access: read once what rules read at the same heads

bench_screening shows judging at about 17 µs of a change's 25: some
two and a half reads at its heads (the policy, bans, its node's member
entry, whether there is a registry, the sealed declarations), each
costing as much as working out what the heads cover. Screening judges
most of a batch at the same heads, and those reads are the same for
every change judged there.

Keep them (rules::Reads), by node and by user where they depend on
one, while the heads stay the same; screening keeps one for a batch.
What the heads hold doesn't change as the document takes changes, so
nothing goes stale but by other heads. The benchmark breakdown shows
judging so and afresh.

3 files changed, 136 insertions, 29 deletions

diff --git a/src/access/rules.rs b/src/access/rules.rs
index ed9ee4b..3aba0ba 100644
--- a/src/access/rules.rs
+++ b/src/access/rules.rs
@@ -29,10 +29,12 @@
 //! any member may. A change written while the document had no policy (at its
 //! dependencies) is judged by none: the rules hold from the policy on.
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, HashMap};
 
-use crate::access::Right;
+use crate::access::policy::Policy;
 use crate::access::registry;
+use crate::access::sealed::Declared;
+use crate::access::{Caps, Right};
 use crate::document::at::At;
 use crate::document::inspect::{Action, Op, Path};
 use crate::document::meta;
@@ -76,8 +78,49 @@ impl std::fmt::Display for Refusal {
     }
 }
 
+/// What `judge` reads at the heads a change is judged at, whatever the change: the same
+/// for every change judged there, so read once (`judge_reading`). Each read at given
+/// heads costs as much as working out what they cover, and screening judges most of a
+/// batch at the same heads (`access::screen`). What the heads hold doesn't change as the
+/// document takes more changes, so nothing here goes stale but by other heads.
+#[derive(Debug, Default)]
+pub(crate) struct Reads {
+    seen: Vec<ChangeHash>,
+    policy: Option<Policy>,
+    registry: Option<bool>,
+    declared: Option<Vec<Declared>>,
+    /// By node: whether it is banned, and whether it is a member.
+    banned: HashMap<String, bool>,
+    member: HashMap<String, bool>,
+    /// By user: what the registry lets it do.
+    caps: HashMap<String, Caps>,
+}
+
+impl Reads {
+    /// Read at `seen`: what was read at other heads forgotten.
+    fn at(&mut self, seen: &[ChangeHash]) -> &mut Self {
+        if self.seen != seen {
+            *self = Self {
+                seen: seen.to_vec(),
+                ..Self::default()
+            };
+        }
+        self
+    }
+}
+
 /// Judges `change`, about to be applied to `doc` (of `epoch`).
 pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result<(), Refusal> {
+    judge_reading(doc, epoch, change, &mut Reads::default())
+}
+
+/// `judge`, with what was read at the same heads before (`Reads`).
+pub(crate) fn judge_reading(
+    doc: &Document,
+    epoch: Epoch,
+    change: &Change<'_>,
+    reads: &mut Reads,
+) -> Result<(), Refusal> {
     if change.deps.is_empty() {
         // The first epoch's: the first change a document holds, as it comes to an empty
         // one, or as it is in one (judged again, `swarm::refusals`).
@@ -95,9 +138,13 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
         };
     }
     let at = At::new(doc, change.seen).ok_or(Refusal::Unready)?;
+    let reads = reads.at(change.seen);
     // Written while the document had no policy (or one this node doesn't know): judged
     // as it was then, by none. History from before the policy stays as it is.
-    let policy = super::policy::of(&at);
+    let policy = reads
+        .policy
+        .get_or_insert_with(|| super::policy::of(&at))
+        .clone();
     if !policy.enforced() {
         return Ok(());
     }
@@ -105,11 +152,18 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
     let Some(node) = change.committer else {
         return refuse("it names no node".to_owned());
     };
