glm-5.3-flash
8a10a678e0
The rider-1 real-client capture (n5 task, executed at the wave-5 gate) corrected the wire emission shape: upstream ls-refs.c sends the literal `unborn` token in the oid field; the zero-oid shape we served since wave 2 made a real clone fetch-want the null oid (`ERR object <zero-oid> does not exist`, captured live). The inbound RefLine parser accepted both shapes all along; the emission pins to the upstream form. Pre-publish draft-shape amendment (the OQ-03 freeze is at the core's publish): ADR-018 gains the §5 amendment note; transport.md §ls-refs records the validation + corrected shape; the pinned rows re-byte-pin the literal (v2_machines renamed, the stateless/parse-back rows updated); the RefLine/LsRefsArgs doc comments and the CHANGELOG wave-2 bullet corrected. Verification: cargo test (workspace 375 default / 378 all-features, -p alkgit 376, -p alkgit --features sha256 379, 0 fail), both clippies -D warnings clean, cargo fmt --check, cargo doc zero warnings, the fuzz corpus gate 13/13, cargo publish --dry-run green.
1830 lines
68 KiB
Rust
1830 lines
68 KiB
Rust
//! The wave-1 in-memory backend doubles — alkgit's contract-reference
|
||
//! test convention (docs/plans/implementation.md §Wave 1, the
|
||
//! test-convention bullet): the alkblobs mem-engine analog. Every
|
||
//! protocol wave's gate runs its machines against these doubles
|
||
//! (`resolve`/`put_repo`/`update_repo`/`remove_repo` on
|
||
//! [`MemRegistry`], advertisement data and CAS transactions on
|
||
//! [`MemRefs`], the ack/done-round seam on [`MemPackGen`]), so the
|
||
//! convention is established here even though the full harness
|
||
//! compiles with wave 2's first machines.
|
||
//!
|
||
//! Each test binary's `mod common;` compiles the whole shared module,
|
||
//! and no single binary exercises every double — dead_code is allowed
|
||
//! module-wide so one machine's harness never fails the next's.
|
||
|
||
#![allow(dead_code)]
|
||
//!
|
||
//! **Location decision (recorded, the OQ-03 freeze-inventory discipline):**
|
||
//! the doubles live *only* here — shared modules under `tests/` so the
|
||
//! integration tests of the wave-2/3 machines reach them with `mod
|
||
//! common;` — and never in the public API: no re-export, no feature
|
||
//! gate, nothing doubles-related reachable from the crate root. The
|
||
//! doubles' shapes are test surface, test-versioned freely; if a
|
||
//! `src/`-unit test ever needs them, the path is a `#[cfg(test)]`-only
|
||
//! module or a `doc(hidden)` one — never a public export.
|
||
//!
|
||
//! **The row/stamp discipline** (adapted from the sibling crates —
|
||
//! the alkstore ADR-017 §4.2 version-stamping convention as alkblobs'
|
||
//! tests/conformance suite carries it):
|
||
//!
|
||
//! - Assertions are **rows**: named, independently-runnable properties,
|
||
//! `snake_case`, the name stating the *property*, not the test or the
|
||
//! caller.
|
||
//! - Every row is **version-stamped** with a doc-comment list item
|
||
//! beginning **`Contract stamp:`** citing the spec § / ADR § that pins
|
||
//! the behavior tested — never a bare wave or date. Several stamps per
|
||
//! row are allowed; a later change that amends a row's property
|
||
//! appends to its stamp list; stamps accumulate, the ADRs record the
|
||
//! history.
|
||
//! - The stamp marker is greppable —
|
||
//! `grep -rn "Contract stamp:" tests/` is the inventory.
|
||
//!
|
||
//! Refs are rows with stamps, here too: [`MemRefs`]' store is a row
|
||
//! table keyed on the resolved id (backend.md §Pinned signatures —
|
||
//! impls key storage on `repo_id`; the record carries no path, ADR-019
|
||
//! §8), and the traits' contract rows carry the stamps.
|
||
//!
|
||
//! **CAS semantics the refs double encodes** (ADR-013 §7, §10; ADR-018
|
||
//! §6; the substrate row CAS per ADR-019 §2): one transaction per
|
||
//! `apply_updates` call. A per-ref `expected` comparison never errors —
|
||
//! a stale expected is a denied [`RefOutcome`] (a business result of
|
||
//! receive-pack), the reason string being the server-chosen, verbatim
|
||
//! `ng` text ([`STALE_REASON`] here). `atomic = true` prepares the whole
|
||
//! transaction and commits it only if every update's CAS succeeds: on
|
||
//! any denial nothing is applied, the denied refs report their own
|
||
//! reasons, and every other ref of the transaction reports
|
||
//! [`ATOMIC_FAILURE_REASON`] (the observed upstream shape,
|
||
//! ADR-013 §10). Updates inside one transaction evaluate sequentially
|
||
//! against the evolving row state, so intra-transaction compositions
|
||
//! (a create then an update of the same ref) compose like a real
|
||
//! transaction.
|
||
//!
|
||
//! Unknown-repo ≡ unauthorized (ADR-008): [`MemRegistry`] returns
|
||
//! [`RegistryError::NotFound`] for an unresolved id and nothing else —
|
||
//! the call sites that consume the double collapse `NotFound` with
|
||
//! `Forbidden` to the generic wire denial; the double itself never
|
||
//! carries the authorization decision.
|
||
//!
|
||
//! **The object-store skeleton's Find-shaped front** ([`MemObjects`]):
|
||
//! `contains` (the existence half of the predicate `common_haves`
|
||
//! recognizes — ADR-014 §5's exists + is-commit; the is-commit half
|
||
//! lives in [`MemPackGen`]) and `find` (the object lookup generation uses) fill
|
||
//! only what the seam needs at this wave. The gix-trait binding over them
|
||
//! is [`MemFind`] — the `gix_object::Find`/`Exists` + `gix_pack::Find`
|
||
//! surface ADR-004's generation pipeline needs; [`MemPackGen`]::
|
||
//! `generate` composes the real pipeline over it (the wave-2 task),
|
||
//! and [`MemPackIngest`]::`prepare` composes the real ingestion double
|
||
//! behavior (the wave-3 task).
|
||
//!
|
||
//! **The ingestion double's composition** (recorded — the wave-3 task's
|
||
//! shape decision): not the POC-O `Bundle::write_to_directory`-over-a-
|
||
//! working-area route (that needs a filesystem staging dir + a gix-odb
|
||
//! front, and is the wave-8 substrate impl's shape with the pin bridge),
|
||
//! but the **streaming input iterators landing objects directly** — the
|
||
//! same internals `Bundle::write_to_directory` composes:
|
||
//! `data::input::BytesToEntriesIter` wrapped by
|
||
//! `data::input::LookupRefDeltaObjectsIter` with [`MemFind`] as the
|
||
//! thin-pack base lookup (ADR-013 §5's server-odb resolution), every
|
||
//! pack object expanded and staged as a loose-form row keyed by its
|
||
//! re-derived oid under the decompressed-size cap
|
||
//! (`Limits::max_object_size`, review 004 R-2 — the expansion reads
|
||
//! incrementally, no allocation sized by a pack-declared size).
|
||
//! Delta application is a local applier (gix-pack's
|
||
//! `data::delta::apply` is crate-private), the zero-object pack's
|
||
//! trailer — never reached by the streaming machinery's verification —
|
||
//! is verified by the double itself, and nothing partial ever lands:
|
||
//! the connectivity walk runs over the staged rows plus the
|
||
//! pre-existing namespace, and the landing happens only once the walk
|
||
//! closes. The mem store has no GC sweep, so no pins: on the `unpack`
|
||
//! ng leg nothing lands at all — this double's strictest honest form
|
||
//! of the working-area discard (ADR-020 §2).
|
||
|
||
use std::collections::BTreeMap;
|
||
use std::collections::HashSet;
|
||
use std::io;
|
||
use std::io::{Read, Write};
|
||
use std::sync::atomic::AtomicBool;
|
||
use std::sync::Arc;
|
||
|
||
use async_trait::async_trait;
|
||
use gix_hash::ObjectId;
|
||
use gix_pack::data::entry::Header as PackHeader;
|
||
use gix_pack::data::input::BytesToEntriesIter;
|
||
use gix_pack::data::input::Mode as InputMode;
|
||
use gix_pack::data::input::{Entry as PackEntry, EntryDataMode, LookupRefDeltaObjectsIter};
|
||
use gix_pack::data::output::entry::iter_from_counts;
|
||
use gix_pack::data::output::entry::iter_from_counts::{Mode, Options, Outcome};
|
||
use gix_pack::data::output::{bytes::FromEntriesIter, count};
|
||
use gix_parallel::Finalize;
|
||
use gix_traverse::commit::{Parents, Simple};
|
||
use parking_lot::Mutex;
|
||
|
||
use alkgit::{
|
||
GitPackGen, GitPackIngest, GitRefs, GitRegistry, GitRegistryStore, Limits, PreparedPush,
|
||
PushOptions, RefLine, RefOutcome, RefUpdate, RegistryError, RepoRecord, StorageError,
|
||
};
|
||
|
||
/// The server-chosen verbatim `ng` reason of a per-ref CAS denial
|
||
/// (ADR-013 §7–8: CAS-stale failures carry a server-chosen reason,
|
||
/// upstream's `stale info`-style internal text).
|
||
pub const STALE_REASON: &str = "stale info";
|
||
|
||
/// The server-chosen verbatim `ng` reason a ref that would otherwise
|
||
/// have succeeded reports when an atomic transaction is aborted by any
|
||
/// denial (ADR-013 §10 — every *other* ref reports `atomic push
|
||
/// failure`; the denied refs keep their own reasons).
|
||
pub const ATOMIC_FAILURE_REASON: &str = "atomic push failure";
|
||
|
||
/// The mem registry double — [`GitRegistry`]/[`GitRegistryStore`] over
|
||
/// an in-memory id → record map (`BTreeMap` for deterministic
|
||
/// iteration, the record's byte-determinism rule carried over from
|
||
/// ADR-015 §8's shape), under a short-held [`parking_lot::Mutex`]
|
||
/// (convention 3; no poisoning recovery posture exists to apply).
|
||
///
|
||
/// `resolve` of an unknown id is [`RegistryError::NotFound`] — the
|
||
/// unknown-repo ≡ unauthorized posture's double form (ADR-008); the
|
||
/// double returns it and nothing else. `put_repo` rejects an id that
|
||
/// is already taken ([`RegistryError::AlreadyExists`]) and an empty
|
||
/// repo id ([`RegistryError::Invalid`]) — the record's key discipline.
|
||
#[derive(Clone, Default)]
|
||
pub struct MemRegistry {
|
||
rows: Arc<Mutex<BTreeMap<String, RepoRecord>>>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitRegistry for MemRegistry {
|
||
async fn resolve(&self, repo_id: &str) -> Result<RepoRecord, RegistryError> {
|
||
match self.rows.lock().get(repo_id) {
|
||
Some(record) => Ok(record.clone()),
|
||
None => Err(RegistryError::NotFound),
|
||
}
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitRegistryStore for MemRegistry {
|
||
async fn put_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
|
||
if record.repo_id.is_empty() {
|
||
return Err(RegistryError::Invalid);
|
||
}
|
||
let mut rows = self.rows.lock();
|
||
if rows.contains_key(&record.repo_id) {
|
||
return Err(RegistryError::AlreadyExists);
|
||
}
|
||
rows.insert(record.repo_id.clone(), record.clone());
|
||
Ok(())
|
||
}
|
||
|
||
async fn update_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
|
||
let mut rows = self.rows.lock();
|
||
match rows.get_mut(&record.repo_id) {
|
||
Some(row) => *row = record.clone(),
|
||
None => return Err(RegistryError::NotFound),
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
async fn remove_repo(&self, repo_id: &str) -> Result<(), RegistryError> {
|
||
match self.rows.lock().remove(repo_id) {
|
||
Some(_) => Ok(()),
|
||
None => Err(RegistryError::NotFound),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// The mem refs double — [`GitRefs`] over a row table keyed on the
|
||
/// resolved repo id: `repo_id → (ref name → oid)`, both `BTreeMap`s
|
||
/// (deterministic advertisement order), under a short-held
|
||
/// [`parking_lot::Mutex`]. The
|
||
/// [`list_refs`][GitRefs::list_refs] and CAS semantics are the module
|
||
/// doc's CAS section; see its stamps.
|
||
///
|
||
/// The HEAD symref rows ([`set_head_symref`][MemRefs::set_head_symref])
|
||
/// carry the unborn-HEAD fixture (review 001 N-5): a repo whose HEAD
|
||
/// points to an unborn branch lists a `HEAD` row with `oid: None` and
|
||
/// the symref target — the row the wire layer maps to the unborn
|
||
/// ls-refs line (ADRs 003/009; the machine shape lands with the wave-2
|
||
/// ls-refs work).
|
||
#[derive(Clone, Default)]
|
||
pub struct MemRefs {
|
||
rows: Arc<Mutex<BTreeMap<String, BTreeMap<String, ObjectId>>>>,
|
||
head_symrefs: Arc<Mutex<BTreeMap<String, Option<String>>>>,
|
||
}
|
||
|
||
impl MemRefs {
|
||
/// Sets (or, with `None`, clears) the repo's unborn-HEAD symref row
|
||
/// — the fixture the unborn ls-refs shape composes against (the
|
||
/// N-5 rider's `RefLine { oid: None, symref_target: Some(..) }`).
|
||
pub fn set_head_symref(&self, repo: &RepoRecord, target: Option<String>) {
|
||
match target {
|
||
Some(target) => {
|
||
self.head_symrefs
|
||
.lock()
|
||
.insert(repo.repo_id.clone(), Some(target));
|
||
}
|
||
None => {
|
||
self.head_symrefs.lock().remove(&repo.repo_id);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitRefs for MemRefs {
|
||
async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
|
||
let head = self
|
||
.head_symrefs
|
||
.lock()
|
||
.get(&repo.repo_id)
|
||
.cloned()
|
||
.flatten();
|
||
let rows = self.rows.lock();
|
||
let Some(tier) = rows.get(&repo.repo_id) else {
|
||
return Ok(match head {
|
||
Some(target) => vec![RefLine {
|
||
name: "HEAD".to_owned(),
|
||
oid: None,
|
||
peeled: None,
|
||
symref_target: Some(target),
|
||
}],
|
||
None => Vec::new(),
|
||
});
|
||
};
|
||
let mut listed = Vec::with_capacity(tier.len() + usize::from(head.is_some()));
|
||
if let Some(target) = head {
|
||
listed.push(RefLine {
|
||
name: "HEAD".to_owned(),
|
||
oid: None,
|
||
peeled: None,
|
||
symref_target: Some(target),
|
||
});
|
||
}
|
||
listed.extend(tier.iter().map(|(name, oid)| RefLine {
|
||
name: name.clone(),
|
||
oid: Some(*oid),
|
||
peeled: None,
|
||
symref_target: None,
|
||
}));
|
||
Ok(listed)
|
||
}
|
||
|
||
async fn apply_updates(
|
||
&self,
|
||
repo: &RepoRecord,
|
||
updates: Vec<RefUpdate>,
|
||
atomic: bool,
|
||
) -> Result<Vec<RefOutcome>, StorageError> {
|
||
let mut rows = self.rows.lock();
|
||
let tier = rows.entry(repo.repo_id.clone()).or_default();
|
||
let mut working = tier.clone();
|
||
|
||
let mut evaluated: Vec<(RefUpdate, Result<(), String>)> = Vec::with_capacity(updates.len());
|
||
for update in &updates {
|
||
let outcome = evaluate(&mut working, update, STALE_REASON);
|
||
evaluated.push((update.clone(), outcome));
|
||
}
|
||
|
||
let any_denial = evaluated.iter().any(|(_, result)| result.is_err());
|
||
if !(atomic && any_denial) {
|
||
*tier = working;
|
||
}
|
||
|
||
let outcomes = evaluated
|
||
.into_iter()
|
||
.map(|(update, result)| match result {
|
||
Ok(()) => {
|
||
if atomic && any_denial {
|
||
RefOutcome {
|
||
name: update.name,
|
||
result: Err(ATOMIC_FAILURE_REASON.to_string()),
|
||
}
|
||
} else {
|
||
RefOutcome {
|
||
name: update.name,
|
||
result: Ok(()),
|
||
}
|
||
}
|
||
}
|
||
Err(reason) => RefOutcome {
|
||
name: update.name,
|
||
result: Err(reason),
|
||
},
|
||
})
|
||
.collect();
|
||
Ok(outcomes)
|
||
}
|
||
}
|
||
|
||
fn evaluate(
|
||
working: &mut BTreeMap<String, ObjectId>,
|
||
update: &RefUpdate,
|
||
stale: &str,
|
||
) -> Result<(), String> {
|
||
let current = working.get(&update.name).copied();
|
||
let matches = match (update.expected, current) {
|
||
(None, None) => true,
|
||
(None, Some(_)) => false,
|
||
(Some(expected), Some(current)) => expected == current,
|
||
(Some(_), None) => false,
|
||
};
|
||
if !matches {
|
||
return Err(stale.to_string());
|
||
}
|
||
match update.new {
|
||
Some(new) => {
|
||
working.insert(update.name.clone(), new);
|
||
}
|
||
None => {
|
||
working.remove(&update.name);
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
type ObjectRows = BTreeMap<(String, ObjectId), Vec<u8>>;
|
||
|
||
/// The object-store double skeleton — an in-memory `repo_id → oid →
|
||
/// object-bytes` map (`BTreeMap`, short-held [`parking_lot::Mutex`])
|
||
/// with the Find-shaped front waves 2–3 compose their machines
|
||
/// against: [`contains`][MemObjects::contains] (the existence predicate
|
||
/// [`common_haves`][alkgit::GitPackGen::common_haves] recognizes —
|
||
/// ADR-014 §5) and [`find`][MemObjects::find] (the object lookup
|
||
/// generation uses). `put` is test support: the doubles' rows seed the
|
||
/// map before composing against it. The `gix_object::Find` trait
|
||
/// binding and the pack-level machinery are waves 2–3; the shape is
|
||
/// fixed here so that composition is mechanical.
|
||
#[derive(Clone, Default)]
|
||
pub struct MemObjects {
|
||
entries: Arc<Mutex<ObjectRows>>,
|
||
}
|
||
|
||
impl MemObjects {
|
||
/// The existence half of the seam's recognized predicate (ADR-014
|
||
/// §5's exists + is-commit — the is-commit half is
|
||
/// [`MemPackGen`]'s `common_haves`, decoding the object type from
|
||
/// the stored bytes).
|
||
pub fn contains(&self, repo: &RepoRecord, oid: &ObjectId) -> bool {
|
||
self.entries
|
||
.lock()
|
||
.contains_key(&(repo.repo_id.clone(), *oid))
|
||
}
|
||
|
||
/// The object lookup for generation: the stored bytes, or `None`
|
||
/// when the repo's namespace lacks the oid.
|
||
pub fn find(&self, repo: &RepoRecord, oid: &ObjectId) -> Option<Vec<u8>> {
|
||
self.entries
|
||
.lock()
|
||
.get(&(repo.repo_id.clone(), *oid))
|
||
.cloned()
|
||
}
|
||
|
||
/// Seeds one object into the repo's namespace (test support).
|
||
pub fn put(&self, repo: &RepoRecord, oid: ObjectId, bytes: Vec<u8>) {
|
||
self.entries
|
||
.lock()
|
||
.insert((repo.repo_id.clone(), oid), bytes);
|
||
}
|
||
}
|
||
|
||
/// The pack-level Find front over one repo's [`MemObjects`] namespace —
|
||
/// the gix-trait binding ADR-004's generation pipeline consumes. The
|
||
/// test-double form of the substrate's Find front (ADR-019 §6, the
|
||
/// `gix_object::Find`/`Exists` + `gix_pack::Find` shapes constructed once
|
||
/// per session): the stored rows are loose-object encodings, decoded back
|
||
/// through [`gix_object::loose_header`] on every lookup.
|
||
///
|
||
/// `location_by_oid`/`pack_offsets_and_oid`/`entry_by_location` answer
|
||
/// with "not in any pack": the doubles carry no packed objects, so every
|
||
/// entry is generated as a whole deflated base (ADR-004; delta synthesis
|
||
/// is the recorded optimization backlog).
|
||
///
|
||
/// A lookup miss is recorded — the last missed oid — so a failed
|
||
/// walk/count stage maps its not-found error back to the
|
||
/// [`StorageError::ObjectMissing`] the seam aborts with (ADR-004), the
|
||
/// recorded id naming the missing object rather than the want.
|
||
#[derive(Clone)]
|
||
pub struct MemFind {
|
||
objects: MemObjects,
|
||
repo: RepoRecord,
|
||
missed: Arc<Mutex<Option<ObjectId>>>,
|
||
}
|
||
|
||
impl MemFind {
|
||
/// The Find front over one repo's object namespace, for one
|
||
/// generation call.
|
||
pub fn new(objects: &MemObjects, repo: &RepoRecord) -> Self {
|
||
MemFind {
|
||
objects: objects.clone(),
|
||
repo: repo.clone(),
|
||
missed: Arc::default(),
|
||
}
|
||
}
|
||
|
||
/// The oid recorded by the latest lookup miss, if any.
|
||
fn missed_object(&self) -> Option<ObjectId> {
|
||
self.missed.lock().take()
|
||
}
|
||
}
|
||
|
||
impl gix_object::Exists for MemFind {
|
||
fn exists(&self, id: &gix_hash::oid) -> bool {
|
||
<Self as gix_pack::Find>::contains(self, id)
|
||
}
|
||
}
|
||
|
||
impl gix_object::Find for MemFind {
|
||
fn try_find<'a>(
|
||
&self,
|
||
id: &gix_hash::oid,
|
||
buffer: &'a mut Vec<u8>,
|
||
) -> gix_error::Result<Option<gix_object::Data<'a>>> {
|
||
let find = <Self as gix_pack::Find>::try_find_cached(
|
||
self,
|
||
id,
|
||
buffer,
|
||
&mut gix_pack::cache::Never,
|
||
)?;
|
||
Ok(find.map(|(data, _location)| data))
|
||
}
|
||
}
|
||
|
||
impl gix_pack::Find for MemFind {
|
||
fn contains(&self, id: &gix_hash::oid) -> bool {
|
||
self.objects
|
||
.contains(&self.repo, &(id as &gix_hash::oid).to_owned())
|
||
}
|
||
|
||
fn try_find_cached<'a>(
|
||
&self,
|
||
id: &gix_hash::oid,
|
||
buffer: &'a mut Vec<u8>,
|
||
_pack_cache: &mut dyn gix_pack::cache::DecodeEntry,
|
||
) -> gix_error::Result<
|
||
Option<(
|
||
gix_object::Data<'a>,
|
||
Option<gix_pack::data::entry::Location>,
|
||
)>,
|
||
> {
|
||
let Some(bytes) = self
|
||
.objects
|
||
.find(&self.repo, &(id as &gix_hash::oid).to_owned())
|
||
else {
|
||
self.missed.lock().replace(id.to_owned());
|
||
return Ok(None);
|
||
};
|
||
let (kind, _size, header_len) = gix_object::decode::loose_header(&bytes)?;
|
||
buffer.clear();
|
||
buffer.extend_from_slice(&bytes[header_len..]);
|
||
Ok(Some((
|
||
gix_object::Data {
|
||
kind,
|
||
object_hash: gix_hash::Kind::Sha1,
|
||
data: buffer,
|
||
},
|
||
None,
|
||
)))
|
||
}
|
||
|
||
fn location_by_oid(
|
||
&self,
|
||
_id: &gix_hash::oid,
|
||
_buf: &mut Vec<u8>,
|
||
) -> gix_error::Result<Option<gix_pack::data::entry::Location>> {
|
||
Ok(None)
|
||
}
|
||
|
||
fn pack_offsets_and_oid(
|
||
&self,
|
||
_pack_id: u32,
|
||
) -> gix_error::Result<Option<Vec<(gix_pack::data::Offset, ObjectId)>>> {
|
||
Ok(None)
|
||
}
|
||
|
||
fn entry_by_location(
|
||
&self,
|
||
_location: &gix_pack::data::entry::Location,
|
||
) -> Option<gix_pack::find::Entry> {
|
||
None
|
||
}
|
||
}
|
||
|
||
fn object_missing(find: &MemFind, fallback: ObjectId) -> StorageError {
|
||
StorageError::ObjectMissing {
|
||
oid: find.missed_object().unwrap_or(fallback),
|
||
}
|
||
}
|
||
|
||
/// Maps a pipeline error to the seam's error model: a not-found is the
|
||
/// missing-object abort (ADR-004), anything else is a backing-store
|
||
/// failure (ADR-018 §7). `fallback` names the want the stage was
|
||
/// working on, only reached when no concrete miss was recorded.
|
||
fn from_pipeline_error(
|
||
find: &MemFind,
|
||
error: gix_error::Error,
|
||
fallback: ObjectId,
|
||
) -> StorageError {
|
||
if error.is_not_found() {
|
||
object_missing(find, fallback)
|
||
} else {
|
||
StorageError::Io(error.to_string())
|
||
}
|
||
}
|
||
|
||
/// The peel result of one want or have: the tag objects walked through
|
||
/// (the chain's closure members) and the terminal object's id.
|
||
struct PeeledTip {
|
||
chain: Vec<ObjectId>,
|
||
terminal: ObjectId,
|
||
terminal_is_commit: bool,
|
||
}
|
||
|
||
/// Peels `tip` through its tag chain (ADR-004: tips must be peeled before
|
||
/// the commit-ancestry walk), returning the chain and the terminal kind.
|
||
///
|
||
/// Errors on the first missing object of the chain with
|
||
/// [`StorageError::ObjectMissing`] (ADR-004) — a tip that cannot be
|
||
/// resolved is the abort, never a partial pack.
|
||
fn peel_tip(find: &MemFind, tip: ObjectId) -> Result<PeeledTip, StorageError> {
|
||
let mut chain = Vec::new();
|
||
let mut current = tip;
|
||
loop {
|
||
let mut buf = Vec::new();
|
||
let data = match gix_object::FindExt::find(find, ¤t, &mut buf) {
|
||
Ok(data) => data,
|
||
Err(error) => return Err(from_pipeline_error(find, error, tip)),
|
||
};
|
||
let decoded = data.decode();
|
||
let next = match decoded {
|
||
Ok(gix_object::ObjectRef::Tag(tag)) => {
|
||
chain.push(current);
|
||
tag.target()
|
||
}
|
||
Ok(gix_object::ObjectRef::Commit(_)) => {
|
||
return Ok(PeeledTip {
|
||
chain,
|
||
terminal: current,
|
||
terminal_is_commit: true,
|
||
});
|
||
}
|
||
Ok(_) => {
|
||
return Ok(PeeledTip {
|
||
chain,
|
||
terminal: current,
|
||
terminal_is_commit: false,
|
||
});
|
||
}
|
||
Err(error) => return Err(StorageError::Io(error.to_string())),
|
||
};
|
||
current = next;
|
||
}
|
||
}
|
||
|
||
/// The generation pipeline over the mem Find — ADR-004's composition
|
||
/// (poc2-findings' validated shape): peel wants to commit tips, the
|
||
/// boundary-bounded commit-ancestry walk ([`Simple`], `Parents::All`),
|
||
/// the two `count::objects_unthreaded` stages over the gix Find, and the
|
||
/// entry/byte stages streaming the V2 sha1 pack into the sink.
|
||
///
|
||
/// The two count stages compose ADR-014 §2's closure subtraction: the
|
||
/// boundary-committed ancestry is excluded by the walk's predicate, and
|
||
/// non-commit want targets / peeled tag objects are counted against the
|
||
/// boundary's own peeled object set (a tag want's terminal commit being
|
||
/// a boundary have ships the tag alone). Counted objects are then
|
||
/// existence-checked before the byte pipeline opens, so a missing
|
||
/// closure member aborts with nothing ever written to the sink; the
|
||
/// entry stage's own `missing_objects` statistics remain the abort
|
||
/// posture ADR-004 pins.
|
||
fn generate_pack(
|
||
objects: &MemObjects,
|
||
repo: &RepoRecord,
|
||
wants: Vec<ObjectId>,
|
||
boundary_haves: Vec<ObjectId>,
|
||
sink: Box<dyn io::Write + Send>,
|
||
) -> Result<(), StorageError> {
|
||
let find = MemFind::new(objects, repo);
|
||
let no_want = ObjectId::null(gix_hash::Kind::Sha1);
|
||
let first_want = wants.first().copied().unwrap_or(no_want);
|
||
|
||
let mut boundary_commits = HashSet::new();
|
||
let mut boundary_objects = HashSet::new();
|
||
for have in boundary_haves {
|
||
let peeled = peel_tip(&find, have)?;
|
||
boundary_objects.extend(peeled.chain.iter().copied());
|
||
boundary_objects.insert(peeled.terminal);
|
||
if peeled.terminal_is_commit {
|
||
boundary_commits.insert(peeled.terminal);
|
||
}
|
||
}
|
||
|
||
let mut commit_tips = Vec::new();
|
||
let mut non_commit_tips = Vec::new();
|
||
let mut tag_chains = Vec::new();
|
||
for want in wants {
|
||
let peeled = peel_tip(&find, want)?;
|
||
for chain_oid in peeled.chain {
|
||
if !boundary_objects.contains(&chain_oid) {
|
||
tag_chains.push(chain_oid);
|
||
}
|
||
}
|
||
if peeled.terminal_is_commit {
|
||
commit_tips.push(peeled.terminal);
|
||
} else if !boundary_objects.contains(&peeled.terminal) {
|
||
non_commit_tips.push(peeled.terminal);
|
||
}
|
||
}
|
||
|
||
let mut walk = Simple::filtered(commit_tips, find.clone(), |tip: &gix_hash::oid| {
|
||
!boundary_commits.contains(tip)
|
||
})
|
||
.parents(Parents::All);
|
||
let mut survivors = Vec::new();
|
||
for info in walk.by_ref() {
|
||
let info = info.map_err(|error| from_pipeline_error(&find, error, first_want))?;
|
||
survivors.push(info.id);
|
||
}
|
||
|
||
let counter = gix_utils::progress::Discard;
|
||
let interrupt = AtomicBool::new(false);
|
||
let mut closure_inputs = survivors;
|
||
closure_inputs.extend(non_commit_tips);
|
||
let (mut counts, _count_outcome) = count::objects_unthreaded(
|
||
&find,
|
||
&mut closure_inputs
|
||
.into_iter()
|
||
.map(Ok::<ObjectId, gix_error::Error>),
|
||
&counter,
|
||
&interrupt,
|
||
count::objects::ObjectExpansion::TreeContents,
|
||
)
|
||
.map_err(|error| from_pipeline_error(&find, error, first_want))?;
|
||
let (tag_counts, _tag_outcome) = count::objects_unthreaded(
|
||
&find,
|
||
&mut tag_chains.into_iter().map(Ok::<ObjectId, gix_error::Error>),
|
||
&counter,
|
||
&interrupt,
|
||
count::objects::ObjectExpansion::AsIs,
|
||
)
|
||
.map_err(|error| from_pipeline_error(&find, error, first_want))?;
|
||
let mut counted: HashSet<ObjectId> = counts.iter().map(|count| count.id).collect();
|
||
for tag_count in tag_counts {
|
||
if counted.contains(&tag_count.id) {
|
||
continue;
|
||
}
|
||
counted.insert(tag_count.id);
|
||
counts.push(tag_count);
|
||
}
|
||
|
||
for count in &counts {
|
||
if !gix_pack::Find::contains(&find, &count.id) {
|
||
return Err(StorageError::ObjectMissing { oid: count.id });
|
||
}
|
||
}
|
||
|
||
let counts_len = counts.len();
|
||
let mut entries = gix_parallel::InOrderIter::from(
|
||
iter_from_counts(
|
||
counts,
|
||
find.clone(),
|
||
Box::new(gix_utils::progress::Discard),
|
||
Options {
|
||
thread_limit: Some(1),
|
||
mode: Mode::PackCopyAndBaseObjects,
|
||
allow_thin_pack: false,
|
||
chunk_size: 10,
|
||
version: gix_pack::data::Version::V2,
|
||
compression: gix_zlib::Compression::DEFAULT,
|
||
},
|
||
)
|
||
.map_err(|error| from_pipeline_error(&find, error, first_want))?,
|
||
);
|
||
let mut from_entries = FromEntriesIter::new(
|
||
entries.by_ref(),
|
||
sink,
|
||
counts_len as u32,
|
||
gix_pack::data::Version::V2,
|
||
gix_hash::Kind::Sha1,
|
||
);
|
||
for step in from_entries.by_ref() {
|
||
step.map_err(|error| from_pipeline_error(&find, error, first_want))?;
|
||
}
|
||
if from_entries.digest().is_none() {
|
||
return Err(StorageError::Io(
|
||
"the pack trailer was never written".to_string(),
|
||
));
|
||
}
|
||
let entry_statistics: Outcome = entries
|
||
.inner
|
||
.finalize()
|
||
.map_err(|error| from_pipeline_error(&find, error, first_want))?;
|
||
if entry_statistics.missing_objects > 0 {
|
||
let missing = find
|
||
.missed_object()
|
||
.unwrap_or_else(|| ObjectId::null(gix_hash::Kind::Sha1));
|
||
return Err(StorageError::ObjectMissing { oid: missing });
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// The pack-generation double — [`GitPackGen`] sharing the
|
||
/// [`MemObjects`] namespace: `common_haves` is wired (the seam's
|
||
/// recognized predicate — exists + is-commit, per ADR-014 §5; the
|
||
/// stored bytes' object type is decodable through
|
||
/// [`gix_object::decode::loose_header`], preserving the caller's order
|
||
/// verbatim) and `generate` composes ADR-004's real pipeline over
|
||
/// [`MemFind`] (sync test code — the blocking-thread posture is the
|
||
/// machine's, ADR-009).
|
||
#[derive(Clone, Default)]
|
||
pub struct MemPackGen {
|
||
pub objects: MemObjects,
|
||
}
|
||
|
||
impl MemPackGen {
|
||
/// The seam's recognized predicate (ADR-014 §5): exists in the
|
||
/// repo's namespace **and** decodes as a commit. Unrecognizable
|
||
/// stored bytes fail closed (never recognized).
|
||
fn recognized(&self, repo: &RepoRecord, oid: &ObjectId) -> bool {
|
||
let Some(bytes) = self.objects.find(repo, oid) else {
|
||
return false;
|
||
};
|
||
matches!(
|
||
gix_object::decode::loose_header(&bytes),
|
||
Ok((gix_object::Kind::Commit, _, _))
|
||
)
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitPackGen for MemPackGen {
|
||
async fn generate(
|
||
&self,
|
||
repo: &RepoRecord,
|
||
wants: Vec<ObjectId>,
|
||
boundary_haves: Vec<ObjectId>,
|
||
_limits: Limits,
|
||
sink: Box<dyn io::Write + Send>,
|
||
) -> Result<(), StorageError> {
|
||
generate_pack(&self.objects, repo, wants, boundary_haves, sink)
|
||
}
|
||
|
||
async fn common_haves(
|
||
&self,
|
||
repo: &RepoRecord,
|
||
haves: Vec<ObjectId>,
|
||
) -> Result<Vec<ObjectId>, StorageError> {
|
||
Ok(haves
|
||
.into_iter()
|
||
.filter(|oid| self.recognized(repo, oid))
|
||
.collect())
|
||
}
|
||
}
|
||
|
||
/// One object expanded out of an ingested pack, staged (not yet
|
||
/// landed): the oid re-derived from the expanded body, the object kind
|
||
/// (a delta shares its base's kind), and the raw body.
|
||
#[derive(Clone)]
|
||
struct Staged {
|
||
oid: ObjectId,
|
||
kind: gix_object::Kind,
|
||
body: Vec<u8>,
|
||
}
|
||
|
||
/// The impl-side pack-size budget (ADR-009's `max_pack_size` — the
|
||
/// wire-side relay count is the primary enforcement, the seam's
|
||
/// `limits` parameter is the impl's own defense): a counting reader
|
||
/// that errors once the stream passes the budget, failing the
|
||
/// ingestion into the `unpack` ng leg, never `Err`. EOF before the
|
||
/// budget is the ordinary stream end — under-length input is the
|
||
/// indexer's own parse error.
|
||
struct BudgetedReader<R> {
|
||
inner: R,
|
||
bytes_seen: u64,
|
||
limit: u64,
|
||
}
|
||
|
||
impl<R> BudgetedReader<R> {
|
||
fn new(inner: R, limit: u64) -> Self {
|
||
BudgetedReader {
|
||
inner,
|
||
bytes_seen: 0,
|
||
limit,
|
||
}
|
||
}
|
||
}
|
||
|
||
impl<R: Read> Read for BudgetedReader<R> {
|
||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||
if self.bytes_seen >= self.limit {
|
||
return Err(io::Error::other(format!(
|
||
"pack stream exceeded the maximum pack size budget of {} bytes",
|
||
self.limit
|
||
)));
|
||
}
|
||
let read = self.inner.read(buf)?;
|
||
self.bytes_seen += u64::try_from(read).unwrap_or(u64::MAX);
|
||
Ok(read)
|
||
}
|
||
}
|
||
|
||
/// One size field of a delta stream: the 7-bit little-endian groups
|
||
/// with the continuation bit, starting at `offset` — the consumed byte
|
||
/// count is returned alongside.
|
||
fn delta_size_at(delta: &[u8], offset: usize) -> Result<(u64, usize), String> {
|
||
let mut size = 0u64;
|
||
let mut shift = 0u32;
|
||
let mut cursor = offset;
|
||
loop {
|
||
let Some(&byte) = delta.get(cursor) else {
|
||
return Err("the delta header is truncated".to_string());
|
||
};
|
||
cursor += 1;
|
||
size |= u64::from(byte & 0x7f) << shift;
|
||
shift += 7;
|
||
if byte & 0x80 == 0 {
|
||
return Ok((size, cursor - offset));
|
||
}
|
||
if shift >= 64 {
|
||
return Err("the delta header size overflow".to_string());
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Applies a git delta stream to its base: the declared base size
|
||
/// must match, the declared result stays under `cap` (the ingestion
|
||
/// decompressed-size cap — the delta's materialized result is one of
|
||
/// the objects the pack produces),
|
||
/// and the instructions (copy from base, insert literal)
|
||
/// must produce exactly the declared result size. The result
|
||
/// accumulates incrementally — the declared size is checked against,
|
||
/// never pre-allocated (the size-bomb class): the loop itself aborts
|
||
/// once the produced bytes pass the declaration, so no delta stream
|
||
/// produces bytes beyond the cap unannounced.
|
||
fn apply_delta(base: &[u8], delta: &[u8], cap: u64) -> Result<Vec<u8>, String> {
|
||
let (base_size, base_header_len) = delta_size_at(delta, 0)?;
|
||
let (result_size, result_header_len) = delta_size_at(delta, base_header_len)?;
|
||
if base_size != u64::try_from(base.len()).unwrap_or(u64::MAX) {
|
||
return Err(format!(
|
||
"the delta declares a base size of {base_size} but the base object is {} bytes",
|
||
base.len()
|
||
));
|
||
}
|
||
if result_size > u64::from(u32::MAX) {
|
||
return Err("the delta declares an out-of-bounds result size".to_string());
|
||
}
|
||
if result_size > cap {
|
||
return Err(format!(
|
||
"the delta declares a materialized result of {result_size} bytes beyond \
|
||
the ingestion cap of {cap} bytes"
|
||
));
|
||
}
|
||
let mut out = Vec::new();
|
||
let mut cursor = base_header_len + result_header_len;
|
||
while let Some(&cmd) = delta.get(cursor) {
|
||
cursor += 1;
|
||
if cmd & 0x80 != 0 {
|
||
let mut ofs = 0u64;
|
||
let mut size = 0u64;
|
||
for (bit, shift) in [(0x01, 0), (0x02, 8), (0x04, 16), (0x08, 24)] {
|
||
if cmd & bit != 0 {
|
||
let Some(&byte) = delta.get(cursor) else {
|
||
return Err("the delta copy instruction is truncated".to_string());
|
||
};
|
||
cursor += 1;
|
||
ofs |= u64::from(byte) << shift;
|
||
}
|
||
}
|
||
for (bit, shift) in [(0x10, 0), (0x20, 8), (0x40, 16)] {
|
||
if cmd & bit != 0 {
|
||
let Some(&byte) = delta.get(cursor) else {
|
||
return Err("the delta copy instruction is truncated".to_string());
|
||
};
|
||
cursor += 1;
|
||
size |= u64::from(byte) << shift;
|
||
}
|
||
}
|
||
if size == 0 {
|
||
size = 0x10000;
|
||
}
|
||
let (ofs, size) = (
|
||
usize::try_from(ofs).map_err(|_| "the delta copy offset overflows".to_string())?,
|
||
usize::try_from(size).map_err(|_| "the delta copy size overflows".to_string())?,
|
||
);
|
||
let Some(end) = ofs.checked_add(size) else {
|
||
return Err("the delta copy range overflows".to_string());
|
||
};
|
||
let Some(bytes) = base.get(ofs..end) else {
|
||
return Err("the delta copy range exceeds the base object size".to_string());
|
||
};
|
||
out.extend_from_slice(bytes);
|
||
} else if cmd == 0 {
|
||
return Err("the delta command 0 is reserved and invalid".to_string());
|
||
} else {
|
||
let end = cursor + usize::from(cmd);
|
||
let Some(bytes) = delta.get(cursor..end) else {
|
||
return Err("the delta insert data is truncated".to_string());
|
||
};
|
||
cursor = end;
|
||
out.extend_from_slice(bytes);
|
||
}
|
||
if u64::try_from(out.len()).unwrap_or(u64::MAX) > result_size {
|
||
return Err(format!(
|
||
"the delta instructions produce more than the declared {result_size} bytes"
|
||
));
|
||
}
|
||
}
|
||
if out.len() as u64 != result_size {
|
||
return Err(format!(
|
||
"the delta instructions produce {} bytes but the delta declares {result_size}",
|
||
out.len()
|
||
));
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
/// Inflates one kept pack entry back to its bytes (the delta payload
|
||
/// for delta entries, the object body for base entries).
|
||
///
|
||
/// The decompressed-size cap (`cap`, the ingestion budget's
|
||
/// `max_object_size`) applies twice: a declaration beyond the cap is
|
||
/// rejected before any reading, and the inflation itself is
|
||
/// incremental — the produced bytes accumulate in bounded chunks and
|
||
/// the read aborts at the cap, so no allocation is ever sized by the
|
||
/// pack-declared size (the size-bomb class, review 004 R-2: a hostile
|
||
/// pack declares an entry and decompresses to gigabytes; the old
|
||
/// allocate-then-fill shape trusted the declaration).
|
||
fn inflate_entry(entry: &PackEntry, cap: u64) -> Result<Vec<u8>, String> {
|
||
let compressed = entry
|
||
.compressed
|
||
.as_ref()
|
||
.ok_or_else(|| "the pack entry lost its compressed bytes".to_string())?;
|
||
if entry.decompressed_size > cap {
|
||
return Err(format!(
|
||
"the pack entry declares {} decompressed bytes beyond the ingestion cap of {cap} bytes",
|
||
entry.decompressed_size
|
||
));
|
||
}
|
||
let mut out: Vec<u8> = Vec::new();
|
||
let mut decompressor = gix_zlib::Decompress::new();
|
||
let mut reader = io::BufReader::new(io::Cursor::new(compressed.as_slice()));
|
||
let mut chunk = vec![0u8; INFLATE_CHUNK_BYTES];
|
||
loop {
|
||
let written = gix_zlib::stream::inflate::read(&mut reader, &mut decompressor, &mut chunk)
|
||
.map_err(|error| format!("the pack entry does not inflate: {error}"))?;
|
||
out.extend_from_slice(&chunk[..written]);
|
||
if u64::try_from(out.len()).unwrap_or(u64::MAX) > cap {
|
||
return Err(format!(
|
||
"the pack entry inflates past the ingestion cap of {cap} bytes"
|
||
));
|
||
}
|
||
if written < chunk.len() {
|
||
break;
|
||
}
|
||
}
|
||
if out.len() as u64 != entry.decompressed_size {
|
||
return Err(format!(
|
||
"the pack entry inflates to {} bytes but {} were declared",
|
||
out.len(),
|
||
entry.decompressed_size
|
||
));
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
/// The object kind of a base-entry header.
|
||
fn base_kind(header: &PackHeader) -> Option<gix_object::Kind> {
|
||
match header {
|
||
PackHeader::Blob => Some(gix_object::Kind::Blob),
|
||
PackHeader::Tree => Some(gix_object::Kind::Tree),
|
||
PackHeader::Commit => Some(gix_object::Kind::Commit),
|
||
PackHeader::Tag => Some(gix_object::Kind::Tag),
|
||
PackHeader::OfsDelta { .. } | PackHeader::RefDelta { .. } => None,
|
||
}
|
||
}
|
||
|
||
/// The inflate chunk the expansion cap reads in (bounded output
|
||
/// slices — the incremental reading itself never buffers beyond the
|
||
/// chunk plus the produced bytes).
|
||
const INFLATE_CHUNK_BYTES: usize = 65536;
|
||
|
||
/// Expands one delta entry against the staged in-pack objects: the
|
||
/// base is resolved by pack offset (ofs delta) or oid (ref delta) —
|
||
/// `None` means the base is not staged (yet), leaving the entry
|
||
/// pending for the fixpoint pass.
|
||
fn expand_delta(
|
||
staged_by_offset: &BTreeMap<u64, Staged>,
|
||
staged_by_oid: &BTreeMap<ObjectId, Staged>,
|
||
entry: &PackEntry,
|
||
cap: u64,
|
||
) -> Result<Option<Staged>, String> {
|
||
let base = match &entry.header {
|
||
PackHeader::OfsDelta { base_distance } => {
|
||
let Some(base_offset) = entry.pack_offset.checked_sub(*base_distance) else {
|
||
return Err(format!(
|
||
"the ofs delta at pack offset {} names an impossible base distance {base_distance}",
|
||
entry.pack_offset
|
||
));
|
||
};
|
||
match staged_by_offset.get(&base_offset) {
|
||
Some(base) => base,
|
||
None => return Ok(None),
|
||
}
|
||
}
|
||
PackHeader::RefDelta { base_id } => match staged_by_oid.get(base_id) {
|
||
Some(base) => base,
|
||
None => return Ok(None),
|
||
},
|
||
PackHeader::Blob | PackHeader::Tree | PackHeader::Commit | PackHeader::Tag => {
|
||
return Err("a base-entry header cannot expand as a delta".to_string());
|
||
}
|
||
};
|
||
let payload = inflate_entry(entry, cap)?;
|
||
let body = apply_delta(&base.body, &payload, cap)?;
|
||
let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, base.kind, &body)
|
||
.map_err(|error| format!("the expanded delta object does not hash: {error}"))?;
|
||
Ok(Some(Staged {
|
||
oid,
|
||
kind: base.kind,
|
||
body,
|
||
}))
|
||
}
|
||
|
||
/// Stages one pack entry: base entries inflate and re-hash, delta
|
||
/// entries expand against the staged bases; an unresolvable delta is
|
||
/// parked for the fixpoint pass (in-pack forward references are legal).
|
||
fn stage_entry(
|
||
staged_by_offset: &mut BTreeMap<u64, Staged>,
|
||
staged_by_oid: &mut BTreeMap<ObjectId, Staged>,
|
||
pending: &mut Vec<PackEntry>,
|
||
entry: PackEntry,
|
||
cap: u64,
|
||
) -> Result<(), String> {
|
||
let staged = match &entry.header {
|
||
PackHeader::Blob | PackHeader::Tree | PackHeader::Commit | PackHeader::Tag => {
|
||
let kind = base_kind(&entry.header)
|
||
.ok_or_else(|| "a base-entry header carries no object kind".to_string())?;
|
||
let body = inflate_entry(&entry, cap)?;
|
||
let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &body)
|
||
.map_err(|error| format!("the pack entry object does not hash: {error}"))?;
|
||
Some(Staged { oid, kind, body })
|
||
}
|
||
_ => expand_delta(staged_by_offset, staged_by_oid, &entry, cap)?,
|
||
};
|
||
match staged {
|
||
Some(staged) => {
|
||
staged_by_offset.insert(entry.pack_offset, staged.clone());
|
||
staged_by_oid.insert(staged.oid, staged);
|
||
}
|
||
None => pending.push(entry),
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// The server-chosen reason for a delta whose base resolved nowhere:
|
||
/// a ref delta may name the missing base oid (the oid came from the
|
||
/// client's own pack, no oracle risk — ADR-018 §7's `ObjectMissing`
|
||
/// note applies at the reason's discretion).
|
||
fn unresolved_delta_reason(entry: &PackEntry) -> String {
|
||
match &entry.header {
|
||
PackHeader::RefDelta { base_id } => {
|
||
format!("the delta base {base_id} is neither in the pack nor in the repository")
|
||
}
|
||
PackHeader::OfsDelta { base_distance } => format!(
|
||
"the ofs delta's base at distance {base_distance} does not resolve within the pack"
|
||
),
|
||
_ => "a base entry cannot remain unresolved".to_string(),
|
||
}
|
||
}
|
||
|
||
/// Lands the staged objects into the repo's namespace as loose-form
|
||
/// rows keyed by their oids — the verification posture the substrate
|
||
/// gets from the store's verification-on-put is test-side re-derivation
|
||
/// (the oids were already re-derived at expansion).
|
||
fn land_staged(objects: &MemObjects, repo: &RepoRecord, staged: &BTreeMap<ObjectId, Staged>) {
|
||
for (oid, object) in staged {
|
||
let header = gix_object::encode::loose_header(object.kind, object.body.len() as u64);
|
||
let mut loose = Vec::new();
|
||
loose.extend_from_slice(&header);
|
||
loose.extend_from_slice(&object.body);
|
||
objects.put(repo, *oid, loose);
|
||
}
|
||
}
|
||
|
||
/// The full-graph connectivity walk from the pushed tips (ADR-020 §2,
|
||
/// parents included — `gix-fsck`'s `Connectivity::check_commit` does
|
||
/// not walk parents; the owning implementation walks): commits walk
|
||
/// parents and trees, trees walk entries, tags walk targets — over the
|
||
/// combined graph of the staged rows and the pre-existing namespace.
|
||
fn check_connectivity(
|
||
objects: &MemObjects,
|
||
repo: &RepoRecord,
|
||
staged: &BTreeMap<ObjectId, Staged>,
|
||
tips: &[ObjectId],
|
||
) -> Result<(), String> {
|
||
let mut seen: HashSet<ObjectId> = HashSet::new();
|
||
let mut stack: Vec<ObjectId> = tips.to_vec();
|
||
while let Some(oid) = stack.pop() {
|
||
if !seen.insert(oid) {
|
||
continue;
|
||
}
|
||
let (kind, body) = if let Some(object) = staged.get(&oid) {
|
||
(object.kind, object.body.clone())
|
||
} else {
|
||
let Some(bytes) = objects.find(repo, &oid) else {
|
||
return Err(format!("missing object {oid}"));
|
||
};
|
||
let (kind, _size, header_len) = gix_object::decode::loose_header(&bytes)
|
||
.map_err(|error| format!("corrupt object {oid}: {error}"))?;
|
||
(kind, bytes[header_len..].to_vec())
|
||
};
|
||
match kind {
|
||
gix_object::Kind::Commit => {
|
||
let commit = gix_object::CommitRef::from_bytes(&body, gix_hash::Kind::Sha1)
|
||
.map_err(|error| format!("corrupt commit {oid}: {error}"))?;
|
||
stack.push(commit.tree());
|
||
for parent in commit.parents() {
|
||
stack.push(parent);
|
||
}
|
||
}
|
||
gix_object::Kind::Tree => {
|
||
let tree = gix_object::TreeRef::from_bytes(&body, gix_hash::Kind::Sha1)
|
||
.map_err(|error| format!("corrupt tree {oid}: {error}"))?;
|
||
for entry in &tree.entries {
|
||
stack.push(entry.oid.into());
|
||
}
|
||
}
|
||
gix_object::Kind::Tag => {
|
||
let tag = gix_object::TagRef::from_bytes(&body, gix_hash::Kind::Sha1)
|
||
.map_err(|error| format!("corrupt tag {oid}: {error}"))?;
|
||
stack.push(tag.target());
|
||
}
|
||
gix_object::Kind::Blob => {}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// The ingestion over one boxed pack stream (the ADR-020 §2
|
||
/// composition, the streaming-input form): the budgeted stream fails
|
||
/// into the ng leg past the size budget; the pack header, entries, and
|
||
/// trailer verify through the machinery (Mode::Verify) — except the
|
||
/// zero-object pack, whose trailer the machinery never reaches, so the
|
||
/// double verifies it itself; thin bases resolve from the repo's
|
||
/// namespace through [MemFind]; in-pack forward ref-delta references
|
||
/// resolve in a bounded fixpoint pass; and every expansion is bounded
|
||
/// by `entry_cap` — the decompressed-size cap (`Limits::max_object_size`)
|
||
/// the staged entries inflate and materialize under, the declared
|
||
/// size never pre-allocated (review 004 R-2's guard). Returns the
|
||
/// staged objects — the caller lands them only after the connectivity
|
||
/// walk closes. Every failure is pack-level validation and reports the
|
||
/// server- chosen `unpack ng` reason; `Err` (the `StorageError`) stays
|
||
/// for infrastructure failure only, which the mem double has none of.
|
||
fn ingest_pack(
|
||
objects: &MemObjects,
|
||
repo: &RepoRecord,
|
||
pack: Box<dyn Read + Send>,
|
||
budget: u64,
|
||
entry_cap: u64,
|
||
) -> Result<BTreeMap<ObjectId, Staged>, String> {
|
||
let mut budgeted = BudgetedReader::new(pack, budget);
|
||
let mut header = [0u8; gix_pack::data::header::SIZE];
|
||
budgeted
|
||
.read_exact(&mut header)
|
||
.map_err(|_| "the pack stream ended in the header".to_string())?;
|
||
let (_version, object_count) = gix_pack::data::header::decode(&header)
|
||
.map_err(|error| format!("the pack header does not decode: {error}"))?;
|
||
|
||
let mut staged_by_offset: BTreeMap<u64, Staged> = BTreeMap::new();
|
||
let mut staged_by_oid: BTreeMap<ObjectId, Staged> = BTreeMap::new();
|
||
|
||
if object_count == 0 {
|
||
// The header bytes were already consumed from `budgeted`, so the
|
||
// trailer — which the streaming machinery never reaches for a
|
||
// zero-object pack — is read from the stream directly.
|
||
let mut trailer_reader = io::BufReader::new(budgeted);
|
||
let mut trailer = [0u8; gix_hash::Kind::Sha1.len_in_bytes()];
|
||
trailer_reader
|
||
.read_exact(&mut trailer)
|
||
.map_err(|_| "the zero-object pack stream ends before the trailer".to_string())?;
|
||
let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
|
||
hasher.update(&header);
|
||
let digest = hasher
|
||
.try_finalize()
|
||
.map_err(|error| format!("the pack trailer does not hash: {error}"))?;
|
||
if digest != ObjectId::from(trailer) {
|
||
return Err(
|
||
"the zero-object pack's trailer does not verify against the pack bytes".to_string(),
|
||
);
|
||
}
|
||
return Ok(staged_by_oid);
|
||
}
|
||
|
||
let reader = io::BufReader::new(io::Cursor::new(header).chain(budgeted));
|
||
let entries = BytesToEntriesIter::new_from_header(
|
||
reader,
|
||
InputMode::Verify,
|
||
EntryDataMode::Keep,
|
||
gix_hash::Kind::Sha1,
|
||
)
|
||
.map_err(|error| format!("the pack header does not decode: {error}"))?;
|
||
let lookup = MemFind::new(objects, repo);
|
||
let mut thin = LookupRefDeltaObjectsIter::new(entries, lookup, gix_zlib::Compression::DEFAULT);
|
||
let mut pending: Vec<PackEntry> = Vec::new();
|
||
for step in thin.by_ref() {
|
||
let entry = step.map_err(|error| format!("the pack stream fails verification: {error}"))?;
|
||
stage_entry(
|
||
&mut staged_by_offset,
|
||
&mut staged_by_oid,
|
||
&mut pending,
|
||
entry,
|
||
entry_cap,
|
||
)?;
|
||
}
|
||
while !pending.is_empty() {
|
||
let drained = std::mem::take(&mut pending);
|
||
let count = drained.len();
|
||
for entry in drained {
|
||
stage_entry(
|
||
&mut staged_by_offset,
|
||
&mut staged_by_oid,
|
||
&mut pending,
|
||
entry,
|
||
entry_cap,
|
||
)?;
|
||
}
|
||
if pending.len() == count {
|
||
return Err(unresolved_delta_reason(&pending[0]));
|
||
}
|
||
}
|
||
Ok(staged_by_oid)
|
||
}
|
||
|
||
/// The pack-ingestion double — [`GitPackIngest`] sharing the
|
||
/// [`MemObjects`] namespace: `prepare` composes the ADR-020 §2
|
||
/// ingestion over the streaming input iterators (the module doc's
|
||
/// recorded composition) — the budgeted stream, thin bases from
|
||
/// [`MemFind`] (ADR-013 §5), every pack object expanded, and the
|
||
/// full-graph connectivity walk from the pushed tips with parents
|
||
/// included. Sync test code — the blocking-thread posture is the
|
||
/// machine's (ADR-009, same as [`MemPackGen::generate`]). Pack-level
|
||
/// validation failures (parse, index, thin-base miss, connectivity,
|
||
/// budget, the decompressed-size cap —
|
||
/// `Limits::max_object_size`'s incremental inflate, review 004 R-2)
|
||
/// land in [`PreparedPush::unpack`]'s ng leg, never `Err`; the
|
||
/// session continues to the report per ADR-013 §8. `updates` returns
|
||
/// the input list verbatim on every path, including the empty pack
|
||
/// (zero objects is valid and reported as such) and every ng leg — the
|
||
/// wire layer needs the ref names for the all-refs unpacker-error
|
||
/// report (the amended seam's round-trip rule, ADR-018 §5).
|
||
#[derive(Clone, Default)]
|
||
pub struct MemPackIngest {
|
||
pub objects: MemObjects,
|
||
}
|
||
|
||
impl MemPackIngest {
|
||
/// The double's object namespace, for the ingested objects' loose
|
||
/// rows to be inspected against.
|
||
pub fn objects(&self) -> &MemObjects {
|
||
&self.objects
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitPackIngest for MemPackIngest {
|
||
async fn prepare(
|
||
&self,
|
||
repo: &RepoRecord,
|
||
updates: Vec<RefUpdate>,
|
||
pack: Box<dyn Read + Send>,
|
||
_push_options: Option<PushOptions>,
|
||
limits: Limits,
|
||
) -> Result<PreparedPush, StorageError> {
|
||
let unpack = match ingest_pack(
|
||
&self.objects,
|
||
repo,
|
||
pack,
|
||
limits.max_pack_size,
|
||
limits.max_object_size,
|
||
) {
|
||
Ok(staged) => {
|
||
let mut tips: Vec<ObjectId> = Vec::new();
|
||
for update in &updates {
|
||
if let Some(new) = update.new {
|
||
tips.push(new);
|
||
}
|
||
}
|
||
match check_connectivity(&self.objects, repo, &staged, &tips) {
|
||
Ok(()) => {
|
||
land_staged(&self.objects, repo, &staged);
|
||
Ok(())
|
||
}
|
||
Err(reason) => Err(reason),
|
||
}
|
||
}
|
||
Err(reason) => Err(reason),
|
||
};
|
||
Ok(PreparedPush { unpack, updates })
|
||
}
|
||
}
|
||
|
||
/// A sink whose written bytes the rows inspect after the fact — a
|
||
/// shared buffer wrapped in the [`io::Write`] the seam's boxed sink
|
||
/// carries (`Send`), so an aborted generation's emptiness is provable.
|
||
#[derive(Clone, Default)]
|
||
pub struct SharedSink {
|
||
bytes: Arc<Mutex<Vec<u8>>>,
|
||
}
|
||
|
||
impl SharedSink {
|
||
/// The bytes written so far, in write order.
|
||
pub fn snapshot(&self) -> Vec<u8> {
|
||
self.bytes.lock().clone()
|
||
}
|
||
}
|
||
|
||
impl io::Write for SharedSink {
|
||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||
self.bytes.lock().extend_from_slice(buf);
|
||
Ok(buf.len())
|
||
}
|
||
|
||
fn flush(&mut self) -> io::Result<()> {
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
fn fixture_signature() -> gix_actor::Signature {
|
||
gix_actor::Signature {
|
||
name: b"alkgit doubles".into(),
|
||
email: b"doubles@alk.dev".into(),
|
||
time: gix_object::date::Time {
|
||
seconds: 1_700_000_000,
|
||
offset: 0,
|
||
},
|
||
}
|
||
}
|
||
|
||
fn seeded_loose(kind: gix_object::Kind, body: &[u8]) -> (ObjectId, Vec<u8>) {
|
||
let id = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, body)
|
||
.expect("fixture hashes its own object");
|
||
let mut bytes = Vec::new();
|
||
bytes.extend_from_slice(&gix_object::encode::loose_header(kind, body.len() as u64));
|
||
bytes.extend_from_slice(body);
|
||
(id, bytes)
|
||
}
|
||
|
||
/// Builds a blob fixture and returns `(oid, loose bytes)`.
|
||
pub fn blob_fixture(content: &[u8]) -> (ObjectId, Vec<u8>) {
|
||
seeded_loose(gix_object::Kind::Blob, content)
|
||
}
|
||
|
||
/// Decodes one loose object's bytes back to `(kind, body)` — the
|
||
/// pack-entry fixture source (the fixture objects' loose encoding is
|
||
/// the seeded row; the entries encode from the body).
|
||
pub fn loose_body(bytes: &[u8]) -> (gix_object::Kind, Vec<u8>) {
|
||
let (kind, _size, header_len) =
|
||
gix_object::decode::loose_header(bytes).expect("fixture objects decode");
|
||
(kind, bytes[header_len..].to_vec())
|
||
}
|
||
|
||
/// Builds a tree fixture from `(kind, filename, oid)` entries; the
|
||
/// entries are serialized in git's own entry order.
|
||
pub fn tree_fixture(
|
||
entries: &[(gix_object::tree::EntryKind, &str, ObjectId)],
|
||
) -> (ObjectId, Vec<u8>) {
|
||
let mut entries: Vec<gix_object::tree::Entry> = entries
|
||
.iter()
|
||
.map(|(kind, filename, oid)| gix_object::tree::Entry {
|
||
mode: (*kind).into(),
|
||
filename: (*filename).into(),
|
||
oid: *oid,
|
||
})
|
||
.collect();
|
||
entries.sort();
|
||
let tree = gix_object::Tree { entries };
|
||
let mut body = Vec::new();
|
||
gix_object::WriteTo::write_to(&tree, &mut body).expect("fixture trees serialize");
|
||
seeded_loose(gix_object::Kind::Tree, &body)
|
||
}
|
||
|
||
/// Builds a commit fixture over `tree` with `parents`; returns
|
||
/// `(oid, loose bytes)`.
|
||
pub fn commit_fixture(tree: ObjectId, parents: &[ObjectId], message: &str) -> (ObjectId, Vec<u8>) {
|
||
let commit = gix_object::Commit {
|
||
tree,
|
||
parents: parents.iter().copied().collect(),
|
||
author: fixture_signature(),
|
||
committer: fixture_signature(),
|
||
encoding: None,
|
||
message: message.into(),
|
||
extra_headers: Vec::new(),
|
||
};
|
||
let mut body = Vec::new();
|
||
gix_object::WriteTo::write_to(&commit, &mut body).expect("fixture commits serialize");
|
||
seeded_loose(gix_object::Kind::Commit, &body)
|
||
}
|
||
|
||
/// Builds an annotated tag fixture over `target`; returns
|
||
/// `(oid, loose bytes)`.
|
||
pub fn tag_fixture(
|
||
target: ObjectId,
|
||
target_kind: gix_object::Kind,
|
||
name: &str,
|
||
) -> (ObjectId, Vec<u8>) {
|
||
let tag = gix_object::Tag {
|
||
target,
|
||
target_kind,
|
||
name: name.into(),
|
||
tagger: Some(fixture_signature()),
|
||
message: b"a fixture tag\n".into(),
|
||
signature: None,
|
||
};
|
||
let mut body = Vec::new();
|
||
gix_object::WriteTo::write_to(&tag, &mut body).expect("fixture tags serialize");
|
||
seeded_loose(gix_object::Kind::Tag, &body)
|
||
}
|
||
|
||
/// A hostile delta stream: a valid base-size declaration, a declared
|
||
/// materialized result, and no instructions — the applier's
|
||
/// decompressed-size-cap check (and, past it, the exact-produced-size
|
||
/// check) fires before anything accumulates.
|
||
pub fn hostile_delta_stream(base_size: u64, declared_result: u64) -> Vec<u8> {
|
||
let mut out = delta_size_varint(base_size);
|
||
out.extend(delta_size_varint(declared_result));
|
||
out
|
||
}
|
||
|
||
/// Deflates bytes as one zlib stream (the pack entries' compression).
|
||
fn deflate_bytes(bytes: &[u8]) -> Vec<u8> {
|
||
let mut deflate =
|
||
gix_zlib::stream::deflate::Write::new(Vec::new(), gix_zlib::Compression::DEFAULT);
|
||
deflate.write_all(bytes).expect("fixture bytes deflate");
|
||
deflate
|
||
.flush()
|
||
.expect("the fixture deflate stream finishes");
|
||
deflate.into_inner()
|
||
}
|
||
|
||
/// One whole-base pack entry (the canonical encoding of `Commit`,
|
||
/// `Tree`, `Blob`, `Tag` entries) at `pack_offset`.
|
||
pub fn pack_entry_base(kind: gix_object::Kind, body: &[u8], pack_offset: u64) -> PackEntry {
|
||
let header = match kind {
|
||
gix_object::Kind::Blob => PackHeader::Blob,
|
||
gix_object::Kind::Tree => PackHeader::Tree,
|
||
gix_object::Kind::Commit => PackHeader::Commit,
|
||
gix_object::Kind::Tag => PackHeader::Tag,
|
||
};
|
||
let compressed = deflate_bytes(body);
|
||
PackEntry {
|
||
header,
|
||
header_size: header.size(body.len() as u64) as u16,
|
||
pack_offset,
|
||
compressed: Some(compressed.clone()),
|
||
compressed_size: compressed.len() as u64,
|
||
crc32: None,
|
||
decompressed_size: body.len() as u64,
|
||
trailer: None,
|
||
}
|
||
}
|
||
|
||
/// One delta pack entry at `pack_offset`: the delta `payload` is the
|
||
/// zlib-deflated entry content; `header` carries the base reference
|
||
/// (a [`PackHeader::RefDelta`] thin-base or an in-pack
|
||
/// [`PackHeader::OfsDelta`]).
|
||
pub fn pack_entry_delta(header: PackHeader, payload: &[u8], pack_offset: u64) -> PackEntry {
|
||
let compressed = deflate_bytes(payload);
|
||
PackEntry {
|
||
header,
|
||
header_size: header.size(payload.len() as u64) as u16,
|
||
pack_offset,
|
||
compressed: Some(compressed.clone()),
|
||
compressed_size: compressed.len() as u64,
|
||
crc32: None,
|
||
decompressed_size: payload.len() as u64,
|
||
trailer: None,
|
||
}
|
||
}
|
||
|
||
/// One size field of a delta stream header: 7-bit little-endian
|
||
/// groups with the continuation bit.
|
||
fn delta_size_varint(size: u64) -> Vec<u8> {
|
||
let mut out = Vec::new();
|
||
let mut rest = size;
|
||
loop {
|
||
let mut byte = rest as u8 & 0x7f;
|
||
rest >>= 7;
|
||
if rest != 0 {
|
||
byte |= 0x80;
|
||
}
|
||
out.push(byte);
|
||
if rest == 0 {
|
||
return out;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// A delta stream copying the whole base and inserting `result` as
|
||
/// literal instructions — the fixture form of a real client's delta:
|
||
/// the produced object is the whole base followed by `result` (the
|
||
/// copy instruction exercises the applier's copy path, the insertion
|
||
/// the insert path). Bases of exactly 65536 bytes (the copy size's
|
||
/// special case) are rejected — fixture bases are small.
|
||
pub fn delta_stream(base_size: u64, result: &[u8]) -> Vec<u8> {
|
||
assert!(
|
||
base_size < 0x10000,
|
||
"fixture delta bases stay below the copy size's 65536 special case"
|
||
);
|
||
let mut out = delta_size_varint(base_size);
|
||
out.extend(delta_size_varint(base_size + result.len() as u64));
|
||
if base_size > 0 {
|
||
let mut cmd = 0x80u8;
|
||
let mut size_bytes = Vec::new();
|
||
let mut rest = base_size;
|
||
for bit in [0x10u8, 0x20, 0x40] {
|
||
if rest & 0xff != 0 {
|
||
cmd |= bit;
|
||
size_bytes.push(rest as u8);
|
||
}
|
||
rest >>= 8;
|
||
}
|
||
out.push(cmd);
|
||
out.extend(size_bytes);
|
||
}
|
||
for chunk in result.chunks(127) {
|
||
out.push(chunk.len() as u8);
|
||
out.extend_from_slice(chunk);
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Assembles a V2 sha1 pack from `entries`: the `PACK` header, the
|
||
/// entries (header + compressed bytes each), and the sha1 trailer
|
||
/// over the whole stream.
|
||
pub fn pack_bytes(entries: &[PackEntry]) -> Vec<u8> {
|
||
let mut out = Vec::new();
|
||
out.extend_from_slice(&gix_pack::data::header::encode(
|
||
gix_pack::data::Version::V2,
|
||
entries.len() as u32,
|
||
));
|
||
for entry in entries {
|
||
entry
|
||
.header
|
||
.write_to(entry.decompressed_size, &mut out)
|
||
.expect("fixture entry headers write");
|
||
out.extend_from_slice(
|
||
entry
|
||
.compressed
|
||
.as_ref()
|
||
.expect("fixture entries keep bytes"),
|
||
);
|
||
}
|
||
let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
|
||
hasher.update(&out);
|
||
let digest = hasher
|
||
.try_finalize()
|
||
.expect("fixture pack trailers hash over the pack bytes");
|
||
out.extend_from_slice(digest.as_slice());
|
||
out
|
||
}
|
||
|
||
/// The decoded pack contents the rows assert over: the parsed header,
|
||
/// the entry objects re-hashed from their decompressed bytes, and the
|
||
/// verified trailer.
|
||
pub struct DecodedPack {
|
||
/// The pack data version the header declares.
|
||
pub version: gix_pack::data::Version,
|
||
/// The object count the header declares.
|
||
pub object_count: u32,
|
||
/// The entries as `(oid, kind, object bytes)`, in pack order.
|
||
pub objects: Vec<(ObjectId, gix_object::Kind, Vec<u8>)>,
|
||
/// The trailer checksum, verified over the whole pack stream.
|
||
pub trailer: ObjectId,
|
||
}
|
||
|
||
/// The section flush of a push request's command section
|
||
/// (client framing).
|
||
pub const FLUSH: &[u8] = b"0000";
|
||
|
||
/// The V0-framed push request's framed command head (ADR-013 §3,
|
||
/// push-captures.md §Request): the command pkt-lines (caps
|
||
/// NUL-attached on the first line only) and the command-section flush —
|
||
/// the chunk list every substrate's client-side body builder chains the
|
||
/// pack (or the options section) onto.
|
||
pub fn push_commands(caps: &str, commands: &[String]) -> Vec<Vec<u8>> {
|
||
let mut chunks = Vec::new();
|
||
let mut commands = commands.iter();
|
||
if let Some(first) = commands.next() {
|
||
let line = if caps.is_empty() {
|
||
first.clone()
|
||
} else {
|
||
format!("{first}\0{caps}")
|
||
};
|
||
chunks.push(pkt(line.as_bytes()));
|
||
}
|
||
for command in commands {
|
||
chunks.push(pkt(command.as_bytes()));
|
||
}
|
||
chunks.push(FLUSH.to_vec());
|
||
chunks
|
||
}
|
||
|
||
fn pkt(payload: &[u8]) -> Vec<u8> {
|
||
let total = payload.len() + 4;
|
||
let mut out = Vec::with_capacity(total);
|
||
out.extend_from_slice(format!("{total:04x}").as_bytes());
|
||
out.extend_from_slice(payload);
|
||
out
|
||
}
|
||
|
||
pub fn create_command(name: &str, new: ObjectId) -> String {
|
||
format!("{} {new} {name}", "0".repeat(40))
|
||
}
|
||
|
||
pub fn delete_command(name: &str, current: ObjectId) -> String {
|
||
format!("{current} {} {name}", "0".repeat(40))
|
||
}
|
||
|
||
/// A fixture object: its oid and the (kind, body) its loose bytes
|
||
/// carry — the pack-entry source.
|
||
pub struct PackedFixture {
|
||
pub oid: ObjectId,
|
||
pub kind: gix_object::Kind,
|
||
pub body: Vec<u8>,
|
||
}
|
||
|
||
pub fn blob_fix(content: &[u8]) -> PackedFixture {
|
||
let (oid, bytes) = blob_fixture(content);
|
||
let (kind, body) = loose_body(&bytes);
|
||
PackedFixture { oid, kind, body }
|
||
}
|
||
|
||
pub fn tree_fix(entries: &[(gix_object::tree::EntryKind, &str, ObjectId)]) -> PackedFixture {
|
||
let (oid, bytes) = tree_fixture(entries);
|
||
let (kind, body) = loose_body(&bytes);
|
||
PackedFixture { oid, kind, body }
|
||
}
|
||
|
||
pub fn commit_fix(tree: ObjectId, parents: &[ObjectId], message: &str) -> PackedFixture {
|
||
let (oid, bytes) = commit_fixture(tree, parents, message);
|
||
let (kind, body) = loose_body(&bytes);
|
||
PackedFixture { oid, kind, body }
|
||
}
|
||
|
||
/// A two-commit chain + its self-contained V2 sha1 pack — the e2e
|
||
/// pushes' content.
|
||
pub struct ChainPack {
|
||
pub pack: Vec<u8>,
|
||
pub commit_a: ObjectId,
|
||
pub commit_b: ObjectId,
|
||
}
|
||
|
||
pub fn chain_pack() -> ChainPack {
|
||
let blob_a = blob_fix(b"alpha");
|
||
let blob_b = blob_fix(b"beta");
|
||
let tree_a = tree_fix(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid)]);
|
||
let tree_b = tree_fix(&[
|
||
(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid),
|
||
(gix_object::tree::EntryKind::Blob, "b.txt", blob_b.oid),
|
||
]);
|
||
let commit_a = commit_fix(tree_a.oid, &[], "alpha");
|
||
let commit_b = commit_fix(tree_b.oid, &[commit_a.oid], "beta");
|
||
let mut offset = 0u64;
|
||
let entries: Vec<_> = [&blob_a, &blob_b, &tree_a, &tree_b, &commit_a, &commit_b]
|
||
.iter()
|
||
.map(|fixture| {
|
||
let entry = pack_entry_base(fixture.kind, &fixture.body, offset);
|
||
offset += entry.bytes_in_pack();
|
||
entry
|
||
})
|
||
.collect();
|
||
ChainPack {
|
||
pack: pack_bytes(&entries),
|
||
commit_a: commit_a.oid,
|
||
commit_b: commit_b.oid,
|
||
}
|
||
}
|
||
|
||
/// Pkt-line payloads of a framed wire body (a flush as a trailing
|
||
/// empty payload).
|
||
pub fn parse_pkt_lines(body: &[u8]) -> Vec<Vec<u8>> {
|
||
let mut lines = Vec::new();
|
||
let mut cursor = body;
|
||
while !cursor.is_empty() {
|
||
let (header, rest) = cursor.split_at(4);
|
||
let len = usize::from_str_radix(std::str::from_utf8(header).expect("hex header"), 16)
|
||
.expect("hex length");
|
||
cursor = rest;
|
||
if len < 4 {
|
||
lines.push(Vec::new());
|
||
continue;
|
||
}
|
||
let (payload, rest) = cursor.split_at(len - 4);
|
||
cursor = rest;
|
||
lines.push(payload.to_vec());
|
||
}
|
||
lines
|
||
}
|
||
|
||
pub fn wired_text(payload: &[u8]) -> String {
|
||
String::from_utf8_lossy(payload)
|
||
.trim_end_matches('\n')
|
||
.to_owned()
|
||
}
|
||
|
||
/// The sideband-wrapped report's inner pkt-line lines: every data
|
||
/// packet is band-1, exactly one outer flush ends the response, the
|
||
/// band payload's own framing ends at its inner flush.
|
||
pub fn sideband_report_lines(body: &[u8]) -> Vec<String> {
|
||
let mut payload = Vec::new();
|
||
let mut outer_flushes = 0;
|
||
for packet in &parse_pkt_lines(body) {
|
||
if packet.is_empty() {
|
||
outer_flushes += 1;
|
||
continue;
|
||
}
|
||
assert_eq!(packet[0], 1, "band-1 framing carries the report");
|
||
payload.extend_from_slice(&packet[1..]);
|
||
}
|
||
assert_eq!(outer_flushes, 1, "exactly one outer flush");
|
||
let mut inner = parse_pkt_lines(&payload);
|
||
let last = inner.pop().expect("the inner framing is non-empty");
|
||
assert!(
|
||
last.is_empty(),
|
||
"the inner flush terminates the report: {last:?}"
|
||
);
|
||
inner.iter().map(|p| wired_text(p)).collect()
|
||
}
|
||
|
||
/// Decodes the sink's bytes back: the pack header (version, object
|
||
/// count), every entry inflated and re-hashed (deltas would fail — the
|
||
/// doubles carry no packs), and the trailer verified against the
|
||
/// stream. Entries are parsed with the pack machinery's own
|
||
/// `data::input` iterator in `Verify` mode.
|
||
pub fn decode_pack(pack: &[u8]) -> DecodedPack {
|
||
let header: [u8; gix_pack::data::header::SIZE] = pack[..gix_pack::data::header::SIZE]
|
||
.try_into()
|
||
.expect("pack header present");
|
||
let (version, object_count) =
|
||
gix_pack::data::header::decode(&header).expect("the pack header decodes");
|
||
let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
|
||
io::Cursor::new(pack),
|
||
gix_pack::data::input::Mode::Verify,
|
||
gix_pack::data::input::EntryDataMode::Keep,
|
||
gix_hash::Kind::Sha1,
|
||
)
|
||
.expect("the pack stream starts with a decodable header");
|
||
let mut objects = Vec::new();
|
||
let mut trailer = None;
|
||
let mut decompressor = gix_zlib::Decompress::new();
|
||
for entry in &mut entries {
|
||
let entry = entry.expect("every pack entry decodes");
|
||
if entry.trailer.is_some() {
|
||
trailer = entry.trailer;
|
||
}
|
||
let kind = match entry.header {
|
||
gix_pack::data::entry::Header::Commit => gix_object::Kind::Commit,
|
||
gix_pack::data::entry::Header::Tree => gix_object::Kind::Tree,
|
||
gix_pack::data::entry::Header::Blob => gix_object::Kind::Blob,
|
||
gix_pack::data::entry::Header::Tag => gix_object::Kind::Tag,
|
||
delta @ (gix_pack::data::entry::Header::OfsDelta { .. }
|
||
| gix_pack::data::entry::Header::RefDelta { .. }) => {
|
||
panic!("the doubles' packs carry no delta entries, got {delta:?}")
|
||
}
|
||
};
|
||
let compressed = entry.compressed.expect("keep mode retains the entry bytes");
|
||
let mut object = vec![0; entry.decompressed_size as usize];
|
||
let inflated = gix_zlib::stream::inflate::read(
|
||
&mut io::Cursor::new(compressed),
|
||
&mut decompressor,
|
||
&mut object,
|
||
)
|
||
.expect("the entry inflates");
|
||
decompressor.reset();
|
||
assert_eq!(
|
||
inflated,
|
||
object.len(),
|
||
"the inflate filled the entry's size"
|
||
);
|
||
let id = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &object)
|
||
.expect("the object re-hashes");
|
||
objects.push((id, kind, object));
|
||
}
|
||
DecodedPack {
|
||
version,
|
||
object_count,
|
||
objects,
|
||
trailer: trailer.unwrap_or_else(|| {
|
||
let body_len = pack.len() - gix_hash::Kind::Sha1.len_in_bytes();
|
||
let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
|
||
hasher.update(&pack[..body_len]);
|
||
let computed = hasher
|
||
.try_finalize()
|
||
.expect("the zero-entry trailer hashes over the header alone");
|
||
let digest: [u8; 20] =
|
||
<&[u8] as std::convert::TryInto<[u8; 20]>>::try_into(&pack[body_len..])
|
||
.expect("the trailer is a full digest");
|
||
let stored = ObjectId::from(digest);
|
||
assert_eq!(computed, stored, "the zero-entry pack's trailer verifies");
|
||
stored
|
||
}),
|
||
}
|
||
}
|