gunnar_ssh/

seam.rs

//! **The one remaining seam: temporary, and marked so.**
//!
//! # What used to be here, and why it is gone
//!
//! This module carried three traits, because `gunnar-core` (the ACL model) and
//! `gunnar-wire` (the git wire protocol) were being written in parallel with this crate and
//! blocking on either would have meant writing no server at all.
//!
//! **`gunnar-core` has landed, so two of them are deleted.** `AuthDirectory` and
//! `RepoResolver` are now [`crate::control::Control`] over the real
//! [`AuthTable`](gunnar_core::AuthTable) and [`RepoCatalog`](gunnar_core::RepoCatalog),
//! and their test doubles went with them. They were not thin adapters worth
//! keeping: both returned `Option`, and `gunnar-core` deliberately returns
//! `Result` because the table lives in storage and a read can fail. Adapting one
//! to the other means choosing an answer for the failure case, and the
//! safe-looking choice (`None`) turns a storage fault into a silent permission
//! denial. That is precisely what the ACL model says not to do, and it is what the
//! seam was doing.
//!
//! **`gunnar-wire` has not landed**, so [`GitTransport`] stays exactly as it
//! was. It is the seam the protocol state machines plug into, and wiring it is
//! a separate job.
//!
//! | trait | delegates to | status |
//! |---|---|---|
//! | ~~`AuthDirectory`~~ | `gunnar_core::AuthTable` | **deleted** |
//! | ~~`RepoResolver`~~ | `gunnar_core::RepoCatalog` | **deleted** |
//! | [`GitTransport`] | `gunnar-wire`'s `upload_pack` / `receive_pack` | still a seam |
//!
//! # The test double
//!
//! [`BannerTransport`] is a **test double, not a backend**. It exists so this
//! crate's tests can prove the SSH machinery end to end without `gunnar-wire`.
//! Nothing outside a test should construct it.

use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;

use gunnar_core::{Fingerprint, Level, Principal, Repo};
use tokio::io::{AsyncRead, AsyncWrite};

use crate::command::GitService;

/// A bidirectional byte stream: what an SSH channel is once the exec has been
/// accepted.
///
/// Blanket-implemented, so `Box<dyn DuplexStream>` accepts russh's
/// `ChannelStream` and, in a test, a `tokio::io::DuplexStream`.
pub trait DuplexStream: AsyncRead + AsyncWrite + Unpin + Send {}

impl<T: AsyncRead + AsyncWrite + Unpin + Send> DuplexStream for T {}

/// Which part of a git conversation one call to the seam carries.
///
/// # Why an SSH crate names an HTTP shape
///
/// This module is **the seam**, not the SSH transport: it is the join every
/// front end plugs the protocol state machines into, and `gunnar-http` plugs
/// into this one rather than growing a second copy of it. That is the whole
/// reason the crate docs above call `GitTransport` "the seam" and not "the SSH
/// bridge".
///
/// The alternative was a second trait in `gunnar-http` with its own
/// `ServiceRequest`, its own authorisation call and its own join to
/// `gunnar-wire` — which is exactly the shape that produces two answers to
/// *"may this principal push"*. Sharing one seam costs an enum with three
/// variants and no dependencies; `gunnar-http` linking `gunnar-ssh` adds
/// nothing to any build, because a gunnar server that speaks HTTP also speaks
/// SSH and `gunnar-ssh` was already there.
///
/// If a build ever wants HTTP **without** SSH, this module and
/// [`crate::server::Dispatch`] move to their own crate and nothing else
/// changes. That build does not exist today, and inventing the crate for it
/// now would be inventing a boundary before there is anything on both sides of
/// it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Delivery {
    /// One connection carries the advertisement, the negotiation and the pack.
    /// This is SSH, and it is what `gunnar-wire`'s `Session::Connected` means.
    #[default]
    Session,
    /// Smart HTTP `GET /<path>/info/refs?service=…`: write the advertisement
    /// and stop. Nothing is read from the client.
    Advertisement,
    /// Smart HTTP `POST /<path>/<service>`: one stateless-rpc round.
    StatelessRpc,
}

