gunnar-lfs 1.6.0

//! One `git-lfs-transfer` connection, from the capability advertisement to
//! `quit`.
//!
//! Nothing here owns a socket: [`Session::serve`] takes `&mut dyn BufRead` and
//! `&mut dyn Write`, the same shape `gunnar-wire`'s state machines take, so it
//! can be driven from a unit test with no SSH in sight and from a russh channel
//! in production without knowing the difference.
//!
//! # The invariant this module exists to hold
//!
//! **No allocation is ever sized from an object.** Every byte of every object,
//! in both directions, moves through a buffer of [`CHUNK_BYTES`]. The size a
//! client declares is used to *refuse* and to *verify*, never to reserve. That
//! is asserted against real process RSS in `tests/streaming.rs`, because
//! "it streams" is precisely the claim a passing unit test cannot make.
//!
//! # Command order
//!
//! The first command **must** be `version`. The reference server accepts
//! commands in any order; gunnar does not, because a session that transferred
//! objects before any protocol version was agreed has no contract to have
//! served them under. The real client always negotiates first
//! (`PktlineConnection.Start`), so this costs nothing and closes a case nobody
//! would otherwise have decided.

use std::io::{BufRead, Write};

use gunnar_pktline::Packet;

use crate::allowance::Allowance;
use crate::batch::plan_object;
use crate::error::{Error, Result};
use crate::ledger::NamespaceAllowance;
use crate::limits::{Limits, CHUNK_BYTES};
use crate::oid::Oid;
use crate::operation::Operation;
use crate::store::LfsStore;
use crate::verify::VerifyingSink;
use crate::wire::{self, SectionEnd};

/// What a session did. Returned so the caller can log applied output, the
/// bytes and objects that really moved, rather than "a session happened".
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Summary {
    /// Commands dispatched, including the ones refused.
    pub commands: u64,
    /// `batch` requests answered.
    pub batches: u64,
    /// Objects named across every `batch`.
    pub batch_objects: u64,
    /// Objects streamed out by `get-object`.
    pub objects_sent: u64,
    /// Objects committed by `put-object`.
    pub objects_received: u64,
    /// Object payload bytes written to the client.
    pub bytes_sent: u64,
    /// Object payload bytes read from the client and committed.
    pub bytes_received: u64,
    /// Did the client say `quit`, or did the stream simply end?
    pub clean_quit: bool,
}

/// A `git-lfs-transfer` session over one repository.
pub struct Session<'a> {
    store: &'a dyn LfsStore,
    operation: Operation,
    limits: Limits,
    /// The tenant's remaining plan, or `None` on a box that caps nobody — the
    /// community appliance, and every namespace with no signed fact. See
    /// [`Allowance`] for the number and [`NamespaceAllowance`] for why the door
    /// needs to know what else is uploading against the same namespace.
    allowance: Option<NamespaceAllowance>,
    /// **Bytes this session has already committed against [`Self::allowance`].**
    ///
    /// **No longer what the DOOR reads.** `put-object` charges the namespace's
    /// ledger, which counts this session's own objects and every other
    /// transfer's, so the running total below is subsumed by a number that is
    /// also right when two sessions are open at once. What it is still for is
    /// the `batch` answer, which is advisory and per session by nature — see
    /// [`crate::batch::plan_object`].
    ///
    /// A `Cell` because [`Session::serve`] takes `&self` — the whole state
    /// machine does, so it can be driven from a test with no interior mutability
    /// anywhere — and because one session is one channel on one thread: there is
    /// no second writer to race with, and an `AtomicU64` would be a lock nobody
    /// contends spelled as if somebody might.
    ///
    /// It has to exist at all because a session is not one object. `put-object`
    /// may repeat until `quit`, so an allowance checked only per object would
    /// let a tenant with 1 GiB left upload twenty 8 GiB objects — each one
    /// individually under every limit on the box. `tests/namespace_allowance.rs`
    /// is the assertion that a per-object check could not pass.
    taken: std::cell::Cell<u64>,
}

