glm-5.3-flash
9f55302c84
- read_post dispatches on the service: the receive-pack POST body parses V0-framed (command lines, caps NUL-attached on the first, the command-section flush; the gate-on push-options section consumed before the tail per ADR-020 §1 rule (i), decided from the client's capability selection) — replacing the wave-3 typed decline - the ReceivePack branch escapes the raw pack tail and runs serve_receive_pack under PackTailDiscipline::Stateless; the report rides the response's one-shot-flush rules - the entry gains the ingest handle + the push-options negotiated-gate flag (additive pre-publish); budgets: framed sections against max_request_body, the tail against max_pack_size in the relay, the wall clock per POST - RawTail::new went pub (additive pre-publish; the machine-level test compositions compose tails by it) - the machine-level stand-in rows re-based onto the real V0 body shape; e2e rows over the entry: ordinary push, packless delete, probe/nothing-to-do, options routing, both budget breaches, and the advertisement GET -> push POST -> advertisement GET pair Verification: cargo test -p alkgit (pass, incl. --all-features), cargo clippy --all-targets --all-features -- -D warnings (clean), cargo fmt --check (clean), taskgraph validate (39 valid)
1759 lines
59 KiB
Rust
1759 lines
59 KiB
Rust
//! The wave-3 receive-pack state machine rows (core/w3-push-machine):
|
||
//! parsed request → prepare → validate → CAS → report per ADR-013
|
||
//! §3–11 as amended by ADR-020 §1 — the duplex peek's four ordering
|
||
//! rules, the stateless EOF twin, the packless CAS-only path, the
|
||
//! report's shapes byte-against push-captures.md, and the budgets
|
||
//! (the permit around prepare, the pack-size relay, fail-closed
|
||
//! breaches) — over the in-memory backend doubles.
|
||
//!
|
||
//! Rows are names + `Contract stamp:` versions, per
|
||
//! `tests/common` §The row/stamp discipline.
|
||
|
||
mod common;
|
||
|
||
use std::collections::BTreeMap;
|
||
use std::io;
|
||
use std::pin::Pin;
|
||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||
use std::sync::{Arc, Mutex as StdMutex};
|
||
use std::task::{Context, Poll};
|
||
use std::time::Duration;
|
||
|
||
use async_trait::async_trait;
|
||
use common::{
|
||
blob_fixture, commit_fixture, loose_body, pack_bytes, pack_entry_base, tree_fixture,
|
||
MemPackIngest, MemRefs,
|
||
};
|
||
use futures::io::AsyncWrite as FuturesAsyncWrite;
|
||
use futures::stream;
|
||
use gix_hash::ObjectId;
|
||
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
|
||
use tokio::sync::Semaphore;
|
||
|
||
use alkgit::{
|
||
serve_receive_pack, BodyReader, GitPackIngest, GitRefs, Limits, PackTailDiscipline,
|
||
PreparedPush, PushOptions, RawTail, RefUpdate, RepoRecord, Request, Service, Session,
|
||
StatelessPost, StatelessRequest, SubstrateError, Visibility, ATOMIC_PUSH_FAILURE_REASON,
|
||
UNPACKER_ERROR_REASON,
|
||
};
|
||
|
||
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 oid(b: u8) -> ObjectId {
|
||
ObjectId::from([b; 20])
|
||
}
|
||
|
||
/// 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 create command line (`old = zero`, ADR-013 §3's shapes).
|
||
fn create_command(name: &str, new: ObjectId) -> String {
|
||
format!("{} {new} {name}", oid(0))
|
||
}
|
||
|
||
/// A delete command line (`new = zero`).
|
||
fn delete_command(name: &str, current: ObjectId) -> String {
|
||
format!("{current} {} {name}", oid(0))
|
||
}
|
||
|
||
/// An update command line (`old = current`, `new = target`).
|
||
fn update_command(name: &str, current: ObjectId, new: ObjectId) -> String {
|
||
format!("{current} {new} {name}")
|
||
}
|
||
|
||
/// A fixture object: its oid and the (kind, body) its loose bytes
|
||
/// carry — the pack-entry source.
|
||
struct Fixture {
|
||
oid: ObjectId,
|
||
kind: gix_object::Kind,
|
||
body: Vec<u8>,
|
||
}
|
||
|
||
fn blob_fix(content: &[u8]) -> Fixture {
|
||
let (id, bytes) = blob_fixture(content);
|
||
let (kind, body) = loose_body(&bytes);
|
||
Fixture {
|
||
oid: id,
|
||
kind,
|
||
body,
|
||
}
|
||
}
|
||
|
||
fn tree_fix(entries: &[(gix_object::tree::EntryKind, &str, ObjectId)]) -> Fixture {
|
||
let (id, bytes) = tree_fixture(entries);
|
||
let (kind, body) = loose_body(&bytes);
|
||
Fixture {
|
||
oid: id,
|
||
kind,
|
||
body,
|
||
}
|
||
}
|
||
|
||
fn commit_fix(tree: ObjectId, parents: &[ObjectId], message: &str) -> Fixture {
|
||
let (id, bytes) = commit_fixture(tree, parents, message);
|
||
let (kind, body) = loose_body(&bytes);
|
||
Fixture {
|
||
oid: id,
|
||
kind,
|
||
body,
|
||
}
|
||
}
|
||
|
||
/// Assembles a self-contained V2 sha1 pack from the fixtures
|
||
/// (cumulative pack offsets make `pack_bytes` whole).
|
||
fn pack_from_fixtures(fixtures: &[&Fixture]) -> Vec<u8> {
|
||
let mut offset = 0u64;
|
||
let entries: Vec<_> = fixtures
|
||
.iter()
|
||
.map(|fixture| {
|
||
let entry = pack_entry_base(fixture.kind, &fixture.body, offset);
|
||
offset += entry.bytes_in_pack();
|
||
entry
|
||
})
|
||
.collect();
|
||
pack_bytes(&entries)
|
||
}
|
||
|
||
/// A two-commit chain fixture + its pack — the rows' ordinary push.
|
||
struct ChainPack {
|
||
pack: Vec<u8>,
|
||
commit_a: ObjectId,
|
||
commit_b: ObjectId,
|
||
}
|
||
|
||
fn chain_pack() -> ChainPack {
|
||
let blob_a = blob_fix(b"alpha");
|
||
let blob_b = blob_fix(b"beta");
|
||
let tree_a = tree_fix(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid)]);
|
||
let tree_b = tree_fix(&[
|
||
(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid),
|
||
(gix_object::tree::EntryKind::Blob, "b.txt", blob_b.oid),
|
||
]);
|
||
let commit_a = commit_fix(tree_a.oid, &[], "alpha");
|
||
let commit_b = commit_fix(tree_b.oid, &[commit_a.oid], "beta");
|
||
let pack = pack_from_fixtures(&[&blob_a, &blob_b, &tree_a, &tree_b, &commit_a, &commit_b]);
|
||
ChainPack {
|
||
pack,
|
||
commit_a: commit_a.oid,
|
||
commit_b: commit_b.oid,
|
||
}
|
||
}
|
||
|
||
/// The duplex machine-level request bytes: the V2-framed request the
|
||
/// duplex reader consumes until `Session` gains its V0 read shape
|
||
/// (core/w3-duplex-serving re-baselines this onto the real V0 stream
|
||
/// shape) — the command section with the caps NUL-attached on the first
|
||
/// command line, the command-section flush, and — when args are present
|
||
/// (the negotiated push-options gate) — the args section behind the
|
||
/// delimiter, the mapping the push grammar consumes.
|
||
fn request_bytes(caps: &str, commands: &[String], args: &[String]) -> Vec<u8> {
|
||
let mut out = pkt(b"command=git-receive-pack\n");
|
||
let mut commands = commands.iter();
|
||
if let Some(first) = commands.next() {
|
||
let line = if caps.is_empty() {
|
||
first.clone()
|
||
} else {
|
||
format!("{first}\0{caps}")
|
||
};
|
||
out.extend_from_slice(&pkt(line.as_bytes()));
|
||
}
|
||
for command in commands {
|
||
out.extend_from_slice(&pkt(command.as_bytes()));
|
||
}
|
||
if args.is_empty() {
|
||
out.extend_from_slice(FLUSH);
|
||
} else {
|
||
out.extend_from_slice(DELIMITER);
|
||
for arg in args {
|
||
out.extend_from_slice(&pkt(arg.as_bytes()));
|
||
}
|
||
out.extend_from_slice(FLUSH);
|
||
}
|
||
out
|
||
}
|
||
|
||
/// The stateless POST body chunks — the real V0 shape
|
||
/// (push-captures.md §Request; ADR-013 §3): the command section with
|
||
/// the caps NUL-attached on the first command line, the
|
||
/// command-section flush, and the pack tail's chunks after it (raw,
|
||
/// ADR-013 §3). Zero commands on this wire shape is the immediate
|
||
/// flush — the nothing-to-do/probe body.
|
||
fn body_chunks(caps: &str, commands: &[String], pack_chunks: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
|
||
let mut chunks = Vec::new();
|
||
let mut commands = commands.iter();
|
||
if let Some(first) = commands.next() {
|
||
let line = if caps.is_empty() {
|
||
first.clone()
|
||
} else {
|
||
format!("{first}\0{caps}")
|
||
};
|
||
chunks.push(pkt(line.as_bytes()));
|
||
}
|
||
for command in commands {
|
||
chunks.push(pkt(command.as_bytes()));
|
||
}
|
||
chunks.push(FLUSH.to_vec());
|
||
chunks.extend(pack_chunks);
|
||
chunks
|
||
}
|
||
|
||
type ChunkStream = stream::Iter<std::vec::IntoIter<io::Result<bytes::Bytes>>>;
|
||
type PostBody = BodyReader<ChunkStream, io::Error>;
|
||
|
||
fn body_reader(chunks: Vec<Vec<u8>>) -> PostBody {
|
||
let items: Vec<io::Result<bytes::Bytes>> = chunks
|
||
.into_iter()
|
||
.map(|chunk| io::Result::Ok(bytes::Bytes::from(chunk)))
|
||
.collect();
|
||
BodyReader::new(stream::iter(items))
|
||
}
|
||
|
||
/// An async sink whose framed bytes the stateless 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<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<io::Result<()>> {
|
||
Poll::Ready(Ok(()))
|
||
}
|
||
|
||
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||
self.poll_flush(cx)
|
||
}
|
||
}
|
||
|
||
/// Pkt-line payloads of a framed byte string.
|
||
fn parse_pkt_lines(body: &[u8]) -> Vec<Vec<u8>> {
|
||
let mut lines = Vec::new();
|
||
let mut cursor = body;
|
||
while !cursor.is_empty() {
|
||
let (header, rest) = cursor.split_at(4);
|
||
let len = usize::from_str_radix(std::str::from_utf8(header).expect("hex header"), 16)
|
||
.expect("hex length");
|
||
cursor = rest;
|
||
if len < 4 {
|
||
lines.push(Vec::new());
|
||
continue;
|
||
}
|
||
let (payload, rest) = cursor.split_at(len - 4);
|
||
cursor = rest;
|
||
lines.push(payload.to_vec());
|
||
}
|
||
lines
|
||
}
|
||
|
||
fn text(payload: &[u8]) -> String {
|
||
String::from_utf8_lossy(payload)
|
||
.trim_end_matches('\n')
|
||
.to_owned()
|
||
}
|
||
|
||
fn payloads_as_text(frames: &[Vec<u8>]) -> Vec<String> {
|
||
frames.iter().map(|frame| text(frame)).collect()
|
||
}
|
||
|
||
/// The sideband-wrapped report's inner pkt-line lines (ADR-013 §8):
|
||
/// asserts the framing itself — every data packet is band-1, the
|
||
/// response ends at exactly one outer flush, the band payload's own
|
||
/// framing ends at its inner flush — and returns the report's lines.
|
||
fn sideband_report_lines(packets: &[Vec<u8>]) -> Vec<String> {
|
||
let mut payload = Vec::new();
|
||
let mut outer_flushes = 0;
|
||
for packet in packets {
|
||
if packet.is_empty() {
|
||
outer_flushes += 1;
|
||
continue;
|
||
}
|
||
assert_eq!(packet[0], 1, "band-1 framing carries the report");
|
||
payload.extend_from_slice(&packet[1..]);
|
||
}
|
||
assert_eq!(outer_flushes, 1, "exactly one outer flush");
|
||
let mut inner = parse_pkt_lines(&payload);
|
||
let last = inner.pop().expect("the inner framing is non-empty");
|
||
assert!(
|
||
last.is_empty(),
|
||
"the inner flush terminates the report: {last:?}"
|
||
);
|
||
payloads_as_text(&inner)
|
||
}
|
||
|
||
/// The bare report's lines (no sideband was selected): the report's
|
||
/// lines, the terminating flush asserted to be the last packet.
|
||
fn bare_report_lines(packets: &[Vec<u8>]) -> Vec<String> {
|
||
let (last, rest) = packets.split_last().expect("the report is non-empty");
|
||
assert!(last.is_empty(), "the report ends at the flush: {last:?}");
|
||
payloads_as_text(rest)
|
||
}
|
||
|
||
/// Reads pkt-lines from the client end until (and including) a flush.
|
||
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());
|
||
return packets;
|
||
}
|
||
let mut payload = vec![0u8; len - 4];
|
||
reader
|
||
.read_exact(&mut payload)
|
||
.await
|
||
.expect("pkt-line payload");
|
||
packets.push(payload);
|
||
}
|
||
}
|
||
|
||
/// Limits with a short peek window (rows run fast; the 750 ms default
|
||
/// is asserted in `limits.rs`).
|
||
fn fast_limits() -> Limits {
|
||
Limits {
|
||
packless_peek: Duration::from_millis(100),
|
||
..Limits::default()
|
||
}
|
||
}
|
||
|
||
/// Seeds one ref row into the refs double (row support for deletes and
|
||
/// stale-old denials).
|
||
async fn seed_ref(refs: &MemRefs, record: &RepoRecord, name: &str, value: ObjectId) {
|
||
refs.apply_updates(
|
||
record,
|
||
vec![RefUpdate {
|
||
name: name.to_owned(),
|
||
expected: None,
|
||
new: Some(value),
|
||
}],
|
||
false,
|
||
)
|
||
.await
|
||
.unwrap();
|
||
}
|
||
|
||
async fn list_ref(refs: &MemRefs, record: &RepoRecord, name: &str) -> Option<ObjectId> {
|
||
refs.list_refs(record)
|
||
.await
|
||
.unwrap()
|
||
.into_iter()
|
||
.find(|line| line.name == name)
|
||
.and_then(|line| line.oid)
|
||
}
|
||
|
||
/// Spawns the duplex machine — the session machinery built inside the
|
||
/// task over the client's stream. Rows where the client waits for the
|
||
/// report before closing (the packless drain's shape) join after
|
||
/// reading + dropping; every other row writes the request (+ pack)
|
||
/// first and joins immediately.
|
||
fn spawn_duplex_machine(
|
||
stream: tokio::io::DuplexStream,
|
||
refs: Arc<dyn GitRefs>,
|
||
ingest: Arc<dyn GitPackIngest>,
|
||
record: RepoRecord,
|
||
limits: Limits,
|
||
push_options_negotiated: bool,
|
||
pipeline: Arc<Semaphore>,
|
||
) -> tokio::task::JoinHandle<Result<(), SubstrateError>> {
|
||
tokio::spawn(async move {
|
||
let mut session = Session::new(stream, limits, Service::ReceivePack);
|
||
let request = session
|
||
.read_request()
|
||
.await
|
||
.unwrap()
|
||
.expect("the invocation's request");
|
||
let tail = session.take_raw_tail().unwrap();
|
||
serve_receive_pack(
|
||
&mut session,
|
||
&request,
|
||
tail,
|
||
push_options_negotiated,
|
||
refs,
|
||
ingest,
|
||
&record,
|
||
&limits,
|
||
PackTailDiscipline::Duplex,
|
||
pipeline,
|
||
)
|
||
.await
|
||
})
|
||
}
|
||
|
||
/// Drives the stateless machine inline over one POST body (no packless
|
||
/// drain exists on this discipline — no client cooperation needed);
|
||
/// returns the outcome and the response sink.
|
||
async fn stateless_machine(
|
||
chunks: Vec<Vec<u8>>,
|
||
refs: Arc<dyn GitRefs>,
|
||
ingest: Arc<dyn GitPackIngest>,
|
||
record: &RepoRecord,
|
||
limits: Limits,
|
||
push_options_negotiated: bool,
|
||
) -> (Result<(), SubstrateError>, VecSink) {
|
||
let sink = VecSink::default();
|
||
let mut post = StatelessRequest::new(
|
||
body_reader(chunks),
|
||
sink.clone(),
|
||
limits,
|
||
Service::ReceivePack,
|
||
);
|
||
// The stateless POST body parses V0-framed (ADR-013 §3); the
|
||
// nothing-to-do body (an immediate flush) resolves to the probe at
|
||
// the substrate (ADR-005 §2's service-agnostic probe rule), so the
|
||
// machine-level nothing-to-do shape composes here directly: the
|
||
// empty request, no tail.
|
||
let (command, tail) = match post.read_post().await.expect("the POST parses") {
|
||
StatelessPost::Command(command) => (command, post.take_raw_tail().unwrap()),
|
||
StatelessPost::Probe => (
|
||
Request {
|
||
command: "git-receive-pack".to_owned(),
|
||
pre_lines: Vec::new(),
|
||
args: Vec::new(),
|
||
},
|
||
RawTail::new(body_reader(Vec::new()), Arc::new(AtomicU64::new(0))),
|
||
),
|
||
};
|
||
let outcome = serve_receive_pack(
|
||
post.response(),
|
||
&command,
|
||
tail,
|
||
push_options_negotiated,
|
||
refs,
|
||
ingest,
|
||
record,
|
||
&limits,
|
||
PackTailDiscipline::Stateless,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
)
|
||
.await;
|
||
(outcome, sink)
|
||
}
|
||
|
||
/// A prepare double that counts its invocations (rows asserting the
|
||
/// machine's call discipline: prepare runs under the permit; never on
|
||
/// the packless path).
|
||
struct PrepareSignal {
|
||
inner: MemPackIngest,
|
||
runs: Arc<AtomicUsize>,
|
||
}
|
||
|
||
impl PrepareSignal {
|
||
fn new(run_count: Arc<AtomicUsize>) -> Self {
|
||
PrepareSignal {
|
||
inner: MemPackIngest::default(),
|
||
runs: run_count,
|
||
}
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl GitPackIngest for PrepareSignal {
|
||
async fn prepare(
|
||
&self,
|
||
record: &RepoRecord,
|
||
updates: Vec<RefUpdate>,
|
||
pack: Box<dyn io::Read + Send>,
|
||
push_options: Option<PushOptions>,
|
||
limits: Limits,
|
||
) -> Result<PreparedPush, alkgit::StorageError> {
|
||
self.runs.fetch_add(1, Ordering::SeqCst);
|
||
self.inner
|
||
.prepare(record, updates, pack, push_options, limits)
|
||
.await
|
||
}
|
||
}
|
||
|
||
fn session_error(result: Result<(), SubstrateError>) -> SubstrateError {
|
||
match result {
|
||
Err(error) => error,
|
||
Ok(()) => panic!("expected a session error"),
|
||
}
|
||
}
|
||
|
||
fn protocol_message(error: SubstrateError) -> String {
|
||
let SubstrateError::Protocol(message) = error else {
|
||
panic!("expected a protocol-class session error: {error:?}");
|
||
};
|
||
message
|
||
}
|
||
|
||
/// An ordinary push (commands + pack, dribbled three ways) ingests via
|
||
/// `prepare` under the permit, validates the names, applies via the
|
||
/// one CAS call, and reports `unpack ok` + per-ref `ok` lines in
|
||
/// request order — sideband-framed (band-1 payload self-framed, the
|
||
/// inner flush inside the band, one outer flush last) against
|
||
/// push-captures.md §Status report; the pack's objects land in the
|
||
/// backend namespace and the refs row-table reflects the CAS.
|
||
///
|
||
/// Contract stamp: ADR-013 §3 (the request shape), §6–7 (prepare →
|
||
/// name → CAS — the one CAS home), §8 (the sideband report, request
|
||
/// order); ADR-020 §1 (the peek's full-header branch — the peeked
|
||
/// bytes prepend the bridged tail; the first pack byte rides its own
|
||
/// chunk, reassembled with zero loss); ADR-009 (the permit wraps the
|
||
/// prepare call); push-captures.md §Status report (the byte-level
|
||
/// basis); tests/common §MemPackIngest (the ingestion double).
|
||
#[tokio::test]
|
||
async fn ordinary_push_ingests_applies_and_reports_in_request_order() {
|
||
let chain = chain_pack();
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let mem_ingest = MemPackIngest::default();
|
||
let ingest = Arc::new(mem_ingest.clone()) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/main", chain.commit_b),
|
||
create_command("refs/heads/dev", chain.commit_a),
|
||
];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k",
|
||
&commands,
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack[..1]).await.unwrap();
|
||
client.write_all(&chain.pack[1..24]).await.unwrap();
|
||
client.write_all(&chain.pack[24..]).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
"ok refs/heads/main".to_owned(),
|
||
"ok refs/heads/dev".to_owned(),
|
||
],
|
||
"the report's lines in request order, band-1 framed with the inner flush"
|
||
);
|
||
drop(client);
|
||
let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.expect("the machine completes")
|
||
.unwrap();
|
||
assert!(outcome.is_ok(), "the machine ends cleanly: {outcome:?}");
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
Some(chain.commit_b),
|
||
"the first command applied through the CAS home"
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/dev").await,
|
||
Some(chain.commit_a),
|
||
"the second command applied"
|
||
);
|
||
assert!(
|
||
mem_ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pushed objects landed in the backend namespace"
|
||
);
|
||
}
|
||
|
||
/// A packless delete-only push (duplex) takes the peek's packless
|
||
/// path: no prepare call runs (CAS-only), the CAS deletes through the
|
||
/// same ref home, the report is the same shape, and after the report
|
||
/// the tail drains to EOF — the client stays open until the report and
|
||
/// closing afterward ends the session cleanly, never a mid-stream RST
|
||
/// (the drain's Ok completion is the observable).
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rule (ii) (a timeout with zero pack
|
||
/// bytes takes the packless, CAS-only path) and rule (iv) (drain after
|
||
/// the report, never a mid-stream RST — the client stays open until
|
||
/// the report); ADR-020 §3 (the client's write-side-open shape);
|
||
/// ADR-013 §7 (the same one-transaction CAS home, applied exactly as a
|
||
/// pushed pack's updates would be; the same report); ADR-009 (the peek
|
||
/// window is `Limits::packless_peek`); poco-findings.md (the ~40 ms
|
||
/// datum — the report shape is the capture-level acceptance);
|
||
/// push-captures.md §Status report.
|
||
#[tokio::test]
|
||
async fn packless_delete_push_takes_the_cas_only_path_with_a_clean_drain() {
|
||
let record = repo("owner/doubled");
|
||
let current = oid(0x1b);
|
||
let mem_refs = MemRefs::default();
|
||
seed_ref(&mem_refs, &record, "refs/heads/gone", current).await;
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k",
|
||
&[delete_command("refs/heads/gone", current)],
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
// The client waits for the report (ADR-020 §3), then closes.
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
|
||
"the packless report shape, no pack involved"
|
||
);
|
||
drop(client);
|
||
let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.expect("the drain ends when the client closes")
|
||
.unwrap();
|
||
assert!(outcome.is_ok(), "the session drains cleanly: {outcome:?}");
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
0,
|
||
"no prepare call runs on the packless path"
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/gone").await,
|
||
None,
|
||
"the CAS deleted the ref"
|
||
);
|
||
}
|
||
|
||
/// A packless create-from-nothing command applies through the same one
|
||
/// CAS home (ADR-013 §7 exactly as a pushed pack's updates would — the
|
||
/// v1 policy leg is CAS-old-value only, non-FF allowed, ADR-013 §7).
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rule (ii) (delete/create-from-nothing
|
||
/// are the packless path's in-scope commands); ADR-013 §7 (CAS-only,
|
||
/// the packless create reaches the same CAS home); push-captures.md
|
||
/// §CAS / race semantics.
|
||
#[tokio::test]
|
||
async fn packless_create_applies_through_the_cas_home() {
|
||
let record = repo("owner/doubled");
|
||
let target = oid(0x2c);
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status",
|
||
&[create_command("refs/heads/born", target)],
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
bare_report_lines(&packets),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/born".to_owned()],
|
||
"the bare report shape without sideband selection"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/born").await,
|
||
Some(target),
|
||
"the create applied through the CAS home"
|
||
);
|
||
}
|
||
|
||
/// A timeout with a partial pack-header byte string is the malformed-
|
||
/// push session error — never a silent packless downgrade (a half-sent
|
||
/// pack cannot be re-interpreted as absence): the session errors, no
|
||
/// ref applies, no report runs.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rule (iii); ADR-005 §4 (the session
|
||
/// error's wire mapping is the entries'); ADR-009 (fail-closed);
|
||
/// push-captures.md (vs. the stdio shape — the duplex amendment is the
|
||
/// pinned machine behavior).
|
||
#[tokio::test]
|
||
async fn a_partial_header_at_peek_timeout_is_a_malformed_push_session_error() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(b"PA").await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.expect("the peek's window times out — no park forever")
|
||
.unwrap();
|
||
let message = protocol_message(session_error(outcome));
|
||
assert!(message.contains("pack-header"), "{message}");
|
||
assert!(message.contains("malformed push"), "{message}");
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
None,
|
||
"no ref applied through the malformed push"
|
||
);
|
||
}
|
||
|
||
/// An atomic push whose CAS denies one ref reports the CAS's outcomes
|
||
/// verbatim in request order: the denied ref keeps its own server-
|
||
/// chosen reason, the otherwise-succeeding refs report `atomic push
|
||
/// failure`, and nothing applies (the one-transaction home rolled the
|
||
/// whole push back).
|
||
///
|
||
/// Contract stamp: ADR-013 §10 (the atomic-abort mapping, verbatim);
|
||
/// ADR-013 §7 (the single CAS home + per-ref denials as outcomes);
|
||
/// push-captures.md §Status report (the captured pair), §CAS / race
|
||
/// semantics; tests/common §CAS semantics (the double's transaction
|
||
/// shape).
|
||
#[tokio::test]
|
||
async fn an_atomic_cas_denial_frames_the_outcomes_verbatim() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/main", chain.commit_b),
|
||
update_command("refs/heads/dev", oid(0xaa), chain.commit_a),
|
||
];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k atomic",
|
||
&commands,
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
format!("ng refs/heads/main {ATOMIC_PUSH_FAILURE_REASON}"),
|
||
"ng refs/heads/dev stale info".to_owned(),
|
||
],
|
||
"the aborted ref reports atomic push failure; the denied ref keeps its own reason"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
None,
|
||
"the atomic rollback applied nothing"
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/dev").await,
|
||
None,
|
||
"the denied ref did not apply"
|
||
);
|
||
}
|
||
|
||
/// With `atomic` negotiated and the name validation rejecting a ref,
|
||
/// the transaction aborts whole — no CAS call runs at all (a partial
|
||
/// apply is precisely what atomic forbids): the rejected ref keeps
|
||
/// `funny refname` (its own pinned reason, the denied-ref rule) and
|
||
/// the valid refs report `atomic push failure` — the captured upstream
|
||
/// shape.
|
||
///
|
||
/// Contract stamp: ADR-013 §10 (the atomic abort on any failure), §7
|
||
/// (a rejected refname keeps `funny refname`), §9 (the rejected name
|
||
/// never reaches the CAS), §8 (reasons verbatim); push-captures.md
|
||
/// §Status report (the captured pair's shape); poco-findings.md (the
|
||
/// live-captured atomic-abort-on-a-name-check).
|
||
#[tokio::test]
|
||
async fn an_atomic_push_with_a_rejected_name_aborts_the_whole_push() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/../evil", chain.commit_b),
|
||
create_command("refs/heads/main", chain.commit_b),
|
||
];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k atomic",
|
||
&commands,
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
"ng refs/heads/../evil funny refname".to_owned(),
|
||
format!("ng refs/heads/main {ATOMIC_PUSH_FAILURE_REASON}"),
|
||
],
|
||
"the rejected name keeps its pinned reason; the aborted refs report the abort"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
None,
|
||
"no ref applied — the atomic push aborted whole"
|
||
);
|
||
assert_eq!(
|
||
mem_refs.list_refs(&record).await.unwrap().len(),
|
||
0,
|
||
"the backend's ref table never saw any name"
|
||
);
|
||
}
|
||
|
||
/// A rejected ref name reports `ng <ref> funny refname` (the pinned
|
||
/// observed text) and is excluded from the CAS call — the backend's
|
||
/// ref table never sees an invalid name — while the remaining refs
|
||
/// apply in their request order.
|
||
///
|
||
/// Contract stamp: ADR-013 §7 (per-ref validation: name first), §9
|
||
/// (the check's home + the exclusion from CAS), §8 (the report's
|
||
/// per-ref reasons verbatim, in request order); push-captures.md
|
||
/// §Raw stdio (the `funny refname` ground truth).
|
||
#[tokio::test]
|
||
async fn a_rejected_refname_reports_funny_refname_and_is_excluded_from_the_cas() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/../evil", chain.commit_b),
|
||
create_command("refs/heads/main", chain.commit_a),
|
||
];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes("report-status", &commands, &[]))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
bare_report_lines(&packets),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
"ng refs/heads/../evil funny refname".to_owned(),
|
||
"ok refs/heads/main".to_owned(),
|
||
],
|
||
"report in request order — the rejected slot keeps its place"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
Some(chain.commit_a),
|
||
"the remaining refs applied"
|
||
);
|
||
assert_eq!(
|
||
mem_refs.list_refs(&record).await.unwrap().len(),
|
||
1,
|
||
"the invalid name never reached the CAS home"
|
||
);
|
||
}
|
||
|
||
/// A pack-level failure (`unpack ng <reason>`) reports the observed
|
||
/// raw-stdio shape — all refs `ng <ref> unpacker error` — with no CAS
|
||
/// call at all: nothing applies through a failed unpack.
|
||
///
|
||
/// Contract stamp: ADR-013 §8 (the unpack ng leg + all-refs
|
||
/// `unpacker error`, the pack-level reason forwarded verbatim); ADR-018
|
||
/// §5 (the updates round-trip verbatim on the ng path — the all-refs
|
||
/// names survive); ADR-013 §7 (per-ref checks run AFTER unpack);
|
||
/// push-captures.md §Raw stdio (corrupt pack), §Pack-read ordering
|
||
/// (definitive unpack-first order).
|
||
#[tokio::test]
|
||
async fn an_unpack_ng_leg_reports_all_refs_unpacker_error_without_a_cas_call() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/main", oid(0x2c)),
|
||
create_command("refs/heads/other", oid(0x3d)),
|
||
];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k",
|
||
&commands,
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
// A corrupt pack header (a full 12-byte header with a bad
|
||
// signature): the pack-level failure is detected without waiting
|
||
// for more bytes — the stdio capture's shape (`unpack protocol
|
||
// error (pack signature mismatch detected)`), not a short stall
|
||
// (a client stalling mid-header parks the blocking ingestion on
|
||
// bytes only its EOF resolves — the entries' wall clock's victim,
|
||
// never an in-machine error path).
|
||
client.write_all(b"NOPE-NOPE-NO").await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
let lines = sideband_report_lines(&packets);
|
||
assert!(
|
||
lines[0].starts_with("unpack ng "),
|
||
"pack-level reason verbatim: {lines:?}"
|
||
);
|
||
assert!(!lines[0].contains(UNPACKER_ERROR_REASON), "{lines:?}");
|
||
assert_eq!(
|
||
lines[1..],
|
||
vec![
|
||
format!("ng refs/heads/main {UNPACKER_ERROR_REASON}"),
|
||
format!("ng refs/heads/other {UNPACKER_ERROR_REASON}"),
|
||
],
|
||
"all refs report the unpacker-error shape"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
1,
|
||
"the prepare call ran once (the pack failed inside it)"
|
||
);
|
||
assert_eq!(
|
||
mem_refs.list_refs(&record).await.unwrap().len(),
|
||
0,
|
||
"no ref applied through the failed unpack"
|
||
);
|
||
}
|
||
|
||
/// The nothing-to-do shape (an immediate flush, no commands) replies
|
||
/// flush with no report and no other byte — no peek, no CAS, no report
|
||
/// section, no prepare call.
|
||
///
|
||
/// Contract stamp: ADR-013 §3 (the nothing-to-do shape: reply flush,
|
||
/// no report); push-captures.md §CAS / race semantics (the empty
|
||
/// command list's client-side shape).
|
||
#[tokio::test]
|
||
async fn the_nothing_to_do_shape_replies_flush_with_no_report() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes("report-status side-band-64k", &[], &[]))
|
||
.await
|
||
.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(packets.len(), 1, "exactly the reply flush");
|
||
assert!(packets[0].is_empty(), "the reply flush has no payload");
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
0,
|
||
"no prepare call ran on the nothing-to-do shape"
|
||
);
|
||
assert_eq!(
|
||
mem_refs.list_refs(&record).await.unwrap().len(),
|
||
0,
|
||
"no CAS ran"
|
||
);
|
||
}
|
||
|
||
/// The grammar's up-front rejections surface as session errors before
|
||
/// any pack byte is read: shallow lines (ADR-013 §4, rejected up
|
||
/// front) and an unadvertised object-format (the push-side check, the
|
||
/// fetch rule). Nothing applies, nothing reports.
|
||
///
|
||
/// Contract stamp: ADR-013 §4 (shallow's up-front rejection), §2/§4
|
||
/// (the object-format check rides the fetch rule); ADR-005 §4 (the
|
||
/// session error's wire mapping is the entries'); ADR-013 §7 (the
|
||
/// per-ref checks run after unpack — none of this ever reaches them).
|
||
#[tokio::test]
|
||
async fn shallow_and_object_format_reject_up_front_as_session_errors() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(MemRefs::default()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let commands = [
|
||
format!("shallow {}", oid(0x1b)),
|
||
create_command("refs/heads/main", oid(0x2c)),
|
||
];
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes("", &commands, &[]))
|
||
.await
|
||
.unwrap();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
fast_limits(),
|
||
false,
|
||
Arc::new(Semaphore::new(1)),
|
||
);
|
||
let message = protocol_message(session_error(machine.await.unwrap()));
|
||
assert!(message.contains("shallow"), "{message}");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
|
||
let commands = [create_command("refs/heads/main", oid(0x2c))];
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes("object-format=sha1024", &commands, &[]))
|
||
.await
|
||
.unwrap();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
fast_limits(),
|
||
false,
|
||
Arc::new(Semaphore::new(1)),
|
||
);
|
||
let message = protocol_message(session_error(machine.await.unwrap()));
|
||
assert!(message.contains("object-format"), "{message}");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
}
|
||
|
||
/// The permit wraps prepare (ADR-009's enforcement point): a saturated
|
||
/// pipeline semaphore delays the prepare call — nothing runs until a
|
||
/// permit frees — and once it frees the push completes with its
|
||
/// report (the pre-written pack parks inside the bounded bridge the
|
||
/// whole time).
|
||
///
|
||
/// Contract stamp: ADR-009 (the pipeline-concurrency budget's
|
||
/// enforcement point around the backend pipeline calls — the same
|
||
/// semaphore the fetch machine's `generate` runs under); backend.md
|
||
/// §concurrency model (the permit stays wire-side, the blocking thread
|
||
/// impl-side); ADR-013 §6's surviving budgets paragraph.
|
||
#[tokio::test]
|
||
async fn the_pipeline_concurrency_permit_wraps_prepare() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let refs = Arc::new(MemRefs::default()) as Arc<dyn GitRefs>;
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let pipeline = Arc::new(Semaphore::new(1));
|
||
let held = pipeline
|
||
.acquire()
|
||
.await
|
||
.expect("the rows hold the permit first");
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::clone(&pipeline),
|
||
);
|
||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
0,
|
||
"a saturated semaphore delays the prepare call"
|
||
);
|
||
drop(held);
|
||
// The report is read while the client end stays alive: the relay
|
||
// (joined by the machine after the report) ends at the tail's EOF
|
||
// — the client's close past the report.
|
||
let packets = read_pkts(&mut client).await;
|
||
drop(client);
|
||
let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.expect("the machine completes once the permit frees")
|
||
.unwrap();
|
||
assert!(outcome.is_ok());
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
1,
|
||
"prepare ran exactly once, under the permit"
|
||
);
|
||
let lines = bare_report_lines(&packets);
|
||
assert_eq!(
|
||
lines,
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()]
|
||
);
|
||
}
|
||
|
||
/// The wire-side pack-size budget breaches fail closed (ADR-009): the
|
||
/// relay stops before the breaching bytes reach the ingestion, the
|
||
/// session returns the budget failure (the client-fault class), and no
|
||
/// ref applies, no report runs.
|
||
///
|
||
/// Contract stamp: ADR-009 (the `max_pack_size` budget, fail-closed,
|
||
/// its client-fault class — 413-mapped at the door on stateless);
|
||
/// pack_source.rs (the relay's wire-side enforcement point); ADR-020 §2
|
||
/// (the impl-side counting-reader is the impl's own defense).
|
||
#[tokio::test]
|
||
async fn a_pack_size_budget_breach_fails_the_session_closed() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = Limits {
|
||
max_pack_size: 2,
|
||
packless_peek: Duration::from_millis(100),
|
||
..Limits::default()
|
||
};
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(4)),
|
||
);
|
||
let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.expect("the breach fails the session promptly")
|
||
.unwrap();
|
||
let error = session_error(outcome);
|
||
let SubstrateError::Protocol(message) = error else {
|
||
panic!("the budget breach is the client-fault class: {error:?}");
|
||
};
|
||
assert!(message.contains("max_pack_size"), "{message}");
|
||
assert_eq!(
|
||
mem_refs.list_refs(&record).await.unwrap().len(),
|
||
0,
|
||
"no ref applied through the breached push"
|
||
);
|
||
}
|
||
|
||
/// The negotiated push-options section (ADR-013 §11, the args section
|
||
/// when the gate is on) is consumed by the grammar **before** the peek
|
||
/// (rule (i)): the section parses, the peek watches only for the pack's
|
||
/// first bytes, and the push proceeds ordinarily.
|
||
///
|
||
/// Contract stamp: ADR-013 §11 (the section's grammar; verbatim,
|
||
/// un-interpreted); ADR-020 §1 rule (i); push-captures.md §Request.
|
||
#[tokio::test]
|
||
async fn the_negotiated_push_options_section_is_consumed_before_the_peek() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
let commands = [create_command("refs/heads/main", chain.commit_b)];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status push-options",
|
||
&commands,
|
||
&["private=yes".to_owned()],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
true,
|
||
Arc::new(Semaphore::new(4)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
assert_eq!(
|
||
bare_report_lines(&packets),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
"the push flowed through, the options section consumed ahead of the peek"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
Some(chain.commit_b)
|
||
);
|
||
}
|
||
|
||
/// Selecting `push-options` when the gate is off fails closed (the
|
||
/// grammar's rule): the session errors before the peek, no ref
|
||
/// applies.
|
||
///
|
||
/// Contract stamp: ADR-013 §11 (the gate's default-off posture — the
|
||
/// selection is out of grammar), §2 (only advertised caps select);
|
||
/// push.rs (the grammar's fail-closed rule).
|
||
#[tokio::test]
|
||
async fn selecting_push_options_with_the_gate_off_fails_closed() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status push-options",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
&["private=yes".to_owned()],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
fast_limits(),
|
||
false,
|
||
Arc::new(Semaphore::new(1)),
|
||
);
|
||
let message = protocol_message(session_error(machine.await.unwrap()));
|
||
assert!(message.contains("not advertised"), "{message}");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
}
|
||
|
||
/// The stateless tail ingests dribbled pack bytes — including a pack
|
||
/// header split across body chunks (the EOF-driven twin of the duplex
|
||
/// peek's reassembly) — and reports the same shapes: `unpack ok` +
|
||
/// per-ref `ok` lines in request order, sideband-framed; the objects
|
||
/// land.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 (the EOF-driven twin — the stateless POST
|
||
/// body's EOF is unambiguous); ADR-013 §3 (one POST body is one
|
||
/// invocation, the pack raw to the body's end), §7–8 (the CAS + the
|
||
/// report, request order); push-captures.md §Status report.
|
||
#[tokio::test]
|
||
async fn the_stateless_tail_ingests_dribbled_pack_bytes() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let mem_ingest = MemPackIngest::default();
|
||
let ingest = Arc::new(mem_ingest.clone()) as Arc<dyn GitPackIngest>;
|
||
let commands = [
|
||
create_command("refs/heads/main", chain.commit_b),
|
||
create_command("refs/heads/dev", chain.commit_a),
|
||
];
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"report-status side-band-64k",
|
||
&commands,
|
||
vec![
|
||
chain.pack[..1].to_vec(),
|
||
chain.pack[1..24].to_vec(),
|
||
chain.pack[24..].to_vec(),
|
||
],
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
assert!(outcome.is_ok(), "the machine ends cleanly: {outcome:?}");
|
||
let packets = parse_pkt_lines(&sink.body());
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec![
|
||
"unpack ok".to_owned(),
|
||
"ok refs/heads/main".to_owned(),
|
||
"ok refs/heads/dev".to_owned(),
|
||
]
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
Some(chain.commit_b)
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/dev").await,
|
||
Some(chain.commit_a)
|
||
);
|
||
assert!(
|
||
mem_ingest.objects().contains(&record, &chain.commit_b),
|
||
"the pushed objects landed"
|
||
);
|
||
}
|
||
|
||
/// EOF at the stateless tail's start (zero pack bytes) is the packless
|
||
/// path: CAS-only (no prepare call), the same report, the delete-only
|
||
/// push served.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 (the EOF twin's zero-bytes mapping — the
|
||
/// same packless, CAS-only path; EOF is instead of a timeout); ADR-013
|
||
/// §7 (the same CAS home); core/w3-push-machine (the EOF twin's rule).
|
||
#[tokio::test]
|
||
async fn eof_at_the_tails_start_takes_the_stateless_packless_path() {
|
||
let record = repo("owner/doubled");
|
||
let current = oid(0x1b);
|
||
let mem_refs = MemRefs::default();
|
||
seed_ref(&mem_refs, &record, "refs/heads/gone", current).await;
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let runs = Arc::new(AtomicUsize::new(0));
|
||
let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"report-status side-band-64k",
|
||
&[delete_command("refs/heads/gone", current)],
|
||
Vec::new(),
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
assert!(outcome.is_ok());
|
||
let packets = parse_pkt_lines(&sink.body());
|
||
assert_eq!(
|
||
sideband_report_lines(&packets),
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
|
||
);
|
||
assert_eq!(
|
||
runs.load(Ordering::SeqCst),
|
||
0,
|
||
"no prepare call runs on the packless path"
|
||
);
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/gone").await,
|
||
None,
|
||
"the CAS deleted the ref"
|
||
);
|
||
}
|
||
|
||
/// EOF after a partial pack header is the stateless malformed-push
|
||
/// error — the timeout rule's EOF-driven twin — never a silent
|
||
/// packless downgrade; no ref applies, no report writes.
|
||
///
|
||
/// Contract stamp: ADR-020 §1 rule (iii) (the same rule, EOF-driven
|
||
/// instead of timeout-driven); ADR-009 (fail-closed).
|
||
#[tokio::test]
|
||
async fn eof_after_a_partial_header_is_the_stateless_malformed_push_error() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"report-status side-band-64k",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
vec![b"PA".to_vec()],
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
let message = protocol_message(session_error(outcome));
|
||
assert!(message.contains("pack-header"), "{message}");
|
||
assert!(message.contains("malformed push"), "{message}");
|
||
assert!(sink.body().is_empty(), "no report bytes were written");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
}
|
||
|
||
/// The stateless nothing-to-do shape replies flush with no report.
|
||
///
|
||
/// Contract stamp: ADR-013 §3 (both substrates).
|
||
#[tokio::test]
|
||
async fn the_stateless_nothing_to_do_shape_replies_flush_with_no_report() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks("report-status side-band-64k", &[], Vec::new()),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
assert!(outcome.is_ok());
|
||
let packets = parse_pkt_lines(&sink.body());
|
||
assert_eq!(packets.len(), 1, "exactly the reply flush");
|
||
assert!(packets[0].is_empty(), "the reply flush has no payload");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
}
|
||
|
||
/// A client that selected neither report cap still gets its CAS
|
||
/// applied — the machine reports nothing (no report, no flush; real
|
||
/// receive-pack's `report()` shape) — with the CAS's outcomes intact.
|
||
///
|
||
/// Contract stamp: ADR-013 §8 (the report exists for a report-status
|
||
/// selection; without it nothing is written — the mirror of
|
||
/// receive-pack.c's report guard); push-captures.md §Status report
|
||
/// (every captured client selected the cap; the no-selection shape is
|
||
/// the guard's own).
|
||
#[tokio::test]
|
||
async fn a_client_not_selecting_report_caps_gets_no_report_bytes() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"",
|
||
&[create_command("refs/heads/main", chain.commit_b)],
|
||
vec![chain.pack.clone()],
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
assert!(outcome.is_ok());
|
||
assert!(sink.body().is_empty(), "no report bytes without the cap");
|
||
assert_eq!(
|
||
list_ref(&mem_refs, &record, "refs/heads/main").await,
|
||
Some(chain.commit_b),
|
||
"the CAS applied regardless of the report selection"
|
||
);
|
||
}
|
||
|
||
/// The duplex machine's report framing follows the client's
|
||
/// `side-band-64k` selection; without the selection the report's
|
||
/// pkt-lines are bare to the single flush (the file:// capture's
|
||
/// shape).
|
||
///
|
||
/// Contract stamp: ADR-013 §8 (the framing selects on the client's
|
||
/// side-band-64k — an unwrapped report under a sideband selection
|
||
/// aborts real clients, observed live); ADR-005 (the substrate owns
|
||
/// the framing); push-captures.md §Status report.
|
||
#[tokio::test]
|
||
async fn the_reports_framing_follows_the_sideband_selection() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
let commands = [create_command("refs/heads/main", chain.commit_b)];
|
||
|
||
let (mut client, stream) = duplex(1 << 20);
|
||
client
|
||
.write_all(&request_bytes(
|
||
"report-status side-band-64k",
|
||
&commands,
|
||
&[],
|
||
))
|
||
.await
|
||
.unwrap();
|
||
client.write_all(&chain.pack).await.unwrap();
|
||
|
||
let limits = fast_limits();
|
||
let machine = spawn_duplex_machine(
|
||
stream,
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
record.clone(),
|
||
limits,
|
||
false,
|
||
Arc::new(Semaphore::new(4)),
|
||
);
|
||
|
||
let packets = read_pkts(&mut client).await;
|
||
let lines = sideband_report_lines(&packets);
|
||
assert_eq!(
|
||
lines,
|
||
vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
|
||
"the sideband-selected client rides band-1 with the self-framed report"
|
||
);
|
||
drop(client);
|
||
assert!(tokio::time::timeout(Duration::from_secs(5), machine)
|
||
.await
|
||
.unwrap()
|
||
.unwrap()
|
||
.is_ok());
|
||
}
|
||
|
||
/// The stateless machine upholds the same up-front rejections (the
|
||
/// grammar's session errors are substrate-blind).
|
||
///
|
||
/// Contract stamp: ADR-013 §4 (the rejections are substrate-blind);
|
||
/// core/w3-push-machine (the EOF twin's parity).
|
||
#[tokio::test]
|
||
async fn the_stateless_machine_rejects_up_front_the_same_way() {
|
||
let record = repo("owner/doubled");
|
||
let mem_refs = MemRefs::default();
|
||
let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
|
||
let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"report-status",
|
||
&[
|
||
format!("shallow {}", oid(0x1b)),
|
||
create_command("refs/heads/main", oid(0x2c)),
|
||
],
|
||
Vec::new(),
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
let message = protocol_message(session_error(outcome));
|
||
assert!(message.contains("shallow"), "{message}");
|
||
assert!(sink.body().is_empty(), "no report on the rejected push");
|
||
|
||
let (outcome, sink) = stateless_machine(
|
||
body_chunks(
|
||
"object-format=sha1024",
|
||
&[create_command("refs/heads/main", oid(0x2c))],
|
||
Vec::new(),
|
||
),
|
||
Arc::clone(&refs),
|
||
Arc::clone(&ingest),
|
||
&record,
|
||
fast_limits(),
|
||
false,
|
||
)
|
||
.await;
|
||
let message = protocol_message(session_error(outcome));
|
||
assert!(message.contains("object-format"), "{message}");
|
||
assert!(sink.body().is_empty(), "no report on the rejected push");
|
||
assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
|
||
}
|
||
|
||
/// The seam the machine composes: `prepare`'s `PreparedPush.updates`
|
||
/// round-trips verbatim and becomes the CAS call's input — the
|
||
/// machine's leg boundary, pinned on the doubles.
|
||
///
|
||
/// Contract stamp: ADR-018 §5 (`PreparedPush::updates` = the prepare/
|
||
/// CAS inputs, verbatim); ADR-013 §6–7 (prepare validates and returns,
|
||
/// refs commits).
|
||
#[tokio::test]
|
||
async fn prepared_updates_are_the_cas_inputs() {
|
||
let record = repo("owner/doubled");
|
||
let chain = chain_pack();
|
||
let ingest = MemPackIngest::default();
|
||
let updates = vec![RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(chain.commit_b),
|
||
}];
|
||
let prepared = ingest
|
||
.prepare(
|
||
&record,
|
||
updates.clone(),
|
||
Box::new(io::Cursor::new(chain.pack)),
|
||
None,
|
||
Limits::default(),
|
||
)
|
||
.await
|
||
.unwrap();
|
||
assert!(prepared.unpack.is_ok());
|
||
assert_eq!(prepared.updates, updates, "verbatim into the CAS call");
|
||
}
|