summaryrefslogtreecommitdiffstats
path: root/src/remake/local.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/local.rs')
-rw-r--r--src/remake/local.rs48
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()