impl<'a> Session<'a> {
    /// A session serving `store` for `operation`.
    ///
    /// **`operation` is the one the SSH exec line named**, which is the one the
    /// grant check in `gunnar_ssh::command` already used to demand Read or
    /// Write. It is passed in rather than re-parsed here so there is one
    /// authorisation decision, not two (one implementation).
    pub fn new(store: &'a dyn LfsStore, operation: Operation) -> Self {
        Session {
            store,
            operation,
            limits: Limits::default(),
            allowance: None,
            taken: std::cell::Cell::new(0),
        }
    }

    /// The same session with different limits.
    pub fn with_limits(mut self, limits: Limits) -> Self {
        self.limits = limits;
        self
    }

    /// The limits in force.
    pub fn limits(&self) -> Limits {
        self.limits
    }

    /// **The same session, bounded by the tenant's remaining plan.**
    ///
    /// Not a [`Limits`] field, though it is also a byte count that refuses an
    /// upload: `Limits` is `Copy`, one set of numbers for the whole box, and is
    /// carried on `WireConfig` from the operator's command line. This is a
    /// different number for every namespace, is re-measured for every transfer,
    /// and carries a sentence composed from a signed entitlement fact. Folding
    /// them into one struct would make the operator's flags look tenant-scoped
    /// and the tenant's plan look like a server default.
    ///
    /// Absent on a community build and on any namespace with no cap — see
    /// [`Allowance`] on why "no allowance" and "an allowance of 0" are
    /// deliberately different states.
    pub fn with_allowance(mut self, allowance: NamespaceAllowance) -> Self {
        self.allowance = Some(allowance);
        self
    }

    /// The tenant's remaining plan **as measured**, if this session has one —
    /// for a log line and for the advisory `batch` answer.
    ///
    /// Deliberately the inner [`Allowance`] and not the
    /// [`NamespaceAllowance`]: this is the reporting accessor, it cannot
    /// charge, and a caller about to take bytes must go through
    /// [`NamespaceAllowance::admit`] instead. Returning the number that was
    /// measured is also what a reader of `lfs_allowance_bytes=` in the
    /// transport's log wants — the plan's room, not the plan's room minus
    /// whatever a sibling connection happened to be doing at that instant.
    pub fn allowance(&self) -> Option<&Allowance> {
        self.allowance.as_ref().map(NamespaceAllowance::allowance)
    }

    /// Serve the connection to completion.
    ///
    /// Returns `Err` only for failures that end the session: I/O and framing.
    /// Everything else has already been answered with a status code on the wire
    /// and is counted in the [`Summary`].
    pub fn serve(&self, input: &mut dyn BufRead, output: &mut dyn Write) -> Result<Summary> {
        let mut summary = Summary::default();
        wire::write_capabilities(output)?;
        output.flush()?;

        let mut negotiated = false;
        loop {
            let packet = match gunnar_pktline::read_packet(input)? {
                Some(packet) => packet,
                // The client closed without `quit`. Legal, if impolite: git-lfs
                // does it when a transfer errors out. Not an error here.
                None => return Ok(summary),
            };
            let line = match packet {
                Packet::Data(data) => data,
                Packet::Flush => {
                    // A bare flush where a command belongs. Answered rather
                    // than treated as end-of-stream, matching the reference
                    // server, so a confused client gets a diagnostic.
                    wire::write_error(output, 400, "expected a command, got a flush-pkt")?;
                    output.flush()?;
                    continue;
                }
                other => {
                    return Err(Error::Framing(format!(
                        "unexpected {other:?} where a command belongs"
                    )))
                }
            };
            let line =
                String::from_utf8_lossy(line.strip_suffix(b"\n").unwrap_or(&line)).to_string();
            let (verb, argument) = match line.split_once(' ') {
                Some((verb, rest)) => (verb, rest),
                None => (line.as_str(), ""),
            };
            summary.commands += 1;

            if !negotiated && verb != "version" {
                tracing::warn!(
                    target: crate::LFS_TARGET,
                    %verb,
                    "a command arrived before the protocol version was negotiated"
                );
                wire::write_error(output, 400, "the first command must be `version`")?;
                output.flush()?;
                continue;
            }

            let outcome = self.dispatch(verb, argument, input, output, &mut summary);
            match outcome {
                Ok(Flow::Continue) => {
                    if verb == "version" {
                        negotiated = true;
                    }
                }
                Ok(Flow::Quit) => {
                    summary.clean_quit = true;
                    output.flush()?;
                    return Ok(summary);
                }
                Err(err) if err.is_fatal() => {
                    // A fatal failure that still knows what it decided says so
                    // before it goes. `write_error` is allowed to fail here and
                    // is not allowed to replace `err`: the commonest way to
                    // reach this line at all is a peer that has already gone,
                    // and "could not tell you about the real error" is not the
                    // fact worth keeping. See `Error::final_report`.
                    if let Some((status, message)) = err.final_report() {
                        let _ = wire::write_error(output, status, &message);
                        let _ = output.flush();
                    }
                    tracing::warn!(target: crate::LFS_TARGET, %verb, %err, "the lfs session failed");
                    return Err(err);
                }
                Err(err) => {
                    tracing::debug!(target: crate::LFS_TARGET, %verb, status = err.status(), %err, "refused");
                    wire::write_error(output, err.status(), &err.client_message())?;
                }
            }
            output.flush()?;
        }
    }

