aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/bootstrap.rs6
-rw-r--r--src/swarm/changes.rs49
-rw-r--r--src/swarm/document_file.rs2
-rw-r--r--src/swarm/relay.rs11
-rw-r--r--src/swarm/stats.rs2
-rw-r--r--src/swarm/sync.rs20
6 files changed, 39 insertions, 51 deletions
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index ec9a355..d2bbe24 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -300,15 +300,13 @@ fn identify(
})
}
-/// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor.
+/// A backup to recover from, as `GET /admin/document` saves it (`document::load`).
fn load_backup(path: &Path, files: &Files) -> Result<Document> {
let bytes = files
.read(FileKind::Backup, path)
.with_context(|| format!("Reading the backup {}", path.display()))?
.with_context(|| format!("The backup {} doesn't exist", path.display()))?;
- Ok(Document::load(&bytes)
- .with_context(|| format!("Loading the backup {}", path.display()))?
- .with_actor(ActorId::random()))
+ crate::document::load(&bytes).with_context(|| format!("Loading the backup {}", path.display()))
}
/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
index 6831081..3a50769 100644
--- a/src/swarm/changes.rs
+++ b/src/swarm/changes.rs
@@ -6,42 +6,47 @@ use std::sync::Arc;
use tracing::{info, warn};
+use crate::document::changes;
use crate::document::store::{self, ChangeSet, InLists};
use crate::state::AppState;
pub(crate) async fn run_change_publisher(state: AppState) {
let mut changed = state.change_tx.subscribe();
- let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
+ // The replica's heads (sorted, as of its last refresh: every change refreshes them).
+ let mut last_heads = state.doc.read().await.heads().to_vec();
info!("Starting change publisher");
while changed.changed().await.is_ok() {
let before = last_heads.clone();
- let (heads, paths, conflicts) = state
- .write_doc(move |doc| {
- let heads = doc.get_heads();
- if heads == before {
- return (heads, Vec::new(), Vec::new());
- }
+ let doc = Arc::clone(&state.doc);
+ // Looking changes up needs the document mutably.
+ let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
+ let mut replica = doc.blocking_write();
+ let heads = replica.heads().to_vec();
+ if heads == before {
+ return (heads, Vec::new(), Vec::new());
+ }
- // After an epoch switch the previous heads belong to the old document:
- // report the root as changed, which every stream is related to.
- if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
- return (heads, vec![Vec::new()], Vec::new());
- }
+ // After an epoch switch the previous heads belong to the old document:
+ // report the root as changed, which every stream is related to.
+ if !changes::missing(&mut replica.doc, &before).is_empty() {
+ return (heads, vec![Vec::new()], Vec::new());
+ }
- let mut paths = Vec::new();
- let mut conflicts = Vec::new();
- for changed in store::changed(doc, &before, &heads, InLists::Index) {
- if changed.conflict {
- conflicts.push(changed.path.clone());
- }
- paths.push(changed.path);
+ let mut paths = Vec::new();
+ let mut conflicts = Vec::new();
+ for changed in store::changed(&mut replica.doc, &before, &heads, InLists::Index) {
+ if changed.conflict {
+ conflicts.push(changed.path.clone());
}
- (heads, paths, conflicts)
- })
- .await;
+ paths.push(changed.path);
+ }
+ (heads, paths, conflicts)
+ })
+ .await
+ .expect("change publisher task panicked");
if heads == last_heads {
continue;
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
index 188b2b6..435ff32 100644
--- a/src/swarm/document_file.rs
+++ b/src/swarm/document_file.rs
@@ -47,7 +47,7 @@ pub(crate) async fn flush(state: &AppState) {
return;
}
let heads = replica.heads().to_vec();
- (heads, replica.doc.save_after(&since))
+ (heads, crate::document::save_since(&mut replica.doc, &since))
};
file.append(since, heads, increment).await;
}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index c28c9f1..d11c3e0 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -130,16 +130,9 @@ pub(crate) async fn pull_if_behind(
}
// Those of the listed heads we lack. With a digest only a sample is listed; if we
// have all of it, a later keepalive lists others.
- let missing: Vec<_> = {
- // Document's lookups take &mut self.
+ let missing = {
let mut replica = state.doc.write().await;
- remote
- .heads
- .listed()
- .iter()
- .filter(|head| replica.doc.get_change_by_hash(head).is_none())
- .copied()
- .collect()
+ crate::document::changes::missing(&mut replica.doc, remote.heads.listed())
};
if missing.is_empty() {
return false;
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index b5548d5..2398b09 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -279,7 +279,7 @@ pub(crate) fn measure_saved_size(state: &AppState) {
// Both steps take time proportional to the document: off the async threads.
let measured = match tokio::task::spawn_blocking(move || {
let mut copy = doc.blocking_read().doc.clone();
- copy.save().len() as u64
+ crate::document::save(&mut copy).len() as u64
})
.await
{
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 40e577a..3cbe5f3 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -389,15 +389,11 @@ pub(crate) async fn handle_sync(
// Now by the peers' clocks (their median): this node's own clock skew
// doesn't count; the writer's still does.
let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
- // Time 0: not stamped (snapshots, older nodes).
- let delays: Vec<i64> = replica
- .doc
- .get_changes(&heads_before)
- .iter()
- .map(|change| change.timestamp())
- .filter(|&time| time > 0)
- .map(|time| now_ms - time)
- .collect();
+ let delays: Vec<i64> =
+ crate::document::changes::commit_times(&mut replica.doc, &heads_before)
+ .into_iter()
+ .map(|time| now_ms - time)
+ .collect();
stats.propagated(delays);
}
@@ -505,11 +501,7 @@ async fn download_and_apply_delta(
let heads_before = replica.heads().to_vec();
crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
replica.refresh_heads();
- let missing = spec
- .to
- .iter()
- .filter(|head| replica.doc.get_change_by_hash(head).is_none())
- .count();
+ let missing = crate::document::changes::missing(&mut replica.doc, &spec.to).len();
if missing > 0 {
bail!("{missing} of the offered heads are still missing after loading");
}