gunnar-core 3.0.0

//! What the control tables cost.
//!
//! The identity table and the catalog are each one JSON document behind one
//! reference (`doc.rs`). That seam is right: no database, content-addressed
//! history, a real compare-and-swap. What was wrong was the **unit**. One
//! document meant one parse of every principal, every key and every grant on
//! the server, four times per SSH connection, plus one parse of every
//! repository on top.
//!
//! # Why these tests count store touches rather than timing anything
//!
//! A wall clock passes on a small fixture no matter what the complexity is. So
//! the assertions here are shaped like `crates/gunnar-wire/tests/
//! connectivity_cost.rs`: **`cost(small N)` against `cost(large N)`**, on a
//! counted store, with the liveness half asserted alongside. A lookup that
//! touched nothing at all would otherwise satisfy every "constant" claim in
//! this file.
//!
//! Four separate claims live here, and they fail independently:
//!
//! 1. a **warm** connection sequence reads **zero** objects, at any table size;
//! 2. a **cold** one reads a bounded slice of the table, not all of it;
//! 3. creating a repository does not rewrite the catalog; and
//! 4. none of the above cost the compare-and-swap its correctness, proven by
//!    running the same concurrency test against a deliberately broken
//!    reference store in the same run.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Barrier};

use gunnar_core::{
    AuthTable, BlobId, ControlStore, DocVersion, Fingerprint, Grant, Level, NewRepo, Principal,
    PrincipalKind, RedbControlStore, RepoCatalog, RepoRef, Scope, Visibility,
};

// ───────────────────────────── the counted store ─────────────────────────────

/// Every touch the control plane makes on its storage, counted.
///
/// Reads and writes are counted both as *calls* and as *bytes*, deliberately
/// separately. A fix that turned one 300 MB read into three 300 MB reads would
/// look like a regression by call count and like nothing at all by neither.
/// A fix that turned one 300 MB read into three 4 KB reads is the actual
/// goal, which only the byte counter can see.
///
/// `blob_reads` is what used to be `object_reads`: the certificate-identity plan moved these
/// documents off the git object store, and this file counts the seam they moved
/// to. The four claims below are unchanged, and so are their numbers.
#[derive(Default)]
struct Counts {
    blob_reads: AtomicUsize,
    blob_read_bytes: AtomicUsize,
    blob_writes: AtomicUsize,
    blob_write_bytes: AtomicUsize,
    /// Reads of a document's *pointer* — the one probe per call that makes the
    /// cache coherent, and the liveness half of the warm-path claim.
    doc_reads: AtomicUsize,
}

impl Counts {
    fn reset(&self) {
        self.blob_reads.store(0, Ordering::SeqCst);
        self.blob_read_bytes.store(0, Ordering::SeqCst);
        self.blob_writes.store(0, Ordering::SeqCst);
        self.blob_write_bytes.store(0, Ordering::SeqCst);
        self.doc_reads.store(0, Ordering::SeqCst);
    }
    fn object_reads(&self) -> usize {
        self.blob_reads.load(Ordering::SeqCst)
    }
    fn object_read_bytes(&self) -> usize {
        self.blob_read_bytes.load(Ordering::SeqCst)
    }
    fn object_write_bytes(&self) -> usize {
        self.blob_write_bytes.load(Ordering::SeqCst)
    }
    fn ref_reads(&self) -> usize {
        self.doc_reads.load(Ordering::SeqCst)
    }
}

/// A [`ControlStore`] that counts what goes through it, and that can be told to
/// **ignore the compare-and-swap expectation** — the exact defect a CAS exists
/// to catch, and the only way to see the guard below fail.
struct CountedStore {
    inner: RedbControlStore,
    counts: Arc<Counts>,
    clobber: bool,
}

