diff --git a/crates/minimald/src/session.rs b/crates/minimald/src/session.rs index 99551b457..b58625650 100644 --- a/crates/minimald/src/session.rs +++ b/crates/minimald/src/session.rs @@ -66,6 +66,34 @@ async fn materialize_patches_into_home( Ok(()) } +/// Load the persisted composition snapshot for an `Active` session +/// brought up from disk after a daemon restart. Returns `None` (with +/// a warning log) when the sidecar is missing or corrupt. The +/// launcher then falls back to its baseline set, preserving the +/// "attach still works" property at the cost of the lost loadout +/// contributions. This is the loud-fallback path: the operator sees +/// the warning instead of a silent drop. +async fn load_composition(record: &SessionRecordHandle) -> Option> { + match record.load_composition().await { + Ok(Some(comp)) => Some(Arc::new(comp)), + Ok(None) => { + tracing::warn!( + session_id = %record.id(), + "no composition snapshot for Active session; falling back to baseline", + ); + None + } + Err(e) => { + tracing::warn!( + session_id = %record.id(), + error = %e, + "failed to load composition snapshot; falling back to baseline", + ); + None + } + } +} + /// The name a session's PTask hostname is registered under, doubling as the /// session host's display name: the session's assigned name, or the project /// directory's basename when unnamed. @@ -185,8 +213,10 @@ enum SessionInner { /// on-disk `Active` record. Active { /// The finalized [`Composition`] this session was created with. - /// `None` after a daemon restart — the composition isn't persisted, - /// and the launcher falls back to its baseline set in that case. + /// `None` only when the sidecar is missing or corrupt on a + /// session brought up from disk after a daemon restart — + /// [`load_composition`] logs a warning and the launcher + /// falls back to its baseline set in that case. /// /// The launcher currently consumes only the composition's packages /// and vars. Patches (need file-upload plumbing) and lifecycle hooks @@ -387,19 +417,23 @@ impl Session { /// Launches the actor for a session — the one path onto which every /// session actor is spawned. The initial state machine state is derived /// from the record alone: an `Active` record comes up ready to attach - /// (without a composition; it isn't persisted, so the launcher falls back - /// to its baseline set), a `Pending` one as an unconfigured `Draft` - /// awaiting `ConfigureLoadout`. + /// with its composition restored from the snapshot sidecar (or, if the + /// sidecar is missing or corrupt, with a logged warning and no + /// composition so the launcher falls back to its baseline set), a + /// `Pending` one as an unconfigured `Draft` awaiting `ConfigureLoadout`. pub(crate) async fn run(conf: SessionConfig) -> Result { let obj = conf.record.object().await?; Self::create_dirs(&obj)?; let inner = match obj.record().status { - SessionStatus::Active => SessionInner::Active { - composition: None, - host: None, - sops: vec![], - }, + SessionStatus::Active => { + let composition = load_composition(&conf.record).await; + SessionInner::Active { + composition, + host: None, + sops: vec![], + } + } SessionStatus::Pending => SessionInner::Draft { pending: None }, // `Materializing` records are only meaningful across a // matching in-memory composition, which is lost on @@ -714,6 +748,13 @@ impl Session { // (a session isn't attachable until then, so // publishing the route would let something reach a // launcher that can't materialize its patches). + // + // Persist the composition snapshot before the record + // write so a crash at any point leaves either a + // reaped session (Pending/Materializing records are + // reaped at startup) or an Active session with its + // sidecar intact. + self.record.store_composition(&composition).await?; let mut record = object.record().clone(); record.status = SessionStatus::Materializing; self.record.write(record.clone()).await?; @@ -786,6 +827,11 @@ impl Session { // `FinalizeSession` after patches upload — see // [`Self::finalize`] for the transition and its // preconditions. + // + // Persist the composition snapshot before the record write + // (same crash-safety reasoning as the Materialized fast + // path above). + self.record.store_composition(&composition).await?; let mut record = self.record.record().await?; record.status = SessionStatus::Materializing; self.record.write(record.clone()).await?; @@ -1991,4 +2037,98 @@ mod tests { "reattaching should flush prior terminal state, got: {flushed:?}", ); } + + /// A session's composition persists across a daemon restart: the + /// sidecar written at composition-assembly time is read back when + /// the actor is respawned from disk, so the launcher sees the same + /// packages and vars instead of falling back to the baseline set. + /// This is the core fix for issue #849 — "session composition state + /// is in-memory only: daemon restart drops loadout packages/vars + /// for existing sessions." + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn composition_survives_actor_restart() { + let server = TestServer::new().await; + let mut client = server.connect().await; + let session_id = create_session(&mut client).await; + + // The actor was spawned during `create_configured_session` + // and holds the composition in memory. Verify it's present. + let manager = server.state.sessions_manager().await; + let handle = manager + .get_session(crate::sessions::SessionKeyPredicate::Id(session_id)) + .await + .unwrap() + .expect("session should resolve while actor is running"); + assert!( + handle.peek_composition().await.is_some(), + "freshly configured session should hold its composition in memory" + ); + + // Stop the actor and evict it from the running map so the + // next `get_session` spawns a fresh actor from the on-disk + // record — simulating a daemon restart. + handle.stop().await; + manager.evict(session_id).await; + + // Re-resolve: spawns a new actor from disk. The composition + // should be restored from the sidecar, not None. + let handle = manager + .get_session(crate::sessions::SessionKeyPredicate::Id(session_id)) + .await + .unwrap() + .expect("session should resolve after eviction"); + assert!( + handle.peek_composition().await.is_some(), + "re-spawned session should have its composition restored from \ + the sidecar, not fall back to baseline" + ); + } + + /// When the sidecar is missing (e.g. a session that predated the + /// sidecar, or a corrupt filesystem), the actor still spawns — but + /// with no composition, so the launcher falls back to its baseline + /// set. The operator sees a warning log rather than a silent drop. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn missing_sidecar_falls_back_to_baseline() { + let server = TestServer::new().await; + let mut client = server.connect().await; + let session_id = create_session(&mut client).await; + + let manager = server.state.sessions_manager().await; + let handle = manager + .get_session(crate::sessions::SessionKeyPredicate::Id(session_id)) + .await + .unwrap() + .expect("session should resolve"); + + // Delete the sidecar to simulate a pre-sidecar session or a + // corrupt filesystem. The composition sidecar lives at + // `/composition.json`, a sibling of `record.json`; + // derive it from the workspace path (`/tree`). + let paths = handle.paths().await.expect("paths should resolve"); + let composition_file = paths + .working + .parent() + .expect("workspace path has a parent") + .join(&paths::DaemonRelPath::try_new("composition.json").unwrap()); + tokio::fs::remove_file(composition_file.as_utf8_path()) + .await + .expect("sidecar should exist to delete"); + + handle.stop().await; + manager.evict(session_id).await; + + // Re-resolve: spawns from disk with no sidecar. The + // composition should be None — loud fallback to baseline. + let handle = manager + .get_session(crate::sessions::SessionKeyPredicate::Id(session_id)) + .await + .unwrap() + .expect("session should resolve after eviction"); + assert!( + handle.peek_composition().await.is_none(), + "session with a missing sidecar should fall back to baseline \ + (composition None), not hold a stale composition" + ); + } } diff --git a/crates/minimald/src/store.rs b/crates/minimald/src/store.rs index 0f700c46e..32279fae0 100644 --- a/crates/minimald/src/store.rs +++ b/crates/minimald/src/store.rs @@ -47,6 +47,15 @@ enum StoreMessage { oneshot::Sender>, ), SessionDelete(DiskSessionKey, oneshot::Sender>), + CompositionLoad( + DiskSessionKey, + oneshot::Sender, std::io::Error>>, + ), + CompositionStore( + DiskSessionKey, + std::sync::Arc, + oneshot::Sender>, + ), } /// The session store actor. Mediates reading, writing, and enumerating @@ -132,6 +141,12 @@ impl Store { StoreMessage::SessionDelete(k, r) => { let _ = r.send(self.store.delete(&k)); } + StoreMessage::CompositionLoad(k, r) => { + let _ = r.send(self.store.load_composition(&k)); + } + StoreMessage::CompositionStore(k, comp, r) => { + let _ = r.send(self.store.store_composition(&k, &comp)); + } } } } @@ -249,6 +264,44 @@ impl SessionRecordHandle { .await; recv.await.expect("corresponding store is dead") } + + /// Loads the persisted composition snapshot for this session, if + /// one exists. Returns `Ok(None)` when the sidecar is absent + /// (pre-sidecar session, or composition was never assembled). A + /// corrupt sidecar surfaces as an error so the caller can log it + /// and fall back to baseline. + pub async fn load_composition( + &self, + ) -> Result, std::io::Error> { + let (send, recv) = oneshot::channel(); + let _ = self + .h + .sender + .send(StoreMessage::CompositionLoad(self.k.clone(), send)) + .await; + recv.await.expect("corresponding store is dead") + } + + /// Atomically persists the composition snapshot for this session + /// (tmp + rename). Called at composition-assembly time so a + /// restart can re-apply the exact composition that was approved at + /// `min activate` time. + pub async fn store_composition( + &self, + composition: &sessions::core::compose::Composition, + ) -> Result<(), std::io::Error> { + let (send, recv) = oneshot::channel(); + let _ = self + .h + .sender + .send(StoreMessage::CompositionStore( + self.k.clone(), + std::sync::Arc::new(composition.clone()), + send, + )) + .await; + recv.await.expect("corresponding store is dead") + } } /// A non-owning handle to the [`Store`] actor. diff --git a/crates/sessions/docs/COMPOSITION.md b/crates/sessions/docs/COMPOSITION.md index 21dce5c1a..e5644cb39 100644 --- a/crates/sessions/docs/COMPOSITION.md +++ b/crates/sessions/docs/COMPOSITION.md @@ -672,14 +672,14 @@ overload: stuck `Materializing` records. - **`Materializing` records don't survive daemon restart.** The `SessionInner::Active { composition, .. }` state that Phase 4a - produces is memory-only: the composition isn't persisted to the - record, only the *status* is. A `Materializing` record whose - actor is respawned from disk after a restart has no - in-memory composition to check against, so `Manager::init` - runs `reap_unresumable_records` at startup and deletes any - `Pending` or `Materializing` record it finds (with an `info!` - log). If a race lets one through, `finalize` refuses with an - `InvalidInput` fault ("session is Materializing but has no + produces is persisted to a `composition.json` sidecar alongside + `record.json` so a restart can restore it for `Active` sessions. + But a `Materializing` record means the patches upload hasn't + completed — the sidecar exists but the patches marker doesn't, + so `Manager::init` runs `reap_unresumable_records` at startup and + deletes any `Pending` or `Materializing` record it finds (with an + `info!` log). If a race lets one through, `finalize` refuses with + an `InvalidInput` fault ("session is Materializing but has no in-memory composition") so the operator sees the problem instead of silently attaching to an empty-home shell. - **Patches unpack is atomic; the marker is the precondition.** diff --git a/crates/sessions/src/core/compose.rs b/crates/sessions/src/core/compose.rs index 1b59b0a3b..20bed3493 100644 --- a/crates/sessions/src/core/compose.rs +++ b/crates/sessions/src/core/compose.rs @@ -1232,6 +1232,38 @@ impl Composition { } } +/// Reconstruct a [`Composition`] from a persisted +/// [`WireComposition`](crate::wire::request::WireComposition) +/// snapshot. The daemon writes the snapshot at composition-assembly +/// time and reads it back at spawn-from-disk so a restart re-applies +/// the exact composition that was approved at `min activate` time. +/// +/// Fallible only on lifecycle hooks (a wire hook with no callbacks +/// is rejected); vars, patches, and packages convert infallibly via +/// their existing `From` impls. +impl TryFrom for Composition { + type Error = ComposeError; + + fn try_from(wire: crate::wire::request::WireComposition) -> Result { + let hooks: Vec = wire + .lifecycle_hooks + .into_iter() + .map(|h| { + h.try_into().map_err(|e| ComposeError::InvalidWireItem { + what: "lifecycle hook with no callbacks", + context: format!("{e}"), + }) + }) + .collect::>()?; + Ok(Self { + vars: wire.vars.into_iter().map(Into::into).collect(), + patches: wire.patches.into_iter().map(Into::into).collect(), + packages: wire.packages.into_iter().map(Into::into).collect(), + lifecycle_hooks: hooks, + }) + } +} + /// Configuration for the compose pipeline. /// /// Defaults to symlink-safe behavior (no following) — appropriate for diff --git a/crates/sessions/src/store.rs b/crates/sessions/src/store.rs index 8ddbc2973..519f5c529 100644 --- a/crates/sessions/src/store.rs +++ b/crates/sessions/src/store.rs @@ -140,6 +140,38 @@ pub trait Loader { /// - Other I/O errors if the record cannot be read or the index cannot be /// flushed. fn delete(&mut self, key: &Self::Key) -> Result<(), std::io::Error>; + + /// Loads the persisted composition snapshot for the session at + /// `key`, if one exists. Returns `Ok(None)` when the sidecar is + /// absent (a session that predates the sidecar, or whose + /// composition was never assembled). A corrupt or unreadable + /// sidecar surfaces as an error so the caller can log it and + /// decide on a fallback. + /// + /// # Errors + /// + /// - `NotFound` if `key` is stale (same semantics as [`Self::save`]). + /// - I/O or deserialization errors if the sidecar exists but + /// cannot be read or parsed. + fn load_composition( + &self, + key: &Self::Key, + ) -> Result, std::io::Error>; + + /// Atomically persists the composition snapshot for the session + /// at `key` (tmp + rename, same crash-safety as [`Self::save`]). + /// Called at composition-assembly time so a restart can re-apply + /// the exact composition that was approved at `min activate` time. + /// + /// # Errors + /// + /// - `NotFound` if `key` is stale (same semantics as [`Self::save`]). + /// - I/O or serialization errors if the sidecar cannot be written. + fn store_composition( + &self, + key: &Self::Key, + composition: &crate::core::compose::Composition, + ) -> Result<(), std::io::Error>; } /// The concrete key used to identify sessions from [`DiskLoader`]. @@ -871,6 +903,68 @@ impl Loader for DiskLoader { Err(e) => Err(e), } } + + fn load_composition( + &self, + key: &Self::Key, + ) -> Result, std::io::Error> { + let short = self.live_short(key)?; + let path = self + .minimal_dir + .as_utf8_path() + .join("sessions") + .join(&short) + .join("composition.json"); + match std::fs::read(&path) { + Ok(bytes) => { + let wire: crate::wire::request::WireComposition = serde_json::from_slice(&bytes) + .map_err(|e| { + std::io::Error::other(format!( + "parsing composition snapshot at {}: {e}", + path.as_str() + )) + })?; + crate::core::compose::Composition::try_from(wire) + .map(Some) + .map_err(|e| { + std::io::Error::other(format!( + "reconstructing composition from snapshot at {}: {e}", + path.as_str() + )) + }) + } + Err(e) if e.kind() == NotFound => Ok(None), + Err(e) => Err(e), + } + } + + fn store_composition( + &self, + key: &Self::Key, + composition: &crate::core::compose::Composition, + ) -> Result<(), std::io::Error> { + let short = self.live_short(key)?; + let session_dir = self + .minimal_dir + .as_utf8_path() + .join("sessions") + .join(&short); + std::fs::create_dir_all(&session_dir)?; + let dest = session_dir.join("composition.json"); + let tmp = session_dir.join("composition.json.tmp"); + + let wire = crate::wire::request::WireComposition::from(composition); + let json = serde_json::to_vec_pretty(&wire) + .map_err(|e| std::io::Error::other(format!("serializing composition snapshot: {e}")))?; + std::fs::write(&tmp, &json)?; + + #[cfg(target_os = "linux")] + common::renameat2::renameat2_cwd(tmp.as_std_path(), dest.as_std_path(), 0)?; + #[cfg(not(target_os = "linux"))] + std::fs::rename(&tmp, &dest)?; + + Ok(()) + } } #[cfg(test)] diff --git a/crates/sessions/src/wire/request.rs b/crates/sessions/src/wire/request.rs index b54058ddf..761e0c8c3 100644 --- a/crates/sessions/src/wire/request.rs +++ b/crates/sessions/src/wire/request.rs @@ -17,6 +17,10 @@ use super::primitives::{ }; use crate::SessionId; +/// Snapshot format version for [`WireComposition`]. Bumped when the +/// on-disk shape changes in a way older daemons can't tolerate. +const COMPOSITION_SNAPSHOT_VERSION: u32 = 1; + /// The client's composed contribution, wire-shaped: var values /// resolved, patch sources expanded to concrete files, every item /// already gated by the user policy. @@ -32,6 +36,56 @@ pub struct WireContribution { pub requested_packages: Vec, } +/// Persisted composition snapshot, written by the daemon as a sidecar +/// to [`record.json`](crate::store) at composition-assembly time so a +/// restart can re-apply the exact [`Composition`] that was approved at +/// `min activate` time. +/// +/// Mirrors [`Composition`](crate::core::compose::Composition) via the +/// same wire primitives used for the live RPC flow +/// ([`WireContribution`], [`ContributionResponse`]). A `version` field +/// gates the on-disk shape so a future format can evolve without +/// ambiguity; the current and only version is +/// [`COMPOSITION_SNAPSHOT_VERSION`] (1). +#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct WireComposition { + /// Snapshot format version. Defaults to + /// [`COMPOSITION_SNAPSHOT_VERSION`] so sidecars written by daemons + /// that predate the field deserialize cleanly. + #[serde(default = "default_composition_version")] + pub version: u32, + /// Variables that survived the policy gate. + pub vars: Vec, + /// Patches that survived the policy gate. + pub patches: Vec, + /// Packages contributed to the session (pass-through; no gate). + pub packages: Vec, + /// Lifecycle hooks contributed to the session (pass-through; no + /// gate). + pub lifecycle_hooks: Vec, +} + +fn default_composition_version() -> u32 { + COMPOSITION_SNAPSHOT_VERSION +} + +impl From<&crate::core::compose::Composition> for WireComposition { + fn from(c: &crate::core::compose::Composition) -> Self { + Self { + version: COMPOSITION_SNAPSHOT_VERSION, + vars: c.vars().iter().cloned().map(Into::into).collect(), + patches: c.patches().iter().cloned().map(Into::into).collect(), + packages: c.packages().iter().cloned().map(Into::into).collect(), + lifecycle_hooks: c + .lifecycle_hooks() + .iter() + .cloned() + .map(Into::into) + .collect(), + } + } +} + /// Daemon → Client: items from the daemon-side closure (packages, /// project config) that need client-side policy + prompts. /// @@ -177,4 +231,36 @@ mod tests { "expected `kind` tag, got: {json}" ); } + + #[test] + fn composition_snapshot_round_trips() { + let c = WireComposition { + version: 1, + vars: vec![super::WireSessionVar { + var: WireResolvedVar { + name: "EDITOR".into(), + value: "hx".into(), + carries_user_data: false, + }, + source: WireSource::UserLoadout { name: "dev".into() }, + }], + patches: vec![], + packages: vec![super::WirePackageRef { + name: "helix".into(), + source: WireSource::UserLoadout { name: "dev".into() }, + }], + lifecycle_hooks: vec![], + }; + assert_eq!(round_trip(&c), c); + } + + #[test] + fn composition_snapshot_defaults_version_to_one() { + // A sidecar written without the `version` field (e.g. by a + // hypothetical future writer that omits it) should default to + // version 1, not 0. + let json = r#"{"vars":[],"patches":[],"packages":[],"lifecycle_hooks":[]}"#; + let c: WireComposition = serde_json::from_str(json).unwrap(); + assert_eq!(c.version, COMPOSITION_SNAPSHOT_VERSION); + } }