signatures: on their own (SIGNATURES)

Signatures no sync message took go on their own within a second, to
each peer not sent them yet (wire version bumped).

Auth plan M5.

13 files changed, 140 insertions, 8 deletions

diff --git a/docs/api.md b/docs/api.md
index cc3cc05..a02d3d5 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -264,6 +264,8 @@ change as it is made, and `--sign-every none` signs only those above.
 Signatures travel with sync messages: a node passes on, for each node's writes, the
 latest signature it checked, and checks one it receives once it holds the change (a
 signature costs about 30 µs to check, and one per node and run vouches for the rest).
+Those no sync message took, as one made after its change went, go on their own within
+a second (SIGNATURES), to each peer that wasn't sent them yet.
 The signatures a node has are kept in its document file (`--document-file`) with the
 document.
 
diff --git a/docs/architecture.md b/docs/architecture.md
index f4ce611..1713d40 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -77,7 +77,7 @@ it switched to the right document without trusting anyone.
 ## Modules
 
 **[`transport/`](../src/transport/)** — [`mod.rs`](../src/transport/mod.rs) defines the message types (HELLO, HELLO_REPLY, sync
-envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, PROBE/PROBE_REPLY) and the `Transport` trait;
+envelope, STATUS/STATUS_REPLY, PING/PONG, blob messages, PROBE/PROBE_REPLY, SIGNATURES) and the `Transport` trait;
 [`wire.rs`](../src/transport/wire.rs) is the wire format (a version byte, bumped with every change to the format,
 and a kind byte, then the message in MessagePack, derived from the types in [`mod.rs`](../src/transport/mod.rs)
 with codecs for hashes and IDs); [`mycelium/`](../src/transport/mycelium/) (with the `mycelium` feature) is the REST
@@ -118,6 +118,7 @@ ignored 25-node soak test with a propagation bound (`cargo test soak -- --ignore
 | `tree` | writers' trees: passing each writer's changes on to the members after us in its tree, the same on every node | [`peers.md`](peers.md) |
 | `reach` | whom each neighbour reaches, as it says in its STATUS: the 2-hop neighbourhood trees are built on | [`peers.md`](peers.md) |
 | `relay` | announcing heads where trees don't reach (at once) and to everyone else (slow round), and pulling what's missing | module docs |
+| `signatures` | sending signatures of changes that no sync message took, every second | [`api.md`](api.md), Signed changes |
 | [`topology/`](../src/swarm/topology/) | which members are neighbours: ring plus low-score links, stable members preferred | module docs |
 | `members` | the member registry: aliases, alive stamps, tiers, the electorate | module docs |
 | `stability` | a node's own connectivity record and its stable/transient tier | module docs |
diff --git a/docs/measurements.md b/docs/measurements.md
index 9036031..4e1099f 100644
--- a/docs/measurements.md
+++ b/docs/measurements.md
@@ -180,7 +180,7 @@ Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind
 (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and
 ME, `paths` for PING and PONG, `probe` for PROBE
 and PROBE_REPLY, `plumtree` for PRUNE and GRAFT, `hyparview` for NEIGHBOUR,
-NEIGHBOUR_REPLY and DISCONNECT). The layers below the node count WHO and ME, and the
+NEIGHBOUR_REPLY and DISCONNECT, `signatures` for SIGNATURES). The layers below the node count WHO and ME, and the
 paths layer's own PINGs and PONGs, under the address they went to or came from; a peer
 (`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire
 encoding): not the transport's framing (for Mycelium, its topic and the base64 of its
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 2e131fb..7aba2fe 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -2046,7 +2046,7 @@ components:
 
     ByKind:
       type: object
-      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, links, probe, plumtree, hyparview]
+      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, probe, plumtree, hyparview, signatures]
       properties:
         hello:
           $ref: '#/components/schemas/Counter'
@@ -2076,6 +2076,9 @@ components:
         hyparview:
           $ref: '#/components/schemas/Counter'
           description: NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT, with --topology hyparview.
+        signatures:
+          $ref: '#/components/schemas/Counter'
+          description: SIGNATURES, signatures of changes no sync message took.
 
     Traffic:
       type: object
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 08928b4..038f739 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -193,6 +193,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
             (direction, "probe", by.probe),
             (direction, "plumtree", by.plumtree),
             (direction, "hyparview", by.hyparview),
+            (direction, "signatures", by.signatures),
         ]
     };
     let mut series = Vec::new();
diff --git a/src/node.rs b/src/node.rs
index 8cf327e..3d3d85d 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -239,6 +239,7 @@ impl Node {
         tokio::spawn(peer::probe_when_reachable(state.clone(), reachable));
 
         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));
         }
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 04b2eb6..585ea1e 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -491,7 +491,7 @@ impl Totals {
 }
 
 /// The kinds, in the order of the report.
