diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs index 8ce6a6fc..1f2b3588 100644 --- a/apps/nesinit/src/lib.rs +++ b/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; diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs index 2952761f..aa157c6a 100644 --- a/apps/nesinit/src/main.rs +++ b/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 outcome?, + outcome = session::run(channel, workload, &mut ports, &mut found_rx) => outcome?, signal = asked_to_stop() => { signal?; tracing::info!("asked to stop"); diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 78e0e28d..8490f499 100644 --- a/apps/nesinit/src/session.rs +++ b/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( channel: C, workload: &mut W, payload: &mut Ports, + addresses: &mut Receiver, ) -> std::io::Result 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( channel: C, workload: &mut W, payload: &mut Ports, + addresses: &mut Receiver, ) -> std::io::Result where C: AsyncRead + AsyncWrite, @@ -91,6 +94,7 @@ where let mut running: Option = 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), Ended(Exit), FromWorkload(Option), + Address(Option), } /// 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 { + 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>, @@ -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 { .. })); diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs new file mode 100644 index 00000000..2d98fba5 --- /dev/null +++ b/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) { + let mut sent: Option = None; + loop { + match read(&path).await { + Ok(current) if Some(¤t) != 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 { + 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>) { + 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); + } +}