From 8725cfbd7c8f8c1ef869ced5f600204e1c88e4ec Mon Sep 17 00:00:00 2001 From: Mario Fetka Date: Mon, 27 Jul 2026 08:59:22 +0200 Subject: [PATCH] Add durable queue retry backoff --- contrib/testing/README.md | 7 +- .../smtp-outbound-opportunistic-tls-check.py | 72 +++++++--- docs/queue.md | 18 +++ docs/test-evidence/0.7-r1.md | 14 +- include/nmap.h | 2 + src/agents/queue/CMakeLists.txt | 10 ++ src/agents/queue/conf.c | 23 ++- src/agents/queue/conf.h | 3 + src/agents/queue/queue.c | 134 ++++++++++++++++-- src/agents/queue/retry-policy.c | 76 ++++++++++ src/agents/queue/retry-policy.h | 36 +++++ src/agents/queue/tests/retry-policy-test.c | 53 +++++++ src/apps/config/config/queue | 5 +- src/apps/config/tests/test_configuration.py | 19 +++ src/libs/python/bongo/configuration/model.py | 9 +- 15 files changed, 443 insertions(+), 38 deletions(-) create mode 100644 src/agents/queue/retry-policy.c create mode 100644 src/agents/queue/retry-policy.h create mode 100644 src/agents/queue/tests/retry-policy-test.c diff --git a/contrib/testing/README.md b/contrib/testing/README.md index 6a2340a..a3c9bfe 100644 --- a/contrib/testing/README.md +++ b/contrib/testing/README.md @@ -83,12 +83,13 @@ one-off files in `/tmp`: delivery when STARTTLS works, plaintext delivery when it is not advertised, and a fresh one-time plaintext reconnect after a failed opportunistic TLS handshake. The fourth MX initially refuses connections; its message must - remain in Queue 7 and be delivered exactly once after the MX becomes + remain in Queue 7 with a persisted initial retry state and be delivered + exactly once by the automatic exponential-backoff path after the MX becomes reachable. Every captured external message must also contain exactly one Bongo DKIM signature with the configured domain and selector. To free standard MX port 25, it temporarily moves Bongo's public listener to port - 10025. It restores the exact SMTP configuration, resolver file, Queue - fixtures, and original Bongo service state. + 10025. It restores the exact SMTP and Queue configurations, resolver file, + Queue fixtures, and original Bongo service state. - `smtp-required-tls-check.py` uses isolated direct-MX fixtures to prove that globally required TLS never falls back after absent or failed STARTTLS. A second phase injects RFC 8689 `REQUIRETLS` over an encrypted trusted diff --git a/contrib/testing/smtp-outbound-opportunistic-tls-check.py b/contrib/testing/smtp-outbound-opportunistic-tls-check.py index b3e12af..b9e8507 100755 --- a/contrib/testing/smtp-outbound-opportunistic-tls-check.py +++ b/contrib/testing/smtp-outbound-opportunistic-tls-check.py @@ -118,25 +118,33 @@ def service_active() -> bool: ).returncode == 0 -def read_smtp_configuration() -> dict: - completed = run([ADMIN, "__config-read", "smtp"]) +def read_configuration(name: str) -> dict: + completed = run([ADMIN, "__config-read", name]) try: configuration = json.loads(completed.stdout) except json.JSONDecodeError as error: raise SMTP07Error( - "bongo-admin returned invalid SMTP configuration") from error + f"bongo-admin returned invalid {name} configuration") from error if not isinstance(configuration, dict): - raise SMTP07Error("SMTP configuration is not an object") + raise SMTP07Error(f"{name} configuration is not an object") return configuration -def replace_smtp_configuration(configuration: dict) -> None: +def replace_configuration(name: str, configuration: dict) -> None: run( - [ADMIN, "__config-replace", "smtp"], + [ADMIN, "__config-replace", name], input_text=json.dumps(configuration, separators=(",", ":")), ) +def read_smtp_configuration() -> dict: + return read_configuration("smtp") + + +def replace_smtp_configuration(configuration: dict) -> None: + replace_configuration("smtp", configuration) + + def wait_for_bongo() -> None: deadline = time.monotonic() + 30 while time.monotonic() < deadline: @@ -216,23 +224,30 @@ def wait_for_queued(marker: str) -> str: f"direct-MX outage did not retain message {marker}") -def retry_when_idle(identifier: str) -> None: +def wait_for_retry_schedule(identifier: str) -> None: + queue_number, entry = identifier.split("-", 1) + control = Path("/var/lib/bongo/spool") / f"c{entry}.{queue_number}" deadline = time.monotonic() + TIMEOUT while time.monotonic() < deadline: - completed = queue("retry", identifier, check=False) - if completed.returncode == 0: + try: + lines = control.read_text( + encoding="ascii", errors="strict").splitlines() + attempts = [ + int(line[1:]) for line in lines + if re.fullmatch(r"Y[0-9]+", line) + ] + delay = control.stat().st_mtime - time.time() + except (FileNotFoundError, ValueError): + time.sleep(0.1) + continue + if attempts == [1] and 0 < delay <= 5: return - if identifier not in queue_ids(): - return - detail = f"{completed.stdout}\n{completed.stderr}".lower() - if "already being processed" not in detail: + if attempts and attempts != [1]: raise SMTP07Error( - f"cannot retry direct-MX queue entry {identifier}: " - f"{completed.stderr.strip() or completed.stdout.strip()}" - ) - time.sleep(0.2) + f"unexpected retry state for {identifier}: {attempts}") + time.sleep(0.1) raise SMTP07Error( - f"direct-MX queue entry {identifier} remained busy") + f"queue entry {identifier} has no persisted initial retry schedule") def wait_for_queue_removal(identifier: str) -> None: @@ -817,6 +832,7 @@ def perform_test() -> None: if not originally_active: raise SMTP07Error("bongo.service must be active before this live test") smtp_configuration = read_smtp_configuration() + queue_configuration = read_configuration("queue") configured_host = str( smtp_configuration.get("internal_relay_bind_address", "") ).strip() @@ -850,8 +866,13 @@ def perform_test() -> None: raise SMTP07Error("SMTP-07 requires a configured DKIM selector") test_smtp_configuration = copy.deepcopy(smtp_configuration) test_smtp_configuration["port"] = TEMPORARY_PUBLIC_PORT + test_queue_configuration = copy.deepcopy(queue_configuration) + test_queue_configuration["queueinterval"] = 1 + test_queue_configuration["retry_initial_seconds"] = 2 + test_queue_configuration["retry_maximum_seconds"] = 4 resolver_changed = False smtp_changed = False + queue_changed = False servers: list[socketserver.BaseServer] = [] queue_clean = False connections: dict[str, int] = {} @@ -865,6 +886,8 @@ def perform_test() -> None: try: replace_smtp_configuration(test_smtp_configuration) smtp_changed = True + replace_configuration("queue", test_queue_configuration) + queue_changed = True run([SYSTEMCTL, "stop", "bongo.service"]) save_resolver(original) atomic_write( @@ -901,6 +924,7 @@ def perform_test() -> None: outage_marker = f"{TOKEN}-{outage_route.mode}" submit(outage_route) outage_queue_id = wait_for_queued(outage_marker) + wait_for_retry_schedule(outage_queue_id) outage_capture = SMTPCapture(outage_route) outage_server = ThreadingTCPServer( @@ -911,7 +935,6 @@ def perform_test() -> None: threading.Thread( target=outage_server.serve_forever, daemon=True ).start() - retry_when_idle(outage_queue_id) encrypted, message = wait_for_marker( outage_capture, outage_marker ) @@ -948,6 +971,13 @@ def perform_test() -> None: except (OSError, SMTP07Error) as error: restoration_errors.append( f"SMTP configuration restore failed: {error}") + if queue_changed: + try: + replace_configuration("queue", queue_configuration) + queue_changed = False + except (OSError, SMTP07Error) as error: + restoration_errors.append( + f"Queue configuration restore failed: {error}") if run( [SYSTEMCTL, "stop", "bongo.service"], check=False ).returncode != 0: @@ -989,13 +1019,15 @@ def perform_test() -> None: raise SMTP07Error("resolver configuration was not restored exactly") if read_smtp_configuration() != smtp_configuration: raise SMTP07Error("SMTP configuration was not restored exactly") + if read_configuration("queue") != queue_configuration: + raise SMTP07Error("Queue configuration was not restored exactly") print( "SMTP-07 PASS " f"starttls=tls/connections:{connections['tls']} " f"plain=plaintext/connections:{connections['plain']} " "handshake-failure=plaintext-reconnect/" f"connections:{connections['fallback']} " - "internet-outage=retained/recovered " + "internet-outage=retained/automatic-backoff/recovered " "dkim=verified " "duplicates=no " f"queue-clean={'yes' if queue_clean else 'no'} " diff --git a/docs/queue.md b/docs/queue.md index e55e768..bb8404f 100644 --- a/docs/queue.md +++ b/docs/queue.md @@ -32,6 +32,24 @@ consuming the filesystem's final space. Per-user Store quota is separate: a full mailbox defers or rejects that recipient without corrupting other recipient state. +Temporary delivery failures use one persistent exponential retry policy, +regardless of whether the next hop is a directly resolved Internet MX, a +configured relayhost, or an internal LMTP destination. The retry counter is +part of the durable Queue envelope and the next eligible attempt survives a +daemon or host restart. `retry_initial_seconds` defaults to 300 seconds and +the delay doubles up to `retry_maximum_seconds`, which defaults to 4000 +seconds. `queueinterval` controls how often the monitor looks for eligible +work; it does not shorten a message's persisted backoff. An administrator can +request an immediate attempt with `bongo-queuetool retry` or `flush`. + +Normal messages expire after `queuetimeout`, while delivery-status messages +and other null-envelope-sender traffic expire after +`bounce_queue_timeout`. Both defaults are 432000 seconds (five days). On +normal-message expiry Bongo uses its existing RFC delivery-status path to +return a failure for the recipients still pending. An expired null-sender +message is discarded after the configured bounce lifetime, preventing a +bounce loop. Held queue entries do not run or expire until released. + Collector backpressure is also separate from the byte reserve. The `collector` document limits globally pending and per-user pending external messages. Queue counts only entries carrying its durable collected-message diff --git a/docs/test-evidence/0.7-r1.md b/docs/test-evidence/0.7-r1.md index 3672a97..ff8eb21 100644 --- a/docs/test-evidence/0.7-r1.md +++ b/docs/test-evidence/0.7-r1.md @@ -2358,13 +2358,15 @@ queries to the required `ns_c_in` class. This fixes direct MX/A resolution and the TXT lookup path shared by mail policy features. The reusable `smtp-outbound-opportunistic-tls-check.py` installs a temporary -loopback DNS resolver and publishes three isolated MX routes: +loopback DNS resolver and publishes four isolated MX routes: - a server which advertises STARTTLS and completes a TLS 1.2-or-newer handshake; - a server which does not advertise STARTTLS; - a server which advertises STARTTLS, accepts the command, then aborts the TLS handshake. +- an initially unreachable MX which becomes available only after Bongo has + durably retained the original Queue 7 entry. ```sh sudo ./contrib/testing/smtp-outbound-opportunistic-tls-check.py --live @@ -2399,6 +2401,16 @@ opportunistic-handshake-fallback deliveries and required exactly one SMTP-07 PASS starttls=tls/connections:1 plain=plaintext/connections:1 handshake-failure=plaintext-reconnect/connections:2 dkim=verified duplicates=no queue-clean=yes resolver-restored=yes config-restored=yes service-restored=active ``` +The outage extension was then run against installed source commit +`9acc2890`. No server listened on the fourth MX address during the first +delivery attempt. Bongo retained one stable Queue 7 entry, the fixture brought +the same MX online, and an explicit retry delivered the original message +exactly once with a valid DKIM signature: + +```text +SMTP-07 PASS starttls=tls/connections:1 plain=plaintext/connections:1 handshake-failure=plaintext-reconnect/connections:2 internet-outage=retained/recovered dkim=verified duplicates=no queue-clean=yes resolver-restored=yes config-restored=yes service-restored=active +``` + ## SMTP-08 Result: **PASS** diff --git a/include/nmap.h b/include/nmap.h index 2c66efe..d83ce93 100644 --- a/include/nmap.h +++ b/include/nmap.h @@ -418,6 +418,7 @@ typedef int NMAPDeliveryCode; #define QUEUE_ADDRESS 'A' #define QUEUE_THIRD_PARTY 'T' #define QUEUE_CLIENT_METADATA 'P' +#define QUEUE_RETRY_STATE 'Y' /* Queue prefixes (String) */ #define QUEUES_ID "I" @@ -432,6 +433,7 @@ typedef int NMAPDeliveryCode; #define QUEUES_ADDRESS "A" #define QUEUES_THIRD_PARTY "T" #define QUEUES_CLIENT_METADATA "P" +#define QUEUES_RETRY_STATE "Y" /* Result codes */ #define NMAP_READY 1000 diff --git a/src/agents/queue/CMakeLists.txt b/src/agents/queue/CMakeLists.txt index 7cc8964..d814c6e 100644 --- a/src/agents/queue/CMakeLists.txt +++ b/src/agents/queue/CMakeLists.txt @@ -9,6 +9,7 @@ add_executable(bongoqueue queued.c mime.c queue.c + retry-policy.c spool-file.c store-delivery-format.c ) @@ -86,4 +87,13 @@ if(BUILD_TESTING) ${CMAKE_CURRENT_SOURCE_DIR} ) add_test(NAME queue-spool-file COMMAND queue-spool-file-test) + + add_executable(queue-retry-policy-test + tests/retry-policy-test.c + retry-policy.c + ) + target_include_directories(queue-retry-policy-test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ) + add_test(NAME queue-retry-policy COMMAND queue-retry-policy-test) endif() diff --git a/src/agents/queue/conf.c b/src/agents/queue/conf.c index 4a20ce2..f6190ad 100644 --- a/src/agents/queue/conf.c +++ b/src/agents/queue/conf.c @@ -42,7 +42,10 @@ #define STACKSPACE_Q (1024*80) #define STACKSPACE_S (1024*80) -#define DEFAULT_MAX_LINGER (4 * 24 * 60 * 60) +#define DEFAULT_MAX_LINGER (5 * 24 * 60 * 60) +#define DEFAULT_BOUNCE_MAX_LINGER (5 * 24 * 60 * 60) +#define DEFAULT_RETRY_INITIAL (5 * 60) +#define DEFAULT_RETRY_MAXIMUM 4000 #define QLIMIT_CONCURRENT 250 #define QLIMIT_SEQUENTIAL 500 @@ -91,11 +94,14 @@ static BongoConfigItem QueueConfig[] = { BONGO_CONFIG_INT("o:queuetuning_load_low/i", &Conf.loadMonitorLow), BONGO_CONFIG_INT("o:queuetuning_trigger/i", &Conf.limitTrigger), BONGO_CONFIG_INT("o:queuetimeout/i", &Conf.maxLinger), + BONGO_CONFIG_INT("o:bounce_queue_timeout/i", &Conf.bounceMaxLinger), { BONGO_JSON_STRING, "o:quotamessage/s", &Conf.quotaMessage }, BONGO_CONFIG_INT("o:port/i", &Agent.agent.port), { BONGO_JSON_BOOL, "o:forwardundeliverable_enabled/b", &Conf.forwardUndeliverableEnabled }, { BONGO_JSON_STRING, "o:forwardundeliverable_to/s", &Conf.forwardUndeliverableAddress }, BONGO_CONFIG_INT("o:queueinterval/i", &Conf.queueInterval), + BONGO_CONFIG_INT("o:retry_initial_seconds/i", &Conf.retryInitial), + BONGO_CONFIG_INT("o:retry_maximum_seconds/i", &Conf.retryMaximum), BONGO_CONFIG_INT("o:minimumfreespace/i", &Conf.minimumFree), { BONGO_JSON_ARRAY, "o:trustedhosts/a", &trustedHostsConfig }, { BONGO_JSON_BOOL, "o:rtsantispamconfig_enabled/b", &Conf.bounceBlockSpam }, @@ -126,6 +132,21 @@ ReadConfiguration (BOOL *recover) if (!ReadBongoConfiguration(GlobalConfig, "global")) { return FALSE; } + + /* + * A pre-0.7 Store document does not contain the retry fields. Keep the + * daemon safe until the configuration editor merges the current template + * and synchronizes it back to Store. + */ + if (Conf.maxLinger <= 0) + Conf.maxLinger = DEFAULT_MAX_LINGER; + if (Conf.bounceMaxLinger <= 0) + Conf.bounceMaxLinger = DEFAULT_BOUNCE_MAX_LINGER; + if (Conf.retryInitial <= 0) + Conf.retryInitial = DEFAULT_RETRY_INITIAL; + if (Conf.retryMaximum < Conf.retryInitial) + Conf.retryMaximum = Conf.retryInitial > DEFAULT_RETRY_MAXIMUM ? + Conf.retryInitial : DEFAULT_RETRY_MAXIMUM; *recover = MsgGetRecoveryFlag("queue"); diff --git a/src/agents/queue/conf.h b/src/agents/queue/conf.h index 162ced0..8430ffd 100644 --- a/src/agents/queue/conf.h +++ b/src/agents/queue/conf.h @@ -63,8 +63,11 @@ typedef struct _QueueConfiguration { char deferEnd[7]; time_t maxLinger; + time_t bounceMaxLinger; time_t queueInterval; + time_t retryInitial; + time_t retryMaximum; /* Forward Undeliverable */ BOOL forwardUndeliverableEnabled; diff --git a/src/agents/queue/queue.c b/src/agents/queue/queue.c index 2124d26..27fd49b 100644 --- a/src/agents/queue/queue.c +++ b/src/agents/queue/queue.c @@ -26,8 +26,10 @@ #include #include #include +#include #include #include +#include #include #include "queue.h" @@ -37,6 +39,7 @@ #include "messages.h" #include "dsn-format.h" #include "envelope-format.h" +#include "retry-policy.h" #include "spool-file.h" #include "store-delivery-format.h" #include "delivery-policy.h" @@ -112,6 +115,82 @@ int IntCmp(const void *key1, const void *key2); static int HandleDSN(FILE *data, FILE *control); static BOOL ProcessQueueEntry(char *entryIn); +static BOOL +ScheduleQueueRetry(const char *entry, int queue) +{ + char controlPath[XPL_MAX_PATH + 1]; + char workPath[XPL_MAX_PATH + 1]; + char line[CONN_BUFSIZE + 1]; + FILE *control = NULL; + FILE *work = NULL; + unsigned long attempt = 0; + unsigned long parsedAttempt; + uint64_t delay; + time_t now; + struct utimbuf timestamps; + BOOL result = FALSE; + + if (!QueueFormat(controlPath, "%s/c%s.%03d", Conf.spoolPath, entry, + queue) || + !QueueFormat(workPath, "%s/w%s.%03d", Conf.spoolPath, entry, queue)) + return FALSE; + + control = fopen(controlPath, "rb"); + work = fopen(workPath, "wb"); + if (!control || !work) + goto cleanup; + + while (fgets(line, sizeof(line), control)) { + if (BongoQueueParseRetryState(line, &parsedAttempt)) { + if (parsedAttempt > attempt) + attempt = parsedAttempt; + continue; + } + if (BongoQueueFileWrite(work, line, strlen(line)) != 0) + goto cleanup; + } + if (ferror(control)) + goto cleanup; + if (fclose(control) != 0) { + control = NULL; + goto cleanup; + } + control = NULL; + + if (attempt != ULONG_MAX) + attempt++; + if (BongoQueueFilePrintf(work, QUEUES_RETRY_STATE"%lu\r\n", + attempt) != 0 || + BongoQueueFileCommit(&work, workPath, controlPath) != 0) + goto cleanup; + + delay = BongoQueueRetryDelay((uint64_t)Conf.retryInitial, + (uint64_t)Conf.retryMaximum, attempt); + now = time(NULL); + timestamps.actime = now; + timestamps.modtime = now + (time_t)delay; + if (utime(controlPath, ×tamps) != 0) { + Log(LOG_WARNING, + "Could not persist retry time for queue entry %03d-%s: %s", + queue, entry, strerror(errno)); + goto cleanup; + } + Log(LOG_NOTICE, + "Deferred queue entry %03d-%s for %" PRIu64 + " seconds after attempt %lu", + queue, entry, delay, attempt); + result = TRUE; + +cleanup: + if (control) + fclose(control); + if (work) + fclose(work); + if (!result) + unlink(workPath); + return result; +} + static void StartImportedQueueEntry(unsigned long id) { @@ -888,6 +967,8 @@ ProcessQueueEntry(char *entryIn) char entry[15]; BOOL keep = TRUE; BOOL bounce = FALSE; + BOOL nullSender = FALSE; + BOOL dateFound = FALSE; time_t date; struct sockaddr_in saddr; struct stat sb; @@ -940,14 +1021,28 @@ StartOver: } if (fh) { - if (fgets(line, CONN_BUFSIZE, fh) == NULL) { + date = 0; + dateFound = FALSE; + nullSender = FALSE; + while (fgets(line, CONN_BUFSIZE, fh) != NULL) { + if (line[0] == QUEUE_DATE) { + date = (time_t)strtoimax(line + 1, NULL, 10); + dateFound = TRUE; + } else if (line[0] == QUEUE_FROM) { + nullSender = line[1] == '-' && + (line[2] == ' ' || line[2] == '\r' || + line[2] == '\n' || line[2] == '\0'); + } + if (dateFound && line[0] == QUEUE_FROM) + break; + } + if (!dateFound) { Log(LOG_ERROR, "Unable to read queue control file '%s': %s", path, ferror(fh) ? strerror(errno) : "unexpected end of file"); FCLOSE_CHECK(fh); ProcessQueueEntryCleanUp(entryID, report); return(FALSE); } - date = atoi(line + 1); FCLOSE_CHECK(fh); } else { ProcessQueueEntryCleanUp(entryID, report); @@ -1256,7 +1351,9 @@ StartOver: /* It's in the deliver queue, check if it's local and deliver it, if not, hand it off to the agents!*/ /* By way of design,this function also removes all entries w/o any receiver from the queue */ case Q_OUTGOING: { - if (date < time(NULL) - Conf.maxLinger) { + time_t lifetime = BongoQueueDeliveryLifetime( + Conf.maxLinger, Conf.bounceMaxLinger, nullSender); + if (date < time(NULL) - lifetime) { /* We move it to the Q_RTS queue and the linger code there will bounce it for us */ QueueFormat(path, "%s/c%s.%03d", Conf.spoolPath, entry, queue); QueueFormat(path2, "%s/c%s.%03d", Conf.spoolPath, entry, Q_RTS); @@ -1754,7 +1851,7 @@ StartOver: entry, Q_DELIVER); RENAME_CHECK(path, Path2); } - Queue.restartNeeded = TRUE; + ScheduleQueueRetry(entry, Q_DELIVER); ProcessQueueEntryCleanUp(entryID, report); return(TRUE); } @@ -2083,7 +2180,9 @@ StartOver: } case Q_RTS: { - if (date < time(NULL) - Conf.maxLinger) { + time_t lifetime = BongoQueueDeliveryLifetime( + Conf.maxLinger, Conf.bounceMaxLinger, nullSender); + if (date < time(NULL) - lifetime) { if ((handle = QDBHandleAlloc()) != NULL) { QDBRemoveID(handle, entryID); @@ -2202,6 +2301,9 @@ StartOver: } } + if (queue == Q_OUTGOING && keep) + ScheduleQueueRetry(entry, Q_OUTGOING); + ProcessQueueEntryCleanUp(entryID, report); return(TRUE); @@ -2857,10 +2959,9 @@ CheckQueue(void *queueIn) flushing = Queue.flushNeeded; Queue.restartNeeded = Queue.flushNeeded = FALSE; + now = time(NULL); if (flushing) { Log(LOG_INFO, "Flushing the queue."); - } else { - now = time(NULL) - Conf.queueInterval + 60; } Log(LOG_DEBUG, "Restarting the queue"); @@ -2892,10 +2993,21 @@ CheckQueue(void *queueIn) continue; } - /* skip recently-touched messages that are in the outgoing - * queue (queue 7) */ - if (!flushing && path[2] == '7' && (XplCalendarTime(dirEntry->d_cdatetime)+600) > (unsigned long)now) { - continue; + /* + * A retained local or remote delivery has a future mtime + * recording its next eligible retry. QFLUSH intentionally + * overrides that schedule for an administrator-requested run. + */ + if (!flushing && + (atoi(path) == Q_DELIVER || atoi(path) == Q_OUTGOING)) { + struct stat retryStat; + char retryPath[XPL_MAX_PATH + 1]; + + QueueFormat(retryPath, "%s/%s", Conf.spoolPath, + dirEntry->d_nameDOS); + if (stat(retryPath, &retryStat) == 0 && + retryStat.st_mtime > now) + continue; } /* wait for an empty queue slot */ diff --git a/src/agents/queue/retry-policy.c b/src/agents/queue/retry-policy.c new file mode 100644 index 0000000..de66618 --- /dev/null +++ b/src/agents/queue/retry-policy.c @@ -0,0 +1,76 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + * To contact Novell about this file by physical or electronic mail, you + * may find current contact information at www.novell.com. + * + ****************************************************************************/ + +#include "retry-policy.h" + +#include + +#include +#include +#include +#include + +uint64_t +BongoQueueRetryDelay(uint64_t initial_seconds, uint64_t maximum_seconds, + unsigned long attempt) +{ + uint64_t delay; + + if (initial_seconds == 0 || maximum_seconds < initial_seconds) + return 0; + delay = initial_seconds; + while (attempt > 1 && delay < maximum_seconds) { + if (delay > maximum_seconds / 2) { + delay = maximum_seconds; + break; + } + delay *= 2; + attempt--; + } + return delay > maximum_seconds ? maximum_seconds : delay; +} + +int +BongoQueueParseRetryState(const char *line, unsigned long *attempt) +{ + char *end; + unsigned long parsed; + + if (!line || !attempt || line[0] != QUEUE_RETRY_STATE || + !isdigit((unsigned char)line[1])) + return 0; + errno = 0; + parsed = strtoul(line + 1, &end, 10); + if (errno == ERANGE || end == line + 1 || + (*end == '\r' && (end[1] != '\n' || end[2] != '\0')) || + (*end == '\n' && end[1] != '\0') || + (*end != '\0' && *end != '\r' && *end != '\n')) + return 0; + *attempt = parsed; + return 1; +} + +time_t +BongoQueueDeliveryLifetime(time_t message_lifetime, time_t bounce_lifetime, + int null_sender) +{ + return null_sender ? bounce_lifetime : message_lifetime; +} diff --git a/src/agents/queue/retry-policy.h b/src/agents/queue/retry-policy.h new file mode 100644 index 0000000..9b64096 --- /dev/null +++ b/src/agents/queue/retry-policy.h @@ -0,0 +1,36 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + * To contact Novell about this file by physical or electronic mail, you + * may find current contact information at www.novell.com. + * + ****************************************************************************/ + +#ifndef BONGO_QUEUE_RETRY_POLICY_H +#define BONGO_QUEUE_RETRY_POLICY_H + +#include +#include + +uint64_t BongoQueueRetryDelay(uint64_t initial_seconds, + uint64_t maximum_seconds, + unsigned long attempt); +int BongoQueueParseRetryState(const char *line, unsigned long *attempt); +time_t BongoQueueDeliveryLifetime(time_t message_lifetime, + time_t bounce_lifetime, + int null_sender); + +#endif diff --git a/src/agents/queue/tests/retry-policy-test.c b/src/agents/queue/tests/retry-policy-test.c new file mode 100644 index 0000000..5d5bcd9 --- /dev/null +++ b/src/agents/queue/tests/retry-policy-test.c @@ -0,0 +1,53 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + * To contact Novell about this file by physical or electronic mail, you + * may find current contact information at www.novell.com. + * + ****************************************************************************/ + +#include "retry-policy.h" + +#include +#include + +int +main(void) +{ + unsigned long attempt = 0; + + assert(BongoQueueRetryDelay(300, 4000, 1) == 300); + assert(BongoQueueRetryDelay(300, 4000, 2) == 600); + assert(BongoQueueRetryDelay(300, 4000, 3) == 1200); + assert(BongoQueueRetryDelay(300, 4000, 4) == 2400); + assert(BongoQueueRetryDelay(300, 4000, 5) == 4000); + assert(BongoQueueRetryDelay(300, 4000, 1000000) == 4000); + assert(BongoQueueRetryDelay(0, 4000, 1) == 0); + assert(BongoQueueRetryDelay(4001, 4000, 1) == 0); + + assert(BongoQueueParseRetryState("Y1", &attempt) && attempt == 1); + assert(BongoQueueParseRetryState("Y42\r\n", &attempt) && attempt == 42); + assert(!BongoQueueParseRetryState("Y", &attempt)); + assert(!BongoQueueParseRetryState("Y-1", &attempt)); + assert(!BongoQueueParseRetryState("Y1 trailing", &attempt)); + assert(!BongoQueueParseRetryState("Y1\rtrailing", &attempt)); + assert(!BongoQueueParseRetryState("Y1\ntrailing", &attempt)); + assert(!BongoQueueParseRetryState("X1", &attempt)); + + assert(BongoQueueDeliveryLifetime(432000, 432000, 0) == 432000); + assert(BongoQueueDeliveryLifetime(432000, 86400, 1) == 86400); + return 0; +} diff --git a/src/apps/config/config/queue b/src/apps/config/config/queue index 824a773..1f4a6ec 100644 --- a/src/apps/config/config/queue +++ b/src/apps/config/config/queue @@ -11,12 +11,15 @@ "queuetuning_load_high" : 0, "queuetuning_load_low" : 0, "queuetuning_trigger" : 0, - "queuetimeout" : 345600, + "queuetimeout" : 432000, + "bounce_queue_timeout" : 432000, "quotamessage" : "", "port" : 8670, "forwardundeliverable_enabled" : false, "forwardundeliverable_to" : "", "queueinterval" : 10, + "retry_initial_seconds" : 300, + "retry_maximum_seconds" : 4000, "minimumfreespace" : 5242880, "trustedhosts" : [], "hosteddomains" : [], diff --git a/src/apps/config/tests/test_configuration.py b/src/apps/config/tests/test_configuration.py index f3f7d03..d3d7dc1 100644 --- a/src/apps/config/tests/test_configuration.py +++ b/src/apps/config/tests/test_configuration.py @@ -162,6 +162,25 @@ class ConfigurationModelTest(unittest.TestCase): for issue in errors(validate_all( configs, defaults, check_files=False)))) + def test_queue_retry_delays_are_positive_and_ordered(self): + defaults = templates() + configs = templates() + queue = configs["queue"] + queue["retry_initial_seconds"] = 600 + queue["retry_maximum_seconds"] = 300 + problems = errors(validate_all(configs, defaults, check_files=False)) + self.assertTrue(any( + issue.document == "queue" and + issue.path == "retry_maximum_seconds" + for issue in problems)) + + queue["retry_maximum_seconds"] = 1200 + self.assertFalse(any( + issue.document == "queue" and + issue.path.startswith("retry_") + for issue in errors(validate_all( + configs, defaults, check_files=False)))) + def test_smtp_socket_timeout_is_bounded(self): defaults = templates() for value in (0, 3601): diff --git a/src/libs/python/bongo/configuration/model.py b/src/libs/python/bongo/configuration/model.py index 7b58d12..464ff3c 100644 --- a/src/libs/python/bongo/configuration/model.py +++ b/src/libs/python/bongo/configuration/model.py @@ -505,9 +505,16 @@ def _validate_queue(config: Mapping[str, Any], issues: list[Issue]) -> None: _validate_domains(issues, "queue", "domains", config.get("domains"), required=True) _validate_domains(issues, "queue", "hosteddomains", config.get("hosteddomains")) _port(issues, "queue", config, "port") - for key in ("queueinterval", "queuetimeout", "minimumfreespace", + for key in ("queueinterval", "queuetimeout", "bounce_queue_timeout", + "retry_initial_seconds", "retry_maximum_seconds", + "minimumfreespace", "queuetuning_concurrent", "queuetuning_sequential"): _positive(issues, "queue", config, key) + initial = config.get("retry_initial_seconds") + maximum = config.get("retry_maximum_seconds") + if isinstance(initial, int) and isinstance(maximum, int) and maximum < initial: + _issue(issues, "queue", "retry_maximum_seconds", + "maximum retry delay must not be shorter than the initial delay") if config.get("forwardundeliverable_enabled") and not config.get("forwardundeliverable_to"): _issue(issues, "queue", "forwardundeliverable_to", "a destination is required when forwarding is enabled")