1 files changed, 50 insertions, 8 deletions

diff --git a/src/compaction/local.rs b/src/compaction/local.rs
index 869c94e..adf8764 100644
--- a/src/compaction/local.rs
+++ b/src/compaction/local.rs
@@ -6,7 +6,8 @@
 
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Mutex, MutexGuard};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
 
 use automerge::AutoCommit;
 use serde::Serialize;
@@ -34,11 +35,24 @@ struct Inner {
     first_seen: HashMap<String, Instant>,
     status: HashMap<String, LocalStatus>,
     archive: Vec<Archived>,
-    /// The snapshot request this node is waiting for (peer, wanted epoch).
-    awaiting: Option<(String, Epoch, oneshot::Sender<SnapshotReply>)>,
+    /// The chunk this node is waiting for: (peer, wanted epoch, offset).
+    awaiting: Option<(String, Epoch, u64, oneshot::Sender<SnapshotReply>)>,
     catch_up: Option<CatchUp>,
+    /// The saved document chunks are served from (one download source at a time).
+    served: Option<Arc<Served>>,
 }
 
+/// A saved copy of this node's document, served to downloading peers.
+pub(crate) struct Served {
+    pub(crate) epoch: Epoch,
+    pub(crate) bytes: Vec<u8>,
+    pub(crate) digest: [u8; 32],
+    pub(crate) saved_at: Instant,
+}
+
+/// A served copy is dropped this long after it was saved.
+const SERVED_KEPT_FOR: Duration = Duration::from_secs(15 * 60);
+
 /// Fetching a newer epoch from a peer.
 #[derive(Clone, Debug)]
 pub(crate) struct CatchUp {
@@ -140,26 +154,41 @@ impl Compaction {
         prune(&mut inner.archive, current_generation);
     }
 
+    /// Drops archives past their retention and a served copy nobody needs any more.
     pub(crate) fn prune_archive(&self, current_generation: u64) {
-        prune(&mut self.lock().archive, current_generation);
+        let mut inner = self.lock();
+        prune(&mut inner.archive, current_generation);
+        if inner
+            .served
+            .as_ref()
+            .is_some_and(|served| served.saved_at.elapsed() > SERVED_KEPT_FOR)
+        {
+            inner.served = None;
+        }
     }
 
-    /// Registers interest in a snapshot reply from `peer`, replacing any earlier one.
+    /// Registers interest in one chunk from `peer`, replacing any earlier request.
     pub(crate) fn await_reply(
         &self,
         peer: &str,
         wanted: Epoch,
+        offset: u64,
     ) -> oneshot::Receiver<SnapshotReply> {
         let (sender, receiver) = oneshot::channel();
-        self.lock().awaiting = Some((peer.to_owned(), wanted, sender));
+        self.lock().awaiting = Some((peer.to_owned(), wanted, offset, sender));
         receiver
     }
 
-    /// Hands a reply to the waiting fetch. False if nobody asked that peer.
+    /// Hands a reply to the waiting download. False unless it answers exactly the
+    /// pending request (late replies to earlier attempts are dropped).
     pub(crate) fn deliver_reply(&self, peer: &str, reply: SnapshotReply) -> bool {
         let mut inner = self.lock();
         match inner.awaiting.take() {
-            Some((expected, _, sender)) if expected == peer => sender.send(reply).is_ok(),
+            Some((expected, wanted, offset, sender))
+                if expected == peer && wanted == reply.wanted && offset == reply.offset =>
+            {
+                sender.send(reply).is_ok()
+            }
             other => {
                 inner.awaiting = other;
                 false
@@ -167,6 +196,19 @@ impl Compaction {
         }
     }
 
+    /// The copy served for `epoch`, if any.
+    pub(crate) fn served(&self, epoch: Epoch) -> Option<Arc<Served>> {
+        self.lock()
+            .served
+            .as_ref()
+            .filter(|served| served.epoch == epoch)
+            .cloned()
+    }
+
+    pub(crate) fn set_served(&self, served: Arc<Served>) {
+        self.lock().served = Some(served);
+    }
+
     pub(crate) fn stop_awaiting(&self) {
         self.lock().awaiting = None;
     }