mod common;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
#[cfg(feature = "sqlite")]
use alkgit_substrate::PackServing;
use gix_hash::ObjectId;
use alkgit_substrate::{MaintenanceOpts, WorkerEvent};
use common::{
extract_and_walk_client_pack, git, git_run, rev_list_oids, sideband_report_lines, DoorOptions,
GateRendezvous, Harness, DEFAULT_REPO_ID,
};
use parking_lot::Mutex;
const FIXTURE_LINES: usize = 300;
fn fixture_line(i: usize) -> String {
format!("alpha line {i} of the thin-push fixture file\n")
}
fn file_content(marker: &str) -> String {
let mut content: String = (0..FIXTURE_LINES).map(fixture_line).collect();
content.push_str(marker);
content
}
struct ClientDir(PathBuf);
impl ClientDir {
fn path(&self) -> &Path {
&self.0
}
}
async fn init_client(base: &Path, name: &str) -> Option<ClientDir> {
let dir = base.join(name);
std::fs::create_dir_all(&dir).expect("the client dir creates");
git(
Some(base),
&["init", "-q", "-b", "main", dir.to_str().expect("utf8 path")],
)
.await?;
git(Some(&dir), &["config", "user.name", "alkgit e2e"]).await?;
git(Some(&dir), &["config", "user.email", "e2e@alk.dev"]).await?;
Some(ClientDir(dir))
}
async fn commit_fixture_file(client: &ClientDir, marker: &str, message: &str) -> Option<String> {
std::fs::write(client.path().join("a.txt"), file_content(marker))
.expect("the fixture file writes");
git(Some(client.path()), &["add", "a.txt"]).await?;
git(Some(client.path()), &["commit", "-q", "-m", message]).await?;
rev_parse(client, "HEAD").await
}
async fn rev_parse(client: &ClientDir, what: &str) -> Option<String> {
let run = git(Some(client.path()), &["rev-parse", what]).await?;
Some(run.stdout.trim().to_owned())
}
async fn introduced_oids(client: &ClientDir, new: &str, base: &str) -> Option<BTreeSet<ObjectId>> {
let run = git(
Some(client.path()),
&["rev-list", "--objects", new, &format!("^{}", base)],
)
.await?;
Some(rev_list_oids(&run.stdout))
}
async fn real_push(
client: &ClientDir,
url: &str,
mut args: Vec<&str>,
refspecs: &[&str],
) -> Option<common::GitRun> {
let mut all = vec!["push"];
all.append(&mut args);
all.push(url);
for refspec in refspecs {
all.push(refspec);
}
git(Some(client.path()), &all).await
}
async fn real_clone(url: &str, dst: &Path) -> Option<common::GitRun> {
git(
None,
&[
"-c",
"protocol.version=2",
"clone",
"-q",
url,
dst.to_str().expect("utf8 dst"),
],
)
.await
}
async fn real_fetch(dst: &Path) -> Option<common::GitRun> {
git(Some(dst), &["fetch", "origin"]).await
}
async fn fsck_strict(dst: &Path) -> Option<common::GitRun> {
git(Some(dst), &["fsck", "--strict"]).await
}
async fn assert_clone_checked_out(dst: &Path, expected_tip: &str, expected_content: &str) {
let client = ClientDir(dst.to_path_buf());
let head = rev_parse(&client, "HEAD")
.await
.expect("the clone's HEAD resolves");
assert_eq!(
head, expected_tip,
"the clone checked out the pushed commit"
);
let bytes = std::fs::read(dst.join("a.txt")).expect("the fixture file checked out");
assert_eq!(
String::from_utf8_lossy(&bytes),
expected_content,
"the checked-out file is the pushed content"
);
fsck_strict(dst)
.await
.expect("the run rode a resolved git binary")
.expect_ok();
}
fn door_url(kind: &str, addr: std::net::SocketAddr, repo_id: &str) -> String {
match kind {
"git" => format!("git://{addr}/{repo_id}"),
_ => format!("http://{addr}/{repo_id}"),
}
}
async fn feed_events(harness: &Harness, since: i64) -> Vec<alkgit_substrate::RegistryEvent> {
let store = harness.assembly.reactor().store().clone();
let feed = store
.stream(alkgit_substrate::EVENTS_STREAM)
.await
.expect("the feed opens");
feed.read_since(since, 100)
.await
.expect("the feed reads")
.iter()
.filter_map(alkgit_substrate::RegistryEvent::decode_event)
.collect()
}
async fn listed_tips(harness: &Harness) -> Vec<(String, Option<ObjectId>)> {
harness
.assembly
.refs()
.list_refs(&harness.record)
.await
.expect("the refs side lists")
.into_iter()
.map(|line| (line.name, line.oid))
.collect()
}
async fn assert_unborn_head_row_persists(harness: &Harness) {
let tips = listed_tips(harness).await;
let head = tips
.iter()
.find(|(name, _)| name == "HEAD")
.expect("the seeded HEAD row persists");
assert!(
head.1.is_none(),
"the seeded HEAD stays the unborn symref row (a push never touches it)"
);
}
async fn assert_work_area_clean(harness: &Harness) {
let root = harness.work_root();
let mut read = tokio::fs::read_dir(&root)
.await
.expect("the working-area root opens");
let mut names = Vec::new();
while let Some(entry) = read.next_entry().await.expect("the work dir reads") {
names.push(entry.file_name());
}
assert!(
names.is_empty(),
"no per-push working directory remains after the pushes: {names:?}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_real_git_push_lands_objects_the_cas_and_the_reactive_legs_over_the_duplex_door() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("ordinary-push", DEFAULT_REPO_ID).await;
let capture = common::StreamCapture::new();
let base =
std::env::temp_dir().join(format!("alkgit-real-git-ordinary-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(
&harness,
DoorOptions {
capture: Some(capture.clone()),
..DoorOptions::default()
},
)
.await;
let client = init_client(&base, "seed").await.expect("the client inits");
commit_fixture_file(&client, "", "the ordinary push's commit")
.await
.expect("the seed commit lands");
let mut wake = harness
.assembly
.reactor()
.store()
.listen(alkgit_substrate::WAKE_CHANNEL)
.await
.expect("the wake source opens");
let url = door_url("git", door.addr, &harness.record.repo_id);
let push = real_push(&client, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let wake_event = tokio::time::timeout(
Duration::from_secs(5),
wake.recv_timeout(Duration::from_secs(5)),
)
.await
.expect("the announce's wake leg delivers inside the deadline")
.expect("the wake source stayed open")
.expect("the wake event delivered");
assert_eq!(wake_event.channel, alkgit_substrate::WAKE_CHANNEL);
assert!(
push.stderr().contains("main") && (push.stderr().contains("new branch")),
"the real client reported the new branch: {}",
push.stderr()
);
let report = sideband_report_lines(&capture.to_client());
assert_eq!(
report,
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned(),],
"the captured status report frames per ADR-013 §8 (band-1 wrapped, inner flush)"
);
assert!(
extract_and_walk_client_pack(&capture.to_server()).is_some(),
"the ordinary push's pack rode the wire (the peek saw the header)"
);
let commit_tip = rev_parse(&client, "HEAD").await.expect("tip");
let tree = rev_parse(&client, "HEAD^{tree}")
.await
.expect("tree resolves");
let blob = rev_parse(&client, "HEAD:a.txt")
.await
.expect("blob resolves");
let tip_oid = gix_hash::ObjectId::from_hex(commit_tip.as_bytes()).expect("the tip oid parses");
let tree_oid = gix_hash::ObjectId::from_hex(tree.as_bytes()).expect("the tree oid parses");
let blob_oid = gix_hash::ObjectId::from_hex(blob.as_bytes()).expect("the blob oid parses");
for (what, oid) in [
("the commit (tip)", tip_oid),
("the tree", tree_oid),
("the blob", blob_oid),
] {
let stat = harness
.assembly
.front()
.stat(&oid)
.await
.expect("the front stats");
assert!(
stat.is_some(),
"the pushed {what} landed in the store under its git oid"
);
}
assert_unborn_head_row_persists(&harness).await;
let tips = listed_tips(&harness).await;
assert_eq!(
tips.iter()
.find(|(name, _)| name == "refs/heads/main")
.and_then(|(_, oid)| *oid),
Some(tip_oid),
"the ref CAS applied the pushed main over the unborn-HEAD seed"
);
assert!(
harness.assembly.refs_version() >= 1,
"the CAS bumped the write counter"
);
assert_eq!(
harness
.assembly
.reactor()
.registry()
.watermark()
.await
.expect("the watermark reads"),
2,
"the marker watermark advanced past the record put (1) to the push's marker (2)"
);
assert_eq!(
feed_events(&harness, 0).await,
vec![
alkgit_substrate::RegistryEvent::Put {
seq: 1,
record: harness.record.clone(),
},
alkgit_substrate::RegistryEvent::Mutation {
seq: 2,
repo_id: harness.record.repo_id.clone(),
},
],
"the reactive legs' durable feed shows the put and the push's Mutation marker"
);
let report = harness
.assembly
.blobs()
.sweep()
.await
.expect("the sweep runs");
assert!(
report.aborted.is_none(),
"the sweep ran with the registered closure"
);
assert_eq!(
report.deleted, 0,
"the pushed objects are liveness-covered — the pins committed after the CAS"
);
assert!(harness
.assembly
.front()
.stat(&tip_oid)
.await
.expect("the front stats")
.is_some());
assert_work_area_clean(&harness).await;
door.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_real_thin_push_pushes_no_base_objects_and_resolves_them_from_the_substrate() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("thin-push", DEFAULT_REPO_ID).await;
let capture = common::StreamCapture::new();
let base = std::env::temp_dir().join(format!("alkgit-real-git-thin-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let door_thin = common::spawn_daemon_door(
&harness,
DoorOptions {
capture: Some(capture.clone()),
..DoorOptions::default()
},
)
.await;
let client = init_client(&base, "origin")
.await
.expect("the client inits");
let tip_c1 = commit_fixture_file(&client, "", "the first push's commit")
.await
.expect("the first commit lands");
let url = door_url("git", door.addr, &harness.record.repo_id);
real_push(&client, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let url_thin = door_url("git", door_thin.addr, &harness.record.repo_id);
let tip_c2 = commit_fixture_file(
&client,
"this second push appends a small marker line to the fixture file\n",
"the thin push's commit",
)
.await
.expect("the second commit lands");
let introduced = introduced_oids(&client, &tip_c2, &tip_c1)
.await
.expect("the introduced set resolves");
assert!(
!introduced.is_empty(),
"the thin push has something to introduce"
);
let old_blob = rev_parse(&client, &format!("{tip_c1}:a.txt"))
.await
.expect("the base blob resolves");
let old_blob_oid =
gix_hash::ObjectId::from_hex(old_blob.as_bytes()).expect("the base blob oid parses");
assert!(
harness
.assembly
.front()
.stat(&old_blob_oid)
.await
.expect("the front stats")
.is_some(),
"the thin base exists server-side before the push"
);
let pack = {
real_push(&client, &url_thin, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
extract_and_walk_client_pack(&capture.to_server())
.expect("the thin push's pack rode the wire")
};
assert_eq!(
pack.object_count as usize, pack.entry_count,
"the captured pack's entries all parsed"
);
let thin_bases: Vec<ObjectId> = pack
.deltas
.iter()
.filter(|(_, kind)| *kind == common::DeltaKind::ThinBase)
.map(|(base, _)| *base)
.collect();
assert!(
!thin_bases.is_empty(),
"the real client's pack is thin: it carries delta entries whose bases are \
not among the pack's own objects (push-captures.md §Thin pack)"
);
for base in &thin_bases {
assert!(
harness
.assembly
.front()
.stat(base)
.await
.expect("the front stats")
.is_some(),
"every thin base resolves from the substrate Find"
);
assert!(
!introduced.contains(base),
"the thin bases are the ancestry the client excluded — never re-sent"
);
}
assert!(
pack.full_oids.iter().all(|oid| introduced.contains(oid)),
"the pack's full entries are exactly the introduced objects"
);
let tips = listed_tips(&harness).await;
assert_eq!(
tips.iter()
.find(|(name, _)| name == "refs/heads/main")
.and_then(|(_, oid)| *oid),
Some(gix_hash::ObjectId::from_hex(tip_c2.as_bytes()).expect("the tip oid parses")),
"the thin push's CAS applied the pushed main"
);
let verify_dir = base.join("verify");
real_clone(&url, &verify_dir)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let content =
std::fs::read_to_string(verify_dir.join("a.txt")).expect("the fresh clone has the file");
assert_eq!(
content,
file_content("this second push appends a small marker line to the fixture file\n"),
"the fresh clone reads the exact pushed content — the thin base materialized server-side"
);
fsck_strict(&verify_dir)
.await
.expect("the run rode a resolved git binary")
.expect_ok();
door.accept.abort();
door_thin.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_real_packless_delete_push_takes_the_bounded_peek_and_the_sweep_reclaims_the_released_refs_objects(
) {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("packless-delete", DEFAULT_REPO_ID).await;
let base = std::env::temp_dir().join(format!("alkgit-real-git-delete-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door_setup = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let delete_capture = common::StreamCapture::new();
let door_delete = common::spawn_daemon_door(
&harness,
DoorOptions {
capture: Some(delete_capture.clone()),
..DoorOptions::default()
},
)
.await;
let client = init_client(&base, "origin")
.await
.expect("the client inits");
let url_setup = door_url("git", door_setup.addr, &harness.record.repo_id);
let url_delete = door_url("git", door_delete.addr, &harness.record.repo_id);
let tip_main = commit_fixture_file(&client, "", "main's commit")
.await
.expect("main's commit lands");
real_push(&client, &url_setup, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
git(Some(client.path()), &["checkout", "-q", "-b", "old"])
.await
.expect("the run rides a resolved git binary")
.expect_ok();
let tip_old = commit_fixture_file(&client, "the old branch's own content\n", "old's commit")
.await
.expect("old's commit lands");
let old_blob = rev_parse(&client, "HEAD:a.txt")
.await
.expect("old's blob resolves");
real_push(&client, &url_setup, Vec::new(), &["old"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
git(Some(client.path()), &["checkout", "-q", "main"])
.await
.expect("the run rides a resolved git binary")
.expect_ok();
let delete = real_push(&client, &url_delete, Vec::new(), &[":refs/heads/old"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert!(
delete.stderr().contains("[deleted]"),
"the real client reported the deletion: {}",
delete.stderr()
);
assert!(
extract_and_walk_client_pack(&delete_capture.to_server()).is_none(),
"the delete-only push sent no pack bytes — the wire shape the \
bounded peek classifies (captured live: the POC's packless fact)"
);
let report = sideband_report_lines(&delete_capture.to_client());
assert_eq!(
report,
vec!["unpack ok".to_owned(), "ok refs/heads/old".to_owned()],
"the packless push's report frames per ADR-013 §8"
);
assert_unborn_head_row_persists(&harness).await;
let tips = listed_tips(&harness).await;
assert!(
tips.iter().all(|(name, _)| name != "refs/heads/old"),
"the delete's CAS removed the ref row"
);
assert_eq!(
tips.iter()
.find(|(name, _)| name == "refs/heads/main")
.and_then(|(_, oid)| *oid),
Some(gix_hash::ObjectId::from_hex(tip_main.as_bytes()).expect("the tip oid parses")),
"main held"
);
assert_eq!(
harness
.assembly
.reactor()
.registry()
.watermark()
.await
.expect("the watermark reads"),
4,
"the deletion ride's the marker legs like a pushed pack's commit \
(1 = the put, 2/3 = the pushes, 4 = the delete)"
);
let report = harness
.assembly
.blobs()
.sweep()
.await
.expect("the sweep runs");
assert!(report.aborted.is_none());
assert!(
report.deleted >= 1,
"the deleted ref's objects lost their coverage — the sweep reclaimed them"
);
let old_tip_oid = gix_hash::ObjectId::from_hex(tip_old.as_bytes()).expect("the tip oid parses");
let old_blob_oid =
gix_hash::ObjectId::from_hex(old_blob.as_bytes()).expect("the blob oid parses");
for oid in [old_tip_oid, old_blob_oid] {
assert!(
harness
.assembly
.front()
.stat(&oid)
.await
.expect("the front stats")
.is_none(),
"the released refs' object reclaimed"
);
}
let main_commit = rev_parse(&client, "main")
.await
.expect("main's tip resolves");
let main_oid = gix_hash::ObjectId::from_hex(main_commit.as_bytes()).expect("the oid parses");
assert!(
harness
.assembly
.front()
.stat(&main_oid)
.await
.expect("the front stats")
.is_some(),
"main's objects remain covered"
);
assert_work_area_clean(&harness).await;
door_setup.accept.abort();
door_delete.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_real_atomic_push_clashed_at_the_cas_rolls_back_whole_and_reports_the_capture_pair() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("atomic-clash", "owner/atomic").await;
let base = std::env::temp_dir().join(format!("alkgit-real-git-atomic-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let url = door_url("git", door.addr, &harness.record.repo_id);
let seed = init_client(&base, "seed").await.expect("the seed inits");
let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
.await
.expect("the seed commit lands");
real_push(&seed, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let clash_dir = base.join("clash");
real_clone(&url, &clash_dir)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let clash = ClientDir(clash_dir);
let tip_a = commit_fixture_file(&clash, "the clashing client's first commit\n", "clash-a")
.await
.expect("the clash client's commits land");
let tip_b = commit_fixture_file(&clash, "the clashing client's second commit\n", "clash-b")
.await
.expect("the clash client's commits land");
git(Some(clash.path()), &["branch", "feature", "HEAD^"])
.await
.expect("the run rides a resolved git binary")
.expect_ok();
let winner_dir = base.join("winner");
real_clone(&url, &winner_dir)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let winner = ClientDir(winner_dir);
let tip_z = commit_fixture_file(&winner, "the winner's commit\n", "winner")
.await
.expect("the winner's commit lands");
let gate = GateRendezvous::new();
let clash_capture = common::StreamCapture::new();
let door_gated = common::spawn_daemon_door(
&harness,
DoorOptions {
capture: Some(clash_capture.clone()),
gate: Some(gate.clone()),
},
)
.await;
let url_gated = door_url("git", door_gated.addr, &harness.record.repo_id);
let clashing = tokio::spawn({
let dir = clash.path().to_path_buf();
async move {
let mut all = vec!["push", "--atomic"];
all.push(&url_gated);
all.push("main");
all.push("feature");
git_run(Some(&dir), &all).await
}
});
tokio::time::timeout(Duration::from_secs(30), gate.held.notified())
.await
.expect("the clashing client's request reached the hold inside the deadline");
real_push(&winner, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert_eq!(
listed_tips(&harness)
.await
.iter()
.find(|(name, _)| name == "refs/heads/main")
.and_then(|(_, oid)| *oid),
Some(gix_hash::ObjectId::from_hex(tip_z.as_bytes()).expect("the winner's tip parses")),
"the plain client's main advance committed first"
);
gate.release();
let clashing = clashing
.await
.expect("the clashing client's task joins")
.expect("the row rides a resolved git binary");
assert_ne!(
clashing.code, 0,
"the clashing push was denied by the CAS: {}{}",
clashing.stdout, clashing.stderr
);
assert!(
clashing.stderr.contains("stale info"),
"the stale ref reports the server-chosen reason: {}",
clashing.stderr
);
assert!(
clashing.stderr.contains("atomic push failure"),
"the would-be-successful ref reports the atomic-abort reason: {}",
clashing.stderr
);
let report = sideband_report_lines(&clash_capture.to_client());
assert_eq!(
report
.iter()
.filter(|line| **line != "unpack ok")
.map(String::as_str)
.collect::<std::collections::BTreeSet<&str>>(),
[
"ng refs/heads/main stale info",
"ng refs/heads/feature atomic push failure"
]
.into_iter()
.collect(),
"the report carries the capture pair verbatim: {:?}",
report
);
let tips = listed_tips(&harness).await;
assert!(
tips.iter().all(|(name, _)| name != "refs/heads/feature"),
"the atomic rollback applied nothing — feature's create never landed"
);
assert_eq!(
harness
.assembly
.reactor()
.registry()
.watermark()
.await
.expect("the watermark reads"),
3,
"the aborted CAS rides no legs: 1 = the put, 2/3 = the two committed \
pushes' markers"
);
let introduced = introduced_oids(&clash, &tip_b, &tip_c1)
.await
.expect("the clashing client's introduced set resolves");
let report = harness
.assembly
.blobs()
.sweep()
.await
.expect("the sweep runs");
assert!(report.aborted.is_none());
assert!(
report.deleted >= 1,
"the dropped guards' objects are orphans — the sweep reclaims them"
);
let tip_a_oid =
gix_hash::ObjectId::from_hex(tip_a.as_bytes()).expect("the clash tip oid parses");
for oid in introduced.iter().chain(std::iter::once(&tip_a_oid)) {
assert!(
harness
.assembly
.front()
.stat(oid)
.await
.expect("the front stats")
.is_none(),
"the rollback's pack objects recovered GC-decreed: {oid}"
);
}
let winner_oid =
gix_hash::ObjectId::from_hex(tip_z.as_bytes()).expect("the winner's tip parses");
assert!(
harness
.assembly
.front()
.stat(&winner_oid)
.await
.expect("the front stats")
.is_some(),
"the winner's objects stay covered"
);
assert_work_area_clean(&harness).await;
door.accept.abort();
door_gated.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_real_git_clone_and_fetch_complete_over_the_duplex_door_with_the_pushed_commit_checked_out(
) {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("clone-fetch-duplex", DEFAULT_REPO_ID).await;
let base = std::env::temp_dir().join(format!(
"alkgit-real-git-clone-duplex-{}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let url = door_url("git", door.addr, &harness.record.repo_id);
let seed = init_client(&base, "seed").await.expect("the seed inits");
let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
.await
.expect("the seed commit lands");
real_push(&seed, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let dst = base.join("clone");
real_clone(&url, &dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;
let tip_c2 = commit_fixture_file(&seed, "the fetch's new tip\n", "the advanced commit")
.await
.expect("the advanced commit lands");
real_push(&seed, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
real_fetch(&dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
.await
.expect("the tracking ref resolves");
assert_eq!(
fetched, tip_c2,
"the real fetch negotiated and advanced the tracking ref"
);
assert_work_area_clean(&harness).await;
door.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn the_real_clone_and_fetch_sessions_complete_over_the_stateless_http_door() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem("clone-fetch-http", DEFAULT_REPO_ID).await;
let base =
std::env::temp_dir().join(format!("alkgit-real-git-clone-http-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let door_http = common::spawn_http_door(&harness).await;
let url_git = door_url("git", door.addr, &harness.record.repo_id);
let url_http = door_url("http", door_http.addr, &harness.record.repo_id);
let seed = init_client(&base, "seed").await.expect("the seed inits");
let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
.await
.expect("the seed commit lands");
real_push(&seed, &url_git, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let dst = base.join("clone");
real_clone(&url_http, &dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;
let tip_c2 = commit_fixture_file(&seed, "the http fetch's new tip\n", "the advanced commit")
.await
.expect("the advanced commit lands");
real_push(&seed, &url_git, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
real_fetch(&dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
.await
.expect("the tracking ref resolves");
assert_eq!(
fetched, tip_c2,
"the real fetch advanced the tracking ref over the stateless door"
);
assert_work_area_clean(&harness).await;
door.accept.abort();
door_http.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[cfg(feature = "sqlite")]
#[tokio::test(flavor = "multi_thread")]
async fn the_sqlite_assembly_serves_a_real_push_and_a_real_clone_through_store() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::sqlite(
"sqlite-through-store",
DEFAULT_REPO_ID,
PackServing::ThroughStore,
)
.await;
let base = std::env::temp_dir().join(format!("alkgit-real-git-sqlite-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let door_http = common::spawn_http_door(&harness).await;
let url_git = door_url("git", door.addr, &harness.record.repo_id);
let url_http = door_url("http", door_http.addr, &harness.record.repo_id);
let seed = init_client(&base, "seed").await.expect("the seed inits");
let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
.await
.expect("the seed commit lands");
real_push(&seed, &url_git, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert_unborn_head_row_persists(&harness).await;
let tips = listed_tips(&harness).await;
assert_eq!(
tips.iter()
.find(|(name, _)| name == "refs/heads/main")
.and_then(|(_, oid)| *oid),
Some(gix_hash::ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses")),
"the refs CAS applied over the co-tenant rows"
);
let dst = base.join("clone");
real_clone(&url_http, &dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;
let tip_c2 = commit_fixture_file(
&seed,
"the through-store fetch's new tip\n",
"the advanced commit",
)
.await
.expect("the advanced commit lands");
real_push(&seed, &url_git, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
real_fetch(&dst)
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
.await
.expect("the tracking ref resolves");
assert_eq!(
fetched, tip_c2,
"the http fetch advanced through the through-store serving shape"
);
assert_work_area_clean(&harness).await;
door.accept.abort();
door_http.accept.abort();
std::fs::remove_dir_all(&base).ok();
drop(harness);
}
#[tokio::test(flavor = "multi_thread")]
async fn the_maintenance_chain_backstops_a_real_pushs_composed_legs() {
if common::git_command().is_none() {
eprintln!("skipping row: no `git` binary on PATH");
return;
}
let harness = Harness::mem_tuned(
"chain-backstop",
DEFAULT_REPO_ID,
MaintenanceOpts {
sweep_spec: "@every 1s".to_owned(),
reconcile_tick: Duration::from_secs(10),
worker_poll: Duration::from_millis(20),
..MaintenanceOpts::default()
},
)
.await;
let base = std::env::temp_dir().join(format!("alkgit-real-git-chain-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&base);
std::fs::create_dir_all(&base).expect("the row's base creates");
let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
let url = door_url("git", door.addr, &harness.record.repo_id);
let seed = init_client(&base, "seed").await.expect("the seed inits");
let _tip = commit_fixture_file(&seed, "", "the chain row's commit")
.await
.expect("the seed commit lands");
real_push(&seed, &url, Vec::new(), &["main"])
.await
.expect("the row rides a resolved git binary")
.expect_ok();
let watermark = harness
.assembly
.reactor()
.registry()
.watermark()
.await
.expect("the watermark reads");
let applied = harness.assembly.applied_watermark().clone();
let missed = harness
.assembly
.reactor()
.reconcile(&applied)
.await
.expect("the reconcile pass closes the record row");
assert!(
!missed.is_empty(),
"the reconcile caught the record row up (the markers never ride rows_since)"
);
assert_eq!(
applied.get(),
watermark,
"the applied watermark anchored at the durable watermark"
);
let seen: Arc<Mutex<Vec<WorkerEvent>>> = Arc::new(Mutex::new(Vec::new()));
let mut sweep_verdict: Option<alkgit_substrate::SweepVerdict> = None;
let stop = alkstore::StopToken::new();
let chain_run = harness.assembly.chain().run("e2e-worker", stop.clone(), {
let seen = seen.clone();
move |event| seen.lock().push(event)
});
tokio::pin!(chain_run);
let started = tokio::time::Instant::now();
let mut result: Option<alkstore::Result<()>> = None;
while result.is_none() {
let housekeeping = seen
.lock()
.iter()
.any(|event| matches!(event, WorkerEvent::PackHousekeeping(_)));
sweep_verdict = seen.lock().iter().find_map(|event| match event {
WorkerEvent::Sweep(verdict) => Some(verdict.clone()),
_ => None,
});
if (housekeeping && sweep_verdict.is_some() || started.elapsed() > Duration::from_secs(10))
&& !stop.is_cancelled()
{
stop.cancel();
}
tokio::select! {
outcome = &mut chain_run => result = Some(outcome),
_ = tokio::time::sleep(Duration::from_millis(50)) => {}
}
}
let done = result.expect("the chain's run reached its end");
assert!(
done.is_ok(),
"the chain ran clean under the cancelled stop token: {done:?}"
);
assert!(
seen.lock()
.iter()
.any(|event| matches!(event, WorkerEvent::PackHousekeeping(_))),
"the real push's enqueued housekeeping job drained: {:?}",
seen.lock()
);
let verdict = sweep_verdict.expect("the chain's scheduled sweep ran");
match verdict {
alkgit_substrate::SweepVerdict::Report(report) => {
assert!(report.aborted.is_none(), "the sweep ran: {report:?}");
assert_eq!(
report.deleted, 0,
"the pushed graph is liveness-covered across the chain's sweep"
);
}
other => panic!("the chain's sweep verdict ran: {other:?}"),
}
door.accept.abort();
std::fs::remove_dir_all(&base).ok();
}