Preserve collector mailbox delivery through queue

contrib/testing/README.md

@@ -90,6 +90,24 @@ The generated short-lived test CA at
 `/tmp/bongo-cyrus-fixture/tls/ca.crt` is the fixture trust anchor. Set
 `BONGO_COLLECTOR_CAINFO` to that path for the Collector process during this
 test; normal installations continue to use the operating-system trust store.
+The packaged `bongo.service` uses `PrivateTmp=true`, so a live systemd test
+must make the fixture visible inside that service namespace. A runtime-only
+drop-in can bind precisely this fixture read-only:
+
+```ini
+[Service]
+BindReadOnlyPaths=/tmp/bongo-cyrus-fixture
+```
+
+Install it with `systemctl edit --runtime bongo.service`, restart Bongo, and
+set the Collector `ca_file` to the path above. Do not weaken certificate or
+host-name verification for the test.
+
+After a live collection cycle, show the source-mailbox counts with:
+
+```sh
+./contrib/testing/cyrus-fixture.py mailbox-status
+```
 
 Stop or remove all fixture state with:
 

contrib/testing/cyrus-fixture.py

@@ -419,16 +419,43 @@ def verify_collector(root: Path, probe: str, password: str, count: int) -> None:
 
 def status(root: Path) -> None:
     pid = read_pid(root)
-    print(f"pid: {pid if pid else 'stopped'}")
-    for name, port in PORTS.items():
+    ready = []
+    for port in PORTS.values():
         try:
             with socket.create_connection(("127.0.0.1", port), timeout=0.25):
-                state = "ready"
+                ready.append(True)
         except OSError:
-            state = "closed"
+            ready.append(False)
+    if pid:
+        print(f"pid: {pid}")
+    elif any(ready):
+        print("pid: unavailable (fixture ports are ready)")
+    else:
+        print("pid: stopped")
+    for (name, port), is_ready in zip(PORTS.items(), ready):
+        state = "ready" if is_ready else "closed"
         print(f"{name}: 127.0.0.1:{port} {state}")
 
 
+def mailbox_status(password: str) -> None:
+    for username, protocol, owner, folder in SOURCE_ACCOUNTS:
+        client = imaplib.IMAP4("127.0.0.1", PORTS["imap"])
+        try:
+            client.login(username, password)
+            result, values = client.status("INBOX", "(MESSAGES)")
+            if result != "OK" or not values or values[0] is None:
+                raise RuntimeError(f"could not inspect Cyrus mailbox for {username}")
+            text = values[0].decode("ascii", "strict")
+            count = int(text.rsplit(" ", 1)[1].rstrip(")"))
+            print(f"{username}: {count} message(s) via {protocol.upper()} -> "
+                  f"{owner}/{folder}")
+        finally:
+            try:
+                client.logout()
+            except OSError:
+                pass
+
+
 def reset(root: Path) -> None:
     require_root()
     stop(root)
@@ -442,7 +469,8 @@ def parser() -> argparse.ArgumentParser:
     result.add_argument("--root", type=checked_root, default=DEFAULT_ROOT)
     result.add_argument("--password", default=DEFAULT_PASSWORD)
     commands = result.add_subparsers(dest="command", required=True)
-    for name in ("prepare", "start", "stop", "status", "reset"):
+    for name in ("prepare", "start", "stop", "status", "mailbox-status",
+                 "reset"):
         commands.add_parser(name)
     delivery = commands.add_parser("inject")
     delivery.add_argument("file", help="RFC 5322 message file or - for stdin")
@@ -477,6 +505,8 @@ def main() -> int:
         stop(args.root)
     elif args.command == "status":
         status(args.root)
+    elif args.command == "mailbox-status":
+        mailbox_status(args.password)
     elif args.command == "reset":
         reset(args.root)
     elif args.command == "seed":

docs/external-accounts.md

@@ -23,6 +23,8 @@ operating system trust store when possible. `BONGO_COLLECTOR_CAINFO` remains
 available as a test-only environment override.
 
 Remote deletion policies are `never`, `after_import`, and `after_days`.
-Messages are recorded as imported only after the Bongo store accepts them over
-NMAP. A failed remote deletion does not cause the message to be delivered
-again; the collector retries the deletion during a later poll.
+Messages are recorded as imported only after Bongo's durable Queue accepts and
+commits the complete message. The normal antivirus, antispam, Sieve/rules,
+quota, and Store delivery pipeline owns it from that point onward. A failed
+remote deletion does not cause the message to be delivered again; the
+collector retries the deletion during a later poll.

src/agents/collector/collector.c

