From 075ee0be09be7ebce846043e5230ffb67a7f9a5c Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 8 Nov 2026 14:14:27 +0000 Subject: screen, signatures: the oldest waiting go, not the newest Review finding 9: once 4096 signatures waited for their changes, new ones were dropped, and the waiting changes were cut at 10 000 keeping the oldest. A member could send junk once (signatures of changes that never come, changes no signature will vouch for) and fill both for good. Every honest signature arriving before its change was then lost, so under signatures required its changes waited for the actor's next signature: a stall for quiet actors. Both now keep the newest: the oldest go. A share per sender, so a member sending junk without pause can't push out what others sent either, is in future-work.md. --- src/access/screen.rs | 13 +++++++--- src/document/signatures.rs | 65 +++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 69 insertions(+), 9 deletions(-) (limited to 'src') diff --git a/src/access/screen.rs b/src/access/screen.rs index a9450a6..4c640b6 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -39,7 +39,7 @@ use crate::document::replica::{Epoch, user_of}; use crate::document::signatures::Signatures; use crate::document::{Change, ChangeHash, Document, ReadDoc}; -/// Changes waiting for their dependencies, kept at most. +/// Changes waiting for their dependencies, kept at most: the oldest go. const MAX_WAITING: usize = 10_000; /// Refused changes kept for operators, at most: the oldest go. pub(crate) const MAX_REFUSED: usize = 1_000; @@ -187,6 +187,13 @@ impl Waiting { pub(crate) fn len(&self) -> usize { self.0.len() } + + /// The newest `MAX_WAITING`: the oldest go, so changes a peer sent that will never + /// be taken (whose dependencies or signatures never come) make way for others. + fn keep_newest(&mut self) { + let excess = self.0.len().saturating_sub(MAX_WAITING); + self.0.drain(..excess); + } } /// What `apply` did: how many changes it applied, and those it refused. @@ -269,7 +276,7 @@ pub(crate) fn apply( let (ready, unready) = in_order(doc, candidates); waiting.0 = unready; if ready.is_empty() { - waiting.0.truncate(MAX_WAITING); + waiting.keep_newest(); return Ok(screened); } @@ -462,7 +469,7 @@ pub(crate) fn apply( screened.roots.insert(root); } } - waiting.0.truncate(MAX_WAITING); + waiting.keep_newest(); Ok(screened) } diff --git a/src/document/signatures.rs b/src/document/signatures.rs index 919e7b7..27bbb9e 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -19,7 +19,7 @@ //! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received //! signature is checked once its change is there (`offer`), and kept if it vouches for //! more of its actor's changes than any before; one of a change not there yet waits -//! (`pending`, a bounded number). +//! (`pending`, a bounded number: the oldest go). use std::collections::HashMap; @@ -193,9 +193,47 @@ pub(crate) fn unpack(bytes: &[u8]) -> Option> { .collect(), ) } -/// Received signatures of changes not there yet, kept at most (`offer`). +/// Received signatures of changes not there yet, kept at most (`offer`): the oldest go. const MAX_PENDING: usize = 4096; +/// Received signatures of changes the document doesn't hold yet, `MAX_PENDING` at most: +/// the oldest go, so junk a peer sent once (signatures of changes that never come) +/// makes way for those that keep coming. +#[derive(Debug, Default, Clone)] +struct Pending { + by_change: HashMap, + /// The changes, oldest first; some may have left `by_change` since. + order: std::collections::VecDeque, +} + +impl Pending { + fn insert(&mut self, hash: ChangeHash, signature: [u8; 64]) { + if self.by_change.insert(hash, signature).is_none() { + self.order.push_back(hash); + } + while self.by_change.len() > MAX_PENDING { + match self.order.pop_front() { + Some(oldest) => { + self.by_change.remove(&oldest); + } + None => break, + } + } + if self.order.len() > 2 * MAX_PENDING { + let by_change = &self.by_change; + self.order.retain(|hash| by_change.contains_key(hash)); + } + } + + fn remove(&mut self, hash: &ChangeHash) -> Option<[u8; 64]> { + self.by_change.remove(hash) + } + + fn keys(&self) -> impl Iterator { + self.by_change.keys() + } +} + /// A document's signatures, by change. #[derive(Debug, Default)] pub(crate) struct Signatures { @@ -211,7 +249,7 @@ pub(crate) struct Signatures { /// (`newer_than`). stamp: u64, /// Received signatures of changes the document doesn't hold yet. - pending: HashMap, + pending: Pending, } /// An actor's latest change a signature verified. @@ -316,9 +354,7 @@ impl Signatures { let mut kept = 0; for (hash, signature) in received { if doc.get_change_meta_by_hash(&hash).is_none() { - if self.pending.len() < MAX_PENDING { - self.pending.insert(hash, signature); - } + self.pending.insert(hash, signature); } else if self.vouch(doc, hash, &signature) { self.insert(hash, signature); kept += 1; @@ -636,6 +672,23 @@ mod tests { assert_eq!(kept.since(0), vec![signed(second)]); } + /// Signatures waiting for their changes: past the bound, the oldest go. + #[test] + fn the_oldest_waiting_signatures_go() { + let mut pending = Pending::default(); + let hash = |i: usize| { + let mut bytes = [0; 32]; + bytes[..8].copy_from_slice(&(i as u64).to_le_bytes()); + ChangeHash(bytes) + }; + for i in 0..=MAX_PENDING { + pending.insert(hash(i), [0; 64]); + } + assert_eq!(pending.keys().count(), MAX_PENDING); + assert!(pending.remove(&hash(0)).is_none()); + assert!(pending.remove(&hash(MAX_PENDING)).is_some()); + } + /// Protocol data, listed paths (within or above them) and asked-for writes are /// signed; others aren't. #[test] -- cgit v1.3.1