Keep progress streams open until the job finishes

The stream used to close after three minutes, which a job queued behind
other users could exceed. It now runs until the job completes or the
client disconnects, sending a keep-alive comment every 15 seconds so
nginx does not time out an idle stream.

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

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)
+	}
+}