/// Everything the wire needs to know about the session it is serving.
#[derive(Debug, Clone)]
pub struct ServiceRequest {
    /// Which service the client asked for.
    pub service: GitService,
    /// The repository, as the catalog holds it, carrying the hash kind the wire
    /// has to echo in `object-format` (the git wire protocol) and the default branch.
    ///
    /// When [`Self::prospective`] is set this is the repository the
    /// first-repository door **would** create — the same value
    /// `RepoCatalog::create` would write — and it is not in the catalog yet.
    pub repo: Repo,
    /// **The first-repository door is open for this push, and nothing has
    /// been written.** Only ever `true` for [`GitService::ReceivePack`] or an
    /// LFS upload (`GitService::LfsTransfer(Upload)`, 2026-09-28) to a
    /// repository that does not exist, by a principal who may create it. An
    /// LFS upload's repository is made when its first object arrives. The
    /// transport advertises `repo` as an empty repository, reads the client's
    /// commands, checks the hash, and only then asks
    /// `Control::admit_push_create` to write the row and makes the store. A
    /// client that hangs up at the advertisement leaves no trace.
    pub prospective: bool,
    /// Who is asking. Carried into the push attestation and the
    /// event stream.
    pub principal: Principal,
    /// The level the principal was found to hold, which the wire may need for
    /// finer decisions than "may push at all".
    pub access: Level,
    /// **The key this session actually authenticated with**.
    ///
    /// `None` means no key did, the anonymous read-only path, which
    /// is a real state and not a missing value, so it is an `Option` rather
    /// than an empty string.
    ///
    /// The server computed this in `auth_publickey` and looked the principal up
    /// by it, then used to drop it. Downstream, `gunnar-cli`'s push attestation
    /// and its `RefUpdated` / `RefDeleted` events had to *guess* it back out of
    /// the identity table, and could only do so when the principal held exactly
    /// one key, so a real deploy account with two keys published an empty
    /// fingerprint, and the audit trail could not answer "which key pushed
    /// this".
    ///
    /// It is a **parameter of [`crate::Dispatch::authorise`]**, not something
    /// attached afterwards like [`ServiceRequest::with_git_protocol`], and the
    /// difference is deliberate: an attach-after-the-fact field can be
    /// forgotten, and a forgotten one is silently `None`, which is exactly the
    /// shape of the bug this closes. The compiler asks for it instead.
    /// `git_protocol` is the other way round for the opposite reason: it comes
    /// from the *client*, and must not be able to reach the grant check.
    pub key_fingerprint: Option<Fingerprint>,
    /// The raw value of the client's `GIT_PROTOCOL` environment variable, if it
    /// sent one and the server's allowlist accepted it (see
    /// [`crate::server::ENV_ALLOWLIST`]).
    ///
    /// **Raw, not parsed**, and deliberately: the grammar is
    /// `gunnar-wire`'s (`ProtocolVersion::from_git_protocol`), and this crate
    /// does not depend on `gunnar-wire`. Parsing it here would be a second
    /// implementation of a version negotiation that already exists, and the two
    /// could disagree about `version=1` (one implementation).
    ///
    /// `None` means the client sent nothing, which is v0. It is an `Option`
    /// rather than a defaulted `ProtocolVersion` so that "the client asked for
    /// v0" and "the client never got to ask" stay distinguishable in a log.
    pub git_protocol: Option<String>,
    /// Which part of the conversation this call carries.
    ///
    /// [`Delivery::Session`] for SSH, always. Attached afterwards like
    /// [`ServiceRequest::git_protocol`] rather than taken as a parameter of
    /// [`crate::server::Dispatch::authorise`], and unlike the key fingerprint
    /// that is safe here for a specific reason: **forgetting it fails loudly.**
    /// An HTTP `POST` served as a `Session` writes a ref advertisement in front
    /// of the negotiation, and `git` rejects that on the first packet. The
    /// fingerprint's failure mode is the opposite — a forgotten one is silently
    /// `None` and shows up months later as an empty field in an audit trail —
    /// which is why that one is a parameter and this one is not.
    pub delivery: Delivery,
    /// **Which credential authenticated this request**, as the audit trail
    /// should name it: `key:SHA256:…` or `token:<hash prefix>`.
    ///
    /// `None` means a caller that presented no credential at all — the
    /// anonymous read path — and it is the ONLY thing `None` means here.
    ///
    /// It exists because [`Self::key_fingerprint`] cannot mean that (GUN-014).
    /// HTTP authenticates with a **token**, so there is no key to put in that
    /// field and it is `None` on every HTTP request, `git-receive-pack`
    /// included — while the code that consumes it states that an empty
    /// fingerprint has "one reason only — the anonymous, read-only path… That
    /// path cannot push". A push therefore reached `RefUpdated`, `RefDeleted`
    /// and the attestation with an empty fingerprint and no other field to say
    /// who had made it.
    ///
    /// A builder and not a parameter, unlike the fingerprint, and the reason
    /// is the opposite of that one's: the SSH front door does not set it at
    /// all. `key_fingerprint` is already the answer there, and deriving
    /// `key:<fp>` in one place (`transport::session_credential`) is one
    /// implementation rather than two front doors spelling the same prefix.
    pub credential: Option<String>,
    /// **The gate chain's warning line**: a Grace tenant's push is allowed and
    /// told `plan expired <date>; renew at <url>`. The transport puts it on the
    /// sideband so `git push` prints `remote: …`. `None` for every other state,
    /// for every read, and on a community build (whose one gate never warns).
    pub plan_notice: Option<String>,
    /// **The gate chain's byte ceiling for this push**, the smallest any gate
    /// asked for. The transport hands it to `gunnar-wire`, which counts the
    /// pack off the wire and cuts the transfer past it — see
    /// [`crate::PackCeiling`] and [`crate::PlanGate::pack_ceiling`].
    ///
    /// `None` for every read, for every write on a namespace with no byte cap,
    /// and on a community build (whose one gate asks for no ceiling), and
    /// `None` then means what it has always meant: read the pack to its end.
    ///
    /// Decided in `authorise_service` rather than looked up in the transport
    /// for the same reason [`ServiceRequest::plan_notice`] is: the gate chain
    /// lives on the front door's `Control`, and the transport has never been
    /// able to see it.
    ///
    /// # What reads it today, said plainly
    ///
    /// **The receive path, and only the receive path.** The chain asks for a
    /// ceiling wherever it asks `pack_quota`, which includes an LFS upload — and
    /// the LFS transfer ignores this field, deliberately. What LFS needed was
    /// its own cut against its own meter (`lfs_bytes`, not `pack_bytes`), and
    /// since 2026-09-07 it has one: [`ServiceRequest::lfs_allowance`]. Half a
    /// ceiling carried here would have been the twin of a limit that fires on
    /// one transport and not the other. A `Machine` principal is never asked at
    /// all ([`crate::PlanGate::applies_to_machine`]), so a primary's replication
    /// push is never cut by a tenant's plan.
    pub pack_ceiling: Option<crate::plan::PackCeiling>,
    /// **Where this push stands in the namespace's ledger of pack bytes** — the
    /// value the `pack_bytes` counter had when
    /// [`crate::server::Dispatch::ask_plan`] measured, taken BEFORE any gate
    /// walked the disk. `Some` exactly when [`Self::pack_ceiling`] is.
    ///
    /// `gunnar-server` binds the two together and hands the result to
    /// `gunnar-wire`, whose receive loop charges the ledger for every chunk
    /// before it takes it — so the ceiling is the room the whole namespace has
    /// left, not the room each connection is separately told it has.
    ///
    /// # Why this is a field beside the ceiling and not a field INSIDE it
    ///
    /// The LFS wire folded its mark into the number
    /// ([`gunnar_lfs::NamespaceAllowance`]) because there the two are used
    /// together by one callee. Here they are not: the number goes into
    /// `gunnar_wire::PackCeiling`, in a crate that cannot name a
    /// `gunnar_lfs::Mark` and must not learn to, and the mark goes into the
    /// same struct as a `dyn gunnar_wire::PackLedger`. They are joined at
    /// `gunnar_server::transport`, the seam that already exists for exactly
    /// this conversion, so folding them together here would be folding them for
    /// the journey and nothing else.
    ///
    /// The other half of the reason is whose fact it is. [`crate::PackCeiling`]
    /// is what a [`crate::PlanGate`] ANSWERED, and `PlanGate` is a public trait
    /// anyone may implement; the mark is the SESSION's standing in a counter
    /// this crate owns, taken before any gate is consulted and existing whether
    /// or not one answers. Putting it in the gate's return type would have
    /// asked every implementor to produce something only the dispatch can, and
    /// changed the type of a field the server's tests read.
    pub pack_mark: Option<gunnar_lfs::Mark>,
    /// **The gate chain's LFS room for this transfer**, the smallest any gate
    /// admitted — the twin of [`ServiceRequest::pack_ceiling`] one wire over.
    /// The transport hands it to `gunnar_lfs::Session` (SSH) and to
    /// [`LfsAccess`] (smart HTTP), and an upload past it is refused 507 with the
    /// plan's own sentence, before the object's bytes.
    ///
    /// `None` for every read, for every namespace with no `lfs_bytes` cap, and
    /// on a community build — and `None` means what the pack ceiling's `None`
    /// means: take what arrives.
    ///
    /// The pack ceiling would not have done: they are different meters on the
    /// signed fact, they are reported separately by `Usage.Read`, and one is
    /// counted in a receive loop while the other is declared up front. See
    /// [`crate::PlanGate::lfs_allowance`].
    ///
    /// # Why a `NamespaceAllowance` and not the plain `gunnar_lfs::Allowance`
    ///
    /// Because a plain one is a number measured for ONE transfer, and two
    /// transfers that started together measured the same "before". This one
    /// carries, beside the number, where the transfer stands in the namespace's
    /// ledger of bytes admitted since — so the door counts what the other
    /// connections are already doing. `gunnar-server` still decides the number
    /// (`PlanGate::lfs_allowance` returns the plain `Allowance` and is
    /// unchanged); [`crate::server::Dispatch::ask_plan`] binds it to the
    /// ledger, which is also the one place that can guarantee the mark is taken
    /// BEFORE the disk is walked. See `gunnar_lfs::ledger`.
    pub lfs_allowance: Option<gunnar_lfs::NamespaceAllowance>,
    /// **Set when the client is gone and the transfer must stop** at its next
    /// safe point (`crate::channel::serve_until_session_ends`). A transport
    /// whose git work runs off the async task checks it there, so the work
    /// ends before the fan worker serving it is given back.
    pub cancel: crate::channel::Cancel,
}

