signatures: sign what is sent

A node signed its changes no rule signs a while after them (--sign-every
1s): the change left at once, unsigned, and its signature followed on
its own a second later, to every neighbour, over every link. A few
writes made a cascade of SIGNATURES. Under a policy requiring
signatures it signed every change as it was made instead.

A node now signs its latest change, if no signature vouches for it, as
a sync message takes it (--sign-every sent, the default): the
signature leaves in the same message, and goes where the change goes.
One signature per message, however many changes it carries, each
vouching for those before it: a slow writer's every change is signed,
a fast writer's batches are, at no threshold to pick. A policy
requiring signatures signs what is sent too, even with `none`. `all`
still signs every change as it is made; changes to protocol data,
listed paths and ?signed=true writes are signed as they are made, as
before. The interval, and the task signing on it, go.

7 files changed, 61 insertions, 79 deletions

diff --git a/docs/example.toml b/docs/example.toml
index 82ba016..9eeee20 100644
--- a/docs/example.toml
+++ b/docs/example.toml
@@ -75,10 +75,12 @@ htpasswd = "/etc/continuum/htpasswd"
 
 # Which of its changes this node signs with its node key besides those touching
 # protocol data, paths listed in _continuum/access/signed, or written with
-# ?signed=true: "all", "none" (one left unsigned is vouched for by the node's next
-# signed change), or its latest change, if unsigned, this often (docs/api.md, Signed
-# changes).
-# sign-every = "1s"
+# ?signed=true: "sent", its latest as a sync message takes it (one signature per
+# message, vouching for the changes before it); "all", every change as it is made; or
+# "none" (one left unsigned is vouched for by the node's next signed change; as "sent"
+# under a policy requiring signatures). docs/api.md, Signed changes, says which to
+# choose.
+# sign-every = "sent"
 
 # How long a token this node issues (POST /auth/login, /auth/refresh) is taken, by every
 # node of the swarm (docs/access.md, Tokens). Refreshed before it ends, it lasts until
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 4230b65..7df68e7 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -126,9 +126,9 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
     option(
         "sign-every",
         Kind::Text,
-        "all|none|DURATION",
-        Some("1s"),
-        "Which of its changes this node signs with its node key besides those touching protocol data, paths listed in _continuum/access/signed, or written with ?signed=true: all of them; none (one left unsigned is vouched for by the node's next signed change); or its latest change, if unsigned, this often, so every change is vouched for within it (docs/api.md, Signed changes).",
+        "all|sent|none",
+        Some("sent"),
+        "Which of its changes this node signs with its node key besides those touching protocol data, paths listed in _continuum/access/signed, or written with ?signed=true: its latest, as a sync message takes it (sent: one signature per message, vouching for the changes before it); every change as it is made (all); or none (one left unsigned is vouched for by the node's next signed change; as sent under a policy requiring signatures). docs/api.md, Signed changes, says which to choose.",
     ),
     option(
         "token-lifetime",
diff --git a/src/document/signatures.rs b/src/document/signatures.rs
index 102cad2..f2809fe 100644
--- a/src/document/signatures.rs
+++ b/src/document/signatures.rs
@@ -12,8 +12,8 @@
 //! touches protocol data (`_continuum`) or a path the swarm lists as signed
 //! (`_continuum/access/signed`, `listed`), or when its write asks for it (`Sign`); a
 //! change left unsigned is vouched for by the node's next signed one, which, as
-//! `--sign-every` says, is its latest change, signed a while after it if nothing signed
-//! it before (`SignEvery`), or every change at once.
+//! `--sign-every` says (`SignEvery`), is its latest change, signed as it leaves in a
+//! sync message (`sent`), or every change at once (`all`).
 //!
 //! Signatures travel with sync messages (`SyncEnvelope::signatures`): each actor's
 //! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received
@@ -86,49 +86,40 @@ pub(crate) fn listed(doc: &Document) -> Vec<Vec<String>> {
 }
 
 /// Which of this node's changes it signs besides those `Sign` says (`--sign-every`).
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub(crate) enum SignEvery {
     /// Every change, as it is made (`all`, `true`).
     Change,
+    /// Its latest change, if unsigned, as a sync message takes it (`sent`): one
+    /// signature per message, however many changes it carries, each vouching for those
+    /// before it.
+    #[default]
+    Sent,
     /// No others (`none`, `false`): one left unsigned waits for the node's next signed
-    /// change.
+    /// change; unless the policy requires signatures, then as `sent`.
     Never,
-    /// Its latest change, if unsigned, this often: every change is vouched for within it.
-    Interval(std::time::Duration),
-}
-
-impl Default for SignEvery {
-    fn default() -> Self {
-        SignEvery::Interval(std::time::Duration::from_secs(1))
-    }
 }
 
 impl std::fmt::Display for SignEvery {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            SignEvery::Change => f.write_str("all"),
-            SignEvery::Never => f.write_str("none"),
-            SignEvery::Interval(every) => f.write_str(&crate::units::format_duration(*every)),
-        }
+        f.write_str(match self {
+            SignEvery::Change => "all",
+            SignEvery::Sent => "sent",
+            SignEvery::Never => "none",
+        })
     }
 }
 
 impl std::str::FromStr for SignEvery {
     type Err = String;
 
-    /// `all` or `true`, `none` or `false`, or a duration (`units::parse_duration`; 0 is
-    /// `none`).
+    /// `all` or `true`, `sent`, `none` or `false`.
     fn from_str(text: &str) -> Result<Self, String> {
         match text.trim() {
             "all" | "true" => Ok(SignEvery::Change),
+            "sent" => Ok(SignEvery::Sent),
             "none" | "false" => Ok(SignEvery::Never),
-            other => match crate::units::parse_duration(other) {
-                Ok(every) if every.is_zero() => Ok(SignEvery::Never),
-                Ok(every) => Ok(SignEvery::Interval(every)),
-                Err(error) => Err(format!(
-                    "{other:?} is neither all, none nor a duration ({error})"
-                )),
-            },
+            other => Err(format!("{other:?} is neither all, sent nor none")),
         }
     }
 }
