inbound-upstream-result: the upstream's answer in the Inbound log

The owner, 2026-10-16: the Inbound log lacks the upstream's answer - "Не
понятна судьба письма: принято, отклонено, хост не доступен"; the result
in the row only, a page per message later if needed. Mockups 5348330 and
46b90d2, plan 295d72e.

- Migration 0012: inbound_spam_log gains queue_id, upstream_status ('',
  queued, held, sent, deferred, bounced; CHECK), upstream_relay,
  upstream_reply, upstream_at, and a partial index on queue_id. Schema 12.
- The inbound journal milter writes an accepted row with the queue id of
  end-of-message and "queued" ("held" when the filter had it held), the
  recipient as Postfix logs it (recipientAsLogged), after detaching older
  rows of a queue id Postfix uses again; no queue id, no result.
- The tailer: a delivery line no send-log row took goes to the journal -
  status, relay= ("" for none), the reply after the first status=, cut to
  255 printable bytes, the line's time; queued/held/deferred rows only,
  sent and bounced final. qmgr's expiry makes them bounced, keeping the
  last reply. cleanup's "removed (" detaches rows of a message never
  queued. Addresses in delivery lines are read by Postfix's quoting
  (loggedAddress): a quoted RCPT TO cannot end one early.
- The page: an Upstream column - a plain tag for queued and held, else a
  tag opening the host, the reply, when and what follows; "—" for refused
  mail and older rows. Help (word for word with the mockup), guide,
  architecture, security, schema-migrations, CHANGELOG.

Reviewed before the commit (technical and security, 2026-10-16); the plan
lists each finding and what it became.

Tests: store (results, finality, expiry, held, a reused id, the queue-id
index in every plan), milter (queue id, held, quoted recipient, no {i}),
logtail (relay and reply, a forged quoted recipient, Run hands the journal
what the send log did not take, cleanup's removal), view (the column, held,
null sender), e2e (both relayed messages sent with the sink's 250; an
upstream refusing connections deferred with the reason and no host, then
expired to bounced; the refused row has none). go vet + go test (26
packages) and make e2e on the test server for this tree.

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

internal/store/inboundspam.go

@@ -25,6 +25,18 @@ const (
 	SpamEngineSelfPost = "selfpost" // nobody: relayed with no filter attached
 )
 
+// What the upstream made of a message accepted on port 25, on its journal
+// rows (migration 0012). Kept in sync with the CHECK constraint of
+// inbound_spam_log; "" is a row with no result (refused mail, or written
+// before 2.10.0). sent and bounced are final.
+const (
+	UpstreamQueued   = "queued"   // accepted, not tried yet
+	UpstreamHeld     = "held"     // accepted into the hold queue by the filter's verdict, not tried until released
+	UpstreamSent     = "sent"     // the upstream took it
+	UpstreamDeferred = "deferred" // put off or not reached; Postfix tries again
+	UpstreamBounced  = "bounced"  // refused for good, or expired in the queue
+)
+
 // 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.
