From a19106db218c9665eac753d3cc2a305fd1cb7fcc Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 15:06:48 +0000 Subject: [PATCH 1/2] feat: run exact warmup charge UTC refund and authoritative legacy result regressions in serialized PostgreSQL CI with isolated upgrade coverage --- .github/workflows/ci.yml | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ae1f25599..764b6cfec 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,6 +47,7 @@ jobs: with: filters: | go-services: + - '.github/workflows/ci.yml' - 'go.mod' - 'go.sum' - 'internal/**' @@ -220,6 +221,13 @@ jobs: go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1 go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1 + - name: Warmup charge and authoritative result regressions (PostgreSQL) + env: + WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable + run: | + go test -race -v -p 1 ./internal/repository ./internal/app/consumer ./internal/app/orgtransfer -run '^TestLive(Warmup(ReservationPreservesOriginalChargeDay|RefundCASProtectsConcurrentAndRestampedTasks|RefundAndResultRollbackTogether|ReplyRefundUsesChargedShapeAfterParentDeletion|ChargeAndRefundUseUTCInNonUTCSession|ChargeProofRollsBackWhenCounterWriteFails|CancellationPathsRefundOnlyProvenCharge|GenericCompletionDoesNotProveChargedAccounting|SendFailureGivesTheDayBack|SendGiveBackIsIdempotent)|GiveBackDailySend(ReversesADispatchedWarmupSend|LeavesOtherDaysAlone)|AccountDailyUsage(CountsWarmupTasksNotTheStatisticsRow|ScopesWarmupToTypeAndDay)|ImportWarmupStatisticsCannotGrantChargeRefundAuthority|Legacy(FailedStatusCannotProveNonExecution|JSONAndAvroFailurePersistConservativeAdmission)|AuthoritativeDeadLetterResultsReconcileWithoutReplay|DeadLetterResultEffectsAndSettlementRollBackTogether|Campaign(DeadLetterWorkerResultReleasesReservationOnce|ForeignKeyCleanupDoesNotProveWakeupOnly)|UserEmailDeadLetterUsesUnderlyingSendTaskType|UnknownSendOperatorResolutionReconcilesAndAudits|RecordedWorkerOutcomeSurvivesLostEvent|UnreservedLegacyFailedTaskRetainsUnknownHold|UnknownResolutionRejectsMissingEvidenceAndLiveExecution|SendRecoveryResolutionRequiresEvidenceAndRetainsUnknown)$' -count=1 + go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveWarmupChargeUpgradeKeepsLegacyUnverifiedAndPreservesEvidence$' -count=1 + - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL) env: WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable From 3888a2e71986c7950213c5389410c8c33f68f84d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 15:16:02 +0000 Subject: [PATCH 2/2] feat: repair already applied warmup failure dead letters only with durable refunded charge proof without replaying callbacks or mailbox holds and cover exact task regressions in PostgreSQL CI --- .github/workflows/ci.yml | 2 +- docs/content/docs/development/send-safety.mdx | 2 + .../warmup_charge_provenance_live_test.go | 113 ++++++++++++++++++ internal/repository/pg_send_recovery.go | 21 +++- .../repository/unknown_send_resolution.go | 8 +- 5 files changed, 140 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 764b6cfec..8b60f17dc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -225,7 +225,7 @@ jobs: env: WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable run: | - go test -race -v -p 1 ./internal/repository ./internal/app/consumer ./internal/app/orgtransfer -run '^TestLive(Warmup(ReservationPreservesOriginalChargeDay|RefundCASProtectsConcurrentAndRestampedTasks|RefundAndResultRollbackTogether|ReplyRefundUsesChargedShapeAfterParentDeletion|ChargeAndRefundUseUTCInNonUTCSession|ChargeProofRollsBackWhenCounterWriteFails|CancellationPathsRefundOnlyProvenCharge|GenericCompletionDoesNotProveChargedAccounting|SendFailureGivesTheDayBack|SendGiveBackIsIdempotent)|GiveBackDailySend(ReversesADispatchedWarmupSend|LeavesOtherDaysAlone)|AccountDailyUsage(CountsWarmupTasksNotTheStatisticsRow|ScopesWarmupToTypeAndDay)|ImportWarmupStatisticsCannotGrantChargeRefundAuthority|Legacy(FailedStatusCannotProveNonExecution|JSONAndAvroFailurePersistConservativeAdmission)|AuthoritativeDeadLetterResultsReconcileWithoutReplay|DeadLetterResultEffectsAndSettlementRollBackTogether|Campaign(DeadLetterWorkerResultReleasesReservationOnce|ForeignKeyCleanupDoesNotProveWakeupOnly)|UserEmailDeadLetterUsesUnderlyingSendTaskType|UnknownSendOperatorResolutionReconcilesAndAudits|RecordedWorkerOutcomeSurvivesLostEvent|UnreservedLegacyFailedTaskRetainsUnknownHold|UnknownResolutionRejectsMissingEvidenceAndLiveExecution|SendRecoveryResolutionRequiresEvidenceAndRetainsUnknown)$' -count=1 + go test -race -v -p 1 ./internal/repository ./internal/app/consumer ./internal/tasks ./internal/app/orgtransfer -run '^TestLive(Warmup(ReservationPreservesOriginalChargeDay|RefundCASProtectsConcurrentAndRestampedTasks|RefundAndResultRollbackTogether|ReplyRefundUsesChargedShapeAfterParentDeletion|ChargeAndRefundUseUTCInNonUTCSession|ChargeProofRollsBackWhenCounterWriteFails|CancellationPathsRefundOnlyProvenCharge|GenericCompletionDoesNotProveChargedAccounting|SendFailureGivesTheDayBack|SendGiveBackIsIdempotent|SafetyHoldReschedulesInsteadOfBlockingTheDispatcher|Recovery(RetiresExistingUnstartedDispatchOnce|HandlesPartialReservations|PreservesExecutionAndReceiptEvidence|KeepsOtherUnknownAndRestrictiveHolds|AndBeginCannotBothWin))|AppliedWarmupFailureRepairsOnlyRefundedChargeDeadLetters|GiveBackDailySend(ReversesADispatchedWarmupSend|LeavesOtherDaysAlone)|AccountDailyUsage(CountsWarmupTasksNotTheStatisticsRow|ScopesWarmupToTypeAndDay)|ImportWarmupStatisticsCannotGrantChargeRefundAuthority|Legacy(FailedStatusCannotProveNonExecution|JSONAndAvroFailurePersistConservativeAdmission)|AuthoritativeDeadLetterResultsReconcileWithoutReplay|DeadLetterResultEffectsAndSettlementRollBackTogether|Campaign(DeadLetterWorkerResultReleasesReservationOnce|ForeignKeyCleanupDoesNotProveWakeupOnly)|UserEmailDeadLetterUsesUnderlyingSendTaskType|UnknownSendOperatorResolutionReconcilesAndAudits|RecordedWorkerOutcomeSurvivesLostEvent|UnreservedLegacyFailedTaskRetainsUnknownHold|UnknownResolutionRejectsMissingEvidenceAndLiveExecution|SendRecoveryResolutionRequiresEvidenceAndRetainsUnknown)$' -count=1 go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveWarmupChargeUpgradeKeepsLegacyUnverifiedAndPreservesEvidence$' -count=1 - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL) diff --git a/docs/content/docs/development/send-safety.mdx b/docs/content/docs/development/send-safety.mdx index 8c44082ac..e048fdab7 100644 --- a/docs/content/docs/development/send-safety.mdx +++ b/docs/content/docs/development/send-safety.mdx @@ -41,6 +41,8 @@ Result reconciliation commits task, campaign progress, variant, suppression and Warmup authorization records the UTC charge day and whether a reply was counted in the same transaction as the statistics increment. Outbound reservation preserves the initial completion timestamp. A definitive failure or safe unstarted cancellation refunds only that recorded charge, at most once, in the same transaction as the task change. Completion timestamps and dispatch/executor nonces alone never authorize a refund. Pre-upgrade and imported tasks have no positive charge record: known outcomes can still settle, but historical statistics are not decremented by inference. Unknown outcomes remain held and unresolved. +If a modern warmup task becomes dead-lettered after its definitive failure and refund were committed, the matching authoritative result repairs only task and dead-letter metadata. It does not repeat accounting, callbacks or mailbox hold changes. Already-applied failures without positive refunded-charge proof remain protected. + ## Evidence-based repair holds Authentication, permanent and conflicting-result holds do not clear merely because time passed, a reconnect occurred or an unrelated send succeeded. An organization-authorized operator can use `send_recovery_resolution` in the existing mailbox PATCH after investigating the current held task and reason: diff --git a/internal/app/consumer/warmup_charge_provenance_live_test.go b/internal/app/consumer/warmup_charge_provenance_live_test.go index aac8d657c..a49e133cd 100644 --- a/internal/app/consumer/warmup_charge_provenance_live_test.go +++ b/internal/app/consumer/warmup_charge_provenance_live_test.go @@ -1,12 +1,15 @@ package jobs import ( + "context" + "encoding/json" "testing" "time" "github.com/google/uuid" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" ) func TestLiveWarmupGenericCompletionDoesNotProveChargedAccounting(t *testing.T) { @@ -46,3 +49,113 @@ func TestLiveWarmupGenericCompletionDoesNotProveChargedAccounting(t *testing.T) }) } } + +func TestLiveAppliedWarmupFailureRepairsOnlyRefundedChargeDeadLetters(t *testing.T) { + for _, source := range []string{"event", "recorded", "rollback"} { + for _, proof := range []string{"refunded", "unrefunded", "legacy", "ambiguous"} { + t.Run(source+"/"+proof, func(t *testing.T) { + s, h := liveWarmupService(t) + f := newWarmupFixture(t, h) + exec := func(sql string, args ...any) { + t.Helper() + if _, err := h.Exec(t.Context(), sql, args...); err != nil { + t.Fatal(err) + } + } + for range 3 { + if err := s.WarmupRepo.IncrementDailyCount(t.Context(), f.sender, time.Now()); err != nil { + t.Fatal(err) + } + } + if proof != "legacy" { + exec(`INSERT INTO warmup_tasks(task_id,warmup_charged_date,warmup_reply_charged) VALUES($1,(NOW() AT TIME ZONE 'UTC')::date,false)`, f.task) + } + result := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendRetry, ObservedAt: time.Now().UTC()}}} + if proof == "unrefunded" { + exec(`UPDATE tasks SET status='failed',send_result_state='failed',send_result_applied_at=NOW() WHERE id=$1`, f.task) + } else if err := s.HandleEmailFailed(t.Context(), result); err != nil { + t.Fatal(err) + } + var applied time.Time + var released, refunded *time.Time + if err := h.QueryRow(t.Context(), `SELECT t.send_result_applied_at,t.send_released_at,w.warmup_refunded_at FROM tasks t LEFT JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&applied, &released, &refunded); err != nil { + t.Fatal(err) + } + if err := s.TaskRepo.UpdateTaskStatus(t.Context(), f.task, "dead_lettered"); err != nil { + t.Fatal(err) + } + exec(`INSERT INTO task_dead_letters(task_id,task_type,payload,last_error,attempts,next_retry_at) VALUES($1,'warmup','{"task_id":"preserved"}','late dispatcher failure',2,NOW())`, f.task) + exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='authentication',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task) + if proof == "ambiguous" { + result.Error.Failure.Disposition = errx.SendAmbiguous + } + raw, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + exec(`UPDATE tasks SET send_executor_result=$2 WHERE id=$1`, f.task, raw) + if source == "recorded" { + reader := s.TaskRepo.(interface { + ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error) + }) + results, err := reader.ListUnappliedSendResults(t.Context(), 200) + if err != nil { + t.Fatal(err) + } + found := false + for _, stored := range results { + if stored.TaskID == f.task { + found = true + result = stored + } + } + if found != (proof == "refunded") { + t.Fatalf("recorded terminal repair selection=%v, proof=%s", found, proof) + } + } + callbacks := 0 + if source == "rollback" && proof == "refunded" { + constraint := "charge_repair_" + uuid.New().String()[:8] + exec(`ALTER TABLE task_dead_letters ADD CONSTRAINT ` + constraint + ` CHECK (task_id<>'` + f.task.String() + `'::uuid OR status<>'resolved') NOT VALID`) + t.Cleanup(func() { + if _, err := h.Exec(context.Background(), `ALTER TABLE task_dead_letters DROP CONSTRAINT IF EXISTS `+constraint); err != nil { + t.Error(err) + } + }) + if err := s.TaskRepo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(context.Context) error { callbacks++; return nil }); err == nil { + t.Fatal("failed dead-letter write committed terminal repair") + } + var status, dlq string + if err := h.QueryRow(t.Context(), `SELECT t.status,d.status FROM tasks t JOIN task_dead_letters d ON d.task_id=t.id WHERE t.id=$1`, f.task).Scan(&status, &dlq); err != nil || status != "dead_lettered" || dlq != "pending" || callbacks != 0 { + t.Fatalf("partial terminal repair: status=%s dlq=%s callbacks=%d err=%v", status, dlq, callbacks, err) + } + exec(`ALTER TABLE task_dead_letters DROP CONSTRAINT ` + constraint) + } + for range 2 { + if err := s.TaskRepo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(context.Context) error { callbacks++; return nil }); err != nil { + t.Fatal(err) + } + } + var status, dlq string + var sent int + var unchanged, evidence bool + if err := h.QueryRow(t.Context(), `SELECT t.status,d.status,ws.emails_sent, + t.send_result_state='failed' AND t.send_result_applied_at=$2 AND t.send_released_at IS NOT DISTINCT FROM $3::timestamptz AND w.warmup_refunded_at IS NOT DISTINCT FROM $4::timestamptz, + d.payload='{"task_id":"preserved"}'::jsonb AND d.last_error='late dispatcher failure' AND d.attempts=2 AND d.replayed_at IS NULL AND e.send_recovery_hold AND e.send_recovery_reason='authentication' AND e.send_recovery_task_id=t.id + FROM tasks t LEFT JOIN warmup_tasks w ON w.task_id=t.id JOIN task_dead_letters d ON d.task_id=t.id JOIN email_accounts e ON e.id=t.email_account_id JOIN warmup_statistics ws ON ws.email_account_id=e.id AND ws.date=(NOW() AT TIME ZONE 'UTC')::date WHERE t.id=$1`, f.task, applied, released, refunded).Scan(&status, &dlq, &sent, &unchanged, &evidence); err != nil { + t.Fatal(err) + } + wantStatus, wantDLQ, wantSent := "dead_lettered", "pending", 3 + if proof == "refunded" { + wantStatus, wantDLQ = "failed", "resolved" + } + if proof == "refunded" || proof == "ambiguous" { + wantSent = 2 + } + if status != wantStatus || dlq != wantDLQ || sent != wantSent || callbacks != 0 || !unchanged || !evidence { + t.Fatalf("terminal repair status=%s dlq=%s sent=%d callbacks=%d unchanged=%v evidence=%v", status, dlq, sent, callbacks, unchanged, evidence) + } + }) + } + } +} diff --git a/internal/repository/pg_send_recovery.go b/internal/repository/pg_send_recovery.go index 6717664b1..123ff10a2 100644 --- a/internal/repository/pg_send_recovery.go +++ b/internal/repository/pg_send_recovery.go @@ -14,6 +14,14 @@ import ( ) type sendResultKey struct{} + +const refundedWarmupChargePredicate = `t.task_type='warmup' AND EXISTS(SELECT 1 FROM warmup_tasks charge + WHERE charge.task_id=t.id AND charge.warmup_charged_date IS NOT NULL + AND charge.warmup_reply_charged IS NOT NULL AND charge.warmup_refunded_at IS NOT NULL)` + +const repairableSendDeadLetterPredicate = `t.status='dead_lettered' AND (t.send_result_state='sent' + OR (t.send_result_state='failed' AND ` + refundedWarmupChargePredicate + `))` + type sendResultContext struct { tx pgx.Tx taskID uuid.UUID @@ -190,7 +198,18 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send } return tx.Commit(ctx) } - if applied != nil && status != "dead_lettered" && previous != nil && state == *previous && state != "unknown" { + if applied != nil && previous != nil && state == *previous && state != "unknown" { + if status == "dead_lettered" { + // Repair metadata only; the positive refund proof forbids repeating accounting. + tag, err := tx.Exec(ctx, `UPDATE tasks t SET status='failed',updated_at=NOW() WHERE t.id=$1 AND t.status='dead_lettered' + AND t.send_result_state='failed' AND t.send_result_applied_at IS NOT NULL AND `+refundedWarmupChargePredicate, result.TaskID) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return nil + } + } if err = resolveSendDeadLetter(ctx, tx, result.TaskID); err != nil { return err } diff --git a/internal/repository/unknown_send_resolution.go b/internal/repository/unknown_send_resolution.go index 970ef7acc..6a063f890 100644 --- a/internal/repository/unknown_send_resolution.go +++ b/internal/repository/unknown_send_resolution.go @@ -92,7 +92,7 @@ func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int rows, err := r.db.Query(ctx, `WITH recorded AS ( SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,t.send_executor_result AS result,1 AS priority FROM tasks t WHERE t.send_executor_result IS NOT NULL AND t.status<>'cancelled' - AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent')) + AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`)) UNION ALL SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,t.send_executor_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL @@ -100,14 +100,14 @@ func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int UNION ALL SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM warmup_tasks w JOIN tasks t ON t.id=w.task_id WHERE w.dispatch_result IS NOT NULL AND t.status<>'cancelled' - AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent')) + AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`)) UNION ALL SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id JOIN warmup_tasks w ON w.task_id=t.id WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL AND w.dispatch_result IS NOT NULL AND t.send_result_applied_at IS NOT NULL AND t.status<>'cancelled' UNION ALL SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,r.confirmed_result,2 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id - WHERE r.confirmed_result IS NOT NULL AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent')) AND t.status<>'cancelled' + WHERE r.confirmed_result IS NOT NULL AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`)) AND t.status<>'cancelled' ), definitive AS (SELECT *,CASE WHEN result->>'success'='true' THEN 'sent' ELSE 'failed' END AS outcome FROM recorded WHERE result->>'success'='true' OR result->'error'->'failure'->>'disposition' IN('retry','throttle','permanent','authentication') @@ -115,7 +115,7 @@ func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int ('RECIPIENT_REJECTED','SEND_REJECTED','DOMAIN_AUTH_REJECTED','AUTHENTICATION_FAILED','INVALID_CREDENTIALS','GOOGLE_AUTHENTICATION_FAILED','SENDING_TOO_FAST','QUOTA_EXCEEDED','UNSUPPORTED'))), pending AS (SELECT DISTINCT ON(id) id,created_at,result FROM definitive WHERE send_result_applied_at IS NULL OR outcome IS DISTINCT FROM send_result_state - OR EXISTS(SELECT 1 FROM tasks t WHERE t.id=definitive.id AND t.status='dead_lettered' AND t.send_result_state='sent') ORDER BY id,priority) + OR EXISTS(SELECT 1 FROM tasks t WHERE t.id=definitive.id AND `+repairableSendDeadLetterPredicate+`) ORDER BY id,priority) SELECT id,result FROM pending ORDER BY created_at LIMIT $1`, limit) if err != nil { return nil, err