1 files changed, 75 insertions, 0 deletions
diff --git a/src/access/screen.rs b/src/access/screen.rs
index dd9ffcd..66b7fc8 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -36,6 +36,52 @@ pub(crate) struct Refused {
     pub(crate) why: String,
 }
 
+/// What screening keeps between batches, for one document (`AppState::screening`): the
+/// changes waiting for their dependencies, and those refused, for operators
+/// (`GET /admin/refused`).
+#[derive(Debug, Default)]
+pub(crate) struct Screening {
+    /// The document's (`Replica::epoch_seq`): another's start empty (`of`).
+    epoch_seq: u32,
+    pub(crate) waiting: Waiting,
+    refused: std::collections::VecDeque<Refused>,
+}
+
+impl Screening {
+    /// Itself for the document `epoch_seq`: emptied if it was another's.
+    pub(crate) fn of(&mut self, epoch_seq: u32) -> &mut Self {
+        if self.epoch_seq != epoch_seq {
+            *self = Screening {
+                epoch_seq,
+                ..Screening::default()
+            };
+        }
+        self
+    }
+
+    /// Keeps `refused`, logged, the oldest going past `MAX_REFUSED`.
+    pub(crate) fn record(&mut self, refused: Vec<Refused>) {
+        for refused in refused {
+            tracing::warn!(
+                change = %refused.change.hash(),
+                committer = refused.committer.as_deref().unwrap_or("none"),
+                user = refused.user.as_deref().unwrap_or("none"),
+                why = %refused.why,
+                "Refused a change (_continuum/access/policy)"
+            );
+            self.refused.push_back(refused);
+            if self.refused.len() > MAX_REFUSED {
+                self.refused.pop_front();
+            }
+        }
+    }
+
+    /// Those refused, oldest first.
+    pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
+        self.refused.iter()
+    }
+}
+
 /// Received changes waiting for their dependencies.
 #[derive(Debug, Default)]
 pub(crate) struct Waiting(Vec<Change>);
@@ -179,6 +225,35 @@ pub(crate) fn apply(
     Ok(screened)
 }
 
+/// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
+/// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load
+/// as changes, or with a misplaced author (`changes::misplaced_author`), are refused
+/// whole.
+pub(crate) fn apply_bytes(
+    doc: &mut Document,
+    epoch: Epoch,
+    epoch_seq: u32,
+    bytes: &[u8],
+    screening: &std::sync::Mutex<Screening>,
+    log: &mut crate::document::PatchLog,
+) -> Result<usize> {
+    use crate::locks::MutexExt;
+    if bytes.is_empty() {
+        return Ok(0);
+    }
+    if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
+        anyhow::bail!("Refused changes: {misplaced}");
+    }
+    let changes = crate::document::changes::changes_in(bytes)
+        .ok_or_else(|| anyhow::anyhow!("Refused changes that don't load"))?;
+    let mut screening = screening.lock_or_recover();
+    let screening = screening.of(epoch_seq);
+    let screened = apply(doc, epoch, changes, &mut screening.waiting, log)?;
+    let applied = screened.applied;
+    screening.record(screened.refused);
+    Ok(applied)
+}
+
 /// Whether a change touching `paths` changes what rules read: the members, the user
 /// registry, remake records, the protocol data as a whole.
 fn relevant(paths: &BTreeSet<Path>) -> bool {