impl ServiceRequest {
    /// Attach the client's `GIT_PROTOCOL` value.
    ///
    /// Separate from the authorisation decision on purpose: what a client may
    /// do is decided from its key and the catalog, and an environment variable
    /// must never be able to reach that decision. This carries a *hint about
    /// framing* to the wire, nothing more.
    pub fn with_git_protocol(mut self, value: Option<String>) -> Self {
        self.git_protocol = value;
        self
    }

    /// Say which part of the conversation this call carries. See
    /// [`Delivery`].
    pub fn with_delivery(mut self, delivery: Delivery) -> Self {
        self.delivery = delivery;
        self
    }

    /// Name the credential this request authenticated with. See
    /// [`ServiceRequest::credential`].
    pub fn with_credential(mut self, credential: Option<String>) -> Self {
        self.credential = credential;
        self
    }
}

/// What went wrong while serving.
#[derive(Debug, thiserror::Error)]
pub enum TransportError {
    /// The protocol state machine refused the request.
    #[error("protocol error: {0}")]
    Protocol(String),
    /// The stream failed.
    #[error("i/o error: {0}")]
    Io(#[from] std::io::Error),
    /// Not wired yet. **This is the honest answer until `gunnar-wire` lands**,
    /// and it must never quietly become a successful empty response: a clone
    /// that returns no objects and exits zero looks exactly like a working
    /// server serving an empty repository (measured, not asserted).
    #[error("not implemented: {0}; the wire seam is not connected to gunnar-wire yet")]
    Unimplemented(&'static str),
    /// The transfer was refused **before a single byte reached the client**, and
    /// the string is the sentence the client is owed.
    ///
    /// This variant exists because the other three could not carry one. A
    /// transfer that failed after the advertisement went out cannot be answered
    /// with a fresh pkt-line — the client is mid-pack and the framing belongs to
    /// `gunnar-wire`'s sideband, which is what `UploadPack::report_error` is
    /// for. A transfer that failed *before* the first write can be answered, and
    /// until this variant existed it was not: the session logged the whole
    /// reason and sent the client an exit status and no words, which git renders
    /// as `the remote end hung up unexpectedly`.
    ///
    /// So the variant is the *fact* "nothing has been written yet", not a
    /// severity. Only a construction site that knows that may use it, which is
    /// why it is a distinct variant rather than a flag on
    /// [`TransportError::Protocol`].
    #[error("{0}")]
    Refused(String),
}

impl TransportError {
    /// What the client is told, or `None` when the client cannot be told
    /// anything.
    ///
    /// Named after [`crate::server::DispatchError::client_message`] and
    /// `gunnar_wire::Error::client_message` because it is the same contract in a
    /// third place (one implementation): a refusal the operator can see and the user cannot
    /// is half a fix.
    ///
    /// `None` for [`TransportError::Protocol`] and [`TransportError::Io`]
    /// **on purpose**, and it is the conservative answer rather than the
    /// convenient one: those two are raised from inside a transfer that may
    /// already have written an advertisement, a `NAK` or half a pack, and a raw
    /// pkt-line appended to that stream is corruption, not an error message.
    pub fn client_message(&self) -> Option<String> {
        match self {
            TransportError::Protocol(_) | TransportError::Io(_) => None,
            // Nothing has been served, by definition of the variant, so the
            // client can be told — and must be, because the alternative is a
            // clone that exits non-zero with no reason at all.
            TransportError::Unimplemented(_) => Some(self.to_string()),
            TransportError::Refused(message) => Some(message.clone()),
        }
    }
}

/// An authorised repository's LFS object store, and what it will accept.
///
/// Returned by [`LfsTransport::open_lfs`], and deliberately not a session: LFS
/// over HTTP is four independent requests (`batch`, `GET`, `PUT`, `verify`),
/// each authorised on its own, so there is nothing for a session to hold
/// between them. What the two surfaces share is the *store* and the *limits*,
/// which is exactly what this carries.
pub struct LfsAccess {
    /// Where the objects are. `Arc` because the caller runs the actual I/O on a
    /// blocking thread and has to own it there.
    pub store: std::sync::Arc<dyn gunnar_lfs::LfsStore>,
    /// The quota and the batch cap in force, from the **same**
    /// `WireConfig::lfs` the SSH session is built with. A second home for these
    /// is how 8 GiB comes to be enforced on one transport and not the other.
    pub limits: gunnar_lfs::Limits,
    /// **The tenant's remaining LFS room**, decided by the same
    /// [`crate::PlanGate`] chain that decided
    /// [`crate::ServiceRequest::pack_ceiling`] for a push, and `None` wherever a
    /// push would get no ceiling: a community build, an uncapped namespace, a
    /// download.
    ///
    /// Beside [`Self::limits`] rather than inside it because they are different
    /// kinds of number. `limits` is the operator's, one set for the whole box,
    /// `Copy`, and comes off the command line; this is one namespace's plan,
    /// re-measured per transfer, and carries a sentence built from a signed
    /// entitlement fact. A tenant who fills it is told to buy more; a client
    /// that trips `limits` is told the box does not take objects that big.
    pub allowance: Option<gunnar_lfs::NamespaceAllowance>,
}

impl std::fmt::Debug for LfsAccess {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LfsAccess")
            .field("limits", &self.limits)
            .finish_non_exhaustive()
    }
}

/// Open the LFS object store of a repository a caller has **already been
/// authorised for**.
///
/// # Why this is a seam and not a store handed to `gunnar-http` directly
///
/// The same reason [`GitTransport`] is. `gunnar-http` knows HTTP and the LFS
/// batch API; it must not know that a repository is a directory, that the
/// objects live under `<git_dir>/lfs/`, or that opening one can take a lock.
/// Those are `gunnar-cli`'s, which owns `ServerStores` and the one
/// `WireConfig` both transports are configured from.
///
/// # `request` is the output of the ACL, not an input to it
///
/// The parameter is a [`ServiceRequest`], which can only be got out of
/// [`crate::server::Dispatch::authorise_service`]. So an implementation cannot
/// be called without the grant check having already happened, and there is no
/// second authorisation path — the same property `GitTransport` has, held the
/// same way. `request.service` is `GitService::LfsTransfer(operation)` and the
/// operation in it is the one that demanded Read or Write.
///
/// # Blocking
///
/// Implementations do filesystem work. The caller runs them on a blocking
/// thread; nothing here is async, for the same reason `gunnar_lfs::Session` is
/// not.
pub trait LfsTransport: Send + Sync + 'static {
    /// The object store for `request`'s repository.
    fn open_lfs(&self, request: &ServiceRequest) -> Result<LfsAccess, TransportError>;
}

