fix(email): worker email subsystem BugBash 2026-05-19 — P0/P1/P2 fixes

P0-1 — Brevo auth error silently dropped all email. brevo_provider.go
classified a 401/403 (bad/expired/revoked API key) and 429 (rate limit)
as Permanent, so the forwarder advanced the cursor past every audit row
in every batch — total, silent, unrecoverable email loss. Reclassify
401/403/429 as Transient (cursor held, recoverable) and emit an
alert-able email.brevo.auth_wall ERROR. 400/422 payload rejects stay
Permanent. Test: TestBrevoProvider_AuthFailureIsTransient.

P1-2 — cursor had no time floor. A Redis wipe reset the cursor to zero
and fetchBatch (no age bound) re-sent the entire audit_log history. Add
a 48h age floor to fetchBatch and seed a missing cursor to now()-grace
instead of the zero value. Tests: TestEventForwarder_FetchBatch_HasAgeFloor,
TestEventForwarder_MissingCursor_SeedsToNow.

P1-3 — no send ledger. Idempotency rested on the Brevo X-Mailin-Custom
header, which is not a delivery-dedup mechanism. Add a forwarder_sent
ledger table (migration 055) claimed ON CONFLICT DO NOTHING before each
send; a duplicate audit_id is skipped. A Transient send releases the
claim so the retry can re-send. Tests: TestEventForwarder_Ledger_SendsOnce,
TestEventForwarder_TransientReleasesLedger.

Log noise (P1, reported symptom) — idle-tick INFO demoted to DEBUG:
event_email_forwarder.no_new_rows, middleware.work_ok, and the
zero-candidate *.completed lines in checkout_reconcile / expiry_reminder /
payment_grace_reminder / payment_grace_terminator / deployment_reminder.
INFO now only fires on non-zero work. Test: TestEventForwarder_IdleTick_NoInfoLog.

P2-1 — builder_skipped_row demoted WARN -> INFO (no owner email is an
expected state for deleted/orphan/test teams). Cursor-advance unchanged.
Test: TestEventForwarder_BuilderSkippedRow_NotWarn.

F5 — suppression bypass. The forwarder checked suppression against
row.OwnerEmail but sent to resolveRecipient(row); for anon-tier
metadata.email recipients the check ran against "" and was bypassed.
Resolve the recipient once and check suppression against the same
address that is emailed. Test: TestEventForwarder_SuppressionUsesSentRecipient.

F4 — silent no-op. A kind in eventEmailBuilders with no renderer fell
through the dead Brevo-template path -> SkippedNoTemplate -> cursor
advanced, zero email, zero error. A missing renderer is now a loud ERROR
and holds the cursor. Test: TestEventForwarder_MissingRenderer_LoudErrorNoAdvance.

F3 — deploy-reminder spam. deployment_reminder fired 6 identical emails
over 12h. Reduced to a 3-stage escalating cadence (maxDeployReminders=3)
with a reminder_index-keyed subject prefix (Heads up / Reminder / Final
reminder), matching anon.expiry_warning. Test: TestDeployExpiringSoon_EscalatingCadence.

Migration 055_forwarder_sent.sql added to worker/sql (canonical) and
mirrored into api/internal/db/migrations + api testhelpers schema mirror
(separate commit in the api repo).

go build ./... && go vet ./... && go test ./... -count=1 — all green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mastermanas805

internal/email/brevo_provider.go

@@ -20,13 +20,20 @@ package email
 // forwarder doesn't need to know the wire format:
 //
 //   2xx                              → nil                                            (success — forwarder advances cursor)
-//   4xx (401 / 422 / 400 etc.)       → *SendError{Class: SendClassPermanent}          (forwarder advances + logs ERROR)
+//   401 / 403 (auth) / 429 (rate)    → *SendError{Class: SendClassTransient}          (forwarder HOLDS cursor — recoverable)
+//   4xx (400 / 422 payload reject)   → *SendError{Class: SendClassPermanent}          (forwarder advances + logs ERROR)
 //   5xx / network / timeout          → *SendError{Class: SendClassTransient}          (forwarder holds cursor)
 //   no template configured for kind  → *SendError{Class: SendClassSkippedNoTemplate}  (forwarder advances silently)
 //