-const KINDS: [&str; 11] = [
+const KINDS: [&str; 12] = [
     "SYNC",
     "STATUS",
     "STATUS_REPLY",
@@ -503,9 +503,10 @@ const KINDS: [&str; 11] = [
     "PROBE",
     "PRUNE/GRAFT",
     "NEIGHBOUR",
+    "SIGNATURES",
 ];
 
-fn counters(by: &ByKind) -> [Counter; 11] {
+fn counters(by: &ByKind) -> [Counter; 12] {
     [
         by.sync,
         by.status,
@@ -518,10 +519,11 @@ fn counters(by: &ByKind) -> [Counter; 11] {
         by.probe,
         by.plumtree,
         by.hyparview,
+        by.signatures,
     ]
 }
 
-fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
+fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] {
     [
         &mut by.sync,
         &mut by.status,
@@ -534,6 +536,7 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
         &mut by.probe,
         &mut by.plumtree,
         &mut by.hyparview,
+        &mut by.signatures,
     ]
 }
 
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 7b5b09c..0cb4fb8 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -17,6 +17,7 @@ pub(crate) mod plumtree;
 pub(crate) mod reach;
 pub(crate) mod relay;
 pub(crate) mod rtt;
+pub(crate) mod signatures;
 pub(crate) mod skew;
 pub(crate) mod source;
 pub(crate) mod stability;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index d25c7de..e261702 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -242,6 +242,16 @@ pub(crate) async fn handle_message(
             heard_heads(state, &peer, &remote);
             on_status(state, &peer, &remote, reactivated).await;
         }
+        transport::ParsedMessage::Signatures(message) => {
+            // From members, of our epoch: checking costs (`Signatures::offer`).
+            let member = matches!(
+                peer_state.lock().await.standing,
+                crate::state::Standing::Member { .. }
+            );
+            if member && same_epoch {
+                crate::swarm::signatures::received(state, message).await;
+            }
+        }
         transport::ParsedMessage::Probe(request) => {
             crate::swarm::working_links::answer(state, &peer, request).await;
         }
diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs
new file mode 100644
index 0000000..c172455
--- /dev/null
+++ b/src/swarm/signatures.rs
@@ -0,0 +1,79 @@
+//! Signatures of changes sent on their own (SIGNATURES), where no sync message took
+//! them: a signature made after its change went (`--sign-every`), or one received
+//! without the change it vouches for, as from a peer that had it already. Every
+//! `SEND_EVERY`, each active peer on our epoch is sent the signatures it wasn't sent yet
+//! (`Signatures::newer_than`), if any; it checks them as it checks those sync messages
+//! bring (`Signatures::offer`), and sends them on in turn: each actor's latest goes to
+//! each node about once.
+
+use std::time::Duration;
+
+use tracing::debug;
+
+use crate::state::AppState;
+use crate::transport::{ParsedMessage, SignaturesMessage};
+
+/// How often the signatures the peers weren't sent yet go out.
+pub(crate) const SEND_EVERY: Duration = Duration::from_secs(1);
+
+/// Sends the signatures due every `SEND_EVERY`. Runs for as long as the node does.
+pub(crate) async fn run(state: AppState) {
+    let mut ticks = tokio::time::interval(SEND_EVERY);
+    ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    loop {
+        ticks.tick().await;
+        send_due(&state).await;
+    }
+}
+
+/// Each active peer on our epoch: the signatures it wasn't sent yet, if any.
+async fn send_due(state: &AppState) {
+    for peer in state.active_peers_on_local_epoch().await {
+        let Some(peer_state) = state.get_peer_state(&peer).await else {
+            continue;
+        };
+        let doc = std::sync::Arc::clone(&state.doc);
+        // Lock order doc -> peer, as in sync generation.
+        let message = tokio::task::spawn_blocking(move || {
+            let replica = doc.blocking_read();
+            let mut peer_state = peer_state.blocking_lock();
+            let (signatures, sent) = replica.signatures.newer_than(peer_state.signatures_sent);
+            if signatures.is_empty() {
+                return None;
+            }
+            peer_state.signatures_sent = sent;
+            Some(ParsedMessage::Signatures(SignaturesMessage {
+                epoch: replica.epoch,
+                epoch_seq: replica.epoch_seq,
+                signatures: serde_bytes::ByteBuf::from(crate::document::signatures::pack(
+                    &signatures,
+                )),
+            }))
+        })
+        .await;
+        if let Ok(Some(message)) = message
+            && let Err(error) = state.transport.send_message(&peer, &message).await
+        {
+            // Lost, they come again with the next sync session (`PeerState::reset_sync`).
+            debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't send signatures");
+        }
+    }
+}
+
+/// SIGNATURES from `peer`, of our epoch: checked and kept as those of sync messages.
+pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
+    let Some(signatures) = crate::document::signatures::unpack(&message.signatures)
+        .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE)
+    else {
+        return;
+    };
+    let doc = std::sync::Arc::clone(&state.doc);
+    let _ = tokio::task::spawn_blocking(move || {
+        let mut replica = doc.blocking_write();
+        if replica.epoch == message.epoch {
+            let replica = &mut *replica;
+            replica.signatures.offer(&replica.doc, signatures);
+        }
+    })
+    .await;
+}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 35e1e5f..a1fb9d5 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -252,6 +252,19 @@ pub(crate) enum ParsedMessage {
     NeighbourReply(NeighbourAnswer),
     /// Tells the recipient we are neighbours no more (`swarm::hyparview`).
     Disconnect(NeighbourAnswer),
+    /// Signatures of changes, sent on their own (`swarm::signatures`).
+    Signatures(SignaturesMessage),
+}
+
+/// SIGNATURES: signatures of changes no sync message took (`swarm::signatures`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct SignaturesMessage {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// As `SyncEnvelope::signatures`.
+    pub(crate) signatures: serde_bytes::ByteBuf,
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -399,6 +412,7 @@ pub(crate) enum MessageKind {
     Neighbour,
     NeighbourReply,
     Disconnect,
+    Signatures,
 }
 
 impl ParsedMessage {
@@ -421,6 +435,7 @@ impl ParsedMessage {
             ParsedMessage::Neighbour(_) => MessageKind::Neighbour,
             ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply,
             ParsedMessage::Disconnect(_) => MessageKind::Disconnect,
+            ParsedMessage::Signatures(_) => MessageKind::Signatures,
         }
     }
 
@@ -446,6 +461,7 @@ impl ParsedMessage {
             ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
                 Some((answer.epoch, answer.epoch_seq))
             }
+            ParsedMessage::Signatures(message) => Some((message.epoch, message.epoch_seq)),
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 13ca9fa..180d3b5 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -57,6 +57,8 @@ pub(crate) struct ByKind {
     pub(crate) plumtree: Counter,
     /// NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT (`swarm::hyparview`).
     pub(crate) hyparview: Counter,
+    /// SIGNATURES (`swarm::signatures`).
+    pub(crate) signatures: Counter,
 }
 
 impl ByKind {
@@ -73,6 +75,7 @@ impl ByKind {
             MessageKind::Probe,
             MessageKind::Prune,
             MessageKind::Neighbour,
+            MessageKind::Signatures,
         ] {
             let theirs = *other.get(kind);
             self.get_mut(kind).absorb(theirs);
@@ -94,6 +97,7 @@ impl ByKind {
             MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
                 &self.hyparview
             }
+            MessageKind::Signatures => &self.signatures,
         }
     }
 
@@ -112,6 +116,7 @@ impl ByKind {
             MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
                 &mut self.hyparview
             }
+            MessageKind::Signatures => &mut self.signatures,
         }
     }
 }