/// Serve `git-upload-pack` / `git-receive-pack` over an accepted channel.
///
/// This is `gunnar-wire`'s entry point seen from the SSH side. The returned
/// future owns the stream for the whole of the transfer; the server spawns it
/// rather than awaiting it inline, because russh dispatches handler calls from
/// the session's own read loop and awaiting there would stop the loop that has
/// to deliver the client's bytes.
pub trait GitTransport: Send + Sync + 'static {
    /// Serve one request to completion.
    fn serve<'a>(
        &'a self,
        request: ServiceRequest,
        stream: Box<dyn DuplexStream>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>>;
}

/// A [`GitTransport`] that writes one line describing the request it was given,
/// then reads the client out. **A test double.**
///
/// It exists to prove a specific thing that a mock recording a call cannot: that
/// the *byte stream* reached the seam, in both directions, after a real SSH
/// handshake. The line it writes names the service and the repository, so a test
/// asserting on it is asserting on applied output rather than on a flag someone
/// set (measured, not asserted).
#[derive(Debug, Default)]
pub struct BannerTransport {
    served: Mutex<Vec<String>>,
    git_protocols: Mutex<Vec<Option<String>>>,
    settle_after_release: bool,
    fail_at_the_end: bool,
    /// Set by [`BannerTransport::refusing`]: the transport writes **nothing** and
    /// returns [`TransportError::Refused`], which is the shape of every
    /// post-authorisation refusal — a held store lock, a repository that will not
    /// open, an unwired service.
    refuse_before_writing: Option<String>,
    /// Set by [`BannerTransport::holding`]: the transport writes its banner and
    /// then does not return until the paired [`TransportHold`] is released.
    hold: Option<tokio::sync::watch::Receiver<bool>>,
}

