W2/supply ingress

crates/before/README.md

@@ -168,7 +168,7 @@ To use `Clock` in this manner:
   Unlike `absorb` (and its operator forms `|` and `|=`), `recv` increments the clock's
   inner `Version` after absorbing the received one.
 
-Consistently using `send` and [`recv`](Clock::recv`) in this
+Consistently using `send` and `recv` in this
 manner ensures that the `Clock` tracks *process causality*, by following
 the ordering of sends and receives.
 

src/tree/mirror/streaming/materialized.rs

@@ -199,33 +199,51 @@ fn violation<T, E>(violation: Violation) -> Result<T, Error<E>> {
 /// An honest replica supplies each leaf at most once (disputed scopes
 /// partition the tree) and only leaves its own set holds, so the running
 /// total of absorbed live leaves is bounded by the `set_len` its greeting
-/// declared — a premise the session's window solve priced. Every
-/// ingestion site charges the ledger exactly where it checks version
-/// containment: the two declared premises are enforced side by side.
+/// declared — a premise the session's window solve priced. Two
+/// instruments enforce it, each holding its own ledger over the same
+/// declaration: the walk charges each supply exactly where it checks
+/// version containment, so the two declared premises are enforced side by
+/// side; the wire decoder charges each supplied record at ingress, before
+/// its payload takes backend custody, so the bound holds at every instant
+/// of a still-open reply.
 #[derive(Clone, Debug)]
 pub(crate) struct SupplyLedger {
     /// The sender's greeting-declared set length.
     declared: u64,
-    /// Live leaves absorbed so far, across every ingestion site.
+    /// Live leaves absorbed so far, across every ingestion site sharing
+    /// this ledger.
     absorbed: Arc<AtomicU64>,
 }
 
 impl SupplyLedger {
-    fn new(declared: u64) -> Self {
+    pub(crate) fn new(declared: u64) -> Self {
         SupplyLedger {
             declared,
             absorbed: Arc::default(),
         }
     }
 
-    /// Charge `leaves` absorbed supplies, failing the session at the first
-    /// leaf past the declaration ([`Violation::OverdrawnSupply`]).
-    pub(crate) fn absorb<E>(&self, leaves: u64) -> Result<(), Error<E>> {
+    /// Charge `leaves` absorbed supplies against the declaration.
+    ///
+    /// Errors with the declared length at the first leaf past it; each
+    /// enforcement point renders the overdraw in its own typed vocabulary
+    /// (the walk's [`Violation::OverdrawnSupply`], the wire decoder's
+    /// ingress rejection).
+    pub(crate) fn charge(&self, leaves: u64) -> Result<(), u64> {
         let prior = self.absorbed.fetch_add(leaves, Ordering::Relaxed);
         match prior.checked_add(leaves) {
             Some(total) if total <= self.declared => Ok(()),
             // A wrapped counter is past any declarable length too.
-            _ => violation(Violation::OverdrawnSupply),
+            _ => Err(self.declared),
+        }
+    }
+
+    /// Charge `leaves` absorbed supplies, failing the session at the first
+    /// leaf past the declaration ([`Violation::OverdrawnSupply`]).
+    pub(crate) fn absorb<E>(&self, leaves: u64) -> Result<(), Error<E>> {
+        match self.charge(leaves) {
+            Ok(()) => Ok(()),
+            Err(_) => violation(Violation::OverdrawnSupply),
         }
     }
 }

src/tree/mirror/streaming/remote/adapter/decode.rs

@@ -11,6 +11,7 @@ use crate::tree::{
         Backend, Leaf,
         backend::BoxNodeStream,
         convert::Convert,
+        materialized::SupplyLedger,
         message::{Reaction as ProtocolReaction, Reply},
         window::FAN,
     },
@@ -72,10 +73,13 @@ where
 ///
 /// `version_bytes` is the peer's greeting-declared `max_version_bytes`:
 /// a supplied version encoding over it is a
-/// [`DecodeError::OversizedVersion`] session violation.
+/// [`DecodeError::OversizedVersion`] session violation. `ledger` is the
+/// session's declared-`set_len` allowance, charged per record before the
+/// payload takes custody ([`DecodeError::OverdrawnSupply`]).
 pub fn early_supplies<B, T, G, F>(
     backend: B,
     version_bytes: u64,
+    ledger: SupplyLedger,
     parent: Prefix<S<G>>,
     frames: F,
 ) -> impl Stream<Item = Result<(u8, B::Node<G>), DecodeError<B::Error>>> + Send
@@ -95,7 +99,13 @@ where
         let leaves = leaves.inspect(|_| fan_probe::on_recv());
         let leaves: BoxNodeStream<'static, B, T, Z> = Box::pin(leaves);
         let mut assembled = pin!(backend.clone().assemble::<G>(leaves));
-        let mut read = pin!(read_early::<B, T, G, _>(version_bytes, parent, frames, tx));
+        let mut read = pin!(read_early::<B, T, G, _>(
+            version_bytes,
+            &ledger,
+            parent,
+            frames,
+            tx
+        ));
         let mut read_result: Option<Result<(), DecodeError<B::Error>>> = None;
         loop {
             let step = futures::future::poll_fn(|cx| {
@@ -133,6 +143,7 @@ where
 /// nothing after it — streaming its leaves to assembly.
 async fn read_early<B, T, G, F>(
     version_bytes: u64,
+    ledger: &SupplyLedger,
     parent: Prefix<S<G>>,
     mut frames: F,
     leaves: mpsc::Sender<Result<(Prefix<Z>, B::Node<Z>), B::Error>>,
@@ -157,6 +168,14 @@ where
                     let (version, message) = record.map_err(DecodeError::Record)?;
                     let (leaf_prefix, _) =
                         supplies.observe::<B::Error, T>(parent, &version, &message)?;
+                    // The set-length half of the greeting's priced
+                    // premises, charged per record before the payload
+                    // takes backend custody: a peer supplying past its
+                    // declaration fails at the offending record, while
+                    // the reply is still open.
+                    ledger
+                        .charge(1)
+                        .map_err(|declared| DecodeError::OverdrawnSupply { declared })?;
                     let leaf = <B::Node<Z> as Leaf<T>>::leaf(version, message)
                         .await
                         .map_err(DecodeError::Backend)?;
@@ -195,6 +214,7 @@ where
 pub async fn decode_reply<B, T, H, F>(
     backend: B,
     version_bytes: u64,
+    ledger: SupplyLedger,
     scope: Scope<S<H>>,
     frames: &mut F,
 ) -> Result<Decoded<B, T, S<H>, Vec<Scope<H>>>, DecodeError<B::Error>>
@@ -206,17 +226,25 @@ where
     S<S<H>>: Height,
     F: Stream<Item = Frame<T>> + Unpin,
 {
-    decode(backend, version_bytes, scope, frames, |scope, listing| {
-        let (_, prefix) = scope.next().ok_or(ScopeError::UnpositionedQuery)?;
-        Ok(Scope::new(prefix, listing))
-    })
+    decode(
+        backend,
+        version_bytes,
+        ledger,
+        scope,
+        frames,
+        |scope, listing| {
+            let (_, prefix) = scope.next().ok_or(ScopeError::UnpositionedQuery)?;
+            Ok(Scope::new(prefix, listing))
+        },
+    )
     .await
 }
 
 /// Decode one leaf-height reply, where only an empty request for the leaf is valid.
 pub async fn decode_leaf_reply<B, T, F>(
     backend: B,
     version_bytes: u64,
+    ledger: SupplyLedger,
     scope: Scope<Z>,
     frames: &mut F,
 ) -> Result<Decoded<B, T, Z, Vec<Scope<Z>>>, DecodeError<B::Error>>
@@ -225,19 +253,27 @@ where
     T: borsh::BorshDeserialize + Send + Sync + 'static,
     F: Stream<Item = Frame<T>> + Unpin,
 {
-    decode(backend, version_bytes, scope, frames, |scope, listing| {
-        if !listing.is_empty() {
-            return Err(ScopeError::NonemptyLeafQuery);
-        }
-        let (_, prefix) = scope.next().ok_or(ScopeError::UnpositionedQuery)?;
-        Ok(Scope::leaf(prefix))
-    })
+    decode(
+        backend,
+        version_bytes,
+        ledger,
+        scope,
+        frames,
+        |scope, listing| {
+            if !listing.is_empty() {
+                return Err(ScopeError::NonemptyLeafQuery);
+            }
+            let (_, prefix) = scope.next().ok_or(ScopeError::UnpositionedQuery)?;
+            Ok(Scope::leaf(prefix))
+        },
+    )
     .await
 }
 
 async fn decode<B, T, H, F, Q, N>(
     backend: B,
     version_bytes: u64,
+    ledger: SupplyLedger,
     scope: Scope<H>,
     frames: &mut F,
     question: Q,
@@ -261,7 +297,7 @@ where
     // reader's hand per reply stream, at `node_bytes(0, version_bound)`
     // plus the slot itself (the window's supply-decode envelope).
     let (tx, rx) = mpsc::channel::<Result<(Prefix<Z>, B::Node<Z>), B::Error>>(FAN);
-    let read = read_reply::<B, T, H, _, _, _>(version_bytes, scope, frames, question, tx);
+    let read = read_reply::<B, T, H, _, _, _>(version_bytes, &ledger, scope, frames, question, tx);
     let assemble = assemble_supplies::<B, T, H>(backend, rx);
     let (read, assembled) = futures::future::join(read, assemble).await;
     let Some(ReadReply {
@@ -280,6 +316,7 @@ where
 /// Read and validate exactly one reply while streaming its leaves to assembly.
 async fn read_reply<B, T, H, F, Q, N>(
     version_bytes: u64,
+    ledger: &SupplyLedger,
     mut scope: Scope<H>,
     frames: &mut F,
     mut question: Q,
@@ -340,6 +377,14 @@ where
                     if let Some((radix, prefix)) = run {
                         read.skeleton.push(Skeleton::Supply { radix, prefix });
                     }
+                    // The set-length half of the greeting's priced
+                    // premises, charged per record before the payload
+                    // takes backend custody: a peer supplying past its
+                    // declaration fails at the offending record, while
+                    // the reply is still open.
+                    ledger
+                        .charge(1)
+                        .map_err(|declared| DecodeError::OverdrawnSupply { declared })?;
                     let leaf = <B::Node<Z> as Leaf<T>>::leaf(version, message)
                         .await
                         .map_err(DecodeError::Backend)?;

src/tree/mirror/streaming/remote/adapter/error.rs

@@ -95,6 +95,20 @@ pub enum DecodeError<E> {
         "supplied version encodes {actual} bytes, over the peer's declared {declared}-byte bound"
     )]
     OversizedVersion { declared: u64, actual: usize },
+    /// A supplied leaf record past the peer's declared `set_len`.
+    ///
+    /// An honest peer supplies each leaf at most once and only leaves its
+    /// own set holds, so its greeting-declared set length bounds the
+    /// session's total supplied records; the local window solve priced
+    /// absorbed-supply volume from that declaration. The charge lands at
+    /// ingress, before the record's payload takes backend custody, so a
+    /// peer supplying past its declaration fails the session at the
+    /// offending record while its reply is still open, never after the
+    /// reply materializes. The in-process walk enforces the same premise
+    /// at absorption as
+    /// [`Violation::OverdrawnSupply`](crate::error::MaterializedViolation::OverdrawnSupply).
+    #[error("supplied leaf overruns the peer's declared set length of {declared}")]
+    OverdrawnSupply { declared: u64 },
     /// A positional wire reaction cannot be scoped without another child.
     #[error(transparent)]
     Scope(#[from] ScopeError),

src/tree/mirror/streaming/remote/adapter/tests.rs

@@ -15,7 +15,7 @@ use before::Version;
 use crate::{
     message::Message,
     tree::{
-        mirror::streaming::remote::codec::LeafRun,
+        mirror::streaming::{materialized::SupplyLedger, remote::codec::LeafRun},
         typed::{Hash, Path, hash::MERKLE_HASH_LEN},
     },
 };
@@ -32,6 +32,12 @@ fn hash(byte: u8) -> Hash {
     Hash([byte; MERKLE_HASH_LEN])
 }
 
+/// A set-length allowance no fixture here can exhaust, for tests whose
+/// subject is not the ingress supply charge.
+fn unbounded() -> SupplyLedger {
+    SupplyLedger::new(u64::MAX)
+}
+
 /// Build a supply run from borrowed leaf records, in the given order.
 fn leaf_run<T>(records: &[(&Version, &Message<T>)]) -> LeafRun<T> {
     let mut run = LeafRun::new();

src/tree/mirror/streaming/remote/adapter/tests/backend_errors.rs

@@ -17,7 +17,7 @@ use crate::tree::{
 
 use super::{
     super::{DecodeError, EncodeError, Scope, decode_reply, encode_reply},
-    LeafCase, hash, leaf_run, runtime,
+    LeafCase, hash, leaf_run, runtime, unbounded,
 };
 use crate::tree::mirror::streaming::{
     convert::Convert,
@@ -132,6 +132,7 @@ where
                 .block_on(decode_reply::<Failing<Local>, u64, H, _>(
                     backend.clone(),
                     u64::MAX,
+                    unbounded(),
                     Scope::new(parent, &[]),
                     &mut frames,
                 ))

src/tree/mirror/streaming/remote/adapter/tests/fan_occupancy.rs

@@ -35,6 +35,7 @@ use crate::{
 };
 
 use super::super::{Scope, decode::fan_probe, decode_reply, early_supplies};
+use super::unbounded;
 
 /// Leaf records per supply frame.
 const PER_FRAME: usize = 16;
@@ -85,6 +86,7 @@ fn peak_occupancy(mut input: impl Stream<Item = Frame<u64>> + Unpin) -> usize {
         decode_reply::<Local, u64, UnderUnderRoot, _>(
             Local,
             u64::MAX,
+            unbounded(),
             Scope::<UnderRoot>::opening(&[]),
             &mut input,
         )
@@ -136,6 +138,7 @@ fn eager_early_supplies_ride_the_same_ceiling() {
         let assembled: Vec<_> = early_supplies::<Local, u64, UnderRoot, _>(
             Local,
             u64::MAX,
+            unbounded(),
             Prefix::new(),
             stream::iter(frames(&leaves)),
         )