feat(nesinit): the box starts one daemon for its whole life, beside the compositor (#434)

Adds an optional `daemon` to the boot descriptor (decision 0085).
nesinit starts it after the compositor, as the given uid, announces it
in /run/nestri/daemon before the box is ready, lists it among the box's
services, and on its exit reports ServiceDied by name and ends the box
(as for the compositor). Optional both ways: an old descriptor reads the
same; an old guest refuses a new descriptor and the host already reports
that as a stale image. Tests: protocol round trip, started after the
compositor, start failure refuses the box, exit reported.

apps/nesinit/src/session.rs

@@ -160,6 +160,9 @@ where
     let mut services_open = true;
     // The box's compositor exit, once there is a compositor.
     let mut compositor_exit: Option<crate::workload::Exited> = None;
+    // The box's daemon exit and name, once there is a daemon.
+    let mut daemon_exit: Option<crate::workload::Exited> = None;
+    let mut daemon_name = String::new();
 
     loop {
         // The exits are guarded like the channels below: an empty set resolves
@@ -182,6 +185,9 @@ where
             gone = async { compositor_exit.as_mut().expect("guarded").await }, if compositor_exit.is_some() => {
                 Event::CompositorDied(gone)
             },
+            gone = async { daemon_exit.as_mut().expect("guarded").await }, if daemon_exit.is_some() => {
+                Event::DaemonDied(gone)
+            },
         };
 
         let line = match event {
@@ -298,6 +304,20 @@ where
                     "the compositor exited ({exit:?}), so the box can no longer draw"
                 ))));
             }
+            Event::DaemonDied(gone) => {
+                // What every launch shares is gone with it (the box's one
+                // Steam sign-in), and a launch that carried on would have
+                // nothing to say why its game cannot reach it. Said as a
+                // service death, then the box ends.
+                let exit = gone.unwrap_or_else(|_| Exit::code(1));
+                let name = daemon_name.clone();
+                tracing::error!(?exit, %name, "the box's daemon has exited");
+                send(&mut writer, &GuestToHost::ServiceDied { name: name.clone(), exit }).await?;
+                workload.signal_stop();
+                return Ok(Outcome::Refused(Failure::new(format!(
+                    "the {name} daemon exited ({exit:?}), so the box can no longer do what it is for"
+                ))));
+            }
             Event::ServiceDied(None) => {
                 // The supervisor is gone, which on the way down is ordinary.
                 // Nothing to report and nothing to end: a box with no service
@@ -425,10 +445,21 @@ where
                         workload.start_compositor(compositor)?;
                         up.push("nescope".to_string());
                     }
+                    // After the compositor: what it starts may draw. And the
+                    // daemon is started, not waited on: it says itself when it
+                    // works, to whoever needs it to.
+                    if let Some(daemon) = &descriptor.daemon {
+                        workload.start_daemon(daemon)?;
+                        up.push(daemon.name.clone());
+                    }
                     Ok(up)
                 }) {
                     Ok(up) => {
                         compositor_exit = workload.compositor_exit();
+                        daemon_exit = workload.daemon_exit();
+                        if let Some(daemon) = &descriptor.daemon {
+                            daemon_name = daemon.name.clone();
+                        }
                         tracing::info!(services = up.len(), "the box is ready to be launched into");
                         send(&mut writer, &GuestToHost::Initialized { services: up }).await?
                     }
@@ -670,6 +701,7 @@ enum Event {
     Address(Option<crate::ticket::Found>),
     ServiceDied(Option<Died>),
     CompositorDied(std::io::Result<Exit>),
+    DaemonDied(std::io::Result<Exit>),
 }
 
 /// Hand an envelope to the relay, and treat a relay that is not there as the
@@ -725,6 +757,7 @@ mod tests {
             images: Vec::new(),
             video: Default::default(),
             compositor: None,
+            daemon: None,
         }
     }
 
@@ -1022,6 +1055,86 @@ mod tests {
         assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}");
     }
 
