delivery-log-storage: findings of the technical and security reviews
test / test (push) Canceled after 0s

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 <noreply@anthropic.com>
This commit is contained in:
mix
2026-10-12 10:41:04 +03:00
co-authored by Claude Opus 5.5
parent 815236a607
commit a990e7cc23
14 changed files with 456 additions and 178 deletions
+17 -6
View File
@@ -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 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 once, to read the lines of a row and to prune
([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)). ([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)).
- **The log tailer keeps what Postfix writes about queued messages.** Every - **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, as `mail.log` line headed by a queue id is stored with the line's own time, one
the tailer reads it; after an unclean stop a line read twice is stored once. transaction per pass over the log; after an unclean stop a line read twice
Lines whose queue id has no send-log row - the row was pruned, or the is stored once. Mail that came in on port 25 is left out: a stranger's
message never had one (inbound relay mail, bounces) - are deleted once they message never has a send-log row, and what he can make Postfix log is not
are an hour old, with every send-log sweep and hourly between. 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 ### 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 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. 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 ## [2.1.0] - 2026-10-12
### Security ### Security
+22 -8
View File
@@ -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 Postfix wrote about it, so the tailer keeps those lines in the database as it
reads them: 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 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 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 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 same place and are not ids; the pattern accepts Postfix's short form
(capital hexadecimal) and its long form (no vowels). (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=`, - **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 `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. 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 Lines that never get a row — bounces Postfix generates, port-25 mail met in
generates — are kept too, for a while. 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 - **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 row carries is deleted (`PruneSendLogLines`), at start and after
send-log sweep and hourly between. That removes the lines of rows the sweep every six-hourly send-log sweep. That removes the lines of rows the sweep
deleted and the lines that never had a row. 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 - **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 (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 recipients is several rows under one id, and each row's page shows all of
the message's lines. the message's lines.
- **Read twice, stored once.** The read offset is saved every few seconds, so - **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 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. 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 - **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 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 **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 serving panel pages on request: `TailLines` (the last *n* lines, for
+51 -19
View File
@@ -29,20 +29,31 @@ rotation, while the row itself lives for ninety days.
### What is stored ### 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 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 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: written, cut to 4096 bytes on a character boundary. The head is matched from
the tail is the far end's words. `NOQUEUE` and `warning` are not queue ids - the start of the line: the tail is the far end's words. `NOQUEUE` and
the pattern takes Postfix's short form (hexadecimal) and its long form (no `warning` are not queue ids - the pattern takes Postfix's short form
vowels) and nothing else. (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 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 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 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 would drop exactly them.
relay mail, bounces Postfix generates - are stored too and removed by the
prune below within the hour or two. **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 ### 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 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`, 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 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 id once the message has left the queue - the id is the microsecond of arrival
falls outside that window. This is the bound the roadmap item asked for. 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 ### Retention
Lines live as long as a send-log row with their queue id does. One rule covers 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 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, 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 and the lines of messages that never had a row. It runs at start and after
sweep and once an hour between. 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 There is no cap of its own: with port-25 mail left out, the table is bounded
queued messages, which its own rate limits bound. by what authenticated applications send, which level 1 and level 2 limit.
### After an unclean stop ### 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 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. 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
The page reads the table: the latest 200 lines of the row, oldest first. When 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 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 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. 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 - 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 ## Done when
@@ -143,10 +175,10 @@ Target version cut: **`2.2.0`** (MINOR). One commit per step;
[development.md](../development.md) § Plan checklists. [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] 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] 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] 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] [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`) - [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** - [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)
- [ ] Security review — **Fable** - [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)
+1 -1
View File
@@ -36,7 +36,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md).
| structured-logging | `log/slog` with levels and fields | candidate | — | — | | structured-logging | `log/slog` with levels and fields | candidate | — | — |
| review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — | | review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — |
| panel-notifications | E-mail notifications about important events | 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 | — | — | | password-reset | Password reset by e-mail | candidate | — | — |
| phone-desktop-mode | A phone browser in "desktop site" mode gets the phone navigation | 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) | | panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) |
+20 -1
View File
@@ -244,6 +244,11 @@ operator can act on it.
private key from `certs/` when present; a domain export carries the DKIM key private key from `certs/` when present; a domain export carries the DKIM key
and **working** application passwords in the clear (otherwise a transfer and **working** application passwords in the clear (otherwise a transfer
without recreating credentials would be impossible). 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 - Both downloads are encrypted with a password unless the box on the form is
unticked (it is ticked by default): unticked (it is ticked by default):
scrypt (N=2¹⁵, r=8, p=1) → AES-256-GCM, streamed in 64 KiB chunks, each 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 the file is inaccessible to others. The log holds envelope addresses and
client IPs, but neither message bodies nor headers; it is excluded from client IPs, but neither message bodies nor headers; it is excluded from
backups (`log/` is skipped) so that a dump stays state rather than 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 - **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 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 and the port is 1–65535; loopback, private and link-local addresses are
+101 -23
View File
@@ -27,44 +27,122 @@ import (
// "warning:" stand there too. A queue id is hexadecimal in capitals, or, with // "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 // enable_long_queue_ids, digits and letters without vowels; neither word is
// either. // 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 // smtpdProgramRe tells an smtpd from the other programs, whatever listener it
// row carries its queue id. The first lines of a message are logged when it // serves: "postfix/smtpd", "postfix/submission/smtpd", "postfix/inbound/smtpd".
// arrives and its row is written at end-of-message; an hour is a message body var smtpdProgramRe = regexp.MustCompile(`^postfix/(?:[a-z0-9-]+/)?smtpd$`)
// sent slowly.
const lineGrace = time.Hour
// linePruneInterval is how often lines without a row are removed between two // port25Program is the smtpd of the inbound relay's listener, which
// send-log sweeps. With the inbound relay on, most of what Postfix logs under // build/postfix-config.sh names so that its lines are told from those of the
// a queue id is mail that never gets a send-log row; it is not left to pile up // submission ports.
// for six hours. A var so tests can shorten it. const port25Program = "postfix/inbound/smtpd"
var linePruneInterval = time.Hour
// storeLine keeps line if it is headed by a queue id. It does not ask whether const (
// a send-log row exists for the id: for a message's first lines none does yet, // lineGrace is how old a line must be before the prune asks whether a
// and the lines of messages that never get one are removed by pruneLines. // 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 // 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. // 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 // 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. // 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) at, rest := lineTime(line, now)
m := queueLineRe.FindStringSubmatch(rest) m := queueLineRe.FindStringSubmatch(rest)
if m == nil { if m == nil {
return return
} }
l := store.SendLogLine{QueueID: m[1], At: at, Line: line} program, queueID, text := m[1], m[2], m[3]
var err error
if catchingUp { if smtpdProgramRe.MatchString(program) && len(text) >= 7 && text[:7] == "client=" {
_, err = st.InsertSendLogLineOnce(l) if program == port25Program {
} else { k.rememberPort25(queueID)
err = st.InsertSendLogLine(l) } else {
delete(k.port25, queueID)
}
} }
if err != nil { if _, inbound := k.port25[queueID]; inbound {
log.Printf("log-tailer: keep line of %s: %v", m[1], err) 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: // pruneLines removes the stored lines no send-log row carries the queue id of:
+102 -12
View File
@@ -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"}, {"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"}, {"cleanup", "mail postfix/cleanup[52]: 41E862C00D9E: message-id=<1@example.com>", "41E862C00D9E"},
{"qmgr removes it", "mail postfix/qmgr[10]: 41E862C00D9E: removed", "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"}, {"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 <n@in.example.com>: Recipient address rejected; from=<a@example.net> to=<n@in.example.com> proto=ESMTP helo=<mx.example.net>", ""}, {"refused before queueing", "mail postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from mx.example.net[203.0.113.9]: 550 5.1.1 <n@in.example.com>: Recipient address rejected; from=<a@example.net> to=<n@in.example.com> proto=ESMTP helo=<mx.example.net>", ""},
{"a warning", "mail postfix/smtpd[10]: warning: hostname mx.example.net does not resolve to address 203.0.113.9", ""}, {"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 { for _, c := range cases {
got := "" got := ""
if m := queueLineRe.FindStringSubmatch(c.rest); m != nil { if m := queueLineRe.FindStringSubmatch(c.rest); m != nil {
got = m[1] got = m[2]
} }
if got != c.want { if got != c.want {
t.Errorf("%s: queue id %q, want %q", c.name, 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 // A line is dated by its own stamp, not by when the tailer got to it, and is
// kept as the log has it. // kept as the log has it. Nothing is written until the pass over the log ends.
func TestStoreLineDatesByTheLine(t *testing.T) { func TestLineKeeperDatesByTheLine(t *testing.T) {
cs := &captureStore{} cs := &captureStore{}
k := newLineKeeper(cs)
now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC)
line := "Oct 12 09:00:00 mail postfix/qmgr[7]: 41E862C00D9E: removed" line := "Oct 12 09:00:00 mail postfix/qmgr[7]: 41E862C00D9E: removed"
storeLine(cs, line, now, false) k.add(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("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 { if len(cs.lines) != 1 {
t.Fatalf("kept %d lines, want 1: %v", len(cs.lines), cs.kept()) 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=<a@example.com>, 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=<a@example.net>, 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=<whatever-the-stranger-wrote@example.net>",
"host postfix/qmgr[10]: 4A1B2C3D: from=<a@example.net>, size=900, nrcpt=1 (queue active)",
outSent,
"host postfix/smtp[27]: 4A1B2C3D: to=<b@in.example.com>, 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 // 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 // stop - its saved place lies before lines it had already read - it does not
// keep them a second time. // 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 // Lines without a send-log row are removed at the start and after every
// sweep and on their own shorter cadence, and only once they are old enough // send-log sweep, and only once they are old enough for the row to have been
// for the row to have been written. // written.
func TestRetentionLoopPrunesLines(t *testing.T) { func TestRetentionLoopPrunesLines(t *testing.T) {
old := linePruneInterval old := retentionInterval
linePruneInterval = 20 * time.Millisecond retentionInterval = 20 * time.Millisecond
t.Cleanup(func() { linePruneInterval = old }) t.Cleanup(func() { retentionInterval = old })
cs := &captureStore{} cs := &captureStore{}
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
+16 -13
View File
@@ -32,8 +32,7 @@ type StatusStore interface {
UpdateStatus(queueID, recipient, status string) (int64, error) UpdateStatus(queueID, recipient, status string) (int64, error)
ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error)
DeleteSendLogBefore(cutoff time.Time) (int64, error) DeleteSendLogBefore(cutoff time.Time) (int64, error)
InsertSendLogLine(l store.SendLogLine) error InsertSendLogLines(lines []store.SendLogLine, once bool) (int, error)
InsertSendLogLineOnce(l store.SendLogLine) (bool, error)
PruneSendLogLines(cutoff time.Time) (int64, error) PruneSendLogLines(cutoff time.Time) (int64, error)
LogtailState(path string) (store.LogtailState, bool, error) LogtailState(path string) (store.LogtailState, bool, error)
SaveLogtailState(path string, st store.LogtailState) 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 // 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
// written while the panel was down. Every line headed by a queue id is kept // written while the panel was down. The lines headed by a queue id are kept
// for the message's page (lines.go). With inbound set it also writes the // 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 journal's rows for refused mail and prunes that journal
// (inbound.go). It returns nil on a clean shutdown. // (inbound.go). It returns nil on a clean shutdown.
func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays, inbound *Inbound) error { 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{}) caughtUp := make(chan struct{})
go reconcileLoop(ctx, st, caughtUp) 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 // 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 // while the panel was down - and, after an unclean stop, some it had
// read already. // read already.
@@ -144,7 +145,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
if journal != nil { if journal != nil {
journal.handle(line, now, catchingUp) journal.handle(line, now, catchingUp)
} }
storeLine(st, line, now, catchingUp) keeper.add(line, now, catchingUp)
queueID, recipient, status, ok := parseDelivery(line) queueID, recipient, status, ok := parseDelivery(line)
if !ok { if !ok {
return return
@@ -225,8 +226,9 @@ func reconcile(st StatusStore, cutoff time.Time) {
} }
// retentionLoop prunes expired send-log rows immediately and then periodically, // retentionLoop prunes expired send-log rows immediately and then periodically,
// and after each sweep the stored log lines those rows leave behind. Lines // and after each sweep the stored log lines no row carries the queue id of:
// without a row are also removed on a shorter cadence of their own (lines.go). // those the deleted rows leave behind, and those that never had a row
// (lines.go).
func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) { func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) {
prune := func() { prune := func() {
retentionDays := retention() retentionDays := retention()
@@ -248,8 +250,6 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays)
pruneLines(st) pruneLines(st)
t := time.NewTicker(retentionInterval) t := time.NewTicker(retentionInterval)
defer t.Stop() defer t.Stop()
lines := time.NewTicker(linePruneInterval)
defer lines.Stop()
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
@@ -257,8 +257,6 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays)
case <-t.C: case <-t.C:
prune() prune()
pruneLines(st) 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 // 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 // the point where every line the panel missed while it was down has been
// handled. // 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 ( var (
f *os.File f *os.File
r *bufio.Reader r *bufio.Reader
@@ -510,6 +512,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
}() }()
drain := func() { drain := func() {
defer flush()
for { for {
line, err := r.ReadString('\n') line, err := r.ReadString('\n')
if err == io.EOF { if err == io.EOF {
+19 -15
View File
@@ -150,27 +150,31 @@ type captureStore struct {
// lines are the log lines kept for messages' pages; pruned is the cutoff of // lines are the log lines kept for messages' pages; pruned is the cutoff of
// each PruneSendLogLines call. // each PruneSendLogLines call.
lines []store.SendLogLine lines []store.SendLogLine
pruned []time.Time 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() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
c.lines = append(c.lines, l) c.batches++
return nil wrote := 0
} next:
for _, l := range lines {
func (c *captureStore) InsertSendLogLineOnce(l store.SendLogLine) (bool, error) { if once {
c.mu.Lock() for _, have := range c.lines {
defer c.mu.Unlock() if have.QueueID == l.QueueID && have.Line == l.Line {
for _, have := range c.lines { continue next
if have.QueueID == l.QueueID && have.Line == l.Line { }
return false, nil }
} }
c.lines = append(c.lines, l)
wrote++
} }
c.lines = append(c.lines, l) return wrote, nil
return true, nil
} }
func (c *captureStore) PruneSendLogLines(cutoff time.Time) (int64, error) { func (c *captureStore) PruneSendLogLines(cutoff time.Time) (int64, error) {
+2 -4
View File
@@ -30,10 +30,8 @@ func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, erro
return 0, nil return 0, nil
} }
func (s *retentionProbeStore) InsertSendLogLine(store.SendLogLine) error { return nil } func (s *retentionProbeStore) InsertSendLogLines([]store.SendLogLine, bool) (int, error) {
return 0, nil
func (s *retentionProbeStore) InsertSendLogLineOnce(store.SendLogLine) (bool, error) {
return false, nil
} }
func (s *retentionProbeStore) PruneSendLogLines(time.Time) (int64, error) { return 0, nil } func (s *retentionProbeStore) PruneSendLogLines(time.Time) (int64, error) { return 0, nil }
+3 -2
View File
@@ -130,8 +130,9 @@ func (p RetryPolicy) BackoffCap() string {
} }
// LongestLifetime is how long Postfix may go on working on a message: the // 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 // larger of the queue lifetime for ordinary mail and for mail from the null
// bounds which stored log lines can be about a send-log row (store.SendLogLines). // 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 { func (p RetryPolicy) LongestLifetime() time.Duration {
if p.BounceQueueLifetime > p.MaximalQueueLifetime { if p.BounceQueueLifetime > p.MaximalQueueLifetime {
return p.BounceQueueLifetime return p.BounceQueueLifetime
+60 -36
View File
@@ -3,6 +3,7 @@ package store
import ( import (
"fmt" "fmt"
"time" "time"
"unicode/utf8"
) )
// SendLogLine is one mail.log line Postfix wrote under a queue id, kept so a // 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. // is accepted, and the message body can take its time after that.
const sendLogLineBefore = time.Hour const sendLogLineBefore = time.Hour
// cutSendLogLine cuts a line to SendLogLineMax bytes, and not through the
// middle of a character.
func cutSendLogLine(line string) string { func cutSendLogLine(line string) string {
if len(line) > SendLogLineMax { if len(line) <= SendLogLineMax {
return 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 // InsertSendLogLines stores lines in one transaction and returns how many it
// by a queue id, whether or not a send-log row exists for it yet. // wrote. The tailer calls it with what it read in one pass over the log,
func (s *Store) InsertSendLogLine(l SendLogLine) error { // whether or not send-log rows exist for the queue ids yet: a commit per line
_, err := s.db.Exec( // would be a disk sync per line, on the one connection the journal milter
`INSERT INTO send_log_lines (queue_id, logged_at, line) VALUES (?, ?, ?)`, // writes through.
l.QueueID, l.At.UTC().Format(time.RFC3339), cutSendLogLine(l.Line), //
) // With once set, a line is written only if the same line is not there already
if err != nil { // under its queue id. That is for the lines the tailer reads again after an
return fmt.Errorf("insert send_log_lines: %w", err) // 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 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)
// InsertSendLogLineOnce stores a line unless the same one is already there, SELECT ?1, ?2, ?3
// and reports whether it wrote. It is for the lines the tailer reads again WHERE NOT EXISTS (SELECT 1 FROM send_log_lines WHERE queue_id = ?1 AND line = ?3)`
// 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)
} }
n, _ := res.RowsAffected() tx, err := s.db.Begin()
return n > 0, nil 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 // 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 // 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 // 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 // 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 // deleted, and the lines of messages that never had a row - the bounces
// mail, the bounces Postfix generates. cutoff leaves alone the lines of a // Postfix generates. cutoff leaves alone the lines of a message still being
// message still being received, whose row the journal milter has yet to write. // 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) { func (s *Store) PruneSendLogLines(cutoff time.Time) (int64, error) {
res, err := s.db.Exec( res, err := s.db.Exec(
`DELETE FROM send_log_lines `DELETE FROM send_log_lines
WHERE logged_at < ? 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), cutoff.UTC().Format(time.RFC3339),
) )
if err != nil { if err != nil {
+39 -33
View File
@@ -4,6 +4,7 @@ import (
"strings" "strings"
"testing" "testing"
"time" "time"
"unicode/utf8"
) )
func sendLogLineTexts(lines []SendLogLine) []string { func sendLogLineTexts(lines []SendLogLine) []string {
@@ -28,15 +29,13 @@ func countSendLogLines(t *testing.T, s *Store) int {
func TestSendLogLinesOfOneMessage(t *testing.T) { func TestSendLogLinesOfOneMessage(t *testing.T) {
s := openTestStore(t) s := openTestStore(t)
at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) 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: "41E862C00D9E", At: at, Line: "client=app"},
{QueueID: "5F00AA", At: at, Line: "someone else's"}, {QueueID: "5F00AA", At: at, Line: "someone else's"},
{QueueID: "41E862C00D9E", At: at.Add(time.Second), Line: "status=sent"}, {QueueID: "41E862C00D9E", At: at.Add(time.Second), Line: "status=sent"},
{QueueID: "41E862C00D9E", At: at.Add(2 * time.Second), Line: "removed"}, {QueueID: "41E862C00D9E", At: at.Add(2 * time.Second), Line: "removed"},
} { }, false); err != nil || n != 4 {
if err := s.InsertSendLogLine(l); err != nil { t.Fatalf("insert: wrote %d, %v", n, err)
t.Fatalf("insert line %d: %v", i, err)
}
} }
row := SendLogRow{QueueID: "41E862C00D9E", CreatedAt: at} 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) first := time.Date(2026, 9, 1, 9, 0, 0, 0, time.UTC)
second := first.AddDate(0, 0, 30) second := first.AddDate(0, 0, 30)
lifetime := 6 * 24 * time.Hour 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(-30 * time.Minute), Line: "first: client"},
{QueueID: "ABC123", At: first.Add(5 * 24 * time.Hour), Line: "first: expired"}, {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.Add(-2 * time.Hour), Line: "between the two"},
{QueueID: "ABC123", At: second, Line: "second: sent"}, {QueueID: "ABC123", At: second, Line: "second: sent"},
} { }, false); err != nil {
if err := s.InsertSendLogLine(l); err != nil { t.Fatalf("insert: %v", err)
t.Fatalf("insert: %v", err)
}
} }
lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: first}, lifetime, 200) 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 // 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. // writes nothing, also when the line comes twice in one batch. The same text
func TestInsertSendLogLineOnce(t *testing.T) { // under another queue id is another line. Without once, what is given is
// written.
func TestInsertSendLogLinesOnce(t *testing.T) {
s := openTestStore(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"} 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 := l
other.QueueID = "DEF456" 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 { if n := countSendLogLines(t, s); n != 2 {
t.Errorf("%d lines stored, want 2", n) 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 // 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) { func TestSendLogLineIsCut(t *testing.T) {
s := openTestStore(t) s := openTestStore(t)
at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC)
l := SendLogLine{QueueID: "ABC123", At: at, Line: strings.Repeat("x", SendLogLineMax+500)} // Two bytes a character, starting at an odd offset: the ceiling falls inside one.
if err := s.InsertSendLogLine(l); err != nil { 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) t.Fatalf("insert: %v", err)
} }
if wrote, err := s.InsertSendLogLineOnce(l); err != nil || wrote { if n, err := s.InsertSendLogLines([]SendLogLine{l}, true); err != nil || n != 0 {
t.Errorf("the same long line again: wrote = %v, %v", wrote, err) t.Errorf("the same long line again: wrote %d, %v", n, err)
} }
lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: at}, time.Hour, 10) lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: at}, time.Hour, 10)
if err != nil || len(lines) != 1 { if err != nil || len(lines) != 1 {
t.Fatalf("lines = %d, %v", len(lines), err) t.Fatalf("lines = %d, %v", len(lines), err)
} }
if len(lines[0].Line) != SendLogLineMax { if got := lines[0].Line; len(got) != SendLogLineMax-1 || !utf8.ValidString(got) {
t.Errorf("kept %d bytes, want %d", len(lines[0].Line), SendLogLineMax) 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 { if err := s.InsertQueued(SendLogEntry{QueueID: "KEPT01", Domain: "example.com", To: "a@example.net"}); err != nil {
t.Fatalf("insert row: %v", err) 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: "KEPT01", At: old, Line: "has a row"},
{QueueID: "INBOUND1", At: old, Line: "never had 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"}, {QueueID: "ARRIVING", At: now.Add(-time.Minute), Line: "its row is not written yet"},
} { }, false); err != nil {
if err := s.InsertSendLogLine(l); err != nil { t.Fatalf("insert: %v", err)
t.Fatalf("insert: %v", err)
}
} }
n, err := s.PruneSendLogLines(now.Add(-time.Hour)) n, err := s.PruneSendLogLines(now.Add(-time.Hour))
@@ -197,15 +197,13 @@ func TestDeliveryPageShowsStoredLogLinesAfterRotation(t *testing.T) {
h, row := serverWithDelivery(t) h, row := serverWithDelivery(t)
h.cfg.RetryPolicy = postfix.DefaultRetryPolicy() h.cfg.RetryPolicy = postfix.DefaultRetryPolicy()
h.cfg.MailLogPath = writeMailLog(t, "2026-08-04T00:00:01.000000+00:00 host postfix/qmgr[10]: 99999999: removed") 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.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: 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=<other@example.ru>, size=500, nrcpt=1 (queue active)"}, {QueueID: "99999999", At: row.CreatedAt, Line: "2026-08-03T05:15:52.300000+00:00 host postfix/qmgr[10]: 99999999: from=<other@example.ru>, 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=<public@example.ru>, dsn=2.0.0, status=sent (250 OK)"}, {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=<public@example.ru>, dsn=2.0.0, status=sent (250 OK)"},
} { }, false); err != nil {
if err := h.store.InsertSendLogLine(l); err != nil { t.Fatalf("InsertSendLogLines: %v", err)
t.Fatalf("InsertSendLogLine: %v", err)
}
} }
out := getBody(t, h.HandleDelivery, "/outbound/log/"+itoa(row.ID)) out := getBody(t, h.HandleDelivery, "/outbound/log/"+itoa(row.ID))