feat: integrate hosted accounting regressions and proof-gated failed dead-letter repair with merged provider fixes

.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/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)
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable

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:

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)
+				}
+			})
+		}
+	}
+}

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
 			}

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