From b2a89f8ce4fda7d4888d48095dc021fa8d06dfb3 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Mon, 19 Oct 2026 11:48:47 +0200 Subject: [PATCH] feat: stamp cloud_link_mailboxes.settings_observed_at before mirroring Warmbly Cloud warmup settings so an older report cannot overwrite a newer one, give each cloud link its own deadline in the settings sync, keep drawer edits made during a save or reverted to the stored value, and correct the cloud_warmup_not_applied docs --- docs/content/docs/api/error-codes.mdx | 2 +- docs/content/docs/guides/warmbly-cloud.mdx | 2 +- internal/app/cloudlink/warmup_settings.go | 64 +++++++++++-------- internal/app/consumer/cloud_settings_sync.go | 3 +- internal/repository/cloud_warmup_mirror.go | 47 ++++++++------ .../components/app/emails/InboxDetails.tsx | 8 ++- 6 files changed, 77 insertions(+), 49 deletions(-) diff --git a/docs/content/docs/api/error-codes.mdx b/docs/content/docs/api/error-codes.mdx index 5b95536b6..b6d747d1b 100644 --- a/docs/content/docs/api/error-codes.mdx +++ b/docs/content/docs/api/error-codes.mdx @@ -189,7 +189,7 @@ Each registered key may make 30 protected calls per hour. Past that the call ans #### Warmbly Cloud warmup refusals -A mailbox warmed by Warmbly Cloud has its warmup settings changed through the cloud (see [warmup settings](/guides/warmbly-cloud/#warmup-settings)). `PATCH /emails/:id` and the warmup start, pause and resume routes answer with these when that cannot happen. In every case nothing is saved on the instance. A replay with the same `Idempotency-Key` returns the stored answer, so retry with a new key. +A mailbox warmed by Warmbly Cloud has its warmup settings changed through the cloud (see [warmup settings](/guides/warmbly-cloud/#warmup-settings)). `PATCH /emails/:id` and the warmup start, pause and resume routes answer with these when that cannot happen. In every case the rest of the patch is not applied on the instance. After `cloud_warmup_not_applied`, the mailbox's copy of its Warmbly Cloud settings is still refreshed to what Warmbly Cloud reported. A replay with the same `Idempotency-Key` returns the stored answer, so retry with a new key. | `code` | Status | Meaning | |--------|--------|---------| diff --git a/docs/content/docs/guides/warmbly-cloud.mdx b/docs/content/docs/guides/warmbly-cloud.mdx index 89a45b106..8abe38474 100644 --- a/docs/content/docs/guides/warmbly-cloud.mdx +++ b/docs/content/docs/guides/warmbly-cloud.mdx @@ -165,7 +165,7 @@ Pause and resume per mailbox from the same row. Once a mailbox is enrolled, its Warmbly Cloud owns an enrolled mailbox's starting volume, daily increase, maximum, reply rate, sending window and sending days. Your instance keeps a copy of them on the mailbox, so the Warmup tab, the API and campaign estimates show what Warmbly Cloud actually runs. - The first enrollment proposes the settings the mailbox has on your instance. Enrolling it again, or refreshing its credential after a password change, never sends them again, so neither can undo a change made in Warmbly Cloud. -- Changing a setting on the Warmup tab, or through `PATCH /emails/:id`, sends it to Warmbly Cloud first. It is saved only once Warmbly Cloud has applied it, and the mailbox then shows the values Warmbly Cloud reports. If Warmbly Cloud cannot be reached, or does not apply a value, nothing is saved and the error names what was not applied. +- Changing a setting on the Warmup tab, or through `PATCH /emails/:id`, sends it to Warmbly Cloud first. It is saved only once Warmbly Cloud has applied it, and the mailbox then shows the values Warmbly Cloud reports. If Warmbly Cloud cannot be reached, nothing changes. If it answers but does not apply a value, the rest of your change is not saved, the error names what was not applied, and the mailbox shows the values Warmbly Cloud reported. - The copy is refreshed whenever the Cloud mailbox list loads, after every change, and every 15 minutes in the background, so a change made in Warmbly Cloud shows up here within 15 minutes. The Warmup tab shows when it was last refreshed. - The timezone stays your instance's, because campaigns use it too. It travels with every change, and the background refresh sends it again whenever Warmbly Cloud's differs. - Warmbly Cloud does not take 0 from an instance for the starting volume, maximum, daily increase or reply rate, so on an enrolled mailbox those fields start at 1. Sending every day is sent as all seven days. diff --git a/internal/app/cloudlink/warmup_settings.go b/internal/app/cloudlink/warmup_settings.go index e4b713af3..d509154aa 100644 --- a/internal/app/cloudlink/warmup_settings.go +++ b/internal/app/cloudlink/warmup_settings.go @@ -13,6 +13,9 @@ import ( "github.com/warmbly/warmbly/internal/repository" ) +// settingsSyncLinkTimeout bounds one link's listing, timezone pushes and mirror write. +const settingsSyncLinkTimeout = time.Minute + var ( errSettingsUnreported = errx.NewWithIdentifier(errx.ServiceUnavailable, "cloud_link_bad_response", "Warmbly Cloud did not report this mailbox's warmup settings. Try again in a moment.") errLinkDisconnecting = errx.NewWithIdentifier(errx.Conflict, "cloud_link_disconnecting", "This workspace is disconnecting from Warmbly Cloud. Try again once it has finished.") @@ -159,37 +162,46 @@ func (s *service) SyncWarmupSettings(ctx context.Context) ([]uuid.UUID, *errx.Er if len(group) == 0 || l.DisconnectPending { continue } - observedAt := time.Now() - var states []models.PoolLinkMailboxState - if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/mailboxes", nil, &states); xerr != nil { - log.Warn().Str("instance_id", l.InstanceID.String()).Str("code", xerr.Identifier).Msg("cloud link: warmup settings could not be refreshed") - lastErr = xerr - continue - } - byRemote := make(map[uuid.UUID]*models.PoolLinkMailboxState, len(states)) - for j := range states { - byRemote[states[j].RemoteID] = &states[j] - } - reported := map[uuid.UUID]models.PoolLinkWarmupSettings{} - for _, m := range group { - state := byRemote[m.RemoteID] - if m.EnrollmentState == "pending_remove" || state == nil || !state.Settings.Reported() { - continue - } - reported[m.EmailAccountID] = state.Settings - s.pushTimezone(ctx, l, m, state.Settings.Timezone) - } - ids, err := s.mirror(ctx, observedAt, reported) - if err != nil { - log.Warn().Err(err).Str("instance_id", l.InstanceID.String()).Msg("cloud link: warmup settings could not be mirrored") - lastErr = errx.InternalError() - continue - } + ids, xerr := s.syncLinkSettings(ctx, l, group) changed = append(changed, ids...) + if xerr != nil { + lastErr = xerr + } } return changed, lastErr } +// syncLinkSettings mirrors one link's mailboxes under its own deadline, so a slow link cannot starve the rest. +func (s *service) syncLinkSettings(ctx context.Context, l *models.CloudLink, group []models.CloudLinkMailbox) ([]uuid.UUID, *errx.Error) { + ctx, cancel := context.WithTimeout(ctx, settingsSyncLinkTimeout) + defer cancel() + observedAt := time.Now() + var states []models.PoolLinkMailboxState + if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/mailboxes", nil, &states); xerr != nil { + log.Warn().Str("instance_id", l.InstanceID.String()).Str("code", xerr.Identifier).Msg("cloud link: warmup settings could not be refreshed") + return nil, xerr + } + byRemote := make(map[uuid.UUID]*models.PoolLinkMailboxState, len(states)) + for j := range states { + byRemote[states[j].RemoteID] = &states[j] + } + reported := map[uuid.UUID]models.PoolLinkWarmupSettings{} + for _, m := range group { + state := byRemote[m.RemoteID] + if m.EnrollmentState == "pending_remove" || state == nil || !state.Settings.Reported() { + continue + } + reported[m.EmailAccountID] = state.Settings + s.pushTimezone(ctx, l, m, state.Settings.Timezone) + } + ids, err := s.mirror(ctx, observedAt, reported) + if err != nil { + log.Warn().Err(err).Str("instance_id", l.InstanceID.String()).Msg("cloud link: warmup settings could not be mirrored") + return nil, errx.InternalError() + } + return ids, nil +} + // pushTimezone keeps the cloud reading the warmup window in the zone this instance owns. func (s *service) pushTimezone(ctx context.Context, l *models.CloudLink, m models.CloudLinkMailbox, cloudTZ string) { acc, xerr := s.emails.GetByID(ctx, m.EmailAccountID) diff --git a/internal/app/consumer/cloud_settings_sync.go b/internal/app/consumer/cloud_settings_sync.go index 626c2598d..97180fca8 100644 --- a/internal/app/consumer/cloud_settings_sync.go +++ b/internal/app/consumer/cloud_settings_sync.go @@ -27,7 +27,8 @@ func (s *JobsService) StartCloudSettingsSync(ctx context.Context, interval time. } func (s *JobsService) syncCloudSettings(ctx context.Context, syncer CloudSettingsSyncer) error { - runCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) + // Each link has its own deadline inside the sync; this only bounds the whole pass below the interval. + runCtx, cancel := context.WithTimeout(ctx, 10*time.Minute) defer cancel() changed, xerr := syncer.SyncWarmupSettings(runCtx) for _, id := range changed { diff --git a/internal/repository/cloud_warmup_mirror.go b/internal/repository/cloud_warmup_mirror.go index 448b18310..2bd8348dc 100644 --- a/internal/repository/cloud_warmup_mirror.go +++ b/internal/repository/cloud_warmup_mirror.go @@ -5,6 +5,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/infrastructure/db" "github.com/warmbly/warmbly/internal/models" @@ -49,41 +50,49 @@ func (r *cloudLinkRepository) MirrorCloudWarmupSettings(ctx context.Context, obs return nil, err } defer tx.Rollback(ctx) - // observedAt orders a slow listing behind a newer relay without holding the reconciliation lock. + // Stamping first locks each observation row, so an older report racing a newer one is re-checked and dropped. + stamp := `UPDATE cloud_link_mailboxes SET settings_observed_at = $1 + WHERE email_account_id = ANY($2::uuid[]) AND enrollment_state <> 'pending_remove' + AND (settings_observed_at IS NULL OR settings_observed_at < $1) + RETURNING email_account_id` + accepted, err := scanIDs(tx.Query(ctx, stamp, observedAt, ids)) + if err != nil { + db.CaptureError(err, stamp, nil, "query") + return nil, err + } + if len(accepted) == 0 { + return nil, tx.Commit(ctx) + } query := ` UPDATE email_accounts ea SET warmup_base = s.base, warmup_max = s.max, warmup_increase = s.inc, warmup_reply_rate = s.reply, warmup_start_time = s.start_at::time, warmup_end_time = s.end_at::time, warmup_days = s.days, updated_at = now() - FROM unnest($2::uuid[], $3::int[], $4::int[], $5::int[], $6::smallint[], $7::text[], $8::text[], $9::smallint[]) + FROM unnest($1::uuid[], $2::int[], $3::int[], $4::int[], $5::smallint[], $6::text[], $7::text[], $8::smallint[]) AS s(id, base, max, inc, reply, start_at, end_at, days) - JOIN cloud_link_mailboxes c ON c.email_account_id = s.id - WHERE ea.id = s.id AND c.enrollment_state <> 'pending_remove' - AND (c.settings_observed_at IS NULL OR c.settings_observed_at < $1) + WHERE ea.id = s.id AND s.id = ANY($9::uuid[]) AND (ea.warmup_base, ea.warmup_max, ea.warmup_increase, ea.warmup_reply_rate, ea.warmup_start_time, ea.warmup_end_time, ea.warmup_days) IS DISTINCT FROM (s.base, s.max, s.inc, s.reply, s.start_at::time, s.end_at::time, s.days) RETURNING ea.id` - rows, err := tx.Query(ctx, query, observedAt, ids, bases, maxes, increases, replies, starts, ends, days) + changed, err := scanIDs(tx.Query(ctx, query, ids, bases, maxes, increases, replies, starts, ends, days, accepted)) if err != nil { db.CaptureError(err, query, nil, "query") return nil, err } - var changed []uuid.UUID + return changed, tx.Commit(ctx) +} + +func scanIDs(rows pgx.Rows, err error) ([]uuid.UUID, error) { + 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 { - rows.Close() return nil, err } - changed = append(changed, id) + ids = append(ids, id) } - rows.Close() - if err := rows.Err(); err != nil { - return nil, err - } - if _, err := tx.Exec(ctx, `UPDATE cloud_link_mailboxes SET settings_observed_at = $1 - WHERE email_account_id = ANY($2::uuid[]) AND enrollment_state <> 'pending_remove' - AND (settings_observed_at IS NULL OR settings_observed_at < $1)`, observedAt, ids); err != nil { - return nil, err - } - return changed, tx.Commit(ctx) + return ids, rows.Err() } diff --git a/web/src/components/app/emails/InboxDetails.tsx b/web/src/components/app/emails/InboxDetails.tsx index 306e7ac45..d58a5df66 100644 --- a/web/src/components/app/emails/InboxDetails.tsx +++ b/web/src/components/app/emails/InboxDetails.tsx @@ -424,7 +424,11 @@ function Detail({ mailbox, onClose, initialTab, tabIntentKey, onTabIntentConsume // Fields the user has edited since the last save; a refresh never overwrites them. const touched = React.useRef(new Set()); const update = (patch: Partial) => { - for (const k of Object.keys(patch)) touched.current.add(k as keyof Inbox); + // A field put back to the stored value is untouched again, so a refresh can still reach it. + for (const k of Object.keys(patch) as (keyof Inbox)[]) { + if (JSON.stringify(patch[k]) === JSON.stringify(mailbox[k])) touched.current.delete(k); + else touched.current.add(k); + } setForm((f) => ({ ...f, ...patch })); }; // A refetch carries newer values into the untouched fields, so Save never writes an older one back. @@ -476,6 +480,8 @@ function Detail({ mailbox, onClose, initialTab, tabIntentKey, onTabIntentConsume setForm((f) => { const next = { ...f } as Record; for (const k of Object.keys(patch) as (keyof Inbox)[]) { + // A field edited again while the save was in flight keeps the newer value. + if (JSON.stringify(f[k]) !== JSON.stringify(patch[k])) continue; touched.current.delete(k); next[k] = saved[k]; }