glm-5.3-flash
1ac4bfb5c4
- Multi-have no-done round acks each recognized have in request order (unrecognized haves silent, NAK absent) on both serving substrates: duplex (tests/duplex_serving.rs) + stateless (tests/stateless_serving.rs) - The follow-up done round's boundary set is the recognized subset: both commons bound the walk; the missing and the exists-but-not-a- commit haves subtract nothing - Disposition: validated, no retraction; tracker §2 records the outcome and books the substrate Find re-run for track B wave 7 - Existing NAK-only / seam rows unmodified and green Verified: cargo test, -p alkgit, --all-features, both clippies (-D warnings), fmt, doc, taskgraph validate
1599 lines
53 KiB
Rust
1599 lines
53 KiB
Rust
//! The wave-2 duplex serving entry (core/w2-duplex-serving): the
|
||
//! advertisement-once + command-loop session the native transport
|
||
//! drives — the marker-gated constructor, the wall clock, the error
|
||
//! mapping, and the resolve→authorize front's generic denial — over
|
||
//! in-memory duplex pairs with the wave-2 backend doubles. Extended by
|
||
//! core/w3-duplex-serving: the `Service::ReceivePack` branch serves the
|
||
//! real push session (the V0 advertisement, one request per stream, the
|
||
//! duplex tail discipline, the rows below the push block).
|
||
//!
|
||
//! Rows are names + `Contract stamp:` versions, per
|
||
//! `tests/common` §The row/stamp discipline.
|
||
|
||
mod common;
|
||
|
||
use std::collections::{BTreeMap, BTreeSet};
|
||
use std::sync::Arc;
|
||
use std::time::Duration;
|
||
|
||
use gix_hash::ObjectId;
|
||
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
|
||
use tokio::sync::Semaphore;
|
||
use tokio::task::JoinHandle;
|
||
|
||
use alkcall::core::Identity;
|
||
use alkgit::{
|
||
serve_duplex, serve_duplex_resolved, AccessAction, AuthorizedRepo, GitPackGen, GitPackIngest,
|
||
GitRefs, GitRegistry, GitRegistryStore, Limits, PreparedPush, PushOptions, RefUpdate,
|
||
RepoRecord, Service, StorageError, SubstrateError, Visibility,
|
||
};
|
||
use common::{
|
||
blob_fixture, chain_pack, commit_fixture, create_command, decode_pack, delete_command,
|
||
push_commands, sideband_report_lines, tag_fixture, tree_fixture, MemObjects, MemPackGen,
|
||
MemPackIngest, MemRefs, MemRegistry,
|
||
};
|
||
|
||
const FLUSH: &[u8] = b"0000";
|
||
const DELIMITER: &[u8] = b"0001";
|
||
|
||
fn repo(id: &str) -> RepoRecord {
|
||
RepoRecord {
|
||
repo_id: id.to_string(),
|
||
visibility: Visibility::Private,
|
||
grants: BTreeMap::new(),
|
||
}
|
||
}
|
||
|
||
fn identity(id: &str) -> Identity {
|
||
Identity {
|
||
id: id.to_string(),
|
||
scopes: Vec::new(),
|
||
resources: std::collections::HashMap::new(),
|
||
}
|
||
}
|
||
|
||
/// The marker the adapter hands the entries — constructible only
|
||
/// through the checked path (`AuthorizedRepo::authorize`); here the
|
||
/// holder identity is granted (at least) the action.
|
||
fn marker(record: RepoRecord, identity: &Identity, action: AccessAction) -> AuthorizedRepo {
|
||
AuthorizedRepo::authorize(record, Some(identity), action).expect("the adapter's check passed")
|
||
}
|
||
|
||
/// Encodes one pkt-line, client-side framing style.
|
||
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
|
||
}
|
||
|
||
/// A V2 command request, client-side framing: the `command=` line, the
|
||
/// `0001` delimiter, the args, the terminating flush.
|
||
fn command_request(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
|
||
[
|
||
pkt(command),
|
||
pre.iter()
|
||
.flat_map(|l| pkt(l.as_bytes()))
|
||
.collect::<Vec<u8>>(),
|
||
DELIMITER.to_vec(),
|
||
args.iter()
|
||
.flat_map(|l| pkt(l.as_bytes()))
|
||
.collect::<Vec<u8>>(),
|
||
FLUSH.to_vec(),
|
||
]
|
||
.concat()
|
||
}
|
||
|
||
/// Reads pkt-lines from the client end until (and including) a flush —
|
||
/// one advertisement or one command response.
|
||
async fn read_pkts(reader: &mut tokio::io::DuplexStream) -> Vec<Vec<u8>> {
|
||
let mut packets = Vec::new();
|
||
loop {
|
||
let mut header = [0u8; 4];
|
||
reader
|
||
.read_exact(&mut header)
|
||
.await
|
||
.expect("pkt-line header");
|
||
let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
|
||
.expect("hex length");
|
||
if len < 4 {
|
||
packets.push(Vec::new());
|
||
break;
|
||
}
|
||
let mut payload = vec![0u8; len - 4];
|
||
reader
|
||
.read_exact(&mut payload)
|
||
.await
|
||
.expect("pkt-line payload");
|
||
packets.push(payload);
|
||
}
|
||
packets
|
||
}
|
||
|
||
/// Reads exactly one pkt-line — the session-end error band's shape (an
|
||
/// error band is written without a trailing flush; the session closes
|
||
/// right after it).
|
||
async fn read_band(reader: &mut tokio::io::DuplexStream) -> Vec<u8> {
|
||
let mut header = [0u8; 4];
|
||
reader.read_exact(&mut header).await.expect("band header");
|
||
let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
|
||
.expect("hex length");
|
||
let mut payload = vec![0u8; len - 4];
|
||
reader.read_exact(&mut payload).await.expect("band payload");
|
||
payload
|
||
}
|
||
|
||
fn text(payload: &[u8]) -> String {
|
||
String::from_utf8_lossy(payload)
|
||
.trim_end_matches('\n')
|
||
.to_owned()
|
||
}
|
||
|
||
fn text_lines(pkts: &[Vec<u8>]) -> Vec<String> {
|
||
pkts.iter().map(|p| text(p)).collect()
|
||
}
|
||
|
||
/// The advertisement's expected line sequence (feature-aware — the
|
||
/// sha256 rider rides the feature per OQ-05's passthrough policy).
|
||
fn advertisement_lines() -> Vec<String> {
|
||
#[cfg(not(feature = "sha256"))]
|
||
let lines: [&str; 4] = [
|
||
"version 2",
|
||
"ls-refs=unborn",
|
||
"fetch=wait-for-done",
|
||
"object-format=sha1",
|
||
];
|
||
#[cfg(feature = "sha256")]
|
||
let lines: [&str; 5] = [
|
||
"version 2",
|
||
"ls-refs=unborn",
|
||
"fetch=wait-for-done",
|
||
"object-format=sha1",
|
||
"object-format=sha256",
|
||
];
|
||
lines.iter().map(|l| l.to_string()).collect()
|
||
}
|
||
|
||
/// The advertisement's expected packet sequence: the lines plus the
|
||
/// terminating flush (read_pkts's trailing empty entry).
|
||
fn advertisement_pkts() -> Vec<String> {
|
||
let mut lines = advertisement_lines();
|
||
lines.push(String::new());
|
||
lines
|
||
}
|
||
|
||
/// The adapter's per-session assembly: the markers (read- and
|
||
/// write-granted identity), the backend handles keyed on the resolved
|
||
/// repo, the wave-2 doubles + the wave-3 ingest double (one object
|
||
/// namespace shared by packgen and ingest).
|
||
struct Fixture {
|
||
identity: Identity,
|
||
read_marker: AuthorizedRepo,
|
||
write_marker: AuthorizedRepo,
|
||
refs: Arc<MemRefs>,
|
||
gen: Arc<MemPackGen>,
|
||
ingest: Arc<MemPackIngest>,
|
||
objects: MemObjects,
|
||
}
|
||
|
||
fn fixture() -> Fixture {
|
||
let record = repo("owner/doubled");
|
||
let mut read_record = record.clone();
|
||
read_record
|
||
.grants
|
||
.insert("u-7731".to_string(), vec![AccessAction::Read]);
|
||
let mut write_record = record;
|
||
write_record
|
||
.grants
|
||
.insert("u-7731".to_string(), vec![AccessAction::Write]);
|
||
let holder = identity("u-7731");
|
||
let objects = MemObjects::default();
|
||
Fixture {
|
||
identity: holder.clone(),
|
||
read_marker: marker(read_record, &holder, AccessAction::Read),
|
||
write_marker: marker(write_record, &holder, AccessAction::Write),
|
||
refs: Arc::new(MemRefs::default()),
|
||
gen: Arc::new(MemPackGen {
|
||
objects: objects.clone(),
|
||
}),
|
||
ingest: Arc::new(MemPackIngest {
|
||
objects: objects.clone(),
|
||
}),
|
||
objects,
|
||
}
|
||
}
|
||
|
||
/// Seeds the two-commit fixture: `old` (a, the common ancestor) →
|
||
/// `main` (b) plus an annotated tag, and refs for the tips; returns the
|
||
/// object ids the conversation rows assert against.
|
||
struct TwoCommits {
|
||
commit_a: ObjectId,
|
||
commit_b: ObjectId,
|
||
tree_b: ObjectId,
|
||
blob_b: ObjectId,
|
||
}
|
||
|
||
async fn seed_two_commits(fx: &Fixture) -> TwoCommits {
|
||
let doubled = repo("owner/doubled");
|
||
let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
|
||
let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
|
||
let (tree_a, tree_a_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
|
||
let (tree_b, tree_b_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "b.txt", blob_b)]);
|
||
let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
|
||
let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
|
||
let (tag, tag_bytes) = tag_fixture(commit_b, gix_object::Kind::Commit, "v1");
|
||
for (id, bytes) in [
|
||
(blob_a, blob_a_bytes),
|
||
(blob_b, blob_b_bytes),
|
||
(tree_a, tree_a_bytes),
|
||
(tree_b, tree_b_bytes),
|
||
(commit_a, commit_a_bytes),
|
||
(commit_b, commit_b_bytes),
|
||
(tag, tag_bytes),
|
||
] {
|
||
fx.objects.put(&doubled, id, bytes);
|
||
}
|
||
fx.refs
|
||
.apply_updates(
|
||
&doubled,
|
||
vec![
|
||
RefUpdate {
|
||
name: "refs/heads/old".to_string(),
|
||
expected: None,
|
||
new: Some(commit_a),
|
||
},
|
||
RefUpdate {
|
||
name: "refs/heads/main".to_string(),
|
||
expected: None,
|
||
new: Some(commit_b),
|
||
},
|
||
RefUpdate {
|
||
name: "refs/tags/v1".to_string(),
|
||
expected: None,
|
||
new: Some(tag),
|
||
},
|
||
],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
TwoCommits {
|
||
commit_a,
|
||
commit_b,
|
||
tree_b,
|
||
blob_b,
|
||
}
|
||
}
|
||
|
||
/// Serves the marker-gated entry against a fresh in-memory duplex pair;
|
||
/// the returned peer end is the client's, the join handle the session's
|
||
/// result.
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn serve(
|
||
marker: &AuthorizedRepo,
|
||
refs: Arc<dyn GitRefs>,
|
||
gen: Arc<dyn GitPackGen>,
|
||
ingest: Arc<dyn GitPackIngest>,
|
||
push_options_negotiated: bool,
|
||
limits: Limits,
|
||
service: Service,
|
||
) -> (
|
||
tokio::io::DuplexStream,
|
||
JoinHandle<Result<(), SubstrateError>>,
|
||
) {
|
||
let (peer, stream) = duplex(1 << 16);
|
||
let joined = tokio::spawn({
|
||
let marker = marker.clone();
|
||
async move {
|
||
serve_duplex(
|
||
&marker,
|
||
stream,
|
||
limits,
|
||
service,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
push_options_negotiated,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
)
|
||
.await
|
||
}
|
||
});
|
||
(peer, joined)
|
||
}
|
||
|
||
/// A read-session fixture serve: the advertisement precedes every
|
||
/// conversation row.
|
||
async fn serve_fixture(
|
||
fx: &Fixture,
|
||
limits: Limits,
|
||
) -> (
|
||
tokio::io::DuplexStream,
|
||
JoinHandle<Result<(), SubstrateError>>,
|
||
) {
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let gen: Arc<dyn GitPackGen> = fx.gen.clone();
|
||
let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
|
||
serve(
|
||
&fx.read_marker,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
false,
|
||
limits,
|
||
Service::UploadPack,
|
||
)
|
||
.await
|
||
}
|
||
|
||
/// A push-session fixture serve: the write marker, the ingest handle,
|
||
/// the `push-options` negotiated-gate flag alongside.
|
||
async fn serve_push(
|
||
fx: &Fixture,
|
||
ingest: Arc<dyn GitPackIngest>,
|
||
push_options_negotiated: bool,
|
||
limits: Limits,
|
||
) -> (
|
||
tokio::io::DuplexStream,
|
||
JoinHandle<Result<(), SubstrateError>>,
|
||
) {
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let gen: Arc<dyn GitPackGen> = fx.gen.clone();
|
||
serve(
|
||
&fx.write_marker,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
push_options_negotiated,
|
||
limits,
|
||
Service::ReceivePack,
|
||
)
|
||
.await
|
||
}
|
||
|
||
/// The advertisement is emitted exactly once per session at session
|
||
/// start, commands dispatch per request, and the reset discipline holds
|
||
/// between commands — three commands in one session (two ls-refs, one
|
||
/// fetch's no-done round) are all heard and answered, never swallowed
|
||
/// and never re-advertised.
|
||
///
|
||
/// Contract stamp: ADR-003 (advertisement-once per duplex session —
|
||
/// real git hangs on re-advertisement); ADR-005 §1 (the command loop
|
||
/// over the request reader, the reset discipline); core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn advertisement_once_with_three_commands_dispatching_over_the_reset_discipline() {
|
||
let fx = fixture();
|
||
let commits = seed_two_commits(&fx).await;
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
let mut conversation = text_lines(&advertisement);
|
||
assert_eq!(
|
||
conversation.as_slice(),
|
||
advertisement_pkts().as_slice(),
|
||
"the advertisement, emitted once at session start"
|
||
);
|
||
|
||
for round in 0..2 {
|
||
peer.write_all(&command_request(
|
||
b"command=ls-refs\n",
|
||
&["agent=git/2.43"],
|
||
&["ref-prefix refs/heads/"],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let response = read_pkts(&mut peer).await;
|
||
let lines = text_lines(&response);
|
||
conversation.extend(lines.iter().cloned());
|
||
assert_eq!(
|
||
lines.last(),
|
||
Some(&String::new()),
|
||
"round {round}'s response ends at flush"
|
||
);
|
||
assert_eq!(
|
||
lines[..lines.len() - 1]
|
||
.iter()
|
||
.map(|line| line
|
||
.split_once(' ')
|
||
.map_or(line.to_owned(), |(_, name)| name.to_owned()))
|
||
.collect::<Vec<_>>(),
|
||
vec!["refs/heads/main", "refs/heads/old"],
|
||
"round {round}'s ref-prefix-filtered ls-refs response"
|
||
);
|
||
}
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1"],
|
||
&[
|
||
&format!("want {}", commits.commit_b),
|
||
&format!("have {}", commits.commit_a),
|
||
],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let response = read_pkts(&mut peer).await;
|
||
let lines = text_lines(&response);
|
||
conversation.extend(lines.clone());
|
||
assert_eq!(lines[0], "acknowledgments", "the no-done round's shape");
|
||
assert_eq!(lines[1], format!("ACK {}", commits.commit_a));
|
||
assert_eq!(lines.last(), Some(&String::new()));
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close ends the session quietly");
|
||
|
||
assert_eq!(
|
||
conversation
|
||
.iter()
|
||
.filter(|line| line == &"version 2")
|
||
.count(),
|
||
1,
|
||
"the advertisement-once rule across the whole conversation"
|
||
);
|
||
}
|
||
|
||
/// A second advertisement between commands never occurs: two ls-refs
|
||
/// rounds ride plain ref listings — no capability dump rides between
|
||
/// them.
|
||
///
|
||
/// Contract stamp: ADR-003 (emitted once per duplex session, never
|
||
/// between commands — a re-advertisement hangs real git);
|
||
/// core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_second_advertisement_never_occurs_between_commands() {
|
||
let fx = fixture();
|
||
seed_two_commits(&fx).await;
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
|
||
.await
|
||
.unwrap();
|
||
let first = text_lines(&read_pkts(&mut peer).await);
|
||
peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
|
||
.await
|
||
.unwrap();
|
||
let second = text_lines(&read_pkts(&mut peer).await);
|
||
|
||
assert_eq!(first, second, "both rounds list the same refs");
|
||
let listing: Vec<String> = first
|
||
.iter()
|
||
.filter(|line| !line.is_empty())
|
||
.map(|line| {
|
||
line.split_once(' ')
|
||
.map_or(line.to_owned(), |(_, name)| name.to_owned())
|
||
})
|
||
.collect();
|
||
assert_eq!(
|
||
listing,
|
||
vec!["refs/heads/main", "refs/heads/old", "refs/tags/v1"],
|
||
"the unfiltered ls-refs listing"
|
||
);
|
||
assert!(
|
||
first
|
||
.iter()
|
||
.chain(second.iter())
|
||
.all(|line| line != "version 2"),
|
||
"no capability dump rode between the commands"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close ends the session quietly");
|
||
}
|
||
|
||
/// A full fetch conversation — advertisement → ls-refs → the fetch done
|
||
/// round with a real pack from the doubles — passes end-to-end over an
|
||
/// in-memory duplex pair; the client's clean close ends the session
|
||
/// quietly.
|
||
///
|
||
/// Contract stamp: ADR-002 (the duplex session entry, the running V2
|
||
/// fetch session); ADR-005 §1 (the command loop); ADR-014 §2 (the done
|
||
/// round's packfile section); core/w2-duplex-serving (the full
|
||
/// conversation over the entry).
|
||
#[tokio::test]
|
||
async fn a_full_fetch_conversation_passes_end_to_end() {
|
||
let fx = fixture();
|
||
let commits = seed_two_commits(&fx).await;
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=ls-refs\n",
|
||
&["symrefs", "unborn"],
|
||
&["ref-prefix refs/heads/"],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let ref_lines = text_lines(&read_pkts(&mut peer).await);
|
||
assert_eq!(
|
||
ref_lines[0],
|
||
format!("{} refs/heads/main", commits.commit_b)
|
||
);
|
||
assert_eq!(ref_lines[1], format!("{} refs/heads/old", commits.commit_a));
|
||
assert_eq!(ref_lines.last(), Some(&String::new()));
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1", "agent=git/2.43"],
|
||
&[
|
||
"thin-pack",
|
||
"no-progress",
|
||
&format!("want {}", commits.commit_b),
|
||
&format!("have {}", commits.commit_a),
|
||
"done",
|
||
],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let fetch_response = read_pkts(&mut peer).await;
|
||
assert_eq!(text(&fetch_response[0]), "packfile");
|
||
let band_chunks: Vec<&[u8]> = fetch_response[1..fetch_response.len() - 1]
|
||
.iter()
|
||
.map(|p| &p[1..])
|
||
.collect();
|
||
let pack: Vec<u8> = band_chunks.concat();
|
||
let decoded = decode_pack(&pack);
|
||
let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
|
||
assert_eq!(
|
||
ids,
|
||
BTreeSet::from([commits.commit_b, commits.tree_b, commits.blob_b]),
|
||
"closure(want) − closure(recognized have) over the entry"
|
||
);
|
||
let _ = decoded.trailer;
|
||
assert!(
|
||
fetch_response.last().unwrap().is_empty(),
|
||
"flush-terminated"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close ends the session quietly");
|
||
}
|
||
|
||
/// The OQ-19 rider-2 positive-ack capture over the duplex session
|
||
/// (the machinery-side run): a multi-have negotiation session where the
|
||
/// no-`done` round acks **each** recognized have — per-have `ACK <oid>`
|
||
/// lines in the request's order, only the seam-recognized subset (the
|
||
/// missing and the exists-but-not-a-commit haves stay unacked), `NAK`
|
||
/// absent — and the done round's boundary set is the recognized subset:
|
||
/// the unrecognized haves subtract nothing, both commons bound the walk.
|
||
///
|
||
/// Contract stamp: OQ-19 rider 2 (the positive-ACK capture,
|
||
/// tasks/architecture/oq-19-wire-validations.md §2 — the
|
||
/// machinery-side run; the substrate-backed re-run is track B wave 7);
|
||
/// ADR-014 §1 (ACK per recognized have, NAK only when no ACKs);
|
||
/// ADR-014 §2 as amended by review 001 A-4 (the boundary set is the
|
||
/// recognized subset); ADR-014 §5 (exists + is-commit — the blob have
|
||
/// drives the negative leg); ADR-005 §1 (both rounds ride one session
|
||
/// over the command loop).
|
||
#[tokio::test]
|
||
async fn a_multi_have_session_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
|
||
) {
|
||
let fx = fixture();
|
||
let doubled = repo("owner/doubled");
|
||
let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
|
||
let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
|
||
let (blob_c, blob_c_bytes) = blob_fixture(b"gamma");
|
||
let (tree_a, tree_a_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
|
||
let (tree_b, tree_b_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "b.txt", blob_b)]);
|
||
let (tree_c, tree_c_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "c.txt", blob_c)]);
|
||
let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
|
||
let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
|
||
let (commit_c, commit_c_bytes) = commit_fixture(tree_c, &[commit_b], "gamma");
|
||
for (id, bytes) in [
|
||
(blob_a, blob_a_bytes),
|
||
(blob_b, blob_b_bytes),
|
||
(blob_c, blob_c_bytes),
|
||
(tree_a, tree_a_bytes),
|
||
(tree_b, tree_b_bytes),
|
||
(tree_c, tree_c_bytes),
|
||
(commit_a, commit_a_bytes),
|
||
(commit_b, commit_b_bytes),
|
||
(commit_c, commit_c_bytes),
|
||
] {
|
||
fx.objects.put(&doubled, id, bytes);
|
||
}
|
||
|
||
let missing = ObjectId::from([9u8; 20]);
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1", "agent=git/2.43"],
|
||
&[
|
||
&format!("want {}", commit_c),
|
||
&format!("have {}", commit_a),
|
||
&format!("have {}", missing),
|
||
&format!("have {}", commit_b),
|
||
],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let ack_round = text_lines(&read_pkts(&mut peer).await);
|
||
assert_eq!(
|
||
ack_round,
|
||
vec![
|
||
"acknowledgments".to_owned(),
|
||
format!("ACK {}", commit_a),
|
||
format!("ACK {}", commit_b),
|
||
String::new(),
|
||
],
|
||
"per-have ACKs in request order, unrecognized haves silent, flush-terminated"
|
||
);
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1", "agent=git/2.43"],
|
||
&[
|
||
"thin-pack",
|
||
"no-progress",
|
||
&format!("want {}", commit_c),
|
||
&format!("have {}", commit_a),
|
||
&format!("have {}", missing),
|
||
&format!("have {}", blob_a),
|
||
&format!("have {}", commit_b),
|
||
"done",
|
||
],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let done_round = read_pkts(&mut peer).await;
|
||
assert_eq!(text(&done_round[0]), "packfile");
|
||
let band_chunks: Vec<&[u8]> = done_round[1..done_round.len() - 1]
|
||
.iter()
|
||
.map(|p| &p[1..])
|
||
.collect();
|
||
let pack: Vec<u8> = band_chunks.concat();
|
||
let decoded = decode_pack(&pack);
|
||
let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
|
||
assert_eq!(
|
||
ids,
|
||
BTreeSet::from([commit_c, tree_c, blob_c]),
|
||
"the boundary set is the recognized subset: both commons bounded the \
|
||
walk, the missing and the non-commit have subtracted nothing"
|
||
);
|
||
let _ = decoded.trailer;
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close ends the session quietly");
|
||
}
|
||
|
||
/// An unknown command declines via the error band and the session ends:
|
||
/// the band is the response, the entry returns the typed failure, and
|
||
/// nothing follows.
|
||
///
|
||
/// Contract stamp: ADR-005 §4 (protocol-shaped failures map to the
|
||
/// pkt-line error band before the session ends); ADR-003 (honest
|
||
/// surface — unimplemented features declined); core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn an_unknown_command_declines_via_the_error_band() {
|
||
let fx = fixture();
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
peer.write_all(&command_request(
|
||
b"command=object-info\n",
|
||
&[],
|
||
&["size 0000"],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(
|
||
band.starts_with("ERR unknown command \"object-info\""),
|
||
"the decline is the band: {band}"
|
||
);
|
||
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("the decline is a typed failure");
|
||
match &result {
|
||
SubstrateError::Protocol(message) => {
|
||
assert!(message.contains("unknown command"), "{message}");
|
||
assert!(message.contains("object-info"), "{message}");
|
||
}
|
||
other => panic!("an unknown command is protocol-shaped: {other:?}"),
|
||
}
|
||
|
||
let mut remaining = [0u8; 1];
|
||
let n = peer.read(&mut remaining).await.expect("read after close");
|
||
assert_eq!(n, 0, "nothing follows the decline band");
|
||
}
|
||
|
||
/// A malformed client request (the machinery's protocol failure)
|
||
/// band-maps and ends the session — the entry's error mapping covers the
|
||
/// substrate's failures, not just the command dispatch's.
|
||
///
|
||
/// Contract stamp: ADR-005 §4 (machinery protocol errors band-map);
|
||
/// ADR-005 §1 (the unconditional error-break rule — the session ends,
|
||
/// never resets-and-retries); core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_malformed_request_bands_and_ends_the_session() {
|
||
let fx = fixture();
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
|
||
read_pkts(&mut peer).await; // the advertisement
|
||
|
||
peer.write_all(b"zzzz").await.unwrap(); // a malformed pkt-line
|
||
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(band.starts_with("ERR "), "the failure bands");
|
||
assert!(
|
||
band.contains("malformed pkt-line"),
|
||
"the machinery's protocol failure bands: {band}"
|
||
);
|
||
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("a machinery protocol failure is a typed failure");
|
||
assert!(
|
||
matches!(result, SubstrateError::Protocol(_)),
|
||
"the typed failure returns to the caller: {result:?}"
|
||
);
|
||
}
|
||
|
||
/// The wall-clock budget (ADR-009) ends a session that exceeds it —
|
||
/// here with a shrunken `Limits`: the stalled read breaches the budget,
|
||
/// the entry bands + closes (the duplex mapping), and the typed
|
||
/// server/session failure returns.
|
||
///
|
||
/// Contract stamp: ADR-009 (the session wall clock — enforced by
|
||
/// transport's session loop; breach is the server/session class, error
|
||
/// band + close on duplex); core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn the_wall_clock_budget_ends_a_session_that_exceeds_it() {
|
||
let fx = fixture();
|
||
let limits = Limits {
|
||
session_wall_clock: Duration::from_millis(100),
|
||
..Limits::default()
|
||
};
|
||
|
||
let (mut peer, joined) = serve_fixture(&fx, limits).await;
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
// No command ever arrives; the stalled read breaches the budget.
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(
|
||
band.starts_with("ERR ") && band.contains("session_wall_clock"),
|
||
"the breach bands the session: {band}"
|
||
);
|
||
|
||
let mut remaining = [0u8; 1];
|
||
let n = peer.read(&mut remaining).await.expect("read after close");
|
||
assert_eq!(n, 0, "the breached session closed");
|
||
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("the breach is a typed failure");
|
||
match &result {
|
||
SubstrateError::Protocol(message) => {
|
||
assert!(message.contains("session_wall_clock"), "{message}");
|
||
}
|
||
other => panic!("a wall-clock breach is protocol-shaped: {other:?}"),
|
||
}
|
||
}
|
||
|
||
/// Unknown-repo ≡ unauthorized: the resolve→authorize front's denial is
|
||
/// byte-identical for a `RegistryError::NotFound` resolution and a
|
||
/// failed `authorize` evaluation, and no protocol byte precedes the
|
||
/// denial.
|
||
///
|
||
/// Contract stamp: ADR-007 steps 2–3 (resolve, then authorize, before
|
||
/// any protocol output); ADR-008/N-2 (unknown ≡ unauthorized — no
|
||
/// existence oracle); core/w2-duplex-serving (the front's collapse).
|
||
#[tokio::test]
|
||
async fn unknown_repo_and_unauthorized_denials_are_indistinguishable() {
|
||
let fx = fixture();
|
||
let holder = fx.identity.clone();
|
||
|
||
let unknown_registry: Arc<dyn GitRegistry> = Arc::new(MemRegistry::default());
|
||
|
||
let unauthorized_registry = MemRegistry::default();
|
||
unauthorized_registry
|
||
.put_repo(&repo("owner/doubled"))
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(
|
||
AuthorizedRepo::authorize(
|
||
unauthorized_registry
|
||
.resolve("owner/doubled")
|
||
.await
|
||
.unwrap(),
|
||
Some(&holder),
|
||
AccessAction::Read
|
||
),
|
||
None,
|
||
"the grantless identity on the private repo denies"
|
||
);
|
||
let unauthorized_registry: Arc<dyn GitRegistry> = Arc::new(unauthorized_registry);
|
||
|
||
let (unknown, unknown_wire) = denial_wire(unknown_registry, Service::UploadPack).await;
|
||
let (unauthorized, unauthorized_wire) =
|
||
denial_wire(unauthorized_registry, Service::UploadPack).await;
|
||
|
||
match (&unknown, &unauthorized) {
|
||
(Err(SubstrateError::AccessDenied), Err(SubstrateError::AccessDenied)) => {}
|
||
other => panic!("both denials collapse to AccessDenied: {other:?}"),
|
||
}
|
||
|
||
assert!(!unknown_wire.is_empty());
|
||
assert_eq!(
|
||
unknown_wire, unauthorized_wire,
|
||
"the denials are byte-identical"
|
||
);
|
||
let denial = pkt_payloads(&unknown_wire);
|
||
assert_eq!(
|
||
denial,
|
||
vec![b"ERR access denied: owner/doubled".to_vec()],
|
||
"the collapsed band names only what the client sent, and no protocol byte precedes it"
|
||
);
|
||
assert!(
|
||
!String::from_utf8_lossy(&unknown_wire).contains("not found")
|
||
&& !String::from_utf8_lossy(&unknown_wire).contains("grant"),
|
||
"no cause leaks through the denial band"
|
||
);
|
||
}
|
||
|
||
/// Parses a raw captured wire byte sequence into pkt-line payloads (the
|
||
/// denial capture's shape: frames only, no trailing flush).
|
||
fn pkt_payloads(wire: &[u8]) -> Vec<Vec<u8>> {
|
||
let mut payloads = Vec::new();
|
||
let mut i = 0;
|
||
while i < wire.len() {
|
||
assert!(i + 4 <= wire.len(), "trailing partial header");
|
||
let len = usize::from_str_radix(
|
||
std::str::from_utf8(&wire[i..i + 4]).expect("hex header"),
|
||
16,
|
||
)
|
||
.expect("hex length");
|
||
assert!(len >= 4 && i + len <= wire.len(), "partial pkt-line");
|
||
payloads.push(wire[i + 4..i + len].to_vec());
|
||
i += len;
|
||
}
|
||
payloads
|
||
}
|
||
|
||
/// Drives the resolve→authorize front to its denial and captures the
|
||
/// denial's wire bytes through a live reader (the client's view: the
|
||
/// band is the first thing on the wire).
|
||
async fn denial_wire(
|
||
registry: Arc<dyn GitRegistry>,
|
||
service: Service,
|
||
) -> (Result<(), SubstrateError>, Vec<u8>) {
|
||
let (mut peer, stream) = duplex(1 << 16);
|
||
let holder = identity("u-7731");
|
||
let refs: Arc<dyn GitRefs> = Arc::new(MemRefs::default());
|
||
let gen: Arc<dyn GitPackGen> = Arc::new(MemPackGen::default());
|
||
let ingest: Arc<dyn GitPackIngest> = Arc::new(MemPackIngest::default());
|
||
let joined = tokio::spawn(async move {
|
||
serve_duplex_resolved(
|
||
"owner/doubled",
|
||
Some(&holder),
|
||
service,
|
||
stream,
|
||
Limits::default(),
|
||
registry,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
false,
|
||
Arc::new(Semaphore::new(
|
||
Limits::default().max_concurrent_blocking_tasks,
|
||
)),
|
||
)
|
||
.await
|
||
});
|
||
|
||
let mut wire = Vec::new();
|
||
let mut chunk = [0u8; 256];
|
||
let drained = tokio::time::timeout(Duration::from_secs(5), async {
|
||
loop {
|
||
let n = peer.read(&mut chunk).await.expect("the denial wire");
|
||
if n == 0 {
|
||
break;
|
||
}
|
||
wire.extend_from_slice(&chunk[..n]);
|
||
}
|
||
})
|
||
.await;
|
||
if drained.is_err() {
|
||
panic!("the denial never closed the stream");
|
||
}
|
||
(joined.await.expect("the front task joins"), wire)
|
||
}
|
||
|
||
/// Reads the raw response bytes up to and including the terminating
|
||
/// flush — one advertisement or one report body.
|
||
async fn read_body_until_flush(reader: &mut tokio::io::DuplexStream) -> Vec<u8> {
|
||
let mut body = Vec::new();
|
||
loop {
|
||
let mut header = [0u8; 4];
|
||
reader.read_exact(&mut header).await.expect("header");
|
||
body.extend_from_slice(&header);
|
||
let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
|
||
.expect("hex length");
|
||
if len < 4 {
|
||
return body;
|
||
}
|
||
let mut payload = vec![0u8; len - 4];
|
||
reader.read_exact(&mut payload).await.expect("payload");
|
||
body.extend_from_slice(&payload);
|
||
}
|
||
}
|
||
|
||
/// A prepare double that counts its invocations — the observable for
|
||
/// the packless peek's path (no prepare call) and for fail-closed
|
||
/// breaches (no backend call).
|
||
struct CountedIngest {
|
||
inner: Arc<MemPackIngest>,
|
||
runs: Arc<std::sync::atomic::AtomicUsize>,
|
||
}
|
||
|
||
fn counting_ingest(
|
||
inner: Arc<MemPackIngest>,
|
||
runs: Arc<std::sync::atomic::AtomicUsize>,
|
||
) -> Arc<dyn GitPackIngest> {
|
||
Arc::new(CountedIngest { inner, runs })
|
||
}
|
||
|
||
#[async_trait::async_trait]
|
||
impl GitPackIngest for CountedIngest {
|
||
async fn prepare(
|
||
&self,
|
||
record: &RepoRecord,
|
||
updates: Vec<RefUpdate>,
|
||
pack: Box<dyn std::io::Read + Send>,
|
||
push_options: Option<PushOptions>,
|
||
limits: Limits,
|
||
) -> Result<PreparedPush, StorageError> {
|
||
use std::sync::atomic::Ordering::SeqCst;
|
||
self.runs.fetch_add(1, SeqCst);
|
||
self.inner
|
||
.prepare(record, updates, pack, push_options, limits)
|
||
.await
|
||
}
|
||
}
|
||
|
||
/// A prepare double that records the push-options it received and
|
||
/// delegates — the observable for the duplex options-section routing.
|
||
struct OptionsRecorder {
|
||
inner: Arc<MemPackIngest>,
|
||
seen: Arc<std::sync::Mutex<Option<PushOptions>>>,
|
||
}
|
||
|
||
#[async_trait::async_trait]
|
||
impl GitPackIngest for OptionsRecorder {
|
||
async fn prepare(
|
||
&self,
|
||
record: &RepoRecord,
|
||
updates: Vec<RefUpdate>,
|
||
pack: Box<dyn std::io::Read + Send>,
|
||
push_options: Option<PushOptions>,
|
||
limits: Limits,
|
||
) -> Result<PreparedPush, StorageError> {
|
||
*self.seen.lock().expect("seen is not poisoned") = push_options.clone();
|
||
self.inner
|
||
.prepare(record, updates, pack, push_options, limits)
|
||
.await
|
||
}
|
||
}
|
||
|
||
/// The full push session over the duplex entry (ADR-013 §3's one
|
||
/// request per stream): the V0 ref advertisement speaks first with no
|
||
/// service prefix (the empty repo's `capabilities^{}` sentinel with the
|
||
/// caps NUL-attached, the honest served set — `quiet` absent), then one
|
||
/// request — the create command + the pack, dribbled in chunks —
|
||
/// ingests and CAS-applies, the report streams back sideband-framed,
|
||
/// and the session ends cleanly after the report (the client's close
|
||
/// releases the relay's park; nothing rides after the report).
|
||
///
|
||
/// Contract stamp: ADR-013 §1–2 (the V0 advertisement, server speaks
|
||
/// first, no service prefix; the sentinel for the no-refs repo; the
|
||
/// honest set), §3 (one POST / one stream), §7 (the one-transaction CAS
|
||
/// home), §8 (the sideband report); ADR-020 §1 (the duplex tail
|
||
/// discipline — the bounded peek; rule (iv)'s clean close after the
|
||
/// report); ADR-005 §4 (the protocol-failure band mapping); ADR-009
|
||
/// (the wall clock over the whole push); core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_full_push_conversation_applies_and_reports_over_the_duplex_entry() {
|
||
let fx = fixture();
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
|
||
let (mut peer, joined) = serve_push(&fx, fx.ingest.clone(), false, Limits::default()).await;
|
||
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
let lines = text_lines(&advertisement);
|
||
assert_eq!(
|
||
lines.len(),
|
||
2,
|
||
"an empty repo advertises exactly the sentinel line + flush: {lines:?}"
|
||
);
|
||
let caps_line = &lines[0];
|
||
assert!(
|
||
caps_line.contains("capabilities^{}"),
|
||
"the empty repo's sentinel: {caps_line}"
|
||
);
|
||
assert!(
|
||
caps_line.contains('\0'),
|
||
"the caps ride NUL-attached on the first emitted line: {caps_line}"
|
||
);
|
||
assert!(
|
||
caps_line.contains("agent=alkgit/"),
|
||
"the agent token is advertised: {caps_line}"
|
||
);
|
||
for served in ["report-status", "side-band-64k", "atomic"] {
|
||
assert!(
|
||
caps_line.contains(served),
|
||
"{served} stays served: {caps_line}"
|
||
);
|
||
}
|
||
assert!(
|
||
!caps_line.contains("quiet"),
|
||
"quiet is not advertised (ADR-013 §2): {caps_line}"
|
||
);
|
||
assert!(
|
||
!lines.iter().all(|line| line.contains("# service=")),
|
||
"no service prefix rides the duplex push advertisement"
|
||
);
|
||
assert_eq!(*lines.last().expect("the flush"), "", "flush-terminated");
|
||
|
||
let request = [
|
||
push_commands(
|
||
"report-status side-band-64k",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
),
|
||
vec![
|
||
chain.pack[..1].to_vec(),
|
||
chain.pack[1..24].to_vec(),
|
||
chain.pack[24..].to_vec(),
|
||
],
|
||
]
|
||
.concat()
|
||
.concat();
|
||
peer.write_all(&request).await.unwrap();
|
||
|
||
let report_body = read_body_until_flush(&mut peer).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&report_body),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
"the report is unpack + per-ref in request order"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close ends the session quietly after the report");
|
||
|
||
assert_eq!(
|
||
fx.refs
|
||
.list_refs(&record)
|
||
.await
|
||
.unwrap()
|
||
.iter()
|
||
.filter_map(|line| Some((line.name.clone(), line.oid?)))
|
||
.collect::<BTreeMap<String, ObjectId>>(),
|
||
BTreeMap::from([("refs/heads/main".to_owned(), chain.commit_b)]),
|
||
"the CAS applied the pushed update"
|
||
);
|
||
assert!(
|
||
fx.ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pushed objects landed via the ingest handle"
|
||
);
|
||
}
|
||
|
||
/// The packless delete-only push (ADR-020 §1 rule (ii)): the client
|
||
/// sends commands + flush and stays open — the bounded peek times out
|
||
/// with zero pack bytes, the packless CAS-only path runs with **no
|
||
/// prepare call**, and the same report shape (the ~40 ms-datum capture)
|
||
/// streams back while the client is still open. The client's close
|
||
/// after the report ends the drain cleanly (rule (iv) — no mid-stream
|
||
/// RST; the session returns quietly, never an error).
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rules (ii)/(iv) (the packless path, the
|
||
/// clean-close discipline), §3 (the client's write-side-open waiting
|
||
/// shape); ADR-013 §7–8 (the CAS-only path, the same report);
|
||
/// ADR-009 (the peek window is the named `packless_peek` budget);
|
||
/// core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_packless_delete_push_takes_the_peek_packless_path_and_drains_cleanly() {
|
||
use std::sync::atomic::Ordering::SeqCst;
|
||
|
||
let fx = fixture();
|
||
let record = repo("owner/doubled");
|
||
let current = ObjectId::from([0x1b; 20]);
|
||
fx.refs
|
||
.apply_updates(
|
||
&record,
|
||
vec![RefUpdate {
|
||
name: "refs/heads/gone".to_owned(),
|
||
expected: None,
|
||
new: Some(current),
|
||
}],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||
let (mut peer, joined) = serve_push(
|
||
&fx,
|
||
counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
|
||
false,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
read_pkts(&mut peer).await;
|
||
peer.write_all(
|
||
&push_commands(
|
||
"report-status side-band-64k",
|
||
&[delete_command("refs/heads/gone", current)],
|
||
)
|
||
.concat(),
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let report_body = read_body_until_flush(&mut peer).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&report_body),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
|
||
"the packless report equals the capture's shape, while the client stays open"
|
||
);
|
||
assert_eq!(
|
||
runs.load(SeqCst),
|
||
0,
|
||
"the packless path runs no prepare call"
|
||
);
|
||
assert!(
|
||
!fx.refs
|
||
.list_refs(&record)
|
||
.await
|
||
.unwrap()
|
||
.iter()
|
||
.any(|line| line.name == "refs/heads/gone"),
|
||
"the CAS-only path deleted the ref"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the packless tail's drain ends cleanly, no mid-stream RST");
|
||
}
|
||
|
||
/// The nothing-to-do shape (ADR-013 §3): an immediate flush with no
|
||
/// commands after the advertisement — the machine replies flush, no
|
||
/// report; the session ends cleanly.
|
||
///
|
||
/// Contract stamp: ADR-013 §3 (the nothing-to-do shape — reply flush,
|
||
/// no report); ADR-020 §1 (its duplex read shape); ADR-009 (no
|
||
/// prepare call, no peek); core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_push_nothing_to_do_flush_replies_flush_with_no_report() {
|
||
use std::sync::atomic::Ordering::SeqCst;
|
||
|
||
let fx = fixture();
|
||
let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||
let (mut peer, joined) = serve_push(
|
||
&fx,
|
||
counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
|
||
false,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
read_pkts(&mut peer).await;
|
||
peer.write_all(FLUSH).await.unwrap();
|
||
|
||
let reply = read_pkts(&mut peer).await;
|
||
assert_eq!(
|
||
reply,
|
||
vec![Vec::<u8>::new()],
|
||
"exactly one reply flush, no report"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the session ends quietly after the reply");
|
||
assert_eq!(
|
||
runs.load(SeqCst),
|
||
0,
|
||
"the nothing-to-do shape runs no prepare call"
|
||
);
|
||
}
|
||
|
||
/// A partial pack header at the peek timeout is the malformed-push
|
||
/// session error (ADR-020 §1 rule (iii) — a half-sent pack cannot be
|
||
/// re-interpreted as absence): the error bands, no report, no ref
|
||
/// applies, no prepare call.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rule (iii); ADR-005 §4 (the
|
||
/// protocol-failure band); ADR-009 (the peek window bounds the gather,
|
||
/// observed bytes still classify);
|
||
/// core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_partial_pack_header_at_the_peek_timeout_is_a_malformed_push() {
|
||
use std::sync::atomic::Ordering::SeqCst;
|
||
|
||
let fx = fixture();
|
||
let record = repo("owner/doubled");
|
||
let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||
let limits = Limits {
|
||
packless_peek: Duration::from_millis(150),
|
||
..Limits::default()
|
||
};
|
||
let (mut peer, joined) = serve_push(
|
||
&fx,
|
||
counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
|
||
false,
|
||
limits,
|
||
)
|
||
.await;
|
||
|
||
read_pkts(&mut peer).await;
|
||
peer.write_all(
|
||
&push_commands(
|
||
"report-status",
|
||
&[create_command(
|
||
"refs/heads/main",
|
||
ObjectId::from([0x2c; 20]),
|
||
)],
|
||
)
|
||
.concat(),
|
||
)
|
||
.await
|
||
.unwrap();
|
||
peer.write_all(b"PA").await.unwrap();
|
||
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(
|
||
band.starts_with("ERR ") && band.contains("half-sent pack"),
|
||
"the partial header is the malformed-push band: {band}"
|
||
);
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("the malformed push is a typed failure");
|
||
assert!(
|
||
matches!(result, SubstrateError::Protocol(_)),
|
||
"a malformed push is protocol-shaped (banded): {result:?}"
|
||
);
|
||
assert_eq!(runs.load(SeqCst), 0, "no prepare call ran");
|
||
assert!(
|
||
fx.refs.list_refs(&record).await.unwrap().is_empty(),
|
||
"no ref applied"
|
||
);
|
||
}
|
||
|
||
/// The negotiated `push-options` section is consumed before the pack
|
||
/// peek (ADR-020 §1 rule (i)) and routed into `prepare`: the
|
||
/// gate-on advertisement carries the `push-options` token; the client
|
||
/// selects the cap and sends the options section + its flush between
|
||
/// the command flush and the pack — the pack still ingests byte-exact
|
||
/// past the second flush.
|
||
///
|
||
/// Contract stamp: ADR-013 §11 (the section's grammar, the
|
||
/// negotiated gate); ADR-020 §1 rule (i) (consumed before the peek);
|
||
/// ADR-013 §2 (the gate rides the advertisement);
|
||
/// core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn the_negotiated_push_options_section_is_consumed_before_the_pack_peek() {
|
||
let fx = fixture();
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
|
||
let seen = Arc::new(std::sync::Mutex::new(None));
|
||
let ingest: Arc<dyn GitPackIngest> = Arc::new(OptionsRecorder {
|
||
inner: Arc::clone(&fx.ingest),
|
||
seen: Arc::clone(&seen),
|
||
});
|
||
let (mut peer, joined) = serve_push(&fx, ingest, true, Limits::default()).await;
|
||
|
||
let advertisement = text_lines(&read_pkts(&mut peer).await);
|
||
assert!(
|
||
advertisement[0].contains("push-options"),
|
||
"the gate's token rides the gate-on advertisement: {:?}",
|
||
advertisement[0]
|
||
);
|
||
|
||
let request = [
|
||
push_commands(
|
||
"report-status side-band-64k push-options",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
),
|
||
vec![pkt(b"deploy-key=abc"), FLUSH.to_vec(), chain.pack.clone()],
|
||
]
|
||
.concat()
|
||
.concat();
|
||
peer.write_all(&request).await.unwrap();
|
||
|
||
let report_body = read_body_until_flush(&mut peer).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&report_body),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
);
|
||
assert_eq!(
|
||
*seen.lock().expect("seen is not poisoned"),
|
||
Some(PushOptions {
|
||
pairs: vec![("deploy-key".to_owned(), "abc".to_owned())]
|
||
}),
|
||
"the negotiated options section routed into prepare"
|
||
);
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect("the clean close after the report");
|
||
assert!(
|
||
fx.ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pack after the options flush ingested byte-exact"
|
||
);
|
||
}
|
||
|
||
/// The denial path (ADR-007/ADR-008): an unauthorized push collapses to
|
||
/// the generic band *before any advertisement byte* — the resolve→
|
||
/// authorize gate precedes the V0 advertisement, unknown ≡ unauthorized.
|
||
///
|
||
/// Contract stamp: ADR-007 steps 2–3 (the gate precedes any protocol
|
||
/// byte — the advertisement included); ADR-008/N-2 (unknown ≡
|
||
/// unauthorized); ADR-011 §3/ADR-015 §2 (push is the write action);
|
||
/// core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_denied_push_is_the_generic_band_before_any_advertisement_byte() {
|
||
let unauthorized_registry = MemRegistry::default();
|
||
unauthorized_registry
|
||
.put_repo(&repo("owner/doubled"))
|
||
.await
|
||
.unwrap();
|
||
let unauthorized_registry: Arc<dyn GitRegistry> = Arc::new(unauthorized_registry);
|
||
|
||
let (result, wire) = denial_wire(unauthorized_registry, Service::ReceivePack).await;
|
||
match &result {
|
||
Err(SubstrateError::AccessDenied) => {}
|
||
other => panic!("the denial is the typed AccessDenied: {other:?}"),
|
||
}
|
||
assert_eq!(
|
||
pkt_payloads(&wire),
|
||
vec![b"ERR access denied: owner/doubled".to_vec()],
|
||
"the generic band names only the client-sent id; no advertisement preceded it"
|
||
);
|
||
}
|
||
|
||
/// The wall clock bounds the whole push (ADR-009): a client that stalls
|
||
/// after the advertisement (no request ever read, the peek never even
|
||
/// reached) breaches the session budget — the error band + close, the
|
||
/// typed server/session failure.
|
||
///
|
||
/// Contract stamp: ADR-009 (the session wall clock over the whole
|
||
/// session, push included; breach is the server/session class — error
|
||
/// band + close on duplex); ADR-020 §1 (the peek window runs inside
|
||
/// the wall clock, never outside it); core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn the_wall_clock_bounds_the_push_session() {
|
||
let fx = fixture();
|
||
let limits = Limits {
|
||
session_wall_clock: Duration::from_millis(200),
|
||
..Limits::default()
|
||
};
|
||
|
||
let (mut peer, joined) = serve_push(&fx, fx.ingest.clone(), false, limits).await;
|
||
read_pkts(&mut peer).await;
|
||
// Nothing is ever sent: the stalled request read breaches the budget.
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(
|
||
band.starts_with("ERR ") && band.contains("session_wall_clock"),
|
||
"the breach bands the push session: {band}"
|
||
);
|
||
|
||
let mut remaining = [0u8; 1];
|
||
assert_eq!(
|
||
peer.read(&mut remaining).await.expect("read after close"),
|
||
0,
|
||
"the breached session closed"
|
||
);
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("the breach is a typed failure");
|
||
assert!(
|
||
matches!(
|
||
&result,
|
||
SubstrateError::Protocol(message) if message.contains("session_wall_clock")
|
||
),
|
||
"the typed failure returns: {result:?}"
|
||
);
|
||
}
|
||
|
||
/// A push advertisement listing failure bands the session (the
|
||
/// backend-failure mapping — ADR-005 §4 through the
|
||
/// storage-to-session map, the ADR-008 posture at the advertisement):
|
||
/// nothing is emitted, never a partial ref list.
|
||
///
|
||
/// Contract stamp: ADR-013 §2 (the advertisement's listing — the one
|
||
/// backend call; failures map through the storage-to-session error
|
||
/// model, review 001 N-1's no-partial-list rule); ADR-005 §4
|
||
/// (backend failures band-map);
|
||
/// core/w3-duplex-serving.
|
||
#[tokio::test]
|
||
async fn a_push_advertisement_listing_failure_bands_the_session() {
|
||
struct ListingFailure;
|
||
#[async_trait::async_trait]
|
||
impl GitRefs for ListingFailure {
|
||
async fn list_refs(
|
||
&self,
|
||
_repo: &RepoRecord,
|
||
) -> Result<Vec<alkgit::RefLine>, StorageError> {
|
||
Err(StorageError::Invalid("listing unavailable".to_string()))
|
||
}
|
||
|
||
async fn apply_updates(
|
||
&self,
|
||
_repo: &RepoRecord,
|
||
_updates: Vec<RefUpdate>,
|
||
_atomic: bool,
|
||
) -> Result<Vec<alkgit::RefOutcome>, StorageError> {
|
||
Err(StorageError::Invalid("listing unavailable".to_string()))
|
||
}
|
||
}
|
||
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = Arc::new(ListingFailure);
|
||
let gen: Arc<dyn GitPackGen> = fx.gen.clone();
|
||
let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
|
||
|
||
let (mut peer, joined) = serve(
|
||
&fx.write_marker,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
false,
|
||
Limits::default(),
|
||
Service::ReceivePack,
|
||
)
|
||
.await;
|
||
|
||
let band = text(&read_band(&mut peer).await);
|
||
assert!(band.starts_with("ERR "), "a listing failure bands: {band}");
|
||
assert!(
|
||
band.contains("listing unavailable"),
|
||
"the backend failure's own text bands through the map: {band}"
|
||
);
|
||
assert_eq!(
|
||
peer.read(&mut [0u8; 1]).await.expect("read after close"),
|
||
0,
|
||
"the banded session closed"
|
||
);
|
||
let result = joined
|
||
.await
|
||
.expect("the session task joins")
|
||
.expect_err("the failure is typed");
|
||
assert!(
|
||
matches!(result, SubstrateError::Protocol(_)),
|
||
"a backend failure band-maps to a protocol-shaped typed failure: {result:?}"
|
||
);
|
||
}
|
||
|
||
/// A resolved front with a passing gate reaches the session: an
|
||
/// anonymous caller on a public repo — the anonymous-first-class read
|
||
/// path — gets the advertisement, and unknown-repo/unauthorized callers
|
||
/// get only the generic denial (nothing in between).
|
||
///
|
||
/// Contract stamp: ADR-011 §3 (anonymous public read is first-class at
|
||
/// the gate); ADR-007 (the gate precedes the advertisement);
|
||
/// core/w2-duplex-serving.
|
||
#[tokio::test]
|
||
async fn the_resolved_front_serves_the_anonymous_public_read_session() {
|
||
let fx = fixture();
|
||
let mut public = repo("owner/doubled");
|
||
public.visibility = Visibility::Public;
|
||
let registry = MemRegistry::default();
|
||
registry.put_repo(&public).await.unwrap();
|
||
let registry: Arc<dyn GitRegistry> = Arc::new(registry);
|
||
|
||
let (mut peer, stream) = duplex(1 << 16);
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let gen: Arc<dyn GitPackGen> = fx.gen.clone();
|
||
let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
|
||
let joined = tokio::spawn(async move {
|
||
serve_duplex_resolved(
|
||
"owner/doubled",
|
||
None,
|
||
Service::UploadPack,
|
||
stream,
|
||
Limits::default(),
|
||
registry,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
false,
|
||
Arc::new(Semaphore::new(
|
||
Limits::default().max_concurrent_blocking_tasks,
|
||
)),
|
||
)
|
||
.await
|
||
});
|
||
|
||
let advertisement = read_pkts(&mut peer).await;
|
||
assert_eq!(text_lines(&advertisement), advertisement_pkts());
|
||
|
||
peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
|
||
.await
|
||
.unwrap();
|
||
let _ = read_pkts(&mut peer).await;
|
||
|
||
drop(peer);
|
||
joined
|
||
.await
|
||
.expect("the front task joins")
|
||
.expect("the clean close ends the session quietly");
|
||
}
|
||
|
||
/// The entry's marker is the type-level ACL ordering: the only
|
||
/// constructor path into the entries' input type runs the checked
|
||
/// policy — a denied evaluation yields no marker value, so an
|
||
/// unauthorized session cannot reach [`serve_duplex`] at any type level.
|
||
///
|
||
/// Contract stamp: ADR-007 step 3 (the marker is the only path in —
|
||
/// resolve → authorize precedes any protocol byte); ADR-011 §3 (the
|
||
/// policy evaluation producing it); core/w2-duplex-serving.
|
||
#[test]
|
||
fn only_the_checked_marker_path_feeds_the_entries() {
|
||
let private = repo("owner/doubled");
|
||
let holder = identity("u-7731");
|
||
assert_eq!(
|
||
AuthorizedRepo::authorize(private.clone(), None, AccessAction::Read),
|
||
None,
|
||
"anonymous on a private repo is no marker"
|
||
);
|
||
assert_eq!(
|
||
AuthorizedRepo::authorize(private.clone(), Some(&holder), AccessAction::Read),
|
||
None,
|
||
"a grantless identity is no marker"
|
||
);
|
||
let mut granted = private;
|
||
granted
|
||
.grants
|
||
.insert("u-7731".to_string(), vec![AccessAction::Read]);
|
||
assert!(
|
||
AuthorizedRepo::authorize(granted, Some(&holder), AccessAction::Read).is_some(),
|
||
"the granted identity is the entry's input"
|
||
);
|
||
}
|