From bfdbd77b0469795b2b84f340552e706b0ac161fa Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Wed, 9 Sep 2026 05:32:30 -0700 Subject: [PATCH] feat: fix the eight defects the second review pass found, including three where the previous fix did not land: the bind-mounted state dir was root-owned so the node running as uid 1000 still could not write its update target, elevating the reserved-worker eviction did nothing because the rotation loop bailed on target-equals-current before the urgency was consulted, and the warmup-pool assertion was vacuous which hid that warmupPoolFor checked the subscription repo before the billing provider and answered free on a self-host install --- admin/src/app/dashboard/WorkerDetailPage.tsx | 3 +- .../fleet/ConvertDedicatedDialog.tsx | 40 ++++--------------- admin/src/lib/api/client/admin/fleet.ts | 5 +-- internal/api/handler/nodescript/join.sh | 7 ++++ .../app/email/worker_removal_live_test.go | 9 +++-- internal/app/fleet/rebalance.go | 23 ++++++++++- internal/app/fleet/rotation_live_test.go | 16 +++++++- internal/app/worker/assignment.go | 38 ++++++++++++------ internal/sandbox/seed.go | 4 +- 9 files changed, 88 insertions(+), 57 deletions(-) diff --git a/admin/src/app/dashboard/WorkerDetailPage.tsx b/admin/src/app/dashboard/WorkerDetailPage.tsx index b84f784a..83e236fc 100644 --- a/admin/src/app/dashboard/WorkerDetailPage.tsx +++ b/admin/src/app/dashboard/WorkerDetailPage.tsx @@ -263,8 +263,9 @@ export default function WorkerDetailPage() { {statsQ.data.emails_sent_today} {statsQ.data.emails_sent_this_week} {statsQ.data.total_emails_sent} + {/* Already a percentage in SQL; multiplying again gives 10000%. */} - {`${Math.round(statsQ.data.success_rate * 100)}%`} + {`${Math.round(statsQ.data.success_rate)}%`} )} diff --git a/admin/src/app/dashboard/fleet/ConvertDedicatedDialog.tsx b/admin/src/app/dashboard/fleet/ConvertDedicatedDialog.tsx index c4903de8..0b21d28b 100644 --- a/admin/src/app/dashboard/fleet/ConvertDedicatedDialog.tsx +++ b/admin/src/app/dashboard/fleet/ConvertDedicatedDialog.tsx @@ -46,7 +46,6 @@ export function ConvertDedicatedDialog({ const [workerId, setWorkerId] = useState(""); const [org, setOrg] = useState(null); const [subscriptionId, setSubscriptionId] = useState(""); - const [drainTo, setDrainTo] = useState(""); const workersQ = useQuery({ queryKey: ["admin", "workers", "managed"], @@ -58,18 +57,14 @@ export function ConvertDedicatedDialog({ // Any worker can be reserved: there is no category to check. const shared = workers; const worker = workers.find((w) => w.id === workerId) ?? null; - const needsDrain = !!worker && (worker.mailbox_count ?? 0) > 0; - const drainTargets = workers.filter((w) => w.id !== workerId); - const subOk = UUID_RE.test(subscriptionId.trim()); - const canSubmit = !!workerId && !!org && subOk && (!needsDrain || !!drainTo); + const canSubmit = !!workerId && !!org && subOk; const mutation = useMutation({ mutationFn: () => convertWorkerToDedicated(workerId, { organization_id: org!.id, subscription_id: subscriptionId.trim(), - drain_to_worker_id: drainTo || null, }), onSuccess: (res) => { toast.success( @@ -89,7 +84,6 @@ export function ConvertDedicatedDialog({ setWorkerId(""); setOrg(null); setSubscriptionId(""); - setDrainTo(""); } return ( @@ -117,8 +111,8 @@ export function ConvertDedicatedDialog({
- - @@ -159,29 +153,11 @@ export function ConvertDedicatedDialog({

-
- - -
+

+ Mailboxes already on this worker are not evicted here. The rotation loop + moves other tenants off it on its own schedule, so the reservation + becomes exclusive without re-authenticating every mailbox at once. +

diff --git a/admin/src/lib/api/client/admin/fleet.ts b/admin/src/lib/api/client/admin/fleet.ts index c182c59a..f8ecaca9 100644 --- a/admin/src/lib/api/client/admin/fleet.ts +++ b/admin/src/lib/api/client/admin/fleet.ts @@ -61,10 +61,9 @@ export interface AdminDedicatedAssignment { account_count: number; } -export interface AdminConvertDedicatedRequest { +export interface AdminReserveWorkerRequest { organization_id: string; subscription_id: string; - drain_to_worker_id?: string | null; } // Mirrors the /reserve handler. Reserving no longer drains anything: the @@ -133,7 +132,7 @@ export function releaseDedicatedWorker(orgId: string): Promise { return Request({ method: "POST", diff --git a/internal/api/handler/nodescript/join.sh b/internal/api/handler/nodescript/join.sh index 222d6d08..0c475f63 100755 --- a/internal/api/handler/nodescript/join.sh +++ b/internal/api/handler/nodescript/join.sh @@ -174,6 +174,12 @@ write_config() { fi mkdir -p "$CONFIG_DIR" "$STATE_DIR" + # The node container runs as uid 1000 (see deploy/docker/worker.Dockerfile), + # so the bind-mounted state dir has to be writable by it. Without this the + # agent's target-version write fails with EACCES, which it only logs, and + # auto-update silently never happens. + chown -R 1000:1000 "$STATE_DIR" 2>/dev/null || true + chmod 0775 "$STATE_DIR" umask 077 { printf '%s\n' "$NODE_ENV" @@ -256,6 +262,7 @@ fi sed -i "s|^WARMBLY_VERSION=.*|WARMBLY_VERSION=$target|" "$CONFIG_DIR/node.env" printf 'WARMBLY_IMAGE_REF=%s:%s\n' "$image" "$target" > "$STATE_DIR/image-ref" +chown -R 1000:1000 "$STATE_DIR" 2>/dev/null || true echo "warmbly-node-update: $current -> $target" systemctl restart "warmbly-$role" UPDATER diff --git a/internal/app/email/worker_removal_live_test.go b/internal/app/email/worker_removal_live_test.go index 61f82df9..f87b4ae1 100644 --- a/internal/app/email/worker_removal_live_test.go +++ b/internal/app/email/worker_removal_live_test.go @@ -62,8 +62,11 @@ func newRemovalLiveFixture(t *testing.T) *removalLiveFixture { 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) + // A worker is a node (the machine) plus a placement row (the mail on it). + exec(`INSERT INTO fleet_nodes (id, role, name, address, active, last_seen_at) + VALUES ($1, 'worker', 'drop-test', '127.0.0.1', true, now())`, f.worker) + exec(`INSERT INTO workers (id, account_count, load_score) + VALUES ($1, 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)`, @@ -76,7 +79,7 @@ func newRemovalLiveFixture(t *testing.T) *removalLiveFixture { arg any }{ {`DELETE FROM email_accounts WHERE id = $1`, f.mailbox}, - {`DELETE FROM workers WHERE id = $1`, f.worker}, + {`DELETE FROM fleet_nodes WHERE id = $1`, f.worker}, {`DELETE FROM organizations WHERE id = $1`, f.org}, {`DELETE FROM users WHERE id = $1`, f.user}, } { diff --git a/internal/app/fleet/rebalance.go b/internal/app/fleet/rebalance.go index 016c2631..d2bd8ea6 100644 --- a/internal/app/fleet/rebalance.go +++ b/internal/app/fleet/rebalance.go @@ -96,12 +96,22 @@ func (r *Rotator) tick(ctx context.Context) error { continue } - res, err := r.Assignment.SelectWorkerFor(ctx, workerapp.PlacementLookup{ + lookup := workerapp.PlacementLookup{ EmailAccountID: state.EmailAccountID, OrgID: *state.OrganizationID, CurrentWorkerID: state.WorkerID, Region: state.WorkerRegion, - }) + } + // When the mailbox has to LEAVE where it is, the current worker must be + // off the table: it is still the incumbent, still carries the + // stickiness bonus, and would win its own scoring, so the loop would + // bail on "target == current" and the mailbox would never go anywhere. + if mustLeave(urgency, state) { + lookup.CurrentWorkerID = nil + lookup.ExcludeWorkerID = state.WorkerID + } + + res, err := r.Assignment.SelectWorkerFor(ctx, lookup) if err != nil || res == nil || res.Worker == nil { continue } @@ -136,6 +146,15 @@ func (r *Rotator) tick(ctx context.Context) error { return nil } +// mustLeave reports whether staying put is not an option, as opposed to merely +// being improvable. A dead or degraded worker cannot do the work, and a worker +// reserved for someone else must not keep a stranger's mail. +func mustLeave(urgency workerapp.RotationUrgency, state repository.MailboxPlacementState) bool { + return urgency == workerapp.RotationImmediate || + urgency == workerapp.RotationElevated || + state.WorkerReservedForOtherOrg +} + // DrainWorker moves every mailbox off one worker, ignoring residency. Used by // the admin drain action and by the quarantine loop when a worker is blocked. func (r *Rotator) DrainWorker(ctx context.Context, workerID uuid.UUID) error { diff --git a/internal/app/fleet/rotation_live_test.go b/internal/app/fleet/rotation_live_test.go index e5ec2b0b..f021068d 100644 --- a/internal/app/fleet/rotation_live_test.go +++ b/internal/app/fleet/rotation_live_test.go @@ -29,6 +29,10 @@ func TestLivePlacementAndRotation(t *testing.T) { if dsn == "" { t.Skip("WARMBLY_TEST_DB unset") } + // Billing off is the self-host default and makes the pool assertion below + // distinguishable from the column default. + t.Setenv("BILLING_PROVIDER", "none") + ctx := context.Background() pool, err := pgxpool.New(ctx, dsn) if err != nil { @@ -123,14 +127,22 @@ func TestLivePlacementAndRotation(t *testing.T) { // 1b. Placement also settles warmup pool membership. It used to fall out of // tier placement; with tiers gone it has to be set explicitly, and // leaving it unset silently warms paying customers in the free pool. + // + // The assertion has to distinguish "set" from "left at the default", + // so it runs with billing disabled, where every org resolves to the + // premium pool and the column default ('free') is a visible failure. for i, mb := range mailboxes { var poolType *string if err := pool.QueryRow(ctx, `SELECT warmup_pool_type FROM email_accounts WHERE id = $1`, mb).Scan(&poolType); err != nil { t.Fatalf("mailbox %d: read warmup pool: %v", i, err) } - if poolType == nil || *poolType == "" { - t.Fatalf("mailbox %d: placement left warmup_pool_type unset", i) + if poolType == nil || *poolType != "premium" { + got := "" + if poolType != nil { + got = *poolType + } + t.Fatalf("mailbox %d: warmup_pool_type is %q, want \"premium\"; placement is not assigning pool membership", i, got) } } diff --git a/internal/app/worker/assignment.go b/internal/app/worker/assignment.go index e9adcd4e..42c91f14 100644 --- a/internal/app/worker/assignment.go +++ b/internal/app/worker/assignment.go @@ -182,7 +182,7 @@ func (s *workerAssignmentService) SelectWorkerFor(ctx context.Context, lookup Pl if err != nil || len(rows) == 0 { // A broken or unpopulated capacity view must not take onboarding down. // Fall back to the least-loaded live worker. - return s.selectFallback(ctx, req) + return s.selectFallback(ctx, req, lookup.ExcludeWorkerID) } candidates := make([]PlacementCandidate, 0, len(rows)) @@ -234,7 +234,7 @@ func (s *workerAssignmentService) SelectWorkerFor(ctx context.Context, lookup Pl best := SelectPlacement(candidates, req) if best == nil { - return s.selectFallback(ctx, req) + return s.selectFallback(ctx, req, lookup.ExcludeWorkerID) } return s.buildResult(ctx, *best, req, candidates) } @@ -267,16 +267,27 @@ func (s *workerAssignmentService) buildResult( } // selectFallback is the no-capacity-view path: least-loaded live worker. -func (s *workerAssignmentService) selectFallback(ctx context.Context, req PlacementRequest) (*PlacementResult, error) { +// +// It still refuses an unhealthy one. Without that, draining a quarantined +// worker could empty the candidate set, fall through here, and place the +// mailboxes onto another blocked machine, which is the opposite of what the +// drain was for. +func (s *workerAssignmentService) selectFallback(ctx context.Context, req PlacementRequest, exclude *uuid.UUID) (*PlacementResult, error) { workers, err := s.workerRepo.ListPlaceableWorkers(ctx) if err != nil { return nil, err } - if len(workers) == 0 { - return nil, ErrNoAvailableWorkers + for i := range workers { + w := workers[i] + if exclude != nil && w.ID == *exclude { + continue + } + switch w.HealthState { + case models.WorkerHealthHealthy, models.WorkerHealthWatch: + return &PlacementResult{Worker: &w}, nil + } } - w := workers[0] - return &PlacementResult{Worker: &w}, nil + return nil, ErrNoAvailableWorkers } // warmupPoolFor resolves which warmup pool a mailbox joins. Unrelated to @@ -286,14 +297,17 @@ func (s *workerAssignmentService) selectFallback(ctx context.Context, req Placem // more slowly, where the reverse would put unproven mail in front of paying // customers. func (s *workerAssignmentService) warmupPoolFor(ctx context.Context, orgID uuid.UUID) string { - if s.subRepo == nil { - return "free" - } - // With billing disabled there is no free/paid split to enforce, so every - // org gets the premium pool. Mirrors feature.gate's self-host unlock. + // Billing first: with it disabled there is no free/paid split to enforce, + // so every org gets the premium pool and the subscription is irrelevant. + // Checking the repository before this made a self-host install with no + // subscription repo wired fall through to "free" and warm every mailbox in + // the wrong pool. Mirrors feature.gate's self-host unlock. if config.BillingProvider() == "none" { return "premium" } + if s.subRepo == nil { + return "free" + } sub, err := s.subRepo.GetByOrganizationID(ctx, orgID) if err != nil || sub == nil || !sub.HasPaidSubscription() { return "free" diff --git a/internal/sandbox/seed.go b/internal/sandbox/seed.go index b8d3b0f6..a3e7d042 100644 --- a/internal/sandbox/seed.go +++ b/internal/sandbox/seed.go @@ -817,8 +817,8 @@ func repairContactVerification(ctx context.Context, pool *pgxpool.Pool) error { // an assignment ping-pong that strands mailboxes mid-send. func deactivateIdleFixtureWorkers(ctx context.Context, pool *pgxpool.Pool) error { tag, err := pool.Exec(ctx, ` - UPDATE workers SET active = FALSE, updated_at = NOW() - WHERE active AND id <> $1`, + UPDATE fleet_nodes SET active = FALSE, updated_at = NOW() + WHERE active AND role = 'worker' AND id <> $1`, sandboxWorker) if err != nil { return err