impl ControlStore for CountedStore {
    fn put_blob(&self, bytes: &[u8]) -> gunnar_core::Result<BlobId> {
        self.counts.blob_writes.fetch_add(1, Ordering::SeqCst);
        self.counts
            .blob_write_bytes
            .fetch_add(bytes.len(), Ordering::SeqCst);
        self.inner.put_blob(bytes)
    }

    fn get_blob(&self, id: &BlobId) -> gunnar_core::Result<Option<Vec<u8>>> {
        self.counts.blob_reads.fetch_add(1, Ordering::SeqCst);
        let out = self.inner.get_blob(id)?;
        if let Some(bytes) = &out {
            self.counts
                .blob_read_bytes
                .fetch_add(bytes.len(), Ordering::SeqCst);
        }
        Ok(out)
    }

    fn read_doc(&self, name: &str) -> gunnar_core::Result<Option<BlobId>> {
        self.counts.doc_reads.fetch_add(1, Ordering::SeqCst);
        self.inner.read_doc(name)
    }

    fn commit_doc(
        &self,
        name: &str,
        expected: Option<BlobId>,
        new: BlobId,
    ) -> gunnar_core::Result<bool> {
        let expected = if self.clobber {
            // Force: whatever is there now is "expected", so the commit always
            // lands and a losing writer silently overwrites the winner.
            self.inner.read_doc(name)?
        } else {
            expected
        };
        self.inner.commit_doc(name, expected, new)
    }

    fn history(&self, name: &str) -> gunnar_core::Result<Vec<DocVersion>> {
        self.inner.history(name)
    }
}

/// A control plane over a counted store.
struct Counted {
    counts: Arc<Counts>,
    store: Arc<dyn ControlStore>,
    auth: AuthTable,
    catalog: RepoCatalog,
}

impl Counted {
    fn new() -> Self {
        Self::with_clobber(false)
    }

    fn with_clobber(clobber: bool) -> Self {
        let counts = Arc::new(Counts::default());
        let store: Arc<dyn ControlStore> = Arc::new(CountedStore {
            inner: RedbControlStore::in_memory().expect("control store"),
            counts: Arc::clone(&counts),
            clobber,
        });
        Counted {
            auth: AuthTable::new(Arc::clone(&store)).expect("auth table"),
            catalog: RepoCatalog::new(Arc::clone(&store)).expect("catalog"),
            counts,
            store,
        }
    }

    /// A second, independent pair of handles onto the same storage: what a
    /// restarted process, or another thread that opened its own, would see.
    /// Its document cache starts cold.
    fn cold_handles(&self) -> (AuthTable, RepoCatalog) {
        (
            AuthTable::new(Arc::clone(&self.store)).expect("auth table"),
            RepoCatalog::new(Arc::clone(&self.store)).expect("catalog"),
        )
    }
}

// ───────────────────────────── synthetic keys ────────────────────────────────

/// Standard base64, written out rather than depended on: this crate's `base64ct`
/// is a normal dependency and integration tests only see dev-dependencies.
fn b64(bytes: &[u8]) -> String {
    const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut out = String::new();
    for chunk in bytes.chunks(3) {
        let b = [
            chunk[0],
            *chunk.get(1).unwrap_or(&0),
            *chunk.get(2).unwrap_or(&0),
        ];
        let n = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
        out.push(ALPHABET[(n >> 18) as usize & 63] as char);
        out.push(ALPHABET[(n >> 12) as usize & 63] as char);
        out.push(if chunk.len() > 1 {
            ALPHABET[(n >> 6) as usize & 63] as char
        } else {
            '='
        });
        out.push(if chunk.len() > 2 {
            ALPHABET[n as usize & 63] as char
        } else {
            '='
        });
    }
    out
}

