Merge cbd367a5461ae8adfe4986cbb8d499888c56eba4 into 4733e56af27e2e8c94b3779151bcaa647ac85432

Cargo.lock

@@ -3617,7 +3617,9 @@ dependencies = [
  "rand 0.8.6",
  "reqwest",
  "serde",
+ "serde_json",
  "serde_yaml",
+ "tempfile",
  "thiserror 2.0.18",
  "tokio",
  "tracing",

testing-framework/core/Cargo.toml

@@ -27,7 +27,11 @@ prometheus-http-query = "0.8"
 rand                  = { workspace = true }
 reqwest               = { features = ["json"], workspace = true }
 serde                 = { workspace = true }
+serde_json            = { workspace = true }
 serde_yaml            = { workspace = true }
 thiserror             = { workspace = true }
 tokio                 = { features = ["macros", "process", "rt-multi-thread", "sync", "time"], workspace = true }
 tracing               = { workspace = true }
+
+[dev-dependencies]
+tempfile = { workspace = true }

testing-framework/core/src/scenario/mod.rs

@@ -15,6 +15,7 @@ mod expectation;
 pub mod internal;
 mod observability;
 mod runtime;
+mod snapshot;
 mod sources;
 mod workload;
 
@@ -48,6 +49,10 @@ pub use runtime::{
     wait_for_http_ports_with_requirement_and_timeout, wait_for_http_ports_with_timeout,
     wait_http_readiness, wait_until_stable,
 };
+pub use snapshot::{
+    NodeStateSource, SnapshotArtifact, SnapshotContext, SnapshotExtension, SnapshotFactory,
+    SnapshotHandle, SnapshotManifest, SnapshotNodeStateAdapter, SnapshotSpec, SnapshotStore,
+};
 pub use sources::{
     ClusterControlProfile, ClusterMode, ExistingCluster, ExternalNodeSource, IntoExistingCluster,
 };

testing-framework/core/src/scenario/snapshot/mod.rs

@@ -0,0 +1,502 @@
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    marker::PhantomData,
+    path::PathBuf,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+
+use crate::scenario::{
+    Application, DynError, NodeClients, PreparedRuntimeExtension, RuntimeExtensionFactory,
+};
+
+mod store;
+
+pub use store::SnapshotStore;
+
+/// Logical name of a snapshot inside the configured snapshot store.
+pub type SnapshotName = String;
+/// Stable identifier for an application-specific snapshot extension artifact.
+pub type SnapshotExtensionId = String;
+/// Stable node identifier used as the key for per-node state in a snapshot.
+pub type SnapshotNodeName = String;
+
+/// Selects which snapshot parts should participate in a save or load operation.
+///
+/// Node selection is intentionally not part of this type. A snapshot can
+/// contain state for many nodes, keyed by node name in
+/// [`SnapshotManifest::node_state`]. When a caller needs one node's local state
+/// as a startup directory, it uses [`NodeStateSource`] instead.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotSpec {
+    include_node_state: bool,
+    extensions: BTreeSet<SnapshotExtensionId>,
+}
+
+impl SnapshotSpec {
+    /// Include only deployer-managed node state.
+    #[must_use]
+    pub fn node_state() -> Self {
+        Self {
+            include_node_state: true,
+            extensions: BTreeSet::new(),
+        }
+    }
+
+    /// Include only one extension artifact.
+    #[must_use]
+    pub fn extension(extension: impl Into<SnapshotExtensionId>) -> Self {
+        Self::extensions([extension])
+    }
+
+    /// Include only the given extension artifacts.
+    #[must_use]
+    pub fn extensions(
+        extensions: impl IntoIterator<Item = impl Into<SnapshotExtensionId>>,
+    ) -> Self {
+        Self {
+            include_node_state: false,
+            extensions: extensions.into_iter().map(Into::into).collect(),
+        }
+    }
+
+    /// Return a copy of this spec without deployer-managed node state.
+    #[must_use]
+    pub fn without_node_state(mut self) -> Self {
+        self.include_node_state = false;
+        self
+    }
+
+    /// Return a copy of this spec with deployer-managed node state included.
+    #[must_use]
+    pub fn with_node_state(mut self) -> Self {
+        self.include_node_state = true;
+        self
+    }
+
+    /// Return a copy of this spec with one additional extension artifact.
+    #[must_use]
+    pub fn with_extension(mut self, extension: impl Into<SnapshotExtensionId>) -> Self {
+        self.extensions.insert(extension.into());
+        self
+    }
+
+    /// Return a copy of this spec with the given extension artifacts.
+    #[must_use]
+    pub fn with_extensions(
+        mut self,
+        extensions: impl IntoIterator<Item = impl Into<SnapshotExtensionId>>,
+    ) -> Self {
+        for extension in extensions {
+            self = self.with_extension(extension);
+        }
+        self
+    }
+
+    /// Whether this spec includes deployer-managed node state.
+    #[must_use]
+    pub fn includes_node_state(&self) -> bool {
+        self.include_node_state
+    }
+
+    /// Whether this spec includes an extension artifact with the given id.
+    #[must_use]
+    pub fn includes_extension(&self, id: &str) -> bool {
+        self.extensions.contains(id)
+    }
+}
+
+impl Default for SnapshotSpec {
+    fn default() -> Self {
+        Self::node_state()
+    }
+}
+
+/// Source for one node's local state when preparing or restoring a runtime dir.
+///
+/// This type is deliberately narrower than a whole snapshot reference.
+/// Extension artifacts and full snapshot loads are addressed by snapshot name
+/// through [`SnapshotHandle::load`]. Node startup/restore needs a concrete
+/// directory, so a snapshot source must name both the snapshot and the source
+/// node.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub enum NodeStateSource {
+    /// Read state for one node from the configured snapshot store.
+    Snapshot {
+        /// Snapshot name under the store root.
+        snapshot: SnapshotName,
+        /// Node key inside that snapshot.
+        node: SnapshotNodeName,
+    },
+    /// Read state directly from an existing node data directory.
+    ExternalDirectory(PathBuf),
+}
+
+impl NodeStateSource {
+    /// Use a node's saved state inside a named snapshot.
+    #[must_use]
+    pub fn snapshot_node(
+        snapshot: impl Into<SnapshotName>,
+        node: impl Into<SnapshotNodeName>,
+    ) -> Self {
+        Self::Snapshot {
+            snapshot: snapshot.into(),
+            node: node.into(),
+        }
+    }
+
+    /// Use an existing node data directory outside the snapshot store.
+    #[must_use]
+    pub fn external_directory(path: impl Into<PathBuf>) -> Self {
+        Self::ExternalDirectory(path.into())
+    }
+}
+
+/// Versioned JSON artifact stored in a snapshot manifest.
+///
+/// Core stores artifacts opaquely. The producer owns the schema encoded in
+/// `payload`, and `version` should be bumped when that schema changes.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotArtifact {
+    /// Producer-defined payload schema version.
+    pub version: u32,
+    /// Small diagnostic/indexing data that can be inspected without decoding
+    /// the full payload.
+    pub metadata: serde_json::Value,
+    /// Producer-defined artifact payload.
+    pub payload: serde_json::Value,
+}
+
+impl SnapshotArtifact {
+    /// Create a snapshot artifact from explicit version, metadata, and payload.
+    #[must_use]
+    pub fn new(version: u32, metadata: serde_json::Value, payload: serde_json::Value) -> Self {
+        Self {
+            version,
+            metadata,
+            payload,
+        }
+    }
+}
+
+/// Manifest file describing the contents of one snapshot.
+///
+/// Node state entries are keyed by node name. Extension entries are keyed by
+/// extension id. The actual node files live beside the manifest in
+/// deployer/application-specific directories.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotManifest {
+    /// Snapshot manifest format version.
+    pub format_version: u32,
+    /// Snapshot name under the configured snapshot store.
+    pub name: SnapshotName,
+    /// Creation timestamp in Unix milliseconds.
+    pub created_unix_millis: u128,
+    /// Per-node state artifacts saved by the deployer/node-state adapter.
+    pub node_state: BTreeMap<SnapshotNodeName, SnapshotArtifact>,
+    /// Application-specific extension artifacts.
+    pub extensions: BTreeMap<SnapshotExtensionId, SnapshotArtifact>,
+}
+
+impl SnapshotManifest {
+    const FORMAT_VERSION: u32 = 1;
+
+    /// Create an empty manifest for a named snapshot.
+    #[must_use]
+    pub fn new(name: impl Into<SnapshotName>) -> Self {
+        Self {
+            format_version: Self::FORMAT_VERSION,
+            name: name.into(),
+            created_unix_millis: unix_millis_now(),
+            node_state: BTreeMap::new(),
+            extensions: BTreeMap::new(),
+        }
+    }
+}
+
+/// Runtime context passed to snapshot extensions.
+pub struct SnapshotContext<E: Application> {
+    deployment: E::Deployment,
+    node_clients: NodeClients<E>,
+}
+
+impl<E: Application> Clone for SnapshotContext<E> {
+    fn clone(&self) -> Self {
+        Self {
+            deployment: self.deployment.clone(),
+            node_clients: self.node_clients.clone(),
+        }
+    }
+}
+
+impl<E: Application> SnapshotContext<E> {
+    /// Create snapshot extension context from the active deployment and
+    /// clients.
+    #[must_use]
+    pub fn new(deployment: E::Deployment, node_clients: NodeClients<E>) -> Self {
+        Self {
+            deployment,
+            node_clients,
+        }
+    }
+
+    /// Active deployment for the scenario.
+    #[must_use]
+    pub const fn deployment(&self) -> &E::Deployment {
+        &self.deployment
+    }
+
+    /// Node clients available in the active scenario.
+    #[must_use]
+    pub const fn node_clients(&self) -> &NodeClients<E> {
+        &self.node_clients
+    }
+}
+
+/// Application-specific snapshot extension.
+///
+/// Extensions save and load artifacts that are not owned by the deployer/node
+/// runtime itself, such as test harness wallet state. Implementations should be
+/// deterministic and treat `spec` as a filter for optional sub-parts they own.
+#[async_trait]
+pub trait SnapshotExtension<E: Application>: Send + Sync {
+    /// Stable id used as the key in [`SnapshotManifest::extensions`].
+    fn id(&self) -> &'static str;
+
+    /// Save this extension's artifact for the active snapshot operation.
+    ///
+    /// Returning `Ok(None)` means this extension has no state to write.
+    async fn save(
+        &self,
+        context: &SnapshotContext<E>,
+        spec: &SnapshotSpec,
+    ) -> Result<Option<SnapshotArtifact>, DynError>;
+
+    /// Load this extension's artifact from the active snapshot operation.
+    async fn load(
+        &self,
+        context: &SnapshotContext<E>,
+        artifact: &SnapshotArtifact,
+        spec: &SnapshotSpec,
+    ) -> Result<(), DynError>;
+}
+
+/// Deployer-owned adapter for saving and loading node runtime state.
+///
+/// Core does not know which files make up a node's durable state. The adapter
+/// bridges that deployer/application detail to the generic snapshot manifest.
+#[async_trait]
+pub trait SnapshotNodeStateAdapter<E: Application>: Send + Sync {
+    /// Save node state for the active deployment into the named snapshot.
+    async fn save_node_state(
+        &self,
+        snapshot: &str,
+        spec: &SnapshotSpec,
+    ) -> Result<BTreeMap<SnapshotNodeName, SnapshotArtifact>, DynError>;
+
+    /// Load node state from a named snapshot manifest into the active
+    /// deployment.
+    async fn load_node_state(
+        &self,
+        snapshot: &str,
+        manifest: &SnapshotManifest,
+        spec: &SnapshotSpec,
+    ) -> Result<(), DynError>;
+
+    /// Resolve one node's state to a local startup directory.
+    ///
+    /// Deployers that cannot expose a local startup directory may keep the
+    /// default unsupported implementation.
+    async fn prepare_node_state_source_dir(
+        &self,
+        source: &NodeStateSource,
+    ) -> Result<PathBuf, DynError> {
+        Err(format!(
+            "preparing node state source '{source:?}' as a startup directory is not supported by this deployer"
+        )
+        .into())
+    }
+
+    /// Persist a snapshot manifest.
+    async fn write_manifest(&self, manifest: &SnapshotManifest) -> Result<(), DynError>;
+
+    /// Read a snapshot manifest by name.
+    async fn read_manifest(&self, snapshot: &str) -> Result<SnapshotManifest, DynError>;
+}
+
+/// Prepared snapshot runtime extension for an active scenario.
+pub struct SnapshotHandle<E: Application> {
+    node_state: Arc<dyn SnapshotNodeStateAdapter<E>>,
+    context: SnapshotContext<E>,
+    extensions: Vec<Arc<dyn SnapshotExtension<E>>>,
+    _phantom: PhantomData<E>,
+}
+
+/// Factory that installs snapshot support into a scenario runtime.
+pub struct SnapshotFactory<E: Application> {
+    node_state: Arc<dyn SnapshotNodeStateAdapter<E>>,
+    extensions: Vec<Arc<dyn SnapshotExtension<E>>>,
+}
+
+impl<E: Application> SnapshotFactory<E> {
+    /// Create a snapshot factory with a node-state adapter and no extensions.
+    #[must_use]
+    pub fn new(node_state: Arc<dyn SnapshotNodeStateAdapter<E>>) -> Self {
+        Self {
+            node_state,
+            extensions: Vec::new(),
+        }
+    }
+
+    /// Register one application-specific snapshot extension.
+    #[must_use]
+    pub fn with_extension(mut self, extension: Arc<dyn SnapshotExtension<E>>) -> Self {
+        self.extensions.push(extension);
+        self
+    }
+
+    /// Register multiple application-specific snapshot extensions.
+    #[must_use]
+    pub fn with_extensions(
+        mut self,
+        extensions: impl IntoIterator<Item = Arc<dyn SnapshotExtension<E>>>,
+    ) -> Self {
+        self.extensions.extend(extensions);
+        self
+    }
+}
+
+#[async_trait]
+impl<E: Application> RuntimeExtensionFactory<E> for SnapshotFactory<E> {
+    async fn prepare(
+        &self,
+        deployment: &E::Deployment,
+        node_clients: NodeClients<E>,
+    ) -> Result<PreparedRuntimeExtension, DynError> {
+        Ok(PreparedRuntimeExtension::new(SnapshotHandle::new(
+            Arc::clone(&self.node_state),
+            SnapshotContext::new(deployment.clone(), node_clients),
+            self.extensions.clone(),
+        )))
+    }
+}
+
+impl<E: Application> Clone for SnapshotHandle<E> {
+    fn clone(&self) -> Self {
+        Self {
+            node_state: Arc::clone(&self.node_state),
+            context: self.context.clone(),
+            extensions: self.extensions.clone(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<E: Application> SnapshotHandle<E> {
+    /// Create a prepared snapshot handle.
+    #[must_use]
+    pub fn new(
+        node_state: Arc<dyn SnapshotNodeStateAdapter<E>>,
+        context: SnapshotContext<E>,
+        extensions: Vec<Arc<dyn SnapshotExtension<E>>>,
+    ) -> Self {
+        Self {
+            node_state,
+            context,
+            extensions,
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Save a named snapshot according to `spec`.
+    ///
+    /// This writes a new manifest containing the requested node-state and
+    /// extension artifacts. Lifecycle concerns such as stopping nodes before
+    /// saving are intentionally owned by the caller/deployer, not this method.
+    pub async fn save(
+        &self,
+        name: impl Into<SnapshotName>,
+        spec: SnapshotSpec,
+    ) -> Result<SnapshotName, DynError> {
+        let snapshot = name.into();
+        let mut manifest = SnapshotManifest::new(snapshot.clone());
+
+        if spec.includes_node_state() {
+            manifest.node_state = self.node_state.save_node_state(&snapshot, &spec).await?;
+        }
+
+        for extension in &self.extensions {
+            if !spec.includes_extension(extension.id()) {
+                continue;
+            }
+
+            if let Some(artifact) = extension.save(&self.context, &spec).await? {
+                manifest
+                    .extensions
+                    .insert(extension.id().to_owned(), artifact);
+            }
+        }
+
+        self.node_state.write_manifest(&manifest).await?;
+
+        Ok(snapshot)
+    }
+
+    /// Load requested snapshot parts into the active scenario runtime.
+    ///
+    /// Loading node state here delegates to the deployer's node-state adapter.
+    /// Preparing a local node startup directory is a separate operation handled
+    /// by [`Self::prepare_node_state_source_dir`].
+    pub async fn load(
+        &self,
+        snapshot: impl AsRef<str>,
+        spec: SnapshotSpec,
+    ) -> Result<(), DynError> {
+        let snapshot = snapshot.as_ref();
+        let manifest = self.node_state.read_manifest(snapshot).await?;
+
+        if spec.includes_node_state() {
+            self.node_state
+                .load_node_state(snapshot, &manifest, &spec)
+                .await?;
+        }
+
+        for extension in &self.extensions {
+            if !spec.includes_extension(extension.id()) {
+                continue;
+            }
+
+            let Some(artifact) = manifest.extensions.get(extension.id()) else {
+                return Err(format!(
+                    "snapshot '{snapshot}' does not contain extension '{}'",
+                    extension.id()
+                )
+                .into());
+            };
+
+            extension.load(&self.context, artifact, &spec).await?;
+        }
+
+        Ok(())
+    }
+
+    /// Resolve one node's saved state to a local startup directory.
+    ///
+    /// This is used when a node process should be started from existing local
+    /// state. It does not load extension artifacts.
+    pub async fn prepare_node_state_source_dir(
+        &self,
+        source: NodeStateSource,
+    ) -> Result<PathBuf, DynError> {
+        self.node_state.prepare_node_state_source_dir(&source).await
+    }
+}
+
+fn unix_millis_now() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |duration| duration.as_millis())
+}

testing-framework/core/src/scenario/snapshot/store.rs

@@ -0,0 +1,364 @@
+use std::{
+    fs, io,
+    io::ErrorKind,
+    path::{Component, Path, PathBuf},
+};
+
+use crate::scenario::{DynError, NodeStateSource, SnapshotArtifact, SnapshotManifest};
+
+const MANIFEST_FILE: &str = "snapshot-manifest.json";
+
+#[derive(Clone, Debug)]
+/// Filesystem-backed snapshot store.
+///
+/// The store owns the on-disk layout under a root directory. It is
+/// intentionally generic over the concrete node state directories: callers
+/// provide the list of subdirectories that make up durable node state.
+pub struct SnapshotStore {
+    root_dir: PathBuf,
+}
+
+impl SnapshotStore {
+    /// Create a snapshot store rooted at `root_dir`.
+    #[must_use]
+    pub fn new(root_dir: impl Into<PathBuf>) -> Self {
+        Self {
+            root_dir: root_dir.into(),
+        }
+    }
+
+    /// Root directory containing named snapshots.
+    #[must_use]
+    pub fn root_dir(&self) -> &Path {
+        &self.root_dir
+    }
+
+    /// Save selected node state subdirectories into a named snapshot.
+    ///
+    /// The destination layout is `<root>/<snapshot_name>/<node_name>/<subdir>`.
+    /// If a manifest already exists it is updated; an invalid existing manifest
+    /// fails the save instead of being overwritten.
+    pub fn save_node_dirs(
+        &self,
+        snapshot_name: &str,
+        node_name: &str,
+        runtime_dir: &Path,
+        state_subdirs: &[&str],
+    ) -> Result<String, DynError> {
+        validate_path_component(snapshot_name, "Snapshot name")?;
+        validate_path_component(node_name, "Node name")?;
+
+        let mut manifest = self.read_manifest_or_new(snapshot_name)?;
+        let destination = self.snapshot_node_dir(snapshot_name, node_name);
+        clear_dir(&destination)?;
+
+        for dir_name in state_subdirs {
+            copy_dir_recursive(&runtime_dir.join(dir_name), &destination.join(dir_name)).map_err(
+                |source| {
+                    io::Error::other(format!(
+                        "failed to copy node `{node_name}` state from '{}' to '{}': {source}",
+                        runtime_dir.display(),
+                        destination.display()
+                    ))
+                },
+            )?;
+        }
+
+        manifest.node_state.insert(
+            node_name.to_owned(),
+            SnapshotArtifact::new(
+                1,
+                serde_json::json!({
+                    "state_subdirs": state_subdirs,
+                }),
+                serde_json::json!({
+                    "artifact": node_name,
+                }),
+            ),
+        );
+        self.write_manifest(&manifest)?;
+
+        Ok(snapshot_name.to_owned())
+    }
+
+    /// Resolve and validate a node-state source directory.
+    ///
+    /// This does not copy files. It only returns the source path after checking
+    /// that all required state subdirectories exist.
+    pub fn prepare_node_dir(
+        &self,
+        source: &NodeStateSource,
+        required_subdirs: &[&str],
+    ) -> Result<PathBuf, DynError> {
+        let source_dir = match source {
+            NodeStateSource::Snapshot { snapshot, node } => self.snapshot_node_dir(snapshot, node),
+            NodeStateSource::ExternalDirectory(path) => path.clone(),
+        };
+
+        ensure_subdirs(&source_dir, required_subdirs)?;
+
+        Ok(source_dir)
+    }
+
+    /// Restore node state subdirectories into a runtime directory.
+    ///
+    /// Each target subdirectory is cleared before the corresponding source
+    /// subdirectory is copied.
+    pub fn restore_node_dirs(
+        &self,
+        source: &NodeStateSource,
+        runtime_dir: &Path,
+        state_subdirs: &[&str],
+    ) -> Result<(), DynError> {
+        let source_dir = self.prepare_node_dir(source, state_subdirs)?;
+
+        for dir_name in state_subdirs {
+            let runtime_state_dir = runtime_dir.join(dir_name);
+            clear_dir(&runtime_state_dir)?;
+            copy_dir_recursive(&source_dir.join(dir_name), &runtime_state_dir)?;
+        }
+
+        Ok(())
+    }
+
+    /// Save or replace one extension artifact in a named snapshot manifest.
+    ///
+    /// If the snapshot manifest does not exist it is created. Invalid existing
+    /// manifests fail the save instead of being overwritten.
+    pub fn save_extension_artifact(
+        &self,
+        snapshot_name: &str,
+        extension_id: &str,
+        artifact: SnapshotArtifact,
+    ) -> Result<String, DynError> {
+        validate_path_component(snapshot_name, "Snapshot name")?;
+        validate_path_component(extension_id, "Snapshot extension id")?;
+
+        let mut manifest = self.read_manifest_or_new(snapshot_name)?;
+
+        manifest
+            .extensions
+            .insert(extension_id.to_owned(), artifact);
+        self.write_manifest(&manifest)?;
+
+        Ok(snapshot_name.to_owned())
+    }
+
+    /// Load one extension artifact from a named snapshot manifest.
+    pub fn load_extension_artifact(
+        &self,
+        snapshot_name: &str,
+        extension_id: &str,
+    ) -> Result<SnapshotArtifact, DynError> {
+        validate_path_component(snapshot_name, "Snapshot name")?;
+        validate_path_component(extension_id, "Snapshot extension id")?;
+
+        let manifest = self.read_manifest(snapshot_name)?;
+        manifest
+            .extensions
+            .get(extension_id)
+            .cloned()
+            .ok_or_else(|| {
+                format!("snapshot '{snapshot_name}' does not contain extension '{extension_id}'")
+                    .into()
+            })
+    }
+
+    /// Read a snapshot manifest by snapshot name.
+    pub fn read_manifest(&self, snapshot: &str) -> Result<SnapshotManifest, DynError> {
+        let bytes = fs::read(self.manifest_path(snapshot))?;
+        Ok(serde_json::from_slice(&bytes)?)
+    }
+
+    fn read_manifest_or_new(&self, snapshot: &str) -> Result<SnapshotManifest, DynError> {
+        match fs::read(self.manifest_path(snapshot)) {
+            Ok(bytes) => Ok(serde_json::from_slice(&bytes)?),
+            Err(error) if error.kind() == ErrorKind::NotFound => {
+                Ok(SnapshotManifest::new(snapshot))
+            }
+            Err(error) => Err(error.into()),
+        }
+    }
+
+    /// Write a snapshot manifest to its canonical path.
+    pub fn write_manifest(&self, manifest: &SnapshotManifest) -> Result<(), DynError> {
+        validate_path_component(&manifest.name, "Snapshot name")?;
+
+        let path = self.manifest_path(&manifest.name);
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent)?;
+        }
+        fs::write(path, serde_json::to_vec_pretty(manifest)?)?;
+        Ok(())
+    }
+
+    /// Return the canonical directory for one node inside a named snapshot.
+    #[must_use]
+    pub fn snapshot_node_dir(&self, snapshot: &str, node_name: &str) -> PathBuf {
+        self.root_dir.join(snapshot).join(node_name)
+    }
+
+    /// Return the canonical manifest path for a named snapshot.
+    #[must_use]
+    pub fn manifest_path(&self, snapshot: &str) -> PathBuf {
+        self.root_dir.join(snapshot).join(MANIFEST_FILE)
+    }
+}
+
+fn validate_path_component(value: &str, field_name: &str) -> Result<(), DynError> {
+    if value.trim().is_empty() {
+        return Err(format!("{field_name} cannot be empty").into());
+    }
+
+    let path = Path::new(value);
+    let mut components = path.components();
+
+    match (components.next(), components.next()) {
+        (Some(Component::Normal(_)), None) if !path.is_absolute() => Ok(()),
+        _ => {
+            Err(format!("{field_name} must be a single safe path component, got `{value}`").into())
+        }
+    }
+}
+
+fn ensure_subdirs(path: &Path, required_subdirs: &[&str]) -> Result<(), DynError> {
+    for dir_name in required_subdirs {
+        let state_dir = path.join(dir_name);
+        if !state_dir.is_dir() {
+            return Err(format!(
+                "snapshot state path '{}' is missing or is not a directory",
+                state_dir.display()
+            )
+            .into());
+        }
+    }
+
+    Ok(())
+}
+
+fn clear_dir(path: &Path) -> io::Result<()> {
+    if path.exists() {
+        if !path.is_dir() {
+            return Err(io::Error::other(format!(
+                "path exists but is not a directory: '{}'",
+                path.display()
+            )));
+        }
+        fs::remove_dir_all(path)?;
+    }
+    fs::create_dir_all(path)
+}
+
+fn copy_dir_recursive(src: &Path, dst: &Path) -> io::Result<()> {
+    if !src.is_dir() {
+        return Err(io::Error::other(format!(
+            "directory '{}' is missing",
+            src.display()
+        )));
+    }
+
+    fs::create_dir_all(dst)?;
+
+    for entry in fs::read_dir(src)? {
+        let entry = entry?;
+        let source_path = entry.path();
+        let destination_path = dst.join(entry.file_name());
+        let metadata = entry.metadata()?;
+
+        if metadata.is_dir() {
+            copy_dir_recursive(&source_path, &destination_path)?;
+        } else if metadata.is_file() {
+            if let Some(parent) = destination_path.parent() {
+                fs::create_dir_all(parent)?;
+            }
+            fs::copy(&source_path, &destination_path)?;
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use tempfile::TempDir;
+
+    use super::*;
+
+    fn write_file(path: &Path, contents: &str) {
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent).expect("parent dir should be created");
+        }
+        fs::write(path, contents).expect("test file should be written");
+    }
+
+    #[test]
+    fn node_state_and_extension_artifacts_share_manifest() {
+        let root = TempDir::new().expect("snapshot root should be created");
+        let runtime = TempDir::new().expect("runtime dir should be created");
+        let store = SnapshotStore::new(root.path());
+
+        write_file(&runtime.path().join("db").join("state.txt"), "db");
+
+        store
+            .save_extension_artifact(
+                "snapshot",
+                "wallet",
+                SnapshotArtifact::new(
+                    1,
+                    serde_json::json!({ "wallet_count": 1 }),
+                    serde_json::json!({ "wallets": ["WALLET_1A"] }),
+                ),
+            )
+            .expect("extension artifact should save");
+
+        store
+            .save_node_dirs("snapshot", "NODE_1", runtime.path(), &["db"])
+            .expect("node state should save");
+
+        let manifest = store
+            .read_manifest("snapshot")
+            .expect("manifest should be readable");
+
+        assert!(manifest.node_state.contains_key("NODE_1"));
+        assert_eq!(
+            manifest
+                .extensions
+                .get("wallet")
+                .expect("wallet artifact should remain in manifest")
+                .metadata,
+            serde_json::json!({ "wallet_count": 1 })
+        );
+        assert!(
+            store
+                .snapshot_node_dir("snapshot", "NODE_1")
+                .join("db")
+                .join("state.txt")
+                .is_file()
+        );
+    }
+
+    #[test]
+    fn saving_node_state_does_not_overwrite_invalid_manifest() {
+        let root = TempDir::new().expect("snapshot root should be created");
+        let runtime = TempDir::new().expect("runtime dir should be created");
+        let store = SnapshotStore::new(root.path());
+
+        write_file(&runtime.path().join("db").join("state.txt"), "db");
+        write_file(&store.manifest_path("snapshot"), "not json");
+
+        let error = store
+            .save_node_dirs("snapshot", "NODE_1", runtime.path(), &["db"])
+            .expect_err("invalid manifest should fail the save");
+
+        assert!(error.to_string().contains("expected ident"));
+        assert_eq!(
+            fs::read_to_string(store.manifest_path("snapshot"))
+                .expect("manifest should still exist"),
+            "not json"
+        );
+        assert!(
+            !store.snapshot_node_dir("snapshot", "NODE_1").exists(),
+            "node state should not be copied when manifest cannot be read"
+        );
+    }
+}