findings of the reviews of audit stage 1

Technical review (Opus) and security review (Fable) of 9efc9c7..a59e3d4;
reports in the session's scratchpad. Fixed:

- B1 (verified bug): store.UpdateStatus read before writing in its
  transaction; a commit by another connection in between (the dmarc-ingest
  process opens the same file) failed it at once with "database is locked"
  and the delivery line was lost. The first statement is now a write.
- R3: a delivery line updated every row of its queue id and recipient; it
  now updates only the newest (Postfix reuses short ids). The sweep closes
  by row id (QueuedDelivery.ID, CloseUnreported(id)).
- R4: a contradicted sweep keeps a mark, reason "reconcile-contradicted",
  instead of losing which row it was.
- R1: contradicted cannot see a lost line (never read late). New counter
  reconcile_missed: at close, the sweep looks for the row's delivery line
  in mail.log and mail.log.1 (QueueLines). 0007 edited - it has run on no
  database but throwaway stands.
- R2: queued_over_lifetime cannot measure "what removing the sweep would
  leave" while the sweep runs; its text now says rows stuck despite the
  sweep, and the rule uses closed - missed - contradicted.
- cert-reload (R5, R6, R7, security medium/low): both files copied to .new
  and the copies checked as a pair before both are renamed; acts on the copy
  only when postconf says Postfix reads it; a failed reload is retried next
  pass; umask 077; says so when the mounted files cannot be read. The e2e
  case asserts Postfix reads the copy, the copy equals the mount, the key is
  600 root.
- Security low: a recipient with a quoted local part - the journal cut
  "a b"@x at the space, deliveryRe cut "a>b"@x at the ">" - matched no
  delivery line and was closed by the sweep. cleanAddress ends a bracketed
  address at the first ">" outside quotes; deliveryRe takes the recipient
  up to the ">, " before relay= (test fixtures given the relay= every real
  delivery line has).
- Security low: postqueue -p's error carries at most 512 bytes of output.
- Nits: addDeliveryLogDay refuses a column not on its list; the truncation
  test checks its message.
- To stage 2b (page wording, mockup-first): the history step of a bounced
  row says the receiving server refused it, false for a row the sweep
  closed. To the owner when stage 2b is reached: since M13 the "deferred"
  figure counts whole messages too, which "Connections turned away" would
  not name.

Found while verifying: the e2e's main positive path failed once with "553
not owned by user app1" - an smtpd idle from before the panel's reload took
the connection in the reload's second. Measured on the stand (one-off
probe, not committed): sending 0, 10, 40, 110 s after the save (two seconds
at the soonest) always met an smtpd started after the reload. The e2e now
waits for the sender map after each application it creates
(waitForSenderMap: AUTH, MAIL, RCPT, RSET - no message, no row), and the
level-1 case uses it instead of its own loop.

Docs: architecture § Queue reconcile, schema-migrations (rows 7 and 8,
chain head 8), CHANGELOG, audit plan (stage 1 note, the measured window,
a new stage 2b line: 62 lines), roadmap 15/62.

Evidence on the test server for this tree: go vet clean, go test ./... 23
packages ok, make e2e exit 0 (renewed certificate, hung panel, level 1 at
49/60).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/logtail/lines_test.go

@@ -176,7 +176,7 @@ func TestRunKeepsQueueLinesOnce(t *testing.T) {
 
 	const (
 		client = "host postfix/submission/smtpd[1]: 41E862C00D9E: client=unknown[203.0.113.7]"
-		sent   = "host postfix/smtp[1]: 41E862C00D9E: to=<a@example.net>, dsn=2.0.0, status=sent (ok)"
+		sent   = "host postfix/smtp[1]: 41E862C00D9E: to=<a@example.net>, relay=x, dsn=2.0.0, status=sent (ok)"
 		later  = "host postfix/qmgr[1]: 41E862C00D9E: removed"
 	)
 

internal/logtail/logtail.go

@@ -31,9 +31,10 @@ import (
 type StatusStore interface {
 	UpdateStatus(queueID, recipient, status string) (int64, error)
 	ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error)
-	CloseUnreported(queueID, recipient string) (int64, error)
+	CloseUnreported(id int64) (int64, error)
 	// The counters of delivery_log_days (store/deliverylog.go).
 	AddReconcileClosed(at time.Time, n int64) error
+	AddReconcileMissed(at time.Time, n int64) error
 	AddTailerBlind(from time.Time, d time.Duration) error
 	CountQueuedOlderThan(cutoff time.Time) (int64, error)
 	SetQueuedOverLifetime(at time.Time, n int64) error
@@ -88,6 +89,10 @@ type RetentionDays func() int
 // The "<queue-id>: to=<addr>, …, status=<word>" shape is specific to the
 // delivery agents; qmgr/smtpd/cleanup lines do not match.
 //
+// The recipient runs to the ">, " that comes before relay= (and orig_to=,
+// when there is one): a quoted local part may hold a ">", and a recipient cut
+// there would match no row and leave it for the sweep (review of stage 1).
+//
 // It is matched at the head of the line - the host name, "program[pid]:",
 // then the queue id - after SplitTimestamp has taken the stamp off, like
 // queueLineRe and the inbound patterns. Anywhere else the text is a
@@ -100,7 +105,7 @@ type RetentionDays func() int
 // the line — and that one can come from the reply text, which the far end
 // controls. A bounce whose reply quoted "status=sent" would then be filed as a
 // success. The real field is always the first one after to=<…>.
-var deliveryRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-Za-z]+): to=<([^>]*)>,.*?\bstatus=(\w+)`)
+var deliveryRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-Za-z]+): to=<(.*?)>, (?:orig_to=<.*?>, )?relay=.*?\bstatus=(\w+)`)
 
 // parseDelivery extracts (queue-id, recipient, status) from a mail.log line.
 // ok is false for lines that are not recognised delivery results.
