znippy_common/

meta_sink.rs

//! `ArchiveMetaSink` — abstraction over the archive's metadata layer.
//!
//! After the (unchanged) compression pipeline writes all blob bytes to disk, the
//! metadata layer — one Arrow IPC sub-index per `(pkg_type, repo)` group, a
//! manifest, and the `MULTI_INDEX_MAGIC` footer — is written through this trait.
//!
//! [`ArrowIpcSink`] reproduces the v0.7 on-disk format byte-for-byte. Future
//! backends (e.g. Iceberg) implement the same trait without touching the blob
//! pipeline.

use std::fs::File;
use std::os::unix::fs::FileExt;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use arrow::array::{
    BooleanArray, BooleanBuilder, FixedSizeBinaryArray, FixedSizeBinaryBuilder, StringArray,
    StringBuilder, UInt32Array, UInt32Builder, UInt64Array, UInt64Builder,
};
use arrow::datatypes::Schema;
use arrow::ipc::writer::StreamWriter;
use arrow::record_batch::RecordBatch;

use crate::index::{
    ChunkLoc, LOOKUP_MODULE, META_MODULE, MULTI_INDEX_MAGIC, ManifestEntry, RESERVED_PKG_TYPE,
    TRIE_MODULE, is_reserved_module, lookup_schema, write_manifest_bytes,
};
use crate::meta_index::{MetaTable, build_meta_batch, meta_schema};
#[cfg(feature = "sign")]
use crate::index::{SIGN_ARCHIVE_MODULE, SIGN_ARTIFACTS_MODULE};

/// Identifies the logical sub-archive a sub-index belongs to.
#[derive(Debug, Clone)]
pub struct GroupKey {
    pub pkg_type: i8,
    pub repo: String,
    pub module_name: String,
}

/// The bytes of one extra reserved section a writer asks the sink to seal.
///
/// `Raw` is a byte blob framed by the manifest entry (what the fst trie and the
/// git oid index are); `Arrow` is a real Arrow IPC stream (what the commit graph
/// and the reachability bitmaps are), so DuckDB/Polars can read it straight out
/// of the archive by its manifest byte range.
///
/// `Clone` is cheap on both arms and it is what lets a writer *keep* the
/// sections it just sealed: [`ReservedSectionBuilder`] consumes what it returns,
/// so a caller that must both seal and hand the sections back (gunnar's
/// `GitOps::seal`) would otherwise have to build them twice — two derivations of
/// one fact, which is the drift LAW 5 forbids. `RecordBatch` clones are Arc
/// bumps; only a `Raw` payload copies bytes.
#[derive(Clone)]
pub enum ReservedPayload {
    Raw(Vec<u8>),
    Arrow { schema: Arc<Schema>, batches: Vec<RecordBatch> },
}

/// One extra reserved section. `module_name` **must** satisfy
/// [`is_reserved_module`] — the sink refuses anything else, because a section
/// the manifest readers do not classify as reserved is merged into the data
/// index and corrupts `list` / `decompress` / the iceberg sink.
#[derive(Clone)]
pub struct ReservedSection {
    pub module_name: String,
    pub payload: ReservedPayload,
}

/// Concise on purpose: a `Raw` payload can be megabytes, and the useful facts in
/// an error message are the module and the size, never the bytes.
impl std::fmt::Debug for ReservedSection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.payload {
            ReservedPayload::Raw(b) => f
                .debug_struct("ReservedSection")
                .field("module_name", &self.module_name)
                .field("raw_bytes", &b.len())
                .finish(),
            ReservedPayload::Arrow { batches, .. } => f
                .debug_struct("ReservedSection")
                .field("module_name", &self.module_name)
                .field("rows", &batches.iter().map(|b| b.num_rows()).sum::<usize>())
                .finish(),
        }
    }
}

impl ReservedSection {
    pub fn raw(module_name: impl Into<String>, bytes: Vec<u8>) -> Self {
        Self { module_name: module_name.into(), payload: ReservedPayload::Raw(bytes) }
    }

    pub fn arrow(
        module_name: impl Into<String>,
        schema: Arc<Schema>,
        batches: Vec<RecordBatch>,
    ) -> Self {
        Self {
            module_name: module_name.into(),
            payload: ReservedPayload::Arrow { schema, batches },
        }
    }
}

