glm-5.3-flash
0dd5721290
The publish bump (core/w5-publish) over the green wave-5 gate:
- manifest: version 0.0.1 -> 0.1.0; the substrate's inter-crate pin
follows (alkgit = { path = "..", version = "0.1.0" })
- the family LICENSE-MIT / LICENSE-APACHE files added (every sibling
crate ships them; the package listing gained the two files)
- CHANGELOG: the 0.1.0 entry dated and the stale 0.0.1-note dropped
- tests: the two agent-string pins moved from the hardcoded version
to env!(CARGO_PKG_VERSION) — the src side builds the agent string
from the manifest version, so the pins now hold across bumps
Verification: cargo test (workspace), cargo test --all-features,
cargo test -p alkgit, both clippies (-D warnings), cargo fmt --check,
cargo doc --no-deps (0 warnings), the fuzz corpus gate (13/13),
cargo publish --dry-run --allow-dirty (77-file package) — all green.
1132 lines
39 KiB
Rust
1132 lines
39 KiB
Rust
//! The wave-4 GitSession push client rows (core/w4-session-push): the
|
|
//! hand-rolled send-pack client (ADR-017 §3 — gitoxide has no
|
|
//! send-pack; a mirror of ADR-013's server shapes at the same honesty
|
|
//! bar, capture-grounded per push-captures.md, POC-X finding 3) driven
|
|
//! against the in-crate receive-pack server in-process — the V0
|
|
//! advertisement parse (the sentinel included), the command section's
|
|
//! zero-id shapes with the caps on the first line only (the wire bytes
|
|
//! byte-pinned), the caller-supplied pack streamed raw after the
|
|
//! command flush (thin included), the status report parsed in both
|
|
//! framings (sideband band-1-wrapped and bare, byte-checked against
|
|
//! our server's emission both ways), the atomic denial's aborted-refs
|
|
//! report, the client riders (the packless write-side-open shape per
|
|
//! ADR-020 §3 and the report-drain RST discipline per ADR-023 §5,
|
|
//! asserted from both sides), and the budgets (wall clock, the
|
|
//! report's max_pack_size mapping, fail-closed, terminal).
|
|
//!
|
|
//! Rows are names + `Contract stamp:` versions, per `tests/common`
|
|
//! §The row/stamp discipline.
|
|
|
|
mod common;
|
|
|
|
use std::collections::{BTreeMap, HashMap};
|
|
use std::pin::Pin;
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::task::{Context, Poll};
|
|
use std::time::Duration;
|
|
|
|
use gix_hash::ObjectId;
|
|
use gix_pack::data::entry::Header as PackHeader;
|
|
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt, ReadBuf};
|
|
use tokio::sync::Semaphore;
|
|
|
|
use alkcall::core::auth::Identity;
|
|
use alkcall::core::types::BiStream;
|
|
use alkgit::{
|
|
serve_duplex, AccessAction, GitPackGen, GitPackIngest, GitRefs, GitRegistryStore, GitSession,
|
|
Limits, PushAdvertisedRef, PushReport, PushSendOptions, RefOutcome, RefUpdate, RepoRecord,
|
|
Service, SessionError, SubstrateError, Visibility,
|
|
};
|
|
use common::{
|
|
blob_fixture, commit_fixture, delta_stream, loose_body, pack_bytes, pack_entry_base,
|
|
pack_entry_delta, parse_pkt_lines, sideband_report_lines, tree_fixture, MemPackGen,
|
|
MemPackIngest, MemRefs, MemRegistry,
|
|
};
|
|
|
|
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 SessionJoiner = tokio::task::JoinHandle<Result<(), SubstrateError>>;
|
|
|
|
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 }),
|
|
)
|
|
}
|
|
|
|
/// The fixture: the doubles' registry double plus the granted repo row.
|
|
async fn fixture_fresh(holder: &str) -> (Handles, RepoRecord, Identity) {
|
|
let handles = handles();
|
|
let record = repo("owner/doubled");
|
|
let holder = identity(holder);
|
|
let mut granted = record.clone();
|
|
granted
|
|
.grants
|
|
.insert(holder.id.clone(), vec![AccessAction::Write]);
|
|
let registry = Arc::clone(&handles.0);
|
|
registry.put_repo(&granted).await.expect("the repo row");
|
|
(handles, granted, holder)
|
|
}
|
|
|
|
// --- the wire spies --------------------------------------------------------
|
|
|
|
struct RecordRead<S> {
|
|
inner: S,
|
|
record: Arc<Mutex<Vec<u8>>>,
|
|
}
|
|
|
|
impl<S: tokio::io::AsyncRead + Unpin> tokio::io::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,
|
|
}
|
|
}
|
|
}
|
|
|
|
struct RecordWrite<S> {
|
|
inner: S,
|
|
record: Arc<Mutex<Vec<u8>>>,
|
|
shutdowns: Arc<AtomicUsize>,
|
|
}
|
|
|
|
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();
|
|
this.shutdowns.fetch_add(1, Ordering::SeqCst);
|
|
Pin::new(&mut this.inner).poll_shutdown(cx)
|
|
}
|
|
}
|
|
|
|
fn shutdown_counter() -> Arc<AtomicUsize> {
|
|
Arc::new(AtomicUsize::new(0))
|
|
}
|
|
|
|
/// Spawns the in-crate duplex serving entry (`serve_duplex`) for the
|
|
/// receive-pack service over a spy-wrapped pair; returns the client
|
|
/// end plus both wire records, the write-shutdown counter, and the
|
|
/// serving task's joiner.
|
|
fn serve_twin(
|
|
handles: &Handles,
|
|
record: RepoRecord,
|
|
holder: &Identity,
|
|
) -> (
|
|
BiStream,
|
|
WireRecord,
|
|
WireRecord,
|
|
Arc<AtomicUsize>,
|
|
SessionJoiner,
|
|
) {
|
|
let checked = alkgit::AuthorizedRepo::authorize(record, Some(holder), AccessAction::Write)
|
|
.expect("the checked write 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 shutdowns = shutdown_counter();
|
|
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),
|
|
shutdowns: Arc::clone(&shutdowns),
|
|
},
|
|
);
|
|
let joiner = tokio::spawn(async move {
|
|
serve_duplex(
|
|
&checked,
|
|
server,
|
|
Limits::default(),
|
|
Service::ReceivePack,
|
|
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,
|
|
shutdowns,
|
|
joiner,
|
|
)
|
|
}
|
|
|
|
/// Opens the push session over the serving twin (the channels-mode
|
|
/// preamble rode the open op — zero preamble bytes) and runs its
|
|
/// handshake; the session carries `session_limits` (the client-side
|
|
/// budgets; the server twin runs on the defaults).
|
|
async fn open_push_session(
|
|
handles: &Handles,
|
|
record: RepoRecord,
|
|
holder: &Identity,
|
|
session_limits: Limits,
|
|
) -> (
|
|
(WireRecord, WireRecord, Arc<AtomicUsize>, SessionJoiner),
|
|
GitSession,
|
|
) {
|
|
let twin = serve_twin(handles, record, holder);
|
|
let mut session = GitSession::open_via_channels(
|
|
twin.0,
|
|
"owner/doubled",
|
|
Service::ReceivePack,
|
|
session_limits,
|
|
)
|
|
.expect("the construction");
|
|
session
|
|
.handshake()
|
|
.await
|
|
.expect("the push handshake drains the advertisement");
|
|
((twin.1, twin.2, twin.3, twin.4), session)
|
|
}
|
|
|
|
/// The client end of a raw duplex pair as the session's BiStream.
|
|
fn client_side(stream: tokio::io::DuplexStream) -> BiStream {
|
|
let (read, write) = tokio::io::split(stream);
|
|
BiStream::from_joined(read, write)
|
|
}
|
|
|
|
async fn seed_ref(handles: &Handles, record: &RepoRecord, name: &str, oid: ObjectId) {
|
|
Arc::clone(&handles.1)
|
|
.apply_updates(
|
|
record,
|
|
vec![RefUpdate {
|
|
name: name.to_owned(),
|
|
expected: None,
|
|
new: Some(oid),
|
|
}],
|
|
false,
|
|
)
|
|
.await
|
|
.expect("the seed");
|
|
}
|
|
|
|
fn create(name: &str, new: ObjectId) -> RefUpdate {
|
|
RefUpdate {
|
|
name: name.to_owned(),
|
|
expected: None,
|
|
new: Some(new),
|
|
}
|
|
}
|
|
|
|
fn delete(name: &str, current: ObjectId) -> RefUpdate {
|
|
RefUpdate {
|
|
name: name.to_owned(),
|
|
expected: Some(current),
|
|
new: None,
|
|
}
|
|
}
|
|
|
|
fn update(name: &str, expected: ObjectId, new: ObjectId) -> RefUpdate {
|
|
RefUpdate {
|
|
name: name.to_owned(),
|
|
expected: Some(expected),
|
|
new: Some(new),
|
|
}
|
|
}
|
|
|
|
/// Frames one command line the client's text-mode writer frames it:
|
|
/// the LF-augmented pkt-line (the writer assures the trailing newline).
|
|
fn pkt_line(payload: &[u8]) -> Vec<u8> {
|
|
let total = payload.len() + 1 + 4;
|
|
let mut out = Vec::with_capacity(total);
|
|
out.extend_from_slice(format!("{total:04x}").as_bytes());
|
|
out.extend_from_slice(payload);
|
|
out.push(b'\n');
|
|
out
|
|
}
|
|
|
|
/// Splits the server's sent record at the advertisement's flush (the
|
|
/// server speaks first: the advertisement precedes the report) — the
|
|
/// flush is the framing boundary, not a substring of the payloads.
|
|
fn split_advertisement(body: &[u8]) -> (&[u8], &[u8]) {
|
|
let mut cursor = body;
|
|
while !cursor.is_empty() {
|
|
let header = std::str::from_utf8(&cursor[..4]).expect("hex header");
|
|
let len = usize::from_str_radix(header, 16).expect("hex length");
|
|
if len < 4 {
|
|
let offset = body.len() - cursor.len();
|
|
return (&body[..offset], &body[offset + 4..]);
|
|
}
|
|
cursor = &cursor[len..];
|
|
}
|
|
(body, &[])
|
|
}
|
|
|
|
/// The first command line's caps segment, test-side (the shape the
|
|
/// wire pins): report-status always, the caller's selections,
|
|
/// delete-refs auto-ridden on deletes, the pushed object format, the
|
|
/// crate agent.
|
|
fn first_caps(options: &PushSendOptions, has_deletes: bool) -> String {
|
|
let mut tokens = vec![if options.report_status_v2 {
|
|
"report-status-v2"
|
|
} else {
|
|
"report-status"
|
|
}];
|
|
if options.side_band_64k {
|
|
tokens.push("side-band-64k");
|
|
}
|
|
if options.atomic {
|
|
tokens.push("atomic");
|
|
}
|
|
if has_deletes {
|
|
tokens.push("delete-refs");
|
|
}
|
|
format!(
|
|
"{} object-format=sha1 agent=alkgit/{}",
|
|
tokens.join(" "),
|
|
env!("CARGO_PKG_VERSION")
|
|
)
|
|
}
|
|
|
|
/// A commit-chain fixture: two commits' worth of objects; the chain
|
|
/// carries the bodies (stripped loose encoding) the thin-pack row
|
|
/// re-assembles pack entries from. `seed_child` seeds the child-side
|
|
/// objects too (the ordinary row's generation source; the thin row
|
|
/// leaves them unseeded so the pack lands them).
|
|
struct Chain {
|
|
commit_a: ObjectId,
|
|
commit_b: ObjectId,
|
|
blob_a: ObjectId,
|
|
blob_b: ObjectId,
|
|
tree_b_body: Vec<u8>,
|
|
commit_b_body: Vec<u8>,
|
|
}
|
|
|
|
fn chain_fixture(objects: &common::MemObjects, doubled: &RepoRecord, seed_child: bool) -> Chain {
|
|
let (blob_a, blob_a_bytes) = blob_fixture(b"root content");
|
|
let blob_b_content: &'static [u8] = b"root content\nchild content";
|
|
let blob_b =
|
|
gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, blob_b_content)
|
|
.expect("the fixture hashes its own object");
|
|
let (tree_a, _) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]);
|
|
let (commit_a, _) = commit_fixture(tree_a, &[], "alpha");
|
|
let (tree_b, tree_b_bytes) = tree_fixture(&[
|
|
(gix_object::tree::EntryKind::Blob, "file", blob_a),
|
|
(gix_object::tree::EntryKind::Blob, "child", blob_b),
|
|
]);
|
|
let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
|
|
for (oid, bytes) in [
|
|
(blob_a, blob_a_bytes),
|
|
(
|
|
tree_a,
|
|
tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]).1,
|
|
),
|
|
(commit_a, commit_fixture(tree_a, &[], "alpha").1),
|
|
] {
|
|
objects.put(doubled, oid, bytes);
|
|
}
|
|
if seed_child {
|
|
let blob_b_bytes = blob_fixture(blob_b_content).1;
|
|
for (oid, bytes) in [
|
|
(blob_b, blob_b_bytes),
|
|
(tree_b, tree_b_bytes),
|
|
(commit_b, commit_b_bytes),
|
|
] {
|
|
objects.put(doubled, oid, bytes);
|
|
}
|
|
}
|
|
Chain {
|
|
commit_a,
|
|
commit_b,
|
|
blob_a,
|
|
blob_b,
|
|
tree_b_body: loose_body(
|
|
&tree_fixture(&[
|
|
(gix_object::tree::EntryKind::Blob, "file", blob_a),
|
|
(gix_object::tree::EntryKind::Blob, "child", blob_b),
|
|
])
|
|
.1,
|
|
)
|
|
.1,
|
|
commit_b_body: loose_body(&commit_fixture(tree_b, &[commit_a], "beta").1).1,
|
|
}
|
|
}
|
|
|
|
/// Contract stamp: ADR-017 §3/§4 (the ordinary push — commands in the
|
|
/// zero-id create shapes, the caller's pack raw after the command
|
|
/// flush, the report parsed to typed outcomes), ADR-013 §2/§3/§8 (the
|
|
/// caps NUL-attached on the first line only — byte-pinned; the
|
|
/// sideband framing byte-checked from the server's emission),
|
|
/// push-captures.md §Request/§Status report; core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn an_ordinary_push_sends_commands_and_pack_then_parses_the_report() {
|
|
let (handles, record, holder) = fixture_fresh("u-write").await;
|
|
let chain = chain_fixture(handles.3.objects(), &record, true);
|
|
let sink = common::SharedSink::default();
|
|
Arc::clone(&handles.2)
|
|
.generate(
|
|
&record,
|
|
vec![chain.commit_b],
|
|
Vec::new(),
|
|
Limits::default(),
|
|
Box::new(sink.clone()),
|
|
)
|
|
.await
|
|
.expect("the caller's pack generation");
|
|
let pack = sink.snapshot();
|
|
|
|
let ((wire_received, wire_sent, write_shutdowns, server_joiner), mut session) =
|
|
open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
|
|
|
|
let advertised = session
|
|
.push_advertisement()
|
|
.await
|
|
.expect("the empty repo's advertisement");
|
|
assert!(
|
|
advertised.is_empty(),
|
|
"the empty repo's sentinel parses as the empty set: {advertised:?}"
|
|
);
|
|
|
|
let options = PushSendOptions {
|
|
side_band_64k: true,
|
|
..Default::default()
|
|
};
|
|
let report = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![create("refs/heads/main", chain.commit_b)],
|
|
Some(&mut &pack[..]),
|
|
&options,
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely")
|
|
.expect("the ordinary push lands");
|
|
|
|
assert_eq!(
|
|
report,
|
|
PushReport {
|
|
unpack: Ok(()),
|
|
refs: vec![RefOutcome {
|
|
name: "refs/heads/main".to_owned(),
|
|
result: Ok(()),
|
|
}],
|
|
},
|
|
"the report parses to the typed per-ref outcome"
|
|
);
|
|
let rows = Arc::clone(&handles.1)
|
|
.list_refs(&record)
|
|
.await
|
|
.expect("rows");
|
|
assert_eq!(rows.len(), 1, "the ref row landed");
|
|
assert_eq!(rows[0].oid, Some(chain.commit_b));
|
|
assert!(
|
|
handles.3.objects().contains(&record, &chain.blob_b),
|
|
"the pack's objects landed"
|
|
);
|
|
|
|
let zero = "0".repeat(40);
|
|
let caps = first_caps(&options, false);
|
|
let expected_client: Vec<u8> = [
|
|
pkt_line(format!("{zero} {} refs/heads/main\0 {caps}", chain.commit_b).as_bytes()),
|
|
b"0000".to_vec(),
|
|
pack,
|
|
]
|
|
.concat();
|
|
assert_eq!(
|
|
*wire_received.lock().expect("spy record"),
|
|
expected_client,
|
|
"the client's wire bytes: the command line (caps NUL-attached, zero-id create shape), \
|
|
the command flush, then the pack raw"
|
|
);
|
|
|
|
let sent = wire_sent.lock().expect("spy record").clone();
|
|
let (_advertisement, report_part) = split_advertisement(&sent);
|
|
assert_eq!(
|
|
sideband_report_lines(report_part),
|
|
vec!["unpack ok", "ok refs/heads/main"],
|
|
"the server's sideband-framed report parses to the same outcomes"
|
|
);
|
|
|
|
// The client drops the stream only after the report drained (the
|
|
// ADR-023 §5 shape); the server's serving session then completes.
|
|
drop(session);
|
|
server_joiner
|
|
.await
|
|
.expect("the server join")
|
|
.expect("the serving task completed cleanly — the push session closed after the report");
|
|
assert_eq!(
|
|
write_shutdowns.load(Ordering::SeqCst),
|
|
0,
|
|
"the client never shut the write half down mid-push (the rider's shape)"
|
|
);
|
|
}
|
|
|
|
/// Contract stamp: ADR-013 §5 (thin packs accepted, bases from the
|
|
/// server odb — the caller's thin pack rides raw, the ref-delta base
|
|
/// resolves server-side), ADR-017 §4 (the caller-supplied pack);
|
|
/// core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn a_thin_push_streams_the_callers_thin_pack() {
|
|
let (handles, record, holder) = fixture_fresh("u-write").await;
|
|
let chain = chain_fixture(handles.3.objects(), &record, false);
|
|
assert!(
|
|
!handles.3.objects().contains(&record, &chain.blob_b),
|
|
"the child-side objects are unseeded — only the pack can land them"
|
|
);
|
|
|
|
let mut offset = 0u64;
|
|
// The delta's base is the server-held blob ("root content", 12
|
|
// bytes — the delta_stream fixture form copies it whole); the
|
|
// insertion appended reconstructs the child blob's body.
|
|
let delta = pack_entry_delta(
|
|
PackHeader::RefDelta {
|
|
base_id: chain.blob_a,
|
|
},
|
|
delta_stream(12, b"\nchild content").as_slice(),
|
|
offset,
|
|
);
|
|
offset += delta.bytes_in_pack();
|
|
let tree_entry = pack_entry_base(gix_object::Kind::Tree, &chain.tree_b_body, offset);
|
|
offset += tree_entry.bytes_in_pack();
|
|
let commit_entry = pack_entry_base(gix_object::Kind::Commit, &chain.commit_b_body, offset);
|
|
let thin_pack = pack_bytes(&[delta, tree_entry, commit_entry]);
|
|
|
|
let (_, mut session) =
|
|
open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
|
|
let advertised = session
|
|
.push_advertisement()
|
|
.await
|
|
.expect("the empty repo's advertisement");
|
|
assert!(advertised.is_empty());
|
|
|
|
let options = PushSendOptions::default();
|
|
let report = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![create("refs/heads/main", chain.commit_b)],
|
|
Some(&mut &thin_pack[..]),
|
|
&options,
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely")
|
|
.expect("the thin push lands");
|
|
assert_eq!(
|
|
report.unpack,
|
|
Ok(()),
|
|
"the thin pack ingests — the ref-delta base resolves from the server's own objects"
|
|
);
|
|
assert_eq!(
|
|
report.refs,
|
|
vec![RefOutcome {
|
|
name: "refs/heads/main".to_owned(),
|
|
result: Ok(()),
|
|
}]
|
|
);
|
|
assert!(
|
|
handles.3.objects().contains(&record, &chain.blob_b),
|
|
"the delta'd child blob landed (reconstructed from the server-held base)"
|
|
);
|
|
}
|
|
|
|
/// Contract stamp: ADR-013 §10 (the atomic semantics client-side: an
|
|
/// aborted transaction reports the aborted-refs shape — `atomic push
|
|
/// failure` on the would-have-succeeded refs, the denied refs keep
|
|
/// their own reasons), ADR-017 §3, ADR-013 §2 (the typed
|
|
/// advertisement rows); push-captures.md §CAS; core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn an_atomic_push_denial_parses_the_aborted_refs_report() {
|
|
let (handles, record, holder) = fixture_fresh("u-write").await;
|
|
let chain = chain_fixture(handles.3.objects(), &record, true);
|
|
seed_ref(&handles, &record, "refs/heads/keep", chain.commit_a).await;
|
|
|
|
let (_, mut session) =
|
|
open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
|
|
let advertised = session
|
|
.push_advertisement()
|
|
.await
|
|
.expect("the advertisement");
|
|
assert_eq!(
|
|
advertised,
|
|
vec![PushAdvertisedRef {
|
|
oid: chain.commit_a,
|
|
name: "refs/heads/keep".to_owned(),
|
|
}],
|
|
"the seeded advertisement parses to the typed ref rows"
|
|
);
|
|
|
|
// One command is CAS-stale (a wrong expected id), one would have
|
|
// succeeded: the atomic transaction aborts whole.
|
|
let stale_expected =
|
|
gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"wrong")
|
|
.expect("a decoy oid");
|
|
let options = PushSendOptions {
|
|
atomic: true,
|
|
..Default::default()
|
|
};
|
|
let report = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![
|
|
create("refs/heads/other", chain.commit_b),
|
|
update("refs/heads/keep", stale_expected, chain.commit_b),
|
|
],
|
|
None::<&mut &[u8]>,
|
|
&options,
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely")
|
|
.expect("the atomic push reports");
|
|
assert_eq!(report.unpack, Ok(()));
|
|
assert_eq!(
|
|
report.refs,
|
|
vec![
|
|
RefOutcome {
|
|
name: "refs/heads/other".to_owned(),
|
|
result: Err("atomic push failure".to_owned()),
|
|
},
|
|
RefOutcome {
|
|
name: "refs/heads/keep".to_owned(),
|
|
result: Err("stale info".to_owned()),
|
|
},
|
|
],
|
|
"the aborted-refs report: would-have-succeeded carries the atomic failure, the denied \
|
|
ref keeps its own reason"
|
|
);
|
|
let rows = Arc::clone(&handles.1)
|
|
.list_refs(&record)
|
|
.await
|
|
.expect("rows");
|
|
assert_eq!(
|
|
rows.len(),
|
|
1,
|
|
"nothing applied — the transaction aborted whole"
|
|
);
|
|
assert_eq!(
|
|
rows[0].oid,
|
|
Some(chain.commit_a),
|
|
"the seeded ref untouched"
|
|
);
|
|
}
|
|
|
|
/// The packless delete-only push, both report framings byte-checked
|
|
/// from our server's emission (sideband selected, sideband not), with
|
|
/// the ADR-020 §3 rider asserted from both sides: the client holds the
|
|
/// write side open until the report (no shutdown), and the server's
|
|
/// bounded peek takes the packless path — the report arrives and the
|
|
/// serving task completes cleanly, never a partial-header EOF error.
|
|
///
|
|
/// Contract stamp: ADR-020 §3 (the write-side-open rider, asserted
|
|
/// from both sides), ADR-020 §1 (our server's peek serves the
|
|
/// packless client's waiting shape), ADR-013 §8 (both framings), §3
|
|
/// (the zero-id delete shape, byte-pinned), ADR-023 §5 (the report is
|
|
/// read in full before the stream could drop); core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn a_packless_delete_push_holds_the_write_side_open_across_both_report_framings() {
|
|
for (why, options) in [
|
|
(
|
|
"the sideband framing",
|
|
PushSendOptions {
|
|
side_band_64k: true,
|
|
..Default::default()
|
|
},
|
|
),
|
|
("the bare framing", PushSendOptions::default()),
|
|
] {
|
|
let (handles, record, holder) = fixture_fresh("u-write").await;
|
|
let chain = chain_fixture(handles.3.objects(), &record, true);
|
|
seed_ref(&handles, &record, "refs/heads/dev", chain.commit_a).await;
|
|
|
|
let ((wire_received, wire_sent, write_shutdowns, server_joiner), mut session) =
|
|
open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
|
|
let advertised = session
|
|
.push_advertisement()
|
|
.await
|
|
.expect("the advertisement");
|
|
assert_eq!(
|
|
advertised,
|
|
vec![PushAdvertisedRef {
|
|
oid: chain.commit_a,
|
|
name: "refs/heads/dev".to_owned(),
|
|
}],
|
|
"{why}: the seeded advertisement"
|
|
);
|
|
|
|
let report = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![delete("refs/heads/dev", chain.commit_a)],
|
|
None::<&mut &[u8]>,
|
|
&options,
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely")
|
|
.expect("the packless delete push lands");
|
|
assert_eq!(
|
|
report,
|
|
PushReport {
|
|
unpack: Ok(()),
|
|
refs: vec![RefOutcome {
|
|
name: "refs/heads/dev".to_owned(),
|
|
result: Ok(()),
|
|
}],
|
|
},
|
|
"{why}: the delete reports ok"
|
|
);
|
|
let rows = Arc::clone(&handles.1)
|
|
.list_refs(&record)
|
|
.await
|
|
.expect("rows");
|
|
assert!(
|
|
rows.is_empty(),
|
|
"{why}: the ref row is deleted (the packless CAS-only path ran)"
|
|
);
|
|
|
|
let zero = "0".repeat(40);
|
|
let caps = first_caps(&options, true);
|
|
let expected_client: Vec<u8> = [
|
|
pkt_line(format!("{} {zero} refs/heads/dev\0 {caps}", chain.commit_a).as_bytes()),
|
|
b"0000".to_vec(),
|
|
]
|
|
.concat();
|
|
assert_eq!(
|
|
*wire_received.lock().expect("spy record"),
|
|
expected_client,
|
|
"{why}: the delete command (old = the current id, new = the zero id — \
|
|
ADR-013 §3's delete shape) with the caps NUL-attached, the command flush, and \
|
|
nothing after — the packless shape holds the write side open (no shutdown, no \
|
|
half-close) until the report"
|
|
);
|
|
assert_eq!(
|
|
write_shutdowns.load(Ordering::SeqCst),
|
|
0,
|
|
"{why}: the client never shut the write half down before the report"
|
|
);
|
|
|
|
let sent = wire_sent.lock().expect("spy record").clone();
|
|
let (_advertisement, report_part) = split_advertisement(&sent);
|
|
if options.side_band_64k {
|
|
assert_eq!(
|
|
sideband_report_lines(report_part),
|
|
vec!["unpack ok", "ok refs/heads/dev"],
|
|
"{why}: the band-1-wrapped report (the inner framing inside the band)"
|
|
);
|
|
} else {
|
|
assert_eq!(
|
|
parse_pkt_lines(report_part),
|
|
vec![
|
|
b"unpack ok\n".to_vec(),
|
|
b"ok refs/heads/dev\n".to_vec(),
|
|
Vec::new(),
|
|
],
|
|
"{why}: the bare report — pkt-lines to the single flush"
|
|
);
|
|
}
|
|
|
|
drop(session);
|
|
server_joiner.await.expect("the server join").expect(
|
|
"{why}: the server took the packless path and closed cleanly — never the \
|
|
partial-header EOF error",
|
|
);
|
|
}
|
|
}
|
|
|
|
/// The ADR-023 §5 discipline against a scripted peer: a packless push
|
|
/// holds the write side open while waiting for the report (the peer
|
|
/// observes no EOF while it delays), and the client reads the full
|
|
/// report — the response side drained to its flush — returning before
|
|
/// the peer's scripted close (no close-with-unread-receive-data on
|
|
/// either end).
|
|
///
|
|
/// Contract stamp: ADR-020 §3 (the write-side-open shape,
|
|
/// peer-observed), ADR-023 §5 (the drain-before-drop discipline);
|
|
/// core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn a_packless_push_holds_the_write_side_open_and_drains_the_full_report() {
|
|
let (client_end, server_end) = duplex(1 << 16);
|
|
let facts: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
|
|
let peer_facts = Arc::clone(&facts);
|
|
tokio::spawn(async move {
|
|
let (mut peer_read, mut peer_write) = tokio::io::split(server_end);
|
|
peer_write
|
|
.write_all(
|
|
&[
|
|
pkt_line(
|
|
format!(
|
|
"{} capabilities^{{}}\0report-status delete-refs object-format=sha1",
|
|
"0".repeat(40)
|
|
)
|
|
.as_bytes(),
|
|
),
|
|
b"0000".to_vec(),
|
|
]
|
|
.concat(),
|
|
)
|
|
.await
|
|
.expect("the advertisement writes");
|
|
// The client's commands, to the command flush.
|
|
loop {
|
|
let mut header = [0u8; 4];
|
|
if peer_read.read_exact(&mut header).await.is_err() {
|
|
peer_facts
|
|
.lock()
|
|
.expect("facts")
|
|
.push("eof before the report".to_owned());
|
|
return;
|
|
}
|
|
let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
|
|
.expect("hex length");
|
|
if len < 4 {
|
|
break;
|
|
}
|
|
let mut payload = vec![0u8; len - 4];
|
|
peer_read
|
|
.read_exact(&mut payload)
|
|
.await
|
|
.expect("the payload");
|
|
}
|
|
// The rider's shape: the client holds its write side open — no
|
|
// EOF arrives while the peer delays (a half-closing client
|
|
// would surface here as an unexpected read end).
|
|
let eof_watcher = tokio::spawn(async move {
|
|
let mut drained = Vec::new();
|
|
match peer_read.read_to_end(&mut drained).await {
|
|
Ok(_) => drained,
|
|
Err(_) => Vec::new(),
|
|
}
|
|
});
|
|
match tokio::time::timeout(Duration::from_millis(900), eof_watcher).await {
|
|
Ok(Ok(buf)) => {
|
|
if buf.is_empty() {
|
|
peer_facts
|
|
.lock()
|
|
.expect("facts")
|
|
.push("eof before the report".to_owned());
|
|
} else {
|
|
peer_facts
|
|
.lock()
|
|
.expect("facts")
|
|
.push("unexpected bytes while waiting".to_owned());
|
|
}
|
|
return;
|
|
}
|
|
Err(_still_open) => {}
|
|
Ok(Err(_)) => {
|
|
peer_facts
|
|
.lock()
|
|
.expect("facts")
|
|
.push("eof before the report".to_owned());
|
|
return;
|
|
}
|
|
}
|
|
// The report, then a scripted delayed close: the client must
|
|
// have returned after draining the full report.
|
|
peer_facts
|
|
.lock()
|
|
.expect("facts")
|
|
.push("report sent".to_owned());
|
|
peer_write
|
|
.write_all(
|
|
&[
|
|
b"000eunpack ok\n".to_vec(),
|
|
b"0016ok refs/heads/dev\n".to_vec(),
|
|
b"0000".to_vec(),
|
|
]
|
|
.concat(),
|
|
)
|
|
.await
|
|
.expect("the report writes");
|
|
tokio::time::sleep(Duration::from_millis(1500)).await;
|
|
let _ = peer_write.shutdown().await;
|
|
});
|
|
|
|
let mut session = GitSession::open_via_channels(
|
|
client_side(client_end),
|
|
"owner/doubled",
|
|
Service::ReceivePack,
|
|
Limits::default(),
|
|
)
|
|
.expect("the construction");
|
|
session.handshake().await.expect("the advertisement drain");
|
|
|
|
let options = PushSendOptions::default();
|
|
let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
|
|
.expect("a decoy oid");
|
|
let report = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![delete("refs/heads/dev", current)],
|
|
None::<&mut &[u8]>,
|
|
&options,
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely — it returned on the report, not on the peer's close")
|
|
.expect("the report drained in full");
|
|
assert_eq!(
|
|
report,
|
|
PushReport {
|
|
unpack: Ok(()),
|
|
refs: vec![RefOutcome {
|
|
name: "refs/heads/dev".to_owned(),
|
|
result: Ok(()),
|
|
}],
|
|
}
|
|
);
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
assert_eq!(
|
|
*facts.lock().expect("facts"),
|
|
vec!["report sent".to_owned()],
|
|
"the client held the write side open until the report and returned before the peer's \
|
|
scripted close (the report side was drained in full)"
|
|
);
|
|
tokio::time::sleep(Duration::from_millis(1600)).await;
|
|
}
|
|
|
|
/// Contract stamp: ADR-017 §6 / ADR-009 (the wall clock bounds the
|
|
/// operation client-side; the response report maps to `max_pack_size`
|
|
/// — the construction task's recorded mapping; breaches are typed and
|
|
/// fail-closed, terminal, re-reported); core/w4-session-push.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn the_push_budgets_fail_closed_typed() {
|
|
// The wall clock: a peer that never sends a report stalls the
|
|
// report read — the breach is the typed wall-clock error, terminal.
|
|
let (client_end, server_end) = duplex(1 << 16);
|
|
tokio::spawn(async move {
|
|
let mut stream = server_end;
|
|
stream
|
|
.write_all(
|
|
&[
|
|
pkt_line(
|
|
format!(
|
|
"{} capabilities^{{}}\0report-status object-format=sha1",
|
|
"0".repeat(40)
|
|
)
|
|
.as_bytes(),
|
|
),
|
|
b"0000".to_vec(),
|
|
]
|
|
.concat(),
|
|
)
|
|
.await
|
|
.expect("the advertisement writes");
|
|
tokio::time::sleep(Duration::from_secs(30)).await;
|
|
let _ = stream.shutdown().await;
|
|
});
|
|
let limits = Limits {
|
|
session_wall_clock: Duration::from_millis(400),
|
|
..Limits::default()
|
|
};
|
|
let mut session = GitSession::open_via_channels(
|
|
client_side(client_end),
|
|
"owner/doubled",
|
|
Service::ReceivePack,
|
|
limits,
|
|
)
|
|
.expect("the construction");
|
|
session.handshake().await.expect("the advertisement drain");
|
|
let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
|
|
.expect("a decoy oid");
|
|
let error = tokio::time::timeout(
|
|
Duration::from_secs(5),
|
|
session.push(
|
|
vec![delete("refs/heads/x", current)],
|
|
None::<&mut &[u8]>,
|
|
&Default::default(),
|
|
),
|
|
)
|
|
.await
|
|
.expect("the breach answers promptly")
|
|
.expect_err("the wall clock breached");
|
|
match &error {
|
|
SessionError::WallClock { budget } => {
|
|
assert_eq!(*budget, Duration::from_millis(400));
|
|
}
|
|
other => panic!("expected the typed wall-clock breach, got {other:?}"),
|
|
}
|
|
let again = session
|
|
.push(
|
|
vec![delete("refs/heads/x", current)],
|
|
None::<&mut &[u8]>,
|
|
&Default::default(),
|
|
)
|
|
.await
|
|
.expect_err("the session is terminal");
|
|
assert_eq!(error, again, "the breach re-reports without re-entry");
|
|
|
|
// The response report's budget: the report bytes map to
|
|
// max_pack_size — a report over the budget ends the session with
|
|
// the typed breach, fail-closed (nothing beyond the budget is
|
|
// ever retained).
|
|
let (handles, record, holder) = fixture_fresh("u-write").await;
|
|
let chain = chain_fixture(handles.3.objects(), &record, true);
|
|
seed_ref(&handles, &record, "refs/heads/dev", chain.commit_a).await;
|
|
let client_limits = Limits {
|
|
max_pack_size: 8,
|
|
..Limits::default()
|
|
};
|
|
let ((_, _, _, server_joiner), mut session) =
|
|
open_push_session(&handles, record.clone(), &holder, client_limits).await;
|
|
let error = tokio::time::timeout(
|
|
Duration::from_secs(10),
|
|
session.push(
|
|
vec![delete("refs/heads/dev", chain.commit_a)],
|
|
None::<&mut &[u8]>,
|
|
&Default::default(),
|
|
),
|
|
)
|
|
.await
|
|
.expect("the push is timely")
|
|
.expect_err("the report outgrew the budget");
|
|
match &error {
|
|
SessionError::BudgetExceeded {
|
|
what,
|
|
budget_name,
|
|
limit,
|
|
} => {
|
|
assert!(what.contains("report"), "{what}");
|
|
assert_eq!(*budget_name, "max_pack_size");
|
|
assert_eq!(*limit, 8);
|
|
}
|
|
other => panic!("expected the typed budget breach, got {other:?}"),
|
|
}
|
|
let again = session
|
|
.push(
|
|
vec![delete("refs/heads/dev", chain.commit_a)],
|
|
None::<&mut &[u8]>,
|
|
&Default::default(),
|
|
)
|
|
.await
|
|
.expect_err("the session is terminal");
|
|
assert_eq!(error, again, "the breach re-reports without re-entry");
|
|
// The client drops (the report side was drained to the breach);
|
|
// the server's serving session then completes.
|
|
drop(session);
|
|
server_joiner
|
|
.await
|
|
.expect("the server join")
|
|
.expect("the server side of the budget breach completes cleanly");
|
|
}
|
|
|
|
/// Local grammar refusals fail before the wire (nothing attached, and
|
|
/// nothing would have been sent): an empty command set (the caller
|
|
/// computes the diff; a nonzero command set is the minimum shape), a
|
|
/// push on an upload-pack session. Both end the session (error-break
|
|
/// terminal).
|
|
///
|
|
/// Contract stamp: ADR-017 §6 (the local `invalid` class,
|
|
/// fail-closed), ADR-016 §3 (one service machine per stream);
|
|
/// core/w4-session-push.
|
|
#[tokio::test]
|
|
async fn local_push_refusals_fail_before_the_wire_and_terminate() {
|
|
let (client_end, _server_end) = duplex(1 << 16);
|
|
let mut session = GitSession::open_via_channels(
|
|
client_side(client_end),
|
|
"owner/doubled",
|
|
Service::ReceivePack,
|
|
Limits::default(),
|
|
)
|
|
.expect("the construction");
|
|
let error = session
|
|
.push(vec![], None::<&mut &[u8]>, &Default::default())
|
|
.await
|
|
.expect_err("an empty command set refuses locally");
|
|
match &error {
|
|
SessionError::InvalidInput(message) => {
|
|
assert!(message.contains("empty"), "{message}");
|
|
}
|
|
other => panic!("expected the local refusal, got {other:?}"),
|
|
}
|
|
let again = session
|
|
.push(vec![], None::<&mut &[u8]>, &Default::default())
|
|
.await
|
|
.expect_err("the session is terminal");
|
|
assert_eq!(error, again, "the refusal re-reports without re-entry");
|
|
|
|
let mut wrong_service = GitSession::open_via_channels(
|
|
client_side(duplex(1 << 16).0),
|
|
"owner/doubled",
|
|
Service::UploadPack,
|
|
Limits::default(),
|
|
)
|
|
.expect("the construction");
|
|
let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
|
|
.expect("a decoy oid");
|
|
let error = wrong_service
|
|
.push(
|
|
vec![delete("refs/heads/x", current)],
|
|
None::<&mut &[u8]>,
|
|
&Default::default(),
|
|
)
|
|
.await
|
|
.expect_err("push never rides an upload-pack session");
|
|
match &error {
|
|
SessionError::InvalidInput(message) => {
|
|
assert!(message.contains("receive-pack"), "{message}");
|
|
}
|
|
other => panic!("expected the local refusal, got {other:?}"),
|
|
}
|
|
}
|