test: cover transfer-helper scan/repair/CLI and agent terminal, reaper, fsevent paths

dory-core/agent/src/reaper.rs

@@ -47,3 +47,54 @@ fn reap_available_children() {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn the_child_wait_lock_is_shared_and_exclusive() {
+        let guard = managed_child_wait_guard().await;
+        // A background reaper sweep only runs while nothing holds the managed wait, so the lock the
+        // sweep polls must be the very lock exec waits take.
+        assert!(child_wait_lock().try_lock().is_err());
+        assert!(tokio::time::timeout(
+            std::time::Duration::from_millis(100),
+            managed_child_wait_guard()
+        )
+        .await
+        .is_err());
+
+        drop(guard);
+        let reacquired = tokio::time::timeout(
+            std::time::Duration::from_secs(20),
+            managed_child_wait_guard(),
+        )
+        .await;
+        assert!(reacquired.is_ok(), "the released lock must be reacquirable");
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn the_pid1_reaper_does_not_start_outside_pid_1() {
+        assert_ne!(unsafe { libc::getpid() }, 1, "tests do not run as PID 1");
+        // Starting a `waitpid(-1, ...)` sweep in a non-init process would steal exit statuses from
+        // the managed exec waits, so the entry point must be a no-op here: an unrelated child stays
+        // waitable by its own spawner.
+        start_pid1_reaper_if_needed();
+
+        let child = std::process::Command::new("/bin/sh")
+            .args(["-c", "exit 3"])
+            .spawn()
+            .expect("spawn child");
+        let pid = child.id() as libc::pid_t;
+        std::mem::forget(child);
+        std::thread::sleep(std::time::Duration::from_millis(400));
+
+        let mut status = 0;
+        let waited = unsafe { libc::waitpid(pid, &mut status, 0) };
+        assert_eq!(waited, pid, "the child was reaped by something else");
+        assert!(libc::WIFEXITED(status));
+        assert_eq!(libc::WEXITSTATUS(status), 3);
+    }
+}

dory-core/agent/src/terminal.rs

@@ -227,6 +227,9 @@ fn terminate_child(pid: libc::pid_t) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tokio::net::UnixStream;
+
+    const STREAM_TIMEOUT: Duration = Duration::from_secs(20);
 
     #[test]
     fn selects_existing_shell() {
@@ -237,4 +240,142 @@ mod tests {
             let _ = libc::waitpid(process.pid, &mut status, 0);
         }
     }
+
+    #[test]
+    fn the_child_shell_receives_the_injected_term() {
+        let shell = spawn_shell().expect("spawn shell");
+        set_nonblocking(shell.master.as_raw_fd()).expect("nonblocking master");
+        write_blocking(
+            shell.master.as_raw_fd(),
+            b"printf 'term=%s\\n' \"$TERM\"\nexit\n",
+        );
+
+        let transcript = read_until(shell.master.as_raw_fd(), "term=xterm-256color");
+        terminate_child_blocking(shell.pid);
+        assert!(transcript.contains("term=xterm-256color"), "{transcript}");
+    }
+
+    #[tokio::test]
+    async fn the_stream_carries_shell_output_and_ends_when_the_shell_exits() {
+        let (mut host, guest) = UnixStream::pair().expect("socket pair");
+        let served = tokio::spawn(serve_shell_stream(guest));
+
+        host.write_all(b"printf 'dory-%s\\n' ready\nexit\n")
+            .await
+            .expect("write command");
+        host.flush().await.expect("flush command");
+
+        let mut transcript = Vec::new();
+        let mut buffer = [0_u8; 4096];
+        let read = tokio::time::timeout(STREAM_TIMEOUT, async {
+            loop {
+                let count = host.read(&mut buffer).await?;
+                if count == 0 {
+                    return Ok::<(), io::Error>(());
+                }
+                transcript.extend_from_slice(&buffer[..count]);
+                if String::from_utf8_lossy(&transcript).contains("dory-ready") {
+                    return Ok(());
+                }
+            }
+        })
+        .await;
+        assert!(read.is_ok(), "timed out reading the shell transcript");
+        assert!(
+            String::from_utf8_lossy(&transcript).contains("dory-ready"),
+            "{}",
+            String::from_utf8_lossy(&transcript)
+        );
+
+        // The shell exited, so the PTY read side reports EOF and both copy loops unwind.
+        let outcome = tokio::time::timeout(STREAM_TIMEOUT, served).await;
+        assert!(outcome.expect("shell stream ended").is_ok());
+    }
+
+    #[tokio::test]
+    async fn closing_the_client_stream_tears_the_session_down() {
+        let (host, guest) = UnixStream::pair().expect("socket pair");
+        let served = tokio::spawn(serve_shell_stream(guest));
+        drop(host);
+
+        let outcome = tokio::time::timeout(STREAM_TIMEOUT, served).await;
+        assert!(outcome.expect("shell stream ended").is_ok());
+    }
+
+    #[tokio::test]
+    async fn pty_reads_and_writes_go_through_the_readiness_guards() {
+        let shell = spawn_shell().expect("spawn shell");
+        set_nonblocking(shell.master.as_raw_fd()).expect("nonblocking master");
+        let pid = shell.pid;
+        let pty = AsyncFd::new(shell.master).expect("register pty");
+
+        write_all_fd(&pty, b"").await.expect("empty write");
+        write_all_fd(&pty, b"printf 'guarded'\n")
+            .await
+            .expect("write command");
+        let mut buffer = [0_u8; 4096];
+        let count = tokio::time::timeout(STREAM_TIMEOUT, read_fd(&pty, &mut buffer))
+            .await
+            .expect("read did not time out")
+            .expect("read");
+        assert!(count > 0);
+        terminate_child_blocking(pid);
+    }
+
+    #[test]
+    fn raw_descriptor_helpers_surface_their_errors() {
+        assert_eq!(dup_fd(-1).unwrap_err().raw_os_error(), Some(libc::EBADF));
+        assert_eq!(
+            set_nonblocking(-1).unwrap_err().raw_os_error(),
+            Some(libc::EBADF)
+        );
+        assert_eq!(
+            read_raw(-1, &mut [0_u8; 1]).unwrap_err().raw_os_error(),
+            Some(libc::EBADF)
+        );
+        assert_eq!(
+            write_raw(-1, b"x").unwrap_err().raw_os_error(),
+            Some(libc::EBADF)
+        );
+    }
+
+    fn write_blocking(fd: RawFd, mut bytes: &[u8]) {
+        while !bytes.is_empty() {
+            match write_raw(fd, bytes) {
+                Ok(count) => bytes = &bytes[count..],
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    std::thread::sleep(Duration::from_millis(20))
+                }
+                Err(error) => panic!("pty write failed: {error}"),
+            }
+        }
+    }
+
+    fn read_until(fd: RawFd, marker: &str) -> String {
+        let deadline = std::time::Instant::now() + STREAM_TIMEOUT;
+        let mut output = Vec::new();
+        let mut buffer = [0_u8; 4096];
+        while std::time::Instant::now() < deadline {
+            match read_raw(fd, &mut buffer) {
+                Ok(0) => break,
+                Ok(count) => output.extend_from_slice(&buffer[..count]),
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    std::thread::sleep(Duration::from_millis(20))
+                }
+                Err(_) => break,
+            }
+            if String::from_utf8_lossy(&output).contains(marker) {
+                break;
+            }
+        }
+        String::from_utf8_lossy(&output).into_owned()
+    }
+
+    fn terminate_child_blocking(pid: libc::pid_t) {
+        unsafe {
+            libc::kill(pid, libc::SIGKILL);
+            let mut status = 0;
+            let _ = libc::waitpid(pid, &mut status, 0);
+        }
+    }
 }