/// Read-only view of the **sorted lookup** exactly as the sink is about to seal
/// it: row `r` of the lookup sub-index is `(path(r), loc(r))`, ordered by
/// `(relative_path, chunk_seq)`.
///
/// Handed to a [`ReservedSectionBuilder`] so an extra index can point at real
/// lookup row numbers instead of re-deriving the sort and hoping the two agree
/// (LAW 5 — fix by construction, do not add a guard watching two copies).
pub struct LookupView<'a> {
    paths: &'a [String],
    locs: &'a [ChunkLoc],
    order: &'a [usize],
}

impl<'a> LookupView<'a> {
    /// Number of lookup rows.
    pub fn len(&self) -> usize {
        self.order.len()
    }

    pub fn is_empty(&self) -> bool {
        self.order.is_empty()
    }

    /// `relative_path` of lookup row `row`.
    pub fn path(&self, row: usize) -> &'a str {
        &self.paths[self.order[row]]
    }

    /// Chunk location of lookup row `row`.
    pub fn loc(&self, row: usize) -> &'a ChunkLoc {
        &self.locs[self.order[row]]
    }

    /// Every distinct `relative_path`, in sorted order, with the lookup row its
    /// contiguous chunk run starts at.
    pub fn first_rows(&self) -> Vec<(&'a str, u64)> {
        let mut out: Vec<(&'a str, u64)> = Vec::new();
        let mut prev: Option<&str> = None;
        for row in 0..self.order.len() {
            let p = self.path(row);
            if prev != Some(p) {
                out.push((p, row as u64));
                prev = Some(p);
            }
        }
        out
    }
}

