feat: reconcile persisted worker send receipts independently of broker replay while retaining ambiguous outcome holds

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

.github/workflows/ci.yml

@@ -204,6 +204,7 @@ jobs:
         run: |
           go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement)' -count=1
           go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults)|WarmupSenderDistinguishes)' -count=1
+          go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1
 
       - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL)
         env:

internal/app/consumer/event_send_result_live_test.go

@@ -2,6 +2,7 @@ package jobs
 
 import (
 	"context"
+	"encoding/json"
 	"os"
 	"testing"
 	"time"
@@ -15,6 +16,93 @@ import (
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
+func TestLiveStoredSendResultRecoveryWithoutBrokerDelivery(t *testing.T) {
+	for _, outcome := range []string{"sent", "failed", "ambiguous", "mismatched", "not started", "no nonce", "cancelled", "deleted worker"} {
+		t.Run(outcome, func(t *testing.T) {
+			s, d := liveWarmupService(t)
+			f := newWarmupFixture(t, d)
+			worker := uuid.New()
+			if _, err := d.Exec(t.Context(), `INSERT INTO fleet_nodes(id,role,name,active) VALUES($1,'worker','result recovery test',true)`, worker); err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _, _ = d.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+			token := f.mintToken(t, s.WarmupRepo)
+			result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: f.sentMsgID}
+			if outcome == "mismatched" {
+				result.TaskID = uuid.New()
+			}
+			if outcome == "failed" || outcome == "ambiguous" {
+				result.Success = false
+				disposition := errx.SendPermanent
+				if outcome == "ambiguous" {
+					disposition = errx.SendAmbiguous
+				}
+				result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: disposition, ObservedAt: time.Now()}}
+			}
+			raw, err := json.Marshal(result)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if _, err := d.Exec(t.Context(), `UPDATE tasks SET send_reserved_at=NOW()-INTERVAL '1 hour',send_result_state='unknown',send_executor_nonce=$2,send_executor_worker=$3,send_executor_started_at=NOW(),send_executor_result=$4 WHERE id=$1`, f.task, uuid.New(), worker, raw); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := d.Exec(t.Context(), `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if outcome == "not started" {
+				if _, err := d.Exec(t.Context(), `UPDATE tasks SET send_executor_started_at=NULL WHERE id=$1`, f.task); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if outcome == "no nonce" {
+				if _, err := d.Exec(t.Context(), `UPDATE tasks SET send_executor_nonce=NULL WHERE id=$1`, f.task); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if outcome == "deleted worker" {
+				if _, err := d.Exec(t.Context(), `DELETE FROM fleet_nodes WHERE id=$1`, worker); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if outcome == "cancelled" {
+				if _, err := d.Exec(t.Context(), `UPDATE tasks SET status='cancelled' WHERE id=$1`, f.task); err != nil {
+					t.Fatal(err)
+				}
+			}
+			_, err = s.recoverStoredSendResults(t.Context(), uuid.Nil)
+			if (outcome == "mismatched") != (err != nil) {
+				t.Fatalf("unexpected replay result: %v", err)
+			}
+			var state string
+			var applied, released, hold bool
+			if err := d.QueryRow(t.Context(), `SELECT t.send_result_state,t.send_result_applied_at IS NOT NULL,t.send_released_at IS NOT NULL,e.send_recovery_hold FROM tasks t JOIN email_accounts e ON e.id=t.email_account_id WHERE t.id=$1`, f.task).Scan(&state, &applied, &released, &hold); err != nil {
+				t.Fatal(err)
+			}
+			switch outcome {
+			case "sent", "deleted worker":
+				if state != "sent" || !applied || released || hold {
+					t.Fatalf("confirmed send not recovered: %s applied=%v released=%v hold=%v", state, applied, released, hold)
+				}
+				if err := s.HandleEmailSent(t.Context(), result); err != nil {
+					t.Fatalf("later broker result not idempotent: %v", err)
+				}
+				verified, err := s.WarmupRepo.FindWarmupToken(t.Context(), token)
+				if err != nil || verified.SentMessageID != f.sentMsgID {
+					t.Fatalf("worker receipt not applied to warmup: %+v %v", verified, err)
+				}
+			case "failed":
+				if state != "failed" || !applied || !released || !hold {
+					t.Fatalf("definite failure not reconciled: %s applied=%v released=%v hold=%v", state, applied, released, hold)
+				}
+			default:
+				if state != "unknown" || applied || released || !hold {
+					t.Fatalf("unsafe outcome released: %s applied=%v released=%v hold=%v", state, applied, released, hold)
+				}
+			}
+		})
+	}
+}
+
 // Live checks of the send-outcome loop against a real Postgres. Skipped unless
 // WARMBLY_TEST_DB is set (same convention as internal/scheduler):
 //

internal/app/consumer/send_result_effects.go

@@ -2,11 +2,72 @@ package jobs
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/rs/zerolog/log"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
 )
 
+func (s *JobsService) runStoredSendResults(ctx context.Context) {
+	if _, ok := s.TaskRepo.(repository.StoredSendResultRepository); !ok {
+		return
+	}
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	var cursor uuid.UUID
+	for {
+		pass, cancel := context.WithTimeout(ctx, 30*time.Second)
+		var err error
+		cursor, err = s.recoverStoredSendResults(pass, cursor)
+		if err != nil {
+			log.Warn().Msg("stored send result reconciliation deferred")
+		}
+		cancel()
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+func (s *JobsService) recoverStoredSendResults(ctx context.Context, after uuid.UUID) (uuid.UUID, error) {
+	results, err := s.TaskRepo.(repository.StoredSendResultRepository).ListPendingSendResults(ctx, after, 200)
+	if err != nil {
+		return after, err
+	}
+	if len(results) == 0 {
+		return uuid.Nil, nil
+	}
+	var first error
+	for _, stored := range results {
+		if err := ctx.Err(); err != nil {
+			return after, err
+		}
+		after = stored.TaskID
+		var result models.SendEmailResult
+		err := json.Unmarshal(stored.Payload, &result)
+		if err == nil && (result.TaskID == uuid.Nil || result.TaskID != stored.TaskID) {
+			err = errors.New("stored send result task mismatch")
+		}
+		if err == nil {
+			if result.Success {
+				err = s.HandleEmailSent(ctx, result)
+			} else {
+				err = s.HandleEmailFailed(ctx, result)
+			}
+		}
+		if err != nil && first == nil {
+			first = err
+		}
+	}
+	return after, first
+}
+
 func (s *JobsService) runSendResultEffects(ctx context.Context) {
 	outbox, ok := s.AdvancedService.(interface{ DeliverSendResultEffects(context.Context) error })
 	if !ok {

internal/app/consumer/send_result_effects_test.go

@@ -0,0 +1,96 @@
+package jobs
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+type storedSendResultStub struct {
+	repository.TaskRepository
+	repository.SendResultRecovery
+	list    func(context.Context, uuid.UUID, int) ([]repository.StoredSendResult, error)
+	applied []models.SendEmailResult
+	fail    uuid.UUID
+}
+
+func (r *storedSendResultStub) ListPendingSendResults(ctx context.Context, after uuid.UUID, limit int) ([]repository.StoredSendResult, error) {
+	return r.list(ctx, after, limit)
+}
+
+func (r *storedSendResultStub) ApplySendResult(_ context.Context, result models.SendEmailResult, _ func(context.Context) error) error {
+	r.applied = append(r.applied, result)
+	if result.TaskID == r.fail {
+		return errors.New("transient apply failure")
+	}
+	return nil
+}
+
+func TestStoredSendResultRecoveryContinuesPastInvalidAndFailedResults(t *testing.T) {
+	malformed, mismatch, failed, sent := uuid.New(), uuid.New(), uuid.New(), uuid.New()
+	payload := func(id uuid.UUID, success bool) json.RawMessage {
+		raw, err := json.Marshal(models.SendEmailResult{TaskID: id, Success: success})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return raw
+	}
+	r := &storedSendResultStub{fail: failed}
+	r.list = func(_ context.Context, after uuid.UUID, limit int) ([]repository.StoredSendResult, error) {
+		if limit != 200 {
+			t.Fatalf("unbounded replay: %d", limit)
+		}
+		if after == sent {
+			return nil, nil
+		}
+		return []repository.StoredSendResult{
+			{TaskID: malformed, Payload: json.RawMessage(`{"success":`)},
+			{TaskID: mismatch, Payload: payload(uuid.New(), true)},
+			{TaskID: failed, Payload: payload(failed, false)},
+			{TaskID: sent, Payload: payload(sent, true)},
+		}, nil
+	}
+	s := &JobsService{TaskRepo: r}
+	after, err := s.recoverStoredSendResults(t.Context(), uuid.Nil)
+	if err == nil || after != sent || len(r.applied) != 2 || r.applied[0].TaskID != failed || r.applied[1].TaskID != sent {
+		t.Fatalf("one failed result starved the rest or an invalid result was applied: %s %v %+v", after, err, r.applied)
+	}
+	if r.applied[0].Success || !r.applied[1].Success {
+		t.Fatal("result authority changed during replay")
+	}
+	after, err = s.recoverStoredSendResults(t.Context(), after)
+	if err != nil || after != uuid.Nil {
+		t.Fatalf("replay cursor did not wrap: %s %v", after, err)
+	}
+}
+
+func TestStoredSendResultRecoveryRetainsCursorOnReadFailureOrCancellation(t *testing.T) {
+	for _, cancelled := range []bool{false, true} {
+		cursor := uuid.New()
+		ctx, cancel := context.WithCancel(t.Context())
+		r := &storedSendResultStub{}
+		r.list = func(_ context.Context, after uuid.UUID, _ int) ([]repository.StoredSendResult, error) {
+			if after != cursor {
+				t.Fatal("cursor not forwarded")
+			}
+			if cancelled {
+				return []repository.StoredSendResult{{TaskID: uuid.New()}}, nil
+			}
+			return nil, errors.New("database unavailable")
+		}
+		if cancelled {
+			cancel()
+		}
+		s := &JobsService{TaskRepo: r}
+		after, err := s.recoverStoredSendResults(ctx, cursor)
+		cancel()
+		if err == nil || after != cursor || len(r.applied) != 0 {
+			t.Fatalf("unfinished batch skipped: %s %v %+v", after, err, r.applied)
+		}
+	}
+}

internal/app/consumer/service.go

@@ -127,6 +127,7 @@ func (s *JobsService) Start(ctx context.Context) {
 	// timestamps.
 	go s.sweepFollowUps(ctx)
 	go s.runSendResultEffects(ctx)
+	go s.runStoredSendResults(ctx)
 	go s.runArrivalOutbox(ctx)
 
 	if err := s.Bus.Subscribe(ctx, []string{kafka.TopicWorkerEvents}, "consumer-group", s.receive); err != nil {

internal/infrastructure/db/migrations/000280_stored_send_result_replay.down.sql

@@ -0,0 +1 @@
+DROP INDEX tasks_pending_executor_results;

internal/infrastructure/db/migrations/000280_stored_send_result_replay.up.sql

@@ -0,0 +1,2 @@
+CREATE INDEX tasks_pending_executor_results ON tasks(id)
+    WHERE send_executor_result IS NOT NULL AND send_result_applied_at IS NULL;

internal/repository/pg_send_recovery.go

@@ -66,6 +66,37 @@ type SendResultRecovery interface {
 	GetSendAdmission(context.Context, uuid.UUID, uuid.UUID, models.InboxProvider, time.Time) (*SendAdmission, error)
 }
 
+type StoredSendResult struct {
+	TaskID  uuid.UUID
+	Payload json.RawMessage
+}
+
+type StoredSendResultRepository interface {
+	SendResultRecovery
+	ListPendingSendResults(context.Context, uuid.UUID, int) ([]StoredSendResult, error)
+}
+
+func (r *taskRepository) ListPendingSendResults(ctx context.Context, after uuid.UUID, limit int) ([]StoredSendResult, error) {
+	rows, err := r.db.Query(ctx, `SELECT id,send_executor_result FROM tasks
+		WHERE id>$1 AND send_executor_result IS NOT NULL AND send_executor_started_at IS NOT NULL
+		AND send_executor_nonce IS NOT NULL
+		AND send_result_applied_at IS NULL AND (send_result_state IS NULL OR send_result_state='unknown')
+		AND status<>'cancelled' ORDER BY id LIMIT $2`, after, min(max(limit, 1), 200))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var results []StoredSendResult
+	for rows.Next() {
+		var result StoredSendResult
+		if err := rows.Scan(&result.TaskID, &result.Payload); err != nil {
+			return nil, err
+		}
+		results = append(results, result)
+	}
+	return results, rows.Err()
+}
+
 type SendAdmission struct {
 	MailboxID    uuid.UUID
 	Provider     models.InboxProvider