Hello
', 0, 0, 'email')`, + f.step, f.campaign, f.org) + for i := 0; i < leads; i++ { + id := uuid.New() + f.leads = append(f.leads, id) + exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, + company, phone, custom_fields, verification_status, created_at) + VALUES ($1, $2, $3, $4, 'Routed', 'Lead', '', '', '{}', 'valid', NOW() + make_interval(secs => $5))`, + id, f.owner, f.org, "routed-lead-"+id.String()[:8]+"@test.local", float64(i)) + exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, $3)`, + f.campaign, id, i) + } + + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM campaign_contact_progress WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaign_leads WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM sequences WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaigns WHERE id = $1`, f.campaign}, + {`DELETE FROM email_accounts WHERE id = $1`, f.mailbox}, + {`DELETE FROM contacts WHERE organization_id = $1`, f.org}, + {`DELETE FROM organizations WHERE id = $1`, f.org}, + {`DELETE FROM users WHERE id = $1`, f.owner}, + } { + if _, err := pool.Exec(c, step.sql, step.arg); err != nil { + t.Errorf("cleanup %q: %v", step.sql, err) + } + } + }) + return f +} + +func (f *routedPairsFixture) find(t *testing.T, paced PacedSenders, limit int) ([]ContactSequencePair, *time.Time, bool) { + t.Helper() + pairs, nextDue, onSender, err := NewCampaignProgressRepository(f.pool).FindRoutedPairs( + context.Background(), f.campaign, "created_at", "asc", "", false, false, paced, limit) + if err != nil { + t.Fatalf("find routed pairs: %v", err) + } + return pairs, nextDue, onSender +} + +// The batch is the first `limit` DUE leads in routing order, and a batch that +// found somebody reports no next-due: the caller is sending, not waiting. +func TestLiveFindRoutedPairsReturnsDueLeadsInOrderUpToTheLimit(t *testing.T) { + _, pool := liveContactDB(t) + f := newRoutedPairsFixture(t, pool, 5) + + pairs, nextDue, onSender := f.find(t, nil, 3) + if len(pairs) != 3 { + t.Fatalf("got %d pairs, want the limit of 3", len(pairs)) + } + for i := range pairs { + if pairs[i].ContactID != f.leads[i] { + t.Fatalf("pair %d is contact %s, want %s: the batch must keep routing order", + i, pairs[i].ContactID, f.leads[i]) + } + if pairs[i].SequenceID != f.step || !pairs[i].IsNewLead { + t.Fatalf("pair %d = %+v, want the entry step for a new lead", i, pairs[i]) + } + } + if nextDue != nil || onSender { + t.Fatalf("a batch that found leads reported next_due=%v waiting_on_sender=%v", nextDue, onSender) + } +} + +// A limit below one is still a request for work, not for nothing. +func TestLiveFindRoutedPairsLimitBelowOneReturnsOne(t *testing.T) { + _, pool := liveContactDB(t) + f := newRoutedPairsFixture(t, pool, 3) + + for _, limit := range []int{0, -1} { + pairs, _, _ := f.find(t, nil, limit) + if len(pairs) != 1 { + t.Fatalf("limit %d returned %d pairs, want 1", limit, len(pairs)) + } + } +} + +// A limit above the number of due leads returns all of them, and stopping the +// scan early must not lose the ones behind it. +func TestLiveFindRoutedPairsReturnsEveryDueLeadBelowTheLimit(t *testing.T) { + _, pool := liveContactDB(t) + f := newRoutedPairsFixture(t, pool, 4) + + pairs, _, _ := f.find(t, nil, 25) + if len(pairs) != 4 { + t.Fatalf("got %d pairs, want all 4 due leads", len(pairs)) + } +} + +// Nothing due: no pairs, and the soonest moment something becomes due, so the +// scheduler defers until then instead of calling the campaign complete. +func TestLiveFindRoutedPairsReportsNextDueWhenNothingIsDue(t *testing.T) { + _, pool := liveContactDB(t) + f := newRoutedPairsFixture(t, pool, 3) + // An entry delay holds every first email back for two hours. + if _, err := pool.Exec(context.Background(), + `UPDATE campaigns SET entry_delay_minutes = 120 WHERE id = $1`, f.campaign); err != nil { + t.Fatalf("set entry delay: %v", err) + } + + pairs, nextDue, onSender := f.find(t, nil, 25) + if len(pairs) != 0 { + t.Fatalf("got %d pairs while every lead was inside the entry delay", len(pairs)) + } + if nextDue == nil { + t.Fatal("nothing due and no next-due time: the scheduler would complete the campaign") + } + if until := time.Until(*nextDue); until < 110*time.Minute || until > 121*time.Minute { + t.Fatalf("next due in %s, want about the two-hour entry delay", until.Round(time.Minute)) + } + if onSender { + t.Fatal("an entry delay is not a lead waiting for its mailbox") + } +} + +// A lead bound to a mailbox that is merely paced waits for it and is reported +// through the next-due time; the leads behind it are still offered, so one busy +// mailbox never parks the campaign. +func TestLiveFindRoutedPairsSkipsALeadWaitingOnItsPacedMailbox(t *testing.T) { + _, pool := liveContactDB(t) + f := newRoutedPairsFixture(t, pool, 2) + ctx := context.Background() + + // The first lead has heard from the mailbox already, so it is bound to it. + if _, err := pool.Exec(ctx, + `UPDATE campaign_leads SET email_account_id = $3, sender_assigned_at = NOW() + WHERE campaign_id = $1 AND contact_id = $2`, f.campaign, f.leads[0], f.mailbox); err != nil { + t.Fatalf("bind lead: %v", err) + } + back := time.Now().Add(3 * time.Hour).Truncate(time.Second) + + pairs, nextDue, onSender := f.find(t, PacedSenders{f.mailbox: back}, 25) + if len(pairs) != 1 { + t.Fatalf("got %d pairs, want only the unbound lead behind the waiting one", len(pairs)) + } + if pairs[0].ContactID != f.leads[1] { + t.Fatalf("offered contact %s, want the unbound lead %s", pairs[0].ContactID, f.leads[1]) + } + // A due pair is what the caller acts on, so the wait is not reported here. + if nextDue != nil || onSender { + t.Fatalf("reported next_due=%v waiting_on_sender=%v alongside a sendable pair", nextDue, onSender) + } + + // With the free lead gone, the wait is the whole answer. + if _, err := pool.Exec(ctx, `DELETE FROM campaign_leads WHERE campaign_id = $1 AND contact_id = $2`, + f.campaign, f.leads[1]); err != nil { + t.Fatalf("drop the free lead: %v", err) + } + pairs, nextDue, onSender = f.find(t, PacedSenders{f.mailbox: back}, 25) + if len(pairs) != 0 { + t.Fatalf("got %d pairs while the only lead was waiting for its own mailbox", len(pairs)) + } + if nextDue == nil || !nextDue.Equal(back) { + t.Fatalf("next due = %v, want the mailbox's reopening %s", nextDue, back) + } + if !onSender { + t.Fatal("the wait must be reported as a lead waiting for its mailbox, not as a step's delay") + } +} diff --git a/internal/repository/pg_campaign_progress.go b/internal/repository/pg_campaign_progress.go index 9fea48fd..c5dad0f0 100644 --- a/internal/repository/pg_campaign_progress.go +++ b/internal/repository/pg_campaign_progress.go @@ -211,28 +211,35 @@ type CampaignProgressRepository interface { CountEmailsSentTodayByOrganization(ctx context.Context, organizationID uuid.UUID) (int, error) GetLatestCampaignSequenceForContact(ctx context.Context, contactID uuid.UUID) (*CampaignSequencePair, error) - // FindNextRoutedPair selects the next (contact, step) to send by following - // each contact's step rules (the branching tree) rather than a flat position - // order. prioritizeNewLeads sorts first-step pairs first; excludeNewLeads - // drops first-step pairs entirely so the new-lead/day cap can be enforced - // while follow-ups keep flowing. The second return value, when the pair is - // nil, is the soonest time a waiting contact's condition window elapses — the + // FindRoutedPairs selects the (contact, step) pairs that are due now by + // following each contact's step rules (the branching tree) rather than a + // flat position order, in routing order, up to limit of them. + // prioritizeNewLeads sorts first-step pairs first; excludeNewLeads drops + // first-step pairs entirely so the new-lead/day cap can be enforced while + // follow-ups keep flowing. The second return value, when NO pair is due, is + // the soonest time a waiting contact's condition window elapses — the // scheduler should defer and re-check then rather than completing. // paced names the pool mailboxes that cannot take a send right now but will // be able to on their own, mapped to when; a lead already bound to one is // skipped for this pass and reported through the next-due time instead of // parking every lead behind it. Nil applies no sender gate. // + // More than one pair is returned because placement can refuse a SINGLE + // lead (ESP-strict finds no same-provider mailbox, the lead's own mailbox + // is busy, the recipient's preferred hours are hours away) and the leads + // behind them are still sendable. One pair and one refusal used to park the + // whole campaign. + // // waitingOnSender reports that the returned next-due moment is a lead // waiting for its own mailbox rather than for a step's wait or a condition // window, so the caller can say which. - FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders) (pair *ContactSequencePair, nextDue *time.Time, waitingOnSender bool, err error) + FindRoutedPairs(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders, limit int) (pairs []ContactSequencePair, nextDue *time.Time, waitingOnSender bool, err error) // RouteContact runs the same routing for ONE contact and reports where // their flow goes next, plus the pre-send gate that excludes them, so a // per-contact preview reads the facts the send path reads. RouteContact(ctx context.Context, campaignID, contactID uuid.UUID) (*ContactRoute, error) - // CountUndeliverableLeads counts the leads FindNextRoutedPair excludes + // CountUndeliverableLeads counts the leads FindRoutedPairs excludes // because address verification refused them. Reported when a campaign // finishes, so "completed" never silently means "skipped everybody". CountUndeliverableLeads(ctx context.Context, campaignID uuid.UUID) (int, error) @@ -968,7 +975,7 @@ func (r *campaignProgressRepository) GetLatestCampaignSequenceForContact(ctx con // undeliverableClause is the ONE definition of "the campaign will never send to // this lead": address verification marked it invalid, or marked it risky while // the campaign's "send to risky emails" toggle is off. Routing excludes these -// (see FindNextRoutedPair), the campaign task's pre-send gates refuse them, and +// (see FindRoutedPairs), the campaign task's pre-send gates refuse them, and // the Leads view reports them as undeliverable, so all three read the same rule // rather than three copies that can drift apart. // @@ -982,7 +989,7 @@ func undeliverableClause(cp string) string { ) } -// FindNextRoutedPair selects the next (contact, step) to send by FOLLOWING THE +// FindRoutedPairs selects the (contact, step) pairs to send by FOLLOWING THE // FLOW graph. For each contact, the next step is the route out of their // last-sent step: // 1. conditional branches (first match wins, evaluated against engagement), @@ -1010,10 +1017,12 @@ func undeliverableClause(cp string) string { // whose next step isn't decidable yet (an engagement window still open) is not // returned. // -// Only a pair that is DUE is returned: a new lead is due once the campaign's +// Only pairs that are DUE are returned: a new lead is due once the campaign's // entry delay has elapsed since they entered it (immediately when there is // none); a routed step is due wait_after days after the contact's last step -// (plus a wait node's minutes). The first due contact in list order wins. +// (plus a wait node's minutes). The first `limit` due contacts in list order +// win, in order, so the caller can move on to the next one when placement +// refuses the one in front. // Contacts whose next step is due later never block the ones behind them: without this, the one lead // routed to a "wait 3 days" follow-up parks the whole campaign for 3 days while // every other lead's first email sits queued. When nothing is due, the second @@ -1028,8 +1037,11 @@ func undeliverableClause(cp string) string { // that is NOT coming back on its own is deliberately absent from `paced`: the // lead is offered, and the placer moves it to another mailbox. // -// Returns a nil pair and a nil next-due when the campaign is genuinely complete. -func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders) (*ContactSequencePair, *time.Time, bool, error) { +// Returns no pairs and a nil next-due when the campaign is genuinely complete. +func (r *campaignProgressRepository) FindRoutedPairs(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders, limit int) ([]ContactSequencePair, *time.Time, bool, error) { + if limit < 1 { + limit = 1 + } router, err := r.loadRouter(ctx, campaignID) if err != nil { return nil, nil, false, err @@ -1125,6 +1137,7 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam // wait elapses or a condition window that closes. Only reported when no // contact is due right now. var nextDue *time.Time + pairs := make([]ContactSequencePair, 0, limit) // waitingOnSender is true while the soonest moment belongs to a lead // waiting for its own mailbox, so the caller can say "waiting for its // mailbox" rather than "waiting for a step's delay". @@ -1169,11 +1182,19 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam noteDue(back, true) continue } - return &ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, NotBefore: res.DueAt, AssignedSender: in.sender}, nil, false, nil + pairs = append(pairs, ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, NotBefore: res.DueAt, AssignedSender: in.sender}) + if len(pairs) >= limit { + break + } } if rerr := rows.Err(); rerr != nil { return nil, nil, false, rerr } + // A due pair makes the deferral times collected so far irrelevant: the + // caller is sending, not waiting. + if len(pairs) > 0 { + return pairs, nil, false, nil + } // Nobody sendable now. Hand back the soonest moment somebody will be so the // scheduler defers until then rather than completing. return nil, nextDue, waitingOnSender, nil @@ -1205,7 +1226,7 @@ type ContactRoute struct { // RouteContact runs the campaign's routing for ONE contact and reports where // their flow goes next, including the gate that would exclude them. It reads -// exactly what FindNextRoutedPair reads, so a preview never disagrees with +// exactly what FindRoutedPairs reads, so a preview never disagrees with // the send path. func (r *campaignProgressRepository) RouteContact(ctx context.Context, campaignID, contactID uuid.UUID) (*ContactRoute, error) { router, err := r.loadRouter(ctx, campaignID) diff --git a/internal/scheduler/behavior.go b/internal/scheduler/behavior.go index 1a8673d7..5ebb7c3f 100644 --- a/internal/scheduler/behavior.go +++ b/internal/scheduler/behavior.go @@ -103,11 +103,24 @@ func (s *schedulerService) placeWithinBehavior(ctx context.Context, r behavior.R // arithmetic sequence; the configured min_wait_time is the fallback for every // mailbox that has not opted in. func (s *schedulerService) behaviorGap(r behavior.Resolved, at time.Time, fallbackSeconds int) int { + return s.behaviorGapWith(r, at, fallbackSeconds, rand.Float64) +} + +// behaviorGapFloor is behaviorGap without the draw: the shortest gap the +// profile allows. A read-only preview uses it so the same unchanged state +// answers with the same time twice in a row — a "not before" that moved on +// every refresh is what made the contact drawer look like it was guessing +// (issue #437). +func (s *schedulerService) behaviorGapFloor(r behavior.Resolved, at time.Time, fallbackSeconds int) int { + return s.behaviorGapWith(r, at, fallbackSeconds, func() float64 { return 0 }) +} + +func (s *schedulerService) behaviorGapWith(r behavior.Resolved, at time.Time, fallbackSeconds int, rnd func() float64) int { if !r.Enabled { return fallbackSeconds } plan := r.PlanOn(behavior.PlanDateFor(at, r.Loc)) - gap := behavior.DrawGap(plan, rand.Float64) + gap := behavior.DrawGap(plan, rnd) secs := int(gap / time.Second) if secs < 1 { return fallbackSeconds diff --git a/internal/scheduler/campaign_scheduler.go b/internal/scheduler/campaign_scheduler.go index baf5870a..cd7c3b17 100644 --- a/internal/scheduler/campaign_scheduler.go +++ b/internal/scheduler/campaign_scheduler.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "errors" "fmt" "math/rand" "strings" @@ -72,7 +73,10 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai excludeNewLeads = true } } - nextPair, recheckAt, senderWait, err := s.campaignProgressRepo.FindNextRoutedPair( + // The due leads this pass may try, in routing order. More than one, because + // placement can refuse a lead for a reason that is that lead's alone while + // the lead behind them is sendable this second (issue #437). + candidates, recheckAt, senderWait, err := s.campaignProgressRepo.FindRoutedPairs( ctx, campaignID, campaign.ContactOrderBy, @@ -81,29 +85,30 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai campaign.PrioritizeNewLeads, excludeNewLeads, paced, + config.CampaignPlacementCandidates, ) if err != nil { return time.Time{}, nil, uuid.Nil, err } - if nextPair == nil { + if len(candidates) == 0 { // When the new-lead cap is active and only new-lead pairs remain, - // FindNextRoutedPair returns nil with exclude on but WOULD return a pair - // without it. In that case defer to the next day so follow-ups keep + // FindRoutedPairs returns nothing with exclude on but WOULD return a + // pair without it. In that case defer to the next day so follow-ups keep // progressing and new leads resume tomorrow — do NOT complete. if excludeNewLeads { - again, againDue, againSenderWait, aerr := s.campaignProgressRepo.FindNextRoutedPair( + again, againDue, againSenderWait, aerr := s.campaignProgressRepo.FindRoutedPairs( ctx, campaignID, campaign.ContactOrderBy, campaign.ContactOrderDir, orderField, - campaign.PrioritizeNewLeads, false, paced, + campaign.PrioritizeNewLeads, false, paced, 1, ) switch { case aerr != nil: // Fall through to the ordinary wait/complete decision below. - case again != nil: + case len(again) > 0: s.logCampaignDecision(ctx, campaignID, "new_lead_cap_reached", "Daily new-lead cap reached; deferring remaining new leads to tomorrow", map[string]interface{}{"max_new_leads_per_day": campaign.MaxNewLeadsPerDay}) - deferTime := s.deferToNextDay(campaign) + deferTime := s.deferToNextDay(campaign, false) // A follow-up that comes due before tomorrow must not wait for // the new-lead cap to reset. if recheckAt != nil && recheckAt.Before(deferTime) { @@ -149,14 +154,37 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai return time.Time{}, nil, uuid.Nil, ErrCampaignCompleted } - // Branch routing is resolved inside FindNextRoutedPair: the chosen step is the + // Branch routing is resolved inside FindRoutedPairs: the chosen step is the // route out of the contact's last-sent step — conditional branches first // (first match wins, evaluated against opened/clicked/replied), then the // explicit "else" catch-all, then linear position+1 only when a step defines // no branches. A step is sent only if the flow reaches it; STOP/end and // already-sent loops drop the contact in the finder. Conditions are evaluated // at schedule time (a known, accepted race vs. last-moment engagement). - return s.placeCampaignSend(ctx, campaign, accounts, senderMetaByID, nextPair, pass, false) + // + // Place the due leads in order and send the first one the pool can take. A + // refusal that is about THIS lead (ErrLeadDeferred: ESP-strict has no + // mailbox for their provider, their own mailbox is busy, their preferred + // hours are hours away) moves to the lead behind them; anything else is the + // pool's answer for every lead and ends the pass immediately. Only when + // every candidate is refused does the campaign defer, at the soonest of + // their slots. + var leadSlot time.Time + leadAccount := accounts[0].ID + for i := range candidates { + at, sendable, accountID, perr := s.placeCampaignSend(ctx, campaign, accounts, senderMetaByID, &candidates[i], pass, false) + if !errors.Is(perr, ErrLeadDeferred) { + return at, sendable, accountID, perr + } + if leadSlot.IsZero() || (!at.IsZero() && at.Before(leadSlot)) { + leadSlot, leadAccount = at, accountID + } + } + // Every due lead was refused for its own reason. Re-check on the deferral + // horizon rather than completing: the leads are still there, and what + // refuses them (a mailbox under budget again, a recipient's morning) comes + // back from outside this chain. + return leadSlot, nil, leadAccount, ErrCampaignDeferred } // humanizeMinutes renders a delay as the largest whole unit it divides into @@ -313,7 +341,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode if pass.risk.BlocksSending() { logDecision("org_suspended", "Sending is paused for this workspace while it is under review", nil) - return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + return s.deferToNextDay(campaign, preview), nil, accounts[0].ID, ErrCampaignDeferred } // providerMatches reports whether a mailbox's provider satisfies the @@ -506,7 +534,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // capped does. var resume time.Time if budgetSpent > 0 { - resume = s.deferToNextDay(campaign) + resume = s.deferToNextDay(campaign, preview) } if hoursClosed > 0 { if open := nextScheduleSlot(reopensAt, windows, campaignTZ); resume.IsZero() || open.Before(resume) { @@ -526,7 +554,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode 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 + return s.deferToNextDay(campaign, preview), nil, accounts[0].ID, ErrCampaignDeferred case healthHeld > 0 || lifecycleGated > 0: var why []string if healthHeld > 0 { @@ -542,7 +570,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode "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 + return s.deferToNextDay(campaign, preview), 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. @@ -574,14 +602,17 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode switch campaign.ESPMatchMode { case "strict": if len(matching) == 0 { - // No matching mailbox under budget today: defer to the next slot - // rather than complete or send cross-provider. - logDecision("provider_match_deferred", - "No same-provider mailbox available; deferring to next slot", + // No matching mailbox under budget today: leave THIS lead for + // later rather than complete or send cross-provider. It is the + // recipient's own domain that has no mailbox, so the lead behind + // them may still be sendable; logged once a day, because the + // pass now re-finds this for every refused lead. + logDecisionOnce("provider_match_deferred", + "No same-provider mailbox available; those leads wait for one", map[string]interface{}{"recipient_provider": recipientProvider}) // Deferral, not a send: nil pair + sentinel so the caller reschedules // instead of sending this contact from a cross-provider mailbox. - return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred + return s.deferToNextDay(campaign, preview), nil, accounts[0].ID, ErrLeadDeferred } candidates = matching case "prefer": @@ -613,12 +644,21 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode map[string]interface{}{"mailbox": bound.Email, "reason": gate.reason}) resume := gate.reopensAt if resume.IsZero() { - resume = s.deferToNextDay(campaign) + resume = s.deferToNextDay(campaign, preview) } return resume, nil, bound.ID, ErrSenderBusy } } - selected = selectAccountByRotationMode(campaign.RotationMode, candidates) + if preview { + // Rotation is a draw, and a read that draws answers a different + // mailbox — and so a different min-gap — every time it is called. + // A preview picks the same one every time instead; which mailbox + // rotation will really hand this lead is not knowable in advance + // anyway, and the drawer is reporting a time, not a sender. + selected = stableCandidate(candidates) + } else { + selected = selectAccountByRotationMode(campaign.RotationMode, candidates) + } } if selected == nil { return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox @@ -628,14 +668,30 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // STEP 8.75: Send-to-send spacing for THIS send. With a behaviour profile // the gap is drawn fresh from the mailbox's range, so the intervals between - // its sends are irregular; otherwise it is the mailbox's fixed min gap. - gapSeconds := s.behaviorGap(selected.Behavior, candidateTime, account.MinWaitTime) + // its sends are irregular; otherwise it is the mailbox's fixed min gap. A + // preview takes the shortest gap the profile allows instead of a draw, so + // it answers the same question the same way twice. + var gapSeconds int + if preview { + gapSeconds = s.behaviorGapFloor(selected.Behavior, candidateTime, account.MinWaitTime) + } else { + gapSeconds = s.behaviorGap(selected.Behavior, candidateTime, account.MinWaitTime) + } + + // leadFloor records that the hard floor below belongs to THIS lead and not + // to the pool, so a refusal moves the pass to the next lead instead of + // parking the campaign (issue #437). + leadFloor := false // Move the candidate onto the mailbox's own workday before the spacing // maths below runs, so distribution is computed against the window the send // will actually land in. if selected.OpenAt != nil && selected.OpenAt.After(candidateTime) { candidateTime = *selected.OpenAt + // Waiting for a mailbox to reopen is this lead's own wait only when the + // lead is BOUND to it, for the reason spelled out at the min-gap below: + // an unbound lead is one rotation places, and rotation gets another go. + leadFloor = bound != nil && bound.ID == account.ID } // hardFloor is the earliest moment this send is ALLOWED: wait_after, @@ -644,6 +700,14 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // (distribution, jitter, curve) shapes the slot but never gates a send. hardFloor := candidateTime + // A preview reports that FLOOR and stops there. Steps 9 to 13 below are + // spacing: drawn fresh from a random source and measured from time.Now(), + // so running them for a read made two reads of one unchanged campaign + // answer minutes apart, and the drawer showed a time that walked forward on + // every refresh while the step sat there marked Due (issue #437). Every + // real constraint — the mailbox min-gap, the recipient's hours, the sending + // windows — still runs. + // STEP 9: Even distribution across the candidate day's sending window. With // a behaviour profile that is the mailbox's own rolled workday (lunch // excluded); otherwise it is the span of the campaign's intervals for that @@ -656,8 +720,7 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // per-mailbox guard still binds afterwards — each mailbox's own daily cap // and hourly ceiling gated it into this set, and the min-gap plus conflict // resolution below space its own sends. - remainingEmails := poolRemainingOn(pool, candidateTime) - if remainingEmails > 0 { + if remainingEmails := poolRemainingOn(pool, candidateTime); !preview && remainingEmails > 0 { remainingMinutes, ok := remainingSendMinutes(selected, candidateTime, windows, campaignTZ) if ok && remainingMinutes > 0 { // Vary the pace multiplicatively (bursts and lulls) — evenly @@ -692,6 +755,21 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode } if hardFloor.Before(earliestNext) { hardFloor = nextScheduleSlot(earliestNext, windows, campaignTZ) + // A bound lead has one address and must wait out ITS mailbox's gap, + // which is this lead's wait and nobody else's: the lead behind it, + // on another mailbox, can still go now. + // + // An unbound lead keeps the pool-wide answer. Selection does not + // look at the min-gap, so rotation can hand an unbound lead a + // mailbox that has just sent — but round_robin and + // least_recently_used both pick the least-used mailbox, which is + // the one that has NOT just sent, and weighted re-draws on the next + // tick. Skipping the lead would spend the whole candidate budget + // re-deriving one shared gap on a single-mailbox campaign, which is + // most of them. + if bound != nil && bound.ID == account.ID { + leadFloor = true + } } } @@ -703,25 +781,27 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // the slot, capped at the original 20 minutes, keeps the irregularity // without erasing the pacing. Deliberately NOT rounded to a 5-minute grid — // a fleet that only ever sends at :x0/:x5 marks is a detectable pattern. - if spread := min(int(time.Until(candidateTime).Minutes())/2, 20); spread > 0 { - candidateTime = candidateTime.Add(time.Minute * time.Duration(randomJitter(-spread, spread))) + if !preview { + if spread := min(int(time.Until(candidateTime).Minutes())/2, 20); spread > 0 { + candidateTime = candidateTime.Add(time.Minute * time.Duration(randomJitter(-spread, spread))) + } } candidateTime = notBefore(candidateTime) // STEP 12: Check conflicts with other scheduled tasks - dateToCheck := candidateTime - scheduledTasks, err := s.taskRepo.GetScheduledTasksForAccount(ctx, account.ID, dateToCheck) - if err != nil { - return time.Time{}, nil, uuid.Nil, err + if !preview { + scheduledTasks, terr := s.taskRepo.GetScheduledTasksForAccount(ctx, account.ID, candidateTime) + if terr != nil { + return time.Time{}, nil, uuid.Nil, terr + } + candidateTime = resolveConflicts(candidateTime, scheduledTasks, gapSeconds) } - candidateTime = resolveConflicts(candidateTime, scheduledTasks, gapSeconds) - // STEP 13: Apply human-like distribution (favor morning/afternoon peaks). // Skipped for behaviour-profiled mailboxes: the profile already describes // this mailbox's workday and its own lunch break, and layering the generic // curve on top would drag sends away from the hours the customer chose. - if !selected.Behavior.Enabled { + if !preview && !selected.Behavior.Enabled { candidateTime = applyDistributionCurve(candidateTime, campaignTZ) } @@ -733,6 +813,9 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode recipientContact, sendPref, campaign.EndDate); ok && snapped.After(candidateTime) { candidateTime = snapped hardFloor = snapped + // One recipient's morning is nobody else's: the lead behind this one + // may be awake right now. + leadFloor = true } } @@ -748,20 +831,66 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode // one. Report it deferred instead: the caller reschedules at the computed // slot without sending. A task that fired at its own slot always passes. if time.Until(hardFloor) > config.CampaignNotDueGraceSeconds*time.Second { - return finalSlot(candidateTime), nil, account.ID, ErrCampaignDeferred + if leadFloor { + return scheduledSlot(candidateTime, preview), nil, account.ID, ErrLeadDeferred + } + return scheduledSlot(candidateTime, preview), nil, account.ID, ErrCampaignDeferred } // STEP 15: Randomise the sub-minute component so sends never land on :00. - return finalSlot(candidateTime), nextPair, account.ID, nil + return scheduledSlot(candidateTime, preview), nextPair, account.ID, nil +} + +// stableCandidate is the preview's stand-in for rotation: the strongest +// candidate, ties broken by mailbox id so the same pool always answers with the +// same mailbox. +func stableCandidate(candidates []AccountCandidate) *AccountCandidate { + var best *AccountCandidate + for i := range candidates { + c := &candidates[i] + if best == nil || c.Weight > best.Weight || + (c.Weight == best.Weight && c.Account.ID.String() < best.Account.ID.String()) { + best = c + } + } + return best +} + +// scheduledSlot is finalSlot for a real scheduled_at and a bare future clamp +// for a preview: the sub-minute randomisation exists so the fleet does not send +// at second :00, and a read-only answer that moved by up to a minute between +// two refreshes was reporting that jitter as if it were news. +func scheduledSlot(t time.Time, preview bool) time.Time { + if preview { + return notBefore(t) + } + return finalSlot(t) } // deferToNextDay pushes a candidate time to the next valid campaign day within // 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 { +// +// A preview gets the FLOOR of the same answer instead: the first open minute of +// the campaign's next sending day. Two things made the scheduler's own value +// wrong to show in a drawer — it is measured from the instant it is asked +// ("24 hours from now", so 3pm today means 3pm tomorrow) and it then adds up to +// half an hour of jitter so a fleet of deferred chains does not all wake +// together. Neither means anything to a reader, and between them they moved the +// drawer's "not before" on every refresh for the commonest waiting reason +// there is: every mailbox having spent its daily budget (issue #437). The +// floor is the honest answer anyway, because what the step is waiting for is +// the day rolling over, not the wake-up the chain happens to have picked. +func (s *schedulerService) deferToNextDay(campaign *models.Campaign, preview bool) time.Time { tz := loadLocation(campaign.Timezone) - t := nextScheduleSlot(time.Now().Add(24*time.Hour), effectiveWindows(campaign), tz) + windows := effectiveWindows(campaign) + if preview { + local := time.Now().In(tz) + midnight := time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, tz).AddDate(0, 0, 1) + return nextScheduleSlot(midnight, windows, tz) + } + t := nextScheduleSlot(time.Now().Add(24*time.Hour), windows, tz) // Add a small jitter so deferred tasks don't all wake at the same instant. return t.Add(time.Minute * time.Duration(randomJitter(0, 30))) } diff --git a/internal/scheduler/campaign_sender.go b/internal/scheduler/campaign_sender.go index 696f9388..5595808e 100644 --- a/internal/scheduler/campaign_sender.go +++ b/internal/scheduler/campaign_sender.go @@ -274,7 +274,7 @@ func (s *schedulerService) pacedSenders(ctx context.Context, p *campaignPass, ac if gate.paced { back := gate.reopensAt if back.IsZero() { - back = s.deferToNextDay(p.campaign) + back = s.deferToNextDay(p.campaign, false) } paced[acct.ID] = back } diff --git a/internal/scheduler/contact_preview.go b/internal/scheduler/contact_preview.go index 7ff28c06..12bfea07 100644 --- a/internal/scheduler/contact_preview.go +++ b/internal/scheduler/contact_preview.go @@ -33,7 +33,14 @@ const ( ) // ContactSendPreview is a read-only "what happens next" for one contact in -// one campaign; ScheduledAt is set only when the step is due now. +// one campaign. It is a pure read in both senses: it writes nothing, and it +// answers the same unchanged state with the same times on every call, because +// the drawer polls it and a figure that walked forward on every refresh read as +// a product that could not make up its mind (issue #437). +// +// ScheduledAt is set only when the step is due now, and is then the campaign +// chain's own next wakeup. NotBefore is the step's hard floor, never a paced +// slot. type ContactSendPreview struct { Route *repository.ContactRoute State models.ContactNextActionState @@ -96,11 +103,19 @@ func (s *schedulerService) PreviewContactSend(ctx context.Context, campaignID, c switch { case perr == nil && sendable != nil: pv.State = models.NextActionDue - slot := at - pv.ScheduledAt = &slot + // A due step is not waiting for a time of its own; it is waiting for the + // campaign's own chain to wake up and reach it. Report THAT instant, + // which is a stored scheduled_at and therefore the same answer on every + // read, rather than rolling a fresh slot for a send that is already due + // (issue #437). No wakeup means the chain is being re-seeded, and the + // honest answer is no time at all. + pv.ScheduledAt = s.campaignWakeup(ctx, campaignID) return pv, nil case errors.Is(perr, ErrCampaignDeferred): pv.State = models.NextActionWaiting + // `at` is the step's hard floor here, not a paced slot: a preview runs + // placement with the spacing, jitter and distribution layers off, so the + // same unchanged campaign answers with the same instant every time. slot := at if route.DueAt != nil && route.DueAt.After(slot) { slot = *route.DueAt @@ -172,3 +187,24 @@ func projectRampLevel(c *models.Campaign, now time.Time) { c.RampLevel = min(c.RampCeiling, max(c.RampLevel, c.RampStart)+c.RampIncrement) c.RampLevelDate = &today } + +// campaignWakeup is when the campaign's chain next runs: the earliest pending +// campaign task. Nil when the campaign has none, which the reconciler fixes +// within its own interval. +func (s *schedulerService) campaignWakeup(ctx context.Context, campaignID uuid.UUID) *time.Time { + tasks, err := s.campaignRepo.GetPendingCampaignTasks(ctx, campaignID) + if err != nil { + return nil + } + var next *time.Time + for i := range tasks { + at := tasks[i].ScheduledAt + if at == nil { + continue + } + if next == nil || at.Before(*next) { + next = at + } + } + return next +} diff --git a/internal/scheduler/contact_preview_live_test.go b/internal/scheduler/contact_preview_live_test.go index 7808c1cf..9fb128bc 100644 --- a/internal/scheduler/contact_preview_live_test.go +++ b/internal/scheduler/contact_preview_live_test.go @@ -42,12 +42,31 @@ func liveLead(t *testing.T, pool *pgxpool.Pool, campaign uuid.UUID) (step1, cont return step1, contact } +// parkWakeup writes the campaign chain's pending wakeup at `at`, the row a +// real tick leaves behind for its successor. +func parkWakeup(t *testing.T, pool *pgxpool.Pool, f *liveFixture, at time.Time) { + t.Helper() + ctx := context.Background() + id := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at, created_at, updated_at) + VALUES ($1, 'campaign', $2, 'pending', '', $3, NOW(), NOW())`, id, f.mailbox, at); err != nil { + t.Fatalf("park wakeup: %v", err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO campaign_tasks (task_id, campaign_id) VALUES ($1, $2)`, id, f.campaign); err != nil { + t.Fatalf("link wakeup: %v", err) + } +} + // A new lead in an always-open campaign is due now: the preview names the -// entry step and carries the slot the scheduler would give it. +// entry step and reports the campaign chain's own next wakeup as the time it +// will be served. func TestLivePreviewDueLeadGetsASlot(t *testing.T) { handle, pool := liveDB(t) f := newLiveFixture(t, pool, "UTC") step1, contact := liveLead(t, pool, f.campaign) + wakeup := time.Now().Add(7 * time.Minute).Truncate(time.Second) + parkWakeup(t, pool, f, wakeup) pv, err := livePreviewer(t, liveScheduler(t, handle, pool)).PreviewContactSend(context.Background(), f.campaign, contact) if err != nil { @@ -59,8 +78,8 @@ func TestLivePreviewDueLeadGetsASlot(t *testing.T) { if pv.State != models.NextActionDue || pv.ScheduledAt == nil { t.Fatalf("want a due step with a slot, got state=%q scheduled_at=%v constraint=%q", pv.State, pv.ScheduledAt, pv.Constraint) } - if pv.ScheduledAt.Before(time.Now().Add(-time.Minute)) { - t.Fatalf("slot %s is in the past", pv.ScheduledAt) + if !pv.ScheduledAt.Equal(wakeup) { + t.Fatalf("slot %s is not the campaign's parked wakeup %s", pv.ScheduledAt.UTC(), wakeup.UTC()) } } @@ -172,3 +191,169 @@ func TestLivePreviewPausedCampaignHasNoSlot(t *testing.T) { t.Fatalf("want paused with no slot, got state=%q constraint=%q scheduled_at=%v", pv.State, pv.Constraint, pv.ScheduledAt) } } + +// The reported symptom of issue #437: the drawer showed a step marked Due whose +// "next slot" walked forward on every refresh — 2:41 PM, then 2:44, then 2:51. +// The preview was running the whole send path for a read, including the layers +// that are not constraints at all: even distribution measured from time.Now() +// with a random multiplier, ±20 minutes of jitter, conflict resolution, the +// distribution curve, a behaviour gap drawn fresh from the mailbox's range, +// weighted rotation re-rolled per call, a randomised sub-minute component, and +// a next-day deferral jittered by up to half an hour so a fleet of chains does +// not wake together. +// +// So the invariant is one sentence and it covers every state: reading an +// unchanged campaign twice gives the same answer twice. It is asserted over the +// whole state table rather than on the one case that was reported, because each +// state reaches the slot through a different set of those layers, and each +// layer was its own drift. +func TestLivePreviewIsStableAcrossReads(t *testing.T) { + ctx := context.Background() + exec := func(t *testing.T, pool *pgxpool.Pool, sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("setup %q: %v", sql[:min(70, len(sql))], err) + } + } + // sent records a completed campaign send from a mailbox a moment ago, which + // is what starts its minimum gap and spends its daily budget. + sent := func(t *testing.T, pool *pgxpool.Pool, mailbox uuid.UUID) { + t.Helper() + exec(t, pool, `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', 'mid@test.local', NOW(), NOW(), NOW(), NOW())`, + uuid.New(), mailbox) + } + + cases := []struct { + name string + state models.ContactNextActionState + setup func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, contact uuid.UUID) + }{ + {"due, chain parked", models.NextActionDue, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + parkWakeup(t, pool, f, time.Now().Add(6*time.Minute).Truncate(time.Second)) + }}, + {"due, chain re-seeding", models.NextActionDue, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) {}}, + {"entry delay", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET entry_delay_minutes = 180 WHERE id = $1`, f.campaign) + }}, + {"start date", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET start_date = NOW() + interval '2 days' WHERE id = $1`, f.campaign) + }}, + {"sending window", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + h := (time.Now().UTC().Hour() + 4) % 24 + exec(t, pool, `UPDATE campaigns SET start_time = $2, end_time = $3 WHERE id = $1`, + f.campaign, fmt.Sprintf("%02d:00", h), fmt.Sprintf("%02d:30", h)) + }}, + {"mailbox minimum gap", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE email_accounts SET min_wait_time = 5400 WHERE id = $1`, f.mailbox) + sent(t, pool, f.mailbox) + }}, + {"daily budget spent", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE email_accounts SET campaign_limit = 1 WHERE id = $1`, f.mailbox) + sent(t, pool, f.mailbox) + }}, + {"daily new-lead cap", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET max_new_leads_per_day = 1 WHERE id = $1`, f.campaign) + exec(t, pool, `INSERT INTO campaign_daily_sends (campaign_id, send_date, emails_sent, new_leads_started) + VALUES ($1, CURRENT_DATE, 1, 1)`, f.campaign) + exec(t, pool, `UPDATE email_accounts SET campaign_limit = 1 WHERE id = $1`, f.mailbox) + sent(t, pool, f.mailbox) + }}, + {"no same-provider mailbox", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE contacts SET email = $2 WHERE id = $1`, c, "strict-"+c.String()[:8]+"@gmail.com") + exec(t, pool, `UPDATE campaigns SET esp_match_mode = 'strict' WHERE id = $1`, f.campaign) + }}, + // Rotation is a draw. Two mailboxes whose gaps differ by an hour, so a + // read that re-rolled rotation answered an hour apart. + {"a pool, not one mailbox", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET rotation_mode = 'weighted' WHERE id = $1`, f.campaign) + second := uuid.New() + exec(t, pool, `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, 'Live Two', '', '', 'smtp_imap', 'active', 50, 7200, 'UTC')`, + second, f.user, f.org, "live2-"+second.String()[:8]+"@test.local") + t.Cleanup(func() { + c := context.Background() + for _, sql := range []string{ + `DELETE FROM campaign_tasks WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, + `DELETE FROM tasks WHERE email_account_id = $1`, + `DELETE FROM email_account_daily_plan WHERE email_account_id = $1`, + `DELETE FROM email_account_behavior WHERE email_account_id = $1`, + `DELETE FROM email_accounts WHERE id = $1`, + } { + if _, err := pool.Exec(c, sql, second); err != nil { + t.Errorf("cleanup %q: %v", sql, err) + } + } + }) + exec(t, pool, `UPDATE email_accounts SET min_wait_time = 3600 WHERE id = $1`, f.mailbox) + sent(t, pool, f.mailbox) + sent(t, pool, second) + }}, + // A behaviour profile draws the send-to-send gap from a range on every + // pass. The floor has to clear the not-due grace or the step reads as + // due and carries no not-before to compare. + {"behaviour profile gap", models.NextActionWaiting, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + b := liveProfile() + b.GapMinSeconds, b.GapMaxSeconds = 600, 7200 + b.WorkStartMin, b.WorkStartMax = 0, 0 + b.WorkEndMin, b.WorkEndMax = 24*60-1, 24*60-1 + b.LunchEnabled = false + b.Weekdays = models.BehaviorWeekdaysAll + f.setProfile(t, b) + sent(t, pool, f.mailbox) + }}, + {"no mailbox at all", models.NextActionBlocked, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE email_accounts SET status = 'inactive' WHERE id = $1`, f.mailbox) + }}, + {"past its end date", models.NextActionBlocked, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET end_date = NOW() - interval '1 day' WHERE id = $1`, f.campaign) + }}, + {"campaign paused", models.NextActionPaused, func(t *testing.T, pool *pgxpool.Pool, f *liveFixture, c uuid.UUID) { + exec(t, pool, `UPDATE campaigns SET status = 'paused' WHERE id = $1`, f.campaign) + }}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + handle, pool := liveDB(t) + f := newLiveFixture(t, pool, "UTC") + _, contact := liveLead(t, pool, f.campaign) + tc.setup(t, pool, f, contact) + + p := livePreviewer(t, liveScheduler(t, handle, pool)) + type answer struct { + state, constraint string + scheduledAt, notBefore string + } + show := func(x *time.Time) string { + if x == nil { + return "none" + } + return x.UTC().Format(time.RFC3339Nano) + } + var first answer + for i := 0; i < 8; i++ { + pv, err := p.PreviewContactSend(context.Background(), f.campaign, contact) + if err != nil { + t.Fatalf("preview %d: %v", i, err) + } + if pv.State != tc.state { + t.Fatalf("read %d: state = %q, want %q (constraint %q)", i, pv.State, tc.state, pv.Constraint) + } + got := answer{string(pv.State), string(pv.Constraint), show(pv.ScheduledAt), show(pv.NotBefore)} + if i == 0 { + first = got + continue + } + if got != first { + t.Fatalf("read %d changed its answer with nothing about the campaign changed:\n first: %+v\n now: %+v", + i, first, got) + } + } + t.Logf("stable over 8 reads: %+v", first) + }) + } +} diff --git a/internal/scheduler/errors.go b/internal/scheduler/errors.go index 000e0268..27bc2cf8 100644 --- a/internal/scheduler/errors.go +++ b/internal/scheduler/errors.go @@ -52,24 +52,42 @@ var ( // ErrDailyLimitReached is returned when the daily limit has been reached ErrDailyLimitReached = errors.New("daily email limit reached") - // ErrCampaignDeferred is returned when there IS a valid contact to send but - // no eligible mailbox right now — ESP-strict has no same-provider mailbox - // under budget, or the daily new-lead cap is reached. The caller must - // reschedule at the returned (defer) time WITHOUT sending. The returned pair - // is always nil on this path so it can never be mistaken for a sendable - // contact; the returned accountID is a nominal pool mailbox for the wakeup - // task only (the next invocation re-evaluates selection from scratch). + // ErrCampaignDeferred is returned when there ARE valid contacts to send but + // nothing can go right now — every usable mailbox has spent its daily + // budget or is outside its hours, the daily new-lead cap is reached, the + // step's own wait has not elapsed, or every due lead was refused for a + // reason of its own (see ErrLeadDeferred). The caller must reschedule at the + // returned (defer) time WITHOUT sending. The returned pair is always nil on + // this path so it can never be mistaken for a sendable contact; the returned + // accountID is a nominal pool mailbox for the wakeup task only (the next + // invocation re-evaluates selection from scratch). ErrCampaignDeferred = errors.New("campaign send deferred - no eligible mailbox for this contact right now") - // ErrSenderBusy is the narrower deferral: this lead's sequence belongs to - // one mailbox, and that mailbox has nothing left today. Every step of a + // ErrLeadDeferred is the deferral that belongs to ONE lead rather than to + // the campaign: this recipient cannot be placed right now (ESP-strict finds + // no same-provider mailbox for their domain, their own mailbox is busy, + // their preferred hours are hours away), while the pool itself is fine and + // the lead behind them may well be sendable this second. + // + // The scheduler answers it by moving to the next routed lead instead of + // parking the campaign. Before that, one such lead at the head of the + // routing order stopped every other lead in the campaign from ever being + // sent: the tick deferred, woke, routed to the same lead, and deferred + // again, forever (issue #437). + // + // It wraps ErrCampaignDeferred so every caller outside the scheduler — + // which only ever reschedules on a deferral — behaves exactly as before. + ErrLeadDeferred = fmt.Errorf("%w: this lead cannot be placed right now", ErrCampaignDeferred) + + // ErrSenderBusy is the narrower lead deferral: this lead's sequence belongs + // to one mailbox, and that mailbox has nothing left today. Every step of a // conversation comes from the address the contact first heard from, so the // lead waits for it rather than being written to by a stranger. // - // It wraps ErrCampaignDeferred, so every caller that reschedules on a - // deferral behaves exactly as before; only the contact drawer, which words - // the reason, tests for it. - ErrSenderBusy = fmt.Errorf("%w: the mailbox this lead's sequence belongs to has no capacity left today", ErrCampaignDeferred) + // It wraps ErrLeadDeferred (and so ErrCampaignDeferred), so the scheduler + // moves on to the next lead and every other caller reschedules exactly as + // before; only the contact drawer, which words the reason, tests for it. + ErrSenderBusy = fmt.Errorf("%w: the mailbox this lead's sequence belongs to has no capacity left today", ErrLeadDeferred) ) // DeferSlot is the wakeup time a caller must use after CalculateNextCampaignTime diff --git a/internal/tasks/campaign_lead_refusal_live_test.go b/internal/tasks/campaign_lead_refusal_live_test.go new file mode 100644 index 00000000..1cc1f1a0 --- /dev/null +++ b/internal/tasks/campaign_lead_refusal_live_test.go @@ -0,0 +1,144 @@ +package tasks + +import ( + "context" + "testing" + + "github.com/google/uuid" +) + +// Live checks for issue #437: a campaign that reads ACTIVE, whose leads read +// "Due" in the contact drawer, and that never sends anything. +// +// The cause was a scope mistake, not a missing gate. Placement can refuse a +// send for a reason that belongs to ONE lead — ESP-strict finds no mailbox for +// that recipient's provider, that lead's own mailbox is busy, that recipient's +// preferred hours are hours away — and the tick answered by deferring the whole +// campaign. Routing hands back the same lead on the next tick, so the campaign +// deferred, woke, refused the same lead and deferred again, forever, while +// every lead behind it sat queued. +// +// ESP-strict is the cheapest way to make placement refuse exactly one lead +// deterministically, so it is what these drive. + +// espStrictOnLeadA points the campaign at same-provider-only sending and moves +// lead A (the first in routing order) onto a domain the pool cannot match: the +// fixture's only mailbox is smtp_imap, which under strict mode is never a +// Gmail match. Lead B keeps a plain domain, which is a wildcard. +func espStrictOnLeadA(t *testing.T, f *sendFixture) { + t.Helper() + ctx := context.Background() + if _, err := f.pool.Exec(ctx, `UPDATE contacts SET email = $2 WHERE id = $1`, + f.leadA, "blocked-"+f.leadA.String()[:8]+"@gmail.com"); err != nil { + t.Fatalf("move lead A to gmail: %v", err) + } + if _, err := f.pool.Exec(ctx, + `UPDATE campaigns SET esp_match_mode = 'strict' WHERE id = $1`, f.campaign); err != nil { + t.Fatalf("enable ESP-strict: %v", err) + } +} + +// pendingWakeups counts the campaign's live chain. +func (f *sendFixture) pendingWakeups(t *testing.T) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), `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(&n); err != nil { + t.Fatalf("count wakeups: %v", err) + } + return n +} + +// logCount counts one kind of activity line on the campaign. +func (f *sendFixture) logCount(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 %q logs: %v", eventType, err) + } + return n +} + +// The headline regression: one lead placement cannot serve must not stop the +// lead behind it from being sent, on the very first tick. +func TestLiveRefusedLeadDoesNotParkTheCampaign(t *testing.T) { + f := newSendFixture(t) + espStrictOnLeadA(t, f) + + f.tick(t) + + if f.sender.count() != 1 { + t.Fatalf("the tick dispatched %d sends; the lead behind the refused one was never served", f.sender.count()) + } + if row := f.progressFor(t, f.leadB); row == nil || row.sentAt == nil { + t.Fatalf("lead B was not sent: %+v", row) + } + if row := f.progressFor(t, f.leadA); row != nil && (row.sentAt != nil || row.dispatchedAt != nil) { + t.Fatalf("the refused lead was sent anyway from a cross-provider mailbox: %+v", row) + } + if n := f.pendingWakeups(t); n != 1 { + t.Fatalf("the chain has %d pending wakeups, want exactly 1", n) + } +} + +// And the campaign stays alive around the lead it cannot serve: it does not +// complete (the lead is still there), it does not pause, and the reason is +// written to the activity log once rather than once per refused lead per tick. +func TestLiveCampaignSurvivesALeadItCannotServe(t *testing.T) { + f := newSendFixture(t) + espStrictOnLeadA(t, f) + + for i := 0; i < 3; i++ { + f.tick(t) + } + + if f.sender.count() != 1 { + t.Fatalf("%d sends over three ticks, want exactly 1 (lead B once, lead A never)", f.sender.count()) + } + 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("read status: %v", err) + } + if status != "active" { + t.Fatalf("campaign is %q; a lead waiting for a mailbox its provider does not have is not the end of the campaign", status) + } + if n := f.logCount(t, "provider_match_deferred"); n != 1 { + t.Fatalf("the ESP-strict deferral was logged %d times; it must be once a day, not once per refused lead per tick", n) + } + if n := f.logCount(t, "completed"); n != 0 { + t.Fatalf("the campaign logged %d completions while a lead was still waiting", n) + } +} + +// A campaign whose EVERY due lead is refused still parks a successor rather +// than completing or pausing, so it resumes the moment a matching mailbox +// appears. +func TestLiveCampaignWithEveryLeadRefusedDefersInsteadOfCompleting(t *testing.T) { + f := newSendFixture(t) + ctx := context.Background() + espStrictOnLeadA(t, f) + if _, err := f.pool.Exec(ctx, `UPDATE contacts SET email = $2 WHERE id = $1`, + f.leadB, "blocked-"+uuid.New().String()[:8]+"@outlook.com"); err != nil { + t.Fatalf("move lead B to outlook: %v", err) + } + + f.tick(t) + + if f.sender.count() != 0 { + t.Fatalf("%d sends went out while no mailbox matched any recipient", f.sender.count()) + } + if n := f.pendingWakeups(t); n != 1 { + t.Fatalf("the chain has %d pending wakeups, want exactly 1: the campaign must come back and look again", n) + } + var status string + if err := f.pool.QueryRow(ctx, `SELECT status FROM campaigns WHERE id = $1`, f.campaign).Scan(&status); err != nil { + t.Fatalf("read status: %v", err) + } + if status != "active" { + t.Fatalf("campaign is %q, want it still active and waiting", status) + } +} diff --git a/internal/tasks/campaign_lead_sender_live_test.go b/internal/tasks/campaign_lead_sender_live_test.go index 91e30d31..bce83340 100644 --- a/internal/tasks/campaign_lead_sender_live_test.go +++ b/internal/tasks/campaign_lead_sender_live_test.go @@ -325,3 +325,38 @@ func TestLiveABusySenderHoldsItsOwnLeadOnly(t *testing.T) { t.Fatalf("lead 2 sent from the mailbox that has no budget left") } } + +// TestLiveBoundLeadInItsMailboxGapDoesNotHoldTheQueue is the same rule for +// spacing rather than budget (issue #437). A lead bound to a mailbox that sent +// a moment ago has to wait out that mailbox's minimum gap; the leads behind it, +// which another mailbox can take right now, must not wait with it. +func TestLiveBoundLeadInItsMailboxGapDoesNotHoldTheQueue(t *testing.T) { + f := newStickyFixture(t, 2, 2) + + f.tick(t) + first := f.sendsByLead(t)[f.leads[0]] + if len(first) != 1 { + t.Fatalf("first step did not send: %v", first) + } + + // An hour of spacing on every mailbox. Lead 1 is bound to one that has just + // sent, so its follow-up cannot go for an hour; lead 2 has no mailbox yet + // and the other one has never sent. + if _, err := f.pool.Exec(context.Background(), + `UPDATE email_accounts SET min_wait_time = 3600 WHERE organization_id = $1`, f.org); err != nil { + t.Fatalf("widen the gap: %v", err) + } + + f.tick(t) + sends := f.sendsByLead(t) + if got := sends[f.leads[0]]; len(got) != 1 { + t.Fatalf("lead 1 was emailed again from %v inside its own mailbox's minimum gap", got) + } + got := sends[f.leads[1]] + if len(got) != 1 { + t.Fatalf("lead 2 did not send while the other mailbox was free and idle: %v", got) + } + if got[0] == first[0] { + t.Fatalf("lead 2 sent from the mailbox that is inside its minimum gap") + } +} diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index 53c07cd7..345fbbcf 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -441,7 +441,7 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error { log.Warn().Err(err).Str("campaign_id", campaign.ID.String()).Str("task_id", taskID.String()).Msg("Failed to update campaign task tracking") } - // stop_on_reply is enforced inside FindNextRoutedPair (STEP 6), and it is now + // stop_on_reply is enforced inside FindRoutedPairs (STEP 6), and it is now // ROUTE-AWARE: a contact who replied is only handed back when their next step // is part of the reply flow (the reply branch's own path). The normal cold // sequence stops there, so there is no longer a blanket "contact has replied, diff --git a/web/src/components/app/contacts/contact-edit/ActivityTab.tsx b/web/src/components/app/contacts/contact-edit/ActivityTab.tsx index 6b3f4f18..b4d19f17 100644 --- a/web/src/components/app/contacts/contact-edit/ActivityTab.tsx +++ b/web/src/components/app/contacts/contact-edit/ActivityTab.tsx @@ -469,8 +469,8 @@ function NextActionFact({ {stateLabel} {next.state === "due" && next.scheduled_at && ( - - next slot {fmtAbsolute(next.scheduled_at)} + + next pass {fmtAbsolute(next.scheduled_at)} )} {next.state !== "due" && next.not_before && ( diff --git a/web/src/lib/api/models/app/contacts/ContactCampaignState.ts b/web/src/lib/api/models/app/contacts/ContactCampaignState.ts index 848d3a31..5c28a58f 100644 --- a/web/src/lib/api/models/app/contacts/ContactCampaignState.ts +++ b/web/src/lib/api/models/app/contacts/ContactCampaignState.ts @@ -20,7 +20,7 @@ export interface ContactCampaignStep { in_flight?: boolean; } -// due: the step is due and the scheduler produced a slot for it. +// due: the step is due; scheduled_at is when the campaign next works its queue. // waiting: a hard constraint holds it back; not_before is the earliest. // paused: the campaign is not active. // blocked: the campaign cannot send at all right now. @@ -35,7 +35,9 @@ export interface ContactNextAction { subject?: string; state: ContactNextActionState; - // Only when due now; contacts ahead in the queue can still push it later. + // Only when due now, and then it is the campaign chain's own next wakeup, + // not a slot reserved for this contact: leads queued ahead can still push + // this step to a later pass. Absent while the chain is being re-seeded. scheduled_at?: string | null; not_before?: string | null; constraint?: string;