    fn dispatch(
        &self,
        verb: &str,
        argument: &str,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
        summary: &mut Summary,
    ) -> Result<Flow> {
        match verb {
            "version" => self.version(argument, input, output),
            "batch" => self.batch(input, output, summary),
            "get-object" => self.get_object(argument, input, output, summary),
            "put-object" => self.put_object(argument, input, output, summary),
            "verify-object" => self.verify_object(argument, input, output),
            // The LFS design's optional half. See `wire::CAPABILITIES` for why 501 and not
            // 405: git-lfs turns 501 into "locking is not implemented here" and
            // disables lock verification silently, which is the true statement.
            "lock" | "unlock" | "list-lock" | "list-locks" => {
                let _ = wire::read_args(input, self.limits.max_args, "a lock command")?;
                Err(Error::NotImplemented(format!(
                    "`{verb}` is not implemented: this server advertises no `locking` capability"
                )))
            }
            "quit" => {
                let _ = wire::read_args(input, self.limits.max_args, "quit")?;
                // No delim. `ReadStatus`, the only reader `quit` uses, calls a
                // delim `unexpected data`.
                wire::write_status(output, 200, &[])?;
                Ok(Flow::Quit)
            }
            // Deliberately not draining an unknown command's section: gunnar
            // cannot know whether it has one, and guessing wrong swallows the
            // next real command. The stray argument lines are each answered
            // `400` in turn and the stream resynchronises at the next flush.
            other => Err(Error::BadRequest(format!("unknown command `{other}`"))),
        }
    }

    /// A command whose section ended with the **wrong terminator**, answered
    /// with a 400 rather than killed with a [`Error::Framing`].
    ///
    /// `Framing` means *the pkt-line boundary is lost and there is no way to
    /// find the next command*. That was not true at any of the four call sites
    /// that used it here: `read_args` had already consumed the section up to
    /// and including its terminator and told the caller which one arrived, so
    /// the reader knew exactly where it stood. `Framing` is fatal, so a client
    /// that ended `version` with a delim-pkt got an EOF where the protocol
    /// defines a status code — the same shape as the quota refusal that
    /// `refuse!`'s drain error replaced.
    ///
    /// A delim-pkt promises a further section. It is read out here, because
    /// **that**, and not the error type, is what keeps the stream in sync; and
    /// if reading it out fails, sync really is lost and the fatal error is
    /// honest. `read_lines` is the same drain the `hash-algo` refusal already
    /// uses (one implementation).
    fn wrong_terminator(
        &self,
        input: &mut dyn BufRead,
        what: &'static str,
        wanted: SectionEnd,
        got: SectionEnd,
    ) -> Error {
        if got == SectionEnd::Delim {
            if let Err(err) = wire::read_lines(
                input,
                self.limits.max_batch_objects,
                "a section this command does not define",
            ) {
                return Error::Framing(format!(
                    "{what} ended with a delim-pkt and the section that followed could not be \
                     read out, so the stream is desynced: {err}"
                ));
            }
        }
        Error::BadRequest(format!(
            "{what} must end with a {}, not a {}",
            wanted.name(),
            got.name()
        ))
    }

