log tailer: expiry, Postfix's time values and refusal lines as Postfix writes them
test / test (push) Canceled after 0s

Expiry (assumptions A-8). Postfix 3.7.11 logs the end of a message's queue
lifetime once per message, "<id>: from=<..>, status=expired, returned to
sender" (src/qmgr/qmgr_active.c, qmgr_active_done_25_generic()), with no
line per recipient; the tailer waited for "to=..., status=expired", which
no Postfix writes, and the sweep closed only queued rows, so such rows
stayed deferred. The tailer now reads qmgr's line (ExpireMessage: each
recipient's newest open row of the id becomes bounced); the sweep closes a
deferred row too when Postfix no longer holds the message, and counts as
missed only a final line (sent, bounced, expiry). The queue listing's '#'
(forced expiry, showq_compat.c showq_message()) is read as a flag (B-PQ1).

Time values (B-PQ10, B-PQ11): conv_time() ported - one number, one
lower-case unit, the parameter's default unit when none (get_def_time_unit:
d for the lifetimes, h for delay_warning_time), the rest ignored; postconf
-h -x expands $name; bounce_queue_lifetime capped at maximal_queue_lifetime
as qmgr_post_init() caps it, so LongestLifetime is the queue lifetime.

Refusals (A-21, A-23, A-24, A-15): smtpd_whatsup() ends the NOQUEUE line
with sasl_username=<login> on an authenticated session; the line is now
matched to its row by client, login and sender, and a line without a login
is no application's. The client= line's login ends at the comma. The
"IPv6:" trim goes (namaddr is built from state->addr). A row recovered from
cleanup's line says that line names the last recipient only.

