mkit_core/store/

source.rs

//! Read-side object sources and adapters over the durable [`ObjectStore`].
//!
//! [`ObjectSource`] is the read trait shared by the store and its overlays;
//! [`EphemeralSink`] is the in-memory snapshot overlay for query commands;
//! [`DisplaySource`] is the display-only adapter that opts a render path out of
//! per-read hash verification (#625). Extracted from `store.rs` (#633): these are
//! consumers of the store's read API, distinct from its on-disk write/fsync
//! machinery.

use crate::hash::Hash;
use crate::object::{Object, object_id_from_parts};
use crate::serialize;

use super::{MAX_RAW_OBJECT_SIZE, ObjectSink, ObjectStore, StoreError, StoreResult};

/// Read source shared by [`ObjectStore`] and snapshot overlays, so
/// tree-diff code can resolve objects from either the durable store or
/// an in-memory [`EphemeralSink`].
pub trait ObjectSource {
    /// Read and integrity-verify the raw bytes of `h`.
    ///
    /// # Errors
    /// [`StoreError::ObjectNotFound`] / [`StoreError::HashMismatch`] /
    /// I/O errors, as for [`ObjectStore::read`].
    fn read(&self, h: &Hash) -> StoreResult<Vec<u8>>;

    /// Read and decode `h` into a typed [`Object`].
    ///
    /// # Errors
    /// As [`Self::read`], plus decode errors.
    fn read_object(&self, h: &Hash) -> StoreResult<Object> {
        Ok(serialize::deserialize(&self.read(h)?)?)
    }

    /// Read `h` without integrity verification, for display-only
    /// rendering — see [`ObjectStore::read_unverified`] for the full
    /// policy (#625). Defaults to the fully-verifying [`Self::read`], so
    /// every existing and third-party [`ObjectSource`] stays verifying
    /// unless it explicitly opts in by overriding this method.
    ///
    /// # Errors
    /// As [`Self::read`] (an opting-in override may narrow this to never
    /// return [`StoreError::HashMismatch`]).
    fn read_unverified(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        self.read(h)
    }
}

impl ObjectSource for ObjectStore {
    fn read(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        ObjectStore::read(self, h)
    }

    fn read_unverified(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        ObjectStore::read_unverified(self, h)
    }

    /// Overrides the trait default (`deserialize(self.read(h)?)`, which
    /// would decode a `Tree`/`ChunkedBlob` twice — once inside `read`'s
    /// id verification, once again here) to delegate to the inherent
    /// [`ObjectStore::read_object`], which decodes once and reuses the
    /// result. Every caller generic over `S: ObjectSource` (`diff::load_tree`,
    /// `LoadedBlob::load`, …) resolves `read_object` through this trait
    /// impl rather than the inherent method even when `S = ObjectStore`,
    /// so without this override those call sites — the primary
    /// Tree/ChunkedBlob decoding paths — would keep paying for the
    /// double decode the inherent method was written to avoid.
    fn read_object(&self, h: &Hash) -> StoreResult<Object> {
        ObjectStore::read_object(self, h)
    }
}

/// In-memory object overlay for **ephemeral worktree snapshots**
/// (`status`, `diff`, conflict/restore safety checks).
///
/// Writes never touch the durable store: objects whose hash already
/// exists on disk are deduplicated against it (the store's
/// visible-implies-durable invariant makes that safe), everything else
/// lives in a private map that vanishes with the sink. This is what
/// keeps query commands from (a) paying any durability cost, (b)
/// growing `objects/` with throwaway snapshot trees, and (c) ever
/// making a non-durable object *visible* where another writer's dedup
/// could durably reference it.
///
/// Reads fall through to the underlying store, so diff walkers can
/// resolve a snapshot tree that references committed objects.
#[derive(Debug)]
pub struct EphemeralSink<'s> {
    store: &'s ObjectStore,
    objects: std::sync::Mutex<std::collections::HashMap<Hash, Vec<u8>>>,
}

impl<'s> EphemeralSink<'s> {
    /// Create an empty overlay over `store`.
    #[must_use]
    pub fn new(store: &'s ObjectStore) -> Self {
        Self {
            store,
            objects: std::sync::Mutex::new(std::collections::HashMap::new()),
        }
    }
}

impl ObjectSink for EphemeralSink<'_> {
    fn put(&self, bytes: &[u8]) -> StoreResult<Hash> {
        self.put_parts(&[bytes])
    }

    fn put_parts(&self, parts: &[&[u8]]) -> StoreResult<Hash> {
        let mut total: usize = 0;
        for p in parts {
            total = total
                .checked_add(p.len())
                .ok_or(StoreError::ObjectTooLarge)?;
        }
        if total > MAX_RAW_OBJECT_SIZE {
            return Err(StoreError::ObjectTooLarge);
        }
        let h = object_id_from_parts(parts);
        // Dedup against the durable store: visible store objects are durable
        // by invariant, and skipping them keeps the overlay's memory bounded
        // by the *changed* content, not the worktree. The overlay only ever
        // materialises the bytes on a dedup miss.
        if self.store.contains(&h) {
            return Ok(h);
        }
        self.objects
            .lock()
            .expect("ephemeral sink mutex")
            .entry(h)
            .or_insert_with(|| {
                let mut buf = Vec::with_capacity(total);
                for p in parts {
                    buf.extend_from_slice(p);
                }
                buf
            });
        Ok(h)
    }

    fn has(&self, h: &Hash) -> bool {
        self.objects
            .lock()
            .expect("ephemeral sink mutex")
            .contains_key(h)
            || self.store.contains(h)
    }
}

