composefs-oci: add support for splitfdstream

Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Giuseppe Scrivano <gscrivan@redhat.com>

Author: giuseppe

Cargo.toml

@@ -22,6 +22,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/composefs-oci/Cargo.toml

@@ -16,11 +16,12 @@ 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 = { version = "1.0", default-features = false, features = ["derive"] }
 serde_json = { version = "1.0", default-features = false, features = ["std"] }
 sha2 = { version = "0.10.1", default-features = false }

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

@@ -594,6 +594,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>,