dory-core/agent/src/vsock_server.rs

@@ -112,10 +112,13 @@ async fn serve_fsevents() -> std::io::Result<()> {
     }
 }
 
-async fn handle_fsevent_batch(
-    mut stream: tokio_vsock::VsockStream,
+async fn handle_fsevent_batch<S>(
+    mut stream: S,
     dedupe: Arc<crate::fsevents::FSEventDedupeStore>,
-) -> std::io::Result<()> {
+) -> std::io::Result<()>
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
     // One deadline covers the complete frame so a peer cannot slow-drip a permitted connection.
     let body = timeout(FSEVENT_READ_TIMEOUT, async {
         let body_len = stream.read_u32_le().await? as usize;
@@ -241,10 +244,140 @@ async fn serve_shell() -> std::io::Result<()> {
 mod tests {
     use super::*;
 
+    use std::path::PathBuf;
+
+    use crate::fsevents::{encode_batch_frame, ResponseStatus, MAX_FRAME_BYTES, PROTOCOL_VERSION};
+
+    struct Response {
+        operation_id: u64,
+        path_count: u32,
+        status: u32,
+        failed_indices: Vec<u32>,
+    }
+
+    fn decode_response(frame: &[u8]) -> Response {
+        let body_len = u32::from_le_bytes(frame[0..4].try_into().unwrap()) as usize;
+        assert_eq!(frame.len(), 4 + body_len);
+        assert_eq!(
+            u32::from_le_bytes(frame[4..8].try_into().unwrap()),
+            PROTOCOL_VERSION
+        );
+        let failed_count = u32::from_le_bytes(frame[24..28].try_into().unwrap()) as usize;
+        let failed_indices = (0..failed_count)
+            .map(|index| {
+                let start = 28 + index * 4;
+                u32::from_le_bytes(frame[start..start + 4].try_into().unwrap())
+            })
+            .collect();
+        Response {
+            operation_id: u64::from_le_bytes(frame[8..16].try_into().unwrap()),
+            path_count: u32::from_le_bytes(frame[16..20].try_into().unwrap()),
+            status: u32::from_le_bytes(frame[20..24].try_into().unwrap()),
+            failed_indices,
+        }
+    }
+
+    /// Feeds `request` to the fsevent handler and returns the handler result plus everything the
+    /// handler wrote back to the peer.
+    async fn exchange(
+        request: &[u8],
+        dedupe: Arc<crate::fsevents::FSEventDedupeStore>,
+    ) -> (std::io::Result<()>, Vec<u8>) {
+        let (mut peer, guest) = tokio::io::duplex(256 * 1024);
+        peer.write_all(request).await.expect("write request");
+        let result = handle_fsevent_batch(guest, dedupe).await;
+        let mut response = Vec::new();
+        peer.read_to_end(&mut response)
+            .await
+            .expect("read response");
+        (result, response)
+    }
+
     #[test]
     fn host_only_listener_policy_accepts_host_and_rejects_guest_cids() {
         assert!(is_host_peer(&VsockAddr::new(VMADDR_CID_HOST, PORT_CONTROL)));
         assert!(!is_host_peer(&VsockAddr::new(3, PORT_CONTROL)));
         assert!(!is_host_peer(&VsockAddr::new(VMADDR_CID_ANY, PORT_CONTROL)));
     }
+
+    #[tokio::test]
+    async fn a_valid_batch_is_nudged_and_answered_with_per_path_results() {
+        let directory = std::env::temp_dir().join(format!("dory-fsevents-{}", std::process::id()));
+        std::fs::create_dir_all(&directory).expect("create batch directory");
+        let existing = directory.join("present");
+        std::fs::write(&existing, b"payload").expect("write batch file");
+        // Nudging walks up to an existing ancestor, so only a path whose every ancestor is missing
+        // (the filesystem root is never nudged) can fail.
+        let unreachable = PathBuf::from("/dory-absent-fsevent-root/child");
+        let paths = vec![existing.clone(), unreachable];
+
+        let request = encode_batch_frame(7, &paths).expect("encode batch");
+        let (result, response) = exchange(&request, Arc::default()).await;
+
+        result.expect("handled batch");
+        let decoded = decode_response(&response);
+        assert_eq!(decoded.operation_id, 7);
+        assert_eq!(decoded.path_count, 2);
+        assert_eq!(decoded.status, ResponseStatus::Success as u32);
+        // Only the path whose parent directory does not exist can fail to be nudged.
+        assert_eq!(decoded.failed_indices, vec![1]);
+
+        std::fs::remove_dir_all(&directory).expect("clean batch directory");
+    }
+
+    #[tokio::test]
+    async fn reusing_an_operation_id_for_different_paths_is_reported_as_a_conflict() {
+        let dedupe = Arc::new(crate::fsevents::FSEventDedupeStore::default());
+        let first = encode_batch_frame(11, &[PathBuf::from("/tmp")]).expect("encode batch");
+        let second = encode_batch_frame(11, &[PathBuf::from("/var")]).expect("encode batch");
+
+        let (first_result, _) = exchange(&first, Arc::clone(&dedupe)).await;
+        first_result.expect("handled first batch");
+        let (second_result, response) = exchange(&second, dedupe).await;
+
+        second_result.expect("handled second batch");
+        let decoded = decode_response(&response);
+        assert_eq!(decoded.operation_id, 11);
+        assert_eq!(decoded.path_count, 1);
+        assert_eq!(
+            decoded.status,
+            ResponseStatus::ConflictingOperationId as u32
+        );
+        assert!(decoded.failed_indices.is_empty());
+    }
+
+    #[tokio::test]
+    async fn oversized_and_undecodable_frames_are_rejected_without_a_response() {
+        let mut oversized = ((MAX_FRAME_BYTES + 1) as u32).to_le_bytes().to_vec();
+        oversized.extend_from_slice(&[0_u8; 16]);
+        let (result, response) = exchange(&oversized, Arc::default()).await;
+        assert_eq!(
+            result.unwrap_err().kind(),
+            std::io::ErrorKind::InvalidData,
+            "declared body length above the frame cap must be refused"
+        );
+        assert!(response.is_empty());
+
+        let mut wrong_version = 16_u32.to_le_bytes().to_vec();
+        wrong_version.extend_from_slice(&(PROTOCOL_VERSION + 1).to_le_bytes());
+        wrong_version.extend_from_slice(&1_u64.to_le_bytes());
+        wrong_version.extend_from_slice(&0_u32.to_le_bytes());
+        let (result, response) = exchange(&wrong_version, Arc::default()).await;
+        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::InvalidData);
+        assert!(response.is_empty());
+    }
+
+    #[tokio::test]
+    async fn a_slow_drip_frame_hits_the_read_deadline() {
+        // The length prefix promises a body that never arrives, which is exactly the stall the
+        // per-connection deadline exists to bound.
+        let (peer, guest) = tokio::io::duplex(1024);
+        let mut peer = peer;
+        peer.write_all(&64_u32.to_le_bytes())
+            .await
+            .expect("write length prefix");
+        let result = handle_fsevent_batch(guest, Arc::default()).await;
+        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut);
+        drop(peer);
+    }
 }