splitstream: Rework file format

This changes the splitstream format a bit. The primary differences are:

 * The header is not compressed
 * All referenced fs-verity objects are stored in the header, including
   external chunks, mapped splitstreams and (a new feature) references
   that are not used in chunks.
 * The mapping table is separate from the reference table (and generally
   smaller), and indexes into it.
 * There is a magic value to detect the file format.
 * There is a magic content type to detect the type wrapped in the stream.
 * We store a tag for what ObjectID format is used
 * The total size of the stream is stored in the header.

The ability to reference file objects in the repo even if they are not
part of the splitstream "content" will be useful for the ostree
support to reference file content objects.

This change also allows More efficient GC enumeration, because we
don't have to parse the entire splitstream to find the referenced
objects.

Author: alexlarsson

crates/cfsctl/src/main.rs

@@ -180,7 +180,7 @@ async fn main() -> Result<()> {
             }
         }
         Command::Cat { name } => {
-            repo.merge_splitstream(&name, None, &mut std::io::stdout())?;
+            repo.merge_splitstream(&name, None, None, &mut std::io::stdout())?;
         }
         Command::ImportImage { reference } => {
             let image_id = repo.import_image(&reference, &mut std::io::stdin())?;

crates/composefs-http/src/lib.rs

@@ -13,9 +13,7 @@ use sha2::{Digest, Sha256};
 use tokio::task::JoinSet;
 
 use composefs::{
-    fsverity::FsVerityHashValue,
-    repository::Repository,
-    splitstream::{DigestMapEntry, SplitStreamReader},
+    fsverity::FsVerityHashValue, repository::Repository, splitstream::SplitStreamReader,
     util::Sha256Digest,
 };
 
@@ -61,7 +59,7 @@ impl<ObjectID: FsVerityHashValue> Downloader<ObjectID> {
     }
 
     fn open_splitstream(&self, id: &ObjectID) -> Result<SplitStreamReader<File, ObjectID>> {
-        SplitStreamReader::new(File::from(self.repo.open_object(id)?))
+        SplitStreamReader::new(File::from(self.repo.open_object(id)?), None)
     }
 
     fn read_object(&self, id: &ObjectID) -> Result<Vec<u8>> {
@@ -107,7 +105,7 @@ impl<ObjectID: FsVerityHashValue> Downloader<ObjectID> {
 
             // this part is fast: it only touches the header
             let mut reader = self.open_splitstream(&id)?;
-            for DigestMapEntry { verity, body } in &reader.refs.map {
+            for (body, verity) in reader.iter_mappings() {
                 match splitstreams.insert(verity.clone(), Some(*body)) {
                     // This is the (normal) case if we encounter a splitstream we didn't see yet...
                     None => {

crates/composefs-oci/src/image.rs

@@ -9,6 +9,7 @@ use composefs::{
     tree::{Directory, FileSystem, Inode, Leaf},
 };
 
+use crate::skopeo::{OCI_CONFIG_CONTENT_TYPE, TAR_LAYER_CONTENT_TYPE};
 use crate::tar::{TarEntry, TarItem};
 
 pub fn process_entry<ObjectID: FsVerityHashValue>(
@@ -74,14 +75,19 @@ pub fn create_filesystem<ObjectID: FsVerityHashValue>(
 ) -> Result<FileSystem<ObjectID>> {
     let mut filesystem = FileSystem::default();
 
-    let mut config_stream = repo.open_stream(config_name, config_verity)?;
+    let mut config_stream =
+        repo.open_stream(config_name, config_verity, Some(OCI_CONFIG_CONTENT_TYPE))?;
     let config = ImageConfiguration::from_reader(&mut config_stream)?;
 
     for diff_id in config.rootfs().diff_ids() {
         let layer_sha256 = super::sha256_from_digest(diff_id)?;
         let layer_verity = config_stream.lookup(&layer_sha256)?;
 
-        let mut layer_stream = repo.open_stream(&hex::encode(layer_sha256), Some(layer_verity))?;
+        let mut layer_stream = repo.open_stream(
+            &hex::encode(layer_sha256),
+            Some(layer_verity),
+            Some(TAR_LAYER_CONTENT_TYPE),
+        )?;
         while let Some(entry) = crate::tar::get_entry(&mut layer_stream)? {
             process_entry(&mut filesystem, entry)?;
         }

crates/composefs-oci/src/lib.rs

@@ -15,6 +15,7 @@ use composefs::{
     util::{parse_sha256, Sha256Digest},
 };
 
+use crate::skopeo::{OCI_CONFIG_CONTENT_TYPE, TAR_LAYER_CONTENT_TYPE};
 use crate::tar::get_entry;
 
 type ContentAndVerity<ObjectID> = (Sha256Digest, ObjectID);
@@ -39,14 +40,19 @@ pub fn import_layer<ObjectID: FsVerityHashValue>(
     name: Option<&str>,
     tar_stream: &mut impl Read,
 ) -> Result<ObjectID> {
-    repo.ensure_stream(sha256, |writer| tar::split(tar_stream, writer), name)
+    repo.ensure_stream(
+        sha256,
+        TAR_LAYER_CONTENT_TYPE,
+        |writer| tar::split(tar_stream, writer),
+        name,
+    )
 }
 
 pub fn ls_layer<ObjectID: FsVerityHashValue>(
     repo: &Repository<ObjectID>,
     name: &str,
 ) -> Result<()> {
-    let mut split_stream = repo.open_stream(name, None)?;
+    let mut split_stream = repo.open_stream(name, None, Some(TAR_LAYER_CONTENT_TYPE))?;
 
     while let Some(entry) = get_entry(&mut split_stream)? {
         println!("{entry}");
@@ -81,9 +87,9 @@ pub fn open_config<ObjectID: FsVerityHashValue>(
                 .with_context(|| format!("Object {name} is unknown to us"))?
         }
     };
-    let mut stream = repo.open_stream(name, Some(id))?;
+    let mut stream = repo.open_stream(name, Some(id), Some(OCI_CONFIG_CONTENT_TYPE))?;
     let config = ImageConfiguration::from_reader(&mut stream)?;
-    Ok((config, stream.refs))
+    Ok((config, stream.get_mappings()))
 }
 
 fn hash(bytes: &[u8]) -> Sha256Digest {
@@ -104,7 +110,7 @@ pub fn open_config_shallow<ObjectID: FsVerityHashValue>(
             // we need to manually check the content digest
             let expected_hash = parse_sha256(name)
                 .context("Containers must be referred to by sha256 if verity is missing")?;
-            let mut stream = repo.open_stream(name, None)?;
+            let mut stream = repo.open_stream(name, None, Some(OCI_CONFIG_CONTENT_TYPE))?;
             let mut raw_config = vec![];
             stream.read_to_end(&mut raw_config)?;
             ensure!(hash(&raw_config) == expected_hash, "Data integrity issue");
@@ -121,7 +127,8 @@ pub fn write_config<ObjectID: FsVerityHashValue>(
     let json = config.to_string()?;
     let json_bytes = json.as_bytes();
     let sha256 = hash(json_bytes);
-    let mut stream = repo.create_stream(Some(sha256), Some(refs));
+    let mut stream = repo.create_stream(OCI_CONFIG_CONTENT_TYPE, Some(sha256));
+    stream.add_sha256_mappings(refs);
     stream.write_inline(json_bytes);
     let id = repo.write_stream(stream, None)?;
     Ok((sha256, id))
@@ -199,7 +206,7 @@ mod test {
         let id = import_layer(&repo, &layer_id, Some("name"), &mut layer.as_slice()).unwrap();
 
         let mut dump = String::new();
-        let mut split_stream = repo.open_stream("refs/name", Some(&id)).unwrap();
+        let mut split_stream = repo.open_stream("refs/name", Some(&id), None).unwrap();
         while let Some(entry) = tar::get_entry(&mut split_stream).unwrap() {
             writeln!(dump, "{entry}").unwrap();
         }

crates/composefs-oci/src/skopeo.rs

@@ -10,12 +10,13 @@ use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest, MediaType};
 use rustix::process::geteuid;
 use tokio::{io::AsyncReadExt, sync::Semaphore};
 
-use composefs::{
-    fsverity::FsVerityHashValue, repository::Repository, splitstream::DigestMap, util::Sha256Digest,
-};
+use composefs::{fsverity::FsVerityHashValue, repository::Repository, util::Sha256Digest};
 
 use crate::{sha256_from_descriptor, sha256_from_digest, tar::split_async, ContentAndVerity};
 
+pub const TAR_LAYER_CONTENT_TYPE: u64 = 0x2a037edfcae1ffea;
+pub const OCI_CONFIG_CONTENT_TYPE: u64 = 0x44218c839727a80b;
+
 struct ImageOp<ObjectID: FsVerityHashValue> {
     repo: Arc<Repository<ObjectID>>,
     proxy: ImageProxy,
@@ -78,7 +79,9 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
             self.progress
                 .println(format!("Fetching layer {}", hex::encode(layer_sha256)))?;
 
-            let mut splitstream = self.repo.create_stream(Some(layer_sha256), None);
+            let mut splitstream = self
+                .repo
+                .create_stream(TAR_LAYER_CONTENT_TYPE, Some(layer_sha256));
             match descriptor.media_type() {
                 MediaType::ImageLayer => {
                     split_async(progress, &mut splitstream).await?;
@@ -155,15 +158,15 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
                 entries.push((layer_sha256, future));
             }
 
+            let mut splitstream = self
+                .repo
+                .create_stream(OCI_CONFIG_CONTENT_TYPE, Some(config_sha256));
+
             // Collect the results.
-            let mut config_maps = DigestMap::new();
             for (layer_sha256, future) in entries {
-                config_maps.insert(&layer_sha256, &future.await??);
+                splitstream.add_sha256_mapping(&layer_sha256, &future.await??);
             }
 
-            let mut splitstream = self
-                .repo
-                .create_stream(Some(config_sha256), Some(config_maps));
             splitstream.write_inline(&raw_config);
             let config_id = self.repo.write_stream(splitstream, None)?;
 

crates/composefs/src/fsverity/hashvalue.rs

@@ -2,6 +2,7 @@ use core::{fmt, hash::Hash};
 
 use hex::FromHexError;
 use sha2::{digest::FixedOutputReset, digest::Output, Digest, Sha256, Sha512};
+use std::cmp::Ord;
 use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Unaligned};
 
 pub trait FsVerityHashValue
@@ -12,6 +13,7 @@ where
     Self: Hash + Eq,
     Self: fmt::Debug,
     Self: Send + Sync + Unpin + 'static,
+    Self: PartialOrd + Ord,
 {
     type Digest: Digest + FixedOutputReset + fmt::Debug;
     const ALGORITHM: u8;
@@ -93,7 +95,19 @@ impl fmt::Debug for Sha512HashValue {
     }
 }
 
-#[derive(Clone, Eq, FromBytes, Hash, Immutable, IntoBytes, KnownLayout, PartialEq, Unaligned)]
+#[derive(
+    Clone,
+    Eq,
+    FromBytes,
+    Hash,
+    Immutable,
+    IntoBytes,
+    KnownLayout,
+    PartialEq,
+    Unaligned,
+    PartialOrd,
+    Ord,
+)]
 #[repr(C)]
 pub struct Sha256HashValue([u8; 32]);
 
@@ -110,7 +124,19 @@ impl FsVerityHashValue for Sha256HashValue {
     const ID: &str = "sha256";
 }
 
-#[derive(Clone, Eq, FromBytes, Hash, Immutable, IntoBytes, KnownLayout, PartialEq, Unaligned)]
+#[derive(
+    Clone,
+    Eq,
+    FromBytes,
+    Hash,
+    Immutable,
+    IntoBytes,
+    KnownLayout,
+    PartialEq,
+    Unaligned,
+    PartialOrd,
+    Ord,
+)]
 #[repr(C)]
 pub struct Sha512HashValue([u8; 64]);
 

crates/composefs/src/repository.rs

@@ -24,7 +24,7 @@ use crate::{
         compute_verity, enable_verity, ensure_verity_equal, measure_verity, FsVerityHashValue,
     },
     mount::mount_composefs_at,
-    splitstream::{DigestMap, SplitStreamReader, SplitStreamWriter},
+    splitstream::{SplitStreamReader, SplitStreamWriter},
     util::{proc_self_fd, replace_symlinkat, ErrnoFilter, Sha256Digest},
 };
 
@@ -184,10 +184,10 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     /// store the result.
     pub fn create_stream(
         self: &Arc<Self>,
+        content_type: u64,
         sha256: Option<Sha256Digest>,
-        maps: Option<DigestMap<ObjectID>>,
     ) -> SplitStreamWriter<ObjectID> {
-        SplitStreamWriter::new(self, maps, sha256)
+        SplitStreamWriter::new(self, content_type, sha256)
     }
 
     fn format_object_path(id: &ObjectID) -> String {
@@ -224,11 +224,11 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
             Ok(stream) => {
                 let measured_verity: ObjectID = measure_verity(&stream)?;
                 let mut context = Sha256::new();
-                let mut split_stream = SplitStreamReader::new(File::from(stream))?;
+                let mut split_stream = SplitStreamReader::new(File::from(stream), None)?;
 
                 // check the verity of all linked streams
-                for entry in &split_stream.refs.map {
-                    if self.check_stream(&entry.body)?.as_ref() != Some(&entry.verity) {
+                for (body, verity) in split_stream.iter_mappings() {
+                    if self.check_stream(body)?.as_ref() != Some(verity) {
                         bail!("reference mismatch");
                     }
                 }
@@ -271,6 +271,12 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         Ok(object_id)
     }
 
+    pub fn has_named_stream(&self, name: &str) -> bool {
+        let stream_path = format!("streams/refs/{}", name);
+
+        readlinkat(&self.repository, &stream_path, []).is_ok()
+    }
+
     /// Assign the given name to a stream.  The stream must already exist.  After this operation it
     /// will be possible to refer to the stream by its new name 'refs/{name}'.
     pub fn name_stream(&self, sha256: Sha256Digest, name: &str) -> Result<()> {
@@ -297,6 +303,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     pub fn ensure_stream(
         self: &Arc<Self>,
         sha256: &Sha256Digest,
+        content_type: u64,
         callback: impl FnOnce(&mut SplitStreamWriter<ObjectID>) -> Result<()>,
         reference: Option<&str>,
     ) -> Result<ObjectID> {
@@ -305,7 +312,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         let object_id = match self.has_stream(sha256)? {
             Some(id) => id,
             None => {
-                let mut writer = self.create_stream(Some(*sha256), None);
+                let mut writer = self.create_stream(content_type, Some(*sha256));
                 callback(&mut writer)?;
                 let object_id = writer.done()?;
 
@@ -327,6 +334,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         &self,
         name: &str,
         verity: Option<&ObjectID>,
+        expected_content_type: Option<u64>,
     ) -> Result<SplitStreamReader<File, ObjectID>> {
         let filename = format!("streams/{name}");
 
@@ -338,7 +346,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
                 .with_context(|| format!("Opening ref 'streams/{name}'"))?
         });
 
-        SplitStreamReader::new(file)
+        SplitStreamReader::new(file, expected_content_type)
     }
 
     pub fn open_object(&self, id: &ObjectID) -> Result<OwnedFd> {
@@ -349,9 +357,10 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
         &self,
         name: &str,
         verity: Option<&ObjectID>,
+        expected_content_type: Option<u64>,
         stream: &mut impl Write,
     ) -> Result<()> {
-        let mut split_stream = self.open_stream(name, verity)?;
+        let mut split_stream = self.open_stream(name, verity, expected_content_type)?;
         split_stream.cat(stream, |id| -> Result<Vec<u8>> {
             let mut data = vec![];
             File::from(self.open_object(id)?).read_to_end(&mut data)?;
@@ -551,7 +560,7 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
             println!("{object:?} lives as a stream");
             objects.insert(object.clone());
 
-            let mut split_stream = self.open_stream(&object.to_hex(), None)?;
+            let mut split_stream = self.open_stream(&object.to_hex(), None, None)?;
             split_stream.get_object_refs(|id| {
                 println!("   with {id:?}");
                 objects.insert(id.clone());