Merge remote-tracking branch 'origin/main' into feature/third-party-data-provider-integrations

.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

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

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

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}},
 	},

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

internal/infrastructure/db/migrations/000289_mailbox_send_claim.down.sql

@@ -0,0 +1 @@
+ALTER TABLE email_accounts DROP COLUMN IF EXISTS send_claimed_at;

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;

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
+}

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")
+	}
+}

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 := `

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")
 		}

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).

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 {

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 {

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)
 	}

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
+}

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', '<earlier@test.local>', $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)
+	}
+}

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)
+		}
+	}
+}

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

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,

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', '<p>Hello</p>', 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)
+	}
+}