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,