    fn version(
        &self,
        argument: &str,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
    ) -> Result<Flow> {
        let (_, end) = wire::read_args(input, self.limits.max_args, "version")?;
        if end != SectionEnd::Flush {
            return Err(self.wrong_terminator(input, "version", SectionEnd::Flush, end));
        }
        if argument != wire::PROTOCOL_VERSION {
            return Err(Error::BadRequest(format!(
                "unsupported protocol version `{argument}`; this server speaks {}",
                wire::PROTOCOL_VERSION
            )));
        }
        wire::write_status_lines(output, 200, &[], &[])?;
        Ok(Flow::Continue)
    }

    /// Answer a `batch`.
    ///
    /// # A `download` batch answers `download` for every object, held or not
    ///
    /// This is the one place gunnar deliberately departs from
    /// `charmbracelet/git-lfs-transfer`, and it is not a guess. RESEARCHED
    /// 2026-08-05, against the specification, the client's source and the
    /// reference server's.
    ///
    /// **The pure-SSH grammar cannot express a per-object error.** From
    /// `docs/proposals/ssh_adapter.md`, which is where the pure-SSH protocol is
    /// specified — the HTTP `docs/api/batch.md` does not govern it:
    ///
    /// ```text
    /// batch-oid-line = PKT-LINE(oid size action *(key=[value]) LF)
    /// ```
    ///
    /// There is no `error` production, no `code`, nowhere to put one. The
    /// document closes the door twice more: *"If the server has no actions that
    /// are valid for an object, it should be listed once in the response with
    /// the `noop` action"*, and *"Unknown arguments should be ignored, as should
    /// unknown key-value pairs in the `oid-line` production"* — so a server that
    /// invented `error=404` would be **specified into being ignored**. The HTTP
    /// API's per-object `{"error":{"code":404}}` with the batch still 200
    /// (`docs/api/batch.md`) has no counterpart here. That is a real gap in the
    /// pure-SSH protocol, not a gap in gunnar.
    ///
    /// **And `noop` is read by the client as "nothing to do", not as "absent".**
    /// `tq/ssh.go`, parsing the response: `if entries[2] == "noop" { continue }`,
    /// which leaves the transfer with an empty `Actions` map; then
    /// `tq/transfer_queue.go`, dispatching it:
    ///
    /// ```text
    /// } else if a == nil && manifest.standaloneTransferAgent == "" {
    ///     q.Skip(o.Size)
    ///     q.wait.Done()
    /// }
    /// ```
    ///
    /// No error, no retry, no line printed. MEASURED against git-lfs 3.7.1: a
    /// `git lfs fetch --all` for two objects of which gunnar held one printed
    /// **"2 objects found, done."** and exited **0**, having fetched one. The
    /// reference server does exactly this —
    /// `BatchData(DownloadOperation, "download", "noop")` — so every deployed
    /// LFS-over-SSH server has the same hole.
    ///
    /// **`get-object` is where the protocol can still tell the truth.** It has a
    /// `status-error-response`, and the client turns any non-2xx from it into a
    /// reported failure for that object alone. So a download batch hands back
    /// the `download` action unconditionally and [`Session::get_object`] answers
    /// `404` for what is not here. That reaches, over SSH, precisely the outcome
    /// git-lfs's own suite demands over HTTP (`t/t-fetch.sh`, "fetch with
    /// missing object"): *"should return non-zero, but should also download all
    /// the other valid files too"*. Pinned in
    /// `tests/interop.rs::a_fetch_of_an_absent_object_fails_and_still_delivers_the_present_one`.
    ///
    /// The cost, stated plainly: the client wraps a failed `get-object` in
    /// `errors.NewRetriableError` unconditionally (`tq/ssh.go`), so it asks for
    /// an absent object up to `lfs.transfer.maxretries` times — 8 by default —
    /// before reporting it. Eight cheap round trips on an already-open channel,
    /// in exchange for the fetch not lying. Refusing the whole batch instead
    /// would report the failure in one trip and cost the client every *other*
    /// object in it, which is the trade the HTTP API exists to avoid.
    ///
    /// An **upload** batch is unchanged, and `noop` there is correct in its own
    /// right: the server holds the bytes, so there is genuinely no action.
    fn batch(
        &self,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
        summary: &mut Summary,
    ) -> Result<Flow> {
        let (args, end) = wire::read_args(input, self.limits.max_args, "the batch arguments")?;
        if end != SectionEnd::Delim {
            return Err(self.wrong_terminator(
                input,
                "the batch arguments",
                SectionEnd::Delim,
                end,
            ));
        }
        // The client sends `hash-algo=sha256` and refuses any other value in the
        // response. An LFS oid is SHA-256 by definition; there is nothing else
        // to agree about, and a server that quietly accepted `sha1` here would
        // be filing objects under ids no client computes.
        match wire::arg(&args, "hash-algo") {
            None | Some("") | Some("sha256") => {}
            Some(other) => {
                let _ = wire::read_lines(input, self.limits.max_batch_objects, "the batch objects");
                // 409, the code `docs/api/batch.md` names for exactly this.
                // See [`Error::Conflict`].
                return Err(Error::Conflict(format!(
                    "unsupported hash algorithm `{other}`; lfs oids are sha256"
                )));
            }
        }

        let lines = wire::read_lines(input, self.limits.max_batch_objects, "the batch objects")?;
        summary.batches += 1;
        summary.batch_objects += lines.len() as u64;

        let mut response = Vec::with_capacity(lines.len());
        for line in &lines {
            let mut fields = line.split(' ');
            let (Some(oid_text), Some(size_text)) = (fields.next(), fields.next()) else {
                return Err(Error::BadRequest(format!(
                    "malformed batch line `{}`",
                    truncate(line)
                )));
            };
            let oid = Oid::parse(oid_text)
                .map_err(|err| Error::BadRequest(format!("batch line: {err}")))?;
            let size: u64 = size_text.parse().map_err(|_| {
                Error::BadRequest(format!(
                    "batch line: `{}` is not a size",
                    truncate(size_text)
                ))
            })?;

            // **One decision, two wires.** The quota, the store lookup and the
            // ordering between them live in `crate::batch`, which the HTTP
            // batch API routes through as well (one implementation). What differs is only
            // what each grammar can *say* about the answer: this one has
            // `upload | download | noop` and nothing else, so `AlreadyPresent`
            // and `Missing` both come out `noop` here. See `batch::Plan`.
            let plan = plan_object(
                self.store,
                &self.limits,
                self.allowance(),
                self.taken.get(),
                self.operation,
                &oid,
                size,
            )?;
            response.push(format!("{oid} {size} {}", plan.ssh_action(self.operation)));
        }

        wire::write_status_lines(output, 200, &[], &response)?;
        Ok(Flow::Continue)
    }

