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.

6 files changed, 59 insertions, 34 deletions

diff --git a/docs/access.md b/docs/access.md
index b853b5a..c5365b7 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -312,15 +312,16 @@ saying the same; and never makes a node drop what its own rules take. If its rul
 take the change, the two policies disagree, which it logs. A change whose
 dependencies haven't come yet waits for them.
 
+A node judges what it receives whether its document has a policy yet or not, each
+change by the policy it was written under: a joining node judges its first sync, the
+policy's own change and everything after it, as every other node does. While neither
+its document nor what it receives holds a policy, it judges only that a root is the
+epoch's, and nothing costs more than applying the changes.
+
 What it doesn't do yet:
 
 * **Prove the user.** A change is its node's (with signatures required, proven) and its
   claimed user's: a member could claim a user its operator doesn't have.
-* **Judge a joining node's first sync.** An empty document has no policy: what it gets
-  from the node it joins through is taken as it is, as that node judged it.
-* **Agree at once.** The policy applies as each node gets it: a change received between
-  the policy's arrival on one node and on another may be refused by one and kept by
-  the other.
 
 A node that doesn't know the policy (another version, or a setting it lacks) takes no
 writes through the API (`409`) and receives without judging, until it is upgraded.
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`).