1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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);
}
|
