add old block detection

TODO.md

@@ -265,6 +265,7 @@ These are roughly in order of completition
 
 These are not yet ordered. There might be duplicates. We might not actually need all of these.
 
+- [ ] get `oldest_allowed` out of config. or calculate automatically based on block time.
 - [ ] `change_user_tier_by_address` script
 - [ ] figure out if "could not get block from params" is a problem worth logging
     - maybe it was an ots request?

web3_proxy/src/bin/web3_proxy_cli/cost_calculator.rs

@@ -61,7 +61,8 @@ pub struct CostCalculatorCommand {
     /// the chain id to check. If none, check all.
     #[argh(option)]
     chain_id: Option<u64>,
-    // TODO: start and end dates?
+    // TODO: query start and end dates
+    // TODO: goal price
 }
 
 impl CostCalculatorCommand {
@@ -161,8 +162,6 @@ impl CostCalculatorCommand {
 
         debug!("cost_seconds: {}", cost_seconds);
 
-        info!("$0.000005 = goal");
-
         let query_seconds: Decimal = query_response
             .last_period_datetime
             .signed_duration_since(query_response.first_period_datetime)
@@ -170,6 +169,8 @@ impl CostCalculatorCommand {
             .into();
         info!("query seconds: {}", query_seconds);
 
+        info!("$0.000005 = goal");
+
         let x = costs(
             query_response.total_frontend_requests,
             query_seconds,

web3_proxy/src/rpcs/blockchain.rs

@@ -7,6 +7,7 @@ use crate::{
     config::BlockAndRpc, jsonrpc::JsonRpcRequest, rpcs::synced_connections::SyncedConnections,
 };
 use anyhow::Context;
+use chrono::{DateTime, NaiveDateTime, Utc};
 use derive_more::From;
 use ethers::prelude::{Block, TxHash, H256, U64};
 use hashbrown::{HashMap, HashSet};
@@ -14,6 +15,7 @@ use log::{debug, warn, Level};
 use moka::future::Cache;
 use serde::Serialize;
 use serde_json::json;
+use std::time::{SystemTime, UNIX_EPOCH};
 use std::{cmp::Ordering, fmt::Display, sync::Arc};
 use tokio::sync::{broadcast, watch};
 use tokio::time::Duration;
@@ -264,12 +266,31 @@ impl Web3Connections {
                 // we don't know if its on the heaviest chain yet
                 self.save_block(&rpc_head_block, false).await?;
 
-                connection_heads.insert(rpc.name.to_owned(), rpc_head_hash);
+                let now = SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .context("no time")?;
 
-                Some(BlockId {
+                // TODO: get this from config
-                    hash: rpc_head_hash,
+                let oldest_allowed = now - Duration::from_secs(120);
-                    num: rpc_head_num,
+
-                })
+                let block_timestamp = Duration::from_secs(rpc_head_block.timestamp.as_u64());
+
+                if block_timestamp < oldest_allowed {
+                    let behind_secs = (oldest_allowed - block_timestamp).as_secs();
+
+                    warn!("rpc is behind by {} seconds", behind_secs);
+
+                    connection_heads.remove(&rpc.name);
+
+                    None
+                } else {
+                    connection_heads.insert(rpc.name.to_owned(), rpc_head_hash);
+
+                    Some(BlockId {
+                        hash: rpc_head_hash,
+                        num: rpc_head_num,
+                    })
+                }
             }
             None => {
                 // TODO: warn is too verbose. this is expected if a node disconnects and has to reconnect

web3_proxy/src/rpcs/connections.rs

@@ -847,10 +847,12 @@ impl Serialize for Web3Connections {
 
 mod tests {
     #![allow(unused_imports)]
+    use std::time::{SystemTime, UNIX_EPOCH};
+
     // TODO: why is this allow needed? does tokio::test get in the way somehow?
     use super::*;
     use crate::rpcs::{blockchain::BlockId, provider::Web3Provider};
-    use ethers::types::Block;
+    use ethers::types::{Block, U256};
     use log::{trace, LevelFilter};
     use parking_lot::RwLock;
 
@@ -863,9 +865,16 @@ mod tests {
             .is_test(true)
             .try_init();
 
+        let now: U256 = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs()
+            .into();
+
         let lagged_block = Block {
             hash: Some(H256::random()),
             number: Some(0.into()),
+            timestamp: now - 1,
             ..Default::default()
         };
 
@@ -873,6 +882,7 @@ mod tests {
             hash: Some(H256::random()),
             number: Some(1.into()),
             parent_hash: lagged_block.hash.unwrap(),
+            timestamp: now,
             ..Default::default()
         };
 
@@ -1093,10 +1103,17 @@ mod tests {
             .is_test(true)
             .try_init();
 
+        let now: U256 = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs()
+            .into();
+
         let head_block: Block<TxHash> = Block {
             hash: Some(H256::random()),
             number: Some(1_000_000.into()),
             parent_hash: H256::random(),
+            timestamp: now,
             ..Default::default()
         };