Skip to main content

triblespace_core/
repo.rs

1#![allow(clippy::type_complexity)]
2//! This module provides a high-level API for storing and retrieving data from repositories.
3//! The design is inspired by Git, but with a focus on object/content-addressed storage.
4//! It separates storage concerns from the data model, and reduces the mutable state of the repository,
5//! to an absolute minimum, making it easier to reason about and allowing for different storage backends.
6//!
7//! Blob repositories are collections of blobs that can be content-addressed by their hash.
8//! This is typically local `.pile` file or a S3 bucket or a similar service.
9//! On their own they have no notion of branches or commits, or other stateful constructs.
10//! As such they also don't have a notion of time, order or history,
11//! massively relaxing the constraints on storage.
12//! This makes it possible to use a wide range of storage services, including those that don't support
13//! atomic transactions or have other limitations.
14//!
15//! Branch repositories on the other hand are a stateful construct that can be used to represent a branch pointing to a specific commit.
16//! They are stored in a separate repository, typically a  local `.pile` file, a database or an S3 compatible service with a compare-and-swap operation,
17//! and can be used to represent the state of a repository at a specific point in time.
18//!
19//! Technically, branches are just a mapping from a branch id to a blob hash,
20//! But because TribleSets are themselves easily stored in a blob, and because
21//! trible commit histories are an append-only chain of TribleSet metadata,
22//! the hash of the head is sufficient to represent the entire history of a branch.
23//!
24//! ## Basic usage
25//!
26//! ```rust,ignore
27//! use ed25519_dalek::SigningKey;
28//! use rand::rngs::OsRng;
29//! use triblespace::prelude::*;
30//! use triblespace::prelude::inlineencodings::{GenId, ShortString};
31//! use triblespace::repo::{memoryrepo::MemoryRepo, Repository};
32//!
33//! let storage = MemoryRepo::default();
34//! let mut repo = Repository::new(storage, SigningKey::generate(&mut OsRng), TribleSet::new()).unwrap();
35//! let branch_id = repo.create_branch("main", None).expect("create branch");
36//! let mut ws = repo.pull(*branch_id).expect("pull branch");
37//!
38//! attributes! {
39//!     "8F180883F9FD5F787E9E0AF0DF5866B9" as pub author: GenId;
40//!     "0DBB530B37B966D137C50B943700EDB2" as pub firstname: ShortString;
41//!     "6BAA463FD4EAF45F6A103DB9433E4545" as pub lastname: ShortString;
42//! }
43//! let author = fucid();
44//! ws.commit(
45//!     entity!{ &author @
46//!         literature::firstname: "Frank",
47//!         literature::lastname: "Herbert",
48//!      },
49//!     "initial commit",
50//! );
51//!
52//! // Single-attempt push: `try_push` uploads local blobs and attempts a
53//! // single CAS update. On conflict it returns a workspace containing the
54//! // new branch state which you should merge into before retrying.
55//! match repo.try_push(&mut ws).expect("try_push") {
56//!     None => {}
57//!     Some(_) => panic!("unexpected conflict"),
58//! }
59//! ```
60//!
61//! `create_branch` registers a new branch and returns an [`ExclusiveId`](crate::id::ExclusiveId) guard.
62//! `pull` creates a new workspace from an existing branch while
63//! `branch_from` can be used to start a new branch from a specific commit
64//! handle. See `examples/workspace.rs` for a more complete example.
65//!
66//! ## Handling conflicts
67//!
68//! The single-attempt primitive is [`Repository::try_push`](crate::repo::Repository::try_push). It returns
69//! `Ok(None)` on success or `Ok(Some(conflict_ws))` when the branch advanced
70//! concurrently. Callers that want explicit conflict handling may use this
71//! form:
72//!
73//! ```rust,ignore
74//! while let Some(mut other) = repo.try_push(&mut ws)? {
75//!     // Merge our staged changes into the incoming workspace and retry.
76//!     other.merge(&mut ws)?;
77//!     ws = other;
78//! }
79//! ```
80//!
81//! For convenience `Repository::push` is provided as a retrying wrapper that
82//! performs the merge-and-retry loop for you. Call `push` when you prefer the
83//! repository to handle conflicts automatically; call `try_push` when you need
84//! to inspect or control the intermediate conflict workspace yourself.
85//!
86//! `push` performs a compare‐and‐swap (CAS) update on the branch metadata.
87//! This optimistic concurrency control keeps branches consistent without
88//! locking and can be emulated by many storage systems (for example by
89//! using conditional writes on S3).
90//!
91//! ## Git parallels
92//!
93//! The API deliberately mirrors concepts from Git to make its usage familiar:
94//!
95//! - A [`Repository`](crate::repo::Repository) stores commits and branch metadata similar to a remote.
96//! - [`Workspace`](crate::repo::Workspace) is akin to a working directory combined with an index. It
97//!   tracks changes against a branch head until you `push` them.
98//! - `create_branch` and `branch_from` correspond to creating new branches from
99//!   scratch or from a specific commit, respectively.
100//! - `push` updates the repository atomically. If the branch advanced in the
101//!   meantime, you receive a conflict workspace which can be merged before
102//!   retrying the push.
103//! - `pull` is similar to cloning a branch into a new workspace.
104//!
105//! `pull` uses the repository's default signing key for new commits. If you
106//! need to work with a different identity, the `_with_key` variants allow providing
107//! an explicit key when creating branches or pulling workspaces.
108//!
109//! These parallels should help readers leverage their Git knowledge when
110//! working with trible repositories.
111//!
112/// Branch metadata construction and signature verification.
113pub mod branch;
114/// Capability-based authorization for triblespace networks.
115pub mod capability;
116/// Commit metadata construction and signature verification.
117pub mod commit;
118/// Storage adapter that delegates blobs and branches to separate backends.
119pub mod hybridstore;
120/// Fully in-memory repository implementation for tests and ephemeral use.
121pub mod memoryrepo;
122#[cfg(feature = "object-store")]
123/// Repository backed by an `object_store`-compatible remote (S3, local FS, etc.).
124pub mod objectstore;
125/// Local file-based pile storage backend.
126pub mod pile;
127
128/// Trait for storage backends that require explicit close/cleanup.
129///
130/// Not all storage backends need to implement this; implementations that have
131/// nothing to do on close may return Ok(()) or use `Infallible` as the error
132/// type.
133pub trait StorageClose {
134    /// Error type returned by `close`.
135    type Error: std::error::Error;
136
137    /// Consume the storage and perform any necessary cleanup.
138    fn close(self) -> Result<(), Self::Error>;
139}
140
141// Convenience impl for repositories whose storage supports explicit close.
142impl<Storage> Repository<Storage>
143where
144    Storage: BlobStore + PinStore + StorageClose,
145{
146    /// Close the repository's underlying storage if it supports explicit
147    /// close operations.
148    ///
149    /// This method is only available when the storage type implements
150    /// [`StorageClose`]. It consumes the repository and delegates to the
151    /// storage's `close` implementation, returning any error produced.
152    pub fn close(self) -> Result<(), <Storage as StorageClose>::Error> {
153        self.storage.close()
154    }
155}
156
157use crate::macros::pattern;
158use std::collections::{HashSet, VecDeque};
159use std::convert::Infallible;
160use std::error::Error;
161use std::fmt::Debug;
162use std::fmt::{self};
163
164use commit::commit_metadata;
165use hifitime::Epoch;
166use itertools::Itertools;
167
168use crate::blob::encodings::simplearchive::UnarchiveError;
169use crate::blob::encodings::UnknownBlob;
170use crate::blob::Blob;
171use crate::blob::BlobEncoding;
172use crate::blob::MemoryBlobStore;
173use crate::blob::IntoBlob;
174use crate::blob::TryFromBlob;
175use crate::find;
176use crate::id::genid;
177use crate::id::Id;
178use crate::patch::Entry;
179use crate::patch::IdentitySchema;
180use crate::patch::PATCH;
181use crate::prelude::inlineencodings::GenId;
182use crate::repo::branch::branch_metadata;
183use crate::trible::TribleSet;
184use crate::inline::encodings::hash::Handle;
185use crate::inline::Inline;
186use crate::inline::InlineEncoding;
187use crate::inline::INLINE_LEN;
188use ed25519_dalek::SigningKey;
189
190use crate::blob::encodings::longstring::LongString;
191use crate::blob::encodings::simplearchive::SimpleArchive;
192use crate::blob::encodings::succinctarchive::SuccinctArchiveBlob;
193use crate::prelude::*;
194use crate::inline::encodings::ed25519 as ed;
195use crate::inline::encodings::shortstring::ShortString;
196
197attributes! {
198    /// The actual data of the commit.
199    "4DD4DDD05CC31734B03ABB4E43188B1F" as pub content: Handle<SimpleArchive>;
200    /// Metadata describing the commit content.
201    "88B59BD497540AC5AECDB7518E737C87" as pub metadata: Handle<SimpleArchive>;
202    /// A commit that this commit is based on.
203    "317044B612C690000D798CA660ECFD2A" as pub parent: Handle<SimpleArchive>;
204    /// A (potentially long) message describing the commit.
205    "B59D147839100B6ED4B165DF76EDF3BB" as pub message: Handle<LongString>;
206    /// A short message describing the commit.
207    "12290C0BE0E9207E324F24DDE0D89300" as pub short_message: ShortString;
208    /// The hash of the first commit in the commit chain of the branch.
209    "272FBC56108F336C4D2E17289468C35F" as pub head: Handle<SimpleArchive>;
210    /// An id used to track the branch.
211    "8694CC73AF96A5E1C7635C677D1B928A" as pub branch: GenId;
212    /// The author of the signature identified by their ed25519 public key.
213    "ADB4FFAD247C886848161297EFF5A05B" as pub signed_by: ed::ED25519PublicKey;
214    /// The `r` part of a ed25519 signature.
215    "9DF34F84959928F93A3C40AEB6E9E499" as pub signature_r: ed::ED25519RComponent;
216    /// The `s` part of a ed25519 signature.
217    "1ACE03BF70242B289FDF00E4327C3BC6" as pub signature_s: ed::ED25519SComponent;
218    /// Optional SuccinctArchive rollup of the branch HEAD's logical contents.
219    ///
220    /// Readers can fetch this blob via the repository's blob store to obtain
221    /// a compact, instantly-queryable representation of the branch's state
222    /// without having to materialise the TribleSet from the commit chain.
223    /// Absent on branches that haven't had a rollup built yet. Soft state:
224    /// the rollup is redundant with (and must agree with) whatever
225    /// `ws.checkout(..)` would return for the same HEAD.
226    "D7D14C6737AA27A51E1E08D380D13EF9" as pub rollup: Handle<SuccinctArchiveBlob>;
227}
228
229/// The `ListBlobs` trait is used to list all blobs in a repository.
230pub trait BlobStoreList {
231    /// Iterator over blob handles in the store.
232    type Iter<'a>: Iterator<Item = Result<Inline<Handle<UnknownBlob>>, Self::Err>>
233    where
234        Self: 'a;
235    /// Error type for listing operations.
236    type Err: Error + Debug + Send + Sync + 'static;
237
238    /// Lists all blobs in the repository.
239    fn blobs<'a>(&'a self) -> Self::Iter<'a>;
240
241    /// Lists blobs in `self` that are not in `old`.
242    ///
243    /// Backends with true snapshot semantics (e.g. [`Pile`],
244    /// where each [`Reader`](BlobStore::Reader) holds a frozen clone of the
245    /// in-memory blob index) compute the difference cheaply via the index's
246    /// own set-difference operation. Backends without snapshot semantics
247    /// (e.g. an object store, where the Reader is just a handle to the live
248    /// remote) fall back to the default implementation, which lists all
249    /// current blobs — over-eager but always correct.
250    ///
251    /// Use this for "what blobs are new since I last looked" patterns
252    /// (e.g. announcing newly-imported blobs to a DHT) where holding the
253    /// previous Reader as a baseline gives you the delta.
254    fn blobs_diff<'a>(&'a self, _old: &Self) -> Self::Iter<'a> {
255        self.blobs()
256    }
257}
258
259/// Metadata about a blob in a repository.
260#[derive(Debug, Clone)]
261pub struct BlobMetadata {
262    /// Timestamp in milliseconds since UNIX epoch when the blob was created/stored.
263    pub timestamp: u64,
264    /// Length of the blob in bytes.
265    pub length: u64,
266}
267
268/// Trait exposing metadata lookup for blobs available in a repository reader.
269pub trait BlobStoreMeta {
270    /// Error type returned by metadata calls.
271    type MetaError: std::error::Error + Send + Sync + 'static;
272
273    /// Returns metadata for the blob identified by `handle`, or `None` if
274    /// the blob is not present.
275    fn metadata<S>(
276        &self,
277        handle: Inline<Handle<S>>,
278    ) -> Result<Option<BlobMetadata>, Self::MetaError>
279    where
280        S: BlobEncoding + 'static,
281        Handle<S>: InlineEncoding;
282}
283
284/// Trait exposing a monotonic "forget" operation.
285///
286/// Forget is idempotent and monotonic: it removes materialization from a
287/// particular repository but does not semantically delete derived facts.
288pub trait BlobStoreForget {
289    /// Error type for forget operations.
290    type ForgetError: std::error::Error + Send + Sync + 'static;
291
292    /// Removes the materialized blob identified by `handle` from this store.
293    fn forget<S>(&mut self, handle: Inline<Handle<S>>) -> Result<(), Self::ForgetError>
294    where
295        S: BlobEncoding + 'static,
296        Handle<S>: InlineEncoding;
297}
298
299/// The `GetBlob` trait is used to retrieve blobs from a repository.
300pub trait BlobStoreGet {
301    /// Error type for get operations, parameterised by the deserialization error.
302    type GetError<E: std::error::Error + Send + Sync + 'static>: Error + Send + Sync + 'static;
303
304    /// Retrieves a blob from the repository by its handle.
305    /// The handle is a unique identifier for the blob, and is used to retrieve it from the repository.
306    /// The blob is returned as a [`Blob`] object, which contains the raw bytes of the blob,
307    /// which can be deserialized via the appropriate schema type, which is specified by the `T` type parameter.
308    ///
309    /// # Errors
310    /// Returns an error if the blob could not be found in the repository.
311    /// The error type is specified by the `Err` associated type.
312    fn get<T, S>(
313        &self,
314        handle: Inline<Handle<S>>,
315    ) -> Result<T, Self::GetError<<T as TryFromBlob<S>>::Error>>
316    where
317        S: BlobEncoding + 'static,
318        T: TryFromBlob<S>,
319        Handle<S>: InlineEncoding;
320}
321
322/// The `PutBlob` trait is used to store blobs in a repository.
323pub trait BlobStorePut {
324    /// Error type for put operations.
325    type PutError: Error + Debug + Send + Sync + 'static;
326
327    /// Serialises `item` as a blob, stores it, and returns its handle.
328    fn put<S, T>(&mut self, item: T) -> Result<Inline<Handle<S>>, Self::PutError>
329    where
330        S: BlobEncoding + 'static,
331        T: IntoBlob<S>,
332        Handle<S>: InlineEncoding;
333}
334
335/// Combined read/write blob storage.
336///
337/// Extends [`BlobStorePut`] with the ability to create a shareable
338/// [`Reader`](BlobStore::Reader) snapshot for concurrent reads.
339pub trait BlobStore: BlobStorePut {
340    /// A clonable reader handle for concurrent blob lookups.
341    type Reader: BlobStoreGet + BlobStoreList + Clone + Send + PartialEq + Eq + 'static;
342    /// Error type for creating a reader.
343    type ReaderError: Error + Debug + Send + Sync + 'static;
344    /// Creates a shareable reader snapshot of the current store state.
345    fn reader(&mut self) -> Result<Self::Reader, Self::ReaderError>;
346}
347
348/// Trait for blob stores that can retain a supplied set of handles.
349pub trait BlobStoreKeep {
350    /// Retain only the blobs identified by `handles`.
351    fn keep<I>(&mut self, handles: I)
352    where
353        I: IntoIterator<Item = Inline<Handle<UnknownBlob>>>;
354}
355
356/// Trait for stores that can enumerate a blob's child references.
357///
358/// "Children" are the 32-byte-aligned values in a blob that correspond
359/// to existing blobs in the store — the conservative set of references.
360///
361/// The default implementation scans the blob's bytes and checks each
362/// 32-byte chunk with [`BlobStoreGet::get`]. Backends with batch
363/// capabilities (e.g. a network store with a SYNC protocol) can
364/// override this for efficiency.
365pub trait BlobChildren: BlobStoreGet {
366    /// Return handles of blobs referenced by `handle` that exist in this store.
367    fn children(
368        &self,
369        handle: Inline<Handle<UnknownBlob>>,
370    ) -> Vec<Inline<Handle<UnknownBlob>>> {
371        let Ok(blob) = self.get::<Blob<UnknownBlob>, UnknownBlob>(handle) else {
372            return Vec::new();
373        };
374        let bytes = blob.bytes.as_ref();
375        let mut result = Vec::new();
376        let mut offset = 0usize;
377        while offset + INLINE_LEN <= bytes.len() {
378            let mut raw = [0u8; INLINE_LEN];
379            raw.copy_from_slice(&bytes[offset..offset + INLINE_LEN]);
380            let candidate = Inline::<Handle<UnknownBlob>>::new(raw);
381            if self.get::<anybytes::Bytes, UnknownBlob>(candidate).is_ok() {
382                result.push(candidate);
383            }
384            offset += INLINE_LEN;
385        }
386        result
387    }
388}
389
390// No blanket impl — types opt in explicitly so they can provide
391// optimized implementations (e.g. network stores with batch protocols).
392// Use `impl_blob_children_default!` for the scan-and-check fallback.
393
394/// Outcome of a compare-and-swap pin update (used by both the
395/// primitive `PinStore::update` and the higher-level
396/// `Repository::push` for content branches).
397#[derive(Debug)]
398pub enum PushResult {
399    /// The CAS succeeded — the pin now points to the new value.
400    Success(),
401    /// The CAS failed — the pin's head had advanced. Contains the
402    /// current head, or `None` if the pin was tombstoned concurrently.
403    Conflict(Option<Inline<Handle<SimpleArchive>>>),
404}
405
406/// A point-in-time snapshot of (pin id → head) mappings.
407///
408/// PATCH keyed by 16-byte pin id, valued by the pinned head's handle.
409/// Cloning is O(1) (refcount bump), so this is the right primitive for
410/// handing pin state across threads or into long-lived serving views.
411///
412/// Returned by [`PinStore::pin_snapshot`].
413pub type PinSnapshot = PATCH<16, IdentitySchema, Inline<Handle<SimpleArchive>>>;
414
415/// Storage backend for pins: named, atomically-updatable handles to
416/// SimpleArchive blobs.
417///
418/// A *pin* is the storage primitive — a named cell holding a single
419/// `Inline<Handle<SimpleArchive>>`, updated via compare-and-swap. The
420/// pile's compaction sweep treats every pin head as a reachability
421/// root: blobs reachable from a pin survive; the rest are reclaimed.
422///
423/// Pins back several specialized use patterns, distinguished at
424/// higher layers via metadata markers:
425/// - A **branch** is a pin whose value resolves to a commit-chain
426///   head (Repository's content abstraction). Branch metadata
427///   carries `metadata::name` for human-readable lookup.
428/// - A **tracking pin** mirrors a remote peer's branch head and
429///   carries `tracking_remote_pin` + `remote_name`.
430/// - A **local-only pin** (renewal policy, pending requests,
431///   per-team cap holdings) carries `local_only_pin: <kind>` and is
432///   excluded from gossip publication.
433///
434/// `PinStore` itself doesn't know about these distinctions — it just
435/// provides the primitive: enumerate ids, read the current head, CAS
436/// an update. The two-level taxonomy lives at higher layers
437/// (decide#6de2dd95).
438///
439/// This trait is the stateful counterpart to [`BlobStore`]: blob
440/// stores are content-addressed and orderless; pin stores track a
441/// single mutable pointer per pin. The update operation uses
442/// compare-and-swap semantics so multiple writers can coordinate
443/// without locks.
444pub trait PinStore {
445    /// Error type for listing pins.
446    type PinsError: Error + Debug + Send + Sync + 'static;
447    /// Error type for head lookups.
448    type HeadError: Error + Debug + Send + Sync + 'static;
449    /// Error type for CAS updates.
450    type UpdateError: Error + Debug + Send + Sync + 'static;
451
452    /// Iterator over pin IDs.
453    type ListIter<'a>: Iterator<Item = Result<Id, Self::PinsError>>
454    where
455        Self: 'a;
456
457    /// Lists every pin in the store. Returns a fallible iterator over
458    /// pin ids (any role — branches, tracking pins, local-only pins).
459    /// Callers that want only content branches filter by checking for
460    /// the `metadata::name` attribute on each pin's head metadata.
461    fn pins<'a>(&'a mut self) -> Result<Self::ListIter<'a>, Self::PinsError>;
462
463    /// Cheap point-in-time snapshot of the (pin id → head) map.
464    ///
465    /// Returns a [`PinSnapshot`] — a PATCH keyed by pin id, valued by
466    /// the pinned head's handle. Cloning the returned PATCH is O(1)
467    /// (refcount bump), so this is also the right primitive for handing
468    /// the pin state to background threads / async tasks without
469    /// re-querying the store.
470    ///
471    /// The default impl walks `pins()` + `head()`. Stores with a
472    /// PATCH-backed pin index ([`crate::repo::pile::Pile`]) override to
473    /// clone the index directly. Head errors during the default walk
474    /// are skipped silently — partial snapshots are acceptable for the
475    /// "serving view" use case; callers that need strict atomicity
476    /// should drive [`pins`](Self::pins) + [`head`](Self::head)
477    /// themselves and handle errors.
478    fn pin_snapshot(&mut self) -> Result<PinSnapshot, Self::PinsError> {
479        let mut out = PinSnapshot::new();
480        let ids: Vec<Id> = self.pins()?.filter_map(|r| r.ok()).collect();
481        for id in ids {
482            if let Ok(Some(h)) = self.head(id) {
483                let bid: [u8; 16] = id.into();
484                let entry = Entry::with_value(&bid, h);
485                out.insert(&entry);
486            }
487        }
488        Ok(out)
489    }
490
491    /// Retrieves the current head of a pin by its id.
492    ///
493    /// Returns `Ok(Some(handle))` if the pin exists and has a head,
494    /// `Ok(None)` if the pin is tombstoned (deleted), and an error if
495    /// the underlying store failed to read.
496    ///
497    /// # Parameters
498    /// * `id` — The id of the pin to look up.
499    fn head(&mut self, id: Id) -> Result<Option<Inline<Handle<SimpleArchive>>>, Self::HeadError>;
500
501    /// Compare-and-swap update of a pin's head.
502    ///
503    /// Used to create a fresh pin, advance an existing one, or
504    /// tombstone (delete) one. The CAS guard (`old`) lets multiple
505    /// writers coordinate without locks: a stale writer's update
506    /// returns `PushResult::Conflict(current)` carrying the actual
507    /// current head for retry / merge.
508    ///
509    /// # Parameters
510    /// * `id` — The id of the pin to update.
511    /// * `old` — Expected current head (`None` when creating a fresh pin).
512    /// * `new` — New head (`None` tombstones the pin).
513    ///
514    /// # Returns
515    /// * `Success` — The pin now points at `new`.
516    /// * `Conflict(current)` — Some other writer advanced first; the
517    ///   pin's current head is `current`.
518    fn update(
519        &mut self,
520        id: Id,
521        old: Option<Inline<Handle<SimpleArchive>>>,
522        new: Option<Inline<Handle<SimpleArchive>>>,
523    ) -> Result<PushResult, Self::UpdateError>;
524}
525
526/// Error returned by [`transfer`] when copying blobs between stores.
527#[derive(Debug)]
528pub enum TransferError<ListErr, LoadErr, StoreErr> {
529    /// Failed to list handles from the source.
530    List(ListErr),
531    /// Failed to load a blob from the source.
532    Load(LoadErr),
533    /// Failed to store a blob in the target.
534    Store(StoreErr),
535}
536
537impl<ListErr, LoadErr, StoreErr> fmt::Display for TransferError<ListErr, LoadErr, StoreErr> {
538    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
539        write!(f, "failed to transfer blob")
540    }
541}
542
543impl<ListErr, LoadErr, StoreErr> Error for TransferError<ListErr, LoadErr, StoreErr>
544where
545    ListErr: Debug + Error + 'static,
546    LoadErr: Debug + Error + 'static,
547    StoreErr: Debug + Error + 'static,
548{
549    fn source(&self) -> Option<&(dyn Error + 'static)> {
550        match self {
551            Self::List(e) => Some(e),
552            Self::Load(e) => Some(e),
553            Self::Store(e) => Some(e),
554        }
555    }
556}
557
558/// Copies the specified blob handles from `source` into `target`.
559pub fn transfer<'a, BS, BT, Handles>(
560    source: &'a BS,
561    target: &'a mut BT,
562    handles: Handles,
563) -> impl Iterator<
564    Item = Result<
565        (
566            Inline<Handle<UnknownBlob>>,
567            Inline<Handle<UnknownBlob>>,
568        ),
569        TransferError<
570            Infallible,
571            <BS as BlobStoreGet>::GetError<Infallible>,
572            <BT as BlobStorePut>::PutError,
573        >,
574    >,
575> + 'a
576where
577    BS: BlobStoreGet + 'a,
578    BT: BlobStorePut + 'a,
579    Handles: IntoIterator<Item = Inline<Handle<UnknownBlob>>> + 'a,
580    Handles::IntoIter: 'a,
581{
582    handles.into_iter().map(move |source_handle| {
583        let blob: Blob<UnknownBlob> = source.get(source_handle).map_err(TransferError::Load)?;
584
585        Ok((
586            source_handle,
587            (target.put(blob).map_err(TransferError::Store)?),
588        ))
589    })
590}
591
592/// Iterator that visits every blob handle reachable from a set of roots.
593///
594/// Uses [`BlobChildren`] to enumerate references at each level,
595/// so backends with batch capabilities get efficient traversal.
596pub struct ReachableHandles<'a, BS>
597where
598    BS: BlobChildren,
599{
600    source: &'a BS,
601    queue: VecDeque<Inline<Handle<UnknownBlob>>>,
602    visited: HashSet<[u8; INLINE_LEN]>,
603}
604
605impl<'a, BS> ReachableHandles<'a, BS>
606where
607    BS: BlobChildren,
608{
609    fn new(source: &'a BS, roots: impl IntoIterator<Item = Inline<Handle<UnknownBlob>>>) -> Self {
610        let mut queue = VecDeque::new();
611        for handle in roots {
612            queue.push_back(handle);
613        }
614
615        Self {
616            source,
617            queue,
618            visited: HashSet::new(),
619        }
620    }
621}
622
623impl<'a, BS> Iterator for ReachableHandles<'a, BS>
624where
625    BS: BlobChildren,
626{
627    type Item = Inline<Handle<UnknownBlob>>;
628
629    fn next(&mut self) -> Option<Self::Item> {
630        while let Some(handle) = self.queue.pop_front() {
631            let raw = handle.raw;
632
633            if !self.visited.insert(raw) {
634                continue;
635            }
636
637            // Use BlobChildren to get references — backends can override
638            // with batch-optimized implementations.
639            for child in self.source.children(handle) {
640                if !self.visited.contains(&child.raw) {
641                    self.queue.push_back(child);
642                }
643            }
644
645            return Some(handle);
646        }
647
648        None
649    }
650}
651
652/// Create a breadth-first iterator over blob handles reachable from `roots`.
653///
654/// Uses [`BlobChildren`] for reference enumeration, so network-backed
655/// stores can provide optimized batch implementations.
656pub fn reachable<'a, BS>(
657    source: &'a BS,
658    roots: impl IntoIterator<Item = Inline<Handle<UnknownBlob>>>,
659) -> ReachableHandles<'a, BS>
660where
661    BS: BlobChildren,
662{
663    ReachableHandles::new(source, roots)
664}
665
666/// Iterate over every 32-byte candidate in the value column of a [`TribleSet`].
667///
668/// This is a conservative conversion used when scanning metadata for potential
669/// blob handles. Each 32-byte chunk is treated as a `Handle<UnknownBlob>`.
670/// Callers can feed the resulting iterator into [`BlobStoreKeep::keep`] or other
671/// helpers that accept collections of handles.
672pub fn potential_handles<'a>(
673    set: &'a TribleSet,
674) -> impl Iterator<Item = Inline<Handle<UnknownBlob>>> + 'a {
675    set.vae.iter().map(|raw| {
676        let mut value = [0u8; INLINE_LEN];
677        value.copy_from_slice(&raw[0..INLINE_LEN]);
678        Inline::<Handle<UnknownBlob>>::new(value)
679    })
680}
681
682/// An error that can occur when creating a commit.
683/// This error can be caused by a failure to store the content or metadata blobs.
684#[derive(Debug)]
685pub enum CreateCommitError<BlobErr: Error + Debug + Send + Sync + 'static> {
686    /// Failed to store the content blob.
687    ContentStorageError(BlobErr),
688    /// Failed to store the commit metadata blob.
689    CommitStorageError(BlobErr),
690}
691
692impl<BlobErr: Error + Debug + Send + Sync + 'static> fmt::Display for CreateCommitError<BlobErr> {
693    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
694        match self {
695            CreateCommitError::ContentStorageError(e) => write!(f, "Content storage failed: {e}"),
696            CreateCommitError::CommitStorageError(e) => {
697                write!(f, "Commit metadata storage failed: {e}")
698            }
699        }
700    }
701}
702
703impl<BlobErr: Error + Debug + Send + Sync + 'static> Error for CreateCommitError<BlobErr> {
704    fn source(&self) -> Option<&(dyn Error + 'static)> {
705        match self {
706            CreateCommitError::ContentStorageError(e) => Some(e),
707            CreateCommitError::CommitStorageError(e) => Some(e),
708        }
709    }
710}
711
712/// Error returned by [`Workspace::merge`].
713#[derive(Debug)]
714pub enum MergeError {
715    /// The merge failed because the workspaces have different base repos.
716    DifferentRepos(),
717    /// The ancestry walk failed because one or more commit blobs along the
718    /// chain weren't readable from the workspace's view. The merge refuses
719    /// to fall through to a divergent-merge in this case — creating a merge
720    /// commit referencing an unknown chain would leave a dangling parent in
721    /// the resulting branch, and the append-only pile keeps that corruption
722    /// forever.
723    ///
724    /// Callers should ensure both heads' full closures are locally present
725    /// (e.g. via `fetch_reachable`) before retrying. The contained string
726    /// is a human-readable description of the underlying read failure.
727    AncestryWalkFailed(String),
728}
729
730/// Error returned by [`Repository::push`] and [`Repository::try_push`].
731/// Error type for [`Repository::compute_rollup`].
732#[derive(Debug)]
733pub enum RollupError<Storage: PinStore + BlobStore> {
734    /// The branch was not found in the underlying storage.
735    UnknownBranch,
736    /// The branch is empty — no HEAD to roll up.
737    EmptyBranch,
738    /// The branch HEAD advanced between checkout and CAS-update. The
739    /// caller may retry (`compute_rollup` is content-addressed so repeat
740    /// calls dedupe against already-uploaded blobs).
741    HeadAdvanced,
742    /// Underlying push / storage error during the attach step.
743    Push(PushError<Storage>),
744    /// Could not pull the branch to obtain a workspace.
745    Pull(PullError<Storage::HeadError,
746                    <Storage as BlobStore>::ReaderError,
747                    <<Storage as BlobStore>::Reader as BlobStoreGet>::GetError<UnarchiveError>>),
748    /// Could not check out the branch state to build the archive.
749    Checkout(WorkspaceCheckoutError<
750        <<Storage as BlobStore>::Reader as BlobStoreGet>::GetError<UnarchiveError>>),
751}
752
753#[derive(Debug)]
754pub enum PushError<Storage: PinStore + BlobStore> {
755    /// An error occurred while enumerating the branch storage branches.
756    StorageBranches(Storage::PinsError),
757    /// An error occurred while creating a blob reader.
758    StorageReader(<Storage as BlobStore>::ReaderError),
759    /// An error occurred while reading metadata blobs.
760    StorageGet(
761        <<Storage as BlobStore>::Reader as BlobStoreGet>::GetError<UnarchiveError>,
762    ),
763    /// An error occurred while transferring blobs to the repository.
764    StoragePut(<Storage as BlobStorePut>::PutError),
765    /// An error occurred while updating the branch storage.
766    BranchUpdate(Storage::UpdateError),
767    /// Malformed branch metadata.
768    BadBranchMetadata(),
769    /// Merge failed while retrying a push.
770    MergeError(MergeError),
771}
772
773// Allow using the `?` operator to convert MergeError into PushError in
774// contexts where PushError is the function error type. This keeps call sites
775// succinct by avoiding manual mapping closures like
776// `.map_err(|e| PushError::MergeError(e))?`.
777impl<Storage> From<MergeError> for PushError<Storage>
778where
779    Storage: PinStore + BlobStore,
780{
781    fn from(e: MergeError) -> Self {
782        PushError::MergeError(e)
783    }
784}
785
786// Note: we intentionally avoid generic `From` impls for storage-associated
787// error types because they can overlap with other blanket implementations
788// and lead to coherence conflicts. Call sites use explicit mapping via the
789// enum variant constructors (e.g. `map_err(PushError::StoragePut)`) where
790// needed which keeps conversions explicit and stable.
791
792/// Error returned by [`Repository::create_branch`] and related methods.
793#[derive(Debug)]
794pub enum BranchError<Storage>
795where
796    Storage: PinStore + BlobStore,
797{
798    /// An error occurred while creating a blob reader.
799    StorageReader(<Storage as BlobStore>::ReaderError),
800    /// An error occurred while reading metadata blobs.
801    StorageGet(
802        <<Storage as BlobStore>::Reader as BlobStoreGet>::GetError<UnarchiveError>,
803    ),
804    /// An error occurred while storing blobs.
805    StoragePut(<Storage as BlobStorePut>::PutError),
806    /// An error occurred while retrieving branch heads.
807    BranchHead(Storage::HeadError),
808    /// An error occurred while updating the branch storage.
809    BranchUpdate(Storage::UpdateError),
810    /// The branch already exists.
811    AlreadyExists(),
812    /// The referenced base branch does not exist.
813    BranchNotFound(Id),
814}
815
816/// Error returned by [`Repository::lookup_branch`].
817#[derive(Debug)]
818pub enum LookupError<Storage>
819where
820    Storage: PinStore + BlobStore,
821{
822    /// Failed to enumerate branches.
823    StorageBranches(Storage::PinsError),
824    /// Failed to read a branch head.
825    BranchHead(Storage::HeadError),
826    /// Failed to create a blob reader.
827    StorageReader(<Storage as BlobStore>::ReaderError),
828    /// Failed to read a metadata blob.
829    StorageGet(
830        <<Storage as BlobStore>::Reader as BlobStoreGet>::GetError<UnarchiveError>,
831    ),
832    /// Multiple branches were found with the given name.
833    NameConflict(Vec<Id>),
834    /// Branch metadata is malformed.
835    BadBranchMetadata(),
836}
837
838/// Error returned by [`Repository::ensure_branch`].
839#[derive(Debug)]
840pub enum EnsureBranchError<Storage>
841where
842    Storage: PinStore + BlobStore,
843{
844    /// Failed to look up the branch.
845    Lookup(LookupError<Storage>),
846    /// Failed to create the branch.
847    Create(BranchError<Storage>),
848}
849
850/// High-level wrapper combining a blob store and branch store into a usable
851/// repository API.
852///
853/// The [`Repository`] type exposes convenience methods for creating branches,
854/// committing data and pushing changes while delegating actual storage to the
855/// given [`BlobStore`] and [`PinStore`] implementations.
856pub struct Repository<Storage: BlobStore + PinStore> {
857    storage: Storage,
858    signing_key: SigningKey,
859    commit_metadata: MetadataHandle,
860}
861
862/// Error returned by [`Repository::pull`].
863pub enum PullError<BranchStorageErr, BlobReaderErr, BlobStorageErr>
864where
865    BranchStorageErr: Error,
866    BlobReaderErr: Error,
867    BlobStorageErr: Error,
868{
869    /// The branch does not exist in the repository.
870    BranchNotFound(Id),
871    /// An error occurred while accessing the branch storage.
872    BranchStorage(BranchStorageErr),
873    /// An error occurred while creating a blob reader.
874    BlobReader(BlobReaderErr),
875    /// An error occurred while accessing the blob storage.
876    BlobStorage(BlobStorageErr),
877    /// The branch metadata is malformed or does not contain the expected fields.
878    BadBranchMetadata(),
879}
880
881impl<B, R, C> fmt::Debug for PullError<B, R, C>
882where
883    B: Error + fmt::Debug,
884    R: Error + fmt::Debug,
885    C: Error + fmt::Debug,
886{
887    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
888        match self {
889            PullError::BranchNotFound(id) => f.debug_tuple("BranchNotFound").field(id).finish(),
890            PullError::BranchStorage(e) => f.debug_tuple("BranchStorage").field(e).finish(),
891            PullError::BlobReader(e) => f.debug_tuple("BlobReader").field(e).finish(),
892            PullError::BlobStorage(e) => f.debug_tuple("BlobStorage").field(e).finish(),
893            PullError::BadBranchMetadata() => f.debug_tuple("BadBranchMetadata").finish(),
894        }
895    }
896}
897
898impl<Storage> Repository<Storage>
899where
900    Storage: BlobStore + PinStore,
901{
902    /// Creates a new repository with the given storage, signing key, and
903    /// repo-wide commit metadata.
904    ///
905    /// `commit_metadata` accepts anything convertible into a [`Fragment`] —
906    /// either a raw [`TribleSet`] (auto-promoted with empty blob store via
907    /// `impl From<TribleSet> for Fragment`), or a Fragment built up via
908    /// `entity!{}` / `attributes!::describe()` that carries auxiliary blobs
909    /// (e.g. `Handle<LongString>` doc strings). The Fragment's blobs are
910    /// absorbed into storage so handles referenced by the metadata facts
911    /// stay resolvable for any downstream reader that pulls a commit and
912    /// calls [`Workspace::checkout_metadata`].
913    ///
914    /// The resulting metadata blob is referenced from every commit produced
915    /// by workspaces of this repository.
916    pub fn new<F: Into<crate::trible::Fragment>>(
917        mut storage: Storage,
918        signing_key: SigningKey,
919        commit_metadata: F,
920    ) -> Result<Self, <Storage as BlobStorePut>::PutError> {
921        let (facts, mut blobs) = commit_metadata.into().into_facts_and_blobs();
922        // Persist any blobs the Fragment carried — typically `Handle<LongString>`
923        // doc strings or other handle-referenced payloads. They're stored as
924        // `UnknownBlob` (raw bytes) because the storage layer is encoding-agnostic;
925        // readers recover the schema via the handle's declared encoding.
926        let reader = blobs
927            .reader()
928            .expect("MemoryBlobStore::reader is infallible");
929        for (_handle, blob) in reader {
930            storage.put::<UnknownBlob, _>(blob)?;
931        }
932        let commit_metadata = storage.put(facts)?;
933        Ok(Self {
934            storage,
935            signing_key,
936            commit_metadata,
937        })
938    }
939
940    /// Consume the repository and return the underlying storage backend.
941    ///
942    /// This is useful for callers that need to take ownership of the storage
943    /// (for example to call `close()` on a [`Pile`]) instead of letting the
944    /// repository drop it implicitly.
945    pub fn into_storage(self) -> Storage {
946        self.storage
947    }
948
949    /// Borrow the underlying storage backend.
950    pub fn storage(&self) -> &Storage {
951        &self.storage
952    }
953
954    /// Borrow the underlying storage backend mutably.
955    pub fn storage_mut(&mut self) -> &mut Storage {
956        &mut self.storage
957    }
958
959    /// Replace the repository signing key.
960    pub fn set_signing_key(&mut self, signing_key: SigningKey) {
961        self.signing_key = signing_key;
962    }
963
964    /// Returns the repository commit metadata handle.
965    pub fn commit_metadata(&self) -> MetadataHandle {
966        self.commit_metadata
967    }
968
969    /// Initializes a new branch in the repository.
970    /// Branches are the only mutable state in the repository,
971    /// and are used to represent the state of a commit chain at a specific point in time.
972    /// A branch must always point to a commit, and this function can be used to create a new branch.
973    ///
974    /// Creates a new branch in the repository.
975    /// This branch is a pointer to a specific commit in the repository.
976    /// The branch is created with name and is initialized to point to the opionally given commit.
977    /// The branch is signed by the branch signing key.
978    ///
979    /// # Parameters
980    /// * `branch_name` - Name of the new branch.
981    /// * `commit` - Commit to initialize the branch from.
982    pub fn create_branch(
983        &mut self,
984        branch_name: &str,
985        commit: Option<CommitHandle>,
986    ) -> Result<ExclusiveId, BranchError<Storage>> {
987        self.create_branch_with_key(branch_name, commit, self.signing_key.clone())
988    }
989
990    /// Same as [`Self::create_branch`] but uses the provided signing key.
991    pub fn create_branch_with_key(
992        &mut self,
993        branch_name: &str,
994        commit: Option<CommitHandle>,
995        signing_key: SigningKey,
996    ) -> Result<ExclusiveId, BranchError<Storage>> {
997        let branch_id = genid();
998        let name_blob: Blob<LongString> = branch_name.to_owned().to_blob();
999        let name_handle = name_blob.get_handle();
1000        self.storage
1001            .put::<LongString, _>(name_blob)
1002            .map_err(|e| BranchError::StoragePut(e))?;
1003
1004        let branch_set = if let Some(commit) = commit {
1005            let reader = self
1006                .storage
1007                .reader()
1008                .map_err(|e| BranchError::StorageReader(e))?;
1009            let set: TribleSet = reader.get(commit).map_err(|e| BranchError::StorageGet(e))?;
1010
1011            branch::branch_metadata(
1012                &signing_key,
1013                *branch_id,
1014                name_handle,
1015                Some(set.to_blob()),
1016                None,
1017            )
1018        } else {
1019            branch::branch_unsigned(*branch_id, name_handle, None, None)
1020        };
1021
1022        let branch_blob = branch_set.to_blob();
1023        let branch_handle = self
1024            .storage
1025            .put(branch_blob)
1026            .map_err(|e| BranchError::StoragePut(e))?;
1027        let push_result = self
1028            .storage
1029            .update(*branch_id, None, Some(branch_handle))
1030            .map_err(|e| BranchError::BranchUpdate(e))?;
1031
1032        match push_result {
1033            PushResult::Success() => Ok(branch_id),
1034            PushResult::Conflict(_) => Err(BranchError::AlreadyExists()),
1035        }
1036    }
1037
1038    /// Look up a branch by name.
1039    ///
1040    /// Iterates all branches, reads each one's metadata, and returns the ID
1041    /// of the branch whose name matches. Returns `Ok(None)` if no branch has
1042    /// that name, or `LookupError::NameConflict` if multiple branches share it.
1043    pub fn lookup_branch(&mut self, name: &str) -> Result<Option<Id>, LookupError<Storage>> {
1044        let branch_ids: Vec<Id> = self
1045            .storage
1046            .pins()
1047            .map_err(LookupError::StorageBranches)?
1048            .collect::<Result<Vec<_>, _>>()
1049            .map_err(LookupError::StorageBranches)?;
1050
1051        let mut matches = Vec::new();
1052
1053        for branch_id in branch_ids {
1054            let Some(meta_handle) = self
1055                .storage
1056                .head(branch_id)
1057                .map_err(LookupError::BranchHead)?
1058            else {
1059                continue;
1060            };
1061
1062            let reader = self.storage.reader().map_err(LookupError::StorageReader)?;
1063            let meta_set: TribleSet = reader.get(meta_handle).map_err(LookupError::StorageGet)?;
1064
1065            let Ok((name_handle,)) = find!(
1066                (n: Inline<Handle<LongString>>),
1067                pattern!(&meta_set, [{ crate::metadata::name: ?n }])
1068            )
1069            .exactly_one() else {
1070                continue;
1071            };
1072
1073            let Ok(branch_name): Result<anybytes::View<str>, _> = reader.get(name_handle) else {
1074                continue;
1075            };
1076
1077            if branch_name.as_ref() == name {
1078                matches.push(branch_id);
1079            }
1080        }
1081
1082        match matches.len() {
1083            0 => Ok(None),
1084            1 => Ok(Some(matches[0])),
1085            _ => Err(LookupError::NameConflict(matches)),
1086        }
1087    }
1088
1089    /// Ensure a branch with the given name exists, creating it if necessary.
1090    ///
1091    /// If a branch named `name` already exists, returns its ID.
1092    /// If no such branch exists, creates a new one (optionally from the given
1093    /// commit) and returns its ID.
1094    ///
1095    /// Errors if multiple branches share the same name (ambiguous).
1096    pub fn ensure_branch(
1097        &mut self,
1098        name: &str,
1099        commit: Option<CommitHandle>,
1100    ) -> Result<Id, EnsureBranchError<Storage>> {
1101        match self
1102            .lookup_branch(name)
1103            .map_err(EnsureBranchError::Lookup)?
1104        {
1105            Some(id) => Ok(id),
1106            None => {
1107                let id = self
1108                    .create_branch(name, commit)
1109                    .map_err(EnsureBranchError::Create)?;
1110                Ok(*id)
1111            }
1112        }
1113    }
1114
1115    /// Pulls an existing branch using the repository's signing key.
1116    /// The workspace inherits the repository default metadata if configured.
1117    pub fn pull(
1118        &mut self,
1119        branch_id: Id,
1120    ) -> Result<
1121        Workspace<Storage>,
1122        PullError<
1123            Storage::HeadError,
1124            Storage::ReaderError,
1125            <Storage::Reader as BlobStoreGet>::GetError<UnarchiveError>,
1126        >,
1127    > {
1128        self.pull_with_key(branch_id, self.signing_key.clone())
1129    }
1130
1131    /// Same as [`Self::pull`] but overrides the signing key.
1132    pub fn pull_with_key(
1133        &mut self,
1134        branch_id: Id,
1135        signing_key: SigningKey,
1136    ) -> Result<
1137        Workspace<Storage>,
1138        PullError<
1139            Storage::HeadError,
1140            Storage::ReaderError,
1141            <Storage::Reader as BlobStoreGet>::GetError<UnarchiveError>,
1142        >,
1143    > {
1144        // 1. Get the branch metadata head from the branch store.
1145        let base_branch_meta_handle = match self.storage.head(branch_id) {
1146            Ok(Some(handle)) => handle,
1147            Ok(None) => return Err(PullError::BranchNotFound(branch_id)),
1148            Err(e) => return Err(PullError::BranchStorage(e)),
1149        };
1150        // 2. Get the current commit from the branch metadata.
1151        let reader = self.storage.reader().map_err(PullError::BlobReader)?;
1152        let base_branch_meta: TribleSet = match reader.get(base_branch_meta_handle) {
1153            Ok(meta_set) => meta_set,
1154            Err(e) => return Err(PullError::BlobStorage(e)),
1155        };
1156
1157        let head_ = match find!(
1158            (head_: Inline<_>),
1159            pattern!(&base_branch_meta, [{ head: ?head_ }])
1160        )
1161        .at_most_one()
1162        {
1163            Ok(Some((h,))) => Some(h),
1164            Ok(None) => None,
1165            Err(_) => return Err(PullError::BadBranchMetadata()),
1166        };
1167        // Create workspace with the current commit and base blobs.
1168        let base_blobs = self.storage.reader().map_err(PullError::BlobReader)?;
1169        Ok(Workspace {
1170            base_blobs,
1171            staged: MemoryBlobStore::new(),
1172            head: head_,
1173            base_head: head_,
1174            base_branch_id: branch_id,
1175            base_branch_meta: base_branch_meta_handle,
1176            signing_key,
1177            commit_metadata: self.commit_metadata,
1178        })
1179    }
1180
1181    /// Pushes the workspace's new blobs and commit to the persistent repository.
1182    /// This syncs the local BlobSet with the repository's BlobStore and performs
1183    /// an atomic branch update (using the stored base_branch_meta).
1184    pub fn push(&mut self, workspace: &mut Workspace<Storage>) -> Result<(), PushError<Storage>> {
1185        // Retrying push: attempt a single push and, on conflict, merge the
1186        // local workspace into the returned conflict workspace and retry.
1187        // This implements the common push-merge-retry loop as a convenience
1188        // wrapper around `try_push`.
1189        while let Some(mut conflict_ws) = self.try_push(workspace)? {
1190            // Keep the previous merge order: merge the caller's staged
1191            // changes into the incoming conflict workspace. This preserves
1192            // the semantic ordering of parents used in the merge commit.
1193            conflict_ws.merge(workspace)?;
1194
1195            // Move the merged incoming workspace into the caller's workspace
1196            // so the next try_push operates against the fresh branch state.
1197            // Using assignment here is equivalent to `swap` but avoids
1198            // retaining the previous `workspace` contents in the temp var.
1199            *workspace = conflict_ws;
1200        }
1201
1202        Ok(())
1203    }
1204
1205    /// Single-attempt push: upload local blobs and try to update the branch
1206    /// head once. Returns `Ok(None)` on success, or `Ok(Some(conflict_ws))`
1207    /// when the branch was updated concurrently and the caller should merge.
1208    pub fn try_push(
1209        &mut self,
1210        workspace: &mut Workspace<Storage>,
1211    ) -> Result<Option<Workspace<Storage>>, PushError<Storage>> {
1212        // 1. Sync `workspace.staged` to repository's BlobStore.
1213        let workspace_reader = workspace.staged.reader().unwrap();
1214        for handle in workspace_reader.blobs() {
1215            let handle = handle.expect("infallible blob enumeration");
1216            let blob: Blob<UnknownBlob> =
1217                workspace_reader.get(handle).expect("infallible blob read");
1218            self.storage
1219                .put::<UnknownBlob, _>(blob)
1220                .map_err(PushError::StoragePut)?;
1221        }
1222
1223        // 1.5 If the workspace's head did not change since the workspace was
1224        // created, there's no commit to reference and therefore no branch
1225        // metadata update is required. This avoids touching the branch store
1226        // in the common case where only blobs were staged or nothing changed.
1227        if workspace.base_head == workspace.head {
1228            return Ok(None);
1229        }
1230
1231        // 2. Create a new branch meta blob referencing the new workspace head.
1232        let repo_reader = self.storage.reader().map_err(PushError::StorageReader)?;
1233        let base_branch_meta: TribleSet = repo_reader
1234            .get(workspace.base_branch_meta)
1235            .map_err(PushError::StorageGet)?;
1236
1237        let Ok((branch_name,)) = find!(
1238            (name: Inline<Handle<LongString>>),
1239            pattern!(base_branch_meta, [{ crate::metadata::name: ?name }])
1240        )
1241        .exactly_one() else {
1242            return Err(PushError::BadBranchMetadata());
1243        };
1244
1245        let head_handle = workspace.head.ok_or(PushError::BadBranchMetadata())?;
1246        let head_: TribleSet = repo_reader
1247            .get(head_handle)
1248            .map_err(PushError::StorageGet)?;
1249
1250        let branch_meta = branch_metadata(
1251            &workspace.signing_key,
1252            workspace.base_branch_id,
1253            branch_name,
1254            Some(head_.to_blob()),
1255            // A fresh commit invalidates any prior rollup (it was computed
1256            // against the old HEAD). Readers fall back to checkout until
1257            // `compute_rollup` runs against the new HEAD.
1258            None,
1259        );
1260
1261        let branch_meta_handle = self
1262            .storage
1263            .put(branch_meta)
1264            .map_err(PushError::StoragePut)?;
1265
1266        // 3. Use CAS (comparing against workspace.base_branch_meta) to update the branch pointer.
1267        let result = self
1268            .storage
1269            .update(
1270                workspace.base_branch_id,
1271                Some(workspace.base_branch_meta),
1272                Some(branch_meta_handle),
1273            )
1274            .map_err(PushError::BranchUpdate)?;
1275
1276        match result {
1277            PushResult::Success() => {
1278                // Update workspace base pointers so subsequent pushes can detect
1279                // that the workspace is already synchronized and avoid re-upload.
1280                workspace.base_branch_meta = branch_meta_handle;
1281                workspace.base_head = workspace.head;
1282                // Refresh the workspace base blob reader to ensure newly
1283                // uploaded blobs are visible to subsequent checkout operations.
1284                workspace.base_blobs = self.storage.reader().map_err(PushError::StorageReader)?;
1285                // Clear staged local blobs now that they have been uploaded and
1286                // the branch metadata updated. This frees memory and prevents
1287                // repeated uploads of the same staged blobs on subsequent pushes.
1288                workspace.staged = MemoryBlobStore::new();
1289                Ok(None)
1290            }
1291            PushResult::Conflict(conflicting_meta) => {
1292                let conflicting_meta = conflicting_meta.ok_or(PushError::BadBranchMetadata())?;
1293
1294                let repo_reader = self.storage.reader().map_err(PushError::StorageReader)?;
1295                let branch_meta: TribleSet = repo_reader
1296                    .get(conflicting_meta)
1297                    .map_err(PushError::StorageGet)?;
1298
1299                let head_ = match find!((head_: Inline<_>),
1300                    pattern!(&branch_meta, [{ head: ?head_ }])
1301                )
1302                .at_most_one()
1303                {
1304                    Ok(Some((h,))) => Some(h),
1305                    Ok(None) => None,
1306                    Err(_) => return Err(PushError::BadBranchMetadata()),
1307                };
1308
1309                let conflict_ws = Workspace {
1310                    base_blobs: self.storage.reader().map_err(PushError::StorageReader)?,
1311                    staged: MemoryBlobStore::new(),
1312                    head: head_,
1313                    base_head: head_,
1314                    base_branch_id: workspace.base_branch_id,
1315                    base_branch_meta: conflicting_meta,
1316                    signing_key: workspace.signing_key.clone(),
1317                    commit_metadata: workspace.commit_metadata,
1318                };
1319
1320                Ok(Some(conflict_ws))
1321            }
1322        }
1323    }
1324
1325    /// Builds a [`SuccinctArchive`](crate::blob::encodings::succinctarchive::SuccinctArchive) rollup of the branch's current HEAD,
1326    /// stores it as a blob in the underlying storage, and attaches the
1327    /// resulting handle to the branch metadata via CAS.
1328    ///
1329    /// Returns the new rollup handle on success.
1330    ///
1331    /// Returns [`RollupError::HeadAdvanced`] if the branch HEAD moved
1332    /// between `pull` and the CAS-update. The caller may retry — the
1333    /// archive blob is content-addressed, so subsequent calls dedupe
1334    /// against already-uploaded blobs.
1335    ///
1336    /// Returns [`RollupError::EmptyBranch`] if the branch has no HEAD
1337    /// commit yet (nothing to roll up).
1338    ///
1339    /// This is the sole public write-path for rollups. The companion
1340    /// read-path is [`Workspace::rollup`].
1341    pub fn compute_rollup(
1342        &mut self,
1343        branch_id: Id,
1344    ) -> Result<
1345        Inline<Handle<crate::blob::encodings::succinctarchive::SuccinctArchiveBlob>>,
1346        RollupError<Storage>,
1347    > {
1348        use crate::blob::encodings::succinctarchive::{OrderedUniverse, SuccinctArchive};
1349        use crate::blob::IntoBlob;
1350
1351        let mut ws = self.pull(branch_id).map_err(RollupError::Pull)?;
1352        let head_handle = ws.head().ok_or(RollupError::EmptyBranch)?;
1353
1354        // Materialise the branch state from its commit chain and build the
1355        // succinct index over it.
1356        let space = ws.checkout(..).map_err(RollupError::Checkout)?;
1357        let archive: SuccinctArchive<OrderedUniverse> = (&*space).into();
1358        drop(space);
1359
1360        // Upload the archive blob directly to storage — no workspace-local
1361        // staging needed; the CAS below references it by handle.
1362        let archive_blob = (&archive).to_blob();
1363        let handle: Inline<
1364            Handle<crate::blob::encodings::succinctarchive::SuccinctArchiveBlob>,
1365        > = self
1366            .storage
1367            .put(archive_blob)
1368            .map_err(|e| RollupError::Push(PushError::StoragePut(e)))?;
1369
1370        // Construct a fresh branch meta that carries the same head as
1371        // `base_branch_meta` plus the new rollup attribute.
1372        let reader = self
1373            .storage
1374            .reader()
1375            .map_err(|e| RollupError::Push(PushError::StorageReader(e)))?;
1376        let base_meta: TribleSet = reader
1377            .get(ws.base_branch_meta)
1378            .map_err(|e| RollupError::Push(PushError::StorageGet(e)))?;
1379        let (branch_name,) = find!(
1380            (name: Inline<Handle<LongString>>),
1381            pattern!(&base_meta, [{ crate::metadata::name: ?name }])
1382        )
1383        .exactly_one()
1384        .map_err(|_| RollupError::Push(PushError::BadBranchMetadata()))?;
1385        let head_blob: TribleSet = reader
1386            .get(head_handle)
1387            .map_err(|e| RollupError::Push(PushError::StorageGet(e)))?;
1388
1389        let new_meta = branch::branch_metadata(
1390            &ws.signing_key,
1391            branch_id,
1392            branch_name,
1393            Some(head_blob.to_blob()),
1394            Some(handle),
1395        );
1396        let new_meta_handle = self
1397            .storage
1398            .put(new_meta)
1399            .map_err(|e| RollupError::Push(PushError::StoragePut(e)))?;
1400
1401        // CAS: swap `base_branch_meta` for the new meta. On conflict, the
1402        // head advanced between our pull and this CAS — the rollup we built
1403        // is stale against the new head, so report upstream.
1404        let update_result = self
1405            .storage
1406            .update(branch_id, Some(ws.base_branch_meta), Some(new_meta_handle))
1407            .map_err(|e| RollupError::Push(PushError::BranchUpdate(e)))?;
1408        match update_result {
1409            PushResult::Success() => Ok(handle),
1410            PushResult::Conflict(_) => Err(RollupError::HeadAdvanced),
1411        }
1412    }
1413}
1414
1415/// A handle to a commit blob in the repository.
1416pub type CommitHandle = Inline<Handle<SimpleArchive>>;
1417type MetadataHandle = Inline<Handle<SimpleArchive>>;
1418/// A set of commit handles, used by [`CommitSelector`] and [`Checkout`].
1419pub type CommitSet = PATCH<INLINE_LEN, IdentitySchema, ()>;
1420type BranchMetaHandle = Inline<Handle<SimpleArchive>>;
1421
1422/// The result of a [`Workspace::checkout`] operation: a [`TribleSet`] paired
1423/// with the set of commits that produced it. Pass the commit set as the start
1424/// of a range selector to obtain incremental deltas on the next checkout.
1425///
1426/// [`Checkout`] dereferences to [`TribleSet`], so it can be used directly with
1427/// `find!`, `pattern!`, and `pattern_changes!`.
1428///
1429/// # Example: incremental updates
1430///
1431/// ```rust,ignore
1432/// let mut changed = repo.pull(branch_id)?.checkout(..)?;
1433/// let mut full = changed.facts().clone();
1434///
1435/// loop {
1436///     // full already includes changed
1437///     for result in pattern_changes!(&full, &changed, [{ ... }]) {
1438///         // process new results
1439///     }
1440///
1441///     // Advance — exclude exactly the commits we already processed.
1442///     changed = repo.pull(branch_id)?.checkout(changed.commits()..)?;
1443///     full += &changed;
1444/// }
1445/// ```
1446#[derive(Debug, Clone)]
1447pub struct Checkout {
1448    facts: TribleSet,
1449    commits: CommitSet,
1450}
1451
1452impl PartialEq<TribleSet> for Checkout {
1453    fn eq(&self, other: &TribleSet) -> bool {
1454        self.facts == *other
1455    }
1456}
1457
1458impl PartialEq<Checkout> for TribleSet {
1459    fn eq(&self, other: &Checkout) -> bool {
1460        *self == other.facts
1461    }
1462}
1463
1464impl Checkout {
1465    /// The checked-out tribles.
1466    pub fn facts(&self) -> &TribleSet {
1467        &self.facts
1468    }
1469
1470    /// The set of commits that produced this checkout. Use as the start of a
1471    /// range selector (`checkout.commits()..`) to exclude these commits
1472    /// on the next checkout and obtain only new data.
1473    pub fn commits(&self) -> CommitSet {
1474        self.commits.clone()
1475    }
1476
1477    /// Consume the checkout and return the inner TribleSet.
1478    pub fn into_facts(self) -> TribleSet {
1479        self.facts
1480    }
1481}
1482
1483impl std::ops::Deref for Checkout {
1484    type Target = TribleSet;
1485    fn deref(&self) -> &TribleSet {
1486        &self.facts
1487    }
1488}
1489
1490impl std::ops::AddAssign<&Checkout> for Checkout {
1491    fn add_assign(&mut self, rhs: &Checkout) {
1492        self.facts += rhs.facts.clone();
1493        self.commits.union(rhs.commits.clone());
1494    }
1495}
1496
1497impl std::ops::Add for Checkout {
1498    type Output = Self;
1499    fn add(mut self, rhs: Self) -> Self {
1500        self.facts += rhs.facts;
1501        self.commits.union(rhs.commits);
1502        self
1503    }
1504}
1505
1506impl std::ops::Add<&Checkout> for Checkout {
1507    type Output = Self;
1508    fn add(mut self, rhs: &Checkout) -> Self {
1509        self += rhs;
1510        self
1511    }
1512}
1513
1514/// The Workspace represents the mutable working area or "staging" state.
1515/// It was formerly known as `Head`. It is sent to worker threads,
1516/// modified (via commits, merges, etc.), and then merged back into the Repository.
1517pub struct Workspace<Blobs: BlobStore> {
1518    /// Staged blobs — added to this workspace but not yet pushed to
1519    /// the underlying repo. Analogous to git's staging area (the
1520    /// index): blobs accumulate here via `put` and friends, then
1521    /// `repo.push(&mut ws)` ships everything as one batch to the
1522    /// durable backend.
1523    pub staged: MemoryBlobStore,
1524    /// The blob storage base for the workspace.
1525    base_blobs: Blobs::Reader,
1526    /// The branch id this workspace is tracking; None for a detached workspace.
1527    base_branch_id: Id,
1528    /// The meta-handle corresponding to the base branch state used for CAS.
1529    base_branch_meta: BranchMetaHandle,
1530    /// Handle to the current commit in the working branch. `None` for an empty branch.
1531    head: Option<CommitHandle>,
1532    /// The branch head snapshot when this workspace was created (pull time).
1533    ///
1534    /// This allows `try_push` to cheaply detect whether the commit head has
1535    /// advanced since the workspace was created without querying the remote
1536    /// branch store.
1537    base_head: Option<CommitHandle>,
1538    /// Signing key used for commit/branch signing.
1539    signing_key: SigningKey,
1540    /// Metadata handle for commits created in this workspace.
1541    commit_metadata: MetadataHandle,
1542}
1543
1544impl<Blobs> fmt::Debug for Workspace<Blobs>
1545where
1546    Blobs: BlobStore,
1547    Blobs::Reader: fmt::Debug,
1548{
1549    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1550        f.debug_struct("Workspace")
1551            .field("staged", &self.staged)
1552            .field("base_blobs", &self.base_blobs)
1553            .field("base_branch_id", &self.base_branch_id)
1554            .field("base_branch_meta", &self.base_branch_meta)
1555            .field("base_head", &self.base_head)
1556            .field("head", &self.head)
1557            .field("commit_metadata", &self.commit_metadata)
1558            .finish()
1559    }
1560}
1561
1562/// Helper trait for [`Workspace::checkout`] specifying commit handles or ranges.
1563pub trait CommitSelector<Blobs: BlobStore> {
1564    fn select(
1565        self,
1566        ws: &mut Workspace<Blobs>,
1567    ) -> Result<
1568        CommitSet,
1569        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1570    >;
1571}
1572
1573/// Selector that returns every commit reachable from a starting selector.
1574pub struct Ancestors<S>(pub S);
1575
1576/// Convenience function to create an [`Ancestors`] selector.
1577pub fn ancestors<S>(selector: S) -> Ancestors<S> {
1578    Ancestors(selector)
1579}
1580
1581/// Selector that walks every commit in the input set back N parent steps,
1582/// following all parent links (including merge parents). Returns the set
1583/// of all commits found at exactly depth N from the starting set.
1584///
1585/// This is a wavefront expansion: at each step, every commit in the current
1586/// frontier is replaced by all of its parents. After N steps the frontier
1587/// is the result.
1588pub struct NthAncestors<S>(pub S, pub usize);
1589
1590/// Walk `selector` back `n` parent steps through all parent links.
1591pub fn nth_ancestors<S>(selector: S, n: usize) -> NthAncestors<S> {
1592    NthAncestors(selector, n)
1593}
1594
1595/// Selector that returns the direct parents of commits from a starting selector.
1596pub struct Parents<S>(pub S);
1597
1598/// Convenience function to create a [`Parents`] selector.
1599pub fn parents<S>(selector: S) -> Parents<S> {
1600    Parents(selector)
1601}
1602
1603/// Selector that returns commits reachable from either of two selectors but
1604/// not both.
1605pub struct SymmetricDiff<A, B>(pub A, pub B);
1606
1607/// Convenience function to create a [`SymmetricDiff`] selector.
1608pub fn symmetric_diff<A, B>(a: A, b: B) -> SymmetricDiff<A, B> {
1609    SymmetricDiff(a, b)
1610}
1611
1612/// Selector that returns the union of commits returned by two selectors.
1613pub struct Union<A, B> {
1614    left: A,
1615    right: B,
1616}
1617
1618/// Convenience function to create a [`Union`] selector.
1619pub fn union<A, B>(left: A, right: B) -> Union<A, B> {
1620    Union { left, right }
1621}
1622
1623/// Selector that returns the intersection of commits returned by two selectors.
1624pub struct Intersect<A, B> {
1625    left: A,
1626    right: B,
1627}
1628
1629/// Convenience function to create an [`Intersect`] selector.
1630pub fn intersect<A, B>(left: A, right: B) -> Intersect<A, B> {
1631    Intersect { left, right }
1632}
1633
1634/// Selector that returns commits from the left selector that are not also
1635/// returned by the right selector.
1636pub struct Difference<A, B> {
1637    left: A,
1638    right: B,
1639}
1640
1641/// Convenience function to create a [`Difference`] selector.
1642pub fn difference<A, B>(left: A, right: B) -> Difference<A, B> {
1643    Difference { left, right }
1644}
1645
1646/// Selector that returns commits with timestamps in the given inclusive range.
1647pub struct TimeRange(pub Epoch, pub Epoch);
1648
1649/// Convenience function to create a [`TimeRange`] selector.
1650pub fn time_range(start: Epoch, end: Epoch) -> TimeRange {
1651    TimeRange(start, end)
1652}
1653
1654/// Selector that filters commits returned by another selector.
1655pub struct Filter<S, F> {
1656    selector: S,
1657    filter: F,
1658}
1659
1660/// Convenience function to create a [`Filter`] selector.
1661pub fn filter<S, F>(selector: S, filter: F) -> Filter<S, F> {
1662    Filter { selector, filter }
1663}
1664
1665impl<Blobs> CommitSelector<Blobs> for CommitHandle
1666where
1667    Blobs: BlobStore,
1668{
1669    fn select(
1670        self,
1671        _ws: &mut Workspace<Blobs>,
1672    ) -> Result<
1673        CommitSet,
1674        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1675    > {
1676        let mut patch = CommitSet::new();
1677        patch.insert(&Entry::new(&self.raw));
1678        Ok(patch)
1679    }
1680}
1681
1682impl<Blobs> CommitSelector<Blobs> for CommitSet
1683where
1684    Blobs: BlobStore,
1685{
1686    fn select(
1687        self,
1688        _ws: &mut Workspace<Blobs>,
1689    ) -> Result<
1690        CommitSet,
1691        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1692    > {
1693        Ok(self)
1694    }
1695}
1696
1697impl<Blobs> CommitSelector<Blobs> for Vec<CommitHandle>
1698where
1699    Blobs: BlobStore,
1700{
1701    fn select(
1702        self,
1703        _ws: &mut Workspace<Blobs>,
1704    ) -> Result<
1705        CommitSet,
1706        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1707    > {
1708        let mut patch = CommitSet::new();
1709        for handle in self {
1710            patch.insert(&Entry::new(&handle.raw));
1711        }
1712        Ok(patch)
1713    }
1714}
1715
1716impl<Blobs> CommitSelector<Blobs> for &[CommitHandle]
1717where
1718    Blobs: BlobStore,
1719{
1720    fn select(
1721        self,
1722        _ws: &mut Workspace<Blobs>,
1723    ) -> Result<
1724        CommitSet,
1725        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1726    > {
1727        let mut patch = CommitSet::new();
1728        for handle in self {
1729            patch.insert(&Entry::new(&handle.raw));
1730        }
1731        Ok(patch)
1732    }
1733}
1734
1735impl<Blobs> CommitSelector<Blobs> for Option<CommitHandle>
1736where
1737    Blobs: BlobStore,
1738{
1739    fn select(
1740        self,
1741        _ws: &mut Workspace<Blobs>,
1742    ) -> Result<
1743        CommitSet,
1744        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1745    > {
1746        let mut patch = CommitSet::new();
1747        if let Some(handle) = self {
1748            patch.insert(&Entry::new(&handle.raw));
1749        }
1750        Ok(patch)
1751    }
1752}
1753
1754impl<S, Blobs> CommitSelector<Blobs> for Ancestors<S>
1755where
1756    S: CommitSelector<Blobs>,
1757    Blobs: BlobStore,
1758{
1759    fn select(
1760        self,
1761        ws: &mut Workspace<Blobs>,
1762    ) -> Result<
1763        CommitSet,
1764        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1765    > {
1766        let seeds = self.0.select(ws)?;
1767        collect_reachable_from_patch(ws, seeds)
1768    }
1769}
1770
1771impl<Blobs, S> CommitSelector<Blobs> for NthAncestors<S>
1772where
1773    Blobs: BlobStore,
1774    S: CommitSelector<Blobs>,
1775{
1776    fn select(
1777        self,
1778        ws: &mut Workspace<Blobs>,
1779    ) -> Result<
1780        CommitSet,
1781        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1782    > {
1783        let mut frontier = self.0.select(ws)?;
1784        let mut remaining = self.1;
1785
1786        while remaining > 0 && !frontier.is_empty() {
1787            // Collect current frontier keys before mutating.
1788            let keys: Vec<[u8; INLINE_LEN]> = frontier.iter().copied().collect();
1789            let mut next_frontier = CommitSet::new();
1790            for raw in keys {
1791                let handle = CommitHandle::new(raw);
1792                let meta: TribleSet = ws.get(handle).map_err(WorkspaceCheckoutError::Storage)?;
1793                for (p,) in find!((p: Inline<_>), pattern!(&meta, [{ parent: ?p }])) {
1794                    next_frontier.insert(&Entry::new(&p.raw));
1795                }
1796            }
1797            frontier = next_frontier;
1798            remaining -= 1;
1799        }
1800
1801        Ok(frontier)
1802    }
1803}
1804
1805impl<S, Blobs> CommitSelector<Blobs> for Parents<S>
1806where
1807    S: CommitSelector<Blobs>,
1808    Blobs: BlobStore,
1809{
1810    fn select(
1811        self,
1812        ws: &mut Workspace<Blobs>,
1813    ) -> Result<
1814        CommitSet,
1815        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1816    > {
1817        let seeds = self.0.select(ws)?;
1818        let mut result = CommitSet::new();
1819        for raw in seeds.iter() {
1820            let handle = Inline::new(*raw);
1821            let meta: TribleSet = ws.get(handle).map_err(WorkspaceCheckoutError::Storage)?;
1822            for (p,) in find!((p: Inline<_>), pattern!(&meta, [{ parent: ?p }])) {
1823                result.insert(&Entry::new(&p.raw));
1824            }
1825        }
1826        Ok(result)
1827    }
1828}
1829
1830impl<A, B, Blobs> CommitSelector<Blobs> for SymmetricDiff<A, B>
1831where
1832    A: CommitSelector<Blobs>,
1833    B: CommitSelector<Blobs>,
1834    Blobs: BlobStore,
1835{
1836    fn select(
1837        self,
1838        ws: &mut Workspace<Blobs>,
1839    ) -> Result<
1840        CommitSet,
1841        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1842    > {
1843        let seeds_a = self.0.select(ws)?;
1844        let seeds_b = self.1.select(ws)?;
1845        let a = collect_reachable_from_patch(ws, seeds_a)?;
1846        let b = collect_reachable_from_patch(ws, seeds_b)?;
1847        let inter = a.intersect(&b);
1848        let mut union = a;
1849        union.union(b);
1850        Ok(union.difference(&inter))
1851    }
1852}
1853
1854impl<A, B, Blobs> CommitSelector<Blobs> for Union<A, B>
1855where
1856    A: CommitSelector<Blobs>,
1857    B: CommitSelector<Blobs>,
1858    Blobs: BlobStore,
1859{
1860    fn select(
1861        self,
1862        ws: &mut Workspace<Blobs>,
1863    ) -> Result<
1864        CommitSet,
1865        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1866    > {
1867        let mut left = self.left.select(ws)?;
1868        let right = self.right.select(ws)?;
1869        left.union(right);
1870        Ok(left)
1871    }
1872}
1873
1874impl<A, B, Blobs> CommitSelector<Blobs> for Intersect<A, B>
1875where
1876    A: CommitSelector<Blobs>,
1877    B: CommitSelector<Blobs>,
1878    Blobs: BlobStore,
1879{
1880    fn select(
1881        self,
1882        ws: &mut Workspace<Blobs>,
1883    ) -> Result<
1884        CommitSet,
1885        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1886    > {
1887        let left = self.left.select(ws)?;
1888        let right = self.right.select(ws)?;
1889        Ok(left.intersect(&right))
1890    }
1891}
1892
1893impl<A, B, Blobs> CommitSelector<Blobs> for Difference<A, B>
1894where
1895    A: CommitSelector<Blobs>,
1896    B: CommitSelector<Blobs>,
1897    Blobs: BlobStore,
1898{
1899    fn select(
1900        self,
1901        ws: &mut Workspace<Blobs>,
1902    ) -> Result<
1903        CommitSet,
1904        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1905    > {
1906        let left = self.left.select(ws)?;
1907        let right = self.right.select(ws)?;
1908        Ok(left.difference(&right))
1909    }
1910}
1911
1912impl<S, F, Blobs> CommitSelector<Blobs> for Filter<S, F>
1913where
1914    Blobs: BlobStore,
1915    S: CommitSelector<Blobs>,
1916    F: for<'x, 'y> Fn(&'x TribleSet, &'y TribleSet) -> bool,
1917{
1918    fn select(
1919        self,
1920        ws: &mut Workspace<Blobs>,
1921    ) -> Result<
1922        CommitSet,
1923        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1924    > {
1925        let patch = self.selector.select(ws)?;
1926        let mut result = CommitSet::new();
1927        let filter = self.filter;
1928        for raw in patch.iter() {
1929            let handle = Inline::new(*raw);
1930            let meta: TribleSet = ws.get(handle).map_err(WorkspaceCheckoutError::Storage)?;
1931
1932            let Ok((content_handle,)) = find!(
1933                (c: Inline<_>),
1934                pattern!(&meta, [{ content: ?c }])
1935            )
1936            .exactly_one() else {
1937                return Err(WorkspaceCheckoutError::BadCommitMetadata());
1938            };
1939
1940            let payload: TribleSet = ws
1941                .get(content_handle)
1942                .map_err(WorkspaceCheckoutError::Storage)?;
1943
1944            if filter(&meta, &payload) {
1945                result.insert(&Entry::new(raw));
1946            }
1947        }
1948        Ok(result)
1949    }
1950}
1951
1952/// Selector that yields commits touching a specific entity.
1953pub struct HistoryOf(pub Id);
1954
1955/// Convenience function to create a [`HistoryOf`] selector.
1956pub fn history_of(entity: Id) -> HistoryOf {
1957    HistoryOf(entity)
1958}
1959
1960impl<Blobs> CommitSelector<Blobs> for HistoryOf
1961where
1962    Blobs: BlobStore,
1963{
1964    fn select(
1965        self,
1966        ws: &mut Workspace<Blobs>,
1967    ) -> Result<
1968        CommitSet,
1969        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1970    > {
1971        let Some(head_) = ws.head else {
1972            return Ok(CommitSet::new());
1973        };
1974        let entity = self.0;
1975        filter(
1976            ancestors(head_),
1977            move |_: &TribleSet, payload: &TribleSet| payload.iter().any(|t| t.e() == &entity),
1978        )
1979        .select(ws)
1980    }
1981}
1982
1983// Generic range selectors: allow any selector type to be used as a range
1984// endpoint. We still walk the history reachable from the end selector but now
1985// stop descending a branch as soon as we encounter a commit produced by the
1986// start selector. This keeps the mechanics explicit—`start..end` literally
1987// walks from `end` until it hits `start`—while continuing to support selectors
1988// such as `Ancestors(...)` at either boundary.
1989
1990fn collect_reachable_from_patch<Blobs: BlobStore>(
1991    ws: &mut Workspace<Blobs>,
1992    patch: CommitSet,
1993) -> Result<
1994    CommitSet,
1995    WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
1996> {
1997    let mut result = CommitSet::new();
1998    for raw in patch.iter() {
1999        let handle = Inline::new(*raw);
2000        let reach = collect_reachable(ws, handle)?;
2001        result.union(reach);
2002    }
2003    Ok(result)
2004}
2005
2006fn collect_reachable_from_patch_until<Blobs: BlobStore>(
2007    ws: &mut Workspace<Blobs>,
2008    seeds: CommitSet,
2009    stop: &CommitSet,
2010) -> Result<
2011    CommitSet,
2012    WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2013> {
2014    let mut visited = HashSet::new();
2015    let mut stack: Vec<CommitHandle> = seeds.iter().map(|raw| Inline::new(*raw)).collect();
2016    let mut result = CommitSet::new();
2017
2018    while let Some(commit) = stack.pop() {
2019        if !visited.insert(commit) {
2020            continue;
2021        }
2022
2023        if stop.get(&commit.raw).is_some() {
2024            continue;
2025        }
2026
2027        result.insert(&Entry::new(&commit.raw));
2028
2029        let meta: TribleSet = ws
2030            .staged
2031            .reader()
2032            .unwrap()
2033            .get(commit)
2034            .or_else(|_| ws.base_blobs.get(commit))
2035            .map_err(WorkspaceCheckoutError::Storage)?;
2036
2037        for (p,) in find!((p: Inline<_>,), pattern!(&meta, [{ parent: ?p }])) {
2038            stack.push(p);
2039        }
2040    }
2041
2042    Ok(result)
2043}
2044
2045impl<T, Blobs> CommitSelector<Blobs> for std::ops::Range<T>
2046where
2047    T: CommitSelector<Blobs>,
2048    Blobs: BlobStore,
2049{
2050    fn select(
2051        self,
2052        ws: &mut Workspace<Blobs>,
2053    ) -> Result<
2054        CommitSet,
2055        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2056    > {
2057        let end_patch = self.end.select(ws)?;
2058        let start_patch = self.start.select(ws)?;
2059
2060        collect_reachable_from_patch_until(ws, end_patch, &start_patch)
2061    }
2062}
2063
2064impl<T, Blobs> CommitSelector<Blobs> for std::ops::RangeFrom<T>
2065where
2066    T: CommitSelector<Blobs>,
2067    Blobs: BlobStore,
2068{
2069    fn select(
2070        self,
2071        ws: &mut Workspace<Blobs>,
2072    ) -> Result<
2073        CommitSet,
2074        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2075    > {
2076        let Some(head_) = ws.head else {
2077            return Ok(CommitSet::new());
2078        };
2079        let exclude_patch = self.start.select(ws)?;
2080
2081        let mut head_patch = CommitSet::new();
2082        head_patch.insert(&Entry::new(&head_.raw));
2083
2084        collect_reachable_from_patch_until(ws, head_patch, &exclude_patch)
2085    }
2086}
2087
2088impl<T, Blobs> CommitSelector<Blobs> for std::ops::RangeTo<T>
2089where
2090    T: CommitSelector<Blobs>,
2091    Blobs: BlobStore,
2092{
2093    fn select(
2094        self,
2095        ws: &mut Workspace<Blobs>,
2096    ) -> Result<
2097        CommitSet,
2098        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2099    > {
2100        let end_patch = self.end.select(ws)?;
2101        collect_reachable_from_patch(ws, end_patch)
2102    }
2103}
2104
2105impl<Blobs> CommitSelector<Blobs> for std::ops::RangeFull
2106where
2107    Blobs: BlobStore,
2108{
2109    fn select(
2110        self,
2111        ws: &mut Workspace<Blobs>,
2112    ) -> Result<
2113        CommitSet,
2114        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2115    > {
2116        let Some(head_) = ws.head else {
2117            return Ok(CommitSet::new());
2118        };
2119        collect_reachable(ws, head_)
2120    }
2121}
2122
2123impl<Blobs> CommitSelector<Blobs> for TimeRange
2124where
2125    Blobs: BlobStore,
2126{
2127    fn select(
2128        self,
2129        ws: &mut Workspace<Blobs>,
2130    ) -> Result<
2131        CommitSet,
2132        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2133    > {
2134        let Some(head_) = ws.head else {
2135            return Ok(CommitSet::new());
2136        };
2137        let start = self.0;
2138        let end = self.1;
2139        filter(
2140            ancestors(head_),
2141            move |meta: &TribleSet, _payload: &TribleSet| {
2142                if let Ok(Some(((ts_start, ts_end),))) =
2143                    find!((t: (Epoch, Epoch)), pattern!(meta, [{ crate::metadata::created_at: ?t }])).at_most_one()
2144                {
2145                    ts_start <= end && ts_end >= start
2146                } else {
2147                    false
2148                }
2149            },
2150        )
2151        .select(ws)
2152    }
2153}
2154
2155/// Minimum number of commits at which `checkout_commits*` switches
2156/// from the serial loop to a `rayon::par_iter().try_reduce()` over
2157/// the commits. Each commit involves one (or two) blob fetches plus
2158/// an unarchive — independent work per commit — so the crossover is
2159/// small. Below this the rayon overhead dominates.
2160#[cfg(feature = "parallel")]
2161const PARALLEL_CHECKOUT_THRESHOLD: usize = 8;
2162
2163impl<Blobs: BlobStore> Workspace<Blobs> {
2164    /// Returns the branch id associated with this workspace.
2165    pub fn branch_id(&self) -> Id {
2166        self.base_branch_id
2167    }
2168
2169    /// Returns the current commit handle if one exists.
2170    pub fn head(&self) -> Option<CommitHandle> {
2171        self.head
2172    }
2173
2174    /// Returns the workspace metadata handle.
2175    pub fn metadata(&self) -> MetadataHandle {
2176        self.commit_metadata
2177    }
2178
2179    /// Reads the rollup handle, if any, from the workspace's base branch
2180    /// metadata. Returns `None` if the branch has no rollup yet or if the
2181    /// metadata is missing the attribute. Readers can use this to fetch
2182    /// the archive blob directly and skip `checkout(..)` for warm queries:
2183    ///
2184    /// ```rust,ignore
2185    /// let mut ws = repo.pull(branch)?;
2186    /// match ws.rollup()? {
2187    ///     Some(h) => {
2188    ///         let archive: SuccinctArchive<_> = ws.get(h)?;
2189    ///         // query archive
2190    ///     }
2191    ///     None => {
2192    ///         let space = ws.checkout(..)?;
2193    ///         // query space (commit-chain materialisation)
2194    ///     }
2195    /// }
2196    /// ```
2197    ///
2198    /// Writers don't go through this — attach a rollup via
2199    /// [`Repository::compute_rollup`] instead.
2200    pub fn rollup(
2201        &mut self,
2202    ) -> Result<
2203        Option<Inline<Handle<SuccinctArchiveBlob>>>,
2204        <Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>,
2205    > {
2206        let base_meta: TribleSet = self.base_blobs.get(self.base_branch_meta)?;
2207        Ok(
2208            find!(
2209                (r: Inline<Handle<SuccinctArchiveBlob>>),
2210                pattern!(&base_meta, [{ rollup: ?r }])
2211            )
2212            .next()
2213            .map(|(r,)| r),
2214        )
2215    }
2216
2217    /// Adds a blob to the workspace's local blob store.
2218    /// Mirrors [`BlobStorePut::put`](crate::repo::BlobStorePut) for ease of use.
2219    pub fn put<S, T>(&mut self, item: T) -> Inline<Handle<S>>
2220    where
2221        S: BlobEncoding + 'static,
2222        T: IntoBlob<S>,
2223        Handle<S>: InlineEncoding,
2224    {
2225        self.staged.put(item).expect("infallible blob put")
2226    }
2227
2228    /// Retrieves a blob from the workspace.
2229    ///
2230    /// The method first checks the workspace's local blob store and falls back
2231    /// to the base blob store if the blob is not found locally.
2232    pub fn get<T, S>(
2233        &mut self,
2234        handle: Inline<Handle<S>>,
2235    ) -> Result<T, <Blobs::Reader as BlobStoreGet>::GetError<<T as TryFromBlob<S>>::Error>>
2236    where
2237        S: BlobEncoding + 'static,
2238        T: TryFromBlob<S>,
2239        Handle<S>: InlineEncoding,
2240    {
2241        self.staged
2242            .reader()
2243            .unwrap()
2244            .get(handle)
2245            .or_else(|_| self.base_blobs.get(handle))
2246    }
2247
2248    /// Performs a commit in the workspace.
2249    ///
2250    /// Accepts anything that converts into a [`Fragment`] — either a
2251    /// raw [`TribleSet`] (auto-promoted to a Fragment with empty blob
2252    /// store), or a Fragment built up via `entity!{}` /
2253    /// `MetaDescribe::describe()` whose embedded blobs get absorbed
2254    /// into `self.staged` alongside the commit-content blob.
2255    /// This method creates a new commit blob (stored in the local
2256    /// blobset) and updates the current commit handle.
2257    pub fn commit(&mut self, content_: impl Into<Fragment>, message_: &str) {
2258        self.commit_internal(content_.into(), Some(self.commit_metadata), Some(message_));
2259    }
2260
2261    /// Like [`commit`](Self::commit) but attaches one-off metadata
2262    /// instead of the repository default.
2263    ///
2264    /// Accepts anything that converts into a [`Fragment`]: the
2265    /// fragment's embedded blobs are absorbed into the staged blob
2266    /// store and its facts are archived as a `SimpleArchive` blob
2267    /// whose handle lands in the commit's metadata slot. Archiving is
2268    /// content-addressed, so committing the same metadata fragment
2269    /// repeatedly converges on one blob — no caller-side caching
2270    /// needed for correctness or storage. When the metadata is
2271    /// already archived (e.g. shared across many commits and you hold
2272    /// its handle), use
2273    /// [`commit_with_metadata_handle`](Self::commit_with_metadata_handle)
2274    /// to skip re-serialization.
2275    pub fn commit_with_metadata(
2276        &mut self,
2277        content_: impl Into<Fragment>,
2278        metadata_: impl Into<Fragment>,
2279        message_: &str,
2280    ) {
2281        let (meta_facts, meta_blobs) = metadata_.into().into_facts_and_blobs();
2282        self.staged.union(meta_blobs);
2283        let metadata_handle = self.put(meta_facts);
2284        self.commit_internal(content_.into(), Some(metadata_handle), Some(message_));
2285    }
2286
2287    /// Like [`commit`](Self::commit) but attaches an already-archived
2288    /// metadata handle instead of the repository default. The handle
2289    /// variant of [`commit_with_metadata`](Self::commit_with_metadata):
2290    /// no serialization happens, so this is the right form when the
2291    /// same metadata archive is shared across many commits.
2292    pub fn commit_with_metadata_handle(
2293        &mut self,
2294        content_: impl Into<Fragment>,
2295        metadata_: MetadataHandle,
2296        message_: &str,
2297    ) {
2298        self.commit_internal(content_.into(), Some(metadata_), Some(message_));
2299    }
2300
2301    fn commit_internal(
2302        &mut self,
2303        content_: Fragment,
2304        metadata_handle: Option<MetadataHandle>,
2305        message_: Option<&str>,
2306    ) {
2307        let (content_facts, content_blobs) = content_.into_facts_and_blobs();
2308        // 0. Absorb any blobs the Fragment carried with it into the
2309        //    staging area before producing the commit blob, so handles
2310        //    inside `content_facts` resolve against `self.staged`.
2311        self.staged.union(content_blobs);
2312        // 1. Create a commit blob from the current head, content, metadata and the commit message.
2313        let content_blob: Blob<SimpleArchive> = content_facts.to_blob();
2314        // If a message is provided, store it as a LongString blob and pass the handle.
2315        let message_handle = message_.map(|m| self.put(m.to_string()));
2316        let parents = self.head.iter().copied();
2317
2318        let commit_set = crate::repo::commit::commit_metadata(
2319            &self.signing_key,
2320            parents,
2321            message_handle,
2322            Some(content_blob.clone()),
2323            metadata_handle,
2324        );
2325        // 2. Store the content and commit blobs in `self.staged`.
2326        let _ = self
2327            .staged
2328            .put::<SimpleArchive, _>(content_blob)
2329            .expect("failed to put content blob");
2330        let commit_handle = self
2331            .staged
2332            .put(commit_set)
2333            .expect("failed to put commit blob");
2334        // 3. Update `self.head` to point to the new commit.
2335        self.head = Some(commit_handle);
2336    }
2337
2338    /// Merge another workspace into this one.
2339    ///
2340    /// Always copies the *staged* blobs from `other.staged` into
2341    /// `self.staged` (so standalone blobs that aren't referenced by any
2342    /// commit chain still come along — useful when the other workspace was
2343    /// being used to stage content).
2344    ///
2345    /// Then integrates `other.head` via [`merge_commit`](Self::merge_commit),
2346    /// which picks no-op / fast-forward / merge commit as appropriate.
2347    ///
2348    /// Returns the workspace's new head, or `None` if both workspaces were
2349    /// empty (nothing to merge into anything).
2350    ///
2351    /// Notes:
2352    /// - The merge does *not* automatically import the entire base history
2353    ///   reachable from `other`'s head. If the incoming parent commits
2354    ///   reference blobs that do not exist in this repository's storage,
2355    ///   reading those commits later will fail until the missing blobs are
2356    ///   explicitly imported (for example via `repo::transfer(reachable(...))`).
2357    /// - This design keeps merge permissive and leaves cross-repository blob
2358    ///   import as an explicit user action.
2359    pub fn merge(
2360        &mut self,
2361        other: &mut Workspace<Blobs>,
2362    ) -> Result<Option<CommitHandle>, MergeError> {
2363        // 1. Always transfer staged blobs from `other`. They may include
2364        //    standalone blobs (no commit referring to them yet) that the
2365        //    caller wanted to stash in the workspace independent of any
2366        //    branch state.
2367        let other_local = other.staged.reader().unwrap();
2368        for r in other_local.blobs() {
2369            let handle = r.expect("infallible blob enumeration");
2370            let blob: Blob<UnknownBlob> = other_local.get(handle).expect("infallible blob read");
2371            self.staged
2372                .put::<UnknownBlob, _>(blob)
2373                .expect("infallible blob put");
2374        }
2375
2376        // 2. Integrate `other`'s head via the smart merge_commit. If `other`
2377        //    has no head, there's nothing further to integrate — just return
2378        //    our current head (which may or may not exist).
2379        match other.head {
2380            Some(other_head) => Ok(Some(self.merge_commit(other_head)?)),
2381            None => Ok(self.head),
2382        }
2383    }
2384
2385    /// Integrate another commit into this workspace's history.
2386    ///
2387    /// Picks the cheapest correct strategy:
2388    ///
2389    /// - **No-op** if the workspace has no head and `other` *is* the head, or
2390    ///   if `other` is already in the current head's ancestry.
2391    /// - **Fast-forward** if the workspace has no head, or if the current head
2392    ///   is in `other`'s ancestry — `self.head` is set to `other` directly.
2393    /// - **Merge commit** otherwise — a new commit with `[current_head, other]`
2394    ///   as parents is created and `self.head` advances to it.
2395    ///
2396    /// Returns the workspace's new head in all cases.
2397    ///
2398    /// The ancestor checks are best-effort: if the relevant commit blobs are
2399    /// missing from the workspace's view, the function falls through to the
2400    /// always-correct merge-commit path. Callers that mirror remote chains
2401    /// should ensure reachable blobs were imported (e.g. via `reachable` +
2402    /// `transfer`) for the optimization to kick in.
2403    pub fn merge_commit(
2404        &mut self,
2405        other: Inline<Handle<SimpleArchive>>,
2406    ) -> Result<CommitHandle, MergeError> {
2407        // Trivial cases first.
2408        let local_head = match self.head {
2409            None => {
2410                // No local head — fast-forward to `other`.
2411                self.head = Some(other);
2412                return Ok(other);
2413            }
2414            Some(h) if h == other => {
2415                // Identical — no-op.
2416                return Ok(h);
2417            }
2418            Some(h) => h,
2419        };
2420
2421        // Walk both ancestry chains. If either walk fails because a commit
2422        // blob is missing locally, refuse to merge — falling through to a
2423        // divergent-merge here would write a new commit referencing an
2424        // unknown parent, which `pile diagnose check` would later report as
2425        // a chain break and which `fetch_reachable`'s Phase-1
2426        // `have_local` short-circuit would never re-fetch. Better to fail
2427        // loudly so the caller can re-sync the missing closure and retry.
2428        let remote_in_local = ancestors(local_head)
2429            .select(self)
2430            .map_err(|e| MergeError::AncestryWalkFailed(format!("walking local ancestry: {e:?}")))?
2431            .get(&other.raw)
2432            .is_some();
2433        if remote_in_local {
2434            // `other` is already in our history → no-op.
2435            return Ok(local_head);
2436        }
2437
2438        let local_in_remote = ancestors(other)
2439            .select(self)
2440            .map_err(|e| MergeError::AncestryWalkFailed(format!("walking remote ancestry: {e:?}")))?
2441            .get(&local_head.raw)
2442            .is_some();
2443        if local_in_remote {
2444            // We're behind `other` → fast-forward.
2445            self.head = Some(other);
2446            return Ok(other);
2447        }
2448
2449        // Truly divergent — create a merge commit.
2450        let parents = self.head.iter().copied().chain(Some(other));
2451        let merge_commit = commit_metadata(&self.signing_key, parents, None, None, None);
2452        let commit_handle = self
2453            .staged
2454            .put(merge_commit)
2455            .expect("failed to put merge commit blob");
2456        self.head = Some(commit_handle);
2457        Ok(commit_handle)
2458    }
2459
2460    /// Move the workspace's head to `commit` without creating a new commit.
2461    ///
2462    /// This is the "fast-forward" case: when the new commit is a descendant
2463    /// of (or equal to) the current head, you can advance directly without
2464    /// a merge commit. The caller is responsible for verifying the
2465    /// descendancy relationship — typically via [`ancestors`] over `commit`.
2466    ///
2467    /// Use this in pull/sync flows to avoid spurious merge commits when one
2468    /// peer is simply behind the other.
2469    pub fn set_head(&mut self, commit: CommitHandle) {
2470        self.head = Some(commit);
2471    }
2472
2473    /// Returns the combined [`TribleSet`] for the specified commits.
2474    ///
2475    /// Each commit handle must reference a commit blob stored either in the
2476    /// workspace's local blob store or the repository's base store. The
2477    /// associated content blobs are loaded and unioned together. An error is
2478    /// returned if any commit or content blob is missing or malformed.
2479    fn checkout_commits<I>(
2480        &mut self,
2481        commits: I,
2482    ) -> Result<
2483        TribleSet,
2484        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2485    >
2486    where
2487        I: IntoIterator<Item = CommitHandle>,
2488    {
2489        let local = self.staged.reader().unwrap();
2490        let commits: Vec<CommitHandle> = commits.into_iter().collect();
2491
2492        #[cfg(feature = "parallel")]
2493        {
2494            if commits.len() >= PARALLEL_CHECKOUT_THRESHOLD {
2495                use rayon::prelude::*;
2496                let base = self.base_blobs.clone();
2497                return commits
2498                    .into_par_iter()
2499                    .map_with(
2500                        (local, base),
2501                        |(local, base), commit| -> Result<TribleSet, _> {
2502                            let meta: TribleSet = local
2503                                .get(commit)
2504                                .or_else(|_| base.get(commit))
2505                                .map_err(WorkspaceCheckoutError::Storage)?;
2506                            let content_opt = match find!(
2507                                (c: Inline<_>),
2508                                pattern!(&meta, [{ content: ?c }])
2509                            )
2510                            .at_most_one()
2511                            {
2512                                Ok(Some((c,))) => Some(c),
2513                                Ok(None) => None,
2514                                Err(_) => {
2515                                    return Err(WorkspaceCheckoutError::BadCommitMetadata())
2516                                }
2517                            };
2518                            if let Some(c) = content_opt {
2519                                let set: TribleSet = local
2520                                    .get(c)
2521                                    .or_else(|_| base.get(c))
2522                                    .map_err(WorkspaceCheckoutError::Storage)?;
2523                                Ok(set)
2524                            } else {
2525                                Ok(TribleSet::new())
2526                            }
2527                        },
2528                    )
2529                    .try_reduce(TribleSet::new, |a, b| Ok(a + b));
2530            }
2531        }
2532
2533        let mut result = TribleSet::new();
2534        for commit in commits {
2535            let meta: TribleSet = local
2536                .get(commit)
2537                .or_else(|_| self.base_blobs.get(commit))
2538                .map_err(WorkspaceCheckoutError::Storage)?;
2539
2540            // Some commits (for example merge commits) intentionally do not
2541            // carry a content blob. Treat those as no-ops during checkout so
2542            // callers can request ancestor ranges without failing when a
2543            // merge commit is encountered.
2544            let content_opt =
2545                match find!((c: Inline<_>), pattern!(&meta, [{ content: ?c }])).at_most_one() {
2546                    Ok(Some((c,))) => Some(c),
2547                    Ok(None) => None,
2548                    Err(_) => return Err(WorkspaceCheckoutError::BadCommitMetadata()),
2549                };
2550
2551            if let Some(c) = content_opt {
2552                let set: TribleSet = local
2553                    .get(c)
2554                    .or_else(|_| self.base_blobs.get(c))
2555                    .map_err(WorkspaceCheckoutError::Storage)?;
2556                result += set;
2557            } else {
2558                // No content for this commit (e.g. merge-only commit); skip it.
2559                continue;
2560            }
2561        }
2562        Ok(result)
2563    }
2564
2565    fn checkout_commits_metadata<I>(
2566        &mut self,
2567        commits: I,
2568    ) -> Result<
2569        TribleSet,
2570        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2571    >
2572    where
2573        I: IntoIterator<Item = CommitHandle>,
2574    {
2575        let local = self.staged.reader().unwrap();
2576        let commits: Vec<CommitHandle> = commits.into_iter().collect();
2577
2578        #[cfg(feature = "parallel")]
2579        {
2580            if commits.len() >= PARALLEL_CHECKOUT_THRESHOLD {
2581                use rayon::prelude::*;
2582                let base = self.base_blobs.clone();
2583                return commits
2584                    .into_par_iter()
2585                    .map_with(
2586                        (local, base),
2587                        |(local, base), commit| -> Result<TribleSet, _> {
2588                            let meta: TribleSet = local
2589                                .get(commit)
2590                                .or_else(|_| base.get(commit))
2591                                .map_err(WorkspaceCheckoutError::Storage)?;
2592                            let metadata_opt = match find!(
2593                                (c: Inline<_>),
2594                                pattern!(&meta, [{ metadata: ?c }])
2595                            )
2596                            .at_most_one()
2597                            {
2598                                Ok(Some((c,))) => Some(c),
2599                                Ok(None) => None,
2600                                Err(_) => {
2601                                    return Err(WorkspaceCheckoutError::BadCommitMetadata())
2602                                }
2603                            };
2604                            if let Some(c) = metadata_opt {
2605                                let set: TribleSet = local
2606                                    .get(c)
2607                                    .or_else(|_| base.get(c))
2608                                    .map_err(WorkspaceCheckoutError::Storage)?;
2609                                Ok(set)
2610                            } else {
2611                                Ok(TribleSet::new())
2612                            }
2613                        },
2614                    )
2615                    .try_reduce(TribleSet::new, |a, b| Ok(a + b));
2616            }
2617        }
2618
2619        let mut result = TribleSet::new();
2620        for commit in commits {
2621            let meta: TribleSet = local
2622                .get(commit)
2623                .or_else(|_| self.base_blobs.get(commit))
2624                .map_err(WorkspaceCheckoutError::Storage)?;
2625
2626            let metadata_opt =
2627                match find!((c: Inline<_>), pattern!(&meta, [{ metadata: ?c }])).at_most_one() {
2628                    Ok(Some((c,))) => Some(c),
2629                    Ok(None) => None,
2630                    Err(_) => return Err(WorkspaceCheckoutError::BadCommitMetadata()),
2631                };
2632
2633            if let Some(c) = metadata_opt {
2634                let set: TribleSet = local
2635                    .get(c)
2636                    .or_else(|_| self.base_blobs.get(c))
2637                    .map_err(WorkspaceCheckoutError::Storage)?;
2638                result += set;
2639            }
2640        }
2641        Ok(result)
2642    }
2643
2644    fn checkout_commits_with_metadata<I>(
2645        &mut self,
2646        commits: I,
2647    ) -> Result<
2648        (TribleSet, TribleSet),
2649        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2650    >
2651    where
2652        I: IntoIterator<Item = CommitHandle>,
2653    {
2654        let local = self.staged.reader().unwrap();
2655        let commits: Vec<CommitHandle> = commits.into_iter().collect();
2656
2657        #[cfg(feature = "parallel")]
2658        {
2659            if commits.len() >= PARALLEL_CHECKOUT_THRESHOLD {
2660                use rayon::prelude::*;
2661                let base = self.base_blobs.clone();
2662                return commits
2663                    .into_par_iter()
2664                    .map_with(
2665                        (local, base),
2666                        |(local, base), commit| -> Result<(TribleSet, TribleSet), _> {
2667                            let meta: TribleSet = local
2668                                .get(commit)
2669                                .or_else(|_| base.get(commit))
2670                                .map_err(WorkspaceCheckoutError::Storage)?;
2671                            let content_opt = match find!(
2672                                (c: Inline<_>),
2673                                pattern!(&meta, [{ content: ?c }])
2674                            )
2675                            .at_most_one()
2676                            {
2677                                Ok(Some((c,))) => Some(c),
2678                                Ok(None) => None,
2679                                Err(_) => {
2680                                    return Err(WorkspaceCheckoutError::BadCommitMetadata())
2681                                }
2682                            };
2683                            let data_set = if let Some(c) = content_opt {
2684                                local
2685                                    .get(c)
2686                                    .or_else(|_| base.get(c))
2687                                    .map_err(WorkspaceCheckoutError::Storage)?
2688                            } else {
2689                                TribleSet::new()
2690                            };
2691                            let metadata_opt = match find!(
2692                                (c: Inline<_>),
2693                                pattern!(&meta, [{ metadata: ?c }])
2694                            )
2695                            .at_most_one()
2696                            {
2697                                Ok(Some((c,))) => Some(c),
2698                                Ok(None) => None,
2699                                Err(_) => {
2700                                    return Err(WorkspaceCheckoutError::BadCommitMetadata())
2701                                }
2702                            };
2703                            let metadata_set = if let Some(c) = metadata_opt {
2704                                local
2705                                    .get(c)
2706                                    .or_else(|_| base.get(c))
2707                                    .map_err(WorkspaceCheckoutError::Storage)?
2708                            } else {
2709                                TribleSet::new()
2710                            };
2711                            Ok((data_set, metadata_set))
2712                        },
2713                    )
2714                    .try_reduce(
2715                        || (TribleSet::new(), TribleSet::new()),
2716                        |(a_data, a_meta), (b_data, b_meta)| {
2717                            Ok((a_data + b_data, a_meta + b_meta))
2718                        },
2719                    );
2720            }
2721        }
2722
2723        let mut data = TribleSet::new();
2724        let mut metadata_set = TribleSet::new();
2725        for commit in commits {
2726            let meta: TribleSet = local
2727                .get(commit)
2728                .or_else(|_| self.base_blobs.get(commit))
2729                .map_err(WorkspaceCheckoutError::Storage)?;
2730
2731            let content_opt =
2732                match find!((c: Inline<_>), pattern!(&meta, [{ content: ?c }])).at_most_one() {
2733                    Ok(Some((c,))) => Some(c),
2734                    Ok(None) => None,
2735                    Err(_) => return Err(WorkspaceCheckoutError::BadCommitMetadata()),
2736                };
2737
2738            if let Some(c) = content_opt {
2739                let set: TribleSet = local
2740                    .get(c)
2741                    .or_else(|_| self.base_blobs.get(c))
2742                    .map_err(WorkspaceCheckoutError::Storage)?;
2743                data += set;
2744            }
2745
2746            let metadata_opt =
2747                match find!((c: Inline<_>), pattern!(&meta, [{ metadata: ?c }])).at_most_one() {
2748                    Ok(Some((c,))) => Some(c),
2749                    Ok(None) => None,
2750                    Err(_) => return Err(WorkspaceCheckoutError::BadCommitMetadata()),
2751                };
2752
2753            if let Some(c) = metadata_opt {
2754                let set: TribleSet = local
2755                    .get(c)
2756                    .or_else(|_| self.base_blobs.get(c))
2757                    .map_err(WorkspaceCheckoutError::Storage)?;
2758                metadata_set += set;
2759            }
2760        }
2761        Ok((data, metadata_set))
2762    }
2763
2764    /// Returns the combined [`TribleSet`] for the specified commits or commit
2765    /// ranges. `spec` can be a single [`CommitHandle`], an iterator of handles
2766    /// or any of the standard range types over [`CommitHandle`].
2767    pub fn checkout<R>(
2768        &mut self,
2769        spec: R,
2770    ) -> Result<
2771        Checkout,
2772        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2773    >
2774    where
2775        R: CommitSelector<Blobs>,
2776    {
2777        let commits = spec.select(self)?;
2778        let facts = self.checkout_commits(commits.iter().map(|raw| Inline::new(*raw)))?;
2779        Ok(Checkout { facts, commits })
2780    }
2781
2782    /// Returns the combined metadata [`TribleSet`] for the specified commits.
2783    /// Commits without metadata handles contribute an empty set.
2784    pub fn checkout_metadata<R>(
2785        &mut self,
2786        spec: R,
2787    ) -> Result<
2788        TribleSet,
2789        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2790    >
2791    where
2792        R: CommitSelector<Blobs>,
2793    {
2794        let patch = spec.select(self)?;
2795        let commits = patch.iter().map(|raw| Inline::new(*raw));
2796        self.checkout_commits_metadata(commits)
2797    }
2798
2799    /// Returns the combined data and metadata [`TribleSet`] for the specified commits.
2800    /// Metadata is loaded from each commit's `metadata` handle, when present.
2801    pub fn checkout_with_metadata<R>(
2802        &mut self,
2803        spec: R,
2804    ) -> Result<
2805        (TribleSet, TribleSet),
2806        WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2807    >
2808    where
2809        R: CommitSelector<Blobs>,
2810    {
2811        let patch = spec.select(self)?;
2812        let commits = patch.iter().map(|raw| Inline::new(*raw));
2813        self.checkout_commits_with_metadata(commits)
2814    }
2815}
2816
2817#[derive(Debug)]
2818pub enum WorkspaceCheckoutError<GetErr: Error> {
2819    /// Error retrieving blobs from storage.
2820    Storage(GetErr),
2821    /// Commit metadata is malformed or ambiguous.
2822    BadCommitMetadata(),
2823}
2824
2825impl<E: Error + fmt::Debug> fmt::Display for WorkspaceCheckoutError<E> {
2826    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2827        match self {
2828            WorkspaceCheckoutError::Storage(e) => write!(f, "storage error: {e}"),
2829            WorkspaceCheckoutError::BadCommitMetadata() => {
2830                write!(f, "commit metadata malformed")
2831            }
2832        }
2833    }
2834}
2835
2836impl<E: Error + fmt::Debug> Error for WorkspaceCheckoutError<E> {}
2837
2838fn collect_reachable<Blobs: BlobStore>(
2839    ws: &mut Workspace<Blobs>,
2840    from: CommitHandle,
2841) -> Result<
2842    CommitSet,
2843    WorkspaceCheckoutError<<Blobs::Reader as BlobStoreGet>::GetError<UnarchiveError>>,
2844> {
2845    let mut visited = HashSet::new();
2846    let mut stack = vec![from];
2847    let mut result = CommitSet::new();
2848
2849    while let Some(commit) = stack.pop() {
2850        if !visited.insert(commit) {
2851            continue;
2852        }
2853        result.insert(&Entry::new(&commit.raw));
2854
2855        let meta: TribleSet = ws
2856            .staged
2857            .reader()
2858            .unwrap()
2859            .get(commit)
2860            .or_else(|_| ws.base_blobs.get(commit))
2861            .map_err(WorkspaceCheckoutError::Storage)?;
2862
2863        for (p,) in find!((p: Inline<_>,), pattern!(&meta, [{ parent: ?p }])) {
2864            stack.push(p);
2865        }
2866    }
2867
2868    Ok(result)
2869}