diff --git a/src/app.rs b/src/app.rs index 8ae21a5..7885930 100644 --- a/src/app.rs +++ b/src/app.rs @@ -341,11 +341,12 @@ 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| { - 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/app_tests.rs b/src/app_tests.rs index 9fb66cc..1bcabb7 100644 --- a/src/app_tests.rs +++ b/src/app_tests.rs @@ -104,21 +104,39 @@ fn shift(code: KeyCode) -> KeyEvent { KeyEvent::new(code, KeyModifiers::SHIFT) } +/// 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, + 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(); + // 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(); assert_eq!(app.views[order[0]].id, 2, "tagged task should sort first"); @@ -134,9 +152,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 +2188,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 +2210,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]); diff --git a/src/daemon.rs b/src/daemon.rs index a7bb80f..4fdd8c1 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 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<()> { - let dir = runtime_dir(); + run_kill_in(&runtime_dir()) +} + +/// 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. - 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,25 @@ 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); + // `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(); + 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 +437,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 +854,25 @@ mod tests { assert!(notice.contains("--kill"), "{notice}"); } - /// A missing hello response times out the kill exchange. + /// 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_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(); + // 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); - 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. 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: ") diff --git a/src/transport.rs b/src/transport.rs index 205e745..9225086 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -5,9 +5,12 @@ use std::{ os::unix::net::UnixStream, sync::{ atomic::AtomicBool, - mpsc::{Receiver, Sender, TryRecvError, channel}, + mpsc::{ + Receiver, RecvTimeoutError, Sender, SyncSender, TryRecvError, channel, sync_channel, + }, }, thread::{self, JoinHandle}, + time::{Duration, Instant}, }; use crate::{ @@ -39,9 +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`). + /// 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); } @@ -149,28 +153,61 @@ impl Drop for ThreadTransport { } } +/// 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 -/// 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; 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. + frame_tx: Option)>>, evt_rx: Receiver, reader: 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`). + writer: Option>, + /// 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 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 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, + 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; + // 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); + break; + } + } + }); let (evt_tx, evt_rx) = channel(); let reader = thread::spawn(move || { let mut read = read; @@ -189,12 +226,53 @@ impl SocketTransport { let _ = wait_tx.send(()); }); Self { - write, + ctrl, + frame_tx: Some(frame_tx), evt_rx, reader: Some(reader), + writer: Some(writer), dead: false, } } + + /// 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 reader did not finish before the deadline. Shut down + // the local socket to release both worker threads. + let _ = self.ctrl.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(); + } + self.join_writer(); + } + + /// 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() { + let _ = h.join(); + } + } } impl Transport for SocketTransport { @@ -204,12 +282,18 @@ 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 { + // 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.write.shutdown(Shutdown::Both); + let _ = self.ctrl.shutdown(Shutdown::Both); } } @@ -223,20 +307,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), + // 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. + // 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.write.shutdown(Shutdown::Both); + let _ = self.ctrl.shutdown(Shutdown::Both); + if let Some(h) = self.reader.take() { + let _ = h.join(); + } + self.join_writer(); } } - // 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(); - } } } @@ -276,13 +362,32 @@ impl Transport for LocalTransport { mod tests { use super::*; - /// A failed send marks the transport disconnected and stops its reader. + /// 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(); + let (wait_tx, _wait_rx) = channel(); + SocketTransport::from_halves(write, ours, ctrl, wait_tx) + } + + /// 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 { + 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 @@ -290,9 +395,69 @@ 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. + // Peer closure or writer shutdown ends the reader before this join. t.reader.take().unwrap().join().unwrap(); + // Dropping the sender releases the writer if it has not observed the + // failed write yet. + t.join_writer(); + } + + /// 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); + + // 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) { + t.send(Command::Paste { + id: 1, + bytes: bytes.clone(), + }); + } + let elapsed = start.elapsed(); + 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" + ); + + // Release any worker blocked on socket I/O before joining it. + t.shutdown(ExitIntent::Disconnect); + drop(theirs); + } + + /// `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); + + // 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(()); + }); + // 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"); + worker.join().unwrap(); + drop(theirs); } }