diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx index 2ec34e0e7..4db2522d4 100644 --- a/docs/content/docs/api/reference/deliverability-ops.mdx +++ b/docs/content/docs/api/reference/deliverability-ops.mdx @@ -292,13 +292,15 @@ Auth: **Scope** `SEND_CAMPAIGNS` ยท **Org permission** `send_campaigns` | Parameter | In | Type | Description | |-----------|----|------|-------------| -| `status` | query | string | Optional status filter (e.g. `pending`, `replayed`). | +| `status` | query | string | Optional status filter (e.g. `pending`, `replayed`, `resolved`). | | `limit` | query | integer | Max rows to return, 1 to 200 (default 100). | ### Response Returns a `data` array of dead-letter records. This endpoint is not cursor-paginated; it returns up to `limit` rows in one response. +Send dead letters become `resolved` only after an authoritative worker or operator-confirmed send outcome is reconciled. Recovery does not resend mail: the original payload, error and attempts stay available for inspection. Unknown outcomes remain pending and held. A missing Sent-folder copy, reservation marker or executor marker is not proof that nothing was sent. + ```json { "data": [ diff --git a/internal/app/consumer/event_send_result.go b/internal/app/consumer/event_send_result.go index eac98c6a6..4ae088e3e 100644 --- a/internal/app/consumer/event_send_result.go +++ b/internal/app/consumer/event_send_result.go @@ -186,12 +186,12 @@ func (s *JobsService) applyEmailFailed(ctx context.Context, result models.SendEm } } switch task.Status { - case "completed": + case "completed", "dead_lettered": // The normal case: stamped by the control plane, refused by the worker. case "active": return errSendResultEarly default: - // Already walked back (duplicate delivery), cancelled, or dead-lettered. + // Already walked back (duplicate delivery) or cancelled. return nil } diff --git a/internal/app/consumer/send_legacy_recovery_live_test.go b/internal/app/consumer/send_legacy_recovery_live_test.go index b179405e2..7f1720e4e 100644 --- a/internal/app/consumer/send_legacy_recovery_live_test.go +++ b/internal/app/consumer/send_legacy_recovery_live_test.go @@ -1,11 +1,15 @@ package jobs import ( + "context" "encoding/json" + "errors" "testing" "time" + "github.com/google/uuid" "github.com/iskorotkov/avro/v2" + "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" ) @@ -63,6 +67,257 @@ func legacySendResultSchema(t *testing.T) (avro.Schema, avro.Schema) { return writer, resolved } +func TestLiveLegacyFailedStatusCannotProveNonExecution(t *testing.T) { + for _, status := range []string{"failed", "dead_lettered"} { + t.Run(status, func(t *testing.T) { + h := liveDB(t) + f := newSendResultFixture(t, h) + s := liveJobsService(h) + id := f.stampSend(t, s) + if _, err := h.Exec(t.Context(), `UPDATE tasks SET status=$2::task_status,send_result_state='unknown' WHERE id=$1`, id, status); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.mailbox, id); err != nil { + t.Fatal(err) + } + reader := s.TaskRepo.(interface { + ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error) + }) + results, err := reader.ListUnappliedSendResults(t.Context(), 200) + if err != nil { + t.Fatal(err) + } + for _, result := range results { + if result.TaskID == id { + t.Fatal("task status and missing modern markers manufactured provider failure evidence") + } + } + ambiguous := models.SendEmailResult{TaskID: id, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendAmbiguous}}} + if err := s.HandleEmailFailed(t.Context(), ambiguous); err != nil { + t.Fatal(err) + } + var state string + var applied, released, hold bool + if err := h.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`, id).Scan(&state, &applied, &released, &hold); err != nil { + t.Fatal(err) + } + if state != "unknown" || applied || released || !hold { + t.Fatalf("legacy outcome guessed: state=%s applied=%t released=%t hold=%t", state, applied, released, hold) + } + }) + } +} + +func TestLiveAuthoritativeDeadLetterResultsReconcileWithoutReplay(t *testing.T) { + for _, outcome := range []string{"sent", "failed", "ambiguous", "previously applied sent", "previously applied failed", "unstamped failure", "legacy failure"} { + t.Run(outcome, func(t *testing.T) { + s, h := liveWarmupService(t) + f := newWarmupFixture(t, h) + token := f.mintToken(t, s.WarmupRepo) + var day time.Time + if err := h.QueryRow(t.Context(), `SELECT completed_at FROM tasks WHERE id=$1`, f.task).Scan(&day); err != nil { + t.Fatal(err) + } + if err := s.WarmupRepo.IncrementDailyCount(t.Context(), f.sender, day); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, f.task); err != nil { + t.Fatal(err) + } + if outcome != "legacy failure" { + if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_reserved_at=completed_at,send_executor_nonce=$2 WHERE id=$1`, f.task, uuid.New()); err != nil { + t.Fatal(err) + } + } + if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type,payload,last_error,attempts,next_retry_at) VALUES($1,'warmup','{"task_id":"preserved"}','legacy dispatch error',1,NOW())`, f.task); err != nil { + t.Fatal(err) + } + if _, err := h.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 == "unstamped failure" { + if _, err := h.Exec(t.Context(), `UPDATE tasks SET completed_at=NULL WHERE id=$1`, f.task); err != nil { + t.Fatal(err) + } + } + result := models.SendEmailResult{TaskID: f.task, Success: outcome == "sent" || outcome == "previously applied sent", MessageID: f.sentMsgID, SentAt: day} + if !result.Success { + disposition := errx.SendRetry + if outcome == "ambiguous" { + disposition = errx.SendAmbiguous + } + result.MessageID = "" + result.Error = &models.EmailSendError{Code: "SERVER_UNREACHABLE", Failure: &errx.SendFailure{Disposition: disposition, Stage: "dial", Protocol: "smtp", Scope: "mailbox"}} + } + if outcome == "previously applied sent" || outcome == "previously applied failed" { + raw, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + state := "failed" + if result.Success { + state = "sent" + } + if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_result_state=$2,send_result_applied_at=NOW(),send_executor_result=$3 WHERE id=$1`, f.task, state, raw); err != nil { + t.Fatal(err) + } + s.reconcileRecordedSendResults(t.Context()) + } + for range 2 { + var err error + if result.Success { + err = s.HandleEmailSent(t.Context(), result) + } else { + err = s.HandleEmailFailed(t.Context(), result) + } + if err != nil { + t.Fatal(err) + } + } + var status, state, dlq, message, lastError string + var applied, released, hold, replayed, evidenceKept bool + var count, attempts int + if err := h.QueryRow(t.Context(), `SELECT t.status,t.send_result_state,t.send_result_applied_at IS NOT NULL,t.send_released_at IS NOT NULL,e.send_recovery_hold,d.status,d.replayed_at IS NOT NULL,d.payload='{"task_id":"preserved"}'::jsonb,d.last_error,d.attempts,wt.sent_message_id,ws.emails_sent FROM tasks t JOIN email_accounts e ON e.id=t.email_account_id JOIN task_dead_letters d ON d.task_id=t.id JOIN warmup_tokens wt ON wt.task_id=t.id JOIN warmup_statistics ws ON ws.email_account_id=e.id AND ws.date=DATE($3::timestamptz) WHERE t.id=$1 AND wt.token=$2`, f.task, token, day).Scan(&status, &state, &applied, &released, &hold, &dlq, &replayed, &evidenceKept, &lastError, &attempts, &message, &count); err != nil { + t.Fatal(err) + } + if replayed || !evidenceKept || lastError != "legacy dispatch error" || attempts != 1 { + t.Fatal("recovery replayed or erased original dead-letter evidence") + } + switch outcome { + case "sent", "previously applied sent": + if status != "completed" || state != "sent" || !applied || released || hold || dlq != "resolved" || message != f.sentMsgID || count != 1 { + t.Fatalf("sent recovery incomplete: status=%s state=%s applied=%t released=%t hold=%t dlq=%s count=%d", status, state, applied, released, hold, dlq, count) + } + case "failed", "unstamped failure", "legacy failure": + wantCount := 0 + if outcome == "unstamped failure" || outcome == "legacy failure" { + wantCount = 1 + } + if status != "failed" || state != "failed" || !applied || !released || hold || dlq != "resolved" || message != "" || count != wantCount { + t.Fatalf("failure recovery incomplete: status=%s state=%s applied=%t released=%t hold=%t dlq=%s count=%d", status, state, applied, released, hold, dlq, count) + } + case "previously applied failed": + if status != "dead_lettered" || state != "failed" || !applied || released || !hold || dlq != "pending" || message != "" || count != 1 { + t.Fatal("already-applied failure refunded without accounting provenance") + } + default: + if status != "dead_lettered" || state != "unknown" || applied || released || !hold || dlq != "pending" || message != "" || count != 1 { + t.Fatal("ambiguous dead letter was guessed resolved") + } + } + }) + } +} + +func TestLiveDeadLetterResultEffectsAndSettlementRollBackTogether(t *testing.T) { + s, h := liveWarmupService(t) + f := newWarmupFixture(t, h) + f.mintToken(t, s.WarmupRepo) + if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, f.task); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'warmup')`, f.task); err != nil { + t.Fatal(err) + } + want := errors.New("effect failed after task transition") + result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: f.sentMsgID} + recovery := s.TaskRepo.(repository.SendResultRecovery) + err := recovery.ApplySendResult(t.Context(), result, func(ctx context.Context) error { + if err := s.applyEmailSent(ctx, result); err != nil { + return err + } + return want + }) + if !errors.Is(err, want) { + t.Fatal(err) + } + var status, state, dlq, message string + var applied bool + if err := h.QueryRow(t.Context(), `SELECT t.status,t.send_result_state,t.send_result_applied_at IS NOT NULL,d.status,wt.sent_message_id FROM tasks t JOIN task_dead_letters d ON d.task_id=t.id JOIN warmup_tokens wt ON wt.task_id=t.id WHERE t.id=$1`, f.task).Scan(&status, &state, &applied, &dlq, &message); err != nil { + t.Fatal(err) + } + if status != "dead_lettered" || state != "unknown" || applied || dlq != "pending" || message != "" { + t.Fatal("partial result effects escaped rollback") + } +} + +func TestLiveCampaignDeadLetterWorkerResultReleasesReservationOnce(t *testing.T) { + h := liveDB(t) + f := newSendResultFixture(t, h) + s := liveJobsService(h) + id := f.stampSend(t, s) + if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, id); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'campaign')`, id); err != nil { + t.Fatal(err) + } + result := models.SendEmailResult{TaskID: id, SentAt: time.Now().UTC(), Error: &models.EmailSendError{Code: "SERVER_UNREACHABLE", Failure: &errx.SendFailure{Protocol: "smtp", Stage: "dial", Disposition: errx.SendRetry, Scope: "mailbox"}}} + raw, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_executor_result=$2 WHERE id=$1`, id, raw); err != nil { + t.Fatal(err) + } + for range 2 { + s.reconcileRecordedSendResults(t.Context()) + if err := s.HandleEmailFailed(t.Context(), result); err != nil { + t.Fatal(err) + } + } + var status, dlq string + var sent, reserved, applied bool + var attempts int + if err := h.QueryRow(t.Context(), `SELECT t.status,d.status,p.sent_at IS NOT NULL,p.dispatched_at IS NOT NULL,t.send_result_applied_at IS NOT NULL,p.send_attempts FROM tasks t JOIN task_dead_letters d ON d.task_id=t.id JOIN campaign_contact_progress p ON p.campaign_id=$2 AND p.contact_id=$3 AND p.sequence_id=$4 WHERE t.id=$1`, id, f.campaign, f.contact, f.step).Scan(&status, &dlq, &sent, &reserved, &applied, &attempts); err != nil { + t.Fatal(err) + } + if status != "failed" || dlq != "resolved" || sent || reserved || !applied || attempts != 1 { + t.Fatalf("definitive campaign refusal not reconciled once: status=%s dlq=%s sent=%t reserved=%t applied=%t attempts=%d", status, dlq, sent, reserved, applied, attempts) + } +} + +func TestLiveUserEmailDeadLetterUsesUnderlyingSendTaskType(t *testing.T) { + h := liveDB(t) + f := newSendResultFixture(t, h) + s := liveJobsService(h) + id := uuid.New() + if _, err := h.Exec(t.Context(), `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','dead_lettered','')`, id, f.mailbox); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'user_email')`, id); err != nil { + t.Fatal(err) + } + if err := s.HandleEmailSent(t.Context(), models.SendEmailResult{TaskID: id, MessageID: "confirmed@example.test"}); err != nil { + t.Fatal(err) + } + var status string + if err := h.QueryRow(t.Context(), `SELECT status FROM task_dead_letters WHERE task_id=$1`, id).Scan(&status); err != nil || status != "resolved" { + t.Fatal("user_email dead letter was not settled", status, err) + } +} + +func TestLiveCampaignForeignKeyCleanupDoesNotProveWakeupOnly(t *testing.T) { + h := liveDB(t) + f := newSendResultFixture(t, h) + s := liveJobsService(h) + id := f.stampSend(t, s) + before, err := s.TaskRepo.GetCampaignTask(t.Context(), id) + if err != nil || before == nil || before.ContactID == nil || before.SequenceID == nil { + t.Fatal("send-capable campaign fixture missing", err) + } + if _, err := h.Exec(t.Context(), `DELETE FROM contacts WHERE id=$1`, f.contact); err != nil { + t.Fatal(err) + } + if _, err := h.Exec(t.Context(), `DELETE FROM sequences WHERE id=$1`, *before.SequenceID); err != nil { + t.Fatal(err) + } + after, err := s.TaskRepo.GetCampaignTask(t.Context(), id) + if err != nil || after == nil || after.ContactID != nil || after.SequenceID != nil || after.CampaignID == nil { + t.Fatal("FK-null former send no longer matches the unsafe wakeup-only predicate", err) + } +} + func TestLiveLegacyJSONAndAvroFailurePersistConservativeAdmission(t *testing.T) { writer, reader := legacySendResultSchema(t) for _, format := range []string{"json", "avro"} { diff --git a/internal/app/consumer/send_recovery_resolution_live_test.go b/internal/app/consumer/send_recovery_resolution_live_test.go index b7fd1b2f5..49e92f41c 100644 --- a/internal/app/consumer/send_recovery_resolution_live_test.go +++ b/internal/app/consumer/send_recovery_resolution_live_test.go @@ -113,9 +113,9 @@ func TestLiveRecordedWorkerOutcomeSurvivesLostEvent(t *testing.T) { } } -func TestLiveUnreservedFailedTaskCannotHoldMailbox(t *testing.T) { +func TestLiveUnreservedLegacyFailedTaskRetainsUnknownHold(t *testing.T) { for _, existing := range []bool{false, true} { - t.Run(map[bool]string{false: "new failure", true: "existing invalid hold"}[existing], func(t *testing.T) { + t.Run(map[bool]string{false: "new ambiguous result", true: "existing unknown hold"}[existing], func(t *testing.T) { h := liveDB(t) f := newSendResultFixture(t, h) s := liveJobsService(h) @@ -140,8 +140,8 @@ func TestLiveUnreservedFailedTaskCannotHoldMailbox(t *testing.T) { if err := h.Pool.QueryRow(ctx, `SELECT ea.send_recovery_hold,t.send_result_applied_at IS NOT NULL,t.send_result_state FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, id).Scan(&hold, &applied, &state); err != nil { t.Fatal(err) } - if hold || !applied || state != "failed" { - t.Fatal("unreserved failure held mailbox", hold, applied, state) + if !hold || applied || state != "unknown" { + t.Fatal("missing authority markers resolved a legacy unknown", hold, applied, state) } }) } diff --git a/internal/repository/pg_send_recovery.go b/internal/repository/pg_send_recovery.go index fd61d38a0..6717664b1 100644 --- a/internal/repository/pg_send_recovery.go +++ b/internal/repository/pg_send_recovery.go @@ -163,11 +163,8 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send var applied *time.Time var mailboxID uuid.UUID var status string - var reserved, warmupAuthorized bool - err = tx.QueryRow(ctx, `SELECT send_result_state, send_result_applied_at, email_account_id, status::text, - send_reserved_at IS NOT NULL OR send_executor_nonce IS NOT NULL OR send_executor_started_at IS NOT NULL OR send_executor_result IS NOT NULL, - EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=tasks.id AND dispatch_nonce IS NOT NULL) - FROM tasks WHERE id = $1`, result.TaskID).Scan(&previous, &applied, &mailboxID, &status, &reserved, &warmupAuthorized) + err = tx.QueryRow(ctx, `SELECT send_result_state, send_result_applied_at, email_account_id, status::text + FROM tasks WHERE id = $1`, result.TaskID).Scan(&previous, &applied, &mailboxID, &status) if errors.Is(err, pgx.ErrNoRows) { return nil } @@ -177,12 +174,10 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 269))`, mailboxID.String()); err != nil { return err } - // A pre-dispatch failure cannot leave ambiguous provider execution behind. - if resultState(result) == "unknown" && status == "failed" && !reserved && !warmupAuthorized { - result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Protocol: "internal", Stage: "prepare", Scope: "mailbox", Disposition: errx.SendRetry}} - } - if applied != nil || (previous != nil && *previous != "unknown") { - state := resultState(result) + state := resultState(result) + // Receipt stamping is idempotent; an already-applied failure cannot be refunded again. + repairDeadLetter := status == "dead_lettered" && previous != nil && *previous == state && state == "sent" + if !repairDeadLetter && (applied != nil || (previous != nil && *previous != "unknown")) { if previous != nil && state != "unknown" && state != *previous { if _, err = tx.Exec(ctx, `UPDATE send_recovery_resolutions SET conflict_detected_at=NOW() WHERE recovery_task_id=$1 AND previous_reason='unknown' AND confirmed_result IS NOT NULL`, result.TaskID); err != nil { @@ -195,12 +190,17 @@ 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 err = resolveSendDeadLetter(ctx, tx, result.TaskID); err != nil { + return err + } + return tx.Commit(ctx) + } return nil } if status == "cancelled" { return nil } - state := resultState(result) if state == "failed" && status == "active" { return errors.New("send result arrived before task stamping") } @@ -217,6 +217,15 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send var effects []func(context.Context) var effectError error if state != "unknown" { + if status == "dead_lettered" && state == "sent" { + sentAt := result.SentAt + if sentAt.IsZero() { + sentAt = time.Now().UTC() + } + if _, err = tx.Exec(ctx, `UPDATE tasks SET status='completed',completed_at=COALESCE(completed_at,$2),updated_at=NOW() WHERE id=$1`, result.TaskID, sentAt); err != nil { + return err + } + } inner := context.WithValue(ctx, sendResultKey{}, sendResultContext{tx: tx, taskID: result.TaskID, effects: &effects, effectError: &effectError}) if err = apply(inner); err != nil { return err @@ -227,6 +236,9 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send if _, err = tx.Exec(ctx, `UPDATE tasks SET send_result_applied_at = NOW(), send_released_at=CASE WHEN send_result_state='failed' THEN NOW() ELSE send_released_at END WHERE id = $1`, result.TaskID); err != nil { return err } + if err = resolveSendDeadLetter(ctx, tx, result.TaskID); err != nil { + return err + } } if err = persistSendHold(ctx, tx, mailboxID, result, state); err != nil { return err @@ -240,6 +252,13 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send return nil } +func resolveSendDeadLetter(ctx context.Context, tx pgx.Tx, taskID uuid.UUID) error { + _, err := tx.Exec(ctx, `UPDATE task_dead_letters SET status='resolved',next_retry_at=NULL,updated_at=NOW() + WHERE task_id=$1 AND status IN ('pending','failed','replayed') + AND EXISTS(SELECT 1 FROM tasks t WHERE t.id=$1 AND t.task_type IN ('warmup','campaign','email','placement'))`, taskID) + return err +} + func persistSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, result models.SendEmailResult, state string) error { if state != "unknown" { _, err := tx.Exec(ctx, `WITH next_hold AS ( diff --git a/internal/repository/pg_warmup.go b/internal/repository/pg_warmup.go index 7303b9121..80ef61fef 100644 --- a/internal/repository/pg_warmup.go +++ b/internal/repository/pg_warmup.go @@ -1094,23 +1094,32 @@ func (r *warmupRepository) FailWarmupSend(ctx context.Context, accountID, taskID defer tx.Rollback(ctx) var status string + var completedAt *time.Time + var dispatchAuthorized bool err = tx.QueryRow(ctx, ` - SELECT status::text + SELECT status::text, completed_at, + send_reserved_at IS NOT NULL OR send_executor_nonce IS NOT NULL + OR EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=tasks.id AND dispatch_nonce IS NOT NULL) FROM tasks WHERE id = $1 AND email_account_id = $2 AND task_type = 'warmup' FOR UPDATE - `, taskID, accountID).Scan(&status) + `, taskID, accountID).Scan(&status, &completedAt, &dispatchAuthorized) if errors.Is(err, pgx.ErrNoRows) { return nil } if err != nil { return err } - if status != "completed" { + if status != "completed" && status != "dead_lettered" { return tx.Commit(ctx) } - if _, err = tx.Exec(ctx, ` + // Legacy dead-letter accounting cannot be reconstructed from status alone. + if status == "completed" || (dispatchAuthorized && completedAt != nil) { + if completedAt != nil { + date = *completedAt + } + if _, err = tx.Exec(ctx, ` UPDATE warmup_statistics ws SET emails_sent = GREATEST(ws.emails_sent - 1, 0), emails_replied = CASE @@ -1122,8 +1131,9 @@ func (r *warmupRepository) FailWarmupSend(ctx context.Context, accountID, taskID END WHERE ws.email_account_id = $1 AND ws.date = DATE($3) - `, accountID, taskID, date); err != nil { - return err + `, accountID, taskID, date); err != nil { + return err + } } if _, err = tx.Exec(ctx, `UPDATE tasks SET status = 'failed', updated_at = NOW() WHERE id = $1`, taskID); err != nil { return err diff --git a/internal/repository/unknown_send_resolution.go b/internal/repository/unknown_send_resolution.go index 8d6861acd..970ef7acc 100644 --- a/internal/repository/unknown_send_resolution.go +++ b/internal/repository/unknown_send_resolution.go @@ -91,33 +91,31 @@ func queueUnknownSendResolution(ctx context.Context, tx pgx.Tx, org, mailbox str func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int) ([]models.SendEmailResult, error) { 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 + 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')) 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 AND t.send_executor_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,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 + 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')) 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 AND t.status<>'cancelled' - UNION ALL - SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,jsonb_build_object('task_id',t.id,'success',false,'legacy_error','Send failed before reservation', - 'error',jsonb_build_object('failure',jsonb_build_object('protocol','internal','stage','prepare','scope','mailbox','disposition','retry'))),3 - FROM tasks t WHERE t.status='failed' AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL - AND t.send_reserved_at IS NULL AND t.send_executor_nonce IS NULL AND t.send_executor_started_at IS NULL - AND NOT EXISTS(SELECT 1 FROM warmup_tasks w WHERE w.task_id=t.id AND w.dispatch_nonce IS NOT NULL)), + 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' + ), 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') OR (result->'error'->>'failure' IS NULL AND result->'error'->>'code' IN ('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 ORDER BY id,priority) + 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) SELECT id,result FROM pending ORDER BY created_at LIMIT $1`, limit) if err != nil { return nil, err