From 3f8e951bd6a2e979348156ee5575e59e42b9da9f Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 13:38:43 +0000 Subject: [PATCH 1/4] feat: guard legacy send dead-letter replay and expose unverified recovery evidence Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../docs/api/reference/deliverability-ops.mdx | 2 +- docs/content/docs/development/admin-panel.mdx | 2 +- .../docs/development/instance-health.mdx | 2 +- docs/content/docs/development/warmblyctl.mdx | 2 +- internal/app/advanced/campaign_replay_test.go | 46 ++++++++++++- internal/app/advanced/service.go | 64 ++++++------------- internal/repository/pg_advanced_outreach.go | 18 ++++-- internal/repository/pg_monitoring.go | 9 ++- .../pg_monitoring_queues_live_test.go | 25 +++++++- 9 files changed, 107 insertions(+), 63 deletions(-) diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx index 2ec34e0e7..82eb08043 100644 --- a/docs/content/docs/api/reference/deliverability-ops.mdx +++ b/docs/content/docs/api/reference/deliverability-ops.mdx @@ -324,7 +324,7 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin `POST /tasks/dlq/:id/replay` -Re-dispatches a dead-lettered task. Because a replay can transmit real mail, this requires the send permission rather than plain write access. +Replays only a non-sending campaign wakeup pass, using its campaign lock. Send-capable tasks, including legacy warmup and campaign contact sends, return `409 Conflict` and remain in the dead-letter queue: a retry date or missing send marker is not proof that the provider did not send. This route requires the send permission rather than plain write access. Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns` diff --git a/docs/content/docs/development/admin-panel.mdx b/docs/content/docs/development/admin-panel.mdx index 794d109cd..80599c185 100644 --- a/docs/content/docs/development/admin-panel.mdx +++ b/docs/content/docs/development/admin-panel.mdx @@ -71,7 +71,7 @@ Relative ages use your browser's current clock when a snapshot arrives, with per Mailbox diagnostic participation, sender/recipient flags and shared/rolling limits are organization-scoped mailbox fields, not new admin-panel switches. Use the existing authorized mailbox PATCH; no new UI is implied. An operator-reviewed `send_recovery_resolution` binds the current held task/reason and definite repair/provider evidence, never an unknown send. See [send safety](/development/send-safety/). | **Seed panel** | The instance's placement seed inboxes by provider, adding and removing them, every workspace's recent [placement tests](/guides/placement-tests/) with their results, and an operator test that counts against no allowance | | **Campaigns** | Every campaign on the instance with its state and owner | -| **Sends** | In-flight reservations no worker has answered, and the reclaimer that resolves them after 30 minutes; dead letters with replay; task failures; customer webhook delivery health, with reclaim | +| **Sends** | In-flight reservations no worker has answered, and the reclaimer that resolves them after 30 minutes; dead letters with replay limited to non-sending campaign wakeups; task failures; customer webhook delivery health, with reclaim | ## Accounts diff --git a/docs/content/docs/development/instance-health.mdx b/docs/content/docs/development/instance-health.mdx index a413b3ac3..832ba11ec 100644 --- a/docs/content/docs/development/instance-health.mdx +++ b/docs/content/docs/development/instance-health.mdx @@ -67,7 +67,7 @@ Two neighbouring endpoints round out the surface, both under **Instance** in the ## Operational monitoring -**Operations** renders the version `"1"` snapshot from `GET /admin/instance/monitoring`. It covers registered workers, worker sample receipt, confirmed sends, campaign reservations, dispatch, safety holds, sync state and fair-use waits, stored mailbox errors, jobs, dead letters, webhooks, customer notification rows, send-result effects, domain checks, warmup mailbox loading and broker queue observations. Unsupported sources remain visible as unavailable, including detailed send-wait attribution and external tracking-broker diagnostics. Optional arrival-outbox measurements are unavailable when their schema is absent; monitoring needs no arrival migration and does not require a new worker wire version. +**Operations** renders the version `"1"` snapshot from `GET /admin/instance/monitoring`. It covers registered workers, worker sample receipt, confirmed sends, campaign reservations, dispatch, safety holds, sync state and fair-use waits, stored mailbox errors, jobs, dead letters, webhooks, customer notification rows, send-result effects, domain checks, warmup mailbox loading and broker queue observations. Dead letters without safe replay proof remain visible as recovery-unverified even when a retry date has passed; the due, waiting and unscheduled counts only cover non-sending campaign wakeup passes. The consumer does not have a task scheduler for automatic replay. Unsupported sources remain visible as unavailable, including detailed send-wait attribution and external tracking-broker diagnostics. Optional arrival-outbox measurements are unavailable when their schema is absent; monitoring needs no arrival migration and does not require a new worker wire version. The backend collects with an eight-second overall timeout, three seconds per source and at most four concurrent collectors. It shares a one-minute snapshot across admins, then filters source permissions for each response. HTTP responses are not cacheable. On collection failure it can retain a source's last good measurement for at most five minutes, marked stale and with its original observation/evidence clocks. After that it is unavailable. Source samples are bounded; measured/expected scope counts and incomplete coverage expose truncation. Counts in partial samples may be lower bounds, never extrapolated fleet totals. diff --git a/docs/content/docs/development/warmblyctl.mdx b/docs/content/docs/development/warmblyctl.mdx index 083f681f3..c01f173a9 100644 --- a/docs/content/docs/development/warmblyctl.mdx +++ b/docs/content/docs/development/warmblyctl.mdx @@ -172,7 +172,7 @@ warmblyctl admin recovery restart-backfill --id --server laptop \ Substitute the same real canonical UUID in both places. Clearing a provider throttle can resume provider work; restarting backfill resets the existing sync backfill state. Neither action establishes sending authority. -Dead-letter replay is never part of diagnosis. Inspect the task, mailbox, provider receipts and uncertain-send protections first. Replay additionally requires a separate risk acknowledgement: +Dead-letter replay is never part of diagnosis. Only non-sending campaign wakeup passes can be replayed; send-capable dead letters remain pending until their outcome is independently reconciled through normal recovery. Inspect the task, mailbox, provider receipts and uncertain-send protections first. Replay additionally requires a separate risk acknowledgement: ```bash warmblyctl admin recovery replay --id --server laptop \ diff --git a/internal/app/advanced/campaign_replay_test.go b/internal/app/advanced/campaign_replay_test.go index da8827668..d5144fc36 100644 --- a/internal/app/advanced/campaign_replay_test.go +++ b/internal/app/advanced/campaign_replay_test.go @@ -19,6 +19,10 @@ type replayRepo struct { replayed int } +func (r *replayRepo) ListRetryableDeadLetters(context.Context, int) ([]models.TaskDeadLetter, error) { + return []models.TaskDeadLetter{*r.dlq}, nil +} + func (r *replayRepo) GetTaskDeadLetter(context.Context, uuid.UUID, uuid.UUID) (*models.TaskDeadLetter, error) { return r.dlq, nil } @@ -32,6 +36,8 @@ type replayTasks struct { repository.TaskRepository task *repository.Task campaignID *uuid.UUID + contactID *uuid.UUID + sequenceID *uuid.UUID ctErr error created bool inserted int @@ -52,7 +58,43 @@ func (t *replayTasks) GetCampaignTask(context.Context, uuid.UUID) (*repository.C if t.ctErr != nil { return nil, t.ctErr } - return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID}, nil + return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID, ContactID: t.contactID, SequenceID: t.sequenceID}, nil +} + +func TestDeadLetterNeverReplaysSendTasksWithoutAuthority(t *testing.T) { + for _, tc := range []struct { + name, dlqType, taskType string + contact, sequence bool + }{ + {name: "legacy warmup send", dlqType: "warmup", taskType: "warmup"}, + {name: "user email send", dlqType: "user_email", taskType: "user_email"}, + {name: "campaign contact send", dlqType: "campaign", taskType: "campaign", contact: true}, + {name: "campaign sequence send", dlqType: "campaign", taskType: "campaign", sequence: true}, + {name: "mismatched dead letter", dlqType: "warmup", taskType: "campaign"}, + } { + t.Run(tc.name, func(t *testing.T) { + id := uuid.New() + campaignID := uuid.New() + repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: id, TaskType: tc.dlqType}} + tasks := &replayTasks{task: &repository.Task{ID: id, TaskType: tc.taskType}, campaignID: &campaignID, created: true} + if tc.contact { + tasks.contactID = &id + } + if tc.sequence { + tasks.sequenceID = &id + } + svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{}} + if xerr := svc.ReplayDeadLetter(t.Context(), uuid.New(), repo.dlq.ID); xerr == nil { + t.Fatal("send dead letter was allowed through manual replay") + } + if n, xerr := svc.ProcessRetryableDeadLetters(t.Context()); xerr != nil || n != 0 { + t.Fatalf("auto replay result = %d, %v", n, xerr) + } + if tasks.inserted != 0 || tasks.repended != 0 || repo.replayed != 0 { + t.Fatalf("send mutated: inserted=%d pending=%d replayed=%d", tasks.inserted, tasks.repended, repo.replayed) + } + }) + } } func (t *replayTasks) CreateTaskWithLock(context.Context, *repository.Task, *repository.CampaignTask) (bool, error) { @@ -103,7 +145,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { task := &repository.Task{ID: uuid.New(), TaskType: "campaign", EmailAccountID: uuid.New()} - repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID}} + repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID, TaskType: "campaign"}} tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, created: tc.created} svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{fail: tc.queueFails}} diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go index fe830727b..bb2d1eb64 100644 --- a/internal/app/advanced/service.go +++ b/internal/app/advanced/service.go @@ -2288,26 +2288,15 @@ func (s *service) ReplayDeadLetter(ctx context.Context, organizationID, deadLett if task == nil { return errx.ErrNotFound } - if _, handled, rerr := s.replayCampaignPass(ctx, task); handled { - if rerr != nil { - return toErrx(rerr) - } - if err := s.repo.MarkTaskDeadLetterReplayed(ctx, deadLetterID); err != nil { - return toErrx(err) - } - return nil + if dlq.TaskType != "campaign" || task.TaskType != "campaign" { + return errx.New(errx.Conflict, "send dead letters require authoritative outcome evidence and cannot be replayed") } - - scheduleAt := time.Now().UTC().Add(10 * time.Second) - cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt) - if err != nil { - return toErrx(err) + _, handled, rerr := s.replayCampaignPass(ctx, task) + if rerr != nil { + return toErrx(rerr) } - if err := s.taskRepo.UpdateTaskScheduledAt(ctx, task.ID, scheduleAt, cloudTaskName); err != nil { - return toErrx(err) - } - if err := s.taskRepo.UpdateTaskStatus(ctx, task.ID, "pending"); err != nil { - return toErrx(err) + if !handled { + return errx.New(errx.Conflict, "only non-sending campaign passes can be replayed") } if err := s.repo.MarkTaskDeadLetterReplayed(ctx, deadLetterID); err != nil { return toErrx(err) @@ -2329,7 +2318,10 @@ func (s *service) replayCampaignPass(ctx context.Context, task *repository.Task) if err != nil { return false, true, err } - if ct == nil || ct.CampaignID == nil { + if ct == nil || ct.ContactID != nil || ct.SequenceID != nil { + return false, false, nil + } + if ct.CampaignID == nil { // The campaign is gone; there is nothing to replay into. return false, true, nil } @@ -2680,10 +2672,14 @@ func (s *service) ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.E retried := 0 for _, dlq := range items { + if dlq.TaskType != "campaign" { + continue + } task, err := s.taskRepo.GetTask(ctx, dlq.TaskID) - if err != nil || task == nil { - // Mark as exhausted if the task no longer exists - _ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID) + if err != nil { + return retried, toErrx(err) + } + if task == nil || task.TaskType != "campaign" { continue } @@ -2700,30 +2696,6 @@ func (s *service) ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.E _ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID) continue } - - // Check if attempts exceeded - if dlq.Attempts >= dlq.MaxAttempts { - _ = s.repo.IncrementDeadLetterAttempt(ctx, dlq.ID, nil) - continue - } - - // Schedule retry via Cloud Tasks - scheduleAt := time.Now().UTC().Add(10 * time.Second) - cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt) - if err != nil { - // Compute next retry with exponential backoff - backoff := time.Duration(30*(1<$1 AND d.attempts$1 AND d.attempts=d.max_attempts`}, - {id: "dlq_unscheduled", title: "Pending dead letters without retry slot", unit: "records", condition: "unknown", severity: "warning", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at IS NULL AND d.attempts Date: Sat, 10 Oct 2026 13:53:18 +0000 Subject: [PATCH 2/4] feat: preserve the unverified dead-letter monitoring identity in redacted administrator CLI output Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cmd/warmblyctl/remote_output.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/warmblyctl/remote_output.go b/cmd/warmblyctl/remote_output.go index e7ad9e866..5c3c13484 100644 --- a/cmd/warmblyctl/remote_output.go +++ b/cmd/warmblyctl/remote_output.go @@ -102,7 +102,7 @@ arrival_aging arrival_due arrival_leased arrival_pending arrival_stage_0 arrival arrival_stage_2 arrival_stage_3 arrival_wait backfill_complete backfill_pending backfill_running backfill_stale_running campaign_reserved campaign_reserved_aging confirmed_other_1h confirmed_warmup_1h confirmed_warmup_proxy_1h credential_errors -dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed +dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed dlq_unverified dlq_unscheduled dlq_wait effects_leased effects_pending effects_repeated effects_unleased failed_results_1h failed_results_24h future_schedule holds_authentication holds_conflict warmup_invalid_window holds_permanent holds_unknown inactive_credential_errors job_error_old job_error_recent From a53de41c92589e10ab49f4eaa6cb77f139d73b38 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 14:16:01 +0000 Subject: [PATCH 3/4] feat: persist monotonic campaign dispatch intent and require locked verified wakeup provenance for every dead-letter replay path --- .../docs/api/reference/deliverability-ops.mdx | 2 +- .../docs/guides/workspace-export-import.mdx | 2 + internal/app/advanced/campaign_replay_test.go | 25 ++- internal/app/advanced/service.go | 14 +- internal/app/orgtransfer/export_test.go | 13 ++ internal/app/orgtransfer/spec.go | 3 +- .../campaign_dispatch_intent_upgrade_test.go | 124 ++++++++++++ .../000282_campaign_dispatch_intent.down.sql | 17 ++ .../000282_campaign_dispatch_intent.up.sql | 24 +++ .../campaign_replay_intent_live_test.go | 186 ++++++++++++++++++ internal/repository/pg_advanced_outreach.go | 4 +- internal/repository/pg_monitoring.go | 2 - .../pg_monitoring_queues_live_test.go | 23 ++- internal/repository/pg_task.go | 69 ++++++- internal/tasks/campaign_send_live_test.go | 41 ++++ internal/tasks/campaign_task.go | 6 +- 16 files changed, 521 insertions(+), 34 deletions(-) create mode 100644 internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go create mode 100644 internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql create mode 100644 internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql create mode 100644 internal/repository/campaign_replay_intent_live_test.go diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx index 82eb08043..ff41364ba 100644 --- a/docs/content/docs/api/reference/deliverability-ops.mdx +++ b/docs/content/docs/api/reference/deliverability-ops.mdx @@ -324,7 +324,7 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin `POST /tasks/dlq/:id/replay` -Replays only a non-sending campaign wakeup pass, using its campaign lock. Send-capable tasks, including legacy warmup and campaign contact sends, return `409 Conflict` and remain in the dead-letter queue: a retry date or missing send marker is not proof that the provider did not send. This route requires the send permission rather than plain write access. +Replays only a dead-lettered campaign wakeup with explicit durable wakeup intent, rechecked under its campaign lock. Once a task binds a contact or sequence for sending, its send intent stays restrictive even after either record is deleted. Legacy or imported tasks without verified wakeup intent and all send-capable tasks return `409 Conflict` and retain their dead-letter evidence. A retry date, JSON payload, null contact/sequence, or missing send marker cannot grant replay. This route requires the send permission rather than plain write access. Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns` diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index 316273817..abeea2c2a 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -57,6 +57,8 @@ A workspace archive is a single file holding everything one workspace owns. It e Diagnostic participation, shared/rolling limits, restrictive send evidence and audited repair-resolution history travel with their registered mailbox/task/organization data. Source-worker executor handles reset; exact provider-message DKIM authorization/nonces are excluded. Portable minimal receipt proof remains historical evidence, never an execution grant. The reconciliation webhook/notification outbox is source-instance-local and excluded to prevent replay on import; drain pending effects before moving. Whole-instance database backups retain it. See [send-safety retention and rollback](/development/send-safety/#upgrade-rollback-and-portability). +Campaign task replay intent resets to unverified on workspace import. The destination cannot replay an imported dead letter as a verified non-sending wakeup. A contact or sequence binding remains restrictive send intent even if that record is later removed. Whole-instance backups retain the durable intent; migration rollback refuses to discard retained send intent or unverified pending campaign dead letters. + This page moves **one workspace** between two running instances, re-sealing its secrets for the destination's keys. To move an entire self-hosted install (every workspace, its users, its platform admins) use `warmblyctl backup` and `warmblyctl restore`, which carry the database, the blob root and the encryption keys as one bundle. See [data control](/development/data-control/#backups). The two are not interchangeable: a bundle cannot be applied to a single workspace, and this archive cannot restore an instance. diff --git a/internal/app/advanced/campaign_replay_test.go b/internal/app/advanced/campaign_replay_test.go index d5144fc36..f8525b321 100644 --- a/internal/app/advanced/campaign_replay_test.go +++ b/internal/app/advanced/campaign_replay_test.go @@ -38,6 +38,8 @@ type replayTasks struct { campaignID *uuid.UUID contactID *uuid.UUID sequenceID *uuid.UUID + intent repository.CampaignDispatchIntent + replayErr error ctErr error created bool inserted int @@ -58,25 +60,28 @@ func (t *replayTasks) GetCampaignTask(context.Context, uuid.UUID) (*repository.C if t.ctErr != nil { return nil, t.ctErr } - return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID, ContactID: t.contactID, SequenceID: t.sequenceID}, nil + return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID, ContactID: t.contactID, SequenceID: t.sequenceID, DispatchIntent: t.intent}, nil } func TestDeadLetterNeverReplaysSendTasksWithoutAuthority(t *testing.T) { for _, tc := range []struct { name, dlqType, taskType string contact, sequence bool + intent repository.CampaignDispatchIntent }{ {name: "legacy warmup send", dlqType: "warmup", taskType: "warmup"}, {name: "user email send", dlqType: "user_email", taskType: "user_email"}, {name: "campaign contact send", dlqType: "campaign", taskType: "campaign", contact: true}, {name: "campaign sequence send", dlqType: "campaign", taskType: "campaign", sequence: true}, {name: "mismatched dead letter", dlqType: "warmup", taskType: "campaign"}, + {name: "legacy null FKs", dlqType: "campaign", taskType: "campaign", intent: repository.CampaignDispatchUnverified}, + {name: "send after FK cleanup", dlqType: "campaign", taskType: "campaign", intent: repository.CampaignDispatchSend}, } { t.Run(tc.name, func(t *testing.T) { id := uuid.New() campaignID := uuid.New() repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: id, TaskType: tc.dlqType}} - tasks := &replayTasks{task: &repository.Task{ID: id, TaskType: tc.taskType}, campaignID: &campaignID, created: true} + tasks := &replayTasks{task: &repository.Task{ID: id, TaskType: tc.taskType, Status: "dead_lettered"}, campaignID: &campaignID, created: true, intent: tc.intent} if tc.contact { tasks.contactID = &id } @@ -104,6 +109,16 @@ func (t *replayTasks) CreateTaskWithLock(context.Context, *repository.Task, *rep return t.created, nil } +func (t *replayTasks) CreateCampaignReplayTask(ctx context.Context, task *repository.Task, ct *repository.CampaignTask, source uuid.UUID) (bool, error) { + if source != t.task.ID { + return false, errors.New("wrong replay source") + } + if t.replayErr != nil { + return false, t.replayErr + } + return t.CreateTaskWithLock(ctx, task, ct) +} + func (t *replayTasks) UpdateTaskScheduledAt(context.Context, uuid.UUID, time.Time, string) error { return nil } @@ -131,6 +146,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) { for _, tc := range []struct { name string ctErr error + replayErr error created bool queueFails bool wantErr bool @@ -142,11 +158,12 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) { {name: "chain already has its next pass", created: false, wantReplayed: 1}, {name: "campaign task unreadable", ctErr: errors.New("connection reset"), wantErr: true}, {name: "queue refuses the new pass", created: true, queueFails: true, wantErr: true, wantInserted: 1, wantDeleted: 1}, + {name: "intent changed before lock", replayErr: repository.ErrCampaignReplayUnverified, wantErr: true}, } { t.Run(tc.name, func(t *testing.T) { - task := &repository.Task{ID: uuid.New(), TaskType: "campaign", EmailAccountID: uuid.New()} + task := &repository.Task{ID: uuid.New(), TaskType: "campaign", EmailAccountID: uuid.New(), Status: "dead_lettered"} repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID, TaskType: "campaign"}} - tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, created: tc.created} + tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, replayErr: tc.replayErr, created: tc.created, intent: repository.CampaignDispatchWakeup} svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{fail: tc.queueFails}} xerr := svc.ReplayDeadLetter(context.Background(), uuid.New(), repo.dlq.ID) diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go index bb2d1eb64..fc3078151 100644 --- a/internal/app/advanced/service.go +++ b/internal/app/advanced/service.go @@ -2311,25 +2311,27 @@ func (s *service) ReplayDeadLetter(ctx context.Context, organizationID, deadLett // campaign pass; replayed, that a new pass was queued. An error means nothing // was queued and the dead letter stays for another try. func (s *service) replayCampaignPass(ctx context.Context, task *repository.Task) (replayed, handled bool, err error) { - if task == nil || task.TaskType != "campaign" { + if task == nil || task.TaskType != "campaign" || task.Status != "dead_lettered" { return false, false, nil } ct, err := s.taskRepo.GetCampaignTask(ctx, task.ID) if err != nil { return false, true, err } - if ct == nil || ct.ContactID != nil || ct.SequenceID != nil { + if ct == nil || ct.DispatchIntent != repository.CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil { return false, false, nil } if ct.CampaignID == nil { - // The campaign is gone; there is nothing to replay into. - return false, true, nil + return false, false, nil } at := time.Now().UTC().Add(10 * time.Second) id := uuid.New() - created, err := s.taskRepo.CreateTaskWithLock(ctx, + created, err := s.taskRepo.CreateCampaignReplayTask(ctx, &repository.Task{ID: id, TaskType: "campaign", EmailAccountID: task.EmailAccountID, Status: "pending", ScheduledAt: &at}, - &repository.CampaignTask{TaskID: id, CampaignID: ct.CampaignID}) + &repository.CampaignTask{TaskID: id, CampaignID: ct.CampaignID}, task.ID) + if errors.Is(err, repository.ErrCampaignReplayUnverified) { + return false, false, nil + } if err != nil { return false, true, err } diff --git a/internal/app/orgtransfer/export_test.go b/internal/app/orgtransfer/export_test.go index 16dd91e41..7c5d636c8 100644 --- a/internal/app/orgtransfer/export_test.go +++ b/internal/app/orgtransfer/export_test.go @@ -2,9 +2,22 @@ package orgtransfer import ( "encoding/json" + "slices" "testing" ) +func TestCampaignReplayIntentResetsOnWorkspaceImport(t *testing.T) { + for _, tbl := range Tables { + if tbl.Name == "campaign_tasks" { + if !slices.Contains(tbl.ResetOnImport, "dispatch_intent") { + t.Fatal("workspace import grants source-instance replay intent") + } + return + } + } + t.Fatal("campaign tasks missing from portable data registry") +} + // An archive is a customer download. The risk verdict and the evidence behind // it are operator-facing, so they must not ride along in the manifest, and the // same list must refuse them on the way back in. diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index 424ab46d9..59e04042c 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -814,7 +814,8 @@ var Tables = []Table{ }, { Name: "campaign_tasks", Group: models.OrgDataGroupSending, - Scope: `task_id IN ` + orgTasks, + Scope: `task_id IN ` + orgTasks, + ResetOnImport: []string{"dispatch_intent"}, }, { Name: "warmup_tasks", Group: models.OrgDataGroupSending, diff --git a/internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go b/internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go new file mode 100644 index 000000000..998c81237 --- /dev/null +++ b/internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go @@ -0,0 +1,124 @@ +package db + +import ( + "context" + "net/url" + "os" + "strings" + "testing" + + "github.com/golang-migrate/migrate/v4/source/iofs" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +func TestLiveCampaignDispatchIntentUpgradePreservesUnverifiedHistoryAndRollback(t *testing.T) { + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + u, err := url.Parse(dsn) + if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") { + t.Fatal("upgrade requires local PostgreSQL") + } + admin, err := pgx.Connect(t.Context(), dsn) + if err != nil { + t.Fatal(err) + } + defer admin.Close(context.Background()) + name := "campaign_intent_upgrade_" + uuid.New().String()[:8] + quoted := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil { + t.Error(err) + } + }() + u.Path = "/" + name + source, err := iofs.New(migrationsFS, "migrations") + if err != nil { + t.Fatal(err) + } + m, err := openMigrate(source, u.String()) + if err != nil { + t.Fatal(err) + } + defer m.Close() + if err := m.Migrate(281); err != nil { + t.Fatal(err) + } + conn, err := pgx.Connect(t.Context(), u.String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + exec := func(sql string, args ...any) { + t.Helper() + if _, err := conn.Exec(t.Context(), sql, args...); err != nil { + t.Fatal(err) + } + } + user, org, mailbox, campaign, contact, step := uuid.New(), uuid.New(), uuid.New(), uuid.New(), uuid.New(), uuid.New() + legacy, bound, wakeup := uuid.New(), uuid.New(), uuid.New() + exec(`INSERT INTO users(id,email,first_name,last_name) VALUES($1,$2,'Upgrade','')`, user, user.String()+"@example.test") + exec(`INSERT INTO organizations(id,name,owner_user_id) VALUES($1,'Upgrade',$2)`, org, user) + exec(`INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,send_recovery_hold,send_recovery_reason,send_recovery_task_id) VALUES($1,$2,$3,$4,'Upgrade','','','smtp_imap',true,'unknown',$5)`, mailbox, user, org, mailbox.String()+"@example.test", legacy) + exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Upgrade','','active',127,'UTC',NOW(),NOW())`, campaign, user, org) + exec(`INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields) VALUES($1,$2,$3,'lead@example.test','','','','','{}')`, contact, user, org) + exec(`INSERT INTO sequences(id,campaign_id,organization_id,name,subject,body_plain,body_html,wait_after,position) VALUES($1,$2,$3,'Step','Subject','Text','',0,1)`, step, campaign, org) + for _, task := range []uuid.UUID{legacy, bound} { + exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, task, mailbox) + exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, task, campaign) + exec(`INSERT INTO task_dead_letters(task_id,task_type,payload,last_error,attempts) VALUES($1,'campaign','{"dispatch_intent":"wakeup"}','original',1)`, task) + } + exec(`UPDATE campaign_tasks SET contact_id=$2,sequence_id=$3 WHERE task_id=$1`, bound, contact, step) + if err := m.Up(); err != nil { + t.Fatal(err) + } + var unchanged bool + if err := conn.QueryRow(t.Context(), `SELECT COUNT(*)=2 AND bool_and(ct.dispatch_intent='unverified' AND t.send_result_state IS NULL AND t.send_result_applied_at IS NULL AND d.status='pending' AND d.attempts=1 AND d.last_error='original' AND d.payload='{"dispatch_intent":"wakeup"}'::jsonb AND ea.send_recovery_hold AND ea.send_recovery_reason='unknown') FROM campaign_tasks ct JOIN tasks t ON t.id=ct.task_id JOIN task_dead_letters d ON d.task_id=t.id JOIN email_accounts ea ON ea.id=t.email_account_id`).Scan(&unchanged); err != nil || !unchanged { + t.Fatal("upgrade granted authority or changed unknown evidence", unchanged, err) + } + exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, legacy) + var intent string + if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent::text FROM campaign_tasks WHERE task_id=$1`, legacy).Scan(&intent); err != nil || intent != "unverified" { + t.Fatal("legacy promoted to verified wakeup", intent, err) + } + exec(`DELETE FROM contacts WHERE id=$1`, contact) + exec(`DELETE FROM sequences WHERE id=$1`, step) + exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, bound) + if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent FROM campaign_tasks WHERE task_id=$1`, bound).Scan(&intent); err != nil || intent != "send" { + t.Fatal("FK cleanup forgot prior bound send", intent, err) + } + exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, wakeup, mailbox) + exec(`INSERT INTO campaign_tasks(task_id,campaign_id,dispatch_intent) VALUES($1,$2,'wakeup')`, wakeup, campaign) + if _, err := conn.Exec(t.Context(), `UPDATE campaign_tasks SET dispatch_intent='arbitrary' WHERE task_id=$1`, wakeup); err == nil { + t.Fatal("intent accepted arbitrary value") + } + down, err := migrationsFS.ReadFile("migrations/000282_campaign_dispatch_intent.down.sql") + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(t.Context(), string(down)); err == nil || !strings.Contains(err.Error(), "restrictive recovery evidence") { + t.Fatal("rollback discarded retained send evidence", err) + } + exec(`DELETE FROM tasks WHERE id=$1`, bound) + if _, err := conn.Exec(t.Context(), string(down)); err == nil || !strings.Contains(err.Error(), "restrictive recovery evidence") { + t.Fatal("rollback discarded unverified pending dead letter", err) + } + // Only this isolated fixture is removed to exercise a clean down/up pair. + exec(`DELETE FROM tasks WHERE id IN($1,$2)`, legacy, wakeup) + exec(string(down)) + up, err := migrationsFS.ReadFile("migrations/000282_campaign_dispatch_intent.up.sql") + if err != nil { + t.Fatal(err) + } + exec(string(up)) + exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, legacy, mailbox) + exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, legacy, campaign) + if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent::text FROM campaign_tasks WHERE task_id=$1`, legacy).Scan(&intent); err != nil || intent != "unverified" { + t.Fatal("clean down/up lost restrictive default", intent, err) + } +} diff --git a/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql new file mode 100644 index 000000000..f09fad647 --- /dev/null +++ b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql @@ -0,0 +1,17 @@ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM campaign_tasks ct + LEFT JOIN task_dead_letters d ON d.task_id = ct.task_id AND d.status = 'pending' + WHERE ct.dispatch_intent = 'send' + OR (ct.dispatch_intent = 'unverified' AND d.id IS NOT NULL) + ) THEN + RAISE EXCEPTION 'campaign dispatch intent is restrictive recovery evidence; retain it and roll forward'; + END IF; +END; +$$; + +DROP TRIGGER campaign_dispatch_intent_monotonic ON campaign_tasks; +DROP FUNCTION retain_campaign_dispatch_intent(); +ALTER TABLE campaign_tasks DROP COLUMN dispatch_intent; +DROP TYPE campaign_dispatch_intent; diff --git a/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql new file mode 100644 index 000000000..dc63c36af --- /dev/null +++ b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql @@ -0,0 +1,24 @@ +CREATE TYPE campaign_dispatch_intent AS ENUM ('unverified', 'wakeup', 'send'); + +ALTER TABLE campaign_tasks + ADD COLUMN dispatch_intent campaign_dispatch_intent NOT NULL DEFAULT 'unverified'; + +CREATE FUNCTION retain_campaign_dispatch_intent() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'UPDATE' THEN + IF OLD.dispatch_intent = 'send' OR OLD.contact_id IS NOT NULL OR OLD.sequence_id IS NOT NULL THEN + NEW.dispatch_intent := 'send'; + ELSIF OLD.dispatch_intent = 'unverified' AND NEW.dispatch_intent = 'wakeup' THEN + NEW.dispatch_intent := 'unverified'; + END IF; + END IF; + IF NEW.contact_id IS NOT NULL OR NEW.sequence_id IS NOT NULL THEN + NEW.dispatch_intent := 'send'; + END IF; + RETURN NEW; +END; +$$; + +CREATE TRIGGER campaign_dispatch_intent_monotonic + BEFORE INSERT OR UPDATE ON campaign_tasks + FOR EACH ROW EXECUTE FUNCTION retain_campaign_dispatch_intent(); diff --git a/internal/repository/campaign_replay_intent_live_test.go b/internal/repository/campaign_replay_intent_live_test.go new file mode 100644 index 000000000..8a83e0bca --- /dev/null +++ b/internal/repository/campaign_replay_intent_live_test.go @@ -0,0 +1,186 @@ +package repository + +import ( + "context" + "errors" + "net/url" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/infrastructure/db" +) + +func liveCampaignReplayDB(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + u, err := url.Parse(dsn) + if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") { + t.Fatal("replay fixture requires local PostgreSQL") + } + admin, err := pgx.Connect(t.Context(), dsn) + if err != nil { + t.Fatal(err) + } + name := "campaign_replay_" + uuid.New().String()[:8] + quoted := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil { + _ = admin.Close(context.Background()) + t.Fatal(err) + } + t.Cleanup(func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil { + t.Error(err) + } + _ = admin.Close(context.Background()) + }) + u.Path = "/" + name + if err := db.RunMigrations(u.String()); err != nil { + t.Fatal(err) + } + pool, err := pgxpool.New(t.Context(), u.String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + return pool +} + +func TestLiveCampaignReplayIntentTransitionsFilterBeforeLimitAndLock(t *testing.T) { + pool := liveCampaignReplayDB(t) + f := newWarmupUsageFixture(t, pool) + r := NewTaskRepository(pool) + campaign, contact, sequence := uuid.New(), uuid.New(), uuid.New() + f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaign, f.user, f.org) + f.exec(`INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields) VALUES($1,$2,$3,'lead@example.test','','','','','{}')`, contact, f.user, f.org) + f.exec(`INSERT INTO sequences(id,campaign_id,organization_id,name,subject,body_plain,body_html,wait_after,position) VALUES($1,$2,$3,'Step','Subject','Text','',0,1)`, sequence, campaign, f.org) + if err := r.UpdateCampaignTaskTracking(t.Context(), uuid.New(), contact, sequence); !errors.Is(err, pgx.ErrNoRows) { + t.Fatal("missing tracking row accepted as durable intent", err) + } + for range 60 { + id := uuid.New() + f.exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, id, f.account) + f.exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, id, campaign) + f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at,payload) VALUES($1,'campaign',NOW()-INTERVAL '2 hours','{"dispatch_intent":"wakeup","safe":true}')`, id) + } + newWakeup := func() uuid.UUID { + t.Helper() + id := uuid.New() + created, err := r.CreateTaskWithLock(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "dead_lettered", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign}) + if err != nil || !created { + t.Fatalf("create wakeup: %t %v", created, err) + } + f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at,payload,last_error,attempts) VALUES($1,'campaign',NOW()-INTERVAL '1 hour','{"original":true}','retained',1)`, id) + return id + } + send := newWakeup() + if err := r.UpdateCampaignTaskTracking(t.Context(), send, contact, sequence); err != nil { + t.Fatal(err) + } + f.exec(`DELETE FROM contacts WHERE id=$1`, contact) + f.exec(`DELETE FROM sequences WHERE id=$1`, sequence) + f.exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, send) + ct, err := r.GetCampaignTask(t.Context(), send) + if err != nil || ct.DispatchIntent != CampaignDispatchSend || ct.ContactID != nil || ct.SequenceID != nil { + t.Fatalf("send capability lost after FK cleanup: %+v %v", ct, err) + } + wakeup := newWakeup() + listed, err := NewAdvancedOutreachRepository(pool).ListRetryableDeadLetters(t.Context(), 50) + if err != nil || len(listed) != 1 || listed[0].TaskID != wakeup { + t.Fatalf("unsafe legacy rows filled LIMIT before wakeup: %+v %v", listed, err) + } + for _, row := range listed { + ct, err := r.GetCampaignTask(t.Context(), row.TaskID) + if err != nil || ct.DispatchIntent != CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil { + t.Fatalf("unverified replay candidate: %+v %v", ct, err) + } + } + for _, source := range []uuid.UUID{send, uuid.New()} { + id := uuid.New() + if _, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign}, source); !errors.Is(err, ErrCampaignReplayUnverified) { + t.Fatalf("unverified source replay: %v", err) + } + } + var created atomic.Int32 + var wg sync.WaitGroup + errs := make(chan error, 8) + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + id := uuid.New() + ok, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign}, wakeup) + if err != nil { + errs <- err + } + if ok { + created.Add(1) + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatal(err) + } + if created.Load() != 1 { + t.Fatal("per-campaign replay created duplicate chains", created.Load()) + } + var intact bool + if err := pool.QueryRow(t.Context(), `SELECT status='pending' AND attempts=1 AND last_error='retained' AND payload='{"original":true}'::jsonb FROM task_dead_letters WHERE task_id=$1`, wakeup).Scan(&intact); err != nil || !intact { + t.Fatal("constructor changed original evidence", intact, err) + } +} + +func TestLiveCampaignReplayRechecksIntentAfterRowLockWait(t *testing.T) { + pool := liveCampaignReplayDB(t) + f := newWarmupUsageFixture(t, pool) + r := NewTaskRepository(pool) + campaign, source, replay := uuid.New(), uuid.New(), uuid.New() + f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaign, f.user, f.org) + if ok, err := r.CreateTaskWithLock(t.Context(), &Task{ID: source, TaskType: "campaign", Status: "dead_lettered", EmailAccountID: f.account}, &CampaignTask{TaskID: source, CampaignID: &campaign}); err != nil || !ok { + t.Fatal(ok, err) + } + f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at) VALUES($1,'campaign',NOW())`, source) + tx, err := pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + if _, err := tx.Exec(t.Context(), `UPDATE campaign_tasks SET dispatch_intent='send' WHERE task_id=$1`, source); err != nil { + t.Fatal(err) + } + result := make(chan error, 1) + go func() { + _, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: replay, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: replay, CampaignID: &campaign}, source) + result <- err + }() + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + for { + var waiting bool + if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE 'SELECT true FROM task_dead_letters%')`).Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + if ctx.Err() != nil { + t.Fatal(ctx.Err()) + } + } + if err := tx.Commit(t.Context()); err != nil { + t.Fatal(err) + } + if err := <-result; !errors.Is(err, ErrCampaignReplayUnverified) { + t.Fatalf("stale wakeup snapshot allowed replay: %v", err) + } +} diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go index 624f66e7a..d28520ed3 100644 --- a/internal/repository/pg_advanced_outreach.go +++ b/internal/repository/pg_advanced_outreach.go @@ -1317,13 +1317,11 @@ func (r *advancedOutreachRepository) ListRetryableDeadLetters(ctx context.Contex SELECT d.id, d.task_id, d.task_type, d.payload, d.last_error, d.attempts, d.max_attempts, d.status, d.next_retry_at, d.replayed_at, d.created_at, d.updated_at FROM task_dead_letters d JOIN tasks t ON t.id = d.task_id - JOIN campaign_tasks ct ON ct.task_id = t.id WHERE d.status = 'pending' AND d.next_retry_at IS NOT NULL AND d.next_retry_at <= NOW() AND d.attempts < d.max_attempts - AND d.task_type = 'campaign' AND t.task_type = 'campaign' - AND ct.contact_id IS NULL AND ct.sequence_id IS NULL + AND ` + safeDLQCampaignPass + ` ORDER BY d.next_retry_at ASC LIMIT $1 ` diff --git a/internal/repository/pg_monitoring.go b/internal/repository/pg_monitoring.go index 9841eeb35..028f785d9 100644 --- a/internal/repository/pg_monitoring.go +++ b/internal/repository/pg_monitoring.go @@ -29,8 +29,6 @@ type monitoringSection struct { queries []monitoringQuery } -const safeDLQCampaignPass = `d.task_type='campaign' AND t.task_type='campaign' AND EXISTS (SELECT 1 FROM campaign_tasks ct WHERE ct.task_id=t.id AND ct.contact_id IS NULL AND ct.sequence_id IS NULL)` - func (r *MonitoringRepository) Sources() []monitoring.Source { out := make([]monitoring.Source, 0, len(monitoringSections)+3) for _, id := range monitoringSectionOrder { diff --git a/internal/repository/pg_monitoring_queues_live_test.go b/internal/repository/pg_monitoring_queues_live_test.go index d3e4ecbc7..e8450d008 100644 --- a/internal/repository/pg_monitoring_queues_live_test.go +++ b/internal/repository/pg_monitoring_queues_live_test.go @@ -11,8 +11,8 @@ import ( ) func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) { - _, pool := liveContactDB(t) - requireSchemaVersion(t, pool, 277) + pool := liveCampaignReplayDB(t) + requireSchemaVersion(t, pool, 282) f := newWarmupUsageFixture(t, pool) now := time.Now().UTC().Truncate(time.Microsecond) r := NewMonitoringRepository(pool) @@ -31,8 +31,11 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) { } t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM task_dead_letters WHERE task_id=$1`, id) }) passID := uuid.New() - f.exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id)VALUES($1,'campaign',$2,'dead_lettered','')`, passID, f.account) - f.exec(`INSERT INTO campaign_tasks(task_id) VALUES($1)`, passID) + campaignID := uuid.New() + f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at)VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaignID, f.user, f.org) + if created, err := NewTaskRepository(pool).CreateTaskWithLock(t.Context(), &Task{ID: passID, TaskType: "campaign", EmailAccountID: f.account, Status: "dead_lettered"}, &CampaignTask{TaskID: passID, CampaignID: &campaignID}); err != nil || !created { + t.Fatal(created, err) + } f.exec(`INSERT INTO task_dead_letters(task_id,task_type,status,next_retry_at,attempts,max_attempts,payload,last_error)VALUES($1,'campaign','pending',$2,1,5,'{}','test')`, passID, now.Add(-2*time.Minute)) t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM task_dead_letters WHERE task_id=$1`, passID) @@ -41,14 +44,20 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) { if err != nil { t.Fatal(err) } - if len(listed) == 0 || listed[0].TaskID != passID { - t.Fatal("replayable campaign pass was missing or an unverified send was chosen") - } + found := false for _, letter := range listed { + found = found || letter.TaskID == passID + ct, err := NewTaskRepository(pool).GetCampaignTask(t.Context(), letter.TaskID) + if err != nil || ct == nil || ct.DispatchIntent != CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil { + t.Fatal("candidate lacks positive wakeup intent", ct, err) + } if letter.TaskID == id { t.Fatal("legacy send was selected for automatic replay") } } + if !found { + t.Fatal("replayable campaign pass missing from isolated fixture") + } job := uuid.NewString() f.exec(`INSERT INTO scheduled_job_runs(name,last_status,last_finished_at,error_count,run_count,last_error)VALUES($1,'error',$2,999,1000,'private error')`, job, now.Add(-time.Minute)) t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM scheduled_job_runs WHERE name=$1`, job) }) diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go index e4a772ea2..13d60205b 100644 --- a/internal/repository/pg_task.go +++ b/internal/repository/pg_task.go @@ -30,12 +30,29 @@ type Task struct { // CampaignTask represents campaign-specific task data type CampaignTask struct { - TaskID uuid.UUID - CampaignID *uuid.UUID - ContactID *uuid.UUID - SequenceID *uuid.UUID + TaskID uuid.UUID + CampaignID *uuid.UUID + ContactID *uuid.UUID + SequenceID *uuid.UUID + DispatchIntent CampaignDispatchIntent } +type CampaignDispatchIntent string + +const ( + CampaignDispatchUnverified CampaignDispatchIntent = "unverified" + CampaignDispatchWakeup CampaignDispatchIntent = "wakeup" + CampaignDispatchSend CampaignDispatchIntent = "send" +) + +var ErrCampaignReplayUnverified = errors.New("campaign replay requires verified wakeup intent") + +const campaignWakeupReplayPredicate = `ct.dispatch_intent='wakeup' AND ct.campaign_id IS NOT NULL AND ct.contact_id IS NULL AND ct.sequence_id IS NULL` + +const campaignReplayTaskPredicate = `d.task_type='campaign' AND t.task_type='campaign' AND t.status='dead_lettered'` + +const safeDLQCampaignPass = campaignReplayTaskPredicate + ` AND EXISTS (SELECT 1 FROM campaign_tasks ct WHERE ct.task_id=t.id AND ` + campaignWakeupReplayPredicate + `)` + // WarmupTask represents warmup-specific task data type WarmupTask struct { TaskID uuid.UUID @@ -170,6 +187,7 @@ type TaskRepository interface { // Task locking CreateTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask) (bool, error) + CreateCampaignReplayTask(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error) CreateWarmupTaskWithLock(ctx context.Context, task *Task, warmupTask *WarmupTask) (bool, error) // DirectPendingWarmupTask points a mailbox's pending warmup task at one // partner and pulls it forward. Returns false when there is no pending task @@ -385,7 +403,7 @@ func (r *taskRepository) GetTaskByMessageID(ctx context.Context, messageID strin // GetCampaignTask retrieves campaign task data func (r *taskRepository) GetCampaignTask(ctx context.Context, taskID uuid.UUID) (*CampaignTask, error) { query := ` - SELECT task_id, campaign_id, contact_id, sequence_id + SELECT task_id, campaign_id, contact_id, sequence_id, dispatch_intent FROM campaign_tasks WHERE task_id = $1 ` @@ -396,6 +414,7 @@ func (r *taskRepository) GetCampaignTask(ctx context.Context, taskID uuid.UUID) &campaignTask.CampaignID, &campaignTask.ContactID, &campaignTask.SequenceID, + &campaignTask.DispatchIntent, ) if errors.Is(err, sql.ErrNoRows) { @@ -945,6 +964,20 @@ func (r *taskRepository) DeleteTask(ctx context.Context, taskID uuid.UUID) error // CreateTaskWithLock creates a campaign task with a PostgreSQL advisory lock. // It returns false when the campaign already has a pending wakeup task. func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask) (bool, error) { + return r.createCampaignTaskWithLock(ctx, task, campaignTask, uuid.Nil) +} + +func (r *taskRepository) CreateCampaignReplayTask(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error) { + if sourceTaskID == uuid.Nil || task == nil || task.TaskType != "campaign" || campaignTask == nil || campaignTask.CampaignID == nil || campaignTask.ContactID != nil || campaignTask.SequenceID != nil { + return false, ErrCampaignReplayUnverified + } + return r.createCampaignTaskWithLock(ctx, task, campaignTask, sourceTaskID) +} + +func (r *taskRepository) createCampaignTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error) { + if task == nil || (campaignTask != nil && campaignTask.TaskID != task.ID) { + return false, ErrCampaignReplayUnverified + } tx, err := r.db.Begin(ctx) if err != nil { return false, err @@ -957,6 +990,19 @@ func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, cam if err != nil { return false, err } + if sourceTaskID != uuid.Nil { + var verified bool + err = tx.QueryRow(ctx, `SELECT true FROM task_dead_letters d JOIN tasks t ON t.id=d.task_id + JOIN campaign_tasks ct ON ct.task_id=t.id WHERE t.id=$1 AND ct.campaign_id=$2 + AND t.email_account_id=$3 AND d.status='pending' AND `+campaignReplayTaskPredicate+` AND `+campaignWakeupReplayPredicate+` + FOR UPDATE OF d,t,ct`, sourceTaskID, *campaignTask.CampaignID, task.EmailAccountID).Scan(&verified) + if errors.Is(err, pgx.ErrNoRows) { + return false, ErrCampaignReplayUnverified + } + if err != nil { + return false, err + } + } var existing uuid.UUID err = tx.QueryRow(ctx, ` @@ -987,8 +1033,12 @@ func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, cam // Create campaign task entry if provided if campaignTask != nil { - ctQuery := `INSERT INTO campaign_tasks (task_id, campaign_id, contact_id, sequence_id) VALUES ($1, $2, $3, $4)` - _, err = tx.Exec(ctx, ctQuery, campaignTask.TaskID, campaignTask.CampaignID, campaignTask.ContactID, campaignTask.SequenceID) + intent := CampaignDispatchUnverified + if task.TaskType == "campaign" && campaignTask.CampaignID != nil && campaignTask.ContactID == nil && campaignTask.SequenceID == nil { + intent = CampaignDispatchWakeup + } + ctQuery := `INSERT INTO campaign_tasks (task_id, campaign_id, contact_id, sequence_id, dispatch_intent) VALUES ($1, $2, $3, $4, $5)` + _, err = tx.Exec(ctx, ctQuery, campaignTask.TaskID, campaignTask.CampaignID, campaignTask.ContactID, campaignTask.SequenceID, intent) if err != nil { return false, err } @@ -1181,7 +1231,10 @@ func (r *taskRepository) UpdateCampaignTaskTracking(ctx context.Context, taskID, WHERE task_id = $1 ` - _, err := r.db.Exec(ctx, query, taskID, contactID, sequenceID) + tag, err := r.db.Exec(ctx, query, taskID, contactID, sequenceID) + if err == nil && tag.RowsAffected() != 1 { + return pgx.ErrNoRows + } return err } diff --git a/internal/tasks/campaign_send_live_test.go b/internal/tasks/campaign_send_live_test.go index ec5be6ae5..ebe6b1ac2 100644 --- a/internal/tasks/campaign_send_live_test.go +++ b/internal/tasks/campaign_send_live_test.go @@ -37,6 +37,47 @@ type breakingProgressRepo struct { breakStamp bool } +type rejectingCampaignTrackingRepo struct { + repository.TaskRepository +} + +func (r rejectingCampaignTrackingRepo) UpdateCampaignTaskTracking(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) error { + return errors.New("simulated durable tracking write failure") +} + +func TestLiveCampaignSendRequiresDurableIntentBeforeReservation(t *testing.T) { + f := newSendFixture(t) + id := uuid.New() + at := time.Now().Add(-time.Minute) + if ok, err := f.svc.taskRepo.CreateTaskWithLock(t.Context(), &Task{ID: id, TaskType: "campaign", EmailAccountID: f.mailbox, Status: "pending", ScheduledAt: &at}, &CampaignTask{TaskID: id, CampaignID: &f.campaign}); err != nil || !ok { + t.Fatal(ok, err) + } + original := f.svc.taskRepo + f.svc.taskRepo = rejectingCampaignTrackingRepo{TaskRepository: original} + if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: id.String()}); xerr == nil { + t.Fatal("failed durable intent write allowed execution") + } + if f.sender.count() != 0 { + t.Fatal("send dispatched without durable send intent") + } + var dispatched int + if err := f.pool.QueryRow(t.Context(), `SELECT COUNT(*) FROM campaign_contact_progress WHERE campaign_id=$1 AND dispatched_at IS NOT NULL`, f.campaign).Scan(&dispatched); err != nil || dispatched != 0 { + t.Fatal("send reserved without durable intent", dispatched, err) + } + ct, err := original.GetCampaignTask(t.Context(), id) + if err != nil || ct.DispatchIntent != repository.CampaignDispatchWakeup { + t.Fatal("unsent wakeup intent changed", ct, err) + } + f.svc.taskRepo = original + if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: id.String()}); xerr != nil { + t.Fatal(xerr) + } + ct, err = original.GetCampaignTask(t.Context(), id) + if err != nil || ct.DispatchIntent != repository.CampaignDispatchSend || f.sender.count() != 1 { + t.Fatal("durable send intent not recorded before successful retry", ct, err) + } +} + func (b *breakingProgressRepo) RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error { if b.breakStamp { return errors.New("simulated transient database error on the progress write") diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index 2d4cf2a5c..6ee4d36ed 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -504,8 +504,10 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx // This allows the tracking consumer to find the correct contact/sequence when // processing open/click events from the tracking pixel service if err := s.taskRepo.UpdateCampaignTaskTracking(ctx, taskID, contact.ID, sequence.ID); err != nil { - // Log but don't fail - tracking can still work via fallback methods - log.Warn().Err(err).Str("campaign_id", campaign.ID.String()).Str("task_id", taskID.String()).Msg("Failed to update campaign task tracking") + errs.CaptureException(err) + s.retryCampaignTickLater(ctx, taskRecord) + executionStatus = "failed" + return errx.InternalError() } // stop_on_reply is enforced inside FindRoutedPairs (STEP 6), and it is now From 5bf8bf6e5176689e2da192b89c898da3a503a14e Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 14:25:16 +0000 Subject: [PATCH 4/4] feat: document the campaign dispatch provenance downgrade guard and roll-forward recovery contract Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- docs/content/docs/api/reference/deliverability-ops.mdx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx index ff41364ba..46c84bbd8 100644 --- a/docs/content/docs/api/reference/deliverability-ops.mdx +++ b/docs/content/docs/api/reference/deliverability-ops.mdx @@ -326,6 +326,8 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin Replays only a dead-lettered campaign wakeup with explicit durable wakeup intent, rechecked under its campaign lock. Once a task binds a contact or sequence for sending, its send intent stays restrictive even after either record is deleted. Legacy or imported tasks without verified wakeup intent and all send-capable tasks return `409 Conflict` and retain their dead-letter evidence. A retry date, JSON payload, null contact/sequence, or missing send marker cannot grant replay. This route requires the send permission rather than plain write access. +Database downgrades refuse to discard this provenance while send-intent or unresolved unverified dead-letter rows exist. Keep the schema and roll forward instead. + Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns` | Parameter | In | Type | Description |