From d220fc404722d4e8b295f2c5fdee8c7f3886ca3e Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 03:59:29 +0000 Subject: [PATCH] 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 | 1 + .../consumer/event_send_result_live_test.go | 88 +++++++++++++++++ internal/app/consumer/send_result_effects.go | 61 ++++++++++++ .../app/consumer/send_result_effects_test.go | 96 +++++++++++++++++++ internal/app/consumer/service.go | 1 + .../000280_stored_send_result_replay.down.sql | 1 + .../000280_stored_send_result_replay.up.sql | 2 + internal/repository/pg_send_recovery.go | 31 ++++++ 8 files changed, 281 insertions(+) create mode 100644 internal/app/consumer/send_result_effects_test.go create mode 100644 internal/infrastructure/db/migrations/000280_stored_send_result_replay.down.sql create mode 100644 internal/infrastructure/db/migrations/000280_stored_send_result_replay.up.sql diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b91966c2..afaecfff9 100644 --- a/.github/workflows/ci.yml +++ b/.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: diff --git a/internal/app/consumer/event_send_result_live_test.go b/internal/app/consumer/event_send_result_live_test.go index f75ddfd30..54c3a7424 100644 --- a/internal/app/consumer/event_send_result_live_test.go +++ b/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): // diff --git a/internal/app/consumer/send_result_effects.go b/internal/app/consumer/send_result_effects.go index 489a1178b..ed963ff71 100644 --- a/internal/app/consumer/send_result_effects.go +++ b/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 { diff --git a/internal/app/consumer/send_result_effects_test.go b/internal/app/consumer/send_result_effects_test.go new file mode 100644 index 000000000..51ec0147c --- /dev/null +++ b/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) + } + } +} diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index 69e7afd7b..debe17f0d 100644 --- a/internal/app/consumer/service.go +++ b/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 { diff --git a/internal/infrastructure/db/migrations/000280_stored_send_result_replay.down.sql b/internal/infrastructure/db/migrations/000280_stored_send_result_replay.down.sql new file mode 100644 index 000000000..efc2f9b22 --- /dev/null +++ b/internal/infrastructure/db/migrations/000280_stored_send_result_replay.down.sql @@ -0,0 +1 @@ +DROP INDEX tasks_pending_executor_results; diff --git a/internal/infrastructure/db/migrations/000280_stored_send_result_replay.up.sql b/internal/infrastructure/db/migrations/000280_stored_send_result_replay.up.sql new file mode 100644 index 000000000..a2cd7a79e --- /dev/null +++ b/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; diff --git a/internal/repository/pg_send_recovery.go b/internal/repository/pg_send_recovery.go index 4eb54f07f..62be4b0d7 100644 --- a/internal/repository/pg_send_recovery.go +++ b/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