diff --git a/Cargo.toml b/Cargo.toml index d9de2e1f0..fe44781f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -162,7 +162,9 @@ libp2p = { version = "0.55", features = [ "ed25519", "gossipsub", "identify", + "kad", "macros", + "mdns", "quic", "tokio", ] } diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index f54b41293..7011577cf 100644 --- a/lez/sequencer/core/Cargo.toml +++ b/lez/sequencer/core/Cargo.toml @@ -53,6 +53,8 @@ default = [] testnet = [] # Generate mock external clients implementations for testing mock = [] +# Enable mDNS-based local peer discovery for gossip. +mdns = [] [dev-dependencies] futures.workspace = true diff --git a/lez/sequencer/core/metrics/src/record.rs b/lez/sequencer/core/metrics/src/record.rs index 96b3304ed..b3848e389 100644 --- a/lez/sequencer/core/metrics/src/record.rs +++ b/lez/sequencer/core/metrics/src/record.rs @@ -16,6 +16,7 @@ use crate::names; pub enum TransactionOrigin { User, Sequencer, + Gossip, } #[derive(Debug, Clone, Copy, strum::IntoStaticStr, strum::EnumIter)] diff --git a/lez/sequencer/core/src/config.rs b/lez/sequencer/core/src/config.rs index 1a97e5f96..a3ae4d0d0 100644 --- a/lez/sequencer/core/src/config.rs +++ b/lez/sequencer/core/src/config.rs @@ -39,20 +39,12 @@ pub enum GenesisAction { /// entirely: no sockets, no background tasks. #[derive(Clone, Serialize, Deserialize)] pub struct GossipConfig { - /// Multiaddr to listen on. Deployments should set a concrete routable - /// address (announced addresses are taken from listeners; unspecified - /// IPs like 0.0.0.0 are filtered out of announcements). + /// Multiaddr to listen on. #[serde(default = "default_gossip_listen_addr")] pub listen_addr: String, /// Peer multiaddrs to dial at startup, optionally with `/p2p/`. #[serde(default)] pub bootstrap_peers: Vec, - /// Announcement heartbeat interval. - #[serde(with = "humantime_serde", default = "default_announce_interval")] - pub announce_interval: Duration, - /// Accredited-keys refresh interval. - #[serde(with = "humantime_serde", default = "default_keys_refresh_interval")] - pub keys_refresh_interval: Duration, } // TODO: Provide default values @@ -126,14 +118,6 @@ fn default_gossip_listen_addr() -> String { "/ip4/0.0.0.0/udp/0/quic-v1".to_owned() } -const fn default_announce_interval() -> Duration { - Duration::from_secs(60) -} - -const fn default_keys_refresh_interval() -> Duration { - Duration::from_secs(300) -} - #[expect(clippy::unnecessary_wraps, reason = "Required by serde")] const fn default_metrics_address() -> Option { Some(SequencerConfig::DEFAULT_METRICS_ADDRESS) @@ -153,22 +137,6 @@ mod tests { let config: GossipConfig = serde_json::from_str("{}").unwrap(); assert_eq!(config.listen_addr, "/ip4/0.0.0.0/udp/0/quic-v1"); assert!(config.bootstrap_peers.is_empty()); - assert_eq!(config.announce_interval, Duration::from_secs(60)); - assert_eq!(config.keys_refresh_interval, Duration::from_secs(300)); - } - - #[test] - fn gossip_config_parses_humantime_intervals() { - let config: GossipConfig = serde_json::from_str( - r#"{"listen_addr": "/ip4/127.0.0.1/udp/7070/quic-v1", - "bootstrap_peers": ["/ip4/127.0.0.1/udp/7071/quic-v1"], - "announce_interval": "5s", - "keys_refresh_interval": "1m"}"#, - ) - .unwrap(); - assert_eq!(config.bootstrap_peers.len(), 1); - assert_eq!(config.announce_interval, Duration::from_secs(5)); - assert_eq!(config.keys_refresh_interval, Duration::from_secs(60)); } #[test] diff --git a/lez/sequencer/core/src/gossip/keys_provider.rs b/lez/sequencer/core/src/gossip/accreditation/mod.rs similarity index 93% rename from lez/sequencer/core/src/gossip/keys_provider.rs rename to lez/sequencer/core/src/gossip/accreditation/mod.rs index abb9bfa81..fbdd4f58d 100644 --- a/lez/sequencer/core/src/gossip/keys_provider.rs +++ b/lez/sequencer/core/src/gossip/accreditation/mod.rs @@ -1,5 +1,6 @@ -//! Source of the channel's current accredited key set, polled by the -//! gossip layer to validate announcements. +//! Source of the channel's current accredited key set, read from L1. Retained +//! for future ChannelConfig-signature work; not currently consumed by the +//! gossip mesh. //! //! FIXME: `NodeKeysProvider` will be replaced by an L2 Join/Leave-derived provider in a follow-up. //! The related PR is . diff --git a/lez/sequencer/core/src/gossip/announcement.rs b/lez/sequencer/core/src/gossip/announcement.rs deleted file mode 100644 index 237d9e1dc..000000000 --- a/lez/sequencer/core/src/gossip/announcement.rs +++ /dev/null @@ -1,208 +0,0 @@ -//! The signed address announcement gossiped on the per-channel topic. -//! -//! Deliberately libp2p-free: pure wire format, signed with the bedrock -//! Ed25519 key so receivers validate against the channel's accredited set. - -use borsh::{BorshDeserialize, BorshSerialize}; -use logos_blockchain_core::mantle::ops::channel::Ed25519PublicKey; -use logos_blockchain_key_management_system_service::keys::{Ed25519Key, Ed25519Signature}; - -/// Caps enforced at validation; a violating message is rejected outright. -pub const MAX_LISTEN_ADDRS: usize = 8; -pub const MAX_ADDR_LEN: usize = 256; - -#[derive(Clone, Debug, PartialEq, Eq, BorshSerialize, BorshDeserialize)] -pub struct Announcement { - /// Cross-channel replay guard: must equal the topic's channel. - pub channel_id: [u8; 32], - /// Announcer's bedrock Ed25519 public key. - pub public_key: [u8; 32], - /// Multiaddrs the announcer listens on. - pub listen_addrs: Vec, - /// Freshness: unix millis at signing; receivers keep only the highest - /// per key. - pub seq: u64, -} - -#[derive(Clone, Debug, BorshSerialize, BorshDeserialize)] -pub struct SignedAnnouncement { - pub announcement: Announcement, - /// Ed25519 signature over `borsh(announcement)`. - pub signature: [u8; 64], -} - -/// Why a message must not be propagated (gossipsub `Reject`). Staleness and -/// unknown keys are `Ignore`, decided by the caller as they are not -/// structural faults of the message. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum RejectReason { - Undecodable, - WrongChannel, - TooManyAddrs, - AddrTooLong, - BadSignature, -} - -impl Announcement { - #[must_use] - pub fn sign(self, key: &Ed25519Key) -> SignedAnnouncement { - let payload = borsh::to_vec(&self).expect("announcement serialization cannot fail"); - let signature = key.sign_payload(&payload).to_bytes(); - SignedAnnouncement { - announcement: self, - signature, - } - } -} - -impl SignedAnnouncement { - #[must_use] - pub fn to_bytes(&self) -> Vec { - borsh::to_vec(self).expect("announcement serialization cannot fail") - } - - /// Structural & signature validation. - /// Accreditation and staleness are the caller's checks. - pub fn decode_and_verify( - bytes: &[u8], - expected_channel: &[u8; 32], - ) -> Result { - let signed: Self = borsh::from_slice(bytes).map_err(|_err| RejectReason::Undecodable)?; - let announcement = &signed.announcement; - - if &announcement.channel_id != expected_channel { - return Err(RejectReason::WrongChannel); - } - if announcement.listen_addrs.len() > MAX_LISTEN_ADDRS { - return Err(RejectReason::TooManyAddrs); - } - if announcement - .listen_addrs - .iter() - .any(|addr| addr.len() > MAX_ADDR_LEN) - { - return Err(RejectReason::AddrTooLong); - } - - let public_key = Ed25519PublicKey::from_bytes(&announcement.public_key) - .map_err(|_err| RejectReason::BadSignature)?; - let payload = borsh::to_vec(announcement).expect("announcement serialization cannot fail"); - let signature = Ed25519Signature::from_bytes(&signed.signature); - public_key - .verify(&payload, &signature) - .map_err(|_err| RejectReason::BadSignature)?; - - Ok(signed.announcement) - } -} - -/// `GossipSub` topic carrying [`SignedAnnouncement`]s for a channel. -#[must_use] -pub fn announcements_topic(channel_id: &[u8; 32]) -> String { - format!("/lez/{}/v1/announcements", hex::encode(channel_id)) -} - -#[cfg(test)] -mod tests { - use logos_blockchain_key_management_system_service::keys::Ed25519Key; - - use super::*; - - const CHANNEL: [u8; 32] = [1; 32]; - - fn signed(key: &Ed25519Key, addrs: Vec) -> SignedAnnouncement { - Announcement { - channel_id: CHANNEL, - public_key: key.public_key().to_bytes(), - listen_addrs: addrs, - seq: 42, - } - .sign(key) - } - - #[test] - fn round_trip_verifies() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let bytes = signed(&key, vec!["/ip4/127.0.0.1/udp/7070/quic-v1".into()]).to_bytes(); - let announcement = SignedAnnouncement::decode_and_verify(&bytes, &CHANNEL).unwrap(); - assert_eq!(announcement.seq, 42); - assert_eq!(announcement.public_key, key.public_key().to_bytes()); - } - - #[test] - fn garbage_is_undecodable() { - assert_eq!( - SignedAnnouncement::decode_and_verify(b"not borsh", &CHANNEL), - Err(RejectReason::Undecodable) - ); - } - - #[test] - fn wrong_channel_is_rejected() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let bytes = signed(&key, vec![]).to_bytes(); - assert_eq!( - SignedAnnouncement::decode_and_verify(&bytes, &[2; 32]), - Err(RejectReason::WrongChannel) - ); - } - - #[test] - fn tampered_payload_fails_signature() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let mut announcement = signed(&key, vec![]); - announcement.announcement.seq = 43; // signature no longer covers this - assert_eq!( - SignedAnnouncement::decode_and_verify(&announcement.to_bytes(), &CHANNEL), - Err(RejectReason::BadSignature) - ); - } - - #[test] - fn signature_from_other_key_fails() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let other = Ed25519Key::from_bytes(&[8; 32]); - // Claims key's identity but signed by other. - let bytes = Announcement { - channel_id: CHANNEL, - public_key: key.public_key().to_bytes(), - listen_addrs: vec![], - seq: 1, - } - .sign(&other) - .to_bytes(); - assert_eq!( - SignedAnnouncement::decode_and_verify(&bytes, &CHANNEL), - Err(RejectReason::BadSignature) - ); - } - - #[test] - fn too_many_addrs_rejected() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let addrs = vec!["/ip4/127.0.0.1/udp/1/quic-v1".to_owned(); MAX_LISTEN_ADDRS + 1]; - let bytes = signed(&key, addrs).to_bytes(); - assert_eq!( - SignedAnnouncement::decode_and_verify(&bytes, &CHANNEL), - Err(RejectReason::TooManyAddrs) - ); - } - - #[test] - fn oversized_addr_rejected() { - let key = Ed25519Key::from_bytes(&[7; 32]); - let bytes = signed(&key, vec!["a".repeat(MAX_ADDR_LEN + 1)]).to_bytes(); - assert_eq!( - SignedAnnouncement::decode_and_verify(&bytes, &CHANNEL), - Err(RejectReason::AddrTooLong) - ); - } - - #[test] - fn topic_is_channel_scoped() { - assert_eq!( - announcements_topic(&CHANNEL), - format!("/lez/{}/v1/announcements", hex::encode(CHANNEL)) - ); - } -} diff --git a/lez/sequencer/core/src/gossip/directory.rs b/lez/sequencer/core/src/gossip/directory.rs deleted file mode 100644 index 8406ecb45..000000000 --- a/lez/sequencer/core/src/gossip/directory.rs +++ /dev/null @@ -1,128 +0,0 @@ -//! Latest validated announcement per accredited key. Entries leave only via -//! [`PeerDirectory::retain_keys`] (accredited-set changes), never by -//! timeout — a silent peer's last known addresses stay dialable. - -use std::collections::{HashMap, HashSet}; - -use libp2p::{Multiaddr, PeerId}; - -pub struct PeerEntry { - pub peer_id: PeerId, - pub listen_addrs: Vec, - pub seq: u64, -} - -/// What [`PeerDirectory::upsert`] did with an announcement. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum UpsertOutcome { - /// Newer than anything stored for this key; entry replaced. - Fresh, - /// At or below the stored seq; entry untouched. - Stale, -} - -#[derive(Default)] -pub struct PeerDirectory { - entries: HashMap<[u8; 32], PeerEntry>, -} - -impl PeerDirectory { - pub fn upsert( - &mut self, - public_key: [u8; 32], - peer_id: PeerId, - listen_addrs: Vec, - seq: u64, - ) -> UpsertOutcome { - match self.entries.get(&public_key) { - Some(entry) if entry.seq >= seq => UpsertOutcome::Stale, - _ => { - self.entries.insert( - public_key, - PeerEntry { - peer_id, - listen_addrs, - seq, - }, - ); - UpsertOutcome::Fresh - } - } - } - - pub fn iter(&self) -> impl Iterator { - self.entries.iter() - } - - /// The stored freshness seq for `public_key`, if an entry exists. - #[must_use] - pub fn seq_of(&self, public_key: &[u8; 32]) -> Option { - self.entries.get(public_key).map(|entry| entry.seq) - } - - #[must_use] - pub fn pubkey_of(&self, peer_id: &PeerId) -> Option<[u8; 32]> { - self.entries - .iter() - .find(|(_, entry)| &entry.peer_id == peer_id) - .map(|(key, _)| *key) - } - - pub fn retain_keys(&mut self, accredited: &HashSet<[u8; 32]>) { - self.entries.retain(|key, _| accredited.contains(key)); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn peer(_n: u8) -> PeerId { - libp2p::identity::Keypair::generate_ed25519() - .public() - .to_peer_id() - } - - #[test] - fn upsert_newer_seq_wins_older_is_stale() { - let mut directory = PeerDirectory::default(); - let id = peer(1); - assert_eq!( - directory.upsert([1; 32], id, vec![], 10), - UpsertOutcome::Fresh - ); - assert_eq!( - directory.upsert([1; 32], id, vec![], 10), - UpsertOutcome::Stale, - "equal seq is a replay" - ); - assert_eq!( - directory.upsert([1; 32], id, vec![], 9), - UpsertOutcome::Stale - ); - assert_eq!( - directory.upsert([1; 32], id, vec![], 11), - UpsertOutcome::Fresh - ); - assert_eq!(directory.iter().next().unwrap().1.seq, 11); - } - - #[test] - fn pubkey_reverse_lookup() { - let mut directory = PeerDirectory::default(); - let id = peer(1); - directory.upsert([3; 32], id, vec![], 1); - assert_eq!(directory.pubkey_of(&id), Some([3; 32])); - assert_eq!(directory.pubkey_of(&peer(2)), None); - } - - #[test] - fn retain_keys_drops_deaccredited() { - let mut directory = PeerDirectory::default(); - directory.upsert([1; 32], peer(1), vec![], 1); - directory.upsert([2; 32], peer(2), vec![], 1); - directory.retain_keys(&std::collections::HashSet::from([[1; 32]])); - assert_eq!(directory.iter().count(), 1); - assert!(directory.iter().all(|(key, _)| key == &[1; 32])); - } -} diff --git a/lez/sequencer/core/src/gossip/mod.rs b/lez/sequencer/core/src/gossip/mod.rs index cd4ff2a42..af9fe4530 100644 --- a/lez/sequencer/core/src/gossip/mod.rs +++ b/lez/sequencer/core/src/gossip/mod.rs @@ -1,16 +1,15 @@ -//! Sequencer p2p gossip: a libp2p swarm gossiping signed sequencer address -//! announcements on a per-channel `GossipSub` topic. +//! Sequencer p2p gossip: a libp2p swarm that discovers peers via Kademlia, +//! Identify, and bootstrap (plus mDNS behind a cargo feature). //! //! p2p is a latency optimization, never a source of truth: gossip being //! down degrades to L1-only behavior, and a gossip failure after startup //! never halts the node. -pub use network::{Libp2pNetwork, PeerNetworkTrait}; +pub use network::{Libp2pNetwork, PeerNetworkTrait, TxPublisher}; -pub mod announcement; -pub mod directory; -pub mod keys_provider; +pub mod accreditation; pub mod network; +pub mod seen_cache; pub mod validation; #[cfg(test)] diff --git a/lez/sequencer/core/src/gossip/network.rs b/lez/sequencer/core/src/gossip/network.rs index 23c2ac7ff..6a7794c94 100644 --- a/lez/sequencer/core/src/gossip/network.rs +++ b/lez/sequencer/core/src/gossip/network.rs @@ -6,53 +6,50 @@ use std::{ collections::{HashMap, HashSet}, - time::{Duration, SystemTime, UNIX_EPOCH}, + time::Duration, }; use anyhow::{Context as _, Result, anyhow}; +use common::transaction::LeeTransaction; use futures::StreamExt as _; +#[cfg(feature = "mdns")] +use libp2p::mdns; use libp2p::{ Multiaddr, PeerId, SwarmBuilder, gossipsub, identify, identity::Keypair, + kad, + multiaddr::Protocol, swarm::{NetworkBehaviour, Swarm, SwarmEvent}, }; use log::{debug, error, info, warn}; -use logos_blockchain_key_management_system_service::keys::Ed25519Key; +use mempool::MemPoolHandle; use tokio::sync::{mpsc, watch}; use tokio_util::sync::CancellationToken; -use crate::{ - config::GossipConfig, - gossip::{ - announcement::{Announcement, MAX_LISTEN_ADDRS, announcements_topic}, - directory::PeerDirectory, - keys_provider::AccreditedKeysProvider, - }, -}; +use crate::{TransactionOrigin, config::GossipConfig, gossip::seen_cache::SeenCache}; /// How long to wait for the first listen address before failing startup. const LISTEN_TIMEOUT: Duration = Duration::from_secs(5); -/// Sweep interval for redialing disconnected accredited peers. -const DIAL_RETRY_INTERVAL: Duration = Duration::from_secs(2); -/// Per-peer dial backoff: base * 2^attempts, capped. -const DIAL_BACKOFF_BASE: Duration = Duration::from_secs(1); -const DIAL_BACKOFF_MAX: Duration = Duration::from_secs(300); -/// Isolation warning cadence and startup grace. -const NO_PEERS_WARN_INTERVAL: Duration = Duration::from_secs(60); -const NO_PEERS_GRACE: Duration = Duration::from_secs(30); /// Cadence of the post-death reminder that the node is running L1-only. const DEATH_REMINDER_INTERVAL: Duration = Duration::from_secs(300); +/// Recently-seen gossiped transaction hashes kept for dedup. +const SEEN_CACHE_CAPACITY: usize = 4096; +/// Outbound local-publish channel depth; `try_send` drops on overflow. +const TX_PUBLISH_CHANNEL_CAPACITY: usize = 1024; #[derive(NetworkBehaviour)] struct GossipBehaviour { gossipsub: gossipsub::Behaviour, identify: identify::Behaviour, + kademlia: kad::Behaviour, + #[cfg(feature = "mdns")] + mdns: mdns::tokio::Behaviour, } /// The seam `SequencerCore` and tests see. Later phases extend it with /// publish operations and inbound sinks. pub trait PeerNetworkTrait { - /// Bedrock Ed25519 public keys of currently connected accredited peers. + /// Ed25519 public keys of currently connected peers. fn connected_peers(&self) -> Vec<[u8; 32]>; /// Cancelled when the drive task terminates. Unlike the publisher's @@ -68,22 +65,35 @@ pub struct Libp2pNetwork { driver: tokio::task::JoinHandle<()>, listen_addrs: Vec, local_peer_id: PeerId, + tx_tx: mpsc::Sender, +} + +/// Handle for publishing locally-submitted transactions to the gossip mesh. +/// `publish` is non-blocking: a full channel drops the transaction rather +/// than back-pressuring the caller. +#[derive(Clone)] +pub struct TxPublisher(mpsc::Sender); + +impl TxPublisher { + pub fn publish(&self, tx: LeeTransaction) { + if let Err(err) = self.0.try_send(tx) { + debug!("Dropping local tx publish: outbound gossip channel full or closed: {err}"); + } + } } impl Libp2pNetwork { - /// Builds the swarm, binds `listen_addr`, subscribes to the channel's - /// announcements topic, dials bootstrap peers, and spawns the drive - /// task. Fails fast on a bad listen/bootstrap multiaddr or bind - /// failure; after that, gossip errors never halt the node. - pub async fn start( + /// Builds the swarm, binds `listen_addr`, seeds Kademlia and dials + /// bootstrap peers, and spawns the drive task. Fails fast on a bad + /// listen/bootstrap multiaddr or bind failure; after that, gossip errors + /// never halt the node. + pub async fn start( config: GossipConfig, channel_id: [u8; 32], secret_key: [u8; 32], - keys_provider: P, + mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, + max_block_size: u64, ) -> Result { - // Build the KMS signing key first: `ed25519_from_bytes` zeroizes - // its input. - let signing_key = Ed25519Key::from_bytes(&secret_key); let mut secret_for_libp2p = secret_key; let keypair = Keypair::ed25519_from_bytes(&mut secret_for_libp2p) .map_err(|err| anyhow!("Invalid bedrock signing key for libp2p identity: {err}"))?; @@ -102,8 +112,16 @@ impl Libp2pNetwork { }) .collect::>()?; + let topic = gossipsub::IdentTopic::new(format!("/lez/{}/v1/txs", hex::encode(channel_id))); + + let message_id_fn = |msg: &gossipsub::Message| { + let id = borsh::from_slice::(&msg.data) + .map_or_else(|_| msg.data.clone(), |tx| tx.hash().0.to_vec()); + gossipsub::MessageId::from(id) + }; let gossipsub_config = gossipsub::ConfigBuilder::default() .validation_mode(gossipsub::ValidationMode::Strict) + .message_id_fn(message_id_fn) .validate_messages() .build() .map_err(|err| anyhow!("Failed to build gossipsub config: {err}"))?; @@ -114,6 +132,15 @@ impl Libp2pNetwork { .map_err(|err| anyhow!("Failed to build gossipsub behaviour: {err}"))?; let identify_behaviour = identify::Behaviour::new(identify::Config::new("/lez/1".to_owned(), keypair.public())); + let kademlia_behaviour = { + let store = kad::store::MemoryStore::new(local_peer_id); + let mut kademlia = kad::Behaviour::new(local_peer_id, store); + kademlia.set_mode(Some(kad::Mode::Server)); + kademlia + }; + #[cfg(feature = "mdns")] + let mdns_behaviour = mdns::tokio::Behaviour::new(mdns::Config::default(), local_peer_id) + .map_err(|err| anyhow!("Failed to build mdns behaviour: {err}"))?; let mut swarm = SwarmBuilder::with_existing_identity(keypair) .with_tokio() @@ -121,11 +148,20 @@ impl Libp2pNetwork { .with_behaviour(|_key| GossipBehaviour { gossipsub: gossipsub_behaviour, identify: identify_behaviour, + kademlia: kademlia_behaviour, + #[cfg(feature = "mdns")] + mdns: mdns_behaviour, }) .expect("behaviour constructor is infallible") .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(60))) .build(); + swarm + .behaviour_mut() + .gossipsub + .subscribe(&topic) + .context("Failed to subscribe to gossip tx topic")?; + swarm .listen_on(listen_addr) .context("Failed to listen on gossip address")?; @@ -134,48 +170,44 @@ impl Libp2pNetwork { let listen_addrs = wait_for_listen_addr(&mut swarm).await?; info!("Gossip listening on {listen_addrs:?} as {local_peer_id}"); - let topic = gossipsub::IdentTopic::new(announcements_topic(&channel_id)); - swarm - .behaviour_mut() - .gossipsub - .subscribe(&topic) - .map_err(|err| anyhow!("Failed to subscribe to announcements topic: {err}"))?; - + // Seed Kademlia with bootstrap peers that carry an embedded peer id; + // dial the rest directly, since Kademlia can't route to an address + // without a known peer id. for addr in &bootstrap { + let embedded_peer_id = match addr.iter().last() { + Some(Protocol::P2p(peer_id)) => Some(peer_id), + _ => None, + }; + if let Some(peer_id) = embedded_peer_id { + swarm + .behaviour_mut() + .kademlia + .add_address(&peer_id, addr.clone()); + continue; + } if let Err(err) = swarm.dial(addr.clone()) { warn!("Failed to dial gossip bootstrap peer {addr}: {err}"); } } + if let Err(err) = swarm.behaviour_mut().kademlia.bootstrap() { + debug!("Kademlia bootstrap skipped (no known peers yet): {err}"); + } let (connected_tx, connected_rx) = watch::channel(Vec::new()); let driver_cancellation = CancellationToken::new(); - - // Accredited keys arrive over a watch fed by a separate refresher - // task, so the swarm loop never awaits an HTTP fetch. - let (keys_tx, keys_rx) = watch::channel(HashSet::new()); - let (refresh_tx, refresh_rx) = mpsc::channel(1); - tokio::spawn(run_keys_refresher( - keys_provider, - config.keys_refresh_interval, - refresh_rx, - keys_tx, - )); + let (tx_tx, tx_rx) = mpsc::channel::(TX_PUBLISH_CHANNEL_CAPACITY); let driver = tokio::spawn(run_drive_task(DriveTask { swarm, - topic, - channel_id, - signing_key, - own_pubkey: Ed25519Key::from_bytes(&secret_key).public_key().to_bytes(), - announce_interval: config.announce_interval, - bootstrap, - directory: PeerDirectory::default(), connected: HashSet::new(), - dial_backoff: HashMap::new(), + pubkeys: HashMap::new(), connected_tx, - keys_rx, - refresh_tx, cancellation: driver_cancellation.clone(), + topic, + mempool: mempool_handle, + seen: SeenCache::new(SEEN_CACHE_CAPACITY), + max_block_size, + tx_rx, })); spawn_death_reminder(driver_cancellation.clone()); @@ -185,6 +217,7 @@ impl Libp2pNetwork { driver, listen_addrs, local_peer_id, + tx_tx, }) } @@ -197,6 +230,12 @@ impl Libp2pNetwork { pub const fn local_peer_id(&self) -> PeerId { self.local_peer_id } + + /// Handle for publishing locally-submitted transactions to the mesh. + #[must_use] + pub fn tx_publisher(&self) -> TxPublisher { + TxPublisher(self.tx_tx.clone()) + } } impl PeerNetworkTrait for Libp2pNetwork { @@ -219,32 +258,27 @@ impl Drop for Libp2pNetwork { /// Everything the drive task owns. struct DriveTask { swarm: Swarm, - topic: gossipsub::IdentTopic, - channel_id: [u8; 32], - signing_key: Ed25519Key, - own_pubkey: [u8; 32], - announce_interval: Duration, - bootstrap: Vec, - directory: PeerDirectory, connected: HashSet, - /// Per-peer dial backoff: (attempts, earliest next attempt). - dial_backoff: HashMap, + /// Ed25519 public keys of peers seen via Identify, keyed by `PeerId`. + pubkeys: HashMap, connected_tx: watch::Sender>, - keys_rx: watch::Receiver>, - refresh_tx: mpsc::Sender<()>, cancellation: CancellationToken, + topic: gossipsub::IdentTopic, + mempool: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, + seen: SeenCache, + max_block_size: u64, + tx_rx: mpsc::Receiver, } impl DriveTask { #[expect( clippy::wildcard_enum_match_arm, - reason = "SwarmEvent is non_exhaustive; only connection and message events are handled" + reason = "SwarmEvent is non_exhaustive; only connection and behaviour events are handled" )] fn on_swarm_event(&mut self, event: SwarmEvent) { match event { SwarmEvent::ConnectionEstablished { peer_id, .. } => { self.connected.insert(peer_id); - self.dial_backoff.remove(&peer_id); self.update_connected_watch(); } SwarmEvent::ConnectionClosed { @@ -253,186 +287,56 @@ impl DriveTask { .. } => { self.connected.remove(&peer_id); + self.pubkeys.remove(&peer_id); self.update_connected_watch(); } - SwarmEvent::Behaviour(GossipBehaviourEvent::Gossipsub(gossipsub::Event::Message { - propagation_source, - message_id, - message, - })) => self.on_gossip_message(propagation_source, &message_id, &message), + SwarmEvent::Behaviour(behaviour_event) => self.on_behaviour_event(behaviour_event), _ => {} } } - fn on_gossip_message( - &mut self, - source: PeerId, - message_id: &gossipsub::MessageId, - message: &gossipsub::Message, - ) { - use crate::gossip::validation::{Evaluation, evaluate_announcement}; - - let evaluation = evaluate_announcement( - &message.data, - &self.channel_id, - &self.own_pubkey, - &self.keys_rx.borrow(), - &self.directory, - ); - - let acceptance = match evaluation { - Evaluation::Reject(reason) => { - debug!("Rejecting gossip announcement from {source}: {reason:?}"); - gossipsub::MessageAcceptance::Reject + // `GossipBehaviourEvent` is generated by `#[derive(NetworkBehaviour)]`; + // clippy does not flag wildcard matches against macro-generated enums, + // so no `#[expect(clippy::wildcard_enum_match_arm)]` is needed here. + fn on_behaviour_event(&mut self, event: GossipBehaviourEvent) { + match event { + GossipBehaviourEvent::Gossipsub(gossipsub::Event::Message { + propagation_source, + message_id, + message, + }) => { + self.on_gossip_message(propagation_source, &message_id, &message.data); } - Evaluation::IgnoreUnknownKey => { - // Our cached accredited set may be stale; nudge the - // refresher (it rate-limits internally). Never blocks; a full - // channel already has a refresh pending, so a dropped nudge is fine. - _ = self.refresh_tx.try_send(()); - gossipsub::MessageAcceptance::Ignore - } - Evaluation::IgnoreOwn | Evaluation::IgnoreStale => gossipsub::MessageAcceptance::Ignore, - Evaluation::Accept { - public_key, - peer_id, - listen_addrs, - seq, - } => { - info!( - "Learned gossip addresses for sequencer {} ({} addrs)", - hex::encode(public_key), - listen_addrs.len() - ); - self.directory - .upsert(public_key, peer_id, listen_addrs, seq); - // A fresh entry may map an already-open connection to its - // key, or name a peer we should dial now. - self.update_connected_watch(); - self.dial_missing_peers(); - gossipsub::MessageAcceptance::Accept - } - }; - - let _ = self - .swarm - .behaviour_mut() - .gossipsub - .report_message_validation_result(message_id, &source, acceptance); - } - - fn publish_announcement(&mut self) { - // Announce concrete listener addresses; unspecified IPs (0.0.0.0) - // are useless to remote peers. - let addrs: Vec = self - .swarm - .listeners() - .chain(self.swarm.external_addresses()) - .filter(|addr| !is_unspecified(addr)) - .map(ToString::to_string) - .collect::>() - .into_iter() - .take(MAX_LISTEN_ADDRS) - .collect(); - if addrs.is_empty() { - return; - } - - let announcement = Announcement { - channel_id: self.channel_id, - public_key: self.own_pubkey, - listen_addrs: addrs, - seq: unix_millis(), - }; - let bytes = announcement.sign(&self.signing_key).to_bytes(); - if let Err(err) = self - .swarm - .behaviour_mut() - .gossipsub - .publish(self.topic.clone(), bytes) - { - // `InsufficientPeers` while alone is the normal quiet state. - debug!("Skipping gossip announcement publish: {err}"); - } - } - - /// Dial every accredited directory entry we are not connected to, with - /// per-peer exponential backoff. Bootstrap addresses are retried only - /// while fully disconnected (they may lack a peer id to track). - fn dial_missing_peers(&mut self) { - let now = tokio::time::Instant::now(); - let accredited = self.keys_rx.borrow().clone(); - - let candidates: Vec<(PeerId, Vec)> = self - .directory - .iter() - .filter(|(key, entry)| { - accredited.contains(*key) && !self.connected.contains(&entry.peer_id) - }) - .map(|(_, entry)| (entry.peer_id, entry.listen_addrs.clone())) - .collect(); - - for (peer_id, addrs) in candidates { - if let Some((_, next_attempt)) = self.dial_backoff.get(&peer_id) - && *next_attempt > now - { - continue; - } - let (attempts, _) = self.dial_backoff.remove(&peer_id).unwrap_or((0, now)); - let delay = DIAL_BACKOFF_BASE - .saturating_mul(2_u32.saturating_pow(attempts)) - .min(DIAL_BACKOFF_MAX); - let next_attempt = now.checked_add(delay).unwrap_or(now); - self.dial_backoff - .insert(peer_id, (attempts.saturating_add(1), next_attempt)); - - let opts = libp2p::swarm::dial_opts::DialOpts::peer_id(peer_id) - .addresses(addrs) - .build(); - if let Err(err) = self.swarm.dial(opts) { - debug!("Gossip dial of {peer_id} failed to start: {err}"); - } - } - - if self.connected.is_empty() { - for addr in self.bootstrap.clone() { - if let Err(err) = self.swarm.dial(addr.clone()) { - debug!("Gossip bootstrap redial of {addr} failed: {err}"); + GossipBehaviourEvent::Identify(identify::Event::Received { peer_id, info, .. }) => { + if let Ok(ed25519_pubkey) = info.public_key.try_into_ed25519() { + self.pubkeys.insert(peer_id, ed25519_pubkey.to_bytes()); + self.update_connected_watch(); + } + for addr in info.listen_addrs { + self.swarm + .behaviour_mut() + .kademlia + .add_address(&peer_id, addr); } } + #[cfg(feature = "mdns")] + GossipBehaviourEvent::Mdns(mdns::Event::Discovered(peers)) => { + for (peer_id, addr) in peers { + if let Err(err) = self.swarm.dial(addr) { + debug!("Failed to dial mdns-discovered peer {peer_id}: {err}"); + } + } + } + _ => {} } } - fn warn_if_isolated(&self, started_at: tokio::time::Instant) { - // Snapshot the watch once: `connected_accredited` must never - // re-borrow `keys_rx` while a guard is held. - let accredited = self.keys_rx.borrow().clone(); - if started_at.elapsed() > NO_PEERS_GRACE - && accredited.len() > 1 - && self.connected_accredited(&accredited).next().is_none() - { - warn!( - "Channel has {} accredited sequencers but no gossip peers are connected — \ - check `gossip.bootstrap_peers` in the sequencer config", - accredited.len() - ); - } - } - - /// Connected peers that map to an accredited key via the directory. - fn connected_accredited<'keys>( - &'keys self, - accredited: &'keys HashSet<[u8; 32]>, - ) -> impl Iterator + 'keys { - self.connected - .iter() - .filter_map(|peer_id| self.directory.pubkey_of(peer_id)) - .filter(move |pubkey| accredited.contains(pubkey)) - } - fn update_connected_watch(&self) { - let accredited = self.keys_rx.borrow().clone(); - let mut peers: Vec<[u8; 32]> = self.connected_accredited(&accredited).collect(); + let mut peers: Vec<[u8; 32]> = self + .connected + .iter() + .filter_map(|peer_id| self.pubkeys.get(peer_id).copied()) + .collect(); peers.sort_unstable(); self.connected_tx.send_if_modified(|current| { if *current == peers { @@ -443,10 +347,71 @@ impl DriveTask { } }); } + + /// Validates an inbound gossiped transaction and reports the mesh + /// acceptance decision, admitting it to the mempool on first sight. + fn on_gossip_message( + &mut self, + source: PeerId, + message_id: &gossipsub::MessageId, + data: &[u8], + ) { + use crate::gossip::validation::{TxEvaluation, evaluate_transaction}; + + let acceptance = match evaluate_transaction(data, self.max_block_size) { + TxEvaluation::Reject(reason) => { + debug!("Rejecting gossiped tx from {source}: {reason}"); + gossipsub::MessageAcceptance::Reject + } + TxEvaluation::Ignore => gossipsub::MessageAcceptance::Ignore, + TxEvaluation::Accept(tx) => { + let hash = tx.hash(); + if !self.seen.insert(hash) { + gossipsub::MessageAcceptance::Ignore + } else if self + .mempool + .try_push((TransactionOrigin::Gossip, tx)) + .is_err() + { + debug!("Mempool full; dropping gossiped tx {hash:?}"); + gossipsub::MessageAcceptance::Ignore + } else { + gossipsub::MessageAcceptance::Accept + } + } + }; + _ = self + .swarm + .behaviour_mut() + .gossipsub + .report_message_validation_result(message_id, &source, acceptance); + } + + /// Publishes a locally-submitted transaction to the mesh. + fn publish_transaction(&mut self, tx: &LeeTransaction) { + let hash = tx.hash(); + self.seen.insert(hash); + let bytes = borsh::to_vec(tx).expect("tx borsh serialization should not fail"); + if let Err(err) = self + .swarm + .behaviour_mut() + .gossipsub + .publish(self.topic.clone(), bytes) + { + debug!("Skipping local tx publish {hash:?}: {err}"); + } + } } /// Derives the libp2p `PeerId` an Ed25519 public key produces. `None` only /// for byte strings that are not a valid curve point. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "unused by the mesh until a later gossip task; exercised by the identity test below" + ) +)] pub(crate) fn peer_id_from_ed25519(pubkey: &[u8; 32]) -> Option { libp2p::identity::ed25519::PublicKey::try_from_bytes(pubkey) .ok() @@ -491,65 +456,10 @@ async fn run_drive_task(mut task: DriveTask) { // Cancelled on return or panic, so observers learn the driver is gone. let _guard = task.cancellation.clone().drop_guard(); - let started_at = tokio::time::Instant::now(); - let mut announce_interval = tokio::time::interval(task.announce_interval); - let mut dial_interval = tokio::time::interval(DIAL_RETRY_INTERVAL); - let mut warn_interval = tokio::time::interval(NO_PEERS_WARN_INTERVAL); - loop { tokio::select! { event = task.swarm.select_next_some() => task.on_swarm_event(event), - _ = announce_interval.tick() => task.publish_announcement(), - _ = dial_interval.tick() => task.dial_missing_peers(), - _ = warn_interval.tick() => task.warn_if_isolated(started_at), - } - } -} - -#[expect( - clippy::integer_division_remainder_used, - reason = "Generated by select! macro, can't be easily rewritten to avoid this lint" -)] -async fn run_keys_refresher( - provider: P, - refresh_interval: Duration, - mut refresh_rx: mpsc::Receiver<()>, - keys_tx: watch::Sender>, -) { - /// Minimum spacing for demand-driven (unknown-key) refreshes. - const MIN_SPACING: Duration = Duration::from_secs(10); - let mut interval = tokio::time::interval(refresh_interval); - let mut last_fetch = tokio::time::Instant::now() - .checked_sub(MIN_SPACING) - .unwrap_or_else(tokio::time::Instant::now); - - loop { - tokio::select! { - _ = interval.tick() => {} - request = refresh_rx.recv() => { - if request.is_none() { - return; // Drive task gone. - } - if last_fetch.elapsed() < MIN_SPACING { - continue; - } - } - } - last_fetch = tokio::time::Instant::now(); - match provider.accredited_keys().await { - // Errors keep the last known set — never shrink on a fetch hiccup. - Err(err) => warn!("Failed to refresh accredited keys for gossip: {err:#}"), - Ok(keys) => { - keys_tx.send_if_modified(|current| { - if *current == keys { - false - } else { - info!("Gossip accredited key set updated ({} keys)", keys.len()); - *current = keys; - true - } - }); - } + Some(tx) = task.tx_rx.recv() => task.publish_transaction(&tx), } } } @@ -568,48 +478,30 @@ fn spawn_death_reminder(cancellation: CancellationToken) { }); } -#[expect( - clippy::wildcard_enum_match_arm, - reason = "Protocol is non_exhaustive with many variants; only IP variants matter here" -)] -fn is_unspecified(addr: &Multiaddr) -> bool { - addr.iter().any(|proto| match proto { - libp2p::multiaddr::Protocol::Ip4(ip) => ip.is_unspecified(), - libp2p::multiaddr::Protocol::Ip6(ip) => ip.is_unspecified(), - _ => false, - }) -} - -fn unix_millis() -> u64 { - u64::try_from( - SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("system clock before unix epoch") - .as_millis(), - ) - .expect("timestamp fits u64") -} - #[cfg(test)] mod tests { use logos_blockchain_key_management_system_service::keys::Ed25519Key; use super::*; - use crate::{config::GossipConfig, gossip::keys_provider::StaticKeysProvider}; + use crate::config::GossipConfig; + + const TEST_MAX_BLOCK_SIZE: u64 = 1 << 20; fn test_config() -> GossipConfig { GossipConfig { listen_addr: "/ip4/127.0.0.1/udp/0/quic-v1".to_owned(), bootstrap_peers: vec![], - announce_interval: std::time::Duration::from_secs(1), - keys_refresh_interval: std::time::Duration::from_secs(3600), } } + fn test_mempool_handle() -> MemPoolHandle<(TransactionOrigin, LeeTransaction)> { + mempool::MemPool::new(1000).1 + } + #[test] fn libp2p_identity_matches_kms_public_key() { - // The PeerId derived from an announcement's public_key must equal - // the PeerId the same secret produces as a libp2p identity. + // The PeerId derived from an Ed25519 public key must equal the + // PeerId the same secret produces as a libp2p identity. let secret = [9; 32]; let kms_pubkey = Ed25519Key::from_bytes(&secret).public_key().to_bytes(); let mut secret_for_libp2p = secret; @@ -627,7 +519,8 @@ mod tests { test_config(), [1; 32], [9; 32], - StaticKeysProvider(std::collections::HashSet::new()), + test_mempool_handle(), + TEST_MAX_BLOCK_SIZE, ) .await .unwrap(); @@ -648,7 +541,8 @@ mod tests { config, [1; 32], [9; 32], - StaticKeysProvider(std::collections::HashSet::new()), + test_mempool_handle(), + TEST_MAX_BLOCK_SIZE ) .await .is_err() @@ -666,7 +560,8 @@ mod tests { config, [1; 32], [9; 32], - StaticKeysProvider(std::collections::HashSet::new()), + test_mempool_handle(), + TEST_MAX_BLOCK_SIZE ) .await .is_err() @@ -679,7 +574,8 @@ mod tests { test_config(), [1; 32], [9; 32], - StaticKeysProvider(std::collections::HashSet::new()), + test_mempool_handle(), + TEST_MAX_BLOCK_SIZE, ) .await .unwrap(); diff --git a/lez/sequencer/core/src/gossip/seen_cache.rs b/lez/sequencer/core/src/gossip/seen_cache.rs new file mode 100644 index 000000000..69fc318c6 --- /dev/null +++ b/lez/sequencer/core/src/gossip/seen_cache.rs @@ -0,0 +1,116 @@ +//! Bounded, FIFO-eviction membership cache over transaction hashes. +//! +//! The mempool is a plain channel with no dedup, so the gossip layer tracks +//! recently seen transactions here to avoid re-admitting duplicates that +//! arrive from multiple peers or echo back after a local publish. Counters +//! are exposed for a future metrics surface. + +use std::collections::{HashSet, VecDeque}; + +use common::HashType; + +pub struct SeenCache { + capacity: usize, + order: VecDeque, + set: HashSet, + hits: u64, + inserts: u64, + evictions: u64, +} + +impl SeenCache { + #[must_use] + pub fn new(capacity: usize) -> Self { + Self { + capacity: capacity.max(1), + order: VecDeque::new(), + set: HashSet::new(), + hits: 0, + inserts: 0, + evictions: 0, + } + } + + /// Records `hash` as seen. Returns `true` if it was newly inserted, + /// `false` if already present (a hit). Evicts the oldest entry when full. + pub fn insert(&mut self, hash: HashType) -> bool { + if self.set.contains(&hash) { + self.hits = self.hits.saturating_add(1); + return false; + } + if self.order.len() >= self.capacity + && let Some(oldest) = self.order.pop_front() + { + self.set.remove(&oldest); + self.evictions = self.evictions.saturating_add(1); + } + self.set.insert(hash); + self.order.push_back(hash); + self.inserts = self.inserts.saturating_add(1); + true + } + + #[must_use] + pub fn contains(&self, hash: &HashType) -> bool { + self.set.contains(hash) + } + + #[must_use] + pub fn len(&self) -> usize { + self.order.len() + } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.order.is_empty() + } + + #[must_use] + pub const fn hits(&self) -> u64 { + self.hits + } + + #[must_use] + pub const fn inserts(&self) -> u64 { + self.inserts + } + + #[must_use] + pub const fn evictions(&self) -> u64 { + self.evictions + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn h(n: u8) -> HashType { + HashType([n; 32]) + } + + #[test] + fn insert_reports_novelty_and_contains() { + let mut cache = SeenCache::new(4); + assert!(cache.insert(h(1))); + assert!(!cache.insert(h(1))); + assert!(cache.contains(&h(1))); + assert!(!cache.contains(&h(2))); + assert_eq!(cache.len(), 1); + assert_eq!(cache.inserts(), 1); + assert_eq!(cache.hits(), 1); + } + + #[test] + fn evicts_oldest_past_capacity() { + let mut cache = SeenCache::new(2); + assert!(cache.insert(h(1))); + assert!(cache.insert(h(2))); + assert!(cache.insert(h(3))); // evicts h(1) + assert!(!cache.contains(&h(1))); + assert!(cache.contains(&h(2))); + assert!(cache.contains(&h(3))); + assert_eq!(cache.len(), 2); + assert_eq!(cache.evictions(), 1); + } +} diff --git a/lez/sequencer/core/src/gossip/tests.rs b/lez/sequencer/core/src/gossip/tests.rs index 00a90642b..e2b11d81f 100644 --- a/lez/sequencer/core/src/gossip/tests.rs +++ b/lez/sequencer/core/src/gossip/tests.rs @@ -1,44 +1,70 @@ //! Multi-node integration tests over real QUIC on 127.0.0.1. -use std::{ - collections::HashSet, - time::{Duration, Instant}, -}; +use std::time::{Duration, Instant}; +use common::transaction::LeeTransaction; use logos_blockchain_key_management_system_service::keys::Ed25519Key; +use mempool::MemPool; +use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts}; use crate::{ + TransactionOrigin, config::GossipConfig, - gossip::{Libp2pNetwork, PeerNetworkTrait as _, keys_provider::StaticKeysProvider}, + gossip::{Libp2pNetwork, PeerNetworkTrait as _}, }; const CHANNEL: [u8; 32] = [1; 32]; +const TEST_MAX_BLOCK_SIZE: u64 = 1 << 20; fn pubkey(secret: [u8; 32]) -> [u8; 32] { Ed25519Key::from_bytes(&secret).public_key().to_bytes() } +/// A real, validly-signed transfer, reusing the same helper the RPC-side +/// admission tests use. +fn valid_transaction() -> LeeTransaction { + let acc1 = initial_public_user_accounts()[0].account_id; + let acc2 = initial_public_user_accounts()[1].account_id; + let sign_key1 = initial_pub_accounts_private_keys()[0].pub_sign_key.clone(); + common::test_utils::create_transaction_native_token_transfer(acc1, 0, acc2, 10, &sign_key1) +} + +/// Structurally well-formed but with a signature/public-key pair that does +/// not match, so it decodes but fails the stateless witness check; used to +/// exercise rejection through the real gossip pipeline rather than +/// `evaluate_transaction` directly. +fn invalidly_signed_transaction() -> LeeTransaction { + let LeeTransaction::Public(mut tx) = valid_transaction() else { + unreachable!("valid_transaction always builds a Public transaction"); + }; + let (signature, _correct_public_key) = tx.witness_set.signatures_and_public_keys()[0].clone(); + let wrong_public_key = + lee::PublicKey::new_from_private_key(&initial_pub_accounts_private_keys()[1].pub_sign_key); + tx.witness_set = + lee::public_transaction::WitnessSet::from_raw_parts(vec![(signature, wrong_public_key)]); + LeeTransaction::Public(tx) +} + async fn start_node( secret: [u8; 32], - accredited: HashSet<[u8; 32]>, bootstrap: Vec, -) -> Libp2pNetwork { +) -> (Libp2pNetwork, MemPool<(TransactionOrigin, LeeTransaction)>) { let config = GossipConfig { listen_addr: "/ip4/127.0.0.1/udp/0/quic-v1".to_owned(), bootstrap_peers: bootstrap, - announce_interval: Duration::from_millis(500), - keys_refresh_interval: Duration::from_secs(3600), }; - Libp2pNetwork::start(config, CHANNEL, secret, StaticKeysProvider(accredited)) - .await - .expect("node should start") + let (mempool, mempool_handle) = MemPool::new(1000); + let network = + Libp2pNetwork::start(config, CHANNEL, secret, mempool_handle, TEST_MAX_BLOCK_SIZE) + .await + .expect("node should start"); + (network, mempool) } -/// Polls `condition` until it holds or `timeout` elapses. async fn wait_for(timeout: Duration, mut condition: impl FnMut() -> bool) -> bool { let deadline = Instant::now() .checked_add(timeout) - .expect("test deadline within Instant range"); + .expect("deadline within Instant range"); while Instant::now() < deadline { if condition() { return true; @@ -49,66 +75,96 @@ async fn wait_for(timeout: Duration, mut condition: impl FnMut() -> bool) -> boo } #[tokio::test] -async fn transitive_discovery_connects_all_accredited_nodes() { +async fn nodes_discover_each_other_via_bootstrap() { let secrets = [[10; 32], [11; 32], [12; 32]]; - let accredited: HashSet<[u8; 32]> = secrets.iter().map(|secret| pubkey(*secret)).collect(); - - // A is the only bootstrap point; B and C never learn each other's - // addresses from config. - let node_a = start_node(secrets[0], accredited.clone(), vec![]).await; + let (node_a, _mempool_a) = start_node(secrets[0], vec![]).await; let a_addr = node_a.listen_addrs()[0].to_string(); - let node_b = start_node(secrets[1], accredited.clone(), vec![a_addr.clone()]).await; - let node_c = start_node(secrets[2], accredited.clone(), vec![a_addr]).await; + let (node_b, _mempool_b) = start_node(secrets[1], vec![a_addr.clone()]).await; + let (node_c, _mempool_c) = start_node(secrets[2], vec![a_addr]).await; - // C must discover B via A's gossip and dial it directly. assert!( wait_for(Duration::from_secs(30), || { - node_c.connected_peers().contains(&pubkey(secrets[1])) + node_a.connected_peers().contains(&pubkey(secrets[1])) + && node_a.connected_peers().contains(&pubkey(secrets[2])) }) .await, - "C never connected to B via gossip; C sees {:?}", - node_c.connected_peers() + "A never connected to both B and C; A sees {:?}", + node_a.connected_peers() ); - assert!(node_b.connected_peers().contains(&pubkey(secrets[2]))); drop((node_a, node_b, node_c)); } #[tokio::test] -async fn non_accredited_node_is_never_a_connected_peer() { - let accredited_secrets = [[20; 32], [21; 32]]; - let outsider_secret = [22; 32]; - let accredited: HashSet<[u8; 32]> = accredited_secrets - .iter() - .map(|secret| pubkey(*secret)) - .collect(); - - let node_a = start_node(accredited_secrets[0], accredited.clone(), vec![]).await; +async fn transaction_submitted_to_one_node_reaches_others() { + let secrets = [[20; 32], [21; 32], [22; 32]]; + let (node_a, _mempool_a) = start_node(secrets[0], vec![]).await; let a_addr = node_a.listen_addrs()[0].to_string(); - let node_b = start_node( - accredited_secrets[1], - accredited.clone(), - vec![a_addr.clone()], - ) - .await; - // The outsider considers everyone accredited and announces eagerly — - // but its own key is in nobody's set. - let outsider = start_node(outsider_secret, accredited.clone(), vec![a_addr]).await; + let (node_b, mut mempool_b) = start_node(secrets[1], vec![a_addr.clone()]).await; + let (node_c, mut mempool_c) = start_node(secrets[2], vec![a_addr]).await; assert!( wait_for(Duration::from_secs(30), || { - node_a - .connected_peers() - .contains(&pubkey(accredited_secrets[1])) + node_a.connected_peers().contains(&pubkey(secrets[1])) + && node_a.connected_peers().contains(&pubkey(secrets[2])) }) .await, - "accredited pair should connect" + "A never connected to both B and C" ); - // Give the outsider ample time to announce, then confirm it was ignored. + + let tx = valid_transaction(); + let expected_hash = tx.hash(); + node_a.tx_publisher().publish(tx.clone()); + + assert!( + wait_for(Duration::from_secs(30), || { + mempool_b + .pop() + .is_some_and(|(_, received)| received.hash() == expected_hash) + }) + .await, + "B never received the gossiped transaction" + ); + assert!( + wait_for(Duration::from_secs(30), || { + mempool_c + .pop() + .is_some_and(|(_, received)| received.hash() == expected_hash) + }) + .await, + "C never received the gossiped transaction" + ); + drop((node_a, node_b, node_c)); +} + +#[tokio::test] +async fn invalid_transaction_is_not_propagated() { + // `TxPublisher::publish` only accepts a `LeeTransaction`, so genuinely + // undecodable bytes are not reachable through the public API; instead we + // publish a structurally well-formed transaction with an invalid + // signature, which still exercises the real gossip pipeline's rejection + // path (`evaluate_transaction`'s stateless check, then + // `MessageAcceptance::Reject`) end-to-end. + let secrets = [[30; 32], [31; 32]]; + let (node_a, _mempool_a) = start_node(secrets[0], vec![]).await; + let a_addr = node_a.listen_addrs()[0].to_string(); + let (node_b, mut mempool_b) = start_node(secrets[1], vec![a_addr]).await; + + assert!( + wait_for(Duration::from_secs(30), || { + node_a.connected_peers().contains(&pubkey(secrets[1])) + }) + .await, + "A never connected to B" + ); + + node_a + .tx_publisher() + .publish(invalidly_signed_transaction()); + tokio::time::sleep(Duration::from_secs(2)).await; assert!( - !node_a.connected_peers().contains(&pubkey(outsider_secret)), - "outsider must not appear as a connected accredited peer" + mempool_b.pop().is_none(), + "an invalidly-signed transaction must not reach the mempool" ); - assert!(!node_b.connected_peers().contains(&pubkey(outsider_secret))); - drop((node_a, node_b, outsider)); + drop((node_a, node_b)); } diff --git a/lez/sequencer/core/src/gossip/validation.rs b/lez/sequencer/core/src/gossip/validation.rs index bda681374..5eda6c3a5 100644 --- a/lez/sequencer/core/src/gossip/validation.rs +++ b/lez/sequencer/core/src/gossip/validation.rs @@ -1,231 +1,91 @@ -//! Pure decision function for inbound announcements: everything except the -//! directory upsert and the gossipsub report, so the whole pipeline is -//! testable without a swarm. +//! Pure decision function for an inbound gossiped transaction. +//! +//! The same stateless admission the RPC performs, minus mempool/seen-cache +//! side effects (those live in the drive task). Testable without a swarm. -use std::collections::HashSet; +use common::transaction::LeeTransaction; -use libp2p::{Multiaddr, PeerId}; - -use crate::gossip::{ - announcement::{RejectReason, SignedAnnouncement}, - directory::PeerDirectory, - network::peer_id_from_ed25519, -}; +/// Reserve ~200 bytes for block header overhead, mirroring the RPC check. +const BLOCK_HEADER_OVERHEAD: u64 = 200; #[derive(Debug)] -pub enum Evaluation { - /// Structurally invalid or forged: penalize the propagating peer. - Reject(RejectReason), - /// Signed by a key outside our (possibly stale) accredited set. - IgnoreUnknownKey, - /// Our own announcement echoed back (or replayed by another peer). - IgnoreOwn, - /// At or below the directory's stored seq for this key. - IgnoreStale, - /// Fresh and accredited: caller upserts the directory and dials. - Accept { - public_key: [u8; 32], - peer_id: PeerId, - listen_addrs: Vec, - seq: u64, - }, +pub enum TxEvaluation { + /// Structurally valid and authenticated; forward and admit. + Accept(LeeTransaction), + /// Malformed / forbidden; log the reason, penalize the peer. + Reject(String), + /// Reserved for the drive task's seen/mempool-full decisions. + Ignore, } +/// Decodes and stateless-checks a gossiped transaction the same way the RPC +/// admits a submitted one: size check, signature/witness check, then the +/// sequencer-only-program guard. #[must_use] -pub fn evaluate_announcement( - data: &[u8], - channel_id: &[u8; 32], - own_pubkey: &[u8; 32], - accredited: &HashSet<[u8; 32]>, - directory: &PeerDirectory, -) -> Evaluation { - let announcement = match SignedAnnouncement::decode_and_verify(data, channel_id) { - Ok(announcement) => announcement, - Err(reason) => return Evaluation::Reject(reason), +pub fn evaluate_transaction(data: &[u8], max_block_size: u64) -> TxEvaluation { + let tx: LeeTransaction = match borsh::from_slice(data) { + Ok(tx) => tx, + Err(err) => return TxEvaluation::Reject(format!("undecodable transaction: {err}")), }; - if &announcement.public_key == own_pubkey { - return Evaluation::IgnoreOwn; - } - if !accredited.contains(&announcement.public_key) { - return Evaluation::IgnoreUnknownKey; + let tx_size = u64::try_from(data.len()).unwrap_or(u64::MAX); + let max_tx_size = max_block_size.saturating_sub(BLOCK_HEADER_OVERHEAD); + if tx_size > max_tx_size { + return TxEvaluation::Reject(format!("transaction too large: {tx_size} > {max_tx_size}")); } - let listen_addrs: Vec = match announcement - .listen_addrs - .iter() - .map(|addr| addr.parse()) - .collect() + let authenticated = match tx.transaction_stateless_check() { + Ok(tx) => tx, + Err(err) => return TxEvaluation::Reject(format!("stateless check failed: {err:?}")), + }; + + if let LeeTransaction::Public(public_tx) = &authenticated + && crate::is_sequencer_only_program(public_tx.message().program_id) { - Ok(addrs) => addrs, - Err(_err) => return Evaluation::Reject(RejectReason::Undecodable), - }; - - // The key was verified as a valid curve point by `decode_and_verify`. - let Some(peer_id) = peer_id_from_ed25519(&announcement.public_key) else { - return Evaluation::Reject(RejectReason::BadSignature); - }; - - if directory - .seq_of(&announcement.public_key) - .is_some_and(|stored| stored >= announcement.seq) - { - return Evaluation::IgnoreStale; + return TxEvaluation::Reject("sequencer-only program".to_owned()); } - Evaluation::Accept { - public_key: announcement.public_key, - peer_id, - listen_addrs, - seq: announcement.seq, - } + TxEvaluation::Accept(authenticated) } #[cfg(test)] mod tests { - use std::collections::HashSet; - - use logos_blockchain_key_management_system_service::keys::Ed25519Key; + use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts}; use super::*; - use crate::gossip::announcement::{Announcement, RejectReason}; - const CHANNEL: [u8; 32] = [1; 32]; - const OWN: [u8; 32] = [0xAA; 32]; - - fn key() -> Ed25519Key { - Ed25519Key::from_bytes(&[7; 32]) - } - - fn bytes(key: &Ed25519Key, seq: u64) -> Vec { - Announcement { - channel_id: CHANNEL, - public_key: key.public_key().to_bytes(), - listen_addrs: vec!["/ip4/127.0.0.1/udp/7070/quic-v1".to_owned()], - seq, - } - .sign(key) - .to_bytes() - } - - fn accredited(key: &Ed25519Key) -> HashSet<[u8; 32]> { - HashSet::from([key.public_key().to_bytes()]) + fn valid_transaction() -> LeeTransaction { + let acc1 = initial_public_user_accounts()[0].account_id; + let acc2 = initial_public_user_accounts()[1].account_id; + let sign_key1 = initial_pub_accounts_private_keys()[0].pub_sign_key.clone(); + common::test_utils::create_transaction_native_token_transfer(acc1, 0, acc2, 10, &sign_key1) } #[test] - fn accredited_fresh_announcement_is_accepted() { - let key = key(); - let directory = PeerDirectory::default(); - let evaluation = evaluate_announcement( - &bytes(&key, 1), - &CHANNEL, - &OWN, - &accredited(&key), - &directory, - ); - let Evaluation::Accept { - public_key, - listen_addrs, - seq, - .. - } = evaluation - else { - panic!("expected Accept, got {evaluation:?}"); - }; - assert_eq!(public_key, key.public_key().to_bytes()); - assert_eq!(listen_addrs.len(), 1); - assert_eq!(seq, 1); - } - - #[test] - fn structural_failure_is_reject() { - let directory = PeerDirectory::default(); + fn well_formed_transaction_is_accepted() { + let tx = valid_transaction(); + let bytes = borsh::to_vec(&tx).unwrap(); assert!(matches!( - evaluate_announcement(b"junk", &CHANNEL, &OWN, &HashSet::new(), &directory), - Evaluation::Reject(RejectReason::Undecodable) + evaluate_transaction(&bytes, 1 << 20), + TxEvaluation::Accept(_) )); } #[test] - fn unknown_key_is_ignored_not_rejected() { - let key = key(); - let directory = PeerDirectory::default(); + fn garbage_bytes_are_rejected() { assert!(matches!( - evaluate_announcement(&bytes(&key, 1), &CHANNEL, &OWN, &HashSet::new(), &directory), - Evaluation::IgnoreUnknownKey + evaluate_transaction(&[0xff, 0xff, 0xff], 1 << 20), + TxEvaluation::Reject(_) )); } #[test] - fn own_echoed_announcement_is_ignored() { - let key = key(); - let own = key.public_key().to_bytes(); - let directory = PeerDirectory::default(); + fn oversize_transaction_is_rejected() { + let tx = valid_transaction(); + let bytes = borsh::to_vec(&tx).unwrap(); assert!(matches!( - evaluate_announcement( - &bytes(&key, 1), - &CHANNEL, - &own, - &accredited(&key), - &directory - ), - Evaluation::IgnoreOwn - )); - } - - #[test] - fn replayed_seq_is_stale() { - let key = key(); - let mut directory = PeerDirectory::default(); - let Evaluation::Accept { - public_key, - peer_id, - listen_addrs, - seq, - } = evaluate_announcement( - &bytes(&key, 5), - &CHANNEL, - &OWN, - &accredited(&key), - &directory, - ) - else { - panic!("expected Accept"); - }; - directory.upsert(public_key, peer_id, listen_addrs, seq); - assert!(matches!( - evaluate_announcement( - &bytes(&key, 5), - &CHANNEL, - &OWN, - &accredited(&key), - &directory - ), - Evaluation::IgnoreStale - )); - } - - #[test] - fn unparseable_multiaddr_is_reject() { - let key = key(); - let announcement_bytes = Announcement { - channel_id: CHANNEL, - public_key: key.public_key().to_bytes(), - listen_addrs: vec!["not a multiaddr".to_owned()], - seq: 1, - } - .sign(&key) - .to_bytes(); - let directory = PeerDirectory::default(); - assert!(matches!( - evaluate_announcement( - &announcement_bytes, - &CHANNEL, - &OWN, - &accredited(&key), - &directory - ), - Evaluation::Reject(RejectReason::Undecodable) + evaluate_transaction(&bytes, 1), + TxEvaluation::Reject(_) )); } } diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 1ae706795..c3f964ad3 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -79,6 +79,8 @@ pub enum TransactionOrigin { User, /// Transactions generated by the sequencer itself. Sequencer, + /// Transactions received via p2p gossip from a peer sequencer. + Gossip, } impl From for sequencer_core_metrics::TransactionOrigin { @@ -86,6 +88,7 @@ impl From for sequencer_core_metrics::TransactionOrigin { match origin { TransactionOrigin::User => Self::User, TransactionOrigin::Sequencer => Self::Sequencer, + TransactionOrigin::Gossip => Self::Gossip, } } } @@ -677,7 +680,9 @@ impl SequencerCore { ) -> bool { let tx_hash = tx.hash(); match origin { - TransactionOrigin::User => { + // Gossiped transactions arrive from untrusted peers, same as + // user-submitted ones, so they get the same full state validation. + TransactionOrigin::User | TransactionOrigin::Gossip => { let validated_diff = match tx.validate_on_state(state, block_height, timestamp) { Ok(diff) => diff, Err(err) => { diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index efbb979af..b35e3be39 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -115,36 +115,6 @@ impl BlockPublisherTrait for MockBlockPublisher { } } -/// No-op peer network: no peers, a token that never fires. -pub struct MockPeerNetwork { - cancellation: CancellationToken, -} - -impl MockPeerNetwork { - #[must_use] - pub fn new() -> Self { - Self { - cancellation: CancellationToken::new(), - } - } -} - -impl Default for MockPeerNetwork { - fn default() -> Self { - Self::new() - } -} - -impl crate::gossip::PeerNetworkTrait for MockPeerNetwork { - fn connected_peers(&self) -> Vec<[u8; 32]> { - Vec::new() - } - - fn driver_cancellation(&self) -> CancellationToken { - self.cancellation.clone() - } -} - /// The notes the mock reports as released by `withdrawals`. /// /// Zone-sdk picks the actual channel notes to release, so a mock has to invent diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 971877708..d3aa77827 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -255,7 +255,8 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result Result Result, listen_addr: SocketAddr, max_block_size: u64, + tx_publisher: Option, ) -> Result<(ServerHandle, SocketAddr)> { let server = jsonrpsee::server::ServerBuilder::with_config( jsonrpsee::server::ServerConfigBuilder::new() @@ -322,7 +328,8 @@ async fn run_server( info!("Starting Sequencer Service RPC server on {addr}"); - let service = service::SequencerService::new(sequencer, mempool_handle, max_block_size); + let service = + service::SequencerService::new(sequencer, mempool_handle, max_block_size, tx_publisher); let handle = server.start(service.into_rpc()); Ok((handle, addr)) } diff --git a/lez/sequencer/service/src/service.rs b/lez/sequencer/service/src/service.rs index e55735c07..a9682e66a 100644 --- a/lez/sequencer/service/src/service.rs +++ b/lez/sequencer/service/src/service.rs @@ -23,6 +23,7 @@ pub struct SequencerService { sequencer: Arc>>, mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, max_block_size: u64, + tx_publisher: Option, } impl SequencerService { @@ -30,11 +31,13 @@ impl SequencerService { sequencer: Arc>>, mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, max_block_size: u64, + tx_publisher: Option, ) -> Self { Self { sequencer, mempool_handle, max_block_size, + tx_publisher, } } } @@ -101,10 +104,14 @@ impl sequencer_service_rpc::Rpc error!("Transaction failed before reaching mempool: {err:#?}"); })?; + let for_gossip = authenticated_tx.clone(); self.mempool_handle .push((TransactionOrigin::User, authenticated_tx)) .await .expect("Mempool is closed, this is a bug"); + if let Some(publisher) = &self.tx_publisher { + publisher.publish(for_gossip); + } Ok(tx_hash) }