@@ -37,6 +49,11 @@ type InboundSpamEntry struct {
 	Decision      string
 	Engine        string
 	Detail        string
+	// QueueID and Upstream are set on mail that was accepted: the queue id
+	// the delivery lines will name, and UpstreamQueued (UpstreamHeld for a
+	// message the filter had held).
+	QueueID  string
+	Upstream string
 	// At is when it happened; zero means now. The log tailer dates a refusal
 	// by its mail.log line, which it may read long after.
 	At time.Time
@@ -56,10 +73,11 @@ 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, inbound_domain_id)
-		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`)`,
+			(inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at, inbound_domain_id,
+			 queue_id, upstream_status)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, `+inboundDomainID+`, ?, ?)`,
 		e.InboundDomain, e.ClientIP, e.From, e.To, e.Subject, e.Decision, e.Engine, e.Detail,
-		at.UTC().Format(time.RFC3339), e.InboundDomain,
+		at.UTC().Format(time.RFC3339), e.InboundDomain, e.QueueID, e.Upstream,
 	)
 	if err != nil {
 		return fmt.Errorf("insert inbound_spam_log: %w", err)
@@ -94,6 +112,74 @@ func (s *Store) InsertInboundSpamOnce(e InboundSpamEntry) (bool, error) {
 	return n > 0, nil
 }
 
+// updateUpstreamQuery puts one delivery attempt's result on the rows of a
+// queue id and recipient whose result is not final; expireUpstreamQuery makes
+// those of a queue id bounced, saying it expired after the last reply. Both
+// read the queue id through idx_inbound_spam_log_queue_id
+// (TestInboundUpstreamUsesItsIndex).
+const (
+	updateUpstreamQuery = `UPDATE inbound_spam_log
+		SET upstream_status = ?, upstream_relay = ?, upstream_reply = ?, upstream_at = ?
+		WHERE queue_id = ? AND queue_id <> '' AND to_addr = ? COLLATE NOCASE
+		  AND upstream_status IN ('queued', 'held', 'deferred')`
+	expireUpstreamQuery = `UPDATE inbound_spam_log
+		SET upstream_status = 'bounced', upstream_at = ?,
+		    upstream_reply = substr(CASE WHEN upstream_reply = '' THEN 'expired in the queue'
+		                                 ELSE 'expired in the queue; last reply: ' || upstream_reply END, 1, ?)
+		WHERE queue_id = ? AND queue_id <> '' AND upstream_status IN ('queued', 'held', 'deferred')`
+	forgetQueueIDQuery = `UPDATE inbound_spam_log
+		SET queue_id = '',
+		    upstream_status = CASE WHEN upstream_status IN ('queued', 'held', 'deferred') THEN '' ELSE upstream_status END
+		WHERE queue_id = ? AND queue_id <> ''`
+)
+
+// ForgetInboundQueueID detaches the journal rows of a queue id from it: those
+// still open lose their result too, as nothing will ever report it. The
+// milter calls it before it writes a new message's rows - Postfix reuses
+// queue ids (enable_long_queue_ids=no, the default: microseconds and the
+// queue file's inode, src/global/mail_queue.h), and only one live queue file
+// has an id, so older rows of it are of a message that is gone and must not
+// take the new one's results. The tailer calls it when cleanup removes a
+// queue file it did not queue ("removed (canceled)"): the milter had written
+// the rows, but the message was never queued.
+func (s *Store) ForgetInboundQueueID(queueID string) error {
+	if _, err := s.db.Exec(forgetQueueIDQuery, queueID); err != nil {
+		return fmt.Errorf("forget inbound queue id %s: %w", queueID, err)
+	}
+	return nil
+}
+
+// UpstreamReplyMax is how much of an upstream's reply a row keeps.
+const UpstreamReplyMax = 255
+
+// UpdateInboundUpstream puts the result of one attempt to hand a message to
+// its upstream on its journal rows - those of queueID and recipient (case
+// folded, as Postfix logs the address it was given) - unless they hold a
+// final one already: a line read twice after a restart, or an older one read
+// after a newer, does not undo a sent. relay is the host that answered, ""
+// when there was no session; reply is cut to UpstreamReplyMax by the caller.
+// It returns how many rows took it: 0 for mail that is not port 25's. A
+// message the filter had held takes its results once it is released.
+func (s *Store) UpdateInboundUpstream(queueID, recipient, status, relay, reply string, at time.Time) (int64, error) {
+	res, err := s.db.Exec(updateUpstreamQuery, status, relay, reply, at.UTC().Format(time.RFC3339), queueID, recipient)
+	if err != nil {
+		return 0, fmt.Errorf("update inbound upstream %s: %w", queueID, err)
+	}
+	return res.RowsAffected()
+}
+
+// ExpireInboundUpstream makes the rows of a message Postfix gave up on
+// bounced (qmgr's "status=expired, returned to sender", which names no
+// recipient): the sender was sent a bounce. The last reply stays, after a
+// note that the message expired.
+func (s *Store) ExpireInboundUpstream(queueID string, at time.Time) (int64, error) {
+	res, err := s.db.Exec(expireUpstreamQuery, at.UTC().Format(time.RFC3339), UpstreamReplyMax, queueID)
+	if err != nil {
+		return 0, fmt.Errorf("expire inbound upstream %s: %w", queueID, err)
+	}
+	return res.RowsAffected()
+}
+
 // PruneInboundSpam deletes journal rows older than cutoff and then, if more
 // than keep rows are left, the oldest of those. It returns how many went.
 func (s *Store) PruneInboundSpam(cutoff time.Time, keep int) (int64, error) {
@@ -130,6 +216,12 @@ type InboundSpamRow struct {
 	Engine        string
 	Detail        string
 	CreatedAt     time.Time
+	// The upstream's answer: its status ("" for none), the host that gave
+	// it, the reply and when.
+	Upstream      string
+	UpstreamRelay string
+	UpstreamReply string
+	UpstreamAt    time.Time
 }
 
 // InboundSpamFilter narrows QueryInboundSpam and CountInboundSpam. Like
@@ -166,7 +258,8 @@ func (s *Store) QueryInboundSpam(filter InboundSpamFilter, limit, offset int) ([
 	where, args := inboundSpamWhere(filter)
 	args = append(args, limit, offset)
 	rows, err := s.db.Query(
-		`SELECT id, inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at
+		`SELECT id, inbound_domain, client_ip, from_addr, to_addr, subject, decision, engine, detail, created_at,
+		        upstream_status, upstream_relay, upstream_reply, upstream_at
 		 FROM inbound_spam_log`+where+`
 		 ORDER BY created_at DESC, id DESC
 		 LIMIT ? OFFSET ?`,
