From 9adb44845cd0010b3365fdf3149719f6f97a2e88 Mon Sep 17 00:00:00 2001 From: Mix Date: Wed, 14 Oct 2026 14:16:26 +0300 Subject: [PATCH] logtail: the sweep closes only queued rows, with reason "reconcile" (audit H3) reconcile() closed a row through UpdateStatus (WHERE queue_id AND to_addr), which could overwrite a "sent" the tailer wrote between the listing and the close, and rewrite an old row of a reused queue id. store.CloseUnreported sets status bounced and reason "reconcile" (store.ReasonReconcile; the 0006 column, no schema change) WHERE ... AND status = 'queued'; the sweep counts rows actually closed. Tests: store - a queued row is closed with the reason, a sent row is left alone (0 rows); logtail - the sweep's close goes through CloseUnreported. CHANGELOG [Unreleased], audit plan line ticked, roadmap 9/61. Evidence: go vet ./... clean; go test ./internal/store ok, ./internal/logtail ok but for TestFollowTailsAndRotates (Windows rename of an open file, fails here always); the server run of the stage follows before the push. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 +++++ docs/audit-2026-10-12.md | 2 +- docs/roadmap.md | 2 +- internal/logtail/logtail.go | 12 ++++++---- internal/logtail/logtail_test.go | 12 ++++++++-- internal/logtail/retention_test.go | 4 ++++ internal/store/sendercheck.go | 4 ++++ internal/store/sendlog.go | 19 ++++++++++++++++ internal/store/sendlog_test.go | 35 ++++++++++++++++++++++++++++++ 9 files changed, 88 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ff5dcc..81ede64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,12 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version marked `bounced`. A listing over the cap now tells the sweep nothing, as it already did for the notification about mail waiting too long; the Mail queue page still shows the cut listing with its note (audit 2026-10-12, H3). +- **The sweep closes only what is still queued, and says it did.** Its close + was the tailer's plain status update, so a `sent` the log wrote an instant + earlier could be overwritten with `bounced`, and a reused queue id could + rewrite an old row of the same recipient. The close is now conditional on + the row still being `queued`, and the row it closes carries the reason + `reconcile` (audit 2026-10-12, H3). ## [2.4.2] - 2026-10-14 diff --git a/docs/audit-2026-10-12.md b/docs/audit-2026-10-12.md index f181077..26202ed 100644 --- a/docs/audit-2026-10-12.md +++ b/docs/audit-2026-10-12.md @@ -596,7 +596,7 @@ has already run is never edited — the next file is added; the published - [x] **`QueueIDs` on a truncated listing returns an error**, as `postqueueJSON` does; `reconcile` already treats an error as no information. Test: a 5 MiB fake `postqueue -p`. — **Opus** -- [ ] **Reconcile only closes `queued` rows** (`AND status = 'queued'` in the +- [x] **Reconcile only closes `queued` rows** (`AND status = 'queued'` in the reconcile path's update) and marks what it closed: `reason = 'reconcile'` on the row (the 0006 column; a new value, no schema change). — **Opus** - [ ] **Measure before deciding the sweep's fate** (owner, 2026-10-12: the diff --git a/docs/roadmap.md b/docs/roadmap.md index 3a46ea8..71760d3 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -42,7 +42,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md). | from-header-binding | Hold a message's `From:` header against the application that sends it | shipped `2.4.0` | 9/9 | [plans/from-header-binding.md](plans/from-header-binding.md) | | panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) | | schema-squash | Squash SQLite migrations into a 2.x baseline | built in panel-redesign (stage 1) | — | [plans/panel-redesign.md](plans/panel-redesign.md) | -| audit-2026-10-12 | Follow-ups of the 2026-10-12 full-tree code audit, seven stages | in progress (stage 0 shipped `2.4.2`; next stages 1+2 → `2.5.0`) | 8/61 | [audit-2026-10-12.md](audit-2026-10-12.md) | +| audit-2026-10-12 | Follow-ups of the 2026-10-12 full-tree code audit, seven stages | in progress (stage 0 shipped `2.4.2`; next stages 1+2 → `2.5.0`) | 9/61 | [audit-2026-10-12.md](audit-2026-10-12.md) | **Recommended order** (not binding; owner, 2026-10-09: feature order follows development convenience): **panel-redesign** shipped as `2.0.0` (2026-10-11) and diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go index ca66231..d0e5759 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -31,6 +31,7 @@ import ( type StatusStore interface { UpdateStatus(queueID, recipient, status string) (int64, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) + CloseUnreported(queueID, recipient string) (int64, error) DeleteSendLogBefore(cutoff time.Time) (int64, error) InsertSendLogLines(lines []store.SendLogLine, once bool) (int, error) PruneSendLogLines(cutoff time.Time) (int64, error) @@ -191,7 +192,9 @@ func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{} } // reconcile marks as bounced every row still "queued" from before cutoff whose -// message Postfix no longer holds. +// message Postfix no longer holds, with reason "reconcile". The close itself +// is conditional on the row still being queued, so a delivery line read +// between the listing and the close wins. // // A row reaches this state only when its delivery lines are gone for good — the // log rotated past its fourteen files while the panel was down, or was deleted @@ -219,16 +222,17 @@ func reconcile(st StatusStore, cutoff time.Time) { return } - var closed int + var closed int64 for _, row := range rows { if _, still := held[row.QueueID]; still { continue } - if _, err := st.UpdateStatus(row.QueueID, row.To, store.StatusBounced); err != nil { + n, err := st.CloseUnreported(row.QueueID, row.To) + if err != nil { log.Printf("log-tailer: reconcile: close %s/%s: %v", row.QueueID, row.To, err) continue } - closed++ + closed += n } if closed > 0 { log.Printf("log-tailer: reconcile: closed %d row(s) Postfix no longer holds and never reported", closed) diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go index d073db3..a1b050e 100644 --- a/internal/logtail/logtail_test.go +++ b/internal/logtail/logtail_test.go @@ -208,6 +208,14 @@ 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) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, queueID+"|"+recipient+"|"+store.ReasonReconcile) + return 1, nil +} + func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil } func (c *captureStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDelivery, error) { @@ -433,8 +441,8 @@ func TestReconcileClosesOnlyWhatPostfixNoLongerHolds(t *testing.T) { reconcile(cs, time.Now().UTC()) got := cs.snapshot() - if len(got) != 1 || got[0] != "GONEQ|lost@example.net|"+store.StatusBounced { - t.Fatalf("got %v, want only the message Postfix dropped closed as bounced", got) + 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) } } diff --git a/internal/logtail/retention_test.go b/internal/logtail/retention_test.go index 9e277da..2f41aa0 100644 --- a/internal/logtail/retention_test.go +++ b/internal/logtail/retention_test.go @@ -23,6 +23,10 @@ func (s *retentionProbeStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDeli return nil, nil } +func (s *retentionProbeStore) CloseUnreported(string, string) (int64, error) { + return 0, nil +} + func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, error) { s.mu.Lock() s.cutoff = cutoff diff --git a/internal/store/sendercheck.go b/internal/store/sendercheck.go index bc25d30..ff4759a 100644 --- a/internal/store/sendercheck.go +++ b/internal/store/sendercheck.go @@ -13,6 +13,10 @@ const ( ReasonFrom = "from" // the From: line is not the application's ) +// ReasonReconcile marks a "bounced" row the reconcile sweep closed (send_log.reason): +// its message left Postfix's queue with no delivery line ever read for it. +const ReasonReconcile = "reconcile" + // Settings keys of the sender check's statistics. const ( // settingSenderCheckSince is the first UTC day the statistics cover. diff --git a/internal/store/sendlog.go b/internal/store/sendlog.go index 96d6205..e810705 100644 --- a/internal/store/sendlog.go +++ b/internal/store/sendlog.go @@ -97,6 +97,25 @@ func (s *Store) UpdateStatus(queueID, recipient, status string) (int64, error) { return n, nil } +// CloseUnreported is the reconcile sweep's close of one row: "bounced", marked +// reason "reconcile" so it is known who decided it (architecture.md § Log +// tailer). Only a row still "queued" is closed - a result the log wrote an +// instant earlier stands, and a months-old row of a reused queue id is not +// rewritten (audit 2026-10-12, H3). It returns the number of rows closed. +func (s *Store) CloseUnreported(queueID, recipient string) (int64, error) { + now := time.Now().UTC().Format(time.RFC3339) + res, err := s.db.Exec( + `UPDATE send_log SET status = ?, reason = ?, updated_at = ? + WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE AND status = ?`, + StatusBounced, ReasonReconcile, now, queueID, recipient, StatusQueued, + ) + if err != nil { + return 0, fmt.Errorf("close unreported send_log row: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} + // QueuedDelivery is a send-log row still waiting for a delivery result, // reduced to what the log-tailer's reconcile sweep needs to look it up in the // Postfix queue and, failing that, to close it (architecture.md § Log tailer). diff --git a/internal/store/sendlog_test.go b/internal/store/sendlog_test.go index 05e4456..67cbe7a 100644 --- a/internal/store/sendlog_test.go +++ b/internal/store/sendlog_test.go @@ -97,6 +97,41 @@ func TestUpdateStatusRecipientCaseInsensitive(t *testing.T) { } } +// The sweep closes only what is still queued, and says it was the sweep: a +// status the log wrote stands, whatever the sweep believed when it listed the +// row. +func TestCloseUnreportedClosesOnlyQueuedRows(t *testing.T) { + st := openTestStore(t) + for _, e := range []SendLogEntry{{QueueID: "Q1", To: "Lost@example.net"}, {QueueID: "Q2", To: "done@example.net"}} { + if err := st.InsertQueued(e); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + } + if _, err := st.UpdateStatus("Q2", "done@example.net", StatusSent); err != nil { + t.Fatal(err) + } + + if n, err := st.CloseUnreported("Q1", "lost@example.net"); err != nil || n != 1 { + t.Fatalf("CloseUnreported of a queued row = %d, %v; want 1", n, err) + } + if n, err := st.CloseUnreported("Q2", "done@example.net"); err != nil || n != 0 { + t.Fatalf("CloseUnreported of a sent row = %d, %v; want 0", n, err) + } + + var status, reason string + for _, c := range []struct{ id, status, reason string }{ + {"Q1", StatusBounced, ReasonReconcile}, + {"Q2", StatusSent, ""}, + } { + if err := st.db.QueryRow(`SELECT status, reason FROM send_log WHERE queue_id = ?`, c.id).Scan(&status, &reason); err != nil { + t.Fatal(err) + } + if status != c.status || reason != c.reason { + t.Errorf("%s: status %q reason %q, want %q %q", c.id, status, reason, c.status, c.reason) + } + } +} + // The reconcile sweep asks for the rows old enough that Postfix should have // reported on them by now. A message accepted moments ago is simply in flight, // and one the milter refused never reached the queue at all, so neither is the