Allow ethereum oracle control from ledger.

apps/src/lib/node/ledger/ethereum_oracle/mod.rs

@@ -243,6 +243,18 @@ impl<C: RpcClient> Oracle<C> {
             _ => None,
         }
     }
+
+    /// If the bridge has been deactivated, block here until a new
+    /// config is passed that reactivates the bridge
+    async fn wait_on_reactivation(&mut self) -> Config {
+        loop {
+            if let Some(Command::UpdateConfig(c)) = self.control.recv().await {
+                if c.active {
+                    return c;
+                }
+            }
+        }
+    }
 }
 
 /// Block until an initial configuration is received via the command channel.
@@ -394,6 +406,9 @@ async fn run_oracle_aux<C: RpcClient>(mut oracle: Oracle<C>) {
         if let Some(new_config) = oracle.update_config() {
             config = new_config;
         }
+        if !config.active {
+            config = oracle.wait_on_reactivation().await;
+        }
         next_block_to_process += 1.into();
     }
 }
@@ -981,6 +996,105 @@ mod test_oracle {
         oracle.await.expect("Test failed");
     }
 
+    /// Test that Ethereum oracle can be deactivate and reactivated
+    /// via config updates.
+    /// NOTE: This test can flake due to async channel race
+    /// conditions.
+    #[tokio::test]
+    async fn test_oracle_reactivation() {
+        let TestPackage {
+            oracle,
+            eth_recv,
+            controller,
+            mut blocks_processed_recv,
+            mut control_sender,
+        } = setup();
+        let config = Config::default();
+        let oracle = start_with_default_config(
+            oracle,
+            &mut control_sender,
+            config.clone(),
+        )
+        .await;
+
+        // set the height of the chain such that there are some blocks deep
+        // enough to be considered confirmed by the oracle
+        let confirmed_block_height = 9; // all blocks up to and including this block will have enough confirmations
+        let min_confirmations = u64::from(config.min_confirmations);
+        controller.apply_cmd(TestCmd::NewHeight(Uint256::from(
+            min_confirmations + confirmed_block_height - 3,
+        )));
+
+        // check that the oracle indeed processes the expected blocks
+        for height in 0u64..(confirmed_block_height - 4) {
+            let block_processed = timeout(
+                std::time::Duration::from_secs(3),
+                blocks_processed_recv.recv(),
+            )
+            .await
+            .expect("Timed out waiting for block to be checked")
+            .unwrap();
+            assert_eq!(block_processed, Uint256::from(height));
+        }
+
+        // Deactivate the bridge before all confirmed events are confirmed and
+        // processed There is a very fine needle to thread here. A block
+        // must be processed **after** this config is sent in order for
+        // the updated config to be received. However, this test can
+        // flake due to channel race conditions.
+        control_sender
+            .try_send(Command::UpdateConfig(Config {
+                active: false,
+                ..Default::default()
+            }))
+            .expect("Test failed");
+        std::thread::sleep(Duration::from_secs(1));
+        controller.apply_cmd(TestCmd::NewHeight(Uint256::from(
+            min_confirmations + confirmed_block_height - 4,
+        )));
+
+        let block_processed = timeout(
+            std::time::Duration::from_secs(3),
+            blocks_processed_recv.recv(),
+        )
+        .await
+        .expect("Timed out waiting for block to be checked")
+        .unwrap();
+        assert_eq!(block_processed, Uint256::from(confirmed_block_height - 4));
+
+        // check that the oracle hasn't yet checked any further blocks
+        // TODO: check this in a deterministic way rather than just waiting a
+        // bit
+        let res = timeout(
+            std::time::Duration::from_secs(3),
+            blocks_processed_recv.recv(),
+        )
+        .await;
+        assert!(res.is_err());
+
+        // reactivate the bridge and check that the oracle
+        // processed the rest of the confirmed blocks
+        control_sender
+            .try_send(Command::UpdateConfig(Default::default()))
+            .expect("Test failed");
+
+        controller.apply_cmd(TestCmd::NewHeight(Uint256::from(
+            min_confirmations + confirmed_block_height,
+        )));
+        for height in (confirmed_block_height - 3)..=confirmed_block_height {
+            let block_processed = timeout(
+                std::time::Duration::from_secs(3),
+                blocks_processed_recv.recv(),
+            )
+            .await
+            .expect("Timed out waiting for block to be checked")
+            .unwrap();
+            assert_eq!(block_processed, Uint256::from(height));
+        }
+        drop(eth_recv);
+        oracle.await.expect("Test failed");
+    }
+
     /// Test that if the Ethereum RPC endpoint returns a latest block that is
     /// more than one block later than the previous latest block we received, we
     /// still check all the blocks in between

apps/src/lib/node/ledger/shell/finalize_block.rs

@@ -136,6 +136,7 @@ where
 
         // Tracks the accepted transactions
         self.wl_storage.storage.block.results = BlockResults::default();
