use std::sync::Arc;
use alkblobs::{Algorithm, EntryMeta, Pin, Store};
use gix_hash::ObjectId;
use gix_object::Kind;
use tokio::runtime::Handle;
use alkgit::StorageError;
pub(crate) const GIT_OBJECT_ALGORITHM: Algorithm = Algorithm::GitObjectSha1;
pub fn oid_key(oid: &gix_hash::oid) -> Result<alkblobs::Key, StorageError> {
if oid.kind() != gix_hash::Kind::Sha1 {
return Err(StorageError::Invalid(format!(
"oid {oid} is not a sha1 oid; the object front addresses only the sha1 family (OQ-05)"
)));
}
let mut bytes = Vec::with_capacity(1 + oid.as_bytes().len());
bytes.push(GIT_OBJECT_ALGORITHM.byte());
bytes.extend_from_slice(oid.as_bytes());
alkblobs::Key::from_bytes(&bytes)
.map_err(|e| StorageError::Invalid(format!("the oid-tagged key does not decode: {e}")))
}
pub(crate) const LOOSE_HEADER_PEEK_MAX: u64 = 128;
pub fn key_to_oid(key: &alkblobs::Key) -> Option<ObjectId> {
if key.algorithm() != GIT_OBJECT_ALGORITHM {
return None;
}
let raw = key.digest().bytes();
if raw.len() != gix_hash::Kind::Sha1.len_in_bytes() {
return None;
}
let mut bytes = [0u8; 20];
bytes.copy_from_slice(raw);
Some(ObjectId::from(bytes))
}
pub fn loose_form(kind: Kind, content: &[u8]) -> Vec<u8> {
let mut form = gix_object::encode::loose_header(kind, content.len() as u64).into_vec();
form.extend_from_slice(content);
form
}
pub(crate) fn store_leg_error(context: &str, error: alkblobs::Error) -> StorageError {
match error {
alkblobs::Error::Verification | alkblobs::Error::KeyInvalid => {
StorageError::Invalid(format!("{context}: {error}"))
}
alkblobs::Error::Missing
| alkblobs::Error::Io(_)
| alkblobs::Error::GcRefuse
| alkblobs::Error::RegistryFull => StorageError::Io(format!("{context}: {error}")),
}
}
#[derive(Clone)]
pub struct ObjectFront {
store: Arc<Store>,
runtime: Handle,
}
impl ObjectFront {
pub fn new(store: Arc<Store>) -> Self {
ObjectFront {
store,
runtime: Handle::current(),
}
}
pub fn session(&self) -> SessionFind {
SessionFind {
front: self.clone(),
missed: Arc::default(),
}
}
pub fn store(&self) -> &Arc<Store> {
&self.store
}
pub fn block_on<F: std::future::Future>(&self, fut: F) -> F::Output {
self.runtime.block_on(fut)
}
pub fn block_in_place<F>(&self, fut: F) -> F::Output
where
F: std::future::Future + Send,
F::Output: Send,
{
tokio::task::block_in_place(move || self.runtime.block_on(fut))
}
pub async fn put(
&self,
kind: Kind,
content: &[u8],
) -> Result<(alkblobs::Key, Pin), StorageError> {
let form = loose_form(kind, content);
let (key, pin) = self
.store
.put_as(
GIT_OBJECT_ALGORITHM,
Some(form.len() as u64),
std::io::Cursor::new(&form),
)
.await
.map_err(|e| store_leg_error("alkblobs put_as", e))?;
let oid = key_to_oid(&key).ok_or_else(|| {
StorageError::Invalid(format!(
"the put returned key {:?} is not the sha1 git-object family",
key.digest().bytes()
))
})?;
let derived = GIT_OBJECT_ALGORITHM.derive(&form, form.len() as u64);
if oid.as_bytes() != derived.bytes() {
return Err(StorageError::Invalid(format!(
"the put returned key {oid:?} does not re-derive to the object's oid"
)));
}
Ok((key, pin))
}
pub async fn get(&self, oid: &gix_hash::oid) -> Result<Option<(Kind, Vec<u8>)>, StorageError> {
let key = oid_key(oid)?;
let Some((len, blob)) = self
.store
.get(&key.as_bytes())
.await
.map_err(|e| store_leg_error("alkblobs get", e))?
else {
return Ok(None);
};
let value = blob
.read_at(0, len)
.await
.map_err(|e| store_leg_error("alkblobs read_at", e))?;
let (kind, header_len) = parse_loose_form(&value)?;
let derived = GIT_OBJECT_ALGORITHM.derive(&value, value.len() as u64);
if oid.as_bytes() != derived.bytes() {
return Err(StorageError::Invalid(format!(
"corruption detected at {oid}: the delivered value does not re-derive to the oid"
)));
}
Ok(Some((kind, value[header_len..].to_vec())))
}
pub async fn kind(&self, oid: &gix_hash::oid) -> Result<Option<Kind>, StorageError> {
let key = oid_key(oid)?;
let Some((len, blob)) = self
.store
.get(&key.as_bytes())
.await
.map_err(|e| store_leg_error("alkblobs get", e))?
else {
return Ok(None);
};
let peeked = blob
.read_at(0, LOOSE_HEADER_PEEK_MAX.min(len))
.await
.map_err(|e| store_leg_error("alkblobs header peek", e))?;
let (kind, _size, _header_len) =
gix_object::decode::loose_header(&peeked).map_err(|e| {
StorageError::Invalid(format!(
"the stored loose-object form's header does not parse: {e}"
))
})?;
Ok(Some(kind))
}
pub async fn stat(&self, oid: &gix_hash::oid) -> Result<Option<EntryMeta>, StorageError> {
let key = oid_key(oid)?;
let meta = self
.store
.stat(&key.as_bytes())
.await
.map_err(|e| store_leg_error("alkblobs stat", e))?;
match meta {
Some(meta) if meta.algorithm == GIT_OBJECT_ALGORITHM => Ok(Some(meta)),
Some(meta) => Err(StorageError::Invalid(format!(
"the family entry behind {oid} carries algorithm {:?}, not the sha1 git-object family",
meta.algorithm
))),
None => Ok(None),
}
}
}
#[derive(Clone)]
pub struct SessionFind {
front: ObjectFront,
missed: Arc<parking_lot::Mutex<Option<ObjectId>>>,
}
impl SessionFind {
pub fn missed_object(&self) -> Option<ObjectId> {
self.missed.lock().take()
}
}
fn find_error(leg: StorageError) -> gix_error::Error {
match leg {
StorageError::ObjectMissing { oid } => {
gix_error::message!("substrate odb: object {oid} does not exist").not_found_error()
}
StorageError::Invalid(_) => gix_error::message!("substrate odb: {leg}").validation_error(),
StorageError::Io(_) => gix_error::message!("substrate odb: {leg}").retryable_error(),
}
}
impl gix_object::Exists for SessionFind {
fn exists(&self, id: &gix_hash::oid) -> bool {
<Self as gix_pack::Find>::contains(self, id)
}
}
impl gix_object::Find for SessionFind {
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 SessionFind {
fn contains(&self, id: &gix_hash::oid) -> bool {
let oid: ObjectId = (id as &gix_hash::oid).to_owned();
let Ok(key) = oid_key(&oid) else {
return false;
};
self.front
.block_on(self.front.store.has(&key.as_bytes()))
.unwrap_or(false)
}
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 oid: ObjectId = (id as &gix_hash::oid).to_owned();
let found = self.front.block_on(self.front.get(&oid));
match found {
Ok(None) => {
self.missed.lock().replace(oid);
Ok(None)
}
Ok(Some((kind, data))) => {
buffer.clear();
buffer.extend_from_slice(&data);
Ok(Some((
gix_object::Data {
kind,
object_hash: id.kind(),
data: buffer,
},
None,
)))
}
Err(leg) => Err(find_error(leg)),
}
}
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 parse_loose_form(value: &[u8]) -> Result<(Kind, usize), StorageError> {
let (kind, size, header_len) = gix_object::decode::loose_header(value).map_err(|e| {
StorageError::Invalid(format!("the stored loose-object form does not parse: {e}"))
})?;
let body = value.len().saturating_sub(header_len) as u64;
if body != size {
return Err(StorageError::Invalid(format!(
"the stored loose-object form's body is {body} bytes against a header declaring {size}"
)));
}
Ok((kind, header_len))
}
#[cfg(test)]
mod tests {
use alkblobs::{Store, StoreMode};
use gix_hash::ObjectId;
use gix_object::Kind;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use super::{key_to_oid, loose_form, oid_key, ObjectFront, SessionFind};
use crate::testutil::scratch_dir as scratch_root;
use alkgit::StorageError;
fn mem_front() -> ObjectFront {
ObjectFront::new(Arc::new(Store::mem_only()))
}
async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
let (key, _pin) = front.put(kind, content).await.unwrap();
key_to_oid(&key).expect("the put's key is the family key")
}
fn blob_content() -> &'static [u8] {
b"hello\n"
}
fn tree_bytes(blob: &ObjectId) -> Vec<u8> {
let mut tree = b"100644 hello\0".to_vec();
tree.extend_from_slice(blob.as_bytes());
tree
}
fn commit_bytes(tree: &ObjectId, parents: &[&ObjectId]) -> Vec<u8> {
let mut body = format!("tree {tree}\n");
for parent in parents {
body.push_str(&format!("parent {parent}\n"));
}
body.push_str(
"author A <a@example.invalid> 1750000000 +0000\n\
committer A <a@example.invalid> 1750000000 +0000\n\n",
);
body.into_bytes()
}
async fn fixture_front() -> (ObjectFront, ObjectId) {
let front = mem_front();
let blob = put_object(&front, Kind::Blob, blob_content()).await;
let tree = front
.put(Kind::Tree, &tree_bytes(&blob))
.await
.map(|(key, _)| key_to_oid(&key).unwrap())
.unwrap();
let base = front
.put(Kind::Commit, &commit_bytes(&tree, &[]))
.await
.map(|(key, _)| key_to_oid(&key).unwrap())
.unwrap();
let tip = front
.put(Kind::Commit, &commit_bytes(&tree, &[&base]))
.await
.map(|(key, _)| key_to_oid(&key).unwrap())
.unwrap();
(front, tip)
}
#[tokio::test]
async fn put_get_round_trips_through_the_family() {
let front = mem_front();
for (kind, content) in [
(Kind::Blob, b"hello\n".as_slice()),
(Kind::Blob, b"".as_slice()),
(Kind::Blob, vec![0u8, 255, 0, 10, 13, 255].as_slice()),
] {
let (key, _pin) = front.put(kind, content).await.unwrap();
assert_eq!(
key.algorithm(),
alkblobs::Algorithm::GitObjectSha1,
"the family byte the front writes is the sha1 git-object member"
);
let (got_kind, got) = front
.get(&key_to_oid(&key).unwrap())
.await
.unwrap()
.unwrap();
assert_eq!(got_kind, kind);
assert_eq!(got, content);
}
let blob = put_object(&front, Kind::Blob, b"hello\n").await;
let tree = put_object(&front, Kind::Tree, &tree_bytes(&blob)).await;
let commit = put_object(&front, Kind::Commit, &commit_bytes(&tree, &[])).await;
let (got_kind, got) = front.get(&tree).await.unwrap().unwrap();
assert_eq!(got_kind, Kind::Tree);
assert_eq!(got, tree_bytes(&blob));
let (got_kind, got) = front.get(&commit).await.unwrap().unwrap();
assert_eq!(got_kind, Kind::Commit);
assert_eq!(got, commit_bytes(&tree, &[]));
}
#[tokio::test]
async fn the_loose_preimage_hashes_to_git_s_oid() {
let front = mem_front();
let vectors: &[(&[u8], Kind, &str)] = &[
(b"", Kind::Blob, "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391"),
(
b"hello world\n",
Kind::Blob,
"3b18e512dba79e4c8300dd08aeb37f8e728b8dad",
),
(b"", Kind::Tree, "4b825dc642cb6eb9a060e54bf8d69288fbee4904"),
];
for (content, kind, oid_hex) in vectors {
let oid = put_object(&front, *kind, content).await;
assert_eq!(
oid.to_string(),
*oid_hex,
"the derived oid is git's hash-object oid"
);
}
}
#[tokio::test]
async fn the_put_oid_matches_the_git_cli() {
if std::process::Command::new("git")
.arg("--version")
.output()
.is_err()
{
eprintln!("skipping (no git binary installed)");
return;
}
let dir = scratch_root("odb", "cli");
let dir = dir.display().to_string();
let out = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(&dir)
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.output()
.unwrap();
assert!(out.status.success(), "git init failed");
for content in [b"hello\n".as_slice(), b"hello world\n".as_slice()] {
let mut child = std::process::Command::new("git")
.current_dir(&dir)
.args(["hash-object", "--stdin"])
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.spawn()
.unwrap();
use std::io::Write as _;
child.stdin.as_mut().unwrap().write_all(content).unwrap();
let out = child.wait_with_output().unwrap();
assert!(out.status.success(), "git hash-object failed");
let oid_hex = String::from_utf8(out.stdout).unwrap().trim().to_owned();
let front = mem_front();
let oid = put_object(&front, Kind::Blob, content).await;
assert_eq!(oid.to_string(), oid_hex, "the put's oid matches git's");
}
std::fs::remove_dir_all(&dir).unwrap_or_default();
}
#[tokio::test]
async fn stat_reports_the_preimage_length() {
let front = mem_front();
let content = b"hello".as_slice();
let oid = put_object(&front, Kind::Blob, content).await;
let form = loose_form(Kind::Blob, content);
assert_ne!(form.len(), content.len(), "the header is inside the value");
let meta = front.stat(&oid).await.unwrap().expect("the entry stats");
assert_eq!(meta.len, form.len() as u64, "stat is the preimage length");
assert_eq!(meta.len, 12, "header (7) + content (5) = the preimage");
assert_eq!(
meta.algorithm,
alkblobs::Algorithm::GitObjectSha1,
"stat's algorithm is the sha1 family member"
);
let missing = ObjectId::from([7u8; 20]);
assert!(
front.stat(&missing).await.unwrap().is_none(),
"stat is a definitive miss, never a guessed zero"
);
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn corrupt_kvtier_entry_fails_the_read() {
let dir = scratch_root("odb", "kv-corrupt");
let db = dir.join("blobs.sqlite3");
let store = Arc::new(
Store::kv_only_sqlite(&db, 128 * 1024, alkblobs::DEFAULT_DELETE_WINDOW_DELAY)
.await
.unwrap(),
);
let front = ObjectFront::new(store);
let oid = put_object(&front, Kind::Blob, blob_content()).await;
assert!(front.get(&oid).await.is_ok(), "the pristine entry reads");
let key = oid_key(&oid).unwrap();
let conn = rusqlite::Connection::open(&db).unwrap();
conn.execute_batch("PRAGMA journal_mode = WAL;").unwrap();
let value = conn
.query_row(
"SELECT value FROM blobs WHERE key = ?",
[key.as_bytes()],
|row| row.get::<_, Vec<u8>>(0),
)
.unwrap();
assert_eq!(
value.len(),
loose_form(Kind::Blob, blob_content()).len(),
"the preimage is in the row"
);
let mut corrupted = value.clone();
corrupted[2] = b'9';
conn.execute(
"UPDATE blobs SET value = ? WHERE key = ?",
rusqlite::params![corrupted, key.as_bytes()],
)
.unwrap();
match front.get(&oid).await {
Err(StorageError::Invalid(reason)) => {
assert!(
reason.contains("verification failed: content does not match key")
|| reason.contains("re-derive")
|| reason.contains("parse"),
"the failure names the corruption, not a backing failure: {reason}"
);
}
Err(e) => panic!("corruption is a validation failure, not {e}"),
Ok(_) => panic!("a corrupted entry cannot read back"),
}
let find = front.session();
assert!(
tokio::task::spawn_blocking(move || {
gix_object::Find::try_find(&find, oid.as_ref(), &mut Vec::new()).is_err()
})
.await
.unwrap(),
"the Find leg fails the corrupted entry too"
);
std::fs::remove_dir_all(&dir).unwrap_or_default();
}
#[tokio::test]
async fn corrupt_largetier_entry_fails_the_read() {
let dir = scratch_root("odb", "large-corrupt");
let store = Arc::new(Store::dual_tier_mem_local(&dir, 4).await.unwrap());
const THRESHOLD: u64 = 4;
assert_eq!(store.mode(), StoreMode::DualTier, "the large tier is live");
let form_threshold = loose_form(Kind::Blob, b"!").len() as u64;
assert!(form_threshold > THRESHOLD, "any loose form routes large");
let front = ObjectFront::new(store);
let oid = put_object(&front, Kind::Blob, blob_content()).await;
assert!(front.get(&oid).await.is_ok(), "the pristine entry reads");
let key = oid_key(&oid).unwrap();
let hex = hex::encode(key.as_bytes());
let entry_path = dir.join(&hex[..2]).join(&hex[2..4]).join(&hex);
assert!(entry_path.is_file(), "the large tier's committed file");
let mut value = std::fs::read(&entry_path).unwrap();
let last = value.len() - 1;
value[last] ^= 0xff;
std::fs::write(&entry_path, &value).unwrap();
match front.get(&oid).await {
Err(StorageError::Invalid(reason)) => {
assert!(
reason.contains("re-derive"),
"the failure names the corruption, not a backing failure: {reason}"
);
}
Err(e) => panic!("corruption is a validation failure, not {e}"),
Ok(_) => panic!("a corrupted entry cannot read back"),
}
std::fs::remove_dir_all(&dir).unwrap_or_default();
}
#[tokio::test(flavor = "multi_thread")]
async fn the_kind_leg_answer_from_the_header_not_the_body() {
let dir = scratch_root("odb", "kind-leg");
let store = Arc::new(Store::dual_tier_mem_local(&dir, 4).await.unwrap());
let front = ObjectFront::new(store);
for (kind, content) in [
(Kind::Blob, b"hello\n".as_slice()),
(Kind::Tree, tree_bytes(&ObjectId::from([3u8; 20])).as_slice()),
(Kind::Commit, commit_bytes(&ObjectId::from([3u8; 20]), &[]).as_slice()),
(Kind::Tag, b"object \0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\ntype commit\ntag v\ntagger A <a@example.invalid> 1750000000 +0000\n\nm\n".as_slice()),
] {
let oid = put_object(&front, kind, content).await;
assert_eq!(front.kind(&oid).await.unwrap(), Some(kind), "{kind:?}");
}
let missing = ObjectId::from([0xE2u8; 20]);
assert!(
front.kind(&missing).await.unwrap().is_none(),
"a miss is None"
);
let big: Vec<u8> = std::iter::repeat_n(b'x', 4096).collect();
let oid = put_object(&front, Kind::Blob, &big).await;
assert_eq!(front.kind(&oid).await.unwrap(), Some(Kind::Blob));
std::fs::remove_dir_all(&dir).unwrap_or_default();
}
#[tokio::test]
async fn the_front_serves_only_the_sha1_family() {
let front = mem_front();
put_object(&front, Kind::Blob, b"one\n").await;
put_object(&front, Kind::Blob, b"two\n").await;
for key in front.store().list().await.unwrap() {
assert_eq!(key.algorithm(), alkblobs::Algorithm::GitObjectSha1);
assert_eq!(key.as_bytes()[0], 0x03, "the sha1 family byte");
}
let (key, _pin) = front.put(Kind::Blob, b"three\n").await.unwrap();
assert_eq!(
key.as_bytes()[0],
0x03,
"the sha256 member (0x04) is never written"
);
}
#[tokio::test]
async fn the_find_legs_answer_with_no_packed_objects() {
let (front, _tip) = fixture_front().await;
let blob = put_object(&front, Kind::Blob, blob_content()).await;
let find = front.session();
tokio::task::spawn_blocking(move || {
assert!(gix_object::Exists::exists(&find, blob.as_ref()));
let missing = ObjectId::from([9u8; 20]);
assert!(!gix_object::Exists::exists(&find, missing.as_ref()));
let mut buffer = Vec::new();
let data = gix_object::Find::try_find(&find, blob.as_ref(), &mut buffer)
.unwrap()
.expect("the blob is found");
assert_eq!(data.kind, Kind::Blob);
assert_eq!(data.data, blob_content());
assert_eq!(data.object_hash, gix_hash::Kind::Sha1);
use gix_object::FindExt as _;
let mut buffer = Vec::new();
let blob_ref = find.find_blob(blob.as_ref(), &mut buffer).unwrap();
assert_eq!(
blob_ref.data,
blob_content(),
"the FindExt blanket finds too"
);
use gix_pack::Find as _;
let id = blob.as_ref() as &gix_hash::oid;
assert!(
find.contains(id),
"the pack-level contains is the same lookup"
);
let locate = find.location_by_oid(id, &mut Vec::new()).unwrap();
assert!(locate.is_none(), "no pack holds the object");
let offsets = find.pack_offsets_and_oid(0).unwrap();
assert!(offsets.is_none(), "no pack exists");
let mut buffer = Vec::new();
let resolved = find.try_find(id, &mut buffer).unwrap().expect("found");
assert!(
resolved.1.is_none(),
"the location leg is none even on a resolved object"
);
})
.await
.unwrap();
}
#[tokio::test]
async fn entry_by_location_is_unanswered() {
let (front, _tip) = fixture_front().await;
let find = front.session();
let location = gix_pack::data::entry::Location {
pack_id: 0,
entry_size: 1,
pack_offset: 0,
};
assert!(gix_pack::Find::entry_by_location(&find, &location).is_none());
}
#[tokio::test]
async fn lookup_misses_record_per_session() {
let (front, _tip) = fixture_front().await;
let missing = ObjectId::from([11u8; 20]);
let blob = put_object(&front, Kind::Blob, b"again\n").await;
tokio::task::spawn_blocking(move || {
let missing_id = missing.as_ref() as &gix_hash::oid;
let one = front.session();
let two = front.session();
let mut buffer = Vec::new();
assert_eq!(
gix_object::Find::try_find(&one, missing_id, &mut buffer).unwrap(),
None,
"a miss is a recorded result, not an error"
);
assert_eq!(one.missed_object(), Some(missing));
assert!(two.missed_object().is_none(), "the miss is session-private");
assert!(one.missed_object().is_none(), "the slot drains on take");
let find = front.session();
assert!(
gix_object::Find::try_find(&find, missing.as_ref(), &mut Vec::new())
.unwrap()
.is_none()
);
assert_eq!(find.missed_object(), Some(missing));
assert!(
gix_object::Find::try_find(&find, blob.as_ref(), &mut Vec::new())
.unwrap()
.is_some(),
"a real lookup still succeeds after a recorded miss"
);
})
.await
.unwrap();
}
#[tokio::test]
#[should_panic]
async fn the_blocking_bridge_panics_in_async_contexts() {
let (front, tip) = fixture_front().await;
let _ = front.block_on(front.get(&tip));
}
#[tokio::test(flavor = "multi_thread")]
async fn the_async_context_bridges_with_block_in_place() {
let (front, tip) = fixture_front().await;
assert!(front.block_in_place(front.get(&tip)).unwrap().is_some());
}
type PipelineResult = Result<(Vec<u8>, gix_hash::ObjectId, u32), StorageError>;
fn pipeline_error(find: &SessionFind, e: gix_error::Error, fallback: ObjectId) -> StorageError {
if e.is_not_found() {
StorageError::ObjectMissing {
oid: find.missed_object().unwrap_or(fallback),
}
} else {
StorageError::Io(e.to_string())
}
}
fn closure_of(find: &SessionFind, tip: ObjectId) -> Result<Vec<ObjectId>, StorageError> {
let mut closure = Vec::new();
let mut stack = vec![tip];
while let Some(current) = stack.pop() {
if closure.contains(¤t) {
continue;
}
let mut buf = Vec::new();
let commit = gix_object::FindExt::find_commit(find, ¤t, &mut buf)
.map_err(|e| pipeline_error(find, e, tip))?;
for parent in &commit.parents {
let parent =
gix_hash::ObjectId::from_hex(parent).expect("the stored parent is hex");
stack.push(parent);
}
closure.push(current);
}
Ok(closure)
}
#[tokio::test]
async fn the_generation_pipeline_smokes_over_the_front() {
let (front, tip) = fixture_front().await;
let find = front.session();
let pack = tokio::task::spawn_blocking(move || smoke_pack(&find, tip))
.await
.unwrap();
let (pack_bytes, digest, _objects) =
pack.unwrap_or_else(|e| panic!("the pipeline packs over the front: {e}"));
assert!(pack_bytes.starts_with(b"PACK"), "the pack header wrote");
let trailing = &pack_bytes[pack_bytes.len() - 20..];
assert_eq!(
trailing,
digest.as_slice(),
"the trailer is the pipeline's digest"
);
}
fn smoke_pack(find: &SessionFind, want: ObjectId) -> PipelineResult {
use gix_pack::data::output::bytes::FromEntriesIter;
use gix_pack::data::output::count;
use gix_pack::data::output::entry::iter_from_counts;
use gix_pack::data::output::entry::iter_from_counts::{Mode, Options, Outcome};
use gix_parallel::Finalize as _;
let commits = closure_of(find, want)?;
let counter = gix_utils::progress::Discard;
let interrupt = AtomicBool::new(false);
let (counts, _outcome) = count::objects_unthreaded(
find,
&mut commits.into_iter().map(Ok::<ObjectId, gix_error::Error>),
&counter,
&interrupt,
count::objects::ObjectExpansion::TreeContents,
)
.map_err(|e| pipeline_error(find, e, want))?;
let counts_len = counts.len() as u32;
let sink = Vec::<u8>::new();
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(|e| pipeline_error(find, e, want))?,
);
let mut from_entries = FromEntriesIter::new(
entries.by_ref(),
sink,
counts_len,
gix_pack::data::Version::V2,
gix_hash::Kind::Sha1,
);
for step in from_entries.by_ref() {
step.map_err(|e| pipeline_error(find, e, want))?;
}
let digest = from_entries
.digest()
.ok_or_else(|| StorageError::Io("the trailer was never written".to_string()))?;
let pack_bytes = from_entries.into_write();
let statistics: Outcome = entries
.inner
.finalize()
.map_err(|e| pipeline_error(find, e, want))?;
assert_eq!(
statistics.missing_objects, 0,
"ADR-004's abort posture: no missing objects in the smoke"
);
Ok((pack_bytes, digest, counts_len))
}
}