From a990e7cc23ea8b3a3e9ef0513cbc14623d7e0a8c Mon Sep 17 00:00:00 2001 From: Mix Date: Mon, 12 Oct 2026 10:41:04 +0300 Subject: [PATCH] delivery-log-storage: findings of the technical and security reviews Neither review (read-only agents, Opus and Fable, over ae06d8b..815236a) has a Critical or High finding. What they found and what was done: - A stranger on port 25 could grow the table: every queue-id line of inbound mail was stored and pruned an hour or more later, his own message-id among them. The tailer now leaves out the lines of a queue id that postfix/inbound/smtpd opened, until qmgr removes the message or another listener opens one under the id; 4096 such ids are remembered. - A commit, so a disk sync, per stored line on the connection the journal milter shares, and a backlog written line by line. Lines are now written in one transaction per pass over the log, 500 at most, before the read offset is saved (InsertSendLogLines; follow() calls the keeper's flush). - The prune asked the send log once per line and ran hourly. It asks once per queue id and runs at start and with the six-hourly sweep; with port-25 mail left out there is little without a row to wait for. - The 4096-byte cut could split a character; it keeps characters whole. - security.md did not say that a backup now carries these lines, and the plan said the time window excludes a reused queue id where it only limits the chance to the row's queue lifetime. Both are corrected; the collision is recorded as an accepted risk with its trigger. - Tests that did not prove their comments: the prune is now tested with the sweep, a rejected row next to the prune, batching and the change from catching up to following, port-25 mail left out and its id reused. Left as found and written down in the plan: lines the tailer never reads (a file rotated away while the panel was down), and a message in the queue across the upgrade, which shows only its later lines. On the test server, for this tree: go vet clean, go test ./... 22 packages ok, make e2e ok. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 23 +++- docs/architecture.md | 30 +++-- docs/plans/delivery-log-storage.md | 70 +++++++--- docs/roadmap.md | 2 +- docs/security.md | 21 ++- internal/logtail/lines.go | 124 ++++++++++++++---- internal/logtail/lines_test.go | 114 ++++++++++++++-- internal/logtail/logtail.go | 29 ++-- internal/logtail/logtail_test.go | 34 ++--- internal/logtail/retention_test.go | 6 +- internal/postfix/retry.go | 5 +- internal/store/sendloglines.go | 96 +++++++++----- internal/store/sendloglines_test.go | 72 +++++----- .../web/handlers/handlers_monitor_test.go | 8 +- 14 files changed, 456 insertions(+), 178 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6137f52..174bf16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,12 +12,14 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version its send-log row lives, with the store methods to write a line, to write it once, to read the lines of a row and to prune ([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)). -- **The log tailer keeps what Postfix writes about queued messages.** Every - `mail.log` line headed by a queue id is stored with the line's own time, as - the tailer reads it; after an unclean stop a line read twice is stored once. - Lines whose queue id has no send-log row - the row was pruned, or the - message never had one (inbound relay mail, bounces) - are deleted once they - are an hour old, with every send-log sweep and hourly between. +- **The log tailer keeps what Postfix writes about queued messages.** A + `mail.log` line headed by a queue id is stored with the line's own time, one + transaction per pass over the log; after an unclean stop a line read twice + is stored once. Mail that came in on port 25 is left out: a stranger's + message never has a send-log row, and what he can make Postfix log is not + written to the database. Lines whose queue id has no send-log row - the row + was pruned, or the message never had one (bounces) - are deleted once they + are an hour old, at start and with every six-hourly send-log sweep. ### Changed @@ -31,6 +33,15 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version lines only while they are still in the current file, as before; the note shown when there are none and the *Outbound log* help topic say so. +### Security + +- **A full backup now carries the log lines of messages in the send log** - + client addresses, SASL logins, message ids, receivers' replies - where + `mail.log` itself is not in a backup. Recorded in + [security.md](docs/security.md) with the accepted risk of a queue-id + collision showing one message's lines on another's page. Technical and + security reviews of 2026-10-12: no Critical or High finding. + ## [2.1.0] - 2026-10-12 ### Security diff --git a/docs/architecture.md b/docs/architecture.md index 696c36e..c66b075 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -243,34 +243,48 @@ day before it is rotated; a message's page (`/outbound/log/{id}`) shows what Postfix wrote about it, so the tailer keeps those lines in the database as it reads them: -- **What is kept:** every line headed by a queue id — after the timestamp, the +- **What is kept:** a line headed by a queue id — after the timestamp, the host name and `program[pid]:` — with that id, the line's own time and the line as written, cut to 4096 bytes. The id is read from the head of the line only: the tail is the far end's text. `NOQUEUE` and `warning` stand in the same place and are not ids; the pattern accepts Postfix's short form (capital hexadecimal) and its long form (no vowels). +- **Port-25 mail is left out.** A queue id opened by `postfix/inbound/smtpd` + is a stranger's message and never gets a send-log row; none of its lines + are stored, up to qmgr's `removed` or until another listener opens a + message under the id (4096 such ids are remembered). What a stranger can + make Postfix log is therefore not written to the database at his rate. - **No check for a row at write time.** A message's first lines (`client=`, `message-id=`) are logged before the journal milter writes its row at end-of-message, so the tailer cannot ask whether the id is in the send log. - Lines of messages that never get a row — inbound relay mail, bounces Postfix - generates — are kept too, for a while. + Lines that never get a row — bounces Postfix generates, port-25 mail met in + mid-flight after a restart — are kept for a while. +- **Written per pass, not per line:** the lines of one pass over the log go + in one transaction (`InsertSendLogLines`), 500 at most, before the read + offset is saved. A commit per line would be a disk sync per line on the + connection the journal milter shares. - **Retention is one rule:** a line older than an hour whose queue id no - send-log row carries is deleted (`PruneSendLogLines`), at start, after every - send-log sweep and hourly between. That removes the lines of rows the sweep - deleted and the lines that never had a row. + send-log row carries is deleted (`PruneSendLogLines`), at start and after + every six-hourly send-log sweep. That removes the lines of rows the sweep + deleted and the lines that never had a row. The send log is asked once per + queue id; the statement still walks the table, about a second per million + lines, with the connection taken meanwhile. - **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 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 - `bounce_queue_lifetime`) plus a day after it. A message with several + `bounce_queue_lifetime`) 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 recipients is several rows under one id, and each row's page shows all of the message's lines. - **Read twice, stored once.** The read offset is saved every few seconds, so after an unclean stop some lines are read again; while the tailer is catching up at start it writes a line only if the same line is not there. +- **Not stored:** what the tailer never read — the tail of a file rotated away + while the panel was down. - **Size:** six to ten lines of about 200 bytes per message — more than the row itself. The table is in the database and therefore in a full backup, which - `log/` is not. + `log/` is not ([security.md](security.md) § Backup and domain export). **Two one-shot reads** sit beside the follow loop and are unrelated to it, both serving panel pages on request: `TailLines` (the last *n* lines, for diff --git a/docs/plans/delivery-log-storage.md b/docs/plans/delivery-log-storage.md index 92beec0..1ba93aa 100644 --- a/docs/plans/delivery-log-storage.md +++ b/docs/plans/delivery-log-storage.md @@ -29,20 +29,31 @@ rotation, while the row itself lives for ninety days. ### What is stored -Every `mail.log` line whose head is a queue id - after the timestamp, the host +A `mail.log` line whose head is a queue id - after the timestamp, the host name and `program[pid]:` - goes into a new table with that id, the line's own time (`lineTime`, the same reading the inbound journal uses) and the line as -written, cut to 4096 bytes. The head is matched from the start of the line: -the tail is the far end's words. `NOQUEUE` and `warning` are not queue ids - -the pattern takes Postfix's short form (hexadecimal) and its long form (no -vowels) and nothing else. +written, cut to 4096 bytes on a character boundary. The head is matched from +the start of the line: the tail is the far end's words. `NOQUEUE` and +`warning` are not queue ids - the pattern takes Postfix's short form +(hexadecimal) and its long form (no vowels) and nothing else. The tailer does not ask whether the id belongs to a send-log row before it writes. The first lines of a message (`client=`, `message-id=`) are logged before the journal milter writes the row at end-of-message, so such a check -would drop exactly them. Lines of messages that never get a row - inbound -relay mail, bounces Postfix generates - are stored too and removed by the -prune below within the hour or two. +would drop exactly them. + +**Mail that came in on port 25 is left out** (security review, 2026-10-12). A +queue id that `postfix/inbound/smtpd` opened is a stranger's message; it never +gets a send-log row, and what a stranger can make Postfix log - his own +`message-id`, up to the cut - is not stored to be pruned later. The tailer +remembers such ids (4096 at most) until qmgr's `removed` or until another +listener opens a message under the id. What is still stored without a row is +little: the bounces Postfix generates, and port-25 mail met in mid-flight +after a restart. + +Lines are written in one transaction per pass over the log, 500 at most +(technical review): a commit per line is a disk sync per line on the one +connection the journal milter writes through. ### Which lines are a row's @@ -50,19 +61,26 @@ A line is shown on a row's page when the queue id is the same **and** the line's time lies between one hour before the row was written and the queue lifetime (the larger of `maximal_queue_lifetime` and `bounce_queue_lifetime`, as `postconf` reports them) plus a day after it. Postfix reuses a short queue -id once the message has left the queue; a later message under the same id -falls outside that window. This is the bound the roadmap item asked for. +id once the message has left the queue - the id is the microsecond of arrival +and the inode of the queue file - so a row that lives ninety days can meet a +later message under its id. The window does not rule that out, it limits it +to the days the row's own message could still be in the queue; this is the +bound the roadmap item asked for. The search of the current file that the +page did before had the same exposure for a day. `enable_long_queue_ids` +would end it and is a change to the mail path, not part of this item. ### Retention Lines live as long as a send-log row with their queue id does. One rule covers both ends: a line older than an hour whose queue id has no send-log row is deleted. That removes the lines of a row the retention sweep has just deleted, -and the lines of messages that never had a row. It runs with the send-log -sweep and once an hour between. +and the lines of messages that never had a row. It runs at start and after +every send-log sweep, six-hourly. It asks the send log once per queue id; at a +million lines that is still a scan of the table, of the order of a second, +during which the one connection is taken. -There is no cap of its own: the table is bounded by what Postfix logs about -queued messages, which its own rate limits bound. +There is no cap of its own: with port-25 mail left out, the table is bounded +by what authenticated applications send, which level 1 and level 2 limit. ### After an unclean stop @@ -70,6 +88,16 @@ The tailer saves its place every few seconds, so after a crash it reads some lines twice. While it is catching up at start a line is written only if the same line is not there already - the rule the inbound journal follows. +### What is not kept + +- Lines the tailer never reads are not stored: the tail of a file that was + rotated away while the panel was down, as with delivery statuses + (architecture.md § Log tailer). +- A message in the queue across the upgrade to 2.2.0 has only its later lines + stored, and the page then shows those alone: the search of the file runs + only for a row with nothing stored. The only installation is the owner's + test server; no merge of the two sources is built for one moment of it. + ### The page The page reads the table: the latest 200 lines of the row, oldest first. When @@ -119,9 +147,13 @@ rotation or to the Server › Mail log page. today from the file through the same escaping; storing them changes where they are read from, not what is shown or to whom. A domain administrator sees the lines of messages of their own domains, as now. -- They now enter the database and so a full backup, which `log/` does not. +- They now enter the database and so a full backup, which `log/` does not: + submitting clients' addresses and SASL logins, message ids, receivers' + replies. [security.md](../security.md) says so. - A line is matched from its head, so a reply cannot make itself a line of - another message; Postfix does not log raw line breaks. + another message; Postfix does not log raw line breaks, and `postlogd` is the + file's only writer. +- A stranger on port 25 does not write to the table (above). ## Done when @@ -143,10 +175,10 @@ Target version cut: **`2.2.0`** (MINOR). One commit per step; [development.md](../development.md) § Plan checklists. - [x] Migration `0003_send_log_lines.sql`, store methods (insert, insert once, lines of a row, prune) and their tests — **Opus** -- [x] Log tailer: store queue-id lines, write once while catching up, prune with the send-log sweep and hourly — **Opus** +- [x] Log tailer: store queue-id lines, write once while catching up, prune with the send-log sweep — **Opus** (hourly as first built; six-hourly after the reviews) - [x] The two texts, mockup first: the *Outbound log* help topic and the empty note (`design:` commit) — **Opus** (the help topic is in the mockups; the empty note is not drawn there - the `out-message` mockup shows a message with lines - and changes with the page) - [x] Message page reads the table, falls back to the current `mail.log`; help text in `help.go` — **Opus** - [x] [architecture.md](../architecture.md), [guide.md](../guide.md), [schema-migrations.md](../schema-migrations.md) — **Opus** (schema-migrations.md went in with the migration) - [x] `go vet`, `go test`, e2e: a message's page still lists its lines after a forced rotation — **Opus** (on the test server: vet ok, 22 packages ok, e2e ok with the new step `delivery_log_survives_rotation`) -- [ ] Technical review — **Opus** -- [ ] Security review — **Fable** +- [x] Technical review — **Opus** (2026-10-12; no Critical or High. Fixed: one transaction per pass instead of a commit per line, the prune asks the send log once per queue id and runs six-hourly, the cut keeps characters whole, tests for what the comments claimed. Left as found and written down under *What is not kept*: lines the tailer never read, and a message in the queue across the upgrade) +- [x] Security review — **Fable** (2026-10-12; no Critical or High. Fixed: port-25 mail is not stored, so a stranger cannot grow the table; [security.md](../security.md) states what a backup now carries. The window against queue-id reuse is described as the limit it is) diff --git a/docs/roadmap.md b/docs/roadmap.md index 268e7b4..1a632ad 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -36,7 +36,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md). | structured-logging | `log/slog` with levels and fields | candidate | — | — | | review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — | | panel-notifications | E-mail notifications about important events | candidate | — | — | -| delivery-log-storage | Keep a message's `mail.log` lines with its send-log row | in progress | 6/8 | [plans/delivery-log-storage.md](plans/delivery-log-storage.md) | +| delivery-log-storage | Keep a message's `mail.log` lines with its send-log row | in progress | 8/8 | [plans/delivery-log-storage.md](plans/delivery-log-storage.md) | | password-reset | Password reset by e-mail | candidate | — | — | | phone-desktop-mode | A phone browser in "desktop site" mode gets the phone navigation | candidate | — | — | | panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) | diff --git a/docs/security.md b/docs/security.md index 89fa94c..20ea774 100644 --- a/docs/security.md +++ b/docs/security.md @@ -244,6 +244,11 @@ operator can act on it. private key from `certs/` when present; a domain export carries the DKIM key and **working** application passwords in the clear (otherwise a transfer without recreating credentials would be impossible). +- Since 2.2.0 a full backup also carries, in the database, what Postfix logged + about each message still in the send log (`send_log_lines`): the submitting + client's address and SASL login, the message id, the receiving server and + its reply. `mail.log` itself stays out of a backup; these lines are state + the panel shows, kept for the send log's retention and no longer. - Both downloads are encrypted with a password unless the box on the form is unticked (it is ticked by default): scrypt (N=2¹⁵, r=8, p=1) → AES-256-GCM, streamed in 64 KiB chunks, each @@ -404,7 +409,21 @@ deferred item from the roadmap. the file is inaccessible to others. The log holds envelope addresses and client IPs, but neither message bodies nor headers; it is excluded from backups (`log/` is skipped) so that a dump stays state rather than - diagnostics. + diagnostics. Since 2.2.0 the tailer copies the lines of queued outbound + messages into the database for the message's page, and those are in a + backup (§ Backup and domain export); mail offered on port 25 is not copied. +- **A message's page can show another message's log lines after a queue-id + collision.** Postfix's short queue ids repeat: the id is the microsecond of + arrival and the queue file's inode, and inodes come round within seconds. + The page takes the stored lines of the row's queue id dated within the + row's own queue lifetime (about six days), so a second message has to get + the same microsecond on the same inode in that span - about one in a + million per reuse of the inode, and not something a sender can aim. When it + happens, a `domain` user could read the log lines of a message that is not + theirs. The search of the current `mail.log` that the page did before + 2.2.0 had the same exposure over a day. Trigger to revisit: a report of + foreign lines on a page, or a reason to turn on `enable_long_queue_ids`, + which ends it. - **A `domain` user may point an inbound domain's upstream at any host and port.** The upstream form checks that the host is a well-formed name or IP and the port is 1–65535; loopback, private and link-local addresses are diff --git a/internal/logtail/lines.go b/internal/logtail/lines.go index 2a29e0a..f05699a 100644 --- a/internal/logtail/lines.go +++ b/internal/logtail/lines.go @@ -27,44 +27,122 @@ import ( // "warning:" stand there too. A queue id is hexadecimal in capitals, or, with // enable_long_queue_ids, digits and letters without vowels; neither word is // either. -var queueLineRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-F]{6,24}|[0-9B-DF-HJ-NP-TV-Zb-df-hj-np-tv-z]{10,24}): `) +var queueLineRe = regexp.MustCompile(`^\S+ ([^\s\[]+)\[\d+\]: ([0-9A-F]{6,24}|[0-9B-DF-HJ-NP-TV-Zb-df-hj-np-tv-z]{10,24}): (.*)`) -// lineGrace is how old a line must be before the prune asks whether a send-log -// row carries its queue id. The first lines of a message are logged when it -// arrives and its row is written at end-of-message; an hour is a message body -// sent slowly. -const lineGrace = time.Hour +// smtpdProgramRe tells an smtpd from the other programs, whatever listener it +// serves: "postfix/smtpd", "postfix/submission/smtpd", "postfix/inbound/smtpd". +var smtpdProgramRe = regexp.MustCompile(`^postfix/(?:[a-z0-9-]+/)?smtpd$`) -// linePruneInterval is how often lines without a row are removed between two -// send-log sweeps. With the inbound relay on, most of what Postfix logs under -// a queue id is mail that never gets a send-log row; it is not left to pile up -// for six hours. A var so tests can shorten it. -var linePruneInterval = time.Hour +// port25Program is the smtpd of the inbound relay's listener, which +// build/postfix-config.sh names so that its lines are told from those of the +// submission ports. +const port25Program = "postfix/inbound/smtpd" -// storeLine keeps line if it is headed by a queue id. It does not ask whether -// a send-log row exists for the id: for a message's first lines none does yet, -// and the lines of messages that never get one are removed by pruneLines. +const ( + // lineGrace is how old a line must be before the prune asks whether a + // send-log row carries its queue id. The first lines of a message are + // logged when it arrives and its row is written at end-of-message; an hour + // is a message body sent slowly. + lineGrace = time.Hour + + // lineBatch is how many lines are held before they are written without + // waiting for the end of the pass over the log. A pass is a second's worth + // of lines while following, and the whole backlog when the tailer starts. + lineBatch = 500 + + // port25IDs is how many queue ids opened on port 25 are remembered so that + // their lines are not kept. Beyond it the oldest knowledge is given up, and + // those messages' lines are kept until the prune finds them without a row. + port25IDs = 4096 +) + +// lineKeeper collects the lines headed by a queue id and writes them in +// batches. It is used from the follow loop only, so it needs no locking. +type lineKeeper struct { + st StatusStore + // buf holds the lines not yet written; once says they were read while + // catching up and may have been read before the last stop. + buf []store.SendLogLine + once bool + // port25 holds the queue ids the port-25 smtpd opened: mail offered by + // strangers, which never gets a send-log row. Its lines are not kept - what + // a stranger can make Postfix log is not something to store on his say-so + // and prune later. An id leaves when qmgr removes the message or when + // another listener opens a message under it (Postfix reuses ids). + port25 map[string]struct{} +} + +func newLineKeeper(st StatusStore) *lineKeeper { + return &lineKeeper{st: st, port25: make(map[string]struct{})} +} + +// add takes one mail.log line and keeps it if it is headed by the queue id of +// a message that did not come in on port 25. It does not ask whether a +// send-log row exists for the id: for a message's first lines none does yet, +// and the lines of messages that never get one - the bounces Postfix +// generates, port-25 mail met in mid-flight after a restart - are removed by +// pruneLines. // // catchingUp is as in inboundJournal.handle: the line may have been read // before the last stop, and is then written only if it is not there already. // The line is dated by its own stamp - after the panel was down, that and now // are hours apart, and a line is matched to its row by time. -func storeLine(st StatusStore, line string, now time.Time, catchingUp bool) { +func (k *lineKeeper) add(line string, now time.Time, catchingUp bool) { at, rest := lineTime(line, now) m := queueLineRe.FindStringSubmatch(rest) if m == nil { return } - l := store.SendLogLine{QueueID: m[1], At: at, Line: line} - var err error - if catchingUp { - _, err = st.InsertSendLogLineOnce(l) - } else { - err = st.InsertSendLogLine(l) + program, queueID, text := m[1], m[2], m[3] + + if smtpdProgramRe.MatchString(program) && len(text) >= 7 && text[:7] == "client=" { + if program == port25Program { + k.rememberPort25(queueID) + } else { + delete(k.port25, queueID) + } } - if err != nil { - log.Printf("log-tailer: keep line of %s: %v", m[1], err) + if _, inbound := k.port25[queueID]; inbound { + if program == "postfix/qmgr" && text == "removed" { + delete(k.port25, queueID) + } + return } + + if catchingUp != k.once { + k.flush() + k.once = catchingUp + } + k.buf = append(k.buf, store.SendLogLine{QueueID: queueID, At: at, Line: line}) + if len(k.buf) >= lineBatch { + k.flush() + } +} + +func (k *lineKeeper) rememberPort25(queueID string) { + if len(k.port25) >= port25IDs { + // Which one goes does not matter: the cost of forgetting an id is its + // remaining lines kept until the prune. + for id := range k.port25 { + delete(k.port25, id) + break + } + } + k.port25[queueID] = struct{}{} +} + +// flush writes the lines held. The follow loop calls it after every pass over +// the log and before it saves its place, so the place saved is never ahead of +// what was written. Lines that cannot be written are dropped and said so: the +// tailer goes on, and a page without them falls back to the log file. +func (k *lineKeeper) flush() { + if len(k.buf) == 0 { + return + } + if _, err := k.st.InsertSendLogLines(k.buf, k.once); err != nil { + log.Printf("log-tailer: keep %d log lines: %v", len(k.buf), err) + } + k.buf = k.buf[:0] } // pruneLines removes the stored lines no send-log row carries the queue id of: diff --git a/internal/logtail/lines_test.go b/internal/logtail/lines_test.go index d9c9310..325b56a 100644 --- a/internal/logtail/lines_test.go +++ b/internal/logtail/lines_test.go @@ -20,7 +20,7 @@ func TestQueueLineHead(t *testing.T) { {"port-25 smtpd", "mail postfix/inbound/smtpd[41]: 4A1B2C3D: client=mx.example.net[203.0.113.9]", "4A1B2C3D"}, {"cleanup", "mail postfix/cleanup[52]: 41E862C00D9E: message-id=<1@example.com>", "41E862C00D9E"}, {"qmgr removes it", "mail postfix/qmgr[10]: 41E862C00D9E: removed", "41E862C00D9E"}, - {"another program", "mail opendkim[12]: 41E862C00D9E: DKIM-Signature field added (s=sp, d=example.com)", "41E862C00D9E"}, + {"a bounce is sent: the id at the head, not the new message's", "mail postfix/bounce[30]: 41E862C00D9E: sender non-delivery notification: 5F00AA12", "41E862C00D9E"}, {"long queue id", "mail postfix/qmgr[10]: 3sJ8Xh2Cgqz4Xy1: removed", "3sJ8Xh2Cgqz4Xy1"}, {"refused before queueing", "mail postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from mx.example.net[203.0.113.9]: 550 5.1.1 : Recipient address rejected; from= to= proto=ESMTP helo=", ""}, {"a warning", "mail postfix/smtpd[10]: warning: hostname mx.example.net does not resolve to address 203.0.113.9", ""}, @@ -38,7 +38,7 @@ func TestQueueLineHead(t *testing.T) { for _, c := range cases { got := "" if m := queueLineRe.FindStringSubmatch(c.rest); m != nil { - got = m[1] + got = m[2] } if got != c.want { t.Errorf("%s: queue id %q, want %q", c.name, got, c.want) @@ -47,14 +47,19 @@ func TestQueueLineHead(t *testing.T) { } // A line is dated by its own stamp, not by when the tailer got to it, and is -// kept as the log has it. -func TestStoreLineDatesByTheLine(t *testing.T) { +// kept as the log has it. Nothing is written until the pass over the log ends. +func TestLineKeeperDatesByTheLine(t *testing.T) { cs := &captureStore{} + k := newLineKeeper(cs) now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) line := "Oct 12 09:00:00 mail postfix/qmgr[7]: 41E862C00D9E: removed" - storeLine(cs, line, now, false) - storeLine(cs, "Oct 12 09:00:01 mail postfix/smtpd[7]: connect from unknown[203.0.113.7]", now, false) + k.add(line, now, false) + k.add("Oct 12 09:00:01 mail postfix/smtpd[7]: connect from unknown[203.0.113.7]", now, false) + if len(cs.lines) != 0 { + t.Fatalf("written before the pass ended: %v", cs.kept()) + } + k.flush() if len(cs.lines) != 1 { t.Fatalf("kept %d lines, want 1: %v", len(cs.lines), cs.kept()) @@ -68,6 +73,91 @@ func TestStoreLineDatesByTheLine(t *testing.T) { } } +// A backlog is written in batches of lineBatch, not held whole, and a pass +// that changes from catching up to following does not write the later lines +// under the earlier rule. +func TestLineKeeperWritesInBatches(t *testing.T) { + cs := &captureStore{} + k := newLineKeeper(cs) + now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) + line := func(i int) string { + return "host postfix/qmgr[7]: 41E862C00D9E: from=, size=" + itoa(i) + ", nrcpt=1 (queue active)" + } + + for i := 0; i < lineBatch+10; i++ { + k.add(line(i), now, true) + } + if len(cs.lines) != lineBatch || cs.batches != 1 { + t.Fatalf("after %d lines: %d written in %d batches, want %d in 1", lineBatch+10, len(cs.lines), cs.batches, lineBatch) + } + + // The same line again, now as one read while following: it is a new line, + // and the ten still held from the catch-up are written first, under theirs. + k.add(line(0), now, false) + k.flush() + if got, want := len(cs.lines), lineBatch+11; got != want { + t.Errorf("%d lines written, want %d", got, want) + } + if cs.batches != 3 { + t.Errorf("%d batches, want 3", cs.batches) + } +} + +// Mail that comes in on port 25 is a stranger's and never gets a send-log row: +// none of its lines are kept, from the smtpd that opened the queue id to +// qmgr's "removed". The id is then anybody's again. +func TestLineKeeperLeavesOutPort25Mail(t *testing.T) { + cs := &captureStore{} + k := newLineKeeper(cs) + now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) + + const ( + outClient = "host postfix/submission/smtpd[10]: 41E862C00D9E: client=app.example.com[203.0.113.7], sasl_method=PLAIN, sasl_username=app" + outSent = "host postfix/smtp[26]: 41E862C00D9E: to=, relay=mx.example.net[203.0.113.9]:25, dsn=2.0.0, status=sent (250 OK)" + reused = "host postfix/submission/smtpd[10]: 4A1B2C3D: client=app.example.com[203.0.113.7], sasl_method=PLAIN, sasl_username=app" + bounce = "host postfix/cleanup[52]: 5F00AA12: message-id=<20261012180000.5F00AA12@mail.example.com>" + ) + for _, line := range []string{ + "host postfix/inbound/smtpd[41]: 4A1B2C3D: client=mx.example.net[198.51.100.9]", + outClient, + "host postfix/cleanup[52]: 4A1B2C3D: message-id=", + "host postfix/qmgr[10]: 4A1B2C3D: from=, size=900, nrcpt=1 (queue active)", + outSent, + "host postfix/smtp[27]: 4A1B2C3D: to=, relay=upstream.example.com[192.0.2.1]:25, dsn=2.0.0, status=sent (250 OK)", + "host postfix/qmgr[10]: 4A1B2C3D: removed", + // The id comes round again for a submitted message, and a bounce has no + // smtpd at all: both are kept. + reused, + bounce, + } { + k.add(line, now, false) + } + k.flush() + + want := []string{"41E862C00D9E|" + outClient, "41E862C00D9E|" + outSent, "4A1B2C3D|" + reused, "5F00AA12|" + bounce} + if got := cs.kept(); strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Errorf("kept:\n%s\nwant:\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) + } + + // A port-25 id that a submission listener opens before qmgr's "removed" was + // seen (the line was lost, the panel restarted) is the submitted message's. + k.add("host postfix/inbound/smtpd[41]: 6C00BB34: client=mx.example.net[198.51.100.9]", now, false) + again := "host postfix/smtps/smtpd[11]: 6C00BB34: client=app.example.com[203.0.113.7], sasl_method=PLAIN, sasl_username=app" + k.add(again, now, false) + k.flush() + if got := cs.kept(); got[len(got)-1] != "6C00BB34|"+again { + t.Errorf("a reused port-25 id was not kept for the submitted message: %v", got[len(got)-1]) + } + + // What is remembered of port 25 has a ceiling. + for i := 0; i < port25IDs+50; i++ { + k.rememberPort25("7D" + itoa(100000+i)) + } + if len(k.port25) > port25IDs { + t.Errorf("%d port-25 ids remembered, ceiling is %d", len(k.port25), port25IDs) + } +} + // The tailer keeps queue-id lines as it follows the log, and after an unclean // stop - its saved place lies before lines it had already read - it does not // keep them a second time. @@ -115,13 +205,13 @@ func TestRunKeepsQueueLinesOnce(t *testing.T) { } } -// Lines without a send-log row are removed at the start, with every send-log -// sweep and on their own shorter cadence, and only once they are old enough -// for the row to have been written. +// Lines without a send-log row are removed at the start and after every +// send-log sweep, and only once they are old enough for the row to have been +// written. func TestRetentionLoopPrunesLines(t *testing.T) { - old := linePruneInterval - linePruneInterval = 20 * time.Millisecond - t.Cleanup(func() { linePruneInterval = old }) + old := retentionInterval + retentionInterval = 20 * time.Millisecond + t.Cleanup(func() { retentionInterval = old }) cs := &captureStore{} ctx, cancel := context.WithCancel(context.Background()) diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go index 183b72c..52bd0a9 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -32,8 +32,7 @@ type StatusStore interface { UpdateStatus(queueID, recipient, status string) (int64, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) DeleteSendLogBefore(cutoff time.Time) (int64, error) - InsertSendLogLine(l store.SendLogLine) error - InsertSendLogLineOnce(l store.SendLogLine) (bool, error) + InsertSendLogLines(lines []store.SendLogLine, once bool) (int, error) PruneSendLogLines(cutoff time.Time) (int64, error) LogtailState(path string) (store.LogtailState, bool, error) SaveLogtailState(path string, st store.LogtailState) error @@ -111,8 +110,9 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) { // Run follows path and updates send-log statuses until ctx is cancelled, while // a background sweep prunes rows older than retention(). Reading resumes at // the offset the previous run persisted, so a restart parses the delivery lines -// written while the panel was down. Every line headed by a queue id is kept -// for the message's page (lines.go). With inbound set it also writes the +// written while the panel was down. The lines headed by a queue id are kept +// for the message's page, except those of mail that came in on port 25 +// (lines.go). With inbound set it also writes the // inbound journal's rows for refused mail and prunes that journal // (inbound.go). It returns nil on a clean shutdown. func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays, inbound *Inbound) error { @@ -130,7 +130,8 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa caughtUp := make(chan struct{}) go reconcileLoop(ctx, st, caughtUp) - return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) { + keeper := newLineKeeper(st) + return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, keeper.flush, func(line string) { // Until the first read reaches end-of-file the lines are those written // while the panel was down - and, after an unclean stop, some it had // read already. @@ -144,7 +145,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa if journal != nil { journal.handle(line, now, catchingUp) } - storeLine(st, line, now, catchingUp) + keeper.add(line, now, catchingUp) queueID, recipient, status, ok := parseDelivery(line) if !ok { return @@ -225,8 +226,9 @@ func reconcile(st StatusStore, cutoff time.Time) { } // retentionLoop prunes expired send-log rows immediately and then periodically, -// and after each sweep the stored log lines those rows leave behind. Lines -// without a row are also removed on a shorter cadence of their own (lines.go). +// and after each sweep the stored log lines no row carries the queue id of: +// those the deleted rows leave behind, and those that never had a row +// (lines.go). func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) { prune := func() { retentionDays := retention() @@ -248,8 +250,6 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) pruneLines(st) t := time.NewTicker(retentionInterval) defer t.Stop() - lines := time.NewTicker(linePruneInterval) - defer lines.Stop() for { select { case <-ctx.Done(): @@ -257,8 +257,6 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) case <-t.C: prune() pruneLines(st) - case <-lines.C: - pruneLines(st) } } } @@ -460,7 +458,11 @@ func SplitTimestamp(line string) (stamp, rest string) { // caughtUp is closed after the first read that reaches end-of-file, which is // the point where every line the panel missed while it was down has been // handled. -func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{}, handle func(string)) error { +// +// flush is called at the end of every pass over the file, before the read +// position is saved: a handler that gathers lines to write them together +// writes them there, so the saved position is never ahead of what was written. +func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{}, flush func(), handle func(string)) error { var ( f *os.File r *bufio.Reader @@ -510,6 +512,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ }() drain := func() { + defer flush() for { line, err := r.ReadString('\n') if err == io.EOF { diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go index 3287031..d4c0756 100644 --- a/internal/logtail/logtail_test.go +++ b/internal/logtail/logtail_test.go @@ -150,27 +150,31 @@ type captureStore struct { // lines are the log lines kept for messages' pages; pruned is the cutoff of // each PruneSendLogLines call. - lines []store.SendLogLine - pruned []time.Time + lines []store.SendLogLine + batches int + pruned []time.Time } -func (c *captureStore) InsertSendLogLine(l store.SendLogLine) error { +// InsertSendLogLines keeps what the store would: with once, a line is not +// kept a second time under its queue id. batches counts the calls. +func (c *captureStore) InsertSendLogLines(lines []store.SendLogLine, once bool) (int, error) { c.mu.Lock() defer c.mu.Unlock() - c.lines = append(c.lines, l) - return nil -} - -func (c *captureStore) InsertSendLogLineOnce(l store.SendLogLine) (bool, error) { - c.mu.Lock() - defer c.mu.Unlock() - for _, have := range c.lines { - if have.QueueID == l.QueueID && have.Line == l.Line { - return false, nil + c.batches++ + wrote := 0 +next: + for _, l := range lines { + if once { + for _, have := range c.lines { + if have.QueueID == l.QueueID && have.Line == l.Line { + continue next + } + } } + c.lines = append(c.lines, l) + wrote++ } - c.lines = append(c.lines, l) - return true, nil + return wrote, nil } func (c *captureStore) PruneSendLogLines(cutoff time.Time) (int64, error) { diff --git a/internal/logtail/retention_test.go b/internal/logtail/retention_test.go index 470db3d..9e277da 100644 --- a/internal/logtail/retention_test.go +++ b/internal/logtail/retention_test.go @@ -30,10 +30,8 @@ func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, erro return 0, nil } -func (s *retentionProbeStore) InsertSendLogLine(store.SendLogLine) error { return nil } - -func (s *retentionProbeStore) InsertSendLogLineOnce(store.SendLogLine) (bool, error) { - return false, nil +func (s *retentionProbeStore) InsertSendLogLines([]store.SendLogLine, bool) (int, error) { + return 0, nil } func (s *retentionProbeStore) PruneSendLogLines(time.Time) (int64, error) { return 0, nil } diff --git a/internal/postfix/retry.go b/internal/postfix/retry.go index 7354518..bf86fcb 100644 --- a/internal/postfix/retry.go +++ b/internal/postfix/retry.go @@ -130,8 +130,9 @@ func (p RetryPolicy) BackoffCap() string { } // LongestLifetime is how long Postfix may go on working on a message: the -// larger of the queue lifetime for mail and for the bounces it generates. It -// bounds which stored log lines can be about a send-log row (store.SendLogLines). +// larger of the queue lifetime for ordinary mail and for mail from the null +// sender, which an application may submit too. It bounds which stored log +// lines can be about a send-log row (store.SendLogLines). func (p RetryPolicy) LongestLifetime() time.Duration { if p.BounceQueueLifetime > p.MaximalQueueLifetime { return p.BounceQueueLifetime diff --git a/internal/store/sendloglines.go b/internal/store/sendloglines.go index 1f7c0f7..cf63279 100644 --- a/internal/store/sendloglines.go +++ b/internal/store/sendloglines.go @@ -3,6 +3,7 @@ package store import ( "fmt" "time" + "unicode/utf8" ) // SendLogLine is one mail.log line Postfix wrote under a queue id, kept so a @@ -25,44 +26,62 @@ const SendLogLineMax = 4096 // is accepted, and the message body can take its time after that. const sendLogLineBefore = time.Hour +// cutSendLogLine cuts a line to SendLogLineMax bytes, and not through the +// middle of a character. func cutSendLogLine(line string) string { - if len(line) > SendLogLineMax { - return line[:SendLogLineMax] + if len(line) <= SendLogLineMax { + return line } - return line + end := SendLogLineMax + for end > 0 && !utf8.RuneStart(line[end]) { + end-- + } + return line[:end] } -// InsertSendLogLine stores one line. The tailer calls it for every line headed -// by a queue id, whether or not a send-log row exists for it yet. -func (s *Store) InsertSendLogLine(l SendLogLine) error { - _, err := s.db.Exec( - `INSERT INTO send_log_lines (queue_id, logged_at, line) VALUES (?, ?, ?)`, - l.QueueID, l.At.UTC().Format(time.RFC3339), cutSendLogLine(l.Line), - ) - if err != nil { - return fmt.Errorf("insert send_log_lines: %w", err) +// InsertSendLogLines stores lines in one transaction and returns how many it +// wrote. The tailer calls it with what it read in one pass over the log, +// whether or not send-log rows exist for the queue ids yet: a commit per line +// would be a disk sync per line, on the one connection the journal milter +// writes through. +// +// With once set, a line is written only if the same line is not there already +// under its queue id. That is for the lines the tailer reads again after an +// unclean stop; a line carries its time and its process id, so two that are +// equal are one line read twice. +func (s *Store) InsertSendLogLines(lines []SendLogLine, once bool) (int, error) { + if len(lines) == 0 { + return 0, nil } - return nil -} - -// InsertSendLogLineOnce stores a line unless the same one is already there, -// and reports whether it wrote. It is for the lines the tailer reads again -// after an unclean stop; a line carries its time and its process id, so two -// that are equal are one line read twice. -func (s *Store) InsertSendLogLineOnce(l SendLogLine) (bool, error) { - line := cutSendLogLine(l.Line) - res, err := s.db.Exec( - `INSERT INTO send_log_lines (queue_id, logged_at, line) - SELECT ?, ?, ? - WHERE NOT EXISTS (SELECT 1 FROM send_log_lines WHERE queue_id = ? AND line = ?)`, - l.QueueID, l.At.UTC().Format(time.RFC3339), line, - l.QueueID, line, - ) - if err != nil { - return false, fmt.Errorf("insert send_log_lines once: %w", err) + query := `INSERT INTO send_log_lines (queue_id, logged_at, line) VALUES (?1, ?2, ?3)` + if once { + query = `INSERT INTO send_log_lines (queue_id, logged_at, line) + SELECT ?1, ?2, ?3 + WHERE NOT EXISTS (SELECT 1 FROM send_log_lines WHERE queue_id = ?1 AND line = ?3)` } - n, _ := res.RowsAffected() - return n > 0, nil + tx, err := s.db.Begin() + if err != nil { + return 0, fmt.Errorf("insert send_log_lines: %w", err) + } + defer tx.Rollback() + stmt, err := tx.Prepare(query) + if err != nil { + return 0, fmt.Errorf("insert send_log_lines: %w", err) + } + defer stmt.Close() + wrote := 0 + for _, l := range lines { + res, err := stmt.Exec(l.QueueID, l.At.UTC().Format(time.RFC3339), cutSendLogLine(l.Line)) + if err != nil { + return 0, fmt.Errorf("insert send_log_lines: %w", err) + } + n, _ := res.RowsAffected() + wrote += int(n) + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("insert send_log_lines: %w", err) + } + return wrote, nil } // SendLogLines returns the stored lines of the message a send-log row is @@ -110,14 +129,19 @@ func (s *Store) SendLogLines(row SendLogRow, lifetime time.Duration, n int) ([]S // PruneSendLogLines deletes the lines dated before cutoff whose queue id has // no send-log row, and returns how many went. That one rule is the whole of // the table's retention: it removes the lines of rows the send-log sweep has -// deleted, and the lines of messages that never had a row - inbound relay -// mail, the bounces Postfix generates. cutoff leaves alone the lines of a -// message still being received, whose row the journal milter has yet to write. +// deleted, and the lines of messages that never had a row - the bounces +// Postfix generates. cutoff leaves alone the lines of a message still being +// received, whose row the journal milter has yet to write. +// +// The send log is asked once per queue id, not once per line: the ids are read +// off the index, and a message is several lines. func (s *Store) PruneSendLogLines(cutoff time.Time) (int64, error) { res, err := s.db.Exec( `DELETE FROM send_log_lines WHERE logged_at < ? - AND NOT EXISTS (SELECT 1 FROM send_log WHERE send_log.queue_id = send_log_lines.queue_id)`, + AND queue_id IN ( + SELECT q.queue_id FROM (SELECT DISTINCT queue_id FROM send_log_lines) AS q + WHERE NOT EXISTS (SELECT 1 FROM send_log WHERE send_log.queue_id = q.queue_id))`, cutoff.UTC().Format(time.RFC3339), ) if err != nil { diff --git a/internal/store/sendloglines_test.go b/internal/store/sendloglines_test.go index 9770e77..996d868 100644 --- a/internal/store/sendloglines_test.go +++ b/internal/store/sendloglines_test.go @@ -4,6 +4,7 @@ import ( "strings" "testing" "time" + "unicode/utf8" ) func sendLogLineTexts(lines []SendLogLine) []string { @@ -28,15 +29,13 @@ func countSendLogLines(t *testing.T, s *Store) int { func TestSendLogLinesOfOneMessage(t *testing.T) { s := openTestStore(t) at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) - for i, l := range []SendLogLine{ + if n, err := s.InsertSendLogLines([]SendLogLine{ {QueueID: "41E862C00D9E", At: at, Line: "client=app"}, {QueueID: "5F00AA", At: at, Line: "someone else's"}, {QueueID: "41E862C00D9E", At: at.Add(time.Second), Line: "status=sent"}, {QueueID: "41E862C00D9E", At: at.Add(2 * time.Second), Line: "removed"}, - } { - if err := s.InsertSendLogLine(l); err != nil { - t.Fatalf("insert line %d: %v", i, err) - } + }, false); err != nil || n != 4 { + t.Fatalf("insert: wrote %d, %v", n, err) } row := SendLogRow{QueueID: "41E862C00D9E", CreatedAt: at} @@ -75,15 +74,13 @@ func TestSendLogLinesBoundedByTheRowsLifetime(t *testing.T) { first := time.Date(2026, 9, 1, 9, 0, 0, 0, time.UTC) second := first.AddDate(0, 0, 30) lifetime := 6 * 24 * time.Hour - for _, l := range []SendLogLine{ + if _, err := s.InsertSendLogLines([]SendLogLine{ {QueueID: "ABC123", At: first.Add(-30 * time.Minute), Line: "first: client"}, {QueueID: "ABC123", At: first.Add(5 * 24 * time.Hour), Line: "first: expired"}, {QueueID: "ABC123", At: second.Add(-2 * time.Hour), Line: "between the two"}, {QueueID: "ABC123", At: second, Line: "second: sent"}, - } { - if err := s.InsertSendLogLine(l); err != nil { - t.Fatalf("insert: %v", err) - } + }, false); err != nil { + t.Fatalf("insert: %v", err) } lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: first}, lifetime, 200) @@ -103,45 +100,52 @@ func TestSendLogLinesBoundedByTheRowsLifetime(t *testing.T) { } // After an unclean stop the tailer reads some lines twice; the second reading -// writes nothing. The same text under another queue id is another line. -func TestInsertSendLogLineOnce(t *testing.T) { +// writes nothing, also when the line comes twice in one batch. The same text +// under another queue id is another line. Without once, what is given is +// written. +func TestInsertSendLogLinesOnce(t *testing.T) { s := openTestStore(t) l := SendLogLine{QueueID: "ABC123", At: time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC), Line: "Oct 12 09:00:00 mail postfix/qmgr[7]: ABC123: removed"} - - for i, want := range []bool{true, false} { - wrote, err := s.InsertSendLogLineOnce(l) - if err != nil || wrote != want { - t.Errorf("reading %d: wrote = %v, %v; want %v", i+1, wrote, err, want) - } - } other := l other.QueueID = "DEF456" - if wrote, err := s.InsertSendLogLineOnce(other); err != nil || !wrote { - t.Errorf("another queue id: wrote = %v, %v", wrote, err) + + if n, err := s.InsertSendLogLines([]SendLogLine{l, l, other}, true); err != nil || n != 2 { + t.Errorf("first reading wrote %d, %v; want 2", n, err) + } + if n, err := s.InsertSendLogLines([]SendLogLine{l, other}, true); err != nil || n != 0 { + t.Errorf("second reading wrote %d, %v; want 0", n, err) } if n := countSendLogLines(t, s); n != 2 { t.Errorf("%d lines stored, want 2", n) } + if n, err := s.InsertSendLogLines([]SendLogLine{l}, false); err != nil || n != 1 { + t.Errorf("a line as it is written: wrote %d, %v; want 1", n, err) + } + if n, err := s.InsertSendLogLines(nil, true); err != nil || n != 0 { + t.Errorf("nothing to write: %d, %v", n, err) + } } // The far end writes the reply a delivery line quotes; what is kept of a line -// has a ceiling, and a line read twice is still recognised past it. +// has a ceiling, not set through the middle of a character, and a line read +// twice is still recognised past it. func TestSendLogLineIsCut(t *testing.T) { s := openTestStore(t) at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) - l := SendLogLine{QueueID: "ABC123", At: at, Line: strings.Repeat("x", SendLogLineMax+500)} - if err := s.InsertSendLogLine(l); err != nil { + // Two bytes a character, starting at an odd offset: the ceiling falls inside one. + l := SendLogLine{QueueID: "ABC123", At: at, Line: "x" + strings.Repeat("é", SendLogLineMax)} + if _, err := s.InsertSendLogLines([]SendLogLine{l}, false); err != nil { t.Fatalf("insert: %v", err) } - if wrote, err := s.InsertSendLogLineOnce(l); err != nil || wrote { - t.Errorf("the same long line again: wrote = %v, %v", wrote, err) + if n, err := s.InsertSendLogLines([]SendLogLine{l}, true); err != nil || n != 0 { + t.Errorf("the same long line again: wrote %d, %v", n, err) } lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: at}, time.Hour, 10) if err != nil || len(lines) != 1 { t.Fatalf("lines = %d, %v", len(lines), err) } - if len(lines[0].Line) != SendLogLineMax { - t.Errorf("kept %d bytes, want %d", len(lines[0].Line), SendLogLineMax) + if got := lines[0].Line; len(got) != SendLogLineMax-1 || !utf8.ValidString(got) { + t.Errorf("kept %d bytes (valid UTF-8: %v), want %d whole characters", len(got), utf8.ValidString(got), SendLogLineMax-1) } } @@ -156,14 +160,16 @@ func TestPruneSendLogLines(t *testing.T) { if err := s.InsertQueued(SendLogEntry{QueueID: "KEPT01", Domain: "example.com", To: "a@example.net"}); err != nil { t.Fatalf("insert row: %v", err) } - for _, l := range []SendLogLine{ + // A row the milter refused has no queue id; it keeps no line alive. + if err := s.InsertRejected(SendLogEntry{Domain: "example.com", From: "a@example.com"}); err != nil { + t.Fatalf("insert rejected row: %v", err) + } + if _, err := s.InsertSendLogLines([]SendLogLine{ {QueueID: "KEPT01", At: old, Line: "has a row"}, {QueueID: "INBOUND1", At: old, Line: "never had a row"}, {QueueID: "ARRIVING", At: now.Add(-time.Minute), Line: "its row is not written yet"}, - } { - if err := s.InsertSendLogLine(l); err != nil { - t.Fatalf("insert: %v", err) - } + }, false); err != nil { + t.Fatalf("insert: %v", err) } n, err := s.PruneSendLogLines(now.Add(-time.Hour)) diff --git a/internal/web/handlers/handlers_monitor_test.go b/internal/web/handlers/handlers_monitor_test.go index 5aa40bc..09696b8 100644 --- a/internal/web/handlers/handlers_monitor_test.go +++ b/internal/web/handlers/handlers_monitor_test.go @@ -197,15 +197,13 @@ func TestDeliveryPageShowsStoredLogLinesAfterRotation(t *testing.T) { h, row := serverWithDelivery(t) h.cfg.RetryPolicy = postfix.DefaultRetryPolicy() h.cfg.MailLogPath = writeMailLog(t, "2026-08-04T00:00:01.000000+00:00 host postfix/qmgr[10]: 99999999: removed") - for _, l := range []store.SendLogLine{ + if _, err := h.store.InsertSendLogLines([]store.SendLogLine{ {QueueID: row.QueueID, At: row.CreatedAt.AddDate(0, 0, -40), Line: "2026-06-24T05:00:00.000000+00:00 host postfix/qmgr[10]: 4A1B2C3D: an earlier message under the same id"}, {QueueID: row.QueueID, At: row.CreatedAt.Add(-time.Second), Line: "2026-08-03T05:15:52.219218+00:00 host postfix/smtpd[20]: 4A1B2C3D: client=mail.example.com[203.0.113.4]"}, {QueueID: "99999999", At: row.CreatedAt, Line: "2026-08-03T05:15:52.300000+00:00 host postfix/qmgr[10]: 99999999: from=, size=500, nrcpt=1 (queue active)"}, {QueueID: row.QueueID, At: row.CreatedAt.Add(11 * time.Second), Line: "2026-08-03T05:16:03.884210+00:00 host postfix/smtp[26]: 4A1B2C3D: to=, dsn=2.0.0, status=sent (250 OK)"}, - } { - if err := h.store.InsertSendLogLine(l); err != nil { - t.Fatalf("InsertSendLogLine: %v", err) - } + }, false); err != nil { + t.Fatalf("InsertSendLogLines: %v", err) } out := getBody(t, h.HandleDelivery, "/outbound/log/"+itoa(row.ID))