@@ -321,6 +326,7 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
         MessageKind::Neighbour => ("NEIGHBOUR", "hyparview"),
         MessageKind::NeighbourReply => ("NEIGHBOUR_REPLY", "hyparview"),
         MessageKind::Disconnect => ("DISCONNECT", "hyparview"),
+        MessageKind::Signatures => ("SIGNATURES", "signatures"),
     }
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 97f8819..88b2916 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 40;
+pub(crate) const VERSION: u8 = 41;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -36,6 +36,7 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::Neighbour => 16,
         MessageKind::NeighbourReply => 17,
         MessageKind::Disconnect => 18,
+        MessageKind::Signatures => 19,
     }
 }
 
@@ -97,6 +98,7 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
         ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
             rmp_serde::encode::write(&mut payload, answer)?
         }
+        ParsedMessage::Signatures(message) => rmp_serde::encode::write(&mut payload, message)?,
     }
     Ok(payload)
 }
@@ -176,6 +178,8 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::NeighbourReply(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Disconnect) {
         ParsedMessage::Disconnect(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Signatures) {
+        ParsedMessage::Signatures(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -535,6 +539,11 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             agreed: false,
         }),
+        ParsedMessage::Signatures(crate::transport::SignaturesMessage {
+            epoch,
+            epoch_seq: 1,
+            signatures: serde_bytes::ByteBuf::from(vec![9; 96]),
+        }),
     ]
 }
 
@@ -566,7 +575,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 17, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 18, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();