diff --git a/contrib/testing/collector-flood-quota-check.py b/contrib/testing/collector-flood-quota-check.py index 6e65c0a..bcc5ab1 100755 --- a/contrib/testing/collector-flood-quota-check.py +++ b/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}", diff --git a/src/agents/queue/collected-count.c b/src/agents/queue/collected-count.c index b09c023..84ce4f2 100644 --- a/src/agents/queue/collected-count.c +++ b/src/agents/queue/collected-count.c @@ -18,15 +18,40 @@ #include +#include #include #include +#include #include #include +#include #include #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) { diff --git a/src/agents/queue/collected-count.h b/src/agents/queue/collected-count.h index 9052121..717d25f 100644 --- a/src/agents/queue/collected-count.h +++ b/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 diff --git a/src/agents/queue/queue.c b/src/agents/queue/queue.c index 9b55769..46390d8 100644 --- a/src/agents/queue/queue.c +++ b/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); diff --git a/src/agents/queue/tests/collected-count-test.c b/src/agents/queue/tests/collected-count-test.c index efa060c..9ba4f4e 100644 --- a/src/agents/queue/tests/collected-count-test.c +++ b/src/agents/queue/tests/collected-count-test.c @@ -18,7 +18,9 @@ #include #include +#include #include +#include #include @@ -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; }