From 25c8c45de4b4c7b392f375afc579c0856ac84157 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Sat, 21 Feb 2026 19:55:38 +0000 Subject: [PATCH] Limit concurrent indexing to two runs Further jobs report a queued phase until a slot frees, so a burst of users cannot exhaust the container's CPU and memory at once. --- main.go | 22 ++++++++++++++++------ main_test.go | 29 +++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/main.go b/main.go index 2cd7284..ce71ad5 100644 --- a/main.go +++ b/main.go @@ -45,7 +45,8 @@ type Job struct { const ( idleExpiry = time.Hour // snapshots unused and jobs untouched for this long are dropped - maxSnapshots = 8 // ponytail: fixed cap sized for the 2GB container; make it configurable if memory changes + maxSnapshots = 8 // ponytail: fixed caps sized for the 2GB container; make them configurable if memory changes + maxIndexing = 2 // concurrent index runs; each already uses every CPU ) type Server struct { @@ -56,6 +57,7 @@ type Server struct { workspace string // parent of every job directory; only paths under it are ever removed cors string static http.Handler + indexSlots chan struct{} } func NewServer(workspace string) *Server { @@ -65,11 +67,12 @@ func NewServer(workspace string) *Server { } web, _ := fs.Sub(webFiles, "web") return &Server{ - jobs: map[string]*Job{}, - snapshots: map[string]*Snapshot{}, - workspace: workspace, - cors: cors, - static: http.FileServerFS(web), + jobs: map[string]*Job{}, + snapshots: map[string]*Snapshot{}, + workspace: workspace, + cors: cors, + static: http.FileServerFS(web), + indexSlots: make(chan struct{}, maxIndexing), } } @@ -517,6 +520,13 @@ func (s *Server) localJobRoot(id string) (string, bool) { } func (s *Server) indexJob(id, root string) { + select { + case s.indexSlots <- struct{}{}: + default: + s.updateJob(id, "queued", 0, 0, "Waiting for other indexing to finish…") + s.indexSlots <- struct{}{} + } + defer func() { <-s.indexSlots }() s.updateJob(id, "scanning", 0, 0, "Discovering source files…") snapshot, err := indexDirectory(root, func(phase string, completed, total int, message string) { s.updateJob(id, phase, completed, total, message) diff --git a/main_test.go b/main_test.go index 5c040fc..70925ef 100644 --- a/main_test.go +++ b/main_test.go @@ -361,3 +361,32 @@ func TestSnapshotCapEvictsLeastRecentlyUsed(t *testing.T) { } first.json("GET", "/api/snapshots/"+oldest, "", 404, nil) } + +func TestIndexingQueuesBeyondConcurrencyLimit(t *testing.T) { + server, c := newTestServer(t) + for range maxIndexing { + server.indexSlots <- struct{}{} + } + var created struct{ JobID string } + c.json("POST", "/api/jobs", `{"kind":"local"}`, 201, &created) + c.json("POST", "/api/jobs/"+created.JobID+"/files?path=a.go", "package a", 200, nil) + c.json("POST", "/api/jobs/"+created.JobID+"/commit", "", 202, nil) + var job Job + for range 100 { + if c.json("GET", "/api/jobs/"+created.JobID, "", 200, &job); job.Phase == "queued" { + break + } + time.Sleep(5 * time.Millisecond) + } + if job.Phase != "queued" { + t.Fatalf("phase = %q, want queued", job.Phase) + } + <-server.indexSlots + for range 200 { + if c.json("GET", "/api/jobs/"+created.JobID, "", 200, &job); job.SnapshotID != nil { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("job never finished after a slot was freed: %+v", job) +}