Merge pull request #1195 from eqlabs/pending_consistency

feat(sync): ensure pending data is consistent

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Fixed
+
+- pending data from the gateway is inconsistent
+  - this could exhibit as RPC data changing status between `pending | L2 accepted | not found`, especially noticeable for transactions.
+
 ## [0.6.4] - 2023-07-05
 
 ### Fixed

crates/common/src/state_update.rs

@@ -117,4 +117,62 @@ impl StateUpdate {
         self.declared_cairo_classes.insert(cairo);
         self
     }
+
+    /// The number of individual changes in this state update.
+    ///
+    /// The total amount of:
+    /// - system storage updates
+    /// - contract storage updates
+    /// - contract nonce updates
+    /// - contract deployments
+    /// - contract class replacements
+    /// - class declarations
+    pub fn change_count(&self) -> usize {
+        self.declared_cairo_classes.len()
+            + self.declared_sierra_classes.len()
+            + self
+                .system_contract_updates
+                .iter()
+                .map(|x| x.1.storage.len())
+                .sum::<usize>()
+            + self
+                .contract_updates
+                .iter()
+                .map(|x| {
+                    x.1.storage.len()
+                        + x.1.class.as_ref().map(|_| 1).unwrap_or_default()
+                        + x.1.nonce.as_ref().map(|_| 1).unwrap_or_default()
+                })
+                .sum::<usize>()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::felt;
+
+    use super::*;
+
+    #[test]
+    fn change_count() {
+        let state_update = StateUpdate::default()
+            .with_contract_nonce(ContractAddress(felt!("0x1")), ContractNonce(felt!("0x2")))
+            .with_contract_nonce(ContractAddress(felt!("0x4")), ContractNonce(felt!("0x5")))
+            .with_declared_cairo_class(ClassHash(felt!("0x3")))
+            .with_declared_sierra_class(SierraHash(felt!("0x4")), CasmHash(felt!("0x5")))
+            .with_deployed_contract(ContractAddress(felt!("0x1")), ClassHash(felt!("0x3")))
+            .with_replaced_class(ContractAddress(felt!("0x33")), ClassHash(felt!("0x35")))
+            .with_system_storage_update(
+                ContractAddress::ONE,
+                StorageAddress(felt!("0x10")),
+                StorageValue(felt!("0x99")),
+            )
+            .with_storage_update(
+                ContractAddress(felt!("0x33")),
+                StorageAddress(felt!("0x10")),
+                StorageValue(felt!("0x99")),
+            );
+
+        assert_eq!(state_update.change_count(), 8);
+    }
 }

crates/gateway-client/src/lib.rs

@@ -103,6 +103,118 @@ pub trait GatewayApi: Sync {
     }
 }
 
+#[async_trait::async_trait]
+impl<T: GatewayApi + Sync + Send> GatewayApi for std::sync::Arc<T> {
+    async fn block(&self, block: BlockId) -> Result<reply::MaybePendingBlock, SequencerError> {
+        self.as_ref().block(block).await
+    }
+
+    async fn block_without_retry(
+        &self,
+        block: BlockId,
+    ) -> Result<reply::MaybePendingBlock, SequencerError> {
+        self.as_ref().block_without_retry(block).await
+    }
+
+    async fn class_by_hash(&self, class_hash: ClassHash) -> Result<bytes::Bytes, SequencerError> {
+        self.as_ref().class_by_hash(class_hash).await
+    }
+
+    async fn pending_class_by_hash(
+        &self,
+        class_hash: ClassHash,
+    ) -> Result<bytes::Bytes, SequencerError> {
+        self.as_ref().pending_class_by_hash(class_hash).await
+    }
+
+    async fn transaction(
+        &self,
+        transaction_hash: TransactionHash,
+    ) -> Result<reply::Transaction, SequencerError> {
+        self.as_ref().transaction(transaction_hash).await
+    }
+
+    async fn state_update(&self, block: BlockId) -> Result<StateUpdate, SequencerError> {
+        self.as_ref().state_update(block).await
+    }
+
+    async fn eth_contract_addresses(&self) -> Result<reply::EthContractAddresses, SequencerError> {
+        self.as_ref().eth_contract_addresses().await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    async fn add_invoke_transaction(
+        &self,
+        version: TransactionVersion,
+        max_fee: Fee,
+        signature: Vec<TransactionSignatureElem>,
+        nonce: TransactionNonce,
+        contract_address: ContractAddress,
+        calldata: Vec<CallParam>,
+    ) -> Result<reply::add_transaction::InvokeResponse, SequencerError> {
+        self.as_ref()
+            .add_invoke_transaction(
+                version,
+                max_fee,
+                signature,
+                nonce,
+                contract_address,
+                calldata,
+            )
+            .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    async fn add_declare_transaction(
+        &self,
+        version: TransactionVersion,
+        max_fee: Fee,
+        signature: Vec<TransactionSignatureElem>,
+        nonce: TransactionNonce,
+        contract_definition: ContractDefinition,
+        sender_address: ContractAddress,
+        compiled_class_hash: Option<CasmHash>,
+        token: Option<String>,
+    ) -> Result<reply::add_transaction::DeclareResponse, SequencerError> {
+        self.as_ref()
+            .add_declare_transaction(
+                version,
+                max_fee,
+                signature,
+                nonce,
+                contract_definition,
+                sender_address,
+                compiled_class_hash,
+                token,
+            )
+            .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    async fn add_deploy_account(
+        &self,
+        version: TransactionVersion,
+        max_fee: Fee,
+        signature: Vec<TransactionSignatureElem>,
+        nonce: TransactionNonce,
+        contract_address_salt: ContractAddressSalt,
+        class_hash: ClassHash,
+        calldata: Vec<CallParam>,
+    ) -> Result<reply::add_transaction::DeployAccountResponse, SequencerError> {
+        self.as_ref()
+            .add_deploy_account(
+                version,
+                max_fee,
+                signature,
+                nonce,
+                contract_address_salt,
+                class_hash,
+                calldata,
+            )
+            .await
+    }
+}
+
 /// Starknet sequencer client using REST API.
 ///
 /// Retry is performed on __all__ types of errors __except for__

crates/pathfinder/src/state/sync/l2.rs

@@ -99,7 +99,7 @@ pub async fn sync<GatewayClient>(
     mut blocks: BlockChain,
 ) -> anyhow::Result<()>
 where
-    GatewayClient: GatewayApi,
+    GatewayClient: GatewayApi + Clone + Send + 'static,
 {
     let L2SyncContext {
         websocket_txs,
@@ -146,7 +146,7 @@ where
                                 .expect("Head hash should exist when entering pending mode");
                             (next_block, next_state_update) = pending::poll_pending(
                                 tx_event.clone(),
-                                &sequencer,
+                                sequencer.clone(),
                                 (head.1, head.2),
                                 interval,
                                 chain,
@@ -632,6 +632,7 @@ mod tests {
             sequencer: MockGatewayApi,
         ) -> JoinHandle<anyhow::Result<()>> {
             let storage = Storage::in_memory().unwrap();
+            let sequencer = std::sync::Arc::new(sequencer);
             let context = L2SyncContext {
                 websocket_txs: WebsocketSenders::for_test(),
                 sequencer,
@@ -983,6 +984,7 @@ mod tests {
                 );
 
                 // Let's run the UUT
+                let mock = std::sync::Arc::new(mock);
                 let context = L2SyncContext {
                     websocket_txs: WebsocketSenders::for_test(),
                     sequencer: mock,