From ed2027b2020baa8863ad221f884c740de20d9f5a Mon Sep 17 00:00:00 2001 From: Dai Ha Date: Thu, 10 Sep 2026 13:46:06 +0700 Subject: [PATCH] fleetd #435: make FixedPlacementPolicy honor maxLoad FixedPlacementPolicy (the default placement policy) never consulted maxLoad, so an at-cap default was chosen anyway on every unqualified spawn -- the cap was advisory, not enforced, for the one policy every config uses by default. weighted/round-robin already gated on it via PlacementPolicyUtil.available(). Extract the "at cap" predicate into PlacementPolicyUtil.atCap(ctx, c) so all three policies share one definition, and consult it at both of FixedPlacementPolicy's filter sites (the default fast path and the candidate walk), mirroring the existing weightExcluded pattern. An at-cap default now falls through to the next candidate instead of refusing the spawn -- only when every candidate is unusable does the policy still throw, naming the cap in the message. Update the class javadoc (five exceptions -> six) and the reason-priority comments to match CompositePeerLauncher's explicit-spawn order (quarantine, cooling off, max load, model-off). --- .../fleet/placement/FixedPlacementPolicy.java | 90 +++++++++++++------ .../fleet/placement/PlacementPolicyUtil.java | 32 ++++--- .../member/CompositePeerLauncherTest.java | 25 ++++++ .../fleet/placement/PlacementPolicyTest.java | 87 ++++++++++++++++++ 4 files changed, 196 insertions(+), 38 deletions(-) diff --git a/fleetd/src/main/java/dev/ltms/fleet/placement/FixedPlacementPolicy.java b/fleetd/src/main/java/dev/ltms/fleet/placement/FixedPlacementPolicy.java index 4a0ca61..79cb0a1 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/placement/FixedPlacementPolicy.java +++ b/fleetd/src/main/java/dev/ltms/fleet/placement/FixedPlacementPolicy.java @@ -5,14 +5,12 @@ import java.util.List; /** * Backward-compatible placement: an unqualified spawn always resolves to the configured default - * profile, exactly as {@code CompositePeerLauncher} did before CB-518. This ignores caps - * ({@code maxLoad}) so that a pre-existing config behaves identically after upgrade — capacity - * gating for automatic placement is deliberately out of scope for {@code fixed}, exactly as it - * always has been. Reachability is a narrower exception (fleetd #315, below): a profile is never - * checked for reachability up front, only skipped once it has already failed in this same - * spawn call's retry loop — see the unreachable case below. + * profile, exactly as {@code CompositePeerLauncher} did before CB-518. Reachability is a narrower + * exception (fleetd #315, below): a profile is never checked for reachability up front, only + * skipped once it has already failed in this same spawn call's retry loop — see the + * unreachable case below. * - *

Five exceptions walk past the default instead of returning it unconditionally: + *

Six exceptions walk past the default instead of returning it unconditionally: *

