diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 43e45f2..3a2df77 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -600,6 +600,19 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP "rate_aware", retryCfg.RateHint != nil) } + // Orphaned-runner sweep: destroys dispatched runners that were never + // assigned a job within the grace window. GitHub schedules JIT + // runners onto any queued job with matching labels, so runner + // teardown is keyed to observed assignments; the sweep is the + // cleanup path for runners whose intended job ran elsewhere. + orphanCfg := scheduler.OrphanSweepConfig{ + Enabled: cfg.Runner.OrphanSweepEnabled(), + Grace: cfg.Runner.OrphanSweepGrace(), + } + if orphanCfg.Enabled { + log.Info("orphaned-runner sweep enabled", "grace", orphanCfg.Grace) + } + // Start scheduler (ties CI provider jobs to container lifecycle) sched := scheduler.New(scheduler.Config{ Runtime: rt, @@ -626,6 +639,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP NativeMacUser: cfg.Runner.MacOS.User, RunnerDir: rm.Dir(), Retry: retryCfg, + OrphanSweep: orphanCfg, Log: log, }) diff --git a/config.example.toml b/config.example.toml index 701edd2..f56fa91 100644 --- a/config.example.toml +++ b/config.example.toml @@ -84,26 +84,21 @@ shutdown_timeout = "5m" # schedule = ["30s", "1m", "2m", "5m", "10m"] # jitter = 0.2 -# Claim retry queue. +# Orphaned-runner sweep. # -# GitHub does not re-deliver workflow_job webhooks. When ephemerd's -# initial claim/provision attempt fails for a transient reason -# (rate-limit exhausted, transient 5xx, network blip), the job would -# otherwise be lost. The retry queue schedules re-attempts on a -# jittered exponential backoff ladder and gives up after max_age. -# -# When the last GitHub API response says remaining==0 with a known -# reset time, the next attempt is snapped to just after reset instead -# of falling through the ladder, so we don't burn attempts against a -# provably-exhausted budget. +# ephemerd registers one just-in-time runner per queued job, but GitHub +# assigns a registered runner to ANY queued job with matching labels. +# Runner teardown therefore follows the assignment GitHub reports in +# workflow_job webhooks (runner_name), not the job the runner was +# dispatched for. A runner whose intended job was picked up elsewhere +# and that never received a job of its own is destroyed and +# deregistered after the grace window. # -# Enabled by default. Set enabled = false to restore the pre-existing -# "log and drop" behavior. -# [runner.claim_retry] -# enabled = true -# max_age = "90m" -# schedule = ["30s", "1m", "2m", "5m", "10m"] -# jitter = 0.2 +# Only active in webhook mode (polling mode has no in_progress events +# to observe assignments with). Enabled by default. +# [runner.orphan_sweep] +# enabled = true +# grace = "10m" [vm.linux] # Enable a Linux VM for running Linux jobs on Windows/macOS hosts. diff --git a/pkg/config/config.go b/pkg/config/config.go index f4ced68..9fff5b5 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -469,6 +469,32 @@ type RunnerConfig struct { // hit an API blip at claim time would be lost until human // intervention. ClaimRetry ClaimRetryToml `toml:"claim_retry"` + + // OrphanSweep controls teardown of runners that were dispatched for + // a job but never observed picking one up. GitHub assigns JIT + // runners to ANY queued job with matching labels, so runner + // lifecycle is keyed to the observed assignment; a runner whose + // intended job went elsewhere and that never got a job of its own is + // destroyed after the grace window. + OrphanSweep OrphanSweepToml `toml:"orphan_sweep"` +} + +// OrphanSweepToml configures the scheduler's orphaned-runner sweep. +// +// Enabled defaults to true when the [runner.orphan_sweep] table is +// omitted. The sweep only acts in webhook mode (in polling mode there +// are no in_progress events, so ephemerd cannot tell an orphaned runner +// from a busy one) and only for providers that report runner +// assignments (GitHub). +type OrphanSweepToml struct { + // Enabled toggles the sweep. Nil = default true; operators disable + // by explicitly setting enabled = false. + Enabled *bool `toml:"enabled"` + + // Grace is how long a dispatched runner may sit without being + // assigned a job before it is destroyed and deregistered. Accepts + // Go duration strings ("10m", "1h"). Default 10m. + Grace string `toml:"grace"` } // ClaimRetryToml configures the scheduler's claim-retry queue. @@ -675,6 +701,29 @@ func (r *RunnerConfig) ClaimRetrySchedule() []time.Duration { return out } +// OrphanSweepEnabled reports whether the orphaned-runner sweep should +// run. Defaults to true when the table is omitted or Enabled is nil; +// operators opt out with `enabled = false`. +func (r *RunnerConfig) OrphanSweepEnabled() bool { + if r == nil || r.OrphanSweep.Enabled == nil { + return true + } + return *r.OrphanSweep.Enabled +} + +// OrphanSweepGrace returns how long a dispatched runner may remain +// unassigned before the sweep destroys it. Defaults to 10 minutes when +// unset or unparseable. +func (r *RunnerConfig) OrphanSweepGrace() time.Duration { + if r == nil || r.OrphanSweep.Grace == "" { + return 10 * time.Minute + } + if d, err := time.ParseDuration(r.OrphanSweep.Grace); err == nil && d > 0 { + return d + } + return 10 * time.Minute +} + // ClaimRetryJitter returns the retry backoff jitter fraction. Defaults // to 0.2 (+/-20%). Values outside [0,1] are clamped to the default. func (r *RunnerConfig) ClaimRetryJitter() float64 { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index f92c8fc..dcde014 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -2078,3 +2078,45 @@ func TestMacOSRunnerConfig_ResolvedMaxNative(t *testing.T) { }) } } + +func TestOrphanSweepEnabled(t *testing.T) { + boolPtr := func(b bool) *bool { return &b } + tests := []struct { + name string + cfg *RunnerConfig + want bool + }{ + {"nil config defaults to true", nil, true}, + {"omitted table defaults to true", &RunnerConfig{}, true}, + {"explicit true", &RunnerConfig{OrphanSweep: OrphanSweepToml{Enabled: boolPtr(true)}}, true}, + {"explicit false disables", &RunnerConfig{OrphanSweep: OrphanSweepToml{Enabled: boolPtr(false)}}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.cfg.OrphanSweepEnabled(); got != tt.want { + t.Errorf("OrphanSweepEnabled() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestOrphanSweepGrace(t *testing.T) { + tests := []struct { + name string + cfg *RunnerConfig + want time.Duration + }{ + {"nil config defaults to 10m", nil, 10 * time.Minute}, + {"empty defaults to 10m", &RunnerConfig{}, 10 * time.Minute}, + {"custom duration", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "25m"}}, 25 * time.Minute}, + {"unparseable defaults to 10m", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "soon"}}, 10 * time.Minute}, + {"non-positive defaults to 10m", &RunnerConfig{OrphanSweep: OrphanSweepToml{Grace: "-5m"}}, 10 * time.Minute}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.cfg.OrphanSweepGrace(); got != tt.want { + t.Errorf("OrphanSweepGrace() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/pkg/github/client.go b/pkg/github/client.go index 44103fb..132b4ee 100644 --- a/pkg/github/client.go +++ b/pkg/github/client.go @@ -505,6 +505,7 @@ func (c *Client) WebhookHandler(secret string) (http.Handler, <-chan JobEvent) { "repo", repoName, "job_id", payload.WorkflowJob.GetID(), "labels", payload.WorkflowJob.Labels, + "runner_name", payload.WorkflowJob.GetRunnerName(), ) events <- JobEvent{ diff --git a/pkg/providers/github/github.go b/pkg/providers/github/github.go index 53fe3f4..54163f3 100644 --- a/pkg/providers/github/github.go +++ b/pkg/providers/github/github.go @@ -35,8 +35,9 @@ type Provider struct { // Compile-time interface checks. var ( - _ providers.Poll = (*Provider)(nil) - _ providers.Webhook = (*Provider)(nil) + _ providers.Poll = (*Provider)(nil) + _ providers.Webhook = (*Provider)(nil) + _ providers.RunnerNameReporter = (*Provider)(nil) ) // New creates a GitHub provider wrapping an existing GitHub client. @@ -231,6 +232,17 @@ func (p *Provider) convertEvent(ev github.JobEvent) providers.JobEvent { fe.JobID = ev.Job.GetID() fe.RunID = ev.Job.GetRunID() fe.Conclusion = ev.Job.GetConclusion() + // GitHub populates runner_name on in_progress and completed + // actions; empty on queued and for jobs cancelled before any + // runner picked them up. + fe.RunnerName = ev.Job.GetRunnerName() } return fe } + +// ReportsRunnerNames implements providers.RunnerNameReporter: GitHub +// workflow_job webhooks carry runner_name on in_progress and completed +// actions, so the scheduler may key runner teardown and the orphan +// sweep on observed assignments for runners dispatched via this +// provider. +func (p *Provider) ReportsRunnerNames() bool { return true } diff --git a/pkg/providers/github/github_test.go b/pkg/providers/github/github_test.go index 9bfd805..37fef18 100644 --- a/pkg/providers/github/github_test.go +++ b/pkg/providers/github/github_test.go @@ -110,6 +110,46 @@ func TestConvertEvent_NilJob(t *testing.T) { } } +func TestConvertEvent_RunnerName(t *testing.T) { + p := New(nil, slog.Default(), "", "") + + id := int64(86444761252) + runnerName := "ephemerd-github-ephpm-kind_cope" + job := &gh.WorkflowJob{ + ID: &id, + RunnerName: &runnerName, + Labels: []string{"self-hosted", "linux"}, + } + + ev := p.convertEvent(ghclient.JobEvent{ + Action: "in_progress", + Repo: "ephpm/ephemerd", + Job: job, + }) + + if ev.RunnerName != runnerName { + t.Errorf("RunnerName = %q, want %q", ev.RunnerName, runnerName) + } + + // Queued events (and cancelled-before-assignment completions) carry + // no runner_name — the field must stay empty, not panic. + ev = p.convertEvent(ghclient.JobEvent{ + Action: "queued", + Repo: "ephpm/ephemerd", + Job: &gh.WorkflowJob{ID: &id}, + }) + if ev.RunnerName != "" { + t.Errorf("RunnerName = %q for queued event, want empty", ev.RunnerName) + } +} + +func TestReportsRunnerNames(t *testing.T) { + p := New(nil, slog.Default(), "", "") + if !p.ReportsRunnerNames() { + t.Error("ReportsRunnerNames() = false, want true — GitHub webhooks carry runner_name") + } +} + func TestStop_NilCancel(t *testing.T) { p := New(nil, slog.Default(), "", "") if err := p.Stop(context.Background()); err != nil { diff --git a/pkg/providers/provider.go b/pkg/providers/provider.go index 753de10..eedc708 100644 --- a/pkg/providers/provider.go +++ b/pkg/providers/provider.go @@ -109,6 +109,19 @@ type Webhook interface { DeregisterWebhooks(ctx context.Context) error } +// RunnerNameReporter is optionally implemented by providers whose +// in_progress / completed JobEvents carry the assigned runner's name +// (JobEvent.RunnerName). The scheduler only trusts "this runner was +// never assigned a job" conclusions — e.g. the orphaned-runner sweep — +// for runners dispatched via providers that report assignments. +type RunnerNameReporter interface { + Provider + + // ReportsRunnerNames returns true when the provider populates + // JobEvent.RunnerName on in_progress and completed events. + ReportsRunnerNames() bool +} + // PollConfig provides settings for poll-based job discovery. type PollConfig struct { PollInterval int // seconds between polls (0 = provider default) @@ -128,6 +141,17 @@ type JobEvent struct { Labels []string // runner labels/tags (e.g., ["self-hosted", "linux", "x64"]) Conclusion string // for completed events: "success", "failure", "cancelled" + // RunnerName is the name of the runner the platform actually assigned + // the job to. GitHub populates it on "in_progress" and "completed" + // workflow_job webhook actions; it is empty on "queued" events, when a + // job was cancelled before any runner picked it up, and for providers + // that don't report it. The scheduler uses this to key runner teardown + // on the OBSERVED assignment rather than the dispatch intent — GitHub + // treats registered JIT runners with matching labels as fungible, so + // the runner ephemerd dispatched "for" a job is not necessarily the + // runner that ran it. + RunnerName string + // Raw holds the original forge-specific object for edge cases. // GitHub: *github.WorkflowJob, Forgejo: task proto, GitLab: job payload. Raw any diff --git a/pkg/scheduler/reassignment_test.go b/pkg/scheduler/reassignment_test.go new file mode 100644 index 0000000..3c8852a --- /dev/null +++ b/pkg/scheduler/reassignment_test.go @@ -0,0 +1,517 @@ +package scheduler + +// Tests for the runner-reassignment race: GitHub treats registered JIT +// runners with matching labels as fungible, so the runner ephemerd +// dispatches "for" job A can be assigned job B by GitHub's scheduler. +// Teardown must follow the OBSERVED assignment (the runner named in the +// in_progress / completed webhook), never the dispatch intent — +// destroying job.dispatched on job A's completion used to kill whatever +// job that runner was actually executing (observed in production as +// jobs failing at ~10m01s with no logs after their runner vanished). + +import ( + "context" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +// reportingProvider wraps claimCountingProvider with +// providers.RunnerNameReporter so runners dispatched through it are +// eligible for the orphan sweep (like the real GitHub provider). +type reportingProvider struct { + *claimCountingProvider +} + +func (p *reportingProvider) ReportsRunnerNames() bool { return true } + +var _ providers.RunnerNameReporter = (*reportingProvider)(nil) + +// dispatchLinuxJob drives a queued Linux job through handleQueued +// against the fake dispatch server and returns the name of the runner +// that was dispatched for it. +func dispatchLinuxJob(t *testing.T, s *Scheduler, prov providers.Provider, jobID int64) string { + t.Helper() + + event := providers.JobEvent{ + Provider: prov, + Action: "queued", + Repo: "myrepo", + JobID: jobID, + Labels: []string{"self-hosted", "linux"}, + } + s.handleQueued(context.Background(), event) + + s.mu.Lock() + rj, ok := s.running[keyFor(event)] + s.mu.Unlock() + if !ok { + t.Fatalf("job %d not tracked in running after handleQueued", jobID) + } + if rj.dispatched == "" { + t.Fatalf("job %d has no dispatched runner name", jobID) + } + return rj.dispatched +} + +// destroyedNames snapshots the ids the fake dispatch server has been +// asked to destroy so far. +func destroyedNames(fake *fakeDispatchServer) map[string]bool { + fake.mu.Lock() + defer fake.mu.Unlock() + out := map[string]bool{} + for _, req := range fake.destroyRequests { + out[req.Id] = true + } + return out +} + +func waitForDestroy(t *testing.T, fake *fakeDispatchServer, name string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + if destroyedNames(fake)[name] { + return + } + if time.Now().After(deadline) { + t.Fatalf("runner %q was not destroyed within deadline (destroyed: %v)", name, destroyedNames(fake)) + } + time.Sleep(10 * time.Millisecond) + } +} + +// TestHandleCompleted_ReassignedRunner_ProductionScenario replays the +// production race end to end: +// +// dispatch runner R for job A, runner S for job B +// GitHub swaps them: in_progress(B, R), in_progress(A, S) +// completed(A, S) → must destroy S, must NOT touch R (mid-flight on B) +// completed(B, R) → must destroy R +func TestHandleCompleted_ReassignedRunner_ProductionScenario(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 4, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + + const jobA, jobB int64 = 101, 202 + runnerR := dispatchLinuxJob(t, s, prov, jobA) + runnerS := dispatchLinuxJob(t, s, prov, jobB) + keyA := jobKey{Provider: prov.Name(), JobID: jobA} + keyB := jobKey{Provider: prov.Name(), JobID: jobB} + + // GitHub swaps the assignments. + s.handleInProgress(providers.JobEvent{ + Provider: prov, Action: "in_progress", Repo: "myrepo", JobID: jobB, RunnerName: runnerR, + }) + s.handleInProgress(providers.JobEvent{ + Provider: prov, Action: "in_progress", Repo: "myrepo", JobID: jobA, RunnerName: runnerS, + }) + + s.mu.Lock() + rbR, okR := s.runners[runnerR] + rbS, okS := s.runners[runnerS] + s.mu.Unlock() + if !okR || !rbR.bound || rbR.boundKey != keyB { + t.Fatalf("runner R binding = %+v, want bound to job B", rbR) + } + if !okS || !rbS.bound || rbS.boundKey != keyA { + t.Fatalf("runner S binding = %+v, want bound to job A", rbS) + } + + // Job A completes — on runner S. The old code destroyed + // job.dispatched (= R) here, killing job B mid-flight. + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobA, RunnerName: runnerS, Conclusion: "success", + }) + + waitForDestroy(t, fake, runnerS) + if destroyedNames(fake)[runnerR] { + t.Fatalf("runner R was destroyed on job A's completion — the reassignment race regressed") + } + + // R's bookkeeping (filed under its dispatch intent, job A) must survive. + s.mu.Lock() + rj, rStillTracked := s.running[keyA] + _, ledgerHasR := s.runners[runnerR] + _, bStillTracked := s.running[keyB] + s.mu.Unlock() + if !rStillTracked || rj.dispatched != runnerR { + t.Fatalf("runner R's entry (under intent key A) should survive job A's completion") + } + if !ledgerHasR { + t.Fatal("runner R's ledger entry should survive job A's completion") + } + if bStillTracked { + t.Fatal("runner S's entry (under intent key B) should be removed once S is destroyed") + } + + // Job B completes — on runner R. Now R goes down. + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobB, RunnerName: runnerR, Conclusion: "success", + }) + + waitForDestroy(t, fake, runnerR) + + s.mu.Lock() + nRunning, nLedger := len(s.running), len(s.runners) + s.mu.Unlock() + if nRunning != 0 { + t.Errorf("running map has %d entries after both completions, want 0", nRunning) + } + if nLedger != 0 { + t.Errorf("runner ledger has %d entries after both completions, want 0", nLedger) + } + + close(fake.waitBlock) +} + +// TestHandleCompleted_NoReassignment_HappyPath pins that the common case +// (runner runs exactly the job it was dispatched for) behaves as before. +func TestHandleCompleted_NoReassignment_HappyPath(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 4, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + + const jobA int64 = 303 + runnerR := dispatchLinuxJob(t, s, prov, jobA) + + s.handleInProgress(providers.JobEvent{ + Provider: prov, Action: "in_progress", Repo: "myrepo", JobID: jobA, RunnerName: runnerR, + }) + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobA, RunnerName: runnerR, Conclusion: "success", + }) + + waitForDestroy(t, fake, runnerR) + + s.mu.Lock() + nRunning, nLedger := len(s.running), len(s.runners) + s.mu.Unlock() + if nRunning != 0 { + t.Errorf("running map has %d entries, want 0", nRunning) + } + if nLedger != 0 { + t.Errorf("runner ledger has %d entries, want 0", nLedger) + } + + close(fake.waitBlock) +} + +// TestHandleCompleted_CancelledBeforeAssignment pins the fallback: a +// completed event with no runner name (job cancelled before any runner +// picked it up) still tears down the dispatch-intent runner, as long as +// that runner was never observed running a different job. +func TestHandleCompleted_CancelledBeforeAssignment(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 4, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + + const jobA int64 = 404 + runnerR := dispatchLinuxJob(t, s, prov, jobA) + + // No in_progress ever arrives; the job is cancelled unassigned. + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobA, RunnerName: "", Conclusion: "cancelled", + }) + + waitForDestroy(t, fake, runnerR) + + s.mu.Lock() + nRunning := len(s.running) + s.mu.Unlock() + if nRunning != 0 { + t.Errorf("running map has %d entries, want 0", nRunning) + } + + close(fake.waitBlock) +} + +// TestHandleCompleted_EmptyRunnerName_IntentRunnerBoundElsewhere pins +// the guard on the fallback: when the completed event carries no runner +// name but the dispatch-intent runner is known to be running a DIFFERENT +// job, it must be left alone. +func TestHandleCompleted_EmptyRunnerName_IntentRunnerBoundElsewhere(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 4, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + + const jobA, jobB int64 = 505, 606 + runnerR := dispatchLinuxJob(t, s, prov, jobA) + keyA := jobKey{Provider: prov.Name(), JobID: jobA} + + // GitHub gives job B to runner R (dispatched for A). + s.handleInProgress(providers.JobEvent{ + Provider: prov, Action: "in_progress", Repo: "myrepo", JobID: jobB, RunnerName: runnerR, + }) + + // Job A finishes with no runner name (e.g. cancelled while queued + // after its intended runner was stolen). + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobA, RunnerName: "", Conclusion: "cancelled", + }) + + // R must not be destroyed and its bookkeeping must survive: job B's + // completion is what tears it down. + time.Sleep(50 * time.Millisecond) + if destroyedNames(fake)[runnerR] { + t.Fatal("runner R destroyed by empty-runner-name fallback while bound to job B") + } + s.mu.Lock() + _, tracked := s.running[keyA] + s.mu.Unlock() + if !tracked { + t.Fatal("runner R's entry should survive job A's runnerless completion") + } + + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobB, RunnerName: runnerR, Conclusion: "success", + }) + waitForDestroy(t, fake, runnerR) + + close(fake.waitBlock) +} + +// TestHandleCompleted_ForeignRunner pins that a completed event naming a +// runner ephemerd does not own destroys nothing: the job ran elsewhere +// (peer daemon / GitHub-hosted), and our dispatch-intent runner stays up +// for another assignment or the orphan sweep. +func TestHandleCompleted_ForeignRunner(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 4, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + + const jobA int64 = 707 + runnerR := dispatchLinuxJob(t, s, prov, jobA) + keyA := jobKey{Provider: prov.Name(), JobID: jobA} + + s.handleCompleted(context.Background(), providers.JobEvent{ + Provider: prov, Action: "completed", Repo: "myrepo", + JobID: jobA, RunnerName: "somebody-elses-runner", Conclusion: "success", + }) + + time.Sleep(50 * time.Millisecond) + if len(destroyedNames(fake)) != 0 { + t.Fatalf("nothing should be destroyed for a foreign runner, got %v", destroyedNames(fake)) + } + s.mu.Lock() + _, tracked := s.running[keyA] + _, ledger := s.runners[runnerR] + s.mu.Unlock() + if !tracked || !ledger { + t.Fatal("dispatch-intent runner bookkeeping should survive a foreign-runner completion") + } + + close(fake.waitBlock) +} + +// TestHandleInProgress_UnknownOrEmptyRunner pins the no-op paths. +func TestHandleInProgress_UnknownOrEmptyRunner(t *testing.T) { + s := New(Config{Log: quietLogger()}) + + // Empty runner name: nothing to record. + s.handleInProgress(providers.JobEvent{Action: "in_progress", Repo: "r", JobID: 1}) + // Runner we never dispatched: nothing to record. + s.handleInProgress(providers.JobEvent{Action: "in_progress", Repo: "r", JobID: 1, RunnerName: "not-ours"}) + + s.mu.Lock() + defer s.mu.Unlock() + if len(s.runners) != 0 { + t.Errorf("runner ledger has %d entries, want 0", len(s.runners)) + } +} + +// --- orphan sweep --- + +// seedDispatchedRunner inserts a running entry + ledger entry directly, +// the way trackRunning would, with a controllable dispatch timestamp. +func seedDispatchedRunner(s *Scheduler, prov providers.Provider, jobID int64, name string, dispatchedAt time.Time, bound, observable bool) jobKey { + key := jobKey{Provider: prov.Name(), JobID: jobID} + _, cancel := context.WithCancel(context.Background()) + s.mu.Lock() + s.running[key] = &runningJob{ + provider: prov, + claim: &providers.Claim{RunnerID: jobID * 10, RunnerName: name, Repo: "myrepo"}, + repo: "myrepo", + cancel: cancel, + dispatched: name, + startedAt: dispatchedAt, + } + s.runners[name] = &runnerBinding{ + intentKey: key, + dispatchedAt: dispatchedAt, + bound: bound, + observable: observable, + } + if bound { + s.runners[name].boundKey = jobKey{Provider: prov.Name(), JobID: jobID + 1000} + } + s.mu.Unlock() + return key +} + +// TestSweepOrphanRunners_GraceWindow pins the sweep matrix: only +// runners that are (a) past the grace window, (b) never bound, and +// (c) dispatched via an assignment-reporting provider in webhook mode +// are destroyed and deregistered. +func TestSweepOrphanRunners_GraceWindow(t *testing.T) { + fake := &fakeDispatchServer{} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + OrphanSweep: OrphanSweepConfig{Enabled: true, Grace: 10 * time.Minute}, + Log: quietLogger(), + }) + s.webhookMode = true + + old := time.Now().Add(-11 * time.Minute) + fresh := time.Now().Add(-1 * time.Minute) + + orphanKey := seedDispatchedRunner(s, prov, 1, "runner-orphan", old, false, true) + boundKey := seedDispatchedRunner(s, prov, 2, "runner-bound", old, true, true) + freshKey := seedDispatchedRunner(s, prov, 3, "runner-fresh", fresh, false, true) + pollKey := seedDispatchedRunner(s, prov, 4, "runner-unobservable", old, false, false) + + s.sweepOrphanRunners() + + waitForDestroy(t, fake, "runner-orphan") + got := destroyedNames(fake) + for _, name := range []string{"runner-bound", "runner-fresh", "runner-unobservable"} { + if got[name] { + t.Errorf("sweep destroyed %q, which should have been exempt", name) + } + } + + s.mu.Lock() + _, orphanTracked := s.running[orphanKey] + _, boundTracked := s.running[boundKey] + _, freshTracked := s.running[freshKey] + _, pollTracked := s.running[pollKey] + s.mu.Unlock() + if orphanTracked { + t.Error("orphaned runner should be removed from running") + } + if !boundTracked || !freshTracked || !pollTracked { + t.Error("exempt runners should stay tracked") + } + + // The orphan never ran a job, so it must be deregistered from the + // provider (JIT runners only auto-remove after running a job). + if rel := base.releases.Load(); rel != 1 { + t.Errorf("ReleaseJob called %d times, want 1 (the orphan)", rel) + } +} + +// TestSweepOrphanRunners_DisabledOrPolling pins that the sweep is inert +// when disabled and when in polling mode (no in_progress events means +// "never bound" carries no information). +func TestSweepOrphanRunners_DisabledOrPolling(t *testing.T) { + tests := []struct { + name string + enabled bool + webhookMode bool + }{ + {"disabled", false, true}, + {"polling mode", true, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fake := &fakeDispatchServer{} + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + base := newClaimCountingProvider("github") + prov := &reportingProvider{claimCountingProvider: base} + + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + OrphanSweep: OrphanSweepConfig{Enabled: tt.enabled, Grace: 10 * time.Minute}, + Log: quietLogger(), + }) + s.webhookMode = tt.webhookMode + + key := seedDispatchedRunner(s, prov, 1, "runner-x", time.Now().Add(-1*time.Hour), false, true) + + s.sweepOrphanRunners() + time.Sleep(50 * time.Millisecond) + + if len(destroyedNames(fake)) != 0 { + t.Errorf("sweep destroyed runners while %s: %v", tt.name, destroyedNames(fake)) + } + s.mu.Lock() + _, tracked := s.running[key] + s.mu.Unlock() + if !tracked { + t.Error("runner should stay tracked") + } + }) + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index f182b00..824f24c 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -55,6 +55,18 @@ type Config struct { // (Enabled=false) to keep the pre-existing "log and drop" behavior. Retry RetryConfig + // OrphanSweep configures teardown of dispatched runners that were + // never observed picking up a job. GitHub schedules JIT runners onto + // ANY queued job with matching labels, so the runner dispatched "for" + // a job may end up running a different one — leaving the runner that + // was dispatched for THAT job idle with no job-completion event ever + // pointing at it. The sweep destroys such runners once they have been + // idle-unbound for Grace. Only active in webhook mode and only for + // runners dispatched via providers that report runner assignments + // (providers.RunnerNameReporter) — otherwise "never observed bound" + // would just mean "we had no way to observe it". + OrphanSweep OrphanSweepConfig + // RunnerImageForRepo resolves the per-repo, per-OS image override // configured under [runner.images]. Returns "" when no override is // set; the scheduler then falls back to the provider per-OS default @@ -92,6 +104,22 @@ func (s *Scheduler) resolveImage(ctx context.Context, event *providers.JobEvent, return event.Provider.DefaultImageFor(os) } +// OrphanSweepConfig tunes the orphaned-runner sweep. Zero-valued = +// disabled (matching pre-existing behavior); the CLI enables it by +// default with a 10-minute grace window. +type OrphanSweepConfig struct { + // Enabled toggles the sweep. + Enabled bool + + // Grace is how long a dispatched runner may remain unbound (never + // seen in an in_progress event) before it is destroyed. Defaults to + // 10 minutes when zero. + Grace time.Duration +} + +// defaultOrphanGrace is applied when OrphanSweepConfig.Grace is zero. +const defaultOrphanGrace = 10 * time.Minute + // jobKey uniquely identifies a job across providers. Different providers // can return the same int64 job ID, so we include the provider name. type jobKey struct { @@ -117,6 +145,8 @@ type Scheduler struct { seen map[jobKey]time.Time // recently handled jobs for dedup pending map[jobKey]struct{} // jobs dispatched to a handler but not yet holding sem attempts map[jobKey]int // provisioning passes per job, for zombie detection + runners map[string]*runnerBinding // dispatched runners by name; tracks observed job assignment + webhookMode bool // true when job events arrive via webhooks (in_progress observable) mu sync.Mutex sem chan struct{} // local/native job concurrency limiter linuxSem chan struct{} // Linux dispatch (VM) concurrency limiter @@ -197,7 +227,69 @@ type runningJob struct { startedAt time.Time } +// runnerName returns the name the runner was registered under with the +// provider. Every provisioning path stores it on the claim; the Linux +// dispatch path mirrors it in dispatched. +func (rj *runningJob) runnerName() string { + if rj.dispatched != "" { + return rj.dispatched + } + if rj.claim != nil { + return rj.claim.RunnerName + } + return "" +} + +// runnerBinding tracks which job a dispatched runner ACTUALLY picked up. +// +// ephemerd registers one JIT runner per queued job, but the platform's +// scheduler does not honor that pairing: GitHub hands a registered +// runner ANY queued job with matching labels. When several same-label +// jobs queue at once (every multi-job workflow run), the runner +// dispatched "for" job A routinely ends up running job B. Teardown must +// therefore be keyed on the observed assignment (in_progress / +// completed runner_name), not the dispatch intent — destroying +// job.dispatched when job A completes kills whatever job the runner is +// actually executing. +type runnerBinding struct { + intentKey jobKey // job the runner was dispatched for (key into s.running) + boundKey jobKey // job the platform assigned (valid when bound) + bound bool // true once an in_progress event named this runner + dispatchedAt time.Time // when the runner was provisioned (orphan sweep) + observable bool // provider reports runner assignments (RunnerNameReporter) +} + +// trackRunning files a provisioned runner's bookkeeping under its +// dispatch-intent job key and opens a runner-name ledger entry so +// in_progress / completed events can locate the runner by NAME +// regardless of which job the platform actually assigned to it. +func (s *Scheduler) trackRunning(key jobKey, rj *runningJob, provider providers.Provider) { + observable := false + if rnr, ok := provider.(providers.RunnerNameReporter); ok { + observable = rnr.ReportsRunnerNames() + } + s.mu.Lock() + s.running[key] = rj + if name := rj.runnerName(); name != "" { + s.runners[name] = &runnerBinding{ + intentKey: key, + dispatchedAt: time.Now(), + observable: observable, + } + } + s.mu.Unlock() + metrics.JobsActive.Inc() +} +// untrackRunningLocked removes a runner's bookkeeping (running entry + +// runner-name ledger). Caller holds s.mu and is responsible for the +// actual teardown of the runner's resources. +func (s *Scheduler) untrackRunningLocked(key jobKey, rj *runningJob) { + delete(s.running, key) + if name := rj.runnerName(); name != "" { + delete(s.runners, name) + } +} // New creates a scheduler. func New(cfg Config) *Scheduler { @@ -234,6 +326,7 @@ func New(cfg Config) *Scheduler { seen: make(map[jobKey]time.Time), pending: make(map[jobKey]struct{}), attempts: make(map[jobKey]int), + runners: make(map[string]*runnerBinding), sem: make(chan struct{}, cfg.MaxConcurrent), linuxSem: make(chan struct{}, cfg.MaxConcurrent), macSem: make(chan struct{}, macVMs), @@ -466,7 +559,14 @@ func (s *Scheduler) Run(ctx context.Context) error { }() } - // Periodically clean up the seen-jobs dedup map + // Record the discovery mode: the orphaned-runner sweep is only safe + // when in_progress events are observable, i.e. in webhook mode. + s.mu.Lock() + s.webhookMode = useWebhook + s.mu.Unlock() + + // Periodically clean up the seen-jobs dedup map and sweep orphaned + // runners (dispatched but never assigned a job by the platform). cleanupTicker := time.NewTicker(5 * time.Minute) defer cleanupTicker.Stop() @@ -475,6 +575,7 @@ func (s *Scheduler) Run(ctx context.Context) error { select { case <-cleanupTicker.C: s.cleanSeen() + s.sweepOrphanRunners() case <-ctx.Done(): s.cfg.Log.Info("shutting down scheduler") @@ -487,13 +588,7 @@ func (s *Scheduler) Run(ctx context.Context) error { metrics.JobsQueuedTotal.Inc() go s.handleQueued(ctx, event) case "in_progress": - // Job was picked up (possibly by a peer daemon - // sharing the installation token). Drop any - // outstanding retry so we stop reattempting a job - // that's already running elsewhere. Nil-safe. - if s.retry != nil { - s.retry.Drop(keyFor(event)) - } + s.handleInProgress(event) case "completed": go s.handleCompleted(ctx, event) } @@ -730,8 +825,7 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent } // Track the dispatched job (env is nil — lifecycle managed by Linux VM worker) - s.mu.Lock() - s.running[key] = &runningJob{ + s.trackRunning(key, &runningJob{ provider: event.Provider, claim: claim, repo: event.Repo, @@ -739,9 +833,7 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent cancel: cancel, dispatched: claim.RunnerName, startedAt: time.Now(), - } - s.mu.Unlock() - metrics.JobsActive.Inc() + }, event.Provider) log.Info("Linux runner dispatched", "name", claim.RunnerName) @@ -762,13 +854,17 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent log.Info("dispatched runner exited", "exit_code", exitCode) } - // Always clean up + // Always clean up. The entry may already be gone if a completed + // event tore this runner down by name (see handleCompleted). s.mu.Lock() rj, exists := s.running[key] if exists { - delete(s.running, key) + s.untrackRunningLocked(key, rj) } s.mu.Unlock() + if exists { + metrics.JobsActive.Dec() + } if err := s.cfg.LinuxDispatcher.Destroy(context.Background(), claim.RunnerName); err != nil { log.Warn("dispatch destroy failed", "error", err) @@ -914,9 +1010,8 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent } // Track the running job - s.mu.Lock() - s.running[key] = &runningJob{ - provider: event.Provider, + s.trackRunning(key, &runningJob{ + provider: event.Provider, claim: claim, repo: event.Repo, image: image, @@ -924,9 +1019,7 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent artifactsDir: artifactsDir, macosVM: macVM, startedAt: time.Now(), - } - s.mu.Unlock() - metrics.JobsActive.Inc() + }, event.Provider) log.Info("macOS VM runner ready", "name", claim.RunnerName, "ip", ip) @@ -951,8 +1044,9 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent s.mu.Lock() rj, exists := s.running[key] if exists { - delete(s.running, key) + s.untrackRunningLocked(key, rj) s.mu.Unlock() + metrics.JobsActive.Dec() rj.macosVM.Stop() if rj.artifactsDir != "" { artifacts.Cleanup(rj.artifactsDir, s.cfg.Log) @@ -1044,17 +1138,14 @@ func (s *Scheduler) handleNativeMacOSJob(ctx context.Context, event providers.Jo } // Track the running job - s.mu.Lock() - s.running[key] = &runningJob{ + s.trackRunning(key, &runningJob{ provider: event.Provider, claim: claim, repo: event.Repo, cancel: cancel, nativeRunner: nr, startedAt: time.Now(), - } - s.mu.Unlock() - metrics.JobsActive.Inc() + }, event.Provider) log.Info("native macOS runner started", "name", claim.RunnerName) @@ -1079,8 +1170,9 @@ func (s *Scheduler) handleNativeMacOSJob(ctx context.Context, event providers.Jo s.mu.Lock() rj, exists := s.running[key] if exists { - delete(s.running, key) + s.untrackRunningLocked(key, rj) s.mu.Unlock() + metrics.JobsActive.Dec() nr.Stop() if rj.provider != nil && rj.claim != nil { if err := rj.provider.ReleaseJob(context.Background(), rj.claim); err != nil { @@ -1218,18 +1310,16 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent } // Track the running job - s.mu.Lock() - s.running[key] = &runningJob{ + s.trackRunning(key, &runningJob{ env: env, + provider: event.Provider, claim: claim, repo: event.Repo, image: image, cancel: cancel, artifactsDir: artifactsDir, startedAt: time.Now(), - } - s.mu.Unlock() - metrics.JobsActive.Inc() + }, event.Provider) log.Info("runner environment ready", "name", claim.RunnerName) @@ -1258,8 +1348,9 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent s.mu.Lock() rj, exists := s.running[key] if exists { - delete(s.running, key) + s.untrackRunningLocked(key, rj) s.mu.Unlock() + metrics.JobsActive.Dec() if err := s.cfg.Runtime.Destroy(context.Background(), env); err != nil { log.Warn("failed to destroy runner environment", "error", err) } @@ -1277,6 +1368,51 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent }() } +// handleInProgress records which runner the platform ACTUALLY assigned +// a job to. GitHub schedules JIT runners onto any queued job with +// matching labels, so this routinely differs from the dispatch intent; +// the binding recorded here is what keeps handleCompleted from +// destroying a runner that is mid-flight on someone else's job. It also +// drops any outstanding claim retry for the job (it's running — possibly +// on a peer daemon — so re-attempting would register ghost runners). +func (s *Scheduler) handleInProgress(event providers.JobEvent) { + key := keyFor(event) + if s.retry != nil { + s.retry.Drop(key) + } + + name := event.RunnerName + if name == "" { + return + } + + s.mu.Lock() + rb, owned := s.runners[name] + var intentKey jobKey + if owned { + rb.bound = true + rb.boundKey = key + intentKey = rb.intentKey + } + s.mu.Unlock() + + if !owned { + return + } + if intentKey != key { + // The observability line for the fungibility race: GitHub gave + // our runner a different job than the one it was dispatched for. + s.cfg.Log.Info("runner picked up a different job than it was dispatched for", + "runner", name, + "dispatched_for_job", intentKey.JobID, + "assigned_job", key.JobID, + "repo", event.Repo, + "detail", "GitHub treats same-label JIT runners as fungible; teardown will follow the observed assignment") + } else { + s.cfg.Log.Debug("runner bound to its dispatched job", "runner", name, "job_id", key.JobID) + } +} + func (s *Scheduler) handleCompleted(ctx context.Context, event providers.JobEvent) { jobID := event.JobID key := keyFor(event) @@ -1289,10 +1425,45 @@ func (s *Scheduler) handleCompleted(ctx context.Context, event providers.JobEven s.retry.Drop(key) } + // Resolve which runner to tear down. The runner that ran this job is + // the one NAMED IN THE EVENT — not necessarily the one dispatched + // when the job queued (GitHub reassigns same-label JIT runners + // freely). Destroying job.dispatched here used to kill whichever job + // the reassigned runner was actually executing. s.mu.Lock() - job, exists := s.running[key] + var job *runningJob + exists := false + ownerKey := key + if name := event.RunnerName; name != "" { + if rb, owned := s.runners[name]; owned { + if rj, ok := s.running[rb.intentKey]; ok && rj.runnerName() == name { + job, ownerKey, exists = rj, rb.intentKey, true + } else { + // Ledger points at an entry the wait-goroutine already + // cleaned up — drop the stale ledger record. + delete(s.runners, name) + } + } + // Not ours (peer daemon / foreign runner): nothing to destroy. + // If we dispatched an intent runner for this job it is still + // alive and unbound — it will pick up another queued job, exit + // on its own, or be culled by the orphan sweep. + } else if rj, ok := s.running[key]; ok { + // No runner named in the event (job cancelled before any runner + // picked it up, or a provider that doesn't report runner names). + // Fall back to the dispatch-intent runner, but only when it was + // never observed running a DIFFERENT job. + rb := s.runners[rj.runnerName()] + if rb == nil || !rb.bound || rb.boundKey == key { + job, exists = rj, true + } else { + log.Info("job completed without a runner name; leaving dispatch-intent runner alone (it is bound to another job)", + "runner", rj.runnerName(), + "bound_job", rb.boundKey.JobID) + } + } if exists { - delete(s.running, key) + s.untrackRunningLocked(ownerKey, job) } s.mu.Unlock() @@ -1303,7 +1474,13 @@ func (s *Scheduler) handleCompleted(ctx context.Context, event providers.JobEven conclusion := event.Conclusion log.Info("job completed, destroying runner environment", "conclusion", conclusion, + "runner", job.runnerName(), ) + if ownerKey != key { + log.Info("destroying runner under its observed assignment, not its dispatch intent", + "runner", job.runnerName(), + "dispatched_for_job", ownerKey.JobID) + } // Record metrics providerName := "" @@ -1383,6 +1560,7 @@ func (s *Scheduler) destroyAll() { jobs[k] = v } s.running = make(map[jobKey]*runningJob) + s.runners = make(map[string]*runnerBinding) s.mu.Unlock() for key, job := range jobs { @@ -1449,6 +1627,94 @@ func (s *Scheduler) cleanSeen() { } } +// sweepOrphanRunners destroys dispatched runners that were never +// observed picking up a job within the configured grace window. +// +// Before teardown was keyed on observed assignments, every completed +// event destroyed the runner dispatched for that job — which implicitly +// (and often wrongly) cleaned up runners whose job was taken by a peer. +// Now that a completed event only touches the runner named in it, a +// runner whose intended job was cancelled before assignment (with no +// runner_name in the completed event and a binding elsewhere), or whose +// job was grabbed by another daemon's runner, has no event that will +// ever destroy it. The sweep is that replacement cleanup. +// +// Safety: only runs in webhook mode (in poll mode there are no +// in_progress events, so "never observed bound" is meaningless) and only +// for runners dispatched via providers that report runner assignments. +func (s *Scheduler) sweepOrphanRunners() { + if !s.cfg.OrphanSweep.Enabled { + return + } + grace := s.cfg.OrphanSweep.Grace + if grace <= 0 { + grace = defaultOrphanGrace + } + + type victim struct { + name string + key jobKey + rj *runningJob + } + var victims []victim + + s.mu.Lock() + if !s.webhookMode { + s.mu.Unlock() + return + } + for name, rb := range s.runners { + if rb.bound || !rb.observable || time.Since(rb.dispatchedAt) < grace { + continue + } + rj, ok := s.running[rb.intentKey] + if !ok { + // Stale ledger entry — the wait-goroutine already cleaned up. + delete(s.runners, name) + continue + } + if rj.runnerName() != name { + continue + } + s.untrackRunningLocked(rb.intentKey, rj) + victims = append(victims, victim{name: name, key: rb.intentKey, rj: rj}) + } + s.mu.Unlock() + + for _, v := range victims { + s.cfg.Log.Warn("destroying orphaned runner: dispatched but never assigned a job within the grace window", + "runner", v.name, + "dispatched_for_job", v.key.JobID, + "grace", grace) + metrics.JobsActive.Dec() + v.rj.cancel() + if v.rj.macosVM != nil { + v.rj.macosVM.Stop() + } else if v.rj.nativeRunner != nil { + v.rj.nativeRunner.Stop() + } else if v.rj.dispatched != "" && s.cfg.LinuxDispatcher != nil { + if err := s.cfg.LinuxDispatcher.Destroy(context.Background(), v.rj.dispatched); err != nil { + s.cfg.Log.Warn("failed to destroy orphaned dispatched runner", "runner", v.name, "error", err) + } + } else if v.rj.env != nil { + if err := s.cfg.Runtime.Destroy(context.Background(), v.rj.env); err != nil { + s.cfg.Log.Warn("failed to destroy orphaned runner environment", "runner", v.name, "error", err) + } + } + if v.rj.artifactsDir != "" { + artifacts.Cleanup(v.rj.artifactsDir, s.cfg.Log) + } + // Deregister the JIT runner from the provider: it never ran a + // job, so it will not auto-remove itself and would otherwise + // linger as an offline ghost. + if v.rj.provider != nil && v.rj.claim != nil { + if err := v.rj.provider.ReleaseJob(context.Background(), v.rj.claim); err != nil { + s.cfg.Log.Debug("deregister orphaned runner", "runner", v.name, "error", err) + } + } + } +} + // enqueueRetryIfEligible passes err through the retry queue, if enabled. // The retryHandler callback re-invokes handleQueued with the ORIGINAL // event when the backoff timer fires. Non-retryable errors and disabled