crates/gunnar-server/src/transport.rs
//! **The join.** `gunnar-ssh`'s [`GitTransport`] seam, implemented over
//! `gunnar-wire`'s state machines, with `gunnar-wire`'s [`PushGate`] seam
//! implemented over `gunnar-policy`'s engine.
//!
//! Until this file existed, those three crates had each been proven alone:
//! `gunnar-wire` against stock `git` over pipes, `gunnar-ssh` against a
//! `BannerTransport`, `gunnar-policy` against crafted push fixtures. Nothing
//! had joined them, so nothing had proved a real `git clone` over a real SSH
//! connection. `tests/ssh_end_to_end.rs` does.
//!
//! # Async on one side, blocking on the other
//!
//! russh hands over a `tokio` duplex stream; `gunnar-wire` wants
//! `&mut dyn BufRead` and `&mut dyn Write`, deliberately, so it can be driven
//! by a pipe or a subprocess in tests (the git wire protocol). The stream is therefore
//! [`tokio::io::split`] into halves — one object cannot be borrowed mutably as
//! both reader and writer — each wrapped in a [`SyncIoBridge`], and the whole
//! transfer runs inside [`tokio::task::spawn_blocking`]. `SyncIoBridge` blocks
//! on the runtime, so running it on a runtime worker would deadlock the reactor
//! that has to feed it.
//!
//! # The protocol version is the client's to choose
//!
//! A v2 client announces itself by sending `GIT_PROTOCOL=version=2` as an SSH
//! **environment request**. `gunnar-ssh` accepts that one variable and no
//! other (`gunnar_ssh::server::ENV_ALLOWLIST`), and carries the raw value
//! on [`ServiceRequest::git_protocol`]. [`protocol_version`] turns it
//! into a [`ProtocolVersion`] with `gunnar-wire`'s own parser, so there is one
//! implementation of the grammar rather than two.
//!
//! A client that sends nothing gets **v0**, which is what `ProtocolVersion`'s
//! default already means and what `gunnar-wire`'s v0 fallback exists for. That
//! path is not vestigial: it is what an old client, or one behind a transport
//! that drops the variable, actually gets, and `tests/ssh_end_to_end.rs`
//! exercises it explicitly rather than assuming it.
//!
//! **What this unblocked, measured 2026-08-03.** While every session was v0,
//! `gunnar clone` could not clone from `gunnar serve` at all:
//!
//! ```text
//! gunnar clone ssh://git@127.0.0.1:2299/team/project.git
//! → Could not decode server reply: Currently we require feature
//! "multi_ack_detailed", which is not supported by the server
//! ```
//!
//! `gunnar-client` fetches through `nornir-git` → `gix`'s blocking transport,
//! which demands `multi_ack_detailed` in v0. `gunnar_wire::caps` used not to
//! advertise it, on the grounds that gunnar negotiated in single-ACK mode and
//! advertising more than is implemented is a bug. **It advertises and
//! implements it now** (smart HTTP): single-ACK cannot complete a stateless-rpc
//! negotiation, so smart HTTP forced the mode to be built properly rather than
//! declined. Delivering `GIT_PROTOCOL` still matters — v2 is the better
//! protocol and `gix` asks for it by itself — but the v0 fallback is no longer
//! a capability gunnar refuses `gix` on paper. That fallback is not exercised
//! here, so treat this paragraph as a statement about the advertisement, which
//! `gunnar-wire/tests/advertisement.rs` asserts, and not as a measurement of
//! `gix`'s v0 client.
//!
//! `receive-pack` is **not** given the negotiated version, and that is not an
//! oversight: protocol v2 has no push side. `git` sends `GIT_PROTOCOL` on a
//! push too and then speaks v0 anyway, so passing it on would be inventing a
//! protocol neither end implements.
//!
//! # This module is the event-stream producer, and the attestation one
//!
//! `Events.Watch` had a bus, a filter, a gap marker and its own tests. For
//! `PushAttestation` it had no producer at all. A stream with nothing behind it
//! connects, delivers nothing and is indistinguishable from a quiet server.
//! [`serve_receive_pack`] is where a push becomes events:
//!
//! ```text
//! refs applied ── status reported ──┬── RefUpdated / RefDeleted / PushRejected
//! └── PushAttestation ──┬── the event bus
//! └── attestations.jsonl
//! ```
//!
//! Everything is published **after** the report is on the wire. A push must
//! never fail because nobody was watching, and it must never be *reported* as
//! having done something the client has not yet been told about.
//!
//! ## Why there is no publish callback in `gunnar-wire`
//!
//! There does not need to be one. `ReceivePack::serve` already returns a
//! [`PushOutcome`] carrying every command, its status and the indexed pack,
//! which is the input an event needs. A `Publisher` trait threaded
//! through `gunnar-wire` would be a second path to the same facts, live at a
//! point where the report has *not* yet been written, and a second thing to
//! keep in step (one implementation). The coupling `gunnar-wire` must not grow is a
//! dependency on `gunnar-grpc`, and returning a value cannot create one.
//!
//! ## The dependency edge push attestation forbids
//!
//! gunnar does not link the Build Thing, nornir, skade, arrow or parquet for
//! any of this. The event is a proto message and the log is a file of JSON
//! lines; anything wanting a warehouse row subscribes and writes it itself.
use std::collections::HashMap;
use std::future::Future;
use std::io::{BufRead, BufReader, BufWriter, Write as _};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use gix_hash::ObjectId;
use gunnar_core::observe::{ServerMetrics, TARGET_GIT, TARGET_POLICY, TARGET_STORE};
use gunnar_grpc::EventBus;
use gunnar_policy::{
KeyDirectory, PolicyConfig, PolicyEngine, PushContext, RefUpdate, RepoId, StaticKeyDirectory,
};
use gunnar_proto::{event, PushAttestation, PushRejected, RefDeleted, RefUpdated, RepoPath};
use gunnar_ssh::seam::{Delivery, DuplexStream, GitTransport, ServiceRequest, TransportError};
use gunnar_ssh::GitService;
use gunnar_store::git_storage_trait::GitServe;
use gunnar_wire::{
PackReceipt, ProtocolVersion, PushFacts, PushGate, PushOutcome, ReceivePack, RefCommand,
RefStatus, Session, UploadPack,
};
use tokio_util::io::SyncIoBridge;
use crate::attest::AttestationLog;
use crate::storage::OpenRepo;
use crate::stores::ServerStores;
/// The per-repository policy document, read from the repository's git dir.
///
/// `gunnar-core` has no storage for branch protection or ref-name policy (see
/// the report in the crate docs), so the gate configuration comes from a file
/// beside the repository until it does. When the catalog grows a policy
/// document, this constant is the only thing that changes.
pub const POLICY_FILE: &str = "gunnar-policy.json";
/// Which protocol version to answer a request in.
///
/// The client chooses, by sending `GIT_PROTOCOL` as an SSH environment request;
/// `gunnar-ssh` allowlists that one name and carries the raw value here. The
/// grammar is parsed **once**, by `gunnar-wire`, so a v1 announcement cannot
/// mean one thing to the SSH layer and another to the wire (one implementation).
///
/// A request that carries nothing is v0, and not by a default that happens to
/// land on v0: that is what `from_git_protocol(None)` says.
fn protocol_version(request: &ServiceRequest) -> ProtocolVersion {
ProtocolVersion::from_git_protocol(request.git_protocol.as_deref())
}
/// **The push's standing in its namespace's byte ledger, as the receive loop
/// asks about it.**
///
/// Four lines, and they are the whole of the third seam this file holds. The
/// other two are [`protocol_version`] and [`wire_session`], and this one exists
/// for the same reason they do: `gunnar-ssh` and `gunnar-wire` deliberately do
/// not depend on each other, and this crate is the only one that holds both.
///
/// The bug it closes is a time-of-check/time-of-use on a quota. Every push was
/// handed `PlanGate::pack_ceiling`'s number — the namespace's remaining room
/// plus slack — and held to the whole of it, so N pushes that arrived together
/// were each admitted against the same measurement and the room was spent N
/// times. `gunnar_lfs::Mark` makes the number a budget the namespace shares:
/// the receive loop charges every chunk before it takes it, against everything
/// admitted since this push measured.
///
/// # What happened to the slack
///
/// Nothing, and that is deliberate. [`crate::entitlement`] still computes
/// `remaining + max(cap / 16, 64 MiB)` and this file still passes that number
/// through untouched; the ledger adds no arithmetic of its own and never
/// re-adds the slack per connection. What changed is only WHO spends it: the
/// 6.25 % (floor 64 MiB) that covers the difference between a thin pack's wire
/// bytes and its stored bytes is now one allowance for the namespace, charged
/// once, instead of one allowance per simultaneous push. That is the direction
/// the slack was argued for in the first place — it exists so the push that is
/// landing is not cut by arithmetic nobody can do in advance, not so that eight
/// pushes may each have one.
///
/// # Why an adapter rather than the trait on `Mark` itself
///
/// Because `gunnar-lfs` would then have to name `gunnar_wire::PackLedger`, and
/// `gunnar-lfs` is a leaf (44 packages) that `gunnar-lfs-serve` ships on its
/// own; `gunnar-wire` cannot name `gunnar_lfs::Mark` either, being bound by the
/// no-OS rule and the dependency budget to plumbing. An `Arc<dyn PackLedger>`
/// built here costs one allocation per push, adds no edge to either crate's
/// graph, and puts the conversion where every other conversion between those two
/// already is.
#[derive(Debug)]
pub(crate) struct PackMark {
pub(crate) mark: gunnar_lfs::Mark,
}
impl gunnar_wire::PackLedger for PackMark {
/// Pure forwarding, on purpose: `room` is the receive loop's number (the
/// ceiling plus the one byte that makes it reachable) and the arithmetic is
/// the ledger's. Anything computed here would be a third place where a byte
/// ceiling is decided.
fn admit_upto(&self, want: u64, room: u64) -> u64 {
self.mark.admit_upto(want, room)
}
}
/// How much of a conversation this call carries, as `gunnar-wire` names it.
///
/// [`Delivery`] is the seam's word and [`Session`] is the wire's, and they are
/// two enums rather than one because `gunnar-ssh` deliberately does not depend
/// on `gunnar-wire` (see `ServiceRequest::git_protocol`). This is the whole of
/// the translation, in one place, so the mapping cannot be made differently in
/// two.
///
/// [`Delivery::Advertisement`] has no `Session` at all: it is not a session,
/// it is a `GET` that writes the advertisement and reads nothing, and it is
/// handled before this is consulted.
fn wire_session(request: &ServiceRequest) -> Session {
match request.delivery {
Delivery::Session | Delivery::Advertisement => Session::Connected,
Delivery::StatelessRpc => Session::StatelessRpc,
}
}
/// **What `gunnar-wire` is configured with, for both directions.**
///
/// `gunnar-wire` has had these knobs since it was written, and a
/// `with_config` on both state machines to receive them. Nothing ever called
/// either: this module said `UploadPack::new(…)` and `ReceivePack::new(…)`, so
/// every deployment ran on `Config::default()` and **an operator could not
/// decline `filter`** — the one capability the type's own doc comment says is
/// "the one capability an operator plausibly wants to decline".
///
/// A configuration surface that cannot affect behaviour is worse than none.
/// It reads as a control that works: an appliance that had to state it does
/// not serve partial clones had `Offered { filter: false }` sitting right
/// there, fully implemented in both protocol versions, and no way at all to
/// reach it. So the knobs are wired rather than deleted, and this struct is
/// the one place the two `gunnar_wire::…::Config` types are held (one implementation) — a
/// second copy of `agent` or `budget` on `ServeConfig` is how a value ends up
/// honoured on one direction and stubbed on the other.
#[derive(Debug, Clone)]
pub struct WireConfig {
/// What `git-upload-pack` serves under: `offered`, `want_policy`, `budget`
/// and `limits`.
pub upload: gunnar_wire::upload_pack::Config,
/// What `git-receive-pack` serves under: the connectivity budget.
pub receive: gunnar_wire::receive_pack::Config,
/// What `git-lfs-transfer` serves under.
///
/// Held here and not beside the store for the same reason the two above
/// are: the LFS session is a third service on this one listener, and a
/// second home for its quota is how 8 GiB comes to be enforced by the SSH
/// path and not by the next caller. `gunnar_lfs::Session::with_limits` and
/// the three builders behind it had **zero** call sites outside
/// `tests/quota.rs` — every deployment ran on `Limits::default()` and the
/// numbers in the doc comments were not reachable from any command line.
pub lfs: gunnar_lfs::Limits,
/// Enumerate the objects each push introduced, so the push attestation can
/// name them. **On by default.**
///
/// The ids are not a by-product of storing a pack — a pack entry does not
/// carry its oid — so answering this waits for the engine's indexer to
/// drain, which is real latency on the push path
/// (`storage::znippy_engine::ingest` has the measured multiples). Before
/// this flag existed the enumeration happened only when a policy gate
/// needed it, and an attestation on a repository with no gates therefore
/// carried an empty object list that was indistinguishable from a push
/// that introduced nothing.
///
/// Turning it **off** does not restore that silence: the attestation then
/// records the reason the set is absent (`new_object_oids_unknown`), because an
/// absence that renders as a value is the defect this flag was added to
/// remove, not the behaviour it restores.
pub attest_objects: bool,
}
impl Default for WireConfig {
/// Hand-written for one field. `#[derive(Default)]` would make
/// [`WireConfig::attest_objects`] `false`, and the safe default here is the
/// one that produces a complete attestation — the cheap default is the one
/// an operator opts into, never the one they get by forgetting.
fn default() -> Self {
WireConfig {
upload: Default::default(),
receive: Default::default(),
lfs: Default::default(),
attest_objects: true,
}
}
}
impl WireConfig {
/// The knobs an operator set on `gunnar serve`, translated once.
///
/// `None` on every flag means "the operator did not say", which takes
/// `gunnar-wire`'s default — the same convention `--max-concurrent-transfers`
/// already uses. `0` is a *stated* value and is treated as one:
///
/// * on the two byte ceilings it is the documented way to remove them,
/// because `Limits`' fields are `Option` and `None` means unbounded;
/// * on `--fetch-budget` it is **refused**, because the walk stops when it
/// reaches the budget, so `0` would mean a server that answers no fetch
/// at all while reading as "no limit". An unbounded-looking flag that
/// silently bricks the server is the failure this whole item is about.
pub fn from_serve_args(args: &gunnar_common::cli::ServeArgs) -> anyhow::Result<Self> {
let mut wire = WireConfig::default();
if args.no_attest_objects {
wire.attest_objects = false;
}
if args.no_filter {
wire.upload.offered.filter = false;
}
if let Some(policy) = args.want_policy {
wire.upload.want_policy = policy.to_wire();
}
if let Some(budget) = args.fetch_budget {
anyhow::ensure!(
budget > 0,
"--fetch-budget 0 would refuse every fetch rather than allow every fetch: the \
walk stops when it REACHES the budget. Leave the flag off for the default of \
{}, or give the number of objects a fetch may visit.",
gunnar_wire::DEFAULT_BUDGET
);
wire.upload.budget = Some(budget);
}
if let Some(bytes) = args.max_fetch_request_bytes {
wire.upload.limits.request_bytes = (bytes > 0).then_some(bytes);
}
if let Some(bytes) = args.max_fetch_object_bytes {
wire.upload.limits.object_bytes = (bytes > 0).then_some(bytes);
}
if let Some(bytes) = args.max_fetch_walk_bytes {
wire.upload.limits.walk_bytes = (bytes > 0).then_some(bytes);
}
// ── the ceiling that bounds the BOX, not the request ─────────────────
//
// Built here rather than defaulted in `gunnar-wire`, because a pool is
// only a pool if every connection gets the same `Arc`. A default inside
// `Config` would give each session a private purse of its own, which
// reads exactly like a server-wide ceiling and is not one.
//
// `for_this_machine` reads `MemTotal`. The alternative — a constant —
// would be a memory ceiling chosen without knowing how much memory
// there is, which is the mistake `DEFAULT_BUDGET` made in objects.
wire.upload.walk_pool = match args.max_walk_bytes_total {
Some(0) => None,
Some(bytes) => Some(gunnar_wire::WalkPool::new(bytes)),
None => Some(gunnar_wire::WalkPool::for_this_machine()),
};
// The LFS three are `u64`/`usize` and not `Option`, so `0` is a stated
// "refuse everything" rather than the "unbounded" the two flags above
// mean by it. Same call as `--fetch-budget`: refuse it at startup,
// because a flag that reads as "no limit" and bricks the service is the
// failure this whole item is about.
if let Some(bytes) = args.max_lfs_object_bytes {
anyhow::ensure!(
bytes > 0,
"--max-lfs-object-bytes 0 would refuse every upload, not allow every upload. \
Leave the flag off for the default of {} bytes.",
gunnar_lfs::DEFAULT_MAX_OBJECT_BYTES
);
wire.lfs = wire.lfs.max_object_bytes(bytes);
}
if let Some(objects) = args.max_lfs_batch_objects {
anyhow::ensure!(
objects > 0,
"--max-lfs-batch-objects 0 would refuse every batch. Leave the flag off for the \
default of {}.",
gunnar_lfs::DEFAULT_MAX_BATCH_OBJECTS
);
wire.lfs = wire.lfs.max_batch_objects(objects);
}
if let Some(bytes) = args.max_lfs_drain_bytes {
anyhow::ensure!(
bytes > 0,
"--max-lfs-drain-bytes 0 would drop the channel on the first refusal instead of \
draining to the client's flush-pkt, so the refusal itself would not be \
delivered. Leave the flag off for the default."
);
wire.lfs = wire.lfs.max_drain_bytes(bytes);
}
Ok(wire)
}
/// **The one place an LFS session is built.**
///
/// By construction rather than by a guard that watches two call sites
/// agree (one implementation). The defect this closes was exactly a second construction
/// path: `Session::new(…).serve(…)` in the SSH transport, with
/// `with_limits` reachable only from `tests/quota.rs`. A test that built
/// its own session could be green with the server on `Limits::default()`
/// forever, so the test drives *this* and there is nothing else to drive.
pub fn lfs_session<'a>(
&self,
store: &'a dyn gunnar_lfs::LfsStore,
operation: gunnar_lfs::Operation,
) -> gunnar_lfs::Session<'a> {
gunnar_lfs::Session::new(store, operation).with_limits(self.lfs)
}
}
/// `gunnar-wire`, wired to the SSH listener.
pub struct WireTransport {
stores: Arc<ServerStores>,
bus: Option<Arc<EventBus>>,
attestations: AttestationLog,
metrics: Arc<ServerMetrics>,
wire: Arc<WireConfig>,
}
impl WireTransport {
/// Serve repositories out of `stores`.
pub fn new(stores: Arc<ServerStores>) -> Self {
let attestations = AttestationLog::under(stores.root());
WireTransport {
stores,
bus: None,
attestations,
metrics: Arc::new(ServerMetrics::new()),
wire: Arc::new(WireConfig::default()),
}
}
/// Serve under `wire` rather than `gunnar-wire`'s defaults.
///
/// The whole of what makes [`WireConfig`] reachable. Without a call to
/// this, every knob on it is a comment.
pub fn with_wire_config(mut self, wire: WireConfig) -> Self {
self.wire = Arc::new(wire);
self
}
/// Count into the **server's** counters rather than into a private set.
///
/// `gunnar serve` mints one [`ServerMetrics`] and hands the same `Arc` here
/// and to the SSH listener, so `connections_refused` and `pushes_rejected`
/// are two readings of one server. A transport built without this still
/// counts, into a set only it can see, which is what a unit test wants.
pub fn with_metrics(mut self, metrics: Arc<ServerMetrics>) -> Self {
self.metrics = metrics;
self
}
/// Publish `RefUpdated` / `RefDeleted` / `PushRejected` / `PushAttestation`
/// to the event bus.
///
/// **The bus is optional; the attestation log is not.** A server started
/// with `--no-grpc` has nobody to notify, but the attestation's durable record is not a
/// notification. It is the thing the event stream explicitly does not
/// promise to keep (see `EventBus::publish`), so the log is written either
/// way and only the publish is conditional.
pub fn with_bus(mut self, bus: Arc<EventBus>) -> Self {
self.bus = Some(bus);
self
}
/// Where the attestation's JSON lines land.
pub fn attestation_log(&self) -> &AttestationLog {
&self.attestations
}
}
/// **The LFS object store, opened the one way** — for the SSH session and for
/// the HTTP batch API alike (one implementation).
///
/// The two surfaces used to be one call apart from having two answers to *where
/// does this repository keep its LFS objects*: `serve_blocking` reached for
/// `stores.repo_store(...).git_dir()` and anything else would have had to
/// re-derive it. It is `<git_dir>/lfs/`, the symlink check in
/// `ServerStores::repo_dir` is not optional on either path, and the handle the
/// store lock lives on is the same handle, so a repack and an LFS transfer
/// exclude each other the same way on both transports.
impl gunnar_ssh::seam::LfsTransport for WireTransport {
fn open_lfs(
&self,
request: &ServiceRequest,
) -> Result<gunnar_ssh::seam::LfsAccess, TransportError> {
// An upload into a repository that does not exist yet (the
// first-repository door, 2026-09-28): the repository is made by its
// first object, see `ProspectiveLfs`.
if request.prospective {
return Ok(gunnar_ssh::seam::LfsAccess {
store: Arc::new(ProspectiveLfs::new(
Arc::clone(&self.stores),
request.clone(),
)),
limits: self.wire.lfs,
allowance: request.lfs_allowance.clone(),
});
}
let store = open_repo_store(&self.stores, request)?;
let lfs = open_lfs_store(&self.stores, &store, request)?;
Ok(gunnar_ssh::seam::LfsAccess {
store: Arc::new(lfs),
// **`self.wire.lfs`, not `Limits::default()`.** The one place an
// operator's `--max-lfs-object-bytes` is held. A separate default
// here is exactly how a quota comes to be enforced on SSH and not
// on HTTP; see `WireConfig::lfs`.
limits: self.wire.lfs,
// **The tenant's plan**, decided by the gate chain in
// `authorise_service` and carried on the request beside the pack
// ceiling. Cloned rather than moved: `open_lfs` takes the request by
// reference, and one `Allowance` is two small fields.
allowance: request.lfs_allowance.clone(),
})
}
}
/// Open the repository behind an **already authorised** request, or say why not
/// in the sentence the client is owed.
///
/// Shared by `serve_blocking` and [`LfsTransport::open_lfs`]. The distinction
/// between the two failures is the load-bearing part and is why this is one
/// function: a held store lock is a running `gunnar gc` (or a second gunnar)
/// and clears on its own, so the answer is "retry"; every other open failure is
/// a fault an operator has to look at. Answering both with one sentence is how
/// a two-minute wait becomes a support ticket.
///
/// `TransportError::Refused` and not `Protocol`: both call sites reach this
/// before a single byte has been written, which is what that variant *means*.
fn open_repo_store(
stores: &ServerStores,
request: &ServiceRequest,
) -> Result<Arc<OpenRepo>, TransportError> {
stores.repo_store(&request.repo).map_err(|err| {
// Opening the repository failed *after* the grant check said yes, so
// this is a storage fault and not a decision. It used to be returned
// silently: the client saw a dead channel and the server said nothing
// at all.
tracing::error!(
target: TARGET_GIT,
error = format!("{err:#}"),
"the repository could not be opened for a transfer that was already authorised"
);
TransportError::Refused(match locked_path(&err) {
Some(_) => format!(
"the repository {} is locked by maintenance on the server (a repack or a \
second gunnar holds its store lock); the transfer was not started, so retry",
request.repo.key()
),
None => format!(
"the repository {} could not be opened on the server; the transfer was not \
started. The reason is in the server log",
request.repo.key()
),
})
})
}
/// The LFS objects under a repository's own git dir. See
/// [`gunnar_lfs::fs_store`].
fn open_lfs_store(
stores: &ServerStores,
store: &OpenRepo,
request: &ServiceRequest,
) -> Result<crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>, TransportError> {
let repo = format!("{}/{}", request.repo.namespace, request.repo.name);
gunnar_lfs::FsLfsStore::open_or_init(store.git_dir())
.map(|fs| {
crate::txlog_lfs::LoggedLfs::new(fs, Arc::clone(stores.txlog()), repo)
// Held to the data set's room, whole object at a time (cov-13).
.with_space(stores.space_ledger())
})
.map_err(|err| {
tracing::error!(
target: TARGET_GIT,
error = format!("{err}"),
"the lfs object store could not be opened for an authorised transfer"
);
// The same `Refused`, for the same reason, at the twin of the seam
// above: this is still before the first byte, and the LFS grammar has
// no per-object error line to carry it later.
TransportError::Refused(format!(
"the lfs object store for {} could not be opened on the server; the \
transfer was not started. The reason is in the server log",
request.repo.key()
))
})
}
/// **An LFS upload into a repository that does not exist yet** (2026-09-28):
/// the first-repository door for git-lfs's pre-push hook, which runs after
/// git took receive-pack's advertisement and BEFORE it sends a command.
/// MEASURED 2026-09-28: the first `git push` with an LFS object into a new
/// repository died in that hook — `batch request: exec request failed on
/// channel 0: exit status 255` — and neither the repository nor the ref was
/// created, while a precreated repository worked.
///
/// `Dispatch::authorise_service` admitted the upload under the SAME
/// push-to-create rules as the push (`Control::push_create_spec`: namespace
/// Admin, and server Admin for a namespace that does not exist; everyone
/// else the fused not-found). Nothing is written until an OBJECT arrives:
/// until then every object is absent (so the batch asks for all of them),
/// and a client that only batches, or hangs up, leaves nothing. The first
/// `begin_write` makes the repository through [`create_prospect`] — the
/// same writes a push-create makes, the grant asked again — or takes the one
/// a concurrent upload or push made first. After that it is the ordinary
/// store.
///
/// The repository is the push-create prospect, sha1, for the reason
/// `push_create_spec` gives; LFS objects are sha256 by their own spec
/// whatever the repository's hash is.
pub(crate) struct ProspectiveLfs<S> {
stores: S,
request: ServiceRequest,
made: std::sync::OnceLock<(
Arc<OpenRepo>,
crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>,
)>,
making: Mutex<()>,
}
impl<S: std::ops::Deref<Target = ServerStores> + Send + Sync> ProspectiveLfs<S> {
pub(crate) fn new(stores: S, request: ServiceRequest) -> Self {
ProspectiveLfs {
stores,
request,
made: std::sync::OnceLock::new(),
making: Mutex::new(()),
}
}
fn made(&self) -> Option<&crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>> {
self.made.get().map(|(_, lfs)| lfs)
}
/// The store, made on first use.
fn made_or_create(
&self,
) -> gunnar_lfs::Result<&crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>> {
if let Some(lfs) = self.made() {
return Ok(lfs);
}
let _one = self.making.lock().unwrap_or_else(|p| p.into_inner());
if let Some(lfs) = self.made() {
return Ok(lfs);
}
let stores: &ServerStores = &self.stores;
let path = self.request.repo.path();
let (repo, store) = create_prospect(stores, &self.request).map_err(|sentence| {
if sentence.starts_with("repository not found") {
gunnar_lfs::Error::NotFound(sentence)
} else {
gunnar_lfs::Error::Store(sentence)
}
})?;
let mut admitted = self.request.clone();
admitted.repo = repo;
admitted.prospective = false;
let lfs = open_lfs_store(stores, &store, &admitted)
.map_err(|err| gunnar_lfs::Error::Store(err.to_string()))?;
tracing::info!(
target: TARGET_GIT,
event = "git.lfs_transfer.created_repo",
repo = %path,
principal = %self.request.principal.id,
"an LFS upload's first object made the repository (the push-create door)"
);
Ok(&self.made.get_or_init(|| (store, lfs)).1)
}
}
impl<S: std::ops::Deref<Target = ServerStores> + Send + Sync> gunnar_lfs::LfsStore
for ProspectiveLfs<S>
{
fn size_of(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<Option<u64>> {
match self.made() {
Some(lfs) => lfs.size_of(oid),
None => Ok(None),
}
}
fn open(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<Option<gunnar_lfs::LfsObject>> {
match self.made() {
Some(lfs) => lfs.open(oid),
None => Ok(None),
}
}
fn begin_write(
&self,
oid: &gunnar_lfs::Oid,
declared_size: u64,
) -> gunnar_lfs::Result<Box<dyn gunnar_lfs::PendingWrite + '_>> {
self.made_or_create()?.begin_write(oid, declared_size)
}
fn delete(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<bool> {
match self.made() {
Some(lfs) => lfs.delete(oid),
None => Ok(false),
}
}
}
impl GitTransport for WireTransport {
fn serve<'a>(
&'a self,
request: ServiceRequest,
stream: Box<dyn DuplexStream>,
) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
let stores = Arc::clone(&self.stores);
let bus = self.bus.clone();
let attestations = self.attestations.clone();
let metrics = Arc::clone(&self.metrics);
let wire = Arc::clone(&self.wire);
Box::pin(async move {
// `spawn_blocking` does NOT carry the current span across (it is a
// thread hand-off, not a task spawn), so without this the whole
// transfer — every line `gunnar-wire`, the gate and this module
// emit — would fall out of `git.transfer` and lose the connection
// id, the principal and the repository. Captured here, where the
// future is polled inside `gunnar-ssh`'s instrumented task.
let parent = tracing::Span::current();
// One object cannot be borrowed mutably as both `BufRead` and
// `Write`, so the duplex is split before it is bridged.
let (read_half, write_half) = tokio::io::split(stream);
// ★ stinky row 1 (2026-10-10): the git work hears that its client
// is gone. `gunnar-ssh` sets the flag when the session's grace runs
// out and then WAITS for this future, so the fan worker is held
// until the blocking task below has actually ended; the guard sets
// it too if this future is dropped some other way, because a
// dropped `JoinHandle` stops nothing.
let cancel = request.cancel.clone();
let _stop_if_dropped = CancelOnDrop(cancel.clone());
let mut reader = Cancellable {
inner: SyncIoBridge::new(read_half),
cancel: cancel.clone(),
};
let mut writer = Cancellable {
inner: SyncIoBridge::new(write_half),
cancel,
};
// The halves are dropped here, at the end of the blocking task, and
// that is now safe: `gunnar-ssh` splits the russh channel instead of
// consuming it, so no part of this stream closes the channel when it
// goes. A 250 ms `linger` used to hold them open to win a race
// against `ChannelCloseOnDrop`; the race is gone and so is the
// timer.
tokio::task::spawn_blocking(move || {
// The blocking thread is this transfer's alone for the whole
// duration, so entering the span here cannot leak it into
// another task's lines.
let _entered = parent.enter();
serve_blocking(
&Serving {
stores: &stores,
bus: bus.as_deref(),
attestations: &attestations,
metrics: &metrics,
wire: &wire,
},
request,
&mut reader,
&mut writer,
)
})
.await
.map_err(|err| {
TransportError::Protocol(format!("the git transport task did not finish: {err}"))
})?
})
}
}
/// Sets the transfer's [`Cancel`](gunnar_ssh::channel::Cancel) when dropped:
/// a transport future dropped before its blocking task ended tells the task.
struct CancelOnDrop(gunnar_ssh::channel::Cancel);
impl Drop for CancelOnDrop {
fn drop(&mut self) {
self.0.cancel();
}
}
/// **The git work's safe points**: every read from and write to the client.
///
/// Once the transfer is told to stop, the next read or write fails with
/// `ConnectionAborted` — not `Interrupted`, which `read_exact`/`write_all`
/// retry — and the work unwinds from there. The ref updates of a receive-pack
/// transaction do no stream I/O between them, so a stop never lands inside
/// one: the transaction lands whole, or (stopped while the pack was still
/// arriving) not at all.
struct Cancellable<T> {
inner: T,
cancel: gunnar_ssh::channel::Cancel,
}
impl<T> Cancellable<T> {
fn check(&self) -> std::io::Result<()> {
if self.cancel.is_cancelled() {
return Err(told_to_stop());
}
Ok(())
}
}
fn told_to_stop() -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::ConnectionAborted,
"the transfer was told to stop: its client's session ended and the transfer outlived the grace",
)
}
impl<T: std::io::Read> std::io::Read for Cancellable<T> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.check()?;
self.inner.read(buf)
}
}
impl<T: std::io::Write> std::io::Write for Cancellable<T> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.check()?;
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.check()?;
self.inner.flush()
}
}
fn wire_error(err: gunnar_wire::Error) -> TransportError {
TransportError::Protocol(err.to_string())
}
/// Counts the bytes that actually cross the wire.
///
/// **Measured, not inferred.** "Objects sent" comes out of the negotiation and
/// says nothing about what a client paid for: a fetch that sends 12 objects can
/// move 4 KiB or 40 MiB. `bytes_out` is the number an operator answering
/// *"why was that clone slow"* reaches for first, and the only place it can be
/// counted honestly is at the stream.
///
/// One relaxed atomic add per `read`/`write` call, and both are already
/// buffered (`BufReader`/`BufWriter` sit on top), so this is a handful of adds
/// per transfer rather than one per byte.
struct Metered<T> {
inner: T,
total: Arc<AtomicU64>,
}
impl<T> Metered<T> {
fn new(inner: T) -> (Metered<T>, Arc<AtomicU64>) {
let total = Arc::new(AtomicU64::new(0));
(
Metered {
inner,
total: Arc::clone(&total),
},
total,
)
}
}
impl<T: std::io::Read> std::io::Read for Metered<T> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
self.total.fetch_add(n as u64, Ordering::Relaxed);
Ok(n)
}
}
impl<T: std::io::Write> std::io::Write for Metered<T> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let n = self.inner.write(buf)?;
self.total.fetch_add(n as u64, Ordering::Relaxed);
Ok(n)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
/// **THE counting point for per-namespace flow, both directions.**
///
/// It is a guard that records when it is DROPPED rather than a call at the end
/// of [`serve_blocking`], and that is the whole design. `serve_blocking` has
/// three service arms, an early return for the abandoned prospect, an early
/// return for the advertisement, and a `?` on every fallible step; a call at
/// "the end" would be five calls, and the sixth exit somebody adds next year
/// would silently stop counting a whole transport. A guard cannot be
/// forgotten, and it counts the failed transfers too — the bytes a client sent
/// before it hung up were still received, and a tenant who is failing is
/// exactly the tenant an operator wants to see.
///
/// It reads the SAME two counters [`Metered`] already keeps for
/// `ServerMetrics`, so a byte is counted once for the box and once for the
/// tenant off one measurement, and the two can never disagree about how much
/// moved. Nothing here counts bytes itself.
///
/// The three `Cell` fields are what the arm that ran decides:
///
/// * `fetched` — an upload-pack SESSION, set after the advertisement's early
/// return, so `git ls-remote` and a monitoring probe are bytes and not a
/// customer's fetch;
/// * `pushed` — a receive-pack session that came back `Ok`;
/// * the LFS pair — the session's own applied figures, which are a different
/// number from the wire bytes about the same transfer.
struct FlowRecording<'a> {
flow: &'a gunnar_core::FlowLedger,
namespace: String,
principal: String,
bytes_in: Arc<AtomicU64>,
bytes_out: Arc<AtomicU64>,
fetched: std::cell::Cell<bool>,
pushed: std::cell::Cell<bool>,
lfs_bytes_in: std::cell::Cell<u64>,
lfs_bytes_out: std::cell::Cell<u64>,
}
impl Drop for FlowRecording<'_> {
fn drop(&mut self) {
self.flow.record(
&self.namespace,
gunnar_core::now_ms(),
&gunnar_core::Transfer {
bytes_in: self.bytes_in.load(Ordering::Relaxed),
bytes_out: self.bytes_out.load(Ordering::Relaxed),
pushed: self.pushed.get(),
fetched: self.fetched.get(),
lfs_bytes_in: self.lfs_bytes_in.get(),
lfs_bytes_out: self.lfs_bytes_out.get(),
principal: self.principal.clone(),
},
);
}
}
/// The lock file, when `err` is a store whose exclusive lock is already held.
///
/// `gunnar gc` opens the repository the same way a transfer does, and the
/// `O_EXCL` lock the Arrow log takes is what makes "one writer" true across
/// processes — so a repack and a transfer on the same repository are mutually
/// exclusive **by that lock and nothing else**. Whichever loses gets
/// [`gunnar_store::Error::StoreLocked`] out of `open_or_init`, wrapped in
/// `anyhow` context by `ServerStores::repo_store`, which is why this walks the
/// chain rather than matching the top-level error.
///
/// Recognising it is not cosmetic: "retry in a moment" and "an operator has to
/// look at this" are opposite instructions, and the caller cannot tell them
/// apart from a formatted string without matching on prose.
fn locked_path(err: &anyhow::Error) -> Option<&std::path::Path> {
match err.downcast_ref::<gunnar_store::Error>() {
Some(gunnar_store::Error::StoreLocked { path }) => Some(path.as_path()),
_ => None,
}
}
/// Everything a transfer needs that belongs to the **server** rather than to
/// the request.
///
/// One borrow instead of five. `serve_blocking` grew an eighth argument when
/// `WireConfig` was wired through and clippy's `too_many_arguments` caught it,
/// which is the right moment to notice that five of those arguments are the
/// same five fields `WireTransport` holds and are passed together everywhere
/// they are passed at all. Bundling them is what the lint is for; an
/// `#[allow]` would have kept the shape that made a sixth easy to add.
struct Serving<'a> {
stores: &'a ServerStores,
bus: Option<&'a EventBus>,
attestations: &'a AttestationLog,
metrics: &'a Arc<ServerMetrics>,
wire: &'a WireConfig,
}
fn serve_blocking(
serving: &Serving<'_>,
request: ServiceRequest,
reader: &mut dyn std::io::Read,
writer: &mut dyn std::io::Write,
) -> Result<(), TransportError> {
// `bus` and `attestations` are not destructured here: they are used only by
// `serve_receive_pack`, which takes the whole `Serving`.
let Serving {
stores,
metrics,
wire,
..
} = *serving;
// `Refused`, not `Protocol`: this is the one point in `serve_blocking` at
// which **nothing has been written** — `Metered` is constructed on the next
// lines and the state machine has not been entered — so the client can still
// be handed a pkt-line, and `TransportError::Refused` is exactly the claim
// "no byte has gone out yet". `open_repo_store` is shared with the HTTP LFS
// seam so the two transports refuse a locked repository in one sentence and
// not two (one implementation).
//
// A PROSPECTIVE request — the first-repository door — has no store to
// open and must not get one here: `repo_store` would create the directory
// for a push the client may abandon at the advertisement. That path goes
// through `admit_prospective_push` below, once the buffers exist.
if request.cancel.is_cancelled() {
return Err(TransportError::Io(told_to_stop()));
}
let store = if request.prospective {
None
} else {
Some(open_repo_store(stores, &request)?)
};
let (metered_in, bytes_in) = Metered::new(reader);
let (metered_out, bytes_out) = Metered::new(writer);
// **The per-namespace flow counting point.** Created here, beside the two
// counters it reads, and recorded when it is dropped — on every exit from
// this function, including the early returns and the failures. See
// `FlowRecording`. It is created BEFORE the first-repository door below, so
// that a client which took an advertisement of a repository it never
// created still has its bytes charged to the namespace it asked about.
let flow = FlowRecording {
flow: stores.flow().as_ref(),
namespace: request.repo.namespace.clone(),
principal: request
.key_fingerprint
.as_ref()
.map(|f| f.to_string())
.unwrap_or_default(),
bytes_in: Arc::clone(&bytes_in),
bytes_out: Arc::clone(&bytes_out),
fetched: std::cell::Cell::new(false),
pushed: std::cell::Cell::new(false),
lfs_bytes_in: std::cell::Cell::new(0),
lfs_bytes_out: std::cell::Cell::new(0),
};
// **`BufWriter::new` is 8 KiB, and a sideband packet is 65 520.** So the
// default capacity did not buffer the pack at all: every packet flushed the
// 5-byte header as a write of its own and then passed 65 515 bytes straight
// through. Two writes per packet, one of them five bytes — and under this
// transport a write is not a syscall, it is a hand-off across a
// `tokio::io::duplex` from a `spawn_blocking` thread, which is a `block_on`,
// a waker and (whenever the reader is parked) a futex pair.
// `gunnar_wire::TRANSPORT_BUFFER_BYTES` carries the arithmetic and the
// write-counting guard.
let mut input = BufReader::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, metered_in);
let mut output = BufWriter::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, metered_out);
let started = std::time::Instant::now();
// ── the first-repository door: advertise, read, check, THEN create ──────
//
// Nothing has been written to the catalog or the disk for a prospective
// request. The client is shown the empty repository it would be pushing
// into; if it hangs up (a sha1 tree that read `object-format=sha256`,
// `git push` with nothing to do) this returns and nothing exists. Only a
// command list that passes the hash check turns into a repository — and
// that create is `ServerStores::create_repo`, the same transactional verb
// every other create uses.
//
// A prospective LFS UPLOAD (git-lfs's pre-push, before receive-pack's
// commands) goes on with no store at all: `ProspectiveLfs` makes the
// repository through the same writes when the first object arrives.
let (request, store, preamble) = match store {
Some(store) => (request, Some(store), None),
None if matches!(request.service, GitService::LfsTransfer(_)) => (request, None, None),
None => match admit_prospective_push(serving, &request, &mut input, &mut output)? {
Some((request, store, preamble)) => (request, Some(store), Some(preamble)),
None => {
output.flush()?;
metrics.add_bytes_out(bytes_out.load(Ordering::Relaxed));
metrics.add_bytes_in(bytes_in.load(Ordering::Relaxed));
tracing::info!(
target: TARGET_GIT,
event = "git.receive_pack.prospect_abandoned",
repo = %request.repo.key(),
elapsed_ms = started.elapsed().as_millis() as u64,
"the client took the advertisement of a repository that does not exist and \
sent no command; nothing was created"
);
return Ok(());
}
},
};
// Only a prospective LFS upload reaches here without a store.
let git_store = || {
store.clone().ok_or_else(|| {
TransportError::Refused(format!(
"repository not found, or you do not have access: {}",
request.repo.path()
))
})
};
match request.service {
GitService::UploadPack => {
let span = tracing::info_span!(target: TARGET_GIT, "git.upload_pack");
let _entered = span.enter();
let store = git_store()?;
let version = protocol_version(&request);
// ── the znippy engine answers reads ─────────────────────────────────
//
// When `GUNNAR_COLD_TIER=znippy` and the repository has an archive,
// upload-pack serves pack ENTRIES out of that archive rather than
// out of `objects/pack` — no unfold, no file. `packed()` is `Some`
// on both arms, so the entry-copy path is the same code; only where
// the bytes come from changes.
//
// receive-pack is deliberately NOT switched: a push lands on `fs`
// and reaches the archive through the append hook. The archive is
// immutable, and `ZnippyObjects::write` says so rather than
// pretending.
//
// There is no second read side to choose between any more. The
// engine *is* the archive on the znippy arm, so `GUNNAR_COLD_TIER`,
// `cold_objects()` and the `Box<dyn ObjectStore>` that carried a
// `None` for "this binary has no znippy engine" have all gone with
// the old znippy engine crate: one store answers reads and writes, and which
// engine it is was decided at startup.
//
// **One handle, not two.** `UploadPack` used to take an object
// store and a ref store, and an advertisement built from one
// snapshot could promise a pack built from another. `GitServe`
// carries both halves, so the ref list and the pack it promises are
// the same store.
let git = store.serve();
// `hash_kind` is a required argument, never a default: it comes
// from the catalog entry through `OpenRepo`, and answering `sha1`
// for a sha256 repository does not error — it decodes garbage.
//
// `with_config`, not `new`: this call is the whole of what makes
// `--no-filter`, `--want-policy`, `--fetch-budget` and the byte
// ceilings anything other than fields nobody reads.
let upload = UploadPack::new(git, store.hash_kind()).with_config(wire.upload.clone());
// `Delivery::Advertisement` is `GET /info/refs`: write the ref list
// (v0) or the capability list (v2) and stop, reading nothing. The
// same call the connected path makes inline, so the two transports
// cannot advertise differently.
if request.delivery == Delivery::Advertisement {
upload.advertise(version, &mut output).map_err(wire_error)?;
output.flush()?;
let out = bytes_out.load(Ordering::Relaxed);
metrics.add_bytes_out(out);
tracing::info!(
target: TARGET_GIT,
event = "git.upload_pack.advertised",
protocol = version.number(),
bytes_out = out,
elapsed_ms = started.elapsed().as_millis() as u64,
"served the upload-pack advertisement"
);
return Ok(());
}
// Past the advertisement's early return above: this is a SESSION,
// which is what `fetches` counts. A bare `git ls-remote` never
// reaches here.
flow.fetched.set(true);
let summary = upload
.serve_session(version, wire_session(&request), &mut input, &mut output)
.map_err(|err| {
tracing::warn!(
target: TARGET_GIT,
event = "git.upload_pack.failed",
elapsed_ms = started.elapsed().as_millis() as u64,
bytes_out = bytes_out.load(Ordering::Relaxed),
%err,
"git-upload-pack failed"
);
wire_error(err)
})?;
output.flush()?;
let out = bytes_out.load(Ordering::Relaxed);
if summary.objects_sent > 0 {
metrics.add_packs_served(1);
}
metrics.add_objects_sent(summary.objects_sent as u64);
metrics.add_bytes_out(out);
metrics.add_bytes_in(bytes_in.load(Ordering::Relaxed));
// The applied outputs, and the duration. `repo` and `principal` are
// NOT repeated: they are fields of `ssh.connection`, which this span
// descends from.
tracing::info!(
target: TARGET_GIT,
event = "git.upload_pack.served",
// What was actually spoken, taken from the session's own
// summary rather than from the request that asked for it; a v2
// request served as v0 would otherwise log as v2.
protocol = summary.protocol.number(),
ls_refs = summary.ls_refs,
rounds = summary.fetch_rounds,
common = summary.common,
objects = summary.objects_sent,
// **WHICH PATH ANSWERED.** `Selection::answered_from_bitmap`
// exists so a guard can tell a fetch served from the store's
// reachability index from one served by walking, and until now
// nothing anywhere read it: not a log line, not a metric, not a
// test. Its only reader in the constellation was an example
// binary.
//
// That blind spot has already cost something concrete. The
// `--fetch-budget` control charged the ids a WALKER holds, and
// on the store the appliance actually runs there is no walker —
// the index answers whole — so the budget was compared against
// nothing while a red test sat in front of three separate call
// sites and each read it as a policy question. A field nobody can
// observe is a behaviour nobody can assert.
from_index = summary.answered_from_bitmap,
bytes_out = out,
elapsed_ms = started.elapsed().as_millis() as u64,
"served git-upload-pack"
);
}
GitService::ReceivePack => {
let span = tracing::info_span!(target: TARGET_GIT, "git.receive_pack");
let _entered = span.enter();
let store = git_store()?;
let outcome =
serve_receive_pack(serving, &request, &store, preamble, &mut input, &mut output);
let out = bytes_out.load(Ordering::Relaxed);
let inn = bytes_in.load(Ordering::Relaxed);
metrics.add_bytes_out(out);
metrics.add_bytes_in(inn);
// A push is a receive-pack session that CAME BACK. The bytes are
// charged either way — they arrived — but a session that failed is
// not a push, and counting it as one would make a tenant whose every
// push is refused look like a tenant who is working.
flow.pushed.set(outcome.is_ok());
if let Err(err) = &outcome {
tracing::warn!(
target: TARGET_GIT,
event = "git.receive_pack.failed",
elapsed_ms = started.elapsed().as_millis() as u64,
bytes_in = inn,
%err,
"git-receive-pack failed"
);
}
outcome?;
}
// LFS over SSH, git-lfs 3.0's pure-SSH transfer. The grant was already decided
// by `gunnar_ssh::command::parse`, from the same key table and the same
// path sandbox the two verbs above went through: `operation` here is the
// one that demanded Read or Write, not a second decision. It is passed
// straight to the session so there is no second authorisation path.
//
// The objects live under the repository's own git dir, in `lfs/`. See
// `gunnar_lfs::fs_store`.
GitService::LfsTransfer(operation) => {
let span = tracing::info_span!(target: TARGET_GIT, "git.lfs_transfer");
let _entered = span.enter();
let lfs: Box<dyn gunnar_lfs::LfsStore + '_> = match &store {
Some(store) => Box::new(open_lfs_store(stores, store, &request)?),
None => Box::new(ProspectiveLfs::new(stores, request.clone())),
};
let mut session = wire.lfs_session(lfs.as_ref(), operation);
// **The tenant's remaining LFS room**, from the gate chain that ran
// in `authorise_service`. `None` on a community build, on an
// uncapped namespace and on every download; see
// `gunnar_ssh::PlanGate::lfs_allowance`.
if let Some(allowance) = request.lfs_allowance.clone() {
session = session.with_allowance(allowance);
}
// `limits()` and `allowance()` read back off the session, not off
// `wire` and not off the request: what is logged is what the session
// will enforce, so a builder that stopped taking would show here
// rather than in a support ticket.
let effective = session.limits();
tracing::debug!(
target: TARGET_GIT,
max_object_bytes = effective.max_object_bytes,
max_batch_objects = effective.max_batch_objects,
max_drain_bytes = effective.max_drain_bytes,
lfs_allowance_bytes = session.allowance().map(|a| a.bytes),
"lfs session limits"
);
let outcome = session.serve(&mut input, &mut output);
output.flush()?;
let out = bytes_out.load(Ordering::Relaxed);
let inn = bytes_in.load(Ordering::Relaxed);
metrics.add_bytes_out(out);
metrics.add_bytes_in(inn);
let summary = outcome.map_err(|err| {
tracing::warn!(
target: TARGET_GIT,
event = "git.lfs_transfer.failed",
operation = %operation,
elapsed_ms = started.elapsed().as_millis() as u64,
%err,
"git-lfs-transfer failed"
);
TransportError::Protocol(err.to_string())
})?;
// The APPLIED figures, from the session's own count. They are a
// different number from the wire bytes above about the same
// transfer — the wire carries framing and the batch conversation —
// so the flow ledger keeps both rather than one as a subset of the
// other.
flow.lfs_bytes_in.set(summary.bytes_received);
flow.lfs_bytes_out.set(summary.bytes_sent);
// Applied output: objects and bytes that really moved, taken from
// the session's own count rather than from what was asked for.
tracing::info!(
target: TARGET_GIT,
event = "git.lfs_transfer.served",
operation = %operation,
commands = summary.commands,
batches = summary.batches,
batch_objects = summary.batch_objects,
objects_sent = summary.objects_sent,
objects_received = summary.objects_received,
lfs_bytes_out = summary.bytes_sent,
lfs_bytes_in = summary.bytes_received,
clean_quit = summary.clean_quit,
bytes_out = out,
bytes_in = inn,
elapsed_ms = started.elapsed().as_millis() as u64,
"served git-lfs-transfer"
);
}
}
Ok(())
}
// Eight, because the counters joined the six things a push already needed.
// Bundling them into a context struct would be one more type whose
// only job is to be unpacked on the first line of this function.
#[allow(clippy::too_many_arguments)]
/// The door's commit. Advertise the prospect, read the client's commands,
/// and — only if there are any and they pass the hash check — write the
/// catalog row (the namespace with it, if the admin's push names a new one)
/// and make the store. `None` means the client sent nothing and nothing was
/// made.
///
/// A refusal after the preamble (a stale store at the name, a grant revoked
/// since the exec, a store the engine cannot make) is answered in git's own
/// protocol — an `ERR` packet, which `git push` renders as `remote error:` —
/// and leaves nothing: the row is retracted, the namespace too if this call
/// conjured it, the directory removed.
fn admit_prospective_push(
serving: &Serving<'_>,
request: &ServiceRequest,
input: &mut dyn BufRead,
output: &mut dyn std::io::Write,
) -> Result<Option<(ServiceRequest, Arc<OpenRepo>, gunnar_wire::Preamble)>, TransportError> {
let Serving { stores, wire, .. } = *serving;
if request.service != GitService::ReceivePack {
// An LFS upload's door is `ProspectiveLfs`, never this one; anything
// else here is a programming error, refused rather than
// served against a repository that does not exist.
return Err(TransportError::Refused(format!(
"repository not found, or you do not have access: {}",
request.repo.path()
)));
}
let hash_kind: gix_hash::Kind = request.repo.hash_kind.into();
if wire_session(request).advertises_inline() || request.delivery == Delivery::Advertisement {
ReceivePack::write_empty_advertisement(hash_kind, &wire.receive, output)
.map_err(wire_error)?;
if request.delivery == Delivery::Advertisement {
return Ok(None);
}
}
let Some(preamble) = ReceivePack::read_preamble(hash_kind, input).map_err(wire_error)? else {
return Ok(None);
};
// The client means it. Now, and only now, the writes — in the order that
// leaves nothing behind if any of them fails.
match create_prospect(stores, request) {
Ok((repo, store)) => {
let mut admitted = request.clone();
admitted.repo = repo;
admitted.prospective = false;
Ok(Some((admitted, store, preamble)))
}
Err(sentence) => {
let _ = gunnar_wire::err_packet(output, &sentence);
let _ = output.flush();
Err(TransportError::Protocol(sentence))
}
}
}
/// **The first-repository door's writes**, for a prospective request whose
/// client has shown it means it — a receive-pack's commands past the hash
/// check, or an LFS upload's first object ([`ProspectiveLfs`]). The stale
/// store check, the catalog row (and the namespace, if the admin's push names
/// a new one) under [`gunnar_ssh::Control::admit_push_create`]'s re-asked
/// grant, then the store — and if the store cannot be made, the row and the
/// namespace are taken back and the directory removed. `Err` is the sentence
/// the client is owed; nothing is left behind by it.
fn create_prospect(
stores: &ServerStores,
request: &ServiceRequest,
) -> Result<(gunnar_core::Repo, Arc<OpenRepo>), String> {
let path = request.repo.path();
let control = stores.ssh_control();
// **Made meanwhile, by this same push** (2026-09-28): git takes
// receive-pack's advertisement (the prospect) BEFORE it runs the pre-push
// hook, and git-lfs's upload in that hook makes the repository in
// between. The stale-store check below would answer the push's own
// commands with an ERR. A repository that now EXISTS is served, as the
// lost race in `admit_push_create` is, after the same re-asked grant,
// and only in the hash the client was shown.
match control.resolve(&path) {
Ok(Some(existing)) => {
if existing.hash_kind != request.repo.hash_kind {
return Err(format!(
"{path} was created meanwhile as a {:?} repository, and this push was shown {:?}; nothing was written — push again",
existing.hash_kind, request.repo.hash_kind
));
}
return match control.push_create_spec(&request.principal, &path) {
Ok(Some(_)) => stores
.repo_store(&existing)
.map(|store| (existing, store))
.map_err(|err| {
format!("the repository {path} could not be opened on the server: {err:#}")
}),
Ok(None) => Err(format!(
"repository not found, or you do not have access: {path}"
)),
Err(err) => Err(format!(
"the gunnar control plane could not read {path}: {err}"
)),
};
}
Ok(None) => {}
Err(err) => {
return Err(format!(
"the gunnar control plane could not read {path}: {err}"
))
}
}
if let Err(err) = stores.refuse_stale_store(&path, false, request.repo.hash_kind) {
return Err(format!("{err:#}"));
}
let (repo, conjured) = match control.admit_push_create(&request.principal, &path) {
Ok(Some(pair)) => pair,
Ok(None) => {
return Err(format!(
"repository not found, or you do not have access: {path}"
))
}
Err(err) => {
return Err(format!(
"the gunnar control plane could not create {path}: {err}"
))
}
};
match stores.repo_store(&repo) {
Ok(store) => Ok((repo, store)),
Err(err) => {
stores.close_repo(&path);
if let Err(undo) = control.retract_push_create(&path, conjured) {
tracing::error!(
target: TARGET_GIT,
repo = %path,
error = %undo,
"retracting the catalog row of a push-create whose store failed"
);
}
if let Ok(dir) = stores.repo_dir(&path) {
let _ = std::fs::remove_dir_all(dir);
}
tracing::error!(
target: TARGET_GIT,
repo = %path,
error = format!("{err:#}"),
"the push-created repository's store could not be made; nothing was kept"
);
Err(format!("the repository {path} could not be created on the server; nothing was kept. The reason is in the server log"))
}
}
}
/// **A push into an encrypted repository on a build that cannot check it in quarantine**
/// (wiring tracer B 2026-10-08): the quarantine is the znippy engine's
/// (`encrypted_quarantine`), and without it a refused plaintext pack would be stored and framed
/// for the twin before the ref gate refused it. `Some(the line)`: refuse the push by name,
/// before a byte is read.
pub(crate) fn unchecked_encrypted_push(encrypted: bool, quarantine: bool) -> Option<String> {
(encrypted && !quarantine).then(|| {
"this repository is encrypted, and this Gunnar was built without the check that keeps \
plaintext out of it, so it takes no push into it: push to a Gunnar built with its \
standard storage engine. Nothing was received"
.to_string()
})
}
fn serve_receive_pack(
serving: &Serving<'_>,
request: &ServiceRequest,
store: &OpenRepo,
preamble: Option<gunnar_wire::Preamble>,
input: &mut dyn BufRead,
output: &mut dyn std::io::Write,
) -> Result<(), TransportError> {
let Serving {
stores,
bus,
attestations,
metrics,
wire,
} = *serving;
let started = std::time::Instant::now();
// A build that cannot check an encrypted repository's pack in quarantine takes none.
if let Some(line) =
unchecked_encrypted_push(request.repo.encrypted, cfg!(feature = "engine-znippy"))
{
return Err(TransportError::Refused(line));
}
// **Everything below this line is a validation read.** A receive-pack
// ingests, indexes and then walks every commit and tree it was sent to
// prove the push closes — once each — and serves nobody. Those reads are
// indistinguishable from a client's inside the store, so the caller is the
// only thing that can say which they are, and this is the caller.
//
// Without it the push fills the engine's lazy read-cache for a reader that
// does not exist: MEASURED 6 778 047 872 B on `torvalds/linux`, 31 % of the
// finished store (measured 2026-09-03).
//
// Thread-scoped and dropped at the end of this function, so a concurrent
// clone on another connection still warms the cache normally.
#[cfg(feature = "engine-znippy")]
let _validation_reads = crate::storage::znippy_engine::validation_reads();
// **One handle.** `GitServe` carries the reading contract and, through its
// `GitOps` supertrait, the eleven — so receive-pack advertises, checks
// connectivity and writes references against a single store rather than
// against an object handle and a ref handle that could disagree.
let git = store.serve();
let hash_kind = store.hash_kind();
let gate = PolicyGate::open(store.git_dir(), request, stores, Arc::clone(metrics));
// Opened before the ingest, and that ordering is load-bearing: whether the
// policy has any gate at all is what decides if the ingest pays for an
// object-set enumeration. It used to be built after, when nothing needed
// to know.
// No `objects/pack` is created. An empty one is what a later reader uses
// to decide a repository is packfile-shaped, and this server never writes
// a packfile into it.
// ── who asks for the object set, and why it is not only the gates ──────
//
// `want_objects: !gate.engine.is_empty()` stood here, and it had **one**
// consumer in mind. The push attestation is a second one, and it was not
// counted: on a repository with no policy gates the ingest skipped the
// enumeration and `attestation_for` then wrote `new_object_count: 0` with
// an empty `new_object_oids` — a push of nothing and a push nobody
// enumerated rendering as the same record. The log line thirty lines below
// this already got that distinction right (`objects_unknown`); the
// attestation, which is the durable artefact and the one a verifier reads,
// did not. `tests/push_events.rs` saw it as *"the pushed commit is missing
// from the attestation's object list: []"*.
//
// So the attestation asks too, and by default it gets an answer. That is
// not free — see `znippy_engine::ingest`'s note on why the flag exists at
// all: the ids are not a by-product of storing a pack, and asking for them
// waits for the indexer to drain. An operator who would rather have the
// latency than the object list turns it off with
// `--no-attest-objects`, and the attestation then records **why** the set
// is absent instead of implying it was empty.
let objects = if gate.engine.reads_new_objects() {
// A gate reads the set. (Not `!is_empty()`: the attribution-trailer
// gate is live on every repository by default and decides from the
// references and the object graph, so it must not make every push
// demand an enumeration — the gix engine refuses `Required` outright.) An engine that cannot produce one must refuse
// rather than let the gate decide on nothing.
crate::storage::ObjectSet::Required
} else if wire.attest_objects {
// Only the attestation wants it, and the attestation can say it did not
// get it. An engine that cannot enumerate still serves the push.
crate::storage::ObjectSet::Wanted
} else {
crate::storage::ObjectSet::NotNeeded
};
let ingest = PushIngest {
git_dir: store.git_dir().to_path_buf(),
hash_kind,
engine: stores.engine(),
objects,
// **A push has nobody watching an object count.** `git push` renders the
// remote's progress band, and what it wants there is bytes; there is no
// job record and no console poll on this path. The mirror is the caller
// that has both, so the mirror is the caller that passes a sink.
watching: None,
// The pushed pack is a transaction (the twin): framed before the engine stores it. For a
// first push `request.repo` is exactly the row `admit_push_create` writes first.
frame: Some(PackFrame::of(stores, &request.repo)),
// An encrypted repository's pack is checked before it is kept.
encrypted: request.repo.encrypted.then_some(git),
// Held to the data set's room, chunk by chunk (cov-13).
space: stores.space_ledger(),
};
if gate.engine.is_empty() {
tracing::info!(
target: TARGET_POLICY,
event = "policy.gates.none",
policy = %store.git_dir().join(POLICY_FILE).display(),
"no policy gates are configured for this repository; every well-formed push is \
accepted. This is an empty policy, not a passing one."
);
} else {
tracing::info!(
target: TARGET_POLICY,
event = "policy.gates.live",
gates = ?gate.engine.gate_names(),
"policy gates live for this push"
);
}
// **receive-pack reads from the store it writes into.**
//
// This used to be `&objects` unconditionally, under a comment saying
// receive-pack was "deliberately NOT switched" because a push landed on
// `fs` and reached the tier through the append hook. That is no longer the
// route: the pack goes straight into the tier, so checking connectivity
// against the filesystem would ask an empty repository whether the objects
// it just stored are present — and the answer is a refused push, which is
// exactly what it did the first time this ran.
//
// A pack the ingest stored but the indexer has not absorbed yet is **not**
// a stale answer here: the per-pack `indexed` bit makes an index miss fall back to the journal
// rather than report absent, so the read side sees a push the moment
// `put_pack` returns, which is before anything asks.
//
// `write_objects` and not `cold_objects`: the latter asks whether there is
// anything to *serve*, and answers `None` for an empty repository — which
// is every repository's first push.
// The engine is the read side; there is nothing else to open.
// No `#[allow(unused_mut)]` any more: it was here because the only
// assignment below was `cfg(metered)`, so a community build had a `mut`
// binding it never wrote. The ceiling is assigned on both builds, so the
// `mut` is now load-bearing on both — and an allow that no longer suppresses
// anything is a suppression nobody can see expire.
let mut receive_config = wire.receive.clone();
// The plan's warning (Grace), decided in `authorise_service` and carried
// on the request: the wire prints it on the progress band beside the
// report, so `git push` shows `remote: plan expired …; renew at …`.
#[cfg(feature = "metered")]
{
receive_config.notice = request.plan_notice.clone();
}
// **The mid-upload ceiling**, decided by the same gate chain in
// `authorise_service` and carried on the request beside the notice. Not
// `cfg(metered)`: the field is on `ServiceRequest` unconditionally and a
// community build simply carries `None`, so there is one path here rather
// than a second one that only the paid build compiles (one implementation). The two
// `PackCeiling`s are the deliberate twin `gunnar_ssh::PackCeiling`
// documents — `gunnar-ssh` does not depend on `gunnar-wire` — and this is
// the one seam that holds both and converts.
//
// The third field is the one this seam could not have been given by either
// side alone: `gunnar-ssh` measured the namespace and marked its ledger,
// `gunnar-wire` counts the bytes, and neither may name the other's type.
// See `PackMark` below.
receive_config.pack_ceiling =
request
.pack_ceiling
.as_ref()
.map(|ceiling| gunnar_wire::PackCeiling {
bytes: ceiling.bytes,
refusal: ceiling.refusal.clone(),
ledger: request
.pack_mark
.clone()
.map(|mark| Arc::new(PackMark { mark }) as Arc<dyn gunnar_wire::PackLedger>),
});
let receive = ReceivePack::new(git, hash_kind).with_config(receive_config);
// `GET /info/refs?service=git-receive-pack`: the ref list and its
// capabilities, then stop. No command list, no pack, no gate — the push
// itself arrives as a separate POST.
if request.delivery == Delivery::Advertisement && preamble.is_none() {
receive
.write_ref_advertisement(output)
.map_err(wire_error)?;
output.flush()?;
tracing::info!(
target: TARGET_GIT,
event = "git.receive_pack.advertised",
elapsed_ms = started.elapsed().as_millis() as u64,
"served the receive-pack advertisement"
);
return Ok(());
}
// A preamble already read (the first-repository door) skips the
// advertisement and the command read: that half happened before the
// repository existed.
// **Heimdall first** (feature `heimdall`, `crate::heimdall_gate`): with the
// gate installed, the pushed pack is parsed in a Heimdall job and only a
// pack whose signed result checks out reaches the ingest, exactly as a
// fetched one does in `crate::mirror_fetch`.
#[cfg(feature = "heimdall")]
let checked = stores
.pack_gate()
.map(|gate| gate.checked_as(&ingest, hash_kind));
#[cfg(feature = "heimdall")]
let ingest: &dyn gunnar_wire::PackIngest = match &checked {
Some(c) => c,
None => &ingest,
};
#[cfg(not(feature = "heimdall"))]
let ingest: &dyn gunnar_wire::PackIngest = &ingest;
let outcome = match preamble {
Some(preamble) => receive
.serve_after_preamble(preamble, input, output, ingest, &gate)
.map(Some),
None => receive.serve_session(wire_session(request), input, output, ingest, &gate),
}
.map_err(wire_error)?;
output.flush()?;
let Some(outcome) = outcome else {
tracing::info!(
target: TARGET_GIT,
event = "git.receive_pack.advertisement_only",
elapsed_ms = started.elapsed().as_millis() as u64,
"git-receive-pack: the client took the advertisement and had nothing to push"
);
return Ok(());
};
// The key this session authenticated with. Carried on the request since
// The authenticating key itself, not recovered from the identity table.
let fingerprint = session_key_fingerprint(request);
// The answerable one. See `session_credential` and GUN-014.
let credential = session_credential(request);
let gates = gate.gate_that_refused();
// **THE counting point for a policy refusal**, per namespace and per push.
//
// Once per PUSH and not once per refused reference, deliberately: a push
// that names four protected branches met one door four times, and a console
// that reported four refusals would make one mistake look like a campaign.
// The refs themselves are already in the `git.ref.rejected` lines below and
// in the `PushRejected` events, which is where per-reference detail belongs.
if !gates.is_empty() {
serving.stores.flow().refuse(
&request.repo.namespace,
gunnar_core::now_ms(),
gunnar_core::Refusal::Policy,
);
}
let mut refs_ok = 0usize;
let mut refs_rejected = 0usize;
for (command, status) in outcome.commands.iter().zip(&outcome.statuses) {
match status {
RefStatus::Ok => {
refs_ok += 1;
if command.is_delete() {
metrics.add_refs_deleted(1);
} else {
metrics.add_refs_updated(1);
}
tracing::info!(
target: TARGET_GIT,
event = if command.is_delete() { "git.ref.deleted" } else { "git.ref.updated" },
ref_name = %command.name,
old = %command.old,
new = %command.new,
"ref updated"
)
}
RefStatus::Rejected(reason) => {
refs_rejected += 1;
metrics.add_refs_rejected(1);
tracing::warn!(
target: TARGET_GIT,
event = "git.ref.rejected",
ref_name = %command.name,
old = %command.old,
new = %command.new,
// The gate that said no, when one did. Empty means nothing
// named a gate: a connectivity failure, a lost
// compare-and-swap, or the collateral refusal every other
// reference gets in an `--atomic` push. The event stream publishes the
// same distinction; the log must not blur it.
gate = gates.get(&command.name.to_string()).map(String::as_str).unwrap_or(""),
%reason,
"ref rejected"
)
}
}
if let Some(bus) = bus {
// Only a reference that actually moved has a descent question to
// answer, so the walk is skipped for everything else.
let forced = status.is_ok() && was_forced(git, command);
publish(
bus,
request,
command,
status,
outcome.atomic,
&fingerprint,
&credential,
&gates,
forced,
);
}
}
// A push that moved a reference is the pusher's operation (2026-10-21,
// the admin's user list). In memory, at most once a minute per person.
if refs_ok > 0 {
stores
.activity()
.operated(&request.principal.id, gunnar_core::now_ms());
}
// ── a repository nobody could clone, made clonable ─────────────────────
//
// After the references have moved, and only for a repository whose `HEAD`
// still resolves to nothing. See
// [`ServerStores::adopt_default_branch_if_unborn`] for why a push is
// allowed to decide this and for how narrow the condition is. A failure
// here is logged and not fatal: the push itself succeeded and its
// references are durable, and refusing it afterwards would be a lie about
// what happened.
let created: Vec<String> = outcome
.commands
.iter()
.zip(&outcome.statuses)
.filter(|(command, status)| status.is_ok() && !command.is_delete())
.map(|(command, _)| command.name.to_string())
.collect();
if !created.is_empty() {
if let Err(err) = stores.adopt_default_branch_if_unborn(&request.repo.path(), &created) {
tracing::warn!(
target: TARGET_GIT,
event = "git.default_branch.adoption_failed",
repo = %request.repo.key(),
error = format!("{err:#}"),
"the push landed, but this repository's HEAD names a branch that does not \
exist and adopting a pushed one failed; it cannot be cloned until \
`set-default-branch` is run against it"
);
}
}
// ── the one line per push an operator reads ────────────────────────────
//
// Applied outputs: how many references actually moved, how many were
// refused, what the pack cost and how long the whole thing took. A push
// summary that said only "done" would be indistinguishable from one that
// moved nothing (measured, not asserted).
//
// `objects` is what the push INTRODUCED, and an ingest that could not work
// that out logs the reason rather than a plausible zero, because a push of
// nothing and a push whose contents could not be enumerated are different
// facts.
let objects_pushed: &[ObjectId] = outcome.new_objects.as_deref().unwrap_or(&[]);
let objects_unknown = outcome.new_objects.as_ref().err().map(String::as_str);
// Two different numbers, both worth having: `wire_bytes` is what the client
// sent and is what a `max_pack_bytes` limit is enforced against;
// `pack_bytes` is what landed, which is larger for a thin push because
// completing it writes bases the client did not send.
//
// Neither may fall back to `0`. `PushFacts::pack_bytes` is `Option` on
// purpose — "`Some(0)` for a deletions-only push … `None` only if the count
// is genuinely unavailable" — and `unwrap_or(0)` threw away exactly the
// distinction the type exists to carry. `stat` failing on a pack whose
// bytes are then reported as `0` is the same disease: an absence that
// renders as a value (and the `objects_unknown` field above).
let (wire_bytes, wire_bytes_unknown) = measured(
outcome.pack_bytes,
"the ingest did not report how many packfile bytes were read off the wire",
);
let (pack_bytes, pack_bytes_unknown) = measured_pack_bytes(&outcome);
if refs_rejected == 0 && outcome.unpack.is_ok() {
metrics.add_pushes_accepted(1);
} else {
metrics.add_pushes_rejected(1);
}
tracing::info!(
target: TARGET_GIT,
event = "git.receive_pack.served",
refs_ok,
refs_rejected,
atomic = outcome.atomic,
unpack_ok = outcome.unpack.is_ok(),
objects = objects_pushed.len(),
objects_unknown,
wire_bytes,
wire_bytes_unknown,
pack_bytes,
pack_bytes_unknown,
push_options = outcome.push_options.len(),
elapsed_ms = started.elapsed().as_millis() as u64,
"served git-receive-pack"
);
// ── the znippy engine took the objects at ingest ───────────────────────────
//
// There is no second step here any more, and its absence is the change.
//
// What used to sit at this line read the `.pack` and `.idx` that
// `index_thin_pack` had written and appended them to an archive — so the
// tier was a COPY of a filesystem write that had already happened, and the
// packfile it copied from stayed in `objects/pack` for ever. That is what
// made every push pay for git's on-disk format even on a server whose
// storage is Arrow. `PushIngest::ingest_cold` now stores the bytes as they
// arrive and no packfile is written at all.
//
// The `refs_ok > 0` guard that used to wrap this is gone with it, and its
// reasoning does not transfer: it existed because archiving a
// fully-rejected push would make the tier the only place that push
// survived. Storage is no longer a place a push can survive *separately* —
// it is where the push went — so a rejected ref list now leaves exactly
// what a rejected ref list leaves on any git server: objects nothing names,
// for `gunnar gc` to collect.
// ── push attestation ───────────────────────────────────────────────────
//
// Built from what the push actually did, not from what it asked for: a
// refused reference is not in `ref_updates`, so an attestation can never
// claim a reference moved that did not.
//
// The `Result` goes in whole. Handing `objects_pushed` — which is the `Err`
// arm already flattened to `&[]` for the log line above — is what made the
// record claim an empty object list for every push into a repository with
// no policy gates.
let attested_objects = outcome.new_objects.as_deref().map_err(String::as_str);
if let Some(attestation) = attestation_for(
request,
&outcome,
attested_objects,
&fingerprint,
&credential,
) {
tracing::info!(
target: TARGET_GIT,
event = "git.attestation",
refs = attestation.ref_updates.len(),
objects = attestation.new_object_count,
// Absent from the line when the set WAS enumerated (`Option`
// records nothing for `None`), so the field's presence is the fact.
oids_unknown = (!attestation.new_object_oids_unknown.is_empty())
.then_some(attestation.new_object_oids_unknown.as_str()),
pack = %attestation.pack_hash,
key_fingerprint = %attestation.key_fingerprint,
// The field an HTTP push can be attributed by. `key_fingerprint`
// above is empty on that transport (GUN-014).
credential = %attestation.credential,
"push attestation"
);
if let Err(err) = attestations.append(&attestation) {
// the attestation's durable record failed, and the push has already been
// reported to the client as done. Saying so loudly is the only
// correct action; failing here would make gunnar and the client
// disagree about what is in the repository.
tracing::error!(
target: TARGET_GIT,
path = %attestations.path().display(),
%err,
"the push attestation could not be appended to the log; the push itself stands"
);
}
if let Some(bus) = bus {
bus.publish(event::Payload::PushAttestation(attestation));
}
}
Ok(())
}
// ───────────────────────────────────────────────────────────────────────────
// Measurements, and the absence of one
// ───────────────────────────────────────────────────────────────────────────
/// A measurement split into **the number** and **why there isn't one**.
///
/// `tracing`'s `Value` impl for `Option<T>` records nothing at all for `None`,
/// so the numeric field is *absent* from the line when the measurement is,
/// rather than present and zero. That is the whole point. A reader who sees
/// `pack_bytes=0` has to be able to conclude the push carried no bytes, and
/// they cannot if `0` is also what a failed `stat` prints.
///
/// The same shape as `objects` / `objects_unknown` on the line above it:
/// one field for the fact, one for the reason it is missing, never
/// a plausible default standing in for either.
fn measured(value: Option<u64>, why: &str) -> (Option<u64>, Option<String>) {
match value {
Some(bytes) => (Some(bytes), None),
None => (None, Some(why.to_owned())),
}
}
/// How many bytes the pack that landed occupies, or why that is unknown.
///
/// Three different absences used to fold into one `0`:
///
/// * the push carried no pack at all, because it was deletions only;
/// * the pack was indexed without a directory, so nothing was written;
/// * `stat` failed on a path that *was* written.
///
/// Only the first is a zero, and it is a **measured** one. The other two are
/// unknowns and now say so, with the path in the reason: an operator chasing a
/// pack that is not where the server thinks it is needs the path more than they
/// need a number. This is a server log, not a client-facing status line, so the
/// path belongs here (the reverse mistake — a server path on the wire to
/// `git push` — was fixed separately).
fn measured_pack_bytes(outcome: &PushOutcome) -> (Option<u64>, Option<String>) {
let Some(pack) = outcome.pack.as_ref() else {
// Deletions only: nothing was written because nothing was sent.
return (Some(0), None);
};
let Some(path) = pack.data_path.as_deref() else {
// No file — which is the ordinary case, not a degraded one: the znippy
// engine appends the pack verbatim into its archive and writes no
// `.pack`. The store counted what it appended, so the measurement is
// right here and only an engine that reported neither is an unknown.
return match pack.stored_bytes {
Some(bytes) => (Some(bytes), None),
None => (
None,
Some(format!(
"pack {} was indexed without a directory and the store reported no size, so \
there is nothing to measure",
pack.pack_id
)),
),
};
};
match std::fs::metadata(path) {
Ok(meta) => (Some(meta.len()), None),
Err(err) => (
None,
Some(format!("{} could not be stat'ed: {err}", path.display())),
),
}
}
/// How many objects a push introduced, for the attestation record.
///
/// **The count never depends on the list.** `Ok(ids)` is an enumerated set and
/// its length is the count, empty included — a push that introduced nothing is
/// a measured zero. `Err(_)` is "the set was never worked out", and the count
/// then comes off the stored pack's own 12-byte header, which is known either
/// way. Zero only for a push that carried no pack at all.
fn attested_object_count(
new_objects: std::result::Result<&[ObjectId], &str>,
outcome: &PushOutcome,
) -> u64 {
match new_objects {
Ok(ids) => ids.len() as u64,
Err(_) => outcome
.pack
.as_ref()
.map(|pack| u64::from(pack.num_objects))
.unwrap_or(0),
}
}
/// The push attestation record for one push, or `None` when nothing moved.
///
/// A push in which every reference was refused produces `PushRejected` events
/// and no attestation: there is no provenance to attest to, and a record with
/// an empty `ref_updates` would be indistinguishable from a bug in this
/// function.
fn attestation_for(
request: &ServiceRequest,
outcome: &PushOutcome,
new_objects: std::result::Result<&[ObjectId], &str>,
fingerprint: &str,
credential: &str,
) -> Option<PushAttestation> {
let ref_updates: Vec<gunnar_proto::RefUpdate> = outcome
.commands
.iter()
.zip(&outcome.statuses)
.filter(|(_, status)| status.is_ok())
.map(|(command, _)| gunnar_proto::RefUpdate {
ref_name: command.name.to_string(),
old_oid: command.old.to_string(),
new_oid: command.new.to_string(),
})
.collect();
if ref_updates.is_empty() {
return None;
}
// The pack as gunnar stored it. `pack_hash` is that pack's own trailing
// checksum — the `<id>` in `pack-<id>.pack` — in the repository's hash
// kind, so it is verifiable against the file on disk. It is deliberately
// NOT a hash of the bytes the client sent: `index_thin_pack` completes a
// thin pack from the store before writing it, so the received bytes and
// the stored pack are different objects and only one of them still exists.
//
// **Through `measured_pack_bytes`, not through a `stat` of its own.** This
// used to `std::fs::metadata(data_path).ok()…unwrap_or(0)`, which on an
// engine that writes no packfile is `None.unwrap_or(0)` — so every
// attestation this server produced said the push stored zero bytes. Same
// disease as the empty object list below, in the same record, one field
// along. The znippy store writes no pack FILE (`data_path: None`) but it
// counts the bytes it appends, and that count is exactly what
// `measured_pack_bytes` was taught to prefer over an absent file — one
// function, so the log line and the durable record can never disagree
// about how big a push was.
//
// The last resort is the wire count (`PushOutcome::pack_bytes`), for an
// engine that reports neither a file nor a size: a byte count that could
// not be established at all stays `0` here because the proto has no third
// state for it, and `new_object_oids_unknown` is the only field that can
// say the record is partial. What is gone is the case where the count was
// *available* and thrown away for want of a file.
let (pack_hash, pack_bytes) = match &outcome.pack {
Some(pack) => {
let bytes = measured_pack_bytes(outcome)
.0
.or(outcome.pack_bytes)
.unwrap_or(0);
(pack.pack_id.to_string(), bytes)
}
// A deletions-only push carries no pack at all, and that is a fact
// worth attesting rather than a reason to skip the record.
None => (String::new(), 0),
};
// ── the object set, and the reason there isn't one ──────────────────────
//
// The set is not free: on the znippy engine `oids_in_extent` WAITS for the
// indexer to absorb the pack, and that wait is what the deferred indexer
// exists to avoid paying on the push path (`znippy_engine::ingest`'s "why
// this is a flag" note). So `PushIngest` says how badly it wants the ids
// rather than always asking — `ObjectSet::Required` when a policy gate
// reads them (`gate.engine` non-empty), `ObjectSet::Wanted` when only this
// record does, and `ObjectSet::NotNeeded` under `--no-attest-objects`.
// Where the set is not produced — an operator who bought the latency back,
// or an engine that cannot enumerate for a `Wanted` — the ingest says so
// with `Err(reason)` rather than an empty set.
//
// That `Err` used to be flattened to `&[]` by the caller. The record then
// carried an empty list, and an empty list is what a push of zero objects
// carries too, so the two were indistinguishable — and the fix before this
// one made it worse in the honest direction only halfway: the COUNT became
// the pack's, so a reader saw "3 objects" beside a list of none and no word
// about why. The reason travels now, in its own field, and the record says
// what it knows and no more (measured, and never a plausible value where an absence belongs).
let new_object_count = attested_object_count(new_objects, outcome);
let (new_object_oids, new_object_oids_unknown) = match new_objects {
Ok(ids) => (ids.iter().map(ObjectId::to_string).collect(), String::new()),
Err(reason) => (Vec::new(), reason.to_owned()),
};
let received_at_unix_ms = gunnar_grpc::bus::now_unix_ms();
Some(PushAttestation {
// Unique per push: the stored pack's identity is unique to the bytes,
// and the millisecond disambiguates the same pack pushed twice (and a
// deletions-only push, which has no pack at all).
attestation_id: format!(
"{}@{received_at_unix_ms}/{}",
request.repo.key(),
if pack_hash.is_empty() {
"no-pack"
} else {
&pack_hash
}
),
repo: Some(RepoPath::new(&request.repo.namespace, &request.repo.name)),
principal_id: request.principal.id.clone(),
principal_name: request.principal.name.clone(),
key_fingerprint: fingerprint.to_owned(),
// Answerable on both transports, unlike the fingerprint above
// (GUN-014). A verifier reading an older record sees an empty string,
// which is what it saw before this field existed.
credential: credential.to_owned(),
hash_kind: crate::control_plane::hash_kind(request.repo.hash_kind) as i32,
ref_updates,
// The list is never a partial one: it is either the set the store
// enumerated or empty beside a reason that says why nobody asked.
// `new_object_oids_unknown` is what a reader honours; a reader that
// ignores it sees a count that still came from the pack's own header
// rather than from a list that was never built.
new_object_count,
new_object_oids,
new_object_oids_unknown,
new_object_bytes: pack_bytes,
pack_hash,
received_at_unix_ms,
server_agent: gunnar_proto::AGENT.to_owned(),
// The no-OS rule: git traffic has exactly one transport, and this module is the
// SSH one. Not read from the request, because there is nothing else it
// could be, and a field that can only ever hold one value should say
// which one rather than sit empty.
transport: "ssh".to_owned(),
})
}
/// The SHA-256 fingerprint of the key this push authenticated with, when that
/// can be known without guessing.
///
/// The key that authenticated this session, for the attestation and the events.
///
/// # This used to be a guess, and the guess was usually wrong
///
/// The cross-crate gap is closed: `gunnar-ssh` computes the fingerprint in
/// `auth_publickey`, looks the principal up by it, and now **carries it** on
/// [`ServiceRequest`] instead of dropping it.
///
/// What this replaced read the identity table back and answered only when the
/// principal held **exactly one key** — so a machine account with a rotation
/// key, or any human with a laptop key and a desktop key, published an empty
/// fingerprint on every `RefUpdated`, every `RefDeleted` and every attestation.
/// "Which key pushed this" was therefore unanswerable for precisely the
/// accounts an audit trail exists for. Reading it back could never have been
/// made correct: the table cannot say which of two keys made a connection.
///
/// # An empty string here does NOT mean "nobody" (GUN-014)
///
/// It used to say so: *"an empty string means no key, and now means it for one
/// reason only — the anonymous, read-only path, which authenticates with no key
/// at all. That path cannot push, so it cannot reach an attestation."* That was
/// false for smart HTTP. HTTP authenticates with a **token**, so there is no
/// key and `ServiceRequest::key_fingerprint` is `None` on every HTTP request
/// including `git-receive-pack` — and the empty string went into `RefUpdated`,
/// `RefDeleted` and the attestation with nothing else on the record naming who
/// pushed. Two HTTP pushes by two different people were the same audit line.
///
/// The field is still exactly what it says: the SSH key, when there was one.
/// The question "which credential" is [`session_credential`]'s.
fn session_key_fingerprint(request: &ServiceRequest) -> String {
match &request.key_fingerprint {
Some(fingerprint) => fingerprint.as_str().to_owned(),
None => {
tracing::debug!(
target: TARGET_GIT,
principal = %request.principal.id,
"no key fingerprint on this request: the session authenticated anonymously"
);
String::new()
}
}
}
/// **Which credential made this request**, as the audit trail names it:
/// `key:SHA256:…`, `token:<hash prefix>`, or empty for a caller that presented
/// none.
///
/// The field [`session_key_fingerprint`] could not be (GUN-014). It is derived
/// here rather than spelled by each front door so the two transports cannot
/// come to write the prefix differently: SSH already carries the fingerprint
/// and gets `key:` added to it; HTTP has no key and carries
/// [`ServiceRequest::credential`] instead.
///
/// Empty now genuinely has one meaning — no credential — and the anonymous read
/// path is the only thing that produces it, which is what the old note claimed
/// and this makes true.
fn session_credential(request: &ServiceRequest) -> String {
match (&request.key_fingerprint, &request.credential) {
(Some(fingerprint), _) => format!("key:{}", fingerprint.as_str()),
(None, Some(credential)) => credential.clone(),
(None, None) => String::new(),
}
}
// `pack_object_ids` stood here: it opened a written `.idx` with
// `gix_pack::index::File::at` and collected every oid in it. It went with
// `index_thin_pack`, which is the only thing that ever wrote an `.idx` for it
// to read — a push is a verbatim `put_pack` into the engine now, and the ids a
// push introduced come from `PushReceipt::new_oids` off the engine's own index.
//
// It carried the last two live bumps of `gunnar_pack::oidprobe`'s
// `GIX_PACK_INDEX_OPEN` / `GIX_PACK_INDEX_ITER`, which `bench/examples/
// oidlookup-probe.rs` reports. Deleting it changes **no measurement**: the
// function had no caller, so those counters were already zero in every run that
// could be taken. A counter that reads 0 because the path is gone is the truth;
// keeping an uncalled function alive to hold a probe open would be the other
// way round — an instrument attached to code that cannot execute.
/// Taking a pushed pack into `objects/pack`, and saying which objects it
/// **introduced**.
///
/// # The subtraction, and why it is the only honest source
///
/// A gate cannot enforce a quota against a set that includes objects the
/// repository already held. This used to hand the gate every id in the written
/// `.idx`, which for a thin push is the client's objects *plus* the bases
/// `index_thin_pack` fetched from the store to complete it — so a
/// `max_objects` gate refused pushes it need not have, and no amount of
/// re-reading the index afterwards could tell the two apart, because by then
/// they are the same file.
///
/// The distinction exists in exactly one place: the `gix_object::Find` that
/// resolution asks for a missing base. Every id it is asked for and finds is,
/// by definition, an object the repository already had. So the lookup is
/// wrapped, the ids it served are recorded, and
///
/// ```text
/// introduced = everything in the .idx − the bases we handed back
/// ```
///
/// is exact rather than conservative, and needs no second enumeration of
/// anything (one implementation).
/// **Where a pushed or fetched pack is framed**: the box's transaction log, and the catalog
/// key and hash kind the twin's apply opens the repository by (`gunnar_core::txlog`).
#[derive(Clone)]
pub(crate) struct PackFrame {
log: Arc<gunnar_core::txlog::Log>,
repo: String,
hash: String,
}
impl PackFrame {
/// The frame for `repo` on `stores`.
pub(crate) fn of(stores: &crate::stores::ServerStores, repo: &gunnar_core::Repo) -> PackFrame {
PackFrame {
log: Arc::clone(stores.txlog()),
repo: repo.key(),
hash: repo.hash_kind.to_string(),
}
}
/// These exact bytes, as a `Pack` frame, durable before this returns.
fn append(&self, bytes: &[u8]) -> anyhow::Result<()> {
let meta = gunnar_core::txlog::Meta::Pack {
repo: self.repo.clone(),
hash: self.hash.clone(),
};
self.log
.append(gunnar_core::txlog::Kind::Pack, &meta, bytes)?;
Ok(())
}
}
pub(crate) struct PushIngest<'a> {
git_dir: PathBuf,
hash_kind: gix_hash::Kind,
/// Which engine stores this push. Carried on the value rather than read
/// from a global, because the process's engine is a field on
/// [`ServerStores`](crate::stores::ServerStores) now and a push is served
/// out of one of those.
engine: crate::storage::Engine,
/// How badly this push needs the object set it introduced: a policy gate
/// that reads the objects cannot decide without it, the push attestation
/// would rather have it than apologise for it, and an operator can decline
/// it altogether. Asking for it waits on index work the ack path defers on
/// purpose, which is why this is three answers and not a `bool`.
objects: crate::storage::ObjectSet,
/// **Where the running object count goes, or `None`.**
///
/// The numerator of a transfer's progress, and the reason it is on the
/// *ingest* rather than on `gunnar-wire`'s `FetchProgress` beside the
/// denominator: the count exists only inside the engine's pack walk, and
/// there is exactly one of those (one implementation). `fetch_pack` reads the header and
/// can publish `objects_total` because twelve bytes are all that costs;
/// counting the objects themselves would be a second decoder, which its own
/// module doc refuses. So the two halves of one progress pair are published
/// from two places, and this is the half storage owns.
watching: Option<&'a dyn git_storage_trait::PackProgress>,
/// **The transaction log this pack is a frame in**, or `None` (a test, a tool). Handed
/// to `Engine::ingest_push` as `before_store`: the pack's exact bytes are a `Pack` frame
/// before the engine stores them — this path never passes the `OpenRepo::logged` tee.
frame: Option<PackFrame>,
/// **An encrypted repository's objects, or `None`** (crypt-level2 item 4,
/// 2026-10-06): when set, the pushed pack is checked in quarantine
/// ([`crate::encrypted_quarantine`]) BEFORE it is framed or stored — these
/// are the bases a thin pack deltas against. A refusal keeps nothing.
encrypted: Option<&'a dyn git_storage_trait::GitServe>,
/// **The data set's space ledger**, or `None` (a test, a tool, a server
/// whose doors were never wired): every chunk of the pack is admitted
/// against it BEFORE it is read, and a pack that does not fit beside what
/// is already arriving is refused by name — never failed by a full disk
/// mid-receive (cov-13, [`crate::push_space`]).
space: Option<Arc<gunnar_ssh::SpaceLedger>>,
}
impl<'a> PushIngest<'a> {
/// **The ingest a MIRROR lands its fetched pack through** — this one, the
/// push's own, and not a second sink.
///
/// `fetch_pack`'s module doc states the principle it is built on: *a
/// fetched pack and a pushed pack are the same bytes*, so both land through
/// the identical [`gunnar_wire::PackIngest`] and znippy implements it once.
/// A private constructor rather than a public struct because the fields are
/// this module's business; what a mirror needs to say is the four facts
/// below.
pub(crate) fn new(
git_dir: PathBuf,
hash_kind: gix_hash::Kind,
engine: crate::storage::Engine,
objects: crate::storage::ObjectSet,
watching: Option<&'a dyn git_storage_trait::PackProgress>,
frame: Option<PackFrame>,
) -> Self {
PushIngest {
git_dir,
hash_kind,
engine,
objects,
watching,
frame,
// A mirror's fetched pack is not checked here: the repository it
// lands in is the mirror's own (an encrypted repository is not a
// mirror target in Level 1).
encrypted: None,
space: None,
}
}
/// Hold the pack to the data set's room — `stores.space_ledger()`.
pub(crate) fn with_space(mut self, space: Option<Arc<gunnar_ssh::SpaceLedger>>) -> Self {
self.space = space;
self
}
/// **The push, stored without a packfile.**
///
/// The bytes go from the wire into the znippy engine's own storage and nothing
/// is written to `objects/pack` — no `.pack`, no `.idx`, no `.keep`. That is
/// the whole point of it, and it is why this does not call
/// [`index_thin_pack`]: `gix_pack::Bundle::write_to_directory` takes a
/// directory, so every route through it puts a git packfile on disk before
/// the tier ever sees the bytes, and the tier's copy is then a second one.
///
/// # What is given up, said out loud
///
/// `index_thin_pack` **resolved** the pack: it completed thin deltas against
/// the store and produced a real `.idx`. This does not. The closure check
/// inside `put_pack` refuses a pack whose bases are missing, which is the
/// correctness half, but the pack stays thin and stays verbatim — the whole
/// claim of *verbatim is the truth*. A reader resolves chains on the way
/// out instead.
///
/// # The `.keep` that is not there
///
/// `index_thin_pack` wrote one and the caller dropped it once the refs
/// landed, guarding the new pack from a concurrent `gc` in the window
/// between. There is no window here: `put_pack` returns only after the
/// journal row is fsynced, and `gc` reads that journal. The marker's job is
/// done by the ordering rather than by a file.
fn ingest_engine(&self, pack: &mut dyn BufRead) -> gunnar_wire::Result<PackReceipt> {
// ── space admission (cov-13) ─────────────────────────────────────
//
// The promise this pack holds on /data: spool + store, + the log's
// frame when it is framed. Given back copy by copy as each lands (the
// spool at the reader's `consume`, the frame in the hook below, the
// store when this function returns and the hold drops).
let hold = self.space.as_ref().map(|ledger| {
ledger.hold(if self.frame.is_some() {
gunnar_ssh::PUSH_COPIES
} else {
gunnar_ssh::PUSH_COPIES - 1
})
});
let mut gated = None;
let reader: &mut dyn BufRead = match &hold {
Some(hold) => gated.insert(crate::push_space::Gated::new(
pack,
hold,
crate::storage::SPOOL_ABOVE as u64,
)),
None => pack,
};
let hook = |bytes: &[u8]| -> anyhow::Result<()> {
// ★ Before the frame and before the store: an encrypted
// repository's pack is checked in quarantine, and a refusal keeps
// nothing — not on this disk, not in the log the twin replays.
#[cfg(feature = "engine-znippy")]
if let Some(repo) = self.encrypted {
let checked = crate::encrypted_quarantine::check(bytes, self.hash_kind, repo)?;
tracing::info!(
target: TARGET_STORE,
event = "store.encrypted.quarantine_passed",
objects = checked,
"an encrypted repository's pushed pack passed its check in quarantine; it is framed and stored now"
);
}
match &self.frame {
Some(frame) => {
frame.append(bytes)?;
// On the disk now: the frame's promise is `statvfs`'s.
if let Some(hold) = &hold {
hold.realize(bytes.len() as u64);
}
Ok(())
}
None => Ok(()),
}
};
let hooked = self.frame.is_some() || self.encrypted.is_some();
// The mutation gate spans frame through store (`Log::mutation`): a snapshot's freeze
// waits for this pack to be in the engine before it reads the head.
let _framed = self.frame.as_ref().map(|f| f.log.mutation());
let stored = self.engine.ingest_push(
self.git_dir.as_path(),
self.hash_kind,
reader,
self.objects,
self.watching,
hooked.then_some(&hook as &dyn Fn(&[u8]) -> anyhow::Result<()>),
);
// ★ Cut for room: the ledger's line, verbatim, whatever the ingest made
// of the cut stream. Its spool is already gone (the failed read dropped
// it), nothing was framed, nothing stored, and its promise is given back.
let consumed_by_cut = gated
.as_ref()
.map(|g| (g.cut().map(str::to_owned), g.consumed()));
drop(gated);
if let Some((Some(line), sent)) = consumed_by_cut.clone() {
tracing::warn!(
target: TARGET_STORE,
event = "store.push.no_room",
pack_bytes = sent,
refusal = %line,
"a pushed pack was refused for room mid-receive: it did not fit beside the \
pushes already arriving; its spool is deleted and nothing of it was kept"
);
return Err(gunnar_wire::Error::Refused(line));
}
let received = stored
.map_err(|err| {
// ★ The disk filled ANYWAY (a writer outside the ledger): said in
// the ledger's words, never `uring: op failed: Operation canceled`.
if let Some(hold) = &hold {
let freed = consumed_by_cut.as_ref().map_or(0, |(_, sent)| *sent);
if crate::push_space::is_space_failure(&err, || hold.at_the_floor(freed)) {
let line = hold.full_now();
tracing::warn!(
target: TARGET_STORE,
event = "store.push.disk_full",
error = %format!("{err:#}"),
refusal = %line,
"the store failed a pushed pack because the data set filled; told by name"
);
return gunnar_wire::Error::Refused(line);
}
}
#[cfg(feature = "engine-znippy")]
if let Some(refused) = crate::encrypted_quarantine::refusal_in(&err) {
tracing::info!(
target: TARGET_STORE,
event = "store.encrypted.quarantine_refused",
reason = %refused,
"an encrypted repository's pushed pack was refused in quarantine: nothing was framed or stored"
);
return gunnar_wire::Error::Refused(refused);
}
// `{err:#}`, not `{err}`. `anyhow`'s plain `Display` prints the
// OUTERMOST context and nothing under it, so a failure three
// `.context()` calls deep arrived as "storing the pushed pack:
// enumerating the objects this push introduced" — the two
// labels this code added, and never the cause the store gave.
// The alternate form joins the whole chain, which is the only
// version of this line an operator can act on.
gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
"storing the pushed pack: {err:#}"
)))
})?;
let pack_id = gix_hash::ObjectId::try_from(received.pack_id.as_slice()).map_err(|_| {
gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
"the stored pack's trailer is {} bytes, which is not a {} object id",
received.pack_id.len(),
self.hash_kind
)))
})?;
tracing::info!(
target: TARGET_STORE,
// Renamed from `store.cold_tier.ingested` 2026-08-14. The cold tier
// was purged and this event outlived it, so every push in the log
// still announced a tier that does not exist. Nothing asserted on
// the old name — only a doc-comment example, corrected with it.
event = "store.znippy.ingested",
objects = received.num_objects,
bytes = received.pack_bytes,
elapsed_ms = received.elapsed_ms,
object_set = received.new_oids.is_some(),
"the push was stored in the znippy-git store; no packfile was written"
);
let new_objects = match received.new_oids {
Some(oids) => oids
.iter()
.map(|raw| {
gix_hash::ObjectId::try_from(raw.as_slice())
.map_err(|_| format!("the store returned a {}-byte object id", raw.len()))
})
.collect::<std::result::Result<Vec<_>, _>>(),
// Not "no objects" — see `ColdReceipt::new_oids`. A gate reading
// this refuses, which is the correct answer to "I was never asked
// to find out".
None => Err(
"the policy has no gate over objects, so the object set was not \
enumerated"
.to_owned(),
),
};
Ok(PackReceipt {
indexed: gunnar_wire::pack::IndexedPack {
num_objects: received.num_objects,
pack_id,
// There is no index file, and `ObjectId::null` is this tree's
// existing spelling for that — the same value the git-store
// bench server and the store's own end-to-end test use.
index_id: gix_hash::ObjectId::null(self.hash_kind),
pack_version: gunnar_wire::pack::PackVersion::from_u32(received.version),
hash_kind: self.hash_kind,
// The three paths that no longer exist. `None` is not a
// degraded answer here: it is the fact, and `cold_tier_append`
// used to read exactly these two to build its second copy.
index_path: None,
data_path: None,
keep_path: None,
// …but the SIZE is not missing with them. The store counted the
// bytes it appended, and until this line nobody carried the
// number out, so `measured_pack_bytes` fell through to "there
// is nothing on disk to measure" on every push into the engine
// the appliance ships. There is no file; there is a
// measurement. Nothing on disk has to be stat'ed to learn it,
// and on this engine nothing on disk could be.
stored_bytes: Some(received.pack_bytes),
},
new_objects,
})
}
}
impl gunnar_wire::PackIngest for PushIngest<'_> {
/// Straight into the engine. There is no second arm.
///
/// What stood beside this ran `index_thin_pack` — `gix_pack::Bundle::
/// write_to_directory` — writing a real `.pack`, `.idx` and `.keep` into
/// `objects/pack`, reading the `.idx` back to learn which objects arrived,
/// and dropping the `.keep` afterwards. Every line of it was the packfile
/// engine's, and the engine is amputated.
fn ingest(&self, pack: &mut dyn BufRead) -> gunnar_wire::Result<PackReceipt> {
self.ingest_engine(pack)
}
}
/// Did this reference move to a tip that does **not** descend from the one it
/// held? The event stream's `RefUpdated.forced`.
///
/// # This is one ancestry walk, not a second one
///
/// [`gunnar_policy::ancestry::is_descendant_of`] is the same function
/// `BranchProtectionGate` decides a non-fast-forward with, and it is the one
/// cross-checked against `git merge-base --is-ancestor` in that crate's tests.
/// Re-deriving the answer here — `old != new`, a hex compare, a first-parent
/// walk — is the mistake that gate's module docs enumerate, and two
/// implementations that disagree would put a UI and a policy on different sides
/// of the same push (one implementation).
///
/// The gate cannot simply be asked, because it only runs where a protection
/// rule matches. The audit trail has to answer for **every** reference on every
/// repository, including the ones nothing protects, which is the case where
/// nobody is watching.
///
/// # An undecidable walk reports `true`
///
/// A missing object or an exhausted budget is not evidence of a fast-forward.
/// `BranchProtectionGate` refuses on undecided rather than allowing; the same
/// judgement here means flagging rather than clearing, with the reason logged at
/// `warn`. Reporting `false` for a walk that never finished is exactly the
/// wrong-answer this field is being fixed for.
fn was_forced(git: &dyn GitServe, command: &RefCommand) -> bool {
// A create has no previous tip, and a delete has no new one. Neither is a
// rewrite of anything.
if command.is_create() || command.is_delete() {
return false;
}
match gunnar_policy::ancestry::is_descendant_of(
git,
&command.new,
&command.old,
gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
) {
Ok(descends) => !descends,
Err(err) => {
tracing::warn!(
target: TARGET_POLICY,
ref_name = %command.name,
old = %command.old,
new = %command.new,
%err,
"the ancestry walk behind RefUpdated.forced could not be completed; the update \
is published as forced, because `not forced` is a claim this could not make"
);
true
}
}
}
#[allow(clippy::too_many_arguments)]
fn publish(
bus: &EventBus,
request: &ServiceRequest,
command: &RefCommand,
status: &RefStatus,
atomic: bool,
fingerprint: &str,
credential: &str,
gates: &HashMap<String, String>,
forced: bool,
) {
let repo = Some(RepoPath::new(&request.repo.namespace, &request.repo.name));
let payload = match status {
RefStatus::Ok if command.is_delete() => event::Payload::RefDeleted(RefDeleted {
repo,
ref_name: command.name.to_string(),
old_oid: command.old.to_string(),
principal_id: request.principal.id.clone(),
key_fingerprint: fingerprint.to_owned(),
credential: credential.to_owned(),
}),
RefStatus::Ok => event::Payload::RefUpdated(RefUpdated {
repo,
ref_name: command.name.to_string(),
old_oid: command.old.to_string(),
new_oid: command.new.to_string(),
principal_id: request.principal.id.clone(),
key_fingerprint: fingerprint.to_owned(),
credential: credential.to_owned(),
// Answered by [`was_forced`], not assumed. This was a hardcoded
// `false`, which made every force-push on the server
// indistinguishable from an ordinary one on `Events.Watch`.
forced,
}),
RefStatus::Rejected(reason) => event::Payload::PushRejected(PushRejected {
repo,
ref_name: command.name.to_string(),
// Empty when nothing named a gate: a connectivity failure, a
// compare-and-swap that lost, or the collateral refusal every other
// reference gets in an `atomic` push. Those are real refusals and
// they belong on the stream, but attributing them to a gate that
// never ran would be a lie a UI would repeat.
gate: gates
.get(&command.name.to_string())
.cloned()
.unwrap_or_default(),
reason: reason.clone(),
principal_id: request.principal.id.clone(),
whole_push_rejected: atomic,
}),
};
bus.publish(payload);
}
// ───────────────────────────────────────────────────────────────────────────
// the policy seam
// ───────────────────────────────────────────────────────────────────────────
/// `gunnar-wire`'s [`PushGate`] over `gunnar-policy`'s [`PolicyEngine`].
struct PolicyGate {
engine: PolicyEngine,
/// Why this repository's policy could not be read, if it could not be.
///
/// `Some` makes [`PolicyGate::check`] refuse **every** reference (GUN-007).
/// An unreadable policy used to become `PolicyConfig::default()`, and an
/// empty engine is let through by the `is_empty()` branch below — so a
/// document with a typo, or a storage fault, quietly turned branch
/// protection and `require_signature` off for the duration. This is the
/// field that keeps "no gates are configured" and "the gates could not be
/// read" from being the same value.
unreadable: Option<String>,
principal: gunnar_policy::Principal,
repo: RepoId,
keys: Box<dyn KeyDirectory>,
metrics: Arc<ServerMetrics>,
policy_path: PathBuf,
/// Reference name → the `Gate::name()` that refused it.
///
/// `PushGate::check` may only answer `Err(String)`, so the *name* of the
/// gate that said no is lost at that boundary; it survives only inside the
/// `Decision`. The event stream's `PushRejected.gate` needs it, so it is recorded here
/// on the way past rather than parsed back out of the reason string; a
/// reason is written for a human and its shape is free to change.
refused_by: Mutex<HashMap<String, String>>,
}
/// `gunnar-core`'s principal kind as `gunnar-policy` names it.
///
/// # The fourth arm, which must never be taken
///
/// A [`PolicyGate`] is built only for a **push**, and a push needs
/// [`Level::Write`](gunnar_core::Level::Write), which
/// `gunnar_core::AuthTable::may` refuses to the anonymous principal
/// unconditionally, before `Dispatch::authorise` returns and long before this
/// runs. So the anonymous arm is unreachable through every route that exists.
///
/// It is loud rather than plausible. `gunnar-policy` has three kinds and none of
/// them means "nobody"; quietly answering `User` would put an anonymous push
/// through gates written about humans, and do it silently. If this line is ever
/// reached, the ACL has stopped being consulted before the gate. That is a
/// security incident, and it reads like one in the log.
pub(crate) fn policy_principal_kind(
kind: gunnar_core::PrincipalKind,
) -> gunnar_policy::PrincipalKind {
match kind {
gunnar_core::PrincipalKind::User => gunnar_policy::PrincipalKind::User,
gunnar_core::PrincipalKind::Machine => gunnar_policy::PrincipalKind::Machine,
gunnar_core::PrincipalKind::Deploy => gunnar_policy::PrincipalKind::Deploy,
// A group holds no keys and is never authenticated as (punkt 9), so a
// push can never be attributed to one — the same unreachable-by-design
// shape as the anonymous arm below, and loud for the same reason: a
// group at the pre-receive gate means a keyless identity was handed a
// push, which is the ACL being bypassed.
gunnar_core::PrincipalKind::Group => {
tracing::error!(
target: gunnar_core::TARGET_AUDIT,
event = "policy.gate.group_push",
"a GROUP principal reached the pre-receive gate. A group holds no keys and \
cannot authenticate, so this cannot happen unless the grant check has been \
bypassed. Treating it as a machine identity for the gate; investigate \
immediately."
);
debug_assert!(
false,
"a group principal reached the pre-receive gate (punkt 9)"
);
gunnar_policy::PrincipalKind::Machine
}
gunnar_core::PrincipalKind::Anonymous => {
tracing::error!(
target: gunnar_core::TARGET_AUDIT,
event = "policy.gate.anonymous_push",
"an ANONYMOUS principal reached the pre-receive gate. Anonymous access is \
read-only and the ACL refuses it Write, so this cannot happen unless the \
grant check has been bypassed. Treating it as a machine identity for the \
gate; investigate immediately."
);
debug_assert!(false, "an anonymous principal reached the pre-receive gate");
gunnar_policy::PrincipalKind::Machine
}
}
}
impl PolicyGate {
fn open(
git_dir: &Path,
request: &ServiceRequest,
stores: &ServerStores,
metrics: Arc<ServerMetrics>,
) -> Self {
let policy_path = git_dir.join(POLICY_FILE);
let (config, unreadable) = match load_policy(stores, &request.repo, &policy_path) {
Ok(config) => (config, None),
// An empty engine on purpose: nothing should be evaluated against
// a policy that could not be read, and `check` refuses before it
// would be.
Err(why) => (PolicyConfig::default(), Some(why)),
};
// An ENCRYPTED repository adds the plaintext check to whatever policy
// it has: the flag is the account admin's, not the policy file's, so
// no policy edit can switch it off.
let mut gates = config.gates();
if request.repo.encrypted {
gates.push(Box::new(gunnar_policy::EncryptedGate::new()));
}
PolicyGate {
engine: PolicyEngine::new(gates),
unreadable,
principal: gunnar_policy::Principal {
id: request.principal.id.clone(),
name: request.principal.name.clone(),
kind: policy_principal_kind(request.principal.kind),
},
repo: RepoId::new(&request.repo.namespace, &request.repo.name),
keys: Box::new(key_directory(stores)),
metrics,
policy_path,
refused_by: Mutex::new(HashMap::new()),
}
}
/// Which gate refused each reference, for the event-stream `PushRejected` events.
///
/// Read after `ReceivePack::serve` has returned, so every `check` that was
/// going to happen has happened.
fn gate_that_refused(&self) -> HashMap<String, String> {
self.refused_by
.lock()
.expect("the refusing-gate lock")
.clone()
}
}
impl PushGate for PolicyGate {
fn check(
&self,
command: &RefCommand,
facts: &PushFacts<'_>,
) -> std::result::Result<(), String> {
// One span per gate evaluation. It is a child of
// `git.receive_pack`, so the connection, principal and repository come
// for free; what is new is the reference and how long deciding it took.
// Branch protection walks ancestry, and an ancestry walk on a large
// repository is one of the two places a push spends unexplained
// seconds.
let span = tracing::debug_span!(
target: TARGET_POLICY,
"policy.gate",
ref_name = %command.name,
);
let _entered = span.enter();
let started = std::time::Instant::now();
// ── the policy could not be READ, which is not "there is none" ─────
//
// First, before `is_empty()`, because `is_empty()` is the branch that
// lets a push through and an unreadable policy produced exactly that
// value (GUN-007). A pre-receive refusal costs pushes and nothing
// else: clone, fetch and browse are untouched, and the operator is
// told which document and why.
if let Some(why) = &self.unreadable {
self.metrics.add_gate_refusals(1);
tracing::error!(
target: TARGET_POLICY,
event = "policy.gate.unreadable",
ref_name = %command.name,
%why,
"the policy for this repository could not be read, so every gate refuses"
);
return Err(format!(
"gunnar could not read the policy for this repository, so it refuses this push \
rather than serving it with no gates at all: {why}. Reads are unaffected; fix \
the document (or `gunnar control policy set-protection`) and push again."
));
}
if self.engine.is_empty() {
// No gates are configured, so there is nothing to decide. Said out
// loud rather than reached by an engine that happens to return an
// empty decision, because "allowed by every gate" and "there are no
// gates" are different facts.
return Ok(());
}
// A gate that cannot see what was pushed cannot decide, and an
// undecidable gate refuses (the pre-receive gate). This is the branch that keeps a
// configured limit from being silently satisfied by an empty object set.
let new_objects = match facts.new_objects {
Ok(ids) => ids,
// No live gate reads the set, so its absence decides nothing.
// The empty slice is never read: `reads_new_objects()` is exactly
// the question "would any gate look at it".
Err(_) if !self.engine.reads_new_objects() => &[],
Err(reason) => {
self.metrics.add_gate_refusals(1);
tracing::error!(
target: TARGET_POLICY,
event = "policy.gate.undecidable",
%reason,
"the pushed objects could not be enumerated, so every configured gate refuses"
);
return Err(format!(
"gunnar could not enumerate the pushed objects, so the policy gates in {} \
could not decide: {reason}",
self.policy_path.display()
));
}
};
let updates = [RefUpdate::from_wire(
command.name.clone(),
command.old,
command.new,
)];
let mut ctx = PushContext::new(
&self.principal,
&self.repo,
&updates,
new_objects,
facts.objects,
self.keys.as_ref(),
);
// The byte count reaches the gate. Without it
// `LimitsGate` refuses any push at all whenever `max_pack_bytes` is
// configured, because it will not pretend an unmeasured pack is under a
// limit.
if let Some(bytes) = facts.pack_bytes {
ctx = ctx.with_pack_bytes(bytes);
}
// The push options reach the gate. Unconditionally, including the empty
// slice: `Some(&[])` is "the client sent none" and `None` is "nobody
// looked", and `PushOptionsGate` refuses the second. Before this line
// the whole tree's only consumer of a parsed push option was `.len()`
// in the log field twenty lines below, so `push-options` was advertised
// on the wire and discarded.
ctx = ctx.with_push_options(facts.push_options);
let decision = self.engine.evaluate(&ctx);
let elapsed_ms = started.elapsed().as_millis() as u64;
if decision.is_allowed() {
tracing::debug!(
target: TARGET_POLICY,
event = "policy.gate.allowed",
gates = self.engine.gate_names().len(),
new_objects = new_objects.len(),
elapsed_ms,
"every gate allowed this reference"
);
Ok(())
} else {
// `evaluate` runs every gate, so a reference can be refused by more
// than one. The first in gate order is the one named, and the
// summary leads with the same one, so a UI and the client's `ng`
// line agree about who said no.
self.metrics.add_gate_refusals(1);
if let Some(first) = decision.rejections.first() {
// **`warn!`, and it NAMES THE GATE.** The audit trail's `Verify` asks for
// exactly one error line per refused push naming the gate and
// the reference; before this, a refused push produced a
// `PushRejected` event with an empty `gate` field and no log
// line at all that said who refused it.
tracing::warn!(
target: TARGET_POLICY,
event = "policy.gate.refused",
gate = %first.gate,
reason = %first.reason,
refused_by_gates = decision.rejections.len(),
elapsed_ms,
"a policy gate refused this reference"
);
self.refused_by
.lock()
.expect("the refusing-gate lock")
.insert(command.name.to_string(), first.gate.clone());
} else {
// `is_allowed()` said no and named nobody. That is a bug in the
// engine, not a refusal, and it must not read as a normal one.
tracing::error!(
target: TARGET_POLICY,
event = "policy.gate.refused",
gate = "",
elapsed_ms,
"the policy engine refused a reference without naming a gate"
);
}
Err(decision.summary())
}
}
}
/// Read the repository's policy document.
///
/// **`Err` is not an empty policy** (GUN-007). Until 2026-09-11 every failure
/// here — a document that did not parse, a control store that could not be
/// read, a file that could not be opened — returned `PolicyConfig::default()`,
/// and `PolicyGate::check` lets an empty engine through because "no gates are
/// configured" is a real and common answer. So a typo in a policy document, or
/// a transient storage fault, silently switched branch protection and the
/// signature requirement OFF for that push, at `error!` level in a log nobody
/// was reading at the time.
///
/// The old note here said refusing would be worse, because a typo would take
/// the repository offline. It does not: this is the PRE-RECEIVE gate, so a
/// refusal costs pushes and nothing else — clone, fetch, browse and every read
/// are untouched, and the operator is told which document and why. Degraded
/// operation must not be indistinguishable from full operation, which is the
/// same judgement `ancestry` already makes when a walk cannot finish.
///
/// `Ok(PolicyConfig::default())` survives for exactly one case and it is a
/// fact rather than a fault: nothing is stored and no file exists, so nothing
/// has been configured.
/// **The control store first, the file beside the repository second.**
///
/// `POLICY_FILE`'s constant used to carry the note *"When the catalog grows a
/// policy document, this constant is the only thing that changes."* The catalog
/// has grown one — `gunnar_core::PolicyStore`, at `refs/gunnar/policy` — and
/// this is that change. What did **not** change is the file: a repository
/// configured before the store existed keeps working, and an operator who edits
/// the JSON on disk still gets what they edited, until they write a policy over
/// the control plane and the stored one takes over.
///
/// The order is deliberate and is the only one that can be right. The stored
/// document is the one `Policy.SetProtection` writes and the one an operator
/// can see over gRPC; if a stale file could shadow it, turning protection on
/// through the console would appear to work and change nothing.
pub(crate) fn load_policy(
stores: &ServerStores,
repo: &gunnar_core::Repo,
path: &Path,
) -> std::result::Result<PolicyConfig, String> {
match stores.policies().get(&repo.path()) {
Ok(Some(document)) => match PolicyConfig::from_json(&document) {
Ok(config) => return Ok(config),
Err(err) => {
tracing::error!(
target: TARGET_POLICY,
repo = %repo.key(),
source = "control-store",
%err,
"the stored policy document did not parse; pushes to this repository are \
refused until it does"
);
return Err(format!("the stored policy document did not parse: {err}"));
}
},
// Nothing stored. The file below is the pre-store shape and is still
// the answer for a repository configured before `PolicyStore` existed.
Ok(None) => {}
Err(err) => {
// NOT a fall-through to the file any more. The file "may be older
// than what an operator configured" — its own words — so serving
// it when the authoritative store is unreachable is a silent
// downgrade to a policy nobody chose.
tracing::error!(
target: TARGET_POLICY,
repo = %repo.key(),
%err,
"the policy store could not be read; pushes to this repository are refused \
rather than gated by a file that may be older than what an operator \
configured"
);
return Err(format!("the policy store could not be read: {err}"));
}
}
let text = match std::fs::read_to_string(path) {
Ok(text) => text,
// The one `Ok(default)`: nothing stored and no file. Nothing has been
// configured, which is a fact and not a fault.
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return Ok(PolicyConfig::default())
}
Err(err) => {
tracing::error!(
target: TARGET_POLICY,
path = %path.display(),
%err,
"the policy document could not be read; pushes to this repository are refused"
);
return Err(format!(
"the policy document at {} could not be read: {err}",
path.display()
));
}
};
match PolicyConfig::from_json(&text) {
Ok(config) => Ok(config),
Err(err) => {
tracing::error!(
target: TARGET_POLICY,
path = %path.display(),
%err,
"the policy document did not parse; pushes to this repository are refused until \
it does"
);
Err(format!(
"the policy document at {} did not parse: {err}",
path.display()
))
}
}
}
/// Every key gunnar knows **that has not expired**, for the ssh-sig gate (the pre-receive gate).
///
/// `signing_keys()` and not `keys()`, and that one word is a defect fixed on
/// 2026-08-05: this function used to load every row in the table, so a key past
/// its `expires` could no longer authenticate an SSH session and could still
/// satisfy `require_signature` on a push. The expiry was enforced on exactly one
/// of the two things a key does.
pub(crate) fn key_directory(stores: &ServerStores) -> StaticKeyDirectory {
let mut directory = StaticKeyDirectory::new();
let keys = match stores.auth().signing_keys() {
Ok(keys) => keys,
Err(err) => {
tracing::error!(
target: TARGET_POLICY,
%err,
"the identity table could not be read for signature verification"
);
return directory;
}
};
for key in keys {
// The identity table holds two credential kinds in one hash space
// (auth.rs, the mTLS plan): an `x509` row is the bootstrap certificate,
// not an ssh key, and cannot verify an ssh-sig — skipped by name, so
// the WARN below is kept for a key that IS ssh and DID go bad
// (measured 2026-08-29: the band warned at every login).
if !ssh_verifiable_kind(&key.algo) {
tracing::debug!(target: TARGET_POLICY, algo = %key.algo, fingerprint = %key.fingerprint_sha256, "not an ssh key kind; not a signature verifier");
continue;
}
match gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh) {
Ok(parsed) => {
directory.add(&key.principal_id, parsed);
}
Err(err) => tracing::warn!(
target: TARGET_POLICY,
fingerprint = %key.fingerprint_sha256,
%err,
"a registered key did not parse and cannot verify a signature"
),
}
}
// GUN-MERGE-SIGN: gunnar's own merge-signing key signs only the merge commits gunnar
// wrote; it counts for every principal and scope, so a branch holding them can be
// pushed and merged where signatures are required. Read, never minted, here.
match stores.merge_signing_public() {
Ok(Some(key)) => {
directory.add_server_key(key);
}
Ok(None) => {}
Err(err) => tracing::error!(
target: TARGET_POLICY,
err = %format!("{err:#}"),
"the merge-signing key could not be read; gunnar's merge commits will not verify"
),
}
directory
}
/// Is a registered key's `algo` (the name as OpenSSH writes it) something
/// the ssh-sig verifier can hold? `ssh-*`, `ecdsa-*`, `sk-*` are; `x509`
/// (the bootstrap certificate stored beside them) is not.
pub fn ssh_verifiable_kind(algo: &str) -> bool {
algo.starts_with("ssh-") || algo.starts_with("ecdsa-") || algo.starts_with("sk-")
}
// ───────────────────────────────────────────────────────────────────────────
// Guards
// ───────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
/// ★ A build without the quarantine refuses an encrypted repository's push by name, and a
/// plain repository's push is never refused for it (wiring tracer B 2026-10-08).
#[test]
fn a_build_without_the_quarantine_refuses_an_encrypted_push_by_name() {
let line = super::unchecked_encrypted_push(true, false).expect("refused");
assert!(
line.contains("encrypted") && line.contains("this Gunnar"),
"{line}"
);
assert_eq!(
super::unchecked_encrypted_push(true, true),
None,
"the quarantine checks it"
);
assert_eq!(
super::unchecked_encrypted_push(false, false),
None,
"a plain repository"
);
}
use super::*;
use gunnar_wire::pack::{IndexedPack, PackVersion};
#[test]
fn the_signature_directory_takes_ssh_kinds_and_skips_the_x509_row_by_name() {
for ssh in [
"ssh-ed25519",
"ssh-rsa",
"ecdsa-sha2-nistp256",
"sk-ssh-ed25519@openssh.com",
"sk-ecdsa-sha2-nistp256@openssh.com",
] {
assert!(ssh_verifiable_kind(ssh), "{ssh}");
}
assert!(
!ssh_verifiable_kind("x509"),
"the bootstrap certificate is not tried as an ssh key"
);
assert!(!ssh_verifiable_kind(""));
}
/// A `PushOutcome` carrying whatever pack the caller wants to ask about.
fn outcome_with(pack: Option<IndexedPack>) -> PushOutcome {
PushOutcome {
commands: Vec::new(),
statuses: Vec::new(),
unnameable: Vec::new(),
push_options: Vec::new(),
pack,
pack_bytes: Some(0),
new_objects: Ok(Vec::new()),
atomic: false,
unpack: Ok(()),
}
}
/// `stored_bytes: None` on purpose: these arms are about what
/// `measured_pack_bytes` does when the ENGINE reported no count, which is
/// the only case where stat'ing a path is still the answer.
fn indexed(data_path: Option<std::path::PathBuf>) -> IndexedPack {
indexed_with(data_path, None)
}
fn indexed_with(
data_path: Option<std::path::PathBuf>,
stored_bytes: Option<u64>,
) -> IndexedPack {
IndexedPack {
num_objects: 1,
pack_id: ObjectId::empty_blob(gix_hash::Kind::Sha1),
index_id: ObjectId::empty_blob(gix_hash::Kind::Sha1),
pack_version: PackVersion::V2,
hash_kind: gix_hash::Kind::Sha1,
index_path: None,
data_path,
keep_path: None,
stored_bytes,
}
}
/// **An engine that writes no packfile still knows how big the pack was.**
///
/// The znippy store appends the received bytes verbatim and writes no
/// `.pack`, so `data_path` is `None` — and every push into the engine the
/// appliance ships logged `pack_bytes_unknown="pack … was indexed without a
/// directory, so there is nothing on disk to measure"`. That sentence was
/// FALSE: the store counted the bytes it appended
/// (`PushReceipt::pack_bytes`) and nobody carried the number this far. An
/// absence rendered where a measurement exists is the same defect as the
/// reverse, and it is the one this whole family of tests is about.
///
/// Both directions, or a `stored_bytes` that were ignored would still pass
/// the first line: a store that reports NEITHER a file nor a size is still
/// an honest unknown, and must still say so.
#[test]
fn a_pack_with_no_file_is_measured_by_the_store_that_stored_it() {
let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed_with(None, Some(4096)))));
assert_eq!(
bytes,
Some(4096),
"the store's own count is the measurement"
);
assert_eq!(
why, None,
"a measured pack carries no reason for not being measured"
);
let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed_with(None, None))));
assert_eq!(
bytes, None,
"neither a file nor a count is genuinely unknown"
);
assert!(
why.as_deref().unwrap_or_default().contains("no size"),
"and the reason must say the store reported no size, not merely that there is no \
directory: {why:?}"
);
}
/// **The defect this replaced.** `stat` failing rendered as `pack_bytes=0`,
/// which is indistinguishable from a push that genuinely carried nothing.
///
/// Seen red: with the body restored to
/// `.and_then(|path| std::fs::metadata(path).ok()).map(len).unwrap_or(0)`
/// this fails with *"a pack whose bytes could not be read reported
/// Some(0)"* — the old code's exact answer.
#[test]
fn a_pack_that_cannot_be_stated_reports_no_number_and_names_the_path() {
let missing = std::path::PathBuf::from("/nonexistent/gunnar/pack-deadbeef.pack");
let outcome = outcome_with(Some(indexed(Some(missing.clone()))));
let (bytes, why) = measured_pack_bytes(&outcome);
assert_eq!(
bytes, None,
"a pack whose bytes could not be read reported {bytes:?}; the only honest answer is \
no number at all, because 0 is a legal pack size for a deletions-only push"
);
let why = why.expect("an unreadable pack must say why it is unreadable");
assert!(
why.contains(missing.to_str().unwrap()),
"the reason must name the path an operator has to go and look at, got {why:?}"
);
}
/// The one arm that IS a zero, and must stay one: a deletions-only push
/// carries no pack, and reporting that as *unknown* would be the mirror
/// mistake.
#[test]
fn a_deletions_only_push_reports_a_measured_zero_and_no_reason() {
let (bytes, why) = measured_pack_bytes(&outcome_with(None));
assert_eq!(
bytes,
Some(0),
"a push with no pack carries zero pack bytes"
);
assert_eq!(why, None, "zero bytes is a measurement, not an absence");
}
/// **Three different answers over the same empty list**, which is the whole
/// reason the count takes a `Result` and not a slice.
///
/// * not enumerated, 3-object pack → 3. Seen red on the rig's SSE feed
/// ("pushed 0 objects (0 B)" for a push that carried a commit).
/// * not enumerated, no pack at all → 0, because there is nothing to count.
/// * ENUMERATED and empty → 0 even beside a 3-object pack. That is a
/// measured zero: the store said which objects the push introduced and
/// the answer was none. Under the old `is_empty()` fallback this case
/// reported 3, i.e. the record disagreed with the store that had been
/// asked. It is the arm that makes this test unable to pass against the
/// code it replaces.
#[test]
fn the_object_count_reads_the_reason_not_the_length() {
let mut pack = indexed(None);
pack.num_objects = 3;
let three = outcome_with(Some(pack));
let none = outcome_with(None);
let unknown = Err("the policy has no gate over objects");
assert_eq!(attested_object_count(unknown, &three), 3);
assert_eq!(attested_object_count(unknown, &none), 0);
assert_eq!(
attested_object_count(Ok(&[]), &three),
0,
"an enumerated empty set is a MEASURED zero and must not be overruled by the pack"
);
let oid = gix_hash::ObjectId::null(gix_hash::Kind::Sha1);
assert_eq!(attested_object_count(Ok(&[oid, oid]), &none), 2);
}
/// A pack indexed with no directory wrote nothing, so there is nothing to
/// measure — and that is not zero either.
#[test]
fn a_pack_never_written_to_disk_reports_no_number() {
let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed(None))));
assert_eq!(bytes, None);
assert!(
why.expect("must say why").contains("without a directory"),
"the reason must distinguish 'never written' from 'written and unreadable'"
);
}
/// The number is the file's real length, not a plausible one. Seen red by
/// writing 7 bytes and asserting 6.
#[test]
fn a_pack_on_disk_reports_the_bytes_the_file_actually_holds() {
let dir = tempfile::tempdir().expect("a temp dir");
let path = dir.path().join("pack-x.pack");
std::fs::write(&path, b"1234567").expect("write");
let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed(Some(path)))));
assert_eq!(bytes, Some(7), "the length must come off the filesystem");
assert_eq!(why, None);
}
/// `wire_bytes` had the same `unwrap_or(0)`. `PushFacts::pack_bytes`
/// documents `None` as "genuinely unavailable" and `Some(0)` as a
/// deletions-only push; the log collapsed both onto `0`.
#[test]
fn an_uncounted_wire_is_not_a_wire_that_carried_nothing() {
let (bytes, why) = measured(None, "the ingest did not report");
assert_eq!(bytes, None, "an uncounted wire must not render as 0 bytes");
assert_eq!(why.as_deref(), Some("the ingest did not report"));
let (bytes, why) = measured(Some(0), "unused");
assert_eq!(bytes, Some(0), "a counted zero stays a zero");
assert_eq!(why, None, "a counted zero carries no reason");
}
/// The ssh-sig gate is handed **only unexpired** keys.
///
/// The defect, found 2026-08-05: `key_directory` loaded `AuthTable::keys()`,
/// which filters nothing, so a key past its `expires` could no longer open an
/// SSH session and could still satisfy `require_signature` on a push. One
/// credential with two lifetimes.
///
/// `gunnar-core`'s `an_expired_key_signs_nothing_and_is_still_listed` guards
/// the *predicate*. This guards the **wiring**, which is the half that was
/// actually wrong: the predicate already existed and worked, and this call
/// site simply did not ask it. A guard on the predicate alone would have been
/// green over the whole bug.
///
/// # Seen red
///
/// With `signing_keys()` changed back to `keys()`, `all_keys()` has length 1
/// and the first assertion fails. The valid-key half is the control: it fails
/// if `key_directory` returns an empty directory for any other reason —
/// an unreadable table, a key that will not parse — which would otherwise
/// make the first assertion pass for the wrong reason.
/// **Un-ignored 2026-08-10.** The reason it was skipped — *"this crate
/// deliberately carries no storage engine"* — stopped being true when the
/// engine became a feature of this crate with znippy as the default. There
/// is no engine-free build of `gunnar-server` any longer to skip it for.
#[test]
fn the_signature_gate_is_handed_no_expired_key() {
// The known-answer key `gunnar-core`'s own tests use, and a second one, so
// the two principals below hold different key material.
const EXPIRED_KEY: &str = "ssh-ed25519 \
AAAAC3NzaC1lZDI1NTE5AAAAIMW4aqVr75w+3DWorOCbfwl5XCDCq1yCQ6puzU4qzol2 gunnar-kat";
let minted = gunnar_core::mint_ed25519().expect("mint a key");
let live_key = gunnar_core::ssh_key::PrivateKey::from_openssh(&minted)
.expect("parse the minted key")
.public_key()
.to_openssh()
.expect("render the public half");
let dir = tempfile::tempdir().expect("a temp dir");
let stores = ServerStores::open(dir.path()).expect("open the stores");
for (id, key, expires) in [
// Already expired: the unix epoch.
("stale", EXPIRED_KEY, Some(1_000)),
("live", live_key.as_str(), None),
] {
stores
.auth()
.add_principal(
gunnar_core::Principal::new(id, id, gunnar_core::PrincipalKind::Machine)
.expect("a valid id"),
)
.expect("add the principal");
stores
.auth()
.register_key(id, key, expires)
.expect("register the key");
}
let directory = key_directory(&stores);
assert!(
directory.keys_for("stale").is_empty(),
"an expired key must not be offered to the signature gate"
);
// The control. Without it an always-empty directory — an unreadable
// table, a parse failure, a `key_directory` that returns early — passes
// the line above while proving nothing.
assert_eq!(
directory.keys_for("live").len(),
1,
"an unexpired key must still be offered, or this test proves nothing"
);
assert_eq!(
directory.all_keys().len(),
1,
"exactly the unexpired key, and nothing else, reaches the gate"
);
}
}