Keep Collector quota retries in local queue

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()

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);