diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 10 | ||||
| -rw-r--r-- | src/swarm/signatures.rs | 109 |
3 files changed, 0 insertions, 120 deletions
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index e08bc62..ad7f83e 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -19,7 +19,6 @@ pub(crate) mod refusals; pub(crate) mod relay; pub(crate) mod roots; 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 3ddaf08..4a5b6ac 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -274,16 +274,6 @@ 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::Refusals(message) => { // From members, of our epoch: judging costs (`swarm::refusals`). let member = matches!( diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs deleted file mode 100644 index 111b3da..0000000 --- a/src/swarm/signatures.rs +++ /dev/null @@ -1,109 +0,0 @@ -//! 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 screening = std::sync::Arc::clone(&state.screening); - // How many waiting changes went in, if any: published as received ones are. - let applied = tokio::task::spawn_blocking(move || { - let mut replica = doc.blocking_write(); - if replica.epoch != message.epoch { - return 0; - } - let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); - let replica = &mut *replica; - replica.signatures.offer(&replica.doc, signatures); - // Changes waiting for them judged again (`access::screen`): those the policy they - // were written under requires signed. - let applied = replica.logged_signed(|doc, known, log| { - let applied = crate::access::screen::apply_bytes( - doc, - epoch, - epoch_seq, - &[], - &screening, - Some(known), - log, - ); - (applied, true) - }); - replica.refresh_heads(); - match applied { - Ok(outcome) => outcome.applied, - Err(error) => { - tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures"); - 0 - } - } - }) - .await - .unwrap_or(0); - if applied > 0 { - state.publish_change(); - } -} |
