From c452606175744474805c5162b3563d36318bbb7e Mon Sep 17 00:00:00 2001 From: Mix Date: Wed, 14 Oct 2026 16:28:13 +0300 Subject: [PATCH] logtail: the deferred counter remembers a place in the log, not a second (stage 2b) The counter passed over, while catching up after a start, every warning no later than the last one counted - by time; mail.log stamps whole seconds, so a second warning of that second, later in the file, was lost. - The tailer gives each line its place (Place: the file's fingerprint and the offset just past the line): follow tracks the bytes the reader hands over, readRotated likewise in mail.log.1. Run hands tr.line to the counter. - store.AddSenderCheckDeferred(at, n, place) counts, moves the last-deferral mark (never back - an upsert WHERE) and saves the place (settings sender_check_counted_place) in one transaction. - Catching up, a warning in the file of the last one counted is passed over by place (end <= counted end); in another file - a rotated log read again - by time, as before. The fingerprint carries the inode, so the place stays true across a rotation. Tests: two warnings of one second, an unclean stop between them, all lines read again - each counted once; the store saves the place with the count and keeps the latest deferral time. Existing counter tests pass a zero place (time rule). CHANGELOG [Unreleased], audit plan line ticked, roadmap 24/62. Evidence: go vet ./... clean; go test ./internal/store, ./internal/logtail ok (but for the Windows-only test); the server run 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 | 16 ++++++-- internal/logtail/offset.go | 32 ++++++++++++++++ internal/logtail/sendercheck.go | 57 +++++++++++++++++++--------- internal/logtail/sendercheck_test.go | 56 ++++++++++++++++++++++++--- internal/store/sendercheck.go | 53 +++++++++++++++++++------- internal/store/sendercheck_test.go | 25 ++++++++++-- 9 files changed, 204 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d09362..46a95e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,12 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version in the log, with the time from which nothing was read. The log's fingerprint now includes the file's inode, so two files that begin alike are still told apart (audit 2026-10-12, stage 2b). +- **A second deferral of the same second is counted after a restart.** The + Sender check's count of deferred mail remembered the time of the last + line it counted, to pass over lines read again after an unclean stop; + `mail.log` stamps whole seconds, so a second warning of that second was + passed over too. It now remembers the place in the log, saved with the + count in one transaction (audit 2026-10-12, stage 2b). ## [2.5.0] - 2026-10-14 diff --git a/docs/audit-2026-10-12.md b/docs/audit-2026-10-12.md index 9f32e75..7b4abcf 100644 --- a/docs/audit-2026-10-12.md +++ b/docs/audit-2026-10-12.md @@ -751,7 +751,7 @@ order - each stands on the one before. rotated twice: logged, with from when to when nothing was read. A test that stops the tailer, appends, rotates, appends, starts it, and finds both halves once. - **Opus** -- [ ] **The deferred counter remembers a place in the log, not a second** +- [x] **The deferred counter remembers a place in the log, not a second** (owner, 2026-10-14, from the Not done list of from-header-binding; he chose it over leaving the case written down). After an unclean stop the tailer re-reads lines it had read, and the counter passes over every diff --git a/docs/roadmap.md b/docs/roadmap.md index 6f0caa6..021371d 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`, stages 1+2 `2.5.0`; next stage 2b → `2.5.1`) | 23/62 | [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`, stages 1+2 `2.5.0`; next stage 2b → `2.5.1`) | 24/62 | [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 38b9e9e..2125491 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -162,7 +162,8 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa } keeper := newLineKeeper(st) - return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, keeper.flush, func(line string) { + tr := &tracker{st: st, path: path} + return follow(ctx, path, tr, caughtUp, keeper.flush, func(line string) { // Until the first read reaches end-of-file the lines are those written // while the panel was down - and, after an unclean stop, some it had // read already. @@ -177,7 +178,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa journal.handle(line, now, catchingUp) } if deferred != nil { - deferred.handle(line, now, catchingUp) + deferred.handle(line, now, catchingUp, tr.line) } keeper.add(line, now, catchingUp) queueID, recipient, status, ok := parseDelivery(line) @@ -569,6 +570,9 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ r *bufio.Reader info os.FileInfo pending string + // consumed is the offset just past what the reader has handed over, + // pending included: where a complete line ends. + consumed int64 ) openAt := func(offset int64, whence int) error { nf, err := os.Open(path) @@ -588,6 +592,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ f.Close() } f, r, info, pending = nf, bufio.NewReader(nf), ni, "" + consumed = offset return nil } @@ -602,9 +607,12 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ case <-time.After(pollInterval): } } - if _, err := f.Seek(tr.resume(f), io.SeekStart); err != nil { + start := tr.resume(f) + if _, err := f.Seek(start, io.SeekStart); err != nil { log.Printf("log-tailer: seek %s: %v", path, err) + start = 0 } + consumed = start r.Reset(f) // the reader buffered from the pre-seek position // Rotated while the panel was down: the end of the old file first. if tr.readRotated(handle) { @@ -620,6 +628,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ defer flush() for { line, err := r.ReadString('\n') + consumed += int64(len(line)) if err == io.EOF { pending += line // hold the partial line until it completes return @@ -630,6 +639,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{ } full := pending + line pending = "" + tr.line = Place{FP: tr.fp, End: consumed} handle(strings.TrimRight(full, "\r\n")) } } diff --git a/internal/logtail/offset.go b/internal/logtail/offset.go index 8e494f7..b30e6f4 100644 --- a/internal/logtail/offset.go +++ b/internal/logtail/offset.go @@ -46,6 +46,34 @@ type tracker struct { // no longer the log - rotated while the panel was down; prev is that place. behind bool prev store.LogtailState + + // line is the place of the line being handled: set by follow and + // readRotated before each call of handle. + line Place +} + +// Place is where a line of the log ends: the fingerprint of its file and the +// byte offset just past it. A rotated file keeps its fingerprint (the inode +// is part of it), so a place stays true across a rotation. +type Place struct { + FP string + End int64 +} + +// String is the place as it is saved: "@". +func (p Place) String() string { return fmt.Sprintf("%s@%d", p.FP, p.End) } + +// parsePlace reads a saved place; the zero Place for anything else. +func parsePlace(s string) Place { + i := strings.LastIndexByte(s, '@') + if i < 0 { + return Place{} + } + var end int64 + if _, err := fmt.Sscan(s[i+1:], &end); err != nil { + return Place{} + } + return Place{FP: s[:i], End: end} } // resume returns the byte offset the tailer should start reading f at, having @@ -125,10 +153,14 @@ func (t *tracker) readRotated(handle func(string)) bool { lost(fmt.Sprintf("seek %s: %v", rotated, err)) return false } + fp := fingerprintOf(f) + end := t.prev.Offset r := bufio.NewReader(f) for { line, err := r.ReadString('\n') + end += int64(len(line)) if line != "" { + t.line = Place{FP: fp, End: end} handle(strings.TrimRight(line, "\r\n")) } if err != nil { diff --git a/internal/logtail/sendercheck.go b/internal/logtail/sendercheck.go index e75f743..0d68c73 100644 --- a/internal/logtail/sendercheck.go +++ b/internal/logtail/sendercheck.go @@ -46,8 +46,11 @@ import ( // SenderCheckStore is where the count goes, and where the time of the last // counted line is kept. *store.Store satisfies it. type SenderCheckStore interface { - AddSenderCheckDeferred(at time.Time, n int64) error + // AddSenderCheckDeferred counts and saves the place in the log reached, + // in one transaction. + AddSenderCheckDeferred(at time.Time, n int64, place string) error SenderCheckDeferredAt() (time.Time, error) + SenderCheckCountedPlace() (string, error) } // SenderCheck turns the counting on: the store and the socket Postfix was @@ -72,13 +75,16 @@ type deferredCounter struct { // listening, and a reply that did not come. connect, broken string failed map[string]bool // smtpd pid -> its current client was counted already - // last is the time of the latest line counted, by this run or an earlier - // one. While the tailer catches up after a start it may be reading lines - // it had read before an unclean stop; a warning no later than this was - // counted then. mail.log's stamps are whole seconds, so a warning of the - // very second of the last one counted can be passed over - once, at a - // restart - which is the cheaper mistake than counting a stretch twice. - last time.Time + // counted is the place in the log of the latest warning counted, by this + // run or an earlier one, saved with the count. While the tailer catches up + // after a start it may be reading lines it had read before an unclean + // stop; a warning at or before that place was counted then. Until 2.5.1 + // this was a time, and mail.log's stamps are whole seconds: a second + // warning of the same second was passed over (audit 2026-10-12, stage + // 2b). A place in another file - read again from a rotated log - has no + // order to compare with, and last, its time, decides. + counted Place + last time.Time } func newDeferredCounter(sc *SenderCheck) *deferredCounter { @@ -86,8 +92,13 @@ func newDeferredCounter(sc *SenderCheck) *deferredCounter { if err != nil { log.Printf("log-tailer: sender check statistics: %v", err) } + place, err := sc.Store.SenderCheckCountedPlace() + if err != nil { + log.Printf("log-tailer: sender check statistics: %v", err) + } return &deferredCounter{ last: last, + counted: parsePlace(place), st: sc.Store, connect: "warning: connect to Milter service unix:" + sc.Socket + ":", broken: "warning: milter unix:" + sc.Socket + ":", @@ -95,9 +106,9 @@ func newDeferredCounter(sc *SenderCheck) *deferredCounter { } } -// handle looks at one mail.log line. catchingUp is true until the tailer has -// read to the end of the log for the first time. -func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) { +// handle looks at one mail.log line, which ends at place. catchingUp is true +// until the tailer has read to the end of the log for the first time. +func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool, place Place) { at, rest := lineTime(line, now) m := milterLineRe.FindStringSubmatch(rest) if m == nil { @@ -107,7 +118,7 @@ func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) { if program == "cleanup" { // A message cleanup could not finish because the check did not reply. if strings.HasPrefix(text, c.broken) { - c.count(at, catchingUp) + c.count(at, catchingUp, place) } return } @@ -126,19 +137,31 @@ func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) { c.failed = make(map[string]bool) } c.failed[pid] = true - c.count(at, catchingUp) + c.count(at, catchingUp, place) } } -// count adds one deferred message at at, unless it is a line read again. -func (c *deferredCounter) count(at time.Time, catchingUp bool) { - if catchingUp && !at.After(c.last) { +// count adds one deferred message at at, the line ending at place, unless it +// is a line read again. +func (c *deferredCounter) count(at time.Time, catchingUp bool, place Place) { + if catchingUp && c.countedAlready(at, place) { return } if at.After(c.last) { c.last = at } - if err := c.st.AddSenderCheckDeferred(at, 1); err != nil { + c.counted = place + if err := c.st.AddSenderCheckDeferred(at, 1, place.String()); err != nil { log.Printf("log-tailer: sender check statistics: %v", err) } } + +// countedAlready says whether a warning read while catching up was counted +// before: by its place when it is in the file of the last one counted, by its +// time otherwise. +func (c *deferredCounter) countedAlready(at time.Time, place Place) bool { + if place.FP != "" && place.FP == c.counted.FP { + return place.End <= c.counted.End + } + return !at.After(c.last) +} diff --git a/internal/logtail/sendercheck_test.go b/internal/logtail/sendercheck_test.go index 8d19c8e..c64e5ff 100644 --- a/internal/logtail/sendercheck_test.go +++ b/internal/logtail/sendercheck_test.go @@ -6,13 +6,17 @@ import ( ) type deferredStore struct { - days map[string]int64 - last time.Time + days map[string]int64 + last time.Time + place string } func (d *deferredStore) SenderCheckDeferredAt() (time.Time, error) { return d.last, nil } -func (d *deferredStore) AddSenderCheckDeferred(at time.Time, n int64) error { +func (d *deferredStore) SenderCheckCountedPlace() (string, error) { return d.place, nil } + +func (d *deferredStore) AddSenderCheckDeferred(at time.Time, n int64, place string) error { + d.place = place if d.days == nil { d.days = make(map[string]int64) } @@ -77,7 +81,7 @@ func TestDeferredCounter(t *testing.T) { // cleanup that writes it. `2026-10-13T00:01:00.000000+00:00 mail postfix/cleanup[1195]: B6B484831946: milter-reject: END-OF-MESSAGE from unknown[10.77.0.1]: 5.7.1 From: director@other.e2e.test - not an address this application sends as; from= to= proto=ESMTP helo=`, } { - c.handle(line, now, false) + c.handle(line, now, false, Place{}) } if len(st.days) != 2 || st.days["2026-10-12"] != 1 || st.days["2026-10-13"] != 2 { t.Fatalf("deferred per day = %v, want one on the first day and two on the second", st.days) @@ -100,7 +104,7 @@ func TestDeferredCounterDoesNotCountALogReadAgain(t *testing.T) { c := newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"}) for _, line := range first { - c.handle(line, now, false) + c.handle(line, now, false, Place{}) } if st.days["2026-10-12"] != 2 { t.Fatalf("first run counted %d, want 2", st.days["2026-10-12"]) @@ -109,9 +113,49 @@ func TestDeferredCounterDoesNotCountALogReadAgain(t *testing.T) { // A new run: the same lines again, then what the outage added. c = newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"}) for _, line := range append(first, outage("13", "2026-10-12T16:00:09.00000")...) { - c.handle(line, now, true) + c.handle(line, now, true, Place{}) } if st.days["2026-10-12"] != 3 { t.Fatalf("after the restart the day has %d, want 3", st.days["2026-10-12"]) } } + +// Two warnings of the same second, the panel stopped uncleanly between them: +// the next start reads both again, and the place of the first - not its +// second - tells them apart. Each is counted once (audit 2026-10-12, stage +// 2b; a time alone passed the second over). +func TestDeferredCounterRemembersAPlaceNotASecond(t *testing.T) { + st := &deferredStore{} + now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) + lines := []string{ + `2026-10-12T16:00:00+00:00 mail postfix/smtpd[11]: connect from unknown[10.77.0.1]`, + `2026-10-12T16:00:01+00:00 mail postfix/smtpd[11]: warning: connect to Milter service unix:/run/selfpost/sender.sock: Connection refused`, + `2026-10-12T16:00:01+00:00 mail postfix/smtpd[12]: connect from unknown[10.77.0.1]`, + `2026-10-12T16:00:01+00:00 mail postfix/smtpd[12]: warning: connect to Milter service unix:/run/selfpost/sender.sock: Connection refused`, + } + // The place each line ends at, in one file. + places := make([]Place, len(lines)) + end := int64(0) + for i, l := range lines { + end += int64(len(l)) + 1 + places[i] = Place{FP: "abc:1.2", End: end} + } + + c := newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"}) + for i := 0; i < 2; i++ { // stopped after the first warning + c.handle(lines[i], now, false, places[i]) + } + if st.days["2026-10-12"] != 1 { + t.Fatalf("before the stop: %d, want 1", st.days["2026-10-12"]) + } + + // Started again: the saved offset was from before the stop, so all four + // lines are read again. + c = newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"}) + for i, l := range lines { + c.handle(l, now, true, places[i]) + } + if st.days["2026-10-12"] != 2 { + t.Fatalf("after the restart the day has %d, want 2", st.days["2026-10-12"]) + } +} diff --git a/internal/store/sendercheck.go b/internal/store/sendercheck.go index 3861631..d30cc5f 100644 --- a/internal/store/sendercheck.go +++ b/internal/store/sendercheck.go @@ -30,6 +30,9 @@ const ( // settingSenderCheckDeferredAt is when a message was last deferred because // the check did not answer (RFC3339 UTC). settingSenderCheckDeferredAt = "sender_check_deferred_at" + // settingSenderCheckCountedPlace is the place in mail.log of the last + // warning counted ("@"), saved with the count. + settingSenderCheckCountedPlace = "sender_check_counted_place" ) // senderCheckKeepDays is how long a day of the statistics is kept. @@ -80,33 +83,55 @@ func (s *Store) AddSenderCheckDown(at time.Time, d time.Duration) error { } // AddSenderCheckDeferred counts, on the day of at, n messages deferred because -// the check did not answer, and remembers at as the last such moment. -func (s *Store) AddSenderCheckDeferred(at time.Time, n int64) error { +// the check did not answer, remembers at as the last such moment, and saves +// place - where in mail.log the counted line ends - in the same transaction: +// a count without its place, or the other way round, would be counted again +// or lost at the next start (audit 2026-10-12, stage 2b). +func (s *Store) AddSenderCheckDeferred(at time.Time, n int64, place string) error { if n <= 0 { return nil } - _, err := s.db.Exec( - `INSERT INTO sender_check_days (day, deferred) VALUES (?, ?) - ON CONFLICT (day) DO UPDATE SET deferred = deferred + excluded.deferred`, - at.UTC().Format(dayLayout), n, - ) + tx, err := s.db.Begin() if err != nil { return fmt.Errorf("sender check deferred: %w", err) } - // Lines are read in order, but a re-read log must not move the mark back. - last, err := s.GetSetting(settingSenderCheckDeferredAt) - if err != nil { - return err + defer tx.Rollback() + if _, err := tx.Exec( + `INSERT INTO sender_check_days (day, deferred) VALUES (?, ?) + ON CONFLICT (day) DO UPDATE SET deferred = deferred + excluded.deferred`, + at.UTC().Format(dayLayout), n, + ); err != nil { + return fmt.Errorf("sender check deferred: %w", err) } // To the fraction of a second the log gives, and of a fixed width so that - // the strings compare as the times do. + // the strings compare as the times do; a re-read log must not move the + // mark back. stamp := at.UTC().Format("2006-01-02T15:04:05.000000Z") - if stamp > last { - return s.SetSetting(settingSenderCheckDeferredAt, stamp) + setting := func(key, value, cond string) error { + _, err := tx.Exec(`INSERT INTO settings (key, value) VALUES (?, ?) + ON CONFLICT (key) DO UPDATE SET value = excluded.value`+cond, key, value) + return err + } + if err := setting(settingSenderCheckDeferredAt, stamp, " WHERE excluded.value > settings.value"); err != nil { + return fmt.Errorf("sender check deferred: %w", err) + } + if place != "" { + if err := setting(settingSenderCheckCountedPlace, place, ""); err != nil { + return fmt.Errorf("sender check deferred: %w", err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("sender check deferred: %w", err) } return nil } +// SenderCheckCountedPlace is the place in mail.log of the last warning +// counted; "" before any was. +func (s *Store) SenderCheckCountedPlace() (string, error) { + return s.GetSetting(settingSenderCheckCountedPlace) +} + // SenderCheckDeferredAt is when a message was last counted as deferred; zero // if none was. func (s *Store) SenderCheckDeferredAt() (time.Time, error) { diff --git a/internal/store/sendercheck_test.go b/internal/store/sendercheck_test.go index 1543acb..a0683ab 100644 --- a/internal/store/sendercheck_test.go +++ b/internal/store/sendercheck_test.go @@ -42,12 +42,12 @@ func TestSenderCheckStats(t *testing.T) { must(st.AddSenderCheckDown(now, 3*time.Second)) must(st.AddSenderCheckDown(now, 1500*time.Millisecond)) // whole seconds: one must(st.AddSenderCheckDown(now, 400*time.Millisecond)) // nothing - must(st.AddSenderCheckDeferred(now.Add(-time.Hour), 2)) + must(st.AddSenderCheckDeferred(now.Add(-time.Hour), 2, "")) must(st.AddSenderCheckDown(day(10), 30*time.Second)) - must(st.AddSenderCheckDeferred(day(10), 1)) + must(st.AddSenderCheckDeferred(day(10), 1, "")) must(st.AddSenderCheckDown(day(29), time.Minute)) must(st.AddSenderCheckDown(day(30), time.Hour)) // the 31st day back: the year's only - must(st.AddSenderCheckDeferred(day(200), 5)) + must(st.AddSenderCheckDeferred(day(200), 5, "")) must(st.AddSenderCheckDown(day(365), 24*time.Hour)) // a year and a day: nobody's got, err = st.SenderCheckStats(now) @@ -112,3 +112,22 @@ func TestRefusedRowKeepsItsReason(t *testing.T) { t.Errorf("GetSendLog = %+v, %v", one, err) } } + +// The place in the log is saved with the count, and the moment of the last +// deferral does not move back when a log is read again. +func TestSenderCheckDeferredSavesItsPlace(t *testing.T) { + st := openTestStore(t) + later := time.Date(2026, 10, 14, 12, 0, 1, 0, time.UTC) + if err := st.AddSenderCheckDeferred(later, 1, "abc:1.2@100"); err != nil { + t.Fatal(err) + } + if err := st.AddSenderCheckDeferred(later.Add(-time.Hour), 1, "abc:1.2@200"); err != nil { + t.Fatal(err) + } + if p, err := st.SenderCheckCountedPlace(); err != nil || p != "abc:1.2@200" { + t.Fatalf("place = %q, %v", p, err) + } + if at, err := st.SenderCheckDeferredAt(); err != nil || !at.Equal(later) { + t.Fatalf("last deferral = %s, %v; want %s", at, err, later) + } +}