From d71c1c3b9b9726988ee24f2cd170f1b549c7ccf7 Mon Sep 17 00:00:00 2001 From: jiashuoz Date: Thu, 9 Jul 2026 12:20:45 -0700 Subject: [PATCH 1/3] chore(webhookpub): remove dead code + fix docstring drift MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Readability pass on webhookpub (no behavior change): - Delete committed AI-scratchpad text + 3 no-op `var _` compile guards in worker.go, and the now-unused errors/encoding-json/pgconn imports. - Delete unused `nowUTC` var (+ its time import) in outbox.go. - Delete the dead `oldest` telemetry scaffolding in Tick (declared, ranged into nothing, discarded). - Fix docstring drift: writeOutboxRow's "(eventually) PublishBestEffortTx" (already wired) + acknowledge that in river mode nothing LISTENs the NOTIFY; PublishBestEffortTx's reference to a webhook_publish_failures table that was never built + the "legacy-fallback" line that contradicts the file's own "no legacy fallback path left"; the pg_notify block now notes it's a harmless no-op in river mode. - Rewrite the event.go package header to describe the current durable three-layer outbox→fan-out→River pipeline (it still described the retired in-process post-commit-goroutine publisher + a tmp/ design path). - Drop stale internal "slice-N" markers; note both fan-out engines share fanOutEventCore in the matches/Publisher docs. - Standardize the fan-out job Kind() "webhook_fan_out" → "webhook_fanout" to match webhook_fanout_reconcile + the rest of the "fanout" naming (safe: E2A_WEBHOOK_FANOUT_MODE defaults legacy, so no such jobs exist in prod yet). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01DfAXBRMJ3KPuU3UEgwqjhS --- internal/config/config.go | 2 +- internal/webhookpub/event.go | 42 +++++++++++++------------ internal/webhookpub/fanout_worker.go | 2 +- internal/webhookpub/outbox.go | 34 ++++++++++---------- internal/webhookpub/publisher.go | 6 ++-- internal/webhookpub/worker.go | 47 +++++++--------------------- 6 files changed, 54 insertions(+), 79 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index f608f279..05798db9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -153,7 +153,7 @@ type InboundConfig struct { // river-migration.md). Mode="legacy" (the default) drains webhook_events → // webhook_subscriber_deliveries via the in-process webhookpub.OutboxWorker // (LISTEN/NOTIFY + poll + SKIP-LOCKED lease). Mode="river" opts into the River -// pipeline: PublishTx/PublishBestEffortTx enqueue a webhook_fan_out job in the event's +// pipeline: PublishTx/PublishBestEffortTx enqueue a webhook_fanout job in the event's // tx and the webhookpub.FanOutWorker does the match/insert/enqueue off the drain loop. // Override with E2A_WEBHOOK_FANOUT_MODE. Any value other than "river" is treated as // "legacy" (fail-safe to the unchanged path). Wired in Slice 2; unused under legacy. diff --git a/internal/webhookpub/event.go b/internal/webhookpub/event.go index 841204d6..ecfb7ffd 100644 --- a/internal/webhookpub/event.go +++ b/internal/webhookpub/event.go @@ -1,20 +1,21 @@ -// Package webhookpub publishes events from the e2a core (relay, -// outbound sender, HITL flow) to subscribers registered via the new -// /v1/webhooks resource. It runs in-process and post-commit -// async: trigger code commits its primary DB write, then calls -// Publisher.Publish in a goroutine. The publisher matches the event -// against enabled subscribers (event type + filters), inserts one -// webhook_subscriber_deliveries row per match, and returns; actual -// HTTP delivery is the retry worker's job. +// Package webhookpub publishes events from the e2a core (relay, outbound sender, +// HITL flow) to subscribers registered via the /v1/webhooks resource, using a durable +// three-layer pipeline: // -// Slice 1 only fires email.received from the relay. Later slices add -// email.sent and the unified review-hold events (email.pending_review, -// email.review_approved, email.review_rejected). +// Layer 1 — webhook_events (the outbox): trigger code writes the event row inside its +// own DB transaction (Outbox.PublishTx / PublishBestEffortTx), so the event +// is durable the moment the business state commits. +// Layer 2 — webhook_subscriber_deliveries: the event is fanned out to one row per +// matching enabled subscriber (event type + filters). Two interchangeable +// fan-out engines drain Layer 1: the legacy in-process OutboxWorker +// (LISTEN/NOTIFY + poll, the default) and the River FanOutWorker +// (E2A_WEBHOOK_FANOUT_MODE=river). Both share fanOutEventCore. +// Layer 3 — HTTP POST: internal/webhookdelivery's River DeliverWorker delivers each +// Layer 2 row with retries + HMAC signing. // -// This is the sole push path: the legacy per-agent -// agent_identities.webhook_url + agent_mode columns (and the -// PersistentDeliverer that served them) were removed in slice 3 -// (migration 029). See the final design at tmp/e2a_webhooks_design.md. +// This is the sole push path: the legacy per-agent agent_identities.webhook_url + +// agent_mode columns (and the PersistentDeliverer that served them) were removed in +// migration 029. package webhookpub import ( @@ -39,8 +40,9 @@ const ( // - email.deferred: a transient delay the worker is still retrying (SES 4xx // / ramp deferral) — not terminal; a later email.sent or email.failed // follows. Peers (Resend/SendGrid) push the same delayed/deferred signal. - // Emitted synchronously today (the sync server already knows the outcome); - // they become the primary async signal once the worker pool ships (slice 3). + // Emitted synchronously in sync outbound mode (the server already knows the + // outcome); the async send pipeline (E2A_OUTBOUND_MODE=async) makes them the + // primary signal. EventEmailFailed = "email.failed" EventEmailDeferred = "email.deferred" // Review-hold lifecycle (unified, direction-aware — design 2026-06-22). A held @@ -84,9 +86,9 @@ const ( EventEmailPendingReview = "email.pending_review" ) -// AllEventTypes is the canonical allowlist of event names. Used by -// the slice-2 handler validation. Adding a new event type means -// adding a constant above AND extending this slice. +// AllEventTypes is the canonical allowlist of event names. Used by the handler's +// event-type validation. Adding a new event type means adding a constant above AND +// extending this slice. var AllEventTypes = []string{ EventEmailReceived, EventEmailSent, diff --git a/internal/webhookpub/fanout_worker.go b/internal/webhookpub/fanout_worker.go index 4ac3b722..18b814e8 100644 --- a/internal/webhookpub/fanout_worker.go +++ b/internal/webhookpub/fanout_worker.go @@ -26,7 +26,7 @@ type FanOutArgs struct { EventID string `json:"event_id"` } -func (FanOutArgs) Kind() string { return "webhook_fan_out" } +func (FanOutArgs) Kind() string { return "webhook_fanout" } const ( // maxFanOutAttempts bounds River's retries of a fan-out job. Fan-out failures are diff --git a/internal/webhookpub/outbox.go b/internal/webhookpub/outbox.go index 7d36141e..5d7bbdaf 100644 --- a/internal/webhookpub/outbox.go +++ b/internal/webhookpub/outbox.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "log" - "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" @@ -39,8 +38,8 @@ type Outbox interface { // PublishBestEffortTx attempts the outbox write inside the // caller's transaction but never returns an error. On failure, - // logs to webhook_publish_failures (slice 4 will add the table) - // and lets the caller's tx commit anyway. + // logs and lets the caller's tx commit anyway (a future slice may pipe these + // to a webhook_publish_failures table — not yet built). // // Returns `wrote=true` iff the row was actually written (flag // enabled AND writeOutboxRow succeeded). The wrote signal is retained for @@ -66,7 +65,7 @@ type Outbox interface { // SetFanOutEnqueuer wires the River fan-out enqueuer (two-phase, called after the // shared client exists). When set (E2A_WEBHOOK_FANOUT_MODE=river), PublishTx / - // PublishBestEffortTx enqueue a webhook_fan_out job in the event's own tx and stamp + // PublishBestEffortTx enqueue a webhook_fanout job in the event's own tx and stamp // fanout_job_id — the River FanOutWorker then does the fan-out, replacing the // legacy in-process OutboxWorker drain. nil (default/legacy) ⇒ the event row is // written as before and the OutboxWorker fans it out via LISTEN/NOTIFY + poll. @@ -230,10 +229,8 @@ func (o *outbox) PublishBestEffortTx(ctx context.Context, tx pgx.Tx, e Event) (w // Best-effort: log and return wrote=false. The caller's tx // commits the business state regardless because the // irreversible action (SES.Send) already happened — rolling - // back would orphan a sent email. A future slice will pipe - // these failures to a webhook_publish_failures table; for now - // the log is the only signal AND the legacy-fallback signal - // for the caller. + // back would orphan a sent email. For now the log is the only + // signal (a future slice may add a webhook_publish_failures table). log.Printf("[outbox] PublishBestEffortTx err (event=%s type=%s): %v", e.ID, e.Type, err) return false } @@ -246,10 +243,11 @@ func (o *outbox) PublishBestEffortTx(ctx context.Context, tx pgx.Tx, e Event) (w return true } -// writeOutboxRow is the SQL body shared by PublishTx and (eventually) -// PublishBestEffortTx. Idempotent on (id): a retried trigger with the -// same deterministic id no-ops the second INSERT. Issues pg_notify so -// the slice-2 worker wakes immediately on commit. +// writeOutboxRow is the SQL body shared by PublishTx and PublishBestEffortTx. +// Idempotent on (id): a retried trigger with the same deterministic id no-ops the +// second INSERT (RowsAffected reports 0, returned as inserted=false). Issues pg_notify +// so the legacy OutboxWorker (LISTEN/NOTIFY mode) wakes immediately on commit; in +// river fan-out mode nothing LISTENs and the NOTIFY is a harmless no-op. func writeOutboxRow(ctx context.Context, exec outboxExecutor, e Event) (inserted bool, err error) { if e.ID == "" { return false, fmt.Errorf("webhookpub: outbox event must have non-empty ID") @@ -302,9 +300,14 @@ func writeOutboxRow(ctx context.Context, exec outboxExecutor, e Event) (inserted // insert — a deduped event already has one. inserted = tag.RowsAffected() > 0 + // In river fan-out mode no worker LISTENs webhook_events_new (the OutboxWorker + // isn't started), so this NOTIFY is wasted — one cheap statement per event. It + // stays unconditional to keep the legacy default working; drop it when the legacy + // OutboxWorker is deleted post-cutover. + // // pg_notify is best-effort: NOTIFY only fires on COMMIT (Postgres // queues it). If COMMIT fails, no notification is emitted. The - // slice-2 worker's 1s fallback poll catches missed wakeups + // legacy OutboxWorker's 1s fallback poll catches missed wakeups // (deploy windows, LISTEN reconnect races). Payload is empty // because the worker rescans the table anyway. // @@ -327,8 +330,3 @@ func writeOutboxRow(ctx context.Context, exec outboxExecutor, e Event) (inserted } return inserted, nil } - -// Time helpers — kept here rather than relying on time.Now() in -// callers so test code can swap them. Not currently used; slice 2's -// worker will need a clock abstraction. -var nowUTC = func() time.Time { return time.Now().UTC() } diff --git a/internal/webhookpub/publisher.go b/internal/webhookpub/publisher.go index 6bd718b9..93cceb19 100644 --- a/internal/webhookpub/publisher.go +++ b/internal/webhookpub/publisher.go @@ -79,8 +79,8 @@ func (f StaticFlag) Enabled() bool { return bool(f) } // a "labels = [urgent]" filter does NOT match an event whose Labels // slice is empty/nil — i.e. unlabelled inbound mail is silently // skipped when the subscriber has a label filter. This is the -// explicit semantic chosen in the design (H5). Shared with the outbox -// drain (OutboxWorker.fanOutOne). +// explicit semantic chosen in the design (H5). Shared by both fan-out +// engines via fanOutEventCore (legacy OutboxWorker + River FanOutWorker). func matches(e Event, f identity.WebhookFilters) bool { if len(f.AgentIDs) > 0 { if e.AgentID == "" { @@ -119,7 +119,7 @@ func contains(haystack []string, needle string) bool { } func intersects(a, b []string) bool { - // O(len(a) * len(b)) — fine at slice 1 scale (filters cap 50, + // O(len(a) * len(b)) — fine at this scale (filters cap 50, // event labels cap 100 per the labels feature design). for _, x := range a { for _, y := range b { diff --git a/internal/webhookpub/worker.go b/internal/webhookpub/worker.go index ae0b5f64..6922e00a 100644 --- a/internal/webhookpub/worker.go +++ b/internal/webhookpub/worker.go @@ -2,8 +2,6 @@ package webhookpub import ( "context" - "encoding/json" - "errors" "fmt" "log" "strings" @@ -13,7 +11,6 @@ import ( "github.com/Mnexa-AI/e2a/internal/identity" "github.com/Mnexa-AI/e2a/internal/telemetry" "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) @@ -25,10 +22,10 @@ import ( // §4.4 for the full architecture. // // Wakeup paths: -// 1. LISTEN webhook_events_new — dedicated connection, sub-50ms -// latency from trigger COMMIT to fan-out. -// 2. 1s fallback poll — catches notifications missed during deploy -// or LISTEN reconnect. +// 1. LISTEN webhook_events_new — dedicated connection, sub-50ms +// latency from trigger COMMIT to fan-out. +// 2. 1s fallback poll — catches notifications missed during deploy +// or LISTEN reconnect. // // Multi-replica safety: // - FOR UPDATE SKIP LOCKED + next_poll_at bump in GetPending leases @@ -88,7 +85,7 @@ type identityReader interface { ListEnabledWebhooksForRouting(ctx context.Context, userID, eventType string) ([]identity.Webhook, error) } -// NewOutboxWorker constructs the slice-2 worker. Production wiring +// NewOutboxWorker constructs the legacy LISTEN/NOTIFY fan-out drain. Production wiring // passes the real pool and identity.Store; tests can pass fakes. func NewOutboxWorker(pool *pgxpool.Pool, store identityReader) *OutboxWorker { return &OutboxWorker{ @@ -232,17 +229,10 @@ func (w *OutboxWorker) Tick(ctx context.Context) { if !notifyWake { w.metrics.NotifyMissed() } - // Slice 10 telemetry hook: log batch size + age of oldest row so - // publisher lag can be derived from access logs. A future - // follow-up wires real Prometheus/OTLP counters. - var oldest time.Time - for _, ev := range events { - _ = ev - } - log.Printf("[outbox-worker-metrics] tick batch=%d oldest_age_estimate=lease-bound elapsed_ms_so_far=%d", + // Publisher-lag is emitted via SetPublisherLag above; this line logs batch size + + // tick latency so lag can also be derived from access logs. + log.Printf("[outbox-worker-metrics] tick batch=%d elapsed_ms_so_far=%d", len(events), time.Since(tickStart).Milliseconds()) - _ = oldest - _ = oldestAge // already emitted via SetPublisherLag sem := make(chan struct{}, w.concurrency) var wg sync.WaitGroup @@ -526,20 +516,9 @@ func derefString(p *string) string { return *p } -// BeginFunc helper for pgxpool.Pool — pgx v5 doesn't export the v4 -// shorthand, so we replicate it here. Used by fanOutOne so the tx -// boundary is implicit in the lambda's lifetime. -// -// Defined on *pgxpool.Pool via an alias is not Go-legal; instead this -// is a free function we call as w.pool.BeginFunc(ctx, fn). Wait — -// that's also not legal. Let me inline this where needed... actually -// pgx v5 does have BeginFunc-style helpers. Just use Begin + manual -// commit/rollback. -var _ = errors.New -var _ pgconn.CommandTag - -// poolBeginFunc emulates pgx v4's BeginFunc against a *pgxpool.Pool -// (pgx v5 dropped the helper; we wrote our own). +// poolBeginFunc emulates pgx v4's BeginFunc against a *pgxpool.Pool (pgx v5 dropped +// the helper): Begin → fn → Commit, rolling back on error. Shared by both fan-out +// workers via fanOutEventCore. func poolBeginFunc(ctx context.Context, pool *pgxpool.Pool, fn func(tx pgx.Tx) error) error { tx, err := pool.Begin(ctx) if err != nil { @@ -551,7 +530,3 @@ func poolBeginFunc(ctx context.Context, pool *pgxpool.Pool, fn func(tx pgx.Tx) e } return tx.Commit(ctx) } - -// Compile guard to remind us we removed the BeginFunc method -// reference above (kept the helper for clarity). -var _ = json.Marshal From 5b4aa1a2557ba11e06b15f7609a3d72e527e3cd5 Mon Sep 17 00:00:00 2001 From: jiashuoz Date: Thu, 9 Jul 2026 12:26:19 -0700 Subject: [PATCH 2/3] refactor(jobs): hoist the copy-pasted reconciler into jobs.ReconcilePending MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stranded-row reconciler was duplicated across five packages (outboundsend, inboundprocess, hitlnotify, webhookdelivery, webhookpub fan-out) — ~240 lines with byte-identical control flow: scan `WHERE AND IS NULL LIMIT N` → per-row `FOR UPDATE` re-check → skip-if-set → EnqueueXTx → stamp the job id. Only the table/predicate/job-column/log-prefix varied. Extract a single `jobs.ReconcilePending(pool, spec, enqueueTx)` + `jobs.ReconcileSpec` + `jobs.DefaultReconcileBatch`. Each package's ReconcilePending collapses to a spec literal + one call; the domain method signatures are unchanged, so every caller (main.go cutover + each ReconcileWorker) is untouched. Behavior-identical: the two packages that stamped via a store method (inboundprocess/hitlnotify) stamped with the exact same `UPDATE SET =$2 WHERE id=$1` the helper now inlines; those store methods stay for the accept-tx path. Log output (prefixes) is preserved verbatim. The interpolated identifiers are compile-time constants (documented in ReconcileSpec) — the row id is always a bound param. Net ~240 → ~40 shared lines, and the reconcile invariants (FOR UPDATE re-check, IS NULL guard) now live in one hot-path-tested place instead of five. Each package keeps its own reconcile_test.go — all green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01DfAXBRMJ3KPuU3UEgwqjhS --- internal/hitlnotify/jobs.go | 65 +++------------- internal/inboundprocess/jobs.go | 64 ++-------------- internal/jobs/reconcile.go | 110 +++++++++++++++++++++++++++ internal/outboundsend/jobs.go | 62 ++------------- internal/webhookdelivery/jobs.go | 59 ++------------ internal/webhookpub/fanout_worker.go | 61 +++------------ 6 files changed, 148 insertions(+), 273 deletions(-) create mode 100644 internal/jobs/reconcile.go diff --git a/internal/hitlnotify/jobs.go b/internal/hitlnotify/jobs.go index 6727498b..634e9b13 100644 --- a/internal/hitlnotify/jobs.go +++ b/internal/hitlnotify/jobs.go @@ -3,7 +3,6 @@ package hitlnotify import ( "context" "errors" - "log" "sync" "github.com/jackc/pgx/v5" @@ -14,12 +13,6 @@ import ( "github.com/Mnexa-AI/e2a/internal/jobs" ) -// reconcileBatch bounds one cutover scan of pending_review rows without a -// notification job — mirrors outboundsend/inboundprocess. In steady state the set -// is ~empty (the accept-tx stamps notify_job_id atomically); this caps how many -// rows one pass re-drives if the feature was just enabled or a crash left a backlog. -const reconcileBatch = 1000 - // Jobs is the HITL-notification integration on the shared River client: a // jobs.Registrar (contributes NotifyWorker) plus the transactional enqueue entry // point the hold accept-tx calls. Both the shared client and the concrete Deliverer @@ -103,53 +96,13 @@ func (j *Jobs) EnqueueNotifyTx(ctx context.Context, tx pgx.Tx, messageID string) // notify_job_id IS NULL re-check means a re-run (or a concurrent replica) never // double-enqueues. func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) { - rows, err := pool.Query(ctx, - `SELECT id FROM messages - WHERE status='pending_review' AND notify_job_id IS NULL AND notified_at IS NULL - LIMIT $1`, reconcileBatch) - if err != nil { - return 0, err - } - var ids []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return 0, err - } - ids = append(ids, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return 0, err - } - - n := 0 - for _, id := range ids { - if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { - // Re-check under a row lock: another process (or a prior run) may have - // enqueued it already. Skip if notify_job_id is now set. - var jobID *int64 - if err := tx.QueryRow(ctx, - `SELECT notify_job_id FROM messages WHERE id=$1 FOR UPDATE`, id, - ).Scan(&jobID); err != nil { - return err - } - if jobID != nil { - return nil // already enqueued - } - newJobID, err := j.EnqueueNotifyTx(ctx, tx, id) - if err != nil { - return err - } - if err := j.store.StampNotifyJobIDTx(ctx, tx, id, newJobID); err != nil { - return err - } - n++ - return nil - }); err != nil { - log.Printf("[hitl-notify] reconcile enqueue %s: %v", id, err) - } - } - return n, nil + // The stamp is an inline UPDATE inside jobs.ReconcilePending — identical SQL to + // store.StampNotifyJobIDTx (which the accept-tx still uses). The notified_at IS NULL + // guard keeps already-emailed holds out of the reconcile set. + return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{ + Table: "messages", + JobColumn: "notify_job_id", + Where: "status='pending_review' AND notified_at IS NULL", + LogPrefix: "[hitl-notify] reconcile", + }, j.EnqueueNotifyTx) } diff --git a/internal/inboundprocess/jobs.go b/internal/inboundprocess/jobs.go index 820fc012..6aea55bc 100644 --- a/internal/inboundprocess/jobs.go +++ b/internal/inboundprocess/jobs.go @@ -3,7 +3,6 @@ package inboundprocess import ( "context" "errors" - "log" "sync" "github.com/jackc/pgx/v5" @@ -14,12 +13,6 @@ import ( "github.com/Mnexa-AI/e2a/internal/jobs" ) -// reconcileBatch bounds one cutover scan of stranded accepted intake rows — mirrors -// outboundsend.reconcileBatch. In steady state the stranded set is ~empty (the -// accept-tx stamps process_job_id atomically); this caps how many rows one pass -// re-drives if a systemic enqueue failure ever left a backlog. -const reconcileBatch = 1000 - // Jobs is the inbound-processing integration on the shared River client: a // jobs.Registrar (contributes InboundProcessWorker) plus the transactional enqueue // entry point the SMTP accept-tx calls. The shared client + the Processor are both @@ -96,55 +89,14 @@ func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob { // worker's MarkInboundIntakeFailed somehow never lands — rare, and the startup pass // re-drives NULL-job rows on the next deploy. func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) { - rows, err := pool.Query(ctx, - `SELECT id FROM inbound_intake - WHERE status='accepted' AND process_job_id IS NULL - LIMIT $1`, reconcileBatch) - if err != nil { - return 0, err - } - var ids []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return 0, err - } - ids = append(ids, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return 0, err - } - - n := 0 - for _, id := range ids { - if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { - // Re-check under a row lock: another process (or a prior run) may have - // enqueued it already. Skip if process_job_id is now set. - var jobID *int64 - if err := tx.QueryRow(ctx, - `SELECT process_job_id FROM inbound_intake WHERE id=$1 FOR UPDATE`, id, - ).Scan(&jobID); err != nil { - return err - } - if jobID != nil { - return nil // already enqueued - } - newJobID, err := j.EnqueueInboundProcessTx(ctx, tx, id) - if err != nil { - return err - } - if err := j.store.StampInboundIntakeJobIDTx(ctx, tx, id, newJobID); err != nil { - return err - } - n++ - return nil - }); err != nil { - log.Printf("[inbound-reconcile] enqueue %s: %v", id, err) - } - } - return n, nil + // The stamp is an inline UPDATE inside jobs.ReconcilePending — identical SQL to + // store.StampInboundIntakeJobIDTx (which the accept-tx still uses). + return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{ + Table: "inbound_intake", + JobColumn: "process_job_id", + Where: "status='accepted'", + LogPrefix: "[inbound-reconcile]", + }, j.EnqueueInboundProcessTx) } // EnqueueInboundProcessTx inserts the inbound_process job in the caller's accept-tx diff --git a/internal/jobs/reconcile.go b/internal/jobs/reconcile.go new file mode 100644 index 00000000..025a89ef --- /dev/null +++ b/internal/jobs/reconcile.go @@ -0,0 +1,110 @@ +package jobs + +import ( + "context" + "fmt" + "log" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// DefaultReconcileBatch bounds one reconcile scan. In steady state the stranded set is +// ~empty; under a systemic enqueue failure it caps how many rows one pass re-drives +// (one tx each) so an unhealthy River isn't amplified by fanning the whole backlog +// every tick — the remainder is picked up on the next pass. +const DefaultReconcileBatch = 1000 + +// ReconcileSpec describes one table's stranded-row reconcile. +// +// SECURITY: Table, JobColumn, and Where are COMPILE-TIME CONSTANTS supplied by the +// calling package — never runtime or user input. They are interpolated directly into +// SQL (there is no parameter form for identifiers), so callers must pass only literal +// strings. The only runtime value (the row id) is always a bound parameter. +type ReconcileSpec struct { + // Table is the row table (e.g. "messages", "webhook_events"). + Table string + // JobColumn is the nullable bigint column holding the enqueued River job id + // (e.g. "send_job_id", "fanout_job_id"). A row is "stranded" when it matches + // Where AND JobColumn IS NULL. + JobColumn string + // Where is the domain predicate identifying rows that SHOULD carry a job + // (e.g. "direction='outbound' AND delivery_status='accepted'"). ANDed with + // " IS NULL"; pass no leading/trailing "AND". + Where string + // LogPrefix tags per-row enqueue-failure logs (e.g. "[outbound-reconcile]"). + LogPrefix string + // Batch caps rows scanned per pass; 0 uses DefaultReconcileBatch. + Batch int +} + +// ReconcilePending re-drives every row matching spec.Where whose spec.JobColumn IS +// NULL: it scans up to spec.Batch ids, then per id opens a tx, re-checks the job column +// under FOR UPDATE (skipping rows another process or a prior pass already enqueued), +// calls enqueueTx to insert the River job in that tx, and stamps the returned job id +// back onto the row — all atomically. Idempotent: the FOR UPDATE + IS NULL guard means +// a re-run (or a concurrent replica) never double-enqueues. A per-row failure is logged +// and skipped (the next pass retries it); the returned count is the rows enqueued. +// +// This is the shared body behind every domain's startup cutover + live reconcile +// worker (outboundsend, inboundprocess, hitlnotify, webhookdelivery, webhookpub +// fan-out). Each domain supplies a ReconcileSpec + its own EnqueueXTx. +func ReconcilePending(ctx context.Context, pool *pgxpool.Pool, spec ReconcileSpec, + enqueueTx func(ctx context.Context, tx pgx.Tx, id string) (int64, error)) (int, error) { + + batch := spec.Batch + if batch <= 0 { + batch = DefaultReconcileBatch + } + + rows, err := pool.Query(ctx, + fmt.Sprintf(`SELECT id FROM %s WHERE %s AND %s IS NULL LIMIT $1`, + spec.Table, spec.Where, spec.JobColumn), + batch) + if err != nil { + return 0, err + } + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return 0, err + } + ids = append(ids, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return 0, err + } + + recheckSQL := fmt.Sprintf(`SELECT %s FROM %s WHERE id=$1 FOR UPDATE`, spec.JobColumn, spec.Table) + stampSQL := fmt.Sprintf(`UPDATE %s SET %s=$2 WHERE id=$1`, spec.Table, spec.JobColumn) + + n := 0 + for _, id := range ids { + if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { + // Re-check under a row lock: another process (or a prior run) may have + // enqueued it already. Skip if the job column is now set. + var jobID *int64 + if err := tx.QueryRow(ctx, recheckSQL, id).Scan(&jobID); err != nil { + return err + } + if jobID != nil { + return nil // already enqueued + } + newJobID, err := enqueueTx(ctx, tx, id) + if err != nil { + return err + } + if _, err := tx.Exec(ctx, stampSQL, id, newJobID); err != nil { + return err + } + n++ + return nil + }); err != nil { + log.Printf("%s enqueue %s: %v", spec.LogPrefix, id, err) + } + } + return n, nil +} diff --git a/internal/outboundsend/jobs.go b/internal/outboundsend/jobs.go index 5c987b54..e7b07ec6 100644 --- a/internal/outboundsend/jobs.go +++ b/internal/outboundsend/jobs.go @@ -2,7 +2,6 @@ package outboundsend import ( "context" - "log" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -11,12 +10,6 @@ import ( "github.com/Mnexa-AI/e2a/internal/jobs" ) -// reconcileBatch bounds one cutover scan of stranded accepted rows — mirrors -// webhookdelivery.reconcileBatch. In steady state the stranded set is ~empty (the -// accept-tx stamps send_job_id atomically); this caps how many rows one pass -// re-drives if a systemic enqueue failure ever stranded a backlog. -const reconcileBatch = 1000 - // Jobs is the outbound-send integration on the shared River client: a // jobs.Registrar (contributes SendWorker) plus the transactional enqueue entry // point the accept-tx calls. The shared client is injected via SetEnqueuer after @@ -62,55 +55,12 @@ func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob { // double-enqueues. Mirrors webhookdelivery.ReconcilePending. Returns the count // enqueued. func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) { - rows, err := pool.Query(ctx, - `SELECT id FROM messages - WHERE direction='outbound' AND delivery_status='accepted' AND send_job_id IS NULL - LIMIT $1`, reconcileBatch) - if err != nil { - return 0, err - } - var ids []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return 0, err - } - ids = append(ids, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return 0, err - } - - n := 0 - for _, id := range ids { - if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { - // Re-check under a row lock: another process (or a prior run) may have - // enqueued it already. Skip if send_job_id is now set. - var jobID *int64 - if err := tx.QueryRow(ctx, - `SELECT send_job_id FROM messages WHERE id=$1 FOR UPDATE`, id, - ).Scan(&jobID); err != nil { - return err - } - if jobID != nil { - return nil // already enqueued - } - newJobID, err := j.EnqueueSendTx(ctx, tx, id) - if err != nil { - return err - } - _, err = tx.Exec(ctx, `UPDATE messages SET send_job_id=$2 WHERE id=$1`, id, newJobID) - if err == nil { - n++ - } - return err - }); err != nil { - log.Printf("[outbound-reconcile] enqueue %s: %v", id, err) - } - } - return n, nil + return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{ + Table: "messages", + JobColumn: "send_job_id", + Where: "direction='outbound' AND delivery_status='accepted'", + LogPrefix: "[outbound-reconcile]", + }, j.EnqueueSendTx) } // EnqueueSendTx enqueues a send job WITHIN the caller's transaction — the outbox diff --git a/internal/webhookdelivery/jobs.go b/internal/webhookdelivery/jobs.go index d5b8e111..6b1e4c7b 100644 --- a/internal/webhookdelivery/jobs.go +++ b/internal/webhookdelivery/jobs.go @@ -20,12 +20,6 @@ import ( // re-driven within this bound rather than waiting for a process restart. const reconcileInterval = 1 * time.Minute -// reconcileBatch bounds one reconcile tick's scan. In steady state the stranded -// set is ~empty; under a systemic enqueue failure it caps how many rows one tick -// re-drives (one tx each), so an unhealthy River can't be amplified by fanning the -// whole backlog every minute. The remainder is picked up on the next tick. -const reconcileBatch = 1000 - // Jobs is the webhook-delivery integration on the shared River client: a // jobs.Registrar (contributes DeliverWorker + the reconcile periodic) plus the // transactional enqueue entry point the outbox drain + redelivery API call. The @@ -98,53 +92,12 @@ func (w *ReconcileWorker) Work(ctx context.Context, _ *river.Job[WebhookReconcil // guard means a re-run (or a concurrent replica) never double-enqueues. Returns // the number of rows enqueued. func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) { - rows, err := pool.Query(ctx, - `SELECT id FROM webhook_subscriber_deliveries WHERE status='pending' AND job_id IS NULL LIMIT $1`, reconcileBatch) - if err != nil { - return 0, err - } - var ids []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return 0, err - } - ids = append(ids, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return 0, err - } - - n := 0 - for _, id := range ids { - if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { - // Re-check under a row lock: another process (or a prior run) may have - // enqueued it already. Skip if job_id is now set. - var jobID *int64 - if err := tx.QueryRow(ctx, - `SELECT job_id FROM webhook_subscriber_deliveries WHERE id=$1 FOR UPDATE`, id, - ).Scan(&jobID); err != nil { - return err - } - if jobID != nil { - return nil // already enqueued - } - newJobID, err := j.EnqueueDeliveryTx(ctx, tx, id) - if err != nil { - return err - } - _, err = tx.Exec(ctx, `UPDATE webhook_subscriber_deliveries SET job_id=$2 WHERE id=$1`, id, newJobID) - if err == nil { - n++ - } - return err - }); err != nil { - log.Printf("[webhook-reconcile] enqueue %s: %v", id, err) - } - } - return n, nil + return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{ + Table: "webhook_subscriber_deliveries", + JobColumn: "job_id", + Where: "status='pending'", + LogPrefix: "[webhook-reconcile]", + }, j.EnqueueDeliveryTx) } // EnqueueDelivery enqueues a River delivery job for an ALREADY-INSERTED pending diff --git a/internal/webhookpub/fanout_worker.go b/internal/webhookpub/fanout_worker.go index 18b814e8..a6dd660c 100644 --- a/internal/webhookpub/fanout_worker.go +++ b/internal/webhookpub/fanout_worker.go @@ -35,12 +35,10 @@ const ( // retry forever the way the legacy pending-row poll did. maxFanOutAttempts = 10 - // fanOutReconcileInterval / fanOutReconcileBatch mirror the delivery reconciler: - // frequent + cheap (a partial index backs the status='pending' AND fanout_job_id - // IS NULL scan) and bounded per tick so a systemic enqueue failure can't be - // amplified by fanning the whole backlog every minute. + // fanOutReconcileInterval is how often the reconciler re-drives pending events with + // no fan-out job — frequent + cheap (a partial index backs the status='pending' AND + // fanout_job_id IS NULL scan). The per-pass row cap is jobs.DefaultReconcileBatch. fanOutReconcileInterval = 1 * time.Minute - fanOutReconcileBatch = 1000 ) // FanOutJobs is the webhook fan-out integration on the shared River client: a @@ -198,51 +196,10 @@ func (w *FanOutReconcileWorker) Work(ctx context.Context, _ *river.Job[FanOutRec // guard means a re-run (or a concurrent replica) never double-enqueues. Returns the // number of events enqueued. func (j *FanOutJobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) { - rows, err := pool.Query(ctx, - `SELECT id FROM webhook_events WHERE status='pending' AND fanout_job_id IS NULL LIMIT $1`, fanOutReconcileBatch) - if err != nil { - return 0, err - } - var ids []string - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - rows.Close() - return 0, err - } - ids = append(ids, id) - } - rows.Close() - if err := rows.Err(); err != nil { - return 0, err - } - - n := 0 - for _, id := range ids { - if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error { - // Re-check under a row lock: another process (or a prior run) may have - // enqueued it already. Skip if fanout_job_id is now set. - var jobID *int64 - if err := tx.QueryRow(ctx, - `SELECT fanout_job_id FROM webhook_events WHERE id=$1 FOR UPDATE`, id, - ).Scan(&jobID); err != nil { - return err - } - if jobID != nil { - return nil // already enqueued - } - newJobID, err := j.EnqueueFanOutTx(ctx, tx, id) - if err != nil { - return err - } - _, err = tx.Exec(ctx, `UPDATE webhook_events SET fanout_job_id=$2 WHERE id=$1`, id, newJobID) - if err == nil { - n++ - } - return err - }); err != nil { - log.Printf("[webhook-fanout-reconcile] enqueue %s: %v", id, err) - } - } - return n, nil + return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{ + Table: "webhook_events", + JobColumn: "fanout_job_id", + Where: "status='pending'", + LogPrefix: "[webhook-fanout-reconcile]", + }, j.EnqueueFanOutTx) } From db77b5ec2c8452ba2edf2101c57b92e395b5017d Mon Sep 17 00:00:00 2001 From: jiashuoz Date: Thu, 9 Jul 2026 12:31:54 -0700 Subject: [PATCH 3/3] chore(jobs): document Timeout choices + small consistency polish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Readability polish across the River worker packages (no behavior change), from the modularity review: - Document the deliberate reliance on River's 60s default JobTimeout on the four per-message workers that DON'T override Timeout() (SendWorker, DeliverWorker, InboundProcessWorker, NotifyWorker) + RetentionWorker — so the omission reads as a decision, not an oversight (the repo has an explicit "River JobTimeout trap" lesson). InboundProcessWorker notes the content screen is bounded by piguard's own 5s per-detector timeout; RetentionWorker notes its single unbounded DELETE is ctx-safe under a 60s cut and flags batching (like #397) as a future option. - jobs.New: pass Config to defaultQueueConfig(cfg) instead of six positional ints (removes an invisible-transposition hazard). - Have RegisterJobs call the NewMaintenanceWorker constructors (webhookdelivery, hitlworker, janitor) instead of rebuilding the struct literal by hand — the constructors exist precisely so tests build an identical worker; now RegisterJobs can't drift from them. - outboundsend: name the markFailed terminal-write retry magic numbers (terminalWriteRetries / terminalWriteBackoff); drop landed "(slice B/C/D)" internal markers from comments. - hitlworker: "the legacy publisher" → "the webhook publisher" (the wired OutboxPublisher is the current outbox path, not the retired one). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01DfAXBRMJ3KPuU3UEgwqjhS --- internal/hitlnotify/worker.go | 2 ++ internal/hitlworker/events.go | 3 ++- internal/hitlworker/maintenance.go | 2 +- internal/inboundprocess/retention.go | 6 ++++++ internal/inboundprocess/worker.go | 4 ++++ internal/janitor/janitor.go | 2 +- internal/jobs/jobs.go | 2 +- internal/jobs/queues.go | 14 +++++++------- internal/outboundsend/jobs.go | 4 ++-- internal/outboundsend/worker.go | 21 ++++++++++++++++----- internal/webhookdelivery/maintenance.go | 2 +- internal/webhookdelivery/worker.go | 2 ++ 12 files changed, 45 insertions(+), 19 deletions(-) diff --git a/internal/hitlnotify/worker.go b/internal/hitlnotify/worker.go index d8c2d18a..d2084dc8 100644 --- a/internal/hitlnotify/worker.go +++ b/internal/hitlnotify/worker.go @@ -99,6 +99,8 @@ func (w *NotifyWorker) NextRetry(job *river.Job[HITLNotifyArgs]) time.Time { return time.Now().Add(notifyRetryBackoffs[i]) } +// Work intentionally has no Timeout() override — a single SMTP submit of the approval +// notification fits River's 60s default JobTimeout. func (w *NotifyWorker) Work(ctx context.Context, job *river.Job[HITLNotifyArgs]) error { pn, err := w.store.LoadPendingNotify(ctx, job.Args.MessageID) if err != nil { diff --git a/internal/hitlworker/events.go b/internal/hitlworker/events.go index d1ff2f72..e60ff7c8 100644 --- a/internal/hitlworker/events.go +++ b/internal/hitlworker/events.go @@ -25,7 +25,8 @@ import ( // reviewed_by_user_id is omitted and `auto_resolved: true` is set; and the // routing key (Event.UserID) is the agent OWNER (the sweep is system-scoped). -// publish fires e through the legacy publisher, detached from the sweep's +// publish fires e through the webhook publisher (the OutboxPublisher, which writes the +// webhook_events outbox that a fan-out worker drains), detached from the sweep's // context so a cancelled sweep doesn't drop an already-committed resolution's // notification. No-op when no publisher is wired. The deterministic id (set by // the caller) makes a re-swept row's event idempotent. diff --git a/internal/hitlworker/maintenance.go b/internal/hitlworker/maintenance.go index 5d2656cb..890d2b98 100644 --- a/internal/hitlworker/maintenance.go +++ b/internal/hitlworker/maintenance.go @@ -105,7 +105,7 @@ func NewMaintenanceJobs(sweeper Sweeper, asyncSends bool) *MaintenanceJobs { // idempotent + cross-replica-safe so scheduling is the only concern). Implements // jobs.Registrar. func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob { - river.AddWorker(w, &MaintenanceWorker{sweeper: m.sweeper, asyncSends: m.asyncSends}) + river.AddWorker(w, NewMaintenanceWorker(m.sweeper, m.asyncSends)) return []*river.PeriodicJob{ river.NewPeriodicJob( river.PeriodicInterval(maintenanceInterval), diff --git a/internal/inboundprocess/retention.go b/internal/inboundprocess/retention.go index fc5b4e02..b75c3842 100644 --- a/internal/inboundprocess/retention.go +++ b/internal/inboundprocess/retention.go @@ -35,6 +35,12 @@ type RetentionWorker struct { pruner Pruner } +// Work intentionally has no Timeout() override — it relies on River's 60s default +// JobTimeout. PruneProcessedIntake is a single unbounded DELETE; inbound_intake is +// short-lived (processed rows pruned every tick), so the set is normally tiny. On a +// large backlog a 60s cut is safe (the DELETE is ctx-aware — it rolls back and the next +// tick resumes), just slow. TODO: batch it like the janitor's ctid-LIMIT prunes (#397) +// if a backlog ever makes the single DELETE a problem. func (w *RetentionWorker) Work(ctx context.Context, _ *river.Job[InboundRetentionArgs]) error { n, err := w.pruner.PruneProcessedIntake(ctx, retentionWindow) if err != nil { diff --git a/internal/inboundprocess/worker.go b/internal/inboundprocess/worker.go index 7cd89921..fcdebea5 100644 --- a/internal/inboundprocess/worker.go +++ b/internal/inboundprocess/worker.go @@ -93,6 +93,10 @@ func (w *InboundProcessWorker) NextRetry(job *river.Job[InboundProcessArgs]) tim return time.Now().Add(inboundRetryBackoffs[i]) } +// Work intentionally has no Timeout() override — parse/SPF-DKIM/screen/persist fits +// River's 60s default JobTimeout. The heaviest step, the content screen, is an external +// LLM call bounded by its own per-detector timeout (piguard, 5s default) that fails +// open, so it can't hang the job past 60s. func (w *InboundProcessWorker) Work(ctx context.Context, job *river.Job[InboundProcessArgs]) error { it, err := w.store.LoadInboundIntake(ctx, job.Args.IntakeID) if err != nil { diff --git a/internal/janitor/janitor.go b/internal/janitor/janitor.go index 5d638bf7..37ed5060 100644 --- a/internal/janitor/janitor.go +++ b/internal/janitor/janitor.go @@ -244,7 +244,7 @@ func NewMaintenanceJobs(j *Janitor) *MaintenanceJobs { return &MaintenanceJobs{j // unbounded-DELETE hazard is resolved — each is now a ctx-aware ctid-LIMIT drain // loop (see the DeleteExpired* store methods) bounded by MaintenanceWorker.Timeout. func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob { - river.AddWorker(w, &MaintenanceWorker{janitor: m.janitor}) + river.AddWorker(w, NewMaintenanceWorker(m.janitor)) return []*river.PeriodicJob{ river.NewPeriodicJob( river.PeriodicInterval(janitorInterval), diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 63328bd2..dd2adf9b 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -107,7 +107,7 @@ func New(pool *pgxpool.Pool, cfg Config, registrars ...Registrar) (*Client, erro } rc, err := river.NewClient(riverpgxv5.New(pool), &river.Config{ - Queues: defaultQueueConfig(cfg.OutboundWorkers, cfg.InboundWorkers, cfg.WebhookWorkers, cfg.MaintenanceWorkers, cfg.NotifyWorkers, cfg.DefaultWorkers), + Queues: defaultQueueConfig(cfg), Workers: workers, PeriodicJobs: periodic, }) diff --git a/internal/jobs/queues.go b/internal/jobs/queues.go index 14fc9bdd..8a64c45f 100644 --- a/internal/jobs/queues.go +++ b/internal/jobs/queues.go @@ -34,13 +34,13 @@ const ( // MaxWorkers is deliberately generous for I/O-bound work (SMTP / HTTP); tune // against real throughput. Every queue a domain enqueues into MUST appear here or // its jobs sit unworked. -func defaultQueueConfig(outbound, inbound, webhook, maintenance, notify, deflt int) map[string]river.QueueConfig { +func defaultQueueConfig(cfg Config) map[string]river.QueueConfig { return map[string]river.QueueConfig{ - QueueOutbound: {MaxWorkers: outbound}, - QueueInbound: {MaxWorkers: inbound}, - QueueWebhook: {MaxWorkers: webhook}, - QueueMaintenance: {MaxWorkers: maintenance}, - QueueNotify: {MaxWorkers: notify}, - QueueDefault: {MaxWorkers: deflt}, + QueueOutbound: {MaxWorkers: cfg.OutboundWorkers}, + QueueInbound: {MaxWorkers: cfg.InboundWorkers}, + QueueWebhook: {MaxWorkers: cfg.WebhookWorkers}, + QueueMaintenance: {MaxWorkers: cfg.MaintenanceWorkers}, + QueueNotify: {MaxWorkers: cfg.NotifyWorkers}, + QueueDefault: {MaxWorkers: cfg.DefaultWorkers}, } } diff --git a/internal/outboundsend/jobs.go b/internal/outboundsend/jobs.go index e7b07ec6..5d04fdd8 100644 --- a/internal/outboundsend/jobs.go +++ b/internal/outboundsend/jobs.go @@ -31,7 +31,7 @@ func (j *Jobs) SetEnqueuer(e jobs.Enqueuer) { j.enq = e } // RegisterJobs adds the SendWorker to the shared client's bundle. Implements // jobs.Registrar. // -// No live periodic reconciler yet (slice D). The one residual it would close: if a +// No live periodic reconciler yet (startup cutover only). The one residual it would close: if a // job's terminal write (markFailed) fails on all its retries, the worker still // cancels/discards the River job, leaving the row `accepted` with a stamped-but-dead // job — which ReconcilePending (keyed on send_job_id IS NULL) does not catch. That @@ -66,7 +66,7 @@ func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, e // EnqueueSendTx enqueues a send job WITHIN the caller's transaction — the outbox // pattern: the accept-tx's messages-row insert and this job commit together, so an // `accepted` message can never exist without a send job (or vice versa). The -// accept-tx stamps the returned river_job id on messages.send_job_id (slice C) so +// accept-tx stamps the returned river_job id on messages.send_job_id so // the reconciler can find stranded rows (`accepted` with no job). Mirrors // webhookdelivery.EnqueueDeliveryTx. func (j *Jobs) EnqueueSendTx(ctx context.Context, tx pgx.Tx, messageID string) (int64, error) { diff --git a/internal/outboundsend/worker.go b/internal/outboundsend/worker.go index 9815bfcb..216c35d9 100644 --- a/internal/outboundsend/worker.go +++ b/internal/outboundsend/worker.go @@ -29,7 +29,7 @@ import ( // sendRetryBackoffs is the per-attempt delay schedule for a failed outbound send — // the decided envelope (design §4). River drives it via NextRetry; indexed by -// attempt. Provider-outage errors snooze instead of counting an attempt (slice D). +// attempt. Provider-outage errors snooze instead of counting an attempt. var sendRetryBackoffs = []time.Duration{ 30 * time.Second, 2 * time.Minute, @@ -107,13 +107,13 @@ type DeliverOutcome struct { } // Deliverer performs a SINGLE SMTP submit — River owns re-attempts. Implemented in -// the binary over internal/outbound's single-attempt path (slice B). +// the binary over internal/outbound's single-attempt path. type Deliverer interface { Deliver(ctx context.Context, j *SendJob) DeliverOutcome } // Store is the messages-store surface the worker needs. Implemented over -// internal/identity in the binary (slice C). MarkSent/MarkFailed each own their own +// internal/identity in the binary. MarkSent/MarkFailed each own their own // transaction and emit email.sent / email.failed via the webhook outbox in it. type Store interface { // LoadForSend returns the send payload, or (nil, nil) if the message is gone @@ -148,6 +148,9 @@ func (w *SendWorker) NextRetry(job *river.Job[OutboundSendArgs]) time.Time { return time.Now().Add(sendRetryBackoffs[i]) } +// Work intentionally has no Timeout() override — a single SES submit comfortably fits +// River's 60s default JobTimeout. (Contrast the maintenance/sweep workers, which +// override it because they can run for minutes.) func (w *SendWorker) Work(ctx context.Context, job *river.Job[OutboundSendArgs]) error { j, err := w.store.LoadForSend(ctx, job.Args.MessageID) if err != nil { @@ -195,16 +198,24 @@ func (w *SendWorker) Work(ctx context.Context, job *river.Job[OutboundSendArgs]) // markFailed writes the terminal 'failed' status, retrying a transient DB error a // few times so the row's status never desyncs from a discarded River job (mirrors // webhookdelivery.markFailedReliably). +const ( + // terminalWriteRetries / terminalWriteBackoff bound the retry of the terminal + // 'failed' write in markFailed — a best-effort last resort when the DB write of the + // final send outcome itself fails. Backoff is linear (i+1)×base. + terminalWriteRetries = 3 + terminalWriteBackoff = 150 * time.Millisecond +) + func (w *SendWorker) markFailed(ctx context.Context, messageID string, attempt int, detail string) { var err error - for i := 0; i < 3; i++ { + for i := 0; i < terminalWriteRetries; i++ { if err = w.store.MarkFailed(ctx, messageID, attempt, detail); err == nil { return } select { case <-ctx.Done(): return - case <-time.After(time.Duration(i+1) * 150 * time.Millisecond): + case <-time.After(time.Duration(i+1) * terminalWriteBackoff): } } log.Printf("[outbound-send] CRITICAL: terminal 'failed' write for %s failed after retries: %v", messageID, err) diff --git a/internal/webhookdelivery/maintenance.go b/internal/webhookdelivery/maintenance.go index 80a481eb..7381d784 100644 --- a/internal/webhookdelivery/maintenance.go +++ b/internal/webhookdelivery/maintenance.go @@ -60,7 +60,7 @@ func NewMaintenanceJobs(sweeper Sweeper) *MaintenanceJobs { return &MaintenanceJ // periodic scheduler already inserts at most one per interval and a completed run // must not dedup-block the next), RunOnStart:false (first sweep after one interval). func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob { - river.AddWorker(w, &MaintenanceWorker{sweeper: m.sweeper}) + river.AddWorker(w, NewMaintenanceWorker(m.sweeper)) return []*river.PeriodicJob{ river.NewPeriodicJob( river.PeriodicInterval(maintenanceInterval), diff --git a/internal/webhookdelivery/worker.go b/internal/webhookdelivery/worker.go index 093b9b1f..ddbff2b8 100644 --- a/internal/webhookdelivery/worker.go +++ b/internal/webhookdelivery/worker.go @@ -91,6 +91,8 @@ func (w *DeliverWorker) NextRetry(job *river.Job[WebhookDeliverArgs]) time.Time return time.Now().Add(retryBackoffs[i]) } +// Work intentionally has no Timeout() override — a single HTTP POST (bounded by the +// deliverer's own client timeout) fits River's 60s default JobTimeout. func (w *DeliverWorker) Work(ctx context.Context, job *river.Job[WebhookDeliverArgs]) error { d, err := w.subStore.GetSubscriberDeliveryByID(ctx, job.Args.DeliveryID) if errors.Is(err, pgx.ErrNoRows) {