From bd435a2687b5020787143784633093b723c71e63 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 00:40:39 -0700 Subject: [PATCH 1/2] feat: tell the worker to drop a mailbox the customer disabled or disconnected, the two paths #217 left out: emailService.Update now publishes REMOVE_EMAIL when a status leaves active (and re-ships the mailbox when it returns to active instead of waiting on the reconciler), Delete publishes it before the row goes and refuses to delete when it cannot, since after the row there is no assignment left to read; delete also stopped failing outright, because tasks and warmup_admin_actions referenced email_accounts with no delete action so disconnecting anything that had ever warmed up or sent a step raised a foreign key violation (migration 000098), read the mailbox through an org-scoped query while passing a user id so the warmup pool cleanup, realtime event and webhook never ran, and never refunded the worker's account count or load score --- docs/content/docs/api/error-codes.mdx | 16 + docs/content/docs/guides/mailboxes.mdx | 12 + internal/app/email/handler.go | 86 +++- internal/app/email/loader.go | 42 ++ .../app/email/worker_removal_live_test.go | 256 ++++++++++ internal/app/email/worker_removal_test.go | 438 ++++++++++++++++++ internal/errx/common.go | 6 + .../000098_mailbox_delete_cascades.down.sql | 9 + .../000098_mailbox_delete_cascades.up.sql | 16 + web/src/app/app/emails/page.tsx | 18 +- 10 files changed, 893 insertions(+), 6 deletions(-) create mode 100644 internal/app/email/worker_removal_live_test.go create mode 100644 internal/app/email/worker_removal_test.go create mode 100644 internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.down.sql create mode 100644 internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.up.sql diff --git a/docs/content/docs/api/error-codes.mdx b/docs/content/docs/api/error-codes.mdx index d9da143e..8b909fbd 100644 --- a/docs/content/docs/api/error-codes.mdx +++ b/docs/content/docs/api/error-codes.mdx @@ -306,6 +306,22 @@ A `503` whose `code` is `mailbox_provider_not_configured` is not transient and r - Or connect the mailbox over SMTP and IMAP instead, which needs no configuration - Full walkthrough: [connect mailboxes](/development/deployment-guide/#connect-mailboxes) +#### `mailbox_worker_unreachable` + +A `503` whose `code` is `mailbox_worker_unreachable` comes from `DELETE /emails/{id}`. Disconnecting a mailbox has to reach the machine that syncs it before the record goes, because once the record is gone nothing can tell that machine to stop. When the instruction cannot be delivered, nothing is removed and the mailbox is left exactly as it was. + +```json +{ + "error": "Service Unavailable", + "message": "This mailbox could not be disconnected right now because the machine syncing it could not be reached. Nothing was removed, so try again in a moment.", + "code": "mailbox_worker_unreachable", + "request_id": "4bbbd1b2-8f86-47dd-8a7f-9476501ad20e" +} +``` + +**How to fix:** +- Retry the delete. It is safe to repeat: a mailbox that is already gone returns `404`, and one that is still there is untouched + ## Error handling best practices ### Implement retry logic diff --git a/docs/content/docs/guides/mailboxes.mdx b/docs/content/docs/guides/mailboxes.mdx index 28793537..c0d656b9 100644 --- a/docs/content/docs/guides/mailboxes.mdx +++ b/docs/content/docs/guides/mailboxes.mdx @@ -102,6 +102,18 @@ Keep warmup running after campaigns begin rather than switching it off once a ma The model still supports separate free and premium pools, but access is gated to paid workspaces today. +## Pausing and disconnecting + +A mailbox can be set inactive (`PATCH /emails/{id}` with `status`) when you want it to stop without losing its settings, its history, or its place in the fleet. + +Switching it off takes effect immediately: the machine syncing it is told to drop it, so it stops importing mail and stops being picked for campaign sends and warmup within seconds rather than at that machine's next restart. Warmup pool membership is dropped, and any warmup chain it had winds down on its next step. A campaign email already handed over is answered as a failure and the step is retried later on a mailbox that is still active, so nobody receives it twice and no lead is stranded. + +It keeps its worker assignment while off, so switching it back on puts it back on the same machine, sending from the same IP, and it resumes syncing from where it stopped instead of re-importing. + +**Disconnecting** removes the mailbox for good, from the Accounts selection bar. The machine syncing it is told to drop it before the record is removed, because afterwards there is nothing left to tell. If that instruction cannot be delivered, the disconnect fails with a `503` and nothing is removed, so retry it in a moment rather than assuming it worked. + +Everything belonging to that mailbox goes with it: its imported mail in the unibox, its warmup history and pool membership, its credentials, its sender links, and any send still scheduled for it. A campaign that was using it keeps running on its remaining senders. Export the workspace first if you want a copy. Disable the mailbox instead when you only want it to stop. + ## Worker assignment You never pick a worker. Warmbly assigns each mailbox to a sending worker automatically and can move it later. Workers are the machines that send and sync, so spreading mailboxes across them spreads sending across network identities and IPs. diff --git a/internal/app/email/handler.go b/internal/app/email/handler.go index 7a97f8bc..3325155a 100644 --- a/internal/app/email/handler.go +++ b/internal/app/email/handler.go @@ -7,6 +7,7 @@ import ( "time" "github.com/google/uuid" + "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/infrastructure/pubsub" @@ -50,10 +51,38 @@ func (s *emailService) Update(ctx context.Context, userID, emailAccountID string } s.syncWarmupPoolMembership(ctx, account) + s.applyStatusToWorker(ctx, userID, account, udata.Status) s.publishAccountEvent(ctx, pubsub.EventAccountSynced, account) return account, nil } +// applyStatusToWorker carries a mailbox status change through to the machine +// running it. Only the row was written before this, so a mailbox the customer +// switched off kept syncing and sending on its old schedule until its worker +// restarted, and one switched back on waited for the reconciler's next pass. +// +// Best-effort in both directions, unlike the delete path: the status is +// already written, no load path ships a mailbox that is not active, and the +// reconciler re-places an active one either way. +func (s *emailService) applyStatusToWorker(ctx context.Context, userID string, account *models.Email, status *string) { + if account == nil || status == nil { + return + } + + if *status == "active" { + s.loadAccountBestEffort(ctx, account.ID) + return + } + + if xerr := s.dropFromWorker(ctx, userID, account.ID); xerr != nil { + log.Warn(). + Str("error", xerr.Message). + Str("email_id", account.ID.String()). + Str("status", *status). + Msg("could not tell the worker to drop the disabled mailbox; it stops syncing when that worker next restarts") + } +} + // BulkUpdateTags is a pure tag-link rewrite: no warmup pool or worker state // depends on tags, so no per-account fanout is needed (the caller audits // once and the spine refreshes lists). @@ -242,20 +271,62 @@ func (s *emailService) resolveDomainAuth(ctx context.Context, userID, emailAccou return domain, &res, nil } +// Delete disconnects a mailbox for good. +// +// The worker holding it is told to drop it BEFORE the row goes, because after +// the delete there is no assignment left to read and no reconciler that can +// repair a missed removal: a worker that was never told keeps syncing an +// account that no longer exists and emits events the consumer resolves against +// nothing. That makes the removal part of the delete rather than a follow-up. +// When it cannot be sent, nothing is deleted and the customer can try again. func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string) *errx.Error { - account, xerr := s.emailRepository.Get(ctx, userID, emailAccountID) - if xerr != nil && xerr != errx.ErrNotFound { + accountID, err := uuid.Parse(emailAccountID) + if err != nil { + return errx.ErrUuid + } + + // Read by id rather than through Get: Get is scoped by organization and was + // being handed a user id, so it never found the mailbox and every side + // effect below was silently skipped. Ownership is checked here instead, + // against the same owner the delete itself filters on, so the removal + // published below is never reachable for someone else's mailbox. + account, xerr := s.emailRepository.GetByID(ctx, accountID) + if xerr != nil { + return xerr + } + if account == nil || !sameUser(account.UserID, userID) { + return errx.ErrNotFound + } + + if xerr := s.dropFromWorker(ctx, userID, accountID); xerr != nil { return xerr } + // Give the worker back the capacity this mailbox was counted for. Nothing + // else does it on delete: the row's worker_id is only nulled by the foreign + // key, so the worker's account count and load score would stay charged for + // a mailbox that no longer exists and its usable capacity would shrink with + // every disconnect. + if s.workerAssignment != nil && account.WorkerID != nil { + if err := s.workerAssignment.UnassignWorkerFromEmail(ctx, accountID); err != nil { + log.Warn().Err(err). + Str("email_id", accountID.String()). + Msg("could not release the deleted mailbox's worker capacity") + } + } + if xerr := s.emailRepository.Delete(ctx, userID, emailAccountID); xerr != nil { + // The removal has already gone out and the mailbox is still active, so + // put it back now rather than leaving it dark until the reconciler's + // next pass, minutes later. + s.loadAccountBestEffort(ctx, accountID) return xerr } s.removeFromAllWarmupPools(ctx, account) s.publishAccountEvent(ctx, pubsub.EventAccountDisconnected, account) - if s.webhookService != nil && account != nil && account.OrganizationID != nil { + if s.webhookService != nil && account.OrganizationID != nil { _, _ = s.webhookService.Dispatch(ctx, *account.OrganizationID, models.WebhookEventEmailAccountRemoved, map[string]any{ "email_account_id": account.ID, "email": account.Email, @@ -265,6 +336,15 @@ func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string return nil } +// sameUser compares two user ids as uuids, so a differently formatted but equal +// id is not read as a different owner (the delete itself compares in SQL, +// where Postgres does the same). +func sameUser(a, b string) bool { + left, aerr := uuid.Parse(a) + right, berr := uuid.Parse(b) + return aerr == nil && berr == nil && left == right +} + func (s *emailService) syncWarmupPoolMembership(ctx context.Context, account *models.Email) { if s.warmupService == nil || account == nil { return diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index 5f491e4b..dffd140d 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/app/instancesettings" + "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" "golang.org/x/oauth2" @@ -171,6 +172,47 @@ func (s *emailService) LoadAccountOntoWorker(ctx context.Context, accountID uuid return s.publisher.PublishAddEmail(ctx, *workerID, payload) } +// dropFromWorker tells the worker holding a mailbox to drop it from memory. +// +// Workers keep accounts in memory, and the per-worker load filters on status +// only when a worker starts, so a mailbox the customer disabled or disconnected +// keeps syncing on its old schedule until that worker restarts. This is the +// control-plane half of that: the same removal the consumer sends when a +// provider error deactivates a mailbox. +// +// The assignment is asked for on its own rather than read off an updated row: +// the mailbox as the dashboard sees it carries no worker_id, which is exactly +// what made the consumer's removal unreachable the first time. +// +// A send already dispatched to that worker is not lost: it is answered with +// EMAIL_FAILED, which walks the reservation back and lets routing retry the +// step on a mailbox that is still active. +func (s *emailService) dropFromWorker(ctx context.Context, userID string, accountID uuid.UUID) *errx.Error { + if s.publisher == nil { + return nil + } + + workerID, xerr := s.emailRepository.GetWorkerID(ctx, accountID) + if xerr != nil { + return xerr + } + if workerID == nil { + return nil + } + + if err := s.publisher.PublishRemoveEmail(ctx, *workerID, &models.RemoveWorkerEmail{ + UserID: userID, + EmailID: accountID.String(), + }); err != nil { + log.Warn().Err(err). + Str("email_id", accountID.String()). + Str("worker_id", workerID.String()). + Msg("could not tell the worker to drop the mailbox") + return errx.ErrEmailWorkerUnreachable + } + return nil +} + // releaseDeadWorker returns the worker a mailbox should load onto, releasing it // first when the one it holds can no longer receive anything. // diff --git a/internal/app/email/worker_removal_live_test.go b/internal/app/email/worker_removal_live_test.go new file mode 100644 index 00000000..b93100fc --- /dev/null +++ b/internal/app/email/worker_removal_live_test.go @@ -0,0 +1,256 @@ +package email + +import ( + "context" + "os" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/warmbly/warmbly/internal/app/worker" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/db" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +// Live cover for the disable and disconnect paths against real SQL. Skipped +// unless WARMBLY_TEST_DB is set, so `go test ./...` stays hermetic: +// +// WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \ +// go test ./internal/app/email/ -run Live -v +// +// Stubs cannot see any of what these check: the delete path used to read the +// mailbox through a query scoped by organization while handing it a user id, +// so it never found the row, and the worker assignment lives in a column no +// update returns. + +type removalLiveFixture struct { + pool *pgxpool.Pool + svc *emailService + pub *stubEventPublisher + user uuid.UUID + org uuid.UUID + mailbox uuid.UUID + worker uuid.UUID +} + +func newRemovalLiveFixture(t *testing.T) *removalLiveFixture { + t.Helper() + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + handle, err := db.New(context.Background(), dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(func() { handle.Pool.Close() }) + + ctx := context.Background() + f := &removalLiveFixture{pool: handle.Pool, user: uuid.New(), org: uuid.New(), mailbox: uuid.New(), worker: uuid.New()} + exec := func(sql string, args ...any) { + t.Helper() + if _, err := f.pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture: %v", err) + } + } + exec(`INSERT INTO users (id, email, first_name, last_name) VALUES ($1, $2, 'Drop', 'Test')`, + f.user, "drop-"+f.user.String()[:8]+"@test.local") + exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Drop Test', $2, $3)`, + f.org, "drop-"+f.org.String()[:8], f.user) + // One mailbox's worth of load: an smtp_imap mailbox that is not warming + // weighs 1.0, which is what the delete has to refund. + exec(`INSERT INTO workers (id, name, ip_addr, active, account_count, load_score) + VALUES ($1, 'drop-test', '127.0.0.1', true, 1, 1)`, f.worker) + exec(`INSERT INTO email_accounts (id, user_id, organization_id, worker_id, email, name, + signature_plain, signature_html, provider, status, campaign_limit, min_wait_time) + VALUES ($1, $2, $3, $4, $5, 'Drop', '', '', 'smtp_imap', 'active', 50, 600)`, + f.mailbox, f.user, f.org, f.worker, "drop-"+f.mailbox.String()[:8]+"@test.local") + + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM email_accounts WHERE id = $1`, f.mailbox}, + {`DELETE FROM workers WHERE id = $1`, f.worker}, + {`DELETE FROM organizations WHERE id = $1`, f.org}, + {`DELETE FROM users WHERE id = $1`, f.user}, + } { + if _, err := f.pool.Exec(c, step.sql, step.arg); err != nil { + t.Errorf("cleanup %q: %v", step.sql, err) + } + } + }) + + f.pub = &stubEventPublisher{} + f.svc = &emailService{ + emailRepository: repository.NewEmailRepostory(handle, nil), + publisher: f.pub, + workerAssignment: worker.NewAssignmentService(repository.NewWorkerRepository(f.pool), nil, nil), + } + return f +} + +func (f *removalLiveFixture) mailboxExists(t *testing.T) bool { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), + `SELECT count(*) FROM email_accounts WHERE id = $1`, f.mailbox).Scan(&n); err != nil { + t.Fatalf("read back: %v", err) + } + return n == 1 +} + +func (f *removalLiveFixture) workerLoad(t *testing.T) (int, float64) { + t.Helper() + var count int + var score float64 + if err := f.pool.QueryRow(context.Background(), + `SELECT account_count, load_score FROM workers WHERE id = $1`, f.worker).Scan(&count, &score); err != nil { + t.Fatalf("read worker: %v", err) + } + return count, score +} + +// Disabling writes the status AND tells the worker holding the mailbox, with +// the assignment read from the column rather than from the row the update +// returns (which does not carry one). +func TestLiveDisablingAMailboxRemovesItFromItsWorker(t *testing.T) { + f := newRemovalLiveFixture(t) + inactive := "inactive" + + account, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &inactive}) + if xerr != nil { + t.Fatalf("update: %v", xerr) + } + if account.Status != "inactive" { + t.Fatalf("status = %q, want inactive", account.Status) + } + if len(f.pub.removed) != 1 { + t.Fatalf("published %d removals, want 1", len(f.pub.removed)) + } + if f.pub.removed[0].workerID != f.worker || f.pub.removed[0].emailID != f.mailbox.String() { + t.Errorf("removal = %+v, want worker %s and mailbox %s", f.pub.removed[0], f.worker, f.mailbox) + } + + // The mailbox keeps its placement while disabled, so re-enabling puts it + // back on the same worker and the same sending IP. + var assigned *uuid.UUID + if err := f.pool.QueryRow(context.Background(), + `SELECT worker_id FROM email_accounts WHERE id = $1`, f.mailbox).Scan(&assigned); err != nil { + t.Fatalf("read assignment: %v", err) + } + if assigned == nil || *assigned != f.worker { + t.Errorf("assignment = %v, want it kept at %s", assigned, f.worker) + } +} + +// Deleting has to reach the worker BEFORE the row goes: afterwards there is no +// worker_id left to read. +func TestLiveDeletingAMailboxRemovesItFromItsWorkerFirst(t *testing.T) { + f := newRemovalLiveFixture(t) + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + + if len(f.pub.removed) != 1 { + t.Fatalf("published %d removals, want 1: the worker keeps syncing a mailbox that no longer exists", len(f.pub.removed)) + } + if f.pub.removed[0].workerID != f.worker { + t.Errorf("removal sent to worker %s, want %s", f.pub.removed[0].workerID, f.worker) + } + if f.mailboxExists(t) { + t.Error("the mailbox row survived the delete") + } + + // And the worker gets its capacity back, or every disconnect permanently + // shrinks what that machine can be given. + count, score := f.workerLoad(t) + if count != 0 || score != 0 { + t.Errorf("worker still charged for the deleted mailbox: account_count=%d load_score=%v", count, score) + } +} + +// The lookup that finds the mailbox is deliberately unscoped, so ownership is +// checked in the service. A teammate's user id must not delete this mailbox or +// publish a removal for it. +func TestLiveDeleteRefusesAMailboxTheCallerDoesNotOwn(t *testing.T) { + f := newRemovalLiveFixture(t) + + if xerr := f.svc.Delete(context.Background(), uuid.New().String(), f.mailbox.String()); xerr != errx.ErrNotFound { + t.Fatalf("error = %v, want not found", xerr) + } + if len(f.pub.removed) != 0 { + t.Errorf("published %d removals for someone else's mailbox, want 0", len(f.pub.removed)) + } + if !f.mailboxExists(t) { + t.Error("someone else's delete removed the mailbox") + } +} + +// A removal that cannot be published leaves the mailbox exactly as it was: +// still there, still assigned, still counted. +func TestLiveDeleteKeepsEverythingWhenTheWorkerCannotBeTold(t *testing.T) { + f := newRemovalLiveFixture(t) + f.pub.removeErr = errBusDown + + xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()) + if xerr == nil || xerr.Code != errx.ServiceUnavailable { + t.Fatalf("error = %v, want a 503 so the client retries", xerr) + } + if !f.mailboxExists(t) { + t.Fatal("the mailbox was deleted even though the worker was never told") + } + if count, score := f.workerLoad(t); count != 1 || score != 1 { + t.Errorf("worker load changed for a mailbox that still exists: account_count=%d load_score=%v", count, score) + } +} + +// A mailbox that has ever been scheduled work carries task rows, and a platform +// admin's warmup enforcement carries its own. Both used to reference the +// mailbox with no delete action, so disconnecting anything that had ever warmed +// up or sent a campaign step raised a foreign key violation: the customer got a +// server error and the mailbox stayed connected. Migration 000098 makes that +// work go with the mailbox it belongs to. +// +// Run cmd/migrate against WARMBLY_TEST_DB first; a database still on the old +// constraint fails this the way production did. +func TestLiveDeletingAMailboxWithScheduledWork(t *testing.T) { + f := newRemovalLiveFixture(t) + ctx := context.Background() + + if _, err := f.pool.Exec(ctx, + `INSERT INTO tasks (task_type, email_account_id, status, message_id, scheduled_at) + VALUES ('warmup', $1, 'pending', '', now()), ('campaign', $1, 'completed', '', now())`, + f.mailbox); err != nil { + t.Fatalf("fixture task: %v", err) + } + if _, err := f.pool.Exec(ctx, + `INSERT INTO warmup_admin_actions (admin_user_id, email_account_id, action, reason) + VALUES ($1, $2, 'block', 'fixture')`, f.user, f.mailbox); err != nil { + t.Fatalf("fixture admin action: %v", err) + } + + if xerr := f.svc.Delete(ctx, f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("disconnecting a mailbox that has scheduled work failed: %v", xerr) + } + if f.mailboxExists(t) { + t.Fatal("the mailbox survived its own delete") + } + + for _, table := range []string{"tasks", "warmup_admin_actions"} { + var left int + if err := f.pool.QueryRow(ctx, + `SELECT count(*) FROM `+table+` WHERE email_account_id = $1`, f.mailbox).Scan(&left); err != nil { + t.Fatalf("read %s: %v", table, err) + } + if left != 0 { + t.Errorf("%d %s rows point at a mailbox that no longer exists", left, table) + } + } +} diff --git a/internal/app/email/worker_removal_test.go b/internal/app/email/worker_removal_test.go new file mode 100644 index 00000000..1dfa65b4 --- /dev/null +++ b/internal/app/email/worker_removal_test.go @@ -0,0 +1,438 @@ +package email + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/events" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +// stubRemovalRepo answers the handful of calls the disable and disconnect +// paths make. Embedding the interface keeps the stub to those; anything else +// panics loudly rather than passing silently. +type stubRemovalRepo struct { + repository.EmailRepository + + trace *[]string + + account *models.Email + getErr *errx.Error + workerID *uuid.UUID + workerErr *errx.Error + updateErr *errx.Error + deleteErr *errx.Error + statusSet []string + deleteCalls int + workerCalls int +} + +func (s *stubRemovalRepo) record(step string) { + if s.trace != nil { + *s.trace = append(*s.trace, step) + } +} + +func (s *stubRemovalRepo) Update(ctx context.Context, userID, emailAccountID string, udata *models.UpdateEmail) (*models.Email, *errx.Error) { + if udata.Status != nil { + s.statusSet = append(s.statusSet, *udata.Status) + } + if s.updateErr != nil { + return nil, s.updateErr + } + // Deliberately mirrors the real repository: its RETURNING list carries the + // mailbox as the dashboard sees it and no worker assignment, so reading + // WorkerID off this row would find nothing. + id, _ := uuid.Parse(emailAccountID) + status := "inactive" + if udata.Status != nil { + status = *udata.Status + } + return &models.Email{ID: id, Status: status}, nil +} + +func (s *stubRemovalRepo) GetByID(ctx context.Context, emailAccountID uuid.UUID) (*models.Email, *errx.Error) { + if s.getErr != nil { + return nil, s.getErr + } + return s.account, nil +} + +func (s *stubRemovalRepo) GetWorkerID(ctx context.Context, emailAccountID uuid.UUID) (*uuid.UUID, *errx.Error) { + s.workerCalls++ + return s.workerID, s.workerErr +} + +func (s *stubRemovalRepo) GetSMTPCredentials(ctx context.Context, emailAccountID uuid.UUID) (*repository.SMTPCredentials, *errx.Error) { + return &repository.SMTPCredentials{SMTPHost: "smtp.test.local", SMTPPort: 587, IMAPHost: "imap.test.local", IMAPPort: 993}, nil +} + +func (s *stubRemovalRepo) Delete(ctx context.Context, userID, emailAccountID string) *errx.Error { + s.deleteCalls++ + s.record("delete") + return s.deleteErr +} + +// stubEventPublisher records what was shipped to workers. +type stubEventPublisher struct { + events.Publisher + + trace *[]string + + removeErr error + removed []workerRemoval + added []uuid.UUID +} + +var errBusDown = errors.New("bus down") + +type workerRemoval struct { + workerID uuid.UUID + userID string + emailID string +} + +func (p *stubEventPublisher) PublishRemoveEmail(ctx context.Context, workerID uuid.UUID, remove *models.RemoveWorkerEmail) error { + if p.trace != nil { + *p.trace = append(*p.trace, "remove") + } + p.removed = append(p.removed, workerRemoval{workerID: workerID, userID: remove.UserID, emailID: remove.EmailID}) + return p.removeErr +} + +func (p *stubEventPublisher) PublishAddEmail(ctx context.Context, workerID uuid.UUID, email *models.AddWorkerEmail) error { + if p.trace != nil { + *p.trace = append(*p.trace, "add") + } + p.added = append(p.added, email.ID) + return nil +} + +type removalFixture struct { + svc *emailService + repo *stubRemovalRepo + pub *stubEventPublisher + assign *stubAssignment + trace []string + user uuid.UUID + org uuid.UUID + mailbox uuid.UUID + worker uuid.UUID +} + +func newRemovalFixture(t *testing.T) *removalFixture { + t.Helper() + f := &removalFixture{user: uuid.New(), org: uuid.New(), mailbox: uuid.New(), worker: uuid.New()} + f.repo = &stubRemovalRepo{ + trace: &f.trace, + workerID: &f.worker, + account: &models.Email{ + ID: f.mailbox, + UserID: f.user.String(), + OrganizationID: &f.org, + WorkerID: &f.worker, + Email: "box@test.local", + Provider: "smtp_imap", + Status: "active", + }, + } + f.pub = &stubEventPublisher{trace: &f.trace} + f.assign = &stubAssignment{live: true} + f.svc = &emailService{emailRepository: f.repo, publisher: f.pub, workerAssignment: f.assign} + return f +} + +// The defect: a mailbox switched off in the dashboard kept syncing on its old +// schedule, because Update wrote the row and told nobody. +func TestDisablingAMailboxTellsTheWorkerToDropIt(t *testing.T) { + f := newRemovalFixture(t) + inactive := "inactive" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &inactive}); xerr != nil { + t.Fatalf("update: %v", xerr) + } + + if len(f.pub.removed) != 1 { + t.Fatalf("published %d removals, want 1: the worker keeps syncing a disabled mailbox until it restarts", len(f.pub.removed)) + } + got := f.pub.removed[0] + if got.workerID != f.worker { + t.Errorf("removal sent to worker %s, want %s", got.workerID, f.worker) + } + if got.emailID != f.mailbox.String() || got.userID != f.user.String() { + t.Errorf("removal carried user=%s email=%s, want user=%s email=%s", got.userID, got.emailID, f.user, f.mailbox) + } + // The assignment has to be asked for on its own: the row Update returns + // carries no worker_id, which is what made the consumer's removal + // unreachable the first time. + if f.repo.workerCalls != 1 { + t.Errorf("asked for the assignment %d times, want 1", f.repo.workerCalls) + } + if len(f.pub.added) != 0 { + t.Errorf("a disabled mailbox was shipped back to a worker: %v", f.pub.added) + } +} + +// A revoked mailbox is just as unusable as a disabled one, and the status +// column carries both. +func TestRevokingAMailboxAlsoDropsItFromTheWorker(t *testing.T) { + f := newRemovalFixture(t) + revoked := "revoked" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &revoked}); xerr != nil { + t.Fatalf("update: %v", xerr) + } + if len(f.pub.removed) != 1 { + t.Fatalf("published %d removals for a revoked mailbox, want 1", len(f.pub.removed)) + } +} + +// Re-enabling is the other half of the same silence: without this the mailbox +// waits on the reconciler's next pass, up to five minutes, before it syncs or +// sends again. +func TestReenablingAMailboxShipsItBackToItsWorker(t *testing.T) { + f := newRemovalFixture(t) + active := "active" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &active}); xerr != nil { + t.Fatalf("update: %v", xerr) + } + + if len(f.pub.added) != 1 || f.pub.added[0] != f.mailbox { + t.Fatalf("shipped %v back to the worker, want one %s", f.pub.added, f.mailbox) + } + if len(f.pub.removed) != 0 { + t.Errorf("a re-enabled mailbox was also removed: %v", f.pub.removed) + } +} + +// Every other PATCH (a signature, a daily cap, a tag) must leave the worker +// alone: re-shipping decrypted credentials on every keystroke is not free. +func TestAPatchThatLeavesTheStatusAloneDoesNotTouchTheWorker(t *testing.T) { + f := newRemovalFixture(t) + name := "New name" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Name: &name}); xerr != nil { + t.Fatalf("update: %v", xerr) + } + if len(f.pub.removed) != 0 || len(f.pub.added) != 0 { + t.Errorf("a plain edit reached the worker: %d removals, %d loads", len(f.pub.removed), len(f.pub.added)) + } +} + +// A status write that failed means the mailbox is still active, so nothing +// should be told to drop it. +func TestAFailedStatusWriteNeverReachesTheWorker(t *testing.T) { + f := newRemovalFixture(t) + f.repo.updateErr = errx.InternalError() + inactive := "inactive" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &inactive}); xerr == nil { + t.Fatal("a failed status write was reported as success") + } + if len(f.pub.removed) != 0 || f.repo.workerCalls != 0 { + t.Errorf("acted on a failed status write: %d removals, %d assignment lookups", len(f.pub.removed), f.repo.workerCalls) + } +} + +// Disabling stays best-effort: the status is written, no load path ships a +// mailbox that is not active, and the customer's edit must still succeed. +func TestDisablingSucceedsEvenWhenTheBusIsDown(t *testing.T) { + f := newRemovalFixture(t) + f.pub.removeErr = errBusDown + inactive := "inactive" + + if _, xerr := f.svc.Update(context.Background(), f.user.String(), f.mailbox.String(), &models.UpdateEmail{Status: &inactive}); xerr != nil { + t.Fatalf("a bus failure blocked the status change: %v", xerr) + } + if len(f.pub.removed) != 1 { + t.Errorf("published %d removals, want 1 attempt", len(f.pub.removed)) + } +} + +// Deleting is the path with no safety net: once the row is gone there is no +// assignment left to read and no reconciler that can repair a missed removal. +func TestDeleteTellsTheWorkerBeforeTheRowGoes(t *testing.T) { + f := newRemovalFixture(t) + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + + if len(f.pub.removed) != 1 { + t.Fatalf("published %d removals, want 1: the worker keeps syncing an account that no longer exists", len(f.pub.removed)) + } + if f.pub.removed[0].workerID != f.worker || f.pub.removed[0].emailID != f.mailbox.String() { + t.Errorf("removal = %+v, want worker %s and mailbox %s", f.pub.removed[0], f.worker, f.mailbox) + } + if len(f.trace) != 2 || f.trace[0] != "remove" || f.trace[1] != "delete" { + t.Errorf("order was %v, want the removal published before the row is deleted", f.trace) + } + if f.repo.deleteCalls != 1 { + t.Errorf("delete called %d times, want 1", f.repo.deleteCalls) + } +} + +// Reliable, not best-effort: a removal that could not be sent leaves the +// mailbox in place so the customer can try again, rather than stranding it on +// a worker forever. +func TestDeleteKeepsTheMailboxWhenTheWorkerCannotBeTold(t *testing.T) { + f := newRemovalFixture(t) + f.pub.removeErr = errBusDown + + xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()) + if xerr == nil { + t.Fatal("the mailbox was deleted without the worker ever being told") + } + if xerr.Code != errx.ServiceUnavailable { + t.Errorf("error code = %d, want 503 so the client knows to retry", xerr.Code) + } + if f.repo.deleteCalls != 0 { + t.Errorf("the row was deleted %d times after the removal failed, want 0", f.repo.deleteCalls) + } + if f.assign.unassignCalls != 0 { + t.Errorf("worker capacity was released for a mailbox that still exists (%d calls)", f.assign.unassignCalls) + } +} + +// The assignment lookup failing is the same situation: we do not know which +// worker to tell. +func TestDeleteKeepsTheMailboxWhenTheAssignmentCannotBeRead(t *testing.T) { + f := newRemovalFixture(t) + f.repo.workerErr = errx.InternalError() + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr == nil { + t.Fatal("the mailbox was deleted on an unreadable assignment") + } + if f.repo.deleteCalls != 0 { + t.Errorf("the row was deleted %d times, want 0", f.repo.deleteCalls) + } +} + +// A mailbox on no worker has nothing to remove, and must still delete. +func TestDeleteWithoutAWorkerStillRemovesTheRow(t *testing.T) { + f := newRemovalFixture(t) + f.repo.workerID = nil + f.repo.account.WorkerID = nil + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + if len(f.pub.removed) != 0 { + t.Errorf("published %d removals for a mailbox on no worker, want 0", len(f.pub.removed)) + } + if f.repo.deleteCalls != 1 { + t.Errorf("delete called %d times, want 1", f.repo.deleteCalls) + } +} + +// The worker keeps counting a deleted mailbox against its capacity otherwise: +// the foreign key nulls worker_id and nothing refunds the load. +func TestDeleteGivesTheWorkerItsCapacityBack(t *testing.T) { + f := newRemovalFixture(t) + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + if f.assign.unassignCalls != 1 { + t.Errorf("released the assignment %d times, want 1", f.assign.unassignCalls) + } +} + +// The removal must never be reachable for a mailbox the caller does not own: +// the lookup that finds it is unscoped, so ownership is checked here. +func TestDeleteRefusesAMailboxTheCallerDoesNotOwn(t *testing.T) { + f := newRemovalFixture(t) + f.repo.account.UserID = uuid.New().String() + + xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()) + if xerr != errx.ErrNotFound { + t.Fatalf("error = %v, want not found", xerr) + } + if len(f.pub.removed) != 0 || f.repo.deleteCalls != 0 { + t.Errorf("acted on someone else's mailbox: %d removals, %d deletes", len(f.pub.removed), f.repo.deleteCalls) + } +} + +// Owner ids arriving in different letter case are the same owner; Postgres +// compares them as uuids and so does this. +func TestDeleteAcceptsTheOwnerInAnyCase(t *testing.T) { + f := newRemovalFixture(t) + f.repo.account.UserID = uuidUpper(f.user) + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("the owner was refused their own mailbox: %v", xerr) + } +} + +func TestDeleteRejectsAMalformedID(t *testing.T) { + f := newRemovalFixture(t) + + if xerr := f.svc.Delete(context.Background(), f.user.String(), "not-a-uuid"); xerr != errx.ErrUuid { + t.Fatalf("error = %v, want a uuid error", xerr) + } + if f.repo.deleteCalls != 0 { + t.Errorf("a malformed id reached the delete (%d calls)", f.repo.deleteCalls) + } +} + +// Deleting a mailbox that is already gone is a 404, not a removal published +// into the void. +func TestDeleteOfAMissingMailboxPublishesNothing(t *testing.T) { + f := newRemovalFixture(t) + f.repo.getErr = errx.ErrNotFound + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != errx.ErrNotFound { + t.Fatalf("error = %v, want not found", xerr) + } + if len(f.pub.removed) != 0 || f.repo.deleteCalls != 0 { + t.Errorf("acted on a missing mailbox: %d removals, %d deletes", len(f.pub.removed), f.repo.deleteCalls) + } +} + +// The service is built without a publisher in jobs and reduced deployments; +// deleting must still work there. +func TestDeleteWithNoPublisherWired(t *testing.T) { + f := newRemovalFixture(t) + f.svc.publisher = nil + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + if f.repo.deleteCalls != 1 { + t.Errorf("delete called %d times, want 1", f.repo.deleteCalls) + } + if f.repo.workerCalls != 0 { + t.Errorf("looked up the assignment %d times with no publisher wired, want 0", f.repo.workerCalls) + } +} + +// uuidUpper renders an id the way a caller that upper-cases its ids would. +func uuidUpper(id uuid.UUID) string { + out := []rune(id.String()) + for i, r := range out { + if r >= 'a' && r <= 'f' { + out[i] = r - 32 + } + } + return string(out) +} + +// A delete that fails after the removal was published leaves a mailbox that is +// still active but no longer loaded anywhere. It goes straight back on rather +// than waiting minutes for the reconciler. +func TestAFailedDeletePutsTheMailboxBackOnItsWorker(t *testing.T) { + f := newRemovalFixture(t) + f.repo.deleteErr = errx.InternalError() + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr == nil { + t.Fatal("a failed delete was reported as success") + } + if len(f.pub.added) != 1 || f.pub.added[0] != f.mailbox { + t.Errorf("shipped %v back to a worker, want one %s", f.pub.added, f.mailbox) + } +} diff --git a/internal/errx/common.go b/internal/errx/common.go index f000002a..d158a58b 100644 --- a/internal/errx/common.go +++ b/internal/errx/common.go @@ -132,6 +132,12 @@ var ( ErrEmailWarmupIncrease = New(BadRequest, "Warmup increase amount must be between 0 and 100.") ErrEmailReplyRate = New(BadRequest, "Warmup reply rate must be between 0 and 100.") + // Disconnecting a mailbox has to reach the machine syncing it before the + // row goes: afterwards there is no assignment left to read and nothing that + // can repair a missed removal, so the mailbox would sync on forever. + ErrEmailWorkerUnreachable = NewWithIdentifier(ServiceUnavailable, "mailbox_worker_unreachable", + "This mailbox could not be disconnected right now because the machine syncing it could not be reached. Nothing was removed, so try again in a moment.") + // Campaign ErrCampaignName = New(BadRequest, "Campaign name length must be between 3 and 50 characters.") ErrCampaignDescription = New(BadRequest, "Campaign description length must be below 300 characters.") diff --git a/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.down.sql b/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.down.sql new file mode 100644 index 00000000..ed5f3cea --- /dev/null +++ b/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.down.sql @@ -0,0 +1,9 @@ +ALTER TABLE tasks DROP CONSTRAINT IF EXISTS tasks_email_account_id_fkey; +ALTER TABLE tasks + ADD CONSTRAINT tasks_email_account_id_fkey + FOREIGN KEY (email_account_id) REFERENCES email_accounts (id); + +ALTER TABLE warmup_admin_actions DROP CONSTRAINT IF EXISTS warmup_admin_actions_email_account_id_fkey; +ALTER TABLE warmup_admin_actions + ADD CONSTRAINT warmup_admin_actions_email_account_id_fkey + FOREIGN KEY (email_account_id) REFERENCES email_accounts (id); diff --git a/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.up.sql b/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.up.sql new file mode 100644 index 00000000..22beb65b --- /dev/null +++ b/internal/infrastructure/db/migrations/000098_mailbox_delete_cascades.up.sql @@ -0,0 +1,16 @@ +-- Disconnecting a mailbox failed for any mailbox that had ever been scheduled +-- work. tasks.email_account_id and warmup_admin_actions.email_account_id both +-- referenced email_accounts with no delete action, so the DELETE raised a +-- foreign key violation and the customer got a server error while the mailbox +-- stayed connected. Both rows describe work for, or enforcement against, one +-- mailbox and mean nothing once it is gone, so they go with it. + +ALTER TABLE tasks DROP CONSTRAINT IF EXISTS tasks_email_account_id_fkey; +ALTER TABLE tasks + ADD CONSTRAINT tasks_email_account_id_fkey + FOREIGN KEY (email_account_id) REFERENCES email_accounts (id) ON DELETE CASCADE; + +ALTER TABLE warmup_admin_actions DROP CONSTRAINT IF EXISTS warmup_admin_actions_email_account_id_fkey; +ALTER TABLE warmup_admin_actions + ADD CONSTRAINT warmup_admin_actions_email_account_id_fkey + FOREIGN KEY (email_account_id) REFERENCES email_accounts (id) ON DELETE CASCADE; diff --git a/web/src/app/app/emails/page.tsx b/web/src/app/app/emails/page.tsx index 9f8db63b..74aca733 100644 --- a/web/src/app/app/emails/page.tsx +++ b/web/src/app/app/emails/page.tsx @@ -147,12 +147,18 @@ export default function AddressesPage() { async () => { setRemoving(true); const results = await Promise.allSettled(selected.map((id) => removeEmail(id))); - const failed = results.filter((r) => r.status === "rejected").length; + const failed = results.filter((r) => r.status === "rejected"); await queryClient.invalidateQueries({ queryKey: ["emails"] }); setSelected([]); setRemoving(false); - if (failed > 0) toast.error(`${failed} mailbox${failed > 1 ? "es" : ""} couldn't be removed`); - else toast.success(`Removed ${n} mailbox${n > 1 ? "es" : ""}`); + if (failed.length > 0) { + // Surface the server's reason when there is one to show: a + // disconnect that could not reach the machine syncing the + // mailbox is worth retrying, and "couldn't be removed" + // alone does not say so. + const reason = failed.length === 1 ? removeErrorMessage(failed[0].reason) : undefined; + toast.error(reason ?? `${failed.length} mailbox${failed.length > 1 ? "es" : ""} couldn't be removed`); + } else toast.success(`Removed ${n} mailbox${n > 1 ? "es" : ""}`); }, ); }; @@ -450,6 +456,12 @@ export default function AddressesPage() { /* ── one mailbox row + its warmup dropdown ───────────────────────────── */ +// removeErrorMessage pulls the API's own explanation out of a failed request. +function removeErrorMessage(err: unknown): string | undefined { + const e = err as { response?: { data?: { message?: string } } }; + return e?.response?.data?.message; +} + function MailboxRow({ box, tags, From 36224c8caf53103506eb96b7d4ebdd468d366b52 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 00:49:11 -0700 Subject: [PATCH 2/2] feat: give the deleted mailbox's worker its capacity back inside the delete's own transaction, because a refund made beside it can be lost for good: once the row is gone nothing records which worker was charged for that mailbox, so a partially applied unassign left account_count or load_score charged with nothing able to repair it; emailRepository.Delete now deletes the row RETURNING worker_id and credits the worker in the same transaction, taking the placement weight from the caller so it matches what assignment charged, and the long comments this PR added are condensed to the constraint they exist to state --- internal/app/email/handler.go | 62 +++++++------------ internal/app/email/loader.go | 22 +++---- .../app/email/worker_removal_live_test.go | 17 +++++ internal/app/email/worker_removal_test.go | 34 ++++++++-- internal/repository/pg_email.go | 61 +++++++++++++----- 5 files changed, 119 insertions(+), 77 deletions(-) diff --git a/internal/app/email/handler.go b/internal/app/email/handler.go index 3325155a..962414c3 100644 --- a/internal/app/email/handler.go +++ b/internal/app/email/handler.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/app/worker" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/infrastructure/pubsub" @@ -56,14 +57,10 @@ func (s *emailService) Update(ctx context.Context, userID, emailAccountID string return account, nil } -// applyStatusToWorker carries a mailbox status change through to the machine -// running it. Only the row was written before this, so a mailbox the customer -// switched off kept syncing and sending on its old schedule until its worker -// restarted, and one switched back on waited for the reconciler's next pass. -// -// Best-effort in both directions, unlike the delete path: the status is -// already written, no load path ships a mailbox that is not active, and the -// reconciler re-places an active one either way. +// applyStatusToWorker carries a status change through to the machine running +// the mailbox, which only the row recorded before. Best-effort both ways: no +// load path ships a mailbox that is not active, and the reconciler re-places an +// active one anyway. func (s *emailService) applyStatusToWorker(ctx context.Context, userID string, account *models.Email, status *string) { if account == nil || status == nil { return @@ -271,25 +268,19 @@ func (s *emailService) resolveDomainAuth(ctx context.Context, userID, emailAccou return domain, &res, nil } -// Delete disconnects a mailbox for good. -// -// The worker holding it is told to drop it BEFORE the row goes, because after -// the delete there is no assignment left to read and no reconciler that can -// repair a missed removal: a worker that was never told keeps syncing an -// account that no longer exists and emits events the consumer resolves against -// nothing. That makes the removal part of the delete rather than a follow-up. -// When it cannot be sent, nothing is deleted and the customer can try again. +// Delete disconnects a mailbox. The worker is told to drop it BEFORE the row +// goes, because afterwards no assignment is left to read and nothing can repair +// a missed removal, so a removal that cannot be sent fails the whole delete. func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string) *errx.Error { accountID, err := uuid.Parse(emailAccountID) if err != nil { return errx.ErrUuid } - // Read by id rather than through Get: Get is scoped by organization and was - // being handed a user id, so it never found the mailbox and every side - // effect below was silently skipped. Ownership is checked here instead, - // against the same owner the delete itself filters on, so the removal - // published below is never reachable for someone else's mailbox. + // Read by id: Get is scoped by organization and was being handed a user id, + // so it never found the mailbox and every side effect below was skipped. + // Ownership moves here, or the removal below would be publishable for a + // mailbox the caller does not own. account, xerr := s.emailRepository.GetByID(ctx, accountID) if xerr != nil { return xerr @@ -302,23 +293,13 @@ func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string return xerr } - // Give the worker back the capacity this mailbox was counted for. Nothing - // else does it on delete: the row's worker_id is only nulled by the foreign - // key, so the worker's account count and load score would stay charged for - // a mailbox that no longer exists and its usable capacity would shrink with - // every disconnect. - if s.workerAssignment != nil && account.WorkerID != nil { - if err := s.workerAssignment.UnassignWorkerFromEmail(ctx, accountID); err != nil { - log.Warn().Err(err). - Str("email_id", accountID.String()). - Msg("could not release the deleted mailbox's worker capacity") - } - } - - if xerr := s.emailRepository.Delete(ctx, userID, emailAccountID); xerr != nil { - // The removal has already gone out and the mailbox is still active, so - // put it back now rather than leaving it dark until the reconciler's - // next pass, minutes later. + // The refund travels inside the delete's transaction: the foreign key only + // nulls worker_id, so a worker not credited here stays charged for a + // mailbox that no longer exists, unrepairably. + refund := worker.MailboxWeight(account.Provider, account.Warmup != nil) + if xerr := s.emailRepository.Delete(ctx, userID, emailAccountID, refund); xerr != nil { + // The removal already went out and the mailbox is still active: put it + // back now instead of leaving it dark until the reconciler's next pass. s.loadAccountBestEffort(ctx, accountID) return xerr } @@ -336,9 +317,8 @@ func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string return nil } -// sameUser compares two user ids as uuids, so a differently formatted but equal -// id is not read as a different owner (the delete itself compares in SQL, -// where Postgres does the same). +// sameUser compares user ids as uuids, the way the delete's own WHERE clause +// does, so formatting alone never reads as a different owner. func sameUser(a, b string) bool { left, aerr := uuid.Parse(a) right, berr := uuid.Parse(b) diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index dffd140d..35a5521e 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -172,21 +172,15 @@ func (s *emailService) LoadAccountOntoWorker(ctx context.Context, accountID uuid return s.publisher.PublishAddEmail(ctx, *workerID, payload) } -// dropFromWorker tells the worker holding a mailbox to drop it from memory. +// dropFromWorker tells the worker holding a mailbox to drop it from memory, +// the same removal the consumer sends when a provider error deactivates one. +// Workers hold accounts in memory and only filter on status at startup, so +// without this a disabled or disconnected mailbox syncs until that worker +// restarts. A send already dispatched is answered with EMAIL_FAILED, which +// walks its reservation back. // -// Workers keep accounts in memory, and the per-worker load filters on status -// only when a worker starts, so a mailbox the customer disabled or disconnected -// keeps syncing on its old schedule until that worker restarts. This is the -// control-plane half of that: the same removal the consumer sends when a -// provider error deactivates a mailbox. -// -// The assignment is asked for on its own rather than read off an updated row: -// the mailbox as the dashboard sees it carries no worker_id, which is exactly -// what made the consumer's removal unreachable the first time. -// -// A send already dispatched to that worker is not lost: it is answered with -// EMAIL_FAILED, which walks the reservation back and lets routing retry the -// step on a mailbox that is still active. +// The assignment is read on its own because the row an update returns carries +// no worker_id, which is what made the consumer's removal unreachable in #218. func (s *emailService) dropFromWorker(ctx context.Context, userID string, accountID uuid.UUID) *errx.Error { if s.publisher == nil { return nil diff --git a/internal/app/email/worker_removal_live_test.go b/internal/app/email/worker_removal_live_test.go index b93100fc..61f82df9 100644 --- a/internal/app/email/worker_removal_live_test.go +++ b/internal/app/email/worker_removal_live_test.go @@ -176,6 +176,23 @@ func TestLiveDeletingAMailboxRemovesItFromItsWorkerFirst(t *testing.T) { } } +// The refund shares the delete's transaction, so a delete that matches no row +// must leave the worker's capacity exactly as it was. Driven through the +// repository, because the service refuses a foreign owner before it gets here. +func TestLiveDeleteThatMatchesNoRowRefundsNothing(t *testing.T) { + f := newRemovalLiveFixture(t) + + if xerr := f.svc.emailRepository.Delete(context.Background(), uuid.New().String(), f.mailbox.String(), 1); xerr != errx.ErrNotFound { + t.Fatalf("error = %v, want not found", xerr) + } + if count, score := f.workerLoad(t); count != 1 || score != 1 { + t.Errorf("capacity was refunded for a mailbox that was not deleted: account_count=%d load_score=%v", count, score) + } + if !f.mailboxExists(t) { + t.Error("the mailbox was deleted by a caller that does not own it") + } +} + // The lookup that finds the mailbox is deliberately unscoped, so ownership is // checked in the service. A teammate's user id must not delete this mailbox or // publish a removal for it. diff --git a/internal/app/email/worker_removal_test.go b/internal/app/email/worker_removal_test.go index 1dfa65b4..51f9b56c 100644 --- a/internal/app/email/worker_removal_test.go +++ b/internal/app/email/worker_removal_test.go @@ -4,8 +4,10 @@ import ( "context" "errors" "testing" + "time" "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/worker" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/events" "github.com/warmbly/warmbly/internal/models" @@ -27,6 +29,7 @@ type stubRemovalRepo struct { updateErr *errx.Error deleteErr *errx.Error statusSet []string + refunded []float64 deleteCalls int workerCalls int } @@ -71,8 +74,9 @@ func (s *stubRemovalRepo) GetSMTPCredentials(ctx context.Context, emailAccountID return &repository.SMTPCredentials{SMTPHost: "smtp.test.local", SMTPPort: 587, IMAPHost: "imap.test.local", IMAPPort: 993}, nil } -func (s *stubRemovalRepo) Delete(ctx context.Context, userID, emailAccountID string) *errx.Error { +func (s *stubRemovalRepo) Delete(ctx context.Context, userID, emailAccountID string, workerLoadRefund float64) *errx.Error { s.deleteCalls++ + s.refunded = append(s.refunded, workerLoadRefund) s.record("delete") return s.deleteErr } @@ -294,8 +298,8 @@ func TestDeleteKeepsTheMailboxWhenTheWorkerCannotBeTold(t *testing.T) { if f.repo.deleteCalls != 0 { t.Errorf("the row was deleted %d times after the removal failed, want 0", f.repo.deleteCalls) } - if f.assign.unassignCalls != 0 { - t.Errorf("worker capacity was released for a mailbox that still exists (%d calls)", f.assign.unassignCalls) + if len(f.repo.refunded) != 0 { + t.Errorf("worker capacity was refunded for a mailbox that still exists (%v)", f.repo.refunded) } } @@ -331,15 +335,33 @@ func TestDeleteWithoutAWorkerStillRemovesTheRow(t *testing.T) { } // The worker keeps counting a deleted mailbox against its capacity otherwise: -// the foreign key nulls worker_id and nothing refunds the load. +// the foreign key nulls worker_id and nothing refunds the load. The refund +// rides inside the delete, so it cannot be left half-applied against a row that +// no longer says which worker was charged. func TestDeleteGivesTheWorkerItsCapacityBack(t *testing.T) { f := newRemovalFixture(t) if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { t.Fatalf("delete: %v", xerr) } - if f.assign.unassignCalls != 1 { - t.Errorf("released the assignment %d times, want 1", f.assign.unassignCalls) + if len(f.repo.refunded) != 1 || f.repo.refunded[0] != worker.MailboxWeight("smtp_imap", false) { + t.Errorf("refunds handed to the delete = %v, want one smtp_imap weight", f.repo.refunded) + } +} + +// A warming Gmail mailbox is charged less than a cold SMTP one, and has to be +// refunded what it was actually charged. +func TestDeleteRefundsTheWeightTheMailboxWasChargedAt(t *testing.T) { + f := newRemovalFixture(t) + warming := time.Now() + f.repo.account.Provider = "gmail-api" + f.repo.account.Warmup = &warming + + if xerr := f.svc.Delete(context.Background(), f.user.String(), f.mailbox.String()); xerr != nil { + t.Fatalf("delete: %v", xerr) + } + if len(f.repo.refunded) != 1 || f.repo.refunded[0] != worker.MailboxWeight("gmail-api", true) { + t.Errorf("refund = %v, want the warmup weight", f.repo.refunded) } } diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index d7a61f3c..7c939842 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -117,7 +117,10 @@ type EmailRepository interface { // domains. It returns the mailboxes that entered the failing state on THIS // call, which is what the sweep notifies on. UpdateDomainAuthState(ctx context.Context, domain, state string, spf, dkim, dmarc bool, dmarcPolicy, reason string, checkedAt time.Time) ([]models.EmailAuthTransition, *errx.Error) - Delete(ctx context.Context, userID, emailAccountID string) *errx.Error + // Delete removes a mailbox and refunds workerLoadRefund of its worker's + // load in the same transaction, so a deleted mailbox can never leave a + // worker permanently charged for it. + Delete(ctx context.Context, userID, emailAccountID string, workerLoadRefund float64) *errx.Error NewOauthAccount(ctx context.Context, userID string, data models.NewOauthAccount) (*models.Email, *errx.Error) NewSMTPIMAPAccount(ctx context.Context, userID string, data models.NewSMTPIMAPAccount) (*models.Email, *errx.Error) @@ -1228,28 +1231,54 @@ func (r *emailRepository) UpdateDomainAuthState(ctx context.Context, domain, sta return transitions, nil } -func (r *emailRepository) Delete(ctx context.Context, userID, emailAccountID string) *errx.Error { +// Delete removes a mailbox and refunds its worker's capacity in ONE +// transaction. Split in two, the refund can be lost for good: after the row is +// gone nothing records which worker was charged for that mailbox, so a refund +// that failed on its own could never be repaired and the worker would carry a +// deleted mailbox's load forever. workerLoadRefund is the mailbox's placement +// weight, computed by the caller from the same provider and warmup flag +// assignment charged it with. +func (r *emailRepository) Delete(ctx context.Context, userID, emailAccountID string, workerLoadRefund float64) *errx.Error { + tx, err := r.DB.Begin(ctx) + if err != nil { + db.CaptureError(err, "", nil, "begin") + return errx.InternalError() + } + defer tx.Rollback(ctx) + query := ` DELETE FROM email_accounts WHERE user_id = $1 AND id = $2 + RETURNING worker_id ` + params := []any{userID, emailAccountID} - params := []any{ - userID, - emailAccountID, - } - - cmd, err := r.DB.Exec( - ctx, - query, - params..., - ) - if err != nil { - db.CaptureError(err, query, params, "exec") + var workerID *uuid.UUID + if err := tx.QueryRow(ctx, query, params...).Scan(&workerID); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return errx.ErrNotFound + } + db.CaptureError(err, query, params, "queryrow") return errx.InternalError() } - if cmd.RowsAffected() == 0 { - return errx.ErrNotFound + + if workerID != nil { + refund := ` + UPDATE workers + SET account_count = GREATEST(account_count - 1, 0), + load_score = GREATEST(0, load_score - $2), + updated_at = now() + WHERE id = $1 + ` + if _, err := tx.Exec(ctx, refund, *workerID, workerLoadRefund); err != nil { + db.CaptureError(err, refund, []any{*workerID, workerLoadRefund}, "exec") + return errx.InternalError() + } + } + + if err := tx.Commit(ctx); err != nil { + db.CaptureError(err, "", nil, "commit") + return errx.InternalError() } return nil }