/// The release lever of [`BannerTransport::holding`].
///
/// Dropping it releases every held transfer, so a test that panics mid-way
/// cannot wedge a server task forever.
#[derive(Debug)]
pub struct TransportHold {
    tx: tokio::sync::watch::Sender<bool>,
}

impl TransportHold {
    /// Let every held transfer finish.
    pub fn release(&self) {
        let _ = self.tx.send(true);
    }
}

impl Drop for TransportHold {
    fn drop(&mut self) {
        self.release();
    }
}

impl BannerTransport {
    /// A transport that has served nothing yet.
    pub fn new() -> Self {
        Self::default()
    }

    /// The same transport, but it **drops the stream and then waits** before it
    /// returns.
    ///
    /// This is the red half of the exit-status ordering guard, and it is worth
    /// spelling out why it is not a contrivance.
    ///
    /// russh's `Channel::into_stream()` wraps the channel in
    /// `ChannelCloseOnDrop`, whose `Drop` **spawns a task** that sends
    /// `ChannelMsg::Close`. A server that hands the transport such a stream and
    /// sends `exit-status` afterwards is in a race with that task: the ordinary
    /// path wins it on a current-thread runtime (the send takes an available
    /// permit without yielding) and loses it often enough on a multi-threaded
    /// one to make a *successful* push report failure (measured 2026-08-03,
    /// `ssh` exit 255 with a complete 244-byte advertisement).
    ///
    /// A test on the ordinary path therefore proves nothing: it is green with
    /// the bug present. This mode gives the spawned task the scheduling it needs
    /// to run, which turns "usually" into "always" **in the failing direction
    /// only**. A server that closes nothing on drop is unaffected by any amount
    /// of waiting here, so a green under this transport is not a timing
    /// accident; it is the absence of the close.
    pub fn settling_after_it_releases_the_stream() -> Self {
        BannerTransport {
            settle_after_release: true,
            ..Self::default()
        }
    }

