diff --git a/Cargo.lock b/Cargo.lock index 489a87d..46b8cd0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2583,6 +2583,7 @@ dependencies = [ "pmacs-protocol", "pollster", "sys-locale", + "tempfile", "unicode-width", "wgpu", "winit", diff --git a/pmacs-gpu/Cargo.toml b/pmacs-gpu/Cargo.toml index c00923d..40bc3ee 100644 --- a/pmacs-gpu/Cargo.toml +++ b/pmacs-gpu/Cargo.toml @@ -62,3 +62,6 @@ pollster = "0.4.0" wgpu = "29.0.3" winit = "0.30.13" unicode-width = "0.2" + +[dev-dependencies] +tempfile = "3" diff --git a/pmacs-gpu/src/attach.rs b/pmacs-gpu/src/attach.rs index 2b8a0e9..35a9ba6 100644 --- a/pmacs-gpu/src/attach.rs +++ b/pmacs-gpu/src/attach.rs @@ -126,6 +126,8 @@ pub enum ManagedAttachError { connect: io::Error, /// Observed daemon process outcome, when it exited early. daemon_status: Option, + /// Startup deadline used for this attempt. + timeout: Duration, }, } @@ -155,10 +157,11 @@ impl std::fmt::Display for ManagedAttachError { socket, connect, daemon_status, + timeout, } => { write!( f, - "daemon did not become attachable on {} within 5 seconds: {connect}", + "daemon did not become attachable on {} within {timeout:?}: {connect}", socket.display() )?; if let Some(status) = daemon_status { @@ -608,7 +611,7 @@ fn spawn_daemon(daemon_executable: &Path, socket_path: &Path) -> io::Result, facts: &ManagedDaemonFacts) { + if let Some(child) = child.take() { + start_daemon_reaper(child, facts.clone()); + } +} + #[allow(clippy::too_many_arguments)] fn connect_managed_inner( socket_path: &Path, @@ -682,49 +691,60 @@ where } } - let mut child = spawner(daemon_executable, socket_path).map_err(|source| { + let child = spawner(daemon_executable, socket_path).map_err(|source| { ManagedAttachError::SpawnDaemon { executable: daemon_executable.to_owned(), source, } })?; let daemon = ManagedDaemonFacts::spawned(child.id()); + let mut child = Some(child); let deadline = Instant::now() + timeout; loop { match connector(socket_path) { Ok(stream) => { let attached = connect_stream_with_sink(stream, sink); - start_daemon_reaper(child, daemon.clone()); + hand_off_daemon_child(&mut child, &daemon); return Ok(ManagedAttach { client: attached?, daemon, }); } Err(error) => { - if !post_spawn_retryable(socket_path, &error)? { + let retryable = match post_spawn_retryable(socket_path, &error) { + Ok(retryable) => retryable, + Err(classification_error) => { + hand_off_daemon_child(&mut child, &daemon); + return Err(classification_error); + } + }; + if !retryable { + hand_off_daemon_child(&mut child, &daemon); return Err(AttachClientError::Connect(error).into()); } - if let Some(status) = child + let wait_status = match child + .as_mut() + .expect("managed daemon child handed off only on return") .try_wait() - .map_err(ManagedAttachError::ObserveDaemon)? { + Ok(status) => status, + Err(observe_error) => { + hand_off_daemon_child(&mut child, &daemon); + return Err(ManagedAttachError::ObserveDaemon(observe_error)); + } + }; + if let Some(status) = wait_status { daemon.record_wait(status.to_string()); } - if daemon.daemon_reaped() { - let status = daemon.daemon_wait_result(); - if Instant::now() >= deadline { - return Err(ManagedAttachError::StartupTimeout { - socket: socket_path.to_owned(), - connect: error, - daemon_status: status, - }); - } - } else if Instant::now() >= deadline { + if Instant::now() >= deadline { + let daemon_status = daemon.daemon_wait_result(); + hand_off_daemon_child(&mut child, &daemon); return Err(ManagedAttachError::StartupTimeout { socket: socket_path.to_owned(), connect: error, - daemon_status: None, + daemon_status, + timeout, }); } thread::sleep( @@ -907,8 +927,8 @@ impl AttachClient { cvar.notify_one(); Ok(()) } else { - // Refused because the outbox is closed — a lossless overflow - // against a stalled daemon (or an earlier writer failure). + // Refusing a lossless event prevents replica divergence against + // a stalled daemon (or an earlier writer failure). // Tear the session down actively (F-008): shut the socket so // the reader wakes with EOF and fires `Disconnected`, giving a // visible "(daemon disconnected)" instead of a GPU that keeps @@ -927,7 +947,7 @@ impl AttachClient { #[cfg(test)] mod tests { use super::*; - use pmacs_protocol::{InstanceCapabilities, MouseButton}; + use pmacs_protocol::{InstanceCapabilities, InstanceIdentity, MouseButton}; fn caps( multi_frontend: bool, @@ -1197,7 +1217,9 @@ mod tests { } #[test] fn managed_attach_starts_only_for_absent_or_refused_sockets() { - let socket = Path::new("/tmp/pmacs-managed-test.sock"); + let temp = tempfile::tempdir().expect("tempdir"); + let socket_path = temp.path().join("managed.sock"); + let socket = socket_path.as_path(); assert!( initial_startup_authorized(socket, &io::Error::new(io::ErrorKind::NotFound, "absent")) .expect("classify absent socket") @@ -1225,7 +1247,9 @@ mod tests { #[test] fn managed_retry_adds_only_interrupted_and_would_block() { - let socket = Path::new("/tmp/pmacs-managed-test.sock"); + let temp = tempfile::tempdir().expect("tempdir"); + let socket_path = temp.path().join("managed.sock"); + let socket = socket_path.as_path(); for kind in [io::ErrorKind::Interrupted, io::ErrorKind::WouldBlock] { assert!( post_spawn_retryable(socket, &io::Error::new(kind, "transient")) @@ -1266,8 +1290,10 @@ mod tests { #[test] fn managed_attach_fails_closed_on_non_retryable_connect_errors() { + let temp = tempfile::tempdir().expect("tempdir"); + let socket = temp.path().join("managed.sock"); let result = connect_managed_inner( - Path::new("/tmp/pmacs-managed-test.sock"), + &socket, Path::new("/unused/pmacs"), |_| { Err(io::Error::new( @@ -1286,4 +1312,69 @@ mod tests { if error.kind() == io::ErrorKind::PermissionDenied )); } + + #[test] + fn managed_retry_survives_transients_and_uses_the_successful_stream() { + let temp = tempfile::tempdir().expect("tempdir"); + let socket = temp.path().join("managed.sock"); + let (client_stream, mut server_stream) = UnixStream::pair().expect("socket pair"); + let server = thread::spawn(move || { + let hello = Hello { + protocol_version: PROTOCOL_VERSION, + assigned_frontend_id: FrontendId::LOCAL, + instance_identity: InstanceIdentity { + pmacs_version: "managed-retry-test".to_owned(), + build_hash: None, + instance_name: None, + uptime_secs: 0, + working_directory: "/tmp".to_owned(), + }, + instance_capabilities: caps(true, true, true), + }; + write_message(&mut server_stream, &hello).expect("write Hello"); + let _: AttachRequest = + read_message(&mut server_stream).expect("read real AttachRequest"); + }); + let mut attempts = 0; + let mut client_stream = Some(client_stream); + let managed = connect_managed_inner( + &socket, + Path::new("/bin/sh"), + |_| { + attempts += 1; + match attempts { + 1 => Err(io::Error::new(io::ErrorKind::NotFound, "initial miss")), + 2 => Err(io::Error::new(io::ErrorKind::Interrupted, "signal")), + 3 => Err(io::Error::new(io::ErrorKind::WouldBlock, "backlog")), + 4 => Ok(client_stream.take().expect("single successful stream")), + _ => panic!("unexpected connection attempt"), + } + }, + |_, _| Command::new("/bin/sh").args(["-c", "exit 0"]).spawn(), + Duration::from_secs(1), + Duration::ZERO, + |_| true, + ) + .expect("transient sequence must attach"); + assert_eq!(attempts, 4); + assert!(managed.daemon.spawned_daemon()); + assert_eq!(managed.client.server_protocol_version(), PROTOCOL_VERSION); + server.join().expect("handshake server"); + } + + #[test] + fn startup_timeout_reports_the_configured_duration() { + let error = ManagedAttachError::StartupTimeout { + socket: PathBuf::from("/tmp/unused.sock"), + connect: io::Error::new(io::ErrorKind::NotFound, "still absent"), + daemon_status: Some("exit status: 17".to_owned()), + timeout: Duration::from_millis(1), + }; + let message = error.to_string(); + assert!( + message.contains("1ms"), + "unexpected timeout message: {message}" + ); + assert!(!message.contains("5 seconds")); + } } diff --git a/pmacs-gpu/src/main.rs b/pmacs-gpu/src/main.rs index a312108..0fb98df 100644 --- a/pmacs-gpu/src/main.rs +++ b/pmacs-gpu/src/main.rs @@ -651,6 +651,7 @@ fn main() { proxy: Some(proxy), state: None, attach_client, + pending_events: Vec::new(), modifiers: winit::keyboard::ModifiersState::empty(), }; event_loop @@ -1016,10 +1017,15 @@ fn frame_probe_text(frame: &TerminalFrame) -> String { const GPU_USAGE: &str = "\ pmacs-gpu — GPU frontend for pmacs -USAGE: - pmacs-gpu --attach attach to an existing daemon - pmacs-gpu --help print this help - pmacs-gpu --version print package and protocol versions"; +NORMAL STARTUP: + pmacs --gpu [--socket NAME|PATH] start or reuse a managed daemon + +ADVANCED DIRECT ATTACH: + pmacs-gpu --attach attach to an existing daemon only + +OPTIONS: + pmacs-gpu --help print this help + pmacs-gpu --version print package and protocol versions"; /// Strict parser for direct, managed, and headless GPU entry points. fn parse_args(args: &[String]) -> Result { @@ -1046,7 +1052,13 @@ fn parse_args(args: &[String]) -> Result { daemon_executable: PathBuf::from(daemon_executable), }) } - [] => Err("an explicit mode is required; use --attach ".to_owned()), + [] => Err( + "managed startup is provided by `pmacs --gpu`; direct use requires --attach " + .to_owned(), + ), + [flag, ..] if matches!(flag.as_str(), "--help" | "-h" | "--version" | "-V") => { + Err(format!("{flag} does not accept operands")) + } [flag, ..] if matches!( flag.as_str(), @@ -1070,6 +1082,10 @@ struct App { /// a non-Option in a borrow. proxy: Option>, state: Option, + /// User events received before winit creates `state`. Managed attach + /// starts its reader before `run_app`, so the initial snapshot may arrive + /// before `resumed` on backends with a different callback order. + pending_events: Vec, /// Held both for stream lifetime and for the main loop's /// `send_viewport` / `send_key` write-back path. attach_client: Option, @@ -1079,6 +1095,19 @@ struct App { modifiers: winit::keyboard::ModifiersState, } +fn defer_app_event( + state_ready: bool, + pending: &mut Vec, + event: AppEvent, +) -> Option { + if state_ready { + Some(event) + } else { + pending.push(event); + None + } +} + type LoroTextDeltaBatches = Arc>>>; /// All resources owned by one running pmacs-gpu instance. @@ -1759,6 +1788,69 @@ impl App { eprintln!("pmacs-gpu: send_menu_pointer failed: {e}"); } } + + fn dispatch_app_event(&mut self, event: AppEvent) { + let state = self + .state + .as_mut() + .expect("app events dispatch only after state initialization"); + match event { + AppEvent::Attach(AttachEvent::Message(msg)) => { + let debug_apply = debug_apply(); + let apply_start = debug_apply.then(std::time::Instant::now); + let label = debug_apply.then(|| instance_message_label(msg.as_ref())); + let follow_up = state.apply_attach_message(*msg); + if let (Some(start), Some(label)) = (apply_start, label) { + eprintln!( + "pmacs-gpu apply: {label}={}us", + std::time::Instant::now().duration_since(start).as_micros() + ); + } + // If the message triggered a follow-up Viewport + // (currently: every BufferSnapshot does), emit it back + // to the daemon. The daemon's `SemanticRenderState` + // produces no styling until a viewport is declared. + if let Some(ViewportSend { + buffer_id, + visible, + generation, + }) = follow_up + && let Some(client) = self.attach_client.as_ref() + && let Err(e) = client.send_viewport(buffer_id, visible, generation) + { + eprintln!("pmacs-gpu: send Viewport failed: {e}"); + } + // Vterm Stage 3 — the dual declaration. After every + // snapshot the frontend re-declares BOTH its byte + // viewport (above) and its terminal cell size, because + // an empty terminal identity snapshot does not announce + // itself as a terminal. The daemon keeps whichever one + // matches the buffer's kind, which is what breaks the + // otherwise circular "need a frame to know to ask for + // one" dependency. + self.flush_terminal_declaration(); + let Some(state) = self.state.as_mut() else { + return; + }; + state.release_timed_out_floor(); + let ready_keys = state.take_ready_round_trip_keys(); + if let Some(client) = self.attach_client.as_ref() { + for (key, mods) in ready_keys { + if debug_input() { + eprintln!("pmacs-gpu flush_key: {key:?} mods={mods:?}"); + } + if let Err(e) = client.send_key(key, mods) { + eprintln!("pmacs-gpu: flush send_key failed: {e}"); + } + } + } + } + AppEvent::Attach(AttachEvent::Disconnected(reason)) => { + eprintln!("pmacs-gpu: daemon disconnected ({reason})"); + state.on_daemon_disconnected("(daemon disconnected)"); + } + } + } } impl ApplicationHandler for App { @@ -1797,6 +1889,9 @@ impl ApplicationHandler for App { } } } + for event in std::mem::take(&mut self.pending_events) { + self.dispatch_app_event(event); + } } #[allow(clippy::too_many_lines)] // linear per-event dispatch; splitting hides the input flow. @@ -2386,64 +2481,9 @@ impl ApplicationHandler for App { } fn user_event(&mut self, _event_loop: &ActiveEventLoop, event: AppEvent) { - let Some(state) = self.state.as_mut() else { - return; - }; - match event { - AppEvent::Attach(AttachEvent::Message(msg)) => { - let debug_apply = debug_apply(); - let apply_start = debug_apply.then(std::time::Instant::now); - let label = debug_apply.then(|| instance_message_label(msg.as_ref())); - let follow_up = state.apply_attach_message(*msg); - if let (Some(start), Some(label)) = (apply_start, label) { - eprintln!( - "pmacs-gpu apply: {label}={}us", - std::time::Instant::now().duration_since(start).as_micros() - ); - } - // If the message triggered a follow-up Viewport - // (currently: every BufferSnapshot does), emit it back - // to the daemon. The daemon's `SemanticRenderState` - // produces no styling until a viewport is declared. - if let Some(ViewportSend { - buffer_id, - visible, - generation, - }) = follow_up - && let Some(client) = self.attach_client.as_ref() - && let Err(e) = client.send_viewport(buffer_id, visible, generation) - { - eprintln!("pmacs-gpu: send Viewport failed: {e}"); - } - // Vterm Stage 3 — the dual declaration. After every - // snapshot the frontend re-declares BOTH its byte - // viewport (above) and its terminal cell size, because - // an empty terminal identity snapshot does not announce - // itself as a terminal. The daemon keeps whichever one - // matches the buffer's kind, which is what breaks the - // otherwise circular "need a frame to know to ask for - // one" dependency. - self.flush_terminal_declaration(); - let Some(state) = self.state.as_mut() else { - return; - }; - state.release_timed_out_floor(); - let ready_keys = state.take_ready_round_trip_keys(); - if let Some(client) = self.attach_client.as_ref() { - for (key, mods) in ready_keys { - if debug_input() { - eprintln!("pmacs-gpu flush_key: {key:?} mods={mods:?}"); - } - if let Err(e) = client.send_key(key, mods) { - eprintln!("pmacs-gpu: flush send_key failed: {e}"); - } - } - } - } - AppEvent::Attach(AttachEvent::Disconnected(reason)) => { - eprintln!("pmacs-gpu: daemon disconnected ({reason})"); - state.on_daemon_disconnected("(daemon disconnected)"); - } + if let Some(event) = defer_app_event(self.state.is_some(), &mut self.pending_events, event) + { + self.dispatch_app_event(event); } } } @@ -14287,4 +14327,40 @@ mod tests { ); } } + + #[test] + fn pre_state_app_events_are_buffered_in_arrival_order() { + let mut pending = Vec::new(); + for reason in ["snapshot-predecessor", "snapshot-successor"] { + let event = AppEvent::Attach(AttachEvent::Disconnected(reason.to_owned())); + assert!(defer_app_event(false, &mut pending, event).is_none()); + } + assert_eq!(pending.len(), 2); + let reasons = pending + .into_iter() + .map(|event| match event { + AppEvent::Attach(AttachEvent::Disconnected(reason)) => reason, + AppEvent::Attach(AttachEvent::Message(_)) => panic!("unexpected message"), + }) + .collect::>(); + assert_eq!(reasons, ["snapshot-predecessor", "snapshot-successor"]); + + let immediate = AppEvent::Attach(AttachEvent::Disconnected("ready".to_owned())); + assert!(defer_app_event(true, &mut Vec::new(), immediate).is_some()); + } + + #[test] + fn gpu_cli_points_bare_invocation_to_broker_and_labels_direct_attach() { + let bare = parse_args(&[]).expect_err("bare GPU invocation must fail"); + assert!( + bare.contains("pmacs --gpu"), + "unexpected bare error: {bare}" + ); + assert!(GPU_USAGE.contains("NORMAL STARTUP")); + assert!(GPU_USAGE.contains("ADVANCED DIRECT ATTACH")); + + let extra = ["--help", "extra"].map(str::to_owned); + let error = parse_args(&extra).expect_err("help operands must fail"); + assert_eq!(error, "--help does not accept operands"); + } } diff --git a/src/main.rs b/src/main.rs index 167ed63..881b3da 100644 --- a/src/main.rs +++ b/src/main.rs @@ -301,7 +301,7 @@ fn gpu_binary(current_exe: &Path, override_bin: Option) -> (PathBuf, Pa if let Some(override_bin) = override_bin { return (override_bin, sibling); } - if sibling.exists() { + if sibling.is_file() { return (sibling.clone(), sibling); } (PathBuf::from("pmacs-gpu"), sibling) @@ -859,6 +859,11 @@ mod tests { assert_eq!(selected, override_bin); assert_eq!(reported_sibling, sibling); + std::fs::create_dir(&sibling).expect("create sibling directory"); + let (selected, _) = gpu_binary(&root, None); + assert_eq!(selected, PathBuf::from("pmacs-gpu")); + std::fs::remove_dir(&sibling).expect("remove sibling directory"); + std::fs::write(&sibling, b"gpu").expect("create sibling"); let (selected, _) = gpu_binary(&root, None); assert_eq!(selected, sibling); diff --git a/tests/gpu_invocation_acceptance.rs b/tests/gpu_invocation_acceptance.rs index 57db913..efdaa34 100644 --- a/tests/gpu_invocation_acceptance.rs +++ b/tests/gpu_invocation_acceptance.rs @@ -173,11 +173,22 @@ mod crdt { impl ManagedProbe { fn spawn(socket: &Path, report: &Path, daemon_executable: &Path, home: &Path) -> Self { + Self::spawn_with_env(socket, report, daemon_executable, home, &[]) + } + + fn spawn_with_env( + socket: &Path, + report: &Path, + daemon_executable: &Path, + home: &Path, + envs: &[(&str, &Path)], + ) -> Self { assert!( gpu_binary().is_file(), "build pmacs-gpu before this acceptance suite" ); - let mut child = Command::new(gpu_binary()) + let mut command = Command::new(gpu_binary()); + command .args(["--headless-managed-probe"]) .arg(socket) .arg(report) @@ -186,9 +197,11 @@ mod crdt { .env("XDG_CONFIG_HOME", home) .stdin(Stdio::piped()) .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("spawn managed probe"); + .stderr(Stdio::null()); + for (key, value) in envs { + command.env(key, value); + } + let mut child = command.spawn().expect("spawn managed probe"); let stdin = child.stdin.take().expect("probe stdin"); Self { child, @@ -325,37 +338,78 @@ mod crdt { } #[test] - fn concurrent_managed_launches_converge_on_one_socket_owner() { + fn concurrent_managed_launches_converge_and_reap_the_lock_loser() { let temp = secure_tempdir(); let socket = temp.path().join("race.sock"); - let mut first = ManagedProbe::spawn( + let first_ready = temp.path().join("first-ready"); + let second_ready = temp.path().join("second-ready"); + let first_wrapper = temp.path().join("first-daemon"); + let second_wrapper = temp.path().join("second-daemon"); + let wrapper = "touch \"$PMACS_BARRIER_SELF\"\n\ + while [ ! -e \"$PMACS_BARRIER_PEER\" ]; do sleep 0.01; done\n\ + exec \"$PMACS_REAL_DAEMON\" \"$@\""; + write_script(&first_wrapper, wrapper); + write_script(&second_wrapper, wrapper); + let real_daemon = pmacs_binary(); + + let mut first = ManagedProbe::spawn_with_env( &socket, &temp.path().join("first-report"), - &pmacs_binary(), + &first_wrapper, temp.path(), + &[ + ("PMACS_BARRIER_SELF", &first_ready), + ("PMACS_BARRIER_PEER", &second_ready), + ("PMACS_REAL_DAEMON", &real_daemon), + ], ); - let mut second = ManagedProbe::spawn( + let mut second = ManagedProbe::spawn_with_env( &socket, &temp.path().join("second-report"), - &pmacs_binary(), + &second_wrapper, temp.path(), + &[ + ("PMACS_BARRIER_SELF", &second_ready), + ("PMACS_BARRIER_PEER", &first_ready), + ("PMACS_REAL_DAEMON", &real_daemon), + ], ); let first_facts = first.wait_ready(); let second_facts = second.wait_ready(); assert_eq!( - first_facts.get("buffer_snapshot").map(String::as_str), + first_facts.get("spawned_daemon").map(String::as_str), Some("true") ); assert_eq!( - second_facts.get("buffer_snapshot").map(String::as_str), + second_facts.get("spawned_daemon").map(String::as_str), Some("true") ); assert!(UnixStream::connect(&socket).is_ok()); - let pids = [first.daemon_pid, second.daemon_pid]; - assert!(first.close().success()); - assert!(second.close().success()); - for pid in pids.into_iter().flatten() { - signal_pid(pid, Signal::SIGTERM); + + let deadline = Instant::now() + Duration::from_secs(5); + let first_lost = loop { + let first_reaped = parse_report(&first.report) + .get("daemon_reaped") + .is_some_and(|value| value == "true"); + let second_reaped = parse_report(&second.report) + .get("daemon_reaped") + .is_some_and(|value| value == "true"); + if first_reaped ^ second_reaped { + break first_reaped; + } + assert!( + Instant::now() < deadline, + "exactly one losing daemon child was not reaped" + ); + thread::sleep(Duration::from_millis(20)); + }; + + if first_lost { + assert!(first.close().success()); + assert!(second.close().success()); + } else { + assert!(second.close().success()); + assert!(first.close().success()); } } @@ -527,7 +581,9 @@ mod crdt { .output() .expect("GPU help"); assert!(help.status.success()); - assert!(String::from_utf8_lossy(&help.stdout).contains("--attach ")); + let help_text = String::from_utf8_lossy(&help.stdout); + assert!(help_text.contains("pmacs --gpu")); + assert!(help_text.contains("ADVANCED DIRECT ATTACH")); let version = Command::new(gpu_binary()) .arg("--version") @@ -536,8 +592,20 @@ mod crdt { assert!(version.status.success()); assert!(String::from_utf8_lossy(&version.stdout).contains("protocol v")); + let bare = Command::new(gpu_binary()) + .output() + .expect("bare GPU invocation"); + assert_eq!(bare.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&bare.stderr).contains("pmacs --gpu")); + + let help_extra = Command::new(gpu_binary()) + .args(["--help", "extra"]) + .output() + .expect("GPU help with extra operand"); + assert_eq!(help_extra.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&help_extra.stderr).contains("does not accept operands")); + for argv in [ - vec![], vec!["--attach"], vec!["--attach", "/tmp/x.sock", "ignored"], vec!["unexpected"],