From 9dcd350b7e93cc6c29fca8d7107a993872cd6cfe Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 19:34:04 +0500 Subject: [PATCH 1/2] feat(observe): expose terminal tombstone lifecycle --- 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 | 8 +++--- internal/fleetobserve/metrics.go | 2 ++ internal/fleetobserve/metrics_test.go | 2 ++ internal/fleetobserve/observe.go | 20 ++++++++------ internal/queueintent/queueintent.go | 26 +++++++++++++++---- internal/queueintent/queueintent_test.go | 22 ++++++++++++++++ scripts/build-garm-nddev.sh | 6 ++--- .../overlay/workers/scaleset/queue_intent.go | 9 +++++++ 13 files changed, 80 insertions(+), 25 deletions(-) diff --git a/config/example-runner-1.yaml b/config/example-runner-1.yaml index 88251aa..867fa9d 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.82 + manager_version: v0.2.1-nddev.83 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.79 diff --git a/config/example-runner-2.yaml b/config/example-runner-2.yaml index c326256..e86d6f5 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.82 + manager_version: v0.2.1-nddev.83 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.79 diff --git a/config/example-runner-3.yaml b/config/example-runner-3.yaml index 32f3f80..f0614fe 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.82 + manager_version: v0.2.1-nddev.83 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.79 diff --git a/config/example-runner-4.yaml b/config/example-runner-4.yaml index 1e2b942..b9e1690 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.82 + manager_version: v0.2.1-nddev.83 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.79 diff --git a/config/example-services.yaml b/config/example-services.yaml index 917e8f0..ff6610f 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.82 + manager_version: v0.2.1-nddev.83 scheduling_mode: scale-set provider: incus provider_version: v0.1.5-nddev.79 diff --git a/config/garm-derivative.yaml b/config/garm-derivative.yaml index 12f2dd4..e52c839 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.82 +derivative_version: v0.2.1-nddev.83 upstream: repository: https://github.com/cloudbase/garm release: v0.2.1 @@ -90,8 +90,8 @@ 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: 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. + sha256: 31486d264f0a2b8dc63bb902ee49163c32e4c8774a6ad5b6d29efad92a228a03 + purpose: Fsync-backed central queue admission where bounded terminal tombstones prevent delayed JobAssigned redelivery from resurrecting completed work and emit one structured suppression event, 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: 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. @@ -115,7 +115,7 @@ build: - sqlite_omit_load_extension reproducible_rebuilds: 2 maximum_required_glibc: "2.34" - binary_sha256: 18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741 + binary_sha256: f9594a5ff67643a3b1f9e3357db6a96b45e1ec83797ff9ad0189302a264bb238 runtime_contract: queue_intent_schema_version: 5 event_driven_scale_set_wake: true diff --git a/internal/fleetobserve/metrics.go b/internal/fleetobserve/metrics.go index 87f3d20..cdb69fa 100644 --- a/internal/fleetobserve/metrics.go +++ b/internal/fleetobserve/metrics.go @@ -139,6 +139,8 @@ func RenderPrometheus(snapshot Snapshot, now time.Time, maxStaleness time.Durati gauge(&output, "gha_fleet_queue_intents_stored", "Queue intents stored in the current journal generation.", float64(snapshot.Queue.Stored)) gauge(&output, "gha_fleet_queue_intents_active", "Unexpired queue intents owned by central admission.", float64(snapshot.Queue.Active)) gauge(&output, "gha_fleet_queue_intents_expired", "Expired queue intents awaiting the next writer cleanup transaction.", float64(snapshot.Queue.Expired)) + gauge(&output, "gha_fleet_queue_terminal_tombstones", "Completed job identities retained to suppress delayed lifecycle redelivery.", float64(snapshot.Queue.TerminalJobs)) + gauge(&output, "gha_fleet_queue_terminal_tombstone_next_expiry_seconds", "Seconds until the next bounded terminal job tombstone expires.", float64(snapshot.Queue.TerminalNextExpirySeconds)) gauge(&output, "gha_fleet_queue_intents_in_flight", "Acquiring, acquired, assigned or running queue intents.", float64(snapshot.Queue.InFlight)) gauge(&output, "gha_fleet_queue_oldest_age_seconds", "Age of the oldest active GitHub queue intent.", float64(snapshot.Queue.OldestQueueAgeSeconds)) gauge(&output, "gha_fleet_queue_uncovered_running", "Running queue intents without a durable created or warm-claimed execution lease.", float64(snapshot.Queue.UncoveredRunning)) diff --git a/internal/fleetobserve/metrics_test.go b/internal/fleetobserve/metrics_test.go index db49a3d..f98bc6d 100644 --- a/internal/fleetobserve/metrics_test.go +++ b/internal/fleetobserve/metrics_test.go @@ -45,6 +45,8 @@ func TestRenderPrometheusIsDeterministicAndBounded(t *testing.T) { "gha_fleet_provider_warm_preemptions_total 0\n", "gha_fleet_queue_journal_generation 1\n", "gha_fleet_queue_intents_active 0\n", + "gha_fleet_queue_terminal_tombstones 0\n", + "gha_fleet_queue_terminal_tombstone_next_expiry_seconds 0\n", "gha_fleet_queue_intents_in_flight 0\n", "gha_fleet_queue_uncovered_running 0\n", "gha_fleet_queue_missing_runner_request_id 0\n", diff --git a/internal/fleetobserve/observe.go b/internal/fleetobserve/observe.go index b6292e6..28053b4 100644 --- a/internal/fleetobserve/observe.go +++ b/internal/fleetobserve/observe.go @@ -199,6 +199,8 @@ type QueueSummary struct { Stored int `json:"stored"` Active int `json:"active"` Expired int `json:"expired"` + TerminalJobs int `json:"terminal_jobs"` + TerminalNextExpirySeconds int64 `json:"terminal_next_expiry_seconds"` InFlight int `json:"in_flight"` OldestQueueAgeSeconds int64 `json:"oldest_queue_age_seconds"` UncoveredRunning int `json:"uncovered_running"` @@ -560,14 +562,16 @@ func summarizeQueue(snapshot queueintent.Snapshot, platform config.Config, now t knownScaleSets[pool.ScaleSetName] = struct{}{} } summary := QueueSummary{ - Generation: snapshot.Generation, - Stored: snapshot.Stored, - Active: len(snapshot.Active), - Expired: snapshot.Expired, - ByState: make(map[string]int), - OldestStateAgeSeconds: make(map[string]int64), - ByPriority: make(map[int]int), - ByScaleSet: make(map[string]int), + Generation: snapshot.Generation, + Stored: snapshot.Stored, + Active: len(snapshot.Active), + Expired: snapshot.Expired, + TerminalJobs: snapshot.TerminalJobs, + TerminalNextExpirySeconds: snapshot.TerminalNextExpirySeconds, + ByState: make(map[string]int), + OldestStateAgeSeconds: make(map[string]int64), + ByPriority: make(map[int]int), + ByScaleSet: make(map[string]int), } for scaleSet := range knownScaleSets { summary.ByScaleSet[scaleSet] = 0 diff --git a/internal/queueintent/queueintent.go b/internal/queueintent/queueintent.go index 82a9623..61ff3c9 100644 --- a/internal/queueintent/queueintent.go +++ b/internal/queueintent/queueintent.go @@ -88,10 +88,12 @@ type Journal struct { } type Snapshot struct { - Generation uint64 - Stored int - Expired int - Active []Intent + Generation uint64 + Stored int + Expired int + TerminalJobs int + TerminalNextExpirySeconds int64 + Active []Intent } func (r Reader) HasActive(ctx context.Context) (bool, error) { @@ -118,6 +120,16 @@ func (r Reader) ReadActive(ctx context.Context) (Snapshot, error) { now := r.now() active := make([]Intent, 0, len(journal.Intents)) expired := 0 + terminalNextExpirySeconds := int64(0) + for _, expiry := range journal.TerminalJobs { + remaining := int64(expiry.Sub(now).Seconds()) + if remaining < 0 { + remaining = 0 + } + if terminalNextExpirySeconds == 0 || remaining < terminalNextExpirySeconds { + terminalNextExpirySeconds = remaining + } + } for _, intent := range journal.Intents { if intent.ExpiresAt.After(now) { active = append(active, intent) @@ -134,7 +146,11 @@ func (r Reader) ReadActive(ctx context.Context) (Snapshot, error) { } return active[left].Key < active[right].Key }) - return Snapshot{Generation: journal.Generation, Stored: len(journal.Intents), Expired: expired, Active: active}, nil + return Snapshot{ + Generation: journal.Generation, Stored: len(journal.Intents), Expired: expired, + TerminalJobs: len(journal.TerminalJobs), TerminalNextExpirySeconds: terminalNextExpirySeconds, + Active: active, + }, nil } // ActiveForScaleSet reports whether the named scale set of the named tenant diff --git a/internal/queueintent/queueintent_test.go b/internal/queueintent/queueintent_test.go index 29b02aa..ca2235a 100644 --- a/internal/queueintent/queueintent_test.go +++ b/internal/queueintent/queueintent_test.go @@ -73,6 +73,28 @@ func TestReaderReturnsOnlyActiveIntentsInDeterministicOrder(t *testing.T) { } } +func TestReaderReportsBoundedTerminalTombstones(t *testing.T) { + now := time.Date(2026, 8, 26, 14, 0, 0, 0, time.UTC) + path := writeFixture(t, `{ + "schema_version": 5, + "generation": 9, + "updated_at": "2026-08-26T13:59:00Z", + "intents": {}, + "repositories": {}, + "terminal_jobs": { + "11111111-1111-4111-8111-111111111111": "2026-08-27T14:00:00Z", + "22222222-2222-4222-8222-222222222222": "2026-08-26T15:00:00Z" + } +}`) + snapshot, err := (Reader{Path: path, Now: func() time.Time { return now }}).ReadActive(context.Background()) + if err != nil { + t.Fatal(err) + } + if snapshot.TerminalJobs != 2 || snapshot.TerminalNextExpirySeconds != 3600 { + t.Fatalf("terminal snapshot = %#v", snapshot) + } +} + func TestReaderRejectsFinalComponentSymlink(t *testing.T) { directory := t.TempDir() target := filepath.Join(directory, "real-journal.json") diff --git a/scripts/build-garm-nddev.sh b/scripts/build-garm-nddev.sh index c4718bc..1bd13b6 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.82" +readonly derivative_version="v0.2.1-nddev.83" 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="18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741" +readonly expected_binary_sha256="f9594a5ff67643a3b1f9e3357db6a96b45e1ec83797ff9ad0189302a264bb238" readonly patch_paths=( "third_party/garm/patches/0001-event-driven-reconciliation.patch" "third_party/garm/patches/0002-central-queue-admission.patch" @@ -98,7 +98,7 @@ readonly overlay_paths=( "third_party/garm/overlay/workers/provider/nddev_create_retry_test.go" ) readonly overlay_sha256s=( - "180e60554d264358d5a71c3b16b90d897a82cdb51a7b3d70b587029f97e5e3fe" + "31486d264f0a2b8dc63bb902ee49163c32e4c8774a6ad5b6d29efad92a228a03" "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 5ba2cc2..d77a7b9 100644 --- a/third_party/garm/overlay/workers/scaleset/queue_intent.go +++ b/third_party/garm/overlay/workers/scaleset/queue_intent.go @@ -558,10 +558,12 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti var orphanedStarts []string var uncorrelatedStarts []string var reservationTransfers [][2]string + var suppressedTerminalAssignments []string err = c.update(config, func(journal *queueIntentJournal, now time.Time) error { orphanedStarts = orphanedStarts[:0] uncorrelatedStarts = uncorrelatedStarts[:0] reservationTransfers = reservationTransfers[:0] + suppressedTerminalAssignments = suppressedTerminalAssignments[:0] completedKeys := make(map[string]struct{}, len(completed)) startedKeys := make(map[string]struct{}, len(started)) for _, job := range completed { @@ -656,6 +658,7 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti } for _, job := range assigned { if expiry, terminal := journal.TerminalJobs[job.JobID]; terminal && expiry.After(now) { + suppressedTerminalAssignments = append(suppressedTerminalAssignments, job.JobID) continue } intent, err := queueIntentFromLifecycle(config, scaleSet, entity, job, now, time.Duration(config.QueuedTTLSeconds)*time.Second) @@ -712,6 +715,12 @@ func (c *queueIntentCoordinator) ObserveLifecycle(scaleSet params.ScaleSet, enti "key", key, "scale_set", scaleSet.Name, "scale_set_id", scaleSet.ScaleSetID, ) } + for _, jobID := range suppressedTerminalAssignments { + slog.Info( + "suppressed delayed terminal job assignment", + "job_id", jobID, "scale_set", scaleSet.Name, "scale_set_id", scaleSet.ScaleSetID, + ) + } // A durable assigned/queued intent is the crash-safe ownership transfer. // Retaining JobAssigned at GitHub only causes head-of-line redelivery and // prevents later JobCompleted messages from reaching this listener. From 4bb93be0c3eebcf2059c3c1efea82e490b63f894 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Wed, 26 Aug 2026 19:37:14 +0500 Subject: [PATCH 2/2] fix(recovery): resume interrupted attempts --- internal/schedulerrecovery/controller.go | 23 ++++++++++++++++++ internal/schedulerrecovery/controller_test.go | 21 ++++++++++++++++ internal/schedulerrecovery/recover.go | 22 +++++++++++++++++ internal/schedulerrecovery/recover_test.go | 24 +++++++++++++++---- internal/schedulerrecovery/store.go | 15 ++++++++++++ internal/schedulerrecovery/store_test.go | 3 +++ 6 files changed, 104 insertions(+), 4 deletions(-) diff --git a/internal/schedulerrecovery/controller.go b/internal/schedulerrecovery/controller.go index bc66fc0..fb55f9c 100644 --- a/internal/schedulerrecovery/controller.go +++ b/internal/schedulerrecovery/controller.go @@ -41,6 +41,29 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error) if controller.Observer == nil || controller.Heartbeat == nil || controller.Attempts == nil || controller.Executor == nil || controller.Events == nil || controller.Now == nil { return Decision{}, Result{}, fmt.Errorf("scheduler recovery controller is incomplete") } + active, err := controller.Attempts.Active(ctx) + if err != nil { + return Decision{}, Result{}, fmt.Errorf("read active recovery attempt: %w", err) + } + if len(active) > 1 { + return Decision{}, Result{}, fmt.Errorf("multiple active recovery attempts require deterministic repair") + } + if len(active) == 1 { + attempt := active[0] + if err := controller.emit(ctx, Event{At: controller.Now().UTC(), State: "recovering", Reason: "resume-interrupted-recovery", AttemptID: attempt.ID, Stuck: attempt.Stuck}); err != nil { + return Decision{}, Result{}, err + } + result, recoveryErr := resumeAcquired(ctx, attempt, controller.Attempts, controller.Executor, controller.Now) + terminal := Event{At: controller.Now().UTC(), State: "recovered", Reason: "resume-interrupted-recovery", AttemptID: attempt.ID, Stuck: result.Remaining} + if recoveryErr != nil { + terminal.State = "failed" + terminal.Error = recoveryErr.Error() + } + if err := controller.emit(ctx, terminal); err != nil { + return Decision{}, result, err + } + return Decision{Reason: "resume-interrupted-recovery", Stuck: attempt.Stuck}, result, recoveryErr + } observation, err := controller.Observer.Observe(ctx) if err != nil { return Decision{}, Result{}, fmt.Errorf("observe scheduler: %w", err) diff --git a/internal/schedulerrecovery/controller_test.go b/internal/schedulerrecovery/controller_test.go index b2b752c..a792ae6 100644 --- a/internal/schedulerrecovery/controller_test.go +++ b/internal/schedulerrecovery/controller_test.go @@ -73,3 +73,24 @@ func TestControllerReportsHealthyCurrentHeartbeat(t *testing.T) { require.Equal(t, "dispatcher-heartbeat-current", decision.Reason) require.Equal(t, "healthy", events.events[0].State) } + +func TestControllerFinishesInterruptedRecoveryAfterRestartProgressed(t *testing.T) { + t.Parallel() + at := time.Date(2026, 8, 26, 14, 13, 56, 0, time.UTC) + attempt := NewAttempt(at, []string{"retry-1"}) + store := &memoryAttempts{active: map[string]Attempt{attempt.ID: attempt}} + executor := &faultExecutor{progressed: []string{"retry-1"}} + events := &eventRecorder{} + controller := Controller{ + Policy: Policy{MinimumStuckAge: time.Minute, MinimumUptime: time.Minute, Cooldown: time.Minute, HeartbeatStale: time.Minute}, + Observer: staticObserver{}, Heartbeat: staticHeartbeat{}, Attempts: store, Executor: executor, + Events: events, Now: func() time.Time { return at.Add(time.Minute) }, + } + decision, result, err := controller.Tick(context.Background()) + require.NoError(t, err) + require.Equal(t, "resume-interrupted-recovery", decision.Reason) + require.True(t, result.Recovered) + require.Zero(t, executor.restarts) + require.Empty(t, store.active) + require.Equal(t, []string{"recovering", "recovered"}, []string{events.events[0].State, events.events[1].State}) +} diff --git a/internal/schedulerrecovery/recover.go b/internal/schedulerrecovery/recover.go index 6e2b368..7aca415 100644 --- a/internal/schedulerrecovery/recover.go +++ b/internal/schedulerrecovery/recover.go @@ -28,10 +28,32 @@ type Result struct { } type AttemptStore interface { + Active(context.Context) ([]Attempt, error) Begin(context.Context, Attempt) (bool, error) Finish(context.Context, Result) error } +func resumeAcquired(ctx context.Context, attempt Attempt, store AttemptStore, executor Executor, now func() time.Time) (Result, error) { + progressed, remaining, progressErr := executor.AwaitProgress(ctx, attempt) + if progressErr == nil && len(remaining) == 0 { + result := Result{ + AttemptID: attempt.ID, Progressed: slices.Clone(progressed), Recovered: true, + FinishedAt: now().UTC(), + } + if err := store.Finish(ctx, result); err != nil { + return result, fmt.Errorf("finish resumed recovery attempt: %w", err) + } + return result, nil + } + // The previous process may have died before or during the manager restart. + // Re-running the checkpoint-first sequence is idempotent; restricting it to + // the still-stuck identities avoids replaying work already proven progressed. + if len(remaining) > 0 { + attempt.Stuck = slices.Clone(remaining) + } + return recoverAcquired(ctx, attempt, store, executor, now) +} + type Executor interface { Checkpoint(context.Context, Attempt) (string, error) RestartDispatcher(context.Context, Attempt) error diff --git a/internal/schedulerrecovery/recover_test.go b/internal/schedulerrecovery/recover_test.go index c85c7c2..b343132 100644 --- a/internal/schedulerrecovery/recover_test.go +++ b/internal/schedulerrecovery/recover_test.go @@ -12,7 +12,7 @@ import ( type memoryAttempts struct { mu sync.Mutex - active map[string]bool + active map[string]Attempt finished []Result } @@ -20,12 +20,17 @@ func (store *memoryAttempts) Begin(_ context.Context, attempt Attempt) (bool, er store.mu.Lock() defer store.mu.Unlock() if store.active == nil { - store.active = map[string]bool{} + store.active = map[string]Attempt{} } - if store.active[attempt.ID] { + if _, exists := store.active[attempt.ID]; exists { return false, nil } - store.active[attempt.ID] = true + for _, result := range store.finished { + if result.AttemptID == attempt.ID { + return false, nil + } + } + store.active[attempt.ID] = attempt return true, nil } @@ -33,9 +38,20 @@ func (store *memoryAttempts) Finish(_ context.Context, result Result) error { store.mu.Lock() defer store.mu.Unlock() store.finished = append(store.finished, result) + delete(store.active, result.AttemptID) return nil } +func (store *memoryAttempts) Active(_ context.Context) ([]Attempt, error) { + store.mu.Lock() + defer store.mu.Unlock() + attempts := make([]Attempt, 0, len(store.active)) + for _, attempt := range store.active { + attempts = append(attempts, attempt) + } + return attempts, nil +} + type faultExecutor struct { mu sync.Mutex checkpoints int diff --git a/internal/schedulerrecovery/store.go b/internal/schedulerrecovery/store.go index 8aedf40..f783b00 100644 --- a/internal/schedulerrecovery/store.go +++ b/internal/schedulerrecovery/store.go @@ -74,6 +74,21 @@ func (store FileStore) Begin(_ context.Context, attempt Attempt) (bool, error) { return acquired, err } +func (store FileStore) Active(_ context.Context) ([]Attempt, error) { + var attempts []Attempt + err := store.locked(func(state *fileState) error { + attempts = make([]Attempt, 0, len(state.Active)) + for _, attempt := range state.Active { + attempts = append(attempts, attempt) + } + slices.SortFunc(attempts, func(left, right Attempt) int { + return left.StartedAt.Compare(right.StartedAt) + }) + return nil + }) + return attempts, err +} + func (store FileStore) Finish(_ context.Context, result Result) error { return store.locked(func(state *fileState) error { if _, exists := state.Active[result.AttemptID]; !exists { diff --git a/internal/schedulerrecovery/store_test.go b/internal/schedulerrecovery/store_test.go index 4d105e6..a3712fe 100644 --- a/internal/schedulerrecovery/store_test.go +++ b/internal/schedulerrecovery/store_test.go @@ -20,6 +20,9 @@ func TestFileStorePersistsAttemptAndSuppressesReplay(t *testing.T) { require.True(t, acquired) reopened := FileStore{Path: store.Path, LockPath: store.LockPath} + active, err := reopened.Active(context.Background()) + require.NoError(t, err) + require.Equal(t, []Attempt{attempt}, active) acquired, err = reopened.Begin(context.Background(), attempt) require.NoError(t, err) require.False(t, acquired)