Merge pull request #420 from warmbly/feat/worker-capacity-soft-target

Worker capacity is a placement target, not a hard gate

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.
 

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
 

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

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

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
 }
 

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)
+	}
+}

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.

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")
+	}
+}

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,