refactor: fix the annoying process bug. misc rust updates

rust/Cargo.lock

@@ -1930,6 +1930,7 @@ dependencies = [
  "async-trait",
  "color-eyre",
  "ethers",
+ "hex",
  "optics-core",
  "serde 1.0.125",
  "serde_json",

rust/optics-base/src/replica.rs

@@ -106,7 +106,7 @@ impl Replica for Replicas {
         }
     }
 
-    async fn next_to_process(&self) -> Result<U256, ChainCommunicationError> {
+    async fn next_to_process(&self) -> Result<u32, ChainCommunicationError> {
         match self {
             Replicas::Ethereum(replica) => replica.next_to_process().await,
             Replicas::Mock(mock_replica) => mock_replica.next_to_process().await,

rust/optics-core/src/traits/replica.rs

@@ -40,7 +40,7 @@ pub trait Replica: Common + Send + Sync + std::fmt::Debug {
     async fn previous_root(&self) -> Result<H256, ChainCommunicationError>;
 
     /// Fetch the last processed sequence number
-    async fn next_to_process(&self) -> Result<U256, ChainCommunicationError>;
+    async fn next_to_process(&self) -> Result<u32, ChainCommunicationError>;
 
     /// Dispatch a transaction to prove inclusion of some leaf in the replica.
     async fn prove(&self, proof: &Proof) -> Result<TxOutcome, ChainCommunicationError>;

rust/optics-ethereum/Cargo.toml

@@ -16,3 +16,5 @@ tracing = "0.1.22"
 color-eyre = "0.5.0"
 
 optics-core = { path = "../optics-core" }
+
+hex = "*"

rust/optics-ethereum/src/replica.rs

@@ -7,6 +7,7 @@ use optics_core::{
     traits::{ChainCommunicationError, Common, DoubleUpdate, Replica, State, TxOutcome},
     Encode, OpticsMessage, SignedUpdate, Update,
 };
+use tracing::info;
 
 use std::{convert::TryFrom, error::Error as StdError, sync::Arc};
 
@@ -217,7 +218,7 @@ where
     }
 
     #[tracing::instrument(err)]
-    async fn next_to_process(&self) -> Result<U256, ChainCommunicationError> {
+    async fn next_to_process(&self) -> Result<u32, ChainCommunicationError> {
         Ok(self.contract.next_to_process().call().await?)
     }
 
@@ -238,6 +239,7 @@ where
 
     #[tracing::instrument(err)]
     async fn process(&self, message: &OpticsMessage) -> Result<TxOutcome, ChainCommunicationError> {
+        info!("{}", hex::encode(message.to_vec()));
         let tx = self.contract.process(message.to_vec());
         Ok(report_tx!(tx).into())
     }

rust/optics-test/src/mocks/replica.rs

@@ -26,7 +26,7 @@ mock! {
 
         pub fn _previous_root(&self) -> Result<H256, ChainCommunicationError> {}
 
-        pub fn _next_to_process(&self) -> Result<U256, ChainCommunicationError> {}
+        pub fn _next_to_process(&self) -> Result<u32, ChainCommunicationError> {}
 
         pub fn _prove(&self, proof: &Proof) -> Result<TxOutcome, ChainCommunicationError> {}
 
@@ -104,7 +104,7 @@ impl Replica for MockReplicaContract {
         self._previous_root()
     }
 
-    async fn next_to_process(&self) -> Result<U256, ChainCommunicationError> {
+    async fn next_to_process(&self) -> Result<u32, ChainCommunicationError> {
         self._next_to_process()
     }
 

rust/processor/src/processor.rs

@@ -19,7 +19,10 @@ use optics_base::{
     replica::Replicas,
     reset_loop_if,
 };
-use optics_core::traits::{Home, Replica};
+use optics_core::{
+    accumulator::merkle::Proof,
+    traits::{CommittedMessage, Common, Home, MessageStatus, Replica},
+};
 
 use crate::{
     prover::{Prover, ProverSync},
@@ -62,14 +65,18 @@ impl ReplicaProcessor {
             //      - If not, wait and poll again
             // 4. Submit the proof to the replica
             loop {
-                let next_to_process = self.replica.next_to_process().await?;
-                let sequence = next_to_process.as_u32();
+                let sequence = self.replica.next_to_process().await?;
+                info!(
+                    "Next to process for replica {} is {}",
+                    self.replica.name(),
+                    sequence
+                );
 
                 let message = self.home.message_by_sequence(domain, sequence).await?;
                 reset_loop_if!(
                     message.is_none(),
                     interval,
-                    "Remote does not contain message at {}:{}",
+                    "Home does not contain message at {}:{}",
                     domain,
                     sequence
                 );
@@ -86,24 +93,42 @@ impl ReplicaProcessor {
                 );
 
                 let proof = proof_res.unwrap();
-                if proof.leaf != message.message.to_leaf() {
-                    let err = format!("Leaf in prover does not match retrieved message. Index: {}. Retrieved: {}. Local: {}.", message.leaf_index, message.message.to_leaf(), proof.leaf);
+                if proof.leaf != message.to_leaf() {
+                    let err = format!("Leaf in prover does not match retrieved message. Index: {}. Calculated: {}. Prover: {}.", message.leaf_index, message.to_leaf(), proof.leaf);
                     tracing::error!("{}", err);
                     color_eyre::eyre::bail!(err);
                 }
 
+                // Dispatch for processing
                 info!(
                     "Dispatching a message for processing {}:{}",
                     domain, sequence
                 );
-                self.replica
-                    .prove_and_process(message.as_ref(), &proof)
-                    .await?;
+                self.process(message, proof).await?;
 
                 sleep(std::time::Duration::from_secs(interval)).await;
             }
         })
     }
+
+    /// Dispatch a message for processing. If the message is already proven, process only.
+    async fn process(&self, message: CommittedMessage, proof: Proof) -> Result<()> {
+        let status = self.replica.message_status(message.to_leaf()).await?;
+
+        match status {
+            MessageStatus::None => {
+                self.replica
+                    .prove_and_process(message.as_ref(), &proof)
+                    .await?;
+            }
+            MessageStatus::Pending => {
+                self.replica.process(message.as_ref()).await?;
+            }
+            MessageStatus::Processed => {} // Indicates race condition?
+        }
+
+        Ok(())
+    }
 }
 
 decl_agent!(

solidity/optics-core/contracts/Replica.sol

@@ -209,9 +209,6 @@ contract Replica is Initializable, Common, QueueManager {
         // fail.
         messages[_m.keccak()] = MessageStatus.Processed;
 
-        bytes memory _payload = _m.body().clone();
-        address _recipient = _m.recipientAddress();
-
         // NB:
         // A call running out of gas TYPICALLY errors the whole tx. We want to
         // a) ensure the call has a sufficient amount of gas to make a
@@ -223,19 +220,18 @@ contract Replica is Initializable, Common, QueueManager {
         require(gasleft() >= PROCESS_GAS + RESERVE_GAS, "!gas");
         // transparently return.
 
-        try
-            IMessageRecipient(_recipient).handle{gas: PROCESS_GAS}(
-                _m.origin(),
-                _m.sender(),
+        address _recipient = _m.recipientAddress();
+
+        bytes memory _payload =
+            abi.encode(_m.origin(), _m.sender(), _m.body().clone());
+
+        bytes memory _calldata =
+            abi.encodePacked(
+                IMessageRecipient(_recipient).handle.selector,
                 _payload
-            )
-        returns (bytes memory _response) {
-            _success = true;
-            _result = _response;
-        } catch (bytes memory _err) {
-            _success = false;
-            _result = _err;
-        }
+            );
+
+        (_success, _result) = _recipient.call{gas: PROCESS_GAS}(_calldata);
 
         nextToProcess = _sequence + 1;
     }

solidity/optics-core/hardhat.config.js

@@ -40,4 +40,7 @@ module.exports = {
     target: 'ethers-v5',
     alwaysGenerateOverloads: false, // should overloads with full signatures like deposit(uint256) be generated always, even if there are no overloads?
   },
+  mocha: {
+    bail: true,
+  },
 };

solidity/optics-core/test/Replica.test.js

@@ -324,6 +324,10 @@ describe('Replica', async () => {
 
     // Ensure proper static call return value
     let [success, ret] = await replica.callStatic.process(opticsMessage);
+
+    // remove the extra encoding layer
+    [ret] = ethers.utils.defaultAbiCoder.decode(['bytes'], ret);
+
     expect(success).to.be.true;
     expect(ret).to.equal(mockVal);
 

solidity/optics-core/test/cross-chain/GovernanceRouter.test.js

@@ -247,8 +247,15 @@ describe('GovernanceRouter', async () => {
     ] = await governorReplicaOnNonGovernorChain.callStatic.testProcess(
       opticsMessage,
     );
+
+    // remove the extra encoding layer
+    [ret] = ethers.utils.defaultAbiCoder.decode(
+      ['bytes'],
+      ethers.utils.toUtf8Bytes(ret),
+    );
+
     expect(success).to.be.true;
-    expect(ret).to.be.empty;
+    expect(ret).to.equal('0x');
   });
 
   it('Transfers governorship', async () => {