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}