    fn get_object(
        &self,
        argument: &str,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
        summary: &mut Summary,
    ) -> Result<Flow> {
        let (_, end) = wire::read_args(input, self.limits.max_args, "get-object")?;
        if end != SectionEnd::Flush {
            return Err(self.wrong_terminator(input, "get-object", SectionEnd::Flush, end));
        }
        if !self.operation.may_download() {
            return Err(Error::NotAllowed(
                "get-object is only available on a `download` connection".to_string(),
            ));
        }
        let oid =
            Oid::parse(argument).map_err(|err| Error::BadRequest(format!("get-object: {err}")))?;
        let Some(object) = self.store.open(&oid)? else {
            return Err(Error::NotFound(format!(
                "object {oid} is not on this server"
            )));
        };

        // `size=` is mandatory and must precede the data: the client errors with
        // "no size argument seen" before it reads a byte.
        wire::write_text(output, &format!("status {}", 200))?;
        wire::write_text(output, &format!("size={}", object.size))?;
        wire::write_delim(output)?;

        let mut reader = object.reader;
        let mut framer = wire::DataFramer::new(output);
        // `copy` with an explicit slab rather than `std::io::copy`: the slab
        // size is the memory bound this module promises, and it must be stated
        // here rather than inherited from whatever `copy` happens to pick.
        let mut slab = vec![0u8; CHUNK_BYTES];
        let mut sent = 0u64;
        loop {
            let n = reader.read(&mut slab)?;
            if n == 0 {
                break;
            }
            framer.write_all(&slab[..n])?;
            sent += n as u64;
        }
        let framed = framer.finish()?;
        wire::write_flush(output)?;

        debug_assert_eq!(framed, sent);
        if sent != object.size {
            // The store lied about the size it just reported. The client has
            // already been told `size=`, so the only honest thing left is to
            // say so in the log. The wire is committed.
            tracing::error!(
                target: crate::LFS_TARGET,
                %oid, declared = object.size, sent,
                "the object store returned a different number of bytes than it declared"
            );
        }
        summary.objects_sent += 1;
        summary.bytes_sent += sent;
        Ok(Flow::Continue)
    }