@@ -149,7 +154,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
 	// the sweep would otherwise close. follow() closes this once it has read to
 	// end-of-file for the first time.
 	caughtUp := make(chan struct{})
-	go reconcileLoop(ctx, st, caughtUp)
+	go reconcileLoop(ctx, st, caughtUp, path)
 
 	var deferred *deferredCounter
 	if sender != nil {
@@ -190,7 +195,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
 // up with the log, and then leaves the first sweep a full interval away, so a
 // restart resolves rows from the log — the accurate source — before the sweep
 // gets to guess at whatever the log could not explain.
-func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{}) {
+func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{}, path string) {
 	select {
 	case <-ctx.Done():
 		return
@@ -210,7 +215,7 @@ func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{}
 			return
 		case <-t.C:
 			now := time.Now().UTC()
-			reconcile(st, now.Add(-reconcileGrace))
+			reconcile(st, now.Add(-reconcileGrace), path)
 			countStuck(st, now, overLifetime)
 		}
 	}
@@ -242,7 +247,12 @@ func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) {
 // success because a delivery the panel cannot evidence must not be shown as
 // one. Rows whose message is still in the queue, and every row when the queue
 // cannot be listed at all, are left exactly as they are.
-func reconcile(st StatusStore, cutoff time.Time) {
+//
+// A closed row whose delivery line is in mail.log or mail.log.1 after all is
+// counted as missed: the line was there and the tailer did not read it. That
+// is what tells a tailer that loses lines from a log that lost them, which
+// reconcile_contradicted cannot - a lost line is never read late.
+func reconcile(st StatusStore, cutoff time.Time, path string) {
 	rows, err := st.ListQueuedOlderThan(cutoff)
 	if err != nil {
 		log.Printf("log-tailer: reconcile: list queued rows: %v", err)
@@ -260,26 +270,54 @@ func reconcile(st StatusStore, cutoff time.Time) {
 		return
 	}
 
-	var closed int64
+	var closed, missed int64
 	for _, row := range rows {
 		if _, still := held[row.QueueID]; still {
 			continue
 		}
-		n, err := st.CloseUnreported(row.QueueID, row.To)
+		n, err := st.CloseUnreported(row.ID)
 		if err != nil {
 			log.Printf("log-tailer: reconcile: close %s/%s: %v", row.QueueID, row.To, err)
 			continue
 		}
 		closed += n
+		if n > 0 && deliveryInLog(path, row) {
+			log.Printf("log-tailer: reconcile: %s/%s had a delivery line in the log the tailer did not read", row.QueueID, row.To)
+			missed++
+		}
 	}
 	if closed > 0 {
 		log.Printf("log-tailer: reconcile: closed %d row(s) Postfix no longer holds and never reported", closed)
 		if err := st.AddReconcileClosed(time.Now(), closed); err != nil {
 			log.Printf("log-tailer: reconcile: count: %v", err)
 		}
+		if err := st.AddReconcileMissed(time.Now(), missed); err != nil {
+			log.Printf("log-tailer: reconcile: count: %v", err)
+		}
 	}
 }
 
+// deliveryInLog says whether the current log or the one rotated last holds a
+// delivery line for row. Rows reach the sweep minutes after they were
+// written, so the end of those two files is where such a line would be.
+func deliveryInLog(path string, row store.QueuedDelivery) bool {
+	if path == "" {
+		return false
+	}
+	for _, p := range []string{path, path + ".1"} {
+		lines, err := QueueLines(p, row.QueueID, 256)
+		if err != nil {
+			continue
+		}
+		for _, line := range lines {
+			if q, r, _, ok := parseDelivery(line); ok && q == row.QueueID && strings.EqualFold(r, row.To) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // retentionLoop prunes expired send-log rows immediately and then periodically,
 // 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

internal/logtail/logtail_test.go

@@ -116,6 +116,16 @@ func TestParseDelivery(t *testing.T) {
 			line:   "host postfix/cleanup[12]: BEEF03: message-id=<x@example.com>",
 			wantOK: false,
 		},
+		{
+			// A quoted local part may hold ">": the recipient runs to the
+			// ">, " before relay=.
+			name:      "a recipient with > in a quoted local part",
+			line:      `host postfix/smtp[26]: C0FFEE: to=<"a>b"@example.net>, relay=mx[203.0.113.9]:25, dsn=2.0.0, status=sent (250 OK)`,
+			wantOK:    true,
+			queueID:   "C0FFEE",
+			recipient: `"a>b"@example.net`,
+			status:    store.StatusSent,
+		},
 		{
 			// The stamp Postfix 3.7's postlogd writes, as on the stand.
 			name:      "syslog stamp",
@@ -178,6 +188,7 @@ type captureStore struct {
 	// closed sums AddReconcileClosed; stuckCount is what CountQueuedOlderThan
 	// answers and stuck what SetQueuedOverLifetime was given.
 	closed     int64
+	missed     int64
 	stuckCount int64
 	stuck      []int64
 
@@ -241,12 +252,18 @@ func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, e
 	return 1, nil
 }
 
-// CloseUnreported records the sweep's close as "queue-id|recipient|reconcile".
-func (c *captureStore) CloseUnreported(queueID, recipient string) (int64, error) {
+// CloseUnreported records the sweep's close as "queue-id|recipient|reconcile",
+// finding the row by its id among queued.
+func (c *captureStore) CloseUnreported(id int64) (int64, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.calls = append(c.calls, queueID+"|"+recipient+"|"+store.ReasonReconcile)
-	return 1, nil
+	for _, q := range c.queued {
+		if q.ID == id {
+			c.calls = append(c.calls, q.QueueID+"|"+q.To+"|"+store.ReasonReconcile)
+			return 1, nil
+		}
+	}
+	return 0, nil
 }
 
 func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil }
@@ -259,6 +276,12 @@ func (c *captureStore) AddReconcileClosed(_ time.Time, n int64) error {
 	c.closed += n
 	return nil
 }
+func (c *captureStore) AddReconcileMissed(_ time.Time, n int64) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.missed += n
+	return nil
+}
 func (c *captureStore) AddTailerBlind(time.Time, time.Duration) error { return nil }
 func (c *captureStore) CountQueuedOlderThan(time.Time) (int64, error) {
 	c.mu.Lock()
@@ -332,7 +355,7 @@ func TestFollowTailsAndRotates(t *testing.T) {
 	// Give follow() time to open at EOF (it seeks to end immediately on start,
 	// so the seed line above is ignored), then append a delivery line.
 	time.Sleep(50 * time.Millisecond)
-	appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, dsn=2.0.0, status=sent (ok)")
+	appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
 	waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
 
 	// Rotate: move the current file aside and create a fresh one (logrotate
@@ -343,7 +366,7 @@ func TestFollowTailsAndRotates(t *testing.T) {
 	if err := os.WriteFile(path, nil, 0o644); err != nil {
 		t.Fatalf("recreate: %v", err)
 	}
-	appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, dsn=5.1.1, status=bounced (nope)")
+	appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, relay=x, dsn=5.1.1, status=bounced (nope)")
 	waitFor(t, func() bool { return contains(cs.snapshot(), "Q2|b@example.net|bounced") })
 
 	cancel()
@@ -375,12 +398,12 @@ func TestFollowResumesAfterRestart(t *testing.T) {
 
 	cs := &captureStore{}
 	stop := startRun(t, path, cs)
-	appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, dsn=2.0.0, status=sent (ok)")
+	appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
 	waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
 	stop() // persists the offset past Q1
 
 	// Down: Postfix keeps delivering.
-	appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, dsn=2.0.0, status=sent (ok)")
+	appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
 
 	cs.reset()
 	stop = startRun(t, path, cs)
@@ -394,7 +417,7 @@ func TestFollowResumesAfterRestart(t *testing.T) {
 	// The stored offset belongs to a file that no longer exists, so the new one
 	// must be read from the start.
 	if err := os.WriteFile(path, []byte(strings.Repeat("host postfix/qmgr[1]: fresh log after rotation\n", 20)+
-		"host postfix/smtp[1]: Q3: to=<c@example.net>, dsn=5.1.1, status=bounced (nope)\n"), 0o644); err != nil {
+		"host postfix/smtp[1]: Q3: to=<c@example.net>, relay=x, dsn=5.1.1, status=bounced (nope)\n"), 0o644); err != nil {
 		t.Fatalf("recreate log: %v", err)
 	}
 
@@ -490,17 +513,39 @@ func TestReconcileClosesOnlyWhatPostfixNoLongerHolds(t *testing.T) {
 	stubQueue(t, nil, "STILLQ")
 
 	cs := &captureStore{queued: []store.QueuedDelivery{
-		{QueueID: "STILLQ", To: "inflight@example.net"},
-		{QueueID: "GONEQ", To: "lost@example.net"},
+		{ID: 1, QueueID: "STILLQ", To: "inflight@example.net"},
+		{ID: 2, QueueID: "GONEQ", To: "lost@example.net"},
 	}}
-	reconcile(cs, time.Now().UTC())
+	reconcile(cs, time.Now().UTC(), "")
 
 	got := cs.snapshot()
 	if len(got) != 1 || got[0] != "GONEQ|lost@example.net|"+store.ReasonReconcile {
 		t.Fatalf("got %v, want only the message Postfix dropped closed by the sweep", got)
 	}
-	if cs.closed != 1 {
-		t.Fatalf("the day counts %d closed row(s), want 1", cs.closed)
+	if cs.closed != 1 || cs.missed != 0 {
+		t.Fatalf("the day counts %d closed and %d missed, want 1 and 0", cs.closed, cs.missed)
+	}
+}
+
+// A closed row whose delivery line is in the log after all - here in the file
+// rotated last - is counted as missed: the tailer did not read it.
+func TestReconcileCountsALineTheTailerMissed(t *testing.T) {
+	stubQueue(t, nil)
+	dir := t.TempDir()
+	path := filepath.Join(dir, "mail.log")
+	if err := os.WriteFile(path, []byte("Oct 14 11:00:00 mail postfix/qmgr[10]: OTHERQ: removed\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(path+".1", []byte("Oct 14 10:59:58 mail postfix/smtp[26]: SEENQ: to=<Seen@example.net>, relay=mx[203.0.113.9]:25, dsn=2.0.0, status=sent (250 OK)\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	cs := &captureStore{queued: []store.QueuedDelivery{
+		{ID: 1, QueueID: "SEENQ", To: "seen@example.net"},
+		{ID: 2, QueueID: "GONEQ", To: "lost@example.net"},
+	}}
+	reconcile(cs, time.Now().UTC(), path)
+	if cs.closed != 2 || cs.missed != 1 {
+		t.Fatalf("closed %d, missed %d; want 2 and 1", cs.closed, cs.missed)
 	}
 }
 
@@ -560,9 +605,9 @@ func TestReconcileLeavesRowsAloneWhenTheQueueCannotBeRead(t *testing.T) {
 	stubQueue(t, errors.New("postqueue: Permission denied"))
 
 	cs := &captureStore{queued: []store.QueuedDelivery{
-		{QueueID: "GONEQ", To: "lost@example.net"},
+		{ID: 1, QueueID: "GONEQ", To: "lost@example.net"},
 	}}
-	reconcile(cs, time.Now().UTC())
+	reconcile(cs, time.Now().UTC(), "")
 
 	if got := cs.snapshot(); len(got) != 0 {
 		t.Fatalf("got %v, want no row touched", got)
@@ -574,7 +619,7 @@ func TestReconcileLeavesRowsAloneWhenTheQueueCannotBeRead(t *testing.T) {
 func TestReconcileSkipsTheQueueWhenNoRowIsStuck(t *testing.T) {
 	calls := stubQueue(t, nil)
 
-	reconcile(&captureStore{}, time.Now().UTC())
+	reconcile(&captureStore{}, time.Now().UTC(), "")
 
 	if *calls != 0 {
 		t.Fatalf("queue listed %d time(s), want none", *calls)

internal/logtail/retention_test.go

@@ -23,10 +23,11 @@ func (s *retentionProbeStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDeli
 	return nil, nil
 }
 
-func (s *retentionProbeStore) CloseUnreported(string, string) (int64, error) {
+func (s *retentionProbeStore) CloseUnreported(int64) (int64, error) {
 	return 0, nil
 }
 
+func (s *retentionProbeStore) AddReconcileMissed(time.Time, int64) error     { return nil }
 func (s *retentionProbeStore) AddReconcileClosed(time.Time, int64) error     { return nil }
 func (s *retentionProbeStore) AddTailerBlind(time.Time, time.Duration) error { return nil }
 func (s *retentionProbeStore) CountQueuedOlderThan(time.Time) (int64, error) { return 0, nil }