+    fn daemon_given() -> nesprotocol::lifecycle::Daemon {
+        nesprotocol::lifecycle::Daemon {
+            name: "steam".into(),
+            exec: Exec {
+                argv: vec!["/payload/client".into()],
+                env: Default::default(),
+                cwd: None,
+                uid: 1001,
+                gid: 1001,
+            },
+        }
+    }
+
+    /// The box's daemon is started by the box, after the compositor, and is
+    /// listed among its services. ref(d-0085)
+    #[tokio::test]
+    async fn the_daemon_is_started_by_the_box_after_the_compositor() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0))));
+        let mut given = descriptor();
+        given.compositor = Some(nesprotocol::lifecycle::Compositor {
+            width: 1280, height: 720, fps: 30, uid: 1001, gid: 1001, env: Default::default(),
+        });
+        given.daemon = Some(daemon_given());
+        caller.expect_ready().await;
+        caller.say(&HostToGuest::Boot { descriptor: Box::new(given.clone()) }).await;
+        assert_eq!(caller.expect().await, GuestToHost::Mounted);
+        let GuestToHost::Initialized { services } = caller.expect().await else {
+            panic!("the box did not come up")
+        };
+        assert!(services.contains(&"steam".to_string()), "{services:?}");
+        caller.say(&HostToGuest::Shutdown).await;
+        let (_, workload, _) = session.await.unwrap();
+        assert_eq!(workload.daemons, vec![given.daemon.unwrap()]);
+        let order: Vec<_> = workload.order.iter().filter(|o| ["compositor", "daemon"].contains(o)).collect();
+        assert_eq!(order, vec![&"compositor", &"daemon"]);
+    }
+
+    /// A daemon that will not start refuses the box.
+    #[tokio::test]
+    async fn a_daemon_that_will_not_start_refuses_the_box() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let mut double = Double::exits_when_stopped(Exit::code(0));
+        double.daemon_failure = Some(Failure::new("no client"));
+        let session = spawn(guest, Given::new(double));
+        let mut given = descriptor();
+        given.daemon = Some(daemon_given());
+        caller.expect_ready().await;
+        caller.say(&HostToGuest::Boot { descriptor: Box::new(given) }).await;
+        assert_eq!(caller.expect().await, GuestToHost::Mounted);
+        assert_eq!(caller.expect().await, GuestToHost::InitFailed { reason: "no client".into() });
+        let (outcome, _, _) = session.await.unwrap();
+        assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}");
+    }
+
+    /// A daemon that exits is reported by name and ends the box.
+    #[tokio::test]
+    async fn a_daemon_that_exits_is_reported_and_ends_the_box() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        let mut double = Double::exits_when_stopped(Exit::code(0));
+        double.daemon_exit = Some(rx);
+        let session = spawn(guest, Given::new(double));
+        let mut given = descriptor();
+        given.daemon = Some(daemon_given());
+        caller.expect_ready().await;
+        caller.say(&HostToGuest::Boot { descriptor: Box::new(given) }).await;
+        caller.expect_booted().await;
+        tx.send(Exit::code(3)).unwrap();
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::ServiceDied { name: "steam".into(), exit: Exit::code(3) }
+        );
+        let (outcome, _, _) = session.await.unwrap();
+        assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}");
+    }
+
     #[tokio::test]
     async fn the_descriptors_video_limits_reach_the_services() {
         // The ceiling is useless if it stops at the descriptor. `neshub` is the

apps/nesinit/src/workload.rs

@@ -13,7 +13,7 @@ use std::os::unix::ffi::OsStringExt as _;
 use std::pin::Pin;
 
 use nesprotocol::lifecycle::{
-    AttachId, Compositor, Exec, Exit, Image, LAUNCH_ID_ENV, LaunchId, Mount, Overlay, Winsize,
+    AttachId, Compositor, Daemon, Exec, Exit, Image, LAUNCH_ID_ENV, LaunchId, Mount, Overlay, Winsize,
 };
 
 use std::os::unix::process::CommandExt;
@@ -99,6 +99,20 @@ pub trait Workload {
         None
     }
 
+    /// Start the box's daemon, and return once it is spawned.
+    ///
+    /// Owned by the box like the compositor, and for the same reason: every
+    /// launch shares what it holds (the box's one Steam sign-in), so no launch's
+    /// stop may reach it. Whether it is *working* is its own to say; this is
+    /// only that it started. ref(d-0085)
+    fn start_daemon(&mut self, daemon: &Daemon) -> Result<(), Failure>;
+
+    /// The daemon's exit, once, after [`Workload::start_daemon`] has returned.
+    /// `None` when there is no daemon or it was already taken.
+    fn daemon_exit(&mut self) -> Option<Exited> {
+        None
+    }
+
     /// Ask one launch to stop: its whole process group, not its top process.
     /// Idempotent, and never ends the session by itself. A launch that is not
     /// running is not signalled: its pid may already belong to something else.
@@ -129,8 +143,25 @@ pub struct Process {
     /// The compositor's exit, until the session takes it. See
     /// [`Workload::compositor_exit`].
     compositor_exit: Option<Exited>,
+    /// The box's daemon, if it has one. Like the compositor, not a launch.
+    daemon: Option<Watched>,
+    /// The daemon's exit, until the session takes it.
+    daemon_exit: Option<Exited>,
+    /// Where the daemon is announced to the box's launches: [`DAEMON_MARKER`],
+    /// except in a test.
+    daemon_marker: String,
 }
 
+/// The file that says this box has a daemon, holding its name.
+///
+/// Written before the daemon is spawned and so before the box is ready, which
+/// is before any launch can exist: a launch that finds it knows the daemon is
+/// there to be waited for, and one that does not find it knows there is none to
+/// wait for. Anything else a launch could look at (the daemon's own state) is
+/// not there yet in the first moments of a box, and a launch that read its
+/// absence as "no daemon" would do what the daemon is for by itself.
+pub const DAEMON_MARKER: &str = "/run/nestri/daemon";
+
 /// How long a launch's group has to leave after being asked before it is
 /// killed.
 pub(crate) const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
@@ -419,6 +450,9 @@ impl Process {
             cgroup_root: CGROUP_ROOT.to_string(),
             compositor: None,
             compositor_exit: None,
+            daemon: None,
+            daemon_exit: None,
+            daemon_marker: DAEMON_MARKER.to_string(),
         }
     }
 
