diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/main.rs | 2 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 32 | ||||
| -rw-r--r-- | src/remake/delta.rs | 17 | ||||
| -rw-r--r-- | src/remake/local.rs | 17 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 3 | ||||
| -rw-r--r-- | src/remake/switch.rs | 5 |
6 files changed, 56 insertions, 20 deletions
diff --git a/src/main.rs b/src/main.rs index 88cbc6f..278a0f4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -146,6 +146,7 @@ async fn main() -> Result<()> { Some(backup) => (backup.doc, Some(backup.epoch), Some(backup.swarm_id)), None => (AutoCommit::new(), None, None), }; + let initial_actor = initial_doc.get_actor().clone(); let initial_epoch = match (recovered_epoch, args.debug_epoch) { (Some(epoch), _) => epoch, (None, Some(generation)) => Epoch { @@ -210,6 +211,7 @@ async fn main() -> Result<()> { }; trace!("Application state initialized"); + state.remake.add_own_actor(initial_actor); // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from // names its swarm itself (checked against the remembered one above). diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index cad5a22..4feb70e 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -31,7 +31,7 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; -use automerge::{AutoCommit, ChangeHash}; +use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::time::Instant; use tracing::{debug, error, info, warn}; @@ -177,8 +177,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { host.set_stale_view(Some((local, copy))); } let was_ready = host.suspend().await; + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); let old = match host - .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new())) + .switch_epoch(target, move |_live| { + Ok::<_, Infallible>(AutoCommit::new().with_actor(actor)) + }) .await { Ok(old) => old, @@ -276,7 +280,9 @@ async fn finish<H: Host>(host: &H) { // Everything heavy runs on a copy, without a lock. let copy = host.read_doc(|doc| doc.clone()).await; - let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await; + let own = host.remake().own_actors(); + let prepared = + tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own)).await; let (old, metadata, plan, own_missing, had_history) = match prepared { Ok(Prepared::Ready { @@ -395,7 +401,15 @@ async fn finish<H: Host>(host: &H) { } /// Verifies the synced document and decides about the old document's own changes. -fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared { +/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor +/// counts in any case. +fn prepare( + mut copy: AutoCommit, + mut old: AutoCommit, + previous: Epoch, + target: Epoch, + mut own: Vec<ActorId>, +) -> Prepared { if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() { return Prepared::NotYet(old); } @@ -421,7 +435,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E .ok() .filter(|next| next.previous == previous); - let own = old.get_actor().clone(); + own.push(old.get_actor().clone()); let clock = next .as_ref() .map_or_else(BTreeMap::new, |next| next.clock.clone()); @@ -573,7 +587,7 @@ mod tests { #[test] fn replays_within_the_window() { let (old, swarm, target) = scenario(&json!({})); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Replay(paths, _), own_missing, @@ -589,7 +603,7 @@ mod tests { #[test] fn quarantines_outside_the_window() { let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } })); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Quarantine { paths, .. }, .. @@ -602,7 +616,7 @@ mod tests { fn new_node_has_nothing_to_replay() { let (_, swarm, target) = scenario(&json!({})); assert!(matches!( - prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target), + prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target, Vec::new()), Prepared::Ready { plan: Plan::Nothing, had_history: false, @@ -615,7 +629,7 @@ mod tests { fn waits_for_the_target_epoch() { let (old, _, target) = scenario(&json!({})); assert!(matches!( - prepare(AutoCommit::new(), old, Epoch::GENESIS, target), + prepare(AutoCommit::new(), old, Epoch::GENESIS, target, Vec::new()), Prepared::NotYet(_) )); } diff --git a/src/remake/delta.rs b/src/remake/delta.rs index c8f6b7c..51cc409 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -159,22 +159,23 @@ pub(crate) struct Missing { /// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the /// snapshot iff seq <= clock[actor]. /// -/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes' -/// changes after the base were replayed by those nodes when they switched, and replaying -/// them again would overwrite newer writes in the new epoch with stale ones. +/// Only changes this node wrote itself (any of the actors in `own`: one per run and +/// epoch) are worth replaying when it rejoins: other nodes' changes after the base were +/// replayed by the nodes that switched, and replaying them again would overwrite newer +/// writes in the new epoch with stale ones. pub(crate) fn missing_from_snapshot( doc: &mut AutoCommit, clock: &BTreeMap<String, u64>, - own: &ActorId, + own: &[ActorId], ) -> Missing { - let own_hex = own.to_string(); + let own_hex: HashSet<String> = own.iter().map(ActorId::to_string).collect(); let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc .get_changes(&[]) .iter() .map(|change| { let actor = change.actor_id().to_string(); let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq); - let own_missing = actor == own_hex && !in_snapshot; + let own_missing = own_hex.contains(&actor) && !in_snapshot; (change.hash(), change.deps().to_vec(), own_missing) }) .collect(); @@ -370,7 +371,7 @@ mod tests { doc.commit(); let after = doc.get_heads(); - let missing = missing_from_snapshot(&mut doc, &clock, &own); + let missing = missing_from_snapshot(&mut doc, &clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); assert_eq!(missing.before, with_foreign); assert_ne!(missing.before, base); @@ -381,7 +382,7 @@ mod tests { // Nothing of ours missing: nothing to replay. let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect(); - let none = missing_from_snapshot(&mut doc, &full, &own); + let none = missing_from_snapshot(&mut doc, &full, std::slice::from_ref(&own)); assert_eq!(none.own, 0); let mut heads = after.clone(); heads.sort(); diff --git a/src/remake/local.rs b/src/remake/local.rs index 04ac0b0..09c99bd 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; -use automerge::AutoCommit; +use automerge::{ActorId, AutoCommit}; use serde::Serialize; use tokio::time::Instant; @@ -55,6 +55,9 @@ struct Inner { content_fetches: HashSet<[u8; 32]>, /// Remakes this node coordinates now (`coordinator`). coordinating: HashSet<String>, + /// Every actor this node writes or wrote with: one per run and epoch, and those of + /// earlier runs it knows of (`--document-file`). Its own writes, for a rejoin. + own_actors: Vec<ActorId>, } /// Content no remake refers to any more is kept this long after it was stored, for @@ -147,6 +150,18 @@ impl RemakeState { self.started_at } + /// This node writes with `actor` from now on (or did, in an earlier run). + pub(crate) fn add_own_actor(&self, actor: ActorId) { + let mut inner = self.lock(); + if !inner.own_actors.contains(&actor) { + inner.own_actors.push(actor); + } + } + + pub(crate) fn own_actors(&self) -> Vec<ActorId> { + self.lock().own_actors.clone() + } + /// Whether this node starts coordinating `id` now (false if it already does). pub(crate) fn start_coordinating(&self, id: &str) -> bool { self.lock().coordinating.insert(id.to_owned()) diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index 3c180ba..20569e0 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -750,7 +750,8 @@ mod tests { assert!(verify_document(&mut fetched, wrong).is_err()); let own = late.get_actor().clone(); - let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own); + let missing = + delta::missing_from_snapshot(&mut late, &metadata.clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); let heads = late.get_heads(); let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..])); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index cf5ea75..e78643a 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -205,6 +205,9 @@ async fn switch<H: Host>( .take_prebuilt(id) .filter(|snapshot| snapshot.epoch == expected); + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); + // 2. Build (or reuse), verify and replay without holding any lock. let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking( move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> { @@ -222,7 +225,7 @@ async fn switch<H: Host>( } // A fresh actor: this node's writes in the new epoch. - let mut doc = snapshot.doc.with_actor(ActorId::random()); + let mut doc = snapshot.doc.with_actor(actor); // Keep this remake's record visible in the new epoch, for its history. if carry_record { |