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.

1 files changed, 59 insertions, 6 deletions

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<Vec<(ChangeHash, [u8; 64])>> {
             .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<ChangeHash, [u8; 64]>,
+    /// The changes, oldest first; some may have left `by_change` since.
+    order: std::collections::VecDeque<ChangeHash>,
+}
+
+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<Item = &ChangeHash> {
+        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<ChangeHash, [u8; 64]>,
+    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]