Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 39 additions & 6 deletions docs/arch/webhook-job-lifecycle.md
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,9 @@ N_B exits -> reprovisionIfStranded(B): started[B] set -> no-op
(e.g. a superseded workflow run GitHub keeps listing as queued but never
dispatches) would otherwise re-provision on every runner exit forever.
- **draining** — shutting down.
- **the platform says the job no longer needs a runner** — asked via
`JobStateReporter` after the guards above pass, before re-queuing. See
[Stale dispatch decisions](#stale-dispatch-decisions).

Re-dispatch is launched as `go s.handleQueued(...)` so it never blocks on the
concurrency slot the exiting wait-goroutine is about to release, and it uses
Expand Down Expand Up @@ -150,6 +153,34 @@ original claim-retry path, which would misroute a re-claim failure.
after dispatch. Pruning `started[X]` earlier would let that late exit falsely
re-provision the already-run X.

## Stale dispatch decisions

`started` is in memory and expires; a dispatch decision can be far older. On
2026-10-05 the Windows node (2 slots) was admitting dispatches 24h and 28h into
their slot wait, for jobs GitHub had cancelled three days earlier. `started`
had long forgotten them, so `admitDispatch` admitted each one; the runner idled
out the 90-minute orphan grace; `reprovisionIfStranded` saw `started` unset and
re-queued it; repeat. ~23 dead runners a day against ~10 real jobs, and live
jobs queued for hours behind them. Three changes close it:

- **Ask the platform.** Providers implementing `JobStateReporter`
(`JobAwaitingRunner`: GitHub reads `GET /repos/{o}/{r}/actions/jobs/{id}`)
are consulted by `admitDispatch` when the dispatch *waited* for its slot,
and by `reprovisionIfStranded` before every re-queue. A "gone" answer
(in_progress, completed, or 404) abandons the dispatch and is recorded in
`started`. It **fails open**: an API error provisions as before, because a
stranded live job costs more than one wasted runner. A dispatch that found a
free slot at once skips the call — its webhook is seconds old.
- **Bound the wait.** `acquireSlot` gives up after `maxSlotWait` (6h). The
job loses nothing: if it is still queued, the reconcile poll or its next
`queued` event dispatches it fresh.
- **Keep the zombie counter while the job is held.** `cleanSeen` used to reset
`attempts[key]` whenever `seen[key]` aged past `seenTTL` (10m), including
for jobs still pending on a slot or running. Every strand cycle is longer
than 10 minutes, so the counter was zero at each re-provision and
`maxProvisionAttempts` never fired. It now survives while the key is in
`pending` or `running`.

## The reconcile poll is now a backstop only

The periodic catch-up poll (`runReconcileLoop`, `[webhook] reconcile_interval`)
Expand Down Expand Up @@ -221,12 +252,14 @@ sorted, deduped) into a **fungibility class**. `Scheduler.jobLabels` records the
class of every accepted job; `runnerBinding.labelSet` records the class each
dispatched runner serves.

1. **`admitDispatch(key)`** — the last gate before claiming, called by all four
provisioning paths immediately after they acquire their concurrency slot.
It returns false when the job was observed running while the handler sat
blocked, so the dispatch is *discharged by the sibling's execution* instead
of becoming an orphan. The caller releases its slot, keeping
`max_concurrent` accounting correct.
1. **`admitDispatch(ctx, event, waited)`** — the last gate before claiming,
called by all provisioning paths immediately after they acquire their
concurrency slot. It abandons the dispatch when the job was observed running
while the handler sat blocked, so the dispatch is *discharged by the
sibling's execution* instead of becoming an orphan — or, when the handler
waited, when the platform reports the job no longer needs a runner (see
[Stale dispatch decisions](#stale-dispatch-decisions)). The caller releases
its slot, keeping `max_concurrent` accounting correct.

2. **Label-set reconciliation in `sweepOrphanRunners`** — an unbound runner
whose intent job ran elsewhere is a *spare*. Spares are allocated against
Expand Down
32 changes: 32 additions & 0 deletions pkg/github/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,38 @@ func (c *Client) RunnerBusy(ctx context.Context, repo string, runnerID int64) (b
return runner.GetBusy(), nil
}

// JobAwaitingRunner reports whether a workflow job is still waiting for a
// runner, from GitHub's own view of the job.
//
// Anything short of in_progress or completed counts as waiting: queued is the
// common case, and waiting/pending/requested (environment approvals, run
// concurrency groups) are jobs that will still need a runner once released.
// A job that no longer exists (404 — its run was deleted) does not need one.
// Every other error is returned, and the caller must treat it as unknown.
//
// The jobs endpoint is repo-scoped whether runners are registered at the org
// or the repo level, so there is no org-level variant to choose between.
func (c *Client) JobAwaitingRunner(ctx context.Context, repo string, jobID int64) (bool, error) {
job, resp, err := c.client.Actions.GetWorkflowJobByID(ctx, c.cfg.Owner, repo, jobID)
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return false, nil
}
return false, fmt.Errorf("reading job %d: %w", jobID, err)
}
return jobAwaitsRunner(job.GetStatus()), nil
}

