From 374d41feaed2942238a4c23e76b200acceebd8a5 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Thu, 10 Sep 2026 18:12:38 +0100 Subject: [PATCH 1/2] Carry block metadata on gossip relay requests Attach GossipBlock { slot, block_root } to existing SendGossip requests at both block relay sites. RPC retains its empty callback, and other topics carry no block record. This prepares SSE publication without a new block stage or second spine notification. Relay selection, import receipts, and networking behaviour remain unchanged. Eight block_relay tests cover metadata in both sync modes, RPC and replay silence, staging, missing-parent retry, rejected imports, bad signatures, and disabled relay. They require ef_tests, enabled by just nextest. Fixture slots come from the decoded blocks. Existing producer tests check absent metadata for nine non-block topic families. The proposer-lookahead test checks eligibility in both sync modes while bypassing signature verification. A Controller test injects opaque payloads through the spine and enters loop_body. With metadata absent and present, forwarding preserves bytes and recipient selection, and IWANT service works. The peer-manager test retains sender and IDONTWANT exclusions. Fixture limits: gossip tests reuse SSZ handles as unread protobuf handles. DA completion enters its handler without running the columns tile. No fixture covers the full Gloas AwaitParentPayload retry; the relay-disabled test checks its handler flag directly. Validation, all exit zero: - just fmt-check - just clippy - just nextest - git diff --check 1,357 tests passed; 5 skipped. Assisted-by: Codex:gpt-6-astra --- crates/beacon_state/tile/src/tile/block.rs | 9 +- crates/beacon_state/tile/src/tile/gossip.rs | 28 +- .../beacon_state/tile/src/tile/orphan_pool.rs | 2 +- crates/beacon_state/tile/src/tile/tests.rs | 162 ++++++---- .../tile/src/tile/tests/block_relay.rs | 288 ++++++++++++++++++ crates/columns/src/tile.rs | 1 + crates/common/src/lib.rs | 21 +- crates/common/src/spine.rs | 4 +- crates/common/src/spine/messages.rs | 10 +- crates/control/src/tile.rs | 5 + crates/control/src/tile/tests.rs | 153 ++++++++++ crates/peer/src/manager/mod.rs | 1 + crates/peer/src/manager/promises.rs | 4 +- 13 files changed, 603 insertions(+), 85 deletions(-) create mode 100644 crates/beacon_state/tile/src/tile/tests/block_relay.rs create mode 100644 crates/control/src/tile/tests.rs diff --git a/crates/beacon_state/tile/src/tile/block.rs b/crates/beacon_state/tile/src/tile/block.rs index da497c14..41a26521 100644 --- a/crates/beacon_state/tile/src/tile/block.rs +++ b/crates/beacon_state/tile/src/tile/block.rs @@ -6,7 +6,7 @@ use silver_beacon_state_data::{ }; use silver_common::{ BeaconStateEvent, BlockSource, BlockStage, EngineFcuReq, EngineNewPayloadReq, EngineReq, - SyncNeed, SyncUpdate, TCacheRead, TRandomAccess, hex32, + GossipBlock, SyncNeed, SyncUpdate, TCacheRead, TRandomAccess, hex32, ssz_view::{ self, BEACON_BLOCK_BODY_FIXED, BeaconBlockBodyFuluView, BeaconBlockBodyGloasView, SignedBeaconBlockView, @@ -96,7 +96,7 @@ impl BeaconStateTile { source: BlockSource, pre_verified: bool, producers: &mut Producers, - mut send_gossip: impl FnMut(&mut Producers), + mut send_gossip: impl FnMut(&mut Producers, GossipBlock), ) -> Feedback { if let Err(e) = Self::check_block_size(data) { tracing::warn!(?source, "{e}"); @@ -107,7 +107,10 @@ impl BeaconStateTile { let parsed = match self.parse_and_verify_block(data, pre_verified) { Ok(parsed) => { if parsed.relay_eligible { - send_gossip(producers); + send_gossip(producers, GossipBlock { + slot: block_slot, + block_root: parsed.block_root, + }); } parsed } diff --git a/crates/beacon_state/tile/src/tile/gossip.rs b/crates/beacon_state/tile/src/tile/gossip.rs index 8c5d802f..3589250e 100644 --- a/crates/beacon_state/tile/src/tile/gossip.rs +++ b/crates/beacon_state/tile/src/tile/gossip.rs @@ -6,8 +6,8 @@ use silver_beacon_state_data::{ }; use silver_common::{ ATTESTATION_SUBNETS, BeaconStateEvent, BlockSource, EngineNewPayloadEnvelopeReq, EngineReq, - GossipTopic, LOCAL_GOSSIP_STREAM_ID, MAX_BLOBS_PER_BLOCK, NewGossipMsg, PeerEvent, SyncNeed, - TCacheRead, TRead, hex32, + GossipBlock, GossipTopic, LOCAL_GOSSIP_STREAM_ID, MAX_BLOBS_PER_BLOCK, NewGossipMsg, PeerEvent, + SyncNeed, TCacheRead, TRead, hex32, metrics::timed, ssz_view::{ AttestationDataView, AttesterSlashingView, ExecutionPayloadEnvelopeView as Envelope, @@ -428,7 +428,7 @@ impl BeaconStateTile { committed_ptc = true; } } - Self::relay_gossip(&m, producers); + Self::relay_gossip(&m, None, producers); accepted = true; } else { Self::reject_gossip(&m, producers); @@ -1220,9 +1220,14 @@ impl BeaconStateTile { let feedback = match m.topic { GossipTopic::BeaconBlock if !self.sync_target.is_following() => { match self.parse_and_verify_block(data, pre_verified) { - Ok(parsed) if do_relay && parsed.relay_eligible => { - Self::relay_gossip(&m, producers) - } + Ok(parsed) if do_relay && parsed.relay_eligible => Self::relay_gossip( + &m, + Some(GossipBlock { + slot: parsed.header.slot, + block_root: parsed.block_root, + }), + producers, + ), Err(err) if matches!(err.feedback(), Feedback::Reject(_)) => { producers.produce(PeerEvent::P2pGossipInvalidMsg { p2p_peer: m.stream_id.peer(), @@ -1241,9 +1246,9 @@ impl BeaconStateTile { BlockSource::Gossip, pre_verified, producers, - |p| { + |p, block| { if do_relay { - Self::relay_gossip(&m, p); + Self::relay_gossip(&m, Some(block), p); } }, ); @@ -1273,7 +1278,7 @@ impl BeaconStateTile { }), Feedback::Accept(block_root) => { if do_relay { - Self::relay_gossip(&m, producers); + Self::relay_gossip(&m, None, producers); } self.on_accept(block_root, producers); } @@ -1282,7 +1287,7 @@ impl BeaconStateTile { } Feedback::AwaitParentPayload { .. } => { if do_relay { - Self::relay_gossip(&m, producers); + Self::relay_gossip(&m, None, producers); } self.park_block(feedback, BlockSourceMsg::Gossip(m), data, producers); } @@ -1294,13 +1299,14 @@ impl BeaconStateTile { true } - fn relay_gossip(m: &NewGossipMsg, producers: &mut Producers) { + fn relay_gossip(m: &NewGossipMsg, block: Option, producers: &mut Producers) { producers.produce(PeerEvent::SendGossip { originator_stream_id: m.stream_id, topic: m.topic, msg_hash: m.msg_hash, recv_ts: m.recv_ts, protobuf: m.protobuf, + block, }); } diff --git a/crates/beacon_state/tile/src/tile/orphan_pool.rs b/crates/beacon_state/tile/src/tile/orphan_pool.rs index 8f670c7a..93a3878c 100644 --- a/crates/beacon_state/tile/src/tile/orphan_pool.rs +++ b/crates/beacon_state/tile/src/tile/orphan_pool.rs @@ -233,7 +233,7 @@ impl BeaconStateTile { } let feedback = - self.apply_block(data, read, BlockSource::Rpc, pre_verified, producers, |_| {}); + self.apply_block(data, read, BlockSource::Rpc, pre_verified, producers, |_, _| {}); match feedback { Feedback::Accept(block_root) => self.on_accept(block_root, producers), Feedback::Reject(_) => producers.produce(PeerEvent::RpcMisbehaviour { diff --git a/crates/beacon_state/tile/src/tile/tests.rs b/crates/beacon_state/tile/src/tile/tests.rs index a8ffa9be..e1a4735b 100644 --- a/crates/beacon_state/tile/src/tile/tests.rs +++ b/crates/beacon_state/tile/src/tile/tests.rs @@ -10,8 +10,9 @@ use silver_beacon_state_data::{ StateReadView, ValSeed, Withdrawals, }; use silver_common::{ - BlockStage, EngineNewPayloadResp, GossipTopic, LOCAL_GOSSIP_STREAM_ID, MessageId, P2pStreamId, - PeerEvent, StreamProtocol, SyncNeed, TCache, TCacheProducer, TCacheRead, TProducer, + BlockStage, EngineNewPayloadResp, GossipBlock, GossipTopic, LOCAL_GOSSIP_STREAM_ID, MessageId, + P2pStreamId, PeerEvent, StreamProtocol, SyncNeed, TCache, TCacheProducer, TCacheRead, + TProducer, column_util::block_root_fulu, ssz_view::{ ATTESTATION_DATA_SIZE, AttestationView, BEACON_BLOCK_BODY_FIXED, BYTES_PER_KZG_COMMITMENT, @@ -124,6 +125,16 @@ fn make_tile_with_gossip( wall_slot: u64, state: BeaconState, ) -> (BeaconStateTile, TProducer, TProducer) { + let (tile, gossip, rpc, _replay) = + make_tile_with_producers(wall_slot, state, SpecConfig::mainnet()); + (tile, gossip, rpc) +} + +fn make_tile_with_producers( + wall_slot: u64, + state: BeaconState, + spec: SpecConfig, +) -> (BeaconStateTile, TProducer, TProducer, TProducer) { let secs_per_slot = 12u64; let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(); let genesis = now.saturating_sub(wall_slot * secs_per_slot + 1); @@ -138,7 +149,7 @@ fn make_tile_with_gossip( let replay_c = replay_p.cache_ref().random_access("test_replay_buf", true).unwrap(); let tile = BeaconStateTile::new( ticker, - Arc::new(SpecConfig::mainnet()), + Arc::new(spec), &SyncingConfig::default(), gossip_c, rpc_c, @@ -147,7 +158,7 @@ fn make_tile_with_gossip( true, state, ); - (tile, gossip_p, event_p) + (tile, gossip_p, event_p, replay_p) } /// Publish a minimal block (slot at offset 100) into `producer` and wrap it @@ -498,7 +509,7 @@ fn short_gossip_block_rejected_before_any_field_read() { BlockSource::Gossip, false, &mut adapter.producers, - |_| panic!("a malformed block must never be relayed"), + |_, _| panic!("a malformed block must never be relayed"), ); assert!(matches!(feedback, Feedback::Reject(None)), "len {len}: {feedback:?}"); } @@ -509,10 +520,14 @@ fn short_gossip_block_rejected_before_any_field_read() { bytes[100..108].copy_from_slice(&11u64.to_le_bytes()); bytes[116] = 0xFF; // unknown parent_root let (data, read) = publish_block_bytes(&mut gp, &bytes); - let feedback = - tile.apply_block(&data, read, BlockSource::Gossip, false, &mut adapter.producers, |_| { - panic!("an unimportable block must never be relayed") - }); + let feedback = tile.apply_block( + &data, + read, + BlockSource::Gossip, + false, + &mut adapter.producers, + |_, _| panic!("an unimportable block must never be relayed"), + ); assert!(matches!(feedback, Feedback::RequestParent { .. }), "{feedback:?}"); } @@ -586,7 +601,7 @@ fn a_block_already_in_fork_choice_is_reported_already_known() { let (data, read) = publish_block_bytes(&mut gp, &bytes); let feedback = - tile.apply_block(&data, read, BlockSource::Rpc, false, &mut adapter.producers, |_| { + tile.apply_block(&data, read, BlockSource::Rpc, false, &mut adapter.producers, |_, _| { panic!("a repeat is never relayed") }); assert_eq!(feedback, Feedback::AlreadyKnown(block_root)); @@ -618,13 +633,13 @@ fn a_block_is_applied_once_and_already_known_on_repeat() { // bypasses proposer-signature verification. let (data, read) = publish_block_bytes(&mut gp, &block_ssz); let feedback = - tile.apply_block(&data, read, BlockSource::Gossip, true, &mut adapter.producers, |_| {}); + tile.apply_block(&data, read, BlockSource::Gossip, true, &mut adapter.producers, |_, _| {}); let Feedback::Accept(Some(block_root)) = feedback else { panic!("{feedback:?}") }; assert_eq!(block_stages(&mut sink), [(block_root, BlockStage::Applied)]); let (data, read) = publish_block_bytes(&mut gp, &block_ssz); let feedback = - tile.apply_block(&data, read, BlockSource::Rpc, false, &mut adapter.producers, |_| { + tile.apply_block(&data, read, BlockSource::Rpc, false, &mut adapter.producers, |_, _| { panic!("a repeat is never relayed") }); assert_eq!(feedback, Feedback::AlreadyKnown(block_root)); @@ -663,7 +678,7 @@ fn payload_timestamp_and_blob_count_are_checked_before_relay() { BlockSource::Gossip, true, // pre_verified: the signature is not what these cases are about &mut adapter.producers, - |_| relayed = true, + |_, _| relayed = true, ); // The well-formed case still fails the STF (synthetic parent root), so @@ -743,7 +758,7 @@ fn non_canonical_body_is_rejected_before_relay() { BlockSource::Gossip, true, &mut adapter.producers, - |_| relayed = true, + |_, _| relayed = true, ); assert_eq!(matches!(feedback, Feedback::Reject(None)), want_reject, "{feedback:?}"); @@ -773,51 +788,51 @@ fn block_at_the_finalized_start_slot_is_ignored() { BlockSource::Gossip, true, &mut adapter.producers, - |_| {}, + |_, _| {}, ); assert_eq!(matches!(feedback, Feedback::Ignore), want_ignore, "slot {slot}: {feedback:?}"); } } -/// `proposer_lookahead` reaches only the parent's current and next epoch. A -/// block past a long skipped-slot run has no entry, so silver cannot check its -/// proposer — it still imports, but must not go on the wire as if it had. +/// Fixture bypass: signature checking is skipped to isolate proposer lookup. #[test] -fn block_past_the_lookahead_window_imports_but_is_not_relayed() { +fn block_relay_requires_a_resolved_proposer() { const PARENT_SLOT: u64 = 10; - - // The parent sits in epoch 0, so the lookahead index is the slot itself - // and the window is `[0, PROPOSER_LOOKAHEAD_SIZE)`. let last_in_window = PROPOSER_LOOKAHEAD_SIZE as u64 - 1; - for (slot, want_relay) in [ - (PARENT_SLOT + 1, true), - (last_in_window, true), - (last_in_window + 1, false), - (last_in_window + SLOTS_PER_EPOCH, false), - ] { - let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(slot); - seed_tile(&mut tile, 4, PARENT_SLOT); - tile.sync_target = SyncUpdate::Following; - - let stamp = slot * SpecConfig::mainnet().seconds_per_slot(); - let bytes = fulu_block_with_payload(slot, stamp, 0); - let (data, read) = publish_block_bytes(&mut gp, &bytes); - - let mut relayed = false; - let feedback = tile.apply_block( - &data, - read, - BlockSource::Gossip, - true, - &mut adapter.producers, - |_| relayed = true, - ); - - assert_eq!(relayed, want_relay, "slot {slot}: {feedback:?}"); - // Either way the block is not thrown away: the proposer window is our - // limit, not grounds to refuse the block. - assert!(!matches!(feedback, Feedback::Ignore), "slot {slot}: {feedback:?}"); + for target in + [SyncUpdate::Following, SyncUpdate::SyncingHead { head_slot: 400, head_root: [9; 32] }] + { + for (slot, want_relay) in [ + (PARENT_SLOT + 1, true), + (last_in_window, true), + (last_in_window + 1, false), + (last_in_window + SLOTS_PER_EPOCH, false), + ] { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(slot); + seed_tile(&mut tile, 4, PARENT_SLOT); + tile.sync_target = target; + adapter.consume(|_: PeerEvent, _| {}); + + let stamp = slot * SpecConfig::mainnet().seconds_per_slot(); + let bytes = fulu_block_with_payload(slot, stamp, 0); + let msg = gossip_msg(&mut gp, &bytes, GossipTopic::BeaconBlock); + assert!(tile.handle_gossip(msg.ssz, msg, true, true, &mut adapter.producers)); + + let mut relays = Vec::new(); + adapter.consume(|event: PeerEvent, _| { + if let PeerEvent::SendGossip { topic, block, .. } = event { + assert_eq!(topic, GossipTopic::BeaconBlock); + relays.push(block); + } + }); + let expected = GossipBlock { slot, block_root: block_root_fulu(&bytes) }; + assert_eq!( + relays, + if want_relay { vec![Some(expected)] } else { vec![] }, + "{target:?}, slot {slot}" + ); + } } } @@ -1187,7 +1202,7 @@ fn ve_accept() { seed_tile_with_keys(&mut tile, 4, 256 * SLOTS_PER_EPOCH); let imm = seed_immutable(&tile); let buf = test_signing::sign_voluntary_exit(0, 0, 0, &imm); - assert_eq!(tile.handle_voluntary_exit(&buf), Feedback::Accept(None)); + assert_non_block_relay(&mut tile, &buf, GossipTopic::VoluntaryExit); } #[test] @@ -1215,7 +1230,7 @@ fn ps_accept() { seed_tile_with_keys(&mut tile, 4, 0); let imm = seed_immutable(&tile); let buf = test_signing::sign_proposer_slashing(0, 0, 0, &imm); - assert_eq!(tile.handle_proposer_slashing(&buf), Feedback::Accept(None)); + assert_non_block_relay(&mut tile, &buf, GossipTopic::ProposerSlashing); } #[test] @@ -1278,7 +1293,7 @@ fn as_accept() { seed_tile_with_keys(&mut tile, 4, 0); let imm = seed_immutable(&tile); let buf = test_signing::sign_attester_slashing_double_vote(0, 0, 0, 0, &imm); - assert_eq!(tile.handle_attester_slashing(&buf), Feedback::Accept(None)); + assert_non_block_relay(&mut tile, &buf, GossipTopic::AttesterSlashing); } #[test] @@ -1332,7 +1347,7 @@ fn bls_change_accept() { let imm = seed_immutable(&tile); let to_addr = [0x42u8; 20]; let buf = test_signing::sign_bls_to_execution_change(0, 0, &to_addr, &imm); - assert_eq!(tile.handle_bls_to_execution_change(&buf), Feedback::Accept(None)); + assert_non_block_relay(&mut tile, &buf, GossipTopic::BlsToExecutionChange); } #[test] @@ -1460,6 +1475,8 @@ fn attestation_updates_vote_tracker() { assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_epoch, want_epoch); } +/// Fixture bypass: `protobuf` reuses the SSZ handle; beacon-state forwards it +/// without reading its bytes. fn gossip_msg(producer: &mut TProducer, bytes: &[u8], topic: GossipTopic) -> NewGossipMsg { let mut r = producer.reserve(bytes.len(), true).expect("reserve"); r.buffer().unwrap()[..bytes.len()].copy_from_slice(bytes); @@ -1524,10 +1541,13 @@ fn batched_att( fn attestation_batch_flush_applies_all() { let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); seed_tile_with_keys(&mut tile, 128, 0); + adapter.consume(|_: PeerEvent, _| {}); let bbr = tile.last_applied_block_root; + let mut expected_topics = Vec::new(); for vi in [0u32, 1] { let (buf, subnet) = batched_att(&tile, vi as usize, vi); + expected_topics.push(GossipTopic::BeaconAttestation(subnet)); let m = gossip_att_msg(&mut gp, &buf, subnet); tile.defer_vote(m, &mut adapter.producers); } @@ -1538,6 +1558,7 @@ fn attestation_batch_flush_applies_all() { assert_eq!(tile.fork_choice.vote_tracker.votes[1].latest_root, bbr); assert!(tile.vote_batch.is_empty()); assert!(tile.vote_pending.is_empty()); + assert_eq!(non_block_relays(&mut adapter), expected_topics); } /// A forged signature (valid G2 point, wrong key) fails the batch verify; @@ -1632,6 +1653,7 @@ fn sync_message_batch_applies_and_marks_seen() { let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); seed_tile_with_keys(&mut tile, 128, 0); let imm = seed_immutable(&tile); + adapter.consume(|_: PeerEvent, _| {}); let bbr = tile.last_applied_block_root; let wall = tile.ticker.current_slot(); @@ -1649,6 +1671,7 @@ fn sync_message_batch_applies_and_marks_seen() { // 128 positions of each subcommittee. assert_eq!(SyncCommitteeContributionView::aggregation_bits(&contribution), &[0xff; 16]); assert!(tile.vote_batch.is_empty() && tile.vote_pending.is_empty()); + assert_eq!(non_block_relays(&mut adapter), [GossipTopic::SyncCommittee(1)]); } #[test] @@ -1797,7 +1820,7 @@ fn sync_contribution_accepted_then_superset_ignored() { let bbr = tile.last_applied_block_root; let buf = test_signing::sign_contribution_and_proof(0, 0, slot, sub, 3, 0, bbr, &imm); - assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Accept(None))); + assert_non_block_relay(&mut tile, &buf, GossipTopic::SyncCommitteeContributionAndProof); assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Ignore)); } @@ -1881,6 +1904,7 @@ fn ptc_vote_records_every_matching_committee_position() { let slot = 31; let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(slot); seed_tile_with_keys(&mut tile, 128, slot); + adapter.consume(|_: PeerEvent, _| {}); let root = tile.last_applied_block_root; let msg = test_signing::sign_payload_attestation_message( 0, @@ -1908,6 +1932,7 @@ fn ptc_vote_records_every_matching_committee_position() { assert!( tile.fork_choice.ptc_data_availability_votes(&root).iter().all(|vote| *vote == Some(true)) ); + assert_eq!(non_block_relays(&mut adapter), [GossipTopic::PayloadAttestationMessage]); } /// Spec `validate_on_attestation`: a single attestation for a block we @@ -2234,7 +2259,7 @@ fn agg_accept() { let buf = build_agg_for_vi0(&tile); let beacon_block_root = tile.last_applied_block_root; let slot = SignedAggregateAndProofView::agg_slot(&buf); - assert_eq!(tile.handle_aggregate_and_proof(&buf), Feedback::Accept(None)); + assert_non_block_relay(&mut tile, &buf, GossipTopic::BeaconAggregateAndProof); assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, beacon_block_root); assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_epoch, slot / SLOTS_PER_EPOCH); } @@ -3236,3 +3261,28 @@ fn finalize_promotes_every_tier_into_checkpoint_encode() { ); } } + +#[cfg(feature = "ef_tests")] +mod block_relay; + +fn non_block_relays(adapter: &mut SpineAdapter) -> Vec { + let mut topics = Vec::new(); + adapter.consume(|event: PeerEvent, _| { + if let PeerEvent::SendGossip { topic, block, .. } = event { + assert_eq!(block, None, "{topic:?} carries no block metadata"); + topics.push(topic); + } + }); + topics +} + +fn assert_non_block_relay(tile: &mut BeaconStateTile, bytes: &[u8], topic: GossipTopic) { + let mut gossip = TCache::producer("non_block_relay", 1 << 20); + tile.gossip_consumer = gossip.cache_ref().random_access("non_block_relay", true).unwrap(); + let (_spine, mut adapter) = spine_adapter(tile); + adapter.consume(|_: PeerEvent, _| {}); + let msg = gossip_msg(&mut gossip, bytes, topic); + tile.on_gossip(msg, &mut adapter.producers); + tile.flush_votes(&mut adapter.producers); + assert_eq!(non_block_relays(&mut adapter), [topic]); +} diff --git a/crates/beacon_state/tile/src/tile/tests/block_relay.rs b/crates/beacon_state/tile/src/tile/tests/block_relay.rs new file mode 100644 index 00000000..137fe0a2 --- /dev/null +++ b/crates/beacon_state/tile/src/tile/tests/block_relay.rs @@ -0,0 +1,288 @@ +use super::*; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Receipt { + slot: Slot, + block_root: B256, + stage: BlockStage, + source: BlockSource, +} + +struct Published { + events: Vec, + relays: Vec, +} + +impl Published { + fn drain(sink: &mut SpineAdapter) -> Self { + let mut events = Vec::new(); + sink.consume(|event: BeaconStateEvent, _| events.push(event)); + let mut relays = Vec::new(); + sink.consume(|event: PeerEvent, _| { + if let PeerEvent::SendGossip { topic, block, .. } = event { + assert_eq!(topic, GossipTopic::BeaconBlock); + relays.push(block.expect("every block relay carries its metadata")); + } + }); + Self { events, relays } + } + + fn receipts(&self) -> Vec { + self.events + .iter() + .filter_map(|event| match *event { + BeaconStateEvent::BlockReceived { slot, block_root, stage, source, .. } => { + Some(Receipt { slot, block_root, stage, source }) + } + _ => None, + }) + .collect() + } +} + +fn fulu_from_genesis() -> SpecConfig { + SpecConfig { fulu_fork_epoch: 0, ..SpecConfig::mainnet() } +} + +fn sanity_file(name: &str, file: &str) -> Vec { + let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("consensus-spec-tests/tests/mainnet/fulu/sanity/blocks/pyspec_tests") + .join(name) + .join(file); + let raw = fs::read(&path) + .unwrap_or_else(|e| panic!("{}: {e} (run `just ef-tests-download`)", path.display())); + snap::Decoder::new().decompress_vec(&raw).unwrap_or_else(|e| panic!("{}: {e}", path.display())) +} + +fn sanity_fixture(name: &str) -> (Vec, Vec) { + (sanity_file(name, "pre.ssz_snappy"), sanity_file(name, "blocks_0.ssz_snappy")) +} + +struct BlockPublications { + tile: BeaconStateTile, + gossip: TProducer, + rpc: TProducer, + replay: TProducer, + adapter: SpineAdapter, + sink: SpineAdapter, + _spine: Box, +} + +impl BlockPublications { + fn new(pre_ssz: &[u8], block_ssz: &[u8], target: SyncUpdate) -> Self { + let state = BeaconState::from_checkpoint(pre_ssz, &fulu_from_genesis(), &[]) + .unwrap_or_else(|e| panic!("decompose checkpoint: {e}")); + let block_slot = SignedBeaconBlockView::slot(block_ssz); + let (mut tile, gossip, rpc, replay) = + make_tile_with_producers(block_slot + 1, state, fulu_from_genesis()); + tile.sync_target = target; + let (mut spine, adapter) = spine_adapter(&tile); + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine); + sink.consume(|_: BeaconStateEvent, _| {}); + sink.consume(|_: PeerEvent, _| {}); + Self { tile, gossip, rpc, replay, adapter, sink, _spine: spine } + } + + fn on_gossip(&mut self, block_ssz: &[u8]) { + let m = gossip_msg(&mut self.gossip, block_ssz, GossipTopic::BeaconBlock); + self.tile.on_gossip(m, &mut self.adapter.producers); + } + + fn on_gossip_unrelayed(&mut self, block_ssz: &[u8]) { + let m = gossip_msg(&mut self.gossip, block_ssz, GossipTopic::BeaconBlock); + self.tile.handle_gossip(m.ssz, m, false, false, &mut self.adapter.producers); + } + + fn on_rpc_block(&mut self, block_ssz: &[u8]) { + let (_, read) = publish_block_bytes(&mut self.rpc, block_ssz); + self.tile.on_rpc_inbound(live_block_response(read), &mut self.adapter.producers); + } + + fn on_replay(&mut self, block_ssz: &[u8]) { + let (_, read) = publish_block_bytes(&mut self.replay, block_ssz); + self.tile.on_replay(ReplayBlock::Block { ssz: read }, &mut self.adapter.producers); + } + + fn drain(&mut self) -> Published { + Published::drain(&mut self.sink) + } +} + +fn stages_of(receipts: &[Receipt]) -> Vec { + receipts.iter().map(|r| r.stage).collect() +} + +fn fulu_gossip_block(bytes: &[u8]) -> GossipBlock { + GossipBlock { slot: SignedBeaconBlockView::slot(bytes), block_root: block_root_fulu(bytes) } +} + +#[test] +fn gossip_relay_carries_block_metadata_in_either_sync_mode() { + let (pre_ssz, block_ssz) = sanity_fixture("attestation"); + let expected = fulu_gossip_block(&block_ssz); + for target in + [SyncUpdate::Following, SyncUpdate::SyncingHead { head_slot: 400, head_root: [9; 32] }] + { + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, target); + rig.on_gossip(&block_ssz); + + let published = rig.drain(); + assert_eq!(published.relays, [expected], "{target:?}"); + if target.is_following() { + assert!(published.receipts().contains(&Receipt { + slot: expected.slot, + block_root: expected.block_root, + stage: BlockStage::Applied, + source: BlockSource::Gossip, + })); + } + } +} + +#[test] +fn an_rpc_block_is_imported_without_a_gossip_notification() { + let (pre_ssz, block_ssz) = sanity_fixture("attestation"); + let expected = fulu_gossip_block(&block_ssz); + for target in + [SyncUpdate::Following, SyncUpdate::SyncingHead { head_slot: 400, head_root: [9; 32] }] + { + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, target); + rig.on_rpc_block(&block_ssz); + + let published = rig.drain(); + assert!(published.relays.is_empty(), "{target:?}: RPC never requests relay"); + assert!( + published + .receipts() + .iter() + .any(|r| { r.block_root == expected.block_root && r.stage == BlockStage::Applied }), + "{target:?}: the RPC block was imported" + ); + } +} + +#[test] +fn a_blob_block_is_relayed_once_across_staging_and_import() { + let (pre_ssz, block_ssz) = sanity_fixture("one_blob"); + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, SyncUpdate::Following); + let expected = fulu_gossip_block(&block_ssz); + + rig.on_gossip(&block_ssz); + let published = rig.drain(); + assert_eq!(published.relays, [expected]); + assert_eq!(stages_of(&published.receipts()), [BlockStage::AwaitData]); + + rig.on_gossip(&block_ssz); + let repeated = rig.drain(); + assert!(repeated.relays.is_empty()); + + rig.tile.handle_data_columns_available( + expected.block_root, + expected.slot, + &mut rig.adapter.producers, + ); + let imported = rig.drain(); + assert_eq!(stages_of(&imported.receipts()), [BlockStage::Applied]); + assert!(imported.relays.is_empty(), "DA completion does not relay the block again"); +} + +/// `drain_awaiting_payload` disables relay when retrying a Gloas block. +/// No fixture covers that retry, so this checks the handler's flag directly. +#[test] +fn disabling_relay_suppresses_the_gossip_notification() { + let (pre_ssz, block_ssz) = sanity_fixture("attestation"); + let expected = fulu_gossip_block(&block_ssz); + for target in + [SyncUpdate::Following, SyncUpdate::SyncingHead { head_slot: 400, head_root: [9; 32] }] + { + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, target); + rig.on_gossip_unrelayed(&block_ssz); + + let published = rig.drain(); + assert!(published.relays.is_empty(), "{target:?}: relay is disabled"); + if target.is_following() { + assert!( + published.receipts().iter().any(|r| { + r.block_root == expected.block_root && r.stage == BlockStage::Applied + }), + "disabling relay still imports the block" + ); + } + } +} + +/// A missing parent fails precheck before signature verification. The parent's +/// import retries the child through the block handler. +#[test] +fn a_parked_block_is_relayed_by_the_retry_that_admits_it() { + let (pre_ssz, first) = sanity_fixture("attestation"); + let second = sanity_file("attestation", "blocks_1.ssz_snappy"); + let mut rig = BlockPublications::new(&pre_ssz, &second, SyncUpdate::Following); + + rig.on_gossip(&second); + let parked = rig.drain(); + assert_eq!(stages_of(&parked.receipts()), [BlockStage::AwaitParent]); + assert!(parked.relays.is_empty(), "the missing parent prevents validation"); + let child = parked.receipts()[0].block_root; + + rig.on_gossip(&first); + let released = rig.drain(); + assert_eq!(released.relays, [fulu_gossip_block(&first), fulu_gossip_block(&second)]); + let of_child = + released.receipts().into_iter().filter(|r| r.block_root == child).collect::>(); + assert_eq!(stages_of(&of_child), [BlockStage::Applied]); +} + +#[test] +fn a_relay_request_does_not_imply_successful_import() { + let pre_ssz = sanity_file("invalid_incorrect_state_root", "pre.ssz_snappy"); + let block_ssz = sanity_file("invalid_incorrect_state_root", "blocks_0.ssz_snappy"); + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, SyncUpdate::Following); + + rig.on_gossip(&block_ssz); + let published = rig.drain(); + let expected = fulu_gossip_block(&block_ssz); + assert_eq!(published.relays, [expected]); + assert!(published.receipts().is_empty()); + assert!( + published.events.iter().any(|event| matches!(event, + BeaconStateEvent::BlockRejected { block_root, source: BlockSource::Gossip } + if *block_root == expected.block_root + )), + "state transition rejected the relayed block" + ); +} + +#[test] +fn a_block_with_a_bad_signature_reports_nothing() { + let (pre_ssz, block_ssz) = sanity_fixture("attestation"); + let mut forged = block_ssz.clone(); + forged[4] ^= 0xFF; // the proposer signature occupies [4..100) + + for target in + [SyncUpdate::Following, SyncUpdate::SyncingHead { head_slot: 400, head_root: [9; 32] }] + { + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, target); + rig.on_gossip(&forged); + let published = rig.drain(); + assert!(published.receipts().is_empty(), "{target:?}"); + assert!(published.relays.is_empty(), "{target:?}: a bad signature cannot be relayed"); + } +} + +#[test] +fn a_replayed_block_reports_no_gossip() { + let (pre_ssz, block_ssz) = sanity_fixture("attestation"); + let mut rig = BlockPublications::new(&pre_ssz, &block_ssz, SyncUpdate::Following); + + rig.on_replay(&block_ssz); + + assert_eq!( + rig.tile.head_state_slot(), + SignedBeaconBlockView::slot(&block_ssz), + "replay imported the block" + ); + let published = rig.drain(); + assert!(published.receipts().is_empty(), "replay emits no block receipts"); + assert!(published.relays.is_empty(), "replay requests no gossip relay"); +} diff --git a/crates/columns/src/tile.rs b/crates/columns/src/tile.rs index 09409c41..2bbde276 100644 --- a/crates/columns/src/tile.rs +++ b/crates/columns/src/tile.rs @@ -529,6 +529,7 @@ impl DataColumnsTile { msg_hash, recv_ts, protobuf, + block: None, }); } RelayMeta::Rpc { ssz } if self.sync_state.is_synced() => { diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index ccf10fc0..045d723b 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -23,16 +23,17 @@ pub use crate::{ EngineGetPayloadBodiesByRangeReq, EngineGetPayloadBodiesResp, EngineGetPayloadReq, EngineGetPayloadResp, EngineHealthEvent, EngineNewPayloadEnvelopeReq, EngineNewPayloadReq, EngineNewPayloadResp, EnginePreparePayloadReq, EngineReq, EngineResp, Error as TCacheError, - GossipMsgIn, GossipMsgOut, IpBytes, LOCAL_GOSSIP_STREAM_ID, LocalAttestationFailure, - LocalAttestationResult, MAX_BLOBS_PER_BLOCK, MAX_PAYLOAD_BODIES_PER_REQ, MULTISTREAM_V1, - MultiProducer as TMultiProducer, NewGossipMsg, P2pConnectionStats, P2pSend, P2pStreamId, - PREFILL_SLOTS, PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, PeerStats, - PeerStatus, PeerTopicScores, Prefill, Producer as TProducer, REJECT_RESPONSE, - RPC_PROTOCOLS, RandomAccessConsumer as TRandomAccess, ReplayBlock, - Reservation as TReservation, RpcInbound, RpcOutbound, RpcRequest, RpcRequestInbound, - RpcRequestOutbound, RpcResponse, RpcResponseInbound, RpcResponseOutbound, RpcSeverity, - SilverSpine, SilverSpineProducers, StreamProtocol, SyncNeed, SyncUpdate, SyncingStrategy, - TCache, TCacheProducer, TCacheRead, TCacheRef, WithdrawalInline, + GossipBlock, GossipMsgIn, GossipMsgOut, IpBytes, LOCAL_GOSSIP_STREAM_ID, + LocalAttestationFailure, LocalAttestationResult, MAX_BLOBS_PER_BLOCK, + MAX_PAYLOAD_BODIES_PER_REQ, MULTISTREAM_V1, MultiProducer as TMultiProducer, NewGossipMsg, + P2pConnectionStats, P2pSend, P2pStreamId, PREFILL_SLOTS, PayloadValidationStatus, + PeerControl, PeerEvent, PeerScores, PeerStats, PeerStatus, PeerTopicScores, Prefill, + Producer as TProducer, REJECT_RESPONSE, RPC_PROTOCOLS, + RandomAccessConsumer as TRandomAccess, ReplayBlock, Reservation as TReservation, + RpcInbound, RpcOutbound, RpcRequest, RpcRequestInbound, RpcRequestOutbound, RpcResponse, + RpcResponseInbound, RpcResponseOutbound, RpcSeverity, SilverSpine, SilverSpineProducers, + StreamProtocol, SyncNeed, SyncUpdate, SyncingStrategy, TCache, TCacheProducer, TCacheRead, + TCacheRef, WithdrawalInline, }, util::{create_self_signed_certificate, decode_varint, encode_varint, hex32}, wheel::Wheel, diff --git a/crates/common/src/spine.rs b/crates/common/src/spine.rs index 543084ab..b391a05d 100644 --- a/crates/common/src/spine.rs +++ b/crates/common/src/spine.rs @@ -7,8 +7,8 @@ pub use messages::{ EngineGetBlobsResp, EngineGetPayloadBodiesByHashReq, EngineGetPayloadBodiesByRangeReq, EngineGetPayloadBodiesResp, EngineGetPayloadReq, EngineGetPayloadResp, EngineHealthEvent, EngineNewPayloadEnvelopeReq, EngineNewPayloadReq, EngineNewPayloadResp, - EnginePreparePayloadReq, EngineReq, EngineResp, GossipMsgIn, GossipMsgOut, IpBytes, - LocalAttestationFailure, LocalAttestationResult, MAX_BLOBS_PER_BLOCK, + EnginePreparePayloadReq, EngineReq, EngineResp, GossipBlock, GossipMsgIn, GossipMsgOut, + IpBytes, LocalAttestationFailure, LocalAttestationResult, MAX_BLOBS_PER_BLOCK, MAX_PAYLOAD_BODIES_PER_REQ, NewGossipMsg, P2pConnectionStats, P2pSend, PREFILL_SLOTS, PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, PeerStats, PeerStatus, PeerTopicScores, Prefill, ReplayBlock, RpcInbound, RpcOutbound, RpcRequest, RpcRequestInbound, diff --git a/crates/common/src/spine/messages.rs b/crates/common/src/spine/messages.rs index 44e82fa1..ac3e6c66 100644 --- a/crates/common/src/spine/messages.rs +++ b/crates/common/src/spine/messages.rs @@ -5,7 +5,7 @@ use std::{ }; use flux::timing::Nanos; -use silver_beacon_state_data::SLOTS_PER_EPOCH; +use silver_beacon_state_data::{B256, SLOTS_PER_EPOCH}; use crate::{ DataKind, Enr, GossipTopic, Identify, MessageId, Origin, P2pStreamId, PeerId, StreamProtocol, @@ -318,6 +318,13 @@ impl RpcOutbound { } } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(C)] +pub struct GossipBlock { + pub slot: u64, + pub block_root: B256, +} + #[derive(Clone, Copy, Debug)] #[repr(C, u8)] #[allow(clippy::large_enum_variant)] @@ -466,6 +473,7 @@ pub enum PeerEvent { msg_hash: MessageId, recv_ts: Nanos, protobuf: TCacheRead, + block: Option, }, /// Misbehaviour observed on the RPC (req/resp) sub-protocol. The peer /// manager translates `severity` into a P5 application-score delta; diff --git a/crates/control/src/tile.rs b/crates/control/src/tile.rs index 4ea64fd2..199b2b1b 100644 --- a/crates/control/src/tile.rs +++ b/crates/control/src/tile.rs @@ -14,6 +14,9 @@ use crate::sync_engine::{SyncAction, SyncEngine}; const PEER_PERSIST_INTERVAL: Duration = Duration::from_secs(300); +#[cfg(test)] +mod tests; + pub struct Controller { peer_manager: PeerManager, gossip_handler: GossipHandler, @@ -184,6 +187,7 @@ impl Tile for Controller { msg_hash, recv_ts: Nanos::now(), protobuf, + block: None, }, now, &mut |evt| { @@ -210,6 +214,7 @@ impl Tile for Controller { msg_hash, recv_ts: _, protobuf, + block: _, } = &event { self.gossip_handler.mcache_insert(*msg_hash, *topic, *protobuf); diff --git a/crates/control/src/tile/tests.rs b/crates/control/src/tile/tests.rs new file mode 100644 index 00000000..de512aa2 --- /dev/null +++ b/crates/control/src/tile/tests.rs @@ -0,0 +1,153 @@ +use std::{ + io::Write, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, +}; + +use silver_chain_spec::SpecConfig; +use silver_common::{ + GossipBlock, GossipMsgIn, GossipMsgOut, IpBytes, Keypair, MessageId, P2pStreamId, PeerId, + StreamProtocol, TCache, TCacheProducer, TCacheRead, TProducer, +}; +use silver_peer::SyncingConfig; + +use super::*; + +struct GossipPublications { + controller: Controller, + adapter: SpineAdapter, + observer: SpineAdapter, + incoming: TProducer, + payload: TCacheRead, + outbound: TRandomAccess, + _spine: Box, +} + +struct Observer; +impl Tile for Observer { + fn loop_body(&mut self, _: &mut SpineAdapter) {} +} + +impl GossipPublications { + fn new(topic: GossipTopic, bytes: &[u8]) -> Self { + let incoming = TCache::producer("publication_in", 1 << 16); + let rpc = TCache::producer("publication_rpc", 1 << 16); + let mut protobuf = TCache::producer("publication_out", 1 << 16); + let payload = write_bytes(&mut protobuf, bytes); + let outbound = protobuf.cache_ref().random_access("publication_observer", true).unwrap(); + let controller = Controller::new( + PeerManager::new( + PeerId::default(), + vec![], + vec![topic], + Default::default(), + SyncingConfig::default(), + [0; 4], + [0; METADATA_SIZE], + 0, + ), + GossipHandler::new( + incoming.cache_ref().random_access("publication_in", true).unwrap(), + TCache::producer("publication_ssz", 1 << 16), + protobuf, + "00000000".to_owned(), + ) + .unwrap(), + TCache::multi_producer("publication_rpc_out", 1 << 16), + rpc.cache_ref().random_access("publication_rpc", true).unwrap(), + SyncEngine::new(SyncingConfig::default(), false, 0, Arc::new(SpecConfig::mainnet())), + ); + static SEQUENCE: AtomicU64 = AtomicU64::new(0); + let base = std::env::temp_dir().join(format!( + "silver-publications-{}-{}", + std::process::id(), + SEQUENCE.fetch_add(1, Ordering::Relaxed), + )); + std::fs::create_dir_all(&base).unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(&base, None)); + let adapter = SpineAdapter::connect_tile(&controller, &mut spine); + let mut observer = SpineAdapter::connect_tile(&Observer, &mut spine); + observer.consume(|_: P2pSend, _| {}); + let mut capture = + Self { controller, adapter, observer, incoming, payload, outbound, _spine: spine }; + capture.crank(); + for peer in 1..=2u8 { + capture.observer.produce(PeerEvent::P2pNewConnection { + p2p_peer_id: peer as usize, + peer_id_full: Keypair::from_secret(&[peer; 32]).unwrap().peer_id(), + ip: IpBytes::V4([10, 0, 0, peer]), + port: 4000 + peer as u16, + local_dial: false, + }); + capture + .observer + .produce(PeerEvent::P2pGossipTopicSubscribe { p2p_peer: peer as usize, topic }); + } + capture.crank(); + capture.sent(); + capture + } + + fn crank(&mut self) { + self.controller.loop_body(&mut self.adapter); + } + + fn sent(&mut self) -> Vec<(usize, Vec)> { + let mut frames = Vec::new(); + self.observer.consume(|event: P2pSend, _| { + if let P2pSend::Gossip(GossipMsgOut { peer_id, tcache }) = event { + let read = self.outbound.acquire(tcache); + frames.push((peer_id, read.buffer().unwrap().0.to_vec())); + } + }); + frames + } + + fn iwant(&mut self, peer: usize, hash: MessageId) { + let stream = P2pStreamId::new(peer, 0, StreamProtocol::GossipSub, true); + let mut bytes = stream.as_ref().to_vec(); + // RPC.control → ControlMessage.iwant → ControlIWant.message_ids, each + // length-delimited. + bytes.extend_from_slice(&[0x1a, 24, 0x12, 22, 0x0a, 20]); + bytes.extend_from_slice(&hash.id); + let tcache = write_bytes(&mut self.incoming, &bytes); + self.observer.produce(GossipMsgIn { p2p_id: stream, tcache }); + self.crank(); + } +} + +fn write_bytes(producer: &mut TProducer, bytes: &[u8]) -> TCacheRead { + let mut reservation = producer.reserve(bytes.len(), false).unwrap(); + reservation.write_all(bytes).unwrap(); + reservation.flush().unwrap(); + reservation.read() +} + +#[test] +fn relay_metadata_preserves_routing_and_iwant_service() { + // Forwarding and IWANT service treat the payload as opaque bytes. + let bytes = b"relay payload"; + let hash = MessageId { id: [0xCD; 20] }; + for block in [None, Some(GossipBlock { slot: 37, block_root: [0xAB; 32] })] { + let mut capture = GossipPublications::new(GossipTopic::BeaconBlock, bytes); + capture.observer.produce(PeerEvent::SendGossip { + originator_stream_id: P2pStreamId::new(1, 0, StreamProtocol::GossipSub, true), + topic: GossipTopic::BeaconBlock, + msg_hash: hash, + recv_ts: Nanos::now(), + protobuf: capture.payload, + block, + }); + capture.crank(); + assert_eq!(capture.sent(), [(2, bytes.to_vec())], "the sender is excluded"); + + capture.iwant(1, hash); + assert_eq!( + capture.sent(), + [(1, bytes.to_vec())], + "the relay request populated the message cache" + ); + } +} diff --git a/crates/peer/src/manager/mod.rs b/crates/peer/src/manager/mod.rs index 62a6b492..81af2965 100644 --- a/crates/peer/src/manager/mod.rs +++ b/crates/peer/src/manager/mod.rs @@ -424,6 +424,7 @@ impl PeerManager { msg_hash, recv_ts: _, protobuf, + block: _, } => { // TODO recv_ts elapsed metric self.on_send_gossip(originator_stream_id.peer(), msg_hash, topic, protobuf, emit); diff --git a/crates/peer/src/manager/promises.rs b/crates/peer/src/manager/promises.rs index 9b869e02..a024e592 100644 --- a/crates/peer/src/manager/promises.rs +++ b/crates/peer/src/manager/promises.rs @@ -342,7 +342,7 @@ impl PeerManager { mod tests { use std::time::Duration; - use silver_common::{PeerEvent, TCacheProducer}; + use silver_common::{GossipBlock, PeerEvent, TCacheProducer}; use silver_config::ScoreParams; use super::*; @@ -1026,6 +1026,7 @@ mod tests { msg_hash: hash, recv_ts: silver_common::Nanos::now(), protobuf: mk_tcache_read(), + block: Some(GossipBlock { slot: 37, block_root: [0xAB; 32] }), }, now, &mut |c| cap.0.push(c), @@ -1079,6 +1080,7 @@ mod tests { msg_hash: silver_common::MessageId { id: [0xCD; 20] }, recv_ts: silver_common::Nanos::now(), protobuf: mk_tcache_read(), + block: None, }, now, &mut |event| cap.0.push(event), From 719476262daad0323e8121a00226f0399ccf73a8 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Thu, 10 Sep 2026 18:49:36 +0100 Subject: [PATCH 2/2] Serve block_gossip events over SSE Publish block_gossip from block metadata on existing SendGossip requests. The boundary consumes PeerEvent outside the engine capacity gate. It selects the metadata without reading the gossip payload. This deliberately narrows the Beacon API's validation contract. RPC imports remain silent because they do not request gossip publication. A publication request does not guarantee delivery to peers. Keep block tied to Applied receipts. Repeated publication requests produce repeated events, and late subscribers receive no event replay. The two spine queues establish no shared ordering. Wire event: block_gossip Data: {"slot":"","block":"0x<64 lowercase hex>"} Amend ADR-0004 with the publication contract and its exclusions. Subscribers to both topics can reach the existing send cap sooner. Message layouts and the peer_events ring are unchanged. Extend topic parsing coverage with standalone and mixed subscriptions. Check the renderer's required fields as parsed JSON. Socket tests enter through the boundary's loop_body with spine inputs. One scenario checks separate and mixed subscriptions, repeated requests, and isolation from unrelated publications and block imports. Trailing events keep unwanted publications within the events read. The existing block test retains its receipt-stage coverage. A late subscriber connects after an earlier subscriber receives an event. Another test serves gossip notifications with unanswered FCU requests occupying engine capacity. Startup requests receive responses. Opaque payload bytes avoid coupling these fixtures to block serialization. The test client decodes HTTP and SSE before checking event names and required JSON fields. It permits additional fields and does not require an ordering between the two topics. A temporary renderer with reordered keys, extra whitespace, and an additional nested field passed all 1,361 tests. Two fault controls each failed one test, with just nextest exiting 100: - Publishing Applied imports on both topics failed block_subscriptions_select_imports_and_preserve_repeated_relay_requests. - Gating the peer-event consumer on engine capacity failed a_gossip_event_is_served_while_the_engine_pool_is_saturated. All controls were restored. Final validation, each exiting zero: - just fmt-check - just clippy - just nextest: 1,361 passed, 5 skipped - git diff --check Assisted-by: Codex:gpt-6-astra --- Cargo.lock | 1 + crates/application_boundary/Cargo.toml | 1 + crates/application_boundary/src/lib.rs | 18 +- crates/application_boundary/tests/tile.rs | 310 ++++++++++++++++++---- crates/beacon_api/src/events.rs | 16 +- crates/beacon_api/src/json.rs | 17 ++ crates/beacon_api/src/server.rs | 7 + docs/adr/0004-sync-materialized-api.md | 14 + 8 files changed, 325 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4233c92e..6ad2bf88 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4546,6 +4546,7 @@ dependencies = [ "silver_engine_api", "silver_httpcore", "tempfile", + "ureq", ] [[package]] diff --git a/crates/application_boundary/Cargo.toml b/crates/application_boundary/Cargo.toml index d123bea8..6ff861b9 100644 --- a/crates/application_boundary/Cargo.toml +++ b/crates/application_boundary/Cargo.toml @@ -19,6 +19,7 @@ hex.workspace = true serde_json.workspace = true silver_engine_api = { workspace = true, features = ["test-el"] } tempfile = "3" +ureq.workspace = true [lints] workspace = true diff --git a/crates/application_boundary/src/lib.rs b/crates/application_boundary/src/lib.rs index b439b2cf..502883c5 100644 --- a/crates/application_boundary/src/lib.rs +++ b/crates/application_boundary/src/lib.rs @@ -4,8 +4,8 @@ use flux::{spine::SpineAdapter, tile::Tile}; use silver_beacon_api::{BeaconApi, SlotStatus}; use silver_beacon_state_data::{BeaconStateReader, SpecConfig}; use silver_common::{ - BeaconStateEvent, BlockStage, Enr, Identify, Keypair, SilverSpine, SyncUpdate, TProducer, - TRandomAccess, + BeaconStateEvent, BlockStage, Enr, GossipBlock, Identify, Keypair, PeerEvent, SilverSpine, + SyncUpdate, TProducer, TRandomAccess, }; use silver_config::EngineConfig; use silver_engine_api::EngineApi; @@ -83,10 +83,9 @@ impl ApplicationBoundaryTile { fn consume_spine_events(&mut self, adapter: &mut SpineAdapter) { let beacon = &mut self.beacon; - // Consumed every iteration, and never behind the engine's capacity - // gate: a consumer's first `consume` jumps its cursor to the - // producer's write head, so a queue left unread while the pool is - // saturated loses everything published in the meantime. + // A consumer's first consume starts at the producer's write head. + // Keep both event queues active during engine saturation; delaying + // their first consume would discard notifications already queued. adapter.consume(|event: BeaconStateEvent, _| match event { BeaconStateEvent::Status { latest_block_slot, wall_slot, head_optimistic, .. } => { beacon.node_status_mut().slots = @@ -100,6 +99,13 @@ impl ApplicationBoundaryTile { } => beacon.publish_block(slot, &block_root), _ => {} }); + adapter.consume(|event: PeerEvent, _| { + if let PeerEvent::SendGossip { block: Some(GossipBlock { slot, block_root }), .. } = + event + { + beacon.publish_block_gossip(slot, &block_root); + } + }); let status = beacon.node_status_mut(); adapter.consume(|update: SyncUpdate, _| { status.syncing = !matches!(update, SyncUpdate::Following); diff --git a/crates/application_boundary/tests/tile.rs b/crates/application_boundary/tests/tile.rs index 8fceb7ba..1a1f03fd 100644 --- a/crates/application_boundary/tests/tile.rs +++ b/crates/application_boundary/tests/tile.rs @@ -1,19 +1,21 @@ use std::{ - io::{Read, Write}, + io::{BufRead, BufReader, Read, Write}, net::{SocketAddr, TcpStream}, os::unix::net::UnixStream, - sync::mpsc::{self, Receiver}, + sync::mpsc::{self, Receiver, TryRecvError}, thread::JoinHandle, time::{Duration, Instant}, }; -use flux::{spine::SpineAdapter, tile::Tile}; +use flux::{spine::SpineAdapter, tile::Tile, timing::Nanos}; +use serde_json::Value; use silver_application_boundary::ApplicationBoundaryTile; use silver_beacon_api::SlotStatus; use silver_beacon_state_data::{BeaconStateOwner, SpecConfig}; use silver_common::{ BeaconStateEvent, BlockSource, BlockStage, ELSyncStatus, EngineFcuReq, EngineReq, EngineResp, - Enr, Identify, Keypair, PayloadValidationStatus, SilverSpine, SyncUpdate, TCache, + Enr, GossipBlock, GossipTopic, Identify, Keypair, MessageId, P2pStreamId, + PayloadValidationStatus, PeerEvent, SilverSpine, StreamProtocol, SyncUpdate, TCache, TCacheProducer, ssz_view::STATUS_V2_SIZE, }; use silver_config::EngineConfig; @@ -135,8 +137,6 @@ fn drain_fcu_completions( }); } -const SSE_HEAD: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nX-Accel-Buffering: no\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n"; - fn block_received(slot: u64, byte: u8, stage: BlockStage) -> BeaconStateEvent { BeaconStateEvent::BlockReceived { slot, @@ -147,35 +147,106 @@ fn block_received(slot: u64, byte: u8, stage: BlockStage) -> BeaconStateEvent { } } -fn block_frame(slot: u64, byte: u8) -> Vec { - let data = format!( - "event: block\ndata: {{\"slot\":\"{slot}\",\"block\":\"0x{}\",\"execution_optimistic\":true}}\n\n", - hex::encode([byte; 32]) - ); - let mut frame = format!("{:x}\r\n", data.len()).into_bytes(); - frame.extend_from_slice(data.as_bytes()); - frame.extend_from_slice(b"\r\n"); - frame +fn block_relay(slot: u64, byte: u8) -> PeerEvent { + // SSE uses the metadata, so the fixture needs no encoded gossip object. + let payload = b"opaque relay payload"; + let mut producer = TCache::producer("cs_relay_metadata", 1 << 12); + let mut reservation = producer.reserve(payload.len(), false).unwrap(); + reservation.write_all(payload).unwrap(); + reservation.flush().unwrap(); + let protobuf = reservation.read(); + PeerEvent::SendGossip { + originator_stream_id: P2pStreamId::new(0, 0, StreamProtocol::GossipSub, false), + topic: GossipTopic::BeaconBlock, + msg_hash: MessageId { id: [byte; 20] }, + recv_ts: Nanos::now(), + protobuf, + block: Some(GossipBlock { slot, block_root: [byte; 32] }), + } } -/// Signals after receiving the response head so events are not published -/// before the subscription is active. -fn events_subscriber(addr: SocketAddr, frame_len: usize) -> (JoinHandle>, Receiver<()>) { - let (subscribed, on_subscribed) = mpsc::channel(); - let client = std::thread::spawn(move || { - let mut stream = TcpStream::connect(addr).unwrap(); - stream.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); - write!(stream, "GET /eth/v1/events?topics=block HTTP/1.1\r\nHost: localhost\r\n\r\n") - .unwrap(); - let mut head = vec![0; SSE_HEAD.len()]; - stream.read_exact(&mut head).unwrap(); - assert_eq!(head, SSE_HEAD, "{}", String::from_utf8_lossy(&head)); - subscribed.send(()).unwrap(); - let mut frame = vec![0; frame_len]; - stream.read_exact(&mut frame).unwrap(); - frame - }); - (client, on_subscribed) +#[derive(Debug)] +struct SseEvent { + name: String, + data: Value, +} + +impl SseEvent { + fn assert_block(&self, name: &str, slot: u64, byte: u8) { + assert_eq!(self.name, name); + assert_eq!(self.data["slot"], slot.to_string()); + assert_eq!(self.data["block"], format!("0x{}", hex::encode([byte; 32]))); + } +} + +struct EventsSubscriber { + client: JoinHandle<()>, + events: Receiver, +} + +impl EventsSubscriber { + fn new(addr: SocketAddr, topics: &str, count: usize, pump: impl FnMut()) -> Self { + let (subscribed, on_subscribed) = mpsc::channel(); + let (send, events) = mpsc::channel(); + let url = format!("http://{addr}/eth/v1/events?topics={topics}"); + let client = std::thread::spawn(move || { + let response = ureq::get(&url).timeout(Duration::from_secs(10)).call().unwrap(); + assert_eq!(response.status(), 200); + assert_eq!(response.content_type(), "text/event-stream"); + subscribed.send(()).unwrap(); + + let mut name = String::new(); + let mut data = String::new(); + let mut received = 0; + for line in BufReader::new(response.into_reader()).lines() { + let line = line.unwrap(); + if line.is_empty() { + if !data.is_empty() { + send.send(SseEvent { + name: if name.is_empty() { "message".to_owned() } else { name.clone() }, + data: serde_json::from_str(&data).expect("event data is JSON"), + }) + .unwrap(); + received += 1; + if received == count { + return; + } + } + name.clear(); + data.clear(); + } else if let Some((field, value)) = line.split_once(':') { + let value = value.strip_prefix(' ').unwrap_or(value); + match field { + "event" => name = value.to_owned(), + "data" => { + data.push_str(value); + data.push('\n'); + } + _ => {} + } + } + } + panic!("event stream closed before {count} events arrived"); + }); + receive_while_pumping(&on_subscribed, pump); + Self { client, events } + } + + fn next(&self, pump: impl FnMut()) -> SseEvent { + receive_while_pumping(&self.events, pump) + } +} + +fn receive_while_pumping(receiver: &Receiver, mut pump: impl FnMut()) -> T { + loop { + match receiver.try_recv() { + Ok(value) => return value, + Err(TryRecvError::Empty) => pump(), + Err(TryRecvError::Disconnected) => { + panic!("subscriber stopped before sending its result") + } + } + } } fn status_event(head_slot: u64, wall_slot: u64, head_optimistic: bool) -> BeaconStateEvent { @@ -726,36 +797,171 @@ fn an_applied_block_on_the_spine_reaches_an_events_subscriber() { ]); let mut adapter = SpineAdapter::connect_tile(&tile, &mut *spine); let mut inj = SpineAdapter::connect_tile(&Injector, &mut *spine); - // Initialize the consumer before publishing: its first consume skips - // events already on the spine. tile.loop_body(&mut adapter); let [Bind::Tcp(addr)] = tile.beacon.local_addrs()[..] else { panic!("expected one tcp bind") }; - let expected = block_frame(10, 0xab); - let (client, on_subscribed) = events_subscriber(addr, expected.len()); - let deadline = Instant::now() + Duration::from_secs(10); - let mut crank = |tile: &mut ApplicationBoundaryTile, msg: &str| { - assert!(Instant::now() < deadline, "timeout: {msg}"); + let mut crank = || { + assert!(Instant::now() < deadline, "timeout: block subscription"); tile.loop_body(&mut adapter); std::thread::sleep(Duration::from_millis(1)); }; - while on_subscribed.try_recv().is_err() { - crank(&mut tile, "stream head reaches the subscriber"); - } + let client = EventsSubscriber::new(addr, "block", 1, &mut crank); inj.produce(block_received(7, 0x07, BlockStage::AlreadyKnown)); inj.produce(block_received(8, 0x08, BlockStage::AwaitParent)); inj.produce(block_received(9, 0x09, BlockStage::AwaitData)); inj.produce(block_received(10, 0xab, BlockStage::Applied)); - while !client.is_finished() { - crank(&mut tile, "block frame reaches the subscriber"); + let event = client.next(&mut crank); + event.assert_block("block", 10, 0xab); + assert_eq!(event.data["execution_optimistic"], true); + client.client.join().unwrap(); +} + +#[test] +fn block_subscriptions_select_imports_and_preserve_repeated_relay_requests() { + let base = TempDir::new().unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(base.path(), None)); + let mut tile = boundary_tile(&Bind::parse("127.0.0.1:0"), no_el(), [ + "cs_two_streams_gossip", + "cs_two_streams_rpc", + "cs_two_streams_resp", + ]); + let mut adapter = SpineAdapter::connect_tile(&tile, &mut *spine); + let mut inj = SpineAdapter::connect_tile(&Injector, &mut *spine); + tile.loop_body(&mut adapter); + + let [Bind::Tcp(addr)] = tile.beacon.local_addrs()[..] else { panic!("expected one tcp bind") }; + let deadline = Instant::now() + Duration::from_secs(10); + let mut crank = || { + assert!(Instant::now() < deadline, "timeout: block subscription routing"); + tile.loop_body(&mut adapter); + std::thread::sleep(Duration::from_millis(1)); + }; + let block = EventsSubscriber::new(addr, "block", 2, &mut crank); + let gossip = EventsSubscriber::new(addr, "block_gossip", 3, &mut crank); + let mixed = EventsSubscriber::new(addr, "block,block_gossip", 5, &mut crank); + + let mut unrelated = block_relay(9, 0xaf); + let PeerEvent::SendGossip { topic, block: metadata, .. } = &mut unrelated else { + unreachable!() + }; + *topic = GossipTopic::BeaconAttestation(0); + *metadata = None; + inj.produce(unrelated); + inj.produce(block_relay(10, 0xac)); + inj.produce(block_relay(10, 0xac)); + inj.produce(block_received(11, 0xab, BlockStage::Applied)); + + // Observe both topics before sending sentinels, keeping leaks inside the events + // read. The two queues promise no ordering relative to each other. + let mut initial = (0..3).map(|_| mixed.next(&mut crank)).collect::>(); + initial.sort_by(|a, b| a.name.cmp(&b.name)); + initial[0].assert_block("block", 11, 0xab); + initial[1].assert_block("block_gossip", 10, 0xac); + initial[2].assert_block("block_gossip", 10, 0xac); + inj.produce(block_relay(12, 0xae)); + inj.produce(block_received(14, 0xb0, BlockStage::Applied)); + + block.next(&mut crank).assert_block("block", 11, 0xab); + block.next(&mut crank).assert_block("block", 14, 0xb0); + gossip.next(&mut crank).assert_block("block_gossip", 10, 0xac); + gossip.next(&mut crank).assert_block("block_gossip", 10, 0xac); + gossip.next(&mut crank).assert_block("block_gossip", 12, 0xae); + let mut trailing = [mixed.next(&mut crank), mixed.next(&mut crank)]; + trailing.sort_by(|a, b| a.name.cmp(&b.name)); + trailing[0].assert_block("block", 14, 0xb0); + trailing[1].assert_block("block_gossip", 12, 0xae); + for subscriber in [block, gossip, mixed] { + subscriber.client.join().unwrap(); } - let got = client.join().unwrap(); - assert!( - got == expected, - "\n got: {:?}\nexpected: {:?}", - String::from_utf8_lossy(&got), - String::from_utf8_lossy(&expected) - ); +} + +#[test] +fn a_late_subscriber_receives_only_relay_requests_published_after_it() { + let base = TempDir::new().unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(base.path(), None)); + let mut tile = boundary_tile(&Bind::parse("127.0.0.1:0"), no_el(), [ + "cs_late_gossip", + "cs_late_rpc", + "cs_late_resp", + ]); + let mut adapter = SpineAdapter::connect_tile(&tile, &mut *spine); + let mut inj = SpineAdapter::connect_tile(&Injector, &mut *spine); + tile.loop_body(&mut adapter); + + let [Bind::Tcp(addr)] = tile.beacon.local_addrs()[..] else { panic!("expected one tcp bind") }; + let deadline = Instant::now() + Duration::from_secs(10); + let mut crank = || { + assert!(Instant::now() < deadline, "timeout: late subscription"); + tile.loop_body(&mut adapter); + std::thread::sleep(Duration::from_millis(1)); + }; + let early = EventsSubscriber::new(addr, "block_gossip", 1, &mut crank); + inj.produce(block_relay(20, 0x11)); + early.next(&mut crank).assert_block("block_gossip", 20, 0x11); + early.client.join().unwrap(); + + let late = EventsSubscriber::new(addr, "block_gossip", 1, &mut crank); + inj.produce(block_relay(21, 0x22)); + late.next(&mut crank).assert_block("block_gossip", 21, 0x22); + late.client.join().unwrap(); +} + +#[test] +fn a_gossip_event_is_served_while_the_engine_pool_is_saturated() { + let base = TempDir::new().unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(base.path(), None)); + let (mut el, endpoint) = FakeEl::tcp(); + let jwt_path = write_jwt(base.path()); + let capacity = 8; + let config = EngineConfig { + execution_endpoint: endpoint, + jwt_secret: jwt_path.to_str().unwrap().to_string(), + max_connections: capacity, + ..EngineConfig::default() + }; + let mut tile = boundary_tile(&Bind::parse("127.0.0.1:0"), config, [ + "cs_gsat_gossip", + "cs_gsat_rpc", + "cs_gsat_resp", + ]); + let mut adapter = SpineAdapter::connect_tile(&tile, &mut *spine); + let mut inj = SpineAdapter::connect_tile(&Injector, &mut *spine); + tile.loop_body(&mut adapter); + for byte in 0..capacity { + inj.produce(fcu_req(byte as u8)); + } + + let deadline = Instant::now() + Duration::from_secs(10); + let mut handled = 0; + let mut pending = 0; + let mut crank = |tile: &mut ApplicationBoundaryTile, el: &mut FakeEl| { + assert!(Instant::now() < deadline, "timeout: saturated engine pool"); + tile.loop_body(&mut adapter); + el.pump(); + for i in handled..el.requests.len() { + let method = el.requests[i].method.as_str(); + if method.starts_with("engine_forkchoiceUpdated") { + pending += 1; + } else { + el.respond(i, if method == "eth_syncing" { "false" } else { "[]" }); + } + } + handled = el.requests.len(); + std::thread::sleep(Duration::from_millis(1)); + pending + }; + while crank(&mut tile, &mut el) < capacity {} + inj.produce(fcu_req(0xff)); + + let [Bind::Tcp(addr)] = tile.beacon.local_addrs()[..] else { panic!("expected one tcp bind") }; + let mut pump = || { + crank(&mut tile, &mut el); + }; + let client = EventsSubscriber::new(addr, "block_gossip", 1, &mut pump); + inj.produce(block_relay(30, 0x33)); + client.next(&mut pump).assert_block("block_gossip", 30, 0x33); + client.client.join().unwrap(); + assert_eq!(pending, capacity, "the additional FCU stays queued while SSE is served"); } diff --git a/crates/beacon_api/src/events.rs b/crates/beacon_api/src/events.rs index 23db79e1..d82a0f8c 100644 --- a/crates/beacon_api/src/events.rs +++ b/crates/beacon_api/src/events.rs @@ -13,6 +13,7 @@ pub(crate) const KEEP_ALIVE: &[u8] = b": keep-alive\n\n"; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum Channel { Block, + BlockGossip, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -68,6 +69,7 @@ fn topics(query: &str) -> Result { fn channel(topic: &str) -> Option { match topic { "block" => Some(Channel::Block), + "block_gossip" => Some(Channel::BlockGossip), _ => None, } } @@ -126,7 +128,19 @@ mod tests { assert_eq!(topics("topics=block"), Ok(block_only())); assert_eq!(topics("topics=block,block"), Ok(block_only())); assert_eq!(topics("topics=block&topics=block"), Ok(block_only())); - assert_eq!(topics("topics=block%2Cblock"), Ok(block_only()), "percent-encoded comma"); + let mut gossip = ChannelSet::default(); + gossip.insert(Channel::BlockGossip); + assert_eq!(topics("topics=block_gossip"), Ok(gossip)); + + let mut both = gossip; + both.insert(Channel::Block); + for query in [ + "topics=block,block_gossip", + "topics=block_gossip&topics=block", + "topics=block%2Cblock_gossip", + ] { + assert_eq!(topics(query), Ok(both), "{query}"); + } } #[test] diff --git a/crates/beacon_api/src/json.rs b/crates/beacon_api/src/json.rs index 3e2a187e..b86107b7 100644 --- a/crates/beacon_api/src/json.rs +++ b/crates/beacon_api/src/json.rs @@ -231,6 +231,15 @@ impl Json<'_> { self.end_object(); } + pub(crate) fn block_gossip_event(&mut self, slot: u64, block_root: &[u8; 32]) { + self.begin_object(); + self.key("slot"); + self.quoted_u64(slot); + self.key("block"); + self.hex(block_root); + self.end_object(); + } + pub(crate) fn finality_checkpoints(&mut self, checkpoints: &FinalityCheckpoints) { self.begin_object(); self.key("previous_justified"); @@ -481,6 +490,14 @@ mod tests { assert!(!json_safe("back\\slash")); } + #[test] + fn block_gossip_event_carries_the_slot_and_the_root() { + let body = write(|json| json.block_gossip_event(10, &[0x9a; 32])); + let parsed: serde_json::Value = serde_json::from_str(&body).expect("valid JSON"); + assert_eq!(parsed["slot"], "10"); + assert_eq!(parsed["block"], format!("0x{}", hex::encode([0x9a; 32]))); + } + #[test] fn block_event_quotes_the_slot_and_hexes_the_root() { let mut out = Vec::new(); diff --git a/crates/beacon_api/src/server.rs b/crates/beacon_api/src/server.rs index 72a870eb..a30bad2c 100644 --- a/crates/beacon_api/src/server.rs +++ b/crates/beacon_api/src/server.rs @@ -397,6 +397,13 @@ impl BeaconApi { self.publish(Channel::Block, "block", &data); } + /// Repeated roots are not deduplicated. + pub fn publish_block_gossip(&mut self, slot: u64, block_root: &[u8; 32]) { + let mut data = Vec::new(); + Json::new(&mut data).block_gossip_event(slot, block_root); + self.publish(Channel::BlockGossip, "block_gossip", &data); + } + fn publish(&mut self, channel: Channel, event: &str, data: &[u8]) { let mut frame = Vec::new(); events::frame(&mut frame, event, data); diff --git a/docs/adr/0004-sync-materialized-api.md b/docs/adr/0004-sync-materialized-api.md index 3ee52e91..d30f621b 100644 --- a/docs/adr/0004-sync-materialized-api.md +++ b/docs/adr/0004-sync-materialized-api.md @@ -78,3 +78,17 @@ are published. The tile calls `publish_block` without accessing connections. `/eth/v1/events` serves `block` and rejects other topics with 400. Further topics and silver-specific SSE routes can use the same subscription mechanism. + +Amended 2026-09-10: `/eth/v1/events` also serves `block_gossip` for block +publication requests following silver's gossip checks. A request precedes +payload notification, state transition, and import; it does not guarantee +delivery to peers. This deliberately narrows the Beacon API's validation +contract: RPC block imports remain silent because they do not request relay. +The topic follows silver's relay policy, keeping its promise tied to gossip +publication without duplicating an observation on the spine. The boundary +selects block metadata on the existing `SendGossip` request, without reading +its payload; producers own topic consistency. The `block` topic still follows +`Applied` import receipts. These queues establish no shared ordering. +Repeated requests for one root are not deduplicated, and late subscribers +receive no replay. Subscribers to both topics can reach the existing send +cap sooner.