Ignore orphaned collector queue controls

contrib/testing/collector-flood-quota-check.py

@@ -535,11 +535,11 @@ def main() -> int:
         )
         blocked_collector["maximum_pending_messages_per_user"] = 1
         replace_configuration("collector", blocked_collector)
+        collector_changed = True
+        restart_bongo()
         wait_local_configuration(
             "collector", "maximum_pending_messages_per_user", 1
         )
-        collector_changed = True
-        restart_bongo()
         api = WebApi()
         seed("normal", 2, HOLD_SOURCE[0])
         hold_id = api.create_account(HOLD_SOURCE)
@@ -563,24 +563,24 @@ def main() -> int:
         api.delete_account(hold_id)
         account_ids.remove(hold_id)
         replace_configuration("collector", original_collector)
+        restart_bongo()
         wait_local_configuration(
             "collector", "maximum_pending_messages_per_user",
             original_collector.get("maximum_pending_messages_per_user"),
         )
         collector_changed = False
-        restart_bongo()
         clear_remote(HOLD_SOURCE[0])
     finally:
         if collector_changed:
             try:
                 replace_configuration("collector", original_collector)
+                restart_bongo()
                 wait_local_configuration(
                     "collector", "maximum_pending_messages_per_user",
                     original_collector.get(
                         "maximum_pending_messages_per_user"
                     ),
                 )
-                restart_bongo()
             except Exception as cleanup_error:
                 print(
                     f"STQ-12 cleanup warning (Collector): {cleanup_error}",

src/agents/queue/collected-count.c

@@ -18,15 +18,40 @@
 
 #include <config.h>
 
+#include <ctype.h>
 #include <errno.h>
 #include <limits.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/stat.h>
 
 #include <nmap.h>
 
 #include "collected-count.h"
 
+int
+BongoQueueCollectedDataExists(const char *spool_path, const char *id)
+{
+    char path[XPL_MAX_PATH + 1];
+    struct stat status;
+    size_t position;
+    int written;
+
+    if (spool_path == NULL || spool_path[0] == '\0' ||
+        id == NULL || strlen(id) != 7) {
+        return 0;
+    }
+    for (position = 0; position < 7; position++) {
+        if (!isxdigit((unsigned char)id[position]))
+            return 0;
+    }
+    written = snprintf(path, sizeof(path), "%s/d%s.msg", spool_path, id);
+    if (written < 0 || (size_t)written >= sizeof(path))
+        return 0;
+    return stat(path, &status) == 0 && S_ISREG(status.st_mode);
+}
+
 static BOOL
 RecipientMatches(const char *line, const char *owner)
 {

src/agents/queue/collected-count.h

@@ -25,5 +25,6 @@
 
 int BongoQueueCollectedEnvelope(FILE *control, const char *owner,
                                 BOOL *owner_matches);
+int BongoQueueCollectedDataExists(const char *spool_path, const char *id);
 
 #endif

src/agents/queue/queue.c

@@ -3540,6 +3540,13 @@ CommandQcountCollected(void *param)
         id[7] = '\0';
         if (g_hash_table_contains(counted, id))
             continue;
+        /*
+         * A worker control file can briefly outlive its data file after a
+         * successful delivery or an administrative QDELE.  Such an orphan is
+         * not a pending message and must not hold the Collector at its cap.
+         */
+        if (!BongoQueueCollectedDataExists(Conf.spoolPath, id))
+            continue;
 
         QueueFormat(path, "%s/%s", Conf.spoolPath, name);
         control = fopen(path, "rb");
@@ -3588,6 +3595,12 @@ CommandQdele(void *param)
     QueueFormat(client->path, "%s/c%s.%s", Conf.spoolPath, ptr + 4, ptr);
     UNLINK_CHECK(client->path);
 
+    QueueFormat(client->path, "%s/w%s.%s", Conf.spoolPath, ptr + 4, ptr);
+    if (unlink(client->path) != 0 && errno != ENOENT) {
+        Log(LOG_WARNING, "Unable to delete Queue work file %s: %d",
+            client->path, errno);
+    }
+
     QueueFormat(client->path, "%s/d%s.msg", Conf.spoolPath, ptr + 4);
     UNLINK_CHECK(client->path);
 

src/agents/queue/tests/collected-count-test.c

@@ -18,7 +18,9 @@
 
 #include <assert.h>
 #include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
+#include <unistd.h>
 
 #include <nmap.h>
 
@@ -43,6 +45,9 @@ int
 main(void)
 {
     char collected[256];
+    char directory[] = "/tmp/bongo-collected-count-XXXXXX";
+    char data_path[256];
+    FILE *data;
 
     snprintf(collected, sizeof(collected),
              "D1\r\nX%u\r\nMtest1 test1 32 INBOX 0\r\n",
@@ -54,5 +59,18 @@ main(void)
     Check("D1\r\nX32768junk\r\nMtest1 test1 32 INBOX 0\r\n", "test1", -1,
           FALSE);
     Check("D1\r\nX32768\r\nLtest2 test2 0\r\n", "test2", 1, TRUE);
+    assert(mkdtemp(directory) != NULL);
+    assert(snprintf(data_path, sizeof(data_path), "%s/d0123abc.msg",
+                    directory) < (int)sizeof(data_path));
+    assert(!BongoQueueCollectedDataExists(directory, "0123abc"));
+    data = fopen(data_path, "wb");
+    assert(data != NULL);
+    assert(fputs("message", data) >= 0);
+    assert(fclose(data) == 0);
+    assert(BongoQueueCollectedDataExists(directory, "0123abc"));
+    assert(!BongoQueueCollectedDataExists(directory, "0123abz"));
+    assert(!BongoQueueCollectedDataExists(directory, "short"));
+    assert(unlink(data_path) == 0);
+    assert(rmdir(directory) == 0);
     return 0;
 }