From 8c4a59f6522a405a5364b5e7c54245cfd47ca031 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Fri, 10 Jul 2026 17:51:13 -0700 Subject: [PATCH] fix(scheduler): correct claim retry queue (ladder, give-up, native) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The claim retry queue merged in #104 has three correctness bugs; fix them in place (config-gated and enabled by default, so this runs on every fleet host): 1. Backoff ladder never advances / give-up never fires. fireDue deleted the popped item from the index, so runOne's failure re-Add built a fresh item — attempts reset to 1 (stuck at schedule[0] ~30s) and firstFailure reset every fire (MaxAge never triggered). A job that couldn't be claimed retried every 30s forever, burning the shared installation token it was waiting on. Fix: fireDue keeps the item in the index across a fire. 2. Retries dropped after one attempt. With (1) half-fixed you get the opposite: retryHandler always returned nil, and runOne dropping on nil removed the item handleQueued had just re-Added synchronously — so a job self-healed at most once. Fix: retryHandler returns the real claim error (captured via a context pointer that enqueueRetryIfEligible fills in, which also suppresses the duplicate enqueue). runOne drops on success / advances on failure, with the rate-limit error class preserved so snap-to-reset still applies on later attempts. 3. Native macOS claim failures didn't self-heal — handleNativeMacOSJob was the one claim path not wired to the queue (Windows already routes through handleLocalJob). Wired it up. Tests: TestRetryQueue_StillFailingRetryStaysEnqueued drives the real fireDue->retryHandler->claim path (fails before this change); TestRetryQueue_LadderAdvancesThroughFirePath asserts the delay grows 30s->1m->2m and firstFailure is preserved. Full scheduler suite green, -race and go vet clean. 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/retry_queue.go | 9 +- pkg/scheduler/retry_queue_test.go | 129 ++++++++++++++++++++++++++++ pkg/scheduler/scheduler.go | 32 +++++-- 5 files changed, 211 insertions(+), 8 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 2bf834c..3408629 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/retry_queue.go b/pkg/scheduler/retry_queue.go index ec69cd0..adbd57a 100644 --- a/pkg/scheduler/retry_queue.go +++ b/pkg/scheduler/retry_queue.go @@ -390,7 +390,10 @@ 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: leaving the + // popped item registered lets a failure re-Add (via runOne) find + // it and advance attempts/preserve firstFailure, rather than + // building a fresh item that resets the ladder and MaxAge. due = append(due, it) } q.mu.Unlock() @@ -411,6 +414,10 @@ func (q *retryQueue) runOne(ctx context.Context, it *retryItem) { "attempt", it.attempts+1) err := it.handler(ctx, it.event) if err == nil { + // fireDue leaves the popped item in q.index so a failure re-Add + // advances the ladder; on success nothing re-Adds it, so drop the + // lingering index entry here. removeLocked tolerates index < 0. + 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..4b53349 100644 --- a/pkg/scheduler/retry_queue_test.go +++ b/pkg/scheduler/retry_queue_test.go @@ -361,3 +361,132 @@ func TestRetryQueue_Disabled_IsNoOp(t *testing.T) { t.Error("Enabled() = true, want false") } } + +// 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): + } + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 824f24c..e795001 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -801,7 +801,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 } @@ -934,7 +934,7 @@ func (s *Scheduler) handleMacOSJob(ctx context.Context, event providers.JobEvent } unsee() <-s.macSem - s.enqueueRetryIfEligible(event, err) + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -1094,8 +1094,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 @@ -1275,7 +1276,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 } @@ -1719,7 +1720,15 @@ func (s *Scheduler) sweepOrphanRunners() { // 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 } @@ -1736,6 +1745,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() @@ -1745,8 +1756,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. On a claim failure enqueueRetryIfEligible writes the + // error into claimErr (and suppresses a duplicate enqueue), so we can + // return it: nil => claimed/dispatched OK (runOne drops the retry); + // non-nil => still failing (runOne advances the ladder with the real + // error class preserved). + var claimErr error + rctx := context.WithValue(ctx, retryAttemptCtxKey{}, &claimErr) + s.handleQueued(rctx, event) + return claimErr } // buildLabelsForOS builds runner labels for a given target OS.