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.
 * 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/composefs-http/src/lib.rs

@@ -15,7 +15,7 @@ use tokio::task::JoinSet;
 use composefs::{
     fsverity::FsVerityHashValue,
     repository::Repository,
-    splitstream::{DigestMapEntry, SplitStreamReader},
+    splitstream::{SplitStreamReader},
     util::Sha256Digest,
 };
 
@@ -107,7 +107,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/lib.rs

@@ -83,7 +83,7 @@ pub fn open_config<ObjectID: FsVerityHashValue>(
     };
     let mut stream = repo.open_stream(name, Some(id))?;
     let config = ImageConfiguration::from_reader(&mut stream)?;
-    Ok((config, stream.refs))
+    Ok((config, stream.get_mappings()))
 }
 
 fn hash(bytes: &[u8]) -> Sha256Digest {
@@ -121,7 +121,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(Some(sha256));
+    stream.add_sha256_mappings(refs);
     stream.write_inline(json_bytes);
     let id = repo.write_stream(stream, None)?;
     Ok((sha256, id))

crates/composefs-oci/src/skopeo.rs

@@ -10,7 +10,7 @@ use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest, MediaType};
 use tokio::{io::AsyncReadExt, sync::Semaphore};
 
 use composefs::{
-    fsverity::FsVerityHashValue, repository::Repository, splitstream::DigestMap, util::Sha256Digest,
+    fsverity::FsVerityHashValue, repository::Repository, util::Sha256Digest,
 };
 
 use crate::{sha256_from_descriptor, sha256_from_digest, tar::split_async, ContentAndVerity};
@@ -77,7 +77,7 @@ 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(Some(layer_sha256));
             match descriptor.media_type() {
                 MediaType::ImageLayer => {
                     split_async(progress, &mut splitstream).await?;
@@ -154,15 +154,15 @@ impl<ObjectID: FsVerityHashValue> ImageOp<ObjectID> {
                 entries.push((layer_sha256, future));
             }
 
+            let mut splitstream = self
+                .repo
+                .create_stream(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,7 @@ 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 +112,7 @@ 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::{filter_errno, proc_self_fd, replace_symlinkat, Sha256Digest},
 };
 
@@ -185,9 +185,8 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
     pub fn create_stream(
         self: &Arc<Self>,
         sha256: Option<Sha256Digest>,
-        maps: Option<DigestMap<ObjectID>>,
     ) -> SplitStreamWriter<ObjectID> {
-        SplitStreamWriter::new(self, maps, sha256)
+        SplitStreamWriter::new(self, sha256)
     }
 
     fn format_object_path(id: &ObjectID) -> String {
@@ -227,8 +226,8 @@ impl<ObjectID: FsVerityHashValue> Repository<ObjectID> {
                 let mut split_stream = SplitStreamReader::new(File::from(stream))?;
 
                 // 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");
                     }
                 }
@@ -305,7 +304,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(Some(*sha256));
                 callback(&mut writer)?;
                 let object_id = writer.done()?;