test(runtime-core): bind stream relay publisher test ports in the publisher

packages/runtime-core/guest/stream-relay.c

@@ -228,8 +228,22 @@ static int stop_owned(long pid, long start) {
     return 78;
 #endif
 }
+/*
+ * Publisher readiness contract: after the frontend and optional control socket
+ * are bound, the publisher writes exactly one readiness line to stdout and keeps
+ * serving. A Unix frontend or an explicit TCP port (1-65535) writes "ready\n",
+ * unchanged for existing callers. `--publish 0` (exactly "0") asks the kernel for
+ * an ephemeral loopback port while the listener is bound, closing the window in
+ * which a caller-probed free port can be taken by another process; that form
+ * writes "ready <port>\n" with the bound decimal port. Port 0 is accepted only by
+ * `--publish`; the relay and `--publish-unix` forms keep rejecting it with 64.
+ * Exit 73 still means the frontend could not be bound (or a bound Unix frontend
+ * path could not be verified), and no readiness line is written for any refused
+ * startup.
+ */
 int main(int argc, char **argv) {
     char *publish_args[8];
+    int ephemeral_port = 0;
     struct sockaddr_un control = {0};
     struct stat control_owned, receipt_owned, frontend_receipt_owned;
     char receipt_path[PATH_MAX] = {0};
@@ -263,7 +277,9 @@ int main(int argc, char **argv) {
         }
         /* Both publishers share upstream parsing; the Unix frontend has no TCP port. */
         publish_args[0] = argv[0]; publish_args[1] = argv[3];
-        publish_args[2] = "127.0.0.1"; publish_args[3] = publish_unix ? "1" : argv[2];
+        ephemeral_port = !publish_unix && !strcmp(argv[2], "0");
+        publish_args[2] = "127.0.0.1";
+        publish_args[3] = publish_unix || ephemeral_port ? "1" : argv[2];
         publish_args[4] = argv[5]; publish_args[5] = "--reservation";
         publish_args[6] = argv[4]; publish_args[7] = NULL;
         argv = publish_args; argc = 7; publish_mode = 1;
@@ -297,7 +313,7 @@ int main(int argc, char **argv) {
     }
     if (argc != option) return 64;
     target.sin_family = AF_INET;
-    target.sin_port = htons((uint16_t)port);
+    target.sin_port = ephemeral_port ? 0 : htons((uint16_t)port);
     local.sun_family = AF_UNIX;
     memcpy(local.sun_path, argv[1], strlen(argv[1]) + 1);
     if (publish_mode) { upstream = local; if (private_upstream(&current)) return 78; }
@@ -338,7 +354,12 @@ int main(int argc, char **argv) {
                     receipt_path, &receipt_owned, &receipt_bound)) failed = 1;
         }
     }
