From 92840172da68adcc2019f86cf980fc56ce91979d Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 18:50:27 -0700 Subject: [PATCH 1/6] fix(transport): bound the Quit-path reader join so a wedged daemon cannot strand the terminal --- src/transport.rs | 85 +++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 74 insertions(+), 11 deletions(-) diff --git a/src/transport.rs b/src/transport.rs index 205e745..2a94479 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -5,9 +5,10 @@ use std::{ os::unix::net::UnixStream, sync::{ atomic::AtomicBool, - mpsc::{Receiver, Sender, TryRecvError, channel}, + mpsc::{Receiver, RecvTimeoutError, Sender, TryRecvError, channel}, }, thread::{self, JoinHandle}, + time::{Duration, Instant}, }; use crate::{ @@ -41,7 +42,9 @@ pub trait Transport { fn connected(&self) -> bool; /// Tear down per `intent`, blocking until it's done, so the client restores /// the terminal only after the core has acted (tasks killed on `Quit`, the - /// connection closed on `Disconnect`). + /// connection closed on `Disconnect`). The wait is not unconditional: an + /// implementation may bound it and hang up on a wedged core — the terminal + /// restore is owed to the user either way. fn shutdown(&mut self, intent: ExitIntent); } @@ -195,6 +198,37 @@ impl SocketTransport { dead: false, } } + + /// `Quit` teardown: send `Shutdown`, then wait at most `bound` for the + /// daemon to close the socket — the reader ending is the proof the tasks + /// died. The reader owns `evt_tx`, so `evt_rx` disconnecting is exactly the + /// reader ending; events arriving meanwhile are discarded (the client is + /// past polling). On expiry, force our socket shut: the halves are clones + /// of one descriptor, so this errors the reader's blocking `read_frame` out + /// immediately (dropping the write half alone would not interrupt it), + /// which makes the final join bounded. Production passes `SEND_TIMEOUT`; + /// tests pass a small budget. + fn quit_within(&mut self, bound: Duration) { + self.send(Command::Shutdown); + let deadline = Instant::now() + bound; + loop { + let now = Instant::now(); + if now >= deadline { + // The daemon never closed the socket: it lost its right to be + // waited on. Hang up so the reader errors out. + let _ = self.write.shutdown(Shutdown::Both); + break; + } + match self.evt_rx.recv_timeout(deadline - now) { + Ok(_) => {} // discard + Err(RecvTimeoutError::Timeout) => {} // deadline re-checked above + Err(RecvTimeoutError::Disconnected) => break, // reader ended + } + } + if let Some(h) = self.reader.take() { + let _ = h.join(); + } + } } impl Transport for SocketTransport { @@ -223,20 +257,22 @@ impl Transport for SocketTransport { fn shutdown(&mut self, intent: ExitIntent) { match intent { - // Group-kill every task and stop the daemon; the socket then closes - // (daemon gone = tasks killed). - ExitIntent::Quit => self.send(Command::Shutdown), + // Group-kill every task and stop the daemon, then wait for our + // reader to see the daemon close the socket (daemon gone = tasks + // killed) — but only up to `SEND_TIMEOUT`. A wedged daemon that + // never closes does not get a veto on restoring the terminal. + ExitIntent::Quit => self.quit_within(SEND_TIMEOUT), // Close the connection without a Shutdown: the daemon sees EOF and - // keeps the tasks running for the next client to reattach. + // keeps the tasks running for the next client to reattach. The + // close we just forced errors the reader out of its blocking read, + // so this join is bounded. ExitIntent::Disconnect => { let _ = self.write.shutdown(Shutdown::Both); + if let Some(h) = self.reader.take() { + let _ = h.join(); + } } } - // Either way, wait for our reader to see the socket close before the - // client restores the terminal. On Quit that means the tasks are dead. - if let Some(h) = self.reader.take() { - let _ = h.join(); - } } } @@ -295,4 +331,31 @@ mod tests { // exited: joining it cannot hang. t.reader.take().unwrap().join().unwrap(); } + + /// `Quit` teardown is bounded: a daemon that accepts the `Shutdown` frame + /// but never closes the socket cannot block the terminal restore. + #[test] + fn quit_shutdown_is_bounded_when_the_daemon_never_closes() { + let (ours, theirs) = UnixStream::pair().unwrap(); + let write = ours.try_clone().unwrap(); + let (wait_tx, _wait_rx) = channel(); + let mut t = SocketTransport::from_halves(write, ours, wait_tx); + + // Hold `theirs` open, never writing and never closing: the `Shutdown` + // frame lands in the socket buffer, but no close ever arrives, so the + // reader stays blocked in `read_frame` until the transport hangs up. + let (done_tx, done_rx) = channel(); + let worker = thread::spawn(move || { + t.quit_within(Duration::from_millis(150)); + let _ = done_tx.send(()); + }); + // Test-side deadline well above the bound: on unfixed code the worker + // blocks in the reader join forever, and this fails the test instead of + // hanging the suite (unwinding drops `theirs`, which unblocks it). + done_rx + .recv_timeout(Duration::from_secs(2)) + .expect("Quit shutdown must return within its bound"); + worker.join().unwrap(); + drop(theirs); + } } From f3ccb2cceb2947408e939506b9e534a3f0136c37 Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 18:52:09 -0700 Subject: [PATCH 2/6] fix(daemon): --kill reports honestly while a daemon is starting; delete the unreachable socket fallback --- src/daemon.rs | 209 ++++++++++++-------------------------------------- 1 file changed, 51 insertions(+), 158 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index a7bb80f..2c2ff13 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -17,7 +17,7 @@ use std::{ fs, - io::{self, ErrorKind, Read, Write}, + io::{self, ErrorKind, Read, Seek, Write}, net::Shutdown, os::unix::{ fs::{DirBuilderExt, MetadataExt, PermissionsExt}, @@ -33,7 +33,7 @@ use std::{ mpsc::channel, }, thread, - time::{Duration, Instant}, + time::Duration, }; use nix::{ @@ -48,7 +48,7 @@ use crate::{ path::FLEETCOM_RUNTIME_DIR, protocol::{ Command, Event, LaunchContext, PROTOCOL_VERSION, decode_command, decode_event, - decode_hello, encode_command, encode_event, encode_hello, hello_version, + decode_hello, encode_event, encode_hello, hello_version, }, supervisor::{self, Supervisor}, }; @@ -367,12 +367,18 @@ fn no_daemon() -> io::Result<()> { /// `--kill` must work while someone else is attached. The pid comes from the /// lock file (trustworthy while the flock is held: the holder wrote it), and /// daemon exit releases the flock, so acquiring it is the completion signal. -/// A no-op (with a message) if no daemon is running. +/// A no-op (with a message) if no daemon is running. A held flock with no +/// readable pid is a daemon mid-startup: that is an error, never the no-op, +/// because someone provably holds the lock. pub fn run_kill() -> io::Result<()> { - let dir = runtime_dir(); + run_kill_in(&runtime_dir()) +} + +/// `run_kill` against an explicit runtime directory. +fn run_kill_in(dir: &Path) -> io::Result<()> { // The lock PID is a signal target, and the socket receives the client's // environment, so validate the directory before reading either file. - ensure_runtime_dir(&dir)?; + ensure_runtime_dir(dir)?; let lock_path = dir.join("daemon.lock"); let Ok(file) = fs::OpenOptions::new() .read(true) @@ -387,13 +393,27 @@ pub fn run_kill() -> io::Result<()> { Err((file, _)) => file, }; - let mut pid_str = String::new(); - file.read_to_string(&mut pid_str)?; - let Some(pid) = crate::task::positive_pid(pid_str.trim()) else { - // Without a usable pid, fall back to a Shutdown frame over the socket. - // Bound the fallback because an attached client can keep the daemon - // from accepting this connection. - return kill_via_socket_at(&socket_path(), KILL_SOCKET_TIMEOUT); + // The flock is held, but the pid can be momentarily unreadable: + // `run_daemon` acquires the lock, then truncates and writes the pid, so + // the only held-lock window without one is those two syscalls. Poll the + // file briefly rather than guess. + let mut pid = None; + for _ in 0..20 { + let mut pid_str = String::new(); + file.seek(io::SeekFrom::Start(0))?; + file.read_to_string(&mut pid_str)?; + pid = crate::task::positive_pid(pid_str.trim()); + if pid.is_some() { + break; + } + thread::sleep(Duration::from_millis(10)); + } + let Some(pid) = pid else { + return Err(io::Error::new( + ErrorKind::TimedOut, + "the daemon holds the lock but has not written its pid (it may \ + still be starting); retry", + )); }; // ESRCH means the daemon exited between the lock probe and here; the flock @@ -419,93 +439,6 @@ pub fn run_kill() -> io::Result<()> { )) } -/// Timeout applied to blocking socket-fallback kill operations. -const KILL_SOCKET_TIMEOUT: Duration = Duration::from_secs(10); - -/// Timeout reported when the socket-fallback kill exchange does not finish. -fn kill_handshake_timeout() -> io::Error { - io::Error::new( - ErrorKind::TimedOut, - "the daemon is running but did not complete the kill handshake in \ - time (another client may be attached); retry after it detaches, or \ - send SIGTERM to the daemon process directly", - ) -} - -/// Set the read timeout to the remaining deadline budget. -fn arm_read_deadline(s: &UnixStream, deadline: Instant) -> io::Result<()> { - let left = deadline.saturating_duration_since(Instant::now()); - if left.is_zero() { - return Err(kill_handshake_timeout()); - } - s.set_read_timeout(Some(left)) -} - -/// Convert either platform representation of a socket timeout into the -/// kill-handshake timeout. -fn deadline_mapped(e: io::Error) -> io::Error { - if is_timeout(&e) { - kill_handshake_timeout() - } else { - e - } -} - -/// When the lock lacks a valid PID, send `Shutdown` over the socket and bound -/// handshake and completion I/O by `budget`. -fn kill_via_socket_at(path: &Path, budget: Duration) -> io::Result<()> { - match UnixStream::connect(path) { - Ok(mut s) => { - let (kind, payload) = encode_hello(&LaunchContext::here()); - kill_exchange(&mut s, budget, kind, &payload) - } - Err(_) => no_daemon(), - } -} - -/// Drive the bounded Shutdown exchange with a pre-encoded hello frame. -fn kill_exchange( - s: &mut UnixStream, - budget: Duration, - hello_kind: u8, - hello_payload: &[u8], -) -> io::Result<()> { - let deadline = Instant::now() + budget; - s.set_write_timeout(Some(budget))?; - - write_frame(s, hello_kind, hello_payload).map_err(deadline_mapped)?; - arm_read_deadline(s, deadline)?; - let (kind, payload) = read_frame(s).map_err(deadline_mapped)?; - check_hello_ack(kind, &payload)?; - - let (kind, payload) = encode_command(&Command::Shutdown); - write_frame(s, kind, &payload).map_err(deadline_mapped)?; - // Socket closure signals completion. Re-arm each read with the remaining - // budget so the loop cannot outlive the deadline. - let mut buf = [0u8; 256]; - loop { - arm_read_deadline(s, deadline)?; - match s.read(&mut buf) { - Ok(0) => return Ok(()), - Ok(_) => {} - Err(e) if is_timeout(&e) => return Err(kill_handshake_timeout()), - // Retry interrupted reads; the deadline still bounds the loop. - Err(e) if e.kind() == ErrorKind::Interrupted => {} - // The daemon closing mid-drain is completion, same as `Ok(0)`. - Err(e) - if matches!( - e.kind(), - ErrorKind::ConnectionReset | ErrorKind::BrokenPipe | ErrorKind::UnexpectedEof - ) => - { - return Ok(()); - } - // Propagate other errors because they do not confirm daemon exit. - Err(e) => return Err(e), - } - } -} - /// The daemon entry point (`fleetcom --daemon`). Binds the socket and serves clients /// until an explicit shutdown. The supervisor is created once and persists across /// reconnects: tasks outlive any single client. @@ -923,66 +856,26 @@ mod tests { assert!(notice.contains("--kill"), "{notice}"); } - /// A missing hello response times out the kill exchange. + /// A held flock with no pid is a daemon between lock acquisition and pid + /// write: `--kill` must report an error, not the Ok "no daemon running" + /// no-op, because the holder proves a daemon is starting. #[test] - fn kill_via_socket_bounds_the_handshake_wait() { - let base = temp("kill_socket_mute"); - fs::create_dir_all(&*base).unwrap(); - let sock = base.join("mute.sock"); - // Leave the connection queued in the listener backlog. - let _listener = UnixListener::bind(&sock).unwrap(); - let start = Instant::now(); - let err = kill_via_socket_at(&sock, Duration::from_millis(200)).unwrap_err(); + fn kill_with_a_held_lock_and_no_pid_is_an_error() { + let base = temp("kill_lock_no_pid"); + let dir = base.join("runtime"); + ensure_runtime_dir(&dir).unwrap(); + let holder = fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(false) + .open(dir.join("daemon.lock")) + .unwrap(); + // flock is per open-file-description, so this handle contends with + // the one `run_kill_in` opens, even within one process. + let _held = Flock::lock(holder, FlockArg::LockExclusiveNonblock).unwrap(); + let err = run_kill_in(&dir).unwrap_err(); assert_eq!(err.kind(), ErrorKind::TimedOut); - assert!(err.to_string().contains("kill handshake"), "{err}"); - assert!( - start.elapsed() < Duration::from_secs(5), - "the deadline must fire, not the test's timeout" - ); - } - - /// A blocked hello write reports the kill-handshake timeout. - #[test] - fn kill_exchange_maps_a_write_timeout() { - let base = temp("kill_socket_bigenv"); - fs::create_dir_all(&*base).unwrap(); - let sock = base.join("mute.sock"); - let _listener = UnixListener::bind(&sock).unwrap(); - let mut s = UnixStream::connect(&sock).unwrap(); - // The listener never accepts, so this payload fills the send buffer. - let oversized = vec![0u8; 8 * 1024 * 1024]; - let err = kill_exchange(&mut s, Duration::from_millis(200), 0, &oversized).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::TimedOut); - assert!(err.to_string().contains("kill handshake"), "{err}"); - } - - /// A daemon that keeps the socket open after Shutdown times out the drain. - #[test] - fn kill_via_socket_bounds_the_drain_wait() { - let base = temp("kill_socket_drain"); - fs::create_dir_all(&*base).unwrap(); - let sock = base.join("stuck.sock"); - let listener = UnixListener::bind(&sock).unwrap(); - let server = thread::spawn(move || { - let (mut s, _) = listener.accept().unwrap(); - let _ = read_frame(&mut s); // hello - let (kind, payload) = encode_event(&Event::HelloOk); - let _ = write_frame(&mut s, kind, &payload); - let _ = read_frame(&mut s); // Shutdown, swallowed - // Hold the socket open until the client drops its end. - let _ = s.read(&mut [0u8; 16]); - }); - let err = kill_via_socket_at(&sock, Duration::from_millis(300)).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::TimedOut); - server.join().unwrap(); - } - - /// A missing socket makes the fallback a no-op. - #[test] - fn kill_via_socket_without_a_socket_is_a_noop() { - let base = temp("kill_socket_absent"); - fs::create_dir_all(&*base).unwrap(); - assert!(kill_via_socket_at(&base.join("absent.sock"), Duration::from_millis(100)).is_ok()); + assert!(err.to_string().contains("pid"), "{err}"); } /// Remove group and other read/execute permissions from a valid directory. From 6d39b62a1b9a0ab3e41a13a16bc7394b3c14935f Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 18:52:52 -0700 Subject: [PATCH 3/6] test(app): inject TaskView fixtures in the four order-asserting dashboard tests --- src/app_tests.rs | 57 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 40 insertions(+), 17 deletions(-) diff --git a/src/app_tests.rs b/src/app_tests.rs index 9fb66cc..f41a31b 100644 --- a/src/app_tests.rs +++ b/src/app_tests.rs @@ -104,21 +104,42 @@ fn shift(code: KeyCode) -> KeyEvent { KeyEvent::new(code, KeyModifiers::SHIFT) } +/// Minimal live task for injecting into `app.views` directly: carries the +/// fields `sections`/`display_order` sort on (id, cwd, tagged, group) with +/// everything else inert. Ordering tests use these instead of real children +/// because a spawned child that exits mid-test re-buckets its row into +/// Completed and breaks a fixed expected order under load. +fn view(id: u64, cwd: PathBuf, tagged: bool, group: Option<&str>) -> TaskView { + TaskView { + id, + command: "true".to_string(), + cwd, + tagged, + group: group.map(str::to_string), + name: None, + lifecycle: Lifecycle::Active, + parked: false, + preview: Preview::floor(String::new()), + started_ago: Duration::ZERO, + quiet_ago: Some(Duration::ZERO), + finished_ago: None, + } +} + /// Selection is bound to a task id, so a reorder (here: tagging a task into /// the "In use" bucket) must not move the highlight to a different task. #[test] fn selection_follows_task_across_reorder() { let mut app = App::new_local(30, 100); let dir = app.invocation_dir.clone(); - app.spawn_in("sleep 5", dir.clone()); // id 1 - app.spawn_in("sleep 5", dir); // id 2 - app.pump(); + app.views = vec![view(1, dir.clone(), false, None), view(2, dir, false, None)]; app.resolve_selection(); assert_eq!(app.selected_id, Some(1)); - // Tag id 2 -> it sorts into the "In use" bucket, ahead of id 1. - app.transport.send(Command::Tag { id: 2, on: true }); - app.pump(); + // Tag id 2 -> it sorts into the "In use" bucket, ahead of id 1. Mutated + // in place: the Tag round-trip is other tests' subject, and a pump would + // overwrite the injected views with the core's empty snapshot. + app.views[1].tagged = true; let order = app.display_order(); assert_eq!(app.views[order[0]].id, 2, "tagged task should sort first"); @@ -134,9 +155,10 @@ fn selection_follows_task_across_reorder() { fn dir_mode_groups_by_cwd() { let mut app = App::new_local(30, 100); let inv = app.invocation_dir.clone(); - app.spawn_in("sleep 5", inv); // id 1, invocation dir - app.spawn_in("sleep 5", PathBuf::from("/tmp")); // id 2, /tmp - app.pump(); + app.views = vec![ + view(1, inv, false, None), // invocation dir + view(2, PathBuf::from("/tmp"), false, None), // /tmp + ]; app.group_mode = GroupMode::State; let s = app.sections(); @@ -2169,11 +2191,11 @@ fn find_palette_opens_on_slash_only_with_tasks() { fn find_candidates_follow_display_order() { let mut app = App::new_local(30, 100); let inv = app.invocation_dir.clone(); - app.spawn_in("sleep 5", inv.clone()); // id 1 - app.spawn_in("sleep 5", inv.clone()); // id 2 - app.spawn_in("sleep 5", inv); // id 3 - app.transport.send(Command::Tag { id: 3, on: true }); - app.pump(); + app.views = vec![ + view(1, inv.clone(), false, None), + view(2, inv.clone(), false, None), + view(3, inv, true, None), + ]; let order: Vec = app .display_order() .into_iter() @@ -2191,9 +2213,10 @@ fn find_candidates_follow_display_order() { fn find_empty_input_lists_every_task() { let mut app = App::new_local(30, 100); let inv = app.invocation_dir.clone(); - app.spawn_in("sleep 5", inv.clone()); // id 1 - app.spawn_grouped("sleep 5", inv, "alpha"); // id 2 - app.pump(); + app.views = vec![ + view(1, inv.clone(), false, None), + view(2, inv, false, Some("alpha")), + ]; app.on_key_dashboard(key(KeyCode::Char('/'))); assert_eq!(find_ids(&app), vec![1, 2]); From 6259f51cdb7660a13c86135369b9407357f1097b Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 19:02:12 -0700 Subject: [PATCH 4/6] fix(transport): move client frame writes to a bounded writer thread so a stalled daemon cannot freeze the UI --- src/app.rs | 6 +- src/transport.rs | 204 ++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 172 insertions(+), 38 deletions(-) diff --git a/src/app.rs b/src/app.rs index 8ae21a5..6b18488 100644 --- a/src/app.rs +++ b/src/app.rs @@ -344,8 +344,9 @@ impl App { // Split the stream here (the fallible part) so the transport factory in // `assemble` (which owns the wake sender) stays infallible. let read = stream.try_clone()?; + let ctrl = stream.try_clone()?; let mut app = Self::assemble(rows, cols, move |_, _, wait_tx| { - Box::new(SocketTransport::from_halves(stream, read, wait_tx)) + Box::new(SocketTransport::from_halves(stream, read, ctrl, wait_tx)) }); app.daemon_backed = true; // Report when a running daemon could not apply this invocation's @@ -361,7 +362,8 @@ impl App { // Reconnection must not block the active UI indefinitely. let stream = crate::daemon::connect_ready_bounded()?; let read = stream.try_clone()?; - Ok(SocketTransport::from_halves(stream, read, wait_tx)) + let ctrl = stream.try_clone()?; + Ok(SocketTransport::from_halves(stream, read, ctrl, wait_tx)) }; match build() { Ok(t) => { diff --git a/src/transport.rs b/src/transport.rs index 2a94479..de4c68e 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -5,7 +5,9 @@ use std::{ os::unix::net::UnixStream, sync::{ atomic::AtomicBool, - mpsc::{Receiver, RecvTimeoutError, Sender, TryRecvError, channel}, + mpsc::{ + Receiver, RecvTimeoutError, Sender, SyncSender, TryRecvError, channel, sync_channel, + }, }, thread::{self, JoinHandle}, time::{Duration, Instant}, @@ -152,28 +154,69 @@ impl Drop for ThreadTransport { } } +/// Frames a `send` may queue ahead of the writer thread. 64 bounds the run +/// loop's exposure to 64 frames of delivery latency — not 64 × `SEND_TIMEOUT` +/// of blocking, because `send` never waits on the queue: a full queue means +/// the peer let 64 frames pile up against a 5 s-per-write budget, and the +/// transport declares it dead instead (see `SocketTransport::send`). +const SEND_QUEUE: usize = 64; + /// The core as a separate process (`fleetcom --daemon`), reached over a Unix -/// socket. Commands are written as frames on the connection; a reader thread -/// turns inbound event frames back into `Event`s on a channel, so `poll` drains -/// the channel exactly like `ThreadTransport`. +/// socket. Commands are encoded on the caller's thread and queued to a writer +/// thread that frames them onto the connection, so a daemon that stops reading +/// can never block the client's run loop; a reader thread turns inbound event +/// frames back into `Event`s on a channel, so `poll` drains the channel +/// exactly like `ThreadTransport`. pub struct SocketTransport { - write: UnixStream, + /// Control handle for forced shutdowns only — frame writes happen on the + /// writer thread. All handles are `try_clone`s of one socket (dup'd FDs + /// share the open socket description), so a shutdown here errors the + /// reader and writer out of their blocking calls. + ctrl: UnixStream, + /// Encoded frames to the writer thread. `None` once teardown takes it: + /// dropping the sender is what ends an idle writer's `recv` loop. + frame_tx: Option)>>, evt_rx: Receiver, reader: Option>, + writer: Option>, /// Set when the reader thread ends on socket EOF (the daemon is gone), or - /// when a `send` fails (the connection is unrecoverable; see `send`). + /// when `send` cannot queue a frame (the connection is unrecoverable; see + /// `send`). A write failure on the writer thread lands here indirectly: + /// the writer shuts the socket down, the reader exits on it, and `poll` + /// observes the disconnect. dead: bool, } impl SocketTransport { - /// Build over pre-split stream halves (`write`, `read`). The `try_clone` that - /// can fail is the caller's job: done outside the transport so the App's - /// transport factory stays infallible. `wait_tx` wakes the client's run loop - /// on each inbound event. - pub fn from_halves(write: UnixStream, read: UnixStream, wait_tx: Sender<()>) -> Self { + /// Build over pre-split clones of one stream: `write` feeds the writer + /// thread, `read` the reader thread, `ctrl` stays behind for forced + /// shutdowns. The `try_clone`s that can fail are the caller's job: done + /// outside the transport so the App's transport factory stays infallible. + /// `wait_tx` wakes the client's run loop on each inbound event. + pub fn from_halves( + write: UnixStream, + read: UnixStream, + ctrl: UnixStream, + wait_tx: Sender<()>, + ) -> Self { // Keep construction infallible; if this best-effort setup fails, the // stream retains its existing write-timeout setting. let _ = write.set_write_timeout(Some(SEND_TIMEOUT)); + let (frame_tx, frame_rx) = sync_channel::<(u8, Vec)>(SEND_QUEUE); + let writer = thread::spawn(move || { + let mut write = write; + // Ends when the queue sender drops (teardown, or the transport + // itself dropped) or a frame write fails. `SEND_TIMEOUT` on the + // stream bounds each write; on failure, shut the socket down so + // the reader's blocking `read_frame` errors out too and the `dead` + // flag reaches the client through the existing `poll` path. + while let Ok((kind, payload)) = frame_rx.recv() { + if write_frame(&mut write, kind, &payload).is_err() { + let _ = write.shutdown(Shutdown::Both); + break; + } + } + }); let (evt_tx, evt_rx) = channel(); let reader = thread::spawn(move || { let mut read = read; @@ -192,21 +235,25 @@ impl SocketTransport { let _ = wait_tx.send(()); }); Self { - write, + ctrl, + frame_tx: Some(frame_tx), evt_rx, reader: Some(reader), + writer: Some(writer), dead: false, } } - /// `Quit` teardown: send `Shutdown`, then wait at most `bound` for the + /// `Quit` teardown: queue `Shutdown`, then wait at most `bound` for the /// daemon to close the socket — the reader ending is the proof the tasks - /// died. The reader owns `evt_tx`, so `evt_rx` disconnecting is exactly the - /// reader ending; events arriving meanwhile are discarded (the client is - /// past polling). On expiry, force our socket shut: the halves are clones - /// of one descriptor, so this errors the reader's blocking `read_frame` out + /// died. `Shutdown` rides the writer queue like any frame; whether it + /// flushes or not, the bounded `evt_rx` wait is the backstop. The reader + /// owns `evt_tx`, so `evt_rx` disconnecting is exactly the reader ending; + /// events arriving meanwhile are discarded (the client is past polling). + /// On expiry, force our socket shut: the handles are clones of one + /// descriptor, so this errors the reader's blocking `read_frame` out /// immediately (dropping the write half alone would not interrupt it), - /// which makes the final join bounded. Production passes `SEND_TIMEOUT`; + /// which makes the final joins bounded. Production passes `SEND_TIMEOUT`; /// tests pass a small budget. fn quit_within(&mut self, bound: Duration) { self.send(Command::Shutdown); @@ -216,7 +263,7 @@ impl SocketTransport { if now >= deadline { // The daemon never closed the socket: it lost its right to be // waited on. Hang up so the reader errors out. - let _ = self.write.shutdown(Shutdown::Both); + let _ = self.ctrl.shutdown(Shutdown::Both); break; } match self.evt_rx.recv_timeout(deadline - now) { @@ -228,6 +275,20 @@ impl SocketTransport { if let Some(h) = self.reader.take() { let _ = h.join(); } + self.join_writer(); + } + + /// End the writer thread with a bounded join. Every caller has a socket + /// closure already in force (the daemon closed it, or we forced `ctrl` + /// shut), so a writer blocked mid-write errors out immediately; dropping + /// the queue sender is what ends an idle writer's `recv`. Without a prior + /// closure this join could wait a full `SEND_TIMEOUT` — never call it on a + /// live socket. + fn join_writer(&mut self) { + self.frame_tx.take(); + if let Some(h) = self.writer.take() { + let _ = h.join(); + } } } @@ -238,12 +299,21 @@ impl Transport for SocketTransport { return; } let (kind, payload) = encode_command(&cmd); - if write_frame(&mut self.write, kind, &payload).is_err() { - // A failed frame write may leave a partial frame on the stream. - // Mark the connection dead and close both halves so the reader - // exits and the client can reconnect. + // Queue, never block: this runs on the client's run-loop thread, where + // any wait on the peer freezes painting, input, and exit. + let queued = self + .frame_tx + .as_ref() + .is_some_and(|tx| tx.try_send((kind, payload)).is_ok()); + if !queued { + // Full or disconnected. A full queue means the peer let + // `SEND_QUEUE` frames pile up against a `SEND_TIMEOUT`-per-write + // budget: for our purposes that is a dead peer, the same verdict + // as a failed synchronous write. Mark the connection dead and + // close the socket so the reader and writer exit and the client + // can reconnect. self.dead = true; - let _ = self.write.shutdown(Shutdown::Both); + let _ = self.ctrl.shutdown(Shutdown::Both); } } @@ -264,13 +334,14 @@ impl Transport for SocketTransport { ExitIntent::Quit => self.quit_within(SEND_TIMEOUT), // Close the connection without a Shutdown: the daemon sees EOF and // keeps the tasks running for the next client to reattach. The - // close we just forced errors the reader out of its blocking read, - // so this join is bounded. + // close we just forced errors the reader out of its blocking read + // and any in-flight frame write, so both joins are bounded. ExitIntent::Disconnect => { - let _ = self.write.shutdown(Shutdown::Both); + let _ = self.ctrl.shutdown(Shutdown::Both); if let Some(h) = self.reader.take() { let _ = h.join(); } + self.join_writer(); } } } @@ -312,13 +383,33 @@ impl Transport for LocalTransport { mod tests { use super::*; - /// A failed send marks the transport disconnected and stops its reader. + /// Build a transport over one end of a socketpair, doing the `try_clone` + /// splitting the production call sites do. + fn transport_over(ours: UnixStream) -> SocketTransport { + let write = ours.try_clone().unwrap(); + let ctrl = ours.try_clone().unwrap(); + let (wait_tx, _wait_rx) = channel(); + SocketTransport::from_halves(write, ours, ctrl, wait_tx) + } + + /// Poll until the transport reports dead or `bound` expires. Dead now + /// propagates asynchronously (writer error → socket shutdown → reader + /// exit → `poll` sees the disconnect), so tests must wait, bounded. + fn wait_dead(t: &mut SocketTransport, bound: Duration) { + let deadline = Instant::now() + bound; + while t.connected() && Instant::now() < deadline { + let _ = t.poll(); + thread::yield_now(); + } + } + + /// A send the writer thread cannot deliver marks the transport + /// disconnected: the failed write shuts the socket down, the reader exits + /// on it, and `poll` surfaces the drop. #[test] fn failed_send_marks_the_transport_dead() { let (ours, theirs) = UnixStream::pair().unwrap(); - let write = ours.try_clone().unwrap(); - let (wait_tx, _wait_rx) = channel(); - let mut t = SocketTransport::from_halves(write, ours, wait_tx); + let mut t = transport_over(ours); assert!(t.connected()); drop(theirs); // the daemon is gone @@ -326,10 +417,53 @@ mod tests { id: None, attached: false, }); + wait_dead(&mut t, Duration::from_secs(2)); assert!(!t.connected(), "a failed send must mark the transport dead"); - // The stream was shut down with it, so the reader thread saw EOF and - // exited: joining it cannot hang. + // The stream was shut down along the way, so the reader thread saw + // EOF and exited: joining it cannot hang. t.reader.take().unwrap().join().unwrap(); + // The writer ended too — its write failed against the closed peer, or + // its sender is about to drop; either way this join is bounded. + t.join_writer(); + } + + /// The regression this design exists for: a peer that stops reading must + /// not block `send` — the run loop's thread is the UI. Fill the socket + /// send buffer and the whole frame queue; every `send` must return + /// promptly and the transport must declare itself dead, leaving recovery + /// to the reconnect path. + #[test] + fn send_burst_against_a_stalled_peer_never_blocks_the_caller() { + let (ours, theirs) = UnixStream::pair().unwrap(); + let mut t = transport_over(ours); + + // 64 KiB per frame: one or two fill the socket send buffer (single- + // digit KiB by default on macOS), the rest fill the `SEND_QUEUE` + // slots, and the overflow must be refused, not waited on. + let bytes = vec![b'p'; 64 * 1024]; + let start = Instant::now(); + for _ in 0..(SEND_QUEUE + 8) { + t.send(Command::Paste { + id: 1, + bytes: bytes.clone(), + }); + } + let elapsed = start.elapsed(); + // Unfixed, the first buffer-filling write alone blocks `SEND_TIMEOUT` + // (5 s); the whole burst must stay far under that. + assert!( + elapsed < Duration::from_secs(2), + "send burst blocked the run loop for {elapsed:?}" + ); + assert!( + !t.connected(), + "a full frame queue must mark the transport dead" + ); + + // Force the socket shut so the writer blocked mid-frame errors out: + // no thread outlives the test unbounded. + t.shutdown(ExitIntent::Disconnect); + drop(theirs); } /// `Quit` teardown is bounded: a daemon that accepts the `Shutdown` frame @@ -337,9 +471,7 @@ mod tests { #[test] fn quit_shutdown_is_bounded_when_the_daemon_never_closes() { let (ours, theirs) = UnixStream::pair().unwrap(); - let write = ours.try_clone().unwrap(); - let (wait_tx, _wait_rx) = channel(); - let mut t = SocketTransport::from_halves(write, ours, wait_tx); + let mut t = transport_over(ours); // Hold `theirs` open, never writing and never closing: the `Shutdown` // frame lands in the socket buffer, but no close ever arrives, so the From c6b85002bfb3986a30b550108375057bba60d2f9 Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 19:16:11 -0700 Subject: [PATCH 5/6] refactor(comments): clarify comments in task view construction and selection tests --- src/app.rs | 4 +- src/app_tests.rs | 13 ++-- src/daemon.rs | 23 ++++---- src/transport.rs | 150 +++++++++++++++++++---------------------------- 4 files changed, 77 insertions(+), 113 deletions(-) diff --git a/src/app.rs b/src/app.rs index 6b18488..7885930 100644 --- a/src/app.rs +++ b/src/app.rs @@ -341,8 +341,8 @@ impl App { /// the socket. pub fn connect(rows: u16, cols: u16) -> io::Result { let (stream, origin) = crate::daemon::connect_ready()?; - // Split the stream here (the fallible part) so the transport factory in - // `assemble` (which owns the wake sender) stays infallible. + // Create the reader and control handles before `assemble`: its + // transport factory cannot return an `io::Result`. let read = stream.try_clone()?; let ctrl = stream.try_clone()?; let mut app = Self::assemble(rows, cols, move |_, _, wait_tx| { diff --git a/src/app_tests.rs b/src/app_tests.rs index f41a31b..1bcabb7 100644 --- a/src/app_tests.rs +++ b/src/app_tests.rs @@ -104,11 +104,9 @@ fn shift(code: KeyCode) -> KeyEvent { KeyEvent::new(code, KeyModifiers::SHIFT) } -/// Minimal live task for injecting into `app.views` directly: carries the -/// fields `sections`/`display_order` sort on (id, cwd, tagged, group) with -/// everything else inert. Ordering tests use these instead of real children -/// because a spawned child that exits mid-test re-buckets its row into -/// Completed and breaks a fixed expected order under load. +/// Construct a stable live task without spawning a child. `Active` and +/// unparked keep an untagged view in Running; a child can exit during the test +/// and move its row to Completed. fn view(id: u64, cwd: PathBuf, tagged: bool, group: Option<&str>) -> TaskView { TaskView { id, @@ -136,9 +134,8 @@ fn selection_follows_task_across_reorder() { app.resolve_selection(); assert_eq!(app.selected_id, Some(1)); - // Tag id 2 -> it sorts into the "In use" bucket, ahead of id 1. Mutated - // in place: the Tag round-trip is other tests' subject, and a pump would - // overwrite the injected views with the core's empty snapshot. + // Tagging id 2 moves it into "In use," ahead of id 1. Mutate the injected + // snapshot directly: a pump would replace it with the empty core snapshot. app.views[1].tagged = true; let order = app.display_order(); diff --git a/src/daemon.rs b/src/daemon.rs index 2c2ff13..4fdd8c1 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -367,14 +367,14 @@ fn no_daemon() -> io::Result<()> { /// `--kill` must work while someone else is attached. The pid comes from the /// lock file (trustworthy while the flock is held: the holder wrote it), and /// daemon exit releases the flock, so acquiring it is the completion signal. -/// A no-op (with a message) if no daemon is running. A held flock with no -/// readable pid is a daemon mid-startup: that is an error, never the no-op, -/// because someone provably holds the lock. +/// A no-op (with a message) if no daemon is running. A held flock without a +/// usable pid is an error: it may be the interval between lock acquisition and +/// pid publication, so it cannot be treated as the no-daemon case. pub fn run_kill() -> io::Result<()> { run_kill_in(&runtime_dir()) } -/// `run_kill` against an explicit runtime directory. +/// Run the `--kill` operation against an explicit runtime directory. fn run_kill_in(dir: &Path) -> io::Result<()> { // The lock PID is a signal target, and the socket receives the client's // environment, so validate the directory before reading either file. @@ -393,10 +393,8 @@ fn run_kill_in(dir: &Path) -> io::Result<()> { Err((file, _)) => file, }; - // The flock is held, but the pid can be momentarily unreadable: - // `run_daemon` acquires the lock, then truncates and writes the pid, so - // the only held-lock window without one is those two syscalls. Poll the - // file briefly rather than guess. + // `run_daemon` acquires the flock before replacing the pid. Retry briefly + // to cover that publication interval. let mut pid = None; for _ in 0..20 { let mut pid_str = String::new(); @@ -856,9 +854,8 @@ mod tests { assert!(notice.contains("--kill"), "{notice}"); } - /// A held flock with no pid is a daemon between lock acquisition and pid - /// write: `--kill` must report an error, not the Ok "no daemon running" - /// no-op, because the holder proves a daemon is starting. + /// A held flock without a usable pid is an error: an existing lock file + /// reaches the no-daemon path only when its flock is acquirable. #[test] fn kill_with_a_held_lock_and_no_pid_is_an_error() { let base = temp("kill_lock_no_pid"); @@ -870,8 +867,8 @@ mod tests { .truncate(false) .open(dir.join("daemon.lock")) .unwrap(); - // flock is per open-file-description, so this handle contends with - // the one `run_kill_in` opens, even within one process. + // The second open creates a distinct open-file description, so it + // contends with this lock even within one process. let _held = Flock::lock(holder, FlockArg::LockExclusiveNonblock).unwrap(); let err = run_kill_in(&dir).unwrap_err(); assert_eq!(err.kind(), ErrorKind::TimedOut); diff --git a/src/transport.rs b/src/transport.rs index de4c68e..9225086 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -42,11 +42,10 @@ pub trait Transport { /// disconnects (the daemon died, or an in-process core panicked), which the /// client surfaces instead of freezing on a stale mirror. fn connected(&self) -> bool; - /// Tear down per `intent`, blocking until it's done, so the client restores - /// the terminal only after the core has acted (tasks killed on `Quit`, the - /// connection closed on `Disconnect`). The wait is not unconditional: an - /// implementation may bound it and hang up on a wedged core — the terminal - /// restore is owed to the user either way. + /// Tear down per `intent` before the client restores the terminal. `Quit` + /// requests core shutdown; `Disconnect` closes only the client connection. + /// An implementation may impose a deadline, then close its connection so a + /// stalled core cannot block terminal restoration indefinitely. fn shutdown(&mut self, intent: ExitIntent); } @@ -154,11 +153,8 @@ impl Drop for ThreadTransport { } } -/// Frames a `send` may queue ahead of the writer thread. 64 bounds the run -/// loop's exposure to 64 frames of delivery latency — not 64 × `SEND_TIMEOUT` -/// of blocking, because `send` never waits on the queue: a full queue means -/// the peer let 64 frames pile up against a 5 s-per-write budget, and the -/// transport declares it dead instead (see `SocketTransport::send`). +/// Maximum encoded command frames awaiting the writer. `send` uses `try_send`: +/// saturation closes the transport instead of blocking the caller. const SEND_QUEUE: usize = 64; /// The core as a separate process (`fleetcom --daemon`), reached over a Unix @@ -168,10 +164,9 @@ const SEND_QUEUE: usize = 64; /// frames back into `Event`s on a channel, so `poll` drains the channel /// exactly like `ThreadTransport`. pub struct SocketTransport { - /// Control handle for forced shutdowns only — frame writes happen on the - /// writer thread. All handles are `try_clone`s of one socket (dup'd FDs - /// share the open socket description), so a shutdown here errors the - /// reader and writer out of their blocking calls. + /// Control handle for forced shutdowns; frame writes happen on the writer + /// thread. Shutting down this handle interrupts socket I/O through the + /// duplicated reader and writer handles. ctrl: UnixStream, /// Encoded frames to the writer thread. `None` once teardown takes it: /// dropping the sender is what ends an idle writer's `recv` loop. @@ -179,20 +174,18 @@ pub struct SocketTransport { evt_rx: Receiver, reader: Option>, writer: Option>, - /// Set when the reader thread ends on socket EOF (the daemon is gone), or - /// when `send` cannot queue a frame (the connection is unrecoverable; see - /// `send`). A write failure on the writer thread lands here indirectly: - /// the writer shuts the socket down, the reader exits on it, and `poll` - /// observes the disconnect. + /// Set by `send` when it cannot queue a frame, or by `poll` when the event + /// channel disconnects. A writer failure shuts down the socket, which ends + /// the reader and disconnects that channel. dead: bool, } impl SocketTransport { - /// Build over pre-split clones of one stream: `write` feeds the writer - /// thread, `read` the reader thread, `ctrl` stays behind for forced - /// shutdowns. The `try_clone`s that can fail are the caller's job: done - /// outside the transport so the App's transport factory stays infallible. - /// `wait_tx` wakes the client's run loop on each inbound event. + /// Build from three handles to one stream: `write` feeds the writer thread, + /// `read` feeds the reader thread, and `ctrl` remains available for forced + /// shutdowns. Callers duplicate the handles before construction so cloning + /// errors remain at the call site. `wait_tx` wakes the client for each + /// inbound event. pub fn from_halves( write: UnixStream, read: UnixStream, @@ -205,11 +198,9 @@ impl SocketTransport { let (frame_tx, frame_rx) = sync_channel::<(u8, Vec)>(SEND_QUEUE); let writer = thread::spawn(move || { let mut write = write; - // Ends when the queue sender drops (teardown, or the transport - // itself dropped) or a frame write fails. `SEND_TIMEOUT` on the - // stream bounds each write; on failure, shut the socket down so - // the reader's blocking `read_frame` errors out too and the `dead` - // flag reaches the client through the existing `poll` path. + // The loop ends when the queue sender drops or a frame write + // fails. A write failure shuts down the socket, which releases the + // reader and lets `poll` observe the event-channel disconnect. while let Ok((kind, payload)) = frame_rx.recv() { if write_frame(&mut write, kind, &payload).is_err() { let _ = write.shutdown(Shutdown::Both); @@ -244,25 +235,20 @@ impl SocketTransport { } } - /// `Quit` teardown: queue `Shutdown`, then wait at most `bound` for the - /// daemon to close the socket — the reader ending is the proof the tasks - /// died. `Shutdown` rides the writer queue like any frame; whether it - /// flushes or not, the bounded `evt_rx` wait is the backstop. The reader - /// owns `evt_tx`, so `evt_rx` disconnecting is exactly the reader ending; - /// events arriving meanwhile are discarded (the client is past polling). - /// On expiry, force our socket shut: the handles are clones of one - /// descriptor, so this errors the reader's blocking `read_frame` out - /// immediately (dropping the write half alone would not interrupt it), - /// which makes the final joins bounded. Production passes `SEND_TIMEOUT`; - /// tests pass a small budget. + /// Queue `Shutdown`, then wait at most `bound` for the event channel to + /// disconnect. The reader owns its sender, so disconnection means the + /// reader ended after socket closure or a read failure. Events received + /// before then are discarded because teardown has started. On expiry, + /// shut down the local socket to release the reader and writer before + /// joining them. fn quit_within(&mut self, bound: Duration) { self.send(Command::Shutdown); let deadline = Instant::now() + bound; loop { let now = Instant::now(); if now >= deadline { - // The daemon never closed the socket: it lost its right to be - // waited on. Hang up so the reader errors out. + // The reader did not finish before the deadline. Shut down + // the local socket to release both worker threads. let _ = self.ctrl.shutdown(Shutdown::Both); break; } @@ -278,12 +264,9 @@ impl SocketTransport { self.join_writer(); } - /// End the writer thread with a bounded join. Every caller has a socket - /// closure already in force (the daemon closed it, or we forced `ctrl` - /// shut), so a writer blocked mid-write errors out immediately; dropping - /// the queue sender is what ends an idle writer's `recv`. Without a prior - /// closure this join could wait a full `SEND_TIMEOUT` — never call it on a - /// live socket. + /// Drop the queue sender and join the writer after socket I/O has ended. An + /// idle writer exits `recv`; socket closure releases an in-flight write. + /// Call only after the reader ends or `ctrl` shuts down the socket. fn join_writer(&mut self) { self.frame_tx.take(); if let Some(h) = self.writer.take() { @@ -306,12 +289,9 @@ impl Transport for SocketTransport { .as_ref() .is_some_and(|tx| tx.try_send((kind, payload)).is_ok()); if !queued { - // Full or disconnected. A full queue means the peer let - // `SEND_QUEUE` frames pile up against a `SEND_TIMEOUT`-per-write - // budget: for our purposes that is a dead peer, the same verdict - // as a failed synchronous write. Mark the connection dead and - // close the socket so the reader and writer exit and the client - // can reconnect. + // A full queue cannot accept this command without blocking; a + // disconnected queue has no writer. Mark the transport dead and + // shut down the socket so both worker threads exit. self.dead = true; let _ = self.ctrl.shutdown(Shutdown::Both); } @@ -327,15 +307,14 @@ impl Transport for SocketTransport { fn shutdown(&mut self, intent: ExitIntent) { match intent { - // Group-kill every task and stop the daemon, then wait for our - // reader to see the daemon close the socket (daemon gone = tasks - // killed) — but only up to `SEND_TIMEOUT`. A wedged daemon that - // never closes does not get a veto on restoring the terminal. + // Request daemon shutdown and wait up to `SEND_TIMEOUT` for the + // peer to close. On timeout, local shutdown releases socket I/O + // before the client restores the terminal. ExitIntent::Quit => self.quit_within(SEND_TIMEOUT), // Close the connection without a Shutdown: the daemon sees EOF and - // keeps the tasks running for the next client to reattach. The - // close we just forced errors the reader out of its blocking read - // and any in-flight frame write, so both joins are bounded. + // keeps the tasks running for the next client to reattach. Local + // shutdown releases the blocking read and any in-flight write + // before both threads are joined. ExitIntent::Disconnect => { let _ = self.ctrl.shutdown(Shutdown::Both); if let Some(h) = self.reader.take() { @@ -383,8 +362,7 @@ impl Transport for LocalTransport { mod tests { use super::*; - /// Build a transport over one end of a socketpair, doing the `try_clone` - /// splitting the production call sites do. + /// Duplicate one socketpair endpoint into write, read, and control handles. fn transport_over(ours: UnixStream) -> SocketTransport { let write = ours.try_clone().unwrap(); let ctrl = ours.try_clone().unwrap(); @@ -392,9 +370,9 @@ mod tests { SocketTransport::from_halves(write, ours, ctrl, wait_tx) } - /// Poll until the transport reports dead or `bound` expires. Dead now - /// propagates asynchronously (writer error → socket shutdown → reader - /// exit → `poll` sees the disconnect), so tests must wait, bounded. + /// Poll until the transport reports dead or `bound` expires. Writer errors + /// propagate through socket shutdown, reader exit, and event-channel + /// disconnection. fn wait_dead(t: &mut SocketTransport, bound: Duration) { let deadline = Instant::now() + bound; while t.connected() && Instant::now() < deadline { @@ -419,27 +397,23 @@ mod tests { }); wait_dead(&mut t, Duration::from_secs(2)); assert!(!t.connected(), "a failed send must mark the transport dead"); - // The stream was shut down along the way, so the reader thread saw - // EOF and exited: joining it cannot hang. + // Peer closure or writer shutdown ends the reader before this join. t.reader.take().unwrap().join().unwrap(); - // The writer ended too — its write failed against the closed peer, or - // its sender is about to drop; either way this join is bounded. + // Dropping the sender releases the writer if it has not observed the + // failed write yet. t.join_writer(); } - /// The regression this design exists for: a peer that stops reading must - /// not block `send` — the run loop's thread is the UI. Fill the socket - /// send buffer and the whole frame queue; every `send` must return - /// promptly and the transport must declare itself dead, leaving recovery - /// to the reconnect path. + /// A peer that stops reading cannot block `send` on the run-loop thread. + /// Saturating the socket and frame queue must return promptly and mark the + /// transport dead. #[test] fn send_burst_against_a_stalled_peer_never_blocks_the_caller() { let (ours, theirs) = UnixStream::pair().unwrap(); let mut t = transport_over(ours); - // 64 KiB per frame: one or two fill the socket send buffer (single- - // digit KiB by default on macOS), the rest fill the `SEND_QUEUE` - // slots, and the overflow must be refused, not waited on. + // Repeated 64 KiB frames saturate the non-reading peer's socket buffer; + // the remaining frames fill the bounded queue. let bytes = vec![b'p'; 64 * 1024]; let start = Instant::now(); for _ in 0..(SEND_QUEUE + 8) { @@ -449,8 +423,6 @@ mod tests { }); } let elapsed = start.elapsed(); - // Unfixed, the first buffer-filling write alone blocks `SEND_TIMEOUT` - // (5 s); the whole burst must stay far under that. assert!( elapsed < Duration::from_secs(2), "send burst blocked the run loop for {elapsed:?}" @@ -460,30 +432,28 @@ mod tests { "a full frame queue must mark the transport dead" ); - // Force the socket shut so the writer blocked mid-frame errors out: - // no thread outlives the test unbounded. + // Release any worker blocked on socket I/O before joining it. t.shutdown(ExitIntent::Disconnect); drop(theirs); } - /// `Quit` teardown is bounded: a daemon that accepts the `Shutdown` frame - /// but never closes the socket cannot block the terminal restore. + /// `Quit` remains bounded when the peer keeps the socket open after the + /// `Shutdown` frame is queued. #[test] fn quit_shutdown_is_bounded_when_the_daemon_never_closes() { let (ours, theirs) = UnixStream::pair().unwrap(); let mut t = transport_over(ours); - // Hold `theirs` open, never writing and never closing: the `Shutdown` - // frame lands in the socket buffer, but no close ever arrives, so the - // reader stays blocked in `read_frame` until the transport hangs up. + // Keep the peer open without sending frames. The reader remains in + // `read_frame` until the quit deadline shuts down the local socket. let (done_tx, done_rx) = channel(); let worker = thread::spawn(move || { t.quit_within(Duration::from_millis(150)); let _ = done_tx.send(()); }); - // Test-side deadline well above the bound: on unfixed code the worker - // blocks in the reader join forever, and this fails the test instead of - // hanging the suite (unwinding drops `theirs`, which unblocks it). + // The outer deadline exceeds the transport bound and prevents the test + // suite from hanging. Unwinding drops `theirs`, which releases the + // worker if the assertion fails. done_rx .recv_timeout(Duration::from_secs(2)) .expect("Quit shutdown must return within its bound"); From e2d8fb01c54a5f667e6d9667b63f21d42382ee44 Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Sun, 16 Aug 2026 19:18:22 -0700 Subject: [PATCH 6/6] refactor(docs): improve clarity in comments for session handling and codex functions --- src/harness/omp.rs | 4 ++-- src/harness/summary.rs | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/harness/omp.rs b/src/harness/omp.rs index 59f98fa..0892f1e 100644 --- a/src/harness/omp.rs +++ b/src/harness/omp.rs @@ -15,8 +15,8 @@ //! Sessions live at `//_.jsonl`. //! The harness home *is* the sessions root: `PI_CODING_AGENT_SESSION_DIR` //! names a sessions directory outright, so no agent-dir value can express it. -//! That override also flattens the store — it is passed straight through as the -//! session file's parent and the bucket level is never computed — so +//! That override also flattens the store: it is passed straight through as the +//! session file's parent and the bucket level is never computed, so //! correlation scans the root and one level below it. //! //! Correlation does not derive bucket names. It enumerates the root and its diff --git a/src/harness/summary.rs b/src/harness/summary.rs index 8443786..378c8c4 100644 --- a/src/harness/summary.rs +++ b/src/harness/summary.rs @@ -456,8 +456,8 @@ fn codex_model_with_reasoning(item: &str) -> bool { matches!(words.len(), 2 | 3) && CODEX_EFFORT.contains(&words[1]) } -/// The composer: the bottom-most column-0 [`CODEX_PROMPT`] row — the glyph -/// alone or the glyph and a space — that is not a modal selector. Rows +/// The composer: the bottom-most column-0 [`CODEX_PROMPT`] row that is not a +/// modal selector. The row is the glyph alone or the glyph and a space. Rows /// below it are tolerated, never required: blank rows, indented affordance /// hints (`tab to queue message`), or the status line. The working layout can /// paint hints below the composer with no status line at all. Prompt echoes in @@ -549,8 +549,8 @@ fn codex_interrupt_paren(s: &str) -> bool { /// The text after codex's compact elapsed counter, or `None` when `s` does /// not open with one: space-separated `{digits}{unit}` fields in strictly -/// descending `h`, `m`, `s` order, ending at the seconds field — `0s`, -/// `1m 00s`, `25h 02m 03s`. A field that is not digits plus a unit (`1/3`, +/// descending `h`, `m`, `s` order, ending at the seconds field (`0s`, +/// `1m 00s`, `25h 02m 03s`). A field that is not digits plus a unit (`1/3`, /// `9.9s`) fails. fn codex_elapsed(s: &str) -> Option<&str> { let mut rest = s; @@ -805,7 +805,7 @@ fn omp_approval(rows: &[String]) -> Option<(String, &'static str)> { } /// The selector's chosen row: a cursor spelling, a space, then `Approve` and -/// nothing more. Equality after the cursor is the whole check — the ascii +/// nothing more. Equality after the cursor is the whole check: the ascii /// cursor `>` also opens a quoted line, so the row's remainder has to be /// exact. fn omp_approve_row(row: &str) -> bool { @@ -816,7 +816,7 @@ fn omp_approve_row(row: &str) -> bool { } /// The selector's head row: `Allow tool: {name}`. The prefix's trailing -/// space carries the name requirement — a trimmed row cannot end in one — so +/// space carries the name requirement: a trimmed row cannot end in one, so /// a bare `Allow tool:` fails. fn omp_allow_head(row: &str) -> bool { row.trim().starts_with("Allow tool: ")