feat(queue): add full jitter to the reclaimer interval

CHANGELOG.md

@@ -9,6 +9,10 @@ and uses [semantic versioning](https://semver.org/).
 
 ### Added
 
+- **Reclaimer jitter** — the queue reclaimer now waits a random 0–30s between
+  cycles instead of a fixed 30s, so nodes that restart together no longer scan
+  Redis in lock-step. A new `ac_reclaimer_claims_total{stream, result}` counter
+  records each claimed message as `recovered` or `poison`.
 - **Richer results** — each test in the `VERDICT` event, the webhook and the
   stored results now carries `exit_status`, `sandbox_status` and `message`
   beside the existing CPU, wall-clock (`run_time_ns`) and memory figures.

docs/ARCHITECTURE.md

@@ -49,7 +49,7 @@ go-judge sandbox (separate process / container)
 ### Queue (`internal/queue`)
 
 - **`DualConsumer`**: weighted fair queuing over two Redis Streams. Each worker goroutine tries `submitWeight` (default 3) messages from the submit stream before one from the run stream. When both are empty, it blocks for `AC_BLOCK_READ_TIMEOUT` (default 500 ms).
-- **Reclaimer**: every 30 seconds, `XAUTOCLAIM` re-claims messages idle for >30 seconds and reprocesses them. Messages exceeding `maxDeliveryCount=5` deliveries are dead-lettered (ACKed without processing) and counted in `ac_poison_messages_total`.
+- **Reclaimer**: after a random wait between 0 and 30 seconds (full jitter, re-drawn every cycle so nodes that restart together drift apart), `XAUTOCLAIM` re-claims messages idle for >30 seconds and reprocesses them. Messages exceeding `maxDeliveryCount=5` deliveries are dead-lettered (ACKed without processing) and counted in `ac_poison_messages_total`. Each claim is counted in `ac_reclaimer_claims_total` as `recovered` or `poison`.
 - **Legacy drain**: on startup, any messages remaining in the old single stream (`AC_STREAM_NAME`) are drained once and then the goroutine exits.
 - **PubSub**: Redis Pub/Sub + list-backed event log for SSE delivery. `PrepareAndEnqueue` pipelines test storage + QUEUED event + stream enqueue in one round trip.
 
@@ -91,6 +91,7 @@ All metrics are Prometheus counters/histograms exposed on `127.0.0.1:6060/metric
 | `ac_queue_depth` | Gauge | stream |
 | `ac_rate_limit_rejections_total` | Counter | type (user/ip) |
 | `ac_poison_messages_total` | Counter | — |
+| `ac_reclaimer_claims_total` | Counter | stream, result (recovered/poison) |
 
 ---
 

internal/metrics/metrics.go

@@ -63,6 +63,12 @@ var (
 		Help:      "Messages dead-lettered after exceeding max delivery count.",
 	})
 
+	ReclaimerClaimsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "ac",
+		Name:      "reclaimer_claims_total",
+		Help:      "Stuck messages the reclaimer took over, by stream and result (recovered: reprocessed and acked; poison: dead-lettered).",
+	}, []string{"stream", "result"})
+
 	CompileCacheHitsTotal = promauto.NewCounter(prometheus.CounterOpts{
 		Namespace: "ac",
 		Name:      "compile_cache_hits_total",

internal/queue/reclaimer_test.go

@@ -0,0 +1,244 @@
+package queue
+
+import (
+	"context"
+	"math/rand/v2"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	goredis "github.com/redis/go-redis/v9"
+
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
+)
+
+func TestFullJitter_StaysBelowBase(t *testing.T) {
+	base := 30 * time.Second
+	cases := []struct {
+		r    float64
+		want time.Duration
+	}{
+		{0, 0},
+		{0.5, 15 * time.Second},
+		{0.25, 7500 * time.Millisecond},
+	}
+	for _, c := range cases {
+		if got := fullJitter(base, func() float64 { return c.r }); got != c.want {
+			t.Errorf("fullJitter(30s, %v) = %v, want %v", c.r, got, c.want)
+		}
+	}
+
+	r := rand.New(rand.NewPCG(1, 2)).Float64 //nolint:gosec // G404: seeded so the test is repeatable
+	for i := 0; i < 10_000; i++ {
+		if d := fullJitter(base, r); d < 0 || d >= base {
+			t.Fatalf("draw %d = %v, want within [0, %v)", i, d, base)
+		}
+	}
+}
+
+// Ten equal-width buckets over [0, 30s) should each get about a tenth of the
+// draws. A fixed seed keeps the test deterministic.
+func TestFullJitter_IsUniform(t *testing.T) {
+	const draws, buckets = 10_000, 10
+	base := 30 * time.Second
+	r := rand.New(rand.NewPCG(7, 11)).Float64 //nolint:gosec // G404: seeded so the test is repeatable
+
+	var counts [buckets]int
+	for i := 0; i < draws; i++ {
+		counts[int(fullJitter(base, r)*buckets/base)]++
+	}
+	for i, n := range counts {
+		// Expected 1000 per bucket; ±20% is far outside normal variation.
+		if n < 800 || n > 1200 {
+			t.Errorf("bucket %d holds %d draws, want 800..1200 (counts %v)", i, n, counts)
+		}
+	}
+}
+
+// Twenty reclaimers that start in the same instant must not fire their
+// first cycle together: no one-second window may hold more than a handful.
+func TestFullJitter_SpreadsSimultaneousStarts(t *testing.T) {
+	const nodes = 20
+	base := 30 * time.Second
+
+	perSecond := map[int64]int{}
+	for node := 0; node < nodes; node++ {
+		r := rand.New(rand.NewPCG(uint64(node), 99)).Float64 //nolint:gosec // G404: seeded so the test is repeatable
+		perSecond[int64(fullJitter(base, r)/time.Second)]++
+	}
+	for sec, n := range perSecond {
+		if n > 4 {
+			t.Errorf("%d of %d reclaimers fire in second %d, want at most 4", n, nodes, sec)
+		}
+	}
+	if len(perSecond) < nodes/2 {
+		t.Errorf("first cycles fall into only %d distinct seconds, want at least %d", len(perSecond), nodes/2)
+	}
+}
+
+// The loop must draw a new wait after every cycle (a Ticker would keep one
+// fixed phase). Two zero waits make two cycles run at once; the third draw
+// parks the loop for most of an hour.
+func TestReclaimer_RedrawsJitterEveryCycle(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	dc := NewDualConsumer(rdb, "ac:sub-jit", "ac:run-jit", "workers", 3, 300*time.Millisecond)
+	dc.reclaimMinIdle = 0
+	dc.reclaimInterval = time.Hour
+	var draws atomic.Int32
+	dc.jitter = func() float64 {
+		if draws.Add(1) <= 2 {
+			return 0
+		}
+		return 0.99
+	}
+	if err := dc.CreateGroups(ctx); err != nil {
+		t.Fatalf("CreateGroups: %v", err)
+	}
+	addStuckMessage(ctx, t, rdb, "ac:sub-jit", "stuck-jit")
+
+	var handled atomic.Int32
+	var wg sync.WaitGroup
+	dc.StartReclaimer(ctx, &wg, func(_ context.Context, id string) error {
+		if id == "stuck-jit" {
+			handled.Add(1)
+		}
+		return nil
+	})
+
+	deadline := time.Now().Add(5 * time.Second)
+	for draws.Load() < 3 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	wg.Wait()
+
+	if got := draws.Load(); got != 3 {
+		t.Errorf("jitter drawn %d times, want 3 (initial wait + one per finished cycle)", got)
+	}
+	if got := handled.Load(); got != 1 {
+		t.Errorf("stuck message handled %d times, want 1", got)
+	}
+}
+
+func TestReclaimer_CountsRecoveredClaims(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ctx := context.Background()
+
+	dc := NewDualConsumer(rdb, "ac:sub-rec", "ac:run-rec", "workers", 3, 300*time.Millisecond)
+	dc.reclaimMinIdle = 0
+	if err := dc.CreateGroups(ctx); err != nil {
+		t.Fatalf("CreateGroups: %v", err)
+	}
+	addStuckMessage(ctx, t, rdb, "ac:sub-rec", "stuck-rec")
+
+	recovered := metrics.ReclaimerClaimsTotal.WithLabelValues("ac:sub-rec", "recovered")
+	poison := metrics.ReclaimerClaimsTotal.WithLabelValues("ac:sub-rec", "poison")
+	before, beforePoison := testutil.ToFloat64(recovered), testutil.ToFloat64(poison)
+
+	dc.reclaimStream(ctx, "ac:sub-rec", func(context.Context, string) error { return nil })
+
+	if got := testutil.ToFloat64(recovered) - before; got != 1 {
+		t.Errorf("recovered claims grew by %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(poison) - beforePoison; got != 0 {
+		t.Errorf("poison claims grew by %v, want 0", got)
+	}
+}
+
+// A message whose handler fails is left pending for the next cycle, so it
+// must not count as recovered.
+func TestReclaimer_FailedHandlerIsNotCounted(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ctx := context.Background()
+
+	dc := NewDualConsumer(rdb, "ac:sub-fail", "ac:run-fail", "workers", 3, 300*time.Millisecond)
+	dc.reclaimMinIdle = 0
+	if err := dc.CreateGroups(ctx); err != nil {
+		t.Fatalf("CreateGroups: %v", err)
+	}
+	addStuckMessage(ctx, t, rdb, "ac:sub-fail", "stuck-fail")
+
+	recovered := metrics.ReclaimerClaimsTotal.WithLabelValues("ac:sub-fail", "recovered")
+	before := testutil.ToFloat64(recovered)
+
+	dc.reclaimStream(ctx, "ac:sub-fail", func(context.Context, string) error {
+		return context.DeadlineExceeded
+	})
+
+	if got := testutil.ToFloat64(recovered) - before; got != 0 {
+		t.Errorf("recovered claims grew by %v after a handler error, want 0", got)
+	}
+}
+
+func TestReclaimer_CountsPoisonClaims(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ctx := context.Background()
+
+	dc := NewDualConsumer(rdb, "ac:sub-poi", "ac:run-poi", "workers", 3, 300*time.Millisecond)
+	dc.reclaimMinIdle = 0
+	if err := dc.CreateGroups(ctx); err != nil {
+		t.Fatalf("CreateGroups: %v", err)
+	}
+	id := addStuckMessage(ctx, t, rdb, "ac:sub-poi", "poison-job")
+	// The read above was delivery 1; each XCLAIM adds one more.
+	for i := 1; i < maxDeliveryCount; i++ {
+		if err := rdb.XClaim(ctx, &goredis.XClaimArgs{
+			Stream:   "ac:sub-poi",
+			Group:    "workers",
+			Consumer: "crashing-worker",
+			Messages: []string{id},
+		}).Err(); err != nil {
+			t.Fatalf("XClaim: %v", err)
+		}
+	}
+
+	poison := metrics.ReclaimerClaimsTotal.WithLabelValues("ac:sub-poi", "poison")
+	recovered := metrics.ReclaimerClaimsTotal.WithLabelValues("ac:sub-poi", "recovered")
+	before, beforeRecovered := testutil.ToFloat64(poison), testutil.ToFloat64(recovered)
+
+	var handled atomic.Int32
+	dc.reclaimStream(ctx, "ac:sub-poi", func(context.Context, string) error {
+		handled.Add(1)
+		return nil
+	})
+
+	if n := handled.Load(); n != 0 {
+		t.Errorf("handler ran %d times for a poison message, want 0", n)
+	}
+	if got := testutil.ToFloat64(poison) - before; got != 1 {
+		t.Errorf("poison claims grew by %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(recovered) - beforeRecovered; got != 0 {
+		t.Errorf("recovered claims grew by %v, want 0", got)
+	}
+	if n, err := rdb.XLen(ctx, "ac:dlq:poison").Result(); err != nil || n != 1 {
+		t.Errorf("DLQ length = %d (err %v), want 1", n, err)
+	}
+}
+
+// addStuckMessage enqueues a message and reads it as a worker that never
+// acks, leaving it in the group's pending list. It returns the message ID.
+func addStuckMessage(ctx context.Context, t *testing.T, rdb *goredis.Client, stream, submissionID string) string {
+	t.Helper()
+	id, err := rdb.XAdd(ctx, &goredis.XAddArgs{
+		Stream: stream,
+		Values: map[string]any{"submission_id": submissionID},
+	}).Result()
+	if err != nil {
+		t.Fatalf("XAdd: %v", err)
+	}
+	if err := rdb.XReadGroup(ctx, &goredis.XReadGroupArgs{
+		Group:    "workers",
+		Consumer: "crashed-worker",
+		Streams:  []string{stream, ">"},
+		Count:    1,
+	}).Err(); err != nil {
+		t.Fatalf("XReadGroup: %v", err)
+	}
+	return id
+}

internal/queue/stream.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"math/rand/v2"
 	"strconv"
 	"strings"
 	"sync"
@@ -132,7 +133,9 @@ type DualConsumer struct {
 	submitWeight     int
 	bulkWeight       int
 	blockReadTimeout time.Duration
-	reclaimMinIdle   time.Duration // how long a message must be idle before reclaim; defaults to 30s
+	reclaimMinIdle   time.Duration  // how long a message must be idle before reclaim; defaults to 30s
+	reclaimInterval  time.Duration  // upper bound of the jittered wait between reclaim cycles; defaults to 30s
+	jitter           func() float64 // returns a value in [0, 1); tests swap it for a fixed sequence
 }
 
 func NewDualConsumer(rdb *redis.Client, submitStream, runStream, group string, submitWeight int, blockReadTimeout time.Duration) *DualConsumer {
@@ -160,6 +163,8 @@ func NewTieredConsumer(rdb *redis.Client, submitStream, runStream, bulkStream, g
 		bulkWeight:       bulkWeight,
 		blockReadTimeout: blockReadTimeout,
 		reclaimMinIdle:   30 * time.Second,
+		reclaimInterval:  30 * time.Second,
+		jitter:           rand.Float64, //nolint:gosec // G404: spreading timers needs no cryptographic randomness
 	}
 }
 
@@ -318,8 +323,10 @@ func (dc *DualConsumer) blockRead(ctx context.Context, consumer string, handler
 }
 
 // StartReclaimer runs a background goroutine that reclaims stuck messages
-// from both streams every 30s using XAUTOCLAIM. Messages idle for > 30s
-// are re-processed through the handler.
+// from every stream using XAUTOCLAIM. Messages idle for > 30s are
+// re-processed through the handler. Each wait between cycles is drawn fresh
+// from [0, 30s) (full jitter), so nodes that start together do not keep
+// hitting Redis in lock-step.
 func (dc *DualConsumer) StartReclaimer(ctx context.Context, wg *sync.WaitGroup, handler func(ctx context.Context, submissionID string) error) {
 	wg.Add(1)
 	go func() {
@@ -333,7 +340,13 @@ func (dc *DualConsumer) StartReclaimer(ctx context.Context, wg *sync.WaitGroup,
 			time.Sleep(time.Second)
 		}
 	}()
-	slog.Info("reclaimer started", "interval", "30s")
+	slog.Info("reclaimer started", "max_interval", dc.reclaimInterval, "jitter", "full")
+}
+
+// fullJitter returns a random wait in [0, base): the "full jitter" strategy.
+// r must return a value in [0, 1), like rand.Float64.
+func fullJitter(base time.Duration, r func() float64) time.Duration {
+	return time.Duration(r() * float64(base))
 }
 
 func (dc *DualConsumer) runReclaimer(ctx context.Context, handler func(ctx context.Context, submissionID string) error) (panicked bool) {
@@ -343,16 +356,20 @@ func (dc *DualConsumer) runReclaimer(ctx context.Context, handler func(ctx conte
 			panicked = true
 		}
 	}()
-	ticker := time.NewTicker(30 * time.Second)
-	defer ticker.Stop()
+	// A Ticker keeps a fixed phase forever, so reclaimers that start in the
+	// same instant stay synchronised. A Timer re-armed with a new random wait
+	// after every cycle breaks that phase lock.
+	timer := time.NewTimer(fullJitter(dc.reclaimInterval, dc.jitter))
+	defer timer.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return false
-		case <-ticker.C:
+		case <-timer.C:
 			for _, stream := range dc.streams() {
 				dc.reclaimStream(ctx, stream, handler)
 			}
+			timer.Reset(fullJitter(dc.reclaimInterval, dc.jitter))
 		}
 	}
 }
@@ -447,6 +464,7 @@ func (dc *DualConsumer) reclaimStream(ctx context.Context, stream string, handle
 					continue
 				}
 				dc.ack(ctx, stream, msg.ID)
+				metrics.ReclaimerClaimsTotal.WithLabelValues(stream, "poison").Inc()
 				continue
 			}
 
@@ -459,6 +477,7 @@ func (dc *DualConsumer) reclaimStream(ctx context.Context, stream string, handle
 			}
 
 			dc.ack(ctx, stream, msg.ID)
+			metrics.ReclaimerClaimsTotal.WithLabelValues(stream, "recovered").Inc()
 		}
 
 		if newCursor == "0-0" {