diff options
| -rw-r--r-- | src/node.rs | 4 | ||||
| -rw-r--r-- | src/swarm/changes.rs | 88 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 81 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 2 |
5 files changed, 92 insertions, 84 deletions
diff --git a/src/node.rs b/src/node.rs index 8adab2b..acf1dc5 100644 --- a/src/node.rs +++ b/src/node.rs @@ -21,7 +21,7 @@ use crate::state::{AppState, NodeLifecycle, Timing}; use crate::swarm::auth::SwarmKey; use crate::swarm::state_file::StateFile; use crate::swarm::sync::SyncEvent; -use crate::swarm::{self, peer, sync}; +use crate::swarm::{self, changes, peer, sync}; use crate::transport::Transport; use crate::transport::firewall::Firewall; use crate::transport::identity::{IdentityConfig, IdentityTransport}; @@ -194,6 +194,6 @@ impl Node { /// from then on, and it avoids diffing the initial sync) and the document file writer. pub(crate) fn spawn_ready_workers(state: &AppState) { let publisher_state = state.clone(); - tokio::spawn(async move { sync::run_change_publisher(publisher_state).await }); + tokio::spawn(async move { changes::run_change_publisher(publisher_state).await }); tokio::spawn(crate::document::file::run(state.clone())); } diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs new file mode 100644 index 0000000..10a3bfe --- /dev/null +++ b/src/swarm/changes.rs @@ -0,0 +1,88 @@ +//! The change publisher: after document changes, which store paths they touched +//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and +//! a nudge to topology when the member list changed. + +use std::sync::Arc; + +use tracing::{info, warn}; + +use crate::document::store::ChangeSet; +use crate::state::AppState; + +fn patch_store_path(patch: &automerge::Patch) -> Vec<String> { + let mut path: Vec<String> = patch + .path + .iter() + .map(|(_, prop)| match prop { + automerge::Prop::Map(key) => key.clone(), + automerge::Prop::Seq(index) => index.to_string(), + }) + .collect(); + + match &patch.action { + automerge::PatchAction::PutMap { key, .. } + | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()), + // List, text and counter changes: the containing object changed. + _ => {} + } + path +} + +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; + + 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()); + } + + // 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()); + } + + let mut paths = Vec::new(); + let mut conflicts = Vec::new(); + for patch in doc.diff(&before, &heads) { + let path = patch_store_path(&patch); + if matches!( + patch.action, + automerge::PatchAction::PutMap { conflict: true, .. } + ) { + conflicts.push(path.clone()); + } + paths.push(path); + } + (heads, paths, conflicts) + }) + .await; + + if heads == last_heads { + continue; + } + last_heads = heads; + + for path in &conflicts { + warn!(path = ?path, "Concurrent writes produced a conflict"); + } + + if paths + .iter() + .any(|path| crate::swarm::topology::affects_neighbours(path)) + { + state.topology_changed.notify_one(); + } + + // Fails only when nobody is subscribed. + let _ = state.change_events.send(Arc::new(ChangeSet { paths })); + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 7fb8e16..9cdf154 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -4,6 +4,7 @@ pub(crate) mod auth; pub(crate) mod blob; pub(crate) mod bootstrap; +pub(crate) mod changes; pub(crate) mod links; pub(crate) mod members; pub(crate) mod pair; diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 05c0c34..e89c2e2 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -10,7 +10,6 @@ use tokio::sync::mpsc; use tokio::time::Instant; use tracing::{debug, info, trace, warn}; -use crate::document::store::ChangeSet; use crate::locks::MutexExt; use crate::state::{AppState, Liveness, PeerState, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; @@ -618,83 +617,3 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece } } } - -// --- change publisher -------------------------------------------------------- - -fn patch_store_path(patch: &automerge::Patch) -> Vec<String> { - let mut path: Vec<String> = patch - .path - .iter() - .map(|(_, prop)| match prop { - automerge::Prop::Map(key) => key.clone(), - automerge::Prop::Seq(index) => index.to_string(), - }) - .collect(); - - match &patch.action { - automerge::PatchAction::PutMap { key, .. } - | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()), - // List, text and counter changes: the containing object changed. - _ => {} - } - path -} - -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; - - 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()); - } - - // 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()); - } - - let mut paths = Vec::new(); - let mut conflicts = Vec::new(); - for patch in doc.diff(&before, &heads) { - let path = patch_store_path(&patch); - if matches!( - patch.action, - automerge::PatchAction::PutMap { conflict: true, .. } - ) { - conflicts.push(path.clone()); - } - paths.push(path); - } - (heads, paths, conflicts) - }) - .await; - - if heads == last_heads { - continue; - } - last_heads = heads; - - for path in &conflicts { - warn!(path = ?path, "Concurrent writes produced a conflict"); - } - - if paths - .iter() - .any(|path| crate::swarm::topology::affects_neighbours(path)) - { - state.topology_changed.notify_one(); - } - - // Fails only when nobody is subscribed. - let _ = state.change_events.send(Arc::new(ChangeSet { paths })); - } -} diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 063fc41..c861101 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -161,7 +161,7 @@ pub(crate) struct NeighbourChange { } /// Recomputes the neighbours whenever something they depend on may have changed -/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`), +/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`), /// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a /// fallback. A burst of notifications coalesces into one recomputation. pub(crate) async fn run_worker(state: AppState) { |