crates/gunnar-lfs/src/session.rs
//! 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()
}