aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api/document.rs7
-rw-r--r--src/document/changes.rs22
-rw-r--r--src/document/file.rs9
-rw-r--r--src/document/mod.rs24
-rw-r--r--src/document/replica.rs2
-rw-r--r--src/node.rs2
-rw-r--r--src/simtest/mod.rs4
-rw-r--r--src/state/mod.rs19
-rw-r--r--src/state/session.rs2
-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
15 files changed, 95 insertions, 86 deletions
diff --git a/src/api/document.rs b/src/api/document.rs
index 20bbc7b..08fd21f 100644
--- a/src/api/document.rs
+++ b/src/api/document.rs
@@ -50,12 +50,11 @@ pub(crate) async fn get_document(State(state): State<AppState>) -> Response {
// Both steps take time proportional to the document: off the async threads, and
// the lock only for the copy.
let saved = tokio::task::spawn_blocking(move || {
- let (mut copy, epoch) = {
+ let (mut copy, epoch, heads) = {
let replica = doc.blocking_read();
- (replica.doc.clone(), replica.epoch)
+ (replica.doc.clone(), replica.epoch, replica.heads().to_vec())
};
- let heads = copy.get_heads();
- (copy.save(), heads, epoch)
+ (crate::document::save(&mut copy), heads, epoch)
})
.await;
let (bytes, heads, epoch) = match saved {
diff --git a/src/document/changes.rs b/src/document/changes.rs
index d369c79..655063b 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -8,6 +8,26 @@ use anyhow::{Context, Result, anyhow, bail};
use super::{Change, ChangeHash, Document};
+/// Those of `heads` that `doc` doesn't have. (Looking a change up needs the document
+/// mutably.)
+pub(crate) fn missing(doc: &mut Document, heads: &[ChangeHash]) -> Vec<ChangeHash> {
+ heads
+ .iter()
+ .copied()
+ .filter(|head| doc.get_change_by_hash(head).is_none())
+ .collect()
+}
+
+/// When the changes after `since` were committed, in Unix milliseconds by their
+/// writers' clocks (`Replica::commit`); changes without a time (snapshots) left out.
+pub(crate) fn commit_times(doc: &mut Document, since: &[ChangeHash]) -> Vec<i64> {
+ doc.get_changes(since)
+ .iter()
+ .map(|change| change.timestamp())
+ .filter(|&time| time > 0)
+ .collect()
+}
+
/// The change `hash` as the document encodes it; None if `doc` doesn't have it.
pub(crate) fn change_bytes(doc: &mut Document, hash: &ChangeHash) -> Option<Vec<u8>> {
doc.get_change_by_hash(hash)
@@ -33,7 +53,7 @@ pub(crate) fn delta_bytes(
from: &[ChangeHash],
to: &[ChangeHash],
) -> Option<Vec<u8>> {
- if to.iter().any(|head| doc.get_change_by_hash(head).is_none()) {
+ if !missing(doc, to).is_empty() {
return None;
}
// Heads of `from` we don't have can't hide anything we have.
diff --git a/src/document/file.rs b/src/document/file.rs
index 47d7310..e03c2e6 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -99,7 +99,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
}
let (record, after) = after.split_at(length);
match &mut doc {
- None => doc = Some(AutoCommit::load(record).context("the full save")?),
+ None => doc = Some(super::load(record).context("the full save")?),
Some(doc) => {
doc.load_incremental(record).context("an increment")?;
}
@@ -109,10 +109,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
let Some(doc) = doc else {
bail!("no full save");
};
- let loaded = Loaded {
- doc: doc.with_actor(ActorId::random()),
- actors,
- };
+ let loaded = Loaded { doc, actors };
Ok((loaded, records.len()))
}
@@ -250,7 +247,7 @@ impl DocumentFile {
.into_iter()
.filter(|actor| *actor == current || present.contains(actor))
.collect();
- let save = copy.save();
+ let save = super::save(&mut copy);
let bytes = whole(&actors, &save)?;
writer
.files
diff --git a/src/document/mod.rs b/src/document/mod.rs
index 1612145..2230e47 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -6,6 +6,30 @@
/// Only `document` and `remake` use Automerge itself (`architecture.md`, The document).
pub(crate) use automerge::{ActorId, AutoCommit as Document, Change, ChangeHash};
+/// The whole document as bytes, as the engine saves it: a backup, the document
+/// file's full save, the size a joining node downloads.
+pub(crate) fn save(doc: &mut Document) -> Vec<u8> {
+ doc.save()
+}
+
+/// The changes after `since` as bytes that, appended to a save, load with it (the
+/// document file's appends).
+pub(crate) fn save_since(doc: &mut Document, since: &[ChangeHash]) -> Vec<u8> {
+ doc.save_after(since)
+}
+
+/// A saved document (`save`), loaded to write to under a writer ID of its own: one ID
+/// must never write in two places, and a save may be loaded on several nodes, or again
+/// after the node that saved it wrote on.
+pub(crate) fn load(bytes: &[u8]) -> anyhow::Result<Document> {
+ Ok(Document::load(bytes)?.with_actor(ActorId::random()))
+}
+
+/// The actor (writer ID) `doc` writes under.
+pub(crate) fn actor(doc: &Document) -> ActorId {
+ doc.get_actor().clone()
+}
+
pub(crate) mod changes;
pub(crate) mod file;
pub(crate) mod lock;
diff --git a/src/document/replica.rs b/src/document/replica.rs
index 5d3d7a7..5a87016 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -101,7 +101,7 @@ impl fmt::Debug for Epoch {
/// The heads are cached: `AutoCommit::get_heads` takes `&mut self` (it closes a pending
/// transaction), so without the cache every keepalive and status read would need the
/// write lock. Every path that changes the document under the write lock calls
-/// [`Replica::refresh_heads`] before releasing it (`AppState::write_doc`, `mutate_doc`,
+/// [`Replica::refresh_heads`] before releasing it (`AppState::mutate_doc`,
/// `switch_epoch`, the sync apply); readers take the read lock and use
/// [`Replica::heads`].
///
diff --git a/src/node.rs b/src/node.rs
index f09de5b..8224c7a 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -72,7 +72,7 @@ impl Node {
let (epoch_tx, _) = watch::channel(config.initial_epoch);
let firewall = Arc::new(Firewall::default());
let traffic = Arc::new(Traffic::default());
- let initial_actor = config.initial_doc.get_actor().clone();
+ let initial_actor = crate::document::actor(&config.initial_doc);
let transport = match config.identity {
Some(identity) => Arc::new(IdentityTransport::new(
config.transport,
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 9642f28..aa30e7c 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -253,11 +253,11 @@ impl Sim {
ago(p.last_seen),
ago(p.last_sync_rx),
p.is_unreachable(),
- p.sync.automerge_view().0.map(|heads| heads
+ p.sync.sync_view().0.map(|heads| heads
.iter()
.map(|head| head.to_string()[..8].to_owned())
.collect::<Vec<_>>()),
- p.sync.automerge_view().1,
+ p.sync.sync_view().1,
p.sync.sending.is_some(),
p.sync
.reply
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 673bd56..8ac1124 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -394,25 +394,6 @@ impl AppState {
.expect("document read task panicked")
}
- /// Mutates the document on the blocking pool.
- pub(crate) async fn write_doc<T>(
- &self,
- f: impl FnOnce(&mut Document) -> T + Send + 'static,
- ) -> T
- where
- T: Send + 'static,
- {
- let doc = Arc::clone(&self.doc);
- tokio::task::spawn_blocking(move || {
- let mut replica = doc.blocking_write();
- let out = f(&mut replica.doc);
- replica.refresh_heads();
- out
- })
- .await
- .expect("document write task panicked")
- }
-
/// Mutates the document on the blocking pool as one change, then
/// publishes it. Write and notification run in the same blocking task, so a
/// cancelled request can't leave a committed write unannounced.
diff --git a/src/state/session.rs b/src/state/session.rs
index bf5c83f..7305c58 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -245,7 +245,7 @@ impl SyncSession {
/// For the simulator's dump: the peer's heads as Automerge knows them, and whether
/// it thinks a message is in flight.
#[cfg(test)]
- pub(crate) fn automerge_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) {
+ pub(crate) fn sync_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) {
(self.state.their_heads.as_deref(), self.state.in_flight)
}
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");
}