1 files changed, 85 insertions, 1 deletions

diff --git a/src/state/host.rs b/src/state/host.rs
index fb2e2e9..d23e07d 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 use anyhow::Result;
 use tokio::sync::broadcast;
 use tokio::time::Instant;
-use tracing::debug;
+use tracing::{debug, error, info};
 
 use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing};
 use crate::document::replica::{Epoch, Replica};
@@ -90,6 +90,90 @@ impl AppState {
         (doc, true)
     }
 
+    /// Drops `root`, a change the swarm's rules refuse that this node took unjudged
+    /// (`swarm::refusals`), and every change built on it: the document becomes the one
+    /// without them (forked where none is), written by a fresh actor of this node's, as
+    /// the old one's later changes are among those dropped; this node's own writes among
+    /// them go to the quarantine. Every peer syncs anew. Nothing while a rejoin runs, or
+    /// if the document doesn't hold `root`.
+    pub(crate) async fn drop_taken(&self, root: ChangeHash) {
+        let own = self.remake.own_actors();
+        let state = self.clone();
+        let dropped = tokio::task::spawn_blocking(move || {
+            let mut replica = state.doc.blocking_write();
+            if !replica.writable {
+                return None;
+            }
+            let changes = replica.doc.get_changes(&[]);
+            if !changes.iter().any(|change| change.hash() == root) {
+                return None;
+            }
+            // In causal order: a change after those it builds on.
+            let mut tainted: std::collections::HashSet<ChangeHash> = [root].into();
+            for change in &changes {
+                if change.deps().iter().any(|dep| tainted.contains(dep)) {
+                    tainted.insert(change.hash());
+                }
+            }
+            let kept: Vec<_> = changes
+                .iter()
+                .filter(|change| !tainted.contains(&change.hash()))
+                .collect();
+            let built_on: std::collections::HashSet<ChangeHash> = kept
+                .iter()
+                .flat_map(|change| change.deps().iter().copied())
+                .collect();
+            let heads: Vec<ChangeHash> = kept
+                .iter()
+                .map(|change| change.hash())
+                .filter(|hash| !built_on.contains(hash))
+                .collect();
+            // This node's own writes among those dropped: the paths they touched.
+            let mut actors = own;
+            actors.push(replica.doc.get_actor().clone());
+            let ours: Vec<crate::document::ExpandedChange> = changes
+                .iter()
+                .filter(|change| tainted.contains(&change.hash()))
+                .filter(|change| actors.contains(change.actor_id()))
+                .map(|change| change.decode())
+                .collect();
+            let mut paths: Vec<Vec<String>> =
+                crate::document::inspect::Inspector::new(&replica.doc)
+                    .ops(&ours)
+                    .into_iter()
+                    .flatten()
+                    .map(|op| op.path)
+                    .collect();
+            paths.sort();
+            paths.dedup();
+            let actor = crate::document::ActorId::random();
+            let fresh = match replica.doc.fork_at(&heads) {
+                Ok(fresh) => fresh.with_actor(actor.clone()),
+                Err(error) => {
+                    error!(error = %error, "Couldn't drop a change taken unjudged");
+                    return None;
+                }
+            };
+            state.remake.add_own_actor(actor);
+            let epoch = replica.epoch;
+            let old = state.swap_document(&mut replica, epoch, fresh);
+            Some((epoch, old, paths, tainted.len()))
+        })
+        .await
+        .ok()
+        .flatten();
+        let Some((epoch, old, paths, count)) = dropped else {
+            return;
+        };
+        info!(
+            changes = count,
+            own_paths = paths.len(),
+            "Dropped a change taken unjudged and what built on it"
+        );
+        crate::remake::quarantine_taken(self, epoch, &old, paths);
+        self.after_epoch_switch().await;
+    }
+
     /// After the document was replaced: every peer's sync state refers to the old
     /// document, so start over with all of them. Peers still on the old epoch are
     /// skipped by the sync worker until they switch too.