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 <noreply@anthropic.com>
2026-10-14 14:20:44 +03:00
parent 9adb44845c
commit ffefa9d6df
13 changed files with 459 additions and 14 deletions
@@ -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
+}
@@ -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
@@ -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