From fe328eca7e40fbdfd44bef785e00e420ac82fdeb Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 06:13:06 +0000 Subject: [PATCH] feat: retain missing Cloud report classification when scheduler health batches fall back to a single mailbox read --- .../campaign_pool_batch_live_test.go | 8 ++- internal/scheduler/campaign_sender.go | 6 +- internal/scheduler/cloud_status_gate_test.go | 63 +++++++++++++++++++ 3 files changed, 72 insertions(+), 5 deletions(-) diff --git a/internal/scheduler/campaign_pool_batch_live_test.go b/internal/scheduler/campaign_pool_batch_live_test.go index 730503215..0dd604087 100644 --- a/internal/scheduler/campaign_pool_batch_live_test.go +++ b/internal/scheduler/campaign_pool_batch_live_test.go @@ -52,7 +52,9 @@ func (r *campaignPoolHealthReads) GetHealthState(ctx context.Context, id uuid.UU func (r *campaignPoolHealthReads) GetHealthStates(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]repository.WarmupHealthRead, error) { r.batch++ if r.batchErr != nil { - return nil, r.batchErr + err := r.batchErr + r.batchErr = nil + return nil, err } return r.WarmupRepository.GetHealthStates(ctx, ids) } @@ -111,8 +113,8 @@ func TestLiveCampaignPoolBatchFailuresKeepSafety(t *testing.T) { if !errors.Is(err, failure) || pair != nil || tasks.single != 0 || health.batch != 0 { t.Fatalf("unknown budget must not send: pair=%v err=%v task singles=%d health batches=%d", pair, err, tasks.single, health.batch) } - } else if err != nil && !errors.Is(err, ErrCampaignDeferred) || pair == nil || health.single != 1 || health.batch != 1 { - t.Fatalf("health must fall back to the individual read: pair=%v err=%v health singles=%d batches=%d", pair, err, health.single, health.batch) + } else if err != nil && !errors.Is(err, ErrCampaignDeferred) || pair == nil || health.single != 0 || health.batch != 2 { + t.Fatalf("health must fall back to a one-mailbox rich read: pair=%v err=%v health singles=%d batches=%d", pair, err, health.single, health.batch) } }) } diff --git a/internal/scheduler/campaign_sender.go b/internal/scheduler/campaign_sender.go index 9f715cd54..cc52bf229 100644 --- a/internal/scheduler/campaign_sender.go +++ b/internal/scheduler/campaign_sender.go @@ -249,8 +249,10 @@ func (s *schedulerService) healthFor(ctx context.Context, p *campaignPass, id uu } h := healthRead{} if s.warmupRepo != nil { - if state, until, err := s.warmupRepo.GetHealthState(ctx, id); err == nil { - h = healthRead{state: state, blockedUntil: until, known: true} + if states, err := s.warmupRepo.GetHealthStates(ctx, []uuid.UUID{id}); err == nil { + if state, ok := states[id]; ok { + h = healthRead{state: state.State, blockedUntil: state.BlockedUntil, known: true, cloudStatus: state.CloudStatusOnly, cloudMissing: state.CloudReportMissing} + } } } p.health[id] = h diff --git a/internal/scheduler/cloud_status_gate_test.go b/internal/scheduler/cloud_status_gate_test.go index f4923478a..ab226b004 100644 --- a/internal/scheduler/cloud_status_gate_test.go +++ b/internal/scheduler/cloud_status_gate_test.go @@ -2,13 +2,76 @@ package scheduler import ( "context" + "errors" "testing" "time" "github.com/google/uuid" "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" ) +type fallbackCloudHealth struct { + repository.WarmupRepository + states map[uuid.UUID]repository.WarmupHealthRead + reads int +} + +func (r *fallbackCloudHealth) GetHealthState(_ context.Context, id uuid.UUID) (models.WarmupHealthState, *time.Time, error) { + h := r.states[id] + return h.State, h.BlockedUntil, nil +} + +func (r *fallbackCloudHealth) GetHealthStates(_ context.Context, ids []uuid.UUID) (map[uuid.UUID]repository.WarmupHealthRead, error) { + r.reads++ + if r.reads == 1 { + return nil, errors.New("health batch unavailable") + } + return map[uuid.UUID]repository.WarmupHealthRead{ids[0]: r.states[ids[0]]}, nil +} + +type fallbackCloudCounts struct{ repository.TaskRepository } + +func (*fallbackCloudCounts) CountCampaignEmailsSentTodayByAccounts(context.Context, []uuid.UUID) (map[uuid.UUID]int, error) { + return map[uuid.UUID]int{}, nil +} + +func TestMissingCloudClassificationSurvivesHealthBatchFailure(t *testing.T) { + for _, realHold := range []bool{false, true} { + t.Run(map[bool]string{false: "missing report", true: "real hold"}[realHold], func(t *testing.T) { + now := time.Now() + ended := now.Add(-time.Minute) + acct := models.Email{ID: uuid.New(), CampaignLimit: 50} + repo := &fallbackCloudHealth{states: map[uuid.UUID]repository.WarmupHealthRead{acct.ID: { + State: models.WarmupHealthBlocked, BlockedUntil: &ended, CloudStatusOnly: !realHold, CloudReportMissing: true, + }}} + s := &schedulerService{warmupRepo: repo, taskRepo: &fallbackCloudCounts{}} + pass := &campaignPass{campaign: &models.Campaign{DailyLimit: 50}, risk: models.OrgRiskTrusted, + health: map[uuid.UUID]healthRead{}, sentToday: map[uuid.UUID]int{}} + if err := s.prefillPool(t.Context(), pass, []models.Email{acct}); err != nil { + t.Fatal(err) + } + if len(pass.health) != 0 { + t.Fatal("failed batch populated the health memo") + } + d := s.planMailbox(t.Context(), pass, acct, 0, 0, nil, now, 10*3600, now.Add(10*time.Hour)) + wantState, wantBand := models.MailboxPlanCloudStatus, "" + if realHold { + wantState, wantBand = models.MailboxPlanHealthHold, string(models.WarmupHealthBlocked) + } + if d.state != wantState || d.healthBand() != wantBand || !d.health.cloudMissing { + t.Fatalf("fallback plan: state=%q band=%q health=%+v", d.state, d.healthBand(), d.health) + } + if gate, remaining := s.gateFor(t.Context(), pass, acct, 10); gate.reason != gateHealth || remaining != 0 { + t.Fatalf("fallback opened send gate: %+v remaining=%d", gate, remaining) + } + if repo.reads != 2 { + t.Fatalf("health reads=%d, want one failed batch and one memoized fallback", repo.reads) + } + }) + } +} + // A hold that is only a missing Warmbly Cloud report stays a health gate, so // nothing is sent, but carries its own mark for the sending plan. func TestHealthGateMarksAMissingCloudReport(t *testing.T) {