From ffefa9d6df0babe65fab347ac88d3f6b4324085a Mon Sep 17 00:00:00 2001 From: Mix Date: Wed, 14 Oct 2026 14:20:44 +0300 Subject: [PATCH] delivery log: measure the sweep before deciding its fate (audit H3, 0007) Migration 0007 (DDL only): delivery_log_days, a row per UTC day, kept a year, shown nowhere - read off a copy of the database after a month by the decision rule written in the audit plan. - reconcile_closed: the sweep adds the rows it actually closed. - reconcile_contradicted: store.UpdateStatus, in one transaction, counts the rows with reason "reconcile" a delivery line now speaks of, writes the log's status and clears the reason - so a re-read line counts once. - tailer_blind_seconds: follow() tracks stretches in which the file is gone or reopening fails (logtail.blindness); stored when a stretch ends and every minute while it lasts; under 3 s not counted (the rotation's tick between rename and create loses nothing). - queued_over_lifetime: every sweep overwrites the day's level with the rows queued past LongestLifetime + MaximalBackoff (a message Postfix still works on stays queued until its first retry after the lifetime). - retentionLoop prunes days older than a year. Tests: store (sums per day, the level overwritten, under a second nothing, prune; a contradicted sweep counted once and its reason cleared; count of queued rows); logtail (the sweep counts what it closed; countStuck records the level; blindness - a one-tick gap not counted, 75 s stored as 60 + 15 from where the last part ended). Schema tests at user_version 7. schema-migrations.md: chain head 7 (it still said 5 since 2.4.0), row 7. CHANGELOG [Unreleased], audit plan line ticked, roadmap 10/61. Evidence: go vet ./... clean; go test ./internal/store ok, ./internal/logtail ok but for TestFollowTailsAndRotates (Windows); the server run of the stage follows before the push. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 ++ docs/audit-2026-10-12.md | 2 +- docs/roadmap.md | 2 +- docs/schema-migrations.md | 8 +- internal/logtail/logtail.go | 91 ++++++++++++++- internal/logtail/logtail_test.go | 80 +++++++++++++ internal/logtail/retention_test.go | 6 + internal/store/deliverylog.go | 106 ++++++++++++++++++ internal/store/deliverylog_test.go | 69 ++++++++++++ .../migrations/0007_delivery_log_days.sql | 19 ++++ internal/store/schema_test.go | 9 +- internal/store/sendlog.go | 38 ++++++- internal/store/sendlog_test.go | 32 ++++++ 13 files changed, 459 insertions(+), 14 deletions(-) create mode 100644 internal/store/deliverylog.go create mode 100644 internal/store/deliverylog_test.go create mode 100644 internal/store/migrations/0007_delivery_log_days.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 81ede64..61ee4b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,17 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version the row still being `queued`, and the row it closes carries the reason `reconcile` (audit 2026-10-12, H3). +### Added + +- **The delivery log measures its own guesses** (migration `0007`). A table + `delivery_log_days` counts, per UTC day, the rows the reconcile sweep + closed, those of them a delivery line later contradicted (the log's status + then replaces the sweep's), the seconds the log tailer could not read + `mail.log`, and the rows still queued past Postfix's queue lifetime. No page + shows it: it is read off a copy of the database after a month, to decide + by a rule written in the audit whether the sweep stays (audit 2026-10-12, + H3). + ## [2.4.2] - 2026-10-14 Release hygiene: stage 0 of the 2026-10-12 audit. Nothing the panel or the diff --git a/docs/audit-2026-10-12.md b/docs/audit-2026-10-12.md index 26202ed..ff0076a 100644 --- a/docs/audit-2026-10-12.md +++ b/docs/audit-2026-10-12.md @@ -599,7 +599,7 @@ has already run is never edited — the next file is added; the published - [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 +- [x] **Measure before deciding the sweep's fate** (owner, 2026-10-12: the keep / remove / `lost` question needs real data, and today the only record of a sweep is a stdout line that dies with the container). A table `delivery_log_days` (migration `0007`, DDL only; a row per UTC day, kept a diff --git a/docs/roadmap.md b/docs/roadmap.md index 71760d3..8ecec57 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`) | 9/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`) | 10/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/docs/schema-migrations.md b/docs/schema-migrations.md index a78afdc..f7dfac9 100644 --- a/docs/schema-migrations.md +++ b/docs/schema-migrations.md @@ -17,8 +17,8 @@ When you add or review a migration, update this file in the same change. | Field | Value | |---|---| -| Chain head | `user_version = 5` | -| Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql`, `0003_send_log_lines.sql`, `0004_notifications.sql`, `0005_send_log_rejected_index.sql`, `0006_sender_check.sql` | +| Chain head | `user_version = 7` | +| Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql`, `0003_send_log_lines.sql`, `0004_notifications.sql`, `0005_send_log_rejected_index.sql`, `0006_sender_check.sql`, `0007_delivery_log_days.sql` | | File mark | `PRAGMA application_id = 0x53503230` ("SP20") | | Database file | `/data/selfpost.db` (bind mount) | | Compatibility | 2.x MINOR releases must boot a `2.0.0` data directory. **Nothing is carried over from 1.x.** | @@ -65,7 +65,8 @@ features after them add files, the first being `0002_inbound_spam_log.sql` of inbound-antispam-panel (`2.1.0`), then `0003_send_log_lines.sql` of delivery-log-storage (`2.2.0`), `0004_notifications.sql` and `0005_send_log_rejected_index.sql` of panel-notifications (`2.3.0`), and -`0006_sender_check.sql` of from-header-binding (`2.4.0`). +`0006_sender_check.sql` of from-header-binding (`2.4.0`), and +`0007_delivery_log_days.sql` of the 2026-10-12 audit (`2.5.0`). --- @@ -79,6 +80,7 @@ delivery-log-storage (`2.2.0`), `0004_notifications.sql` and | 4 | `0004_notifications.sql` | 2.3.0 | DDL | `users.notify` (default 1: every user starts with notifications on) and `notify_state` | | 5 | `0005_send_log_rejected_index.sql` | 2.3.0 | DDL | `idx_send_log_rejected`, a partial index of the refused rows of `send_log` by time; a file of its own because the test installation had run 0004 before the review asked for it | | 6 | `0006_sender_check.sql` | 2.4.0 | DDL | `send_log.reason` and `send_log.header_from` (both default `''`: a row refused earlier keeps no reason), and `sender_check_days` | +| 7 | `0007_delivery_log_days.sql` | 2.5.0 | DDL | `delivery_log_days`: what the reconcile sweep and the log tailer did, a row per UTC day, kept a year, shown nowhere - the measurement that decides the sweep's fate (audit 2026-10-12, stage 1). `send_log.reason` gains the value `reconcile` without a schema change | **Kind:** *DDL* — schema only; *data* — `INSERT`/`UPDATE` that must stay correct for operators upgrading from older 2.x images. diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go index d0e5759..226ff0b 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -32,6 +32,12 @@ type StatusStore interface { UpdateStatus(queueID, recipient, status string) (int64, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) CloseUnreported(queueID, recipient string) (int64, error) + // The counters of delivery_log_days (store/deliverylog.go). + AddReconcileClosed(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 + PruneDeliveryLogDays(now time.Time) error DeleteSendLogBefore(cutoff time.Time) (int64, error) InsertSendLogLines(lines []store.SendLogLine, once bool) (int, error) PruneSendLogLines(cutoff time.Time) (int64, error) @@ -56,6 +62,10 @@ func RetentionInterval() time.Duration { // It is a var so tests can answer without a running Postfix. var queueIDs = postfix.QueueIDs +// retryPolicy reads Postfix's queue lifetimes, for counting the rows left +// queued past them. A var for the same reason. +var retryPolicy = postfix.LoadRetryPolicy + const ( // reconcileInterval is how often the sweep compares stuck rows against the // Postfix queue, and reconcileGrace how long a row is left alone first. @@ -179,6 +189,11 @@ func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{} case <-caughtUp: } + // A message Postfix still works on stays queued until its first retry after + // the queue lifetime, which can be a whole backoff later. + policy := retryPolicy() + overLifetime := policy.LongestLifetime() + policy.MaximalBackoff + t := time.NewTicker(reconcileInterval) defer t.Stop() for { @@ -186,11 +201,26 @@ func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{} case <-ctx.Done(): return case <-t.C: - reconcile(st, time.Now().UTC().Add(-reconcileGrace)) + now := time.Now().UTC() + reconcile(st, now.Add(-reconcileGrace)) + countStuck(st, now, overLifetime) } } } +// countStuck records how many rows are still queued past Postfix's queue +// lifetime - what the log would leave unresolved if there were no sweep. Each +// sweep overwrites the day's value, so a past day keeps its last sweep's. +func countStuck(st StatusStore, now time.Time, overLifetime time.Duration) { + n, err := st.CountQueuedOlderThan(now.Add(-overLifetime)) + if err == nil { + err = st.SetQueuedOverLifetime(now, n) + } + if err != nil { + log.Printf("log-tailer: count rows queued past the queue lifetime: %v", err) + } +} + // reconcile marks as bounced every row still "queued" from before cutoff whose // message Postfix no longer holds, with reason "reconcile". The close itself // is conditional on the row still being queued, so a delivery line read @@ -236,6 +266,9 @@ func reconcile(st StatusStore, cutoff time.Time) { } 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) + } } } @@ -260,8 +293,15 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) } } + pruneDays := func() { + if err := st.PruneDeliveryLogDays(time.Now()); err != nil { + log.Printf("log-tailer: %v", err) + } + } + prune() pruneLines(st) + pruneDays() t := time.NewTicker(retentionInterval) defer t.Stop() for { @@ -271,6 +311,7 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) case <-t.C: prune() pruneLines(st) + pruneDays() } } } @@ -552,12 +593,14 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ } var once sync.Once + blind := &blindness{add: tr.st.AddTailerBlind} ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { select { case <-ctx.Done(): + blind.see(time.Now(), false) tr.record(f, read(), true) // shutdown: the next start resumes here return nil case <-ticker.C: @@ -565,6 +608,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ once.Do(func() { close(caughtUp) }) ni, err := os.Stat(path) if err != nil { + blind.see(time.Now(), true) continue // file briefly gone mid-rotation; try again next tick } pos, _ := f.Seek(0, io.SeekCurrent) @@ -576,12 +620,57 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ drain() if err := openAt(0, io.SeekStart); err != nil { log.Printf("log-tailer: reopen %s: %v", path, err) + blind.see(time.Now(), true) continue } + blind.see(time.Now(), false) tr.adopt(f) continue } + blind.see(time.Now(), false) tr.record(f, read(), false) } } } + +// blindness measures the stretches in which follow cannot read mail.log - the +// file gone, reopening it failing - for delivery_log_days. A stretch is +// stored when it ends, and every minute while it lasts, so a long one is not +// lost to a restart and lands on the days it covered. A stretch shorter than +// blindMin is not counted: a rotation leaves the name missing for a tick +// between rename and create, and that loses nothing - the new file is read +// from its start. +type blindness struct { + add func(from time.Time, d time.Duration) error + since time.Time // start of the current stretch; zero when reading + long bool // the current stretch has passed blindMin +} + +const blindMin = 3 * time.Second + +func (b *blindness) see(now time.Time, blind bool) { + if blind && b.since.IsZero() { + b.since = now + return + } + if b.since.IsZero() { + return + } + d := now.Sub(b.since) + if d >= blindMin { + b.long = true + } + if blind && d < time.Minute { + return + } + if b.long { + if err := b.add(b.since, d); err != nil { + log.Printf("log-tailer: count the time mail.log could not be read: %v", err) + } + } + if blind { + b.since = now // the stretch goes on; the next part starts here + return + } + b.since, b.long = time.Time{}, false +} diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go index a1b050e..cf6fc46 100644 --- a/internal/logtail/logtail_test.go +++ b/internal/logtail/logtail_test.go @@ -147,6 +147,11 @@ type captureStore struct { // queued is what ListQueuedOlderThan returns, for the reconcile sweep. queued []store.QueuedDelivery queuedErr error + // closed sums AddReconcileClosed; stuckCount is what CountQueuedOlderThan + // answers and stuck what SetQueuedOverLifetime was given. + closed int64 + stuckCount int64 + stuck []int64 // lines are the log lines kept for messages' pages; pruned is the cutoff of // each PruneSendLogLines call. @@ -218,6 +223,28 @@ func (c *captureStore) CloseUnreported(queueID, recipient string) (int64, error) func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil } +// The counters of delivery_log_days: closed and stuck are kept for the sweep's +// tests, the rest is not looked at. +func (c *captureStore) AddReconcileClosed(_ time.Time, n int64) error { + c.mu.Lock() + defer c.mu.Unlock() + c.closed += 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() + defer c.mu.Unlock() + return c.stuckCount, nil +} +func (c *captureStore) SetQueuedOverLifetime(_ time.Time, n int64) error { + c.mu.Lock() + defer c.mu.Unlock() + c.stuck = append(c.stuck, n) + return nil +} +func (c *captureStore) PruneDeliveryLogDays(time.Time) error { return nil } + func (c *captureStore) ListQueuedOlderThan(time.Time) ([]store.QueuedDelivery, error) { c.mu.Lock() defer c.mu.Unlock() @@ -444,6 +471,59 @@ func TestReconcileClosesOnlyWhatPostfixNoLongerHolds(t *testing.T) { 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) + } +} + +// Each sweep records how many rows are queued past the queue lifetime, as a +// level for the day. +func TestCountStuckRecordsTheLevel(t *testing.T) { + cs := &captureStore{stuckCount: 3} + countStuck(cs, time.Now(), time.Hour) + cs.stuckCount = 0 + countStuck(cs, time.Now(), time.Hour) + if len(cs.stuck) != 2 || cs.stuck[0] != 3 || cs.stuck[1] != 0 { + t.Fatalf("levels recorded = %v, want [3 0]", cs.stuck) + } +} + +// The tailer's blind time: a rotation's one-tick gap is not counted, a longer +// stretch is, a stretch that lasts is stored a minute at a time from where the +// last part ended, and reading again ends it. +func TestBlindnessCountsStretchesThatMatter(t *testing.T) { + type part struct { + from time.Time + d time.Duration + } + var got []part + b := &blindness{add: func(from time.Time, d time.Duration) error { + got = append(got, part{from, d}) + return nil + }} + t0 := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + at := func(s int) time.Time { return t0.Add(time.Duration(s) * time.Second) } + + b.see(at(0), true) // rotation: gone for a tick + b.see(at(1), false) + b.see(at(2), false) + if len(got) != 0 { + t.Fatalf("a one-second gap was counted: %v", got) + } + + for s := 10; s <= 80; s++ { // blind for 75 s + b.see(at(s), true) + } + b.see(at(85), false) + want := []part{{at(10), 60 * time.Second}, {at(70), 15 * time.Second}} + if len(got) != len(want) { + t.Fatalf("parts = %v, want %v", got, want) + } + for i := range want { + if !got[i].from.Equal(want[i].from) || got[i].d != want[i].d { + t.Errorf("part %d = %v, want %v", i, got[i], want[i]) + } + } } // A queue that cannot be listed says nothing about any message — treating the diff --git a/internal/logtail/retention_test.go b/internal/logtail/retention_test.go index 2f41aa0..cef26b9 100644 --- a/internal/logtail/retention_test.go +++ b/internal/logtail/retention_test.go @@ -27,6 +27,12 @@ func (s *retentionProbeStore) CloseUnreported(string, string) (int64, error) { return 0, 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 } +func (s *retentionProbeStore) SetQueuedOverLifetime(time.Time, int64) error { return nil } +func (s *retentionProbeStore) PruneDeliveryLogDays(time.Time) error { return nil } + func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, error) { s.mu.Lock() s.cutoff = cutoff diff --git a/internal/store/deliverylog.go b/internal/store/deliverylog.go new file mode 100644 index 0000000..390a7f4 --- /dev/null +++ b/internal/store/deliverylog.go @@ -0,0 +1,106 @@ +package store + +import ( + "fmt" + "time" +) + +// The counters of delivery_log_days (migration 0007): what the reconcile sweep +// and the log tailer did, a row per UTC day, read off a copy of the database +// to decide the sweep's fate (audit 2026-10-12, stage 1). No page shows them. + +// deliveryLogKeepDays is how long a day of the counters is kept. +const deliveryLogKeepDays = 366 + +// addDeliveryLogDay adds n to one counter of the day of at. column is one of +// the fixed names below, never anything a request carries. +func (s *Store) addDeliveryLogDay(at time.Time, column string, n int64) error { + if n <= 0 { + return nil + } + _, err := s.db.Exec(fmt.Sprintf( + `INSERT INTO delivery_log_days (day, %[1]s) VALUES (?, ?) + ON CONFLICT (day) DO UPDATE SET %[1]s = %[1]s + excluded.%[1]s`, column), + at.UTC().Format(dayLayout), n, + ) + if err != nil { + return fmt.Errorf("delivery log day, %s: %w", column, err) + } + return nil +} + +// AddReconcileClosed counts n rows the reconcile sweep closed. +func (s *Store) AddReconcileClosed(at time.Time, n int64) error { + return s.addDeliveryLogDay(at, "reconcile_closed", n) +} + +// AddTailerBlind counts a stretch the log tailer could not read mail.log, +// on the day it began. Under a second counts nothing. +func (s *Store) AddTailerBlind(from time.Time, d time.Duration) error { + return s.addDeliveryLogDay(from, "tailer_blind_seconds", int64(d/time.Second)) +} + +// SetQueuedOverLifetime records how many rows are still queued past Postfix's +// queue lifetime. It is a level, not a sum: each sweep overwrites the day's +// value, so a past day keeps what its last sweep saw. +func (s *Store) SetQueuedOverLifetime(at time.Time, n int64) error { + _, err := s.db.Exec( + `INSERT INTO delivery_log_days (day, queued_over_lifetime) VALUES (?, ?) + ON CONFLICT (day) DO UPDATE SET queued_over_lifetime = excluded.queued_over_lifetime`, + at.UTC().Format(dayLayout), n, + ) + if err != nil { + return fmt.Errorf("delivery log day, queued_over_lifetime: %w", err) + } + return nil +} + +// CountQueuedOlderThan counts the rows still "queued" that were accepted +// before cutoff. created_at is RFC3339 UTC, so the strings compare as times. +func (s *Store) CountQueuedOlderThan(cutoff time.Time) (int64, error) { + var n int64 + err := s.db.QueryRow( + `SELECT COUNT(*) FROM send_log WHERE status = ? AND queue_id <> '' AND created_at < ?`, + StatusQueued, cutoff.UTC().Format(time.RFC3339), + ).Scan(&n) + if err != nil { + return 0, fmt.Errorf("count queued send_log rows: %w", err) + } + return n, nil +} + +// DeliveryLogDay is one day of the counters, for tests and for reading a copy. +type DeliveryLogDay struct { + Day string + Closed, Contradicted, BlindSeconds, Stuck int64 +} + +// DeliveryLogDays returns the counted days, oldest first. +func (s *Store) DeliveryLogDays() ([]DeliveryLogDay, error) { + rows, err := s.db.Query( + `SELECT day, reconcile_closed, reconcile_contradicted, tailer_blind_seconds, queued_over_lifetime + FROM delivery_log_days ORDER BY day`) + if err != nil { + return nil, fmt.Errorf("delivery log days: %w", err) + } + defer rows.Close() + var out []DeliveryLogDay + for rows.Next() { + var d DeliveryLogDay + if err := rows.Scan(&d.Day, &d.Closed, &d.Contradicted, &d.BlindSeconds, &d.Stuck); err != nil { + return nil, fmt.Errorf("scan delivery log day: %w", err) + } + out = append(out, d) + } + return out, rows.Err() +} + +// PruneDeliveryLogDays deletes the days older than a year. +func (s *Store) PruneDeliveryLogDays(now time.Time) error { + _, err := s.db.Exec(`DELETE FROM delivery_log_days WHERE day < ?`, + now.UTC().AddDate(0, 0, -deliveryLogKeepDays).Format(dayLayout)) + if err != nil { + return fmt.Errorf("prune delivery log days: %w", err) + } + return nil +} diff --git a/internal/store/deliverylog_test.go b/internal/store/deliverylog_test.go new file mode 100644 index 0000000..4297960 --- /dev/null +++ b/internal/store/deliverylog_test.go @@ -0,0 +1,69 @@ +package store + +import ( + "testing" + "time" +) + +// The counters add up per UTC day, the stuck level is overwritten rather than +// added, nothing is written for nothing, and a year is kept. +func TestDeliveryLogDays(t *testing.T) { + st := openTestStore(t) + day1 := time.Date(2026, 10, 14, 23, 59, 0, 0, time.UTC) + day2 := day1.Add(2 * time.Minute) + + must := func(err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } + } + must(st.AddReconcileClosed(day1, 2)) + must(st.AddReconcileClosed(day1, 3)) + must(st.AddReconcileClosed(day1, 0)) + must(st.AddTailerBlind(day1, 1500*time.Millisecond)) + must(st.AddTailerBlind(day1, 900*time.Millisecond)) // under a second: nothing + must(st.SetQueuedOverLifetime(day1, 4)) + must(st.SetQueuedOverLifetime(day1, 1)) + must(st.AddTailerBlind(day2, time.Minute)) + + days, err := st.DeliveryLogDays() + must(err) + want := []DeliveryLogDay{ + {Day: "2026-10-14", Closed: 5, BlindSeconds: 1, Stuck: 1}, + {Day: "2026-10-15", BlindSeconds: 60}, + } + if len(days) != len(want) { + t.Fatalf("days = %+v, want %+v", days, want) + } + for i := range want { + if days[i] != want[i] { + t.Errorf("day %d = %+v, want %+v", i, days[i], want[i]) + } + } + + must(st.PruneDeliveryLogDays(day2.AddDate(1, 0, 1))) + days, err = st.DeliveryLogDays() + must(err) + if len(days) != 1 || days[0].Day != "2026-10-15" { + t.Fatalf("after pruning a year and a day later: %+v, want 2026-10-15 alone", days) + } +} + +func TestCountQueuedOlderThan(t *testing.T) { + st := openTestStore(t) + for _, e := range []SendLogEntry{{QueueID: "Q1", To: "a@example.net"}, {QueueID: "Q2", To: "b@example.net"}} { + if err := st.InsertQueued(e); err != nil { + t.Fatal(err) + } + } + if _, err := st.UpdateStatus("Q2", "b@example.net", StatusSent); err != nil { + t.Fatal(err) + } + if n, err := st.CountQueuedOlderThan(time.Now().Add(time.Hour)); err != nil || n != 1 { + t.Fatalf("queued rows before an hour from now = %d, %v; want 1", n, err) + } + if n, err := st.CountQueuedOlderThan(time.Now().Add(-time.Hour)); err != nil || n != 0 { + t.Fatalf("queued rows before an hour ago = %d, %v; want 0", n, err) + } +} diff --git a/internal/store/migrations/0007_delivery_log_days.sql b/internal/store/migrations/0007_delivery_log_days.sql new file mode 100644 index 0000000..a17ab4f --- /dev/null +++ b/internal/store/migrations/0007_delivery_log_days.sql @@ -0,0 +1,19 @@ +-- What the delivery log's own machinery did, a row per UTC day that has +-- something to say, kept a year (audit 2026-10-12, stage 1). These are the +-- numbers the reconcile sweep's fate is decided by; no page reads them, they +-- are read off a copy of the database. DDL only; nothing existing changes. +-- reconcile_closed rows the sweep closed as bounced (reason 'reconcile') +-- reconcile_contradicted rows closed by the sweep that a delivery line later +-- rewrote - the log wins, and is counted here +-- tailer_blind_seconds seconds the log tailer could not read mail.log +-- (the file gone, reopening it failing) +-- queued_over_lifetime rows still queued past Postfix's queue lifetime, +-- as measured by the day's last sweep (a level, not +-- a sum) +CREATE TABLE delivery_log_days ( + day TEXT PRIMARY KEY, + reconcile_closed INTEGER NOT NULL DEFAULT 0, + reconcile_contradicted INTEGER NOT NULL DEFAULT 0, + tailer_blind_seconds INTEGER NOT NULL DEFAULT 0, + queued_over_lifetime INTEGER NOT NULL DEFAULT 0 +); diff --git a/internal/store/schema_test.go b/internal/store/schema_test.go index 7073991..820cb64 100644 --- a/internal/store/schema_test.go +++ b/internal/store/schema_test.go @@ -60,6 +60,7 @@ func TestBaselineSchema(t *testing.T) { "send_log_lines": "id queue_id logged_at line", "notify_state": "key phase count mark since sent_at", "sender_check_days": "day down_seconds deferred", + "delivery_log_days": "day reconcile_closed reconcile_contradicted tailer_blind_seconds queued_over_lifetime", } rows, err := st.db.Query("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name") if err != nil { @@ -114,8 +115,8 @@ func TestBaselineSchema(t *testing.T) { } var version, appID int64 - if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 6 { - t.Errorf("user_version = %d, %v; want 6", version, err) + if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 7 { + t.Errorf("user_version = %d, %v; want 7", version, err) } if err := st.db.QueryRow("PRAGMA application_id").Scan(&appID); err != nil || appID != applicationID { t.Errorf("application_id = %#x, %v; want %#x", appID, err, applicationID) @@ -191,8 +192,8 @@ func TestOpenMigratesA20Database(t *testing.T) { } defer st.Close() var version int - if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 6 { - t.Errorf("user_version = %d, %v; want 6", version, err) + if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 7 { + t.Errorf("user_version = %d, %v; want 7", version, err) } if ok, err := st.UserExists(); err != nil || !ok { t.Errorf("the user of the 2.0 database: exists = %v, %v", ok, err) diff --git a/internal/store/sendlog.go b/internal/store/sendlog.go index e810705..3e7ecc9 100644 --- a/internal/store/sendlog.go +++ b/internal/store/sendlog.go @@ -83,16 +83,46 @@ func (s *Store) InsertRejected(e SendLogEntry) error { // case between the milter (envelope) and the delivery log. It returns the // number of rows updated so the caller can tell whether the line matched a // journal entry. +// +// A row the reconcile sweep closed (reason "reconcile") that a delivery line +// now speaks of was closed on a guess the log could have settled: the log +// wins, the reason is cleared, and the day's reconcile_contradicted counts it +// (delivery_log_days) - once, since the row no longer carries the reason. func (s *Store) UpdateStatus(queueID, recipient, status string) (int64, error) { - now := time.Now().UTC().Format(time.RFC3339) - res, err := s.db.Exec( - `UPDATE send_log SET status = ?, updated_at = ? + now := time.Now().UTC() + tx, err := s.db.Begin() + if err != nil { + return 0, fmt.Errorf("update send_log status: %w", err) + } + defer tx.Rollback() + var contradicted int64 + if err := tx.QueryRow( + `SELECT COUNT(*) FROM send_log WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE AND reason = ?`, + queueID, recipient, ReasonReconcile, + ).Scan(&contradicted); err != nil { + return 0, fmt.Errorf("update send_log status: %w", err) + } + res, err := tx.Exec( + `UPDATE send_log SET status = ?, updated_at = ?, + reason = CASE WHEN reason = ? THEN '' ELSE reason END WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE`, - status, now, queueID, recipient, + status, now.Format(time.RFC3339), ReasonReconcile, queueID, recipient, ) if err != nil { return 0, fmt.Errorf("update send_log status: %w", err) } + if contradicted > 0 { + if _, err := tx.Exec( + `INSERT INTO delivery_log_days (day, reconcile_contradicted) VALUES (?, ?) + ON CONFLICT (day) DO UPDATE SET reconcile_contradicted = reconcile_contradicted + excluded.reconcile_contradicted`, + now.Format(dayLayout), contradicted, + ); err != nil { + return 0, fmt.Errorf("update send_log status: count the contradicted sweep: %w", err) + } + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("update send_log status: %w", err) + } n, _ := res.RowsAffected() return n, nil } diff --git a/internal/store/sendlog_test.go b/internal/store/sendlog_test.go index 67cbe7a..3c9aae1 100644 --- a/internal/store/sendlog_test.go +++ b/internal/store/sendlog_test.go @@ -132,6 +132,38 @@ func TestCloseUnreportedClosesOnlyQueuedRows(t *testing.T) { } } +// A delivery line about a row the sweep closed proves the sweep guessed what +// the log could have said: the log's status wins, the reason goes, and the day +// counts one contradiction - once, however often the line is read again. +func TestUpdateStatusCountsAContradictedSweep(t *testing.T) { + st := openTestStore(t) + if err := st.InsertQueued(SendLogEntry{QueueID: "Q1", To: "a@example.net"}); err != nil { + t.Fatal(err) + } + if _, err := st.CloseUnreported("Q1", "a@example.net"); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if n, err := st.UpdateStatus("Q1", "A@example.net", StatusSent); err != nil || n != 1 { + t.Fatalf("UpdateStatus = %d, %v", n, err) + } + } + var status, reason string + if err := st.db.QueryRow(`SELECT status, reason FROM send_log WHERE queue_id = 'Q1'`).Scan(&status, &reason); err != nil { + t.Fatal(err) + } + if status != StatusSent || reason != "" { + t.Fatalf("status %q reason %q, want sent and no reason", status, reason) + } + days, err := st.DeliveryLogDays() + if err != nil { + t.Fatal(err) + } + if len(days) != 1 || days[0].Contradicted != 1 { + t.Fatalf("days = %+v, want one day with one contradiction", days) + } +} + // 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