-// The "4xx-advances" behaviour is deliberate: a single audit row with
-// malformed content shouldn't block every event behind it forever. We
-// log loudly so the poisoned row is visible in the structured-log stream.
+// The "payload-4xx-advances" behaviour is deliberate: a single audit row
+// with malformed content (400/422) shouldn't block every event behind it
+// forever. We log loudly so the poisoned row is visible in the log stream.
+//
+// P0-1 (2026-05-19): 401/403/429 are NOT advanced. A bad/expired/revoked
+// API key (401/403) or a rate-limit (429) is an ACCOUNT-level condition,
+// recoverable without operator data loss — classifying it Permanent made
+// the forwarder silently drop every audit row in every batch. These now
+// map to Transient (cursor held) and log the alert-able auth_wall ERROR.
 
 import (
 	"bytes"
@@ -376,11 +383,53 @@ func (p *BrevoProvider) doRequest(ctx context.Context, evt EventEmail, body []by
 		)
 		return nil
 
+	case resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden:
+		// P0-1 FIX (2026-05-19): 401/403 is an ACCOUNT-LEVEL auth failure
+		// (bad / expired / revoked BREVO_API_KEY) — NOT a per-row payload
+		// problem. It is transient from the queue's point of view: it
+		// resolves the instant an operator rotates the secret. Classifying
+		// it Permanent made the forwarder advance the cursor past every
+		// row in every batch, silently and unrecoverably dropping all
+		// email. Transient holds the cursor so nothing is lost and the
+		// backlog drains once the key is fixed.
+		//
+		// This is logged at ERROR with a distinct, alert-able key
+		// (auth_wall) so an operator is paged before a batch is burned —
+		// the generic email.brevo.permanent_4xx key is for per-row
+		// payload rejects, not account auth failure.
+		slog.Error("email.brevo.auth_wall",
+			"kind", evt.Kind,
+			"recipient", evt.Recipient,
+			"status", resp.StatusCode,
+			"path", string(path),
+			"body", string(respBody),
+			"note", "account-level Brevo auth failure (bad/expired/revoked BREVO_API_KEY) — classified Transient, cursor held, email NOT lost; rotate the secret",
+		)
+		return &SendError{
+			Class:   SendClassTransient,
+			Message: fmt.Sprintf("brevo: auth failure %d %s", resp.StatusCode, string(respBody)),
+		}
+
+	case resp.StatusCode == http.StatusTooManyRequests:
+		// P0-1 FIX (2026-05-19): 429 rate-limited is also recoverable —
+		// the row is fine, Brevo just wants us to back off. Transient so
+		// the cursor holds and the row retries next tick.
+		slog.Warn("email.brevo.rate_limited",
+			"kind", evt.Kind,
+			"recipient", evt.Recipient,
+			"status", resp.StatusCode,
+			"path", string(path),
+			"body", string(respBody),
+		)
+		return &SendError{
+			Class:   SendClassTransient,
+			Message: fmt.Sprintf("brevo: rate limited %d %s", resp.StatusCode, string(respBody)),
+		}
+
 	case resp.StatusCode >= 400 && resp.StatusCode < 500:
-		// 401/403 = bad API key (every row will fail the same way until
-		// someone rotates the secret), 400/422 = bad payload for this row.
-		// In both cases advancing the cursor is correct: holding it pins
-		// the whole queue on one bad row.
+		// 400/422 and other 4xx = genuine per-row payload rejection. The
+		// row will never produce a valid send, so advancing the cursor is
+		// correct: holding it pins the whole queue on one bad row.
 		slog.Error("email.brevo.permanent_4xx",
 			"kind", evt.Kind,
 			"recipient", evt.Recipient,

internal/email/brevo_provider_test.go

@@ -131,9 +131,12 @@ func TestBrevoProvider_2xxReturnsNil(t *testing.T) {
 }
 
 // TestBrevoProvider_4xxReturnsPermanent verifies the "advance past poisoned
-// row" contract: a 401 (or any 4xx) → *SendError{Class: SendClassPermanent}.
+// row" contract: a genuine per-row payload reject (400/422) →
+// *SendError{Class: SendClassPermanent}. NOTE 401/403/429 are deliberately
+// EXCLUDED here — they are account-level/recoverable and now classify
+// Transient; see TestBrevoProvider_AuthFailureIsTransient (P0-1 regression).
 func TestBrevoProvider_4xxReturnsPermanent(t *testing.T) {
-	for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusUnprocessableEntity} {
+	for _, status := range []int{http.StatusBadRequest, http.StatusUnprocessableEntity, http.StatusNotFound, http.StatusConflict} {
 		srv, _ := fakeBrevo(t, status)
 		p := newTestProvider(t, srv, nil)
 
@@ -151,7 +154,40 @@ func TestBrevoProvider_4xxReturnsPermanent(t *testing.T) {
 			continue
 		}
 		if se.Class != SendClassPermanent {
-			t.Errorf("SendEvent on %d → Class=%v; want SendClassPermanent — holding cursor on auth errors pins the queue forever", status, se.Class)
+			t.Errorf("SendEvent on %d → Class=%v; want SendClassPermanent — a payload reject must advance past the poisoned row", status, se.Class)
+		}
+	}
+}
+
+// TestBrevoProvider_AuthFailureIsTransient is the P0-1 regression test.
+//
+// BUG: a bad/expired/revoked BREVO_API_KEY returns 401/403. The provider
+// classified that Permanent → the forwarder advanced the cursor → every
+// audit row in every batch was silently, unrecoverably dropped. A 429
+// rate-limit had the same fate. This test FAILS against the pre-fix code
+// (which returned SendClassPermanent for 401/403/429) and PASSES only
+// when those statuses are classified Transient so the cursor is held and
+// the email is recoverable once the operator rotates the key / backs off.
+func TestBrevoProvider_AuthFailureIsTransient(t *testing.T) {
+	for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden, http.StatusTooManyRequests} {
+		srv, _ := fakeBrevo(t, status)
+		p := newTestProvider(t, srv, nil)
+
+		err := p.SendEvent(context.Background(), EventEmail{
+			Kind:      "subscription.upgraded",
+			Recipient: "x@example.com",
+		})
+		if err == nil {
+			t.Errorf("SendEvent on %d = nil; want SendError(Transient)", status)
+			continue
+		}
+		var se *SendError
+		if !errors.As(err, &se) {
+			t.Errorf("SendEvent on %d returned %T; want *SendError", status, err)
+			continue
+		}
+		if se.Class != SendClassTransient {
+			t.Errorf("SendEvent on %d → Class=%v; want SendClassTransient — auth/rate failure must HOLD the cursor, not advance and drop email silently", status, se.Class)
 		}
 	}
 }

internal/jobs/checkout_reconcile.go

@@ -175,7 +175,11 @@ func (w *CheckoutReconcileWorker) Work(ctx context.Context, job *river.Job[Check
 	rows.Close()
 
 	if len(candidates) == 0 {
-		slog.Info("jobs.checkout_reconcile.completed",
+		// P1-1 (BugBash 2026-05-19): idle tick — zero candidates carries
+		// no operational signal. Demoted INFO → DEBUG; liveness is
+		// covered by jobs.middleware.work_ok. INFO is reserved for a
+		// tick that actually did work.
+		slog.Debug("jobs.checkout_reconcile.completed",
 			"emailed", 0,
 			"resolved_late", 0,
 			"skipped", 0,

internal/jobs/deployment_reminder.go

@@ -7,9 +7,15 @@ package jobs
 //   - ttl_policy != 'permanent'
 //   - status NOT IN ('deleted', 'expired')
 //   - expires_at falls within the next 12h
-//   - reminders_sent < 6
+//   - reminders_sent < maxDeployReminders (3)
 //   - last_reminder_at IS NULL OR last_reminder_at < now() - 2h  (cooldown)
 //
+// F3 (BugBash 2026-05-19): the cadence used to be SIX identical emails
+// over the final 12h (T+12/14/16/18/20/22h) — read as spam. It is now a
+// 3-stage escalating cadence ("Heads up" → "Reminder" → "Final reminder")
+// matching anon.expiry_warning's shape. See maxDeployReminders and the
+// reminder_index-keyed subject in renderDeployExpiringSoon.
+//
 // For each candidate, CAS-advance reminders_sent (so two ticks don't fire
 // the same reminder twice), then write a deploy.expiring_soon audit_log
 // row carrying every param the email template needs. The BrevoForwarder
@@ -30,8 +36,9 @@ package jobs
 //     below.
 //
 // Cadence in practice: a deploy that lands at T0 with auto_24h TTL fires
-// reminders at T+12h, T+14h, T+16h, T+18h, T+20h, T+22h — six emails over
-// the final 12h.
+// 3 reminders in the final 12h window — roughly T+12h ("Heads up"),
+// T+14h ("Reminder"), T+16h ("Final reminder") — with the 2h cooldown
+// gating the gap. Three escalating emails, not six identical ones.
 //
 // Audit kind: deploy.expiring_soon. Metadata: {deploy_id, team_id,
 // reminder_index, hours_remaining, expires_at, app_id, deploy_url,
@@ -53,6 +60,12 @@ import (
 	"instant.dev/worker/internal/metrics"
 )
 
+// maxDeployReminders caps how many deploy-expiry reminders fire per
+// deployment (F3, BugBash 2026-05-19). 3 stages — "Heads up" / "Reminder"
+// / "Final reminder" — matching the anon.expiry_warning escalating
+// cadence. Was 6, which produced six identical emails over 12h.
+const maxDeployReminders = 3
+
 // DeploymentReminderArgs is the River job payload (no fields — runs as a sweep).
 type DeploymentReminderArgs struct{}
 
@@ -128,11 +141,11 @@ func (w *DeploymentReminderWorker) Work(ctx context.Context, job *river.Job[Depl
 		  AND d.status NOT IN ('deleted', 'expired')
 		  AND d.expires_at > $1
 		  AND d.expires_at <= $2
-		  AND d.reminders_sent < 6
+		  AND d.reminders_sent < $4
 		  AND (d.last_reminder_at IS NULL OR d.last_reminder_at <= $3)
 		ORDER BY d.expires_at ASC
 		LIMIT 500
-	`, now, windowEnd, cooldownBefore)
+	`, now, windowEnd, cooldownBefore, maxDeployReminders)
 	if err != nil {
 		return fmt.Errorf("DeploymentReminderWorker: query failed: %w", err)
 	}
@@ -160,7 +173,10 @@ func (w *DeploymentReminderWorker) Work(ctx context.Context, job *river.Job[Depl
 	w.sampleTTLGauge(ctx)
 
 	if len(candidates) == 0 {
-		slog.Info("jobs.deployment_reminder.completed",
+		// P1-1 (BugBash 2026-05-19): idle tick — demoted INFO → DEBUG.
+		// deployment_reminder runs every 60s; an idle INFO every minute
+		// is heartbeat noise. Liveness via jobs.middleware.work_ok.
+		slog.Debug("jobs.deployment_reminder.completed",
 			"sent", 0,
 			"candidates", 0,
 			"duration_ms", time.Since(start).Milliseconds(),
@@ -284,9 +300,9 @@ func advanceReminderCAS(ctx context.Context, db *sql.DB, deployIDStr string, exp
 		    last_reminder_at = now()
 		WHERE id = $1
 		  AND reminders_sent = $2
-		  AND reminders_sent < 6
+		  AND reminders_sent < $4
 		  AND (last_reminder_at IS NULL OR last_reminder_at <= $3)
-	`, deployIDStr, expectedRemindersSent, cooldownBefore)
+	`, deployIDStr, expectedRemindersSent, cooldownBefore, maxDeployReminders)
 	if err != nil {
 		return false, err
 	}

internal/jobs/deployment_reminder_test.go

@@ -93,7 +93,7 @@ func TestDeploymentReminderWorker_WritesAuditWithFullMetadata(t *testing.T) {
 
 	// CAS UPDATE.
 	mock.ExpectExec(`UPDATE deployments\s+SET reminders_sent`).
-		WithArgs("deploy-1", 2, sqlmock.AnyArg()).
+		WithArgs("deploy-1", 2, sqlmock.AnyArg(), 3 /* maxDeployReminders */).
 		WillReturnResult(sqlmock.NewResult(0, 1))
 
 	// Audit INSERT — the BrevoForwarder picks this up on its next tick.
@@ -153,7 +153,7 @@ func TestDeploymentReminderWorker_CASRaceLostNoAudit(t *testing.T) {
 
 	// CAS returns rowsAffected=0 — another tick won.
 	mock.ExpectExec(`UPDATE deployments\s+SET reminders_sent`).
-		WithArgs("deploy-race", 1, sqlmock.AnyArg()).
+		WithArgs("deploy-race", 1, sqlmock.AnyArg(), 3 /* maxDeployReminders */).
 		WillReturnResult(sqlmock.NewResult(0, 0))
 
 	// NO INSERT INTO audit_log expected here.