fix(scheduler): claim retry queue with jittered backoff + fresh rate gauges

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.

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.

pkg/config/config.go

@@ -461,6 +461,42 @@ 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"`
+}
+
+// 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"`
 }
 
 // MacOSRunnerConfig controls macOS job routing. It lives under [runner]
@@ -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 == "" {

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) {