1 files changed, 95 insertions, 0 deletions
diff --git a/src/remake/content.rs b/src/remake/content.rs
new file mode 100644
index 0000000..e5d78c5
--- /dev/null
+++ b/src/remake/content.rs
@@ -0,0 +1,95 @@
+//! Content of remakes: the new document of a `replace` or `import` remake.
+//!
+//! The proposal records only a reference (BLAKE3 hash and size). The coordinator keeps
+//! the bytes in memory; every member fetches them before voting (in chunks, from
+//! several peers at once: `swarm::blob`) and serves them in turn. A committed remake
+//! therefore has its content on a majority of the electorate. Nodes that switch need
+//! it to build the snapshot; nodes that rejoin later don't (they download the snapshot
+//! itself).
+//!
+//! Content no open remake refers to is dropped an hour after it was stored.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use super::Host;
+use super::records::{ContentRef, Record};
+
+/// The content `reference` names, if this node has it; otherwise starts fetching it in
+/// the background (once) and returns None.
+pub(super) fn available_or_fetch<H: Host>(
+    host: &H,
+    reference: &ContentRef,
+) -> Option<Arc<Vec<u8>>> {
+    let hash = reference.hash()?;
+    if let Some(bytes) = host.remake().content(&hash) {
+        return Some(bytes);
+    }
+    if host.remake().start_content_fetch(hash) {
+        let host = host.clone();
+        let reference = reference.clone();
+        tokio::spawn(async move {
+            fetch(&host, hash, &reference).await;
+            host.remake().end_content_fetch(&hash);
+        });
+    }
+    None
+}
+
+/// The content `reference` names: from this node, or fetched now. Waits for a download
+/// already running.
+pub(super) async fn ensure<H: Host>(
+    host: &H,
+    reference: &ContentRef,
+) -> Result<Arc<Vec<u8>>, String> {
+    let hash = reference
+        .hash()
+        .ok_or_else(|| format!("unreadable content hash {:?}", reference.blake3))?;
+    loop {
+        if let Some(bytes) = host.remake().content(&hash) {
+            return Ok(bytes);
+        }
+        if host.remake().start_content_fetch(hash) {
+            let fetched = fetch(host, hash, reference).await;
+            host.remake().end_content_fetch(&hash);
+            return fetched
+                .ok_or_else(|| format!("couldn't download content {}", reference.blake3));
+        }
+        // Someone else is downloading it.
+        while host.remake().content_fetch_running(&hash) {
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    }
+}
+
+async fn fetch<H: Host>(host: &H, hash: [u8; 32], reference: &ContentRef) -> Option<Arc<Vec<u8>>> {
+    match host.fetch_content(hash).await {
+        Ok(bytes) if bytes.len() as u64 == reference.size => {
+            info!(content = %reference.blake3, bytes = bytes.len(), "Downloaded remake content");
+            let bytes = Arc::new(bytes);
+            host.remake().put_content(hash, Arc::clone(&bytes));
+            Some(bytes)
+        }
+        Ok(bytes) => {
+            warn!(content = %reference.blake3, bytes = bytes.len(), expected = reference.size, "Remake content has the wrong size");
+            None
+        }
+        Err(error) => {
+            warn!(content = %reference.blake3, error = %format!("{error:#}"), "Couldn't download remake content");
+            None
+        }
+    }
+}
+
+/// Drops content no open remake in `records` refers to (after its grace period).
+pub(super) fn prune<H: Host>(host: &H, records: &[(String, Record)], now: u64, grace: u64) {
+    let referenced: HashSet<[u8; 32]> = records
+        .iter()
+        .filter(|(_, record)| record.phase(now, grace).is_open())
+        .filter_map(|(_, record)| record.proposal.content.as_ref()?.hash())
+        .collect();
+    host.remake().prune_contents(&referenced);
+}