From 4bc971eecbd2d75e35950ff2b554128ecc6eeab4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:26:21 -0700 Subject: [PATCH] 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) {