    fn put_object(
        &self,
        argument: &str,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
        summary: &mut Summary,
    ) -> Result<Flow> {
        let (args, end) = wire::read_args(input, self.limits.max_args, "put-object")?;

        // Everything that can be refused *before* the data section is read is
        // refused here, but each refusal must still consume the object the
        // client is already writing, or the next read finds object bytes where
        // a command belongs. `refuse` does both, in that order.
        //
        // **The drain's failure must not become the answer.** `?` on the drain
        // discarded the refusal that was already decided and returned the
        // drain's own `Framing` error, which is fatal and carries no status —
        // so a client that sent past the drain budget got a dropped channel
        // instead of its 413. That budget is 128 KiB and the quota is 8 GiB, so
        // it was every refusal an operator would ever configure a quota for.
        // See [`Error::RefusedThenClosed`].
        //
        // **The macro is a thin wrapper over `drain_then`, not a second copy of
        // it.** When the drain-swallows-the-refusal bug was fixed it was fixed
        // *here*, in the macro, and the two refusals further down that drain by
        // hand — `begin_write` failing, and the sink rejecting a chunk — were
        // left spelled `self.drain_data_section(input)?`, still swallowing.
        // Nothing reached them at the time. `Error::OutOfSpace` reaches both.
        // One writer now, so there is no third copy to forget (one implementation).
        macro_rules! refuse {
            ($err:expr) => {{
                let err: Error = $err;
                if end == SectionEnd::Delim {
                    return Err(self.drain_then(input, err));
                }
                return Err(err);
            }};
        }

        if !self.operation.may_upload() {
            refuse!(Error::NotAllowed(
                "put-object is only available on an `upload` connection".to_string(),
            ));
        }
        let oid = match Oid::parse(argument) {
            Ok(oid) => oid,
            Err(err) => refuse!(Error::BadRequest(format!("put-object: {err}"))),
        };
        let Some(size_text) = wire::arg(&args, "size") else {
            refuse!(Error::BadRequest(
                "put-object requires a `size` argument".to_string()
            ));
        };
        let Ok(declared) = size_text.parse::<u64>() else {
            refuse!(Error::BadRequest(format!(
                "put-object: `{}` is not a size",
                truncate(size_text)
            )));
        };
        if declared > self.limits.max_object_bytes {
            refuse!(Error::TooLarge(format!(
                "object {oid} is {declared} bytes; this server accepts at most {}",
                self.limits.max_object_bytes
            )));
        }
        // **The tenant's own room, and this is the load-bearing copy of that
        // check** for the same reason the line above is the load-bearing copy of
        // the operator's: a client is free to skip `batch` entirely, so a door
        // that only stood in `batch` would stand in front of the polite clients
        // and nobody else.
        //
        // Counted against the NAMESPACE's ledger and not this session's
        // `taken`, which is the fix for the concurrent half of the same
        // question. `taken` answered "what have I committed" and every session
        // could only ever answer it about itself, so two sessions opened at the
        // same instant were both admitted against an empty namespace. The
        // ledger counts this session's earlier objects too, so the
        // split-into-legal-pieces refusal that `taken` used to make is not lost
        // — see [`NamespaceAllowance::admit`].
        //
        // ⚠ AND IT IS CHARGED ON A NUMBER THE CLIENT CHOSE. The payload has not
        // arrived, so `declared` is all there is to judge by, and `ledger.rs`
        // is a monotone counter with nothing to give back — deliberately, and
        // its module comment argues the case at length: a reservation that can
        // be leaked locks a tenant out of quota they own, forever, on one
        // crash. That argument holds, and it assumes the over-charge is
        // "over-conservative, transient, and self-healing".
        //
        // A client that declares and then does not deliver breaks the
        // assumption rather than the design. Nothing bytes back a lie, so see
        // `charged_but_not_delivered` below: the object's refusal is fatal to
        // the connection, which bounds a lying client to ONE charge per
        // connection instead of one per round trip.
        let charged = match &self.allowance {
            Some(allowance) => {
                if let Err(err) = allowance.admit(declared) {
                    refuse!(err);
                }
                true
            }
            None => false,
        };
        if end != SectionEnd::Delim {
            return Err(Error::BadRequest(
                "put-object must be followed by a delim-pkt and the object".to_string(),
            ));
        }

        let pending = match self.store.begin_write(&oid, declared) {
            Ok(pending) => pending,
            Err(err) => return Err(self.drain_then(input, err)),
        };
        let mut sink = VerifyingSink::new(pending, oid, declared);

        loop {
            match gunnar_pktline::expect_packet(input, "the object payload")? {
                Packet::Flush => break,
                Packet::Data(data) => {
                    if let Err(err) = sink.accept(&data) {
                        // Over-long already, or the store has no room: stop
                        // feeding it, discard what was written, and drain the
                        // rest so the session survives to answer.
                        drop(sink);
                        let refusal = self.drain_then(input, err);
                        return Err(self.charged_but_not_delivered(charged, declared, refusal));
                    }
                }
                other => {
                    return Err(Error::Framing(format!(
                        "unexpected {other:?} inside the object payload"
                    )))
                }
            }
        }

        let received = sink.received();
        if let Err(err) = sink.commit() {
            // The object did not turn out to be the one that was declared: too
            // short, or the wrong hash. The allowance was charged for
            // `declared` bytes and none of them landed.
            return Err(self.charged_but_not_delivered(charged, declared, err));
        }
        summary.objects_received += 1;
        summary.bytes_received += received;
        // **After `commit`, never before.** What the allowance counts is bytes
        // the tenant now HOLDS, and an object that failed verification or that
        // the store refused holds nothing — charging for it would refuse a
        // tenant's next legitimate upload on the strength of one that never
        // landed. `received` and not `declared` for the same reason: the two are
        // equal here only because `VerifyingSink` just proved they are.
        self.taken.set(self.taken.get().saturating_add(received));
        // Success carries no lines. `SuccessStatus()` in the reference server
        // is the no-delim form, and the client reads this one with
        // `ReadStatusWithLines`, which accepts either.
        wire::write_status(output, 200, &[])?;
        Ok(Flow::Continue)
    }

