diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 6a8b9ae0a..18c5b3b02 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; @@ -10,6 +10,7 @@ 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; @@ -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/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..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), } @@ -223,6 +228,16 @@ pub enum QueryExecuteError { pub type StorageResult = Result; +pub enum StorageOperation { + Store(K, V), + 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() @@ -231,6 +246,24 @@ 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/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..4d67678d5 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, + 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!( @@ -158,6 +176,89 @@ where Ok(()) } + fn apply_batch(&mut self, operations: Vec>) -> 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", + 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", + self.table_name + ); + let delete_sql = format!( + "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| { + 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)?; + } + } + 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, "*atomic-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)?; @@ -345,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 8b025fd94..5eec23983 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, + 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!( @@ -273,6 +293,94 @@ where Ok(()) } + fn apply_batch(&mut self, operations: Vec>) -> StorageResult<()> { + 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 + ); + let delete_sql = format!( + "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| { + 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)?; + } + } + 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, "*atomic-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)?; @@ -449,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) @@ -459,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 { @@ -469,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()?; @@ -588,4 +712,113 @@ 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()); + } + + 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/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..d3c4444ee 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, 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, @@ -420,6 +439,59 @@ fn commit_overlay_storage( Ok(()) } +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(()); + }; + 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_with_cross_table_remove(operations, cross_table_remove)?; + 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 @@ -838,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)? @@ -907,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>> = @@ -2562,7 +2642,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,22 +2654,72 @@ impl SurfnetSvm { &mut self, 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 { + ) -> SurfpoolResult> { + let Some(overrides) = self.scheduled_overrides.get(&target_slot)? else { // No overrides for this slot - return Ok(()); + 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 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?; + + if let (Some(staged_db), Some(target_db)) = + (staged_inner.db.as_ref(), self.inner.db.as_mut()) + { + 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) + } + + 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 +2745,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 +2759,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 +2781,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 +2804,7 @@ impl SurfnetSvm { "Failed to fetch account {} from remote: {}", account_pubkey, e ); + fetch_error = Some(format!("fetchBeforeUse fetch failed: {e}")); } } } else { @@ -2671,6 +2813,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 +2839,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 +2857,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 +2876,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 +2888,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 +2903,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 +2925,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 +2947,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 +2971,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 +2983,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 +4216,106 @@ 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 replace_scheduled_overrides( &mut self, - scenario: surfpool_types::Scenario, - slot: Option, + base_slot: Slot, + override_ids: &HashSet, + 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); + 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) + .ok_or_else(|| { + SurfpoolError::invalid_params(format!( + "Scenario override {} exceeds the maximum slot", + override_instance.id + )) + })?; + 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); + } + + apply_scheduled_override_updates(self.scheduled_overrides.as_mut(), replacements) + } + + 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 +4325,132 @@ 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.replace_scheduled_overrides(base_slot, &override_ids, 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.saturating_add(1) < *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)); + } + } + + apply_scheduled_override_updates(self.scheduled_overrides.as_mut(), updates)?; + + 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 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.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; + self.latest_epoch_info.slot_index = clock.slot; + self.latest_epoch_info.epoch = clock.epoch; + self.latest_epoch_info.absolute_slot = target_slot; + self.simnet_events_tx.system_clock_updated(clock); + + 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 { + 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"), + )); + } + } } } @@ -4083,7 +4466,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; @@ -4094,11 +4477,805 @@ 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, StorageOperation, StorageResult, tests::TestType}; + + #[derive(Clone)] + struct FailOnceTakeStorage { + slot: u64, + overrides: Option>, + fail_next_take: bool, + } + + impl Storage> for FailOnceTakeStorage { + 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>> { + if *key == self.slot { + Ok(self.overrides.clone()) + } else { + Ok(None) + } + } + + 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(self + .overrides + .as_ref() + .map(|_| vec![self.slot]) + .unwrap_or_default()) + } + + fn into_iter( + &self, + ) -> StorageResult)> + '_>> { + Ok(Box::new(std::iter::empty())) + } + + fn count(&self) -> StorageResult { + Ok(u64::from(self.overrides.is_some())) + } + + fn clone_box(&self) -> Box>> { + Box::new(self.clone()) + } + } + + #[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()) + } + } + + #[derive(Clone, Default)] + struct RejectingAccountBatchStorage { + entries: HashMap, + atomic_supported: bool, + panic_on_clone: bool, + } + + 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 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(()) + } + + 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> { + assert!(!self.panic_on_clone, "unsupported storage was cloned"); + 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 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; + 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"] + ); + } + + #[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(); + 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_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(); + 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_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(); + 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_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; + let target_slot = current_slot + 450; + let target_clock = clock_for_absolute_slot(&svm, target_slot); + + let (account_pubkey, original_data, next_override) = + test_account_override(&mut svm, "next", 1); + 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.clone()) + .await + .unwrap_err(); + + assert!(error.to_string().contains("Storage error")); + assert_eq!(svm.latest_epoch_info.absolute_slot, next_slot); + let stored_clock = svm.inner.get_sysvar::(); + assert_eq!( + stored_clock.slot, + 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 + ); + + 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")] + #[test_case(TestType::in_memory(); "with in-memory sqlite db")] + #[tokio::test] + 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) = + test_account_override(&mut svm, "current", 0); + svm.scheduled_overrides + .store(current_slot, vec![override_instance]) + .unwrap(); + svm.inner.db = Some(Box::new(RejectingAccountBatchStorage { + atomic_supported: true, + ..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 + ); + } + + #[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() { @@ -4921,6 +6098,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) { 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", ];