protocol.rs
In rust/crates/mkit-core/src/
Preview
Code · 49 KiB//! Cross-transport types: error taxonomy, the [`Transport`] trait, the
//! [`PackKey`] digest wrapper, and the retry/backoff helpers used by
//! every transport implementation (memory, file, HTTP, S3, SSH, enc).
//! [`retrying`] is the single shared driver of the SPEC-TRANSPORT §7
//! ladder — every transport's `retrying`/`with_retry` method is a thin
//! wrapper around it, so the backoff/classification policy lives in
//! exactly one place instead of being reimplemented per crate.
//!
//! The SSH wire format is defined in `mkit-rpc`'s `ssh.proto` and
//! lives in `mkit_rpc::mkit::rpc::v1::ssh`; transport-ssh consumes
//! the schema directly. The hand-rolled `OP_HELLO` byte format that
//! used to live in this module has been retired.
// SPEC-TRANSPORT §7 calls out the exponential ladder in seconds
// (1, 2, 4, …, 300). Expressing those values with `Duration::from_secs`
// is deliberate — switching to `from_mins` loses the one-to-one match
// with the spec text.
#![allow(clippy::duration_suboptimal_units)]
use core::fmt;
use core::time::Duration;
use crate::hash::{FromHexError, Hash, to_hex};
use crate::refs::Ref;
pub use crate::refs::RefWriteCondition;
// ---------------------------------------------------------------------------
// Error taxonomy
// ---------------------------------------------------------------------------
/// Errors that any transport may surface across the [`Transport`]
/// boundary. Implementations MAY wrap transport-specific errors
/// internally but MUST map them to one of these variants before
/// returning.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum TransportError {
/// Verification is still pending. Scheduling belongs to the caller.
#[error("remote error: verification pending, retry after {retry_after:?}")]
PendingVerification { retry_after: Duration },
/// Repository membership has not become visible yet.
#[error("remote error: repository membership not yet visible")]
MembershipNotVisible,
/// A remote supplied a bounded delay hint, in milliseconds.
#[error("remote error: retry after {ms} ms")]
RetryAfter { ms: u64 },
/// `download_pack` called on a digest the remote does not hold.
#[error("pack not found on remote")]
PackNotFound,
/// Authentication or ACL failure (HTTP 401/403, SSH auth refusal,
/// S3 `SignatureDoesNotMatch`, …).
#[error("access denied by remote")]
AccessDenied,
/// A remote requires admission before this write may proceed.
#[error("{0}")]
AdmissionRequired(Box<AdmissionRequired>),
/// User admission configuration or helper headers were rejected.
#[error("admission configuration: {0}")]
AdmissionConfiguration(String),
/// Local HTTPS trust configuration could not be loaded or validated.
#[error("HTTPS trust configuration: {0}")]
TlsConfiguration(String),
/// The configured admission helper failed.
#[error("admission helper failed: {0}")]
AdmissionHelperFailed(String),
/// Catch-all remote-side failure carrying an advisory message. The
/// message is for operators; programs MUST NOT pattern-match on its
/// contents.
#[error("remote error: {0}")]
RemoteError(String),
/// `update_ref` CAS precondition was not satisfied. Per
/// SPEC-TRANSPORT §7, callers MUST treat this as
/// "possibly-success on retry" for `.missing` / `.match` and
/// confirm with `read_ref`.
#[error("ref CAS precondition failed")]
RefConflict,
/// Caller passed a ref name failing SPEC-REFS §3.
#[error("invalid ref name: {0}")]
InvalidRef(String),
/// Network-level failure: DNS, TCP connect, TLS handshake, SSH
/// subprocess spawn. Retryable (see [`is_retryable`]).
#[error("connection to remote failed")]
ConnectionFailed,
/// Unexpected HTTP status or transport-protocol error. 5xx and 429
/// are retryable; 4xx (except 401/403/404/409/412) is not.
#[error("server error (status {status})")]
ServerError {
/// Numeric status code. HTTP uses its native codes; transports
/// without a status integer use `0`.
status: u16,
},
/// Server response did not match the wire contract (truncated
/// frame, unknown opcode, bad JSON, …).
#[error("invalid response from remote")]
InvalidResponse,
/// Generic protocol-level failure — malformed frame, unexpected
/// opcode order, or failed handshake.
#[error("protocol error")]
ProtocolError,
/// Payload exceeded a transport-specific cap.
#[error("payload too large: {0} bytes")]
PayloadTooLarge(usize),
/// An insecure URL scheme (plain `http://`) was supplied for a
/// non-loopback host. Plain HTTP is restricted to loopback addresses
/// (`127.0.0.1`, `::1`, `localhost`) so production traffic is never
/// transported in the clear.
#[error("insecure scheme: plain http:// is allowed only for loopback hosts")]
InsecureScheme,
}
/// One opaque challenge advertised by the remote.
#[derive(Clone, PartialEq, Eq)]
pub struct AdmissionChallengeEntry {
/// Lowercase scheme identifier.
pub scheme: String,
/// Opaque challenge value. Do not display or log it.
pub value: String,
}
impl fmt::Debug for AdmissionChallengeEntry {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AdmissionChallengeEntry")
.field("scheme", &self.scheme)
.field("value_len", &self.value.len())
.finish()
}
}
/// A remote admission challenge and the bounded response headers needed by a helper.
#[derive(Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct AdmissionRequired {
/// Challenges in server preference order.
pub challenges: Vec<AdmissionChallengeEntry>,
/// Untrusted server description, retained for the helper.
pub description: String,
/// Bounded `WWW-Authenticate` header values.
pub www_authenticate: Vec<String>,
/// Bounded `PAYMENT-REQUIRED` header values.
pub payment_required: Vec<String>,
/// Terminal admission state, if a helper already ran or hit its cap.
pub reason: Option<&'static str>,
}
impl AdmissionRequired {
/// Construct a remote challenge from already-bounded fields. The Connect
/// client validates the bounds before calling this.
#[must_use]
pub fn new(
challenges: Vec<AdmissionChallengeEntry>,
description: String,
www_authenticate: Vec<String>,
payment_required: Vec<String>,
) -> Self {
Self {
challenges,
description,
www_authenticate,
payment_required,
reason: None,
}
}
/// Add a safe terminal explanation without exposing challenge values.
#[must_use]
pub fn with_reason(mut self, reason: &'static str) -> Self {
self.reason = Some(reason);
self
}
}
impl fmt::Debug for AdmissionRequired {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AdmissionRequired")
.field("challenges", &self.challenges)
.field("description_len", &self.description.len())
.field("www_authenticate_count", &self.www_authenticate.len())
.field("payment_required_count", &self.payment_required.len())
.field("reason", &self.reason)
.finish_non_exhaustive()
}
}
impl fmt::Display for AdmissionRequired {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("admission required by remote: ")?;
write_safe_remote_text(f, &self.description)?;
if self.challenges.is_empty() {
f.write_str(" (no challenges)")?;
} else {
f.write_str(" (schemes: ")?;
for (i, challenge) in self.challenges.iter().enumerate() {
if i != 0 {
f.write_str(", ")?;
}
write_safe_remote_text(f, &challenge.scheme)?;
}
f.write_str(")")?;
}
if let Some(reason) = self.reason {
write!(f, ": {reason}")?;
}
Ok(())
}
}
fn write_safe_remote_text(f: &mut fmt::Formatter<'_>, value: &str) -> fmt::Result {
for c in value.chars() {
if c.is_control() || matches!(c, '\u{202a}'..='\u{202e}' | '\u{2066}'..='\u{2069}') {
write!(f, "\\u{{{:x}}}", c as u32)?;
} else {
write!(f, "{c}")?;
}
}
Ok(())
}
/// Result alias used throughout this module.
pub type TransportResult<T> = Result<T, TransportError>;
// ---------------------------------------------------------------------------
// PackKey — 32-byte digest wrapper
// ---------------------------------------------------------------------------
/// A 32-byte pack digest used as the content-address for an uploaded
/// pack. This is the same 32 bytes as [`Hash`](tyalias@Hash) but wrapped so pack
/// digests and object hashes do not silently cross purposes at API
/// boundaries.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct PackKey(pub [u8; 32]);
impl PackKey {
/// Build a [`PackKey`] from a raw 32-byte digest.
#[must_use]
pub const fn new(bytes: [u8; 32]) -> Self {
Self(bytes)
}
/// Borrow the underlying 32 bytes.
#[must_use]
pub const fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
/// Lowercase 64-char hex.
#[must_use]
pub fn to_hex(&self) -> String {
to_hex(&self.0)
}
/// Build a [`PackKey`] from a [`Hash`](tyalias@Hash) (alias for [`From`]).
#[must_use]
pub const fn from_hash(h: Hash) -> Self {
Self(h)
}
/// Convert back to a plain [`Hash`](tyalias@Hash).
#[must_use]
pub const fn into_hash(self) -> Hash {
self.0
}
/// Verify downloaded bytes against this independently requested address.
/// Covers the entire byte sequence, including a pack's internal trailer.
/// Also applies to auxiliary blobs: a lookup URL or a self-consistent
/// manifest cannot authenticate the returned bytes on its own.
///
/// # Errors
/// Returns [`TransportError::InvalidResponse`] on a digest mismatch.
pub fn verify_bytes(self, bytes: &[u8]) -> TransportResult<()> {
if crate::hash::hash(bytes) != self.0 {
return Err(TransportError::InvalidResponse);
}
Ok(())
}
}
impl From<Hash> for PackKey {
fn from(h: Hash) -> Self {
Self(h)
}
}
impl From<PackKey> for Hash {
fn from(k: PackKey) -> Hash {
k.0
}
}
impl fmt::Display for PackKey {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.to_hex())
}
}
/// Parse a [`PackKey`] from a 64-char lowercase hex string.
///
/// Accepts uppercase too (matches the permissive [`crate::hash::from_hex`]
/// semantics); callers that require lowercase MUST validate the input
/// independently.
pub fn pack_key_from_hex(s: &str) -> Result<PackKey, FromHexError> {
let h = crate::hash::from_hex(s)?;
Ok(PackKey(h))
}
// ---------------------------------------------------------------------------
// Retry / backoff
// ---------------------------------------------------------------------------
/// Return `true` if a transport should retry after seeing `err`.
///
/// Retryable per SPEC-TRANSPORT §7:
/// - [`TransportError::ConnectionFailed`]
/// - [`TransportError::ServerError`] with a 5xx status OR HTTP 429.
///
/// Explicitly non-retryable:
/// - [`TransportError::PackNotFound`]
/// - [`TransportError::AccessDenied`]
/// - [`TransportError::AdmissionRequired`] — requires an explicit helper response
/// - [`TransportError::RefConflict`] (CAS retry is a caller-level policy)
/// - [`TransportError::InvalidRef`]
/// - [`TransportError::InvalidResponse`] / [`TransportError::ProtocolError`]
/// - [`TransportError::PayloadTooLarge`]
/// - [`TransportError::TlsConfiguration`] — local trust setup failed.
/// - [`TransportError::RemoteError`] — the remote chose not to be specific;
/// we do not guess.
/// - [`TransportError::ServerError`] with any 4xx status.
#[must_use]
pub fn is_retryable(err: &TransportError) -> bool {
match err {
TransportError::ConnectionFailed => true,
TransportError::ServerError { status } => *status >= 500 || *status == 429,
_ => false,
}
}
/// Max attempts for the default backoff ladder.
///
/// SPEC-TRANSPORT §7: `attempt = 1; while attempt ≤ 5`.
pub const BACKOFF_MAX_ATTEMPTS: u32 = 5;
/// Initial sleep between attempts.
pub const BACKOFF_INITIAL: Duration = Duration::from_secs(1);
/// Upper bound on any individual sleep.
pub const BACKOFF_CAP: Duration = Duration::from_secs(300);
/// Per-pack body size ceiling enforced by every transport that ingests
/// pack bytes (HTTP `Content-Length`, S3 `GetObject`, SSH
/// `DownloadPackHeader.total_bytes`). On 64-bit targets, 4 GiB matches
/// the pack-format addressable range; pointer-width-limited targets cap
/// at their maximum addressable buffer size instead of failing to compile.
#[cfg(target_pointer_width = "64")]
pub const PACK_BODY_LIMIT: u64 = 4 * 1024 * 1024 * 1024;
#[cfg(not(target_pointer_width = "64"))]
pub const PACK_BODY_LIMIT: u64 = usize::MAX as u64;
/// `usize`-typed mirror of [`PACK_BODY_LIMIT`] for `Vec`-shaped buffer
/// caps. The assertion below prevents silent truncation on any target.
#[allow(clippy::cast_possible_truncation)]
pub const PACK_BODY_LIMIT_USIZE: usize = PACK_BODY_LIMIT as usize;
const _: () = assert!(
(PACK_BODY_LIMIT_USIZE as u64) == PACK_BODY_LIMIT,
"PACK_BODY_LIMIT does not fit in usize on this target",
);
/// Exponential-backoff iterator used by all transports.
///
/// Yields `[1s, 2s, 4s, 8s, 16s]` (5 attempts) for the default ladder,
/// doubling each step and capping at 300s. This is the ladder mandated
/// by SPEC-TRANSPORT §7 for `ConnectionFailed`, 5xx, and HTTP 429.
///
/// The iterator is self-contained — it holds no reference to a clock,
/// so it can be constructed in tests and exhaustively enumerated.
#[derive(Debug, Clone)]
pub struct BackoffIterator {
next_delay: Duration,
attempts_remaining: u32,
cap: Duration,
}
impl BackoffIterator {
/// Default ladder: 5 attempts, starting at 1s, doubling, capped at 300s.
#[must_use]
pub const fn new() -> Self {
Self {
next_delay: BACKOFF_INITIAL,
attempts_remaining: BACKOFF_MAX_ATTEMPTS,
cap: BACKOFF_CAP,
}
}
/// Custom ladder for tests.
#[must_use]
pub const fn with(initial: Duration, cap: Duration, attempts: u32) -> Self {
Self {
next_delay: initial,
attempts_remaining: attempts,
cap,
}
}
}
impl Default for BackoffIterator {
fn default() -> Self {
Self::new()
}
}
impl Iterator for BackoffIterator {
type Item = Duration;
fn next(&mut self) -> Option<Self::Item> {
if self.attempts_remaining == 0 {
return None;
}
self.attempts_remaining -= 1;
let current = self.next_delay;
let doubled = current.saturating_mul(2);
self.next_delay = if doubled > self.cap {
self.cap
} else {
doubled
};
Some(current)
}
}
/// Transport-agnostic retry driver shared by every [`Transport`]
/// implementation, so the SPEC-TRANSPORT §7 ladder lives in exactly one
/// place instead of being reimplemented per crate. Extracted from what
/// was `HttpTransport::retrying` in `mkit-transport-http`.
///
/// `op` is re-invoked from scratch on every attempt — it MUST perform a
/// fresh, self-contained unit of work each call (a new HTTP request on
/// the existing connection pool, a freshly-reconnected SSH child, a
/// redialed encrypted session, …) rather than assuming any state left
/// over from a failed prior attempt is still valid. This matters most
/// for connection-oriented transports: a frame-level failure can leave
/// a stream mid-message-desynced, so `op` reconnecting before retrying
/// (rather than resuming on the same broken handle) is what makes the
/// retry safe, not just present.
///
/// `backoff` is a ladder *factory* (not a live iterator) so a fresh
/// ladder starts on every call to `retrying` — production uses
/// [`BackoffIterator::new`], tests inject a short/deterministic ladder.
/// `sleep` is the delay hook between attempts; production sleeps for
/// the full duration, tests typically inject a no-op or recorder.
///
/// Retries only the classes [`is_retryable`] accepts
/// (`ConnectionFailed`, `ServerError{5xx}`, `ServerError{429}`); every
/// other error returns immediately on the first attempt.
pub fn retrying<T>(
mut op: impl FnMut() -> TransportResult<T>,
backoff: fn() -> BackoffIterator,
sleep: fn(Duration),
) -> TransportResult<T> {
let mut ladder = backoff();
loop {
match op() {
Ok(v) => return Ok(v),
Err(err) => {
if is_retryable(&err)
&& let Some(delay) = ladder.next()
{
sleep(delay);
continue;
}
return Err(err);
}
}
}
}
// ---------------------------------------------------------------------------
// PackChunk — transport-agnostic streaming segment
// ---------------------------------------------------------------------------
/// One bounded-size segment of a streamed pack transfer.
///
/// Mirrors the wire-level `PackChunk` shape shared by the SSH and enc
/// transports (`offset`, `data`, `last` — see
/// `mkit-rpc/proto/mkit/rpc/v1/ssh/ssh.proto`) and the `mkit.transport.v1`
/// Connect proto being designed for the HTTP
/// reference worker, without this crate depending on any
/// protobuf-generated type: `mkit-core` is the dependency root that
/// `mkit-rpc` builds on, not the reverse (see this module's header
/// comment), so the canonical protobuf `PackChunk` cannot be named here.
/// Transports convert 1:1 between this type and their own wire
/// representation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PackChunk {
/// Byte offset of `data` within the pack. Consecutive chunks in one
/// transfer MUST have ascending, contiguous offsets starting at 0 —
/// i.e. chunk *n*'s `offset` equals the sum of every prior chunk's
/// `data.len()`.
pub offset: u64,
/// Chunk payload. Transports typically bound this to a fixed
/// per-frame maximum (e.g. `mkit_rpc::CHUNK_DATA_MAX`, 800 KiB for
/// SSH/enc) so no single chunk forces a large allocation.
pub data: Vec<u8>,
/// `true` on the final chunk of the stream. An empty pack is still
/// represented as exactly one chunk with `last = true` and empty
/// `data` — a stream MUST NOT end silently without a `last = true`
/// chunk.
pub last: bool,
}
// ---------------------------------------------------------------------------
// Transport trait
// ---------------------------------------------------------------------------
/// The mkit transport vtable.
///
/// Every transport (memory, file, HTTP, S3, SSH) implements this trait.
/// Methods are synchronous and take `&self`; transports that need
/// interior mutability (e.g. connection pools) MUST use a `Mutex` /
/// `RwLock` internally. This keeps the trait object-safe.
///
/// All implementations MUST honour the retry policy in
/// SPEC-TRANSPORT §7 internally OR document that the caller is
/// responsible — the abstract trait takes no position. The
/// [`is_retryable`] and [`BackoffIterator`] helpers are provided for
/// implementations that embed the policy.
/// The repository a transport addresses, for remote error context.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct RepositoryAddress<'a> {
/// Repository identity as sent on the wire (STC §7.4).
pub repository: &'a str,
/// Origin (scheme and authority) of the remote.
pub origin: &'a str,
}
impl<'a> RepositoryAddress<'a> {
/// Build an address from a repository identity and a remote origin.
#[must_use]
pub const fn new(repository: &'a str, origin: &'a str) -> Self {
Self { repository, origin }
}
}
pub trait Transport: Send + Sync {
/// Repository identity and origin used for remote error context, if available.
/// Existing transports carry no addressing context by default.
fn repository_address(&self) -> Option<RepositoryAddress<'_>> {
None
}
/// Upload a pack. The digest is computed by the caller (BLAKE3 of
/// the full pack bytes) and used as the object key — servers MAY
/// dedupe on this key.
fn upload_pack(&self, bytes: &[u8], key: &PackKey) -> TransportResult<()>;
/// Upload a pack for an intended head ref. Transports without upload
/// tickets ignore the hint and retain their existing upload behavior.
fn upload_pack_via_ref(
&self,
bytes: &[u8],
key: &PackKey,
_head_ref: &str,
) -> TransportResult<()> {
self.upload_pack(bytes, key)
}
/// Download a pack by its digest.
///
/// Returns [`TransportError::PackNotFound`] if the remote does not
/// hold this digest.
fn download_pack(&self, key: &PackKey) -> TransportResult<Vec<u8>>;
/// Download a pack with a read-your-writes ref hint. Defaults to ignoring it.
fn download_pack_via_ref(&self, key: &PackKey, _ref_name: &str) -> TransportResult<Vec<u8>> {
self.download_pack(key)
}
/// Upload a pack by streaming bounded-size [`PackChunk`]s instead of
/// requiring the whole pack materialized as one `&[u8]` up front.
///
/// `total_bytes` is the caller-declared pack length — the wire
/// header most streaming transports send before the first chunk.
/// `chunks` MUST yield its segments in ascending contiguous `offset`
/// order and end with exactly one item whose `last` field is `true`
/// (an empty pack still yields one `last = true` chunk with empty
/// `data`); the accumulated `data` length across every yielded chunk
/// MUST equal `total_bytes`. The digest (`key`) is still computed by
/// the caller up front, exactly as for [`Self::upload_pack`] — this
/// method does not hash the stream itself.
///
/// This is an additive, opt-in entry point: no existing transport is
/// forced to implement real streaming. The default impl buffers
/// `chunks` into one `Vec` (bounded by [`PACK_BODY_LIMIT_USIZE`]) and
/// delegates to [`Self::upload_pack`], so every transport gets a
/// working implementation with zero code — callers may always use
/// this entry point, even against a transport that has not opted
/// into streaming. Transports that can forward chunks directly to
/// their own wire (SSH, enc — see `mkit-transport-ssh`'s existing
/// `PackChunk` frame loop) SHOULD override this to avoid the buffer
/// and stay in bounded memory regardless of pack size.
///
/// # Errors
///
/// Returns [`TransportError::ProtocolError`] if `chunks` never
/// yields a `last = true` item, or if the accumulated byte count
/// does not equal `total_bytes`. Returns
/// [`TransportError::PayloadTooLarge`] if `total_bytes` (or the
/// accumulated count) would exceed [`PACK_BODY_LIMIT`]. Propagates
/// any error yielded by `chunks` itself (e.g. the caller's own I/O
/// error while reading a pack off disk).
fn upload_pack_streaming(
&self,
key: &PackKey,
total_bytes: u64,
chunks: &mut dyn Iterator<Item = TransportResult<PackChunk>>,
) -> TransportResult<()> {
if total_bytes > PACK_BODY_LIMIT {
return Err(TransportError::PayloadTooLarge(PACK_BODY_LIMIT_USIZE));
}
// `total_bytes <= PACK_BODY_LIMIT` was just checked, and
// `PACK_BODY_LIMIT_USIZE as u64 == PACK_BODY_LIMIT` is asserted
// at the constant's definition, so this conversion never
// truncates — `try_from` (rather than `as`) makes that provable
// to clippy instead of asserted in a comment.
let initial = usize::try_from(total_bytes).unwrap_or(PACK_BODY_LIMIT_USIZE);
let mut buf = Vec::with_capacity(initial);
let mut saw_last = false;
for chunk in chunks {
let c = chunk?;
if c.data.len() > PACK_BODY_LIMIT_USIZE.saturating_sub(buf.len()) {
return Err(TransportError::PayloadTooLarge(PACK_BODY_LIMIT_USIZE));
}
buf.extend_from_slice(&c.data);
if c.last {
saw_last = true;
break;
}
}
if !saw_last || buf.len() as u64 != total_bytes {
return Err(TransportError::ProtocolError);
}
self.upload_pack(&buf, key)
}
/// Download a pack as a lazy stream of bounded-size [`PackChunk`]s
/// instead of one big `Vec<u8>`.
///
/// This is an additive, opt-in entry point mirroring
/// [`Self::upload_pack_streaming`]. The default impl calls
/// [`Self::download_pack`] eagerly (so it does not save memory by
/// itself) and wraps the whole result as a single `last = true`
/// chunk — every transport gets a working implementation with zero
/// code. Transports that can read their own wire incrementally (SSH,
/// enc) SHOULD override this to yield each wire chunk as it arrives,
/// keeping memory bounded to roughly one chunk at a time regardless
/// of total pack size.
///
/// # Errors
///
/// Returns [`TransportError::PackNotFound`] immediately if the
/// remote does not hold `key` — a conforming implementation never
/// returns a stream that then fails its first item with
/// `PackNotFound`. Errors surfacing mid-stream (a malformed frame, a
/// connection drop) are yielded as `Err` items from the returned
/// iterator rather than failing this call itself, since an
/// overridden implementation may not know the transfer will fail
/// until partway through.
fn download_pack_streaming(
&self,
key: &PackKey,
) -> TransportResult<Box<dyn Iterator<Item = TransportResult<PackChunk>> + '_>> {
let bytes = self.download_pack(key)?;
Ok(Box::new(core::iter::once(Ok(PackChunk {
offset: 0,
data: bytes,
last: true,
}))))
}
/// HEAD-check a pack. Cheaper than [`Self::download_pack`] on
/// network transports.
fn pack_exists(&self, key: &PackKey) -> TransportResult<bool>;
/// Check a pack with a read-your-writes ref hint. Defaults to ignoring it.
fn pack_exists_via_ref(&self, key: &PackKey, _ref_name: &str) -> TransportResult<bool> {
self.pack_exists(key)
}
/// Upload a content-addressed **auxiliary blob** — transfer metadata
/// that is NOT a packfile (e.g. a packlist chain node, SPEC-PACKFILE is
/// silent on these). The key is BLAKE3 of `bytes`, exactly like a pack.
///
/// Auxiliary blobs share the digest-keyed content-addressed store with
/// packs (the store is a general blob store; "pack" is just the primary
/// content kind), so the default impl delegates to [`Self::upload_pack`].
/// The distinct verb keeps the *kind* explicit at the call site so a
/// caller never has to infer "is this blob a packfile or metadata?".
fn upload_blob(&self, bytes: &[u8], key: &PackKey) -> TransportResult<()> {
self.upload_pack(bytes, key)
}
/// Upload auxiliary content for an intended head ref. The default keeps
/// the transport's existing blob behavior.
fn upload_blob_via_ref(
&self,
bytes: &[u8],
key: &PackKey,
_head_ref: &str,
) -> TransportResult<()> {
self.upload_blob(bytes, key)
}
/// Download an auxiliary blob by digest. Counterpart to
/// [`Self::upload_blob`]; default impl delegates to
/// [`Self::download_pack`]. Returns [`TransportError::PackNotFound`] if
/// the remote does not hold this digest.
fn download_blob(&self, key: &PackKey) -> TransportResult<Vec<u8>> {
self.download_pack(key)
}
/// Download auxiliary metadata with a ref hint. Defaults to ignoring it.
fn download_blob_via_ref(&self, key: &PackKey, _ref_name: &str) -> TransportResult<Vec<u8>> {
self.download_blob(key)
}
/// Unconditional ref write — equivalent to
/// `update_ref(name, RefWriteCondition::Any, hash)`.
///
/// Default impl delegates to [`Self::update_ref`] so transports only
/// implement one entry point.
fn write_ref(&self, name: &str, hash: &Hash) -> TransportResult<()> {
self.update_ref(name, RefWriteCondition::Any, hash)
}
/// CAS ref write. See [`RefWriteCondition`].
///
/// On `.missing` / `.match` CAS failure, returns
/// [`TransportError::RefConflict`]. Callers retrying after a
/// timeout MUST follow up with [`Self::read_ref`] to confirm
/// whether the first attempt actually landed (SPEC-TRANSPORT §7).
fn update_ref(
&self,
name: &str,
condition: RefWriteCondition,
hash: &Hash,
) -> TransportResult<()>;
/// Read the current value of a ref, or `None` if it does not exist.
fn read_ref(&self, name: &str) -> TransportResult<Option<Hash>>;
/// List refs whose full name starts with `prefix`. Returned names
/// have `prefix` stripped per SPEC-REFS §4. An empty prefix lists
/// every ref.
fn list_refs(&self, prefix: &str) -> TransportResult<Vec<Ref>>;
/// Advance a branch by updating its **head ref and its packmap ref
/// together**, each under its own CAS precondition.
///
/// This exists so the delta-transfer invariant — "if `head_ref` resolves
/// to T, the packmap reconstructs `closure(T)`" — can be upheld without a
/// window where the two refs disagree. A transport backed by a
/// transactional ref store SHOULD override this to apply both writes in
/// ONE transaction: then a failed advance changes nothing, and the head
/// is never observed past a packmap that can't yet reconstruct it.
///
/// The default impl is the safe non-transactional approximation used by
/// stores without multi-ref transactions: it writes the **packmap first**
/// (durable before the head moves) then the head. A crash in between
/// leaves the head at its prior value and the packmap a superset — still
/// consistent for fetch. The [`AdvanceOutcome`] distinguishes a packmap
/// precondition failure (caller re-reads and retries the chain) from a
/// head precondition failure (caller treats it as non-fast-forward),
/// which a single `RefConflict` could not.
fn advance_refs(
&self,
head_ref: &str,
head_condition: RefWriteCondition,
head_value: &Hash,
packmap_ref: &str,
packmap_condition: RefWriteCondition,
packmap_value: &Hash,
) -> TransportResult<AdvanceOutcome> {
match self.update_ref(packmap_ref, packmap_condition, packmap_value) {
Ok(()) => {}
Err(TransportError::RefConflict) => return Ok(AdvanceOutcome::PackmapConflict),
Err(e) => return Err(e),
}
match self.update_ref(head_ref, head_condition, head_value) {
Ok(()) => Ok(AdvanceOutcome::Committed),
Err(TransportError::RefConflict) => Ok(AdvanceOutcome::HeadConflict),
Err(e) => Err(e),
}
}
/// Advance both refs while committing any upload tickets for `commit`.
/// Other transports ignore the pack keys and use their existing advance.
#[allow(clippy::too_many_arguments)]
fn advance_refs_committing(
&self,
head_ref: &str,
head_condition: RefWriteCondition,
head_value: &Hash,
packmap_ref: &str,
packmap_condition: RefWriteCondition,
packmap_value: &Hash,
_commit: &[PackKey],
) -> TransportResult<CommitOutcome> {
self.advance_refs(
head_ref,
head_condition,
head_value,
packmap_ref,
packmap_condition,
packmap_value,
)
.map(CommitOutcome::Advanced)
}
/// Limits the remote advertises to the push planner. `None` means the
/// transport does not impose an additional limit.
fn upload_limits(&self) -> UploadLimits {
UploadLimits::default()
}
/// Whether [`Self::advance_refs`] commits the head + packmap advance as
/// one indivisible transaction, rather than the default's ordered
/// packmap-then-head writes.
///
/// The default (non-transactional) `advance_refs` is safe for an
/// **appending** packmap write: per its doc comment, a crash or lost
/// head-CAS race between the two writes leaves the packmap a strict
/// superset of what the (unmoved) head needs — still reconstructable.
/// That safety argument does NOT extend to a packmap **reset** (a fresh
/// node with `prev = None`, produced by the pack-chain re-baseline,
/// mkit #406): a reset is not a superset of the prior chain, so a
/// packmap write that commits while the paired head write loses its CAS
/// would strand the (still-unmoved) head pointing at a commit whose
/// closure the reset packmap can no longer reconstruct (mkit #521).
///
/// Callers MUST treat `false` (the default) as "never request a
/// packmap reset against this transport" — see
/// `remote_dispatch::push_branch`'s re-baseline gate. Override to
/// `true` ONLY when [`Self::advance_refs`] is overridden with a
/// genuinely transactional implementation (e.g. the HTTP transport's
/// single-request `/refs/advance` endpoint, mkit #408).
fn supports_atomic_advance(&self) -> bool {
false
}
}
/// Result of [`Transport::advance_refs`] — a two-ref branch advance.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AdvanceOutcome {
/// Both refs were updated.
Committed,
/// The head precondition did not hold. An atomic transport leaves
/// nothing changed. Callers re-read the head (SPEC-TRANSPORT §7): if it
/// already holds their target, a retried write landed and the advance
/// succeeded; otherwise the branch moved under them (non-fast-forward).
HeadConflict,
/// The packmap precondition did not hold (a concurrent pusher advanced
/// the chain). Callers re-read the packmap and retry.
PackmapConflict,
}
/// Result of an advance that can consume upload tickets.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum CommitOutcome {
/// The advance reached the ordinary two-ref outcome.
Advanced(AdvanceOutcome),
/// A ticket was invalid, expired, incomplete, or was already consumed.
TicketRejected,
/// A delta base is not yet available to this repository.
DeltaBaseUnavailable,
/// The packlist names content not in this repository.
PacklistNotInRepository,
}
/// Server limits relevant to splitting a push into packs and advances.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct UploadLimits {
/// Maximum accepted serialized pack bytes, if advertised.
pub max_pack_bytes: Option<u64>,
/// Maximum number of upload tickets one advance may consume.
pub tickets_per_advance: Option<usize>,
/// First serialized pack size that requires a ticket, when known.
/// A transport may stop requiring tickets after discovery fallback.
pub ticket_threshold_bytes: Option<u64>,
}
// ---------------------------------------------------------------------------
// async_shim — sync/async bridge for transports that wrap an async cipher
// ---------------------------------------------------------------------------
/// Sync-over-async shim for transports whose underlying cipher / I/O is
/// async (e.g. `commonware-stream::encrypted`) but whose
/// [`Transport`] trait surface is intentionally sync.
///
/// Lives in `mkit-core` (the trait crate) because it is generic
/// infrastructure — multiple transports and sparse-checkout's transport
/// layer will reuse the same plug-in point. It does **not** depend on
/// `tokio`, `commonware-runtime`, or any concrete executor; callers
/// pick the runner.
///
/// # Why a trait
///
/// `mkit-transport-enc` and (once its transport layer lands) `mkit-core::sparse` need to
/// drive `async fn` bodies from a sync method. Hard-coding
/// `tokio::runtime::Handle::block_on` would bleed tokio across the
/// workspace; hard-coding `commonware_runtime::deterministic` would
/// mean production = tests. A pluggable `Executor` keeps the
/// runtime-choice at the consumer crate.
pub mod async_shim {
/// Drives an async future to completion synchronously. Pluggable so
/// callers can choose between `tokio`, `commonware-runtime`'s
/// deterministic runner (tests), or the planned production tokio
/// runner without `mkit-core` having to compile-time depend on a
/// specific runtime crate.
///
/// Implementations MUST be re-entrancy-safe in the sense expected
/// by the chosen runtime — calling `block_on` from inside an
/// already-running task on the same runtime will typically panic
/// or deadlock. The shim's contract is "synchronous external API
/// wraps async internals", not "arbitrary async-from-sync
/// recursion".
pub trait Executor: Send + Sync {
/// Block the current thread until `fut` resolves.
fn block_on<F, T>(&self, fut: F) -> T
where
F: core::future::Future<Output = T> + Send,
T: Send;
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn pack_key_hex_roundtrip() {
let bytes = [0x42u8; 32];
let pk = PackKey::new(bytes);
let hex = pk.to_hex();
assert_eq!(hex.len(), 64);
let pk2 = pack_key_from_hex(&hex).unwrap();
assert_eq!(pk, pk2);
}
#[test]
fn is_retryable_matches_spec() {
assert!(is_retryable(&TransportError::ConnectionFailed));
assert!(is_retryable(&TransportError::ServerError { status: 500 }));
assert!(is_retryable(&TransportError::ServerError { status: 503 }));
assert!(is_retryable(&TransportError::ServerError { status: 429 }));
assert!(!is_retryable(&TransportError::ServerError { status: 404 }));
assert!(!is_retryable(&TransportError::ServerError { status: 401 }));
assert!(!is_retryable(&TransportError::PackNotFound));
assert!(!is_retryable(&TransportError::AccessDenied));
assert!(!is_retryable(&TransportError::TlsConfiguration(
"bad CA file".into()
)));
assert!(!is_retryable(&TransportError::RefConflict));
}
#[test]
fn backoff_default_ladder_is_1_2_4_8_16() {
let delays: Vec<Duration> = BackoffIterator::new().collect();
assert_eq!(
delays,
vec![
Duration::from_secs(1),
Duration::from_secs(2),
Duration::from_secs(4),
Duration::from_secs(8),
Duration::from_secs(16),
]
);
}
#[test]
fn backoff_caps_at_max() {
let cap = Duration::from_secs(10);
let delays: Vec<Duration> = BackoffIterator::with(Duration::from_secs(8), cap, 5).collect();
// 8s, then cap (16s would exceed 10s cap; clamped to 10s)
assert_eq!(delays[0], Duration::from_secs(8));
for d in &delays[1..] {
assert!(*d <= cap);
}
}
// -----------------------------------------------------------------
// upload_pack_streaming / download_pack_streaming default impls
// -----------------------------------------------------------------
/// Minimal in-memory [`Transport`] that only implements the
/// required whole-buffer methods, so its `*_streaming` behavior is
/// entirely the trait's default impl under test.
#[derive(Default)]
struct RecordingTransport {
uploaded: std::sync::Mutex<Option<(Vec<u8>, PackKey)>>,
stored: std::sync::Mutex<std::collections::HashMap<[u8; 32], Vec<u8>>>,
}
impl Transport for RecordingTransport {
fn upload_pack(&self, bytes: &[u8], key: &PackKey) -> TransportResult<()> {
*self.uploaded.lock().unwrap() = Some((bytes.to_vec(), *key));
self.stored
.lock()
.unwrap()
.insert(*key.as_bytes(), bytes.to_vec());
Ok(())
}
fn download_pack(&self, key: &PackKey) -> TransportResult<Vec<u8>> {
self.stored
.lock()
.unwrap()
.get(key.as_bytes())
.cloned()
.ok_or(TransportError::PackNotFound)
}
fn pack_exists(&self, _key: &PackKey) -> TransportResult<bool> {
unimplemented!("not exercised by these tests")
}
fn update_ref(
&self,
_name: &str,
_condition: RefWriteCondition,
_hash: &Hash,
) -> TransportResult<()> {
unimplemented!("not exercised by these tests")
}
fn read_ref(&self, _name: &str) -> TransportResult<Option<Hash>> {
unimplemented!("not exercised by these tests")
}
fn list_refs(&self, _prefix: &str) -> TransportResult<Vec<Ref>> {
unimplemented!("not exercised by these tests")
}
}
#[test]
fn ticket_hooks_default_to_existing_transport_behavior() {
let transport = RecordingTransport::default();
let bytes = b"test pack";
let key = PackKey::new(crate::hash::hash(bytes));
transport
.upload_pack_via_ref(bytes, &key, "refs/heads/main")
.unwrap();
assert_eq!(transport.download_pack(&key).unwrap(), bytes);
transport
.upload_blob_via_ref(bytes, &key, "refs/heads/main")
.unwrap();
assert_eq!(transport.upload_limits(), UploadLimits::default());
}
fn chunks_of(data: &[u8], chunk_len: usize) -> Vec<PackChunk> {
if data.is_empty() {
return vec![PackChunk {
offset: 0,
data: Vec::new(),
last: true,
}];
}
let mut out = Vec::new();
let mut offset = 0usize;
while offset < data.len() {
let end = core::cmp::min(offset + chunk_len, data.len());
out.push(PackChunk {
offset: offset as u64,
data: data[offset..end].to_vec(),
last: end == data.len(),
});
offset = end;
}
out
}
#[test]
fn upload_pack_streaming_default_delegates_to_upload_pack() {
let t = RecordingTransport::default();
let payload = b"hello mkit pack bytes".repeat(100);
let key = PackKey::new([0x11; 32]);
let mut it = chunks_of(&payload, 7).into_iter().map(Ok);
t.upload_pack_streaming(&key, payload.len() as u64, &mut it)
.expect("streaming upload via default impl");
let (got_bytes, got_key) = t.uploaded.lock().unwrap().clone().expect("upload recorded");
assert_eq!(got_bytes, payload);
assert_eq!(got_key, key);
}
#[test]
fn upload_pack_streaming_default_rejects_missing_last_chunk() {
let t = RecordingTransport::default();
let key = PackKey::new([0x22; 32]);
// No chunk at all — total_bytes = 0 still requires one `last =
// true` chunk per the trait contract.
let mut it = core::iter::empty();
let err = t
.upload_pack_streaming(&key, 0, &mut it)
.expect_err("must reject a stream with no last=true chunk");
assert!(matches!(err, TransportError::ProtocolError));
}
#[test]
fn upload_pack_streaming_default_rejects_total_bytes_mismatch() {
let t = RecordingTransport::default();
let key = PackKey::new([0x33; 32]);
let mut it = core::iter::once(Ok(PackChunk {
offset: 0,
data: vec![1, 2, 3],
last: true,
}));
// Declared total (10) does not match the 3 bytes actually
// streamed.
let err = t
.upload_pack_streaming(&key, 10, &mut it)
.expect_err("must reject a total_bytes/accumulated-length mismatch");
assert!(matches!(err, TransportError::ProtocolError));
}
#[test]
fn upload_pack_streaming_default_propagates_chunk_error() {
let t = RecordingTransport::default();
let key = PackKey::new([0x44; 32]);
let mut it = core::iter::once(Err(TransportError::ConnectionFailed));
let err = t
.upload_pack_streaming(&key, 0, &mut it)
.expect_err("must propagate an error yielded mid-stream");
assert!(matches!(err, TransportError::ConnectionFailed));
}
#[test]
fn upload_pack_streaming_default_rejects_oversize_total() {
let t = RecordingTransport::default();
let key = PackKey::new([0x55; 32]);
let mut it = core::iter::empty();
let err = t
.upload_pack_streaming(&key, PACK_BODY_LIMIT + 1, &mut it)
.expect_err("must reject total_bytes above PACK_BODY_LIMIT");
assert!(matches!(err, TransportError::PayloadTooLarge(_)));
}
#[test]
fn download_pack_streaming_default_wraps_whole_pack() {
let t = RecordingTransport::default();
let key = PackKey::new([0x66; 32]);
let payload = vec![9u8; 4096];
t.upload_pack(&payload, &key).unwrap();
let mut stream = t.download_pack_streaming(&key).expect("stream opens");
let first = stream.next().expect("one chunk").expect("no error");
assert_eq!(first.data, payload);
assert!(first.last);
assert!(
stream.next().is_none(),
"default impl yields exactly one chunk"
);
}
#[test]
fn download_pack_streaming_default_propagates_not_found() {
let t = RecordingTransport::default();
let key = PackKey::new([0x77; 32]);
// `Box<dyn Iterator<..>>`'s `Ok` type isn't `Debug`, so match
// instead of `expect_err`.
match t.download_pack_streaming(&key) {
Err(TransportError::PackNotFound) => {}
Err(other) => panic!("expected PackNotFound, got {other:?}"),
Ok(_) => panic!("missing pack must fail before any chunk is produced"),
}
}
// -----------------------------------------------------------------
// retrying() driver
// -----------------------------------------------------------------
fn test_backoff() -> BackoffIterator {
BackoffIterator::with(Duration::from_millis(1), Duration::from_millis(1), 5)
}
fn no_sleep(_delay: Duration) {}
#[test]
fn retrying_succeeds_on_first_try_without_sleeping() {
use core::sync::atomic::{AtomicUsize, Ordering};
fn record_sleep(_delay: Duration) {
SLEEPS.fetch_add(1, Ordering::SeqCst);
}
static CALLS: AtomicUsize = AtomicUsize::new(0);
static SLEEPS: AtomicUsize = AtomicUsize::new(0);
CALLS.store(0, Ordering::SeqCst);
SLEEPS.store(0, Ordering::SeqCst);
let result = retrying::<u32>(
|| {
CALLS.fetch_add(1, Ordering::SeqCst);
Ok(7)
},
test_backoff,
record_sleep,
);
assert_eq!(result.unwrap(), 7);
assert_eq!(CALLS.load(Ordering::SeqCst), 1);
assert_eq!(SLEEPS.load(Ordering::SeqCst), 0);
}
#[test]
fn retrying_recovers_after_transient_connection_failures() {
use core::sync::atomic::{AtomicUsize, Ordering};
static CALLS: AtomicUsize = AtomicUsize::new(0);
CALLS.store(0, Ordering::SeqCst);
let result = retrying::<u32>(
|| {
let n = CALLS.fetch_add(1, Ordering::SeqCst);
if n < 3 {
Err(TransportError::ConnectionFailed)
} else {
Ok(42)
}
},
test_backoff,
no_sleep,
);
assert_eq!(result.unwrap(), 42);
assert_eq!(CALLS.load(Ordering::SeqCst), 4);
}
#[test]
fn retrying_gives_up_after_ladder_exhausts() {
use core::sync::atomic::{AtomicUsize, Ordering};
static CALLS: AtomicUsize = AtomicUsize::new(0);
CALLS.store(0, Ordering::SeqCst);
let result = retrying::<u32>(
|| {
CALLS.fetch_add(1, Ordering::SeqCst);
Err(TransportError::ConnectionFailed)
},
test_backoff,
no_sleep,
);
assert!(matches!(result, Err(TransportError::ConnectionFailed)));
// 5-attempt ladder => 1 initial + 5 retries = 6 total calls.
assert_eq!(CALLS.load(Ordering::SeqCst), 6);
}
#[test]
fn retrying_does_not_retry_non_retryable_errors() {
use core::sync::atomic::{AtomicUsize, Ordering};
static CALLS: AtomicUsize = AtomicUsize::new(0);
CALLS.store(0, Ordering::SeqCst);
let result = retrying::<u32>(
|| {
CALLS.fetch_add(1, Ordering::SeqCst);
Err(TransportError::PackNotFound)
},
test_backoff,
no_sleep,
);
assert!(matches!(result, Err(TransportError::PackNotFound)));
assert_eq!(CALLS.load(Ordering::SeqCst), 1);
}
}