diff options
Diffstat (limited to 'src/remake/catchup.rs')
| -rw-r--r-- | src/remake/catchup.rs | 49 |
1 files changed, 47 insertions, 2 deletions
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index db25251..42b6526 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -7,8 +7,10 @@ //! 1. stops being ready and keeps its document aside; reads are answered from a copy //! of it, marked stale (`X-Continuum-Stale`), writes get `503`; //! 2. switches to an empty document labelled with the newer epoch: an empty history -//! conflicts with nothing, so ordinary sync fills it from peers on that epoch, -//! exactly like a new node joining; +//! conflicts with nothing. It downloads the epoch's snapshot change in chunks from +//! several peers on that epoch at once (`swarm::blob`; sync waits meanwhile), then +//! ordinary sync adds the changes made since, exactly like a new node joining. If +//! the download fails, sync transfers the snapshot too, in one message; //! 3. once caught up with a peer, checks that the history starts with the epoch's //! snapshot change; //! 4. decides about its own changes the swarm never saw (none, for a new node): @@ -50,6 +52,10 @@ const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60); pub(super) async fn check<H: Host>(host: &H) { if let Some((target, started)) = host.remake().rejoin_progress() { let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT); + if host.remake().fetching_snapshot() { + // The download has its own stall timeout; the clock restarts after it. + return; + } if host.initial_sync_done() { finish(host).await; } else if started.elapsed() > timeout { @@ -147,6 +153,45 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { started: Instant::now(), was_ready, }); + + // Sync waits until the snapshot is here (or its download failed). + host.remake().set_fetching_snapshot(Some(target)); + tokio::spawn(fetch_snapshot(host.clone(), target)); +} + +/// Downloads the snapshot change of `target` and applies it, then lets sync go on. +async fn fetch_snapshot<H: Host>(host: H, target: Epoch) { + let outcome = host.fetch_change(ChangeHash(target.id)).await; + let still_rejoining = host + .remake() + .rejoin_progress() + .is_some_and(|(rejoining, _)| rejoining == target) + && host.current_epoch() == target; + if still_rejoining { + match outcome { + Ok(change) => { + let applied = host + .commit_doc(move |doc| doc.apply_changes([change])) + .await; + match applied { + Ok(()) => { + info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since") + } + Err(error) => { + warn!(epoch = %target, error = %error, "Couldn't apply the downloaded snapshot; syncing it instead") + } + } + } + Err(error) => warn!( + epoch = %target, + error = %format!("{error:#}"), + "Couldn't download the epoch's snapshot; syncing it instead" + ), + } + host.remake().restart_rejoin_clock(target); + } + host.remake().set_fetching_snapshot(None); + host.request_sync(); } /// What to do with this node's own changes the new epoch lacks. |
