diff --git a/docs/arch/webhook-job-lifecycle.md b/docs/arch/webhook-job-lifecycle.md index 0ff5ef9..563a21d 100644 --- a/docs/arch/webhook-job-lifecycle.md +++ b/docs/arch/webhook-job-lifecycle.md @@ -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 @@ -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`) @@ -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 diff --git a/pkg/github/client.go b/pkg/github/client.go index 820e681..bf64371 100644 --- a/pkg/github/client.go +++ b/pkg/github/client.go @@ -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 diff --git a/pkg/github/jobawaiting_test.go b/pkg/github/jobawaiting_test.go new file mode 100644 index 0000000..2ff53e3 --- /dev/null +++ b/pkg/github/jobawaiting_test.go @@ -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") + } +} diff --git a/pkg/providers/github/github.go b/pkg/providers/github/github.go index 601f956..367729a 100644 --- a/pkg/providers/github/github.go +++ b/pkg/providers/github/github.go @@ -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. @@ -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) } diff --git a/pkg/providers/provider.go b/pkg/providers/provider.go index 4886614..3d32c45 100644 --- a/pkg/providers/provider.go +++ b/pkg/providers/provider.go @@ -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) diff --git a/pkg/scheduler/cordon_gate_test.go b/pkg/scheduler/cordon_gate_test.go index ede75ca..d396e6d 100644 --- a/pkg/scheduler/cordon_gate_test.go +++ b/pkg/scheduler/cordon_gate_test.go @@ -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) } @@ -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) } } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 0bac88e..c0b1a5a 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -1118,6 +1118,11 @@ const ( // dispatchAbandonCordoned: the node was cordoned while this dispatch // waited for a concurrency slot. dispatchAbandonCordoned + // dispatchAbandonGone: the platform says the job no longer needs a runner + // — it finished, was cancelled, or is running elsewhere — and this node + // never saw the event that would have said so (or saw it so long ago that + // the record expired). + dispatchAbandonGone ) // log emits the verdict's explanation on the caller's job-scoped logger, so @@ -1130,6 +1135,9 @@ func (v dispatchVerdict) log(log *slog.Logger) { case dispatchAbandonCordoned: log.Info("abandoning dispatch: scheduler was cordoned while this dispatch waited for a concurrency slot", "detail", "cordon means stop claiming NEW work; the job stays queued for another node or for this node after uncordon") + case dispatchAbandonGone: + log.Info("abandoning dispatch: the platform reports the job no longer needs a runner", + "detail", "it finished, was cancelled, or ran elsewhere while this dispatch waited for a concurrency slot") } } @@ -1172,23 +1180,71 @@ func (v dispatchVerdict) log(log *slog.Logger) { // // Jobs ALREADY RUNNING are unaffected — this gate is only ever consulted on the // path to a new claim, never on a running job's lifecycle. -func (s *Scheduler) admitDispatch(key jobKey) dispatchVerdict { +// +// 3. Staleness (dispatchAbandonGone). Check 1 trusts the started map, which is +// in memory and pruned after roughly JobTimeout. A slot wait had no bound, +// so a dispatch could outlive that record by days: on 2026-10-05 the +// Windows node admitted dispatches 24h and 28h old, for jobs cancelled on +// 10-02, and every one booted a runner that idled out the 90-minute orphan +// grace while live jobs queued behind it. So when the dispatch actually +// WAITED (waited > 0), ask the platform. A dispatch that found a free slot +// at once skips the call: its webhook is seconds old. +func (s *Scheduler) admitDispatch(ctx context.Context, event providers.JobEvent, waited time.Duration) dispatchVerdict { + key := keyFor(event) s.mu.Lock() - defer s.mu.Unlock() delete(s.pending, key) if s.draining { + s.mu.Unlock() return dispatchAbandonCordoned } - if !s.webhookMode { + if _, done := s.started[key]; done && s.webhookMode { // started is only populated from in_progress/completed webhooks. - return dispatchAdmit - } - if _, done := s.started[key]; done { + s.mu.Unlock() return dispatchAbandonSatisfied } + s.mu.Unlock() + + if waited > 0 && !s.jobStillWanted(ctx, event) { + return dispatchAbandonGone + } return dispatchAdmit } +// jobStillWantedTimeout bounds the platform lookup so a slow API cannot stall +// a dispatch that has a slot in hand. +const jobStillWantedTimeout = 15 * time.Second + +// jobStillWanted asks the job's provider whether the job still needs a runner. +// +// It fails OPEN: a provider that cannot answer, or an API error, returns true +// and the dispatch proceeds exactly as it did before this check existed. The +// asymmetry is deliberate — a wrong "gone" strands a live job, while a wrong +// "wanted" costs one runner that the orphan sweep reclaims. +// +// A "gone" answer is recorded in started, the same satisfied signal a +// completed webhook leaves, so the re-provision path and later gates see it +// without asking again. +func (s *Scheduler) jobStillWanted(ctx context.Context, event providers.JobEvent) bool { + reporter, ok := event.Provider.(providers.JobStateReporter) + if !ok { + return true + } + ctx, cancel := context.WithTimeout(ctx, jobStillWantedTimeout) + defer cancel() + wanted, err := reporter.JobAwaitingRunner(ctx, &event) + if err != nil { + s.cfg.Log.Warn("could not confirm the job still needs a runner; provisioning anyway", + "job_id", event.JobID, "repo", event.Repo, "error", err) + return true + } + if !wanted { + s.mu.Lock() + s.started[keyFor(event)] = time.Now() + s.mu.Unlock() + } + return wanted +} + // handleLinuxJob dispatches a Linux job to the Linux VM worker via gRPC. // The host registers the JIT runner (with Linux labels) and sends // Create/Wait/Destroy RPCs to the dispatch server running inside the VM @@ -1222,7 +1278,7 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent } }() - if v := s.admitDispatch(key); v != dispatchAdmit { + if v := s.admitDispatch(ctx, event, slot.waited); v != dispatchAdmit { v.log(log) slot.release() return @@ -1421,7 +1477,7 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent } }() - if v := s.admitDispatch(key); v != dispatchAdmit { + if v := s.admitDispatch(ctx, event, slot.waited); v != dispatchAdmit { v.log(log) slot.release() return @@ -1850,7 +1906,7 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent } }() - if v := s.admitDispatch(key); v != dispatchAdmit { + if v := s.admitDispatch(ctx, event, slot.waited); v != dispatchAdmit { v.log(log) slot.release() return @@ -2369,6 +2425,20 @@ func (s *Scheduler) cleanSeen() { for id, t := range s.seen { if time.Since(t) > seenTTL { delete(s.seen, id) + // A job this node is still holding — blocked on a slot (pending) + // or with a runner up (running) — has NOT left the queue; its seen + // stamp merely aged. Resetting its zombie counter here is what + // made maxProvisionAttempts unreachable: every strand cycle lasts + // longer than seenTTL (a 90-minute orphan grace, or hours on a + // slot), so the counter was back to zero by each re-provision and + // cancelled jobs looped for days. Its labels stay too — a waiting + // job is still demand for its fungibility class. + if _, ok := s.pending[id]; ok { + continue + } + if _, ok := s.running[id]; ok { + continue + } // Job stopped appearing in the queue (finished/cancelled) — reset // its zombie counter so a future legitimate rerun starts fresh. delete(s.attempts, id) @@ -2853,11 +2923,25 @@ func (s *Scheduler) reprovisionIfStranded(ctx context.Context, event providers.J dispatchCtx = ctx } - s.cfg.Log.Info("dispatched runner exited but its job was never observed running; re-provisioning", - "job_id", key.JobID, - "repo", event.Repo, - "detail", "same-label JIT runners are fungible; the runner likely ran a sibling job and left this one queued") - go s.handleQueued(dispatchCtx, event) + go func() { + // Ask before re-queuing. "Never observed running" is a statement about + // this node's memory, not about the job: the completed webhook may have + // been missed, or recorded so long ago that started has forgotten it. + // Without this check the loop below re-provisioned cancelled jobs for + // days — runner boots, idles out the orphan grace, exits unassigned, + // re-provision — and the attempt cap never fired (see cleanSeen). + if !s.jobStillWanted(dispatchCtx, event) { + s.cfg.Log.Info("dispatched runner exited unassigned, and the platform reports the job no longer needs a runner; not re-provisioning", + "job_id", key.JobID, + "repo", event.Repo) + return + } + s.cfg.Log.Info("dispatched runner exited but its job was never observed running; re-provisioning", + "job_id", key.JobID, + "repo", event.Repo, + "detail", "same-label JIT runners are fungible; the runner likely ran a sibling job and left this one queued") + s.handleQueued(dispatchCtx, event) + }() } // runReconcileLoop periodically re-runs each provider's catch-up poll while in diff --git a/pkg/scheduler/slots.go b/pkg/scheduler/slots.go index 23a507b..b53f86a 100644 --- a/pkg/scheduler/slots.go +++ b/pkg/scheduler/slots.go @@ -61,6 +61,17 @@ var ( // so waiting this long against a pool that nothing is tracked against is // reported at Error. slotLeakSuspectAfter = 15 * time.Minute + + // maxSlotWait bounds how long a dispatch may sit blocked on a full pool + // before it gives up. Without it a wait had no end at all: on 2026-10-05 + // the Windows node had dispatches 24h and 28h into their wait, for jobs + // GitHub had cancelled days before, and each one that won a slot booted a + // runner that idled out the full orphan grace window. Giving up loses + // nothing — a job that is still genuinely queued is rediscovered by the + // reconcile poll and the next queued event and dispatched fresh. Six hours + // is long enough to queue behind any real build (GitHub's own job limit) + // and short enough that the backlog cannot outlive the jobs in it. + maxSlotWait = 6 * time.Hour ) // slotToken is a one-shot handle to an acquired concurrency slot. @@ -72,6 +83,11 @@ var ( type slotToken struct { sem chan struct{} once sync.Once + + // waited is how long acquiring this slot blocked; zero on the fast path. + // admitDispatch uses it to decide whether the dispatch decision is stale + // enough to re-check with the platform before claiming a runner. + waited time.Duration } // release returns the slot to its pool. Safe to call any number of times from @@ -86,8 +102,9 @@ func (t *slotToken) release() { t.once.Do(func() { <-t.sem }) } -// acquireSlot takes a slot from sem, blocking until one is free or ctx is -// done. It returns nil when ctx ended first; the caller must not provision. +// acquireSlot takes a slot from sem, blocking until one is free, ctx is done, +// or maxSlotWait elapses. It returns nil when it did not get a slot; the caller +// must not provision. // // The only behavioural difference from the raw `select { case sem <- ... }` // it replaces is that a wait which actually BLOCKS becomes visible in the log. @@ -110,15 +127,24 @@ func (s *Scheduler) acquireSlot(ctx context.Context, sem chan struct{}, pool str start := time.Now() timer := time.NewTimer(slotWaitLogAfter) defer timer.Stop() + giveUp := time.NewTimer(maxSlotWait) + defer giveUp.Stop() for { select { case sem <- struct{}{}: + waited := time.Since(start) log.Info("acquired concurrency slot after waiting", - "pool", pool, "waited", time.Since(start).Truncate(time.Second), + "pool", pool, "waited", waited.Truncate(time.Second), "held", len(sem), "capacity", cap(sem)) - return &slotToken{sem: sem} + return &slotToken{sem: sem, waited: waited} case <-ctx.Done(): return nil + case <-giveUp.C: + log.Warn("abandoning dispatch: waited too long for a concurrency slot", + "pool", pool, "waited", time.Since(start).Truncate(time.Second), + "max_wait", maxSlotWait, + "detail", "if the job is still queued, the reconcile poll or its next queued event dispatches it fresh") + return nil case <-timer.C: // PER-POOL, not global. The first version of this compared the // pool's held count against s.ActiveJobs(), which is diff --git a/pkg/scheduler/stale_dispatch_test.go b/pkg/scheduler/stale_dispatch_test.go new file mode 100644 index 0000000..eb1ae25 --- /dev/null +++ b/pkg/scheduler/stale_dispatch_test.go @@ -0,0 +1,401 @@ +package scheduler + +// Tests for dropping STALE dispatch decisions. +// +// On 2026-10-05 the Windows node spent ~34 of its 48 slot-hours a day booting +// runners for jobs GitHub had cancelled up to three days earlier. Three things +// combined: +// +// - admitDispatch's satisfied check reads the in-memory started map, which +// forgets a finished job after ~JobTimeout, while a slot wait had no bound +// at all (dispatches were 24h and 28h into their wait); +// - reprovisionIfStranded re-queued any job not in started, so each idle +// runner's exit put the dead job straight back in line; +// - cleanSeen reset the zombie counter of jobs still pending or running, so +// maxProvisionAttempts never fired. +// +// The platform is now asked before a stale decision becomes a runner, waits +// are bounded, and the counter survives while the node still holds the job. + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +// jobStateProvider answers JobAwaitingRunner from a test-controlled value. +type jobStateProvider struct { + *reportingProvider + awaiting atomic.Bool + err error + calls atomic.Int32 +} + +func (p *jobStateProvider) JobAwaitingRunner(context.Context, *providers.JobEvent) (bool, error) { + p.calls.Add(1) + if p.err != nil { + return false, p.err + } + return p.awaiting.Load(), nil +} + +var _ providers.JobStateReporter = (*jobStateProvider)(nil) + +func newJobStateProvider(awaiting bool, err error) (*jobStateProvider, *claimCountingProvider) { + base := newClaimCountingProvider("github") + p := &jobStateProvider{reportingProvider: &reportingProvider{claimCountingProvider: base}, err: err} + p.awaiting.Store(awaiting) + return p, base +} + +// waitForStarted blocks until started contains key (or the deadline elapses). +func waitForStarted(t *testing.T, s *Scheduler, key jobKey) bool { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + s.mu.Lock() + _, ok := s.started[key] + s.mu.Unlock() + if ok { + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(5 * time.Millisecond) + } +} + +// TestAdmitDispatch_AsksPlatformAfterWaiting is the production shape: a job +// is blocked behind a full pool, finishes at GitHub while it waits, and this +// node holds no record of that (the completed webhook was missed, or started +// already forgot it). When the slot frees, the dispatch must ask and drop it. +func TestAdmitDispatch_AsksPlatformAfterWaiting(t *testing.T) { + tests := []struct { + name string + webhookMode bool + awaiting bool + apiErr error + wantClaims int32 + wantRunning bool + wantStarted bool + }{ + { + name: "job finished at the platform while we waited", + webhookMode: true, awaiting: false, + wantClaims: 0, wantRunning: false, wantStarted: true, + }, + { + name: "job still queued when the slot frees", + webhookMode: true, awaiting: true, + wantClaims: 1, wantRunning: true, + }, + { + // Fail open: an API outage must never strand a live job. + name: "platform cannot answer", + webhookMode: true, apiErr: errors.New("api down"), + wantClaims: 1, wantRunning: true, + }, + { + // Unlike the started map, the platform's answer is valid in poll + // mode too. + name: "poll mode also drops a finished job", + webhookMode: false, awaiting: false, + wantClaims: 0, wantRunning: false, wantStarted: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + defer close(fake.waitBlock) + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + prov, base := newJobStateProvider(tt.awaiting, tt.apiErr) + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 1, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + s.webhookMode = tt.webhookMode + + const jobID int64 = 910 + key := jobKey{Provider: prov.Name(), JobID: jobID} + + s.linuxSem <- struct{}{} // the pool is full + done := make(chan struct{}) + go func() { + defer close(done) + s.handleQueued(context.Background(), linuxQueuedEvent(prov, jobID)) + }() + if !waitForPending(t, s, key) { + t.Fatal("handler never reached the semaphore wait") + } + // pending is set just BEFORE acquireSlot. Freeing the slot that + // instant can let the handler take the fast path, which correctly + // skips the platform check — so give it time to actually block. + time.Sleep(50 * time.Millisecond) + <-s.linuxSem + <-done + + if got := prov.calls.Load(); got != 1 { + t.Errorf("platform consulted %d time(s), want 1", got) + } + if got := base.claims.Load(); got != tt.wantClaims { + t.Errorf("claims = %d, want %d", got, tt.wantClaims) + } + s.mu.Lock() + _, running := s.running[key] + _, pending := s.pending[key] + _, started := s.started[key] + s.mu.Unlock() + if running != tt.wantRunning { + t.Errorf("running = %v, want %v", running, tt.wantRunning) + } + if pending { + t.Error("pending was not cleared") + } + if started != tt.wantStarted { + t.Errorf("started = %v, want %v (a gone verdict is recorded so later gates need not ask again)", started, tt.wantStarted) + } + if !tt.wantRunning && len(s.linuxSem) != 0 { + t.Errorf("abandoned dispatch leaked its slot: linuxSem holds %d", len(s.linuxSem)) + } + }) + } +} + +// A dispatch that found a free slot at once is acting on a webhook seconds +// old. It must not spend an API call per job asking about it. +func TestAdmitDispatch_FastPathDoesNotAsk(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + defer close(fake.waitBlock) + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + prov, base := newJobStateProvider(false, nil) // would say "gone" if asked + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 1, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + s.webhookMode = true + + s.handleQueued(context.Background(), linuxQueuedEvent(prov, 920)) + + if got := prov.calls.Load(); got != 0 { + t.Errorf("platform consulted %d time(s) on the fast path, want 0", got) + } + if got := base.claims.Load(); got != 1 { + t.Errorf("claims = %d, want 1", got) + } +} + +// TestReprovisionIfStranded_AsksPlatformFirst: a runner exits unassigned and +// the node never saw its job run. Re-queue only if the platform agrees the job +// still needs a runner. +func TestReprovisionIfStranded_AsksPlatformFirst(t *testing.T) { + tests := []struct { + name string + awaiting bool + apiErr error + wantClaims int32 + }{ + {name: "job is gone", awaiting: false, wantClaims: 1}, + {name: "job still queued", awaiting: true, wantClaims: 2}, + {name: "platform cannot answer", apiErr: errors.New("api down"), wantClaims: 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + defer close(fake.waitBlock) + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + prov, base := newJobStateProvider(true, nil) + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 8, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + s.webhookMode = true + + const jobID int64 = 930 + key := jobKey{Provider: prov.Name(), JobID: jobID} + event := linuxQueuedEvent(prov, jobID) + + s.handleQueued(context.Background(), event) + if got := base.claims.Load(); got != 1 { + t.Fatalf("first dispatch: claims = %d, want 1", got) + } + + // The runner exits without the job ever being observed; by now the + // platform has its own view of the job. + s.mu.Lock() + s.untrackRunningLocked(key, s.running[key]) + s.mu.Unlock() + prov.awaiting.Store(tt.awaiting) + prov.err = tt.apiErr + + s.reprovisionIfStranded(context.Background(), event) + + if tt.wantClaims == 1 { + if !waitForStarted(t, s, key) { + t.Fatal("gone verdict was never recorded") + } + time.Sleep(100 * time.Millisecond) // let any (bug) re-dispatch land + } else { + waitForClaims(t, base, tt.wantClaims) + } + if got := base.claims.Load(); got != tt.wantClaims { + t.Errorf("claims = %d, want %d", got, tt.wantClaims) + } + }) + } +} + +// TestStrandLoop_StopsWhenPlatformSaysGone drives the real wait-goroutine: +// every runner exits at once, unassigned, exactly like the idle runners on the +// Windows node. Before the fix this looped to the zombie cap (and in +// production past it, see cleanSeen). With the platform reporting the job +// finished, the first exit must end it. +func TestStrandLoop_StopsWhenPlatformSaysGone(t *testing.T) { + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + close(fake.waitBlock) // Wait returns immediately: the runner exits unassigned + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + prov, base := newJobStateProvider(false, nil) + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 8, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + s.webhookMode = true + + const jobID int64 = 940 + key := jobKey{Provider: prov.Name(), JobID: jobID} + s.handleQueued(context.Background(), linuxQueuedEvent(prov, jobID)) + + if !waitForStarted(t, s, key) { + t.Fatal("the strand loop never consulted the platform") + } + time.Sleep(150 * time.Millisecond) + if got := base.claims.Load(); got != 1 { + t.Errorf("claims = %d, want 1: a job the platform reports finished was re-provisioned", got) + } +} + +// TestAcquireSlot_GivesUpAfterMaxWait: a wait has an end. The dispatch is +// dropped cleanly — no claim, no leaked slot, and its dedup cleared so the job +// is dispatched afresh if it turns out to be still queued. +func TestAcquireSlot_GivesUpAfterMaxWait(t *testing.T) { + old := maxSlotWait + maxSlotWait = 100 * time.Millisecond + t.Cleanup(func() { maxSlotWait = old }) + + fake := &fakeDispatchServer{waitBlock: make(chan struct{})} + defer close(fake.waitBlock) + _, dc, stopDispatch := startFakeDispatchServer(t, fake) + defer stopDispatch() + + prov, base := newJobStateProvider(true, nil) + s := New(Config{ + Providers: []providers.Provider{prov}, + LinuxDispatcher: dc, + MaxConcurrent: 1, + JobTimeout: 30 * time.Second, + Log: quietLogger(), + }) + s.webhookMode = true + + const jobID int64 = 950 + key := jobKey{Provider: prov.Name(), JobID: jobID} + event := linuxQueuedEvent(prov, jobID) + + s.linuxSem <- struct{}{} // full, and nothing will free it + done := make(chan struct{}) + go func() { + defer close(done) + s.handleQueued(context.Background(), event) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("dispatch was still blocked long after maxSlotWait") + } + + if got := base.claims.Load(); got != 0 { + t.Errorf("claims = %d, want 0", got) + } + if len(s.linuxSem) != 1 { + t.Errorf("linuxSem holds %d, want 1 (only the occupier): the abandoned wait took or freed a slot", len(s.linuxSem)) + } + s.mu.Lock() + _, pending := s.pending[key] + _, seen := s.seen[key] + s.mu.Unlock() + if pending || seen { + t.Errorf("pending=%v seen=%v after give-up, want both cleared so the job can be re-dispatched", pending, seen) + } + + // Still queued and now there is room: the next event dispatches it. + <-s.linuxSem + s.handleQueued(context.Background(), event) + if got := base.claims.Load(); got != 1 { + t.Errorf("re-dispatch after give-up: claims = %d, want 1", got) + } +} + +// TestCleanSeen_KeepsAttemptsForHeldJobs: the zombie counter survives a seen +// expiry while the node still holds the job. Resetting it there is what let +// strand cycles (each longer than seenTTL) re-provision forever. +func TestCleanSeen_KeepsAttemptsForHeldJobs(t *testing.T) { + s := New(Config{Log: quietLogger()}) + pendingKey := jobKey{Provider: "github", JobID: 1} + runningKey := jobKey{Provider: "github", JobID: 2} + goneKey := jobKey{Provider: "github", JobID: 3} + expired := time.Now().Add(-2 * seenTTL) + + s.mu.Lock() + for _, k := range []jobKey{pendingKey, runningKey, goneKey} { + s.seen[k] = expired + s.attempts[k] = 3 + s.jobLabels[k] = "linux,self-hosted" + } + s.pending[pendingKey] = struct{}{} + s.running[runningKey] = &runningJob{} + s.mu.Unlock() + + s.cleanSeen() + + s.mu.Lock() + defer s.mu.Unlock() + for _, k := range []jobKey{pendingKey, runningKey} { + if s.attempts[k] != 3 { + t.Errorf("job %d: attempts = %d, want 3 — reset while the node still holds the job", k.JobID, s.attempts[k]) + } + if _, ok := s.jobLabels[k]; !ok { + t.Errorf("job %d: labels dropped while the job is still demand", k.JobID) + } + } + if _, ok := s.attempts[goneKey]; ok { + t.Error("attempts kept for a job that left the queue; a later rerun would start over the cap") + } + if _, ok := s.seen[pendingKey]; ok { + t.Error("expired seen stamp was not pruned") + } +}