put remake content into blobs

1 files changed, 44 insertions, 4 deletions

diff --git a/src/remake/local.rs b/src/remake/local.rs
index 787bd5e..ec2f70a 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -1,12 +1,13 @@
 //! Per-node, in-memory remake state: snapshots built ahead of the switch, switch
-//! progress for reports, a rejoin in progress, quarantined changes, and archived
-//! documents of previous epochs.
+//! progress for reports, a rejoin in progress, quarantined changes, archived documents
+//! of previous epochs, and the content of remakes (`content`).
 //!
 //! Nothing here is replicated or durable. After a restart the node simply has no
 //! archive or quarantine, and rebuilds snapshots when needed.
 
-use std::collections::HashMap;
-use std::sync::{Mutex, MutexGuard};
+use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
 
 use automerge::AutoCommit;
 use serde::Serialize;
@@ -37,8 +38,16 @@ struct Inner {
     /// While the rejoin downloads the target epoch's snapshot (`catchup`), sync waits.
     fetching_snapshot: Option<Epoch>,
     quarantine: Vec<Quarantined>,
+    /// Content of remakes, by BLAKE3 hash, and when it was stored (served to peers).
+    contents: HashMap<[u8; 32], (Arc<Vec<u8>>, Instant)>,
+    /// Content being downloaded.
+    content_fetches: HashSet<[u8; 32]>,
 }
 
+/// Content no remake refers to any more is kept this long after it was stored, for
+/// nodes that switch late.
+const CONTENT_GRACE: Duration = Duration::from_secs(3600);
+
 /// Catching up with a newer epoch: attempts and the last failure.
 #[derive(Clone, Debug)]
 pub(crate) struct CatchUp {
@@ -213,6 +222,37 @@ impl RemakeState {
         self.lock().rejoin.take()
     }
 
+    pub(crate) fn content(&self, hash: &[u8; 32]) -> Option<Arc<Vec<u8>>> {
+        self.lock()
+            .contents
+            .get(hash)
+            .map(|(bytes, _)| Arc::clone(bytes))
+    }
+
+    pub(crate) fn put_content(&self, hash: [u8; 32], bytes: Arc<Vec<u8>>) {
+        self.lock().contents.insert(hash, (bytes, Instant::now()));
+    }
+
+    /// Marks a download of `hash` as started; false if one already is.
+    pub(crate) fn start_content_fetch(&self, hash: [u8; 32]) -> bool {
+        self.lock().content_fetches.insert(hash)
+    }
+
+    pub(crate) fn end_content_fetch(&self, hash: &[u8; 32]) {
+        self.lock().content_fetches.remove(hash);
+    }
+
+    pub(crate) fn content_fetch_running(&self, hash: &[u8; 32]) -> bool {
+        self.lock().content_fetches.contains(hash)
+    }
+
+    /// Drops content that no remake in `referenced` needs, once its grace has passed.
+    pub(crate) fn prune_contents(&self, referenced: &HashSet<[u8; 32]>) {
+        self.lock().contents.retain(|hash, (_, stored)| {
+            referenced.contains(hash) || stored.elapsed() < CONTENT_GRACE
+        });
+    }
+
     /// Whether sync should wait: a rejoin is downloading its epoch's snapshot.
     pub(crate) fn fetching_snapshot(&self) -> bool {
         self.lock().fetching_snapshot.is_some()