diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index de342a99d..5b838866c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -289,8 +289,15 @@ jobs: env: WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable run: | + set -o pipefail go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement|WarmupFilingAuthority|WarmupTimeWindow|WarmupInvalidWindow)' -count=1 - go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1 + go test -race -v ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|UnknownRetirement|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1 | tee warmup-recovery.log + if grep -q -- '--- SKIP: TestLiveWarmupUnknownRetirement' warmup-recovery.log; then exit 1; fi + for test in TestLiveWarmupUnknownRetirementPreservesEvidenceAndFencesReplayOnce TestLiveWarmupUnknownRetirementAllowsDistinctNewSendAndKeepsCooldown TestLiveWarmupUnknownRetirementAndDefinitiveResultSerialize TestLiveWarmupUnknownRetirementSkipsProtectedRowsWithoutStarvingLaterMailboxes TestLiveWarmupUnknownRetirementReconcilerSeedsOnceAndClearsActiveWarnings; do + grep -q -- "--- PASS: $test " warmup-recovery.log || exit 1 + done + go test -race -v ./internal/infrastructure/db -run '^TestLiveWarmupUnknownRetirementUpgrade' -count=1 | tee warmup-retirement-upgrade.log + grep -q -- '--- PASS: TestLiveWarmupUnknownRetirementUpgrade ' warmup-retirement-upgrade.log go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1 - name: Warmup database query-pressure regressions (PostgreSQL) diff --git a/docs/content/docs/development/send-safety.mdx b/docs/content/docs/development/send-safety.mdx index bc74f89a0..0c11a51e6 100644 --- a/docs/content/docs/development/send-safety.mdx +++ b/docs/content/docs/development/send-safety.mdx @@ -72,6 +72,18 @@ Authentication, permanent and conflicting-result holds do not clear merely becau ### Resolving an unknown send +#### Automatic warmup retirement + +Warmup-only unknown sends have an automatic alternative to confirming the old outcome: permanently retire the individual task without replaying it, then admit new, distinct tasks. After ten minutes, the warmup scheduler checks for an existing execution authorization and attempt, no definitive recorded result or receipt, and an exact `unknown` hold for that task. It rechecks under scheduling, task and mailbox send locks. Unstarted dispatches use their separate safe cancellation and refund path first. + +Retirement records `tasks.warmup_send_retired_at`, cancels execution permanently, preserves `send_result_state='unknown'`, and keeps all original reservation, charge, message ID, nonce, worker and result evidence. It neither refunds nor reconstructs charges, marks the message sent or failed, nor retries it. Late results cannot apply accounting, reauthorize the retired task or change a later task's hold. Other unresolved sends and conflict, authentication or permanent holds remain restrictive. The normal scheduler respects all existing warmup eligibility, cooldown and volume limits. + +This recovery includes eligible historical holds after upgrade. Legacy sends without execution fencing, messages with confirmed receipt/provider evidence, and tasks with a definitive result or operator confirmation awaiting reconciliation are not retired. Campaign and other non-warmup sends keep the provider-confirmation-only policy below. No additional customer action or configuration is required for eligible warmup sends. + +Apply migration 315 before running the updated backend, consumer and scheduler. Upgrade all control-plane replicas before relying on the new policy. An old replica may temporarily reassert a retired task's unknown hold; the updated scheduler repairs that hold without replaying the task or touching a newer restrictive hold. Whole-instance backups and organization send-history exports retain the retirement marker. The down migration refuses to remove recorded retirements; retain the schema and roll forward rather than discarding the replay fence. + +#### Provider-confirmed resolution + Unknown means the provider outcome is unproven, not that sending failed. Neither an expired timeout nor the absence of a synced Sent copy is proof that nothing was sent. Obtain explicit provider evidence for the exact task before using the mailbox PATCH: ```json @@ -102,7 +114,7 @@ Campaign routing, placement, day plans and estimates share durable mailbox send Provider execution gets 120 seconds, which covers the SMTP client's own 90-second budget for a slow server's verdict after the message, so a send is not cut off waiting for an answer the server was about to give. After provider execution, workers record and publish the result on an independent 45-second context with at most four attempts, each bounded to 10 seconds. Failed reports retain the exact outcome in memory for command redelivery without another provider send. Persisted results survive worker replacement and lost broker events: the consumer replays them through the same idempotent reconciliation. A worker crash before durable recording can still leave an unknown outcome requiring investigation. -Already-failed tasks with no outbound reservation or execution authorization cannot create unknown provider holds. The recorded-result sweep also reconciles existing invalid holds for such tasks. Real reserved or started unknown sends remain held; upgrades never infer failure from elapsed time. Deploy the backend/consumer (including the additive migration) before updated workers. No new configuration is required; existing genuine unknown holds need definitive worker evidence or the explicit operator confirmation above. +Already-failed tasks with no outbound reservation or execution authorization cannot create unknown provider holds. The recorded-result sweep also reconciles existing invalid holds for such tasks. Real reserved or started unknown sends remain held until a definitive result is reconciled or an eligible warmup task is permanently retired; upgrades never infer failure from elapsed time. Deploy the backend/consumer (including the additive migration) before updated workers. No new configuration is required. Non-warmup and ineligible unknown holds need definitive worker evidence or the explicit operator confirmation above. ## Actual-message authentication diff --git a/docs/content/docs/guides/mailboxes.mdx b/docs/content/docs/guides/mailboxes.mdx index 88c596a61..2cbe654a1 100644 --- a/docs/content/docs/guides/mailboxes.mdx +++ b/docs/content/docs/guides/mailboxes.mdx @@ -596,10 +596,12 @@ Repeated reports of the same error code count once toward the health score, at t A later completed sync automatically clears earlier temporary sync errors. A provider-confirmed send clears earlier temporary sending errors, including provider throttling and quota warnings. Recovered reports no longer appear under **Needs attention** or reduce the health score. Older reports that do not identify which operation failed clear once both sending and sync have succeeded after the report; existing reports with that evidence are cleared on upgrade too. A newer failure remains visible, and simply waiting or completing a queued task does not count as recovery. Recoverable connection failures retry automatically while the mailbox is on; a connection warning alone is not a reason to reconnect. Authentication, credential and policy errors retain their own instructions, and a switched-off mailbox must be switched back on before it can retry. -A send safety hold is different from a retryable connection failure. If Warmbly cannot confirm whether the provider accepted a message, sending pauses to avoid duplicates until the outcome is confirmed. The mailbox is shown as needing attention even without a separate connection warning, and its health details explain the hold. A successful inbox sync or the passage of time does not release it. See [send safety](/development/send-safety/) for provider-confirmed resolution. +A send safety hold is different from a retryable connection failure. If Warmbly cannot confirm whether the provider accepted a message, sending pauses to avoid retrying a message that might already have been sent. The mailbox is shown as needing attention even without a separate connection warning, and its health details explain the hold. A successful inbox sync does not confirm a send. Campaign and other non-warmup sends still require provider-confirmed resolution. See [send safety](/development/send-safety/). Warmup scheduling automatically retires abandoned preparation tasks after ten minutes only when they have no generated message ID, verification token, reservation, dispatch or execution evidence. The normal scheduler can then continue with a new task. Tasks that might have sent a message remain protected; they are not retried just because they are old. +For an authorized warmup send whose outcome remains unknown after ten minutes, Warmbly can permanently retire that task and resume with a new, distinct message. The old message is never retried or counted as confirmed success or failure; its original charge and send evidence are kept. This recovery runs automatically, including for existing eligible holds. It does not clear sign-in failures, provider restrictions, conflicting results or another unresolved send, and it never overrides a confirmed receipt or a definitive result waiting to be applied. New sending still follows your schedule, daily limits and cooldowns. + The **Inbox** column is the share of the mailbox's warmup mail that reached partners' inboxes at Google, Microsoft and Yahoo over the last seven days (smaller mail hosts run their own filters and are never counted), and the score is capped at it, so a mailbox landing some mail in spam no longer reads `100`. Click it, or open the drawer's **Deliverability** tab, for the day-by-day history and the split per provider. See [seeing where warmup lands](/guides/warmup/#seeing-where-warmup-lands). The detail drawer runs a live `SPF`, `DKIM`, and `DMARC` check on your sending domain. Confirm it is green before sending; major providers require alignment for bulk and cold mail. A domain that stays Failing eventually stops cold sending and warmup from every mailbox on it, so fix the records and press **Check** to clear it straight away; for anyone who can manage mailboxes, the check records its verdict. A lookup DNS did not answer shows the records as Not verified and is never held against the domain. DKIM reads **Not verified** rather than Missing when no key answers, because its selector is not something DNS can be asked for; that never gates sending. See [Deliverability](/guides/deliverability/) for the grace period and exactly what is stopped. diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index d9d82ad10..fa0bcb0ec 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -58,6 +58,8 @@ A workspace archive is a single file holding everything one workspace owns. It e Diagnostic participation, shared/rolling limits, restrictive send evidence and audited repair-resolution history travel with their registered mailbox/task/organization data. Source-worker executor handles reset; exact provider-message DKIM authorization/nonces are excluded. Portable minimal receipt proof remains historical evidence, never an execution grant. The reconciliation webhook/notification outbox is source-instance-local and excluded to prevent replay on import; drain pending effects before moving. Whole-instance database backups retain it. See [send-safety retention and rollback](/development/send-safety/#upgrade-rollback-and-portability). +Permanent unknown-warmup retirement markers travel with send history. Imported retired tasks remain cancelled with their outcome recorded as unknown; they never authorize a retry. Use a destination with the compatible retirement schema and recovery code. + Campaign task replay intent resets to unverified on workspace import. The destination cannot replay an imported dead letter as a verified non-sending wakeup. A contact or sequence binding remains restrictive send intent even if that record is later removed. Whole-instance backups retain the durable intent; migration rollback refuses to discard retained send intent or unverified pending campaign dead letters. diff --git a/internal/app/analytics/account_errors_test.go b/internal/app/analytics/account_errors_test.go index ac31e1c53..e0e1b1354 100644 --- a/internal/app/analytics/account_errors_test.go +++ b/internal/app/analytics/account_errors_test.go @@ -53,6 +53,22 @@ func TestSendSafetyHoldCannotAppearHealthyOrPromiseAutomaticSendRetries(t *testi } } +func TestUnknownWarmupHoldExplainsAutomaticRetirementWithoutPromisingResend(t *testing.T) { + id := uuid.New() + s := &analyticsService{} + status := s.assembleAccountStatus(t.Context(), &models.Email{ID: id, Status: "active", Provider: "smtp_imap"}, + &models.AccountDailyUsage{}, []models.AccountError{{ErrorCode: string(errx.MailErrorCodeServerUnreachable), Operation: models.MailTransportSend}}, + nil, false, nil, false, &statusReads{sendHolds: map[uuid.UUID]repository.SendRecoveryHold{id: {Reason: "unknown", Warmup: true}}}) + if status.Health.Status != "error" || status.Health.Score > 50 { + t.Fatalf("recovery has not completed: %+v", status.Health) + } + for _, message := range []string{status.Health.Issues[len(status.Health.Issues)-1], *status.Errors[0].ActionRequired} { + if !strings.Contains(message, "automatically") || !strings.Contains(message, "new sends") || !strings.Contains(message, "never retried") || strings.Contains(message, "Provider confirmation is required") { + t.Fatalf("incorrect warmup recovery guidance: %s", message) + } + } +} + type statusSendHoldReader struct { repository.EmailRepository calls int diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 68e87296d..6984303f0 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -450,6 +450,9 @@ func (s *analyticsService) assembleAccountStatus(ctx context.Context, email *mod func sendRecoveryHoldMessage(hold repository.SendRecoveryHold) string { switch hold.Reason { case "unknown": + if hold.Warmup { + return "Sending is paused after an unconfirmed warmup send. Warmbly checks it automatically and, when safe, retires that message so new sends can resume. The uncertain message is never retried." + } return "Sending is paused because the previous send outcome is unconfirmed. Warmbly will not retry it automatically, to avoid duplicate mail. Provider confirmation is required to resolve this hold; a successful inbox sync does not confirm the send." case "authentication": return "Sending is paused after an authentication failure. Repair the sending credentials and confirm outbound recovery before releasing this hold." diff --git a/internal/app/orgtransfer/send_recovery_live_test.go b/internal/app/orgtransfer/send_recovery_live_test.go index b87bfa1bd..5d99e691e 100644 --- a/internal/app/orgtransfer/send_recovery_live_test.go +++ b/internal/app/orgtransfer/send_recovery_live_test.go @@ -59,6 +59,11 @@ func TestLiveArchiveSendRecoveryLocalOriginResets(t *testing.T) { t.Fatal(err) } } + retiredTask := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,warmup_send_retired_at) + VALUES($1,$2,'warmup','cancelled','','unknown',NOW())`, retiredTask, mailbox); err != nil { + t.Fatal(err) + } if _, err := pool.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, mailbox, heldTask); err != nil { t.Fatal(err) } @@ -102,9 +107,17 @@ func TestLiveArchiveSendRecoveryLocalOriginResets(t *testing.T) { if _, err := pool.Exec(ctx, `DELETE FROM send_recovery_resolutions WHERE organization_id=$1`, f.org); err != nil { t.Fatal(err) } + if _, err := pool.Exec(ctx, `DELETE FROM tasks WHERE id=$1`, retiredTask); err != nil { + t.Fatal(err) + } if _, err := svc.ImportFrom(ctx, f.org, bytes.NewReader(archive.Bytes()), ImportOptions{Groups: groups, Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.owner}, nil); err != nil { t.Fatal(err) } + var retirementRetained bool + if err := pool.QueryRow(ctx, `SELECT warmup_send_retired_at IS NOT NULL AND status='cancelled' AND send_result_state='unknown' + AND send_result_applied_at IS NULL AND send_released_at IS NULL AND send_executor_nonce IS NULL FROM tasks WHERE id=$1`, retiredTask).Scan(&retirementRetained); err != nil || !retirementRetained { + t.Fatalf("retirement replay fence lost on archive roundtrip: retained=%t err=%v", retirementRetained, err) + } var count, reset, confirmations, automatic int if err := pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE recorded_locally_at IS NULL AND conflict_resolved_at IS NULL AND conflict_detected_at='2099-01-01Z'), count(*) FILTER(WHERE confirmed_result IS NOT NULL AND confirmation_reference='Source confirmation fixture'), diff --git a/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.down.sql b/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.down.sql new file mode 100644 index 000000000..11e6a7124 --- /dev/null +++ b/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.down.sql @@ -0,0 +1,10 @@ +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM tasks WHERE warmup_send_retired_at IS NOT NULL) THEN + RAISE EXCEPTION 'cannot remove permanent warmup send retirement evidence; retain the schema and roll forward'; + END IF; +END $$; + +DROP INDEX tasks_retirable_unknown_warmup; +ALTER TABLE tasks DROP CONSTRAINT tasks_retired_warmup_send_check, + DROP COLUMN warmup_send_retired_at; diff --git a/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.up.sql b/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.up.sql new file mode 100644 index 000000000..67714f3a5 --- /dev/null +++ b/internal/infrastructure/db/migrations/000315_warmup_unknown_send_retirement.up.sql @@ -0,0 +1,10 @@ +ALTER TABLE tasks ADD COLUMN warmup_send_retired_at timestamptz, + ADD CONSTRAINT tasks_retired_warmup_send_check CHECK ( + warmup_send_retired_at IS NULL OR ( + task_type='warmup' AND status='cancelled' + AND send_result_state IS NOT DISTINCT FROM 'unknown' + AND send_result_applied_at IS NULL AND send_released_at IS NULL)); + +CREATE INDEX tasks_retirable_unknown_warmup ON tasks (COALESCE(send_reserved_at,completed_at,updated_at),id) + WHERE task_type='warmup' AND send_result_state='unknown' + AND warmup_send_retired_at IS NULL AND send_result_applied_at IS NULL; diff --git a/internal/infrastructure/db/warmup_unknown_retirement_upgrade_test.go b/internal/infrastructure/db/warmup_unknown_retirement_upgrade_test.go new file mode 100644 index 000000000..3e8948c28 --- /dev/null +++ b/internal/infrastructure/db/warmup_unknown_retirement_upgrade_test.go @@ -0,0 +1,106 @@ +package db + +import ( + "context" + "net/url" + "os" + "strings" + "testing" + + "github.com/golang-migrate/migrate/v4/source/iofs" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +func TestLiveWarmupUnknownRetirementUpgrade(t *testing.T) { + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + u, err := url.Parse(dsn) + if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") { + t.Fatal("upgrade requires local PostgreSQL") + } + admin, err := pgx.Connect(t.Context(), dsn) + if err != nil { + t.Fatal(err) + } + defer admin.Close(context.Background()) + name := "warmup_retirement_upgrade_" + uuid.New().String()[:8] + quoted := pgx.Identifier{name}.Sanitize() + if _, err = admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil { + t.Error(err) + } + }() + u.Path = "/" + name + source, err := iofs.New(migrationsFS, "migrations") + if err != nil { + t.Fatal(err) + } + m, err := openMigrate(source, u.String()) + if err != nil { + t.Fatal(err) + } + defer m.Close() + if err = m.Migrate(314); err != nil { + t.Fatal(err) + } + conn, err := pgx.Connect(t.Context(), u.String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + exec := func(sql string, args ...any) { + t.Helper() + if _, err := conn.Exec(t.Context(), sql, args...); err != nil { + t.Fatal(err) + } + } + user, org, mailbox, task := uuid.New(), uuid.New(), uuid.New(), uuid.New() + exec(`INSERT INTO users(id,email,first_name,last_name) VALUES($1,$2,'Upgrade','')`, user, user.String()+"@test.local") + exec(`INSERT INTO organizations(id,name,owner_user_id) VALUES($1,'Upgrade',$2)`, org, user) + exec(`INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,send_recovery_hold,send_recovery_reason,send_recovery_task_id) VALUES($1,$2,$3,$4,'Upgrade','','','smtp_imap',true,'unknown',$5)`, mailbox, user, org, mailbox.String()+"@test.local", task) + exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_reserved_at,send_executor_nonce,send_executor_started_at) VALUES($1,$2,'warmup','completed','','unknown',NOW(),gen_random_uuid(),NOW())`, task, mailbox) + var before, after string + query := `SELECT jsonb_build_object('task',to_jsonb(t)-'warmup_send_retired_at','account',to_jsonb(ea))::text FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1` + if err = conn.QueryRow(t.Context(), query, task).Scan(&before); err != nil { + t.Fatal(err) + } + if err = m.Migrate(315); err != nil { + t.Fatal(err) + } + if err = conn.QueryRow(t.Context(), query, task).Scan(&after); err != nil || before != after { + t.Fatalf("migration rewrote historical outcome or hold: err=%v", err) + } + if err = m.Steps(-1); err != nil { + t.Fatalf("empty retirement downgrade: %v", err) + } + if err = m.Steps(1); err != nil { + t.Fatal(err) + } + exec(`UPDATE tasks SET status='cancelled',warmup_send_retired_at=NOW() WHERE id=$1`, task) + for _, change := range []string{"status='pending'", "task_type='email'", "send_result_state='sent'", "send_result_applied_at=NOW()", "send_released_at=NOW()"} { + if _, err = conn.Exec(t.Context(), `UPDATE tasks SET `+change+` WHERE id=$1`, task); err == nil { + t.Fatalf("retirement invariant accepted %s", change) + } + } + down, err := migrationsFS.ReadFile("migrations/000315_warmup_unknown_send_retirement.down.sql") + if err != nil { + t.Fatal(err) + } + if _, err = conn.Exec(t.Context(), string(down)); err == nil || !strings.Contains(err.Error(), "cannot remove permanent warmup send retirement evidence") { + t.Fatalf("rollback discarded replay fence: %v", err) + } + var retained bool + if err = conn.QueryRow(t.Context(), `SELECT warmup_send_retired_at IS NOT NULL AND status='cancelled' AND send_result_state='unknown' FROM tasks WHERE id=$1`, task).Scan(&retained); err != nil || !retained { + t.Fatalf("refused down changed retirement: retained=%t err=%v", retained, err) + } + version, dirty, err := m.Version() + if err != nil || version != 315 || dirty { + t.Fatalf("schema version=%d dirty=%t err=%v", version, dirty, err) + } +} diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index 4a9d44b1d..f6a0d9b1c 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -1686,7 +1686,7 @@ func (r *emailRepository) Update(ctx context.Context, orgID, emailAccountID stri FROM email_accounts ea LEFT JOIN tasks t ON t.id=ea.send_recovery_task_id WHERE ea.organization_id=$1 AND ea.id=$2 AND ea.status='active' AND ea.send_recovery_hold AND ea.send_recovery_task_id=$6 AND ea.send_recovery_reason=$7 AND ea.send_recovery_reason IN('authentication','permanent','conflict') AND (ea.send_cooldown_provider IS NULL OR ea.send_cooldown_provider=ea.provider::text) FOR UPDATE OF ea), - valid AS(SELECT * FROM held WHERE NOT EXISTS(SELECT 1 FROM tasks u WHERE u.email_account_id=held.id AND u.send_result_state='unknown' AND u.send_result_applied_at IS NULL) + valid AS(SELECT * FROM held WHERE NOT EXISTS(SELECT 1 FROM tasks u WHERE u.email_account_id=held.id AND u.send_result_state='unknown' AND u.send_result_applied_at IS NULL AND u.warmup_send_retired_at IS NULL) AND (($3='authentication_repaired' AND held.send_recovery_reason='authentication' AND $4::uuid IS NOT NULL AND EXISTS(SELECT 1 FROM tasks e WHERE e.id=$4 AND e.email_account_id=held.id AND e.send_result_state='sent' AND e.send_result_applied_at>held.completed_at AND e.send_executor_started_at>held.completed_at)) OR ($3='operator_provider_confirmation' AND held.send_recovery_reason IN('authentication','permanent','conflict') AND length($5)>0 AND $4::uuid IS NOT NULL diff --git a/internal/repository/pg_monitoring.go b/internal/repository/pg_monitoring.go index e4ce6b3c9..f733bc653 100644 --- a/internal/repository/pg_monitoring.go +++ b/internal/repository/pg_monitoring.go @@ -193,10 +193,10 @@ var monitoringSections = map[string]monitoringSection{ {id: "confirmed_warmup_1h", title: "Provider-confirmed warmup tasks (1h)", unit: "tasks", condition: "healthy", severity: "info", note: "Provider acceptance only; legacy completed_at is a time proxy after independent token confirmation.", window: 3600 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(COALESCE(t.send_result_applied_at,t.completed_at)::timestamptz), MAX(COALESCE(t.send_result_applied_at,t.completed_at)::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.task_type='warmup' AND EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id AND wt.sent_message_id<>'') AND COALESCE(t.send_result_applied_at,t.completed_at) BETWEEN $1-INTERVAL '1 hour' AND $1`}, {id: "confirmed_other_1h", title: "Reconciled non-warmup sent tasks (1h)", unit: "tasks", condition: "healthy", severity: "info", note: "Native reconciliation evidence, not receiver delivery or inbox placement.", window: 3600 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_result_applied_at::timestamptz), MAX(t.send_result_applied_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.task_type<>'warmup' AND t.send_result_state='sent' AND t.send_result_applied_at BETWEEN $1-INTERVAL '1 hour' AND $1`}, {id: "confirmed_warmup_proxy_1h", title: "Legacy-clock confirmed warmup tasks (1h)", unit: "tasks", condition: "healthy", severity: "info", note: "Subset of confirmed warmup using completed_at proxy.", window: 3600 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.completed_at::timestamptz), MAX(t.completed_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.task_type='warmup' AND t.send_result_applied_at IS NULL AND EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id AND wt.sent_message_id<>'') AND t.completed_at BETWEEN $1-INTERVAL '1 hour' AND $1`}, - {id: "unknown_results", title: "Unverified send results", unit: "tasks", condition: "recovery_unverified", severity: "warning", note: "Includes legacy uncertainty; unknown is not a failed send.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_reserved_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_result_state='unknown'`}, - {id: "reservations_under_5m", title: "Outstanding reservation: under_5m", unit: "tasks", condition: "in_flight", severity: "info", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL AND t.send_reserved_at>$1-INTERVAL '5 minutes' AND t.send_reserved_at<=$1`}, - {id: "reservations_5_to_30m", title: "Outstanding reservation: 5_to_30m", unit: "tasks", condition: "in_flight", severity: "info", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL AND t.send_reserved_at<=$1-INTERVAL '5 minutes' AND t.send_reserved_at>$1-INTERVAL '30 minutes'`}, - {id: "reservations_over_30m", title: "Outstanding reservation: over_30m", unit: "tasks", condition: "recovery_unverified", severity: "warning", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL AND t.send_reserved_at<=$1-INTERVAL '30 minutes'`}, + {id: "unknown_results", title: "Unverified send results", unit: "tasks", condition: "recovery_unverified", severity: "warning", note: "Excludes permanently retired warmup sends; unknown is not a failed send.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_reserved_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL`}, + {id: "reservations_under_5m", title: "Outstanding reservation: under_5m", unit: "tasks", condition: "in_flight", severity: "info", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL AND t.send_result_applied_at IS NULL AND t.send_reserved_at>$1-INTERVAL '5 minutes' AND t.send_reserved_at<=$1`}, + {id: "reservations_5_to_30m", title: "Outstanding reservation: 5_to_30m", unit: "tasks", condition: "in_flight", severity: "info", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL AND t.send_result_applied_at IS NULL AND t.send_reserved_at<=$1-INTERVAL '5 minutes' AND t.send_reserved_at>$1-INTERVAL '30 minutes'`}, + {id: "reservations_over_30m", title: "Outstanding reservation: over_30m", unit: "tasks", condition: "recovery_unverified", severity: "warning", note: "Reservation precedes provider execution; age is not a native timeout guarantee.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_reserved_at::timestamptz), MAX(t.send_executor_started_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_reserved_at IS NOT NULL AND t.send_released_at IS NULL AND t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL AND t.send_result_applied_at IS NULL AND t.send_reserved_at<=$1-INTERVAL '30 minutes'`}, {id: "failed_results_1h", title: "Reconciled failed send results (1h)", unit: "tasks", condition: "recent_failure", severity: "warning", note: "Failure of these tasks, not a continuing mailbox outage.", window: 3600 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_result_applied_at::timestamptz), MAX(t.send_result_applied_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_result_state='failed' AND t.send_result_applied_at BETWEEN $1-INTERVAL '1 hour' AND $1`}, {id: "failed_results_24h", title: "Reconciled failed send results (24h)", unit: "tasks", condition: "recovery_unverified", severity: "warning", note: "Historical task outcomes; mailbox recovery requires scoped confirmation.", window: 86400 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.send_result_applied_at::timestamptz), MAX(t.send_result_applied_at::timestamptz), MIN(NULL::timestamptz) FROM tasks t LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.send_result_state='failed' AND t.send_result_applied_at BETWEEN $1-INTERVAL '24 hours' AND $1`}, }}, diff --git a/internal/repository/pg_send_recovery.go b/internal/repository/pg_send_recovery.go index b5916adb9..8bafedd98 100644 --- a/internal/repository/pg_send_recovery.go +++ b/internal/repository/pg_send_recovery.go @@ -117,11 +117,12 @@ type SendAdmission struct { type SendRecoveryHold struct { Reason string + Warmup bool } func (r *emailRepository) GetSendRecoveryHolds(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]SendRecoveryHold, error) { // Normal in-flight sends also hold admission; only surface settled or stuck ones. - query := `SELECT ea.id,COALESCE(ea.send_recovery_reason,'') FROM email_accounts ea + query := `SELECT ea.id,COALESCE(ea.send_recovery_reason,''),COALESCE(t.task_type='warmup',false) FROM email_accounts ea LEFT JOIN tasks t ON t.id=ea.send_recovery_task_id AND t.email_account_id=ea.id WHERE ea.id=ANY($1) AND ea.send_recovery_hold AND (ea.send_recovery_reason IS DISTINCT FROM 'unknown' OR t.id IS NULL @@ -142,7 +143,7 @@ func (r *emailRepository) GetSendRecoveryHolds(ctx context.Context, ids []uuid.U for rows.Next() { var id uuid.UUID var hold SendRecoveryHold - if err := rows.Scan(&id, &hold.Reason); err != nil { + if err := rows.Scan(&id, &hold.Reason, &hold.Warmup); err != nil { return nil, err } holds[id] = hold @@ -337,7 +338,7 @@ func refreshSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, taskID _, err := tx.Exec(ctx, `WITH next_hold AS ( SELECT id,reason FROM ( SELECT t.id,'unknown' AS reason,1 AS priority,t.created_at FROM tasks t - WHERE t.email_account_id=$1 AND t.send_result_state='unknown' + WHERE t.email_account_id=$1 AND t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL UNION ALL SELECT t.id,'conflict',2,t.created_at FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id WHERE t.email_account_id=$1 AND (r.previous_reason='unknown' OR r.evidence_type='authoritative_result_conflict') diff --git a/internal/repository/pg_warmup.go b/internal/repository/pg_warmup.go index 083938ed3..45487336b 100644 --- a/internal/repository/pg_warmup.go +++ b/internal/repository/pg_warmup.go @@ -1629,7 +1629,8 @@ func (r *warmupRepository) ProcessWarmupReceipt(ctx context.Context, tokenID, re if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=$1 AND lineage_version=1), EXISTS(SELECT 1 FROM warmup_tasks w JOIN tasks sent ON sent.id=w.task_id WHERE w.task_id=$1 AND w.lineage_version=1 AND (sent.status<>'completed' OR $2='')), - EXISTS(SELECT 1 FROM warmup_tokens WHERE token=$3 AND sent_retired_at IS NOT NULL)`, token.TaskID, token.SentMessageID, token.Token).Scan(&lineage, &awaitingSend, &retired); err != nil { + EXISTS(SELECT 1 FROM warmup_tokens WHERE token=$3 AND sent_retired_at IS NOT NULL) + OR EXISTS(SELECT 1 FROM tasks WHERE id=$1 AND warmup_send_retired_at IS NOT NULL)`, token.TaskID, token.SentMessageID, token.Token).Scan(&lineage, &awaitingSend, &retired); err != nil { return false, err } if retired { diff --git a/internal/repository/warmup_unknown_retirement.go b/internal/repository/warmup_unknown_retirement.go new file mode 100644 index 000000000..1354f3922 --- /dev/null +++ b/internal/repository/warmup_unknown_retirement.go @@ -0,0 +1,131 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/warmbly/warmbly/internal/models" +) + +const retirableUnknownWarmupSend = `t.task_type='warmup' AND t.status IN ('completed','failed','dead_lettered') + AND t.send_result_state='unknown' AND t.warmup_send_retired_at IS NULL + AND t.send_result_applied_at IS NULL AND t.send_released_at IS NULL + AND ea.send_recovery_hold AND ea.send_recovery_reason='unknown' AND ea.send_recovery_task_id=t.id + AND (t.send_executor_nonce IS NOT NULL OR w.dispatch_nonce IS NOT NULL) + AND (t.send_executor_started_at IS NOT NULL OR w.dispatch_started_at IS NOT NULL + OR EXISTS(SELECT 1 FROM outbound_attempts a WHERE a.task_id=t.id)) + AND COALESCE(t.send_reserved_at,t.completed_at,t.updated_at)<$1 + AND (t.send_executor_started_at IS NULL OR t.send_executor_started_at<$1) + AND (w.dispatch_started_at IS NULL OR w.dispatch_started_at<$1) + AND NOT EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id AND wt.sent_message_id<>'') + AND NOT EXISTS(SELECT 1 FROM warmup_received wr WHERE wr.task_id=t.id) + AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions sr WHERE sr.recovery_task_id=t.id + AND (sr.confirmed_result IS NOT NULL OR sr.conflict_detected_at IS NOT NULL))` + +const retiredWarmupHold = `t.warmup_send_retired_at IS NOT NULL + AND ea.send_recovery_hold AND ea.send_recovery_reason='unknown' AND ea.send_recovery_task_id=t.id` + +func (r *taskRepository) RetireUnknownWarmupSends(ctx context.Context, before time.Time, limit int) (int, error) { + limit = min(max(limit, 1), 500) + var after *time.Time + afterTask := uuid.Nil + retired := 0 + for retired < limit { + rows, err := r.db.Query(ctx, `SELECT t.id,t.email_account_id,COALESCE(t.send_reserved_at,t.completed_at,t.updated_at) FROM tasks t + JOIN warmup_tasks w ON w.task_id=t.id JOIN email_accounts ea ON ea.id=t.email_account_id + WHERE ((`+retirableUnknownWarmupSend+`) OR (`+retiredWarmupHold+`)) + AND ($3::timestamptz IS NULL OR (COALESCE(t.send_reserved_at,t.completed_at,t.updated_at),t.id)>($3,$4::uuid)) + ORDER BY COALESCE(t.send_reserved_at,t.completed_at,t.updated_at),t.id LIMIT $2`, before, limit-retired, after, afterTask) + if err != nil { + return retired, err + } + type candidate struct { + task, mailbox uuid.UUID + at time.Time + } + var candidates []candidate + for rows.Next() { + var c candidate + if err := rows.Scan(&c.task, &c.mailbox, &c.at); err != nil { + rows.Close() + return retired, err + } + candidates = append(candidates, c) + } + err = rows.Err() + rows.Close() + if err != nil { + return retired, err + } + if len(candidates) == 0 { + break + } + for _, c := range candidates { + ok, err := r.retireUnknownWarmupSend(ctx, c.task, c.mailbox, before) + if err != nil { + return retired, err + } + if ok { + retired++ + } + } + // Protected historical payloads must not starve later recoverable mailboxes. + last := candidates[len(candidates)-1] + after, afterTask = &last.at, last.task + } + return retired, nil +} + +func (r *taskRepository) retireUnknownWarmupSend(ctx context.Context, task, mailbox uuid.UUID, before time.Time) (bool, error) { + tx, err := r.warmupSchedulingTx(ctx, task) + if err != nil { + return false, err + } + defer tx.Rollback(ctx) + if err = lockSend(ctx, tx, task, mailbox); err != nil { + return false, err + } + var executorResult, dispatchResult []byte + var alreadyRetired bool + err = tx.QueryRow(ctx, `SELECT t.send_executor_result,w.dispatch_result,t.warmup_send_retired_at IS NOT NULL FROM tasks t + JOIN warmup_tasks w ON w.task_id=t.id JOIN email_accounts ea ON ea.id=t.email_account_id + WHERE t.id=$2 AND t.email_account_id=$3 AND ((`+retirableUnknownWarmupSend+`) OR (`+retiredWarmupHold+`)) + FOR UPDATE OF t,w,ea`, before, task, mailbox).Scan(&executorResult, &dispatchResult, &alreadyRetired) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + if alreadyRetired { + // Older replicas may reassert a historical unknown hold during a rolling upgrade. + if err = refreshUnknownSendHold(ctx, tx, mailbox, task); err != nil { + return false, err + } + return true, tx.Commit(ctx) + } + for _, raw := range [][]byte{executorResult, dispatchResult} { + if len(raw) == 0 { + continue + } + var result models.SendEmailResult + if json.Unmarshal(raw, &result) != nil || result.TaskID != task || resultState(result) != "unknown" { + return false, nil + } + } + // Retirement fences replay, not provider outcome or accounting: retain every original send marker. + if _, err = tx.Exec(ctx, `UPDATE tasks SET status='cancelled',warmup_send_retired_at=NOW(),updated_at=NOW() WHERE id=$1`, task); err != nil { + return false, err + } + if err = resolveSendDeadLetter(ctx, tx, task); err != nil { + return false, err + } + if err = refreshUnknownSendHold(ctx, tx, mailbox, task); err != nil { + return false, err + } + return true, tx.Commit(ctx) +} diff --git a/internal/tasks/warmup_dispatch_recovery_live_test.go b/internal/tasks/warmup_dispatch_recovery_live_test.go index c26e95506..23bdee8c0 100644 --- a/internal/tasks/warmup_dispatch_recovery_live_test.go +++ b/internal/tasks/warmup_dispatch_recovery_live_test.go @@ -21,6 +21,7 @@ type warmupRecoveryRepository interface { repository.WarmupDispatchRepository RecoverUnstartedWarmupDispatches(context.Context, time.Time, int) (int, error) RecoverAbandonedWarmupPreparations(context.Context, time.Time, int) (int, error) + RetireUnknownWarmupSends(context.Context, time.Time, int) (int, error) } func (f *warmupRecoveryFixture) abandonedPreparation(t *testing.T) uuid.UUID { diff --git a/internal/tasks/warmup_reconciler.go b/internal/tasks/warmup_reconciler.go index dca99d233..0c1537795 100644 --- a/internal/tasks/warmup_reconciler.go +++ b/internal/tasks/warmup_reconciler.go @@ -66,6 +66,17 @@ func (s *tasksService) ReconcileWarmupSchedules(ctx context.Context, limit int) log.Info().Int("recovered", recovered).Msg("warmup reconcile retired unstarted dispatches") } } + if recovery, ok := s.taskRepo.(interface { + RetireUnknownWarmupSends(context.Context, time.Time, int) (int, error) + }); ok { + n, err := recovery.RetireUnknownWarmupSends(ctx, time.Now().Add(-10*time.Minute), limit) + if err != nil { + return 0, err + } + if n > 0 { + log.Info().Int("retired", n).Msg("warmup reconcile permanently retired unknown sends without replay or refund") + } + } if lineage, ok := s.taskRepo.(repository.WarmupLineageRepository); ok { pending, err := lineage.UnqueuedWarmupTasks(ctx, limit) if err != nil { diff --git a/internal/tasks/warmup_reconciler_recovery_test.go b/internal/tasks/warmup_reconciler_recovery_test.go index 2ee373e90..2a9879ff7 100644 --- a/internal/tasks/warmup_reconciler_recovery_test.go +++ b/internal/tasks/warmup_reconciler_recovery_test.go @@ -35,6 +35,34 @@ func (r *reconcileRecoveryRepo) RecoverUnstartedWarmupDispatches(_ context.Conte type reconcileEmptyEmailRepo struct{ repository.EmailRepository } +type unknownWarmupRecoveryRepo struct { + reconcileRecoveryRepo + retirementCalls int + retirementErr error +} + +func (r *unknownWarmupRecoveryRepo) RetireUnknownWarmupSends(_ context.Context, before time.Time, limit int) (int, error) { + if d := time.Since(before); d < 9*time.Minute || d > 11*time.Minute || limit != 500 || r.calls != 1 { + return 0, errors.New("unknown retirement must follow unstarted recovery with the safety window and batch bound") + } + r.retirementCalls++ + return 1, r.retirementErr +} + +func TestWarmupReconcilerRetiresUnknownSendsAfterUnstartedRecovery(t *testing.T) { + r := &unknownWarmupRecoveryRepo{} + s := &tasksService{taskRepo: r, emailRepo: &reconcileEmptyEmailRepo{}} + if _, err := s.ReconcileWarmupSchedules(t.Context(), 500); err != nil || r.retirementCalls != 1 { + t.Fatalf("retirements=%d err=%v", r.retirementCalls, err) + } + r.calls = 0 + want := errors.New("retirement unavailable") + r.retirementErr = want + if _, err := s.ReconcileWarmupSchedules(t.Context(), 500); !errors.Is(err, want) || r.retirementCalls != 2 { + t.Fatalf("failed retirement must stop the recovery pass: retirements=%d err=%v", r.retirementCalls, err) + } +} + type abandonedPreparationRecoveryRepo struct { reconcileRecoveryRepo preparationCalls int diff --git a/internal/tasks/warmup_unknown_retirement_live_test.go b/internal/tasks/warmup_unknown_retirement_live_test.go new file mode 100644 index 000000000..65e6fb66d --- /dev/null +++ b/internal/tasks/warmup_unknown_retirement_live_test.go @@ -0,0 +1,364 @@ +package tasks + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" + "github.com/warmbly/warmbly/internal/scheduler" +) + +func newUnknownWarmupFixture(t *testing.T) *warmupRecoveryFixture { + t.Helper() + f := newWarmupRecoveryFixture(t) + result := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Code: "SERVER_UNREACHABLE", Failure: &errx.SendFailure{ + Protocol: "smtp", Stage: "message accept", Disposition: errx.SendAmbiguous, Scope: "mailbox", + }}} + raw, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + f.exec(t, `UPDATE tasks SET message_id='',send_executor_started_at=NOW()-INTERVAL '1 hour',send_executor_result=$2 WHERE id=$1`, f.task, raw) + f.exec(t, `UPDATE warmup_tasks SET dispatch_started_at=NOW()-INTERVAL '1 hour',dispatch_result=$2 WHERE task_id=$1`, f.task, raw) + f.exec(t, `INSERT INTO outbound_attempts(nonce,task_id,email_account_id,provider,attempted_at,recipient_count) VALUES($1,$2,$3,'smtp_imap',NOW()-INTERVAL '1 hour',1)`, f.nonce, f.task, f.sender.ID) + account, xerr := f.svc.emailRepo.GetByID(t.Context(), f.sender.ID) + if xerr != nil { + t.Fatal(xerr) + } + f.sender = *account + return f +} + +func unknownWarmupSnapshot(t *testing.T, f *warmupRecoveryFixture) string { + t.Helper() + var snapshot string + if err := f.pool.QueryRow(t.Context(), `SELECT jsonb_build_object( + 'task',to_jsonb(t)-'status'-'updated_at'-'warmup_send_retired_at','warmup',to_jsonb(w), + 'tokens',(SELECT jsonb_agg(to_jsonb(wt)) FROM warmup_tokens wt WHERE wt.task_id=t.id), + 'attempts',(SELECT jsonb_agg(to_jsonb(a)) FROM outbound_attempts a WHERE a.task_id=t.id), + 'charges',(SELECT jsonb_agg(to_jsonb(s)) FROM warmup_statistics s WHERE s.email_account_id=t.email_account_id))::text + FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&snapshot); err != nil { + t.Fatal(err) + } + return snapshot +} + +func assertUnknownWarmupRetired(t *testing.T, f *warmupRecoveryFixture) { + t.Helper() + var retired, held, unknown bool + if err := f.pool.QueryRow(t.Context(), `SELECT t.warmup_send_retired_at IS NOT NULL AND t.status='cancelled', + t.send_result_state='unknown' AND t.send_result_applied_at IS NULL AND t.send_released_at IS NULL,ea.send_recovery_hold + FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, f.task).Scan(&retired, &unknown, &held); err != nil || !retired || !unknown || held { + t.Fatalf("retired=%t unknown=%t held=%t err=%v", retired, unknown, held, err) + } + for _, inspect := range []func(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (*repository.WarmupDispatchState, error){f.r.InspectOutbound, f.r.InspectWarmupDispatch} { + state, err := inspect(t.Context(), f.task, f.sender.ID, f.worker) + if err != nil || state.State != "denied" { + t.Fatalf("retired command became sendable: %+v err=%v", state, err) + } + } +} + +func TestLiveWarmupUnknownRetirementPreservesEvidenceAndFencesReplayOnce(t *testing.T) { + f := newUnknownWarmupFixture(t) + before := unknownWarmupSnapshot(t, f) + var wg sync.WaitGroup + var mu sync.Mutex + total := 0 + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100) + if err != nil { + t.Error(err) + } + mu.Lock() + total += n + mu.Unlock() + }() + } + wg.Wait() + if total != 1 { + t.Fatalf("retired %d times", total) + } + assertUnknownWarmupRetired(t, f) + if after := unknownWarmupSnapshot(t, f); before != after { + t.Fatalf("retirement changed send evidence or accounting: before=%s after=%s", before, after) + } + for _, begin := range []func(context.Context, uuid.UUID, uuid.UUID, uuid.UUID, uuid.UUID) (*repository.WarmupDispatchState, error){f.r.BeginOutbound, f.r.BeginWarmupDispatch} { + state, err := begin(t.Context(), f.task, f.sender.ID, f.worker, f.nonce) + if err != nil || state.State != "denied" { + t.Fatalf("late command executed: %+v err=%v", state, err) + } + } + if _, err := f.r.AuthorizeWarmupDispatch(t.Context(), f.task, f.sender.ID, f.worker); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("retired task reauthorized: %v", err) + } + if _, err := f.pool.Exec(t.Context(), `UPDATE tasks SET status='pending' WHERE id=$1`, f.task); err == nil { + t.Fatal("retired task can be revived") + } +} + +func TestLiveWarmupUnknownRetirementLeavesOtherHoldsAndConfirmedEvidenceProtected(t *testing.T) { + for _, tc := range []struct{ name, query string }{ + {"recent reservation", `UPDATE tasks SET send_reserved_at=NOW() WHERE id=$1`}, + {"recent execution", `UPDATE tasks SET send_executor_started_at=NOW() WHERE id=$1`}, + {"recent warmup execution", `UPDATE warmup_tasks SET dispatch_started_at=NOW() WHERE task_id=$1`}, + {"campaign", `UPDATE tasks SET task_type='campaign' WHERE id=$1`}, + {"manual send", `UPDATE tasks SET task_type='email' WHERE id=$1`}, + {"placement", `UPDATE tasks SET task_type='placement' WHERE id=$1`}, + {"authentication hold", `UPDATE email_accounts SET send_recovery_reason='authentication' WHERE id=(SELECT email_account_id FROM tasks WHERE id=$1)`}, + {"permanent hold", `UPDATE email_accounts SET send_recovery_reason='permanent' WHERE id=(SELECT email_account_id FROM tasks WHERE id=$1)`}, + {"conflict hold", `UPDATE email_accounts SET send_recovery_reason='conflict' WHERE id=(SELECT email_account_id FROM tasks WHERE id=$1)`}, + {"changed held task", `UPDATE email_accounts SET send_recovery_task_id=gen_random_uuid() WHERE id=(SELECT email_account_id FROM tasks WHERE id=$1)`}, + {"applied result", `UPDATE tasks SET send_result_applied_at=NOW() WHERE id=$1`}, + {"confirmed sent", `UPDATE tasks SET send_result_state='sent' WHERE id=$1`}, + {"released reservation", `UPDATE tasks SET send_released_at=NOW() WHERE id=$1`}, + {"provider message ID", `UPDATE warmup_tokens SET sent_message_id='' WHERE task_id=$1`}, + {"confirmed receipt", `INSERT INTO warmup_received(email_account_id,internal_id,sender_account_id,message_id,task_id) SELECT recipient_account_id,gen_random_uuid(),sender_account_id,'',task_id FROM warmup_tokens WHERE task_id=$1`}, + {"unfenced legacy", `WITH updated AS(UPDATE tasks SET send_executor_nonce=NULL WHERE id=$1) UPDATE warmup_tasks SET dispatch_nonce=NULL WHERE task_id=$1`}, + {"definitive executor success", `UPDATE tasks SET send_executor_result=jsonb_build_object('task_id',id,'success',true) WHERE id=$1`}, + {"definitive dispatch success", `UPDATE warmup_tasks SET dispatch_result=jsonb_build_object('task_id',task_id,'success',true) WHERE task_id=$1`}, + {"definitive executor refusal", `UPDATE tasks SET send_executor_result=jsonb_build_object('task_id',id,'error',jsonb_build_object('code','RECIPIENT_REJECTED')) WHERE id=$1`}, + {"invalid result lineage", `UPDATE tasks SET send_executor_result=jsonb_build_object('task_id',gen_random_uuid()) WHERE id=$1`}, + {"malformed stored result", `UPDATE tasks SET send_executor_result='[]' WHERE id=$1`}, + {"queued confirmation", `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result,recorded_locally_at) SELECT ea.organization_id,ea.id,t.id,'unknown','operator_confirmed_sent','test evidence',jsonb_build_object('task_id',t.id,'success',true),NOW() FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newUnknownWarmupFixture(t) + f.exec(t, tc.query, f.task) + before := unknownWarmupSnapshot(t, f) + n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100) + if err != nil || n != 0 { + t.Fatalf("unsafe retirement: n=%d err=%v", n, err) + } + if after := unknownWarmupSnapshot(t, f); after != before { + t.Fatal("protected evidence changed") + } + var held bool + if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, f.sender.ID).Scan(&held); err != nil || !held { + t.Fatalf("protected mailbox released: held=%t err=%v", held, err) + } + }) + } +} + +func TestLiveWarmupUnknownRetirementLateResultsCannotChangeNewAdmission(t *testing.T) { + f := newUnknownWarmupFixture(t) + if n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100); err != nil || n != 1 { + t.Fatalf("retired=%d err=%v", n, err) + } + next := uuid.New() + f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state) VALUES($1,$2,'warmup','completed','','unknown')`, next, f.sender.ID) + f.exec(t, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender.ID, next) + for _, result := range []models.SendEmailResult{ + {TaskID: f.task, Success: true}, + {TaskID: f.task, Error: &models.EmailSendError{Code: "RECIPIENT_REJECTED"}}, + {TaskID: f.task}, + } { + called := false + err := f.r.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(context.Context) error { called = true; return nil }) + if err != nil || called { + t.Fatalf("late result applied accounting: called=%t err=%v", called, err) + } + } + var heldTask uuid.UUID + if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_task_id FROM email_accounts WHERE id=$1 AND send_recovery_hold`, f.sender.ID).Scan(&heldTask); err != nil || heldTask != next { + t.Fatalf("late results changed new hold: task=%s err=%v", heldTask, err) + } +} + +func TestLiveWarmupUnknownRetirementAllowsDistinctNewSendAndKeepsCooldown(t *testing.T) { + f := newUnknownWarmupFixture(t) + until := time.Now().Add(time.Hour) + f.exec(t, `UPDATE email_accounts SET send_cooldown_until=$2,send_cooldown_provider='smtp_imap' WHERE id=$1`, f.sender.ID, until) + if n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100); err != nil || n != 1 { + t.Fatalf("retired=%d err=%v", n, err) + } + r := f.r.(repository.SendResultRecovery) + a, err := r.GetSendAdmission(t.Context(), f.org, f.sender.ID, "smtp_imap", time.Now()) + if err != nil || a.Allowed || a.RecoveryHold || a.RetryAt == nil || !a.RetryAt.Equal(until.Truncate(time.Microsecond)) { + t.Fatalf("cooldown bypassed: admission=%+v err=%v", a, err) + } + f.exec(t, `UPDATE email_accounts SET send_cooldown_until=NULL WHERE id=$1`, f.sender.ID) + next, token := uuid.New(), uuid.New() + f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'warmup','active','')`, next, f.sender.ID) + f.exec(t, `INSERT INTO warmup_tasks(task_id,lineage_version,target_account_id) VALUES($1,1,$2)`, next, f.atWorkspace) + f.exec(t, `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id) VALUES($1,$2,$3,$4)`, token, next, f.sender.ID, f.atWorkspace) + nonce, err := f.r.AuthorizeWarmupDispatch(t.Context(), next, f.sender.ID, f.worker) + if err != nil || nonce == f.nonce || nonce == uuid.Nil { + t.Fatalf("new task was not authorized independently: nonce=%s err=%v", nonce, err) + } + nonce, err = f.r.ReserveOutbound(t.Context(), repository.OutboundReservation{ + TaskID: next, MailboxID: f.sender.ID, OrganizationID: f.org, WorkerID: f.worker, Provider: "smtp_imap", + Recipients: []string{"pick-" + f.atWorkspace.String()[:8] + "@acme.test"}, + }) + if err != nil { + t.Fatalf("new distinct task could not reserve outbound admission: %v", err) + } + state, err := f.r.BeginOutbound(t.Context(), next, f.sender.ID, f.worker, nonce) + if err != nil || state.State != "execute" { + t.Fatalf("distinct new send blocked: state=%+v err=%v", state, err) + } + old, err := f.r.BeginWarmupDispatch(t.Context(), f.task, f.sender.ID, f.worker, f.nonce) + if err != nil || old.State != "denied" { + t.Fatalf("old task revived by new authorization: state=%+v err=%v", old, err) + } +} + +func TestLiveWarmupUnknownRetirementAndDefinitiveResultSerialize(t *testing.T) { + for i := range 10 { + t.Run(fmt.Sprint(i), func(t *testing.T) { + f := newUnknownWarmupFixture(t) + f.exec(t, `UPDATE tasks SET send_executor_result=NULL WHERE id=$1`, f.task) + f.exec(t, `UPDATE warmup_tasks SET dispatch_result=NULL WHERE task_id=$1`, f.task) + var wg sync.WaitGroup + var n int + var called bool + wg.Add(2) + go func() { + defer wg.Done() + var err error + n, err = f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100) + if err != nil { + t.Error(err) + } + }() + go func() { + defer wg.Done() + if err := f.r.(repository.SendResultRecovery).ApplySendResult(t.Context(), models.SendEmailResult{TaskID: f.task, Success: true}, func(context.Context) error { called = true; return nil }); err != nil { + t.Error(err) + } + }() + wg.Wait() + if n == 1 && called || n == 0 && !called { + t.Fatalf("retirement and reconciliation must not both win or lose: retired=%d applied=%t", n, called) + } + }) + } +} + +func TestLiveWarmupUnknownRetirementMovesHoldWithoutResurrectingRetiredTasks(t *testing.T) { + f := newUnknownWarmupFixture(t) + other := uuid.New() + f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state) VALUES($1,$2,'email','completed','','unknown')`, other, f.sender.ID) + if n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100); err != nil || n != 1 { + t.Fatalf("retired=%d err=%v", n, err) + } + var heldTask uuid.UUID + if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_task_id FROM email_accounts WHERE id=$1 AND send_recovery_hold`, f.sender.ID).Scan(&heldTask); err != nil || heldTask != other { + t.Fatalf("other unknown lost: task=%s err=%v", heldTask, err) + } + if err := f.r.(repository.SendResultRecovery).ApplySendResult(t.Context(), models.SendEmailResult{TaskID: other, Success: true}, func(context.Context) error { return nil }); err != nil { + t.Fatal(err) + } + assertUnknownWarmupRetired(t, f) + // Simulate an outgoing old replica reasserting an unknown hold during deployment. + f.exec(t, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender.ID, f.task) + if n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100); err != nil || n != 1 { + t.Fatalf("stale hold repair=%d err=%v", n, err) + } + assertUnknownWarmupRetired(t, f) +} + +func TestLiveWarmupUnknownRetirementLateReceiptDoesNotWaitOrTriggerActions(t *testing.T) { + f := newUnknownWarmupFixture(t) + if n, err := f.r.RetireUnknownWarmupSends(t.Context(), f.before, 100); err != nil || n != 1 { + t.Fatalf("retired=%d err=%v", n, err) + } + var token uuid.UUID + if err := f.pool.QueryRow(t.Context(), `SELECT token FROM warmup_tokens WHERE task_id=$1`, f.task).Scan(&token); err != nil { + t.Fatal(err) + } + called := false + matched, err := f.svc.warmupRepo.(repository.WarmupReceiptProcessor).ProcessWarmupReceipt(t.Context(), token, f.atWorkspace, + repository.WarmupReceiptProof{SenderAddress: f.sender.Email, Token: token, MessageID: ""}, + func(*models.WarmupToken) error { called = true; return nil }) + if err != nil || !matched || called { + t.Fatalf("retired receipt deferred or authorized actions: matched=%t called=%t err=%v", matched, called, err) + } +} + +func TestLiveWarmupUnknownRetirementSkipsProtectedRowsWithoutStarvingLaterMailboxes(t *testing.T) { + protected := newUnknownWarmupFixture(t) + protected.exec(t, `UPDATE tasks SET send_reserved_at=NOW()-INTERVAL '2 hours',send_executor_result='[]' WHERE id=$1`, protected.task) + next := uuid.New() + protected.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_reserved_at,send_executor_nonce,send_executor_started_at) + VALUES($1,$2,'warmup','completed','','unknown',NOW()-INTERVAL '1 hour',gen_random_uuid(),NOW()-INTERVAL '1 hour')`, next, protected.atSmallHost) + protected.exec(t, `INSERT INTO warmup_tasks(task_id) VALUES($1)`, next) + protected.exec(t, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, protected.atSmallHost, next) + t.Cleanup(func() { _, _ = protected.pool.Exec(context.Background(), `DELETE FROM tasks WHERE id=$1`, next) }) + if n, err := protected.r.RetireUnknownWarmupSends(t.Context(), protected.before, 1); err != nil || n != 1 { + t.Fatalf("protected result starved eligible mailbox: retired=%d err=%v", n, err) + } + var retired bool + if err := protected.pool.QueryRow(t.Context(), `SELECT t.warmup_send_retired_at IS NOT NULL AND t.status='cancelled' AND NOT ea.send_recovery_hold + FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, next).Scan(&retired); err != nil || !retired { + t.Fatalf("later eligible mailbox remained stuck: retired=%t err=%v", retired, err) + } + var held bool + if err := protected.pool.QueryRow(t.Context(), `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, protected.sender.ID).Scan(&held); err != nil || !held { + t.Fatalf("protected mailbox unblocked: held=%t err=%v", held, err) + } +} + +func TestLiveWarmupUnknownRetirementReconcilerSeedsOnceAndClearsActiveWarnings(t *testing.T) { + f := newUnknownWarmupFixture(t) + f.svc.taskRepo = repository.NewTaskRepository(f.pool) + f.svc.scheduler = scheduler.NewSchedulerService(f.svc.taskRepo, f.svc.warmupRepo, nil, f.svc.emailRepo, nil, nil, nil) + f.svc.tasksClient = noopTaskScheduler{} + counts := func() map[string]int64 { + t.Helper() + for _, source := range repository.NewMonitoringRepository(f.pool).Sources() { + if source.ID != "sends" { + continue + } + out, err := source.Collect(t.Context(), time.Now()) + if err != nil { + t.Fatal(err) + } + result := map[string]int64{} + for _, metric := range out.Metrics { + if metric.ID == "unknown_results" || metric.ID == "reservations_over_30m" { + if metric.Count == nil { + t.Fatalf("recovery warning %s unavailable", metric.ID) + } + result[metric.ID] = *metric.Count + } + } + return result + } + t.Fatal("send monitoring source absent") + return nil + } + before := counts() + for range 2 { + if _, err := f.svc.ReconcileWarmupSchedules(t.Context(), 500); err != nil { + t.Fatal(err) + } + } + assertUnknownWarmupRetired(t, f) + var pending int + if err := f.pool.QueryRow(t.Context(), `SELECT count(*) FROM tasks WHERE email_account_id=$1 AND task_type='warmup' AND status='pending' AND id<>$2`, f.sender.ID, f.task).Scan(&pending); err != nil || pending != 1 { + t.Fatalf("normal scheduling failed to seed exactly one distinct wakeup: pending=%d err=%v", pending, err) + } + after := counts() + if len(before) != 2 || len(after) != 2 { + t.Fatalf("missing recovery warnings: before=%v after=%v", before, after) + } + for id, count := range before { + if after[id] != count-1 { + t.Fatalf("retired history remains an active warning: %s before=%d after=%d", id, count, after[id]) + } + } +}