feat: keep deferred warmup commands free of terminal send results and recover existing never-started preparation conflicts without releasing real delivery holds

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -202,8 +202,8 @@ jobs:
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
         run: |
-          go test -race ./internal/repository -run '^TestLive(LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions)' -count=1
-          go test -race ./internal/tasks -run '^TestLiveWarmup(SafetyHoldReschedules|Recovery)' -count=1
+          go test -race ./internal/repository -run '^TestLive(LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains)' -count=1
+          go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults)|WarmupSenderDistinguishes)' -count=1
 
       - name: Test the Kafka codec and client lifecycle
         run: go test -race -tags kafka ./internal/infrastructure/codec/ ./internal/infrastructure/kafka/ ./internal/infrastructure/eventbus/

docs/content/docs/guides/warmup.mdx

@@ -37,6 +37,8 @@ Persistent loading failures and pending warmup tasks at least an hour past their
 
 A missing mailbox also requests another reload through the worker heartbeat. Existing warmup sends stuck before execution recover automatically: after ten minutes, the scheduler retires a nonce-protected command only when neither worker execution nor a delivery receipt exists, refunds its reserved count, and resumes the mailbox's normal schedule. Delayed copies of the retired command cannot send. Customers do not need to disconnect or re-add their mailboxes. Actually started sends with missing outcomes, conflicting outcomes, provider cooldowns, authentication failures and other safety holds are not cleared by this repair.
 
+If outbound admission refuses a warmup command before publication, a successful deferral returns it to the schedule without recording a failed send. A deferral whose database outcome is uncertain is not treated as proof of failure. Existing records affected by an older preparation-result race also recover automatically, but only when the recorded internal preparation failure predates the new authorization and there is no execution, outbound reservation, provider-confirmed send or receipt. Only that provably unstarted command's false conflict is released; genuine conflicting outcomes remain held.
+
 Follow a provider-specific error when one is shown. Google and Microsoft sign-ins use their native APIs, not SMTP settings. Reconnect those mailboxes only for a reported sign-in or permission error. SMTP host, port, password and IP allowlist checks apply to SMTP mailboxes when the error reports a mail-server connection problem. Warmbly Cloud makes those connections from its own network, not your self-hosted instance's address.
 
 Recovery does not bypass paused participation, provider limits or deliverability holds. If the warning persists, give support the mailbox address and the failure time so it can be matched to the worker's loading and sending records.

internal/repository/warmup_dispatch_recovery.go

