feat(nesinit): carry the session's address out of the guest

Whatever serves media in a box knows how it can be reached, and the person who
needs to know is not in the box. Standard output here is a log file inside a
VM, so the control channel is the delivery path rather than a convenience --
which makes this init's job and not a detail of whichever component happens to
bind the port. The socket it reads was already documented as being read this
way; nothing read it.

Polled rather than read once, because an address is not a value but the best
answer so far. An endpoint discovers more ways to reach it after it binds, so
the first answer is the one that works on a local network and fails from
anywhere else. Only a changed answer is forwarded.

It dials rather than listens, which is the opposite of the relay next door and
deliberate: there the guest listens because the workload starts later, and here
the server is the long-lived one. Dialling also makes a server that has not
bound yet something to retry rather than something to wait for without knowing
whether it is coming.

The address itself is never logged. It is a capability to reach the session,
and a log inside the guest is the one place it has no reason to be.

A carrier that stops does not end a session: whatever was already reported is
still correct, and the workload's exit still has to be.

apps/nesinit/src/lib.rs

@@ -12,4 +12,5 @@ pub mod payload;
 pub mod reap;
 pub mod session;
 pub mod shutdown;
+pub mod ticket;
 pub mod workload;

apps/nesinit/src/main.rs

@@ -4,13 +4,14 @@
 // the workload the channel describes, and turn the end of either into an
 // ordered shutdown.
 
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use nesinit::payload::{self, Ports};
 use nesinit::reap::{self, Waiters};
 use nesinit::session::{self, Outcome};
 use nesinit::shutdown::{self, Machine};
+use nesinit::ticket;
 use nesinit::workload::{Process, Workload};
 use nesprotocol::lifecycle::CONTROL_PORT;
 use tokio::signal::unix::{SignalKind, signal};
@@ -25,6 +26,13 @@ const GRACE: Duration = Duration::from_secs(10);
 /// deep queue holds stale copies of it rather than protecting anything.
 const RELAY_DEPTH: usize = 8;
 
+/// How many addresses may be waiting to be forwarded.
+///
+/// Two, because only the newest one matters: an address is superseded by the
+/// next one rather than added to, so a deeper queue holds stale copies of it
+/// and delays the one that is current.
+const ADDRESS_DEPTH: usize = 2;
+
 fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt()
         .with_env_filter(
@@ -97,8 +105,14 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result<Outc
         from_workload: up_rx,
     };
 
