feat(fortuna): More metrics around revelation (#2726)

* feat(fortuna): More metrics around revelation

apps/fortuna/Cargo.lock

@@ -1647,7 +1647,7 @@ dependencies = [
 
 [[package]]
 name = "fortuna"
-version = "7.6.2"
+version = "7.6.3"
 dependencies = [
  "anyhow",
  "axum",

apps/fortuna/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fortuna"
-version = "7.6.2"
+version = "7.6.3"
 edition = "2021"
 
 [lib]

apps/fortuna/src/api.rs

@@ -2,7 +2,7 @@ use {
     crate::{
         chain::reader::{BlockNumber, BlockStatus, EntropyReader},
         history::History,
-        state::HashChainState,
+        state::MonitoredHashChainState,
     },
     anyhow::Result,
     axum::{
@@ -91,7 +91,7 @@ pub struct BlockchainState {
     /// Obtained from the response of eth_chainId rpc call
     pub network_id: u64,
     /// The hash chain(s) required to serve random numbers for this blockchain
-    pub state: Arc<HashChainState>,
+    pub state: Arc<MonitoredHashChainState>,
     /// The contract that the server is fulfilling requests for.
     pub contract: Arc<dyn EntropyReader>,
     /// The address of the provider that this server is operating for.
@@ -212,7 +212,7 @@ mod test {
             },
             chain::reader::{mock::MockEntropyReader, BlockStatus},
             history::History,
-            state::{HashChainState, PebbleHashChain},
+            state::{HashChainState, MonitoredHashChainState, PebbleHashChain},
         },
         axum::http::StatusCode,
         axum_test::{TestResponse, TestServer},
@@ -241,10 +241,17 @@ mod test {
     async fn test_server() -> (TestServer, Arc<MockEntropyReader>, Arc<MockEntropyReader>) {
         let eth_read = Arc::new(MockEntropyReader::with_requests(10, &[]));
 
+        let eth_state = MonitoredHashChainState::new(
+            ETH_CHAIN.clone(),
+            Default::default(),
+            "ethereum".into(),
+            PROVIDER,
+        );
+
         let eth_state = BlockchainState {
             id: "ethereum".into(),
             network_id: 1,
-            state: ETH_CHAIN.clone(),
+            state: Arc::new(eth_state),
             contract: eth_read.clone(),
             provider_address: PROVIDER,
             reveal_delay_blocks: 1,
@@ -255,10 +262,17 @@ mod test {
 
         let avax_read = Arc::new(MockEntropyReader::with_requests(10, &[]));
 
+        let avax_state = MonitoredHashChainState::new(
+            AVAX_CHAIN.clone(),
+            Default::default(),
+            "avalanche".into(),
+            PROVIDER,
+        );
+
         let avax_state = BlockchainState {
             id: "avalanche".into(),
             network_id: 43114,
-            state: AVAX_CHAIN.clone(),
+            state: Arc::new(avax_state),
             contract: avax_read.clone(),
             provider_address: PROVIDER,
             reveal_delay_blocks: 2,

apps/fortuna/src/chain/ethereum.rs

@@ -309,6 +309,7 @@ impl<T: JsonRpcClient + 'static> EntropyReader for PythRandom<Provider<T>> {
         let mut event = self.requested_with_callback_filter();
         event.filter = event
             .filter
+            .address(self.address())
             .from_block(from_block)
             .to_block(to_block)
             .topic1(provider);

apps/fortuna/src/command/generate.rs

@@ -48,6 +48,7 @@ pub async fn generate(opts: &GenerateOptions) -> Result<()> {
         let mut event = contract.revealed_with_callback_filter();
         event.filter = event
             .filter
+            .address(contract.address())
             .from_block(last_block_number)
             .to_block(current_block_number);
 

apps/fortuna/src/command/run.rs

@@ -7,7 +7,7 @@ use {
         eth_utils::traced_client::RpcMetrics,
         history::History,
         keeper::{self, keeper_metrics::KeeperMetrics},
-        state::{HashChainState, PebbleHashChain},
+        state::{HashChainState, MonitoredHashChainState, PebbleHashChain},
     },
     anyhow::{anyhow, Error, Result},
     axum::Router,
@@ -183,6 +183,7 @@ async fn setup_chain_and_run_keeper(
         chain_id,
         &chain_config,
         rpc_metrics.clone(),
+        keeper_metrics.clone(),
     )
     .await?;
     chains.write().await.insert(
@@ -210,6 +211,7 @@ async fn setup_chain_state(
     chain_id: &ChainId,
     chain_config: &EthereumConfig,
     rpc_metrics: Arc<RpcMetrics>,
+    keeper_metrics: Arc<KeeperMetrics>,
 ) -> Result<BlockchainState> {
     let contract = Arc::new(InstrumentedPythContract::from_config(
         chain_config,
@@ -284,10 +286,7 @@ async fn setup_chain_state(
         hash_chains.push(pebble_hash_chain);
     }
 
-    let chain_state = HashChainState {
-        offsets,
-        hash_chains,
-    };
+    let chain_state = HashChainState::new(offsets, hash_chains)?;
 
     if chain_state.reveal(provider_info.original_commitment_sequence_number)?
         != provider_info.original_commitment
@@ -297,9 +296,16 @@ async fn setup_chain_state(
         tracing::info!("Root of chain id {} matches commitment", &chain_id);
     }
 
+    let monitored_chain_state = MonitoredHashChainState::new(
+        Arc::new(chain_state),
+        keeper_metrics.clone(),
+        chain_id.clone(),
+        *provider,
+    );
+
     let state = BlockchainState {
         id: chain_id.clone(),
-        state: Arc::new(chain_state),
+        state: Arc::new(monitored_chain_state),
         network_id,
         contract,
         provider_address: *provider,

apps/fortuna/src/command/setup_provider.rs

@@ -122,12 +122,12 @@ async fn setup_chain_provider(
                 provider_config.chain_sample_interval,
             )
             .await?;
-            let chain_state = HashChainState {
-                offsets: vec![provider_info
+            let chain_state = HashChainState::new(
+                vec![provider_info
                     .original_commitment_sequence_number
                     .try_into()?],
-                hash_chains: vec![hash_chain],
-            };
+                vec![hash_chain],
+            )?;
 
             if chain_state.reveal(provider_info.original_commitment_sequence_number)?
                 != provider_info.original_commitment

apps/fortuna/src/keeper/keeper_metrics.rs

@@ -41,6 +41,7 @@ pub struct KeeperMetrics {
     pub final_gas_multiplier: Family<AccountLabel, Histogram>,
     pub final_fee_multiplier: Family<AccountLabel, Histogram>,
     pub gas_price_estimate: Family<AccountLabel, Gauge<f64, AtomicU64>>,
+    pub highest_revealed_sequence_number: Family<AccountLabel, Gauge>,
     pub accrued_pyth_fees: Family<ChainIdLabel, Gauge<f64, AtomicU64>>,
     pub block_timestamp_lag: Family<ChainIdLabel, Gauge>,
     pub latest_block_timestamp: Family<ChainIdLabel, Gauge>,
@@ -88,6 +89,7 @@ impl Default for KeeperMetrics {
                 Histogram::new(vec![100.0, 110.0, 120.0, 140.0, 160.0, 180.0, 200.0].into_iter())
             }),
             gas_price_estimate: Family::default(),
+            highest_revealed_sequence_number: Family::default(),
             accrued_pyth_fees: Family::default(),
             block_timestamp_lag: Family::default(),
             latest_block_timestamp: Family::default(),
@@ -223,6 +225,12 @@ impl KeeperMetrics {
             keeper_metrics.gas_price_estimate.clone(),
         );
 
+        writable_registry.register(
+            "highest_revealed_sequence_number",
+            "The highest sequence number revealed by the keeper either via callbacks or manual reveal",
+            keeper_metrics.highest_revealed_sequence_number.clone(),
+        );
+
         writable_registry.register(
             "accrued_pyth_fees",
             "Accrued Pyth fees on the contract",

apps/fortuna/src/state.rs

@@ -1,8 +1,12 @@
 use {
-    crate::api::ChainId,
+    crate::{
+        api::ChainId,
+        keeper::keeper_metrics::{AccountLabel, KeeperMetrics},
+    },
     anyhow::{ensure, Result},
     ethers::types::Address,
     sha3::{Digest, Keccak256},
+    std::sync::Arc,
     tokio::task::spawn_blocking,
 };
 
@@ -127,11 +131,22 @@ impl PebbleHashChain {
 /// which requires tracking multiple hash chains here.
 pub struct HashChainState {
     // The sequence number where the hash chain starts. Must be stored in sorted order.
-    pub offsets: Vec<usize>,
-    pub hash_chains: Vec<PebbleHashChain>,
+    offsets: Vec<usize>,
+    hash_chains: Vec<PebbleHashChain>,
 }
 
 impl HashChainState {
+    pub fn new(offsets: Vec<usize>, hash_chains: Vec<PebbleHashChain>) -> Result<HashChainState> {
+        if offsets.len() != hash_chains.len() {
+            return Err(anyhow::anyhow!(
+                "Offsets and hash chains must have the same length."
+            ));
+        }
+        Ok(HashChainState {
+            offsets,
+            hash_chains,
+        })
+    }
     pub fn from_chain_at_offset(offset: usize, chain: PebbleHashChain) -> HashChainState {
         HashChainState {
             offsets: vec![offset],
@@ -152,12 +167,54 @@ impl HashChainState {
     }
 }
 
+pub struct MonitoredHashChainState {
+    hash_chain_state: Arc<HashChainState>,
+    metrics: Arc<KeeperMetrics>,
+    account_label: AccountLabel,
+}
+impl MonitoredHashChainState {
+    pub fn new(
+        hash_chain_state: Arc<HashChainState>,
+        metrics: Arc<KeeperMetrics>,
+        chain_id: ChainId,
+        provider_address: Address,
+    ) -> Self {
+        Self {
+            hash_chain_state,
+            metrics,
+            account_label: AccountLabel {
+                chain_id,
+                address: provider_address.to_string(),
+            },
+        }
+    }
+
+    pub fn reveal(&self, sequence_number: u64) -> Result<[u8; 32]> {
+        let res = self.hash_chain_state.reveal(sequence_number);
+        if res.is_ok() {
+            let metric = self
+                .metrics
+                .highest_revealed_sequence_number
+                .get_or_create(&self.account_label);
+            if metric.get() < sequence_number as i64 {
+                metric.set(sequence_number as i64);
+            }
+        }
+        res
+    }
+}
+
 #[cfg(test)]
 mod test {
     use {
-        crate::state::{HashChainState, PebbleHashChain},
+        crate::{
+            keeper::keeper_metrics::{AccountLabel, KeeperMetrics},
+            state::{HashChainState, MonitoredHashChainState, PebbleHashChain},
+        },
         anyhow::Result,
+        ethers::types::Address,
         sha3::{Digest, Keccak256},
+        std::{sync::Arc, vec},
     };
 
     fn run_hash_chain_test(secret: [u8; 32], length: usize, sample_interval: usize) {
@@ -294,4 +351,65 @@ mod test {
 
         Ok(())
     }
+    #[test]
+    fn test_inconsistent_lengths() -> Result<()> {
+        let chain1 = PebbleHashChain::new([0u8; 32], 10, 1);
+        let chain2 = PebbleHashChain::new([1u8; 32], 10, 1);
+
+        let hash_chain_state = HashChainState::new(vec![5], vec![chain1.clone(), chain2.clone()]);
+        assert!(hash_chain_state.is_err());
+        let hash_chain_state = HashChainState::new(vec![5, 10], vec![chain1.clone()]);
+        assert!(hash_chain_state.is_err());
+        let hash_chain_state =
+            HashChainState::new(vec![5, 10], vec![chain1.clone(), chain2.clone()]);
+        assert!(hash_chain_state.is_ok());
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_highest_revealed_sequence_number() {
+        let chain = PebbleHashChain::new([0u8; 32], 100, 1);
+        let hash_chain_state = HashChainState::new(vec![0], vec![chain]).unwrap();
+        let metrics = Arc::new(KeeperMetrics::default());
+        let provider = Address::random();
+        let monitored = MonitoredHashChainState::new(
+            Arc::new(hash_chain_state),
+            metrics.clone(),
+            "ethereum".to_string(),
+            provider,
+        );
+        let label = AccountLabel {
+            chain_id: "ethereum".to_string(),
+            address: provider.to_string(),
+        };
+
+        assert!(monitored.reveal(5).is_ok());
+        let current = metrics
+            .highest_revealed_sequence_number
+            .get_or_create(&label)
+            .get();
+        assert_eq!(current, 5);
+
+        assert!(monitored.reveal(15).is_ok());
+        let current = metrics
+            .highest_revealed_sequence_number
+            .get_or_create(&label)
+            .get();
+        assert_eq!(current, 15);
+
+        assert!(monitored.reveal(10).is_ok());
+        let current = metrics
+            .highest_revealed_sequence_number
+            .get_or_create(&label)
+            .get();
+        assert_eq!(current, 15);
+
+        assert!(monitored.reveal(1000).is_err());
+        let current = metrics
+            .highest_revealed_sequence_number
+            .get_or_create(&label)
+            .get();
+        assert_eq!(current, 15);
+    }
 }