feat(inbound): at-least-once inbound on River (queue-first, QueueInbound)#391
Merged
Conversation
…bound slice 1) deliverToAgent swallowed the messages-insert error (log + return) and deliverMessages counted it delivered anyway, so a failed persist still returned 250 OK — the sending MTA treats 250 as done and never retries, silently losing the mail (docs/events.md even advertised email.received as 'at-least-once', which this contradicted). Now deliverToAgent returns the persist error and deliverMessages maps it to a transient 451 (4.3.0), so the sender retries the whole message per RFC 5321 instead of dropping it. Screening/metering still fail open (they don't error out here); only a genuine persist failure yields 451. Multi-recipient caveat: a retry re-delivers to already-succeeded recipients (the sync path has no dedup) — duplicate beats loss, and the queue-first inbound path (E2A_INBOUND_MODE=async, upcoming slices) collapses it. Regression test forces the insert to fail (agent-scoped trigger) and asserts the SMTP client gets 451 with nothing persisted. Also lands the migration design doc. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Layer 1 of the queue-first inbound pipeline: the durable landing pad the SMTP session writes before 250. Migration 056 adds inbound_intake (raw MIME + envelope + remote_ip for SPF + dedup key + status/job/message_fk), a UNIQUE(recipient,message_id, content_hash) dedup index, and partial indexes for the reconciler + retention sweep. Store methods: InsertInboundIntakeTx (ON CONFLICT DO NOTHING → inserted bool so a lost-ack MTA retry is an idempotent no-op accept), StampInboundIntakeJobIDTx, LoadInboundIntake (nil when pruned), MarkInboundIntakeProcessedTx (the worker's idempotency gate), MarkInboundIntakeFailed. DB-backed tests cover dedup, load round-trip, and the processed/failed transitions. Not a hand-rolled queue — River owns claim/retry via the job; this table is the record the job reads. No wire/API change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ync/async (inbound slice 3a) Pulls the full inbound processing (parse, SPF/DKIM, HMAC sign, ingestion gate, content screening, persist, event publish) out of session.deliverToAgent into (*Server).processInbound(ctx, inboundInput, postPersistHook). Everything it needs is derivable from the connection facts (body, envelope_from, remote_ip, recipient) — the exact contents of an inbound_intake row — so the SAME implementation serves the synchronous SMTP session (hook=nil) and the upcoming async River worker (hook = MarkInboundIntakeProcessedTx, run inside the persist tx so the intake flips to processed atomically with the messages insert + events — its idempotency gate). envelopeFromTrusted/senderResolvable become *Server methods taking the envelope explicitly. No behavior change on the sync path: the full relay suite passes unchanged. No worker yet (slice 3b). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Layer 3 of the queue-first inbound pipeline: a River worker that takes an accepted inbound_intake row and runs the shared processInbound chain off the SMTP critical path. Mirrors internal/outboundsend. - QueueInbound added to the shared jobs package (isolated concurrency so an inbound spike / the per-message Gemini screen can't starve outbound or webhook lanes) + InboundWorkers config (default 8). - InboundProcessWorker: load intake → no-op if pruned or already processed/failed (idempotency gate) → ProcessIntake → transient errors retry the bounded envelope, exhaustion marks the intake failed (ops visibility; we already 250'd). One pass per attempt; River owns the retry envelope via NextRetry. - Jobs registrar + EnqueueInboundProcessTx (InsertTx on QueueInbound) for the accept-tx. relay.Server.ProcessIntake is the Processor adapter (rebuilds the connection context from the row, hooks MarkInboundIntakeProcessedTx into the persist tx). Worker unit tests cover success / gone / already-processed / transient retry / final-attempt-fails. Not yet wired into the relay (slice 4 does the accept-tx + E2A_INBOUND_MODE flag). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…nbound slice 4)
Wires the async inbound path end to end, gated by E2A_INBOUND_MODE (default sync,
unchanged). When async, session.Data calls acceptInbound: for each recipient, INSERT
inbound_intake (dedup on recipient+message_id+content_hash) + EnqueueInboundProcessTx
+ stamp job id, ALL in one tx, before 250. Commit ⇒ 250 (durably accepted; River
delivers at-least-once); commit failure ⇒ 451 (sender retries, no silent loss). A
lost-ack resend hits the dedup key → idempotent 250, no second row/job.
- config.InboundConfig{Mode} + E2A_INBOUND_MODE (default sync).
- relay: inboundAsync + InboundEnqueuer + SetInboundEnqueuer; acceptInbound +
content-hash; ProcessIntake (the Processor adapter) already added in 3b.
- inboundprocess.Jobs late-binds the Processor (SetProcessor, mutex-guarded) — the
relay Server is built after jobs.New; the worker tolerates the brief startup window.
- main.go registers inboundJobs when async, sets processor + enqueuer after the relay
is built.
Integration test: async send → one accepted intake row + stamped job, NO messages
row (deferred); resend → dedup (one row, no second enqueue). Default sync path + full
relay suite unchanged.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ReconcilePending enqueues an inbound_process job for every accepted intake row with no job yet (process_job_id IS NULL) — the cutover for pre-async rows at the mode-flip and any row stranded by a crash between insert and enqueue. Per-row FOR UPDATE + NULL-job guard makes it idempotent across re-runs / replicas. Mirrors outboundsend.ReconcilePending. Wired in main.go after SetProcessor so re-driven jobs find a wired processor. Store interface gains StampInboundIntakeJobIDTx. Live periodic reconciler deferred as a follow-up (matching outboundsend): the accept- tx is atomic so the NULL-job set is ~empty in steady state; the startup pass re-drives on the next deploy. DB test: stranded accepted row → stamped, re-run idempotent. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A River periodic on QueueMaintenance prunes processed inbound_intake rows older than 72h (raw MIME also lives in messages.raw_message once processed, so this is non- destructive; failed rows are retained for inspection). Sweep runs every 6h, RunOnStart:false — mirrors the webhook auto-disable janitor. RegisterJobs now contributes the RetentionWorker + its periodic alongside the InboundProcessWorker, so it activates automatically in async mode. identity.PruneProcessedIntake + a DB test covering old-pruned / recent-kept / accepted-kept. Completes the queue-first inbound pipeline (slices 1-6): at-least-once from the SMTP edge (451 on accept failure), durable intake + dedup, River processing off the critical path, startup cutover, and retention. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…idempotency guard Both review passes (no blockers, at-least-once holds, sync default unchanged): - DELETED-AGENT + OVER-BILLING (both reviewers, the key one): GetAgentByID returns pgx.ErrNoRows (not (nil,nil)) for a gone agent, so the no-op drop branch was dead — a recipient deleted between accept and processing burned the full ~5.5h retry envelope AND, because usage.RecordAndCheck ran BEFORE the persist tx on every attempt, was metered ~6x while never delivered. Fixed: errors.Is(ErrNoRows) → no-op drop (restores the historical sync skip+continue + design §5); metering moved to AFTER the persist tx so it counts once per delivered message (mirrors outbound). - IDEMPOTENCY GUARD (independent): MarkInboundIntakeProcessedTx now flips WHERE status='accepted' and returns ErrIntakeAlreadyProcessed on 0 rows → the worker's persist tx rolls back (no duplicate message/event) and the worker treats it as done. - COVERAGE: ^internal/inboundprocess$ floor (65) + a real-path test exercising the actual ProcessIntake + hook atomicity + the status-guard re-drive no-op (was only fake-tested) + late-bind + retention-worker tests. - NITS: remote_ip nil-guard (store '' not '<nil>'); removed dead session thread-info fields + the now-unused senderEmail param; live-periodic-reconciler marked deferred in the design doc. Multi-recipient sync 451→duplicate is documented (by-design, duplicate beats loss; narrowed now that deleted-agent no longer 451s). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Extensive local e2e surfaced a hygiene gap: a message accepted (250) whose agent is
deleted before the worker processes it was dropped correctly (no message, not metered,
no retry storm — the prior review fix) BUT left the intake row lingering 'accepted'
forever with a completed job — orphaned raw MIME that the reconciler (job NOT null)
and retention (prunes 'processed' only) never reclaim, accumulating on agent churn.
Fix: processInbound returns the identity.ErrRecipientGone sentinel for a gone agent
(instead of a bare nil); the async worker marks the intake terminally failed
('recipient agent no longer exists') so it stops lingering, and the sync path skips
the recipient (preserving the historical skip+continue, no 451). Worker unit test +
live re-verify: the dropped intake now reaches 'failed', zero 'accepted' orphans.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…eview) Fresh review re-pass of PR #391's fix commits (both independent + adversarial: no blockers). The one should-fix: processInbound logged 'failed to record inbound message' for ErrIntakeAlreadyProcessed — but that sentinel is an EXPECTED at-least-once re-drive outcome (a prior attempt already processed the intake; the persist tx rolls back with no duplicate; the worker treats it as done), not a failure. Every worker re-drive of an already-processed intake would emit a scary error line. Special-case it before the generic error log; return behavior is identical. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
jiashuoz
added a commit
that referenced
this pull request
Jul 9, 2026
The HITL approval-notification email — the reviewer's primary signal that an outbound message is held in pending_review — was a detached goroutine (context.Background, in-process SMTP retries). A crash or SMTP outage between the 202 pending_review response and the send dropped it silently; the reviewer was never told. cmd/e2a/main.go flagged this as a known gap. Move it onto River, same three-layer pattern as inbound (#391) / outbound (#388): - The hold accept-tx (HoldForApprovalCore) now writes the pending_review row, enqueues a hitl_notify job on the new isolated QueueNotify, and stamps notify_job_id — all in ONE transaction. The notification is durably queued before the 202, so a crash can't lose it. - NotifyWorker (internal/hitlnotify) re-reads the message, no-ops on gone/resolved/expired/already-notified, sends via the no-loop SMTPRelay.SendOnce (River owns the retry envelope), and stamps notified_at only AFTER a successful send — the send-dedup marker. Loss is impossible; a rare crash-window duplicate "please review" email is the accepted at-least-once trade. - ReconcilePending re-drives holds stranded without a job at startup. At-least-once from the 202 — closes the last hole after inbound + outbound. Migrations 057 (notify_job_id + notified_at columns + a cutover UPDATE stamping notified_at on pre-existing pending_review rows, so the first deploy doesn't re-notify holds the old code already emailed) and 058 (the reconciler's partial index, built CREATE INDEX CONCURRENTLY under e2a:no-transaction so the deploy doesn't write-lock messages). Gated on the same relay+public-URL config as the notifier; unconfigured deployments keep the plain hold path (no notification), byte-for-byte unchanged. Tests: 11 worker unit tests, reconciler + cutover-skip integration, and a driven end-to-end (accept-tx → real River → fake SMTP → notified_at stamped). Design: docs/design/hitl-notify-river.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
jiashuoz
added a commit
that referenced
this pull request
Jul 9, 2026
The HITL approval-notification email — the reviewer's primary signal that an outbound message is held in pending_review — was a detached goroutine (context.Background, in-process SMTP retries). A crash or SMTP outage between the 202 pending_review response and the send dropped it silently; the reviewer was never told. cmd/e2a/main.go flagged this as a known gap. Move it onto River, same three-layer pattern as inbound (#391) / outbound (#388): - The hold accept-tx (HoldForApprovalCore) now writes the pending_review row, enqueues a hitl_notify job on the new isolated QueueNotify, and stamps notify_job_id — all in ONE transaction. The notification is durably queued before the 202, so a crash can't lose it. - NotifyWorker (internal/hitlnotify) re-reads the message, no-ops on gone/resolved/expired/already-notified, sends via the no-loop SMTPRelay.SendOnce (River owns the retry envelope), and stamps notified_at only AFTER a successful send — the send-dedup marker. Loss is impossible; a rare crash-window duplicate "please review" email is the accepted at-least-once trade. - ReconcilePending re-drives holds stranded without a job at startup. At-least-once from the 202 — closes the last hole after inbound + outbound. Migrations 057 (notify_job_id + notified_at columns + a cutover UPDATE stamping notified_at on pre-existing pending_review rows, so the first deploy doesn't re-notify holds the old code already emailed) and 058 (the reconciler's partial index, built CREATE INDEX CONCURRENTLY under e2a:no-transaction so the deploy doesn't write-lock messages). Gated on the same relay+public-URL config as the notifier; unconfigured deployments keep the plain hold path (no notification), byte-for-byte unchanged. Tests: 11 worker unit tests, reconciler + cutover-skip integration, and a driven end-to-end (accept-tx → real River → fake SMTP → notified_at stamped). Design: docs/design/hitl-notify-river.md Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This was referenced Jul 9, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Pushes the durability boundary of the inbound path to the SMTP edge, mirroring what #388 did for outbound — the symmetric "both lanes at-least-once" half. Gated behind
E2A_INBOUND_MODE=async; the defaultsyncpath is byte-for-byte unchanged (full relay suite passes untouched). Design:docs/design/inbound-message-pipeline-river.md.The problem (today, sync)
The SMTP session does ALL processing inline before
250— parse, SPF/DKIM, sender-gate, content screen (up-to-10s Gemini), persist, deliver. AnddeliverToAgentswallowed the persist error, so a failed insert still returned250 OK→ the sending MTA never retries → silent mail loss (whiledocs/events.mdadvertisedemail.receivedas "at-least-once"). No inbound dedup either — an MTA retry made a second message.The pipeline (async)
Three layers, like webhook/outbound:
The only work gating
250is the durable accept. A commit failure ⇒451(sender retries, no loss). The dedup key(recipient, message_id, content_hash)makes a lost-ack retry an idempotent250. The worker flips the intake toprocessedatomically with the messages insert, so a crash re-drive is a no-op. Delivery reuses the existing outbox→River webhook path unchanged.Slices
deliverToAgentpropagates the persist error →Datareturns 451 (engine-agnostic, ships the silent-loss fix immediately).inbound_intaketable (migration 056) + store surface (insert/dedup/load/mark/stamp).processInboundextraction — the full chain pulled out of the session into one*Servermethod shared by the sync session (hook=nil) and the async worker (hook=MarkInboundIntakeProcessedTx, run in the persist tx). No sync behavior change. Then the River worker onQueueInbound(internal/inboundprocess).E2A_INBOUND_MODE—session.Databranches toacceptInboundwhen async;contentHashHexdedup; wired in main.go (the Processor is late-bound — the relay is built afterjobs.New).Key design notes
inbound_intaketable, not a minimalmessagesrow: inbound can't produce a complete messages row at accept (subject/sender/auth/conversation come from the worker's parse), so a placeholder row would pollute the product's central read model. Unlike outbound, whoseacceptedrow is complete-but-undelivered.remote_ippersisted because SPF needs the connecting IP, unavailable to the async worker.Verification
go build ./..., full affected suite (inboundprocess,identity,relay,config,jobs), spec-check — all green. Unit tests: 451-on-persist-failure, dedup, worker idempotency (success/gone/already-processed/retry/final-fail), reconciler, prune.250→ intakeaccepted→processed+linked → messages row created off the SMTP path →email.receiveddelivered to the subscriber; resend of the same Message-ID → one intake + one message (dedup); sync-mode (flag off) → inline processing, no intake row (default unchanged). Clean logs; torn down.Deferred (follow-ups; not blocking the guarantee)
Live periodic reconciler (matching outbound — the accept-tx is atomic so the NULL-job set is ~empty; startup pass re-drives on deploy); an
email.rejectedevent for terminally-unparseable bodies (v1 marks intakefailed+ drops silently, as we already250'd).🤖 Generated with Claude Code