+    // Started before the workload, like the relay, and for the same reason:
+    // whatever serves the address may bind the moment it comes up, and nothing
+    // here should be the reason a session waits to be reachable.
+    let (found_tx, mut found_rx) = tokio::sync::mpsc::channel(ADDRESS_DEPTH);
+    tokio::spawn(ticket::carry(PathBuf::from(ticket::SOCKET), found_tx));
+
     let outcome = tokio::select! {
-        outcome = session::run(channel, workload, &mut ports) => outcome?,
+        outcome = session::run(channel, workload, &mut ports, &mut found_rx) => outcome?,
         signal = asked_to_stop() => {
             signal?;
             tracing::info!("asked to stop");

apps/nesinit/src/session.rs

@@ -14,6 +14,7 @@ use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader
 
 use crate::payload::Ports;
 use crate::workload::{Exited, Failure, Workload};
+use tokio::sync::mpsc::Receiver;
 
 /// How a session ended.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -39,12 +40,13 @@ pub async fn run<C, W>(
     channel: C,
     workload: &mut W,
     payload: &mut Ports,
+    addresses: &mut Receiver<String>,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
     W: Workload,
 {
-    match converse(channel, workload, payload).await {
+    match converse(channel, workload, payload, addresses).await {
         Err(error) if channel_gone(&error) => {
             // A caller that has stopped reading has also stopped being able to
             // tell us to stop, which is the same situation as the channel
@@ -70,6 +72,7 @@ async fn converse<C, W>(
     channel: C,
     workload: &mut W,
     payload: &mut Ports,
+    addresses: &mut Receiver<String>,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
@@ -91,6 +94,7 @@ where
 
     let mut running: Option<Exited> = None;
     let mut relay_open = true;
+    let mut carrier_open = true;
 
     loop {
         let event = match running.as_mut() {
@@ -98,10 +102,12 @@ where
                 ended = exited => Event::Ended(ended?),
                 line = lines.next_line() => Event::Line(line?),
                 up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up),
+                found = addresses.recv(), if carrier_open => Event::Address(found),
             },
             None => tokio::select! {
                 line = lines.next_line() => Event::Line(line?),
                 up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up),
+                found = addresses.recv(), if carrier_open => Event::Address(found),
             },
         };
 
@@ -115,6 +121,20 @@ where
                 send(&mut writer, &GuestToHost::Payload { payload }).await?;
                 continue;
             }
+            Event::Address(Some(ticket)) => {
+                // Sent whenever a better one is found, not only the first time:
+                // a caller that keeps the first address it is given works on a
+                // local network and fails from anywhere else.
+                send(&mut writer, &GuestToHost::Ticket { ticket }).await?;
+                continue;
+            }
+            Event::Address(None) => {
+                // Nothing will look for an address again. Not an ending: the
+                // session has whatever it was already told, and the workload
+                // still has to be stopped and its exit reported.
+                carrier_open = false;
+                continue;
+            }
             Event::FromWorkload(None) => {
                 // The relay is gone. The session is not: the workload can
                 // still be stopped, and its exit still has to be reported.
@@ -195,6 +215,7 @@ enum Event {
     Line(Option<String>),
     Ended(Exit),
     FromWorkload(Option<Payload>),
+    Address(Option<String>),
 }
 
 /// Hand an envelope to the relay, and treat a relay that is not there as the
@@ -276,6 +297,17 @@ mod tests {
         )
     }
 
+    /// A carrier that never finds an address, for the tests that are not
+    /// about one. Held open rather than closed: a closed channel is itself a
+    /// case, and it is tested on purpose below.
+    fn nowhere() -> mpsc::Receiver<String> {
+        let (tx, rx) = mpsc::channel(1);
+        // Kept alive for the process, so `recv` pends rather than resolving
+        // `None` and taking a branch these tests are not exercising.
+        Box::leak(Box::new(tx));
+        rx
+    }
+
     /// The other end of the channel, as a caller would drive it.
     struct Caller {
         lines: tokio::io::Lines<BufReader<DuplexStream>>,
@@ -322,7 +354,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -339,6 +373,87 @@ mod tests {
         assert_eq!(outcome, Outcome::Shutdown);
     }
 
+    /// The address goes up the channel as it is found. This is the only way
+    /// out: standard output here is a log file inside a VM and the person who
+    /// needs the address is outside it.
+    #[tokio::test]
+    async fn an_address_that_is_found_is_reported_to_the_caller() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (found_tx, mut found_rx) = mpsc::channel(4);
+
+        let session = tokio::spawn(async move {
+            let (mut ports, _to_workload, _from_workload) = ports();
+            let mut workload = Double::exits_when_stopped(Exit::code(0));
+            run(guest, &mut workload, &mut ports, &mut found_rx)
+                .await
+                .unwrap()
+        });
+
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ready {
+                protocol_version: 2
+            }
+        );
+
+        found_tx
+            .send("nestri:local-only".to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ticket {
+                ticket: "nestri:local-only".into()
+            }
+        );
+
+        // And a better one replaces it rather than being the caller's problem
+        // to have missed.
+        found_tx
+            .send("nestri:with-relays".to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ticket {
+                ticket: "nestri:with-relays".into()
+            }
+        );
+
+        caller.say(&HostToGuest::Shutdown).await;
+        assert_eq!(session.await.unwrap(), Outcome::Shutdown);
+    }
+
+    /// Nothing looking for an address any more is not an ending. The session
+    /// keeps whatever it was already told and still has to report an exit.
+    #[tokio::test]
+    async fn a_carrier_that_stops_does_not_end_the_session() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (found_tx, mut found_rx) = mpsc::channel(4);
+
+        let session = tokio::spawn(async move {
+            let (mut ports, _to_workload, _from_workload) = ports();
+            let mut workload = Double::exits_when_stopped(Exit::code(0));
+            run(guest, &mut workload, &mut ports, &mut found_rx)
+                .await
+                .unwrap()
+        });
+
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ready {
+                protocol_version: 2
+            }
+        );
+        drop(found_tx);
+
+        // Still answering, which is the whole assertion.
+        caller.say(&HostToGuest::Shutdown).await;
+        assert_eq!(session.await.unwrap(), Outcome::Shutdown);
+    }
+
     #[tokio::test]
     async fn the_descriptor_mounts_and_starts_what_it_names() {
         let (guest, host) = tokio::io::duplex(4096);
@@ -347,7 +462,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -373,7 +490,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(3));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -409,7 +528,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::signal(9));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -440,7 +561,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -466,7 +589,9 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.mount_failure = Some(Failure::new("EACCES: /mnt/user"));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -501,7 +626,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -524,7 +651,9 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -556,7 +685,9 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.start_failure = Some(Failure::new("ENOENT: /usr/bin/workload"));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap();
             (outcome, workload)
         });
 
