From db3427223af90a4589eaf92e579d331d06ded179 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Sun, 6 Sep 2026 15:45:13 +0500 Subject: [PATCH] fix(garm): keep GitHub-queued waiters through assigned TTL Never-started assigned intents demote to queued instead of disappearing after 600s, so a still-queued GitHub job cannot lose its journal slot to a later UUID on the same label. Queued waiters refresh until the 24-hour execution horizon. FIFO follows FirstQueuedAt when reconciliation rewrites QueueTime. Job-keyed capacity refusals are cluster backpressure and no longer open a 24-hour create circuit. This is GARM v0.2.1-nddev.90. Fixes #406 Fixes #407 --- CHANGELOG.md | 9 ++ 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 | 20 +-- internal/queueadmission/config.go | 2 + scripts/build-garm-nddev.sh | 12 +- .../workers/provider/nddev_create_retry.go | 20 ++- .../provider/nddev_create_retry_test.go | 32 ++-- .../overlay/workers/scaleset/queue_intent.go | 78 ++++++++-- .../workers/scaleset/queue_intent_test.go | 143 ++++++++++++++++++ 13 files changed, 272 insertions(+), 54 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b148eec..04d5ba9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +- Keep a GitHub-queued job represented after its 600-second assigned TTL. + Never-started assigned intents demote to queued instead of disappearing, + queued waiters refresh until the 24-hour execution horizon, and FIFO + follows FirstQueuedAt so a rewritten QueueTime cannot skip an older job. + A `:job:` capacity refusal is cluster backpressure like an instance + refusal and no longer opens a 24-hour create circuit after three packed + warm-pool failures. This is GARM `v0.2.1-nddev.90`. Closes the skip/refill + class reported as public #406 and #407. + - Release a warm worker's exact memory reservation when placement refuses creation. Successful capacity deferrals previously skipped cleanup and retained uncreated workers until the admission lease expired. Early diff --git a/config/example-runner-1.yaml b/config/example-runner-1.yaml index 656974e3..500b73ac 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.89 + manager_version: v0.2.1-nddev.90 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.125 diff --git a/config/example-runner-2.yaml b/config/example-runner-2.yaml index 3769557c..674e9f90 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.89 + manager_version: v0.2.1-nddev.90 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.125 diff --git a/config/example-runner-3.yaml b/config/example-runner-3.yaml index 4d5982f6..d4b41de2 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.89 + manager_version: v0.2.1-nddev.90 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.125 diff --git a/config/example-runner-4.yaml b/config/example-runner-4.yaml index e3f794e1..dbbb2cf6 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.89 + manager_version: v0.2.1-nddev.90 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.125 diff --git a/config/example-services.yaml b/config/example-services.yaml index 99c88770..f1ef4bcf 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.89 + manager_version: v0.2.1-nddev.90 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.125 diff --git a/config/garm-derivative.yaml b/config/garm-derivative.yaml index 12965851..3a056501 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.89 +derivative_version: v0.2.1-nddev.90 upstream: repository: https://github.com/cloudbase/garm release: v0.2.1 @@ -93,17 +93,17 @@ 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. overlays: - path: third_party/garm/overlay/workers/scaleset/queue_intent.go - sha256: 675e114073f1673b013bb75fd48c2ed48d83ca8d9bd26cbf8d09314f7f599224 - purpose: Fsync-backed central queue admission where a GitHub-replaced, never-started waiter inherits its exact workflow-job lineage and original queue time instead of starving behind every five-minute UUID replacement, while terminal execution suppression, repository binding, running rehydration and stable phase-entry semantics remain intact, and every intent carries an immutable first-queued stamp that reconciliation cannot move so a wait can be measured at all. + sha256: 509de591878bd7b2976233a668ae050032b936db29e0de3bdb23d484ffe93503 + purpose: Fsync-backed central queue admission that 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: 7670e2e2d5972fee177d3439bcb0f55fb1e39247e03520c4c69fab79135ee1d0 - purpose: Deterministic terminal waiter lineage transfer and delayed-redelivery suppression, disabled-scale-set isolation, provisional bootstrap, authoritative repository enrichment, orphan-start rehydration, phase-entry and expiry, priority, fairness, width, idempotency, acknowledgement and concurrent-selection tests. + sha256: abbcd57768579f464f46f0d410a236f60c695fddc815232ed95002126b025446 + purpose: Prove 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: a12afd7c9a12f52fd62f4f3eda6a88be3477d96d98cd6b03b765ce32ff6bb27e - purpose: Fsync-backed schema-v2 instance-to-intent reservations and per-job attempt leases, pruning inactive terminal jobs while retaining active state, and selecting the oldest eligible tenant-qualified retry domain so same-named scale sets cannot alias across forge owners. + sha256: f00d9ac5ecc2b79ab0584c56db8890e0ea8e116a608d7ed2ee26f00056bc0f69 + purpose: Fsync-backed schema-v2 instance-to-intent reservations where job-keyed capacity refusals are cluster backpressure like instance-keyed capacity, never opening a 24-hour circuit, so a packed warm pool cannot strand a GitHub-queued job for a day after three placement failures. - path: third_party/garm/overlay/workers/provider/nddev_create_retry_test.go - sha256: f523f68a24f5888ce907f1f5aa23cda0008b29b22e74c6583207daca7670f318 - purpose: Prove unique-owner reconstruction, tenant-qualified shared-capacity ownership, stable per-job budgets, selective terminal pruning, blocked-intent preservation, next-eligible reservation, and fail-closed behavior when every exact intent is blocked or the queue journal is invalid. + sha256: 054941a9c78c82164b8d710450246d32e6fff84ada56ca4ebb12751fa9eb1dce + purpose: Prove a job-keyed capacity refusal never opens a 24-hour circuit across 32 attempts, unique-owner reconstruction, tenant-qualified shared-capacity ownership, selective terminal pruning, blocked-intent preservation, and fail-closed behavior when every exact intent is blocked. build: container_image: docker.io/library/golang@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36 go_version: go1.26.6 @@ -118,7 +118,7 @@ build: - sqlite_omit_load_extension reproducible_rebuilds: 2 maximum_required_glibc: "2.34" - binary_sha256: c94249b642821e209e747c7117982a7704c7ce1cf413a66efd0dc25b969af168 + binary_sha256: 8d55695cad2265d0cb2ad8dfaf9d49b88f61b3a9bef54539b7684c926145bf64 runtime_contract: queue_intent_schema_version: 6 event_driven_scale_set_wake: true diff --git a/internal/queueadmission/config.go b/internal/queueadmission/config.go index c2201cf6..85e19160 100644 --- a/internal/queueadmission/config.go +++ b/internal/queueadmission/config.go @@ -80,6 +80,8 @@ func (c Config) Validate() error { c.DefaultWeight < 1 || c.DefaultWeight > 100 { return fmt.Errorf("queue admission identity or count limits are invalid") } + // 600s is the never-started refresh cadence, not the drop horizon. + // GitHub-queued waiters remain represented until ExecutionTTL. if c.QueuedTTLSeconds != 600 || c.AcquiringTTLSeconds != 120 || c.AcquiredTTLSeconds != 600 || c.ExecutionTTLSeconds != 86400 || c.PriorityAgingSeconds != 300 || c.MaxRepositorySharePercent != 75 { diff --git a/scripts/build-garm-nddev.sh b/scripts/build-garm-nddev.sh index c34c37d2..10ffa907 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.89" +readonly derivative_version="v0.2.1-nddev.90" 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="c94249b642821e209e747c7117982a7704c7ce1cf413a66efd0dc25b969af168" +readonly expected_binary_sha256="8d55695cad2265d0cb2ad8dfaf9d49b88f61b3a9bef54539b7684c926145bf64" readonly patch_paths=( "third_party/garm/patches/0001-event-driven-reconciliation.patch" "third_party/garm/patches/0002-central-queue-admission.patch" @@ -100,10 +100,10 @@ readonly overlay_paths=( "third_party/garm/overlay/workers/provider/nddev_create_retry_test.go" ) readonly overlay_sha256s=( - "675e114073f1673b013bb75fd48c2ed48d83ca8d9bd26cbf8d09314f7f599224" - "7670e2e2d5972fee177d3439bcb0f55fb1e39247e03520c4c69fab79135ee1d0" - "a12afd7c9a12f52fd62f4f3eda6a88be3477d96d98cd6b03b765ce32ff6bb27e" - "f523f68a24f5888ce907f1f5aa23cda0008b29b22e74c6583207daca7670f318" + "509de591878bd7b2976233a668ae050032b936db29e0de3bdb23d484ffe93503" + "abbcd57768579f464f46f0d410a236f60c695fddc815232ed95002126b025446" + "f00d9ac5ecc2b79ab0584c56db8890e0ea8e116a608d7ed2ee26f00056bc0f69" + "054941a9c78c82164b8d710450246d32e6fff84ada56ca4ebb12751fa9eb1dce" ) readonly overlay_targets=( "workers/scaleset/queue_intent.go" diff --git a/third_party/garm/overlay/workers/provider/nddev_create_retry.go b/third_party/garm/overlay/workers/provider/nddev_create_retry.go index a3d1157b..8ea3c55d 100644 --- a/third_party/garm/overlay/workers/provider/nddev_create_retry.go +++ b/third_party/garm/overlay/workers/provider/nddev_create_retry.go @@ -330,7 +330,7 @@ func nddevBeforeProviderCreate(ctx context.Context, key string, scaleSetNames .. if record.NextAllowedAt.After(now) { return fmt.Errorf("provider create retry is deferred until %s", record.NextAllowedAt.Format(time.RFC3339)) } - capacityBackpressure := record.LastErrorClass == "capacity" && !strings.Contains(key, ":job:") + capacityBackpressure := record.LastErrorClass == "capacity" if record.Attempts >= nddevRetryMaximum && !capacityBackpressure && record.LastErrorClass != "intent" { record.TerminalUntil = now.Add(nddevRetryExecutionTTL) record.UpdatedAt = now @@ -365,13 +365,13 @@ func nddevRecordProviderCreateFailure(ctx context.Context, key string, providerE if record.LastErrorClass == "capacity" || record.LastErrorClass == "intent" { // Saturation and a canceled job whose pre-AcquireJobs intent has // already disappeared are bounded backpressure, not a broken provider. - // Capacity accumulates delay without ever opening a circuit; a completed - // provider deletion clears it immediately. Intent cancellation stays on - // the short fixed delay because no scarce resource must be awaited. - if record.LastErrorClass == "capacity" && strings.Contains(key, ":job:") && record.Attempts >= nddevRetryMaximum { - record.NextAllowedAt = now - record.TerminalUntil = now.Add(nddevRetryExecutionTTL) - } else if record.LastErrorClass == "capacity" { + // Capacity, including a job-keyed create that failed to place, + // accumulates delay without ever opening a 24h circuit; a completed + // provider deletion clears it immediately. Opening TerminalUntil on + // a job-keyed capacity refusal stranded GitHub-queued work for a day + // after three packed warm-pool refusals. Intent cancellation stays + // on the short fixed delay because no scarce resource must be awaited. + if record.LastErrorClass == "capacity" { record.NextAllowedAt = now.Add(nddevCapacityRetryDelay(key, record.Attempts)) owner := record.Owner if owner == "" { @@ -384,9 +384,7 @@ func nddevRecordProviderCreateFailure(ctx context.Context, key string, providerE record.Attempts = 1 record.NextAllowedAt = now.Add(nddevRetryBase) } - if record.LastErrorClass != "capacity" || !strings.Contains(key, ":job:") || record.Attempts < nddevRetryMaximum { - record.TerminalUntil = time.Time{} - } + record.TerminalUntil = time.Time{} journal.Records[key] = record domainKey := nddevRetryDomainKey(key) if domainKey != key { diff --git a/third_party/garm/overlay/workers/provider/nddev_create_retry_test.go b/third_party/garm/overlay/workers/provider/nddev_create_retry_test.go index e67cf0cb..446282ad 100644 --- a/third_party/garm/overlay/workers/provider/nddev_create_retry_test.go +++ b/third_party/garm/overlay/workers/provider/nddev_create_retry_test.go @@ -483,7 +483,7 @@ func TestNDDevProviderCapacityBackpressureNeverOpensCircuit(t *testing.T) { } } -func TestNDDevJobCapacityFailureStopsAfterTwoRetries(t *testing.T) { +func TestNDDevJobCapacityFailureNeverOpensCircuit(t *testing.T) { now := time.Date(2026, 8, 24, 7, 0, 0, 0, time.UTC) originalNow := nddevRetryNow nddevRetryNow = func() time.Time { return now } @@ -492,30 +492,36 @@ func TestNDDevJobCapacityFailureStopsAfterTwoRetries(t *testing.T) { t.Setenv(nddevRetryFileEnv, filepath.Join(directory, "retry.json")) t.Setenv(nddevRetryLockEnv, filepath.Join(directory, "retry.lock")) key := "scale-set:example-entity:17:job:example-job" + domainKey := "scale-set:example-entity:17" - for attempt := 1; attempt <= nddevRetryMaximum; attempt++ { + for attempt := 0; attempt < 32; attempt++ { if err := nddevBeforeProviderCreate(context.Background(), key); err != nil { - t.Fatalf("attempt %d preflight: %v", attempt, err) + t.Fatalf("job capacity attempt %d preflight: %v", attempt, err) } if err := nddevRecordProviderCreateFailure(context.Background(), key, errors.New("provider admission rejected pool: insufficient-memory")); err != nil { - t.Fatalf("attempt %d failure: %v", attempt, err) + t.Fatalf("job capacity attempt %d failure: %v", attempt, err) } journal, err := nddevReadRetryJournal(os.Getenv(nddevRetryFileEnv)) if err != nil { t.Fatal(err) } + for _, recordKey := range []string{key, domainKey, nddevCapacityDomainKey} { + record := journal.Records[recordKey] + if record.LastErrorClass != "capacity" || !record.TerminalUntil.IsZero() { + t.Fatalf("job capacity record %q opened a circuit: %#v", recordKey, record) + } + } record := journal.Records[key] - if record.Attempts != attempt || record.LastErrorClass != "capacity" { - t.Fatalf("attempt %d record=%#v", attempt, record) + if record.Attempts != min(attempt+1, nddevRetryMaximum) { + t.Fatalf("job capacity attempts=%d want=%d record=%#v", record.Attempts, min(attempt+1, nddevRetryMaximum), record) } - if attempt < nddevRetryMaximum { - now = record.NextAllowedAt - } else if record.TerminalUntil.IsZero() { - t.Fatalf("third capacity failure did not terminate the job record: %#v", record) + now = journal.Records[key].NextAllowedAt + if domainNext := journal.Records[domainKey].NextAllowedAt; domainNext.After(now) { + now = domainNext + } + if sharedNext := journal.Records[nddevCapacityDomainKey].NextAllowedAt; sharedNext.After(now) { + now = sharedNext } - } - if err := nddevBeforeProviderCreate(context.Background(), key); err == nil { - t.Fatal("fourth capacity attempt passed an open job circuit") } } diff --git a/third_party/garm/overlay/workers/scaleset/queue_intent.go b/third_party/garm/overlay/workers/scaleset/queue_intent.go index 33b33f7f..5031330b 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent.go @@ -1010,7 +1010,7 @@ func (c *queueIntentCoordinator) update(config queueAdmissionConfig, mutate func } now := c.nowUTC() migrateLegacyQueueIntentOwnership(&journal, config, now) - cleanupExpiredQueueIntents(&journal, now) + cleanupExpiredQueueIntents(&journal, config, now) cleanupExpiredTerminalJobs(&journal, now) // A restart can occur after JobAssigned was durably acknowledged but before // the scale-up worker observed its provisional token. Promote from durable @@ -1295,8 +1295,8 @@ func eligibleQueueCandidates(journal *queueIntentJournal, config queueAdmissionC if leftRepository.Pass != rightRepository.Pass { return leftRepository.Pass < rightRepository.Pass } - if !candidates[left].QueueTime.Equal(candidates[right].QueueTime) { - return candidates[left].QueueTime.Before(candidates[right].QueueTime) + if leftWait, rightWait := queueWaitSince(candidates[left]), queueWaitSince(candidates[right]); !leftWait.Equal(rightWait) { + return leftWait.Before(rightWait) } return candidates[left].Key < candidates[right].Key }) @@ -1331,13 +1331,25 @@ func queueBackgroundInFlight(journal *queueIntentJournal) int { } func effectiveQueuePriority(intent queueIntent, config queueAdmissionConfig, now time.Time) int { - if intent.Priority <= 1 || !now.After(intent.QueueTime) { + waitSince := queueWaitSince(intent) + if intent.Priority <= 1 || !now.After(waitSince) { return intent.Priority } - aged := int(now.Sub(intent.QueueTime) / (time.Duration(config.PriorityAgingSeconds) * time.Second)) + aged := int(now.Sub(waitSince) / (time.Duration(config.PriorityAgingSeconds) * time.Second)) return maxQueuePriority(1, intent.Priority-aged) } +// queueWaitSince is the FIFO clock. QueueTime is rewritten by authoritative +// reconciliation from job.CreatedAt, and where that is absent it falls forward +// to now, so a still-queued waiter can lose its place to a job that arrived +// later. FirstQueuedAt is written once. +func queueWaitSince(intent queueIntent) time.Time { + if !intent.FirstQueuedAt.IsZero() { + return intent.FirstQueuedAt + } + return intent.QueueTime +} + func queueInFlight(journal *queueIntentJournal) (int, map[string]int) { total := 0 byRepository := make(map[string]int) @@ -1479,12 +1491,58 @@ func repositoryPolicy(config queueAdmissionConfig, repository string) queueRepos return queueRepositoryPolicy{Weight: config.DefaultWeight, MaxInFlight: config.DefaultRepositoryLimit} } -func cleanupExpiredQueueIntents(journal *queueIntentJournal, now time.Time) { +func cleanupExpiredQueueIntents(journal *queueIntentJournal, config queueAdmissionConfig, now time.Time) { for key, intent := range journal.Intents { - if !intent.ExpiresAt.After(now) { - delete(journal.Intents, key) + if intent.ExpiresAt.After(now) { + continue + } + if retainNeverStartedQueueIntent(&intent, config, now) { + journal.Intents[key] = intent + continue } + delete(journal.Intents, key) + } +} + +// retainNeverStartedQueueIntent keeps a GitHub-queued waiter in the journal +// after its phase TTL. Deleting it was the skip/refill defect: GitHub still +// had the job queued, JobAssigned had already been acknowledged, and a later +// UUID on the same label took the slot. Assigned-without-a-request-id demotes +// to queued so it stops charging capacity; queued waiters refresh until the +// execution horizon. Started work and mid-acquire/acquired stalls still drop. +func retainNeverStartedQueueIntent(intent *queueIntent, config queueAdmissionConfig, now time.Time) bool { + if intent == nil || intent.RunnerName != "" || intent.State == queueStateRunning { + return false + } + if intent.FirstQueuedAt.IsZero() { + intent.FirstQueuedAt = intent.QueueTime + if intent.FirstQueuedAt.IsZero() { + intent.FirstQueuedAt = now + } + } + horizon := intent.FirstQueuedAt.Add(time.Duration(config.ExecutionTTLSeconds) * time.Second) + if !horizon.After(now) { + return false + } + switch intent.State { + case queueStateAssigned: + if intent.RunnerRequestID != 0 { + return false + } + intent.State = queueStateQueued + intent.StateEnteredAt = now + intent.UpdatedAt = now + case queueStateQueued: + intent.UpdatedAt = now + default: + return false + } + next := expiryForState(config, queueStateQueued, now) + if next.After(horizon) { + next = horizon } + intent.ExpiresAt = next + return true } func expiryForState(config queueAdmissionConfig, state queueIntentState, now time.Time) time.Time { @@ -1501,7 +1559,9 @@ func expiryForState(config queueAdmissionConfig, state queueIntentState, now tim // set. It can own a real cold provider create before JobAvailable exists; // measured release/container registration crossed 150 seconds, so the // 120-second API-call horizon expired valid ownership and hid running - // work. Use the bounded pre-start horizon, never the execution horizon. + // work. Use the bounded pre-start horizon for the capacity charge, never + // the execution horizon. Never-started assigned waiters demote to queued + // at this TTL instead of disappearing while GitHub still has the job. seconds = config.AcquiredTTLSeconds case queueStateRunning: // Running is the one state that legitimately lasts as long as a job. 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 fb2a48ec..4b2b6700 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent_test.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent_test.go @@ -128,6 +128,149 @@ func TestRedeliveredWaiterInheritsOriginalQueueAge(t *testing.T) { } } +func TestExpiredNeverStartedAssignedIsDemotedAndKeepsFIFO(t *testing.T) { + now := time.Date(2026, 9, 6, 8, 37, 30, 0, time.UTC) + coordinator := testQueueCoordinator(t, &now, nil) + scaleSet := testQueueScaleSet(5, "nddev-linux-integration") + older := testQueueJob(401, "example-owner", "example-repository", now) + older.MessageType = params.MessageTypeJobAssigned + older.RunnerRequestID = 0 + older.RunnerName = "" + entity := testQueueEntityForJob(older) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{older}, nil, nil); err != nil { + t.Fatal(err) + } + now = now.Add(time.Minute) + newer := testQueueJob(402, "example-owner", "example-repository", now) + newer.MessageType = params.MessageTypeJobAssigned + newer.RunnerRequestID = 0 + newer.RunnerName = "" + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{newer}, nil, nil); err != nil { + t.Fatal(err) + } + journal, err := readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + olderKey := queueIntentKey(int64(scaleSet.ScaleSetID), older.JobID) + newerKey := queueIntentKey(int64(scaleSet.ScaleSetID), newer.JobID) + if journal.Intents[olderKey].State != queueStateAssigned { + t.Fatalf("older waiter was not admitted: %#v", journal.Intents[olderKey]) + } + if journal.Intents[newerKey].State != queueStateQueued { + t.Fatalf("newer waiter consumed the only slot: %#v", journal.Intents[newerKey]) + } + olderFirst := journal.Intents[olderKey].FirstQueuedAt + olderQueue := journal.Intents[olderKey].QueueTime + now = now.Add(10 * time.Minute) + if err := coordinator.Validate(); err != nil { + t.Fatal(err) + } + journal, err = readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + olderIntent, exists := journal.Intents[olderKey] + if !exists { + t.Fatal("expired never-started assigned waiter was deleted while GitHub still queued it") + } + if olderIntent.State != queueStateAssigned { + t.Fatalf("older waiter lost the slot after expiry: %#v", olderIntent) + } + if !olderIntent.FirstQueuedAt.Equal(olderFirst) || !olderIntent.QueueTime.Equal(olderQueue) { + t.Fatalf("older waiter lost its wait clock: %#v", olderIntent) + } + if journal.Intents[newerKey].State != queueStateQueued { + t.Fatalf("newer waiter skipped the expired original: %#v", journal.Intents[newerKey]) + } +} + +func TestExpiredQueuedWaiterRefreshesUntilExecutionHorizon(t *testing.T) { + now := time.Date(2026, 9, 6, 8, 37, 30, 0, time.UTC) + coordinator := testQueueCoordinator(t, &now, nil) + scaleSet := testQueueScaleSet(5, "nddev-linux-integration") + running := testQueueJob(501, "example-owner", "example-repository", now) + entity := testQueueEntityForJob(running) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{running}, nil, nil); err != nil { + t.Fatal(err) + } + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, nil, []params.ScaleSetJobMessage{running}, nil); err != nil { + t.Fatal(err) + } + now = now.Add(time.Minute) + waiter := testQueueJob(502, "example-owner", "example-repository", now) + waiter.MessageType = params.MessageTypeJobAssigned + waiter.RunnerRequestID = 0 + waiter.RunnerName = "" + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{waiter}, nil, nil); err != nil { + t.Fatal(err) + } + journal, err := readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + waiterKey := queueIntentKey(int64(scaleSet.ScaleSetID), waiter.JobID) + if journal.Intents[waiterKey].State != queueStateQueued { + t.Fatalf("waiter was admitted over a running job: %#v", journal.Intents[waiterKey]) + } + firstQueued := journal.Intents[waiterKey].FirstQueuedAt + now = now.Add(10 * time.Minute) + if err := coordinator.Validate(); err != nil { + t.Fatal(err) + } + journal, err = readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + intent, exists := journal.Intents[waiterKey] + if !exists { + t.Fatal("queued waiter was dropped at the 600s cadence while GitHub still queued it") + } + if intent.State != queueStateQueued || !intent.ExpiresAt.After(now) || !intent.FirstQueuedAt.Equal(firstQueued) { + t.Fatalf("queued waiter was not refreshed: %#v", intent) + } + now = firstQueued.Add(24*time.Hour + time.Second) + if err := coordinator.Validate(); err != nil { + t.Fatal(err) + } + journal, err = readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + if _, exists := journal.Intents[waiterKey]; exists { + t.Fatalf("queued waiter survived the execution horizon: %#v", journal.Intents[waiterKey]) + } +} + +func TestQueueFIFOUsesFirstQueuedAtWhenQueueTimeMoves(t *testing.T) { + now := time.Date(2026, 9, 6, 9, 0, 0, 0, time.UTC) + config := queueAdmissionConfig{ + MaxInFlight: 2, MaxBackgroundInFlight: 2, DefaultRepositoryLimit: 2, DefaultWeight: 1, + MaxRepositorySharePercent: 75, PriorityAgingSeconds: 300, + ScaleSets: testQueueScaleSetResourceMap(), + Repositories: map[string]queueRepositoryPolicy{}, + } + journal := queueIntentJournal{ + Intents: map[string]queueIntent{ + "moved": { + Key: "moved", State: queueStateQueued, Repository: "owner/moved", + ScaleSetName: "nddev-linux-standard", Priority: 1, + QueueTime: now, FirstQueuedAt: now.Add(-15 * time.Minute), + }, + "fresh": { + Key: "fresh", State: queueStateQueued, Repository: "owner/fresh", + ScaleSetName: "nddev-linux-standard", Priority: 1, + QueueTime: now.Add(-time.Minute), FirstQueuedAt: now.Add(-time.Minute), + }, + }, + Repositories: map[string]queueRepositoryState{}, + } + candidates := eligibleQueueCandidates(&journal, config, map[string]int{}, now) + if len(candidates) != 2 || candidates[0].Key != "moved" { + t.Fatalf("FIFO followed rewritten QueueTime: %#v", candidates) + } +} + func TestAuthoritativeReconciliationReleasesOneExactIntentIdempotently(t *testing.T) { now := time.Now().UTC() coordinator := testQueueCoordinator(t, &now, nil)