Skip to content

Honor full JSON-RPC wait timeout#57

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zackees merged 1 commit into from
Apr 17, 2026
Merged

Honor full JSON-RPC wait timeout #57

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 69 additions & 17 deletions crates/fbuild-python/src/lib.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -405,23 +405,14 @@ impl SerialMonitor {
let data = format!("{}\n", json_str);
self.write(&data);

let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout);

while std::time::Instant::now() < deadline {
let remaining = (deadline - std::time::Instant::now()).as_secs_f64();
if let Some(json_part) = wait_for_remote_json_rpc_response(timeout, |remaining| {
// read_lines takes &mut self but we only have &self here —
// use the raw WS read directly
let lines = self.read_lines_inner(remaining.min(1.0));
for line in &lines {
if let Some(json_part) = line.strip_prefix("REMOTE:") {
let json_module = py.import_bound("json")?;
let result = json_module.call_method1("loads", (json_part.trim(),))?;
return Ok(result.to_object(py));
}
}
if lines.is_empty() {
break;
}
// use the raw WS read directly.
self.read_lines_inner(remaining.min(1.0))
}) {
let json_module = py.import_bound("json")?;
let result = json_module.call_method1("loads", (json_part.trim(),))?;
return Ok(result.to_object(py));
}
Comment on lines +408 to 416

@coderabbitaicoderabbitaiBotApr 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Risk of tight-spin until timeout when the WebSocket is closed/errored.

The refactor correctly stops treating a single empty batch as terminal, but it also drops the only signal the outer loop had for a dead stream. read_lines_inner returns an empty Vec<String> promptly (without consuming wall time) whenever the underlying ws_read.next() yields Ok(None) or Ok(Some(Message::Close(_))) (see lines 525), i.e. after the daemon disconnects. In that scenario wait_for_remote_json_rpc_response will now re-invoke poll immediately, every iteration returning instantly with vec![], burning a core until the caller's timeout elapses — and only then surfacing PyTimeoutError instead of a prompt connection error.

The previous code at least bailed out of the loop on the first empty batch, so a severed session failed fast. The new logic needs an equivalent "no more data will ever come" exit path.

Suggested shape: let the poller distinguish "waited and got nothing" from "stream is done", e.g. return Option<Vec<String>> (None = terminal) or a small enum, and have wait_for_remote_json_rpc_response break on the terminal variant:

♻️ Sketch of a terminal-aware poll contract
-fn wait_for_remote_json_rpc_response<F>(timeout: f64, mut poll: F) -> Option<String>-where- F: FnMut(f64) -> Vec<String>,-{+fn wait_for_remote_json_rpc_response<F>(timeout: f64, mut poll: F) -> Option<String>+where+ F: FnMut(f64) -> Option<Vec<String>>, // None => stream terminated+{
let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout);
while std::time::Instant::now() < deadline {
let remaining = (deadline - std::time::Instant::now()).as_secs_f64();
- let lines = poll(remaining);+ let Some(lines) = poll(remaining) else {+ return None; // peer gone, don't spin+ };
if let Some(json_part) = extract_remote_json_rpc_response(&lines) {
return Some(json_part);
}
}
None
}

…and have read_lines_inner (or a thin wrapper used only by write_json_rpc) report the Ok(None) / Close case as None. As a cheaper alternative, detect "poll returned faster than remaining" and insert a bounded sleep / break, but an explicit terminal signal is more robust.

Also applies to: 542-557

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/fbuild-python/src/lib.rs` around lines 408 - 416, The loop currently
spin-waits when the websocket is closed because read_lines_inner returns an
empty Vec and wait_for_remote_json_rpc_response keeps polling; fix by making the
poll contract terminal-aware: change read_lines_inner (or add a thin wrapper
used by write_json_rpc) to return Option<Vec<String>> (None meaning stream
terminated on Ok(None)/Close) or a small enum, and update
wait_for_remote_json_rpc_response to treat None as a terminal condition and
break/propagate a connection error immediately instead of retrying until
timeout; update call sites (the block that calls
wait_for_remote_json_rpc_response and any other callers around lines near
read_lines_inner / write_json_rpc) to handle the terminal signal accordingly.


Err(pyo3::exceptions::PyTimeoutError::new_err(format!(
Expand DownExpand Up@@ -541,6 +532,30 @@ impl SerialMonitor {
}
}

fn extract_remote_json_rpc_response(lines: &[String]) -> Option<String> {
lines.iter().find_map(|line| {
line.strip_prefix("REMOTE:")
.map(|json_part| json_part.to_string())
})
}

fn wait_for_remote_json_rpc_response<F>(timeout: f64, mut poll: F) -> Option<String>
where
F: FnMut(f64) -> Vec<String>,
{
let deadline = std::time::Instant::now() + std::time::Duration::from_secs_f64(timeout);

while std::time::Instant::now() < deadline {
let remaining = (deadline - std::time::Instant::now()).as_secs_f64();
let lines = poll(remaining);
if let Some(json_part) = extract_remote_json_rpc_response(&lines) {
return Some(json_part);
}
}

None
}

/// Python-visible Daemon class (high-level API).
#[pyclass]
struct Daemon;
Expand DownExpand Up@@ -777,7 +792,9 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> {

#[cfg(test)]
mod tests {
use super::PYTHON_MODULE_VERSION;
use super::{
extract_remote_json_rpc_response, wait_for_remote_json_rpc_response, PYTHON_MODULE_VERSION,
};

/// Ensures the Python-visible `__version__` is sourced from Cargo and not
/// a stale hardcoded literal. The previous value `"2.0.0"` diverged from
Expand DownExpand Up@@ -822,4 +839,39 @@ mod tests {
);
}
}

#[test]
fn extract_remote_json_rpc_response_skips_empty_batches() {
let empty: Vec<String> = vec![];
assert_eq!(extract_remote_json_rpc_response(&empty), None);
}

#[test]
fn extract_remote_json_rpc_response_finds_remote_payload() {
let lines = vec![
"noise".to_string(),
r#"REMOTE: {"ok": true}"#.to_string(),
"more noise".to_string(),
];
assert_eq!(
extract_remote_json_rpc_response(&lines).as_deref(),
Some(r#" {"ok": true}"#)
);
}

#[test]
fn wait_for_remote_json_rpc_response_keeps_polling_after_empty_batch() {
let mut polls = 0usize;
let result = wait_for_remote_json_rpc_response(0.05, |_| {
polls += 1;
match polls {
1 => vec![],
2 => vec!["REMOTE: {\"ok\": true}".to_string()],
_ => vec![],
}
});

assert_eq!(polls, 2, "an empty batch must not end the overall wait");
assert_eq!(result.as_deref(), Some(r#" {"ok": true}"#));
}
}
Loading