    fn verify_object(
        &self,
        argument: &str,
        input: &mut dyn BufRead,
        output: &mut dyn Write,
    ) -> Result<Flow> {
        let (args, end) = wire::read_args(input, self.limits.max_args, "verify-object")?;
        if end != SectionEnd::Flush {
            return Err(self.wrong_terminator(input, "verify-object", SectionEnd::Flush, end));
        }
        if !self.operation.may_upload() {
            return Err(Error::NotAllowed(
                "verify-object is only available on an `upload` connection".to_string(),
            ));
        }
        let oid = Oid::parse(argument)
            .map_err(|err| Error::BadRequest(format!("verify-object: {err}")))?;
        let declared: u64 = wire::arg(&args, "size")
            .ok_or_else(|| {
                Error::BadRequest("verify-object requires a `size` argument".to_string())
            })?
            .parse()
            .map_err(|_| Error::BadRequest("verify-object: malformed `size`".to_string()))?;

        match self.store.size_of(&oid)? {
            None => Err(Error::NotFound(format!(
                "object {oid} was not stored by this server"
            ))),
            Some(stored) if stored != declared => Err(Error::Corrupt(format!(
                "object {oid} is {stored} bytes on this server, not {declared}"
            ))),
            Some(_) => {
                wire::write_status(output, 200, &[])?;
                Ok(Flow::Continue)
            }
        }
    }

