From 1075996e5282ffa3eb0d51e6414542705618cecc Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Thu, 13 Aug 2026 16:23:20 +0300 Subject: [PATCH 1/7] feat(core): report scenario override outcomes --- crates/core/src/error.rs | 37 +- crates/core/src/rpc/surfnet_cheatcodes.rs | 98 ++- crates/core/src/runloops/mod.rs | 29 +- crates/core/src/surfnet/locker.rs | 154 +++- crates/core/src/surfnet/svm.rs | 695 ++++++++++++++++-- crates/core/src/tests/integration.rs | 119 ++- .../kit/generated/OverrideOutcome.ts | 24 + .../surfpool-sdk/kit/generated/index.ts | 1 + .../surfpool-sdk/kit/generated/methods.ts | 2 + crates/sdk-node/surfpool-sdk/kit/types/api.ts | 13 +- crates/types/src/rpc_endpoints.json | 28 +- crates/types/src/scenarios.rs | 41 ++ crates/types/src/types.rs | 20 +- 13 files changed, 1131 insertions(+), 130 deletions(-) create mode 100644 crates/sdk-node/surfpool-sdk/kit/generated/OverrideOutcome.ts diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 6a8b9ae0a..ef6806e4e 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -1,7 +1,7 @@ use std::{fmt::Display, future::Future, pin::Pin}; use crossbeam_channel::TrySendError; -use jsonrpc_core::{Error, Result}; +use jsonrpc_core::{Error, ErrorCode, Result}; use litesvm::error::LiteSVMError; use serde::Serialize; use serde_json::json; @@ -12,6 +12,7 @@ use solana_transaction::TransactionError; use solana_transaction_status::EncodeError; use crate::storage::StorageError; +use surfpool_types::SimnetCommandError; pub type SurfpoolResult = std::result::Result; @@ -84,6 +85,14 @@ impl From for SurfpoolError { } impl SurfpoolError { + pub fn into_simnet_command_error(self) -> SimnetCommandError { + let message = self.to_string(); + match self.0.code { + ErrorCode::InvalidParams => SimnetCommandError::InvalidParams(message), + _ => SimnetCommandError::Internal(message), + } + } + pub fn from_try_send_error(e: TrySendError) -> Self { let mut error = Error::internal_error(); error.data = Some(json!(format!( @@ -416,6 +425,15 @@ impl SurfpoolError { Self(error) } + pub fn internal_message(message: M) -> Self + where + M: Into, + { + let mut error = Error::internal_error(); + error.message = message.into(); + Self(error) + } + pub fn sig_verify_replace_recent_blockhash_collision() -> Self { Self(Error::invalid_params( "sigVerify may not be used with replaceRecentBlockhash", @@ -564,3 +582,20 @@ impl Display for AirdropError { } impl std::error::Error for AirdropError {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn simnet_command_errors_preserve_rpc_classification() { + assert!(matches!( + SurfpoolError::invalid_params("bad slot").into_simnet_command_error(), + SimnetCommandError::InvalidParams(message) if message.contains("bad slot") + )); + assert!(matches!( + SurfpoolError::from(StorageError::LockError).into_simnet_command_error(), + SimnetCommandError::Internal(message) if message.contains("Storage error") + )); + } +} diff --git a/crates/core/src/rpc/surfnet_cheatcodes.rs b/crates/core/src/rpc/surfnet_cheatcodes.rs index 154057c51..311d024f0 100644 --- a/crates/core/src/rpc/surfnet_cheatcodes.rs +++ b/crates/core/src/rpc/surfnet_cheatcodes.rs @@ -1,5 +1,5 @@ use std::{ - collections::BTreeMap, + collections::{BTreeMap, HashSet}, sync::{Arc, RwLock}, }; @@ -18,8 +18,8 @@ use spl_associated_token_account_interface::address::get_associated_token_addres use surfpool_types::{ AccountSnapshot, CheatcodeControlConfig, CheatcodeFilter, ClockCommand, ExportSnapshotConfig, GetStreamedAccountsResponse, GetSurfnetInfoResponse, Idl, OfflineAccountConfig, - ResetAccountConfig, RpcProfileResultConfig, Scenario, SimnetCommand, StreamAccountConfig, - StreamAccountsEntry, UiKeyedProfileResult, + OverrideOutcome, ResetAccountConfig, RpcProfileResultConfig, Scenario, SimnetCommand, + StreamAccountConfig, StreamAccountsEntry, UiKeyedProfileResult, types::{AccountUpdate, SetSomeAccount, SupplyUpdate, TokenAccountUpdate, UuidOrSignature}, }; @@ -781,6 +781,13 @@ pub trait SurfnetCheatcodes { config: Option, ) -> Result; + #[rpc(meta, name = "surfnet_timeTravelWithOverrideOutcomes")] + fn time_travel_with_override_outcomes( + &self, + meta: Self::Metadata, + config: Option, + ) -> Result>>; + /// A cheat code to freeze the Surfnet clock on the local network. /// All time progression halts until resumed. /// @@ -1278,13 +1285,13 @@ pub trait SurfnetCheatcodes { /// - `scenarioRelativeSlot`: The relative slot offset (from base slot) when this override should be applied /// - `label`: Optional label for this override /// - `enabled`: Whether this override is active - /// - `fetchBeforeUse`: If true, fetch fresh account data just before transaction execution (useful for price feeds, oracle updates, and dynamic balances) + /// - `fetchBeforeUse`: If true, require fresh remote account data before applying the override. The override is skipped when no remote client is configured or the fetch fails. /// - `account`: Account address (either `{ "pubkey": "..." }` or `{ "pda": { "programId": "...", "seeds": [...] } }`) /// - `tags`: Array of tags for categorization /// - `slot` (optional): The base slot from which relative slot offsets are calculated. If omitted, uses the current slot. /// /// ## Returns - /// A `RpcResponse<()>` indicating whether the Scenario registration was successful. + /// A `RpcResponse>` for overrides at the base slot. /// /// ## Example Request (with slot) /// ```json @@ -1362,7 +1369,13 @@ pub trait SurfnetCheatcodes { /// "slot": 355684457, /// "apiVersion": "2.2.2" /// }, - /// "value": null, + /// "value": [ + /// { + /// "overrideId": "override-1", + /// "label": "Set BTC price", + /// "applied": true + /// } + /// ], /// "id": 1 /// } /// ``` @@ -1372,7 +1385,14 @@ pub trait SurfnetCheatcodes { meta: Self::Metadata, scenario: Scenario, slot: Option, - ) -> BoxFuture>>; + ) -> BoxFuture>>>; + + #[rpc(meta, name = "surfnet_cancelScenarioOverrides")] + fn cancel_scenario_overrides( + &self, + meta: Self::Metadata, + override_ids: Vec, + ) -> Result>; } #[derive(Clone)] @@ -2089,6 +2109,26 @@ impl SurfnetCheatcodes for SurfnetCheatcodesRpc { Ok(epoch_info) } + fn time_travel_with_override_outcomes( + &self, + meta: Self::Metadata, + config: Option, + ) -> Result>> { + let key = meta.as_ref().map(|ctx| ctx.id.clone()).unwrap_or_default(); + let simnet_command_tx = meta.get_surfnet_command_tx()?; + let svm_locker = meta.get_svm_locker()?; + let (epoch_info, outcomes) = svm_locker.time_travel_with_override_outcomes( + key, + simnet_command_tx, + config.unwrap_or_default(), + )?; + + Ok(RpcResponse { + context: RpcResponseContext::new(epoch_info.absolute_slot), + value: outcomes, + }) + } + fn reset_account( &self, meta: Self::Metadata, @@ -2318,7 +2358,7 @@ impl SurfnetCheatcodes for SurfnetCheatcodesRpc { meta: Self::Metadata, scenario: Scenario, slot: Option, - ) -> BoxFuture>> { + ) -> BoxFuture>>> { let SurfnetRpcContext { svm_locker, remote_ctx, @@ -2328,35 +2368,33 @@ impl SurfnetCheatcodes for SurfnetCheatcodesRpc { }; Box::pin(async move { - // Get the base slot for registration (either provided or current) - let base_slot = slot.unwrap_or_else(|| svm_locker.get_latest_absolute_slot()); - - // Register the scenario with explicit base slot - svm_locker - .register_scenario(scenario, Some(base_slot)) - .map_err(|e| jsonrpc_core::Error { - code: jsonrpc_core::ErrorCode::InternalError, - message: format!("Failed to register scenario: {}", e), - data: None, - })?; - - // Immediately materialize overrides for the BASE slot (not current slot) - // This ensures slot 0's override is applied right away - svm_locker - .materialize_overrides_for_slot(&remote_ctx, base_slot) + let outcomes = svm_locker + .register_scenario_and_materialize(&remote_ctx, scenario, slot) .await - .map_err(|e| jsonrpc_core::Error { - code: jsonrpc_core::ErrorCode::InternalError, - message: format!("Failed to materialize initial overrides: {}", e), - data: None, - })?; + .map_err(jsonrpc_core::Error::from)?; Ok(RpcResponse { context: RpcResponseContext::new(svm_locker.get_latest_absolute_slot()), - value: (), + value: outcomes, }) }) } + + fn cancel_scenario_overrides( + &self, + meta: Self::Metadata, + override_ids: Vec, + ) -> Result> { + let svm_locker = meta.get_svm_locker()?; + let removed = svm_locker + .cancel_scheduled_overrides(override_ids.into_iter().collect::>()) + .map_err(jsonrpc_core::Error::from)?; + + Ok(RpcResponse { + context: RpcResponseContext::new(svm_locker.get_latest_absolute_slot()), + value: removed, + }) + } } #[cfg(test)] diff --git a/crates/core/src/runloops/mod.rs b/crates/core/src/runloops/mod.rs index f4366adcc..b9af65065 100644 --- a/crates/core/src/runloops/mod.rs +++ b/crates/core/src/runloops/mod.rs @@ -477,27 +477,14 @@ pub async fn start_block_production_runloop( }); } SimnetCommand::UpdateInternalClockWithConfirmation(_, clock, response_tx) => { - // Confirm the current block to materialize any scheduled overrides for this slot - if let Err(e) = svm_locker.confirm_current_block(&remote_client_with_commitment).await { - svm_locker.simnet_events_tx().error(format!( - "Failed to confirm block after time travel: {}", e - )); - } - - let epoch_info = svm_locker.with_svm_writer(|svm_writer| { - svm_writer.inner.set_sysvar(&clock); - svm_writer.updated_at = clock.unix_timestamp as u64 * 1_000; - svm_writer.latest_epoch_info.absolute_slot = clock.slot; - svm_writer.latest_epoch_info.epoch = clock.epoch; - svm_writer.latest_epoch_info.slot_index = clock.slot; - svm_writer.latest_epoch_info.epoch = clock.epoch; - svm_writer.latest_epoch_info.absolute_slot = clock.slot + clock.epoch * svm_writer.latest_epoch_info.slots_in_epoch; - svm_writer.simnet_events_tx.system_clock_updated(clock); - svm_writer.latest_epoch_info.clone() - }); - - // Send confirmation back - let _ = response_tx.send(epoch_info); + let result = { + let mut svm_writer = svm_locker.0.write().await; + svm_writer + .time_travel_to_clock(&remote_client_with_commitment, clock) + .await + .map_err(crate::error::SurfpoolError::into_simnet_command_error) + }; + let _ = response_tx.send(result); } SimnetCommand::UpdateBlockProductionMode(update) => { block_production_mode = update; diff --git a/crates/core/src/surfnet/locker.rs b/crates/core/src/surfnet/locker.rs index 1606b40bb..43c0f71fe 100644 --- a/crates/core/src/surfnet/locker.rs +++ b/crates/core/src/surfnet/locker.rs @@ -55,9 +55,9 @@ use solana_transaction_status::{ use surfpool_types::{ AccountSnapshot, ComputeUnitsEstimationResult, ExecutionCapture, ExportSnapshotConfig, Idl, KeyedProfileResult, ProfileResult, RpcProfileResultConfig, RunbookExecutionStatusReport, - SimnetCommand, SimnetEventsTx, StartupError, SurfnetStartupStatus, SurfnetStartupTask, - TransactionConfirmationStatus, TransactionStatusEvent, UiKeyedProfileResult, UuidOrSignature, - VersionedIdl, + SimnetCommand, SimnetCommandError, SimnetEventsTx, StartupError, SurfnetStartupStatus, + SurfnetStartupTask, TransactionConfirmationStatus, TransactionStatusEvent, + UiKeyedProfileResult, UuidOrSignature, VersionedIdl, }; use tokio::sync::RwLock; use txtx_addon_kit::indexmap::IndexSet; @@ -2626,21 +2626,28 @@ impl SurfnetSvmLocker { }) } - /// Registers a scenario for execution - pub fn register_scenario( + pub async fn register_scenario_and_materialize( &self, + remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, scenario: surfpool_types::Scenario, slot: Option, - ) -> SurfpoolResult<()> { - self.with_svm_writer(move |svm_writer| svm_writer.register_scenario(scenario, slot)) + ) -> SurfpoolResult> { + let mut svm_writer = self.0.write().await; + let base_slot = slot.unwrap_or(svm_writer.latest_epoch_info.absolute_slot); + svm_writer + .register_scenario_and_materialize(remote_ctx, scenario, base_slot) + .await + } + + pub fn cancel_scheduled_overrides(&self, override_ids: HashSet) -> SurfpoolResult { + self.with_svm_writer(move |svm_writer| svm_writer.cancel_scheduled_overrides(&override_ids)) } - /// Materializes overrides for a specific slot (not necessarily the current slot) pub async fn materialize_overrides_for_slot( &self, remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, slot: Slot, - ) -> SurfpoolResult<()> { + ) -> SurfpoolResult> { let mut svm_writer = self.0.write().await; svm_writer .materialize_overrides_for_slot(remote_ctx, slot) @@ -3716,6 +3723,16 @@ impl SurfnetSvmLocker { simnet_command_tx: Sender, config: TimeTravelConfig, ) -> SurfpoolResult { + self.time_travel_with_override_outcomes(key, simnet_command_tx, config) + .map(|(epoch_info, _)| epoch_info) + } + + pub fn time_travel_with_override_outcomes( + &self, + key: Option<(blake3::Hash, String)>, + simnet_command_tx: Sender, + config: TimeTravelConfig, + ) -> SurfpoolResult<(EpochInfo, Vec)> { let (epoch_info, slot_time, updated_at) = self.with_svm_reader(|svm_reader| { ( svm_reader.latest_epoch_info.clone(), @@ -3726,7 +3743,7 @@ impl SurfnetSvmLocker { let clock_update: Clock = calculate_time_travel_clock(&config, updated_at, slot_time, &epoch_info) - .map_err(|e| SurfpoolError::internal(e.to_string()))?; + .map_err(|e| SurfpoolError::invalid_params(e.to_string()))?; let formated_time = chrono::DateTime::from_timestamp(clock_update.unix_timestamp, 0) .unwrap_or_else(|| chrono::DateTime::from_timestamp(0, 0).unwrap()) @@ -3737,17 +3754,22 @@ impl SurfnetSvmLocker { let (response_tx, response_rx) = crossbeam_channel::bounded(1); // Send the command with confirmation - let _ = simnet_command_tx.send(SimnetCommand::UpdateInternalClockWithConfirmation( - key, - clock_update, - response_tx, - )); - - // Wait for confirmation with timeout - let updated_epoch_info = response_rx - .recv_timeout(std::time::Duration::from_secs(2)) - .map_err(|e| { - SurfpoolError::internal(format!("Failed to confirm clock update: {}", e)) + simnet_command_tx + .send(SimnetCommand::UpdateInternalClockWithConfirmation( + key, + clock_update, + response_tx, + )) + .map_err(|e| SurfpoolError::internal(format!("Failed to request clock update: {e}")))?; + + let (updated_epoch_info, outcomes) = response_rx + .recv() + .map_err(|e| SurfpoolError::internal(format!("Failed to confirm clock update: {}", e)))? + .map_err(|e| match e { + SimnetCommandError::InvalidParams(message) => { + SurfpoolError::invalid_params(message) + } + SimnetCommandError::Internal(message) => SurfpoolError::internal_message(message), })?; self.simnet_events_tx().info(format!( @@ -3755,7 +3777,7 @@ impl SurfnetSvmLocker { formated_time, updated_epoch_info.epoch, updated_epoch_info.absolute_slot )); - Ok(updated_epoch_info) + Ok((updated_epoch_info, outcomes)) } /// Retrieves the latest absolute slot from the underlying SVM. @@ -3807,7 +3829,10 @@ impl SurfnetSvmLocker { // This prevents lock contention and potential deadlocks from mixing blocking and async locks let mut svm_writer = self.0.write().await; svm_writer.confirm_current_block()?; - svm_writer.materialize_overrides(remote_ctx).await + // Clock-driven materialization has no RPC response for reporting skipped outcomes. + let outcomes = svm_writer.materialize_overrides(remote_ctx).await?; + svm_writer.emit_skipped_override_events(&outcomes); + Ok(()) } /// Subscribes for signature updates (confirmed/finalized) and returns a receiver of events. @@ -4423,8 +4448,13 @@ pub fn format_ui_amount(amount: u64, decimals: u8) -> f64 { #[cfg(test)] mod tests { - use std::collections::HashMap; + use std::{ + collections::HashMap, + thread, + time::{Duration, Instant}, + }; + use crossbeam_channel::unbounded; use solana_account::Account; use solana_account_decoder::UiAccountEncoding; use solana_epoch_schedule::EpochSchedule; @@ -4437,6 +4467,82 @@ mod tests { surfnet::{BlockHeader, SurfnetSvm, svm::apply_override_to_decoded_account}, }; + #[test] + fn time_travel_waits_for_delayed_confirmation() { + let (svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let locker = SurfnetSvmLocker::new(svm); + let current_epoch_info = locker.get_epoch_info(); + let current_slot = current_epoch_info.absolute_slot; + let (command_tx, command_rx) = unbounded::(); + + let responder = thread::spawn(move || { + let SimnetCommand::UpdateInternalClockWithConfirmation(_, _, response_tx) = + command_rx.recv().unwrap() + else { + panic!("unexpected command"); + }; + thread::sleep(Duration::from_millis(2_100)); + response_tx + .send(Ok((current_epoch_info, Vec::new()))) + .unwrap(); + }); + + let started_at = Instant::now(); + let result = locker.time_travel_with_override_outcomes( + None, + command_tx, + TimeTravelConfig::AbsoluteSlot(current_slot), + ); + + assert!(result.is_ok()); + assert!(started_at.elapsed() >= Duration::from_millis(2_100)); + responder.join().unwrap(); + } + + #[test] + fn time_travel_preserves_command_error_classification() { + let cases = [ + ( + SimnetCommandError::InvalidParams("bad target".to_string()), + jsonrpc_core::ErrorCode::InvalidParams, + "bad target", + ), + ( + SimnetCommandError::Internal("storage unavailable".to_string()), + jsonrpc_core::ErrorCode::InternalError, + "storage unavailable", + ), + ]; + + for (command_error, expected_code, expected_message) in cases { + let (svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let locker = SurfnetSvmLocker::new(svm); + let current_slot = locker.get_epoch_info().absolute_slot; + let (command_tx, command_rx) = unbounded::(); + let responder = thread::spawn(move || { + let SimnetCommand::UpdateInternalClockWithConfirmation(_, _, response_tx) = + command_rx.recv().unwrap() + else { + panic!("unexpected command"); + }; + response_tx.send(Err(command_error)).unwrap(); + }); + + let error = locker + .time_travel_with_override_outcomes( + None, + command_tx, + TimeTravelConfig::AbsoluteSlot(current_slot), + ) + .unwrap_err(); + let rpc_error: jsonrpc_core::Error = error.into(); + + assert_eq!(rpc_error.code, expected_code); + assert!(rpc_error.to_string().contains(expected_message)); + responder.join().unwrap(); + } + } + /// A real `PriceUpdateV2` account. Its `VerificationLevel` is the one-byte `Full` variant and /// it ends in a padding byte, which is what fixes the offsets the tests below assert on. fn pyth_price_update_v2_fixture() -> Vec { diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index b48d8a685..8521cd93d 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -2562,7 +2562,7 @@ impl SurfnetSvm { pub async fn materialize_overrides( &mut self, remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, - ) -> SurfpoolResult<()> { + ) -> SurfpoolResult> { let current_slot = self.latest_epoch_info.absolute_slot; self.materialize_overrides_for_slot(remote_ctx, current_slot) @@ -2574,11 +2574,11 @@ impl SurfnetSvm { &mut self, remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, target_slot: Slot, - ) -> SurfpoolResult<()> { + ) -> SurfpoolResult> { // Remove and get overrides for this slot let Some(overrides) = self.scheduled_overrides.take(&target_slot)? else { // No overrides for this slot - return Ok(()); + return Ok(Vec::new()); }; debug!( @@ -2587,9 +2587,29 @@ impl SurfnetSvm { target_slot ); + self.materialize_override_batch(remote_ctx, overrides).await + } + + pub async fn materialize_override_batch( + &mut self, + remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, + overrides: Vec, + ) -> SurfpoolResult> { + use surfpool_types::OverrideOutcome; + + let mut outcomes = Vec::new(); + for override_instance in overrides { + let outcome_id = override_instance.id.clone(); + let outcome_label = override_instance.label.clone(); + if !override_instance.enabled { debug!("Skipping disabled override: {}", override_instance.id); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + "override is disabled", + )); continue; } @@ -2615,6 +2635,11 @@ impl SurfnetSvm { "Failed to resolve account address for override {}", override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + "could not resolve the account address", + )); continue; } }; @@ -2624,8 +2649,11 @@ impl SurfnetSvm { override_instance.id, account_pubkey, override_instance.label ); - // Fetch fresh account data from remote if requested + // Never apply values over stale data when fresh data was explicitly requested. if override_instance.fetch_before_use { + let mut fetched_fresh = false; + let mut fetch_error: Option = None; + if let Some((client, _)) = remote_ctx { debug!( "Fetching fresh account data for {} from remote", @@ -2643,13 +2671,16 @@ impl SurfnetSvm { remote_account.lamports(), remote_account.data().len() ); - - // Set the fresh account data in the SVM - if let Err(e) = self.inner.set_account(account_pubkey, remote_account) { - warn!( - "Failed to set account {} from remote: {}", - account_pubkey, e - ); + match self.inner.set_account(account_pubkey, remote_account) { + Ok(()) => fetched_fresh = true, + Err(e) => { + warn!( + "Failed to set account {} from remote: {}", + account_pubkey, e + ); + fetch_error = + Some(format!("failed to store fetched account data: {e}")); + } } } Ok(GetAccountResult::None(_)) => { @@ -2663,6 +2694,7 @@ impl SurfnetSvm { "Failed to fetch account {} from remote: {}", account_pubkey, e ); + fetch_error = Some(format!("fetchBeforeUse fetch failed: {e}")); } } } else { @@ -2671,6 +2703,14 @@ impl SurfnetSvm { override_instance.id ); } + + if !fetched_fresh { + let reason = fetch_error.unwrap_or_else(|| { + "fetchBeforeUse requested but fresh data was unavailable".to_string() + }); + outcomes.push(OverrideOutcome::skipped(outcome_id, outcome_label, reason)); + continue; + } } // Apply the override values to the account data @@ -2689,6 +2729,7 @@ impl SurfnetSvm { "Override {} has no account data modifications (all values are PDA seeds)", override_instance.id ); + outcomes.push(OverrideOutcome::applied(outcome_id, outcome_label)); continue; } @@ -2706,6 +2747,11 @@ impl SurfnetSvm { "Account {} not found in SVM for override {}, skipping modifications", account_pubkey, override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + "account not found in the local SVM (enable fetchBeforeUse to load it)", + )); continue; }; @@ -2720,6 +2766,11 @@ impl SurfnetSvm { "No IDL registered for program {} (owner of account {}), skipping override {}", owner_program_id, account_pubkey, override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + format!("no IDL registered for the owner program {owner_program_id}"), + )); continue; } Err(e) => { @@ -2727,6 +2778,11 @@ impl SurfnetSvm { "Failed to get IDL for program {}: {}, skipping override {}", owner_program_id, e, override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + format!("failed to load the IDL for program {owner_program_id}: {e}"), + )); continue; } }; @@ -2737,6 +2793,11 @@ impl SurfnetSvm { "IDL versions empty for program {}, skipping override {}", owner_program_id, override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + format!("no IDL version available for program {owner_program_id}"), + )); continue; }; @@ -2754,6 +2815,11 @@ impl SurfnetSvm { account_data.len(), override_instance.id ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + "account data too small to override (enable fetchBeforeUse to load it first)", + )); continue; } @@ -2771,6 +2837,11 @@ impl SurfnetSvm { If the account doesn't exist locally, enable fetchBeforeUse: true.", account_pubkey, override_instance.id, e ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + format!("failed to apply the overrides to the account data: {e}"), + )); continue; } }; @@ -2790,6 +2861,11 @@ impl SurfnetSvm { "Failed to set modified account {} in SVM: {}", account_pubkey, e ); + outcomes.push(OverrideOutcome::skipped( + outcome_id, + outcome_label, + format!("failed to store the modified account: {e}"), + )); } else { debug!( "Successfully applied {} override(s) to account {} (override {})", @@ -2797,11 +2873,15 @@ impl SurfnetSvm { account_pubkey, override_instance.id ); + outcomes.push(OverrideOutcome::applied(outcome_id, outcome_label)); } + } else { + // Reaching this branch completes an address-only or fetch-only override. + outcomes.push(OverrideOutcome::applied(outcome_id, outcome_label)); } } - Ok(()) + Ok(outcomes) } /// Forges account data by applying overrides to existing account data @@ -4026,17 +4106,78 @@ impl SurfnetSvm { Ok(fixtures) } - /// Registers a scenario for execution by scheduling its overrides - /// - /// The `slot` parameter is the base slot from which relative override slot heights are calculated. - /// If not provided, uses the current slot. - pub fn register_scenario( + fn validate_scenario_registration( + &self, + scenario: &surfpool_types::Scenario, + base_slot: Slot, + ) -> SurfpoolResult> { + let current_slot = self.latest_epoch_info.absolute_slot; + if base_slot < current_slot { + return Err(SurfpoolError::invalid_params(format!( + "Scenario base slot {base_slot} is behind current slot {current_slot}" + ))); + } + + let mut override_ids = HashSet::new(); + for override_instance in &scenario.overrides { + if override_instance.id.trim().is_empty() { + return Err(SurfpoolError::invalid_params( + "Scenario override ids cannot be empty", + )); + } + if !override_ids.insert(override_instance.id.clone()) { + return Err(SurfpoolError::invalid_params(format!( + "Duplicate scenario override id: {}", + override_instance.id + ))); + } + base_slot + .checked_add(override_instance.scenario_relative_slot) + .ok_or_else(|| { + SurfpoolError::invalid_params(format!( + "Scenario override {} exceeds the maximum slot", + override_instance.id + )) + })?; + } + + Ok(override_ids) + } + + fn schedule_overrides( &mut self, - scenario: surfpool_types::Scenario, - slot: Option, + base_slot: Slot, + overrides: Vec, ) -> SurfpoolResult<()> { - // Use provided slot or current slot as the base for relative slot heights - let base_slot = slot.unwrap_or(self.latest_epoch_info.absolute_slot); + for override_instance in overrides { + let absolute_slot = base_slot + .checked_add(override_instance.scenario_relative_slot) + .ok_or_else(|| { + SurfpoolError::invalid_params(format!( + "Scenario override {} exceeds the maximum slot", + override_instance.id + )) + })?; + let mut slot_overrides = self + .scheduled_overrides + .get(&absolute_slot)? + .unwrap_or_default(); + slot_overrides.push(override_instance); + self.scheduled_overrides + .store(absolute_slot, slot_overrides)?; + } + + Ok(()) + } + + pub async fn register_scenario_and_materialize( + &mut self, + remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, + scenario: surfpool_types::Scenario, + base_slot: Slot, + ) -> SurfpoolResult> { + let override_ids = self.validate_scenario_registration(&scenario, base_slot)?; + let current_slot = self.latest_epoch_info.absolute_slot; info!( "Registering scenario: {} ({}) with {} overrides at base slot {}", @@ -4046,28 +4187,141 @@ impl SurfnetSvm { base_slot ); - // Schedule overrides by adding base slot to their scenario-relative slots - for override_instance in scenario.overrides { - let scenario_relative_slot = override_instance.scenario_relative_slot; - let absolute_slot = base_slot + scenario_relative_slot; + let (initial_overrides, future_overrides): (Vec<_>, Vec<_>) = scenario + .overrides + .into_iter() + .partition(|override_instance| { + base_slot == current_slot && override_instance.scenario_relative_slot == 0 + }); - debug!( - "Scheduling override at absolute slot {} (base {} + relative {})", - absolute_slot, base_slot, scenario_relative_slot - ); + self.cancel_scheduled_overrides(&override_ids)?; + self.schedule_overrides(base_slot, future_overrides)?; + self.materialize_override_batch(remote_ctx, initial_overrides) + .await + } - let mut slot_overrides = self - .scheduled_overrides - .get(&absolute_slot) - .ok() - .flatten() - .unwrap_or_default(); - slot_overrides.push(override_instance); - self.scheduled_overrides - .store(absolute_slot, slot_overrides)?; + pub fn next_pending_override_slot_between( + &self, + current_slot: Slot, + target_slot: Slot, + ) -> SurfpoolResult> { + Ok(self + .scheduled_overrides + .keys()? + .into_iter() + .filter(|slot| current_slot < *slot && *slot < target_slot) + .min()) + } + + pub fn cancel_scheduled_overrides( + &mut self, + override_ids: &HashSet, + ) -> SurfpoolResult { + if override_ids.is_empty() { + return Ok(0); } - Ok(()) + let mut updates = Vec::new(); + let mut removed = 0u64; + + for slot in self.scheduled_overrides.keys()? { + let Some(slot_overrides) = self.scheduled_overrides.get(&slot)? else { + continue; + }; + let original_len = slot_overrides.len(); + let remaining: Vec<_> = slot_overrides + .into_iter() + .filter(|override_instance| !override_ids.contains(&override_instance.id)) + .collect(); + + if remaining.len() != original_len { + removed += (original_len - remaining.len()) as u64; + updates.push((slot, remaining)); + } + } + + for (slot, remaining) in updates { + if remaining.is_empty() { + self.scheduled_overrides.take(&slot)?; + } else { + self.scheduled_overrides.store(slot, remaining)?; + } + } + + Ok(removed) + } + + pub async fn time_travel_to_clock( + &mut self, + remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, + clock: Clock, + ) -> SurfpoolResult<(EpochInfo, Vec)> { + let slots_in_epoch = self.latest_epoch_info.slots_in_epoch; + let target_slot = clock + .epoch + .checked_mul(slots_in_epoch) + .and_then(|epoch_start| epoch_start.checked_add(clock.slot)) + .ok_or_else(|| SurfpoolError::invalid_params("Time travel target slot overflowed"))?; + let current_slot = self.latest_epoch_info.absolute_slot; + + if target_slot < current_slot { + return Err(SurfpoolError::invalid_params(format!( + "Time travel target slot {target_slot} is behind current slot {current_slot}" + ))); + } + + if let Some(next_pending_slot) = + self.next_pending_override_slot_between(current_slot, target_slot)? + { + return Err(SurfpoolError::invalid_params(format!( + "Time travel from slot {current_slot} to {target_slot} would skip pending scenario overrides at slot {next_pending_slot}" + ))); + } + + let materialization_result = if target_slot == current_slot { + Ok(Vec::new()) + } else { + self.confirm_current_block()?; + self.materialize_overrides(remote_ctx).await + }; + + self.inner.set_sysvar(&clock); + self.updated_at = clock.unix_timestamp as u64 * 1_000; + self.latest_epoch_info.slot_index = clock.slot; + self.latest_epoch_info.epoch = clock.epoch; + self.latest_epoch_info.absolute_slot = target_slot; + let _ = self + .simnet_events_tx + .send(SimnetEvent::SystemClockUpdated(clock)); + + let mut outcomes = materialization_result?; + if target_slot == current_slot || target_slot > current_slot.saturating_add(1) { + outcomes.extend( + self.materialize_overrides_for_slot(remote_ctx, target_slot) + .await?, + ); + } + + self.emit_skipped_override_events(&outcomes); + Ok((self.latest_epoch_info.clone(), outcomes)) + } + + pub(crate) fn emit_skipped_override_events( + &self, + outcomes: &[surfpool_types::OverrideOutcome], + ) { + for outcome in outcomes { + if !outcome.applied { + let _ = self.simnet_events_tx.send(SimnetEvent::warn(format!( + "Override {} was skipped: {}", + outcome + .label + .as_deref() + .unwrap_or(outcome.override_id.as_str()), + outcome.reason.as_deref().unwrap_or("unknown reason"), + ))); + } + } } } @@ -4094,11 +4348,370 @@ mod tests { use solana_transaction::Transaction; use solana_transaction_error::TransactionError; use spl_token_interface::state::{Account as TokenAccount, AccountState}; - use surfpool_types::ExportSnapshotFilter; + use surfpool_types::{AccountAddress, ExportSnapshotFilter}; use test_case::test_case; use super::*; - use crate::storage::tests::TestType; + use crate::storage::{Storage, StorageError, StorageResult, tests::TestType}; + + #[derive(Clone)] + struct FailingTakeStorage; + + impl Storage> for FailingTakeStorage { + fn store(&mut self, _key: u64, _value: Vec) -> StorageResult<()> { + Ok(()) + } + + fn clear(&mut self) -> StorageResult<()> { + Ok(()) + } + + fn get(&self, _key: &u64) -> StorageResult>> { + Ok(None) + } + + fn take(&mut self, _key: &u64) -> StorageResult>> { + Err(StorageError::LockError) + } + + fn keys(&self) -> StorageResult> { + Ok(Vec::new()) + } + + fn into_iter( + &self, + ) -> StorageResult)> + '_>> { + Ok(Box::new(std::iter::empty())) + } + + fn count(&self) -> StorageResult { + Ok(0) + } + + fn clone_box(&self) -> Box>> { + Box::new(self.clone()) + } + } + + fn test_override(id: &str, relative_slot: Slot) -> OverrideInstance { + OverrideInstance { + id: id.to_string(), + template_id: "test-template".to_string(), + values: HashMap::new(), + scenario_relative_slot: relative_slot, + label: None, + enabled: true, + fetch_before_use: false, + account: AccountAddress::Pubkey(Pubkey::new_unique().to_string()), + } + } + + fn clock_for_absolute_slot(svm: &SurfnetSvm, target_slot: Slot) -> Clock { + let slots_in_epoch = svm.latest_epoch_info.slots_in_epoch; + let epoch = target_slot / slots_in_epoch; + let slot = target_slot % slots_in_epoch; + let elapsed_slots = target_slot.saturating_sub(svm.latest_epoch_info.absolute_slot); + let unix_timestamp = svm + .updated_at + .saturating_add(elapsed_slots.saturating_mul(svm.slot_time)) + / 1_000; + + Clock { + slot, + epoch_start_timestamp: unix_timestamp as i64, + epoch, + leader_schedule_epoch: 0, + unix_timestamp: unix_timestamp as i64, + } + } + + #[tokio::test] + async fn materialize_override_batch_returns_exactly_one_outcome_per_override() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let mut disabled = test_override("disabled", 0); + disabled.enabled = false; + let mut absent = test_override("absent", 0); + absent + .values + .insert("account.value".to_string(), serde_json::json!(1)); + let applied = test_override("applied", 0); + + let outcomes = svm + .materialize_override_batch(&None, vec![disabled, absent, applied]) + .await + .unwrap(); + + assert_eq!(outcomes.len(), 3); + assert_eq!(outcomes[0].override_id, "disabled"); + assert!(!outcomes[0].applied); + assert!(outcomes[0].reason.is_some()); + assert_eq!(outcomes[1].override_id, "absent"); + assert!(!outcomes[1].applied); + assert!(outcomes[1].reason.is_some()); + assert_eq!(outcomes[2].override_id, "applied"); + assert!(outcomes[2].applied); + assert!(outcomes[2].reason.is_none()); + } + + #[tokio::test] + async fn fetch_before_use_without_remote_context_skips_the_override() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let mut override_instance = test_override("fresh", 0); + override_instance.fetch_before_use = true; + + let outcomes = svm + .materialize_override_batch(&None, vec![override_instance]) + .await + .unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!(!outcomes[0].applied); + assert_eq!( + outcomes[0].reason.as_deref(), + Some("fetchBeforeUse requested but fresh data was unavailable") + ); + } + + #[test_case(TestType::sqlite(); "with on-disk sqlite db")] + #[test_case(TestType::in_memory(); "with in-memory sqlite db")] + #[test_case(TestType::no_db(); "with no db")] + #[cfg_attr(feature = "postgres", test_case(TestType::postgres(); "with postgres db"))] + fn pending_override_helpers_preserve_boundaries_and_unrelated_entries(test_type: TestType) { + let (mut svm, _events_rx, _geyser_rx) = test_type.initialize_svm(); + svm.scheduled_overrides + .store(450, vec![test_override("cancel", 450)]) + .unwrap(); + svm.scheduled_overrides + .store( + 750, + vec![test_override("cancel", 750), test_override("keep", 750)], + ) + .unwrap(); + + assert_eq!( + svm.next_pending_override_slot_between(0, 750).unwrap(), + Some(450) + ); + assert_eq!( + svm.next_pending_override_slot_between(450, 750).unwrap(), + None + ); + assert_eq!( + svm.next_pending_override_slot_between(750, 750).unwrap(), + None + ); + + let cancel_ids = HashSet::from(["cancel".to_string()]); + assert_eq!(svm.cancel_scheduled_overrides(&cancel_ids).unwrap(), 2); + assert_eq!(svm.cancel_scheduled_overrides(&cancel_ids).unwrap(), 0); + assert!(svm.scheduled_overrides.get(&450).unwrap().is_none()); + assert_eq!( + svm.scheduled_overrides + .get(&750) + .unwrap() + .unwrap() + .iter() + .map(|override_instance| override_instance.id.as_str()) + .collect::>(), + vec!["keep"] + ); + } + + #[tokio::test] + async fn register_scenario_materializes_only_initial_overrides_and_replaces_pending_ids() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let base_slot = svm.latest_epoch_info.absolute_slot; + let scenario = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("initial", 0), test_override("future", 450)], + tags: Vec::new(), + }; + + let outcomes = svm + .register_scenario_and_materialize(&None, scenario, base_slot) + .await + .unwrap(); + assert_eq!(outcomes.len(), 1); + assert_eq!(outcomes[0].override_id, "initial"); + assert!(outcomes[0].applied); + assert!(svm.scheduled_overrides.get(&base_slot).unwrap().is_none()); + assert_eq!( + svm.scheduled_overrides + .get(&(base_slot + 450)) + .unwrap() + .unwrap()[0] + .id, + "future" + ); + + let replacement = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("future", 750)], + tags: Vec::new(), + }; + svm.register_scenario_and_materialize(&None, replacement, base_slot) + .await + .unwrap(); + + assert!( + svm.scheduled_overrides + .get(&(base_slot + 450)) + .unwrap() + .is_none() + ); + assert_eq!( + svm.scheduled_overrides + .get(&(base_slot + 750)) + .unwrap() + .unwrap()[0] + .id, + "future" + ); + } + + #[tokio::test] + async fn register_scenario_rejects_invalid_input_before_mutation() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + svm.latest_epoch_info.absolute_slot = 10; + let current_slot = svm.latest_epoch_info.absolute_slot; + let duplicate_scenario = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("duplicate", 1), test_override("duplicate", 2)], + tags: Vec::new(), + }; + + assert!( + svm.register_scenario_and_materialize(&None, duplicate_scenario, current_slot) + .await + .is_err() + ); + assert_eq!(svm.scheduled_overrides.count().unwrap(), 0); + + let past_scenario = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("past", 0)], + tags: Vec::new(), + }; + assert!( + svm.register_scenario_and_materialize(&None, past_scenario, current_slot - 1) + .await + .is_err() + ); + + let overflow_scenario = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("overflow", u64::MAX)], + tags: Vec::new(), + }; + assert!( + svm.register_scenario_and_materialize(&None, overflow_scenario, current_slot) + .await + .is_err() + ); + assert_eq!(svm.scheduled_overrides.count().unwrap(), 0); + } + + #[tokio::test] + async fn time_travel_same_slot_materializes_once_without_advancing() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let current_slot = svm.latest_epoch_info.absolute_slot; + svm.scheduled_overrides + .store(current_slot, vec![test_override("current", 0)]) + .unwrap(); + let clock = clock_for_absolute_slot(&svm, current_slot); + + let (epoch_info, outcomes) = svm + .time_travel_to_clock(&None, clock.clone()) + .await + .unwrap(); + assert_eq!(epoch_info.absolute_slot, current_slot); + assert_eq!(outcomes.len(), 1); + assert_eq!(outcomes[0].override_id, "current"); + + let (_, retry_outcomes) = svm.time_travel_to_clock(&None, clock).await.unwrap(); + assert!(retry_outcomes.is_empty()); + } + + #[tokio::test] + async fn time_travel_materializes_next_and_sparse_target_slots_once() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let base_slot = svm.latest_epoch_info.absolute_slot; + svm.scheduled_overrides + .store(base_slot + 1, vec![test_override("next", 1)]) + .unwrap(); + + let next_clock = clock_for_absolute_slot(&svm, base_slot + 1); + let (_, next_outcomes) = svm.time_travel_to_clock(&None, next_clock).await.unwrap(); + assert_eq!(next_outcomes.len(), 1); + assert_eq!(next_outcomes[0].override_id, "next"); + + svm.scheduled_overrides + .store(base_slot + 450, vec![test_override("sparse", 450)]) + .unwrap(); + let sparse_clock = clock_for_absolute_slot(&svm, base_slot + 450); + let (epoch_info, sparse_outcomes) = + svm.time_travel_to_clock(&None, sparse_clock).await.unwrap(); + assert_eq!(epoch_info.absolute_slot, base_slot + 450); + assert_eq!(sparse_outcomes.len(), 1); + assert_eq!(sparse_outcomes[0].override_id, "sparse"); + + let retry_clock = clock_for_absolute_slot(&svm, base_slot + 450); + let (_, retry_outcomes) = svm.time_travel_to_clock(&None, retry_clock).await.unwrap(); + assert!(retry_outcomes.is_empty()); + } + + #[tokio::test] + async fn time_travel_rejects_skipping_pending_slot_before_mutation() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let base_slot = svm.latest_epoch_info.absolute_slot; + svm.scheduled_overrides + .store(base_slot + 450, vec![test_override("pending", 450)]) + .unwrap(); + let target_clock = clock_for_absolute_slot(&svm, base_slot + 750); + + let error = svm + .time_travel_to_clock(&None, target_clock) + .await + .unwrap_err(); + + assert!(error.to_string().contains(&(base_slot + 450).to_string())); + assert_eq!(svm.latest_epoch_info.absolute_slot, base_slot); + assert!( + svm.scheduled_overrides + .get(&(base_slot + 450)) + .unwrap() + .is_some() + ); + } + + #[tokio::test] + async fn time_travel_keeps_clock_state_synchronized_when_materialization_fails() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let target_slot = svm.latest_epoch_info.absolute_slot + 1; + let target_clock = clock_for_absolute_slot(&svm, target_slot); + svm.scheduled_overrides = Box::new(FailingTakeStorage); + + let error = svm + .time_travel_to_clock(&None, target_clock.clone()) + .await + .unwrap_err(); + + assert!(error.to_string().contains("Storage error")); + assert_eq!(svm.latest_epoch_info.absolute_slot, target_slot); + let stored_clock = svm.inner.get_sysvar::(); + assert_eq!(stored_clock.slot, target_clock.slot); + assert_eq!(stored_clock.epoch, target_clock.epoch); + } #[test] fn startup_status_subscription_tracks_accepted_transitions() { diff --git a/crates/core/src/tests/integration.rs b/crates/core/src/tests/integration.rs index 03247694c..75e695bf6 100644 --- a/crates/core/src/tests/integration.rs +++ b/crates/core/src/tests/integration.rs @@ -1,4 +1,4 @@ -use std::{str::FromStr, sync::Arc, time::Duration}; +use std::{collections::HashMap, str::FromStr, sync::Arc, time::Duration}; use base64::Engine; use crossbeam_channel::{ @@ -56,11 +56,11 @@ use spl_token_2022_interface::{ pod::PodMint, }; use surfpool_types::{ - AccountSnapshot, CheatcodeConfig, CheatcodeControlConfig, CheatcodeFilter, - DEFAULT_SLOT_TIME_MS, Idl, RpcProfileDepth, RpcProfileResultConfig, SimnetCommand, SimnetEvent, - StartupPlanner, SurfnetStartupPhase, SurfnetStartupStatus, SurfnetStartupTask, - SurfnetStartupTaskState, SurfpoolConfig, UiAccountChange, UiAccountProfileState, - UiKeyedProfileResult, + AccountAddress, AccountSnapshot, CheatcodeConfig, CheatcodeControlConfig, CheatcodeFilter, + DEFAULT_SLOT_TIME_MS, Idl, OverrideInstance, OverrideOutcome, RpcProfileDepth, + RpcProfileResultConfig, Scenario, SimnetCommand, SimnetEvent, StartupPlanner, + SurfnetStartupPhase, SurfnetStartupStatus, SurfnetStartupTask, SurfnetStartupTaskState, + SurfpoolConfig, UiAccountChange, UiAccountProfileState, UiKeyedProfileResult, types::{ BlockProductionMode, RpcConfig, SimnetConfig, SubgraphConfig, TransactionStatusEvent, UuidOrSignature, @@ -4886,6 +4886,113 @@ fn test_time_travel_absolute_slot(test_type: TestType) { println!("Time travel to absolute slot test passed successfully!"); } +#[test] +fn test_time_travel_reports_target_outcomes_and_rejects_skipped_pending_slot() { + let rpc_server = SurfnetCheatcodesRpc::empty(); + let simnet = boot_simnet(BlockProductionMode::Manual, Some(400), TestType::no_db()) + .expect("the simnet should boot"); + let (plugin_commands_tx, _plugin_commands_rx) = crossbeam_channel::unbounded::(); + let runloop_context = RunloopContext { + id: None, + svm_locker: simnet.locker.clone(), + simnet_commands_tx: simnet.commands.clone(), + remote_rpc_client: None, + rpc_config: RpcConfig::default(), + cheatcode_config: CheatcodeConfig::new(), + plugin_commands_tx, + }; + let base_slot = simnet.locker.get_latest_absolute_slot(); + let scenario = Scenario { + id: "sparse-scenario".to_string(), + name: "Sparse scenario".to_string(), + description: String::new(), + overrides: vec![OverrideInstance { + id: "slot-450".to_string(), + template_id: "test-template".to_string(), + values: HashMap::new(), + scenario_relative_slot: 450, + label: None, + enabled: true, + fetch_before_use: false, + account: AccountAddress::Pubkey(Pubkey::new_unique().to_string()), + }], + tags: Vec::new(), + }; + + let register_response = jsonrpc_core::futures::executor::block_on( + rpc_server.register_scenario(Some(runloop_context.clone()), scenario, Some(base_slot)), + ) + .unwrap(); + assert!(register_response.value.is_empty()); + + let error = rpc_server + .time_travel( + Some(runloop_context.clone()), + Some(TimeTravelConfig::AbsoluteSlot(base_slot + 750)), + ) + .unwrap_err(); + assert_eq!(error.code, jsonrpc_core::ErrorCode::InvalidParams); + assert!(error.message.contains(&(base_slot + 450).to_string())); + assert_eq!(simnet.locker.get_latest_absolute_slot(), base_slot); + + let outcome_response: RpcResponse> = rpc_server + .time_travel_with_override_outcomes( + Some(runloop_context.clone()), + Some(TimeTravelConfig::AbsoluteSlot(base_slot + 450)), + ) + .unwrap(); + assert_eq!(outcome_response.context.slot, base_slot + 450); + assert_eq!(outcome_response.value.len(), 1); + assert_eq!(outcome_response.value[0].override_id, "slot-450"); + assert!(outcome_response.value[0].applied); + + let current_slot = simnet.locker.get_latest_absolute_slot(); + let cancel_scenario = Scenario { + id: "cancel-scenario".to_string(), + name: "Cancel scenario".to_string(), + description: String::new(), + overrides: vec![OverrideInstance { + id: "cancel-before-750".to_string(), + template_id: "test-template".to_string(), + values: HashMap::new(), + scenario_relative_slot: 300, + label: None, + enabled: true, + fetch_before_use: false, + account: AccountAddress::Pubkey(Pubkey::new_unique().to_string()), + }], + tags: Vec::new(), + }; + jsonrpc_core::futures::executor::block_on(rpc_server.register_scenario( + Some(runloop_context.clone()), + cancel_scenario, + Some(current_slot), + )) + .unwrap(); + let cancel_response = rpc_server + .cancel_scenario_overrides( + Some(runloop_context.clone()), + vec!["cancel-before-750".to_string()], + ) + .unwrap(); + assert_eq!(cancel_response.value, 1); + let retry_cancel_response = rpc_server + .cancel_scenario_overrides( + Some(runloop_context.clone()), + vec!["cancel-before-750".to_string()], + ) + .unwrap(); + assert_eq!(retry_cancel_response.value, 0); + + let epoch_info = rpc_server + .time_travel( + Some(runloop_context), + Some(TimeTravelConfig::AbsoluteSlot(base_slot + 750)), + ) + .unwrap(); + assert_eq!(epoch_info.absolute_slot, base_slot + 750); +} + #[test_case(TestType::sqlite(); "with on-disk sqlite db")] #[test_case(TestType::in_memory(); "with in-memory sqlite db")] #[test_case(TestType::no_db(); "with no db")] diff --git a/crates/sdk-node/surfpool-sdk/kit/generated/OverrideOutcome.ts b/crates/sdk-node/surfpool-sdk/kit/generated/OverrideOutcome.ts new file mode 100644 index 000000000..0f7c4ec85 --- /dev/null +++ b/crates/sdk-node/surfpool-sdk/kit/generated/OverrideOutcome.ts @@ -0,0 +1,24 @@ +// @generated by ts-rs from the Rust types in crates/types. +// Do not edit; run `npm run generate:kit-types` in crates/sdk-node instead. +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Outcome of materializing a single override during scenario application. + */ +export type OverrideOutcome = { +/** + * The id of the override this outcome refers to. + */ +overrideId: string, +/** + * Human-readable label copied from the override, when it had one. + */ +label?: string, +/** + * Whether the override's write landed. False means it was skipped. + */ +applied: boolean, +/** + * Why the override was skipped, present only when it was not applied. + */ +reason?: string, }; diff --git a/crates/sdk-node/surfpool-sdk/kit/generated/index.ts b/crates/sdk-node/surfpool-sdk/kit/generated/index.ts index dbbe0cd89..f03568589 100644 --- a/crates/sdk-node/surfpool-sdk/kit/generated/index.ts +++ b/crates/sdk-node/surfpool-sdk/kit/generated/index.ts @@ -12,6 +12,7 @@ export type * from "./GetStreamedAccountsResponse.js"; export type * from "./GetSurfnetInfoResponse.js"; export type * from "./OfflineAccountConfig.js"; export type * from "./OverrideInstance.js"; +export type * from "./OverrideOutcome.js"; export type * from "./ParsedAccount.js"; export type * from "./PdaSeed.js"; export type * from "./ResetAccountConfig.js"; diff --git a/crates/sdk-node/surfpool-sdk/kit/generated/methods.ts b/crates/sdk-node/surfpool-sdk/kit/generated/methods.ts index 7f4613f17..170aafa13 100644 --- a/crates/sdk-node/surfpool-sdk/kit/generated/methods.ts +++ b/crates/sdk-node/surfpool-sdk/kit/generated/methods.ts @@ -1,6 +1,7 @@ // @generated by ts-rs from the Rust types in crates/types. // Do not edit; run `npm run generate:kit-types` in crates/sdk-node instead. export const SURFNET_CHEATCODE_METHODS = [ + "surfnet_cancelScenarioOverrides", "surfnet_cloneProgramAccount", "surfnet_disableCheatcode", "surfnet_enableCheatcode", @@ -26,6 +27,7 @@ export const SURFNET_CHEATCODE_METHODS = [ "surfnet_streamAccount", "surfnet_streamAccounts", "surfnet_timeTravel", + "surfnet_timeTravelWithOverrideOutcomes", "surfnet_writeProgram", ] as const; diff --git a/crates/sdk-node/surfpool-sdk/kit/types/api.ts b/crates/sdk-node/surfpool-sdk/kit/types/api.ts index 0e29dbea3..ad695cd9d 100644 --- a/crates/sdk-node/surfpool-sdk/kit/types/api.ts +++ b/crates/sdk-node/surfpool-sdk/kit/types/api.ts @@ -1,4 +1,5 @@ import type { Address, GetEpochInfoApi } from '@solana/kit'; +import type { OverrideOutcome } from '../generated/OverrideOutcome.js'; import type { AccountSnapshot, @@ -89,6 +90,9 @@ export type LocalSignatureEntry = Readonly<{ export type SurfnetTimeTravelApi = { timeTravel(config?: TimeTravelConfig): EpochInfo; }; +export type SurfnetTimeTravelWithOverrideOutcomesApi = { + timeTravelWithOverrideOutcomes(config?: TimeTravelConfig): readonly OverrideOutcome[]; +}; export type SurfnetPauseClockApi = { pauseClock(): EpochInfo; }; @@ -173,7 +177,10 @@ export type SurfnetExportSnapshotApi = { // Scenario export type SurfnetRegisterScenarioApi = { - registerScenario(scenario: Scenario, slot?: number | bigint): null; + registerScenario(scenario: Scenario, slot?: number | bigint): readonly OverrideOutcome[]; +}; +export type SurfnetCancelScenarioOverridesApi = { + cancelScenarioOverrides(overrideIds: readonly string[]): bigint; }; // Local @@ -187,7 +194,8 @@ export type SurfnetGetLocalSignaturesApi = { * All `surfnet_*` cheatcode methods, with the `surfnet_` prefix stripped * (it is re-added on the wire by the request transformer). */ -export type SurfnetCheatcodesApi = SurfnetCloneProgramAccountApi & +export type SurfnetCheatcodesApi = SurfnetCancelScenarioOverridesApi & + SurfnetCloneProgramAccountApi & SurfnetDisableCheatcodeApi & SurfnetEnableCheatcodeApi & SurfnetExportSnapshotApi & @@ -212,6 +220,7 @@ export type SurfnetCheatcodesApi = SurfnetCloneProgramAccountApi & SurfnetStreamAccountApi & SurfnetStreamAccountsApi & SurfnetTimeTravelApi & + SurfnetTimeTravelWithOverrideOutcomesApi & SurfnetWriteProgramApi; // ── Manifest coverage assertion ───────────────────────────────────────────── diff --git a/crates/types/src/rpc_endpoints.json b/crates/types/src/rpc_endpoints.json index d863d035a..1a81818eb 100644 --- a/crates/types/src/rpc_endpoints.json +++ b/crates/types/src/rpc_endpoints.json @@ -822,6 +822,18 @@ ], "returns": "An `EpochInfo` object describing the clock after the jump." }, + { + "method": "surfnet_timeTravelWithOverrideOutcomes", + "description": "Moves the Surfnet clock forward and reports overrides materialized at the target slot. The request is rejected before mutation when it would skip a pending override slot.", + "params": [ + { + "name": "config", + "type": "Option", + "description": "The same target configuration accepted by surfnet_timeTravel." + } + ], + "returns": "A `RpcResponse>` containing outcomes materialized while reaching the target slot." + }, { "method": "surfnet_pauseClock", "description": "A cheat code to freeze the Surfnet clock. Slot production and time progression halt until resumed. The paused state is not reported by any other method.", @@ -834,6 +846,18 @@ "params": [], "returns": "An `EpochInfo` object describing the clock after resuming." }, + { + "method": "surfnet_cancelScenarioOverrides", + "description": "Cancels pending scenario overrides by override id. Overrides already materialized are unaffected.", + "params": [ + { + "name": "overrideIds", + "type": "Vec", + "description": "Override ids to remove from the pending schedule. Repeated calls are safe." + } + ], + "returns": "A `RpcResponse` containing the number of pending overrides removed." + }, { "method": "surfnet_registerScenario", "description": "A cheat code to register a scenario with account overrides.", @@ -860,7 +884,7 @@ "scenarioRelativeSlot": "u64 (Relative slot when this override should be applied (relative to scenario registration slot))", "label": "Option (An optional label for this override instance)", "enabled": "bool (Indicates whether this override instance is enabled)", - "fetchBeforeUse": "bool (Indicates whether to fetch the latest on-chain account data before applying overrides)", + "fetchBeforeUse": "bool (Requires the latest remote account data before applying overrides. The override is skipped when no remote client is configured or the fetch fails.)", "account": "AccountAddress (The account this override targets, as {\"pubkey\": \"\"} or {\"pda\": {\"programId\": \"\", \"seeds\": [ ... ]}})" } } @@ -873,7 +897,7 @@ "description": "The base slot from which relative slot offsets are calculated. If omitted, uses the current slot." } ], - "returns": "A `RpcResponse<()>` indicating whether the write was successful." + "returns": "A `RpcResponse>` carrying the per-override outcomes for the base slot. Each `OverrideOutcome` is {overrideId, label?, applied, reason?}; applied=false with a reason means the override was skipped. Use surfnet_timeTravelWithOverrideOutcomes to receive later-slot outcomes synchronously as each target slot is reached." } ] }, diff --git a/crates/types/src/scenarios.rs b/crates/types/src/scenarios.rs index fb3859572..03d4f34ca 100644 --- a/crates/types/src/scenarios.rs +++ b/crates/types/src/scenarios.rs @@ -532,6 +532,47 @@ impl OverrideInstance { } } +/// Outcome of materializing a single override during scenario application. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "ts-bindings", derive(ts_rs::TS), ts(export))] +pub struct OverrideOutcome { + /// The id of the override this outcome refers to. + #[schemars(description = "The id of the override this outcome refers to")] + pub override_id: String, + /// Human-readable label copied from the override, when it had one. + #[schemars(description = "Human-readable label copied from the override, when set")] + #[cfg_attr(feature = "ts-bindings", ts(optional))] + pub label: Option, + /// Whether the override's write landed. False means it was skipped. + #[schemars(description = "Whether the override was applied; false means it was skipped")] + pub applied: bool, + /// Why the override was skipped, present only when it was not applied. + #[schemars(description = "Why the override was skipped, present only when not applied")] + #[cfg_attr(feature = "ts-bindings", ts(optional))] + pub reason: Option, +} + +impl OverrideOutcome { + pub fn applied(override_id: String, label: Option) -> Self { + Self { + override_id, + label, + applied: true, + reason: None, + } + } + + pub fn skipped(override_id: String, label: Option, reason: impl Into) -> Self { + Self { + override_id, + label, + applied: false, + reason: Some(reason.into()), + } + } +} + /// A scenario containing a timeline of overrides #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase")] diff --git a/crates/types/src/types.rs b/crates/types/src/types.rs index ae86afd2d..c4696e5e0 100644 --- a/crates/types/src/types.rs +++ b/crates/types/src/types.rs @@ -27,8 +27,8 @@ use solana_transaction_error::TransactionError; use txtx_addon_kit::indexmap::IndexMap; use uuid::Uuid; -use crate::DEFAULT_MAINNET_RPC_URL; pub use crate::startup::*; +use crate::{DEFAULT_MAINNET_RPC_URL, OverrideOutcome}; pub const DEFAULT_RPC_PORT: u16 = 8899; pub const DEFAULT_WS_PORT: u16 = 8900; @@ -738,13 +738,23 @@ pub enum TransactionStatusEvent { VerificationFailure(String), } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SimnetCommandError { + InvalidParams(String), + Internal(String), +} + #[derive(Debug)] pub enum SimnetCommand { SlotForward(Option), SlotBackward(Option), CommandClock(Option<(Hash, String)>, ClockCommand), UpdateInternalClock(Option<(Hash, String)>, Clock), - UpdateInternalClockWithConfirmation(Option<(Hash, String)>, Clock, Sender), + UpdateInternalClockWithConfirmation( + Option<(Hash, String)>, + Clock, + Sender), SimnetCommandError>>, + ), UpdateBlockProductionMode(BlockProductionMode), /// Executes a transaction. `sendTransaction` enqueues this on the same /// channel as the startup commands below, so channel order decides which @@ -1011,6 +1021,8 @@ impl CloudSurfnetRpcGating { "surfnet_getActiveIdl".into(), "surfnet_getLocalSignatures".into(), "surfnet_timeTravel".into(), + "surfnet_timeTravelWithOverrideOutcomes".into(), + "surfnet_cancelScenarioOverrides".into(), "surfnet_pauseClock".into(), "surfnet_resumeClock".into(), "surfnet_resetAccount".into(), @@ -1715,7 +1727,8 @@ pub enum CheatcodeFilter { /// `surfpool-core/src/rpc/surfnet_cheatcodes.rs` asserts it matches the /// methods actually registered by the `SurfnetCheatcodes` trait, so adding, /// removing, or renaming a cheatcode without updating this list fails CI. -pub const SURFNET_CHEATCODE_METHODS: [&str; 26] = [ +pub const SURFNET_CHEATCODE_METHODS: [&str; 28] = [ + "surfnet_cancelScenarioOverrides", "surfnet_cloneProgramAccount", "surfnet_disableCheatcode", "surfnet_enableCheatcode", @@ -1741,6 +1754,7 @@ pub const SURFNET_CHEATCODE_METHODS: [&str; 26] = [ "surfnet_streamAccount", "surfnet_streamAccounts", "surfnet_timeTravel", + "surfnet_timeTravelWithOverrideOutcomes", "surfnet_writeProgram", ]; From 691bb82813017edee04dd40aff032ef98c62ba82 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Fri, 14 Aug 2026 17:29:40 +0300 Subject: [PATCH 2/7] fix(core): make scenario scheduling failure-safe --- crates/core/src/storage/fifo_map.rs | 17 ++ crates/core/src/storage/hash_map.rs | 17 ++ crates/core/src/storage/mod.rs | 6 + crates/core/src/storage/overlay.rs | 38 ++++- crates/core/src/storage/postgres.rs | 49 +++++- crates/core/src/storage/sqlite.rs | 74 ++++++++- crates/core/src/surfnet/svm.rs | 243 +++++++++++++++++++++++----- 7 files changed, 401 insertions(+), 43 deletions(-) diff --git a/crates/core/src/storage/fifo_map.rs b/crates/core/src/storage/fifo_map.rs index 461290c59..48316f980 100644 --- a/crates/core/src/storage/fifo_map.rs +++ b/crates/core/src/storage/fifo_map.rs @@ -13,6 +13,23 @@ where Ok(()) } + fn apply_batch( + &mut self, + operations: Vec>, + ) -> super::StorageResult<()> { + for operation in operations { + match operation { + super::StorageOperation::Store(key, value) => { + self.insert(key, value); + } + super::StorageOperation::Remove(key) => { + self.remove(&key); + } + } + } + Ok(()) + } + fn clear(&mut self) -> super::StorageResult<()> { self.clear(); Ok(()) diff --git a/crates/core/src/storage/hash_map.rs b/crates/core/src/storage/hash_map.rs index 2af4c2448..1ed4366a1 100644 --- a/crates/core/src/storage/hash_map.rs +++ b/crates/core/src/storage/hash_map.rs @@ -13,6 +13,23 @@ where Ok(()) } + fn apply_batch( + &mut self, + operations: Vec>, + ) -> super::StorageResult<()> { + for operation in operations { + match operation { + super::StorageOperation::Store(key, value) => { + self.insert(key, value); + } + super::StorageOperation::Remove(key) => { + self.remove(&key); + } + } + } + Ok(()) + } + fn clear(&mut self) -> super::StorageResult<()> { self.clear(); Ok(()) diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs index 505d60f94..3ca7dd920 100644 --- a/crates/core/src/storage/mod.rs +++ b/crates/core/src/storage/mod.rs @@ -223,6 +223,11 @@ pub enum QueryExecuteError { pub type StorageResult = Result; +pub enum StorageOperation { + Store(K, V), + Remove(K), +} + impl From for jsonrpc_core::Error { fn from(err: StorageError) -> Self { SurfpoolError::from(err).into() @@ -231,6 +236,7 @@ impl From for jsonrpc_core::Error { pub trait Storage: Send + Sync { fn store(&mut self, key: K, value: V) -> StorageResult<()>; + fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()>; fn clear(&mut self) -> StorageResult<()>; fn get(&self, key: &K) -> StorageResult>; fn take(&mut self, key: &K) -> StorageResult>; diff --git a/crates/core/src/storage/overlay.rs b/crates/core/src/storage/overlay.rs index ee956c835..60ee37841 100644 --- a/crates/core/src/storage/overlay.rs +++ b/crates/core/src/storage/overlay.rs @@ -6,7 +6,7 @@ use std::{ use serde::{Deserialize, Serialize}; -use super::{OverlayDelta, OverlayLike, Storage, StorageError, StorageResult}; +use super::{OverlayDelta, OverlayLike, Storage, StorageError, StorageOperation, StorageResult}; /// Represents the state of a key in the overlay #[derive(Clone)] @@ -73,6 +73,21 @@ where Ok(()) } + fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { + let mut overlay = self.overlay.write().map_err(|_| StorageError::LockError)?; + for operation in operations { + match operation { + StorageOperation::Store(key, value) => { + overlay.insert(key, OverlayEntry::Written(value)); + } + StorageOperation::Remove(key) => { + overlay.insert(key, OverlayEntry::Deleted); + } + } + } + Ok(()) + } + fn get(&self, key: &K) -> StorageResult> { // First check overlay let overlay = self.overlay.read().map_err(|_| StorageError::LockError)?; @@ -356,6 +371,27 @@ mod tests { assert_eq!(base.get(&"key1".into()).unwrap(), Some("base_value".into())); } + #[test] + fn test_overlay_apply_batch_updates_all_entries() { + let mut base: Box> = + Box::new(StorageHashMap::::new()); + base.store("remove".into(), "base_value".into()).unwrap(); + let mut overlay = OverlayStorage::new(base); + + overlay + .apply_batch(vec![ + StorageOperation::Remove("remove".into()), + StorageOperation::Store("store".into(), "overlay_value".into()), + ]) + .unwrap(); + + assert_eq!(overlay.get(&"remove".into()).unwrap(), None); + assert_eq!( + overlay.get(&"store".into()).unwrap(), + Some("overlay_value".into()) + ); + } + #[test] fn test_overlay_keys_merges_correctly() { let mut base: Box> = diff --git a/crates/core/src/storage/postgres.rs b/crates/core/src/storage/postgres.rs index 65b0603ef..f8663ed52 100644 --- a/crates/core/src/storage/postgres.rs +++ b/crates/core/src/storage/postgres.rs @@ -6,7 +6,7 @@ use std::{ use log::debug; use serde::{Deserialize, Serialize}; use surfpool_db::diesel::{ - self, RunQueryDsl, + self, Connection, RunQueryDsl, connection::SimpleConnection, r2d2::{ConnectionManager, Pool}, sql_query, @@ -14,7 +14,7 @@ use surfpool_db::diesel::{ }; use crate::storage::{ - Storage, StorageConstructor, StorageError, StorageResult, + Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -158,6 +158,51 @@ where Ok(()) } + fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { + let mut serialized = Vec::with_capacity(operations.len()); + for operation in operations { + match operation { + StorageOperation::Store(key, value) => serialized.push(( + serialize_key(NAME, &self.table_name, &key)?, + Some(serialize_value(NAME, &self.table_name, &value)?), + )), + StorageOperation::Remove(key) => { + serialized.push((serialize_key(NAME, &self.table_name, &key)?, None)) + } + } + } + + let upsert_sql = format!( + "INSERT INTO {} (surfnet_id, key, value, updated_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP) \ + ON CONFLICT (surfnet_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP", + self.table_name + ); + let delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = $1 AND key = $2", + self.table_name + ); + let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; + + conn.transaction::<_, diesel::result::Error, _>(|conn| { + for (key, value) in &serialized { + if let Some(value) = value { + sql_query(upsert_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .bind::(value) + .execute(conn)?; + } else { + sql_query(delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .execute(conn)?; + } + } + Ok(()) + }) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + } + fn get(&self, key: &K) -> StorageResult> { debug!("Getting value from table '{}", self.table_name); let key_str = serialize_key(NAME, &self.table_name, key)?; diff --git a/crates/core/src/storage/sqlite.rs b/crates/core/src/storage/sqlite.rs index 8b025fd94..1c30390f2 100644 --- a/crates/core/src/storage/sqlite.rs +++ b/crates/core/src/storage/sqlite.rs @@ -9,7 +9,7 @@ use std::{ use log::debug; use serde::{Deserialize, Serialize}; use surfpool_db::diesel::{ - self, RunQueryDsl, + self, Connection, RunQueryDsl, connection::SimpleConnection, r2d2::{ConnectionManager, Pool}, sql_query, @@ -17,7 +17,7 @@ use surfpool_db::diesel::{ }; use crate::storage::{ - Storage, StorageConstructor, StorageError, StorageResult, + Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -273,6 +273,50 @@ where Ok(()) } + fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { + let mut serialized = Vec::with_capacity(operations.len()); + for operation in operations { + match operation { + StorageOperation::Store(key, value) => serialized.push(( + serialize_key(NAME, &self.table_name, &key)?, + Some(serialize_value(NAME, &self.table_name, &value)?), + )), + StorageOperation::Remove(key) => { + serialized.push((serialize_key(NAME, &self.table_name, &key)?, None)) + } + } + } + + let upsert_sql = format!( + "INSERT OR REPLACE INTO {} (surfnet_id, key, value, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP)", + self.table_name + ); + let delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = ? AND key = ?", + self.table_name + ); + let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; + + conn.transaction::<_, diesel::result::Error, _>(|conn| { + for (key, value) in &serialized { + if let Some(value) = value { + sql_query(upsert_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .bind::(value) + .execute(conn)?; + } else { + sql_query(delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .execute(conn)?; + } + } + Ok(()) + }) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + } + fn get(&self, key: &K) -> StorageResult> { debug!("Getting value from table '{}", self.table_name); let key_str = serialize_key(NAME, &self.table_name, key)?; @@ -588,4 +632,30 @@ mod tests { let result = storage2.get(&"key1".to_string()).unwrap(); assert!(result.is_none(), "in-memory databases should be isolated"); } + + #[test] + fn test_apply_batch_rolls_back_every_operation_when_one_fails() { + let mut storage: SqliteStorage = + SqliteStorage::connect(":memory:", "batch_test", "surfnet1").unwrap(); + storage.store(1, "old".to_string()).unwrap(); + { + let mut conn = storage.pool.get().unwrap(); + conn.batch_execute( + "CREATE TRIGGER reject_batch_value BEFORE INSERT ON batch_test \ + WHEN NEW.key = '999' BEGIN SELECT RAISE(ABORT, 'rejected'); END;", + ) + .unwrap(); + } + + let result = storage.apply_batch(vec![ + StorageOperation::Remove(1), + StorageOperation::Store(2, "new".to_string()), + StorageOperation::Store(999, "rejected".to_string()), + ]); + + assert!(result.is_err()); + assert_eq!(storage.get(&1).unwrap().as_deref(), Some("old")); + assert!(storage.get(&2).unwrap().is_none()); + assert!(storage.get(&999).unwrap().is_none()); + } } diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index 8521cd93d..b020ff47c 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -85,7 +85,7 @@ use crate::{ error::{AirdropError, SurfpoolError, SurfpoolResult}, rpc::utils::convert_transaction_metadata_from_canonical, scenarios::TemplateRegistry, - storage::{OverlayStorage, Storage, new_kv_store, new_kv_store_with_default}, + storage::{OverlayStorage, Storage, StorageOperation, new_kv_store, new_kv_store_with_default}, surfnet::{ LogsSubscriptionData, locker::is_supported_token_program, surfnet_lite_svm::SurfnetLiteSvm, }, @@ -2575,8 +2575,7 @@ impl SurfnetSvm { remote_ctx: &Option<(SurfnetRemoteClient, CommitmentConfig)>, target_slot: Slot, ) -> SurfpoolResult> { - // Remove and get overrides for this slot - let Some(overrides) = self.scheduled_overrides.take(&target_slot)? else { + let Some(overrides) = self.scheduled_overrides.get(&target_slot)? else { // No overrides for this slot return Ok(Vec::new()); }; @@ -2587,7 +2586,11 @@ impl SurfnetSvm { target_slot ); - self.materialize_override_batch(remote_ctx, overrides).await + let outcomes = self + .materialize_override_batch(remote_ctx, overrides) + .await?; + self.scheduled_overrides.take(&target_slot)?; + Ok(outcomes) } pub async fn materialize_override_batch( @@ -4144,11 +4147,32 @@ impl SurfnetSvm { Ok(override_ids) } - fn schedule_overrides( + fn replace_scheduled_overrides( &mut self, base_slot: Slot, + override_ids: &HashSet, overrides: Vec, ) -> SurfpoolResult<()> { + let mut replacements = BTreeMap::new(); + + for slot in self.scheduled_overrides.keys()? { + let Some(slot_overrides) = self.scheduled_overrides.get(&slot)? else { + continue; + }; + if slot_overrides + .iter() + .any(|override_instance| override_ids.contains(&override_instance.id)) + { + replacements.insert( + slot, + slot_overrides + .into_iter() + .filter(|override_instance| !override_ids.contains(&override_instance.id)) + .collect::>(), + ); + } + } + for override_instance in overrides { let absolute_slot = base_slot .checked_add(override_instance.scenario_relative_slot) @@ -4158,15 +4182,35 @@ impl SurfnetSvm { override_instance.id )) })?; - let mut slot_overrides = self - .scheduled_overrides - .get(&absolute_slot)? - .unwrap_or_default(); - slot_overrides.push(override_instance); - self.scheduled_overrides - .store(absolute_slot, slot_overrides)?; + if !replacements.contains_key(&absolute_slot) { + let slot_overrides = self + .scheduled_overrides + .get(&absolute_slot)? + .unwrap_or_default() + .into_iter() + .filter(|scheduled| !override_ids.contains(&scheduled.id)) + .collect(); + replacements.insert(absolute_slot, slot_overrides); + } + replacements + .get_mut(&absolute_slot) + .expect("replacement slot was inserted") + .push(override_instance); } + self.scheduled_overrides.apply_batch( + replacements + .into_iter() + .map(|(slot, overrides)| { + if overrides.is_empty() { + StorageOperation::Remove(slot) + } else { + StorageOperation::Store(slot, overrides) + } + }) + .collect(), + )?; + Ok(()) } @@ -4194,8 +4238,7 @@ impl SurfnetSvm { base_slot == current_slot && override_instance.scenario_relative_slot == 0 }); - self.cancel_scheduled_overrides(&override_ids)?; - self.schedule_overrides(base_slot, future_overrides)?; + self.replace_scheduled_overrides(base_slot, &override_ids, future_overrides)?; self.materialize_override_batch(remote_ctx, initial_overrides) .await } @@ -4209,7 +4252,7 @@ impl SurfnetSvm { .scheduled_overrides .keys()? .into_iter() - .filter(|slot| current_slot < *slot && *slot < target_slot) + .filter(|slot| current_slot.saturating_add(1) < *slot && *slot < target_slot) .min()) } @@ -4278,12 +4321,18 @@ impl SurfnetSvm { ))); } - let materialization_result = if target_slot == current_slot { - Ok(Vec::new()) - } else { + let mut outcomes = Vec::new(); + if target_slot != current_slot { self.confirm_current_block()?; - self.materialize_overrides(remote_ctx).await - }; + outcomes.extend(self.materialize_overrides(remote_ctx).await?); + } + + if target_slot == current_slot || target_slot > current_slot.saturating_add(1) { + outcomes.extend( + self.materialize_overrides_for_slot(remote_ctx, target_slot) + .await?, + ); + } self.inner.set_sysvar(&clock); self.updated_at = clock.unix_timestamp as u64 * 1_000; @@ -4294,14 +4343,6 @@ impl SurfnetSvm { .simnet_events_tx .send(SimnetEvent::SystemClockUpdated(clock)); - let mut outcomes = materialization_result?; - if target_slot == current_slot || target_slot > current_slot.saturating_add(1) { - outcomes.extend( - self.materialize_overrides_for_slot(remote_ctx, target_slot) - .await?, - ); - } - self.emit_skipped_override_events(&outcomes); Ok((self.latest_epoch_info.clone(), outcomes)) } @@ -4352,22 +4393,31 @@ mod tests { use test_case::test_case; use super::*; - use crate::storage::{Storage, StorageError, StorageResult, tests::TestType}; + use crate::storage::{Storage, StorageError, StorageOperation, StorageResult, tests::TestType}; #[derive(Clone)] - struct FailingTakeStorage; + struct FailingTakeStorage { + overrides: Vec, + } impl Storage> for FailingTakeStorage { fn store(&mut self, _key: u64, _value: Vec) -> StorageResult<()> { Ok(()) } + fn apply_batch( + &mut self, + _operations: Vec>>, + ) -> StorageResult<()> { + Ok(()) + } + fn clear(&mut self) -> StorageResult<()> { Ok(()) } fn get(&self, _key: &u64) -> StorageResult>> { - Ok(None) + Ok(Some(self.overrides.clone())) } fn take(&mut self, _key: &u64) -> StorageResult>> { @@ -4393,6 +4443,56 @@ mod tests { } } + #[derive(Clone)] + struct RejectingBatchStorage { + entries: HashMap>, + } + + impl Storage> for RejectingBatchStorage { + fn store(&mut self, key: u64, value: Vec) -> StorageResult<()> { + self.entries.insert(key, value); + Ok(()) + } + + fn apply_batch( + &mut self, + _operations: Vec>>, + ) -> StorageResult<()> { + Err(StorageError::LockError) + } + + fn clear(&mut self) -> StorageResult<()> { + self.entries.clear(); + Ok(()) + } + + fn get(&self, key: &u64) -> StorageResult>> { + Ok(self.entries.get(key).cloned()) + } + + fn take(&mut self, key: &u64) -> StorageResult>> { + Ok(self.entries.remove(key)) + } + + fn keys(&self) -> StorageResult> { + Ok(self.entries.keys().copied().collect()) + } + + fn into_iter( + &self, + ) -> StorageResult)> + '_>> { + Ok(Box::new(self.entries.clone().into_iter())) + } + + fn count(&self) -> StorageResult { + Ok(self.entries.len() as u64) + } + + fn clone_box(&self) -> Box>> { + Box::new(self.clone()) + } + } + fn test_override(id: &str, relative_slot: Slot) -> OverrideInstance { OverrideInstance { id: id.to_string(), @@ -4573,6 +4673,43 @@ mod tests { ); } + #[tokio::test] + async fn register_scenario_keeps_previous_schedule_when_replacement_batch_fails() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let base_slot = svm.latest_epoch_info.absolute_slot; + let previous_slot = base_slot + 450; + let replacement_slot = base_slot + 750; + svm.scheduled_overrides = Box::new(RejectingBatchStorage { + entries: HashMap::from([(previous_slot, vec![test_override("future", 450)])]), + }); + let replacement = surfpool_types::Scenario { + id: "scenario".to_string(), + name: "Scenario".to_string(), + description: String::new(), + overrides: vec![test_override("future", 750)], + tags: Vec::new(), + }; + + svm.register_scenario_and_materialize(&None, replacement, base_slot) + .await + .unwrap_err(); + + assert_eq!( + svm.scheduled_overrides + .get(&previous_slot) + .unwrap() + .unwrap()[0] + .scenario_relative_slot, + 450 + ); + assert!( + svm.scheduled_overrides + .get(&replacement_slot) + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn register_scenario_rejects_invalid_input_before_mutation() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); @@ -4670,6 +4807,29 @@ mod tests { assert!(retry_outcomes.is_empty()); } + #[tokio::test] + async fn time_travel_to_sparse_target_materializes_pending_next_slot() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let base_slot = svm.latest_epoch_info.absolute_slot; + svm.scheduled_overrides + .store(base_slot + 1, vec![test_override("next", 1)]) + .unwrap(); + let target_slot = base_slot + 450; + let target_clock = clock_for_absolute_slot(&svm, target_slot); + + let (epoch_info, outcomes) = svm.time_travel_to_clock(&None, target_clock).await.unwrap(); + + assert_eq!(epoch_info.absolute_slot, target_slot); + assert_eq!(outcomes.len(), 1); + assert_eq!(outcomes[0].override_id, "next"); + assert!( + svm.scheduled_overrides + .get(&(base_slot + 1)) + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn time_travel_rejects_skipping_pending_slot_before_mutation() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); @@ -4695,22 +4855,29 @@ mod tests { } #[tokio::test] - async fn time_travel_keeps_clock_state_synchronized_when_materialization_fails() { + async fn time_travel_does_not_skip_to_target_when_materialization_dequeue_fails() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); - let target_slot = svm.latest_epoch_info.absolute_slot + 1; + let current_slot = svm.latest_epoch_info.absolute_slot; + let next_slot = current_slot + 1; + let target_slot = current_slot + 450; let target_clock = clock_for_absolute_slot(&svm, target_slot); - svm.scheduled_overrides = Box::new(FailingTakeStorage); + svm.scheduled_overrides = Box::new(FailingTakeStorage { + overrides: vec![test_override("next", 1)], + }); let error = svm - .time_travel_to_clock(&None, target_clock.clone()) + .time_travel_to_clock(&None, target_clock) .await .unwrap_err(); assert!(error.to_string().contains("Storage error")); - assert_eq!(svm.latest_epoch_info.absolute_slot, target_slot); + assert_eq!(svm.latest_epoch_info.absolute_slot, next_slot); let stored_clock = svm.inner.get_sysvar::(); - assert_eq!(stored_clock.slot, target_clock.slot); - assert_eq!(stored_clock.epoch, target_clock.epoch); + assert_eq!( + stored_clock.slot, + next_slot % svm.latest_epoch_info.slots_in_epoch + ); + assert!(svm.scheduled_overrides.get(&next_slot).unwrap().is_some()); } #[test] From 63dc1fec2fc69e69c6a19f2c3a1cebcac5194f73 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Sat, 15 Aug 2026 00:47:12 +0300 Subject: [PATCH 3/7] fix(core): stage override materialization --- crates/core/src/surfnet/svm.rs | 68 ++++++++++++++++++++++++++++++++-- 1 file changed, 64 insertions(+), 4 deletions(-) diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index b020ff47c..66f36dc34 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -2586,10 +2586,20 @@ impl SurfnetSvm { target_slot ); - let outcomes = self - .materialize_override_batch(remote_ctx, overrides) - .await?; + let mut staged_inner = self.inner.clone_for_profiling(); + std::mem::swap(&mut self.inner, &mut staged_inner); + let materialization_result = self.materialize_override_batch(remote_ctx, overrides).await; + std::mem::swap(&mut self.inner, &mut staged_inner); + + let outcomes = materialization_result?; self.scheduled_overrides.take(&target_slot)?; + + if let (Some(staged_db), Some(target_db)) = + (staged_inner.db.as_ref(), self.inner.db.as_mut()) + { + commit_overlay_storage(staged_db.as_ref(), target_db.as_mut())?; + } + self.inner.svm = staged_inner.svm; Ok(outcomes) } @@ -4861,8 +4871,50 @@ mod tests { let next_slot = current_slot + 1; let target_slot = current_slot + 450; let target_clock = clock_for_absolute_slot(&svm, target_slot); + + let idl: Idl = + serde_json::from_slice(&include_bytes!("../tests/assets/idl_v1.json").to_vec()) + .unwrap(); + svm.register_idl(idl.clone(), None).unwrap(); + + #[derive(BorshSerialize)] + struct CustomAccount { + my_custom_data: u64, + another_field: String, + bool: bool, + pubkey: Pubkey, + } + + let account_pubkey = Pubkey::new_unique(); + let mut original_data = idl.accounts[0].discriminator.clone(); + CustomAccount { + my_custom_data: 42, + another_field: "original".to_string(), + bool: true, + pubkey: Pubkey::new_unique(), + } + .serialize(&mut original_data) + .unwrap(); + svm.inner + .set_account( + account_pubkey, + Account { + lamports: 1_000, + data: original_data.clone(), + owner: idl.address.parse().unwrap(), + executable: false, + rent_epoch: 0, + }, + ) + .unwrap(); + + let mut next_override = test_override("next", 1); + next_override.account = AccountAddress::Pubkey(account_pubkey.to_string()); + next_override + .values + .insert("my_custom_data".to_string(), serde_json::json!(99)); svm.scheduled_overrides = Box::new(FailingTakeStorage { - overrides: vec![test_override("next", 1)], + overrides: vec![next_override], }); let error = svm @@ -4878,6 +4930,14 @@ mod tests { next_slot % svm.latest_epoch_info.slots_in_epoch ); assert!(svm.scheduled_overrides.get(&next_slot).unwrap().is_some()); + assert_eq!( + svm.inner + .get_account(&account_pubkey) + .unwrap() + .unwrap() + .data(), + original_data + ); } #[test] From 6a71ec24922a1695f90d4d4c167334dd08bf7ba4 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Sat, 15 Aug 2026 00:58:53 +0300 Subject: [PATCH 4/7] fix(core): restore overrides after commit failure --- crates/core/src/surfnet/svm.rs | 220 ++++++++++++++++++++++++++------- 1 file changed, 177 insertions(+), 43 deletions(-) diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index 66f36dc34..e36292168 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -420,6 +420,35 @@ fn commit_overlay_storage( Ok(()) } +fn commit_overlay_storage_batch( + sandbox_storage: &dyn Storage, + target_storage: &mut dyn Storage, +) -> SurfpoolResult<()> { + let Some(overlay) = sandbox_storage.as_overlay() else { + return Ok(()); + }; + let delta = overlay.extract_overlay()?; + if delta.base_cleared { + return Err(SurfpoolError::internal( + "Cleared storage overlays cannot be committed as an atomic batch", + )); + } + + let operations = delta + .deletes + .into_iter() + .map(StorageOperation::Remove) + .chain( + delta + .writes + .into_iter() + .map(|(key, value)| StorageOperation::Store(key, value)), + ) + .collect(); + target_storage.apply_batch(operations)?; + Ok(()) +} + /// Composes a [`FeatureSet`] from a user-supplied [`SvmFeatureConfig`]. /// /// The starting baseline is LiteSVM's mainnet-beta feature set (see @@ -2586,6 +2615,7 @@ impl SurfnetSvm { target_slot ); + let scheduled_batch = overrides.clone(); let mut staged_inner = self.inner.clone_for_profiling(); std::mem::swap(&mut self.inner, &mut staged_inner); let materialization_result = self.materialize_override_batch(remote_ctx, overrides).await; @@ -2597,7 +2627,18 @@ impl SurfnetSvm { if let (Some(staged_db), Some(target_db)) = (staged_inner.db.as_ref(), self.inner.db.as_mut()) { - commit_overlay_storage(staged_db.as_ref(), target_db.as_mut())?; + if let Err(commit_error) = + commit_overlay_storage_batch(staged_db.as_ref(), target_db.as_mut()) + { + if let Err(restore_error) = + self.scheduled_overrides.store(target_slot, scheduled_batch) + { + return Err(SurfpoolError::internal(format!( + "Failed to commit staged account overrides: {commit_error}; failed to restore their schedule: {restore_error}" + ))); + } + return Err(commit_error); + } } self.inner.svm = staged_inner.svm; Ok(outcomes) @@ -4388,7 +4429,7 @@ mod tests { use base64::{Engine, engine::general_purpose}; use borsh::BorshSerialize; // use test_log::test; // uncomment to get logs from litesvm - use solana_account::Account; + use solana_account::{Account, AccountSharedData}; use solana_hash::Hash; use solana_keypair::Keypair; use solana_loader_v3_interface::get_program_data_address; @@ -4503,6 +4544,56 @@ mod tests { } } + #[derive(Clone, Default)] + struct RejectingAccountBatchStorage { + entries: HashMap, + } + + impl Storage for RejectingAccountBatchStorage { + fn store(&mut self, key: String, value: AccountSharedData) -> StorageResult<()> { + self.entries.insert(key, value); + Ok(()) + } + + fn apply_batch( + &mut self, + _operations: Vec>, + ) -> StorageResult<()> { + Err(StorageError::LockError) + } + + fn clear(&mut self) -> StorageResult<()> { + self.entries.clear(); + Ok(()) + } + + fn get(&self, key: &String) -> StorageResult> { + Ok(self.entries.get(key).cloned()) + } + + fn take(&mut self, key: &String) -> StorageResult> { + Ok(self.entries.remove(key)) + } + + fn keys(&self) -> StorageResult> { + Ok(self.entries.keys().cloned().collect()) + } + + fn into_iter( + &self, + ) -> StorageResult + '_>> { + Ok(Box::new(self.entries.clone().into_iter())) + } + + fn count(&self) -> StorageResult { + Ok(self.entries.len() as u64) + } + + fn clone_box(&self) -> Box> { + Box::new(self.clone()) + } + } + fn test_override(id: &str, relative_slot: Slot) -> OverrideInstance { OverrideInstance { id: id.to_string(), @@ -4516,6 +4607,55 @@ mod tests { } } + fn test_account_override( + svm: &mut SurfnetSvm, + id: &str, + relative_slot: Slot, + ) -> (Pubkey, Vec, OverrideInstance) { + let idl: Idl = + serde_json::from_slice(&include_bytes!("../tests/assets/idl_v1.json").to_vec()) + .unwrap(); + svm.register_idl(idl.clone(), None).unwrap(); + + #[derive(BorshSerialize)] + struct CustomAccount { + my_custom_data: u64, + another_field: String, + bool: bool, + pubkey: Pubkey, + } + + let account_pubkey = Pubkey::new_unique(); + let mut original_data = idl.accounts[0].discriminator.clone(); + CustomAccount { + my_custom_data: 42, + another_field: "original".to_string(), + bool: true, + pubkey: Pubkey::new_unique(), + } + .serialize(&mut original_data) + .unwrap(); + svm.inner + .set_account( + account_pubkey, + Account { + lamports: 1_000, + data: original_data.clone(), + owner: idl.address.parse().unwrap(), + executable: false, + rent_epoch: 0, + }, + ) + .unwrap(); + + let mut override_instance = test_override(id, relative_slot); + override_instance.account = AccountAddress::Pubkey(account_pubkey.to_string()); + override_instance + .values + .insert("my_custom_data".to_string(), serde_json::json!(99)); + (account_pubkey, original_data, override_instance) + } + fn clock_for_absolute_slot(svm: &SurfnetSvm, target_slot: Slot) -> Clock { let slots_in_epoch = svm.latest_epoch_info.slots_in_epoch; let epoch = target_slot / slots_in_epoch; @@ -4872,47 +5012,8 @@ mod tests { let target_slot = current_slot + 450; let target_clock = clock_for_absolute_slot(&svm, target_slot); - let idl: Idl = - serde_json::from_slice(&include_bytes!("../tests/assets/idl_v1.json").to_vec()) - .unwrap(); - svm.register_idl(idl.clone(), None).unwrap(); - - #[derive(BorshSerialize)] - struct CustomAccount { - my_custom_data: u64, - another_field: String, - bool: bool, - pubkey: Pubkey, - } - - let account_pubkey = Pubkey::new_unique(); - let mut original_data = idl.accounts[0].discriminator.clone(); - CustomAccount { - my_custom_data: 42, - another_field: "original".to_string(), - bool: true, - pubkey: Pubkey::new_unique(), - } - .serialize(&mut original_data) - .unwrap(); - svm.inner - .set_account( - account_pubkey, - Account { - lamports: 1_000, - data: original_data.clone(), - owner: idl.address.parse().unwrap(), - executable: false, - rent_epoch: 0, - }, - ) - .unwrap(); - - let mut next_override = test_override("next", 1); - next_override.account = AccountAddress::Pubkey(account_pubkey.to_string()); - next_override - .values - .insert("my_custom_data".to_string(), serde_json::json!(99)); + let (account_pubkey, original_data, next_override) = + test_account_override(&mut svm, "next", 1); svm.scheduled_overrides = Box::new(FailingTakeStorage { overrides: vec![next_override], }); @@ -4940,6 +5041,39 @@ mod tests { ); } + #[tokio::test] + async fn materialization_commit_failure_restores_schedule_and_account_state() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let current_slot = svm.latest_epoch_info.absolute_slot; + let (account_pubkey, original_data, override_instance) = + test_account_override(&mut svm, "current", 0); + svm.scheduled_overrides + .store(current_slot, vec![override_instance]) + .unwrap(); + svm.inner.db = Some(Box::new(RejectingAccountBatchStorage::default())); + + let error = svm + .materialize_overrides_for_slot(&None, current_slot) + .await + .unwrap_err(); + + assert!(error.to_string().contains("Storage error")); + assert!( + svm.scheduled_overrides + .get(¤t_slot) + .unwrap() + .is_some() + ); + assert_eq!( + svm.inner + .get_account(&account_pubkey) + .unwrap() + .unwrap() + .data(), + original_data + ); + } + #[test] fn startup_status_subscription_tracks_accepted_transitions() { use surfpool_types::SurfnetStartupPhase; From 408076c5f00546e67107787cfb58686975e2f421 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Sat, 15 Aug 2026 02:19:32 +0300 Subject: [PATCH 5/7] fix(core): cancel scheduled overrides atomically --- crates/core/src/error.rs | 2 +- crates/core/src/surfnet/svm.rs | 78 +++++++++++++++++++++++++--------- 2 files changed, 58 insertions(+), 22 deletions(-) diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index ef6806e4e..18c5b3b02 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -10,9 +10,9 @@ use solana_clock::Slot; use solana_pubkey::Pubkey; use solana_transaction::TransactionError; use solana_transaction_status::EncodeError; +use surfpool_types::SimnetCommandError; use crate::storage::StorageError; -use surfpool_types::SimnetCommandError; pub type SurfpoolResult = std::result::Result; diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index e36292168..c49f86260 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -449,6 +449,29 @@ fn commit_overlay_storage_batch( Ok(()) } +fn apply_scheduled_override_updates( + scheduled_overrides: &mut dyn Storage>, + updates: impl IntoIterator)>, +) -> SurfpoolResult<()> { + let operations = updates + .into_iter() + .map(|(slot, overrides)| { + if overrides.is_empty() { + StorageOperation::Remove(slot) + } else { + StorageOperation::Store(slot, overrides) + } + }) + .collect::>(); + + if operations.is_empty() { + return Ok(()); + } + + scheduled_overrides.apply_batch(operations)?; + Ok(()) +} + /// Composes a [`FeatureSet`] from a user-supplied [`SvmFeatureConfig`]. /// /// The starting baseline is LiteSVM's mainnet-beta feature set (see @@ -4249,20 +4272,7 @@ impl SurfnetSvm { .push(override_instance); } - self.scheduled_overrides.apply_batch( - replacements - .into_iter() - .map(|(slot, overrides)| { - if overrides.is_empty() { - StorageOperation::Remove(slot) - } else { - StorageOperation::Store(slot, overrides) - } - }) - .collect(), - )?; - - Ok(()) + apply_scheduled_override_updates(self.scheduled_overrides.as_mut(), replacements) } pub async fn register_scenario_and_materialize( @@ -4334,13 +4344,7 @@ impl SurfnetSvm { } } - for (slot, remaining) in updates { - if remaining.is_empty() { - self.scheduled_overrides.take(&slot)?; - } else { - self.scheduled_overrides.store(slot, remaining)?; - } - } + apply_scheduled_override_updates(self.scheduled_overrides.as_mut(), updates)?; Ok(removed) } @@ -4767,6 +4771,38 @@ mod tests { ); } + #[test] + fn cancellation_batch_failure_preserves_every_scheduled_slot() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let first_slot = 450; + let second_slot = 750; + let first_overrides = vec![test_override("cancel", first_slot)]; + let second_overrides = vec![ + test_override("cancel", second_slot), + test_override("keep", second_slot), + ]; + svm.scheduled_overrides = Box::new(RejectingBatchStorage { + entries: HashMap::from([ + (first_slot, first_overrides.clone()), + (second_slot, second_overrides.clone()), + ]), + }); + + let missing_ids = HashSet::from(["missing".to_string()]); + assert_eq!(svm.cancel_scheduled_overrides(&missing_ids).unwrap(), 0); + + let cancel_ids = HashSet::from(["cancel".to_string()]); + assert!(svm.cancel_scheduled_overrides(&cancel_ids).is_err()); + assert_eq!( + svm.scheduled_overrides.get(&first_slot).unwrap().unwrap(), + first_overrides + ); + assert_eq!( + svm.scheduled_overrides.get(&second_slot).unwrap().unwrap(), + second_overrides + ); + } + #[tokio::test] async fn register_scenario_materializes_only_initial_overrides_and_replaces_pending_ids() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); From 3601a68d478440d32e1d5dee802b5b7f31dbb0e2 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Mon, 17 Aug 2026 08:09:55 +0300 Subject: [PATCH 6/7] fix(core): atomically materialize scheduled overrides --- crates/core/src/storage/mod.rs | 27 ++++ crates/core/src/storage/postgres.rs | 119 ++++++++++++++-- crates/core/src/storage/sqlite.rs | 187 ++++++++++++++++++++++++-- crates/core/src/surfnet/svm.rs | 201 ++++++++++++++++++++++++---- 4 files changed, 483 insertions(+), 51 deletions(-) diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs index 3ca7dd920..0c642ffd5 100644 --- a/crates/core/src/storage/mod.rs +++ b/crates/core/src/storage/mod.rs @@ -120,6 +120,11 @@ pub enum StorageError { DeserializeValueError(String, serde_json::Error), #[error("Failed to acquire lock for database")] LockError, + #[error("Atomic cross-table batches are not supported by {backend} ({configuration})")] + AtomicCrossTableBatchUnsupported { + backend: String, + configuration: String, + }, #[error("Query failed for table '{0}' in '{1}' database: {2}")] QueryError(String, String, #[source] QueryExecuteError), } @@ -228,6 +233,11 @@ pub enum StorageOperation { Remove(K), } +pub struct CrossTableRemove { + pub(crate) table_name: &'static str, + pub(crate) serialized_key: String, +} + impl From for jsonrpc_core::Error { fn from(err: StorageError) -> Self { SurfpoolError::from(err).into() @@ -237,6 +247,23 @@ impl From for jsonrpc_core::Error { pub trait Storage: Send + Sync { fn store(&mut self, key: K, value: V) -> StorageResult<()>; fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()>; + fn ensure_atomic_cross_table_batch_supported(&self) -> StorageResult<()> { + Err(StorageError::AtomicCrossTableBatchUnsupported { + backend: std::any::type_name::().to_string(), + configuration: "default storage implementation".to_string(), + }) + } + fn apply_batch_with_cross_table_remove( + &mut self, + _operations: Vec>, + _cross_table_remove: CrossTableRemove, + ) -> StorageResult<()> { + self.ensure_atomic_cross_table_batch_supported()?; + Err(StorageError::AtomicCrossTableBatchUnsupported { + backend: std::any::type_name::().to_string(), + configuration: "atomic operation is not implemented".to_string(), + }) + } fn clear(&mut self) -> StorageResult<()>; fn get(&self, key: &K) -> StorageResult>; fn take(&mut self, key: &K) -> StorageResult>; diff --git a/crates/core/src/storage/postgres.rs b/crates/core/src/storage/postgres.rs index f8663ed52..4d67678d5 100644 --- a/crates/core/src/storage/postgres.rs +++ b/crates/core/src/storage/postgres.rs @@ -14,7 +14,7 @@ use surfpool_db::diesel::{ }; use crate::storage::{ - Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, + CrossTableRemove, Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -72,6 +72,24 @@ where K: Serialize + for<'de> Deserialize<'de>, V: Serialize + for<'de> Deserialize<'de> + Clone, { + fn serialize_operations( + &self, + operations: Vec>, + ) -> StorageResult)>> { + operations + .into_iter() + .map(|operation| match operation { + StorageOperation::Store(key, value) => Ok(( + serialize_key(NAME, &self.table_name, &key)?, + Some(serialize_value(NAME, &self.table_name, &value)?), + )), + StorageOperation::Remove(key) => { + Ok((serialize_key(NAME, &self.table_name, &key)?, None)) + } + }) + .collect() + } + fn ensure_table_exists(&self) -> StorageResult<()> { debug!("Ensuring table '{}' exists", self.table_name); let create_table_sql = format!( @@ -159,19 +177,49 @@ where } fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { - let mut serialized = Vec::with_capacity(operations.len()); - for operation in operations { - match operation { - StorageOperation::Store(key, value) => serialized.push(( - serialize_key(NAME, &self.table_name, &key)?, - Some(serialize_value(NAME, &self.table_name, &value)?), - )), - StorageOperation::Remove(key) => { - serialized.push((serialize_key(NAME, &self.table_name, &key)?, None)) + let serialized = self.serialize_operations(operations)?; + + let upsert_sql = format!( + "INSERT INTO {} (surfnet_id, key, value, updated_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP) \ + ON CONFLICT (surfnet_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP", + self.table_name + ); + let delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = $1 AND key = $2", + self.table_name + ); + let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; + + conn.transaction::<_, diesel::result::Error, _>(|conn| { + for (key, value) in &serialized { + if let Some(value) = value { + sql_query(upsert_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .bind::(value) + .execute(conn)?; + } else { + sql_query(delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .execute(conn)?; } } - } + Ok(()) + }) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + } + + fn ensure_atomic_cross_table_batch_supported(&self) -> StorageResult<()> { + Ok(()) + } + fn apply_batch_with_cross_table_remove( + &mut self, + operations: Vec>, + cross_table_remove: CrossTableRemove, + ) -> StorageResult<()> { + let serialized = self.serialize_operations(operations)?; let upsert_sql = format!( "INSERT INTO {} (surfnet_id, key, value, updated_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP) \ ON CONFLICT (surfnet_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP", @@ -181,6 +229,10 @@ where "DELETE FROM {} WHERE surfnet_id = $1 AND key = $2", self.table_name ); + let cross_table_delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = $1 AND key = $2", + cross_table_remove.table_name + ); let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; conn.transaction::<_, diesel::result::Error, _>(|conn| { @@ -198,9 +250,13 @@ where .execute(conn)?; } } + sql_query(cross_table_delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(&cross_table_remove.serialized_key) + .execute(conn)?; Ok(()) }) - .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*atomic-batch*", e)) } fn get(&self, key: &K) -> StorageResult> { @@ -390,3 +446,42 @@ where Ok(storage) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_atomic_cross_table_batch_rolls_back_postgres() { + let Ok(database_url) = std::env::var("SURFPOOL_TEST_POSTGRES_URL") else { + eprintln!("SURFPOOL_TEST_POSTGRES_URL is not set; skipping PostgreSQL fault test"); + return; + }; + let surfnet_id = uuid::Uuid::new_v4().to_string(); + let mut accounts: PostgresStorage = + PostgresStorage::connect(&database_url, "accounts", &surfnet_id).unwrap(); + accounts + .store("account".to_string(), "old".to_string()) + .unwrap(); + let missing_table = Box::leak( + format!("missing_atomic_table_{}", uuid::Uuid::new_v4().simple()).into_boxed_str(), + ); + + let result = accounts.apply_batch_with_cross_table_remove( + vec![StorageOperation::Store( + "account".to_string(), + "new".to_string(), + )], + CrossTableRemove { + table_name: missing_table, + serialized_key: serde_json::to_string(&1_u64).unwrap(), + }, + ); + + assert!(result.is_err()); + assert_eq!( + accounts.get(&"account".to_string()).unwrap().as_deref(), + Some("old") + ); + } +} diff --git a/crates/core/src/storage/sqlite.rs b/crates/core/src/storage/sqlite.rs index 1c30390f2..5eec23983 100644 --- a/crates/core/src/storage/sqlite.rs +++ b/crates/core/src/storage/sqlite.rs @@ -17,7 +17,7 @@ use surfpool_db::diesel::{ }; use crate::storage::{ - Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, + CrossTableRemove, Storage, StorageConstructor, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -112,6 +112,8 @@ pub struct SqliteStorage { is_file_based: bool, /// The connection string for creating direct connections during cleanup connection_string: String, + cross_table_transactions_supported: bool, + storage_configuration: &'static str, } const NAME: &str = "SQLite"; @@ -190,6 +192,24 @@ where K: Serialize + for<'de> Deserialize<'de>, V: Serialize + for<'de> Deserialize<'de> + Clone, { + fn serialize_operations( + &self, + operations: Vec>, + ) -> StorageResult)>> { + operations + .into_iter() + .map(|operation| match operation { + StorageOperation::Store(key, value) => Ok(( + serialize_key(NAME, &self.table_name, &key)?, + Some(serialize_value(NAME, &self.table_name, &value)?), + )), + StorageOperation::Remove(key) => { + Ok((serialize_key(NAME, &self.table_name, &key)?, None)) + } + }) + .collect() + } + fn ensure_table_exists(&self) -> StorageResult<()> { debug!("Ensuring table '{}' exists", self.table_name); let create_table_sql = format!( @@ -274,19 +294,55 @@ where } fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { - let mut serialized = Vec::with_capacity(operations.len()); - for operation in operations { - match operation { - StorageOperation::Store(key, value) => serialized.push(( - serialize_key(NAME, &self.table_name, &key)?, - Some(serialize_value(NAME, &self.table_name, &value)?), - )), - StorageOperation::Remove(key) => { - serialized.push((serialize_key(NAME, &self.table_name, &key)?, None)) + let serialized = self.serialize_operations(operations)?; + + let upsert_sql = format!( + "INSERT OR REPLACE INTO {} (surfnet_id, key, value, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP)", + self.table_name + ); + let delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = ? AND key = ?", + self.table_name + ); + let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; + + conn.transaction::<_, diesel::result::Error, _>(|conn| { + for (key, value) in &serialized { + if let Some(value) = value { + sql_query(upsert_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .bind::(value) + .execute(conn)?; + } else { + sql_query(delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(key) + .execute(conn)?; } } + Ok(()) + }) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + } + + fn ensure_atomic_cross_table_batch_supported(&self) -> StorageResult<()> { + if !self.cross_table_transactions_supported { + return Err(StorageError::AtomicCrossTableBatchUnsupported { + backend: NAME.to_string(), + configuration: self.storage_configuration.to_string(), + }); } + Ok(()) + } + fn apply_batch_with_cross_table_remove( + &mut self, + operations: Vec>, + cross_table_remove: CrossTableRemove, + ) -> StorageResult<()> { + self.ensure_atomic_cross_table_batch_supported()?; + let serialized = self.serialize_operations(operations)?; let upsert_sql = format!( "INSERT OR REPLACE INTO {} (surfnet_id, key, value, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP)", self.table_name @@ -295,6 +351,10 @@ where "DELETE FROM {} WHERE surfnet_id = ? AND key = ?", self.table_name ); + let cross_table_delete_sql = format!( + "DELETE FROM {} WHERE surfnet_id = ? AND key = ?", + cross_table_remove.table_name + ); let mut conn = self.pool.get().map_err(|_| StorageError::LockError)?; conn.transaction::<_, diesel::result::Error, _>(|conn| { @@ -312,9 +372,13 @@ where .execute(conn)?; } } + sql_query(cross_table_delete_sql.clone()) + .bind::(&self.surfnet_id) + .bind::(&cross_table_remove.serialized_key) + .execute(conn)?; Ok(()) }) - .map_err(|e| StorageError::store(&self.table_name, NAME, "*batch*", e)) + .map_err(|e| StorageError::store(&self.table_name, NAME, "*atomic-batch*", e)) } fn get(&self, key: &K) -> StorageResult> { @@ -493,6 +557,8 @@ where // Unique name per storage instance; cache=shared so pool connections share it let id = MEMORY_DB_COUNTER.fetch_add(1, Ordering::Relaxed); format!("file:memdb{}?mode=memory&cache=shared", id) + } else if database_url.starts_with("file:") && database_url.contains("mode=memory") { + database_url.to_string() } else if database_url.starts_with("file:") { if database_url.contains('?') { format!("{}&mode=rwc", database_url) @@ -503,7 +569,19 @@ where format!("file:{}?mode=rwc", database_url) }; - let is_file_based = database_url != ":memory:"; + let is_file_based = !connection_string.contains("mode=memory"); + let has_shared_memory_cache = connection_string.contains("cache=shared"); + let cross_table_transactions_supported = + is_file_based || (database_url != ":memory:" && has_shared_memory_cache); + let storage_configuration = if is_file_based { + "file-backed database" + } else if database_url == ":memory:" { + "isolated :memory: storage instance" + } else if has_shared_memory_cache { + "named shared-cache in-memory database" + } else { + "named in-memory database without cache=shared" + }; let pool = get_or_create_shared_pool(&connection_string, is_file_based)?; let storage = SqliteStorage { @@ -513,6 +591,8 @@ where surfnet_id: surfnet_id.to_string(), is_file_based, connection_string, + cross_table_transactions_supported, + storage_configuration, }; storage.ensure_table_exists()?; @@ -658,4 +738,87 @@ mod tests { assert!(storage.get(&2).unwrap().is_none()); assert!(storage.get(&999).unwrap().is_none()); } + + fn assert_atomic_cross_table_rollback(database_url: &str) { + let mut accounts: SqliteStorage = + SqliteStorage::connect(database_url, "accounts", "surfnet1").unwrap(); + let mut scheduled_overrides: SqliteStorage = + SqliteStorage::connect(database_url, "scheduled_overrides", "surfnet1").unwrap(); + accounts + .store("account".to_string(), "old".to_string()) + .unwrap(); + scheduled_overrides.store(1, "pending".to_string()).unwrap(); + { + let mut conn = scheduled_overrides.pool.get().unwrap(); + conn.batch_execute( + "CREATE TRIGGER reject_schedule_delete BEFORE DELETE ON scheduled_overrides \ + WHEN OLD.key = '1' BEGIN SELECT RAISE(ABORT, 'rejected'); END;", + ) + .unwrap(); + } + + let result = accounts.apply_batch_with_cross_table_remove( + vec![StorageOperation::Store( + "account".to_string(), + "new".to_string(), + )], + CrossTableRemove { + table_name: "scheduled_overrides", + serialized_key: serde_json::to_string(&1_u64).unwrap(), + }, + ); + + assert!(result.is_err()); + assert_eq!( + accounts.get(&"account".to_string()).unwrap().as_deref(), + Some("old") + ); + assert_eq!( + scheduled_overrides.get(&1).unwrap().as_deref(), + Some("pending") + ); + } + + #[test] + fn test_atomic_cross_table_batch_rolls_back_file_sqlite() { + let temp_file = tempfile::NamedTempFile::new().unwrap(); + assert_atomic_cross_table_rollback(temp_file.path().to_str().unwrap()); + } + + #[test] + fn test_atomic_cross_table_batch_rolls_back_named_memory_sqlite() { + let id = MEMORY_DB_COUNTER.fetch_add(1, Ordering::Relaxed); + let database_url = format!("file:atomic-test-{id}?mode=memory&cache=shared"); + assert_atomic_cross_table_rollback(&database_url); + } + + #[test] + fn test_named_memory_database_survives_one_storage_pool_drop() { + let id = MEMORY_DB_COUNTER.fetch_add(1, Ordering::Relaxed); + let database_url = format!("file:lifecycle-test-{id}?mode=memory&cache=shared"); + let accounts: SqliteStorage = + SqliteStorage::connect(&database_url, "accounts", "surfnet1").unwrap(); + let mut scheduled_overrides: SqliteStorage = + SqliteStorage::connect(&database_url, "scheduled_overrides", "surfnet1").unwrap(); + scheduled_overrides.store(1, "pending".to_string()).unwrap(); + + drop(accounts); + + assert_eq!( + scheduled_overrides.get(&1).unwrap().as_deref(), + Some("pending") + ); + let mut reconnected_accounts: SqliteStorage = + SqliteStorage::connect(&database_url, "accounts", "surfnet1").unwrap(); + reconnected_accounts + .store("account".to_string(), "value".to_string()) + .unwrap(); + assert_eq!( + reconnected_accounts + .get(&"account".to_string()) + .unwrap() + .as_deref(), + Some("value") + ); + } } diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index c49f86260..91172d6e8 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -2,7 +2,10 @@ use std::{ cmp::max, collections::{BTreeMap, HashMap, HashSet, VecDeque}, str::FromStr, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, time::SystemTime, }; @@ -85,7 +88,10 @@ use crate::{ error::{AirdropError, SurfpoolError, SurfpoolResult}, rpc::utils::convert_transaction_metadata_from_canonical, scenarios::TemplateRegistry, - storage::{OverlayStorage, Storage, StorageOperation, new_kv_store, new_kv_store_with_default}, + storage::{ + CrossTableRemove, OverlayStorage, Storage, StorageOperation, new_kv_store, + new_kv_store_with_default, + }, surfnet::{ LogsSubscriptionData, locker::is_supported_token_program, surfnet_lite_svm::SurfnetLiteSvm, }, @@ -117,6 +123,19 @@ lazy_static::lazy_static! { }; } +static MATERIALIZATION_MEMORY_DB_COUNTER: AtomicU64 = AtomicU64::new(0); + +fn materialization_database_url(database_url: Option<&str>) -> Option { + database_url.map(|url| { + if url == ":memory:" { + let id = MATERIALIZATION_MEMORY_DB_COUNTER.fetch_add(1, Ordering::Relaxed); + format!("file:surfpool-materialization-{id}?mode=memory&cache=shared") + } else { + url.to_string() + } + }) +} + /// Helper function to apply an override to a decoded account value using dot notation pub fn apply_override_to_decoded_account( decoded_value: &mut Value, @@ -423,6 +442,7 @@ fn commit_overlay_storage( fn commit_overlay_storage_batch( sandbox_storage: &dyn Storage, target_storage: &mut dyn Storage, + cross_table_remove: CrossTableRemove, ) -> SurfpoolResult<()> { let Some(overlay) = sandbox_storage.as_overlay() else { return Ok(()); @@ -445,7 +465,7 @@ fn commit_overlay_storage_batch( .map(|(key, value)| StorageOperation::Store(key, value)), ) .collect(); - target_storage.apply_batch(operations)?; + target_storage.apply_batch_with_cross_table_remove(operations, cross_table_remove)?; Ok(()) } @@ -890,13 +910,18 @@ impl SurfnetSvm { let (simnet_events_tx, simnet_events_rx) = SimnetEventsTx::channel(1024); let (geyser_events_tx, geyser_events_rx) = crossbeam_channel::bounded(1024); let surfnet_id = config.surfnet_id; + let materialization_database_url = materialization_database_url(database_url); // Compose the final feature set up front (mainnet baseline + // config.enable - config.disable) so that the inner LiteSVM is // constructed exactly once, with the correct features and feature // accounts loaded. See `compose_feature_set` for the composition rules. let feature_set = compose_feature_set(&config.feature_config); - let inner = SurfnetLiteSvm::new(database_url, &surfnet_id, feature_set.clone())?; + let inner = SurfnetLiteSvm::new( + materialization_database_url.as_deref(), + &surfnet_id, + feature_set.clone(), + )?; let native_mint_account = inner .get_account(&spl_token_interface::native_mint::ID)? @@ -959,8 +984,11 @@ impl SurfnetSvm { new_kv_store(&database_url, "token_accounts_by_mint", &surfnet_id)?; let streamed_accounts_db: Box> = new_kv_store(&database_url, "streamed_accounts", &surfnet_id)?; - let scheduled_overrides_db: Box>> = - new_kv_store(&database_url, "scheduled_overrides", &surfnet_id)?; + let scheduled_overrides_db: Box>> = new_kv_store( + &materialization_database_url.as_deref(), + "scheduled_overrides", + &surfnet_id, + )?; let offline_accounts_db: Box> = new_kv_store(&database_url, "offline_accounts", &surfnet_id)?; let registered_idls_db: Box>> = @@ -2632,36 +2660,41 @@ impl SurfnetSvm { return Ok(Vec::new()); }; + if let Some(target_db) = self.inner.db.as_ref() { + target_db.ensure_atomic_cross_table_batch_supported()?; + } + debug!( "Materializing {} override(s) for slot {}", overrides.len(), target_slot ); - let scheduled_batch = overrides.clone(); let mut staged_inner = self.inner.clone_for_profiling(); std::mem::swap(&mut self.inner, &mut staged_inner); let materialization_result = self.materialize_override_batch(remote_ctx, overrides).await; std::mem::swap(&mut self.inner, &mut staged_inner); let outcomes = materialization_result?; - self.scheduled_overrides.take(&target_slot)?; if let (Some(staged_db), Some(target_db)) = (staged_inner.db.as_ref(), self.inner.db.as_mut()) { - if let Err(commit_error) = - commit_overlay_storage_batch(staged_db.as_ref(), target_db.as_mut()) - { - if let Err(restore_error) = - self.scheduled_overrides.store(target_slot, scheduled_batch) - { - return Err(SurfpoolError::internal(format!( - "Failed to commit staged account overrides: {commit_error}; failed to restore their schedule: {restore_error}" - ))); - } - return Err(commit_error); - } + let serialized_slot = serde_json::to_string(&target_slot).map_err(|error| { + SurfpoolError::internal(format!( + "Failed to serialize scheduled override slot {target_slot}: {error}" + )) + })?; + commit_overlay_storage_batch( + staged_db.as_ref(), + target_db.as_mut(), + CrossTableRemove { + table_name: "scheduled_overrides", + serialized_key: serialized_slot, + }, + )?; + } else { + self.scheduled_overrides.take(&target_slot)?; } self.inner.svm = staged_inner.svm; Ok(outcomes) @@ -4394,9 +4427,7 @@ impl SurfnetSvm { self.latest_epoch_info.slot_index = clock.slot; self.latest_epoch_info.epoch = clock.epoch; self.latest_epoch_info.absolute_slot = target_slot; - let _ = self - .simnet_events_tx - .send(SimnetEvent::SystemClockUpdated(clock)); + self.simnet_events_tx.system_clock_updated(clock); self.emit_skipped_override_events(&outcomes); Ok((self.latest_epoch_info.clone(), outcomes)) @@ -4408,14 +4439,14 @@ impl SurfnetSvm { ) { for outcome in outcomes { if !outcome.applied { - let _ = self.simnet_events_tx.send(SimnetEvent::warn(format!( + self.simnet_events_tx.warn(format!( "Override {} was skipped: {}", outcome .label .as_deref() .unwrap_or(outcome.override_id.as_str()), outcome.reason.as_deref().unwrap_or("unknown reason"), - ))); + )); } } } @@ -4551,6 +4582,8 @@ mod tests { #[derive(Clone, Default)] struct RejectingAccountBatchStorage { entries: HashMap, + atomic_supported: bool, + panic_on_clone: bool, } impl Storage for RejectingAccountBatchStorage { @@ -4566,6 +4599,26 @@ mod tests { Err(StorageError::LockError) } + fn ensure_atomic_cross_table_batch_supported(&self) -> StorageResult<()> { + if self.atomic_supported { + Ok(()) + } else { + Err(StorageError::AtomicCrossTableBatchUnsupported { + backend: "test-account-storage".to_string(), + configuration: "atomic batches disabled".to_string(), + }) + } + } + + fn apply_batch_with_cross_table_remove( + &mut self, + _operations: Vec>, + _cross_table_remove: CrossTableRemove, + ) -> StorageResult<()> { + self.ensure_atomic_cross_table_batch_supported()?; + Err(StorageError::LockError) + } + fn clear(&mut self) -> StorageResult<()> { self.entries.clear(); Ok(()) @@ -4594,6 +4647,7 @@ mod tests { } fn clone_box(&self) -> Box> { + assert!(!self.panic_on_clone, "unsupported storage was cloned"); Box::new(self.clone()) } } @@ -5077,8 +5131,44 @@ mod tests { ); } + #[test_case(TestType::sqlite(); "with on-disk sqlite db")] + #[test_case(TestType::in_memory(); "with in-memory sqlite db")] #[tokio::test] - async fn materialization_commit_failure_restores_schedule_and_account_state() { + async fn materialization_atomically_commits_account_and_schedule(test_type: TestType) { + let (mut svm, _events_rx, _geyser_rx) = test_type.initialize_svm(); + let current_slot = svm.latest_epoch_info.absolute_slot; + let (account_pubkey, original_data, override_instance) = + test_account_override(&mut svm, "current", 0); + svm.scheduled_overrides + .store(current_slot, vec![override_instance]) + .unwrap(); + + let outcomes = svm + .materialize_overrides_for_slot(&None, current_slot) + .await + .unwrap(); + + assert_eq!(outcomes.len(), 1); + assert!(outcomes[0].applied); + assert!( + svm.scheduled_overrides + .get(¤t_slot) + .unwrap() + .is_none() + ); + let stored_account = svm + .inner + .db + .as_ref() + .unwrap() + .get(&account_pubkey.to_string()) + .unwrap() + .unwrap(); + assert_ne!(stored_account.data(), original_data); + } + + #[tokio::test] + async fn materialization_atomic_commit_failure_preserves_schedule_and_account_state() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); let current_slot = svm.latest_epoch_info.absolute_slot; let (account_pubkey, original_data, override_instance) = @@ -5086,7 +5176,10 @@ mod tests { svm.scheduled_overrides .store(current_slot, vec![override_instance]) .unwrap(); - svm.inner.db = Some(Box::new(RejectingAccountBatchStorage::default())); + svm.inner.db = Some(Box::new(RejectingAccountBatchStorage { + atomic_supported: true, + ..RejectingAccountBatchStorage::default() + })); let error = svm .materialize_overrides_for_slot(&None, current_slot) @@ -5110,6 +5203,43 @@ mod tests { ); } + #[tokio::test] + async fn materialization_rejects_unsupported_storage_before_mutation() { + let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); + let current_slot = svm.latest_epoch_info.absolute_slot; + let (account_pubkey, original_data, override_instance) = + test_account_override(&mut svm, "current", 0); + svm.scheduled_overrides + .store(current_slot, vec![override_instance]) + .unwrap(); + svm.inner.db = Some(Box::new(RejectingAccountBatchStorage { + panic_on_clone: true, + ..RejectingAccountBatchStorage::default() + })); + + let error = svm + .materialize_overrides_for_slot(&None, current_slot) + .await + .unwrap_err(); + + assert!(error.to_string().contains("test-account-storage")); + assert!(error.to_string().contains("atomic batches disabled")); + assert!( + svm.scheduled_overrides + .get(¤t_slot) + .unwrap() + .is_some() + ); + assert_eq!( + svm.inner + .get_account(&account_pubkey) + .unwrap() + .unwrap() + .data(), + original_data + ); + } + #[test] fn startup_status_subscription_tracks_accepted_transitions() { use surfpool_types::SurfnetStartupPhase; @@ -5931,6 +6061,23 @@ mod tests { assert!(svm.inner.db.is_some()); } + #[cfg(feature = "sqlite")] + #[test] + fn test_in_memory_materialization_domains_are_isolated_per_surfnet() { + let (mut first, _events_rx, _geyser_rx) = + SurfnetSvm::new_with_db(Some(":memory:"), SurfnetSvmConfig::default()).unwrap(); + let (second, _events_rx, _geyser_rx) = + SurfnetSvm::new_with_db(Some(":memory:"), SurfnetSvmConfig::default()).unwrap(); + let slot = first.latest_epoch_info.absolute_slot; + first + .scheduled_overrides + .store(slot, vec![test_override("first", 0)]) + .unwrap(); + + assert!(first.scheduled_overrides.get(&slot).unwrap().is_some()); + assert!(second.scheduled_overrides.get(&slot).unwrap().is_none()); + } + #[test_case(TestType::sqlite(); "with on-disk sqlite db")] #[cfg_attr(feature = "postgres", test_case(TestType::postgres(); "with postgres db"))] fn test_new_with_db_restores_slot_checkpoint(test_type: TestType) { From 2fdec3d440250a576ae88593652fe8e16e11e1e8 Mon Sep 17 00:00:00 2001 From: 92Infinitus92 <92georgepetroff92@gmail.com> Date: Mon, 17 Aug 2026 12:25:51 +0300 Subject: [PATCH 7/7] fix(core): retry current-slot overrides before time travel --- crates/core/src/surfnet/svm.rs | 67 ++++++++++++++++++++++++++-------- 1 file changed, 52 insertions(+), 15 deletions(-) diff --git a/crates/core/src/surfnet/svm.rs b/crates/core/src/surfnet/svm.rs index 91172d6e8..d3c4444ee 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -4409,13 +4409,15 @@ impl SurfnetSvm { ))); } - let mut outcomes = Vec::new(); + let mut outcomes = self + .materialize_overrides_for_slot(remote_ctx, current_slot) + .await?; if target_slot != current_slot { self.confirm_current_block()?; outcomes.extend(self.materialize_overrides(remote_ctx).await?); } - if target_slot == current_slot || target_slot > current_slot.saturating_add(1) { + if target_slot > current_slot.saturating_add(1) { outcomes.extend( self.materialize_overrides_for_slot(remote_ctx, target_slot) .await?, @@ -4482,11 +4484,13 @@ mod tests { use crate::storage::{Storage, StorageError, StorageOperation, StorageResult, tests::TestType}; #[derive(Clone)] - struct FailingTakeStorage { - overrides: Vec, + struct FailOnceTakeStorage { + slot: u64, + overrides: Option>, + fail_next_take: bool, } - impl Storage> for FailingTakeStorage { + impl Storage> for FailOnceTakeStorage { fn store(&mut self, _key: u64, _value: Vec) -> StorageResult<()> { Ok(()) } @@ -4502,16 +4506,31 @@ mod tests { Ok(()) } - fn get(&self, _key: &u64) -> StorageResult>> { - Ok(Some(self.overrides.clone())) + fn get(&self, key: &u64) -> StorageResult>> { + if *key == self.slot { + Ok(self.overrides.clone()) + } else { + Ok(None) + } } - fn take(&mut self, _key: &u64) -> StorageResult>> { - Err(StorageError::LockError) + fn take(&mut self, key: &u64) -> StorageResult>> { + if *key != self.slot { + return Ok(None); + } + if self.fail_next_take { + self.fail_next_take = false; + return Err(StorageError::LockError); + } + Ok(self.overrides.take()) } fn keys(&self) -> StorageResult> { - Ok(Vec::new()) + Ok(self + .overrides + .as_ref() + .map(|_| vec![self.slot]) + .unwrap_or_default()) } fn into_iter( @@ -4521,7 +4540,7 @@ mod tests { } fn count(&self) -> StorageResult { - Ok(0) + Ok(u64::from(self.overrides.is_some())) } fn clone_box(&self) -> Box>> { @@ -5095,7 +5114,7 @@ mod tests { } #[tokio::test] - async fn time_travel_does_not_skip_to_target_when_materialization_dequeue_fails() { + async fn time_travel_retries_failed_current_slot_before_advancing() { let (mut svm, _events_rx, _geyser_rx) = SurfnetSvm::default(); let current_slot = svm.latest_epoch_info.absolute_slot; let next_slot = current_slot + 1; @@ -5104,12 +5123,14 @@ mod tests { let (account_pubkey, original_data, next_override) = test_account_override(&mut svm, "next", 1); - svm.scheduled_overrides = Box::new(FailingTakeStorage { - overrides: vec![next_override], + svm.scheduled_overrides = Box::new(FailOnceTakeStorage { + slot: next_slot, + overrides: Some(vec![next_override]), + fail_next_take: true, }); let error = svm - .time_travel_to_clock(&None, target_clock) + .time_travel_to_clock(&None, target_clock.clone()) .await .unwrap_err(); @@ -5129,6 +5150,22 @@ mod tests { .data(), original_data ); + + let (epoch_info, outcomes) = svm.time_travel_to_clock(&None, target_clock).await.unwrap(); + + assert_eq!(epoch_info.absolute_slot, target_slot); + assert_eq!(outcomes.len(), 1); + assert_eq!(outcomes[0].override_id, "next"); + assert!(outcomes[0].applied); + assert!(svm.scheduled_overrides.get(&next_slot).unwrap().is_none()); + assert_ne!( + svm.inner + .get_account(&account_pubkey) + .unwrap() + .unwrap() + .data(), + original_data + ); } #[test_case(TestType::sqlite(); "with on-disk sqlite db")]