gunnar-server 3.0.0

//! **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(
    connection: &mut std::net::TcpStream,
    git: &dyn GitServe,
    hash: gix_hash::Kind,
    announce: bool,
    require: Option<&str>,
) -> 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;
        }
        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(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 mut body = vec![0u8; length];
    reader.read_exact(&mut body)?;

    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
    );
}