@@ -180,14 +273,18 @@ func (s *Store) QueryInboundSpam(filter InboundSpamFilter, limit, offset int) ([
 	var out []InboundSpamRow
 	for rows.Next() {
 		var (
-			row     InboundSpamRow
-			created string
+			row          InboundSpamRow
+			created, upd string
 		)
 		if err := rows.Scan(&row.ID, &row.InboundDomain, &row.ClientIP, &row.From, &row.To,
-			&row.Subject, &row.Decision, &row.Engine, &row.Detail, &created); err != nil {
+			&row.Subject, &row.Decision, &row.Engine, &row.Detail, &created,
+			&row.Upstream, &row.UpstreamRelay, &row.UpstreamReply, &upd); err != nil {
 			return nil, fmt.Errorf("scan inbound_spam_log row: %w", err)
 		}
 		row.CreatedAt, _ = time.Parse(time.RFC3339, created)
+		if upd != "" {
+			row.UpstreamAt, _ = time.Parse(time.RFC3339, upd)
+		}
 		out = append(out, row)
 	}
 	return out, rows.Err()

internal/store/inboundspam_test.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"strings"
 	"testing"
 	"time"
 )
@@ -344,3 +345,143 @@ func TestInsertInboundSpamOnce(t *testing.T) {
 		t.Errorf("rows = %d, %v; want 3", n, err)
 	}
 }