@@ -2,15 +2,21 @@ package repository
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/warmbly/warmbly/internal/models"
 )
 
 const unstartedWarmupDispatch = `t.task_type='warmup' AND t.status='completed'
-	AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL
-	AND t.send_result_evidence IS 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_result_evidence IS NULL AND t.send_released_at IS NULL)
+	 OR (t.send_executor_nonce IS NULL AND t.send_reserved_at IS NULL AND w.dispatch_nonce IS NOT NULL
+	     AND t.send_result_applied_at<t.completed_at AND t.send_released_at<t.completed_at
+	     AND t.send_result_evidence->>'protocol'='internal' AND t.send_result_evidence->>'stage'='prepare'
+	     AND t.send_result_evidence->>'disposition'='retry' AND t.send_result_evidence->>'scope'='mailbox'))
 	AND COALESCE(t.send_reserved_at,t.completed_at)<$1
 	AND (t.send_executor_nonce IS NOT NULL OR w.dispatch_nonce IS NOT NULL)
 	AND t.send_executor_started_at IS NULL AND t.send_executor_result IS NULL
@@ -63,6 +69,15 @@ func (r *taskRepository) retireUnstartedWarmupDispatch(ctx context.Context, task
 	if err = lockSend(ctx, tx, task, mailbox); err != nil {
 		return false, err
 	}
+	var stalePreparation bool
+	err = tx.QueryRow(ctx, `SELECT t.send_result_applied_at IS NOT NULL FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id
+		WHERE t.id=$2 AND t.email_account_id=$3 AND `+unstartedWarmupDispatch, before, task, mailbox).Scan(&stalePreparation)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
 	tag, err := tx.Exec(ctx, `UPDATE tasks t SET status='cancelled',updated_at=NOW(),send_result_state='failed',send_result_applied_at=NOW(),send_released_at=NOW()
 		FROM warmup_tasks w WHERE w.task_id=t.id AND t.id=$2 AND t.email_account_id=$3 AND `+unstartedWarmupDispatch, before, task, mailbox)
 	if err != nil {
@@ -80,6 +95,12 @@ func (r *taskRepository) retireUnstartedWarmupDispatch(ctx context.Context, task
 	if _, err = tx.Exec(ctx, `DELETE FROM warmup_tokens WHERE task_id=$1`, task); err != nil {
 		return false, err
 	}
+	if stalePreparation {
+		if _, err = tx.Exec(ctx, `UPDATE email_accounts SET send_recovery_reason='unknown'
+			WHERE id=$1 AND send_recovery_task_id=$2 AND send_recovery_reason='conflict'`, mailbox, task); err != nil {
+			return false, err
+		}
+	}
 	if err = persistSendHold(ctx, tx, mailbox, models.SendEmailResult{TaskID: task}, "failed"); err != nil {
 		return false, err
 	}

internal/repository/warmup_preparation_recovery_live_test.go

@@ -0,0 +1,143 @@
+package repository
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func preparationRecoveryFixture(t *testing.T) (*poolLinkFixture, *taskRepository, uuid.UUID, uuid.UUID) {
+	t.Helper()
+	f, r := lineageFixture(t)
+	ctx := t.Context()
+	worker := uuid.New()
+	queries := []struct {
+		query string
+		args  []any
+	}{
+		{`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol) VALUES($1,'worker',true,NOW(),1)`, []any{worker}},
+		{`INSERT INTO workers(id) VALUES($1)`, []any{worker}},
+		{`UPDATE email_accounts SET worker_id=$1,test_mode='legacy' WHERE id=$2`, []any{worker, f.sender}},
+		{`INSERT INTO warmup_tasks(task_id,target_account_id,lineage_version) VALUES($1,$2,1)`, []any{f.task, f.recipient}},
+		{`UPDATE tasks SET status='active' WHERE id=$1`, []any{f.task}},
+	}
+	for _, q := range queries {
+		if _, err := f.pool.Exec(ctx, q.query, q.args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	t.Cleanup(func() { _, _ = f.pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+	f.token(t, "", "Preparation recovery", false)
+	nonce, err := r.AuthorizeWarmupDispatch(ctx, f.task, f.sender, worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return f, r, worker, nonce
+}
+
+func TestLiveWarmupRecoversOldPreparationResultAfterDeferral(t *testing.T) {
+	f, r, worker, nonce := preparationRecoveryFixture(t)
+	ctx := t.Context()
+	if err := r.DeferWarmupDispatch(ctx, f.task, f.sender, worker, nonce, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	result := models.SendEmailResult{TaskID: f.task, SentAt: time.Now().UTC(), Error: &models.EmailSendError{Failure: &errx.SendFailure{
+		Protocol: "internal", Stage: "prepare", Disposition: errx.SendRetry, Scope: "mailbox", ObservedAt: time.Now().UTC(),
+	}}}
+	// Reproduce the released backend's result write after the task was requeued.
+	apply := func(ctx context.Context) error {
+		return f.warmup.FailWarmupSend(ctx, f.sender, f.task, result.SentAt, "Send not published", "Worker authority unavailable")
+	}
+	if err := r.ApplySendResult(ctx, result, apply); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.pool.Exec(ctx, `UPDATE email_accounts SET send_cooldown_until=NULL WHERE id=$1`, f.sender); err != nil {
+		t.Fatal(err)
+	}
+	if claimed, err := r.ClaimWarmupTask(ctx, f.task, time.Now().Add(time.Second)); err != nil || !claimed {
+		t.Fatalf("claim=%v error=%v", claimed, err)
+	}
+	f.token(t, "", "Preparation recovery", false)
+	if _, err := r.AuthorizeWarmupDispatch(ctx, f.task, f.sender, worker); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplySendResult(ctx, result, apply); err != nil {
+		t.Fatal(err)
+	}
+	var reason string
+	if err := f.pool.QueryRow(ctx, `SELECT send_recovery_reason FROM email_accounts WHERE id=$1`, f.sender).Scan(&reason); err != nil || reason != "conflict" {
+		t.Fatalf("old backend race not reproduced: reason=%s error=%v", reason, err)
+	}
+	var wg sync.WaitGroup
+	var recovered atomic.Int32
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			retired, err := r.retireUnstartedWarmupDispatch(ctx, f.task, f.sender, time.Now().Add(time.Second))
+			if err != nil {
+				t.Error(err)
+			} else if retired {
+				recovered.Add(1)
+			}
+		}()
+	}
+	wg.Wait()
+	if recovered.Load() != 1 {
+		t.Fatalf("never-published command retired %d times", recovered.Load())
+	}
+	var held bool
+	var status string
+	var count, tokens int
+	if err := f.pool.QueryRow(ctx, `SELECT ea.send_recovery_hold,t.status,ws.emails_sent,
+		(SELECT count(*) FROM warmup_tokens WHERE task_id=t.id)
+		FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id JOIN warmup_statistics ws ON ws.email_account_id=ea.id AND ws.date=CURRENT_DATE
+		WHERE t.id=$1`, f.task).Scan(&held, &status, &count, &tokens); err != nil || held || status != "cancelled" || count != 0 || tokens != 0 {
+		t.Fatalf("recovery held=%v status=%s sent=%d tokens=%d error=%v", held, status, count, tokens, err)
+	}
+	state, err := r.InspectWarmupDispatch(ctx, f.task, f.sender, worker)
+	if err != nil || state.State != "denied" {
+		t.Fatalf("retired command replay: %+v %v", state, err)
+	}
+}
+
+func TestLiveWarmupPreparationRecoveryRetainsExecutionAndProviderEvidence(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		query string
+	}{
+		{"worker began", `UPDATE warmup_tasks SET dispatch_started_at=NOW() WHERE task_id=$1`},
+		{"executor began", `UPDATE tasks SET send_executor_started_at=NOW() WHERE id=$1`},
+		{"outbound nonce", `UPDATE tasks SET send_executor_nonce=gen_random_uuid() WHERE id=$1`},
+		{"outbound result", `UPDATE tasks SET send_executor_result='{"success":false}' WHERE id=$1`},
+		{"worker result", `UPDATE warmup_tasks SET dispatch_result='{"success":false}' WHERE task_id=$1`},
+		{"reserved outbound", `UPDATE tasks SET send_reserved_at=NOW()-INTERVAL '2 hours' WHERE id=$1`},
+		{"provider confirmed", `UPDATE warmup_tokens SET sent_message_id='accepted@example.test' WHERE task_id=$1`},
+		{"outbound attempt", `INSERT INTO outbound_attempts(nonce,task_id,email_account_id,provider,attempted_at,recipient_count) SELECT gen_random_uuid(),id,email_account_id,'smtp_imap',NOW(),1 FROM tasks WHERE 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,'confirmed@example.test',task_id FROM warmup_tokens WHERE task_id=$1`},
+		{"provider failure", `UPDATE tasks SET send_result_evidence=jsonb_set(send_result_evidence,'{protocol}','"smtp"') WHERE id=$1`},
+		{"current preparation", `UPDATE tasks SET send_result_applied_at=NOW() WHERE id=$1`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f, r, _, _ := preparationRecoveryFixture(t)
+			ctx := t.Context()
+			if _, err := f.pool.Exec(ctx, `UPDATE tasks SET completed_at=NOW()-INTERVAL '1 hour',
+				send_result_applied_at=NOW()-INTERVAL '2 hours',send_released_at=NOW()-INTERVAL '2 hours',
+				send_result_evidence='{"protocol":"internal","stage":"prepare","disposition":"retry","scope":"mailbox"}' WHERE id=$1`, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := f.pool.Exec(ctx, tc.query, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if retired, err := r.retireUnstartedWarmupDispatch(ctx, f.task, f.sender, time.Now()); err != nil || retired {
+				t.Fatalf("protected send retired=%v error=%v", retired, err)
+			}
+		})
+	}
+}

internal/tasks/email_sender.go

@@ -72,6 +72,8 @@ var ErrWorkerUnconfirmed = errors.New("the mailbox's sending worker could not be
 // being offered again.
 var ErrSendDispatchUnknown = errors.New("the send was not confirmed as queued and may already be on its way to a worker")
 
+var ErrWarmupDispatchDeferred = errors.New("the warmup send was not published and has already been rescheduled")
+
 type emailSender struct {
 	emailRepo repository.EmailRepository
 	publisher events.Publisher
@@ -192,8 +194,9 @@ func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessa
 			if r, ok := s.admission.(repository.WarmupDispatchRepository); ok {
 				if n, nerr := uuid.Parse(msg.DispatchNonce); nerr == nil {
 					if derr := r.DeferWarmupDispatch(ctx, taskID, account.ID, *workerID, n, time.Now().Add(5*time.Minute)); derr != nil {
-						return derr
+						return fmt.Errorf("%w: defer warmup: %v", ErrSendDispatchUnknown, derr)
 					}
+					return fmt.Errorf("%w: %v", ErrWarmupDispatchDeferred, err)
 				}
 			}
 		}

internal/tasks/email_sender_deferral_test.go

@@ -0,0 +1,121 @@
+package tasks
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+	"github.com/warmbly/warmbly/internal/scheduler"
+	"github.com/warmbly/warmbly/internal/tasks/proto"
+)
+
+type deferredWarmupAdmission struct {
+	repository.OutboundAdmissionRepository
+	repository.WarmupDispatchRepository
+	err      error
+	deferred bool
+}
+
+func (a *deferredWarmupAdmission) ReserveOutbound(context.Context, repository.OutboundReservation) (uuid.UUID, error) {
+	return uuid.Nil, repository.ErrSendAdmissionDenied
+}
+
+func (a *deferredWarmupAdmission) DeferWarmupDispatch(context.Context, uuid.UUID, uuid.UUID, uuid.UUID, uuid.UUID, time.Time) error {
+	a.deferred = true
+	return a.err
+}
+
+func TestWarmupSenderDistinguishesRequeueFromUnknownDeferral(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		err  error
+		want error
+	}{
+		{"confirmed deferral", nil, ErrWarmupDispatchDeferred},
+		{"uncertain transaction", errors.New("commit response lost"), ErrSendDispatchUnknown},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			admission := &deferredWarmupAdmission{err: tc.err}
+			worker, org := uuid.New(), uuid.New()
+			sender := &emailSender{admission: admission}
+			err := sender.Send(t.Context(), uuid.New(), EmailMessage{
+				IsWarmup: true, DispatchNonce: uuid.NewString(), To: []string{"partner@example.test"},
+			}, models.Email{ID: uuid.New(), Email: "sender@example.test", WorkerID: &worker, OrganizationID: &org})
+			if !admission.deferred || !errors.Is(err, tc.want) {
+				t.Fatalf("deferred=%v error=%v want=%v", admission.deferred, err, tc.want)
+			}
+		})
+	}
+}
+
+type preparationDeferringSender struct {
+	EmailSender
+	repo  repository.WarmupDispatchRepository
+	calls int
+}
+
+func (*preparationDeferringSender) WarmupWorkerReady(context.Context, models.Email) (bool, error) {
+	return true, nil
+}
+
+func (s *preparationDeferringSender) Send(ctx context.Context, task uuid.UUID, msg EmailMessage, account models.Email) error {
+	nonce, err := uuid.Parse(msg.DispatchNonce)
+	if err != nil {
+		return err
+	}
+	if err := s.repo.DeferWarmupDispatch(ctx, task, account.ID, *account.WorkerID, nonce, time.Now().Add(5*time.Minute)); err != nil {
+		return err
+	}
+	s.calls++
+	return ErrWarmupDispatchDeferred
+}
+
+type preparationScheduler struct{ scheduler.SchedulerService }
+
+func (preparationScheduler) WarmupDispatchNotBefore(_ context.Context, _ uuid.UUID, now time.Time) (time.Time, error) {
+	return now, nil
+}
+
+func TestLiveWarmupTaskDoesNotApplyResultsAfterConfirmedDeferral(t *testing.T) {
+	f := newPartnerRoutingFixture(t)
+	ctx := t.Context()
+	worker, task := uuid.New(), uuid.New()
+	exec := func(query string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, query, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol) VALUES($1,'worker',true,NOW(),2)`, worker)
+	exec(`INSERT INTO workers(id) VALUES($1)`, worker)
+	t.Cleanup(func() { _, _ = f.pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+	exec(`UPDATE email_accounts SET worker_id=$1,warmup=NOW()-INTERVAL '30 days',warmup_days=127,warmup_pool_type='free' WHERE organization_id=$2`, worker, f.org)
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,scheduled_at) VALUES($1,$2,'warmup','pending','',NOW()-INTERVAL '1 minute')`, task, f.sender.ID)
+	exec(`INSERT INTO warmup_tasks(task_id) VALUES($1)`, task)
+	f.svc.taskRepo = repository.NewTaskRepository(f.pool)
+	f.svc.scheduler = preparationScheduler{}
+	sender := &preparationDeferringSender{repo: f.svc.taskRepo.(repository.WarmupDispatchRepository)}
+	f.svc.emailSender = sender
+	for i := 1; i <= 2; i++ {
+		if xerr := f.svc.HandleEmailTask(&proto.ProcessTask{TaskId: task.String()}); xerr != nil {
+			t.Fatal(xerr)
+		}
+		var status string
+		var clean, held bool
+		var reserved int
+		if err := f.pool.QueryRow(ctx, `SELECT t.status,
+			t.send_result_state IS NULL AND t.send_result_applied_at IS NULL AND t.send_released_at IS NULL AND t.send_result_evidence IS NULL,
+			ea.send_recovery_hold,ws.emails_sent FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id
+			JOIN warmup_statistics ws ON ws.email_account_id=ea.id AND ws.date=CURRENT_DATE WHERE t.id=$1`, task).Scan(&status, &clean, &held, &reserved); err != nil {
+			t.Fatal(err)
+		}
+		if sender.calls != i || status != "pending" || !clean || held || reserved != 0 {
+			t.Fatalf("attempt=%d calls=%d status=%s clean=%v held=%v reserved=%d", i, sender.calls, status, clean, held, reserved)
+		}
+		exec(`UPDATE tasks SET scheduled_at=NOW()-INTERVAL '1 minute' WHERE id=$1`, task)
+	}
+}

internal/tasks/email_task.go

@@ -408,7 +408,7 @@ func (s *tasksService) HandleEmailTask(task *proto.ProcessTask) *errx.Error {
 	}
 	emailMsg.DispatchNonce = nonce.String()
 	if err := s.emailSender.Send(ctx, taskID, emailMsg, *account); err != nil {
-		if !errors.Is(err, ErrSendDispatchUnknown) {
+		if !errors.Is(err, ErrSendDispatchUnknown) && !errors.Is(err, ErrWarmupDispatchDeferred) {
 			result := models.SendEmailResult{TaskID: taskID, Error: &models.EmailSendError{Failure: &errx.SendFailure{Protocol: "internal", Stage: "prepare", Disposition: errx.SendRetry, Scope: "mailbox", ObservedAt: time.Now().UTC()}}, SentAt: time.Now().UTC()}
 			recovery := s.taskRepo.(repository.SendResultRecovery)
 			if applyErr := recovery.ApplySendResult(ctx, result, func(ctx context.Context) error {