+        let mut changed_keys = BTreeSet::new();
         for (tx_index, processed_tx) in req.txs.iter().enumerate() {
             let tx = if let Ok(tx) = Tx::try_from(processed_tx.tx.as_ref()) {
                 tx
@@ -421,14 +422,15 @@ where
             )
             .map_err(Error::TxApply)
             {
-                Ok(result) => {
+                Ok(ref mut result) => {
                     if result.is_accepted() {
                         tracing::trace!(
                             "all VPs accepted transaction {} storage \
                              modification {:#?}",
                             tx_event["hash"],
                             result
                         );
+                        changed_keys.append(&mut result.changed_keys);
                         stats.increment_successful_txs();
                         self.wl_storage.commit_tx();
                         if !tx_event.contains_key("code") {
@@ -531,7 +533,7 @@ where
             self.update_epoch(&mut response);
             // send the latest oracle configs. These may have changed due to
             // governance.
-            self.update_eth_oracle();
+            self.update_eth_oracle(&changed_keys);
         }
 
         if !req.proposer_address.is_empty() {

apps/src/lib/node/ledger/shell/init_chain.rs

@@ -204,7 +204,7 @@ where
         if let Some(config) = genesis.ethereum_bridge_params {
             tracing::debug!("Initializing Ethereum bridge storage.");
             config.init_storage(&mut self.wl_storage);
-            self.update_eth_oracle();
+            self.update_eth_oracle(&Default::default());
         } else {
             self.wl_storage
                 .write_bytes(

apps/src/lib/node/ledger/shell/mod.rs

@@ -548,7 +548,7 @@ where
             // TODO: config event log params
             event_log: EventLog::default(),
         };
-        shell.update_eth_oracle();
+        shell.update_eth_oracle(&Default::default());
         shell
     }
 
@@ -928,14 +928,18 @@ where
     }
 
     /// If a handle to an Ethereum oracle was provided to the [`Shell`], attempt
-    /// to send it an updated configuration, using an initial configuration
+    /// to send it an updated configuration, using a configuration
     /// based on Ethereum bridge parameters in blockchain storage.
     ///
     /// This method must be safe to call even before ABCI `InitChain` has been
     /// called (i.e. when storage is empty), as we may want to do this check
     /// every time the shell starts up (including the first time ever at which
     /// time storage will be empty).
-    fn update_eth_oracle(&mut self) {
+    ///
+    /// This method is also called during `FinalizeBlock` to update the oracle
+    /// if relevant storage changes have occurred. This includes deactivating
+    /// and reactivating the bridge.
+    fn update_eth_oracle(&mut self, changed_keys: &BTreeSet<Key>) {
         if let ShellMode::Validator {
             eth_oracle: Some(EthereumOracleChannels { control_sender, .. }),
             ..
@@ -961,18 +965,32 @@ where
                 );
                 return;
             }
-            if !self.wl_storage.ethbridge_queries().is_bridge_active() {
-                tracing::info!(
-                    "Not starting oracle as the Ethereum bridge is disabled"
-                );
-                return;
-            }
             let Some(config) = EthereumBridgeConfig::read(&self.wl_storage) else {
                 tracing::info!(
                     "Not starting oracle as the Ethereum bridge config couldn't be found in storage"
                 );
                 return;
             };
+            let active =
+                if !self.wl_storage.ethbridge_queries().is_bridge_active() {
+                    if !changed_keys
+                        .contains(&eth_bridge::storage::active_key())
+                    {
+                        tracing::info!(
+                            "Not starting oracle as the Ethereum bridge is \
+                             disabled"
+                        );
+                        return;
+                    } else {
+                        tracing::info!(
+                            "Disabling oracle as the bridge has been disabled"
+                        );
+                        false
+                    }
+                } else {
+                    true
+                };
+
             let start_block = self
                 .wl_storage
                 .storage
@@ -998,6 +1016,7 @@ where
                 bridge_contract: config.contracts.bridge.address,
                 governance_contract: config.contracts.governance.address,
                 start_block,
+                active,
             };
             tracing::info!(
                 ?config,

ethereum_bridge/src/oracle/config.rs

@@ -16,6 +16,8 @@ pub struct Config {
     pub governance_contract: EthAddress,
     /// The earliest Ethereum block from which events may be processed.
     pub start_block: ethereum_structs::BlockHeight,
+    /// The status of the Ethereum bridge (active / inactive)
+    pub active: bool,
 }
 
 // TODO: this production Default implementation is temporary, there should be no
@@ -29,6 +31,7 @@ impl std::default::Default for Config {
             bridge_contract: EthAddress([0; 20]),
             governance_contract: EthAddress([1; 20]),
             start_block: 0.into(),
+            active: true,
         }
     }
 }

ethereum_bridge/src/storage/eth_bridge_queries.rs

@@ -37,14 +37,34 @@ pub enum SendValsetUpd {
     AtPrevHeight,
 }
 
-#[derive(Debug, Clone, BorshDeserialize, BorshSerialize)]
+#[derive(
+    Debug,
+    Clone,
+    BorshDeserialize,
+    BorshSerialize,
+    PartialEq,
+    Eq,
+    Hash,
+    Ord,
+    PartialOrd,
+)]
 /// An enum indicating if the Ethereum bridge is enabled.
 pub enum EthBridgeStatus {
     Disabled,
     Enabled(EthBridgeEnabled),
 }
 
-#[derive(Debug, Clone, BorshDeserialize, BorshSerialize)]
+#[derive(
+    Debug,
+    Clone,
+    BorshDeserialize,
+    BorshSerialize,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Hash,
+)]
 /// Enum indicating if the bridge was initialized at genesis
 /// or a later epoch.
 pub enum EthBridgeEnabled {