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]