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 diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx index 2ec34e0e7..46c84bbd8 100644 --- a/docs/content/docs/api/reference/deliverability-ops.mdx +++ b/docs/content/docs/api/reference/deliverability-ops.mdx @@ -324,7 +324,9 @@ 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 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` 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/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 da8827668..f8525b321 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,10 @@ type replayTasks struct { repository.TaskRepository task *repository.Task campaignID *uuid.UUID + contactID *uuid.UUID + sequenceID *uuid.UUID + intent repository.CampaignDispatchIntent + replayErr error ctErr error created bool inserted int @@ -52,7 +60,46 @@ 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, 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, Status: "dead_lettered"}, campaignID: &campaignID, created: true, intent: tc.intent} + 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) { @@ -62,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 } @@ -89,6 +146,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) { for _, tc := range []struct { name string ctErr error + replayErr error created bool queueFails bool wantErr bool @@ -100,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()} - repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID}} - tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, created: tc.created} + 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, 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 fe830727b..fc3078151 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) @@ -2322,22 +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.CampaignID == nil { - // The campaign is gone; there is nothing to replay into. - return false, true, nil + if ct == nil || ct.DispatchIntent != repository.CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil { + return false, false, nil + } + if ct.CampaignID == 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 } @@ -2680,10 +2674,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 +2698,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