-    if super::bans::banned(&at, node) {
+    let banned = *reads
+        .banned
+        .entry(node.to_owned())
+        .or_insert_with(|| super::bans::banned(&at, node));
+    if banned {
         return refuse(format!("{node} is banned"));
     }
     let own_entry = meta::path(&[meta::MEMBERS, node]);
-    let member = store::read_path(&at, &own_entry).ok().flatten().is_some();
+    let member = *reads
+        .member
+        .entry(node.to_owned())
+        .or_insert_with(|| store::read_path(&at, &own_entry).ok().flatten().is_some());
     let registering = change
         .paths
         .iter()
@@ -117,12 +171,25 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
     if !member && !registering {
         return refuse(format!("{node} isn't a member"));
     }
-    let registry = registry::exists(&at);
+    let registry = *reads.registry.get_or_insert_with(|| registry::exists(&at));
+    if registry && let Some(user) = change.user {
+        reads
+            .caps
+            .entry(user.to_owned())
+            .or_insert_with(|| registry::caps_of(&at, user));
+    }
+    reads
+        .declared
+        .get_or_insert_with(|| super::sealed::declared(&at));
+    let reads = &*reads;
     // Whether the user may write `path`: as the registry at its dependencies says, or,
     // without one, any member.
     let user_may = |path: &Path| match (registry, change.user) {
         (false, _) => true,
-        (true, Some(user)) => registry::caps_of(&at, user).allows(Right::Write, path),
+        (true, Some(user)) => reads
+            .caps
+            .get(user)
+            .is_some_and(|caps| caps.allows(Right::Write, path)),
         (true, None) => false,
     };
     // Maps made on the way to what the change writes, where there were none
@@ -252,9 +319,9 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
     if policy.strict() {
         signed_values(&at, change.ops).or_else(refuse)?;
     }
-    let declared = super::sealed::declared(&at);
+    let declared = reads.declared.as_deref().unwrap_or_default();
     if !declared.is_empty() {
-        sealed_values(&at, &declared, change.ops).or_else(refuse)?;
+        sealed_values(&at, declared, change.ops).or_else(refuse)?;
     }
     Ok(())
 }
@@ -732,6 +799,39 @@ mod tests {
         assert!(earlier.is_ok());
     }
 
+    /// What was read at some heads, kept (`Reads`): read again at others, a ban since
+    /// counts; read at the first again, it doesn't.
+    #[test]
+    fn reads_kept_by_heads() {
+        let mut doc = swarm(true);
+        let before = doc.get_heads();
+        write(&mut doc, "_continuum/banned/b", json!({"by": "alice"}));
+        let after = doc.get_heads();
+        let touched = paths(&["_continuum/members/b/alive"]);
+        let mut reads = Reads::default();
+        let mut at = |heads: &[ChangeHash]| {
+            judge_reading(
+                &doc,
+                Epoch::GENESIS,
+                &Change {
+                    hash: ChangeHash([9; 32]),
+                    deps: heads,
+                    seen: heads,
+                    committer: Some("b"),
+                    user: None,
+                    paths: &touched,
+                    ops: &[],
+                },
+                &mut reads,
+            )
+        };
+        assert!(at(&before).is_ok());
+        assert!(at(&before).is_ok());
+        assert!(refused(at(&after)));
+        assert!(refused(at(&after)));
+        assert!(at(&before).is_ok());
+    }
+
     /// A remake's decision: its coordinator's, or one who may write it; no other
     /// member's, whatever its time.
     #[test]
