From 5e45662475509ac3fbade0732d707924371c18a7 Mon Sep 17 00:00:00 2001 From: Benoit Sigoure Date: Thu, 30 Jul 2026 00:25:07 +0000 Subject: [PATCH 1/2] webapp: add per scale set job queue view When all runners of a scale set are busy, it is hard to tell where a given workflow job sits in the queue, or why capacity is not being used. This adds a "Job Queue" view to the web UI that groups queued and running jobs per scale set (and per pool, matched by labels), ordered by request time, with links to the GitHub run/job pages and to the scale set/pool detail pages. To support the view: * Jobs recorded by scale set listeners now carry the garm scale set ID (new WorkflowJob.ScaleSetFkID column, exposed as scale_set_id on the Job API model). Previously the scale set ID was dropped when recording jobs. * The full RunnerScaleSetStatistic from each session message is persisted on the scale set and exposed as "statistics" on the API (previously only TotalAssignedJobs survived as desired_runner_count). The view shows GitHub's numbers (assigned jobs, busy/idle runners) next to GARM's instance counts, making divergence visible. * consolidateRunnerState syncs GitHub's per-runner view (online idle/busy, offline) onto instances that finished installing. Without this, a runner whose agent died after setup stayed "idle" in GARM forever while GitHub considered it offline and never assigned it jobs. The view updates live via the existing job/instance websocket events. Co-Authored-By: Claude Fable 5 --- database/common/mocks/Store.go | 64 +-- database/common/store.go | 2 +- database/sql/jobs.go | 14 + database/sql/models.go | 9 + database/sql/scalesets.go | 10 +- database/sql/scalesets_test.go | 2 +- database/sql/util.go | 8 + database/watcher/watcher_store_test.go | 5 +- params/params.go | 8 +- webapp/src/lib/api/generated-client.ts | 11 + webapp/src/lib/api/generated/api.ts | 26 ++ webapp/src/lib/components/Navigation.svelte | 5 + webapp/src/routes/queue/+page.svelte | 408 ++++++++++++++++++ .../src/routes/queue/page.integration.test.ts | 295 +++++++++++++ webapp/swagger.yaml | 41 ++ workers/scaleset/interfaces.go | 2 +- workers/scaleset/scaleset.go | 36 +- workers/scaleset/scaleset_helper.go | 7 +- workers/scaleset/scaleset_listener.go | 6 +- 19 files changed, 913 insertions(+), 46 deletions(-) create mode 100644 webapp/src/routes/queue/+page.svelte create mode 100644 webapp/src/routes/queue/page.integration.test.ts diff --git a/database/common/mocks/Store.go b/database/common/mocks/Store.go index 9d91d5e5a..b4c9e8924 100644 --- a/database/common/mocks/Store.go +++ b/database/common/mocks/Store.go @@ -5572,17 +5572,17 @@ func (_c *Store_SetEntityPoolManagerStatus_Call) RunAndReturn(run func(context.C return _c } -// SetScaleSetDesiredRunnerCount provides a mock function with given fields: ctx, scaleSetID, desiredRunnerCount -func (_m *Store) SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID uint, desiredRunnerCount int) error { - ret := _m.Called(ctx, scaleSetID, desiredRunnerCount) +// SetScaleSetLastMessageID provides a mock function with given fields: ctx, scaleSetID, lastMessageID +func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error { + ret := _m.Called(ctx, scaleSetID, lastMessageID) if len(ret) == 0 { - panic("no return value specified for SetScaleSetDesiredRunnerCount") + panic("no return value specified for SetScaleSetLastMessageID") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, uint, int) error); ok { - r0 = rf(ctx, scaleSetID, desiredRunnerCount) + if rf, ok := ret.Get(0).(func(context.Context, uint, int64) error); ok { + r0 = rf(ctx, scaleSetID, lastMessageID) } else { r0 = ret.Error(0) } @@ -5590,47 +5590,47 @@ func (_m *Store) SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID u return r0 } -// Store_SetScaleSetDesiredRunnerCount_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetDesiredRunnerCount' -type Store_SetScaleSetDesiredRunnerCount_Call struct { +// Store_SetScaleSetLastMessageID_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetLastMessageID' +type Store_SetScaleSetLastMessageID_Call struct { *mock.Call } -// SetScaleSetDesiredRunnerCount is a helper method to define mock.On call +// SetScaleSetLastMessageID is a helper method to define mock.On call // - ctx context.Context // - scaleSetID uint -// - desiredRunnerCount int -func (_e *Store_Expecter) SetScaleSetDesiredRunnerCount(ctx interface{}, scaleSetID interface{}, desiredRunnerCount interface{}) *Store_SetScaleSetDesiredRunnerCount_Call { - return &Store_SetScaleSetDesiredRunnerCount_Call{Call: _e.mock.On("SetScaleSetDesiredRunnerCount", ctx, scaleSetID, desiredRunnerCount)} +// - lastMessageID int64 +func (_e *Store_Expecter) SetScaleSetLastMessageID(ctx interface{}, scaleSetID interface{}, lastMessageID interface{}) *Store_SetScaleSetLastMessageID_Call { + return &Store_SetScaleSetLastMessageID_Call{Call: _e.mock.On("SetScaleSetLastMessageID", ctx, scaleSetID, lastMessageID)} } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) Run(run func(ctx context.Context, scaleSetID uint, desiredRunnerCount int)) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) Run(run func(ctx context.Context, scaleSetID uint, lastMessageID int64)) *Store_SetScaleSetLastMessageID_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(uint), args[2].(int)) + run(args[0].(context.Context), args[1].(uint), args[2].(int64)) }) return _c } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) Return(_a0 error) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) Return(_a0 error) *Store_SetScaleSetLastMessageID_Call { _c.Call.Return(_a0) return _c } -func (_c *Store_SetScaleSetDesiredRunnerCount_Call) RunAndReturn(run func(context.Context, uint, int) error) *Store_SetScaleSetDesiredRunnerCount_Call { +func (_c *Store_SetScaleSetLastMessageID_Call) RunAndReturn(run func(context.Context, uint, int64) error) *Store_SetScaleSetLastMessageID_Call { _c.Call.Return(run) return _c } -// SetScaleSetLastMessageID provides a mock function with given fields: ctx, scaleSetID, lastMessageID -func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error { - ret := _m.Called(ctx, scaleSetID, lastMessageID) +// SetScaleSetRunnerStatistics provides a mock function with given fields: ctx, scaleSetID, stats +func (_m *Store) SetScaleSetRunnerStatistics(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) error { + ret := _m.Called(ctx, scaleSetID, stats) if len(ret) == 0 { - panic("no return value specified for SetScaleSetLastMessageID") + panic("no return value specified for SetScaleSetRunnerStatistics") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, uint, int64) error); ok { - r0 = rf(ctx, scaleSetID, lastMessageID) + if rf, ok := ret.Get(0).(func(context.Context, uint, params.RunnerScaleSetStatistic) error); ok { + r0 = rf(ctx, scaleSetID, stats) } else { r0 = ret.Error(0) } @@ -5638,32 +5638,32 @@ func (_m *Store) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, return r0 } -// Store_SetScaleSetLastMessageID_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetLastMessageID' -type Store_SetScaleSetLastMessageID_Call struct { +// Store_SetScaleSetRunnerStatistics_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetScaleSetRunnerStatistics' +type Store_SetScaleSetRunnerStatistics_Call struct { *mock.Call } -// SetScaleSetLastMessageID is a helper method to define mock.On call +// SetScaleSetRunnerStatistics is a helper method to define mock.On call // - ctx context.Context // - scaleSetID uint -// - lastMessageID int64 -func (_e *Store_Expecter) SetScaleSetLastMessageID(ctx interface{}, scaleSetID interface{}, lastMessageID interface{}) *Store_SetScaleSetLastMessageID_Call { - return &Store_SetScaleSetLastMessageID_Call{Call: _e.mock.On("SetScaleSetLastMessageID", ctx, scaleSetID, lastMessageID)} +// - stats params.RunnerScaleSetStatistic +func (_e *Store_Expecter) SetScaleSetRunnerStatistics(ctx interface{}, scaleSetID interface{}, stats interface{}) *Store_SetScaleSetRunnerStatistics_Call { + return &Store_SetScaleSetRunnerStatistics_Call{Call: _e.mock.On("SetScaleSetRunnerStatistics", ctx, scaleSetID, stats)} } -func (_c *Store_SetScaleSetLastMessageID_Call) Run(run func(ctx context.Context, scaleSetID uint, lastMessageID int64)) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) Run(run func(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic)) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(uint), args[2].(int64)) + run(args[0].(context.Context), args[1].(uint), args[2].(params.RunnerScaleSetStatistic)) }) return _c } -func (_c *Store_SetScaleSetLastMessageID_Call) Return(_a0 error) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) Return(_a0 error) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Return(_a0) return _c } -func (_c *Store_SetScaleSetLastMessageID_Call) RunAndReturn(run func(context.Context, uint, int64) error) *Store_SetScaleSetLastMessageID_Call { +func (_c *Store_SetScaleSetRunnerStatistics_Call) RunAndReturn(run func(context.Context, uint, params.RunnerScaleSetStatistic) error) *Store_SetScaleSetRunnerStatistics_Call { _c.Call.Return(run) return _c } diff --git a/database/common/store.go b/database/common/store.go index 49e17873a..6d978e6b3 100644 --- a/database/common/store.go +++ b/database/common/store.go @@ -166,7 +166,7 @@ type ScaleSetsStore interface { GetScaleSetByID(ctx context.Context, scaleSet uint) (params.ScaleSet, error) DeleteScaleSetByID(ctx context.Context, scaleSetID uint) (err error) SetScaleSetLastMessageID(ctx context.Context, scaleSetID uint, lastMessageID int64) error - SetScaleSetDesiredRunnerCount(ctx context.Context, scaleSetID uint, desiredRunnerCount int) error + SetScaleSetRunnerStatistics(ctx context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) error } type ScaleSetInstanceStore interface { diff --git a/database/sql/jobs.go b/database/sql/jobs.go index ac1709427..5bbe50320 100644 --- a/database/sql/jobs.go +++ b/database/sql/jobs.go @@ -72,6 +72,10 @@ func sqlWorkflowJobToParamsJob(job WorkflowJob) (params.Job, error) { jobParam.RunnerName = job.Instance.Name } + if job.ScaleSetFkID != nil { + jobParam.ScaleSetID = *job.ScaleSetFkID + } + return jobParam, nil } @@ -104,6 +108,11 @@ func (s *sqlDatabase) paramsJobToWorkflowJob(ctx context.Context, conn *gorm.DB, LockedBy: job.LockedBy, } + if job.ScaleSetID != 0 { + scaleSetID := job.ScaleSetID + workflofJob.ScaleSetFkID = &scaleSetID + } + if job.RunnerName != "" { instance, err := s.getInstance(s.ctx, conn, job.RunnerName) if err != nil { @@ -360,6 +369,11 @@ func (s *sqlDatabase) CreateOrUpdateJob(ctx context.Context, job params.Job) (pa if job.ForgeInstanceID != nil { workflowJob.ForgeInstanceID = job.ForgeInstanceID } + + if job.ScaleSetID != 0 { + scaleSetID := job.ScaleSetID + workflowJob.ScaleSetFkID = &scaleSetID + } if err := tx.Save(&workflowJob).Error; err != nil { return fmt.Errorf("error saving job: %w", err) } diff --git a/database/sql/models.go b/database/sql/models.go index 2dec823e5..be00b868e 100644 --- a/database/sql/models.go +++ b/database/sql/models.go @@ -219,6 +219,10 @@ type ScaleSet struct { Enabled bool LastMessageID int64 DesiredRunnerCount int + // RunnerStatistics is the last RunnerScaleSetStatistic received from + // GitHub on the message session (busy/idle/assigned counts as GitHub + // sees them). + RunnerStatistics datatypes.JSON // ExtraSpecs is an opaque json that gets sent to the provider // as part of the bootstrap params for instances. It can contain // any kind of data needed by providers. @@ -474,6 +478,11 @@ type WorkflowJob struct { // ScaleSetJobID is the job ID for a scaleset job. ScaleSetJobID string `gorm:"index:scaleset_job_id_idx"` + // ScaleSetFkID is the ID of the scale set that this job was assigned to, + // if the job came in through a scale set listener. + ScaleSetFkID *uint `gorm:"index"` + ScaleSet ScaleSet `gorm:"foreignKey:ScaleSetFkID"` + // RunID is the ID of the workflow run. A run may have multiple jobs. RunID int64 // Action is the specific activity that triggered the event. diff --git a/database/sql/scalesets.go b/database/sql/scalesets.go index 6e597c602..275aeb809 100644 --- a/database/sql/scalesets.go +++ b/database/sql/scalesets.go @@ -16,6 +16,7 @@ package sql import ( "context" + "encoding/json" "errors" "fmt" @@ -525,7 +526,7 @@ func (s *sqlDatabase) SetScaleSetLastMessageID(_ context.Context, scaleSetID uin return nil } -func (s *sqlDatabase) SetScaleSetDesiredRunnerCount(_ context.Context, scaleSetID uint, desiredRunnerCount int) (err error) { +func (s *sqlDatabase) SetScaleSetRunnerStatistics(_ context.Context, scaleSetID uint, stats params.RunnerScaleSetStatistic) (err error) { var scaleSet params.ScaleSet var rowsAffected int64 defer func() { @@ -538,8 +539,13 @@ func (s *sqlDatabase) SetScaleSetDesiredRunnerCount(_ context.Context, scaleSetI if err != nil { return fmt.Errorf("error fetching scale set: %w", err) } + asJSON, err := json.Marshal(stats) + if err != nil { + return fmt.Errorf("error marshaling runner statistics: %w", err) + } result := tx.Model(&dbSet).Updates(map[string]interface{}{ - "desired_runner_count": desiredRunnerCount, + "desired_runner_count": stats.TotalAssignedJobs, + "runner_statistics": datatypes.JSON(asJSON), }) if result.Error != nil { return fmt.Errorf("error saving database entry: %w", result.Error) diff --git a/database/sql/scalesets_test.go b/database/sql/scalesets_test.go index 7b5a46bbc..df3b3bf0a 100644 --- a/database/sql/scalesets_test.go +++ b/database/sql/scalesets_test.go @@ -257,7 +257,7 @@ func (s *ScaleSetsTestSuite) TestScaleSetOperations() { s.T().Run("Set scale set last message ID and desired count", func(_ *testing.T) { err = s.Store.SetScaleSetLastMessageID(s.adminCtx, orgScaleSet.ID, 20) s.Require().NoError(err) - err = s.Store.SetScaleSetDesiredRunnerCount(s.adminCtx, orgScaleSet.ID, 5) + err = s.Store.SetScaleSetRunnerStatistics(s.adminCtx, orgScaleSet.ID, params.RunnerScaleSetStatistic{TotalAssignedJobs: 5}) s.Require().NoError(err) orgScaleSetByID, err := s.Store.GetScaleSetByID(s.adminCtx, orgScaleSet.ID) s.Require().NoError(err) diff --git a/database/sql/util.go b/database/sql/util.go index 0432519c8..9bd1a5169 100644 --- a/database/sql/util.go +++ b/database/sql/util.go @@ -489,6 +489,14 @@ func (s *sqlDatabase) sqlToCommonScaleSet(scaleSet ScaleSet) (params.ScaleSet, e ret.ProxyName = scaleSet.Proxy.Name } + if len(scaleSet.RunnerStatistics) > 0 { + var stats params.RunnerScaleSetStatistic + if err := json.Unmarshal(scaleSet.RunnerStatistics, &stats); err != nil { + return params.ScaleSet{}, fmt.Errorf("error unmarshaling runner statistics: %w", err) + } + ret.Statistics = &stats + } + var ep GithubEndpoint if scaleSet.RepoID != nil { ret.RepoID = scaleSet.RepoID.String() diff --git a/database/watcher/watcher_store_test.go b/database/watcher/watcher_store_test.go index 9e3ab750b..2a4388ce8 100644 --- a/database/watcher/watcher_store_test.go +++ b/database/watcher/watcher_store_test.go @@ -582,7 +582,7 @@ func (s *WatcherStoreTestSuite) TestScaleSetWatcher() { s.T().Fatal("expected payload not received") } - err = s.store.SetScaleSetDesiredRunnerCount(s.ctx, updatedScaleSet.ID, 5) + err = s.store.SetScaleSetRunnerStatistics(s.ctx, updatedScaleSet.ID, params.RunnerScaleSetStatistic{TotalAssignedJobs: 5}) s.Require().NoError(err) select { @@ -600,8 +600,9 @@ func (s *WatcherStoreTestSuite) TestScaleSetWatcher() { select { case event := <-consumer.Watch(): - // We updated last message ID and desired runner count above. + // We updated last message ID and runner statistics above. updatedScaleSet.DesiredRunnerCount = 5 + updatedScaleSet.Statistics = ¶ms.RunnerScaleSetStatistic{TotalAssignedJobs: 5} updatedScaleSet.LastMessageID = 99 payloadFromEvent, ok := event.Payload.(params.ScaleSet) s.Require().True(ok) diff --git a/params/params.go b/params/params.go index b7a80aa3e..dc8086031 100644 --- a/params/params.go +++ b/params/params.go @@ -719,7 +719,10 @@ type ScaleSet struct { Enabled bool `json:"enabled,omitempty"` Instances []Instance `json:"instances,omitempty"` DesiredRunnerCount int `json:"desired_runner_count,omitempty"` - EnableShell bool `json:"enable_shell"` + // Statistics is the last runner scale set statistic received from GitHub + // on the message session (busy/idle/assigned counts as GitHub sees them). + Statistics *RunnerScaleSetStatistic `json:"statistics,omitempty"` + EnableShell bool `json:"enable_shell"` // Generation holds the numeric generation of the scaleset. This number // will be incremented, every time certain settings of the scaleset, which @@ -1418,6 +1421,9 @@ type Job struct { WorkflowJobID int64 `json:"workflow_job_id,omitempty"` // ScaleSetJobID is the job ID when generated for a scale set. ScaleSetJobID string `json:"scaleset_job_id,omitempty"` + // ScaleSetID is the garm ID of the scale set this job was assigned to, + // if it came in through a scale set listener. + ScaleSetID uint `json:"scale_set_id,omitempty"` // RunID is the ID of the workflow run. A run may have multiple jobs. RunID int64 `json:"run_id,omitempty"` // Action is the specific activity that triggered the event. diff --git a/webapp/src/lib/api/generated-client.ts b/webapp/src/lib/api/generated-client.ts index 3c050843e..b6e5dfef1 100644 --- a/webapp/src/lib/api/generated-client.ts +++ b/webapp/src/lib/api/generated-client.ts @@ -21,7 +21,9 @@ import { ProxiesApi, ObjectsApi, ToolsApi, + JobsApi, type GARMAgentRelease, + type Job, type Repository, type Organization, type Enterprise, @@ -152,6 +154,7 @@ export class GeneratedGarmApiClient { private proxiesApi: ProxiesApi; private objectsApi: ObjectsApi; private toolsApi: ToolsApi; + private jobsApi: JobsApi; constructor(baseUrl: string = '') { this.baseUrl = baseUrl || window.location.origin; @@ -188,6 +191,7 @@ export class GeneratedGarmApiClient { this.proxiesApi = new ProxiesApi(this.config); this.objectsApi = new ObjectsApi(this.config); this.toolsApi = new ToolsApi(this.config); + this.jobsApi = new JobsApi(this.config); } // Set authentication token @@ -222,6 +226,7 @@ export class GeneratedGarmApiClient { this.providersApi = new ProvidersApi(this.config); this.firstRunApi = new FirstRunApi(this.config); this.hooksApi = new HooksApi(this.config); + this.jobsApi = new JobsApi(this.config); } // Authentication @@ -644,6 +649,12 @@ export class GeneratedGarmApiClient { await this.scaleSetsApi.deleteScaleSet(id.toString()); } + // Jobs + async listJobs(): Promise { + const response = await this.jobsApi.listJobs(); + return response.data || []; + } + // Instances async listInstances(): Promise { const response = await this.instancesApi.listInstances(); diff --git a/webapp/src/lib/api/generated/api.ts b/webapp/src/lib/api/generated/api.ts index 5f7760990..200da388a 100644 --- a/webapp/src/lib/api/generated/api.ts +++ b/webapp/src/lib/api/generated/api.ts @@ -2325,6 +2325,12 @@ export interface Job { * @memberof Job */ 'runner_name'?: string; + /** + * ScaleSetID is the garm ID of the scale set this job was assigned to, if it came in through a scale set listener. + * @type {number} + * @memberof Job + */ + 'scale_set_id'?: number; /** * ScaleSetJobID is the job ID when generated for a scale set. * @type {string} @@ -3003,6 +3009,20 @@ export interface RunnerPrefix { * @export * @interface ScaleSet */ +/** + * RunnerScaleSetStatistic is the last runner scale set statistic received from GitHub on the message session. + * @export + * @interface RunnerScaleSetStatistic + */ +export interface RunnerScaleSetStatistic { + 'totalAvailableJobs'?: number; + 'totalAcquiredJobs'?: number; + 'totalAssignedJobs'?: number; + 'totalRunningJobs'?: number; + 'totalRegisteredRunners'?: number; + 'totalBusyRunners'?: number; + 'totalIdleRunners'?: number; +} export interface ScaleSet { /** * @@ -3160,6 +3180,12 @@ export interface ScaleSet { * @memberof ScaleSet */ 'proxy_name'?: string; + /** + * Statistics is the last runner scale set statistic received from GitHub on the message session. + * @type {RunnerScaleSetStatistic} + * @memberof ScaleSet + */ + 'statistics'?: RunnerScaleSetStatistic; /** * * @type {string} diff --git a/webapp/src/lib/components/Navigation.svelte b/webapp/src/lib/components/Navigation.svelte index 61cd5496c..ac27400b6 100644 --- a/webapp/src/lib/components/Navigation.svelte +++ b/webapp/src/lib/components/Navigation.svelte @@ -77,6 +77,11 @@ label: 'Scale Sets', icon: 'M4 7v10c0 2.21 3.582 4 8 4s8-1.79 8-4V7M4 7c0 2.21 3.582 4 8 4s8-1.79 8-4M4 7c0-2.21 3.582-4 8-4s8 1.79 8 4m0 5c0 2.21-3.582 4-8 4s-8-1.79-8-4' // Database layers icon }, + { + href: resolve('/queue'), + label: 'Job Queue', + icon: 'M4 6h16M4 10h16M4 14h10M4 18h6' // Queue/list icon + }, { href: resolve('/instances'), label: 'Runners', diff --git a/webapp/src/routes/queue/+page.svelte b/webapp/src/routes/queue/+page.svelte new file mode 100644 index 000000000..91405f639 --- /dev/null +++ b/webapp/src/routes/queue/+page.svelte @@ -0,0 +1,408 @@ + + + + Job Queue - GARM + + +
+ + + {#if error} +
+
+
+

Error

+
{error}
+ +
+
+
+ {/if} + +
+ + +
+ + {#if loading} +
+
+
+ {:else if queueGroups.length === 0} +
+

No queued or running jobs.

+
+ {:else} + {#each queueGroups as group (group.key)} +
+
+
+

+ {#if group.href} + {group.title} + {:else} + {group.title} + {/if} +

+

{group.subtitle}

+
+
+ + {group.queued.length} queued + + {#if group.githubStats} + + GitHub: {group.githubStats.totalAssignedJobs ?? 0} assigned ({group.githubStats.totalBusyRunners ?? 0} busy, {group.githubStats.totalIdleRunners ?? 0} idle) + + {/if} + + {group.runners} runners{group.maxRunners ? ` / ${group.maxRunners} max` : ''} ({group.busyRunners} busy, {group.idleRunners} idle{group.offlineRunners ? `, ${group.offlineRunners} offline` : ''}{group.provisioning ? `, ${group.provisioning} provisioning` : ''}) + +
+
+ + {#if group.queued.length === 0} +
+ Queue is empty. +
+ {:else} +
+ + + + + + + + + + + + {#each group.queued as job, idx (job.id)} + + + + + + + + {/each} + +
#JobRepositoryLabelsWaiting
{idx + 1} + {#if jobUrl(job)} + {job.name || '(unnamed job)'} + {:else} + {job.name || '(unnamed job)'} + {/if} + + {job.repository_owner}/{job.repository_name} + +
+ {#each job.labels || [] as label} + {label} + {/each} +
+ 		+ {waitingFor(job)} +
+
+ {/if} +
+ {/each} + {/if} +
diff --git a/webapp/src/routes/queue/page.integration.test.ts b/webapp/src/routes/queue/page.integration.test.ts new file mode 100644 index 000000000..0dbc37e6d --- /dev/null +++ b/webapp/src/routes/queue/page.integration.test.ts @@ -0,0 +1,295 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { render, screen, waitFor } from '@testing-library/svelte'; + +// Unmock real components that setup.ts might mock +vi.unmock('$lib/components/PageHeader.svelte'); + +vi.mock('$app/navigation', () => ({ + goto: vi.fn() +})); + +vi.mock('$app/stores', () => ({})); + +vi.mock('$lib/api/client.js', () => ({ + garmApi: { + listJobs: vi.fn(), + listInstances: vi.fn() + } +})); + +vi.mock('$lib/stores/eager-cache.js', () => ({ + eagerCache: { + subscribe: vi.fn() + }, + eagerCacheManager: { + getScaleSets: vi.fn(), + getPools: vi.fn(), + retryResource: vi.fn() + } +})); + +vi.mock('$lib/stores/toast.js', () => ({ + toastStore: { + success: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warning: vi.fn(), + add: vi.fn() + } +})); + +vi.mock('$lib/stores/websocket.js', () => ({ + websocketStore: { + subscribeToEntity: vi.fn(() => vi.fn()) + } +})); + +import QueuePage from './+page.svelte'; +import { garmApi } from '$lib/api/client.js'; + +const mockScaleSets = [ + { + id: 5, + name: 'cloudstack-ubuntu24-micro', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 100, + min_idle_runners: 32, + desired_runner_count: 60, + statistics: { totalAssignedJobs: 60, totalBusyRunners: 40, totalIdleRunners: 2 }, + enabled: true + }, + { + id: 7, + name: 'cloudstack-ubuntu24-large', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 10, + enabled: true + } +]; + +const mockPools = [ + { + id: 'aaaabbbb-cccc-dddd-eeee-ffff00001111', + image: 'ubuntu:22.04', + flavor: 'default', + org_id: 'org-uuid', + org_name: 'nexthop', + max_runners: 4, + tags: [{ id: 't1', name: 'self-hosted' }, { id: 't2', name: 'pool-label' }] + } +]; + +const mockInstances = [ + { name: 'r1', scale_set_id: 5, status: 'running', runner_status: 'active' }, + { name: 'r2', scale_set_id: 5, status: 'running', runner_status: 'idle' }, + { name: 'r5', scale_set_id: 5, status: 'running', runner_status: 'offline' }, + { name: 'r3', scale_set_id: 5, status: 'pending_create', runner_status: 'pending' }, + { name: 'r4', scale_set_id: 5, status: 'creating', runner_status: 'pending' }, + { name: 'p1', pool_id: 'aaaabbbb-cccc-dddd-eeee-ffff00001111', status: 'running', runner_status: 'active' } +]; + +const mockJobs = [ + // Two queued scale set jobs (ordering by created_at: job 2 queued first) + { + id: 1, + scaleset_job_id: 'ss-job-1', + scale_set_id: 5, + name: 'build', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:10:00Z' + }, + { + id: 2, + scaleset_job_id: 'ss-job-2', + scale_set_id: 5, + name: 'test', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-b', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:05:00Z' + }, + // A running scale set job + { + id: 3, + scaleset_job_id: 'ss-job-3', + scale_set_id: 5, + name: 'lint', + status: 'in_progress', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-micro'], + created_at: '2026-07-29T00:01:00Z' + }, + // A queued webhook job that label-matches the pool + { + id: 4, + workflow_job_id: 1234, + name: 'pool-job', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-c', + labels: ['self-hosted', 'pool-label'], + created_at: '2026-07-29T00:02:00Z' + }, + // A completed job that must not show up + { + id: 5, + scaleset_job_id: 'ss-job-5', + scale_set_id: 7, + name: 'done-job', + status: 'completed', + repository_owner: 'nexthop', + repository_name: 'repo-a', + labels: ['cloudstack-ubuntu24-large'], + created_at: '2026-07-29T00:00:00Z' + }, + // A queued job matching nothing + { + id: 6, + workflow_job_id: 5678, + name: 'orphan-job', + status: 'queued', + repository_owner: 'nexthop', + repository_name: 'repo-d', + labels: ['no-such-label'], + created_at: '2026-07-29T00:03:00Z' + } +]; + +describe('Queue Page - Integration Tests', () => { + beforeEach(async () => { + vi.clearAllMocks(); + + const cacheModule = await import('$lib/stores/eager-cache.js'); + vi.mocked(cacheModule.eagerCacheManager.getScaleSets).mockResolvedValue(mockScaleSets); + vi.mocked(cacheModule.eagerCacheManager.getPools).mockResolvedValue(mockPools); + vi.mocked(garmApi.listJobs).mockResolvedValue(mockJobs as any); + vi.mocked(garmApi.listInstances).mockResolvedValue(mockInstances as any); + }); + + it('renders page title', async () => { + render(QueuePage); + await waitFor(() => { + expect(screen.getByRole('heading', { name: 'Job Queue' })).toBeInTheDocument(); + }); + }); + + it('groups queued jobs per scale set with queue ordering and counts', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByRole('heading', { name: 'cloudstack-ubuntu24-micro' })).toBeInTheDocument(); + }); + + // Scale set group shows queued and running counts (incl. max runners) + expect(screen.getByText('2 queued')).toBeInTheDocument(); + // Runner counts come from instances (2 running: 1 busy / 1 idle; 2 provisioning) + expect( + screen.getByText(/3 runners \/ 100 max \(1 busy, 1 idle, 1 offline, 2 provisioning\)/) + ).toBeInTheDocument(); + + // GitHub's view from the scale set statistics + expect(screen.getByText(/GitHub: 60 assigned \(40 busy, 2 idle\)/)).toBeInTheDocument(); + expect( + screen.getByRole('heading', { name: 'cloudstack-ubuntu24-micro' }).querySelector('a') + ).toHaveAttribute('href', '/scalesets/5'); + + // Queue order: 'test' (older) before 'build' + const rows = screen.getAllByRole('row'); + const rowText = rows.map((r) => r.textContent || ''); + const testIdx = rowText.findIndex((t) => t.includes('test')); + const buildIdx = rowText.findIndex((t) => t.includes('build')); + expect(testIdx).toBeGreaterThan(0); + expect(buildIdx).toBeGreaterThan(testIdx); + + // Completed jobs are not shown + expect(screen.queryByText('done-job')).not.toBeInTheDocument(); + }); + + it('matches webhook jobs to pools by labels', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('pool-job')).toBeInTheDocument(); + }); + expect(screen.getByText(/Pool aaaabbbb/)).toBeInTheDocument(); + }); + + it('shows unattributed jobs in their own group', async () => { + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('Unattributed jobs')).toBeInTheDocument(); + }); + expect(screen.getByText('orphan-job')).toBeInTheDocument(); + }); + + it('shows empty state when there are no active jobs', async () => { + vi.mocked(garmApi.listJobs).mockResolvedValue([]); + vi.mocked(garmApi.listInstances).mockResolvedValue([]); + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('No queued or running jobs.')).toBeInTheDocument(); + }); + }); +}); + +describe('Queue Page - job links', () => { + beforeEach(async () => { + vi.clearAllMocks(); + const cacheModule = await import('$lib/stores/eager-cache.js'); + vi.mocked(cacheModule.eagerCacheManager.getScaleSets).mockResolvedValue(mockScaleSets); + vi.mocked(cacheModule.eagerCacheManager.getPools).mockResolvedValue(mockPools); + vi.mocked(garmApi.listInstances).mockResolvedValue([]); + }); + + it('deep-links webhook jobs to the GitHub job page and scale set jobs to the run page', async () => { + vi.mocked(garmApi.listJobs).mockResolvedValue([ + { + id: 10, + workflow_job_id: 90673415740, + run_id: 30480594764, + name: 'pool-job', + status: 'queued', + repository_owner: 'nexthop-ai', + repository_name: 'nhsystem-ops', + labels: ['self-hosted', 'pool-label'], + created_at: '2026-07-29T00:02:00Z' + }, + { + id: 11, + scaleset_job_id: 'guid-1', + scale_set_id: 5, + run_id: 1234, + workflow_run_url: 'https://github.com/nexthop-ai/repo-a/actions/runs/1234', + name: 'ss-job', + status: 'queued', + repository_owner: 'nexthop-ai', + repository_name: 'repo-a', + labels: [], + created_at: '2026-07-29T00:02:00Z' + } + ] as any); + + render(QueuePage); + + await waitFor(() => { + expect(screen.getByText('pool-job')).toBeInTheDocument(); + }); + expect(screen.getByText('pool-job').closest('a')).toHaveAttribute( + 'href', + 'https://github.com/nexthop-ai/nhsystem-ops/actions/runs/30480594764/job/90673415740' + ); + expect(screen.getByText('ss-job').closest('a')).toHaveAttribute( + 'href', + 'https://github.com/nexthop-ai/repo-a/actions/runs/1234' + ); + }); +}); diff --git a/webapp/swagger.yaml b/webapp/swagger.yaml index 6a13ec8ce..b9456d619 100644 --- a/webapp/swagger.yaml +++ b/webapp/swagger.yaml @@ -1515,6 +1515,13 @@ definitions: runner_name: type: string x-go-name: RunnerName + scale_set_id: + description: |- + ScaleSetID is the garm ID of the scale set this job was assigned to, + if it came in through a scale set listener. + format: uint64 + type: integer + x-go-name: ScaleSetID scaleset_job_id: description: ScaleSetJobID is the job ID when generated for a scale set. type: string @@ -1980,6 +1987,38 @@ definitions: x-go-name: Prefix type: object x-go-package: github.com/cloudbase/garm/params + RunnerScaleSetStatistic: + properties: + totalAcquiredJobs: + format: int64 + type: integer + x-go-name: TotalAcquiredJobs + totalAssignedJobs: + format: int64 + type: integer + x-go-name: TotalAssignedJobs + totalAvailableJobs: + format: int64 + type: integer + x-go-name: TotalAvailableJobs + totalBusyRunners: + format: int64 + type: integer + x-go-name: TotalBusyRunners + totalIdleRunners: + format: int64 + type: integer + x-go-name: TotalIdleRunners + totalRegisteredRunners: + format: int64 + type: integer + x-go-name: TotalRegisteredRunners + totalRunningJobs: + format: int64 + type: integer + x-go-name: TotalRunningJobs + type: object + x-go-package: github.com/cloudbase/garm/params RunnerStatus: type: string x-go-package: github.com/cloudbase/garm/params @@ -2108,6 +2147,8 @@ definitions: x-go-name: ScaleSetID state: $ref: '#/definitions/ScaleSetState' + statistics: + $ref: '#/definitions/RunnerScaleSetStatistic' status_messages: items: $ref: '#/definitions/StatusMessage' diff --git a/workers/scaleset/interfaces.go b/workers/scaleset/interfaces.go index b8acfceb9..5be303a27 100644 --- a/workers/scaleset/interfaces.go +++ b/workers/scaleset/interfaces.go @@ -22,7 +22,7 @@ type scaleSetHelper interface { GetScaleSet() params.ScaleSet GetScaleSetClient() (*scalesets.ScaleSetClient, error) SetLastMessageID(id int64) error - SetDesiredRunnerCount(count int) error + SetRunnerStatistics(stats params.RunnerScaleSetStatistic) error Owner() string HandleJobsCompleted(jobs []params.ScaleSetJobMessage) error HandleJobsStarted(jobs []params.ScaleSetJobMessage) error diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 418756c0d..8d80e2d19 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -457,7 +457,8 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error // Cross check what exists in github with what we have in the database. for name, runner := range ghRunnersByName { status := runner.GetStatus() - if _, ok := dbRunnersByName[name]; !ok { + dbRunner, ok := dbRunnersByName[name] + if !ok { // runner appears to be active. Is it not managed by GARM? if status != params.RunnerIdle && status != params.RunnerActive { slog.InfoContext(w.ctx, "runner does not exist in GARM; removing from github", "runner_name", name) @@ -470,6 +471,39 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error } continue } + + // Sync github's view of the runner (online idle/busy or offline) onto + // the instance. Without this, a runner whose agent died after setup + // stays "idle" in GARM forever, while github considers it offline and + // never assigns it jobs. Only touch runners that finished installing; + // runners in earlier lifecycle states are expected to be offline. + switch dbRunner.RunnerStatus { + case params.RunnerIdle, params.RunnerActive, params.RunnerOffline: + default: + continue + } + switch status { + case params.RunnerIdle, params.RunnerActive, params.RunnerOffline: + default: + continue + } + if dbRunner.RunnerStatus == status { + continue + } + if ok := locking.TryLock(name, w.consumerID); !ok { + slog.DebugContext(w.ctx, "runner is locked; skipping runner status sync", "runner_name", name) + continue + } + slog.InfoContext(w.ctx, "syncing runner status from github", "runner_name", name, "old_status", dbRunner.RunnerStatus, "new_status", status) + updatedRunner, err := w.store.UpdateInstance(w.ctx, name, params.UpdateInstanceParams{RunnerStatus: status}) + locking.Unlock(name, false) + if err != nil { + if !errors.Is(err, runnerErrors.ErrNotFound) { + slog.ErrorContext(w.ctx, "error updating runner status", "runner_name", name, "error", err) + } + continue + } + w.runners[updatedRunner.ID] = updatedRunner } unlockFn, err := w.reapTimedOutRunners(ghRunnersByName) diff --git a/workers/scaleset/scaleset_helper.go b/workers/scaleset/scaleset_helper.go index ea74abb74..c27f017b4 100644 --- a/workers/scaleset/scaleset_helper.go +++ b/workers/scaleset/scaleset_helper.go @@ -74,6 +74,7 @@ func (w *Worker) recordOrUpdateJob(job params.ScaleSetJobMessage) error { baseURL := strings.TrimRight(w.entity.Credentials.BaseURL, "/") jobParams := job.ToJob() jobParams.RunnerGroupName = w.scaleSet.GitHubRunnerGroup + jobParams.ScaleSetID = w.scaleSet.ID switch entity.EntityType { case params.ForgeEntityTypeEnterprise: @@ -227,9 +228,9 @@ func (w *Worker) HandleJobsAvailable(jobs []params.ScaleSetJobMessage) error { return nil } -func (w *Worker) SetDesiredRunnerCount(count int) error { - if err := w.store.SetScaleSetDesiredRunnerCount(w.ctx, w.scaleSet.ID, count); err != nil { - return fmt.Errorf("setting desired runner count: %w", err) +func (w *Worker) SetRunnerStatistics(stats params.RunnerScaleSetStatistic) error { + if err := w.store.SetScaleSetRunnerStatistics(w.ctx, w.scaleSet.ID, stats); err != nil { + return fmt.Errorf("setting runner statistics: %w", err) } return nil } diff --git a/workers/scaleset/scaleset_listener.go b/workers/scaleset/scaleset_listener.go index 3fc00e0da..e8325707d 100644 --- a/workers/scaleset/scaleset_listener.go +++ b/workers/scaleset/scaleset_listener.go @@ -228,8 +228,10 @@ func (l *scaleSetListener) handleSessionMessage(msg params.RunnerScaleSetMessage l.lastMessageID = msg.MessageID } - if err := l.scaleSetHelper.SetDesiredRunnerCount(msg.Statistics.TotalAssignedJobs); err != nil { - slog.ErrorContext(l.ctx, "setting desired runner count", "error", err) + if msg.Statistics != nil { + if err := l.scaleSetHelper.SetRunnerStatistics(*msg.Statistics); err != nil { + slog.ErrorContext(l.ctx, "setting runner statistics", "error", err) + } } if err := l.messageSession.DeleteMessage(l.listenerCtx, msg.MessageID); err != nil { From 54148ccef768d739bfbfa86b6949b06eacd1f0fe Mon Sep 17 00:00:00 2001 From: Benoit Sigoure Date: Fri, 7 Aug 2026 06:09:27 +0000 Subject: [PATCH 2/2] scaleset: reap runners stuck offline during GitHub Actions outages During the 2026-08-06 GitHub Actions outage, hundreds of scale set runners ended up stranded in running/offline: the agent registered, then died or lost its connection to the actions service. GARM only tracked GitHub's offline status as metadata and never recycled such runners, pinning every scale set at max_runners and starving the fleet. Changes: - Track when each runner is first observed in running/offline (in memory on the scale set worker; there is no status-changed timestamp in the DB and UpdatedAt is bumped by every write) and recycle it after DefaultRunnerOfflineTimeout (10 minutes) plus a deterministic per-runner jitter (fnv hash of the name, up to 5 minutes). The jitter prevents runners that went offline together from being reaped, respawned and going offline together in ever more synchronized batches. Offline status is only refreshed by the periodic consolidation pass, so effective reap latency is 10-15+ minutes. - Cap offline reaps at max(5, max_runners/4) per consolidation pass. Provider delete/create operations fan out one goroutine per instance, and unbounded churn can overwhelm the provider API. - When GitHub refuses to deregister a runner (TaskAgentJobStillRunningException, seen when the broker lost a job assignment the runner never acquired), log the refusal and proceed with pending_delete instead of aborting. Destroying the instance is exactly what un-sticks GitHub's state: the agent disappears, GitHub fails the stuck job and releases the runner. Note: the github runners list is NOT a reliable busy indicator for scale set runners (it reports idle for runners actively executing a job), so runners GARM considers active are never touched based on it. With this, scale sets keep cycling stranded runners during an outage so fresh runners can pick up whatever jobs do go through, and the fleet recovers on its own once GitHub stabilizes. Battle-tested in production during the outage itself. Co-Authored-By: Claude Fable 5 --- util/appdefaults/appdefaults.go | 7 ++ workers/scaleset/scaleset.go | 135 +++++++++++++++++++++++++++----- 2 files changed, 121 insertions(+), 21 deletions(-) diff --git a/util/appdefaults/appdefaults.go b/util/appdefaults/appdefaults.go index faf93ad85..8697e2d95 100644 --- a/util/appdefaults/appdefaults.go +++ b/util/appdefaults/appdefaults.go @@ -25,6 +25,13 @@ const ( // of time and no new updates have been made to it's state, it will be removed. DefaultRunnerBootstrapTimeout = 20 + // DefaultRunnerOfflineTimeout is the amount of time a scale set runner may + // remain in the running/offline state before it is recycled. Runners end up + // in this state when the github agent dies or fails to (re)connect to the + // actions service (e.g. during a github outage). Recycling them frees up + // capacity so fresh runners can pick up jobs. + DefaultRunnerOfflineTimeout = 10 * time.Minute + // DefaultGithubURL is the default URL where Github or Github Enterprise can be accessed. DefaultGithubURL = "https://github.com" diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 8d80e2d19..818c5c24e 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -17,6 +17,7 @@ import ( "context" "errors" "fmt" + "hash/fnv" "log/slog" "strings" "sync" @@ -33,6 +34,7 @@ import ( "github.com/cloudbase/garm/params" "github.com/cloudbase/garm/runner/common" garmUtil "github.com/cloudbase/garm/util" + "github.com/cloudbase/garm/util/appdefaults" ) func NewWorker(ctx context.Context, store dbCommon.Store, scaleSet params.ScaleSet, provider common.Provider) (*Worker, error) { @@ -62,6 +64,7 @@ func NewWorker(ctx context.Context, store dbCommon.Store, scaleSet params.ScaleS scaleSet: scaleSet, entity: entity, runners: make(map[string]params.Instance), + offlineSince: make(map[string]time.Time), }, nil } @@ -75,6 +78,14 @@ type Worker struct { scaleSet params.ScaleSet entity params.ForgeEntity runners map[string]params.Instance + // offlineSince tracks when we first observed a runner in the + // running/offline state, keyed by runner name. Instances have no + // "runner status changed at" timestamp in the DB and UpdatedAt is + // bumped by every write, so we track this in memory. Entries are + // cleared when the runner leaves the offline state or is removed; + // a GARM restart resets the clock, which at worst delays reaping + // by one timeout period. Protected by mux. + offlineSince map[string]time.Time consumer dbCommon.Consumer @@ -363,7 +374,13 @@ func (w *Worker) removeRunnerFromGithubAndSetPendingDelete(runnerName string, ag } if err := scaleSetCli.RemoveRunner(w.ctx, agentID); err != nil { if !errors.Is(err, runnerErrors.ErrNotFound) { - return fmt.Errorf("removing runner %s: %w", runnerName, err) + // Github may refuse to deregister a runner it believes is running a + // job (TaskAgentJobStillRunningException) even when the agent never + // actually acquired it (seen during github outages). Destroying the + // instance is what un-sticks that state on github's side: the agent + // disappears, github fails the stuck job and releases the runner. + // So log and proceed with pending_delete instead of aborting. + slog.WarnContext(w.ctx, "github refused to remove runner; deleting instance anyway", "runner_name", runnerName, "error", err) } } instance, err := w.setRunnerDBStatus(runnerName, commonParams.InstancePendingDelete) @@ -380,8 +397,34 @@ func (w *Worker) removeRunnerFromGithubAndSetPendingDelete(runnerName string, ag return nil } +// offlineTimeoutJitter returns a deterministic per-runner offset in +// [0, DefaultRunnerOfflineTimeout/2) added to the offline timeout. Runners +// that go offline together (e.g. during a github outage) would otherwise be +// reaped together, respawned together and go offline together again, with the +// batches growing more synchronized every cycle and hammering the provider +// API. Hashing the name spreads each batch over a jitter window without +// keeping extra state. +func offlineTimeoutJitter(runnerName string) time.Duration { + h := fnv.New64a() + h.Write([]byte(runnerName)) + return time.Duration(h.Sum64() % uint64(appdefaults.DefaultRunnerOfflineTimeout/2)) +} + +// maxOfflineReapsPerPass bounds how many offline runners we recycle in one +// consolidation pass, so a mass-offline event (github outage) trickles +// delete/create calls to the provider over several passes instead of issuing +// them all at once. +func (w *Worker) maxOfflineReapsPerPass() int { + limit := int(w.scaleSet.MaxRunners) / 4 + if limit < 5 { + limit = 5 + } + return limit +} + func (w *Worker) reapTimedOutRunners(runners map[string]params.RunnerReference) (func(), error) { lockNames := []string{} + offlineReaps := 0 unlockFn := func() { for _, name := range lockNames { @@ -390,10 +433,9 @@ func (w *Worker) reapTimedOutRunners(runners map[string]params.RunnerReference) } } + currentNames := make(map[string]struct{}, len(w.runners)) for _, runner := range w.runners { - if time.Since(runner.CreatedAt).Minutes() < float64(w.scaleSet.RunnerTimeout()) { - continue - } + currentNames[runner.Name] = struct{}{} switch runner.Status { case commonParams.InstancePendingDelete, commonParams.InstancePendingForceDelete, commonParams.InstanceDeleting, commonParams.InstanceDeleted: @@ -411,30 +453,72 @@ func (w *Worker) reapTimedOutRunners(runners map[string]params.RunnerReference) continue } - if runner.RunnerStatus != params.RunnerPending && runner.RunnerStatus != params.RunnerInstalling && runner.RunnerStatus != params.RunnerFailed { - slog.DebugContext(w.ctx, "runner is not pending, installing or failed; skipping", "runner_name", runner.Name) - continue + // A runner in running/offline is one whose agent registered with github + // but then died or lost its connection to the actions service (this + // happens en masse during github outages). It will never receive jobs + // again, but it occupies a max_runners slot. Track when we first saw it + // offline and recycle it once it exceeds the offline timeout. + offlineTimedOut := false + if runner.Status == commonParams.InstanceRunning && runner.RunnerStatus == params.RunnerOffline { + since, seen := w.offlineSince[runner.Name] + if !seen { + w.offlineSince[runner.Name] = time.Now() + } else { + timeout := appdefaults.DefaultRunnerOfflineTimeout + offlineTimeoutJitter(runner.Name) + offlineTimedOut = time.Since(since) >= timeout && offlineReaps < w.maxOfflineReapsPerPass() + } + } else { + delete(w.offlineSince, runner.Name) } - if ghRunner, ok := runners[runner.Name]; !ok || ghRunner.GetStatus() == params.RunnerOffline { - if ok := locking.TryLock(runner.Name, w.consumerID); !ok { - slog.DebugContext(w.ctx, "runner is locked; skipping", "runner_name", runner.Name) - continue + + bootstrapTimedOut := false + if time.Since(runner.CreatedAt).Minutes() >= float64(w.scaleSet.RunnerTimeout()) { + if runner.RunnerStatus == params.RunnerPending || runner.RunnerStatus == params.RunnerInstalling || runner.RunnerStatus == params.RunnerFailed { + ghRunner, ok := runners[runner.Name] + bootstrapTimedOut = !ok || ghRunner.GetStatus() == params.RunnerOffline } + } + + if !offlineTimedOut && !bootstrapTimedOut { + continue + } + + if ok := locking.TryLock(runner.Name, w.consumerID); !ok { + slog.DebugContext(w.ctx, "runner is locked; skipping", "runner_name", runner.Name) + continue + } + if offlineTimedOut { + slog.InfoContext( + w.ctx, "reaping runner offline for too long", + "runner_name", runner.Name, + "offline_for", time.Since(w.offlineSince[runner.Name]).String()) + } else { slog.InfoContext( w.ctx, "reaping timed-out/failed runner", "runner_name", runner.Name) + } - if err := w.removeRunnerFromGithubAndSetPendingDelete(runner.Name, runner.AgentID); err != nil { - // Don't let a single poisoned runner (e.g. one that raced into a - // status that can no longer transition to pending_delete through - // some other codepath) abort reaping for the rest of the batch. - // Log it, release just this runner's lock, and keep going. - slog.ErrorContext(w.ctx, "error removing runner", "runner_name", runner.Name, "error", err) - locking.Unlock(runner.Name, false) - continue - } - lockNames = append(lockNames, runner.Name) + if err := w.removeRunnerFromGithubAndSetPendingDelete(runner.Name, runner.AgentID); err != nil { + // Don't let a single poisoned runner (e.g. one that raced into a + // status that can no longer transition to pending_delete through + // some other codepath) abort reaping for the rest of the batch. + // Log it, release just this runner's lock, and keep going. + slog.ErrorContext(w.ctx, "error removing runner", "runner_name", runner.Name, "error", err) + locking.Unlock(runner.Name, false) + continue + } + if offlineTimedOut { + offlineReaps++ + } + delete(w.offlineSince, runner.Name) + lockNames = append(lockNames, runner.Name) + } + + // Prune tracking entries for runners that no longer exist. + for name := range w.offlineSince { + if _, ok := currentNames[name]; !ok { + delete(w.offlineSince, name) } } return unlockFn, nil @@ -490,6 +574,15 @@ func (w *Worker) consolidateRunnerState(runners []params.RunnerReference) error if dbRunner.RunnerStatus == status { continue } + if dbRunner.RunnerStatus == params.RunnerActive && status == params.RunnerIdle { + // The github runners list is NOT a reliable busy indicator for + // scale set runners: it reports "idle" for runners that are + // actively executing a job (observed 2026-08-07: reaping on this + // signal cancelled ~150 running jobs). Never touch active runners + // based on the list; job completion/failure is handled by the + // listener and the offline reaper covers dead agents. + continue + } if ok := locking.TryLock(name, w.consumerID); !ok { slog.DebugContext(w.ctx, "runner is locked; skipping runner status sync", "runner_name", name) continue