crates/gunnar-server/tests/mirror_import.rs
//! **Importing a repository from another git server, the way everybody meets
//! it** (owner order 2026-10-18).
//!
//! `mirror_sync.rs` proves the fetch against gunnar's own smart-HTTP answer.
//! That answer is not what GitHub, Codeberg (Forgejo) or stock
//! `git http-backend` send, and an audit of real imports found every https
//! import from GitHub and Codeberg failing on it. This file stands up an
//! upstream that answers the way those servers do, and asserts the import as
//! a person sees it:
//!
//! * the advertisement opens with the smart-HTTP service announcement
//! (`# service=git-upload-pack` and a flush) before `version 2`;
//! * the upstream's default branch is `master`, not `main`;
//! * a private upstream answers 401 until a token is set, and the first
//! import of it fails;
//! * the upstream keeps some files in Git LFS.
//!
//! Every upstream here is on loopback, so nothing reaches the network.
#![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, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gix_object::Kind;
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;
const TOKEN: &str = "glpat_ImportTestNeverShowsThis";
// ════════════════════════════════════════════════════════════════════════════
// The upstream: a repository, and an HTTP listener that answers like a forge
// ════════════════════════════════════════════════════════════════════════════
fn compute(hash: gix_hash::Kind, kind: Kind, content: &[u8]) -> Oid {
gix_object::compute_hash(hash, kind, content).expect("hashing an object")
}
/// A repository on its own stores, its default branch `branch`, holding one
/// commit of `files` on that branch. Returns the commit.
fn upstream_repo(
hash: HashKind,
branch: &str,
files: &[(&str, &[u8])],
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
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
.create_repo(
NewRepo::new(repo_ref)
.hash_kind(hash)
.default_branch(branch),
false,
)
.expect("create the upstream repository");
let store = stores.repo_store(&repo).expect("materialise it");
let kind = store.hash_kind();
let mut pack = PackBuilder::new(kind);
let mut sorted: Vec<&(&str, &[u8])> = files.iter().collect();
sorted.sort_by_key(|(name, _)| *name);
let mut tree = Vec::new();
for (name, content) in sorted {
let id = compute(kind, Kind::Blob, content);
pack.object(Kind::Blob, content.to_vec());
tree.extend_from_slice(format!("100644 {name}\0").as_bytes());
tree.extend_from_slice(id.as_bytes());
}
let tree_id = compute(kind, Kind::Tree, &tree);
pack.object(Kind::Tree, tree);
let who = "Alice <alice@example.com> 1700000000 +0000";
let commit = format!(
"tree {}\nauthor {who}\ncommitter {who}\n\nfirst\n",
tree_id.to_hex()
)
.into_bytes();
let commit_id = compute(kind, Kind::Commit, &commit);
pack.object(Kind::Commit, commit);
let built = pack.build().expect("building the pack");
store
.serve()
.put_pack(&built.bytes)
.expect("storing the pack");
store
.serve()
.update_ref(branch, None, Some(commit_id.as_bytes()))
.expect("pointing the branch at the commit");
(dir, stores, repo, commit_id)
}
/// A loopback smart-HTTP server in front of one repository.
struct Forge {
addr: SocketAddr,
stop: Arc<AtomicBool>,
_dir: tempfile::TempDir,
}
impl Forge {
/// `announce`: open the advertisement with `# service=git-upload-pack`
/// and a flush, as GitHub, Forgejo and `git http-backend` all do.
/// `require`: the `Authorization` value it insists on; anything else is a
/// 401 before the git seam, as a private repository answers.
fn start(
upstream: (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
Oid,
),
announce: bool,
require: Option<String>,
) -> (Forge, Oid) {
let (dir, stores, repo, head) = upstream;
let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind a loopback port");
let addr = listener.local_addr().expect("the bound address");
let stop = Arc::new(AtomicBool::new(false));
let stopping = Arc::clone(&stop);
std::thread::spawn(move || {
for connection in listener.incoming() {
if stopping.load(Ordering::SeqCst) {
return;
}
let Ok(mut connection) = connection else {
return;
};
let store = stores.repo_store(&repo).expect("open the upstream store");
let _ = serve(
&mut connection,
store.serve(),
store.hash_kind(),
announce,
require.as_deref(),
);
}
});
(
Forge {
addr,
stop,
_dir: dir,
},
head,
)
}
fn url(&self) -> String {
format!("http://{}/someone/project.git", self.addr)
}
}
impl Drop for Forge {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
let _ = std::net::TcpStream::connect(self.addr);
}
}
fn serve<S: Read + Write>(
connection: &mut S,
git: &dyn GitServe,
hash: gix_hash::Kind,
announce: bool,
require: Option<&str>,
) -> std::io::Result<()> {
// The request is read whole first (one request per connection), then answered on
// the same stream: a plain TCP stream or a TLS one (`Forge::start_tls`).
let (method, target, authorization, body) = {
let mut reader = BufReader::new(&mut *connection);
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;
}
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());
}
}
let mut body = vec![0u8; length];
reader.read_exact(&mut body)?;
(method, target, authorization, body)
};
if let Some(want) = require {
if authorization.as_deref() != Some(want) {
connection.write_all(
b"HTTP/1.1 401 Unauthorized\r\nWWW-Authenticate: Basic realm=\"forge\"\r\n\
Content-Length: 0\r\nConnection: close\r\n\r\n",
)?;
return connection.flush();
}
}
let upload = gunnar_wire::UploadPack::new(git, hash);
let mut out = Vec::new();
let content_type = if method == "GET" && target.contains("/info/refs") {
if announce {
out.extend_from_slice(b"001e# service=git-upload-pack\n0000");
}
upload
.advertise(gunnar_wire::ProtocolVersion::V2, &mut out)
.map_err(std::io::Error::other)?;
"application/x-git-upload-pack-advertisement"
} else {
let mut input = std::io::Cursor::new(body);
upload
.serve_session(
gunnar_wire::ProtocolVersion::V2,
gunnar_wire::Session::StatelessRpc,
&mut input,
&mut out,
)
.map_err(std::io::Error::other)?;
"application/x-git-upload-pack-result"
};
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(),
)?;
connection.write_all(&out)?;
connection.flush()
}
// ════════════════════════════════════════════════════════════════════════════
// The box importing it: a namespace and no repository yet
// ════════════════════════════════════════════════════════════════════════════
struct Box_ {
_dir: tempfile::TempDir,
stores: Arc<ServerStores>,
plane: CoreControlPlane,
path: RepoPath,
}
impl Box_ {
fn open() -> Box_ {
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 loopback upstreams on this box");
let plane = CoreControlPlane::new(Arc::clone(&stores));
Box_ {
_dir: dir,
stores,
plane,
path: RepoPath::new("mirror", "copy"),
}
}
/// What the console does for one import: Add (creating), the token if
/// any, Sync — then wait for the job.
fn import(&self, url: &str, token: Option<&str>) -> gunnar_proto::MirrorJob {
self.plane
.mirror_add(&self.path, url, false, true)
.expect("the import is registered");
if let Some(token) = token {
self.plane
.mirror_set_credential(&self.path, MirrorCredentialKind::Token, token, "")
.expect("the token is stored");
}
self.plane
.mirror_sync(&self.path)
.expect("the sync is accepted");
self.settle()
}
fn settle(&self) -> gunnar_proto::MirrorJob {
let deadline = Instant::now() + Duration::from_secs(120);
while Instant::now() < deadline {
let job = self
.config()
.job
.expect("a job that was just enqueued is known");
if job.state != MirrorJobState::Queued as i32
&& job.state != MirrorJobState::Running as i32
{
return job;
}
std::thread::sleep(Duration::from_millis(5));
}
panic!("the import never settled");
}
fn config(&self) -> gunnar_proto::MirrorConfig {
self.plane
.mirror_status(None, Some(&self.path))
.expect("mirror_status")
.into_iter()
.next()
.expect("the registration is there")
}
fn row(&self) -> Option<gunnar_core::model::Repo> {
self.stores
.catalog()
.get(&RepoRef::new("mirror", "copy").expect("a legal path"))
.expect("the catalog answers")
}
fn tip(&self, name: &str) -> Option<Oid> {
let row = self.row()?;
let store = self.stores.repo_store(&row).expect("open the store");
GitOps::refs(store.serve())
.expect("refs")
.into_iter()
.find(|r| r.name == name)
.and_then(|r| r.oid)
.map(|raw| Oid::try_from(&raw[..]).expect("an oid"))
}
fn head_target(&self) -> Option<String> {
let row = self.row()?;
let store = self.stores.repo_store(&row).expect("open the store");
store
.serve()
.head()
.expect("HEAD")
.and_then(|h| h.symref_target)
}
}
fn assert_done(job: &gunnar_proto::MirrorJob) {
assert_eq!(
job.state,
MirrorJobState::Done as i32,
"the import failed: {}",
job.error
);
}
// ════════════════════════════════════════════════════════════════════════════
// M1: the service announcement before `version 2`
// ════════════════════════════════════════════════════════════════════════════
/// ★ **Every https import from GitHub and Codeberg failed** with "the remote
/// opened with `# service=git-upload-pack` and not `version 2`". Smart HTTP
/// allows (and those servers send) the announcement and a flush before the
/// v2 capability advertisement; gunnar's own server does not, so nothing
/// here saw it.
#[test]
fn an_upstream_that_announces_its_service_first_is_imported_exactly() {
for hash in [HashKind::Sha1, HashKind::Sha256] {
let (forge, head) = Forge::start(
upstream_repo(hash, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
None,
);
let down = Box_::open();
let job = down.import(&forge.url(), None);
assert_done(&job);
assert_eq!(
down.row()
.expect("the import made its repository")
.hash_kind,
hash
);
assert_eq!(down.tip("refs/heads/main"), Some(head), "{hash:?}");
}
}
// ════════════════════════════════════════════════════════════════════════════
// M2: the upstream's default branch
// ════════════════════════════════════════════════════════════════════════════
/// ★ **An import of a `master` repository named `main` as its default
/// branch**, so its HEAD pointed at nothing and a clone checked out nothing.
/// The upstream's HEAD (the `symref-target` ls-refs gives) is the default
/// branch.
#[test]
fn an_import_takes_the_upstreams_default_branch() {
let (forge, head) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/master", &[("a.txt", b"one\n")]),
true,
None,
);
let down = Box_::open();
assert_done(&down.import(&forge.url(), None));
assert_eq!(down.tip("refs/heads/master"), Some(head));
assert_eq!(
down.row().expect("the repository").default_branch,
"refs/heads/master",
"the catalog's default branch is the upstream's"
);
assert_eq!(
down.head_target().as_deref(),
Some("refs/heads/master"),
"HEAD names the upstream's default branch"
);
}
// ════════════════════════════════════════════════════════════════════════════
// M3 / M5 / M6: a failed first import leaves nothing behind
// ════════════════════════════════════════════════════════════════════════════
/// ★ **A failed first import (a private repository, no token) used to leave
/// an empty private repository at sha256**, and a later import under the same
/// name failed with "object-format=sha256 but remote advertised sha1".
/// Now: no repository until the upstream has answered, the failure is in
/// plain words, and the same name imports cleanly afterwards.
#[test]
fn a_failed_first_import_leaves_no_repository_and_the_name_imports_later() {
let (private, _) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
Some(format!("Bearer {TOKEN}")),
);
let down = Box_::open();
let job = down.import(&private.url(), None);
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(
down.row().is_none(),
"a failed import left a repository behind"
);
let config = down.config();
assert!(
config.repo_pending,
"the registration says its repository is still to come"
);
let said = job.error.to_lowercase();
assert!(
said.contains("private") && said.contains("token"),
"the person is told what to do: {}",
job.error
);
for engineer in ["SetCredential", "Mirror.", "--kind", "`"] {
assert!(
!job.error.contains(engineer),
"engineer text {engineer:?} in a person's error: {}",
job.error
);
}
// The same name, the right (public, sha1) address.
let (public, head) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("b.txt", b"two\n")]),
true,
None,
);
assert_done(&down.import(&public.url(), None));
assert_eq!(
down.row().expect("the repository").hash_kind,
HashKind::Sha1
);
assert_eq!(down.tip("refs/heads/main"), Some(head));
assert!(!down.config().repo_pending);
}
/// …and a private upstream imports once its token is set: the credential
/// goes onto a registration whose repository does not exist yet.
#[test]
fn a_private_upstream_imports_with_its_token() {
let (forge, head) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
Some(format!("Bearer {TOKEN}")),
);
let down = Box_::open();
let job = down.import(&forge.url(), Some(TOKEN));
assert_done(&job);
assert_eq!(
down.row().expect("the repository").hash_kind,
HashKind::Sha1
);
assert_eq!(down.tip("refs/heads/main"), Some(head));
assert!(!format!("{:?}", down.config()).contains(TOKEN));
}
/// "Remove mirror" on a failed import removes it whole: no registration, no
/// repository.
#[test]
fn removing_a_failed_import_leaves_nothing() {
let (private, _) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
Some(format!("Bearer {TOKEN}")),
);
let down = Box_::open();
let job = down.import(&private.url(), None);
assert_eq!(job.state, MirrorJobState::Failed as i32);
down.plane
.mirror_remove(&down.path)
.expect("the failed import is removed");
assert!(down
.plane
.mirror_status(None, Some(&down.path))
.expect("status")
.is_empty());
assert!(down.row().is_none());
}
// ════════════════════════════════════════════════════════════════════════════
// M4: an ssh import is not created at a guessed format
// ════════════════════════════════════════════════════════════════════════════
/// ★ **ssh imports were created as sha256** whatever the upstream was: the
/// format was only ever asked over https. Now no import creates its
/// repository before the upstream has said its format, over any transport.
/// (Nothing is dialled by `Mirror.Add`; the address is a public IP literal.)
#[test]
fn an_ssh_import_does_not_create_its_repository_at_a_guessed_format() {
let down = Box_::open();
let config = down
.plane
.mirror_add(
&down.path,
"ssh://git@93.184.216.34/someone/project.git",
false,
true,
)
.expect("registered");
assert!(config.repo_pending);
assert!(
down.row().is_none(),
"the repository was created before the upstream said its format"
);
}
// ════════════════════════════════════════════════════════════════════════════
// M7: Git LFS is named, not silently skipped
// ════════════════════════════════════════════════════════════════════════════
/// ★ **LFS content is not mirrored** — and until now nothing said so. A
/// repository whose `.gitattributes` routes files through the LFS filter is
/// flagged on its mirror.
#[test]
fn an_upstream_using_lfs_is_flagged_and_one_without_is_not() {
let (lfs, _) = Forge::start(
upstream_repo(
HashKind::Sha1,
"refs/heads/main",
&[
(".gitattributes", b"*.bin filter=lfs diff=lfs merge=lfs -text\n"),
(
"model.bin",
b"version https://git-lfs.github.com/spec/v1\noid sha256:4d7a214614ab2935c943f9e0ff69d22eadbb8f32b1258daaa5e2ca24d17e2393\nsize 12345\n",
),
],
),
true,
None,
);
let down = Box_::open();
assert_done(&down.import(&lfs.url(), None));
assert!(
down.config().lfs_not_copied,
"the mirror must say its LFS files were not copied"
);
let (plain, _) = Forge::start(
upstream_repo(
HashKind::Sha1,
"refs/heads/main",
&[(".gitattributes", b"*.txt text\n")],
),
true,
None,
);
let other = Box_::open();
assert_done(&other.import(&plain.url(), None));
assert!(!other.config().lfs_not_copied);
}
// ════════════════════════════════════════════════════════════════════════════
// The doors: the namespace's owner tends an import whose repository is to come
// ════════════════════════════════════════════════════════════════════════════
fn owner_request<T>(message: T) -> tonic::Request<T> {
let mut request = tonic::Request::new(message);
request.extensions_mut().insert(gunnar_grpc::Caller {
principal_id: "mirror".to_owned(),
fingerprint: "SHA256:owner".to_owned(),
machine: false,
authenticated: true,
});
request
}
/// ★ The console signs as the person, and gunnar's repository-scoped grant
/// check answers "no" for a repository that is not in the catalog. A pending
/// import is the namespace's, as `Mirror.Add { create_repo }` is, so its owner
/// may set its token, sync it and remove it.
#[test]
fn the_namespace_owner_tends_an_import_whose_repository_is_still_to_come() {
use gunnar_proto::mirror_server::Mirror as _;
let down = Box_::open();
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 runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let repo = Some(down.path.clone());
runtime
.block_on(service.add(owner_request(gunnar_proto::AddMirrorRequest {
repo: repo.clone(),
remote_url: "https://93.184.216.34/someone/project.git".to_owned(),
prune: false,
create_repo: true,
adopt: false,
})))
.expect("the owner adds an import");
runtime
.block_on(
service.set_credential(owner_request(gunnar_proto::SetMirrorCredentialRequest {
repo: repo.clone(),
kind: MirrorCredentialKind::Token as i32,
secret: TOKEN.to_owned(),
username: String::new(),
})),
)
.expect("the owner sets the token of a pending import");
runtime
.block_on(service.remove(owner_request(gunnar_proto::RemoveMirrorRequest { repo })))
.expect("the owner removes a pending import");
}
// ════════════════════════════════════════════════════════════════════════════
// import-big (2026-10-26): an import is DONE only when its references are here
// ════════════════════════════════════════════════════════════════════════════
/// An upstream repository with no references at all.
fn empty_upstream(
hash: HashKind,
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
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
.create_repo(NewRepo::new(repo_ref).hash_kind(hash), false)
.expect("create the upstream repository");
stores.repo_store(&repo).expect("materialise it");
let null = Oid::null(gix_hash::Kind::Sha1);
(dir, stores, repo, null)
}
/// ★ **DONE with nothing in it** (owner test 2026-10-26, alica/linux on
/// rickardtest): a fetch whose plan came out empty applied nothing and the job
/// settled DONE over a repository with no references. An import that brought
/// no branch and no tag is a FAILED import that says why, and it leaves no
/// empty repository behind.
#[test]
fn an_import_that_brings_no_references_is_failed_and_leaves_nothing() {
let (forge, _) = Forge::start(empty_upstream(HashKind::Sha1), true, None);
let down = Box_::open();
let job = down.import(&forge.url(), None);
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"an import with 0 references settled {:?}: {job:?}",
job.state
);
assert!(job.error.contains("no branches or tags"), "{}", job.error);
assert!(down.row().is_none(), "an empty repository was left behind");
let config = down.config();
assert!(config.repo_pending, "the import is still to come");
assert_eq!(
config.last_error, job.error,
"the registration keeps the reason"
);
}
/// What a server that stopped in the middle of an import leaves on disk: the
/// registration, its in-flight mark, and the repository the import had made,
/// with nothing in it. The job queue — memory — is gone with the process.
fn interrupted_import(down: &Box_, url: &str) {
down.plane
.mirror_add(&down.path, url, false, true)
.expect("the import is registered");
let repo_ref = RepoRef::new("mirror", "copy").expect("a legal path");
down.stores
.mirrors()
.mark_started(&repo_ref, 1_700_000_000_000)
.expect("marked")
.expect("registered");
down.stores
.create_repo(NewRepo::new(repo_ref).hash_kind(HashKind::Sha1), false)
.expect("the repository the import had made");
}
/// ★ **An import the server did not live to finish is resumed, not left
/// empty** (import-big). The restarted box settles it as interrupted — the
/// reason is on the registration and the account's audit log — and runs it
/// again, and it lands whole.
#[test]
fn an_import_interrupted_by_a_restart_is_resumed_and_lands() {
let (forge, head) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/master", &[("a.txt", b"one\n")]),
true,
None,
);
let down = Box_::open();
interrupted_import(&down, &forge.url());
// The restart: a new control plane over the same data.
let restarted = CoreControlPlane::new(Arc::clone(&down.stores));
assert_eq!(
restarted.resume_interrupted_mirrors(),
1,
"the interrupted import is resumed"
);
let deadline = Instant::now() + Duration::from_secs(120);
let job = loop {
let config = restarted
.mirror_status(None, Some(&down.path))
.expect("status")
.into_iter()
.next()
.expect("registered");
let job = config.job.expect("the resumed import has a job");
if job.state != MirrorJobState::Queued as i32 && job.state != MirrorJobState::Running as i32
{
break job;
}
assert!(
Instant::now() < deadline,
"the resumed import never settled"
);
std::thread::sleep(Duration::from_millis(5));
};
assert_done(&job);
assert_eq!(down.tip("refs/heads/master"), Some(head));
let record = down
.stores
.mirrors()
.get(&RepoRef::new("mirror", "copy").unwrap())
.unwrap()
.unwrap();
assert_eq!(record.sync_started_at_unix_ms, None, "settled");
assert_eq!(
record.interrupted, 0,
"a sync that settled resets the count"
);
assert!(record.last_error.is_empty());
}
/// ★ **A box that restarts during every attempt stops trying, and says so**:
/// past `MAX_RESUMES` interruptions in a row the import is FAILED with the
/// reason, shown as a FAILED job although the queue holds none — so the
/// repository page says why it is empty instead of "idle".
#[test]
fn an_import_interrupted_again_and_again_is_failed_with_its_reason() {
let down = Box_::open();
interrupted_import(&down, "https://93.184.216.34/someone/project.git");
let repo_ref = RepoRef::new("mirror", "copy").unwrap();
for round in 1..=gunnar_core::MAX_RESUMES {
let settled = down
.stores
.mirrors()
.settle_interrupted(1_700_000_600_000)
.unwrap();
assert_eq!(settled.len(), 1);
assert_eq!(settled[0].interrupted, round);
down.stores
.mirrors()
.mark_started(&repo_ref, 1_700_000_000_000)
.unwrap();
}
let restarted = CoreControlPlane::new(Arc::clone(&down.stores));
assert_eq!(
restarted.resume_interrupted_mirrors(),
0,
"not tried again by itself"
);
let config = restarted
.mirror_status(None, Some(&down.path))
.unwrap()
.into_iter()
.next()
.unwrap();
let job = config.job.expect("the failure is shown as a job");
assert_eq!(job.state, MirrorJobState::Failed as i32);
assert!(
job.error.contains("restarted") && job.error.contains("not tried again"),
"{}",
job.error
);
assert_eq!(config.last_error, job.error);
}
/// **A synthetic big upstream**: `commits` commits on `refs/heads/master`,
/// each adding an 8 KiB file of incompressible bytes, and a tag on every
/// `tag_every`-th commit. Returns the tip.
fn big_upstream(
commits: usize,
tag_every: usize,
) -> (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
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
.create_repo(
NewRepo::new(repo_ref)
.hash_kind(HashKind::Sha1)
.default_branch("refs/heads/master"),
false,
)
.expect("create the upstream repository");
let store = stores.repo_store(&repo).expect("materialise it");
let kind = store.hash_kind();
let mut pack = PackBuilder::new(kind);
let mut seed: u64 = 0x9e37_79b9_7f4a_7c15;
let mut parent: Option<Oid> = None;
let mut tags = Vec::new();
for n in 0..commits {
let mut blob = Vec::with_capacity(8192);
while blob.len() < 8192 {
seed ^= seed << 13;
seed ^= seed >> 7;
seed ^= seed << 17;
blob.extend_from_slice(&seed.to_le_bytes());
}
let blob_id = compute(kind, Kind::Blob, &blob);
pack.object(Kind::Blob, blob);
let mut tree = Vec::new();
tree.extend_from_slice(format!("100644 f{n:05}\0").as_bytes());
tree.extend_from_slice(blob_id.as_bytes());
let tree_id = compute(kind, Kind::Tree, &tree);
pack.object(Kind::Tree, tree);
let who = format!("Alice <alice@example.com> {} +0000", 1_700_000_000 + n);
let parent_line = parent
.map(|p| format!("parent {}\n", p.to_hex()))
.unwrap_or_default();
let commit = format!(
"tree {}\n{parent_line}author {who}\ncommitter {who}\n\ncommit {n}\n",
tree_id.to_hex()
)
.into_bytes();
let commit_id = compute(kind, Kind::Commit, &commit);
pack.object(Kind::Commit, commit);
if n % tag_every == 0 {
tags.push((format!("refs/tags/v{n}"), commit_id));
}
parent = Some(commit_id);
}
let built = pack.build().expect("building the pack");
store
.serve()
.put_pack(&built.bytes)
.expect("storing the pack");
let tip = parent.expect("at least one commit");
store
.serve()
.update_ref("refs/heads/master", None, Some(tip.as_bytes()))
.expect("the branch");
for (name, id) in &tags {
store
.serve()
.update_ref(name, None, Some(id.as_bytes()))
.expect("a tag");
}
(dir, stores, repo, tip)
}
#[derive(Debug)]
struct FixedDisk(u64, u64);
impl gunnar_ssh::SpaceProbe for FixedDisk {
fn measure(&self) -> std::io::Result<(u64, u64)> {
Ok((self.0, self.1))
}
}
/// This box's data set, measured as `free` bytes over a 64 MiB floor.
fn with_room(down: &Box_, free_above_floor: u64) {
const MIB: u64 = 1 << 20;
down.stores.install_space_ledger(Arc::new(
gunnar_ssh::SpaceLedger::with_probe(
down.stores.root(),
gunnar_ssh::DiskFloor {
percent: 0,
bytes: 64 * MIB,
},
Arc::new(FixedDisk(64 * MIB + free_above_floor, 10 * 1024 * MIB)),
)
.with_small_lane(0),
));
}
/// ★ **A big import that does not fit is refused by name** — FAILED with the
/// room's own line, no empty repository left behind, and the reason still on
/// the registration once the queue has forgotten the job. The room is lowered
/// in the test (a 3 MiB pack against 512 KiB) to force the path a kernel-sized
/// import takes on a small box.
#[test]
fn a_big_import_past_the_room_is_failed_by_name_and_leaves_nothing() {
let (forge, _) = Forge::start(big_upstream(400, 10), true, None);
let down = Box_::open();
with_room(&down, 512 << 10);
let job = down.import(&forge.url(), None);
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(!job.error.is_empty());
assert!(down.row().is_none(), "an empty repository was left behind");
assert!(down.config().repo_pending);
assert_eq!(down.config().last_error, job.error);
// …and with the room, the same upstream lands whole: every tag.
let roomy = Box_::open();
with_room(&roomy, 1 << 30);
let (forge, tip) = Forge::start(big_upstream(400, 10), true, None);
let job = roomy.import(&forge.url(), None);
assert_done(&job);
assert_eq!(roomy.tip("refs/heads/master"), Some(tip));
assert!(roomy.tip("refs/tags/v390").is_some());
assert_eq!(job.updates.len(), 41, "the branch and 40 tags");
}
/// Every `index.scanned` event the storage engine says, by the repository it
/// names: a reader that resolved a pending pack for itself. One process-wide
/// sink; each test reads only its own box's repositories out of it.
fn scans_seen() -> &'static std::sync::Mutex<Vec<String>> {
use tracing_subscriber::layer::SubscriberExt as _;
static SEEN: std::sync::OnceLock<&'static std::sync::Mutex<Vec<String>>> =
std::sync::OnceLock::new();
SEEN.get_or_init(|| {
let seen: &'static std::sync::Mutex<Vec<String>> =
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
struct Scans(&'static std::sync::Mutex<Vec<String>>);
#[derive(Default)]
struct Fields {
event: String,
repo: String,
}
impl tracing::field::Visit for Fields {
fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
match field.name() {
"event" => self.event = value.to_owned(),
"repo" => self.repo = value.to_owned(),
_ => {}
}
}
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
match field.name() {
"event" => self.event = format!("{value:?}").trim_matches('"').to_owned(),
"repo" => self.repo = format!("{value:?}").trim_matches('"').to_owned(),
_ => {}
}
}
}
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Scans {
fn on_event(
&self,
event: &tracing::Event<'_>,
_: tracing_subscriber::layer::Context<'_, S>,
) {
let mut fields = Fields::default();
event.record(&mut fields);
if fields.event == "index.scanned" {
self.0.lock().unwrap().push(fields.repo);
}
}
}
let _ = tracing::subscriber::set_global_default(
tracing_subscriber::registry().with(Scans(seen)),
);
seen
})
}
/// ★ **The fetched pack is resolved once, by the indexer, and never a second
/// time by the import itself** (import-big, the cause of the empty alica/linux).
///
/// The reference write checks every new tip exists, and before the pack was
/// indexed that check made the engine resolve the whole pending pack on the
/// import's thread beside the indexer — two whole-pack resolvers, 19.4 GB peak
/// for torvalds/linux, past a 16 GB box. `receive_pack` indexes first
/// (`index_now`); the import now does too. Asserted on the engine's own
/// `index.scanned` line, which only the read-time fallback says.
#[test]
fn an_import_indexes_its_pack_once_and_never_scans_it_for_itself() {
let seen = scans_seen();
let (forge, tip) = Forge::start(big_upstream(1500, 25), true, None);
let down = Box_::open();
let job = down.import(&forge.url(), None);
assert_done(&job);
assert_eq!(down.tip("refs/heads/master"), Some(tip));
let root = down.stores.root().display().to_string();
let ours: Vec<String> = seen
.lock()
.unwrap()
.iter()
.filter(|repo| repo.contains(&root))
.cloned()
.collect();
assert!(
ours.is_empty(),
"the import resolved its pending pack for itself {} time(s) beside the indexer: {ours:?}",
ours.len()
);
}
/// ★ **A resumed import does not fetch its pack again** when the restart came
/// after the pack landed — the index, the longest step of a big import. The
/// objects are durable; only the references were never written. The resumed
/// sync offers what the store already has, the upstream sends nothing it sent
/// before, and the references land.
#[test]
fn an_import_interrupted_after_its_pack_landed_resumes_without_fetching_it_again() {
let (forge, tip) = Forge::start(big_upstream(300, 30), true, None);
let down = Box_::open();
let first = down.import(&forge.url(), None);
assert_done(&first);
assert!(
first.bytes > 2 << 20,
"the first fetch brought the pack: {} bytes",
first.bytes
);
// The state a restart in the index leaves: the pack stored, no reference
// written, the sync in flight.
let row = down.row().expect("the repository");
let store = down.stores.repo_store(&row).expect("open the store");
let refs = GitOps::refs(store.serve()).expect("refs");
let deletes: Vec<gunnar_store::git_storage_trait::RefCas<'_>> = refs
.iter()
.filter(|r| r.name.starts_with("refs/"))
.map(|r| gunnar_store::git_storage_trait::RefCas {
name: r.name.clone(),
old: r.oid.as_deref(),
new: None,
peeled: None,
})
.collect();
GitOps::put_refs_cas(store.serve(), &deletes).expect("the references are gone");
assert_eq!(down.tip("refs/heads/master"), None);
let repo_ref = RepoRef::new("mirror", "copy").unwrap();
down.stores
.mirrors()
.mark_started(&repo_ref, 1_700_000_000_000)
.unwrap();
let restarted = CoreControlPlane::new(Arc::clone(&down.stores));
assert_eq!(restarted.resume_interrupted_mirrors(), 1);
let deadline = Instant::now() + Duration::from_secs(120);
let job = loop {
let job = restarted
.mirror_status(None, Some(&down.path))
.unwrap()
.into_iter()
.next()
.unwrap()
.job
.expect("a job");
if job.state != MirrorJobState::Queued as i32 && job.state != MirrorJobState::Running as i32
{
break job;
}
assert!(
Instant::now() < deadline,
"the resumed import never settled"
);
std::thread::sleep(Duration::from_millis(5));
};
assert_done(&job);
assert_eq!(down.tip("refs/heads/master"), Some(tip));
assert!(down.tip("refs/tags/v270").is_some());
assert!(
job.bytes < 4096,
"the pack was fetched again: {} bytes",
job.bytes
);
}
// ════════════════════════════════════════════════════════════════════════════
// Heimdall milestone 5: the fetched pack is checked in a Heimdall job first
// ════════════════════════════════════════════════════════════════════════════
#[cfg(feature = "heimdall")]
mod heimdall_gate {
use super::*;
use gunnar_heimdall::{checker, PackGate, CHECKER_PATH, PACK, REPORT};
use heimdall_core::*;
use std::sync::Mutex;
/// Plays Heimdall in-process: the real checker on /in/pack, a signed
/// result; or refuses every pack.
struct Fake {
key: BootKey,
refuse: bool,
jobs: Mutex<u32>,
}
fn features() -> Features {
Features {
runner: "fake".into(),
container: true,
seccomp: Protection::Default,
landlock: true,
broker: false,
wasm: false,
image_verified_each_start: true,
}
}
impl JobRunner for Fake {
fn run(&self, job: Job) -> Result<SignedResult, RunError> {
*self.jobs.lock().unwrap() += 1;
let JobKind::Native { argv, .. } = &job.kind else {
panic!("a native job")
};
assert_eq!(argv[0], CHECKER_PATH);
let mut outputs = vec![];
let verdict = if self.refuse {
Verdict::Refused {
code: "heimdall-exit-1".into(),
reason: "the checker said no".into(),
}
} else {
let kind = checker::parse_kind(&argv[1]).unwrap();
let r = checker::check_pack_file(&job.input_dir.join(PACK), kind).unwrap();
let b = serde_json::to_vec(&r).unwrap();
std::fs::write(job.output_dir.join(REPORT), &b).unwrap();
outputs.push(Output::of(REPORT, &b));
Verdict::Accepted
};
let result = JobResult {
job_id: job.id,
nonce: job.nonce,
verdict,
outputs,
exit: ExitInfo::exited(0),
started_unix: 1,
finished_unix: 2,
runner: features(),
events: vec![],
kernel: None,
};
Ok(self.key.sign(result).unwrap())
}
fn verifying_key(&self) -> VerifyingKey {
self.key.verifying_key()
}
fn features(&self) -> Features {
features()
}
}
fn gate(down: &Box_, refuse: bool) -> Arc<Fake> {
let fake = Arc::new(Fake {
key: BootKey::generate(),
refuse,
jobs: Mutex::new(0),
});
let image = ImageRef {
name: "gunnar/pack-check".into(),
digest: format!("sha256:{}", "ab".repeat(32)),
};
let g = PackGate::new(
fake.clone(),
image,
down.stores.root().join("heimdall/spool"),
gix_hash::Kind::Sha1,
);
assert!(down.stores.set_pack_gate(Arc::new(g)));
fake
}
/// ★ The fetched pack goes through the gate before the ingest: a gate
/// that refuses it lands nothing, and says Heimdall refused it.
#[test]
fn an_import_whose_pack_heimdall_refuses_lands_nothing() {
let (forge, _) = Forge::start(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
None,
);
let down = Box_::open();
let fake = gate(&down, true);
let job = down.import(&forge.url(), None);
assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
assert!(job.error.contains("Heimdall"), "{}", job.error);
assert_eq!(down.tip("refs/heads/main"), None);
assert!(*fake.jobs.lock().unwrap() >= 1, "the gate ran a job");
}
/// The same import through a gate whose check passes lands exactly, at
/// both hash kinds (the gate checks at the repository's kind).
#[test]
fn an_import_through_a_passing_gate_lands_exactly() {
for hash in [HashKind::Sha1, HashKind::Sha256] {
let (forge, head) = Forge::start(
upstream_repo(hash, "refs/heads/main", &[("a.txt", b"one\n")]),
true,
None,
);
let down = Box_::open();
let fake = gate(&down, false);
assert_done(&down.import(&forge.url(), None));
assert_eq!(down.tip("refs/heads/main"), Some(head), "{hash:?}");
assert_eq!(*fake.jobs.lock().unwrap(), 1, "{hash:?}");
}
}
}
// ════════════════════════════════════════════════════════════════════════════
// GUN-MIRROR-CA: an https upstream behind a private CA
// ════════════════════════════════════════════════════════════════════════════
/// A private CA and a `localhost` server certificate under it: `(ca pem, leaf pem,
/// the TLS server's config)`.
fn private_ca() -> (String, String, Arc<rustls::ServerConfig>) {
let ca_key = rcgen::KeyPair::generate().expect("a CA key");
let mut ca_params = rcgen::CertificateParams::new(Vec::<String>::new()).expect("CA params");
ca_params
.distinguished_name
.push(rcgen::DnType::CommonName, "Acme Private Root");
ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
ca_params.key_usages = vec![
rcgen::KeyUsagePurpose::KeyCertSign,
rcgen::KeyUsagePurpose::CrlSign,
];
ca_params.not_after = rcgen::date_time_ymd(2040, 1, 1);
let ca_cert = ca_params.self_signed(&ca_key).expect("the CA");
let issuer = rcgen::Issuer::new(ca_params, ca_key);
let leaf_key = rcgen::KeyPair::generate().expect("a server key");
let mut leaf_params =
rcgen::CertificateParams::new(vec!["localhost".to_owned()]).expect("leaf params");
leaf_params.not_after = rcgen::date_time_ymd(2040, 1, 1);
let leaf = leaf_params
.signed_by(&leaf_key, &issuer)
.expect("the server certificate");
let config = rustls::ServerConfig::builder_with_provider(Arc::new(
rustls::crypto::ring::default_provider(),
))
.with_safe_default_protocol_versions()
.expect("TLS versions")
.with_no_client_auth()
.with_single_cert(
vec![leaf.der().clone(), ca_cert.der().clone()],
rustls::pki_types::PrivateKeyDer::Pkcs8(leaf_key.serialize_der().into()),
)
.expect("the TLS server config");
(ca_cert.pem(), leaf.pem(), Arc::new(config))
}
impl Forge {
/// [`Forge::start`] over **https**, its certificate issued by a private CA.
fn start_tls(
upstream: (
tempfile::TempDir,
Arc<ServerStores>,
gunnar_core::model::Repo,
Oid,
),
tls: Arc<rustls::ServerConfig>,
) -> (Forge, Oid) {
let (dir, stores, repo, head) = upstream;
let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind a loopback port");
let addr = listener.local_addr().expect("the bound address");
let stop = Arc::new(AtomicBool::new(false));
let stopping = Arc::clone(&stop);
std::thread::spawn(move || {
for connection in listener.incoming() {
if stopping.load(Ordering::SeqCst) {
return;
}
let Ok(connection) = connection else {
return;
};
let Ok(session) = rustls::ServerConnection::new(Arc::clone(&tls)) else {
continue;
};
let mut stream = rustls::StreamOwned::new(session, connection);
let store = stores.repo_store(&repo).expect("open the upstream store");
let _ = serve(&mut stream, store.serve(), store.hash_kind(), true, None);
stream.conn.send_close_notify();
let _ = stream.flush();
}
});
(
Forge {
addr,
stop,
_dir: dir,
},
head,
)
}
fn https_url(&self) -> String {
format!("https://localhost:{}/someone/project.git", self.addr.port())
}
}
impl Box_ {
fn add(&self, path: &RepoPath, url: &str) {
self.plane
.mirror_add(path, url, false, true)
.expect("the import is registered");
}
fn sync_and_settle(&self, path: &RepoPath) -> gunnar_proto::MirrorJob {
self.plane.mirror_sync(path).expect("the sync is accepted");
let deadline = Instant::now() + Duration::from_secs(120);
while Instant::now() < deadline {
let job = self
.plane
.mirror_status(None, Some(path))
.expect("mirror_status")
.into_iter()
.next()
.and_then(|c| c.job)
.expect("a job");
if job.state != MirrorJobState::Queued as i32
&& job.state != MirrorJobState::Running as i32
{
return job;
}
std::thread::sleep(Duration::from_millis(5));
}
panic!("the import never settled");
}
}
/// ★ **GUN-MIRROR-CA** (field test t-mirror.sh 2026-11-01, `invalid peer certificate:
/// UnknownIssuer`): an https upstream behind a private CA is refused while the mirror
/// trusts only the public roots, and imports once THAT mirror holds the CA. The mirror
/// shows the CA's subject and SHA-256 fingerprint.
#[test]
fn a_private_ca_upstream_is_refused_without_its_ca_and_imports_with_it() {
let (ca_pem, _leaf, tls) = private_ca();
let (forge, head) = Forge::start_tls(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
tls,
);
let without = Box_::open();
without.add(&without.path, &forge.https_url());
let job = without.sync_and_settle(&without.path);
assert_eq!(
job.state,
MirrorJobState::Failed as i32,
"a private CA's certificate was trusted with no CA set"
);
assert!(
job.error.contains("UnknownIssuer") || job.error.to_lowercase().contains("certificate"),
"the refusal says it is the certificate: {}",
job.error
);
let with = Box_::open();
with.add(&with.path, &forge.https_url());
let config = with
.plane
.mirror_set_ca(&with.path, &ca_pem)
.expect("the mirror takes its CA");
assert!(
config.ca_subject.contains("Acme Private Root"),
"the CA's subject is shown: {:?}",
config.ca_subject
);
assert!(
config.ca_fingerprint_sha256.len() == 64,
"the CA's SHA-256 fingerprint is shown: {:?}",
config.ca_fingerprint_sha256
);
assert_done(&with.sync_and_settle(&with.path));
assert_eq!(with.tip("refs/heads/main"), Some(head));
}
/// ★ **A CA set on one mirror is trusted by no other**: two mirrors of the same private-CA
/// upstream on one box, the CA on the first only — the second is still refused.
#[test]
fn a_ca_set_on_one_mirror_is_trusted_by_no_other() {
let (ca_pem, _leaf, tls) = private_ca();
let (forge, head) = Forge::start_tls(
upstream_repo(HashKind::Sha1, "refs/heads/main", &[("a.txt", b"one\n")]),
tls,
);
let down = Box_::open();
let other = RepoPath::new("mirror", "other");
down.add(&down.path, &forge.https_url());
down.add(&other, &forge.https_url());
down.plane
.mirror_set_ca(&down.path, &ca_pem)
.expect("the first mirror takes the CA");
assert_done(&down.sync_and_settle(&down.path));
assert_eq!(down.tip("refs/heads/main"), Some(head));
let refused = down.sync_and_settle(&other);
assert_eq!(
refused.state,
MirrorJobState::Failed as i32,
"the second mirror trusted a CA set on the first"
);
let shown = down
.plane
.mirror_status(None, Some(&other))
.expect("status")
.into_iter()
.next()
.expect("the second mirror");
assert!(shown.ca_subject.is_empty() && shown.ca_fingerprint_sha256.is_empty());
}
/// ★ **Only a CA is taken, by name**: a server's own (leaf) certificate, a private key
/// and text with no certificate are each refused with a sentence saying which; an empty
/// text forgets the CA.
#[test]
fn a_leaf_a_key_or_no_certificate_is_refused_by_name_and_empty_forgets_the_ca() {
let (ca_pem, leaf_pem, _tls) = private_ca();
let down = Box_::open();
down.add(&down.path, "https://localhost:1/someone/project.git");
let refused = |pem: &str| {
format!(
"{}",
down.plane
.mirror_set_ca(&down.path, pem)
.expect_err("refused")
)
};
assert!(
refused(&leaf_pem).contains("not a CA"),
"{}",
refused(&leaf_pem)
);
let key = "-----BEGIN PRIVATE KEY-----\nAAAA\n-----END PRIVATE KEY-----\n";
assert!(refused(key).contains("private key"), "{}", refused(key));
assert!(
refused("hello").contains("no certificate"),
"{}",
refused("hello")
);
let set = down
.plane
.mirror_set_ca(&down.path, &ca_pem)
.expect("a CA is taken");
assert!(!set.ca_subject.is_empty());
let cleared = down
.plane
.mirror_set_ca(&down.path, "")
.expect("empty forgets it");
assert!(cleared.ca_subject.is_empty() && cleared.ca_fingerprint_sha256.is_empty());
}