From 81df1312171fa25ea29cf850be6fd41ff2059cad Mon Sep 17 00:00:00 2001 From: Mix Date: Mon, 12 Oct 2026 10:11:28 +0300 Subject: [PATCH] delivery-log-storage: the tailer keeps the lines of queued messages Every mail.log line headed by a queue id is written to send_log_lines as the tailer reads it, dated by the line's own stamp. The id is taken from the head of the line only - the tail is the far end's text - and has to look like one: "NOQUEUE:" and "warning:" stand in the same place. The tailer does not check for a send-log row first: a message's first lines come before the journal milter writes it. Lines that never get a row are removed by the prune, which deletes lines older than an hour whose queue id no row carries. It runs at start, after each send-log sweep and hourly. While the tailer catches up at start a line is written only if it is not there already, as the inbound journal does: its place is saved every few seconds, so an unclean stop means reading some lines twice. Checked on Windows: go vet clean; go test ./internal/logtail passes except TestFollowTailsAndRotates, which fails there on renaming an open file and always did. New tests: which line heads are queue ids, a line dated by its stamp, no second copy after a rewound offset, the prune cadence and cutoff. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++- docs/plans/delivery-log-storage.md | 2 +- docs/roadmap.md | 2 +- internal/logtail/lines.go | 82 +++++++++++++++++ internal/logtail/lines_test.go | 140 +++++++++++++++++++++++++++++ internal/logtail/logtail.go | 44 ++++++--- internal/logtail/logtail_test.go | 48 ++++++++++ internal/logtail/retention_test.go | 8 ++ 8 files changed, 319 insertions(+), 17 deletions(-) create mode 100644 internal/logtail/lines.go create mode 100644 internal/logtail/lines_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 93954d1..759fbe9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +11,14 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version The table that will hold what Postfix wrote about a message for as long as its send-log row lives, with the store methods to write a line, to write it once, to read the lines of a row and to prune - ([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)). Nothing - writes or reads it yet. + ([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)). +- **The log tailer keeps what Postfix writes about queued messages.** Every + `mail.log` line headed by a queue id is stored with the line's own time, as + the tailer reads it; after an unclean stop a line read twice is stored once. + Lines whose queue id has no send-log row - the row was pruned, or the + message never had one (inbound relay mail, bounces) - are deleted once they + are an hour old, with every send-log sweep and hourly between. No page reads + them yet. ## [2.1.0] - 2026-10-12 diff --git a/docs/plans/delivery-log-storage.md b/docs/plans/delivery-log-storage.md index c454944..b07948d 100644 --- a/docs/plans/delivery-log-storage.md +++ b/docs/plans/delivery-log-storage.md @@ -143,7 +143,7 @@ Target version cut: **`2.2.0`** (MINOR). One commit per step; [development.md](../development.md) § Plan checklists. - [x] Migration `0003_send_log_lines.sql`, store methods (insert, insert once, lines of a row, prune) and their tests — **Opus** -- [ ] Log tailer: store queue-id lines, write once while catching up, prune with the send-log sweep and hourly — **Opus** +- [x] Log tailer: store queue-id lines, write once while catching up, prune with the send-log sweep and hourly — **Opus** - [ ] The two texts, mockup first: the *Outbound log* help topic and the empty note (`design:` commit) — **Opus** - [ ] Message page reads the table, falls back to the current `mail.log`; help text in `help.go` — **Opus** - [ ] [architecture.md](../architecture.md), [guide.md](../guide.md), [schema-migrations.md](../schema-migrations.md) — **Opus** diff --git a/docs/roadmap.md b/docs/roadmap.md index 4ceba2a..695b97c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -36,7 +36,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md). | structured-logging | `log/slog` with levels and fields | candidate | — | — | | review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — | | panel-notifications | E-mail notifications about important events | candidate | — | — | -| delivery-log-storage | Keep a message's `mail.log` lines with its send-log row | in progress | 1/8 | [plans/delivery-log-storage.md](plans/delivery-log-storage.md) | +| delivery-log-storage | Keep a message's `mail.log` lines with its send-log row | in progress | 2/8 | [plans/delivery-log-storage.md](plans/delivery-log-storage.md) | | password-reset | Password reset by e-mail | candidate | — | — | | phone-desktop-mode | A phone browser in "desktop site" mode gets the phone navigation | candidate | — | — | | panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) | diff --git a/internal/logtail/lines.go b/internal/logtail/lines.go new file mode 100644 index 0000000..2a29e0a --- /dev/null +++ b/internal/logtail/lines.go @@ -0,0 +1,82 @@ +package logtail + +import ( + "log" + "regexp" + "time" + + "github.com/mixeme/selfpost/internal/store" +) + +// A message's page shows what Postfix wrote about it. mail.log is rotated +// daily and fourteen files are kept, while a send-log row lives for the +// retention window, so the lines are kept in the database as the tailer reads +// them (architecture.md § Log tailer). +// +// queueLineRe finds the queue id at the head of a line, after the timestamp +// (already taken off by lineTime), the host name and "program[pid]:": +// +// mail postfix/smtp[26]: 41E862C00D9E: to=, relay=…, status=sent (250 OK) +// +// It is matched from the start of the line and nowhere else: the tail of a +// line is the far end's words - a reply, a HELO name - and a pattern that +// could start anywhere would let them file a line under a queue id of their +// choosing. +// +// What stands in that place is not always a queue id: "NOQUEUE:" and +// "warning:" stand there too. A queue id is hexadecimal in capitals, or, with +// enable_long_queue_ids, digits and letters without vowels; neither word is +// either. +var queueLineRe = regexp.MustCompile(`^\S+ [^\s\[]+\[\d+\]: ([0-9A-F]{6,24}|[0-9B-DF-HJ-NP-TV-Zb-df-hj-np-tv-z]{10,24}): `) + +// lineGrace is how old a line must be before the prune asks whether a send-log +// row carries its queue id. The first lines of a message are logged when it +// arrives and its row is written at end-of-message; an hour is a message body +// sent slowly. +const lineGrace = time.Hour + +// linePruneInterval is how often lines without a row are removed between two +// send-log sweeps. With the inbound relay on, most of what Postfix logs under +// a queue id is mail that never gets a send-log row; it is not left to pile up +// for six hours. A var so tests can shorten it. +var linePruneInterval = time.Hour + +// storeLine keeps line if it is headed by a queue id. It does not ask whether +// a send-log row exists for the id: for a message's first lines none does yet, +// and the lines of messages that never get one are removed by pruneLines. +// +// catchingUp is as in inboundJournal.handle: the line may have been read +// before the last stop, and is then written only if it is not there already. +// The line is dated by its own stamp - after the panel was down, that and now +// are hours apart, and a line is matched to its row by time. +func storeLine(st StatusStore, line string, now time.Time, catchingUp bool) { + at, rest := lineTime(line, now) + m := queueLineRe.FindStringSubmatch(rest) + if m == nil { + return + } + l := store.SendLogLine{QueueID: m[1], At: at, Line: line} + var err error + if catchingUp { + _, err = st.InsertSendLogLineOnce(l) + } else { + err = st.InsertSendLogLine(l) + } + if err != nil { + log.Printf("log-tailer: keep line of %s: %v", m[1], err) + } +} + +// pruneLines removes the stored lines no send-log row carries the queue id of: +// those of rows the retention sweep deleted, and those of messages that never +// had a row. +func pruneLines(st StatusStore) { + n, err := st.PruneSendLogLines(time.Now().UTC().Add(-lineGrace)) + if err != nil { + log.Printf("log-tailer: prune stored log lines: %v", err) + return + } + if n > 0 { + log.Printf("log-tailer: pruned %d stored log lines with no send-log row", n) + } +} diff --git a/internal/logtail/lines_test.go b/internal/logtail/lines_test.go new file mode 100644 index 0000000..d9c9310 --- /dev/null +++ b/internal/logtail/lines_test.go @@ -0,0 +1,140 @@ +package logtail + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// Which lines are kept for a message's page, and under which queue id: the id +// at the head of the line and nothing the far end wrote further along it. +func TestQueueLineHead(t *testing.T) { + cases := []struct { + name, rest, want string + }{ + {"delivery", "mail postfix/smtp[26]: 41E862C00D9E: to=, relay=mx.example.net[203.0.113.9]:25, dsn=2.0.0, status=sent (250 OK)", "41E862C00D9E"}, + {"submission smtpd opens the id", "mail postfix/submission/smtpd[10]: 41E862C00D9E: client=unknown[203.0.113.7], sasl_method=PLAIN, sasl_username=app", "41E862C00D9E"}, + {"port-25 smtpd", "mail postfix/inbound/smtpd[41]: 4A1B2C3D: client=mx.example.net[203.0.113.9]", "4A1B2C3D"}, + {"cleanup", "mail postfix/cleanup[52]: 41E862C00D9E: message-id=<1@example.com>", "41E862C00D9E"}, + {"qmgr removes it", "mail postfix/qmgr[10]: 41E862C00D9E: removed", "41E862C00D9E"}, + {"another program", "mail opendkim[12]: 41E862C00D9E: DKIM-Signature field added (s=sp, d=example.com)", "41E862C00D9E"}, + {"long queue id", "mail postfix/qmgr[10]: 3sJ8Xh2Cgqz4Xy1: removed", "3sJ8Xh2Cgqz4Xy1"}, + {"refused before queueing", "mail postfix/inbound/smtpd[41]: NOQUEUE: reject: RCPT from mx.example.net[203.0.113.9]: 550 5.1.1 : Recipient address rejected; from= to= proto=ESMTP helo=", ""}, + {"a warning", "mail postfix/smtpd[10]: warning: hostname mx.example.net does not resolve to address 203.0.113.9", ""}, + {"a connection", "mail postfix/smtpd[10]: connect from unknown[203.0.113.7]", ""}, + {"no program", "preexisting line, ignored on start", ""}, + { + // The reply is the receiving server's text. One that spells the head + // of a line is still the tail of this one: the line is the message's + // it was written about, and no other's. + "a reply that quotes a line head", + "mail postfix/smtp[26]: 9F1A2C00D9E: to=, relay=mx.example.net[203.0.113.9]:25, dsn=5.1.1, status=bounced (host said: 550 mail postfix/smtp[1]: 41E862C00D9E: to=, status=sent)", + "9F1A2C00D9E", + }, + } + for _, c := range cases { + got := "" + if m := queueLineRe.FindStringSubmatch(c.rest); m != nil { + got = m[1] + } + if got != c.want { + t.Errorf("%s: queue id %q, want %q", c.name, got, c.want) + } + } +} + +// A line is dated by its own stamp, not by when the tailer got to it, and is +// kept as the log has it. +func TestStoreLineDatesByTheLine(t *testing.T) { + cs := &captureStore{} + now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC) + line := "Oct 12 09:00:00 mail postfix/qmgr[7]: 41E862C00D9E: removed" + + storeLine(cs, line, now, false) + storeLine(cs, "Oct 12 09:00:01 mail postfix/smtpd[7]: connect from unknown[203.0.113.7]", now, false) + + if len(cs.lines) != 1 { + t.Fatalf("kept %d lines, want 1: %v", len(cs.lines), cs.kept()) + } + got := cs.lines[0] + if got.QueueID != "41E862C00D9E" || got.Line != line { + t.Errorf("kept %q under %q", got.Line, got.QueueID) + } + if want := time.Date(2026, 10, 12, 9, 0, 0, 0, time.Local).UTC(); !got.At.Equal(want) { + t.Errorf("dated %v, want %v", got.At, want) + } +} + +// The tailer keeps queue-id lines as it follows the log, and after an unclean +// stop - its saved place lies before lines it had already read - it does not +// keep them a second time. +func TestRunKeepsQueueLinesOnce(t *testing.T) { + old := pollInterval + pollInterval = 10 * time.Millisecond + t.Cleanup(func() { pollInterval = old }) + + dir := t.TempDir() + path := filepath.Join(dir, "mail.log") + // Longer than a fingerprint, so the file is recognised across the restart. + seed := strings.Repeat("host postfix/smtpd[1]: connect from unknown[203.0.113.7]\n", 20) + if err := os.WriteFile(path, []byte(seed), 0o644); err != nil { + t.Fatalf("seed log: %v", err) + } + + const ( + client = "host postfix/submission/smtpd[1]: 41E862C00D9E: client=unknown[203.0.113.7]" + sent = "host postfix/smtp[1]: 41E862C00D9E: to=, dsn=2.0.0, status=sent (ok)" + later = "host postfix/qmgr[1]: 41E862C00D9E: removed" + ) + + cs := &captureStore{} + stop := startRun(t, path, cs) + appendLine(t, path, client) + appendLine(t, path, "host postfix/smtpd[1]: disconnect from unknown[203.0.113.7]") + appendLine(t, path, sent) + waitFor(t, func() bool { return len(cs.kept()) == 2 }) + stop() + + // The process died before it saved its place: what it has is the place it + // was at before those lines. + cs.mu.Lock() + cs.state.Offset = int64(len(seed)) + cs.mu.Unlock() + appendLine(t, path, later) + + stop = startRun(t, path, cs) + waitFor(t, func() bool { return len(cs.kept()) >= 3 }) + stop() + + want := []string{"41E862C00D9E|" + client, "41E862C00D9E|" + sent, "41E862C00D9E|" + later} + if got := cs.kept(); strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Errorf("kept:\n%s\nwant:\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) + } +} + +// Lines without a send-log row are removed at the start, with every send-log +// sweep and on their own shorter cadence, and only once they are old enough +// for the row to have been written. +func TestRetentionLoopPrunesLines(t *testing.T) { + old := linePruneInterval + linePruneInterval = 20 * time.Millisecond + t.Cleanup(func() { linePruneInterval = old }) + + cs := &captureStore{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go retentionLoop(ctx, cs, func() int { return 90 }) + + waitFor(t, func() bool { return cs.prunes() >= 3 }) + + cs.mu.Lock() + defer cs.mu.Unlock() + for _, cutoff := range cs.pruned { + if age := time.Since(cutoff); age < lineGrace || age > lineGrace+time.Minute { + t.Errorf("prune cutoff %v ago, want %v", age, lineGrace) + } + } +} diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go index 22ca840..183b72c 100644 --- a/internal/logtail/logtail.go +++ b/internal/logtail/logtail.go @@ -4,7 +4,8 @@ // Postfix only decides sent / deferred / bounced later, per recipient, and // reports it in mail.log. This package parses those lines by queue-id + // recipient and advances the matching rows, and prunes rows past the retention -// window. +// window. It also keeps every line Postfix wrote under a queue id, so that a +// message's page can show them after the log has rotated (lines.go). package logtail import ( @@ -25,12 +26,15 @@ import ( // StatusStore is the slice of the store the log-tailer needs: advancing // delivery statuses, finding the rows still waiting for one, pruning the -// retention window, and remembering how far into mail.log it has read. -// *store.Store satisfies it. +// retention window, keeping the lines of queued messages, and remembering how +// far into mail.log it has read. *store.Store satisfies it. type StatusStore interface { UpdateStatus(queueID, recipient, status string) (int64, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) DeleteSendLogBefore(cutoff time.Time) (int64, error) + InsertSendLogLine(l store.SendLogLine) error + InsertSendLogLineOnce(l store.SendLogLine) (bool, error) + PruneSendLogLines(cutoff time.Time) (int64, error) LogtailState(path string) (store.LogtailState, bool, error) SaveLogtailState(path string, st store.LogtailState) error } @@ -107,7 +111,8 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) { // Run follows path and updates send-log statuses until ctx is cancelled, while // a background sweep prunes rows older than retention(). Reading resumes at // the offset the previous run persisted, so a restart parses the delivery lines -// written while the panel was down. With inbound set it also writes the +// written while the panel was down. Every line headed by a queue id is kept +// for the message's page (lines.go). With inbound set it also writes the // inbound journal's rows for refused mail and prunes that journal // (inbound.go). It returns nil on a clean shutdown. func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays, inbound *Inbound) error { @@ -126,15 +131,20 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa go reconcileLoop(ctx, st, caughtUp) return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) { - if journal != nil { - catchingUp := true - select { - case <-caughtUp: - catchingUp = false - default: - } - journal.handle(line, time.Now().UTC(), catchingUp) + // 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. + catchingUp := true + select { + case <-caughtUp: + catchingUp = false + default: } + now := time.Now().UTC() + if journal != nil { + journal.handle(line, now, catchingUp) + } + storeLine(st, line, now, catchingUp) queueID, recipient, status, ok := parseDelivery(line) if !ok { return @@ -214,7 +224,9 @@ func reconcile(st StatusStore, cutoff time.Time) { } } -// retentionLoop prunes expired send-log rows immediately and then periodically. +// retentionLoop prunes expired send-log rows immediately and then periodically, +// and after each sweep the stored log lines those rows leave behind. Lines +// without a row are also removed on a shorter cadence of their own (lines.go). func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) { prune := func() { retentionDays := retention() @@ -233,14 +245,20 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) } prune() + pruneLines(st) t := time.NewTicker(retentionInterval) defer t.Stop() + lines := time.NewTicker(linePruneInterval) + defer lines.Stop() for { select { case <-ctx.Done(): return case <-t.C: prune() + pruneLines(st) + case <-lines.C: + pruneLines(st) } } } diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go index 0d2059f..3287031 100644 --- a/internal/logtail/logtail_test.go +++ b/internal/logtail/logtail_test.go @@ -147,6 +147,54 @@ type captureStore struct { // queued is what ListQueuedOlderThan returns, for the reconcile sweep. queued []store.QueuedDelivery queuedErr error + + // lines are the log lines kept for messages' pages; pruned is the cutoff of + // each PruneSendLogLines call. + lines []store.SendLogLine + pruned []time.Time +} + +func (c *captureStore) InsertSendLogLine(l store.SendLogLine) error { + c.mu.Lock() + defer c.mu.Unlock() + c.lines = append(c.lines, l) + return nil +} + +func (c *captureStore) InsertSendLogLineOnce(l store.SendLogLine) (bool, error) { + c.mu.Lock() + defer c.mu.Unlock() + for _, have := range c.lines { + if have.QueueID == l.QueueID && have.Line == l.Line { + return false, nil + } + } + c.lines = append(c.lines, l) + return true, nil +} + +func (c *captureStore) PruneSendLogLines(cutoff time.Time) (int64, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.pruned = append(c.pruned, cutoff) + return 0, nil +} + +// kept returns the stored lines as "queue-id|line". +func (c *captureStore) kept() []string { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]string, len(c.lines)) + for i, l := range c.lines { + out[i] = l.QueueID + "|" + l.Line + } + return out +} + +func (c *captureStore) prunes() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.pruned) } func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) { diff --git a/internal/logtail/retention_test.go b/internal/logtail/retention_test.go index 78b7296..470db3d 100644 --- a/internal/logtail/retention_test.go +++ b/internal/logtail/retention_test.go @@ -30,6 +30,14 @@ func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, erro return 0, nil } +func (s *retentionProbeStore) InsertSendLogLine(store.SendLogLine) error { return nil } + +func (s *retentionProbeStore) InsertSendLogLineOnce(store.SendLogLine) (bool, error) { + return false, nil +} + +func (s *retentionProbeStore) PruneSendLogLines(time.Time) (int64, error) { return 0, nil } + func (s *retentionProbeStore) LogtailState(string) (store.LogtailState, bool, error) { return store.LogtailState{}, false, nil }