@@ -808,6 +842,70 @@ impl Workload for Process {
         self.compositor_exit.take()
     }
 
+    fn start_daemon(&mut self, daemon: &Daemon) -> Result<(), Failure> {
+        let exec = &daemon.exec;
+        let Some((program, args)) = exec.argv.split_first() else {
+            return Err(Failure::new(format!("the {} daemon has no command", daemon.name)));
+        };
+        let (uid, gid) = (exec.uid, exec.gid);
+        let runtime = crate::system::runtime_dir(uid, gid).map_err(|error| {
+            Failure::new(format!(
+                "no runtime directory for the {} daemon (uid {uid}): {error}",
+                daemon.name
+            ))
+        })?;
+        // Before the spawn: see [`DAEMON_MARKER`]. A failure here refuses the
+        // box, because a launch that could not tell would start its own.
+        if let Some(directory) = std::path::Path::new(&self.daemon_marker).parent() {
+            std::fs::create_dir_all(directory).map_err(|error| {
+                Failure::new(format!("could not make {}: {error}", directory.display()))
+            })?;
+        }
+        std::fs::write(&self.daemon_marker, &daemon.name).map_err(|error| {
+            Failure::new(format!("could not write {}: {error}", self.daemon_marker))
+        })?;
+
+        let mut command = std::process::Command::new(program);
+        command.args(args);
+        command.env_clear();
+        if let Some(cwd) = &exec.cwd {
+            command.current_dir(cwd);
+        }
+        command.envs(environment(exec, Some(&runtime)));
+        // SAFETY: async-signal-safe calls only, nothing allocated.
+        unsafe {
+            command.pre_exec(move || {
+                if libc::setsid() == -1 {
+                    return Err(io::Error::last_os_error());
+                }
+                if libc::setgid(gid) != 0 {
+                    return Err(io::Error::last_os_error());
+                }
+                if libc::setuid(uid) != 0 {
+                    return Err(io::Error::last_os_error());
+                }
+                Ok(())
+            });
+        }
+        let mut watched = self
+            .waiters
+            .watch(|| Ok(command.spawn()?.id() as i32))
+            .map_err(|error| Failure::new(format!("{program} as {uid}:{gid}: {error}")))?;
+        let exit = watched.take_exit().expect("a new watch has its exit");
+        let name = daemon.name.clone();
+        tracing::info!(pid = watched.pid, %name, "the box's daemon is started");
+        self.daemon_exit = Some(Box::pin(async move {
+            exit.await
+                .map_err(|_| io::Error::other(format!("the {name} daemon's exit was not delivered")))
+        }));
+        self.daemon = Some(watched);
+        Ok(())
+    }
+
+    fn daemon_exit(&mut self) -> Option<Exited> {
+        self.daemon_exit.take()
+    }
+
     fn stop_launch(&mut self, id: &LaunchId) {
         if let Some(group) = self.groups.get(id) {
             // Whether or not the top process is still there: its children may
@@ -2082,6 +2180,11 @@ pub mod double {
         pub compositor_failure: Option<Failure>,
         /// When the compositor "exits": the session takes this after it starts.
         pub compositor_exit: Option<oneshot::Receiver<Exit>>,
+        /// Every daemon asked for.
+        pub daemons: Vec<Daemon>,
+        pub daemon_failure: Option<Failure>,
+        /// When the daemon "exits": the session takes this after it starts.
+        pub daemon_exit: Option<oneshot::Receiver<Exit>>,
         pub mount_failure: Option<Failure>,
         pub image_failure: Option<Failure>,
         /// Like `image_failure`, but only for a call that names an image, so a
@@ -2122,6 +2225,9 @@ pub mod double {
                 compositors: Vec::new(),
                 compositor_failure: None,
                 compositor_exit: None,
+                daemons: Vec::new(),
+                daemon_failure: None,
+                daemon_exit: None,
                 mount_failure: None,
                 image_failure: None,
                 named_image_failure: None,
@@ -2227,6 +2333,22 @@ pub mod double {
             }))
         }
 
+        fn start_daemon(&mut self, daemon: &Daemon) -> Result<(), Failure> {
+            self.order.push("daemon");
+            self.daemons.push(daemon.clone());
+            match &self.daemon_failure {
+                Some(failure) => Err(failure.clone()),
+                None => Ok(()),
+            }
+        }
+
+        fn daemon_exit(&mut self) -> Option<Exited> {
+            let rx = self.daemon_exit.take()?;
+            Some(Box::pin(async move {
+                rx.await.map_err(|_| io::Error::other("the daemon was dropped"))
+            }))
+        }
+
         fn stop_launch(&mut self, id: &LaunchId) {
             self.stops += 1;
             self.stopped.push(Some(id.clone()));

crates/nesprotocol/src/lifecycle.rs

@@ -294,6 +294,30 @@ pub struct BootDescriptor {
     /// what a workload that draws nothing wants.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub compositor: Option<Compositor>,
+    /// One process the box runs for as long as it lives, such as the box's
+    /// Steam client. ref(d-0085)
+    ///
+    /// Started after the compositor and before the box is ready, as a user of
+    /// the workload's: the sign-in it holds is the box's, shared by every launch
+    /// in it, and a launch that signed in for itself would replace the one
+    /// beside it. If it exits the box ends and the host is told which it was,
+    /// as for the compositor. `None` is a box with nothing of the kind, and
+    /// reads the same as a descriptor from before the field existed.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub daemon: Option<Daemon>,
+}
+
+/// A process the box keeps for its whole life. See [`BootDescriptor::daemon`].
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Daemon {
+    /// What the host is told it was if it dies, and what the box lists among
+    /// its services once it is started.
+    pub name: String,
+    /// How it is started. The same shape as a launch's, because it is started
+    /// the same way: its user's runtime directory made, the environment cleared
+    /// and this one on top.
+    pub exec: Exec,
 }
 
 /// What the compositor is started with.
@@ -607,6 +631,7 @@ mod tests {
             images: Vec::new(),
             video: VideoLimits::default(),
             compositor: None,
+            daemon: None,
         }
     }
 
@@ -913,6 +938,7 @@ mod video_limits_tests {
             images: Vec::new(),
             video: VideoLimits::default(),
             compositor: None,
+            daemon: None,
         };
         let json = serde_json::to_string(&d).unwrap();
         assert!(!json.contains("bitrate"), "{json}");
@@ -928,6 +954,7 @@ mod video_limits_tests {
                 bitrate_kbps: Some(8_000),
             },
             compositor: None,
+            daemon: None,
         };
         let back: BootDescriptor =
             serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap();
