From d4e150cab4aab658de044ca8a474d9c41585db18 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:23:30 -0700 Subject: [PATCH 1/5] 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 2bf834c..bc32409 100644 --- a/pkg/scheduler/handle_queued_test.go +++ b/pkg/scheduler/handle_queued_test.go @@ -163,7 +163,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{ @@ -193,9 +199,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 b236830..0d97404 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -532,13 +532,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() @@ -555,6 +554,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 949325314ac6a2c06fd6f27b39d0744ece9b1348 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:25:22 -0700 Subject: [PATCH 2/5] 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 0f48ddb5d50c75060e4016665e5c8d1f4c11f385 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 15:42:45 -0700 Subject: [PATCH 3/5] 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 7f4d921..43e45f2 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -580,6 +580,26 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP log.Info("webhook tunnel configured", "provider", cfg.Webhook.Tunnel) } + // 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, @@ -605,6 +625,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP MacOSModeForRepo: cfg.Runner.MacOS.ModeForRepo, NativeMacUser: cfg.Runner.MacOS.User, RunnerDir: rm.Dir(), + Retry: retryCfg, Log: log, }) @@ -803,6 +824,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 824b6ae..6b2257c 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -461,6 +461,14 @@ type RunnerConfig struct { ShutdownTimeout string `toml:"shutdown_timeout"` Windows WindowsRunnerToml `toml:"windows"` MacOS MacOSRunnerConfig `toml:"macos"` + + // 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"` } // MacOSRunnerConfig controls macOS job routing. It lives under [runner] @@ -541,6 +549,34 @@ func (m *MacOSRunnerConfig) ResolvedMaxNative() int { return m.MaxNative } +// 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 // runner containers. Without limits Hyper-V containers default to ~1 GB RAM, // which is too small for MSVC + parallel cl.exe builds. @@ -591,6 +627,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 0d97404..bc1f5d2 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -47,6 +47,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 @@ -116,6 +124,10 @@ type Scheduler struct { nativeMacSem chan struct{} // native macOS job concurrency limiter (separate from VM limit) 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 @@ -216,7 +228,7 @@ func New(cfg Config) *Scheduler { nativeMac = 4 } - return &Scheduler{ + s := &Scheduler{ cfg: cfg, running: make(map[jobKey]*runningJob), seen: make(map[jobKey]time.Time), @@ -228,6 +240,18 @@ func New(cfg Config) *Scheduler { nativeMacSem: make(chan struct{}, nativeMac), 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 @@ -252,6 +276,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") @@ -456,6 +486,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) } @@ -671,10 +709,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 } @@ -800,13 +843,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 } @@ -1142,13 +1185,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 } @@ -1247,6 +1293,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 { @@ -1407,6 +1460,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 { From 2e10de3287db822f8c9682e1e7f4d11c9f468a42 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 21:10:48 -0700 Subject: [PATCH 4/5] fix(scheduler): advance retry ladder through the real fire path fireDue popped a due item and delete()d it from q.index before spawning runOne. On handler failure runOne re-enqueued via Add, which found an empty index and rebuilt a FRESH item (attempts reset to 1, firstFailure reset to now). The backoff ladder therefore never advanced past schedule[0] (~30s forever) and the MaxAge give-up never triggered. Fix: fireDue no longer deletes the popped item from q.index. Pop already sets it.index = -1, so the item is out of the heap but stays registered; the re-Add in runOne now hits the existed branch, incrementing attempts and preserving firstFailure so the ladder advances and give-up fires. On handler SUCCESS runOne calls Drop(key) to clean up the lingering index entry (removeLocked tolerates index < 0, so a concurrent Drop is a no-op). Add TestRetryQueue_LadderAdvancesThroughFirePath, which drives fireDue -> runOne -> re-Add with an always-failing handler and asserts the delays grow 30s -> 1m -> 2m and that the item is dropped once age exceeds MaxAge. It fails against the un-fixed code (ladder pinned at 30s). --- pkg/scheduler/retry_queue.go | 22 ++++- pkg/scheduler/retry_queue_test.go | 129 ++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 4 deletions(-) diff --git a/pkg/scheduler/retry_queue.go b/pkg/scheduler/retry_queue.go index ec69cd0..aa3e75e 100644 --- a/pkg/scheduler/retry_queue.go +++ b/pkg/scheduler/retry_queue.go @@ -390,7 +390,14 @@ func (q *retryQueue) fireDue(ctx context.Context) { 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) + // Intentionally DO NOT delete(q.index, it.key) here. Popping + // already set it.index = -1 (see retryHeap.Pop), so the item is + // out of the heap but stays registered in the index. That way a + // re-Add from runOne (on handler failure) hits the `existed` + // branch: it increments attempts and preserves firstFailure so + // the backoff ladder advances and MaxAge give-up eventually + // fires. runOne calls Drop on success to clean up the lingering + // index entry. due = append(due, it) } q.mu.Unlock() @@ -401,9 +408,10 @@ func (q *retryQueue) fireDue(ctx context.Context) { } // 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. +// backoff ladder advances. On success it Drops the item's lingering index +// entry (fireDue leaves it registered so a failure re-Add can advance the +// ladder); the handler has taken over lifecycle tracking, so 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, @@ -411,6 +419,12 @@ func (q *retryQueue) runOne(ctx context.Context, it *retryItem) { "attempt", it.attempts+1) err := it.handler(ctx, it.event) if err == nil { + // fireDue intentionally leaves the popped item in q.index (so a + // failure re-Add advances the ladder). On success nothing will + // re-Add it, so drop the lingering index entry here to avoid a + // map leak. removeLocked tolerates index < 0 (the item is not in + // the heap), so a concurrent Drop that already fired is a no-op. + q.Drop(it.key) return } // Re-enqueue on failure. Add() decides retry/give-up. diff --git a/pkg/scheduler/retry_queue_test.go b/pkg/scheduler/retry_queue_test.go index a30e1c9..9958bc4 100644 --- a/pkg/scheduler/retry_queue_test.go +++ b/pkg/scheduler/retry_queue_test.go @@ -347,6 +347,135 @@ func TestRetryQueue_FireDueRunsHandler(t *testing.T) { } } +// TestRetryQueue_LadderAdvancesThroughFirePath is the regression test for +// the bug where fireDue deleted the popped item from q.index, so the +// re-Add in runOne always hit the `!existed` branch and rebuilt a FRESH +// item (attempts reset to 1, firstFailure reset to now). The observable +// symptoms were: the backoff ladder never advanced past schedule[0] +// (~30s forever) and the MaxAge give-up never triggered. +// +// This test drives the REAL fire path (fireDue -> runOne -> re-Add) with +// an always-failing handler and asserts: +// 1. The re-schedule delay grows 30s -> 1m -> 2m across successive fires. +// 2. attempts increments monotonically and firstFailure is preserved. +// 3. The item is dropped once its age exceeds MaxAge. +// +// Against the un-fixed code the delay stays pinned at 30s and the item +// never gives up, so both the ladder and give-up assertions fail. +func TestRetryQueue_LadderAdvancesThroughFirePath(t *testing.T) { + base := time.Date(2026, 7, 8, 12, 0, 0, 0, time.UTC) + clk := newFakeClock(base) + // MaxAge chosen so the ladder (30s,1m,2m clamped) runs several rungs + // before we deliberately blow past it at the end. + q := newRetryQueue(RetryConfig{ + Enabled: true, + Schedule: []time.Duration{30 * time.Second, 1 * time.Minute, 2 * time.Minute}, + Jitter: 0, + MaxAge: 10 * time.Minute, + Now: clk.Now, + }, testLogger()) + + ev := mkEvent(42) + + var calls atomic.Int32 + handler := func(_ context.Context, _ providers.JobEvent) error { + calls.Add(1) + return errors.New("HTTP 500 server error") + } + + // Seed the queue as the claim path would: first failure -> attempt 1, + // scheduled 30s out. + q.Add(ev, handler, errors.New("HTTP 500 server error")) + key := keyFor(ev) + it := q.index[key] + if it == nil { + t.Fatal("expected item in index after seed Add") + } + if it.attempts != 1 { + t.Fatalf("attempts after seed = %d, want 1", it.attempts) + } + if got := it.nextAttempt.Sub(clk.Now()); got != 30*time.Second { + t.Fatalf("seed delay = %v, want 30s", got) + } + firstFailure := it.firstFailure + + // fireAndWait drives fireDue and blocks until the goroutine it spawns + // (runOne) has finished its re-Add, detected by attempts reaching the + // expected value. This avoids racing the fire goroutine. + fireAndWait := func(expectAttempts int) { + q.fireDue(context.Background()) + deadline := time.After(2 * time.Second) + for { + q.mu.Lock() + cur, ok := q.index[key] + var attempts int + if ok { + attempts = cur.attempts + } + q.mu.Unlock() + if ok && attempts == expectAttempts { + return + } + select { + case <-deadline: + t.Fatalf("timed out waiting for attempts=%d (have ok=%v attempts=%d)", expectAttempts, ok, attempts) + case <-time.After(2 * time.Millisecond): + } + } + } + + // Fire 1: 30s scheduled -> advance to it, fire. runOne fails and + // re-Adds: attempts 1->2, next delay should be 1m (ladder[1]). + clk.Advance(30 * time.Second) + fireAndWait(2) + it = q.index[key] + if got := it.nextAttempt.Sub(clk.Now()); got != 1*time.Minute { + t.Errorf("after fire 1: delay = %v, want 1m (ladder must advance, not reset to 30s)", got) + } + if !it.firstFailure.Equal(firstFailure) { + t.Errorf("after fire 1: firstFailure changed %v -> %v (must be preserved)", firstFailure, it.firstFailure) + } + + // Fire 2: advance the 1m, fire. attempts 2->3, next delay 2m (ladder[2]). + clk.Advance(1 * time.Minute) + fireAndWait(3) + it = q.index[key] + if got := it.nextAttempt.Sub(clk.Now()); got != 2*time.Minute { + t.Errorf("after fire 2: delay = %v, want 2m", got) + } + + // Fire 3: advance the 2m, fire. attempts 3->4, past ladder end so + // clamps to last rung (2m). Still under MaxAge (total ~3.5m). + clk.Advance(2 * time.Minute) + fireAndWait(4) + it = q.index[key] + if got := it.nextAttempt.Sub(clk.Now()); got != 2*time.Minute { + t.Errorf("after fire 3: delay = %v, want 2m (clamped)", got) + } + + // Now blow past MaxAge: jump the clock well beyond the 10m budget + // from firstFailure, then fire once more. runOne fails, re-Adds, and + // Add's give-up check fires -> item removed from the queue entirely. + clk.Advance(20 * time.Minute) + q.fireDue(context.Background()) + deadline := time.After(2 * time.Second) + for { + if q.Len() == 0 { + q.mu.Lock() + _, stillIndexed := q.index[key] + q.mu.Unlock() + if !stillIndexed { + break + } + } + select { + case <-deadline: + t.Fatalf("timed out waiting for MaxAge give-up; Len=%d", q.Len()) + case <-time.After(2 * time.Millisecond): + } + } +} + // 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) { From eb892ff0975ccdcab2ac4154ba78a301ca89ce17 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Wed, 8 Jul 2026 21:33:05 -0700 Subject: [PATCH 5/5] fix(scheduler): retries survive repeated failures; wire native macOS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-on fixes to the claim retry queue: 1. A retry that fires and fails AGAIN was being dropped instead of re-scheduled, so during a real rate-limit window (which needs several retries until the hourly reset) a job self-healed at most once then was lost. Cause: retryHandler always returned nil, but runOne now Drops on nil — and handleQueued's own enqueueRetryIfEligible had already re-added the item synchronously, so the Drop removed it. Fix: retryHandler now returns the real claim error (captured via a context pointer that enqueueRetryIfEligible fills in, which also suppresses the duplicate enqueue on the retry path). runOne then drops on success and advances the ladder on failure — with the original error class preserved, so rate-aware snap-to-reset still applies on later attempts. New test TestRetryQueue_StillFailingRetry- StaysEnqueued drives the real fireDue->retryHandler->claim path and asserts the job stays enqueued; it fails before this change. 2. handleNativeMacOSJob did not enqueue a retry on claim failure, so native-mode macOS claim failures didn't self-heal (Windows already routes through handleLocalJob, which does). Wired it up to match. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ChVRC9ZrvQarrctDwSy1S5 --- pkg/scheduler/handle_queued_test.go | 2 ++ pkg/scheduler/retry_proof_test.go | 47 +++++++++++++++++++++++++++++ pkg/scheduler/scheduler.go | 32 +++++++++++++++----- 3 files changed, 74 insertions(+), 7 deletions(-) create mode 100644 pkg/scheduler/retry_proof_test.go diff --git a/pkg/scheduler/handle_queued_test.go b/pkg/scheduler/handle_queued_test.go index bc32409..85a173a 100644 --- a/pkg/scheduler/handle_queued_test.go +++ b/pkg/scheduler/handle_queued_test.go @@ -15,6 +15,7 @@ type claimErrorProvider struct { mockProvider err error releaseCount atomic.Int32 + claims atomic.Int32 } func newClaimErrorProvider(name string, err error) *claimErrorProvider { @@ -25,6 +26,7 @@ func newClaimErrorProvider(name string, err error) *claimErrorProvider { } func (p *claimErrorProvider) ClaimJob(_ context.Context, _ *providers.JobEvent, _ string, _ []string) (*providers.Claim, error) { + p.claims.Add(1) return nil, p.err } diff --git a/pkg/scheduler/retry_proof_test.go b/pkg/scheduler/retry_proof_test.go new file mode 100644 index 0000000..45dd63f --- /dev/null +++ b/pkg/scheduler/retry_proof_test.go @@ -0,0 +1,47 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ephpm/ephemerd/pkg/providers" +) + +// TestRetryQueue_StillFailingRetryStaysEnqueued proves that when a retry +// FIRES and the claim fails again (the real production path via +// retryHandler->handleQueued->enqueueRetryIfEligible), the job stays in +// the queue with an advanced attempt count instead of being dropped. +func TestRetryQueue_StillFailingRetryStaysEnqueued(t *testing.T) { + clk := newFakeClock(time.Now()) + 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, 1 * time.Minute}, + Jitter: 0, MaxAge: 1 * time.Hour, Now: clk.Now}, + }) + ev := providers.JobEvent{Provider: p, Action: "queued", Repo: "myrepo", JobID: 100} + + s.handleLocalJob(context.Background(), ev) // seed: claim #1 fails -> enqueue + if got := s.retry.Len(); got != 1 { + t.Fatalf("seed: Len=%d want 1", got) + } + + clk.Advance(30 * time.Second) + s.retry.fireDue(context.Background()) // fires runOne -> retryHandler -> claim #2 fails + + // wait until the re-dispatched claim actually happened + deadline := time.Now().Add(2 * time.Second) + for p.claims.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + time.Sleep(50 * time.Millisecond) // let runOne finish + + if got := p.claims.Load(); got < 2 { + t.Fatalf("retry never re-attempted the claim (claims=%d)", got) + } + if got := s.retry.Len(); got != 1 { + t.Fatalf("BUG: after a still-failing retry, Len=%d want 1 (job was dropped instead of re-scheduled)", got) + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index bc1f5d2..07f6b63 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -717,7 +717,7 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent // 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) + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -849,7 +849,7 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent } unsee() <-s.macSem - s.enqueueRetryIfEligible(event, err) + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -1011,8 +1011,9 @@ func (s *Scheduler) handleNativeMacOSJob(ctx context.Context, event providers.Jo 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() + s.enqueueRetryIfEligible(ctx, event, err) time.Sleep(backoffDuration(event.Repo)) <-s.nativeMacSem return @@ -1194,7 +1195,7 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent // 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) + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -1464,7 +1465,15 @@ func (s *Scheduler) cleanSeen() { // 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) { +func (s *Scheduler) enqueueRetryIfEligible(ctx context.Context, event providers.JobEvent, err error) { + // On the retry path, retryHandler put a *error in the context: hand the + // claim error back to it (preserving the error class) instead of + // enqueuing. Enqueuing here as well would be undone by runOne's + // success-cleanup and lose the job. + if errPtr, ok := ctx.Value(retryAttemptCtxKey{}).(*error); ok && errPtr != nil { + *errPtr = err + return + } if s.retry == nil { return } @@ -1481,6 +1490,8 @@ func (s *Scheduler) enqueueRetryIfEligible(event providers.JobEvent, err error) // succeeded; on failure the handler self-enqueues via // enqueueRetryIfEligible, and on success any future completed webhook // harmlessly Drops the (already-absent) key. +type retryAttemptCtxKey struct{} + func (s *Scheduler) retryHandler(ctx context.Context, event providers.JobEvent) error { key := keyFor(event) s.mu.Lock() @@ -1490,8 +1501,15 @@ func (s *Scheduler) retryHandler(ctx context.Context, event providers.JobEvent) delete(s.seen, key) delete(s.pending, key) s.mu.Unlock() - s.handleQueued(ctx, event) - return nil + // Re-dispatch. enqueueRetryIfEligible, reached on a claim failure, + // writes the claim error into claimErr (and suppresses a duplicate + // enqueue) so we can return it. nil means the claim succeeded (or the + // job was picked up elsewhere) -> runOne drops the retry; a non-nil + // error -> runOne advances the backoff ladder with the real class. + var claimErr error + rctx := context.WithValue(ctx, retryAttemptCtxKey{}, &claimErr) + s.handleQueued(rctx, event) + return claimErr } // buildLabelsForOS builds runner labels for a given target OS.