nvx-rs, runtime-adapter: run a clone's restore sequence to its sandbox's start

nvx-rs/README.md

@@ -694,6 +694,30 @@ abrupt exit leaves, and the release makes the socket's absence durable before
 the lease ends, and removes the record and `restore.sock`, as it removes
 `clone-scratch.raw`.
 
+`Restoring::activate` then runs the clone's restore sequence through the
+adapter's framing of its control session, a `CloneControl`, by a
+restore-to-Ready deadline that runs from the launch, `RESTORE_TO_READY_TIMEOUT`,
+60 seconds by default: it reads the restore-ready event as soon as it comes,
+refusing another process's connection and bytes that differ from the event, and
+reading nothing past it, since OpenVMM on Windows waits until the host has read
+it, and then closes its end; only then connects to the clone's control endpoint,
+whose console bounds the host's authentication from the connection on; attaches
+with the launch's capability; sends the record, read again from
+`clone-activation.bin`, which must still be byte for byte the launch's, in the
+adapter's Activate, as the session's first request; and checks that Ready
+reports the record's transaction ID. It replaces a session that the adapter
+lost, as `CloneControl::lost` tells, with a new one that sends the same record,
+in at most `MAX_ACTIVATION_SESSIONS` sessions by the same deadline, and retries
+nothing else.
+
+`Activated::start` then starts the sandbox by a Ready-to-Start deadline from the
+check of Ready, `READY_TO_START_TIMEOUT`, 30 seconds by default, never sending
+the start again, and `Started::commit` commits the launch, which writes its
+runtime record. A failure ends the launch as `Starting::abandon` does; a crash
+before the commit leaves the launch's journal, whose recovery ends the VMM,
+while the lease lasts until the release, and a crash after it leaves a VM that
+reattaches, and keeps its lease while it runs.
+
 `Store::tombstone` quarantines a generation, which takes no new lease while its
 leases keep working, and `Store::retire` deletes it once its leases have
 drained, crash-safely, removing the records of leases that no clone binding

nvx-rs/src/host/platform/linux.rs

@@ -857,11 +857,13 @@ pub(crate) fn control_endpoint(socket_path: &Path) -> io::Result<String> {
 
 /// The listener at which the launch of a clone awaits the restore-ready event of its VMM: a Unix
 /// socket, which the VMM connects to once, and which queues its connection until the host
-/// accepts it.
+/// accepts it, then the connection that the host accepted, from which it reads the event.
 #[derive(Debug)]
 pub(crate) struct ReadyListener {
     /// The listening socket, which the VMM finds as long as it is open.
-    _socket: UnixListener,
+    socket: UnixListener,
+    /// The VMM's connection, once the host accepted it.
+    connection: Option<UnixStream>,
 }
 
 /// Listens at the Unix socket `socket_path`, in place of a socket that an earlier launch left
@@ -877,7 +879,90 @@ pub(crate) fn listen_ready(socket_path: &Path) -> io::Result<(String, ReadyListe
         _ => {}
     }
     let socket = UnixListener::bind(socket_path)?;
-    Ok((endpoint, ReadyListener { _socket: socket }))
+    // The host polls the socket before it accepts, so an accept never blocks.
+    socket.set_nonblocking(true)?;
+    let listener = ReadyListener {
+        socket,
+        connection: None,
+    };
+    Ok((endpoint, listener))
+}
+
+impl ReadyListener {
+    /// Waits at most `timeout` for what the VMM, the process `pid`, sends to the endpoint, and
+    /// appends it to `received`, which then holds at most `limit` bytes. Returns whether more may
+    /// arrive: `false` once the VMM closed its end of the connection.
+    ///
+    /// The endpoint's first connection must be the VMM's: one of another process fails with
+    /// [`io::ErrorKind::PermissionDenied`].
+    pub(crate) fn receive(
+        &mut self,
+        pid: u32,
+        received: &mut Vec<u8>,
+        limit: usize,
+        timeout: Duration,
+    ) -> io::Result<bool> {
+        let deadline = Instant::now() + timeout;
+        if self.connection.is_none() {
+            if !sys::wait_readable(&self.socket, timeout)? {
+                return Ok(true);
+            }
+            let connection = match self.socket.accept() {
+                Ok((connection, _)) => connection,
+                Err(error)
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                    ) =>
+                {
+                    return Ok(true);
+                }
+                Err(error) => return Err(error),
+            };
+            let peer = sys::peer_pid(&connection)?;
+            if peer != pid {
+                return Err(io::Error::new(
+                    io::ErrorKind::PermissionDenied,
+                    format!(
+                        "process {peer} connected to the restore-ready endpoint, not the VMM, \
+                         process {pid}"
+                    ),
+                ));
+            }
+            connection.set_nonblocking(true)?;
+            self.connection = Some(connection);
+        }
+        let connection = self
+            .connection
+            .as_mut()
+            .expect("the host accepted the VMM's connection");
+        let mut buffer = [0; 64];
+        let room = limit.saturating_sub(received.len()).min(buffer.len());
+        if room == 0
+            || !sys::wait_readable(
+                connection,
+                deadline.saturating_duration_since(Instant::now()),
+            )?
+        {
+            return Ok(true);
+        }
+        match connection.read(&mut buffer[..room]) {
+            Ok(0) => Ok(false),
+            Ok(count) => {
+                received.extend_from_slice(&buffer[..count]);
+                Ok(true)
+            }
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                ) =>
+            {
+                Ok(true)
+            }
+            Err(error) => Err(error),
+        }
+    }
 }
 
 /// Connects to a control endpoint served by the process `expected_pid`.

nvx-rs/src/host/platform/other.rs

@@ -252,6 +252,18 @@ pub(crate) fn listen_ready(_socket_path: &Path) -> io::Result<(String, ReadyList
     Err(unsupported())
 }
 
+impl ReadyListener {
+    pub(crate) fn receive(
+        &mut self,
+        _pid: u32,
+        _received: &mut Vec<u8>,
+        _limit: usize,
+        _timeout: Duration,
+    ) -> io::Result<bool> {
+        Err(unsupported())
+    }
+}
+
 pub(crate) fn connect_endpoint(
     _endpoint: &str,
     _expected_pid: u32,