Stand (2026-10-14, make e2e, exit 0): new expired_message_is_closed -
defer_transports=smtp makes the retry service defer the message, postsuper
-e and postqueue -i force its expiry; mail.log has "<id>:
from=<alerts@sender.e2e.test>, status=force-expired, returned to sender"
and the row goes deferred -> bounced; refusals_show_postfixs_line passes
with the login match. go vet and go test ./... pass on the server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mix
2026-10-14 20:21:51 +03:00
co-authored by Claude Opus 5.5
parent 55aef6e721
commit 0ed03b2094
24 changed files with 594 additions and 212 deletions
+27
View File
@@ -32,6 +32,33 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version
### Fixed ### Fixed
- **A message Postfix gave up on no longer stays "deferred" for ever.**
Postfix records the end of a message's queue lifetime in one line for the
whole message, without the recipients; the panel waited for a line per
recipient that Postfix never writes, and the reconcile sweep closed only
`queued` rows, so such rows said `deferred` until they were pruned. The
log tailer now reads that line and closes the message's rows as
`bounced`; the sweep also closes a `deferred` row whose message Postfix no
longer holds (which covers the rows left so far, and `postsuper -d`); a
message forced to expire (`postsuper -e`, `#` in the queue listing) is
still counted as held (audit 2026-10-12, stage 2c; assumptions A-8,
B-PQ1).
- **Postfix's time values are read as Postfix reads them.** A bare number
took seconds, so `maximal_queue_lifetime = 3` was three seconds to the
panel and three days to Postfix; `1h7m` was 67 minutes to the panel and
one hour to Postfix. A value now takes the parameter's own default unit
and one unit only; `$name` references are expanded; a
`bounce_queue_lifetime` longer than `maximal_queue_lifetime` is cut to it,
as Postfix's queue manager does (assumptions B-PQ10, B-PQ11).
- **A refusal's line goes on the right application's row.** Postfix's line
about a refusal at `MAIL FROM` was matched to its row by client address,
sender and time; two applications behind one address sending as one
sender could take each other's line. The line carries the login after
all, and the match now uses it. A login read from smtpd's `client=` line
no longer keeps a trailing comma when the client also gave `AUTH=` on
`MAIL FROM`. A refusal written from `mail.log` after a restart says that
Postfix's line names the last recipient only (assumptions A-15, A-21,
A-23).
- **A bounce is judged by the filter lists as the panel counts it.** - **A bounce is judged by the filter lists as the panel counts it.**
rspamd's sender rules took a bounce's `From:` header when the envelope rspamd's sender rules took a bounce's `From:` header when the envelope
sender was empty, so a bounce could be refused or allowed by an entry the sender was empty, so a bounce could be refused or allowed by an entry the
+41 -17
View File
@@ -186,10 +186,15 @@ domain is added or deleted.
One process, five roles (six with the inbound relay on): One process, five roles (six with the inbound relay on):
1. **HTTP server** — `:8080` (`PANEL_HTTP_ADDR`); HTTPS terminated by reverse 1. **HTTP server** — `:8080` (`PANEL_HTTP_ADDR`); HTTPS terminated by reverse
proxy only. On start it runs `postconf -h` once for the deferred-mail retry proxy only. On start it runs `postconf -h -x` once for the deferred-mail retry
parameters (`queue_run_delay`, `minimal_backoff_time`, parameters (`queue_run_delay`, `minimal_backoff_time`,
`maximal_backoff_time`, `maximal_queue_lifetime`, `bounce_queue_lifetime`, `maximal_backoff_time`, `maximal_queue_lifetime`, `bounce_queue_lifetime`,
`delay_warning_time`) and caches the snapshot on the handlers config. The `delay_warning_time`), reads each as Postfix does - one number and one
unit, the parameter's own default unit when there is none (`5d` for the
lifetimes, so a bare `3` is three days), `bounce_queue_lifetime` capped at
`maximal_queue_lifetime` as qmgr caps it (Postfix 3.7.11,
`src/global/conv_time.c`, `src/qmgr/qmgr.c`) - and caches the snapshot on
the handlers config. The
Outbound queue box and a delivery's `deferred` / `bounced` history print those Outbound queue box and a delivery's `deferred` / `bounced` history print those
numbers; they never call `postconf` per request. If `postconf` is missing, numbers; they never call `postconf` per request. If `postconf` is missing,
the panel logs a warning and uses Postfix 3.x compiled-in defaults the panel logs a warning and uses Postfix 3.x compiled-in defaults
@@ -432,8 +437,11 @@ judged); the sender check writes it on the row, and the message's page shows
the lines stored under it - only those of the minutes around the refusal, since the lines stored under it - only those of the minutes around the refusal, since
Postfix frees the id at once. A message the journal milter refused at Postfix frees the id at once. A message the journal milter refused at
`MAIL FROM` has none: the tailer reads smtpd's `NOQUEUE: milter-reject: MAIL `MAIL FROM` has none: the tailer reads smtpd's `NOQUEUE: milter-reject: MAIL
from <client>[<address>]: …; from=<…>` - it names no login - and puts it on from <client>[<address>]: …; from=<…> … sasl_username=<login>` - the login is
the newest refused row of that client address and sender within two minutes Postfix's last field on an authenticated session (Postfix 3.7.11,
`src/smtpd/smtpd.c`, `smtpd_whatsup()`), and a line without one is no
application's - and puts it on the nearest refused row of that client
address, login and sender within two minutes
(`send_log.refusal_line`; the milter writes `send_log.client_ip`), retrying (`send_log.refusal_line`; the milter writes `send_log.client_ip`), retrying
for 30 s when the row is not written yet; a line already on a row goes on no for 30 s when the row is not written yet; a line already on a row goes on no
other. The sender check writes its other. The sender check writes its
@@ -441,13 +449,23 @@ row through a buffer after answering 550; a panel stopped in between wrote
none. The tailer remembers each queue id's login from smtpd's `client=` line, none. The tailer remembers each queue id's login from smtpd's `client=` line,
and 30 s after cleanup's `milter-reject: END-OF-MESSAGE … 5.7.1 From:` line it and 30 s after cleanup's `milter-reject: END-OF-MESSAGE … 5.7.1 From:` line it
writes the row from the line if no refused row of that id exists writes the row from the line if no refused row of that id exists
(`send_log.recovered`, shown on the page). It reads from its saved offset, so (`send_log.recovered`, shown on the page). cleanup's line names one
recipient, the message's last (`src/cleanup/cleanup_milter.c`), so a row
written from it has that one, which its page says. It reads from its saved offset, so
a kill is covered as well as a clean stop. a kill is covered as well as a clean stop.
**Expiry.** A message Postfix gives up on at the end of its queue lifetime has
no line per recipient: qmgr writes one line for the message, `<id>:
from=<...>, status=expired, returned to sender` (`force-expired` after
`postsuper -e`; Postfix 3.7.11, `src/qmgr/qmgr_active.c`,
`qmgr_active_done_25_generic()`), and the recipients' last lines say
`deferred`. The tailer reads that line and closes the message's open rows -
each recipient's newest, `queued` or `deferred` - as `bounced`.
**Queue reconcile** is the backstop for what the log cannot explain at all: a **Queue reconcile** is the backstop for what the log cannot explain at all: a
row still `queued` more than two minutes after it was accepted, whose queue id row still open - `queued`, or `deferred` and waiting - more than two minutes
`postqueue -p` no longer lists, is marked `bounced` with the reason after it was accepted, whose queue id `postqueue -p` no longer lists, is
`reconcile` (swept every five minutes, marked `bounced` with the reason `reconcile` (swept every five minutes,
[internal/logtail](../internal/logtail/logtail.go), [internal/logtail](../internal/logtail/logtail.go),
[postfix.QueueIDs](../internal/postfix/queue.go)). Postfix having dropped the [postfix.QueueIDs](../internal/postfix/queue.go)). Postfix having dropped the
message means nothing more will ever be reported about it, so the row can only message means nothing more will ever be reported about it, so the row can only
@@ -458,8 +476,12 @@ end-of-file once (on a restart the log itself holds the answer); the two-minute
grace covers messages merely in flight; a `postqueue` that cannot be read, or grace covers messages merely in flight; a `postqueue` that cannot be read, or
whose listing is over the 4 MiB cap, leaves every row untouched rather than whose listing is over the 4 MiB cap, leaves every row untouched rather than
closing them all or those past the cut; and the close, by the row's id, is closing them all or those past the cut; and the close, by the row's id, is
conditional on the row still being `queued`, so a status the log wrote in the conditional on the row still being open, so a final status the log wrote in
meantime stands. A delivery line updates only the newest row of its queue id the meantime stands. A deferred row is the sweep's too because Postfix can
drop a deferred message without a final line for its recipients: at expiry
before `2.5.2` read that line, and with `postsuper -d`, which logs only
`removed`. The listing's `#` - a message forced to expire - is read as part
of the id's flags (`src/postqueue/showq_compat.c`, `showq_message()`). A delivery line updates only the newest row of its queue id
and recipient: Postfix reuses a short queue id once the message has left the and recipient: Postfix reuses a short queue id once the message has left the
queue, and an old row must not be rewritten. `UpdateStatus`'s transaction queue, and an old row must not be rewritten. `UpdateStatus`'s transaction
starts with a write, so a commit by the `dmarc-ingest` process in between starts with a write, so a commit by the `dmarc-ingest` process in between
@@ -471,16 +493,18 @@ or deleted — or that the tailer was not reading.
Whether the sweep earns its place is measured, not argued (audit 2026-10-12, Whether the sweep earns its place is measured, not argued (audit 2026-10-12,
H3). `delivery_log_days` (migration `0007`) keeps, per UTC day for a year: H3). `delivery_log_days` (migration `0007`) keeps, per UTC day for a year:
`reconcile_closed`, the rows the sweep closed; `reconcile_missed`, those of `reconcile_closed`, the rows the sweep closed; `reconcile_missed`, those of
them whose delivery line was in `mail.log` or `mail.log.1` when the sweep them whose final line - sent, bounced, or the message's expiry - was in
closed them - the tailer did not read a line that was there, which is what a `mail.log` or `mail.log.1` when the sweep closed them - the tailer did not
lost line looks like (it is never read late); `reconcile_contradicted`, those read a line that was there, which is what a lost line looks like (it is never
read late; a deferred line is not one, a deferred row has it by definition); `reconcile_contradicted`, those
a delivery line read later gave a status - the log's status replaces the a delivery line read later gave a status - the log's status replaces the
sweep's and the row's reason becomes `reconcile-contradicted`, so a row counts sweep's and the row's reason becomes `reconcile-contradicted`, so a row counts
once and can be found; `tailer_blind_seconds`, the time `follow()` could not once and can be found; `tailer_blind_seconds`, the time `follow()` could not
read `mail.log` (the file gone, reopening it failing; stretches under three read `mail.log` (the file gone, reopening it failing; stretches under three
seconds - a rotation's tick between rename and create, which loses nothing - seconds - a rotation's tick between rename and create, which loses nothing -
are not counted); and `queued_over_lifetime`, the rows still `queued` past the are not counted); and `queued_over_lifetime`, the rows still open (`queued`
queue lifetime plus a maximal backoff, as the day's last sweep saw them - rows or `deferred`; `queued` only before `2.5.2`) past the queue lifetime plus a
maximal backoff, as the day's last sweep saw them - rows
stuck although the sweep runs (mail on hold, or a queue that could not be stuck although the sweep runs (mail on hold, or a queue that could not be
listed). What removing the sweep would leave queued is closed − missed − listed). What removing the sweep would leave queued is closed − missed −
contradicted: the rows nothing in the log ever resolved. No page shows the contradicted: the rows nothing in the log ever resolved. No page shows the
@@ -529,8 +553,8 @@ reads them:
- **Which lines are a row's.** Postfix reuses a queue id once the message has - **Which lines are a row's.** Postfix reuses a queue id once the message has
left the queue, and a row outlives that by months. `SendLogLines` gives a row left the queue, and a row outlives that by months. `SendLogLines` gives a row
the lines of its queue id dated from one hour before the row was written to the lines of its queue id dated from one hour before the row was written to
the queue lifetime (the larger of `maximal_queue_lifetime` and the queue lifetime (`maximal_queue_lifetime`, which `bounce_queue_lifetime`
`bounce_queue_lifetime`) plus a day after it. That limits a chance reuse of cannot exceed) plus a day after it. That limits a chance reuse of
the id to those days; it does not exclude it. A message with several the id to those days; it does not exclude it. A message with several
recipients is several rows under one id, and each row's page shows all of recipients is several rows under one id, and each row's page shows all of
the message's lines. the message's lines.
+4 -2
View File
@@ -1268,8 +1268,10 @@ chosen. A row identifies its message and nothing more — time,
sender, recipient, subject and status `queued` (accepted, not yet sender, recipient, subject and status `queued` (accepted, not yet
delivered), `sent` (handed off successfully), `deferred` (Postfix is delivered), `sent` (handed off successfully), `deferred` (Postfix is
retrying), `bounced` (final failure), or refused before it was queued. retrying), `bounced` (final failure), or refused before it was queued.
`bounced` is usually Postfix's word; it is the panel's when Postfix no longer `bounced` is usually Postfix's word - a message it gave up on at the end of
holds a message and `mail.log` never said what became of it (the log rotated its queue lifetime is `bounced` too, since Postfix returns it to the sender;
it is the panel's when Postfix no longer holds a message and `mail.log` never
said what became of it (the log rotated
away while the panel was down, or was not readable) - the panel then closes away while the panel was down, or was not readable) - the panel then closes
the row as a failure rather than show a delivery it cannot evidence, and the row as a failure rather than show a delivery it cannot evidence, and
counts that it did. A counts that it did. A
+50 -19
View File
@@ -30,13 +30,14 @@ import (
// far into mail.log it has read. *store.Store satisfies it. // far into mail.log it has read. *store.Store satisfies it.
type StatusStore interface { type StatusStore interface {
UpdateStatus(queueID, recipient, status string) (int64, error) UpdateStatus(queueID, recipient, status string) (int64, error)
ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) ExpireMessage(queueID string) (int64, error)
ListOpenOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error)
CloseUnreported(id int64) (int64, error) CloseUnreported(id int64) (int64, error)
// The counters of delivery_log_days (store/deliverylog.go). // The counters of delivery_log_days (store/deliverylog.go).
AddReconcileClosed(at time.Time, n int64) error AddReconcileClosed(at time.Time, n int64) error
AddReconcileMissed(at time.Time, n int64) error AddReconcileMissed(at time.Time, n int64) error
AddTailerBlind(from time.Time, d time.Duration) error AddTailerBlind(from time.Time, d time.Duration) error
CountQueuedOlderThan(cutoff time.Time) (int64, error) CountOpenOlderThan(cutoff time.Time) (int64, error)
SetQueuedOverLifetime(at time.Time, n int64) error SetQueuedOverLifetime(at time.Time, n int64) error
PruneDeliveryLogDays(now time.Time) error PruneDeliveryLogDays(now time.Time) error
RefusalStore RefusalStore
@@ -132,15 +133,35 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
status = store.StatusDeferred status = store.StatusDeferred
case "bounced": case "bounced":
status = store.StatusBounced status = store.StatusBounced
case "expired":
// Postfix gave up after the queue lifetime; a final failure for us.
status = store.StatusBounced
default: default:
return "", "", "", false return "", "", "", false
} }
return m[1], m[2], status, true return m[1], m[2], status, true
} }
// expiryRe matches qmgr's line for a message it gave up on:
//
// postfix/qmgr[93]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender
//
// Postfix 3.7.11, src/qmgr/qmgr_active.c, qmgr_active_done_25_generic():
// msg_info("%s: from=<%s>, status=%s, returned to sender") with "expired"
// once the queue lifetime has passed, "force-expired" after postsuper -e.
// It is the only line about the expiry, one per message, and no recipient is
// named (assumptions A-8). The line must end there: the sender is the
// client's text, and only qmgr writes this shape.
var expiryRe = regexp.MustCompile(`^\S+ \S*/qmgr\[\d+\]: ([0-9A-Za-z]+): from=<.*>, status=(?:force-)?expired, returned to sender\s*$`)
// parseExpiry returns the queue id of a message Postfix gave up on, from its
// expiry line; ok is false for any other line.
func parseExpiry(line string) (queueID string, ok bool) {
_, rest := SplitTimestamp(line)
m := expiryRe.FindStringSubmatch(rest)
if m == nil {
return "", false
}
return m[1], true
}
// Run follows path and updates send-log statuses until ctx is cancelled, while // Run follows path and updates send-log statuses until ctx is cancelled, while
// a background sweep prunes rows older than retention(). Reading resumes at // a background sweep prunes rows older than retention(). Reading resumes at
// the offset the previous run persisted, so a restart parses the delivery lines // the offset the previous run persisted, so a restart parses the delivery lines
@@ -199,6 +220,11 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
refusals.handle(line, now, Place{FP: tr.line.FP, End: tr.lineStart}) refusals.handle(line, now, Place{FP: tr.line.FP, End: tr.lineStart})
queueID, recipient, status, ok := parseDelivery(line) queueID, recipient, status, ok := parseDelivery(line)
if !ok { if !ok {
if queueID, ok := parseExpiry(line); ok {
if _, err := st.ExpireMessage(queueID); err != nil {
log.Printf("log-tailer: expire %s: %v", queueID, err)
}
}
return return
} }
if _, err := st.UpdateStatus(queueID, recipient, status); err != nil { if _, err := st.UpdateStatus(queueID, recipient, status); err != nil {
@@ -243,11 +269,11 @@ func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{}
} }
} }
// countStuck records how many rows are still queued past Postfix's queue // countStuck records how many rows are still open past Postfix's queue
// lifetime - what the log would leave unresolved if there were no sweep. Each // lifetime - what the log would leave unresolved if there were no sweep. Each
// sweep overwrites the day's value, so a past day keeps its last sweep's. // sweep overwrites the day's value, so a past day keeps its last sweep's.
func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) { func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) {
n, err := st.CountQueuedOlderThan(now.Add(-overLifetime)) n, err := st.CountOpenOlderThan(now.Add(-overLifetime))
if err == nil { if err == nil {
err = st.SetQueuedOverLifetime(now, n) err = st.SetQueuedOverLifetime(now, n)
} }
@@ -256,10 +282,10 @@ func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) {
} }
} }
// reconcile marks as bounced every row still "queued" from before cutoff whose // reconcile marks as bounced every row still open ("queued" or "deferred")
// message Postfix no longer holds, with reason "reconcile". The close itself // from before cutoff whose message Postfix no longer holds, with reason
// is conditional on the row still being queued, so a delivery line read // "reconcile". The close itself is conditional on the row still being open,
// between the listing and the close wins. // so a final delivery line read between the listing and the close wins.
// //
// A row reaches this state only when its delivery lines are gone for good — the // A row reaches this state only when its delivery lines are gone for good — the
// log rotated past its fourteen files while the panel was down, or was deleted // log rotated past its fourteen files while the panel was down, or was deleted
@@ -279,9 +305,9 @@ func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) {
// is what tells a tailer that loses lines from a log that lost them, which // is what tells a tailer that loses lines from a log that lost them, which
// reconcile_contradicted cannot - a lost line is never read late. // reconcile_contradicted cannot - a lost line is never read late.
func reconcile(st StatusStore, cutoff time.Time, path string) { func reconcile(st StatusStore, cutoff time.Time, path string) {
rows, err := st.ListQueuedOlderThan(cutoff) rows, err := st.ListOpenOlderThan(cutoff)
if err != nil { if err != nil {
log.Printf("log-tailer: reconcile: list queued rows: %v", err) log.Printf("log-tailer: reconcile: list open rows: %v", err)
return return
} }
if len(rows) == 0 { if len(rows) == 0 {
@@ -307,7 +333,7 @@ func reconcile(st StatusStore, cutoff time.Time, path string) {
continue continue
} }
closed += n closed += n
if n > 0 && deliveryInLog(path, row) { if n > 0 && resultInLog(path, row) {
log.Printf("log-tailer: reconcile: %s/%s had a delivery line in the log the tailer did not read", row.QueueID, row.To) log.Printf("log-tailer: reconcile: %s/%s had a delivery line in the log the tailer did not read", row.QueueID, row.To)
missed++ missed++
} }
@@ -323,13 +349,15 @@ func reconcile(st StatusStore, cutoff time.Time, path string) {
} }
} }
// deliveryInLog says whether the current log or the one rotated last holds a // resultInLog says whether the current log or the one rotated last holds a
// delivery line for row. Rows reach the sweep minutes after they were // final result for row - a delivery line that says sent or bounced, or the
// written, so the end of those two files is where such a line would be. // message's expiry line. A deferred line is not one: a deferred row has one
// by definition. Rows reach the sweep minutes after they were written, so
// the end of those two files is where such a line would be.
// logrotate 3.21.0, logrotate.c, prerotateSingleLog(): the first rotated // logrotate 3.21.0, logrotate.c, prerotateSingleLog(): the first rotated
// name is "<log>.1", without the compression suffix under delaycompress // name is "<log>.1", without the compression suffix under delaycompress
// (build/logrotate-mail.conf). // (build/logrotate-mail.conf).
func deliveryInLog(path string, row store.QueuedDelivery) bool { func resultInLog(path string, row store.QueuedDelivery) bool {
if path == "" { if path == "" {
return false return false
} }
@@ -339,7 +367,10 @@ func deliveryInLog(path string, row store.QueuedDelivery) bool {
continue continue
} }
for _, line := range lines { for _, line := range lines {
if q, r, _, ok := parseDelivery(line); ok && q == row.QueueID && strings.EqualFold(r, row.To) { if q, r, s, ok := parseDelivery(line); ok && q == row.QueueID && strings.EqualFold(r, row.To) && s != store.StatusDeferred {
return true
}
if q, ok := parseExpiry(line); ok && q == row.QueueID {
return true return true
} }
} }
+81 -15
View File
@@ -17,10 +17,69 @@ import (
// The mail.log lines in this file are composed by hand, after the formats the // The mail.log lines in this file are composed by hand, after the formats the
// code cites (Postfix 3.7.11, src/global/log_adhoc.c and the others), unless a // code cites (Postfix 3.7.11, src/global/log_adhoc.c and the others), unless a
// comment says where one was seen. Some are shapes Postfix does not write: // comment says where one was seen. Some are shapes Postfix does not write:
// "to=..., status=expired" (qmgr logs expiry once per message as "from=<..>, // ISO stamps (msg_logger.c writes "%b %d %H:%M:%S" - A-12), and opendkim
// status=expired, returned to sender" - assumptions A-8), ISO stamps // lines in mail.log (build/opendkim.conf has Syslog no).
// (msg_logger.c writes "%b %d %H:%M:%S" - A-12), and opendkim lines in // qmgr's expiry line, as Postfix 3.7.11 writes it (src/qmgr/qmgr_active.c,
// mail.log (build/opendkim.conf has Syslog no). // qmgr_active_done_25_generic(): "%s: from=<%s>, status=%s, returned to
// sender"); composed by hand after that format. The sender is the client's
// text, so a line only resembles one when it ends as qmgr's does.
func TestParseExpiry(t *testing.T) {
for _, c := range []struct {
line string
id string
}{
{"Oct 14 09:01:02 relay postfix/qmgr[93]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender", "41E862C00D9E"},
{"Oct 14 09:01:02 relay postfix/qmgr[93]: 3Fq1Lw0bZ8z9tN: from=<>, status=force-expired, returned to sender", "3Fq1Lw0bZ8z9tN"},
// Not qmgr.
{"Oct 14 09:01:02 relay postfix/smtp[26]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender", ""},
// qmgr's ordinary from= line, whose sender carries the text.
{"Oct 14 09:01:02 relay postfix/qmgr[93]: 41E862C00D9E: from=<x>, status=expired, returned to sender>, size=500, nrcpt=1 (queue active)", ""},
{"Oct 14 09:01:02 relay postfix/qmgr[93]: 41E862C00D9E: removed", ""},
} {
id, ok := parseExpiry(c.line)
if ok != (c.id != "") || id != c.id {
t.Errorf("parseExpiry(%q) = %q, %v; want %q", c.line, id, ok, c.id)
}
}
}
// The tailer closes an expired message's rows when it reads qmgr's line.
func TestExpiryLineClosesTheMessage(t *testing.T) {
path := filepath.Join(t.TempDir(), "mail.log")
if err := os.WriteFile(path, nil, 0o644); err != nil {
t.Fatal(err)
}
cs := &captureStore{}
stop := startRun(t, path, cs)
defer stop()
appendLine(t, path, "Oct 14 09:01:00 relay postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=none, delay=9, dsn=4.4.1, status=deferred (connect timed out)")
appendLine(t, path, "Oct 14 09:01:02 relay postfix/qmgr[93]: 41E862C00D9E: from=<app@example.org>, status=expired, returned to sender")
waitFor(t, func() bool { return contains(cs.snapshot(), "41E862C00D9E|*|expired") })
if !contains(cs.snapshot(), "41E862C00D9E|a@example.net|"+store.StatusDeferred) {
t.Fatalf("calls = %v, want the deferral before the expiry", cs.snapshot())
}
}
// A deferred row the sweep closes is not "missed" for its deferred line - it
// has one by definition - but is for its message's expiry line.
func TestReconcileMissedCountsOnlyFinalLines(t *testing.T) {
stubQueue(t, nil)
path := filepath.Join(t.TempDir(), "mail.log")
if err := os.WriteFile(path, []byte(
"Oct 14 10:59:58 mail postfix/smtp[26]: DEFQ: to=<late@example.net>, relay=none, delay=9, dsn=4.4.1, status=deferred (connect timed out)\n"+
"Oct 14 10:59:59 mail postfix/qmgr[10]: EXPQ: from=<app@example.org>, status=expired, returned to sender\n"), 0o644); err != nil {
t.Fatal(err)
}
cs := &captureStore{queued: []store.QueuedDelivery{
{ID: 1, QueueID: "DEFQ", To: "late@example.net"},
{ID: 2, QueueID: "EXPQ", To: "gone@example.net"},
}}
reconcile(cs, time.Now().UTC(), path)
if cs.closed != 2 || cs.missed != 1 {
t.Fatalf("closed %d, missed %d; want 2 and 1", cs.closed, cs.missed)
}
}
func TestParseDelivery(t *testing.T) { func TestParseDelivery(t *testing.T) {
cases := []struct { cases := []struct {
name string name string
@@ -53,12 +112,11 @@ func TestParseDelivery(t *testing.T) {
status: store.StatusBounced, status: store.StatusBounced,
}, },
{ {
name: "expired maps to bounced", // Expiry is not a delivery line (TestParseExpiry); qmgr's line
line: "host postfix/smtp[26]: DEF: to=<slow@example.net>, relay=none, status=expired (delivery temporarily suspended)", // for it names no recipient.
wantOK: true, name: "qmgr expiry line ignored",
queueID: "DEF", line: "host postfix/qmgr[10]: DEF: from=<app@example.org>, status=expired, returned to sender",
recipient: "slow@example.net", wantOK: false,
status: store.StatusBounced,
}, },
{ {
name: "qmgr from-line ignored", name: "qmgr from-line ignored",
@@ -189,10 +247,10 @@ type captureStore struct {
haveState bool haveState bool
stateErr error stateErr error
// queued is what ListQueuedOlderThan returns, for the reconcile sweep. // queued is what ListOpenOlderThan returns, for the reconcile sweep.
queued []store.QueuedDelivery queued []store.QueuedDelivery
queuedErr error queuedErr error
// closed sums AddReconcileClosed; stuckCount is what CountQueuedOlderThan // closed sums AddReconcileClosed; stuckCount is what CountOpenOlderThan
// answers and stuck what SetQueuedOverLifetime was given. // answers and stuck what SetQueuedOverLifetime was given.
closed int64 closed int64
missed int64 missed int64
@@ -259,6 +317,14 @@ func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, e
return 1, nil return 1, nil
} }
// ExpireMessage records an expiry line's close as "queue-id|*|expired".
func (c *captureStore) ExpireMessage(queueID string) (int64, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.calls = append(c.calls, queueID+"|*|expired")
return 1, nil
}
// CloseUnreported records the sweep's close as "queue-id|recipient|reconcile", // CloseUnreported records the sweep's close as "queue-id|recipient|reconcile",
// finding the row by its id among queued. // finding the row by its id among queued.
func (c *captureStore) CloseUnreported(id int64) (int64, error) { func (c *captureStore) CloseUnreported(id int64) (int64, error) {
@@ -290,11 +356,11 @@ func (c *captureStore) AddReconcileMissed(_ time.Time, n int64) error {
return nil return nil
} }
func (c *captureStore) AddTailerBlind(time.Time, time.Duration) error { return nil } func (c *captureStore) AddTailerBlind(time.Time, time.Duration) error { return nil }
func (c *captureStore) AttachRefusalLine(string, string, time.Time, string) (bool, error) { func (c *captureStore) AttachRefusalLine(string, string, string, time.Time, string) (bool, error) {
return true, nil return true, nil
} }
func (c *captureStore) RecoverRefusal(store.SendLogEntry, time.Time) (bool, error) { return false, nil } func (c *captureStore) RecoverRefusal(store.SendLogEntry, time.Time) (bool, error) { return false, nil }
func (c *captureStore) CountQueuedOlderThan(time.Time) (int64, error) { func (c *captureStore) CountOpenOlderThan(time.Time) (int64, error) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
return c.stuckCount, nil return c.stuckCount, nil
@@ -307,7 +373,7 @@ func (c *captureStore) SetQueuedOverLifetime(_ time.Time, n int64) error {
} }
func (c *captureStore) PruneDeliveryLogDays(time.Time) error { return nil } func (c *captureStore) PruneDeliveryLogDays(time.Time) error { return nil }
func (c *captureStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDelivery, error) { func (c *captureStore) ListOpenOlderThan(time.Time) ([]store.QueuedDelivery, error) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
return c.queued, c.queuedErr return c.queued, c.queuedErr
+28 -20
View File
@@ -14,20 +14,24 @@ import (
// level-2 limit, a client address off the application's list - is put on the // level-2 limit, a client address off the application's list - is put on the
// row the milter wrote, so the message's page shows it (audit 2026-10-12, // row the milter wrote, so the message's page shows it (audit 2026-10-12,
// stage 2b). The refusal has no queue id; it is matched to its row by client // stage 2b). The refusal has no queue id; it is matched to its row by client
// address, sender and time. As the stand writes it (e2e stand, date not // address, login, sender and time:
// recorded):
// //
// postfix/smtpd[1149]: NOQUEUE: milter-reject: MAIL from unknown[10.77.0.1]: 451 4.7.1 Service unavailable - try again later; from=<billing@sender.e2e.test> proto=ESMTP helo=<e2e-client.e2e.test> // postfix/smtpd[1149]: NOQUEUE: milter-reject: MAIL from unknown[10.77.0.1]: 451 4.7.1 Service unavailable - try again later; from=<billing@sender.e2e.test> proto=ESMTP helo=<e2e-client.e2e.test> sasl_username=<l2app>
// //
// Postfix 3.7.11, src/smtpd/smtpd.c, check_milter_reply(): "%s: %s: %s from // Postfix 3.7.11, src/smtpd/smtpd.c, check_milter_reply(): "%s: %s: %s from
// %s: %s;%s" with "NOQUEUE" before the queue file is opened, and // %s: %s;%s" with "NOQUEUE" before the queue file is opened, and
// smtpd_whatsup() for the rest - " from=<..>" first. smtpd_whatsup() also // smtpd_whatsup() for the rest: " from=<..>", " to=<..>", " proto=..",
// ends the line with " sasl_username=<login>" on an authenticated session, // " helo=<..>", then " sasl_username=<..>" when the session authenticated
// which the quote above lacks and the regexp does not use (assumptions A-20, // (assumptions A-20, A-21). The login is read at the end of the line, where
// A-21). // Postfix puts it last; a line without one is no application's refusal - the
// journal milter refuses at MAIL FROM only an authenticated session
// (milter/ratelimit.go). The HELO name before it is the client's text and can
// end in what looks like the field, but only in a session that has no login,
// and so no refused row to take the line; a row must also be of the line's
// client address and sender, within minutes.
// //
// Port 25's smtpd (postfix/inbound/smtpd) is a stranger's and has no such row. // Port 25's smtpd (postfix/inbound/smtpd) is a stranger's and has no such row.
var refusalLineRe = regexp.MustCompile(`^\S+ postfix/(?:([a-z0-9-]+)/)?smtpd\[\d+\]: NOQUEUE: milter-reject: MAIL from [^\[\s]*\[([^\]]+)\]: [45]\d\d .*?; from=<([^>]*)>`) var refusalLineRe = regexp.MustCompile(`^\S+ postfix/(?:([a-z0-9-]+)/)?smtpd\[\d+\]: NOQUEUE: milter-reject: MAIL from [^\[\s]*\[([^\]]+)\]: [45]\d\d .*?; from=<([^>]*)>(?:.* sasl_username=<([^<>]*)>|.*)$`)
// refusalLineWait is how long a line waits for its row: the milter writes the // refusalLineWait is how long a line waits for its row: the milter writes the
// row when it refuses, which can be read a moment after the line. // row when it refuses, which can be read a moment after the line.
@@ -35,7 +39,7 @@ const refusalLineWait = 30 * time.Second
// RefusalStore is where the lines go. *store.Store satisfies it. // RefusalStore is where the lines go. *store.Store satisfies it.
type RefusalStore interface { type RefusalStore interface {
AttachRefusalLine(clientIP, from string, at time.Time, line string) (bool, error) AttachRefusalLine(clientIP, login, from string, at time.Time, line string) (bool, error)
RecoverRefusal(e store.SendLogEntry, at time.Time) (bool, error) RecoverRefusal(e store.SendLogEntry, at time.Time) (bool, error)
} }
@@ -56,18 +60,19 @@ var fromRefusalRe = regexp.MustCompile(`^\S+ postfix/(?:[a-z0-9-]+/)?cleanup\[\d
// clientLoginRe is smtpd's line naming the login a queue id was submitted // clientLoginRe is smtpd's line naming the login a queue id was submitted
// under: "<id>: client=host[addr], sasl_method=PLAIN, sasl_username=<login>". // under: "<id>: client=host[addr], sasl_method=PLAIN, sasl_username=<login>".
// Postfix 3.7.11, src/smtpd/smtpd.c, mail_open_stream(): ", sasl_sender=<..>" // Postfix 3.7.11, src/smtpd/smtpd.c, mail_open_stream(): ", sasl_sender=.."
// follows when the client gave AUTH= on MAIL FROM, and "\S+" then takes // follows when the client gave AUTH= on MAIL FROM, then ", orig_queue_id=.."
// "login," with the comma (assumptions A-23). // and ", orig_client=.." for XFORWARD; the login ends at the comma (a login
var clientLoginRe = regexp.MustCompile(`^\S+ postfix/(?:([a-z0-9-]+)/)?smtpd\[\d+\]: ([0-9A-Za-z]+): client=.*?, sasl_username=(\S+)`) // holds none - internal/postfix assertMapSafe; assumptions A-23).
var clientLoginRe = regexp.MustCompile(`^\S+ postfix/(?:([a-z0-9-]+)/)?smtpd\[\d+\]: ([0-9A-Za-z]+): client=.*?, sasl_username=([^,\s]+)`)
// refusalLogins bounds the queue ids whose login is remembered. // refusalLogins bounds the queue ids whose login is remembered.
const refusalLogins = 4096 const refusalLogins = 4096
type pendingRefusal struct { type pendingRefusal struct {
clientIP, from, line string clientIP, login, from, line string
at, until time.Time at, until time.Time
start Place // where the line starts in the log start Place // where the line starts in the log
} }
// pendingRecovery is a From: refusal whose row is looked for once the panel // pendingRecovery is a From: refusal whose row is looked for once the panel
@@ -139,14 +144,17 @@ func (r *refusalLines) handle(line string, now time.Time, start Place) {
return return
} }
m := refusalLineRe.FindStringSubmatch(rest) m := refusalLineRe.FindStringSubmatch(rest)
if m == nil || m[1] == "inbound" { if m == nil || m[1] == "inbound" || m[4] == "" {
return return
} }
ip := net.ParseIP(strings.TrimPrefix(m[2], "IPv6:")) // The address in "name[addr]" is smtpd's state->addr, which has no
// "IPv6:" (Postfix 3.7.11, src/smtpd/smtpd_peer.c, smtpd_peer_init();
// assumptions A-24).
ip := net.ParseIP(m[2])
if ip == nil { if ip == nil {
return return
} }
p := pendingRefusal{clientIP: ip.String(), from: m[3], line: line, at: at, until: now.Add(refusalLineWait), start: start} p := pendingRefusal{clientIP: ip.String(), login: m[4], from: m[3], line: line, at: at, until: now.Add(refusalLineWait), start: start}
if !r.attach(p) { if !r.attach(p) {
r.pending = append(r.pending, p) r.pending = append(r.pending, p)
} }
@@ -206,7 +214,7 @@ func domainOf(addr string) string {
} }
func (r *refusalLines) attach(p pendingRefusal) bool { func (r *refusalLines) attach(p pendingRefusal) bool {
ok, err := r.st.AttachRefusalLine(p.clientIP, p.from, p.at, p.line) ok, err := r.st.AttachRefusalLine(p.clientIP, p.login, p.from, p.at, p.line)
if err != nil { if err != nil {
log.Printf("log-tailer: refusal line: %v", err) log.Printf("log-tailer: refusal line: %v", err)
return true // not retried: the store said no for its own reason return true // not retried: the store said no for its own reason
+6 -3
View File
@@ -19,13 +19,13 @@ func (r *refusalProbe) RecoverRefusal(e store.SendLogEntry, _ time.Time) (bool,
return true, nil return true, nil
} }
func (r *refusalProbe) AttachRefusalLine(clientIP, from string, _ time.Time, line string) (bool, error) { func (r *refusalProbe) AttachRefusalLine(clientIP, login, from string, _ time.Time, line string) (bool, error) {
if r.rows == 0 { if r.rows == 0 {
return false, nil return false, nil
} }
r.rows-- r.rows--
r.lines = append(r.lines, line) r.lines = append(r.lines, line)
r.who = append(r.who, clientIP+"|"+from) r.who = append(r.who, clientIP+"|"+login+"|"+from)
return true, nil return true, nil
} }
@@ -46,6 +46,9 @@ func TestRefusalLines(t *testing.T) {
`2026-10-14T12:00:00+00:00 mail postfix/inbound/smtpd[372]: NOQUEUE: milter-reject: MAIL from unknown[203.0.113.7]: 451 4.7.1 Service unavailable - try again later; from=<a@stranger.example> proto=ESMTP helo=<x>`, `2026-10-14T12:00:00+00:00 mail postfix/inbound/smtpd[372]: NOQUEUE: milter-reject: MAIL from unknown[203.0.113.7]: 451 4.7.1 Service unavailable - try again later; from=<a@stranger.example> proto=ESMTP helo=<x>`,
`2026-10-14T12:00:00+00:00 mail postfix/smtpd[11]: NOQUEUE: milter-reject: CONNECT from unknown[10.77.0.1]: 451 4.7.1 Service unavailable - try again later; proto=SMTP`, `2026-10-14T12:00:00+00:00 mail postfix/smtpd[11]: NOQUEUE: milter-reject: CONNECT from unknown[10.77.0.1]: 451 4.7.1 Service unavailable - try again later; proto=SMTP`,
`2026-10-14T12:00:00+00:00 mail postfix/smtpd[11]: NOQUEUE: reject: RCPT from unknown[10.77.0.1]: 553 5.7.1 <a@x>: Sender address rejected: not owned by user app1; from=<a@x> to=<b@y> proto=ESMTP helo=<x>`, `2026-10-14T12:00:00+00:00 mail postfix/smtpd[11]: NOQUEUE: reject: RCPT from unknown[10.77.0.1]: 553 5.7.1 <a@x>: Sender address rejected: not owned by user app1; from=<a@x> to=<b@y> proto=ESMTP helo=<x>`,
// No login: no application's refusal (the panel down refuses every
// session, an unauthenticated one too).
`2026-10-14T12:00:00+00:00 mail postfix/smtpd[12]: NOQUEUE: milter-reject: MAIL from unknown[10.77.0.1]: 451 4.7.1 Service unavailable - try again later; from=<billing@sender.e2e.test> proto=ESMTP helo=<x>`,
} { } {
r.handle(other, now, Place{}) r.handle(other, now, Place{})
} }
@@ -59,7 +62,7 @@ func TestRefusalLines(t *testing.T) {
} }
st.rows = 1 // the milter's row is written st.rows = 1 // the milter's row is written
r.retry(now.Add(time.Second)) r.retry(now.Add(time.Second))
if len(r.pending) != 0 || len(st.lines) != 1 || st.who[0] != "10.77.0.1|billing@sender.e2e.test" { if len(r.pending) != 0 || len(st.lines) != 1 || st.who[0] != "10.77.0.1|l2app|billing@sender.e2e.test" {
t.Fatalf("after the retry: pending %d, attached %v", len(r.pending), st.who) t.Fatalf("after the retry: pending %d, attached %v", len(r.pending), st.who)
} }
+5 -3
View File
@@ -19,10 +19,12 @@ func (s *retentionProbeStore) UpdateStatus(string, string, string) (int64, error
return 0, nil return 0, nil
} }
func (s *retentionProbeStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDelivery, error) { func (s *retentionProbeStore) ListOpenOlderThan(time.Time) ([]store.QueuedDelivery, error) {
return nil, nil return nil, nil
} }
func (s *retentionProbeStore) ExpireMessage(string) (int64, error) { return 0, nil }
func (s *retentionProbeStore) CloseUnreported(int64) (int64, error) { func (s *retentionProbeStore) CloseUnreported(int64) (int64, error) {
return 0, nil return 0, nil
} }
@@ -30,7 +32,7 @@ func (s *retentionProbeStore) CloseUnreported(int64) (int64, error) {
func (s *retentionProbeStore) AddReconcileMissed(time.Time, int64) error { return nil } func (s *retentionProbeStore) AddReconcileMissed(time.Time, int64) error { return nil }
func (s *retentionProbeStore) AddReconcileClosed(time.Time, int64) error { return nil } func (s *retentionProbeStore) AddReconcileClosed(time.Time, int64) error { return nil }
func (s *retentionProbeStore) AddTailerBlind(time.Time, time.Duration) error { return nil } func (s *retentionProbeStore) AddTailerBlind(time.Time, time.Duration) error { return nil }
func (s *retentionProbeStore) CountQueuedOlderThan(time.Time) (int64, error) { return 0, nil } func (s *retentionProbeStore) CountOpenOlderThan(time.Time) (int64, error) { return 0, nil }
func (s *retentionProbeStore) SetQueuedOverLifetime(time.Time, int64) error { return nil } func (s *retentionProbeStore) SetQueuedOverLifetime(time.Time, int64) error { return nil }
func (s *retentionProbeStore) PruneDeliveryLogDays(time.Time) error { return nil } func (s *retentionProbeStore) PruneDeliveryLogDays(time.Time) error { return nil }
@@ -99,7 +101,7 @@ func TestRetentionLoopUsesUpdatedValue(t *testing.T) {
t.Fatalf("second prune cutoff age %v, want about 7 days", age) t.Fatalf("second prune cutoff age %v, want about 7 days", age)
} }
func (s *retentionProbeStore) AttachRefusalLine(string, string, time.Time, string) (bool, error) { func (s *retentionProbeStore) AttachRefusalLine(string, string, string, time.Time, string) (bool, error) {
return true, nil return true, nil
} }
+59 -64
View File
@@ -2,85 +2,80 @@ package postfix
import ( import (
"fmt" "fmt"
"math"
"strconv" "strconv"
"strings" "strings"
"time" "time"
"unicode"
) )
// ParseDuration converts a Postfix time value to a duration. Units are those // ParseTime reads a Postfix time value as Postfix 3.7.11 reads it:
// postconf prints and accepts: s, m, h, d, w. A bare number is seconds. Values // src/global/conv_time.c, conv_time(), called by get_mail_conf_time() in
// may concatenate units (`1h7m`), matching Postfix's own conv_time. // src/global/mail_conf_time.c. One number - strtol(), so leading blanks and a
func ParseDuration(s string) (time.Duration, error) { // "+" are taken, a negative number or one past an int is invalid - followed
s = strings.TrimSpace(s) // by one unit, w d h m or s in lower case; with no unit, defUnit, which
// Postfix takes from the parameter's default value (get_def_time_unit(): "5d"
// gives d, "300s" s, "0h" h). Whatever follows the unit is ignored, so "1h7m"
// is one hour (assumptions B-PQ10, B-PQ11). A string with no digits is read as
// 0 of the unit it starts with, as strtol() leaves it there.
func ParseTime(s string, defUnit byte) (time.Duration, error) {
if s == "" { if s == "" {
return 0, fmt.Errorf("postfix: empty time value") return 0, fmt.Errorf("postfix: empty time value")
} }
var total time.Duration
i := 0 i := 0
for i < len(s) { for i < len(s) && isCSpace(s[i]) {
for i < len(s) && unicode.IsSpace(rune(s[i])) { i++
i++
}
if i >= len(s) {
break
}
if s[i] == '-' {
return 0, fmt.Errorf("postfix: negative time value %q", s)
}
if s[i] == '+' {
i++
}
start := i
for i < len(s) && s[i] >= '0' && s[i] <= '9' {
i++
}
if i == start {
return 0, fmt.Errorf("postfix: invalid time value %q", s)
}
n, err := strconv.ParseInt(s[start:i], 10, 64)
if err != nil {
return 0, fmt.Errorf("postfix: invalid time value %q", s)
}
unit := byte('s')
if i < len(s) && isTimeUnit(s[i]) {
unit = s[i] | 0x20 // ASCII fold to lowercase
i++
}
part, err := durationForUnit(n, unit)
if err != nil {
return 0, err
}
total += part
} }
return total, nil neg := false
} if i < len(s) && (s[i] == '+' || s[i] == '-') {
neg = s[i] == '-'
func isTimeUnit(c byte) bool { i++
switch c | 0x20 {
case 's', 'm', 'h', 'd', 'w':
return true
} }
return false start := i
} var n int64
for i < len(s) && s[i] >= '0' && s[i] <= '9' {
func durationForUnit(n int64, unit byte) (time.Duration, error) { n = n*10 + int64(s[i]-'0')
var unitDur time.Duration if n > math.MaxInt32 {
return 0, fmt.Errorf("postfix: time value %q out of range", s)
}
i++
}
end := i
if i == start {
// strtol() converted nothing: the end is the start of the string.
end, n, neg = 0, 0, false
}
if neg && n > 0 {
return 0, fmt.Errorf("postfix: negative time value %q", s)
}
unit := defUnit
if end < len(s) {
unit = s[end]
}
var per int64
switch unit { switch unit {
case 's':
unitDur = time.Second
case 'm':
unitDur = time.Minute
case 'h':
unitDur = time.Hour
case 'd':
unitDur = 24 * time.Hour
case 'w': case 'w':
unitDur = 7 * 24 * time.Hour per = 7 * 24 * 3600
case 'd':
per = 24 * 3600
case 'h':
per = 3600
case 'm':
per = 60
case 's':
per = 1
default: default:
return 0, fmt.Errorf("postfix: unknown time unit %q", string(unit)) return 0, fmt.Errorf("postfix: invalid time value %q", s)
} }
return time.Duration(n) * unitDur, nil // conv_time(): "if (intval < INT_MAX / WEEK)" and the like.
if per > 1 && n >= math.MaxInt32/per {
return 0, fmt.Errorf("postfix: time value %q out of range", s)
}
return time.Duration(n*per) * time.Second, nil
}
// isCSpace is C's isspace() in the "C" locale, which strtol() skips.
func isCSpace(c byte) bool {
return c == ' ' || c == '\t' || c == '\n' || c == '\v' || c == '\f' || c == '\r'
} }
// FormatDuration renders a Postfix interval the way the Mail queue card and // FormatDuration renders a Postfix interval the way the Mail queue card and
+27 -19
View File
@@ -5,39 +5,47 @@ import (
"time" "time"
) )
func TestParseDuration(t *testing.T) { // Postfix 3.7.11, src/global/conv_time.c, conv_time(): one number, one
// lower-case unit, the default unit when there is none, the rest ignored.
func TestParseTime(t *testing.T) {
cases := []struct { cases := []struct {
in string in string
unit byte
want time.Duration want time.Duration
}{ }{
{"5d", 5 * 24 * time.Hour}, {"5d", 's', 5 * 24 * time.Hour},
{"300s", 300 * time.Second}, {"300s", 'd', 300 * time.Second},
{"4000s", 4000 * time.Second}, {"4000s", 's', 4000 * time.Second},
{"1h", time.Hour}, {"1h", 's', time.Hour},
{"0", 0}, {"0", 's', 0},
{"0s", 0}, {"0s", 's', 0},
{"300", 300 * time.Second}, {"300", 's', 300 * time.Second},
{" 300s ", 300 * time.Second}, {"3", 'd', 3 * 24 * time.Hour}, // maximal_queue_lifetime = 3
{"1h7m", time.Hour + 7*time.Minute}, {"2", 'h', 2 * time.Hour}, // delay_warning_time = 2
{"1w", 7 * 24 * time.Hour}, {" 300s ", 's', 300 * time.Second},
{"2m", 2 * time.Minute}, {"+5m", 's', 5 * time.Minute},
{"1h7m", 's', time.Hour}, // what follows the unit is ignored
{"1h 5x", 's', time.Hour},
{"1w", 's', 7 * 24 * time.Hour},
{"2m", 's', 2 * time.Minute},
{"s", 'd', 0}, // strtol() reads nothing and stops at the "s"
} }
for _, tc := range cases { for _, tc := range cases {
got, err := ParseDuration(tc.in) got, err := ParseTime(tc.in, tc.unit)
if err != nil { if err != nil {
t.Errorf("ParseDuration(%q): %v", tc.in, err) t.Errorf("ParseTime(%q, %c): %v", tc.in, tc.unit, err)
continue continue
} }
if got != tc.want { if got != tc.want {
t.Errorf("ParseDuration(%q) = %v, want %v", tc.in, got, tc.want) t.Errorf("ParseTime(%q, %c) = %v, want %v", tc.in, tc.unit, got, tc.want)
} }
} }
} }
func TestParseDurationRejectsInvalid(t *testing.T) { func TestParseTimeRejectsInvalid(t *testing.T) {
for _, in := range []string{"", "foo", "5x", "-300s", "s", "1h 5x"} { for _, in := range []string{"", "foo", "5x", "-300s", "5D", "1H", " s", "99999999999s", "400000w"} {
if _, err := ParseDuration(in); err == nil { if _, err := ParseTime(in, 's'); err == nil {
t.Errorf("ParseDuration(%q) = nil, want error", in) t.Errorf("ParseTime(%q) = nil, want error", in)
} }
} }
} }
+5 -2
View File
@@ -125,12 +125,15 @@ func QueueIDs() (map[string]struct{}, error) {
// 3C5B04E6C1* 446 Thu Aug 7 10:12:31 app@example.com // 3C5B04E6C1* 446 Thu Aug 7 10:12:31 app@example.com
// //
// The id is at the start of the line, optionally flagged '*' (in the active // The id is at the start of the line, optionally flagged '*' (in the active
// queue) or '!' (on hold), and is followed by the message size. Requiring the // queue) or '!' (on hold), then '#' when the message was forced to expire
// (Postfix 3.7.11, src/postqueue/showq_compat.c, showq_message(): the flags
// are appended to the id in that order; assumptions B-PQ1), and is followed
// by the message size. Requiring the
// size is what separates an entry from the listing's other left-margin lines: // size is what separates an entry from the listing's other left-margin lines:
// the '-Queue ID-' header, the '-- 5 Kbytes in 2 Requests.' trailer, a deferred // the '-Queue ID-' header, the '-- 5 Kbytes in 2 Requests.' trailer, a deferred
// entry's '(connect timed out)' reason, and 'Mail queue is empty'. Recipient // entry's '(connect timed out)' reason, and 'Mail queue is empty'. Recipient
// lines are indented and never match. // lines are indented and never match.
var queueEntryRe = regexp.MustCompile(`^([0-9A-Za-z]+)[*!]?\s+\d+\s`) var queueEntryRe = regexp.MustCompile(`^([0-9A-Za-z]+)[*!]?#?\s+\d+\s`)
func parseQueueIDs(listing string) map[string]struct{} { func parseQueueIDs(listing string) map[string]struct{} {
ids := make(map[string]struct{}) ids := make(map[string]struct{})
+8 -2
View File
@@ -92,10 +92,16 @@ func TestParseQueueIDs(t *testing.T) {
A1B2C3D4E5F! 891 Fri Aug 8 10:14:31 app@example.com A1B2C3D4E5F! 891 Fri Aug 8 10:14:31 app@example.com
held@example.net held@example.net
-- 1 Kbytes in 3 Requests. 6D7E8F9A0B*# 512 Fri Aug 8 10:15:31 app@example.com
expiring@example.net
7E8F9A0B1C# 512 Fri Aug 8 10:16:31 app@example.com
expiring2@example.net
-- 2 Kbytes in 5 Requests.
` `
ids := parseQueueIDs(listing) ids := parseQueueIDs(listing)
want := []string{"3C5B04E6C1", "5B4A2C1D3E", "A1B2C3D4E5F"} want := []string{"3C5B04E6C1", "5B4A2C1D3E", "A1B2C3D4E5F", "6D7E8F9A0B", "7E8F9A0B1C"}
for _, id := range want { for _, id := range want {
if _, ok := ids[id]; !ok { if _, ok := ids[id]; !ok {
t.Errorf("queue id %s not found in %v", id, ids) t.Errorf("queue id %s not found in %v", id, ids)
+21 -9
View File
@@ -55,6 +55,13 @@ var retryConfKeys = []string{
"delay_warning_time", "delay_warning_time",
} }
// retryConfUnits is each key's default unit: the last letter of its default
// value (Postfix 3.7.11, src/global/mail_params.h, DEF_QUEUE_RUN_DELAY
// "300s", DEF_MIN_BACKOFF_TIME, DEF_MAX_BACKOFF_TIME "4000s",
// DEF_MAX_QUEUE_TIME "5d", DEF_DSN_QUEUE_TIME "5d", DEF_DELAY_WARN_TIME
// "0h"; src/global/mail_conf_time.c, get_def_time_unit()).
var retryConfUnits = []byte{'s', 's', 's', 'd', 'd', 'h'}
// readRetryConf runs `postconf -h` for the retry-policy keys. Tests replace it // readRetryConf runs `postconf -h` for the retry-policy keys. Tests replace it
// the same way logtail stubs queueIDs. // the same way logtail stubs queueIDs.
var readRetryConf = postconfRetryValues var readRetryConf = postconfRetryValues
@@ -63,8 +70,9 @@ func postconfRetryValues() ([]string, error) {
// Fixed argv, no user input — same pattern as Queue (security.md). // Fixed argv, no user input — same pattern as Queue (security.md).
// Postfix 3.7.11, src/postconf/postconf_main.c, pcf_show_parameters(): // Postfix 3.7.11, src/postconf/postconf_main.c, pcf_show_parameters():
// named parameters are printed in argument order, with -h one "%s\n" // named parameters are printed in argument order, with -h one "%s\n"
// value each, as written in main.cf (not expanded without -x). // value each; -x expands "$name" as Postfix itself does when it reads
cmd := exec.Command("postconf", "-h", // the value (mail_conf_eval()).
cmd := exec.Command("postconf", "-h", "-x",
"queue_run_delay", "queue_run_delay",
"minimal_backoff_time", "minimal_backoff_time",
"maximal_backoff_time", "maximal_backoff_time",
@@ -109,12 +117,18 @@ func parseRetryPolicy(lines []string) (RetryPolicy, error) {
} }
var durs [6]time.Duration var durs [6]time.Duration
for i, line := range lines { for i, line := range lines {
d, err := ParseDuration(line) d, err := ParseTime(line, retryConfUnits[i])
if err != nil { if err != nil {
return RetryPolicy{}, fmt.Errorf("%s: %w", retryConfKeys[i], err) return RetryPolicy{}, fmt.Errorf("%s: %w", retryConfKeys[i], err)
} }
durs[i] = d durs[i] = d
} }
// Postfix 3.7.11, src/qmgr/qmgr.c, qmgr_post_init(): "bounce_queue_lifetime
// is larger than maximal_queue_lifetime - adjusting" - qmgr uses the
// smaller.
if durs[4] > durs[3] {
durs[4] = durs[3]
}
return RetryPolicy{ return RetryPolicy{
QueueRunDelay: durs[0], QueueRunDelay: durs[0],
MinimalBackoff: durs[1], MinimalBackoff: durs[1],
@@ -141,13 +155,11 @@ func (p RetryPolicy) BackoffCap() string {
// sender, which an application may submit too. It bounds which stored log // sender, which an application may submit too. It bounds which stored log
// lines can be about a send-log row (store.SendLogLines). // lines can be about a send-log row (store.SendLogLines).
// //
// Postfix 3.7.11, src/qmgr/qmgr_active.c, qmgr_active_done(): a message // Postfix 3.7.11, src/qmgr/qmgr_active.c, qmgr_active_done_25_generic(): a
// expires after var_max_queue_time, or var_dsn_queue_time when its sender is // message expires after var_max_queue_time, or var_dsn_queue_time when its
// null - whoever submitted it (qmgr.c caps the second at the first). // sender is null - whoever submitted it; qmgr.c caps the second at the
// first, and so does parseRetryPolicy, which leaves the first the longer.
func (p RetryPolicy) LongestLifetime() time.Duration { func (p RetryPolicy) LongestLifetime() time.Duration {
if p.BounceQueueLifetime > p.MaximalQueueLifetime {
return p.BounceQueueLifetime
}
return p.MaximalQueueLifetime return p.MaximalQueueLifetime
} }
+17 -2
View File
@@ -29,14 +29,29 @@ func TestLoadRetryPolicyUsesPostconfValues(t *testing.T) {
if p.MaximalQueueLifetime != 2*24*time.Hour { if p.MaximalQueueLifetime != 2*24*time.Hour {
t.Errorf("MaximalQueueLifetime = %v, want 2d", p.MaximalQueueLifetime) t.Errorf("MaximalQueueLifetime = %v, want 2d", p.MaximalQueueLifetime)
} }
if p.BounceQueueLifetime != 5*24*time.Hour { // 5d over a maximal 2d: qmgr uses 2d (Postfix 3.7.11, src/qmgr/qmgr.c,
t.Errorf("BounceQueueLifetime = %v, want 5d", p.BounceQueueLifetime) // qmgr_post_init(), "adjusting bounce_queue_lifetime").
if p.BounceQueueLifetime != 2*24*time.Hour {
t.Errorf("BounceQueueLifetime = %v, want 2d", p.BounceQueueLifetime)
} }
if p.DelayWarningTime != 0 { if p.DelayWarningTime != 0 {
t.Errorf("DelayWarningTime = %v, want 0", p.DelayWarningTime) t.Errorf("DelayWarningTime = %v, want 0", p.DelayWarningTime)
} }
} }
// A message lives at most maximal_queue_lifetime: qmgr caps
// bounce_queue_lifetime at it (Postfix 3.7.11, src/qmgr/qmgr.c,
// qmgr_post_init()), so a larger bounce lifetime does not lengthen it.
func TestLongestLifetimeIsTheQueueLifetime(t *testing.T) {
p, err := parseRetryPolicy([]string{"300s", "300s", "4000s", "1d", "5d", "0h"})
if err != nil {
t.Fatal(err)
}
if p.LongestLifetime() != 24*time.Hour || p.BounceQueueLifetime != 24*time.Hour {
t.Fatalf("LongestLifetime %v, BounceQueueLifetime %v; want 24h both", p.LongestLifetime(), p.BounceQueueLifetime)
}
}
func TestLoadRetryPolicyFallsBackWhenPostconfFails(t *testing.T) { func TestLoadRetryPolicyFallsBackWhenPostconfFails(t *testing.T) {
old := readRetryConf old := readRetryConf
readRetryConf = func() ([]string, error) { readRetryConf = func() ([]string, error) {
+6 -5
View File
@@ -69,13 +69,14 @@ func (s *Store) SetQueuedOverLifetime(at time.Time, n int64) error {
return nil return nil
} }
// CountQueuedOlderThan counts the rows still "queued" that were accepted // CountOpenOlderThan counts the rows still open - "queued" or "deferred" -
// before cutoff. created_at is RFC3339 UTC, so the strings compare as times. // that were accepted before cutoff. created_at is RFC3339 UTC, so the strings
func (s *Store) CountQueuedOlderThan(cutoff time.Time) (int64, error) { // compare as times.
func (s *Store) CountOpenOlderThan(cutoff time.Time) (int64, error) {
var n int64 var n int64
err := s.db.QueryRow( err := s.db.QueryRow(
`SELECT COUNT(*) FROM send_log WHERE status = ? AND queue_id <> '' AND created_at < ?`, `SELECT COUNT(*) FROM send_log WHERE status IN (?, ?) AND queue_id <> '' AND created_at < ?`,
StatusQueued, cutoff.UTC().Format(time.RFC3339), StatusQueued, StatusDeferred, cutoff.UTC().Format(time.RFC3339),
).Scan(&n) ).Scan(&n)
if err != nil { if err != nil {
return 0, fmt.Errorf("count queued send_log rows: %w", err) return 0, fmt.Errorf("count queued send_log rows: %w", err)
+3 -3
View File
@@ -59,7 +59,7 @@ func TestAddDeliveryLogDayRefusesAnotherColumn(t *testing.T) {
} }
} }
func TestCountQueuedOlderThan(t *testing.T) { func TestCountOpenOlderThan(t *testing.T) {
st := openTestStore(t) st := openTestStore(t)
for _, e := range []SendLogEntry{{QueueID: "Q1", To: "a@example.net"}, {QueueID: "Q2", To: "b@example.net"}} { for _, e := range []SendLogEntry{{QueueID: "Q1", To: "a@example.net"}, {QueueID: "Q2", To: "b@example.net"}} {
if err := st.InsertQueued(e); err != nil { if err := st.InsertQueued(e); err != nil {
@@ -69,10 +69,10 @@ func TestCountQueuedOlderThan(t *testing.T) {
if _, err := st.UpdateStatus("Q2", "b@example.net", StatusSent); err != nil { if _, err := st.UpdateStatus("Q2", "b@example.net", StatusSent); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if n, err := st.CountQueuedOlderThan(time.Now().Add(time.Hour)); err != nil || n != 1 { if n, err := st.CountOpenOlderThan(time.Now().Add(time.Hour)); err != nil || n != 1 {
t.Fatalf("queued rows before an hour from now = %d, %v; want 1", n, err) t.Fatalf("queued rows before an hour from now = %d, %v; want 1", n, err)
} }
if n, err := st.CountQueuedOlderThan(time.Now().Add(-time.Hour)); err != nil || n != 0 { if n, err := st.CountOpenOlderThan(time.Now().Add(-time.Hour)); err != nil || n != 0 {
t.Fatalf("queued rows before an hour ago = %d, %v; want 0", n, err) t.Fatalf("queued rows before an hour ago = %d, %v; want 0", n, err)
} }
} }
+8 -4
View File
@@ -28,19 +28,23 @@ func TestAttachRefusalLine(t *testing.T) {
} }
first := "Oct 14 12:00:00 mail postfix/smtpd[1]: NOQUEUE: milter-reject: MAIL from unknown[10.0.0.1]: 451 4.7.1 first" first := "Oct 14 12:00:00 mail postfix/smtpd[1]: NOQUEUE: milter-reject: MAIL from unknown[10.0.0.1]: 451 4.7.1 first"
second := "Oct 14 12:00:00 mail postfix/smtpd[2]: NOQUEUE: milter-reject: MAIL from unknown[10.0.0.1]: 451 4.7.1 second" second := "Oct 14 12:00:00 mail postfix/smtpd[2]: NOQUEUE: milter-reject: MAIL from unknown[10.0.0.1]: 451 4.7.1 second"
// Another application at the same client, as the same sender: not its.
if ok, _ := st.AttachRefusalLine("10.0.0.1", "other", "billing@example.com", now, "another login"); ok {
t.Fatal("a line of another login took a row")
}
for _, line := range []string{first, second} { for _, line := range []string{first, second} {
if ok, err := st.AttachRefusalLine("10.0.0.1", "Billing@example.com", now, line); err != nil || !ok { if ok, err := st.AttachRefusalLine("10.0.0.1", "L2App", "Billing@example.com", now, line); err != nil || !ok {
t.Fatalf("attach %q = %v, %v", line, ok, err) t.Fatalf("attach %q = %v, %v", line, ok, err)
} }
} }
// Read again after a restart: no row takes it a second time. // Read again after a restart: no row takes it a second time.
if ok, _ := st.AttachRefusalLine("10.0.0.1", "billing@example.com", now, first); ok { if ok, _ := st.AttachRefusalLine("10.0.0.1", "l2app", "billing@example.com", now, first); ok {
t.Fatal("a line already on a row took another") t.Fatal("a line already on a row took another")
} }
if ok, _ := st.AttachRefusalLine("10.0.0.3", "billing@example.com", now, "a third"); ok { if ok, _ := st.AttachRefusalLine("10.0.0.3", "l2app", "billing@example.com", now, "a third"); ok {
t.Fatal("a line of another client took a row") t.Fatal("a line of another client took a row")
} }
if ok, _ := st.AttachRefusalLine("10.0.0.2", "billing@example.com", now.Add(-time.Hour), "an hour off"); ok { if ok, _ := st.AttachRefusalLine("10.0.0.2", "other", "billing@example.com", now.Add(-time.Hour), "an hour off"); ok {
t.Fatal("a line an hour off took a row") t.Fatal("a line an hour off took a row")
} }
lines := map[string]string{} lines := map[string]string{}
+48 -17
View File
@@ -162,14 +162,14 @@ func (s *Store) UpdateStatus(queueID, recipient, status string) (int64, error) {
// CloseUnreported is the reconcile sweep's close of one row, by its id: // CloseUnreported is the reconcile sweep's close of one row, by its id:
// "bounced", marked reason "reconcile" so it is known who decided it // "bounced", marked reason "reconcile" so it is known who decided it
// (architecture.md § Log tailer). Only a row still "queued" is closed - a // (architecture.md § Log tailer). Only a row still open - "queued" or
// result the log wrote an instant earlier stands (audit 2026-10-12, H3). It // "deferred" - is closed; a final result the log wrote an instant earlier
// returns the number of rows closed, 0 or 1. // stands (audit 2026-10-12, H3). It returns the number of rows closed, 0 or 1.
func (s *Store) CloseUnreported(id int64) (int64, error) { func (s *Store) CloseUnreported(id int64) (int64, error) {
now := time.Now().UTC().Format(time.RFC3339) now := time.Now().UTC().Format(time.RFC3339)
res, err := s.db.Exec( res, err := s.db.Exec(
`UPDATE send_log SET status = ?, reason = ?, updated_at = ? WHERE id = ? AND status = ?`, `UPDATE send_log SET status = ?, reason = ?, updated_at = ? WHERE id = ? AND status IN (?, ?)`,
StatusBounced, ReasonReconcile, now, id, StatusQueued, StatusBounced, ReasonReconcile, now, id, StatusQueued, StatusDeferred,
) )
if err != nil { if err != nil {
return 0, fmt.Errorf("close unreported send_log row: %w", err) return 0, fmt.Errorf("close unreported send_log row: %w", err)
@@ -178,6 +178,30 @@ func (s *Store) CloseUnreported(id int64) (int64, error) {
return n, nil return n, nil
} }
// ExpireMessage closes the rows of a message Postfix gave up on at the end of
// its queue lifetime: each recipient's newest row with that queue id, if it
// is still open ("queued" or "deferred"), becomes "bounced" - Postfix returns
// the message to its sender. Postfix 3.7.11, src/qmgr/qmgr_active.c,
// qmgr_active_done_25_generic(), logs expiry once per message, "<id>:
// from=<..>, status=expired, returned to sender", with no line per
// recipient: the recipients' last lines say "deferred" (assumptions A-8). It
// returns the number of rows closed.
func (s *Store) ExpireMessage(queueID string) (int64, error) {
now := time.Now().UTC().Format(time.RFC3339)
res, err := s.db.Exec(
`UPDATE send_log SET status = ?, updated_at = ?
WHERE status IN (?, ?) AND id IN (
SELECT MAX(id) FROM send_log WHERE queue_id = ? AND status <> 'rejected'
GROUP BY to_addr COLLATE NOCASE)`,
StatusBounced, now, StatusQueued, StatusDeferred, queueID,
)
if err != nil {
return 0, fmt.Errorf("expire send_log rows of %s: %w", queueID, err)
}
n, _ := res.RowsAffected()
return n, nil
}
// QueuedDelivery is a send-log row still waiting for a delivery result, // QueuedDelivery is a send-log row still waiting for a delivery result,
// reduced to what the log-tailer's reconcile sweep needs to look it up in the // reduced to what the log-tailer's reconcile sweep needs to look it up in the
// Postfix queue and, failing that, to close it (architecture.md § Log tailer). // Postfix queue and, failing that, to close it (architecture.md § Log tailer).
@@ -187,22 +211,25 @@ type QueuedDelivery struct {
To string To string
} }
// ListQueuedOlderThan returns the rows still marked "queued" that were accepted // ListOpenOlderThan returns the rows still open - "queued", or "deferred"
// before cutoff — old enough that Postfix should long since have reported a // and waiting for the next attempt - that were accepted before cutoff, old
// result for them. Rows without a queue-id are skipped: the milter refused // enough that Postfix should long since have reported a result for them or
// those before Postfix ever saw the message, so the queue has nothing to say // still hold the message. Rows without a queue-id are skipped: the milter
// about them. // refused those before Postfix ever saw the message, so the queue has
// nothing to say about them. A deferred row is among them because Postfix
// can drop a deferred message without a line per recipient: at expiry (one
// line per message, ExpireMessage) and by postsuper -d (only "removed").
// //
// created_at is stored as RFC3339 UTC, so a lexical comparison against the same // created_at is stored as RFC3339 UTC, so a lexical comparison against the same
// format is chronologically correct. SQLite 3.53.4 compares TEXT with the // format is chronologically correct. SQLite 3.53.4 compares TEXT with the
// BINARY collation, memcmp() (https://sqlite.org/datatype3.html § 7), and // BINARY collation, memcmp() (https://sqlite.org/datatype3.html § 7), and
// time.RFC3339 of a UTC time is fixed-width, YYYY-MM-DDTHH:MM:SSZ. // time.RFC3339 of a UTC time is fixed-width, YYYY-MM-DDTHH:MM:SSZ.
func (s *Store) ListQueuedOlderThan(cutoff time.Time) ([]QueuedDelivery, error) { func (s *Store) ListOpenOlderThan(cutoff time.Time) ([]QueuedDelivery, error) {
rows, err := s.db.Query( rows, err := s.db.Query(
`SELECT id, queue_id, to_addr FROM send_log `SELECT id, queue_id, to_addr FROM send_log
WHERE status = ? AND queue_id <> '' AND created_at < ? WHERE status IN (?, ?) AND queue_id <> '' AND created_at < ?
ORDER BY id`, ORDER BY id`,
StatusQueued, cutoff.UTC().Format(time.RFC3339), StatusQueued, StatusDeferred, cutoff.UTC().Format(time.RFC3339),
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("list queued send_log rows: %w", err) return nil, fmt.Errorf("list queued send_log rows: %w", err)
@@ -253,8 +280,10 @@ type SendLogRow struct {
// AttachRefusalLine puts Postfix's line about a refusal at MAIL FROM on the // AttachRefusalLine puts Postfix's line about a refusal at MAIL FROM on the
// row the journal milter wrote for it (audit 2026-10-12, stage 2b). Such a // row the journal milter wrote for it (audit 2026-10-12, stage 2b). Such a
// refusal has no queue id, and the line names no login, so the row is found // refusal has no queue id, and the line names no login, so the row is found
// by what the line says: a refused row of that client and sender without a // by what the line says: a refused row of that client, login and sender
// line yet, written within two minutes of the line's time; the nearest in // without a line yet - the login is smtpd's sasl_username= on the line, so
// two applications behind one address sending as one sender are told apart
// (assumptions A-21) - written within two minutes of the line's time; the nearest in
// time, the oldest of equals, when there are several - lines are read in the // time, the oldest of equals, when there are several - lines are read in the
// order the rows were written, so two refusals of one second take one line // order the rows were written, so two refusals of one second take one line
// each in turn (review of stage 2b). A line already on a row - read again // each in turn (review of stage 2b). A line already on a row - read again
@@ -265,12 +294,13 @@ type SendLogRow struct {
// reads YYYY-MM-DDTHH:MM:SSZ, the T and the Z included (lang_datefunc.html // reads YYYY-MM-DDTHH:MM:SSZ, the T and the Z included (lang_datefunc.html
// § 2); NOCASE folds ASCII only (datatype3.html § 7.1); ?1 is bound once and // § 2); NOCASE folds ASCII only (datatype3.html § 7.1); ?1 is bound once and
// used twice (lang_expr.html § 4). // used twice (lang_expr.html § 4).
func (s *Store) AttachRefusalLine(clientIP, from string, at time.Time, line string) (bool, error) { func (s *Store) AttachRefusalLine(clientIP, login, from string, at time.Time, line string) (bool, error) {
res, err := s.db.Exec( res, err := s.db.Exec(
`UPDATE send_log SET refusal_line = ?1 WHERE id = ( `UPDATE send_log SET refusal_line = ?1 WHERE id = (
SELECT id FROM send_log SELECT id FROM send_log
WHERE status = ?2 AND queue_id = '' AND refusal_line = '' WHERE status = ?2 AND queue_id = '' AND refusal_line = ''
AND reason IN ('', ?3, ?4) AND from_addr = ?5 COLLATE NOCASE AND client_ip = ?6 AND reason IN ('', ?3, ?4) AND from_addr = ?5 COLLATE NOCASE AND client_ip = ?6
AND app_login = ?10 COLLATE NOCASE
AND created_at BETWEEN ?7 AND ?8 AND created_at BETWEEN ?7 AND ?8
ORDER BY ABS(julianday(created_at) - julianday(?9)), id ORDER BY ABS(julianday(created_at) - julianday(?9)), id
LIMIT 1) LIMIT 1)
@@ -279,6 +309,7 @@ func (s *Store) AttachRefusalLine(clientIP, from string, at time.Time, line stri
at.Add(-refusalLineWindow).UTC().Format(time.RFC3339), at.Add(-refusalLineWindow).UTC().Format(time.RFC3339),
at.Add(refusalLineWindow).UTC().Format(time.RFC3339), at.Add(refusalLineWindow).UTC().Format(time.RFC3339),
at.UTC().Format(time.RFC3339), at.UTC().Format(time.RFC3339),
login,
) )
if err != nil { if err != nil {
return false, fmt.Errorf("attach refusal line: %w", err) return false, fmt.Errorf("attach refusal line: %w", err)
@@ -452,7 +483,7 @@ func sendLogWhere(f SendLogFilter) (string, []any) {
// implementing the configurable retention window (architecture.md § // implementing the configurable retention window (architecture.md §
// Persistence, SEND_LOG_RETENTION_DAYS). It returns the number of rows pruned. // Persistence, SEND_LOG_RETENTION_DAYS). It returns the number of rows pruned.
// created_at is stored as RFC3339 UTC, so a lexical comparison against the // created_at is stored as RFC3339 UTC, so a lexical comparison against the
// same format is chronologically correct (see ListQueuedOlderThan). // same format is chronologically correct (see ListOpenOlderThan).
func (s *Store) DeleteSendLogBefore(cutoff time.Time) (int64, error) { func (s *Store) DeleteSendLogBefore(cutoff time.Time) (int64, error) {
res, err := s.db.Exec( res, err := s.db.Exec(
`DELETE FROM send_log WHERE created_at < ?`, `DELETE FROM send_log WHERE created_at < ?`,
+69 -3
View File
@@ -142,6 +142,72 @@ func TestCloseUnreportedClosesOnlyQueuedRows(t *testing.T) {
} }
} }
// Postfix's expiry line closes the message's open rows - each recipient's
// newest, queued or deferred - and leaves a final result, a refusal and an
// older message that had the same queue id alone (assumptions A-8).
func TestExpireMessageClosesOnlyOpenRows(t *testing.T) {
st := openTestStore(t)
for _, e := range []SendLogEntry{
{QueueID: "OLDQ", To: "a@example.net"}, // an earlier message, same id
{QueueID: "Q1", To: "a@example.net"},
{QueueID: "Q1", To: "b@example.net"},
{QueueID: "Q1", To: "c@example.net"},
} {
if err := st.InsertQueued(e); err != nil {
t.Fatal(err)
}
}
if _, err := st.db.Exec(`UPDATE send_log SET queue_id = 'Q1' WHERE queue_id = 'OLDQ'`); err != nil {
t.Fatal(err)
}
if _, err := st.UpdateStatus("Q1", "a@example.net", StatusDeferred); err != nil {
t.Fatal(err)
}
if _, err := st.UpdateStatus("Q1", "c@example.net", StatusSent); err != nil {
t.Fatal(err)
}
if n, err := st.ExpireMessage("Q1"); err != nil || n != 2 {
t.Fatalf("ExpireMessage = %d, %v; want 2 (a deferred, b queued)", n, err)
}
rows, err := st.db.Query(`SELECT to_addr, status FROM send_log ORDER BY id`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
var got []string
for rows.Next() {
var to, status string
if err := rows.Scan(&to, &status); err != nil {
t.Fatal(err)
}
got = append(got, to+" "+status)
}
want := []string{
"a@example.net " + StatusQueued, // the earlier message's row
"a@example.net " + StatusBounced,
"b@example.net " + StatusBounced,
"c@example.net " + StatusSent,
}
if strings.Join(got, "; ") != strings.Join(want, "; ") {
t.Fatalf("rows = %v, want %v", got, want)
}
}
// The sweep closes a deferred row as it closes a queued one: Postfix can drop
// a deferred message without a line per recipient.
func TestCloseUnreportedClosesADeferredRow(t *testing.T) {
st := openTestStore(t)
if err := st.InsertQueued(SendLogEntry{QueueID: "Q1", To: "a@example.net"}); err != nil {
t.Fatal(err)
}
if _, err := st.UpdateStatus("Q1", "a@example.net", StatusDeferred); err != nil {
t.Fatal(err)
}
if n, err := st.CloseUnreported(rowID(t, st, "Q1")); err != nil || n != 1 {
t.Fatalf("CloseUnreported of a deferred row = %d, %v; want 1", n, err)
}
}
// A delivery line about a row the sweep closed proves the sweep guessed what // A delivery line about a row the sweep closed proves the sweep guessed what
// the log could have said: the log's status wins, the row is marked // the log could have said: the log's status wins, the row is marked
// reconcile-contradicted, and the day counts one contradiction - once, // reconcile-contradicted, and the day counts one contradiction - once,
@@ -205,7 +271,7 @@ func TestUpdateStatusRewritesOnlyTheNewestRow(t *testing.T) {
// reported on them by now. A message accepted moments ago is simply in flight, // reported on them by now. A message accepted moments ago is simply in flight,
// and one the milter refused never reached the queue at all, so neither is the // and one the milter refused never reached the queue at all, so neither is the
// sweep's business. // sweep's business.
func TestListQueuedOlderThan(t *testing.T) { func TestListOpenOlderThan(t *testing.T) {
st := openTestStore(t) st := openTestStore(t)
for _, e := range []SendLogEntry{ for _, e := range []SendLogEntry{
@@ -232,9 +298,9 @@ func TestListQueuedOlderThan(t *testing.T) {
backdate(t, st, "OLD1", cutoff.Add(-time.Hour)) backdate(t, st, "OLD1", cutoff.Add(-time.Hour))
backdate(t, st, "DONE1", cutoff.Add(-time.Hour)) backdate(t, st, "DONE1", cutoff.Add(-time.Hour))
got, err := st.ListQueuedOlderThan(cutoff) got, err := st.ListOpenOlderThan(cutoff)
if err != nil { if err != nil {
t.Fatalf("ListQueuedOlderThan: %v", err) t.Fatalf("ListOpenOlderThan: %v", err)
} }
if len(got) != 1 || got[0] != (QueuedDelivery{ID: got[0].ID, QueueID: "OLD1", To: "stale@example.net"}) { if len(got) != 1 || got[0] != (QueuedDelivery{ID: got[0].ID, QueueID: "OLD1", To: "stale@example.net"}) {
t.Fatalf("got %+v, want only the stale queued row", got) t.Fatalf("got %+v, want only the stale queued row", got)
+1 -1
View File
@@ -103,7 +103,7 @@ func (s *Store) DeleteUserSessions(userID int64) error {
// forever. It piggybacks on session creation rather than running as its own // forever. It piggybacks on session creation rather than running as its own
// background loop. // background loop.
// expires_at is RFC3339 UTC text, so < compares moments (see // expires_at is RFC3339 UTC text, so < compares moments (see
// ListQueuedOlderThan). // ListOpenOlderThan).
func (s *Store) DeleteExpiredSessions(now time.Time) (int64, error) { func (s *Store) DeleteExpiredSessions(now time.Time) (int64, error) {
res, err := s.db.Exec(`DELETE FROM sessions WHERE expires_at < ?`, now.UTC().Format(time.RFC3339)) res, err := s.db.Exec(`DELETE FROM sessions WHERE expires_at < ?`, now.UTC().Format(time.RFC3339))
if err != nil { if err != nil {
+1 -1
View File
@@ -167,7 +167,7 @@ func TestARefusedMessageShowsPostfixsLine(t *testing.T) {
} }
if subject == "with its line" { if subject == "with its line" {
line := time.Now().UTC().Format("2006-01-02T15:04:05Z") + " mail postfix/smtpd[2131]: NOQUEUE: milter-reject: MAIL from unknown[198.51.100.7]: 451 4.7.1 Service unavailable - try again later; from=<alerts@example.org> proto=ESMTP helo=<app>" line := time.Now().UTC().Format("2006-01-02T15:04:05Z") + " mail postfix/smtpd[2131]: NOQUEUE: milter-reject: MAIL from unknown[198.51.100.7]: 451 4.7.1 Service unavailable - try again later; from=<alerts@example.org> proto=ESMTP helo=<app>"
if ok, err := st.AttachRefusalLine("198.51.100.7", "alerts@example.org", time.Now(), line); err != nil || !ok { if ok, err := st.AttachRefusalLine("198.51.100.7", "alerts", "alerts@example.org", time.Now(), line); err != nil || !ok {
t.Fatalf("attach = %v, %v", ok, err) t.Fatalf("attach = %v, %v", ok, err)
} }
} }
+4 -1
View File
@@ -262,7 +262,10 @@ func NewOutMessage(m Meta, in MessageInput) *OutMessage {
history.Help = topicLink(HelpSenderCheck, "What an application may write in From:") history.Help = topicLink(HelpSenderCheck, "What an application may write in From:")
} }
if in.Recovered { if in.Recovered {
facts = append(facts, Fact{Label: "Written", Value: Rich("from ", Code("mail.log"), " after a restart: the panel stopped before it wrote the refusal")}) // cleanup's line names the message's last recipient only (Postfix
// 3.7.11, src/cleanup/cleanup_milter.c, cleanup_milter_apply(),
// " to=<%s>" of state->recip; assumptions A-15).
facts = append(facts, Fact{Label: "Written", Value: Rich("from ", Code("mail.log"), " after a restart: the panel stopped before it wrote the refusal. Postfix's line names the last recipient only")})
} }
if in.QueueID != "" { if in.QueueID != "" {
facts = append(facts, Fact{Label: "Queue id", Value: Rich(Mono(in.QueueID), " - given, the message was not queued")}) facts = append(facts, Fact{Label: "Queue id", Value: Rich(Mono(in.QueueID), " - given, the message was not queued")})
+72
View File
@@ -78,3 +78,75 @@ func testDeliveryLogSurvivesRotation(t *testing.T, sc *scenario) {
t.Fatalf("after mail.log was rotated: %v", err) t.Fatalf("after mail.log was rotated: %v", err)
} }
} }
// testExpiredMessageIsClosed: a message Postfix gives up on is closed as
// bounced. Postfix logs expiry once per message, "<id>: from=<..>,
// status=expired, returned to sender" (here "force-expired"), and no line per
// recipient (Postfix 3.7.11, src/qmgr/qmgr_active.c,
// qmgr_active_done_25_generic(); assumptions A-8). The queue lifetime cannot
// be waited out, so the message is deferred by hand - defer_transports=smtp,
// under which qmgr hands the recipient to the retry service, "4.3.2 deferred
// transport" (src/qmgr/qmgr_message.c, qmgr_message_resolve()) - and then
// forced to expire: postsuper -e marks it, postqueue -i has qmgr look at it
// at once (man/man1/postsuper.1, postqueue.1).
func testExpiredMessageIsClosed(t *testing.T, sc *scenario) {
exec := func(args ...string) {
t.Helper()
if out, err := h.execIn("selfpost", args...); err != nil {
t.Fatalf("%s: %v\n%s", strings.Join(args, " "), err, out)
}
}
exec("postconf", "-e", "defer_transports=smtp")
exec("postfix", "reload")
defer func() {
// The bounce to the sender waits behind the same setting.
h.execIn("selfpost", "postconf", "-e", "defer_transports=")
h.execIn("selfpost", "postfix", "reload")
h.execIn("selfpost", "postqueue", "-f")
}()
token := uniqueToken("expire")
if res := attemptSend(sendAttempt{
authLogin: sc.appLogin, authPassword: sc.appPassword,
from: "alerts@" + senderDomain, headerFrom: "alerts@" + senderDomain, to: recipient,
subject: "e2e " + token, body: "deferred, then expired",
}); !res.ok() {
t.Fatalf("send: %v", res.firstErr())
}
row := func() (string, error) {
return asPanel(`import sqlite3,sys
r=sqlite3.connect("/data/selfpost.db").execute("select queue_id, status from send_log where subject like ? order by id desc",("%"+sys.argv[1]+"%",)).fetchone()
print("" if r is None else r[0]+" "+r[1])`, token)
}
var queueID string
if err := waitFor("the row to be deferred", 30*time.Second, 500*time.Millisecond, func() (bool, error) {
out, err := row()
if err != nil {
return false, err
}
if f := strings.Fields(out); len(f) == 2 && f[1] == "deferred" {
queueID = f[0]
return true, nil
}
return false, fmt.Errorf("row %q", out)
}); err != nil {
t.Fatal(err)
}
exec("postsuper", "-e", queueID)
exec("postqueue", "-i", queueID)
if err := waitFor("the expired message's row to be bounced", 30*time.Second, 500*time.Millisecond, func() (bool, error) {
out, err := row()
if err != nil {
return false, err
}
return out == queueID+" bounced", fmt.Errorf("row %q", out)
}); err != nil {
out, _ := h.execIn("selfpost", "grep", queueID, "/data/log/mail.log")
t.Fatalf("%v\nmail.log:\n%s", err, out)
}
out, err := h.execIn("selfpost", "grep", queueID+": from=<alerts@"+senderDomain+">, status=force-expired, returned to sender", "/data/log/mail.log")
if err != nil {
t.Fatalf("no expiry line in mail.log: %v\n%s", err, out)
}
}
+3
View File
@@ -345,6 +345,9 @@ func TestE2E(t *testing.T) {
// From here on the installation is no longer the one the run started: it // From here on the installation is no longer the one the run started: it
// is restored from its own backup, then switched to accept mail on port 25. // is restored from its own backup, then switched to accept mail on port 25.
run("expired_message_is_closed", func(t *testing.T) {
testExpiredMessageIsClosed(t, sc)
})
run("full_backup_restore", func(t *testing.T) { run("full_backup_restore", func(t *testing.T) {
testFullBackupRestore(t, sc) testFullBackupRestore(t, sc)
}) })