From 35fbf733478dd1dc116f034381d6e6b7303bc589 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Wed, 3 Jun 2026 11:14:02 +0200 Subject: [PATCH] feat: persist warmup engagement dwell Store delayed warmup engagement actions in Postgres and drain them from the consumer so read, important, and star actions survive worker restarts. Keep foldering and spam rescue immediate while routing delayed actions to the mailbox's current worker at fire time. --- cmd/consumer/main.go | 6 ++ internal/app/consumer/event_new_email.go | 66 +++++++++++---- internal/app/consumer/service.go | 1 + internal/app/consumer/warmup_engagement.go | 17 ++++ .../app/consumer/warmup_engagement_poller.go | 66 +++++++++++++++ ...000008_warmup_pending_engagements.down.sql | 1 + .../000008_warmup_pending_engagements.up.sql | 23 ++++++ internal/repository/pg_warmup_engagement.go | 81 +++++++++++++++++++ 8 files changed, 247 insertions(+), 14 deletions(-) create mode 100644 internal/app/consumer/warmup_engagement_poller.go create mode 100644 internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql create mode 100644 internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql create mode 100644 internal/repository/pg_warmup_engagement.go diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 57ba736f..cb4eea6b 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -240,6 +240,7 @@ func main() { EmailAccountErrorRepository: emailAccountErrorRepo, WarmupRepo: warmupRepo, WarmupContentRepo: repository.NewWarmupContentRepository(primaryDB.Pool), + WarmupEngagementRepo: repository.NewWarmupEngagementRepository(primaryDB.Pool), WarmupService: warmupService, WorkerRepo: workerRepo, Publisher: eventsPublisher, @@ -268,6 +269,11 @@ func main() { // Start warmup health evaluation sweep (every hour) go jobsService.StartWarmupHealthSweep(ctx, 1*time.Hour) + // Drains the durable delayed-engagement schedule (read/important/star) so the + // recipient-side dwell survives worker restarts. Short interval keeps the + // effective dwell close to the requested value. + go jobsService.StartWarmupEngagementPoller(ctx, 30*time.Second) + // Start dead worker detection (every 5 minutes) go jobsService.StartDeadWorkerDetection(ctx, 5*time.Minute) diff --git a/internal/app/consumer/event_new_email.go b/internal/app/consumer/event_new_email.go index 890e5720..28d54a0c 100644 --- a/internal/app/consumer/event_new_email.go +++ b/internal/app/consumer/event_new_email.go @@ -2,9 +2,11 @@ package jobs import ( "context" + "encoding/json" "fmt" "slices" "strings" + "time" "github.com/google/uuid" "github.com/rs/zerolog/log" @@ -151,31 +153,67 @@ func (s *JobsService) performWarmupActions(ctx context.Context, e *models.JobEve settings := s.getGenerationSettings(ctx) actions, delaySeconds := engagementPlan(e.Message.EmailID, settings.Engagement) + immediate, delayed := splitEngagementLegs(actions) - action := &models.WarmupEmailAction{ + base := models.WarmupEmailAction{ UserID: e.UserID, EmailID: e.Message.EmailID, GmailID: e.Message.GmailID, UID: e.Message.UID, MailboxUIDValidity: e.Message.Mailbox, - Actions: actions, - DelaySeconds: delaySeconds, } - // Look up the worker ID from the email account + // Resolve the receiving mailbox's worker once. + var workerID *uuid.UUID if s.EmailRepository != nil { - account, xerr := s.EmailRepository.GetByID(ctx, e.Message.EmailID) - if xerr == nil && account != nil && account.WorkerID != nil { - s.Publisher.PublishWarmupAction(ctx, *account.WorkerID, action) - } else if account != nil && account.WorkerID == nil { - // No assigned worker (mid-migration / just-unassigned / assignment - // lag): the warmup mail can't be foldered or engaged with. Log it - // instead of dropping silently so the gap is observable. - log.Warn(). - Str("email_id", e.Message.EmailID.String()). - Msg("Warmup actions skipped: recipient mailbox has no assigned worker") + if account, xerr := s.EmailRepository.GetByID(ctx, e.Message.EmailID); xerr == nil && account != nil { + workerID = account.WorkerID } } + if workerID == nil { + // No assigned worker (mid-migration / just-unassigned / assignment lag): + // the warmup mail can't be foldered or engaged with. Log instead of + // dropping silently so the gap is observable. + log.Warn(). + Str("email_id", e.Message.EmailID.String()). + Msg("Warmup actions skipped: recipient mailbox has no assigned worker") + return + } + + // Immediate, durable leg (folder + spam-rescue): publish to the worker now. + if len(immediate) > 0 { + act := base + act.Actions = immediate + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + } + + if len(delayed) == 0 { + return + } + + act := base + act.Actions = delayed + + // Delayed leg (read / important / star): with no dwell (or no durable store + // available) publish immediately; otherwise persist it to the durable + // schedule so a worker restart mid-dwell can't drop it. The poller publishes + // it when fire_at passes. + if delaySeconds <= 0 || s.WarmupEngagementRepo == nil { + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + return + } + + payload, err := json.Marshal(act) + if err != nil { + log.Warn().Err(err).Str("email_id", e.Message.EmailID.String()).Msg("Failed to marshal delayed warmup engagement; publishing immediately") + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + return + } + fireAt := time.Now().Add(time.Duration(delaySeconds) * time.Second) + if err := s.WarmupEngagementRepo.EnqueuePendingEngagement(ctx, e.Message.EmailID, payload, fireAt); err != nil { + log.Warn().Err(err).Str("email_id", e.Message.EmailID.String()).Msg("Failed to enqueue delayed warmup engagement; publishing immediately") + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + } } // recipientProviderDomain best-effort resolves a recipient mailbox's provider diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index e6168062..e71d5b5d 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -23,6 +23,7 @@ type JobsService struct { EmailAccountErrorRepository repository.EmailAccountErrorRepository WarmupRepo repository.WarmupRepository WarmupContentRepo repository.WarmupContentRepository + WarmupEngagementRepo repository.WarmupEngagementRepository WarmupService warmupapp.Service WorkerRepo repository.WorkerRepository diff --git a/internal/app/consumer/warmup_engagement.go b/internal/app/consumer/warmup_engagement.go index 2ff8b4c6..111183fb 100644 --- a/internal/app/consumer/warmup_engagement.go +++ b/internal/app/consumer/warmup_engagement.go @@ -83,6 +83,23 @@ func engagementPlan(accountID uuid.UUID, e models.WarmupEngagementSettings) (act return actions, delaySeconds } +// splitEngagementLegs separates the reputation-critical, durable actions +// (foldering + spam-rescue, published immediately) from the low-stakes +// engagement-timing signals (read / important / star) that carry the +// recipient-side dwell. The dwell is now applied durably in the control plane +// (a fire_at row drained by the poller), not by an in-process worker timer, so +// a worker restart can no longer drop the delayed leg. +func splitEngagementLegs(actions []string) (immediate, delayed []string) { + for _, a := range actions { + if a == "move_to_warmbly" || a == "remove_from_spam" { + immediate = append(immediate, a) + } else { + delayed = append(delayed, a) + } + } + return immediate, delayed +} + // rollPct rolls a biased percentage chance. The persona bias nudges a given // mailbox consistently above/below the configured rate so mailboxes differ. func rollPct(rate int, bias float64) bool { diff --git a/internal/app/consumer/warmup_engagement_poller.go b/internal/app/consumer/warmup_engagement_poller.go new file mode 100644 index 00000000..503e662e --- /dev/null +++ b/internal/app/consumer/warmup_engagement_poller.go @@ -0,0 +1,66 @@ +package jobs + +import ( + "context" + "encoding/json" + "time" + + "github.com/rs/zerolog/log" + + "github.com/warmbly/warmbly/internal/models" +) + +// StartWarmupEngagementPoller drains due delayed-engagement rows and publishes +// them to the worker. This is the durable replacement for the worker's old +// in-process dwell timer: because the schedule lives in Postgres, a worker (or +// consumer) restart can no longer drop the delayed read/important/star signals. +func (s *JobsService) StartWarmupEngagementPoller(ctx context.Context, interval time.Duration) { + if s.WarmupEngagementRepo == nil || s.Publisher == nil { + return + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.drainDueEngagements(ctx) + } + } +} + +func (s *JobsService) drainDueEngagements(ctx context.Context) { + cctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + + due, err := s.WarmupEngagementRepo.ClaimDuePendingEngagements(cctx, 200) + if err != nil { + log.Warn().Err(err).Msg("warmup engagement poller: claim failed") + return + } + + for _, p := range due { + var action models.WarmupEmailAction + if err := json.Unmarshal(p.Payload, &action); err != nil { + log.Warn().Err(err).Str("id", p.ID.String()).Msg("warmup engagement poller: bad payload, dropping") + continue + } + + // Re-resolve the worker at fire time so a mid-dwell reassignment routes + // to the current worker (the payload deliberately doesn't bake one in). + if s.EmailRepository == nil { + continue + } + account, xerr := s.EmailRepository.GetByID(cctx, action.EmailID) + if xerr != nil || account == nil || account.WorkerID == nil { + // Mailbox now unassigned — drop (best-effort low-stakes engagement). + continue + } + + action.DelaySeconds = 0 // dwell already elapsed; run immediately + s.Publisher.PublishWarmupAction(cctx, *account.WorkerID, &action) + } +} diff --git a/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql new file mode 100644 index 00000000..5ad6d30b --- /dev/null +++ b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS public.warmup_pending_engagements; diff --git a/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql new file mode 100644 index 00000000..14ffff42 --- /dev/null +++ b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql @@ -0,0 +1,23 @@ +-- Durable dwell for delayed warmup engagement actions. +-- +-- The recipient-side "dwell" delay for the low-stakes engagement signals +-- (mark_read / mark_important / star) previously lived only in a worker-process +-- time.AfterFunc timer, so a worker restart mid-dwell dropped those signals with +-- no trace. This table is the durable schedule: the consumer enqueues the +-- delayed leg here with a fire_at, and a consumer-side poller publishes it to +-- the worker when due. The reputation-critical leg (folder + spam-rescue) is +-- still published immediately and is unaffected. +-- +-- Control-plane only: written and drained by the consumer (Postgres-backed), +-- never by the worker. + +CREATE TABLE public.warmup_pending_engagements ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + email_account_id uuid NOT NULL, + payload jsonb NOT NULL, + fire_at timestamp with time zone NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT warmup_pending_engagements_pkey PRIMARY KEY (id) +); + +CREATE INDEX idx_warmup_pending_engagements_due ON public.warmup_pending_engagements USING btree (fire_at); diff --git a/internal/repository/pg_warmup_engagement.go b/internal/repository/pg_warmup_engagement.go new file mode 100644 index 00000000..ce527a1c --- /dev/null +++ b/internal/repository/pg_warmup_engagement.go @@ -0,0 +1,81 @@ +package repository + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// PendingEngagement is a delayed warmup engagement action awaiting its dwell. +// Payload is a JSON-encoded models.WarmupEmailAction (the delayed leg, with +// DelaySeconds already consumed — the worker runs it immediately on receipt). +type PendingEngagement struct { + ID uuid.UUID + EmailAccountID uuid.UUID + Payload []byte + FireAt time.Time +} + +// WarmupEngagementRepository is the durable schedule for dwell-delayed warmup +// engagement actions, so a worker restart can't drop them (the old in-process +// timer did). Control-plane only; drained by the consumer-side poller. +type WarmupEngagementRepository interface { + // EnqueuePendingEngagement stores a delayed engagement leg to fire at fireAt. + EnqueuePendingEngagement(ctx context.Context, accountID uuid.UUID, payload []byte, fireAt time.Time) error + // ClaimDuePendingEngagements atomically removes and returns up to limit rows + // whose fire_at has passed, so each is delivered exactly once across pollers. + ClaimDuePendingEngagements(ctx context.Context, limit int) ([]PendingEngagement, error) +} + +type warmupEngagementRepository struct { + db *pgxpool.Pool +} + +// NewWarmupEngagementRepository creates a new warmup engagement repository. +func NewWarmupEngagementRepository(db *pgxpool.Pool) WarmupEngagementRepository { + return &warmupEngagementRepository{db: db} +} + +func (r *warmupEngagementRepository) EnqueuePendingEngagement(ctx context.Context, accountID uuid.UUID, payload []byte, fireAt time.Time) error { + _, err := r.db.Exec(ctx, + `INSERT INTO warmup_pending_engagements (email_account_id, payload, fire_at) VALUES ($1, $2, $3)`, + accountID, payload, fireAt) + return err +} + +// ClaimDuePendingEngagements uses DELETE ... RETURNING over a FOR UPDATE SKIP +// LOCKED subselect so concurrent pollers never claim the same row and a claimed +// row is removed in the same statement (delivered at most once). +func (r *warmupEngagementRepository) ClaimDuePendingEngagements(ctx context.Context, limit int) ([]PendingEngagement, error) { + if limit <= 0 { + limit = 100 + } + query := ` + DELETE FROM warmup_pending_engagements + WHERE id IN ( + SELECT id FROM warmup_pending_engagements + WHERE fire_at <= NOW() + ORDER BY fire_at + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, email_account_id, payload, fire_at + ` + rows, err := r.db.Query(ctx, query, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []PendingEngagement + for rows.Next() { + var p PendingEngagement + if err := rows.Scan(&p.ID, &p.EmailAccountID, &p.Payload, &p.FireAt); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +}