From a6c60087f746747d24654273383c903b580c76a0 Mon Sep 17 00:00:00 2001 From: harichillin Date: Thu, 12 Nov 2026 16:26:58 +0530 Subject: [PATCH] feat(queue): add full jitter to the reclaimer interval The reclaimer used a fixed 30s ticker, so nodes that restarted together ran XAUTOCLAIM and XPENDING against Redis in the same instant on every cycle. It now re-arms a timer with a fresh random wait in [0, 30s) after each cycle, which breaks the phase lock. Adds ac_reclaimer_claims_total{stream, result} counting each claimed message as recovered (handler succeeded and acked) or poison (dead-lettered). Closes #122 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + docs/ARCHITECTURE.md | 3 +- internal/metrics/metrics.go | 6 + internal/queue/reclaimer_test.go | 244 +++++++++++++++++++++++++++++++ internal/queue/stream.go | 33 ++++- 5 files changed, 282 insertions(+), 8 deletions(-) create mode 100644 internal/queue/reclaimer_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 452454c..a427195 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 58d9d3f..5b1ceac 100644 --- a/docs/ARCHITECTURE.md +++ b/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) | --- diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index b06ab29..a01b5a3 100644 --- a/internal/metrics/metrics.go +++ b/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", diff --git a/internal/queue/reclaimer_test.go b/internal/queue/reclaimer_test.go new file mode 100644 index 0000000..dbc30dc --- /dev/null +++ b/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 +} diff --git a/internal/queue/stream.go b/internal/queue/stream.go index 05783ca..3e77400 100644 --- a/internal/queue/stream.go +++ b/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" {