fix(core): announce slot lifecycle to geyser plugins

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::<PluginCommand>();
 
-    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,20 +311,43 @@ 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}"));
         }
     }
     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));
                             }
                         }

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<dyn Fn(SurfnetSvmLocker) -> 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<Slot>,
-        status: GeyserSlotStatus,
+        status: SlotStatus,
     },
     /// Notify plugins of block metadata.
     NotifyBlockMetadata(GeyserBlockMetadata),

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()
+        );
+    }
 }