diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index fc1ec14c..f70b760a 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; @@ -11,6 +11,7 @@ use solana_pubkey::Pubkey; use solana_signature::Signature; use solana_transaction::TransactionError; use solana_transaction_status::EncodeError; +use surfpool_types::SimnetCommandError; use crate::storage::StorageError; @@ -85,6 +86,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!( @@ -429,6 +438,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", @@ -577,3 +595,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 445aaf5c..0319d3a2 100644 --- a/crates/core/src/rpc/surfnet_cheatcodes.rs +++ b/crates/core/src/rpc/surfnet_cheatcodes.rs @@ -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, TimeTravelResult, UiKeyedProfileResult, types::{ AccountUpdate, ConfidentialBalanceKeys, DeriveConfidentialKeysResponse, GetConfidentialBalanceResponse, SetSomeAccount, SupplyUpdate, TokenAccountUpdate, @@ -754,7 +754,7 @@ pub trait SurfnetCheatcodes { /// - `absoluteEpoch(u64)`: Advances time to the specified epoch (each epoch = 432,000 slots). /// /// ## Returns - /// An `EpochInfo` object reflecting the updated clock state. + /// The updated epoch state and outcomes for overrides materialized at the target slot. /// /// ## Example Request /// ```json @@ -776,7 +776,8 @@ pub trait SurfnetCheatcodes { /// "slotsInEpoch": 432000, /// "absoluteSlot": 221184000, /// "blockHeight": 650000000, - /// "transactionCount": 923472834 + /// "transactionCount": 923472834, + /// "overrideOutcomes": [] /// }, /// "id": 1 /// } @@ -785,7 +786,7 @@ pub trait SurfnetCheatcodes { &self, meta: Self::Metadata, config: Option, - ) -> Result; + ) -> Result; /// A cheat code to freeze the Surfnet clock on the local network. /// All time progression halts until resumed. @@ -1421,13 +1422,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 @@ -1505,7 +1506,13 @@ pub trait SurfnetCheatcodes { /// "slot": 355684457, /// "apiVersion": "2.2.2" /// }, - /// "value": null, + /// "value": [ + /// { + /// "overrideId": "override-1", + /// "label": "Set BTC price", + /// "applied": true + /// } + /// ], /// "id": 1 /// } /// ``` @@ -1515,7 +1522,7 @@ pub trait SurfnetCheatcodes { meta: Self::Metadata, scenario: Scenario, slot: Option, - ) -> BoxFuture>>; + ) -> BoxFuture>>>; } #[derive(Clone)] @@ -2236,15 +2243,22 @@ impl SurfnetCheatcodes for SurfnetCheatcodesRpc { &self, meta: Self::Metadata, config: Option, - ) -> Result { + ) -> Result { let key = meta.as_ref().map(|ctx| ctx.id.clone()).unwrap_or_default(); let time_travel_config = config.unwrap_or_default(); let simnet_command_tx = meta.get_surfnet_command_tx()?; let svm_locker = meta.get_svm_locker()?; - let epoch_info = svm_locker.time_travel(key, simnet_command_tx, time_travel_config)?; + let (epoch_info, outcomes) = svm_locker.time_travel_with_override_outcomes( + key, + simnet_command_tx, + time_travel_config, + )?; - Ok(epoch_info) + Ok(TimeTravelResult { + epoch_info, + override_outcomes: outcomes, + }) } fn reset_account( @@ -2541,7 +2555,7 @@ impl SurfnetCheatcodes for SurfnetCheatcodesRpc { meta: Self::Metadata, scenario: Scenario, slot: Option, - ) -> BoxFuture>> { + ) -> BoxFuture>>> { let SurfnetRpcContext { svm_locker, remote_ctx, @@ -2551,32 +2565,14 @@ 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, }) }) } diff --git a/crates/core/src/runloops/mod.rs b/crates/core/src/runloops/mod.rs index 46d3eda6..7e6ce1ca 100644 --- a/crates/core/src/runloops/mod.rs +++ b/crates/core/src/runloops/mod.rs @@ -476,27 +476,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 461290c5..48316f98 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 2af4c244..1ed4366a 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 aa695fbe..6d6d9b5e 100644 --- a/crates/core/src/storage/mod.rs +++ b/crates/core/src/storage/mod.rs @@ -65,6 +65,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), } @@ -168,6 +173,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() @@ -176,6 +191,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 ee956c83..60ee3784 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 28d49a7e..7ef492c7 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, StorageError, StorageResult, + CrossTableRemove, Storage, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -116,6 +116,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!( @@ -202,6 +220,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)?; @@ -358,3 +459,41 @@ where Ok(Box::new(iter)) } } +#[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 backend = PostgresBackend::open(&database_url, &surfnet_id).unwrap(); + let mut accounts: PostgresStorage = backend.open_store("accounts").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 11774c49..bbd9bb81 100644 --- a/crates/core/src/storage/sqlite.rs +++ b/crates/core/src/storage/sqlite.rs @@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use log::debug; use serde::{Deserialize, Serialize}; use surfpool_db::diesel::{ - self, RunQueryDsl, SqliteConnection, + self, Connection, RunQueryDsl, SqliteConnection, connection::SimpleConnection, r2d2::{ConnectionManager, Pool}, sql_query, @@ -11,7 +11,7 @@ use surfpool_db::diesel::{ }; use crate::storage::{ - Storage, StorageError, StorageResult, + CrossTableRemove, Storage, StorageError, StorageOperation, StorageResult, diesel_common::{ CountRecord, KeyRecord, KvRecord, ValueRecord, deserialize_value, serialize_key, serialize_value, @@ -192,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!( @@ -275,6 +293,88 @@ 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<()> { + 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)?; @@ -586,4 +686,109 @@ mod tests { "stores on one backend should see each other's writes" ); } + + #[test] + fn test_apply_batch_rolls_back_every_operation_when_one_fails() { + let backend = SqliteBackend::open(":memory:", "surfnet1").unwrap(); + let mut storage: SqliteStorage = backend.open_store("batch_test").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 backend = SqliteBackend::open(database_url, "surfnet1").unwrap(); + let mut accounts: SqliteStorage = backend.open_store("accounts").unwrap(); + let mut scheduled_overrides: SqliteStorage = + backend.open_store("scheduled_overrides").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_memory_sqlite() { + assert_atomic_cross_table_rollback(":memory:"); + } + + #[test] + fn test_memory_database_survives_one_storage_drop() { + let backend = SqliteBackend::open(":memory:", "surfnet1").unwrap(); + let accounts: SqliteStorage = backend.open_store("accounts").unwrap(); + let mut scheduled_overrides: SqliteStorage = + backend.open_store("scheduled_overrides").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 = + backend.open_store("accounts").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 fcdb7dd7..e4f9f908 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; @@ -2681,11 +2681,11 @@ impl SurfnetSvmLocker { }; epoch_info.transaction_count = None; - self.with_svm_writer(move |svm_writer| { - let _ = svm_writer.reset_network(epoch_info, epoch_schedule); - let _ = svm_writer.offline_accounts.clear(); - }); - Ok(()) + self.with_svm_writer(move |svm_writer| -> SurfpoolResult<()> { + svm_writer.reset_network(epoch_info, epoch_schedule)?; + svm_writer.offline_accounts.clear()?; + Ok(()) + }) } /// Marks an account as offline, preventing it from being downloaded from the remote RPC. @@ -2795,21 +2795,24 @@ impl SurfnetSvmLocker { self.with_svm_reader(Self::offline_account_owners) } - /// 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 } - /// 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) @@ -3902,6 +3905,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(), @@ -3912,7 +3925,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()) @@ -3923,17 +3936,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!( @@ -3941,7 +3959,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. @@ -3993,7 +4011,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. @@ -4617,9 +4638,12 @@ mod tests { Arc, atomic::{AtomicUsize, Ordering}, }, + thread, + time::{Duration, Instant}, }; use async_trait::async_trait; + use crossbeam_channel::unbounded; use solana_account::Account; use solana_account_decoder::UiAccountEncoding; use solana_client::{ @@ -4646,6 +4670,82 @@ mod tests { }, }; + #[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 0e65972c..7a821c29 100644 --- a/crates/core/src/surfnet/svm.rs +++ b/crates/core/src/surfnet/svm.rs @@ -89,7 +89,7 @@ use crate::{ error::{AirdropError, SurfpoolError, SurfpoolResult}, rpc::utils::convert_transaction_metadata_from_canonical, scenarios::TemplateRegistry, - storage::{OverlayStorage, Storage, StorageBackend}, + storage::{CrossTableRemove, OverlayStorage, Storage, StorageBackend, StorageOperation}, surfnet::{ LogsSubscriptionData, locker::is_supported_token_program, surfnet_lite_svm::SurfnetLiteSvm, }, @@ -487,6 +487,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 @@ -913,7 +966,6 @@ 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; - // 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 @@ -2683,7 +2735,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) @@ -2695,22 +2747,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; } @@ -2736,6 +2838,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; } }; @@ -2745,8 +2852,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", @@ -2782,6 +2892,7 @@ impl SurfnetSvm { "Failed to fetch account {} from remote: {}", account_pubkey, e ); + fetch_error = Some(format!("fetchBeforeUse fetch failed: {e}")); None } }; @@ -2819,11 +2930,16 @@ impl SurfnetSvm { } // 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}")); + } } } } else { @@ -2832,6 +2948,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 @@ -2850,6 +2974,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; } @@ -2867,6 +2992,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; }; @@ -2898,6 +3028,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) => { @@ -2905,6 +3040,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; } }; @@ -2915,6 +3055,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; }; @@ -2932,6 +3077,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; } @@ -2949,6 +3099,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; } }; @@ -2968,6 +3123,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 {})", @@ -2975,11 +3135,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 @@ -4244,17 +4408,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 {}", @@ -4264,28 +4517,100 @@ 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 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}" + ))); } - Ok(()) + 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"), + )); + } + } } } @@ -4301,7 +4626,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; @@ -4312,11 +4637,758 @@ 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_helper_preserves_boundaries(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 + ); + } + + #[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() { @@ -4450,7 +5522,8 @@ mod tests { let target = Pubkey::new_unique(); locker - .register_scenario(fetch_before_use_scenario(target), Some(100)) + .register_scenario_and_materialize(&None, fetch_before_use_scenario(target), Some(100)) + .await .unwrap(); locker .materialize_overrides_for_slot(&Some(remote), 100) @@ -4498,7 +5571,8 @@ mod tests { }); locker - .register_scenario(fetch_before_use_scenario(target), Some(100)) + .register_scenario_and_materialize(&None, fetch_before_use_scenario(target), Some(100)) + .await .unwrap(); locker .materialize_overrides_for_slot(&Some(remote), 100) @@ -5440,6 +6514,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 f4e9a537..41e8d054 100644 --- a/crates/core/src/tests/integration.rs +++ b/crates/core/src/tests/integration.rs @@ -1,4 +1,5 @@ use std::{ + collections::HashMap, io::{Read, Write}, net::TcpListener, str::FromStr, @@ -74,11 +75,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, RpcProfileDepth, RpcProfileResultConfig, Scenario, + SimnetCommand, SimnetEvent, StartupPlanner, SurfnetStartupPhase, SurfnetStartupStatus, + SurfnetStartupTask, SurfnetStartupTaskState, SurfpoolConfig, TimeTravelResult, UiAccountChange, + UiAccountProfileState, UiKeyedProfileResult, types::{ BlockProductionMode, RpcConfig, SimnetConfig, SubgraphConfig, TransactionStatusEvent, UuidOrSignature, @@ -4787,7 +4788,7 @@ fn test_time_travel_absolute_timestamp(test_type: TestType) { // Test time travel to absolute timestamp // Note: time_travel now uses confirmation mechanism, so it waits internally - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteTimestamp(target_timestamp)), ); @@ -4803,11 +4804,11 @@ fn test_time_travel_absolute_timestamp(test_type: TestType) { // Verify the epoch info reflects the time travel assert_ne!( - new_epoch_info.epoch, initial_epoch_info.epoch, + new_epoch_info.epoch_info.epoch, initial_epoch_info.epoch, "Epoch should change after time travel" ); assert_ne!( - new_epoch_info.absolute_slot, initial_epoch_info.absolute_slot, + new_epoch_info.epoch_info.absolute_slot, initial_epoch_info.absolute_slot, "Slot should change after time travel" ); @@ -4815,10 +4816,10 @@ fn test_time_travel_absolute_timestamp(test_type: TestType) { let current_epoch_info = svm_locker.get_epoch_info(); println!("Updated epoch info: {:?}", current_epoch_info); - assert_eq!(current_epoch_info.epoch, new_epoch_info.epoch); + assert_eq!(current_epoch_info.epoch, new_epoch_info.epoch_info.epoch); assert_eq!( current_epoch_info.absolute_slot, - new_epoch_info.absolute_slot + new_epoch_info.epoch_info.absolute_slot ); println!("Time travel to absolute timestamp test passed successfully!"); @@ -4870,7 +4871,7 @@ fn test_time_travel_absolute_slot(test_type: TestType) { // Test time travel to absolute slot // Note: time_travel now uses confirmation mechanism, so it waits internally - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteSlot(target_slot)), ); @@ -4885,26 +4886,98 @@ fn test_time_travel_absolute_slot(test_type: TestType) { // Verify the epoch info reflects the time travel assert_eq!( - new_epoch_info.absolute_slot, target_slot, + new_epoch_info.epoch_info.absolute_slot, target_slot, "Slot should match target slot" ); assert!( - new_epoch_info.epoch > initial_epoch_info.epoch, + new_epoch_info.epoch_info.epoch > initial_epoch_info.epoch, "Epoch should change after time travel" ); assert!( - new_epoch_info.absolute_slot > initial_epoch_info.absolute_slot, + new_epoch_info.epoch_info.absolute_slot > initial_epoch_info.absolute_slot, "Epoch should change after time travel" ); // Verify the current epoch info in SVM matches let current_epoch_info = svm_locker.get_epoch_info(); assert_eq!(current_epoch_info.absolute_slot, target_slot); - assert_eq!(current_epoch_info.epoch, new_epoch_info.epoch); + assert_eq!(current_epoch_info.epoch, new_epoch_info.epoch_info.epoch); 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 time_travel_result = rpc_server + .time_travel( + Some(runloop_context.clone()), + Some(TimeTravelConfig::AbsoluteSlot(base_slot + 450)), + ) + .unwrap(); + assert_eq!(time_travel_result.epoch_info.absolute_slot, base_slot + 450); + assert_eq!(time_travel_result.override_outcomes.len(), 1); + assert_eq!( + time_travel_result.override_outcomes[0].override_id, + "slot-450" + ); + assert!(time_travel_result.override_outcomes[0].applied); + + let epoch_info = rpc_server + .time_travel( + Some(runloop_context), + Some(TimeTravelConfig::AbsoluteSlot(base_slot + 750)), + ) + .unwrap(); + assert_eq!(epoch_info.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")] @@ -4951,7 +5024,7 @@ fn test_time_travel_absolute_epoch(test_type: TestType) { // Test time travel to absolute epoch // Note: time_travel now uses confirmation mechanism, so it waits internally - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteEpoch(target_epoch)), ); @@ -4966,15 +5039,15 @@ fn test_time_travel_absolute_epoch(test_type: TestType) { // Verify the epoch info reflects the time travel assert_eq!( - new_epoch_info.epoch, target_epoch, + new_epoch_info.epoch_info.epoch, target_epoch, "Epoch should match target epoch" ); assert_ne!( - new_epoch_info.epoch, initial_epoch_info.epoch, + new_epoch_info.epoch_info.epoch, initial_epoch_info.epoch, "Epoch should change after time travel" ); assert_ne!( - new_epoch_info.absolute_slot, initial_epoch_info.absolute_slot, + new_epoch_info.epoch_info.absolute_slot, initial_epoch_info.absolute_slot, "Slot should change after time travel" ); @@ -4983,7 +5056,7 @@ fn test_time_travel_absolute_epoch(test_type: TestType) { assert_eq!(current_epoch_info.epoch, target_epoch); assert_eq!( current_epoch_info.absolute_slot, - new_epoch_info.absolute_slot + new_epoch_info.epoch_info.absolute_slot ); println!("Time travel to absolute epoch test passed successfully!"); @@ -5554,6 +5627,22 @@ async fn test_reset_streamed_account_cascade(test_type: TestType) { .with_svm_writer(|svm_writer| { svm_writer.set_account(&owner, owner_account).unwrap(); svm_writer.set_account(&owned, owned_account).unwrap(); + svm_writer + .scheduled_overrides + .store( + svm_writer.latest_epoch_info.absolute_slot + 1, + vec![OverrideInstance { + id: "pending-reset".to_string(), + template_id: "test-template".to_string(), + values: HashMap::new(), + scenario_relative_slot: 1, + label: None, + enabled: true, + fetch_before_use: false, + account: AccountAddress::Pubkey(Pubkey::new_unique().to_string()), + }], + ) + .unwrap(); Ok::<(), SurfpoolError>(()) }) .unwrap(); @@ -5573,6 +5662,9 @@ async fn test_reset_streamed_account_cascade(test_type: TestType) { // Owner is deleted, owned account is deleted assert!(svm_locker.get_account_local(&owner).inner.is_none()); assert!(svm_locker.get_account_local(&owned).inner.is_none()); + svm_locker.with_svm_reader(|svm_reader| { + assert!(svm_reader.scheduled_overrides.keys().unwrap().is_empty()); + }); } #[test_case(TestType::sqlite(); "with on-disk sqlite db")] @@ -5865,7 +5957,7 @@ fn test_reset_network_time_travel_timestamp(test_type: TestType) { // First time travel to target timestamp // Note: time_travel now uses confirmation mechanism, so it waits internally - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteTimestamp(target_timestamp)), ); @@ -5881,7 +5973,7 @@ fn test_reset_network_time_travel_timestamp(test_type: TestType) { // Second time travel to the same timestamp should now succeed after reset // because updated_at was reset to current time - let time_travel_response2: JsonRpcResult = rpc_server.time_travel( + let time_travel_response2: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteTimestamp(target_timestamp)), ); @@ -5924,7 +6016,7 @@ fn test_reset_network_time_travel_slot(test_type: TestType) { // First time travel to target slot // Note: time_travel now uses confirmation mechanism, so it waits internally - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteSlot(target_slot)), ); @@ -5940,7 +6032,7 @@ fn test_reset_network_time_travel_slot(test_type: TestType) { // Second time travel to the same slot should now succeed after reset // because latest_epoch_info.absolute_slot was reset to the offline baseline - let time_travel_response2: JsonRpcResult = rpc_server.time_travel( + let time_travel_response2: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteSlot(target_slot)), ); @@ -5982,7 +6074,7 @@ fn test_reset_network_time_travel_epoch(test_type: TestType) { let target_epoch = 5; // First time travel to target epoch - let time_travel_response: JsonRpcResult = rpc_server.time_travel( + let time_travel_response: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteEpoch(target_epoch)), ); @@ -5999,7 +6091,7 @@ fn test_reset_network_time_travel_epoch(test_type: TestType) { // Second time travel to the same epoch should now succeed after reset // because latest_epoch_info.epoch was reset to 0 - let time_travel_response2: JsonRpcResult = rpc_server.time_travel( + let time_travel_response2: JsonRpcResult = rpc_server.time_travel( Some(runloop_context.clone()), Some(TimeTravelConfig::AbsoluteEpoch(target_epoch)), ); diff --git a/crates/core/src/tests/pump/mod.rs b/crates/core/src/tests/pump/mod.rs index 960ce8e2..ffc712b4 100644 --- a/crates/core/src/tests/pump/mod.rs +++ b/crates/core/src/tests/pump/mod.rs @@ -846,11 +846,13 @@ async fn test_pump_token2022_graduation_lifecycle() { let global_data = rpc.get_account(&PUMP_GLOBAL).await.unwrap().data; let fixture = GraduationFixture::new(user.pubkey(), mint, &curve_account.data, &global_data); + let base_slot = locker.get_latest_absolute_slot(); locker - .register_scenario(preparation.scenario, Some(0)) + .register_scenario_and_materialize(&None, preparation.scenario, Some(base_slot)) + .await .unwrap(); locker - .materialize_overrides_for_slot(&None, 1) + .materialize_overrides_for_slot(&None, base_slot + 1) .await .unwrap(); fund_user(&rpc, &fixture).await; @@ -1074,16 +1076,20 @@ async fn price_shock_changes_a_live_pool_swap() { let mut pool_override = OverrideInstance::new(template.id.clone(), 0, template.address.clone()) .with_values(values) .with_label("PumpSwap virtual quote reserve shock".to_string()); - pool_override.fetch_before_use = true; + pool_override.fetch_before_use = false; let mut price_shock = Scenario::new( "PumpSwap Price Shock".to_string(), "Shift a canonical PumpSwap pool price through its virtual quote reserves.".to_string(), ); price_shock.tags = vec!["pumpswap".to_string(), "price-shock".to_string()]; price_shock.add_override(pool_override); - locker.register_scenario(price_shock, Some(100)).unwrap(); + let base_slot = locker.get_latest_absolute_slot() + 1; + locker + .register_scenario_and_materialize(&None, price_shock, Some(base_slot)) + .await + .unwrap(); locker - .materialize_overrides_for_slot(&None, 100) + .materialize_overrides_for_slot(&None, base_slot) .await .unwrap(); diff --git a/crates/sdk-node/scripts/kit-unit.js b/crates/sdk-node/scripts/kit-unit.js index 6f868e8e..91a045ee 100644 --- a/crates/sdk-node/scripts/kit-unit.js +++ b/crates/sdk-node/scripts/kit-unit.js @@ -58,6 +58,9 @@ test("cheatcodes RPC passes bare (non-enveloped) results through untouched", asy slotIndex: 42, slotsInEpoch: 432000, transactionCount: 5, + overrideOutcomes: [ + { overrideId: "override-1", label: "Price shock", applied: true, reason: null }, + ], }, })); try { @@ -66,6 +69,8 @@ test("cheatcodes RPC passes bare (non-enveloped) results through untouched", asy // Kit's default transport parses all JSON integers as bigint. assert.equal(result.absoluteSlot, 42n); assert.equal(result.transactionCount, 5n); + assert.equal(result.overrideOutcomes[0].overrideId, "override-1"); + assert.equal(result.overrideOutcomes[0].applied, true); } finally { restore(); } 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 00000000..0f7c4ec8 --- /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 e610e03d..db2d9aa8 100644 --- a/crates/sdk-node/surfpool-sdk/kit/generated/index.ts +++ b/crates/sdk-node/surfpool-sdk/kit/generated/index.ts @@ -15,6 +15,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/types/api.ts b/crates/sdk-node/surfpool-sdk/kit/types/api.ts index 0c8de043..fe3d8f74 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, @@ -36,6 +37,10 @@ import type { */ export type EpochInfo = ReturnType; +export type TimeTravelResult = EpochInfo & { + overrideOutcomes: readonly OverrideOutcome[]; +}; + /** * Config for `timeTravel`. Mirrors `TimeTravelConfig` in * `crates/core/src/types.rs` (externally tagged, camelCase); its wire shape @@ -90,7 +95,7 @@ export type LocalSignatureEntry = Readonly<{ // Clock export type SurfnetTimeTravelApi = { - timeTravel(config?: TimeTravelConfig): EpochInfo; + timeTravel(config?: TimeTravelConfig): TimeTravelResult; }; export type SurfnetPauseClockApi = { pauseClock(): EpochInfo; @@ -182,9 +187,8 @@ export type SurfnetExportSnapshotApi = { // Scenario export type SurfnetRegisterScenarioApi = { - registerScenario(scenario: Scenario, slot?: number | bigint): null; + registerScenario(scenario: Scenario, slot?: number | bigint): readonly OverrideOutcome[]; }; - // Local export type SurfnetGetLocalSignaturesApi = { getLocalSignatures(limit?: number | bigint): readonly LocalSignatureEntry[]; diff --git a/crates/types/Cargo.toml b/crates/types/Cargo.toml index 779f815e..82feea74 100644 --- a/crates/types/Cargo.toml +++ b/crates/types/Cargo.toml @@ -30,7 +30,7 @@ solana-account = { workspace = true, features = ["serde"] } solana-account-decoder-client-types = { workspace = true } solana-clock = { workspace = true } solana-commitment-config = { workspace = true } -solana-epoch-info = { workspace = true } +solana-epoch-info = { workspace = true, features = ["serde"] } solana-message = { workspace = true } solana-pubkey = { workspace = true, features = ["curve25519"] } solana-rpc-client-api = { workspace = true } diff --git a/crates/types/src/rpc_endpoints.json b/crates/types/src/rpc_endpoints.json index d863d035..f668f9a8 100644 --- a/crates/types/src/rpc_endpoints.json +++ b/crates/types/src/rpc_endpoints.json @@ -812,7 +812,7 @@ }, { "method": "surfnet_timeTravel", - "description": "A cheat code to move the Surfnet clock forward. The clock only moves forward: a target in the past is rejected. Returns the epoch state after the jump.", + "description": "A cheat code to move the Surfnet clock forward. The clock only moves forward: a target in the past is rejected. Returns the epoch state and outcomes for overrides materialized at the target slot.", "params": [ { "name": "config", @@ -820,7 +820,7 @@ "description": "One of {\"absoluteTimestamp\": }, {\"absoluteSlot\": } or {\"absoluteEpoch\": }. Note that absoluteTimestamp is in MILLISECONDS, while the Clock sysvar reports seconds. If omitted, jumps one hour ahead." } ], - "returns": "An `EpochInfo` object describing the clock after the jump." + "returns": "An `EpochInfo`-compatible object describing the clock after the jump, extended with `overrideOutcomes`, the outcomes materialized while reaching the target slot." }, { "method": "surfnet_pauseClock", @@ -860,7 +860,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 +873,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. Later-slot outcomes are returned by surfnet_timeTravel as each target slot is reached." } ] }, diff --git a/crates/types/src/scenarios.rs b/crates/types/src/scenarios.rs index f328773e..1fad6189 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 c951dd8e..fb36a71d 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; @@ -39,6 +39,14 @@ pub const DEFAULT_SLOT_TIME_MS: u64 = 400; pub type Idl = anchor_lang_idl::types::Idl; pub const DEFAULT_PROFILING_MAP_CAPACITY: usize = 200; +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TimeTravelResult { + #[serde(flatten)] + pub epoch_info: EpochInfo, + pub override_outcomes: Vec, +} + #[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)] pub struct TransactionMetadata { pub signature: Signature, @@ -749,13 +757,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 @@ -1894,6 +1912,34 @@ mod tests { use super::*; + #[test] + fn time_travel_result_extends_epoch_info_without_breaking_epoch_info_clients() { + let epoch_info = EpochInfo { + epoch: 2, + slot_index: 3, + slots_in_epoch: 432_000, + absolute_slot: 864_003, + block_height: 800_000, + transaction_count: Some(42), + }; + let result = TimeTravelResult { + epoch_info: epoch_info.clone(), + override_outcomes: vec![OverrideOutcome { + override_id: "override-1".to_string(), + label: Some("Price shock".to_string()), + applied: true, + reason: None, + }], + }; + + let json = serde_json::to_value(result).unwrap(); + assert_eq!(json["absoluteSlot"], 864_003); + assert_eq!(json["overrideOutcomes"][0]["overrideId"], "override-1"); + + let legacy_epoch_info: EpochInfo = serde_json::from_value(json).unwrap(); + assert_eq!(legacy_epoch_info, epoch_info); + } + #[test] fn test_disable_cheatcode_with_lockout_allows_protected_methods() { // This test catches the bug where lockout was not propagated to