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.

3 files changed, 90 insertions, 0 deletions

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;
+}