diff --git a/crates/core/src/runloops/mod.rs b/crates/core/src/runloops/mod.rs index f4366adc..15913a9e 100644 --- a/crates/core/src/runloops/mod.rs +++ b/crates/core/src/runloops/mod.rs @@ -238,15 +238,6 @@ pub async fn start_local_surfnet_runloop( let (plugin_commands_tx, plugin_commands_rx) = unbounded::(); - let (_rpc_handle, _ws_handle, shutdown_rpc_servers) = start_rpc_servers_runloop( - &config, - &simnet_commands_tx, - svm_locker.clone(), - &remote_rpc_client, - plugin_commands_tx, - ) - .await?; - let simnet_config = simnet.clone(); match start_geyser_runloop( @@ -320,10 +311,36 @@ pub async fn start_local_surfnet_runloop( count }); + let _ = svm_locker.with_svm_reader(|svm| { + // Notify geyser plugins that startup is complete + svm.geyser_events_tx.send(GeyserEvent::EndOfStartup)?; + // Announce the genesis slot so plugins begin tracking it. Block production + // announces slot N+1 while closing slot N, so without this the first slot is + // never created from a plugin's perspective and its block data gets dropped. + let slot = svm.get_latest_absolute_slot(); + svm.geyser_events_tx.send(GeyserEvent::UpdateSlotStatus { + slot, + parent: slot.checked_sub(1), + status: SlotStatus::CreatedBank, + }) + }); + + // Do not accept RPC traffic until Geyser plugins have been told about the + // current slot. A transaction submitted as soon as the listener binds can + // otherwise emit block data for a slot the plugin is not tracking yet. + let (_rpc_handle, _ws_handle, shutdown_rpc_servers) = start_rpc_servers_runloop( + &config, + &simnet_commands_tx, + svm_locker.clone(), + &remote_rpc_client, + plugin_commands_tx, + ) + .await?; + // `Runloop` means nobody declares startup work, so the plan is empty. - // Seal it before CoreStarted, and an embedder waiting on that event - // observes a publicly ready surfnet. An external planner instead - // inspects the project after this point and seals its own plan. + // Seal it before CoreStarted, and emit that public readiness signal only + // after the RPC listeners are accepting connections. An external planner + // instead inspects the project after this point and seals its own plan. if startup_planner == StartupPlanner::Runloop { if let Err(error) = svm_locker.seal_startup_plan(vec![]) { simnet_events_tx_cc.error(format!("Failed to seal startup plan: {error}")); @@ -331,9 +348,6 @@ pub async fn start_local_surfnet_runloop( } simnet_events_tx_cc.core_started(initial_transaction_count); - // Notify geyser plugins that startup is complete - let _ = svm_locker.with_svm_reader(|svm| svm.geyser_events_tx.send(GeyserEvent::EndOfStartup)); - start_block_production_runloop( clock_event_rx, clock_command_tx, @@ -829,14 +843,8 @@ fn start_geyser_runloop( } } Ok(GeyserEvent::UpdateSlotStatus { slot, parent, status }) => { - let slot_status = match status { - crate::surfnet::GeyserSlotStatus::Processed => SlotStatus::Processed, - crate::surfnet::GeyserSlotStatus::Confirmed => SlotStatus::Confirmed, - crate::surfnet::GeyserSlotStatus::Rooted => SlotStatus::Rooted, - }; - for plugin in managed_plugins.iter().map(|p| &*p.plugin) { - if let Err(e) = plugin.update_slot_status(slot, parent, &slot_status) { + if let Err(e) = plugin.update_slot_status(slot, parent, &status) { simnet_events_tx.error(format!("Failed to update slot status in Geyser plugin: {:?}", e)); } } diff --git a/crates/core/src/surfnet/mod.rs b/crates/core/src/surfnet/mod.rs index 55fc5320..559b9197 100644 --- a/crates/core/src/surfnet/mod.rs +++ b/crates/core/src/surfnet/mod.rs @@ -1,5 +1,6 @@ use std::{collections::HashMap, fmt::Display, sync::Arc}; +use agave_geyser_plugin_interface::geyser_plugin_interface::SlotStatus; use crossbeam_channel::Sender; use jsonrpc_core::Result as RpcError; use locker::SurfnetSvmLocker; @@ -38,18 +39,6 @@ pub const SLOTS_PER_EPOCH: u64 = 432000; pub type AccountFactory = Box GetAccountResult + Send + Sync>; -/// Slot status for geyser plugin notifications. -/// Mirrors `agave_geyser_plugin_interface::geyser_plugin_interface::SlotStatus`. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum GeyserSlotStatus { - /// Slot is being processed - Processed, - /// Slot has been rooted (finalized) - Rooted, - /// Slot has been confirmed - Confirmed, -} - /// Block metadata for geyser plugin notifications. #[derive(Debug, Clone)] pub struct GeyserBlockMetadata { @@ -88,7 +77,7 @@ pub enum GeyserEvent { UpdateSlotStatus { slot: Slot, parent: Option, - status: GeyserSlotStatus, + status: SlotStatus, }, /// Notify plugins of block metadata. NotifyBlockMetadata(GeyserBlockMetadata), diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index b48d8a68..81097129 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -7,6 +7,7 @@ use std::{ }; use agave_feature_set::FeatureSet; +use agave_geyser_plugin_interface::geyser_plugin_interface::SlotStatus; use base64::{Engine, prelude::BASE64_STANDARD}; use chrono::Utc; use convert_case::Casing; @@ -77,9 +78,9 @@ use uuid::Uuid; use super::{ AccountSubscriptionData, BlockHeader, BlockIdentifier, FINALIZATION_SLOT_THRESHOLD, - GetAccountResult, GeyserBlockMetadata, GeyserEntryInfo, GeyserEvent, GeyserSlotStatus, - ProgramSubscriptionData, SignatureSubscriptionData, SignatureSubscriptionType, - SlotsUpdatesSubscriptionData, remote::SurfnetRemoteClient, + GetAccountResult, GeyserBlockMetadata, GeyserEntryInfo, GeyserEvent, ProgramSubscriptionData, + SignatureSubscriptionData, SignatureSubscriptionType, SlotsUpdatesSubscriptionData, + remote::SurfnetRemoteClient, }; use crate::{ error::{AirdropError, SurfpoolError, SurfpoolResult}, @@ -2461,14 +2462,31 @@ impl SurfnetSvm { timestamp: slots_update_ts, }); + self.geyser_events_tx + .send(GeyserEvent::UpdateSlotStatus { + slot: new_slot, + parent: Some(parent_slot), + status: SlotStatus::CreatedBank, + }) + .ok(); + let geyser_parent_slot = slot.saturating_sub(1); + // Emit `Processed` for the slot that just executed + self.geyser_events_tx + .send(GeyserEvent::UpdateSlotStatus { + slot, + parent: slot.checked_sub(1), + status: SlotStatus::Processed, + }) + .ok(); + // Emit confirmation for the same slot used by processed account/transaction updates. self.geyser_events_tx .send(GeyserEvent::UpdateSlotStatus { slot, parent: slot.checked_sub(1), - status: GeyserSlotStatus::Confirmed, + status: SlotStatus::Confirmed, }) .ok(); // Mirror the Confirmed Geyser event as an `OptimisticConfirmation` @@ -2529,7 +2547,7 @@ impl SurfnetSvm { .send(GeyserEvent::UpdateSlotStatus { slot: root, parent: root.checked_sub(1), - status: GeyserSlotStatus::Rooted, + status: SlotStatus::Rooted, }) .ok(); // Mirror the Rooted Geyser event as a `Root` notification for @@ -6829,4 +6847,69 @@ mod tests { .expect("Valid account should be restored"); assert_eq!(restored_account.lamports, 1_000_000); } + + /// Geyser consumers that reconstruct blocks (yellowstone-grpc, for one) start + /// tracking a slot only once a lifecycle status announces it, and discard block + /// data that arrives for a slot they are not tracking. So every slot must be + /// announced with `CreatedBank` before any of its block data is emitted. + /// + /// The genesis slot is announced by the runloop rather than here, since block + /// production only ever announces the *next* slot — see the `CreatedBank` send + /// after `GeyserEvent::EndOfStartup` in `runloops::start_local_surfnet_runloop`. + /// That first slot is therefore excluded below. + #[test] + fn test_slot_is_announced_before_its_block_data_is_emitted() { + let (mut svm, _events_rx, geyser_rx) = SurfnetSvm::default(); + let genesis_slot = svm.get_latest_absolute_slot(); + + for _ in 0..5 { + svm.confirm_current_block() + .expect("block confirmation should succeed"); + } + + let mut announced = HashSet::new(); + let mut slots_with_block_data = HashSet::new(); + + while let Ok(event) = geyser_rx.try_recv() { + match event { + GeyserEvent::UpdateSlotStatus { + slot, + status: SlotStatus::CreatedBank, + .. + } => { + assert!( + announced.insert(slot), + "slot {slot} was announced more than once" + ); + } + GeyserEvent::NotifyBlockMetadata(metadata) => { + if metadata.slot != genesis_slot { + assert!( + announced.contains(&metadata.slot), + "block metadata for slot {} was emitted before the slot was announced", + metadata.slot + ); + } + slots_with_block_data.insert(metadata.slot); + } + GeyserEvent::NotifyEntry(entry) => { + if entry.slot != genesis_slot { + assert!( + announced.contains(&entry.slot), + "entry for slot {} was emitted before the slot was announced", + entry.slot + ); + } + slots_with_block_data.insert(entry.slot); + } + _ => {} + } + } + + assert!( + slots_with_block_data.len() > 1, + "expected block data for several slots, got {}", + slots_with_block_data.len() + ); + } }