fix: prevent stale reads (#62)

client/src/client.rs

@@ -152,7 +152,7 @@ impl<DB: Database> Client<DB> {
         self.node.read().await.chain_id()
     }
 
-    pub async fn get_header(&self) -> Header {
-        self.node.read().await.get_header().clone()
+    pub async fn get_header(&self) -> Result<Header> {
+        self.node.read().await.get_header()
     }
 }

client/src/node.rs

@@ -96,36 +96,48 @@ impl Node {
     }
 
     pub fn call(&self, opts: &CallOpts, block: &BlockTag) -> Result<Vec<u8>> {
+        self.check_blocktag_age(block)?;
+
         let payload = self.get_payload(block)?;
         let mut evm = Evm::new(self.execution.clone(), payload.clone(), self.chain_id());
         evm.call(opts)
     }
 
     pub fn estimate_gas(&self, opts: &CallOpts) -> Result<u64> {
+        self.check_head_age()?;
+
         let payload = self.get_payload(&BlockTag::Latest)?;
         let mut evm = Evm::new(self.execution.clone(), payload.clone(), self.chain_id());
         evm.estimate_gas(opts)
     }
 
     pub async fn get_balance(&self, address: &Address, block: &BlockTag) -> Result<U256> {
+        self.check_blocktag_age(block)?;
+
         let payload = self.get_payload(block)?;
         let account = self.execution.get_account(&address, None, payload).await?;
         Ok(account.balance)
     }
 
     pub async fn get_nonce(&self, address: &Address, block: &BlockTag) -> Result<u64> {
+        self.check_blocktag_age(block)?;
+
         let payload = self.get_payload(block)?;
         let account = self.execution.get_account(&address, None, payload).await?;
         Ok(account.nonce)
     }
 
     pub async fn get_code(&self, address: &Address, block: &BlockTag) -> Result<Vec<u8>> {
+        self.check_blocktag_age(block)?;
+
         let payload = self.get_payload(block)?;
         let account = self.execution.get_account(&address, None, payload).await?;
         Ok(account.code)
     }
 
     pub async fn get_storage_at(&self, address: &Address, slot: H256) -> Result<U256> {
+        self.check_head_age()?;
+
         let payload = self.get_payload(&BlockTag::Latest)?;
         let account = self
             .execution
@@ -159,6 +171,8 @@ impl Node {
     }
 
     pub fn get_gas_price(&self) -> Result<U256> {
+        self.check_head_age()?;
+
         let payload = self.get_payload(&BlockTag::Latest)?;
         let base_fee = U256::from_little_endian(&payload.base_fee_per_gas.to_bytes_le());
         let tip = U256::from(10_u64.pow(9));
@@ -171,11 +185,15 @@ impl Node {
     }
 
     pub fn get_block_number(&self) -> Result<u64> {
+        self.check_head_age()?;
+
         let payload = self.get_payload(&BlockTag::Latest)?;
         Ok(payload.block_number)
     }
 
     pub fn get_block_by_number(&self, block: &BlockTag) -> Result<Option<ExecutionBlock>> {
+        self.check_blocktag_age(block)?;
+
         match self.get_payload(block) {
             Ok(payload) => self.execution.get_block(payload).map(|b| Some(b)),
             Err(_) => Ok(None),
@@ -199,8 +217,9 @@ impl Node {
         self.config.general.chain_id
     }
 
-    pub fn get_header(&self) -> &Header {
-        self.consensus.get_header()
+    pub fn get_header(&self) -> Result<Header> {
+        self.check_head_age()?;
+        Ok(self.consensus.get_header().clone())
     }
 
     pub fn get_last_checkpoint(&self) -> Option<Vec<u8>> {
@@ -223,6 +242,26 @@ impl Node {
             }
         }
     }
+
+    fn check_head_age(&self) -> Result<()> {
+        let synced_slot = self.consensus.get_header().slot;
+        let expected_slot = self.consensus.expected_current_slot();
+        let slot_delay = expected_slot - synced_slot;
+
+        if slot_delay > 10 {
+            return Err(eyre!("Out of Sync"));
+        }
+
+        Ok(())
+    }
+
+    fn check_blocktag_age(&self, block: &BlockTag) -> Result<()> {
+        match block {
+            BlockTag::Latest => self.check_head_age(),
+            BlockTag::Finalized => Ok(()),
+            BlockTag::Number(_) => Ok(()),
+        }
+    }
 }
 
 pub enum BlockTag {

consensus/src/consensus.rs

@@ -160,7 +160,7 @@ impl<R: Rpc> ConsensusClient<R> {
         }
 
         let update_finalized_slot = update.finalized_header.clone().unwrap_or_default().slot;
-        let valid_time = self.current_slot() >= update.signature_slot
+        let valid_time = self.expected_current_slot() >= update.signature_slot
             && update.signature_slot > update.attested_header.slot
             && update.attested_header.slot >= update_finalized_slot;
 
@@ -410,7 +410,7 @@ impl<R: Rpc> ConsensusClient<R> {
         chrono::Duration::from_std(delay).unwrap()
     }
 
-    fn current_slot(&self) -> u64 {
+    pub fn expected_current_slot(&self) -> u64 {
         let now = std::time::SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .unwrap();
@@ -428,7 +428,7 @@ impl<R: Rpc> ConsensusClient<R> {
     /// Gets the duration until the next update
     /// Updates are scheduled for 4 seconds into each slot
     pub fn duration_until_next_update(&self) -> Duration {
-        let current_slot = self.current_slot();
+        let current_slot = self.expected_current_slot();
         let next_slot = current_slot + 1;
         let next_slot_timestamp = self.slot_timestamp(next_slot);