impl EphemeralSink<'_> {
    /// The private-map half of a read: bytes this process already built
    /// (via `put`/`put_parts`) and never persisted anywhere else to be
    /// corrupted, shared by both [`ObjectSource::read`] and
    /// [`ObjectSource::read_unverified`] below — the two differ only in
    /// what they do on a miss.
    fn overlay_get(&self, h: &Hash) -> Option<Vec<u8>> {
        self.objects
            .lock()
            .expect("ephemeral sink mutex")
            .get(h)
            .cloned()
    }
}

impl ObjectSource for EphemeralSink<'_> {
    fn read(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        match self.overlay_get(h) {
            Some(bytes) => Ok(bytes),
            None => self.store.read(h),
        }
    }

    /// Overrides the trait default for the same reason `impl ObjectSource
    /// for ObjectStore` does: a private-map hit only ever needs one
    /// decode (the default's `deserialize(self.read(h)?)` already gives
    /// that), but a store fall-through would otherwise re-decode a
    /// `Tree`/`ChunkedBlob` a second time on top of the decode
    /// `ObjectStore::read`'s id verification already does internally.
    /// Delegating to `self.store.read_object` on a miss reuses that
    /// store-side decode instead — `diff`/`status`, which resolve staged
    /// trees through this overlay, are the paths this matters for.
    fn read_object(&self, h: &Hash) -> StoreResult<Object> {
        match self.overlay_get(h) {
            Some(bytes) => Ok(serialize::deserialize(&bytes)?),
            None => self.store.read_object(h),
        }
    }

    fn read_unverified(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        // Private-map hits are already trusted, same as the verifying
        // `read` path above. Only the store fall-through actually skips a
        // hash check.
        match self.overlay_get(h) {
            Some(bytes) => Ok(bytes),
            None => self.store.read_unverified(h),
        }
    }
}

/// Adapter that makes any [`ObjectSource`] read without BLAKE3
/// verification, for display-only rendering (`diff`, `show`, and the
/// commit/merge/pull post-op summaries). Wrap the source once at the
/// render call site — `DisplaySource::new(&store)` — and pass the
/// wrapper wherever a generic `S: ObjectSource` render function expects
/// its source. Every existing render function (`render_stat`,
/// `emit_entry_patch`, `LoadedBlob::load`/`prefix`/`into_content`) works
/// unchanged: the verify/no-verify policy lives entirely in which source
/// gets passed in, never in a flag threaded through render code.
///
/// # When to use
/// mkit never feeds unverified bytes into state that mkit itself writes,
/// or into output that mkit's own tooling round-trips (format-patch →
/// `git am`) — a diffstat or patch preview that only a human reads and
/// discards is neither, so it's fair game. NEVER wrap a source feeding
/// publication (commit/merge/rebase tree writes), dedup, fetch/apply, or
/// any path whose output becomes durable state or gets applied elsewhere
/// (e.g. the format-patch body in `git_tools.rs`, which is deliberately
/// NOT wrapped because `git am` applies it into new commits), including
/// [`crate::verify::build_disclosure_from`], whose bundles are published
/// proofs. See
/// [`ObjectStore::read_unverified`] for the full policy this wrapper
/// exists to apply consistently (#625).
pub struct DisplaySource<'a, S: ObjectSource + ?Sized> {
    inner: &'a S,
}

impl<'a, S: ObjectSource + ?Sized> DisplaySource<'a, S> {
    /// Wrap `inner` so reads through this adapter skip verification.
    #[must_use]
    pub fn new(inner: &'a S) -> Self {
        Self { inner }
    }
}

// Manual `Debug` (rather than `derive`) so this doesn't force an
// unnecessary `S: Debug` bound on every generic render fn that only
// needs `S: ObjectSource`.
impl<S: ObjectSource + ?Sized> std::fmt::Debug for DisplaySource<'_, S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DisplaySource").finish_non_exhaustive()
    }
}

