[WIP] composefs-splitfdstream: new crate

Cargo.toml

@@ -21,6 +21,7 @@ composefs-ioctls = { version = "0.3.0", path = "crates/composefs-ioctls", defaul
 composefs-oci = { version = "0.3.0", path = "crates/composefs-oci", default-features = false }
 composefs-boot = { version = "0.3.0", path = "crates/composefs-boot", default-features = false }
 composefs-http = { version = "0.3.0", path = "crates/composefs-http", default-features = false }
+splitfdstream = { version = "0.3.0", path = "crates/splitfdstream", default-features = false }
 
 [profile.dev.package.sha2]
 # this is *really* slow otherwise

crates/cfsctl/Cargo.toml

@@ -13,7 +13,7 @@ version.workspace = true
 [features]
 default = ['pre-6.15', 'oci']
 http = ['composefs-http']
-oci = ['composefs-oci']
+oci = ['composefs-oci', 'splitfdstream']
 rhel9 = ['composefs/rhel9']
 'pre-6.15' = ['composefs/pre-6.15']
 
@@ -25,6 +25,7 @@ composefs = { workspace = true }
 composefs-boot = { workspace = true }
 composefs-oci = { workspace = true, optional = true }
 composefs-http = { workspace = true, optional = true }
+splitfdstream = { workspace = true, optional = true }
 env_logger = { version = "0.11.0", default-features = false }
 hex = { version = "0.4.0", default-features = false }
 rustix = { version = "1.0.0", default-features = false, features = ["fs", "process"] }

crates/cfsctl/src/main.rs

@@ -84,6 +84,16 @@ enum OciCommand {
         digest: String,
         name: Option<String>,
     },
+    /// Imports a complete image from a splitfdstream server into the repository.
+    ImportImageSplitfdstream {
+        /// Path to the splitfdstream server socket
+        socket: PathBuf,
+        /// The image ID (manifest digest or tag)
+        image_id: String,
+        /// Tag name for imported image
+        #[clap(long)]
+        tag: Option<String>,
+    },
     /// Lists the contents of a tar stream
     LsLayer {
         /// the name of the stream to list, either a stream ID in format oci-config-<hash_type>:<hash_digest> or a reference in 'ref/'
@@ -316,6 +326,27 @@ where
                 )?;
                 println!("{}", object_id.to_id());
             }
+            OciCommand::ImportImageSplitfdstream {
+                socket,
+                image_id,
+                tag,
+            } => {
+                let result = composefs_oci::import_complete_image_from_splitfdstream(
+                    &Arc::new(repo),
+                    &socket,
+                    &image_id,
+                    tag.as_deref(),
+                )?;
+
+                println!("Imported complete image:");
+                println!("  Manifest: {}", result.manifest_digest);
+                println!("  Config:   {}", result.config_digest);
+                println!("  Layers:   {}", result.layers_imported);
+                println!("  Size:     {} bytes", result.total_size_bytes);
+                if let Some(tag_name) = tag {
+                    println!("  Tagged:   {}", tag_name);
+                }
+            }
             OciCommand::LsLayer { name } => {
                 composefs_oci::ls_layer(&repo, &name)?;
             }

crates/composefs-oci/Cargo.toml

@@ -16,11 +16,13 @@ fn-error-context = "0.2"
 async-compression = { version = "0.4.0", default-features = false, features = ["tokio", "zstd", "gzip"] }
 bytes = { version = "1", default-features = false }
 composefs = { workspace = true }
+splitfdstream = { workspace = true }
 containers-image-proxy = { version = "0.9.2", default-features = false }
 hex = { version = "0.4.0", default-features = false }
 indicatif = { version = "0.17.0", default-features = false, features = ["tokio"] }
 oci-spec = { version = "0.8.0", default-features = false }
-rustix = { version = "1.0.0", features = ["fs"] }
+rustix = { version = "1.0.0", features = ["fs", "net"] }
+serde_json = { version = "1.0", default-features = false, features = ["std"] }
 sha2 = { version = "0.10.1", default-features = false }
 tar = { version = "0.4.38", default-features = false }
 tokio = { version = "1.24.2", features = ["rt-multi-thread"] }

crates/composefs-oci/src/lib.rs

@@ -15,8 +15,13 @@
 pub mod image;
 pub mod oci_image;
 pub mod skopeo;
+pub mod splitfdstream;
 pub mod tar;
 
+pub use splitfdstream::{
+    import_complete_image_from_splitfdstream, import_from_splitfdstream, CompleteImageImportResult,
+};
+
 use std::{collections::HashMap, io::Read, sync::Arc};
 
 use anyhow::{bail, ensure, Context, Result};
@@ -39,7 +44,7 @@ pub use skopeo::{pull_image, PullResult};
 
 type ContentAndVerity<ObjectID> = (String, ObjectID);
 
-fn layer_identifier(diff_id: &str) -> String {
+pub(crate) fn layer_identifier(diff_id: &str) -> String {
     format!("oci-layer-{diff_id}")
 }
 

crates/composefs-oci/src/oci_image.rs

@@ -530,6 +530,53 @@ pub fn write_manifest<ObjectID: FsVerityHashValue>(
     Ok((manifest_digest.to_string(), id))
 }
 
+/// Writes a manifest to the repository from raw JSON bytes.
+///
+/// Unlike [`write_manifest`], this preserves the exact JSON bytes from the
+/// original source, avoiding digest mismatches from re-serialization.
+pub fn write_manifest_raw<ObjectID: FsVerityHashValue>(
+    repo: &Arc<Repository<ObjectID>>,
+    manifest_json: &[u8],
+    manifest_digest: &str,
+    config_verity: &ObjectID,
+    layer_verities: &HashMap<Box<str>, ObjectID>,
+    reference: Option<&str>,
+) -> Result<(String, ObjectID)> {
+    let content_id = manifest_identifier(manifest_digest);
+
+    if let Some(verity) = repo.has_stream(&content_id)? {
+        if let Some(name) = reference {
+            tag_image(repo, manifest_digest, name)?;
+        }
+        return Ok((manifest_digest.to_string(), verity));
+    }
+
+    let computed = hash(manifest_json);
+    ensure!(
+        manifest_digest == computed,
+        "Manifest digest mismatch: expected {manifest_digest}, got {computed}"
+    );
+
+    let manifest: ImageManifest =
+        serde_json::from_slice(manifest_json).context("parsing manifest JSON")?;
+
+    let mut stream = repo.create_stream(OCI_MANIFEST_CONTENT_TYPE);
+
+    let config_key = format!("config:{}", manifest.config().digest());
+    stream.add_named_stream_ref(&config_key, config_verity);
+
+    for (diff_id, verity) in layer_verities {
+        stream.add_named_stream_ref(diff_id, verity);
+    }
+
+    stream.write_external(manifest_json)?;
+
+    let oci_ref = reference.map(oci_ref_path);
+    let id = repo.write_stream(stream, &content_id, oci_ref.as_deref())?;
+
+    Ok((manifest_digest.to_string(), id))
+}
+
 /// Checks if a manifest exists.
 pub fn has_manifest<ObjectID: FsVerityHashValue>(
     repo: &Repository<ObjectID>,