Add testing framework for events that update Account

.github/workflows/rust.yml

@@ -20,3 +20,13 @@ jobs:
           crate: taplo-cli
       - name: Run taplo linter
         run: taplo format --check --verbose
+
+  cargo-test-workspace:
+    name: test workspace crates
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install toolchain
+        uses: dsherret/rust-toolchain-file@v1
+      - name: Test Workspace
+        run: cargo test --workspace --features mock_storage

crates/actors/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 default = ["local"]
 local = []
 remote = []
+mock_storage = []
 
 [dependencies]
 ethereum-types = "0.14.1"
@@ -69,3 +70,4 @@ tracing = "0.1.40"
 
 [dev-dependencies]
 anyhow = "1"
+serial_test = "3.1.1"

crates/actors/src/account_cache.rs

@@ -5,7 +5,9 @@ use lasr_messages::{
     AccountCacheMessage, ActorName, ActorType, RpcMessage, RpcResponseError, SupervisorType,
     TransactionResponse,
 };
-use lasr_types::{Account, AccountType, Address};
+#[cfg(feature = "mock_storage")]
+use lasr_types::MockPersistenceStore;
+use lasr_types::{Account, AccountType, Address, PersistenceStore};
 use ractor::{
     concurrency::OneshotReceiver, Actor, ActorCell, ActorProcessingErr, ActorRef, SupervisionEvent,
 };
@@ -14,12 +16,15 @@ use std::{
     time::{Duration, Instant},
 };
 use thiserror::Error;
+#[cfg(not(feature = "mock_storage"))]
 use tikv_client::RawClient as TikvClient;
 use tokio::sync::mpsc::Sender;
 
 #[derive(Debug, Clone, Default)]
 pub struct AccountCacheActor;
 