    /// **A client that charged the allowance and did not deliver does not get
    /// to do it again on this connection.**
    ///
    /// `put-object` admits against the size the client DECLARED, because the
    /// payload has not arrived yet and there is nothing else to judge by, and
    /// `ledger.rs` is a monotone counter with nothing to give back. Its module
    /// comment makes that case well and the case rests on one assumption: that
    /// the over-charge is "over-conservative, transient, and self-healing". A
    /// declaration the client then does not honour is not any of those. The
    /// refusal for a corrupt object is a 422, which is NOT fatal, so the same
    /// connection could declare `max_object_bytes`, send garbage, be told 422,
    /// and declare again — as fast as it could loop, charging a namespace's
    /// whole plan without a byte of it landing. The counter is per NAMESPACE,
    /// so what it exhausts is every other session in that namespace whose mark
    /// was taken first, not only the liar's own.
    ///
    /// This does not make the charge correct; it bounds it. One lie costs one
    /// object's declaration and then the connection, which is exactly the size
    /// of over-charge `ledger.rs` already reasons about and calls self-healing.
    /// Charging what was RECEIVED instead — the obvious reading — would mean
    /// admitting before the bytes are counted and settling afterwards, which is
    /// a reservation, which is the hazard that file refuses by construction.
    ///
    /// `RefusedThenClosed` and not a bare fatal error, because the client is
    /// still owed the 422: `final_report` writes it before the session goes.
    /// Untouched when no allowance was charged — a server with no plan has
    /// nothing to exhaust, and ending its sessions over a corrupt object would
    /// be a new refusal wearing this one's clothes.
    fn charged_but_not_delivered(&self, charged: bool, declared: u64, refusal: Error) -> Error {
        if !charged {
            return refusal;
        }
        Error::RefusedThenClosed {
            refusal: Box::new(refusal),
            closing: Box::new(Error::Corrupt(format!(
                "the namespace's allowance was charged {declared} bytes for an object that                  did not arrive; this connection does not get to declare another"
            ))),
        }
    }

    /// Read out the object a refusal has already been decided for, and give back
    /// the error to answer with.
    ///
    /// **The refusal always survives.** If the drain fails, the two facts are
    /// kept apart in [`Error::RefusedThenClosed`] — what the client is told, and
    /// why the session then ends — rather than the drain's fatal `Framing`
    /// replacing the status the client was owed. This is the one place that
    /// decision is made; see the note on `put-object`'s `refuse!`.
    fn drain_then(&self, input: &mut dyn BufRead, refusal: Error) -> Error {
        match self.drain_data_section(input) {
            Ok(()) => refusal,
            Err(desync) => Error::RefusedThenClosed {
                refusal: Box::new(refusal),
                closing: Box::new(desync),
            },
        }
    }

    /// Read and discard an object the server has already refused.
    ///
    /// Bounded by [`Limits::max_drain_bytes`]: past that the session ends,
    /// because a client that keeps sending after a refusal is not one this
    /// server owes an orderly conversation to. Memory is O(one pkt-line)
    /// throughout; the packets are dropped as they arrive.
    fn drain_data_section(&self, input: &mut dyn BufRead) -> Result<()> {
        let mut drained = 0u64;
        loop {
            match gunnar_pktline::expect_packet(input, "a refused object payload")? {
                Packet::Flush => return Ok(()),
                Packet::Data(data) => {
                    drained += data.len() as u64;
                    if drained > self.limits.max_drain_bytes {
                        return Err(Error::Framing(format!(
                            "a refused upload kept sending past {} bytes; \
                             the session is closing rather than reading it out",
                            self.limits.max_drain_bytes
                        )));
                    }
                }
                other => {
                    return Err(Error::Framing(format!(
                        "unexpected {other:?} inside a refused object payload"
                    )))
                }
            }
        }
    }
}

enum Flow {
    Continue,
    Quit,
}

fn truncate(text: &str) -> String {
    text.chars().take(64).collect()
}