    /// A transport whose transfers **stay in flight** until the returned
    /// [`TransportHold`] is released.
    ///
    /// This is what makes the admission gate testable at all. A limit on
    /// *concurrent* transfers cannot be reached by a transport that finishes
    /// instantly: a hundred sequential clones never put two in flight together,
    /// and a test written against the ordinary banner would be asserting on
    /// scheduling luck. Here the first N transfers are genuinely simultaneous
    /// and stay that way until the test says otherwise.
    ///
    /// The banner is written **before** the wait, so a client knows its transfer
    /// was admitted rather than merely accepted.
    pub fn holding() -> (Self, TransportHold) {
        let (tx, rx) = tokio::sync::watch::channel(false);
        (
            BannerTransport {
                hold: Some(rx),
                ..Self::default()
            },
            TransportHold { tx },
        )
    }

    /// A transport that refuses **without writing a byte**, the way a transfer
    /// whose repository will not open does.
    ///
    /// Distinct from [`BannerTransport::that_then_fails`] on the one axis that
    /// matters: that one writes its banner first, so the client has already had
    /// output and the framing is no longer the session's to add to. This one has
    /// written nothing, which is the *only* condition under which the session may
    /// still hand the client a reason.
    pub fn refusing(message: impl Into<String>) -> Self {
        Self {
            refuse_before_writing: Some(message.into()),
            ..Self::default()
        }
    }

