Phase 6: journal-milter + send-log status tailer + retention

Implement the structured send log (spec 7.3), the project's highest-risk component since a milter bug can break the relay itself. - internal/milter: go-milter v0.4.1 journal-milter. Per-connection session collects SASL login, From, recipients and Subject across callbacks and writes one send_log "queued" row per (queue-id, recipient) at EOM (spec 7.3.3). Monitoring only: callbacks return Continue/Accept, recorder errors are logged never propagated, so it can never block mail. - internal/logtail: polling mail.log tailer with rotation handling (inode change / truncation), parses sent/deferred/bounced/expired by queue-id + recipient and advances rows; background retention sweep prunes rows past SEND_LOG_RETENTION_DAYS (default 90) at startup and every 6h. - internal/store/sendlog.go: InsertQueued, UpdateStatus (case-insensitive recipient match), DeleteSendLogBefore + status constants. - cmd/panel: open the store once and share it across http/milter/tailer; replace the journal/logtail stubs with the real roles. - build/postfix-config.sh: bounded milter timeouts (15/15/30s) so a hung milter also fails open in seconds, not the 300s default. Fix found in-container: SASL login (app_login) was empty because go-milter keys macros exactly as Postfix sends them, and multi-character macro names arrive brace-wrapped ({auth_authen}); the SASL-less Phase 0 spike could not observe this. Added a brace-tolerant macro lookup. Verified on selfpost.mixfed.ru: gofmt/vet/unit tests green; container e2e records rows with correct fields and advances status via the tailer; fail-open confirmed for both an unreachable and a hung milter; retention prunes at start. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 22:58:34 +03:00
parent 2dbd8d009e
commit dc08ccbf7c
13 changed files with 1033 additions and 51 deletions
@@ -0,0 +1,209 @@
+// Package logtail follows Postfix's mail.log and reconciles the send-log
+// delivery statuses the journal-milter could not know at receive time (spec
+// 7.3). A milter row starts life as "queued"; 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.
+package logtail
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"log"
+	"os"
+	"regexp"
+	"strings"
+	"time"
+
+	"codeberg.org/mix/selfpost/internal/store"
+)
+
+// StatusStore is the slice of the store the log-tailer needs. *store.Store
+// satisfies it.
+type StatusStore interface {
+	UpdateStatus(queueID, recipient, status string) (int64, error)
+	DeleteSendLogBefore(cutoff time.Time) (int64, error)
+}
+
+// pollInterval is how often the tail loop checks for new bytes / rotation. It
+// is a var so tests can shorten it.
+var pollInterval = time.Second
+
+const (
+	// retentionInterval is how often the retention sweep runs (also once at
+	// startup). The window itself is configurable; the cadence need not be.
+	retentionInterval = 6 * time.Hour
+	// defaultRetentionDays applies when the configured value is unset/invalid
+	// (spec 7.3).
+	defaultRetentionDays = 90
+)
+
+// deliveryRe matches a Postfix delivery line and captures queue-id, recipient
+// and status, e.g.
+//
+//	postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=…, dsn=2.0.0, status=sent (250 OK)
+//
+// The "<queue-id>: to=<addr>, …, status=<word>" shape is specific to the
+// delivery agents; qmgr/smtpd/cleanup lines do not match.
+var deliveryRe = regexp.MustCompile(`\b([0-9A-Za-z]+): to=<([^>]*)>,.*\bstatus=(\w+)`)
+
+// parseDelivery extracts (queue-id, recipient, status) from a mail.log line.
+// ok is false for lines that are not recognised delivery results.
+func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
+	m := deliveryRe.FindStringSubmatch(line)
+	if m == nil {
+		return "", "", "", false
+	}
+	switch m[3] {
+	case "sent":
+		status = store.StatusSent
+	case "deferred":
+		status = store.StatusDeferred
+	case "bounced":
+		status = store.StatusBounced
+	case "expired":
+		// Postfix gave up after the queue lifetime; a final failure for us.
+		status = store.StatusBounced
+	default:
+		return "", "", "", false
+	}
+	return m[1], m[2], status, true
+}
+
+// Run follows path and updates send-log statuses until ctx is cancelled, while
+// a background sweep prunes rows older than retentionDays. It returns nil on a
+// clean shutdown.
+func Run(ctx context.Context, path string, st StatusStore, retentionDays int) error {
+	go retentionLoop(ctx, st, retentionDays)
+
+	return follow(ctx, path, func(line string) {
+		queueID, recipient, status, ok := parseDelivery(line)
+		if !ok {
+			return
+		}
+		if _, err := st.UpdateStatus(queueID, recipient, status); err != nil {
+			log.Printf("log-tailer: update %s/%s -> %s: %v", queueID, recipient, status, err)
+		}
+	})
+}
+
+// retentionLoop prunes expired send-log rows immediately and then periodically.
+func retentionLoop(ctx context.Context, st StatusStore, retentionDays int) {
+	if retentionDays <= 0 {
+		retentionDays = defaultRetentionDays
+	}
+	prune := func() {
+		cutoff := time.Now().UTC().AddDate(0, 0, -retentionDays)
+		n, err := st.DeleteSendLogBefore(cutoff)
+		if err != nil {
+			log.Printf("log-tailer: retention prune: %v", err)
+			return
+		}
+		if n > 0 {
+			log.Printf("log-tailer: pruned %d send-log rows older than %d days", n, retentionDays)
+		}
+	}
+
+	prune()
+	t := time.NewTicker(retentionInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			prune()
+		}
+	}
+}
+
+// follow tails path line by line, calling handle for each complete line, until
+// ctx is cancelled. It starts at end-of-file (so a restart does not reprocess
+// history) and reopens the file when it is rotated (inode change from
+// logrotate's create, or truncation from copytruncate) so nothing is missed.
+func follow(ctx context.Context, path string, handle func(string)) error {
+	var (
+		f       *os.File
+		r       *bufio.Reader
+		info    os.FileInfo
+		pending string
+	)
+	openAt := func(offset int64, whence int) error {
+		nf, err := os.Open(path)
+		if err != nil {
+			return err
+		}
+		if _, err := nf.Seek(offset, whence); err != nil {
+			nf.Close()
+			return err
+		}
+		ni, err := nf.Stat()
+		if err != nil {
+			nf.Close()
+			return err
+		}
+		if f != nil {
+			f.Close()
+		}
+		f, r, info, pending = nf, bufio.NewReader(nf), ni, ""
+		return nil
+	}
+
+	// The container may start before Postfix has created mail.log; wait for it.
+	for {
+		if err := openAt(0, io.SeekEnd); err == nil {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(pollInterval):
+		}
+	}
+	defer func() {
+		if f != nil {
+			f.Close()
+		}
+	}()
+
+	drain := func() {
+		for {
+			line, err := r.ReadString('\n')
+			if err == io.EOF {
+				pending += line // hold the partial line until it completes
+				return
+			}
+			if err != nil {
+				log.Printf("log-tailer: read %s: %v", path, err)
+				return
+			}
+			full := pending + line
+			pending = ""
+			handle(strings.TrimRight(full, "\r\n"))
+		}
+	}
+
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			drain()
+			ni, err := os.Stat(path)
+			if err != nil {
+				continue // file briefly gone mid-rotation; try again next tick
+			}
+			pos, _ := f.Seek(0, io.SeekCurrent)
+			if !os.SameFile(info, ni) || ni.Size() < pos {
+				// Rotated away or truncated: reopen from the start of the new
+				// file. Any tail of the old file was already drained above.
+				if err := openAt(0, io.SeekStart); err != nil {
+					log.Printf("log-tailer: reopen %s: %v", path, err)
+				}
+			}
+		}
+	}
+}
@@ -0,0 +1,176 @@
+package logtail
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"codeberg.org/mix/selfpost/internal/store"
+)
+
+func TestParseDelivery(t *testing.T) {
+	cases := []struct {
+		name                       string
+		line                       string
+		wantOK                     bool
+		queueID, recipient, status string
+	}{
+		{
+			name:      "sent",
+			line:      "2026-07-11T11:55:34 host postfix/smtp[26]: 41E862C00D9E: to=<a@example.net>, relay=mx.example.net[203.0.113.9]:25, delay=0.5, dsn=2.0.0, status=sent (250 OK)",
+			wantOK:    true,
+			queueID:   "41E862C00D9E",
+			recipient: "a@example.net",
+			status:    store.StatusSent,
+		},
+		{
+			name:      "deferred",
+			line:      "host postfix/smtp[26]: 5900C2C00D9E: to=<y@example.net>, relay=none, delay=30, dsn=4.4.1, status=deferred (connect timed out)",
+			wantOK:    true,
+			queueID:   "5900C2C00D9E",
+			recipient: "y@example.net",
+			status:    store.StatusDeferred,
+		},
+		{
+			name:      "bounced",
+			line:      "host postfix/smtp[26]: ABC: to=<no@example.net>, relay=…, dsn=5.1.1, status=bounced (user unknown)",
+			wantOK:    true,
+			queueID:   "ABC",
+			recipient: "no@example.net",
+			status:    store.StatusBounced,
+		},
+		{
+			name:      "expired maps to bounced",
+			line:      "host postfix/smtp[26]: DEF: to=<slow@example.net>, relay=none, status=expired (delivery temporarily suspended)",
+			wantOK:    true,
+			queueID:   "DEF",
+			recipient: "slow@example.net",
+			status:    store.StatusBounced,
+		},
+		{
+			name:   "qmgr from-line ignored",
+			line:   "host postfix/qmgr[10]: 41E862C00D9E: from=<noreply@example.com>, size=500, nrcpt=1 (queue active)",
+			wantOK: false,
+		},
+		{
+			name:   "smtpd client-line ignored",
+			line:   "host postfix/smtpd[10]: 41E862C00D9E: client=unknown[203.0.113.7]",
+			wantOK: false,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			q, r, s, ok := parseDelivery(c.line)
+			if ok != c.wantOK {
+				t.Fatalf("ok = %v, want %v", ok, c.wantOK)
+			}
+			if !ok {
+				return
+			}
+			if q != c.queueID || r != c.recipient || s != c.status {
+				t.Fatalf("got (%q,%q,%q), want (%q,%q,%q)", q, r, s, c.queueID, c.recipient, c.status)
+			}
+		})
+	}
+}
+
+// captureStore records UpdateStatus calls for the follow integration test.
+type captureStore struct {
+	mu    sync.Mutex
+	calls []string
+}
+
+func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.calls = append(c.calls, queueID+"|"+recipient+"|"+status)
+	return 1, nil
+}
+
+func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil }
+
+func (c *captureStore) snapshot() []string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]string(nil), c.calls...)
+}
+
+// TestFollowTailsAndRotates writes delivery lines to a log file, then rotates
+// it (rename + fresh create, as logrotate does) and writes more, asserting the
+// tailer picks up lines from both the original and rotated file.
+func TestFollowTailsAndRotates(t *testing.T) {
+	old := pollInterval
+	pollInterval = 10 * time.Millisecond
+	t.Cleanup(func() { pollInterval = old })
+
+	dir := t.TempDir()
+	path := filepath.Join(dir, "mail.log")
+	if err := os.WriteFile(path, []byte("preexisting line, ignored on start\n"), 0o644); err != nil {
+		t.Fatalf("seed log: %v", err)
+	}
+
+	cs := &captureStore{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- Run(ctx, path, cs, 90) }()
+
+	// Give follow() time to open at EOF (it seeks to end immediately on start,
+	// so the seed line above is ignored), then append a delivery line.
+	time.Sleep(50 * time.Millisecond)
+	appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, dsn=2.0.0, status=sent (ok)")
+	waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
+
+	// Rotate: move the current file aside and create a fresh one (logrotate
+	// "create"), then append to the new file.
+	if err := os.Rename(path, path+".1"); err != nil {
+		t.Fatalf("rotate: %v", err)
+	}
+	if err := os.WriteFile(path, nil, 0o644); err != nil {
+		t.Fatalf("recreate: %v", err)
+	}
+	appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, dsn=5.1.1, status=bounced (nope)")
+	waitFor(t, func() bool { return contains(cs.snapshot(), "Q2|b@example.net|bounced") })
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+}
+
+func appendLine(t *testing.T, path, line string) {
+	t.Helper()
+	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
+	if err != nil {
+		t.Fatalf("open for append: %v", err)
+	}
+	defer f.Close()
+	if _, err := f.WriteString(line + "\n"); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+}
+
+func contains(ss []string, want string) bool {
+	for _, s := range ss {
+		if s == want {
+			return true
+		}
+	}
+	return false
+}
+
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("condition not met within timeout")
+}