screen: what a node receives judged, policy or not

Review 2, N7: a node screened received changes only while its own
document had a policy it enforces. A joiner's empty document has none,
so its first sync, as much history as the node it joins through chose,
was applied unjudged, and under signatures required unvouched; the same
for any node while the policy's own change arrived with later ones.

Sync messages, deltas and downloaded changes are now always screened,
each change judged by the policy at its dependencies (1135), so the
verdicts are the swarm's. While neither the document nor the changes
hold a policy (none touches a "policy" key: inspect::touch_key), only
an epoch's root is judged and nothing is placed: bench_screening, after
1188, measured placing everything at about as much again as applying
(564 ms + 543 ms for 100 000 changes), judging by the rules at about
five times (+2.5 s), which a document with a policy pays anyway.

access.md no longer lists the joiner's first sync, nor nodes agreeing
only as the policy reaches them (judged at the dependencies, they
agree), as left to do; changes::load_delta is for tests now.

5 files changed, 53 insertions, 29 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index 162528b..758c83b 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -288,7 +288,15 @@ pub(crate) fn apply(
 
     let decoded: Vec<crate::document::ExpandedChange> =
         ready.iter().map(|change| change.decode()).collect();
-    let ops: Vec<Vec<Op>> = Inspector::new(doc).ops(&decoded);
+    // Without a policy here, nor one among these: none judged but an epoch's root, and
+    // nothing placed. A change writing the policy writes at its key.
+    let judging = super::policy::of(&*doc).enforced()
+        || crate::document::inspect::touch_key(&decoded, "policy");
+    let ops: Vec<Vec<Op>> = if judging {
+        Inspector::new(doc).ops(&decoded)
+    } else {
+        vec![Vec::new(); decoded.len()]
+    };
     let paths: Vec<BTreeSet<Path>> = ops
         .iter()
         .map(|ops| ops.iter().map(|op| op.path.clone()).collect())
@@ -431,26 +439,32 @@ pub(crate) fn apply(
             }
         }
         let seen_at: Vec<ChangeHash> = seen_at.into_iter().collect();
-        let judged = rules::judge(
-            doc,
-            epoch,
-            &rules::Change {
-                hash,
-                deps: change.deps(),
-                seen: &seen_at,
-                committer,
-                user: user.as_deref(),
-                paths: &paths[index],
-                ops: &ops[index],
-            },
-        );
+        let root = change.deps().is_empty();
+        let judged = if judging || root {
+            rules::judge(
+                doc,
+                epoch,
+                &rules::Change {
+                    hash,
+                    deps: change.deps(),
+                    seen: &seen_at,
+                    committer,
+                    user: user.as_deref(),
+                    paths: &paths[index],
+                    ops: &ops[index],
+                },
+            )
+        } else {
+            Ok(())
+        };
         match judged {
             Ok(()) => {
                 accepted.push(change.clone());
                 if let Some(committer) = committer {
                     screened.taken_from.insert(committer.to_owned());
                 }
-                if relevant(&paths[index]) || reaches_signed(doc, &paths[index]) {
+                // A root too: whether another may be one is read in the document.
+                if root || relevant(&paths[index]) || reaches_signed(doc, &paths[index]) {
                     // What rules read changed: in before the next is judged.
                     screened.applied += accepted.len();
                     apply_taken(doc, std::mem::take(&mut accepted), log)?;
diff --git a/src/document/changes.rs b/src/document/changes.rs
index eef7dbc..44d30bd 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -156,7 +156,9 @@ pub(crate) fn delta_bytes(
     Some(bytes)
 }
 
-/// Loads a delta (`delta_bytes`) into `doc`.
+/// Loads a delta (`delta_bytes`) into `doc`, unjudged: for tests; a node screens what it
+/// downloads (`access::screen::apply_bytes`).
+#[cfg(test)]
 pub(crate) fn load_delta(doc: &mut Document, bytes: &[u8]) -> Result<()> {
     if let Some(misplaced) = misplaced_author(bytes) {
         bail!("Refused the downloaded changes: {misplaced}");
diff --git a/src/document/inspect.rs b/src/document/inspect.rs
index c42de82..f0356de 100644
--- a/src/document/inspect.rs
+++ b/src/document/inspect.rs
@@ -86,6 +86,17 @@ impl Place {
     }
 }
 
+/// Whether any of `changes` does anything at a map key `key`: a cheap look, without
+/// placing anything, for whether they may write the policy (`access::screen`).
+pub(crate) fn touch_key(changes: &[ExpandedChange], key: &str) -> bool {
+    changes.iter().any(|change| {
+        change
+            .operations
+            .iter()
+            .any(|op| matches!(&op.key, Key::Map(name) if name.as_str() == key))
+    })
+}
+
 /// Where a write at `path` in `doc` is, as rules see it (`Inspector`): placed by the
 /// same rule (`Place::child`), so cut after the first list or text on the way, which
 /// what is inside one touches. Where `doc` holds nothing (or a scalar) on the way, the
diff --git a/src/state/host.rs b/src/state/host.rs
index e7fc61a..01cbaa2 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -571,8 +571,9 @@ impl Host for AppState {
                 let result = {
                     let mut replica = state.doc.blocking_write();
                     let policy = crate::access::policy::of(&replica.doc);
-                    let result = if policy.enforced() {
-                        // Judged first (`access::screen`).
+                    // Judged first (`access::screen`), whether this document has a policy
+                    // yet or not.
+                    let result = {
                         let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
                         let replica = &mut *replica;
                         crate::access::screen::apply_bytes(
@@ -586,10 +587,6 @@ impl Host for AppState {
                         )
                         .map(|_| ())
                         .map_err(|error| format!("{error:#}"))
-                    } else {
-                        replica
-                            .apply_changes([change])
-                            .map_err(|error| error.to_string())
                     };
                     replica.refresh_heads();
                     result
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index ffdbf07..75c20f1 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -641,10 +641,11 @@ pub(crate) async fn handle_sync(
             replica
                 .logged_signed(|doc, known, log| {
                     let result = (|| {
-                        // Judged first, as the policy has it (`access::screen`); with
-                        // signatures required, by those it sent too.
+                        // Judged first, each as the policy it was written under has it
+                        // (`access::screen`), whether this document has one yet or not;
+                        // with signatures required, by those it sent too.
                         let policy = crate::access::policy::of(&*doc);
-                        if policy.enforced() {
+                        {
                             let bytes = message.take_changes();
                             let required = policy.required();
                             if required {
@@ -955,8 +956,9 @@ async fn download_and_apply_delta(
         }
         let heads_before = replica.heads().to_vec();
         let policy = crate::access::policy::of(&replica.doc);
-        if policy.enforced() {
-            // Judged first (`access::screen`).
+        {
+            // Judged first (`access::screen`), whether this document has a policy yet or
+            // not.
             let epoch_seq = replica.epoch_seq;
             let replica = &mut *replica;
             crate::access::screen::apply_bytes(
@@ -968,8 +970,6 @@ async fn download_and_apply_delta(
                 policy.required().then_some(&mut replica.signatures),
                 &mut crate::document::PatchLog::inactive(),
             )?;
-        } else {
-            crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
         }
         replica.refresh_heads();
         // Signatures waiting for what it brought (`Signatures::offer`).