fix(scheduler): correct claim retry queue (ladder, give-up, native)

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
 }
 

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)
+	}
+}

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.

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):
+		}
+	}
+}

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.