test: reduce persistence integration test sleeps and harness duplication

1 files changed, 61 insertions, 312 deletions

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,
@@ -211,228 +190,6 @@ fn first_pane_id(response: &Value) -> 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");