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.
1428 lines
46 KiB
Rust
1428 lines
46 KiB
Rust
//! The wave-4 GitSession fetch client (core/w4-session-fetch): the
|
||
//! consumer half's ADR-017 §1 operations riding the wave-4
|
||
//! construction — `ls_refs` (the `ls-refs` argument grammar our own
|
||
//! server machine parses) and `fetch` (the full ack loop as full
|
||
//! stateless commands per ADR-014, the done round's sideband pack
|
||
//! pumped to the caller's sink) — rows driven in-process,
|
||
//! client-vs-our-server, both halves in one test over the wave-2/3
|
||
//! backend doubles (the POC-X gate shape re-landed on the real code:
|
||
//! findings 2/4/5 of pocx-findings.md). The direct-mode rows pair the
|
||
//! session with the real `GitAdapter` (the request line is the
|
||
//! adapter's consumption, ADR-016 §2); the machinery rows pair
|
||
//! channels-mode sessions with `serve_duplex` (zero preamble — the
|
||
//! preamble rode the open op).
|
||
//!
|
||
//! Rows are names + `Contract stamp:` versions, per `tests/common`
|
||
//! §The row/stamp discipline.
|
||
|
||
mod common;
|
||
|
||
use std::collections::{BTreeMap, BTreeSet, HashMap};
|
||
use std::pin::Pin;
|
||
use std::sync::{Arc, Mutex};
|
||
use std::task::{Context, Poll};
|
||
use std::time::Duration;
|
||
|
||
use alkcall::core::auth::Identity;
|
||
use alkcall::core::types::{BiStream, Connection, ProtocolHandler};
|
||
use tokio::io::{duplex, AsyncRead, AsyncWriteExt, ReadBuf};
|
||
use tokio::sync::Semaphore;
|
||
|
||
use gix_hash::ObjectId;
|
||
|
||
use alkgit::{
|
||
serve_duplex, AccessAction, AuthorizedRepo, FetchRound, GitAdapter, GitPackGen, GitPackIngest,
|
||
GitRefs, GitRegistry, GitRegistryStore, GitSession, Limits, LsRefsArgs, PreparedPush, RefLine,
|
||
RefUpdate, RepoRecord, Service, SessionError, SubstrateError, Visibility,
|
||
};
|
||
use common::{
|
||
blob_fixture, commit_fixture, decode_pack, tree_fixture, MemPackGen, MemPackIngest, MemRefs,
|
||
MemRegistry,
|
||
};
|
||
|
||
type Handles = (
|
||
Arc<MemRegistry>,
|
||
Arc<MemRefs>,
|
||
Arc<MemPackGen>,
|
||
Arc<MemPackIngest>,
|
||
);
|
||
|
||
fn handles() -> Handles {
|
||
let objects = common::MemObjects::default();
|
||
(
|
||
Arc::new(MemRegistry::default()),
|
||
Arc::new(MemRefs::default()),
|
||
Arc::new(MemPackGen {
|
||
objects: objects.clone(),
|
||
}),
|
||
Arc::new(MemPackIngest { objects }),
|
||
)
|
||
}
|
||
|
||
fn repo(id: &str) -> RepoRecord {
|
||
RepoRecord {
|
||
repo_id: id.to_string(),
|
||
visibility: Visibility::Private,
|
||
grants: BTreeMap::new(),
|
||
}
|
||
}
|
||
|
||
fn identity(id: &str) -> Identity {
|
||
Identity {
|
||
id: id.to_string(),
|
||
scopes: Vec::new(),
|
||
resources: HashMap::new(),
|
||
}
|
||
}
|
||
|
||
type WireRecord = Arc<Mutex<Vec<u8>>>;
|
||
|
||
type StreamJoiner = tokio::task::JoinHandle<Result<(), SubstrateError>>;
|
||
|
||
/// Records every byte read off the wire — the client-sent side (what
|
||
/// the server RECEIVED), for the request-grammar byte pins.
|
||
struct RecordRead<S> {
|
||
inner: S,
|
||
record: WireRecord,
|
||
}
|
||
|
||
impl<S: AsyncRead + Unpin> AsyncRead for RecordRead<S> {
|
||
fn poll_read(
|
||
mut self: Pin<&mut Self>,
|
||
cx: &mut Context<'_>,
|
||
buf: &mut ReadBuf<'_>,
|
||
) -> Poll<std::io::Result<()>> {
|
||
let this = self.as_mut().get_mut();
|
||
let before = buf.filled().len();
|
||
match Pin::new(&mut this.inner).poll_read(cx, buf) {
|
||
Poll::Ready(Ok(())) => {
|
||
this.record
|
||
.lock()
|
||
.expect("spy record")
|
||
.extend_from_slice(&buf.filled()[before..]);
|
||
Poll::Ready(Ok(()))
|
||
}
|
||
other => other,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Records every byte written to the wire (the server-sent side).
|
||
struct RecordWrite<S> {
|
||
inner: S,
|
||
record: WireRecord,
|
||
}
|
||
|
||
impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for RecordWrite<S> {
|
||
fn poll_write(
|
||
mut self: Pin<&mut Self>,
|
||
cx: &mut Context<'_>,
|
||
buf: &[u8],
|
||
) -> Poll<std::io::Result<usize>> {
|
||
let this = self.as_mut().get_mut();
|
||
this.record
|
||
.lock()
|
||
.expect("spy record")
|
||
.extend_from_slice(buf);
|
||
Pin::new(&mut this.inner).poll_write(cx, buf)
|
||
}
|
||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
||
let this = self.as_mut().get_mut();
|
||
Pin::new(&mut this.inner).poll_flush(cx)
|
||
}
|
||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
|
||
let this = self.as_mut().get_mut();
|
||
Pin::new(&mut this.inner).poll_shutdown(cx)
|
||
}
|
||
}
|
||
|
||
/// The fixture: the doubles' registry double plus the granted repo row.
|
||
async fn fetch_fixture(holder: &str, action: AccessAction) -> (Handles, RepoRecord, Identity) {
|
||
let handles = handles();
|
||
let mut granted = repo("owner/doubled");
|
||
granted.grants.insert(holder.to_owned(), vec![action]);
|
||
let registry = Arc::clone(&handles.0);
|
||
registry.put_repo(&granted).await.expect("the repo row");
|
||
(handles, granted, identity(holder))
|
||
}
|
||
|
||
/// A two-commit chain fixture (fetch_machines' rows' shape): a child
|
||
/// commit on its root, both trees/blobs seeded in one namespace.
|
||
struct Chain {
|
||
commit_a: ObjectId,
|
||
commit_b: ObjectId,
|
||
tree_a: ObjectId,
|
||
tree_b: ObjectId,
|
||
blob_a: ObjectId,
|
||
blob_b: ObjectId,
|
||
}
|
||
|
||
fn two_commit_chain(objects: &common::MemObjects, doubled: &RepoRecord) -> Chain {
|
||
let (blob_a, blob_a_bytes) = blob_fixture(b"root content");
|
||
let (blob_b, blob_b_bytes) = blob_fixture(b"child content");
|
||
let (tree_a, tree_a_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]);
|
||
let (tree_b, tree_b_bytes) =
|
||
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", 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 (oid, 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),
|
||
] {
|
||
objects.put(doubled, oid, bytes);
|
||
}
|
||
Chain {
|
||
commit_a,
|
||
commit_b,
|
||
tree_a,
|
||
tree_b,
|
||
blob_a,
|
||
blob_b,
|
||
}
|
||
}
|
||
|
||
/// Seeds the chain's commits as two ref rows (base + main) for the
|
||
/// listing rows.
|
||
async fn seed_chain_refs(handles: &Handles, record: &RepoRecord, chain: &Chain) {
|
||
Arc::clone(&handles.1)
|
||
.apply_updates(
|
||
record,
|
||
vec![
|
||
RefUpdate {
|
||
name: "refs/heads/base".to_owned(),
|
||
expected: None,
|
||
new: Some(chain.commit_a),
|
||
},
|
||
RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(chain.commit_b),
|
||
},
|
||
],
|
||
false,
|
||
)
|
||
.await
|
||
.expect("the seed");
|
||
}
|
||
|
||
/// The channels-mode serving twin: spawns the in-crate duplex serving
|
||
/// entry (`serve_duplex`) over a spy-wrapped pair whose server records
|
||
/// both directions. Pairs with `open_via_channels` sessions (the
|
||
/// preamble rode the open op — zero preamble bytes here).
|
||
async fn serve_twin(
|
||
handles: &Handles,
|
||
record: RepoRecord,
|
||
holder: &Identity,
|
||
action: AccessAction,
|
||
limits: Limits,
|
||
) -> (BiStream, WireRecord, WireRecord, StreamJoiner) {
|
||
let checked =
|
||
AuthorizedRepo::authorize(record, Some(holder), action).expect("the checked marker");
|
||
let refs: Arc<dyn GitRefs> = handles.1.clone();
|
||
let packgen: Arc<dyn GitPackGen> = handles.2.clone();
|
||
let ingest: Arc<dyn GitPackIngest> = handles.3.clone();
|
||
let (client_end, server_end) = duplex(1 << 16);
|
||
let received = Arc::new(Mutex::new(Vec::new()));
|
||
let sent = Arc::new(Mutex::new(Vec::new()));
|
||
let (server_read, server_write) = tokio::io::split(server_end);
|
||
let server = BiStream::from_joined(
|
||
RecordRead {
|
||
inner: server_read,
|
||
record: Arc::clone(&received),
|
||
},
|
||
RecordWrite {
|
||
inner: server_write,
|
||
record: Arc::clone(&sent),
|
||
},
|
||
);
|
||
let joiner = tokio::spawn(async move {
|
||
serve_duplex(
|
||
&checked,
|
||
server,
|
||
limits,
|
||
Service::UploadPack,
|
||
refs,
|
||
packgen,
|
||
ingest,
|
||
false,
|
||
Arc::new(Semaphore::new(
|
||
Limits::default().max_concurrent_blocking_tasks,
|
||
)),
|
||
)
|
||
.await
|
||
});
|
||
let (client_read, client_write) = tokio::io::split(client_end);
|
||
(
|
||
BiStream::from_joined(client_read, client_write),
|
||
received,
|
||
sent,
|
||
joiner,
|
||
)
|
||
}
|
||
|
||
/// The `accept_bi` queue source (session_construction's shape): yields
|
||
/// the server-side stream (spy-wrapped on the client-sent side) once.
|
||
struct YieldQueued {
|
||
streams: tokio::sync::Mutex<std::collections::VecDeque<BiStream>>,
|
||
received: WireRecord,
|
||
}
|
||
|
||
#[async_trait::async_trait]
|
||
impl alkcall::core::BidiStreamSource for YieldQueued {
|
||
async fn accept_bi(&self) -> Result<BiStream, alkcall::core::StreamError> {
|
||
let stream = self
|
||
.streams
|
||
.lock()
|
||
.await
|
||
.pop_front()
|
||
.ok_or(alkcall::core::StreamError::ConnectionClosed)?;
|
||
let (server_read, server_write) = tokio::io::split(stream);
|
||
Ok(BiStream::from_joined(
|
||
RecordRead {
|
||
inner: server_read,
|
||
record: Arc::clone(&self.received),
|
||
},
|
||
server_write,
|
||
))
|
||
}
|
||
async fn open_bi(&self) -> Result<BiStream, alkcall::core::StreamError> {
|
||
Err(alkcall::core::StreamError::StreamClosed)
|
||
}
|
||
fn remote_addr(&self) -> Option<std::net::SocketAddr> {
|
||
None
|
||
}
|
||
fn close(&self, _code: u32, _reason: &str) {}
|
||
}
|
||
|
||
/// The direct-mode serving twin: the real `GitAdapter` (the request
|
||
/// line is the adapter's consumption — the preamble rides out from the
|
||
/// session's `connect_direct` handshake) over a one-stream source.
|
||
/// Pairs with `connect_direct` sessions; the holder rides the
|
||
/// `AuthContext` identity (the adapter resolve→authorizes through the
|
||
/// doubles' registry row).
|
||
async fn adapter_twin(
|
||
handles: &Handles,
|
||
holder: &Identity,
|
||
) -> (
|
||
BiStream,
|
||
WireRecord,
|
||
WireRecord,
|
||
tokio::task::JoinHandle<()>,
|
||
) {
|
||
let (client_end, server_end) = duplex(1 << 16);
|
||
let received = Arc::new(Mutex::new(Vec::new()));
|
||
let sent = Arc::new(Mutex::new(Vec::new()));
|
||
// The served stream carries the sent-side spy; the adapter's gate
|
||
// reads the request line off the received-side spy.
|
||
let (server_read, server_write) = tokio::io::split(server_end);
|
||
let served = BiStream::from_joined(
|
||
server_read,
|
||
RecordWrite {
|
||
inner: server_write,
|
||
record: Arc::clone(&sent),
|
||
},
|
||
);
|
||
let adapter = GitAdapter::new(
|
||
Arc::clone(&handles.0) as Arc<dyn GitRegistry>,
|
||
Arc::clone(&handles.1) as Arc<dyn GitRefs>,
|
||
Arc::clone(&handles.2) as Arc<dyn GitPackGen>,
|
||
Arc::clone(&handles.3) as Arc<dyn GitPackIngest>,
|
||
false,
|
||
Limits::default(),
|
||
Arc::new(Semaphore::new(
|
||
Limits::default().max_concurrent_blocking_tasks,
|
||
)),
|
||
);
|
||
let connection = Connection::from_source(
|
||
YieldQueued {
|
||
streams: tokio::sync::Mutex::new([served].into()),
|
||
received: Arc::clone(&received),
|
||
},
|
||
b"alk/git".to_vec(),
|
||
);
|
||
let mut auth = alkcall::core::auth::AuthContext::anonymous(b"alk/git");
|
||
auth.identity = Some(holder.clone());
|
||
let joiner = tokio::spawn(async move {
|
||
let _ = ProtocolHandler::handle(&adapter, connection, &auth).await;
|
||
});
|
||
let (client_read, client_write) = tokio::io::split(client_end);
|
||
(
|
||
BiStream::from_joined(client_read, client_write),
|
||
received,
|
||
sent,
|
||
joiner,
|
||
)
|
||
}
|
||
|
||
/// The fake wire peer: writes the scripted bytes, then holds the
|
||
/// stream open (the client's drop closes it) — the hostile/stalling
|
||
/// peer rows' serving side.
|
||
async fn spawn_fake_peer(script: Vec<Vec<u8>>) -> BiStream {
|
||
let (client_end, server_end) = duplex(1 << 16);
|
||
let mut server = server_end;
|
||
let _ = server.write_all(script.concat().as_slice()).await;
|
||
let _ = server.flush().await;
|
||
tokio::spawn(async move {
|
||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||
let _ = server.shutdown().await;
|
||
});
|
||
let (client_read, client_write) = tokio::io::split(client_end);
|
||
BiStream::from_joined(client_read, client_write)
|
||
}
|
||
|
||
/// Encodes one pkt-line (client framing).
|
||
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
|
||
}
|
||
|
||
/// One text command line (the client's text codec appends the newline
|
||
/// the wire grammar carries).
|
||
fn text_line(text: &[u8]) -> Vec<u8> {
|
||
let mut payload = text.to_vec();
|
||
payload.push(b'\n');
|
||
pkt(&payload)
|
||
}
|
||
|
||
fn want_line(oid: &ObjectId) -> Vec<u8> {
|
||
text_line(format!("want {oid}").as_bytes())
|
||
}
|
||
|
||
fn have_line(oid: &ObjectId) -> Vec<u8> {
|
||
text_line(format!("have {oid}").as_bytes())
|
||
}
|
||
|
||
const DELIM: &[u8] = b"0001";
|
||
const FLUSH: &[u8] = b"0000";
|
||
|
||
/// The commands' capability line set (the agent + the object-format
|
||
/// echo, negotiation-captures' grammar).
|
||
fn capability_lines() -> Vec<Vec<u8>> {
|
||
vec![
|
||
text_line(
|
||
concat!(
|
||
"agent=",
|
||
env!("CARGO_PKG_NAME"),
|
||
"/",
|
||
env!("CARGO_PKG_VERSION")
|
||
)
|
||
.as_bytes(),
|
||
),
|
||
text_line(b"object-format=sha1"),
|
||
]
|
||
}
|
||
|
||
// --- ls-refs ----------------------------------------------------------------
|
||
|
||
/// Contract stamp: ADR-017 §1 (ls_refs), ADR-016 §2 (the request
|
||
/// line is the adapter's — byte-pinned), the negotiation-captures
|
||
/// §Round grammar (caps then `0001` then args then flush) on the
|
||
/// `ls-refs` command; core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn ls_refs_direct_returns_the_ref_list_and_the_request_is_the_pinned_grammar() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let seeded_a = ObjectId::from([7u8; 20]);
|
||
let seeded_b = ObjectId::from([8u8; 20]);
|
||
Arc::clone(&handles.1)
|
||
.apply_updates(
|
||
&record,
|
||
vec![
|
||
RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(seeded_a),
|
||
},
|
||
RefUpdate {
|
||
name: "refs/tags/v1".to_owned(),
|
||
expected: None,
|
||
new: Some(seeded_b),
|
||
},
|
||
],
|
||
false,
|
||
)
|
||
.await
|
||
.expect("the seed");
|
||
handles
|
||
.1
|
||
.set_head_symref(&record, Some("refs/heads/main".to_owned()));
|
||
let (client, received, _sent, _joiner) = adapter_twin(&handles, &holder).await;
|
||
|
||
let mut session = GitSession::connect_direct(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Some("example.net"),
|
||
Limits::default(),
|
||
)
|
||
.expect("the direct construction");
|
||
let listed = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.ls_refs(&LsRefsArgs {
|
||
symrefs: true,
|
||
peel: true,
|
||
ref_prefixes: vec!["refs/heads/".to_owned()],
|
||
unborn: false,
|
||
}),
|
||
)
|
||
.await
|
||
.expect("ls-refs is timely")
|
||
.expect("the ref list returns");
|
||
|
||
assert_eq!(
|
||
listed,
|
||
vec![RefLine {
|
||
name: "refs/heads/main".to_owned(),
|
||
oid: Some(seeded_a),
|
||
peeled: None,
|
||
symref_target: None,
|
||
}],
|
||
"the ref-prefix filter honors the argument (the unborn HEAD row is not requested, so the \
|
||
row is omitted there); the tags row filters out"
|
||
);
|
||
|
||
assert_eq!(
|
||
*received.lock().expect("spy record"),
|
||
[
|
||
pkt(b"git-upload-pack owner/doubled\0host=example.net\0\0version=2\0"),
|
||
text_line(b"command=ls-refs"),
|
||
agent_line_placeholder(),
|
||
text_line(b"object-format=sha1"),
|
||
DELIM.to_vec(),
|
||
text_line(b"symrefs"),
|
||
text_line(b"peel"),
|
||
text_line(b"ref-prefix refs/heads/"),
|
||
FLUSH.to_vec(),
|
||
]
|
||
.concat(),
|
||
"the request bytes are the ADR-016 §2 preamble line plus the capture grammar's command shape"
|
||
);
|
||
let _ = seeded_b;
|
||
}
|
||
|
||
fn agent_line_placeholder() -> Vec<u8> {
|
||
text_line(
|
||
concat!(
|
||
"agent=",
|
||
env!("CARGO_PKG_NAME"),
|
||
"/",
|
||
env!("CARGO_PKG_VERSION")
|
||
)
|
||
.as_bytes(),
|
||
)
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §1 (ls_refs — the `unborn` argument
|
||
/// honored), ADR-016 §2's V2 ref grammar (the unborn symref line
|
||
/// parses oid-less — both the literal `unborn` emission and the
|
||
/// zero-oid/attribute-only peer shapes), review 001 N-5 (the unborn
|
||
/// HEAD fixture); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn ls_refs_honors_the_unborn_argument_and_the_bare_command_shape() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
handles
|
||
.1
|
||
.set_head_symref(&record, Some("refs/heads/unborn".to_owned()));
|
||
let (client, _received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the channels-mode construction");
|
||
let listed = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.ls_refs(&LsRefsArgs {
|
||
symrefs: true,
|
||
unborn: true,
|
||
..LsRefsArgs::default()
|
||
}),
|
||
)
|
||
.await
|
||
.expect("ls-refs is timely")
|
||
.expect("the argument-bearing ls-refs runs against the serving twin");
|
||
assert_eq!(
|
||
listed,
|
||
vec![RefLine {
|
||
name: "HEAD".to_owned(),
|
||
oid: None,
|
||
peeled: None,
|
||
symref_target: Some("refs/heads/unborn".to_owned()),
|
||
}],
|
||
"the unborn HEAD row rides the grammar line and parses oid-less"
|
||
);
|
||
|
||
let bare = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.ls_refs(&LsRefsArgs::default()),
|
||
)
|
||
.await
|
||
.expect("ls-refs is timely")
|
||
.expect("the bare command shape (no argument lines, no delimiter)");
|
||
assert!(
|
||
bare.is_empty(),
|
||
"the empty-argument command's response is refs-less (the unborn row only on the request)"
|
||
);
|
||
}
|
||
|
||
// --- the ack loop and the done round ----------------------------------------
|
||
|
||
/// Contract stamp: ADR-017 §1 (fetch — the client side of the loop we
|
||
/// serve), ADR-014 §1/§5 (the acknowledgments grammar + the ack rule —
|
||
/// the exists + is-commit predicate), ADR-023 §6 (rounds are full
|
||
/// stateless commands); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn the_ack_loop_rounds_parse_both_acknowledgment_shapes() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let chain = two_commit_chain(&handles.2.objects, &record);
|
||
let (client, received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the channels-mode construction");
|
||
let reports = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![
|
||
// A recognized common (the root commit exists and is a
|
||
// commit) — the positive ack.
|
||
FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: vec![chain.commit_a],
|
||
done: false,
|
||
},
|
||
// An unknown and a non-commit have — NAK alone.
|
||
FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: vec![chain.blob_a, ObjectId::from([9u8; 20])],
|
||
done: false,
|
||
},
|
||
],
|
||
&mut Vec::new(),
|
||
),
|
||
)
|
||
.await
|
||
.expect("the loop is timely")
|
||
.expect("both rounds acknowledged");
|
||
|
||
assert_eq!(
|
||
reports,
|
||
vec![
|
||
alkgit::FetchRoundReport {
|
||
acks: vec![chain.commit_a],
|
||
pack_bytes: None,
|
||
},
|
||
alkgit::FetchRoundReport {
|
||
acks: Vec::new(),
|
||
pack_bytes: None,
|
||
},
|
||
],
|
||
"the positive ACK shape and the NAK-only shape both parsed, one report per round"
|
||
);
|
||
let wire = received.lock().expect("spy record").clone();
|
||
assert_eq!(
|
||
wire,
|
||
[
|
||
// Round 1's full stateless command (caps, delim, the
|
||
// thin-pack/ofs-delta initial args, then the caller's wants
|
||
// and haves).
|
||
text_line(b"command=fetch"),
|
||
capability_lines().concat(),
|
||
DELIM.to_vec(),
|
||
text_line(b"thin-pack"),
|
||
text_line(b"ofs-delta"),
|
||
want_line(&chain.commit_b),
|
||
have_line(&chain.commit_a),
|
||
FLUSH.to_vec(),
|
||
// Round 2 re-sends the want, new haves.
|
||
text_line(b"command=fetch"),
|
||
capability_lines().concat(),
|
||
DELIM.to_vec(),
|
||
text_line(b"thin-pack"),
|
||
text_line(b"ofs-delta"),
|
||
want_line(&chain.commit_b),
|
||
have_line(&chain.blob_a),
|
||
have_line(&ObjectId::from([9u8; 20])),
|
||
FLUSH.to_vec(),
|
||
]
|
||
.concat(),
|
||
"each round's request bytes are the capture grammar: wants re-sent verbatim, the \
|
||
acknowledgments ride gix's send machinery"
|
||
);
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §1 (the done round), ADR-017 §4 (the
|
||
/// storage-agnostic pack delivery — the caller's sink, byte-counted),
|
||
/// ADR-014 §2 (the done round's packfile section, sideband + flush),
|
||
/// pocx-findings finding 4 (the pump); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn the_done_round_terminates_fetch_and_delivers_the_packed_bytes_to_the_consumer() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let chain = two_commit_chain(&handles.2.objects, &record);
|
||
let (client, received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the channels-mode construction");
|
||
let mut pack = Vec::new();
|
||
let reports = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: Vec::new(),
|
||
done: true,
|
||
}],
|
||
&mut pack,
|
||
),
|
||
)
|
||
.await
|
||
.expect("the fetch is timely")
|
||
.expect("the done round delivered the pack");
|
||
|
||
assert_eq!(
|
||
reports,
|
||
vec![alkgit::FetchRoundReport {
|
||
acks: Vec::new(),
|
||
pack_bytes: Some(pack.len() as u64),
|
||
}],
|
||
"the report's byte count is exactly the consumer's byte count"
|
||
);
|
||
assert!(
|
||
!pack.is_empty(),
|
||
"the sideband-unwrapped pack reached the consumer intact"
|
||
);
|
||
|
||
let decoded = decode_pack(&pack);
|
||
assert_eq!(
|
||
decoded
|
||
.objects
|
||
.iter()
|
||
.map(|(oid, _kind, _body)| *oid)
|
||
.collect::<BTreeSet<_>>(),
|
||
BTreeSet::from([
|
||
chain.commit_a,
|
||
chain.commit_b,
|
||
chain.tree_a,
|
||
chain.tree_b,
|
||
chain.blob_a,
|
||
chain.blob_b,
|
||
]),
|
||
"the full closure of commit_b was delivered (no boundary haves), the trailer verified by \
|
||
the decode"
|
||
);
|
||
|
||
// The request byte pin (the capture grammar's done-round shape).
|
||
assert_eq!(
|
||
*received.lock().expect("spy record"),
|
||
[
|
||
text_line(b"command=fetch"),
|
||
capability_lines().concat(),
|
||
DELIM.to_vec(),
|
||
text_line(b"thin-pack"),
|
||
text_line(b"ofs-delta"),
|
||
want_line(&chain.commit_b),
|
||
text_line(b"done"),
|
||
FLUSH.to_vec(),
|
||
]
|
||
.concat(),
|
||
"the fetch request bytes are the capture grammar (command, caps, delim, args, done, flush)"
|
||
);
|
||
|
||
// The pack round-trips through the doubles' ingest path: the
|
||
// objects land, the connectivity walk closes from the tips.
|
||
let prepared: PreparedPush = handles
|
||
.3
|
||
.prepare(
|
||
&record,
|
||
vec![RefUpdate {
|
||
name: "refs/heads/main".to_owned(),
|
||
expected: None,
|
||
new: Some(chain.commit_b),
|
||
}],
|
||
Box::new(std::io::Cursor::new(pack.clone())),
|
||
None,
|
||
Limits::default(),
|
||
)
|
||
.await
|
||
.expect("the ingestion accepts the delivered pack");
|
||
assert!(
|
||
prepared.unpack.is_ok(),
|
||
"the delivered pack ingests cleanly: {:?}",
|
||
prepared.unpack
|
||
);
|
||
for object in [chain.commit_b, chain.tree_b, chain.blob_b] {
|
||
assert!(
|
||
handles.3.objects().contains(&record, &object),
|
||
"the ingested landed row holds {object}"
|
||
);
|
||
}
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §1 (the done round as the negotiation
|
||
/// terminator), ADR-014 §2/§7 (the done request's haves are the
|
||
/// complete boundary view — the negotiation's payoff is the trimmed
|
||
/// pack), ADR-023 §6 (2 no-done rounds + done, the POC-X gate shape);
|
||
/// core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn multi_round_negotiation_converges_with_two_no_done_rounds_plus_the_done_round() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let chain = two_commit_chain(&handles.2.objects, &record);
|
||
let (client, received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the channels-mode construction");
|
||
let mut pack = Vec::new();
|
||
let reports = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![
|
||
FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: vec![chain.commit_a],
|
||
done: false,
|
||
},
|
||
// The stateless re-send: wants + commons again, plus
|
||
// one unrecognized extra.
|
||
FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: vec![chain.commit_a, chain.blob_b],
|
||
done: false,
|
||
},
|
||
FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: vec![chain.commit_a],
|
||
done: true,
|
||
},
|
||
],
|
||
&mut pack,
|
||
),
|
||
)
|
||
.await
|
||
.expect("the loop is timely")
|
||
.expect("the negotiation converges on the done round");
|
||
|
||
assert_eq!(
|
||
reports[0].acks,
|
||
vec![chain.commit_a],
|
||
"the root commit acks on the first round"
|
||
);
|
||
assert_eq!(
|
||
reports[1].acks,
|
||
vec![chain.commit_a],
|
||
"the common re-sent every round re-acks (the non-commit have does not)"
|
||
);
|
||
let trimmed = reports[2].pack_bytes.expect("the done round delivered");
|
||
assert_eq!(trimmed, pack.len() as u64);
|
||
|
||
// The negotiation's payoff: the boundary's closure (commit_a's
|
||
// chain) subtracted from the pack — measured against the plain
|
||
// full fetch of the same want.
|
||
let (client_full, _received_full, _sent_full, _joiner_full) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
let mut session_full = GitSession::open_via_channels(
|
||
client_full,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the full-fetch construction");
|
||
let full_packed = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session_full.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: Vec::new(),
|
||
done: true,
|
||
}],
|
||
&mut Vec::new(),
|
||
),
|
||
)
|
||
.await
|
||
.expect("the full fetch is timely")
|
||
.expect("the full pack delivered");
|
||
|
||
assert!(
|
||
trimmed < full_packed[0].pack_bytes.expect("the full fetch delivered"),
|
||
"the done round's haves trimmed the pack below the no-haves shape"
|
||
);
|
||
let decoded = decode_pack(&pack);
|
||
assert_eq!(
|
||
decoded
|
||
.objects
|
||
.iter()
|
||
.map(|(oid, _kind, _body)| *oid)
|
||
.collect::<BTreeSet<_>>(),
|
||
BTreeSet::from([chain.commit_b, chain.tree_b, chain.blob_b]),
|
||
"closure(commit_b) − closure(commit_a)"
|
||
);
|
||
let grammar = String::from_utf8(received.lock().expect("spy record").clone())
|
||
.expect("the request grammar is text");
|
||
assert_eq!(
|
||
grammar.matches("want ").count(),
|
||
3,
|
||
"each stateless round re-sent the want (three rounds)"
|
||
);
|
||
}
|
||
|
||
// --- the session persistence and the budget rows -----------------------------
|
||
|
||
/// Contract stamp: ADR-023 §2 (one connection = one service machine —
|
||
/// the V2 session persists across ls-refs → fetch), ADR-003's
|
||
/// advertisement-once discipline (the duplex entry's serving shape);
|
||
/// core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn the_session_persists_across_ls_refs_fetch_on_one_stream() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let chain = two_commit_chain(&handles.2.objects, &record);
|
||
seed_chain_refs(&handles, &record, &chain).await;
|
||
let (client, received, sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the channels-mode construction");
|
||
let listed = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.ls_refs(&LsRefsArgs {
|
||
symrefs: true,
|
||
..LsRefsArgs::default()
|
||
}),
|
||
)
|
||
.await
|
||
.expect("ls-refs is timely")
|
||
.expect("the listing");
|
||
assert_eq!(listed.len(), 2, "the chain's two commit refs listed");
|
||
|
||
let mut pack = Vec::new();
|
||
let reports = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![chain.commit_a],
|
||
haves: Vec::new(),
|
||
done: true,
|
||
}],
|
||
&mut pack,
|
||
),
|
||
)
|
||
.await
|
||
.expect("the fetch is timely")
|
||
.expect("the fetch on the same stream");
|
||
assert_eq!(reports.len(), 1);
|
||
assert!(
|
||
reports[0].pack_bytes.is_some(),
|
||
"the pack delivered on the persisted session"
|
||
);
|
||
|
||
let sent_bytes = sent.lock().expect("spy record").clone();
|
||
assert_eq!(
|
||
sent_bytes
|
||
.windows(b"version 2".len())
|
||
.filter(|window| window == b"version 2")
|
||
.count(),
|
||
1,
|
||
"one advertisement was emitted for the session (advertisement-once)"
|
||
);
|
||
let received_bytes = received.lock().expect("spy record").clone();
|
||
let grammar = String::from_utf8(received_bytes).expect("the request grammar is text");
|
||
assert!(
|
||
grammar.contains("command=ls-refs") && grammar.contains("command=fetch"),
|
||
"both commands rode the one stream: {grammar}"
|
||
);
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §6 / ADR-009 (the negotiation-rounds
|
||
/// budget, client-side; a breach ends the session with a typed error —
|
||
/// fail-closed and terminal); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn the_negotiation_rounds_budget_fails_closed_typed() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let (client, received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
let limits = Limits {
|
||
max_negotiation_rounds: 2,
|
||
..Limits::default()
|
||
};
|
||
let round = || FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: false,
|
||
};
|
||
let mut session =
|
||
GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
|
||
.expect("the construction");
|
||
let error = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(vec![round(), round(), round()], &mut Vec::new()),
|
||
)
|
||
.await
|
||
.expect("the refusal is immediate")
|
||
.expect_err("the plan is over the rounds budget");
|
||
match &error {
|
||
SessionError::BudgetExceeded {
|
||
what,
|
||
budget_name,
|
||
limit,
|
||
} => {
|
||
assert!(what.contains("negotiation"), "{what}");
|
||
assert_eq!(*budget_name, "max_negotiation_rounds");
|
||
assert_eq!(*limit, 2);
|
||
}
|
||
other => panic!("expected the typed rounds breach, got {other:?}"),
|
||
}
|
||
assert!(
|
||
received.lock().expect("spy record").is_empty(),
|
||
"the breach refused before anything touched the wire"
|
||
);
|
||
let again = session
|
||
.fetch(Vec::new(), &mut Vec::new())
|
||
.await
|
||
.unwrap_err();
|
||
assert_eq!(error, again, "the breach is terminal (error-break)");
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §6 / ADR-009 (the wall clock bounds every
|
||
/// operation, fetch included — the stalled peer is the client-side
|
||
/// breach); core/w4-session-fetch.
|
||
#[tokio::test]
|
||
async fn the_wall_clock_bounds_the_fetch_operation_and_the_breach_is_terminal() {
|
||
// The peer answers the handshake's advertisement, then stalls (no
|
||
// fetch response ever arrives).
|
||
let script = vec![
|
||
text_line(b"version 2"),
|
||
text_line(b"ls-refs=unborn"),
|
||
text_line(b"fetch=wait-for-done"),
|
||
text_line(b"object-format=sha1"),
|
||
FLUSH.to_vec(),
|
||
];
|
||
let client = spawn_fake_peer(script).await;
|
||
let limits = Limits {
|
||
session_wall_clock: Duration::from_millis(300),
|
||
..Limits::default()
|
||
};
|
||
let mut session =
|
||
GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
|
||
.expect("the construction");
|
||
tokio::time::timeout(Duration::from_secs(5), session.handshake())
|
||
.await
|
||
.expect("the handshake is timely")
|
||
.expect("the advertisement consumed");
|
||
let start = tokio::time::Instant::now();
|
||
let done_round = || FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: true,
|
||
};
|
||
let error = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(vec![done_round()], &mut Vec::new()),
|
||
)
|
||
.await
|
||
.expect("the fetch answers at the breach's scale")
|
||
.expect_err("the stalled peer breached the wall clock");
|
||
assert!(
|
||
start.elapsed() < Duration::from_secs(5),
|
||
"the breach answers at the budget's scale"
|
||
);
|
||
match &error {
|
||
SessionError::WallClock { budget } => assert_eq!(*budget, Duration::from_millis(300)),
|
||
other => panic!("expected the wall-clock breach, got {other:?}"),
|
||
}
|
||
let again = session
|
||
.fetch(vec![done_round()], &mut Vec::new())
|
||
.await
|
||
.unwrap_err();
|
||
assert_eq!(error, again, "the breach re-reports without re-entry");
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §6 / ADR-009 (the response-pack budget —
|
||
/// `max_pack_size`, the construction task's recorded mapping; the sink
|
||
/// never receives a byte past the budget); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn the_response_pack_budget_fails_closed_typed() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let chain = two_commit_chain(&handles.2.objects, &record);
|
||
let (client, _received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
let limits = Limits {
|
||
max_pack_size: 32,
|
||
..Limits::default()
|
||
};
|
||
let mut session =
|
||
GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
|
||
.expect("the construction");
|
||
let mut pack = Vec::new();
|
||
let done_round = || FetchRound {
|
||
wants: vec![chain.commit_b],
|
||
haves: Vec::new(),
|
||
done: true,
|
||
};
|
||
let error = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(vec![done_round()], &mut pack),
|
||
)
|
||
.await
|
||
.expect("the breach is timely")
|
||
.expect_err("the pack is over the budget");
|
||
match &error {
|
||
SessionError::BudgetExceeded {
|
||
what,
|
||
budget_name,
|
||
limit,
|
||
} => {
|
||
assert!(what.contains("pack"), "{what}");
|
||
assert_eq!(*budget_name, "max_pack_size");
|
||
assert_eq!(*limit, 32);
|
||
}
|
||
other => panic!("expected the typed pack budget breach, got {other:?}"),
|
||
}
|
||
assert!(
|
||
pack.len() as u64 <= 32,
|
||
"the sink never received a byte past the budget: {}",
|
||
pack.len()
|
||
);
|
||
let again = session
|
||
.fetch(vec![done_round()], &mut Vec::new())
|
||
.await
|
||
.unwrap_err();
|
||
assert_eq!(error, again, "the breach is terminal (error-break)");
|
||
}
|
||
|
||
/// Contract stamp: ADR-017 §1 (the caller's negotiation shape), the
|
||
/// fail-before-the-wire posture (the `invalid` class, nothing on the
|
||
/// wire); core/w4-session-fetch.
|
||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||
async fn a_no_done_round_without_haves_and_a_fetch_on_a_push_session_refuse_locally() {
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let (client, received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the construction");
|
||
let error = session
|
||
.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: Vec::new(),
|
||
done: false,
|
||
}],
|
||
&mut Vec::new(),
|
||
)
|
||
.await
|
||
.expect_err("a no-done round without haves is out of grammar (gix's assert shape)");
|
||
match &error {
|
||
SessionError::InvalidInput(message) => {
|
||
assert!(message.contains("at least one have"), "{message}");
|
||
}
|
||
other => panic!("expected the local refusal, got {other:?}"),
|
||
}
|
||
assert!(
|
||
received.lock().expect("spy record").is_empty(),
|
||
"the refusal happened before anything touched the wire"
|
||
);
|
||
drop(session);
|
||
|
||
// The done round terminates the negotiation: a round after it is
|
||
// out of grammar.
|
||
let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
|
||
let (client, _received, _sent, _joiner) = serve_twin(
|
||
&handles,
|
||
record.clone(),
|
||
&holder,
|
||
AccessAction::Read,
|
||
Limits::default(),
|
||
)
|
||
.await;
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the construction");
|
||
match session
|
||
.fetch(
|
||
vec![
|
||
FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: true,
|
||
},
|
||
FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: false,
|
||
},
|
||
],
|
||
&mut Vec::new(),
|
||
)
|
||
.await
|
||
{
|
||
Err(SessionError::InvalidInput(message)) => {
|
||
assert!(message.contains("terminates"), "{message}");
|
||
}
|
||
other => panic!("expected the post-done refusal, got {other:?}"),
|
||
}
|
||
drop(session);
|
||
|
||
// Wrong-service misuse: the fetch machine refuses a push session.
|
||
let (client_end, _server_end) = duplex(1 << 16);
|
||
let (client_read, client_write) = tokio::io::split(client_end);
|
||
let mut push_session = GitSession::connect_direct(
|
||
BiStream::from_joined(client_read, client_write),
|
||
"owner/doubled",
|
||
Service::ReceivePack,
|
||
None,
|
||
Limits::default(),
|
||
)
|
||
.expect("the push construction");
|
||
match push_session
|
||
.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: false,
|
||
}],
|
||
&mut Vec::new(),
|
||
)
|
||
.await
|
||
{
|
||
Err(SessionError::InvalidInput(message)) => {
|
||
assert!(message.contains("upload-pack"), "{message}");
|
||
}
|
||
other => panic!("expected the wrong-service refusal, got {other:?}"),
|
||
}
|
||
match push_session.ls_refs(&LsRefsArgs::default()).await {
|
||
Err(SessionError::InvalidInput(_)) => {}
|
||
other => panic!("expected the wrong-service refusal for ls-refs, got {other:?}"),
|
||
}
|
||
}
|
||
|
||
/// The hostile-shape row: a peer that sends `ready` on an ack round and
|
||
/// a peer whose done round carries no packfile section each fail
|
||
/// closed with a typed protocol error — the wait-for-done posture has
|
||
/// no ready, and the done round's pack is its terminator.
|
||
///
|
||
/// Contract stamp: ADR-014 §4 (ready is never in the grammar), ADR-014
|
||
/// §2 (the working done round's response is the packfile section);
|
||
/// core/w4-session-fetch.
|
||
#[tokio::test]
|
||
async fn the_hostile_shape_rows_refuse_the_off_contract_responses_typed() {
|
||
// ready on the ack round.
|
||
let script = vec![
|
||
text_line(b"version 2"),
|
||
text_line(b"ls-refs=unborn"),
|
||
text_line(b"fetch=wait-for-done"),
|
||
text_line(b"object-format=sha1"),
|
||
FLUSH.to_vec(),
|
||
text_line(b"acknowledgments"),
|
||
text_line(b"ready"),
|
||
DELIM.to_vec(),
|
||
text_line(b"packfile"),
|
||
pkt(b"\x01packdata"),
|
||
FLUSH.to_vec(),
|
||
];
|
||
let client = spawn_fake_peer(script).await;
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the construction");
|
||
tokio::time::timeout(Duration::from_secs(5), session.handshake())
|
||
.await
|
||
.expect("the handshake is timely")
|
||
.expect("the advertisement consumed");
|
||
let error = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: false,
|
||
}],
|
||
&mut Vec::new(),
|
||
),
|
||
)
|
||
.await
|
||
.expect("the round is timely")
|
||
.expect_err("ready is off grammar (the wait-for-done posture)");
|
||
match &error {
|
||
SessionError::Protocol(message) => assert!(message.contains("ready"), "{message}"),
|
||
other => panic!("expected the typed protocol refusal, got {other:?}"),
|
||
}
|
||
assert!(
|
||
session
|
||
.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: true,
|
||
}],
|
||
&mut Vec::new(),
|
||
)
|
||
.await
|
||
.is_err(),
|
||
"the session is terminal after the refusal"
|
||
);
|
||
|
||
// A done round answered with acknowledgments only.
|
||
let script = vec![
|
||
text_line(b"version 2"),
|
||
text_line(b"ls-refs=unborn"),
|
||
text_line(b"fetch=wait-for-done"),
|
||
text_line(b"object-format=sha1"),
|
||
FLUSH.to_vec(),
|
||
text_line(b"acknowledgments"),
|
||
text_line(b"NAK"),
|
||
FLUSH.to_vec(),
|
||
];
|
||
let client = spawn_fake_peer(script).await;
|
||
let mut session = GitSession::open_via_channels(
|
||
client,
|
||
"owner/doubled",
|
||
Service::UploadPack,
|
||
Limits::default(),
|
||
)
|
||
.expect("the construction");
|
||
tokio::time::timeout(Duration::from_secs(5), session.handshake())
|
||
.await
|
||
.expect("the handshake is timely")
|
||
.expect("the advertisement consumed");
|
||
let error = tokio::time::timeout(
|
||
Duration::from_secs(5),
|
||
session.fetch(
|
||
vec![FetchRound {
|
||
wants: vec![ObjectId::from([1u8; 20])],
|
||
haves: vec![ObjectId::from([2u8; 20])],
|
||
done: true,
|
||
}],
|
||
&mut Vec::new(),
|
||
),
|
||
)
|
||
.await
|
||
.expect("the round is timely")
|
||
.expect_err("the done round's packfile section is mandatory");
|
||
match &error {
|
||
SessionError::Protocol(message) => {
|
||
assert!(message.contains("packfile"), "{message}")
|
||
}
|
||
other => panic!("expected the typed protocol refusal, got {other:?}"),
|
||
}
|
||
}
|
||
|
||
/// The session surface's honesty pin (ADR-017 §4's storage-agnostic
|
||
/// shape): the machine module imports and names no object-store or
|
||
/// pack machinery of any kind — the session is the transport + grammar
|
||
/// layer, the pack bytes stream to the caller's consumer.
|
||
///
|
||
/// Contract stamp: ADR-017 §4 (storage-agnostic), ADR-010 (the
|
||
/// odb-free protocol core); core/w4-session-fetch.
|
||
#[test]
|
||
fn the_session_machine_carries_no_object_store_surface() {
|
||
let source = include_str!("../src/session.rs");
|
||
for banned in [
|
||
"gix_odb",
|
||
"gix_object",
|
||
"gix_pack::",
|
||
"gix_traverse",
|
||
"use gix_pack",
|
||
] {
|
||
assert!(
|
||
!source.contains(banned),
|
||
"the session machine names the object-store/pack machinery {banned:?} — ADR-017 §4's \
|
||
storage-agnostic shape forbids it"
|
||
);
|
||
}
|
||
}
|