From ab70314281257f19ee516f55d748d2ee2d771053 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Mon, 7 Sep 2026 16:05:23 +0500 Subject: [PATCH] fix(garm): confirm current demand before registering capacity Signed-off-by: rldyourmnd --- CHANGELOG.md | 22 +++- config/example-runner-1.yaml | 2 +- config/example-runner-2.yaml | 2 +- config/example-runner-3.yaml | 2 +- config/example-runner-4.yaml | 2 +- config/example-services.yaml | 2 +- config/garm-derivative.yaml | 23 +++- docs/runbooks/fleet-alerts.md | 22 ++++ scripts/build-garm-nddev.sh | 16 ++- .../workers/scaleset/confirmed_demand.go | 59 +++++++++ .../workers/scaleset/confirmed_demand_test.go | 123 ++++++++++++++++++ .../overlay/workers/scaleset/queue_intent.go | 30 +++-- .../workers/scaleset/queue_intent_test.go | 72 +++++++++- .../0030-confirm-live-scale-set-demand.patch | 28 ++++ 14 files changed, 365 insertions(+), 40 deletions(-) create mode 100644 third_party/garm/overlay/workers/scaleset/confirmed_demand.go create mode 100644 third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go create mode 100644 third_party/garm/patches/0030-confirm-live-scale-set-demand.patch diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c9fc33b..f84b24a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## Unreleased +- Confirm current GitHub scale-set demand before each new JIT registration. + A retained local assignment requests reconciliation but cannot allocate a + runner against a fresh zero-demand snapshot. Reads have a five-second + deadline and a thirty-second backoff after zero, invalid or failed evidence; + a stale local zero can still recover against fresh positive demand. Existing + jobs, queue timestamps and runners are preserved. Fast jobs whose start and + completion share one message batch no longer retain running capacity, and + terminal records from earlier writers cannot request more runners. This is + GARM `v0.2.1-nddev.92`; runtime adoption is a separate deployment operation. + - Recovery requires exact complete progress evidence, preserves unresolved identities on checkpoint/restart failures, and never replays interrupted restart authorization. Blocked/suppressed stalled work remains unhealthy; @@ -17,14 +27,12 @@ lost POST reply. This publisher does not invent an agent consumer or execute log text. -- Scale up from durable admitted queue ownership, not GitHub - `DesiredRunnerCount`. After a sibling runner is deleted, GitHub reports - zero assigned jobs while `JobAssigned` waiters still need a runner, so - autoscale never called create and Almaty `Candidate certified` cycled a - new UUID every five minutes with the scale set at 0. Pre-job creates +- GARM `v0.2.1-nddev.91` introduced demand reconciliation from durable admitted + queue ownership when the persisted `DesiredRunnerCount` was zero. The later + `.92` change above requires current GitHub evidence before actual creation. + Pre-job creates also bind a queued non-terminal waiter when the journal is not yet - assigned, and terminal lineage is omitted from retry inventory. This is - GARM `v0.2.1-nddev.91`. + assigned, and terminal lineage is omitted from retry inventory. - Pack 4 GiB workers onto the member with the least remaining memory that still fits, instead of spreading onto empty 16 GiB hosts. The emptiest- diff --git a/config/example-runner-1.yaml b/config/example-runner-1.yaml index 79da8130..844073ac 100644 --- a/config/example-runner-1.yaml +++ b/config/example-runner-1.yaml @@ -6,7 +6,7 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.91 + manager_version: v0.2.1-nddev.92 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.129 diff --git a/config/example-runner-2.yaml b/config/example-runner-2.yaml index faa0b8b5..9e09bd0f 100644 --- a/config/example-runner-2.yaml +++ b/config/example-runner-2.yaml @@ -6,7 +6,7 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.91 + manager_version: v0.2.1-nddev.92 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.129 diff --git a/config/example-runner-3.yaml b/config/example-runner-3.yaml index 3bb49b3e..f6124603 100644 --- a/config/example-runner-3.yaml +++ b/config/example-runner-3.yaml @@ -6,7 +6,7 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.91 + manager_version: v0.2.1-nddev.92 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.129 diff --git a/config/example-runner-4.yaml b/config/example-runner-4.yaml index 48381da1..a48e7bb8 100644 --- a/config/example-runner-4.yaml +++ b/config/example-runner-4.yaml @@ -6,7 +6,7 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.91 + manager_version: v0.2.1-nddev.92 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.129 diff --git a/config/example-services.yaml b/config/example-services.yaml index dcda52f8..6701a587 100644 --- a/config/example-services.yaml +++ b/config/example-services.yaml @@ -24,7 +24,7 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.91 + manager_version: v0.2.1-nddev.92 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.129 diff --git a/config/garm-derivative.yaml b/config/garm-derivative.yaml index 778b386a..99cd6c73 100644 --- a/config/garm-derivative.yaml +++ b/config/garm-derivative.yaml @@ -1,6 +1,6 @@ schema_version: 1 artifact: garm -derivative_version: v0.2.1-nddev.91 +derivative_version: v0.2.1-nddev.92 upstream: repository: https://github.com/cloudbase/garm release: v0.2.1 @@ -93,20 +93,29 @@ patches: purpose: Treat a completed workflow run as authoritative over a queued child job so cancelled runs release their exact queue intent and cannot be resurrected during scale-set reconciliation. - path: third_party/garm/patches/0029-scale-up-from-admitted-intent.patch sha256: e63c57a3c0a9d492ba45bc35f02e1f9abbc84e7e7923717b3be6ff45206f8a0d - purpose: Scale up from durable admitted queue ownership capped by MaxRunners, not GitHub DesiredRunnerCount. Desired drops to zero after a sibling runner is deleted while JobAssigned waiters still need a runner, which is the assigned-without-instance stall. + purpose: Enter demand reconciliation from durable admitted ownership even when the persisted GitHub count is zero; the later confirmed-demand patch gates actual JIT registration. + - path: third_party/garm/patches/0030-confirm-live-scale-set-demand.patch + sha256: 102aebd1fb51a7d4bd619fbbb5ab6e3e5706677e35cdbd6c76663640a24f1497 + purpose: Confirm current GitHub scale-set demand with a bounded read before JIT registration, preserving admission limits and queued identities while rejecting stale local-only demand. overlays: - path: third_party/garm/overlay/workers/scaleset/queue_intent.go - sha256: b20c6261ad7896ae3ca56befb64fefb5cd565322d77976af2135ca9e386643f1 - purpose: Fsync-backed central queue admission that scales up from admitted ownership when GitHub desired is zero, demotes an expired never-started assigned waiter to queued instead of deleting it, refreshes queued TTL up to the execution horizon, and FIFO-orders by FirstQueuedAt so a rewritten QueueTime cannot starve a GitHub-queued job behind a later UUID on the same label. + sha256: be85eac2126d7723483467958222f401061bd08a68113af61ff4b51208f717d7 + purpose: Fsync-backed central queue admission that excludes terminal capacity, gives completion precedence in fast start/complete batches, requests fresh demand when persisted desired is zero, demotes an expired never-started assigned waiter to queued instead of deleting it, refreshes queued TTL up to the execution horizon, and FIFO-orders by FirstQueuedAt so a rewritten QueueTime cannot starve a GitHub-queued job behind a later UUID on the same label. - path: third_party/garm/overlay/workers/scaleset/queue_intent_test.go - sha256: a060182f2105307c265e6d8c8064bb4850f21af465bcbeaae3502f2a1360795b - purpose: Prove scale-up from an admitted waiter when GitHub desired is zero, expired never-started assigned waiters keep FIFO over later JobAssigned tokens, queued waiters refresh until the execution horizon then drop, FirstQueuedAt wins when QueueTime moves, plus lineage transfer, bootstrap, fairness, width and acknowledgement coverage. + sha256: e4680119fd004f56db9801583de3781e70597b1a0f7c79eee9edd7cb95d80cdf + purpose: Prove completed batched execution cannot retain running capacity, an admitted waiter requests reconciliation when persisted desired is zero, expired never-started assigned waiters keep FIFO over later JobAssigned tokens, queued waiters refresh until the execution horizon then drop, FirstQueuedAt wins when QueueTime moves, plus lineage transfer, bootstrap, fairness, width and acknowledgement coverage. - path: third_party/garm/overlay/workers/provider/nddev_create_retry.go sha256: 2fd202d890088680ef9257e0a3366855b6e4d216c41a7a8936d7866998021219 purpose: Fsync-backed schema-v2 instance-to-intent reservations that bind a pre-job create to a queued non-terminal waiter when no assigned intent exists yet, skip terminal lineage from active inventory, and treat job-keyed capacity refusals as cluster backpressure that never opens a 24-hour circuit. - path: third_party/garm/overlay/workers/provider/nddev_create_retry_test.go sha256: c52b783f1b420a3bb15fdc3ca7a445e295251e82e64a85f4d2b41de474546eff purpose: Prove a pre-job create binds the live queued waiter rather than terminal lineage, unique-owner reconstruction, tenant-qualified shared-capacity ownership, selective terminal pruning, blocked-intent preservation, and fail-closed behavior when every exact intent is blocked. + - path: third_party/garm/overlay/workers/scaleset/confirmed_demand.go + sha256: 5c289dc30a52e5329e3d4887660629e7d12f855b5c8204d2d99e68c003f39458 + purpose: Bound each new JIT registration by current GitHub demand with a five-second request deadline and thirty-second negative/error backoff; preserve existing jobs and workers. + - path: third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go + sha256: 32f7485aeae73c15b28148d8d244e93a35cd8fb6dccea99a418db890107604d8 + purpose: Prove fresh zero demand prevents allocation, stale persisted zero can recover against fresh positive demand, positive observations are not reused, and missing, mismatched or unavailable evidence fails closed. build: container_image: docker.io/library/golang@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36 go_version: go1.26.6 @@ -121,7 +130,7 @@ build: - sqlite_omit_load_extension reproducible_rebuilds: 2 maximum_required_glibc: "2.34" - binary_sha256: f57cf8d0860745106d21b38cfa097f284b710906ea6d8303b3be4fd913b99d97 + binary_sha256: c112777344debf7e9c0b82524b6de381872ff6576e700cc4c4469b136770b560 runtime_contract: queue_intent_schema_version: 6 event_driven_scale_set_wake: true diff --git a/docs/runbooks/fleet-alerts.md b/docs/runbooks/fleet-alerts.md index 057d612d..0fe04c09 100644 --- a/docs/runbooks/fleet-alerts.md +++ b/docs/runbooks/fleet-alerts.md @@ -3,6 +3,28 @@ Every fleet alert is symptom-oriented. Confirm the exact metric and correlated queue intent, GitHub job, provider lease and Incus instance before recovery. +For queue and lifecycle age alerts, `observed` is seconds, not a job or runner +count. The notification's `since` value is the alert observation timestamp; +use the exact journal identity and its first-queued/state-entry timestamps to +reconstruct the wait. A lower aggregate maximum can mean that one job advanced +or was cancelled while other work remains stalled. Check its actual outcome. + +A REST workflow job can remain queued while the scale-set API reports no jobs +assigned to that scale set. Preserve both observations. GARM confirms the +scale-set API's current `statistics.TotalAssignedJobs` before creating a new +JIT registration, bounded by local admission and the configured maximum. An +old local `JobAssigned` record is a reason to reconcile; it is not independent +proof of dispatchable demand. Conversely, an old persisted zero must not block +an authoritative read that could show new demand. GitHub documents this +distinction in the [scale-set client autoscaling contract](https://github.com/actions/scaleset#autoscaling). + +Compare a start with its completion in the same message batch when diagnosing +fast jobs. Completion is terminal even if no intermediate running observation +was captured. Never interpret a retained terminal record, a refreshed admission +TTL, a normal process heartbeat or a quieter alert as proof that the exact job +started. The confirmed-demand guard does not cancel or rerun those jobs, delete +online runners, repair a lost upstream assignment, or prove runtime recovery. + OpenTelemetry collects and transforms every fleet signal, OTLP/HTTP transports it, and OpenObserve stores, queries, dashboards and alerts it. PromQL in this contract is OpenObserve's query syntax; no Prometheus server, agent, diff --git a/scripts/build-garm-nddev.sh b/scripts/build-garm-nddev.sh index 5d07be44..02725425 100755 --- a/scripts/build-garm-nddev.sh +++ b/scripts/build-garm-nddev.sh @@ -19,7 +19,7 @@ set -Eeuo pipefail # Every value below is the manifest's. Editing one here detaches the build # from the provenance it is reviewed against, which is why the region is # regenerated and compared rather than maintained. -readonly derivative_version="v0.2.1-nddev.91" +readonly derivative_version="v0.2.1-nddev.92" readonly upstream_repository="https://github.com/cloudbase/garm" readonly upstream_commit="154638445c3949c1958b01812f69d9a1e4d82684" readonly build_image="docker.io/library/golang@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36" @@ -32,7 +32,7 @@ readonly build_module_mode="vendor" readonly build_tags="osusergo,netgo,sqlite_omit_load_extension" readonly build_reproducible_rebuilds="2" readonly build_maximum_required_glibc="2.34" -readonly expected_binary_sha256="f57cf8d0860745106d21b38cfa097f284b710906ea6d8303b3be4fd913b99d97" +readonly expected_binary_sha256="c112777344debf7e9c0b82524b6de381872ff6576e700cc4c4469b136770b560" readonly patch_paths=( "third_party/garm/patches/0001-event-driven-reconciliation.patch" "third_party/garm/patches/0002-central-queue-admission.patch" @@ -63,6 +63,7 @@ readonly patch_paths=( "third_party/garm/patches/0027-bound-upstream-download-attempts.patch" "third_party/garm/patches/0028-terminal-workflow-run-reconciliation.patch" "third_party/garm/patches/0029-scale-up-from-admitted-intent.patch" + "third_party/garm/patches/0030-confirm-live-scale-set-demand.patch" ) readonly patch_sha256s=( "2f0571f141e7388d6ea0cb0341549ba5bf5dab26d0006382a71b76655e272d34" @@ -94,24 +95,31 @@ readonly patch_sha256s=( "35f257c8016c7999c8780970f673c5b35e04b3b3301565cffea0bd242e168456" "3bc2722e87f1721c6094130d0f07310a796bf8074504ef121acac8ca7402887c" "e63c57a3c0a9d492ba45bc35f02e1f9abbc84e7e7923717b3be6ff45206f8a0d" + "102aebd1fb51a7d4bd619fbbb5ab6e3e5706677e35cdbd6c76663640a24f1497" ) readonly overlay_paths=( "third_party/garm/overlay/workers/scaleset/queue_intent.go" "third_party/garm/overlay/workers/scaleset/queue_intent_test.go" "third_party/garm/overlay/workers/provider/nddev_create_retry.go" "third_party/garm/overlay/workers/provider/nddev_create_retry_test.go" + "third_party/garm/overlay/workers/scaleset/confirmed_demand.go" + "third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go" ) readonly overlay_sha256s=( - "b20c6261ad7896ae3ca56befb64fefb5cd565322d77976af2135ca9e386643f1" - "a060182f2105307c265e6d8c8064bb4850f21af465bcbeaae3502f2a1360795b" + "be85eac2126d7723483467958222f401061bd08a68113af61ff4b51208f717d7" + "e4680119fd004f56db9801583de3781e70597b1a0f7c79eee9edd7cb95d80cdf" "2fd202d890088680ef9257e0a3366855b6e4d216c41a7a8936d7866998021219" "c52b783f1b420a3bb15fdc3ca7a445e295251e82e64a85f4d2b41de474546eff" + "5c289dc30a52e5329e3d4887660629e7d12f855b5c8204d2d99e68c003f39458" + "32f7485aeae73c15b28148d8d244e93a35cd8fb6dccea99a418db890107604d8" ) readonly overlay_targets=( "workers/scaleset/queue_intent.go" "workers/scaleset/queue_intent_test.go" "workers/provider/nddev_create_retry.go" "workers/provider/nddev_create_retry_test.go" + "workers/scaleset/confirmed_demand.go" + "workers/scaleset/confirmed_demand_test.go" ) # END GENERATED REGION diff --git a/third_party/garm/overlay/workers/scaleset/confirmed_demand.go b/third_party/garm/overlay/workers/scaleset/confirmed_demand.go new file mode 100644 index 00000000..e041627a --- /dev/null +++ b/third_party/garm/overlay/workers/scaleset/confirmed_demand.go @@ -0,0 +1,59 @@ +package scaleset + +import ( + "context" + "fmt" + "time" + + "github.com/cloudbase/garm/params" +) + +const ( + demandReadTimeout = 5 * time.Second + demandReadBackoff = 30 * time.Second +) + +type scaleSetDemandReader interface { + GetRunnerScaleSetByID(context.Context, int) (params.RunnerScaleSet, error) +} + +// confirmedDemandGate bounds new JIT registrations by GitHub's current demand. +// Retained JobAssigned records preserve audit/FIFO identity, but cannot prove +// that GitHub still assigns that work to this scale set. Conversely, the last +// message's persisted count may be stale-low. Read the authoritative snapshot +// before allocating; keep every existing job and runner untouched on refusal. +// The owning Worker's mutex serializes access to nextRead. +type confirmedDemandGate struct { + nextRead time.Time +} + +func (g *confirmedDemandGate) target(ctx context.Context, now time.Time, reader scaleSetDemandReader, scaleSet params.ScaleSet, current, admitted int) (int, error) { + if !scaleSet.Enabled || admitted <= current || admitted < 1 || scaleSet.MaxRunners < 1 { + return current, nil + } + if now.Before(g.nextRead) { + return current, nil + } + // Failed, missing and zero-demand observations all have bounded read cost. + // A successful positive observation is never cached across a new create. + g.nextRead = now.Add(demandReadBackoff) + readCtx, cancel := context.WithTimeout(ctx, demandReadTimeout) + defer cancel() + remote, err := reader.GetRunnerScaleSetByID(readCtx, scaleSet.ScaleSetID) + if err != nil { + return current, fmt.Errorf("read current scale-set demand: %w", err) + } + if remote.ID != scaleSet.ScaleSetID || remote.Name != scaleSet.Name || remote.Enabled == nil || !*remote.Enabled || remote.Statistics == nil { + return current, fmt.Errorf("current scale-set demand has missing or mismatched identity, enabled state or statistics") + } + desired := remote.Statistics.TotalAssignedJobs + if desired < 0 || remote.Statistics.TotalRunningJobs < 0 || remote.Statistics.TotalRunningJobs > desired { + return current, fmt.Errorf("current scale-set demand has invalid job counts") + } + target := min(admitted, desired, int(scaleSet.MaxRunners)) + if target <= current { + return current, nil + } + g.nextRead = time.Time{} + return target, nil +} diff --git a/third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go b/third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go new file mode 100644 index 00000000..dc985bb1 --- /dev/null +++ b/third_party/garm/overlay/workers/scaleset/confirmed_demand_test.go @@ -0,0 +1,123 @@ +package scaleset + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/cloudbase/garm/params" +) + +type demandReaderFunc func(context.Context, int) (params.RunnerScaleSet, error) + +func (f demandReaderFunc) GetRunnerScaleSetByID(ctx context.Context, id int) (params.RunnerScaleSet, error) { + return f(ctx, id) +} + +func demandFixture() (params.ScaleSet, params.RunnerScaleSet) { + enabled := true + local := params.ScaleSet{ScaleSetID: 5, Name: "example-integration", Enabled: true, MaxRunners: 8, DesiredRunnerCount: 0} + remote := params.RunnerScaleSet{ID: 5, Name: local.Name, Enabled: &enabled, Statistics: ¶ms.RunnerScaleSetStatistic{TotalAssignedJobs: 3}} + return local, remote +} + +func TestConfirmedDemandRetainedWaiterCannotCreateAgainstFreshZero(t *testing.T) { + local, remote := demandFixture() + remote.Statistics.TotalAssignedJobs = 0 + now := time.Now() + var gate confirmedDemandGate + reads := 0 + reader := demandReaderFunc(func(ctx context.Context, id int) (params.RunnerScaleSet, error) { + reads++ + if id != local.ScaleSetID { + t.Fatalf("wrong scale-set read: %d", id) + } + deadline, ok := ctx.Deadline() + if !ok || time.Until(deadline) > demandReadTimeout { + t.Fatal("demand read must have a bounded deadline") + } + return remote, nil + }) + for _, elapsed := range []time.Duration{0, time.Second, 5 * time.Second, 29 * time.Second} { + target, err := gate.target(context.Background(), now.Add(elapsed), reader, local, 0, 4) + if err != nil || target != 0 { + t.Fatalf("retained waiters allocated capacity: target=%d err=%v", target, err) + } + } + if reads != 1 { + t.Fatalf("zero-demand read storm: %d reads", reads) + } + remote.Statistics.TotalAssignedJobs = 2 + target, err := gate.target(context.Background(), now.Add(demandReadBackoff), reader, local, 0, 4) + if err != nil || target != 2 || reads != 2 { + t.Fatalf("new positive demand did not reopen: target=%d reads=%d err=%v", target, reads, err) + } +} + +func TestConfirmedDemandRefreshesStaleZeroAndDoesNotCachePositive(t *testing.T) { + local, remote := demandFixture() + var gate confirmedDemandGate + reads := 0 + reader := demandReaderFunc(func(context.Context, int) (params.RunnerScaleSet, error) { reads++; return remote, nil }) + target, err := gate.target(context.Background(), time.Now(), reader, local, 0, 5) + if err != nil || target != 3 { + t.Fatalf("stale local zero stranded current work: target=%d err=%v", target, err) + } + remote.Statistics.TotalAssignedJobs = 0 + target, err = gate.target(context.Background(), time.Now(), reader, local, 1, 5) + if err != nil || target != 1 || reads != 2 { + t.Fatalf("stale positive demand permitted another create: target=%d reads=%d err=%v", target, reads, err) + } +} + +func TestConfirmedDemandPreservesAdmissionAndMaximum(t *testing.T) { + for _, tc := range []struct{ admitted, maximum, current, want int }{{2, 8, 0, 2}, {5, 1, 0, 1}, {0, 8, 0, 0}, {2, 8, 2, 2}, {4, 8, 5, 5}} { + local, remote := demandFixture() + local.MaxRunners = uint(tc.maximum) + var gate confirmedDemandGate + target, err := gate.target(context.Background(), time.Now(), demandReaderFunc(func(context.Context, int) (params.RunnerScaleSet, error) { return remote, nil }), local, tc.current, tc.admitted) + if err != nil || target != tc.want { + t.Fatalf("%+v: target=%d err=%v", tc, target, err) + } + } +} + +func TestConfirmedDemandUnknownOrWrongIdentityFailsClosed(t *testing.T) { + for _, kind := range []string{"read-error", "missing-stats", "wrong-id", "wrong-name", "disabled", "unknown-enabled", "negative", "running-exceeds-assigned"} { + t.Run(kind, func(t *testing.T) { + local, remote := demandFixture() + var readErr error + switch kind { + case "read-error": + readErr = errors.New("unavailable") + case "missing-stats": + remote.Statistics = nil + case "wrong-id": + remote.ID++ + case "wrong-name": + remote.Name = "example-other" + case "disabled": + *remote.Enabled = false + case "unknown-enabled": + remote.Enabled = nil + case "negative": + remote.Statistics.TotalAssignedJobs = -1 + case "running-exceeds-assigned": + remote.Statistics.TotalRunningJobs = 4 + } + var gate confirmedDemandGate + reads := 0 + now := time.Now() + reader := demandReaderFunc(func(context.Context, int) (params.RunnerScaleSet, error) { reads++; return remote, readErr }) + target, err := gate.target(context.Background(), now, reader, local, 1, 4) + if err == nil || target != 1 { + t.Fatalf("unverified demand permitted create: target=%d err=%v", target, err) + } + _, _ = gate.target(context.Background(), now.Add(time.Second), reader, local, 1, 4) + if reads != 1 { + t.Fatal("failed read was not bounded") + } + }) + } +} diff --git a/third_party/garm/overlay/workers/scaleset/queue_intent.go b/third_party/garm/overlay/workers/scaleset/queue_intent.go index 4b2a3b7b..4df4cf0a 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent.go @@ -403,23 +403,24 @@ func (c *queueIntentCoordinator) HasQueuedAvailable(scaleSet params.ScaleSet, jo return pending, err } -// AdmittedCapacityTarget is the exact current runner target from durable queue -// ownership. GitHub DesiredRunnerCount is not an upper bound: it drops to zero -// while JobAssigned waiters still need a runner, which is the -// assigned-without-instance stall. Cancellations leave the journal, so this -// count already clips stale-high desired. Cap it with MaxRunners at the -// scale-up call site via admittedScaleUpTarget. +// AdmittedCapacityTarget counts non-terminal local resource ownership. It is +// an admission ceiling, not proof that GitHub can dispatch the retained jobs. +// The scale-up path confirms current GitHub demand before creating a runner; +// a stale-low persisted message count alone must not prevent that read. func (c *queueIntentCoordinator) AdmittedCapacityTarget(scaleSet params.ScaleSet, entity params.ForgeEntity) (int, error) { config, err := c.loadConfig() if err != nil { return 0, err } target := 0 - err = c.update(config, func(journal *queueIntentJournal, _ time.Time) error { + err = c.update(config, func(journal *queueIntentJournal, now time.Time) error { for _, intent := range journal.Intents { if intent.ScaleSetID != int64(scaleSet.ScaleSetID) || intent.ScaleSetName != scaleSet.Name || intent.Owner != entity.Owner { continue } + if expiry, terminal := journal.TerminalJobs[intent.JobID]; terminal && expiry.After(now) { + continue + } switch intent.State { case queueStateAssigned, queueStateAcquiring, queueStateAcquired, queueStateRunning: target++ @@ -442,9 +443,9 @@ func admittedScaleUpTarget(intentTarget, maxRunners int) int { return intentTarget } -// shouldScaleUp is true when GitHub desired or durable admitted ownership -// still needs a runner. Admitted is the floor that GitHub TotalAssignedJobs -// does not provide after the runner for a sibling job is deleted. +// shouldScaleUp requests reconciliation when either local observation suggests +// missing capacity. The create path still requires both local admission and a +// fresh GitHub demand snapshot; neither stale counter authorizes a new runner. func shouldScaleUp(current, githubDesired, admitted int) bool { return current < admitted || current < githubDesired } @@ -619,18 +620,25 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti suppressedTerminalAssignments = suppressedTerminalAssignments[:0] completedKeys := make(map[string]struct{}, len(completed)) startedKeys := make(map[string]struct{}, len(started)) + for _, job := range started { + startedKeys[queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID)] = struct{}{} + } for _, job := range completed { if !validQueueText(job.JobID) { return fmt.Errorf("completed job has invalid job ID") } key := queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID) intent, exists := journal.Intents[key] + _, startedInBatch := startedKeys[key] // GitHub may retire a capacity waiter before it ever emits // JobAvailable, then immediately assign the same workflow job under a // new UUID. Retain that unstarted intent as a terminal lineage marker; // it is excluded from admission below and lets the replacement inherit // the original queue time instead of starving at the tail forever. - if exists && intent.RunnerRequestID == 0 && intent.RunnerName == "" && intent.State != queueStateRunning { + // A fast job may start and finish in this one batch. Completion + // must win; keeping its pre-start lineage would let the started + // loop resurrect a terminal running intent and request spare VMs. + if exists && !startedInBatch && intent.RunnerRequestID == 0 && intent.RunnerName == "" && intent.State != queueStateRunning { intent.State = queueStateQueued intent.StateEnteredAt = now intent.UpdatedAt = now diff --git a/third_party/garm/overlay/workers/scaleset/queue_intent_test.go b/third_party/garm/overlay/workers/scaleset/queue_intent_test.go index 28f190f0..c940a272 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent_test.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent_test.go @@ -14,6 +14,68 @@ import ( "github.com/cloudbase/garm/params" ) +func TestFastStartedCompletedBatchDoesNotRetainRunningCapacity(t *testing.T) { + now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC) + coordinator := testQueueCoordinator(t, &now, nil) + scaleSet := testQueueScaleSet(11, "example-integration") + job := testQueueJob(101, "example-owner", "example-repository", now) + job.RunnerRequestID = 0 + job.RunnerName = "" + entity := testQueueEntityForJob(job) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{job}, nil, nil); err != nil { + t.Fatal(err) + } + started := job + started.RunnerName = "example-runner" + started.RunnerID = 202 + now = now.Add(time.Minute) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, nil, []params.ScaleSetJobMessage{started}, []params.ScaleSetJobMessage{started}); err != nil { + t.Fatal(err) + } + journal, err := readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + if _, exists := journal.Intents[queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID)]; exists { + t.Fatal("completed execution was retained as running capacity") + } + if !journal.TerminalJobs[job.JobID].After(now) { + t.Fatal("completed execution lost its terminal receipt") + } + target, err := coordinator.AdmittedCapacityTarget(scaleSet, entity) + if err != nil || target != 0 { + t.Fatalf("completed execution still requests a runner: target=%d err=%v", target, err) + } +} + +func TestTerminalRunningRecordFromOlderWriterDoesNotRequestCapacity(t *testing.T) { + now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC) + coordinator := testQueueCoordinator(t, &now, nil) + scaleSet := testQueueScaleSet(11, "example-integration") + job := testQueueJob(101, "example-owner", "example-repository", now) + job.RunnerRequestID = 0 + job.RunnerName = "example-runner" + job.RunnerID = 202 + entity := testQueueEntityForJob(job) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{job}, []params.ScaleSetJobMessage{job}, nil); err != nil { + t.Fatal(err) + } + // Older writers could leave a running record alongside its terminal + // receipt after a batched start/complete. Read it without erasing evidence. + journal, err := readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + markTerminalJob(&journal, job.JobID, now.Add(time.Hour)) + if err := writeQueueIntentJournal(coordinator.journalPath, journal); err != nil { + t.Fatal(err) + } + target, err := coordinator.AdmittedCapacityTarget(scaleSet, entity) + if err != nil || target != 0 { + t.Fatalf("legacy terminal execution requested capacity: target=%d err=%v", target, err) + } +} + func TestAdmittedCapacityIntentDisappearsOnCompletion(t *testing.T) { now := time.Date(2026, 8, 19, 7, 0, 0, 0, time.UTC) coordinator := testQueueCoordinator(t, &now, nil) @@ -66,14 +128,12 @@ func TestAdmittedScaleUpTargetCapsAtMaxRunners(t *testing.T) { } } -func TestScaleUpUsesAdmittedWhenGitHubDesiredIsZero(t *testing.T) { +func TestAdmittedWaiterRequestsDemandReconciliationWhenPersistedDesiredIsZero(t *testing.T) { t.Parallel() - // Live 2026-09-06: Candidate certified sat assigned-without-instance for - // 17 minutes because handleAutoScale compared runnerCount to GitHub - // DesiredRunnerCount (0 after the sibling runner was deleted) and never - // called handleScaleUp. + // A persisted zero must not prevent the authoritative demand read. The + // confirmed-demand tests separately prove that a fresh zero forbids create. if !shouldScaleUp(0, 0, 1) { - t.Fatal("assigned-without-instance must scale up when GitHub desired is 0") + t.Fatal("an admitted waiter must request current demand when persisted desired is 0") } if shouldScaleDown(0, 0, 1) { t.Fatal("empty pool with an admitted waiter must not scale down") diff --git a/third_party/garm/patches/0030-confirm-live-scale-set-demand.patch b/third_party/garm/patches/0030-confirm-live-scale-set-demand.patch new file mode 100644 index 00000000..988dab27 --- /dev/null +++ b/third_party/garm/patches/0030-confirm-live-scale-set-demand.patch @@ -0,0 +1,28 @@ +diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go +index e19186e..dd6a316 100644 +--- a/workers/scaleset/scaleset.go ++++ b/workers/scaleset/scaleset.go +@@ -89,6 +89,7 @@ type Worker struct { + // the autoscaler immediately, while its periodic ticker remains the + // reconciliation safety net. + autoscaleWake chan struct{} ++ demandGate confirmedDemandGate + } + + func (w *Worker) ensureScaleSetInGitHub() error { +@@ -888,6 +889,15 @@ func (w *Worker) handleScaleUp() { + slog.ErrorContext(w.ctx, "error getting scale set client", "error", err) + return + } ++ confirmed, err := w.demandGate.target(w.ctx, time.Now(), scaleSetCli, w.scaleSet, current, target) ++ if err != nil { ++ slog.ErrorContext(w.ctx, "current GitHub demand could not be confirmed; retaining queued jobs", "error", err) ++ return ++ } ++ if confirmed <= current { ++ return ++ } ++ target = confirmed + for i := current; i < target; i++ { + newRunnerName := strings.ToLower(fmt.Sprintf("%s-%s", w.scaleSet.GetRunnerPrefix(), util.NewID())) + jitConfig, err := scaleSetCli.GenerateJitRunnerConfig(w.ctx, newRunnerName, w.scaleSet.ScaleSetID)