Files
selfpost/internal/store/sendlog.go
T
mixandClaude Opus 5.5 ffefa9d6df delivery log: measure the sweep before deciding its fate (audit H3, 0007)
Migration 0007 (DDL only): delivery_log_days, a row per UTC day, kept a
year, shown nowhere - read off a copy of the database after a month by the
decision rule written in the audit plan.

- reconcile_closed: the sweep adds the rows it actually closed.
- reconcile_contradicted: store.UpdateStatus, in one transaction, counts the
  rows with reason "reconcile" a delivery line now speaks of, writes the
  log's status and clears the reason - so a re-read line counts once.
- tailer_blind_seconds: follow() tracks stretches in which the file is gone
  or reopening fails (logtail.blindness); stored when a stretch ends and
  every minute while it lasts; under 3 s not counted (the rotation's tick
  between rename and create loses nothing).
- queued_over_lifetime: every sweep overwrites the day's level with the rows
  queued past LongestLifetime + MaximalBackoff (a message Postfix still
  works on stays queued until its first retry after the lifetime).
- retentionLoop prunes days older than a year.

Tests: store (sums per day, the level overwritten, under a second nothing,
prune; a contradicted sweep counted once and its reason cleared; count of
queued rows); logtail (the sweep counts what it closed; countStuck records
the level; blindness - a one-tick gap not counted, 75 s stored as 60 + 15
from where the last part ended). Schema tests at user_version 7.

schema-migrations.md: chain head 7 (it still said 5 since 2.4.0), row 7.
CHANGELOG [Unreleased], audit plan line ticked, roadmap 10/61.
Evidence: go vet ./... clean; go test ./internal/store ok, ./internal/logtail
ok but for TestFollowTailsAndRotates (Windows); the server run of the stage
follows before the push.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 14:20:44 +03:00

350 lines
13 KiB
Go