/// The `i`th of an arbitrary number of distinct `authorized_keys` lines.
///
/// Built by hand from the ed25519 wire encoding (`string "ssh-ed25519"` then
/// `string <32 bytes>`) because the point of the fixture is *a table with N
/// rows in it*, and generating N real keypairs would spend the test's time on
/// elliptic curves instead.
fn synthetic_key(i: usize) -> String {
    let mut blob = Vec::new();
    blob.extend_from_slice(&11u32.to_be_bytes());
    blob.extend_from_slice(b"ssh-ed25519");
    blob.extend_from_slice(&32u32.to_be_bytes());
    let mut material = [0u8; 32];
    // A cheap deterministic spread, so two indices never collide.
    let mut h: u64 = 0xcbf2_9ce4_8422_2325 ^ (i as u64);
    for slot in material.chunks_mut(8) {
        h = h.wrapping_mul(0x0100_0000_01b3).rotate_left(13) ^ (i as u64);
        slot.copy_from_slice(&h.to_be_bytes()[..slot.len()]);
    }
    blob.extend_from_slice(&material);
    format!("ssh-ed25519 {} synthetic-{i}", b64(&blob))
}

/// A table with `n` principals, one key each, and one repository-scoped grant
/// each, plus `n` repositories in `bulk`, and one known repository in `ns`
/// that the connection sequence asks about.
fn populate(fixture: &Counted, n: usize) -> (Fingerprint, RepoRef) {
    fixture.catalog.create_namespace("ns").expect("namespace");
    fixture.catalog.create_namespace("bulk").expect("namespace");
    let target = RepoRef::new("ns", "app").expect("repo ref");
    fixture
        .catalog
        .create(NewRepo::new(target.clone()).visibility(Visibility::Private))
        .expect("create the target repository");

    for i in 0..n {
        let id = format!("p{i:06}");
        fixture
            .auth
            .add_principal(Principal {
                id: id.clone(),
                name: format!("principal {i}"),
                kind: PrincipalKind::User,
                created: 0,
            })
            .expect("add principal");
        fixture
            .auth
            .register_key(&id, &synthetic_key(i), None)
            .expect("register key");
        fixture
            .auth
            .grant(Grant {
                principal: id.clone(),
                scope: Scope::Repo(RepoRef::new("ns", "app").expect("repo ref")),
                level: Level::Write,
            })
            .expect("grant");
        fixture
            .catalog
            .create(NewRepo::new(
                RepoRef::new("bulk", &format!("r{i:06}")).expect("repo ref"),
            ))
            .expect("create repository");
    }

    // The principal the sequence authenticates as is the *first* one, so a
    // linear scan cannot get lucky by finding it early. Index 0 is the best
    // case for a scan, which is what makes an unchanged number damning.
    let fingerprint = Fingerprint::of_openssh(&synthetic_key(0)).expect("fingerprint");
    (fingerprint, target)
}

/// Exactly what one SSH connection asks the control plane, in order: who is
/// this key, does that repository exist, and what level do they hold on it
/// (`Control::access_for` walks `[Admin, Write, Read]`).
fn connection_sequence(auth: &AuthTable, catalog: &RepoCatalog, fp: &Fingerprint, path: &RepoRef) {
    let principal = auth
        .principal_by_fingerprint(fp)
        .expect("identity lookup")
        .expect("the fixture registered this key");
    let repo = catalog
        .get(path)
        .expect("catalog lookup")
        .expect("the fixture created this repository");
    let mut level = None;
    for want in [Level::Admin, Level::Write, Level::Read] {
        if auth.may(&principal.id, &repo, want).expect("acl") {
            level = Some(want);
            break;
        }
    }
    assert_eq!(
        level,
        Some(Level::Write),
        "the fixture grants Write; a sequence that answers anything else measured the wrong thing"
    );
}

// ───────────────────────────── 1. the warm path ──────────────────────────────