+
+// Accepted mail starts queued with its queue id; each delivery line moves it
+// on until sent or bounced, which no later or repeated line undoes. A
+// refusal, which has no queue id, and another recipient of the message are
+// left alone; the queue id goes by its own index.
+func TestInboundUpstream(t *testing.T) {
+	st := openTestStore(t)
+	accepted := InboundSpamEntry{
+		InboundDomain: "in.example.com", ClientIP: "203.0.113.7", From: "s@example.net", To: "a@in.example.com",
+		Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, QueueID: "4A1B2C3D", Upstream: UpstreamQueued,
+	}
+	other := accepted
+	other.To = "b@in.example.com"
+	refused := InboundSpamEntry{InboundDomain: "in.example.com", From: "s@example.net", To: "a@in.example.com",
+		Decision: SpamDecisionReject, Engine: SpamEnginePostfix, Detail: "550 5.1.1 unknown"}
+	for _, e := range []InboundSpamEntry{accepted, other, refused} {
+		if err := st.InsertInboundSpam(e); err != nil {
+			t.Fatal(err)
+		}
+	}
+	result := func(to string) InboundSpamRow {
+		t.Helper()
+		rows, err := st.QueryInboundSpam(InboundSpamFilter{}, 10, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, r := range rows {
+			if r.To == to && r.Decision == SpamDecisionAccept {
+				return r
+			}
+		}
+		t.Fatalf("no accepted row for %s", to)
+		return InboundSpamRow{}
+	}
+	if r := result("a@in.example.com"); r.Upstream != UpstreamQueued || !r.UpstreamAt.IsZero() {
+		t.Errorf("a new row = %q at %v, want queued and no time", r.Upstream, r.UpstreamAt)
+	}
+
+	at := time.Date(2026, 10, 16, 10, 16, 42, 0, time.UTC)
+	step := func(status, reply string, want int64) {
+		t.Helper()
+		n, err := st.UpdateInboundUpstream("4A1B2C3D", "A@IN.example.com", status, "mx.up.example[198.51.100.5]:25", reply, at)
+		if err != nil || n != want {
+			t.Fatalf("UpdateInboundUpstream(%s) = %d, %v; want %d", status, n, err, want)
+		}
+		at = at.Add(time.Minute)
+	}
+	step(UpstreamDeferred, "451 4.7.1 Greylisted", 1)
+	if r := result("a@in.example.com"); r.Upstream != UpstreamDeferred || r.UpstreamReply != "451 4.7.1 Greylisted" ||
+		r.UpstreamRelay != "mx.up.example[198.51.100.5]:25" || !r.UpstreamAt.Equal(time.Date(2026, 10, 16, 10, 16, 42, 0, time.UTC)) {
+		t.Errorf("after a deferral: %+v", r)
+	}
+	step(UpstreamSent, "250 2.0.0 Ok: queued as CE05D10261D", 1)
+	step(UpstreamDeferred, "451 4.7.1 Greylisted", 0) // read again after a restart
+	if r := result("a@in.example.com"); r.Upstream != UpstreamSent || r.UpstreamReply != "250 2.0.0 Ok: queued as CE05D10261D" {
+		t.Errorf("a sent row was moved back: %+v", r)
+	}
+	if r := result("b@in.example.com"); r.Upstream != UpstreamQueued {
+		t.Errorf("another recipient took the result: %+v", r)
+	}
+	if n, err := st.UpdateInboundUpstream("9999", "a@in.example.com", UpstreamSent, "", "250 ok", at); err != nil || n != 0 {
+		t.Errorf("an unknown queue id = %d, %v; want 0 (outbound mail)", n, err)
+	}
+	if n, err := st.UpdateInboundUpstream("", "a@in.example.com", UpstreamSent, "", "250 ok", at); err != nil || n != 0 {
+		t.Errorf("an empty queue id took %d rows, %v - a refusal has none", n, err)
+	}
+
+	// Expiry: the other recipient, still queued, bounces, and says so.
+	if _, err := st.UpdateInboundUpstream("4A1B2C3D", "b@in.example.com", UpstreamDeferred, "", "connect to mx.up.example[198.51.100.5]:25: Connection refused", at); err != nil {
+		t.Fatal(err)
+	}
+	if n, err := st.ExpireInboundUpstream("4A1B2C3D", at); err != nil || n != 1 {
+		t.Fatalf("ExpireInboundUpstream = %d, %v; want 1 (the sent row stays)", n, err)
+	}
+	if r := result("b@in.example.com"); r.Upstream != UpstreamBounced ||
+		r.UpstreamReply != "expired in the queue; last reply: connect to mx.up.example[198.51.100.5]:25: Connection refused" {
+		t.Errorf("after the expiry: %+v", r)
+	}
+	if err := st.InsertInboundSpam(InboundSpamEntry{InboundDomain: "in.example.com", To: "c@in.example.com",
+		Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, Upstream: "lost"}); err == nil {
+		t.Error("an unknown upstream status was stored")
+	}
+
+	for name, q := range map[string]string{"update": updateUpstreamQuery, "expire": expireUpstreamQuery} {
+		args := []any{"sent", "", "", "", "4A1B2C3D", "a@in.example.com"}
+		if name == "expire" {
+			args = []any{"", 255, "4A1B2C3D"}
+		}
+		if plan := explain(t, st, q, args...); !strings.Contains(plan, "USING INDEX idx_inbound_spam_log_queue_id") {
+			t.Errorf("%s does not use idx_inbound_spam_log_queue_id:\n%s", name, plan)
+		}
+	}
+}
+
+// A held message takes its result once released; a queue id Postfix uses
+// again detaches the rows of the message that had it - the open ones lose
+// their result, nothing reports it now - so the new message's lines and its
+// expiry reach only its own rows.
+func TestInboundUpstreamHeldAndReusedQueueID(t *testing.T) {
+	st := openTestStore(t)
+	old := InboundSpamEntry{InboundDomain: "in.example.com", To: "a@in.example.com",
+		Decision: SpamDecisionQuarantine, Engine: SpamEngineRspamd, QueueID: "4A1B", Upstream: UpstreamHeld}
+	if err := st.InsertInboundSpam(old); err != nil {
+		t.Fatal(err)
+	}
+	at := time.Date(2026, 10, 16, 10, 0, 0, 0, time.UTC)
+	if n, err := st.UpdateInboundUpstream("4A1B", "a@in.example.com", UpstreamDeferred, "", "451 later", at); err != nil || n != 1 {
+		t.Fatalf("a released held message: %d, %v", n, err)
+	}
+	if err := st.ForgetInboundQueueID("4A1B"); err != nil {
+		t.Fatal(err)
+	}
+	fresh := InboundSpamEntry{InboundDomain: "in.example.com", To: "a@in.example.com",
+		Decision: SpamDecisionAccept, Engine: SpamEngineSelfPost, QueueID: "4A1B", Upstream: UpstreamQueued}
+	if err := st.InsertInboundSpam(fresh); err != nil {
+		t.Fatal(err)
+	}
+	if n, err := st.ExpireInboundUpstream("4A1B", at); err != nil || n != 1 {
+		t.Fatalf("expiry of the reused id took %d rows, %v; want the new message's one", n, err)
+	}
+	rows, err := st.QueryInboundSpam(InboundSpamFilter{}, 10, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, r := range rows {
+		switch r.Decision {
+		case SpamDecisionQuarantine:
+			if r.Upstream != "" || r.UpstreamReply != "451 later" {
+				t.Errorf("the detached row = %+v, want no result and its last reply kept", r)
+			}
+		case SpamDecisionAccept:
+			if r.Upstream != UpstreamBounced {
+				t.Errorf("the new row = %+v, want bounced", r)
+			}
+		}
+	}
+	if plan := explain(t, st, forgetQueueIDQuery, "4A1B"); !strings.Contains(plan, "USING INDEX idx_inbound_spam_log_queue_id") {
+		t.Errorf("ForgetInboundQueueID does not use idx_inbound_spam_log_queue_id:\n%s", plan)
+	}
+}

internal/store/migrations/0012_inbound_upstream.sql

@@ -0,0 +1,20 @@
+-- What the upstream answered, on the inbound journal's rows
+-- (inbound-upstream-result, 2.10.0; owner, 2026-10-16). A row of mail
+-- accepted on port 25 keeps the message's queue id, which the inbound journal
+-- milter reads at end-of-message, and starts "queued" ("held" when the filter
+-- had it held in the queue); the log tailer puts
+-- each delivery attempt's result on it - the status, the host that answered
+-- ("" when Postfix had no session), its reply and when - from the postfix/smtp
+-- line of that queue id and recipient. sent and bounced are final. A refused
+-- message has no queue id and no result, and neither has a row written before
+-- this file.
+ALTER TABLE inbound_spam_log ADD COLUMN queue_id TEXT NOT NULL DEFAULT '';
+ALTER TABLE inbound_spam_log ADD COLUMN upstream_status TEXT NOT NULL DEFAULT ''
+    CHECK (upstream_status IN ('', 'queued', 'held', 'sent', 'deferred', 'bounced'));
+ALTER TABLE inbound_spam_log ADD COLUMN upstream_relay TEXT NOT NULL DEFAULT '';
+ALTER TABLE inbound_spam_log ADD COLUMN upstream_reply TEXT NOT NULL DEFAULT '';
+ALTER TABLE inbound_spam_log ADD COLUMN upstream_at TEXT NOT NULL DEFAULT '';
+
+-- The tailer finds a row by its queue id at every delivery line; only rows
+-- of accepted mail have one.
+CREATE INDEX idx_inbound_spam_log_queue_id ON inbound_spam_log(queue_id) WHERE queue_id <> '';

internal/store/schema_test.go

@@ -57,7 +57,7 @@ func TestBaselineSchema(t *testing.T) {
 		"user_inbound_domains":  "user_id inbound_domain_id",
 		"dmarc_reports":         "id domain reporter report_id period_begin period_end received_at contact_email policy_p policy_sp policy_pct policy_adkim policy_aspf pass_count fail_count recipient",
 		"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_domain_id",
+		"inbound_spam_log":      "id inbound_domain client_ip from_addr to_addr subject decision engine detail created_at inbound_domain_id queue_id upstream_status upstream_relay upstream_reply upstream_at",
 		"inbound_spam_list":     "id list_type entry_type value note created_at",
 		"send_log_lines":        "id queue_id logged_at line",
 		"notify_state":          "key phase count mark since sent_at",
@@ -108,7 +108,7 @@ func TestBaselineSchema(t *testing.T) {
 	rows.Close()
 	wantIdx := []string{
 		"idx_applications_login_nocase", "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_inbound_spam_log_domain_id_created_at",
+		"idx_inbound_spam_log_created_at", "idx_inbound_spam_log_domain_created_at", "idx_inbound_spam_log_domain_id_created_at", "idx_inbound_spam_log_queue_id",
 		"idx_send_log_app_login", "idx_send_log_app_login_created_at", "idx_send_log_created_at", "idx_send_log_domain", "idx_send_log_domain_id", "idx_send_log_domain_id_created_at",
 		"idx_send_log_lines_queue_id", "idx_send_log_open", "idx_send_log_queue_id", "idx_send_log_refusal_line", "idx_send_log_rejected",
 		"idx_sessions_expires_at",
@@ -119,8 +119,8 @@ func TestBaselineSchema(t *testing.T) {
 	}
 
 	var version, appID int64
-	if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 11 {
-		t.Errorf("user_version = %d, %v; want 11", version, err)
+	if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 12 {
+		t.Errorf("user_version = %d, %v; want 12", 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)
@@ -196,8 +196,8 @@ func TestOpenMigratesA20Database(t *testing.T) {
 	}
 	defer st.Close()
 	var version int
-	if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 11 {
-		t.Errorf("user_version = %d, %v; want 11", version, err)
+	if err := st.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil || version != 12 {
+		t.Errorf("user_version = %d, %v; want 12", version, err)
 	}
 	if ok, err := st.UserExists(); err != nil || !ok {
 		t.Errorf("the user of the 2.0 database: exists = %v, %v", ok, err)