diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/stats/prometheus.rs | 1 | ||||
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/simtest/bench.rs | 9 | ||||
| -rw-r--r-- | src/simtest/signatures.rs | 12 | ||||
| -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 | ||||
| -rw-r--r-- | src/transport/mod.rs | 16 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 6 | ||||
| -rw-r--r-- | src/transport/wire.rs | 13 |
10 files changed, 11 insertions, 167 deletions
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index dd49751..0d79d2c 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -193,7 +193,6 @@ 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 742da09..2f82c96 100644 --- a/src/node.rs +++ b/src/node.rs @@ -274,7 +274,6 @@ 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())); trace!("Remake worker task spawned"); tokio::spawn(swarm::topology::run_worker(state.clone())); diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs index 585ea1e..04b2eb6 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; 12] = [ +const KINDS: [&str; 11] = [ "SYNC", "STATUS", "STATUS_REPLY", @@ -503,10 +503,9 @@ const KINDS: [&str; 12] = [ "PROBE", "PRUNE/GRAFT", "NEIGHBOUR", - "SIGNATURES", ]; -fn counters(by: &ByKind) -> [Counter; 12] { +fn counters(by: &ByKind) -> [Counter; 11] { [ by.sync, by.status, @@ -519,11 +518,10 @@ fn counters(by: &ByKind) -> [Counter; 12] { by.probe, by.plumtree, by.hyparview, - by.signatures, ] } -fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] { +fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] { [ &mut by.sync, &mut by.status, @@ -536,7 +534,6 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] { &mut by.probe, &mut by.plumtree, &mut by.hyparview, - &mut by.signatures, ] } diff --git a/src/simtest/signatures.rs b/src/simtest/signatures.rs index c6232e7..b65b3a7 100644 --- a/src/simtest/signatures.rs +++ b/src/simtest/signatures.rs @@ -1,7 +1,7 @@ -//! Signed changes end to end (`document::signatures`, `swarm::signatures`): nodes under -//! the identity layer, whose node keys sign, write through the API as a user would, -//! and every node comes to see the writes verified, by signatures that came with sync -//! messages or on their own. +//! Signed changes end to end (`document::signatures`): nodes under the identity layer, +//! whose node keys sign, write through the API as a user would, and every node comes +//! to see the writes verified, by signatures that came with the changes in sync +//! messages. use std::time::Duration; @@ -80,8 +80,8 @@ async fn a_signed_write_vouches_for_the_writes_before() { let (actor, first) = write_as_user(&b, "metrics/cpu", json!(42), false).await; sim.wait_converged(&[&a, &b, &c], SOON).await; - // Signatures go within a second: none came for it. Unless b wrote protocol data - // since (always signed), whose signature vouches for it too. + // None is made for it, nor sent. Unless b wrote protocol data since (always + // signed), whose signature vouches for it too. tokio::time::sleep(Duration::from_secs(3)).await; let latest = b .doc 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(); - } -} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 640754a..d794271 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -252,8 +252,6 @@ 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), /// Changes the sender refused, of those the recipient brought (`swarm::refusals`). Refusals(RefusalsMessage), } @@ -273,17 +271,6 @@ pub(crate) struct RefusalsMessage { pub(crate) changes: Vec<crate::document::ChangeHash>, } -/// 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 /// transport address to the recipient's: a signature over both addresses /// (`transport::identity::signed`). WHO and ME end at the identity layer; the rest of @@ -429,7 +416,6 @@ pub(crate) enum MessageKind { Neighbour, NeighbourReply, Disconnect, - Signatures, Refusals, } @@ -453,7 +439,6 @@ impl ParsedMessage { ParsedMessage::Neighbour(_) => MessageKind::Neighbour, ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply, ParsedMessage::Disconnect(_) => MessageKind::Disconnect, - ParsedMessage::Signatures(_) => MessageKind::Signatures, ParsedMessage::Refusals(_) => MessageKind::Refusals, } } @@ -480,7 +465,6 @@ impl ParsedMessage { ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => { Some((answer.epoch, answer.epoch_seq)) } - ParsedMessage::Signatures(message) => Some((message.epoch, message.epoch_seq)), ParsedMessage::Refusals(message) => Some((message.epoch, message.epoch_seq)), ParsedMessage::Who(_) | ParsedMessage::Me(_) diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index e810595..b9506a9 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -57,8 +57,6 @@ 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, /// REFUSALS (`swarm::refusals`). pub(crate) refusals: Counter, } @@ -77,7 +75,6 @@ impl ByKind { MessageKind::Probe, MessageKind::Prune, MessageKind::Neighbour, - MessageKind::Signatures, MessageKind::Refusals, ] { let theirs = *other.get(kind); @@ -100,7 +97,6 @@ impl ByKind { MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { &self.hyparview } - MessageKind::Signatures => &self.signatures, MessageKind::Refusals => &self.refusals, } } @@ -120,7 +116,6 @@ impl ByKind { MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => { &mut self.hyparview } - MessageKind::Signatures => &mut self.signatures, MessageKind::Refusals => &mut self.refusals, } } @@ -331,7 +326,6 @@ 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"), MessageKind::Refusals => ("REFUSALS", "refusals"), } } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 5763407..450ecb0 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 = 45; +pub(crate) const VERSION: u8 = 46; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -36,7 +36,6 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Neighbour => 16, MessageKind::NeighbourReply => 17, MessageKind::Disconnect => 18, - MessageKind::Signatures => 19, MessageKind::Refusals => 20, } } @@ -99,7 +98,6 @@ 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)?, ParsedMessage::Refusals(message) => rmp_serde::encode::write(&mut payload, message)?, } Ok(payload) @@ -180,8 +178,6 @@ 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 if is(MessageKind::Refusals) { ParsedMessage::Refusals(rmp_serde::from_slice(body)?) } else { @@ -545,11 +541,6 @@ 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]), - }), ParsedMessage::Refusals(crate::transport::RefusalsMessage { epoch, epoch_seq: 1, @@ -586,7 +577,7 @@ mod tests { #[test] fn every_kind_round_trips() { let samples = samples(); - assert_eq!(samples.len(), 19, "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(); |