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 <noreply@anthropic.com>

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:

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 <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", ""},
@@ -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=<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
 // 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())

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 {

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) {

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 }

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

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 {

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))

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=<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)"},
-	} {
-		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))