inbound-antispam-panel: inbound journal milter on port 25

Step 3/12 of docs/plans/inbound-antispam-panel.md.

The panel serves a second milter while the inbound relay is on
(internal/milter/inbound.go, socket INBOUND_JOURNAL_MILTER_SOCKET,
default /run/selfpost/inbound-journal.sock). build/postfix-config.sh
puts it last on port 25, after the optional antispam milter, with
default_action=accept; postfix-wrapper.sh waits for its socket. At
end-of-message it writes one inbound_spam_log row per recipient at an
inbound domain - other port-25 recipients (DMARC report addresses) get
none - and always accepts. With a filter attached it reads the verdict
from X-Rspamd-Action and X-Spamd-Result (action, score, up to six
weighted symbols, only what the patterns recognise, 255 bytes); without
one those headers are the sender's and are not read. The plan named
X-Rspamd-Score, which rspamd does not have.

The antispam compose fragment mounts rspamd/milter_headers.conf into
the sidecar so that it adds the two headers (checked in rspamd 3.11:
both routines exist and remove a sender's own headers of those names).

Store: InsertInboundSpam, IsInboundDomain.

Checked on the test server, uncommitted tree: go vet ./... ok,
go test ./... ok (22 packages), make e2e ok - the relayed message of
inbound_relay_delivers_and_refuses has its journal row (accept by
selfpost) and the DMARC report sent through the same listener has none.

Not yet seen in a real run: a verdict written by rspamd reaching this
milter. The e2e stand has no filter, and the test server's only inbound
domain has no upstream, so it accepts no mail.

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

CHANGELOG.md

@@ -10,7 +10,22 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version
 - Schema: migration `0002_inbound_spam_log.sql`, the first after the 2.0
   baseline - two empty tables for the inbound antispam journal and the filter
   lists (inbound-antispam-panel). A 2.0.x database is migrated when the panel
-  starts; nothing reads or writes the tables yet.
+  starts.
+- **Inbound mail is recorded.** With the inbound relay on, the panel runs a
+  second milter, the last one on port 25, and writes a row for every recipient
+  of a message that was accepted for an inbound domain: client address,
+  envelope, subject and - with a spam filter attached - the filter's action,
+  score and heaviest symbols. It only watches: it refuses nothing and Postfix
+  accepts mail when it is down. Refused mail is not there yet (it comes from
+  the mail log, next), and there is no page for the journal yet.
+- The antispam compose fragment mounts `antispam/rspamd/milter_headers.conf`
+  into the rspamd sidecar, which makes it write its verdict into
+  `X-Rspamd-Action` and `X-Spamd-Result`. **Copy `deploy/antispam/rspamd/`
+  next to the fragment before starting it** - Docker creates a directory in
+  place of a missing file and the sidecar then fails to read its
+  configuration.
+- New setting `INBOUND_JOURNAL_MILTER_SOCKET` (default
+  `/run/selfpost/inbound-journal.sock`).
 
 ## [2.0.1] - 2026-10-11
 

build/postfix-config.sh

@@ -117,6 +117,9 @@ SUBMISSION_ENABLE="${SUBMISSION_ENABLE:-false}"
 INBOUND_RELAY_ENABLE="${INBOUND_RELAY_ENABLE:-false}"
 INBOUND_ANTISPAM_MILTER="${INBOUND_ANTISPAM_MILTER:-}"
 INBOUND_ANTISPAM_MILTER_ACTION="${INBOUND_ANTISPAM_MILTER_ACTION:-accept}"
+# The panel's inbound journal milter (monitoring, fail-open), on port 25 only
+# and only while the inbound relay is on. Matched by postfix-wrapper.sh.
+INBOUND_JOURNAL_SOCK="${INBOUND_JOURNAL_MILTER_SOCKET:-/run/selfpost/inbound-journal.sock}"
 INBOUND_RATE_LIMIT_MESSAGES_PER_IP="${INBOUND_RATE_LIMIT_MESSAGES_PER_IP:-20}"
 INBOUND_MESSAGE_SIZE_LIMIT="${INBOUND_MESSAGE_SIZE_LIMIT:-26214400}"
 RELAY_DOMAINS_MAP="${POSTFIX_RELAY_DOMAINS:-/data/postfix/relay_domains}"
@@ -357,7 +360,8 @@ if [ "${INBOUND_RELAY_ENABLE}" = "true" ] || [ "${DMARC_REPORTS_ENABLE}" = "true
 	# would file inbound mail into the outbound send log under the *sender's*
 	# domain — polluting Deliveries and letting a forged From: consume that
 	# domain's level-2 rate-limit budget. So smtpd_milters is always overridden
-	# here: either the optional antispam milter, or nothing at all.
+	# here: the optional antispam milter, then - with the inbound relay on - the
+	# panel's inbound journal milter, or nothing at all.
 	INBOUND_MILTERS=""
 	if [ "${INBOUND_RELAY_ENABLE}" = "true" ] && [ -n "${INBOUND_ANTISPAM_MILTER}" ]; then
 		case "${INBOUND_ANTISPAM_MILTER}" in
@@ -376,6 +380,22 @@ if [ "${INBOUND_RELAY_ENABLE}" = "true" ] || [ "${DMARC_REPORTS_ENABLE}" = "true
 		esac
 		INBOUND_MILTERS="{${INBOUND_ANTISPAM_MILTER},default_action=${INBOUND_ANTISPAM_MILTER_ACTION}}"
 	fi
+	if [ "${INBOUND_RELAY_ENABLE}" = "true" ]; then
+		# Last in the chain: Postfix hands a message to its milters one after
+		# another at end-of-message, so this one reads the verdict headers the
+		# filter added and sees only what the filter let through. It watches and
+		# never refuses, and must never hold up someone else's mail: accept when
+		# it is down. A comma, not a space, between the two - see
+		# master_cf_set_smtp_inet_o.
+		case "${INBOUND_JOURNAL_SOCK}" in
+			/[A-Za-z0-9._/-]*) ;;
+			*)
+				echo "FATAL: INBOUND_JOURNAL_MILTER_SOCKET must be an absolute path, got: ${INBOUND_JOURNAL_SOCK}" >&2
+				exit 1
+				;;
+		esac
+		INBOUND_MILTERS="${INBOUND_MILTERS:+${INBOUND_MILTERS},}{unix:${INBOUND_JOURNAL_SOCK},default_action=accept}"
+	fi
 
 	PORT25_RATE="${INBOUND_RATE_LIMIT_MESSAGES_PER_IP}"
 	PORT25_SIZE="${INBOUND_MESSAGE_SIZE_LIMIT}"

build/postfix-wrapper.sh

@@ -2,8 +2,9 @@
 # Postfix start wrapper: solves the cold-start race where Postfix would try to
 # reach the milter sockets before they are listening.
 #
-# It blocks until BOTH milter sockets — OpenDKIM and the panel's journal-milter
+# It blocks until the milter sockets — OpenDKIM, the panel's journal-milter
-# — are present, then execs `postfix start-fg`. If they are not ready within the
+# and, with the inbound relay on, its inbound journal milter — are present,
+# then execs `postfix start-fg`. If they are not ready within the
 # timeout it exits non-zero WITHOUT starting Postfix, so supervisord/Docker see
 # an explicit startup failure instead of a relay running blind.
 #
@@ -16,8 +17,13 @@ JOURNAL_SOCK="${JOURNAL_MILTER_SOCKET:-/run/selfpost/journal.sock}"
 TIMEOUT="${MILTER_WAIT_TIMEOUT:-30}"
 INTERVAL=1
 
+SOCKS="$OPENDKIM_SOCK $JOURNAL_SOCK"
+if [ "${INBOUND_RELAY_ENABLE:-false}" = "true" ]; then
+	SOCKS="$SOCKS ${INBOUND_JOURNAL_MILTER_SOCKET:-/run/selfpost/inbound-journal.sock}"
+fi
+
 elapsed=0
-for sock in "$OPENDKIM_SOCK" "$JOURNAL_SOCK"; do
+for sock in $SOCKS; do
 	while [ ! -S "$sock" ]; do
 		if [ "$elapsed" -ge "$TIMEOUT" ]; then
 			echo "postfix-wrapper: timed out after ${TIMEOUT}s waiting for milter socket $sock" >&2
@@ -29,5 +35,5 @@ for sock in "$OPENDKIM_SOCK" "$JOURNAL_SOCK"; do
 	echo "postfix-wrapper: milter socket ready: $sock"
 done
 
-echo "postfix-wrapper: both milter sockets ready, starting postfix"
+echo "postfix-wrapper: milter sockets ready, starting postfix"
 exec postfix start-fg

cmd/panel/journal.go

@@ -15,21 +15,42 @@ import (
 // serveJournal opens the journal-milter Unix socket and runs the real milter
 // (architecture.md § Mail path), recording accepted messages into the send
 // log. Socket lifecycle (creation, stale cleanup, group permissions) lives
-// here; the protocol handler lives in internal/milter.
+// in listenMilter; the protocol handler lives in internal/milter.
 func serveJournal(ctx context.Context, cfg config, st *store.Store) error {
-	socketPath := cfg.journalSocket
+	ln, err := listenMilter(cfg.journalSocket)
-	if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil {
+	if err != nil {
 		return err
 	}
+	log.Printf("journal-milter listening on %s", cfg.journalSocket)
+	return milter.Serve(ctx, ln, st)
+}
+
+// serveInboundJournal runs the inbound journal milter, the last milter on
+// port 25 (internal/milter/inbound.go). It exists only while the inbound relay
+// is on: build/postfix-config.sh attaches the socket under the same condition.
+func serveInboundJournal(ctx context.Context, cfg config, st *store.Store) error {
+	ln, err := listenMilter(cfg.inboundJournalSocket)
+	if err != nil {
+		return err
+	}
+	log.Printf("inbound-journal-milter listening on %s", cfg.inboundJournalSocket)
+	return milter.ServeInbound(ctx, ln, st, cfg.inboundAntispamMilter != "")
+}
+
+// listenMilter opens a milter's Unix socket where Postfix can reach it.
+func listenMilter(socketPath string) (net.Listener, error) {
+	if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil {
+		return nil, err
+	}
 	// Clear a stale socket left behind by an unclean shutdown, otherwise the
 	// listen below fails with "address already in use".
 	if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) {
-		return err
+		return nil, err
 	}
 
 	ln, err := net.Listen("unix", socketPath)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	// Postfix (user `postfix`) connects to this socket as a milter and needs
 	// group access. The socket inherits group `selfpost` from the setgid parent
@@ -37,9 +58,7 @@ func serveJournal(ctx context.Context, cfg config, st *store.Store) error {
 	// it (connecting to a Unix socket needs write permission on the node).
 	if err := os.Chmod(socketPath, 0o660); err != nil {
 		ln.Close()
-		return err
+		return nil, err
 	}
-
+	return ln, nil
-	log.Printf("journal-milter listening on %s", socketPath)
-	return milter.Serve(ctx, ln, st)
 }

cmd/panel/main.go

@@ -96,7 +96,10 @@ type config struct {
 	// what attaches it; the panel only says so. See antispamAction.
 	inboundAntispamMilter string
 	inboundAntispamAction string
-	dmarcEnabled          bool
+	// The inbound journal milter's socket; the default mirrors
+	// build/postfix-config.sh. Served only while the inbound relay is on.
+	inboundJournalSocket string
+	dmarcEnabled         bool
 }
 
 func loadConfig() config {
@@ -162,6 +165,7 @@ func loadConfig() config {
 		inboundEnabled:        os.Getenv("INBOUND_RELAY_ENABLE") == "true",
 		inboundAntispamMilter: os.Getenv("INBOUND_ANTISPAM_MILTER"),
 		inboundAntispamAction: antispamAction(os.Getenv("INBOUND_ANTISPAM_MILTER_ACTION")),
+		inboundJournalSocket:  envDefault("INBOUND_JOURNAL_MILTER_SOCKET", "/run/selfpost/inbound-journal.sock"),
 		// Optional DMARC aggregate ingest on port 25. Off unless exactly "true".
 		dmarcEnabled: os.Getenv("DMARC_REPORTS_ENABLE") == "true",
 	}
@@ -294,7 +298,7 @@ func run() error {
 	}
 
 	var wg sync.WaitGroup
-	errc := make(chan error, 4)
+	errc := make(chan error, 5)
 
 	roles := []struct {
 		name string
@@ -318,6 +322,13 @@ func run() error {
 		}},
 	}
 
+	if cfg.inboundEnabled {
+		roles = append(roles, struct {
+			name string
+			fn   func(context.Context) error
+		}{"inbound-journal-milter", func(ctx context.Context) error { return serveInboundJournal(ctx, cfg, st) }})
+	}
+
 	for _, r := range roles {
 		wg.Add(1)
 		go func(name string, fn func(context.Context) error) {

deploy/antispam/docker-compose.antispam.yml

@@ -27,3 +27,6 @@ services:
     image: rspamd/rspamd:3.11
     restart: unless-stopped
     # No host ports: only SelfPost talks to the milter on the compose network.
+    volumes:
+      # The verdict headers SelfPost's inbound journal reads.
+      - ./antispam/rspamd/milter_headers.conf:/etc/rspamd/local.d/milter_headers.conf:ro

deploy/antispam/rspamd/milter_headers.conf

@@ -0,0 +1,11 @@
+# Mounted into the sidecar as /etc/rspamd/local.d/milter_headers.conf.
+#
+# rspamd writes its verdict into two headers of every message it lets through:
+# the action it chose and the score with the symbols behind it. SelfPost's
+# inbound journal reads them (Inbound > Log shows them next to the message),
+# and the upstream mailbox server receives them with the mail.
+#
+# rspamd removes headers of these names that a sender wrote himself before it
+# adds its own. Without this file the journal still lists accepted mail, with
+# no verdict next to it.
+use = ["x-spamd-result", "x-rspamd-action"];

docs/architecture.md

@@ -27,7 +27,7 @@ Managed programs ([build/supervisord.conf](../build/supervisord.conf)):
 | Program | User | Priority | Role |
 |---|---|---|---|
 | `opendkim` | root → `opendkim` (`UserID` in opendkim.conf) | 100 | DKIM signing milter |
-| `panel` | panel | 200 | HTTP UI + journal-milter + log-tailer + rate-limit recalc |
+| `panel` | panel | 200 | HTTP UI + journal-milter + log-tailer + rate-limit recalc; with the inbound relay on also the inbound journal milter |
 | `postfix` | root (wrapper) | 300 | MTA — started only after both milter sockets exist |
 | `postfix-reload` | root | — | On-demand `postfix reload` (autostart off) |
 | `cert-reload` | root | 400 | Daily `postfix reload` for renewed TLS certs |
@@ -74,10 +74,21 @@ smtp/inet` removes Debian's stock smtpd). Outbound delivery still uses the
   `smtpd_tls_security_level=encrypt`.
 - **25/smtp inet** — when `INBOUND_RELAY_ENABLE=true` and/or
   `DMARC_REPORTS_ENABLE=true`. No SASL, no OpenDKIM, no journal-milter —
-  `smtpd_milters` is always overridden on this service (to the antispam milter
+  `smtpd_milters` is always overridden on this service, because main.cf's
-  or to nothing), because main.cf's submission chain would otherwise let a
+  submission chain would otherwise let a signing outage defer relayed mail and
-  signing outage defer relayed mail and would file inbound mail into the
+  would file inbound mail into the outbound send log under the sender's
-  outbound send log under the sender's domain.
+  domain. With the inbound relay on the chain is the optional antispam milter,
+  then the panel's **inbound journal milter** (unix socket
+  `INBOUND_JOURNAL_MILTER_SOCKET`, default
+  `/run/selfpost/inbound-journal.sock`, `default_action=accept`); with DMARC
+  ingest alone it is empty. Postfix hands a message to its milters one after
+  another at end-of-message, each seeing what the one before added, so the
+  journal milter sees only mail the filter let through and reads the filter's
+  verdict headers (`X-Rspamd-Action`, `X-Spamd-Result`) off it. It writes one
+  `inbound_spam_log` row per recipient at an inbound domain - a DMARC report
+  address is not one - and never refuses anything
+  ([internal/milter/inbound.go](../internal/milter/inbound.go)). The headers
+  are read only while a filter is attached; refused mail never reaches it.
   **Inbound relay** (`INBOUND_RELAY_ENABLE=true`): accepts only
   `relay_domains` + `relay_recipient_maps` (`reject_unauth_destination`,
   `reject_unlisted_recipient`). Maps under `/data/postfix/` (`relay_domains`,

docs/guide.md

@@ -209,6 +209,7 @@ expected to set; defaults match the code exactly.
 | `DMARC_MESSAGE_SIZE_LIMIT` | Maximum report message size in bytes when DMARC ingest is on. | `5242880` (5 MiB) | `.env` |
 | `INBOUND_ANTISPAM_MILTER` | Optional milter on the inbound relay listener only (requires `INBOUND_RELAY_ENABLE=true`; not 465/587 or DMARC-only port 25). Empty = off. Format `inet:host:port` or `unix:/path`. Example with [deploy/antispam/docker-compose.antispam.yml](../deploy/antispam/docker-compose.antispam.yml): `inet:antispam:11332`. The panel reads it to report on the *Spam filter* box of the inbound pages and to check that the filter answers (Server › Health). | *(empty)* | `.env` |
 | `INBOUND_ANTISPAM_MILTER_ACTION` | What Postfix does if that milter is down: `accept` (fail-open) or `tempfail` (defer). The *Spam filter* box states which. | `accept` | `.env` |
+| `INBOUND_JOURNAL_MILTER_SOCKET` | Unix socket of the panel's inbound journal milter, the last milter on port 25 while the inbound relay is on. It records accepted inbound mail and never refuses any; there is no reason to change it. | `/run/selfpost/inbound-journal.sock` | compose |
 | `INBOUND_RATE_LIMIT_MESSAGES_PER_IP` | Coarse per-client-IP cap on inbound smtpd (`smtpd_client_message_rate_limit`). Uses the same window as `RATE_LIMIT_WINDOW_SECONDS`. | `20` | `.env` |
 | `INBOUND_MESSAGE_SIZE_LIMIT` | Maximum message size in bytes on inbound smtpd (`message_size_limit`). When both inbound relay and DMARC ingest are on, the shared listener uses the **greater** of this and `DMARC_MESSAGE_SIZE_LIMIT`. | `26214400` (25 MiB) | `.env` |
 | `RATE_LIMIT_MESSAGES_PER_IP` | Level-1 backstop: maximum messages one client IP may submit per window (Postfix `smtpd_client_message_rate_limit`). See [Rate limiting — level 1](#rate-limiting--level-1-ip-backstop). | `100` | `.env` |
@@ -841,6 +842,14 @@ only sees SelfPost. Default action is fail-open (`accept`) so a down sidecar
 does not block backup-MX; set `INBOUND_ANTISPAM_MILTER_ACTION=tempfail` to
 defer instead.
 
+The fragment mounts one file into the sidecar,
+[deploy/antispam/rspamd/milter_headers.conf](../deploy/antispam/rspamd/milter_headers.conf):
+it makes rspamd write its verdict - the action and the score with its symbols
+- into two headers of every message it lets through. SelfPost records them
+with the message, and the upstream receives them with the mail. Copy the
+`antispam/rspamd/` directory next to the fragment; without the file accepted
+mail is still recorded, with no verdict.
+
 The panel only reports this setting; it does not change it. Every inbound
 domain page has a **Spam filter** box — on or off, what happens to mail while
 the filter is down, and that the filter belongs to the server, not to the

docs/plans/inbound-antispam-panel.md

@@ -57,8 +57,9 @@ smtpd_milters = { antispam, inbound-journal }
 ```
 
 Antispam runs first; inbound-journal runs at end-of-message on the **accept**
-path and reads rspamd-added headers (`X-Rspamd-Action`, `X-Rspamd-Score`, symbol
+path and reads rspamd-added headers (`X-Rspamd-Action`, and `X-Spamd-Result`
-summary when present).
+for the score and symbols - rspamd has no `X-Rspamd-Score`; the sidecar adds
+both through `deploy/antispam/rspamd/milter_headers.conf`).
 
 **Antispam rejects** do not reach end-of-message on the journal milter. v1 also
 tails `mail.log` for inbound `postfix/smtpd` **`milter-reject`** / **`reject`**
@@ -244,7 +245,7 @@ Target version cut: **`2.1.0`** (MINOR), after `2.0.0`. One commit per step;
 
 - [x] Agree journal fields, retention, and RBAC (this plan § Decisions) — **Composer**
 - [x] Migration `0002_inbound_spam_log.sql` on the 2.0 baseline — **Composer**
-- [ ] Inbound journal-milter + Postfix wiring — **Composer**
+- [x] Inbound journal-milter + Postfix wiring — **Composer**
 - [ ] mail.log tailer: inbound reject rows — **Composer**
 - [ ] List CRUD + validation + atomic rspamd map sync — **Composer**
 - [ ] Panel: Inbound › Log and Inbound › Filter lists from the kit partials, mockups `in-log` / `in-filter-lists` (add client IP and engine columns mockup-first), evidence pairs — **Composer**

docs/roadmap.md

@@ -30,7 +30,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md).
 |---|---|---|---|---|
 | preflight | Installation check page (`/preflight`) | candidate | — | [plans/preflight.md](plans/preflight.md) |
 | contributing | `CONTRIBUTING.md` | candidate | — | — |
-| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | in progress | 2/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) |
+| inbound-antispam-panel | Inbound antispam journal + allow/deny lists | in progress | 3/12 | [plans/inbound-antispam-panel.md](plans/inbound-antispam-panel.md) |
 | inbound-quarantine | Inbound spam quarantine (hold / review / release) | candidate | — | [plans/inbound-quarantine.md](plans/inbound-quarantine.md) |
 | csrf-tokens | Per-session CSRF tokens on state-changing forms | candidate | — | — |
 | template-data-typing | Typed structs for template data instead of `map[string]any` | candidate | — | — |

internal/milter/inbound.go

@@ -0,0 +1,229 @@
+package milter
+
+import (
+	"context"
+	"log"
+	"net"
+	"net/textproto"
+	"regexp"
+	"strings"
+
+	"github.com/emersion/go-milter"
+
+	"github.com/mixeme/selfpost/internal/mailhdr"
+	"github.com/mixeme/selfpost/internal/store"
+)
+
+// InboundStore is what the inbound journal milter needs: the journal itself
+// and the list of inbound domains, since port 25 also takes mail that is not
+// the relay's. *store.Store satisfies it; tests substitute a fake.
+type InboundStore interface {
+	InsertInboundSpam(e store.InboundSpamEntry) error
+	IsInboundDomain(name string) (bool, error)
+}
+
+// inboundSession is the inbound journal milter: the last milter on port 25,
+// after the optional external filter. It writes one journal row per recipient
+// of a message that reached end-of-message - that is, one the filter did not
+// refuse; refusals never get this far and are read from mail.log instead.
+//
+// Like the send-log milter it only watches: every callback continues or
+// accepts, and Postfix attaches the socket with default_action=accept, so
+// neither an error here nor the panel being down holds up inbound mail.
+type inboundSession struct {
+	milter.NoOpMilter
+	rec InboundStore
+	// filter says whether an external filter is attached in front of this
+	// milter. Without one the verdict headers below can only have been written
+	// by the sender, and are not read.
+	filter bool
+
+	clientIP string // once per connection
+
+	from    string
+	rcpts   []string
+	subject string
+	action  string // X-Rspamd-Action
+	result  string // X-Spamd-Result
+	seen    bool   // any verdict header of the filter
+}
+
+func (s *inboundSession) Connect(host, family string, port uint16, addr net.IP, m *milter.Modifier) (milter.Response, error) {
+	if addr != nil {
+		s.clientIP = addr.String()
+	}
+	return milter.RespContinue, nil
+}
+
+// MailFrom starts a message: a connection may carry several.
+func (s *inboundSession) MailFrom(from string, m *milter.Modifier) (milter.Response, error) {
+	s.reset()
+	s.from = cleanAddress(from)
+	return milter.RespContinue, nil
+}
+
+func (s *inboundSession) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) {
+	s.rcpts = append(s.rcpts, cleanAddress(rcpt))
+	return milter.RespContinue, nil
+}
+
+// Header keeps the first Subject and the filter's verdict. Postfix hands the
+// message to its milters one after another at end-of-message, each seeing what
+// the one before added, which is how the filter's headers arrive here. The
+// filter replaces headers of these names, so the last one read is its own.
+func (s *inboundSession) Header(name, value string, m *milter.Modifier) (milter.Response, error) {
+	switch textproto.CanonicalMIMEHeaderKey(name) {
+	case "Subject":
+		if s.subject == "" {
+			s.subject = mailhdr.DecodeSubject(value)
+		}
+	case "X-Rspamd-Action":
+		if s.filter {
+			s.action, s.seen = value, true
+		}
+	case "X-Spamd-Result":
+		if s.filter {
+			s.result, s.seen = value, true
+		}
+	}
+	return milter.RespContinue, nil
+}
+
+// Body fires at end-of-message: the message is about to be queued for its
+// upstream.
+func (s *inboundSession) Body(m *milter.Modifier) (milter.Response, error) {
+	s.record()
+	s.reset()
+	return milter.RespAccept, nil
+}
+
+func (s *inboundSession) Abort(m *milter.Modifier) error {
+	s.reset()
+	return nil
+}
+
+func (s *inboundSession) reset() {
+	s.from, s.rcpts, s.subject = "", nil, ""
+	s.action, s.result, s.seen = "", "", false
+}
+
+// record writes one row per recipient at an inbound domain. Other recipients
+// of port 25 (a DMARC report address at the server's own name) are not the
+// relay's mail and get no row. Failures are logged, never propagated.
+func (s *inboundSession) record() {
+	e := store.InboundSpamEntry{
+		ClientIP: s.clientIP,
+		From:     s.from,
+		Subject:  s.subject,
+		Decision: store.SpamDecisionAccept,
+		Engine:   store.SpamEngineSelfPost,
+	}
+	if s.seen {
+		action, detail := rspamdVerdict(s.action, s.result)
+		e.Engine, e.Detail = store.SpamEngineRspamd, detail
+		if action == "quarantine" {
+			e.Decision = store.SpamDecisionQuarantine
+		}
+	}
+	for _, to := range s.rcpts {
+		domain := domainOf(to)
+		ok, err := s.rec.IsInboundDomain(domain)
+		if err != nil {
+			log.Printf("inbound-journal: look up %s: %v", domain, err)
+			continue
+		}
+		if !ok {
+			continue
+		}
+		e.InboundDomain, e.To = domain, to
+		if err := s.rec.InsertInboundSpam(e); err != nil {
+			log.Printf("inbound-journal: record %s -> %s: %v", s.from, to, err)
+		}
+	}
+}
+
+// What the journal keeps of a verdict: enough to see why, on one table row.
+const (
+	verdictMaxSymbols = 6
+	verdictMaxBytes   = 255
+)
+
+var (
+	// "default: False [7.20 / 15.00]" - the score and the reject threshold.
+	rspamdScore = regexp.MustCompile(`\[(-?[0-9]+(?:\.[0-9]+)?) / (-?[0-9]+(?:\.[0-9]+)?)\]`)
+	// "BAYES_SPAM(5.10)[100.00%]" - a symbol and its weight.
+	rspamdSymbol = regexp.MustCompile(`([A-Z][A-Z0-9_]{1,63})\((-?[0-9]+(?:\.[0-9]+)?)\)`)
+)
+
+// rspamdVerdict reduces the filter's headers to an action and a one-line
+// summary: the action, the score against the threshold, and the first few
+// symbols that carried weight. It reads only what it recognises - the headers
+// are the filter's when it is up, but the sender's own when it is not (it is
+// attached fail-open), so nothing of them is kept verbatim except through
+// these patterns.
+func rspamdVerdict(actionHeader, resultHeader string) (action, detail string) {
+	action = strings.ToLower(strings.TrimSpace(actionHeader))
+	for _, r := range action {
+		if (r < 'a' || r > 'z') && r != ' ' {
+			action = ""
+			break
+		}
+	}
+	if len(action) > 32 {
+		action = ""
+	}
+
+	var parts []string
+	if action != "" {
+		parts = append(parts, action)
+	}
+	if m := rspamdScore.FindStringSubmatch(resultHeader); m != nil {
+		parts = append(parts, "score "+m[1]+" / "+m[2])
+	}
+	detail = strings.Join(parts, ", ")
+
+	var symbols []string
+	for _, m := range rspamdSymbol.FindAllStringSubmatch(resultHeader, -1) {
+		if strings.Trim(m[2], "-0.") == "" { // weight zero: informational
+			continue
+		}
+		symbols = append(symbols, m[1]+"("+m[2]+")")
+		if len(symbols) == verdictMaxSymbols {
+			break
+		}
+	}
+	if len(symbols) > 0 {
+		if detail != "" {
+			detail += "; "
+		}
+		detail += strings.Join(symbols, " ")
+	}
+	if len(detail) > verdictMaxBytes {
+		detail = detail[:verdictMaxBytes]
+	}
+	return action, detail
+}
+
+// ServeInbound runs the inbound journal milter on ln until ctx is cancelled.
+// filter says whether an external filter is attached in front of it
+// (INBOUND_ANTISPAM_MILTER). It returns nil on a clean shutdown.
+func ServeInbound(ctx context.Context, ln net.Listener, rec InboundStore, filter bool) error {
+	srv := &milter.Server{
+		NewMilter: func() milter.Milter { return &inboundSession{rec: rec, filter: filter} },
+		Actions:   0,                // read-only
+		Protocol:  milter.OptNoBody, // headers and end-of-message, not the body
+	}
+
+	go func() {
+		<-ctx.Done()
+		_ = srv.Close()
+	}()
+
+	if err := srv.Serve(ln); err != nil {
+		if ctx.Err() != nil {
+			return nil
+		}
+		return err
+	}
+	return nil
+}

internal/milter/inbound_test.go

@@ -0,0 +1,224 @@
+package milter
+
+import (
+	"errors"
+	"net"
+	"strings"
+	"testing"
+
+	"github.com/emersion/go-milter"
+
+	"github.com/mixeme/selfpost/internal/store"
+)
+
+// fakeInbound is the journal and the inbound-domain list of one test.
+type fakeInbound struct {
+	domains   map[string]bool
+	rows      []store.InboundSpamEntry
+	insertErr error
+	lookupErr error
+}
+
+func (f *fakeInbound) InsertInboundSpam(e store.InboundSpamEntry) error {
+	if f.insertErr != nil {
+		return f.insertErr
+	}
+	f.rows = append(f.rows, e)
+	return nil
+}
+
+func (f *fakeInbound) IsInboundDomain(name string) (bool, error) {
+	if f.lookupErr != nil {
+		return false, f.lookupErr
+	}
+	return f.domains[name], nil
+}
+
+// A header of the kind the documented rspamd sidecar adds.
+const spamdResult = "default: False [7.20 / 15.00];\n\tBAYES_SPAM(5.10)[100.00%];\n\tMIME_GOOD(-0.10)[text/plain];\n\tARC_NA(0.00)[];\n\tRCVD_COUNT_ZERO(0.00)[0];\n\tR_DKIM_NA(2.20)[]"
+
+// message replays one message through s: envelope, headers in order, end.
+func message(t *testing.T, s *inboundSession, from string, rcpts []string, headers [][2]string) milter.Response {
+	t.Helper()
+	if _, err := s.MailFrom(from, mods(nil)); err != nil {
+		t.Fatalf("MailFrom: %v", err)
+	}
+	for _, r := range rcpts {
+		if _, err := s.RcptTo(r, mods(nil)); err != nil {
+			t.Fatalf("RcptTo: %v", err)
+		}
+	}
+	for _, h := range headers {
+		if _, err := s.Header(h[0], h[1], mods(nil)); err != nil {
+			t.Fatalf("Header: %v", err)
+		}
+	}
+	resp, err := s.Body(mods(map[string]string{"i": "ABC123"}))
+	if err != nil {
+		t.Fatalf("Body: %v", err)
+	}
+	return resp
+}
+
+func inboundSessionFor(t *testing.T, rec InboundStore, filter bool) *inboundSession {
+	t.Helper()
+	s := &inboundSession{rec: rec, filter: filter}
+	if _, err := s.Connect("mx.example.net", "tcp4", 0, net.ParseIP("203.0.113.7"), mods(nil)); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	return s
+}
+
+// With no filter attached an accepted message is a plain row, one per
+// recipient, decided by nobody.
+func TestInboundRecordsAcceptedMessage(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, false)
+	resp := message(t, s, "<sender@example.net> SIZE=1200",
+		[]string{"<a@in.example.com>", "b@IN.example.com"},
+		[][2]string{{"Subject", "=?utf-8?Q?Sch=C3=B6n?="}})
+	if resp != milter.RespAccept {
+		t.Errorf("end-of-message response = %v, want accept", resp)
+	}
+	if len(rec.rows) != 2 {
+		t.Fatalf("rows = %+v, want one per recipient", rec.rows)
+	}
+	want := store.InboundSpamEntry{
+		InboundDomain: "in.example.com", ClientIP: "203.0.113.7",
+		From: "sender@example.net", To: "a@in.example.com", Subject: "Schön",
+		Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost,
+	}
+	if rec.rows[0] != want {
+		t.Errorf("row = %+v\nwant  %+v", rec.rows[0], want)
+	}
+	if rec.rows[1].To != "b@IN.example.com" || rec.rows[1].InboundDomain != "in.example.com" {
+		t.Errorf("second row = %+v", rec.rows[1])
+	}
+}
+
+// Port 25 also takes DMARC reports for the server's own name. They are not
+// the relay's mail and the journal says nothing about them.
+func TestInboundSkipsRecipientsOutsideInboundDomains(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, false)
+	message(t, s, "noreply-dmarc@google.com",
+		[]string{"dmarc-example.com@mail.example.org", "a@in.example.com"}, nil)
+	if len(rec.rows) != 1 || rec.rows[0].To != "a@in.example.com" {
+		t.Errorf("rows = %+v, want the inbound recipient alone", rec.rows)
+	}
+}
+
+// With the filter attached its verdict headers become the engine and detail.
+func TestInboundReadsTheFiltersVerdict(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, true)
+	message(t, s, "sender@example.net", []string{"a@in.example.com"}, [][2]string{
+		{"Subject", "Cheap pills"},
+		{"X-Rspamd-Action", "add header"},
+		{"X-Spamd-Result", spamdResult},
+	})
+	if len(rec.rows) != 1 {
+		t.Fatalf("rows = %+v", rec.rows)
+	}
+	got := rec.rows[0]
+	if got.Engine != store.SpamEngineRspamd || got.Decision != store.SpamDecisionAccept {
+		t.Errorf("engine %q, decision %q", got.Engine, got.Decision)
+	}
+	if want := "add header, score 7.20 / 15.00; BAYES_SPAM(5.10) MIME_GOOD(-0.10) R_DKIM_NA(2.20)"; got.Detail != want {
+		t.Errorf("detail = %q\nwant     %q", got.Detail, want)
+	}
+}
+
+func TestInboundQuarantineAction(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, true)
+	message(t, s, "sender@example.net", []string{"a@in.example.com"},
+		[][2]string{{"x-rspamd-action", "quarantine"}})
+	if len(rec.rows) != 1 || rec.rows[0].Decision != store.SpamDecisionQuarantine || rec.rows[0].Detail != "quarantine" {
+		t.Errorf("rows = %+v", rec.rows)
+	}
+}
+
+// Without a filter in front, a verdict header is the sender's own writing.
+func TestInboundIgnoresVerdictHeadersWithoutAFilter(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, false)
+	message(t, s, "sender@example.net", []string{"a@in.example.com"}, [][2]string{
+		{"X-Rspamd-Action", "no action"},
+		{"X-Spamd-Result", spamdResult},
+	})
+	if len(rec.rows) != 1 || rec.rows[0].Engine != store.SpamEngineSelfPost || rec.rows[0].Detail != "" {
+		t.Errorf("rows = %+v", rec.rows)
+	}
+}
+
+// The filter is attached fail-open, so with it down the headers are again the
+// sender's: nothing of them reaches the journal except through the patterns.
+func TestRspamdVerdictKeepsOnlyWhatItRecognises(t *testing.T) {
+	action, detail := rspamdVerdict("<script>alert(1)</script>", "default: False [1 / 15]; <b>HTML</b> lower_case(9.9) \x1b[31m")
+	if action != "" {
+		t.Errorf("action = %q, want it dropped", action)
+	}
+	if detail != "score 1 / 15" {
+		t.Errorf("detail = %q", detail)
+	}
+
+	var many strings.Builder
+	for i := 0; i < 400; i++ {
+		many.WriteString("SYMBOL_WITH_A_LONG_NAME_" + strings.Repeat("X", 30) + "(1.00)[];")
+	}
+	if _, detail := rspamdVerdict("reject", many.String()); len(detail) > verdictMaxBytes || strings.Count(detail, "(") > verdictMaxSymbols {
+		t.Errorf("detail of %d bytes with %d symbols", len(detail), strings.Count(detail, "("))
+	}
+	if action, _ := rspamdVerdict(strings.Repeat("a", 33), ""); action != "" {
+		t.Errorf("an action of 33 letters was kept")
+	}
+}
+
+// A second message on the same connection starts clean.
+func TestInboundResetsBetweenMessages(t *testing.T) {
+	rec := &fakeInbound{domains: map[string]bool{"in.example.com": true}}
+	s := inboundSessionFor(t, rec, true)
+	message(t, s, "one@example.net", []string{"a@in.example.com"}, [][2]string{
+		{"Subject", "First"}, {"X-Rspamd-Action", "add header"},
+	})
+	message(t, s, "two@example.net", []string{"b@in.example.com"}, nil)
+	if len(rec.rows) != 2 {
+		t.Fatalf("rows = %+v", rec.rows)
+	}
+	want := store.InboundSpamEntry{
+		InboundDomain: "in.example.com", ClientIP: "203.0.113.7",
+		From: "two@example.net", To: "b@in.example.com",
+		Decision: store.SpamDecisionAccept, Engine: store.SpamEngineSelfPost,
+	}
+	if rec.rows[1] != want {
+		t.Errorf("second row = %+v\nwant         %+v", rec.rows[1], want)
+	}
+
+	// An aborted transaction leaves nothing behind for the next one either.
+	if _, err := s.MailFrom("three@example.net", mods(nil)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.RcptTo("c@in.example.com", mods(nil)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Abort(mods(nil)); err != nil {
+		t.Fatal(err)
+	}
+	if len(rec.rows) != 2 || len(s.rcpts) != 0 {
+		t.Errorf("after an abort: rows %d, pending recipients %v", len(rec.rows), s.rcpts)
+	}
+}
+
+// The journal failing is never the mail's problem.
+func TestInboundAcceptsWhenTheStoreFails(t *testing.T) {
+	for name, rec := range map[string]*fakeInbound{
+		"insert": {domains: map[string]bool{"in.example.com": true}, insertErr: errors.New("boom")},
+		"lookup": {lookupErr: errors.New("boom")},
+	} {
+		s := inboundSessionFor(t, rec, false)
+		if resp := message(t, s, "sender@example.net", []string{"a@in.example.com"}, nil); resp != milter.RespAccept {
+			t.Errorf("%s failure: response = %v, want accept", name, resp)
+		}
+	}
+}

internal/store/inboundspam.go

@@ -0,0 +1,72 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// Decisions of the inbound antispam journal. Kept in sync with the CHECK
+// constraint of inbound_spam_log.
+const (
+	SpamDecisionAccept     = "accept"
+	SpamDecisionReject     = "reject"
+	SpamDecisionTempfail   = "tempfail"
+	SpamDecisionQuarantine = "quarantine"
+)
+
+// Who made a decision of the inbound antispam journal. Kept in sync with the
+// CHECK constraint of inbound_spam_log.
+const (
+	SpamEngineRspamd   = "rspamd"   // the external filter
+	SpamEnginePostfix  = "postfix"  // Postfix's own policy: unknown recipient, size, rate
+	SpamEngineSelfPost = "selfpost" // nobody: relayed with no filter attached
+)
+
+// InboundSpamEntry is one row of the inbound antispam journal: what happened
+// to a message offered on port 25 for one recipient. Metadata only - the
+// message itself is relayed, not stored. Subject and Detail may be empty.
+type InboundSpamEntry struct {
+	InboundDomain string
+	ClientIP      string
+	From          string
+	To            string
+	Subject       string
+	Decision      string
+	Engine        string
+	Detail        string
+}
+
+// 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.
+func (s *Store) InsertInboundSpam(e InboundSpamEntry) error {
+	_, err := s.db.Exec(
+		`INSERT INTO inbound_spam_log
+			(inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		e.InboundDomain, e.ClientIP, e.From, e.To, e.Subject, e.Decision, e.Engine, e.Detail,
+		time.Now().UTC().Format(time.RFC3339),
+	)
+	if err != nil {
+		return fmt.Errorf("insert inbound_spam_log: %w", err)
+	}
+	return nil
+}
+
+// IsInboundDomain reports whether name is a configured inbound domain. Port 25
+// also takes mail that is not the relay's - DMARC reports for the server's own
+// name - and the journal keeps rows for inbound domains only.
+func (s *Store) IsInboundDomain(name string) (bool, error) {
+	var one int
+	err := s.db.QueryRow("SELECT 1 FROM inbound_domains WHERE name = ?", strings.ToLower(name)).Scan(&one)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("look up inbound domain: %w", err)
+	}
+	return true, nil
+}

internal/store/inboundspam_test.go

@@ -0,0 +1,61 @@
+package store
+
+import (
+	"testing"
+	"time"
+)
+
+func TestInsertInboundSpam(t *testing.T) {
+	st := openTestStore(t)
+	e := InboundSpamEntry{
+		InboundDomain: "in.example.com", ClientIP: "203.0.113.7",
+		From: "sender@example.net", To: "a@in.example.com", Subject: "Hello",
+		Decision: SpamDecisionAccept, Engine: SpamEngineRspamd, Detail: "no action, score 0.10 / 15.00",
+	}
+	if err := st.InsertInboundSpam(e); err != nil {
+		t.Fatalf("InsertInboundSpam: %v", err)
+	}
+	var got InboundSpamEntry
+	var created string
+	err := st.db.QueryRow(
+		"SELECT inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at FROM inbound_spam_log",
+	).Scan(&got.InboundDomain, &got.ClientIP, &got.From, &got.To, &got.Subject, &got.Decision, &got.Engine, &got.Detail, &created)
+	if err != nil {
+		t.Fatalf("read back: %v", err)
+	}
+	if got != e {
+		t.Errorf("row = %+v\nwant  %+v", got, e)
+	}
+	if at, err := time.Parse(time.RFC3339, created); err != nil || time.Since(at) > time.Minute {
+		t.Errorf("created_at = %q (%v)", created, err)
+	}
+
+	// Every decision and engine the code names is one the table takes.
+	for _, d := range []string{SpamDecisionAccept, SpamDecisionReject, SpamDecisionTempfail, SpamDecisionQuarantine} {
+		for _, eng := range []string{SpamEngineRspamd, SpamEnginePostfix, SpamEngineSelfPost} {
+			e.Decision, e.Engine = d, eng
+			if err := st.InsertInboundSpam(e); err != nil {
+				t.Errorf("decision %s by %s: %v", d, eng, err)
+			}
+		}
+	}
+	e.Decision = "bounce"
+	if err := st.InsertInboundSpam(e); err == nil {
+		t.Error("an unknown decision was stored")
+	}
+}
+
+func TestIsInboundDomain(t *testing.T) {
+	st := openTestStore(t)
+	if _, err := st.AddInboundDomain("in.example.com"); err != nil {
+		t.Fatalf("AddInboundDomain: %v", err)
+	}
+	for name, want := range map[string]bool{
+		"in.example.com": true, "IN.Example.COM": true,
+		"example.com": false, "": false, "other.example.com": false,
+	} {
+		if got, err := st.IsInboundDomain(name); err != nil || got != want {
+			t.Errorf("IsInboundDomain(%q) = %v, %v; want %v", name, got, err, want)
+		}
+	}
+}

test/e2e/port25_config_check.go

@@ -47,9 +47,10 @@ var port25Cases = []port25Case{
 			{param: "smtpd_sasl_auth_enable", contains: "no", wantExact: true},
 			{param: "smtpd_relay_restrictions", contains: "reject_unauth_destination", wantExact: true},
 			{param: "smtpd_recipient_restrictions", contains: "reject_unauth_destination,reject_unlisted_recipient", wantExact: true},
-			// No antispam milter configured: the submission chain (strict
+			// No antispam milter configured: the inbound journal milter
-			// OpenDKIM + journal-milter) must not reach inbound mail.
+			// alone. The submission chain (strict OpenDKIM + journal-milter)
-			{param: "smtpd_milters", contains: "", wantExact: true},
+			// must not reach inbound mail.
+			{param: "smtpd_milters", contains: "{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true},
 		},
 	},
 	{
@@ -72,7 +73,7 @@ var port25Cases = []port25Case{
 		},
 		check: []optionCheck{
 			{param: "smtpd_recipient_restrictions", contains: "check_recipient_access,texthash:/data/postfix/dmarc_recipients,reject_unauth_destination,reject_unlisted_recipient", wantExact: true},
-			{param: "smtpd_milters", contains: "", wantExact: true},
+			{param: "smtpd_milters", contains: "{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true},
 		},
 		// Both on: the report addresses sit on a relay domain, and the relay's
 		// unlisted-recipient check runs after the restriction list. Unless their
@@ -94,7 +95,7 @@ var port25Cases = []port25Case{
 		check: []optionCheck{
 			// Exact equality also covers the absence of OpenDKIM/journal-milter
 			// sockets — anything else there would fail the comparison.
-			{param: "smtpd_milters", contains: "{inet:antispam:11332,default_action=accept}", wantExact: true},
+			{param: "smtpd_milters", contains: "{inet:antispam:11332,default_action=accept},{unix:/run/selfpost/inbound-journal.sock,default_action=accept}", wantExact: true},
 		},
 	},
 }

test/e2e/port25_features_test.go

@@ -265,6 +265,30 @@ func testInboundRelay(t *testing.T, sc *scenario) {
 		t.Fatalf("mail for the inbound domain never reached its upstream: %v\n%s", err, mailLogTail(h, "postfix/"))
 	}
 
+	// The inbound journal milter saw it go through. No page shows the journal
+	// yet, so the row is read where it is kept. No filter is attached on the
+	// stand: the row is an accept by nobody.
+	want := fmt.Sprintf("%s|someone@stranger.e2e.test|%s|e2e %s|accept|selfpost", inboundDomain, inboundRecipient, token)
+	var rows string
+	for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(time.Second) {
+		rows, err = h.execIn("selfpost", "python3", "-c",
+			`import sqlite3; print("\n".join("|".join(r) for r in sqlite3.connect("file:/data/selfpost.db?mode=ro", uri=True).execute("select inbound_domain, from_addr, to_addr, subject, decision, engine from inbound_spam_log")))`)
+		if err != nil {
+			t.Fatalf("read the inbound journal: %v\n%s", err, rows)
+		}
+		if strings.Contains(rows, want) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("no journal row for the relayed message, want %q, journal:\n%s\n%s", want, rows, h.logs("selfpost"))
+		}
+	}
+	// A report mailed to the server's own name earlier in the run came through
+	// the same listener and is not the relay's mail.
+	if n := strings.Count(strings.TrimSpace(rows), "\n") + 1; n != 1 {
+		t.Fatalf("the journal holds %d rows, want the relayed message alone:\n%s", n, rows)
+	}
+
 	for _, to := range []string{"nobody@" + inboundDomain, "someone@elsewhere.e2e.test"} {
 		res := deliverPort25("someone@stranger.e2e.test", to, message(to, uniqueToken("refused")))
 		if res.rcptErr == nil {