inbound journal: one row per recipient, and every row counts to the cap (audit M3)

Security audit 2026-10-17, finding M3, stage 1. inboundSession.RcptTo
appended every RCPT (the outbound session de-duplicates), and the end of
the message inserted a row per recipient one statement at a time inside
Postfix's callback; the cut to 10 000 rows every 500 counted only the log
tailer's rows. Postfix accepts a repeated recipient up to
smtpd_recipient_limit (1 000) and sends each to the milter, so one message
was a thousand rows, uncut until the six-hourly prune.

Now: the inbound milter folds a repeated recipient (case-insensitive, as
cleanup's been_here() with BH_FLAG_FOLD does, and as session.RcptTo does);
a message's rows go in one transaction (store.InsertInboundSpamMessage,
which replaces InsertInboundSpam in milter.InboundStore); the count and the
cut moved from the tailer into the store (countInboundWrites, an atomic
counter on Store) so the milter's rows, the tailer's and the tailer's
start-up re-reads all count. store.InboundSpamLogMaxRows is the one cap.

Tests: TestInboundRecipientRepeatedIsOneRow (1 000 RCPT in two cases, one
row); TestInboundJournalIsCutWhicheverWriterWrites (a small cap: a message's
rows and refusals together cross the mark, the table is cut); the tailer's
own counting test went with its counter. go test ./internal/store
./internal/milter ./internal/logtail (Windows-only Follow failures aside).
Also gofmt of a test line from the M2 commit.

Docs: security.md § Mail path, architecture.md; CHANGELOG; plan.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/store/inboundspam.go

