diff --git a/main.go b/main.go index 6221268..c6cb7e2 100644 --- a/main.go +++ b/main.go @@ -51,6 +51,8 @@ const ( maxIndexing = 2 // concurrent index runs; each already uses every CPU ) +var sseKeepAlive = 15 * time.Second // variable so tests can shorten it + type Server struct { jobsMu sync.Mutex jobs map[string]*Job @@ -488,20 +490,28 @@ func (s *Server) streamJobEvents(w http.ResponseWriter, r *http.Request, id stri last := "" ticker := time.NewTicker(150 * time.Millisecond) defer ticker.Stop() - for range 1200 { + lastWrite := time.Now() + // Runs until the job finishes, disappears, or the client leaves; queued jobs can wait a while. + for { job, ok := s.jobSnapshot(id) if !ok { return } encoded := marshal(job) + chunk := "" if string(encoded) != last { - if _, err := fmt.Fprintf(w, "event: progress\ndata: %s\n\n", encoded); err != nil { + chunk, last = fmt.Sprintf("event: progress\ndata: %s\n\n", encoded), string(encoded) + } else if time.Since(lastWrite) >= sseKeepAlive { + chunk = ": keep-alive\n\n" // stops proxies timing out an idle stream + } + if chunk != "" { + if _, err := io.WriteString(w, chunk); err != nil { return } if flusher != nil { flusher.Flush() } - last = string(encoded) + lastWrite = time.Now() } if job.SnapshotID != nil || job.Error != nil { return diff --git a/main_test.go b/main_test.go index 70925ef..af8fb9c 100644 --- a/main_test.go +++ b/main_test.go @@ -390,3 +390,47 @@ func TestIndexingQueuesBeyondConcurrencyLimit(t *testing.T) { } t.Fatalf("job never finished after a slot was freed: %+v", job) } + +func TestProgressStreamSendsKeepAlivesWhileQueued(t *testing.T) { + defer func(old time.Duration) { sseKeepAlive = old }(sseKeepAlive) + sseKeepAlive = 300 * time.Millisecond + 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) + + res, err := c.http.Get(c.url + "/api/jobs/" + created.JobID + "/events") + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + lines := bufio.NewScanner(res.Body) + deadline := time.After(5 * time.Second) + for { + select { + case <-deadline: + t.Fatal("no keep-alive before the deadline") + default: + } + if !lines.Scan() { + t.Fatal("stream closed while job was queued") + } + if lines.Text() == ": keep-alive" { + break + } + } + <-server.indexSlots + var job Job + for lines.Scan() { + if data, ok := strings.CutPrefix(lines.Text(), "data: "); ok { + json.Unmarshal([]byte(data), &job) + } + } + if job.SnapshotID == nil { + t.Fatalf("stream ended without a snapshot: %+v", job) + } +}