+pub type StorageRef = <AccountCacheActor as Actor>::Arguments;
+
 impl ActorName for AccountCacheActor {
     fn name(&self) -> ractor::ActorName {
         ActorType::AccountCache.to_string()
@@ -44,15 +49,15 @@ impl Default for AccountCacheError {
     }
 }
 
-pub struct AccountCache {
+pub struct AccountCache<S: PersistenceStore> {
     inner: AccountCacheInner,
-    tikv_client: TikvClient,
+    storage: S,
 }
-impl AccountCache {
-    pub fn new(tikv_client: TikvClient) -> Self {
+impl<S: PersistenceStore> AccountCache<S> {
+    pub fn new(storage: S) -> Self {
         Self {
             inner: AccountCacheInner::new(),
-            tikv_client,
+            storage,
         }
     }
 }
@@ -192,8 +197,11 @@ impl AccountCacheActor {
 #[async_trait]
 impl Actor for AccountCacheActor {
     type Msg = AccountCacheMessage;
-    type State = AccountCache;
+    type State = AccountCache<Self::Arguments>;
+    #[cfg(not(feature = "mock_storage"))]
     type Arguments = TikvClient;
+    #[cfg(feature = "mock_storage")]
+    type Arguments = MockPersistenceStore<String, Vec<u8>>;
 
     async fn pre_start(
         &self,
@@ -243,9 +251,10 @@ impl Actor for AccountCacheActor {
                     let acc_key = address.to_full_string();
 
                     // Pull `Account` data from persistence store
-                    state
-                    .tikv_client
-                    .get(acc_key.to_owned())
+                    PersistenceStore::get(
+                        &state.storage,
+                        acc_key.to_owned().into()
+                    )
                     .await
                     .typecast()
                     .log_err(|e| AccountCacheError::Custom(format!("failed to find Account with address: {hex_address} in persistence store: {e:?}")))

crates/actors/src/batcher.rs

@@ -30,7 +30,6 @@ use serde_json::Value;
 use sha3::{Digest, Keccak256};
 use std::io::Write;
 use thiserror::Error;
-use tikv_client::RawClient as TikvClient;
 use tokio::{
     sync::{
         mpsc::{Receiver, Sender, UnboundedSender},
@@ -42,8 +41,8 @@ use web3::types::BlockNumber;
 
 use crate::{
     account_cache, get_account, get_actor_ref, handle_actor_response, process_group_changed,
-    AccountCacheError, ActorExt, Coerce, DaClientError, EoClientError, PendingTransactionError,
-    SchedulerError, StaticFuture, UnorderedFuturePool,
+    AccountCacheActor, AccountCacheError, ActorExt, Coerce, DaClientError, EoClientError,
+    PendingTransactionError, SchedulerError, StaticFuture, StorageRef, UnorderedFuturePool,
 };
 use lasr_messages::{
     AccountCacheMessage, ActorName, ActorType, BatcherMessage, DaClientMessage, EoMessage,
@@ -55,8 +54,9 @@ use lasr_contract::create_program_id;
 use lasr_types::{
     Account, AccountBuilder, AccountType, Address, AddressOrNamespace, ArbitraryData,
     BurnInstruction, ContractLogType, CreateInstruction, Instruction, Metadata, MetadataValue,
-    Namespace, Outputs, ProgramAccount, ProgramUpdate, TokenDistribution, TokenOrProgramUpdate,
-    TokenUpdate, Transaction, TransactionType, TransferInstruction, UpdateInstruction, U256,
+    Namespace, Outputs, PersistenceStore, ProgramAccount, ProgramUpdate, TokenDistribution,
+    TokenOrProgramUpdate, TokenUpdate, Transaction, TransactionType, TransferInstruction,
+    UpdateInstruction, U256,
 };
 
 use derive_builder::Builder;
@@ -442,7 +442,7 @@ impl Batcher {
         Ok(())
     }
 
-    pub(super) async fn add_transaction_to_account(
+    pub async fn add_transaction_to_account(
         batcher: Arc<Mutex<Batcher>>,
         transaction: Transaction,
     ) -> Result<(), BatcherError> {
@@ -1331,7 +1331,7 @@ impl Batcher {
         }
     }
 
-    async fn apply_program_registration(
+    pub async fn apply_program_registration(
         batcher: Arc<Mutex<Batcher>>,
         transaction: Transaction,
     ) -> Result<(), BatcherError> {
@@ -1496,7 +1496,7 @@ impl Batcher {
         }
     }
 
-    async fn apply_instructions_to_accounts(
+    pub async fn apply_instructions_to_accounts(
         batcher: Arc<Mutex<Batcher>>,
         transaction: Transaction,
         outputs: Outputs,
@@ -1701,7 +1701,7 @@ impl Batcher {
 
     async fn handle_next_batch_request(
         batcher: Arc<Mutex<Batcher>>,
-        tikv_client: TikvClient,
+        storage_ref: StorageRef,
     ) -> Result<(), BatcherError> {
         if let Some(blob_response) = {
             let mut guard = batcher.lock().await;
@@ -1720,7 +1720,10 @@ impl Batcher {
                         let acc_val = AccountValue { account: data };
                         // Serialize `Account` data to be stored.
                         if let Some(val) = bincode::serialize(&acc_val).ok() {
-                            if tikv_client.put(addr.clone(), val).await.is_ok() {
+                            if PersistenceStore::put(&storage_ref, addr.clone().into(), val)
+                                .await
+                                .is_ok()
+                            {
                                 tracing::warn!(
                                     "Inserted Account with address of {addr:?} to persistence layer",
                                 )
@@ -1903,8 +1906,8 @@ impl Actor for BatcherActor {
     ) -> Result<(), ActorProcessingErr> {
         let batcher_ptr = Arc::clone(state);
         match message {
-            BatcherMessage::GetNextBatch { tikv_client } => {
-                Batcher::handle_next_batch_request(batcher_ptr, tikv_client).await?;
+            BatcherMessage::GetNextBatch { storage_ref } => {
+                Batcher::handle_next_batch_request(batcher_ptr, storage_ref).await?;
                 // let mut guard = self.future_pool.lock().await;
                 // guard.push(fut.boxed());
             }
@@ -2053,7 +2056,7 @@ impl Actor for BatcherSupervisor {
 
 pub async fn batch_requestor(
     mut stopper: tokio::sync::mpsc::Receiver<u8>,
-    tikv_client: TikvClient,
+    storage_ref: StorageRef,
 ) {
     if let Some(batcher) = ractor::registry::where_is(ActorType::Batcher.to_string()) {
         let batcher: ActorRef<BatcherMessage> = batcher.into();
@@ -2065,7 +2068,7 @@ pub async fn batch_requestor(
             tracing::info!("SLEEPING THEN REQUESTING NEXT BATCH");
             tokio::time::sleep(tokio::time::Duration::from_secs(batch_interval_secs)).await;
             let message = BatcherMessage::GetNextBatch {
-                tikv_client: tikv_client.clone(),
+                storage_ref: storage_ref.clone(),
             };
             tracing::warn!("requesting next batch");
             if let Err(err) = batcher.cast(message) {
@@ -2090,7 +2093,6 @@ mod batcher_tests {
     use futures::{FutureExt, StreamExt};
     use lasr_types::TransactionType;
     use std::sync::Arc;
-    use tikv_client::RawClient as TikvClient;
     use tokio::sync::Mutex;
 
     /// Minimal reproduction of the `ractor::Actor` trait for testing the `handle`
@@ -2107,8 +2109,8 @@ mod batcher_tests {
         async fn handle(&self, message: Self::Msg, state: &mut Self::State) -> Result<()> {
             let batcher_ptr = Arc::clone(state);
             match message {
-                BatcherMessage::GetNextBatch { tikv_client } => {
-                    let fut = Batcher::handle_next_batch_request(batcher_ptr, tikv_client);
+                BatcherMessage::GetNextBatch { storage_ref } => {
+                    let fut = Batcher::handle_next_batch_request(batcher_ptr, storage_ref);
                     let mut guard = self.future_pool.lock().await;
                     guard.push(fut.boxed());
                 }