glm-5.3-flash
8a10a678e0
The rider-1 real-client capture (n5 task, executed at the wave-5 gate) corrected the wire emission shape: upstream ls-refs.c sends the literal `unborn` token in the oid field; the zero-oid shape we served since wave 2 made a real clone fetch-want the null oid (`ERR object <zero-oid> does not exist`, captured live). The inbound RefLine parser accepted both shapes all along; the emission pins to the upstream form. Pre-publish draft-shape amendment (the OQ-03 freeze is at the core's publish): ADR-018 gains the §5 amendment note; transport.md §ls-refs records the validation + corrected shape; the pinned rows re-byte-pin the literal (v2_machines renamed, the stateless/parse-back rows updated); the RefLine/LsRefsArgs doc comments and the CHANGELOG wave-2 bullet corrected. Verification: cargo test (workspace 375 default / 378 all-features, -p alkgit 376, -p alkgit --features sha256 379, 0 fail), both clippies -D warnings clean, cargo fmt --check, cargo doc zero warnings, the fuzz corpus gate 13/13, cargo publish --dry-run green.
1789 lines
59 KiB
Rust
1789 lines
59 KiB
Rust
//! The wave-2 stateless serving entry (core/w2-stateless-serving): the
|
||
//! per-POST dispatch the http door mounts, the info/refs advertisement
|
||
//! emission, the per-request budgets, and the no-cross-request-state
|
||
//! property — the same V2 machines the duplex session feeds, driven
|
||
//! over in-memory reader/writer ends with the wave-2 backend doubles.
|
||
//!
|
||
//! Rows are names + `Contract stamp:` versions, per
|
||
//! `tests/common` §The row/stamp discipline.
|
||
|
||
mod common;
|
||
|
||
use std::collections::BTreeSet;
|
||
use std::pin::Pin;
|
||
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
|
||
use std::sync::{Arc, Mutex as StdMutex};
|
||
use std::task::{Context, Poll};
|
||
use std::time::Duration;
|
||
|
||
use async_trait::async_trait;
|
||
use futures::io::AsyncWrite as FuturesAsyncWrite;
|
||
use futures::stream;
|
||
use gix_hash::ObjectId;
|
||
use tokio::sync::{oneshot, Semaphore};
|
||
|
||
use alkgit::{
|
||
serve_stateless_advertisement, serve_stateless_post, AccessAction, AuthorizedRepo, BodyReader,
|
||
GitPackGen, GitPackIngest, GitRefs, Limits, PreparedPush, PushOptions, RefLine, RefOutcome,
|
||
RefUpdate, RepoRecord, Service, StorageError, SubstrateError, Visibility,
|
||
};
|
||
use common::{
|
||
blob_fixture, chain_pack, commit_fixture, create_command, decode_pack, delete_command,
|
||
parse_pkt_lines, push_commands, sideband_report_lines, tree_fixture, wired_text as text,
|
||
MemObjects, MemPackGen, MemPackIngest, MemRefs,
|
||
};
|
||
|
||
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: std::collections::BTreeMap::new(),
|
||
}
|
||
}
|
||
|
||
fn oid(b: u8) -> ObjectId {
|
||
ObjectId::from([b; 20])
|
||
}
|
||
|
||
fn hex_oid(b: u8) -> String {
|
||
let mut hex = String::with_capacity(40);
|
||
for _ in 0..20 {
|
||
hex.push_str(&format!("{b:02x}"));
|
||
}
|
||
hex
|
||
}
|
||
|
||
/// The marker the door hands the entries — constructible only through
|
||
/// the checked path; here for a private repo with a read-granted identity.
|
||
fn marker(record: RepoRecord, identity: Option<&alkcall::core::Identity>) -> AuthorizedRepo {
|
||
AuthorizedRepo::authorize(record, identity, AccessAction::Read)
|
||
.expect("the door'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
|
||
}
|
||
|
||
/// The first POST body: the smart prefix section, its flush, the V2
|
||
/// capability dump section, its flush, then the command invocation.
|
||
fn first_post_body(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
|
||
[
|
||
pkt(b"# service=git-upload-pack\n"),
|
||
FLUSH.to_vec(),
|
||
pkt(b"version 2\n"),
|
||
pkt(b"agent=git/2.43\n"),
|
||
pkt(b"object-format=sha1\n"),
|
||
FLUSH.to_vec(),
|
||
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()
|
||
}
|
||
|
||
/// A subsequent POST body (negotiation rounds): only the command.
|
||
fn command_post_body(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()
|
||
}
|
||
|
||
/// An async sink whose framed bytes the rows inspect after the fact —
|
||
/// the http door's response body surface shape.
|
||
#[derive(Clone, Default)]
|
||
struct VecSink(Arc<StdMutex<Vec<u8>>>);
|
||
|
||
impl VecSink {
|
||
fn body(&self) -> Vec<u8> {
|
||
self.0.lock().expect("sink is not poisoned").clone()
|
||
}
|
||
}
|
||
|
||
impl FuturesAsyncWrite for VecSink {
|
||
fn poll_write(
|
||
self: Pin<&mut Self>,
|
||
_cx: &mut Context<'_>,
|
||
buf: &[u8],
|
||
) -> Poll<std::io::Result<usize>> {
|
||
self.0
|
||
.lock()
|
||
.expect("sink is not poisoned")
|
||
.extend_from_slice(buf);
|
||
Poll::Ready(Ok(buf.len()))
|
||
}
|
||
|
||
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
||
Poll::Ready(Ok(()))
|
||
}
|
||
|
||
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
||
self.poll_flush(cx)
|
||
}
|
||
}
|
||
|
||
/// A request-body end that never yields and never ends — the wall-clock
|
||
/// budget's victim (a stalled client or a stalled door body).
|
||
struct PendingReader;
|
||
|
||
impl futures::AsyncRead for PendingReader {
|
||
fn poll_read(
|
||
self: Pin<&mut Self>,
|
||
_cx: &mut Context<'_>,
|
||
_buf: &mut [u8],
|
||
) -> Poll<std::io::Result<usize>> {
|
||
Poll::Pending
|
||
}
|
||
}
|
||
|
||
type ChunkStream = stream::Iter<std::vec::IntoIter<std::io::Result<bytes::Bytes>>>;
|
||
type PostBody = BodyReader<ChunkStream, std::io::Error>;
|
||
|
||
fn body_parts(parts: Vec<Vec<u8>>) -> PostBody {
|
||
let items: Vec<std::io::Result<bytes::Bytes>> = parts
|
||
.into_iter()
|
||
.map(|c| Ok(bytes::Bytes::from(c)))
|
||
.collect();
|
||
BodyReader::new(stream::iter(items))
|
||
}
|
||
|
||
/// The door's per-POST assembly: the marker, the backend handles keyed
|
||
/// on the resolved repo, the shared semaphore — the shape
|
||
/// serve_stateless_post consumes.
|
||
struct Fixture {
|
||
marker: AuthorizedRepo,
|
||
refs: Arc<MemRefs>,
|
||
gen: Arc<dyn GitPackGen>,
|
||
ingest: Arc<MemPackIngest>,
|
||
objects: MemObjects,
|
||
repo_id: &'static str,
|
||
}
|
||
|
||
fn fixture() -> Fixture {
|
||
let repo_id = "owner/doubled";
|
||
let mut record = repo(repo_id);
|
||
record
|
||
.grants
|
||
.insert("u-7731".to_string(), vec![AccessAction::Read]);
|
||
let holder = alkcall::core::Identity {
|
||
id: "u-7731".to_string(),
|
||
scopes: Vec::new(),
|
||
resources: std::collections::HashMap::new(),
|
||
};
|
||
let objects = MemObjects::default();
|
||
Fixture {
|
||
marker: marker(record, Some(&holder)),
|
||
refs: Arc::new(MemRefs::default()),
|
||
gen: Arc::new(MemPackGen {
|
||
objects: objects.clone(),
|
||
}),
|
||
ingest: Arc::new(MemPackIngest {
|
||
objects: objects.clone(),
|
||
}),
|
||
objects,
|
||
repo_id,
|
||
}
|
||
}
|
||
|
||
impl Fixture {
|
||
fn record(&self) -> RepoRecord {
|
||
repo(self.repo_id)
|
||
}
|
||
|
||
fn with_gen(&self, gen: Arc<dyn GitPackGen>) -> Fixture {
|
||
Fixture {
|
||
marker: self.marker.clone(),
|
||
refs: Arc::clone(&self.refs),
|
||
gen,
|
||
ingest: Arc::clone(&self.ingest),
|
||
objects: self.objects.clone(),
|
||
repo_id: self.repo_id,
|
||
}
|
||
}
|
||
|
||
async fn serve(
|
||
&self,
|
||
body: PostBody,
|
||
limits: Limits,
|
||
service: Service,
|
||
pipeline: Arc<Semaphore>,
|
||
) -> (Result<(), SubstrateError>, VecSink) {
|
||
let sink = VecSink::default();
|
||
let refs: Arc<dyn GitRefs> = self.refs.clone();
|
||
let result = serve_stateless_post(
|
||
&self.marker,
|
||
body,
|
||
sink.clone(),
|
||
limits,
|
||
service,
|
||
refs,
|
||
Arc::clone(&self.gen),
|
||
self.ingest.clone() as Arc<dyn GitPackIngest>,
|
||
false,
|
||
pipeline,
|
||
)
|
||
.await;
|
||
(result, sink)
|
||
}
|
||
|
||
/// The push POST's serving call: `Service::ReceivePack`, the
|
||
/// `push-options` negotiated-gate flag alongside (the same
|
||
/// assembler config the GET carries).
|
||
async fn serve_push(
|
||
&self,
|
||
body: PostBody,
|
||
limits: Limits,
|
||
push_options_negotiated: bool,
|
||
refs: Arc<dyn GitRefs>,
|
||
ingest: Arc<dyn GitPackIngest>,
|
||
pipeline: Arc<Semaphore>,
|
||
) -> (Result<(), SubstrateError>, VecSink) {
|
||
let sink = VecSink::default();
|
||
let result = serve_stateless_post(
|
||
&self.marker,
|
||
body,
|
||
sink.clone(),
|
||
limits,
|
||
Service::ReceivePack,
|
||
refs,
|
||
Arc::clone(&self.gen),
|
||
ingest,
|
||
push_options_negotiated,
|
||
pipeline,
|
||
)
|
||
.await;
|
||
(result, sink)
|
||
}
|
||
}
|
||
|
||
fn default_pipeline() -> Arc<Semaphore> {
|
||
Arc::new(Semaphore::new(
|
||
Limits::default().max_concurrent_blocking_tasks,
|
||
))
|
||
}
|
||
|
||
async fn assert_protocol_error(result: Result<(), SubstrateError>, needle: &str) {
|
||
let error = result.expect_err("expected a typed protocol failure");
|
||
assert!(
|
||
error.is_terminal(),
|
||
"the typed failure is terminal: {error:?}"
|
||
);
|
||
match &error {
|
||
SubstrateError::Protocol(message) => {
|
||
assert!(message.contains(needle), "{message}");
|
||
}
|
||
other => panic!("expected a protocol-shaped failure, got {other:?}"),
|
||
}
|
||
}
|
||
|
||
/// The first POST carries the cached capability dump — the entry skips
|
||
/// both flush-delimited sections and serves the ls-refs command: ref
|
||
/// lines from the backend listing, attributes on the same line, the
|
||
/// unborn HEAD row emitted as the literal-unborn line, the response
|
||
/// ending at flush.
|
||
///
|
||
/// Contract stamp: ADR-005 §2 (one POST body is one command invocation,
|
||
/// the capability-dump skip is the machinery's); doors.md §alkhttp git
|
||
/// feature (POST /{repo}/git-upload-pack, one V2 command per POST);
|
||
/// ADR-003 (the honest advertisement the client re-sends); review 001
|
||
/// N-5 (the unborn-HEAD line shape, wire form captured literal-unborn —
|
||
/// the rider-1 real-client sessions, wave 5, corrected the zero-oid
|
||
/// form); core/w2-stateless-serving (the entry dispatches the same
|
||
/// machines).
|
||
#[tokio::test]
|
||
async fn the_first_posts_capability_dump_is_skipped_and_ls_refs_serves() {
|
||
let fx = fixture();
|
||
let doubled = fx.record();
|
||
fx.refs
|
||
.apply_updates(
|
||
&doubled,
|
||
vec![
|
||
RefUpdate {
|
||
name: "refs/heads/main".to_string(),
|
||
expected: None,
|
||
new: Some(oid(1)),
|
||
},
|
||
RefUpdate {
|
||
name: "refs/tags/v1".to_string(),
|
||
expected: None,
|
||
new: Some(oid(2)),
|
||
},
|
||
],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
fx.refs
|
||
.set_head_symref(&doubled, Some("refs/heads/main".to_string()));
|
||
|
||
let body = body_parts(vec![first_post_body(
|
||
b"command=ls-refs\n",
|
||
&["agent=git/2.43"],
|
||
&["symrefs", "unborn"],
|
||
)]);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "unborn HEAD symref-target:refs/heads/main");
|
||
assert_eq!(text(&lines[1]), format!("{} refs/heads/main", oid(1)));
|
||
assert_eq!(text(&lines[2]), format!("{} refs/tags/v1", oid(2)));
|
||
assert!(lines[3].is_empty(), "the response ends at flush");
|
||
assert_eq!(lines.len(), 4);
|
||
}
|
||
|
||
/// No cross-request state (ADR-014 §7): two consecutive POSTs ride two
|
||
/// separate entry invocations and re-derive identically — byte-equal
|
||
/// response bodies, no session to restore anywhere.
|
||
///
|
||
/// Contract stamp: ADR-002 (the stateless session entry, per-request
|
||
/// state); ADR-014 §7 (statelessness is a substrate property — the
|
||
/// client re-sends wants + commons every round);
|
||
/// core/w2-stateless-serving (the entry holds nothing between
|
||
/// invocations).
|
||
#[tokio::test]
|
||
async fn two_consecutive_posts_re_derive_identically() {
|
||
let fx = fixture();
|
||
let body = || {
|
||
body_parts(vec![first_post_body(
|
||
b"command=ls-refs\n",
|
||
&[],
|
||
&["peel", "ref-prefix refs/heads/"],
|
||
)])
|
||
};
|
||
|
||
let (first, first_sink) = fx
|
||
.serve(
|
||
body(),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
first.unwrap();
|
||
|
||
let (second, second_sink) = fx
|
||
.serve(
|
||
body(),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
second.unwrap();
|
||
|
||
assert_eq!(
|
||
first_sink.body(),
|
||
second_sink.body(),
|
||
"two POSTs, one deterministic response"
|
||
);
|
||
}
|
||
|
||
/// A full first-POST fetch round (want + done, capability dump ahead of
|
||
/// the command) serves the done round through the entry: the packfile
|
||
/// section streams as band-1 over the response rules and decodes as
|
||
/// closure(wants) − closure(recognized haves).
|
||
///
|
||
/// Contract stamp: ADR-014 §2 (the done round, the packfile section —
|
||
/// `packfile` header, band-1 sideband, section flush, no
|
||
/// acknowledgments); ADR-005 §2 (the first POST's dump skip composes
|
||
/// with the machines); ADR-014 §5 (the ack seam — the recognized have
|
||
/// is subtracted); core/w2-stateless-serving (the entry composes the
|
||
/// negotiation machine per POST).
|
||
#[tokio::test]
|
||
async fn a_full_first_post_fetch_round_serves_the_done_round_pack() {
|
||
let fx = fixture();
|
||
let doubled = fx.record();
|
||
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");
|
||
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),
|
||
] {
|
||
fx.objects.put(&doubled, id, bytes);
|
||
}
|
||
let fx = fx.with_gen(Arc::new(MemPackGen {
|
||
objects: fx.objects.clone(),
|
||
}));
|
||
|
||
let body = body_parts(vec![first_post_body(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1"],
|
||
&[
|
||
&format!("want {commit_b}"),
|
||
&format!("have {commit_a}"),
|
||
"thin-pack",
|
||
"no-progress",
|
||
"done",
|
||
],
|
||
)]);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "packfile");
|
||
let band_chunks: Vec<&[u8]> = lines[1..lines.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_b, tree_b, blob_b]),
|
||
"closure(commit_b) − closure(commit_a) over the stateless entry"
|
||
);
|
||
let _ = decoded.trailer;
|
||
assert!(
|
||
lines.last().unwrap().is_empty(),
|
||
"the response ends at flush"
|
||
);
|
||
}
|
||
|
||
/// The OQ-19 rider-2 positive-ack capture over the stateless entry (the
|
||
/// machinery-side run): a no-`done` POST with multiple real haves 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 POST'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-014 §7 (each POST is one round, no
|
||
/// cross-request state).
|
||
#[tokio::test]
|
||
async fn a_multi_have_post_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
|
||
) {
|
||
let fx = fixture();
|
||
let doubled = fx.record();
|
||
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 fx = fx.with_gen(Arc::new(MemPackGen {
|
||
objects: fx.objects.clone(),
|
||
}));
|
||
|
||
let missing = oid(9u8);
|
||
let body = body_parts(vec![command_post_body(
|
||
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),
|
||
],
|
||
)]);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
let ack_round: Vec<String> = parse_pkt_lines(&sink.body())
|
||
.iter()
|
||
.map(|p| text(p))
|
||
.collect();
|
||
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"
|
||
);
|
||
|
||
let body = body_parts(vec![command_post_body(
|
||
b"command=fetch\n",
|
||
&["object-format=sha1", "agent=git/2.43"],
|
||
&[
|
||
&format!("want {}", commit_c),
|
||
&format!("have {}", commit_a),
|
||
&format!("have {}", missing),
|
||
&format!("have {}", blob_a),
|
||
&format!("have {}", commit_b),
|
||
"done",
|
||
],
|
||
)]);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "packfile");
|
||
let band_chunks: Vec<&[u8]> = lines[1..lines.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;
|
||
}
|
||
|
||
/// A no-`done` fetch POST re-derives the acknowledgments section per
|
||
/// POST — the client re-sent its wants + haves, the entry held nothing.
|
||
///
|
||
/// Contract stamp: ADR-014 §1 (the exact acknowledgments grammar,
|
||
/// `ready` never sent); ADR-014 §7 (the stateless rounds re-derive);
|
||
/// core/w2-stateless-serving.
|
||
#[tokio::test]
|
||
async fn a_no_done_fetch_post_re_derives_the_acknowledgments_per_post() {
|
||
let fx = fixture();
|
||
let doubled = fx.record();
|
||
let (blob, blob_bytes) = blob_fixture(b"paid");
|
||
let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
|
||
let (commit_a, commit_a_bytes) = commit_fixture(tree, &[], "alpha");
|
||
let (commit_b, commit_b_bytes) = commit_fixture(tree, &[commit_a], "beta");
|
||
for (id, bytes) in [
|
||
(blob, blob_bytes),
|
||
(tree, tree_bytes),
|
||
(commit_a, commit_a_bytes),
|
||
(commit_b, commit_b_bytes),
|
||
] {
|
||
fx.objects.put(&doubled, id, bytes);
|
||
}
|
||
let fx = fx.with_gen(Arc::new(MemPackGen {
|
||
objects: fx.objects.clone(),
|
||
}));
|
||
|
||
let body = body_parts(vec![command_post_body(
|
||
b"command=fetch\n",
|
||
&[],
|
||
&[
|
||
&format!("want {}", hex_oid(0x0d)),
|
||
&format!("have {commit_a}"),
|
||
],
|
||
)]);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "acknowledgments");
|
||
assert_eq!(text(&lines[1]), format!("ACK {commit_a}"));
|
||
assert!(lines.last().unwrap().is_empty());
|
||
}
|
||
|
||
/// The flush-only probe resolves to the empty-OK shape: `Ok` with
|
||
/// nothing written — the 200 mapping is the door's (the machinery
|
||
/// resolves; the entry surfaces nothing).
|
||
///
|
||
/// Contract stamp: ADR-005 §2 (the probe rule — answering an error
|
||
/// status breaks the whole clone, poc3-findings §probe_rpc);
|
||
/// core/w2-stateless-serving (`StatelessPost::Probe` → the empty-OK
|
||
/// shape, the response rules honored with nothing written).
|
||
#[tokio::test]
|
||
async fn the_probe_resolves_to_the_empty_ok_shape_with_nothing_written() {
|
||
let fx = fixture();
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body_parts(vec![FLUSH.to_vec()]),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
assert!(sink.body().is_empty(), "nothing written on the probe");
|
||
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body_parts(vec![Vec::new()]),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
assert!(sink.body().is_empty(), "the body-less probe as well");
|
||
}
|
||
|
||
/// An unknown command declines as a typed protocol failure — the
|
||
/// stateless side has no error band; the entry surfaces the failure the
|
||
/// door maps to an http status, writing nothing.
|
||
///
|
||
/// Contract stamp: ADR-005 §4 (the error taxonomy: protocol-shaped
|
||
/// failures map to http statuses at the door); ADR-003 (honest surface,
|
||
/// unimplemented features declined); core/w2-stateless-serving
|
||
/// (unknown command → the door-mapped protocol failure).
|
||
#[tokio::test]
|
||
async fn an_unknown_command_declines_as_the_typed_failure() {
|
||
let fx = fixture();
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body_parts(vec![command_post_body(b"command=object-info\n", &[], &[])]),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
assert_protocol_error(result, "unknown command").await;
|
||
assert!(sink.body().is_empty(), "the decline writes nothing");
|
||
}
|
||
|
||
/// A `git-receive-pack` probe POST resolves to the empty-OK shape
|
||
/// before any service dispatch — the probe shape is service-agnostic
|
||
/// (breaking it breaks the large-request path of *any* client), and on
|
||
/// the push path the large-request probe is ADR-013 §12's fact. The
|
||
/// same flush-only body is the V0 nothing-to-do shape over http
|
||
/// (upstream http-backend answers it 200-empty all the same) — the
|
||
/// reply-flush report shape of that body is the machine-level
|
||
/// discipline, pinned by core/w3-push-machine rows; nothing reaches
|
||
/// any backend handle here.
|
||
///
|
||
/// Contract stamp: ADR-005 §2 (the service-agnostic probe rule — the
|
||
/// empty-OK answer precedes any service dispatch); ADR-013 §12 (the
|
||
/// probe POST on the push path); ADR-013 §3 (the immediate-flush
|
||
/// nothing-to-do shape; over http it shares the probe's bytes);
|
||
/// ADR-020 §1; core/w3-stateless-serving.
|
||
#[tokio::test]
|
||
async fn the_flush_only_receive_pack_post_is_the_probe_shape_before_any_dispatch() {
|
||
let fx = fixture();
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = counting_ingest(fx.ingest.clone(), Arc::clone(&runs));
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
|
||
let (result, sink) = fx
|
||
.serve_push(
|
||
body_parts(vec![FLUSH.to_vec()]),
|
||
Limits::default(),
|
||
false,
|
||
Arc::clone(&refs),
|
||
ingest,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
assert!(
|
||
sink.body().is_empty(),
|
||
"the probe answer writes nothing (the empty-OK shape)"
|
||
);
|
||
assert_eq!(
|
||
runs.load(AtomicOrdering::SeqCst),
|
||
0,
|
||
"nothing was dispatched to any backend handle"
|
||
);
|
||
}
|
||
|
||
/// The wall-clock budget applies per request (the stateless substrate
|
||
/// has no session to amortize it): a stalled request body breaches the
|
||
/// budget and the entry surfaces the typed server/session failure the
|
||
/// door maps to 503.
|
||
///
|
||
/// Contract stamp: ADR-009 (the wall clock, per-request on http —
|
||
/// ADR-002's cost note; server/session budgets map to 503);
|
||
/// core/w2-stateless-serving (the wall clock applies per request).
|
||
#[tokio::test]
|
||
async fn the_wall_clock_budget_fails_closed_per_request() {
|
||
let fx = fixture();
|
||
let limits = Limits {
|
||
session_wall_clock: Duration::from_millis(20),
|
||
..Limits::default()
|
||
};
|
||
let sink = VecSink::default();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let result = serve_stateless_post(
|
||
&fx.marker,
|
||
PendingReader,
|
||
sink.clone(),
|
||
limits,
|
||
Service::UploadPack,
|
||
refs,
|
||
Arc::clone(&fx.gen),
|
||
fx.ingest.clone() as Arc<dyn GitPackIngest>,
|
||
false,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
match result {
|
||
Err(SubstrateError::Protocol(message)) => {
|
||
assert!(error_contains(&message, "session_wall_clock"), "{message}");
|
||
}
|
||
other => panic!("a wall-clock breach is a typed failure: {other:?}"),
|
||
}
|
||
assert!(sink.body().is_empty(), "the breach wrote nothing");
|
||
}
|
||
|
||
fn error_contains(message: &str, needle: &str) -> bool {
|
||
message.contains(needle)
|
||
}
|
||
|
||
/// A saturated pipeline-concurrency semaphore delays the entry's
|
||
/// generate call — the door's shared semaphore is handed through and
|
||
/// the permit wraps the done round; releasing a permit lets the POST
|
||
/// complete.
|
||
///
|
||
/// Contract stamp: ADR-009 (the blocking-pool budget, the permit
|
||
/// acquired around each `GitPackGen::generate` call — backend.md
|
||
/// §concurrency model); core/w2-stateless-serving (the shared
|
||
/// concurrency semaphore is the entry's input).
|
||
#[tokio::test]
|
||
async fn a_saturated_concurrency_permit_delays_the_entry_generate() {
|
||
let fx = fixture();
|
||
let doubled = fx.record();
|
||
let (blob, blob_bytes) = blob_fixture(b"paid");
|
||
let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
|
||
let (commit, commit_bytes) = commit_fixture(tree, &[], "paid");
|
||
for (id, bytes) in [
|
||
(blob, blob_bytes),
|
||
(tree, tree_bytes),
|
||
(commit, commit_bytes),
|
||
] {
|
||
fx.objects.put(&doubled, id, bytes);
|
||
}
|
||
let gen: Arc<dyn GitPackGen> = Arc::new(MemPackGen {
|
||
objects: fx.objects.clone(),
|
||
});
|
||
|
||
let pipeline = Arc::new(Semaphore::new(1));
|
||
let held = pipeline.clone().acquire_owned().await.unwrap();
|
||
let body = body_parts(vec![command_post_body(
|
||
b"command=fetch\n",
|
||
&[],
|
||
&[&format!("want {commit}"), "done"],
|
||
)]);
|
||
let sink = VecSink::default();
|
||
let (done_tx, mut done_rx) = oneshot::channel();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let marker = fx.marker.clone();
|
||
let ingest = fx.ingest.clone() as Arc<dyn GitPackIngest>;
|
||
tokio::spawn(async move {
|
||
let result = serve_stateless_post(
|
||
&marker,
|
||
body,
|
||
sink.clone(),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
refs,
|
||
gen,
|
||
ingest,
|
||
false,
|
||
pipeline,
|
||
)
|
||
.await;
|
||
done_tx.send((result, sink)).ok();
|
||
});
|
||
|
||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||
assert!(
|
||
done_rx.try_recv().is_err(),
|
||
"the saturated permit delays the done round's generate"
|
||
);
|
||
|
||
drop(held);
|
||
let (result, sink) = done_rx
|
||
.await
|
||
.expect("the entry completes once a permit releases");
|
||
result.unwrap();
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "packfile");
|
||
}
|
||
|
||
/// The entry is IO-abstract (convention 15): a tokio read end adapted
|
||
/// through compat plugs in like any door stream — no http type anywhere
|
||
/// below the stream.
|
||
///
|
||
/// Contract stamp: ADR-002 (front-door-blind core — the entry consumes
|
||
/// reader/writer ends); ADR-005 §2 (the machinery is generic over both
|
||
/// ends); convention 15 (no http/channel types below the stream);
|
||
/// core/w2-stateless-serving.
|
||
#[tokio::test]
|
||
async fn the_entry_runs_over_arbitrary_reader_writer_ends() {
|
||
use tokio::io::AsyncWriteExt as TokioAsyncWriteExt;
|
||
use tokio_util::compat::TokioAsyncReadCompatExt;
|
||
|
||
let fx = fixture();
|
||
let (mut peer, stream) = tokio::io::duplex(1 << 15);
|
||
TokioAsyncWriteExt::write_all(
|
||
&mut peer,
|
||
&command_post_body(b"command=ls-refs\n", &[], &["ref-prefix refs/heads/"]),
|
||
)
|
||
.await
|
||
.unwrap();
|
||
drop(peer);
|
||
|
||
let sink = VecSink::default();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let result = serve_stateless_post(
|
||
&fx.marker,
|
||
stream.compat(),
|
||
sink.clone(),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
refs,
|
||
Arc::clone(&fx.gen),
|
||
fx.ingest.clone() as Arc<dyn GitPackIngest>,
|
||
false,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert!(
|
||
lines.last().unwrap().is_empty(),
|
||
"the response ends at flush"
|
||
);
|
||
}
|
||
|
||
/// The info/refs advertisement emission: the smart prefix line, its
|
||
/// flush, then the V2 capability advertisement (the honest list) ending
|
||
/// at flush — the per-request-set shape the door's GET route serves,
|
||
/// the response body remote-curl caches and re-sends in its first POST.
|
||
///
|
||
/// Contract stamp: doors.md §alkhttp git feature (GET info/refs → smart
|
||
/// prefix + capability dump + flush); transport.md §Advertisement
|
||
/// (per-request-set on http); ADR-003 (the honest capability list);
|
||
/// ADR-016 §3 (the service rides the route); poc3-findings §1 (the
|
||
/// two-flush info/refs body is the shape the client caches);
|
||
/// core/w2-stateless-serving (the emission exists for the door's
|
||
/// route).
|
||
#[tokio::test]
|
||
async fn the_info_refs_advertisement_emits_the_smart_prefix_and_the_capability_dump() {
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::UploadPack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "# service=git-upload-pack");
|
||
assert!(lines[1].is_empty(), "the prefix section ends at flush");
|
||
assert_eq!(text(&lines[2]), "version 2");
|
||
// The honest capability list, in emit order — the sha256 rider
|
||
// rides the feature per OQ-05's passthrough policy.
|
||
#[cfg(not(feature = "sha256"))]
|
||
{
|
||
assert_eq!(text(&lines[3]), "ls-refs=unborn");
|
||
assert_eq!(text(&lines[4]), "fetch=wait-for-done");
|
||
assert_eq!(text(&lines[5]), "object-format=sha1");
|
||
assert!(lines[6].is_empty(), "the dump ends at flush");
|
||
assert_eq!(lines.len(), 7);
|
||
}
|
||
#[cfg(feature = "sha256")]
|
||
{
|
||
assert_eq!(text(&lines[3]), "ls-refs=unborn");
|
||
assert_eq!(text(&lines[4]), "fetch=wait-for-done");
|
||
assert_eq!(text(&lines[5]), "object-format=sha1");
|
||
assert_eq!(text(&lines[6]), "object-format=sha256");
|
||
assert!(lines[7].is_empty(), "the dump ends at flush");
|
||
assert_eq!(lines.len(), 8);
|
||
}
|
||
assert!(
|
||
!sink.body().windows(4).any(|w| w == b"0002"),
|
||
"no 0002 anywhere in the response body"
|
||
);
|
||
}
|
||
|
||
/// The receive-pack advertisement GET: the smart prefix + flush, then
|
||
/// the V0 ref advertisement over the backend's listed refs — caps
|
||
/// NUL-attached on the first ref line, bare ref lines, flush-terminated
|
||
/// — the shape the push client caches and re-sends inside its first
|
||
/// POST. A listed ref rides the emission; the ACL-first marker precedes
|
||
/// any byte.
|
||
///
|
||
/// Contract stamp: ADR-013 §1–2 (the V0 framing, the served set);
|
||
/// push-captures.md §Advertisement (the normative shapes); ADR-007
|
||
/// (ACL first via the marker); ADR-005 §2 (the two-flush info/refs
|
||
/// body is the shape the client caches); core/w3-v0-advertisement.
|
||
#[tokio::test]
|
||
async fn the_receive_pack_advertisement_emits_the_prefix_and_the_v0_ref_list() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let outcomes = fx
|
||
.refs
|
||
.apply_updates(
|
||
&record,
|
||
vec![RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(oid(1)),
|
||
}],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(outcomes.len(), 1, "the fixture's ref was created");
|
||
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "# service=git-receive-pack");
|
||
assert!(lines[1].is_empty(), "the prefix section ends at flush");
|
||
// The sha256 passthrough rider rides the feature (OQ-05), exactly
|
||
// as on the V2 advertisement.
|
||
#[allow(unused_variables)]
|
||
let rider = if cfg!(feature = "sha256") {
|
||
" object-format=sha256"
|
||
} else {
|
||
""
|
||
};
|
||
assert_eq!(
|
||
text(&lines[2]),
|
||
format!("{} refs/heads/main\0report-status report-status-v2 delete-refs side-band-64k atomic ofs-delta object-format=sha1{rider} agent=alkgit/{}", hex_oid(1), env!("CARGO_PKG_VERSION")),
|
||
"caps NUL-attached on the FIRST ref line (the served set, agent=alkgit/<version>)"
|
||
);
|
||
assert!(lines[3].is_empty(), "the advertisement ends at flush");
|
||
assert_eq!(lines.len(), 4);
|
||
assert!(
|
||
!sink.body().windows(4).any(|w| w == b"0002"),
|
||
"no 0002 anywhere in the response body"
|
||
);
|
||
}
|
||
|
||
/// An empty repo (no refs, no unborn-HEAD row) advertises exactly the
|
||
/// `capabilities^{}` zero-id sentinel line — never a bare capability
|
||
/// dump.
|
||
///
|
||
/// Contract stamp: ADR-013 §2 (the sentinel rule);
|
||
/// push-captures.md §Advertisement; core/w3-v0-advertisement.
|
||
#[tokio::test]
|
||
async fn the_receive_pack_advertisement_for_an_empty_repo_emits_the_sentinel() {
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
assert_eq!(text(&lines[0]), "# service=git-receive-pack");
|
||
assert!(lines[1].is_empty(), "the prefix section ends at flush");
|
||
#[allow(unused_variables)]
|
||
let rider = if cfg!(feature = "sha256") {
|
||
" object-format=sha256"
|
||
} else {
|
||
""
|
||
};
|
||
assert_eq!(
|
||
text(&lines[2]),
|
||
format!(
|
||
"{} capabilities^{{}}\0report-status report-status-v2 delete-refs side-band-64k atomic ofs-delta object-format=sha1{rider} agent=alkgit/{}",
|
||
"0".repeat(40),
|
||
env!("CARGO_PKG_VERSION"),
|
||
),
|
||
"the sentinel line carries the caps"
|
||
);
|
||
assert!(lines[3].is_empty(), "the advertisement ends at flush");
|
||
assert_eq!(lines.len(), 4);
|
||
}
|
||
|
||
/// The GET emission's `push-options` gate (ADR-013 §11, the assembler's
|
||
/// config; default off in v1): the token appears in the advertised text
|
||
/// only under the gate flag — additive to open later as a value change.
|
||
///
|
||
/// Contract stamp: ADR-013 §2, §11 (the gate-gated token);
|
||
/// core/w3-v0-advertisement.
|
||
#[tokio::test]
|
||
async fn the_push_options_token_rides_the_gate_only() {
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
|
||
let gated = {
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
Arc::clone(&refs),
|
||
&Limits::default(),
|
||
true,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
String::from_utf8_lossy(&sink.body()).into_owned()
|
||
};
|
||
assert!(
|
||
gated.contains("\0report-status"),
|
||
"caps ride the first ref line"
|
||
);
|
||
assert!(
|
||
gated.contains(" push-options "),
|
||
"the gate adds the push-options token: {gated}"
|
||
);
|
||
|
||
let ungated = {
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
String::from_utf8_lossy(&sink.body()).into_owned()
|
||
};
|
||
assert!(
|
||
!ungated.contains("push-options"),
|
||
"push-options is absent by default (v1's gate is off): {ungated}"
|
||
);
|
||
}
|
||
|
||
/// The V0 emission projects the seam rows to oid+name: the unborn-HEAD
|
||
/// symref row (oid: None) is filtered — a V0 advertisement line has no
|
||
/// symref shape.
|
||
///
|
||
/// Contract stamp: ADR-013 §2 (the V0 ref-line shape);
|
||
/// push-captures.md §Advertisement; core/w3-v0-advertisement (the
|
||
/// filtered_head posture; the OQ-19 rider (a) covers the ls-refs side).
|
||
#[tokio::test]
|
||
async fn the_unborn_head_row_is_filtered_from_the_push_advertisement() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
fx.refs
|
||
.set_head_symref(&record, Some("refs/heads/main".to_owned()));
|
||
let outcomes = fx
|
||
.refs
|
||
.apply_updates(
|
||
&record,
|
||
vec![RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(oid(1)),
|
||
}],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(outcomes.len(), 1, "the fixture's ref was created");
|
||
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let sink = VecSink::default();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
|
||
let lines = parse_pkt_lines(&sink.body());
|
||
let body = String::from_utf8_lossy(&sink.body()).into_owned();
|
||
assert!(
|
||
!body.contains("symref-target"),
|
||
"a V0 advertisement line has no symref shape — the unborn row is filtered: {body}"
|
||
);
|
||
assert!(
|
||
!lines
|
||
.iter()
|
||
.any(|line| line.len() >= 4 && String::from_utf8_lossy(&line[4..]).starts_with("HEAD")),
|
||
"the no-oid HEAD row cannot be advertised: {body}"
|
||
);
|
||
}
|
||
|
||
/// The entries' marker is the type-level ACL ordering: a denied check
|
||
/// yields the `None` the door collapses to its generic denial before
|
||
/// any entry call — no unchecked session value can exist to feed the
|
||
/// entries.
|
||
///
|
||
/// 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); ADR-008/N-2 (the collapse rule is
|
||
/// the door's, before the marker exists);
|
||
/// core/w2-stateless-serving (the entries never resolve or authorize).
|
||
#[test]
|
||
fn only_the_checked_marker_path_feeds_the_entries() {
|
||
let mut private = repo("owner/doubled");
|
||
private
|
||
.grants
|
||
.insert("u-7731".to_string(), vec![AccessAction::Read]);
|
||
let holder = alkcall::core::Identity {
|
||
id: "u-7731".to_string(),
|
||
scopes: Vec::new(),
|
||
resources: std::collections::HashMap::new(),
|
||
};
|
||
assert_eq!(
|
||
AuthorizedRepo::authorize(private.clone(), None, AccessAction::Read),
|
||
None,
|
||
"anonymous on a private repo is no marker"
|
||
);
|
||
assert!(
|
||
AuthorizedRepo::authorize(private, Some(&holder), AccessAction::Read).is_some(),
|
||
"the granted identity is the entry's input"
|
||
);
|
||
}
|
||
|
||
/// A backend listing failure surfaces as the typed session error —
|
||
/// never a partial ref list.
|
||
///
|
||
/// Contract stamp: ADR-005 §4 (backend failures band-map through the
|
||
/// entries); review 001 N-1 (fail closed — a partial ref list is the
|
||
/// worst advertisement failure mode); core/w2-stateless-serving.
|
||
#[tokio::test]
|
||
async fn a_backend_listing_failure_serves_nothing() {
|
||
struct ListingFailure;
|
||
|
||
#[async_trait::async_trait]
|
||
impl GitRefs for ListingFailure {
|
||
async fn list_refs(&self, _repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
|
||
Err(StorageError::Invalid("listing unavailable".to_string()))
|
||
}
|
||
|
||
async fn apply_updates(
|
||
&self,
|
||
_repo: &RepoRecord,
|
||
_updates: Vec<RefUpdate>,
|
||
_atomic: bool,
|
||
) -> Result<Vec<RefOutcome>, StorageError> {
|
||
Err(StorageError::Invalid("listing unavailable".to_string()))
|
||
}
|
||
}
|
||
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = Arc::new(ListingFailure);
|
||
let sink = VecSink::default();
|
||
let result = serve_stateless_post(
|
||
&fx.marker,
|
||
body_parts(vec![command_post_body(b"command=ls-refs\n", &[], &[])]),
|
||
sink.clone(),
|
||
Limits::default(),
|
||
Service::UploadPack,
|
||
refs,
|
||
Arc::clone(&fx.gen),
|
||
fx.ingest.clone() as Arc<dyn GitPackIngest>,
|
||
false,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
let error = result.expect_err("a listing failure must not serve");
|
||
assert!(
|
||
error.is_terminal(),
|
||
"backend failures band-map as terminal: {error:?}"
|
||
);
|
||
assert!(sink.body().is_empty(), "a failed listing wrote nothing");
|
||
}
|
||
|
||
/// The GET emission's one backend call is under the per-request wall
|
||
/// clock: a hung listing times out into the typed breach error — the
|
||
/// advertisement route is never the session surface without a budget
|
||
/// (the POST entry's wall-clock discipline, mirrored).
|
||
///
|
||
/// Contract stamp: ADR-009 (the wall clock is a budget of every
|
||
/// session/request surface; fail-closed breach behavior); ADR-005 §4
|
||
/// (the typed failure maps at the door).
|
||
#[tokio::test]
|
||
async fn the_advertisement_listing_hangs_until_the_wall_clock_not_beyond() {
|
||
struct HangingListing;
|
||
|
||
#[async_trait::async_trait]
|
||
impl GitRefs for HangingListing {
|
||
async fn list_refs(&self, _repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
|
||
futures::future::pending().await
|
||
}
|
||
|
||
async fn apply_updates(
|
||
&self,
|
||
_repo: &RepoRecord,
|
||
_updates: Vec<RefUpdate>,
|
||
_atomic: bool,
|
||
) -> Result<Vec<RefOutcome>, StorageError> {
|
||
futures::future::pending().await
|
||
}
|
||
}
|
||
|
||
let fx = fixture();
|
||
let refs: Arc<dyn GitRefs> = Arc::new(HangingListing);
|
||
let sink = VecSink::default();
|
||
let limits = Limits {
|
||
session_wall_clock: std::time::Duration::from_millis(100),
|
||
..Limits::default()
|
||
};
|
||
let started = std::time::Instant::now();
|
||
let result = serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&limits,
|
||
false,
|
||
)
|
||
.await;
|
||
assert!(
|
||
started.elapsed() < std::time::Duration::from_secs(5),
|
||
"the listing did not hang past the wall clock"
|
||
);
|
||
let error = result.expect_err("a hung listing breaches the wall clock");
|
||
match &error {
|
||
SubstrateError::Protocol(message) => {
|
||
assert!(message.contains("session_wall_clock"), "{message}");
|
||
}
|
||
other => panic!("a wall-clock breach is protocol-shaped: {other:?}"),
|
||
}
|
||
assert!(sink.body().is_empty(), "a breached emission wrote nothing");
|
||
}
|
||
|
||
/// A prepare double that counts its invocations — the observable for
|
||
/// the packless path (no prepare call) and the probe (no dispatch).
|
||
struct CountedIngest {
|
||
inner: Arc<MemPackIngest>,
|
||
runs: Arc<AtomicUsize>,
|
||
}
|
||
|
||
fn counting_ingest(inner: Arc<MemPackIngest>, runs: Arc<AtomicUsize>) -> Arc<dyn GitPackIngest> {
|
||
Arc::new(CountedIngest { inner, runs })
|
||
}
|
||
|
||
#[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> {
|
||
self.runs.fetch_add(1, AtomicOrdering::SeqCst);
|
||
self.inner
|
||
.prepare(record, updates, pack, push_options, limits)
|
||
.await
|
||
}
|
||
}
|
||
|
||
/// A prepare double that records the push-options it received — the
|
||
/// observable for the V0 POST body's options-section routing.
|
||
struct OptionsRecorder {
|
||
inner: Arc<MemPackIngest>,
|
||
seen: Arc<StdMutex<Option<PushOptions>>>,
|
||
}
|
||
|
||
#[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 one POST / one stream push flow over the entry (ADR-013 §3): an
|
||
/// ordinary push POST — two create commands with the caps
|
||
/// NUL-attached on the first, the command-section flush, then the pack
|
||
/// tail raw — dribbled in chunks across the body. The commands + tail
|
||
/// ride the push machine under the stateless tail discipline; the
|
||
/// report streams back sideband-framed (exactly one outer flush — the
|
||
/// one-shot-flush guard holds), the refs row-table reflects the CAS,
|
||
/// and the pack's objects land via the ingest handle.
|
||
///
|
||
/// Contract stamp: ADR-013 §3 (one POST / one stream, the raw tail),
|
||
/// §7 (the one-transaction CAS home), §8 (the sideband report,
|
||
/// request order), §12 (the http framing — the substrate sees bytes
|
||
/// in/out); ADR-020 §1 (the stateless EOF twin); ADR-005 §2 (the
|
||
/// response ends at the flush, the one-shot-flush guard, no 0002);
|
||
/// ADR-009 (the pack tail counts against max_pack_size in the relay);
|
||
/// core/w3-stateless-serving (the entry's ReceivePack branch serves
|
||
/// the machine).
|
||
#[tokio::test]
|
||
async fn an_ordinary_push_post_applies_and_reports_sideband_framed() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let chain = chain_pack();
|
||
let commands = [
|
||
create_command("refs/heads/main", chain.commit_b),
|
||
create_command("refs/heads/dev", chain.commit_a),
|
||
];
|
||
|
||
let body = body_parts(
|
||
push_commands("report-status side-band-64k", &commands)
|
||
.into_iter()
|
||
.chain([
|
||
chain.pack[..1].to_vec(),
|
||
chain.pack[1..24].to_vec(),
|
||
chain.pack[24..].to_vec(),
|
||
])
|
||
.collect::<Vec<Vec<u8>>>(),
|
||
);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::ReceivePack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
let packets = parse_pkt_lines(&sink.body());
|
||
assert_eq!(
|
||
sideband_report_lines(&sink.body()),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
"ok refs/heads/main".to_owned(),
|
||
"ok refs/heads/dev".to_owned(),
|
||
]
|
||
);
|
||
assert!(
|
||
!sink.body().windows(4).any(|w| w == b"0002"),
|
||
"no 0002 anywhere in the response body"
|
||
);
|
||
assert!(packets.last().unwrap().is_empty(), "one flush ends it");
|
||
assert_eq!(
|
||
fx.refs
|
||
.list_refs(&record)
|
||
.await
|
||
.unwrap()
|
||
.iter()
|
||
.map(|line| (line.name.clone(), line.oid))
|
||
.collect::<BTreeSet<_>>(),
|
||
BTreeSet::from([
|
||
("refs/heads/main".to_owned(), Some(chain.commit_b)),
|
||
("refs/heads/dev".to_owned(), Some(chain.commit_a)),
|
||
]),
|
||
"the CAS applied the pushed updates"
|
||
);
|
||
assert!(
|
||
fx.ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pushed objects landed via the ingest handle"
|
||
);
|
||
}
|
||
|
||
/// The packless delete POST (ADR-020 §1's EOF-driven twin): the
|
||
/// command section's flush ends the body — the tail's EOF at its start
|
||
/// takes the packless, CAS-only path (no prepare call), and the same
|
||
/// sideband report shape the duplex capture pinned streams back.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 (rule (ii)'s EOF mapping — zero tail
|
||
/// bytes = the packless CAS-only path; the report shape equals the
|
||
/// duplex capture); ADR-013 §3, §7–8; core/w3-stateless-serving.
|
||
#[tokio::test]
|
||
async fn a_packless_delete_post_takes_the_eof_driven_packless_path() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let current = oid(0x1b);
|
||
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(AtomicUsize::new(0));
|
||
let ingest = counting_ingest(Arc::clone(&fx.ingest), Arc::clone(&runs));
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let body = body_parts(vec![push_commands(
|
||
"report-status side-band-64k",
|
||
&[delete_command("refs/heads/gone", current)],
|
||
)
|
||
.concat()]);
|
||
let (result, sink) = fx
|
||
.serve_push(
|
||
body,
|
||
Limits::default(),
|
||
false,
|
||
refs,
|
||
ingest,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
assert_eq!(
|
||
sideband_report_lines(&sink.body()),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
|
||
"the same report shape the duplex capture pinned"
|
||
);
|
||
assert_eq!(
|
||
runs.load(AtomicOrdering::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 deleted the ref"
|
||
);
|
||
}
|
||
|
||
/// The gate-on push-options section is consumed before the pack tail
|
||
/// (ADR-020 §1 rule (i)): a push POST whose body carries commands +
|
||
/// flush + the options section + flush + the pack parses the options
|
||
/// through to `prepare` and the pack still ingests byte-exact — the
|
||
/// escape sits byte-exact past the *second* flush.
|
||
///
|
||
/// Contract stamp: ADR-013 §11 (the section's grammar — bare
|
||
/// pkt-lines, its flush, the negotiated gate); ADR-020 §1 rule (i)
|
||
/// (consumed before the peek); ADR-005 §2 (the body's framing);
|
||
/// core/w3-stateless-serving (the reader's section-two decision rides
|
||
/// the client's capability selection).
|
||
#[tokio::test]
|
||
async fn the_negotiated_push_options_section_is_consumed_before_the_pack_tail() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let chain = chain_pack();
|
||
|
||
let seen = Arc::new(StdMutex::new(None));
|
||
let ingest = Arc::new(OptionsRecorder {
|
||
inner: Arc::clone(&fx.ingest),
|
||
seen: Arc::clone(&seen),
|
||
}) as Arc<dyn GitPackIngest>;
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
|
||
let body = body_parts(
|
||
push_commands(
|
||
"report-status push-options side-band-64k",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
)
|
||
.into_iter()
|
||
.chain(vec![
|
||
pkt(b"deploy-key=abc"),
|
||
FLUSH.to_vec(),
|
||
chain.pack.clone(),
|
||
])
|
||
.collect::<Vec<Vec<u8>>>(),
|
||
);
|
||
let (result, sink) = fx
|
||
.serve_push(
|
||
body,
|
||
Limits::default(),
|
||
true,
|
||
refs,
|
||
ingest,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
|
||
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"
|
||
);
|
||
assert_eq!(
|
||
sideband_report_lines(&sink.body()),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
);
|
||
assert!(
|
||
fx.ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pack after the options flush ingested byte-exact"
|
||
);
|
||
}
|
||
|
||
/// The request-body budget fails closed on the push POST's framed
|
||
/// section (the client-fault class, 413-mapped at the door): a command
|
||
/// list beyond the budget errors before any backend call, nothing
|
||
/// written.
|
||
///
|
||
/// Contract stamp: ADR-009 (max_request_body's breach behavior — the
|
||
/// client-fault budget class); ADR-013 §3 (the framed command
|
||
/// sections are the body budget's scope; the pack tail counts against
|
||
/// max_pack_size in the relay); core/w3-stateless-serving.
|
||
#[tokio::test]
|
||
async fn a_push_post_body_budget_breach_fails_closed() {
|
||
let fx = fixture();
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = counting_ingest(Arc::clone(&fx.ingest), Arc::clone(&runs));
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
let limits = Limits {
|
||
max_request_body: 5,
|
||
..Limits::default()
|
||
};
|
||
let (result, sink) = fx
|
||
.serve_push(
|
||
body_parts(vec![push_commands(
|
||
"report-status",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
)
|
||
.concat()]),
|
||
limits,
|
||
false,
|
||
refs,
|
||
ingest,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
assert_protocol_error(result, "max_request_body").await;
|
||
assert!(sink.body().is_empty(), "the breach wrote nothing");
|
||
assert_eq!(
|
||
runs.load(AtomicOrdering::SeqCst),
|
||
0,
|
||
"the breach failed before any backend call"
|
||
);
|
||
}
|
||
|
||
/// The push tail's pack-size budget fails closed (ADR-009's
|
||
/// max_pack_size in the relay — the client-fault class, 413-mapped at
|
||
/// the door): a pack beyond the budget errors the session before any
|
||
/// report is written, no ref applies.
|
||
///
|
||
/// Contract stamp: ADR-009 (the pack-size breach — the client-fault
|
||
/// budget class, fail-closed); ADR-013 §12; ADR-005 §2 (the pack is
|
||
/// raw in the same body — the relay counts its bytes);
|
||
/// core/w3-stateless-serving.
|
||
#[tokio::test]
|
||
async fn a_push_post_pack_size_breach_fails_closed() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let limits = Limits {
|
||
max_pack_size: 1000,
|
||
..Limits::default()
|
||
};
|
||
let oversized: Vec<u8> = (0..10_000u32).map(|i| (i % 251) as u8).collect();
|
||
let body = body_parts(
|
||
push_commands(
|
||
"report-status side-band-64k",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
)
|
||
.into_iter()
|
||
.chain(oversized.chunks(300).map(|c| c.to_vec()))
|
||
.collect::<Vec<Vec<u8>>>(),
|
||
);
|
||
let (result, sink) = fx
|
||
.serve(body, limits, Service::ReceivePack, default_pipeline())
|
||
.await;
|
||
assert_protocol_error(result, "max_pack_size").await;
|
||
assert!(sink.body().is_empty(), "no report over the broken stream");
|
||
assert!(
|
||
fx.refs.list_refs(&record).await.unwrap().is_empty(),
|
||
"the breach applied no ref"
|
||
);
|
||
}
|
||
|
||
/// The door-facing push pair over the entry: the advertisement GET
|
||
/// serves the V0 emission (smart prefix + flush + the ref
|
||
/// advertisement + flush), the POST — one body, commands + flush +
|
||
/// pack — applies and reports, and the next advertisement GET lists
|
||
/// the pushed ref: the stateless composition real clients cache and
|
||
/// re-derive per request, with no cross-request state.
|
||
///
|
||
/// Contract stamp: ADR-013 §1–2 (the V0 advertisement), §3, §7–8 (the
|
||
/// request + CAS + report), §12 (the http framing facts — the
|
||
/// Content-Types are the door's mapping, the substrate sees bytes);
|
||
/// ADR-005 §2 (the stateless response rules); ADR-007 (the marker
|
||
/// gates both emissions); core/w3-stateless-serving (the entry-level
|
||
/// composition the task verifies).
|
||
#[tokio::test]
|
||
async fn the_advertisement_get_and_push_post_pair_serve_the_v0_flow() {
|
||
let fx = fixture();
|
||
let record = fx.record();
|
||
let chain = chain_pack();
|
||
|
||
async fn advert(fx: &Fixture) -> Vec<String> {
|
||
let sink = VecSink::default();
|
||
let refs: Arc<dyn GitRefs> = fx.refs.clone();
|
||
serve_stateless_advertisement(
|
||
&fx.marker,
|
||
sink.clone(),
|
||
Service::ReceivePack,
|
||
refs,
|
||
&Limits::default(),
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
parse_pkt_lines(&sink.body())
|
||
.iter()
|
||
.map(|p| text(p))
|
||
.collect()
|
||
}
|
||
|
||
let before = advert(&fx).await;
|
||
assert_eq!(
|
||
*before.last().expect("the advertisement is non-empty"),
|
||
String::new(),
|
||
"the advertisement ends at flush"
|
||
);
|
||
assert!(
|
||
before[2].contains("capabilities^{"),
|
||
"the empty repo's sentinel line: {:?}",
|
||
before[2]
|
||
);
|
||
|
||
let body = body_parts(
|
||
push_commands(
|
||
"report-status side-band-64k",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
)
|
||
.into_iter()
|
||
.chain([chain.pack.clone()])
|
||
.collect::<Vec<Vec<u8>>>(),
|
||
);
|
||
let (result, sink) = fx
|
||
.serve(
|
||
body,
|
||
Limits::default(),
|
||
Service::ReceivePack,
|
||
default_pipeline(),
|
||
)
|
||
.await;
|
||
result.unwrap();
|
||
assert_eq!(
|
||
sideband_report_lines(&sink.body()),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
);
|
||
assert_eq!(
|
||
fx.refs
|
||
.list_refs(&record)
|
||
.await
|
||
.unwrap()
|
||
.iter()
|
||
.find(|line| line.name == "refs/heads/main")
|
||
.and_then(|line| line.oid),
|
||
Some(chain.commit_b),
|
||
"the CAS applied the pushed update"
|
||
);
|
||
|
||
let mut after = advert(&fx).await;
|
||
assert_eq!(
|
||
after.pop(),
|
||
Some(String::new()),
|
||
"the advertisement ends at flush"
|
||
);
|
||
assert_eq!(
|
||
after.len(),
|
||
3,
|
||
"the prefix flush + one ref line follow the prefix section"
|
||
);
|
||
let pushed_line = format!("{} refs/heads/main", chain.commit_b);
|
||
assert!(
|
||
after[2]
|
||
.split('\0')
|
||
.next()
|
||
.expect("the ref line has its caps section")
|
||
.starts_with(&pushed_line),
|
||
"the pushed ref appears in the next advertisement: {:?}",
|
||
after[2]
|
||
);
|
||
}
|