// jobAwaitsRunner maps a workflow job status onto "still needs a runner".
// Split out so the mapping is testable without an API server.
func jobAwaitsRunner(status string) bool {
switch status {
case "in_progress", "completed":
return false
}
return true
}

// FetchJobImage fetches the workflow run's job definition and reads the
// container image declared in the job's `container:` field. This requires an
// extra API call per job but lets users specify the image directly in their
Expand Down
80 changes: 80 additions & 0 deletions pkg/github/jobawaiting_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
package github

import (
"context"
"encoding/json"
"net/http"
"testing"
)

// TestJobAwaitingRunner pins the status mapping the scheduler relies on to
// drop stale dispatches. Getting a live status wrong strands a real job (a
// "gone" answer is never second-guessed), so every status that can still end
// up needing a runner must read as awaiting.
func TestJobAwaitingRunner(t *testing.T) {
tests := []struct {
status string
want bool
}{
{"queued", true},
{"waiting", true}, // environment approval: will need a runner once approved
{"pending", true}, // concurrency group: ditto
{"requested", true}, // ditto
{"in_progress", false},
{"completed", false},
}
for _, tt := range tests {
t.Run(tt.status, func(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/repos/testorg/repo1/actions/jobs/77", func(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(map[string]any{"id": 77, "status": tt.status}); err != nil {
t.Logf("encoding: %v", err)
}
})
c, srv := newTestClientWithServer(t, mux)
defer srv.Close()

got, err := c.JobAwaitingRunner(context.Background(), "repo1", 77)
if err != nil {
t.Fatalf("JobAwaitingRunner: %v", err)
}
if got != tt.want {
t.Errorf("JobAwaitingRunner(status=%q) = %v, want %v", tt.status, got, tt.want)
}
})
}
}

// A job whose run was deleted 404s. It cannot need a runner, and reporting it
// as an error would make the caller fail open and boot one for it anyway.
func TestJobAwaitingRunner_NotFoundIsGone(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/repos/testorg/repo1/actions/jobs/77", func(w http.ResponseWriter, r *http.Request) {
http.Error(w, `{"message":"Not Found"}`, http.StatusNotFound)
})
c, srv := newTestClientWithServer(t, mux)
defer srv.Close()

got, err := c.JobAwaitingRunner(context.Background(), "repo1", 77)
if err != nil {
t.Fatalf("JobAwaitingRunner on 404: %v, want nil error", err)
}
if got {
t.Error("JobAwaitingRunner on 404 = true, want false")
}
}

// Any other failure must surface as an error, so the scheduler can fail open
// rather than mistake an API outage for "the job is gone".
func TestJobAwaitingRunner_ServerErrorIsUnknown(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/repos/testorg/repo1/actions/jobs/77", func(w http.ResponseWriter, r *http.Request) {
http.Error(w, `{"message":"boom"}`, http.StatusInternalServerError)
})
c, srv := newTestClientWithServer(t, mux)
defer srv.Close()

if _, err := c.JobAwaitingRunner(context.Background(), "repo1", 77); err == nil {
t.Error("JobAwaitingRunner on 500 returned nil error; an outage would read as a definite answer")
}
}
11 changes: 11 additions & 0 deletions pkg/providers/github/github.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ var (
_ providers.Webhook = (*Provider)(nil)
_ providers.RunnerNameReporter = (*Provider)(nil)
_ providers.RunnerBusyReporter = (*Provider)(nil)
_ providers.JobStateReporter = (*Provider)(nil)
)