- * A fleet where nothing is ever quarantined, cooling off, model-off, unreachable, or weight-0 never - * exercises any of these paths, so today's behaviour is unchanged — in particular, the very first - * selection of a spawn call always sees an empty {@code unreachable} set, so the first choice is - * untouched. + * A fleet where nothing is ever quarantined, cooling off, at cap, model-off, unreachable, or + * weight-0 never exercises any of these paths, so today's behaviour is unchanged — in particular, + * the very first selection of a spawn call always sees an empty {@code unreachable} set, so the + * first choice is untouched. */ final class FixedPlacementPolicy implements PlacementPolicy { @@ -51,13 +59,14 @@ final class FixedPlacementPolicy implements PlacementPolicy { public PlacementCandidate select(PlacementContext ctx) { String d = ctx.defaultProfile(); if (d != null && !d.isBlank() && !ctx.quarantined().contains(d) && !ctx.coolingOff().contains(d) - && !ctx.modelOff().contains(d) && !ctx.unreachable().contains(d) && !weightExcluded(ctx, d)) { + && !ctx.modelOff().contains(d) && !ctx.unreachable().contains(d) && !weightExcluded(ctx, d) + && !capExcluded(ctx, d)) { return new PlacementCandidate(d, null, 1.0f, null); } for (PlacementCandidate c : ctx.candidates()) { if (!ctx.quarantined().contains(c.profile()) && !ctx.coolingOff().contains(c.profile()) && !ctx.modelOff().contains(c.profile()) && !ctx.unreachable().contains(c.profile()) - && !c.excluded()) { + && !c.excluded() && !PlacementPolicyUtil.atCap(ctx, c)) { return new PlacementCandidate(c.profile(), null, c.weight(), c.maxLoad()); } } @@ -66,14 +75,18 @@ final class FixedPlacementPolicy implements PlacementPolicy { // Exhaustion quarantine takes priority: reported only when quarantine is absent, so the // message never claims "cooling off" for a profile that is really backend-exhausted. boolean dCoolingOff = !dQuarantined && ctx.coolingOff().contains(d); - // fleetd #422: model-off is a fourth, independent source (an operator decision) — but - // quarantine/cooling-off still take priority when more than one applies, matching + // fleetd #435: at-cap sits between cooling off and model-off, matching + // CompositePeerLauncher's explicit-spawn check order (quarantine, cooling off, max load, + // then model-off) — reported only when quarantine/cooling-off are both absent. + boolean dAtCap = !dQuarantined && !dCoolingOff && capExcluded(ctx, d); + // fleetd #422: model-off is a fifth, independent source (an operator decision) — but + // quarantine/cooling-off/at-cap still take priority when more than one applies, matching // CompositePeerLauncher's explicit-spawn check order (quarantine, cooling off, max load, // then model-off). - boolean dModelOff = !dQuarantined && !dCoolingOff && ctx.modelOff().contains(d); + boolean dModelOff = !dQuarantined && !dCoolingOff && !dAtCap && ctx.modelOff().contains(d); boolean dUnreachable = ctx.unreachable().contains(d); boolean dWeightExcluded = weightExcluded(ctx, d); - if (dQuarantined || dCoolingOff || dModelOff || dUnreachable || dWeightExcluded) { + if (dQuarantined || dCoolingOff || dAtCap || dModelOff || dUnreachable || dWeightExcluded) { List reasons = new ArrayList<>(); if (dQuarantined) { reasons.add("is quarantined (backend exhausted)"); @@ -81,6 +94,11 @@ final class FixedPlacementPolicy implements PlacementPolicy { if (dCoolingOff) { reasons.add("is cooling off after repeated backend errors"); } + if (dAtCap) { + PlacementCandidate c = candidateFor(ctx, d); + int live = ctx.liveCount().apply(d); + reasons.add("is at maxLoad (" + live + " live >= " + c.maxLoad() + " cap)"); + } if (dModelOff) { reasons.add("names a model the operator has turned off in models.allow"); } @@ -96,18 +114,38 @@ final class FixedPlacementPolicy implements PlacementPolicy { } if (!ctx.candidates().isEmpty()) { throw new PlacementException("all worker profiles are excluded from automatic " - + "selection (quarantined, cooling off, model-off, unreachable, or weight-0)"); + + "selection (quarantined, cooling off, at cap, model-off, unreachable, or weight-0)"); } throw new PlacementException("no worker profiles configured"); } /** Whether {@code profile} carries {@code weight <= 0} (CB-554) among {@code ctx}'s candidates. */ private static boolean weightExcluded(PlacementContext ctx, String profile) { + PlacementCandidate c = candidateFor(ctx, profile); + return c != null && c.excluded(); + } + + /** + * Whether {@code profile} has reached its {@code maxLoad} cap (fleetd #435), using the shared + * {@link PlacementPolicyUtil#atCap} definition — the same one {@code weighted}/{@code + * round-robin} already consult via {@link PlacementPolicyUtil#available}. Looked up by name, + * the same way {@link #weightExcluded} is: the default fast path above builds its own {@link + * PlacementCandidate} with {@code maxLoad} forced to {@code null} (it carries no cap of its + * own), so the candidate actually configured for {@code profile} has to be found in {@code + * ctx.candidates()} first. + */ + private static boolean capExcluded(PlacementContext ctx, String profile) { + PlacementCandidate c = candidateFor(ctx, profile); + return c != null && PlacementPolicyUtil.atCap(ctx, c); + } + + /** The configured candidate named {@code profile} in {@code ctx}, or {@code null} if none. */ + private static PlacementCandidate candidateFor(PlacementContext ctx, String profile) { for (PlacementCandidate c : ctx.candidates()) { if (c.profile().equals(profile)) { - return c.excluded(); + return c; } } - return false; + return null; } } diff --git a/fleetd/src/main/java/dev/ltms/fleet/placement/PlacementPolicyUtil.java b/fleetd/src/main/java/dev/ltms/fleet/placement/PlacementPolicyUtil.java index d80a51c..9539be3 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/placement/PlacementPolicyUtil.java +++ b/fleetd/src/main/java/dev/ltms/fleet/placement/PlacementPolicyUtil.java @@ -11,14 +11,29 @@ final class PlacementPolicyUtil { private PlacementPolicyUtil() { } + /** + * True when {@code c} has reached its {@code maxLoad} cap: {@code liveCount(c.profile()) >= + * c.maxLoad()}. A {@code null} maxLoad means unlimited, so it is never at cap. + * + *

Extracted as the single shared definition of "at cap" (fleetd #435): before this fix it + * was computed inline in both {@link #available} and {@link #emptyException}, and {@code + * FixedPlacementPolicy} — not a caller of either — quietly kept its own {@code select} free of + * any cap check at all, so a capped default profile was chosen anyway under the default + * placement policy. Every automatic policy must call this, not re-derive it. + */ + static boolean atCap(PlacementContext ctx, PlacementCandidate c) { + Integer cap = c.maxLoad(); + return cap != null && ctx.liveCount().apply(c.profile()) >= cap; + } + /** * Candidates that are not weight-excluded (CB-554: explicit {@code weight <= 0}, checked * first because it is a static config choice rather than transient state), not * known-unreachable, not quarantined (CB-578 stage B), not cooling off after repeated backend * errors (fleetd #201 Unit 5 — a separate, shorter-lived source from quarantine), not naming a * model the operator has turned off (fleetd #422 — a third, independent source: an operator - * decision, never a backend-reported outage), and have not reached their maxLoad. A {@code - * null} maxLoad means unlimited. + * decision, never a backend-reported outage), and have not reached their maxLoad (see {@link + * #atCap}). A {@code null} maxLoad means unlimited. */ static List available(PlacementContext ctx) { List out = new ArrayList<>(); @@ -26,16 +41,10 @@ final class PlacementPolicyUtil { if (c.excluded() || ctx.unreachable().contains(c.profile()) || ctx.quarantined().contains(c.profile()) || ctx.coolingOff().contains(c.profile()) - || ctx.modelOff().contains(c.profile())) { + || ctx.modelOff().contains(c.profile()) + || atCap(ctx, c)) { continue; } - Integer cap = c.maxLoad(); - if (cap != null) { - int live = ctx.liveCount().apply(c.profile()); - if (live >= cap) { - continue; - } - } out.add(c); } return out; @@ -59,7 +68,6 @@ final class PlacementPolicyUtil { int coolingOff = 0; int modelOff = 0; for (PlacementCandidate c : ctx.candidates()) { - Integer cap = c.maxLoad(); if (c.excluded()) { weightExcluded++; } else if (ctx.quarantined().contains(c.profile())) { @@ -70,7 +78,7 @@ final class PlacementPolicyUtil { modelOff++; } else if (ctx.unreachable().contains(c.profile())) { unreachable++; - } else if (cap != null && ctx.liveCount().apply(c.profile()) >= cap) { + } else if (atCap(ctx, c)) { atCap++; } } diff --git a/fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java b/fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java index 9d695b2..26fdd29 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java @@ -536,6 +536,31 @@ class CompositePeerLauncherTest { assertEquals("claude", h.profile(), "the returned handle carries the resolved default profile"); } + /** + * fleetd #435: {@code FixedPlacementPolicy} — the default placement policy every config uses + * unless {@code placement:} is set — never consulted {@code maxLoad}, so an unqualified spawn + * (a blank profile, the normal delegation path) landed on a capped default anyway. Measured on + * 7667727: a single dev profile at {@code maxLoad: 1} with 1 live, under {@code fixed()}, + * returned "SPAWNED on profile=a". This test goes through {@code CompositePeerLauncher.spawn} + * with a blank profile, not the policy in isolation, so it proves the caller actually reaches + * the fixed default's new cap check rather than only the {@code select} method. + */ + @Test + void fixedPolicyGatesDefaultProfileAtMaxLoadOnUnqualifiedSpawn() { + FakeHerdr herdr = new FakeHerdr(); + Map profiles = ordered( + "a", stubWorker("a", 1.0f, 1), + "b", stubWorker("b", 1.0f, null)); + StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "a", Set.of()); + CompositePeerLauncher composite = new CompositePeerLauncher( + List.of(adapter), "a", profiles, PlacementPolicies.fixed(), name -> "a".equals(name) ? 1 : 0); + + PeerHandle h = composite.spawn(new SpawnRequest(null, null, null)); + assertEquals("b", h.profile(), + "the default profile a is at maxLoad, so fixed placement must fall through to b"); + assertEquals(0, adapter.spawnCount("a"), "a is never spawned — it is already at cap"); + } + @Test void weightedPolicyGatesProfileAtMaxLoad() { FakeHerdr herdr = new FakeHerdr(); diff --git a/fleetd/src/test/java/dev/ltms/fleet/placement/PlacementPolicyTest.java b/fleetd/src/test/java/dev/ltms/fleet/placement/PlacementPolicyTest.java index 8f0f73b..16fcfd3 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/placement/PlacementPolicyTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/placement/PlacementPolicyTest.java @@ -461,6 +461,93 @@ class PlacementPolicyTest { assertTrue(e.getMessage().contains("weight 0"), e.getMessage()); } + // --- fleetd #435: FixedPlacementPolicy must consult maxLoad too, at BOTH filter sites ------- + + /** + * The default-profile fast path must skip a capped default. Measured on 7667727 before this + * fix: a single dev profile at {@code maxLoad: 1} with 1 live, under {@code fixed()}, still + * returned "SPAWNED on profile=a" — the cap was advisory for every unqualified spawn. + */ + @Test + void fixedSkipsCappedDefault() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext("b", + List.of(PlacementCandidate.profile("a", 1.0f, null), + PlacementCandidate.profile("b", 1.0f, 1)), + name -> "b".equals(name) ? 1 : 0, Set.of(), Set.of(), Set.of()); + assertEquals("a", policy.select(ctx).profile(), + "the default 'b' is at its maxLoad cap, so fixed falls through to the free candidate 'a'"); + } + + /** + * The fallback walk must skip a capped candidate too — exercised independently of the + * default-profile fast path by using no default at all, so this is the only filter that runs. + */ + @Test + void fixedFallbackWalkSkipsCappedCandidate() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext(null, + List.of(PlacementCandidate.profile("a", 1.0f, 1), + PlacementCandidate.profile("b", 1.0f, null)), + name -> "a".equals(name) ? 1 : 0, Set.of(), Set.of(), Set.of()); + assertEquals("b", policy.select(ctx).profile(), + "candidate 'a' is at its maxLoad cap, so the fallback walk skips it and picks 'b'"); + } + + @Test + void fixedThrowsWhenDefaultAndEveryCandidateAtCap() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext("b", + List.of(PlacementCandidate.profile("a", 1.0f, 1), + PlacementCandidate.profile("b", 1.0f, 1)), + _ -> 1, Set.of(), Set.of(), Set.of()); + PlacementException e = assertThrows(PlacementException.class, () -> policy.select(ctx)); + assertTrue(e.getMessage().contains("maxLoad"), "message names the cap: " + e.getMessage()); + assertTrue(e.getMessage().contains("1 cap"), "message names the cap value: " + e.getMessage()); + assertFalse(e.getMessage().contains("quarantined"), "must not read like quarantine: " + e.getMessage()); + assertFalse(e.getMessage().contains("turned off"), "must not read like model-off: " + e.getMessage()); + } + + /** The mirror: an uncapped default is still chosen, so the new term cannot exclude everything. */ + @Test + void fixedStillReturnsUncappedDefault() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext("b", + List.of(PlacementCandidate.profile("a", 1.0f, null), + PlacementCandidate.profile("b", 1.0f, null)), + noSessions(), Set.of(), Set.of(), Set.of()); + assertEquals("b", policy.select(ctx).profile(), "an uncapped default is returned unconditionally"); + } + + /** CB-585: an explicit {@code maxLoad: 0} on the default caps it at zero live members. */ + @Test + void fixedSkipsMaxLoadZeroDefaultEvenWithZeroLiveWorkers() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext("b", + List.of(PlacementCandidate.profile("a", 1.0f, null), + PlacementCandidate.profile("b", 1.0f, 0)), + _ -> 0, Set.of(), Set.of(), Set.of()); + assertEquals("a", policy.select(ctx).profile(), + "the default 'b' has maxLoad 0, so it is already at its cap with nobody live"); + } + + /** + * Quarantine still wins when a profile is both quarantined and at cap (mirrors {@code + * fixedReportsQuarantineNotModelOffWhenBothApply}'s priority over model-off). + */ + @Test + void fixedReportsQuarantineNotAtCapWhenBothApply() { + PlacementPolicy policy = PlacementPolicies.fixed(); + PlacementContext ctx = new PlacementContext("b", + List.of(PlacementCandidate.profile("a", 1.0f, 1), + PlacementCandidate.profile("b", 1.0f, 1)), + _ -> 1, Set.of(), Set.of("a", "b"), Set.of()); + PlacementException e = assertThrows(PlacementException.class, () -> policy.select(ctx)); + assertTrue(e.getMessage().contains("quarantined"), e.getMessage()); + assertFalse(e.getMessage().contains("maxLoad"), + "quarantine takes priority over at-cap in the message: " + e.getMessage()); + } + @Test void unknownPolicyNameThrows() { assertThrows(IllegalArgumentException.class, () -> PlacementPolicies.fromName("random"));