@@ -63,9 +63,86 @@ type InboundSpamEntry struct {
 // inbound domain registered under the row's name now (audit 2026-10-12, L12).
 const inboundDomainID = `(SELECT d.id FROM inbound_domains d WHERE d.name = ?)`
 
-// InsertInboundSpam records one journal row. It is called from the inbound
-// journal milter's hot path and from the log tailer, so it returns any error
-// for the caller to log; neither of them may hold up mail over it.
+// InboundSpamLogMaxRows is how many rows the journal keeps whatever their
+// age: refusals are written at the rate strangers knock on port 25, and
+// accepted mail at the rate they send recipients.
+const InboundSpamLogMaxRows = 10000
+
+// inboundCapRows and inboundCapEvery are the cap and how many rows - from
+// either writer - are written before the journal is cut back to it, without
+// waiting for the six-hourly prune. Postfix's one limit on the rate of RCPT
+// TO, smtpd_client_recipient_rate_limit, is per client address and off by
+// default (postconf(5), Postfix 3.10.13) and not set here. Until 2.14.4 only
+// the log tailer counted its rows; the journal milter's - one per accepted
+// recipient - were cut only every six hours (security audit 2026-10-17, M3).
+// Variables so that a test can make them small.
+var (
+	inboundCapRows  = InboundSpamLogMaxRows
+	inboundCapEvery = int64(500)
+)
+
+// countInboundWrites adds n written rows to the count and, once it reaches
+// inboundCapEvery, cuts the journal to its cap. Two writers crossing the mark
+// together may both cut; the second finds nothing to do.
+func (s *Store) countInboundWrites(n int) error {
+	if n <= 0 || s.inboundWrites.Add(int64(n)) < inboundCapEvery {
+		return nil
+	}
+	s.inboundWrites.Store(0)
+	// The zero time removes nothing by age: this is the cap alone.
+	if _, err := s.PruneInboundSpam(time.Time{}, inboundCapRows); err != nil {
+		return fmt.Errorf("cut inbound_spam_log to %d rows: %w", inboundCapRows, err)
+	}
+	return nil
+}
+
+// InboundRecipient is one recipient of an accepted message at an inbound
+// domain: a row of InsertInboundSpamMessage.
+type InboundRecipient struct {
+	Domain string
+	To     string
+}
+
+// InsertInboundSpamMessage records the rows of one accepted message, one per
+// recipient, e with InboundDomain and To taken from each, in one transaction:
+// the journal milter writes inside Postfix's end-of-message callback on the
+// panel's one connection, and a thousand recipients are one commit, not a
+// thousand. It returns any error for the caller to log.
+func (s *Store) InsertInboundSpamMessage(e InboundSpamEntry, rcpts []InboundRecipient) error {
+	if len(rcpts) == 0 {
+		return nil
+	}
+	at := e.At
+	if at.IsZero() {
+		at = time.Now()
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return fmt.Errorf("insert inbound_spam_log: begin: %w", err)
+	}
+	defer tx.Rollback() //nolint:errcheck // a no-op after Commit
+	for _, r := range rcpts {
+		if _, err := tx.Exec(
+			`INSERT INTO inbound_spam_log
+				(inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at, inbound_domain_id,
+				 queue_id, upstream_status)
+			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`, ?, ?)`,
+			r.Domain, e.ClientIP, e.From, r.To, e.Subject, e.Decision, e.Engine, e.Detail,
+			at.UTC().Format(time.RFC3339), r.Domain, e.QueueID, e.Upstream,
+		); err != nil {
+			return fmt.Errorf("insert inbound_spam_log: %w", err)
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("insert inbound_spam_log: commit: %w", err)
+	}
+	return s.countInboundWrites(len(rcpts))
+}
+
+// InsertInboundSpam records one journal row. It is called from the log
+// tailer (and was from the journal milter, which now writes a message's rows
+// at once), so it returns any error for the caller to log; it may not hold up
+// mail over it.
 func (s *Store) InsertInboundSpam(e InboundSpamEntry) error {
 	at := e.At
 	if at.IsZero() {
@@ -82,7 +159,7 @@ func (s *Store) InsertInboundSpam(e InboundSpamEntry) error {
 	if err != nil {
 		return fmt.Errorf("insert inbound_spam_log: %w", err)
 	}
-	return nil
+	return s.countInboundWrites(1)
 }
 
 // InsertInboundSpamOnce records a journal row unless the same row is there
@@ -109,7 +186,10 @@ func (s *Store) InsertInboundSpamOnce(e InboundSpamEntry) (bool, error) {
 		return false, fmt.Errorf("insert inbound_spam_log once: %w", err)
 	}
 	n, _ := res.RowsAffected()
-	return n > 0, nil
+	if n == 0 {
+		return false, nil
+	}
+	return true, s.countInboundWrites(1)
 }
 
 // updateUpstreamQuery puts one delivery attempt's result on the rows of a

internal/store/inboundspam_test.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -495,3 +496,59 @@ func TestInboundUpstreamHeldAndReusedQueueID(t *testing.T) {
 		t.Errorf("ForgetInboundQueueID does not use idx_inbound_spam_log_queue_id:\n%s", plan)
 	}
 }
+
+// The journal is cut to its cap every so many rows, whichever writer wrote
+// them: the journal milter's rows of accepted mail count as the tailer's
+// refusals do (security audit 2026-10-17, M3). One message's rows are one
+// call. Here the cap is 5 rows and the cut every 4.
+func TestInboundJournalIsCutWhicheverWriterWrites(t *testing.T) {
+	st := openTestStore(t)
+	rowsBefore, everyBefore := inboundCapRows, inboundCapEvery
+	inboundCapRows, inboundCapEvery = 5, 4
+	t.Cleanup(func() { inboundCapRows, inboundCapEvery = rowsBefore, everyBefore })
+	if _, err := st.AddInboundDomain("in.example.com"); err != nil {
+		t.Fatal(err)
+	}
+	count := func() int {
+		var n int
+		if err := st.db.QueryRow("SELECT COUNT(*) FROM inbound_spam_log").Scan(&n); err != nil {
+			t.Fatal(err)
+		}
+		return n
+	}
+	accepted := InboundSpamEntry{ClientIP: "203.0.113.7", From: "a@example.net", Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost}
+	var rcpts []InboundRecipient
+	for i := 0; i < 3; i++ {
+		rcpts = append(rcpts, InboundRecipient{Domain: "in.example.com", To: fmt.Sprintf("r%d@in.example.com", i)})
+	}
+	// Three rows of one message, then two refusals: the fourth row crosses
+	// the mark, and the journal is cut to five.
+	if err := st.InsertInboundSpamMessage(accepted, rcpts); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 2; i++ {
+		if err := st.InsertInboundSpam(InboundSpamEntry{InboundDomain: "in.example.com", ClientIP: "203.0.113.9", To: "x@in.example.com",
+			Decision: SpamDecisionReject, Engine: SpamEnginePostfix}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if n := count(); n != 5 {
+		t.Fatalf("after 5 rows, %d (no cut yet past the cap)", n)
+	}
+	// A message of six more crosses the mark again: cut to five.
+	rcpts = nil
+	for i := 0; i < 6; i++ {
+		rcpts = append(rcpts, InboundRecipient{Domain: "in.example.com", To: fmt.Sprintf("m%d@in.example.com", i)})
+	}
+	if err := st.InsertInboundSpamMessage(accepted, rcpts); err != nil {
+		t.Fatal(err)
+	}
+	if n := count(); n != 5 {
+		t.Errorf("after a message of six over the mark, %d rows, want the cap of 5", n)
+	}
+	// The message's rows carry the domain by id, like the tailer's.
+	var missing int
+	if err := st.db.QueryRow("SELECT COUNT(*) FROM inbound_spam_log WHERE inbound_domain_id IS NULL").Scan(&missing); err != nil || missing != 0 {
+		t.Errorf("%d rows without a domain id (%v)", missing, err)
+	}
+}

internal/store/store.go

@@ -13,6 +13,7 @@ import (
 	"io/fs"
 	"os"
 	"sort"
+	"sync/atomic"
 
 	_ "modernc.org/sqlite" // pure-Go SQLite driver (no cgo), keeps the static build
 )
@@ -23,6 +24,10 @@ var migrationsFS embed.FS
 // Store wraps the database connection pool.
 type Store struct {
 	db *sql.DB
+
+	// inboundWrites counts journal rows written since the journal was last
+	// cut to its row cap, by both of its writers (countInboundWrites).
+	inboundWrites atomic.Int64
 }
 
 // Open opens (creating if needed) the SQLite database at path, enables WAL and