/// A connection whose control documents have not changed
/// since the last one must not read the object store at all, and that must
/// hold at ten principals and at a thousand.
///
/// Both halves matter. `object_reads == 0` is the claim; `ref_reads > 0` is the
/// proof that the sequence still consults storage rather than answering from a
/// frozen snapshot, which is the identity-value trap this whole file is shaped
/// against. Cache coherence itself is a separate test below.
#[test]
fn a_warm_connection_reads_no_objects_at_any_table_size() {
    let mut sizes = Vec::new();
    for n in [10usize, 1000] {
        let fixture = Counted::new();
        let (fp, path) = populate(&fixture, n);

        // Warm: the first sequence is allowed to parse.
        connection_sequence(&fixture.auth, &fixture.catalog, &fp, &path);
        fixture.counts.reset();
        connection_sequence(&fixture.auth, &fixture.catalog, &fp, &path);

        sizes.push((
            n,
            fixture.counts.object_reads(),
            fixture.counts.object_read_bytes(),
            fixture.counts.ref_reads(),
        ));
    }

    for (n, reads, bytes, refs) in &sizes {
        assert!(
            *refs > 0,
            "at N={n} the sequence read no references at all. It is not consulting storage, so \
             a zero object count proves nothing"
        );
        assert_eq!(
            *reads, 0,
            "at N={n} one warm connection still read {reads} objects ({bytes} bytes) from the \
             control documents"
        );
    }
    assert_eq!(
        sizes[0].3, sizes[1].3,
        "the number of reference reads per connection changed between N=10 and N=1000: {sizes:?}"
    );
}

/// The cache is keyed on the reference target, so a write by anyone, this
/// process or another, must be seen by the next read.
///
/// Without this, the test above is satisfied by a cache that never invalidates,
/// which is not a cache but a bug.
#[test]
fn a_warm_handle_sees_another_writers_change() {
    let fixture = Counted::new();
    let (_, _) = populate(&fixture, 4);
    let (other_auth, other_catalog) = fixture.cold_handles();

    // Warm both handles.
    assert!(fixture.auth.principal("p000001").expect("read").is_some());
    assert!(other_auth.principal("p000001").expect("read").is_some());
    assert!(fixture
        .catalog
        .contains(&RepoRef::new("ns", "app").unwrap())
        .unwrap());
    assert!(other_catalog
        .contains(&RepoRef::new("ns", "app").unwrap())
        .unwrap());

    // A write through one handle.
    other_auth
        .add_principal(Principal {
            id: "latecomer".to_owned(),
            name: "arrived after the cache was warm".to_owned(),
            kind: PrincipalKind::Machine,
            created: 0,
        })
        .expect("add principal");
    other_catalog
        .create(NewRepo::new(RepoRef::new("ns", "late").unwrap()))
        .expect("create repository");

    // Must be visible through the other.
    assert!(
        fixture.auth.principal("latecomer").expect("read").is_some(),
        "a warm identity table did not see a principal another handle added"
    );
    assert!(
        fixture
            .catalog
            .contains(&RepoRef::new("ns", "late").unwrap())
            .expect("read"),
        "a warm catalog did not see a repository another handle created"
    );

    // And a removal, which a cache keyed on anything but the target could miss.
    other_auth.remove_principal("latecomer").expect("remove");
    assert!(
        fixture.auth.principal("latecomer").expect("read").is_none(),
        "a warm identity table still reports a principal that was removed"
    );
}

// ───────────────────────────── 2. the cold path ──────────────────────────────

/// A handle whose cache is cold (a freshly started server,
/// the first connection after a restart) must read a bounded slice of the
/// control tables, not all of them.
///
/// The table grows 100× between the two measurements. Whole-document parsing
/// grows with it; a trie of shard blobs grows by the depth of the trie, which
/// is bounded. The threshold is deliberately loose: the two behaviours are two
/// orders of magnitude apart, so anything in between is still a fix and
/// anything at 100× is still the bug.
#[test]
fn a_cold_connection_reads_a_bounded_slice_of_the_table() {
    let mut bytes = Vec::new();
    for n in [10usize, 1000] {
        let fixture = Counted::new();
        let (fp, path) = populate(&fixture, n);
        let (auth, catalog) = fixture.cold_handles();
        fixture.counts.reset();
        connection_sequence(&auth, &catalog, &fp, &path);
        bytes.push((n, fixture.counts.object_read_bytes()));
    }

    assert!(
        bytes[0].1 > 0,
        "a cold connection read nothing at all; the measurement is vacuous: {bytes:?}"
    );
    let ratio = bytes[1].1 as f64 / bytes[0].1 as f64;
    assert!(
        ratio < 10.0,
        "a 100× larger control plane made one cold connection read {ratio:.1}× more bytes \
         ({bytes:?}), so the lookup is still reading the whole table"
    );
}