@@ -598,7 +729,9 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap()
         });
         let mut to_relay = down_rx;
 
@@ -647,7 +780,9 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -693,7 +828,9 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(guest, &mut workload, &mut ports, &mut nowhere())
+                .await
+                .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));

apps/nesinit/src/ticket.rs

@@ -0,0 +1,224 @@
+// Carrying the address a client needs from inside the guest to outside it.
+//
+// Whatever serves media in the guest knows how it can be reached, and the
+// person who needs to know is not in the guest. Standard output here is a log
+// file inside a VM, so the control channel is the delivery path rather than a
+// convenience — which is why this is init's job and not a detail of whichever
+// component happens to bind the port.
+//
+// # Why it is polled rather than read once
+//
+// An address is not a value, it is the best answer so far. An endpoint
+// discovers more ways to reach it after it binds — a local one immediately, a
+// relayed or hole-punched one some seconds later — so the first answer is the
+// one that works on a local network and fails from anywhere else. Re-reading
+// and forwarding only what changed keeps that from being decided by whoever
+// asked first.
+//
+// # Why it dials rather than listens
+//
+// The opposite of the payload relay next door, and deliberately: there the
+// guest listens because the workload starts later, and here the server is the
+// long-lived one. Dialling also means a server that has not bound yet is an
+// error this retries, rather than a connection that has to be waited for
+// without knowing whether it is coming.
+
+use std::io;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
+use tokio::net::UnixStream;
+use tokio::sync::mpsc::Sender;
+
+/// Where the address is read from.
+pub const SOCKET: &str = "/tmp/nestri-ticket.sock";
+
+/// How often to look for a better address.
+///
+/// Short, because the window this closes is the first few seconds of a session
+/// and a player waiting to connect is waiting on exactly this. It costs one
+/// connection to a unix socket per interval.
+const EVERY: Duration = Duration::from_secs(2);
+
+/// The longest address this will read.
+///
+/// One line of text. A cap rather than a preference: the process on the other
+/// end can write without ever sending a newline, and the process holding the
+/// buffer is the one the kernel has been told not to kill.
+const LONGEST: u64 = 8 * 1024;
+
+/// Forward every new address for as long as the session lasts.
+///
+/// Never returns on its own. A server that is not there yet, or has gone away,
+/// is retried at the next interval — there is nothing here worth ending a
+/// running session over, and an address that stops being re-offered does not
+/// stop being correct.
+pub async fn carry(path: PathBuf, out: Sender<String>) {
+    let mut sent: Option<String> = None;
+    loop {
+        match read(&path).await {
+            Ok(current) if Some(&current) != sent.as_ref() => {
+                // The address itself is not logged. It is a capability to reach
+                // this session, and a log inside the guest is the one place it
+                // has no reason to be.
+                tracing::info!(
+                    first = sent.is_none(),
+                    "forwarding an address for this session"
+                );
+                if out.send(current.clone()).await.is_err() {
+                    // The session is over. Nothing else reads this.
+                    return;
+                }
+                sent = Some(current);
+            }
+            Ok(_) => {}
+            Err(error) => {
+                // Expected until whatever serves the address has bound, so it
+                // is not a warning the first several times. It stays at this
+                // level afterwards too: a session that already has an address
+                // is not harmed by failing to look for a better one.
+                tracing::debug!(%error, "no address available yet");
+            }
+        }
+        tokio::time::sleep(EVERY).await;
+    }
+}
+
+/// One line from the socket, which is the whole protocol.
+async fn read(path: &Path) -> io::Result<String> {
+    let stream = UnixStream::connect(path).await?;
+    let mut line = String::new();
+    BufReader::new(stream.take(LONGEST))
+        .read_line(&mut line)
+        .await?;
+    let line = line.trim().to_string();
+    if line.is_empty() {
+        return Err(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "the socket answered with nothing",
+        ));
+    }
+    Ok(line)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::AsyncWriteExt;
+    use tokio::net::UnixListener;
+    use tokio::sync::mpsc;
+
+    /// Serve a sequence of addresses, one per connection, the way a real one
+    /// does: it answers every dial with what it currently knows.
+    fn serve(path: PathBuf, answers: Vec<Option<String>>) {
+        let listener = UnixListener::bind(&path).unwrap();
+        tokio::spawn(async move {
+            let mut answers = answers.into_iter().cycle();
+            loop {
+                let Ok((mut stream, _)) = listener.accept().await else {
+                    return;
+                };
+                if let Some(answer) = answers.next().flatten() {
+                    let _ = stream.write_all(format!("{answer}\n").as_bytes()).await;
+                }
+            }
+        });
+    }
+
+    fn scratch(name: &str) -> PathBuf {
+        let dir =
+            std::env::temp_dir().join(format!("nesinit-ticket-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        dir.join("ticket.sock")
+    }
+
+    #[tokio::test]
+    async fn an_address_reaches_the_channel() {
+        let path = scratch("one");
+        serve(path.clone(), vec![Some("nestri:abc".into())]);
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx));
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("no address arrived")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:abc");
+
+        // And it is not sent again. An address that has not changed is the same
+        // address, and re-sending it is a write per interval for nothing.
+        assert!(
+            tokio::time::timeout(EVERY * 3, rx.recv()).await.is_err(),
+            "an unchanged address was forwarded again"
+        );
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// The case that decides whether this works from anywhere but a local
+    /// network: a better address arrives after the first one has been sent.
+    #[tokio::test]
+    async fn a_better_address_replaces_the_one_before_it() {
+        let path = scratch("better");
+        serve(
+            path.clone(),
+            vec![
+                Some("nestri:local-only".into()),
+                Some("nestri:with-relays".into()),
+            ],
+        );
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx));
+
+        let mut seen = Vec::new();
+        while seen.len() < 2 {
+            let next = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+                .await
+                .expect("the second address never arrived")
+                .expect("the channel closed");
+            seen.push(next);
+        }
+        assert_eq!(seen, ["nestri:local-only", "nestri:with-relays"]);
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// Nothing serving the socket yet is the ordinary case at boot, not a
+    /// failure: this starts before whatever binds it.
+    #[tokio::test]
+    async fn a_socket_that_is_not_there_yet_is_waited_out_rather_than_failed() {
+        let path = scratch("late");
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx));
+
+        tokio::time::sleep(EVERY * 2).await;
+        serve(path.clone(), vec![Some("nestri:late".into())]);
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("an address that arrived late was never picked up")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:late");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// An answer with nothing in it is not an address. Forwarding one would
+    /// publish an empty string as somewhere to connect.
+    #[tokio::test]
+    async fn an_empty_answer_is_not_an_address() {
+        let path = scratch("empty");
+        serve(path.clone(), vec![None, Some("nestri:real".into())]);
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx));
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("no address arrived")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:real");
+        let _ = std::fs::remove_file(&path);
+    }
+}