feat(streams): operator control server - #353
Conversation
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
b551a1b to
f42d312
Compare
f42d312 to
9b45422
Compare
fpacifici
left a comment
fpacifici
left a comment
There was a problem hiding this comment.
Please see the comments in line
| /// The ProcessorHandle allows the main thread to stop the StreamingProcessor | ||
| /// from a different thread. | ||
| handle: Option<ProcessorHandle>, | ||
| handle: Mutex<Option<ProcessorHandle>>, |
There was a problem hiding this comment.
Why do you need a Mutex here ?
| // The GIL is released around the run loop so that a Python control | ||
| // thread can call shutdown while this consumer is running: | ||
|
|
||
| if let Err(e) = processor.run() { | ||
| tracing::error!("StreamProcessor error: {:?}", e); | ||
| sentry::capture_error(&e); | ||
| let run_error = py.detach(|| match processor.run() { |
There was a problem hiding this comment.
So here is the issue. processor.run() takes the GIL in several places. Did you see this code running successfully with a real pipeline that runs python code ?
There was a problem hiding this comment.
Yes, we need detach because otherwise the control server in Python would be locked and not respond.
| *self.handle.lock().unwrap() = Some(processor.get_handle()); | ||
|
|
||
| if self.shutdown_requested.load(Ordering::Acquire) { | ||
| processor.get_handle().signal_shutdown(); | ||
| } |
There was a problem hiding this comment.
I don;'t think you need this. You can get multiple handles from the processor. You do not need to share it between run and shutdown.
There was a problem hiding this comment.
The run call has a reference to the processor on one thread, shutdown does not have a reference (processor is a local to run) and is on another thread. Run needs to use a mutex to store the handle so that shutdown can access it.
|
|
||
| serve_thread = threading.Thread( | ||
| target=_serve_until_shutdown, | ||
| name="control-server", | ||
| daemon=True, |
There was a problem hiding this comment.
Bug: The pipeline-signal thread is non-daemon, which can prevent the process from exiting cleanly if the controller.request_stop() call hangs during shutdown.
Severity: HIGH
Suggested Fix
Change the shutdown_thread to be a daemon thread by setting daemon=True during its initialization. This will allow the Python process to exit even if this thread hangs, as daemon threads are abruptly stopped at shutdown.
Prompt for AI Agent
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.
Location: sentry_streams/sentry_streams/runner.py#L90-L94
Potential issue: The `shutdown_thread` in `runner.py` is created as a non-daemon thread
(`daemon=False`). This thread's purpose is to call `controller.request_stop()` upon
receiving a shutdown signal. The main thread waits for this thread to complete using
`join()` with a timeout. However, if the `controller.request_stop()` call hangs (for
instance, due to a deadlock), the `join()` will time out, but the non-daemon thread will
persist. A running non-daemon thread prevents the Python process from exiting, which
will require an external kill signal to terminate the service.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit a7b4cda. Configure here.
| ) | ||
| serve_thread.start() | ||
| else: | ||
| controller.request_start() |
There was a problem hiding this comment.
Startup signal race crashes process
Medium Severity
In deployment mode, the shutdown thread is started before request_start. If SIGINT/SIGTERM arrives in that window, request_stop moves the runtime to STOPPED, then request_start raises RuntimeStateError (cannot restart runtime that is stopped). The process exits with a traceback instead of a clean shutdown, and this path sits outside the try/finally cleanup.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit a7b4cda. Configure here.
bafc46f to
3e4187d
Compare
a7b4cda to
4937081
Compare
2 participants
Basic control server for the operator to use for blue-green deployments (readyz, status start, stop). Adds --control-port option to runner to determine if the process should use deployment mode (autostart) or operator mode (wait for /start).