/// **The second half.** `may()` scanned every grant on the server because
/// `grants` was a `Vec`. Asking one principal about one repository must not
/// cost the whole grant table.
#[test]
fn an_acl_answer_does_not_cost_the_whole_grant_table() {
    let mut bytes = Vec::new();
    for n in [10usize, 1000] {
        let fixture = Counted::new();
        let (_, path) = populate(&fixture, n);
        let repo = fixture.catalog.open(&path).expect("repo");
        let (auth, _) = fixture.cold_handles();
        fixture.counts.reset();
        assert!(
            auth.may("p000000", &repo, Level::Write).expect("acl"),
            "the fixture grants Write to p000000"
        );
        bytes.push((n, fixture.counts.object_read_bytes()));
    }

    assert!(
        bytes[0].1 > 0,
        "the ACL answer read nothing at all; the measurement is vacuous: {bytes:?}"
    );
    let ratio = bytes[1].1 as f64 / bytes[0].1 as f64;
    assert!(
        ratio < 10.0,
        "a 100× larger grant table made one ACL answer read {ratio:.1}× more bytes ({bytes:?}) \
         and may() is still scanning every grant"
    );
}

// ───────────────────────────── 3. the write path ─────────────────────────────

/// Creating one repository must not rewrite the catalog.
///
/// Measured as *bytes written per create*, which is the quantity that was
/// quadratic: rewriting the whole document makes the average create cost grow
/// in step with the catalog, so a 10× longer run costs ~10× more per create.
/// A sharded document pays the trie plus one leaf, which barely moves.
#[test]
fn creating_a_repository_does_not_rewrite_the_catalog() {
    let mut per_create = Vec::new();
    for n in [100usize, 1000] {
        let fixture = Counted::new();
        fixture.catalog.create_namespace("bulk").expect("namespace");
        fixture.counts.reset();
        for i in 0..n {
            fixture
                .catalog
                .create(NewRepo::new(
                    RepoRef::new("bulk", &format!("r{i:06}")).expect("repo ref"),
                ))
                .expect("create repository");
        }
        per_create.push((n, fixture.counts.object_write_bytes() / n));
    }

    assert!(
        per_create[0].1 > 0,
        "creating repositories wrote nothing; the measurement is vacuous: {per_create:?}"
    );
    let ratio = per_create[1].1 as f64 / per_create[0].1 as f64;
    assert!(
        ratio < 3.0,
        "creating a repository into a 10× larger catalog cost {ratio:.1}× more bytes written \
         ({per_create:?} bytes per create), so the whole catalog is still being rewritten"
    );
}

// ───────────────────────────── 4. the CAS survives ───────────────────────────