// New creates a GitHub provider wrapping an existing GitHub client.
Expand Down Expand Up @@ -165,6 +166,16 @@ func (p *Provider) RunnerBusy(ctx context.Context, claim *providers.Claim) (bool
return p.client.RunnerBusy(ctx, claim.Repo, claim.RunnerID)
}

// JobAwaitingRunner implements providers.JobStateReporter. One GET per stale
// dispatch decision — a dispatch that waited for a slot, or a re-provision —
// never on the fast path of a job that found a free slot immediately.
func (p *Provider) JobAwaitingRunner(ctx context.Context, event *providers.JobEvent) (bool, error) {
if event == nil {
return false, fmt.Errorf("nil event")
}
return p.client.JobAwaitingRunner(ctx, event.Repo, event.JobID)
}

func (p *Provider) FetchJobImage(ctx context.Context, event *providers.JobEvent) string {
return p.client.FetchJobImage(ctx, event.Repo, event.RunID, event.JobID)
}
Expand Down
22 changes: 22 additions & 0 deletions pkg/providers/provider.go
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,28 @@ type RunnerBusyReporter interface {
RunnerBusy(ctx context.Context, claim *Claim) (bool, error)
}

// JobStateReporter is optionally implemented by providers that can be asked,
// on demand, whether a job still needs a runner.
//
// It is the authority the scheduler consults before turning a STALE dispatch
// decision into a runner: one that sat blocked on a concurrency slot, or one
// being re-provisioned after its runner exited unassigned. Both decisions were
// taken from a webhook that may be hours or days old, and the scheduler's own
// record of the job finishing (the started map) is in memory and expires. On
// 2026-10-05 the Windows node was booting runners for jobs that had been
// cancelled three days earlier, each one idling for the full orphan grace
// window while live jobs queued behind it.
type JobStateReporter interface {
Provider

// JobAwaitingRunner reports whether the job is still waiting for a runner
// to pick it up. false means it is running elsewhere, finished, or gone.
// An error means "could not determine" — callers must fail OPEN and
// provision anyway, because a skipped live job is worse than one wasted
// runner.
JobAwaitingRunner(ctx context.Context, event *JobEvent) (bool, error)
}

// PollConfig provides settings for poll-based job discovery.
type PollConfig struct {
PollInterval int // seconds between polls (0 = provider default)
Expand Down
11 changes: 6 additions & 5 deletions pkg/scheduler/cordon_gate_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -315,19 +315,20 @@ func TestUncordon_RestoresDispatchGate(t *testing.T) {
s := New(Config{Log: quietLogger()})
s.bindContexts(context.Background())
s.webhookMode = true
key := jobKey{Provider: "github", JobID: 601}
event := providers.JobEvent{Provider: newMockProvider("github"), Repo: "myrepo", JobID: 601}
key := keyFor(event)

if got := s.admitDispatch(key); got != dispatchAdmit {
if got := s.admitDispatch(context.Background(), event, 0); got != dispatchAdmit {
t.Fatalf("admitDispatch on a healthy scheduler = %v, want dispatchAdmit", got)
}

s.Cordon()
if got := s.admitDispatch(key); got != dispatchAbandonCordoned {
if got := s.admitDispatch(context.Background(), event, 0); got != dispatchAbandonCordoned {
t.Errorf("admitDispatch while cordoned = %v, want dispatchAbandonCordoned", got)
}

s.Uncordon()
if got := s.admitDispatch(key); got != dispatchAdmit {
if got := s.admitDispatch(context.Background(), event, 0); got != dispatchAdmit {
t.Errorf("admitDispatch after Uncordon = %v, want dispatchAdmit; claiming was not restored", got)
}

Expand All @@ -337,7 +338,7 @@ func TestUncordon_RestoresDispatchGate(t *testing.T) {
s.mu.Lock()
s.started[key] = time.Now()
s.mu.Unlock()
if got := s.admitDispatch(key); got != dispatchAbandonSatisfied {
if got := s.admitDispatch(context.Background(), event, 0); got != dispatchAbandonSatisfied {
t.Errorf("admitDispatch for an already-satisfied job = %v, want dispatchAbandonSatisfied", got)
}
}
Expand Down
Loading