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.

9 files changed, 131 insertions, 5 deletions

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();