diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/access/screen.rs | 72 | ||||
| -rw-r--r-- | src/state/host.rs | 6 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 10 |
3 files changed, 85 insertions, 3 deletions
diff --git a/src/access/screen.rs b/src/access/screen.rs index 3ea03e0..010ef49 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -684,6 +684,9 @@ pub(crate) struct Outcome { pub(crate) roots: Vec<ChangeHash>, /// Changes the bytes carried forged under another node's actor: their sender's. pub(crate) forged: usize, + /// The fresh actor the node writes as from now on, where the bytes carried a change + /// of its own actor that it doesn't hold: for its own actors (`add_own_actor`). + pub(crate) fresh_actor: Option<crate::document::ActorId>, } /// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with @@ -710,6 +713,27 @@ pub(crate) fn apply_bytes( } let mut changes = crate::document::changes::changes_in(bytes) .ok_or_else(|| anyhow::anyhow!("Refused changes that don't load"))?; + // A change of the actor this node writes as, and not one it made: another node writes + // as it too (two started from one document file with `--keep-actor`, a copied data + // directory). Each would make changes the other's peers refuse as forged; the node + // writes on as a fresh actor at once. The change is judged as any other. + let fresh_actor = changes + .iter() + .find(|change| { + change.actor_id() == doc.get_actor() + && !crate::document::changes::holds(&*doc, &change.hash()) + }) + .map(|change| { + let actor = crate::document::ActorId::random(); + tracing::warn!( + actor = %doc.get_actor(), + seq = change.seq(), + fresh = %actor, + "A change of this node's actor that it didn't make: another node writes as it (its files copied?); writing as a fresh actor" + ); + doc.set_actor(actor.clone()); + actor + }); let mut screening = screening.lock_or_recover(); let screening = screening.of(epoch_seq); // Into an empty document of the first epoch, a root only as pinned: one given or @@ -752,7 +776,10 @@ pub(crate) fn apply_bytes( } screening.record(refused_roots); if changes.is_empty() && screening.waiting.len() == 0 { - return Ok(Outcome::default()); + return Ok(Outcome { + fresh_actor, + ..Outcome::default() + }); } let received = changes.len(); let policed = screening.policed(doc); @@ -770,6 +797,7 @@ pub(crate) fn apply_bytes( refused_again: received > 0 && screened.refused_again == received, roots: screened.roots.iter().copied().collect(), forged: screened.forged, + fresh_actor, }; screening.judge_members(&screened); screening.record(screened.refused); @@ -1100,6 +1128,48 @@ mod tests { assert!(building.refused[0].why.contains("builds on refused change")); } + /// A change of the node's own actor that it didn't make: it writes on as a fresh + /// actor. Its own changes coming back change nothing. + #[test] + fn another_writing_as_this_node() { + let mut theirs = genesis(); + let mut ours = theirs.fork().with_actor(theirs.get_actor().clone()); + let screening = std::sync::Mutex::new(Screening::default()); + let take = |doc: &mut Document, from: &Document, since: &[ChangeHash]| { + let bytes: Vec<u8> = received(from, since) + .iter() + .flat_map(|change| change.raw_bytes().to_vec()) + .collect(); + apply_bytes( + doc, + Epoch::GENESIS, + 0, + &bytes, + &screening, + None, + &mut crate::document::PatchLog::inactive(), + ) + .unwrap() + }; + let heads = ours.get_heads(); + write(&mut ours, "a", None, "_continuum/members/a/alive", json!(1)); + let echo = ours.clone(); + assert!(take(&mut ours, &echo, &heads).fresh_actor.is_none()); + + let actor = ours.get_actor().clone(); + let before = theirs.get_heads(); + write( + &mut theirs, + "a", + None, + "_continuum/members/a/alive", + json!(2), + ); + let outcome = take(&mut ours, &theirs, &before); + assert_eq!(outcome.fresh_actor.as_ref(), Some(ours.get_actor())); + assert_ne!(ours.get_actor(), &actor); + } + /// Judged as its writer saw the registry: a grant earlier in the same batch counts. #[test] fn a_grant_in_the_same_batch_counts() { diff --git a/src/state/host.rs b/src/state/host.rs index 8522657..5b52e1d 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -590,7 +590,11 @@ impl Host for AppState { Some(&mut replica.signatures), &mut crate::document::PatchLog::inactive(), ) - .map(|_| ()) + .map(|outcome| { + if let Some(actor) = outcome.fresh_actor { + state.remake.add_own_actor(actor); + } + }) .map_err(|error| format!("{error:#}")) }; replica.refresh_heads(); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 3a3d92c..5550c27 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -597,6 +597,7 @@ pub(crate) async fn handle_sync( let peer = peer.to_owned(); let clock = Arc::clone(&state.clock); let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); + let remake = Arc::clone(&state.remake); // Some((caught_up, heads changed, carried changes, the new changes' authors with // theirs, and whether any has none, how long to wait before answering if it @@ -681,6 +682,9 @@ pub(crate) async fn handle_sync( if outcome.forged > 0 { tracing::warn!(peer = %peer, forged = outcome.forged, "A peer brought changes forged under another node's actor: refused"); } + if let Some(actor) = outcome.fresh_actor { + remake.add_own_actor(actor); + } } sync.receive_logged(doc, message, log) })(); @@ -965,6 +969,7 @@ async fn download_and_apply_delta( let clock = Arc::clone(&state.clock); let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); let screening = Arc::clone(&state.screening); + let remake = Arc::clone(&state.remake); tokio::task::spawn_blocking(move || -> Result<bool> { let mut replica = doc.blocking_write(); if replica.epoch != epoch { @@ -976,7 +981,7 @@ async fn download_and_apply_delta( // not. let epoch_seq = replica.epoch_seq; let replica = &mut *replica; - crate::access::screen::apply_bytes( + let outcome = crate::access::screen::apply_bytes( &mut replica.doc, epoch, epoch_seq, @@ -985,6 +990,9 @@ async fn download_and_apply_delta( Some(&mut replica.signatures), &mut crate::document::PatchLog::inactive(), )?; + if let Some(actor) = outcome.fresh_actor { + remake.add_own_actor(actor); + } } replica.refresh_heads(); // Signatures waiting for what it brought (`Signatures::offer`). |