-    if (!failed) { puts("ready"); fflush(stdout); }
+    struct sockaddr_in bound = {0};
+    socklen_t bound_size = sizeof(bound);
+    if (!failed && ephemeral_port && (getsockname(listener, (struct sockaddr *)&bound, &bound_size) ||
+            bound_size != sizeof(bound) || bound.sin_family != AF_INET || !bound.sin_port)) failed = 1;
+    if (!failed && ephemeral_port) { printf("ready %u\n", (unsigned)ntohs(bound.sin_port)); fflush(stdout); }
+    else if (!failed) { puts("ready"); fflush(stdout); }
     while (!stopping && !failed) {
         struct pollfd descriptors[4 + 2 * CONNECTIONS];
         descriptors[3 + 2 * CONNECTIONS] = (struct pollfd){control_fd, POLLIN, 0};

packages/runtime-core/tests/stream_relay_contract.rs

@@ -23,6 +23,8 @@ struct Relay {
     child: Child,
     root: PathBuf,
     socket: PathBuf,
+    /// The TCP port a `--publish` publisher reported or was given; None otherwise.
+    port: Option<u16>,
 }
 impl Relay {
     fn start(port: u16, idle: u32) -> Self {
@@ -93,6 +95,7 @@ impl Relay {
             child,
             root,
             socket,
+            port: None,
         }
     }
     fn connect(&self) -> UnixStream {
@@ -576,9 +579,6 @@ fn malformed_reservations_refuse_before_socket_creation() {
     fs::remove_dir_all(root).unwrap();
 }
 
-fn publisher(socket: &std::path::Path, token: &str, port: u16) -> Relay {
-    controlled_publisher(socket, token, port, None)
-}
 fn controlled_publisher(
     socket: &std::path::Path,
     token: &str,
@@ -676,9 +676,15 @@ fn try_frontend_publisher(
     startup.child = Some(command.stdout(Stdio::piped()).spawn().unwrap());
     let mut output = startup.child.as_mut().unwrap().stdout.take().unwrap();
     let deadline = Instant::now() + Duration::from_secs(3);
-    let expected = b"ready\n";
+    // Port 0 asks a TCP publisher to bind an ephemeral port and report it as
+    // "ready <port>\n"; every other form keeps the bare "ready\n" line.
+    let ephemeral = port == 0 && !unix;
+    let mut line = Vec::with_capacity(READINESS_LIMIT);
     // Read one byte per readiness notification: no unbounded read_line after POLLIN.
-    for expected_byte in expected {
+    while line.last() != Some(&b'\n') {
+        if line.len() == READINESS_LIMIT {
+            return Err(startup.failure("readiness marker"));
+        }
         loop {
             let remaining = deadline.saturating_duration_since(Instant::now());
             if remaining.is_zero() {
@@ -705,21 +711,57 @@ fn try_frontend_publisher(
             }
             let mut byte = [0];
             match output.read(&mut byte) {
-                Ok(1) if byte[0] == *expected_byte => break,
+                Ok(1) => {
+                    line.push(byte[0]);
+                    break;
+                }
                 Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
                 Ok(0) => return Err(startup.failure("readiness EOF")),
                 _ => return Err(startup.failure("readiness marker")),
             }
         }
     }
+    let published = match (ephemeral, line.as_slice()) {
+        (false, b"ready\n") => (!unix).then_some(port),
+        (true, reported) => match parse_reported_port(reported) {
+            Some(reported) => Some(reported),
+            None => return Err(startup.failure("readiness marker")),
+        },
+        _ => return Err(startup.failure("readiness marker")),
+    };
     let root = std::mem::take(&mut startup.root);
     Ok(Relay {
         child: startup.child.take().unwrap(),
         socket: root.join(if unix { "frontend.sock" } else { "unused" }),
         root,
+        port: published,
     })
 }
 
+/// "ready 65535\n" is the longest valid readiness line.
+const READINESS_LIMIT: usize = 12;
+
+/// Parses "ready <port>\n" with a canonical, nonzero decimal port.
+fn parse_reported_port(line: &[u8]) -> Option<u16> {
+    let digits = line.strip_prefix(b"ready ")?.strip_suffix(b"\n")?;
+    if digits.is_empty() || digits[0] == b'0' || !digits.iter().all(u8::is_ascii_digit) {
+        return None;
+    }
+    std::str::from_utf8(digits).ok()?.parse::<u16>().ok()
+}
+
+/// Starts a TCP publisher on a kernel-chosen port that stays bound from the
+/// publisher's bind() onward, so parallel tests cannot take it first.
+fn ephemeral_publisher(
+    socket: &std::path::Path,
+    token: &str,
+    control_token: Option<&str>,
+) -> (Relay, u16) {
+    let publication = controlled_publisher(socket, token, 0, control_token);
+    let port = publication.port.expect("TCP publication reports its port");
+    (publication, port)
+}
+
 #[test]
 fn publisher_startup_reports_port_conflict_and_reaps_child() {
     let backend = TcpListener::bind("127.0.0.1:0").unwrap();
@@ -758,12 +800,103 @@ fn publisher_startup_reports_port_conflict_and_reaps_child() {
     assert!(upstream.socket.exists());
 }
 
-fn free_loopback_port() -> u16 {
-    TcpListener::bind("127.0.0.1:0")
-        .unwrap()
-        .local_addr()
+/// Runs a command that must refuse promptly and returns its exit code and stdout.
+/// A child that is still running after the deadline was admitted instead of
+/// refused; it is killed and reaped so the test fails rather than hangs.
+fn bounded_refusal(command: &mut Command) -> (Option<i32>, Vec<u8>) {
+    let mut child = command
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .spawn()
+        .unwrap();
+    let deadline = Instant::now() + Duration::from_secs(3);
+    let status = loop {
+        if let Some(status) = child.try_wait().unwrap() {
+            break status;
+        }
+        if Instant::now() >= deadline {
+            let _ = child.kill();
+            let _ = child.wait();
+            panic!("command was admitted instead of refused: {command:?}");
+        }
+        thread::sleep(Duration::from_millis(5));
+    };
+    let mut stdout = Vec::new();
+    child
+        .stdout
+        .take()
         .unwrap()
-        .port()
+        .read_to_end(&mut stdout)
+        .unwrap();
+    (status.code(), stdout)
+}
+
+#[test]
+fn ephemeral_publication_reports_its_bound_port_and_explicit_ports_keep_bare_ready() {
+    let backend = TcpListener::bind("127.0.0.1:0").unwrap();
+    let token = "dddddddddddddddddddddddddddddddd";
+    let guard = Relay::start_options(
+        backend.local_addr().unwrap().port(),
+        10000,
+        None,
+        Some(token),
+        None,
+    );
+    // Only the exact `--publish 0` form selects an ephemeral port.
+    for (mode, endpoint) in [("--publish", "00"), ("--publish-unix", "0")] {
+        let (code, stdout) = bounded_refusal(
+            Command::new(BINARY)
+                .args([mode, endpoint])
+                .arg(&guard.socket)
+                .args([token, "10000"]),
+        );
+        assert_eq!(code, Some(64), "{mode} {endpoint}");
+        assert!(stdout.is_empty());
+    }
+    let (code, stdout) = bounded_refusal(
+        Command::new(BINARY)
+            .arg(guard.root.join("relay-port-zero.sock"))
+            .args(["127.0.0.1", "0", "1000"]),
+    );
+    assert_eq!(code, Some(64));
+    assert!(stdout.is_empty());
+    assert!(!guard.root.join("relay-port-zero.sock").exists());
+
+    let (mut first, first_port) = ephemeral_publisher(&guard.socket, token, None);
+    let (mut second, second_port) = ephemeral_publisher(&guard.socket, token, None);
+    assert_ne!(first_port, second_port);
+    for port in [first_port, second_port] {
+        // The reported port is the publisher's live listener, not a released probe.
+        assert_eq!(
+            TcpListener::bind(("127.0.0.1", port)).unwrap_err().kind(),
+            io::ErrorKind::AddrInUse
+        );
+        std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
+    }
+    first.stop();
+    second.stop();
+
+    // Explicit ports keep the bare "ready\n" line for existing callers. Picking a
+    // concrete port needs a released probe, so retry only on the publisher's
+    // confirmed bind refusal (no readiness line, exit 73) and never otherwise.
+    let mut attempts = 0;
+    let explicit = loop {
+        attempts += 1;
+        let probe = TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = probe.local_addr().unwrap().port();
+        drop(probe);
+        match try_frontend_publisher(&guard.socket, token, port, None, false) {
+            Ok(publication) => break publication,
+            Err(failure)
+                if attempts < 3
+                    && failure.stage == "readiness EOF"
+                    && failure.status.and_then(|status| status.code()) == Some(73) => {}
+            Err(failure) => panic!("explicit-port publisher refused: {failure:?}"),
+        }
+    };
+    let port = explicit.port.expect("explicit TCP port recorded");
+    std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
+    assert!(guard.socket.exists());
 }
 
 #[test]
@@ -777,8 +910,7 @@ fn loopback_publication_streams_without_exposing_handshake_and_releases_port() {
         Some(token),
         None,
     );
-    let port = free_loopback_port();
-    let mut publication = publisher(&guard.socket, token, port);
+    let (mut publication, port) = ephemeral_publisher(&guard.socket, token, None);
     let payload = vec![0x5a; 2 * 1024 * 1024];
     let expected = payload.clone();
     let worker = thread::spawn(move || {
@@ -835,8 +967,7 @@ fn publisher_refuses_port_conflicts_and_stale_reused_upstream() {
         Some(73)
     );
     assert!(std::net::TcpStream::connect(("127.0.0.1", busy_port)).is_ok());
-    let port = free_loopback_port();
-    let _publication = publisher(&guard.socket, old, port);
+    let (_publication, port) = ephemeral_publisher(&guard.socket, old, None);
     guard.stop();
     let _replacement = Relay::start_options(
         backend.local_addr().unwrap().port(),
@@ -915,17 +1046,27 @@ fn publisher_refuses_nonprivate_or_aliased_upstream_before_listening() {
     fs::create_dir(&root).unwrap();
     let path = root.join("socket");
     let _listener = UnixListener::bind(&path).unwrap();
-    let port = free_loopback_port();
+    // Hold the requested port for the whole test: a publisher that reached bind()
+    // would exit 73 on it, so exit 78 proves the refusal precedes listening without
+    // probing a released port that a parallel test could have taken.
+    let occupied = TcpListener::bind("127.0.0.1:0").unwrap();
+    occupied.set_nonblocking(true).unwrap();
+    let port = occupied.local_addr().unwrap().port();
     let attempt = |upstream: &std::path::Path| {
         let result = Command::new(BINARY)
             .arg("--publish")
             .arg(port.to_string())
             .arg(upstream)
             .args(["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "10000"])
-            .status()
+            .stdin(Stdio::null())
+            .output()
             .unwrap();
-        assert_eq!(result.code(), Some(78));
-        assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_err());
+        assert_eq!(result.status.code(), Some(78));
+        assert!(result.stdout.is_empty());
+        assert_eq!(
+            occupied.accept().unwrap_err().kind(),
+            std::io::ErrorKind::WouldBlock
+        );
     };
     fs::set_permissions(&path, fs::Permissions::from_mode(0o777)).unwrap();
     attempt(&path);
@@ -955,8 +1096,7 @@ fn private_control_stops_only_matching_publisher_and_preserves_replaced_path() {
     fs::set_permissions(&upstream, fs::Permissions::from_mode(0o700)).unwrap();
     let token = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
     for replace in [false, true] {
-        let port = free_loopback_port();
-        let mut publication = controlled_publisher(&upstream, token, port, Some(token));
+        let (mut publication, port) = ephemeral_publisher(&upstream, token, Some(token));
         let path = publication.root.join("control.sock");
         assert_eq!(
             fs::metadata(&path).unwrap().permissions().mode() & 0o777,
@@ -1030,44 +1170,34 @@ fn control_startup_refuses_occupied_and_nonprivate_paths_without_retaining_publi
     fs::write(&control, b"foreign").unwrap();
     let control_identity = fs::symlink_metadata(&control).unwrap();
     let frontend = root.join("frontend");
-    // The TCP leg publishes on a port that free_loopback_port() has already
-    // released, so a parallel test may re-take it before the relay binds and the
-    // relay then exits 73 at bind instead of reaching the control path. Retry a
-    // bounded number of times on a fresh port; the 70/78 expectations stay exact.
-    const PORT_ATTEMPTS: u32 = 3;
     for unix in [false, true] {
         for (mode, expected_exit) in [(0o700, 70), (0o755, 78)] {
             fs::set_permissions(&root, fs::Permissions::from_mode(mode)).unwrap();
-            let mut attempt = 1;
-            let (pid, output) = loop {
-                let mut command = Command::new(BINARY);
-                command
-                    .arg(if unix { "--publish-unix" } else { "--publish" })
-                    .arg(if unix {
-                        frontend.as_os_str().to_owned()
-                    } else {
-                        free_loopback_port().to_string().into()
-                    })
-                    .arg(&upstream)
-                    .args(["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "10000", "--control"])
-                    .arg(&control)
-                    .arg("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
-                    .stdin(Stdio::null())
-                    .stdout(Stdio::piped())
-                    .stderr(Stdio::piped());
-                let child = command.spawn().unwrap();
-                let pid = child.id();
-                let output = child.wait_with_output().unwrap();
-                if !unix && output.status.code() == Some(73) && attempt < PORT_ATTEMPTS {
-                    attempt += 1;
-                    continue;
-                }
-                break (pid, output);
-            };
+            let mut command = Command::new(BINARY);
+            command
+                .arg(if unix { "--publish-unix" } else { "--publish" })
+                // TCP port 0 binds an ephemeral port inside the publisher, so the TCP
+                // leg always reaches the control path instead of racing a parallel
+                // test for a released port and exiting 73 at bind.
+                .arg(if unix {
+                    frontend.as_os_str().to_owned()
+                } else {
+                    "0".into()
+                })
+                .arg(&upstream)
+                .args(["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "10000", "--control"])
+                .arg(&control)
+                .arg("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+                .stdin(Stdio::null())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped());
+            let child = command.spawn().unwrap();
+            let pid = child.id();
+            let output = child.wait_with_output().unwrap();
             assert_eq!(
                 output.status.code(),
                 Some(expected_exit),
-                "unix={unix} mode={mode:o} attempt={attempt}"
+                "unix={unix} mode={mode:o}"
             );
             assert!(output.stdout.is_empty());
             let mut status = 0;