diff --git a/CHANGELOG.md b/CHANGELOG.md index ab600a2..93954d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version ## [Unreleased] +### Added + +- **Schema: `send_log_lines`** (migration `0003_send_log_lines.sql`, DDL only). + 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. + ## [2.1.0] - 2026-10-12 ### Security diff --git a/docs/plans/delivery-log-storage.md b/docs/plans/delivery-log-storage.md index 559cc35..c454944 100644 --- a/docs/plans/delivery-log-storage.md +++ b/docs/plans/delivery-log-storage.md @@ -142,7 +142,7 @@ ordinary case. Target version cut: **`2.2.0`** (MINOR). One commit per step; [development.md](../development.md) § Plan checklists. -- [ ] 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** - [ ] 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** diff --git a/docs/roadmap.md b/docs/roadmap.md index 9e7fce3..4ceba2a 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 | agreed | 0/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 | 1/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/docs/schema-migrations.md b/docs/schema-migrations.md index 7f1c2c4..2aa7b54 100644 --- a/docs/schema-migrations.md +++ b/docs/schema-migrations.md @@ -17,8 +17,8 @@ When you add or review a migration, update this file in the same change. | Field | Value | |---|---| -| Chain head | `user_version = 2` | -| Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql` | +| Chain head | `user_version = 3` | +| Files | `0001_init.sql` (the 2.0 baseline), `0002_inbound_spam_log.sql`, `0003_send_log_lines.sql` | | File mark | `PRAGMA application_id = 0x53503230` ("SP20") | | Database file | `/data/selfpost.db` (bind mount) | | Compatibility | 2.x MINOR releases must boot a `2.0.0` data directory. **Nothing is carried over from 1.x.** | @@ -62,7 +62,8 @@ What that means in practice: again and no file is deleted, renamed or reordered — every 2.x database has already run it. `2.0.0` and `2.0.1` shipped with the baseline alone; the features after them add files, the first being `0002_inbound_spam_log.sql` of -inbound-antispam-panel (`2.1.0`). +inbound-antispam-panel (`2.1.0`), then `0003_send_log_lines.sql` of +delivery-log-storage (`2.2.0`). --- @@ -72,6 +73,7 @@ inbound-antispam-panel (`2.1.0`). |-----|------|---------|------|---------| | 1 | `0001_init.sql` | 2.0.0 | DDL | The whole 2.0 schema, below | | 2 | `0002_inbound_spam_log.sql` | 2.1.0 | DDL | `inbound_spam_log`, `inbound_spam_list` and two indexes; nothing existing changes | +| 3 | `0003_send_log_lines.sql` | 2.2.0 | DDL | `send_log_lines` and one index; nothing existing changes | **Kind:** *DDL* — schema only; *data* — `INSERT`/`UPDATE` that must stay correct for operators upgrading from older 2.x images. @@ -100,6 +102,7 @@ for operators upgrading from older 2.x images. | `dmarc_report_records` | Per-source rows inside a report | | `inbound_spam_log` | Inbound antispam journal: one row per recipient of a message offered on port 25, with the decision (`accept`, `reject`, `tempfail`, `quarantine`) and who made it (`rspamd`, `postfix`, `selfpost`). Metadata only (v2) | | `inbound_spam_list` | Allow / deny entries for the external filter (`address`, `domain`, `ip`), instance-wide; unique per list (v2) | +| `send_log_lines` | The `mail.log` lines Postfix wrote under a queue id, with the line's own time; what a message's page shows once the log has rotated. No reference to `send_log`: a line belongs to a row by queue id and time, and goes when no row carries its id (v3) | `TestBaselineSchema` (`internal/store/schema_test.go`) pins every table, column and index of this list; `TestOpenMigratesA20Database` opens a database that diff --git a/internal/store/migrations/0003_send_log_lines.sql b/internal/store/migrations/0003_send_log_lines.sql new file mode 100644 index 0000000..42f11fa --- /dev/null +++ b/internal/store/migrations/0003_send_log_lines.sql @@ -0,0 +1,22 @@ +-- What Postfix wrote about a message, kept with its send-log rows +-- (delivery-log-storage, 2.2.0). DDL only: a 2.1.0 database gains one empty +-- table and nothing in it changes. + +-- One row per mail.log line headed by a queue id, written by the log tailer as +-- it reads the log. mail.log is rotated daily and fourteen files are kept; a +-- send-log row lives for the retention window, ninety days by default, and +-- its page shows these lines for as long. +-- +-- There is no reference to send_log: the first lines of a message are logged +-- before the journal milter writes its row, and a message with several +-- recipients is several rows under one queue id. A line belongs to a row by +-- queue id and by time (Postfix reuses queue ids), and a line whose queue id +-- has no send-log row is deleted by the tailer's prune. +CREATE TABLE send_log_lines ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + queue_id TEXT NOT NULL, + logged_at TEXT NOT NULL, + line TEXT NOT NULL +); +-- A message's page reads the lines of one queue id within a span of time. +CREATE INDEX idx_send_log_lines_queue_id ON send_log_lines (queue_id, logged_at); diff --git a/internal/store/schema_test.go b/internal/store/schema_test.go index d05b7cb..d0d86a0 100644 --- a/internal/store/schema_test.go +++ b/internal/store/schema_test.go @@ -57,6 +57,7 @@ func TestBaselineSchema(t *testing.T) { "dmarc_report_records": "id report_row_id source_ip count disposition spf_result dkim_result header_from", "inbound_spam_log": "id inbound_domain client_ip from_addr to_addr subject decision engine detail created_at", "inbound_spam_list": "id list_type entry_type value note created_at", + "send_log_lines": "id queue_id logged_at line", } rows, err := st.db.Query("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name") if err != nil { @@ -102,7 +103,7 @@ func TestBaselineSchema(t *testing.T) { wantIdx := []string{ "idx_dmarc_report_records_report", "idx_dmarc_reports_domain", "idx_dmarc_reports_received", "idx_inbound_spam_log_created_at", "idx_inbound_spam_log_domain_created_at", - "idx_send_log_app_login_created_at", "idx_send_log_created_at", "idx_send_log_domain", "idx_send_log_queue_id", + "idx_send_log_app_login_created_at", "idx_send_log_created_at", "idx_send_log_domain", "idx_send_log_lines_queue_id", "idx_send_log_queue_id", "idx_sessions_expires_at", } sort.Strings(wantIdx) @@ -111,8 +112,8 @@ func TestBaselineSchema(t *testing.T) { } var version, appID int64 - if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 2 { - t.Errorf("user_version = %d, %v; want 2", version, err) + if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 3 { + t.Errorf("user_version = %d, %v; want 3", version, err) } if err := st.db.QueryRow("PRAGMA application_id").Scan(&appID); err != nil || appID != applicationID { t.Errorf("application_id = %#x, %v; want %#x", appID, err, applicationID) @@ -188,13 +189,13 @@ func TestOpenMigratesA20Database(t *testing.T) { } defer st.Close() var version int - if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 2 { - t.Errorf("user_version = %d, %v; want 2", version, err) + if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 3 { + t.Errorf("user_version = %d, %v; want 3", version, err) } if ok, err := st.UserExists(); err != nil || !ok { t.Errorf("the user of the 2.0 database: exists = %v, %v", ok, err) } - for _, table := range []string{"inbound_spam_log", "inbound_spam_list"} { + for _, table := range []string{"inbound_spam_log", "inbound_spam_list", "send_log_lines"} { if len(tableColumns(t, st, table)) == 0 { t.Errorf("table %s was not created", table) } diff --git a/internal/store/sendloglines.go b/internal/store/sendloglines.go new file mode 100644 index 0000000..1f7c0f7 --- /dev/null +++ b/internal/store/sendloglines.go @@ -0,0 +1,128 @@ +package store + +import ( + "fmt" + "time" +) + +// SendLogLine is one mail.log line Postfix wrote under a queue id, kept so a +// message's page can show it after mail.log has rotated (architecture.md § Log +// tailer). At is the line's own time, not when the tailer read it. +type SendLogLine struct { + QueueID string + At time.Time + Line string +} + +// SendLogLineMax is how many bytes of a line are kept. Postfix quotes the far +// end's reply in a delivery line, so a line is as long as someone else cared +// to make it; what a page needs of it is the beginning. +const SendLogLineMax = 4096 + +// sendLogLineBefore is how long before a row was written a line may be dated +// and still be the row's. The journal milter writes the row at end-of-message; +// the smtpd line that opens the queue id is written when the first recipient +// is accepted, and the message body can take its time after that. +const sendLogLineBefore = time.Hour + +func cutSendLogLine(line string) string { + if len(line) > SendLogLineMax { + return line[:SendLogLineMax] + } + return line +} + +// InsertSendLogLine stores one line. The tailer calls it for every line headed +// by a queue id, whether or not a send-log row exists for it yet. +func (s *Store) InsertSendLogLine(l SendLogLine) error { + _, err := s.db.Exec( + `INSERT INTO send_log_lines (queue_id, logged_at, line) VALUES (?, ?, ?)`, + l.QueueID, l.At.UTC().Format(time.RFC3339), cutSendLogLine(l.Line), + ) + if err != nil { + return fmt.Errorf("insert send_log_lines: %w", err) + } + return nil +} + +// InsertSendLogLineOnce stores a line unless the same one is already there, +// and reports whether it wrote. It is for the lines the tailer reads again +// after an unclean stop; a line carries its time and its process id, so two +// that are equal are one line read twice. +func (s *Store) InsertSendLogLineOnce(l SendLogLine) (bool, error) { + line := cutSendLogLine(l.Line) + res, err := s.db.Exec( + `INSERT INTO send_log_lines (queue_id, logged_at, line) + SELECT ?, ?, ? + WHERE NOT EXISTS (SELECT 1 FROM send_log_lines WHERE queue_id = ? AND line = ?)`, + l.QueueID, l.At.UTC().Format(time.RFC3339), line, + l.QueueID, line, + ) + if err != nil { + return false, fmt.Errorf("insert send_log_lines once: %w", err) + } + n, _ := res.RowsAffected() + return n > 0, nil +} + +// SendLogLines returns the stored lines of the message a send-log row is +// about, oldest first and at most n of them - the latest n when there are +// more, since what a message did last is what its page is opened for. +// +// Postfix reuses a queue id once the message has left the queue, and a row +// lives far longer than that, so the id alone does not say whose a line is. +// A line is the row's when it is dated from shortly before the row was written +// to lifetime after it; lifetime is how long Postfix may keep working on a +// message, with a margin, and the caller's to state (postfix.RetryPolicy). +func (s *Store) SendLogLines(row SendLogRow, lifetime time.Duration, n int) ([]SendLogLine, error) { + if row.QueueID == "" || n <= 0 { + return nil, nil + } + rows, err := s.db.Query( + `SELECT logged_at, line FROM ( + SELECT id, logged_at, line FROM send_log_lines + WHERE queue_id = ? AND logged_at >= ? AND logged_at <= ? + ORDER BY id DESC LIMIT ?) + ORDER BY id`, + row.QueueID, + row.CreatedAt.Add(-sendLogLineBefore).UTC().Format(time.RFC3339), + row.CreatedAt.Add(lifetime).UTC().Format(time.RFC3339), + n, + ) + if err != nil { + return nil, fmt.Errorf("query send_log_lines: %w", err) + } + defer rows.Close() + + var out []SendLogLine + for rows.Next() { + var at string + l := SendLogLine{QueueID: row.QueueID} + if err := rows.Scan(&at, &l.Line); err != nil { + return nil, fmt.Errorf("scan send_log_lines row: %w", err) + } + l.At, _ = time.Parse(time.RFC3339, at) + out = append(out, l) + } + return out, rows.Err() +} + +// PruneSendLogLines deletes the lines dated before cutoff whose queue id has +// no send-log row, and returns how many went. That one rule is the whole of +// the table's retention: it removes the lines of rows the send-log sweep has +// deleted, and the lines of messages that never had a row - inbound relay +// mail, the bounces Postfix generates. cutoff leaves alone the lines of a +// message still being received, whose row the journal milter has yet to write. +func (s *Store) PruneSendLogLines(cutoff time.Time) (int64, error) { + res, err := s.db.Exec( + `DELETE FROM send_log_lines + WHERE logged_at < ? + AND NOT EXISTS (SELECT 1 FROM send_log WHERE send_log.queue_id = send_log_lines.queue_id)`, + cutoff.UTC().Format(time.RFC3339), + ) + if err != nil { + return 0, fmt.Errorf("prune send_log_lines: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} diff --git a/internal/store/sendloglines_test.go b/internal/store/sendloglines_test.go new file mode 100644 index 0000000..9770e77 --- /dev/null +++ b/internal/store/sendloglines_test.go @@ -0,0 +1,197 @@ +package store + +import ( + "strings" + "testing" + "time" +) + +func sendLogLineTexts(lines []SendLogLine) []string { + out := make([]string, len(lines)) + for i, l := range lines { + out[i] = l.Line + } + return out +} + +func countSendLogLines(t *testing.T, s *Store) int { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM send_log_lines`).Scan(&n); err != nil { + t.Fatalf("count send_log_lines: %v", err) + } + return n +} + +// A row's page gets the lines of its queue id, oldest first, and not the lines +// of another message. +func TestSendLogLinesOfOneMessage(t *testing.T) { + s := openTestStore(t) + at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) + for i, l := range []SendLogLine{ + {QueueID: "41E862C00D9E", At: at, Line: "client=app"}, + {QueueID: "5F00AA", At: at, Line: "someone else's"}, + {QueueID: "41E862C00D9E", At: at.Add(time.Second), Line: "status=sent"}, + {QueueID: "41E862C00D9E", At: at.Add(2 * time.Second), Line: "removed"}, + } { + if err := s.InsertSendLogLine(l); err != nil { + t.Fatalf("insert line %d: %v", i, err) + } + } + row := SendLogRow{QueueID: "41E862C00D9E", CreatedAt: at} + + lines, err := s.SendLogLines(row, 6*24*time.Hour, 200) + if err != nil { + t.Fatalf("SendLogLines: %v", err) + } + if got := strings.Join(sendLogLineTexts(lines), "|"); got != "client=app|status=sent|removed" { + t.Errorf("lines = %q", got) + } + if len(lines) > 0 && !lines[0].At.Equal(at) { + t.Errorf("first line dated %v, want %v", lines[0].At, at) + } + + // More than the page shows: the latest are kept, still oldest first. + lines, err = s.SendLogLines(row, 6*24*time.Hour, 2) + if err != nil { + t.Fatalf("SendLogLines: %v", err) + } + if got := strings.Join(sendLogLineTexts(lines), "|"); got != "status=sent|removed" { + t.Errorf("latest two = %q", got) + } + + // A row the milter refused has no queue id and so no lines - not every line. + lines, err = s.SendLogLines(SendLogRow{CreatedAt: at}, 6*24*time.Hour, 200) + if err != nil || len(lines) != 0 { + t.Errorf("lines of a row without a queue id = %v, %v", lines, err) + } +} + +// Postfix reuses a queue id once the message has left the queue. The lines of +// a later message under the same id are not the earlier row's, nor the other +// way round. +func TestSendLogLinesBoundedByTheRowsLifetime(t *testing.T) { + s := openTestStore(t) + first := time.Date(2026, 9, 1, 9, 0, 0, 0, time.UTC) + second := first.AddDate(0, 0, 30) + lifetime := 6 * 24 * time.Hour + for _, l := range []SendLogLine{ + {QueueID: "ABC123", At: first.Add(-30 * time.Minute), Line: "first: client"}, + {QueueID: "ABC123", At: first.Add(5 * 24 * time.Hour), Line: "first: expired"}, + {QueueID: "ABC123", At: second.Add(-2 * time.Hour), Line: "between the two"}, + {QueueID: "ABC123", At: second, Line: "second: sent"}, + } { + if err := s.InsertSendLogLine(l); err != nil { + t.Fatalf("insert: %v", err) + } + } + + lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: first}, lifetime, 200) + if err != nil { + t.Fatalf("SendLogLines: %v", err) + } + if got := strings.Join(sendLogLineTexts(lines), "|"); got != "first: client|first: expired" { + t.Errorf("the first message's lines = %q", got) + } + lines, err = s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: second}, lifetime, 200) + if err != nil { + t.Fatalf("SendLogLines: %v", err) + } + if got := strings.Join(sendLogLineTexts(lines), "|"); got != "second: sent" { + t.Errorf("the second message's lines = %q", got) + } +} + +// After an unclean stop the tailer reads some lines twice; the second reading +// writes nothing. The same text under another queue id is another line. +func TestInsertSendLogLineOnce(t *testing.T) { + s := openTestStore(t) + l := SendLogLine{QueueID: "ABC123", At: time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC), Line: "Oct 12 09:00:00 mail postfix/qmgr[7]: ABC123: removed"} + + for i, want := range []bool{true, false} { + wrote, err := s.InsertSendLogLineOnce(l) + if err != nil || wrote != want { + t.Errorf("reading %d: wrote = %v, %v; want %v", i+1, wrote, err, want) + } + } + other := l + other.QueueID = "DEF456" + if wrote, err := s.InsertSendLogLineOnce(other); err != nil || !wrote { + t.Errorf("another queue id: wrote = %v, %v", wrote, err) + } + if n := countSendLogLines(t, s); n != 2 { + t.Errorf("%d lines stored, want 2", n) + } +} + +// The far end writes the reply a delivery line quotes; what is kept of a line +// has a ceiling, and a line read twice is still recognised past it. +func TestSendLogLineIsCut(t *testing.T) { + s := openTestStore(t) + at := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) + l := SendLogLine{QueueID: "ABC123", At: at, Line: strings.Repeat("x", SendLogLineMax+500)} + if err := s.InsertSendLogLine(l); err != nil { + t.Fatalf("insert: %v", err) + } + if wrote, err := s.InsertSendLogLineOnce(l); err != nil || wrote { + t.Errorf("the same long line again: wrote = %v, %v", wrote, err) + } + lines, err := s.SendLogLines(SendLogRow{QueueID: "ABC123", CreatedAt: at}, time.Hour, 10) + if err != nil || len(lines) != 1 { + t.Fatalf("lines = %d, %v", len(lines), err) + } + if len(lines[0].Line) != SendLogLineMax { + t.Errorf("kept %d bytes, want %d", len(lines[0].Line), SendLogLineMax) + } +} + +// The prune is the table's retention: lines go when no send-log row carries +// their queue id - the row was pruned, or there never was one - and a message +// still being received is left alone. +func TestPruneSendLogLines(t *testing.T) { + s := openTestStore(t) + now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + old := now.Add(-3 * time.Hour) + + if err := s.InsertQueued(SendLogEntry{QueueID: "KEPT01", Domain: "example.com", To: "a@example.net"}); err != nil { + t.Fatalf("insert row: %v", err) + } + for _, l := range []SendLogLine{ + {QueueID: "KEPT01", At: old, Line: "has a row"}, + {QueueID: "INBOUND1", At: old, Line: "never had a row"}, + {QueueID: "ARRIVING", At: now.Add(-time.Minute), Line: "its row is not written yet"}, + } { + if err := s.InsertSendLogLine(l); err != nil { + t.Fatalf("insert: %v", err) + } + } + + n, err := s.PruneSendLogLines(now.Add(-time.Hour)) + if err != nil || n != 1 { + t.Fatalf("prune removed %d, %v; want 1", n, err) + } + var left []string + rows, err := s.db.Query(`SELECT queue_id FROM send_log_lines ORDER BY id`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + t.Fatal(err) + } + left = append(left, id) + } + rows.Close() + if got := strings.Join(left, " "); got != "KEPT01 ARRIVING" { + t.Errorf("left after the prune: %q", got) + } + + // The send-log sweep deletes the row; the next prune takes its lines. + if _, err := s.DeleteSendLogBefore(time.Now().UTC().Add(time.Hour)); err != nil { + t.Fatalf("delete rows: %v", err) + } + if n, err := s.PruneSendLogLines(now.Add(-time.Hour)); err != nil || n != 1 { + t.Errorf("prune after the row went removed %d, %v; want 1", n, err) + } +}