@@ -149,11 +140,8 @@ impl<'de> serde::Deserialize<'de> for SignEvery {
                 SignEvery::Never
             }),
             serde_json::Value::String(text) => text.parse().map_err(D::Error::custom),
-            serde_json::Value::Number(number) => {
-                number.to_string().parse().map_err(D::Error::custom)
-            }
             other => Err(D::Error::custom(format!(
-                "{other} is neither all, none nor a duration"
+                "{other} is neither all, sent nor none"
             ))),
         }
     }
@@ -524,7 +512,8 @@ impl Signatures {
         self.unsigned = None;
     }
 
-    /// Whether this node's latest change waits for a signature (`sign_latest`).
+    /// Whether this node's latest change waits for a signature (`sign_latest`, as a sync
+    /// message takes it).
     pub(crate) fn pending(&self) -> bool {
         self.unsigned.is_some()
     }
@@ -625,23 +614,18 @@ mod tests {
         assert!(check(&named, &hash, &key.sign(&message(&hash))).is_none());
     }
 
-    /// `--sign-every`: `all`, `none`, a duration; 0 is `none`.
+    /// `--sign-every`: `all`, `sent`, `none`.
     #[test]
     fn sign_every_reads() {
-        use std::time::Duration;
         let read = |text: &str| text.parse::<SignEvery>();
         assert_eq!(read("all"), Ok(SignEvery::Change));
         assert_eq!(read("true"), Ok(SignEvery::Change));
+        assert_eq!(read("sent"), Ok(SignEvery::Sent));
         assert_eq!(read("none"), Ok(SignEvery::Never));
-        assert_eq!(read("0"), Ok(SignEvery::Never));
-        assert_eq!(
-            read("500ms"),
-            Ok(SignEvery::Interval(Duration::from_millis(500)))
-        );
-        assert!(read("sometimes").is_err());
+        assert!(read("1s").is_err());
         let toml: SignEvery = serde_json::from_value(json!(false)).unwrap();
         assert_eq!(toml, SignEvery::Never);
-        assert_eq!(SignEvery::default().to_string(), "1s");
+        assert_eq!(SignEvery::default().to_string(), "sent");
     }
 
     /// The latest change left unsigned is signed later, once; a signature since covers it.
