test: wait for each Windows milestone to be seen before the next

crates/vt_bin/src/vtt/stalled_remote_cache.rs

@@ -13,10 +13,6 @@ use std::{
 /// forwarded or answered: each emits a "stalled" milestone when it arrives,
 /// and its connection is held until the client closes it. Other requests are
 /// forwarded, each on its own connection. Ctrl-C is left to the command.
-///
-/// On Windows a milestone is the console title, which `ConPTY` sends when it
-/// next renders, so if the command sets one at about the same time, the
-/// earlier title can be lost.
 pub fn run(args: &[String]) -> Result<(), Box<dyn std::error::Error>> {
     let mut args = args;
     let mut stalled_routes = Vec::new();

crates/vt_bin/tests/e2e_snapshots/fixtures/remote_cache/snapshots.toml

@@ -341,10 +341,6 @@ steps = [
 
 [[e2e]]
 name = "pending_uploads"
-# Skipped on Windows, where a milestone is the console title and ConPTY sends
-# only the latest one when it next renders: the proxy's `stalled` milestone for
-# check's upload can replace `uploads-pending`.
-cfg = "not(windows)"
 ignore = true
 steps = [
   { argv = [
@@ -615,9 +611,6 @@ steps = [
 
 [[e2e]]
 name = "hide_pending_uploads"
-# Skipped on Windows, where a milestone is the console title and ConPTY sends
-# it only when it next renders: nothing is printed after `uploads-pending`.
-cfg = "not(windows)"
 ignore = true
 steps = [
   { argv = [

crates/vt_pty_terminal_test/Cargo.toml

@@ -13,6 +13,9 @@ portable-pty = { workspace = true }
 vt_pty_terminal = { workspace = true }
 vt_pty_terminal_test_client = { workspace = true }
 
+[target.'cfg(windows)'.dependencies]
+winapi = { workspace = true, features = ["handleapi", "synchapi", "winnt"] }
+
 [dev-dependencies]
 crossterm = { workspace = true }
 ctor = { workspace = true }

crates/vt_pty_terminal_test/README.md

@@ -53,18 +53,29 @@ vt-pty-terminal-test:<32-hex-random-id>:<name-base64url>
 
 `Reader::expect_milestone` works like this:
 
-1. Drain title events captured by `PtyReader`.
+1. Drain title events captured by `PtyReader`, acknowledging each milestone.
 2. Ignore ordinary titles and completed non-target milestones.
 3. If no match exists, continue reading from the PTY and repeat.
 4. Return the current screen once the requested title is observed.
 
+`Reader::wait_for_exit` also acknowledges the milestones it reads.
+
 ## Cross-platform behavior
 
 On Windows the client calls `SetConsoleTitleW`; ConPTY emits the resulting title
 through its asynchronous renderer after preceding text and cursor state. On Unix
 the client emits an OSC 2 title update, which follows normal PTY byte ordering.
 The same token decoder and test API are used on every platform.
 
+ConPTY emits only the latest title when it renders, so a title set by another
+process before then would replace a milestone. `TestTerminal::spawn` sets
+`VT_PTY_TERMINAL_TEST_MILESTONE_LOCK` for the child on Windows. A producer
+that inherits it holds a named lock shared by the processes in the terminal
+from setting its title until the reader acknowledges the milestone by
+signaling a named event derived from the milestone's random id. A process that
+doesn't inherit it, such as a task that `vp` runs with a filtered environment,
+only sets the title.
+
 ## Typical test pattern
 
 In the child process:

crates/vt_pty_terminal_test/src/lib.rs

@@ -1,4 +1,4 @@
-use std::io::{BufReader, Read};
+use std::io::{BufReader, Read as _};
 
 pub use portable_pty::CommandBuilder;
 use vt_pty_terminal::terminal::{PtyReader, Terminal};
@@ -32,6 +32,21 @@ impl TestTerminal {
     ///
     /// Returns an error if the PTY cannot be opened or the command fails to spawn.
     pub fn spawn(size: ScreenSize, cmd: CommandBuilder) -> anyhow::Result<Self> {
+        // Producers on Windows wait for each milestone to be seen while
+        // holding a lock shared by the processes in this terminal.
+        #[cfg(windows)]
+        let cmd = {
+            use std::sync::atomic::{AtomicU64, Ordering};
+
+            static NEXT_TERMINAL: AtomicU64 = AtomicU64::new(0);
+            let mut cmd = cmd;
+            let terminal = NEXT_TERMINAL.fetch_add(1, Ordering::Relaxed);
+            cmd.env(
+                vt_pty_terminal_test_client::MILESTONE_LOCK_ENV,
+                format!("{}-{terminal}", std::process::id()),
+            );
+            cmd
+        };
         let Terminal { pty_reader, pty_writer, child_handle, .. } = Terminal::spawn(size, cmd)?;
         Ok(Self {
             writer: pty_writer,
@@ -61,6 +76,8 @@ impl Reader {
     ///
     /// Milestones use a uniform title token across platforms.
     ///
+    /// Every milestone read before the named one is skipped.
+    ///
     /// # Panics
     ///
     /// Panics if the child process exits (EOF) before the named milestone is received,
@@ -70,19 +87,30 @@ impl Reader {
         let mut buf = [0u8; 4096];
 
         loop {
-            while let Some(title) = self.pty.get_ref().take_window_title() {
-                if vt_pty_terminal_test_client::decode_milestone_title(&title)
-                    .is_some_and(|milestone| milestone == name)
-                {
-                    return self.screen_contents();
-                }
+            if self.take_milestones().any(|milestone| milestone == name) {
+                return self.screen_contents();
             }
 
             let n = self.pty.read(&mut buf).expect("PTY read failed");
             assert!(n > 0, "EOF reached before milestone '{name}'");
         }
     }
 
+    /// Takes the names of the milestones read so far, letting each producer
+    /// continue once its milestone is taken.
+    fn take_milestones(&self) -> impl Iterator<Item = String> {
+        std::iter::from_fn(|| {
+            loop {
+                let title = self.pty.get_ref().take_window_title()?;
+                if let Some(milestone) = vt_pty_terminal_test_client::decode_milestone_title(&title)
+                {
+                    acknowledge(milestone.id);
+                    return Some(milestone.name);
+                }
+            }
+        })
+    }
+
     /// Reads all remaining PTY output until the child exits, then returns the exit status.
     ///
     /// # Errors
@@ -93,8 +121,41 @@ impl Reader {
     ///
     /// Panics if reading from the PTY fails.
     pub fn wait_for_exit(&mut self) -> anyhow::Result<ExitStatus> {
-        let mut discard = Vec::new();
-        self.pty.read_to_end(&mut discard).expect("PTY read_to_end failed");
+        let mut buf = [0u8; 4096];
+        loop {
+            self.take_milestones().for_each(drop);
+            if self.pty.read(&mut buf).expect("PTY read failed") == 0 {
+                break;
+            }
+        }
         self.child_handle.wait()
     }
 }
+
+/// Signals the producer of milestone `id` that the milestone was seen.
+///
+/// See `vt_pty_terminal_test_client::mark_milestone` for why it waits.
+#[cfg(windows)]
+fn acknowledge(id: u128) {
+    use winapi::um::{handleapi, synchapi, winnt};
+
+    let name = vt_pty_terminal_test_client::milestone_ack_event_name(id)
+        .encode_utf16()
+        .chain([0])
+        .collect::<Vec<_>>();
+    // SAFETY: `name` is a valid NUL-terminated UTF-16 name.
+    let event = unsafe { synchapi::OpenEventW(winnt::EVENT_MODIFY_STATE, 0, name.as_ptr()) };
+    // The producer doesn't wait if it didn't inherit the lock environment
+    // variable, and closes the event once it is signaled.
+    if event.is_null() {
+        return;
+    }
+    // SAFETY: `event` is a valid event handle owned here.
+    unsafe {
+        synchapi::SetEvent(event);
+        handleapi::CloseHandle(event);
+    }
+}
+
+#[cfg(not(windows))]
+const fn acknowledge(_id: u128) {}

crates/vt_pty_terminal_test/tests/milestone.rs

@@ -152,3 +152,60 @@ fn descendant_process_can_mark_milestone() {
     let _ = reader.expect_milestone("nested");
     assert!(reader.wait_for_exit().unwrap().success());
 }
+
+/// Two processes mark milestones at the same time. `ConPTY` emits only the
+/// latest console title when it renders, so without the lock in
+/// `mark_milestone` the nested process's titles could replace `target`.
+#[test]
+#[timeout(5000)]
+#[expect(clippy::redundant_clone, reason = "command_for_fn evaluates its argument twice")]
+fn concurrent_milestones_are_not_lost() {
+    let nested = command_for_fn!((), |(): ()| {
+        for _ in 0..20 {
+            vt_pty_terminal_test_client::mark_milestone("noise");
+        }
+    });
+    let nested_program = nested.program.to_string_lossy().into_owned();
+    let nested_args =
+        nested.args.iter().map(|arg| arg.to_string_lossy().into_owned()).collect::<Vec<_>>();
+
+    let cmd = CommandBuilder::from(command_for_fn!(
+        (nested_program.clone(), nested_args.clone()),
+        |(nested_program, nested_args): (String, Vec<String>)| {
+            let mut nested =
+                std::process::Command::new(nested_program).args(nested_args).spawn().unwrap();
+            vt_pty_terminal_test_client::mark_milestone("target");
+            assert!(nested.wait().unwrap().success());
+        }
+    ));
+
+    let TestTerminal { writer: _, mut reader, child_handle: _ } =
+        TestTerminal::spawn(ScreenSize { rows: 80, cols: 80 }, cmd).unwrap();
+    let _ = reader.expect_milestone("target");
+    assert!(reader.wait_for_exit().unwrap().success());
+}
+
+/// The child prints nothing after the milestone until the harness responds.
+#[test]
+#[timeout(5000)]
+fn milestone_without_later_output() {
+    let cmd = CommandBuilder::from(command_for_fn!((), |(): ()| {
+        use std::io::{Write, stdout};
+
+        let mut stdout = stdout();
+        write!(stdout, "waiting").unwrap();
+        stdout.flush().unwrap();
+        vt_pty_terminal_test_client::mark_milestone("waiting");
+
+        let mut line = String::new();
+        std::io::stdin().read_line(&mut line).unwrap();
+    }));
+
+    let TestTerminal { mut writer, mut reader, child_handle: _ } =
+        TestTerminal::spawn(ScreenSize { rows: 80, cols: 80 }, cmd).unwrap();
+    let screen = reader.expect_milestone("waiting");
+    assert_eq!(screen.trim(), "waiting");
+
+    writer.write_line(b"").unwrap();
+    assert!(reader.wait_for_exit().unwrap().success());
+}

crates/vt_pty_terminal_test_client/Cargo.toml

@@ -16,7 +16,7 @@ base64 = { workspace = true }
 getrandom = { workspace = true, optional = true }
 
 [target.'cfg(windows)'.dependencies]
-winapi = { workspace = true, features = ["wincon"], optional = true }
+winapi = { workspace = true, features = ["synchapi", "winbase", "wincon"], optional = true }
 
 [dev-dependencies]
 getrandom = { workspace = true }