Live run logs via SSE log-watch - #325
Conversation
POST /run can return 202 {id} for Accept: text/event-stream; GET
/run/:id/log-watch streams stdout as SSE. Workers emit AMQP log events
while the process runs, then a done payload. JSON clients still wait
for the existing response body.
Co-authored-by: Max Schmitt <max@schmitt.mx>
Restore the truncated C# API example, avoid template-literal parsing issues in the new SSE test, and wait for the duration line instead of EventSource finished() so visual runs are not racy with one worker. Co-authored-by: Max Schmitt <max@schmitt.mx>
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b2b2e5ace3
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| s.mu.Lock() | ||
| s.logs = append(s.logs, line) | ||
| subs := append([]chan runEvent(nil), s.subs...) |
There was a problem hiding this comment.
Bound the replay history by bytes or lines before appending here. A user-controlled snippet can continuously print output for the full execution window, and the shared control-service now retains every line in memory until the session TTL in addition to the worker's output buffer; unlike before, the control process receives this data incrementally even if the disposable worker later exhausts its own 1 GiB limit, so one or several noisy runs can exhaust the control service or node.
Useful? React with 👍 / 👎.
| for _, ch := range subs { | ||
| ch <- ev | ||
| close(ch) | ||
| } | ||
| s.markFinished() |
There was a problem hiding this comment.
Make terminal delivery nonblocking or mark the session finished before notifying subscribers. If an SSE client stops consuming while its 256-entry channel is full, this send blocks after s.done has been set but before finished is closed; the timeout path then sees the existing result and returns without closing finished, so worker cleanup and replacement never run and repeated slow clients can exhaust the worker pool.
Useful? React with 👍 / 👎.
| for _, line := range s.logs { | ||
| ch <- runEvent{Type: workertypes.WorkerEventLog, Line: line} | ||
| } |
There was a problem hiding this comment.
Do not synchronously enqueue an unbounded replay into this fixed 256-entry channel. When a run emits more than 256 lines before the browser attaches—or when a completed run with that history is opened—the 257th send blocks while s.mu is held, preventing Complete or Fail from acquiring the mutex; the SSE request never starts draining the channel, and the associated worker is never cleaned up or replaced.
Useful? React with 👍 / 👎.
| scanner := bufio.NewScanner(pr) | ||
| scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) | ||
| for scanner.Scan() { | ||
| w.emitLog(scanner.Text()) | ||
| } |
There was a problem hiding this comment.
Handle scanner token overflow without abandoning the pipe reader. For a snippet that writes a line longer than 1 MiB (for example, a large serialized value without a newline), Scanner.Scan stops with ErrTooLong; because pr is left open and its error is ignored, the command's stdout-copy goroutine remains blocked writing to pw, causing c.Run() to wait indefinitely and turning an otherwise finite execution into the 60-second timeout.
Useful? React with 👍 / 👎.
2 participants
Summary
Runs can now stream worker stdout/stderr while the process is still executing.
logevents (line-buffered,stdbuf/PYTHONUNBUFFERED/python -u) then adonepayload with files as before.POST /service/control/runwithAccept: text/event-streamreturns202 { id }immediately (Firefox first-byte / no held POST).GET /service/control/run/:id/log-watchis SSE (: connected,event: log,event: done). Late subscribers get a replay.Accept: text/event-streamstill wait for the existing JSON body (success,output,files,version,duration), so current API e2e keeps working.No Redis and no worker self-registration; one-shot pods and RabbitMQ stay as they are.
Test plan
gofmt;go test ./...;go build ./...frontend:npm run build; Playwright CT chromiumlive-logs): JSONPOST /runstill returns"output":"2"; SSElog-watchemitsearlythendone(~3s); POST TTFB ~160ms for 202e2e --project=apiagainst port-forwarded frontend: 9 passed including live-logsRefs #320 #321