diff --git a/src/node.rs b/src/node.rs
index 314433c..742da09 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -275,9 +275,6 @@ impl Node {
 
         tokio::spawn(remake::run_worker(state.clone()));
         tokio::spawn(crate::swarm::signatures::run(state.clone()));
-        if let crate::document::signatures::SignEvery::Interval(every) = state.sign_every {
-            tokio::spawn(state.clone().sign_latest_every(every));
-        }
         trace!("Remake worker task spawned");
 
         tokio::spawn(swarm::topology::run_worker(state.clone()));
diff --git a/src/simtest/signatures.rs b/src/simtest/signatures.rs
index 2425a16..c6232e7 100644
--- a/src/simtest/signatures.rs
+++ b/src/simtest/signatures.rs
@@ -44,11 +44,10 @@ async fn verified(node: &AppState, actor: &ActorId, seq: u64) -> bool {
     node.doc.read().await.signatures.verified(actor, seq)
 }
 
-/// A write no rule signs is signed a while after it (`--sign-every`, 1 s by default),
-/// and the signature reaches every node on its own (SIGNATURES): the change went before
-/// it was signed.
+/// A write no rule signs is signed as a sync message takes it (`--sign-every sent`, the
+/// default), and the signature reaches every node with it.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn unsigned_writes_are_verified_within_the_interval() {
+async fn writes_leave_signed() {
     let sim = Sim::with_identity(61);
     let a = sim.start("a", &[], Start::Genesis);
     let b = sim.start("b", &["a"], Start::Join);
diff --git a/src/state/mod.rs b/src/state/mod.rs
index e079340..26910b6 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -692,8 +692,8 @@ impl AppState {
     }
 
     /// Signs the change just made in `replica`, if any (the one head after `before`, as
-    /// it builds on every head before), as `sign` says, or as `--sign-every` or the
-    /// policy has it; or keeps it as left unsigned.
+    /// it builds on every head before), as `sign` says, or as `--sign-every` has it; or
+    /// keeps it as left unsigned, for the sync message that takes it (`sign_sent`).
     fn sign_made(
         &self,
         replica: &mut crate::document::replica::Replica,
@@ -707,10 +707,7 @@ impl AppState {
         if after != before
             && let [hash] = after[..]
         {
-            if self.sign_every == SignEvery::Change
-                || sign.wanted(&replica.doc)
-                || crate::access::policy::of(&replica.doc).required()
-            {
+            if self.sign_every == SignEvery::Change || sign.wanted(&replica.doc) {
                 replica.signatures.sign(&replica.doc, hash, key);
             } else {
                 replica.signatures.left_unsigned(hash);
@@ -718,27 +715,27 @@ impl AppState {
         }
     }
 
-    /// This node's latest change signed, if no signature vouches for it yet, every
-    /// `every` (`--sign-every`): every change it makes is vouched for within that.
-    pub(crate) async fn sign_latest_every(self, every: Duration) {
-        let Some(key) = self.node_key.clone() else {
+    /// Before a sync message is made from `doc`: this node's latest change signed, if
+    /// no signature vouches for it yet and it signs what it sends (`--sign-every sent`,
+    /// or `none` under a policy requiring signatures), so it leaves with its signature.
+    /// Looked at first under the read lock: mostly there is nothing to sign. Blocking.
+    pub(crate) fn sign_sent(&self, doc: &crate::document::lock::DocumentLock) {
+        let Some(key) = &self.node_key else {
             return;
         };
-        let mut ticks = tokio::time::interval(every);
-        ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-        loop {
-            ticks.tick().await;
-            let state = self.clone();
-            let key = Arc::clone(&key);
-            let _ = tokio::task::spawn_blocking(move || {
-                // Looked at first under the read lock: mostly there is nothing to sign.
-                state.doc.blocking_read().signatures.pending() && {
-                    let mut replica = state.doc.blocking_write();
-                    let replica = &mut *replica;
-                    replica.signatures.sign_latest(&replica.doc, &key)
-                }
-            })
-            .await;
+        if self.sign_every == SignEvery::Change {
+            return;
+        }
+        let due = {
+            let replica = doc.blocking_read();
+            replica.signatures.pending()
+                && (self.sign_every == SignEvery::Sent
+                    || crate::access::policy::of(&replica.doc).required())
+        };
+        if due {
+            let mut replica = doc.blocking_write();
+            let replica = &mut *replica;
+            replica.signatures.sign_latest(&replica.doc, key);
         }
     }
 
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index e76f5f9..0c49fa3 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -167,9 +167,12 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
 
         let stats = Arc::clone(&state.stats);
         let clock = Arc::clone(&state.clock);
+        let signer = state.clone();
         // The session the message comes from, for its send's outcome.
         let (session, generated) = tokio::task::spawn_blocking(
             move || -> (u64, Result<Option<transport::EncodedMessage>>) {
+                // This node's latest change, if this message may take it, leaves signed.
+                signer.sign_sent(&doc);
                 // Label the message with the epoch of the document it was generated
                 // from, read under the same lock.
                 let (session, epoch, epoch_seq, body, writer, delivered, signatures) = {