diff --git a/docs/content/docs/api/reference/analytics.mdx b/docs/content/docs/api/reference/analytics.mdx index 358084cb6..41244db82 100644 --- a/docs/content/docs/api/reference/analytics.mdx +++ b/docs/content/docs/api/reference/analytics.mdx @@ -615,7 +615,8 @@ Returns the detailed status for one email account: a combined health score (fold "target_volume": 33, "max_volume": 40, "reply_rate": 35, - "days_active": 23 + "days_active": 23, + "next_send_at": "2026-06-12T14:37:12Z" }, "warmup_health": { "state": "watch", @@ -643,7 +644,7 @@ Returns the detailed status for one email account: a combined health score (fold } ``` -`warmup_status` is present only when warmup has ever been enabled; `warmup_health` is present only when the mailbox is in a warmup pool. `in_campaign` reports whether the mailbox currently backs a live campaign. +`warmup_status` is present only when warmup has ever been enabled, and its `next_send_at` is when the mailbox's next warmup send is queued for (absent while nothing is queued, for example while warmup is paused); `warmup_health` is present only when the mailbox is in a warmup pool. `in_campaign` reports whether the mailbox currently backs a live campaign. `warmup_placement` is the mailbox's measured warmup inbox rate over the trailing 7 UTC days, the same figure [Get warmup placement](#get-warmup-placement) reports as `rate`, and is absent when nothing was delivered in that window. Once `inbox_rate` is present, `health.score` is capped at it (rounded down), so a mailbox landing any mail in spam no longer reads `100`, and a rate below `90` also sets `health.status` to at least `warning` with an issue naming the rate. diff --git a/docs/content/docs/guides/sending-behavior.mdx b/docs/content/docs/guides/sending-behavior.mdx index 9f80cc8dd..97b3d5f7c 100644 --- a/docs/content/docs/guides/sending-behavior.mdx +++ b/docs/content/docs/guides/sending-behavior.mdx @@ -42,6 +42,8 @@ Check today's rolled workday at the top of the tab to see when the mailbox will - Times are local to the mailbox's timezone, or the workspace timezone when it has none. - Editing the ranges does not change today. The new ranges apply from the next roll. - Warmup keeps the same working days and hours, but does not count against the cold limits. +- Warmup waits for the workday too, so a mailbox connected after its hours end sends its first warmup email at the start of the next working day. The mailbox list shows when the next one is queued. +- Switching to Fixed schedule, back again, or changing the working days moves the mailbox's queued warmup send, and the next send of every active campaign it backs, straight away onto whichever schedule now applies. diff --git a/docs/content/docs/guides/warmup.mdx b/docs/content/docs/guides/warmup.mdx index 8dca0e497..4a30b15fc 100644 --- a/docs/content/docs/guides/warmup.mdx +++ b/docs/content/docs/guides/warmup.mdx @@ -87,7 +87,7 @@ A worker-loading warning appears only for an active, sending-enabled mailbox aft "The sending worker has not loaded this mailbox yet" means that attempt did not send. It is not proof of a deliverability hold, a refused sign-in or an IP allowlist problem. Active mailboxes are reloaded automatically; after a worker restarts, it keeps reporting its restart until the backend acknowledges the request to reload its mailboxes. -Persistent loading failures and pending warmup tasks at least an hour past their intended schedule are checked every five minutes and reported to configured operator channels and error tracking. The latter is a dispatch problem, not evidence of rejected credentials. The local dispatcher persists a five-minute retry delay after internal failures so an old failing batch cannot indefinitely block newer tasks. Backoff does not alter the intended send time, clear a safety hold, or bypass provider limits. +Persistent loading failures and pending warmup tasks at least an hour past their intended schedule are checked every five minutes and reported to configured operator channels and error tracking. The latter is a dispatch problem, not evidence of rejected credentials. The local dispatcher persists a five-minute retry delay after internal failures so an old failing batch cannot indefinitely block newer tasks. Backoff does not alter the intended send time, clear a safety hold, or bypass provider limits. A send whose preparation was interrupted before it was handed to a worker (for example by a restart) is retired after ten minutes, so it never holds up the mailbox's next warmup email. A missing mailbox also requests another reload through the worker heartbeat. Existing warmup sends stuck before execution recover automatically: after ten minutes, the scheduler retires a nonce-protected command only when neither worker execution nor a delivery receipt exists, refunds its reserved count, and resumes the mailbox's normal schedule. Delayed copies of the retired command cannot send. Customers do not need to disconnect or re-add their mailboxes. Actually started sends with missing outcomes, conflicting outcomes, provider cooldowns, authentication failures and other safety holds are not cleared by this repair. @@ -218,6 +218,8 @@ The partner cap exists because a mailbox never writes to the same partner twice Warmup hours are read in the mailbox's timezone: its own when one is set on the **Settings** tab of its drawer, otherwise the workspace timezone under **Settings > Profile > Timezones**, and UTC when neither is set. Set the workspace timezone once and every mailbox that follows it reads its warmup hours in the team's day. A mailbox with its own timezone keeps it, and changing the workspace timezone does not change any mailbox's hours. +While [Sending behavior](/guides/sending-behavior/) is on, which it is for a newly connected mailbox, warmup only sends when both the warmup hours and that workday allow it, and the Warmup tab says so above the hours. A mailbox connected after its workday ends therefore sends its first warmup email the next working morning. Switch the mailbox to **Fixed schedule** on its **Sending** tab to use the warmup hours alone. Changing the warmup hours, days or timezone (the mailbox's own, or the workspace's for a mailbox that follows it), or switching sending behavior on or off, moves the queued warmup send at once. The mailbox list shows when the next warmup send is queued under the day's count. + Warmup hours must end later than they start on the same day. Equal or reversed hours are rejected, including a change to only one endpoint that would invalidate the stored window. Existing invalid windows are not widened or corrected automatically: pending sends stay unstarted and recheck after five minutes, while admin monitoring shows **Invalid warmup sending hours** until the configured window is corrected. Other send protections still apply after correction; this does not replay uncertain sends or confirm mailbox recovery. ### Holding the ramp on an early signal diff --git a/internal/api/handler/email.go b/internal/api/handler/email.go index 584ac8773..19b41f73a 100644 --- a/internal/api/handler/email.go +++ b/internal/api/handler/email.go @@ -1,6 +1,7 @@ package handler import ( + "context" "net/http" "strconv" @@ -80,6 +81,9 @@ func (h *Handler) UpdateEmail(c *gin.Context) { if data.Status != nil && *data.Status == "active" && resp != nil { _ = h.TasksService.EnsureWarmupScheduled(c.Request.Context(), resp.ID) } + if resp != nil { + h.replaceParkedSends(c.Request.Context(), *orgID, resp.ID, data.ChangesWarmupHours(), data.ChangesSendingHours()) + } // Audit log if accountID, err := uuid.Parse(emailAccountID); err == nil { @@ -89,6 +93,16 @@ func (h *Handler) UpdateEmail(c *gin.Context) { c.JSON(http.StatusOK, resp) } +// replaceParkedSends moves work parked under a mailbox's old hours onto its new ones, so a schedule change takes effect now. +func (h *Handler) replaceParkedSends(ctx context.Context, orgID, accountID uuid.UUID, warmup, campaigns bool) { + if warmup && h.TasksService != nil { + _ = h.TasksService.ReplaceWarmupSchedule(ctx, accountID) + } + if campaigns && h.CampaignService != nil { + h.CampaignService.RescheduleForMailboxes(ctx, orgID, accountID) + } +} + // BulkTagEmails adds/removes tags across many mailboxes in one call — the // mailboxes list bulk bar. Naturally idempotent (set semantics), so retries // are safe without an Idempotency-Key. @@ -161,6 +175,11 @@ func (h *Handler) BulkTagEmails(c *gin.Context) { "removed": strconv.Itoa(len(removeTags)), }) + // Added tags can put mailboxes behind campaigns parked for want of one. + if updated > 0 && len(addTags) > 0 && h.CampaignService != nil { + h.CampaignService.RescheduleForMailboxes(c.Request.Context(), *orgID, emailIDs...) + } + c.JSON(http.StatusOK, gin.H{"updated": updated}) } diff --git a/internal/api/handler/email_behavior.go b/internal/api/handler/email_behavior.go index d5be06caa..8b6f099e7 100644 --- a/internal/api/handler/email_behavior.go +++ b/internal/api/handler/email_behavior.go @@ -120,6 +120,11 @@ func (h *Handler) UpdateEmailBehavior(c *gin.Context) { return } + // The profile shapes both warmup's and campaigns' hours, so their parked slots move with it. + if orgID := middleware.GetOrganizationID(c); orgID != nil { + h.replaceParkedSends(c.Request.Context(), *orgID, accountID, true, true) + } + h.auditOrg(c, models.AuditActionUpdate, models.AuditEntityEmailAccount, &accountID, nil, nil) c.JSON(http.StatusOK, profile) } diff --git a/internal/api/handler/organization.go b/internal/api/handler/organization.go index c832964bd..31e668f97 100644 --- a/internal/api/handler/organization.go +++ b/internal/api/handler/organization.go @@ -205,6 +205,10 @@ func (h *Handler) UpdateOrganization(c *gin.Context) { h.auditOrg(c, models.AuditActionUpdate, models.AuditEntityOrganization, orgID, nil, nil) + if req.Timezone != nil { + go h.replaceParkedSendsForWorkspace(*orgID) + } + // A presence privacy change re-gates connected sockets live (re-track / // untrack / strip activity) rather than waiting for members to reconnect. if req.PresenceShowOnline != nil || req.PresenceShowActivity != nil { @@ -721,3 +725,15 @@ func parseUUIDList(raw string) ([]uuid.UUID, *errx.Error) { } return out, nil } + +// replaceParkedSendsForWorkspace moves warmup and campaign work parked under the old workspace timezone; detached so a large workspace does not hold the request. +func (h *Handler) replaceParkedSendsForWorkspace(orgID uuid.UUID) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + if h.CampaignService != nil { + h.CampaignService.RescheduleForOrganization(ctx, orgID) + } + if h.TasksService != nil { + h.TasksService.ReplaceWorkspaceWarmupSchedules(ctx, orgID) + } +} diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 6984303f0..850fa84ee 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -424,6 +424,9 @@ func (s *analyticsService) assembleAccountStatus(ctx context.Context, email *mod } // A day of refusals is what an owner can still act on; older ones were fixed or are stale. warmupStatus.SendFailure = reads.failures[email.ID] + if at, ok := reads.nextWarmup[email.ID]; ok { + warmupStatus.NextSendAt = &at + } } coldRamp := coldRampNotice(email, reads.coldRamp, reads.now) diff --git a/internal/app/analytics/status_reads.go b/internal/app/analytics/status_reads.go index 9b5b2e7c0..8f68b0ac6 100644 --- a/internal/app/analytics/status_reads.go +++ b/internal/app/analytics/status_reads.go @@ -20,6 +20,7 @@ type statusReads struct { cloud map[uuid.UUID]*models.WarmupHealthInfo ramp *rampReads failures map[uuid.UUID]*models.WarmupSendFailure + nextWarmup map[uuid.UUID]time.Time coldRamp map[uuid.UUID]repository.ColdRampState lifecycles map[uuid.UUID]models.SendLifecycleState sendHolds map[uuid.UUID]repository.SendRecoveryHold @@ -85,6 +86,13 @@ func (s *analyticsService) loadStatusReads(ctx context.Context, emails []models. } } reads.ramp = s.loadRampReads(ctx, warming, reads.now) + if reader, ok := s.warmupRepo.(interface { + NextWarmupSends(context.Context, []uuid.UUID) (map[uuid.UUID]time.Time, error) + }); ok && len(warming) > 0 { + if m, err := reader.NextWarmupSends(ctx, warming); err == nil { + reads.nextWarmup = m + } + } if len(failing) > 0 { if m, err := s.warmupRepo.LastWarmupSendFailures(ctx, failing, reads.now.Add(-24*time.Hour)); err == nil { reads.failures = m diff --git a/internal/app/campaign/handlers.go b/internal/app/campaign/handlers.go index 6501d5b27..61ee5e58b 100644 --- a/internal/app/campaign/handlers.go +++ b/internal/app/campaign/handlers.go @@ -141,7 +141,7 @@ func (s *campaignService) Update(ctx context.Context, orgID, query string, data // A schedule edit on a running campaign must move its parked wakeup task: // otherwise a cleared or earlier start date only takes effect when the old // slot fires, and the campaign looks stuck until then (issue #171). - if data.TouchesSchedule() && resp != nil && resp.Status == "active" { + if (data.TouchesSchedule() || data.TouchesPacing()) && resp != nil && resp.Status == "active" { s.rescheduleCampaignWakeup(ctx, resp.ID) } return resp, nil @@ -162,6 +162,45 @@ func (s *campaignService) rescheduleCampaignWakeup(ctx context.Context, campaign _ = s.enqueueCampaignWakeup(ctx, campaignID, false) } +func (s *campaignService) RescheduleForMailboxes(ctx context.Context, orgID uuid.UUID, accountIDs ...uuid.UUID) { + lister, ok := s.campaignRepository.(interface { + ActiveCampaignIDsForAccount(context.Context, uuid.UUID, uuid.UUID) ([]uuid.UUID, error) + }) + if !ok { + return + } + seen := make(map[uuid.UUID]struct{}) + for _, accountID := range accountIDs { + ids, err := lister.ActiveCampaignIDsForAccount(ctx, orgID, accountID) + if err != nil { + continue + } + for _, id := range ids { + if _, done := seen[id]; done { + continue + } + seen[id] = struct{}{} + s.rescheduleCampaignWakeup(ctx, id) + } + } +} + +func (s *campaignService) RescheduleForOrganization(ctx context.Context, orgID uuid.UUID) { + lister, ok := s.campaignRepository.(interface { + ActiveCampaignIDsForOrganization(context.Context, uuid.UUID) ([]uuid.UUID, error) + }) + if !ok { + return + } + ids, err := lister.ActiveCampaignIDsForOrganization(ctx, orgID) + if err != nil { + return + } + for _, id := range ids { + s.rescheduleCampaignWakeup(ctx, id) + } +} + func (s *campaignService) Delete(ctx context.Context, orgID uuid.UUID, campaignID string) (*models.Campaign, *errx.Error) { campaign, cID, xerr := s.campaignForOrg(ctx, orgID, campaignID) if xerr != nil { @@ -994,11 +1033,15 @@ func (s *campaignService) ListCampaignSenders(ctx context.Context, orgID uuid.UU // ReplaceCampaignSenders atomically replaces the explicit sender pool. func (s *campaignService) ReplaceCampaignSenders(ctx context.Context, orgID uuid.UUID, campaignID string, in []models.CampaignSenderInput) ([]models.CampaignSender, *errx.Error) { - _, cID, xerr := s.campaignForOrg(ctx, orgID, campaignID) + campaign, cID, xerr := s.campaignForOrg(ctx, orgID, campaignID) if xerr != nil { return nil, xerr } - return s.campaignRepository.ReplaceCampaignSenders(ctx, cID, in) + senders, xerr := s.campaignRepository.ReplaceCampaignSenders(ctx, cID, in) + if xerr == nil && campaign != nil && campaign.Status == "active" { + s.rescheduleCampaignWakeup(ctx, cID) + } + return senders, xerr } // VerifyCampaignTrackingDomain resolves the campaign-scoped tracking domain diff --git a/internal/app/campaign/service.go b/internal/app/campaign/service.go index cb9c53c2d..8b6cddbb1 100644 --- a/internal/app/campaign/service.go +++ b/internal/app/campaign/service.go @@ -26,6 +26,10 @@ type CampaignService interface { // allowed limits the results to those campaigns; nil is every campaign in the workspace. Search(ctx context.Context, userID, query, cursor, folder, status, limit string, allowed []uuid.UUID) (*models.CampaignsResult, *errx.Error) Overview(ctx context.Context, orgID string, allowed []uuid.UUID) (*models.CampaignsOverview, *errx.Error) + // RescheduleForMailboxes re-places, once each, the parked wakeups of every active campaign the mailboxes back after their sending hours change. + RescheduleForMailboxes(ctx context.Context, orgID uuid.UUID, accountIDs ...uuid.UUID) + // RescheduleForOrganization re-places every active campaign's parked wakeup after the workspace timezone changes. + RescheduleForOrganization(ctx context.Context, orgID uuid.UUID) // Estimate projects how many contacts a set of segments reaches and how // many sending days a mailbox pool needs under the per-mailbox caps. // Read-only; the new-campaign flow shows it as its launch plan. diff --git a/internal/app/consumer/warmup_reply_back.go b/internal/app/consumer/warmup_reply_back.go index 686363475..bce30a1b3 100644 --- a/internal/app/consumer/warmup_reply_back.go +++ b/internal/app/consumer/warmup_reply_back.go @@ -48,7 +48,10 @@ func (s *JobsService) scheduleWarmupReplyBack(ctx context.Context, token *models at := time.Now().Add(replyBackMinDelay + time.Duration(uint64(binary.BigEndian.Uint32(token.Token[4:8]))%uint64(replyBackMaxDelay-replyBackMinDelay))) // A reply at 04:00 from a mailbox that never sends then is worse than none. - at = withinWarmupHours(at, recipient) + if rolled := withinWarmupHours(at, recipient); !rolled.Equal(at) { + // Spread replies rolled to the opening over its first hour, so the pool does not answer on the stroke of it. + at = rolled.Add(time.Duration(binary.BigEndian.Uint32(token.Token[8:12])%3600) * time.Second) + } moved, err := lineage.BindWarmupSuccessor(ctx, token.TaskID, recipientAccountID, receivedIDs[0], at) if err != nil { diff --git a/internal/models/analytics.go b/internal/models/analytics.go index 1e70cbd28..56e01a9c5 100644 --- a/internal/models/analytics.go +++ b/internal/models/analytics.go @@ -323,6 +323,8 @@ type WarmupStatusInfo struct { PartnerLimit *WarmupPartnerLimit `json:"partner_limit,omitempty"` // SendFailure is present while the newest warmup send failed and no later one was confirmed delivered. SendFailure *WarmupSendFailure `json:"send_failure,omitempty"` + // NextSendAt is when the parked warmup wakeup fires; absent while nothing is queued. + NextSendAt *time.Time `json:"next_send_at,omitempty"` } // WarmupSendFailure is why a warmup send failed: the server's answer when it gave one. diff --git a/internal/models/campaign.go b/internal/models/campaign.go index 0f29be3a5..7b492a181 100644 --- a/internal/models/campaign.go +++ b/internal/models/campaign.go @@ -462,6 +462,13 @@ func (u *UpdateCampaign) TouchesSchedule() bool { u.EntryDelayMinutes != nil } +// TouchesPacing reports whether the patch changes how much the campaign may send or from which mailboxes, which a parked wakeup was placed under too. +func (u *UpdateCampaign) TouchesPacing() bool { + return u.DailyLimit != nil || u.EmailTags != nil || u.SenderStrategy != nil || u.RotationMode != nil || + u.RampEnabled != nil || u.RampStart != nil || u.RampIncrement != nil || u.RampCeiling != nil || + u.MaxNewLeadsPerDay != nil || u.ESPMatchMode != nil +} + // CreateCampaign is the payload accepted by POST /campaigns. Name is required; // every other field is optional and only applied if the caller sent a non-nil // value. The wizard sends everything at once; the simple modal can still send diff --git a/internal/models/email.go b/internal/models/email.go index 551d0409c..7d8f9ba6f 100644 --- a/internal/models/email.go +++ b/internal/models/email.go @@ -750,6 +750,18 @@ type UpdateEmail struct { Tags []string `json:"tags"` } +// ChangesWarmupHours reports whether the patch touches anything the parked warmup slot was placed under. +func (u *UpdateEmail) ChangesWarmupHours() bool { + return u.WarmupStartTime != nil || u.WarmupEndTime != nil || u.WarmupDays != nil || u.Timezone != nil || + u.MinWaitTime != nil || u.WarmupBase != nil || u.WarmupMax != nil || u.WarmupIncrease != nil +} + +// ChangesSendingHours reports whether the patch touches anything a parked campaign wakeup was placed under. +func (u *UpdateEmail) ChangesSendingHours() bool { + return u.Timezone != nil || u.MinWaitTime != nil || u.CampaignLimit != nil || u.SharedDailyLimit != nil || + u.RollingRecipientLimit != nil || u.Tags != nil || (u.Status != nil && *u.Status == "active") +} + type SendRecoveryResolution struct { MessageID string `json:"message_id,omitempty"` ProviderMsgID string `json:"provider_msg_id,omitempty"` diff --git a/internal/repository/pg_campaign.go b/internal/repository/pg_campaign.go index e5416f1a4..c6ba74e6c 100644 --- a/internal/repository/pg_campaign.go +++ b/internal/repository/pg_campaign.go @@ -1927,12 +1927,8 @@ func (r *campaignRepository) AccountHasActiveCampaign(ctx context.Context, accou return exists, err } -// CountActiveCampaignsForAccount counts distinct active campaigns the mailbox -// backs, via the campaign's email tags (tag strategy) OR an enabled -// campaign_senders row (explicit strategy). -func (r *campaignRepository) CountActiveCampaignsForAccount(ctx context.Context, accountID uuid.UUID) (int, error) { - query := ` - SELECT COUNT(*) FROM ( +// activeCampaignsForAccountSQL selects the distinct active campaigns mailbox $1 backs: tag-resolved, explicit sender, or a tenant-wide "all" campaign. +const activeCampaignsForAccountSQL = ` -- tag-resolved SELECT c.id FROM campaigns c @@ -1959,7 +1955,31 @@ func (r *campaignRepository) CountActiveCampaignsForAccount(ctx context.Context, AND c.status = 'active' AND NOT EXISTS (SELECT 1 FROM campaign_email_tags cet2 WHERE cet2.campaign_id = c.id) AND NOT EXISTS (SELECT 1 FROM campaign_senders cs2 WHERE cs2.campaign_id = c.id AND cs2.enabled) - ) AS active_campaigns` +` + +// ActiveCampaignIDsForAccount lists orgID's active campaigns the mailbox backs, under the same scope rule as the count. +func (r *campaignRepository) ActiveCampaignIDsForAccount(ctx context.Context, orgID, accountID uuid.UUID) ([]uuid.UUID, error) { + rows, err := resultDB(ctx, r.DB).Query(ctx, `SELECT c.id FROM (`+activeCampaignsForAccountSQL+`) a JOIN campaigns c ON c.id = a.id WHERE c.organization_id = $2`, accountID, orgID) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// CountActiveCampaignsForAccount counts distinct active campaigns the mailbox +// backs, via the campaign's email tags (tag strategy) OR an enabled +// campaign_senders row (explicit strategy). +func (r *campaignRepository) CountActiveCampaignsForAccount(ctx context.Context, accountID uuid.UUID) (int, error) { + query := `SELECT COUNT(*) FROM (` + activeCampaignsForAccountSQL + `) AS active_campaigns` var count int err := resultDB(ctx, r.DB).QueryRow(ctx, query, accountID).Scan(&count) return count, err @@ -2401,3 +2421,21 @@ func connectLinearSequenceTx(ctx context.Context, tx pgx.Tx, stepIDs []uuid.UUID } return nil } + +// ActiveCampaignIDsForOrganization lists the workspace's active campaigns. +func (r *campaignRepository) ActiveCampaignIDsForOrganization(ctx context.Context, orgID uuid.UUID) ([]uuid.UUID, error) { + rows, err := resultDB(ctx, r.DB).Query(ctx, `SELECT id FROM campaigns WHERE organization_id = $1 AND status = 'active'`, orgID) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index f6a0d9b1c..126e2a994 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -2882,3 +2882,22 @@ func (r *emailRepository) ListOrganizationIDs(ctx context.Context) ([]uuid.UUID, } return out, rows.Err() } + +// WarmingOnWorkspaceClock lists the workspace's warming mailboxes that read their hours in the workspace timezone. +func (r *emailRepository) WarmingOnWorkspaceClock(ctx context.Context, orgID uuid.UUID) ([]uuid.UUID, error) { + rows, err := resultDB(ctx, r.DB).Query(ctx, `SELECT id FROM email_accounts + WHERE organization_id = $1 AND COALESCE(timezone, '') = '' AND status = 'active' AND warmup IS NOT NULL AND warmup_paused_at IS NULL`, orgID) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} diff --git a/internal/repository/pg_warmup_status_batch.go b/internal/repository/pg_warmup_status_batch.go index f61d3647e..a0c44be4a 100644 --- a/internal/repository/pg_warmup_status_batch.go +++ b/internal/repository/pg_warmup_status_batch.go @@ -212,3 +212,29 @@ func (r *warmupRepository) LastWarmupSendFailures(ctx context.Context, accountID } return out, rows.Err() } + +// NextWarmupSends returns each mailbox's earliest pending warmup slot. +func (r *warmupRepository) NextWarmupSends(ctx context.Context, accountIDs []uuid.UUID) (map[uuid.UUID]time.Time, error) { + out := make(map[uuid.UUID]time.Time) + if len(accountIDs) == 0 { + return out, nil + } + rows, err := resultDB(ctx, r.db).Query(ctx, ` + SELECT email_account_id, MIN(scheduled_at) FROM tasks + WHERE email_account_id = ANY($1) AND task_type = 'warmup' AND status = 'pending' AND scheduled_at IS NOT NULL + GROUP BY email_account_id + `, accountIDs) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var id uuid.UUID + var at time.Time + if err := rows.Scan(&id, &at); err != nil { + return nil, err + } + out[id] = at + } + return out, rows.Err() +} diff --git a/internal/repository/warmup_dispatch_recovery.go b/internal/repository/warmup_dispatch_recovery.go index bc91d48c7..38fae34bd 100644 --- a/internal/repository/warmup_dispatch_recovery.go +++ b/internal/repository/warmup_dispatch_recovery.go @@ -25,17 +25,20 @@ const unstartedWarmupDispatch = `t.task_type='warmup' AND t.status='completed' AND NOT EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id AND wt.sent_message_id<>'') AND NOT EXISTS(SELECT 1 FROM warmup_received wr WHERE wr.task_id=t.id)` +// A preparation that never got a dispatch nonce was never authorized, so its message id, an unsent token and a refunded charge prove nothing left. const abandonedWarmupPreparation = `t.task_type='warmup' AND t.status='active' AND t.updated_at<$1 - AND t.message_id='' AND t.completed_at IS NULL + AND t.completed_at IS NULL AND t.send_reserved_at IS NULL AND t.send_result_state IS NULL AND t.send_result_evidence IS NULL AND t.send_result_applied_at IS NULL AND t.send_released_at IS NULL AND t.send_executor_nonce IS NULL AND t.send_executor_worker IS NULL AND t.send_executor_started_at IS NULL AND t.send_executor_result IS NULL AND w.dispatch_nonce IS NULL AND w.dispatch_worker_id IS NULL - AND w.dispatch_started_at IS NULL AND w.dispatch_result IS NULL AND w.warmup_charged_date IS NULL - AND NOT EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id) + AND w.dispatch_started_at IS NULL AND w.dispatch_result IS NULL + AND (w.warmup_charged_date IS NULL OR w.warmup_refunded_at IS NOT NULL) + AND NOT EXISTS(SELECT 1 FROM warmup_tokens wt WHERE wt.task_id=t.id AND (wt.sent_message_id<>'' OR wt.consumed_at IS NOT NULL OR wt.sent_retired_at IS NOT NULL)) AND NOT EXISTS(SELECT 1 FROM outbound_attempts a WHERE a.task_id=t.id) - AND NOT EXISTS(SELECT 1 FROM warmup_received wr WHERE wr.task_id=t.id)` + AND NOT EXISTS(SELECT 1 FROM warmup_received wr WHERE wr.task_id=t.id) + AND (t.message_id='' OR NOT EXISTS(SELECT 1 FROM warmup_received wr WHERE btrim(wr.message_id,'<>')=btrim(t.message_id,'<>')))` func (r *taskRepository) RecoverAbandonedWarmupPreparations(ctx context.Context, before time.Time, limit int) (int, error) { rows, err := r.db.Query(ctx, `SELECT t.id FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id @@ -89,7 +92,14 @@ func (r *taskRepository) retireAbandonedWarmupPreparation(ctx context.Context, t if err != nil { return false, err } - return tag.RowsAffected() == 1, tx.Commit(ctx) + if tag.RowsAffected() != 1 { + return false, nil + } + // Its tokens were never sent; leaving them would let a partner's sync verify mail that does not exist. + if _, err := tx.Exec(ctx, `DELETE FROM warmup_tokens WHERE task_id=$1 AND sent_message_id=''`, task); err != nil { + return false, err + } + return true, tx.Commit(ctx) } // Only nonce-gated commands that never began execution can be safely retired. diff --git a/internal/repository/warmup_lineage.go b/internal/repository/warmup_lineage.go index 657be50ca..a0869b3ad 100644 --- a/internal/repository/warmup_lineage.go +++ b/internal/repository/warmup_lineage.go @@ -262,6 +262,26 @@ func (r *taskRepository) RescheduleWarmupTask(ctx context.Context, id uuid.UUID, return tx.Commit(ctx) } +// ReplaceParkedWarmupTask moves a mailbox's plain pending warmup wakeup (no reply target, no parent) to at; the reconciler requeues the new revision. +func (r *taskRepository) ReplaceParkedWarmupTask(ctx context.Context, accountID uuid.UUID, at time.Time) (bool, error) { + tx, err := r.db.Begin(ctx) + if err != nil { + return false, err + } + defer tx.Rollback(ctx) + if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('warmup_task_' || $1::text))`, accountID); err != nil { + return false, err + } + tag, err := tx.Exec(ctx, `WITH moved AS (UPDATE tasks t SET scheduled_at=$2,updated_at=NOW() FROM warmup_tasks w + WHERE t.email_account_id=$1 AND t.task_type='warmup' AND t.status='pending' AND w.task_id=t.id + AND w.target_account_id IS NULL AND w.parent_task_id IS NULL AND w.dispatch_nonce IS NULL RETURNING t.id) + UPDATE warmup_tasks SET schedule_revision=schedule_revision+1 WHERE task_id IN (SELECT id FROM moved)`, accountID, at) + if err != nil { + return false, err + } + return tag.RowsAffected() > 0, tx.Commit(ctx) +} + func (r *taskRepository) UnqueuedWarmupTasks(ctx context.Context, limit int) ([]WarmupQueueRevision, error) { rows, err := r.db.Query(ctx, `SELECT t.id,t.scheduled_at,w.schedule_revision,t.cloud_task_name FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.status='pending' AND t.scheduled_at IS NOT NULL AND (w.schedule_revision>w.queued_revision OR t.scheduled_at' FROM m`}, {"reserved", `UPDATE tasks SET send_reserved_at=NOW() WHERE id=$1`}, {"unknown outcome", `UPDATE tasks SET send_result_state='unknown' WHERE id=$1`}, {"provider evidence", `UPDATE tasks SET send_result_evidence='{}' WHERE id=$1`}, @@ -123,7 +124,8 @@ func TestLiveWarmupRecoveryPreservesPreparationWithAnySendEvidence(t *testing.T) {"dispatch started", `UPDATE warmup_tasks SET dispatch_started_at=NOW() WHERE task_id=$1`}, {"dispatch result", `UPDATE warmup_tasks SET dispatch_result='{}' WHERE task_id=$1`}, {"charged", `UPDATE warmup_tasks SET warmup_charged_date=CURRENT_DATE,warmup_reply_charged=false WHERE task_id=$1`}, - {"token", `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id) SELECT gen_random_uuid(),id,email_account_id,$2 FROM tasks WHERE id=$1`}, + {"sent token", `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id,sent_message_id) SELECT gen_random_uuid(),id,email_account_id,$2,'' FROM tasks WHERE id=$1`}, + {"consumed token", `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id,consumed_at) SELECT gen_random_uuid(),id,email_account_id,$2,NOW() FROM tasks WHERE id=$1`}, {"outbound attempt", `INSERT INTO outbound_attempts(nonce,task_id,email_account_id,provider,attempted_at,recipient_count) SELECT gen_random_uuid(),id,email_account_id,'smtp_imap',NOW(),1 FROM tasks WHERE id=$1`}, {"receipt", `INSERT INTO warmup_received(email_account_id,internal_id,sender_account_id,message_id,task_id) SELECT $2,gen_random_uuid(),email_account_id,'',id FROM tasks WHERE id=$1`}, } { @@ -143,6 +145,36 @@ func TestLiveWarmupRecoveryPreservesPreparationWithAnySendEvidence(t *testing.T) } } +func TestLiveWarmupRecoveryRetiresUnauthorizedPreparationsThatLeftNothing(t *testing.T) { + for _, tc := range []struct{ name, update string }{ + {"refunded charge", `UPDATE warmup_tasks SET warmup_charged_date=CURRENT_DATE,warmup_reply_charged=false,warmup_refunded_at=NOW() WHERE task_id=$1`}, + {"message and unsent token", `WITH m AS (UPDATE tasks SET message_id='' WHERE id=$1 RETURNING id,email_account_id) + INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id) SELECT gen_random_uuid(),id,email_account_id,$2 FROM m`}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newWarmupRecoveryFixture(t) + id := f.abandonedPreparation(t) + if strings.Contains(tc.update, "$2") { + f.exec(t, tc.update, id, f.atWorkspace) + } else { + f.exec(t, tc.update, id) + } + n, err := f.r.RecoverAbandonedWarmupPreparations(t.Context(), f.before, 100) + if err != nil || n != 1 { + t.Fatalf("stranded preparation not retired: n=%d err=%v", n, err) + } + var status string + var tokens int + if err := f.pool.QueryRow(t.Context(), `SELECT status,(SELECT count(*) FROM warmup_tokens WHERE task_id=$1) FROM tasks WHERE id=$1`, id).Scan(&status, &tokens); err != nil || status != "cancelled" || tokens != 0 { + t.Fatalf("status=%s tokens=%d err=%v", status, tokens, err) + } + if _, err := f.r.AuthorizeWarmupDispatch(t.Context(), id, f.sender.ID, f.worker); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("retired preparation can still authorize: %v", err) + } + }) + } +} + func TestLiveWarmupSafetyHoldReschedulesInsteadOfBlockingTheDispatcher(t *testing.T) { f := newWarmupRecoveryFixture(t) f.exec(t, `UPDATE email_accounts SET warmup_start_time='08:00',warmup_end_time='07:00' WHERE id=$1`, f.sender.ID) diff --git a/web/src/app/app/emails/page.tsx b/web/src/app/app/emails/page.tsx index 6ddba09ed..d96b6409c 100644 --- a/web/src/app/app/emails/page.tsx +++ b/web/src/app/app/emails/page.tsx @@ -127,6 +127,7 @@ import { useAdvisorEntityIndex } from "@/lib/api/hooks/app/advisor/useAdvisor"; import type { AdvisorFinding } from "@/lib/api/models/app/advisor/Advisor"; import { Checkbox } from "@/components/ui/checkbox"; import { labelInk } from "@/lib/utils"; +import { nextSlotLabel } from "@/lib/helper/nextSlot"; export default function AddressesPage() { const p = useUserProfile(); @@ -901,6 +902,7 @@ function MailboxRow({ // A refused send outranks the count: it is why the count is not moving. const sendFailure = inCloud ? cloudSendFailure(cloud?.cloud) : warming ? ws?.send_failure : undefined; + const nextWarmup = warming && ws?.next_send_at ? nextSlotLabel(ws.next_send_at) : undefined; // Warmup column: what's flowing today and why. const warmupLabel = inCloud @@ -987,14 +989,23 @@ function MailboxRow({ ) : warming ? ( - {sendFailure ? : } - - - /{ws?.target_volume ?? box.warmup_base} + + {sendFailure ? : } + + + /{ws?.target_volume ?? box.warmup_base} + + {nextWarmup && !sendFailure && ( + next {nextWarmup} + )} ) : ( Today
Sending {ws.current_volume} of {ws.target_volume} · day {ws.days_active} + {ws.next_send_at && · next {nextSlotLabel(ws.next_send_at)}}
@@ -1547,6 +1550,12 @@ function WarmupTab({ form, update, status, mailbox, canWarmup = true }: { form: {/* Schedule */}
Sending window + {behaviorOn && ( +

+ Sending behaviour is on for this mailbox, so warmup only sends when both this window and the workday on the Sending tab + allow it. Switch it to Fixed schedule there to use these hours alone. +

+ )}
update({ warmup_start_time: v })} /> diff --git a/web/src/lib/api/models/app/analytics/AccountStatus.ts b/web/src/lib/api/models/app/analytics/AccountStatus.ts index a14cefcad..5068054da 100644 --- a/web/src/lib/api/models/app/analytics/AccountStatus.ts +++ b/web/src/lib/api/models/app/analytics/AccountStatus.ts @@ -46,6 +46,8 @@ export interface WarmupStatusInfo { partner_limit?: WarmupPartnerLimit; /** Present while the newest warmup send was refused and none has gone out since. */ send_failure?: WarmupSendFailure; + /** When the next warmup send is queued for; absent while nothing is queued. */ + next_send_at?: Date; } // Why a warmup send failed: the mail server's answer when it gave one. diff --git a/web/src/lib/helper/nextSlot.ts b/web/src/lib/helper/nextSlot.ts new file mode 100644 index 000000000..22c293703 --- /dev/null +++ b/web/src/lib/helper/nextSlot.ts @@ -0,0 +1,11 @@ +// A queued send time as a short label: "10:08" today, "tomorrow 09:12", or "Mon 09:12" later in the week. +export function nextSlotLabel(at: Date | string, now: Date = new Date()): string { + const d = new Date(at); + const time = d.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" }); + const day = (x: Date) => new Date(x.getFullYear(), x.getMonth(), x.getDate()).getTime(); + const days = Math.round((day(d) - day(now)) / 86_400_000); + if (days <= 0) return time; + if (days === 1) return `tomorrow ${time}`; + if (days < 7) return `${d.toLocaleDateString([], { weekday: "short" })} ${time}`; + return `${d.toLocaleDateString([], { month: "short", day: "numeric" })} ${time}`; +}