    /// …and then reports that the transfer failed.
    ///
    /// The other direction of the same guard: a **non-zero** status has to
    /// survive the stream being released too. Without this, the exit-status
    /// assertions could all be satisfied by a server that hard-coded zero.
    pub fn that_then_fails(mut self) -> Self {
        self.fail_at_the_end = true;
        self
    }

    /// Everything it has served, as `"<verb> <ns>/<repo> <principal-id>"`.
    pub fn served(&self) -> Vec<String> {
        self.served.lock().expect("banner transport lock").clone()
    }

    /// The `GIT_PROTOCOL` value that arrived with each served request, in the
    /// same order as [`BannerTransport::served`].
    ///
    /// Recorded separately from the banner so that a test can assert *what the
    /// wire would negotiate with* without changing the byte the existing
    /// end-to-end assertions read back over SSH.
    pub fn git_protocols(&self) -> Vec<Option<String>> {
        self.git_protocols
            .lock()
            .expect("banner transport lock")
            .clone()
    }

    /// The line this transport writes for `request`.
    pub fn banner(request: &ServiceRequest) -> String {
        format!(
            "{} {} {}\n",
            request.service.verb(),
            request.repo.key(),
            request.principal.id
        )
    }
}

impl GitTransport for BannerTransport {
    fn serve<'a>(
        &'a self,
        request: ServiceRequest,
        mut stream: Box<dyn DuplexStream>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move {
            use tokio::io::AsyncWriteExt;
            // Before the banner, before `served` is recorded: a refusal that had
            // already written or already counted itself would not be the thing
            // under test.
            if let Some(message) = &self.refuse_before_writing {
                return Err(TransportError::Refused(message.clone()));
            }
            let banner = BannerTransport::banner(&request);
            self.served
                .lock()
                .expect("banner transport lock")
                .push(banner.trim_end().to_string());
            self.git_protocols
                .lock()
                .expect("banner transport lock")
                .push(request.git_protocol.clone());
            stream.write_all(banner.as_bytes()).await?;
            stream.flush().await?;
            if let Some(hold) = &self.hold {
                // Wait, still holding the stream, which is what an admitted,
                // in-flight transfer looks like to the admission gate.
                // A transfer told to stop (its client gone past the grace)
                // stops, as the real transport does at its next safe point.
                let mut hold = hold.clone();
                let released = async {
                    while !*hold.borrow_and_update() {
                        if hold.changed().await.is_err() {
                            break;
                        }
                    }
                };
                tokio::select! {
                    () = released => {}
                    () = request.cancel.cancelled() => {
                        return Err(TransportError::Io(std::io::Error::new(
                            std::io::ErrorKind::ConnectionAborted,
                            "the banner double was told to stop: its client is gone",
                        )));
                    }
                }
            }
            stream.shutdown().await?;
            if self.settle_after_release {
                // Explicit, so the drop happens here rather than at the end of
                // the block, and anything it spawned has somewhere to run.
                drop(stream);
                for _ in 0..16 {
                    tokio::task::yield_now().await;
                }
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }
            if self.fail_at_the_end {
                return Err(TransportError::Protocol(
                    "the banner double was asked to report a failed transfer".to_string(),
                ));
            }
            Ok(())
        })
    }
}