crates/gunnar-server/tests/mirror_sync.rs
//! **`Mirror.Sync` against a real upstream: enqueue, fetch, apply, and read the
//! objects back.**
//!
//! The verb answered `UNIMPLEMENTED` until now, so there is no round-trip to
//! regression-test and every assertion here is on **applied output**: a
//! repository that was empty holding the upstream's commit object, its
//! reference at the upstream's tip, and a job whose progress pair moved from
//! `(0, N)` to `(N, N)` while it did.
//!
//! # What each half proves, and why neither alone is enough
//!
//! * **SSH** (`a_mirror_sync_over_ssh_…`) runs against a whole real gunnar:
//! `Server::start` on a loopback russh listener, the real key table, the real
//! `WireTransport`, the real `upload_pack` over a real znippy store. The
//! client half is `gunnar_client::connect_upload_pack` — the fetch-side twin
//! of the dialler replication pushes with. Nothing about that path is
//! a fixture, which is the only way to know the two ends of `gunnar-wire`
//! actually meet.
//! * **Smart HTTP** (`a_mirror_sync_over_smart_http_…`) runs against a loopback
//! HTTP/1.1 listener in this file that drives the **same**
//! `UploadPack::serve_session(V2, StatelessRpc, …)` gunnar's own
//! `gunnar-http` drives, plus the two routes (`GET /info/refs`,
//! `POST /git-upload-pack`) it serves. What is written here is the HTTP
//! *framing* and nothing else — and framing is precisely what the adapter
//! under test does, so a fixture that reimplemented the protocol would be
//! testing itself.
//!
//! It is written here rather than reached through `gunnar-http` for one
//! reason: this file has to serve a **401**, and refusing a request before it
//! reaches the git seam is not something that crate can be asked to do
//! without inventing an auth mode nothing ships.
//!
//! # Both hash kinds, on purpose
//!
//! The git wire protocol's rule is that nothing may assume sha1, and a mirror is the one verb
//! where the assumption would be invisible: a sha1 client against a sha256
//! remote does not fail, it negotiates an empty answer, and the operator is
//! left with a command that reports success and copied nothing.
//! `fetch_pack::greet` refuses the mismatch by name, and these tests run the
//! whole path at each kind so the refusal is never the thing being exercised.
//!
//! # The credential is asserted ON THE BYTES
//!
//! `a_private_http_upstream_…` stands up an upstream that answers 401 without
//! the token and serves with it. That proves the token is *sent*. What matters
//! as much is that it is never seen anywhere else, so the test taps the tracing
//! subscriber, collects every RPC response and every error string, and asserts
//! the token's bytes appear in **none** of them. A `has_credential` flag that
//! is right while the value leaks through an error message is the failure this
//! shape exists to catch.
#![cfg(feature = "mirror-http")] // A32-18: the airgap default carries no HTTP client; the appliance rows and CI name `mirror-http`.
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use gix_object::Kind;
#[cfg(feature = "replication")]
use gunnar_core::model::{Grant, Level, PrincipalKind, Visibility};
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_grpc::seam::ControlPlane as _;
use gunnar_proto::{MirrorCredentialKind, MirrorJobState, RepoPath};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_store::conformance::pack::PackBuilder;
use gunnar_store::git_storage_trait::{GitOps, GitServe};
type Oid = gix_hash::ObjectId;
// ════════════════════════════════════════════════════════════════════════════
// Fixtures: a repository with real objects in it
// ════════════════════════════════════════════════════════════════════════════
fn compute(hash: gix_hash::Kind, kind: Kind, content: &[u8]) -> Oid {
gix_object::compute_hash(hash, kind, content).expect("hashing an object")
}
fn tree_bytes(entries: &[(&str, &str, Oid)]) -> Vec<u8> {
let mut body = Vec::new();
for (mode, name, oid) in entries {
body.extend_from_slice(mode.as_bytes());
body.push(b' ');
body.extend_from_slice(name.as_bytes());
body.push(0);
body.extend_from_slice(oid.as_bytes());
}
body
}
fn commit_bytes(tree: Oid, message: &str) -> Vec<u8> {
let who = "Alice <alice@example.com> 1700000000 +0000";
let mut body = format!("tree {}\n", tree.to_hex());
body.push_str(&format!("author {who}\n"));
body.push_str(&format!("committer {who}\n\n"));
body.push_str(message);
body.push('\n');
body.into_bytes()
}
/// One commit, one tree, one blob — three objects — under `refs/heads/main`.
///
/// Returns the commit id and the blob id: the commit is what the reference must
/// end up at on the mirror, and the blob is what proves the whole closure was
/// copied rather than only the tip.
///
/// Gated on `replication` because its one caller is `SshUpstream::start`, which
/// is itself `#[cfg(feature = "replication")]`. Ungated it is `dead_code` on a
/// default build — the one warning `cargo check -p gunnar-server --all-targets`
/// still emitted on 2026-09-12. `plant_a_commit_of` below stays ungated: it has
/// callers on both sides of the feature.
#[cfg(feature = "replication")]
fn plant_a_commit(git: &dyn GitServe, hash: gix_hash::Kind) -> (Oid, Oid) {
plant_a_commit_of(git, hash, b"the quick brown fox\n".to_vec())
}
/// **Incompressible bytes**, from a linear congruential generator so the fixture
/// is deterministic and pulls in no dependency.
///
/// Incompressible matters: the packfile is deflated, and a blob of zeroes would
/// produce a pack of a few hundred bytes however many bytes went in. See
/// [`the_progress_pair_moves_from_zero_to_n_while_the_sync_runs`] for why one
/// fixture needs a pack that does not fit in a single sideband packet.
fn noise(bytes: usize) -> Vec<u8> {
let mut state: u64 = 0x2545_F491_4F6C_DD1D;
(0..bytes)
.map(|_| {
state = state
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
(state >> 33) as u8
})
.collect()
}
/// [`plant_a_commit`] over a caller-chosen blob.
fn plant_a_commit_of(git: &dyn GitServe, hash: gix_hash::Kind, blob: Vec<u8>) -> (Oid, Oid) {
let blob_id = compute(hash, Kind::Blob, &blob);
let tree = tree_bytes(&[("100644", "fox.txt", blob_id)]);
let tree_id = compute(hash, Kind::Tree, &tree);
let commit = commit_bytes(tree_id, "first");
let commit_id = compute(hash, Kind::Commit, &commit);
let mut pack = PackBuilder::new(hash);
for (kind, content) in [
(Kind::Blob, blob),
(Kind::Tree, tree),
(Kind::Commit, commit),
] {
pack.object(kind, content);
}
let fixture = pack.build().expect("building the pack");
git.put_pack(&fixture.bytes).expect("storing the pack");
git.update_ref("refs/heads/main", None, Some(commit_id.as_bytes()))
.expect("pointing main at the commit");
(commit_id, blob_id)
}
/// The oid `name` resolves to in a store, or `None` when it holds no such
/// reference. The applied output every assertion below is really about.
fn tip(git: &dyn GitServe, name: &str) -> Option<Oid> {
GitOps::refs(git)
.expect("reading refs")
.into_iter()
.find(|row| row.name == name)
.and_then(|row| row.oid)
.map(|raw| {
gix_hash::oid::try_from_bytes(&raw)
.expect("a stored oid decodes")
.to_owned()
})
}
/// **A downstream box**: stores, an empty repository, and a control plane over
/// them. No listener — a mirror is a client, and what it needs is a place to
/// put what it fetched.
struct Downstream {
_dir: tempfile::TempDir,
stores: Arc<ServerStores>,
plane: CoreControlPlane,
path: RepoPath,
repo: gunnar_core::model::Repo,
}
impl Downstream {
/// Every upstream in this file is on loopback, which a mirror refuses to
/// dial unless the box says otherwise — so the fixture says otherwise, the
/// same way an operator would: by creating the file under the data root.
/// `open_strict` is the same box without it, which is the shipped default.
fn open(hash: HashKind) -> Downstream {
let down = Downstream::open_strict(hash);
std::fs::write(
down.stores
.root()
.join(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
b"the upstreams in this test are on 127.0.0.1\n",
)
.expect("allowing private upstreams on this box");
down
}
fn open_strict(hash: HashKind) -> Downstream {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
stores
.catalog()
.create_namespace("mirror")
.expect("namespace");
let repo_ref = RepoRef::new("mirror", "copy").expect("a legal repo path");
let repo = stores
.create_repo(NewRepo::new(repo_ref).hash_kind(hash), false)
.expect("create the local repository");
// Materialise it, so the mirror is fetching into a store that exists —
// the same thing `mirror_add`'s `self.open(path)` guarantees.
stores.repo_store(&repo).expect("open the local store");
let plane = CoreControlPlane::new(Arc::clone(&stores));
Downstream {
_dir: dir,
stores,
plane,
path: RepoPath::new("mirror", "copy"),
repo,
}
}
fn git(&self) -> Arc<gunnar_server::storage::OpenRepo> {
self.stores.repo_store(&self.repo).expect("open the store")
}
/// Poll `Mirror.Status` until the job settles, failing rather than hanging.
///
/// A bounded wait around an asynchronous verb, not a fixed sleep: the
/// assertions after it decide the test and this only turns "it never
/// finished" into a message. It also **records every progress pair it
/// sees**, because the pair is only meaningful while the job runs and a
/// reading taken after the finish would prove nothing.
fn settle(&self) -> (gunnar_proto::MirrorJob, Vec<(bool, u64, u64)>) {
let deadline = Instant::now() + Duration::from_secs(120);
let mut seen: Vec<(bool, u64, u64)> = Vec::new();
while Instant::now() < deadline {
let job = self.job();
let reading = (job.has_progress, job.progress_done, job.progress_total);
if seen.last() != Some(&reading) {
seen.push(reading);
}
if job.state != MirrorJobState::Queued as i32
&& job.state != MirrorJobState::Running as i32
{
return (job, seen);
}
std::thread::sleep(Duration::from_millis(5));
}
panic!("the mirror sync never settled");
}
fn job(&self) -> gunnar_proto::MirrorJob {
self.plane
.mirror_status(None, Some(&self.path))
.expect("mirror_status is implemented")
.into_iter()
.next()
.expect("the registration is there")
.job
.expect("a job that was just enqueued is still known")
}
}
// ════════════════════════════════════════════════════════════════════════════
// The SSH upstream: a whole real gunnar on a loopback russh listener
// ════════════════════════════════════════════════════════════════════════════
/// A running gunnar serving one repository over SSH, with one key registered
/// against it.
#[cfg(feature = "replication")]
struct SshUpstream {
_dir: tempfile::TempDir,
runtime: tokio::runtime::Runtime,
server: Option<gunnar_server::Server>,
key_pem: String,
url: String,
head: Oid,
blob: Oid,
}
#[cfg(feature = "replication")]
impl SshUpstream {
fn start(hash: HashKind) -> SshUpstream {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
stores
.catalog()
.create_namespace("upstream")
.expect("namespace");
let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
let repo = stores
.catalog()
.create(
NewRepo::new(repo_ref.clone())
.hash_kind(hash)
// **Private**, so the fetch has to authenticate. A public
// repository would be served to an anonymous caller and the
// whole SSH credential path would go untested while looking
// green.
.visibility(Visibility::Private),
)
.expect("create the upstream repository");
let store = stores.repo_store(&repo).expect("materialise it");
let (head, blob) = plant_a_commit(store.serve(), store.hash_kind());
// A real ed25519 key, minted in process. No `ssh-keygen` — no
// shelling out.
let key_pem = String::from_utf8(gunnar_core::mint_ed25519().expect("mint a key"))
.expect("an OpenSSH private key is utf-8 PEM");
let public = gunnar_core::ssh_key::PrivateKey::from_openssh(&key_pem)
.expect("the minted key parses")
.public_key()
.to_openssh()
.expect("render the public half");
stores
.auth()
.add_principal(
gunnar_core::Principal::new("mirrorer", "mirrorer", PrincipalKind::User).unwrap(),
)
.expect("add the principal");
stores
.auth()
.register_key("mirrorer", &public, None)
.expect("register the key");
stores
.auth()
.grant(Grant::repo("mirrorer", repo_ref, Level::Read))
.expect("grant read");
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("a tokio runtime");
let config = gunnar_server::ServeConfig {
no_grpc: true,
// The upstream must not serve this repository to a caller with no
// key at all, or the credential is never exercised.
allow_anonymous: false,
..gunnar_server::ServeConfig::ephemeral()
};
let server = runtime
.block_on(gunnar_server::Server::start(Arc::clone(&stores), config))
.expect("the upstream server starts");
let port = server.ssh_addr().port();
SshUpstream {
_dir: dir,
runtime,
server: Some(server),
key_pem,
url: format!("ssh://mirrorer@127.0.0.1:{port}/upstream/origin.git"),
head,
blob,
}
}
}
#[cfg(feature = "replication")]
impl Drop for SshUpstream {
fn drop(&mut self) {
if let Some(server) = self.server.take() {
let _ = self.runtime.block_on(server.shutdown());
}
}
}
// ════════════════════════════════════════════════════════════════════════════
// The smart-HTTP upstream: the two routes, and a 401 when a token is required
// ════════════════════════════════════════════════════════════════════════════
/// A loopback HTTP/1.1 listener serving `git-upload-pack` for one repository.
///
/// **The git half is not written here.** `advertise` and `serve_session` are
/// `gunnar-wire`'s own, called exactly as `gunnar-http` calls them
/// (`Session::StatelessRpc`, so no advertisement is written into a POST
/// response). What this file contributes is the request line, the headers and
/// the `Content-Length` — which is the part the adapter under test has to get
/// right, and the part a fixture can legitimately own.
struct HttpUpstream {
addr: SocketAddr,
/// Set when the listener should stop; the accept loop checks it between
/// connections.
stop: Arc<AtomicBool>,
/// Requests that arrived with no acceptable `Authorization` header and were
/// answered 401. Applied output for "the credential was actually needed".
refused: Arc<AtomicU64>,
/// Every `Authorization` header value the listener saw, so a test can
/// assert the token really did travel — the other half of proving it does
/// not leak.
seen_auth: Arc<Mutex<Vec<String>>>,
/// **The rendezvous that makes the mid-transfer moment observable.**
///
/// When set, the listener writes every byte of a POST response EXCEPT the
/// last few and then waits for the test to release it. By that point the
/// client has read the packfile header — which is what publishes
/// `objects_total` — and is blocked on the rest, so a poller has all the
/// time it needs to see `(0, N)` while the job is unambiguously RUNNING.
///
/// A sleep would have been the other way to write this and would have been
/// a flake in disguise: a fetch of three objects over loopback finishes in
/// about a millisecond, and a poller that happened to miss the window
/// reports the same red as a progress field that was never published.
/// (MEASURED: it did, on the first run of this file.)
///
/// Held on the struct as well as moved into the listener thread so the
/// fixture's own shape says the gate belongs to this upstream; the thread
/// is what actually waits on it.
#[allow(dead_code)]
hold: Option<Arc<AtomicBool>>,
/// **A dishonest upstream**: `(from, to)`, applied to every response body
/// this listener writes.
///
/// The two must be the SAME LENGTH, because a pkt-line carries its own
/// four-hex length and a substitution that changed it would break the
/// framing rather than test anything. That constraint is what picks the
/// name in `an_upstream_may_not_name_a_reference_gunnar_would_not_accept`:
/// `refs/heads/main` and `refs/heads/../x` are both fifteen bytes.
///
/// Rewriting the real `upload_pack` output rather than hand-writing a
/// protocol answer is deliberate: everything except the one field under
/// test stays the shipped implementation, so the fetch really does
/// negotiate, really does receive a pack, and really does reach `apply`.
#[allow(dead_code)]
rename: Option<(String, String)>,
}
impl HttpUpstream {
/// `require` is the exact header value the upstream will accept; `None`
/// serves anybody.
fn start(
stores: Arc<ServerStores>,
repo: gunnar_core::model::Repo,
require: Option<String>,
) -> HttpUpstream {
HttpUpstream::start_held(stores, repo, require, None)
}
/// The same, answering with `to` everywhere the honest answer said `from`.
/// See [`HttpUpstream::rename`].
fn start_renaming(
stores: Arc<ServerStores>,
repo: gunnar_core::model::Repo,
from: &str,
to: &str,
) -> HttpUpstream {
assert_eq!(
from.len(),
to.len(),
"a pkt-line carries its own length; the substitution must not change it"
);
HttpUpstream::start_with(
stores,
repo,
None,
None,
Some((from.to_owned(), to.to_owned())),
)
}
/// The same, with the mid-transfer rendezvous armed. See
/// [`HttpUpstream::hold`].
fn start_held(
stores: Arc<ServerStores>,
repo: gunnar_core::model::Repo,
require: Option<String>,
hold: Option<Arc<AtomicBool>>,
) -> HttpUpstream {
HttpUpstream::start_with(stores, repo, require, hold, None)
}
fn start_with(
stores: Arc<ServerStores>,
repo: gunnar_core::model::Repo,
require: Option<String>,
hold: Option<Arc<AtomicBool>>,
rename: Option<(String, String)>,
) -> HttpUpstream {
let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind a loopback port");
listener
.set_nonblocking(false)
.expect("a blocking listener");
let addr = listener.local_addr().expect("the bound address");
let stop = Arc::new(AtomicBool::new(false));
let refused = Arc::new(AtomicU64::new(0));
let seen_auth = Arc::new(Mutex::new(Vec::new()));
let worker = HttpUpstream {
addr,
stop: Arc::clone(&stop),
refused: Arc::clone(&refused),
seen_auth: Arc::clone(&seen_auth),
hold: hold.clone(),
rename: rename.clone(),
};
std::thread::spawn(move || {
for connection in listener.incoming() {
if stop.load(Ordering::SeqCst) {
return;
}
let Ok(mut connection) = connection else {
return;
};
let store = stores.repo_store(&repo).expect("open the upstream store");
let _ = serve_one_http(
&mut connection,
store.serve(),
store.hash_kind(),
require.as_deref(),
&refused,
&seen_auth,
hold.as_deref(),
rename.as_ref(),
);
}
});
worker
}
fn base(&self) -> String {
format!("http://{}/upstream/origin.git", self.addr)
}
}
impl Drop for HttpUpstream {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
// Unblock the accept.
let _ = std::net::TcpStream::connect(self.addr);
}
}
/// One HTTP request, answered.
///
/// Eight arguments and an `allow` rather than a struct: every one of them is a
/// fixture knob a single test turns on, and folding them into a options type
/// would put a second shape between a test and the listener it is configuring.
#[allow(clippy::too_many_arguments)]
fn serve_one_http(
connection: &mut std::net::TcpStream,
git: &dyn GitServe,
hash: gix_hash::Kind,
require: Option<&str>,
refused: &AtomicU64,
seen_auth: &Mutex<Vec<String>>,
hold: Option<&AtomicBool>,
rename: Option<&(String, String)>,
) -> std::io::Result<()> {
let mut reader = BufReader::new(connection.try_clone()?);
let mut line = String::new();
reader.read_line(&mut line)?;
let mut parts = line.split_whitespace();
let method = parts.next().unwrap_or_default().to_owned();
let target = parts.next().unwrap_or_default().to_owned();
let mut length = 0usize;
let mut authorization: Option<String> = None;
loop {
let mut header = String::new();
if reader.read_line(&mut header)? == 0 {
break;
}
let header = header.trim_end();
if header.is_empty() {
break;
}
// Header NAMES are case-insensitive (RFC 9110 §5.1), and ureq 3 sends them
// lower-case (`content-length`, `authorization`) where ureq 2 sent them
// as written: the fixture reads them as a server must (2026-10-03).
let (name, value) = header.split_once(':').unwrap_or((header, ""));
if name.eq_ignore_ascii_case("content-length") {
length = value.trim().parse().unwrap_or(0);
}
if name.eq_ignore_ascii_case("authorization") {
authorization = Some(value.trim().to_owned());
}
}
if let Some(value) = &authorization {
seen_auth.lock().unwrap().push(value.clone());
}
// **The 401 comes before the git seam**, which is what makes the credential
// load-bearing rather than decorative: without it there is no
// advertisement to read, not an empty one.
if let Some(want) = require {
if authorization.as_deref() != Some(want) {
refused.fetch_add(1, Ordering::SeqCst);
connection.write_all(
b"HTTP/1.1 401 Unauthorized\r\n\
WWW-Authenticate: Basic realm=\"gunnar\"\r\n\
Content-Length: 0\r\n\
Connection: close\r\n\r\n",
)?;
return connection.flush();
}
}
let mut body = vec![0u8; length];
reader.read_exact(&mut body)?;
// **Only a `command=fetch` response carries a packfile**, and the hold is
// about the packfile header. Holding the `ls-refs` answer instead would
// block the client before the fetch had even been asked for — which is a
// deadlock, and is exactly what the first version of this fixture did.
let is_fetch = body
.windows(b"command=fetch".len())
.any(|window| window == b"command=fetch");
let upload = gunnar_wire::UploadPack::new(git, hash);
let mut out = Vec::new();
let (content_type, outcome) = if method == "GET" && target.contains("/info/refs") {
(
"application/x-git-upload-pack-advertisement",
upload.advertise(gunnar_wire::ProtocolVersion::V2, &mut out),
)
} else {
let mut input = std::io::Cursor::new(body);
(
"application/x-git-upload-pack-result",
upload
.serve_session(
gunnar_wire::ProtocolVersion::V2,
gunnar_wire::Session::StatelessRpc,
&mut input,
&mut out,
)
.map(|_| ()),
)
};
if let Err(err) = outcome {
let body = format!("upload-pack failed: {err}");
connection.write_all(
format!(
"HTTP/1.1 500 Internal Server Error\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
)
.as_bytes(),
)?;
return connection.flush();
}
// **The one field a dishonest upstream gets to choose**, substituted in the
// real answer. Equal lengths, so every pkt-line header stays correct.
if let Some((from, to)) = rename {
let (from, to) = (from.as_bytes(), to.as_bytes());
let mut at = 0;
while at + from.len() <= out.len() {
if &out[at..at + from.len()] == from {
out[at..at + from.len()].copy_from_slice(to);
at += from.len();
} else {
at += 1;
}
}
}
connection.write_all(
format!(
"HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\n\
Cache-Control: no-cache\r\nConnection: close\r\n\r\n",
out.len()
)
.as_bytes(),
)?;
// **Held mid-body when a test asked for it.** The tail is small enough that
// everything before it — the acknowledgments, the `packfile` section header
// and the 12-byte pack header the client reads `objects_total` out of — has
// already gone, and large enough that the client cannot finish without it.
match (hold, is_fetch && out.len() > 32) {
(Some(gate), true) => {
let cut = out.len() - 24;
connection.write_all(&out[..cut])?;
connection.flush()?;
while !gate.load(Ordering::SeqCst) {
std::thread::yield_now();
}
connection.write_all(&out[cut..])?;
}
_ => connection.write_all(&out)?,
}
connection.flush()
}
/// An upstream repository with content, on its own stores, for the HTTP tests.
fn http_upstream_stores(
hash: HashKind,
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
Oid,
Oid,
) {
http_upstream_stores_of(hash, b"the quick brown fox\n".to_vec())
}
/// **An upstream whose pack is wide rather than deep**: one commit, one tree,
/// and `blobs` distinct files under it.
///
/// The shared fixture is three objects, which is the right size for every
/// assertion about *what* was copied and the wrong size for every assertion
/// about the copying being watchable: on a three-object pack a store that
/// published once per pack and a store that published once per object are
/// indistinguishable. Returns the tip and the number of objects the fetched
/// pack will carry — `blobs + 2`, the closure of the one reference.
///
/// The names are zero-padded so the tree's entries are in the order git
/// requires, and the contents are distinct so no two blobs collapse into one
/// object.
fn http_upstream_stores_wide(
hash: HashKind,
blobs: usize,
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
Oid,
u64,
) {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
stores
.catalog()
.create_namespace("upstream")
.expect("namespace");
let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
let repo = stores
.catalog()
.create(NewRepo::new(repo_ref).hash_kind(hash))
.expect("create the upstream repository");
let store = stores.repo_store(&repo).expect("materialise it");
let git = store.serve();
let hash_kind = store.hash_kind();
let mut pack = PackBuilder::new(hash_kind);
let mut entries: Vec<(&'static str, String, Oid)> = Vec::with_capacity(blobs);
for n in 0..blobs {
let content = format!("file number {n}\n").into_bytes();
let id = compute(hash_kind, Kind::Blob, &content);
pack.object(Kind::Blob, content);
entries.push(("100644", format!("f{n:05}"), id));
}
let tree = tree_bytes(
&entries
.iter()
.map(|(mode, name, id)| (*mode, name.as_str(), *id))
.collect::<Vec<_>>(),
);
let tree_id = compute(hash_kind, Kind::Tree, &tree);
let commit = commit_bytes(tree_id, "wide");
let commit_id = compute(hash_kind, Kind::Commit, &commit);
pack.object(Kind::Tree, tree);
pack.object(Kind::Commit, commit);
let fixture = pack.build().expect("building the wide pack");
git.put_pack(&fixture.bytes).expect("storing the pack");
git.update_ref("refs/heads/main", None, Some(commit_id.as_bytes()))
.expect("pointing main at the commit");
(dir, stores, repo, commit_id, blobs as u64 + 2)
}
/// The same, over a caller-chosen blob.
fn http_upstream_stores_of(
hash: HashKind,
blob: Vec<u8>,
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
Oid,
Oid,
) {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
stores
.catalog()
.create_namespace("upstream")
.expect("namespace");
let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
let repo = stores
.catalog()
.create(NewRepo::new(repo_ref).hash_kind(hash))
.expect("create the upstream repository");
let store = stores.repo_store(&repo).expect("materialise it");
let (head, blob_id) = plant_a_commit_of(store.serve(), store.hash_kind(), blob);
(dir, stores, repo, head, blob_id)
}
// ════════════════════════════════════════════════════════════════════════════
// The assertions every transport shares
// ════════════════════════════════════════════════════════════════════════════
/// **The whole property**, asserted once so the SSH and HTTP tests cannot drift
/// apart about what "mirrored" means.
///
/// It is deliberately not "the job says DONE": a control plane that marked a
/// job DONE without fetching anything would pass that, and every one of these
/// is something it could not fake.
fn assert_mirrored(
down: &Downstream,
job: &gunnar_proto::MirrorJob,
progress: &[(bool, u64, u64)],
head: Oid,
blob: Oid,
) {
assert_eq!(
job.state,
MirrorJobState::Done as i32,
"the sync failed: {}",
job.error
);
// ── the reference moved, to the upstream's own tip ───────────────────────
let store = down.git();
assert_eq!(
tip(store.serve(), "refs/heads/main"),
Some(head),
"the mirror's branch is not at the upstream's commit"
);
// ── and the OBJECTS came with it ─────────────────────────────────────────
//
// A reference pointing at an object the store does not hold is exactly the
// failure a mirror that applied refs without a pack would produce, and it
// reads as success at every other seam.
for (what, oid) in [("the commit", head), ("the blob", blob)] {
assert!(
GitOps::extents(store.serve(), &[oid.as_bytes()])
.expect("asking the store for an object")
.first()
.is_some_and(Option::is_some),
"{what} the reference needs is not in the mirror's store"
);
}
// ── three objects, and the job says so ───────────────────────────────────
assert_eq!(
job.objects, 3,
"the fixture is one commit, one tree and one blob: {job:?}"
);
assert!(job.bytes > 0, "a pack of three objects has a size: {job:?}");
assert_eq!(
job.updates.len(),
1,
"one reference moved, and the job must name it: {:?}",
job.updates
);
assert_eq!(job.updates[0].ref_name, "refs/heads/main");
assert_eq!(job.updates[0].new_oid, head.to_string());
assert_eq!(
job.updates[0].old_oid,
"0".repeat(head.to_string().len()),
"a branch that did not exist here before is a CREATE: all-zeroes old"
);
// ── the progress pair ends at (N, N) ────────────────────────────────────
assert_eq!(
(job.has_progress, job.progress_done, job.progress_total),
(true, 3, 3),
"a finished sync reports every object it took: {job:?}"
);
// …and **every reading a poller ever saw carried the right denominator**.
// A transfer of three objects over loopback is about a millisecond, so
// whether the poller catches the `(0, N)` instant is luck — that claim is
// made deterministically by
// `the_progress_pair_moves_from_zero_to_n_while_the_sync_runs`, which holds
// the upstream mid-body. What every run CAN assert is that the denominator
// is never invented: a reading of `(true, d, t)` with `t != 3` would be a
// total that came from somewhere other than the packfile header.
for reading in progress {
assert!(
!reading.0 || (reading.2 == 3 && reading.1 <= 3),
"a progress reading with a denominator the packfile never declared: \
{reading:?} in {progress:?}"
);
}
}
/// Register the upstream on the downstream and set its credential.
fn register(down: &Downstream, url: &str, kind: MirrorCredentialKind, secret: &str, user: &str) {
down.plane
.mirror_add(&down.path, url, false, false)
.expect("register the mirror");
if kind != MirrorCredentialKind::Unspecified {
let config = down
.plane
.mirror_set_credential(&down.path, kind, secret, user)
.expect("store the credential");
assert_eq!(
config.credential, kind as i32,
"the registration must say WHICH kind is stored"
);
}
}
// ════════════════════════════════════════════════════════════════════════════
// SSH
// ════════════════════════════════════════════════════════════════════════════
#[cfg(feature = "replication")]
fn mirror_over_ssh(hash: HashKind) {
let upstream = SshUpstream::start(hash);
let down = Downstream::open(hash);
register(
&down,
&upstream.url,
MirrorCredentialKind::SshKey,
&upstream.key_pem,
"",
);
let accepted = down
.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let job = accepted.job.expect("Mirror.Sync answers with a job");
assert!(
job.state == MirrorJobState::Queued as i32 || job.state == MirrorJobState::Running as i32,
"Mirror.Sync must return with the job still live, not after fetching: {job:?}"
);
assert!(job.job_id.starts_with("mirror-"), "{job:?}");
let (done, progress) = down.settle();
assert_eq!(
done.job_id, job.job_id,
"the finished job must be the one that was accepted"
);
assert_mirrored(&down, &done, &progress, upstream.head, upstream.blob);
}
/// **A box that says so refuses an upstream host key nobody put there.**
///
/// The mirror's SSH dial was hardcoded trust-on-first-use — the policy was a
/// literal in `dial_ssh` and there was no way to say otherwise — so the first
/// sync trusted whatever answered on the upstream's address, on the one path
/// that is about to offer that address a private key. Replication's own
/// targets have carried a `strict_host_key` knob all along, and its comment
/// says on "is the right setting for anything that outlives an afternoon".
///
/// Every other assertion in this file's SSH half is the TOFU case, which still
/// works: `a_mirror_sync_over_ssh_copies_a_sha256_repository` runs against the
/// same upstream with no `mirror_strict_host_key` file and mirrors it. So this
/// is a rule the box chooses, not a mirror that stopped working.
///
/// MEASURED before the knob existed: the sync reached DONE against an upstream
/// whose host key had never been seen, and `mirror_known_hosts` was created by
/// the fetch itself.
#[cfg(feature = "replication")]
#[test]
fn a_box_with_strict_host_keys_refuses_an_upstream_it_has_never_seen() {
let upstream = SshUpstream::start(HashKind::Sha256);
let down = Downstream::open(HashKind::Sha256);
let known_hosts = down
.stores
.root()
.join(gunnar_server::mirror_fetch::MIRROR_KNOWN_HOSTS);
assert!(
!known_hosts.exists(),
"the fixture must start with nothing pinned, or strictness proves nothing"
);
std::fs::write(
down.stores
.root()
.join(gunnar_server::mirror_fetch::MIRROR_STRICT_HOST_KEY),
b"this box does not learn host keys from the network\n",
)
.expect("asking this box to refuse an unknown host key");
register(
&down,
&upstream.url,
MirrorCredentialKind::SshKey,
&upstream.key_pem,
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"a box that refuses unknown host keys mirrored from one anyway: {job:?}"
);
assert!(
job.error.contains("host key") || job.error.contains("known_hosts"),
"the refusal must name what it refused: {}",
job.error
);
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
None,
"nothing was fetched"
);
assert!(
!known_hosts.exists(),
"a strict dial must LEARN NOTHING; `{}` was written by the fetch",
known_hosts.display()
);
}
/// **sha256**, the kind gunnar creates by default.
#[cfg(feature = "replication")]
#[test]
fn a_mirror_sync_over_ssh_copies_a_sha256_repository() {
mirror_over_ssh(HashKind::Sha256);
}
/// **sha1**, and it is not a formality: `fetch_pack` refuses an `object-format`
/// mismatch by name, so a client that quietly ran at one width against a remote
/// of the other would negotiate an EMPTY answer and report success.
#[cfg(feature = "replication")]
#[test]
fn a_mirror_sync_over_ssh_copies_a_sha1_repository() {
mirror_over_ssh(HashKind::Sha1);
}
// ════════════════════════════════════════════════════════════════════════════
// Smart HTTP
// ════════════════════════════════════════════════════════════════════════════
fn mirror_over_http(hash: HashKind) {
let (_dir, stores, repo, head, blob) = http_upstream_stores(hash);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(hash);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (done, progress) = down.settle();
assert_mirrored(&down, &done, &progress, head, blob);
}
/// **The adapter, against the same `upload_pack` gunnar-http drives.**
#[test]
fn a_mirror_sync_over_smart_http_copies_a_sha256_repository() {
mirror_over_http(HashKind::Sha256);
}
#[test]
fn a_mirror_sync_over_smart_http_copies_a_sha1_repository() {
mirror_over_http(HashKind::Sha1);
}
/// **A private upstream: 401 without the token, served with it, and the token
/// appears in nothing anybody can read.**
#[test]
fn a_private_http_upstream_needs_the_token_and_never_shows_it() {
const TOKEN: &str = "ghp_MirrorSyncNeverLeaksThis";
let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(
Arc::clone(&stores),
repo.clone(),
Some(format!("Bearer {TOKEN}")),
);
let down = Downstream::open(HashKind::Sha256);
// ── RED HALF: with no credential the upstream refuses, and the job FAILS ──
//
// Without this, the green half below would pass against a mirror that
// ignored the credential entirely and against an upstream that never
// checked it.
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane.mirror_sync(&down.path).expect("accepted");
let (refused, _) = down.settle();
assert_eq!(
refused.state,
MirrorJobState::Failed as i32,
"a private upstream must refuse an anonymous mirror, not serve it an empty answer"
);
assert!(
refused.error.contains("401"),
"the failure must name what the upstream said: {}",
refused.error
);
assert!(
upstream.refused.load(Ordering::SeqCst) >= 1,
"the upstream never actually refused anything, so this proves nothing"
);
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
None,
"a refused sync must move no reference"
);
// ── GREEN HALF: the same mirror, with the token ──────────────────────────
down.plane
.mirror_set_credential(&down.path, MirrorCredentialKind::Token, TOKEN, "")
.expect("store the token");
down.plane.mirror_sync(&down.path).expect("accepted");
let (done, progress) = down.settle();
assert_mirrored(&down, &done, &progress, head, blob);
// The token really did travel — otherwise the "never shows it" assertions
// below would be about a token nothing ever sent.
let seen = upstream.seen_auth.lock().unwrap().clone();
assert!(
seen.iter().any(|value| value == &format!("Bearer {TOKEN}")),
"the upstream never received the token: {seen:?}"
);
// ── AND IT IS IN NONE OF THE BYTES ANYBODY CAN READ ─────────────────────
//
// Every surface a caller or an operator reaches: the status response, the
// finished job, the failed job from the red half, and the registration.
let mut surfaces: Vec<String> = vec![
format!(
"{:?}",
down.plane.mirror_status(None, Some(&down.path)).unwrap()
),
format!("{done:?}"),
format!("{refused:?}"),
format!("{:?}", down.plane.mirror_status(None, None).unwrap()),
];
// …and the one RPC that CARRIES it, read back: `SetCredential`'s own
// response. If any RPC were going to hand a secret back it would be this
// one.
surfaces.push(format!(
"{:?}",
down.plane
.mirror_set_credential(&down.path, MirrorCredentialKind::Token, TOKEN, "")
.unwrap()
));
for surface in &surfaces {
assert!(
!surface.contains(TOKEN),
"the token appears in something a caller can read:\n{surface}"
);
}
// …and forgetting it is a real forget, not a flag flip: the next sync is
// refused again.
down.plane
.mirror_set_credential(&down.path, MirrorCredentialKind::Unspecified, "", "")
.expect("forget the token");
down.plane.mirror_sync(&down.path).expect("accepted");
let (after, _) = down.settle();
assert_eq!(
after.state,
MirrorJobState::Failed as i32,
"a forgotten credential must actually be gone: {after:?}"
);
}
/// **`(0, N)` is readable WHILE the fetch is still running, and only then does
/// it become `(N, N)`.**
///
/// The claim the queue exists to make: a fetch of a large repository is minutes
/// of network with a console watching, and a job that sat at RUNNING and then
/// jumped to DONE would tell that console nothing. So the denominator has to be
/// published at the *start* of the transfer — which is what
/// `FetchProgress::objects_total` is for, fired the instant the packfile header
/// arrives — and not computed at the end when nobody needs it.
///
/// Asserted against an upstream **held mid-body**, not against a poll that
/// happened to be quick: a three-object fetch over loopback finishes in about a
/// millisecond, and a timing-dependent version of this test is a flake that
/// reads exactly like the bug.
#[test]
fn the_progress_pair_moves_from_zero_to_n_while_the_sync_runs() {
// **A pack that does not fit in one sideband packet.** The hold below cuts
// the response short, and a pkt-line is only readable once it is COMPLETE:
// with a three-hundred-byte pack the whole transfer is one packet, so
// withholding its tail blocks the client BEFORE the packfile header instead
// of after it, and nothing is ever published. (MEASURED: that is exactly
// what the first version of this test did — a 60-second timeout with
// `has_progress: false`.) 200 KiB of incompressible noise is three sideband
// packets, so the header is in the part that has already gone.
let (_dir, stores, repo, head, blob) =
http_upstream_stores_of(HashKind::Sha256, noise(200 * 1024));
let release = Arc::new(AtomicBool::new(false));
let upstream = HttpUpstream::start_held(stores, repo, None, Some(Arc::clone(&release)));
let down = Downstream::open(HashKind::Sha256);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane.mirror_sync(&down.path).expect("accepted");
// The upstream is holding the tail of the pack, so this cannot pass by
// being fast.
let deadline = Instant::now() + Duration::from_secs(60);
loop {
assert!(
Instant::now() < deadline,
"the sync never published a denominator: {:?}",
down.job()
);
let job = down.job();
if job.has_progress {
assert_eq!(
(job.progress_done, job.progress_total),
(0, 3),
"the FIRST reading must be (0, N): the total is known from the packfile \
header, the numerator is not known until the ingest returns"
);
assert_eq!(
job.state,
MirrorJobState::Running as i32,
"progress must be readable WHILE the job runs, which is the only time \
anybody needs it: {job:?}"
);
break;
}
assert_ne!(
job.state,
MirrorJobState::Done as i32,
"the job finished without ever publishing a denominator: {job:?}"
);
std::thread::yield_now();
}
release.store(true, Ordering::SeqCst);
let (done, progress) = down.settle();
assert_mirrored(&down, &done, &progress, head, blob);
}
/// **The numerator MOVES**: more than two distinct `done` values, monotone,
/// ending at `N`.
///
/// # 🔴 The red this was written against
///
/// Until 2026-09-08 this test recorded exactly two readings for a 302-object
/// pack:
///
/// ```text
/// the count has 2 distinct value(s) over a pack of 302 objects: [0, 302]
/// ```
///
/// `(0, N)` from `FetchProgress::objects_total` when the packfile header
/// arrived, `(N, N)` when the ingest returned, and nothing in between — which
/// is what `mirror_queue`'s module doc said in as many words and what a console
/// rendered as a bar that jumped from empty to full. Everything else about the
/// sync was already green: the reference moved, the objects landed, the job
/// reported the right total. Storing a pack correctly and *saying how far you
/// have got while you store it* are different claims, and only one of them was
/// being made.
///
/// It records 304 readings now — `0` from the header, `1..=302` one per object
/// as the engine's pack walk splits them, and `302` again from `finish` —
/// 303 distinct values where there were two.
///
/// # Why the series is recorded at a sink and not polled off the job
///
/// [`Reporter::progress`] overwrites — deliberately; its own doc says a
/// progress that accumulated would be a second counter to keep right — so the
/// job record holds the latest reading and never the sequence. Polling it would
/// make this test a race between a sampler thread and a pack walk that takes
/// microseconds, and a flaky version of this test reads exactly like the bug it
/// is for. `sync_once_watching` hands the sink the same values the `Reporter`
/// is given, in the same order, on the same thread, so what is recorded here is
/// what the job saw.
///
/// # 302 objects, and the real path under them
///
/// The whole production path runs: the real `MirrorQueue` worker, the real
/// `mirror_fetch`, the real `PushIngest`, the real storage engine, a real
/// upload-pack on the other end of a loopback socket. The only thing this test
/// supplies is a second listener.
#[test]
fn the_object_count_moves_while_the_pack_lands() {
/// Every publish, in order.
struct Series(Mutex<Vec<(u64, u64)>>);
impl gunnar_store::git_storage_trait::PackProgress for Series {
fn indexed(&self, done: u64, total: u64) {
self.0
.lock()
.unwrap_or_else(|p| p.into_inner())
.push((done, total));
}
}
let (_dir, up_stores, up_repo, head, objects) =
http_upstream_stores_wide(HashKind::Sha256, 300);
let upstream = HttpUpstream::start(up_stores, up_repo, None);
let down = Downstream::open(HashKind::Sha256);
let series = Arc::new(Series(Mutex::new(Vec::new())));
let sink = Arc::clone(&series);
let stores = Arc::clone(&down.stores);
// The real queue, so the `Reporter` the sink shadows is a real one, taken
// from a real job by the real worker.
let queue = gunnar_server::mirror_queue::MirrorQueue::new(Arc::new(move |order, report| {
gunnar_server::mirror_fetch::sync_once_watching(
&stores,
order,
&gunnar_server::mirror_fetch::Credential::None,
report,
Some(&*sink),
)
}));
let order = gunnar_server::mirror_queue::MirrorOrder {
path: "mirror/copy".to_owned(),
upstream: upstream.base(),
prune: false,
ca: Vec::new(),
};
queue.enqueue(order);
let deadline = Instant::now() + Duration::from_secs(120);
let job = loop {
assert!(Instant::now() < deadline, "the mirror sync never settled");
let job = queue.status("mirror/copy").expect("the job is still known");
if job.state != gunnar_server::job_queue::JobState::Queued
&& job.state != gunnar_server::job_queue::JobState::Running
{
break job;
}
std::thread::sleep(Duration::from_millis(5));
};
queue.stop();
assert_eq!(
job.state,
gunnar_server::job_queue::JobState::Done,
"the sync failed: {:?}",
job.error
);
// The mirror is real before the series is judged: a sync that published a
// beautiful count and copied nothing must fail here, not there.
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
Some(head),
"the mirror's branch is not at the upstream's commit"
);
assert_eq!(job.objects, objects, "the pack's own object count");
let seen = series.0.lock().unwrap_or_else(|p| p.into_inner()).clone();
let dones: Vec<u64> = seen.iter().map(|(done, _)| *done).collect();
assert!(!seen.is_empty(), "the ingest published nothing at all");
for (done, total) in &seen {
assert_eq!(
*total, objects,
"the denominator moved mid-transfer: a reading declared {total} for a pack \
of {objects}"
);
assert!(
*done <= objects,
"a reading claims {done} of {objects} objects are in"
);
}
assert!(
seen.windows(2).all(|w| w[0].0 <= w[1].0),
"the count went backwards: {dones:?}"
);
assert_eq!(
seen.last().map(|(done, _)| *done),
Some(objects),
"the last reading is not ({objects}, {objects}); a bar that stops short of the \
end reads exactly like a transfer that hung"
);
let distinct = {
let mut values = dones.clone();
values.dedup();
values.len()
};
assert!(
distinct > 2,
"the count has {distinct} distinct value(s) over a pack of {objects} objects: \
{:?}. Two is `(0, N)` and `(N, N)` — the shape this test exists to make \
unshippable.",
if dones.len() > 8 {
&dones[..8]
} else {
&dones[..]
}
);
// …and the job carries the last of them, so what a console polls is the
// same number the sink recorded and not a second bookkeeping.
assert_eq!(
job.progress,
Some((objects, objects)),
"the job's own progress pair must end where the series does"
);
}
// ════════════════════════════════════════════════════════════════════════════
// What an upstream may call a reference
// ════════════════════════════════════════════════════════════════════════════
/// **A reference name from an upstream is untrusted input, and goes through the
/// same rules a pushed name does.**
///
/// The upstream here is the real `upload_pack`, serving a real repository, with
/// `refs/heads/main` rewritten to `refs/heads/../x` in every response body —
/// fifteen bytes for fifteen, so the pkt-line framing is untouched and the
/// negotiation, the pack and the ingest are all the shipped ones. Only the name
/// is a lie, which is exactly the shape a hostile or broken upstream has.
///
/// `fetch_pack::mirror` is the only filter such a name met before this: a
/// prefix test, which `refs/heads/../x` passes.
///
/// MEASURED before the check existed: the fetch succeeded, the job reached
/// DONE, and `refs/heads/../x` was in this repository's ref listing.
#[test]
fn an_upstream_may_not_name_a_reference_gunnar_would_not_accept() {
let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start_renaming(stores, repo, "refs/heads/main", "refs/heads/../x");
let down = Downstream::open(HashKind::Sha256);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"a name with `..` in it was accepted from an upstream: {job:?}"
);
assert!(
job.error.contains("will not write") && job.error.contains(".."),
"the refusal must say which rule the name broke: {}",
job.error
);
let held = GitOps::refs(down.git().serve()).expect("reading refs");
assert!(
held.iter().all(|row| !row.name.contains("..")),
"a reference the upstream named is in this store: {:?}",
held.iter().map(|row| &row.name).collect::<Vec<_>>()
);
}
// ════════════════════════════════════════════════════════════════════════════
// Where a mirror may be pointed
// ════════════════════════════════════════════════════════════════════════════
/// **`Mirror.Add` refuses a destination only the server can reach**, on the
/// call that made it rather than four minutes later on a job.
///
/// The URL here is the loopback one the rest of this file mirrors from happily
/// — the difference is the box, not the address: this `Downstream` is the
/// shipped default, with no `mirror_allow_private_destinations` file. That is
/// the whole finding: `Mirror.Add` needs Admin, every principal is
/// intrinsically Admin in its own namespace, and the only check on the URL was
/// that it was not empty.
///
/// MEASURED before the guard existed: `mirror_add` returned `Ok` and the
/// registration was stored, for `http://169.254.169.254/latest/meta-data/` as
/// readily as for anything else.
#[test]
fn a_mirror_may_not_be_pointed_at_an_address_only_the_server_can_reach() {
let down = Downstream::open_strict(HashKind::Sha256);
for url in [
"http://127.0.0.1:9418/ns/repo.git",
"http://169.254.169.254/latest/meta-data/",
"https://10.0.0.5/ns/repo.git",
"ssh://git@192.168.1.1/ns/repo.git",
"ssh://git@[::1]:2222/ns/repo.git",
"git@10.1.2.3:ns/repo.git",
] {
let err = down
.plane
.mirror_add(&down.path, url, false, false)
.expect_err("a destination the caller cannot reach on their own");
match err {
gunnar_grpc::seam::ControlError::InvalidArgument(why) => assert!(
why.contains("network position")
&& why.contains(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
"the refusal must name the address and the way to allow it: {why}"
),
other => panic!("{url} must be INVALID_ARGUMENT, not {other:?}"),
}
assert!(
down.plane
.mirror_status(None, Some(&down.path))
.expect("status")
.is_empty(),
"{url} was refused and still written down"
);
}
// …and a public destination is still accepted, so the guard is a rule and
// not a wall. Nothing is dialled by `Mirror.Add`, so this makes no request.
down.plane
.mirror_add(
&down.path,
"https://93.184.216.34/ns/repo.git",
false,
false,
)
.expect("a routable destination registers");
}
// ════════════════════════════════════════════════════════════════════════════
// Branch protection, over the batch a mirror applies
// ════════════════════════════════════════════════════════════════════════════
/// Put a protection rule on `refs/heads/main`, through the verb an operator
/// uses. `Policy.SetProtection` needs **Admin**; `Mirror.Sync` needs **Write**,
/// which is the whole asymmetry these two tests are about.
fn protect_main(down: &Downstream, allowed: Vec<String>) {
down.plane
.set_protection(
&down.path,
gunnar_proto::BranchProtection {
ref_pattern: "refs/heads/main".to_owned(),
allow_force_push: false,
allow_delete: false,
allowed_principal_ids: allowed,
..Default::default()
},
)
.expect("the protection rule is stored");
}
/// **A mirror sync may not force-push over a protected branch.**
///
/// The downstream's `refs/heads/main` is at a commit the upstream has never
/// heard of, so the upstream's tip is a genuine non-fast-forward onto it — the
/// update a push would get `non-fast-forward` for, from the rule that is
/// configured here. Every assertion is on applied output: the reference the
/// store holds afterwards, which is what a bypass would have moved.
///
/// MEASURED before the gate existed: the job reached DONE and
/// `refs/heads/main` held the upstream's commit, with `Policy.GetProtection`
/// still reporting `allow_force_push: false` on the same reference.
#[test]
fn a_mirror_sync_does_not_force_push_over_a_protected_branch() {
let (_dir, stores, repo, upstream_head, _blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
// A local history the upstream does not share: same shape, different blob,
// therefore a different commit with no ancestry in common.
let store = down.git();
let (local_head, _) = plant_a_commit_of(
store.serve(),
store.hash_kind(),
b"a history of our own\n".to_vec(),
);
assert_ne!(local_head, upstream_head, "the fixture must diverge");
protect_main(&down, Vec::new());
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"a mirror that force-pushes over branch protection is the third independent way past \
the gate; the job must FAIL: {job:?}"
);
assert!(
job.error.contains("non-fast-forward") && job.error.contains("refs/heads/main"),
"the refusal must name what it refused and why: {}",
job.error
);
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
Some(local_head),
"nothing was moved: the protected reference still holds the commit it held"
);
}
/// **A rule that names its principals refuses the sync**, because the caller's
/// identity does not cross the control-plane seam on this verb and a gate that
/// cannot decide must not allow.
///
/// The upstream's tip here would be an ordinary CREATE — nothing about it is a
/// force-push — so the only thing that can refuse it is the undecidable
/// `allowed_principals`, and the message has to say so rather than blame
/// ancestry.
#[test]
fn a_rule_that_names_its_principals_refuses_a_mirror_sync() {
let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
protect_main(&down, vec!["alice".to_owned()]);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"a rule this path cannot evaluate must refuse, not skip: {job:?}"
);
assert!(
job.error.contains("carries no principal"),
"the refusal must say WHY it could not decide, not invent an ancestry reason: {}",
job.error
);
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
None,
"nothing was moved"
);
}
/// **The gate is not a blanket refusal.** The same protected repository, with
/// the same rule, mirrors normally when the upstream's tip is a fast-forward —
/// which is what keeps the two tests above from being satisfied by a mirror
/// that simply stopped working.
#[test]
fn a_protected_repository_still_mirrors_a_fast_forward() {
let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
protect_main(&down, Vec::new());
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (done, progress) = down.settle();
assert_mirrored(&down, &done, &progress, head, blob);
}
// ════════════════════════════════════════════════════════════════════════════
// The queue's own contract, at the door
// ════════════════════════════════════════════════════════════════════════════
/// **Two clicks are one job**, asserted on the id and on the *applied output* —
/// how many objects the upstream was actually asked for. A queue that
/// deduplicated its bookkeeping while running the fetch twice would pass an
/// id comparison.
#[test]
fn two_syncs_of_one_repository_are_one_job() {
let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
let first = down.plane.mirror_sync(&down.path).unwrap().job.unwrap();
let second = down.plane.mirror_sync(&down.path).unwrap().job.unwrap();
assert_eq!(
first.job_id, second.job_id,
"an impatient second click must join the job that is already there, not start another"
);
let (done, progress) = down.settle();
assert_eq!(done.job_id, first.job_id);
assert_mirrored(&down, &done, &progress, head, blob);
}
/// A sync of a repository nobody registered is `NOT_FOUND` **by name**, and it
/// is refused on the request that made it rather than becoming a job that fails
/// later. Those are two different things to tell an operator and only one of
/// them is their mistake.
#[test]
fn a_sync_of_an_unregistered_repository_is_not_found() {
let down = Downstream::open(HashKind::Sha256);
let err = down
.plane
.mirror_sync(&down.path)
.expect_err("nothing was registered");
match err {
gunnar_grpc::seam::ControlError::NotFound(why) => assert!(
why.contains("no mirror registered for mirror/copy"),
"the refusal must name the repository: {why}"
),
other => panic!("an unregistered mirror must be NOT_FOUND, not {other:?}"),
}
}
/// **`Mirror.Remove` takes the sealed credential with it.**
///
/// The registration was removed and the secret was not: the ssh private key or
/// forge token stayed in the sealed store for the life of the box, with nothing
/// on the control plane naming it or able to name it. No RPC lists secrets and
/// no RPC reads one back, so it was un-listable, un-rotatable, and still
/// decryptable by anything that reaches the KEK — while the operator who
/// removed the mirror because the credential leaked had been told it was gone.
///
/// Asserted on the sealed store itself, which is the only place the fact lives.
///
/// MEASURED before the fix: `SecretStore::get("mirror/mirror/copy")` returned
/// the token's bytes after `Mirror.Remove` returned `Ok`.
#[test]
fn removing_a_mirror_deletes_its_sealed_credential() {
const TOKEN: &str = "ghp_ThisMustNotOutliveTheRegistration";
let down = Downstream::open(HashKind::Sha256);
register(
&down,
"https://93.184.216.34/ns/repo.git",
MirrorCredentialKind::Token,
TOKEN,
"",
);
let repo_ref = RepoRef::new("mirror", "copy").expect("a legal repo path");
let key = gunnar_core::credential_secret_key(&repo_ref);
let secrets = down.stores.secrets().expect("the sealed store");
assert_eq!(
gunnar_store::SecretStore::get(&secrets, &key).expect("reading the sealed store"),
Some(TOKEN.as_bytes().to_vec()),
"the fixture must actually have sealed something, or the assertion below is vacuous"
);
down.plane
.mirror_remove(&down.path)
.expect("remove the mirror");
assert_eq!(
gunnar_store::SecretStore::get(&secrets, &key).expect("reading the sealed store"),
None,
"the credential outlived the registration that was the only thing naming it"
);
assert!(
down.plane
.mirror_status(None, Some(&down.path))
.expect("status")
.is_empty(),
"and the registration is gone"
);
}
/// **The wrong kind of credential is refused, by name, before anything is
/// dialled.**
///
/// The failure this prevents is the silent one: an SSH key against an
/// `https://` remote cannot ride in an HTTP request, so the fetch would go out
/// anonymous and come back 401 — against an operator who can see a credential
/// is stored and will therefore rotate a key that was never the problem.
#[test]
fn a_credential_of_the_wrong_kind_is_refused_by_name() {
let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
let key = String::from_utf8(gunnar_core::mint_ed25519().expect("mint a key"))
.expect("an OpenSSH private key is utf-8 PEM");
register(
&down,
&upstream.base(),
MirrorCredentialKind::SshKey,
&key,
"",
);
down.plane.mirror_sync(&down.path).expect("accepted");
let (job, _) = down.settle();
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(
job.error.contains("smart http") && job.error.contains("ssh private key"),
"the refusal must name BOTH the transport and the credential it cannot carry: {}",
job.error
);
assert!(
!job.error.contains("PRIVATE KEY"),
"the refusal quoted the key: {}",
job.error
);
}
/// **A credential value with no kind is refused**, because the alternative is
/// that "set this token" and "forget my key" are the same request read two
/// ways — and one of those readings silently destroys a stored credential.
#[test]
fn a_credential_write_that_does_not_say_which_way_it_means_is_refused() {
let down = Downstream::open(HashKind::Sha256);
down.plane
.mirror_add(&down.path, "https://example.test/ns/repo.git", false, false)
.expect("register");
let err = down
.plane
.mirror_set_credential(&down.path, MirrorCredentialKind::Unspecified, "a-token", "")
.expect_err("a value with no kind");
assert!(
matches!(err, gunnar_grpc::seam::ControlError::InvalidArgument(_)),
"{err:?}"
);
let err = down
.plane
.mirror_set_credential(&down.path, MirrorCredentialKind::Token, "", "")
.expect_err("a kind with no value");
assert!(
matches!(err, gunnar_grpc::seam::ControlError::InvalidArgument(_)),
"{err:?}"
);
}
// ════════════════════════════════════════════════════════════════════════════
// A mirror sync counts against the namespace's room (2026-10-02)
// ════════════════════════════════════════════════════════════════════════════
/// The refusal a capped gate composes — the plan gate's own sentence, the one
/// a push past the same ceiling is told.
const CAP_WORDS: &str = "plan team would pass its pack quota with this push (cap 1 bytes): the \
transfer was stopped and nothing was stored";
/// A gate that holds the downstream's namespace to a fixed ceiling, or says it
/// is already full.
struct Capped {
ceiling: u64,
full: bool,
}
impl gunnar_ssh::PlanGate for Capped {
fn name(&self) -> &'static str {
"capped-for-the-test"
}
fn verdict(&self, _: &str, _: gunnar_ssh::PlanAction) -> gunnar_ssh::PlanVerdict {
gunnar_ssh::PlanVerdict::Allow
}
fn pack_quota(&self, namespace: &str) -> gunnar_ssh::PlanVerdict {
if self.full && namespace == "mirror" {
gunnar_ssh::PlanVerdict::Refuse(CAP_WORDS.to_owned())
} else {
gunnar_ssh::PlanVerdict::Allow
}
}
fn pack_ceiling(&self, namespace: &str) -> Option<gunnar_ssh::PackCeiling> {
(namespace == "mirror").then(|| gunnar_ssh::PackCeiling {
bytes: self.ceiling,
refusal: CAP_WORDS.to_owned(),
})
}
}
fn capped_sync(ceiling: u64, full: bool) -> (Downstream, gunnar_proto::MirrorJob, Oid) {
let (_dir, stores, repo, head, _blob) =
http_upstream_stores_of(HashKind::Sha256, noise(512 * 1024));
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
// The same chain the front doors hold a push to — installed where a
// mirror's worker reads it.
down.stores.install_plan_gates(vec![
Arc::new(Capped { ceiling, full }) as Arc<dyn gunnar_ssh::PlanGate>
]);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
(down, job, head)
}
/// ★ **A sync bigger than the namespace's room stops cleanly at the cap.** The
/// fetch is cut mid-pack at the ceiling, the job fails with the gate's own
/// sentence, and the mirror holds nothing of it — no reference, no object.
/// Before this, a mirror fetch was the one write on the box no ceiling held:
/// an import landed its whole upstream however far past the cap it went.
#[test]
fn a_mirror_sync_past_the_namespaces_room_stops_at_the_cap_and_lands_nothing() {
let ceiling = 64 * 1024;
let (down, job, head) = capped_sync(ceiling, false);
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(
job.error.contains(CAP_WORDS),
"the job says {:?}",
job.error
);
assert_eq!(
job.refused_reason,
gunnar_proto::MirrorRefusal::PackCap as i32,
"the cap stop is a FIELD, not only gunnar's prose: {job:?}"
);
assert!(
job.bytes <= ceiling + 64 * 1024,
"the fetch took {} bytes off the wire past a {ceiling}-byte ceiling",
job.bytes
);
let store = down.git();
assert_eq!(
tip(store.serve(), "refs/heads/main"),
None,
"a cut sync moved a ref"
);
assert!(
GitOps::extents(store.serve(), &[head.as_bytes()])
.expect("asking the store")
.first()
.is_none_or(Option::is_none),
"a cut sync stored the commit"
);
}
/// A namespace already at its cap fetches nothing at all — refused before the
/// dial, in the gate's words.
#[test]
fn a_mirror_sync_into_a_full_namespace_fetches_nothing() {
let (down, job, _) = capped_sync(u64::MAX, true);
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(job.error.contains(CAP_WORDS), "{:?}", job.error);
assert_eq!(
job.refused_reason,
gunnar_proto::MirrorRefusal::PackCap as i32,
"{job:?}"
);
assert_eq!(job.bytes, 0, "a full namespace took bytes off the wire");
assert_eq!(tip(down.git().serve(), "refs/heads/main"), None);
}
/// A ceiling the upstream fits under changes nothing.
#[test]
fn a_mirror_sync_under_the_ceiling_lands_whole() {
let (down, job, head) = capped_sync(64 * 1024 * 1024, false);
assert_eq!(job.state, MirrorJobState::Done as i32, "{}", job.error);
assert_eq!(
job.refused_reason,
gunnar_proto::MirrorRefusal::Unspecified as i32
);
assert_eq!(tip(down.git().serve(), "refs/heads/main"), Some(head));
}
/// A gate whose plan refuses every write — a suspended namespace.
struct Suspended;
const SUSPENDED_WORDS: &str = "plan team is suspended: writes are refused until it is paid";
impl gunnar_ssh::PlanGate for Suspended {
fn name(&self) -> &'static str {
"suspended-for-the-test"
}
fn verdict(&self, namespace: &str, _: gunnar_ssh::PlanAction) -> gunnar_ssh::PlanVerdict {
if namespace == "mirror" {
gunnar_ssh::PlanVerdict::Refuse(SUSPENDED_WORDS.to_owned())
} else {
gunnar_ssh::PlanVerdict::Allow
}
}
fn pack_quota(&self, _: &str) -> gunnar_ssh::PlanVerdict {
gunnar_ssh::PlanVerdict::Allow
}
fn pack_ceiling(&self, _: &str) -> Option<gunnar_ssh::PackCeiling> {
None
}
}
/// ★ **A suspended namespace's sync is refused by its plan, and the job says
/// so as a field** (2026-10-04): `MIRROR_REFUSAL_PLAN`, beside the gate's own
/// sentence — not the cap, and not a fetch that failed.
#[test]
fn a_mirror_sync_into_a_suspended_namespace_is_refused_by_its_plan() {
let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
down.stores
.install_plan_gates(vec![Arc::new(Suspended) as Arc<dyn gunnar_ssh::PlanGate>]);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.mirror_sync(&down.path)
.expect("the sync is accepted");
let (job, _) = down.settle();
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(job.error.contains(SUSPENDED_WORDS), "{:?}", job.error);
assert_eq!(
job.refused_reason,
gunnar_proto::MirrorRefusal::Plan as i32,
"{job:?}"
);
assert_eq!(job.bytes, 0);
}
// ════════════════════════════════════════════════════════════════════════════
// Mirror.Add never takes over a repository that exists (2026-10-04)
// ════════════════════════════════════════════════════════════════════════════
/// `Mirror.Add` through the real gRPC service, as the account's owner.
fn add_as_owner(
down: &Downstream,
url: &str,
create_repo: bool,
adopt: bool,
) -> Result<gunnar_proto::AddMirrorResponse, tonic::Status> {
use gunnar_proto::mirror_server::Mirror as _;
let _ = down.stores.auth().add_principal(gunnar_core::Principal {
id: "mirror".to_owned(),
name: "mirror".to_owned(),
kind: gunnar_core::PrincipalKind::User,
created: 0,
});
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&down.stores)));
let service = gunnar_grpc::MirrorService::new(gunnar_grpc::ServiceContext::new(
plane,
Arc::new(gunnar_grpc::EventBus::new(16)),
gunnar_grpc::ServerConfig::default(),
));
let mut request = tonic::Request::new(gunnar_proto::AddMirrorRequest {
repo: Some(down.path.clone()),
remote_url: url.to_owned(),
prune: true,
create_repo,
adopt,
});
request.extensions_mut().insert(gunnar_grpc::Caller {
principal_id: "mirror".to_owned(),
fingerprint: "SHA256:owner".to_owned(),
machine: false,
authenticated: true,
});
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime")
.block_on(service.add(request))
.map(tonic::Response::into_inner)
}
/// ★ **An import whose name collides does not turn somebody's repository into
/// a mirror.** `Mirror.Add{create_repo}` over an existing repository that is
/// not a mirror is refused by name and nothing is registered — a pruning
/// sync would otherwise have moved its branches to the upstream's. `adopt`
/// says it on purpose; re-adding an existing MIRROR stays a no-op.
#[test]
fn a_mirror_that_would_create_its_repository_does_not_take_over_one_that_exists() {
let down = Downstream::open(HashKind::Sha256);
let url = "https://127.0.0.1:9/upstream.git";
let refused = add_as_owner(&down, url, true, false)
.expect_err("an existing, non-mirror repository is not taken over");
assert_eq!(
refused.code(),
tonic::Code::FailedPrecondition,
"{refused:?}"
);
assert!(
refused.message().contains("mirror/copy"),
"names it: {refused:?}"
);
assert!(
down.plane
.mirror_status(None, Some(&down.path))
.expect("mirror_status")
.is_empty(),
"a refused add registered the mirror anyway"
);
add_as_owner(&down, url, true, true).expect("adopt takes it over on purpose");
add_as_owner(&down, url, true, false).expect("re-adding an existing mirror is unchanged");
assert_eq!(
down.plane
.mirror_status(None, Some(&down.path))
.expect("mirror_status")
.len(),
1
);
}
// ════════════════════════════════════════════════════════════════════════════
// A mirror that creates its repository takes the upstream's object format
// ════════════════════════════════════════════════════════════════════════════
/// The downstream box with NO repository yet: `Mirror.Add { create_repo }`
/// registers the import, and its first sync makes the repository at the
/// upstream's format (2026-10-18) — which is what every console import does.
/// Returns the box, the finished first sync and the progress it showed.
type Imported = (
Downstream,
Oid,
Oid,
gunnar_proto::MirrorJob,
Vec<(bool, u64, u64)>,
);
fn import_creating(hash_of_upstream: HashKind) -> Imported {
let (_dir, stores, repo, head, blob) = http_upstream_stores(hash_of_upstream);
// Leaked on purpose for the test's life: the upstream must outlive the sync.
let upstream = Box::leak(Box::new(HttpUpstream::start(stores, repo, None)));
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
stores
.catalog()
.create_namespace("mirror")
.expect("namespace");
std::fs::write(
stores
.root()
.join(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
b"the upstreams in this test are on 127.0.0.1\n",
)
.expect("allowing private upstreams on this box");
let plane = CoreControlPlane::new(Arc::clone(&stores));
let path = RepoPath::new("mirror", "copy");
let repo_ref = RepoRef::new("mirror", "copy").expect("a legal path");
plane
.mirror_add(&path, &upstream.base(), false, true)
.expect("register the import");
assert!(
stores
.catalog()
.get(&repo_ref)
.expect("the catalog answers")
.is_none(),
"Mirror.Add made the repository before the upstream said its format"
);
plane.mirror_sync(&path).expect("the sync is accepted");
let deadline = Instant::now() + Duration::from_secs(120);
let mut seen: Vec<(bool, u64, u64)> = Vec::new();
let job = loop {
assert!(Instant::now() < deadline, "the import never settled");
let job = plane
.mirror_status(None, Some(&path))
.expect("mirror_status")
.into_iter()
.next()
.and_then(|m| m.job)
.expect("the job");
let reading = (job.has_progress, job.progress_done, job.progress_total);
if seen.last() != Some(&reading) {
seen.push(reading);
}
if job.state != MirrorJobState::Queued as i32 && job.state != MirrorJobState::Running as i32
{
break job;
}
std::thread::sleep(Duration::from_millis(5));
};
let repo = stores
.catalog()
.get(&repo_ref)
.expect("the catalog answers")
.unwrap_or_else(|| panic!("the first sync made no repository: {}", job.error));
let down = Downstream {
_dir: dir,
stores,
plane,
path,
repo,
};
(down, head, blob, job, seen)
}
/// ★ **A SHA-1 upstream — every repository on GitHub — mirrors into a
/// repository the import creates.** Before 2026-10-02 every such repository
/// was created at sha256 whatever the remote was, so the first sync of any
/// sha1 upstream failed with "this repository is object-format=sha256, but the
/// remote advertised sha1" — measured in the console's uitest, importing a
/// stock-git repository served by `git http-backend`.
#[test]
fn a_mirror_that_creates_its_repository_takes_a_sha1_upstreams_format() {
let (down, head, blob, done, progress) = import_creating(HashKind::Sha1);
assert_eq!(
down.repo.hash_kind,
HashKind::Sha1,
"the repository was created at the wrong format"
);
assert_mirrored(&down, &done, &progress, head, blob);
}
/// …and a sha256 upstream still lands at sha256.
#[test]
fn a_mirror_that_creates_its_repository_takes_a_sha256_upstreams_format() {
let (down, head, blob, done, progress) = import_creating(HashKind::Sha256);
assert_eq!(down.repo.hash_kind, HashKind::Sha256);
assert_mirrored(&down, &done, &progress, head, blob);
}
// ════════════════════════════════════════════════════════════════════════════
// An encrypted repository is never fetched into
// ════════════════════════════════════════════════════════════════════════════
/// The Pack frames the transaction log holds (what the twin would replay).
fn pack_frames(stores: &ServerStores) -> usize {
let mut n = 0;
stores
.txlog()
.scan_meta(|_, kind, _| {
if gunnar_core::txlog::Kind::from_u16(kind) == Some(gunnar_core::txlog::Kind::Pack) {
n += 1;
}
})
.expect("scan the transaction log");
n
}
/// **A mirror whose repository is encrypted lands NOTHING** (wiring tracer B
/// 2026-10-08). A mirror's pack is an upstream's plaintext and `PushIngest`
/// runs no encrypted gate on it, so the worker refuses the sync before the
/// dial: the job FAILS saying so, no object is stored, no Pack frame reaches
/// the twin, the reference does not move. The door refuses making such a
/// mirror (`Mirror.Add`, `Repos.SetEncrypted`); this is the registration that
/// predates the rule, or a flag set below the door.
#[test]
fn a_mirror_never_lands_plaintext_in_an_encrypted_repository() {
let (_dir, stores, repo, _head, blob) = http_upstream_stores(HashKind::Sha256);
let upstream = HttpUpstream::start(stores, repo, None);
let down = Downstream::open(HashKind::Sha256);
register(
&down,
&upstream.base(),
MirrorCredentialKind::Unspecified,
"",
"",
);
down.plane
.set_encrypted(&down.path, true)
.expect("the flag, set below the door");
let frames_before = pack_frames(&down.stores);
match down.plane.mirror_sync(&down.path) {
// Refused at enqueue: nothing ran.
Err(err) => assert!(format!("{err}").contains("encrypted"), "{err}"),
// Or enqueued and refused by the worker, before the dial.
Ok(_) => {
let (job, _) = down.settle();
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"the sync must FAIL: {job:?}"
);
assert!(
job.error.contains("encrypted"),
"the job says why: {}",
job.error
);
}
}
assert_eq!(
tip(down.git().serve(), "refs/heads/main"),
None,
"no reference moved"
);
assert!(
!GitOps::extents(down.git().serve(), &[blob.as_bytes()])
.expect("asking the store")
.first()
.is_some_and(Option::is_some),
"the upstream's plaintext blob is in the encrypted repository's store"
);
assert_eq!(
pack_frames(&down.stores),
frames_before,
"a pack was framed for the twin"
);
}