diff --git a/src/access/screen.rs b/src/access/screen.rs
index 690e032..1c6c8c7 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -384,6 +384,8 @@ pub(crate) fn apply(
     // What the accepted, not applied yet, touch: a change making a map at one of these
     // paths is judged with them in (`rules`' maps made where there are none).
     let mut unapplied: BTreeSet<Path> = BTreeSet::new();
+    // What the rules read at the heads judged at, kept while they stay the same.
+    let mut reads = rules::Reads::default();
     for (index, change) in ready.iter().enumerate() {
         let hash = change.hash();
         let committer = committers[index].as_deref();
@@ -464,7 +466,7 @@ pub(crate) fn apply(
         let seen_at: Vec<ChangeHash> = seen_at.into_iter().collect();
         let root = change.deps().is_empty();
         let judged = if judging || root {
-            rules::judge(
+            rules::judge_reading(
                 doc,
                 epoch,
                 &rules::Change {
@@ -476,6 +478,7 @@ pub(crate) fn apply(
                     paths: &paths[index],
                     ops: &ops[index],
                 },
+                &mut reads,
             )
         } else {
             Ok(())
diff --git a/src/document/bench.rs b/src/document/bench.rs
index 026cf7b..2b82036 100644
--- a/src/document/bench.rs
+++ b/src/document/bench.rs
@@ -684,9 +684,10 @@ fn screening_costs(workload: &str, doc: &Document) {
 }
 
 /// Where screening `changes` spends its time, per change judged: decoding them, their
-/// operations placed (`inspect::Inspector`), and judging them (`rules::judge`), of which
-/// reading the policy (`policy::of`) and, for scale, one read (the committer's member
-/// entry). As screening judges most: the first ones in (the maps they make, which flush
+/// operations placed (`inspect::Inspector`), and judging them, as screening does (what
+/// is read at the same heads read once, `rules::Reads`) and afresh (`rules::judge`); of
+/// that, reading the policy (`policy::of`) and, for scale, one read (the committer's
+/// member entry). As screening judges most: the first ones in (the maps they make, which flush
 /// the batch), the rest judged at the heads after them (`screen`'s frontier).
 fn screening_parts(changes: &[automerge::Change]) {
     use crate::access::{policy, rules};
@@ -736,26 +737,28 @@ fn screening_parts(changes: &[automerge::Change]) {
         .iter()
         .map(|ops| ops.iter().map(|op| op.path.clone()).collect())
         .collect();
-    let judge = median(3, || {
+    let judge_all = |reads: &mut Option<rules::Reads>| {
         let started = Instant::now();
         for (index, change) in rest.iter().enumerate() {
-            let judged = rules::judge(
-                &base,
-                Epoch::GENESIS,
-                &rules::Change {
-                    hash: change.hash(),
-                    deps: change.deps(),
-                    seen: &seen,
-                    committer: committers[index].as_deref(),
-                    user: None,
-                    paths: &paths[index],
-                    ops: &ops[index],
-                },
-            );
+            let change = rules::Change {
+                hash: change.hash(),
+                deps: change.deps(),
+                seen: &seen,
+                committer: committers[index].as_deref(),
+                user: None,
+                paths: &paths[index],
+                ops: &ops[index],
+            };
+            let judged = match reads {
+                Some(reads) => rules::judge_reading(&base, Epoch::GENESIS, &change, reads),
+                None => rules::judge(&base, Epoch::GENESIS, &change),
+            };
             assert_eq!(judged, Ok(()), "every change taken");
         }
         started.elapsed()
-    });
+    };
+    let judge = median(3, || judge_all(&mut Some(rules::Reads::default())));
+    let afresh = median(3, || judge_all(&mut None));
     let read_policy = median(3, || {
         let started = Instant::now();
         for _ in rest {
@@ -775,11 +778,12 @@ fn screening_parts(changes: &[automerge::Change]) {
     });
     let per = |d: Duration| d.as_secs_f64() * 1e6 / rest.len().max(1) as f64;
     println!(
-        "{:<40} per change: decode {:.2} µs, place {:.2} µs, judge {:.2} µs (policy::of {:.2} µs, one read {:.2} µs)",
+        "{:<40} per change: decode {:.2} µs, place {:.2} µs, judge {:.2} µs (afresh {:.2} µs: policy::of {:.2} µs, one read {:.2} µs)",
         "",
         per(decode),
         per(place),
         per(judge),
+        per(afresh),
         per(read_policy),
         per(one_read),
     );