1 files changed, 79 insertions, 0 deletions
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;
+}