@@ -979,6 +1006,7 @@ mod image_tests {
             }],
             video: VideoLimits::default(),
             compositor: None,
+            daemon: None,
         };
         let json = serde_json::to_string(&d).unwrap();
         assert!(json.contains(r#""images":[{"device":"/dev/vdd","at":"/nestri/compat"}]"#));
@@ -1065,4 +1093,28 @@ mod attach_tests {
             serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap();
         assert_eq!(back, d);
     }
+
+    /// A daemon is optional and round-trips, so a descriptor from before the
+    /// field reads the same.
+    #[test]
+    fn a_daemon_is_optional_and_round_trips() {
+        let none: BootDescriptor = serde_json::from_str("{}").unwrap();
+        assert!(none.daemon.is_none());
+        assert!(!serde_json::to_string(&none).unwrap().contains("daemon"));
+        let d = BootDescriptor {
+            daemon: Some(Daemon {
+                name: "steam".into(),
+                exec: Exec {
+                    argv: vec!["/bin/x".into()],
+                    env: BTreeMap::from([("A".to_string(), "b".to_string())]),
+                    cwd: None,
+                    uid: 1001,
+                    gid: 1001,
+                },
+            }),
+            ..none
+        };
+        let back: BootDescriptor = serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap();
+        assert_eq!(back, d);
+    }
 }