/// Builds extra reserved sections once the sink knows the final sorted lookup.
///
/// Called from [`ArrowIpcSink::finish`], after the lookup + trie are laid out and
/// before the manifest is written, so returned sections are recorded as reserved
/// manifest entries like every other derived structure.
pub type ReservedSectionBuilder =
    Box<dyn FnOnce(&LookupView<'_>) -> Result<Vec<ReservedSection>> + Send>;

/// Writes the archive metadata layer (sub-indexes + manifest + footer).
///
/// The blob bytes have already been written to the output by the compression
/// pipeline; implementations only decide how the metadata is materialized.
pub trait ArchiveMetaSink {
    /// Serialize one sub-index — an Arrow IPC stream of `batches` (one or more)
    /// — place it after the previously written region, and record a manifest
    /// entry for it.
    fn push_subindex(
        &mut self,
        schema: &Schema,
        batches: &[RecordBatch],
        key: GroupKey,
    ) -> Result<()>;

    /// Write the manifest + footer, fsync, and return the total file length.
    fn finish(self: Box<Self>) -> Result<u64>;
}

/// Builds the metadata sink once the compression pipeline knows the output file
/// handle and the byte offset just past the last blob. The factory shape lets a
/// caller (e.g. the CLI) choose the backend — `ArrowIpcSink` (inline, default)
/// or a tokio-backed `IcebergSink` (off in a warehouse dir) — **without**
/// `znippy-compress` taking a dependency on the heavy/async backend: the
/// `IcebergSink` is constructed by the caller's closure, so its tokio/iceberg
/// deps stay in the binary that opted in.
///
/// `args`: `(output_file, blob_end_offset)`. An `ArrowIpcSink` uses both; an
/// `IcebergSink` ignores them (it writes its own warehouse, not the `.znippy`).
pub type MetaSinkFactory = Box<dyn FnOnce(Arc<File>, u64) -> Box<dyn ArchiveMetaSink> + Send>;

/// The default backend: inline Arrow IPC sub-indexes + manifest + 8-byte footer,
/// i.e. the v0.7 znippy container format. Behaviour is identical to the
/// previously-inlined writer tail in `slot_packer` / `stream_packer`.
pub struct ArrowIpcSink {
    file: Arc<File>,
    cursor: u64,
    entries: Vec<ManifestEntry>,
    /// Accumulated base columns of every data sub-index, used to build the sorted
    /// lookup sub-index + trie in [`finish`](ArrowIpcSink::finish).
    lookup_paths: Vec<String>,
    lookup_locs: Vec<ChunkLoc>,
    /// Optional searchable metadata (the `META_MODULE` sub-index). `None` — the
    /// default — emits no section at all, which is what makes the produced
    /// archive byte-identical to today's AND what a reader later reports as
    /// `ArchiveMeta::NoMetadata`. `Some(empty table)` is a different thing on
    /// purpose: a present, empty index.
    meta: Option<MetaTable>,
    /// Optional builder for extra *reserved* sections (the `git` package format's
    /// oid index / commit graph / reachability bitmaps). `None` — the default —
    /// writes nothing, so an archive sealed without one stays byte-identical.
    reserved_builder: Option<ReservedSectionBuilder>,
    /// Optional provenance signer (feature `sign`). When set, [`finish`] emits the
    /// per-artifact + per-archive detached CMS signatures as two additional
    /// *reserved* manifest sections — additive and backward-compatible. When
    /// `None` (the default), the produced archive is byte-identical to today's.
    #[cfg(feature = "sign")]
    signer: Option<Box<dyn crate::sign::ArchiveSigner + Send>>,
}

impl ArrowIpcSink {
    /// `blob_end_offset` is the byte offset just past the last blob — where the
    /// first sub-index is placed.
    pub fn new(file: Arc<File>, blob_end_offset: u64) -> Self {
        Self {
            file,
            cursor: blob_end_offset,
            entries: Vec::new(),
            lookup_paths: Vec::new(),
            lookup_locs: Vec::new(),
            meta: None,
            reserved_builder: None,
            #[cfg(feature = "sign")]
            signer: None,
        }
    }

    /// Seal extra **reserved** sections alongside the built-in derived ones.
    ///
    /// The builder is invoked in [`finish`](ArrowIpcSink::finish) with the final
    /// sorted [`LookupView`], so an index it emits can address real lookup rows.
    /// This is the injection point the `git` package format uses for
    /// `__gunnar_oid__` / `__gunnar_graph__` / `__gunnar_reach__` — the same
    /// shape as [`with_meta`](ArrowIpcSink::with_meta), so `compress_dir`'s
    /// `MetaSinkFactory` needs no new parameter and there is no second write path.
    pub fn with_reserved_builder(mut self, builder: ReservedSectionBuilder) -> Self {
        self.reserved_builder = Some(builder);
        self
    }

    /// Seal a searchable metadata sub-index alongside the index.
    ///
    /// This is the injection point for the compress path too: `compress_dir`
    /// already takes a `MetaSinkFactory`, so a caller adds metadata by handing it
    /// `Box::new(|f, b| Box::new(ArrowIpcSink::new(f, b).with_meta(table)))` —
    /// no change to the compress signature and no second write path.
    pub fn with_meta(mut self, meta: MetaTable) -> Self {
        self.meta = Some(meta);
        self
    }

    /// Attach a provenance signer (feature `sign`). On [`finish`], per-artifact and
    /// per-archive detached CMS signatures are written as reserved sections.
    #[cfg(feature = "sign")]
    pub fn with_signer(mut self, signer: Box<dyn crate::sign::ArchiveSigner + Send>) -> Self {
        self.signer = Some(signer);
        self
    }

    /// Pull the base index columns from one batch into the lookup accumulator.
    /// Every composed schema carries these columns; if any is absent we skip the
    /// batch (the lookup degrades gracefully — readers fall back to a scan).
    fn accumulate_lookup(&mut self, batch: &RecordBatch) {
        let cols = (|| {
            Some((
                batch.column_by_name("relative_path")?.as_any().downcast_ref::<StringArray>()?,
                batch.column_by_name("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()?,
                batch.column_by_name("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("compressed")?.as_any().downcast_ref::<BooleanArray>()?,
                batch.column_by_name("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()?,
            ))
        })();
        let Some((paths, chunk_seq, fdata, compressed, usz, blob_off, blob_sz, checksum)) = cols
        else { return; };
        for i in 0..batch.num_rows() {
            let mut ck = [0u8; 32];
            ck.copy_from_slice(checksum.value(i));
            self.lookup_paths.push(paths.value(i).to_string());
            self.lookup_locs.push(ChunkLoc {
                chunk_seq: chunk_seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: blob_off.value(i),
                blob_size: blob_sz.value(i),
                uncompressed_size: usz.value(i),
                compressed: compressed.value(i),
                checksum: ck,
            });
        }
    }

    /// The lookup row order: sorted by (path, chunk_seq) so each file's chunks
    /// are contiguous and paths are in byte-lexicographic order (fst requirement).
    ///
    /// Computed once in `finish` and shared by the lookup/trie writer and the
    /// extra reserved-section builder, so an index built over it addresses the
    /// rows the archive actually carries (LAW 5 — one writer, both paths).
    fn lookup_order(&self) -> Vec<usize> {
        let n = self.lookup_paths.len();
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            self.lookup_paths[a].cmp(&self.lookup_paths[b])
                .then(self.lookup_locs[a].chunk_seq.cmp(&self.lookup_locs[b].chunk_seq))
        });
        order
    }

    /// Invoke the caller's reserved-section builder (if any) and seal what it
    /// returns. A non-reserved `module_name` is a hard error: such a section
    /// would be merged into the data index by `read_multi_index` and would
    /// corrupt `list`, `decompress` and the iceberg sink.
    fn write_reserved_sections(&mut self, order: &[usize]) -> Result<()> {
        let Some(builder) = self.reserved_builder.take() else {
            return Ok(());
        };
        let sections = {
            let view = LookupView {
                paths: &self.lookup_paths,
                locs: &self.lookup_locs,
                order,
            };
            builder(&view)?
        };
        for section in sections {
            anyhow::ensure!(
                is_reserved_module(&section.module_name),
                "module '{}' is not a reserved module name; a non-reserved extra \
                 section would be merged into the data index and corrupt list/decompress",
                section.module_name,
            );
            let key = GroupKey {
                pkg_type: RESERVED_PKG_TYPE,
                repo: String::new(),
                module_name: section.module_name,
            };
            match section.payload {
                ReservedPayload::Raw(bytes) => self.write_raw_section(&bytes, key)?,
                ReservedPayload::Arrow { schema, batches } => {
                    self.push_subindex(schema.as_ref(), &batches, key)?
                }
            }
        }
        Ok(())
    }

    /// Write the sorted lookup sub-index + fst trie as two reserved manifest
    /// entries. Called from `finish` before the manifest is emitted.
    fn write_lookup_and_trie(&mut self, order: &[usize]) -> Result<()> {
        let n = self.lookup_paths.len();

        // Build the lookup sub-index batch (base schema, sorted).
        let mut path_b = StringBuilder::with_capacity(n, n * 16);
        let mut seq_b = UInt32Builder::with_capacity(n);
        let mut fdata_b = UInt64Builder::with_capacity(n);
        let mut comp_b = BooleanBuilder::with_capacity(n);
        let mut usz_b = UInt64Builder::with_capacity(n);
        let mut boff_b = UInt64Builder::with_capacity(n);
        let mut bsz_b = UInt64Builder::with_capacity(n);
        let mut ck_b = FixedSizeBinaryBuilder::with_capacity(n, 32);
        for &i in order {
            let loc = &self.lookup_locs[i];
            path_b.append_value(&self.lookup_paths[i]);
            seq_b.append_value(loc.chunk_seq);
            fdata_b.append_value(loc.fdata_offset);
            comp_b.append_value(loc.compressed);
            usz_b.append_value(loc.uncompressed_size);
            boff_b.append_value(loc.blob_offset);
            bsz_b.append_value(loc.blob_size);
            ck_b.append_value(loc.checksum).expect("checksum is 32 bytes");
        }
        let schema = lookup_schema();
        let batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(path_b.finish()),
                Arc::new(seq_b.finish()),
                Arc::new(fdata_b.finish()),
                Arc::new(comp_b.finish()),
                Arc::new(usz_b.finish()),
                Arc::new(boff_b.finish()),
                Arc::new(bsz_b.finish()),
                Arc::new(ck_b.finish()),
            ],
        )?;
        self.push_subindex(&schema, &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: LOOKUP_MODULE.to_string(),
        })?;

        // Build the fst trie: distinct relative_path → first row index in the
        // (sorted) lookup. Keys must be inserted in lexicographic order — `order`
        // already gives that.
        let mut builder = fst::MapBuilder::memory();
        let mut prev: Option<&str> = None;
        for (sorted_idx, &orig) in order.iter().enumerate() {
            let p = self.lookup_paths[orig].as_str();
            if prev != Some(p) {
                builder.insert(p.as_bytes(), sorted_idx as u64)
                    .map_err(|e| anyhow!("trie insert: {e}"))?;
                prev = Some(p);
            }
        }
        let trie_bytes = builder.into_inner().map_err(|e| anyhow!("trie finish: {e}"))?;
        self.write_raw_section(&trie_bytes, GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: TRIE_MODULE.to_string(),
        })
    }

    /// Emit the per-artifact + per-archive detached CMS signatures as two reserved
    /// sections, computed from the already-accumulated chunk hashes (Law 3 order).
    /// Never re-hashes content. Called from `finish` (before the manifest) only
    /// when a signer is attached.
    #[cfg(feature = "sign")]
    fn write_signatures(&mut self) -> Result<()> {
        use std::collections::BTreeMap;
        let Some(signer) = self.signer.take() else {
            return Ok(());
        };

        // Group chunk hashes by path (borrows self immutably). Produce owned
        // outputs so the borrow is released before we write to the file.
        let (file_digests, artifact_paths, artifact_cms): (
            Vec<(String, [u8; 32])>,
            Vec<String>,
            Vec<Vec<u8>>,
        ) = {
            let mut by_path: BTreeMap<&str, Vec<(u32, &[u8; 32])>> = BTreeMap::new();
            for (p, loc) in self.lookup_paths.iter().zip(self.lookup_locs.iter()) {
                by_path.entry(p.as_str()).or_default().push((loc.chunk_seq, &loc.checksum));
            }
            let mut digs = Vec::with_capacity(by_path.len());
            let mut paths = Vec::with_capacity(by_path.len());
            let mut cmss = Vec::with_capacity(by_path.len());
            for (path, mut chunks) in by_path {
                chunks.sort_by_key(|(seq, _)| *seq);
                let n = chunks.len();
                let digest = crate::sign::file_digest_from_parts(
                    path,
                    chunks.iter().map(|(s, c)| (*s, *c)),
                    n,
                );
                let cms = signer.sign_digest(&digest)?;
                digs.push((path.to_string(), digest));
                paths.push(path.to_string());
                cmss.push(cms);
            }
            (digs, paths, cmss)
        };

        // Per-archive root signature. The footer is always Multi (v0.7).
        let footer = crate::index::IndexFooter::Multi { manifest_offset: 0 };
        let root = crate::sign::archive_root(&file_digests, &footer);
        let archive_cms = signer.sign_digest(&root)?;

        let artifacts_bytes = serialize_artifact_signatures(&artifact_paths, &artifact_cms)?;
        self.write_raw_section(
            &artifacts_bytes,
            GroupKey {
                pkg_type: RESERVED_PKG_TYPE,
                repo: String::new(),
                module_name: SIGN_ARTIFACTS_MODULE.to_string(),
            },
        )?;
        self.write_raw_section(
            &archive_cms,
            GroupKey {
                pkg_type: RESERVED_PKG_TYPE,
                repo: String::new(),
                module_name: SIGN_ARCHIVE_MODULE.to_string(),
            },
        )?;
        Ok(())
    }

    /// Emit the searchable metadata sub-index, when there is one to emit.
    ///
    /// Reserved module, so the data readers skip it and an older znippy simply
    /// ignores the entry. `None` writes NOTHING — that absence is exactly what
    /// `ArchiveMeta::NoMetadata` reports, and it is why an archive sealed without
    /// metadata stays byte-identical to one sealed before this module existed.
    fn write_meta_subindex(&mut self) -> Result<()> {
        let Some(table) = self.meta.take() else {
            return Ok(());
        };
        let batch = build_meta_batch(&table)?;
        let schema = meta_schema();
        self.push_subindex(schema.as_ref(), &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: META_MODULE.to_string(),
        })
    }

    /// Write a raw (non-Arrow) byte section at the cursor and record a manifest
    /// entry whose `index_offset`/`index_len` frame it.
    fn write_raw_section(&mut self, bytes: &[u8], key: GroupKey) -> Result<()> {
        let start = self.cursor;
        self.file.write_all_at(bytes, start)?;
        self.cursor += bytes.len() as u64;
        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: start,
            index_len: bytes.len() as u64,
            row_count: 0,
        });
        Ok(())
    }
}

