diff --git a/contrib/testing/collector-quota-retry-check.py b/contrib/testing/collector-quota-retry-check.py index 3501269..f476888 100755 --- a/contrib/testing/collector-quota-retry-check.py +++ b/contrib/testing/collector-quota-retry-check.py @@ -29,6 +29,8 @@ QUEUE_TOOL = os.environ.get( "BONGO_TEST_QUEUE_TOOL", "/usr/bin/bongo-queuetool" ) ALLOW_LIVE = os.environ.get("BONGO_ALLOW_LIVE_USER_TEST") == "1" +TRACE = os.environ.get("BONGO_TEST_TRACE") == "1" +KEEP_FAILURE = os.environ.get("BONGO_TEST_KEEP_FAILURE") == "1" TOKEN = f"stq10-collector-{os.getpid()}-{int(time.time())}" STARTED_AT = time.monotonic() @@ -41,6 +43,12 @@ def elapsed() -> float: return time.monotonic() - STARTED_AT +def trace(message: str) -> None: + if TRACE: + print(f"STQ-10 collector trace {elapsed():.3f}s: {message}", + file=sys.stderr, flush=True) + + def run( arguments: list[str], *, @@ -157,6 +165,8 @@ def token_queue_ids() -> list[str]: def wait_for_retained_delivery_queue(timeout: float) -> str: deadline = time.monotonic() + timeout last_ids: list[str] = [] + retained_id: str | None = None + stable_since: float | None = None while time.monotonic() < deadline: last_ids = token_queue_ids() # Queue 7 performs outgoing preparation and deliberately falls @@ -166,7 +176,16 @@ def wait_for_retained_delivery_queue(timeout: float) -> str: if item.startswith(("006-", "007-")) ] if len(delivery) == 1: - return delivery[0] + if delivery[0] != retained_id: + retained_id = delivery[0] + stable_since = time.monotonic() + trace(f"candidate retained queue {retained_id}") + elif stable_since is not None and time.monotonic() - stable_since >= 1: + trace(f"retained queue settled as {retained_id}") + return retained_id + else: + retained_id = None + stable_since = None time.sleep(0.25) raise CollectorQuotaTestError( "collected message did not settle in a final delivery queue; " @@ -176,6 +195,7 @@ def wait_for_retained_delivery_queue(timeout: float) -> str: def wait_for_store_delivery(store: StoreClient, timeout: float) -> str: deadline = time.monotonic() + timeout + next_retry = 0.0 while time.monotonic() < deadline: documents = token_store_documents(store) if len(documents) == 1: @@ -184,8 +204,17 @@ def wait_for_store_delivery(store: StoreClient, timeout: float) -> str: raise CollectorQuotaTestError( "quota retry duplicated the collected message" ) - for queue_id in token_queue_ids(): - run_queue("retry", queue_id, check=False) + queue_ids = token_queue_ids() + now = time.monotonic() + if queue_ids and now >= next_retry: + for queue_id in queue_ids: + completed = run_queue("retry", queue_id, check=False) + trace( + f"retry {queue_id}: exit={completed.returncode} " + f"stdout={completed.stdout.strip()!r} " + f"stderr={completed.stderr.strip()!r}" + ) + next_retry = now + 2 time.sleep(0.25) raise CollectorQuotaTestError( "collected message was not delivered after quota was released" @@ -211,7 +240,9 @@ def main() -> int: try: store = open_store() before = inbox_documents(store) + trace(f"baseline documents={len(before)} bytes={baseline}") set_admin_quota(baseline if baseline > 0 else 1) + trace("quota constrained") with tempfile.TemporaryDirectory( prefix="bongo-stq10-collector-" @@ -228,6 +259,7 @@ def main() -> int: f"unexpected collected Queue ID: {initial_id!r}" ) known_queue_ids.add(initial_id) + trace(f"enqueued as {initial_id}") retained_id = wait_for_retained_delivery_queue(30) known_queue_ids.add(retained_id) @@ -255,7 +287,9 @@ def main() -> int: ) set_admin_quota(0) + trace("quota released") delivered_document = wait_for_store_delivery(store, 30) + trace(f"delivered as Store document {delivered_document}") created_documents.append(delivered_document) if set(token_store_documents(store)) != {delivered_document}: raise CollectorQuotaTestError( @@ -278,8 +312,15 @@ def main() -> int: f"STQ-10 collector cleanup warning: {error}", file=sys.stderr, ) - for queue_id in set(token_queue_ids()) | known_queue_ids: - run_queue("delete", queue_id, check=False) + failed = sys.exc_info()[0] is not None + if not (failed and KEEP_FAILURE): + for queue_id in set(token_queue_ids()) | known_queue_ids: + run_queue("delete", queue_id, check=False) + elif failed: + trace( + "preserving failed Queue entries: " + f"{sorted(set(token_queue_ids()) | known_queue_ids)!r}" + ) if store is not None: try: store.Quit() diff --git a/src/agents/queue/queue.c b/src/agents/queue/queue.c index 2ed7c20..20dea92 100644 --- a/src/agents/queue/queue.c +++ b/src/agents/queue/queue.c @@ -1038,6 +1038,7 @@ StartOver: unsigned long messageFlags; unsigned long queueMessageFlags = 0; unsigned long flags = DSN_FAILURE | DSN_HEADER | DSN_BODY; + BOOL retryLocalDelivery = FALSE; char *mailbox; char *preMailboxDelim = NULL; char *postMailboxDelim = NULL; @@ -1311,6 +1312,7 @@ StartOver: } fprintf(newFH, "%s\r\n", line); keep = TRUE; + retryLocalDelivery = TRUE; Log(LOG_NOTICE, "Deferring external message %s for %s: " "local Store quota exceeded", @@ -1493,6 +1495,25 @@ StartOver: QueueFormat(Path2, "%s/w%s.%03d", Conf.spoolPath, entry, queue); RENAME_CHECK(Path2, path); + + /* + * Collector removed the remote source only after Queue + * durably committed this message. On local Store quota + * exhaustion, keep that sole copy in the local-delivery + * queue. Continuing through the outgoing SMTP agent would + * hand it a local mailbox recipient and hold this Queue + * entry until the agent's network timeout. + */ + if (retryLocalDelivery) { + if (queue != Q_DELIVER) { + QueueFormat(Path2, "%s/c%s.%03d", Conf.spoolPath, + entry, Q_DELIVER); + RENAME_CHECK(path, Path2); + } + Queue.restartNeeded = TRUE; + ProcessQueueEntryCleanUp(entryID, report); + return(TRUE); + } } else { QueueFormat(path, "%s/w%s.%03d", Conf.spoolPath, entry, queue); UNLINK_CHECK(path);