access: signatures required

"signatures": "required": a change counts only once a signature by its
committer vouches for it, and waits until then; every node then signs
every change it makes.

Auth plan M6.

2 files changed, 58 insertions, 8 deletions

diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs
index c172455..309ccc2 100644
--- a/src/swarm/signatures.rs
+++ b/src/swarm/signatures.rs
@@ -68,12 +68,47 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
         return;
     };
     let doc = std::sync::Arc::clone(&state.doc);
-    let _ = tokio::task::spawn_blocking(move || {
+    let screening = std::sync::Arc::clone(&state.screening);
+    // How many waiting changes went in, if any: published as received ones are.
+    let applied = tokio::task::spawn_blocking(move || {
         let mut replica = doc.blocking_write();
-        if replica.epoch == message.epoch {
+        if replica.epoch != message.epoch {
+            return 0;
+        }
+        {
+            let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
             let replica = &mut *replica;
             replica.signatures.offer(&replica.doc, signatures);
+            // With signatures required, changes waiting for them judged again
+            // (`access::screen`).
+            if crate::access::policy::of(&replica.doc).required() {
+                let applied = replica.logged_signed(|doc, known, log| {
+                    let applied = crate::access::screen::apply_bytes(
+                        doc,
+                        epoch,
+                        epoch_seq,
+                        &[],
+                        &screening,
+                        Some(known),
+                        log,
+                    );
+                    (applied, true)
+                });
+                replica.refresh_heads();
+                return match applied {
+                    Ok(applied) => applied,
+                    Err(error) => {
+                        tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
+                        0
+                    }
+                };
+            }
+            0
         }
     })
-    .await;
+    .await
+    .unwrap_or(0);
+    if applied > 0 {
+        state.publish_change();
+    }
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 8ca7d99..ff7fa5a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -608,13 +608,25 @@ pub(crate) async fn handle_sync(
             let epoch_seq = replica.epoch_seq;
             let mut message = message;
             replica
-                .logged(|doc, log| {
+                .logged_signed(|doc, known, log| {
                     let result = (|| {
-                        // Judged first, as the policy has it (`access::screen`).
-                        if crate::access::policy::of(&*doc).enforced() {
+                        // Judged first, as the policy has it (`access::screen`); 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 {
+                                known.offer(doc, signatures.clone());
+                            }
                             crate::access::screen::apply_bytes(
-                                doc, epoch, epoch_seq, &bytes, &screening, log,
+                                doc,
+                                epoch,
+                                epoch_seq,
+                                &bytes,
+                                &screening,
+                                required.then_some(&mut *known),
+                                log,
                             )?;
                         }
                         sync.receive_logged(doc, message, log)
@@ -879,15 +891,18 @@ async fn download_and_apply_delta(
             bail!("the epoch changed during the download");
         }
         let heads_before = replica.heads().to_vec();
-        if crate::access::policy::of(&replica.doc).enforced() {
+        let policy = crate::access::policy::of(&replica.doc);
+        if policy.enforced() {
             // Judged first (`access::screen`).
             let epoch_seq = replica.epoch_seq;
+            let replica = &mut *replica;
             crate::access::screen::apply_bytes(
                 &mut replica.doc,
                 epoch,
                 epoch_seq,
                 &bytes,
                 &screening,
+                policy.required().then_some(&mut replica.signatures),
                 &mut crate::document::PatchLog::inactive(),
             )?;
         } else {