screen: judged after the policy is gone, too

Review 3, R4: whether a batch was judged followed the receiving
document's policy now (and whether the batch writes one). After the
policy was removed, or replaced by a version the node doesn't know,
changes written under it and received late went in unjudged, while
nodes that had them earlier judged them; REFUSALS gave up likewise.

Screening now remembers that the document held a policy, known or not
(Screening::policed), and judges what it receives while it does, each
change by the policy at its dependencies; REFUSALS judges then too
(not under a policy the node doesn't know, as before). Not across a
restart, which access.md says.

3 files changed, 64 insertions, 6 deletions

diff --git a/src/access/screen.rs b/src/access/screen.rs
index a4ecdc2..6f7866f 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -78,6 +78,9 @@ pub(crate) struct Screening {
     /// Which change an empty document of the first epoch takes as its root
     /// (`swarm::roots`); kept from one document to the next.
     pub(crate) root_pin: RootPin,
+    /// Whether the document held a policy, known or not, since this node took it up
+    /// (`policed`).
+    policed: bool,
 }
 
 /// Which change an empty document of the first epoch takes as its root: any (nothing
@@ -141,6 +144,16 @@ impl Screening {
     }
 
     /// Itself for the document `epoch_seq`: emptied if it was another's.
+    /// Whether `doc` holds a policy, known or not, or did since this node took it up:
+    /// then what it receives is judged (`apply`), each change by the policy at its
+    /// dependencies, though the policy is gone or unknown now; changes written under it
+    /// may still come. Not across a restart: a node that starts with no policy in its
+    /// document judges as one that never had one.
+    pub(crate) fn policed(&mut self, doc: &Document) -> bool {
+        self.policed |= !matches!(super::policy::of(doc), super::policy::Policy::Open);
+        self.policed
+    }
+
     pub(crate) fn of(&mut self, epoch_seq: u32) -> &mut Self {
         if self.epoch_seq != epoch_seq {
             *self = Screening {
@@ -275,10 +288,13 @@ pub(crate) struct Screened {
 /// refused again by their hash, and those building on them, and those refused now go
 /// into it. With `signatures` (the node's always; tests judge without), those written
 /// under a policy requiring signatures that no signature vouches for wait too, and
-/// those forged under another node's actor are refused.
+/// those forged under another node's actor are refused. With `policed` (the document
+/// held a policy, `Screening::policed`), judged though it has none now.
+#[allow(clippy::too_many_arguments)]
 pub(crate) fn apply(
     doc: &mut Document,
     epoch: Epoch,
+    policed: bool,
     received: Vec<Change>,
     waiting: &mut Waiting,
     refusals: &mut Refusals,
@@ -315,9 +331,11 @@ pub(crate) fn apply(
 
     let decoded: Vec<crate::document::ExpandedChange> =
         ready.iter().map(|change| change.decode()).collect();
-    // 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()
+    // Without a policy here, ever (`policed`: changes written under one removed since
+    // still come), 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 = policed
+        || super::policy::of(&*doc).enforced()
         || crate::document::inspect::touch_key(&decoded, "policy");
     let ops: Vec<Vec<Op>> = if judging {
         Inspector::new(doc).ops(&decoded)
@@ -735,9 +753,11 @@ pub(crate) fn apply_bytes(
         return Ok(Outcome::default());
     }
     let received = changes.len();
+    let policed = screening.policed(doc);
     let screened = apply(
         doc,
         epoch,
+        policed,
         changes,
         &mut screening.waiting,
         &mut screening.refusals,
@@ -896,6 +916,7 @@ mod tests {
         apply(
             doc,
             Epoch::GENESIS,
+            false,
             changes,
             waiting,
             &mut Refusals::default(),
@@ -1051,6 +1072,7 @@ mod tests {
             apply(
                 doc,
                 Epoch::GENESIS,
+                false,
                 changes,
                 &mut waiting,
                 &mut refusals,
@@ -1151,6 +1173,7 @@ mod tests {
             apply(
                 ours,
                 Epoch::GENESIS,
+                false,
                 changes,
                 waiting,
                 &mut refusals,
@@ -1229,6 +1252,7 @@ mod tests {
                 apply(
                     doc,
                     Epoch::GENESIS,
+                    false,
                     changes,
                     waiting,
                     &mut Refusals::default(),
@@ -1301,6 +1325,7 @@ mod tests {
         let screened = apply(
             &mut ours,
             Epoch::GENESIS,
+            false,
             received(&theirs, &[]),
             &mut Waiting::default(),
             &mut Refusals::default(),
@@ -1311,4 +1336,26 @@ mod tests {
         assert!(screened.refused.is_empty());
         assert_eq!(ours.get_heads(), theirs.get_heads());
     }
+
+    /// A document that held a policy is judged so after it is gone: changes written
+    /// under it may still come.
+    #[test]
+    fn policed_after_the_policy_is_gone() {
+        assert!(!Screening::default().policed(&Document::new()));
+        let mut doc = genesis();
+        let mut screening = Screening::default();
+        assert!(screening.policed(&doc));
+        transact(
+            &mut doc,
+            UnixMillis(1),
+            &Writer {
+                node: "a",
+                user: None,
+            },
+            |doc| store::delete_path(doc, &path("_continuum/access/policy"), false),
+        )
+        .unwrap();
+        assert!(!crate::access::policy::of(&doc).enforced());
+        assert!(screening.policed(&doc));
+    }
 }
diff --git a/src/document/bench.rs b/src/document/bench.rs
index 887ced0..39a8dc7 100644
--- a/src/document/bench.rs
+++ b/src/document/bench.rs
@@ -664,6 +664,7 @@ fn screening_costs(workload: &str, doc: &Document) {
         let screened = crate::access::screen::apply(
             &mut empty,
             Epoch::GENESIS,
+            false,
             changes,
             &mut crate::access::screen::Waiting::default(),
             &mut crate::access::screen::Refusals::default(),
diff --git a/src/swarm/refusals.rs b/src/swarm/refusals.rs
index af4252c..30c3971 100644
--- a/src/swarm/refusals.rs
+++ b/src/swarm/refusals.rs
@@ -53,15 +53,25 @@ pub(crate) async fn tell(
 /// rules have it now; if they refuse one, we took it unjudged, and drop it.
 pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMessage) {
     let doc = std::sync::Arc::clone(&state.doc);
+    let screening = std::sync::Arc::clone(&state.screening);
     let changes: Vec<ChangeHash> = message.changes.into_iter().take(MAX_LISTED).collect();
     let judged = tokio::task::spawn_blocking(move || {
+        use crate::locks::MutexExt;
         let replica = doc.blocking_read();
         if replica.epoch != message.epoch {
             return None;
         }
+        // Without a policy, ever (`Screening::policed`), nothing to judge by; with one
+        // this node doesn't know now, it can't judge as they do: receiving without
+        // judging until upgraded. A policy gone since still judges what was written under
+        // it, each change at its dependencies.
         let policy = crate::access::policy::of(&replica.doc);
-        if !policy.enforced() || matches!(policy, crate::access::policy::Policy::Unknown(_)) {
-            // We can't judge as they do: receiving without judging until upgraded.
+        if matches!(policy, crate::access::policy::Policy::Unknown(_))
+            || !screening
+                .lock_or_recover()
+                .of(replica.epoch_seq)
+                .policed(&replica.doc)
+        {
             return None;
         }
         let mut refused = None;