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 <noreply@anthropic.com>
This commit is contained in:
mix
2026-10-12 10:11:28 +03:00
co-authored by Claude Opus 5.5
parent 3d17206063
commit 81df131217
8 changed files with 319 additions and 17 deletions
+8 -2
View File
@@ -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 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 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 once, to read the lines of a row and to prune
([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)). Nothing ([plans/delivery-log-storage.md](docs/plans/delivery-log-storage.md)).
writes or reads it yet. - **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 ## [2.1.0] - 2026-10-12
+1 -1
View File
@@ -143,7 +143,7 @@ Target version cut: **`2.2.0`** (MINOR). One commit per step;
[development.md](../development.md) § Plan checklists. [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** - [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** - [ ] 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** - [ ] 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** - [ ] [architecture.md](../architecture.md), [guide.md](../guide.md), [schema-migrations.md](../schema-migrations.md) — **Opus**
+1 -1
View File
@@ -36,7 +36,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md).
| structured-logging | `log/slog` with levels and fields | candidate | — | — | | structured-logging | `log/slog` with levels and fields | candidate | — | — |
| review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — | | review-2026-08-followups | Remaining findings of the 2026-08-19 code review | candidate | — | — |
| panel-notifications | E-mail notifications about important events | 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 | — | — | | password-reset | Password reset by e-mail | candidate | — | — |
| phone-desktop-mode | A phone browser in "desktop site" mode gets the phone navigation | 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) | | panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) |
+82
View File
@@ -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=<a@example.net>, 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)
}
}
+140
View File
@@ -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=<a@example.net>, 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 <n@in.example.com>: Recipient address rejected; from=<a@example.net> to=<n@in.example.com> proto=ESMTP helo=<mx.example.net>", ""},
{"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=<a@example.net>, relay=mx.example.net[203.0.113.9]:25, dsn=5.1.1, status=bounced (host said: 550 mail postfix/smtp[1]: 41E862C00D9E: to=<b@example.net>, 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=<a@example.net>, 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)
}
}
}
+31 -13
View File
@@ -4,7 +4,8 @@
// Postfix only decides sent / deferred / bounced later, per recipient, and // Postfix only decides sent / deferred / bounced later, per recipient, and
// reports it in mail.log. This package parses those lines by queue-id + // 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 // 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 package logtail
import ( import (
@@ -25,12 +26,15 @@ import (
// StatusStore is the slice of the store the log-tailer needs: advancing // StatusStore is the slice of the store the log-tailer needs: advancing
// delivery statuses, finding the rows still waiting for one, pruning the // delivery statuses, finding the rows still waiting for one, pruning the
// retention window, and remembering how far into mail.log it has read. // retention window, keeping the lines of queued messages, and remembering how
// *store.Store satisfies it. // far into mail.log it has read. *store.Store satisfies it.
type StatusStore interface { type StatusStore interface {
UpdateStatus(queueID, recipient, status string) (int64, error) UpdateStatus(queueID, recipient, status string) (int64, error)
ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error) ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error)
DeleteSendLogBefore(cutoff time.Time) (int64, 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) LogtailState(path string) (store.LogtailState, bool, error)
SaveLogtailState(path string, st store.LogtailState) 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 // Run follows path and updates send-log statuses until ctx is cancelled, while
// a background sweep prunes rows older than retention(). Reading resumes at // a background sweep prunes rows older than retention(). Reading resumes at
// the offset the previous run persisted, so a restart parses the delivery lines // 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 journal's rows for refused mail and prunes that journal
// (inbound.go). It returns nil on a clean shutdown. // (inbound.go). It returns nil on a clean shutdown.
func Run(ctx context.Context, path string, st StatusStore, retention RetentionDays, inbound *Inbound) error { 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) go reconcileLoop(ctx, st, caughtUp)
return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) { return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) {
if journal != nil { // Until the first read reaches end-of-file the lines are those written
catchingUp := true // while the panel was down - and, after an unclean stop, some it had
select { // read already.
case <-caughtUp: catchingUp := true
catchingUp = false select {
default: case <-caughtUp:
} catchingUp = false
journal.handle(line, time.Now().UTC(), catchingUp) default:
} }
now := time.Now().UTC()
if journal != nil {
journal.handle(line, now, catchingUp)
}
storeLine(st, line, now, catchingUp)
queueID, recipient, status, ok := parseDelivery(line) queueID, recipient, status, ok := parseDelivery(line)
if !ok { if !ok {
return 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) { func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays) {
prune := func() { prune := func() {
retentionDays := retention() retentionDays := retention()
@@ -233,14 +245,20 @@ func retentionLoop(ctx context.Context, st StatusStore, retention RetentionDays)
} }
prune() prune()
pruneLines(st)
t := time.NewTicker(retentionInterval) t := time.NewTicker(retentionInterval)
defer t.Stop() defer t.Stop()
lines := time.NewTicker(linePruneInterval)
defer lines.Stop()
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return
case <-t.C: case <-t.C:
prune() prune()
pruneLines(st)
case <-lines.C:
pruneLines(st)
} }
} }
} }
+48
View File
@@ -147,6 +147,54 @@ type captureStore struct {
// queued is what ListQueuedOlderThan returns, for the reconcile sweep. // queued is what ListQueuedOlderThan returns, for the reconcile sweep.
queued []store.QueuedDelivery queued []store.QueuedDelivery
queuedErr error 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) { func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) {
+8
View File
@@ -30,6 +30,14 @@ func (s *retentionProbeStore) DeleteSendLogBefore(cutoff time.Time) (int64, erro
return 0, nil 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) { func (s *retentionProbeStore) LogtailState(string) (store.LogtailState, bool, error) {
return store.LogtailState{}, false, nil return store.LogtailState{}, false, nil
} }