1 files changed, 95 insertions, 6 deletions

diff --git a/src/remake/local.rs b/src/remake/local.rs
index 35a4ff1..04ac0b0 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -1,27 +1,39 @@
 //! 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;
 use tokio::time::Instant;
 
-use super::records::{unix_millis, Retention};
+use super::records::{Retention, unix_millis};
 use super::snapshot::Snapshot;
 use crate::document::replica::Epoch;
 
-#[derive(Default)]
 pub(crate) struct RemakeState {
+    /// When this node started (Unix milliseconds): votes it cast before may be lost
+    /// (`vote`).
+    started_at: u64,
     // A std mutex: every access is short and never spans an await.
     inner: Mutex<Inner>,
 }
 
+impl Default for RemakeState {
+    fn default() -> Self {
+        Self {
+            started_at: unix_millis(),
+            inner: Mutex::default(),
+        }
+    }
+}
+
 #[derive(Default)]
 struct Inner {
     /// Built by the coordinator at commit time; consumed by the local switch.
@@ -34,9 +46,21 @@ struct Inner {
     catch_up: Option<CatchUp>,
     /// A rejoin in progress: the node syncs an empty document of a newer epoch.
     rejoin: Option<Rejoin>,
+    /// 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]>,
+    /// Remakes this node coordinates now (`coordinator`).
+    coordinating: HashSet<String>,
 }
 
+/// 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 {
@@ -119,6 +143,23 @@ pub(crate) struct ArchivedEpoch {
 }
 
 impl RemakeState {
+    pub(crate) fn started_at(&self) -> u64 {
+        self.started_at
+    }
+
+    /// Whether this node starts coordinating `id` now (false if it already does).
+    pub(crate) fn start_coordinating(&self, id: &str) -> bool {
+        self.lock().coordinating.insert(id.to_owned())
+    }
+
+    pub(crate) fn end_coordinating(&self, id: &str) {
+        self.lock().coordinating.remove(id);
+    }
+
+    pub(crate) fn is_coordinating(&self, id: &str) -> bool {
+        self.lock().coordinating.contains(id)
+    }
+
     fn lock(&self) -> MutexGuard<'_, Inner> {
         self.inner
             .lock()
@@ -211,6 +252,54 @@ 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()
+    }
+
+    pub(crate) fn set_fetching_snapshot(&self, target: Option<Epoch>) {
+        self.lock().fetching_snapshot = target;
+    }
+
+    /// The rejoin of `target`, if still in progress, gets its full time for the sync
+    /// that follows the download.
+    pub(crate) fn restart_rejoin_clock(&self, target: Epoch) {
+        if let Some(rejoin) = self.lock().rejoin.as_mut().filter(|r| r.target == target) {
+            rejoin.started = Instant::now();
+        }
+    }
+
     pub(crate) fn add_quarantine(&self, quarantined: Quarantined) {
         self.lock().quarantine.push(quarantined);
     }