feat: atomically validate same-day warmup hours and defer invalid configured windows before claiming or sending while preserving visible safety holds

This commit is contained in:
Devin AI
2026-10-10 10:48:15 +00:00
parent 3ececd2934
commit d152c0778a
12 changed files with 276 additions and 4 deletions
+2 -2
View File
@@ -207,8 +207,8 @@ jobs:
env:
WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
run: |
go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement)' -count=1
go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults)|WarmupSenderDistinguishes)' -count=1
go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement|WarmupTimeWindow|WarmupInvalidWindow)' -count=1
go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1
go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1
- name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL)
+1 -1
View File
@@ -104,7 +104,7 @@ backfill_running backfill_stale_running campaign_reserved campaign_reserved_agin
confirmed_other_1h confirmed_warmup_1h confirmed_warmup_proxy_1h credential_errors
dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed
dlq_unscheduled dlq_wait effects_leased effects_pending effects_repeated effects_unleased
failed_results_1h failed_results_24h future_schedule holds_authentication holds_conflict
failed_results_1h failed_results_24h future_schedule holds_authentication holds_conflict warmup_invalid_window
holds_permanent holds_unknown inactive_credential_errors job_error_old job_error_recent
job_future job_ok job_overdue job_running notification_aging notification_due
notification_exhausted notification_retry notification_sending notification_settled_24h
+2
View File
@@ -19,6 +19,8 @@ All paths below are relative to the versioned base URL `https://api.warmbly.com/
### Emails
`PATCH /emails/:id` validates `warmup_start_time` and `warmup_end_time` together: the end must be later than the start on the same day. A partial time patch uses the other stored endpoint; an invalid pair returns `400`. Unrelated patches do not rewrite existing invalid hours. See [warmup hours](/guides/warmup/).
| Method | Path | API Permission |
|--------|------|----------------|
| GET | `/emails` | `READ_EMAILS` |
+2
View File
@@ -208,6 +208,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.
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
If any warmup email lands in the spam folder of a Google, Microsoft or Yahoo recipient, that mailbox stops climbing immediately:
+22
View File
@@ -1647,6 +1647,28 @@ func (r *emailRepository) Update(ctx context.Context, orgID, emailAccountID stri
return nil, errx.InternalError()
}
defer tx.Rollback(ctx)
if udata.WarmupStartTime != nil || udata.WarmupEndTime != nil {
// Lock the stored half of a partial patch so concurrent changes cannot cross.
query := `SELECT warmup_start_time::text, warmup_end_time::text FROM email_accounts WHERE organization_id=$1 AND id=$2 FOR UPDATE`
var start, end string
if err := tx.QueryRow(ctx, query, orgID, emailAccountID).Scan(&start, &end); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, errx.ErrNotFound
}
db.CaptureError(err, query, []any{orgID, emailAccountID}, "queryrow")
return nil, errx.InternalError()
}
if udata.WarmupStartTime != nil {
start = *udata.WarmupStartTime
}
if udata.WarmupEndTime != nil {
end = *udata.WarmupEndTime
}
startMinute, endMinute := models.ClockMinutes(start, -1), models.ClockMinutes(end, -1)
if startMinute < 0 || endMinute < 0 || endMinute <= startMinute {
return nil, errx.New(errx.BadRequest, "warmup end time must be later than start time on the same day")
}
}
if udata.SendRecoveryResolution != nil {
resolution := udata.SendRecoveryResolution
if strings.ContainsAny(resolution.ConfirmationReference, "\r\n") || len(resolution.ConfirmationReference) > 256 {
+1
View File
@@ -174,6 +174,7 @@ var monitoringSections = map[string]monitoringSection{
{id: "worker_sample_sends_1h", title: "Reported worker successful sends (1h)", unit: "reports", condition: models.MonitoringUnknown, severity: "info", note: "Worker-reported counter, not provider-confirmed warmup delivery; token-confirmed tasks are measured separately.", window: time.Hour, query: `SELECT SUM(h.sends_succeeded)::bigint,NULL::bigint,NULL::bigint,NULL::bigint,MIN(h.observed_at),MAX(h.observed_at),NULL::timestamptz FROM worker_health_samples h WHERE h.observed_at BETWEEN $1-INTERVAL '1 hour' AND $1`},
}},
"send_safety": {permission: models.AdminPermViewUsers, investigate: "/mailboxes", queries: []monitoringQuery{
{id: "warmup_invalid_window", title: "Invalid warmup sending hours", unit: "mailboxes", condition: "safety_hold", severity: "warning", note: "Active warming mailboxes whose end time is not later than start time. Correct the configured hours; no provider execution or recovery is implied.", query: `SELECT COUNT(*), COUNT(DISTINCT e.id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(NULL::timestamptz), MAX(NULL::timestamptz), MIN(NULL::timestamptz) FROM email_accounts e WHERE e.status='active' AND e.warmup IS NOT NULL AND e.warmup_paused_at IS NULL AND (e.test_mode IS NULL OR e.test_mode='legacy' OR e.test_mode='diagnostic' AND e.test_send_enabled) AND e.warmup_end_time<=e.warmup_start_time`},
{id: "holds_unknown", title: "Unknown send outcome protection", unit: "mailboxes", condition: "safety_hold", severity: "warning", note: "Protection can be set before execution; not a provider failure or permission to resend.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT e.id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(NULL::timestamptz), MAX(NULL::timestamptz), MIN(NULL::timestamptz) FROM email_accounts e WHERE e.send_recovery_hold AND COALESCE(e.send_recovery_reason,'unknown')='unknown'`},
{id: "holds_conflict", title: "conflict send protection", unit: "mailboxes", condition: "safety_hold", severity: "warning", note: "Current persisted safety restriction, not proof of a new rejection.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT e.id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(NULL::timestamptz), MAX(NULL::timestamptz), MIN(NULL::timestamptz) FROM email_accounts e WHERE e.send_recovery_hold AND e.send_recovery_reason='conflict'`},
{id: "holds_permanent", title: "permanent send protection", unit: "mailboxes", condition: "safety_hold", severity: "warning", note: "Current persisted safety restriction, not proof of a new rejection.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT e.id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(NULL::timestamptz), MAX(NULL::timestamptz), MIN(NULL::timestamptz) FROM email_accounts e WHERE e.send_recovery_hold AND e.send_recovery_reason='permanent'`},
@@ -0,0 +1,124 @@
package repository
import (
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/warmbly/warmbly/internal/errx"
"github.com/warmbly/warmbly/internal/models"
)
func TestLiveWarmupTimeWindowValidatesEffectivePatchAndPreservesHolds(t *testing.T) {
handle, pool := liveContactDB(t)
f := newWarmupUsageFixture(t, pool)
r := NewEmailRepostory(handle, nil)
ptr := func(v string) *string { return &v }
for _, tc := range []struct {
name, end string
patch models.UpdateEmail
invalid bool
}{
{"valid pair", "20:00", models.UpdateEmail{WarmupStartTime: ptr("09:00"), WarmupEndTime: ptr("17:00")}, false},
{"malformed time", "20:00", models.UpdateEmail{WarmupStartTime: ptr("25:00")}, true},
{"reversed pair", "20:00", models.UpdateEmail{WarmupStartTime: ptr("08:00"), WarmupEndTime: ptr("07:00"), Name: ptr("Changed")}, true},
{"equal pair", "20:00", models.UpdateEmail{WarmupStartTime: ptr("08:00"), WarmupEndTime: ptr("08:00")}, true},
{"partial end", "20:00", models.UpdateEmail{WarmupEndTime: ptr("07:00")}, true},
{"partial start", "20:00", models.UpdateEmail{WarmupStartTime: ptr("21:00")}, true},
{"valid partial end", "20:00", models.UpdateEmail{WarmupEndTime: ptr("19:00")}, false},
{"explicit repair", "07:00", models.UpdateEmail{WarmupEndTime: ptr("19:00")}, false},
{"unrelated legacy patch", "07:00", models.UpdateEmail{Name: ptr("Changed")}, false},
} {
t.Run(tc.name, func(t *testing.T) {
f.exec(`UPDATE email_accounts SET warmup_start_time='08:00',warmup_end_time=$2,name='Before',send_recovery_hold=true,send_recovery_reason='unknown' WHERE id=$1`, f.account, tc.end)
_, xerr := r.Update(t.Context(), f.org.String(), f.account.String(), &tc.patch)
if tc.invalid {
if xerr == nil || xerr.Code != errx.BadRequest {
t.Fatalf("invalid window not rejected: %v", xerr)
}
} else if xerr != nil {
t.Fatal(xerr)
}
var start, end, name, reason string
var held bool
if err := pool.QueryRow(t.Context(), `SELECT warmup_start_time::text,warmup_end_time::text,name,send_recovery_hold,send_recovery_reason FROM email_accounts WHERE id=$1`, f.account).Scan(&start, &end, &name, &held, &reason); err != nil {
t.Fatal(err)
}
if !held || reason != "unknown" {
t.Fatal("configuration patch released unknown send protection")
}
if tc.invalid && (models.ClockHHMM(start) != "08:00" || models.ClockHHMM(end) != tc.end || name != "Before") {
t.Fatal("rejected patch changed persisted settings")
}
if !tc.invalid && tc.patch.WarmupEndTime != nil && models.ClockHHMM(end) != *tc.patch.WarmupEndTime {
t.Fatal("explicit hours correction was not saved")
}
if tc.patch.WarmupStartTime == nil && tc.patch.WarmupEndTime == nil && (models.ClockHHMM(end) != tc.end || models.ClockHHMM(start) != "08:00") {
t.Fatal("unrelated update changed legacy invalid hours")
}
})
}
_, xerr := r.Update(t.Context(), uuid.NewString(), f.account.String(), &models.UpdateEmail{WarmupEndTime: ptr("19:00")})
if xerr == nil || xerr.Code != errx.NotFound {
t.Fatal("time patch crossed workspace ownership")
}
}
func TestLiveWarmupTimeWindowSerializesCrossingPartialPatches(t *testing.T) {
handle, pool := liveContactDB(t)
f := newWarmupUsageFixture(t, pool)
f.exec(`UPDATE email_accounts SET warmup_start_time='08:00',warmup_end_time='20:00' WHERE id=$1`, f.account)
r := NewEmailRepostory(handle, nil)
start, end := "18:00", "17:00"
results := make(chan *errx.Error, 2)
begin := make(chan struct{})
var wg sync.WaitGroup
for _, patch := range []models.UpdateEmail{{WarmupStartTime: &start}, {WarmupEndTime: &end}} {
wg.Go(func() {
<-begin
_, xerr := r.Update(t.Context(), f.org.String(), f.account.String(), &patch)
results <- xerr
})
}
close(begin)
wg.Wait()
close(results)
accepted, rejected := 0, 0
for xerr := range results {
if xerr == nil {
accepted++
} else if xerr.Code == errx.BadRequest {
rejected++
} else {
t.Fatal(xerr)
}
}
if accepted != 1 || rejected != 1 {
t.Fatalf("crossing patches accepted=%d rejected=%d", accepted, rejected)
}
var valid bool
if err := pool.QueryRow(t.Context(), `SELECT warmup_start_time<warmup_end_time FROM email_accounts WHERE id=$1`, f.account).Scan(&valid); err != nil || !valid {
t.Fatalf("invalid concurrent window persisted: valid=%v err=%v", valid, err)
}
}
func TestLiveWarmupInvalidWindowRemainsVisibleUntilCorrected(t *testing.T) {
_, pool := liveContactDB(t)
f := newWarmupUsageFixture(t, pool)
r := NewMonitoringRepository(pool)
now := time.Now()
before := measuredMonitoring(t, collectMonitoring(t, r, "send_safety", now), "warmup_invalid_window")
f.exec(`UPDATE email_accounts SET warmup=NOW(),warmup_start_time='08:00',warmup_end_time='07:00',test_mode='diagnostic',test_send_enabled=true WHERE id=$1`, f.account)
bad := measuredMonitoring(t, collectMonitoring(t, r, "send_safety", now), "warmup_invalid_window")
if *bad.Count != *before.Count+1 || bad.Condition != "safety_hold" || *bad.AffectedMailboxes != *before.AffectedMailboxes+1 || *bad.AffectedOrganizations != *before.AffectedOrganizations+1 || bad.EvidenceAt != nil || bad.LatestEvidenceAt != nil {
t.Fatal("invalid configured hours lost their scoped protection classification", bad)
}
for _, update := range []string{"test_send_enabled=false", "test_send_enabled=true,warmup_paused_at=NOW()", "warmup_paused_at=NULL,warmup_end_time='19:00'"} {
f.exec(`UPDATE email_accounts SET `+update+` WHERE id=$1`, f.account)
got := measuredMonitoring(t, collectMonitoring(t, r, "send_safety", now), "warmup_invalid_window")
if *got.Count != *before.Count {
t.Fatal("stopped or corrected mailbox remained an invalid-window hold", got)
}
}
}
+3 -1
View File
@@ -11,6 +11,8 @@ import (
"github.com/warmbly/warmbly/internal/models"
)
var ErrWarmupTimeWindowInvalid = errors.New("invalid warmup time window")
// WarmupDispatchNotBefore rechecks current policy, not the queue's old snapshot.
func (s *schedulerService) WarmupDispatchNotBefore(ctx context.Context, id uuid.UUID, now time.Time) (time.Time, error) {
return s.WarmupExecutionNotBefore(ctx, id, uuid.Nil, now)
@@ -143,7 +145,7 @@ func warmupDispatchWindow(a *models.Email, bhv behavior.Resolved, desired time.T
{
start, end := models.ClockMinutes(a.WarmupStartTime, 8*60), models.ClockMinutes(a.WarmupEndTime, 20*60)
if end <= start {
return time.Time{}, errors.New("invalid warmup time window")
return time.Time{}, ErrWarmupTimeWindowInvalid
}
local := at.In(loc)
minutes := local.Hour()*60 + local.Minute()
@@ -1,12 +1,40 @@
package scheduler
import (
"errors"
"testing"
"time"
"github.com/warmbly/warmbly/internal/app/behavior"
"github.com/warmbly/warmbly/internal/app/warmupramp"
"github.com/warmbly/warmbly/internal/models"
)
func TestWarmupDispatchWindowClassifiesInvalidHoursWithoutOpeningThem(t *testing.T) {
now := time.Date(2026, time.October, 10, 12, 0, 0, 0, time.UTC)
for _, tc := range []struct {
start, end string
invalid bool
}{
{"08:00", "07:00", true},
{"08:00:00", "07:00:00", true},
{"08:00", "08:00", true},
{"08:00:00", "19:00:00", false},
} {
t.Run(tc.start+"/"+tc.end, func(t *testing.T) {
a := &models.Email{WarmupStartTime: tc.start, WarmupEndTime: tc.end, WarmupDays: 127}
at, err := warmupDispatchWindow(a, behavior.Resolved{}, now)
if tc.invalid {
if !errors.Is(err, ErrWarmupTimeWindowInvalid) || !at.IsZero() {
t.Fatalf("invalid hours opened: at=%v err=%v", at, err)
}
} else if err != nil || !at.Equal(now) {
t.Fatalf("valid hours rejected: at=%v err=%v", at, err)
}
})
}
}
func TestAdjustmentFor(t *testing.T) {
tests := []struct {
name string
+7
View File
@@ -15,6 +15,7 @@ import (
"github.com/warmbly/warmbly/internal/observability/errs"
"github.com/warmbly/warmbly/internal/pkg/mailhost"
"github.com/warmbly/warmbly/internal/repository"
"github.com/warmbly/warmbly/internal/scheduler"
"github.com/warmbly/warmbly/internal/tasks/proto"
)
@@ -228,6 +229,12 @@ func (s *tasksService) HandleEmailTask(task *proto.ProcessTask) *errx.Error {
WarmupDispatchNotBefore(context.Context, uuid.UUID, time.Time) (time.Time, error)
}); ok {
at, err := gate.WarmupDispatchNotBefore(ctx, account.ID, time.Now())
if errors.Is(err, scheduler.ErrWarmupTimeWindowInvalid) {
if err := lineage.RescheduleWarmupTask(ctx, taskID, time.Now().Add(5*time.Minute)); err != nil {
return errx.InternalError()
}
return nil
}
if err != nil {
return errx.InternalError()
}
@@ -1,14 +1,97 @@
package tasks
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/warmbly/warmbly/internal/pkg/generation"
"github.com/warmbly/warmbly/internal/repository"
"github.com/warmbly/warmbly/internal/scheduler"
"github.com/warmbly/warmbly/internal/tasks/proto"
)
func TestLiveWarmupDispatchDefersInvalidHoursBeforeClaimOrSend(t *testing.T) {
f := newPartnerRoutingFixture(t)
ctx := t.Context()
task := uuid.New()
exec := func(query string, args ...any) {
t.Helper()
if _, err := f.pool.Exec(ctx, query, args...); err != nil {
t.Fatal(err)
}
}
exec(`UPDATE email_accounts SET warmup=NOW(),warmup_start_time='08:00',warmup_end_time='07:00',warmup_pool_type='free' WHERE id=$1`, f.sender.ID)
exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,scheduled_at)VALUES($1,$2,'warmup','pending','',NOW()-INTERVAL '1 hour')`, task, f.sender.ID)
exec(`INSERT INTO warmup_tasks(task_id)VALUES($1)`, task)
base := repository.NewTaskRepository(f.pool)
observed := &windowCheckedTasks{TaskRepository: base, WarmupLineageRepository: base.(repository.WarmupLineageRepository), SendResultRecovery: base.(repository.SendResultRecovery)}
f.svc.taskRepo = observed
f.svc.scheduler = scheduler.NewSchedulerService(f.svc.taskRepo, f.svc.warmupRepo, nil, f.svc.emailRepo, nil, nil, nil)
sender := &preparationDeferringSender{}
f.svc.emailSender = sender
if xerr := f.svc.HandleEmailTask(&proto.ProcessTask{TaskId: task.String()}); xerr != nil {
t.Fatalf("invalid configured hours remained a dispatcher error: %v", xerr)
}
var status string
var scheduled time.Time
var unstarted bool
if err := f.pool.QueryRow(ctx, `SELECT status,scheduled_at,send_reserved_at IS NULL AND send_executor_started_at IS NULL AND send_result_state IS NULL FROM tasks WHERE id=$1`, task).Scan(&status, &scheduled, &unstarted); err != nil {
t.Fatal(err)
}
if status != "pending" || !scheduled.After(time.Now().Add(4*time.Minute)) || !unstarted || sender.calls != 0 {
t.Fatalf("status=%s scheduled=%v unstarted=%v calls=%d", status, scheduled, unstarted, sender.calls)
}
if observed.claimed {
t.Fatal("invalid window reached the task claim")
}
var start, end string
if err := f.pool.QueryRow(ctx, `SELECT warmup_start_time::text,warmup_end_time::text FROM email_accounts WHERE id=$1`, f.sender.ID).Scan(&start, &end); err != nil || start != "08:00:00" || end != "07:00:00" {
t.Fatalf("configured hours silently changed: %s/%s err=%v", start, end, err)
}
// Real database failures must still fail dispatch rather than count as deferral.
exec(`UPDATE tasks SET scheduled_at=NOW()-INTERVAL '1 hour' WHERE id=$1`, task)
f.svc.scheduler = failedWindowScheduler{}
if xerr := f.svc.HandleEmailTask(&proto.ProcessTask{TaskId: task.String()}); xerr == nil {
t.Fatal("authority failure became a successful configuration deferral")
}
f.svc.scheduler = scheduler.NewSchedulerService(f.svc.taskRepo, f.svc.warmupRepo, nil, f.svc.emailRepo, nil, nil, nil)
observed.deferralErr = context.DeadlineExceeded
if xerr := f.svc.HandleEmailTask(&proto.ProcessTask{TaskId: task.String()}); xerr == nil {
t.Fatal("failed reschedule was acknowledged as successful configuration deferral")
}
if observed.claimed || sender.calls != 0 {
t.Fatal("failed authority or reschedule reached claim or provider execution")
}
}
type windowCheckedTasks struct {
repository.TaskRepository
repository.WarmupLineageRepository
repository.SendResultRecovery
claimed bool
deferralErr error
}
func (r *windowCheckedTasks) ClaimWarmupTask(ctx context.Context, id uuid.UUID, at time.Time) (bool, error) {
r.claimed = true
return r.WarmupLineageRepository.ClaimWarmupTask(ctx, id, at)
}
func (r *windowCheckedTasks) RescheduleWarmupTask(ctx context.Context, id uuid.UUID, at time.Time) error {
if r.deferralErr != nil {
return r.deferralErr
}
return r.WarmupLineageRepository.RescheduleWarmupTask(ctx, id, at)
}
type failedWindowScheduler struct{ scheduler.SchedulerService }
func (failedWindowScheduler) WarmupDispatchNotBefore(context.Context, uuid.UUID, time.Time) (time.Time, error) {
return time.Time{}, context.DeadlineExceeded
}
func TestLiveWarmupDispatchRechecksOpeningSourceAndExactRecipient(t *testing.T) {
f := newPartnerRoutingFixture(t)
ctx := t.Context()
@@ -23,6 +23,7 @@ type warmupRecoveryRepository interface {
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)
f.svc.taskRepo = repository.NewTaskRepository(f.pool)
f.svc.warmupHealth = nil
pending := uuid.New()