From 37d328c86a1682b998630a9a9866e9e1c0b27014 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 01/10] feat: add a partial index on campaign_contact_progress.dispatch_task_id so the daily send counters can look a task's reservation up without scanning progress --- .../000127_campaign_progress_dispatch_task_index.down.sql | 1 + .../000127_campaign_progress_dispatch_task_index.up.sql | 8 ++++++++ 2 files changed, 9 insertions(+) create mode 100644 internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.down.sql create mode 100644 internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.up.sql diff --git a/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.down.sql b/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.down.sql new file mode 100644 index 00000000..8b091c01 --- /dev/null +++ b/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_campaign_progress_dispatch_task; diff --git a/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.up.sql b/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.up.sql new file mode 100644 index 00000000..937591b4 --- /dev/null +++ b/internal/infrastructure/db/migrations/000127_campaign_progress_dispatch_task_index.up.sql @@ -0,0 +1,8 @@ +-- A campaign task counts as a send only when it holds a step's reservation +-- (issue #306): the chain's wake-ups complete without sending and must not +-- spend the mailbox's daily budget. The counters look the reservation up by +-- its task, so that lookup needs an index. Built concurrently, on its own, +-- because progress is a live table and a plain CREATE INDEX would block it. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_campaign_progress_dispatch_task + ON campaign_contact_progress (dispatch_task_id) + WHERE dispatch_task_id IS NOT NULL; From ec9a3147b6aa373a65d40295142fc75552b0e6ae Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 02/10] feat: count a completed campaign task as a mailbox send only when it holds a step reservation or a worker-confirmed Message-ID, so the chain's deferral and pause wake-ups no longer spend the daily budget or reset the min-gap and rotation clocks (issue #306) --- internal/repository/pg_task.go | 60 +++++++++++++++++++++------------- 1 file changed, 38 insertions(+), 22 deletions(-) diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go index e172dde6..2db2816b 100644 --- a/internal/repository/pg_task.go +++ b/internal/repository/pg_task.go @@ -398,15 +398,27 @@ func (r *taskRepository) GetEmailTask(ctx context.Context, taskID uuid.UUID) (*E return emailTask, err } -// CountCampaignEmailsSentToday counts only campaign tasks completed today (excludes warmup) +// taskDispatchedEmail is the WHERE fragment for "this completed task put an +// email on the wire", for every query that counts or times a mailbox's sends +// (alias t). Only campaign tasks need it: a campaign is one self-perpetuating +// task, and its wake-ups complete without sending (a deferral, an auto-pause, +// an action step), so status alone charged each of them to the mailbox's daily +// budget and reset its min-gap clock (issue #306). A campaign send is the task +// holding a step's reservation, or one the worker answered with a Message-ID. +const taskDispatchedEmail = `(t.task_type <> 'campaign' OR t.message_id <> '' OR EXISTS ( + SELECT 1 FROM campaign_contact_progress ccp WHERE ccp.dispatch_task_id = t.id))` + +// CountCampaignEmailsSentToday counts the campaign emails a mailbox dispatched +// today (excludes warmup, and the campaign chain's own wake-ups). func (r *taskRepository) CountCampaignEmailsSentToday(ctx context.Context, accountID uuid.UUID) (int, error) { query := ` SELECT COUNT(*) - FROM tasks - WHERE email_account_id = $1 - AND status = 'completed' - AND task_type = 'campaign' - AND DATE(completed_at) = CURRENT_DATE + FROM tasks t + WHERE t.email_account_id = $1 + AND t.status = 'completed' + AND t.task_type = 'campaign' + AND DATE(t.completed_at) = CURRENT_DATE + AND ` + taskDispatchedEmail + ` ` var count int @@ -467,10 +479,11 @@ func (r *taskRepository) CreateEmailTaskFull(ctx context.Context, task *Task, em func (r *taskRepository) CountEmailsSentToday(ctx context.Context, accountID uuid.UUID) (int, error) { query := ` SELECT COUNT(*) - FROM tasks - WHERE email_account_id = $1 - AND status = 'completed' - AND DATE(completed_at) = CURRENT_DATE + FROM tasks t + WHERE t.email_account_id = $1 + AND t.status = 'completed' + AND DATE(t.completed_at) = CURRENT_DATE + AND ` + taskDispatchedEmail + ` ` var count int @@ -493,13 +506,15 @@ func (r *taskRepository) CountWarmupEmailsSentToday(ctx context.Context, account return count, err } -// GetLastEmailTime gets the last email send time for an account +// GetLastEmailTime gets the last email send time for an account. It is the +// min-gap clock, so it reads real sends only. func (r *taskRepository) GetLastEmailTime(ctx context.Context, accountID uuid.UUID) (*time.Time, error) { query := ` - SELECT MAX(completed_at) - FROM tasks - WHERE email_account_id = $1 - AND status = 'completed' + SELECT MAX(t.completed_at) + FROM tasks t + WHERE t.email_account_id = $1 + AND t.status = 'completed' + AND ` + taskDispatchedEmail + ` ` var lastTime *time.Time @@ -528,13 +543,14 @@ func (r *taskRepository) GetLastSendTimes(ctx context.Context, accountIDs []uuid } query := ` - SELECT email_account_id, MAX(completed_at) - FROM tasks - WHERE email_account_id = ANY($1) - AND status = 'completed' - AND task_type = $2::task_type - AND completed_at IS NOT NULL - GROUP BY email_account_id + 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.task_type = $2::task_type + AND t.completed_at IS NOT NULL + AND ` + taskDispatchedEmail + ` + GROUP BY t.email_account_id ` rows, err := r.db.Query(ctx, query, accountIDs, taskType) From 3204cf41bdca4b36d9da79e1ae9f71d4cb04de09 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 03/10] feat: apply the reserved-or-confirmed send filter to the workspace sent-today count, the advisor mailbox volume windows, and the deliverability dashboard denominator so campaign wake-ups stop inflating sent totals --- internal/repository/pg_advanced_outreach.go | 3 ++- internal/repository/pg_advisor_snapshot.go | 1 + internal/repository/pg_organization.go | 1 + 3 files changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go index 2db57bd4..f73c2414 100644 --- a/internal/repository/pg_advanced_outreach.go +++ b/internal/repository/pg_advanced_outreach.go @@ -664,7 +664,8 @@ func (r *advancedOutreachRepository) GetDeliverabilityDashboard(ctx context.Cont SELECT COUNT(*) FROM tasks t JOIN email_accounts ea ON ea.id = t.email_account_id WHERE ea.organization_id = $1 AND t.task_type = 'campaign' AND t.status = 'completed' - AND t.completed_at >= $2 AND t.completed_at <= $3` + AND t.completed_at >= $2 AND t.completed_at <= $3 + AND ` + taskDispatchedEmail _ = r.db.QueryRow(ctx, sentQuery, organizationID, from, to).Scan(&out.EmailsSent) out.BounceRate = models.Rate(out.BounceCount, out.EmailsSent) out.ComplaintRate = models.Rate(out.ComplaintCount, out.EmailsSent) diff --git a/internal/repository/pg_advisor_snapshot.go b/internal/repository/pg_advisor_snapshot.go index 920d2d7a..8f540fd6 100644 --- a/internal/repository/pg_advisor_snapshot.go +++ b/internal/repository/pg_advisor_snapshot.go @@ -70,6 +70,7 @@ func (r *advisorRepository) loadMailboxes(ctx context.Context, orgID uuid.UUID) WHERE t.email_account_id = ea.id AND t.task_type = 'campaign' AND t.status = 'completed' AND t.completed_at > NOW() - INTERVAL '30 days' + AND ` + taskDispatchedEmail + ` ) sent ON true LEFT JOIN LATERAL ( SELECT diff --git a/internal/repository/pg_organization.go b/internal/repository/pg_organization.go index 28a0c057..e355562a 100644 --- a/internal/repository/pg_organization.go +++ b/internal/repository/pg_organization.go @@ -641,6 +641,7 @@ func (r *organizationRepository) GetEmailsSentTodayCount(ctx context.Context, or AND t.task_type = 'campaign' AND t.status = 'completed' AND t.completed_at >= CURRENT_DATE + AND `+taskDispatchedEmail+` `, orgID).Scan(&count) return count, err } From 620e89d528f86bf9bb5be2aaffff5ae4321278c5 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 04/10] feat: exclude action and wait steps from the workspace daily email limit by joining the step kind, since they stamp sent_at for routing without sending mail --- internal/repository/pg_campaign_progress.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/repository/pg_campaign_progress.go b/internal/repository/pg_campaign_progress.go index 17e45e90..697d5812 100644 --- a/internal/repository/pg_campaign_progress.go +++ b/internal/repository/pg_campaign_progress.go @@ -819,12 +819,15 @@ func (r *campaignProgressRepository) CheckContactHasReplied(ctx context.Context, return hasReplied, err } -// CountEmailsSentTodayByOrganization returns how many campaign emails were sent today by an organization. +// CountEmailsSentTodayByOrganization returns how many campaign emails were sent +// today by an organization. Action and wait steps stamp sent_at too, for +// routing, but send nothing, so only email steps count. func (r *campaignProgressRepository) CountEmailsSentTodayByOrganization(ctx context.Context, organizationID uuid.UUID) (int, error) { query := ` SELECT COUNT(*) FROM campaign_contact_progress ccp JOIN campaigns c ON c.id = ccp.campaign_id + JOIN sequences s ON s.id = ccp.sequence_id AND s.kind = 'email' WHERE c.organization_id = $1 AND ccp.sent_at IS NOT NULL AND DATE(ccp.sent_at) = CURRENT_DATE From 711be1f8c2552e94f998c650c77266fd309dac14 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 05/10] feat: defer a campaign whose mailboxes are all at their daily cap, outside their own hours, resting, or held by warmup health instead of auto-pausing it, replace the unreachable push-to-tomorrow recompute, and log the daily-cap and unavailable-pool decisions once per UTC day --- internal/scheduler/campaign_scheduler.go | 157 ++++++++++++++--------- internal/scheduler/errors.go | 12 +- internal/scheduler/helpers.go | 12 ++ 3 files changed, 118 insertions(+), 63 deletions(-) diff --git a/internal/scheduler/campaign_scheduler.go b/internal/scheduler/campaign_scheduler.go index 38d4144e..c0229063 100644 --- a/internal/scheduler/campaign_scheduler.go +++ b/internal/scheduler/campaign_scheduler.go @@ -2,7 +2,9 @@ package scheduler import ( "context" + "fmt" "math/rand" + "strings" "time" "github.com/google/uuid" @@ -199,6 +201,11 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode s.logCampaignDecision(ctx, campaignID, eventType, message, metadata) } } + logDecisionOnce := func(eventType, message string, metadata map[string]interface{}) { + if !preview { + s.logCampaignDecisionOnce(ctx, campaignID, eventType, message, metadata) + } + } // STEP 3.5: Resolve the recipient ESP/provider for ESP matching. Cheap: // prefer the cached contact.esp_provider, else derive from the domain @@ -368,6 +375,16 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode authGated := 0 lifecycleGated := 0 + // Why the other mailboxes were left out. A reason that clears on its own + // (the budget resets at midnight, the mailbox's hours reopen, a health hold + // expires) makes an empty pool a deferral; only one that never clears makes + // it a pause. reopensAt is the earliest reopening among hours-closed + // mailboxes. + budgetSpent := 0 + hoursClosed := 0 + healthHeld := 0 + var reopensAt time.Time + var candidates []AccountCandidate for _, acct := range accounts { // Sending-domain authentication. This runs before the daily-count query @@ -401,6 +418,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // Skip accounts that have reached their daily limit if remaining <= 0 { + budgetSpent++ continue } @@ -419,11 +437,13 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode switch state { case models.WarmupHealthQuarantined, models.WarmupHealthBlocked: if blockedUntil == nil || blockedUntil.After(time.Now()) { + healthHeld++ continue } case models.WarmupHealthWatch, models.WarmupHealthThrottled: remaining = int(float64(remaining) * adjustmentFor(state).volumeMultiplier) if remaining <= 0 { + budgetSpent++ continue } } @@ -452,6 +472,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // cold cap. remaining = s.behaviorDailyCap(ctx, bhv, remaining, openAt) if remaining <= 0 { + budgetSpent++ continue } } else if acct.Timezone != "" && acct.Timezone != campaign.Timezone { @@ -459,7 +480,12 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode acctLocal := candidateTime.In(acctTZ) acctHour := acctLocal.Hour() if acctHour < 8 || acctHour >= 20 { - continue // outside account's business hours + // Outside the account's business hours. + hoursClosed++ + if open := businessHoursReopen(candidateTime, acctTZ); reopensAt.IsZero() || open.Before(reopensAt) { + reopensAt = open + } + continue } } @@ -512,13 +538,51 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode return time.Time{}, nil, uuid.Nil, ErrDomainAuthFailing } - // Every mailbox is out of cold rotation. Say so rather than letting the - // campaign look stalled for no visible reason; they return on their own. - if len(candidates) == 0 && lifecycleGated == len(accounts) { - logDecision("mailboxes_resting", - "No mailbox is in cold rotation: all are resting or held in reserve", - map[string]interface{}{"resting_mailboxes": lifecycleGated, "pool_size": len(accounts)}) - return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + // An empty pool is a pause only when nothing in it will change on its own. + // A pool with every mailbox at its daily cap used to fall through to the + // pause below, and the campaign had to be restarted by hand the next + // morning (issue #306); it is a deferral, like every other gate that lifts + // by itself. The conditions a deferred chain re-finds every few minutes + // until midnight are logged once a day, or the feed drowns in them. + if len(candidates) == 0 { + switch { + case budgetSpent > 0: + logDecisionOnce("daily_cap_reached", + "Every mailbox has used its daily budget; sending resumes tomorrow", + map[string]interface{}{"capped_mailboxes": budgetSpent, "pool_size": len(accounts)}) + return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + case hoursClosed > 0: + // Routine and short: the mailbox's own 8am-8pm band reopens within + // hours, so it earns no line in the activity log. + return nextScheduleSlot(reopensAt, windows, campaignTZ), nil, accounts[0].ID, ErrCampaignDeferred + case lifecycleGated == len(accounts): + // Every mailbox is out of cold rotation. Say so rather than letting + // the campaign look stalled for no visible reason; they return on + // their own. + logDecisionOnce("mailboxes_resting", + "No mailbox is in cold rotation: all are resting or held in reserve", + map[string]interface{}{"resting_mailboxes": lifecycleGated, "pool_size": len(accounts)}) + return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + case healthHeld > 0 || lifecycleGated > 0: + var why []string + if healthHeld > 0 { + why = append(why, fmt.Sprintf("%d held by warmup health", healthHeld)) + } + if lifecycleGated > 0 { + why = append(why, fmt.Sprintf("%d resting or in reserve", lifecycleGated)) + } + if authGated > 0 { + why = append(why, fmt.Sprintf("%d failing domain authentication", authGated)) + } + logDecisionOnce("mailboxes_unavailable", + "No mailbox can send right now: "+strings.Join(why, ", "), + map[string]interface{}{"health_held": healthHeld, "resting_mailboxes": lifecycleGated, + "auth_gated": authGated, "pool_size": len(accounts)}) + return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + } + // What is left was gated by a sending-behaviour profile with no working + // days, which no amount of waiting fixes. + return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox } // STEP 8.25: Apply ESP matching to the under-budget candidate set. @@ -555,57 +619,12 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // STEP 8.5: Select best account per the campaign's rotation mode. pool is // the set selection actually ran over, kept for the pacing maths below. + // Every candidate has budget left today, so it has weight and the selector + // always picks one; the guard only keeps a nil from being dereferenced. pool := candidates selected := selectAccountByRotationMode(campaign.RotationMode, candidates) if selected == nil { - // ALL accounts at capacity today — push to next day and recompute with - // tomorrow's full (ramp-clamped) capacity. The ramp clamp AND the ESP - // filter MUST be re-applied here, or tomorrow's recompute over-budgets a - // mailbox past its ramp ceiling / picks a cross-provider sender. - candidateTime = candidateTime.Add(24 * time.Hour) - candidateTime = nextScheduleSlot(candidateTime, windows, campaignTZ) - - var tomorrow []AccountCandidate - for i := range candidates { - acct := candidates[i].Account - // ESP-strict: keep only matching mailboxes for tomorrow too. - if campaign.ESPMatchMode == "strict" && recipientProvider != "" && !candidates[i].ProviderMatch { - continue - } - acctLimit := effectiveCap(acct) // same ramp clamp as STEP 8 - c := candidates[i] - c.RemainingToday = acctLimit - c.Weight = computeWeight(acctLimit, candidates[i].WarmupAgeDays) - tomorrow = append(tomorrow, c) - } - // ESP-prefer: restrict tomorrow to matching mailboxes when any exist. - if campaign.ESPMatchMode == "prefer" && recipientProvider != "" { - var matchingTomorrow []AccountCandidate - for _, c := range tomorrow { - if c.ProviderMatch { - matchingTomorrow = append(matchingTomorrow, c) - } - } - if len(matchingTomorrow) > 0 { - tomorrow = matchingTomorrow - } - } - - pool = tomorrow - selected = selectAccountByRotationMode(campaign.RotationMode, tomorrow) - if selected == nil { - // ESP-strict with no matching mailbox at all: defer rather than - // complete or send cross-provider. - if campaign.ESPMatchMode == "strict" && recipientProvider != "" { - logDecision("provider_match_deferred", - "No same-provider mailbox available tomorrow; deferring", - map[string]interface{}{"recipient_provider": recipientProvider}) - // Deferral, not a send (see above). - return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred - } - // The pool was not empty, every mailbox in it was gated out. - return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox - } + return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox } account := &selected.Account @@ -740,8 +759,9 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode } // deferToNextDay pushes a candidate time to the next valid campaign day within -// the campaign's send window. Used by the ESP-strict and new-lead-cap deferral -// paths so a campaign reschedules instead of completing or busy-looping. +// the campaign's send window. Used by the ESP-strict, new-lead-cap and +// daily-cap deferral paths so a campaign reschedules instead of completing, +// pausing or busy-looping. func (s *schedulerService) deferToNextDay(campaign *models.Campaign) time.Time { tz := loadLocation(campaign.Timezone) t := nextScheduleSlot(time.Now().Add(24*time.Hour), effectiveWindows(campaign), tz) @@ -763,3 +783,24 @@ func (s *schedulerService) logCampaignDecision(ctx context.Context, campaignID u Metadata: metadata, }) } + +// logCampaignDecisionOnce is logCampaignDecision for a condition the deferred +// chain re-finds on every wake-up until the day rolls over: one line per UTC +// day, the day the budgets reset on. Best-effort and nil-safe. +func (s *schedulerService) logCampaignDecisionOnce(ctx context.Context, campaignID uuid.UUID, eventType, message string, metadata map[string]interface{}) { + if s.campaignLogRepo == nil { + return + } + if metadata == nil { + metadata = map[string]interface{}{} + } + dayStart := time.Now().UTC().Truncate(24 * time.Hour) + day := dayStart.Format("2006-01-02") + metadata["day"] = day + _, _ = s.campaignLogRepo.CreateLogOnce(ctx, &repository.CampaignLogEntry{ + CampaignID: campaignID, + EventType: eventType, + Message: message, + Metadata: metadata, + }, "day", day, dayStart) +} diff --git a/internal/scheduler/errors.go b/internal/scheduler/errors.go index 54a9a63e..e9d2f7f6 100644 --- a/internal/scheduler/errors.go +++ b/internal/scheduler/errors.go @@ -25,15 +25,17 @@ var ( ErrNoEmailAccounts = errors.New("no email accounts available for this campaign") // ErrNoEligibleMailbox is the narrower case: the campaign HAS mailboxes, - // but every one was gated out for both today and tomorrow (daily cap - // reached, warmup health, or outside its own sending window). Reporting - // that as ErrNoEmailAccounts sent people looking at their tag configuration - // for a problem that was never there. + // but none can send under its current settings, and waiting will not + // change that (a sending-behaviour profile with no working days). A gate + // that lifts on its own, such as a spent daily budget, a closed sending + // window or a warmup health hold, is a deferral instead, never this. + // Reporting this as ErrNoEmailAccounts sent people looking at their tag + // configuration for a problem that was never there. // // It wraps ErrNoEmailAccounts so existing callers that pause the campaign // on errors.Is(err, ErrNoEmailAccounts) keep behaving exactly as before. ErrNoEligibleMailbox = fmt.Errorf( - "%w: every mailbox is outside its sending window or over its daily budget", ErrNoEmailAccounts) + "%w: no mailbox can send under its current sending settings", ErrNoEmailAccounts) // ErrDomainAuthFailing is the narrower case again: every mailbox in the // campaign's pool was gated by the sending-domain authentication check. diff --git a/internal/scheduler/helpers.go b/internal/scheduler/helpers.go index 7b2b3403..2c2ec00e 100644 --- a/internal/scheduler/helpers.go +++ b/internal/scheduler/helpers.go @@ -132,6 +132,18 @@ func ensureBusinessHours(t time.Time, timezone string) time.Time { return ensureTimeWindow(t, "08:00", "20:00", loc) } +// businessHoursReopen is when the 8am-8pm band next opens for an instant +// that sits outside it: 8am the same local day before the band, 8am the next +// local day after it. +func businessHoursReopen(t time.Time, loc *time.Location) time.Time { + local := t.In(loc) + open := time.Date(local.Year(), local.Month(), local.Day(), 8, 0, 0, 0, loc) + if local.Hour() >= 20 { + open = open.AddDate(0, 0, 1) + } + return open +} + // calculateHoursRemainingUntil calculates hours remaining until a specific end time func calculateHoursRemainingUntil(timezone, endTime string) float64 { loc := loadLocation(timezone) From 6bc6bf43457dcf6eb366c5a3b7952b6a0e7d351e Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 06/10] feat: reword the no-eligible-mailbox pause on the start endpoint, the task auto-pause, and its test to name the only cause left (a sending behaviour profile with no working days) now that budget and window gates defer --- internal/app/campaign/handlers.go | 4 ++-- internal/tasks/auto_pause_reason_test.go | 2 +- internal/tasks/campaign_task.go | 8 ++++++-- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/internal/app/campaign/handlers.go b/internal/app/campaign/handlers.go index 0f062ad8..e0956d74 100644 --- a/internal/app/campaign/handlers.go +++ b/internal/app/campaign/handlers.go @@ -604,8 +604,8 @@ func (s *campaignService) enqueueCampaignWakeup(ctx context.Context, campaignID case errors.Is(err, scheduler.ErrNoEligibleMailbox): _ = s.campaignRepository.UpdateStatusWithLock(ctx, campaignID, "paused_no_accounts") return errx.New(errx.BadRequest, - "this campaign's mailboxes are all outside their sending window or over their daily limit right now; "+ - "check each mailbox's timezone, sending behaviour and daily cap") + "no mailbox on this campaign can send under its current sending settings; "+ + "check each mailbox's sending behaviour profile (working days) and timezone") case errors.Is(err, scheduler.ErrNoEmailAccounts): _ = s.campaignRepository.UpdateStatusWithLock(ctx, campaignID, "paused_no_accounts") return errx.New(errx.BadRequest, "no active email accounts found for campaign's email tags") diff --git a/internal/tasks/auto_pause_reason_test.go b/internal/tasks/auto_pause_reason_test.go index 5d78e32a..6e63a461 100644 --- a/internal/tasks/auto_pause_reason_test.go +++ b/internal/tasks/auto_pause_reason_test.go @@ -25,7 +25,7 @@ func TestAutoPauseReason(t *testing.T) { { name: "no eligible mailbox", err: scheduler.ErrNoEligibleMailbox, - want: "Campaign auto-paused: every mailbox is outside its sending window or over its daily budget", + want: "Campaign auto-paused: no mailbox can send under its current sending settings (check each mailbox's sending behaviour profile and timezone)", }, { name: "generic no accounts", diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index e2fe204d..9da8deb0 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -204,9 +204,13 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error { } if errors.Is(err, scheduler.ErrCampaignDeferred) { // A valid contact exists but no eligible mailbox right now (ESP-strict - // has no same-provider mailbox, or the daily new-lead cap is reached). + // has no same-provider mailbox, the daily new-lead cap is reached, or + // every mailbox has spent its daily budget or is outside its hours). // Reschedule at the deferred slot WITHOUT sending and WITHOUT touching // progress / daily counters / rotation — mirrors the daily-limit path. + // This task completes without a send, and completing it must not + // spend the mailbox's budget either (issue #306): the budget counts + // reserved sends, never bare wake-ups. // Capped: the next-due moment can be days out, and until this chain // wakes nothing re-reads the campaign, so leads imported meanwhile // would sit queued until then. @@ -908,7 +912,7 @@ func autoPauseReason(err error) string { case errors.Is(err, scheduler.ErrDomainAuthFailing): return "Campaign auto-paused: every mailbox is sending from a domain that fails SPF or DMARC authentication" case errors.Is(err, scheduler.ErrNoEligibleMailbox): - return "Campaign auto-paused: every mailbox is outside its sending window or over its daily budget" + return "Campaign auto-paused: no mailbox can send under its current sending settings (check each mailbox's sending behaviour profile and timezone)" default: return "Campaign auto-paused: no active email accounts available" } From 1c88c2196b0227952de84c82ee3a9884350b64db Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 07/10] feat: add live tests proving wake-ups leave the daily budget and min-gap untouched, real sends still count, and a capped campaign parks a wake-up with a single daily log line instead of pausing --- internal/scheduler/daily_budget_live_test.go | 208 ++++++++++++++++++ .../tasks/campaign_daily_budget_live_test.go | 133 +++++++++++ 2 files changed, 341 insertions(+) create mode 100644 internal/scheduler/daily_budget_live_test.go create mode 100644 internal/tasks/campaign_daily_budget_live_test.go diff --git a/internal/scheduler/daily_budget_live_test.go b/internal/scheduler/daily_budget_live_test.go new file mode 100644 index 00000000..50788645 --- /dev/null +++ b/internal/scheduler/daily_budget_live_test.go @@ -0,0 +1,208 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/warmbly/warmbly/internal/pkg/encrypt" + "github.com/warmbly/warmbly/internal/repository" +) + +// Live checks for issue #306: the per-mailbox daily budget counted every +// completed campaign task as a sent email, so the chain's own wake-ups (a +// deferral, a pause) spent the budget and reset the min-gap clock, and a pool +// with every mailbox at its cap paused the campaign instead of waiting for +// tomorrow. Same harness and env var as live_integration_test.go. + +// completeCampaignTask writes one campaign task the mailbox completed today. +// With a step it is a send: the step's reservation points at the task, the way +// ReserveSend leaves it. Without one it is a wake-up that sent nothing. +func (f *liveFixture) completeCampaignTask(t *testing.T, contact, step *uuid.UUID, completedAt time.Time) uuid.UUID { + t.Helper() + ctx := context.Background() + taskID := uuid.New() + if _, err := f.pool.Exec(ctx, ` + INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at, completed_at, created_at, updated_at) + VALUES ($1, 'campaign', $2, 'completed', '', $3, $3, $3, $3)`, taskID, f.mailbox, completedAt); err != nil { + t.Fatalf("complete task: %v", err) + } + if _, err := f.pool.Exec(ctx, `INSERT INTO campaign_tasks (task_id, campaign_id, contact_id, sequence_id) + VALUES ($1, $2, $3, $4)`, taskID, f.campaign, contact, step); err != nil { + t.Fatalf("link task: %v", err) + } + if contact != nil && step != nil { + if _, err := f.pool.Exec(ctx, `INSERT INTO campaign_contact_progress + (campaign_id, contact_id, sequence_id, sent_at, dispatched_at, dispatch_task_id) + VALUES ($1, $2, $3, $4, $4, $5)`, f.campaign, *contact, *step, completedAt, taskID); err != nil { + t.Fatalf("reserve step: %v", err) + } + } + return taskID +} + +// addSentLead attaches a second lead whose first step this mailbox already +// sent today, so the pool has one real send on the books while the fixture's +// own lead is still waiting to go. +func (f *liveFixture) addSentLead(t *testing.T) { + t.Helper() + ctx := context.Background() + var step uuid.UUID + if err := f.pool.QueryRow(ctx, `SELECT id FROM sequences WHERE campaign_id = $1 ORDER BY position LIMIT 1`, f.campaign).Scan(&step); err != nil { + t.Fatalf("find step: %v", err) + } + contact := uuid.New() + if _, err := f.pool.Exec(ctx, `INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields) + VALUES ($1, $2, $3, $4, 'Sent', 'Lead', '', '', '{}')`, + contact, f.user, f.org, "sent-"+contact.String()[:8]+"@test.local"); err != nil { + t.Fatalf("add contact: %v", err) + } + if _, err := f.pool.Exec(ctx, `INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, 1)`, + f.campaign, contact); err != nil { + t.Fatalf("link lead: %v", err) + } + f.completeCampaignTask(t, &contact, &step, time.Now().Add(-2*time.Hour)) +} + +// loggedScheduler is liveScheduler with the activity log wired, for the +// checks on what a deferral records. +func loggedScheduler(t *testing.T, f *liveFixture) SchedulerService { + t.Helper() + handle, pool := liveDB(t) + enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatalf("encrypter: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM campaign_logs WHERE campaign_id = $1`, f.campaign) + }) + return NewSchedulerService( + repository.NewTaskRepository(pool), + repository.NewWarmupRepository(pool), + repository.NewCampaignProgressRepository(pool), + repository.NewEmailRepostory(handle, enc), + repository.NewCampaignRepostory(handle), + repository.NewContactRepostory(handle), + repository.NewCampaignLogRepository(handle), + ) +} + +// TestLiveWakeupsDoNotSpendTheDailyBudget: a mailbox whose chain woke up fifty +// times today without sending has its whole budget left, and no min-gap to +// wait out. +func TestLiveWakeupsDoNotSpendTheDailyBudget(t *testing.T) { + handle, pool := liveDB(t) + f := newLiveFixture(t, pool, "UTC") + ctx := context.Background() + + // Fifty completed wake-ups, the last one seconds ago: exactly the cap, and + // inside the 600s min-gap. + for i := 0; i < 50; i++ { + f.completeCampaignTask(t, nil, nil, time.Now().Add(-time.Duration(i)*time.Second)) + } + + taskRepo := repository.NewTaskRepository(pool) + sent, err := taskRepo.CountCampaignEmailsSentToday(ctx, f.mailbox) + if err != nil { + t.Fatalf("count: %v", err) + } + if sent != 0 { + t.Fatalf("CountCampaignEmailsSentToday = %d for a mailbox that sent nothing (issue #306)", sent) + } + last, err := taskRepo.GetLastEmailTime(ctx, f.mailbox) + if err != nil { + t.Fatalf("last email time: %v", err) + } + if last != nil { + t.Fatalf("GetLastEmailTime = %s for a mailbox that sent nothing; a wake-up reset the min-gap clock", last) + } + + at, pair, accountID, err := liveScheduler(t, handle, pool).CalculateNextCampaignTime(ctx, f.campaign) + if err != nil { + t.Fatalf("the campaign should be sendable, got %v", err) + } + if pair == nil || accountID != f.mailbox { + t.Fatalf("no sendable pair from the fixture mailbox: pair=%v account=%s", pair, accountID) + } + assertFuture(t, at) +} + +// TestLiveRealSendsStillSpendTheDailyBudget is the other half: the counter has +// to keep seeing the sends it exists for, both the reserved one and the one +// only the worker's Message-ID vouches for. +func TestLiveRealSendsStillSpendTheDailyBudget(t *testing.T) { + _, pool := liveDB(t) + f := newLiveFixture(t, pool, "UTC") + ctx := context.Background() + + f.addSentLead(t) + // A send whose reservation is gone but whose task carries the worker's + // Message-ID (a step walked back and re-sent, or one that predates the + // reservation). + confirmed := f.completeCampaignTask(t, nil, nil, time.Now().Add(-time.Hour)) + if _, err := pool.Exec(ctx, `UPDATE tasks SET message_id = '' WHERE id = $1`, confirmed); err != nil { + t.Fatal(err) + } + // And one wake-up, which must not count. + f.completeCampaignTask(t, nil, nil, time.Now().Add(-time.Minute)) + + taskRepo := repository.NewTaskRepository(pool) + sent, err := taskRepo.CountCampaignEmailsSentToday(ctx, f.mailbox) + if err != nil { + t.Fatalf("count: %v", err) + } + if sent != 2 { + t.Fatalf("CountCampaignEmailsSentToday = %d, want 2 (the reserved send and the confirmed one)", sent) + } + last, err := taskRepo.GetLastEmailTime(ctx, f.mailbox) + if err != nil { + t.Fatalf("last email time: %v", err) + } + if last == nil || time.Since(*last) < 50*time.Minute || time.Since(*last) > 70*time.Minute { + t.Fatalf("GetLastEmailTime = %v, want the confirmed send an hour ago, not the wake-up a minute ago", last) + } +} + +// TestLiveDailyCapDefersInsteadOfPausing: with every mailbox at its cap the +// campaign waits for tomorrow, says so once, and is never paused. +func TestLiveDailyCapDefersInsteadOfPausing(t *testing.T) { + _, pool := liveDB(t) + f := newLiveFixture(t, pool, "UTC") + ctx := context.Background() + + if _, err := pool.Exec(ctx, `UPDATE campaigns SET daily_limit = 1 WHERE id = $1`, f.campaign); err != nil { + t.Fatal(err) + } + f.addSentLead(t) + + s := loggedScheduler(t, f) + at, pair, _, err := s.CalculateNextCampaignTime(ctx, f.campaign) + if errors.Is(err, ErrNoEmailAccounts) { + t.Fatalf("a pool at its daily cap paused the campaign: %v", err) + } + if !errors.Is(err, ErrCampaignDeferred) { + t.Fatalf("want ErrCampaignDeferred, got err=%v pair=%v", err, pair) + } + if pair != nil { + t.Fatal("a deferral must never hand back a sendable pair") + } + if !at.After(time.Now().Add(23 * time.Hour)) { + t.Fatalf("deferred to %s, want tomorrow", at) + } + + // A second pass on the same day logs nothing new. + if _, _, _, err := s.CalculateNextCampaignTime(ctx, f.campaign); !errors.Is(err, ErrCampaignDeferred) { + t.Fatalf("second pass: %v", err) + } + var logged int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_logs WHERE campaign_id = $1 AND event_type = 'daily_cap_reached'`, + f.campaign).Scan(&logged); err != nil { + t.Fatal(err) + } + if logged != 1 { + t.Fatalf("daily_cap_reached logged %d times over two passes, want once per day", logged) + } +} diff --git a/internal/tasks/campaign_daily_budget_live_test.go b/internal/tasks/campaign_daily_budget_live_test.go new file mode 100644 index 00000000..feef821b --- /dev/null +++ b/internal/tasks/campaign_daily_budget_live_test.go @@ -0,0 +1,133 @@ +package tasks + +import ( + "context" + "testing" +) + +// End-to-end checks for issue #306 over the real campaign tick: the chain's +// own wake-ups (a deferral, a pause) must not spend the mailbox's daily +// budget, and a campaign whose mailboxes are all at their cap waits for +// tomorrow instead of pausing. Same harness and env var as +// campaign_send_live_test.go. + +// setCaps sets the per-mailbox cap on both the mailbox and the campaign, so +// the effective cap is exactly n. +func (f *sendFixture) setCaps(t *testing.T, n int) { + t.Helper() + ctx := context.Background() + if _, err := f.pool.Exec(ctx, `UPDATE email_accounts SET campaign_limit = $2 WHERE id = $1`, f.mailbox, n); err != nil { + t.Fatalf("set mailbox cap: %v", err) + } + if _, err := f.pool.Exec(ctx, `UPDATE campaigns SET daily_limit = $2 WHERE id = $1`, f.campaign, n); err != nil { + t.Fatalf("set campaign cap: %v", err) + } +} + +func (f *sendFixture) setMinGap(t *testing.T, seconds int) { + t.Helper() + if _, err := f.pool.Exec(context.Background(), `UPDATE email_accounts SET min_wait_time = $2 WHERE id = $1`, f.mailbox, seconds); err != nil { + t.Fatalf("set min gap: %v", err) + } +} + +func (f *sendFixture) campaignStatus(t *testing.T) string { + t.Helper() + var status string + if err := f.pool.QueryRow(context.Background(), `SELECT status FROM campaigns WHERE id = $1`, f.campaign).Scan(&status); err != nil { + t.Fatalf("campaign status: %v", err) + } + return status +} + +func (f *sendFixture) countLogs(t *testing.T, eventType string) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), `SELECT COUNT(*) FROM campaign_logs WHERE campaign_id = $1 AND event_type = $2`, + f.campaign, eventType).Scan(&n); err != nil { + t.Fatalf("count logs: %v", err) + } + return n +} + +// TestLiveDeferredTicksDoNotSpendTheDailyBudget: one send, then three ticks +// that defer on the mailbox's min-gap, then the second lead must still go out. +// Before the fix the three deferrals were three sends against a cap of two, +// and the fourth tick paused the campaign with a lead never emailed. +func TestLiveDeferredTicksDoNotSpendTheDailyBudget(t *testing.T) { + f := newSendFixture(t) + f.setCaps(t, 2) + + f.tick(t) + if f.sender.count() != 1 { + t.Fatalf("first tick dispatched %d sends, want 1", f.sender.count()) + } + + // A 600s gap after that send: every tick inside it defers without sending. + f.setMinGap(t, 600) + for i := 0; i < 3; i++ { + f.tick(t) + } + if f.sender.count() != 1 { + t.Fatalf("deferred ticks dispatched sends: total %d, want 1", f.sender.count()) + } + if status := f.campaignStatus(t); status != "active" { + t.Fatalf("campaign is %q after three deferrals, want active", status) + } + sent, err := f.svc.taskRepo.CountCampaignEmailsSentToday(context.Background(), f.mailbox) + if err != nil { + t.Fatal(err) + } + if sent != 1 { + t.Fatalf("the mailbox is charged %d sends today, want 1 (the deferrals were counted, issue #306)", sent) + } + + // Gap lifted: the second lead is still within budget and goes out. + f.setMinGap(t, 0) + f.tick(t) + if f.sender.count() != 2 { + t.Fatalf("the second lead was not sent after the deferrals (total %d); campaign is %q", f.sender.count(), f.campaignStatus(t)) + } + if row := f.progressFor(t, f.leadB); row == nil || row.sentAt == nil { + t.Fatalf("lead B was not served: %+v", row) + } +} + +// TestLiveCampaignAtDailyCapWaitsForTomorrow: with the cap spent the campaign +// stays active with a parked wake-up, and says why once. +func TestLiveCampaignAtDailyCapWaitsForTomorrow(t *testing.T) { + f := newSendFixture(t) + f.setCaps(t, 1) + ctx := context.Background() + + f.tick(t) + if f.sender.count() != 1 { + t.Fatalf("first tick dispatched %d sends, want 1", f.sender.count()) + } + + for i := 0; i < 2; i++ { + f.tick(t) + if status := f.campaignStatus(t); status != "active" { + t.Fatalf("tick %d at the daily cap left the campaign %q, want active (it used to be paused_no_accounts)", i+2, status) + } + } + if f.sender.count() != 1 { + t.Fatalf("ticks at the cap dispatched sends: total %d, want 1", f.sender.count()) + } + + // The chain is parked, not dropped. + var pending int + if err := f.pool.QueryRow(ctx, `SELECT COUNT(*) FROM tasks t JOIN campaign_tasks ct ON ct.task_id = t.id + WHERE ct.campaign_id = $1 AND t.status = 'pending'`, f.campaign).Scan(&pending); err != nil { + t.Fatal(err) + } + if pending != 1 { + t.Fatalf("%d pending wake-ups after the cap was reached, want 1", pending) + } + if n := f.countLogs(t, "daily_cap_reached"); n != 1 { + t.Fatalf("daily_cap_reached logged %d times over two ticks, want once per day", n) + } + if n := f.countLogs(t, "auto_paused"); n != 0 { + t.Fatalf("the campaign was auto-paused %d time(s) at its daily cap", n) + } +} From c298d898a7027227d66b43552324ced6ab710fb3 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:18:43 -0700 Subject: [PATCH 08/10] feat: document in the campaigns guide what counts against a mailbox's daily budget and that a capped campaign waits for the next day rather than pausing --- docs/content/docs/guides/campaigns.mdx | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index 4d134812..59890f57 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -63,6 +63,8 @@ Warmbly is mailbox-first: safe volume is the sum of each mailbox's budget, not o The campaign daily limit applies as a minimum against the cold cap, so it can lower a mailbox's volume but never raise it above the mailbox's own daily cap (default `50`/day). +Only an email actually handed to a sending worker counts against a mailbox's daily budget and its minimum gap. The scheduler's own wake-ups, a step it deferred because its slot was not due yet, and action or wait steps never do, so a campaign started outside its sending window still has its whole budget when the window opens. When every mailbox on a campaign has used its budget for the day, the campaign stays active and waits for the next day; the activity log notes it once per day. It also waits, rather than pausing, when every mailbox is outside its own hours, resting, or held back by warmup health. + Anything above `50`/day per cold mailbox needs positive reputation signals and a low complaint rate behind it. Adding mailboxes is safer than forcing a few to send more. @@ -153,7 +155,7 @@ Optional **campaign dates** bound when it may send. Leave both blank to run open The play and pause buttons work from the list row or the detail view. Starting moves the campaign to **active** and begins scheduling inside your windows and limits; pausing stops new scheduling immediately and resumes from where it left off. -A campaign can pause itself: **paused, no accounts** when it loses every sender, **paused, trial expired** when a trial ends, **auto-paused** when a guardrail trips, **needs verification** when address verification has refused every remaining lead (the campaign offers to re-verify them or send anyway; see [address verification](/guides/deliverability/#address-verification)), and plain **paused** if it ever loses its workspace, because unsubscribes, bounces and complaints are checked per workspace and cannot be honoured without one. It moves to **finished** once every contact completes the sequence or its end date passes. Configured to do so, it also stops following up with a contact the moment they reply. +A campaign can pause itself: **paused, no accounts** when it loses every sender or no sender can send under its settings (a sending behaviour profile with no working days), **paused, trial expired** when a trial ends, **auto-paused** when a guardrail trips, **needs verification** when address verification has refused every remaining lead (the campaign offers to re-verify them or send anyway; see [address verification](/guides/deliverability/#address-verification)), and plain **paused** if it ever loses its workspace, because unsubscribes, bounces and complaints are checked per workspace and cannot be honoured without one. It moves to **finished** once every contact completes the sequence or its end date passes. Configured to do so, it also stops following up with a contact the moment they reply. A finished campaign can be started again: after extending or clearing its end date, or adding new leads, pressing play resumes it. If there is genuinely nothing left to send it finishes again immediately with a message saying so. From 4bc971eecbd2d75e35950ff2b554128ecc6eeab4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:26:21 -0700 Subject: [PATCH 09/10] feat: resume a campaign whose pool mixes capped and hours-closed mailboxes at the earlier of tomorrow and the closed mailbox's reopening, log the daily cap only when every usable mailbox is capped, and add the mixed-pool live regression --- internal/scheduler/campaign_scheduler.go | 30 ++++++--- internal/scheduler/daily_budget_live_test.go | 70 ++++++++++++++++++++ 2 files changed, 91 insertions(+), 9 deletions(-) diff --git a/internal/scheduler/campaign_scheduler.go b/internal/scheduler/campaign_scheduler.go index c0229063..ed3c44c9 100644 --- a/internal/scheduler/campaign_scheduler.go +++ b/internal/scheduler/campaign_scheduler.go @@ -546,15 +546,27 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // until midnight are logged once a day, or the feed drowns in them. if len(candidates) == 0 { switch { - case budgetSpent > 0: - logDecisionOnce("daily_cap_reached", - "Every mailbox has used its daily budget; sending resumes tomorrow", - map[string]interface{}{"capped_mailboxes": budgetSpent, "pool_size": len(accounts)}) - return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred - case hoursClosed > 0: - // Routine and short: the mailbox's own 8am-8pm band reopens within - // hours, so it earns no line in the activity log. - return nextScheduleSlot(reopensAt, windows, campaignTZ), nil, accounts[0].ID, ErrCampaignDeferred + case budgetSpent > 0 || hoursClosed > 0: + // Resume when the first of them can send again: tomorrow for a + // spent budget, the reopening of the mailbox's own 8am-8pm band + // otherwise. A closed band is routine and short, so it earns no + // line in the activity log; a pool whose every usable mailbox is + // capped does. + var resume time.Time + if budgetSpent > 0 { + resume = s.deferToNextDay(campaign) + } + if hoursClosed > 0 { + if open := nextScheduleSlot(reopensAt, windows, campaignTZ); resume.IsZero() || open.Before(resume) { + resume = open + } + } + if hoursClosed == 0 { + logDecisionOnce("daily_cap_reached", + "Every available mailbox has used its daily budget; sending resumes tomorrow", + map[string]interface{}{"capped_mailboxes": budgetSpent, "pool_size": len(accounts)}) + } + return resume, nil, accounts[0].ID, ErrCampaignDeferred case lifecycleGated == len(accounts): // Every mailbox is out of cold rotation. Say so rather than letting // the campaign look stalled for no visible reason; they return on diff --git a/internal/scheduler/daily_budget_live_test.go b/internal/scheduler/daily_budget_live_test.go index 50788645..19e67e76 100644 --- a/internal/scheduler/daily_budget_live_test.go +++ b/internal/scheduler/daily_budget_live_test.go @@ -166,6 +166,76 @@ func TestLiveRealSendsStillSpendTheDailyBudget(t *testing.T) { } } +// addClosedHoursMailbox attaches a second active mailbox whose own 8am-8pm +// band is closed right now, and returns its timezone. Picked from a spread of +// offsets, so one is always closed at any hour of the day. +func (f *liveFixture) addClosedHoursMailbox(t *testing.T) *time.Location { + t.Helper() + ctx := context.Background() + var loc *time.Location + for _, name := range []string{"Pacific/Honolulu", "America/Los_Angeles", "America/New_York", "Europe/London", + "Europe/Berlin", "Asia/Dubai", "Asia/Tokyo", "Pacific/Auckland"} { + l, err := time.LoadLocation(name) + if err != nil { + continue + } + if h := time.Now().In(l).Hour(); h < 8 || h >= 20 { + loc = l + break + } + } + if loc == nil { + t.Fatal("no timezone in the spread is outside 8am-8pm right now") + } + mailbox := uuid.New() + if _, err := f.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, 'Closed', '', '', 'smtp_imap', 'active', 50, 600, $5)`, + mailbox, f.user, f.org, "closed-"+mailbox.String()[:8]+"@test.local", loc.String()); err != nil { + t.Fatalf("add mailbox: %v", err) + } + t.Cleanup(func() { + c := context.Background() + _, _ = f.pool.Exec(c, `DELETE FROM campaign_tasks WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, mailbox) + _, _ = f.pool.Exec(c, `DELETE FROM tasks WHERE email_account_id = $1`, mailbox) + _, _ = f.pool.Exec(c, `DELETE FROM email_accounts WHERE id = $1`, mailbox) + }) + return loc +} + +// TestLiveMixedPoolResumesAtTheEarlierMailbox: one mailbox at its cap and one +// merely outside its own hours resume when the second reopens, not tomorrow. +func TestLiveMixedPoolResumesAtTheEarlierMailbox(t *testing.T) { + _, pool := liveDB(t) + f := newLiveFixture(t, pool, "UTC") + ctx := context.Background() + + if _, err := pool.Exec(ctx, `UPDATE campaigns SET daily_limit = 1 WHERE id = $1`, f.campaign); err != nil { + t.Fatal(err) + } + f.addSentLead(t) + loc := f.addClosedHoursMailbox(t) + + s := loggedScheduler(t, f) + at, pair, _, err := s.CalculateNextCampaignTime(ctx, f.campaign) + if !errors.Is(err, ErrCampaignDeferred) || pair != nil { + t.Fatalf("want a deferral, got err=%v pair=%v", err, pair) + } + reopen := businessHoursReopen(time.Now(), loc) + if at.After(reopen.Add(time.Minute)) { + t.Fatalf("deferred to %s, but the second mailbox reopens at %s", at, reopen) + } + assertFuture(t, at) + var logged int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_logs WHERE campaign_id = $1 AND event_type = 'daily_cap_reached'`, + f.campaign).Scan(&logged); err != nil { + t.Fatal(err) + } + if logged != 0 { + t.Fatalf("daily_cap_reached logged for a pool that still has a mailbox coming back today") + } +} + // TestLiveDailyCapDefersInsteadOfPausing: with every mailbox at its cap the // campaign waits for tomorrow, says so once, and is never paused. func TestLiveDailyCapDefersInsteadOfPausing(t *testing.T) { From 5c9d36516358f681c5d894d98e66d209d183f8c5 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 4 Sep 2026 02:59:00 -0700 Subject: [PATCH 10/10] feat: pick a closed-hours mailbox that stays closed for the next ten minutes so the scheduling pass cannot run after it opens, and assert the mixed-pool deferral lands on the reopening from both sides instead of accepting any earlier time --- internal/scheduler/daily_budget_live_test.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/internal/scheduler/daily_budget_live_test.go b/internal/scheduler/daily_budget_live_test.go index 19e67e76..2974a055 100644 --- a/internal/scheduler/daily_budget_live_test.go +++ b/internal/scheduler/daily_budget_live_test.go @@ -168,10 +168,13 @@ func TestLiveRealSendsStillSpendTheDailyBudget(t *testing.T) { // addClosedHoursMailbox attaches a second active mailbox whose own 8am-8pm // band is closed right now, and returns its timezone. Picked from a spread of -// offsets, so one is always closed at any hour of the day. +// offsets no wider than the closed band, so one always qualifies whatever the +// hour. It must still be closed when the scheduler runs a moment later, or a +// mailbox picked at 07:59 would be open by then and the pass would send. func (f *liveFixture) addClosedHoursMailbox(t *testing.T) *time.Location { t.Helper() ctx := context.Background() + closed := func(at time.Time) bool { return at.Hour() < 8 || at.Hour() >= 20 } var loc *time.Location for _, name := range []string{"Pacific/Honolulu", "America/Los_Angeles", "America/New_York", "Europe/London", "Europe/Berlin", "Asia/Dubai", "Asia/Tokyo", "Pacific/Auckland"} { @@ -179,13 +182,14 @@ func (f *liveFixture) addClosedHoursMailbox(t *testing.T) *time.Location { if err != nil { continue } - if h := time.Now().In(l).Hour(); h < 8 || h >= 20 { + now := time.Now().In(l) + if closed(now) && closed(now.Add(10*time.Minute)) { loc = l break } } if loc == nil { - t.Fatal("no timezone in the spread is outside 8am-8pm right now") + t.Fatal("no timezone in the spread stays outside 8am-8pm for the next ten minutes") } mailbox := uuid.New() if _, err := f.pool.Exec(ctx, `INSERT INTO email_accounts (id, user_id, organization_id, email, name, @@ -221,8 +225,10 @@ func TestLiveMixedPoolResumesAtTheEarlierMailbox(t *testing.T) { if !errors.Is(err, ErrCampaignDeferred) || pair != nil { t.Fatalf("want a deferral, got err=%v pair=%v", err, pair) } + // Exactly the reopening, within a minute: later means the capped mailbox's + // "tomorrow" won, earlier means the pass never really deferred. reopen := businessHoursReopen(time.Now(), loc) - if at.After(reopen.Add(time.Minute)) { + if at.Before(reopen.Add(-time.Minute)) || at.After(reopen.Add(time.Minute)) { t.Fatalf("deferred to %s, but the second mailbox reopens at %s", at, reopen) } assertFuture(t, at)