diff --git a/CHANGELOG.md b/CHANGELOG.md index 68ed908..dab2e2b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,10 @@ and uses [semantic versioning](https://semver.org/). ### Changed +- **One compile for the same source** — jobs with the same source, language, files and flags that arrive together share a single compile instead of each starting their own. A source that does not compile fails the same way for all of them; a sandbox error in the one compiling makes the others try for themselves. +- **`output` mode** — `POST /api/v1/execute` accepts `output`: `full` (default), `preview` or `none`, to keep full stdout and stderr out of the event log, SSE and webhook when only verdicts are needed. +- **Pool sizes** — `AC_DB_MAX_CONNS`, `AC_REDIS_POOL_SIZE` and `AC_REDIS_MIN_IDLE_CONNS` (defaults 128, 256, 32, as before). +- **Latency alerts** — Prometheus rules for queue-wait p99, slow database or Redis stages, and a compile cache that almost never hits. - **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. diff --git a/README.md b/README.md index 1798f6d..547481e 100644 --- a/README.md +++ b/README.md @@ -314,6 +314,7 @@ Submit code for evaluation against test cases. | `priority` | string | No | Priority tier: `run` (P0 default), `submit` (P1), or `bulk` (P2) | | `time_limit_ms` | integer | No | Per-test CPU time ceiling in milliseconds (100 to 20000) | | `memory_limit_kb` | integer | No | Per-test memory ceiling in kilobytes (16384 to 2097152) | +| `output` | string | No | How much output each test's result carries: `full` (default; stdout and stderr in full plus 256-byte previews), `preview` (previews only) or `none` | | `stack_limit_kb` | integer | No | Per-test stack size limit in kilobytes (1024 to 1048576) | | `max_processes` | integer | No | Per-test process/thread limit (1 to 512); the language default otherwise | | `args` | array | No | Command-line arguments passed to the program on every test, as separate argv entries, never through a shell (max 16, 4 KB in total; not for `sqlite`) | @@ -622,6 +623,12 @@ If queue wait dominates, add capacity; if `run` dominates, look at the language - 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. +**Pools.** The database pool defaults to 128 connections (`AC_DB_MAX_CONNS`) and the Redis pool to 256 (`AC_REDIS_POOL_SIZE`, with `AC_REDIS_MIN_IDLE_CONNS` kept open). Ingest can use up to 512 database connections at once, so under heavy submission traffic raise `AC_DB_MAX_CONNS` (and PostgreSQL's `max_connections`) first; watch `ac_job_stage_seconds{stage="claim"}` and `{stage="persist"}`. + +**Compiles.** Jobs with the same source, language, files and flags that arrive together share one compile (the rest wait for it), and later ones hit the compile cache (`AC_COMPILE_CACHE_TTL`). A caller that submits one job per test of the same program therefore compiles once, but a single job with all the tests is cheaper still: one queue entry, one event stream and one compile. + +**Output.** A job's full stdout and stderr go through Redis, the event log and every SSE client. Send `"output": "preview"` or `"none"` when only verdicts matter. + **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 diff --git a/api/openapi.yaml b/api/openapi.yaml index 882af4d..c3319cc 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -175,6 +175,17 @@ components: Per-test memory limit override in kilobytes (16384–2097152 = 16 MB–2 GB). Omit to use the language default. example: 262144 + output: + type: string + enum: [full, preview, none] + default: full + description: | + How much of each test's output the VERDICT event, the event log and the + webhook carry. `full` includes stdout and stderr in full (up to 8 MB per + test) beside the 256-byte previews. `preview` keeps only the previews and + `none` drops all output text. A large job's full output goes through Redis + and every SSE client, so callers that only need verdicts should not ask for it. + example: preview stack_limit_kb: type: integer format: int64 diff --git a/cmd/stj-exec/main.go b/cmd/stj-exec/main.go index 996611f..48d3d5d 100644 --- a/cmd/stj-exec/main.go +++ b/cmd/stj-exec/main.go @@ -73,7 +73,7 @@ func main() { slog.Info("go-judge connected", "url", cfg.JudgeURL) // Database - pool, err := db.Connect(ctx, cfg.DatabaseURL) + pool, err := db.ConnectWithMaxConns(ctx, cfg.DatabaseURL, cfg.DBMaxConns) if err != nil { slog.Error("database connection failed", "err", err) os.Exit(1) @@ -89,8 +89,8 @@ func main() { // Redis rdb := redis.NewClient(&redis.Options{ Addr: cfg.RedisAddr, - PoolSize: 256, - MinIdleConns: 32, + PoolSize: cfg.RedisPool, + MinIdleConns: cfg.RedisMinIdle, }) if err := rdb.Ping(ctx).Err(); err != nil { slog.Error("redis connection failed", "addr", cfg.RedisAddr, "err", err) diff --git a/deploy/monitoring/alerts.yml b/deploy/monitoring/alerts.yml index d9d0077..5143ee9 100644 --- a/deploy/monitoring/alerts.yml +++ b/deploy/monitoring/alerts.yml @@ -78,3 +78,43 @@ groups: description: > Prometheus cannot scrape the aethercode-exec metrics endpoint. The process may have crashed or the host may be unreachable. + + # ── Latency (see README, "Performance Tuning") ────────────────────────── + - alert: QueueWaitP99High + expr: histogram_quantile(0.99, sum by (le, stream) (rate(ac_queue_wait_seconds_bucket[5m]))) > 30 + for: 5m + labels: + severity: warning + annotations: + summary: "Jobs in {{ $labels.stream }} wait more than 30s at the p99" + description: > + The p99 wait between enqueue and a worker reading the job is + {{ $value | humanizeDuration }}. Workers or sandbox slots are the + bottleneck: compare AC_WORKER_COUNT with go-judge -parallelism and CPU. + + - alert: JobStageP99High + expr: histogram_quantile(0.99, sum by (le, stage) (rate(ac_job_stage_seconds_bucket{stage=~"claim|load_tests|persist|publish"}[5m]))) > 1 + for: 5m + labels: + severity: warning + annotations: + summary: "The {{ $labels.stage }} stage takes more than 1s at the p99" + description: > + claim, persist and publish are database and Redis round trips and + load_tests is a Redis read. Check PostgreSQL and Redis latency and + the connection pool sizes (AC_DB_MAX_CONNS, AC_REDIS_POOL_SIZE). + + - alert: CompileCacheMostlyMissing + expr: > + sum(rate(ac_compile_cache_misses_total[15m])) + / (sum(rate(ac_compile_cache_hits_total[15m])) + sum(rate(ac_compile_cache_misses_total[15m]))) > 0.9 + and sum(rate(ac_compile_cache_misses_total[15m])) > 0.5 + for: 15m + labels: + severity: info + annotations: + summary: "Almost every compile is a cache miss" + description: > + Expected when every job carries different source. If one source is + sent as many jobs (one per test), check that the caller is not + defeating the compile cache. diff --git a/deploy/systemd/env.example b/deploy/systemd/env.example index 5afe474..433498b 100644 --- a/deploy/systemd/env.example +++ b/deploy/systemd/env.example @@ -4,6 +4,10 @@ AC_REDIS_ADDR=127.0.0.1:6379 AC_JUDGE_URL=http://127.0.0.1:5050 AC_JUDGE_TOKEN=CHANGEME AC_WORKER_COUNT=64 +# Connection pools; raise AC_DB_MAX_CONNS (and PostgreSQL max_connections) under heavy submission traffic. +# AC_DB_MAX_CONNS=128 +# AC_REDIS_POOL_SIZE=256 +# AC_REDIS_MIN_IDLE_CONNS=32 AC_LOG_LEVEL=info # json | text | gcp (Google Cloud Logging) | cloudwatch (AWS CloudWatch Logs) AC_LOG_FORMAT=json diff --git a/internal/config/config.go b/internal/config/config.go index 5c6c3e8..97e1937 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,12 +12,18 @@ import ( ) type Config struct { - GatewayAddr string - DatabaseURL string - RedisAddr string - JudgeURL string - JudgeToken string - WorkerCount int + GatewayAddr string + DatabaseURL string + RedisAddr string + JudgeURL string + JudgeToken string + WorkerCount int + // Connection pool sizes. The defaults suit up to about 128 concurrent + // database users; ingest alone can use 512 at once (see the admission + // semaphore), so raise DBMaxConns before raising ingest concurrency. + DBMaxConns int + RedisPool int + RedisMinIdle int StreamName string ConsumerGroup string LogLevel string @@ -83,6 +89,9 @@ func Load() Config { JudgeURL: env("AC_JUDGE_URL", "http://127.0.0.1:5050"), JudgeToken: env("AC_JUDGE_TOKEN", ""), WorkerCount: envInt("AC_WORKER_COUNT", 8), + DBMaxConns: envInt("AC_DB_MAX_CONNS", 128), + RedisPool: envInt("AC_REDIS_POOL_SIZE", 256), + RedisMinIdle: envInt("AC_REDIS_MIN_IDLE_CONNS", 32), StreamName: env("AC_STREAM_NAME", "ac:submissions"), ConsumerGroup: env("AC_CONSUMER_GROUP", "workers"), LogLevel: env("AC_LOG_LEVEL", "info"), @@ -136,6 +145,9 @@ func (c Config) Validate() error { } check(c.WorkerCount > 0 && c.WorkerCount <= 512, "WorkerCount must be 1-512") + check(c.DBMaxConns >= 4 && c.DBMaxConns <= 2000, "DBMaxConns must be 4-2000") + check(c.RedisPool >= 4 && c.RedisPool <= 10000, "RedisPool must be 4-10000") + check(c.RedisMinIdle >= 0 && c.RedisMinIdle <= c.RedisPool, "RedisMinIdle must be 0 up to RedisPool") check(c.SubmitWeight > 0, "SubmitWeight must be > 0") check(c.BulkWeight > 0, "BulkWeight must be > 0") check(c.AdmissionMaxQueueDepth > 0, "AdmissionMaxQueueDepth must be > 0") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5a1d0cd..726c256 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -14,6 +14,9 @@ func validConfig() Config { JudgeURL: "http://127.0.0.1:5050", JudgeToken: "a-secret-token", WorkerCount: 8, + DBMaxConns: 128, + RedisPool: 256, + RedisMinIdle: 32, TestParallelism: 4, StreamName: "ac:submissions", ConsumerGroup: "workers", @@ -112,3 +115,22 @@ func TestEnvDuration(t *testing.T) { t.Fatalf("envDuration unset = %v, want fallback 3s", got) } } + +func TestValidate_PoolSizes(t *testing.T) { + for name, mutate := range map[string]func(*Config){ + "db too small": func(c *Config) { c.DBMaxConns = 1 }, + "db too large": func(c *Config) { c.DBMaxConns = 5000 }, + "redis too small": func(c *Config) { c.RedisPool = 1 }, + "idle above pool": func(c *Config) { c.RedisMinIdle = c.RedisPool + 1 }, + "negative idle": func(c *Config) { c.RedisMinIdle = -1 }, + } { + cfg := validConfig() + mutate(&cfg) + if err := cfg.Validate(); err == nil { + t.Errorf("%s: Validate accepted it", name) + } + } + if err := validConfig().Validate(); err != nil { + t.Errorf("valid config rejected: %v", err) + } +} diff --git a/internal/db/db.go b/internal/db/db.go index 0c9e951..29dc5c8 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -28,13 +28,25 @@ func buildPool(ctx context.Context, config *pgxpool.Config, connectPrefix, pingP return pool, nil } +// DefaultMaxConns is the main pool's size unless the caller sets one. +const DefaultMaxConns = 128 + func Connect(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) { + return ConnectWithMaxConns(ctx, databaseURL, DefaultMaxConns) +} + +// ConnectWithMaxConns is Connect with the pool's size set by the caller. The +// pool keeps a sixteenth of it open (at least 4, at most 16). +func ConnectWithMaxConns(ctx context.Context, databaseURL string, maxConns int) (*pgxpool.Pool, error) { config, err := pgxpool.ParseConfig(databaseURL) if err != nil { return nil, fmt.Errorf("db: parse config: %w", err) } - config.MinConns = 16 - config.MaxConns = 128 + if maxConns < 4 || maxConns > 2000 { + return nil, fmt.Errorf("db: max connections %d outside 4-2000", maxConns) + } + config.MinConns = int32(max(4, min(16, maxConns/16))) + config.MaxConns = int32(maxConns) // Optimize query execution by caching prepared statements and query descriptions config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeCacheDescribe return buildPool(ctx, config, "db: connect", "db: ping") diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go index 19bb671..3ae58b7 100644 --- a/internal/gateway/handler.go +++ b/internal/gateway/handler.go @@ -177,6 +177,8 @@ type executeRequest struct { CallbackSecret string `json:"callback_secret,omitempty"` // Optional priority tier: "run" (default / interactive), "submit" (contest / assessment), or "bulk" (batch re-judge). Priority string `json:"priority"` + // Optional output mode: "full" (default), "preview" (256-byte previews only) or "none". + Output string `json:"output"` // Mode and ProblemVersionID exist only to reject legacy submit-mode requests. Mode string `json:"mode"` @@ -354,6 +356,10 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) { jsonError(w, "callback_secret exceeds 256 bytes", http.StatusBadRequest) return } + if !model.ValidOutputMode(req.Output) { + jsonError(w, "output must be one of: full, preview, none", http.StatusBadRequest) + return + } if !model.ValidPriority(req.Priority) { jsonError(w, "priority must be one of: run, submit, bulk", http.StatusBadRequest) return @@ -517,6 +523,7 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) { // CallbackSecret is validated above (length cap). Encryption at rest // before writing to Redis is handled by the pubsub layer (F18/F8). CallbackSecret: req.CallbackSecret, + Output: req.Output, } if err := h.pubsub.PrepareAndEnqueue(ctx, jobID.String(), testSet, queuedEvt, targetStream.StreamName()); err != nil { slog.Error("enqueue failed", "submission_id", jobID, "err", err) diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go index d0d69cf..5178b2c 100644 --- a/internal/gateway/handler_test.go +++ b/internal/gateway/handler_test.go @@ -653,6 +653,9 @@ func TestHandleExecute_RejectsInvalidRunControls(t *testing.T) { {"too many args", `{"language":"c","source_code":"int main(){}","args":["1","2","3","4","5","6","7","8","9","10","11","12","13","14","15","16","17"],"tests":[{"input":"","expected_output":""}]}`, "args: at most 16 arguments"}, + {"unknown output mode", + `{"language":"c","source_code":"int main(){}","output":"everything","tests":[{"input":"","expected_output":""}]}`, + "output must be one of: full, preview, none"}, {"args for sqlite", `{"language":"sqlite","source_code":"select 1;","args":["x"],"tests":[{"input":"","expected_output":"1"}]}`, "args are not supported for sqlite"}, diff --git a/internal/model/submission.go b/internal/model/submission.go index f016d7c..f7488a8 100644 --- a/internal/model/submission.go +++ b/internal/model/submission.go @@ -90,6 +90,23 @@ func ValidPriority(s string) bool { return false } +// Output modes: how much of each test's output the VERDICT event, the event +// log and the webhook carry. +const ( + OutputFull = "full" // stdout and stderr in full, plus 256-byte previews (default) + OutputPreview = "preview" // 256-byte previews only + OutputNone = "none" // no output text at all +) + +// ValidOutputMode reports whether s is an output mode. The empty string is valid and means full. +func ValidOutputMode(s string) bool { + switch s { + case "", OutputFull, OutputPreview, OutputNone: + return true + } + return false +} + // ValidChecker reports whether s names a supported checker. The empty string // is valid and means exact. Custom checkers can be specified as "custom" or when checker code is provided. func ValidChecker(s string) bool { diff --git a/internal/queue/pubsub.go b/internal/queue/pubsub.go index c73a8c1..7ed4e12 100644 --- a/internal/queue/pubsub.go +++ b/internal/queue/pubsub.go @@ -110,6 +110,7 @@ func (p *PubSub) PrepareAndEnqueue(ctx context.Context, jobID string, set TestSe "interactor_language", set.InteractorLanguage, "callback_url", set.CallbackURL, "callback_secret", callbackSecretField, + "output_mode", set.Output, ) if len(set.Files) > 0 { if filesJSON, err := json.Marshal(set.Files); err == nil { @@ -379,6 +380,7 @@ func (p *PubSub) GetTests(ctx context.Context, jobID string) (TestSet, error) { CompilerFlags: compilerFlags, CallbackURL: fields["callback_url"], CallbackSecret: callbackSecret, + Output: fields["output_mode"], // absent on jobs enqueued before output modes: empty means full }, nil } @@ -402,4 +404,5 @@ type TestSet struct { CompilerFlags []string // optional whitelisted compiler flags CallbackURL string // optional webhook notification callback URL CallbackSecret string // optional HMAC signing secret for webhook + Output string // how much output the verdict event carries: full (empty), preview or none } diff --git a/internal/queue/pubsub_test.go b/internal/queue/pubsub_test.go index 61fb633..b74358e 100644 --- a/internal/queue/pubsub_test.go +++ b/internal/queue/pubsub_test.go @@ -572,6 +572,7 @@ func TestPrepareAndGetTests_RoundTripsRunControls(t *testing.T) { set := TestSet{ Tests: []TestData{{Input: "1", ExpectedOutput: "1"}}, Limits: sandbox.RunLimits{CPU: 1_000_000_000, Memory: 64 << 20, Stack: 8 << 20, Procs: 4, Args: []string{"-n", "5"}}, + Output: "preview", } evt := eventlog.New("job-controls", eventlog.EventQueued, nil) if err := ps.PrepareAndEnqueue(ctx, "job-controls", set, evt, "ac:submissions:run"); err != nil { @@ -584,6 +585,9 @@ func TestPrepareAndGetTests_RoundTripsRunControls(t *testing.T) { if !reflect.DeepEqual(got.Limits, set.Limits) { t.Errorf("Limits = %+v, want %+v", got.Limits, set.Limits) } + if got.Output != "preview" { + t.Errorf("Output = %q, want preview", got.Output) + } } func TestGetTests_RejectsCorruptArgs(t *testing.T) { diff --git a/internal/worker/compilecache.go b/internal/worker/compilecache.go index e3347b9..2cea1c2 100644 --- a/internal/worker/compilecache.go +++ b/internal/worker/compilecache.go @@ -1,6 +1,7 @@ package worker import ( + "context" "crypto/sha256" "encoding/binary" "encoding/hex" @@ -15,18 +16,91 @@ type cacheEntry struct { } type compileCache struct { - mu sync.Mutex - m map[string]cacheEntry - ttl time.Duration + mu sync.Mutex + m map[string]cacheEntry + inflight map[string]*compileFlight + ttl time.Duration } func newCompileCache(ttl time.Duration) *compileCache { return &compileCache{ - m: make(map[string]cacheEntry), - ttl: ttl, + m: make(map[string]cacheEntry), + inflight: make(map[string]*compileFlight), + ttl: ttl, } } +// compileOutcome is what one compile produced: the cached binary's file id, or +// the compiler's diagnostics when the source does not compile. +type compileOutcome struct { + fileID string + failed bool + stderr string +} + +// compileFlight is one compile in progress, shared by every job that asked for +// the same source while it ran. +type compileFlight struct { + done chan struct{} + outcome compileOutcome + err error +} + +// compile returns the outcome for key, running run at most once for jobs that +// arrive together. Jobs that arrive while a compile is running wait for it +// instead of repeating it (shared is true for them), and jobs that arrive after +// it succeeds hit the cache. A source that does not compile is shared with the +// jobs that waited, because the same source fails the same way; it is not +// cached. If the leader's run fails for a reason of its own (the sandbox was +// unreachable, say), each waiter tries again for itself. With a zero ttl the +// cache is off and every call runs run. +func (c *compileCache) compile(ctx context.Context, key string, run func() (compileOutcome, error)) (outcome compileOutcome, shared bool, err error) { + if c.ttl == 0 { + outcome, err = run() + return outcome, false, err + } + for { + c.mu.Lock() + if e, ok := c.m[key]; ok && time.Since(e.created) <= c.ttl { + c.mu.Unlock() + return compileOutcome{fileID: e.fileID}, true, nil + } + flight, running := c.inflight[key] + if !running { + flight = &compileFlight{done: make(chan struct{})} + c.inflight[key] = flight + c.mu.Unlock() + return c.lead(key, flight, run) + } + c.mu.Unlock() + + select { + case <-flight.done: + if flight.err == nil { + return flight.outcome, true, nil + } + // The leader failed for its own reasons: try again, as the leader if no one else has. + case <-ctx.Done(): + return compileOutcome{}, false, ctx.Err() + } + } +} + +func (c *compileCache) lead(key string, flight *compileFlight, run func() (compileOutcome, error)) (compileOutcome, bool, error) { + defer func() { + c.mu.Lock() + delete(c.inflight, key) + c.mu.Unlock() + close(flight.done) + }() + outcome, err := run() + flight.outcome, flight.err = outcome, err + if err == nil && !outcome.failed { + c.put(key, outcome.fileID) + } + return outcome, false, err +} + func compileCacheKey(language, source string) string { return compileCacheKeyWithOptions(language, source, nil, nil) } diff --git a/internal/worker/compilecache_test.go b/internal/worker/compilecache_test.go index 0cb461e..c37c7a8 100644 --- a/internal/worker/compilecache_test.go +++ b/internal/worker/compilecache_test.go @@ -1,6 +1,10 @@ package worker import ( + "context" + "errors" + "sync" + "sync/atomic" "testing" "time" ) @@ -161,3 +165,159 @@ func BenchmarkCompileCacheKey(b *testing.B) { _ = compileCacheKey("python", largeSource) } } + +func TestCompileCache_ConcurrentJobsShareOneCompile(t *testing.T) { + t.Parallel() + c := newCompileCache(time.Minute) + var runs atomic.Int32 + release := make(chan struct{}) + run := func() (compileOutcome, error) { + runs.Add(1) + <-release + return compileOutcome{fileID: "bin-1"}, nil + } + + const callers = 20 + var wg sync.WaitGroup + shared := make([]bool, callers) + ids := make([]string, callers) + for i := range callers { + wg.Add(1) + go func() { + defer wg.Done() + out, sh, err := c.compile(context.Background(), "k", run) + if err != nil { + t.Errorf("compile: %v", err) + } + shared[i], ids[i] = sh, out.fileID + }() + } + time.Sleep(100 * time.Millisecond) // let every caller reach the wait + close(release) + wg.Wait() + + if runs.Load() != 1 { + t.Fatalf("compile ran %d times for %d concurrent jobs, want once", runs.Load(), callers) + } + leaders := 0 + for i := range callers { + if ids[i] != "bin-1" { + t.Errorf("caller %d got %q", i, ids[i]) + } + if !shared[i] { + leaders++ + } + } + if leaders != 1 { + t.Errorf("%d callers report having compiled, want exactly 1", leaders) + } + // Later jobs hit the cache without compiling. + if out, sh, _ := c.compile(context.Background(), "k", run); !sh || out.fileID != "bin-1" || runs.Load() != 1 { + t.Errorf("a later job should hit the cache: %+v shared=%v runs=%d", out, sh, runs.Load()) + } +} + +func TestCompileCache_ASourceThatDoesNotCompileIsSharedButNotCached(t *testing.T) { + t.Parallel() + c := newCompileCache(time.Minute) + var runs atomic.Int32 + release := make(chan struct{}) + run := func() (compileOutcome, error) { + runs.Add(1) + <-release + return compileOutcome{failed: true, stderr: "syntax error"}, nil + } + results := make(chan compileOutcome, 2) + for range 2 { + go func() { + out, _, _ := c.compile(context.Background(), "bad", run) + results <- out + }() + } + time.Sleep(100 * time.Millisecond) + close(release) + for range 2 { + if out := <-results; !out.failed || out.stderr != "syntax error" { + t.Fatalf("outcome = %+v, want the shared compile failure", out) + } + } + if runs.Load() != 1 { + t.Fatalf("ran %d times, want once", runs.Load()) + } + // A failure is not cached: the next job compiles again. + if _, shared, _ := c.compile(context.Background(), "bad", func() (compileOutcome, error) { runs.Add(1); return compileOutcome{failed: true}, nil }); shared || runs.Load() != 2 { + t.Fatalf("the failed source was cached: shared=%v runs=%d", shared, runs.Load()) + } +} + +func TestCompileCache_WaitersRetryWhenTheLeaderHitsATransientError(t *testing.T) { + t.Parallel() + c := newCompileCache(time.Minute) + var runs atomic.Int32 + started := make(chan struct{}) + release := make(chan struct{}) + leader := func() (compileOutcome, error) { + runs.Add(1) + close(started) + <-release + return compileOutcome{}, errors.New("sandbox unreachable") + } + errCh := make(chan error, 1) + go func() { _, _, err := c.compile(context.Background(), "k", leader); errCh <- err }() + <-started + waiter := make(chan compileOutcome, 1) + go func() { + out, _, _ := c.compile(context.Background(), "k", func() (compileOutcome, error) { + runs.Add(1) + return compileOutcome{fileID: "bin-2"}, nil + }) + waiter <- out + }() + time.Sleep(100 * time.Millisecond) + close(release) + if err := <-errCh; err == nil { + t.Fatal("the leader's error was swallowed") + } + if out := <-waiter; out.fileID != "bin-2" { + t.Fatalf("waiter got %+v, want its own successful compile after the leader failed", out) + } + if runs.Load() != 2 { + t.Fatalf("ran %d times, want the failed leader plus one retry", runs.Load()) + } +} + +func TestCompileCache_AWaiterGivesUpWhenItsContextEnds(t *testing.T) { + t.Parallel() + c := newCompileCache(time.Minute) + started := make(chan struct{}) + release := make(chan struct{}) + go func() { + _, _, _ = c.compile(context.Background(), "k", func() (compileOutcome, error) { + close(started) + <-release + return compileOutcome{fileID: "x"}, nil + }) + }() + <-started + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, _, err := c.compile(ctx, "k", nil); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want the waiter's deadline", err) + } + close(release) +} + +func TestCompileCache_ZeroTTLCompilesEveryTime(t *testing.T) { + t.Parallel() + c := newCompileCache(0) + var runs int + for range 3 { + _, shared, _ := c.compile(context.Background(), "k", func() (compileOutcome, error) { runs++; return compileOutcome{fileID: "x"}, nil }) + if shared { + t.Fatal("a disabled cache reported a shared compile") + } + } + if runs != 3 { + t.Fatalf("ran %d times, want 3", runs) + } +} diff --git a/internal/worker/verdict_test.go b/internal/worker/verdict_test.go index bbd2458..c170e7c 100644 --- a/internal/worker/verdict_test.go +++ b/internal/worker/verdict_test.go @@ -52,3 +52,35 @@ func TestVerdictResults_TruncatesTheSandboxMessage(t *testing.T) { t.Fatalf("message length = %d, want 512", len(s)) } } + +func TestVerdictResultsFor_OutputModes(t *testing.T) { + results := []model.TestResult{{TestIndex: 0, Verdict: model.VerdictAccepted, StdoutPreview: strings.Repeat("o", 300), StderrPreview: "warn", ExitStatus: 0}} + keys := func(mode string) map[string]bool { + out := map[string]bool{} + for k := range verdictResultsFor(results, mode)[0] { + out[k] = true + } + return out + } + for _, mode := range []string{"", model.OutputFull} { + k := keys(mode) + if !k["stdout"] || !k["stderr"] || !k["stdout_preview"] || !k["stderr_preview"] { + t.Errorf("mode %q dropped output: %v", mode, k) + } + } + k := keys(model.OutputPreview) + if k["stdout"] || k["stderr"] || !k["stdout_preview"] || !k["stderr_preview"] { + t.Errorf("preview mode keys = %v, want previews only", k) + } + k = keys(model.OutputNone) + if k["stdout"] || k["stderr"] || k["stdout_preview"] || k["stderr_preview"] { + t.Errorf("none mode keys = %v, want no output text", k) + } + // Verdicts and measurements always stay. + for _, mode := range []string{model.OutputPreview, model.OutputNone} { + k := keys(mode) + if !k["verdict"] || !k["cpu_time_ns"] || !k["exit_status"] || !k["test_index"] { + t.Errorf("mode %q lost the verdict or measurements: %v", mode, k) + } + } +} diff --git a/internal/worker/worker.go b/internal/worker/worker.go index df7e41a..f018d70 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -2,6 +2,7 @@ package worker import ( "context" + "errors" "fmt" "log/slog" "strings" @@ -112,6 +113,7 @@ type jobSpec struct { compilerFlags []string callbackURL string callbackSecret string + output string } func isAlreadyTerminal(err error) bool { @@ -234,29 +236,35 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { } cacheKey := compileCacheKeyWithOptions(sub.Language, sub.SourceCode, spec.files, spec.compilerFlags) - if cached, hit := w.cc.get(cacheKey); hit { - fileID = cached - metrics.CompileCacheHitsTotal.Inc() - log.Info("compile cache hit", "fileId", fileID) - } else { + // Jobs with the same source that arrive together share one compile. + outcome, shared, err := w.cc.compile(jobCtx, cacheKey, func() (compileOutcome, error) { compileStart := time.Now() compileResult, err := w.compileWithOptions(jobCtx, lang, sub.SourceCode, spec.files, spec.compilerFlags) metrics.CompileDuration.WithLabelValues(sub.Language).Observe(time.Since(compileStart).Seconds()) if err != nil { - return w.failSubmission(jobCtx, id, model.VerdictInternalError, err.Error(), log) + return compileOutcome{}, err } - if compileResult.Status != sandbox.StatusAccepted { - stderr := compileResult.Files["stderr"] - finalVerdict = string(model.VerdictCompilationError) - return w.failSubmission(jobCtx, id, model.VerdictCompilationError, stderr, log) + return compileOutcome{failed: true, stderr: compileResult.Files["stderr"]}, nil } - - fileID = firstFileID(compileResult.FileIDs) - if fileID == "" { - return w.failSubmission(jobCtx, id, model.VerdictInternalError, "no cached file returned from compile", log) + compiledID := firstFileID(compileResult.FileIDs) + if compiledID == "" { + return compileOutcome{}, errors.New("no cached file returned from compile") } - w.cc.put(cacheKey, fileID) + return compileOutcome{fileID: compiledID}, nil + }) + if err != nil { + return w.failSubmission(jobCtx, id, model.VerdictInternalError, err.Error(), log) + } + if outcome.failed { + finalVerdict = string(model.VerdictCompilationError) + return w.failSubmission(jobCtx, id, model.VerdictCompilationError, outcome.stderr, log) + } + fileID = outcome.fileID + if shared { + metrics.CompileCacheHitsTotal.Inc() + log.Info("compile cache hit", "fileId", fileID) + } else { metrics.CompileCacheMissesTotal.Inc() log.Info("compiled", "fileId", fileID) } @@ -301,7 +309,7 @@ func (w *Worker) process(ctx context.Context, submissionID string) error { } metrics.JobStageSeconds.WithLabelValues("persist").Observe(time.Since(persistStart).Seconds()) - eventResults := verdictResults(results) + eventResults := verdictResultsFor(results, spec.output) effectiveCPU, effectiveMem := lang.Effective(spec.limits) verdictData := map[string]any{ @@ -365,6 +373,7 @@ func (w *Worker) loadTests(ctx context.Context, jobID string) (jobSpec, error) { compilerFlags: set.CompilerFlags, callbackURL: set.CallbackURL, callbackSecret: set.CallbackSecret, + output: set.Output, }, nil } @@ -372,6 +381,30 @@ func (w *Worker) loadTests(ctx context.Context, jobID string) (jobSpec, error) { // supplied the tests, so every test carries its full output (bounded by the // sandbox output limit) plus a 256-byte preview. func verdictResults(results []model.TestResult) []map[string]any { + return verdictResultsFor(results, model.OutputFull) +} + +// verdictResultsFor is verdictResults with the caller's output mode: "preview" +// drops the full stdout and stderr, "none" drops the output text altogether. +// A large job's full output can run to megabytes per test, all of which goes +// through Redis and every SSE client, so callers that only need verdicts can +// leave it out. +func verdictResultsFor(results []model.TestResult, mode string) []map[string]any { + out := verdictResultsFull(results) + if mode == model.OutputPreview || mode == model.OutputNone { + for _, entry := range out { + delete(entry, "stdout") + delete(entry, "stderr") + if mode == model.OutputNone { + delete(entry, "stdout_preview") + delete(entry, "stderr_preview") + } + } + } + return out +} + +func verdictResultsFull(results []model.TestResult) []map[string]any { out := make([]map[string]any, len(results)) for i, r := range results { out[i] = map[string]any{