From 3d6f3e78624f1308a360cbf02eeca3fec2434436 Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Wed, 22 Apr 2026 21:23:18 +0300 Subject: [PATCH] test: reduce persistence integration test sleeps and harness duplication --- tests/client_mode.rs | 303 +++++-------------------------- tests/detach_reattach.rs | 373 +++++++-------------------------------- tests/support/mod.rs | 267 ++++++++++++++++++++++++++++ 3 files changed, 371 insertions(+), 572 deletions(-) diff --git a/tests/client_mode.rs b/tests/client_mode.rs index 882bcb1c..8860f65d 100644 --- a/tests/client_mode.rs +++ b/tests/client_mode.rs @@ -12,8 +12,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; use support::{ - cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, - unregister_spawned_herdr_pid, + cleanup_test_base, client_handshake, encode_varint_u16, encode_varint_u32, frame_message, + read_server_message, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, wait_for_file, wait_for_message_variant, wait_for_socket, + wait_until, }; fn unique_test_dir() -> PathBuf { @@ -66,28 +68,6 @@ fn test_lock() -> MutexGuard<'static, ()> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } -fn wait_for_socket(path: &PathBuf, timeout: Duration) { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if path.exists() && UnixStream::connect(path).is_ok() { - return; - } - thread::sleep(Duration::from_millis(25)); - } - panic!("socket did not appear at {}", path.display()); -} - -fn wait_for_file(path: &PathBuf, timeout: Duration) { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if path.exists() { - return; - } - thread::sleep(Duration::from_millis(25)); - } - panic!("file did not appear at {}", path.display()); -} - fn spawn_client_process( config_home: &PathBuf, runtime_dir: &PathBuf, @@ -177,203 +157,6 @@ fn ping_socket(socket_path: &PathBuf) -> String { response.trim().to_string() } -// --------------------------------------------------------------------------- -// Bincode v2 varint encoding helpers (same as server_headless.rs tests) -// --------------------------------------------------------------------------- - -/// Encode a varint u32 value according to bincode v2 VarintEncoding. -fn encode_varint_u32(v: u32) -> Vec { - if v < 251 { - vec![v as u8] - } else if v < 65536 { - let mut buf = vec![251u8]; - buf.extend_from_slice(&(v as u16).to_le_bytes()); - buf - } else { - let mut buf = vec![252u8]; - buf.extend_from_slice(&v.to_le_bytes()); - buf - } -} - -/// Encode a varint u16 value. -fn encode_varint_u16(v: u16) -> Vec { - if v < 251 { - vec![v as u8] - } else { - let mut buf = vec![251u8]; - buf.extend_from_slice(&v.to_le_bytes()); - buf - } -} - -/// Encode an enum variant with its fields. -fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec { - let mut buf = encode_varint_u32(variant_idx); - for field in fields { - buf.extend_from_slice(field); - } - buf -} - -/// Frame a message with u32LE length prefix. -fn frame_message(payload: &[u8]) -> Vec { - let len = payload.len() as u32; - let mut framed = len.to_le_bytes().to_vec(); - framed.extend_from_slice(payload); - framed -} - -/// Decode a varint u32 from a byte slice at the given offset. -fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { - if offset >= payload.len() { - return Err("payload too short for varint".into()); - } - let first_byte = payload[offset]; - match first_byte { - 0..=250 => Ok((first_byte as u32, 1)), - 251 => { - if offset + 3 > payload.len() { - return Err("payload too short for u16 varint".into()); - } - let v = u16::from_le_bytes( - payload[offset + 1..offset + 3] - .try_into() - .map_err(|e: std::array::TryFromSliceError| e.to_string())?, - ); - Ok((v as u32, 3)) - } - 252 => { - if offset + 5 > payload.len() { - return Err("payload too short for u32 varint".into()); - } - let v = u32::from_le_bytes( - payload[offset + 1..offset + 5] - .try_into() - .map_err(|e: std::array::TryFromSliceError| e.to_string())?, - ); - Ok((v, 5)) - } - _ => Err(format!("unsupported varint tag: {first_byte}")), - } -} - -/// Sends a Hello message over the client socket and reads the Welcome response. -fn client_handshake( - stream: &mut UnixStream, - version: u32, - cols: u16, - rows: u16, -) -> Result<(u32, Option), String> { - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .map_err(|e| e.to_string())?; - - // Encode Hello message using bincode v2 varint format. - // ClientMessage::Hello is variant 0. - let hello_payload = encode_varint_enum( - 0, - &[ - &encode_varint_u32(version), - &encode_varint_u16(cols), - &encode_varint_u16(rows), - ], - ); - let framed = frame_message(&hello_payload); - stream.write_all(&framed).map_err(|e| e.to_string())?; - stream.flush().map_err(|e| e.to_string())?; - - // Read the framed response. - let mut len_buf = [0u8; 4]; - stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; - let len = u32::from_le_bytes(len_buf) as usize; - - if len > 2 * 1024 * 1024 { - return Err(format!("oversized response: {len}")); - } - - let mut payload = vec![0u8; len]; - stream.read_exact(&mut payload).map_err(|e| e.to_string())?; - - // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option } - decode_welcome(&payload) -} - -/// Decode a ServerMessage::Welcome from bincode v2 payload. -fn decode_welcome(payload: &[u8]) -> Result<(u32, Option), String> { - let mut offset = 0; - - // Variant index (should be 0 for Welcome) - let (variant, consumed) = decode_varint_u32(payload, offset)?; - offset += consumed; - if variant != 0 { - return Err(format!( - "expected Welcome (variant 0), got variant {variant}" - )); - } - - // version: u32 - let (version, consumed) = decode_varint_u32(payload, offset)?; - offset += consumed; - - // error: Option — discriminant is always 1 byte - if offset >= payload.len() { - return Err("payload too short for Option tag".into()); - } - let option_tag = payload[offset]; - offset += 1; - - let error = if option_tag == 1 { - // Some(String) — length as varint + UTF-8 bytes - let (str_len, consumed) = decode_varint_u32(payload, offset)?; - offset += consumed; - let str_len = str_len as usize; - - if offset + str_len > payload.len() { - return Err("payload too short for string content".into()); - } - let s = String::from_utf8(payload[offset..offset + str_len].to_vec()) - .map_err(|e| e.to_string())?; - Some(s) - } else { - None - }; - - Ok((version, error)) -} - -/// Read a framed ServerMessage from the stream, returning the variant index -/// and the raw payload for further decoding. -fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec), String> { - stream - .set_read_timeout(Some(Duration::from_secs(10))) - .map_err(|e| e.to_string())?; - - let mut len_buf = [0u8; 4]; - stream - .read_exact(&mut len_buf) - .map_err(|e| format!("read length prefix: {e}"))?; - let len = u32::from_le_bytes(len_buf) as usize; - - if len > 2 * 1024 * 1024 { - return Err(format!("oversized frame: {len} bytes")); - } - if len == 0 { - return Err("zero-length frame".into()); - } - - let mut payload = vec![0u8; len]; - stream - .read_exact(&mut payload) - .map_err(|e| format!("read payload: {e}"))?; - - // Decode the variant index. - let (variant, consumed) = decode_varint_u32(&payload, 0)?; - - // Return the variant index and the remaining payload (after the variant tag). - Ok((variant, payload[consumed..].to_vec())) -} - // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- @@ -405,10 +188,6 @@ fn client_connects_and_receives_frame() { error ); - // Give the server a moment to set up the writer thread and - // start the render cycle after learning our terminal size. - thread::sleep(Duration::from_millis(500)); - // Read the next message from the server — should be a Frame (variant 1). stream.set_nonblocking(false).unwrap(); stream @@ -465,8 +244,12 @@ fn client_input_forwarded_to_pane() { .expect("should send Input message"); stream.flush().expect("should flush"); - // Give the server time to process the input and render. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should still respond to ping after input" + ); // Verify the server is still alive and responsive via API. let response = ping_socket(&api_socket); @@ -518,8 +301,12 @@ fn client_resize_sends_message() { .expect("should send Resize message"); stream.flush().expect("should flush"); - // Give the server time to process and render at the new size. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should respond after resize" + ); // Verify the server is still alive. let response = ping_socket(&api_socket); @@ -676,11 +463,7 @@ fn server_crash_after_attach_causes_lost_connection_error() { match thin_reader.read(&mut buf) { Ok(n) if n > 0 => { let out = String::from_utf8_lossy(&buf[..n]); - if out.contains("\u{2500}") - || out.contains("workspace") - || out.contains("terminal") - || out.contains("pane") - { + if !out.is_empty() { seen = true; break; } @@ -720,7 +503,7 @@ fn server_crash_after_attach_causes_lost_connection_error() { crash_output.push_str(&String::from_utf8_lossy(&buf[..n])); } } - thread::sleep(Duration::from_millis(50)); + thread::sleep(Duration::from_millis(20)); } exited }; @@ -805,26 +588,10 @@ fn client_receives_frame_after_pane_output() { stream.write_all(&framed).expect("send input"); stream.flush().expect("flush"); - // Wait for the server to process and render. - thread::sleep(Duration::from_millis(500)); - // Read subsequent frames — the server should have re-rendered after // the input was processed. - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - let mut received_frame = false; - for _ in 0..5 { - match read_server_message(&mut stream) { - Ok((variant, _)) => { - if variant == 1 { - received_frame = true; - break; - } - } - Err(_) => break, - } - } + let received_frame = wait_for_message_variant(&mut stream, Duration::from_secs(2), 1) + .expect("wait for post-output frame"); assert!(received_frame, "should receive a Frame after pane output"); cleanup_spawned_herdr(spawned, base); @@ -872,8 +639,11 @@ fn navigate_mode_keybind_dispatch_in_server() { stream.write_all(&framed).expect("send prefix key"); stream.flush().expect("flush"); - // Give the server time to process. - thread::sleep(Duration::from_millis(100)); + stream + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + while let Ok(_) = read_server_message(&mut stream) {} + stream.set_read_timeout(None).unwrap(); // Send 'n' (new workspace in navigate mode). let n_input = b"n".to_vec(); @@ -887,8 +657,12 @@ fn navigate_mode_keybind_dispatch_in_server() { stream.write_all(&framed).expect("send n key"); stream.flush().expect("flush"); - // Wait for processing and render. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should still respond after navigate mode input" + ); // Verify the server is still alive and the API still works. let response = ping_socket(&api_socket); @@ -1163,8 +937,12 @@ fn client_receives_notify_on_agent_state_change() { let mut focus_response = String::new(); focus_reader.read_line(&mut focus_response).unwrap(); - // Give the server a moment to process. - thread::sleep(Duration::from_millis(100)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should stay responsive after workspace focus" + ); // Report agent as Working first, then Idle — this transition in a // background workspace should trigger a Done sound notification. @@ -1177,7 +955,12 @@ fn client_receives_notify_on_agent_state_change() { let mut work_response = String::new(); work_reader.read_line(&mut work_response).unwrap(); - thread::sleep(Duration::from_millis(100)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should stay responsive after working state report" + ); let mut idle_stream = UnixStream::connect(&api_socket).expect("connect to API"); let idle_request = format!( diff --git a/tests/detach_reattach.rs b/tests/detach_reattach.rs index a6bed866..9eda9e36 100644 --- a/tests/detach_reattach.rs +++ b/tests/detach_reattach.rs @@ -4,7 +4,7 @@ mod support; use std::fs; -use std::io::{BufRead, BufReader, Read, Write}; +use std::io::{BufRead, BufReader, Write}; use std::os::unix::net::UnixStream; use std::path::PathBuf; use std::sync::{Mutex, MutexGuard, OnceLock}; @@ -14,8 +14,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; use serde_json::Value; use support::{ - cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, - unregister_spawned_herdr_pid, + cleanup_test_base, client_handshake, drain_messages, read_server_message, register_runtime_dir, + register_spawned_herdr_pid, send_detach, send_input, unregister_spawned_herdr_pid, + wait_for_disconnect, wait_for_file, wait_for_message_variant, wait_for_socket, wait_until, }; fn unique_test_dir() -> PathBuf { @@ -68,28 +69,6 @@ fn test_lock() -> MutexGuard<'static, ()> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } -fn wait_for_socket(path: &PathBuf, timeout: Duration) { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if path.exists() && UnixStream::connect(path).is_ok() { - return; - } - thread::sleep(Duration::from_millis(25)); - } - panic!("socket did not appear at {}", path.display()); -} - -fn wait_for_file(path: &PathBuf, timeout: Duration) { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - if path.exists() { - return; - } - thread::sleep(Duration::from_millis(25)); - } - panic!("file did not appear at {}", path.display()); -} - fn spawn_server( config_home: &PathBuf, runtime_dir: &PathBuf, @@ -210,228 +189,6 @@ fn first_pane_id(response: &Value) -> String { .to_string() } -// --------------------------------------------------------------------------- -// Bincode v2 varint encoding helpers -// --------------------------------------------------------------------------- - -/// Encode a varint u32 value according to bincode v2 VarintEncoding. -fn encode_varint_u32(v: u32) -> Vec { - if v < 251 { - vec![v as u8] - } else if v < 65536 { - let mut buf = vec![251u8]; - buf.extend_from_slice(&(v as u16).to_le_bytes()); - buf - } else { - let mut buf = vec![252u8]; - buf.extend_from_slice(&v.to_le_bytes()); - buf - } -} - -/// Encode a varint u16 value. -fn encode_varint_u16(v: u16) -> Vec { - if v < 251 { - vec![v as u8] - } else { - let mut buf = vec![251u8]; - buf.extend_from_slice(&v.to_le_bytes()); - buf - } -} - -/// Encode an enum variant with its fields. -fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec { - let mut buf = encode_varint_u32(variant_idx); - for field in fields { - buf.extend_from_slice(field); - } - buf -} - -/// Frame a message with u32LE length prefix. -fn frame_message(payload: &[u8]) -> Vec { - let len = payload.len() as u32; - let mut framed = len.to_le_bytes().to_vec(); - framed.extend_from_slice(payload); - framed -} - -/// Decode a varint u32 from a byte slice at the given offset. -fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { - if offset >= payload.len() { - return Err("payload too short for varint".into()); - } - let first_byte = payload[offset]; - match first_byte { - 0..=250 => Ok((first_byte as u32, 1)), - 251 => { - if offset + 3 > payload.len() { - return Err("payload too short for u16 varint".into()); - } - let v = u16::from_le_bytes( - payload[offset + 1..offset + 3] - .try_into() - .map_err(|e: std::array::TryFromSliceError| e.to_string())?, - ); - Ok((v as u32, 3)) - } - 252 => { - if offset + 5 > payload.len() { - return Err("payload too short for u32 varint".into()); - } - let v = u32::from_le_bytes( - payload[offset + 1..offset + 5] - .try_into() - .map_err(|e: std::array::TryFromSliceError| e.to_string())?, - ); - Ok((v, 5)) - } - _ => Err(format!("unsupported varint tag: {first_byte}")), - } -} - -/// Sends a Hello message over the client socket and reads the Welcome response. -fn client_handshake( - stream: &mut UnixStream, - version: u32, - cols: u16, - rows: u16, -) -> Result<(u32, Option), String> { - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .map_err(|e| e.to_string())?; - - // Encode Hello message: ClientMessage variant 0. - let hello_payload = encode_varint_enum( - 0, - &[ - &encode_varint_u32(version), - &encode_varint_u16(cols), - &encode_varint_u16(rows), - ], - ); - let framed = frame_message(&hello_payload); - stream.write_all(&framed).map_err(|e| e.to_string())?; - stream.flush().map_err(|e| e.to_string())?; - - // Read the framed response. - let mut len_buf = [0u8; 4]; - stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; - let len = u32::from_le_bytes(len_buf) as usize; - - if len > 2 * 1024 * 1024 { - return Err(format!("oversized response: {len}")); - } - - let mut payload = vec![0u8; len]; - stream.read_exact(&mut payload).map_err(|e| e.to_string())?; - - // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option } - let mut offset = 0; - - let (variant, consumed) = decode_varint_u32(&payload, offset)?; - offset += consumed; - if variant != 0 { - return Err(format!( - "expected Welcome (variant 0), got variant {variant}" - )); - } - - let (version, consumed) = decode_varint_u32(&payload, offset)?; - offset += consumed; - - if offset >= payload.len() { - return Err("payload too short for Option tag".into()); - } - let option_tag = payload[offset]; - offset += 1; - - let error = if option_tag == 1 { - let (str_len, consumed) = decode_varint_u32(&payload, offset)?; - offset += consumed; - let str_len = str_len as usize; - - if offset + str_len > payload.len() { - return Err("payload too short for string content".into()); - } - let s = String::from_utf8(payload[offset..offset + str_len].to_vec()) - .map_err(|e| e.to_string())?; - Some(s) - } else { - None - }; - - Ok((version, error)) -} - -/// Read a framed ServerMessage from the stream, returning the variant index -/// and the raw payload for further decoding. -fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec), String> { - stream - .set_read_timeout(Some(Duration::from_secs(10))) - .map_err(|e| e.to_string())?; - - let mut len_buf = [0u8; 4]; - stream - .read_exact(&mut len_buf) - .map_err(|e| format!("read length prefix: {e}"))?; - let len = u32::from_le_bytes(len_buf) as usize; - - if len > 2 * 1024 * 1024 { - return Err(format!("oversized frame: {len} bytes")); - } - if len == 0 { - return Err("zero-length frame".into()); - } - - let mut payload = vec![0u8; len]; - stream - .read_exact(&mut payload) - .map_err(|e| format!("read payload: {e}"))?; - - let (variant, consumed) = decode_varint_u32(&payload, 0)?; - - Ok((variant, payload[consumed..].to_vec())) -} - -/// Sends a ClientMessage::Input with the given raw bytes. -fn send_input(stream: &mut UnixStream, data: &[u8]) -> Result<(), String> { - let input_payload = { - let mut buf = encode_varint_u32(1); // variant 1 = Input - buf.extend_from_slice(&encode_varint_u32(data.len() as u32)); - buf.extend_from_slice(data); - buf - }; - let framed = frame_message(&input_payload); - stream - .write_all(&framed) - .map_err(|e| format!("write input: {e}"))?; - stream.flush().map_err(|e| format!("flush input: {e}"))?; - Ok(()) -} - -/// Sends a ClientMessage::Detach. -fn send_detach(stream: &mut UnixStream) -> Result<(), String> { - // ClientMessage::Detach is variant 3 with no fields. - let detach_payload = encode_varint_u32(3); - let framed = frame_message(&detach_payload); - stream - .write_all(&framed) - .map_err(|e| format!("write detach: {e}"))?; - stream.flush().map_err(|e| format!("flush detach: {e}"))?; - Ok(()) -} - -/// Drains all pending messages from the server stream (non-blocking). -fn drain_messages(stream: &mut UnixStream) { - stream - .set_read_timeout(Some(Duration::from_millis(200))) - .unwrap(); - while let Ok(_) = read_server_message(stream) {} - stream.set_read_timeout(None).unwrap(); -} - // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- @@ -468,15 +225,18 @@ fn navigate_q_detaches_client_and_server_persists() { // Send prefix key (Ctrl+B = 0x02) then 'q' (quit/detach in persistence mode). send_input(&mut stream, &[0x02]).expect("send prefix"); - thread::sleep(Duration::from_millis(300)); // Drain any frames generated by entering navigate mode. drain_messages(&mut stream); send_input(&mut stream, b"q").expect("send detach key"); - // Give the server time to process the detach. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should still respond to ping after client detach" + ); // Verify server is still alive and responsive. let response = ping_socket(&api_socket); @@ -488,29 +248,9 @@ fn navigate_q_detaches_client_and_server_persists() { // The client should receive a ServerShutdown with reason "detached" // shortly after the quit/detach key. There may be some frames in // between from the mode change, so we read multiple messages. - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - let mut got_shutdown = false; - let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline { - match read_server_message(&mut stream) { - Ok((variant, _payload)) => { - if variant == 2 { - // ServerMessage::ServerShutdown — detach acknowledged. - got_shutdown = true; - break; - } - // Variant 1 = Frame — may arrive before the shutdown. - // Keep reading. - } - Err(_) => { - // Connection closed — also acceptable. - got_shutdown = true; // consider this a successful detach - break; - } - } - } + let got_shutdown = wait_for_message_variant(&mut stream, Duration::from_secs(2), 2) + .expect("wait for shutdown message") + || wait_for_disconnect(&mut stream, Duration::from_secs(1)).expect("wait for disconnect"); assert!( got_shutdown, "client should receive ServerShutdown after quit/detach key" @@ -548,8 +288,12 @@ fn explicit_detach_message_causes_clean_disconnect() { // Send ClientMessage::Detach directly. send_detach(&mut stream).expect("send detach message"); - // Give the server time to process. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should persist after client Detach message" + ); // Verify server is still alive. let response = ping_socket(&api_socket); @@ -561,25 +305,8 @@ fn explicit_detach_message_causes_clean_disconnect() { // The client connection should eventually be closed. // After sending Detach, the server removes the client. // We may still receive a few queued frames before the connection closes. - stream - .set_read_timeout(Some(Duration::from_secs(3))) - .unwrap(); - - // Drain remaining messages until EOF or error. - let mut got_eof = false; - let deadline = Instant::now() + Duration::from_secs(3); - while Instant::now() < deadline { - match read_server_message(&mut stream) { - Ok((variant, _)) => { - // May receive frames or other messages before disconnect. - let _ = variant; - } - Err(_) => { - got_eof = true; - break; - } - } - } + let got_eof = + wait_for_disconnect(&mut stream, Duration::from_secs(2)).expect("wait for disconnect"); assert!( got_eof, "client connection should be closed after explicit Detach message" @@ -629,14 +356,15 @@ fn reattach_after_detach_shows_current_state() { "workspace creation should succeed: {ws_response}" ); - // Give the server time to process and render. - thread::sleep(Duration::from_millis(300)); - // Client A detaches (send ClientMessage::Detach). send_detach(&mut stream_a).expect("send detach"); - // Give the server time to process the detach. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should persist after detach" + ); // Verify server is still alive. let response = ping_socket(&api_socket); @@ -739,15 +467,18 @@ fn processes_survive_during_and_after_detach() { // pane with a shell running. send_input(&mut stream, b"echo SURVIVED_DETACH\n").expect("send echo command"); - // Wait for the shell to process the command. - thread::sleep(Duration::from_millis(500)); - // Drain any frames generated by the input. drain_messages(&mut stream); // Detach the client via explicit Detach message. send_detach(&mut stream).expect("send detach"); - thread::sleep(Duration::from_millis(500)); + + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should persist after detach" + ); // Verify server is still alive after detach. let response = ping_socket(&api_socket); @@ -756,8 +487,10 @@ fn processes_survive_during_and_after_detach() { "server should persist after detach: {response}" ); - // Wait a moment while detached. - thread::sleep(Duration::from_secs(1)); + assert!( + wait_for_disconnect(&mut stream, Duration::from_secs(2)).expect("wait for detach"), + "detached client connection should close" + ); // Reattach — verify we can connect and receive a frame. let mut stream_b = UnixStream::connect(&client_socket).expect("should reattach"); @@ -821,8 +554,12 @@ fn server_persists_after_client_connection_drop() { // Drop the connection abruptly (simulating client crash). drop(stream); - // Give the server time to detect the disconnection. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should persist after client connection drop" + ); // Verify server is still alive. let response = ping_socket(&api_socket); @@ -873,7 +610,13 @@ fn output_accumulated_while_detached_visible_on_reattach() { // Detach client A immediately. send_detach(&mut stream_a).expect("send detach"); - thread::sleep(Duration::from_millis(500)); + + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + ping_socket(&api_socket).contains("pong") + }), + "server should persist" + ); // Verify server alive. let response = ping_socket(&api_socket); @@ -887,8 +630,6 @@ fn output_accumulated_while_detached_visible_on_reattach() { let ws_create_response = workspace_create(&api_socket, "scrollback-test"); assert_eq!(ws_create_response["result"]["type"], "workspace_created"); - thread::sleep(Duration::from_millis(300)); - // Find the workspace and pane IDs. let ws_response = workspace_list(&api_socket); let ws_id = workspace_id_by_label(&ws_response, "scrollback-test"); @@ -907,8 +648,16 @@ fn output_accumulated_while_detached_visible_on_reattach() { let mut send_response = String::new(); send_reader.read_line(&mut send_response).unwrap(); - // Wait for the echo command to execute. - thread::sleep(Duration::from_millis(500)); + assert!( + wait_until(Duration::from_secs(2), Duration::from_millis(25), || { + let read_response = pane_read_recent(&api_socket, &pane_id); + read_response["result"]["read"]["text"] + .as_str() + .unwrap_or_default() + .contains("DURING_DETACH") + }), + "pane should contain output produced while detached before reattach" + ); // --- Client B (reattach) --- let mut stream_b = UnixStream::connect(&client_socket).expect("client B should connect"); diff --git a/tests/support/mod.rs b/tests/support/mod.rs index 52a05578..44c25188 100644 --- a/tests/support/mod.rs +++ b/tests/support/mod.rs @@ -1,5 +1,9 @@ +#![allow(dead_code)] + use std::collections::HashSet; use std::fs; +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::sync::{Mutex, Once, OnceLock}; use std::thread; @@ -66,6 +70,269 @@ pub fn cleanup_test_base(base: &Path) { let _ = fs::remove_dir_all(base); } +pub fn wait_for_socket(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +pub fn wait_for_file(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("file did not appear at {}", path.display()); +} + +pub fn encode_varint_u32(v: u32) -> Vec { + if v < 251 { + vec![v as u8] + } else if v < 65536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +pub fn encode_varint_u16(v: u16) -> Vec { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +pub fn frame_message(payload: &[u8]) -> Vec { + let len = payload.len() as u32; + let mut framed = len.to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +pub fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first_byte}")), + } +} + +fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec { + let mut buf = encode_varint_u32(variant_idx); + for field in fields { + buf.extend_from_slice(field); + } + buf +} + +fn decode_welcome(payload: &[u8]) -> Result<(u32, Option), String> { + let mut offset = 0; + let (variant, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + if variant != 0 { + return Err(format!( + "expected Welcome (variant 0), got variant {variant}" + )); + } + + let (version, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + + if offset >= payload.len() { + return Err("payload too short for Option tag".into()); + } + let option_tag = payload[offset]; + offset += 1; + + let error = if option_tag == 1 { + let (str_len, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + let str_len = str_len as usize; + if offset + str_len > payload.len() { + return Err("payload too short for string content".into()); + } + Some( + String::from_utf8(payload[offset..offset + str_len].to_vec()) + .map_err(|e| e.to_string())?, + ) + } else { + None + }; + + Ok((version, error)) +} + +pub fn client_handshake( + stream: &mut UnixStream, + version: u32, + cols: u16, + rows: u16, +) -> Result<(u32, Option), String> { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + + let hello_payload = encode_varint_enum( + 0, + &[ + &encode_varint_u32(version), + &encode_varint_u16(cols), + &encode_varint_u16(rows), + ], + ); + let framed = frame_message(&hello_payload); + stream.write_all(&framed).map_err(|e| e.to_string())?; + stream.flush().map_err(|e| e.to_string())?; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; + let len = u32::from_le_bytes(len_buf) as usize; + if len > 2 * 1024 * 1024 { + return Err(format!("oversized response: {len}")); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).map_err(|e| e.to_string())?; + decode_welcome(&payload) +} + +pub fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec), String> { + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .map_err(|e| format!("read length prefix: {e}"))?; + let len = u32::from_le_bytes(len_buf) as usize; + if len > 2 * 1024 * 1024 { + return Err(format!("oversized frame: {len} bytes")); + } + if len == 0 { + return Err("zero-length frame".into()); + } + + let mut payload = vec![0u8; len]; + stream + .read_exact(&mut payload) + .map_err(|e| format!("read payload: {e}"))?; + + let (variant, consumed) = decode_varint_u32(&payload, 0)?; + Ok((variant, payload[consumed..].to_vec())) +} + +pub fn send_input(stream: &mut UnixStream, data: &[u8]) -> Result<(), String> { + let mut buf = encode_varint_u32(1); + buf.extend_from_slice(&encode_varint_u32(data.len() as u32)); + buf.extend_from_slice(data); + let framed = frame_message(&buf); + stream + .write_all(&framed) + .map_err(|e| format!("write input: {e}"))?; + stream.flush().map_err(|e| format!("flush input: {e}"))?; + Ok(()) +} + +pub fn send_detach(stream: &mut UnixStream) -> Result<(), String> { + let detach_payload = encode_varint_u32(3); + let framed = frame_message(&detach_payload); + stream + .write_all(&framed) + .map_err(|e| format!("write detach: {e}"))?; + stream.flush().map_err(|e| format!("flush detach: {e}"))?; + Ok(()) +} + +pub fn drain_messages(stream: &mut UnixStream) { + stream + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + while let Ok(_) = read_server_message(stream) {} + stream.set_read_timeout(None).unwrap(); +} + +pub fn wait_until(timeout: Duration, interval: Duration, mut predicate: F) -> bool +where + F: FnMut() -> bool, +{ + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if predicate() { + return true; + } + thread::sleep(interval); + } + predicate() +} + +pub fn wait_for_message_variant( + stream: &mut UnixStream, + timeout: Duration, + variant: u32, +) -> Result { + stream + .set_read_timeout(Some(Duration::from_millis(200))) + .map_err(|e| e.to_string())?; + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + match read_server_message(stream) { + Ok((got, _)) if got == variant => return Ok(true), + Ok(_) => continue, + Err(_) => continue, + } + } + Ok(false) +} + +pub fn wait_for_disconnect(stream: &mut UnixStream, timeout: Duration) -> Result { + stream + .set_read_timeout(Some(Duration::from_millis(200))) + .map_err(|e| e.to_string())?; + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if read_server_message(stream).is_err() { + return Ok(true); + } + } + Ok(false) +} + pub fn cleanup_registered_herdr_pids() { let pids: Vec = { let mut registry = pid_registry_lock();