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 <noreply@anthropic.com>
This commit is contained in:
mix
2026-10-14 14:16:26 +03:00
co-authored by Claude Opus 5.5
parent dfe10f6c27
commit 9adb44845c
9 changed files with 88 additions and 8 deletions
+8 -4
View File
@@ -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)
+10 -2
View File
@@ -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)
}
}
+4
View File
@@ -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