+ 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