logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-08-26 03:41:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sequencer): gossip tx mempool, rm announce logic just use Kademlia

Browse Source
This commit is contained in:
erhant
2026-08-07 17:56:55 +03:00
parent 2110381fe2
commit 9f46ea5f1c
16 changed files with 534 additions and 980 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+2
View File
@@ -162,7 +162,9 @@ libp2p = { version = "0.55", features = [
"ed25519",
"gossipsub",
"identify",
"kad",
"macros",
"mdns",
"quic",
"tokio",
] }
lez/sequencer/core/Cargo.toml
+2
View File
@@ -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
lez/sequencer/core/metrics/src/record.rs
+1
View File
@@ -16,6 +16,7 @@ use crate::names;
pub enum TransactionOrigin {
User,
Sequencer,
Gossip,
}
#[derive(Debug, Clone, Copy, strum::IntoStaticStr, strum::EnumIter)]
lez/sequencer/core/src/config.rs
+1 -33
View File
@@ -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/<peer_id>`.
#[serde(default)]
pub bootstrap_peers: Vec<String>,
/// 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<SocketAddr> {
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]
lez/sequencer/core/src/gossip/keys_provider.rs → lez/sequencer/core/src/gossip/accreditation/mod.rs
+3 -2
View File
@@ -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 <https://github.com/logos-blockchain/logos-execution-zone/pull/653>.
lez/sequencer/core/src/gossip/announcement.rs
-208
View File
@@ -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<String>,
/// 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<u8> {
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<Announcement, RejectReason> {
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<String>) -> 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))
);
}
}
lez/sequencer/core/src/gossip/directory.rs
-128
View File
@@ -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<Multiaddr>,
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<Multiaddr>,
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<Item = (&[u8; 32], &PeerEntry)> {
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<u64> {
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]));
}
}
lez/sequencer/core/src/gossip/mod.rs
+5 -6
View File
@@ -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)]
lez/sequencer/core/src/gossip/network.rs
+219 -323
View File
@@ -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<kad::store::MemoryStore>,
#[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<Multiaddr>,
local_peer_id: PeerId,
tx_tx: mpsc::Sender<LeeTransaction>,
}
/// 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<LeeTransaction>);
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<P: AccreditedKeysProvider>(
/// 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<Self> {
// 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::<Result<_>>()?;
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::<LeeTransaction>(&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::<LeeTransaction>(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<GossipBehaviour>,
topic: gossipsub::IdentTopic,
channel_id: [u8; 32],
signing_key: Ed25519Key,
own_pubkey: [u8; 32],
announce_interval: Duration,
bootstrap: Vec<Multiaddr>,
directory: PeerDirectory,
connected: HashSet<PeerId>,
/// Per-peer dial backoff: (attempts, earliest next attempt).
dial_backoff: HashMap<PeerId, (u32, tokio::time::Instant)>,
/// Ed25519 public keys of peers seen via Identify, keyed by `PeerId`.
pubkeys: HashMap<PeerId, [u8; 32]>,
connected_tx: watch::Sender<Vec<[u8; 32]>>,
keys_rx: watch::Receiver<HashSet<[u8; 32]>>,
refresh_tx: mpsc::Sender<()>,
cancellation: CancellationToken,
topic: gossipsub::IdentTopic,
mempool: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
seen: SeenCache,
max_block_size: u64,
tx_rx: mpsc::Receiver<LeeTransaction>,
}
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<GossipBehaviourEvent>) {
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<String> = self
.swarm
.listeners()
.chain(self.swarm.external_addresses())
.filter(|addr| !is_unspecified(addr))
.map(ToString::to_string)
.collect::<std::collections::BTreeSet<_>>()
.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<Multiaddr>)> = 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<Item = [u8; 32]> + '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<PeerId> {
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<P: AccreditedKeysProvider>(
provider: P,
refresh_interval: Duration,
mut refresh_rx: mpsc::Receiver<()>,
keys_tx: watch::Sender<HashSet<[u8; 32]>>,
) {
/// 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();
lez/sequencer/core/src/gossip/seen_cache.rs
+116
View File
@@ -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<HashType>,
set: HashSet<HashType>,
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);
}
}
lez/sequencer/core/src/gossip/tests.rs
+110 -54
View File
@@ -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<String>,
) -> 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));
}
lez/sequencer/core/src/gossip/validation.rs
+53 -193
View File
@@ -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<Multiaddr>,
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<Multiaddr> = 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<u8> {
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(_)
));
}
}
lez/sequencer/core/src/lib.rs
+6 -1
View File
@@ -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<TransactionOrigin> for sequencer_core_metrics::TransactionOrigin {
@@ -86,6 +88,7 @@ impl From<TransactionOrigin> for sequencer_core_metrics::TransactionOrigin {
match origin {
TransactionOrigin::User => Self::User,
TransactionOrigin::Sequencer => Self::Sequencer,
TransactionOrigin::Gossip => Self::Gossip,
}
}
}
@@ -677,7 +680,9 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
) -> 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) => {
lez/sequencer/core/src/mock.rs
-30
View File
@@ -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
lez/sequencer/service/src/lib.rs
+9 -2
View File
@@ -255,7 +255,8 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
gossip_config,
channel_id,
secret,
sequencer_core::gossip::keys_provider::NodeKeysProvider::new(&bedrock_config),
mempool_handle.clone(),
max_block_size.as_u64(),
)
.await
.context("Failed to start sequencer gossip network")?;
@@ -263,6 +264,9 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
Some(network)
}
};
let tx_publisher = gossip_network
.as_ref()
.map(sequencer_core::gossip::Libp2pNetwork::tx_publisher);
let driver_cancellation = sequencer_core.block_publisher().driver_cancellation();
// Taken while the core is still owned here: once it is behind the `Arc`
@@ -278,6 +282,7 @@ pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<Seq
mempool_handle_for_server,
listen_addr,
max_block_size.as_u64(),
tx_publisher,
)
.await?;
info!("RPC server started");
@@ -303,6 +308,7 @@ async fn run_server(
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
listen_addr: SocketAddr,
max_block_size: u64,
tx_publisher: Option<sequencer_core::gossip::network::TxPublisher>,
) -> 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))
}
lez/sequencer/service/src/service.rs
+7
View File
@@ -23,6 +23,7 @@ pub struct SequencerService<BC: BlockPublisherTrait> {
sequencer: Arc<Mutex<SequencerCore<BC>>>,
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
max_block_size: u64,
tx_publisher: Option<sequencer_core::gossip::network::TxPublisher>,
}
impl<BC: BlockPublisherTrait> SequencerService<BC> {
@@ -30,11 +31,13 @@ impl<BC: BlockPublisherTrait> SequencerService<BC> {
sequencer: Arc<Mutex<SequencerCore<BC>>>,
mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
max_block_size: u64,
tx_publisher: Option<sequencer_core::gossip::network::TxPublisher>,
) -> Self {
Self {
sequencer,
mempool_handle,
max_block_size,
tx_publisher,
}
}
}
@@ -101,10 +104,14 @@ impl<BC: BlockPublisherTrait + Send + Sync + 'static> 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)
}