perf: stage timings, shared event subscription and honest Retry-After

CHANGELOG.md

@@ -21,6 +21,14 @@ and uses [semantic versioning](https://semver.org/).
   /api/v1/execute` answers 400 for a language that is not installed instead of
   failing the job later.
 
+### Changed
+
+- **Where the time goes** — new histograms `ac_queue_wait_seconds{stream}` (time between enqueue and a worker reading the job) and `ac_job_stage_seconds{stage}` (`claim`, `load_tests`, `run`, `persist`, `publish`), beside the existing `ac_compile_duration_seconds`.
+- **Lower latency per job** — a worker now fetches a job's tests from Redis while it claims the job in PostgreSQL, instead of one after the other.
+- **One Redis subscription per gateway** — SSE clients share a single pattern subscription instead of one subscription each. A client that falls behind is closed so it reconnects and replays from the event log; no event is dropped silently.
+- **Honest `Retry-After`** — when the queue is full, `429` carries the time the queue needs to fall under the admission limit at the measured completion rate (1 to 60 s), not a fixed 10.
+- **Leaner JVM** — `java` and `kotlin` start with the serial collector and shared class data, which also lowers the thread count of each run.
+
 ## [0.1.0] — 2026-10-04
 
 First open-source release.

README.md

@@ -599,6 +599,31 @@ feature/* / fix/*  ──>  development (CI Gate)  ──>  production (Protecte
 
 ---
 
+## Performance Tuning
+
+**Find the stage first.** A slow p99 is usually queueing, not execution. Per job, the engine records where the time went:
+
+| Metric | Meaning |
+|---|---|
+| `ac_queue_wait_seconds{stream}` | Enqueue to a worker reading the job |
+| `ac_job_stage_seconds{stage}` | `claim`, `load_tests`, `run`, `persist`, `publish` |
+| `ac_compile_duration_seconds{language}` | The compile step, per language |
+| `ac_job_processing_duration_seconds{language,verdict}` | Dequeue to verdict |
+
+```promql
+histogram_quantile(0.99, sum by (le) (rate(ac_queue_wait_seconds_bucket[5m])))
+histogram_quantile(0.99, sum by (le, stage) (rate(ac_job_stage_seconds_bucket[5m])))
+```
+
+If queue wait dominates, add capacity; if `run` dominates, look at the language and the limits; if `persist` or `publish` do, look at PostgreSQL and Redis.
+
+**Sizing.**
+- Keep `AC_WORKER_COUNT` equal to go-judge's `-parallelism`; more workers than sandbox slots only queue inside go-judge.
+- On a multi-socket host, run one engine plus one go-judge per NUMA node (for example `numactl --cpunodebind=0 --membind=0` for each pair, each with its own listen address and `-parallelism` set to that node's physical cores) rather than one pair on one node. The published benchmark used a single node of a 256-thread host.
+- Raise go-judge's `-pre-fork` toward its parallelism so a burst does not wait for sandbox creation.
+
+**Overload.** `AC_ADMISSION_MAX_QUEUE_DEPTH` bounds how long a job can wait. A job admitted at the limit waits about depth ÷ completion rate, so lower the limit if that exceeds your verdict SLO. Rejected callers receive `Retry-After` computed from the measured completion rate.
+
 ## Repository layout
 
 | Path | Contents |

go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.3 // indirect
 	github.com/klauspost/compress v1.20.3 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e // indirect
 	github.com/magiconair/properties v1.8.10 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect

internal/gateway/handler.go

@@ -439,7 +439,7 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		slog.Error("admission check failed, allowing", "err", err)
 	} else if !admitted {
-		w.Header().Set("Retry-After", "10")
+		w.Header().Set("Retry-After", strconv.Itoa(h.busyRetryAfter(ctx, targetStream.StreamName())))
 		jsonError(w, "server is busy, try again later", http.StatusTooManyRequests)
 		return
 	}
@@ -860,3 +860,17 @@ func (h *Handler) StartLanguageProbe(ctx context.Context, interval time.Duration
 		}
 	}()
 }
+
+// busyRetryAfter is how long a rejected caller should wait: the time for the
+// queue to fall back under the admission limit at the rate jobs are finishing,
+// between 1 and 60 seconds. Without a measured rate it is the old fixed 10.
+func (h *Handler) busyRetryAfter(ctx context.Context, stream string) int {
+	if metrics.JobsDrain.Rate() <= 0 {
+		return 10
+	}
+	depth, err := h.rdb.XLen(ctx, stream).Result()
+	if err != nil {
+		return 10
+	}
+	return metrics.JobsDrain.RetryAfter(depth-int64(h.admissionMaxQueueDepth)+1, 1, 60)
+}

internal/gateway/handler_test.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/db"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/model"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/queue"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/sandbox"
@@ -716,3 +717,31 @@ func TestLanguageAvailability(t *testing.T) {
 		}
 	}
 }
+
+func TestBusyRetryAfterFollowsTheDrainRate(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	h := NewHandler(nil, queue.NewStream(rdb, "ac:submissions:submit", "workers"), queue.NewStream(rdb, "ac:submissions:run", "workers"), nil, nil, rdb, nil, nil, 10, 8, nil)
+	ctx := context.Background()
+	for i := 0; i < 70; i++ {
+		rdb.XAdd(ctx, &goredis.XAddArgs{Stream: "ac:submissions:run", Values: map[string]any{"submission_id": fmt.Sprint(i)}})
+	}
+
+	original := metrics.JobsDrain
+	t.Cleanup(func() { metrics.JobsDrain = original })
+
+	now := time.Unix(5000, 0)
+	metrics.JobsDrain = metrics.NewDrain(func() time.Time { return now })
+	if got := h.busyRetryAfter(ctx, "ac:submissions:run"); got != 10 {
+		t.Fatalf("Retry-After with no measured rate = %d, want the old fixed 10", got)
+	}
+
+	for i := 0; i < 201; i++ {
+		metrics.JobsDrain.Complete()
+	}
+	now = now.Add(time.Second) // the rate settles at about 60 jobs/s (alpha 0.3)
+	got := h.busyRetryAfter(ctx, "ac:submissions:run")
+	// 70 queued, limit 10: it must fall by 61 jobs, about a second at 60/s.
+	if got < 1 || got > 3 {
+		t.Fatalf("Retry-After = %d, want 1 to 3 seconds for a 61-job excess at about 60 jobs/s", got)
+	}
+}

internal/metrics/drain.go

@@ -0,0 +1,74 @@
+package metrics
+
+import (
+	"math"
+	"sync"
+	"time"
+)
+
+// Drain measures how fast jobs finish, as a smoothed rate per second. The
+// gateway turns it into an honest Retry-After when it sheds load.
+type Drain struct {
+	mu      sync.Mutex
+	rate    float64 // smoothed completions per second
+	window  time.Time
+	count   int
+	now     func() time.Time
+	started bool
+}
+
+const (
+	drainWindow = time.Second
+	// drainAlpha weights the newest second against the history.
+	drainAlpha = 0.3
+)
+
+// NewDrain returns a tracker that reads time from now.
+func NewDrain(now func() time.Time) *Drain {
+	return &Drain{now: now}
+}
+
+// Complete records one finished job.
+func (d *Drain) Complete() {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.advance()
+	d.count++
+}
+
+// Rate is the smoothed completions per second; zero until a full second passed.
+func (d *Drain) Rate() float64 {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.advance()
+	return d.rate
+}
+
+// RetryAfter is how many seconds a caller should wait for depth queued jobs to
+// drain, between lo and hi. With no measured rate it returns lo.
+func (d *Drain) RetryAfter(depth int64, lo, hi int) int {
+	rate := d.Rate()
+	if rate <= 0 || depth <= 0 {
+		return lo
+	}
+	seconds := int(math.Ceil(float64(depth) / rate))
+	return max(lo, min(hi, seconds))
+}
+
+// advance folds every full elapsed window into the smoothed rate. Idle windows
+// decay it toward zero.
+func (d *Drain) advance() {
+	now := d.now()
+	if !d.started {
+		d.started, d.window = true, now
+		return
+	}
+	for now.Sub(d.window) >= drainWindow {
+		d.rate = drainAlpha*float64(d.count) + (1-drainAlpha)*d.rate
+		d.count = 0
+		d.window = d.window.Add(drainWindow)
+	}
+}
+
+// JobsDrain is the process-wide tracker: workers feed it, the gateway reads it.
+var JobsDrain = NewDrain(time.Now)

internal/metrics/drain_test.go

@@ -0,0 +1,37 @@
+package metrics
+
+import (
+	"testing"
+	"time"
+)
+
+func TestDrainRateAndRetryAfter(t *testing.T) {
+	now := time.Unix(1000, 0)
+	d := NewDrain(func() time.Time { return now })
+
+	if got := d.RetryAfter(500, 1, 60); got != 1 {
+		t.Fatalf("RetryAfter with no measurements = %d, want the lower bound", got)
+	}
+	d.Complete() // starts the first window
+	for i := 0; i < 100; i++ {
+		d.Complete()
+	}
+	now = now.Add(time.Second)
+	rate := d.Rate()
+	if rate < 25 || rate > 35 {
+		t.Fatalf("rate after one 101-job second = %.1f, want about 30 (alpha 0.3)", rate)
+	}
+	if got := d.RetryAfter(600, 1, 60); got < 15 || got > 25 {
+		t.Fatalf("RetryAfter(600) = %d, want about 20 at 30 jobs/s", got)
+	}
+	if got := d.RetryAfter(1_000_000, 1, 60); got != 60 {
+		t.Fatalf("RetryAfter above the cap = %d, want 60", got)
+	}
+	if got := d.RetryAfter(0, 2, 60); got != 2 {
+		t.Fatalf("RetryAfter on an empty queue = %d, want the lower bound", got)
+	}
+	now = now.Add(30 * time.Second)
+	if d.Rate() > 0.1 {
+		t.Fatalf("rate after 30 idle seconds = %.3f, want it to have decayed", d.Rate())
+	}
+}

internal/metrics/metrics.go

@@ -74,4 +74,21 @@ var (
 		Name:      "compile_cache_misses_total",
 		Help:      "Compile cache misses — source was not in cache, compilation required.",
 	})
+
+	QueueWaitSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ac",
+		Name:      "queue_wait_seconds",
+		Help:      "Time a job waited in its stream between being enqueued and being read by a worker.",
+		Buckets:   []float64{0.005, 0.025, 0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120},
+	}, []string{"stream"})
+
+	// JobStageSeconds splits a job's time into its stages so a slow p99 can be
+	// attributed: claim, load_tests, run, persist and publish. Compile has its
+	// own histogram (CompileDuration) because it is labelled by language.
+	JobStageSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ac",
+		Name:      "job_stage_seconds",
+		Help:      "Time spent in each stage of processing a job.",
+		Buckets:   []float64{0.001, 0.005, 0.025, 0.1, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"stage"})
 )