diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index afe40e559..1db14b183 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -208,6 +208,12 @@ jobs: go run ./cmd/migrate go test -race ./internal/app/behavior ./internal/repository ./internal/tasks -run 'TestDefaultPlan|TestLiveMailboxPlan|TestLiveCampaignAdmission|TestLiveSharedAdmission|TestLiveDeferredTicksDoNotSpendTheDailyBudget|TestLiveCampaignAtDailyCapWaitsForTomorrow' -count=1 + - name: Shared mailbox pool pacing and send claim regressions (PostgreSQL) + env: + WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable + run: | + go test -race -p 1 ./internal/repository ./internal/scheduler ./internal/tasks -run '^TestLive(MailboxSendClaim(AdmitsOnePerGap|WaitsOutAReservedSend)|BookedSendsLeaveOutCampaignWakeups|ParkedCampaignDoesNotThrottleItsPoolMate|BookedSendsStillSpaceTheCampaign|ClaimedMailboxIsPassedOver|CampaignsSharingAMailboxSendOncePerGap|LostMailboxClaimReschedulesWithoutSending)$' -count=1 + - name: Invited signup atomic provisioning regressions (PostgreSQL) env: WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable diff --git a/AGENTS.md b/AGENTS.md index 95ca9f2cc..8d27b2311 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1030,6 +1030,7 @@ What resolves a reservation is the worker's per-task result on `jobs.worker-even Rules that follow from this: - never dispatch a send that is not reserved. A reservation that cannot be written means the task retries; a claim refused (`reserved == false`) means another tick already has that pair in flight, and the task ends `skipped_duplicate` +- a pending `campaign` task is a chain wake-up, never a booked send: it re-picks its mailbox when it fires. Slot conflict resolution reads `GetBookedSendsForAccount`, which leaves them out; spacing around another campaign's parked wake-up throttled every campaign sharing that pool (issue #942). The per-mailbox gap across campaigns is held by `ClaimMailboxSend` (`email_accounts.send_claimed_at`), taken before the send is built and visible on the min-gap clock; a lost claim reschedules the pass within seconds and sends nothing - resolve every reservation exactly once: `RecordEmailSent` on a successful hand-off, `ReleaseSend` when the command PROVABLY never left (no worker, worker offline), `RecordSendFailure` on a worker failure. A failure of the publish call itself is ambiguous (`tasks.ErrSendDispatchUnknown`) and must NOT be released - a reservation nobody answers is resolved by `StartStuckSendReclaimer` in the consumer after `config.CampaignSendReclaimAfterMinutes`: it believes a task that carries a Message-ID (stamps it) and otherwise walks the step back as a failed attempt. Without it a dead worker parks a lead in flight forever - the day's counters belong to the reservation, not the stamp, so a lost stamp can never let the daily cap over-send diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index 6ec2dfd6c..3635188c0 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -170,6 +170,14 @@ Whichever mode you pick, it decides which mailbox **starts** a lead. Every mailb The Leads tab shows each lead's mailbox in its **Sender** column, and a contact's Activity tab shows it beside their progress. A lead has none until its first email goes out. +### Campaigns sharing mailboxes + +Several campaigns can send from the same mailboxes, for example by sharing a tag. What they share is each mailbox's minimum gap and daily cap. A campaign that is outside its sending window, or has nothing due, waits without holding any mailbox, so a campaign that is sending gets the pool's full pace in the meantime. + +Two campaigns never send from one mailbox inside its minimum gap. When both reach for the same mailbox at the same moment, one sends and the other looks again a few seconds later, taking another mailbox whose gap has passed or waiting for this one. Rotation also breaks ties differently for each campaign, so campaigns sharing a pool tend to start on different mailboxes rather than queueing for the same one. + +Because the daily cap belongs to the mailbox, the campaign whose window opens first can use most of a shared pool's day, which matters when the campaigns send in different timezones. To keep part of the pool for a later campaign, lower the earlier campaign's [daily limit per mailbox](#sending-limits), which caps what it takes from each mailbox. To split the pool outright, give each campaign its own tag. + **ESP matching** optionally aligns sender and recipient providers (Google to Google, Microsoft to Microsoft). The recipient's provider is read from their domain's mail records, so a lead on a company domain hosted by Google Workspace or Microsoft 365 is matched too, not only `gmail.com` and `outlook.com` addresses; the contacts list shows it as a badge on each contact's avatar. A mailbox connected with an app password to Google Workspace or Microsoft 365 counts as that provider. Under **Settings** > **ESP matching**, the coverage panel shows how many of the campaign's leads are on Google, Microsoft and other providers, and which of your mailboxes serve each. **Prefer same** falls back to any mailbox when no same-provider one is free; **Strict same** makes the contact wait instead. A contact left waiting this way waits alone: the campaign moves on to the next lead in the queue rather than stopping, and comes back to them when a matching mailbox is free. An unknown recipient provider never blocks a send. Like rotation, it applies to a lead's first email: a follow-up goes out from the mailbox the contact already knows. ### When a lead's mailbox is busy, or gone diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index b0498fd78..78c2f2a59 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -252,11 +252,13 @@ var Tables = []Table{ // avatar_checked_at is this instance's photo sweep checkpoint; the photo travels. // A tracking host is verified per instance and per workspace, so the // destination's own sweep verifies it again. + // send_claimed_at is this instance's per-send gap lock. ResetOnImport: []string{ "transport_generation", "worker_id", "auth_checked_at", "auth_failing_since", "cold_ramp_started_at", "send_lifecycle", "send_lifecycle_since", "send_lifecycle_reason", "seed_scope", "avatar_checked_at", "tracking_domain_verified", "tracking_domain_verified_at", + "send_claimed_at", }, Blobs: []BlobColumn{{Column: "avatar_url", Kind: BlobKindPublicURL}}, }, diff --git a/internal/config/constants.go b/internal/config/constants.go index 16fbe1adc..09381470a 100644 --- a/internal/config/constants.go +++ b/internal/config/constants.go @@ -238,6 +238,10 @@ const ( // task every second, so a retry kept at its old slot would run in a loop. CampaignTickRetrySeconds = 60 + // CampaignSendClaimRetrySeconds is how soon a pass that lost its mailbox to + // another campaign looks again, plus up to as much again of spread. + CampaignSendClaimRetrySeconds = 10 + // CampaignStaleParkHours is when the reconciler starts distrusting a parked // wakeup. Even-distribution can only push a successor to the end of the // mailbox's current day, so a pending tick further out than this was parked diff --git a/internal/infrastructure/db/migrations/000289_mailbox_send_claim.down.sql b/internal/infrastructure/db/migrations/000289_mailbox_send_claim.down.sql new file mode 100644 index 000000000..c9518d9c4 --- /dev/null +++ b/internal/infrastructure/db/migrations/000289_mailbox_send_claim.down.sql @@ -0,0 +1 @@ +ALTER TABLE email_accounts DROP COLUMN IF EXISTS send_claimed_at; diff --git a/internal/infrastructure/db/migrations/000289_mailbox_send_claim.up.sql b/internal/infrastructure/db/migrations/000289_mailbox_send_claim.up.sql new file mode 100644 index 000000000..c75999ae1 --- /dev/null +++ b/internal/infrastructure/db/migrations/000289_mailbox_send_claim.up.sql @@ -0,0 +1,3 @@ +-- When a campaign pass last took this mailbox for a send, so two campaigns +-- sharing it cannot both send inside its minimum gap. +ALTER TABLE email_accounts ADD COLUMN IF NOT EXISTS send_claimed_at timestamptz; diff --git a/internal/repository/mailbox_send_claim.go b/internal/repository/mailbox_send_claim.go new file mode 100644 index 000000000..047e4a66d --- /dev/null +++ b/internal/repository/mailbox_send_claim.go @@ -0,0 +1,44 @@ +package repository + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +// sendClaimClock feeds a claimed-but-not-yet-handed-off campaign send into the +// min-gap clock ($1 is the mailbox id array), so the next pass of any campaign +// sees the mailbox as just used. +const sendClaimClock = `SELECT ea.id, ea.send_claimed_at FROM email_accounts ea + WHERE ea.id = ANY($1) AND ea.send_claimed_at IS NOT NULL` + +// MailboxSendClaimer takes a mailbox for one campaign send. +type MailboxSendClaimer interface { + ClaimMailboxSend(ctx context.Context, mailboxID uuid.UUID, window time.Duration) (bool, error) +} + +// ClaimMailboxSend takes the mailbox for a send now, refused while another +// claim or a reserved send of any lane sits inside window. The conditional +// update serialises concurrent passes on the row, so two campaigns sharing a +// mailbox cannot both win the same gap. +func (r *taskRepository) ClaimMailboxSend(ctx context.Context, mailboxID uuid.UUID, window time.Duration) (bool, error) { + var claimed bool + err := r.db.QueryRow(ctx, ` + UPDATE email_accounts ea SET send_claimed_at = NOW() + WHERE ea.id = $1 + AND (ea.send_claimed_at IS NULL OR ea.send_claimed_at <= NOW() - make_interval(secs => $2)) + AND NOT EXISTS ( + SELECT 1 FROM tasks t + WHERE t.email_account_id = $1 + AND t.send_reserved_at > NOW() - make_interval(secs => $2) + AND t.send_released_at IS NULL) + RETURNING true + `, mailboxID, window.Seconds()).Scan(&claimed) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + return claimed, err +} diff --git a/internal/repository/mailbox_send_claim_live_test.go b/internal/repository/mailbox_send_claim_live_test.go new file mode 100644 index 000000000..942606b79 --- /dev/null +++ b/internal/repository/mailbox_send_claim_live_test.go @@ -0,0 +1,129 @@ +package repository + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" +) + +// Live checks of the per-mailbox send claim and of what counts as a booked +// send (issue #942). Skipped unless WARMBLY_TEST_DB is set. + +func TestLiveMailboxSendClaimAdmitsOnePerGap(t *testing.T) { + _, pool := liveContactDB(t) + requireSchemaVersion(t, pool, 289) + ctx := context.Background() + f := newWarmupUsageFixture(t, pool) + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM tasks WHERE email_account_id = $1`, f.account) + }) + r := NewTaskRepository(pool).(*taskRepository) + window := 9 * time.Minute + + // Passes of several campaigns reach for the same mailbox at once. + var won atomic.Int32 + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + ok, err := r.ClaimMailboxSend(ctx, f.account, window) + if err != nil { + t.Errorf("claim: %v", err) + } + if ok { + won.Add(1) + } + }() + } + wg.Wait() + if n := won.Load(); n != 1 { + t.Fatalf("%d concurrent claims won the same gap, want exactly 1", n) + } + + // The claim is on the min-gap clock, single and batch, before any send. + last, err := r.GetLastEmailTime(ctx, f.account) + if err != nil || last == nil || time.Since(*last) > time.Minute { + t.Fatalf("last email time = %v (err %v), want the claim just taken", last, err) + } + lasts, err := r.GetLastEmailTimes(ctx, []uuid.UUID{f.account}) + if err != nil || time.Since(lasts[f.account]) > time.Minute { + t.Fatalf("batch last email times = %v (err %v), want the claim just taken", lasts, err) + } + + // Once the window has passed the mailbox can be taken again. + f.exec(`UPDATE email_accounts SET send_claimed_at = NOW() - INTERVAL '10 minutes' WHERE id = $1`, f.account) + if ok, err := r.ClaimMailboxSend(ctx, f.account, window); err != nil || !ok { + t.Fatalf("claim after the window: ok=%v err=%v, want it taken", ok, err) + } +} + +func TestLiveMailboxSendClaimWaitsOutAReservedSend(t *testing.T) { + _, pool := liveContactDB(t) + requireSchemaVersion(t, pool, 289) + ctx := context.Background() + f := newWarmupUsageFixture(t, pool) + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM tasks WHERE email_account_id = $1`, f.account) + }) + r := NewTaskRepository(pool).(*taskRepository) + + // A warmup send reserved a minute ago holds the mailbox like a claim does. + reserved := uuid.New() + f.exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id, completed_at, send_reserved_at) + VALUES ($1, 'warmup', $2, 'completed', '', NOW() - INTERVAL '1 minute', NOW() - INTERVAL '1 minute')`, + reserved, f.account) + if ok, err := r.ClaimMailboxSend(ctx, f.account, 5*time.Minute); err != nil || ok { + t.Fatalf("claim inside a reserved send's gap: ok=%v err=%v, want it refused", ok, err) + } + + // A reservation given back never left, so it holds nothing. + f.exec(`UPDATE tasks SET send_released_at = NOW() WHERE id = $1`, reserved) + if ok, err := r.ClaimMailboxSend(ctx, f.account, 5*time.Minute); err != nil || !ok { + t.Fatalf("claim after the reservation was released: ok=%v err=%v, want it taken", ok, err) + } +} + +func TestLiveBookedSendsLeaveOutCampaignWakeups(t *testing.T) { + _, pool := liveContactDB(t) + ctx := context.Background() + f := newWarmupUsageFixture(t, pool) + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM tasks WHERE email_account_id = $1`, f.account) + }) + r := NewTaskRepository(pool) + now := time.Now() + + park, warmup, nextDay, farOff := uuid.New(), uuid.New(), uuid.New(), uuid.New() + f.exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at) + VALUES ($1, 'campaign', $2, 'pending', '', $3)`, park, f.account, now.Add(10*time.Minute)) + f.exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at) + VALUES ($1, 'warmup', $2, 'pending', '', $3)`, warmup, f.account, now.Add(12*time.Minute)) + // Later the next day is still within reach of the candidate. + f.exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at) + VALUES ($1, 'email', $2, 'pending', '', $3)`, nextDay, f.account, now.Add(20*time.Hour)) + f.exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at) + VALUES ($1, 'warmup', $2, 'pending', '', $3)`, farOff, f.account, now.Add(50*time.Hour)) + + booked, err := r.GetBookedSendsForAccount(ctx, f.account, now) + if err != nil { + t.Fatalf("booked sends: %v", err) + } + got := map[uuid.UUID]bool{} + for _, b := range booked { + got[b.ID] = true + } + if got[park] { + t.Fatal("another campaign's parked wake-up was returned as a booked send") + } + if !got[warmup] || !got[nextDay] { + t.Fatalf("booked sends %v, want the warmup send and the next-day email", got) + } + if got[farOff] { + t.Fatal("a send two days out was returned as near the candidate") + } +} diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go index 450436617..a1dc70d85 100644 --- a/internal/repository/pg_task.go +++ b/internal/repository/pg_task.go @@ -144,8 +144,7 @@ type TaskRepository interface { // GetLastSendTimes is the batch form, for rotation across a campaign's // whole sender pool. Accounts that have never sent are absent from the map. GetLastSendTimes(ctx context.Context, accountIDs []uuid.UUID, taskType string) (map[uuid.UUID]time.Time, error) - GetScheduledTasksForAccount(ctx context.Context, accountID uuid.UUID, date time.Time) ([]Task, error) - GetScheduledTasksToday(ctx context.Context, accountID uuid.UUID) ([]Task, error) + GetBookedSendsForAccount(ctx context.Context, accountID uuid.UUID, around time.Time) ([]Task, error) // ListDuePendingTaskIDs returns pending tasks whose scheduled_at has passed, // oldest first, capped at limit. Drives the in-process (TASKS_PROVIDER=local) // dispatcher, which fires each due task by id. @@ -521,13 +520,17 @@ func (r *taskRepository) GetLastEmailTimes(ctx context.Context, accountIDs []uui return out, nil } query := ` - SELECT t.email_account_id, MAX(t.completed_at) - FROM tasks t - WHERE t.email_account_id = ANY($1) - AND t.status = 'completed' - AND t.completed_at IS NOT NULL - AND ` + taskDispatchedEmail + ` - GROUP BY t.email_account_id + SELECT id, MAX(at) FROM ( + SELECT t.email_account_id AS id, t.completed_at AS at + FROM tasks t + WHERE t.email_account_id = ANY($1) + AND t.status = 'completed' + AND t.completed_at IS NOT NULL + AND ` + taskDispatchedEmail + ` + UNION ALL + ` + sendClaimClock + ` + ) last + GROUP BY id ` rows, err := r.db.Query(ctx, query, accountIDs) if err != nil { @@ -736,14 +739,17 @@ func (r *taskRepository) CountWarmupEmailsSentToday(ctx context.Context, account } // GetLastEmailTime gets the last email send time for an account. It is the -// min-gap clock, so it reads real sends only. +// min-gap clock, so it reads real sends and campaign sends claimed but not yet +// handed off. func (r *taskRepository) GetLastEmailTime(ctx context.Context, accountID uuid.UUID) (*time.Time, error) { query := ` - SELECT MAX(t.completed_at) - FROM tasks t - WHERE t.email_account_id = $1 - AND t.status = 'completed' - AND ` + taskDispatchedEmail + ` + SELECT GREATEST( + (SELECT MAX(t.completed_at) + FROM tasks t + WHERE t.email_account_id = $1 + AND t.status = 'completed' + AND ` + taskDispatchedEmail + `), + (SELECT send_claimed_at FROM email_accounts WHERE id = $1)) ` var lastTime *time.Time @@ -799,19 +805,23 @@ func (r *taskRepository) GetLastSendTimes(ctx context.Context, accountIDs []uuid return out, rows.Err() } -// GetScheduledTasksForAccount gets scheduled tasks for an account on a specific date -func (r *taskRepository) GetScheduledTasksForAccount(ctx context.Context, accountID uuid.UUID, date time.Time) ([]Task, error) { +// GetBookedSendsForAccount returns the pending sends booked on a mailbox within +// a day either side of `around`. A pending campaign task is a chain wake-up, +// not a send: it picks its mailbox again when it fires, so it is left out. +func (r *taskRepository) GetBookedSendsForAccount(ctx context.Context, accountID uuid.UUID, around time.Time) ([]Task, error) { query := ` SELECT id, task_type, email_account_id, status, message_id, scheduled_at, completed_at, cloud_task_name, created_at, updated_at FROM tasks WHERE email_account_id = $1 AND status = 'pending' - AND DATE(scheduled_at) = DATE($2) + AND task_type <> 'campaign' + AND scheduled_at > $2::timestamptz - INTERVAL '1 day' + AND scheduled_at < $2::timestamptz + INTERVAL '1 day' ORDER BY scheduled_at ASC ` - rows, err := r.db.Query(ctx, query, accountID, date) + rows, err := r.db.Query(ctx, query, accountID, around) if err != nil { return nil, err } @@ -841,11 +851,6 @@ func (r *taskRepository) GetScheduledTasksForAccount(ctx context.Context, accoun return tasks, rows.Err() } -// GetScheduledTasksToday gets scheduled tasks for an account today -func (r *taskRepository) GetScheduledTasksToday(ctx context.Context, accountID uuid.UUID) ([]Task, error) { - return r.GetScheduledTasksForAccount(ctx, accountID, time.Now()) -} - // UpdateTaskStatus updates the status of a task func (r *taskRepository) UpdateTaskStatus(ctx context.Context, taskID uuid.UUID, status string) error { query := ` diff --git a/internal/scheduler/behavior_test.go b/internal/scheduler/behavior_test.go index 728aa7b36..aff7dea0a 100644 --- a/internal/scheduler/behavior_test.go +++ b/internal/scheduler/behavior_test.go @@ -360,7 +360,7 @@ func TestRoundRobinFallsBackToTodaysSendCount(t *testing.T) { middle := candidateWithoutSenderRow("22222222-2222-2222-2222-222222222222", 4, nil) idle := candidateWithoutSenderRow("33333333-3333-3333-3333-333333333333", 0, nil) - got := selectAccountByRotationMode("round_robin", []AccountCandidate{busiest, middle, idle}) + got := selectAccountByRotationMode("round_robin", []AccountCandidate{busiest, middle, idle}, uuid.Nil) if got == nil { t.Fatal("expected a selection") } @@ -377,7 +377,7 @@ func TestLeastRecentlyUsedFallsBackToRealSendHistory(t *testing.T) { a := candidateWithoutSenderRow("11111111-1111-1111-1111-111111111111", 0, &recent) b := candidateWithoutSenderRow("22222222-2222-2222-2222-222222222222", 0, &older) - got := selectAccountByRotationMode("least_recently_used", []AccountCandidate{a, b}) + got := selectAccountByRotationMode("least_recently_used", []AccountCandidate{a, b}, uuid.Nil) if got == nil { t.Fatal("expected a selection") } @@ -391,7 +391,7 @@ func TestLeastRecentlyUsedPrefersAMailboxThatNeverSent(t *testing.T) { used := candidateWithoutSenderRow("11111111-1111-1111-1111-111111111111", 0, &sent) fresh := candidateWithoutSenderRow("22222222-2222-2222-2222-222222222222", 0, nil) - got := selectAccountByRotationMode("least_recently_used", []AccountCandidate{used, fresh}) + got := selectAccountByRotationMode("least_recently_used", []AccountCandidate{used, fresh}, uuid.Nil) if got == nil || got.Account.ID != fresh.Account.ID { t.Fatalf("a mailbox with no send history should sort first, got %+v", got) } @@ -413,7 +413,7 @@ func TestRotationRotatesAcrossRepeatedPicks(t *testing.T) { for _, id := range ids { pool = append(pool, candidateWithoutSenderRow(id, sentToday[id], nil)) } - got := selectAccountByRotationMode("round_robin", pool) + got := selectAccountByRotationMode("round_robin", pool, uuid.Nil) if got == nil { t.Fatal("expected a selection") } diff --git a/internal/scheduler/campaign_scheduler.go b/internal/scheduler/campaign_scheduler.go index 6d45b1ecb..c61046a78 100644 --- a/internal/scheduler/campaign_scheduler.go +++ b/internal/scheduler/campaign_scheduler.go @@ -751,7 +751,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode pick = clear } } - selected = selectAccountByRotationMode(campaign.RotationMode, pick) + selected = selectAccountByRotationMode(campaign.RotationMode, pick, campaign.ID) } } if selected == nil { @@ -887,13 +887,16 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode } candidateTime = notBefore(candidateTime) - // STEP 12: Check conflicts with other scheduled tasks + // STEP 12: Keep clear of sends already booked on this mailbox. Another + // campaign's parked wake-up is not one (issue #942): it re-picks its + // mailbox when it fires, and the gap is enforced then by STEP 10 and the + // send claim, so spacing around it only throttles campaigns sharing a pool. if !preview { - scheduledTasks, terr := s.taskRepo.GetScheduledTasksForAccount(ctx, account.ID, candidateTime) + booked, terr := s.taskRepo.GetBookedSendsForAccount(ctx, account.ID, candidateTime) if terr != nil { return time.Time{}, nil, uuid.Nil, terr } - candidateTime = resolveConflicts(candidateTime, scheduledTasks, gapSeconds) + candidateTime = resolveConflicts(candidateTime, booked, gapSeconds) } // STEP 13: Apply human-like distribution (favor morning/afternoon peaks). diff --git a/internal/scheduler/email_scheduler.go b/internal/scheduler/email_scheduler.go index ee471a860..7e151b671 100644 --- a/internal/scheduler/email_scheduler.go +++ b/internal/scheduler/email_scheduler.go @@ -52,12 +52,12 @@ func (s *schedulerService) CalculateNextEmailTime(ctx context.Context, accountID jitter := randomJitter(-10, 10) candidateTime = notBefore(candidateTime.Add(time.Minute * time.Duration(jitter))) - // STEP 5: Resolve conflicts with scheduled tasks - scheduledTasks, err := s.taskRepo.GetScheduledTasksForAccount(ctx, accountID, candidateTime) + // STEP 5: Resolve conflicts with sends booked on this account + booked, err := s.taskRepo.GetBookedSendsForAccount(ctx, accountID, candidateTime) if err != nil { return time.Time{}, err } - candidateTime = resolveConflicts(candidateTime, scheduledTasks, gapSeconds) + candidateTime = resolveConflicts(candidateTime, booked, gapSeconds) // STEP 6: Apply distribution curve. Skipped when a profile owns the day. if !bhv.Enabled { diff --git a/internal/scheduler/errors.go b/internal/scheduler/errors.go index b664d2f33..8d7c71d8c 100644 --- a/internal/scheduler/errors.go +++ b/internal/scheduler/errors.go @@ -3,6 +3,7 @@ package scheduler import ( "errors" "fmt" + "math/rand" "time" "github.com/warmbly/warmbly/internal/config" @@ -111,6 +112,13 @@ func DeferSlot(at time.Time) time.Time { return at } +// ClaimRetrySlot is where a pass that lost its mailbox to another campaign +// looks again: soon, and spread so the losers do not collide a second time. +func ClaimRetrySlot() time.Time { + secs := config.CampaignSendClaimRetrySeconds + rand.Intn(config.CampaignSendClaimRetrySeconds+1) + return time.Now().Add(time.Duration(secs) * time.Second) +} + // WakeSlot is where a chain seeded outside a tick first wakes: now for a // sendable pair, whose time is the paced slot of the send after it. func WakeSlot(at time.Time, err error) time.Time { diff --git a/internal/scheduler/helpers.go b/internal/scheduler/helpers.go index 94e1de72b..19f862f83 100644 --- a/internal/scheduler/helpers.go +++ b/internal/scheduler/helpers.go @@ -1,11 +1,13 @@ package scheduler import ( + "hash/fnv" "math" "math/rand" "sort" "time" + "github.com/google/uuid" "github.com/warmbly/warmbly/internal/app/behavior" "github.com/warmbly/warmbly/internal/bitmask" "github.com/warmbly/warmbly/internal/models" @@ -492,38 +494,38 @@ func effectiveWeight(c AccountCandidate) float64 { // selectAccountLeastRecentlyUsed picks the viable candidate (Weight>0) whose // sender last_sent_at is oldest; a nil last_sent_at (never used) sorts first. -// Ties broken by account id for determinism. -func selectAccountLeastRecentlyUsed(candidates []AccountCandidate) *AccountCandidate { +// Ties go to tieRank under the campaign's salt. +func selectAccountLeastRecentlyUsed(candidates []AccountCandidate, salt uuid.UUID) *AccountCandidate { var best *AccountCandidate for i := range candidates { if candidates[i].Weight <= 0 { continue } - if best == nil || lruLess(&candidates[i], best) { + if best == nil || lruLess(&candidates[i], best, salt) { best = &candidates[i] } } return best } -func lruLess(a, b *AccountCandidate) bool { +func lruLess(a, b *AccountCandidate, salt uuid.UUID) bool { switch { case a.SenderLastSentAt == nil && b.SenderLastSentAt == nil: - return a.Account.ID.String() < b.Account.ID.String() + return tieRank(salt, a.Account.ID) < tieRank(salt, b.Account.ID) case a.SenderLastSentAt == nil: return true case b.SenderLastSentAt == nil: return false case a.SenderLastSentAt.Equal(*b.SenderLastSentAt): - return a.Account.ID.String() < b.Account.ID.String() + return tieRank(salt, a.Account.ID) < tieRank(salt, b.Account.ID) default: return a.SenderLastSentAt.Before(*b.SenderLastSentAt) } } // selectAccountRoundRobin picks the viable candidate (Weight>0) with the lowest -// rotation_position; ties broken by account id for determinism. -func selectAccountRoundRobin(candidates []AccountCandidate) *AccountCandidate { +// rotation_position; ties go to tieRank under the campaign's salt. +func selectAccountRoundRobin(candidates []AccountCandidate, salt uuid.UUID) *AccountCandidate { var best *AccountCandidate for i := range candidates { if candidates[i].Weight <= 0 { @@ -532,13 +534,22 @@ func selectAccountRoundRobin(candidates []AccountCandidate) *AccountCandidate { if best == nil || candidates[i].RotationPosition < best.RotationPosition || (candidates[i].RotationPosition == best.RotationPosition && - candidates[i].Account.ID.String() < best.Account.ID.String()) { + tieRank(salt, candidates[i].Account.ID) < tieRank(salt, best.Account.ID)) { best = &candidates[i] } } return best } +// tieRank orders tied mailboxes per campaign: stable for one campaign, but +// different campaigns sharing a pool do not all reach for the same mailbox. +func tieRank(salt, id uuid.UUID) uint64 { + h := fnv.New64a() + _, _ = h.Write(salt[:]) + _, _ = h.Write(id[:]) + return h.Sum64() +} + // needsRotationFallback reports whether a rotation mode depends on per-sender // bookkeeping that only explicitly-picked senders carry. Weighted selection is // driven by remaining capacity, so it already spreads without a cursor. @@ -548,13 +559,14 @@ func needsRotationFallback(rotationMode string) bool { // selectAccountByRotationMode dispatches to the chosen rotation strategy. For // explicit-strategy campaigns rotationMode is one of weighted / round_robin / -// least_recently_used; anything else falls back to weighted. -func selectAccountByRotationMode(rotationMode string, candidates []AccountCandidate) *AccountCandidate { +// least_recently_used; anything else falls back to weighted. salt is the +// campaign id, which decides ties. +func selectAccountByRotationMode(rotationMode string, candidates []AccountCandidate, salt uuid.UUID) *AccountCandidate { switch rotationMode { case "round_robin": - return selectAccountRoundRobin(candidates) + return selectAccountRoundRobin(candidates, salt) case "least_recently_used": - return selectAccountLeastRecentlyUsed(candidates) + return selectAccountLeastRecentlyUsed(candidates, salt) default: return selectAccountWeighted(candidates) } diff --git a/internal/scheduler/send_claim.go b/internal/scheduler/send_claim.go new file mode 100644 index 000000000..9c2e97e3d --- /dev/null +++ b/internal/scheduler/send_claim.go @@ -0,0 +1,35 @@ +package scheduler + +import ( + "context" + "time" + + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +// SendClaimer is the optional capability a campaign pass uses to take its +// mailbox before building the send. +type SendClaimer interface { + ClaimSend(ctx context.Context, account *models.Email) (bool, error) +} + +// ClaimSend takes the mailbox for a send now. False means another pass took +// it inside its gap since this pass read the pool, and this one should look +// again rather than send. +func (s *schedulerService) ClaimSend(ctx context.Context, account *models.Email) (bool, error) { + claimer, ok := s.taskRepo.(repository.MailboxSendClaimer) + if !ok { + return true, nil + } + gap := s.behaviorGapFloor(s.behaviorFor(ctx, account), time.Now(), account.MinWaitTime) + return claimer.ClaimMailboxSend(ctx, account.ID, sendClaimWindow(gap)) +} + +// sendClaimWindow is how long one claim holds a mailbox: its gap less the +// grace a due pass may send early by, never under half the gap. +func sendClaimWindow(gapSeconds int) time.Duration { + window := max(gapSeconds-config.CampaignNotDueGraceSeconds, gapSeconds/2) + return time.Duration(window) * time.Second +} diff --git a/internal/scheduler/shared_pool_live_test.go b/internal/scheduler/shared_pool_live_test.go new file mode 100644 index 000000000..822b2c40e --- /dev/null +++ b/internal/scheduler/shared_pool_live_test.go @@ -0,0 +1,129 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" +) + +// longOpenTimezone is a zone whose 8am-8pm mailbox band stays open for the +// next six hours, so a slot pushed past four hours is the conflict step's +// doing and not the band closing or the distribution curve. +func longOpenTimezone(t *testing.T) string { + t.Helper() + open := func(at time.Time) bool { return at.Hour() >= 8 && at.Hour() < 20 } + for _, name := range []string{"UTC", "America/New_York", "America/Los_Angeles", "Asia/Tokyo", + "Europe/Berlin", "Asia/Dubai", "Pacific/Auckland", "Pacific/Honolulu", "Asia/Kolkata", "America/Sao_Paulo"} { + l, err := time.LoadLocation(name) + if err != nil { + continue + } + now := time.Now().In(l) + if open(now) && open(now.Add(6*time.Hour)) { + return name + } + } + t.Skip("no timezone in the spread stays inside 8am-8pm for the next six hours") + return "" +} + +// longOpenFixture is the live fixture with the campaign on the mailbox's own +// clock, so the distribution curve moves a slot by two hours at most. +func longOpenFixture(t *testing.T) (*liveFixture, SchedulerService) { + t.Helper() + handle, pool := liveDB(t) + tz := longOpenTimezone(t) + f := newLiveFixture(t, pool, tz) + if _, err := pool.Exec(context.Background(), `UPDATE campaigns SET timezone = $2 WHERE id = $1`, f.campaign, tz); err != nil { + t.Fatalf("campaign timezone: %v", err) + } + return f, liveScheduler(t, handle, pool) +} + +// fillMailboxDay books a task of taskType on the mailbox every five minutes +// for a day, closer together than its 600s gap, so any slot near one of them +// conflicts. +func fillMailboxDay(t *testing.T, f *liveFixture, taskType string) { + t.Helper() + if _, err := f.pool.Exec(context.Background(), ` + INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at) + SELECT gen_random_uuid(), $1::task_type, $2, 'pending', '', s + FROM generate_series(NOW() - INTERVAL '1 hour', NOW() + INTERVAL '24 hours', INTERVAL '5 minutes') s`, + taskType, f.mailbox); err != nil { + t.Fatalf("book %s tasks: %v", taskType, err) + } +} + +// TestLiveParkedCampaignDoesNotThrottleItsPoolMate is issue #942: another +// campaign's deferral parks on the shared mailbox are wake-ups, not sends, and +// must not push the active campaign's slot behind them. +func TestLiveParkedCampaignDoesNotThrottleItsPoolMate(t *testing.T) { + f, s := longOpenFixture(t) + fillMailboxDay(t, f, "campaign") + + at, pair, _, err := s.CalculateNextCampaignTime(context.Background(), f.campaign) + if err != nil || pair == nil { + t.Fatalf("pair=%v err=%v, want a sendable lead", pair, err) + } + if limit := time.Now().Add(4 * time.Hour); at.After(limit) { + t.Fatalf("slot %s was pushed behind another campaign's wake-ups (limit %s)", at, limit) + } +} + +// TestLiveBookedSendsStillSpaceTheCampaign is the control: a real send booked +// on the mailbox still moves the slot. +func TestLiveBookedSendsStillSpaceTheCampaign(t *testing.T) { + f, s := longOpenFixture(t) + fillMailboxDay(t, f, "warmup") + + at, pair, _, err := s.CalculateNextCampaignTime(context.Background(), f.campaign) + if err != nil || pair == nil { + t.Fatalf("pair=%v err=%v, want a sendable lead", pair, err) + } + if limit := time.Now().Add(4 * time.Hour); !at.After(limit) { + t.Fatalf("slot %s ignored the warmup sends booked on the mailbox", at) + } +} + +// TestLiveClaimedMailboxIsPassedOver: a mailbox another campaign has just +// taken counts on the min-gap clock, so the pass sends from the other one. +func TestLiveClaimedMailboxIsPassedOver(t *testing.T) { + handle, pool := liveDB(t) + tz := openHoursTimezone(t) + f := newLiveFixture(t, pool, tz) + ctx := context.Background() + + second := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO email_accounts (id, user_id, organization_id, email, name, + signature_plain, signature_html, provider, status, campaign_limit, min_wait_time, timezone) + VALUES ($1, $2, $3, $4, 'Second', '', '', 'smtp_imap', 'active', 50, 600, $5)`, + second, f.user, f.org, "second-"+second.String()[:8]+"@test.local", tz); err != nil { + t.Fatalf("add mailbox: %v", err) + } + t.Cleanup(func() { + c := context.Background() + _, _ = pool.Exec(c, `DELETE FROM tasks WHERE email_account_id = $1`, second) + _, _ = pool.Exec(c, `DELETE FROM email_accounts WHERE id = $1`, second) + }) + // Least-recently-used would pick the mailbox that never sent, and that is + // the one another campaign has just taken. + if _, err := pool.Exec(ctx, ` + INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at, completed_at) + VALUES ($1, 'campaign', $2, 'completed', '', $3, $3)`, + uuid.New(), second, earlierToday(2*time.Hour)); err != nil { + t.Fatalf("earlier send: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE email_accounts SET send_claimed_at = NOW() WHERE id = $1`, f.mailbox); err != nil { + t.Fatalf("claim: %v", err) + } + + _, pair, accountID, err := liveScheduler(t, handle, pool).CalculateNextCampaignTime(ctx, f.campaign) + if err != nil || pair == nil { + t.Fatalf("pair=%v err=%v, want a send from the unclaimed mailbox", pair, err) + } + if accountID != second { + t.Fatalf("picked %s, want the mailbox nobody had claimed (%s)", accountID, second) + } +} diff --git a/internal/scheduler/shared_pool_test.go b/internal/scheduler/shared_pool_test.go new file mode 100644 index 000000000..28558ef7e --- /dev/null +++ b/internal/scheduler/shared_pool_test.go @@ -0,0 +1,50 @@ +package scheduler + +import ( + "testing" + "time" + + "github.com/google/uuid" +) + +// Campaigns sharing a pool used to break rotation ties by mailbox id, so every +// one of them reached for the same "next" mailbox (issue #942). +func TestRotationTiesSpreadAcrossCampaigns(t *testing.T) { + pool := make([]AccountCandidate, 0, 12) + for range 12 { + pool = append(pool, candidateWithoutSenderRow(uuid.NewString(), 0, nil)) + } + for _, mode := range []string{"round_robin", "least_recently_used"} { + picked := map[uuid.UUID]int{} + for range 60 { + campaign := uuid.New() + got := selectAccountByRotationMode(mode, pool, campaign) + if got == nil { + t.Fatalf("%s: expected a selection", mode) + } + if again := selectAccountByRotationMode(mode, pool, campaign); again.Account.ID != got.Account.ID { + t.Fatalf("%s: one campaign broke the same tie two ways", mode) + } + picked[got.Account.ID]++ + } + if len(picked) < 6 { + t.Fatalf("%s: 60 campaigns shared only %d of 12 tied mailboxes: %v", mode, len(picked), picked) + } + } +} + +func TestSendClaimWindowLeavesTheGraceAndCatchesCollisions(t *testing.T) { + for _, c := range []struct { + gap int + want time.Duration + }{ + {600, 540 * time.Second}, + {90, 45 * time.Second}, + {60, 30 * time.Second}, + {0, 0}, + } { + if got := sendClaimWindow(c.gap); got != c.want { + t.Fatalf("sendClaimWindow(%d) = %s, want %s", c.gap, got, c.want) + } + } +} diff --git a/internal/scheduler/warmup_scheduler.go b/internal/scheduler/warmup_scheduler.go index 3cdd2a33e..b00feda67 100644 --- a/internal/scheduler/warmup_scheduler.go +++ b/internal/scheduler/warmup_scheduler.go @@ -330,13 +330,13 @@ func (s *schedulerService) CalculateNextWarmupTime(ctx context.Context, accountI // STEP 10: Ensure within configured warmup time window candidateTime = ensureTimeWindow(candidateTime, warmupStart, warmupEnd, loadLocation(account.ClockTimezone())) - // STEP 11: Check conflicts with other tasks on this account - scheduledTasks, err := s.taskRepo.GetScheduledTasksToday(ctx, accountID) + // STEP 11: Check conflicts with other sends booked on this account + booked, err := s.taskRepo.GetBookedSendsForAccount(ctx, accountID, candidateTime) if err != nil { return time.Time{}, err } - candidateTime = resolveConflicts(candidateTime, scheduledTasks, minWaitSeconds) + candidateTime = resolveConflicts(candidateTime, booked, minWaitSeconds) // STEP 12: Apply human-like distribution curve. Skipped when a behaviour // profile is driving the day — it already carries this mailbox's own peak diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index 6ee4d36ed..b40fac690 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -495,6 +495,34 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx return nil } + // STEP 7.4: Load the sending mailbox and take it for this send. Campaigns + // sharing a pool read it independently, so two passes can pick the same + // mailbox in the same moment; only one claim wins its gap (issue #942). + account, xerr := s.emailRepo.GetByID(ctx, accountID) + if xerr != nil { + return xerr + } + if claimer, ok := s.scheduler.(scheduler.SendClaimer); ok { + claimed, cerr := claimer.ClaimSend(ctx, account) + if cerr != nil { + errs.CaptureException(cerr) + s.taskRepo.RecordTaskFailure(ctx, taskID, "Could not take the mailbox for the send", cerr.Error()) + s.retryCampaignTickLater(ctx, taskRecord) + executionStatus = "failed" + return errx.InternalError() + } + if !claimed { + // Nothing is reserved yet. The next pass sees the claim, so it + // picks a clear mailbox or waits out this one's gap. + if terr := s.createCampaignTask(ctx, campaign.ID, account.ID, scheduler.ClaimRetrySlot()); terr != nil { + log.Warn().Err(terr).Str("campaign_id", campaign.ID.String()).Str("task_id", taskID.String()).Msg("Failed to schedule the campaign's next pass after losing its mailbox to another campaign") + } + s.taskRepo.UpdateTaskStatus(ctx, taskID, "completed") + executionStatus = "completed" + return nil + } + } + // Load this step's attachments (campaign-wide files plus the step's own; // metadata only — the worker fetches the bytes from object storage by S3 // key at send time). @@ -516,12 +544,6 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx // sequence stops there, so there is no longer a blanket "contact has replied, // skip" check here — that would also kill the reply branch's follow-up emails. - // STEP 9: Load email account - account, xerr := s.emailRepo.GetByID(ctx, accountID) - if xerr != nil { - return xerr - } - // STEP 9.25: Stamp the mailbox this send actually goes out from. The chain // seeds a task with the previous tick's pick, and rotation chooses again // here, so tasks.email_account_id (today's budget, the min-gap clock, diff --git a/internal/tasks/shared_mailbox_live_test.go b/internal/tasks/shared_mailbox_live_test.go new file mode 100644 index 000000000..15b31d04c --- /dev/null +++ b/internal/tasks/shared_mailbox_live_test.go @@ -0,0 +1,192 @@ +package tasks + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" + "github.com/warmbly/warmbly/internal/scheduler" + "github.com/warmbly/warmbly/internal/tasks/proto" +) + +// Live checks that campaigns sharing a mailbox cannot both send inside its +// gap once their wake-ups no longer space each other (issue #942). + +// siblingCampaign adds a second campaign, step and lead to the fixture's +// workspace, sending from the same mailbox. +func (f *campaignSendFixture) siblingCampaign(t *testing.T) *campaignSendFixture { + t.Helper() + ctx := context.Background() + b := *f + b.campaign, b.step, b.contact = uuid.New(), uuid.New(), uuid.New() + b.contactEmail = "lead-" + b.contact.String()[:8] + "@test.local" + exec := func(sql string, args ...any) { + t.Helper() + if _, err := f.pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture %q: %v", sql[:min(60, len(sql))], err) + } + } + exec(`INSERT INTO campaigns (id, user_id, organization_id, name, description, status, + daily_limit, timezone, days, start_time, end_time, rotation_mode, updated_at, created_at) + VALUES ($1, $2, $3, 'Live Sibling', '', 'active', 50, 'UTC', 127, '00:00', '23:59', + 'least_recently_used', NOW(), NOW())`, b.campaign, b.user, b.org) + exec(`INSERT INTO sequences (id, campaign_id, organization_id, name, subject, + body_plain, body_html, wait_after, position, kind) + VALUES ($1, $2, $3, 'Step 1', 'Hi', 'Hello', '

Hello

', 0, 0, 'email')`, + b.step, b.campaign, b.org) + exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields) + VALUES ($1, $2, $3, $4, 'Live', 'Sibling', '', '', '{}')`, + b.contact, b.user, b.org, b.contactEmail) + exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, 0)`, b.campaign, b.contact) + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM task_execution_keys WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox}, + {`DELETE FROM task_failures WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox}, + {`DELETE FROM campaign_tasks WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox}, + {`DELETE FROM tasks WHERE email_account_id = $1`, f.mailbox}, + {`DELETE FROM campaign_logs WHERE campaign_id = $1`, b.campaign}, + {`DELETE FROM campaign_daily_sends WHERE campaign_id = $1`, b.campaign}, + {`DELETE FROM campaign_contact_progress WHERE campaign_id = $1`, b.campaign}, + {`DELETE FROM campaign_leads WHERE campaign_id = $1`, b.campaign}, + {`DELETE FROM sequences WHERE campaign_id = $1`, b.campaign}, + {`DELETE FROM campaigns WHERE id = $1`, b.campaign}, + } { + if _, err := f.pool.Exec(c, step.sql, step.arg); err != nil { + t.Errorf("cleanup %q: %v", step.sql, err) + } + } + }) + return &b +} + +// pendingPasses returns when each of the campaign's pending passes is due. +func (f *campaignSendFixture) pendingPasses(t *testing.T) []time.Time { + t.Helper() + rows, err := f.pool.Query(context.Background(), ` + SELECT t.scheduled_at FROM tasks t JOIN campaign_tasks ct ON ct.task_id = t.id + WHERE ct.campaign_id = $1 AND t.status = 'pending'`, f.campaign) + if err != nil { + t.Fatalf("read passes: %v", err) + } + defer rows.Close() + var out []time.Time + for rows.Next() { + var at time.Time + if err := rows.Scan(&at); err != nil { + t.Fatalf("scan pass: %v", err) + } + out = append(out, at) + } + return out +} + +func requireGap(t *testing.T, pool *pgxpool.Pool, mailbox uuid.UUID) { + t.Helper() + if _, err := pool.Exec(context.Background(), `UPDATE email_accounts SET min_wait_time = 600 WHERE id = $1`, mailbox); err != nil { + t.Fatalf("gap: %v", err) + } +} + +func TestLiveCampaignsSharingAMailboxSendOncePerGap(t *testing.T) { + handle := liveCampaignDB(t) + sender := &recordingSender{} + svc := liveCampaignService(t, handle, sender) + a := newCampaignSendFixture(t, handle.Pool) + requireGap(t, handle.Pool, a.mailbox) + b := a.siblingCampaign(t) + + ticks := []uuid.UUID{a.queueTick(t, svc.taskRepo), b.queueTick(t, svc.taskRepo)} + var wg sync.WaitGroup + start := make(chan struct{}) + for _, id := range ticks { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if xerr := svc.HandleCampaignTask(&proto.ProcessTask{TaskId: id.String()}); xerr != nil { + t.Errorf("campaign tick returned an error: %v", xerr) + } + }() + } + close(start) + wg.Wait() + + if n := sender.count(); n != 1 { + t.Fatalf("the shared mailbox sent %d emails inside its 600s gap, want 1", n) + } + // Whichever lost keeps its chain and comes back for its lead. + if len(a.pendingPasses(t)) != 1 || len(b.pendingPasses(t)) != 1 { + t.Fatalf("pending passes: a=%d b=%d, want one each", len(a.pendingPasses(t)), len(b.pendingPasses(t))) + } +} + +// racingScheduler lets another campaign take the mailbox between this pass's +// read of the pool and its claim. +type racingScheduler struct { + scheduler.SchedulerService + pool *pgxpool.Pool + mailbox uuid.UUID +} + +func (r racingScheduler) CalculateNextCampaignTime(ctx context.Context, id uuid.UUID) (time.Time, *repository.ContactSequencePair, uuid.UUID, error) { + at, pair, account, err := r.SchedulerService.CalculateNextCampaignTime(ctx, id) + if _, xerr := r.pool.Exec(ctx, `UPDATE email_accounts SET send_claimed_at = NOW() WHERE id = $1`, r.mailbox); xerr != nil { + return at, pair, account, xerr + } + return at, pair, account, err +} + +func (r racingScheduler) ClaimSend(ctx context.Context, account *models.Email) (bool, error) { + return r.SchedulerService.(scheduler.SendClaimer).ClaimSend(ctx, account) +} + +func TestLiveLostMailboxClaimReschedulesWithoutSending(t *testing.T) { + handle := liveCampaignDB(t) + sender := &recordingSender{} + svc := liveCampaignService(t, handle, sender) + f := newCampaignSendFixture(t, handle.Pool) + requireGap(t, handle.Pool, f.mailbox) + svc.scheduler = racingScheduler{SchedulerService: svc.scheduler, pool: handle.Pool, mailbox: f.mailbox} + + taskID := f.queueTick(t, svc.taskRepo) + if xerr := svc.HandleCampaignTask(&proto.ProcessTask{TaskId: taskID.String()}); xerr != nil { + t.Fatalf("campaign tick returned an error: %v", xerr) + } + + if n := sender.count(); n != 0 { + t.Fatalf("sent %d emails from a mailbox another campaign had just taken", n) + } + var dispatched, failures int + var status string + if err := handle.Pool.QueryRow(context.Background(), ` + SELECT (SELECT COUNT(*) FROM campaign_contact_progress WHERE campaign_id = $1 AND dispatched_at IS NOT NULL), + (SELECT COUNT(*) FROM task_failures WHERE task_id = $2), + (SELECT status::text FROM tasks WHERE id = $2)`, f.campaign, taskID).Scan(&dispatched, &failures, &status); err != nil { + t.Fatalf("read outcome: %v", err) + } + if dispatched != 0 || failures != 0 || status != "completed" { + t.Fatalf("dispatched=%d failures=%d status=%s, want a quiet pass that reserved nothing", dispatched, failures, status) + } + for _, e := range f.logEvents(t) { + if e == "email_failed" { + t.Fatal("a lost claim was reported to the campaign feed as a failed send") + } + } + passes := f.pendingPasses(t) + if len(passes) != 1 { + t.Fatalf("%d pending passes, want the one that looks again", len(passes)) + } + if wait := time.Until(passes[0]); wait < 5*time.Second || wait > 30*time.Second { + t.Fatalf("next pass in %s, want it back within seconds", wait) + } +}