diff options
Diffstat (limited to 'src/compaction/switch.rs')
| -rw-r--r-- | src/compaction/switch.rs | 248 |
1 files changed, 0 insertions, 248 deletions
diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs deleted file mode 100644 index 24bcee0..0000000 --- a/src/compaction/switch.rs +++ /dev/null @@ -1,248 +0,0 @@ -//! Switching this node to a committed epoch. -//! -//! The worker watches the compaction records. When a compaction of the local epoch is -//! committed, it waits one keepalive interval (so the commit record can reach peers -//! through ordinary sync before this node stops syncing the old epoch), then: -//! -//! 1. clones the live document under the read lock; -//! 2. without any lock: takes the coordinator's prebuilt snapshot or builds it, checks -//! that its hash is the committed epoch ID, and replays the clone's changes since -//! the base into it; -//! 3. under the write lock: replays whatever arrived since the clone, swaps document and -//! epoch, and hands back the old document; -//! 4. archives the old document according to the policy. -//! -//! The write lock is held only for step 3, which is proportional to the changes made -//! during steps 1-2. -//! -//! A node whose build doesn't match the committed epoch, or that never saw the commit, -//! rejoins the new epoch from peers instead (`catchup`). - -use std::time::Duration; - -use automerge::{ActorId, AutoCommit, ChangeHash}; -use tokio::sync::broadcast; -use tracing::{error, info, warn}; - -use super::catchup; -use super::delta::{self, ReplayStats}; -use super::local::LocalStatus; -use super::records::{self, unix_millis, EpochRecord, Record}; -use super::snapshot::{self, COMPACTIONS, META_ROOT}; -use super::Host; -use crate::replica::Epoch; -use crate::store; -use crate::sync::paths_related; - -const TICK: Duration = Duration::from_secs(1); - -pub(crate) async fn run_worker<H: Host>(host: H) { - info!("Starting compaction worker"); - - let records_dir = vec![META_ROOT.to_owned(), COMPACTIONS.to_owned()]; - let mut changes = host.subscribe_changes(); - let mut changes_closed = false; - let mut tick = tokio::time::interval(TICK); - tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - - loop { - // React promptly to changes of the records; the timer covers the switch delay. - let should_check = tokio::select! { - _ = tick.tick() => true, - received = changes.recv(), if !changes_closed => match received { - Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)), - Err(broadcast::error::RecvError::Lagged(_)) => true, - Err(broadcast::error::RecvError::Closed) => { - warn!("Change events closed; compaction worker continues on its timer only"); - changes_closed = true; - false - } - } - }; - - if should_check { - check(&host).await; - } - } -} - -async fn check<H: Host>(host: &H) { - let local = host.current_epoch(); - host.compaction().prune_archive(local.generation); - - // While rejoining, the document is being filled from peers: its records are not - // complete enough to act on. - if host.compaction().is_rejoining() { - return catchup::check(host).await; - } - - let all = host.read_doc(records::load_all).await; - let local_record = EpochRecord::from(local); - - let committed: Vec<(String, Record)> = all - .into_iter() - .filter(|(_, record)| { - record.proposal.from_epoch == local_record && record.committed_epoch().is_some() - }) - .collect(); - - let (id, record) = match committed.as_slice() { - // Nothing committed here; maybe peers moved on without us. - [] => return catchup::check(host).await, - [(id, record)] => (id.clone(), record.clone()), - several => { - // The settle protocol prevents this; never guess which one is right. Peers - // that did switch can still be followed. - let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect(); - error!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own"); - return catchup::check(host).await; - } - }; - - if host - .compaction() - .status(&id) - .is_some_and(|status| status.error.is_some()) - { - // Our own switch failed (reported in the compaction's status): fetch the epoch - // from a peer that made it. - return catchup::check(host).await; - } - - let switch_delay = host.dormant_after() / 4; // one keepalive interval - if host.compaction().first_seen(&id).elapsed() < switch_delay { - return; - } - - match switch(host, &id, &record).await { - Ok(stats) => { - host.compaction().set_status( - &id, - LocalStatus { - switched_at: Some(unix_millis()), - replayed_paths: Some(stats.replayed), - failed_paths: Some(stats.failed), - quarantine: None, - error: None, - }, - ); - } - Err(reason) => { - error!(compaction = %id, reason = %reason, "Epoch switch failed"); - host.compaction().set_status( - &id, - LocalStatus { - error: Some(reason), - ..LocalStatus::default() - }, - ); - } - } -} - -async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> { - let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id)) - else { - return Err("unreadable compaction record".to_owned()); - }; - let previous = request.previous; - let base = request.base_heads.clone(); - let retention = record.proposal.policy.archive_retention.clone(); - // Only the coordinator carries the record into the new epoch: one writer, so the - // nodes don't all write the same path concurrently right after switching (a merge - // conflict, plus one change per node to sync). The epoch metadata in the snapshot - // keeps the essentials on every node anyway. - let carry_record = record.proposal.coordinator == host.node_id(); - let record_path = records::record_path(id, ""); - - info!(compaction = %id, from = %previous, to = %expected, "Switching epoch"); - - // 1. Consistent copy, under the read lock only. - let copy = host.read_doc(|doc| doc.clone()).await; - let prebuilt = host - .compaction() - .take_prebuilt(id) - .filter(|snapshot| snapshot.epoch == expected); - - // 2. Build (or reuse), verify and replay without holding any lock. - let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking( - move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> { - let mut copy = copy; - let snapshot = match prebuilt { - Some(snapshot) => snapshot, - None => snapshot::build(&mut copy, &request.params()) - .map_err(|error| format!("snapshot build failed: {error}"))?, - }; - if snapshot.epoch != expected { - return Err(format!( - "built snapshot {} does not match the committed epoch {expected}", - snapshot.epoch - )); - } - - // A fresh actor: this node's writes in the new epoch. - let mut doc = snapshot.doc.with_actor(ActorId::random()); - - // Keep this compaction's record visible in the new epoch, for its history. - if carry_record { - if let Err(error) = store::copy_path(©, &mut doc, &record_path) { - warn!(error = %error, "Could not carry the compaction record into the new epoch"); - } - } - - let heads = copy.get_heads(); - let stats = delta::replay(&mut copy, &base, &heads, &mut doc); - Ok((doc, heads, stats)) - }, - ) - .await - .map_err(|error| format!("switch task failed: {error}"))??; - - // 3. Swap, replaying what arrived since the copy. - let old = swap_in(host, expected, new_doc, copy_heads).await?; - - // 4. Keep the old document as configured. - host.compaction() - .archive(previous, old, retention, expected.generation); - - info!( - compaction = %id, - epoch = %expected, - replayed = stats.replayed, - failed = stats.failed, - "Switched to the new epoch" - ); - Ok(stats) -} - -/// Under the write lock: replays into `new_doc` whatever the live document received -/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it -/// in as `target`. Returns the old document. -pub(super) async fn swap_in<H: Host>( - host: &H, - target: Epoch, - new_doc: AutoCommit, - copy_heads: Vec<ChangeHash>, -) -> Result<AutoCommit, String> { - host.switch_epoch(target, move |live| { - // Only one worker switches, but be sure the copy still describes the live document. - if !copy_heads - .iter() - .all(|head| live.get_change_by_hash(head).is_some()) - { - return Err("the document was replaced while preparing the switch".to_owned()); - } - let mut new_doc = new_doc; - let live_heads = live.get_heads(); - let late = delta::replay(live, ©_heads, &live_heads, &mut new_doc); - if late.replayed + late.failed > 0 { - info!( - replayed = late.replayed, - failed = late.failed, - "Replayed late changes" - ); - } - Ok(new_doc) - }) - .await -} |