package store
import (
"database/sql"
"errors"
"fmt"
"strings"
"time"
)
// Send-log status values (architecture.md § Persistence). "queued" is written
// by the journal-milter when a message is accepted; the log-tailer advances it
// to one of the final states as Postfix reports delivery per recipient.
const (
StatusQueued = "queued"
StatusSent = "sent"
StatusDeferred = "deferred"
StatusBounced = "bounced"
// StatusRejected marks a message the journal-milter refused with a 4xx under
// a level-2 rate limit (guide § Rate limiting). Such a row never gets a
// queue-id and is excluded from the level-2 message count (it was never
// sent).
StatusRejected = "rejected"
)
// SendLogEntry is a single queued send-log row. The journal-milter creates one
// per (queue-id, recipient) pair at end-of-message (architecture.md §
// Persistence); every field except the status/timestamps comes from the
// accepted message.
type SendLogEntry struct {
QueueID string
Domain string
AppLogin string
From string
To string
Subject string
// Reason and HeaderFrom are set on refused rows only: why the message
// was refused (Reason*), and for ReasonFrom what its From: line said.
Reason string
HeaderFrom string
}
// InsertQueued records an accepted message in the send log with status
// "queued". It is called from the journal-milter hot path, so it returns any
// error for the caller to log rather than deciding policy here; the milter
// must stay fail-open regardless (architecture.md § Persistence).
func (s *Store) InsertQueued(e SendLogEntry) error {
now := time.Now().UTC().Format(time.RFC3339)
_, err := s.db.Exec(
`INSERT INTO send_log
(queue_id, domain, app_login, from_addr, to_addr, subject, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
e.QueueID, e.Domain, e.AppLogin, e.From, e.To, e.Subject, StatusQueued, now, now,
)
if err != nil {
return fmt.Errorf("insert send_log: %w", err)
}
return nil
}
// InsertRejected records a message the journal-milter refused under a level-2
// rate limit (guide § Rate limiting), so the rejection is visible in the
// send-log UI. Only the fields known at MAIL FROM are set (domain, sender, app
// login); there is no queue-id or recipient because the message was rejected
// before it was queued.
func (s *Store) InsertRejected(e SendLogEntry) error {
now := time.Now().UTC().Format(time.RFC3339)
_, err := s.db.Exec(
`INSERT INTO send_log
(queue_id, domain, app_login, from_addr, to_addr, subject, status, reason, header_from, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
e.QueueID, e.Domain, e.AppLogin, e.From, e.To, e.Subject, StatusRejected, e.Reason, e.HeaderFrom, now, now,
)
if err != nil {
return fmt.Errorf("insert rejected send_log: %w", err)
}
return nil
}
// UpdateStatus advances the delivery status of the send-log rows matching a
// (queue-id, recipient) pair, which the log-tailer parses out of mail.log.
// Recipient matching is case-insensitive because Postfix may normalise address
// case between the milter (envelope) and the delivery log. It returns the
// number of rows updated so the caller can tell whether the line matched a
// journal entry.
//
// A row the reconcile sweep closed (reason "reconcile") that a delivery line
// now speaks of was closed on a guess the log could have settled: the log
// wins, the reason is cleared, and the day's reconcile_contradicted counts it
// (delivery_log_days) - once, since the row no longer carries the reason.
func (s *Store) UpdateStatus(queueID, recipient, status string) (int64, error) {
now := time.Now().UTC()
tx, err := s.db.Begin()
if err != nil {
return 0, fmt.Errorf("update send_log status: %w", err)
}
defer tx.Rollback()
var contradicted int64
if err := tx.QueryRow(
`SELECT COUNT(*) FROM send_log WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE AND reason = ?`,
queueID, recipient, ReasonReconcile,
).Scan(&contradicted); err != nil {
return 0, fmt.Errorf("update send_log status: %w", err)
}
res, err := tx.Exec(
`UPDATE send_log SET status = ?, updated_at = ?,
reason = CASE WHEN reason = ? THEN '' ELSE reason END
WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE`,
status, now.Format(time.RFC3339), ReasonReconcile, queueID, recipient,
)
if err != nil {
return 0, fmt.Errorf("update send_log status: %w", err)
}
if contradicted > 0 {
if _, err := tx.Exec(
`INSERT INTO delivery_log_days (day, reconcile_contradicted) VALUES (?, ?)
ON CONFLICT (day) DO UPDATE SET reconcile_contradicted = reconcile_contradicted + excluded.reconcile_contradicted`,
now.Format(dayLayout), contradicted,
); err != nil {
return 0, fmt.Errorf("update send_log status: count the contradicted sweep: %w", err)
}
}
if err := tx.Commit(); err != nil {
return 0, fmt.Errorf("update send_log status: %w", err)
}
n, _ := res.RowsAffected()
return n, nil
}
// CloseUnreported is the reconcile sweep's close of one row: "bounced", marked
// reason "reconcile" so it is known who decided it (architecture.md § Log
// tailer). Only a row still "queued" is closed - a result the log wrote an
// instant earlier stands, and a months-old row of a reused queue id is not
// rewritten (audit 2026-10-12, H3). It returns the number of rows closed.
func (s *Store) CloseUnreported(queueID, recipient string) (int64, error) {
now := time.Now().UTC().Format(time.RFC3339)
res, err := s.db.Exec(
`UPDATE send_log SET status = ?, reason = ?, updated_at = ?
WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE AND status = ?`,
StatusBounced, ReasonReconcile, now, queueID, recipient, StatusQueued,
)
if err != nil {
return 0, fmt.Errorf("close unreported send_log row: %w", err)
}
n, _ := res.RowsAffected()
return n, nil
}
// QueuedDelivery is a send-log row still waiting for a delivery result,
// reduced to what the log-tailer's reconcile sweep needs to look it up in the
// Postfix queue and, failing that, to close it (architecture.md § Log tailer).
type QueuedDelivery struct {
QueueID string
To string
}
// ListQueuedOlderThan returns the rows still marked "queued" that were accepted
// before cutoff — old enough that Postfix should long since have reported a
// result for them. Rows without a queue-id are skipped: the milter refused
// those before Postfix ever saw the message, so the queue has nothing to say
// about them.
//
// created_at is stored as RFC3339 UTC, so a lexical comparison against the same
// format is chronologically correct.
func (s *Store) ListQueuedOlderThan(cutoff time.Time) ([]QueuedDelivery, error) {
rows, err := s.db.Query(
`SELECT queue_id, to_addr FROM send_log
WHERE status = ? AND queue_id <> '' AND created_at < ?
ORDER BY id`,
StatusQueued, cutoff.UTC().Format(time.RFC3339),
)
if err != nil {
return nil, fmt.Errorf("list queued send_log rows: %w", err)
}
defer rows.Close()
var out []QueuedDelivery
for rows.Next() {
var d QueuedDelivery
if err := rows.Scan(&d.QueueID, &d.To); err != nil {
return nil, fmt.Errorf("scan queued send_log row: %w", err)
}
out = append(out, d)
}
return out, rows.Err()
}
// SendLogRow is one row as returned to the monitoring UI (architecture.md §
// Persistence): a SendLogEntry plus the fields that only exist once a row has
// been written (id, current status, timestamps).
type SendLogRow struct {
ID int64
QueueID string
Domain string
AppLogin string
From string
To string
Subject string
Status string
// Reason and HeaderFrom: see SendLogEntry.
Reason string
HeaderFrom string
CreatedAt time.Time
// UpdatedAt is when the status last changed — the log-tailer's report of
// the delivery attempt. It equals CreatedAt while a row is still queued.
UpdatedAt time.Time
}
// ErrSendLogNotFound is returned by GetSendLog for an id that no longer exists.
// Send-log rows are pruned on the retention window, so a bookmarked delivery
// disappearing is expected, not a fault.
var ErrSendLogNotFound = errors.New("send-log entry not found")
// GetSendLog returns a single send-log row by id, for the delivery detail page
// the log links each row to.
func (s *Store) GetSendLog(id int64) (SendLogRow, error) {
var (
row SendLogRow
createdAt, updatedAt string
)
err := s.db.QueryRow(
`SELECT id, queue_id, domain, app_login, from_addr, to_addr, subject, status, reason, header_from, created_at, updated_at
FROM send_log WHERE id = ?`, id,
).Scan(&row.ID, &row.QueueID, &row.Domain, &row.AppLogin, &row.From, &row.To,
&row.Subject, &row.Status, &row.Reason, &row.HeaderFrom, &createdAt, &updatedAt)
if errors.Is(err, sql.ErrNoRows) {
return SendLogRow{}, ErrSendLogNotFound
}
if err != nil {
return SendLogRow{}, fmt.Errorf("get send_log %d: %w", id, err)
}
row.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
row.UpdatedAt, _ = time.Parse(time.RFC3339, updatedAt)
return row, nil
}
// SendLogFilter narrows QuerySendLog/CountSendLog. It carries two kinds of
// narrowing, and they behave in opposite ways on purpose.
//
// Domain and AppLogin are the operator's own filters, chosen in the UI: an
// empty field matches everything.
//
// Domains and AllDomains are the authorization scope, which no query parameter
// may widen. Domains is the exhaustive set of domain names the caller is
// entitled to read, applied as an IN constraint; AllDomains lifts that
// restriction and is the only way to read the whole journal. A zero-valued
// filter therefore matches *no* rows: a caller that forgets to state a scope
// gets an empty log rather than every tenant's mail, which is the failure mode
// this struct exists to make impossible.
type SendLogFilter struct {
Domain string
AppLogin string
Domains []string
AllDomains bool
}
// QuerySendLog returns send-log rows matching filter, newest first, for the
// monitoring screen's server-side pagination (product.md's send-log view).
// The filter's authorization scope is mandatory: see SendLogFilter.
func (s *Store) QuerySendLog(filter SendLogFilter, limit, offset int) ([]SendLogRow, error) {
where, args := sendLogWhere(filter)
args = append(args, limit, offset)
rows, err := s.db.Query(
`SELECT id, queue_id, domain, app_login, from_addr, to_addr, subject, status, reason, header_from, created_at, updated_at
FROM send_log`+where+`
ORDER BY id DESC
LIMIT ? OFFSET ?`,
args...,
)
if err != nil {
return nil, fmt.Errorf("query send_log: %w", err)
}
defer rows.Close()
var out []SendLogRow
for rows.Next() {
var (
row SendLogRow
createdAt, updatedAt string
)
if err := rows.Scan(&row.ID, &row.QueueID, &row.Domain, &row.AppLogin,
&row.From, &row.To, &row.Subject, &row.Status, &row.Reason, &row.HeaderFrom, &createdAt, &updatedAt); err != nil {
return nil, fmt.Errorf("scan send_log row: %w", err)
}
row.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
row.UpdatedAt, _ = time.Parse(time.RFC3339, updatedAt)
out = append(out, row)
}
return out, rows.Err()
}
// CountSendLog returns how many send-log rows match filter, so the monitoring
// screen can render page numbers/next-prev links.
func (s *Store) CountSendLog(filter SendLogFilter) (int64, error) {
where, args := sendLogWhere(filter)
var n int64
if err := s.db.QueryRow(`SELECT COUNT(*) FROM send_log`+where, args...).Scan(&n); err != nil {
return 0, fmt.Errorf("count send_log: %w", err)
}
return n, nil
}
func sendLogWhere(f SendLogFilter) (string, []any) {
var clauses []string
var args []any
if !f.AllDomains {
// No scope is not "no restriction": a domain administrator whose last
// assignment was deleted owns nothing in the journal and must see
// nothing, and the same clause catches a caller that never set a scope.
if len(f.Domains) == 0 {
return " WHERE 1 = 0", nil
}
marks := make([]string, len(f.Domains))
for i, name := range f.Domains {
marks[i] = "?"
args = append(args, name)
}
clauses = append(clauses, "domain IN ("+strings.Join(marks, ", ")+")")
}
if f.Domain != "" {
clauses = append(clauses, "domain = ?")
args = append(args, f.Domain)
}
if f.AppLogin != "" {
clauses = append(clauses, "app_login = ?")
args = append(args, f.AppLogin)
}
if len(clauses) == 0 {
return "", nil
}
return " WHERE " + strings.Join(clauses, " AND "), args
}
// DeleteSendLogBefore removes send-log rows created before cutoff,
// implementing the configurable retention window (architecture.md §
// Persistence, SEND_LOG_RETENTION_DAYS). It returns the number of rows pruned.
// created_at is stored as RFC3339 UTC, so a lexical comparison against the
// same format is chronologically correct.
func (s *Store) DeleteSendLogBefore(cutoff time.Time) (int64, error) {
res, err := s.db.Exec(
`DELETE FROM send_log WHERE created_at < ?`,
cutoff.UTC().Format(time.RFC3339),
)
if err != nil {
return 0, fmt.Errorf("prune send_log: %w", err)
}
n, _ := res.RowsAffected()
return n, nil
}