surfpool_core/surfnet/
mod.rs

1use std::{collections::HashMap, fmt::Display, sync::Arc};
2
3use crossbeam_channel::Sender;
4use jsonrpc_core::Result as RpcError;
5use locker::SurfnetSvmLocker;
6use solana_account::Account;
7use solana_account_decoder::{UiAccount, UiAccountEncoding};
8use solana_client::{
9    rpc_config::RpcTransactionLogsFilter,
10    rpc_filter::RpcFilterType,
11    rpc_response::{RpcKeyedAccount, RpcLogsResponse},
12};
13use solana_clock::Slot;
14use solana_commitment_config::CommitmentLevel;
15use solana_epoch_info::EpochInfo;
16use solana_pubkey::Pubkey;
17use solana_rpc_client_api::response::SlotUpdate;
18use solana_signature::Signature;
19use solana_transaction::versioned::VersionedTransaction;
20use solana_transaction_error::TransactionError;
21use solana_transaction_status::{EncodedConfirmedTransactionWithStatusMeta, TransactionStatus};
22use svm::SurfnetSvm;
23
24use crate::{
25    PluginInfo,
26    error::{SurfpoolError, SurfpoolResult},
27    types::{GeyserAccountUpdate, TransactionWithStatusMeta},
28};
29
30pub mod locker;
31pub mod noop_program;
32pub mod remote;
33pub mod surfnet_lite_svm;
34pub mod svm;
35
36pub const FINALIZATION_SLOT_THRESHOLD: u64 = 31;
37pub const SLOTS_PER_EPOCH: u64 = 432000;
38
39pub type AccountFactory = Box<dyn Fn(SurfnetSvmLocker) -> GetAccountResult + Send + Sync>;
40
41/// Slot status for geyser plugin notifications.
42/// Mirrors `agave_geyser_plugin_interface::geyser_plugin_interface::SlotStatus`.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum GeyserSlotStatus {
45    /// Slot is being processed
46    Processed,
47    /// Slot has been rooted (finalized)
48    Rooted,
49    /// Slot has been confirmed
50    Confirmed,
51}
52
53/// Block metadata for geyser plugin notifications.
54#[derive(Debug, Clone)]
55pub struct GeyserBlockMetadata {
56    pub slot: Slot,
57    pub blockhash: String,
58    pub parent_slot: Slot,
59    pub parent_blockhash: String,
60    pub block_time: Option<i64>,
61    pub block_height: Option<u64>,
62    pub executed_transaction_count: u64,
63    pub entry_count: u64,
64}
65
66/// Entry info for geyser plugin notifications.
67/// Surfpool emits one entry per block (simplified model).
68#[derive(Debug, Clone)]
69pub struct GeyserEntryInfo {
70    pub slot: Slot,
71    pub index: usize,
72    pub num_hashes: u64,
73    pub hash: Vec<u8>,
74    pub executed_transaction_count: u64,
75    pub starting_transaction_index: usize,
76}
77
78#[allow(clippy::large_enum_variant)]
79pub enum GeyserEvent {
80    NotifyTransaction(TransactionWithStatusMeta, Option<VersionedTransaction>),
81    UpdateAccount(GeyserAccountUpdate),
82    /// Account update sent at startup (before block production begins).
83    /// These updates should be sent to geyser plugins with is_startup=true.
84    StartupAccountUpdate(GeyserAccountUpdate),
85    /// Notify plugins that startup is complete.
86    EndOfStartup,
87    /// Update slot status (processed, confirmed, rooted/finalized).
88    UpdateSlotStatus {
89        slot: Slot,
90        parent: Option<Slot>,
91        status: GeyserSlotStatus,
92    },
93    /// Notify plugins of block metadata.
94    NotifyBlockMetadata(GeyserBlockMetadata),
95    /// Notify plugins of entry execution.
96    NotifyEntry(GeyserEntryInfo),
97}
98
99/// Commands sent from RPC to the geyser runloop for plugin management.
100pub enum PluginCommand {
101    Load {
102        config_file: String,
103        response_tx: Sender<Result<PluginInfo, String>>,
104    },
105    Unload {
106        name: String,
107        response_tx: Sender<Result<(), String>>,
108    },
109    Reload {
110        name: String,
111        config_file: String,
112        response_tx: Sender<Result<(), String>>,
113    },
114    List {
115        response_tx: Sender<Vec<PluginInfo>>,
116    },
117}
118
119#[derive(Debug, Eq, PartialEq, Hash, Clone)]
120pub struct BlockIdentifier {
121    pub index: u64,
122    pub hash: String,
123}
124
125impl BlockIdentifier {
126    pub fn zero() -> Self {
127        Self::new(
128            0,
129            "0000000000000000000000000000000000000000000000000000000000000000",
130        )
131    }
132
133    pub fn new(index: u64, hash: &str) -> Self {
134        Self {
135            index,
136            hash: hash.to_string(),
137        }
138    }
139}
140
141#[derive(Debug, Clone, Serialize, Deserialize)]
142pub struct BlockHeader {
143    pub hash: String,
144    pub previous_blockhash: String,
145    pub parent_slot: Slot,
146    pub block_time: i64,
147    pub block_height: u64,
148    pub signatures: Vec<Signature>,
149}
150
151#[derive(PartialEq, Eq, Clone)]
152pub enum SurfnetDataConnection {
153    Offline,
154    Connected(String, EpochInfo),
155}
156
157pub type SignatureSubscriptionData = (
158    SignatureSubscriptionType,
159    Sender<(Slot, Option<TransactionError>)>,
160);
161
162pub type AccountSubscriptionData =
163    HashMap<Pubkey, Vec<(Option<UiAccountEncoding>, Sender<UiAccount>)>>;
164
165pub type ProgramSubscriptionData = HashMap<
166    Pubkey,
167    Vec<(
168        Option<UiAccountEncoding>,
169        Option<Vec<RpcFilterType>>,
170        Sender<RpcKeyedAccount>,
171    )>,
172>;
173
174pub type LogsSubscriptionData = (
175    CommitmentLevel,
176    RpcTransactionLogsFilter,
177    Sender<(Slot, RpcLogsResponse)>,
178);
179
180pub type SnapshotSubscriptionData = Sender<SnapshotImportNotification>;
181
182/// Subscription channel for `slotsUpdatesSubscribe` notifications.
183///
184/// Each subscribed client gets one `Sender<Arc<SlotUpdate>>`; the SVM fans
185/// out tagged slot-lifecycle updates (`createdBank`, `frozen`,
186/// `optimisticConfirmation`, `root`) to every active sender. Updates are
187/// wrapped in `Arc` so we only allocate the payload once per emission and
188/// share it across all subscribers.
189pub type SlotsUpdatesSubscriptionData = Sender<Arc<SlotUpdate>>;
190
191#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
192pub struct SnapshotImportNotification {
193    pub snapshot_id: String,
194    pub status: SnapshotImportStatus,
195    pub accounts_loaded: u64,
196    pub total_accounts: u64,
197    pub error: Option<String>,
198}
199
200#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
201pub enum SnapshotImportStatus {
202    Started,
203    InProgress,
204    Completed,
205    Failed,
206}
207
208#[derive(Debug, Clone, PartialEq)]
209pub enum SignatureSubscriptionType {
210    Received,
211    Commitment(CommitmentLevel),
212}
213
214impl Display for SignatureSubscriptionType {
215    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
216        match self {
217            SignatureSubscriptionType::Received => write!(f, "received"),
218            SignatureSubscriptionType::Commitment(level) => write!(f, "{level}"),
219        }
220    }
221}
222
223type DoUpdateSvm = bool;
224
225#[derive(Clone, Debug)]
226/// Represents the result of a get_account operation.
227pub enum GetAccountResult {
228    /// Represents that the account was not found.
229    None(Pubkey),
230    /// Represents that the account was found.
231    /// The `DoUpdateSvm` flag indicates whether the SVM should be updated after this account is found.
232    /// This is useful for cases where the account was fetched from a remote source and needs to be
233    /// updated in the SVM to reflect the latest state. However, when the account is found locally,
234    /// it likely does not need to be updated in the SVM.
235    FoundAccount(Pubkey, Account, DoUpdateSvm),
236    FoundProgramAccount((Pubkey, Account), (Pubkey, Option<Account>)),
237    FoundTokenAccount((Pubkey, Account), (Pubkey, Option<Account>)),
238}
239
240impl GetAccountResult {
241    pub fn expected_data(&self) -> &Vec<u8> {
242        match &self {
243            Self::None(_) => unreachable!(),
244            Self::FoundAccount(_, account, _)
245            | Self::FoundProgramAccount((_, account), _)
246            | Self::FoundTokenAccount((_, account), _) => &account.data,
247        }
248    }
249
250    pub fn apply_update<T>(&mut self, update: T) -> RpcError<()>
251    where
252        T: Fn(&mut Account) -> RpcError<()>,
253    {
254        match self {
255            Self::None(_) => unreachable!(),
256            Self::FoundAccount(_, account, do_update_account) => {
257                update(account)?;
258                *do_update_account = true;
259            }
260            Self::FoundProgramAccount((_, account), _) => {
261                update(account)?;
262            }
263            Self::FoundTokenAccount((_, account), _) => {
264                update(account)?;
265            }
266        }
267        Ok(())
268    }
269
270    pub fn map_account(self) -> SurfpoolResult<Account> {
271        match self {
272            Self::None(pubkey) => Err(SurfpoolError::account_not_found(pubkey)),
273            Self::FoundAccount(_, account, _)
274            | Self::FoundProgramAccount((_, account), _)
275            | Self::FoundTokenAccount((_, account), _) => Ok(account),
276        }
277    }
278
279    #[allow(clippy::type_complexity)]
280    pub fn map_account_with_token_data(
281        self,
282    ) -> Option<((Pubkey, Account), Option<(Pubkey, Option<Account>)>)> {
283        match self {
284            Self::None(_) => None,
285            Self::FoundAccount(pubkey, account, _) => Some(((pubkey, account), None)),
286            Self::FoundProgramAccount((pubkey, account), _) => Some(((pubkey, account), None)),
287            Self::FoundTokenAccount((pubkey, account), token_data) => {
288                Some(((pubkey, account), Some(token_data)))
289            }
290        }
291    }
292
293    pub const fn is_none(&self) -> bool {
294        matches!(self, Self::None(_))
295    }
296
297    pub const fn requires_update(&self) -> bool {
298        match self {
299            Self::None(_) => false,
300            Self::FoundAccount(_, _, do_update) => *do_update,
301            Self::FoundProgramAccount(_, _) => true,
302            Self::FoundTokenAccount(_, _) => true,
303        }
304    }
305}
306
307impl From<GetAccountResult> for Result<Account, SurfpoolError> {
308    fn from(value: GetAccountResult) -> Self {
309        value.map_account()
310    }
311}
312
313impl SignatureSubscriptionType {
314    pub const fn received() -> Self {
315        SignatureSubscriptionType::Received
316    }
317
318    pub const fn processed() -> Self {
319        SignatureSubscriptionType::Commitment(CommitmentLevel::Processed)
320    }
321
322    pub const fn confirmed() -> Self {
323        SignatureSubscriptionType::Commitment(CommitmentLevel::Confirmed)
324    }
325
326    pub const fn finalized() -> Self {
327        SignatureSubscriptionType::Commitment(CommitmentLevel::Finalized)
328    }
329}
330
331#[allow(clippy::large_enum_variant)]
332pub enum GetTransactionResult {
333    None(Signature),
334    FoundTransaction(
335        Signature,
336        EncodedConfirmedTransactionWithStatusMeta,
337        TransactionStatus,
338    ),
339}
340
341impl GetTransactionResult {
342    pub fn found_transaction(
343        signature: Signature,
344        tx: EncodedConfirmedTransactionWithStatusMeta,
345        latest_absolute_slot: u64,
346    ) -> Self {
347        let is_finalized = latest_absolute_slot >= tx.slot + FINALIZATION_SLOT_THRESHOLD;
348        let is_confirmed = latest_absolute_slot >= tx.slot + 1;
349        let (confirmation_status, confirmations) = if is_finalized {
350            (
351                Some(solana_transaction_status::TransactionConfirmationStatus::Finalized),
352                None,
353            )
354        } else if is_confirmed {
355            (
356                Some(solana_transaction_status::TransactionConfirmationStatus::Confirmed),
357                Some((latest_absolute_slot - tx.slot) as usize),
358            )
359        } else {
360            (
361                Some(solana_transaction_status::TransactionConfirmationStatus::Processed),
362                Some((latest_absolute_slot - tx.slot) as usize),
363            )
364        };
365        let status = TransactionStatus {
366            slot: tx.slot,
367            confirmations,
368            status: tx
369                .transaction
370                .clone()
371                .meta
372                .map_or(Ok(()), |m| m.status.map_err(|e| e.into())),
373            err: tx
374                .transaction
375                .clone()
376                .meta
377                .and_then(|m| m.err.map(|e| e.into())),
378            confirmation_status,
379        };
380
381        Self::FoundTransaction(signature, tx, status)
382    }
383
384    pub const fn is_none(&self) -> bool {
385        matches!(self, Self::None(_))
386    }
387
388    pub fn map_found_transaction(&self) -> SurfpoolResult<TransactionStatus> {
389        match self {
390            Self::None(sig) => Err(SurfpoolError::transaction_not_found(sig)),
391            Self::FoundTransaction(_, _, status) => Ok(status.clone()),
392        }
393    }
394
395    pub fn map_some_transaction_status(&self) -> Option<TransactionStatus> {
396        match self {
397            Self::None(_) => None,
398            Self::FoundTransaction(_, _, status) => Some(status.clone()),
399        }
400    }
401}
surfpool_core/surfnet/mod.rs

surfpool_core/surfnet/
mod.rs
