xiaoxue_admin
/
herdr
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test: reduce persistence integration test sleeps and harness duplication

This commit is contained in:
Ogulcan Celik 2026-04-22 21:23:18 +03:00
parent e9a3e3784c
commit 3d6f3e7862
3 changed files with 371 additions and 572 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

303
tests/client_mode.rs
View File

@ -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<u8> {
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<u8> {
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<u8> {
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<u8> {
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>), 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<String> }
decode_welcome(&payload)
}
/// Decode a ServerMessage::Welcome from bincode v2 payload.
fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), 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<String> — 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<u8>), 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!(

373
tests/detach_reattach.rs
View File

@ -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<u8> {
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<u8> {
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<u8> {
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<u8> {
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>), 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<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());
}
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<u8>), 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");

267
tests/support/mod.rs
View File

@ -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<u8> {
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<u8> {
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<u8> {
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<u8> {
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>), 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>), 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<u8>), 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<F>(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<bool, String> {
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<bool, String> {
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<u32> = {
let mut registry = pid_registry_lock();