screen: write as a fresh actor when another node writes as this one

With --keep-actor, a node on a document file that ends with a clean stop
writes on as the actor the file names. Two nodes started on copies of
one file (a copied data directory, a restored disk image while the
original runs) both do, and both make that actor's next change number:
peers holding one refuse the other as forged, and everything built on
it.

screen::apply_bytes, the way every received change comes in, now looks
for a change of the actor the node writes as that the node doesn't hold:
only another node can have made it. The node then writes on as a fresh
actor at once, logs a warning, and its callers add the actor to the
node's own (RemakeState::add_own_actor), which the document file's
header follows. The change is judged as any other. lifecycle.md says not
to start a copy of a node's data directory with --keep-actor, and what
the switch leaves: what both wrote before stays refused where the
other's came first.

1 files changed, 9 insertions, 1 deletions

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`).