impl ArchiveMetaSink for ArrowIpcSink {
    fn push_subindex(
        &mut self,
        schema: &Schema,
        batches: &[RecordBatch],
        key: GroupKey,
    ) -> Result<()> {
        let sub_start = self.cursor;
        let mut sub_bytes: Vec<u8> = Vec::new();
        let mut sw = StreamWriter::try_new(&mut sub_bytes, schema)
            .map_err(|e| anyhow!("sub-index writer: {e}"))?;
        let mut row_count = 0u64;
        for batch in batches {
            row_count += batch.num_rows() as u64;
            sw.write(batch).map_err(|e| anyhow!("sub-index write: {e}"))?;
        }
        sw.finish().map_err(|e| anyhow!("sub-index finish: {e}"))?;

        // Accumulate base columns for the lookup layer — but not from the reserved
        // lookup sub-index itself (that would recurse / double-count).
        // Accumulate base columns for the lookup layer from DATA sub-indexes only.
        // Widened from "not lookup, not trie" to "not reserved" when META_MODULE
        // arrived: the metadata sub-index is Arrow IPC and does come through here,
        // and its rows are key/value facts, not chunk locations — folding them
        // into the lookup would corrupt random access. Behaviour for every
        // pre-existing module is unchanged (sign sections are raw, never pushed).
        if !is_reserved_module(&key.module_name) {
            for batch in batches {
                self.accumulate_lookup(batch);
            }
        }

        let sub_len = sub_bytes.len() as u64;
        self.file.write_all_at(&sub_bytes, sub_start)?;
        self.cursor += sub_len;

        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: sub_start,
            index_len: sub_len,
            row_count,
        });
        Ok(())
    }

    fn finish(mut self: Box<Self>) -> Result<u64> {
        // Emit the sorted lookup sub-index + trie before the manifest so their
        // byte ranges are recorded as (reserved) manifest entries.
        let order = self.lookup_order();
        self.write_lookup_and_trie(&order)?;

        // Emit the caller's extra reserved sections (the `git` package format's
        // oid index / commit graph / reachability bitmaps). A no-op unless a
        // caller attached a builder with `with_reserved_builder`.
        self.write_reserved_sections(&order)?;

        // Emit the searchable metadata sub-index (reserved). A no-op unless a
        // caller attached one with `with_meta`, so the default archive is
        // byte-identical to today's.
        self.write_meta_subindex()?;

        // Emit the detached provenance signatures (feature `sign`) — also reserved
        // sections, also before the manifest. A no-op when no signer is attached,
        // so unsigned archives stay byte-identical to today's format.
        #[cfg(feature = "sign")]
        self.write_signatures()?;

        let manifest_offset = self.cursor;
        let manifest_bytes =
            write_manifest_bytes(&self.entries).map_err(|e| anyhow!("manifest: {e}"))?;
        self.file.write_all_at(&manifest_bytes, manifest_offset)?;

        let after = manifest_offset + manifest_bytes.len() as u64;
        self.file.write_all_at(&MULTI_INDEX_MAGIC, after)?;
        self.file.write_all_at(
            &manifest_offset.to_le_bytes(),
            after + MULTI_INDEX_MAGIC.len() as u64,
        )?;
        self.file.sync_all()?;

        Ok(after + MULTI_INDEX_MAGIC.len() as u64 + 8)
    }
}

