From d152c0778a3f77801cedff2e8938874ee1ce300e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 10:48:15 +0000 Subject: [PATCH] feat: atomically validate same-day warmup hours and defer invalid configured windows before claiming or sending while preserving visible safety holds --- .github/workflows/ci.yml | 4 +- cmd/warmblyctl/remote_output.go | 2 +- docs/content/docs/api/endpoints.mdx | 2 + docs/content/docs/guides/warmup.mdx | 2 + internal/repository/pg_email.go | 22 ++++ internal/repository/pg_monitoring.go | 1 + .../warmup_time_window_live_test.go | 124 ++++++++++++++++++ internal/scheduler/warmup_dispatch.go | 4 +- internal/scheduler/warmup_scheduler_test.go | 28 ++++ internal/tasks/email_task.go | 7 + internal/tasks/warmup_dispatch_live_test.go | 83 ++++++++++++ .../warmup_dispatch_recovery_live_test.go | 1 + 12 files changed, 276 insertions(+), 4 deletions(-) create mode 100644 internal/repository/warmup_time_window_live_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b4615b524..cc94b110c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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) diff --git a/cmd/warmblyctl/remote_output.go b/cmd/warmblyctl/remote_output.go index b5755ee30..e7ad9e866 100644 --- a/cmd/warmblyctl/remote_output.go +++ b/cmd/warmblyctl/remote_output.go @@ -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 diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index dbce521f4..b502f6d9b 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -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` | diff --git a/docs/content/docs/guides/warmup.mdx b/docs/content/docs/guides/warmup.mdx index 1edb920bf..5365eb53b 100644 --- a/docs/content/docs/guides/warmup.mdx +++ b/docs/content/docs/guides/warmup.mdx @@ -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: diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index fa52cb089..d540fd085 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -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 { diff --git a/internal/repository/pg_monitoring.go b/internal/repository/pg_monitoring.go index d637a76a3..080e9e093 100644 --- a/internal/repository/pg_monitoring.go +++ b/internal/repository/pg_monitoring.go @@ -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'`}, diff --git a/internal/repository/warmup_time_window_live_test.go b/internal/repository/warmup_time_window_live_test.go new file mode 100644 index 000000000..25832fb2e --- /dev/null +++ b/internal/repository/warmup_time_window_live_test.go @@ -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