@@ -190,8 +190,14 @@ CollectorLoop(void *ignored)
                 Log(LOG_ERROR, "Unable to query due external mail accounts");
             } else if (due > 0) {
                 Log(LOG_DEBUG, "%d external mail account(s) are due for collection", due);
-                CollectorExternalAccount account;
-                if (CollectorAccountsNextDue(accounts, (long) now, &account) == 1) {
+                while (due-- > 0 &&
+                       BongoAgentGetState(&Collector.agent) <
+                           BONGO_AGENT_STATE_STOPPING) {
+                    CollectorExternalAccount account;
+                    if (CollectorAccountsNextDue(accounts, (long) now,
+                                                 &account) != 1) {
+                        break;
+                    }
                     CollectExternalAccount(accounts, &account, (long) now);
                 }
             }

src/agents/queue/envelope-format.c

@@ -25,8 +25,16 @@
 #include <stdlib.h>
 #include <string.h>
 
+char
+BongoQueueLocalAliasRecipientType(char original_type)
+{
+    return original_type == 'M' ? 'M' : 'L';
+}
+
 int
-BongoQueueParseMailboxRecipient(char *line, char **mailbox,
+BongoQueueParseMailboxRecipient(char *line,
+                                unsigned long *delivery_flags,
+                                char **mailbox,
                                 unsigned long *message_flags,
                                 char **pre_mailbox_delimiter,
                                 char **post_mailbox_delimiter)
@@ -36,9 +44,11 @@ BongoQueueParseMailboxRecipient(char *line, char **mailbox,
     char *pre;
     char *post;
     char *end;
-    unsigned long flags;
+    unsigned long delivery;
+    unsigned long message;
 
-    if (line == NULL || line[0] != 'M' || mailbox == NULL ||
+    if (line == NULL || line[0] != 'M' || delivery_flags == NULL ||
+        mailbox == NULL ||
         message_flags == NULL || pre_mailbox_delimiter == NULL ||
         post_mailbox_delimiter == NULL) return 0;
     first = strchr(line + 1, ' ');
@@ -49,13 +59,17 @@ BongoQueueParseMailboxRecipient(char *line, char **mailbox,
     post = strrchr(pre + 1, ' ');
     if (post == NULL || post == pre + 1 || post[1] == '\0') return 0;
     errno = 0;
-    flags = strtoul(post + 1, &end, 10);
+    delivery = strtoul(second + 1, &end, 10);
+    if (errno == ERANGE || end == second + 1 || end != pre) return 0;
+    errno = 0;
+    message = strtoul(post + 1, &end, 10);
     if (errno == ERANGE || end == post + 1 || *end != '\0') return 0;
 
     *pre = '\0';
     *post = '\0';
+    *delivery_flags = delivery;
     *mailbox = pre + 1;
-    *message_flags = flags;
+    *message_flags = message;
     *pre_mailbox_delimiter = pre;
     *post_mailbox_delimiter = post;
     return 1;

src/agents/queue/envelope-format.h

@@ -22,9 +22,13 @@
 #ifndef BONGO_QUEUE_ENVELOPE_FORMAT_H
 #define BONGO_QUEUE_ENVELOPE_FORMAT_H
 
-int BongoQueueParseMailboxRecipient(char *line, char **mailbox,
+int BongoQueueParseMailboxRecipient(char *line,
+                                    unsigned long *delivery_flags,
+                                    char **mailbox,
                                     unsigned long *message_flags,
                                     char **pre_mailbox_delimiter,
                                     char **post_mailbox_delimiter);
 
+char BongoQueueLocalAliasRecipientType(char original_type);
+
 #endif

src/agents/queue/queue.c

@@ -871,7 +871,9 @@ StartOver:
                             case 1000:
                                 /* this is a local user, change the envelope to reflect that
                                  * in this case ptr2+1 is the orecip which will also include the flags */
-                                fprintf(newFH, QUEUES_RECIP_LOCAL"%s %s\r\n", addrptr, ptr2+1);
+                                fprintf(newFH, "%c%s %s\r\n",
+                                        BongoQueueLocalAliasRecipientType(*cur),
+                                        addrptr, ptr2+1);
                                 break;
                             case 1002:
                                 /* in this case ptr2+1 is the orecip which will also contain the flags */
@@ -1163,7 +1165,7 @@ StartOver:
 
                         case QUEUE_RECIP_MBOX_LOCAL: {    /* Local w/ Mailbox; syntax: MRecip ORecip flags MBox MsgFlags */
                             if (!BongoQueueParseMailboxRecipient(
-                                    line, &mailbox, &messageFlags,
+                                    line, &flags, &mailbox, &messageFlags,
                                     &preMailboxDelim,
                                     &postMailboxDelim)) {
                                 Log(LOG_WARNING,

src/agents/queue/tests/envelope-format-test.c

@@ -26,19 +26,22 @@
 
 static void
 Check(const char *source, const char *expected_mailbox,
-      unsigned long expected_flags)
+      unsigned long expected_delivery_flags,
+      unsigned long expected_message_flags)
 {
     char line[256];
     char *mailbox = NULL;
     char *pre = NULL;
     char *post = NULL;
-    unsigned long flags = 0;
+    unsigned long delivery_flags = 0;
+    unsigned long message_flags = 0;
 
     strcpy(line, source);
     assert(BongoQueueParseMailboxRecipient(
-        line, &mailbox, &flags, &pre, &post));
+        line, &delivery_flags, &mailbox, &message_flags, &pre, &post));
     assert(!strcmp(mailbox, expected_mailbox));
-    assert(flags == expected_flags);
+    assert(delivery_flags == expected_delivery_flags);
+    assert(message_flags == expected_message_flags);
     assert(pre != NULL && post != NULL);
     *pre = ' ';
     *post = ' ';
@@ -52,13 +55,17 @@ main(void)
     char *mailbox = NULL;
     char *pre = NULL;
     char *post = NULL;
-    unsigned long flags = 0;
+    unsigned long delivery_flags = 0;
+    unsigned long message_flags = 0;
 
-    Check("Mtest1 test1 32 INBOX 0", "INBOX", 0);
-    Check("Mtest1 test1 32 Cyrus IMAP 17", "Cyrus IMAP", 17);
-    Check("Mtest1 test1 32 Archive/2026 July 4", "Archive/2026 July", 4);
+    Check("Mtest1 test1 32 INBOX 0", "INBOX", 32, 0);
+    Check("Mtest1 test1 32 Cyrus IMAP 17", "Cyrus IMAP", 32, 17);
+    Check("Mtest1 test1 32 Archive/2026 July 4", "Archive/2026 July", 32, 4);
+    assert(BongoQueueLocalAliasRecipientType('M') == 'M');
+    assert(BongoQueueLocalAliasRecipientType('L') == 'L');
+    assert(BongoQueueLocalAliasRecipientType('R') == 'L');
     assert(!BongoQueueParseMailboxRecipient(
-        malformed, &mailbox, &flags, &pre, &post));
+        malformed, &delivery_flags, &mailbox, &message_flags, &pre, &post));
     assert(!strcmp(malformed, "Mtest1 test1 32 Cyrus IMAP"));
     return 0;
 }

src/agents/rules/CMakeLists.txt

@@ -1,6 +1,11 @@
 add_executable(bongorules
 	rules.c
 	stream.c
+	../queue/envelope-format.c
+)
+
+target_include_directories(bongorules PRIVATE
+	${CMAKE_CURRENT_SOURCE_DIR}/../queue
 )
 
 target_link_libraries(bongorules

src/agents/rules/rules.c

@@ -34,6 +34,7 @@
 #include <connio.h>
 
 #include "rules.h"
+#include "envelope-format.h"
 
 RulesAgentGlobals RulesAgent = {0};
 
@@ -1501,13 +1502,33 @@ ProcessEntry(void *clientp, Connection *conn)
                 Queue->sender = envelopeLine;
             }
             break;
+        case QUEUE_RECIP_MBOX_LOCAL:
         case QUEUE_RECIP_LOCAL:
             /* clean up an previous run */
             RulesCleanupInformation(Queue);
 
             /* isolate out any flags */
-            ptr = strrchr(envelopeLine, ' ');
-            Queue->flags = (ptr) ? atoi(ptr+1) : 0;
+            if (*envelopeLine == QUEUE_RECIP_MBOX_LOCAL) {
+                char *mailboxLine = MemStrdup(envelopeLine);
+                char *mailbox = NULL;
+                char *preMailbox = NULL;
+                char *postMailbox = NULL;
+                unsigned long deliveryFlags = 0;
+                unsigned long messageFlags = 0;
+
+                if (mailboxLine == NULL || !BongoQueueParseMailboxRecipient(
+                        mailboxLine, &deliveryFlags, &mailbox, &messageFlags,
+                        &preMailbox, &postMailbox)) {
+                    MemFree(mailboxLine);
+                    ConnWriteF(Queue->conn, "QMOD RAW %s\r\n", envelopeLine);
+                    break;
+                }
+                Queue->flags = deliveryFlags;
+                MemFree(mailboxLine);
+            } else {
+                ptr = strrchr(envelopeLine, ' ');
+                Queue->flags = (ptr) ? atoi(ptr+1) : 0;
+            }
 
             if (Queue->flags & NO_RULEPROCESSING) {
                 ConnWriteF(Queue->conn, "QMOD RAW %s\r\n", envelopeLine);
@@ -1516,7 +1537,7 @@ ProcessEntry(void *clientp, Connection *conn)
 
             Queue->envelopeLine = MemStrdup(envelopeLine);
 
-            /* increment past the L */
+            /* increment past the local-recipient type */
             envelopeLine++;
 
             /* isolate out the recipient address */