aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/sync.rs')
-rw-r--r--src/swarm/sync.rs203
1 files changed, 55 insertions, 148 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index df4bb2d..3bfbf7d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -11,7 +11,7 @@ use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
use crate::locks::MutexExt;
-use crate::state::{AppState, Liveness, PeerState, Standing};
+use crate::state::{AppState, PeerState, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
use crate::swarm::source;
use crate::swarm::throughput::SyncBytes;
@@ -25,12 +25,6 @@ pub(crate) enum SyncEvent {
PeerNeedsSync(String),
}
-/// Store paths touched by one batch of document changes.
-#[derive(Debug)]
-pub(crate) struct ChangeSet {
- pub(crate) paths: Vec<Vec<String>>,
-}
-
async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
if state.remake.fetching_snapshot() {
// Rejoining: the snapshot comes as a download, not in a sync message. Sync
@@ -48,7 +42,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let validated_swarm = {
let peer_state = peer_state_mutex.lock().await;
- if matches!(peer_state.liveness, Liveness::Dormant { .. }) {
+ if peer_state.liveness.is_dormant() {
// Reactivation re-handshakes, and PeerValidated then restarts sync.
trace!("Skipping sync with dormant peer");
return Ok(());
@@ -74,12 +68,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
);
// Done earlier when the initial sync completes (every peer is enqueued
// then); this only matters if the source goes quiet.
- let tx = state.sync_events.clone();
- let peer = peer.to_owned();
- tokio::spawn(async move {
- tokio::time::sleep(wait).await;
- let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
- });
+ state.request_sync_after(peer, wait);
return Ok(());
}
@@ -139,7 +128,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let ready = state.is_ready().await;
// Before this message is counted.
let bytes_before = SyncBytes::of(&state.traffic, peer);
- let delta_threshold = state.blobs.delta_threshold();
+ // Lower towards a peer a message that large wouldn't reach whole: a slow link
+ // takes a delta chunk by chunk, where a message whole would fail and be sent
+ // again, whole, every time.
+ let rate = state
+ .throughput
+ .lock_or_recover()
+ .get(peer)
+ .map(|estimate| estimate.bytes_per_sec);
+ let failed = peer_state_mutex.lock().await.failed_whole(Instant::now());
+ let delta_threshold = state.blobs.delta_threshold_for(rate, failed);
// The session the message comes from, for its send's outcome.
let (session, generated) = tokio::task::spawn_blocking(
@@ -147,7 +145,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
// Label the message with the epoch of the document it was generated
// from, read under the same lock.
let (session, epoch, epoch_seq, body) = {
- let mut replica = doc.blocking_write();
+ let replica = doc.blocking_read();
let mut peer_state = psm.blocking_lock();
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
// Under the same locks as the epoch label: the message belongs to
@@ -155,9 +153,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let message =
peer_state
.sync
- .generate(&mut replica.doc, Instant::now(), bytes_before);
- let body = message.map(|message| {
- let payload = transport::SyncPayload::encode(message);
+ .generate(&replica.doc, Instant::now(), bytes_before);
+ let body = message.map(|payload| {
if payload.0.len() <= delta_threshold {
return transport::SyncBody::Message(payload);
}
@@ -210,6 +207,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let Some(encoded) = generated else {
return Ok(());
};
+ // Its answer is due.
+ let reply_due = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .reply
+ .as_ref()
+ .map(|wait| wait.deadline);
+ if let Some(at) = reply_due {
+ state.peers_due.by(at);
+ }
(session, encoded)
};
@@ -247,27 +255,22 @@ async fn send_generated(
);
if regenerate {
- state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
- .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+ state.request_sync(peer);
}
Ok(())
}
Err(error) => {
- let Some(delay) = peer_state_mutex
- .lock()
- .await
- .sync
- .send_failed(session, Instant::now())
- else {
+ let delay = {
+ let mut peer_state = peer_state_mutex.lock().await;
+ let now = Instant::now();
+ peer_state.whole_failed(encoded.payload_len, now);
+ peer_state.sync.send_failed(session, now)
+ };
+ let Some(delay) = delay else {
// The new session may not have opened yet.
debug!("Sync message refused, but a reset already discarded it");
- state
- .sync_events
- .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
- .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+ state.request_sync(peer);
return Ok(());
};
@@ -277,12 +280,7 @@ async fn send_generated(
"Sync delivery failed; generating a fresh message later"
);
- let tx = state.sync_events.clone();
- let peer = peer.to_owned();
- tokio::spawn(async move {
- tokio::time::sleep(delay).await;
- let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
- });
+ state.request_sync_after(peer, delay);
Ok(())
}
@@ -372,12 +370,13 @@ pub(crate) async fn handle_sync(
let message = payload;
let peer = peer.to_owned();
let clock = Arc::clone(&state.clock);
+ let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
// Some((caught_up, heads changed)) when applied, None when the message belongs
// to another epoch.
tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
// Before the lock: decoding a large message takes a while.
- let message = message.decode()?;
+ let message = crate::document::sync::decode(&message)?;
let mut replica = doc.blocking_write();
// Authoritative check, under the same lock as the apply: the document may
@@ -390,9 +389,9 @@ pub(crate) async fn handle_sync(
peer_state.reports_ready = Some(from_ready);
let heads_before = replica.heads().to_vec();
- peer_state
- .sync
- .receive(&mut replica.doc, message)
+ let sync = &mut peer_state.sync;
+ replica
+ .logged(|doc, log| (sync.receive_logged(doc, message, log), true))
.context("Automerge sync failed")?;
peer_state.last_sync_rx = Some(Instant::now());
if let Some(wait) = &peer_state.sync.reply {
@@ -408,16 +407,14 @@ pub(crate) async fn handle_sync(
stats.heads_seen(heads_after.len());
let changed = heads_after != heads_before;
if changed && measure_propagation {
- let now_ms = clock.now_millis().as_u64() as i64;
- // 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();
+ // 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;
+ let delays: Vec<i64> =
+ crate::document::changes::commit_times(&replica.doc, &heads_before)
+ .into_iter()
+ .map(|time| now_ms - time)
+ .collect();
stats.propagated(delays);
}
@@ -427,7 +424,7 @@ pub(crate) async fn handle_sync(
// catching up has them, and so does a ready peer whose document is empty
// (a misconfigured second swarm, or a genesis node in its first seconds,
// before it registered itself).
- check_caught_up && from_ready && peer_state.sync.have_all_theirs(&mut replica.doc),
+ check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
changed,
)))
})
@@ -449,6 +446,7 @@ pub(crate) async fn handle_sync(
.stats
.caught_up(state.traffic.total().received.sync.bytes);
source::finished(state, peer);
+ state.join_progress.notify_one();
}
state.notify_store_changed();
@@ -492,7 +490,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran
}
}
// Our next message says what we have now; the peer answers with the rest.
- let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer));
+ state.request_sync(&peer);
});
}
@@ -523,16 +521,9 @@ async fn download_and_apply_delta(
bail!("the epoch changed during the download");
}
let heads_before = replica.heads().to_vec();
- replica
- .doc
- .load_incremental(&bytes)
- .context("Loading the downloaded changes")?;
+ 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(&replica.doc, &spec.to).len();
if missing > 0 {
bail!("{missing} of the offered heads are still missing after loading");
}
@@ -623,87 +614,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) fn paths_related(a: &[String], b: &[String]) -> bool {
- a.starts_with(b) || b.starts_with(a)
-}
-
-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 }));
- }
-}