diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 49596b91d..34d731e0d 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -1316,8 +1316,9 @@ func main() { emailService.WireCloudLink(cloudLinkRepository) cloudLinkService = cloudlink.NewService(cloudLinkRepository, emailRepostory, emailService) cloudLinkService.OnStandingChange(func(ctx context.Context, change models.CloudLinkStandingChange) { - state, _, err := warmupRepository.GetHealthState(ctx, change.EmailAccountID) - if err == nil { + state, cloudUnknown, err := warmupRepository.GetKnownHealthState(ctx, change.EmailAccountID) + // A missing Cloud report is no verdict; only a real hold beside it moves the band. + if err == nil && (!cloudUnknown || state.Holds()) { err = workerRepository.SetEmailAccountRiskBand(ctx, change.EmailAccountID, models.RiskBandFromHealth(state)) } if err != nil { diff --git a/docs/content/docs/api/reference/campaigns.mdx b/docs/content/docs/api/reference/campaigns.mdx index edfc1f7a0..619a98309 100644 --- a/docs/content/docs/api/reference/campaigns.mdx +++ b/docs/content/docs/api/reference/campaigns.mdx @@ -1089,7 +1089,7 @@ The arithmetic adds up: `configured_ceiling` minus every `limits[].emails` minus } ``` -`limits[].kind` is one of `campaign_daily_limit`, `campaign_ramp`, `warmup_graduation`, `workspace_risk`, `domain_auth`, `resting`, `warmup_health_hold`, `other_campaigns`, `warmup_health_pace`, `mailbox_hours`, `sending_behavior`, `spacing`, `sending_window`, `not_running`, `org_daily_limit`, `new_lead_cap` or `leads`. `mailboxes[].state` is one of `sending`, `budget_spent`, `hours_closed`, `no_working_day`, `domain_auth`, `resting`, `health_hold`, `no_worker` (no running sending worker holds the mailbox right now; its day is still counted, since it is placed on one again within minutes) or `window_closed`; `mailboxes[].limited_by` names the clamp that set `cap_today`. See the [campaigns guide](/guides/campaigns/#todays-sending-plan) for what each limit means. +`limits[].kind` is one of `campaign_daily_limit`, `campaign_ramp`, `warmup_graduation`, `workspace_risk`, `domain_auth`, `resting`, `warmup_health_hold`, `cloud_status_hold`, `send_recovery`, `send_cooldown`, `other_campaigns`, `warmup_health_pace`, `mailbox_hours`, `sending_behavior`, `spacing`, `sending_window`, `not_running`, `org_daily_limit`, `new_lead_cap` or `leads`. `mailboxes[].state` is one of `sending`, `budget_spent`, `hours_closed`, `no_working_day`, `domain_auth`, `resting`, `health_hold`, `cloud_status_hold` (Warmbly Cloud has not reported a current status, so cold sends from the mailbox wait for one; it is not a Cloud quarantine or block), `send_recovery`, `send_cooldown`, `send_authority`, `no_worker` (no running sending worker holds the mailbox right now; its day is still counted, since it is placed on one again within minutes) or `window_closed`; `mailboxes[].limited_by` names the clamp that set `cap_today`. See the [campaigns guide](/guides/campaigns/#todays-sending-plan) for what each limit means. ## List campaign senders diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index 3635188c0..32eecf13e 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -226,6 +226,7 @@ The number is the sends already out plus what is still expected, and the list un | Easing out of warmup | A mailbox that warmed starts cold at `5` to `20` a day and adds `5` each clean day (see [easing out of warmup](#easing-out-of-warmup)) | | Workspace sending posture | A restricted workspace sends a fraction of every cap (see [workspace sending posture](/guides/deliverability/#workspace-sending-posture)) | | Domain authentication failing, Resting or in reserve, Held by warmup health | A mailbox that sends nothing cold today, and why | +| Waiting for Warmbly Cloud status | [Warmbly Cloud](/guides/warmbly-cloud/#safety-and-enforcement) has not reported a current status for a mailbox it warms. Cold sends from it wait until one arrives; it is not a Cloud quarantine or block | | Used by other campaigns | A mailbox's daily cap is shared by every campaign it is on; these sends went to another campaign today | | Slowed by warmup health | A mailbox on watch or throttled has its sends spaced wider | | Outside the mailbox's hours, Sending behaviour plan | A mailbox in another timezone or with its own [workday](/guides/sending-behavior/) is closed for the rest of today, or its rolled plan gives it fewer sends than its cap | @@ -235,7 +236,7 @@ The number is the sends already out plus what is still expected, and the list un | New leads per day | Only this many contacts may receive their first email today; follow-ups keep going | | Not enough leads due | The mailboxes could send more, but no more steps are due today | -Under the list, the card says how many leads are due now, due later today, waiting on a step's delay, inside an undecided branch window, waiting for their own mailbox (each contact keeps the address they first heard from, so a follow-up whose mailbox is spent waits rather than switching) or held, and when the campaign next works through its queue. Daily budgets reset at midnight UTC whatever the campaign's timezone, so a campaign far from UTC sees its "sent today" reset partway through its own day. Expand the mailbox rows to see each mailbox's cap for this campaign today and which limit set it, what it has sent (for this campaign, and for others), what it is still expected to send, and whether it is sending, closed for the day, out of budget, resting, held, failing authentication or reconnecting (no running sending worker holds it right now; it is placed on one again within minutes and the other mailboxes keep sending meanwhile). Each limit links to the page where it is changed. +Under the list, the card says how many leads are due now, due later today, waiting on a step's delay, inside an undecided branch window, waiting for their own mailbox (each contact keeps the address they first heard from, so a follow-up whose mailbox is spent waits rather than switching) or held, and when the campaign next works through its queue. Daily budgets reset at midnight UTC whatever the campaign's timezone, so a campaign far from UTC sees its "sent today" reset partway through its own day. Expand the mailbox rows to see each mailbox's cap for this campaign today and which limit set it, what it has sent (for this campaign, and for others), what it is still expected to send, and whether it is sending, closed for the day, out of budget, resting, held, waiting for a [Warmbly Cloud](/guides/warmbly-cloud/#safety-and-enforcement) status, failing authentication or reconnecting (no running sending worker holds it right now; it is placed on one again within minutes and the other mailboxes keep sending meanwhile). Each limit links to the page where it is changed. The sidebar's **sent today** meter reads its denominator the same way: what the workspace's mailboxes can send today under these limits, not their caps added up. The two agree by construction, because both are read through the scheduler. diff --git a/docs/content/docs/guides/mailboxes.mdx b/docs/content/docs/guides/mailboxes.mdx index 7fdc1e251..a28c95f6d 100644 --- a/docs/content/docs/guides/mailboxes.mdx +++ b/docs/content/docs/guides/mailboxes.mdx @@ -562,11 +562,13 @@ The row shows **Resting** or **Reserve** even when no campaign is active on that A mailbox rests when it is quarantined or blocked from warmup, and returns on its own three days after that lifts. One good hour does not put it back in rotation. Spam placement never rests a mailbox: a `watch` or `throttled` mailbox stays in rotation with its daily cold volume dampened (to 70% and 50%) until placement recovers. +A missing or stale [Warmbly Cloud](/guides/warmbly-cloud/#safety-and-enforcement) status is not a health signal. It never rests a mailbox, and campaign sends from the mailbox wait until a current status arrives. A rest that is already running stays in place, and its three days count only time with a current status, so a hold that ends while the status is missing does not put the mailbox back on the first fresh report. + Resting only makes sense while the mailbox is in a warmup pool, because pool health is the signal it recovers on. If a resting mailbox leaves its pool (the workspace loses warmup when a trial ends or a plan changes), there is no health signal to wait for, so the mailbox returns on its own three days after its last rest stamp rather than staying out indefinitely. Pausing warmup on the mailbox does not remove it from the pool, so its health keeps being tracked and the normal three days after it lifts still apply. ### Putting a resting mailbox back yourself -You do not have to wait. A resting mailbox's notice in the drawer has a **Put back into campaigns** button. It lands in `active` unless it is still quarantined or blocked from warmup, in which case it stays resting and the drawer says so: a mailbox the pool has taken out should not send cold mail, and there is no override for that short of the standing lifting. +You do not have to wait. A resting mailbox's notice in the drawer has a **Put back into campaigns** button. It lands in `active` unless it is still quarantined or blocked from warmup, in which case it stays resting and the drawer says so: a mailbox the pool has taken out should not send cold mail, and there is no override for that short of the standing lifting. A Warmbly Cloud mailbox whose status is unavailable lands in `active` too; its campaign sends wait for a fresh status instead of keeping it resting. API keys get the same exit through `POST /emails/:id/release`, which works on a resting mailbox as well as a held one. diff --git a/docs/content/docs/guides/warmbly-cloud.mdx b/docs/content/docs/guides/warmbly-cloud.mdx index e47d62e65..e779dd448 100644 --- a/docs/content/docs/guides/warmbly-cloud.mdx +++ b/docs/content/docs/guides/warmbly-cloud.mdx @@ -209,6 +209,8 @@ The same standing feeds the mailbox's health on the **Mailboxes** page, the **Wa Cloud standing is observed evidence, not a permanent permission. Healthy, watch and throttled observations have a 15-minute freshness limit. Missing, unrecognized or stale observations stop new control-plane admission with `cloud_evidence_unavailable` until a recognized cloud observation arrives. This availability hold does not claim the provider blocked the mailbox. Existing enrollments start with unknown freshness after upgrading, then the next successful standing sync establishes it. +An unavailable status holds only two things: campaign sends from the mailbox, and the access tokens your instance draws for a mailbox signed in through Warmbly Cloud. Both wait for a current status. Nothing else reads it as a verdict. The mailbox is not [rested](/guides/mailboxes/#resting-a-tired-mailbox), putting it back into campaigns is not refused, its health score is not lowered, and it counts as a warning rather than an error on the **Mailboxes** page and in the health summary. A rest that is already running stays in place and counts only time with a current status toward its three days. The mailbox drawer shows it as **Status unavailable**, **Settings > Warmbly Cloud** as **Unavailable**, and a campaign's sending plan as **Waiting for Warmbly Cloud status**. A real quarantine or block, whether the cloud reported it or the local reputation ledger still holds it, applies in full while the status is unavailable. + The Advisor reports an unavailable Cloud status separately from a real pool quarantine or block. A mailbox may still be warming on Cloud while its self-hosted instance lacks a current status. Open **Settings > Warmbly Cloud** and refresh the mailbox list. If the status remains unavailable, check Cloud connectivity and the consumer's `cloud_standing_sync` job, and keep the backend and consumer on the same release. Do not disconnect or re-enroll the mailbox to refresh its status. The **Warming** count indicates warmup is enabled, not that a provider has confirmed a send; the warmup count and send warnings show progress and failures. An active quarantine or block remains in force even when the cloud cannot be reached. A bounded hold still ends on its own date; an unbounded block needs review. Removing a mailbox from the cloud does not lift an active hold. Restrictive cloud evidence is retained by workspace and address in the local reputation ledger, so leaving or changing pools, reconnecting, or [exporting and importing the workspace](/guides/workspace-export-import/) does not reset it. The cloud link itself is instance-local and does not travel with the archive. Fresh positive observations from a new link do not erase an unexpired restriction from the previous link. Appeals go through the mailbox on Warmbly Cloud, where the verdict was reached. diff --git a/docs/public/openapi.json b/docs/public/openapi.json index 9c9c0fe22..638c0b32d 100644 --- a/docs/public/openapi.json +++ b/docs/public/openapi.json @@ -30667,6 +30667,9 @@ "domain_auth", "resting", "warmup_health_hold", + "cloud_status_hold", + "send_recovery", + "send_cooldown", "other_campaigns", "warmup_health_pace", "mailbox_hours", @@ -30852,6 +30855,10 @@ "domain_auth", "resting", "health_hold", + "cloud_status_hold", + "send_recovery", + "send_cooldown", + "send_authority", "no_worker", "window_closed" ] diff --git a/internal/app/advisor/detect_warmup.go b/internal/app/advisor/detect_warmup.go index 9b03747cb..5ff37c9c8 100644 --- a/internal/app/advisor/detect_warmup.go +++ b/internal/app/advisor/detect_warmup.go @@ -275,7 +275,7 @@ func detectWarmupReplyRateLow(s *repository.AdvisorSnapshot) []Finding { func detectCloudStandingUnavailable(s *repository.AdvisorSnapshot) []Finding { out := []Finding{} for _, m := range s.Mailboxes { - if m.PoolHealthReason == "cloud_evidence_unavailable" { + if m.PoolHealthReason == models.WarmupReasonCloudEvidenceUnavailable { out = append(out, Finding{ Key: "cloud_standing_unavailable", GroupTitle: "{count} mailboxes need a fresh Cloud pool status", Category: models.AdvisorCategoryWarmup, Severity: models.AdvisorHigh, Surface: models.AdvisorSurfaceMailboxes, @@ -294,7 +294,7 @@ func detectCloudStandingUnavailable(s *repository.AdvisorSnapshot) []Finding { func detectWarmupPoolThrottled(s *repository.AdvisorSnapshot) []Finding { out := []Finding{} for _, m := range s.Mailboxes { - if m.PoolHealth != string(models.WarmupHealthThrottled) || m.PoolHealthReason == "cloud_evidence_unavailable" { + if m.PoolHealth != string(models.WarmupHealthThrottled) || m.PoolHealthReason == models.WarmupReasonCloudEvidenceUnavailable { continue } remedy := "Keep warmup running and let Warmbly manage the reduced sending pace. Do not raise the daily cap to bypass the slowdown. Recovery follows the reason shown, without manual pool re-entry; separate ramp limits still apply." @@ -335,7 +335,7 @@ func detectWarmupPoolThrottled(s *repository.AdvisorSnapshot) []Finding { func detectWarmupPoolBlocked(s *repository.AdvisorSnapshot) []Finding { out := []Finding{} for _, m := range s.Mailboxes { - if m.PoolHealthReason == "cloud_evidence_unavailable" { + if m.PoolHealthReason == models.WarmupReasonCloudEvidenceUnavailable { continue } // Recoverable bands also carry blocked_until; their timer is not pool exclusion. diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 2e240a8cf..1d0c34471 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -429,7 +429,11 @@ func (s *analyticsService) assembleAccountStatus(ctx context.Context, email *mod // warmupHealthState reads the band off the API shape, defaulting to healthy // for a mailbox in no pool — the same default resolveHealthState uses. +// Unknown cloud evidence is no band at all, so it does not dampen the ramp. func warmupHealthState(h *models.WarmupHealthInfo) models.WarmupHealthState { + if h.CloudEvidenceUnavailable() { + return "" + } if h == nil || h.State == "" { return models.WarmupHealthHealthy } @@ -478,6 +482,14 @@ func applyWarmupHealth(health *models.AccountHealth, wh *models.WarmupHealthInfo if wh == nil { return } + // Sends wait for a fresh report, but a missing report says nothing about reputation. + if wh.CloudEvidenceUnavailable() { + if health.Status == "healthy" { + health.Status = "warning" + } + health.Issues = append(health.Issues, "Warmbly Cloud status unavailable") + return + } var penalty int var issue string diff --git a/internal/app/analytics/warmup_placement_test.go b/internal/app/analytics/warmup_placement_test.go index e29f4a609..9366ea8d7 100644 --- a/internal/app/analytics/warmup_placement_test.go +++ b/internal/app/analytics/warmup_placement_test.go @@ -95,3 +95,25 @@ func TestWarmupHealthKeepsReducedVolumeAsWarningWithoutDoublePenalty(t *testing. }) } } + +// A missing Warmbly Cloud report holds sends but is no verdict on reputation: +// a warning with the reason, never an error or a lower score. +func TestWarmupHealthTreatsUnknownCloudEvidenceAsAWarning(t *testing.T) { + for _, tc := range []struct { + name, initialStatus, wantStatus string + }{ + {"healthy", "healthy", "warning"}, + {"existing error", "error", "error"}, + } { + t.Run(tc.name, func(t *testing.T) { + h := models.AccountHealth{Status: tc.initialStatus, Score: 90} + applyWarmupHealth(&h, models.CloudStandingUnavailable()) + if h.Score != 90 || h.Status != tc.wantStatus || len(h.Issues) != 1 || h.Issues[0] != "Warmbly Cloud status unavailable" { + t.Fatalf("health = %+v, want %s at 90 with the unavailable reason", h, tc.wantStatus) + } + }) + } + if got := warmupHealthState(models.CloudStandingUnavailable()); got != "" { + t.Fatalf("ramp band = %q, want none", got) + } +} diff --git a/internal/app/cloudlink/list_standing_test.go b/internal/app/cloudlink/list_standing_test.go index 2c9a7e00f..98d0e81fe 100644 --- a/internal/app/cloudlink/list_standing_test.go +++ b/internal/app/cloudlink/list_standing_test.go @@ -196,3 +196,38 @@ func TestMailboxListRefreshesOnlyValidatedStandingOnItsOwningLink(t *testing.T) }) } } + +type listedStatusRepo struct { + *listedStandingRepo + statuses map[uuid.UUID]*models.WarmupHealthInfo +} + +func (r *listedStatusRepo) CloudStatuses(context.Context, []models.CloudLinkMailbox) (map[uuid.UUID]*models.WarmupHealthInfo, error) { + return r.statuses, nil +} + +// Settings shows what the mailbox drawer shows: a ledger hold beside a stale +// Cloud report is the hold, not "unavailable". +func TestMailboxListShowsTheStatusTheDrawerShows(t *testing.T) { + org, account, instance := uuid.New(), uuid.New(), uuid.New() + old := time.Now().Add(-time.Hour) + until := time.Now().Add(48 * time.Hour) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode([]models.PoolLinkMailboxState{{RemoteID: account, Health: &models.WarmupHealthInfo{State: "future_state"}}}) + })) + t.Cleanup(srv.Close) + m := &models.CloudLinkMailbox{EmailAccountID: account, RemoteID: account, InstanceID: instance, EnrollmentState: "active", Standing: &models.WarmupHealthInfo{State: "healthy"}, StandingObservedAt: &old} + hold := &models.WarmupHealthInfo{Source: models.WarmupHealthSourceCloud, State: "quarantined", Reason: "complaints", BlockedUntil: &until} + r := &listedStatusRepo{ + listedStandingRepo: &listedStandingRepo{workspaceLinkRepo: &workspaceLinkRepo{ + links: map[uuid.UUID]*models.CloudLink{instance: {InstanceID: instance, CloudURL: srv.URL, Token: "owning-link"}}, + mailboxes: map[uuid.UUID]*models.CloudLinkMailbox{account: m}, + }}, + statuses: map[uuid.UUID]*models.WarmupHealthInfo{account: hold}, + } + s := &service{repo: r, emails: enrollmentEmails{stubEmails{account: &models.Email{ID: account, OrganizationID: &org, Status: "active"}}}} + rows, xerr := s.ListMailboxes(context.Background(), org) + if xerr != nil || len(rows) != 1 || rows[0].Cloud == nil || rows[0].Cloud.Health != hold { + t.Fatalf("Settings row = %+v, %v; want the drawer's ledger hold", rows, xerr) + } +} diff --git a/internal/app/cloudlink/service.go b/internal/app/cloudlink/service.go index c3fb968fb..e08e3f66c 100644 --- a/internal/app/cloudlink/service.go +++ b/internal/app/cloudlink/service.go @@ -492,6 +492,7 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models. } enrolled, err := s.repo.ListForOrg(ctx, orgID, nil) if err != nil { + log.Error().Err(err).Str("organization_id", orgID.String()).Msg("cloud link: enrolled mailboxes could not be listed") return nil, errx.InternalError() } byAccount := make(map[uuid.UUID]models.CloudLinkMailbox, len(enrolled)) @@ -519,6 +520,7 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models. cloudByAccount[e.EmailAccountID] = state if state == nil || state.Health == nil || !knownHealthState(state.Health.State) { if err := s.repo.InvalidateStanding(ctx, e.EmailAccountID); err != nil { + log.Error().Err(err).Str("account_id", e.EmailAccountID.String()).Msg("cloud link: stale standing could not be invalidated") return nil, errx.InternalError() } e.StandingObservedAt = nil @@ -544,6 +546,26 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models. } } + // Settings shows what the mailbox drawer shows, so a ledger hold is never called unavailable. + var statuses map[uuid.UUID]*models.WarmupHealthInfo + if reader, ok := s.repo.(repository.CloudStatusReader); ok { + listed := make([]models.CloudLinkMailbox, 0, len(byAccount)) + for _, e := range byAccount { + listed = append(listed, e) + } + var err error + if statuses, err = reader.CloudStatuses(ctx, listed); err != nil { + log.Error().Err(err).Str("organization_id", orgID.String()).Msg("cloud link: mailbox statuses could not be read") + return nil, errx.InternalError() + } + } + healthFor := func(e models.CloudLinkMailbox) *models.WarmupHealthInfo { + if statuses != nil { + return statuses[e.EmailAccountID] + } + return e.EffectiveStanding(time.Now()) + } + rows := make([]models.CloudLinkMailboxRow, 0, len(accounts)) seen := make(map[uuid.UUID]bool, len(accounts)) add := func(a *models.Email) { @@ -560,7 +582,7 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models. row.Cloud = cloudByAccount[a.ID] // The recorded standing covers a mailbox the cloud holds out of its pool. if row.Cloud != nil { - row.Cloud.Health = e.EffectiveStanding(time.Now()) + row.Cloud.Health = healthFor(e) } } rows = append(rows, row) diff --git a/internal/app/consumer/lifecycle_rebalancer.go b/internal/app/consumer/lifecycle_rebalancer.go index dd91d08d3..229145a4e 100644 --- a/internal/app/consumer/lifecycle_rebalancer.go +++ b/internal/app/consumer/lifecycle_rebalancer.go @@ -41,7 +41,7 @@ func (s *JobsService) rebalanceLifecycles(ctx context.Context) { seen := make([]uuid.UUID, 0, len(candidates)) for _, c := range candidates { seen = append(seen, c.EmailAccountID) - d := lifecycle.Decide(c.Current, c.Since, c.HealthState, now) + d := lifecycle.Decide(c.Current, c.Since, c.HealthState, c.CloudEvidenceUnknown, now) if d.RestartProbation { if err := s.LifecycleRepo.RestartProbation(ctx, c.EmailAccountID); err != nil { log.Warn().Err(err).Str("email_account_id", c.EmailAccountID.String()). diff --git a/internal/app/consumer/lifecycle_rebalancer_test.go b/internal/app/consumer/lifecycle_rebalancer_test.go new file mode 100644 index 000000000..13200a5b3 --- /dev/null +++ b/internal/app/consumer/lifecycle_rebalancer_test.go @@ -0,0 +1,67 @@ +package jobs + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type lifecycleRebalanceRepo struct { + repository.SendLifecycleRepository + candidates []repository.LifecycleCandidate + moved map[uuid.UUID]models.SendLifecycle + restarted []uuid.UUID + checked []uuid.UUID +} + +func (r *lifecycleRebalanceRepo) ListLifecycleCandidates(context.Context, int) ([]repository.LifecycleCandidate, error) { + return r.candidates, nil +} + +func (r *lifecycleRebalanceRepo) SetSendLifecycle(_ context.Context, id uuid.UUID, state models.SendLifecycle, _ string, _ bool) (bool, error) { + r.moved[id] = state + return true, nil +} + +func (r *lifecycleRebalanceRepo) RestartProbation(_ context.Context, id uuid.UUID) error { + r.restarted = append(r.restarted, id) + return nil +} + +func (r *lifecycleRebalanceRepo) MarkLifecycleChecked(_ context.Context, ids []uuid.UUID) error { + r.checked = append(r.checked, ids...) + return nil +} + +// A missing or stale Warmbly Cloud report never rests or resumes a mailbox, and +// restarts a running rest's probation; a real hold beside it still rests. +func TestLifecycleRebalancerNeverRestsOnUnknownCloudEvidence(t *testing.T) { + served := time.Now().Add(-2 * models.RestProbation) + active, resting, held := uuid.New(), uuid.New(), uuid.New() + repo := &lifecycleRebalanceRepo{moved: map[uuid.UUID]models.SendLifecycle{}, candidates: []repository.LifecycleCandidate{ + {EmailAccountID: active, Current: models.SendLifecycleActive, CloudEvidenceUnknown: true}, + {EmailAccountID: resting, Current: models.SendLifecycleResting, Since: &served, CloudEvidenceUnknown: true}, + {EmailAccountID: held, Current: models.SendLifecycleActive, HealthState: models.WarmupHealthQuarantined, CloudEvidenceUnknown: true}, + }} + (&JobsService{LifecycleRepo: repo}).rebalanceLifecycles(context.Background()) + + if _, ok := repo.moved[active]; ok { + t.Fatalf("unknown evidence moved an active mailbox to %q", repo.moved[active]) + } + if _, ok := repo.moved[resting]; ok { + t.Fatalf("unknown evidence moved a resting mailbox to %q", repo.moved[resting]) + } + if len(repo.restarted) != 1 || repo.restarted[0] != resting { + t.Fatalf("restarted probation for %v, want only the resting mailbox", repo.restarted) + } + if repo.moved[held] != models.SendLifecycleResting { + t.Fatalf("a real quarantine moved to %q, want resting", repo.moved[held]) + } + if len(repo.checked) != 3 { + t.Fatalf("checked %d mailboxes, want all 3", len(repo.checked)) + } +} diff --git a/internal/app/consumer/risk_mark.go b/internal/app/consumer/risk_mark.go index bcbaf3851..4c2a1dd41 100644 --- a/internal/app/consumer/risk_mark.go +++ b/internal/app/consumer/risk_mark.go @@ -18,6 +18,9 @@ func (s *JobsService) markRiskBandFromWarmupHealth(ctx context.Context, accountI state = health.HealthState } else if resolved, ok := s.resolveWarmupHealthState(ctx, accountID); ok { state = resolved + } else { + // No reading is no verdict: an error or a missing Cloud report keeps the band. + return } band := models.RiskBandFromHealth(state) @@ -37,8 +40,8 @@ func (s *JobsService) resolveWarmupHealthState(ctx context.Context, accountID uu // The mailbox's standing, including one Warmbly Cloud reported for a // mailbox it warms, so a local event cannot reset a cloud hold's band. - state, _, err := s.WarmupRepo.GetHealthState(ctx, accountID) - if err != nil { + state, cloudUnknown, err := s.WarmupRepo.GetKnownHealthState(ctx, accountID) + if err != nil || (cloudUnknown && !state.Holds()) { return "", false } return state, true diff --git a/internal/app/consumer/risk_rebalancer.go b/internal/app/consumer/risk_rebalancer.go index c1254a1cb..067bea141 100644 --- a/internal/app/consumer/risk_rebalancer.go +++ b/internal/app/consumer/risk_rebalancer.go @@ -50,6 +50,10 @@ func (s *JobsService) rebalanceRisk(ctx context.Context) { var updatedBand int for _, c := range candidates { + // A missing Cloud report is no verdict; only a real hold beside it moves the band. + if c.CloudEvidenceUnknown && !c.HealthState.Holds() { + continue + } newBand := models.RiskBandFromHealth(c.HealthState) if newBand == c.CurrentBand { continue diff --git a/internal/app/consumer/risk_rebalancer_test.go b/internal/app/consumer/risk_rebalancer_test.go new file mode 100644 index 000000000..fbc5b8994 --- /dev/null +++ b/internal/app/consumer/risk_rebalancer_test.go @@ -0,0 +1,82 @@ +package jobs + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type riskRebalanceRepo struct { + repository.WorkerRepository + candidates []repository.RiskCandidate + bands map[uuid.UUID]models.EmailRiskBand +} + +func (r *riskRebalanceRepo) ListRiskCandidates(context.Context, int) ([]repository.RiskCandidate, error) { + return r.candidates, nil +} + +func (r *riskRebalanceRepo) SetEmailAccountRiskBand(_ context.Context, id uuid.UUID, band models.EmailRiskBand) error { + r.bands[id] = band + return nil +} + +// A missing Warmbly Cloud report keeps the band it had; a real hold beside it still moves it. +func TestRiskRebalancerSkipsUnknownCloudEvidence(t *testing.T) { + unknown, held := uuid.New(), uuid.New() + repo := &riskRebalanceRepo{bands: map[uuid.UUID]models.EmailRiskBand{}, candidates: []repository.RiskCandidate{ + {EmailAccountID: unknown, CurrentBand: models.EmailRiskBandRisky, CloudEvidenceUnknown: true}, + {EmailAccountID: held, CurrentBand: models.EmailRiskBandClean, HealthState: models.WarmupHealthBlocked, CloudEvidenceUnknown: true}, + }} + (&JobsService{WorkerRepo: repo}).rebalanceRisk(context.Background()) + + if band, ok := repo.bands[unknown]; ok { + t.Fatalf("unknown evidence moved the band to %q", band) + } + if repo.bands[held] != models.EmailRiskBandQuarantine { + t.Fatalf("a real block set %q, want quarantine", repo.bands[held]) + } +} + +type riskMarkWarmupRepo struct { + repository.WarmupRepository + state models.WarmupHealthState + unknown bool + err error +} + +func (r riskMarkWarmupRepo) GetKnownHealthState(context.Context, uuid.UUID) (models.WarmupHealthState, bool, error) { + return r.state, r.unknown, r.err +} + +// A local event re-reads the standing; a missing Cloud report or a failed read must not reset the band. +func TestRiskMarkKeepsTheBandOnUnknownCloudEvidence(t *testing.T) { + for _, tc := range []struct { + name string + state models.WarmupHealthState + unknown bool + err error + want models.EmailRiskBand + writes bool + }{ + {"unknown", models.WarmupHealthHealthy, true, nil, "", false}, + {"unknown beside a real hold", models.WarmupHealthQuarantined, true, nil, models.EmailRiskBandQuarantine, true}, + {"known", models.WarmupHealthWatch, false, nil, models.EmailRiskBandRisky, true}, + {"read error", models.WarmupHealthHealthy, false, errors.New("database unavailable"), "", false}, + } { + t.Run(tc.name, func(t *testing.T) { + id := uuid.New() + repo := &riskRebalanceRepo{bands: map[uuid.UUID]models.EmailRiskBand{}} + s := &JobsService{WorkerRepo: repo, WarmupRepo: riskMarkWarmupRepo{state: tc.state, unknown: tc.unknown, err: tc.err}} + s.markRiskBandFromWarmupHealth(context.Background(), id, nil) + band, wrote := repo.bands[id] + if wrote != tc.writes || band != tc.want { + t.Fatalf("band = %q (written %v), want %q (written %v)", band, wrote, tc.want, tc.writes) + } + }) + } +} diff --git a/internal/app/email/handler.go b/internal/app/email/handler.go index 9aff7bd80..7dce4bba1 100644 --- a/internal/app/email/handler.go +++ b/internal/app/email/handler.go @@ -151,7 +151,7 @@ func (s *emailService) SetSendHold(ctx context.Context, orgID, emailAccountID st log.Error().Err(err).Str("email_account_id", account.ID.String()).Msg("could not read send lifecycle") return nil, errx.InternalError() } - d := lifecycle.Decide(models.SendLifecycleActive, nil, candidate.HealthState, time.Now()) + d := lifecycle.Decide(models.SendLifecycleActive, nil, candidate.HealthState, candidate.CloudEvidenceUnknown, time.Now()) next, reason = d.Next, "released by its owner" if candidate.Current == models.SendLifecycleResting { reason = "put back by its owner" diff --git a/internal/app/email/send_hold_test.go b/internal/app/email/send_hold_test.go new file mode 100644 index 000000000..30b01c69a --- /dev/null +++ b/internal/app/email/send_hold_test.go @@ -0,0 +1,84 @@ +package email + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type sendHoldEmailRepo struct { + repository.EmailRepository + account *models.Email +} + +func (r *sendHoldEmailRepo) Get(context.Context, string, string) (*models.Email, *errx.Error) { + return r.account, nil +} + +type sendHoldLifecycleRepo struct { + repository.SendLifecycleRepository + candidate repository.LifecycleCandidate + set []models.SendLifecycle + reasons []string +} + +func (r *sendHoldLifecycleRepo) GetLifecycleCandidate(context.Context, uuid.UUID) (*repository.LifecycleCandidate, error) { + c := r.candidate + return &c, nil +} + +func (r *sendHoldLifecycleRepo) SetSendLifecycle(_ context.Context, _ uuid.UUID, state models.SendLifecycle, reason string, _ bool) (bool, error) { + r.set = append(r.set, state) + r.reasons = append(r.reasons, reason) + r.candidate.Current = state + return true, nil +} + +func (r *sendHoldLifecycleRepo) GetSendLifecycles(_ context.Context, ids []uuid.UUID) (map[uuid.UUID]models.SendLifecycleState, error) { + out := map[uuid.UUID]models.SendLifecycleState{} + for _, id := range ids { + out[id] = models.SendLifecycleState{State: r.candidate.Current} + } + return out, nil +} + +func sendHoldFixture(candidate repository.LifecycleCandidate) (*emailService, *sendHoldLifecycleRepo, *models.Email) { + account := &models.Email{ID: uuid.New()} + candidate.EmailAccountID = account.ID + lifecycle := &sendHoldLifecycleRepo{candidate: candidate} + return &emailService{emailRepository: &sendHoldEmailRepo{account: account}, lifecycleRepo: lifecycle}, lifecycle, account +} + +// Putting a mailbox back while Warmbly Cloud has not reported lands it in +// active; the send gate still holds its sends until a fresh report arrives. +func TestSendHoldReleaseWithUnknownCloudEvidenceLandsActive(t *testing.T) { + since := time.Now().Add(-time.Hour) + svc, lifecycle, account := sendHoldFixture(repository.LifecycleCandidate{ + Current: models.SendLifecycleResting, Since: &since, CloudEvidenceUnknown: true, + }) + state, xerr := svc.SetSendHold(context.Background(), uuid.NewString(), account.ID.String(), false) + if xerr != nil { + t.Fatalf("SetSendHold: %v", xerr) + } + if len(lifecycle.set) != 1 || lifecycle.set[0] != models.SendLifecycleActive || state.State != models.SendLifecycleActive { + t.Fatalf("release moved to %v (state %+v), want active", lifecycle.set, state) + } +} + +// A real hold beside unknown Cloud evidence still keeps the mailbox resting. +func TestSendHoldReleaseWithARealHoldStaysResting(t *testing.T) { + svc, lifecycle, account := sendHoldFixture(repository.LifecycleCandidate{ + Current: models.SendLifecycleResting, HealthState: models.WarmupHealthQuarantined, CloudEvidenceUnknown: true, + }) + if _, xerr := svc.SetSendHold(context.Background(), uuid.NewString(), account.ID.String(), false); xerr != nil { + t.Fatalf("SetSendHold: %v", xerr) + } + if len(lifecycle.set) != 1 || lifecycle.set[0] != models.SendLifecycleResting { + t.Fatalf("release moved to %v, want resting", lifecycle.set) + } +} diff --git a/internal/app/lifecycle/lifecycle.go b/internal/app/lifecycle/lifecycle.go index 04e7fd9f5..daf8f4cb7 100644 --- a/internal/app/lifecycle/lifecycle.go +++ b/internal/app/lifecycle/lifecycle.go @@ -28,7 +28,8 @@ func (d Decision) Changed(current models.SendLifecycle) bool { return d.Next != // Decide maps warmup health onto the cold lifecycle. Rests only at quarantined // and blocked: watch and throttled are spam placement, which slows cold volume // through the send budget and never takes a mailbox out of rotation. -func Decide(current models.SendLifecycle, since *time.Time, health models.WarmupHealthState, now time.Time) Decision { +// health is real evidence only; cloudUnknown means Warmbly Cloud has no current report. +func Decide(current models.SendLifecycle, since *time.Time, health models.WarmupHealthState, cloudUnknown bool, now time.Time) Decision { if !current.AutoManaged() { return Decision{Next: current} } @@ -40,6 +41,12 @@ func Decide(current models.SendLifecycle, since *time.Time, health models.Warmup Reason: "warmup health is " + string(health) + "; out of cold rotation until it recovers"} } + // A missing report never starts a rest, and time without one is not probation. + if cloudUnknown { + return Decision{Next: current, RestartProbation: current == models.SendLifecycleResting, + Reason: "Warmbly Cloud has not reported a current status"} + } + // No pool, no signal: never rest, and let a resting clock run rather than strand the mailbox. if health == "" { if current == models.SendLifecycleResting { diff --git a/internal/app/lifecycle/lifecycle_test.go b/internal/app/lifecycle/lifecycle_test.go index d434d56d1..22552f14c 100644 --- a/internal/app/lifecycle/lifecycle_test.go +++ b/internal/app/lifecycle/lifecycle_test.go @@ -12,7 +12,7 @@ func TestDecideRestsOnRealTrouble(t *testing.T) { for _, h := range []models.WarmupHealthState{ models.WarmupHealthQuarantined, models.WarmupHealthBlocked, } { - d := Decide(models.SendLifecycleActive, nil, h, now) + d := Decide(models.SendLifecycleActive, nil, h, false, now) if d.Next != models.SendLifecycleResting { t.Errorf("health %q gave %q, want resting", h, d.Next) } @@ -26,7 +26,7 @@ func TestDecideRestsOnRealTrouble(t *testing.T) { // cold budget is dampened, and it stays in rotation. func TestDecideDoesNotRestOnPlacementBands(t *testing.T) { for _, h := range []models.WarmupHealthState{models.WarmupHealthWatch, models.WarmupHealthThrottled} { - if d := Decide(models.SendLifecycleActive, nil, h, time.Now()); d.Next != models.SendLifecycleActive { + if d := Decide(models.SendLifecycleActive, nil, h, false, time.Now()); d.Next != models.SendLifecycleActive { t.Errorf("%s gave %q, want active", h, d.Next) } } @@ -36,19 +36,19 @@ func TestDecideResumesOnlyAfterProbation(t *testing.T) { now := time.Date(2026, 3, 10, 12, 0, 0, 0, time.UTC) served := now.Add(-models.RestProbation) - if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthHealthy, now); d.Next != models.SendLifecycleActive { + if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthHealthy, false, now); d.Next != models.SendLifecycleActive { t.Errorf("a recovered mailbox that served its rest gave %q, want active", d.Next) } fresh := now.Add(-time.Hour) - if d := Decide(models.SendLifecycleResting, &fresh, models.WarmupHealthHealthy, now); d.Next != models.SendLifecycleResting { + if d := Decide(models.SendLifecycleResting, &fresh, models.WarmupHealthHealthy, false, now); d.Next != models.SendLifecycleResting { t.Errorf("an hour of rest gave %q, want it still resting", d.Next) } - if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthQuarantined, now); d.Next != models.SendLifecycleResting { + if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthQuarantined, false, now); d.Next != models.SendLifecycleResting { t.Errorf("a still-quarantined mailbox gave %q, want resting", d.Next) } - if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthThrottled, now); d.Next != models.SendLifecycleActive { + if d := Decide(models.SendLifecycleResting, &served, models.WarmupHealthThrottled, false, now); d.Next != models.SendLifecycleActive { t.Errorf("a throttled mailbox that served its rest gave %q, want active at its dampened volume", d.Next) } } @@ -59,7 +59,7 @@ func TestDecideNeverTouchesReserve(t *testing.T) { for _, h := range []models.WarmupHealthState{ models.WarmupHealthHealthy, models.WarmupHealthThrottled, models.WarmupHealthBlocked, } { - if d := Decide(models.SendLifecycleReserve, nil, h, now); d.Next != models.SendLifecycleReserve { + if d := Decide(models.SendLifecycleReserve, nil, h, false, now); d.Next != models.SendLifecycleReserve { t.Errorf("health %q moved a reserved mailbox to %q", h, d.Next) } } @@ -68,7 +68,7 @@ func TestDecideNeverTouchesReserve(t *testing.T) { // A row written before the column existed reads as empty and must be treated // as active rather than left in limbo. func TestDecideTreatsAnUnsetStateAsActive(t *testing.T) { - if d := Decide("", nil, models.WarmupHealthHealthy, time.Now()); d.Next != models.SendLifecycleActive { + if d := Decide("", nil, models.WarmupHealthHealthy, false, time.Now()); d.Next != models.SendLifecycleActive { t.Errorf("an unset lifecycle gave %q, want active", d.Next) } } @@ -80,7 +80,7 @@ func TestDecideRestartsProbationWhileStillOut(t *testing.T) { now := time.Date(2026, 3, 10, 12, 0, 0, 0, time.UTC) longAgo := now.Add(-10 * 24 * time.Hour) - d := Decide(models.SendLifecycleResting, &longAgo, models.WarmupHealthQuarantined, now) + d := Decide(models.SendLifecycleResting, &longAgo, models.WarmupHealthQuarantined, false, now) if d.Next != models.SendLifecycleResting { t.Errorf("next = %q, want it still resting", d.Next) } @@ -89,7 +89,7 @@ func TestDecideRestartsProbationWhileStillOut(t *testing.T) { } // Once healthy, the clock runs and is not restarted. - healthy := Decide(models.SendLifecycleResting, &longAgo, models.WarmupHealthHealthy, now) + healthy := Decide(models.SendLifecycleResting, &longAgo, models.WarmupHealthHealthy, false, now) if healthy.RestartProbation { t.Error("a healthy mailbox must not have its probation restarted") } @@ -104,7 +104,7 @@ func TestDecideDoesNotRestartProbationForOtherStates(t *testing.T) { for _, state := range []models.SendLifecycle{ models.SendLifecycleActive, models.SendLifecycleReserve, "", } { - if d := Decide(state, nil, models.WarmupHealthWatch, now); d.RestartProbation { + if d := Decide(state, nil, models.WarmupHealthWatch, false, now); d.RestartProbation { t.Errorf("state %q asked to restart probation", state) } } @@ -119,7 +119,7 @@ func TestUnknownHealthResumesOnlyAfterTheRestWindow(t *testing.T) { now := time.Date(2026, 3, 10, 12, 0, 0, 0, time.UTC) fresh := now.Add(-time.Hour) - d := Decide(models.SendLifecycleResting, &fresh, "", now) + d := Decide(models.SendLifecycleResting, &fresh, "", false, now) if d.Next != models.SendLifecycleResting { t.Fatalf("resting mailbox with no pool resumed after an hour: %v", d.Next) } @@ -128,7 +128,7 @@ func TestUnknownHealthResumesOnlyAfterTheRestWindow(t *testing.T) { } served := now.Add(-models.RestProbation) - d = Decide(models.SendLifecycleResting, &served, "", now) + d = Decide(models.SendLifecycleResting, &served, "", false, now) if d.Next != models.SendLifecycleActive { t.Fatalf("resting mailbox with no pool stayed %v after the rest window", d.Next) } @@ -136,7 +136,7 @@ func TestUnknownHealthResumesOnlyAfterTheRestWindow(t *testing.T) { t.Fatal("resuming with no signal needs a reason the owner can read") } - if d := Decide(models.SendLifecycleActive, &served, "", now); d.Next != models.SendLifecycleActive { + if d := Decide(models.SendLifecycleActive, &served, "", false, now); d.Next != models.SendLifecycleActive { t.Fatalf("active mailbox with no pool moved to %v", d.Next) } } @@ -144,7 +144,41 @@ func TestUnknownHealthResumesOnlyAfterTheRestWindow(t *testing.T) { // A resting mailbox with no clock cannot serve a window, so it stays put // rather than resuming on a missing stamp. func TestUnknownHealthWithNoClockStaysResting(t *testing.T) { - if d := Decide(models.SendLifecycleResting, nil, "", time.Now()); d.Next != models.SendLifecycleResting { + if d := Decide(models.SendLifecycleResting, nil, "", false, time.Now()); d.Next != models.SendLifecycleResting { t.Fatalf("resting mailbox with no since resumed: %v", d.Next) } } + +// A missing Warmbly Cloud report holds sends at the gate, but it is not a +// verdict: it never starts a rest, and time without one never counts as probation. +func TestCloudUnknownNeverStartsARestOrServesProbation(t *testing.T) { + now := time.Date(2026, 3, 10, 12, 0, 0, 0, time.UTC) + served := now.Add(-models.RestProbation) + + d := Decide(models.SendLifecycleActive, &served, "", true, now) + if d.Next != models.SendLifecycleActive || d.RestartProbation { + t.Fatalf("active with unknown cloud evidence gave %+v, want it left active", d) + } + + // A Cloud hold that lapsed during an outage must not resume on the first fresh report. + d = Decide(models.SendLifecycleResting, &served, "", true, now) + if d.Next != models.SendLifecycleResting || !d.RestartProbation { + t.Fatalf("resting with unknown cloud evidence gave %+v, want it resting with probation restarted", d) + } + if d.Reason == "" { + t.Fatal("holding on unknown cloud evidence needs a reason the owner can read") + } + + d = Decide(models.SendLifecycleActive, nil, models.WarmupHealthQuarantined, true, now) + if d.Next != models.SendLifecycleResting { + t.Fatalf("a real quarantine beside unknown cloud evidence gave %q, want resting", d.Next) + } + d = Decide(models.SendLifecycleResting, &served, models.WarmupHealthBlocked, true, now) + if d.Next != models.SendLifecycleResting || !d.RestartProbation { + t.Fatalf("a real block beside unknown cloud evidence gave %+v, want resting with probation restarted", d) + } + + if d := Decide(models.SendLifecycleReserve, nil, "", true, now); d.Next != models.SendLifecycleReserve { + t.Fatalf("unknown cloud evidence moved a reserved mailbox to %q", d.Next) + } +} diff --git a/internal/models/campaign_send_plan.go b/internal/models/campaign_send_plan.go index 241a51c03..1170993ef 100644 --- a/internal/models/campaign_send_plan.go +++ b/internal/models/campaign_send_plan.go @@ -84,6 +84,7 @@ const ( SendLimitDomainAuth = "domain_auth" SendLimitResting = "resting" SendLimitHealthHold = "warmup_health_hold" + SendLimitCloudStatusHold = "cloud_status_hold" SendLimitRecovery = "send_recovery" SendLimitCooldown = "send_cooldown" SendLimitOtherCampaigns = "other_campaigns" @@ -166,7 +167,7 @@ type CampaignMailboxPlan struct { ExpectedRemaining int `json:"expected_remaining"` // State is why the mailbox is or is not sending right now. One of // "sending", "budget_spent", "hours_closed", "no_working_day", - // "domain_auth", "resting", "health_hold". + // "domain_auth", "resting", "health_hold", "cloud_status_hold". State string `json:"state"` // ReopensAt is when a mailbox outside its own hours is next open. ReopensAt *time.Time `json:"reopens_at,omitempty"` @@ -189,6 +190,7 @@ const ( MailboxPlanDomainAuth = "domain_auth" MailboxPlanResting = "resting" MailboxPlanHealthHold = "health_hold" + MailboxPlanCloudStatus = "cloud_status_hold" MailboxPlanRecovery = "send_recovery" MailboxPlanCooldown = "send_cooldown" MailboxPlanAdmission = "send_authority" diff --git a/internal/models/cloud_standing.go b/internal/models/cloud_standing.go index d78b26076..abe2e7eb8 100644 --- a/internal/models/cloud_standing.go +++ b/internal/models/cloud_standing.go @@ -2,12 +2,25 @@ package models import "time" +// WarmupReasonCloudEvidenceUnavailable marks a missing Cloud report: it holds sends but is no verdict. +const WarmupReasonCloudEvidenceUnavailable = "cloud_evidence_unavailable" + +// CloudStandingUnavailable is the standing shown while Warmbly Cloud has no current report. +func CloudStandingUnavailable() *WarmupHealthInfo { + return &WarmupHealthInfo{State: string(WarmupHealthBlocked), Reason: WarmupReasonCloudEvidenceUnavailable, Source: WarmupHealthSourceCloud} +} + +// CloudEvidenceUnavailable reports whether h is CloudStandingUnavailable rather than a real standing. +func (h *WarmupHealthInfo) CloudEvidenceUnavailable() bool { + return h != nil && h.Reason == WarmupReasonCloudEvidenceUnavailable +} + func (m CloudLinkMailbox) EffectiveStanding(now time.Time) *WarmupHealthInfo { if h := m.Standing; h != nil && (h.State == string(WarmupHealthBlocked) || h.State == string(WarmupHealthQuarantined)) && (h.BlockedUntil == nil || h.BlockedUntil.After(now)) { return h } if m.EnrollmentState != "active" || m.Standing == nil || m.StandingObservedAt == nil || m.StandingObservedAt.After(now) || !m.StandingObservedAt.After(now.Add(-15*time.Minute)) { - return &WarmupHealthInfo{State: string(WarmupHealthBlocked), Reason: "cloud_evidence_unavailable"} + return CloudStandingUnavailable() } return m.Standing } diff --git a/internal/models/warmup.go b/internal/models/warmup.go index b9a1864a8..45628d2e2 100644 --- a/internal/models/warmup.go +++ b/internal/models/warmup.go @@ -317,6 +317,11 @@ const ( WarmupHealthBlocked WarmupHealthState = "blocked" ) +// Holds reports whether s takes a mailbox out of the pool and out of cold rotation. +func (s WarmupHealthState) Holds() bool { + return s == WarmupHealthQuarantined || s == WarmupHealthBlocked +} + type WarmupParticipantHealth struct { PoolID uuid.UUID `json:"pool_id"` PoolType string `json:"pool_type"` diff --git a/internal/repository/cloud_link_standing_live_test.go b/internal/repository/cloud_link_standing_live_test.go index c222b8ebb..3f2b083dc 100644 --- a/internal/repository/cloud_link_standing_live_test.go +++ b/internal/repository/cloud_link_standing_live_test.go @@ -1,6 +1,7 @@ package repository import ( + "cmp" "context" "errors" "testing" @@ -100,19 +101,62 @@ func TestLiveCloudStandingFreshnessIsBoundedWithoutInventingAProviderBlock(t *te } } } + // The gates above fail closed; what acts on reputation reads real evidence only. + handle, _ := liveContactDB(t) + lifecycle := NewSendLifecycleRepository(handle) + assertActsOn := func(want models.WarmupHealthState, unknown bool) { + t.Helper() + cand, err := lifecycle.GetLifecycleCandidate(ctx, f.sender) + if err != nil || cand.HealthState != want || cand.CloudEvidenceUnknown != unknown { + t.Fatalf("lifecycle candidate = %+v, %v; want %q with unknown=%v", cand, err, want, unknown) + } + known, gotUnknown, err := f.warmup.GetKnownHealthState(ctx, f.sender) + if wantKnown := cmp.Or(want, models.WarmupHealthHealthy); err != nil || known != wantKnown || gotUnknown != unknown { + t.Fatalf("known standing = %s, %v, %v; want %s with unknown=%v", known, gotUnknown, err, wantKnown, unknown) + } + wantWarnings := 0 + if unknown || want == models.WarmupHealthWatch { + wantWarnings = 1 + } + summary, xerr := NewAnalyticsRepository(handle).GetAccountHealthSummary(ctx, f.org, nil) + if xerr != nil || summary.ErrorAccounts != 0 || summary.WarningAccounts != wantWarnings { + t.Fatalf("summary = %+v, %v; want %d warning and no error", summary, xerr, wantWarnings) + } + risk, err := NewWorkerRepository(f.pool).ListRiskCandidates(ctx, 100000) + if err != nil { + t.Fatal(err) + } + found := false + for _, c := range risk { + if c.EmailAccountID == f.sender { + found = c.HealthState == want && c.CloudEvidenceUnknown == unknown + } + } + if !found { + t.Fatalf("risk candidate does not read %q with unknown=%v", want, unknown) + } + states, err := f.warmup.GetHealthStates(ctx, []uuid.UUID{f.sender}) + if err != nil || states[f.sender].CloudStatusOnly != unknown || states[f.sender].CloudReportMissing != unknown { + t.Fatalf("send gate read = %+v, %v; want CloudStatusOnly and CloudReportMissing %v", states[f.sender], err, unknown) + } + } assertState(models.WarmupHealthBlocked, "cloud_evidence_unavailable") + assertActsOn("", true) if _, err := links.SetStanding(ctx, f.sender, &models.WarmupHealthInfo{State: "healthy"}, false); err != nil { t.Fatal(err) } assertState(models.WarmupHealthHealthy, "") + assertActsOn(models.WarmupHealthHealthy, false) if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET standing_observed_at=NOW()+INTERVAL '1 minute' WHERE email_account_id=$1`, f.sender); err != nil { t.Fatal(err) } assertState(models.WarmupHealthBlocked, "cloud_evidence_unavailable") + assertActsOn("", true) if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET standing_observed_at = NOW() - INTERVAL '16 minutes' WHERE email_account_id = $1`, f.sender); err != nil { t.Fatal(err) } assertState(models.WarmupHealthBlocked, "cloud_evidence_unavailable") + assertActsOn("", true) var stored string if err := f.pool.QueryRow(ctx, `SELECT health_state FROM cloud_link_mailboxes WHERE email_account_id = $1`, f.sender).Scan(&stored); err != nil || stored != "healthy" { t.Fatalf("availability invented provider evidence: %s, %v", stored, err) @@ -121,10 +165,231 @@ func TestLiveCloudStandingFreshnessIsBoundedWithoutInventingAProviderBlock(t *te t.Fatal(err) } assertState(models.WarmupHealthWatch, "") + assertActsOn(models.WarmupHealthWatch, false) if err := links.InvalidateStanding(ctx, f.sender); err != nil { t.Fatal(err) } assertState(models.WarmupHealthBlocked, "cloud_evidence_unavailable") + assertActsOn("", true) +} + +// A stale Cloud report must not outrank a real hold: the gates still fail +// closed, and what acts on reputation still sees the quarantine. +func TestLiveCloudUnknownEvidenceDoesNotHideARealHold(t *testing.T) { + f := newPoolLinkFixture(t) + ctx := context.Background() + links := NewCloudLinkRepository(f.pool, nil) + handle, _ := liveContactDB(t) + if _, err := links.Enroll(ctx, f.sender, f.sender, liveCloudLink(t, f), false); err != nil { + t.Fatal(err) + } + if _, err := links.SetStanding(ctx, f.sender, &models.WarmupHealthInfo{State: "healthy"}, false); err != nil { + t.Fatal(err) + } + until := time.Now().Add(48 * time.Hour) + if _, err := f.pool.Exec(ctx, `INSERT INTO warmup_reputation_ledger (organization_id, email, cloud_health_state, cloud_blocked_until, cloud_health_reason, standing_until) + VALUES ($1, $2, 'quarantined', $3, 'complaints', NOW())`, f.org, f.senderTo, until); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET standing_observed_at = NOW() - INTERVAL '16 minutes' WHERE email_account_id = $1`, f.sender); err != nil { + t.Fatal(err) + } + + // The gate reads the real quarantine, open-ended: the missing report outlasts the ledger's end. + if state, gateUntil, err := f.warmup.GetHealthState(ctx, f.sender); err != nil || state != models.WarmupHealthQuarantined || gateUntil != nil { + t.Fatalf("send gate = %s until %v, %v; want the ledger quarantine with no end", state, gateUntil, err) + } + cand, err := NewSendLifecycleRepository(handle).GetLifecycleCandidate(ctx, f.sender) + if err != nil || cand.HealthState != models.WarmupHealthQuarantined || !cand.CloudEvidenceUnknown { + t.Fatalf("lifecycle candidate = %+v, %v; want quarantined with unknown cloud evidence", cand, err) + } + if state, unknown, err := f.warmup.GetKnownHealthState(ctx, f.sender); err != nil || state != models.WarmupHealthQuarantined || !unknown { + t.Fatalf("known standing = %s, %v, %v; want quarantined with unknown cloud evidence", state, unknown, err) + } + summary, xerr := NewAnalyticsRepository(handle).GetAccountHealthSummary(ctx, f.org, nil) + if xerr != nil || summary.ErrorAccounts != 1 { + t.Fatalf("summary = %+v, %v; want the quarantined mailbox in error", summary, xerr) + } + risk, err := NewWorkerRepository(f.pool).ListRiskCandidates(ctx, 100000) + if err != nil { + t.Fatal(err) + } + found := false + for _, c := range risk { + if c.EmailAccountID == f.sender { + found = c.HealthState == models.WarmupHealthQuarantined && c.CloudEvidenceUnknown + } + } + if !found { + t.Fatal("risk candidate does not see the quarantine beside unknown cloud evidence") + } + assertCloudStatus(t, f, models.WarmupHealthQuarantined, "complaints") + if states, err := f.warmup.GetHealthStates(ctx, []uuid.UUID{f.sender}); err != nil || states[f.sender].CloudStatusOnly || states[f.sender].State != models.WarmupHealthQuarantined || + !states[f.sender].CloudReportMissing || states[f.sender].BlockedUntil != nil { + t.Fatalf("send gate read = %+v, %v; want the real quarantine, open-ended", states[f.sender], err) + } + snap, err := NewAdvisorRepository(handle).LoadSnapshot(ctx, f.org, time.Now()) + if err != nil { + t.Fatal(err) + } + for _, m := range snap.Mailboxes { + if m.ID == f.sender && (m.PoolHealth != string(models.WarmupHealthQuarantined) || m.PoolHealthReason != "complaints" || !m.PoolBlocked) { + t.Fatalf("advisor standing = %s (%s, blocked %v), want the ledger quarantine", m.PoolHealth, m.PoolHealthReason, m.PoolBlocked) + } + } + + // A stale Cloud report of the same state that has ended must not stand in for the live hold. + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET health_state = 'quarantined', blocked_until = NOW() - INTERVAL '1 hour', health_reason = 'bounces' WHERE email_account_id = $1`, f.sender); err != nil { + t.Fatal(err) + } + assertCloudStatus(t, f, models.WarmupHealthQuarantined, "complaints") + + // Nor may a fresh one that ends before the ledger's: the status shows the end the gate enforces. + earlier := time.Now().Add(time.Hour) + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET blocked_until = $2, standing_observed_at = NOW() WHERE email_account_id = $1`, f.sender, earlier); err != nil { + t.Fatal(err) + } + assertCloudStatus(t, f, models.WarmupHealthQuarantined, "complaints") + if got, _ := f.warmup.GetCloudStanding(ctx, f.sender); !sameInstant(got.BlockedUntil, &until) { + t.Fatalf("status ends %v, want the ledger's %v", got.BlockedUntil, until) + } +} + +// assertCloudStatus checks what the drawer and the mailbox list show for the sender. +func assertCloudStatus(t *testing.T, f *poolLinkFixture, want models.WarmupHealthState, reason string) { + t.Helper() + ctx := context.Background() + one, err := f.warmup.GetCloudStanding(ctx, f.sender) + if err != nil || one == nil || one.State != string(want) || one.Reason != reason || one.Source != models.WarmupHealthSourceCloud { + t.Fatalf("GetCloudStanding = %+v, %v; want %s (%s) from cloud", one, err, want, reason) + } + many, err := f.warmup.GetCloudStandings(ctx, []uuid.UUID{f.sender}) + if got := many[f.sender]; err != nil || got == nil || got.State != string(want) || got.Reason != reason { + t.Fatalf("GetCloudStandings = %+v, %v; want %s (%s)", got, err, want, reason) + } + links := NewCloudLinkRepository(f.pool, nil) + m, err := links.GetByAccount(ctx, f.sender) + if err != nil || m == nil { + t.Fatalf("GetByAccount = %+v, %v", m, err) + } + listed, err := links.(CloudStatusReader).CloudStatuses(ctx, []models.CloudLinkMailbox{*m}) + if got := listed[f.sender]; err != nil || got == nil || got.State != string(want) || got.Reason != reason { + t.Fatalf("Settings status = %+v, %v; want %s (%s)", got, err, want, reason) + } +} + +// A live Cloud hold whose report went stale is still real evidence: it is not +// unknown, so the lifecycle and risk keep acting on it. +func TestLiveCloudStaleLiveHoldStaysKnown(t *testing.T) { + for _, state := range []models.WarmupHealthState{models.WarmupHealthQuarantined, models.WarmupHealthBlocked} { + t.Run(string(state), func(t *testing.T) { + f := newPoolLinkFixture(t) + ctx := context.Background() + links := NewCloudLinkRepository(f.pool, nil) + handle, _ := liveContactDB(t) + if _, err := links.Enroll(ctx, f.sender, f.sender, liveCloudLink(t, f), false); err != nil { + t.Fatal(err) + } + hold := &models.WarmupHealthInfo{State: string(state), Reason: "complaints"} + if state == models.WarmupHealthQuarantined { + until := time.Now().Add(72 * time.Hour) + hold.BlockedUntil = &until + } + if _, err := links.SetStanding(ctx, f.sender, hold, false); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET standing_observed_at = NOW() - INTERVAL '20 minutes' WHERE email_account_id = $1`, f.sender); err != nil { + t.Fatal(err) + } + assertKnownHold(t, f, handle, state) + assertCloudStatus(t, f, state, "complaints") + // The report is stale, so the hold must not end at its date while it stays stale. + if gs, until, err := f.warmup.GetHealthState(ctx, f.sender); err != nil || gs != state || until != nil { + t.Fatalf("send gate = %s until %v, %v; want %s with no end", gs, until, err, state) + } + if states, err := f.warmup.GetHealthStates(ctx, []uuid.UUID{f.sender}); err != nil || !states[f.sender].CloudReportMissing || states[f.sender].BlockedUntil != nil { + t.Fatalf("send gate read = %+v, %v; want it open-ended on the stale report", states[f.sender], err) + } + }) + } +} + +// assertKnownHold checks that lifecycle, risk and the send gate read a real hold, not unknown evidence. +func assertKnownHold(t *testing.T, f *poolLinkFixture, handle *db.DB, want models.WarmupHealthState) { + t.Helper() + ctx := context.Background() + cand, err := NewSendLifecycleRepository(handle).GetLifecycleCandidate(ctx, f.sender) + if err != nil || cand.HealthState != want || cand.CloudEvidenceUnknown { + t.Fatalf("lifecycle candidate = %+v, %v; want %s with known evidence", cand, err, want) + } + if state, unknown, err := f.warmup.GetKnownHealthState(ctx, f.sender); err != nil || state != want || unknown { + t.Fatalf("known standing = %s, %v, %v; want %s with known evidence", state, unknown, err, want) + } + risk, err := NewWorkerRepository(f.pool).ListRiskCandidates(ctx, 100000) + if err != nil { + t.Fatal(err) + } + found := false + for _, c := range risk { + if c.EmailAccountID == f.sender { + found = c.HealthState == want && !c.CloudEvidenceUnknown + } + } + if !found { + t.Fatalf("risk candidate does not read %s with known evidence", want) + } + if states, err := f.warmup.GetHealthStates(ctx, []uuid.UUID{f.sender}); err != nil || states[f.sender].State != want || states[f.sender].CloudStatusOnly { + t.Fatalf("send gate read = %+v, %v; want %s as a real hold", states[f.sender], err, want) + } +} + +// An enrollment that is not active, or a link being disconnected, is unknown +// evidence for everything that acts on reputation, unless Cloud holds the mailbox. +func TestLiveCloudPendingEnrollmentAndDisconnectAreUnknown(t *testing.T) { + f := newPoolLinkFixture(t) + ctx := context.Background() + links := NewCloudLinkRepository(f.pool, nil) + handle, _ := liveContactDB(t) + instance := liveCloudLink(t, f) + lifecycle := NewSendLifecycleRepository(handle) + assertUnknown := func(label string) { + t.Helper() + cand, err := lifecycle.GetLifecycleCandidate(ctx, f.sender) + if err != nil || cand.HealthState != "" || !cand.CloudEvidenceUnknown { + t.Fatalf("%s: lifecycle candidate = %+v, %v; want unknown with no state", label, cand, err) + } + if state, unknown, err := f.warmup.GetKnownHealthState(ctx, f.sender); err != nil || state != models.WarmupHealthHealthy || !unknown { + t.Fatalf("%s: known standing = %s, %v, %v; want unknown", label, state, unknown, err) + } + if states, err := f.warmup.GetHealthStates(ctx, []uuid.UUID{f.sender}); err != nil || states[f.sender].State != models.WarmupHealthBlocked || !states[f.sender].CloudStatusOnly { + t.Fatalf("%s: send gate read = %+v, %v; want held on the missing report", label, states[f.sender], err) + } + } + if _, err := links.Enroll(ctx, f.sender, f.sender, instance, false); err != nil { + t.Fatal(err) + } + if _, err := links.SetStanding(ctx, f.sender, &models.WarmupHealthInfo{State: "healthy"}, false); err != nil { + t.Fatal(err) + } + for _, state := range []string{"pending_enroll", "pending_remove"} { + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET enrollment_state = $2, standing_observed_at = NOW() WHERE email_account_id = $1`, f.sender, state); err != nil { + t.Fatal(err) + } + assertUnknown(state) + } + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET enrollment_state = 'active', standing_observed_at = NOW() WHERE email_account_id = $1`, f.sender); err != nil { + t.Fatal(err) + } + if err := links.SetDisconnectPending(ctx, instance); err != nil { + t.Fatal(err) + } + assertUnknown("disconnect pending") + + until := time.Now().Add(48 * time.Hour) + if _, err := f.pool.Exec(ctx, `UPDATE cloud_link_mailboxes SET health_state = 'quarantined', blocked_until = $2, health_reason = 'complaints' WHERE email_account_id = $1`, f.sender, until); err != nil { + t.Fatal(err) + } + assertKnownHold(t, f, handle, models.WarmupHealthQuarantined) } func TestLiveCloudPositiveObservationCannotClearAHoldFromAnotherLink(t *testing.T) { diff --git a/internal/repository/cloud_status_test.go b/internal/repository/cloud_status_test.go new file mode 100644 index 000000000..4fa3941ed --- /dev/null +++ b/internal/repository/cloud_status_test.go @@ -0,0 +1,44 @@ +package repository + +import ( + "testing" + "time" + + "github.com/warmbly/warmbly/internal/models" +) + +// Cloud's own report stands only when it is the live hold the gate enforces; +// any other live hold is shown, with its own end. +func TestCloudStatusShowsTheHoldTheGateEnforces(t *testing.T) { + now := time.Now() + later, earlier, past := now.Add(48*time.Hour), now.Add(time.Hour), now.Add(-time.Hour) + quarantined := string(models.WarmupHealthQuarantined) + reason := "complaints" + known := cloudStatusRow{state: &quarantined, until: &later, reason: &reason} + for _, tc := range []struct { + name string + report *models.WarmupHealthInfo + unknown bool + want *time.Time + cloud bool + }{ + {"same live row", &models.WarmupHealthInfo{State: quarantined, BlockedUntil: &later, PoolType: "premium"}, false, &later, true}, + {"earlier end", &models.WarmupHealthInfo{State: quarantined, BlockedUntil: &earlier}, false, &later, false}, + {"lapsed end", &models.WarmupHealthInfo{State: quarantined, BlockedUntil: &past}, false, &later, false}, + {"milder report", &models.WarmupHealthInfo{State: "healthy"}, false, &later, false}, + {"unknown evidence", &models.WarmupHealthInfo{State: quarantined, BlockedUntil: &later}, true, &later, false}, + } { + t.Run(tc.name, func(t *testing.T) { + c := known + c.unknown = tc.unknown + m := &models.CloudLinkMailbox{Standing: tc.report} + got := c.status(m, now) + if (got == tc.report) != tc.cloud || got.State != quarantined || got.BlockedUntil == nil || !got.BlockedUntil.Equal(*tc.want) { + t.Fatalf("status = %+v, want quarantined until %v (Cloud's report: %v)", got, tc.want, tc.cloud) + } + }) + } + if got := (cloudStatusRow{unknown: true}).status(&models.CloudLinkMailbox{}, now); !got.CloudEvidenceUnavailable() { + t.Fatalf("unknown with no hold = %+v, want unavailable", got) + } +} diff --git a/internal/repository/pg_advisor_snapshot.go b/internal/repository/pg_advisor_snapshot.go index 41122abd1..db2784ed7 100644 --- a/internal/repository/pg_advisor_snapshot.go +++ b/internal/repository/pg_advisor_snapshot.go @@ -90,8 +90,9 @@ func (r *advisorRepository) loadMailboxes(ctx context.Context, orgID uuid.UUID, COALESCE(dl.bounces, 0), COALESCE(dl.complaints, 0), COALESCE(w.sent7, 0), COALESCE(w.recv7, 0), ws.major, ws.major_spam, ws.other, ws.other_spam, - COALESCE(p.health_state, ''), COALESCE(p.last_health_score, 0), COALESCE(p.last_health_reason, ''), - COALESCE(p.blocked_until > NOW(), false) AS pool_blocked, + CASE WHEN cstatus.cloud_only THEN 'blocked' ELSE COALESCE(p.health_state, '') END, COALESCE(p.last_health_score, 0), + CASE WHEN cstatus.cloud_only THEN '` + models.WarmupReasonCloudEvidenceUnavailable + `' ELSE COALESCE(p.last_health_reason, '') END, + (NOT cstatus.cloud_only AND COALESCE(p.blocked_until > NOW(), false)) AS pool_blocked, COALESCE(err.n, 0), COALESCE(camp.active, false), ea.send_lifecycle, @@ -130,8 +131,10 @@ func (r *advisorRepository) loadMailboxes(ctx context.Context, orgID uuid.UUID, ) w ON true LEFT JOIN LATERAL (` + placementEvidenceSQL("ea.id", "$2") + ` ) ws ON true - LEFT JOIN LATERAL (` + warmupStandingSQL("ea.id") + ` + LEFT JOIN LATERAL (` + warmupKnownStandingSQL("ea.id") + ` ) p ON true + -- Unknown Cloud evidence reads as unavailable unless a real hold stands beside it. + CROSS JOIN LATERAL (SELECT (` + cloudStandingUnknownSQL("ea.id") + ` AND NOT ` + liveHoldSQL("p") + `) AS cloud_only) cstatus LEFT JOIN LATERAL ( SELECT COUNT(*) AS n FROM email_account_errors eae diff --git a/internal/repository/pg_analytics.go b/internal/repository/pg_analytics.go index 3d4e63b95..5a8168902 100644 --- a/internal/repository/pg_analytics.go +++ b/internal/repository/pg_analytics.go @@ -1063,6 +1063,7 @@ func (r *analyticsRepository) GetAccountHealthSummary(ctx context.Context, orgID AND eae.resolved_at IS NULL AND eae.severity = 'CRITICAL' ) THEN 'error' WHEN wh.health_state = 'watch' + OR ` + cloudStandingUnknownSQL("ea.id") + ` OR EXISTS ( SELECT 1 FROM email_account_errors eae WHERE eae.email_account_id = ea.id @@ -1071,7 +1072,7 @@ func (r *analyticsRepository) GetAccountHealthSummary(ctx context.Context, orgID ELSE 'healthy' END AS status FROM email_accounts ea - LEFT JOIN LATERAL (` + warmupStandingSQL("ea.id") + ` + LEFT JOIN LATERAL (` + warmupKnownStandingSQL("ea.id") + ` ) wh ON true WHERE ea.organization_id = $1` + inMailboxes("ea.id", mailboxes, 2) + ` ) diff --git a/internal/repository/pg_cloudlink.go b/internal/repository/pg_cloudlink.go index c9b5bbac0..6857fb366 100644 --- a/internal/repository/pg_cloudlink.go +++ b/internal/repository/pg_cloudlink.go @@ -59,6 +59,19 @@ type cloudLinkRepository struct { var errNoLinkEncrypter = errors.New("credential encrypter not configured (set CREDENTIALS_ENCRYPTION_KEY)") +// CloudStatusReader reads what each Cloud-warmed mailbox shows, by the rule GetCloudStandings uses. +type CloudStatusReader interface { + CloudStatuses(ctx context.Context, mailboxes []models.CloudLinkMailbox) (map[uuid.UUID]*models.WarmupHealthInfo, error) +} + +func (r *cloudLinkRepository) CloudStatuses(ctx context.Context, mailboxes []models.CloudLinkMailbox) (map[uuid.UUID]*models.WarmupHealthInfo, error) { + linked := make(map[uuid.UUID]*models.CloudLinkMailbox, len(mailboxes)) + for i := range mailboxes { + linked[mailboxes[i].EmailAccountID] = &mailboxes[i] + } + return cloudStatuses(ctx, r.db, linked) +} + // NewCloudLinkRepository seals the instance token with the mailbox credential key. func NewCloudLinkRepository(db *pgxpool.Pool, enc *encrypt.Encrypter) CloudLinkRepository { return &cloudLinkRepository{db: db, encrypt: enc} diff --git a/internal/repository/pg_send_lifecycle.go b/internal/repository/pg_send_lifecycle.go index 1f4b4b441..53dccab56 100644 --- a/internal/repository/pg_send_lifecycle.go +++ b/internal/repository/pg_send_lifecycle.go @@ -39,8 +39,10 @@ type LifecycleCandidate struct { Current models.SendLifecycle Since *time.Time // HealthState is empty when the mailbox is in no warmup pool, meaning no - // signal rather than a healthy one. + // signal rather than a healthy one. It is read from real evidence only. HealthState models.WarmupHealthState + // CloudEvidenceUnknown is set while Warmbly Cloud has no current report for the mailbox. + CloudEvidenceUnknown bool } type sendLifecycleRepository struct { @@ -108,9 +110,9 @@ func (r *sendLifecycleRepository) ListLifecycleCandidates(ctx context.Context, l // clean bill that would auto-resume it. rows, err := r.DB.Pool.Query(ctx, ` SELECT ea.id, ea.send_lifecycle, ea.send_lifecycle_since, - wh.health_state + wh.health_state, `+cloudStandingUnknownSQL("ea.id")+` FROM email_accounts ea - LEFT JOIN LATERAL (`+warmupStandingSQL("ea.id")+` + LEFT JOIN LATERAL (`+warmupKnownStandingSQL("ea.id")+` ) wh ON true WHERE ea.status = 'active' AND ea.send_lifecycle IN ('active', 'resting') @@ -127,7 +129,7 @@ func (r *sendLifecycleRepository) ListLifecycleCandidates(ctx context.Context, l var c LifecycleCandidate var state string var health *string - if err := rows.Scan(&c.EmailAccountID, &state, &c.Since, &health); err != nil { + if err := rows.Scan(&c.EmailAccountID, &state, &c.Since, &health, &c.CloudEvidenceUnknown); err != nil { return nil, err } c.Current = models.SendLifecycle(state) @@ -144,12 +146,13 @@ func (r *sendLifecycleRepository) GetLifecycleCandidate(ctx context.Context, acc var state string var health *string err := r.DB.Pool.QueryRow(ctx, ` - SELECT ea.id, ea.send_lifecycle, ea.send_lifecycle_since, wh.health_state + SELECT ea.id, ea.send_lifecycle, ea.send_lifecycle_since, wh.health_state, + `+cloudStandingUnknownSQL("ea.id")+` FROM email_accounts ea - LEFT JOIN LATERAL (`+warmupStandingSQL("ea.id")+` + LEFT JOIN LATERAL (`+warmupKnownStandingSQL("ea.id")+` ) wh ON true WHERE ea.id = $1 - `, accountID).Scan(&c.EmailAccountID, &state, &c.Since, &health) + `, accountID).Scan(&c.EmailAccountID, &state, &c.Since, &health, &c.CloudEvidenceUnknown) if err != nil { return nil, err } diff --git a/internal/repository/pg_warmup.go b/internal/repository/pg_warmup.go index 12ef3a304..083938ed3 100644 --- a/internal/repository/pg_warmup.go +++ b/internal/repository/pg_warmup.go @@ -156,6 +156,8 @@ type WarmupRepository interface { // GetParticipantHealthForAccounts is the batched form: one pool row per // mailbox in the set, keyed by account id. GetParticipantHealthForAccounts(ctx context.Context, accountIDs []uuid.UUID) (map[uuid.UUID]*models.WarmupParticipantHealth, error) + // GetKnownHealthState is GetHealthState over real evidence, plus whether Cloud's evidence is unknown. + GetKnownHealthState(ctx context.Context, accountID uuid.UUID) (models.WarmupHealthState, bool, error) // GetCloudStanding is the standing Warmbly Cloud last reported for a // mailbox it warms; nil for any other mailbox. GetCloudStanding(ctx context.Context, accountID uuid.UUID) (*models.WarmupHealthInfo, error) @@ -525,6 +527,10 @@ func (r *warmupRepository) BlockFromPool(ctx context.Context, accountID uuid.UUI type WarmupHealthRead struct { State models.WarmupHealthState BlockedUntil *time.Time + // CloudStatusOnly: the hold is only a missing Warmbly Cloud report, with no real hold beside it. + CloudStatusOnly bool + // CloudReportMissing: Warmbly Cloud has no current report, so a hold has no end whatever BlockedUntil says. + CloudReportMissing bool } // IsPoolEligible checks current role, authority and health independently of price. @@ -551,9 +557,9 @@ func (r *warmupRepository) GetHealthStates(ctx context.Context, accountIDs []uui return out, nil } query := ` - SELECT a.id, h.health_state, h.blocked_until + SELECT a.id, h.health_state, h.blocked_until, h.cloud_only, h.cloud_missing FROM unnest($1::uuid[]) AS a(id) - CROSS JOIN LATERAL (` + warmupStandingSQL("a.id") + ` + CROSS JOIN LATERAL (` + warmupGateStandingSQL("a.id") + ` ) h ` rows, err := resultDB(ctx, r.db).Query(ctx, query, accountIDs) @@ -565,10 +571,11 @@ func (r *warmupRepository) GetHealthStates(ctx context.Context, accountIDs []uui var id uuid.UUID var state string var until *time.Time - if err := rows.Scan(&id, &state, &until); err != nil { + var cloudOnly, cloudMissing bool + if err := rows.Scan(&id, &state, &until, &cloudOnly, &cloudMissing); err != nil { return nil, err } - out[id] = WarmupHealthRead{State: models.WarmupHealthState(state), BlockedUntil: until} + out[id] = WarmupHealthRead{State: models.WarmupHealthState(state), BlockedUntil: until, CloudStatusOnly: cloudOnly, CloudReportMissing: cloudMissing} } return out, rows.Err() } @@ -576,7 +583,7 @@ func (r *warmupRepository) GetHealthStates(ctx context.Context, accountIDs []uui // GetHealthState returns the account's warmup health state and blocked_until without the // caller naming a pool. The ordering keeps the worst state winning. func (r *warmupRepository) GetHealthState(ctx context.Context, accountID uuid.UUID) (models.WarmupHealthState, *time.Time, error) { - query := `SELECT health_state, blocked_until FROM (` + warmupStandingSQL("$1::uuid") + ` + query := `SELECT health_state, blocked_until FROM (` + warmupGateStandingSQL("$1::uuid") + ` ) h` var state string var blockedUntil *time.Time @@ -590,6 +597,19 @@ func (r *warmupRepository) GetHealthState(ctx context.Context, accountID uuid.UU return models.WarmupHealthState(state), blockedUntil, nil } +func (r *warmupRepository) GetKnownHealthState(ctx context.Context, accountID uuid.UUID) (models.WarmupHealthState, bool, error) { + var state *string + var unknown bool + err := resultDB(ctx, r.db).QueryRow(ctx, `SELECT (SELECT k.health_state FROM (`+warmupKnownStandingSQL("$1::uuid")+`) k), `+cloudStandingUnknownSQL("$1::uuid"), accountID).Scan(&state, &unknown) + if err != nil { + return models.WarmupHealthHealthy, false, err + } + if state == nil { + return models.WarmupHealthHealthy, unknown, nil + } + return models.WarmupHealthState(*state), unknown, nil +} + // UnblockFromPool unblocks an account from all warmup pools func (r *warmupRepository) UnblockFromPool(ctx context.Context, accountID uuid.UUID) error { query := ` @@ -694,15 +714,12 @@ func (r *warmupRepository) GetCloudStanding(ctx context.Context, accountID uuid. if err != nil { return nil, err } - var state, reason string - var until *time.Time - if err := resultDB(ctx, r.db).QueryRow(ctx, `SELECT health_state, blocked_until, COALESCE(last_health_reason, '') FROM (`+warmupStandingSQL("$1")+`) standing`, accountID).Scan(&state, &until, &reason); err != nil { + var c cloudStatusRow + if err := resultDB(ctx, r.db).QueryRow(ctx, `SELECT `+cloudStatusSelect("$1::uuid")+` + FROM (SELECT 1) one LEFT JOIN LATERAL (`+warmupKnownStandingSQL("$1::uuid")+`) k ON true`, accountID).Scan(c.dest()...); err != nil { return nil, err } - if state == string(models.WarmupHealthBlocked) && reason == "cloud_evidence_unavailable" { - return &models.WarmupHealthInfo{State: state, Reason: reason, BlockedUntil: until}, nil - } - return m.Standing, nil + return c.status(m, time.Now()), nil } // GetParticipantHealthForAccount returns the participant row from whichever pool the mailbox diff --git a/internal/repository/pg_warmup_status_batch.go b/internal/repository/pg_warmup_status_batch.go index e2da87f83..f61d3647e 100644 --- a/internal/repository/pg_warmup_status_batch.go +++ b/internal/repository/pg_warmup_status_batch.go @@ -100,33 +100,7 @@ func (r *warmupRepository) GetCloudStandings(ctx context.Context, accountIDs []u if err := rows.Err(); err != nil || len(linked) == 0 { return out, err } - ids := make([]uuid.UUID, 0, len(linked)) - for id := range linked { - ids = append(ids, id) - } - rows, err = resultDB(ctx, r.db).Query(ctx, ` - SELECT a.id, h.health_state, h.blocked_until, COALESCE(h.last_health_reason, '') - FROM unnest($1::uuid[]) AS a(id) - CROSS JOIN LATERAL (`+warmupStandingSQL("a.id")+` - ) h`, ids) - if err != nil { - return nil, err - } - defer rows.Close() - for rows.Next() { - var id uuid.UUID - var state, reason string - var until *time.Time - if err := rows.Scan(&id, &state, &until, &reason); err != nil { - return nil, err - } - if state == string(models.WarmupHealthBlocked) && reason == "cloud_evidence_unavailable" { - out[id] = &models.WarmupHealthInfo{State: state, Reason: reason, BlockedUntil: until} - } else if standing := linked[id].Standing; standing != nil { - out[id] = standing - } - } - return out, rows.Err() + return cloudStatuses(ctx, resultDB(ctx, r.db), linked) } // SpamPlacementsSinceForAccounts is SpamPlacementsSince over a set; every diff --git a/internal/repository/pg_worker_risk.go b/internal/repository/pg_worker_risk.go index 765f8ef0b..1a9da7598 100644 --- a/internal/repository/pg_worker_risk.go +++ b/internal/repository/pg_worker_risk.go @@ -48,9 +48,11 @@ type RiskCandidate struct { UserID uuid.UUID OrgID *uuid.UUID CurrentBand models.EmailRiskBand - HealthState models.WarmupHealthState // may be empty if no health row - WorkerID *uuid.UUID - LastEvaluated *time.Time + HealthState models.WarmupHealthState // may be empty if no health row; real evidence only + // CloudEvidenceUnknown is set while Warmbly Cloud has no current report for the mailbox. + CloudEvidenceUnknown bool + WorkerID *uuid.UUID + LastEvaluated *time.Time } // ListRiskCandidates joins email_accounts with warmup health so the risk sweep @@ -71,10 +73,11 @@ func (r *workerRepository) ListRiskCandidates(ctx context.Context, limit int) ([ ea.organization_id, ea.risk_band, COALESCE(wh.health_state, '')::text, + `+cloudStandingUnknownSQL("ea.id")+`, ea.worker_id, ea.risk_evaluated_at FROM email_accounts ea - LEFT JOIN LATERAL (`+warmupStandingSQL("ea.id")+` + LEFT JOIN LATERAL (`+warmupKnownStandingSQL("ea.id")+` ) wh ON true WHERE ea.status = 'active' LIMIT $1 @@ -94,6 +97,7 @@ func (r *workerRepository) ListRiskCandidates(ctx context.Context, limit int) ([ &c.OrgID, &c.CurrentBand, &hs, + &c.CloudEvidenceUnknown, &c.WorkerID, &c.LastEvaluated, ); err != nil { diff --git a/internal/repository/warmup_standing_sql.go b/internal/repository/warmup_standing_sql.go index 29260ba4f..0e41238b9 100644 --- a/internal/repository/warmup_standing_sql.go +++ b/internal/repository/warmup_standing_sql.go @@ -1,5 +1,15 @@ package repository +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + + "github.com/warmbly/warmbly/internal/models" +) + // warmupStandingRankSQL orders health states worst first when sorted DESC. func warmupStandingRankSQL(col string) string { return `CASE ` + col + ` @@ -12,19 +22,149 @@ func warmupStandingRankSQL(col string) string { END` } +// cloudReportMissingPredicate: the clm row has no current, recognized standing. +const cloudReportMissingPredicate = `(clm.enrollment_state <> 'active' OR EXISTS (SELECT 1 FROM cloud_link WHERE instance_id = clm.instance_id AND disconnect_pending) + OR clm.health_state IS NULL OR clm.health_state NOT IN ('healthy', 'watch', 'throttled', 'quarantined', 'blocked') OR clm.standing_observed_at IS NULL + OR clm.standing_observed_at <= NOW() - INTERVAL '15 minutes' OR clm.standing_observed_at > NOW())` + +// cloudStandingUnknownPredicate: the clm row has no current, recognized standing and no live hold. +const cloudStandingUnknownPredicate = `(` + cloudReportMissingPredicate + ` + AND NOT (COALESCE(clm.health_state IN ('quarantined', 'blocked'), false) + AND (clm.blocked_until IS NULL OR clm.blocked_until > NOW())))` + // warmupStandingSQL is a mailbox's worst warmup standing, at most one row: // its local pool row, or the standing Warmbly Cloud last reported for a // mailbox it warms (which has no local pool row). A standing past its term -// ranks below any live one, so it cannot mask one. Every read that gates or -// shows warmup health goes through it, so a cloud verdict holds here too. +// ranks below any live one, so it cannot mask one. Unknown cloud evidence +// reads as blocked, so the gates that use this fail closed. // Columns: health_state, blocked_until, last_health_score, // last_health_reason, last_health_evaluated_at. func warmupStandingSQL(accountExpr string) string { - unknown := `(clm.enrollment_state <> 'active' OR EXISTS (SELECT 1 FROM cloud_link WHERE instance_id = clm.instance_id AND disconnect_pending) - OR clm.health_state IS NULL OR clm.health_state NOT IN ('healthy', 'watch', 'throttled', 'quarantined', 'blocked') OR clm.standing_observed_at IS NULL - OR clm.standing_observed_at <= NOW() - INTERVAL '15 minutes' OR clm.standing_observed_at > NOW()) - AND NOT (COALESCE(clm.health_state IN ('quarantined', 'blocked'), false) - AND (clm.blocked_until IS NULL OR clm.blocked_until > NOW()))` + return warmupStandingUnionSQL(accountExpr, false) +} + +// warmupKnownStandingSQL is warmupStandingSQL without the row unknown cloud evidence synthesises. +func warmupKnownStandingSQL(accountExpr string) string { + return warmupStandingUnionSQL(accountExpr, true) +} + +// cloudReportMissingSQL is a boolean: Warmbly Cloud has no current report for the mailbox, held or not. +func cloudReportMissingSQL(accountExpr string) string { + return `EXISTS (SELECT 1 FROM cloud_link_mailboxes clm WHERE clm.email_account_id = ` + accountExpr + ` AND ` + cloudReportMissingPredicate + `)` +} + +// cloudStandingUnknownSQL is a boolean: the mailbox's Warmbly Cloud evidence is unknown. +func cloudStandingUnknownSQL(accountExpr string) string { + return `EXISTS (SELECT 1 FROM cloud_link_mailboxes clm WHERE clm.email_account_id = ` + accountExpr + ` AND ` + cloudStandingUnknownPredicate + `)` +} + +// liveHoldSQL is a boolean: the standing row aliased alias is a quarantine or block still in term. +func liveHoldSQL(alias string) string { + return `COALESCE(` + alias + `.health_state IN ('quarantined', 'blocked') AND (` + alias + `.blocked_until IS NULL OR ` + alias + `.blocked_until > NOW()), false)` +} + +// warmupGateStandingSQL is the send gate's standing: a live real hold's state, else warmupStandingSQL's row. +// Columns: health_state, blocked_until, cloud_only (held only by a missing Cloud report), cloud_missing. +func warmupGateStandingSQL(accountExpr string) string { + // A missing Cloud report has no end, so any hold beside it is open-ended at the gate. + return ` + SELECT CASE WHEN ` + liveHoldSQL("k") + ` THEN k.health_state ELSE h.health_state END AS health_state, + CASE WHEN u.missing THEN NULL WHEN ` + liveHoldSQL("k") + ` THEN k.blocked_until ELSE h.blocked_until END AS blocked_until, + (u.unknown AND NOT ` + liveHoldSQL("k") + `) AS cloud_only, + u.missing AS cloud_missing + FROM (` + warmupStandingSQL(accountExpr) + `) h + LEFT JOIN LATERAL (` + warmupKnownStandingSQL(accountExpr) + `) k ON true + CROSS JOIN LATERAL (SELECT ` + cloudStandingUnknownSQL(accountExpr) + ` AS unknown, ` + cloudReportMissingSQL(accountExpr) + ` AS missing) u` +} + +// cloudStatusSelect is the columns cloudStatusRow scans, over a warmupKnownStandingSQL lateral aliased k. +func cloudStatusSelect(accountExpr string) string { + return cloudStandingUnknownSQL(accountExpr) + `, k.health_state, k.blocked_until, k.last_health_score, k.last_health_reason, k.last_health_evaluated_at` +} + +type cloudStatusRow struct { + unknown bool + state *string + until *time.Time + score *float64 + reason *string + evaluatedAt *time.Time +} + +func (c *cloudStatusRow) dest() []any { + return []any{&c.unknown, &c.state, &c.until, &c.score, &c.reason, &c.evaluatedAt} +} + +// status is what a Cloud-warmed mailbox shows: a live hold first, then unknown evidence, then Cloud's report. +func (c cloudStatusRow) status(m *models.CloudLinkMailbox, now time.Time) *models.WarmupHealthInfo { + hold := c.state != nil && models.WarmupHealthState(*c.state).Holds() && (c.until == nil || c.until.After(now)) + // Cloud's own report stands only when it is that same live row; it carries the pool and partner detail. + same := !c.unknown && m.Standing != nil && hold && m.Standing.State == *c.state && sameMicrosecond(m.Standing.BlockedUntil, c.until) + if hold && !same { + info := &models.WarmupHealthInfo{Source: models.WarmupHealthSourceCloud, State: *c.state, BlockedUntil: c.until, EvaluatedAt: c.evaluatedAt} + if c.score != nil { + info.Score = *c.score + } + if c.reason != nil { + info.Reason = *c.reason + } + return info + } + if c.unknown { + return models.CloudStandingUnavailable() + } + return m.Standing +} + +// sameMicrosecond compares two optional times at the database's microsecond precision. +func sameMicrosecond(a, b *time.Time) bool { + if a == nil || b == nil { + return a == b + } + return a.Truncate(time.Microsecond).Equal(b.Truncate(time.Microsecond)) +} + +// cloudStatuses is what each Cloud-warmed mailbox in linked shows; absent when there is nothing to show. +func cloudStatuses(ctx context.Context, q interface { + Query(context.Context, string, ...any) (pgx.Rows, error) +}, linked map[uuid.UUID]*models.CloudLinkMailbox) (map[uuid.UUID]*models.WarmupHealthInfo, error) { + out := make(map[uuid.UUID]*models.WarmupHealthInfo, len(linked)) + if len(linked) == 0 { + return out, nil + } + ids := make([]uuid.UUID, 0, len(linked)) + for id := range linked { + ids = append(ids, id) + } + rows, err := q.Query(ctx, ` + SELECT a.id, `+cloudStatusSelect("a.id")+` + FROM unnest($1::uuid[]) AS a(id) + LEFT JOIN LATERAL (`+warmupKnownStandingSQL("a.id")+` + ) k ON true`, ids) + if err != nil { + return nil, err + } + defer rows.Close() + now := time.Now() + for rows.Next() { + var id uuid.UUID + var c cloudStatusRow + if err := rows.Scan(append([]any{&id}, c.dest()...)...); err != nil { + return nil, err + } + if st := c.status(linked[id], now); st != nil { + out[id] = st + } + } + return out, rows.Err() +} + +func warmupStandingUnionSQL(accountExpr string, knownOnly bool) string { + unknown := cloudStandingUnknownPredicate + cloudFilter := "" + if knownOnly { + cloudFilter = ` AND NOT COALESCE(` + unknown + `, false)` + } return ` SELECT st.health_state, st.blocked_until, st.last_health_score, st.last_health_reason, st.last_health_evaluated_at @@ -36,10 +176,10 @@ func warmupStandingSQL(accountExpr string) string { UNION ALL SELECT CASE WHEN ` + unknown + ` THEN 'blocked' ELSE clm.health_state END, CASE WHEN ` + unknown + ` THEN NULL ELSE clm.blocked_until END, - clm.health_score, CASE WHEN ` + unknown + ` THEN 'cloud_evidence_unavailable' ELSE clm.health_reason END, + clm.health_score, CASE WHEN ` + unknown + ` THEN '` + models.WarmupReasonCloudEvidenceUnavailable + `' ELSE clm.health_reason END, clm.health_evaluated_at FROM cloud_link_mailboxes clm - WHERE clm.email_account_id = ` + accountExpr + ` + WHERE clm.email_account_id = ` + accountExpr + cloudFilter + ` UNION ALL SELECT l.cloud_health_state, l.cloud_blocked_until, l.cloud_health_score, l.cloud_health_reason, NULL::timestamptz diff --git a/internal/scheduler/campaign_projection.go b/internal/scheduler/campaign_projection.go index a876d25fc..39bfa33f9 100644 --- a/internal/scheduler/campaign_projection.go +++ b/internal/scheduler/campaign_projection.go @@ -356,7 +356,9 @@ func (s *schedulerService) ProjectCampaign(ctx context.Context, in CampaignProje p.held, p.state = true, models.EstimateSenderResting case gateHealth: p.held, p.state = true, models.EstimateSenderHealthHold - p.heldTill = h.blockedUntil + if !h.cloudMissing { + p.heldTill = h.blockedUntil + } case gateNoWorker: p.state = models.EstimateSenderNoWorker case gateRecovery: diff --git a/internal/scheduler/campaign_sender.go b/internal/scheduler/campaign_sender.go index d9b0aa6dc..9f715cd54 100644 --- a/internal/scheduler/campaign_sender.go +++ b/internal/scheduler/campaign_sender.go @@ -38,6 +38,8 @@ type mailboxGate struct { // reopensAt is when a paced mailbox is expected back; zero when unknown, // which the caller reads as "tomorrow". reopensAt time.Time + // cloudStatus marks a health hold that is only a missing Warmbly Cloud report. + cloudStatus bool } func (g mailboxGate) open() bool { return g.reason == "" } @@ -97,6 +99,9 @@ type healthRead struct { state models.WarmupHealthState blockedUntil *time.Time known bool + cloudStatus bool + // cloudMissing holds a quarantine or block open-ended: a missing Cloud report has no end. + cloudMissing bool } // newCampaignPass resolves the pool-wide state for one scheduling pass. @@ -285,8 +290,8 @@ func (s *schedulerService) gateFor(ctx context.Context, p *campaignPass, acct mo if h := s.healthFor(ctx, p, acct.ID); h.known { switch h.state { case models.WarmupHealthQuarantined, models.WarmupHealthBlocked: - if h.blockedUntil == nil || h.blockedUntil.After(time.Now()) { - return mailboxGate{reason: gateHealth}, 0 + if h.cloudMissing || h.blockedUntil == nil || h.blockedUntil.After(time.Now()) { + return mailboxGate{reason: gateHealth, cloudStatus: h.cloudStatus}, 0 } case models.WarmupHealthWatch, models.WarmupHealthThrottled: remaining = int(float64(remaining) * adjustmentFor(h.state).volumeMultiplier) diff --git a/internal/scheduler/cloud_status_gate_test.go b/internal/scheduler/cloud_status_gate_test.go new file mode 100644 index 000000000..f4923478a --- /dev/null +++ b/internal/scheduler/cloud_status_gate_test.go @@ -0,0 +1,65 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" +) + +// 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) { + cloudOnly, held := uuid.New(), uuid.New() + pass := &campaignPass{health: map[uuid.UUID]healthRead{ + cloudOnly: {state: models.WarmupHealthBlocked, known: true, cloudStatus: true}, + held: {state: models.WarmupHealthBlocked, known: true}, + }} + s := &schedulerService{} + if gate, remaining := s.gateFor(context.Background(), pass, models.Email{ID: cloudOnly}, 10); gate.reason != gateHealth || !gate.cloudStatus || remaining != 0 { + t.Fatalf("missing Cloud report gave %+v with %d left, want a health hold marked as a Cloud status", gate, remaining) + } + if gate, _ := s.gateFor(context.Background(), pass, models.Email{ID: held}, 10); gate.reason != gateHealth || gate.cloudStatus { + t.Fatalf("real block gave %+v, want an unmarked health hold", gate) + } +} + +// The plan's mailbox row names a missing Cloud report as its own state with no +// "blocked" band, and keeps the band for a real hold. +func TestPlanMailboxRowNamesAMissingCloudReport(t *testing.T) { + now := time.Now() + cloudOnly, held := models.Email{ID: uuid.New(), CampaignLimit: 50}, models.Email{ID: uuid.New(), CampaignLimit: 50} + pass := &campaignPass{campaign: &models.Campaign{DailyLimit: 50}, risk: models.OrgRiskTrusted, health: map[uuid.UUID]healthRead{ + cloudOnly.ID: {state: models.WarmupHealthBlocked, known: true, cloudStatus: true}, + held.ID: {state: models.WarmupHealthQuarantined, known: true}, + }} + s := &schedulerService{} + d := s.planMailbox(t.Context(), pass, cloudOnly, 0, 0, nil, now, 10*3600, now.Add(10*time.Hour)) + if d.state != models.MailboxPlanCloudStatus || d.healthBand() != "" { + t.Fatalf("missing Cloud report row = %q band %q, want cloud_status_hold with no band", d.state, d.healthBand()) + } + d = s.planMailbox(t.Context(), pass, held, 0, 0, nil, now, 10*3600, now.Add(10*time.Hour)) + if d.state != models.MailboxPlanHealthHold || d.healthBand() != string(models.WarmupHealthQuarantined) { + t.Fatalf("real hold row = %q band %q, want health_hold quarantined", d.state, d.healthBand()) + } +} + +// A hold beside a missing Cloud report never lapses on the clock: its end can pass +// between the read and the check while the report is still missing. +func TestHealthGateHoldsOpenEndedWhileTheCloudReportIsMissing(t *testing.T) { + ended := time.Now().Add(-time.Minute) + missing, known := uuid.New(), uuid.New() + pass := &campaignPass{health: map[uuid.UUID]healthRead{ + missing: {state: models.WarmupHealthQuarantined, blockedUntil: &ended, known: true, cloudMissing: true}, + known: {state: models.WarmupHealthQuarantined, blockedUntil: &ended, known: true}, + }} + s := &schedulerService{} + if gate, _ := s.gateFor(context.Background(), pass, models.Email{ID: missing}, 10); gate.reason != gateHealth { + t.Fatalf("an ended hold beside a missing Cloud report gave %+v, want it still held", gate) + } + if gate, _ := s.gateFor(context.Background(), pass, models.Email{ID: known}, 10); gate.reason == gateHealth { + t.Fatalf("an ended hold with a current report gave %+v, want the health gate open", gate) + } +} diff --git a/internal/scheduler/send_plan.go b/internal/scheduler/send_plan.go index b094f11fa..590cec821 100644 --- a/internal/scheduler/send_plan.go +++ b/internal/scheduler/send_plan.go @@ -102,6 +102,14 @@ func room(capv, sentThis int) int { return max(0, capv-sentThis) } +// healthBand is the warmup band the row shows; a missing Cloud report is a state, not a band. +func (d mailboxDay) healthBand() string { + if !d.health.known || d.health.cloudStatus || d.health.state == "" || d.health.state == models.WarmupHealthHealthy { + return "" + } + return string(d.health.state) +} + // planMailbox walks one mailbox through the day. windowSecondsLeft is the // campaign's sending time still ahead today; zero means the window is closed // for the rest of the day and the campaign-level clamp reports it instead. @@ -153,6 +161,9 @@ func (s *schedulerService) planMailbox(ctx context.Context, pass *campaignPass, d.state = models.MailboxPlanAdmission default: d.state = models.MailboxPlanHealthHold + if gate.cloudStatus { + d.state = models.MailboxPlanCloudStatus + } } return d } @@ -475,7 +486,7 @@ func (s *schedulerService) PlanCampaignDay(ctx context.Context, campaignID uuid. {kind: models.SendLimitCampaignDailyLimit}, {kind: models.SendLimitCampaignRamp}, {kind: models.SendLimitWarmupGraduation}, {kind: models.SendLimitWorkspaceRisk}, {kind: models.SendLimitDomainAuth}, {kind: models.SendLimitResting}, {kind: models.SendLimitHealthHold}, - {kind: models.SendLimitRecovery}, {kind: models.SendLimitCooldown}, + {kind: models.SendLimitCloudStatusHold}, {kind: models.SendLimitRecovery}, {kind: models.SendLimitCooldown}, {kind: models.SendLimitOtherCampaigns}, {kind: models.SendLimitHealthPace}, {kind: models.SendLimitMailboxHours}, {kind: models.SendLimitSendingBehavior}, {kind: models.SendLimitSpacing}, } @@ -506,7 +517,11 @@ func (s *schedulerService) PlanCampaignDay(ctx context.Context, campaignID uuid. case gateResting: add(models.SendLimitResting, d.byGate) case gateHealth: - add(models.SendLimitHealthHold, d.byGate) + if d.gate.cloudStatus { + add(models.SendLimitCloudStatusHold, d.byGate) + } else { + add(models.SendLimitHealthHold, d.byGate) + } case gateRecovery, gateAdmission: add(models.SendLimitRecovery, d.byGate) case gateCooldown: @@ -528,9 +543,7 @@ func (s *schedulerService) PlanCampaignDay(ctx context.Context, campaignID uuid. t := d.reopensAt mp.ReopensAt = &t } - if d.health.known && d.health.state != "" && d.health.state != models.WarmupHealthHealthy { - mp.Health = string(d.health.state) - } + mp.Health = d.healthBand() plan.Mailboxes = append(plan.Mailboxes, mp) } for _, t := range tallies { @@ -691,7 +704,7 @@ func (s *schedulerService) prefillPool(ctx context.Context, pass *campaignPass, if s.warmupRepo != nil { if states, err := s.warmupRepo.GetHealthStates(ctx, ids); err == nil { for id, h := range states { - pass.health[id] = healthRead{state: h.State, blockedUntil: h.BlockedUntil, known: true} + pass.health[id] = healthRead{state: h.State, blockedUntil: h.BlockedUntil, known: true, cloudStatus: h.CloudStatusOnly, cloudMissing: h.CloudReportMissing} } } } diff --git a/web/src/app/app/emails/page.tsx b/web/src/app/app/emails/page.tsx index a0e567e98..591c94e3b 100644 --- a/web/src/app/app/emails/page.tsx +++ b/web/src/app/app/emails/page.tsx @@ -5,7 +5,7 @@ import { browseString, mailboxSort, mailboxTab, type MailboxTab } from "@/lib/br import { useSearchParams } from "@/hooks/useSearchParams"; import useMailboxTagFilter from "@/hooks/useMailboxTagFilter"; import groupMailboxes from "@/lib/groupMailboxes"; -import { mailboxColdActivity } from "@/lib/mailboxStatus"; +import { cloudStandingUnavailable as cloudHealthUnavailable, mailboxColdActivity } from "@/lib/mailboxStatus"; import { useAppStore } from "@/stores/useAppStore"; import toast from "react-hot-toast/headless"; import { useQueryClient } from "@tanstack/react-query"; @@ -251,7 +251,7 @@ export default function AddressesPage() { next.set(s.id, cur); const before = prev.get(s.id); if (before && (HEALTH_RANK[cur] ?? 0) > (HEALTH_RANK[before] ?? 0)) { - const reason = s.warmup_health?.reason || s.health?.issues?.[0]; + const reason = (!cloudHealthUnavailable(s.warmup_health) && s.warmup_health?.reason) || s.health?.issues?.[0]; toast.error(`${s.email} health dropped to ${cur}${reason ? ` — ${reason}` : ""}`); } } diff --git a/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx b/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx index 87eb150da..a5964786a 100644 --- a/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx +++ b/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx @@ -18,6 +18,7 @@ import { useConfirm } from "@/hooks/context/confirm"; import { TableSurface, Toggle } from "../_components/SectionShell"; import { providerLabel, providerSupported } from "./providers"; import { cloudWarmupPaused } from "@/lib/cloudWarmup"; +import { cloudStandingUnavailable } from "@/lib/mailboxStatus"; export default function MailboxTable() { const rows = useCloudLinkMailboxes(); @@ -164,6 +165,12 @@ export default function MailboxTable() { function HealthPill({ health, paused }: { health: PoolLinkWarmupHealth | null; paused: boolean }) { if (paused) return Paused; if (!health) return Starting; + if (cloudStandingUnavailable(health)) + return ( + + Unavailable + + ); switch (health.state) { case "healthy": return Healthy; diff --git a/web/src/components/app/campaigns/SendPlanCard.tsx b/web/src/components/app/campaigns/SendPlanCard.tsx index 1ded80e67..e1bf04865 100644 --- a/web/src/components/app/campaigns/SendPlanCard.tsx +++ b/web/src/components/app/campaigns/SendPlanCard.tsx @@ -53,6 +53,11 @@ const LIMIT_META: Record = { hint: "A mailbox is quarantined or blocked by its warmup health and sends nothing cold until that lifts.", to: "/app/emails", }, + cloud_status_hold: { + label: "Waiting for Warmbly Cloud status", + hint: "Warmbly Cloud has not reported a current status for a mailbox it warms. It sends nothing cold until one arrives; this is not a block.", + to: "/app/emails", + }, send_recovery: { label: "Send recovery hold", hint: "A mailbox has an unresolved send or provider refusal. Healthy mailboxes remain available; investigate the held send before resolving it.", @@ -122,6 +127,7 @@ const STATE_META: Record domain_auth: { label: "Auth failing", tone: "bg-rose-50 text-rose-700 ring-rose-200" }, resting: { label: "Resting", tone: "bg-slate-100 text-slate-600 ring-slate-200" }, health_hold: { label: "Health hold", tone: "bg-rose-50 text-rose-700 ring-rose-200" }, + cloud_status_hold: { label: "Waiting for Cloud", tone: "bg-amber-50 text-amber-700 ring-amber-200" }, send_recovery: { label: "Send recovery", tone: "bg-rose-50 text-rose-700 ring-rose-200" }, send_cooldown: { label: "Cooling down", tone: "bg-amber-50 text-amber-700 ring-amber-200" }, send_authority: { label: "Send unavailable", tone: "bg-amber-50 text-amber-700 ring-amber-200" }, diff --git a/web/src/components/app/emails/CloudWarmupCard.tsx b/web/src/components/app/emails/CloudWarmupCard.tsx index c3e25edff..4e11dc049 100644 --- a/web/src/components/app/emails/CloudWarmupCard.tsx +++ b/web/src/components/app/emails/CloudWarmupCard.tsx @@ -15,6 +15,7 @@ import { providerSupported } from "@/app/app/settings/warmbly-cloud/providers"; import WarmupPartnerDiversity from "./WarmupPartnerDiversity"; import WarmupSendFailureNote from "./WarmupSendFailureNote"; import { cloudSendFailure, cloudWarmupPaused } from "@/lib/cloudWarmup"; +import { cloudStandingUnavailable } from "@/lib/mailboxStatus"; import { useMutation, useQueryClient } from "@tanstack/react-query"; import { setCloudLinkParticipation } from "@/lib/api/client/app/cloudlink/cloudLink"; import type { DiagnosticParticipation } from "@/lib/api/models/app/cloudlink/CloudLink"; @@ -82,7 +83,7 @@ export default function CloudWarmupCard({ mailboxId, email, provider }: { mailbo const p = cloud?.participation; const paused = cloudWarmupPaused(cloud) || !!p && !p.send; - const health = cloud?.health?.state; + const health = cloudStandingUnavailable(cloud?.health) ? "status unavailable" : cloud?.health?.state; return (
diff --git a/web/src/components/app/emails/InboxDetails.tsx b/web/src/components/app/emails/InboxDetails.tsx index 92c52a709..c3664d61e 100644 --- a/web/src/components/app/emails/InboxDetails.tsx +++ b/web/src/components/app/emails/InboxDetails.tsx @@ -16,6 +16,7 @@ import React, { useMemo, useState } from "react"; import useBrowseState from "@/hooks/useBrowseState"; import { mailboxTab, type MailboxTab } from "@/lib/browse-accounts-analytics"; import { diagnosticSendingAllowed, diagnosticWarmupActive } from "@/lib/diagnosticParticipation"; +import { cloudStandingUnavailable } from "@/lib/mailboxStatus"; import { AnimatePresence, motion } from "framer-motion"; import AdvisorStrip from "@/components/app/advisor/AdvisorStrip"; import { @@ -1250,6 +1251,7 @@ function AuthCheckPanel({ mailbox }: { mailbox: Inbox }) { function WarmupTab({ form, update, status, mailbox, canWarmup = true }: { form: Inbox; update: (p: Partial) => void; status?: AccountStatusModel; mailbox: Inbox; canWarmup?: boolean }) { const ws = status?.warmup_status; const wh = status?.warmup_health; + const whUnavailable = cloudStandingUnavailable(wh); const inCampaign = status?.in_campaign; // Warmup on/off is a lifecycle action (immediate), not a saved form field — @@ -1416,15 +1418,21 @@ function WarmupTab({ form, update, status, mailbox, canWarmup = true }: { form: )}
- - {warmupStateTone[wh.state]?.label ?? wh.state} + + {whUnavailable ? "Status unavailable" : (warmupStateTone[wh.state]?.label ?? wh.state)}
- {wh.reason &&

{wh.reason}

} + {whUnavailable ? ( +

+ Warmbly Cloud has not reported a current status for this mailbox. Campaigns on this instance wait for one before sending from it. This is not a block and does not lower its health. +

+ ) : ( + wh.reason &&

{wh.reason}

+ )} {wh.blocked_until && (

Paused from the pool until {new Date(wh.blocked_until).toLocaleDateString()}.

)} - {wh.source === "cloud" && wh.state !== "healthy" && ( + {wh.source === "cloud" && wh.state !== "healthy" && !whUnavailable && (

{wh.state === "quarantined" || wh.state === "blocked" ? "Campaigns on this instance skip this mailbox while Warmbly Cloud holds it." diff --git a/web/src/components/app/placement/MailboxPlacementTab.tsx b/web/src/components/app/placement/MailboxPlacementTab.tsx index d1914a70c..3f07027e8 100644 --- a/web/src/components/app/placement/MailboxPlacementTab.tsx +++ b/web/src/components/app/placement/MailboxPlacementTab.tsx @@ -11,6 +11,7 @@ import useWarmupPlacement from "@/lib/api/hooks/app/analytics/useWarmupPlacement import type { WarmupHealthInfo } from "@/lib/api/models/app/analytics/AccountStatus"; import type { PlacementGroup } from "@/lib/api/models/app/analytics/WarmupPlacement"; import { cn } from "@/lib/utils"; +import { cloudStandingUnavailable } from "@/lib/mailboxStatus"; import { BAND, GROUP_LABEL, @@ -98,7 +99,11 @@ export default function MailboxPlacementTab({ mailboxId, poolHealth }: { mailbox

- {poolHealth && POOL_STATE_LABEL[poolHealth.state] && ( + {poolHealth && cloudStandingUnavailable(poolHealth) ? ( + + Pool standing: unavailable + + ) : poolHealth && POOL_STATE_LABEL[poolHealth.state] && ( {POOL_STATE_LABEL[poolHealth.state]} diff --git a/web/src/lib/api/models/app/campaigns/SendPlan.ts b/web/src/lib/api/models/app/campaigns/SendPlan.ts index 929839646..366b339cc 100644 --- a/web/src/lib/api/models/app/campaigns/SendPlan.ts +++ b/web/src/lib/api/models/app/campaigns/SendPlan.ts @@ -10,6 +10,7 @@ export type SendLimitKind = | "domain_auth" | "resting" | "warmup_health_hold" + | "cloud_status_hold" | "send_recovery" | "send_cooldown" | "other_campaigns" @@ -33,6 +34,7 @@ export type MailboxPlanState = | "domain_auth" | "resting" | "health_hold" + | "cloud_status_hold" | "send_recovery" | "send_cooldown" | "send_authority" diff --git a/web/src/lib/mailboxStatus.test.ts b/web/src/lib/mailboxStatus.test.ts index e47bc96b1..4d75e04c2 100644 --- a/web/src/lib/mailboxStatus.test.ts +++ b/web/src/lib/mailboxStatus.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { mailboxColdActivity } from "./mailboxStatus"; +import { cloudStandingUnavailable, mailboxColdActivity } from "./mailboxStatus"; describe("cold rotation activity", () => { it.each(["resting", "reserve"] as const)("shows %s without a campaign on this instance", (state) => { @@ -20,3 +20,12 @@ describe("cold rotation activity", () => { expect(mailboxColdActivity({ in_campaign: false, send_lifecycle: { state: "active" } })).toEqual({ resting: false, sending: false }); }); }); + +describe("Cloud standing availability", () => { + it("tells a missing Cloud report from a real block", () => { + expect(cloudStandingUnavailable({ reason: "cloud_evidence_unavailable" })).toBe(true); + expect(cloudStandingUnavailable({ reason: "complaints" })).toBe(false); + expect(cloudStandingUnavailable(null)).toBe(false); + expect(cloudStandingUnavailable(undefined)).toBe(false); + }); +}); diff --git a/web/src/lib/mailboxStatus.ts b/web/src/lib/mailboxStatus.ts index d6af9e46b..e3d6a7038 100644 --- a/web/src/lib/mailboxStatus.ts +++ b/web/src/lib/mailboxStatus.ts @@ -10,6 +10,11 @@ export function mailboxColdActivity(status?: Pick