fix(hvf,hostd): guest PC wrap, vCPU-panic unwind guard, HTTP flow cancellation through response framing (#4839)

crates/deps/libkrun-sys/src/sys.rs

@@ -291,14 +291,6 @@ impl Context {
         check(unsafe { bindings::krun_set_console_output(self.ctx_id, path.as_ptr()) })
     }
 
-    /// Returns the shutdown eventfd. The caller is responsible for
-    /// closing it (libkrun documents that the fd is owned by the
-    /// caller once returned).
-    pub fn shutdown_eventfd(&self) -> Result<i32, Error> {
-        let rc = unsafe { bindings::krun_get_shutdown_eventfd(self.ctx_id) };
-        if rc < 0 { Err(Error::Krun(rc)) } else { Ok(rc) }
-    }
-
     /// Block the calling thread, starting the guest. libkrun's
     /// `krun_start_enter` calls `exit()` on success with the guest's
     /// status; on failure it returns a negative errno. This wrapper is

crates/mvm-agentd/src/builder_session.rs

@@ -17,14 +17,10 @@
 //! path-sanitized [`builder_transfer::receive_into`] — a hostile guest cannot
 //! write outside the host's artifact dir.
 
-use std::io::{Read, Write};
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 
-use anyhow::Result;
 use serde::{Deserialize, Serialize};
 
-use crate::builder_transfer::{read_msg, receive_into, send_dir, write_msg};
-
 /// What the host asks the builder guest to build. `flake_ref`/`attr` are the
 /// same validated inputs the existing builder agent takes; the guest resolves
 /// them against the staged `/work` tree.
@@ -64,37 +60,5 @@ pub struct BuildProduct {
     pub logs: Vec<String>,
 }
 
-/// Guest side: serve one build session on `stream`. Receives `/work` +
-/// `/mvm-bins` into the given destinations, reads the request, runs `build`, then
-/// streams the logs + outcome (+ the artifact tree on success) back.
-pub fn serve_build<S, F>(
-    stream: &mut S,
-    work_dest: &Path,
-    mvm_bins_dest: &Path,
-    build: F,
-) -> Result<()>
-where
-    S: Read + Write,
-    F: FnOnce(&Path, &Path, &BuildRequest) -> BuildProduct,
-{
-    receive_into(stream, work_dest)?;
-    receive_into(stream, mvm_bins_dest)?;
-    let request: BuildRequest = read_msg(stream)?;
-    let product = build(work_dest, mvm_bins_dest, &request);
-    for line in &product.logs {
-        write_msg(stream, &SessionEvent::Log { line: line.clone() })?;
-    }
-    write_msg(
-        stream,
-        &SessionEvent::Done {
-            outcome: product.outcome.clone(),
-        },
-    )?;
-    if matches!(product.outcome, BuildOutcome::Success) {
-        send_dir(stream, &product.out_dir)?;
-    }
-    Ok(())
-}
-
 #[cfg(test)]
 mod tests {}

crates/mvm-agentd/src/trusted_env.rs

@@ -85,17 +85,6 @@ pub fn configure_command(command: &mut Command) {
     configure_with_env(command, std::env::vars_os());
 }
 
-/// Sanitize a fully assembled explicit environment, without importing any
-/// additional variables from the parent. Apply after combining image and
-/// request variables for a trusted mediated launch.
-pub fn configure_explicit_command(command: &mut Command) {
-    let env: Vec<_> = command
-        .get_envs()
-        .filter_map(|(key, value)| Some((key.to_owned(), value?.to_owned())))
-        .collect();
-    configure_with_env(command, env);
-}
-
 fn configure_with_env(
     command: &mut Command,
     env: impl IntoIterator<Item = (std::ffi::OsString, std::ffi::OsString)>,

crates/mvm-build/src/runtime_layer.rs

@@ -509,47 +509,6 @@ pub enum RuntimeLayerStageError {
     Materialize(#[from] crate::rootfs::RootfsError),
 }
 
-/// Verify revocation and snapshot custody around strict, bounded extraction.
-///
-/// # Errors
-///
-/// Refuses withdrawn trust, changed archive bytes or tar entries outside the
-/// runtime-layer subset. This is not a complete composed rootfs or boot grant.
-pub fn stage_runtime_layer(
-    archive: &VerifiedRuntimeLayerArchive,
-) -> Result<StagedRuntimeLayer, RuntimeLayerStageError> {
-    archive.descriptor().check_current_revocations()?;
-    let mut file = archive.open_verified()?;
-    let stage = stage_tar(&mut file)?;
-    archive.reverify()?;
-    archive.descriptor().check_current_revocations()?;
-    Ok(stage)
-}
-
-/// Compose authenticated runtime archives without Nix or a builder VM.
-///
-/// This prepares reusable filesystem input only. It cannot be booted, signed as
-/// an application, or installed as a `.mvmpkg` without the signed base, an
-/// application entrypoint, policy admission, verity and final bundle signing.
-/// A directory may recur across layers, but no file or symlink may overlap.
-///
-/// # Errors
-///
-/// Refuses revoked or changed archives, unresolved dependencies, path
-/// collisions, and any tar semantics not faithfully retained by extraction.
-pub fn compose_runtime_layers(
-    archives: &[&VerifiedRuntimeLayerArchive],
-) -> Result<ComposedRuntimeLayers, RuntimeLayerStageError> {
-    if archives.is_empty() || archives.len() > MAX_LAYERS {
-        return Err(RuntimeLayerStageError::InconsistentLayers);
-    }
-    let arch = &archives[0].descriptor().manifest().runtime_layer.guest_arch;
-    let pins = runtime_pins(archives, arch)?;
-    let mut composed = ComposedRuntimeLayers::new(pins)?;
-    add_runtime_archives(&mut composed, archives)?;
-    Ok(composed)
-}
-
 /// Compose precisely the authenticated layers a v2 application requests.
 ///
 /// The host selects the generic base profile; the author cannot choose a
@@ -749,6 +708,7 @@ fn add_runtime_archives(
     Ok(())
 }
 
+#[cfg(test)]
 fn stage_tar<R: Read + Seek>(reader: &mut R) -> Result<StagedRuntimeLayer, RuntimeLayerStageError> {
     preflight(reader, TarRole::Runtime)?;
     let directory = tempfile::tempdir()?;

crates/mvm-client/src/launch/offline_revocation.rs

@@ -44,12 +44,6 @@ impl OfflineRevocationPolicy {
         Ok(Self { verified })
     }
 
-    /// Load from the host's default MVM_HOME store.
-    pub fn load_from_host() -> Result<Self, ImageSetRevocationStoreError> {
-        let store = ImageSetRevocationStore::in_mvm_home();
-        Self::load_at(&store, Utc::now())
-    }
-
     /// The authenticated signer identity of the loaded list.
     pub fn signer_identity(&self) -> &str {
         self.verified.signer_identity()

crates/mvm-core/src/spawn_scope.rs

@@ -566,15 +566,6 @@ impl BoundCommand {
         &mut self.command
     }
 
-    /// Shorten the creation deadline, so the timeout path is testable without
-    /// a ten-second wait.
-    #[cfg(any(test, feature = "test-support"))]
-    #[must_use]
-    pub fn with_creation_timeout(mut self, timeout: Duration) -> Self {
-        self.creation_timeout = timeout;
-        self
-    }
-
     /// Start the launch and, for a scoped one, wait until the payload is
     /// running inside its scope.
     ///

crates/mvm-hostd/src/supervisor/aggregate.rs

@@ -251,12 +251,6 @@ impl Supervisor {
         self
     }
 
-    /// Wire a prebuilt artifact collector slot.
-    pub fn with_artifact_collector(mut self, artifact: Arc<dyn ArtifactCollector>) -> Self {
-        self.artifact = artifact;
-        self
-    }
-
     /// Wire the platform firewall enforcer slot. The default
     /// [`NoopFirewallEnforcer`] fails closed until a real enforcer is
     /// supplied.

crates/mvm-hostd/src/supervisor/flowmux/http_flow.rs

@@ -293,24 +293,16 @@ pub(super) fn spawn_forward(task: ForwardTask) {
     runtime.spawn(async move {
         let request_url = head.url.clone();
         // A typed HTTP flow names no invocation, so it belongs to no tool.
-        let result = tokio::select! {
-            biased;
-            canceled = cancellation.changed() => {
-                let reason = if canceled.is_ok() {
-                    "guest_reset"
-                } else {
-                    "session_closed"
-                };
-                service.audit_http_stream_failure(&request_url, reason).await;
-                Ok(())
-            }
-            response = service.process_body_stream(head, body) => {
-                match response {
-                    Ok(response) => write_stream_response(&session, &writer, stream_id, response).await,
-                    Err(refusal) => write_response(&session, &writer, stream_id, refusal),
-                }
+        let forward = async {
+            match service.process_body_stream(head, body).await {
+                Ok(response) => write_stream_response(&session, &writer, stream_id, response).await,
+                Err(refusal) => write_response(&session, &writer, stream_id, refusal),
             }
         };
+        let result = run_until_canceled(&mut cancellation, forward, |reason| {
+            service.audit_http_stream_failure(&request_url, reason)
+        })
+        .await;
         cancellations
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner)
@@ -329,6 +321,32 @@ pub(super) fn spawn_forward(task: ForwardTask) {
     });
 }
 
+/// Run `work` until it finishes or the flow is canceled. `work` must include
+/// response framing: a frame sent after a guest `Reset` ends the whole session.
+async fn run_until_canceled<W, A>(
+    cancellation: &mut tokio::sync::watch::Receiver<bool>,
+    work: W,
+    audit_failure: impl FnOnce(&'static str) -> A,
+) -> Result<(), super::FlowMuxError>
+where
+    W: std::future::Future<Output = Result<(), super::FlowMuxError>>,
+    A: std::future::Future<Output = ()>,
+{
+    tokio::select! {
+        biased;
+        canceled = cancellation.changed() => {
+            let reason = if canceled.is_ok() {
+                "guest_reset"
+            } else {
+                "session_closed"
+            };
+            audit_failure(reason).await;
+            Ok(())
+        }
+        result = work => result,
+    }
+}
+
 /// Frame a bounded upstream stream without collecting its body in host
 /// memory. Each receive waits on the channel event with an idle deadline.
 async fn write_stream_response(
@@ -552,6 +570,25 @@ mod tests {
         assert!(!cancel_forwarding(&cancellations, 9));
     }
 
+    #[tokio::test]
+    async fn a_reset_while_the_response_is_still_streaming_stops_the_forward() {
+        let (sender, mut receiver) = tokio::sync::watch::channel(false);
+        // Response framing that has sent a chunk and waits on upstream forever.
+        let streaming = std::future::pending::<Result<(), super::super::FlowMuxError>>();
+        let forward = tokio::spawn(async move {
+            run_until_canceled(&mut receiver, streaming, |reason| async move {
+                assert_eq!(reason, "guest_reset");
+            })
+            .await
+        });
+        sender.send(true).unwrap();
+        tokio::time::timeout(Duration::from_secs(5), forward)
+            .await
+            .expect("a canceled forward must not outlive its stream")
+            .unwrap()
+            .unwrap();
+    }
+
     #[test]
     fn a_forward_task_builder_refuses_missing_runtime_state() {
         assert!(matches!(

crates/mvm-runtime/src/backends/hvf/kernel_boot.rs

@@ -34,11 +34,11 @@ use super::mmio_layout::{
     BALLOON_IRQ, BALLOON_MMIO_BASE, MAX_DISKS, RNG_IRQ, RNG_MMIO_BASE, VSOCK_IRQ, VSOCK_MMIO_BASE,
     disk_mmio,
 };
-use super::smp::{CreationOrder, Release, SecondaryGates, VcpuStart, psci};
+use super::smp::{CreationOrder, OnDrop, Release, SecondaryGates, VcpuStart, psci};
 use super::snapshot::{HVF_SNAPSHOT_BACKEND_KIND, HvfVcpuState};
 use super::stop_wake::{RunOver, wake_until_done};
 use super::sys::*;
-use super::vcpu::esr_ec;
+use super::vcpu::{esr_ec, next_pc};
 use super::{BootFault, default_bootargs};
 use crate::vmm::device::Pl011;
 use crate::vmm::device_state::SnapshotDeviceState;
@@ -186,7 +186,7 @@ fn handle_exception(
         }
         // Advance past the faulting instruction and keep going.
         let pc = vc.get_core(CoreReg::Pc)?;
-        vc.set_core(CoreReg::Pc, pc + 4)?;
+        vc.set_core(CoreReg::Pc, next_pc(pc))?;
         return Ok(RunControl::Continue);
     }
 
@@ -1331,6 +1331,7 @@ fn run_secondary<B: run::DeviceBus>(
     let clock = ThreadCpuHandle::for_current_thread().ok().map(Arc::new);
     let registered = created.send(Ok(clock.clone())).is_ok();
 
+    let _end_on_unwind = OnDrop(|| shared.end());
     let result = (|| {
         if !registered {
             // The boot CPU gave up on the bring-up. Do not enter the guest.
@@ -1980,6 +1981,7 @@ unsafe fn run(
                     .collect();
                 drop(created_tx);
                 drop(ready_tx);
+                let _end_on_unwind = OnDrop(|| shared.end());
 
                 // Wait for the whole machine to exist before starting the
                 // guest. The device tree already describes these CPUs, so a

crates/mvm-runtime/src/backends/hvf/smp.rs

@@ -152,6 +152,17 @@ impl Drop for TurnGuard<'_> {
     }
 }
 
+/// Runs its closure when dropped, including by unwinding. A vCPU thread that
+/// panics would otherwise skip ending the machine and leave every parked CPU
+/// for `thread::scope` to join forever.
+pub(crate) struct OnDrop<F: FnMut()>(pub(crate) F);
+
+impl<F: FnMut()> Drop for OnDrop<F> {
+    fn drop(&mut self) {
+        (self.0)();
+    }
+}
+
 /// Why a parked secondary woke up.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Release {
@@ -441,6 +452,25 @@ mod tests {
 
     /// Shutdown after a release still lets an already-started CPU be woken
     /// again — the second wait returns `Shutdown` rather than blocking.
+    #[test]
+    fn a_panic_while_a_cpu_is_parked_still_releases_it() {
+        let (done, finished) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let gates = SecondaryGates::new(2);
+            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                std::thread::scope(|scope| {
+                    scope.spawn(|| gates.wait_for_release(1));
+                    let _end = OnDrop(|| gates.shutdown());
+                    panic!("boot CPU died");
+                })
+            }));
+            let _ = done.send(());
+        });
+        finished
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .expect("the thread scope must not wedge on a parked CPU");
+    }
+
     #[test]
     fn shutdown_is_idempotent_and_outlives_a_consumed_request() {
         let gates = SecondaryGates::new(2);

crates/mvm-runtime/src/backends/hvf/vcpu.rs

@@ -67,6 +67,11 @@ pub(super) unsafe fn write_gp(vcpu: hv_vcpu_t, reg: u8, value: u64) -> Result<()
     Ok(())
 }
 
+/// The PC after the 4-byte instruction at `pc`; wraps because the guest picks `pc`.
+pub(super) fn next_pc(pc: u64) -> u64 {
+    pc.wrapping_add(4)
+}
+
 /// Advance PC past the faulting (4-byte) instruction.
 ///
 /// # Safety
@@ -78,7 +83,7 @@ pub(super) unsafe fn advance_pc(vcpu: hv_vcpu_t) -> Result<(), HvfError> {
         if hv_vcpu_get_reg(vcpu, HV_REG_PC, &mut pc) != HV_SUCCESS {
             return Err(HvfError::GetReg(0));
         }
-        if hv_vcpu_set_reg(vcpu, HV_REG_PC, pc + 4) != HV_SUCCESS {
+        if hv_vcpu_set_reg(vcpu, HV_REG_PC, next_pc(pc)) != HV_SUCCESS {
             return Err(HvfError::SetReg(0));
         }
     }
@@ -98,6 +103,12 @@ mod tests {
         );
     }
 
+    #[test]
+    fn next_pc_wraps_at_the_top_of_the_address_space() {
+        assert_eq!(next_pc(0x4008_0000), 0x4008_0004);
+        assert_eq!(next_pc(u64::MAX - 3), 0);
+    }
+
     #[test]
     fn decodes_a_32bit_store_to_x0() {
         let iss = (1 << 24) | (2 << 22) | (1 << 6);