diff --git a/AGENTS.md b/AGENTS.md index b3510102..356619be 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, 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 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 7c043459..08c08ecb 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 | +| 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`. +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 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 42c91f14..229175fb 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 @@ -217,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 c4636ce8..4b9a7c58 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,14 @@ type Capacity struct { Effective float64 Load float64 Utilization 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 @@ -69,6 +80,10 @@ func ComputeCapacity(row WorkerCapacityRow) Capacity { if c.Effective > 0 { c.Utilization = c.Load / c.Effective } + c.Target = math.Floor(c.Base * c.HealthMul) + if c.Target <= 0 { + c.Target = 1 + } return c } diff --git a/internal/app/worker/capacity_test.go b/internal/app/worker/capacity_test.go index bb11b792..eb5e66bb 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 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) + } + 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 223b1779..0849e3c0 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 + weightNewWorker = 0.5 + weightOverTarget = 3.0 + weightOverload = 4.0 ) // providerSoftCap is how many mailboxes of ONE provider a single worker is @@ -91,16 +94,35 @@ 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 +} + +// 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 @@ -109,12 +131,28 @@ 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.Projected(req) + if projected <= 1 { + score += weightHeadroom * (1 - projected) + } else { + // 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) } - score += weightHeadroom * (1 - utilization) + + // 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 1983ee5a..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 @@ -20,15 +23,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 +159,64 @@ 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 TestSelectPlacementIsDeterministicOnTies(t *testing.T) { @@ -175,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,