From a243809c148092fb495a57a582919f47c915991b Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:22:25 -0700 Subject: [PATCH 1/4] fix(startup): reflect actual discovery mode in log message The startup log unconditionally printed 'polling mode enabled (tunnel disabled)' whenever no tunnel provider was configured, even when the scheduler was running in webhook mode via an external ingress (cfg.Webhook.Secret set). scheduler.Run gates webhook mode on Tunnel != nil OR WebhookSecret != "", so the two paths must match. Now: log 'webhook mode enabled (external ingress, no tunnel provider)' when a secret is set without a tunnel, and 'polling mode enabled (no tunnel, no webhook secret)' only when neither is configured. --- cmd/ephemerd/main.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index e13895b..435dd5b 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -560,8 +560,14 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP return fmt.Errorf("creating tunnel provider: %w", err) } log.Info("webhook tunnel configured", "provider", cfg.Webhook.Tunnel) + } else if cfg.Webhook.Secret != "" { + // No tunnel, but a webhook secret is configured — the operator is + // running webhooks via an external ingress (reverse proxy, LB, etc.). + // Scheduler treats this as webhook mode (see scheduler.Run: + // useWebhook = Tunnel != nil || WebhookSecret != ""). + log.Info("webhook mode enabled (external ingress, no tunnel provider)") } else { - log.Info("polling mode enabled (tunnel disabled)") + log.Info("polling mode enabled (no tunnel, no webhook secret)") } // Start scheduler (ties CI provider jobs to container lifecycle) From 2e9421a6b2c86f8087fbc626546df85398c7b208 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:23:30 -0700 Subject: [PATCH 2/4] fix(scheduler): record label-mismatched jobs in seen to stop re-adjudication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleQueued ran the canHandleJob label check BEFORE the seen/pending dedup writes, so label-mismatched jobs (e.g. self-hosted+macos+arm64 while the Mac host is down) returned without leaving any trace. On the next poll cycle the same queued job re-appeared and re-triggered the skip path — logs showed 'skipping job, OS labels don't match this platform' repeating every ~15s for hours, and every poll cycle still paid the API cost of listing runs + listing jobs for the run. Reorder handleQueued so the dedup check runs first; on label mismatch, write the key into seen (10min TTL) so re-deliveries short-circuit at the dedup gate. No pending-slot semantics change: we never took the semaphore, so pending stays unset. Update TestHandleQueued_SkipsMismatchedLabels to encode the new contract (seen IS populated, pending is NOT) and add a re-delivery assertion that pins the O(1)-per-event guarantee. --- pkg/scheduler/handle_queued_test.go | 23 ++++++++++++++++++++--- pkg/scheduler/scheduler.go | 25 ++++++++++++++++++------- 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/pkg/scheduler/handle_queued_test.go b/pkg/scheduler/handle_queued_test.go index 0cfaba0..962dea5 100644 --- a/pkg/scheduler/handle_queued_test.go +++ b/pkg/scheduler/handle_queued_test.go @@ -162,7 +162,13 @@ func TestHandleQueued_AlreadyRunningWithProvider(t *testing.T) { } // TestHandleQueued_SkipsMismatchedLabels verifies canHandleJob rejection -// before any side-effects are recorded. +// records the job in `seen` so subsequent poll/webhook re-deliveries +// short-circuit at the dedup check instead of re-adjudicating every tick. +// +// Previously the label check ran before `seen` was written, so a polling +// provider would re-log "skipping job, OS labels don't match" every poll +// cycle for the same unclaimable job (observed as a ~15s re-examination +// loop that also drained the API rate budget). func TestHandleQueued_SkipsMismatchedLabels(t *testing.T) { mp := newMockProvider("github") s := New(Config{ @@ -192,9 +198,20 @@ func TestHandleQueued_SkipsMismatchedLabels(t *testing.T) { } s.mu.Lock() _, seen := s.seen[keyFor(event)] + _, pending := s.pending[keyFor(event)] s.mu.Unlock() - if seen { - t.Error("mismatched-label job should not be added to seen") + if !seen { + t.Error("mismatched-label job should be recorded in seen to prevent re-adjudication loops") + } + if pending { + t.Error("mismatched-label job should not be recorded in pending (we never took a slot)") + } + + // Re-deliver the same event — the second call must short-circuit at + // the dedup check and never reach canHandleJob a second time. + s.handleQueued(context.Background(), event) + if got := len(mp.claims); got != 0 { + t.Errorf("re-delivered mismatched-label job should still not claim, got %d", got) } } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 028d410..5379c27 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -504,13 +504,12 @@ func (s *Scheduler) handleQueued(ctx context.Context, event providers.JobEvent) key := keyFor(event) log := s.cfg.Log.With("job_id", jobID, "repo", event.Repo) - // Skip jobs whose OS labels don't match this platform - if len(event.Labels) > 0 && !s.canHandleJob(event.Labels) { - log.Debug("skipping job, OS labels don't match this platform", "labels", event.Labels) - return - } - - // Dedup: skip if we've already seen this job recently + // Dedup: check the running / pending / seen maps FIRST so a job we've + // already skipped for label mismatch doesn't get re-adjudicated (and + // re-logged) every poll cycle. Previously the label check ran before + // dedup and returned without recording anything in `seen`, so a + // polling provider would surface the same unclaimable queued job on + // every tick, spamming logs and draining API budget. s.mu.Lock() if _, exists := s.running[key]; exists { s.mu.Unlock() @@ -527,6 +526,18 @@ func (s *Scheduler) handleQueued(ctx context.Context, event providers.JobEvent) log.Debug("ignoring duplicate queued event, job recently handled") return } + + // Skip jobs whose OS labels don't match this platform. Record in + // `seen` so subsequent poll/webhook re-deliveries short-circuit at + // the dedup check above; this is what makes label-mismatched jobs + // O(1) per event instead of O(polls-until-seenTTL-expires). + if len(event.Labels) > 0 && !s.canHandleJob(event.Labels) { + s.seen[key] = time.Now() + s.mu.Unlock() + log.Debug("skipping job, OS labels don't match this platform", "labels", event.Labels) + return + } + s.pending[key] = struct{}{} s.seen[key] = time.Now() From 0781a2cb4beaafe7c974634878f7a6ee2f87085b Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:25:22 -0700 Subject: [PATCH 3/4] feat(metrics): fresh GitHub rate-limit gauges from response headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ephemerd_github_api_rate_remaining was defined but never written, so it sat at 0 forever. Operators looking at Grafana couldn't tell 'budget truly exhausted' from 'no data yet' from 'stale reading after window reset'. Wrap the GitHub HTTP client in a rateTrackingTransport that parses X-RateLimit-{Limit,Remaining,Reset} on EVERY response (200 or 429/5xx error — errors carry rate headers too, and that's exactly when we most need them). Introduce three companion gauges: ephemerd_github_api_rate_limit (ceiling for the window) ephemerd_github_api_rate_reset_seconds (unix ts of next reset) ephemerd_github_api_rate_updated_seconds (when we last observed) Use rate_updated_seconds vs time() to age out a stale reading in dashboards / alerts. Also expose Client.RateSnapshot() for callers that want to bias retry backoff based on the live rate state (used by the upcoming scheduler claim-retry queue). Guard against responses that don't emit rate headers (some GraphQL paths) — those must NOT clobber the last known value to zero. --- pkg/github/client.go | 125 +++++++++++++++++++++++++++--- pkg/github/rate_transport_test.go | 106 +++++++++++++++++++++++++ pkg/metrics/metrics.go | 29 ++++++- 3 files changed, 250 insertions(+), 10 deletions(-) create mode 100644 pkg/github/rate_transport_test.go diff --git a/pkg/github/client.go b/pkg/github/client.go index 8e47f7b..44103fb 100644 --- a/pkg/github/client.go +++ b/pkg/github/client.go @@ -10,7 +10,11 @@ import ( "io" "log/slog" "net/http" + "strconv" + "sync/atomic" + "time" + "github.com/ephpm/ephemerd/pkg/metrics" gh "github.com/google/go-github/v72/github" "gopkg.in/yaml.v3" ) @@ -28,9 +32,23 @@ type Config struct { // Client handles GitHub API interactions and webhook events. type Client struct { - cfg Config - client *gh.Client - app *AppAuth // nil when using PAT + cfg Config + client *gh.Client + app *AppAuth // nil when using PAT + lastRate *rateSnapshot // last observed rate-limit state (nil in tests using SetHTTPClient) +} + +// RateSnapshot returns the last-observed rate-limit state. The scheduler +// uses this to bias retry backoff — when remaining==0 and now < reset, +// the next attempt is scheduled just after reset instead of blindly +// falling through the exponential-backoff ladder. Fields are zero-valued +// when the client hasn't yet made a request or was created via the +// bare test constructor. +func (c *Client) RateSnapshot() (remaining, limit int64, reset, updated time.Time) { + if c.lastRate == nil { + return 0, 0, time.Time{}, time.Time{} + } + return c.lastRate.Snapshot() } // JobEvent is emitted when a workflow_job webhook fires. @@ -42,22 +60,40 @@ type JobEvent struct { // New creates a GitHub client. // Uses AppAuth for dynamic token refresh when configured, otherwise a static PAT. +// +// All GitHub API responses flow through rateTrackingTransport, which parses +// the X-RateLimit-* headers and updates the ephemerd_github_api_rate_* +// gauges. Operators use ephemerd_github_api_rate_updated_seconds to tell a +// current-but-exhausted budget from a stale reading. func New(cfg Config) (*Client, error) { var c *gh.Client + var lastRate rateSnapshot if cfg.AppAuth != nil { - // Use a custom transport that injects the latest token on each request. + // Use a custom transport that injects the latest token on each + // request, then wrap it in the rate tracker so every response + // updates the metrics. httpClient := &http.Client{ - Transport: &appAuthTransport{app: cfg.AppAuth}, + Transport: &rateTrackingTransport{ + next: &appAuthTransport{app: cfg.AppAuth}, + last: &lastRate, + }, } c = gh.NewClient(httpClient) } else { - c = gh.NewClient(nil).WithAuthToken(cfg.Token) + httpClient := &http.Client{ + Transport: &rateTrackingTransport{ + next: http.DefaultTransport, + last: &lastRate, + }, + } + c = gh.NewClient(httpClient).WithAuthToken(cfg.Token) } return &Client{ - cfg: cfg, - client: c, - app: cfg.AppAuth, + cfg: cfg, + client: c, + app: cfg.AppAuth, + lastRate: &lastRate, }, nil } @@ -72,6 +108,77 @@ func (t *appAuthTransport) RoundTrip(req *http.Request) (*http.Response, error) return http.DefaultTransport.RoundTrip(req) } +// rateSnapshot is the last observed GitHub rate-limit state. All fields +// are stored as int64 unix seconds / counts so atomic loads and stores +// are lock-free. +type rateSnapshot struct { + remaining atomic.Int64 // X-RateLimit-Remaining + limit atomic.Int64 // X-RateLimit-Limit + resetUnix atomic.Int64 // X-RateLimit-Reset (unix seconds) + updatedUnix atomic.Int64 // when we last parsed rate headers +} + +// Snapshot returns a point-in-time copy for use by the scheduler retry +// queue (rate-aware backoff). +func (r *rateSnapshot) Snapshot() (remaining, limit int64, reset, updated time.Time) { + remaining = r.remaining.Load() + limit = r.limit.Load() + if u := r.resetUnix.Load(); u > 0 { + reset = time.Unix(u, 0) + } + if u := r.updatedUnix.Load(); u > 0 { + updated = time.Unix(u, 0) + } + return remaining, limit, reset, updated +} + +// rateTrackingTransport wraps another RoundTripper and updates the rate +// gauges + a shared rateSnapshot from response headers on every call. +type rateTrackingTransport struct { + next http.RoundTripper + last *rateSnapshot +} + +func (t *rateTrackingTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.next.RoundTrip(req) + if resp != nil { + t.observe(resp) + } + return resp, err +} + +// observe parses X-RateLimit-* headers from a response and updates +// metrics + the shared rateSnapshot. Called on every response, including +// error responses (429/5xx) — those still carry the rate headers. +func (t *rateTrackingTransport) observe(resp *http.Response) { + rem := resp.Header.Get("X-RateLimit-Remaining") + if rem == "" { + // Not every endpoint returns rate headers (e.g. some GraphQL + // paths). Don't touch the gauge — a fresher value is better + // than clobbering with 0. + return + } + if v, err := strconv.ParseInt(rem, 10, 64); err == nil { + t.last.remaining.Store(v) + metrics.GitHubAPIRateRemaining.Set(float64(v)) + } + if lim := resp.Header.Get("X-RateLimit-Limit"); lim != "" { + if v, err := strconv.ParseInt(lim, 10, 64); err == nil { + t.last.limit.Store(v) + metrics.GitHubAPIRateLimit.Set(float64(v)) + } + } + if reset := resp.Header.Get("X-RateLimit-Reset"); reset != "" { + if v, err := strconv.ParseInt(reset, 10, 64); err == nil { + t.last.resetUnix.Store(v) + metrics.GitHubAPIRateResetSeconds.Set(float64(v)) + } + } + now := time.Now().Unix() + t.last.updatedUnix.Store(now) + metrics.GitHubAPIRateUpdatedSeconds.Set(float64(now)) +} + // SetHTTPClient replaces the underlying go-github client. // Used by test infrastructure to point at a fake server. func (c *Client) SetHTTPClient(ghClient *gh.Client) { diff --git a/pkg/github/rate_transport_test.go b/pkg/github/rate_transport_test.go new file mode 100644 index 0000000..f643425 --- /dev/null +++ b/pkg/github/rate_transport_test.go @@ -0,0 +1,106 @@ +package github + +import ( + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" +) + +// TestRateTrackingTransport_UpdatesFromHeaders verifies the transport +// parses X-RateLimit-* headers and updates the shared snapshot on every +// response — including error responses. +func TestRateTrackingTransport_UpdatesFromHeaders(t *testing.T) { + resetAt := time.Now().Add(30 * time.Minute).Unix() + + // A synthetic server that returns rate headers on every response + // (both 200 and 429). We tick the response status through the + // requests to prove the transport observes both. + var reqCount int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + reqCount++ + w.Header().Set("X-RateLimit-Limit", "5000") + if reqCount == 1 { + w.Header().Set("X-RateLimit-Remaining", "4321") + w.Header().Set("X-RateLimit-Reset", strconv.FormatInt(resetAt, 10)) + w.WriteHeader(http.StatusOK) + return + } + // Second request: budget exhausted with headers still present. + w.Header().Set("X-RateLimit-Remaining", "0") + w.Header().Set("X-RateLimit-Reset", strconv.FormatInt(resetAt, 10)) + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + + var snap rateSnapshot + tr := &rateTrackingTransport{next: http.DefaultTransport, last: &snap} + client := &http.Client{Transport: tr} + + // First request: 200 with remaining=4321. + resp, err := client.Get(srv.URL) + if err != nil { + t.Fatalf("first request: %v", err) + } + _ = resp.Body.Close() + + rem, lim, reset, updated := snap.Snapshot() + if rem != 4321 { + t.Errorf("remaining = %d, want 4321", rem) + } + if lim != 5000 { + t.Errorf("limit = %d, want 5000", lim) + } + if reset.Unix() != resetAt { + t.Errorf("reset = %v, want unix %d", reset, resetAt) + } + if updated.IsZero() { + t.Error("updated timestamp should be set after a request") + } + + // Second request: 429 with remaining=0 — the transport must still + // update the snapshot from headers (this is the whole point: an + // error response is exactly when we most need fresh data). + resp, err = client.Get(srv.URL) + if err != nil { + t.Fatalf("second request: %v", err) + } + _ = resp.Body.Close() + + rem, _, _, updated2 := snap.Snapshot() + if rem != 0 { + t.Errorf("after 429, remaining = %d, want 0", rem) + } + if !updated2.After(updated.Add(-time.Second)) { + t.Errorf("updated should have advanced: prev=%v new=%v", updated, updated2) + } +} + +// TestRateTrackingTransport_NoHeaders_LeavesSnapshotAlone pins that a +// response missing the rate headers doesn't clobber the last known +// value with zero — otherwise a stray endpoint that doesn't emit rate +// headers would zero the gauge and confuse operators. +func TestRateTrackingTransport_NoHeaders_LeavesSnapshotAlone(t *testing.T) { + var snap rateSnapshot + snap.remaining.Store(999) + snap.limit.Store(5000) + snap.updatedUnix.Store(time.Now().Unix()) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + // No X-RateLimit-* headers at all. + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + tr := &rateTrackingTransport{next: http.DefaultTransport, last: &snap} + resp, err := (&http.Client{Transport: tr}).Get(srv.URL) + if err != nil { + t.Fatalf("request: %v", err) + } + _ = resp.Body.Close() + + if got := snap.remaining.Load(); got != 999 { + t.Errorf("remaining clobbered to %d — should have stayed at 999 when headers absent", got) + } +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 272f950..4d42895 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -56,9 +56,36 @@ var ( }, []string{"endpoint", "status_code"}) // GitHubAPIRateRemaining tracks the remaining GitHub API rate limit. + // Updated on every GitHub API response from X-RateLimit-Remaining; + // pair with GitHubAPIRateUpdatedSeconds so operators can distinguish + // "0 and current" from "0 and stale". GitHubAPIRateRemaining = promauto.NewGauge(prometheus.GaugeOpts{ Name: "ephemerd_github_api_rate_remaining", - Help: "Remaining GitHub API rate limit quota.", + Help: "Remaining GitHub API rate limit quota (last observed).", + }) + + // GitHubAPIRateLimit reports the ceiling for the current window + // (typically 5000/hr for app installations, 60 for unauthenticated). + GitHubAPIRateLimit = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "ephemerd_github_api_rate_limit", + Help: "GitHub API rate limit ceiling for the current window (last observed).", + }) + + // GitHubAPIRateResetSeconds is the unix timestamp at which the rate + // window resets, taken from X-RateLimit-Reset. + GitHubAPIRateResetSeconds = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "ephemerd_github_api_rate_reset_seconds", + Help: "Unix timestamp of the next GitHub rate-limit window reset.", + }) + + // GitHubAPIRateUpdatedSeconds is the unix timestamp of the most + // recent GitHub API response that carried rate headers. Operators + // use `time() - ephemerd_github_api_rate_updated_seconds` to age + // out a stale-looking `_rate_remaining` reading — if the update + // timestamp is older than the reset timestamp, the gauge is stale. + GitHubAPIRateUpdatedSeconds = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "ephemerd_github_api_rate_updated_seconds", + Help: "Unix timestamp of the last GitHub API response that carried rate headers.", }) // GitHubPollTotal counts polling cycles. From 6e5c1535bfe97d964de0d07ce653a74e20cb6886 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:42:45 -0700 Subject: [PATCH 4/4] feat(scheduler): claim retry queue with jittered backoff and rate-aware scheduling GitHub does not re-deliver workflow_job webhooks. When ephemerd's initial claim/JIT-runner-registration attempt failed with a transient error (rate-limit exhausted, transient 5xx, network blip), the queued job was dropped forever - operators noticed only via missing runs and manually re-queued each job by hand. Add an in-memory priority-heap retry queue keyed by jobKey. Failed claims are re-scheduled on a jittered ladder (30s, 1m, 2m, 5m, 10m by default), giving up after MaxAge (default 90m). Hooks: - handleLocalJob, handleLinuxJob, handleMacOSJob classify the claim error via classifyErr and enqueue on retryable failures. The old blind time.Sleep(backoffDuration) and 5s sleep are gone - the sem is released first so we do not hold a slot idle across the wait. - handleCompleted Drops the retry key so a job picked up elsewhere (peer daemon sharing the installation token, or manual cancel) stops getting reattempted. - The event loop also Drops on 'in_progress', matching the same intent. Rate-aware backoff: when the last GitHub API response reported remaining=0 with a fresh (< 5m old) update timestamp and reset > now, the next attempt is snapped to reset + [5s, 25s) jitter. This avoids burning attempts against a provably-exhausted budget, and the jitter prevents multiple daemons sharing the installation token from stampeding at the exact same second. Error classification (classifyErr) sorts errors into non_retryable / rate_limit / server_side / network / unknown_retryable. 404/422/401/permission-denied are non-retryable; 429 and the go-github RateLimitError/AbuseRateLimitError typed errors are rate_limit; 5xx plus string-match fallbacks ("bad gateway", "gateway timeout") are server_side; net.Error is network. Unknown errors are treated as retryable within backoff/MaxAge - the whole point is to stop losing jobs to unexpected transient failures. Wiring: - New RetryConfig on scheduler.Config, plumbed from [runner.claim_retry] in the TOML (enabled defaults to true - losing jobs is worse than extra retries). - RateHint closure in cmd/ephemerd/main.go walks activeProviders for a GitHub provider and binds to its RateSnapshot; nil when no rate-aware provider is present. - providers/github.Provider exposes RateSnapshot() forwarding to the underlying github.Client (added in the earlier rate-transport commit). Tests: retry_queue_test covers backoff progression (schedule ladder, clamp past end), MaxAge give-up, Drop on completion, non-retryable classification, rate-aware snap-to-reset with a fake clock, and the full classifyErr table. retry_integration_test uses the existing claimErrorProvider to end-to-end verify that a retryable failure in handleLocalJob puts one entry in the queue and that a completed webhook drops it. All new tests use table-driven or deterministic clock patterns; no time.Sleep in scheduling logic. --- cmd/ephemerd/main.go | 43 ++ config.example.toml | 44 +- pkg/config/config.go | 97 +++++ pkg/providers/github/github.go | 9 + pkg/scheduler/retry_integration_test.go | 138 ++++++ pkg/scheduler/retry_queue.go | 530 ++++++++++++++++++++++++ pkg/scheduler/retry_queue_test.go | 363 ++++++++++++++++ pkg/scheduler/scheduler.go | 101 ++++- 8 files changed, 1317 insertions(+), 8 deletions(-) create mode 100644 pkg/scheduler/retry_integration_test.go create mode 100644 pkg/scheduler/retry_queue.go create mode 100644 pkg/scheduler/retry_queue_test.go diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 435dd5b..1a0d3dd 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -570,6 +570,26 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP log.Info("polling mode enabled (no tunnel, no webhook secret)") } + // Build the claim retry queue config. Wired here (not in Load) so + // the RateHint closure can bind to the actual GitHub provider + // instance: the retry queue uses the last-observed rate snapshot to + // snap the next attempt just past the reset window when the API + // budget is exhausted. + retryCfg := scheduler.RetryConfig{ + Enabled: cfg.Runner.ClaimRetryEnabled(), + Schedule: cfg.Runner.ClaimRetrySchedule(), + MaxAge: cfg.Runner.ClaimRetryMaxAge(), + Jitter: cfg.Runner.ClaimRetryJitter(), + RateHint: buildRateHint(activeProviders), + } + if retryCfg.Enabled { + log.Info("claim retry queue enabled", + "max_age", retryCfg.MaxAge, + "schedule", retryCfg.Schedule, + "jitter", retryCfg.Jitter, + "rate_aware", retryCfg.RateHint != nil) + } + // Start scheduler (ties CI provider jobs to container lifecycle) sched := scheduler.New(scheduler.Config{ Runtime: rt, @@ -591,6 +611,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP ShutdownTimeout: cfg.Runner.ParsedShutdownTimeout(), LogRetention: cfg.Log.LogRetentionDuration(), RunnerImageForRepo: cfg.Runner.ImageForRepoOS, + Retry: retryCfg, Log: log, }) @@ -789,6 +810,28 @@ func pollInterval(cfg *config.Config) time.Duration { } } +// buildRateHint returns a rate-hint closure that the scheduler's retry +// queue uses to bias backoff toward the next GitHub rate-limit reset. +// Returns nil when no rate-aware provider is present; the retry queue +// honors nil and falls through to the plain jittered backoff ladder. +// +// Currently only the GitHub provider surfaces rate data. If additional +// providers grow rate tracking, extend this to combine them (typically: +// use the earliest available reset time). +func buildRateHint(active []providers.Provider) func() (int64, time.Time, time.Time) { + for _, p := range active { + gp, ok := p.(*githubProv.Provider) + if !ok { + continue + } + return func() (int64, time.Time, time.Time) { + remaining, _, reset, updated := gp.RateSnapshot() + return remaining, reset, updated + } + } + return nil +} + // crictlCmd exposes the upstream crictl CLI against ephemerd's embedded // containerd CRI socket. The crictl library is linked in-process; no external // binary is required. See docs/arch/crictl.md. diff --git a/config.example.toml b/config.example.toml index a9fb24b..701edd2 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,6 +1,6 @@ [github] # Authentication: personal access token OR GitHub App credentials. -# Use one approach — if app_id is set, it takes priority over token. +# Use one approach - if app_id is set, it takes priority over token. # token = "ghp_your_token_here" # GitHub App authentication (recommended): @@ -63,6 +63,48 @@ job_timeout = "2h" # Time to wait for running jobs during graceful shutdown shutdown_timeout = "5m" +# Claim retry queue. +# +# GitHub does not re-deliver workflow_job webhooks. When ephemerd's +# initial claim/provision attempt fails for a transient reason +# (rate-limit exhausted, transient 5xx, network blip), the job would +# otherwise be lost. The retry queue schedules re-attempts on a +# jittered exponential backoff ladder and gives up after max_age. +# +# When the last GitHub API response reported remaining=0 with a known +# reset time, the next attempt is snapped to just after reset instead +# of falling through the ladder, so we do not burn attempts against a +# provably-exhausted budget. +# +# Enabled by default. Set enabled = false to restore the pre-existing +# "log and drop" behavior. +# [runner.claim_retry] +# enabled = true +# max_age = "90m" +# schedule = ["30s", "1m", "2m", "5m", "10m"] +# jitter = 0.2 + +# Claim retry queue. +# +# GitHub does not re-deliver workflow_job webhooks. When ephemerd's +# initial claim/provision attempt fails for a transient reason +# (rate-limit exhausted, transient 5xx, network blip), the job would +# otherwise be lost. The retry queue schedules re-attempts on a +# jittered exponential backoff ladder and gives up after max_age. +# +# When the last GitHub API response says remaining==0 with a known +# reset time, the next attempt is snapped to just after reset instead +# of falling through the ladder, so we don't burn attempts against a +# provably-exhausted budget. +# +# Enabled by default. Set enabled = false to restore the pre-existing +# "log and drop" behavior. +# [runner.claim_retry] +# enabled = true +# max_age = "90m" +# schedule = ["30s", "1m", "2m", "5m", "10m"] +# jitter = 0.2 + [vm.linux] # Enable a Linux VM for running Linux jobs on Windows/macOS hosts. # On Windows: creates a WSL2 distro with an embedded ephemerd binary. diff --git a/pkg/config/config.go b/pkg/config/config.go index 6d8f81d..18ffcad 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -453,6 +453,42 @@ type RunnerConfig struct { JobTimeout string `toml:"job_timeout"` ShutdownTimeout string `toml:"shutdown_timeout"` Windows WindowsRunnerToml `toml:"windows"` + + // ClaimRetry controls the in-memory retry queue for jobs whose + // initial claim / provision attempt fails with a transient error + // (rate limit, 5xx, network). GitHub does not re-deliver + // workflow_job webhooks, so without this queue any queued job that + // hit an API blip at claim time would be lost until human + // intervention. + ClaimRetry ClaimRetryToml `toml:"claim_retry"` +} + +// ClaimRetryToml configures the scheduler's claim-retry queue. +// +// Enabled defaults to true when the [runner.claim_retry] table is +// omitted (see RunnerConfig.ClaimRetryEnabled). Losing queued jobs to +// transient errors is almost never what an operator wants; set +// enabled = false to restore the pre-existing "log and drop" behavior. +type ClaimRetryToml struct { + // Enabled toggles the retry queue. Nil = default true; operators + // disable by explicitly setting enabled = false. + Enabled *bool `toml:"enabled"` + + // MaxAge is the total time budget from first failure to giving up. + // Accepts Go duration strings ("90m", "1h30m"). Default 90m. + MaxAge string `toml:"max_age"` + + // Schedule is the ordered backoff ladder. Each entry is the base + // delay for one attempt; jitter is applied on top. Default is + // {30s, 1m, 2m, 5m, 10m}. Use a shorter schedule for shorter + // MaxAge, or a longer one to let jobs marinate through extended + // outages. + Schedule []string `toml:"schedule"` + + // Jitter is the +/- fraction applied to each delay (0.0-1.0). + // Default 0.2 (+/-20%). Set 0 to disable jitter (useful in tests, + // rarely useful in production). + Jitter *float64 `toml:"jitter"` } // WindowsRunnerToml configures resource limits for Hyper-V isolated Windows @@ -505,6 +541,67 @@ func (r *RunnerConfig) ImageForRepo(repo string) string { return r.DefaultImage } +// ClaimRetryEnabled reports whether the claim retry queue should run. +// Defaults to true (retries on) when the table is omitted or Enabled is +// nil; operators opt out with `enabled = false`. +func (r *RunnerConfig) ClaimRetryEnabled() bool { + if r == nil || r.ClaimRetry.Enabled == nil { + return true + } + return *r.ClaimRetry.Enabled +} + +// ClaimRetryMaxAge returns the retry queue give-up threshold. Defaults +// to 90 minutes when unset or unparseable. +func (r *RunnerConfig) ClaimRetryMaxAge() time.Duration { + if r == nil || r.ClaimRetry.MaxAge == "" { + return 90 * time.Minute + } + if d, err := time.ParseDuration(r.ClaimRetry.MaxAge); err == nil && d > 0 { + return d + } + return 90 * time.Minute +} + +// ClaimRetrySchedule returns the retry backoff ladder. Defaults to +// {30s, 1m, 2m, 5m, 10m} when unset. Unparseable entries are dropped; +// if all entries fail to parse, the default is returned. +func (r *RunnerConfig) ClaimRetrySchedule() []time.Duration { + def := []time.Duration{ + 30 * time.Second, + 1 * time.Minute, + 2 * time.Minute, + 5 * time.Minute, + 10 * time.Minute, + } + if r == nil || len(r.ClaimRetry.Schedule) == 0 { + return def + } + out := make([]time.Duration, 0, len(r.ClaimRetry.Schedule)) + for _, s := range r.ClaimRetry.Schedule { + if d, err := time.ParseDuration(s); err == nil && d > 0 { + out = append(out, d) + } + } + if len(out) == 0 { + return def + } + return out +} + +// ClaimRetryJitter returns the retry backoff jitter fraction. Defaults +// to 0.2 (+/-20%). Values outside [0,1] are clamped to the default. +func (r *RunnerConfig) ClaimRetryJitter() float64 { + if r == nil || r.ClaimRetry.Jitter == nil { + return 0.2 + } + j := *r.ClaimRetry.Jitter + if j < 0 || j > 1 { + return 0.2 + } + return j +} + // ParsedPollInterval returns the poll interval as a time.Duration. func (g *GitHubConfig) ParsedPollInterval() time.Duration { if g.PollInterval == "" { diff --git a/pkg/providers/github/github.go b/pkg/providers/github/github.go index 9772f99..53fe3f4 100644 --- a/pkg/providers/github/github.go +++ b/pkg/providers/github/github.go @@ -176,6 +176,15 @@ func (p *Provider) CleanStaleWebhooks(ctx context.Context) { p.client.CleanStaleWebhooks(ctx) } +// RateSnapshot exposes the underlying GitHub client's last-observed +// rate-limit state so the scheduler's retry queue can bias backoff: +// when remaining==0 with a fresh update timestamp and reset > now, the +// next attempt is snapped just past the reset. Zero values mean "no +// data yet" (the client has not yet made a request). +func (p *Provider) RateSnapshot() (remaining, limit int64, reset, updated time.Time) { + return p.client.RateSnapshot() +} + // CatchUpPoll fires a single poll to discover jobs queued while ephemerd was // offline. Used in webhook mode (where continuous polling is disabled) to catch // jobs that transitioned to "queued" before webhooks could be registered — diff --git a/pkg/scheduler/retry_integration_test.go b/pkg/scheduler/retry_integration_test.go new file mode 100644 index 0000000..f58bfc0 --- /dev/null +++ b/pkg/scheduler/retry_integration_test.go @@ -0,0 +1,138 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +// TestRetryQueue_Integration_ClaimFailureEnqueues verifies that when +// handleLocalJob's claim fails with a retryable error, the retry queue +// picks it up. Uses a claim provider that always errors and asserts the +// queue depth after the failure path returns. +func TestRetryQueue_Integration_ClaimFailureEnqueues(t *testing.T) { + clk := newFakeClock(time.Now()) + // Retryable error: 500 server side. + p := newClaimErrorProvider("github", errors.New("HTTP 500 internal server error")) + + s := New(Config{ + Providers: []providers.Provider{p}, + MaxConcurrent: 1, + Log: testLogger(), + Retry: RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, + }) + + if s.retry == nil { + t.Fatal("retry queue should be constructed when Retry.Enabled=true") + } + + event := providers.JobEvent{ + Provider: p, + Action: "queued", + Repo: "myrepo", + JobID: 100, + } + + // handleLocalJob runs synchronously here because handleQueued goes + // through unsee -> sem release -> enqueueRetryIfEligible before + // returning. But it does spawn the wait goroutine on success; on + // failure it returns after enqueue. + s.handleLocalJob(context.Background(), event) + + if got := s.retry.Len(); got != 1 { + t.Errorf("retry queue Len() = %d, want 1 after retryable claim failure", got) + } + // Non-retryable error should NOT enqueue. + p2 := newClaimErrorProvider("github", errors.New("POST .../runners: 404 Not Found")) + s2 := New(Config{ + Providers: []providers.Provider{p2}, + MaxConcurrent: 1, + Log: testLogger(), + Retry: RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, + }) + event2 := providers.JobEvent{ + Provider: p2, Action: "queued", Repo: "myrepo", JobID: 101, + } + s2.handleLocalJob(context.Background(), event2) + if got := s2.retry.Len(); got != 0 { + t.Errorf("non-retryable 404 should not enqueue, got Len() = %d", got) + } +} + +// TestRetryQueue_Integration_CompletedDropsRetry pins the interaction +// with handleCompleted: an outstanding retry for a job that then +// completes elsewhere must be dropped so we stop reattempting. +func TestRetryQueue_Integration_CompletedDropsRetry(t *testing.T) { + clk := newFakeClock(time.Now()) + p := newClaimErrorProvider("github", errors.New("HTTP 500")) + s := New(Config{ + Providers: []providers.Provider{p}, + MaxConcurrent: 1, + Log: testLogger(), + Retry: RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, + }) + + event := providers.JobEvent{ + Provider: p, Action: "queued", Repo: "myrepo", JobID: 200, + } + s.handleLocalJob(context.Background(), event) + if s.retry.Len() != 1 { + t.Fatalf("Len() = %d, want 1", s.retry.Len()) + } + + // Completed webhook arrives for the same job (peer picked it up). + completed := providers.JobEvent{ + Provider: p, Action: "completed", Repo: "myrepo", JobID: 200, + Conclusion: "success", + } + s.handleCompleted(context.Background(), completed) + + if got := s.retry.Len(); got != 0 { + t.Errorf("Len() after completed = %d, want 0 (retry should be dropped)", got) + } +} + +// TestRetryQueue_Integration_Disabled verifies the pre-existing "log +// and drop" behavior is preserved when Retry.Enabled=false. Nothing +// enqueues, no goroutine leaks, no crash. +func TestRetryQueue_Integration_Disabled(t *testing.T) { + p := newClaimErrorProvider("github", errors.New("HTTP 500")) + s := New(Config{ + Providers: []providers.Provider{p}, + MaxConcurrent: 1, + Log: testLogger(), + // Retry left zero-valued - Enabled defaults to false. + }) + + if s.retry != nil { + t.Errorf("retry queue should be nil when Retry.Enabled=false, got %v", s.retry) + } + + event := providers.JobEvent{ + Provider: p, Action: "queued", Repo: "myrepo", JobID: 300, + } + s.handleLocalJob(context.Background(), event) // must not panic + + // enqueueRetryIfEligible is a no-op - no queue exists to check. +} diff --git a/pkg/scheduler/retry_queue.go b/pkg/scheduler/retry_queue.go new file mode 100644 index 0000000..ec69cd0 --- /dev/null +++ b/pkg/scheduler/retry_queue.go @@ -0,0 +1,530 @@ +// Package scheduler - retry_queue.go: in-memory claim/provision retry queue. +// +// GitHub does not re-deliver workflow_job webhooks. When ephemerd's initial +// claim attempt fails for a *retryable* reason (rate-limit exhaustion, +// transient network error, 5xx), the queued job is dropped forever unless +// something re-tries it. Before this queue existed, we lost jobs whenever +// the installation-token budget was exhausted at the moment a webhook +// fired; humans then noticed and re-queued each job by hand. +// +// This file implements a single-goroutine retry queue keyed by jobKey. +// Failed claims are re-scheduled on a jittered backoff ladder +// (30s, 1m, 2m, 5m, 10m by default), giving up after RetryConfig.MaxAge. +// When the GitHub rate snapshot says remaining == 0 with a known reset +// time, the next attempt is snapped to just after reset instead of +// falling blindly through the ladder - this avoids wasting attempts +// on a budget that is provably exhausted. +// +// Non-retryable failures (404 job gone, label mismatch, validation) are +// dropped immediately without enqueue. +// +// Completed webhooks call retryQueue.Drop(key) to cancel outstanding +// retries - otherwise a job that completed on a peer daemon would keep +// getting reattempted here until MaxAge. + +package scheduler + +import ( + "container/heap" + "context" + "errors" + "log/slog" + "math/rand/v2" + "net" + "net/http" + "strings" + "sync" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" + gh "github.com/google/go-github/v72/github" +) + +// RetryConfig tunes the claim retry queue. +// +// A zero-valued RetryConfig disables retries entirely, matching the +// pre-existing "log-and-drop" behavior. In practice New() applies +// sensible defaults so callers get retries by default. +type RetryConfig struct { + // Enabled toggles the entire retry queue. When false, failures still + // log and drop as before (no behavior change). + Enabled bool + + // Schedule is the ordered backoff ladder. Each entry is the base + // delay for that attempt; jitter is applied on top. If nil, defaults + // to {30s, 1m, 2m, 5m, 10m}. + Schedule []time.Duration + + // MaxAge is the wall-clock budget from first failure to giving up. + // Once (now - firstFailure) > MaxAge, we log a WARN and drop. Default 90m. + MaxAge time.Duration + + // Jitter is the fraction (0-1) of a delay that's randomized +/- around + // the base value. Set to a NEGATIVE value (e.g. -1) to request the + // default (0.2 = +/-20%). Literal 0 is honored - tests use it for + // deterministic scheduling. + Jitter float64 + + // RateHint returns the last-observed GitHub rate-limit state. When + // remaining == 0 and now < reset and updated is fresh (<5m old), + // the next attempt is snapped to reset + a small jitter instead of + // the Schedule entry. Nil-safe: nil means "no rate awareness". + RateHint func() (remaining int64, reset time.Time, updated time.Time) + + // Now is the clock function. Defaults to time.Now. Tests inject + // a fake clock so backoff scheduling is deterministic. + Now func() time.Time +} + +// defaultRetrySchedule is the fallback backoff ladder - total ~18m +// before the caller runs out of ladder entries; MaxAge caps overall +// budget at 90m by default (repeating the last entry after that). +var defaultRetrySchedule = []time.Duration{ + 30 * time.Second, + 1 * time.Minute, + 2 * time.Minute, + 5 * time.Minute, + 10 * time.Minute, +} + +// applyDefaults fills in zero fields with sensible defaults. Called once +// by newRetryQueue. +func (c *RetryConfig) applyDefaults() { + if len(c.Schedule) == 0 { + c.Schedule = defaultRetrySchedule + } + if c.MaxAge <= 0 { + c.MaxAge = 90 * time.Minute + } + // Jitter defaults only apply when explicitly requested via a + // negative value (sentinel); literal 0 disables jitter. Values > 1 + // are clamped to 0.2 as a safety net so a misconfigured knob doesn't + // produce nonsense delays. + switch { + case c.Jitter < 0: + c.Jitter = 0.2 + case c.Jitter > 1: + c.Jitter = 0.2 + } + if c.Now == nil { + c.Now = time.Now + } +} + +// retryHandler is the callback invoked when a retry fires. Returns nil +// on success (drops the retry). Any error triggers another schedule +// step, unless classifyErr says it's non-retryable. +type retryHandler func(ctx context.Context, event providers.JobEvent) error + +// retryItem is an entry in the priority queue. +type retryItem struct { + key jobKey + event providers.JobEvent + handler retryHandler + attempts int // number of prior failed attempts + firstFailure time.Time // when the FIRST attempt failed (for MaxAge) + nextAttempt time.Time // when to fire next + index int // heap index; -1 when not in the heap +} + +// retryHeap implements heap.Interface, ordered by nextAttempt ascending. +type retryHeap []*retryItem + +func (h retryHeap) Len() int { return len(h) } +func (h retryHeap) Less(i, j int) bool { return h[i].nextAttempt.Before(h[j].nextAttempt) } +func (h retryHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i]; h[i].index = i; h[j].index = j } +func (h *retryHeap) Push(x any) { + it := x.(*retryItem) + it.index = len(*h) + *h = append(*h, it) +} +func (h *retryHeap) Pop() any { + old := *h + n := len(old) + it := old[n-1] + old[n-1] = nil + it.index = -1 + *h = old[:n-1] + return it +} + +// retryQueue is a single-goroutine, priority-heap-driven retry loop. +type retryQueue struct { + cfg RetryConfig + log *slog.Logger + + mu sync.Mutex + heap retryHeap + index map[jobKey]*retryItem // for O(1) Drop / update + + wake chan struct{} // signals the loop that heap head changed +} + +// newRetryQueue constructs a queue. Call Run to start the drain loop. +func newRetryQueue(cfg RetryConfig, log *slog.Logger) *retryQueue { + cfg.applyDefaults() + return &retryQueue{ + cfg: cfg, + log: log, + index: make(map[jobKey]*retryItem), + wake: make(chan struct{}, 1), + } +} + +// Enabled reports whether the queue will actually schedule retries. +func (q *retryQueue) Enabled() bool { return q != nil && q.cfg.Enabled } + +// signal wakes the drain loop. Non-blocking; a coalesced wake is fine. +func (q *retryQueue) signal() { + select { + case q.wake <- struct{}{}: + default: + } +} + +// Add enqueues (or refreshes) a retry for the given event. Called by +// the scheduler when a claim/provision path fails with a retryable +// error. If the queue is disabled or the error is non-retryable, this +// is a no-op after logging. +func (q *retryQueue) Add(event providers.JobEvent, handler retryHandler, err error) { + if q == nil || !q.cfg.Enabled { + return + } + class := classifyErr(err) + if class == errNonRetryable { + q.log.Debug("retry queue: not enqueuing non-retryable error", + "job_id", event.JobID, "provider", providerName(event), "error", err) + return + } + + key := keyFor(event) + now := q.cfg.Now() + + q.mu.Lock() + defer q.mu.Unlock() + + it, existed := q.index[key] + if !existed { + it = &retryItem{ + key: key, + event: event, + handler: handler, + firstFailure: now, + index: -1, + } + q.index[key] = it + } + it.attempts++ + + // Give-up check: too much wall-clock time has passed since the + // first failure. + if now.Sub(it.firstFailure) > q.cfg.MaxAge { + q.log.Warn("retry queue: giving up on job - max age exceeded", + "job_id", event.JobID, + "provider", providerName(event), + "attempts", it.attempts, + "age", now.Sub(it.firstFailure)) + q.removeLocked(it) + return + } + + // Pick the delay: rate-aware if applicable, otherwise ladder + jitter. + delay := q.nextDelayLocked(it, class) + it.nextAttempt = now.Add(delay) + + if it.index < 0 { + heap.Push(&q.heap, it) + } else { + heap.Fix(&q.heap, it.index) + } + q.signal() + + q.log.Info("retry queue: scheduled retry", + "job_id", event.JobID, + "provider", providerName(event), + "attempt", it.attempts, + "delay", delay, + "error_class", class) +} + +// Drop removes any pending retry for the given key. Called from +// handleCompleted / on successful claim / on in_progress webhook so we +// don't keep retrying jobs that were picked up elsewhere. +func (q *retryQueue) Drop(key jobKey) { + if q == nil { + return + } + q.mu.Lock() + defer q.mu.Unlock() + if it, ok := q.index[key]; ok { + q.removeLocked(it) + } +} + +// Len reports the current queue depth (for metrics and tests). +func (q *retryQueue) Len() int { + if q == nil { + return 0 + } + q.mu.Lock() + defer q.mu.Unlock() + return len(q.heap) +} + +// removeLocked pulls an item out of both heap and index. Caller holds mu. +func (q *retryQueue) removeLocked(it *retryItem) { + if it.index >= 0 { + heap.Remove(&q.heap, it.index) + } + delete(q.index, it.key) +} + +// nextDelayLocked returns the delay until the next attempt, jitter +// applied and (optionally) snapped to a rate-limit reset. Caller holds mu. +func (q *retryQueue) nextDelayLocked(it *retryItem, class errClass) time.Duration { + // Rate-limit awareness: only when we're actually retrying because of + // a rate error AND the hint is fresh. Otherwise a 5xx spike would + // pile up behind a distant reset, which is not what we want. + if class == errRateLimit && q.cfg.RateHint != nil { + remaining, reset, updated := q.cfg.RateHint() + now := q.cfg.Now() + if remaining == 0 && !reset.IsZero() && reset.After(now) && + !updated.IsZero() && now.Sub(updated) < 5*time.Minute { + // Sleep until just after reset, with a small [0, 20s) + // jitter so multiple daemons sharing the token don't + // stampede at the exact same second. + jitterMs := int64(20_000) + off := time.Duration(rand.Int64N(jitterMs)) * time.Millisecond + return reset.Sub(now) + 5*time.Second + off + } + } + + // Ladder + jitter. attempts is 1 on the first retry; index 0 into + // the schedule is that first delay. + idx := it.attempts - 1 + if idx < 0 { + idx = 0 + } + if idx >= len(q.cfg.Schedule) { + idx = len(q.cfg.Schedule) - 1 + } + base := q.cfg.Schedule[idx] + return jittered(base, q.cfg.Jitter) +} + +// jittered applies +/-frac jitter to base. Split out for testability. +func jittered(base time.Duration, frac float64) time.Duration { + if frac <= 0 { + return base + } + span := float64(base) * frac + // Range: [-span, +span). + off := (rand.Float64()*2 - 1) * span + d := time.Duration(float64(base) + off) + if d < 0 { + d = 0 + } + return d +} + +// Run drives the drain loop until ctx is cancelled. Safe to call once +// per queue. If the queue is disabled, Run just blocks on ctx. +func (q *retryQueue) Run(ctx context.Context) { + if q == nil || !q.cfg.Enabled { + <-ctx.Done() + return + } + + // A single timer we re-arm every iteration. Nil means "no head". + var timer *time.Timer + stopTimer := func() { + if timer != nil { + timer.Stop() + timer = nil + } + } + defer stopTimer() + + for { + q.mu.Lock() + var next time.Duration = -1 + if len(q.heap) > 0 { + d := q.heap[0].nextAttempt.Sub(q.cfg.Now()) + if d < 0 { + d = 0 + } + next = d + } + q.mu.Unlock() + + stopTimer() + if next >= 0 { + timer = time.NewTimer(next) + } + + var timerCh <-chan time.Time + if timer != nil { + timerCh = timer.C + } + + select { + case <-ctx.Done(): + return + case <-q.wake: + // Heap changed; loop and re-check head. + continue + case <-timerCh: + q.fireDue(ctx) + } + } +} + +// fireDue pops all items whose nextAttempt is <= now and runs their +// handlers. Each handler runs in its own goroutine so a slow handler +// doesn't block the timer loop; on failure the handler is expected to +// call Add again with the new error. +func (q *retryQueue) fireDue(ctx context.Context) { + now := q.cfg.Now() + var due []*retryItem + + q.mu.Lock() + for len(q.heap) > 0 && !q.heap[0].nextAttempt.After(now) { + it := heap.Pop(&q.heap).(*retryItem) + delete(q.index, it.key) + due = append(due, it) + } + q.mu.Unlock() + + for _, it := range due { + go q.runOne(ctx, it) + } +} + +// runOne invokes the handler once, and on failure re-adds via Add so the +// backoff ladder advances. On success we do nothing - the handler took +// over lifecycle tracking and any future completed webhook will Drop +// the (already-absent) key harmlessly. +func (q *retryQueue) runOne(ctx context.Context, it *retryItem) { + q.log.Info("retry queue: firing attempt", + "job_id", it.event.JobID, + "provider", providerName(it.event), + "attempt", it.attempts+1) + err := it.handler(ctx, it.event) + if err == nil { + return + } + // Re-enqueue on failure. Add() decides retry/give-up. + q.Add(it.event, it.handler, err) +} + +// --- error classification --- + +type errClass int + +const ( + errNonRetryable errClass = iota + errRateLimit // 429 / 403 with rate headers + errServerSide // 5xx + errNetwork // network/DNS/timeout + errUnknownRetryable // last-resort: probably safe to retry +) + +func (c errClass) String() string { + switch c { + case errNonRetryable: + return "non_retryable" + case errRateLimit: + return "rate_limit" + case errServerSide: + return "server_side" + case errNetwork: + return "network" + case errUnknownRetryable: + return "unknown_retryable" + default: + return "unknown" + } +} + +// classifyErr sorts a claim/provision error into a retry class. Kept +// public within the package so scheduler.go can decide whether to log +// vs enqueue even before calling Add. +func classifyErr(err error) errClass { + if err == nil { + return errNonRetryable + } + + // GitHub-specific rate limit errors - these are ALWAYS retryable. + var rle *gh.RateLimitError + if errors.As(err, &rle) { + return errRateLimit + } + var arle *gh.AbuseRateLimitError + if errors.As(err, &arle) { + return errRateLimit + } + + // GitHub ErrorResponse: status-code driven. + var ghErr *gh.ErrorResponse + if errors.As(err, &ghErr) && ghErr.Response != nil { + switch ghErr.Response.StatusCode { + case http.StatusNotFound, http.StatusUnprocessableEntity, + http.StatusUnauthorized, http.StatusBadRequest, + http.StatusForbidden: + // 403 without RateLimitError = permission problem, not rate. + // Rate-limit 403s are already caught by the RateLimitError + // branch above. + // 404 job gone: definitely don't retry. + // 401/400/422 validation: fixing requires config change, + // not time. + // (Conflict 409 is handled inside claimJob by name-retry.) + return errNonRetryable + case http.StatusTooManyRequests: + return errRateLimit + } + if ghErr.Response.StatusCode >= 500 { + return errServerSide + } + } + + // Bare network errors - check via net.Error interface. Timeout() + // alone doesn't cover DNS failures, so we also check the string. + var netErr net.Error + if errors.As(err, &netErr) { + return errNetwork + } + + // Fallback string-match heuristics for wrapped errors that don't + // surface via errors.As (some HTTP client wrappers eat types). + msg := strings.ToLower(err.Error()) + switch { + case strings.Contains(msg, "429") || strings.Contains(msg, "rate limit") || + strings.Contains(msg, "secondary rate limit") || strings.Contains(msg, "abuse"): + return errRateLimit + case strings.Contains(msg, "404") || strings.Contains(msg, "not found") || + strings.Contains(msg, "422") || strings.Contains(msg, "validation") || + strings.Contains(msg, "401") || strings.Contains(msg, "unauthorized") || + strings.Contains(msg, "permission denied"): + return errNonRetryable + case strings.Contains(msg, "500") || strings.Contains(msg, "502") || + strings.Contains(msg, "503") || strings.Contains(msg, "504") || + strings.Contains(msg, "bad gateway") || strings.Contains(msg, "gateway timeout"): + return errServerSide + case strings.Contains(msg, "timeout") || strings.Contains(msg, "no such host") || + strings.Contains(msg, "connection refused") || strings.Contains(msg, "eof"): + return errNetwork + } + + // Unknown - treat as retryable but only within backoff/MaxAge. The + // alternative (drop) is worse: the whole point of this queue is to + // stop losing jobs to unexpected transient failures. + return errUnknownRetryable +} + +// providerName is a nil-safe accessor for logging. +func providerName(ev providers.JobEvent) string { + if ev.Provider == nil { + return "" + } + return ev.Provider.Name() +} diff --git a/pkg/scheduler/retry_queue_test.go b/pkg/scheduler/retry_queue_test.go new file mode 100644 index 0000000..a30e1c9 --- /dev/null +++ b/pkg/scheduler/retry_queue_test.go @@ -0,0 +1,363 @@ +package scheduler + +import ( + "context" + "errors" + "net" + "net/http" + "net/url" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" + gh "github.com/google/go-github/v72/github" +) + +// fakeClock is a minimal deterministic clock. The retry queue reads it +// via cfg.Now; we don't need to plug into a full clock interface because +// the queue's only time.NewTimer usage is real-time. Tests that need to +// exercise the timer directly call fireDue instead of waiting on it. +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func newFakeClock(t time.Time) *fakeClock { return &fakeClock{now: t} } + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + c.now = c.now.Add(d) + c.mu.Unlock() +} + +func mkEvent(id int64) providers.JobEvent { + return providers.JobEvent{ + Provider: newMockProvider("github"), + Action: "queued", + Repo: "myrepo", + JobID: id, + } +} + +// TestRetryQueue_BackoffLadder pins the schedule progression. Uses a +// fixed schedule and zero jitter so the delays are exact. +func TestRetryQueue_BackoffLadder(t *testing.T) { + clk := newFakeClock(time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)) + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second, 1 * time.Minute, 2 * time.Minute}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(1) + handler := func(_ context.Context, _ providers.JobEvent) error { + return errors.New("HTTP 500 server error") + } + + // Attempt 1 (first failure): should schedule 30s out. + q.Add(ev, handler, errors.New("HTTP 500 server error")) + it := q.index[keyFor(ev)] + if it == nil { + t.Fatal("expected item in index after Add") + } + if got := it.nextAttempt.Sub(clk.Now()); got != 30*time.Second { + t.Errorf("attempt 1 delay = %v, want 30s", got) + } + if it.attempts != 1 { + t.Errorf("attempts = %d, want 1", it.attempts) + } + + // Simulate the first retry firing and failing again. + q.Add(ev, handler, errors.New("HTTP 500 server error")) + if got := it.nextAttempt.Sub(clk.Now()); got != 1*time.Minute { + t.Errorf("attempt 2 delay = %v, want 1m", got) + } + + // Third failure. + q.Add(ev, handler, errors.New("HTTP 500 server error")) + if got := it.nextAttempt.Sub(clk.Now()); got != 2*time.Minute { + t.Errorf("attempt 3 delay = %v, want 2m", got) + } + + // Fourth failure: beyond schedule length - clamp to last entry. + q.Add(ev, handler, errors.New("HTTP 500 server error")) + if got := it.nextAttempt.Sub(clk.Now()); got != 2*time.Minute { + t.Errorf("attempt 4 (past ladder) delay = %v, want 2m (clamped)", got) + } +} + +// TestRetryQueue_GiveUpOnMaxAge verifies the queue drops an item once +// the age since first failure exceeds MaxAge. +func TestRetryQueue_GiveUpOnMaxAge(t *testing.T) { + clk := newFakeClock(time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)) + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 5 * time.Minute, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(2) + handler := func(_ context.Context, _ providers.JobEvent) error { + return errors.New("HTTP 500") + } + + q.Add(ev, handler, errors.New("HTTP 500")) + if _, ok := q.index[keyFor(ev)]; !ok { + t.Fatal("expected item after first Add") + } + + // Advance past MaxAge and Add again - should drop. + clk.Advance(6 * time.Minute) + q.Add(ev, handler, errors.New("HTTP 500")) + if _, ok := q.index[keyFor(ev)]; ok { + t.Error("expected item dropped after MaxAge exceeded, still present") + } + if q.Len() != 0 { + t.Errorf("Len() = %d, want 0 after give-up", q.Len()) + } +} + +// TestRetryQueue_DropOnCompletion verifies Drop removes an outstanding +// retry - the completed-webhook path. +func TestRetryQueue_DropOnCompletion(t *testing.T) { + clk := newFakeClock(time.Now()) + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(3) + handler := func(_ context.Context, _ providers.JobEvent) error { return errors.New("500") } + q.Add(ev, handler, errors.New("500")) + + if q.Len() != 1 { + t.Fatalf("Len() = %d, want 1", q.Len()) + } + + q.Drop(keyFor(ev)) + + if q.Len() != 0 { + t.Errorf("Len() = %d, want 0 after Drop", q.Len()) + } + if _, ok := q.index[keyFor(ev)]; ok { + t.Error("item still in index after Drop") + } +} + +// TestRetryQueue_NonRetryableErrorNotEnqueued pins that a 404 does not +// enter the queue. +func TestRetryQueue_NonRetryableErrorNotEnqueued(t *testing.T) { + clk := newFakeClock(time.Now()) + q := newRetryQueue(RetryConfig{ + Enabled: true, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(4) + handler := func(_ context.Context, _ providers.JobEvent) error { return nil } + + // 404 via string-match fallback. + q.Add(ev, handler, errors.New("GET .../jobs/999: 404 Not Found")) + if q.Len() != 0 { + t.Errorf("404 should not enqueue, Len() = %d", q.Len()) + } + + // 422 validation. + q.Add(ev, handler, errors.New("POST .../runners: 422 Unprocessable Entity")) + if q.Len() != 0 { + t.Errorf("422 should not enqueue, Len() = %d", q.Len()) + } +} + +// TestRetryQueue_RateAwareSchedulesAfterReset verifies that when the +// rate hint says remaining=0 with a known reset, the next attempt is +// pushed to just after reset instead of the normal ladder delay. +func TestRetryQueue_RateAwareSchedulesAfterReset(t *testing.T) { + base := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC) + clk := newFakeClock(base) + reset := base.Add(45 * time.Minute) // fresh, in the future + updated := base.Add(-30 * time.Second) + + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, // ladder would say 30s + Jitter: 0, + MaxAge: 2 * time.Hour, + Now: clk.Now, + RateHint: func() (int64, time.Time, time.Time) { + return 0, reset, updated + }, + }, testLogger()) + + ev := mkEvent(5) + handler := func(_ context.Context, _ providers.JobEvent) error { return nil } + + // Feed a RateLimitError so class == errRateLimit. + rle := &gh.RateLimitError{ + Rate: gh.Rate{Remaining: 0}, + Response: &http.Response{StatusCode: http.StatusForbidden, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + Message: "rate limit exceeded", + } + q.Add(ev, handler, rle) + + it := q.index[keyFor(ev)] + if it == nil { + t.Fatal("expected item after Add") + } + delay := it.nextAttempt.Sub(clk.Now()) + // Should be ~45m + 5s + [0, 20s) jitter. Definitely much bigger than + // the 30s ladder value. + minWant := 45*time.Minute + 5*time.Second + maxWant := 45*time.Minute + 5*time.Second + 20*time.Second + if delay < minWant || delay >= maxWant { + t.Errorf("rate-aware delay = %v, want in [%v, %v)", delay, minWant, maxWant) + } +} + +// TestRetryQueue_RateAware_StaleHintIgnored verifies we DON'T snap to +// reset if the rate snapshot is older than 5 minutes - the reset time +// may itself be stale. +func TestRetryQueue_RateAware_StaleHintIgnored(t *testing.T) { + base := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC) + clk := newFakeClock(base) + reset := base.Add(45 * time.Minute) + updated := base.Add(-10 * time.Minute) // stale + + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 2 * time.Hour, + Now: clk.Now, + RateHint: func() (int64, time.Time, time.Time) { + return 0, reset, updated + }, + }, testLogger()) + + ev := mkEvent(6) + handler := func(_ context.Context, _ providers.JobEvent) error { return nil } + q.Add(ev, handler, &gh.RateLimitError{ + Response: &http.Response{StatusCode: http.StatusForbidden, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }) + + it := q.index[keyFor(ev)] + if got := it.nextAttempt.Sub(clk.Now()); got != 30*time.Second { + t.Errorf("stale rate hint should fall through to ladder, got delay %v want 30s", got) + } +} + +// TestClassifyErr is a table-driven check of the classification logic. +func TestClassifyErr(t *testing.T) { + cases := []struct { + name string + err error + want errClass + }{ + {"nil", nil, errNonRetryable}, + {"rate_limit_typed", &gh.RateLimitError{ + Response: &http.Response{StatusCode: 403, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errRateLimit}, + {"abuse_typed", &gh.AbuseRateLimitError{ + Response: &http.Response{StatusCode: 403, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errRateLimit}, + {"404_typed", &gh.ErrorResponse{ + Response: &http.Response{StatusCode: 404, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errNonRetryable}, + {"500_typed", &gh.ErrorResponse{ + Response: &http.Response{StatusCode: 500, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errServerSide}, + {"429_typed", &gh.ErrorResponse{ + Response: &http.Response{StatusCode: 429, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errRateLimit}, + {"422_typed", &gh.ErrorResponse{ + Response: &http.Response{StatusCode: 422, Request: &http.Request{URL: &url.URL{Path: "/x"}}}, + }, errNonRetryable}, + {"net_timeout", &net.OpError{Op: "dial", Err: errors.New("i/o timeout")}, errNetwork}, + {"str_500", errors.New("wrapped: 500 internal server error"), errServerSide}, + {"str_502", errors.New("bad gateway 502"), errServerSide}, + {"str_rate", errors.New("secondary rate limit exceeded"), errRateLimit}, + {"str_404", errors.New("registering JIT runner: 404 Not Found"), errNonRetryable}, + {"str_permission", errors.New("permission denied"), errNonRetryable}, + {"str_timeout", errors.New("context deadline exceeded (Client.Timeout)"), errNetwork}, + {"str_unknown", errors.New("something weird happened"), errUnknownRetryable}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := classifyErr(tc.err); got != tc.want { + t.Errorf("classifyErr(%v) = %s, want %s", tc.err, got, tc.want) + } + }) + } +} + +// TestRetryQueue_FireDueRunsHandler drives the drain path directly. +// We add an item whose nextAttempt is already in the past (attempt=0 +// via a manual poke) and call fireDue. +func TestRetryQueue_FireDueRunsHandler(t *testing.T) { + clk := newFakeClock(time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC)) + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second}, + Jitter: 0, + MaxAge: 1 * time.Hour, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(7) + var calls atomic.Int32 + done := make(chan struct{}) + handler := func(_ context.Context, _ providers.JobEvent) error { + if calls.Add(1) == 1 { + close(done) + return nil // success first fire - no re-enqueue + } + return nil + } + + q.Add(ev, handler, errors.New("HTTP 500")) + // Advance the clock past the scheduled attempt so fireDue picks it up. + clk.Advance(1 * time.Minute) + q.fireDue(context.Background()) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("handler never fired within 2s") + } + // Wait a moment for the goroutine to complete so we can safely + // inspect Len(). runOne exits immediately on nil err. + time.Sleep(50 * time.Millisecond) + if q.Len() != 0 { + t.Errorf("Len() = %d, want 0 after successful handler", q.Len()) + } +} + +// TestRetryQueue_Disabled_IsNoOp verifies that when Enabled=false, +// Add is a no-op - the pre-existing "log and drop" behavior is preserved. +func TestRetryQueue_Disabled_IsNoOp(t *testing.T) { + q := newRetryQueue(RetryConfig{Enabled: false}, testLogger()) + ev := mkEvent(8) + q.Add(ev, func(_ context.Context, _ providers.JobEvent) error { return nil }, + errors.New("HTTP 500")) + if q.Len() != 0 { + t.Errorf("disabled queue Len() = %d, want 0", q.Len()) + } + if q.Enabled() { + t.Error("Enabled() = true, want false") + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 5379c27..4324007 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -46,6 +46,14 @@ type Config struct { ShutdownTimeout time.Duration LogRetention time.Duration // max age for job log files (default 7d) + // Retry configures the claim/provision retry queue. When the initial + // attempt to claim a queued job fails with a retryable error + // (rate-limit exhausted, transient 5xx, network), the job is + // enqueued and re-attempted on a backoff ladder rather than lost. + // GitHub does not re-deliver workflow_job webhooks. Leave zero-valued + // (Enabled=false) to keep the pre-existing "log and drop" behavior. + Retry RetryConfig + // RunnerImageForRepo resolves the per-repo, per-OS image override // configured under [runner.images]. Returns "" when no override is // set; the scheduler then falls back to the provider per-OS default @@ -108,6 +116,10 @@ type Scheduler struct { macSem chan struct{} // macOS VM concurrency limiter (Vz has a hard cap) draining bool // true when shutting down, rejects new jobs startTime time.Time + + // retry holds pending re-attempts for jobs whose initial claim + // failed with a retryable error. Nil when Config.Retry.Enabled=false. + retry *retryQueue } const seenTTL = 10 * time.Minute @@ -189,7 +201,7 @@ func New(cfg Config) *Scheduler { } } - return &Scheduler{ + s := &Scheduler{ cfg: cfg, running: make(map[jobKey]*runningJob), seen: make(map[jobKey]time.Time), @@ -199,6 +211,18 @@ func New(cfg Config) *Scheduler { macSem: make(chan struct{}, macVMs), startTime: time.Now(), } + // Only construct the retry queue when the caller explicitly enabled + // it. A disabled queue is safe to leave nil; enqueueRetryIfEligible + // nil-checks so the "log and drop" path is a no-op for opted-out + // callers. + if cfg.Retry.Enabled { + log := cfg.Log + if log == nil { + log = slog.Default() + } + s.retry = newRetryQueue(cfg.Retry, log.With("component", "retry_queue")) + } + return s } // Run starts the scheduler. It discovers jobs via polling (default) or @@ -223,6 +247,12 @@ func (s *Scheduler) Run(ctx context.Context) error { } }() + // Drive the claim/provision retry queue if configured. Nil-safe when + // Retry.Enabled is false. + if s.retry != nil { + go s.retry.Run(ctx) + } + // Clean old job logs on startup, then periodically every hour. // Retention period is configurable via [log] log_retention (default 7d). logDir := filepath.Join(s.cfg.DataDir, "logs") @@ -427,6 +457,14 @@ func (s *Scheduler) Run(ctx context.Context) error { case "queued": metrics.JobsQueuedTotal.Inc() go s.handleQueued(ctx, event) + case "in_progress": + // Job was picked up (possibly by a peer daemon + // sharing the installation token). Drop any + // outstanding retry so we stop reattempting a job + // that's already running elsewhere. Nil-safe. + if s.retry != nil { + s.retry.Drop(keyFor(event)) + } case "completed": go s.handleCompleted(ctx, event) } @@ -615,10 +653,15 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent const maxNameRetries = 3 claim, err := s.claimJob(ctx, &event, labels, log, maxNameRetries) if err != nil { - log.Error("failed to claim job", "error", err) + log.Error("failed to claim job", "error", err, "error_class", classifyErr(err)) unsee() - time.Sleep(backoffDuration(event.Repo)) <-s.linuxSem + // Replaces the old blind time.Sleep(backoffDuration): the + // sem is released FIRST so we do not hold a slot idle across + // the wait, then a rate-aware jittered retry is enqueued + // (no-op when Retry.Enabled is false or the error is not + // retryable). + s.enqueueRetryIfEligible(event, err) return } @@ -744,13 +787,13 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent const maxNameRetries = 3 claim, err := s.claimJob(ctx, &event, labels, log, maxNameRetries) if err != nil { - log.Error("failed to claim job", "error", err) + log.Error("failed to claim job", "error", err, "error_class", classifyErr(err)) if artifactsDir != "" { artifacts.Cleanup(artifactsDir, s.cfg.Log) } unsee() - time.Sleep(backoffDuration(event.Repo)) <-s.macSem + s.enqueueRetryIfEligible(event, err) return } @@ -961,13 +1004,16 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent const maxNameRetries = 3 claim, err := s.claimJob(ctx, &event, labels, log, maxNameRetries) if err != nil { - log.Error("failed to claim job", "error", err) + log.Error("failed to claim job", "error", err, "error_class", classifyErr(err)) if artifactsDir != "" { artifacts.Cleanup(artifactsDir, s.cfg.Log) } unsee() - time.Sleep(5 * time.Second) // back off to avoid tight retry loops on rate limits <-s.sem + // Replaces the old blind 5s sleep with a rate-aware jittered + // retry. No-op when Retry is disabled or the error is not + // retryable. + s.enqueueRetryIfEligible(event, err) return } @@ -1066,6 +1112,13 @@ func (s *Scheduler) handleCompleted(ctx context.Context, event providers.JobEven key := keyFor(event) log := s.cfg.Log.With("job_id", jobID, "repo", event.Repo) + // Drop any outstanding retry attempts: the provider says this job + // is finished, so re-attempting would waste API budget and could + // register ghost runners. Nil-safe. + if s.retry != nil { + s.retry.Drop(key) + } + s.mu.Lock() job, exists := s.running[key] if exists { @@ -1219,6 +1272,40 @@ func (s *Scheduler) cleanSeen() { } } +// enqueueRetryIfEligible passes err through the retry queue, if enabled. +// The retryHandler callback re-invokes handleQueued with the ORIGINAL +// event when the backoff timer fires. Non-retryable errors and disabled +// queues are a no-op. Safe to call with s.retry == nil. +func (s *Scheduler) enqueueRetryIfEligible(event providers.JobEvent, err error) { + if s.retry == nil { + return + } + s.retry.Add(event, s.retryHandler, err) +} + +// retryHandler is the callback the retry queue invokes on each fire. +// It re-enters the top-level dispatch (handleQueued) with the ORIGINAL +// event. Because handleQueued would otherwise dedup our own retry via +// the seen/pending maps, we clear those entries first. +// +// Return value: always nil. handleQueued dispatches asynchronously into +// concurrency slots so we cannot know synchronously whether the retry +// succeeded; on failure the handler self-enqueues via +// enqueueRetryIfEligible, and on success any future completed webhook +// harmlessly Drops the (already-absent) key. +func (s *Scheduler) retryHandler(ctx context.Context, event providers.JobEvent) error { + key := keyFor(event) + s.mu.Lock() + // Clear seen/pending so handleQueued does not dedup our own retry. + // Leave running alone: if the job got picked up elsewhere, we + // want handleQueued's running-check to short-circuit. + delete(s.seen, key) + delete(s.pending, key) + s.mu.Unlock() + s.handleQueued(ctx, event) + return nil +} + // buildLabelsForOS builds runner labels for a given target OS. // Used by the dispatcher to register Linux runners from the Windows host. func buildLabelsForOS(targetOS string, extraLabels []string) []string {