diff --git a/CHANGELOG.md b/CHANGELOG.md index 2036eb45..b7ccb100 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,17 @@ stopped at v0.4 until 2026-09-30, when the entries v0.5 to v0.6.7 were written f ## Unreleased -Nothing yet. +- **Transactions gossip anonymously** (issue #146): a second gossipsub behaviour, + `MessageAuthenticity::Anonymous` under `ValidationMode::Anonymous`, protocol + `/rand//txsub`, topic `rand//tx/2`; consensus, status and peer bindings stay + signed. Rolls node by node with the legacy signed topic still published and relayed; + `--anonymous-tx-only` drops it once no older node is left (`docs/deploy.md`). +- **No duplicate-publish WARN spam** (issue #144): a status's gossip id covers its author and + sequence number, so an unchanged status re-announced on a connect or the 3 s tick goes out + instead of being dropped as "already published"; `PublishError::Duplicate` is a debug line. +- **No ANSI colour in journald** (issue #143): `rand-node` and `rand-prover` colour their log + only at a terminal and never under `NO_COLOR`; `deploy/update-droplet.sh` drops "chain 14" + from an old unit's `Description=`. ## v0.8.2 — 2026-10-27 (chain 21, node-only; the genesis-gated rules wait for the next cut) diff --git a/crates/randprotocol-node/src/main.rs b/crates/randprotocol-node/src/main.rs index 2a0fa85d..368792e8 100644 --- a/crates/randprotocol-node/src/main.rs +++ b/crates/randprotocol-node/src/main.rs @@ -759,6 +759,12 @@ enum Cmd { /// exchange honest gossip both ways. #[arg(long)] strict_gossip: bool, + /// Gossip transactions only on the anonymous lane `rand/{chain}/tx/2` (issue #146): stop + /// subscribing to and publishing on the legacy signed topic `rand/{chain}/tx`, whose + /// messages name the node that first published them. Only once every node on the chain + /// runs v0.8.3 or later: an older node hears transactions on the legacy topic alone. + #[arg(long)] + anonymous_tx_only: bool, #[arg(long, default_value = "127.0.0.1:8545")] rpc: SocketAddr, /// Take part in consensus with this node's key. A key in no current epoch's validator @@ -2490,10 +2496,34 @@ fn opens_storage(cmd: &Cmd) -> bool { matches!(cmd, Cmd::Run { .. } | Cmd::Verify { .. } | Cmd::Db { .. } | Cmd::Init { .. } | Cmd::Safety { .. }) } +/// Colour in the log only for a person at a terminal (issue #143): under systemd stdout is a +/// pipe to journald, and escapes there break `grep`, log shippers and every parser downstream. +/// `NO_COLOR` turns it off anywhere (: present and not empty). +fn log_ansi(is_terminal: bool, no_color: Option) -> bool { + is_terminal && no_color.is_none_or(|v| v.is_empty()) +} + +#[cfg(test)] +mod log_ansi_tests { + use super::log_ansi; + + #[test] + fn colour_only_at_a_terminal_and_never_under_no_color() { + assert!(log_ansi(true, None), "a terminal with no NO_COLOR gets colour"); + assert!(!log_ansi(false, None), "a pipe (journald) gets none"); + assert!(!log_ansi(true, Some("1".into())), "NO_COLOR=1 turns it off at a terminal"); + assert!(!log_ansi(false, Some("1".into()))); + assert!(log_ansi(true, Some("".into())), "an empty NO_COLOR is unset (no-color.org)"); + } +} + #[tokio::main] async fn main() -> Result<()> { + use std::io::IsTerminal; + // `fmt()` writes to stdout, so stdout is the stream whose terminal-ness counts. tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info,libp2p=warn,libp2p_mdns=off".into())) + .with_ansi(log_ansi(std::io::stdout().is_terminal(), std::env::var_os("NO_COLOR"))) .init(); let cli = Cli::parse(); let test_profile_allowed = test_fri_profile_allowed(&cli); @@ -2976,6 +3006,7 @@ async fn main() -> Result<()> { bootstrap, reserved_peer, strict_gossip, + anonymous_tx_only, rpc, validator, no_mdns, @@ -3068,7 +3099,7 @@ async fn main() -> Result<()> { prune_history, gas_policy: randprotocol_core::gas::GasPolicy::from_prices(gas_price, byte_price), msauth: randprotocol_node::msauth_key::MsauthKeyOptions { path: msauth_vk, build: build_msauth_key }, - }, rpc_options, node::NetOptions { reserved_peers: reserved_peer, strict_gossip }) + }, rpc_options, node::NetOptions { reserved_peers: reserved_peer, strict_gossip, anonymous_tx_only }) .await?; // The prover is served once the node's RPC is up and stops with the node; a prover // that exits stops the node too. diff --git a/crates/randprotocol-node/src/network/behaviour.rs b/crates/randprotocol-node/src/network/behaviour.rs index f2be26b1..7182a769 100644 --- a/crates/randprotocol-node/src/network/behaviour.rs +++ b/crates/randprotocol-node/src/network/behaviour.rs @@ -16,7 +16,17 @@ pub struct RandBehaviour { /// Second, for the same reason: the per-source pending cap and the per-peer cap that holds /// reserved peers too (audit v6, NET-1; [`super::edge`]). It emits nothing either. pub guard: super::edge::Guard, + /// Consensus, status, peer bindings and — while [`super::EdgeConfig::anonymous_tx_only`] is + /// off — the legacy transaction topic: `MessageAuthenticity::Signed`, on gossipsub's own + /// `/meshsub` protocol. pub gossipsub: gossipsub::Behaviour, + /// Transactions only (issue #146): `MessageAuthenticity::Anonymous` under + /// `ValidationMode::Anonymous`, so a transaction carries no author, signature or sequence + /// number, and one that does is refused. `MessageAuthenticity` is per behaviour, hence a + /// second one, on a protocol of its own (`/rand/{chain}/txsub`) so the two never compete for + /// one stream. Its events arrive as `RandEvent::Gossipsub` like the first's; the topic tells + /// them apart, because each behaviour subscribes only its own. + pub tx_gossip: gossipsub::Behaviour, pub identify: identify::Behaviour, pub kademlia: kad::Behaviour, pub mdns: Toggle, diff --git a/crates/randprotocol-node/src/network/mod.rs b/crates/randprotocol-node/src/network/mod.rs index 6e1cbfe2..41f8ca6f 100644 --- a/crates/randprotocol-node/src/network/mod.rs +++ b/crates/randprotocol-node/src/network/mod.rs @@ -368,6 +368,14 @@ pub struct EdgeConfig { /// strict node and a permissive one exchange honest messages both ways (pinned by /// `a_strict_node_takes_a_permissive_nodes_signed_gossip_and_drops_a_forged_unsigned_one`). pub strict_gossip: bool, + /// Gossip transactions on the anonymous lane only (issue #146, `--anonymous-tx-only`): no + /// subscription to the legacy signed topic `rand/{chain}/tx`, and nothing published there. Off + /// by default for the rolling upgrade: a node then publishes each transaction on both lanes — + /// the anonymous `rand/{chain}/tx/2` and, for nodes that predate it, the signed legacy topic, + /// which still names this node as the author — and relays on both. Turn it on once no node + /// that predates the anonymous lane is left; until then a transaction published only + /// anonymously never reaches them by gossip. + pub anonymous_tx_only: bool, } #[derive(Clone, Debug, Serialize)] @@ -390,6 +398,18 @@ pub struct PeerInfo { pub struct GossipId { pub message_id: MessageId, pub propagation_source: PeerId, + /// The gossipsub behaviour that delivered it, which is the one its verdict goes back to. + pub lane: GossipLane, +} + +/// Which gossipsub behaviour carried a message (issue #146). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum GossipLane { + /// `MessageAuthenticity::Signed`: consensus, status, peer bindings, legacy transactions. + #[default] + Signed, + /// The anonymous transaction lane, `rand/{chain}/tx/2`. + AnonymousTx, } #[allow(clippy::large_enum_variant)] // moved once, never stored in bulk: boxing buys nothing @@ -499,6 +519,9 @@ struct Topics { /// predates it is not subscribed, so it never receives a `GossipMessage` variant it cannot /// decode — a mixed fleet sees no change on the three topics it shares. peers: IdentTopic, + /// Transactions on the anonymous lane (issue #146), the only topic `tx_gossip` subscribes. + /// `tx` above is the legacy signed topic, kept through the rolling upgrade. + tx_anon: IdentTopic, } impl Topics { @@ -508,9 +531,25 @@ impl Topics { tx: IdentTopic::new(format!("rand/{chain_id}/tx")), status: IdentTopic::new(format!("rand/{chain_id}/status")), peers: IdentTopic::new(format!("rand/{chain_id}/peers")), + tx_anon: IdentTopic::new(format!("rand/{chain_id}/tx/2")), + } + } + + /// The lane a delivery came in on: its topic's. Each behaviour subscribes only its own topics + /// and gossipsub delivers nothing on a topic its behaviour has not subscribed. + fn lane_of(&self, topic: &gossipsub::TopicHash) -> GossipLane { + if *topic == self.tx_anon.hash() { + GossipLane::AnonymousTx + } else { + GossipLane::Signed } } + /// Either transaction topic. + fn is_tx(&self, topic: &gossipsub::TopicHash) -> bool { + *topic == self.tx.hash() || *topic == self.tx_anon.hash() + } + fn for_message(&self, msg: &GossipMessage) -> &IdentTopic { match msg { GossipMessage::Consensus(_) => &self.consensus, @@ -546,24 +585,17 @@ pub async fn start_with( let keypair = identity::Keypair::ed25519_from_bytes(&mut seed)?; let local_peer_id = PeerId::from(keypair.public()); - let gossipsub_config = gossipsub::ConfigBuilder::default() - .heartbeat_interval(Duration::from_millis(500)) - .validation_mode(if edge.strict_gossip { ValidationMode::Strict } else { ValidationMode::Permissive }) - .max_transmit_size(cfg.limits.gossip_max_transmit_size) - .message_id_fn(|m: &gossipsub::Message| MessageId::from(blake3::hash(&m.data).as_bytes().to_vec())) - // Application-level validation: this node forwards a transaction only after it has - // verified here, off the consensus loop. Local to this node — the wire is unchanged, so it - // rolls out onto a mixed fleet by ordinary restart. `ValidationMode` is `Permissive` - // unless `--strict-gossip` ([`EdgeConfig::strict_gossip`]): it too governs only what this - // node accepts, and since every node signs what it publishes, a strict node drops nothing - // an honest permissive one sends (tested, audit v6 CH-7). The cost of the switch is that - // every delivered message must be reported exactly once (see [`GossipId`]) or this node - // stops forwarding it. - .validate_messages() - .build() - .map_err(|e| anyhow::anyhow!("gossipsub config: {e}"))?; + let topics = Topics::new(cfg.chain_id); + let signed_mode = if edge.strict_gossip { ValidationMode::Strict } else { ValidationMode::Permissive }; + let gossipsub_config = gossip_config(&cfg.limits, signed_mode, None, topics.status.hash())?; let gossipsub = gossipsub::Behaviour::new(MessageAuthenticity::Signed(keypair.clone()), gossipsub_config) .map_err(|e| anyhow::anyhow!("gossipsub: {e}"))?; + // The anonymous transaction lane (issue #146). `ValidationMode::Anonymous` refuses a message + // that carries a source, sequence number or signature, so no relay on this lane can attach + // an author either. `--strict-gossip` does not apply: there is no author to check. + let tx_config = gossip_config(&cfg.limits, ValidationMode::Anonymous, Some(tx_protocol_prefix(cfg.chain_id)), topics.status.hash())?; + let tx_gossip = gossipsub::Behaviour::new(MessageAuthenticity::Anonymous, tx_config) + .map_err(|e| anyhow::anyhow!("tx gossipsub: {e}"))?; let identify = identify::Behaviour::new(identify::Config::new( format!("rand/{}/1", cfg.chain_id), @@ -625,7 +657,7 @@ pub async fn start_with( guard.reserve_at(id, [a], edge::SourceKind::Configured); } } - let behaviour = RandBehaviour { limits, guard, gossipsub, identify, kademlia, mdns: Toggle::from(mdns), sync, ping }; + let behaviour = RandBehaviour { limits, guard, gossipsub, tx_gossip, identify, kademlia, mdns: Toggle::from(mdns), sync, ping }; let mut swarm = libp2p::SwarmBuilder::with_existing_identity(keypair) .with_tokio() @@ -637,20 +669,82 @@ pub async fn start_with( .with_connection_timeout(HANDSHAKE_TIMEOUT) .build(); - let topics = Topics::new(cfg.chain_id); - for t in [&topics.consensus, &topics.tx, &topics.status, &topics.peers] { + for t in [&topics.consensus, &topics.status, &topics.peers] { swarm.behaviour_mut().gossipsub.subscribe(t)?; } + if !edge.anonymous_tx_only { + swarm.behaviour_mut().gossipsub.subscribe(&topics.tx)?; + } + swarm.behaviour_mut().tx_gossip.subscribe(&topics.tx_anon)?; for addr in &cfg.listen { swarm.listen_on(addr.clone())?; } let (cmd_tx, cmd_rx) = mpsc::channel(4096); let (evt_tx, evt_rx) = mpsc::channel(4096); - tokio::spawn(run(swarm, cfg, pinned, topics, cmd_rx, evt_tx)); + tokio::spawn(run(swarm, cfg, pinned, topics, !edge.anonymous_tx_only, cmd_rx, evt_tx)); Ok((NetworkHandle { cmd: cmd_tx, local_peer_id }, evt_rx)) } +/// One gossipsub behaviour's configuration. Both lanes share everything but the validation mode +/// and, for the transaction lane, the protocol (`protocol_prefix`, see [`tx_protocol_prefix`]). +fn gossip_config( + limits: &WireLimits, + mode: ValidationMode, + protocol_prefix: Option, + status_topic: gossipsub::TopicHash, +) -> anyhow::Result { + let mut b = gossipsub::ConfigBuilder::default(); + b.heartbeat_interval(Duration::from_millis(500)) + .validation_mode(mode) + .max_transmit_size(limits.gossip_max_transmit_size) + .message_id_fn(move |m: &gossipsub::Message| message_id(m, &status_topic)) + // Application-level validation: this node forwards a transaction only after it has + // verified here, off the consensus loop. Local to this node — the wire is unchanged, so it + // rolls out onto a mixed fleet by ordinary restart. The signed lane's `ValidationMode` is + // `Permissive` unless `--strict-gossip` ([`EdgeConfig::strict_gossip`]): it too governs + // only what this node accepts, and since every node signs what it publishes there, a + // strict node drops nothing an honest permissive one sends (tested, audit v6 CH-7). The + // cost of the switch is that every delivered message must be reported exactly once (see + // [`GossipId`]) or this node stops forwarding it. + .validate_messages(); + if let Some(prefix) = protocol_prefix { + b.protocol_id_prefix(prefix); + } + b.build().map_err(|e| anyhow::anyhow!("gossipsub config: {e}")) +} + +/// The anonymous transaction lane's gossipsub protocol (issue #146): its own, so its streams never +/// meet the signed behaviour's `/meshsub`. A node that predates the lane does not speak it; its +/// connections carry the signed lane only, and gossipsub marks the peer as not supporting this +/// one — nothing is closed. +fn tx_protocol_prefix(chain_id: u64) -> String { + format!("/rand/{chain_id}/txsub") +} + +/// The gossip message id: blake3 of the bytes, so one transaction or vote forwarded by two peers +/// is one message — except on the status topic, where the author and the sequence number come +/// first (issue #144). A `Status` names no author, so two nodes at one head publish identical +/// bytes, and under a content-only id the second one's status was dropped as a duplicate, as was +/// every re-publish of an unchanged status (on each `peer connected` and on the 3 s tick): +/// gossipsub's "Not publishing a message that has already been published" WARN. Every published +/// status is now its own message. The id is local to each node except in IHAVE/IWANT, where a +/// pre-#144 peer computes another id for a status, so at worst it pulls a status it already holds +/// — a few dozen bytes, and only while the fleet is mixed. An unsigned status (no source or +/// sequence number, which only `Permissive` admits) keeps the content id. +fn message_id(m: &gossipsub::Message, status_topic: &gossipsub::TopicHash) -> MessageId { + let mut h = blake3::Hasher::new(); + if m.topic == *status_topic { + if let (Some(source), Some(seq)) = (m.source, m.sequence_number) { + h.update(b"rand-status-id-1"); + h.update(&source.to_bytes()); + h.update(&seq.to_be_bytes()); + } + } + h.update(&m.data); + MessageId::from(h.finalize().as_bytes().to_vec()) +} + /// Hand one verdict to gossipsub. The only place that calls /// `report_message_validation_result`, so the "the message is gone" case is decided once. /// @@ -663,12 +757,113 @@ pub async fn start_with( /// shedding gossip while it catches up would otherwise fill the log with warnings about working /// normally. fn report_to_gossipsub(swarm: &mut Swarm, id: &GossipId, acceptance: MessageAcceptance) { - if !swarm.behaviour_mut().gossipsub.report_message_validation_result(&id.message_id, &id.propagation_source, acceptance) - { + let b = swarm.behaviour_mut(); + let lane = match id.lane { + GossipLane::Signed => &mut b.gossipsub, + GossipLane::AnonymousTx => &mut b.tx_gossip, + }; + if !lane.report_message_validation_result(&id.message_id, &id.propagation_source, acceptance) { tracing::debug!(?id, "validation reported for a message no longer in the cache"); } } +/// [`report_to_gossipsub`], and the same verdict for the other lane's copies of one transaction +/// that were held for it ([`TxCross`]). +fn report(swarm: &mut Swarm, cross: &mut TxCross, id: &GossipId, acceptance: MessageAcceptance) { + report_to_gossipsub(swarm, id, acceptance); + for held in cross.on_verdict(id, acceptance) { + report_to_gossipsub(swarm, &held, acceptance); + } +} + +/// How long [`TxCross`] remembers a transaction: past gossipsub's own window for a verdict +/// (`history_length` 5 × the 500 ms heartbeat), after which a report is a no-op anyway. +const TX_CROSS_TTL: Duration = Duration::from_secs(10); +/// The most transactions [`TxCross`] tracks at once. Past it a copy is not tracked, and the +/// other lane's copy reaches the node as an ordinary duplicate. +const TX_CROSS_MAX: usize = 1 << 16; + +/// One transaction heard on both lanes during the rolling upgrade (issue #146). The node judges +/// one delivery of a transaction and answers any later copy as a duplicate — `Ignore`, not +/// forwarded — so a transaction first heard anonymously would never be relayed on the legacy +/// topic, and nodes that predate the anonymous lane would hear it only from a direct neighbour of +/// its author. So the network task hands the node the first copy only, and reports the verdict +/// the node gives it for every copy the other lane delivered: a transaction this node accepts is +/// relayed on both lanes. Both lanes' message ids are the content's (see [`message_id`]), so one +/// transaction is one id on either. Unused under [`EdgeConfig::anonymous_tx_only`], which has one +/// lane. +#[derive(Default)] +struct TxCross { + seen: HashMap, +} + +struct CrossEntry { + first: GossipLane, + at: Instant, + verdict: Option, + held: Vec, +} + +#[derive(Debug, PartialEq, Eq)] +enum CrossDelivery { + /// Hand it to the node, which reports it. + First, + /// The other lane's copy was judged: report this. + Answer(MessageAcceptance), + /// The other lane's copy is still being judged: its verdict is reported for this one too. + Held, +} + +impl TxCross { + fn on_delivery(&mut self, id: &GossipId, now: Instant) -> CrossDelivery { + if let Some(e) = self.seen.get_mut(&id.message_id) { + if e.first == id.lane { + // Not a second lane: gossipsub delivered one id on one lane twice, which its own + // duplicate cache (longer than our TTL) rules out. The node judges it as it would. + return CrossDelivery::First; + } + return match e.verdict { + Some(v) => CrossDelivery::Answer(v), + None => { + e.held.push(id.clone()); + CrossDelivery::Held + } + }; + } + if self.seen.len() >= TX_CROSS_MAX { + return CrossDelivery::First; + } + self.seen.insert(id.message_id.clone(), CrossEntry { first: id.lane, at: now, verdict: None, held: Vec::new() }); + CrossDelivery::First + } + + /// The verdict on a first copy: every copy held for it, to report the same. + fn on_verdict(&mut self, id: &GossipId, acceptance: MessageAcceptance) -> Vec { + match self.seen.get_mut(&id.message_id) { + Some(e) if e.first == id.lane && e.verdict.is_none() => { + e.verdict = Some(acceptance); + std::mem::take(&mut e.held) + } + _ => Vec::new(), + } + } + + /// Forget what is older than [`TX_CROSS_TTL`]. Copies still held then (a verdict that never + /// came, which every delivery's report-once rule rules out) are returned to be reported + /// `Ignore`, so none is left unreported. + fn expire(&mut self, now: Instant) -> Vec { + let mut orphans = Vec::new(); + self.seen.retain(|_, e| { + let keep = now.duration_since(e.at) < TX_CROSS_TTL; + if !keep { + orphans.append(&mut e.held); + } + keep + }); + orphans + } +} + fn peer_id_of(addr: &Multiaddr) -> Option { addr.iter().find_map(|p| match p { libp2p::multiaddr::Protocol::P2p(id) => Some(id), @@ -726,6 +921,8 @@ async fn run( // Reserved for the life of the process: no `Unreserve` removes one (see [`EdgeConfig`]). pinned: HashSet, topics: Topics, + // Publish and relay transactions on the legacy signed topic too ([`EdgeConfig::anonymous_tx_only`] off). + legacy_tx: bool, mut cmd_rx: mpsc::Receiver, evt_tx: mpsc::Sender, ) { @@ -738,6 +935,10 @@ async fn run( let mut lan_peers: HashSet = HashSet::new(); // What each forwarder may deliver before its frames are decoded (GOSSIP-1). let mut frames = FrameMeter::new(cfg.limits.gossip_max_transmit_size); + // One transaction heard on both lanes (issue #146): only while there are two. + let mut cross = TxCross::default(); + let mut cross_tick = tokio::time::interval(TX_CROSS_TTL / 2); + cross_tick.tick().await; dial_bootstrap(&mut swarm, &cfg, &peers); if !cfg.bootstrap.is_empty() { let _ = swarm.behaviour_mut().kademlia.bootstrap(); @@ -753,6 +954,11 @@ async fn run( dial_bootstrap(&mut swarm, &cfg, &peers); redial_known(&mut swarm, &known, &peers); } + _ = cross_tick.tick() => { + for orphan in cross.expire(Instant::now()) { + report_to_gossipsub(&mut swarm, &orphan, MessageAcceptance::Ignore); + } + } _ = kad_bootstrap.tick() => { if let Err(e) = swarm.behaviour_mut().kademlia.bootstrap() { tracing::debug!("kademlia bootstrap: {e}"); @@ -761,23 +967,7 @@ async fn run( cmd = cmd_rx.recv() => { let Some(cmd) = cmd else { break }; match cmd { - NetworkCommand::Broadcast(msg) => { - let topic = topics.for_message(&msg).clone(); - match bincode::serialize(&msg) { - Ok(data) => { - if let Err(e) = swarm.behaviour_mut().gossipsub.publish(topic, data) { - match e { - gossipsub::PublishError::NoPeersSubscribedToTopic => { - tracing::debug!("publish: no peers yet") - } - gossipsub::PublishError::Duplicate => {} - other => tracing::warn!("publish failed: {other}"), - } - } - } - Err(e) => tracing::error!("encode gossip: {e}"), - } - } + NetworkCommand::Broadcast(msg) => publish(&mut swarm, &topics, legacy_tx, &msg), NetworkCommand::SendSyncRequest { peer, request, reply } => { let id = swarm.behaviour_mut().sync.send_request(&peer, request); let _ = reply.send(id); @@ -808,7 +998,7 @@ async fn run( let _ = reply.send(list); } NetworkCommand::ReportValidation { id, acceptance } => { - report_to_gossipsub(&mut swarm, &id, acceptance); + report(&mut swarm, &mut cross, &id, acceptance); } NetworkCommand::Reserve(id) => { let b = swarm.behaviour_mut(); @@ -830,22 +1020,63 @@ async fn run( } } event = swarm.select_next_some() => { - handle_swarm_event(event, &mut swarm, &mut peers, &mut known, &mut lan_peers, &mut frames, &evt_tx).await; + let mut gossip = GossipState { topics: &topics, frames: &mut frames, cross: &mut cross, legacy_tx }; + handle_swarm_event(event, &mut swarm, &mut peers, &mut known, &mut lan_peers, &mut gossip, &evt_tx).await; } } } tracing::info!("network task stopped"); } +/// Publish one message: a transaction on the anonymous lane, and on the legacy signed topic too +/// while `legacy_tx` (issue #146); everything else on the signed lane, on its own topic. +fn publish(swarm: &mut Swarm, topics: &Topics, legacy_tx: bool, msg: &GossipMessage) { + let data = match bincode::serialize(msg) { + Ok(data) => data, + Err(e) => return tracing::error!("encode gossip: {e}"), + }; + let b = swarm.behaviour_mut(); + if matches!(msg, GossipMessage::Transaction(_)) { + publish_on(&mut b.tx_gossip, &topics.tx_anon, data.clone()); + if legacy_tx { + publish_on(&mut b.gossipsub, &topics.tx, data); + } + } else { + publish_on(&mut b.gossipsub, topics.for_message(msg), data); + } +} + +fn publish_on(lane: &mut gossipsub::Behaviour, topic: &IdentTopic, data: Vec) { + if let Err(e) = lane.publish(topic.clone(), data) { + match e { + gossipsub::PublishError::NoPeersSubscribedToTopic => tracing::debug!(%topic, "publish: no peers yet"), + // gossipsub has already logged its own WARN by now; nothing this node publishes on + // purpose should get here since #144 (see `message_id`). + gossipsub::PublishError::Duplicate => tracing::debug!(%topic, "publish: already published"), + other => tracing::warn!(%topic, "publish failed: {other}"), + } + } +} + +/// What the gossip arm of [`handle_swarm_event`] reads and keeps between events. +struct GossipState<'a> { + topics: &'a Topics, + /// What each forwarder may deliver before its frames are decoded (GOSSIP-1). + frames: &'a mut FrameMeter, + cross: &'a mut TxCross, + legacy_tx: bool, +} + async fn handle_swarm_event( event: SwarmEvent, swarm: &mut Swarm, peers: &mut HashMap, known: &mut HashMap>, lan_peers: &mut HashSet, - frames: &mut FrameMeter, + gossip: &mut GossipState<'_>, evt_tx: &mpsc::Sender, ) { + let frames = &mut *gossip.frames; match event { SwarmEvent::NewListenAddr { address, .. } => { tracing::info!(%address, "listening"); @@ -886,24 +1117,47 @@ async fn handle_swarm_event( message, message_id, })) => { + let lane = gossip.topics.lane_of(&message.topic); + let is_tx = gossip.topics.is_tx(&message.topic); + let id = GossipId { message_id, propagation_source, lane }; + // A transaction's second lane, during the rolling upgrade (issue #146): answered with + // the first copy's verdict, never handed to the node a second time. + if is_tx && gossip.legacy_tx { + match gossip.cross.on_delivery(&id, Instant::now()) { + CrossDelivery::First => {} + CrossDelivery::Answer(a) => return report_to_gossipsub(swarm, &id, a), + CrossDelivery::Held => return, + } + } // Metered by length before anything reads the bytes (audit v6, GOSSIP-1): up to a // transmit size's worth of bincode was decoded, and a transaction hashed, before any // per-peer meter ran. Over the budget: `Ignore` — not relayed, no one penalised, and // reported here because the node loop never sees it. if !frames.admit(propagation_source, message.data.len(), Instant::now()) { tracing::debug!(%propagation_source, bytes = message.data.len(), "gossip frame over the forwarder's byte budget; not decoded"); - report_to_gossipsub(swarm, &GossipId { message_id, propagation_source }, MessageAcceptance::Ignore); + report(swarm, gossip.cross, &id, MessageAcceptance::Ignore); return; } match bincode::deserialize::(&message.data) { + // The anonymous lane carries transactions and nothing else. + Ok(msg) if lane == GossipLane::AnonymousTx && !matches!(msg, GossipMessage::Transaction(_)) => { + tracing::debug!(%propagation_source, "a non-transaction on the anonymous transaction lane"); + report(swarm, gossip.cross, &id, MessageAcceptance::Reject); + } Ok(msg) => { - // `from` falls back to `propagation_source` for a message that carries no - // source, which is the only case where the two coincide by construction. The - // rate limit still keys on `propagation_source`. Under `Permissive` an unsigned - // message's `source` is only claimed, never proven: the node reads a `Status` - // only when the two agree (SYNC-1, `node::on_status_gossip`). - let from = message.source.unwrap_or(propagation_source); - let id = GossipId { message_id, propagation_source }; + // A transaction's author is nobody's business (issue #146): `from` is the + // forwarder, on either lane, and the anonymous lane carries no source at all. + // Otherwise `from` falls back to `propagation_source` for a message that + // carries no source, which is the only case where the two coincide by + // construction. The rate limit keys on `propagation_source` either way. Under + // `Permissive` an unsigned message's `source` is only claimed, never proven: + // the node reads a `Status` only when the two agree (SYNC-1, + // `node::on_status_gossip`). + let from = if matches!(msg, GossipMessage::Transaction(_)) { + propagation_source + } else { + message.source.unwrap_or(propagation_source) + }; let _ = evt_tx.send(NetworkEvent::Gossip { from, msg, id }).await; } Err(e) => { @@ -912,8 +1166,7 @@ async fn handle_swarm_event( // an unreported delivery sits in gossipsub's cache and this node stops // forwarding it for everyone. Reject rather than Ignore — bytes that are not a // `GossipMessage` at all are this peer's fault. - let id = GossipId { message_id, propagation_source }; - report_to_gossipsub(swarm, &id, MessageAcceptance::Reject); + report(swarm, gossip.cross, &id, MessageAcceptance::Reject); } } } @@ -1304,6 +1557,294 @@ mod tests { b.shutdown().await; } + /// Issue #144: a `Status` names no author, so under a content-only message id every node at + /// one head published the *same* id, the second publisher's status was dropped as a duplicate + /// (gossipsub's "Not publishing a message that has already been published" WARN), and so was + /// every re-publish of an unchanged status — on each `peer connected` and on the 3 s tick. A + /// status's id now also covers its author and sequence number; every other topic keeps the + /// content-addressed id, so one transaction forwarded by two peers is still one message. + #[test] + fn a_status_id_covers_its_author_and_sequence_and_every_other_id_is_the_content() { + let topics = Topics::new(7); + let data = bincode::serialize(&GossipMessage::Status(Status { height: 9, head_hash: Hash::digest(b"h"), view: 3, floor: 0 })).unwrap(); + let msg = |topic: &IdentTopic, source: Option, seq: Option| gossipsub::Message { + source, + data: data.clone(), + sequence_number: seq, + topic: topic.hash(), + }; + let (p, q) = (peer_of_seed(1), peer_of_seed(2)); + let id = |m: &gossipsub::Message| message_id(m, &topics.status.hash()); + // Status: two authors at one head, and one author publishing twice, are different messages. + assert_ne!(id(&msg(&topics.status, Some(p), Some(1))), id(&msg(&topics.status, Some(q), Some(1)))); + assert_ne!(id(&msg(&topics.status, Some(p), Some(1))), id(&msg(&topics.status, Some(p), Some(2)))); + assert_eq!(id(&msg(&topics.status, Some(p), Some(1))), id(&msg(&topics.status, Some(p), Some(1))), "a relay of one publish is one message"); + // An unsigned status (no source or sequence) falls back to the content. + assert_eq!(id(&msg(&topics.status, None, None)), MessageId::from(blake3::hash(&data).as_bytes().to_vec())); + // Every other topic: the content alone, whoever sent it (and the pinned blake3 form, so a + // mixed fleet's IHAVE/IWANT ids agree on them). + for t in [&topics.tx, &topics.consensus, &topics.peers] { + let content = MessageId::from(blake3::hash(&data).as_bytes().to_vec()); + assert_eq!(id(&msg(t, Some(p), Some(1))), content, "{t}"); + assert_eq!(id(&msg(t, Some(q), Some(9))), content, "{t}"); + } + } + + /// Issue #144, end to end: B publishes the same status twice and A receives it twice, under + /// two ids. Before, the second publish was gossipsub's duplicate — dropped with a WARN, and + /// the new peer a re-announcement on connect was meant for never heard it. + #[tokio::test] + async fn an_unchanged_status_published_again_is_delivered_again() { + let (a, mut a_rx, b, _b_rx) = two_connected_nodes().await; + let status = Status { height: 9, head_hash: Hash::digest(b"same"), view: 3, floor: 0 }; + let (_, first) = gossip_status_to(&b, &mut a_rx, status.clone()).await; + a.report_validation(first.clone(), MessageAcceptance::Accept).await; + b.broadcast(GossipMessage::Status(status.clone())).await; + let again = wait_for(&mut a_rx, Duration::from_secs(10), |e| match e { + NetworkEvent::Gossip { msg: GossipMessage::Status(s), id, .. } if s.head_hash == status.head_hash && id.message_id != first.message_id => Some(id), + _ => None, + }) + .await + .expect("the same status, published again, was not delivered again"); + assert_eq!(again.propagation_source, b.local_peer_id); + a.report_validation(again, MessageAcceptance::Accept).await; + a.shutdown().await; + b.shutdown().await; + } + + // ------------------------------------------- issue #146: the anonymous transaction lane + + fn tx_of(n: u64) -> randprotocol_core::Transaction { + // `chain_id` only makes the bytes differ: the network task never judges a transaction. + randprotocol_core::Transaction { chain_id: n, bundle: None, action: randprotocol_core::Action::None } + } + + /// A bare gossipsub swarm standing in for another node. `tx_lane`: a listener on the + /// anonymous lane (its protocol, `rand/7/tx/2`); otherwise a node that predates it (`/meshsub`, + /// signed, `rand/7/tx`). `Permissive` either way, so it delivers whatever arrives — a source, + /// had the node attached one, included — and content ids, as the node's. + fn sniffer(tx_lane: bool) -> (Swarm, IdentTopic) { + let kp = identity::Keypair::generate_ed25519(); + let mut b = gossipsub::ConfigBuilder::default(); + b.validation_mode(ValidationMode::Permissive) + .heartbeat_interval(Duration::from_millis(200)) + .max_transmit_size(GOSSIP_MAX_TRANSMIT_SIZE) + .message_id_fn(|m: &gossipsub::Message| MessageId::from(blake3::hash(&m.data).as_bytes().to_vec())); + if tx_lane { + b.protocol_id_prefix(tx_protocol_prefix(7)); + } + let auth = if tx_lane { MessageAuthenticity::Anonymous } else { MessageAuthenticity::Signed(kp.clone()) }; + let gs = gossipsub::Behaviour::new(auth, b.build().unwrap()).unwrap(); + let mut sw = libp2p::SwarmBuilder::with_existing_identity(kp) + .with_tokio() + .with_tcp(tcp::Config::default(), noise::Config::new, yamux::Config::default) + .unwrap() + .with_behaviour(|_| gs) + .unwrap() + .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60))) + .build(); + let topic = IdentTopic::new(if tx_lane { "rand/7/tx/2" } else { "rand/7/tx" }); + sw.behaviour_mut().subscribe(&topic).unwrap(); + (sw, topic) + } + + /// Drive `sw` for `within`, and return every gossip message it received. + async fn sniff(sw: &mut Swarm, within: Duration) -> Vec { + use libp2p::futures::StreamExt; + let until = tokio::time::Instant::now() + within; + let mut got = Vec::new(); + loop { + tokio::select! { + ev = sw.select_next_some() => { + if let SwarmEvent::Behaviour(gossipsub::Event::Message { message, .. }) = ev { + got.push(message); + } + } + _ = tokio::time::sleep_until(until) => return got, + } + } + } + + /// Start a node with `edge` and return it with its full address. + async fn listening(seed: u8, edge: EdgeConfig) -> (NetworkHandle, mpsc::Receiver, Multiaddr) { + let (n, mut rx) = start_with(loopback_cfg(vec![], WireLimits::default()), [seed; 32], edge).await.unwrap(); + let addr = wait_for(&mut rx, Duration::from_secs(5), |e| match e { + NetworkEvent::Listening(a) => Some(a), + _ => None, + }) + .await + .expect("listening"); + let full = addr.with(libp2p::multiaddr::Protocol::P2p(n.local_peer_id)); + (n, rx, full) + } + + /// Issue #146: a transaction gossiped on the anonymous lane carries no `source` and no + /// sequence number — nothing that names the node that published it. A node that predates the + /// lane still gets it, signed, on the legacy topic, unless `--anonymous-tx-only`. + #[tokio::test] + async fn a_gossiped_transaction_carries_no_author_and_the_legacy_copy_follows_the_flag() { + for anonymous_tx_only in [false, true] { + let edge = EdgeConfig { anonymous_tx_only, ..EdgeConfig::default() }; + let (n, _rx, n_addr) = listening(if anonymous_tx_only { 81 } else { 80 }, edge).await; + let (mut anon, _) = sniffer(true); + let (mut old, _) = sniffer(false); + anon.dial(n_addr.clone()).unwrap(); + old.dial(n_addr).unwrap(); + // The meshes form over a few heartbeats; a fresh transaction per try, because the + // content id makes a second publish of one transaction a duplicate. + let (mut on_anon, mut on_old) = (Vec::new(), Vec::new()); + for k in 0..20u64 { + n.broadcast(GossipMessage::Transaction(tx_of(1000 + k))).await; + let (a, o) = tokio::join!(sniff(&mut anon, Duration::from_millis(250)), sniff(&mut old, Duration::from_millis(250))); + on_anon.extend(a); + on_old.extend(o); + if !on_anon.is_empty() && (anonymous_tx_only || !on_old.is_empty()) && k >= 4 { + break; + } + } + assert!(!on_anon.is_empty(), "nothing on the anonymous lane (anonymous_tx_only = {anonymous_tx_only})"); + for m in &on_anon { + assert_eq!(m.source, None, "a transaction named its author"); + assert_eq!(m.sequence_number, None, "a transaction carried a sequence number"); + assert_eq!(m.topic, IdentTopic::new("rand/7/tx/2").hash()); + assert!(matches!(bincode::deserialize::(&m.data), Ok(GossipMessage::Transaction(_)))); + } + if anonymous_tx_only { + assert!(on_old.is_empty(), "--anonymous-tx-only published on the legacy signed topic"); + } else { + assert!(!on_old.is_empty(), "a node that predates the lane got no transaction"); + assert!(on_old.iter().all(|m| m.source == Some(n.local_peer_id)), "the legacy copy is the signed one"); + } + n.shutdown().await; + } + } + + /// Issue #146, the rolling upgrade: B publishes on both lanes, A hears each transaction twice + /// (once per lane) and its node loop sees it **once**, and when A accepts it, A relays the + /// held legacy copy too — so C, a node that predates the lane and is connected to A alone, + /// still gets every transaction. Without the hold, the second copy would reach A's node loop, + /// be judged a pool duplicate (`Ignore`) and never be relayed on the lane it came in on. + #[tokio::test] + async fn during_the_roll_a_transaction_is_judged_once_and_relayed_on_both_lanes() { + let (a, mut a_rx, a_addr) = listening(82, EdgeConfig::default()).await; + let (b, _b_rx) = start(loopback_cfg(vec![a_addr.clone()], WireLimits::default()), [83u8; 32]).await.unwrap(); + assert!(wait_for(&mut a_rx, Duration::from_secs(10), |e| matches!(e, NetworkEvent::PeerConnected(p) if p == b.local_peer_id).then_some(())).await.is_some()); + let (mut c, _) = sniffer(false); + c.dial(a_addr).unwrap(); + let c_id = *c.local_peer_id(); + // C's swarm must be driven for its dial to complete. + let (_, c_connected) = tokio::join!( + sniff(&mut c, Duration::from_secs(3)), + wait_for(&mut a_rx, Duration::from_secs(10), |e| matches!(e, NetworkEvent::PeerConnected(p) if p == c_id).then_some(())) + ); + assert!(c_connected.is_some(), "C never connected to A"); + + let mut judged: Vec = Vec::new(); + let mut at_c = Vec::new(); + for k in 0..30u64 { + b.broadcast(GossipMessage::Transaction(tx_of(2000 + k))).await; + // A's node loop: report Accept for every transaction it is handed, as the node does + // for one that verifies. + let deadline = tokio::time::Instant::now() + Duration::from_millis(400); + loop { + tokio::select! { + ev = a_rx.recv() => { + if let Some(NetworkEvent::Gossip { from, msg: GossipMessage::Transaction(_), id }) = ev { + assert_eq!(from, id.propagation_source, "a transaction's `from` is its forwarder"); + a.report_validation(id.clone(), MessageAcceptance::Accept).await; + judged.push(id); + } + } + got = sniff(&mut c, Duration::from_millis(50)) => at_c.extend(got), + _ = tokio::time::sleep_until(deadline) => break, + } + } + if !at_c.is_empty() && k >= 6 { + break; + } + } + // Let the last holds resolve, and drain. + let tail = sniff(&mut c, Duration::from_millis(800)).await; + at_c.extend(tail); + while let Ok(ev) = a_rx.try_recv() { + if let NetworkEvent::Gossip { msg: GossipMessage::Transaction(_), id, .. } = ev { + judged.push(id); + } + } + assert!(!judged.is_empty(), "A's node loop was never handed a transaction"); + let ids: HashSet<&MessageId> = judged.iter().map(|id| &id.message_id).collect(); + assert_eq!(ids.len(), judged.len(), "one transaction reached A's node loop twice"); + assert!(!at_c.is_empty(), "C, behind A on the legacy topic only, never got a transaction"); + // C is connected to A alone, so whatever it got, A relayed. + for m in &at_c { + assert_eq!(m.source, Some(b.local_peer_id), "the legacy copy C got is B's own, relayed"); + } + a.shutdown().await; + b.shutdown().await; + } + + fn gid(n: u8, lane: GossipLane) -> GossipId { + GossipId { message_id: MessageId::from(vec![n]), propagation_source: peer_of_seed(n), lane } + } + + /// `TxCross`, the rules: the first copy goes to the node, the other lane's copy waits for its + /// verdict or takes it, a verdict releases every held copy once, and an expired entry hands + /// back what it still held so it is reported. + #[test] + fn tx_cross_judges_the_first_copy_and_answers_the_other_lane_with_its_verdict() { + use GossipLane::*; + let t0 = Instant::now(); + let mut x = TxCross::default(); + let first = gid(1, AnonymousTx); + assert_eq!(x.on_delivery(&first, t0), CrossDelivery::First); + let legacy = GossipId { propagation_source: peer_of_seed(2), lane: Signed, ..first.clone() }; + assert_eq!(x.on_delivery(&legacy, t0), CrossDelivery::Held); + // A verdict reported for the held copy itself releases nothing (it is not the first). + assert!(x.on_verdict(&legacy, MessageAcceptance::Reject).is_empty()); + let released = x.on_verdict(&first, MessageAcceptance::Accept); + assert_eq!(released.len(), 1); + assert_eq!(released[0].propagation_source, legacy.propagation_source); + assert_eq!(released[0].lane, Signed); + assert!(x.on_verdict(&first, MessageAcceptance::Accept).is_empty(), "released once"); + // A later copy on the other lane takes the verdict straight away. + let late = GossipId { propagation_source: peer_of_seed(3), lane: Signed, ..first.clone() }; + assert_eq!(x.on_delivery(&late, t0), CrossDelivery::Answer(MessageAcceptance::Accept)); + // The first lane again is no second lane: judged as the node would. + assert_eq!(x.on_delivery(&first, t0), CrossDelivery::First); + + // Expiry: an entry still holding a copy hands it back to be reported; a fresh one stays. + let other = gid(9, Signed); + assert_eq!(x.on_delivery(&other, t0), CrossDelivery::First); + assert_eq!(x.on_delivery(&GossipId { lane: AnonymousTx, ..other.clone() }, t0), CrossDelivery::Held); + assert!(x.expire(t0 + TX_CROSS_TTL / 2).is_empty()); + let orphans = x.expire(t0 + TX_CROSS_TTL); + assert_eq!(orphans.len(), 1); + assert_eq!(orphans[0].lane, AnonymousTx); + assert!(x.seen.is_empty()); + } + + /// The anonymous lane's own validation mode refuses a message that names an author, so no + /// relay can attach one; and its configuration is accepted by gossipsub at all (Anonymous + /// authenticity under Anonymous validation is the only pairing it allows). + #[test] + fn the_transaction_lane_is_anonymous_by_configuration() { + let topics = Topics::new(7); + let cfg = gossip_config(&WireLimits::default(), ValidationMode::Anonymous, Some(tx_protocol_prefix(7)), topics.status.hash()).unwrap(); + assert!(matches!(cfg.validation_mode(), ValidationMode::Anonymous)); + assert!(gossipsub::Behaviour::::new(MessageAuthenticity::Anonymous, cfg.clone()).is_ok()); + let kp = identity::Keypair::generate_ed25519(); + assert!( + gossipsub::Behaviour::::new(MessageAuthenticity::Signed(kp), cfg).is_err(), + "a signing publisher on the anonymous lane is refused at construction" + ); + assert_eq!(topics.lane_of(&topics.tx_anon.hash()), GossipLane::AnonymousTx); + for t in [&topics.tx, &topics.consensus, &topics.status, &topics.peers] { + assert_eq!(topics.lane_of(&t.hash()), GossipLane::Signed); + } + assert!(topics.is_tx(&topics.tx.hash()) && topics.is_tx(&topics.tx_anon.hash())); + assert!(!topics.is_tx(&topics.status.hash())); + } + /// A node with no connection limit accepts every inbound connection, and a fresh libp2p /// identity costs nothing to mint (deep scan 2026-09-24: 200 of them were all accepted). With /// the inbound cap at 3, eight dialers must leave the target holding at most three peers, @@ -1725,13 +2266,17 @@ mod tests { assert_eq!(t7.tx.to_string(), "rand/7/tx"); assert_eq!(t7.status.to_string(), "rand/7/status"); assert_eq!(t7.peers.to_string(), "rand/7/peers"); - let names7: HashSet = [&t7.consensus, &t7.tx, &t7.status, &t7.peers].iter().map(|t| t.to_string()).collect(); - let names8: HashSet = [&t8.consensus, &t8.tx, &t8.status, &t8.peers].iter().map(|t| t.to_string()).collect(); - assert_eq!(names7.len(), 4, "four distinct topics on one chain"); + assert_eq!(t7.tx_anon.to_string(), "rand/7/tx/2"); + let names7: HashSet = [&t7.consensus, &t7.tx, &t7.status, &t7.peers, &t7.tx_anon].iter().map(|t| t.to_string()).collect(); + let names8: HashSet = [&t8.consensus, &t8.tx, &t8.status, &t8.peers, &t8.tx_anon].iter().map(|t| t.to_string()).collect(); + assert_eq!(names7.len(), 5, "five distinct topics on one chain"); assert!(names7.is_disjoint(&names8), "no topic is shared between chains"); // The hashes gossipsub subscribes by are the names' hashes, so they differ the same way. - let hashes: HashSet<_> = [&t7.consensus, &t7.tx, &t7.status, &t7.peers, &t8.consensus, &t8.tx, &t8.status, &t8.peers].iter().map(|t| t.hash()).collect(); - assert_eq!(hashes.len(), 8); + let hashes: HashSet<_> = [&t7.consensus, &t7.tx, &t7.status, &t7.peers, &t7.tx_anon, &t8.consensus, &t8.tx, &t8.status, &t8.peers, &t8.tx_anon] + .iter() + .map(|t| t.hash()) + .collect(); + assert_eq!(hashes.len(), 10); // A chain id at the top of the range is spelled in full, not truncated. assert_eq!(Topics::new(u64::MAX).tx.to_string(), format!("rand/{}/tx", u64::MAX)); } @@ -1880,13 +2425,15 @@ mod tests { } /// `EdgeConfig::default()` is what `start` passes: no reserved or bound peers, permissive - /// gossip. The `--strict-gossip` and `--reserved-peer` paths are opt-in. + /// gossip, both transaction lanes. The `--strict-gossip`, `--reserved-peer` and + /// `--anonymous-tx-only` paths are opt-in. #[test] fn the_default_edge_config_reserves_nobody_and_is_permissive() { let e = EdgeConfig::default(); assert!(e.reserved.is_empty()); assert!(e.bound.is_empty()); assert!(!e.strict_gossip); + assert!(!e.anonymous_tx_only, "both transaction lanes by default, for the rolling upgrade (#146)"); } /// The handle's commands, each one: every method sends the variant the swarm task matches @@ -1911,7 +2458,7 @@ mod tests { h.unreserve_peer(peer).await; assert!(matches!(rx.recv().await, Some(NetworkCommand::Unreserve(p)) if p == peer)); - let id = GossipId { message_id: MessageId::from(vec![1, 2, 3]), propagation_source: peer }; + let id = GossipId { message_id: MessageId::from(vec![1, 2, 3]), propagation_source: peer, lane: GossipLane::AnonymousTx }; h.report_validation(id.clone(), MessageAcceptance::Reject).await; match rx.recv().await { Some(NetworkCommand::ReportValidation { id: got, acceptance: MessageAcceptance::Reject }) => { diff --git a/crates/randprotocol-node/src/node.rs b/crates/randprotocol-node/src/node.rs index cc851f57..3ae649d7 100644 --- a/crates/randprotocol-node/src/node.rs +++ b/crates/randprotocol-node/src/node.rs @@ -2508,6 +2508,9 @@ pub struct NetOptions { /// gossipsub's strict validation (`--strict-gossip`, audit v6, CH-7): see /// [`network::EdgeConfig::strict_gossip`]. pub strict_gossip: bool, + /// Gossip transactions on the anonymous lane only (`--anonymous-tx-only`, issue #146): see + /// [`network::EdgeConfig::anonymous_tx_only`]. + pub anonymous_tx_only: bool, } pub async fn start(cfg: NodeConfig) -> Result { @@ -2641,6 +2644,7 @@ pub async fn start_with(cfg: NodeConfig, rpc_options: RpcOptions, net_options: N reserved: net_options.reserved_peers.clone(), bound: peer_bindings.bound_peers(), strict_gossip: net_options.strict_gossip, + anonymous_tx_only: net_options.anonymous_tx_only, }, ) .await?; @@ -7928,7 +7932,7 @@ mod tests { /// A delivery id from a fresh forwarder. fn gossip_id() -> GossipId { - GossipId { message_id: libp2p::gossipsub::MessageId::new(b"compact"), propagation_source: PeerId::random() } + GossipId { message_id: libp2p::gossipsub::MessageId::new(b"compact"), propagation_source: PeerId::random(), lane: Default::default() } } /// A compact proposal naming only transactions in the pool rebuilds and is handled as a diff --git a/crates/randprotocol-node/tests/network.rs b/crates/randprotocol-node/tests/network.rs index 830f459e..bf08f257 100644 --- a/crates/randprotocol-node/tests/network.rs +++ b/crates/randprotocol-node/tests/network.rs @@ -316,43 +316,64 @@ async fn each_message_rides_its_own_topic_as_the_bincode_of_the_gossip_message() a.h.shutdown().await; } -/// The id a delivery is reported with is content-addressed: blake3 of the message bytes, which is -/// why two forwarders of one message produce one id. Checked from outside against the bytes the -/// publisher put on the wire. +fn tx(n: u64) -> randprotocol_core::Transaction { + // `chain_id` only makes the bytes differ: the network task never judges a transaction. + randprotocol_core::Transaction { chain_id: n, bundle: None, action: randprotocol_core::Action::None } +} + +/// Publish `tx` from `from` until `to` delivers it, as [`gossip_until_delivered`] does a status. +async fn gossip_tx_until_delivered(from: &NetworkHandle, to: &mut Peer, tx: randprotocol_core::Transaction) -> (PeerId, GossipId) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(15); + let want = tx.chain_id; + while tokio::time::Instant::now() < deadline { + from.broadcast(GossipMessage::Transaction(tx.clone())).await; + let got = to + .wait(Duration::from_millis(400), |e| match e { + NetworkEvent::Gossip { from, msg: GossipMessage::Transaction(t), id } if t.chain_id == want => Some((from, id)), + _ => None, + }) + .await; + if let Some(got) = got { + return got; + } + } + panic!("transaction {want} was never delivered"); +} + +/// The id a transaction's delivery is reported with is content-addressed: blake3 of the message +/// bytes, on either transaction lane (issue #146), which is why two forwarders of one transaction +/// produce one id. Checked from outside against the bytes the publisher put on the wire. A status +/// is the exception (issue #144): its id also covers its author and sequence number, so it is not +/// the bytes' hash. #[tokio::test] async fn a_delivered_messages_id_is_the_blake3_of_its_bytes() { let (mut a, b) = pair(21, 22).await; - let s = status(9); - let bytes = bincode::serialize(&GossipMessage::Status(s.clone())).unwrap(); - let (from, id) = gossip_until_delivered(&b.h, &mut a, s).await; - assert_eq!(from, b.h.local_peer_id); + let t = tx(9); + let bytes = bincode::serialize(&GossipMessage::Transaction(t.clone())).unwrap(); + let (from, id) = gossip_tx_until_delivered(&b.h, &mut a, t).await; + assert_eq!(from, b.h.local_peer_id, "one hop: the forwarder, which a transaction's `from` is"); assert_eq!(id.propagation_source, b.h.local_peer_id); assert_eq!(id.message_id.0, blake3::hash(&bytes).as_bytes().to_vec()); a.h.report_validation(id, MessageAcceptance::Accept).await; + + let s = status(9); + let bytes = bincode::serialize(&GossipMessage::Status(s.clone())).unwrap(); + let (_, id) = gossip_until_delivered(&b.h, &mut a, s).await; + assert_ne!(id.message_id.0, blake3::hash(&bytes).as_bytes().to_vec(), "a status's id is its author's too (#144)"); + a.h.report_validation(id, MessageAcceptance::Accept).await; a.h.shutdown().await; b.h.shutdown().await; } -/// The same bytes published three times are delivered once: the content-addressed id makes the -/// repeats duplicates at the publisher and at the receiver alike. -#[tokio::test] -async fn the_same_bytes_published_again_are_delivered_once() { - let (mut a, b) = pair(23, 24).await; - // Warm the mesh with another message first, so the publishes below all go out. - let (_, id) = gossip_until_delivered(&b.h, &mut a, status(1)).await; - a.h.report_validation(id, MessageAcceptance::Accept).await; - - let s = status(2); - for _ in 0..3 { - b.h.broadcast(GossipMessage::Status(s.clone())).await; - } +/// Every delivery of `pred`'s kind within `within`, each reported Accept. +async fn count_deliveries(a: &mut Peer, within: Duration, pred: impl Fn(&GossipMessage) -> bool) -> usize { let mut deliveries = 0; - let deadline = tokio::time::Instant::now() + Duration::from_secs(3); - while tokio::time::Instant::now() < deadline { + let deadline = tokio::time::Instant::now() + within; + loop { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); let got = a .wait(remaining, |e| match e { - NetworkEvent::Gossip { msg: GossipMessage::Status(got), id, .. } if got.height == 2 => Some(id), + NetworkEvent::Gossip { msg, id, .. } if pred(&msg) => Some(id), _ => None, }) .await; @@ -361,10 +382,35 @@ async fn the_same_bytes_published_again_are_delivered_once() { deliveries += 1; a.h.report_validation(id, MessageAcceptance::Accept).await; } - None => break, + None => return deliveries, } } - assert_eq!(deliveries, 1); +} + +/// The same transaction published three times is delivered once — the content-addressed id makes +/// the repeats duplicates at the publisher and at the receiver alike, and the copy on the second +/// lane is held for the first's verdict (issue #146). The same *status* published three times is +/// three statuses (issue #144): each re-announcement reaches its peers. +#[tokio::test] +async fn the_same_bytes_published_again_are_delivered_once() { + let (mut a, b) = pair(23, 24).await; + // Warm both meshes with other messages first, so the publishes below all go out. + let (_, id) = gossip_tx_until_delivered(&b.h, &mut a, tx(1)).await; + a.h.report_validation(id, MessageAcceptance::Accept).await; + let (_, id) = gossip_until_delivered(&b.h, &mut a, status(1)).await; + a.h.report_validation(id, MessageAcceptance::Accept).await; + // Drain what the warm-up's retries left behind. + count_deliveries(&mut a, Duration::from_millis(1500), |_| true).await; + + for _ in 0..3 { + b.h.broadcast(GossipMessage::Transaction(tx(2))).await; + } + assert_eq!(count_deliveries(&mut a, Duration::from_secs(3), |m| matches!(m, GossipMessage::Transaction(t) if t.chain_id == 2)).await, 1); + + for _ in 0..3 { + b.h.broadcast(GossipMessage::Status(status(2))).await; + } + assert_eq!(count_deliveries(&mut a, Duration::from_secs(3), |m| matches!(m, GossipMessage::Status(s) if s.height == 2)).await, 3); a.h.shutdown().await; b.h.shutdown().await; } diff --git a/crates/randprotocol-prover/src/main.rs b/crates/randprotocol-prover/src/main.rs index b16100e1..308a00b4 100644 --- a/crates/randprotocol-prover/src/main.rs +++ b/crates/randprotocol-prover/src/main.rs @@ -169,11 +169,20 @@ fn load_key(path: &Path) -> Result { ProverKey::load(path).with_context(|| format!("loading {} (run `rand-prover keygen` first)", path.display())) } +/// Colour in the log only for a person at a terminal (issue #143): under systemd or launchd +/// stderr is a pipe, and escapes there break `grep` and log parsers. `NO_COLOR` turns it off +/// anywhere (: present and not empty). +fn log_ansi(is_terminal: bool, no_color: Option) -> bool { + is_terminal && no_color.is_none_or(|v| v.is_empty()) +} + #[tokio::main] async fn main() -> Result<()> { + use std::io::IsTerminal; tracing_subscriber::fmt() .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into())) .with_writer(std::io::stderr) + .with_ansi(log_ansi(std::io::stderr().is_terminal(), std::env::var_os("NO_COLOR"))) .init(); let cli = Cli::parse(); let home = expand_home(&cli.home)?; @@ -322,6 +331,15 @@ async fn shutdown_signal() -> std::io::Result<()> { mod tests { use clap::Parser; + #[test] + fn log_colour_only_at_a_terminal_and_never_under_no_color() { + use super::log_ansi; + assert!(log_ansi(true, None), "a terminal with no NO_COLOR gets colour"); + assert!(!log_ansi(false, None), "a pipe (journald) gets none"); + assert!(!log_ansi(true, Some("1".into())), "NO_COLOR=1 turns it off at a terminal"); + assert!(log_ansi(true, Some("".into())), "an empty NO_COLOR is unset (no-color.org)"); + } + #[test] fn run_takes_the_fee_flags_both_or_neither() { let run = |extra: &[&str]| super::Cli::try_parse_from([&["rand-prover", "run"][..], extra].concat()); diff --git a/deploy/caddy/README.md b/deploy/caddy/README.md index ab138205..2e46ac9b 100644 --- a/deploy/caddy/README.md +++ b/deploy/caddy/README.md @@ -72,15 +72,22 @@ designed. What remains for the operator is housekeeping, listed at the end. ## The node's own public listener (audit v6, VK-2 / RPC-4) — point the tunnel at it From v0.6.8 the node has a listener made for this hop: `rand-node run --public-rpc 127.0.0.1:8546`. -It serves a fixed method set (no viewing-key methods, no `rand_mint`, no `rand_getPeers`), takes -no batch and no WebSocket, meters every caller together, and never counts a caller as loopback — -so the method gate and the meter are the node's, not only the sale proxy's `RPC_ALLOWED` in -another repository. **To do on the roll, by the operator** (not yet done; this file records the -target, not the state): +It serves a fixed method set (`rpc::PUBLIC_METHODS`: no viewing-key methods, no `rand_mint`, no +`rand_getPeers`), takes no batch and no WebSocket, meters every caller together, and never counts a +caller as loopback — so the method gate and the meter are the node's, not only the sale proxy's +`RPC_ALLOWED` in another repository. + +**State (2026-10-28): not recorded.** The last recorded state of this hop is the 2026-09-27 tunnel +above, to obs1's operator port `:8545`. obs1 was rebuilt on another host in the 2026-10-22 +DigitalOcean migration, and nothing in this repository records whether its new unit carries +`--public-rpc`. To find out, from the web droplet: `rand_getPeers` through `127.0.0.1:18545` +answers "not served on this node's public listener" once the hop ends at `:8546`, and answers a +peer list while it still ends at `:8545`; on obs1, `systemctl cat rand-node | grep -- --public-rpc`. +Write the answer here. If it is not done, the steps, by the operator, one host at a time: 1. obs1's unit: add `--public-rpc 127.0.0.1:8546` to `ExecStart` — `deploy/enable-public-rpc.sh root@` (edits the unit, idempotent; `RESTART=1` restarts it and checks the listener refuses - `rand_getPeers`; a new host gets it from `PUBLIC_RPC=1 deploy/vps-setup.sh …`). Roll one host at a time. + `rand_getPeers`; a new host gets it from `PUBLIC_RPC=1 deploy/vps-setup.sh …`). 2. obs1's `rpctunnel` key: `permitopen="127.0.0.1:8546"` in place of `:8545`. 3. The web droplet's `rpc-tunnel-obs1.service`: `-L 127.0.0.1:18545:127.0.0.1:8546`. 4. Check from the web droplet: `rand_chainId` answers through `127.0.0.1:18545`; @@ -90,6 +97,28 @@ target, not the state): Until then the section below describes the running chain. +## Which layer gates which method + +The proxy's `RPC_ALLOWED` (randprotocol.org `server/sale/src/rpc.rs` at `8cebd54`) and the +node's `rpc::PUBLIC_METHODS` (since v0.8.0) are **the same 62 methods**; keep them so, and change +both together. `rand_getUnsealed` is on both: it pages the bundles an aggregator may still cover, a public read +the aggregate daemon polls. + +| Method | Sale proxy (`RPC_ALLOWED`) | Node public listener (`--public-rpc`) | Node operator listener (`--rpc`) | +|---|---|---|---| +| the 62 public reads and `rand_sendTransaction` | served; per-IP limits | served; one shared meter | served | +| `rand_mint` | `-32601` | refused | served — the port binds `127.0.0.1`; the genesis pays only `staking.faucet_recipients` | +| `rand_importViewingKey`, `rand_getViewingNotes`, `rand_removeViewingKey` | `-32601` | refused | loopback callers; the token's holders under `--rpc-viewing-token-file`; anyone under `--rpc-viewing-open` (VK-3) | +| `rand_getPeers` | `-32601` | refused | served | +| a batch (top-level array) | refused | refused | up to 20 requests (`MAX_BATCH`) | +| anything added to the node later | `-32601` until named | refused until named | served | + +While the tunnel ends at `:8545` (see the state above) the node's column that applies is the +operator listener's, every proxied caller is loopback to it, and the proxy is the only gate. Once +it ends at `:8546` both gates apply. The sale droplet serves the list its binary was built with, +and the 2026-10-11 sale deploy was built before `8cebd54`; check that the running sale binary is +at or past that commit before relying on the table's first column. + ## Why the proxy is security-critical E's node sees every proxied request from loopback (randscan's Caddy), and it decides two protections @@ -103,12 +132,13 @@ by the TCP peer address, treating loopback as the operator: The node's own per-caller defences are therefore off for public traffic, and the sale service's proxy stands where they would. It does, verified 2026-09-26 against the live endpoint: -1. **A method allowlist** (`RPC_ALLOWED`) — `rand_mint`, `rand_importViewingKey`, - `rand_getViewingNotes`, `rand_removeViewingKey`, `rand_getUnsealed` and `rand_getPeers` are not - on it and answer `-32601 method not allowed here`. An allowlist, so a method added to the node - later stays closed until someone opens it there. +1. **A method allowlist** (`RPC_ALLOWED`, the table above) — `rand_mint`, `rand_importViewingKey`, + `rand_getViewingNotes`, `rand_removeViewingKey` and `rand_getPeers` are not on it and answer + `-32601 method not allowed here` (`rand_getUnsealed` is on it, as on the node's public list). An + allowlist, so a method added to the node later stays closed until someone opens it there. 2. **No batches** — a top-level array is refused, so a refused method cannot ride inside one. -3. **Per-IP limits** — 120 reads and 6 `rand_sendTransaction` a minute, keyed on the real client IP +3. **Per-IP limits** — 1200 reads (`READS_PER_MINUTE`; 120 until a wallet's first sync outgrew it) + and 6 `rand_sendTransaction` a minute, keyed on the real client IP (`client_ip` reads `X-Real-IP`, set by the web droplet's nginx from `CF-Connecting-IP` for Cloudflare's ranges only — a direct connection to the origin is metered as its own TCP peer and cannot name an address), and a 12 MiB body cap. diff --git a/deploy/cutover-droplet-chain14.sh b/deploy/cutover-droplet-chain14.sh index b5ab9314..516341c8 100755 --- a/deploy/cutover-droplet-chain14.sh +++ b/deploy/cutover-droplet-chain14.sh @@ -115,7 +115,7 @@ $SSH "set -e cp -a \$UNIT /root/$SERVICE.service.chain13.bak cat > \$UNIT < [build-host] # @@ -91,6 +92,13 @@ $SSH "set -e sed -i \"s|^ExecStart=\(.*\)\$|ExecStart=\1 ${PRUNE_ARGS}|\" /etc/systemd/system/$SERVICE.service systemctl daemon-reload fi + # A unit written by the chain-14 cutover still says \"RAND full node (d, chain 14)\" (#143): later + # cutovers only swap the datadir suffix. The Description names the host and no chain, so it never + # goes stale again; the chain is in the datadir and in rand_status. + if grep -q '^Description=RAND full node (.*, chain [0-9]*)\$' /etc/systemd/system/$SERVICE.service; then + sed -i 's/^Description=RAND full node (\(.*\), chain [0-9]*)\$/Description=RAND full node (\1)/' /etc/systemd/system/$SERVICE.service + systemctl daemon-reload + fi systemctl restart $SERVICE sleep 4 systemctl is-active $SERVICE diff --git a/docs/architecture.md b/docs/architecture.md index 3b2268ac..baed9644 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -392,8 +392,11 @@ chain id, or to run the old-constraint nodes (or the new ones, on old history) w ## 8. Networking and sync; mempool -libp2p 0.54 over TCP with Noise and Yamux. Behaviours: **gossipsub** (topics -`rand//consensus`, `/tx`, `/status`), **Kademlia** (protocol `/rand//kad/1`) +libp2p 0.54 over TCP with Noise and Yamux. Behaviours: **gossipsub**, twice — signed +(`MessageAuthenticity::Signed`, topics `rand//consensus`, `/status`, `/peers` and the +legacy `/tx`) and anonymous (`MessageAuthenticity::Anonymous` under `ValidationMode::Anonymous`, on +its own protocol `/rand//txsub`, topic `rand//tx/2`: a transaction names no +author, issue #146) — **Kademlia** (protocol `/rand//kad/1`) seeded from `--bootstrap`, **identify**, optional **mDNS** for LAN discovery, **request-response** (`/rand/sync/1`, CBOR) for block sync, and **ping** every 15 s. A node remembers every address it dialled or learnt through identify and redials disconnected known peers every 30 s, in addition to diff --git a/docs/cli.md b/docs/cli.md index 23d758a1..4ce7dc40 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -390,6 +390,7 @@ one the cut announced; a mismatch on one node is almost always an old binary. | `--bootstrap ` | none, repeatable | peer to dial at start and every 30 s while disconnected: `/ip4//tcp//p2p/` | | `--reserved-peer ` | none, repeatable | a peer admitted past the inbound connection cap and served from the validators' share of the sync budget (audit v6, NET-1). Bootstrap peers and every validator whose signed peer binding this node has seen are reserved without it; list the fleet's validators here for a freshly cut chain, whose store holds no bindings yet (`docs/deploy.md`, "Reserved peers") | | `--strict-gossip` | off | gossipsub's strict validation (audit v6, CH-7): accept only messages with a valid author signature, sequence number and source. Every node signs what it publishes, so this drops only forged unsigned messages, and strict and permissive nodes interoperate — it rolls node by node (`docs/deploy.md`) | +| `--anonymous-tx-only` | off | gossip transactions only on the anonymous lane `rand//tx/2` (issue #146): no subscription to, and no publish on, the legacy signed topic `rand//tx`, whose messages name the node that first published them. Off, a node publishes each transaction on both and relays on both. Turn it on only once every node on the chain runs v0.8.3 or later (`docs/deploy.md`, "The anonymous transaction lane") | | `--rpc ` | `127.0.0.1:8545` | the operator's JSON-RPC listener: every method, and it trusts loopback. Keep it on loopback; never forward a public endpoint to it | | `--public-rpc ` | off | a second, public listener (audit v6): a fixed method set (no viewing-key methods, `rand_mint` or `rand_getPeers`), no batches, no WebSocket, one meter for all callers. What a reverse proxy or an SSH forward points at (`docs/rpc.md`, "Two listeners") | | `--rpc-viewing-token-file ` | off | require `Authorization: Bearer ` (32+ characters) on the viewing-key methods of `--rpc` | diff --git a/docs/deploy.md b/docs/deploy.md index 1c9b9f2a..2ced4396 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -172,6 +172,38 @@ node**, and the older advice here (a strict/permissive mix drops messages; only all-stop/all-start) does not hold for this fleet. Turn it on everywhere once it has run on a canary. +### The anonymous transaction lane (v0.8.3, issue #146) + +Before v0.8.3 every transaction rode the signed gossip topic `rand//tx`: each message +carried the publishing node's PeerId, a signature and a sequence number, every relay kept them, +and so every node learned which peer first published each transaction. v0.8.3 adds a second +gossipsub behaviour for transactions alone — `MessageAuthenticity::Anonymous` under +`ValidationMode::Anonymous`, on its own protocol `/rand//txsub`, topic `rand//tx/2`. +A message there has no source, sequence number or signature, and one that carries any is refused. +Consensus, status and peer bindings stay signed. Rate limiting is unchanged: it always keyed on +the forwarder (`propagation_source`), never the author. + +The two lanes cannot share a topic or a protocol, so the roll runs both, then drops the old one: + +1. **Roll v0.8.3 everywhere, flag off (the default).** A v0.8.3 node subscribes both topics, + publishes each transaction on both, and relays on both: a copy heard on one lane is held for + the verdict on the first copy, so the node judges a transaction once and relays it on both + lanes. An older node speaks only `/meshsub` and sees exactly what it saw before — the legacy + copy is still signed, so `--strict-gossip` on an older node drops nothing (a v0.8.3 node + never publishes unsigned on `/meshsub`). Mixed fleets are safe in either direction: an older + node's transactions reach v0.8.3 nodes on the legacy topic. During this phase the legacy copy + still names the author: **the privacy gain starts at step 2.** +2. **After every node on the chain runs v0.8.3 or later** — validators, observers, archives, the + site and bridge nodes — restart each with `--anonymous-tx-only`, one at a time. It stops + publishing on and subscribing to `rand//tx`. Order does not matter: a flagged node + still hears every transaction on `tx/2` from the others, and a not-yet-flagged node still + publishes on both. A node from before v0.8.3 left on the chain would stop hearing the flagged + nodes' transactions (it would still get them in blocks). +3. The next release can make `--anonymous-tx-only` the default and drop the legacy topic. + +The lane does not hide timing: the first node to relay a transaction is still the likeliest +author to a well-connected observer. That is the Dandelion++ issue's to close. + ## Rolling out a new commit 1. `cargo test` locally, commit, push.