From 7fa4fdfbc46ae715b4243215ea072998f8d18707 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 10 Sep 2026 05:38:17 -0700 Subject: [PATCH 1/2] feat: make worker capacity a placement target rather than a hard gate, so Eligible refuses only on health and an over-target worker costs enough score to lose to anything with room instead of returning nil and dropping assignment into selectFallback, score projected utilization including the incoming mailbox's own weight, and penalise the 454/421 auth pressure the capacity view already collected and threw away --- AGENTS.md | 4 +- .../content/docs/development/architecture.mdx | 7 +- internal/app/worker/assignment.go | 5 +- internal/app/worker/capacity.go | 35 ++++++++ internal/app/worker/capacity_test.go | 26 ++++++ internal/app/worker/placement.go | 41 +++++++--- internal/app/worker/placement_test.go | 79 +++++++++++++++++-- 7 files changed, 176 insertions(+), 21 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index dac2f829..2c90f7e1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -312,7 +312,9 @@ In production, workers are treated as individually addressable executors: - worker events are delivered through worker-specific Kafka topics - the platform can rebalance or migrate accounts between workers, reluctantly -Placement is a score, never a filter (`internal/app/worker/placement.go`). Hard constraints cover only whether the work can be done: heartbeating, health in `healthy`/`watch`, and enough capacity headroom for the mailbox's weight. Everything else is a preference term: capacity headroom, incumbency (weighted highest), region match, tenant blast radius, per-provider crowding on one address, and foreign tenants for orgs entitled to isolated egress. +Placement is a score, never a filter (`internal/app/worker/placement.go`). Hard constraints cover only whether the work can be done: heartbeating and health in `healthy`/`watch`. Everything else is a preference term: capacity headroom (projected, so the incoming mailbox's own weight counts), incumbency (weighted highest), region match, tenant blast radius, per-provider crowding on one address, auth pressure from `454`/`421` throttles on that address, and foreign tenants for orgs entitled to isolated egress. + +Capacity is a target, not a ceiling, and nothing refuses a placement for being over it. Over-target costs more score than any bonus a candidate can earn, so a worker with room always wins when one exists; when none does, the least-overloaded wins with every other preference still applied. Do not put capacity back into `Eligible`: `base_capacity` is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and it refused precisely when the fleet was full, dropping assignment into `selectFallback` (first healthy worker, no region, no blast radius, no provider crowding). Capacity is one number for every worker in cold-mailbox equivalents, because each mailbox declares its own cost through `MailboxWeight`: `smtp_imap` = 1.0, `gmail`/`outlook` = 0.05, warmup-only = 0.4. Those are the `email_provider` enum values as stored; do not invent provider strings for them. diff --git a/docs/content/docs/development/architecture.mdx b/docs/content/docs/development/architecture.mdx index 7c043459..f2a28748 100644 --- a/docs/content/docs/development/architecture.mdx +++ b/docs/content/docs/development/architecture.mdx @@ -114,7 +114,7 @@ So the levers invert: IP *stability* per mailbox beats IP diversity, and a migra ### Choosing a worker -Placement scores every live worker and takes the best (`internal/app/worker/placement.go`). Hard constraints are only about whether the work can be done at all: the worker has to be heartbeating, in `healthy` or `watch`, and have capacity headroom for the mailbox's weight. Everything else is a preference: +Placement scores every live worker and takes the best (`internal/app/worker/placement.go`). Hard constraints are only about whether the work can be done at all: the worker has to be heartbeating and in `healthy` or `watch`. Everything else is a preference: | Term | Why | |---|---| @@ -123,9 +123,12 @@ Placement scores every live worker and takes the best (`internal/app/worker/plac | Region match | Sign-ins from where the provider expects them raise fewer challenges | | Tenant blast radius | Spread one customer across workers so a single failure does not stop their sending | | Provider crowding | Many accounts of one provider signing in from one address is what earns a per-IP throttle | +| Auth pressure | `454`/`421` throttles on a worker's address mean the provider is already pushing back on sign-ins from there | | Foreign tenants | Only for organizations entitled to isolated egress | -Capacity is one number for every worker, in cold-mailbox equivalents, because each mailbox already declares its own cost: an `smtp_imap` mailbox weighs `1.0`, a Gmail or Outlook API mailbox `0.05`, and a warmup-only assignment `0.4`. +Capacity is one number for every worker, in cold-mailbox equivalents, because each mailbox already declares its own cost: an `smtp_imap` mailbox weighs `1.0`, a Gmail or Outlook API mailbox `0.05`, and a warmup-only assignment `0.4`. The mailbox being placed counts against the target too, so the same nearly-full worker can be under target for a Gmail mailbox and over it for an `smtp_imap` one. + +**Capacity is a target, not a ceiling.** Nothing refuses a placement for being over it. Going over costs enough score that a worker with room wins whenever one exists, even against an incumbent; once no worker has room, the least-overloaded one still wins with region, blast radius and provider crowding all applied. This is deliberate: the target is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and refusing exactly when the fleet is full is when good placement matters most. ### Moving a mailbox diff --git a/internal/app/worker/assignment.go b/internal/app/worker/assignment.go index 42c91f14..35eeb7c6 100644 --- a/internal/app/worker/assignment.go +++ b/internal/app/worker/assignment.go @@ -91,8 +91,9 @@ type PlacementResult struct { Worker *models.Worker Score float64 IncumbentScore float64 - // IncumbentEligible is false when the current worker could not host the - // mailbox at all, in which case IncumbentScore is meaningless. + // IncumbentEligible is false when the current worker is too unhealthy to + // host the mailbox at all, in which case IncumbentScore is meaningless. + // Being over capacity does not clear it; that shows up in IncumbentScore. IncumbentEligible bool // Mandated is set when the worker was chosen by an entitlement rather than // by scoring, which today means an isolated-egress reservation. Callers diff --git a/internal/app/worker/capacity.go b/internal/app/worker/capacity.go index c4636ce8..20912adf 100644 --- a/internal/app/worker/capacity.go +++ b/internal/app/worker/capacity.go @@ -42,6 +42,9 @@ type WorkerCapacityRow struct { // dimensionless except Effective (mailbox-equivalents) and Load (sum of // mailbox weights). Utilization is Load/Effective and is the value the // scheduler sorts on when picking the next worker. +// +// Effective is a target, not a ceiling. Nothing refuses a placement for being +// over it; see Score in placement.go for what being over it costs. type Capacity struct { Base float64 HealthMul float64 @@ -49,6 +52,10 @@ type Capacity struct { Effective float64 Load float64 Utilization float64 + + // AuthPressure is how hard this worker's address is being pushed back on + // by mailbox providers, in [0, 1]. See computeAuthPressure. + AuthPressure float64 } // ComputeCapacity is the placement math: Base * Health * Age, floored at @@ -69,9 +76,37 @@ func ComputeCapacity(row WorkerCapacityRow) Capacity { if c.Effective > 0 { c.Utilization = c.Load / c.Effective } + c.AuthPressure = computeAuthPressure(row.AuthErrors1h, row.SendsAttempted1h) return c } +// authPressureFullScale is the auth-error rate that saturates the pressure +// signal. Auth errors are the one thing in worker_health_samples that is +// genuinely about the worker's own address rather than about the mailbox or +// its contact list: a per-IP authentication throttle (454 4.7.0) or per-IP +// rate limit (421 4.7.28) is the provider saying this client is signing in +// too much. Bounces and complaints follow the mailbox and say nothing about +// where it is running from. +// +// 5% is already a bad hour, so that is full scale. +const authPressureFullScale = 0.05 + +// computeAuthPressure scales the last hour's auth-error rate into [0, 1]. +// +// The denominator floors at 1 rather than at attempts, because auth errors +// also arrive from sync on a worker that sent nothing. Errors with no attempts +// is the worst case there is, and it saturates, which is the right answer. +func computeAuthPressure(authErrors, sendsAttempted int64) float64 { + if authErrors <= 0 { + return 0 + } + denom := float64(sendsAttempted) + if denom < 1 { + denom = 1 + } + return clampUnit((float64(authErrors) / denom) / authPressureFullScale) +} + // clampUnit pins x into [0, 1]. Negatives are surprisingly easy to feed // in - PostgreSQL's NULLIF/divide-by-zero handling can leak NaN through // in pathological cases - so we coerce defensively here. diff --git a/internal/app/worker/capacity_test.go b/internal/app/worker/capacity_test.go index bb11b792..f40e5f41 100644 --- a/internal/app/worker/capacity_test.go +++ b/internal/app/worker/capacity_test.go @@ -198,3 +198,29 @@ func TestComputeCapacity_HealthStatesArePassedThrough(t *testing.T) { } } } + +func TestComputeAuthPressure(t *testing.T) { + cases := []struct { + name string + authErrors int64 + attempted int64 + want float64 + }{ + {"clean worker", 0, 500, 0}, + {"one error in a thousand", 1, 1000, 0.02}, + {"one percent", 10, 1000, 0.2}, + {"full scale at five percent", 50, 1000, 1}, + {"saturates past full scale", 500, 1000, 1}, + // Sync auth failures arrive on a worker that sent nothing; the + // denominator floor makes that saturate rather than divide by zero. + {"errors with no attempts", 3, 0, 1}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := computeAuthPressure(tc.authErrors, tc.attempted) + if math.Abs(got-tc.want) > 1e-9 { + t.Fatalf("got %v want %v", got, tc.want) + } + }) + } +} diff --git a/internal/app/worker/placement.go b/internal/app/worker/placement.go index 223b1779..ab1df436 100644 --- a/internal/app/worker/placement.go +++ b/internal/app/worker/placement.go @@ -41,6 +41,9 @@ const ( weightIsolation = 1.5 weightBlastRadius = 0.8 weightProviderLoad = 0.7 + weightAuthPressure = 1.2 + weightOverTarget = 3.0 + weightOverload = 4.0 ) // providerSoftCap is how many mailboxes of ONE provider a single worker is @@ -91,16 +94,19 @@ type PlacementRequest struct { IsolatedEgress bool } -// Eligible reports whether a candidate may host the mailbox at all. These are -// the only hard constraints left: the worker has to be able to do the work. -// Everything else is a preference expressed in the score. +// Eligible reports whether a candidate may host the mailbox at all. Health is +// the only hard constraint: capacity is a preference in the score, because +// base_capacity is a flat 16 for every worker and refusing on it refused on a +// guess. It also refused where placement mattered most - a full fleet returned +// nil and fell through to selectFallback, which reads none of region, blast +// radius or provider crowding. func (c PlacementCandidate) Eligible(req PlacementRequest) bool { switch c.Health { case models.WorkerHealthHealthy, models.WorkerHealthWatch: + return true default: return false } - return c.Capacity.Effective-c.Capacity.Load >= req.Weight } // Score ranks an eligible candidate. Higher is better. The terms are additive @@ -109,12 +115,29 @@ func (c PlacementCandidate) Eligible(req PlacementRequest) bool { func (c PlacementCandidate) Score(req PlacementRequest) float64 { var score float64 - // Capacity: prefer the worker with the most room, so the fleet fills evenly. - utilization := c.Capacity.Utilization - if utilization > 1 { - utilization = 1 + // Capacity: prefer the worker with the most room, so the fleet fills + // evenly. Projected, so the incoming mailbox's own weight counts - Eligible + // used to be the only thing that read req.Weight. + projected := c.Capacity.Utilization + if c.Capacity.Effective > 0 { + projected = (c.Capacity.Load + req.Weight) / c.Capacity.Effective } - score += weightHeadroom * (1 - utilization) + if projected <= 1 { + score += weightHeadroom * (1 - projected) + } else { + // A soft wall: the flat cost exceeds the largest bonus a candidate can + // earn (incumbency 2.0 + region 0.6), so anything with room wins when + // it exists, and the ramp still ranks the overloaded against each + // other when nothing does. + over := projected - 1 + score -= weightOverTarget + weightOverload*over + } + + // Provider pushback: 454/421 auth throttles mean this address is signing + // in too much, so steer new mailboxes away. Below incumbency on purpose - + // evacuating a throttled worker is rotation's call, behind a residency + // floor, not a per-mailbox sign-in challenge paid here. + score -= weightAuthPressure * c.Capacity.AuthPressure // Stickiness: the incumbent wins ties and most non-ties. A mailbox that // stays put keeps presenting the same client IP to its provider. diff --git a/internal/app/worker/placement_test.go b/internal/app/worker/placement_test.go index 1983ee5a..368e959a 100644 --- a/internal/app/worker/placement_test.go +++ b/internal/app/worker/placement_test.go @@ -20,15 +20,17 @@ func candidate(id uuid.UUID, effective, load float64) PlacementCandidate { return c } -func TestEligibleRequiresHealthAndHeadroom(t *testing.T) { +func TestEligibleRequiresHealthOnly(t *testing.T) { id := uuid.New() req := PlacementRequest{Weight: 1.0} if c := candidate(id, 16, 4); !c.Eligible(req) { t.Fatal("healthy worker with headroom should be eligible") } - if c := candidate(id, 16, 15.5); c.Eligible(req) { - t.Fatal("worker with less headroom than the mailbox weight should not be eligible") + // Capacity is a preference, not a fence: an over-target worker is still a + // legal home, it just scores badly. See TestOverTargetWorkerIsLastResort. + if c := candidate(id, 16, 40); !c.Eligible(req) { + t.Fatal("being over the capacity target must not refuse a placement") } for _, state := range []models.WorkerHealthState{ @@ -154,14 +156,77 @@ func TestRegionMatchIsAPreferenceNotARequirement(t *testing.T) { } } -func TestSelectPlacementReturnsNilWhenNothingFits(t *testing.T) { - full := candidate(uuid.New(), 16, 16) - if got := SelectPlacement([]PlacementCandidate{full}, PlacementRequest{Weight: 1.0}); got != nil { - t.Fatal("expected no placement when every worker is full") +func TestSelectPlacementReturnsNilOnlyWhenNothingIsHealthy(t *testing.T) { + sick := candidate(uuid.New(), 16, 0) + sick.Health = models.WorkerHealthQuarantined + if got := SelectPlacement([]PlacementCandidate{sick}, PlacementRequest{Weight: 1.0}); got != nil { + t.Fatal("expected no placement when every worker is unhealthy") } if got := SelectPlacement(nil, PlacementRequest{Weight: 1.0}); got != nil { t.Fatal("expected no placement from an empty fleet") } + + // A full fleet is not an empty one. Returning nil here used to drop + // assignment into selectFallback, which ignores every preference term. + full := candidate(uuid.New(), 16, 16) + if got := SelectPlacement([]PlacementCandidate{full}, PlacementRequest{Weight: 1.0}); got == nil { + t.Fatal("a full but healthy fleet must still place the mailbox") + } +} + +func TestOverTargetWorkerIsLastResort(t *testing.T) { + over, room := uuid.New(), uuid.New() + + // The over-target worker is also the incumbent and matches the region, so + // it collects every bonus available. It must still lose to spare capacity. + a := candidate(over, 16, 18) + a.Region = "eu" + b := candidate(room, 16, 15) + + got := SelectPlacement([]PlacementCandidate{a, b}, PlacementRequest{ + Weight: 1.0, CurrentWorkerID: &over, Region: "eu", + }) + if got == nil || got.WorkerID != room { + t.Fatal("a worker with room must beat an over-target one holding every bonus") + } +} + +func TestAmongOverloadedWorkersTheLeastOverloadedWins(t *testing.T) { + bad, worse := uuid.New(), uuid.New() + + got := SelectPlacement([]PlacementCandidate{ + candidate(worse, 16, 40), + candidate(bad, 16, 18), + }, PlacementRequest{Weight: 1.0}) + if got == nil || got.WorkerID != bad { + t.Fatal("with no room anywhere, the least-overloaded worker should win") + } +} + +func TestMailboxWeightCountsAgainstTheTarget(t *testing.T) { + id := uuid.New() + c := candidate(id, 16, 15.5) + + // The same worker is under target for a Gmail mailbox and over it for an + // smtp_imap one. Only the projected utilization can tell them apart. + light := c.Score(PlacementRequest{Weight: MailboxWeight("gmail", false)}) + heavy := c.Score(PlacementRequest{Weight: MailboxWeight("smtp_imap", false)}) + if !(light > 0 && heavy < -weightOverTarget+1) { + t.Fatalf("weight should change the verdict: light=%.3f heavy=%.3f", light, heavy) + } +} + +func TestAuthPressureSteersAwayFromAThrottledAddress(t *testing.T) { + throttled, clean := uuid.New(), uuid.New() + + a := candidate(throttled, 16, 8) + a.Capacity.AuthPressure = 1 + b := candidate(clean, 16, 8) + + got := SelectPlacement([]PlacementCandidate{a, b}, PlacementRequest{Weight: 1.0}) + if got == nil || got.WorkerID != clean { + t.Fatal("placement should avoid an address the provider is auth-throttling") + } } func TestSelectPlacementIsDeterministicOnTies(t *testing.T) { From fa2b5330d78d262e7b526baf9e9c8194a67b08e6 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 10 Sep 2026 05:51:17 -0700 Subject: [PATCH 2/2] feat: drop the auth-pressure placement term because worker_capacity_view aggregates auth_errors (per-mailbox credential failures) and not rate_limit_errors (the 454/421 per-IP throttles it claimed to measure), measure projected utilization against an age-free Capacity.Target so a freshly joined node can relieve a full fleet instead of scoring as 200% loaded after one mailbox, bound the isolated-egress override with an explicit OverTarget check now that Eligible no longer caps it, and cap rotation moves per destination since a tick scores every mailbox against one frozen materialized-view snapshot --- AGENTS.md | 4 +- .../content/docs/development/architecture.mdx | 4 +- internal/app/fleet/rebalance.go | 14 ++++ internal/app/worker/assignment.go | 11 ++- internal/app/worker/capacity.go | 42 +++------- internal/app/worker/capacity_test.go | 44 +++++----- internal/app/worker/placement.go | 47 +++++++---- internal/app/worker/placement_test.go | 81 +++++++++++++++---- internal/repository/pg_worker_placement.go | 6 +- 9 files changed, 161 insertions(+), 92 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2c90f7e1..d27b7b2b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -312,9 +312,9 @@ In production, workers are treated as individually addressable executors: - worker events are delivered through worker-specific Kafka topics - the platform can rebalance or migrate accounts between workers, reluctantly -Placement is a score, never a filter (`internal/app/worker/placement.go`). Hard constraints cover only whether the work can be done: heartbeating and health in `healthy`/`watch`. Everything else is a preference term: capacity headroom (projected, so the incoming mailbox's own weight counts), incumbency (weighted highest), region match, tenant blast radius, per-provider crowding on one address, auth pressure from `454`/`421` throttles on that address, and foreign tenants for orgs entitled to isolated egress. +Placement is a score, never a filter (`internal/app/worker/placement.go`). Hard constraints cover only whether the work can be done: heartbeating and health in `healthy`/`watch`. Everything else is a preference term: capacity headroom (projected, so the incoming mailbox's own weight counts), incumbency (weighted highest), region match, tenant blast radius, per-provider crowding on one address, node youth, and foreign tenants for orgs entitled to isolated egress. -Capacity is a target, not a ceiling, and nothing refuses a placement for being over it. Over-target costs more score than any bonus a candidate can earn, so a worker with room always wins when one exists; when none does, the least-overloaded wins with every other preference still applied. Do not put capacity back into `Eligible`: `base_capacity` is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and it refused precisely when the fleet was full, dropping assignment into `selectFallback` (first healthy worker, no region, no blast radius, no provider crowding). +Capacity is a target, not a ceiling, and nothing refuses a placement for being over it. Over-target costs more score than any bonus a candidate can earn, so stickiness alone can never keep a mailbox on an over-target worker; it does not outweigh the penalty terms, so a worker with room but crowded with foreign tenants can still lose. When nothing has room the least-overloaded wins with every other preference applied. The target deliberately excludes the age ramp (`Capacity.Target`, not `Effective`): age damping collapses `Effective` to its floor for a new node's first hours, and dividing by that made a one-hour-old worker look overloaded after one mailbox, so joining a worker could not relieve a full fleet. Youth is a small score term instead. The isolated-egress override in `assignment.go` skips scoring entirely, so it checks `OverTarget` explicitly; `Eligible` no longer bounds it. Do not put capacity back into `Eligible`: `base_capacity` is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and it refused precisely when the fleet was full, dropping assignment into `selectFallback` (first healthy worker, no region, no blast radius, no provider crowding). Capacity is one number for every worker in cold-mailbox equivalents, because each mailbox declares its own cost through `MailboxWeight`: `smtp_imap` = 1.0, `gmail`/`outlook` = 0.05, warmup-only = 0.4. Those are the `email_provider` enum values as stored; do not invent provider strings for them. diff --git a/docs/content/docs/development/architecture.mdx b/docs/content/docs/development/architecture.mdx index f2a28748..08c08ecb 100644 --- a/docs/content/docs/development/architecture.mdx +++ b/docs/content/docs/development/architecture.mdx @@ -123,12 +123,12 @@ Placement scores every live worker and takes the best (`internal/app/worker/plac | Region match | Sign-ins from where the provider expects them raise fewer challenges | | Tenant blast radius | Spread one customer across workers so a single failure does not stop their sending | | Provider crowding | Many accounts of one provider signing in from one address is what earns a per-IP throttle | -| Auth pressure | `454`/`421` throttles on a worker's address mean the provider is already pushing back on sign-ins from there | +| Node youth | A worker that enrolled minutes ago has proved nothing, so it is probed gently rather than handed every placement for being empty | | Foreign tenants | Only for organizations entitled to isolated egress | Capacity is one number for every worker, in cold-mailbox equivalents, because each mailbox already declares its own cost: an `smtp_imap` mailbox weighs `1.0`, a Gmail or Outlook API mailbox `0.05`, and a warmup-only assignment `0.4`. The mailbox being placed counts against the target too, so the same nearly-full worker can be under target for a Gmail mailbox and over it for an `smtp_imap` one. -**Capacity is a target, not a ceiling.** Nothing refuses a placement for being over it. Going over costs enough score that a worker with room wins whenever one exists, even against an incumbent; once no worker has room, the least-overloaded one still wins with region, blast radius and provider crowding all applied. This is deliberate: the target is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and refusing exactly when the fleet is full is when good placement matters most. +**Capacity is a target, not a ceiling.** Nothing refuses a placement for being over it. Going over costs more than any bonus a candidate can earn, so stickiness alone can never keep a mailbox on an over-target worker; it does not outweigh the penalty terms, so a worker with room but crowded with foreign tenants can still lose. Once no worker has room, the least-overloaded one wins with region, blast radius and provider crowding all applied. This is deliberate: the target is a flat `16` for every worker regardless of the machine, so refusing on it refuses on a guess, and refusing exactly when the fleet is full is when good placement matters most. ### Moving a mailbox diff --git a/internal/app/fleet/rebalance.go b/internal/app/fleet/rebalance.go index eb8a008c..660b4edb 100644 --- a/internal/app/fleet/rebalance.go +++ b/internal/app/fleet/rebalance.go @@ -37,6 +37,12 @@ type Rotator struct { // event (a bad deploy, a provider outage) degrades gracefully instead of // re-placing everything at once. MaxMovesPerTick int + // MaxMovesPerDestination bounds how much of one pass can land on the same + // worker. worker_capacity_view is materialized and refreshed on its own + // minute cadence, so every mailbox in a tick is scored against the same + // frozen load_score: without this, the emptiest worker looks equally empty + // to all of them and a single pass can fill it past its target. + MaxMovesPerDestination int // ScanLimit bounds how many candidate mailboxes one pass examines. ScanLimit int Interval time.Duration @@ -46,6 +52,9 @@ func (r *Rotator) defaults() { if r.MaxMovesPerTick == 0 { r.MaxMovesPerTick = 50 } + if r.MaxMovesPerDestination == 0 { + r.MaxMovesPerDestination = 5 + } if r.ScanLimit == 0 { r.ScanLimit = 500 } @@ -71,6 +80,7 @@ func (r *Rotator) tick(ctx context.Context) error { now := time.Now() moved := 0 + landed := make(map[uuid.UUID]int, r.MaxMovesPerTick) for _, state := range candidates { if moved >= r.MaxMovesPerTick { break @@ -123,6 +133,9 @@ func (r *Rotator) tick(ctx context.Context) error { if res.Worker.ID == *state.WorkerID { continue } + if landed[res.Worker.ID] >= r.MaxMovesPerDestination { + continue + } if !workerapp.WorthMoving(urgency, res.IncumbentScore, res.Score, res.Mandated) { continue } @@ -132,6 +145,7 @@ func (r *Rotator) tick(ctx context.Context) error { log.Warn().Err(err).Str("mailbox", state.EmailAccountID.String()).Msg("rotation move failed") continue } + landed[res.Worker.ID]++ moved++ mailboxID := state.EmailAccountID diff --git a/internal/app/worker/assignment.go b/internal/app/worker/assignment.go index 35eeb7c6..229175fb 100644 --- a/internal/app/worker/assignment.go +++ b/internal/app/worker/assignment.go @@ -218,11 +218,18 @@ func (s *workerAssignmentService) SelectWorkerFor(ctx context.Context, lookup Pl // The reserved worker for an isolated-egress org outranks the score: the // customer is paying for that specific address. It still has to be - // eligible - a reserved worker that is down is not a reason to strand. + // eligible - a reserved worker that is down is not a reason to strand - + // and it still has to have room. Capacity used to bound this branch + // through Eligible; now that Eligible is health-only, the check is + // explicit, because this path skips SelectPlacement and would otherwise + // pile a whole organization onto one box without ever paying the + // over-target cost. An over-target reserved worker falls through to + // scoring, where weightIsolation still prefers it: the entitlement is a + // strong preference, not a pin. if req.IsolatedEgress { if reserved, rerr := s.workerRepo.GetDedicatedWorkerByOrgID(ctx, lookup.OrgID); rerr == nil && reserved != nil { for _, c := range candidates { - if c.WorkerID == reserved.ID && c.Eligible(req) { + if c.WorkerID == reserved.ID && c.Eligible(req) && !c.OverTarget(req) { res, err := s.buildResult(ctx, c, req, candidates) if res != nil { res.Mandated = true diff --git a/internal/app/worker/capacity.go b/internal/app/worker/capacity.go index 20912adf..4b9a7c58 100644 --- a/internal/app/worker/capacity.go +++ b/internal/app/worker/capacity.go @@ -53,9 +53,13 @@ type Capacity struct { Load float64 Utilization float64 - // AuthPressure is how hard this worker's address is being pushed back on - // by mailbox providers, in [0, 1]. See computeAuthPressure. - AuthPressure float64 + // Target is Effective without the age ramp, and is what placement measures + // utilization against. Age damping exists to probe a new worker gently, + // but it collapses Effective to the floor for the first hours of a node's + // life, and dividing by that made a one-hour-old worker look 200% loaded + // after a single mailbox. Placement pays for youth as a score term + // instead; see weightNewWorker. + Target float64 } // ComputeCapacity is the placement math: Base * Health * Age, floored at @@ -76,37 +80,13 @@ func ComputeCapacity(row WorkerCapacityRow) Capacity { if c.Effective > 0 { c.Utilization = c.Load / c.Effective } - c.AuthPressure = computeAuthPressure(row.AuthErrors1h, row.SendsAttempted1h) + c.Target = math.Floor(c.Base * c.HealthMul) + if c.Target <= 0 { + c.Target = 1 + } return c } -// authPressureFullScale is the auth-error rate that saturates the pressure -// signal. Auth errors are the one thing in worker_health_samples that is -// genuinely about the worker's own address rather than about the mailbox or -// its contact list: a per-IP authentication throttle (454 4.7.0) or per-IP -// rate limit (421 4.7.28) is the provider saying this client is signing in -// too much. Bounces and complaints follow the mailbox and say nothing about -// where it is running from. -// -// 5% is already a bad hour, so that is full scale. -const authPressureFullScale = 0.05 - -// computeAuthPressure scales the last hour's auth-error rate into [0, 1]. -// -// The denominator floors at 1 rather than at attempts, because auth errors -// also arrive from sync on a worker that sent nothing. Errors with no attempts -// is the worst case there is, and it saturates, which is the right answer. -func computeAuthPressure(authErrors, sendsAttempted int64) float64 { - if authErrors <= 0 { - return 0 - } - denom := float64(sendsAttempted) - if denom < 1 { - denom = 1 - } - return clampUnit((float64(authErrors) / denom) / authPressureFullScale) -} - // clampUnit pins x into [0, 1]. Negatives are surprisingly easy to feed // in - PostgreSQL's NULLIF/divide-by-zero handling can leak NaN through // in pathological cases - so we coerce defensively here. diff --git a/internal/app/worker/capacity_test.go b/internal/app/worker/capacity_test.go index f40e5f41..eb5e66bb 100644 --- a/internal/app/worker/capacity_test.go +++ b/internal/app/worker/capacity_test.go @@ -199,28 +199,28 @@ func TestComputeCapacity_HealthStatesArePassedThrough(t *testing.T) { } } -func TestComputeAuthPressure(t *testing.T) { - cases := []struct { - name string - authErrors int64 - attempted int64 - want float64 - }{ - {"clean worker", 0, 500, 0}, - {"one error in a thousand", 1, 1000, 0.02}, - {"one percent", 10, 1000, 0.2}, - {"full scale at five percent", 50, 1000, 1}, - {"saturates past full scale", 500, 1000, 1}, - // Sync auth failures arrive on a worker that sent nothing; the - // denominator floor makes that saturate rather than divide by zero. - {"errors with no attempts", 3, 0, 1}, +func TestComputeCapacity_TargetIgnoresTheAgeRamp(t *testing.T) { + // One hour into the 72h ramp. Effective collapses to its floor, which is + // what the placer used to divide by; Target must not. + c := ComputeCapacity(WorkerCapacityRow{ + BaseCapacity: 16, HealthMultiplier: 1, AgeMultiplier: 1.0 / 72, LoadScore: 0, + }) + if c.Effective != 1 { + t.Fatalf("effective: got %v want the floor of 1", c.Effective) } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - got := computeAuthPressure(tc.authErrors, tc.attempted) - if math.Abs(got-tc.want) > 1e-9 { - t.Fatalf("got %v want %v", got, tc.want) - } - }) + if c.Target != 16 { + t.Fatalf("target: got %v want 16", c.Target) + } + + // Health still shrinks the target: that is about the worker, not its age. + sick := ComputeCapacity(WorkerCapacityRow{BaseCapacity: 16, HealthMultiplier: 0.5, AgeMultiplier: 1}) + if sick.Target != 8 { + t.Fatalf("target: got %v want 8", sick.Target) + } + + // And it never reaches zero, or projected utilization divides by nothing. + dead := ComputeCapacity(WorkerCapacityRow{BaseCapacity: 16, HealthMultiplier: 0, AgeMultiplier: 1}) + if dead.Target != 1 { + t.Fatalf("target: got %v want the floor of 1", dead.Target) } } diff --git a/internal/app/worker/placement.go b/internal/app/worker/placement.go index ab1df436..0849e3c0 100644 --- a/internal/app/worker/placement.go +++ b/internal/app/worker/placement.go @@ -41,7 +41,7 @@ const ( weightIsolation = 1.5 weightBlastRadius = 0.8 weightProviderLoad = 0.7 - weightAuthPressure = 1.2 + weightNewWorker = 0.5 weightOverTarget = 3.0 weightOverload = 4.0 ) @@ -109,6 +109,22 @@ func (c PlacementCandidate) Eligible(req PlacementRequest) bool { } } +// Projected is the worker's utilization against its capacity target once this +// mailbox lands on it. +func (c PlacementCandidate) Projected(req PlacementRequest) float64 { + if c.Capacity.Target <= 0 { + return c.Capacity.Utilization + } + return (c.Capacity.Load + req.Weight) / c.Capacity.Target +} + +// OverTarget reports whether taking this mailbox would put the worker past its +// capacity target. Never refuses a placement on its own; it is what the +// isolated-egress override consults before skipping the score entirely. +func (c PlacementCandidate) OverTarget(req PlacementRequest) bool { + return c.Projected(req) > 1 +} + // Score ranks an eligible candidate. Higher is better. The terms are additive // and each one is traceable to a specific provider behaviour, which is what // makes the number explainable in the decision log. @@ -118,26 +134,25 @@ func (c PlacementCandidate) Score(req PlacementRequest) float64 { // Capacity: prefer the worker with the most room, so the fleet fills // evenly. Projected, so the incoming mailbox's own weight counts - Eligible // used to be the only thing that read req.Weight. - projected := c.Capacity.Utilization - if c.Capacity.Effective > 0 { - projected = (c.Capacity.Load + req.Weight) / c.Capacity.Effective - } + projected := c.Projected(req) if projected <= 1 { score += weightHeadroom * (1 - projected) } else { - // A soft wall: the flat cost exceeds the largest bonus a candidate can - // earn (incumbency 2.0 + region 0.6), so anything with room wins when - // it exists, and the ramp still ranks the overloaded against each - // other when nothing does. - over := projected - 1 - score -= weightOverTarget + weightOverload*over + // A soft wall: the flat cost exceeds every bonus a candidate can earn + // (incumbency 2.0 + region 0.6), so being over target can never be + // outweighed by stickiness. It does not dominate the penalty terms, so + // a worker with room but carrying foreign tenants, org concentration + // and provider crowding can still lose - those are real costs too. The + // ramp keeps ranking the overloaded against each other when nothing + // has room, which is the case the old hard check handled by giving up. + score -= weightOverTarget + weightOverload*(projected-1) } - // Provider pushback: 454/421 auth throttles mean this address is signing - // in too much, so steer new mailboxes away. Below incumbency on purpose - - // evacuating a throttled worker is rotation's call, behind a residency - // floor, not a per-mailbox sign-in challenge paid here. - score -= weightAuthPressure * c.Capacity.AuthPressure + // Youth: a node that enrolled minutes ago has proved nothing, so probe it + // gently rather than handing it every placement for being empty. Small on + // purpose - it must never outweigh the over-target cost, or a full fleet + // could not be relieved by joining a worker, which is the whole remedy. + score -= weightNewWorker * (1 - c.Capacity.AgeMul) // Stickiness: the incumbent wins ties and most non-ties. A mailbox that // stays put keeps presenting the same client IP to its provider. diff --git a/internal/app/worker/placement_test.go b/internal/app/worker/placement_test.go index 368e959a..7f28e288 100644 --- a/internal/app/worker/placement_test.go +++ b/internal/app/worker/placement_test.go @@ -12,7 +12,10 @@ func candidate(id uuid.UUID, effective, load float64) PlacementCandidate { c := PlacementCandidate{ WorkerID: id, Health: models.WorkerHealthHealthy, - Capacity: Capacity{Effective: effective, Load: load}, + // AgeMul 1: a mature worker, so the youth term stays out of the way + // unless a test sets it. Target tracks Effective because these helpers + // describe workers past the age ramp, where the two are equal. + Capacity: Capacity{Effective: effective, Target: effective, Load: load, AgeMul: 1}, } if effective > 0 { c.Capacity.Utilization = load / effective @@ -216,19 +219,6 @@ func TestMailboxWeightCountsAgainstTheTarget(t *testing.T) { } } -func TestAuthPressureSteersAwayFromAThrottledAddress(t *testing.T) { - throttled, clean := uuid.New(), uuid.New() - - a := candidate(throttled, 16, 8) - a.Capacity.AuthPressure = 1 - b := candidate(clean, 16, 8) - - got := SelectPlacement([]PlacementCandidate{a, b}, PlacementRequest{Weight: 1.0}) - if got == nil || got.WorkerID != clean { - t.Fatal("placement should avoid an address the provider is auth-throttling") - } -} - func TestSelectPlacementIsDeterministicOnTies(t *testing.T) { a, b := uuid.New(), uuid.New() cands := []PlacementCandidate{candidate(a, 16, 8), candidate(b, 16, 8)} @@ -240,3 +230,66 @@ func TestSelectPlacementIsDeterministicOnTies(t *testing.T) { t.Fatal("tie-breaking must not depend on candidate order") } } + +func TestOverTargetLosesToStickinessButNotToEveryPenalty(t *testing.T) { + over, room := uuid.New(), uuid.New() + + // Stickiness alone never rescues an over-target worker: the flat cost is + // set above incumbency plus a region match. + a := candidate(over, 16, 18) + a.Region = "eu" + b := candidate(room, 16, 15) + got := SelectPlacement([]PlacementCandidate{a, b}, PlacementRequest{ + Weight: 1.0, CurrentWorkerID: &over, Region: "eu", + }) + if got == nil || got.WorkerID != room { + t.Fatal("stickiness must not outweigh being over target") + } + + // It is not a fence, though. Stack every penalty on the worker with room - + // a fresh node full of foreign tenants, already holding all of this org's + // mailboxes and crowded with this provider - and a marginally over-target + // but otherwise clean worker is the better home. That is the intended + // reading: the other terms are real costs, not tie-breakers. + crowded := candidate(room, 16, 15) + crowded.Capacity.AgeMul = 0 + crowded.TotalMailboxes = 1000 + crowded.OrgMailboxesHere = 10 + crowded.ProviderMailboxesHere = 24 + marginal := candidate(over, 16, 16.05) + req := PlacementRequest{Weight: 0.05, IsolatedEgress: true, OrgMailboxesTotal: 10} + if marginal.Score(req) <= crowded.Score(req) { + t.Fatalf("over-target cost should not dominate every penalty: over=%.3f room=%.3f", + marginal.Score(req), crowded.Score(req)) + } +} + +func TestNewNodeRelievesAFullFleet(t *testing.T) { + fresh, mature := uuid.New(), uuid.New() + + // A node enrolled an hour ago: the age ramp has collapsed Effective to its + // floor, but Target is what placement divides by, so it still reads empty. + // Scoring against Effective made this worker look 200% loaded after one + // mailbox and no full fleet could ever be relieved by adding a machine. + n := ComputeCapacity(WorkerCapacityRow{BaseCapacity: 16, HealthMultiplier: 1, AgeMultiplier: 1.0 / 72, LoadScore: 1}) + newNode := PlacementCandidate{WorkerID: fresh, Health: models.WorkerHealthHealthy, Capacity: n} + full := candidate(mature, 16, 20) + + got := SelectPlacement([]PlacementCandidate{full, newNode}, PlacementRequest{Weight: 1.0}) + if got == nil || got.WorkerID != fresh { + t.Fatal("a freshly joined node must be able to relieve an over-target fleet") + } +} + +func TestYouthIsAPreferenceNotABarrier(t *testing.T) { + fresh, mature := uuid.New(), uuid.New() + + young := candidate(fresh, 16, 0) + young.Capacity.AgeMul = 0 + old := candidate(mature, 16, 0) + + got := SelectPlacement([]PlacementCandidate{young, old}, PlacementRequest{Weight: 1.0}) + if got == nil || got.WorkerID != mature { + t.Fatal("between two empty workers the proven one should win") + } +} diff --git a/internal/repository/pg_worker_placement.go b/internal/repository/pg_worker_placement.go index 05a89709..e1e4c083 100644 --- a/internal/repository/pg_worker_placement.go +++ b/internal/repository/pg_worker_placement.go @@ -94,9 +94,9 @@ const placementCandidateSelect = ` ` // ListPlacementCandidates returns every worker that may host a mailbox for the -// given org and provider, with the neighbour counts attached. Filtering on -// capacity headroom and ranking both happen in app code (worker.SelectPlacement) -// so the scoring model can change without a migration. +// given org and provider, with the neighbour counts attached. Ranking happens +// entirely in app code (worker.SelectPlacement) so the scoring model can change +// without a migration; capacity is one of the scored terms, not a filter. func (r *workerRepository) ListPlacementCandidates( ctx context.Context, orgID uuid.UUID,