/// Serialize the per-artifact detached CMS signatures as an Arrow IPC stream with
/// columns `(relative_path: Utf8, cms: Binary)` — one row per file. Itself a
/// valid Arrow IPC file (DuckDB/Polars-queryable), stored as a reserved section.
#[cfg(feature = "sign")]
fn serialize_artifact_signatures(paths: &[String], cms: &[Vec<u8>]) -> Result<Vec<u8>> {
    use arrow::array::BinaryBuilder;
    use arrow::datatypes::{DataType, Field, Schema};

    let n = paths.len();
    let schema = Arc::new(Schema::new(vec![
        Field::new("relative_path", DataType::Utf8, false),
        Field::new("cms", DataType::Binary, false),
    ]));
    let mut path_b = StringBuilder::with_capacity(n, n * 32);
    let mut cms_b = BinaryBuilder::with_capacity(n, n * 512);
    for (p, c) in paths.iter().zip(cms.iter()) {
        path_b.append_value(p);
        cms_b.append_value(c);
    }
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![Arc::new(path_b.finish()), Arc::new(cms_b.finish())],
    )?;
    let mut buf = Vec::new();
    {
        let mut w = StreamWriter::try_new(&mut buf, &schema)
            .map_err(|e| anyhow!("artifact-sig writer: {e}"))?;
        w.write(&batch).map_err(|e| anyhow!("artifact-sig write: {e}"))?;
        w.finish().map_err(|e| anyhow!("artifact-sig finish: {e}"))?;
    }
    Ok(buf)
}