From 2ce03cd8ca00e6f1f08a4d32ca08e291f5bc1f79 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 18:48:55 +0500 Subject: [PATCH 1/2] fix(queue): retain terminal job tombstones --- config/example-runner-1.yaml | 4 +- config/example-runner-2.yaml | 4 +- config/example-runner-3.yaml | 4 +- config/example-runner-4.yaml | 4 +- config/example-services.yaml | 4 +- config/garm-derivative.yaml | 14 +++--- config/provider-derivative.yaml | 4 +- internal/queueintent/absent_journal_test.go | 2 +- internal/queueintent/queueintent.go | 18 ++++++-- internal/queueintent/queueintent_test.go | 2 + internal/queueintent/recovery_linux_test.go | 5 +- internal/queueintent/schema_evolution_test.go | 5 +- internal/queueintent/tenant_scope_test.go | 4 +- scripts/build-garm-nddev.sh | 8 ++-- .../overlay/workers/scaleset/queue_intent.go | 46 +++++++++++++++++-- .../workers/scaleset/queue_intent_test.go | 29 ++++++++++++ 16 files changed, 120 insertions(+), 37 deletions(-) diff --git a/config/example-runner-1.yaml b/config/example-runner-1.yaml index 4b9976a..88251aa 100644 --- a/config/example-runner-1.yaml +++ b/config/example-runner-1.yaml @@ -6,10 +6,10 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.81 + manager_version: v0.2.1-nddev.82 scheduling_mode: scale-set provider: incus - provider_version: v0.1.5-nddev.78 + provider_version: v0.1.5-nddev.79 provider_interface: v0.1.0 worker_kind: incus-container runner: actions/runner diff --git a/config/example-runner-2.yaml b/config/example-runner-2.yaml index 5a8f390..c326256 100644 --- a/config/example-runner-2.yaml +++ b/config/example-runner-2.yaml @@ -6,10 +6,10 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.81 + manager_version: v0.2.1-nddev.82 scheduling_mode: scale-set provider: incus - provider_version: v0.1.5-nddev.78 + provider_version: v0.1.5-nddev.79 provider_interface: v0.1.0 worker_kind: incus-container runner: actions/runner diff --git a/config/example-runner-3.yaml b/config/example-runner-3.yaml index ae0971a..32f3f80 100644 --- a/config/example-runner-3.yaml +++ b/config/example-runner-3.yaml @@ -6,10 +6,10 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.81 + manager_version: v0.2.1-nddev.82 scheduling_mode: scale-set provider: incus - provider_version: v0.1.5-nddev.78 + provider_version: v0.1.5-nddev.79 provider_interface: v0.1.0 worker_kind: incus-container runner: actions/runner diff --git a/config/example-runner-4.yaml b/config/example-runner-4.yaml index cab611e..1e2b942 100644 --- a/config/example-runner-4.yaml +++ b/config/example-runner-4.yaml @@ -6,10 +6,10 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.81 + manager_version: v0.2.1-nddev.82 scheduling_mode: scale-set provider: incus - provider_version: v0.1.5-nddev.78 + provider_version: v0.1.5-nddev.79 provider_interface: v0.1.0 worker_kind: incus-container runner: actions/runner diff --git a/config/example-services.yaml b/config/example-services.yaml index 0019f43..917e8f0 100644 --- a/config/example-services.yaml +++ b/config/example-services.yaml @@ -24,10 +24,10 @@ platform: control_plane: manager: garm - manager_version: v0.2.1-nddev.81 + manager_version: v0.2.1-nddev.82 scheduling_mode: scale-set provider: incus - provider_version: v0.1.5-nddev.78 + provider_version: v0.1.5-nddev.79 provider_interface: v0.1.0 worker_kind: incus-container runner: actions/runner diff --git a/config/garm-derivative.yaml b/config/garm-derivative.yaml index e2bb171..12f2dd4 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.81 +derivative_version: v0.2.1-nddev.82 upstream: repository: https://github.com/cloudbase/garm release: v0.2.1 @@ -90,11 +90,11 @@ patches: purpose: Bound every upstream Linux and Gitea bootstrap curl path to two retries after the initial attempt so GARM and the provider enforce the same three-total-attempt contract. overlays: - path: third_party/garm/overlay/workers/scaleset/queue_intent.go - sha256: 0793ef4a19052849f265dfe40a06a8f3330bef053ea2fff84e080b269a85f7b7 - purpose: Fsync-backed central queue admission where disabled scale sets cannot record or acquire work, sparse JobAssigned owns a bounded provisional token without refreshing its TTL, JobAvailable or the authoritative queued DB row binds repository identity, authoritative JobStarted rehydrates missing running ownership, and state_entered_at remains stable across same-phase lease refreshes. + sha256: 180e60554d264358d5a71c3b16b90d897a82cdb51a7b3d70b587029f97e5e3fe + purpose: Fsync-backed central queue admission where bounded terminal tombstones prevent delayed JobAssigned redelivery from resurrecting completed work, while provisional capacity, repository binding, running rehydration and stable phase-entry semantics remain intact. - path: third_party/garm/overlay/workers/scaleset/queue_intent_test.go - sha256: 02449e9feaf09de10f225323053b1efbc75e810a0f7bba5113b9b27fb1856c65 - purpose: Deterministic disabled-scale-set isolation, provisional bootstrap, authoritative repository enrichment and rebinding refusal, one-time legacy horizon migration, non-refreshing assigned expiry, orphan-start rehydration, phase-entry and expiry, priority, fairness, width, idempotency, acknowledgement, crash expiry and concurrent-selection tests. + sha256: 2f7d98f63033d73d972a9e91ca7cca0934e09abbc62fa971b6ed99e328f531b9 + purpose: Deterministic terminal-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. - path: third_party/garm/overlay/workers/provider/nddev_create_retry.go sha256: 611d67a74ff626f04cc0240409a22848525a06c2ae4dc7180c4b75001fb4b599 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 exact intent so one terminal or deferred job cannot block newer work in the same scale set. @@ -115,9 +115,9 @@ build: - sqlite_omit_load_extension reproducible_rebuilds: 2 maximum_required_glibc: "2.34" - binary_sha256: 118b1ee61acb7e2737f8e2c8c907d6498881d710251a9431a8878358352d2419 + binary_sha256: 18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741 runtime_contract: - queue_intent_schema_version: 4 + queue_intent_schema_version: 5 event_driven_scale_set_wake: true event_driven_instance_wake: true startup_states_protected_from_scale_down: true diff --git a/config/provider-derivative.yaml b/config/provider-derivative.yaml index c6ee5e0..7e86386 100644 --- a/config/provider-derivative.yaml +++ b/config/provider-derivative.yaml @@ -16,7 +16,7 @@ artifact: garm-provider-incus # state all move together, because all three derive from here. A provider change # that does not bump it ships under the previous version, which is exactly how # runner-1 and runner-2 diverged. -derivative_version: v0.1.5-nddev.78 +derivative_version: v0.1.5-nddev.79 # The external-provider protocol GARM speaks to this binary. It moves on its own # schedule -- a provider release does not imply an interface release -- so it is @@ -34,7 +34,7 @@ runtime: # Incus API the fleet can speak, so it belongs to the release identity rather # than to the code that happens to import it. incus_sdk_version: v7.3.0 - queue_intent_schema_version: 4 + queue_intent_schema_version: 5 build: source_commit: f8a91e798f58ca71bcf0204854d387a228262a5c diff --git a/internal/queueintent/absent_journal_test.go b/internal/queueintent/absent_journal_test.go index 4e6e651..75856ab 100644 --- a/internal/queueintent/absent_journal_test.go +++ b/internal/queueintent/absent_journal_test.go @@ -53,7 +53,7 @@ func TestPresentButUnusableJournalStillFails(t *testing.T) { }{ {name: "group readable", contents: `{"schema_version":1,"intents":{},"repositories":{}}`, mode: 0o640, wantError: "private regular file"}, {name: "malformed", contents: `{`, mode: 0o600, wantError: "decode queue-intent journal"}, - {name: "wrong schema", contents: `{"schema_version":5,"intents":{},"repositories":{}}`, mode: 0o600, wantError: "schema_version"}, + {name: "wrong schema", contents: `{"schema_version":99,"intents":{},"repositories":{}}`, mode: 0o600, wantError: "schema_version"}, {name: "null maps", contents: `{"schema_version":1,"intents":null,"repositories":null}`, mode: 0o600, wantError: "must not be null"}, } { t.Run(testCase.name, func(t *testing.T) { diff --git a/internal/queueintent/queueintent.go b/internal/queueintent/queueintent.go index 97fcca4..82a9623 100644 --- a/internal/queueintent/queueintent.go +++ b/internal/queueintent/queueintent.go @@ -21,8 +21,8 @@ import ( const ( LegacySchemaVersion = 1 - PreviousSchemaVersion = 3 - SchemaVersion = 4 + PreviousSchemaVersion = 4 + SchemaVersion = 5 maxJournalBytes = 4 * 1024 * 1024 ) @@ -84,6 +84,7 @@ type Journal struct { UpdatedAt time.Time `json:"updated_at"` Intents map[string]Intent `json:"intents"` Repositories map[string]RepositoryState `json:"repositories"` + TerminalJobs map[string]time.Time `json:"terminal_jobs"` } type Snapshot struct { @@ -235,6 +236,7 @@ func readJournal(path string) (Journal, error) { SchemaVersion: SchemaVersion, Intents: map[string]Intent{}, Repositories: map[string]RepositoryState{}, + TerminalJobs: map[string]time.Time{}, }, nil } return Journal{}, fmt.Errorf("open queue-intent journal: %w", err) @@ -270,12 +272,17 @@ func readJournal(path string) (Journal, error) { } switch journal.SchemaVersion { case LegacySchemaVersion, 2: + journal.TerminalJobs = make(map[string]time.Time) journal.SchemaVersion = SchemaVersion for key, intent := range journal.Intents { intent.StateEnteredAt = intent.UpdatedAt journal.Intents[key] = intent } + case 3: + journal.TerminalJobs = make(map[string]time.Time) + journal.SchemaVersion = SchemaVersion case PreviousSchemaVersion: + journal.TerminalJobs = make(map[string]time.Time) journal.SchemaVersion = SchemaVersion case SchemaVersion: default: @@ -291,7 +298,7 @@ func (j Journal) Validate() error { if j.SchemaVersion != SchemaVersion { return fmt.Errorf("queue-intent journal schema_version must be %d", SchemaVersion) } - if j.Intents == nil || j.Repositories == nil { + if j.Intents == nil || j.Repositories == nil || j.TerminalJobs == nil { return fmt.Errorf("queue-intent journal maps must not be null") } for key, intent := range j.Intents { @@ -322,6 +329,11 @@ func (j Journal) Validate() error { return fmt.Errorf("repository scheduler state %q is invalid", key) } } + for jobID, expiry := range j.TerminalJobs { + if !validText(jobID) || expiry.IsZero() { + return fmt.Errorf("terminal queue job %q is invalid", jobID) + } + } return nil } diff --git a/internal/queueintent/queueintent_test.go b/internal/queueintent/queueintent_test.go index 87082f5..29b02aa 100644 --- a/internal/queueintent/queueintent_test.go +++ b/internal/queueintent/queueintent_test.go @@ -43,6 +43,7 @@ func TestReaderReturnsOnlyActiveIntentsInDeterministicOrder(t *testing.T) { "updated_at": "2026-08-09T13:50:00Z", "expires_at": "2026-08-09T13:59:59Z" } }, + "terminal_jobs": {}, "repositories": { "owner/standard": {"repository": "owner/standard", "weight": 1, "pass": 10}, "owner/release": {"repository": "owner/release", "weight": 2, "pass": 5} @@ -105,6 +106,7 @@ func TestReaderAuthorizesAssignedUUIDWithoutRunnerRequestID(t *testing.T) { "expires_at": "2026-08-10T15:20:46Z" } }, + "terminal_jobs": {}, "repositories": { "example-user/github-actions": {"repository": "example-user/github-actions", "weight": 1, "pass": 1000000} } diff --git a/internal/queueintent/recovery_linux_test.go b/internal/queueintent/recovery_linux_test.go index 4d55ace..d861eff 100644 --- a/internal/queueintent/recovery_linux_test.go +++ b/internal/queueintent/recovery_linux_test.go @@ -28,6 +28,7 @@ func TestRecoverUnboundRunningUsesExactCASAndGeneration(t *testing.T) { StateEnteredAt: updatedAt, UpdatedAt: updatedAt, ExpiresAt: updatedAt.Add(24 * time.Hour), }}, Repositories: map[string]RepositoryState{"example-org": {Repository: "example-org", Weight: 1}}, + TerminalJobs: map[string]time.Time{}, } content, err := json.Marshal(journal) if err != nil { @@ -69,7 +70,7 @@ func TestRecoverUnboundRunningRejectsChangedPrecondition(t *testing.T) { lockPath := filepath.Join(directory, "queue-intents.lock") updatedAt := time.Now().UTC().Add(-time.Hour) key := intentKey(1, "bound-job") - journal := Journal{SchemaVersion: SchemaVersion, Intents: map[string]Intent{key: { + journal := Journal{SchemaVersion: SchemaVersion, TerminalJobs: map[string]time.Time{}, Intents: map[string]Intent{key: { Key: key, ScaleSetID: 1, JobID: "bound-job", RunnerRequestID: 42, ScaleSetName: "nddev-linux-standard", Owner: "example-org", Repository: "example-org/repository", WorkflowRef: "workflow.yml@refs/heads/main", EventName: "push", QueueTime: updatedAt, @@ -96,7 +97,7 @@ func TestRecoverCanceledUnboundAcceptsOnlySparseQueuedOrAssigned(t *testing.T) { lockPath := filepath.Join(directory, "queue-intents.lock") updatedAt := time.Now().UTC().Add(-time.Hour) key := intentKey(6, "canceled-job") - journal := Journal{SchemaVersion: SchemaVersion, Intents: map[string]Intent{key: { + journal := Journal{SchemaVersion: SchemaVersion, TerminalJobs: map[string]time.Time{}, Intents: map[string]Intent{key: { Key: key, ScaleSetID: 6, JobID: "canceled-job", ScaleSetName: "nddev-linux-untrusted", Owner: "example-org", Repository: "example-org", WorkflowRef: "unavailable-before-job-available", EventName: "unavailable-before-job-available", diff --git a/internal/queueintent/schema_evolution_test.go b/internal/queueintent/schema_evolution_test.go index ef597d7..8841c92 100644 --- a/internal/queueintent/schema_evolution_test.go +++ b/internal/queueintent/schema_evolution_test.go @@ -43,7 +43,7 @@ func jsonFieldNames(t *testing.T, value any) []string { // that decision -- internal/providerjournal already carries a worked example of // a version ladder that upgrades in memory. func TestQueueIntentWireShapeIsPinnedToItsSchemaVersion(t *testing.T) { - if SchemaVersion != 4 { + if SchemaVersion != 5 { t.Fatalf("SchemaVersion = %d; update the golden field sets below with it", SchemaVersion) } for _, testCase := range []struct { @@ -55,7 +55,7 @@ func TestQueueIntentWireShapeIsPinnedToItsSchemaVersion(t *testing.T) { name: "Journal", value: Journal{}, wanted: []string{ - "schema_version", "generation", "updated_at", "intents", "repositories", + "schema_version", "generation", "updated_at", "intents", "repositories", "terminal_jobs", }, }, { @@ -119,6 +119,7 @@ func TestUnknownFieldRejectsTheWholeJournalRatherThanOneIntent(t *testing.T) { "repository": "example-org/example-actions", "weight": 1, "pass": 0, }, }, + "terminal_jobs": map[string]any{}, } raw, err := json.Marshal(journal) if err != nil { diff --git a/internal/queueintent/tenant_scope_test.go b/internal/queueintent/tenant_scope_test.go index 2f7d5a6..c3f9520 100644 --- a/internal/queueintent/tenant_scope_test.go +++ b/internal/queueintent/tenant_scope_test.go @@ -29,6 +29,7 @@ func TestActiveForScaleSetSeparatesAccountsSharingAClassName(t *testing.T) { "expires_at": "2026-08-12T14:10:00Z" } }, + "terminal_jobs": {}, "repositories": { "owner/attention": {"repository": "owner/attention", "weight": 1, "pass": 1} } @@ -56,7 +57,8 @@ func TestRepositoryForScaleSetNarrowsOrganizationIntent(t *testing.T) { "repository":"example-org/example-actions","workflow_ref":"x","event_name":"workflow_dispatch", "queue_time":"2026-01-01T00:00:00Z","state":"acquired","priority":1, "updated_at":"2026-01-01T00:00:00Z","expires_at":"2099-01-01T00:00:00Z"}}, - "repositories":{"example-org":{"repository":"example-org","weight":1,"pass":1}} + "repositories":{"example-org":{"repository":"example-org","weight":1,"pass":1}}, + "terminal_jobs":{} }`)} repository, err := reader.RepositoryForScaleSet(context.Background(), "example-org", "nddev-linux-standard") if err != nil { diff --git a/scripts/build-garm-nddev.sh b/scripts/build-garm-nddev.sh index a883286..c4718bc 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.81" +readonly derivative_version="v0.2.1-nddev.82" 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="118b1ee61acb7e2737f8e2c8c907d6498881d710251a9431a8878358352d2419" +readonly expected_binary_sha256="18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741" readonly patch_paths=( "third_party/garm/patches/0001-event-driven-reconciliation.patch" "third_party/garm/patches/0002-central-queue-admission.patch" @@ -98,8 +98,8 @@ readonly overlay_paths=( "third_party/garm/overlay/workers/provider/nddev_create_retry_test.go" ) readonly overlay_sha256s=( - "0793ef4a19052849f265dfe40a06a8f3330bef053ea2fff84e080b269a85f7b7" - "02449e9feaf09de10f225323053b1efbc75e810a0f7bba5113b9b27fb1856c65" + "180e60554d264358d5a71c3b16b90d897a82cdb51a7b3d70b587029f97e5e3fe" + "2f7d98f63033d73d972a9e91ca7cca0934e09abbc62fa971b6ed99e328f531b9" "611d67a74ff626f04cc0240409a22848525a06c2ae4dc7180c4b75001fb4b599" "9379d69b8a40295e176fdfb0587b299fabda69de9708e63672c41546fd9a3508" ) diff --git a/third_party/garm/overlay/workers/scaleset/queue_intent.go b/third_party/garm/overlay/workers/scaleset/queue_intent.go index 85022fb..5ba2cc2 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent.go @@ -22,8 +22,8 @@ import ( const ( queueAdmissionSchemaVersion = 5 queueIntentLegacySchemaVersion = 1 - queueIntentPreviousSchemaVersion = 3 - queueIntentSchemaVersion = 4 + queueIntentPreviousSchemaVersion = 4 + queueIntentSchemaVersion = 5 queueIntentMaxBytes = 4 * 1024 * 1024 queueSchedulerStride = uint64(1_000_000) queueAcquireRetryDelay = 5 * time.Second @@ -141,6 +141,7 @@ type queueIntentJournal struct { UpdatedAt time.Time `json:"updated_at"` Intents map[string]queueIntent `json:"intents"` Repositories map[string]queueRepositoryState `json:"repositories"` + TerminalJobs map[string]time.Time `json:"terminal_jobs"` } type queueIntentCoordinator struct { @@ -187,9 +188,11 @@ func NDDevRemoveQueueIntent(ctx context.Context, jobID string) (bool, error) { matched = key } if matched == "" { + markTerminalJob(journal, jobID, now.Add(time.Duration(config.ExecutionTTLSeconds)*time.Second)) return nil } delete(journal.Intents, matched) + markTerminalJob(journal, jobID, now.Add(time.Duration(config.ExecutionTTLSeconds)*time.Second)) admitQueuedToBudget(journal, config, now) removed = true return nil @@ -566,6 +569,7 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti return fmt.Errorf("completed job has invalid job ID") } completedKeys[queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID)] = struct{}{} + markTerminalJob(journal, job.JobID, now.Add(time.Duration(config.ExecutionTTLSeconds)*time.Second)) } for _, job := range started { key := queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID) @@ -651,6 +655,9 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti journal.Intents[key] = intent } for _, job := range assigned { + if expiry, terminal := journal.TerminalJobs[job.JobID]; terminal && expiry.After(now) { + continue + } intent, err := queueIntentFromLifecycle(config, scaleSet, entity, job, now, time.Duration(config.QueuedTTLSeconds)*time.Second) if err != nil { return err @@ -870,6 +877,7 @@ func (c *queueIntentCoordinator) update(config queueAdmissionConfig, mutate func now := c.nowUTC() migrateLegacyQueueIntentOwnership(&journal, config, now) cleanupExpiredQueueIntents(&journal, 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 // queue state on every serialized journal transaction so recovery never @@ -972,7 +980,7 @@ func (c queueAdmissionConfig) Validate() error { } func (j queueIntentJournal) Validate() error { - if j.SchemaVersion != queueIntentSchemaVersion || j.Intents == nil || j.Repositories == nil { + if j.SchemaVersion != queueIntentSchemaVersion || j.Intents == nil || j.Repositories == nil || j.TerminalJobs == nil { return fmt.Errorf("queue-intent journal identity or maps are invalid") } for key, intent := range j.Intents { @@ -1002,9 +1010,28 @@ func (j queueIntentJournal) Validate() error { return fmt.Errorf("queue repository state %q is invalid", key) } } + for jobID, expiry := range j.TerminalJobs { + if !validQueueText(jobID) || expiry.IsZero() { + return fmt.Errorf("terminal queue job %q is invalid", jobID) + } + } return nil } +func markTerminalJob(journal *queueIntentJournal, jobID string, expiry time.Time) { + if current, exists := journal.TerminalJobs[jobID]; !exists || !current.After(expiry) { + journal.TerminalJobs[jobID] = expiry + } +} + +func cleanupExpiredTerminalJobs(journal *queueIntentJournal, now time.Time) { + for jobID, expiry := range journal.TerminalJobs { + if !expiry.After(now) { + delete(journal.TerminalJobs, jobID) + } + } +} + func queueIntentFromJob(config queueAdmissionConfig, scaleSet params.ScaleSet, job params.ScaleSetJobMessage, now time.Time, ttl time.Duration) (queueIntent, error) { repository := job.OwnerName + "/" + job.RepositoryName if scaleSet.ScaleSetID <= 0 || scaleSet.MaxRunners < 1 || !validQueueText(scaleSet.Name) || !validQueueText(job.JobID) || job.RunnerRequestID <= 0 || !validRepository(repository) || @@ -1374,14 +1401,22 @@ func readQueueIntentJournal(path string) (queueIntentJournal, error) { // v2 adds only the JobStarted runner identity. Old active intents have no // value to synthesize; they remain explicitly uncorrelated until their // lifecycle completes, and the next writer transaction upgrades the file. + journal.TerminalJobs = make(map[string]time.Time) journal.SchemaVersion = queueIntentSchemaVersion for key, intent := range journal.Intents { intent.StateEnteredAt = intent.UpdatedAt journal.Intents[key] = intent } - case queueIntentPreviousSchemaVersion: - // v4 adds correlation fields only. Existing intents remain explicitly + case 3: + // v4 added correlation fields only. Existing intents remain explicitly // incomplete until a later JobAvailable or JobStarted binds them. + journal.TerminalJobs = make(map[string]time.Time) + journal.SchemaVersion = queueIntentSchemaVersion + case queueIntentPreviousSchemaVersion: + // v5 adds bounded terminal tombstones. No prior tombstone can be + // reconstructed, so the map starts empty and only new authoritative + // completion or absence evidence populates it. + journal.TerminalJobs = make(map[string]time.Time) journal.SchemaVersion = queueIntentSchemaVersion case queueIntentSchemaVersion: default: @@ -1477,6 +1512,7 @@ func newQueueIntentJournal() queueIntentJournal { SchemaVersion: queueIntentSchemaVersion, Intents: make(map[string]queueIntent), Repositories: make(map[string]queueRepositoryState), + TerminalJobs: make(map[string]time.Time), } } 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 8d3ecb2..d58d860 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent_test.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent_test.go @@ -50,6 +50,35 @@ func TestAdmittedCapacityIntentDisappearsOnCompletion(t *testing.T) { } } +func TestCompletedJobCannotBeResurrectedByDelayedAssignedRedelivery(t *testing.T) { + now := time.Date(2026, 8, 26, 13, 0, 0, 0, time.UTC) + coordinator := testQueueCoordinator(t, &now, nil) + scaleSet := testQueueScaleSet(11, "nddev-linux-fast") + job := testQueueJob(101, "example-owner", "example-repository", now) + entity := testQueueEntityForJob(job) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{job}, nil, nil); err != nil { + t.Fatal(err) + } + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, nil, nil, []params.ScaleSetJobMessage{job}); err != nil { + t.Fatal(err) + } + now = now.Add(time.Minute) + if _, err := coordinator.ObserveLifecycle(scaleSet, entity, []params.ScaleSetJobMessage{job}, nil, nil); err != nil { + t.Fatal(err) + } + journal, err := readQueueIntentJournal(coordinator.journalPath) + if err != nil { + t.Fatal(err) + } + key := queueIntentKey(int64(scaleSet.ScaleSetID), job.JobID) + if _, exists := journal.Intents[key]; exists { + t.Fatalf("delayed JobAssigned resurrected terminal intent: %#v", journal.Intents[key]) + } + if !journal.TerminalJobs[job.JobID].After(now) { + t.Fatalf("terminal tombstone is absent or expired: %#v", journal.TerminalJobs) + } +} + func TestAuthoritativeReconciliationReleasesOneExactIntentIdempotently(t *testing.T) { now := time.Now().UTC() coordinator := testQueueCoordinator(t, &now, nil) From f183cf8d0d52508b4d88cb5d1b2836de22807b33 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 18:49:46 +0500 Subject: [PATCH 2/2] chore(provider): bind queue schema v5 reader --- config/provider-derivative.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/provider-derivative.yaml b/config/provider-derivative.yaml index 7e86386..b43ec71 100644 --- a/config/provider-derivative.yaml +++ b/config/provider-derivative.yaml @@ -37,8 +37,8 @@ runtime: queue_intent_schema_version: 5 build: - source_commit: f8a91e798f58ca71bcf0204854d387a228262a5c - binary_sha256: 204e3e1a4a4e9b9beeb3516fb8921e58fe9b2f928ed70404a8bdd4497ee0ade2 + source_commit: 2ce03cd8ca00e6f1f08a4d32ca08e291f5bc1f79 + binary_sha256: 9a7f38b3685e3e73fb5d8bcc7a3af1e3070827204ced4de469f573d22005740c go_version: go1.26.6 cgo_enabled: false target_os: linux