diff --git a/crates/application_boundary/src/lib.rs b/crates/application_boundary/src/lib.rs index 67f7bdab..c1901735 100644 --- a/crates/application_boundary/src/lib.rs +++ b/crates/application_boundary/src/lib.rs @@ -7,12 +7,16 @@ use silver_common::{ BeaconStateEvent, BlockStage, Enr, GossipTopic, Identify, Keypair, PeerEvent, SilverSpine, SyncUpdate, TProducer, TRandomAccess, TRead, column_util::{SidecarIdentity, block_root}, - ssz_view::SignedBeaconBlockView, + ssz_view::{SignedBeaconBlockView, StatusView}, }; use silver_config::EngineConfig; use silver_engine_api::EngineApi; use silver_httpcore::{Bind, Readiness, TokenRange}; +use crate::observed_head::{HeadChange, ObservedHead}; + +mod observed_head; + /// A tenant added here takes the next share of a raised `TENANTS`, which keeps /// every share disjoint without a base to compute. const TENANTS: usize = 2; @@ -23,6 +27,7 @@ pub struct ApplicationBoundaryTile { readiness: Readiness, pub beacon: BeaconApi, engine: EngineApi, + head: ObservedHead, spec: SpecConfig, relayed_gossip: TRandomAccess, relayed_rpc: TRandomAccess, @@ -84,11 +89,19 @@ impl ApplicationBoundaryTile { rpc_consumer, resp_producer, ); - Self { readiness, beacon, engine, spec: spec.clone(), relayed_gossip, relayed_rpc } + Self { + readiness, + beacon, + engine, + head: ObservedHead::default(), + spec: spec.clone(), + relayed_gossip, + relayed_rpc, + } } fn consume_spine_events(&mut self, adapter: &mut SpineAdapter) { - let Self { beacon, spec, relayed_gossip, relayed_rpc, .. } = self; + let Self { beacon, head, spec, relayed_gossip, relayed_rpc, .. } = self; // Publish tails even without reads so idle consumers can release cache space. relayed_gossip.free(); relayed_rpc.free(); @@ -97,9 +110,29 @@ impl ApplicationBoundaryTile { // 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, .. } => { + BeaconStateEvent::Status { + ssz, + latest_block_slot, + wall_slot, + head_optimistic, + head_roots, + head_payload, + .. + } => { beacon.node_status_mut().slots = Some(SlotStatus { head_slot: latest_block_slot, wall_slot, head_optimistic }); + if let Some(HeadChange { event, legacy }) = head.observe( + StatusView::head_slot(&ssz), + *StatusView::head_root(&ssz), + head_optimistic, + head_payload, + head_roots, + ) { + if legacy { + beacon.publish_head(&event); + } + beacon.publish_head_v2(&event); + } } BeaconStateEvent::BlockReceived { slot, @@ -130,7 +163,9 @@ impl ApplicationBoundaryTile { }); let status = beacon.node_status_mut(); adapter.consume(|update: SyncUpdate, _| { - status.syncing = !matches!(update, SyncUpdate::Following); + let following = matches!(update, SyncUpdate::Following); + status.syncing = !following; + head.set_following(following); }); status.el = self.engine.sync_status(); diff --git a/crates/application_boundary/src/observed_head.rs b/crates/application_boundary/src/observed_head.rs new file mode 100644 index 00000000..665e1d8b --- /dev/null +++ b/crates/application_boundary/src/observed_head.rs @@ -0,0 +1,286 @@ +use silver_beacon_api::HeadEvent; +use silver_beacon_state_data::{B256, Epoch, SLOTS_PER_EPOCH}; +use silver_common::{HeadRoots, PayloadResolution}; + +/// The last complete observation, including the initial unpublished baseline. +#[derive(Clone, Copy)] +struct Reported { + root: B256, + optimistic: bool, + payload: PayloadResolution, + epoch: Epoch, +} + +/// Every change publishes to `head_v2`; `legacy` also selects `head` when +/// the root or optimism changed. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct HeadChange { + pub(crate) event: HeadEvent, + pub(crate) legacy: bool, +} + +/// Tracks complete head observations in every sync mode, even when no +/// subscribers are connected, and reports changes only while following. +#[derive(Default)] +pub(crate) struct ObservedHead { + reported: Option, + following: bool, +} + +impl ObservedHead { + /// The mode starts as not following, so nothing is reported before + /// Control has concluded. + pub(crate) fn set_following(&mut self, following: bool) { + self.following = following; + } + + /// Incomplete snapshots leave the baseline unchanged. Every complete + /// observation becomes the baseline; only a change observed while + /// following produces an event. + pub(crate) fn observe( + &mut self, + slot: u64, + root: B256, + optimistic: bool, + payload: PayloadResolution, + roots: HeadRoots, + ) -> Option { + if !roots.is_complete() { + return None; + } + let epoch = slot / SLOTS_PER_EPOCH; + let previous = self.reported.replace(Reported { root, optimistic, payload, epoch })?; + if !self.following { + return None; + } + let legacy = previous.root != root || previous.optimistic != optimistic; + if !legacy && previous.payload == payload { + return None; + } + let event = HeadEvent { + slot, + block_root: root, + roots, + payload, + epoch_transition: epoch > previous.epoch, + execution_optimistic: optimistic, + }; + Some(HeadChange { event, legacy }) + } +} + +#[cfg(test)] +mod tests { + use PayloadResolution::{Empty, Full}; + + use super::*; + + const HEAD: B256 = [0x11; 32]; + const OTHER: B256 = [0x22; 32]; + + fn roots(tag: u8) -> HeadRoots { + HeadRoots { + state_root: [tag; 32], + previous_duty_dependent_root: [tag.wrapping_add(1); 32], + current_duty_dependent_root: [tag.wrapping_add(2); 32], + } + } + + fn event( + slot: u64, + root: B256, + tag: u8, + payload: PayloadResolution, + epoch_transition: bool, + optimistic: bool, + ) -> HeadEvent { + HeadEvent { + slot, + block_root: root, + roots: roots(tag), + payload, + epoch_transition, + execution_optimistic: optimistic, + } + } + + fn both_topics(event: HeadEvent) -> Option { + Some(HeadChange { event, legacy: true }) + } + + fn v2_only(event: HeadEvent) -> Option { + Some(HeadChange { event, legacy: false }) + } + + fn following() -> ObservedHead { + let mut head = ObservedHead::default(); + head.set_following(true); + head + } + + fn observed_at( + slot: u64, + root: B256, + optimistic: bool, + payload: PayloadResolution, + ) -> ObservedHead { + let mut head = following(); + assert!( + head.observe(slot, root, optimistic, payload, roots(0x30)).is_none(), + "baseline only" + ); + assert!( + head.observe(slot, root, optimistic, payload, roots(0x30)).is_none(), + "and its repeat" + ); + head + } + + /// An incomplete startup snapshot must not make the first real head appear + /// to advance from epoch zero. + #[test] + fn an_incomplete_status_neither_reports_nor_baselines() { + let mut head = following(); + assert!(head.observe(0, [0u8; 32], true, Empty, HeadRoots::default()).is_none()); + assert!(head.observe(0, [0u8; 32], true, Empty, HeadRoots::default()).is_none(), "repeat"); + + assert!(head.observe(40, HEAD, true, Full, roots(0x30)).is_none()); + assert_eq!( + head.observe(72, OTHER, true, Full, roots(0x40)), + both_topics(event(72, OTHER, 0x40, Full, true, true)), + "epoch 2 against the epoch-1 baseline, not against epoch 0" + ); + } + + /// Every complete observation moves the baseline; only a change observed + /// while following is reported, so a following period starts from the + /// head the node already has. + #[test] + fn changes_are_reported_only_while_following() { + let mut head = ObservedHead::default(); + assert!(head.observe(40, HEAD, true, Full, roots(0x30)).is_none(), "not following yet"); + assert!(head.observe(41, OTHER, true, Full, roots(0x40)).is_none(), "a silent change"); + + head.set_following(true); + assert!(head.observe(41, OTHER, true, Full, roots(0x40)).is_none(), "the baseline repeats"); + assert_eq!( + head.observe(41, OTHER, false, Full, roots(0x40)), + both_topics(event(41, OTHER, 0x40, Full, false, false)) + ); + + head.set_following(false); + assert!(head.observe(64, HEAD, true, Full, roots(0x30)).is_none(), "silent while syncing"); + + head.set_following(true); + assert!(head.observe(64, HEAD, true, Full, roots(0x30)).is_none(), "the baseline moved"); + assert_eq!( + head.observe(65, OTHER, true, Full, roots(0x40)), + both_topics(event(65, OTHER, 0x40, Full, false, true)), + "the epoch transition was observed while syncing, not now" + ); + } + + #[test] + fn a_status_repeating_the_same_head_reports_nothing() { + let mut head = observed_at(40, HEAD, true, Full); + assert!(head.observe(40, HEAD, true, Full, roots(0x50)).is_none(), "other fields changed"); + assert!(head.observe(40, HEAD, true, Full, roots(0x50)).is_none()); + } + + #[test] + fn a_validated_head_reports_once_with_no_epoch_transition() { + let mut head = observed_at(40, HEAD, true, Full); + assert_eq!( + head.observe(40, HEAD, false, Full, roots(0x30)), + both_topics(event(40, HEAD, 0x30, Full, false, false)) + ); + assert!(head.observe(40, HEAD, false, Full, roots(0x30)).is_none(), "repeat"); + } + + #[test] + fn a_payload_resolution_change_alone_reaches_only_head_v2() { + let mut head = observed_at(40, HEAD, true, Empty); + assert_eq!( + head.observe(40, HEAD, true, Full, roots(0x30)), + v2_only(event(40, HEAD, 0x30, Full, false, true)) + ); + assert!(head.observe(40, HEAD, true, Full, roots(0x30)).is_none(), "repeat"); + + assert_eq!( + head.observe(40, HEAD, true, Empty, roots(0x30)), + v2_only(event(40, HEAD, 0x30, Empty, false, true)), + "the reverse transition is published too" + ); + assert!(head.observe(40, HEAD, true, Empty, roots(0x30)).is_none(), "repeat"); + } + + #[test] + fn a_change_in_every_dimension_is_one_event_for_both_topics() { + let mut head = observed_at(40, HEAD, true, Empty); + assert_eq!( + head.observe(41, OTHER, false, Full, roots(0x40)), + both_topics(event(41, OTHER, 0x40, Full, false, false)) + ); + assert!(head.observe(41, OTHER, false, Full, roots(0x40)).is_none(), "repeat"); + } + + #[test] + fn a_head_change_inside_one_epoch_reports_no_transition() { + let mut head = observed_at(40, HEAD, true, Full); + assert_eq!( + head.observe(41, OTHER, true, Full, roots(0x40)), + both_topics(event(41, OTHER, 0x40, Full, false, true)) + ); + assert!(head.observe(41, OTHER, true, Full, roots(0x40)).is_none(), "repeat"); + } + + #[test] + fn a_head_change_into_a_later_epoch_reports_the_transition() { + let mut head = observed_at(40, HEAD, true, Full); + assert_eq!( + head.observe(64, OTHER, true, Full, roots(0x40)), + both_topics(event(64, OTHER, 0x40, Full, true, true)) + ); + assert!(head.observe(64, OTHER, true, Full, roots(0x40)).is_none(), "repeat"); + } + + #[test] + fn a_head_change_into_an_earlier_epoch_reports_no_transition() { + let mut head = observed_at(64, HEAD, true, Full); + assert_eq!( + head.observe(40, OTHER, true, Full, roots(0x40)), + both_topics(event(40, OTHER, 0x40, Full, false, true)) + ); + assert!(head.observe(40, OTHER, true, Full, roots(0x40)).is_none(), "repeat"); + } + + /// The transition flag belongs to the observation that crossed the epoch, + /// not to later updates of the same head. + #[test] + fn updates_right_after_an_epoch_change_report_no_transition() { + let mut head = observed_at(40, HEAD, true, Empty); + assert_eq!( + head.observe(64, OTHER, true, Empty, roots(0x40)), + both_topics(event(64, OTHER, 0x40, Empty, true, true)) + ); + assert_eq!( + head.observe(64, OTHER, true, Full, roots(0x40)), + v2_only(event(64, OTHER, 0x40, Full, false, true)) + ); + assert_eq!( + head.observe(64, OTHER, false, Full, roots(0x40)), + both_topics(event(64, OTHER, 0x40, Full, false, false)) + ); + } + + #[test] + fn an_incomplete_snapshot_between_complete_ones_keeps_the_earlier_baseline() { + let mut head = observed_at(40, HEAD, true, Empty); + assert!(head.observe(41, OTHER, false, Full, HeadRoots::default()).is_none()); + assert_eq!( + head.observe(40, HEAD, true, Full, roots(0x30)), + v2_only(event(40, HEAD, 0x30, Full, false, true)), + "only the payload differs from the last complete observation" + ); + } +} diff --git a/crates/application_boundary/tests/tile.rs b/crates/application_boundary/tests/tile.rs index ec4d0a14..e4100345 100644 --- a/crates/application_boundary/tests/tile.rs +++ b/crates/application_boundary/tests/tile.rs @@ -1,4 +1,5 @@ use std::{ + collections::HashMap, io::{BufRead, BufReader, Read, Write}, net::{SocketAddr, TcpStream}, os::unix::net::UnixStream, @@ -11,12 +12,12 @@ use flux::{spine::SpineAdapter, tile::Tile, timing::Nanos}; use serde_json::{Value, json}; use silver_application_boundary::ApplicationBoundaryTile; use silver_beacon_api::SlotStatus; -use silver_beacon_state_data::{BeaconStateOwner, SpecConfig}; +use silver_beacon_state_data::{BeaconStateOwner, SLOTS_PER_EPOCH, SpecConfig}; use silver_common::{ BeaconStateEvent, BlockSource, BlockStage, ELSyncStatus, EngineFcuReq, EngineReq, EngineResp, - Enr, GossipTopic, Identify, Keypair, MessageId, P2pStreamId, PayloadValidationStatus, - PeerEvent, SilverSpine, StreamProtocol, SyncUpdate, TCache, TCacheProducer, TCacheRead, - TProducer, + Enr, GossipTopic, HeadRoots, Identify, Keypair, MessageId, P2pStreamId, PayloadResolution, + PayloadValidationStatus, PeerEvent, SilverSpine, StreamProtocol, SyncUpdate, TCache, + TCacheProducer, TCacheRead, TProducer, column_util::{block_root_from_sidecar, block_root_fulu}, ssz_view::{ BEACON_BLOCK_BODY_FIXED, DATA_COLUMN_SIDECAR_GLOAS_MIN, DATA_COLUMN_SIDECAR_MIN, @@ -38,13 +39,22 @@ fn boundary_tile( engine_config: EngineConfig, tcache_names: [&'static str; 3], ) -> ApplicationBoundaryTile { - boundary_tile_with_objects(bind, engine_config, tcache_names).0 + boundary_tile_with_spec(bind, engine_config, tcache_names, &SpecConfig::mainnet()).0 } fn boundary_tile_with_objects( bind: &Bind, engine_config: EngineConfig, tcache_names: [&'static str; 3], +) -> (ApplicationBoundaryTile, TProducer, TProducer) { + boundary_tile_with_spec(bind, engine_config, tcache_names, &SpecConfig::mainnet()) +} + +fn boundary_tile_with_spec( + bind: &Bind, + engine_config: EngineConfig, + tcache_names: [&'static str; 3], + spec: &SpecConfig, ) -> (ApplicationBoundaryTile, TProducer, TProducer) { let keypair = Keypair::from_secret(&[1u8; 32]).unwrap(); let local_enr = Enr::empty(keypair.secret_key()).unwrap(); @@ -58,7 +68,7 @@ fn boundary_tile_with_objects( &keypair, local_enr, &Identify::default(), - &SpecConfig::mainnet(), + spec, BeaconStateOwner::empty_test(0).reader(), engine_config, gossip_p.cache_ref().random_access("t", true).unwrap(), @@ -385,9 +395,98 @@ fn status_event(head_slot: u64, wall_slot: u64, head_optimistic: bool) -> Beacon latest_block_slot: head_slot, wall_slot, enr_fork_id: [0u8; 16], + head_roots: HeadRoots::default(), + head_payload: PayloadResolution::Full, + } +} + +fn head_roots() -> HeadRoots { + HeadRoots { + state_root: [0x60; 32], + previous_duty_dependent_root: [0x5e; 32], + current_duty_dependent_root: [0x91; 32], + } +} + +fn head_status( + slot: u64, + block_root: u8, + head_optimistic: bool, + head_payload: PayloadResolution, +) -> BeaconStateEvent { + let mut ssz = [0u8; STATUS_V2_SIZE]; + ssz[44..76].copy_from_slice(&[block_root; 32]); + ssz[76..84].copy_from_slice(&slot.to_le_bytes()); + BeaconStateEvent::Status { + ssz, + head_optimistic, + latest_block_slot: slot, + wall_slot: slot, + enr_fork_id: [0u8; 16], + head_roots: head_roots(), + head_payload, } } +fn head_events_subscriber( + addr: SocketAddr, + topic: &str, + sentinel_slot: u64, +) -> (JoinHandle>, Receiver<()>) { + let (subscribed, on_subscribed) = mpsc::channel(); + let topic = topic.to_string(); + 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={topic} HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line.split_whitespace().nth(1), Some("200")); + let mut headers = HashMap::new(); + loop { + line.clear(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + let (name, value) = line.trim_end().split_once(':').unwrap(); + headers.insert(name.to_ascii_lowercase(), value.trim().to_string()); + } + assert_eq!(headers["content-type"], "text/event-stream"); + assert_eq!(headers["transfer-encoding"], "chunked"); + subscribed.send(()).unwrap(); + + let mut events = Vec::new(); + loop { + line.clear(); + assert!(reader.read_line(&mut line).unwrap() > 0); + let size = usize::from_str_radix(line.trim().split(';').next().unwrap(), 16).unwrap(); + assert!(size > 0, "subscription ended before the sentinel"); + let mut chunk = vec![0; size]; + reader.read_exact(&mut chunk).unwrap(); + let mut end = [0; 2]; + reader.read_exact(&mut end).unwrap(); + assert_eq!(end, *b"\r\n"); + let frame = std::str::from_utf8(&chunk).unwrap().strip_suffix("\n\n").unwrap(); + if frame.starts_with(':') { + continue; + } + let (name, data) = + frame.strip_prefix("event: ").unwrap().split_once("\ndata: ").unwrap(); + assert_eq!(name, topic); + let body: Value = serde_json::from_str(data).unwrap(); + let data = if topic == "head_v2" { &body["data"] } else { &body }; + if data["slot"] == sentinel_slot.to_string() { + return events; + } + events.push(body); + } + }); + (client, on_subscribed) +} + #[test] fn serves_identity_over_tcp() { let base = ShmemDir::new().unwrap(); @@ -1177,3 +1276,202 @@ fn gossip_events_are_served_while_the_engine_pool_is_saturated() { client.client.join().unwrap(); assert_eq!(pending, capacity, "the additional FCU stays queued while SSE is served"); } + +#[test] +fn head_subscribers_receive_changes_for_their_topics() { + let base = ShmemDir::new().unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(base.path(), None)); + let mut spec = SpecConfig::mainnet(); + spec.gloas_fork_epoch = spec.fulu_fork_epoch + 2; + let (mut tile, _gossip, _rpc) = boundary_tile_with_spec( + &Bind::parse("127.0.0.1:0"), + no_el(), + ["cs_head_v2_gossip", "cs_head_v2_rpc", "cs_head_v2_resp"], + &spec, + ); + 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 gloas = spec.gloas_fork_epoch * SLOTS_PER_EPOCH; + let slot = gloas + 8; + let sentinel_slot = slot + 1; + let (legacy, legacy_subscribed) = head_events_subscriber(addr, "head", sentinel_slot); + let (v2, v2_subscribed) = head_events_subscriber(addr, "head_v2", sentinel_slot); + + let deadline = Instant::now() + Duration::from_secs(10); + let mut crank = |tile: &mut ApplicationBoundaryTile, msg: &str| { + assert!(Instant::now() < deadline, "timeout: {msg}"); + tile.loop_body(&mut adapter); + std::thread::sleep(Duration::from_millis(1)); + }; + for subscribed in [legacy_subscribed, v2_subscribed] { + while subscribed.try_recv().is_err() { + crank(&mut tile, "both stream heads reach their subscribers"); + } + } + inj.produce(SyncUpdate::Following); + crank(&mut tile, "the node is following"); + + for status in [ + head_status(gloas + 1, 0x0a, true, PayloadResolution::Full), + head_status(gloas + 1, 0x0a, true, PayloadResolution::Full), + head_status(slot, 0xab, true, PayloadResolution::Empty), + head_status(slot, 0xab, true, PayloadResolution::Empty), + head_status(slot, 0xab, true, PayloadResolution::Full), + head_status(slot, 0xab, true, PayloadResolution::Full), + head_status(slot, 0xab, false, PayloadResolution::Full), + head_status(slot, 0xab, false, PayloadResolution::Full), + ] { + inj.produce(status); + } + crank(&mut tile, "head observations update node status"); + assert_eq!( + tile.beacon.node_status_mut().slots, + Some(SlotStatus { head_slot: slot, wall_slot: slot, head_optimistic: false }) + ); + + // A later head delimits all preceding frames, including unwanted repeats. + inj.produce(head_status(sentinel_slot, 0xcd, false, PayloadResolution::Full)); + while !legacy.is_finished() || !v2.is_finished() { + crank(&mut tile, "every frame reaches its subscriber"); + } + let legacy = legacy.join().unwrap(); + let v2 = v2.join().unwrap(); + assert_eq!(legacy.len(), 2); + assert_eq!(v2.len(), 3); + let roots = head_roots(); + for (events, is_v2) in [(&legacy, false), (&v2, true)] { + for (index, body) in events.iter().enumerate() { + let data = if is_v2 { + assert_eq!(body["version"], "gloas"); + assert_eq!( + body["data"]["payload_status"], + if index == 0 { "empty" } else { "full" } + ); + &body["data"] + } else { + body + }; + assert_eq!(data["slot"], slot.to_string()); + assert_eq!(data["block"], format!("0x{}", hex::encode([0xab; 32]))); + assert_eq!(data["state"], format!("0x{}", hex::encode(roots.state_root))); + assert_eq!(data["epoch_transition"], false); + assert_eq!(data["execution_optimistic"], index + 1 < events.len()); + let (previous, current) = if is_v2 { + ("current_epoch_dependent_root", "next_epoch_dependent_root") + } else { + ("previous_duty_dependent_root", "current_duty_dependent_root") + }; + assert_eq!( + data[previous], + format!("0x{}", hex::encode(roots.previous_duty_dependent_root)) + ); + assert_eq!( + data[current], + format!("0x{}", hex::encode(roots.current_duty_dependent_root)) + ); + } + } +} + +/// Head events describe changes observed while following. Observations in +/// any other mode move the baseline and node status silently. +#[test] +fn head_events_describe_changes_observed_while_following() { + let base = ShmemDir::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_follow_gossip", + "cs_follow_rpc", + "cs_follow_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 sentinel_slot = 40; + let (client, on_subscribed) = head_events_subscriber(addr, "head", sentinel_slot); + + let deadline = Instant::now() + Duration::from_secs(10); + let mut crank = |tile: &mut ApplicationBoundaryTile, msg: &str| { + assert!(Instant::now() < deadline, "timeout: {msg}"); + 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"); + } + + // Restoration and catch-up: Control has not concluded, so heads move silently. + inj.produce(head_status(33, 0xaa, true, PayloadResolution::Full)); + inj.produce(head_status(34, 0xab, true, PayloadResolution::Full)); + crank(&mut tile, "observations outside following update node status"); + assert_eq!( + tile.beacon.node_status_mut().slots, + Some(SlotStatus { head_slot: 34, wall_slot: 34, head_optimistic: true }) + ); + + // Following: the latest observation is already the baseline, so the next + // change is reported at once. + inj.produce(SyncUpdate::Following); + crank(&mut tile, "the mode change is consumed"); + inj.produce(head_status(35, 0xac, true, PayloadResolution::Full)); + inj.produce(head_status(35, 0xac, false, PayloadResolution::Full)); + crank(&mut tile, "changes while following are reported"); + + // Falling behind silences the stream while the head keeps moving. + inj.produce(SyncUpdate::SyncingHead { head_root: [0xff; 32], head_slot: 100 }); + crank(&mut tile, "the mode change is consumed"); + inj.produce(head_status(36, 0xad, true, PayloadResolution::Full)); + crank(&mut tile, "changes while syncing are silent"); + + // Following again: a repeat of the head reached while syncing is no change. + inj.produce(SyncUpdate::Following); + crank(&mut tile, "the mode change is consumed"); + inj.produce(head_status(36, 0xad, true, PayloadResolution::Full)); + inj.produce(head_status(37, 0xae, true, PayloadResolution::Full)); + inj.produce(head_status(sentinel_slot, 0xcd, true, PayloadResolution::Full)); + while !client.is_finished() { + crank(&mut tile, "every frame reaches the subscriber"); + } + + let events = client.join().unwrap(); + assert_eq!(events.len(), 3); + for (event, (slot, optimistic)) in events.iter().zip([(35, true), (35, false), (37, true)]) { + assert_eq!(event["slot"], slot.to_string()); + assert_eq!(event["execution_optimistic"], optimistic); + } +} + +/// The initial head observation emits no event but still updates node status. +#[test] +fn node_status_optimism_follows_a_status_that_publishes_no_head_event() { + let base = ShmemDir::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_optimism_gossip", + "cs_optimism_rpc", + "cs_optimism_resp", + ]); + let mut adapter = SpineAdapter::connect_tile(&tile, &mut *spine); + let mut inj = SpineAdapter::connect_tile(&Injector, &mut *spine); + tile.loop_body(&mut adapter); + + inj.produce(head_status(32, 0x0a, true, PayloadResolution::Full)); + tile.loop_body(&mut adapter); + assert_eq!( + tile.beacon.node_status_mut().slots, + Some(SlotStatus { head_slot: 32, wall_slot: 32, head_optimistic: true }) + ); + + inj.produce(head_status(32, 0x0a, false, PayloadResolution::Full)); + tile.loop_body(&mut adapter); + assert_eq!( + tile.beacon.node_status_mut().slots, + Some(SlotStatus { head_slot: 32, wall_slot: 32, head_optimistic: false }), + "the verdict reaches node status whatever the head filter decides" + ); +} diff --git a/crates/beacon_api/src/events.rs b/crates/beacon_api/src/events.rs index b2edb86a..c69d96e4 100644 --- a/crates/beacon_api/src/events.rs +++ b/crates/beacon_api/src/events.rs @@ -1,5 +1,7 @@ use std::io::Write; +use silver_beacon_state_data::B256; +use silver_common::{HeadRoots, PayloadResolution}; use silver_httpcore::Query; use crate::{response::Response, router::Request, routes::ApiCtx}; @@ -13,10 +15,24 @@ pub(crate) const KEEP_ALIVE: &[u8] = b": keep-alive\n\n"; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum Channel { Block, + Head, + HeadV2, BlockGossip, DataColumnSidecar, } +/// `epoch_transition` compares this head with the publisher's previous +/// complete observation. Only `head_v2` renders `payload`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct HeadEvent { + pub slot: u64, + pub block_root: B256, + pub roots: HeadRoots, + pub payload: PayloadResolution, + pub epoch_transition: bool, + pub execution_optimistic: bool, +} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct ChannelSet(u32); @@ -70,6 +86,8 @@ fn topics(query: &str) -> Result { fn channel(topic: &str) -> Option { match topic { "block" => Some(Channel::Block), + "head" => Some(Channel::Head), + "head_v2" => Some(Channel::HeadV2), "block_gossip" => Some(Channel::BlockGossip), "data_column_sidecar" => Some(Channel::DataColumnSidecar), _ => None, @@ -93,12 +111,18 @@ mod tests { routes::{ROUTES, preboot_ctx}, }; - fn block_only() -> ChannelSet { + fn set(of: &[Channel]) -> ChannelSet { let mut channels = ChannelSet::default(); - channels.insert(Channel::Block); + for &channel in of { + channels.insert(channel); + } channels } + fn block_only() -> ChannelSet { + set(&[Channel::Block]) + } + fn dispatch(query: &str) -> (Served, Vec) { let req = ParsedRequest { method: "GET", @@ -130,6 +154,7 @@ 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)); @@ -148,14 +173,46 @@ mod tests { ] { assert_eq!(topics(query), Ok(all), "{query}"); } + + let every_topic = set(&[ + Channel::Block, + Channel::Head, + Channel::HeadV2, + Channel::BlockGossip, + Channel::DataColumnSidecar, + ]); + for query in [ + "topics=block,head,head_v2,block_gossip,data_column_sidecar", + "topics=data_column_sidecar&topics=block_gossip&topics=head_v2&topics=head&topics=block", + ] { + assert_eq!(topics(query), Ok(every_topic), "{query}"); + } + } + + #[test] + fn head_is_served_alone_and_alongside_block() { + assert_eq!(topics("topics=head"), Ok(set(&[Channel::Head]))); + assert_eq!(topics("topics=block,head"), Ok(set(&[Channel::Block, Channel::Head]))); + assert_eq!(topics("topics=head,block"), Ok(set(&[Channel::Block, Channel::Head]))); + assert_eq!(topics("topics=head&topics=block"), Ok(set(&[Channel::Block, Channel::Head]))); + } + + #[test] + fn head_v2_is_served_alone_and_alongside_the_other_topics() { + let all = set(&[Channel::Block, Channel::Head, Channel::HeadV2]); + assert_eq!(topics("topics=head_v2"), Ok(set(&[Channel::HeadV2]))); + assert_eq!(topics("topics=head_v2,head_v2"), Ok(set(&[Channel::HeadV2]))); + assert_eq!(topics("topics=head,head_v2"), Ok(set(&[Channel::Head, Channel::HeadV2]))); + assert_eq!(topics("topics=block,head,head_v2"), Ok(all)); + assert_eq!(topics("topics=head_v2&topics=block&topics=head"), Ok(all)); } #[test] fn a_topic_silver_does_not_serve_refuses_the_whole_subscription_by_name() { let unknown = |topic: &str| Err(Refused::Unknown(topic.to_string())); - assert_eq!(topics("topics=head"), unknown("head")); - assert_eq!(topics("topics=block,head"), unknown("head")); - assert_eq!(topics("topics=block&topics=chain_reorg"), unknown("chain_reorg")); + assert_eq!(topics("topics=finalized_checkpoint"), unknown("finalized_checkpoint")); + assert_eq!(topics("topics=block,finalized_checkpoint"), unknown("finalized_checkpoint")); + assert_eq!(topics("topics=head_v2&topics=chain_reorg"), unknown("chain_reorg")); } #[test] @@ -194,9 +251,9 @@ mod tests { #[test] fn an_unserved_topic_is_a_400_naming_it_on_an_ordinary_connection() { - let (served, out) = dispatch("topics=block,head"); + let (served, out) = dispatch("topics=block,chain_reorg"); assert_eq!(served, Served::Response); - assert_eq!(out, bad_request(r#"unknown topic \"head\""#)); + assert_eq!(out, bad_request(r#"unknown topic \"chain_reorg\""#)); } #[test] diff --git a/crates/beacon_api/src/json.rs b/crates/beacon_api/src/json.rs index a79c3688..e2b43ffd 100644 --- a/crates/beacon_api/src/json.rs +++ b/crates/beacon_api/src/json.rs @@ -5,6 +5,8 @@ use silver_beacon_state_data::{B256, Checkpoint, Fork, Version}; +use crate::events::HeadEvent; + const HEX_LOWER: &[u8; 16] = b"0123456789abcdef"; /// Appends JSON to a buffer the caller owns — fresh or reused is the caller's @@ -231,6 +233,51 @@ impl Json<'_> { self.end_object(); } + pub(crate) fn head_event(&mut self, head: &HeadEvent) { + self.begin_object(); + self.key("slot"); + self.quoted_u64(head.slot); + self.key("block"); + self.hex(&head.block_root); + self.key("state"); + self.hex(&head.roots.state_root); + self.key("epoch_transition"); + self.bool(head.epoch_transition); + self.key("previous_duty_dependent_root"); + self.hex(&head.roots.previous_duty_dependent_root); + self.key("current_duty_dependent_root"); + self.hex(&head.roots.current_duty_dependent_root); + self.key("execution_optimistic"); + self.bool(head.execution_optimistic); + self.end_object(); + } + + pub(crate) fn head_v2_event(&mut self, head: &HeadEvent, fork_name: &str) { + self.begin_object(); + self.key("version"); + self.string(fork_name); + self.key("data"); + self.begin_object(); + self.key("slot"); + self.quoted_u64(head.slot); + self.key("block"); + self.hex(&head.block_root); + self.key("state"); + self.hex(&head.roots.state_root); + self.key("payload_status"); + self.string(head.payload.name()); + self.key("epoch_transition"); + self.bool(head.epoch_transition); + self.key("current_epoch_dependent_root"); + self.hex(&head.roots.previous_duty_dependent_root); + self.key("next_epoch_dependent_root"); + self.hex(&head.roots.current_duty_dependent_root); + self.key("execution_optimistic"); + self.bool(head.execution_optimistic); + self.end_object(); + self.end_object(); + } + pub(crate) fn block_gossip_event(&mut self, slot: u64, block_root: &[u8; 32]) { self.begin_object(); self.key("slot"); @@ -277,7 +324,9 @@ pub(crate) fn json_safe(text: &str) -> bool { #[cfg(test)] mod tests { + use serde_json::Value; use silver_beacon_state_data::FAR_FUTURE_EPOCH; + use silver_common::{HeadRoots, PayloadResolution}; use super::*; @@ -506,6 +555,73 @@ mod tests { assert!(!json_safe("back\\slash")); } + #[test] + fn head_event_encodes_the_required_fields() { + let head = HeadEvent { + slot: 10, + block_root: [0x9a; 32], + roots: HeadRoots { + state_root: [0x60; 32], + previous_duty_dependent_root: [0x5e; 32], + current_duty_dependent_root: [0x91; 32], + }, + payload: PayloadResolution::Full, + epoch_transition: true, + execution_optimistic: false, + }; + let mut out = Vec::new(); + Json::new(&mut out).head_event(&head); + let data: Value = serde_json::from_slice(&out).unwrap(); + assert_eq!(data["slot"], head.slot.to_string()); + assert_eq!(data["block"], format!("0x{}", hex::encode(head.block_root))); + assert_eq!(data["state"], format!("0x{}", hex::encode(head.roots.state_root))); + assert_eq!(data["epoch_transition"], true); + assert_eq!(data["execution_optimistic"], false); + assert_eq!( + data["previous_duty_dependent_root"], + format!("0x{}", hex::encode(head.roots.previous_duty_dependent_root)) + ); + assert_eq!( + data["current_duty_dependent_root"], + format!("0x{}", hex::encode(head.roots.current_duty_dependent_root)) + ); + } + + #[test] + fn head_v2_event_wraps_the_versioned_data_and_maps_the_dependent_roots() { + let head = HeadEvent { + slot: 10, + block_root: [0x9a; 32], + roots: HeadRoots { + state_root: [0x60; 32], + previous_duty_dependent_root: [0x5e; 32], + current_duty_dependent_root: [0x91; 32], + }, + payload: PayloadResolution::Empty, + epoch_transition: false, + execution_optimistic: true, + }; + let mut out = Vec::new(); + Json::new(&mut out).head_v2_event(&head, "gloas"); + let body: Value = serde_json::from_slice(&out).unwrap(); + assert_eq!(body["version"], "gloas"); + let data = &body["data"]; + assert_eq!(data["slot"], head.slot.to_string()); + assert_eq!(data["block"], format!("0x{}", hex::encode(head.block_root))); + assert_eq!(data["state"], format!("0x{}", hex::encode(head.roots.state_root))); + assert_eq!(data["payload_status"], "empty"); + assert_eq!(data["epoch_transition"], false); + assert_eq!(data["execution_optimistic"], true); + assert_eq!( + data["current_epoch_dependent_root"], + format!("0x{}", hex::encode(head.roots.previous_duty_dependent_root)) + ); + assert_eq!( + data["next_epoch_dependent_root"], + format!("0x{}", hex::encode(head.roots.current_duty_dependent_root)) + ); + } + #[test] fn block_gossip_event_carries_the_slot_and_the_root() { let body = write(|json| json.block_gossip_event(10, &[0x9a; 32])); diff --git a/crates/beacon_api/src/lib.rs b/crates/beacon_api/src/lib.rs index e7475ad0..cc4c80b9 100644 --- a/crates/beacon_api/src/lib.rs +++ b/crates/beacon_api/src/lib.rs @@ -11,5 +11,6 @@ mod routes; mod server; mod statics; +pub use events::HeadEvent; pub use node_status::{NodeStatus, SlotStatus}; pub use server::BeaconApi; diff --git a/crates/beacon_api/src/routes.rs b/crates/beacon_api/src/routes.rs index 8222fd54..2bfaffa2 100644 --- a/crates/beacon_api/src/routes.rs +++ b/crates/beacon_api/src/routes.rs @@ -70,6 +70,7 @@ pub(crate) const ROUTES: &[(Method, &str, Handler)] = &[ pub(crate) struct ApiCtx { pub(crate) statics: StaticBodies, + pub(crate) spec: SpecConfig, pub(crate) state: BeaconStateReader, pub(crate) node_status: NodeStatus, } @@ -84,6 +85,7 @@ impl ApiCtx { ) -> Self { Self { statics: StaticBodies::new(keypair, local_enr, identify, spec), + spec: spec.clone(), state, node_status: NodeStatus::default(), } diff --git a/crates/beacon_api/src/server.rs b/crates/beacon_api/src/server.rs index 849d9df3..14b1a407 100644 --- a/crates/beacon_api/src/server.rs +++ b/crates/beacon_api/src/server.rs @@ -15,7 +15,7 @@ use silver_httpcore::{ use crate::{ NodeStatus, - events::{self, Channel, ChannelSet}, + events::{self, Channel, ChannelSet, HeadEvent}, json::Json, router::{Router, Served}, routes::{ApiCtx, ROUTES}, @@ -397,6 +397,19 @@ impl BeaconApi { self.publish(Channel::Block, "block", &data); } + /// Head-change detection belongs to the caller. + pub fn publish_head(&mut self, head: &HeadEvent) { + let mut data = Vec::new(); + Json::new(&mut data).head_event(head); + self.publish(Channel::Head, "head", &data); + } + + pub fn publish_head_v2(&mut self, head: &HeadEvent) { + let mut data = Vec::new(); + Json::new(&mut data).head_v2_event(head, self.ctx.spec.fork_at_slot(head.slot).name()); + self.publish(Channel::HeadV2, "head_v2", &data); + } + /// Repeated roots are not deduplicated. pub fn publish_block_gossip(&mut self, slot: u64, block_root: &[u8; 32]) { let mut data = Vec::new(); @@ -604,6 +617,7 @@ fn interrupted(err: &io::Error) -> bool { #[cfg(test)] mod tests { use std::{ + io::{BufRead, BufReader}, net::{SocketAddr, TcpStream}, os::unix::net::UnixStream, path::Path, @@ -611,7 +625,9 @@ mod tests { time::Instant, }; - use silver_beacon_state_data::BeaconStateOwner; + use serde_json::Value; + use silver_beacon_state_data::{BeaconStateOwner, SLOTS_PER_EPOCH}; + use silver_common::{HeadRoots, PayloadResolution}; use silver_httpcore::Readiness; use super::*; @@ -1412,6 +1428,148 @@ mod tests { assert_eq!(subscribers(&server), 1, "delivery keeps the subscription"); } + fn head_event(slot: u64, block_root: &[u8; 32], execution_optimistic: bool) -> HeadEvent { + HeadEvent { + slot, + block_root: *block_root, + roots: HeadRoots { + state_root: [0x60; 32], + previous_duty_dependent_root: [0x5e; 32], + current_duty_dependent_root: [0x91; 32], + }, + payload: PayloadResolution::Full, + epoch_transition: false, + execution_optimistic, + } + } + + struct SseEvent { + topic: String, + data: Value, + } + + fn read_events_until_marker(stream: TcpStream) -> JoinHandle> { + std::thread::spawn(move || { + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line.split_whitespace().nth(1), Some("200")); + let mut headers = HashMap::new(); + loop { + line.clear(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + let (name, value) = line.trim_end().split_once(':').unwrap(); + headers.insert(name.to_ascii_lowercase(), value.trim().to_string()); + } + assert_eq!(headers["content-type"], "text/event-stream"); + assert_eq!(headers["transfer-encoding"], "chunked"); + + let mut events = Vec::new(); + loop { + line.clear(); + assert!(reader.read_line(&mut line).unwrap() > 0); + let size = + usize::from_str_radix(line.trim().split(';').next().unwrap(), 16).unwrap(); + assert!(size > 0, "subscription ended before the marker"); + let mut chunk = vec![0; size]; + reader.read_exact(&mut chunk).unwrap(); + let mut end = [0; 2]; + reader.read_exact(&mut end).unwrap(); + assert_eq!(end, *b"\r\n"); + let frame = std::str::from_utf8(&chunk).unwrap().strip_suffix("\n\n").unwrap(); + if frame.starts_with(':') { + continue; + } + let (topic, data) = + frame.strip_prefix("event: ").unwrap().split_once("\ndata: ").unwrap(); + let data = serde_json::from_str(data).unwrap(); + if topic == "test_end" { + return events; + } + events.push(SseEvent { topic: topic.to_string(), data }); + } + }) + } + + // Queued after every publication, so readers can detect leaked or repeated + // frames without relying on a quiet socket or cross-topic ordering. + fn finish_events(server: &mut Server) { + server.api.fan_out(|_| true, b"event: test_end\ndata: {}\n\n", Instant::now()); + } + + #[test] + fn each_subscriber_receives_only_the_channels_it_asked_for() { + let mut server = server_with(64, LONG_TIMEOUT); + let subscriptions = + ["block", "head", "block,head", "head_v2", "head,head_v2", "block,head,head_v2"]; + let readers = subscriptions.map(|topics| { + let mut stream = connect(tcp_addr(&server)); + subscribe(&mut stream, topics); + read_events_until_marker(stream) + }); + pump_until(&mut server, "all subscribed", |server| { + subscribers(server) == subscriptions.len() + }); + + let slot = SpecConfig::mainnet().fulu_fork_epoch * SLOTS_PER_EPOCH; + let head = head_event(slot, &[0xab; 32], true); + server.api.publish_block(slot, &head.block_root); + server.api.publish_head(&head); + server.api.publish_head_v2(&head); + finish_events(&mut server); + pump_until(&mut server, "every subscriber served", |_| { + readers.iter().all(JoinHandle::is_finished) + }); + + for (topics, reader) in subscriptions.into_iter().zip(readers) { + let events = reader.join().unwrap(); + assert_eq!(events.len(), topics.split(',').count(), "{topics}"); + for topic in topics.split(',') { + let matching: Vec<_> = events.iter().filter(|event| event.topic == topic).collect(); + assert_eq!(matching.len(), 1, "{topics}: {topic}"); + let body = &matching[0].data; + let data = if topic == "head_v2" { + assert_eq!(body["version"], "fulu"); + assert_eq!(body["data"]["payload_status"], "full"); + &body["data"] + } else { + body + }; + assert_eq!(data["slot"], slot.to_string()); + assert_eq!(data["block"], format!("0x{}", hex::encode(head.block_root))); + assert_eq!(data["execution_optimistic"], true); + } + } + } + + #[test] + fn head_v2_names_the_fork_at_the_head_slot() { + let mut server = server_with(64, LONG_TIMEOUT); + let mut client = connect(tcp_addr(&server)); + subscribe(&mut client, "head_v2"); + let reader = read_events_until_marker(client); + pump_until(&mut server, "subscribed", |server| subscribers(server) == 1); + + let fulu = SpecConfig::mainnet().fulu_fork_epoch * SLOTS_PER_EPOCH; + let expected = + [(fulu - 1, "electra"), (fulu, "fulu"), (fulu + 1, "fulu"), (fulu - 1, "electra")]; + for (slot, _) in expected { + server.api.publish_head_v2(&head_event(slot, &[0xab; 32], false)); + } + finish_events(&mut server); + let events = serve(&mut server, reader, "versioned frames and marker"); + assert_eq!(events.len(), expected.len()); + for (event, (slot, version)) in events.iter().zip(expected) { + assert_eq!(event.topic, "head_v2"); + assert_eq!(event.data["version"], version); + assert_eq!(event.data["data"]["slot"], slot.to_string()); + assert_eq!(event.data["data"]["payload_status"], "full"); + } + } + #[test] fn a_topic_silver_does_not_serve_is_refused_on_an_ordinary_connection() { let mut server = server_with(64, LONG_TIMEOUT); @@ -1420,7 +1578,7 @@ mod tests { let mut stream = connect(addr); write!( stream, - "GET /eth/v1/events?topics=head HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n" + "GET /eth/v1/events?topics=chain_reorg HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n" ) .unwrap(); read_to_eof(stream) @@ -1429,7 +1587,7 @@ mod tests { let got = serve(&mut server, client, "400 for an unserved topic"); assert_same_bytes( &got, - b"HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: 47\r\n\r\n{\"code\":400,\"message\":\"unknown topic \\\"head\\\"\"}", + b"HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: 54\r\n\r\n{\"code\":400,\"message\":\"unknown topic \\\"chain_reorg\\\"\"}", ); assert_eq!(subscribers(&server), 0); } diff --git a/crates/beacon_state/data/src/column/roots.rs b/crates/beacon_state/data/src/column/roots.rs index 3230878f..fbf694ac 100644 --- a/crates/beacon_state/data/src/column/roots.rs +++ b/crates/beacon_state/data/src/column/roots.rs @@ -1,7 +1,7 @@ use super::{ColumnGroup, ColumnReader, ColumnSpec, ColumnWriteView}; use crate::{ ring::Id, - types::{B256, SLOTS_PER_HISTORICAL_ROOT, Slot}, + types::{B256, Epoch, SLOTS_PER_EPOCH, SLOTS_PER_HISTORICAL_ROOT, Slot}, }; /// `Vector[Root, SLOTS_PER_HISTORICAL_ROOT]`, written pointwise: `process_slot` @@ -39,6 +39,35 @@ impl RootsView<'_, BlockRoots> { self.get(slot as usize % SLOTS_PER_HISTORICAL_ROOT) } + /// Root at the slot before `epoch` starts, saturating to slot zero. + /// + /// A head at the decision slot supplies its own root without a history + /// read. Otherwise, availability is measured from `state_slot`: a + /// checkpoint state can be ahead of its latest block. Returns `None` + /// for overwritten history. + pub fn duty_dependent_root( + &self, + epoch: Epoch, + head_slot: Slot, + head_root: B256, + state_slot: Slot, + ) -> Option { + let decision_slot = (epoch * SLOTS_PER_EPOCH).saturating_sub(1); + debug_assert!( + decision_slot <= head_slot, + "epoch {epoch} decides at slot {decision_slot}, past the head at {head_slot}" + ); + debug_assert!( + head_slot <= state_slot, + "the state at {state_slot} is behind its head at {head_slot}" + ); + if head_slot == decision_slot { + return Some(head_root); + } + (state_slot - decision_slot <= SLOTS_PER_HISTORICAL_ROOT as u64) + .then(|| self.at_slot(decision_slot)) + } + /// Slot of the block with `root`, if the ring holds it at or below /// `from_slot`. Fork choice is what writes a root here, so a hit means /// "seen and validated". diff --git a/crates/beacon_state/data/src/column/tests.rs b/crates/beacon_state/data/src/column/tests.rs index 1a0a8a1a..008e0fcc 100644 --- a/crates/beacon_state/data/src/column/tests.rs +++ b/crates/beacon_state/data/src/column/tests.rs @@ -458,6 +458,105 @@ fn block_roots_slot_of_scans_back_from_the_given_slot() { assert_eq!(reader.slot_of(&[0xCC; 32], 103), None); } +/// A head in epoch 2 uses decision slots 63 and 31. +#[test] +fn duty_dependent_roots_read_the_slots_below_the_two_epoch_starts() { + const HEAD: B256 = [0x46; 32]; + let mut g = BlockRootsGroup::zeroed_vector(); + let id = { + let mut wv = g.roll_fresh(); + wv.set(31, [0x1F; 32]); + wv.set(60, [0x3C; 32]); + // Empty slots 61..=63 carry slot 60's block root forward. + for slot in 61..=63 { + wv.set(slot, [0x3C; 32]); + } + wv.set(70, HEAD); + wv.commit() + }; + let reader = g.view(id); + + assert_eq!( + reader.duty_dependent_root(2, 70, HEAD, 70), + Some([0x3C; 32]), + "epoch 2 decides at slot 63, an empty slot holding slot 60's block" + ); + assert_eq!( + reader.duty_dependent_root(1, 70, HEAD, 70), + Some([0x1F; 32]), + "epoch 1 decides at slot 31" + ); +} + +/// Both lookups use slot zero in epoch 0; the previous lookup also does so +/// in epoch 1. +#[test] +fn early_epochs_saturate_the_decision_slot_to_genesis() { + const GENESIS: B256 = [0x6E; 32]; + const HEAD: B256 = [0x05; 32]; + let mut g = BlockRootsGroup::zeroed_vector(); + let id = { + let mut wv = g.roll_fresh(); + wv.set(0, GENESIS); + wv.set(31, [0x1F; 32]); + wv.commit() + }; + let reader = g.view(id); + + let at = |epoch, head_slot| reader.duty_dependent_root(epoch, head_slot, HEAD, head_slot); + assert_eq!(at(0, 5), Some(GENESIS), "epoch 0 decides at slot 0"); + assert_eq!(at(1, 40), Some([0x1F; 32]), "epoch 1 decides at slot 31"); + assert_eq!(at(0, 40), Some(GENESIS), "epoch 0 stays at slot 0"); +} + +#[test] +fn a_head_at_slot_zero_decides_its_own_shuffling() { + const HEAD: B256 = [0x11; 32]; + let mut g = BlockRootsGroup::zeroed_vector(); + let id = g.roll_fresh().commit(); + let reader = g.view(id); + + assert_eq!(reader.duty_dependent_root(0, 0, HEAD, 0), Some(HEAD)); + assert_ne!(HEAD, reader.at_slot(0), "the ring holds nothing at slot 0 yet"); +} + +#[cfg(debug_assertions)] +#[test] +#[should_panic(expected = "past the head")] +fn a_decision_slot_above_the_head_is_a_bug() { + let mut g = BlockRootsGroup::zeroed_vector(); + let id = g.roll_fresh().commit(); + g.view(id).duty_dependent_root(3, 70, [0x46; 32], 70); +} + +#[cfg(debug_assertions)] +#[test] +#[should_panic(expected = "behind its head")] +fn a_state_behind_its_head_is_a_bug() { + let mut g = BlockRootsGroup::zeroed_vector(); + let id = g.roll_fresh().commit(); + g.view(id).duty_dependent_root(2, 70, [0x46; 32], 69); +} + +/// At state slot S, the oldest retained slot is S − 8192, inclusive. +#[test] +fn a_decision_slot_is_available_until_the_state_moves_a_ring_past_it() { + const HEAD: B256 = [0x46; 32]; + let mut g = BlockRootsGroup::zeroed_vector(); + let id = { + let mut wv = g.roll_fresh(); + wv.set(31, [0x1F; 32]); + wv.commit() + }; + let reader = g.view(id); + // Epoch 1 decides at slot 31; a state at 8223 still holds it as its + // oldest slot. + let edge = SLOTS_PER_HISTORICAL_ROOT as u64 + 31; + + assert_eq!(reader.duty_dependent_root(1, 70, HEAD, edge), Some([0x1F; 32])); + assert_eq!(reader.duty_dependent_root(1, 70, HEAD, edge + 1), None, "overwritten a slot later"); +} + /// A block's reveal accumulates into the current epoch's bucket; the boundary /// copy seeds the next epoch from it, and the next epoch's reveals accumulate /// on top without disturbing the finished epoch. diff --git a/crates/beacon_state/tile/src/fork_choice/head.rs b/crates/beacon_state/tile/src/fork_choice/head.rs index 5b4bc5d0..2cff563f 100644 --- a/crates/beacon_state/tile/src/fork_choice/head.rs +++ b/crates/beacon_state/tile/src/fork_choice/head.rs @@ -1,5 +1,6 @@ use flux_profiler::timed; use silver_beacon_state_data::{B256, Epoch, MIN_SEED_LOOKAHEAD, SLOTS_PER_EPOCH, Slot}; +use silver_common::PayloadResolution; use super::{ExecutionStatus, ForkChoice, GENESIS_EPOCH, NULL, PayloadStatus, node::PTC_SIZE}; @@ -107,6 +108,11 @@ impl ForkChoice { } } + /// Resolves the node's own payload using the same rule as head selection. + pub fn payload_resolution(&self, idx: usize) -> PayloadResolution { + if self.resolves_to_full(idx) { PayloadResolution::Full } else { PayloadResolution::Empty } + } + #[cfg(any(test, feature = "ef_tests"))] pub fn head_payload_present(&self) -> bool { let Some(ji) = self.find_node_idx(&self.justified_checkpoint.root) else { diff --git a/crates/beacon_state/tile/src/fork_choice/mod.rs b/crates/beacon_state/tile/src/fork_choice/mod.rs index 511226ad..423376f8 100644 --- a/crates/beacon_state/tile/src/fork_choice/mod.rs +++ b/crates/beacon_state/tile/src/fork_choice/mod.rs @@ -12,8 +12,8 @@ mod tests; mod vote; pub use lookup::NodeLookup; -use node::{Branch, NodeCheckpoints, PayloadAxis, PtcVotes}; -pub use node::{ExecutionStatus, ForkChoiceNode, PayloadStatus}; +use node::{Branch, NodeCheckpoints, PtcVotes}; +pub use node::{ExecutionStatus, ForkChoiceNode, PayloadAxis, PayloadStatus}; pub use vote::{VoteTracker, WeightDelta}; /// Pre-allocation hint only — the node table and the state rings both grow @@ -74,6 +74,7 @@ pub struct ForkChoice { pub struct BlockImport { pub slot: Slot, pub block_root: B256, + pub state_root: B256, pub parent_root: B256, pub execution_block_hash: B256, pub justified: Checkpoint, @@ -94,6 +95,7 @@ impl ForkChoice { justified_checkpoint: Checkpoint, finalized_slot: Slot, finalized_block_root: B256, + finalized_state_root: B256, finalized_execution_block_hash: B256, anchor_is_gloas: bool, state_id: StateId, @@ -105,6 +107,7 @@ impl ForkChoice { nodes.push(ForkChoiceNode { slot: finalized_slot, block_root: finalized_block_root, + state_root: finalized_state_root, execution_block_hash: finalized_execution_block_hash, parent_ix: NULL, execution_status: ExecutionStatus::Valid, @@ -172,6 +175,7 @@ impl ForkChoice { self.nodes.push(ForkChoiceNode { slot: b.slot, block_root: b.block_root, + state_root: b.state_root, execution_block_hash: b.execution_block_hash, parent_ix: parent, execution_status: ExecutionStatus::Optimistic, diff --git a/crates/beacon_state/tile/src/fork_choice/node.rs b/crates/beacon_state/tile/src/fork_choice/node.rs index 28db4159..0c242a85 100644 --- a/crates/beacon_state/tile/src/fork_choice/node.rs +++ b/crates/beacon_state/tile/src/fork_choice/node.rs @@ -22,6 +22,9 @@ pub enum PayloadStatus { pub struct ForkChoiceNode { pub slot: Slot, pub block_root: B256, + /// Retained because a reorg can select this block after its bytes have + /// left the cache. + pub state_root: B256, pub execution_block_hash: B256, pub parent_ix: usize, diff --git a/crates/beacon_state/tile/src/fork_choice/tests.rs b/crates/beacon_state/tile/src/fork_choice/tests.rs index 8143516f..1b89bc70 100644 --- a/crates/beacon_state/tile/src/fork_choice/tests.rs +++ b/crates/beacon_state/tile/src/fork_choice/tests.rs @@ -1,4 +1,5 @@ use silver_beacon_state_data::MIN_SEED_LOOKAHEAD; +use silver_common::PayloadResolution; use super::{ vote::{Vote, branch_voted_for}, @@ -33,6 +34,13 @@ fn root(b: u8) -> B256 { r } +/// Use a non-zero state root distinct from the block root. +fn state_root_of(block_root: B256) -> B256 { + let mut r = block_root; + r[31] = 0xFF; + r +} + fn cp(epoch: Epoch, b: u8) -> Checkpoint { Checkpoint { epoch, root: root(b) } } @@ -49,6 +57,7 @@ fn block( BlockImport { slot, block_root, + state_root: state_root_of(block_root), parent_root, execution_block_hash: [0u8; 32], justified: jus, @@ -104,7 +113,17 @@ fn compute_deltas( fn single_chain_head() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); fc.on_block(block(2, root(3), root(2), jus, fin)); @@ -116,7 +135,17 @@ fn single_chain_head() { fn fork_heavier_wins() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); fc.on_block(block(1, root(3), root(1), jus, fin)); @@ -136,7 +165,17 @@ fn two_pass_weight_correctness() { // sibling loses weight. let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); // root(1) → root(2) [idx 1] and root(3) [idx 2] fc.on_block(block(1, root(2), root(1), jus, fin)); @@ -162,7 +201,17 @@ fn two_pass_weight_correctness() { fn prune_below_finalized() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); fc.on_block(block(2, root(3), root(2), jus, fin)); @@ -184,7 +233,17 @@ fn prune_drops_later_imported_siblings() { // the promoted delta. let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); // finalized-to-be fc.on_block(block(1, root(3), root(1), jus, fin)); // sibling, imported after @@ -207,7 +266,17 @@ fn prune_drops_later_imported_siblings() { fn deltas_moving_votes() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); let mut votes = vec![Vote::default(); 16]; @@ -240,7 +309,17 @@ fn deltas_different_votes() { // Each validator votes for a different block. let fin = cp(0, 100); let jus = cp(0, 100); - let mut fc = ForkChoice::init(fin, jus, 0, root(100), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(100), + state_root_of(root(100)), + [0u8; 32], + false, + test_state_id(), + 0, + ); for i in 1..=16u8 { fc.on_block(block(i as u64, root(i), root(100), jus, fin)); @@ -271,7 +350,17 @@ fn deltas_different_votes() { fn deltas_move_out_of_tree() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); let mut votes = vec![Vote::default(); 16]; let mut balances = vec![0u64; 16]; @@ -304,7 +393,17 @@ fn deltas_move_out_of_tree() { fn deltas_changing_balances() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); let mut votes = vec![Vote::default(); 16]; @@ -335,7 +434,17 @@ fn deltas_balance_change_no_vote_change() { // Balances change but votes don't — still need deltas. let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); let mut votes = vec![Vote::default(); 16]; let mut old_bal = vec![0u64; 16]; @@ -358,7 +467,17 @@ fn deltas_balance_change_no_vote_change() { fn split_tie_breaker_no_attestations() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); // Two blocks at slot 1 forking from genesis. root(2) < root(3). fc.on_block(block(1, root(2), root(1), jus, fin)); @@ -373,7 +492,17 @@ fn split_tie_breaker_no_attestations() { fn shorter_chain_but_heavier_weight() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); // Long chain: root(1) → root(2) → root(3) → root(4). fc.on_block(block(1, root(2), root(1), jus, fin)); @@ -398,7 +527,17 @@ fn shorter_chain_but_heavier_weight() { fn on_block_duplicate() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); assert_eq!(fc.nodes.len(), 2); @@ -412,7 +551,17 @@ fn on_block_duplicate() { fn on_block_unknown_parent() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); // root(99) is not known. fc.on_block(block(1, root(2), root(99), jus, fin)); @@ -426,7 +575,17 @@ fn on_block_unknown_parent() { fn node_lookup_equivalence_and_prune() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); for i in 2..=20u8 { fc.on_block(block(i as u64, root(i), root(i - 1), jus, fin)); } @@ -451,7 +610,17 @@ fn node_lookup_equivalence_and_prune() { fn proposer_boost_flips_then_expires() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(1, root(2), root(1), jus, fin)); // idx 1 fc.on_block(block(1, root(3), root(1), jus, fin)); // idx 2 @@ -484,7 +653,17 @@ fn proposer_boost_flips_then_expires() { fn compute_deltas_dirty_matches_full() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); for i in 2..=5u8 { fc.on_block(block(1, root(i), root(1), jus, fin)); } @@ -527,7 +706,17 @@ fn compute_deltas_dirty_matches_full() { #[test] fn viability_genesis_exception() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); // Prior-epoch node with a stale (epoch 1) voting source. fc.on_block(block(64, root(2), root(1), cp(1, 9), g)); let a = fc.find_node_idx(&root(2)).unwrap(); @@ -546,7 +735,17 @@ fn viability_unrealized_justified_and_plus_two() { // Genesis finalized isolates the justified rule; store justified // promoted to epoch 2 (node root(2) at slot 64). let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(64, root(2), root(1), cp(2, 2), g)); // A @ epoch 2 fc.justified_checkpoint = cp(2, 2); // C @ epoch 2, child of A, realized justified stale (epoch 1). @@ -583,7 +782,17 @@ fn viability_unrealized_justified_and_plus_two() { #[test] fn gloas_vote_branch_classification() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); fc.on_block(gloas_block(5, root(2), root(1), g, g, PayloadStatus::Full, true)); let gl = fc.node(fc.find_node_idx(&root(2)).unwrap()); assert_eq!(branch_voted_for(gl, 6, true), PayloadStatus::Full); @@ -601,7 +810,17 @@ fn gloas_vote_branch_classification() { #[test] fn gloas_resolves_heavier_payload_branch() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); // A (idx1), then a FULL-edge child C_f (idx2) and EMPTY-edge child C_e (idx3). fc.on_block(gloas_block(1, root(2), root(1), g, g, PayloadStatus::Full, true)); fc.on_block(gloas_block(2, root(3), root(2), g, g, PayloadStatus::Full, true)); @@ -621,6 +840,13 @@ fn gloas_resolves_heavier_payload_branch() { fc.weight_deltas = d; fc.apply_score_changes(); assert_eq!(fc.find_head(), root(4)); + let parent = fc.find_node_idx(&root(2)).unwrap(); + assert_eq!(fc.payload_resolution(parent), PayloadResolution::Empty); + assert_eq!( + fc.payload_resolution(head_idx(&fc)), + PayloadResolution::Full, + "the selected child's own payload is full despite its empty parent edge" + ); } /// On an exactly-tied payload split, `should_extend_payload` decides: with @@ -629,7 +855,17 @@ fn gloas_resolves_heavier_payload_branch() { #[test] fn gloas_tie_broken_by_should_extend_payload() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); fc.on_block(gloas_block(1, root(2), root(1), g, g, PayloadStatus::Full, true)); fc.on_block(gloas_block(2, root(3), root(2), g, g, PayloadStatus::Full, true)); // C_f fc.on_block(gloas_block(2, root(4), root(2), g, g, PayloadStatus::Empty, true)); // C_e @@ -661,7 +897,17 @@ fn gloas_tie_broken_by_should_extend_payload() { fn shuffling_dependent_root_is_ancestor_at_dependent_slot() { let fin = cp(0, 1); let jus = cp(0, 1); - let mut fc = ForkChoice::init(fin, jus, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + fin, + jus, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); let last_of_epoch0 = SLOTS_PER_EPOCH - 1; // Branch A: genesis → 2 (slot 31) → 3 (slot 33) and 6 (slot 34). fc.on_block(block(last_of_epoch0, root(2), root(1), jus, fin)); @@ -693,7 +939,17 @@ fn shuffling_dependent_root_is_ancestor_at_dependent_slot() { #[test] fn gloas_unverified_payload_forces_empty() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); fc.on_block(gloas_block(1, root(2), root(1), g, g, PayloadStatus::Full, false)); fc.on_block(gloas_block(2, root(3), root(2), g, g, PayloadStatus::Full, true)); // C_f fc.on_block(gloas_block(2, root(4), root(2), g, g, PayloadStatus::Empty, true)); // C_e @@ -719,7 +975,17 @@ fn gloas_unverified_payload_forces_empty() { #[test] fn gloas_empty_survives_full_invalid() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); fc.on_block(gloas_block(1, root(2), root(1), g, g, PayloadStatus::Full, true)); fc.on_block(gloas_block(2, root(3), root(2), g, g, PayloadStatus::Full, true)); // C_f fc.on_block(gloas_block(2, root(4), root(2), g, g, PayloadStatus::Empty, true)); // C_e @@ -748,7 +1014,17 @@ fn gloas_empty_survives_full_invalid() { #[test] fn gloas_boost_is_pending_not_empty() { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 8); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); // Boosted current-slot block, envelope not yet revealed. fc.on_block(gloas_block(1, root(2), root(1), g, g, PayloadStatus::Full, false)); fc.proposer_boost_root = root(2); @@ -767,6 +1043,59 @@ fn gloas_boost_is_pending_not_empty() { assert!(fc.head_payload_present()); } +fn head_idx(fc: &ForkChoice) -> usize { + fc.find_node_idx(&fc.find_head()).unwrap() +} + +#[test] +fn payload_resolution_follows_the_selected_node() { + let g = cp(0, 1); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Full, "pre-Gloas anchor"); + + fc.on_block(block(1, root(2), root(1), g, g)); + assert_eq!(fc.find_head(), root(2)); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Full, "pre-Gloas block"); + + fc.on_block(gloas_block(2, root(3), root(2), g, g, PayloadStatus::Full, false)); + assert_eq!(fc.find_head(), root(3)); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Empty, "no envelope yet"); + + fc.mark_payload_verified(&root(3)); + assert_eq!(fc.find_head(), root(3)); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Full); +} + +#[test] +fn a_gloas_anchor_resolves_empty_until_its_envelope_is_verified() { + let g = cp(0, 1); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + true, + test_state_id(), + 8, + ); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Empty); + + fc.mark_payload_verified(&root(1)); + assert_eq!(fc.payload_resolution(head_idx(&fc)), PayloadResolution::Full); +} + /// Two branches meeting at 2@slot 2, with heads level at slot 10 but at /// *different depths* — the left one skips slots 3..9 outright: /// @@ -776,7 +1105,17 @@ fn gloas_boost_is_pending_not_empty() { /// ``` fn skipped_slot_forks() -> ForkChoice { let g = cp(0, 1); - let mut fc = ForkChoice::init(g, g, 0, root(1), [0u8; 32], false, test_state_id(), 0); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 0, + ); fc.on_block(block(2, root(2), root(1), g, g)); fc.on_block(block(10, root(4), root(2), g, g)); fc.on_block(block(9, root(5), root(2), g, g)); @@ -813,3 +1152,33 @@ fn lca_of_a_pruned_head_falls_back_to_finalized() { assert_eq!(fc.lca_slot(root(99), root(6)), Some(SLOTS_PER_EPOCH), "unknown old head"); assert_eq!(fc.lca_slot(root(4), root(99)), None, "unknown new head says nothing"); } + +#[test] +fn a_valid_payload_validates_its_ancestors_and_no_sibling() { + let g = cp(0, 1); + let mut fc = ForkChoice::init( + g, + g, + 0, + root(1), + state_root_of(root(1)), + [0u8; 32], + false, + test_state_id(), + 8, + ); + fc.on_block(block(1, root(2), root(1), g, g)); // A + fc.on_block(block(2, root(3), root(2), g, g)); // C, A's child + fc.on_block(block(1, root(4), root(1), g, g)); // B, A's sibling + let valid = |fc: &ForkChoice, b: u8| { + fc.nodes[fc.find_node_idx(&root(b)).unwrap()].execution_status == ExecutionStatus::Valid + }; + assert!(!valid(&fc, 2) && !valid(&fc, 3) && !valid(&fc, 4), "imports start optimistic"); + + fc.on_payload_valid(&root(3)); + + assert!(valid(&fc, 3)); + assert!(valid(&fc, 2), "A is validated through its child"); + assert!(!valid(&fc, 4), "a sibling is not on the walk"); + assert!(valid(&fc, 1), "the anchor was valid from init"); +} diff --git a/crates/beacon_state/tile/src/tile.rs b/crates/beacon_state/tile/src/tile.rs index b9abcd5a..b5ad969e 100644 --- a/crates/beacon_state/tile/src/tile.rs +++ b/crates/beacon_state/tile/src/tile.rs @@ -11,9 +11,10 @@ use silver_beacon_state_data::{ Slot, SlotState, SpecConfig, StateId, }; use silver_common::{ - BeaconStateEvent, BlockSource, DataColumnsEvent, DataKind, EngineResp, GossipTopic, - NewGossipMsg, Origin, PayloadValidationStatus, ReplayBlock, RequestId, RpcInbound, RpcResponse, - RpcResponseInbound, SilverSpine, SyncUpdate, TRandomAccess, TRead, hex32, + BeaconStateEvent, BlockSource, DataColumnsEvent, DataKind, EngineResp, GossipTopic, HeadRoots, + NewGossipMsg, Origin, PayloadResolution, PayloadValidationStatus, ReplayBlock, RequestId, + RpcInbound, RpcResponse, RpcResponseInbound, SilverSpine, SyncUpdate, TRandomAccess, TRead, + hex32, ssz_view::STATUS_V2_SIZE, ticker::{MAXIMUM_GOSSIP_CLOCK_DISPARITY, SlotTicker, TickEvent}, }; @@ -111,6 +112,21 @@ impl Debug for Feedback { } } +/// Resolved once so each Status uses one fork's head metadata. +#[derive(Clone, Copy)] +struct SelectedHead { + observation: HeadObservation, + idx: usize, +} + +/// Head changes that require a Status even without an import or slot tick. +#[derive(Clone, Copy, PartialEq, Eq)] +struct HeadObservation { + root: B256, + optimistic: bool, + payload: PayloadResolution, +} + pub struct BeaconStateTile { sync_target: SyncUpdate, ticker: SlotTicker, @@ -148,6 +164,9 @@ pub struct BeaconStateTile { precomputed_epochs: PrecomputedEpochs, last_seen_head_root: B256, + /// Kept separate from the reorg marker: an early Status must not hide a + /// reorg that the end-of-loop check has yet to report. + emitted_head: HeadObservation, initial_status_emitted: bool, cached_fork_digest: Option<(Epoch, [u8; 4])>, @@ -225,6 +244,11 @@ impl BeaconStateTile { last_applied_block_root: [0u8; 32], precomputed_epochs: PrecomputedEpochs::default(), last_seen_head_root: [0u8; 32], + emitted_head: HeadObservation { + root: [0u8; 32], + optimistic: true, + payload: PayloadResolution::Empty, + }, initial_status_emitted: false, cached_fork_digest: None, stf_scratch: stf::StfScratch::new(val_cap), @@ -316,7 +340,7 @@ impl BeaconStateTile { // post-bootstrap `process_slot` hashes that canonical state and a // patched value would shift the result. let anchor_is_gloas = self.state.read_view(anchor).is_gloas(); - let (block_root, execution_block_hash) = { + let (header, block_root, execution_block_hash) = { let rv = self.state.read_view(anchor); let state_root = ssz_hash::hash_tree_root_state(&rv); let mut header = rv.slot.state().latest_block_header; @@ -328,18 +352,21 @@ impl BeaconStateTile { } else { rv.slot.state().latest_execution_payload_header.block_hash }; - (ssz_hash::hash_tree_root_block_header(&header), execution_block_hash) + (header, ssz_hash::hash_tree_root_block_header(&header), execution_block_hash) }; let trusted = Checkpoint { epoch: slot.div_ceil(SLOTS_PER_EPOCH), root: block_root }; self.last_applied_block_root = block_root; self.last_seen_head_root = block_root; + // A checkpoint state can be ahead of its latest block. Peers need + // the block's slot in Status. self.fork_choice = ForkChoice::init( trusted, trusted, - slot, + header.slot, block_root, + header.state_root, execution_block_hash, anchor_is_gloas, anchor, @@ -410,19 +437,11 @@ impl BeaconStateTile { ); } - fn status_payload(&mut self, head_root: B256, head_idx: Option) -> [u8; STATUS_V2_SIZE] { + fn status_payload(&mut self, head_root: B256, head_idx: usize) -> [u8; STATUS_V2_SIZE] { let fork_digest = self.fork_digest(); - - let (slot, mut finalized) = match head_idx { - Some(idx) => { - let n = self.fork_choice.node(idx); - (n.slot, n.checkpoints.finalized) - } - None => ( - self.slot_state_at(self.last_applied).latest_block_header.slot, - self.head_finalized_checkpoint(), - ), - }; + let node = self.fork_choice.node(head_idx); + let slot = node.slot; + let mut finalized = node.checkpoints.finalized; if finalized.root == [0u8; 32] { // Genesis placeholder: the head state's finalized root is zero until @@ -451,19 +470,67 @@ impl BeaconStateTile { self.slot_state_at(self.last_applied).latest_block_header.slot } - fn status_event(&mut self) -> BeaconStateEvent { - let head_root = self.fork_choice.find_head(); - let head_idx = self.fork_choice.find_node_idx(&head_root); - let head_optimistic = head_idx.is_none_or(|idx| { - self.fork_choice.node(idx).execution_status != ExecutionStatus::Valid - }); + fn selected_head(&self) -> SelectedHead { + let root = self.fork_choice.find_head(); + let idx = + self.fork_choice.find_node_idx(&root).expect("find_head returns a node-resident root"); + let optimistic = self.fork_choice.node(idx).execution_status != ExecutionStatus::Valid; + let payload = self.fork_choice.payload_resolution(idx); + SelectedHead { observation: HeadObservation { root, optimistic, payload }, idx } + } + + /// Overwritten checkpoint history makes the whole root bundle unavailable; + /// partial metadata cannot describe the head. + fn head_roots(&self, head: SelectedHead) -> HeadRoots { + let node = self.fork_choice.node(head.idx); + let epoch = node.slot / SLOTS_PER_EPOCH; + let view = self.state.read_view(node.state_id); + let state_slot = view.slot.state().slot; + let dependent = |epoch| { + view.block_roots.duty_dependent_root( + epoch, + node.slot, + head.observation.root, + state_slot, + ) + }; + match (dependent(epoch.saturating_sub(1)), dependent(epoch)) { + (Some(previous), Some(current)) => HeadRoots { + state_root: node.state_root, + previous_duty_dependent_root: previous, + current_duty_dependent_root: current, + }, + _ => HeadRoots::default(), + } + } + fn status_event(&mut self, head: SelectedHead) -> BeaconStateEvent { BeaconStateEvent::Status { - ssz: self.status_payload(head_root, head_idx), + ssz: self.status_payload(head.observation.root, head.idx), latest_block_slot: self.last_applied_block_slot(), wall_slot: self.ticker.current_slot(), - head_optimistic, + head_optimistic: head.observation.optimistic, enr_fork_id: self.enr_fork_id(), + head_roots: self.head_roots(head), + head_payload: head.observation.payload, + } + } + + pub(super) fn publish_status(&mut self, producers: &mut Producers) { + self.publish_selected_head(self.selected_head(), producers); + } + + fn publish_selected_head(&mut self, head: SelectedHead, producers: &mut Producers) { + self.emitted_head = head.observation; + let event = self.status_event(head); + producers.produce(event); + } + + /// Covers changes since the last Status, including execution verdicts. + fn publish_status_on_head_change(&mut self, producers: &mut Producers) { + let head = self.selected_head(); + if head.observation != self.emitted_head { + self.publish_selected_head(head, producers); } } @@ -615,7 +682,7 @@ impl BeaconStateTile { let prev_head = self.fork_choice.find_head(); let advanced = self.slot_tick(slot); if advanced || self.fork_choice.find_head() != prev_head { - adapter.produce(self.status_event()); + self.publish_status(&mut adapter.producers); } } TickEvent::StateAdvance(slot) => self.on_state_advance(slot), @@ -734,7 +801,7 @@ impl BeaconStateTile { } ReplayBlock::Done => { producers.produce(BeaconStateEvent::ReplayComplete); - producers.produce(self.status_event()); + self.publish_status(producers); } } } @@ -898,7 +965,7 @@ impl Tile for BeaconStateTile { fn loop_body(&mut self, adapter: &mut SpineAdapter) { if !self.initial_status_emitted { tracing::info!("producing initial status"); - adapter.produce(self.status_event()); + self.publish_status(&mut adapter.producers); self.initial_status_emitted = true; } @@ -910,6 +977,7 @@ impl Tile for BeaconStateTile { if self.fork_choice.take_head_moved() { self.try_detect_reorg(&mut adapter.producers); + self.publish_status_on_head_change(&mut adapter.producers); } } } diff --git a/crates/beacon_state/tile/src/tile/block.rs b/crates/beacon_state/tile/src/tile/block.rs index da497c14..ad0a04da 100644 --- a/crates/beacon_state/tile/src/tile/block.rs +++ b/crates/beacon_state/tile/src/tile/block.rs @@ -509,6 +509,7 @@ impl BeaconStateTile { self.fork_choice.on_block(BlockImport { slot: parsed.header.slot, block_root: parsed.block_root, + state_root: parsed.header.state_root, parent_root: parsed.header.parent_root, execution_block_hash, justified, diff --git a/crates/beacon_state/tile/src/tile/orphan_pool.rs b/crates/beacon_state/tile/src/tile/orphan_pool.rs index 8f670c7a..caa0c5f9 100644 --- a/crates/beacon_state/tile/src/tile/orphan_pool.rs +++ b/crates/beacon_state/tile/src/tile/orphan_pool.rs @@ -147,7 +147,7 @@ impl BeaconStateTile { self.replay_orphans(root, producers); self.drain_pending_envelope(root, producers); } - producers.produce(self.status_event()); + self.publish_status(producers); } pub(super) fn park_block( diff --git a/crates/beacon_state/tile/src/tile/tests.rs b/crates/beacon_state/tile/src/tile/tests.rs index e0f4544b..e82625da 100644 --- a/crates/beacon_state/tile/src/tile/tests.rs +++ b/crates/beacon_state/tile/src/tile/tests.rs @@ -11,7 +11,8 @@ use silver_beacon_state_data::{ }; use silver_common::{ BlockStage, EngineNewPayloadResp, GossipTopic, LOCAL_GOSSIP_STREAM_ID, MessageId, P2pStreamId, - PeerEvent, StreamProtocol, SyncNeed, TCache, TCacheProducer, TCacheRead, TProducer, + PayloadResolution, PayloadValidationStatus, 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, @@ -31,7 +32,7 @@ use super::{ }; use crate::{ error::{PrecheckError, RejectReason}, - fork_choice::{BlockImport, PayloadStatus}, + fork_choice::{BlockImport, PayloadAxis, PayloadStatus}, merkle, ssz_hash, stf::AttestationVote, test_signing, @@ -40,6 +41,14 @@ use crate::{ const MAX_EFFECTIVE_BALANCE: u64 = 32_000_000_000; const TEST_RING_BYTES: usize = 1 << 20; const ANCHOR_ROOT: B256 = [0x01u8; 32]; +/// Non-zero so the anchor has complete head metadata. +const ANCHOR_STATE_ROOT: B256 = [0xA1u8; 32]; + +fn state_root_of(block_root: B256) -> B256 { + let mut r = block_root; + r[31] = 0xFF; + r +} /// Byte position of the body inside a `SignedBeaconBlock`. const BODY: usize = SIGNED_BEACON_BLOCK_MIN; @@ -130,6 +139,7 @@ fn make_tile_with_gossip( (tile, gossip, rpc) } +/// Keeps the replay producer alive for tests that feed cached block bytes. fn make_tile_with_producers( wall_slot: u64, state: BeaconState, @@ -272,8 +282,17 @@ fn arm_tile_state( tile.sync_target = SyncUpdate::Following; let cp = Checkpoint { epoch: 0, root: ANCHOR_ROOT }; - tile.fork_choice = - ForkChoice::init(cp, cp, start_slot, ANCHOR_ROOT, [0u8; 32], false, anchor, seeds.len()); + tile.fork_choice = ForkChoice::init( + cp, + cp, + start_slot, + ANCHOR_ROOT, + ANCHOR_STATE_ROOT, + [0u8; 32], + false, + anchor, + seeds.len(), + ); let view = tile.state.read_view(anchor); tile.shuffling_cache.ensure_window(&view, start_slot / SLOTS_PER_EPOCH); @@ -453,7 +472,8 @@ fn slot_advance_crosses_two_epoch_boundaries() { fn status_event_carries_the_head_s_execution_status() { const CHILD_ROOT: B256 = [0x0C; 32]; - let head_optimistic = |tile: &mut BeaconStateTile| match tile.status_event() { + let head_optimistic = |tile: &mut BeaconStateTile| match tile.status_event(tile.selected_head()) + { BeaconStateEvent::Status { ssz, head_optimistic, .. } => { assert_eq!(*StatusView::head_root(&ssz), tile.fork_choice.find_head()); head_optimistic @@ -465,17 +485,26 @@ fn status_event_carries_the_head_s_execution_status() { seed_tile(&mut tile, 4, 10); assert!(!head_optimistic(&mut tile), "the trusted anchor is valid"); + // The child's post-state sits at its own slot, as an import leaves it. + let child_state = { + let anchor = tile.last_applied; + let mut g = tile.state.write(); + let mut sw = g.slot_states.roll_from(anchor.slot_idx); + sw.state_mut().slot = 11; + StateId { slot_idx: sw.commit(), ..anchor } + }; let anchor_cp = Checkpoint { epoch: 0, root: ANCHOR_ROOT }; tile.fork_choice.on_block(BlockImport { slot: 11, block_root: CHILD_ROOT, + state_root: state_root_of(CHILD_ROOT), parent_root: ANCHOR_ROOT, execution_block_hash: [0u8; 32], justified: anchor_cp, finalized: anchor_cp, unrealized_justified: anchor_cp, unrealized_finalized: anchor_cp, - state_id: tile.last_applied, + state_id: child_state, bid_block_hash: [0u8; 32], parent_payload_status: PayloadStatus::Full, payload_verified: true, @@ -488,6 +517,567 @@ fn status_event_carries_the_head_s_execution_status() { assert!(!head_optimistic(&mut tile)); } +/// A wins the equal-weight root tie-break, regardless of import order. +const A_ROOT: B256 = [0xAA; 32]; +const B_ROOT: B256 = [0x0B; 32]; +/// Decision slots for a head in epoch 2: `start(2) - 1` and `start(1) - 1`. +const CURRENT_DECISION_SLOT: Slot = 63; +const PREVIOUS_DECISION_SLOT: Slot = 31; + +#[derive(Debug, PartialEq, Eq)] +struct StatusHead { + root: B256, + slot: Slot, + optimistic: bool, + roots: HeadRoots, + payload: PayloadResolution, +} + +struct Published(Vec); + +impl Published { + fn drain(sink: &mut SpineAdapter) -> Self { + let mut events = Vec::new(); + sink.consume(|event: BeaconStateEvent, _| events.push(event)); + Self(events) + } + + fn heads(&self) -> Vec { + self.0 + .iter() + .filter_map(|event| match event { + BeaconStateEvent::Status { + ssz, head_optimistic, head_roots, head_payload, .. + } => Some(StatusHead { + root: *StatusView::head_root(ssz), + slot: StatusView::head_slot(ssz), + optimistic: *head_optimistic, + roots: *head_roots, + payload: *head_payload, + }), + _ => None, + }) + .collect() + } + + fn reorgs(&self) -> Vec { + self.0 + .iter() + .filter_map(|event| match event { + BeaconStateEvent::Reorg { lca_slot } => Some(*lca_slot), + _ => None, + }) + .collect() + } + + fn last_head(&self) -> StatusHead { + self.heads().pop().expect("expected a published head") + } +} + +/// Synthetic fork-choice setup with spine injection and publication capture. +/// `crank` runs the tile loop; setup helpers also call tile methods directly. +struct HeadRig { + sink: SpineAdapter, + adapter: SpineAdapter, + tile: BeaconStateTile, + anchor: StateId, + _gossip: TProducer, + _rpc: TProducer, + _spine: TestSpine, +} + +impl HeadRig { + /// Anchored at slot 70 (epoch 2) and cranked once, so every cursor has + /// snapped and the startup Status is behind us. + fn new() -> Self { + const ANCHOR_SLOT: Slot = 70; + let (mut tile, gossip, rpc, mut spine, adapter) = tile_with_producers(ANCHOR_SLOT); + seed_tile(&mut tile, 8, ANCHOR_SLOT); + tile.sync_target = SyncUpdate::Following; + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); + sink.consume(|_: BeaconStateEvent, _| {}); + + // Distinct anchor roots expose accidental reads from a child's history. + let anchor = { + let base = tile.last_applied; + let mut g = tile.state.write(); + let mut w = g.block_roots.roll_from(base.block_roots_idx); + w.set(PREVIOUS_DECISION_SLOT as u32, ANCHOR_PREVIOUS); + w.set(CURRENT_DECISION_SLOT as u32, ANCHOR_CURRENT); + StateId { block_roots_idx: w.commit(), ..base } + }; + tile.last_applied = anchor; + tile.state.publish_state_id(anchor); + let cp = Checkpoint { epoch: 0, root: ANCHOR_ROOT }; + tile.fork_choice = ForkChoice::init( + cp, + cp, + ANCHOR_SLOT, + ANCHOR_ROOT, + ANCHOR_STATE_ROOT, + [0u8; 32], + false, + anchor, + 8, + ); + let mut rig = + Self { sink, adapter, tile, anchor, _gossip: gossip, _rpc: rpc, _spine: spine }; + let _ = rig.crank(); + rig + } + + fn crank(&mut self) -> Published { + self.tile.loop_body(&mut self.adapter); + self.drain() + } + + fn drain(&mut self) -> Published { + Published::drain(&mut self.sink) + } + + /// Builds a synthetic state with distinct roots at the two decision slots. + fn post_state( + &mut self, + parent: StateId, + slot: Slot, + previous: B256, + current: B256, + ) -> StateId { + let mut g = self.tile.state.write(); + let mut sw = g.slot_states.roll_from(parent.slot_idx); + sw.state_mut().slot = slot; + let slot_idx = sw.commit(); + let mut w = g.block_roots.roll_from(parent.block_roots_idx); + w.set(PREVIOUS_DECISION_SLOT as u32, previous); + w.set(CURRENT_DECISION_SLOT as u32, current); + let block_roots_idx = w.commit(); + StateId { slot_idx, block_roots_idx, ..parent } + } + + /// Seeds fork choice and runs the accept notification without parsing a + /// block or executing its state transition. + fn import(&mut self, block_root: B256, slot: Slot, previous: B256, current: B256) -> StateId { + let anchor = self.anchor; + self.import_child(block_root, slot, ANCHOR_ROOT, anchor, previous, current) + } + + fn import_child( + &mut self, + block_root: B256, + slot: Slot, + parent_root: B256, + parent_state: StateId, + previous: B256, + current: B256, + ) -> StateId { + let payload = PayloadAxis { + bid_block_hash: [0u8; 32], + parent_status: PayloadStatus::Full, + verified: true, + is_gloas: false, + }; + self.import_node(block_root, slot, parent_root, parent_state, previous, current, payload) + } + + fn import_gloas( + &mut self, + block_root: B256, + slot: Slot, + previous: B256, + current: B256, + payload_verified: bool, + ) -> StateId { + let payload = PayloadAxis { + bid_block_hash: block_root, + parent_status: PayloadStatus::Full, + verified: payload_verified, + is_gloas: true, + }; + let anchor = self.anchor; + self.import_node(block_root, slot, ANCHOR_ROOT, anchor, previous, current, payload) + } + + /// A verified Gloas block extending `parent_root`'s empty payload. + fn import_empty_child( + &mut self, + block_root: B256, + slot: Slot, + parent_root: B256, + parent_state: StateId, + previous: B256, + current: B256, + ) -> StateId { + let payload = PayloadAxis { + bid_block_hash: block_root, + parent_status: PayloadStatus::Empty, + verified: true, + is_gloas: true, + }; + self.import_node(block_root, slot, parent_root, parent_state, previous, current, payload) + } + + fn import_node( + &mut self, + block_root: B256, + slot: Slot, + parent_root: B256, + parent_state: StateId, + previous: B256, + current: B256, + payload: PayloadAxis, + ) -> StateId { + let state_id = self.post_state(parent_state, slot, previous, current); + let anchor_cp = Checkpoint { epoch: 0, root: ANCHOR_ROOT }; + self.tile.fork_choice.on_block(BlockImport { + slot, + block_root, + state_root: state_root_of(block_root), + parent_root, + execution_block_hash: block_root, + justified: anchor_cp, + finalized: anchor_cp, + unrealized_justified: anchor_cp, + unrealized_finalized: anchor_cp, + state_id, + bid_block_hash: payload.bid_block_hash, + parent_payload_status: payload.parent_status, + payload_verified: payload.verified, + is_gloas: payload.is_gloas, + }); + self.tile.last_applied = state_id; + self.tile.last_applied_block_root = block_root; + self.tile.recompute_head(); + self.tile.on_accept(Some(block_root), &mut self.adapter.producers); + state_id + } + + fn verdict(&mut self, block_root: B256, status: PayloadValidationStatus) { + self.sink.produce(EngineResp::NewPayload(EngineNewPayloadResp { + block_root, + status, + latest_valid_hash: [0u8; 32], + })); + } + + fn advance_to_slot(&mut self, slot: Slot) { + self.tile.ticker.set_since_genesis_ms(slot * 12_000); + } + + /// Bypasses gossip validation and supplies the recomputation normally + /// performed after committing PTC votes. + fn ptc_majority(&mut self, block_root: B256) { + self.tile.fork_choice.record_ptc_votes(&block_root, &[u64::MAX; 8], true, true); + self.tile.recompute_head(); + } + + fn vote_for(&mut self, block_root: B256, validators: std::ops::Range) { + self.vote(block_root, validators, 71, true); + } + + /// Slot-72 votes can distinguish the slot-71 block's empty/full payload. + fn vote_on_payload( + &mut self, + block_root: B256, + validators: std::ops::Range, + present: bool, + ) { + self.vote(block_root, validators, 72, present); + } + + fn vote( + &mut self, + block_root: B256, + validators: std::ops::Range, + attestation_slot: Slot, + payload_present: bool, + ) { + let n = self.tile.head_validator_count(); + for validator in validators { + self.tile.fork_choice.record_vote( + &AttestationVote { + validator, + block_root, + target_epoch: 2, + attestation_slot, + payload_present, + }, + n, + ); + } + } +} + +/// Distinct roots expose snapshots that mix metadata from different forks. +const A_PREVIOUS: B256 = [0xA1; 32]; +const A_CURRENT: B256 = [0xA2; 32]; +const B_PREVIOUS: B256 = [0xB1; 32]; +const B_CURRENT: B256 = [0xB2; 32]; +const ANCHOR_PREVIOUS: B256 = [0x71; 32]; +const ANCHOR_CURRENT: B256 = [0x72; 32]; + +fn head_a(optimistic: bool) -> StatusHead { + StatusHead { + root: A_ROOT, + slot: 71, + optimistic, + roots: HeadRoots { + state_root: state_root_of(A_ROOT), + previous_duty_dependent_root: A_PREVIOUS, + current_duty_dependent_root: A_CURRENT, + }, + payload: PayloadResolution::Full, + } +} + +fn head_b(optimistic: bool) -> StatusHead { + StatusHead { + root: B_ROOT, + slot: 71, + optimistic, + roots: HeadRoots { + state_root: state_root_of(B_ROOT), + previous_duty_dependent_root: B_PREVIOUS, + current_duty_dependent_root: B_CURRENT, + }, + payload: PayloadResolution::Full, + } +} + +fn head_anchor() -> StatusHead { + StatusHead { + root: ANCHOR_ROOT, + slot: 70, + optimistic: false, + roots: HeadRoots { + state_root: ANCHOR_STATE_ROOT, + previous_duty_dependent_root: ANCHOR_PREVIOUS, + current_duty_dependent_root: ANCHOR_CURRENT, + }, + payload: PayloadResolution::Full, + } +} + +#[test] +fn startup_status_uses_the_seeded_anchor_on_both_forks() { + for is_gloas in [false, true] { + for state_slot in [0, 70] { + let mut epoch = EpochState::default(); + if is_gloas { + epoch.fork.current_version = Immutable::default().gloas_fork_version; + } + let state = + BeaconState::for_test(EpochStateFinalized::from_state(epoch), &[], state_slot); + let (mut tile, _gp, _rp, mut spine, mut adapter) = + tile_with_producers_on(state_slot, state, SpecConfig::mainnet()); + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); + sink.consume(|_: BeaconStateEvent, _| {}); + let root = tile.head_block_root(); + let state_root = + ssz_hash::hash_tree_root_state(&tile.state.read_view(tile.last_applied)); + assert_ne!(root, [0; 32], "the constructor seeded a real anchor"); + + tile.loop_body(&mut adapter); + + assert_eq!( + Published::drain(&mut sink).last_head(), + StatusHead { + root, + slot: 0, + optimistic: false, + roots: HeadRoots { + state_root, + previous_duty_dependent_root: root, + current_duty_dependent_root: root, + }, + payload: if is_gloas { + PayloadResolution::Empty + } else { + PayloadResolution::Full + }, + }, + "startup at state slot {state_slot}, Gloas: {is_gloas}" + ); + } + } +} + +#[test] +fn an_import_is_not_republished_by_idle_iterations() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + + assert_eq!(rig.crank().heads(), [head_a(true)]); + for _ in 0..2 { + assert!(rig.crank().heads().is_empty(), "an idle iteration has nothing new to report"); + } +} + +#[test] +fn a_non_head_import_publishes_a_status_naming_the_head_it_did_not_take() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + let _ = rig.crank(); + + rig.import(B_ROOT, 71, B_PREVIOUS, B_CURRENT); + assert_eq!(rig.crank().last_head(), head_a(true)); +} + +#[test] +fn a_verdict_after_an_import_observation_is_not_lost() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + + rig.verdict(A_ROOT, PayloadValidationStatus::Valid); + assert_eq!(rig.crank().last_head(), head_a(false)); +} + +#[test] +fn an_invalid_verdict_publishes_the_snapshot_of_the_head_it_moved_to() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + rig.import(B_ROOT, 71, B_PREVIOUS, B_CURRENT); + let _ = rig.crank(); + + rig.verdict(A_ROOT, PayloadValidationStatus::Invalid); + let events = rig.crank(); + assert_eq!(events.last_head(), head_b(true)); + assert_eq!(events.reorgs(), [70], "the head left A's branch for its sibling"); +} + +#[test] +fn invalidating_the_only_branch_publishes_the_resident_anchor() { + let mut rig = HeadRig::new(); + let a = rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + rig.import_child(B_ROOT, 72, A_ROOT, a, B_PREVIOUS, B_CURRENT); + assert_eq!(rig.crank().last_head().root, B_ROOT); + + rig.verdict(A_ROOT, PayloadValidationStatus::Invalid); + + let events = rig.crank(); + assert_eq!(events.last_head(), head_anchor()); + assert_eq!(events.reorgs(), [70]); +} + +fn head_a_empty(optimistic: bool) -> StatusHead { + StatusHead { payload: PayloadResolution::Empty, ..head_a(optimistic) } +} + +/// Mark verification directly to isolate publication from envelope validation. +#[test] +fn payload_verification_and_execution_validation_update_the_head_independently() { + let mut rig = HeadRig::new(); + rig.import_gloas(A_ROOT, 71, A_PREVIOUS, A_CURRENT, false); + assert_eq!(rig.crank().last_head(), head_a_empty(true)); + + rig.tile.fork_choice.mark_payload_verified(&A_ROOT); + let events = rig.crank(); + assert_eq!(events.last_head(), head_a(true)); + assert!(events.reorgs().is_empty(), "the head block did not move"); + + rig.verdict(A_ROOT, PayloadValidationStatus::Valid); + assert_eq!(rig.crank().last_head(), head_a(false)); +} + +/// An invalid Gloas payload can leave its block selected with an empty +/// resolution. +#[test] +fn an_invalid_verdict_on_a_gloas_head_publishes_the_empty_resolution_without_a_reorg() { + let mut rig = HeadRig::new(); + rig.import_gloas(A_ROOT, 71, A_PREVIOUS, A_CURRENT, true); + assert_eq!(rig.crank().last_head(), head_a(true)); + + rig.verdict(A_ROOT, PayloadValidationStatus::Invalid); + let events = rig.crank(); + assert_eq!(events.last_head(), head_a_empty(true)); + assert!(events.reorgs().is_empty()); +} + +/// Inject slot-72 votes early so their weight is already folded while the +/// previous-slot rule still applies. This isolates the rule's expiry and +/// Status publication; normal gossip would defer these votes until slot 73. +#[test] +fn the_tick_closing_the_previous_slot_window_publishes_the_vote_weighted_resolution() { + let mut rig = HeadRig::new(); + rig.import_gloas(A_ROOT, 71, A_PREVIOUS, A_CURRENT, true); + let _ = rig.crank(); + + rig.vote_on_payload(A_ROOT, 0..8, false); + rig.advance_to_slot(72); + assert_eq!(rig.crank().last_head(), head_a(true)); + + rig.advance_to_slot(73); + let events = rig.crank(); + assert_eq!(events.last_head(), head_a_empty(true)); + assert!(events.reorgs().is_empty()); +} + +/// The boosted empty-edge child makes the parent's resolution depend on PTC +/// votes. Here the child is viable, so the majority also changes the head root. +#[test] +fn a_ptc_majority_returns_the_head_to_the_parent_with_its_full_payload() { + let mut rig = HeadRig::new(); + let a = rig.import_gloas(A_ROOT, 71, A_PREVIOUS, A_CURRENT, true); + rig.advance_to_slot(72); + let _ = rig.crank(); + + rig.tile.fork_choice.set_proposer_boost(B_ROOT); + rig.import_empty_child(B_ROOT, 72, A_ROOT, a, B_PREVIOUS, B_CURRENT); + let b = StatusHead { slot: 72, ..head_b(true) }; + assert_eq!(rig.crank().last_head(), b, "boost on the empty child resolves A empty"); + + rig.ptc_majority(A_ROOT); + let events = rig.crank(); + assert_eq!(events.last_head(), head_a(true)); + assert_eq!(events.reorgs(), [71], "the head left B for its parent"); +} + +/// Votes take effect when the next slot tick recomputes the head. +#[test] +fn a_vote_driven_reorg_publishes_the_new_head_and_reports_the_reorg() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + rig.import(B_ROOT, 71, B_PREVIOUS, B_CURRENT); + let _ = rig.crank(); + + rig.vote_for(B_ROOT, 0..8); + + rig.advance_to_slot(72); + let events = rig.crank(); + assert_eq!(events.last_head(), head_b(true)); + assert_eq!(events.reorgs(), [70]); +} + +#[test] +fn a_status_that_already_named_the_new_head_does_not_hide_the_reorg() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + rig.import(B_ROOT, 71, B_PREVIOUS, B_CURRENT); + let _ = rig.crank(); + + rig.vote_for(B_ROOT, 0..8); + rig.tile.recompute_head(); + rig.tile.on_accept(None, &mut rig.adapter.producers); + let events = rig.crank(); + assert_eq!(events.reorgs(), [70], "the reorg is reported anyway"); + assert_eq!(events.last_head(), head_b(true)); +} + +#[test] +fn validating_a_sibling_does_not_validate_the_head() { + let mut rig = HeadRig::new(); + rig.import(A_ROOT, 71, A_PREVIOUS, A_CURRENT); + rig.import(B_ROOT, 71, B_PREVIOUS, B_CURRENT); + let _ = rig.crank(); + + rig.verdict(B_ROOT, PayloadValidationStatus::Valid); + assert!(rig.crank().heads().iter().all(|head| *head == head_a(true))); + + rig.vote_for(B_ROOT, 0..8); + rig.advance_to_slot(72); + assert_eq!(rig.crank().last_head(), head_b(false)); +} + #[test] fn block_unknown_parent_rejected() { let mut tile = make_tile(); @@ -579,6 +1169,7 @@ fn anchor_child(block_root: B256, state_id: StateId) -> BlockImport { BlockImport { slot: 11, block_root, + state_root: state_root_of(block_root), parent_root: ANCHOR_ROOT, execution_block_hash: [0u8; 32], justified: anchor_cp, @@ -622,24 +1213,56 @@ fn a_block_already_in_fork_choice_is_reported_already_known() { assert_eq!(block_stages(&mut sink), [(block_root, BlockStage::AlreadyKnown)]); } +/// The fixtures sign under Fulu from genesis; mainnet's fork schedule would +/// put a Phase0 domain on their signatures. +#[cfg(feature = "ef_tests")] +fn fulu_from_genesis() -> SpecConfig { + SpecConfig { fulu_fork_epoch: 0, ..SpecConfig::mainnet() } +} + +/// `pre`, `blocks_0` and `post` of a mainnet Fulu sanity fixture, decompressed. +#[cfg(feature = "ef_tests")] +fn sanity_fixture(name: &str) -> (Vec, Vec, Vec) { + let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("consensus-spec-tests/tests/mainnet/fulu/sanity/blocks/pyspec_tests") + .join(name); + let read = |file: &str| { + let path = dir.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())) + }; + (read("pre.ssz_snappy"), read("blocks_0.ssz_snappy"), read("post.ssz_snappy")) +} + +/// Read expected roots from the block header and EF post-state, independently +/// of the duty-dependent lookup under test. +#[cfg(feature = "ef_tests")] +fn fixture_head_roots(block_ssz: &[u8], post_ssz: &[u8]) -> HeadRoots { + let epoch = SignedBeaconBlockView::slot(block_ssz) / SLOTS_PER_EPOCH; + let mut post = BeaconState::from_checkpoint(post_ssz, &fulu_from_genesis(), &[]) + .unwrap_or_else(|e| panic!("decompose post: {e}")); + let id = post.roll_fresh(); + let ring = post.block_roots.view(id.block_roots_idx); + HeadRoots { + state_root: *SignedBeaconBlockView::state_root(block_ssz), + previous_duty_dependent_root: ring + .at_slot((epoch.saturating_sub(1) * SLOTS_PER_EPOCH).saturating_sub(1)), + current_duty_dependent_root: ring.at_slot((epoch * SLOTS_PER_EPOCH).saturating_sub(1)), + } +} + #[cfg(feature = "ef_tests")] #[test] fn a_block_is_applied_once_and_already_known_on_repeat() { - let fixture = PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join("consensus-spec-tests/tests/mainnet/fulu/sanity/blocks/pyspec_tests/attestation"); - let read = |name: &str| { - let path = fixture.join(name); - fs::read(&path) - .unwrap_or_else(|e| panic!("{}: {e} (run `just ef-tests-download`)", path.display())) - }; - let (pre, block) = (read("pre.ssz_snappy"), read("blocks_0.ssz_snappy")); - let pre_ssz = snap::Decoder::new().decompress_vec(&pre).expect("snappy pre"); - let block_ssz = snap::Decoder::new().decompress_vec(&block).expect("snappy block"); + let (pre_ssz, block_ssz, _) = sanity_fixture("attestation"); let state = BeaconState::from_checkpoint(&pre_ssz, &SpecConfig::mainnet(), &[]) .unwrap_or_else(|e| panic!("decompose checkpoint: {e}")); let block_slot = SignedBeaconBlockView::slot(&block_ssz); let (mut tile, mut gp, _rp, mut spine, mut adapter) = - tile_with_producers_on(block_slot + 1, state); + tile_with_producers_on(block_slot + 1, state, SpecConfig::mainnet()); let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); sink.consume(|_: BeaconStateEvent, _| {}); @@ -660,6 +1283,127 @@ fn a_block_is_applied_once_and_already_known_on_repeat() { assert_eq!(block_stages(&mut sink), [(block_root, BlockStage::AlreadyKnown)]); } +#[cfg(feature = "ef_tests")] +#[test] +fn an_imported_block_publishes_its_own_head_metadata() { + let (pre_ssz, block_ssz, post_ssz) = sanity_fixture("attestation"); + let state = BeaconState::from_checkpoint(&pre_ssz, &fulu_from_genesis(), &[]) + .unwrap_or_else(|e| panic!("decompose checkpoint: {e}")); + let expected = fixture_head_roots(&block_ssz, &post_ssz); + let slot = SignedBeaconBlockView::slot(&block_ssz); + let block_root = block_root_fulu(&block_ssz); + let (mut tile, _gp, mut rp, mut spine, mut adapter) = + tile_with_producers_on(slot + 1, state, fulu_from_genesis()); + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); + sink.consume(|_: BeaconStateEvent, _| {}); + tile.loop_body(&mut adapter); + let _ = Published::drain(&mut sink); + + let (_, read) = publish_block_bytes(&mut rp, &block_ssz); + sink.produce(live_block_response(read)); + tile.loop_body(&mut adapter); + + assert_eq!(Published::drain(&mut sink).last_head(), StatusHead { + root: block_root, + slot, + optimistic: true, + roots: expected, + payload: PayloadResolution::Full, + }); +} + +#[cfg(feature = "ef_tests")] +#[test] +fn a_replayed_block_publishes_its_own_head_metadata() { + let (pre_ssz, block_ssz, post_ssz) = sanity_fixture("attestation"); + let state = BeaconState::from_checkpoint(&pre_ssz, &fulu_from_genesis(), &[]) + .unwrap_or_else(|e| panic!("decompose checkpoint: {e}")); + let expected = fixture_head_roots(&block_ssz, &post_ssz); + let slot = SignedBeaconBlockView::slot(&block_ssz); + let (mut tile, _gp, _rp, mut replay) = + make_tile_with_producers(slot + 1, state, fulu_from_genesis()); + let (mut spine, mut adapter) = spine_adapter(&tile); + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); + sink.consume(|_: BeaconStateEvent, _| {}); + tile.loop_body(&mut adapter); + let _ = Published::drain(&mut sink); + + let (_, read) = publish_block_bytes(&mut replay, &block_ssz); + sink.produce(ReplayBlock::Block { ssz: read }); + tile.loop_body(&mut adapter); + + assert_eq!(Published::drain(&mut sink).last_head(), StatusHead { + root: block_root_fulu(&block_ssz), + slot, + optimistic: true, + roots: expected, + payload: PayloadResolution::Full, + }); +} + +#[cfg(feature = "ef_tests")] +#[test] +fn the_anchor_reports_its_block_slot_not_the_checkpoint_state_slot() { + let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join( + "consensus-spec-tests/tests/mainnet/fulu/sanity/slots/pyspec_tests/slots_2/post.ssz_snappy", + ); + let raw = fs::read(&path) + .unwrap_or_else(|e| panic!("{}: {e} (run `just ef-tests-download`)", path.display())); + let ssz = snap::Decoder::new().decompress_vec(&raw).expect("snappy post"); + let state = BeaconState::from_checkpoint(&ssz, &SpecConfig::mainnet(), &[]) + .unwrap_or_else(|e| panic!("decompose checkpoint: {e}")); + + let header = state.slot_states.finalized_view().state().latest_block_header; + let state_slot = state.slot_states.finalized_view().slot_number(); + assert!(state_slot > header.slot, "fixture premise: state {state_slot}, block {}", header.slot); + assert_ne!(header.state_root, [0u8; 32], "fixture premise: the header names its state"); + + let (mut tile, _gp, _rp) = make_tile_with_gossip(state_slot, state); + let BeaconStateEvent::Status { ssz, head_roots, .. } = tile.status_event(tile.selected_head()) + else { + panic!("status_event produces Status") + }; + assert_eq!(StatusView::head_slot(&ssz), header.slot, "p2p Status names the anchor block"); + assert_eq!(*StatusView::head_root(&ssz), tile.head_block_root()); + assert_eq!(head_roots.state_root, header.state_root, "the anchor block's declared state"); +} + +#[test] +fn an_anchor_whose_state_outran_the_ring_reports_no_head_metadata() { + let anchor_at = |state_slot: Slot| { + let mut tile = make_tile(); + seed_tile(&mut tile, 4, state_slot); + let cp = Checkpoint { epoch: 0, root: ANCHOR_ROOT }; + tile.fork_choice = ForkChoice::init( + cp, + cp, + 70, + ANCHOR_ROOT, + ANCHOR_STATE_ROOT, + [0u8; 32], + false, + tile.last_applied, + 4, + ); + tile + }; + // The previous decision slot for a head in epoch 2 is slot 31. + let edge = SLOTS_PER_HISTORICAL_ROOT as u64 + 31; + + let tile = anchor_at(edge); + assert!( + tile.head_roots(tile.selected_head()).is_complete(), + "a state at {edge} still holds slot 31, the oldest slot in its ring" + ); + + let tile = anchor_at(edge + 1); + assert_eq!( + tile.head_roots(tile.selected_head()), + HeadRoots::default(), + "one slot later that root is gone, and so is the whole snapshot" + ); +} + /// Fulu requires the execution timestamp and the active blob limit before a /// block is propagated. The STF checks both, but that runs after the relay. #[test] @@ -878,14 +1622,15 @@ struct TestSpine { fn tile_with_producers( wall_slot: u64, ) -> (BeaconStateTile, TProducer, TProducer, TestSpine, SpineAdapter) { - tile_with_producers_on(wall_slot, BeaconState::empty_test(0)) + tile_with_producers_on(wall_slot, BeaconState::empty_test(0), SpecConfig::mainnet()) } fn tile_with_producers_on( wall_slot: u64, state: BeaconState, + spec: SpecConfig, ) -> (BeaconStateTile, TProducer, TProducer, TestSpine, SpineAdapter) { - let (tile, gp, rp) = make_tile_with_gossip(wall_slot, state); + let (tile, gp, rp, _replay) = make_tile_with_producers(wall_slot, state, spec); let (spine, adapter) = spine_adapter(&tile); (tile, gp, rp, spine, adapter) } @@ -1391,8 +2136,17 @@ fn block_known_parent_bad_sig_rejected() { let parent_root = ssz_hash::hash_tree_root_block_header(&genesis_header); let cp = Checkpoint { epoch: 0, root: parent_root }; - tile.fork_choice = - ForkChoice::init(cp, cp, 10, parent_root, [0u8; 32], false, tile.last_applied, 0); + tile.fork_choice = ForkChoice::init( + cp, + cp, + 10, + parent_root, + state_root_of(parent_root), + [0u8; 32], + false, + tile.last_applied, + 0, + ); // Valid structure, zeroed BLS signature → precheck reaches and fails // signature verification, so no fork-choice node is added. @@ -2766,9 +3520,11 @@ impl ThreeForks { let mut sw = g.slot_states.roll_from(parent.slot_idx); sw.state_mut().slot = slot; let slot_idx = sw.commit(); - // The block-roots column, written where `process_slot` writes it. + // Slot zero is a synthetic marker for detecting reads from the + // wrong bundle after finalization. let mut w = g.block_roots.roll_from(parent.block_roots_idx); w.set((slot % SLOTS_PER_HISTORICAL_ROOT as u64) as u32, root); + w.set(0, root); let block_roots_idx = w.commit(); StateId { balances_idx, slot_idx, block_roots_idx, ..parent } } @@ -2784,6 +3540,7 @@ impl ThreeForks { self.tile.fork_choice.on_block(BlockImport { slot, block_root, + state_root: state_root_of(block_root), parent_root, execution_block_hash: [0u8; 32], justified: cp, @@ -2902,6 +3659,44 @@ fn multi_fork_finalize_promotes_and_rebases() { assert_ne!(tile.last_applied, d_id, "stale head bundle replaced"); } +/// Distinct slot-zero markers identify which state bundle supplies the roots +/// after finalization remaps the surviving head. +#[test] +fn status_reads_the_surviving_head_after_finalization_remaps_its_node() { + let mut forks = ThreeForks::new(); + // Justification follows finality here, as `lift_checkpoints` would have it. + forks.tile.fork_choice.justified_checkpoint = Checkpoint { epoch: 0, root: F_ROOT }; + let (_, _gp, _rp, mut spine, mut adapter) = tile_with_producers(2); + let mut sink = SpineAdapter::connect_tile(&Sink, &mut spine.spine); + sink.consume(|_: BeaconStateEvent, _| {}); + forks.tile.ticker.set_since_genesis_ms(2 * 12_000); + forks.tile.loop_body(&mut adapter); + assert_eq!(Published::drain(&mut sink).last_head().root, D_ROOT); + let old_idx = forks.tile.fork_choice.find_node_idx(&D_ROOT).unwrap(); + + forks.tile.maybe_finalize(); + let new_idx = forks.tile.fork_choice.find_node_idx(&D_ROOT).unwrap(); + assert_ne!(old_idx, new_idx, "pruning moved the surviving head's index"); + forks.tile.ticker.set_since_genesis_ms(3 * 12_000); + forks.tile.loop_body(&mut adapter); + + assert_eq!( + Published::drain(&mut sink).last_head(), + StatusHead { + root: D_ROOT, + slot: 2, + optimistic: true, + roots: HeadRoots { + state_root: state_root_of(D_ROOT), + previous_duty_dependent_root: D_ROOT, + current_duty_dependent_root: D_ROOT, + }, + payload: PayloadResolution::Full, + }, + "epoch 0 decides at slot 0, where each bundle carries its own root" + ); +} + /// A block staged on its data columns holds a committed state off a live /// parent: finalization re-anchors it like a node when its parent survives /// and drops it when its parent is pruned. @@ -3167,6 +3962,7 @@ fn finalize_promotes_every_tier_into_checkpoint_encode() { tile.fork_choice.on_block(BlockImport { slot: 1, block_root: F_ROOT, + state_root: state_root_of(F_ROOT), parent_root: ANCHOR_ROOT, execution_block_hash: [0u8; 32], justified: f_cp, diff --git a/crates/columns/src/tile.rs b/crates/columns/src/tile.rs index 42d1e241..49e52be5 100644 --- a/crates/columns/src/tile.rs +++ b/crates/columns/src/tile.rs @@ -788,8 +788,8 @@ mod tests { use silver_beacon_state_data::{BeaconState, BeaconStateOwner}; use silver_common::{ - BlockSource, BlockStage, EngineGetBlobsResp, EngineReq, MessageId, Nanos, P2pStreamId, - StreamProtocol, TCache, TCacheProducer, TCacheRead, + BlockSource, BlockStage, EngineGetBlobsResp, EngineReq, HeadRoots, MessageId, Nanos, + P2pStreamId, PayloadResolution, StreamProtocol, TCache, TCacheProducer, TCacheRead, column_util::SidecarIdentity, ssz_view::{ DATA_COLUMN_SIDECAR_MIN, DataColumnSidecarFuluView, NUMBER_OF_COLUMNS, @@ -1141,12 +1141,17 @@ mod tests { assert!(!rig.tile.validator.is_validated(&root), "the rejection is forgotten with it"); } - /// A sidecar whose parent is unknown is held. A parent the beacon state - /// has staged (state transition done, its own columns pending) counts as - /// seen, so the child's sidecar validates and is persisted, whichever of - /// the two arrives first. + /// A waiting sidecar becomes persistable when its parent is staged or + /// observed as head, including when the head observation repeats. #[test] - fn sidecar_of_staged_parent_is_accepted() { + fn sidecar_of_observed_parent_is_accepted() { + #[derive(Debug)] + enum ParentObservation { + BeforeSidecar, + AfterSidecar, + RepeatedStatus, + } + // A valid sidecar whose parent root names no block, over the state it // was built on. const CASE: &str = "networking/gossip_data_column_sidecar/pyspec_tests/\ @@ -1159,22 +1164,36 @@ mod tests { let parent_root = *DataColumnSidecarFuluView::parent_root(&sidecar); let staged_parent = || block_received(BlockStage::AwaitData, parent_root, slot - 1); - for parent_first in [false, true] { + for observation in [ + ParentObservation::BeforeSidecar, + ParentObservation::AfterSidecar, + ParentObservation::RepeatedStatus, + ] { let mut rig = Rig::with_state(1 << index, reader.clone(), fulu_from_genesis()); rig.tile.sync_state.set_sync_target(SyncUpdate::Following); rig.tile.sync_state.update(status_ssz(0)); - if parent_first { + if matches!(observation, ParentObservation::BeforeSidecar) { rig.tile.handle_beacon_state_event(staged_parent(), &mut rig.conn.producers); } rig.gossip_sidecar(index, &sidecar); - if !parent_first { - rig.tile.handle_beacon_state_event(staged_parent(), &mut rig.conn.producers); + match observation { + ParentObservation::BeforeSidecar => {} + ParentObservation::AfterSidecar => { + rig.tile.handle_beacon_state_event(staged_parent(), &mut rig.conn.producers); + } + ParentObservation::RepeatedStatus => { + rig.conn.consume(|_: BeaconStateEvent, _| {}); + for _ in 0..2 { + rig.inj.producers.produce(head_status(parent_root, slot - 1)); + rig.tile.loop_body(&mut rig.conn); + } + } } rig.turn(); let out = rig.drain(); - if parent_first { + if matches!(observation, ParentObservation::BeforeSidecar) { assert_eq!(out.publications, [( ColumnSource::Gossip, GossipTopic::DataColumnSidecar(index), @@ -1183,10 +1202,22 @@ mod tests { } else { assert!(out.publications.is_empty(), "a buffered copy is not relayed"); } - assert!( - out.persisted(block_root, index), - "parent_first={parent_first}: the sidecar was processed" - ); + assert!(out.persisted(block_root, index), "{observation:?}: the sidecar was processed"); + } + } + + fn head_status(head_root: BlockRoot, head_slot: u64) -> BeaconStateEvent { + let mut ssz = status_ssz(0); + ssz[44..76].copy_from_slice(&head_root); + ssz[76..84].copy_from_slice(&head_slot.to_le_bytes()); + BeaconStateEvent::Status { + ssz, + latest_block_slot: head_slot, + wall_slot: head_slot, + head_optimistic: false, + enr_fork_id: [0u8; 16], + head_roots: HeadRoots::default(), + head_payload: PayloadResolution::Full, } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 659dbd9b..833da51c 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -29,12 +29,13 @@ 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, PendingSubReservation, Prefill, Producer as TProducer, - REJECT_RESPONSE, RPC_PROTOCOLS, RandomAccessConsumer as TRandomAccess, ReplayBlock, + GossipMsgIn, GossipMsgOut, HeadRoots, 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, PayloadResolution, + PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, PeerStats, PeerStatus, + PeerTopicScores, PendingSubReservation, 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, SubLayout, SubReservation, diff --git a/crates/common/src/spine.rs b/crates/common/src/spine.rs index ec215703..9585ca32 100644 --- a/crates/common/src/spine.rs +++ b/crates/common/src/spine.rs @@ -8,12 +8,12 @@ pub use messages::{ EngineGetPayloadBodiesByHashReq, EngineGetPayloadBodiesByRangeReq, EngineGetPayloadBodiesResp, EngineGetPayloadReq, EngineGetPayloadResp, EngineHealthEvent, EngineNewPayloadEnvelopeReq, EngineNewPayloadReq, EngineNewPayloadResp, EnginePreparePayloadReq, EngineReq, EngineResp, - GossipMsgIn, GossipMsgOut, IpBytes, LocalAttestationFailure, LocalAttestationResult, + GossipMsgIn, GossipMsgOut, HeadRoots, 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, RpcRequestOutbound, RpcResponse, RpcResponseInbound, RpcResponseOutbound, - RpcSeverity, SyncNeed, SyncUpdate, SyncingStrategy, WithdrawalInline, + PREFILL_SLOTS, PayloadResolution, PayloadValidationStatus, PeerControl, PeerEvent, PeerScores, + PeerStats, PeerStatus, PeerTopicScores, Prefill, ReplayBlock, RpcInbound, RpcOutbound, + RpcRequest, RpcRequestInbound, RpcRequestOutbound, RpcResponse, RpcResponseInbound, + RpcResponseOutbound, RpcSeverity, SyncNeed, SyncUpdate, SyncingStrategy, WithdrawalInline, }; pub use stream_id::{LOCAL_GOSSIP_STREAM_ID, P2pStreamId}; pub use stream_protocol::{ diff --git a/crates/common/src/spine/messages.rs b/crates/common/src/spine/messages.rs index 10f6c240..a6aed50b 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::{ CacheFrameRef, DataKind, Enr, GossipTopic, Identify, MessageId, Origin, P2pStreamId, PeerId, @@ -853,6 +853,46 @@ pub enum ColumnSource { El, } +/// A zero `state_root` marks all three roots unavailable. This can occur +/// before seeding or when checkpoint history has overwritten a dependent +/// root. Consumers tracking head changes must ignore incomplete bundles. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[repr(C)] +pub struct HeadRoots { + pub state_root: B256, + /// Root at the slot before the head block's previous epoch starts, + /// saturating to slot zero. + pub previous_duty_dependent_root: B256, + /// Root at the slot before the head block's epoch starts, saturating to + /// slot zero. + pub current_duty_dependent_root: B256, +} + +impl HeadRoots { + pub fn is_complete(&self) -> bool { + self.state_root != B256::default() + } +} + +/// Fork choice's selection of the block's own payload, independent of its +/// execution validation status. Selected pre-Gloas blocks resolve `Full`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub enum PayloadResolution { + Empty, + Full, +} + +impl PayloadResolution { + /// The beacon-API `payload_status` spelling. + pub fn name(self) -> &'static str { + match self { + Self::Empty => "empty", + Self::Full => "full", + } + } +} + #[allow(clippy::large_enum_variant)] #[derive(Clone, Copy, Debug)] #[repr(C)] @@ -864,6 +904,8 @@ pub enum BeaconStateEvent { wall_slot: u64, head_optimistic: bool, enr_fork_id: [u8; 16], + head_roots: HeadRoots, + head_payload: PayloadResolution, }, EnvelopeAvailable { ssz: TCacheRead, diff --git a/crates/control/src/sync_engine/mod.rs b/crates/control/src/sync_engine/mod.rs index fe368d9e..600032cd 100644 --- a/crates/control/src/sync_engine/mod.rs +++ b/crates/control/src/sync_engine/mod.rs @@ -80,7 +80,7 @@ impl ReplayGate { } } - fn blocks_requests(&self) -> bool { + fn is_pending(&self) -> bool { !matches!(self, Self::Open) } @@ -446,6 +446,7 @@ impl SyncEngine { fn on_replay_complete(&mut self) { self.replay.open(); self.awaiting_start = true; + self.mark_dirty(); } pub fn on_terminator(&mut self, request_id: u64, peer: usize, delivered: bool, now: Instant) { @@ -548,7 +549,8 @@ impl SyncEngine { return Some(chosen); } let local = &self.ctx.local; - let comparable = local.have_status && + let comparable = !self.replay.is_pending() && + local.have_status && (self.phase.target().is_some() || self.ctx.peers.received_statuses()); let peers_are_ahead = self.ctx.peers.any_peer_ahead_of(local.head_imported_slot, &self.ctx.cfg); @@ -556,7 +558,7 @@ impl SyncEngine { } pub fn drive_requests(&mut self, now: Instant, emit: &mut impl FnMut(SyncAction) -> bool) { - if self.replay.blocks_requests() { + if self.replay.is_pending() { return; } diff --git a/crates/control/src/sync_engine/tests.rs b/crates/control/src/sync_engine/tests.rs index 9b0b521f..8675f02a 100644 --- a/crates/control/src/sync_engine/tests.rs +++ b/crates/control/src/sync_engine/tests.rs @@ -794,6 +794,10 @@ fn peers_finalized_ahead_skip_the_disk_replay_without_waiting() { peer_finalized(&mut e, PEER, (HEAD_ROOT, 400), 10); local_status(&mut e, 0, 400); + assert!( + matches!(e.advance(), Some(SyncUpdate::SyncingFinalized { target_epoch: 10, .. })), + "a syncing target is announced while replay is pending" + ); assert_eq!( e.maybe_choose_syncing_strategy(t0), Some(SyncingStrategy::SyncFromPeers), @@ -1295,6 +1299,7 @@ fn import_above_the_tail_does_not_settle_the_slots_below_it() { // Beacon state applied a block well above the tail, with nothing // reported for the slots in between. local_status(&mut e, BATCH + 22, 200); + local_status(&mut e, BATCH + 22, 200); drive(&mut e, now); assert_eq!(tail(&e), 0, "an apply above the tail settles nothing below it"); @@ -1317,6 +1322,9 @@ fn replay_head_becomes_the_window_floor() { advance(&mut e); assert!(drive(&mut e, now).is_none(), "the replay gate holds requests"); + local_status(&mut e, 60, 200); + assert!(actions(&mut e, now, true).is_empty(), "intermediate Status leaves requests gated"); + e.on_replay_complete(); local_status(&mut e, 100, 200); diff --git a/crates/e2e/src/bin/da_replay.rs b/crates/e2e/src/bin/da_replay.rs index 4bca5bb7..31f6162b 100644 --- a/crates/e2e/src/bin/da_replay.rs +++ b/crates/e2e/src/bin/da_replay.rs @@ -29,9 +29,9 @@ use silver_columns::tile::{ColumnConsumers, DataColumnsTile}; #[cfg(feature = "alloc-profile")] use silver_common::metrics::CountingAllocator; use silver_common::{ - BeaconStateEvent, DataColumnsEvent, DataKind, EngineReq, GossipTopic, MessageId, Nanos, - NewGossipMsg, P2pStreamId, PeerEvent, SilverSpine, StreamProtocol, SyncNeed, SyncUpdate, - TCache, TCacheProducer, TProducer, + BeaconStateEvent, DataColumnsEvent, DataKind, EngineReq, GossipTopic, HeadRoots, MessageId, + Nanos, NewGossipMsg, P2pStreamId, PayloadResolution, PeerEvent, SilverSpine, StreamProtocol, + SyncNeed, SyncUpdate, TCache, TCacheProducer, TProducer, profiler::InProcessReader, ssz_view::{DataColumnSidecarFuluView, NUMBER_OF_COLUMNS, STATUS_V2_SIZE}, test_util::ShmemDir, @@ -165,6 +165,8 @@ impl Node { wall_slot: slot, head_optimistic: false, enr_fork_id: [0u8; 16], + head_roots: HeadRoots::default(), + head_payload: PayloadResolution::Full, }); self.turn(); } diff --git a/crates/storage/src/tile.rs b/crates/storage/src/tile.rs index 9933b7ca..9d44ae94 100644 --- a/crates/storage/src/tile.rs +++ b/crates/storage/src/tile.rs @@ -217,6 +217,25 @@ impl StorageTile { } impl StorageTile { + fn on_status(&mut self, ssz: &[u8; 92], wall_slot: u64) { + self.wall_slot = wall_slot; + let head_slot = StatusView::head_slot(ssz); + let finalized_epoch = StatusView::finalized_epoch(ssz); + self.store.update_head( + head_slot, + *StatusView::head_root(ssz), + finalized_epoch * SLOTS_PER_EPOCH, + *StatusView::finalized_root(ssz), + ); + + if finalized_epoch > self.checkpointed_epoch && + head_slot + CAUGHT_UP_SLACK_SLOTS >= wall_slot + { + self.checkpointed_epoch = finalized_epoch; + self.persist_pending = true; + } + } + #[timed] fn handle_beacon_state_event( &mut self, @@ -389,24 +408,7 @@ impl Tile for StorageTile { }); if let Some((ssz, wall_slot)) = latest_status_event { - self.wall_slot = wall_slot; - let head_slot = StatusView::head_slot(&ssz); - let head_root = *StatusView::head_root(&ssz); - let finalized_epoch = StatusView::finalized_epoch(&ssz); - let finalized_root = *StatusView::finalized_root(&ssz); - self.store.update_head( - head_slot, - head_root, - finalized_epoch * SLOTS_PER_EPOCH, - finalized_root, - ); - - if finalized_epoch > self.checkpointed_epoch && - head_slot + CAUGHT_UP_SLACK_SLOTS >= wall_slot - { - self.checkpointed_epoch = finalized_epoch; - self.persist_pending = true; - } + self.on_status(&ssz, wall_slot); } adapter.consume(|sync_update: SyncUpdate, _| self.store.sync_update(sync_update)); @@ -499,6 +501,61 @@ mod tests { b } + fn status_ssz(head_slot: u64, finalized_epoch: u64) -> [u8; 92] { + let mut ssz = [0u8; 92]; + ssz[36..44].copy_from_slice(&finalized_epoch.to_le_bytes()); + ssz[44..76].copy_from_slice(&[0xAB; 32]); + ssz[76..84].copy_from_slice(&head_slot.to_le_bytes()); + ssz + } + + fn empty_tile(store_dir: &str) -> StorageTile { + let pg = TCache::producer("st_pg", 1 << 16); + let rpc = TCache::producer("st_rpc", 1 << 16); + let pr = TCache::producer("st_pr", 1 << 16); + let el = TCache::producer("st_el", 1 << 16); + StorageTile::new( + pg.cache_ref().random_access("st_pg", true).unwrap(), + rpc.cache_ref().random_access("st_rpc", true).unwrap(), + pr.cache_ref().random_access("st_pr", true).unwrap(), + el.cache_ref().random_access("st_el", true).unwrap(), + TCache::multi_producer("st_rpc_out", 1 << 16), + TCache::producer("st_replay_out", 1 << 16), + BeaconStateOwner::empty_test(0).reader(), + 0, + Arc::new(SpecConfig::mainnet()), + store_dir.to_string(), + true, + ) + } + + /// Private scheduling state substitutes for a checkpoint writer here; + /// consumption is simulated, so this does not verify disk persistence. + #[test] + fn deferred_checkpoint_remains_eligible_and_repeated_status_does_not_reschedule() { + let store_dir = format!("/tmp/test_storage_status_{}", rand::random::()); + let _ = std::fs::remove_dir_all(&store_dir); + let mut tile = empty_tile(&store_dir); + + tile.on_status(&status_ssz(96, 2), 96 + CAUGHT_UP_SLACK_SLOTS + 1); + assert!(!tile.persist_pending, "checkpoint deferred while catching up"); + + tile.on_status(&status_ssz(96, 2), 96); + assert!( + tile.persist_pending, + "caught up at the same finalization, the checkpoint is not lost" + ); + + tile.persist_pending = false; + tile.on_status(&status_ssz(96, 2), 96); + assert!(!tile.persist_pending, "the same finalization was already scheduled"); + + tile.on_status(&status_ssz(128, 3), 128); + assert!(tile.persist_pending, "an advanced finalization arms the next one"); + + let _ = std::fs::remove_dir_all(&store_dir); + } + #[test] fn replay_skips_blocks_missing_custody_columns() { // Checkpoint at slot 32 plus three unfinalized blocks above it. Custody diff --git a/docs/adr/0004-sync-materialized-api.md b/docs/adr/0004-sync-materialized-api.md index 9ee38058..4ddf2f00 100644 --- a/docs/adr/0004-sync-materialized-api.md +++ b/docs/adr/0004-sync-materialized-api.md @@ -79,6 +79,38 @@ 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-08: `/eth/v1/events` also serves the legacy `head` topic. +`BeaconStateEvent::Status` carries the selected block's declared state root +and both duty-dependent roots from its fork's history. + +Status describes an observation. Each consumer decides which fields require +action. Existing publications remain, and an end-of-loop check covers changes +to the selected head or its execution optimism since the last Status. +The publication marker is separate from the reorg marker, so an earlier +Status cannot hide a reorg notification. Replay can emit intermediate +observations; completion still requires `ReplayComplete`. + +The application boundary publishes a head event when a complete observation +changes the head root or optimism. The first complete observation establishes +a baseline. Incomplete metadata, including overwritten checkpoint history, +leaves that baseline unchanged. Node-status updates continue independently. + +`epoch_transition` compares consecutive complete observations and is true +only when the head epoch advances. Same-block validation updates and backward +reorgs report false. New subscriptions receive future changes without an +initial snapshot. + +Amended 2026-09-09: `/eth/v1/events` also serves `head_v2`. Status carries +fork choice's empty/full resolution of the selected block's own payload. +The end-of-loop check publishes an updated Status when this resolution +changes, even if the root and optimism stay the same. + +The boundary publishes `head_v2` when the root, optimism or resolution +changes, including both empty-to-full and full-to-empty transitions. Legacy +`head` retains its root-and-optimism filter. Both topics use the same +complete observation. The v2 `version` names the configured fork at the head +block's slot; selected pre-Gloas blocks report `full`. + 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 @@ -121,3 +153,24 @@ If the cache has overwritten an object's bytes, the boundary logs a warning and emits no event for that request. It also skips sidecars whose length and column offset identify neither supported layout. These failures do not cancel the publication request. + +Amended 2026-09-11: `head` and `head_v2` describe changes observed while +Control reports following. Disk restoration, the wait for a replay strategy +and network catch-up produce no head notifications. Following is a sync +mode, not a guarantee of zero sync distance or execution validation. + +The boundary keeps every complete observation as its baseline in every mode +and reports a change only while following. A following period therefore +starts from the head the node already has, and its first change is reported. +Status and sync updates travel on separate queues; the boundary reads the +mode once per iteration after draining beacon events, so an observation +drained in the same iteration as a mode change follows the earlier mode, and +one queued between the two drains is reported in the next iteration. The +imprecision is bounded by one iteration. Node-status updates and block +notifications remain independent of the mode. + +Control waits for disk replay to finish or be skipped before reporting +following; previously the gate held only network requests, and disk replay +is chosen exactly when peers look comparable, so following could be announced +during replay. Completion re-evaluates the target. Beacon-state's Status +publications are unchanged.