/// Concurrent writers, run against a sound reference store and against a
/// deliberately broken one **in the same file**, so the assertion is seen to
/// fail before it is believed.
///
/// Returns `(reported success, still in the table)`. Those two numbers, not the
/// number attempted, are what a compare-and-swap promises: an update that was
/// *refused* (`Error::Contended`, eight retries exhausted against seven other
/// writers) is a caller that knows it did not land. An update that returned
/// `Ok` and then vanished is a lost update, and that is what must never happen.
fn concurrent_principal_adds(clobber: bool) -> (usize, usize) {
    const THREADS: usize = 8;
    const EACH: usize = 25;

    let fixture = Counted::with_clobber(clobber);
    let barrier = Arc::new(Barrier::new(THREADS));
    let mut handles = Vec::new();
    for t in 0..THREADS {
        // A separate handle per thread, with its own document cache. Two
        // threads sharing one cache would prove less than two processes do.
        let auth = AuthTable::new(Arc::clone(&fixture.store)).expect("auth table");
        let barrier = Arc::clone(&barrier);
        handles.push(std::thread::spawn(move || {
            let mut acknowledged = Vec::new();
            barrier.wait();
            for i in 0..EACH {
                let id = format!("t{t}n{i:03}");
                if auth
                    .add_principal(Principal {
                        id: id.clone(),
                        name: id.clone(),
                        kind: PrincipalKind::User,
                        created: 0,
                    })
                    .is_ok()
                {
                    acknowledged.push(id);
                }
            }
            acknowledged
        }));
    }
    let acknowledged: Vec<String> = handles
        .into_iter()
        .flat_map(|handle| handle.join().expect("writer thread"))
        .collect();

    let survivors: std::collections::BTreeSet<String> = fixture
        .auth
        .principals()
        .expect("read the table back")
        .into_iter()
        .map(|p| p.id)
        .collect();
    let kept = acknowledged
        .iter()
        .filter(|id| survivors.contains(*id))
        .count();
    (acknowledged.len(), kept)
}

/// **The guard, seen red.** Against a reference store that ignores the expected
/// old target (the exact defect a compare-and-swap exists to catch), writers
/// overwrite each other and acknowledged additions disappear.
#[test]
fn a_broken_compare_and_swap_loses_acknowledged_updates() {
    let (acknowledged, kept) = concurrent_principal_adds(true);
    assert!(
        kept < acknowledged,
        "a reference store that ignores the CAS expectation kept all {acknowledged} \
         acknowledged writes; the test below cannot be trusted, because it would pass on a \
         broken store"
    );
}

/// **The property.** Eight threads, twenty-five principals each, one document:
/// every write that reported success is still there.
#[test]
fn concurrent_writers_do_not_lose_an_update() {
    let (acknowledged, kept) = concurrent_principal_adds(false);
    assert!(
        acknowledged > 0,
        "no writer succeeded at all; the measurement is vacuous"
    );
    assert_eq!(
        kept,
        acknowledged,
        "{} of {acknowledged} acknowledged concurrent additions were lost",
        acknowledged - kept
    );
}

// ───────────────────────────── 5. no silent empties ──────────────────────────

/// A document this gunnar cannot read must be an error, never an empty table.
///
/// The sharded layout changed the stored shape of both control documents. The
/// failure mode that must not exist is the quiet one: an older document parsed
/// as "no principals are registered" is a server that refuses everyone; on the
/// catalog, worse still, one that reports every repository as missing.
#[test]
fn a_document_this_gunnar_cannot_read_is_an_error_not_an_empty_table() {
    let fixture = Counted::new();
    // The pre-shard layout: principals, keys and grants inline in one blob.
    let legacy =
        br#"{"principals":{"a":{"id":"a","name":"a","kind":"User"}},"keys":{},"grants":[]}"#;
    let id = fixture
        .store
        .put_blob(legacy)
        .expect("write the legacy document");
    let current = fixture
        .store
        .read_doc(gunnar_core::AUTH_REF)
        .expect("read the document");
    assert!(fixture
        .store
        .commit_doc(gunnar_core::AUTH_REF, current, id)
        .expect("point the document at it"));

    let (auth, _) = fixture.cold_handles();
    let err = auth
        .principals()
        .expect_err("a document in a layout this gunnar does not write must not read as empty");
    let text = err.to_string();
    assert!(
        text.contains(gunnar_core::AUTH_REF),
        "the error must name the document it could not read, got: {text}"
    );
}