diff --git a/CHANGELOG.md b/CHANGELOG.md index f6be422..68ed908 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index bc5aabc..1798f6d 100644 --- a/README.md +++ b/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 | diff --git a/go.mod b/go.mod index c10a745..923c864 100644 --- a/go.mod +++ b/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 diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go index fed0898..19bb671 100644 --- a/internal/gateway/handler.go +++ b/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) +} diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go index db9f73c..d0d69cf 100644 --- a/internal/gateway/handler_test.go +++ b/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) + } +} diff --git a/internal/metrics/drain.go b/internal/metrics/drain.go new file mode 100644 index 0000000..8947bfc --- /dev/null +++ b/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) diff --git a/internal/metrics/drain_test.go b/internal/metrics/drain_test.go new file mode 100644 index 0000000..61f51ed --- /dev/null +++ b/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()) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index f8300b0..b06ab29 100644 --- a/internal/metrics/metrics.go +++ b/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"}) ) diff --git a/internal/queue/hub.go b/internal/queue/hub.go new file mode 100644 index 0000000..f40968f --- /dev/null +++ b/internal/queue/hub.go @@ -0,0 +1,156 @@ +package queue + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "sync" + + "github.com/redis/go-redis/v9" + + "github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog" +) + +const ( + eventChannelPrefix = "ac:events:" + // subscriberBuffer is how many events one SSE client may be behind before + // it is cut loose to reconnect and replay from the event log. + subscriberBuffer = 16 +) + +// hub fans one Redis pattern subscription out to every SSE client of this +// process. Before it, each client held its own Redis subscription, so 5,000 +// open streams meant 5,000 Redis connections. +type hub struct { + mu sync.Mutex + subs map[string]map[chan eventlog.Event]struct{} + cancel context.CancelFunc + done chan struct{} // closed when the hub stops, after every client channel is closed +} + +// startHub opens the shared subscription and waits for Redis to confirm it, so +// a client that subscribes after this returns cannot miss a later event. +func (p *PubSub) startHub() (*hub, error) { + ctx, cancel := context.WithCancel(context.Background()) + sub := p.rdb.PSubscribe(ctx, eventChannelPrefix+"*") + if _, err := sub.Receive(ctx); err != nil { + cancel() + _ = sub.Close() + return nil, err + } + h := &hub{subs: make(map[string]map[chan eventlog.Event]struct{}), cancel: cancel, done: make(chan struct{})} + p.wg.Add(1) + go func() { + defer p.wg.Done() + defer h.closeAll() + defer sub.Close() //nolint:errcheck // best-effort: the connection is going away + messages := sub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-messages: + if !ok { + return + } + h.dispatch(msg) + } + } + }() + return h, nil +} + +// getHub returns the running hub, starting it on first use. A failed start is +// retried by the next caller. +func (p *PubSub) getHub() (*hub, error) { + p.hubMu.Lock() + defer p.hubMu.Unlock() + if p.hub != nil { + return p.hub, nil + } + h, err := p.startHub() + if err != nil { + return nil, err + } + p.hub = h + return h, nil +} + +// stopHub ends the shared subscription. Wait calls it before waiting. +func (p *PubSub) stopHub() { + p.hubMu.Lock() + defer p.hubMu.Unlock() + if p.hub != nil { + p.hub.cancel() + } +} + +func (h *hub) add(jobID string, ch chan eventlog.Event) { + h.mu.Lock() + defer h.mu.Unlock() + if h.subs[jobID] == nil { + h.subs[jobID] = make(map[chan eventlog.Event]struct{}) + } + h.subs[jobID][ch] = struct{}{} +} + +// remove unregisters a client. It reports whether the client was still +// registered, so the caller closes its channel only if the hub has not already. +func (h *hub) remove(jobID string, ch chan eventlog.Event) bool { + h.mu.Lock() + defer h.mu.Unlock() + if _, ok := h.subs[jobID][ch]; !ok { + return false + } + delete(h.subs[jobID], ch) + if len(h.subs[jobID]) == 0 { + delete(h.subs, jobID) + } + return true +} + +// dispatch hands a message to the clients of its job. A client whose buffer is +// full is closed instead of skipped: the SSE handler turns a closed channel +// into a reconnect, and the reconnect replays every event from the log. +func (h *hub) dispatch(msg *redis.Message) { + jobID, ok := strings.CutPrefix(msg.Channel, eventChannelPrefix) + if !ok || strings.Contains(jobID, ":") { + return + } + h.mu.Lock() + if len(h.subs[jobID]) == 0 { + h.mu.Unlock() + return + } + var evt eventlog.Event + if err := json.Unmarshal([]byte(msg.Payload), &evt); err != nil { + h.mu.Unlock() + return + } + for ch := range h.subs[jobID] { + select { + case ch <- evt: + default: + slog.Warn("sse: client fell behind, closing it to reconnect", "job_id", jobID) + delete(h.subs[jobID], ch) + close(ch) + } + } + if len(h.subs[jobID]) == 0 { + delete(h.subs, jobID) + } + h.mu.Unlock() +} + +func (h *hub) closeAll() { + h.mu.Lock() + defer h.mu.Unlock() + defer close(h.done) + for jobID, set := range h.subs { + for ch := range set { + close(ch) + } + delete(h.subs, jobID) + } +} diff --git a/internal/queue/hub_test.go b/internal/queue/hub_test.go new file mode 100644 index 0000000..2e9580a --- /dev/null +++ b/internal/queue/hub_test.go @@ -0,0 +1,124 @@ +package queue + +import ( + "context" + "testing" + "time" + + "github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog" +) + +func recv(t *testing.T, ch <-chan eventlog.Event) eventlog.Event { + t.Helper() + select { + case evt, ok := <-ch: + if !ok { + t.Fatal("channel closed, want an event") + } + return evt + case <-time.After(3 * time.Second): + t.Fatal("no event within 3s") + } + return eventlog.Event{} +} + +func TestHub_OneRedisSubscriptionServesEveryClient(t *testing.T) { + mr, rdb := newTestRedis(t) + ps := NewPubSub(rdb) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + a1 := ps.Subscribe(ctx, "job-a") + a2 := ps.Subscribe(ctx, "job-a") + b := ps.Subscribe(ctx, "job-b") + // All three clients share one pattern subscription. + if n := mr.PubSubNumPat(); n != 1 { + t.Fatalf("pattern subscriptions = %d, want exactly 1", n) + } + + if err := ps.PublishEvent(ctx, eventlog.New("job-a", eventlog.EventRunning, nil)); err != nil { + t.Fatal(err) + } + if got := recv(t, a1); got.JobID != "job-a" { + t.Errorf("a1 got %q", got.JobID) + } + if got := recv(t, a2); got.JobID != "job-a" { + t.Errorf("a2 got %q", got.JobID) + } + select { + case evt := <-b: + t.Fatalf("job-b received job-a's event: %+v", evt) + case <-time.After(150 * time.Millisecond): + } +} + +func TestHub_CancelledClientIsRemovedAndItsChannelClosed(t *testing.T) { + _, rdb := newTestRedis(t) + ps := NewPubSub(rdb) + ctx, cancel := context.WithCancel(context.Background()) + ch := ps.Subscribe(ctx, "job-c") + cancel() + select { + case _, ok := <-ch: + if ok { + t.Fatal("got an event after cancel, want a closed channel") + } + case <-time.After(3 * time.Second): + t.Fatal("channel not closed after cancel") + } + ps.hubMu.Lock() + defer ps.hubMu.Unlock() + ps.hub.mu.Lock() + defer ps.hub.mu.Unlock() + if len(ps.hub.subs["job-c"]) != 0 { + t.Fatal("cancelled client still registered") + } +} + +func TestHub_AClientThatFallsBehindIsClosedSoItReplays(t *testing.T) { + _, rdb := newTestRedis(t) + ps := NewPubSub(rdb) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + slow := ps.Subscribe(ctx, "job-d") + fast := ps.Subscribe(ctx, "job-e") + + for i := 0; i < subscriberBuffer+5; i++ { + if err := ps.PublishEvent(ctx, eventlog.New("job-d", eventlog.EventRunning, nil)); err != nil { + t.Fatal(err) + } + } + // Drain the buffered events; the channel must then be closed, not stalled. + deadline := time.After(3 * time.Second) + for open := true; open; { + select { + case _, ok := <-slow: + open = ok + case <-deadline: + t.Fatal("slow client's channel was never closed") + } + } + // Other jobs are unaffected. + if err := ps.PublishEvent(ctx, eventlog.New("job-e", eventlog.EventRunning, nil)); err != nil { + t.Fatal(err) + } + recv(t, fast) +} + +func TestHub_WaitStopsTheSharedSubscription(t *testing.T) { + _, rdb := newTestRedis(t) + ps := NewPubSub(rdb) + ctx, cancel := context.WithCancel(context.Background()) + ch := ps.Subscribe(ctx, "job-f") + done := make(chan struct{}) + go func() { ps.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Wait did not return") + } + cancel() + if _, ok := <-ch; ok { + t.Fatal("channel delivered an event after the hub stopped") + } +} diff --git a/internal/queue/pubsub.go b/internal/queue/pubsub.go index 58b3300..c73a8c1 100644 --- a/internal/queue/pubsub.go +++ b/internal/queue/pubsub.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "io" + "log/slog" "strconv" "sync" "time" @@ -168,6 +169,9 @@ type PubSub struct { rdb *redis.Client wg sync.WaitGroup webhookMasterKey []byte // nil means no encryption + + hubMu sync.Mutex + hub *hub // the shared event subscription, started by the first Subscribe } // NewPubSub creates a PubSub with no webhook secret encryption (legacy / no key). @@ -181,9 +185,13 @@ func NewPubSubWithKey(rdb *redis.Client, webhookMasterKey []byte) *PubSub { return &PubSub{rdb: rdb, webhookMasterKey: webhookMasterKey} } -// Wait blocks until all active Subscribe goroutines have exited. -// Call after srv.Shutdown to ensure sub.Close() completes before rdb.Close(). -func (p *PubSub) Wait() { p.wg.Wait() } +// Wait stops the shared event subscription and blocks until every Subscribe +// goroutine has exited. Call after srv.Shutdown so the subscription is closed +// before rdb.Close(). +func (p *PubSub) Wait() { + p.stopHub() + p.wg.Wait() +} func channelName(jobID string) string { return "ac:events:" + jobID @@ -230,40 +238,32 @@ func (p *PubSub) GetEventLog(ctx context.Context, jobID string) ([]eventlog.Even return events, nil } -// Subscribe returns a channel that receives events for the given job. -// Close the context to unsubscribe. +// Subscribe returns a channel that receives events for the given job. It is +// registered with the process-wide hub before Subscribe returns, so events +// published after the call are never missed. Cancel ctx to unsubscribe. The +// channel is closed when Redis is unreachable, when the client falls behind, or +// when the PubSub is stopped; callers treat that as "reconnect and replay". func (p *PubSub) Subscribe(ctx context.Context, jobID string) <-chan eventlog.Event { - ch := make(chan eventlog.Event, 16) - sub := p.rdb.Subscribe(ctx, channelName(jobID)) - + ch := make(chan eventlog.Event, subscriberBuffer) + h, err := p.getHub() + if err != nil { + slog.Error("pubsub: could not open the shared subscription", "err", err) + close(ch) + return ch + } + h.add(jobID, ch) p.wg.Add(1) go func() { defer p.wg.Done() - defer close(ch) - defer sub.Close() //nolint:errcheck // best-effort: unsubscribe when the SSE handler's goroutine exits - - msgCh := sub.Channel() - for { - select { - case <-ctx.Done(): - return - case msg, ok := <-msgCh: - if !ok { - return - } - var evt eventlog.Event - if err := json.Unmarshal([]byte(msg.Payload), &evt); err != nil { - continue - } - select { - case ch <- evt: - case <-ctx.Done(): - return - } + select { + case <-ctx.Done(): + if h.remove(jobID, ch) { + close(ch) } + case <-h.done: + // The hub has closed every client channel. } }() - return ch } diff --git a/internal/queue/stream.go b/internal/queue/stream.go index b37c87e..05783ca 100644 --- a/internal/queue/stream.go +++ b/internal/queue/stream.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "log/slog" + "strconv" + "strings" "sync" "time" @@ -99,6 +101,7 @@ func (s *Stream) Consume(ctx context.Context, consumer string, handler func(ctx continue } + observeQueueWait(s.stream, msg.ID) if err := handler(ctx, subID); err != nil { slog.Error("queue: handler error", "submission_id", subID, "err", err) continue @@ -254,6 +257,7 @@ func (dc *DualConsumer) consumeOne(ctx context.Context, stream, consumer string, return true } + observeQueueWait(stream, msg.ID) if err := handler(ctx, subID); err != nil { slog.Error("queue: handler error", "stream", stream, "submission_id", subID, "err", err) return true // consumed but not ACKed — reclaimer will pick it up @@ -303,6 +307,7 @@ func (dc *DualConsumer) blockRead(ctx context.Context, consumer string, handler dc.ack(ctx, streamName, msg.ID) continue } + observeQueueWait(streamName, msg.ID) if err := handler(ctx, subID); err != nil { slog.Error("queue: handler error", "stream", streamName, "submission_id", subID, "err", err) continue @@ -537,3 +542,19 @@ func DrainLegacyStream(ctx context.Context, wg *sync.WaitGroup, rdb *redis.Clien } }() } + +// observeQueueWait records how long a message sat in its stream. Redis stream +// IDs start with the millisecond they were added, so the wait needs no extra +// field on the message. +func observeQueueWait(stream, id string) { + ms, _, ok := strings.Cut(id, "-") + if !ok { + return + } + added, err := strconv.ParseInt(ms, 10, 64) + if err != nil { + return + } + wait := time.Since(time.UnixMilli(added)) + metrics.QueueWaitSeconds.WithLabelValues(stream).Observe(max(wait, 0).Seconds()) +} diff --git a/internal/queue/wait_test.go b/internal/queue/wait_test.go new file mode 100644 index 0000000..4408baa --- /dev/null +++ b/internal/queue/wait_test.go @@ -0,0 +1,27 @@ +package queue + +import ( + "fmt" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics" +) + +func TestObserveQueueWait(t *testing.T) { + before := testutil.CollectAndCount(metrics.QueueWaitSeconds) + added := time.Now().Add(-3 * time.Second).UnixMilli() + observeQueueWait("ac:test-wait", fmt.Sprintf("%d-0", added)) + if got := testutil.CollectAndCount(metrics.QueueWaitSeconds); got != before+1 { + t.Fatalf("series after a valid id = %d, want %d", got, before+1) + } + // Ids that are not stream ids add nothing and do not panic. + for _, id := range []string{"", "nope", "abc-0", "-"} { + observeQueueWait("ac:test-wait", id) + } + if got := testutil.CollectAndCount(metrics.QueueWaitSeconds); got != before+1 { + t.Fatalf("series after invalid ids = %d, want unchanged %d", got, before+1) + } +} diff --git a/internal/sandbox/args_test.go b/internal/sandbox/args_test.go index 4782c7a..6fe5878 100644 --- a/internal/sandbox/args_test.go +++ b/internal/sandbox/args_test.go @@ -46,3 +46,23 @@ func TestRunCmdCarriesArgsStackAndProcs(t *testing.T) { t.Fatalf("defaults changed: procs = %d, stack = %d", plain.ProcLimit, plain.StackLimit) } } + +func TestJavaStartsLeanAndKeepsItsExecutables(t *testing.T) { + t.Parallel() + for _, id := range []string{"java", "kotlin"} { + run := Languages[id].RunArgs + if run[0] != "/usr/bin/java" { + t.Errorf("%s: run executable = %s, want /usr/bin/java", id, run[0]) + } + joined := strings.Join(run, " ") + for _, flag := range []string{"-XX:+UseSerialGC", "-Xshare:auto", "-XX:TieredStopAtLevel=1"} { + if !strings.Contains(joined, flag) { + t.Errorf("%s: run args %q lack %s", id, joined, flag) + } + } + } + // -J flags must come right after javac and before the options it parses. + if compile := Languages["java"].CompileArgs; compile[0] != "/usr/bin/javac" || !strings.HasPrefix(compile[1], "-J") { + t.Errorf("javac args = %v, want the -J flags straight after the executable", compile) + } +} diff --git a/internal/sandbox/language.go b/internal/sandbox/language.go index e963925..654732f 100644 --- a/internal/sandbox/language.go +++ b/internal/sandbox/language.go @@ -131,14 +131,14 @@ var Languages = map[string]LangConfig{ }, "java": { Compiled: true, - CompileArgs: []string{"/usr/bin/javac", "-d", "/w", "/w/Main.java"}, + CompileArgs: []string{"/usr/bin/javac", "-J-XX:TieredStopAtLevel=1", "-J-XX:+UseSerialGC", "-J-Xshare:auto", "-d", "/w", "/w/Main.java"}, CompileEnv: []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"}, CompileCPU: 15_000_000_000, CompileMem: 536_870_912, CompileProc: 50, SourcePath: "/w/Main.java", BinaryPath: "/w/Main.class", - RunArgs: []string{"/usr/bin/java", "-XX:+TieredCompilation", "-XX:TieredStopAtLevel=1", "-cp", "/w", "Main"}, + RunArgs: []string{"/usr/bin/java", "-XX:+TieredCompilation", "-XX:TieredStopAtLevel=1", "-XX:+UseSerialGC", "-Xshare:auto", "-cp", "/w", "Main"}, RunEnv: []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"}, RunCPU: 2_000_000_000, RunMem: 536_870_912, @@ -207,7 +207,7 @@ var Languages = map[string]LangConfig{ CompileProc: 50, SourcePath: "/w/Main.kt", BinaryPath: "/w/Main.jar", - RunArgs: []string{"/usr/bin/java", "-jar", "/w/Main.jar"}, + RunArgs: []string{"/usr/bin/java", "-XX:TieredStopAtLevel=1", "-XX:+UseSerialGC", "-Xshare:auto", "-jar", "/w/Main.jar"}, RunEnv: []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"}, RunCPU: 2_000_000_000, RunMem: 536_870_912, diff --git a/internal/worker/worker.go b/internal/worker/worker.go index f9a115e..df7e41a 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -131,6 +131,21 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { log := slog.With("submission_id", submissionID) + // The job's tests live in Redis and do not depend on the database claim, so + // the two round trips overlap. The result is read after the claim below. + type loaded struct { + spec jobSpec + err error + took time.Duration + } + testsCh := make(chan loaded, 1) + go func() { + t := time.Now() + spec, err := w.loadTests(jobCtx, submissionID) + testsCh <- loaded{spec, err, time.Since(t)} + }() + + claimStart := time.Now() sub, err := db.GetSubmission(jobCtx, w.pool, id) if err != nil { return fmt.Errorf("worker: get submission: %w", err) @@ -150,9 +165,12 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { return fmt.Errorf("worker: claim submission: %w", err) } + metrics.JobStageSeconds.WithLabelValues("claim").Observe(time.Since(claimStart).Seconds()) + // Count only jobs we've actually claimed; deferred duration also starts here. metrics.ActiveWorkers.Inc() defer metrics.ActiveWorkers.Dec() + defer metrics.JobsDrain.Complete() // a failed job frees its slot too // finalVerdict is set just before each return so the deferred histogram gets labels. finalVerdict := string(model.VerdictInternalError) defer func() { @@ -176,8 +194,10 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { } }() - // Load the job's tests and limits from Redis - spec, err := w.loadTests(jobCtx, submissionID) + // The job's tests and limits, fetched from Redis while the claim ran. + loadedTests := <-testsCh + metrics.JobStageSeconds.WithLabelValues("load_tests").Observe(loadedTests.took.Seconds()) + spec, err := loadedTests.spec, loadedTests.err if err != nil { return w.failSubmission(jobCtx, id, model.VerdictInternalError, fmt.Sprintf("load tests: %v", err), log) } @@ -257,6 +277,7 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { log.Warn("failed to publish RUNNING event", "err", err) } + runStart := time.Now() results, totalCPU, peakMem, passed, overallVerdict, err := w.runTestsBatched(jobCtx, submissionID, lang, sub.SourceCode, spec.tests, fileID, spec.limits, spec.files, log) if err != nil { return w.failSubmission(jobCtx, id, model.VerdictInternalError, fmt.Sprintf("batch run: %v", err), log) @@ -267,6 +288,8 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second) defer writeCancel() + metrics.JobStageSeconds.WithLabelValues("run").Observe(time.Since(runStart).Seconds()) + persistStart := time.Now() if err := db.InsertResultsAndVerdict(writeCtx, w.pool, id, results, model.StatusCompleted, overallVerdict, "", totalCPU, peakMem, len(spec.tests), passed); err != nil { @@ -277,6 +300,7 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { return fmt.Errorf("worker: write results+verdict: %w", err) } + metrics.JobStageSeconds.WithLabelValues("persist").Observe(time.Since(persistStart).Seconds()) eventResults := verdictResults(results) effectiveCPU, effectiveMem := lang.Effective(spec.limits) @@ -290,9 +314,11 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { "applied_memory_limit_bytes": effectiveMem, "results": eventResults, } + publishStart := time.Now() if err := w.pubsub.PublishEvent(writeCtx, eventlog.New(submissionID, eventlog.EventVerdict, verdictData)); err != nil { log.Error("failed to publish VERDICT event (DB has verdict, SSE client may not)", "err", err) } + metrics.JobStageSeconds.WithLabelValues("publish").Observe(time.Since(publishStart).Seconds()) if spec.callbackURL != "" { go dispatchWebhook(context.WithoutCancel(ctx), spec.callbackURL, spec.callbackSecret, WebhookPayload{