impl<S: ObjectSource + ?Sized> ObjectSource for DisplaySource<'_, S> {
    fn read(&self, h: &Hash) -> StoreResult<Vec<u8>> {
        self.inner.read_unverified(h)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::layout::RepoLayout;
    use std::fs::OpenOptions;
    use std::io::{self, Seek, Write};
    use tempfile::TempDir;

    fn fresh_store() -> (TempDir, ObjectStore) {
        let dir = TempDir::new().expect("tempdir");
        let store = ObjectStore::init(&RepoLayout::single(dir.path())).expect("init");
        (dir, store)
    }

    /// Flip the first byte of the on-disk object file for `h`, in place.
    /// Shared by the `DisplaySource` corruption tests below — the same
    /// "the bytes on disk no longer match `h`" setup the corruption tests
    /// in `store.rs` use (deliberately duplicated there rather than
    /// exposing a test-support seam).
    fn corrupt_first_byte(store: &ObjectStore, h: &Hash, first_byte: u8) {
        let path = store.path_for(h);
        let mut f = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        f.seek(io::SeekFrom::Start(0)).unwrap();
        f.write_all(&[first_byte ^ 0xFF]).unwrap();
        f.sync_all().unwrap();
    }

    #[test]
    fn display_source_delegates_to_unverified() {
        // `DisplaySource` is the intended call-site adapter: wrapping the
        // store must make `ObjectSource::read` succeed on an object that
        // `ObjectStore::read` itself would reject.
        let (_dir, store) = fresh_store();
        let bytes = b"trustworthy".to_vec();
        let h = store.write(&bytes).unwrap();
        corrupt_first_byte(&store, &h, bytes[0]);
        assert!(matches!(
            store.read(&h).unwrap_err(),
            StoreError::HashMismatch { .. }
        ));

        let display = DisplaySource::new(&store);
        let mut corrupted = bytes.clone();
        corrupted[0] ^= 0xFF;
        assert_eq!(
            display.read(&h).unwrap(),
            corrupted,
            "DisplaySource::read must delegate to the store's unverified read"
        );
    }

    /// Regression test for the `EphemeralSink::read_object` override:
    /// both a private-map (overlay) hit and a store fall-through must
    /// decode a `Tree` correctly, and the store fall-through must still
    /// surface `HashMismatch` on corruption — through `&dyn ObjectSource`,
    /// the shape every `diff`/`status` caller actually uses.
    #[test]
    fn ephemeral_sink_read_object_tree_overlay_and_fallthrough() {
        use crate::object::{EntryMode, Tree, TreeEntry};
        let tree = |name: &[u8]| {
            Object::Tree(Tree {
                entries: vec![TreeEntry {
                    name: name.to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: crate::hash::hash(b"x"),
                }],
            })
        };

        let (_dir, store) = fresh_store();

        // Store fall-through: object lives only in the durable store.
        let durable_obj = tree(b"durable.txt");
        let durable_bytes = serialize::serialize(&durable_obj).unwrap();
        let durable_h = store.write(&durable_bytes).unwrap();

        // Overlay hit: object lives only in the sink's private map.
        let overlay_obj = tree(b"overlay.txt");
        let overlay_bytes = serialize::serialize(&overlay_obj).unwrap();

        let sink = EphemeralSink::new(&store);
        let overlay_h = sink.put(&overlay_bytes).unwrap();

        let via_trait: &dyn ObjectSource = &sink;
        assert_eq!(via_trait.read_object(&durable_h).unwrap(), durable_obj);
        assert_eq!(via_trait.read_object(&overlay_h).unwrap(), overlay_obj);

        corrupt_first_byte(&store, &durable_h, durable_bytes[0]);
        assert!(matches!(
            via_trait.read_object(&durable_h).unwrap_err(),
            StoreError::HashMismatch { .. }
        ));
    }

    #[test]
    fn ephemeral_sink_unverified_falls_through() {
        // Two shapes an `EphemeralSink` reader can hit: a private-map
        // object built by this snapshot (never touches disk, so nothing to
        // corrupt) and a store-backed object corrupted on disk. Both must
        // read fine through `DisplaySource<EphemeralSink>`.
        let (_dir, store) = fresh_store();
        let durable_bytes = b"trustworthy".to_vec();
        let durable_h = store.write(&durable_bytes).unwrap();
        corrupt_first_byte(&store, &durable_h, durable_bytes[0]);

        let sink = EphemeralSink::new(&store);
        let private_h = sink.put(b"snapshot-only").unwrap();

        let display = DisplaySource::new(&sink);
        assert_eq!(display.read(&private_h).unwrap(), b"snapshot-only");

        let mut corrupted = durable_bytes.clone();
        corrupted[0] ^= 0xFF;
        assert_eq!(display.read(&durable_h).unwrap(), corrupted);
    }

    #[test]
    fn read_unverified_default_is_verifying_read() {
        // A minimal `ObjectSource` impl that only defines `read` must get
        // fully-verifying behaviour from `read_unverified`'s default body
        // — the whole safety net for third-party/future implementors that
        // haven't opted in to the unverified path.
        struct OnlyReadImpl<'s>(&'s ObjectStore);
        impl ObjectSource for OnlyReadImpl<'_> {
            fn read(&self, h: &Hash) -> StoreResult<Vec<u8>> {
                self.0.read(h)
            }
        }

        let (_dir, store) = fresh_store();
        let bytes = b"trustworthy".to_vec();
        let h = store.write(&bytes).unwrap();
        corrupt_first_byte(&store, &h, bytes[0]);

        let only_read = OnlyReadImpl(&store);
        assert!(matches!(
            only_read.read_unverified(&h).unwrap_err(),
            StoreError::HashMismatch { .. }
        ));
    }
}