SYS-3965: added first stage migration of processed events mechanism t…

…o eth-bridge (#390)

Co-authored-by: Thanos <56822898+thadouk@users.noreply.github.com>
Co-authored-by: nahuseyoum <nahu.seyoum@aventus.io>

pallets/avn/src/lib.rs

@@ -706,13 +706,16 @@ impl<ValidatorId: Member> Enforcer<ValidatorId> for () {
 }
 
 pub trait ProcessedEventsChecker {
-    fn check_event(event_id: &EthEventId) -> bool;
+    fn processed_event_exists(event_id: &EthEventId) -> bool;
+    fn add_processed_event(event_id: &EthEventId, accepted: bool);
 }
 
 impl ProcessedEventsChecker for () {
-    fn check_event(_event_id: &EthEventId) -> bool {
-        return false
+    fn processed_event_exists(_event_id: &EthEventId) -> bool {
+        false
     }
+
+    fn add_processed_event(_event_id: &EthEventId, _accepted: bool) {}
 }
 
 pub trait OnGrowthLiftedHandler<Balance> {
@@ -752,7 +755,7 @@ pub trait BridgeInterfaceNotification {
     fn process_lower_proof_result(_: u32, _: Vec<u8>, _: Result<Vec<u8>, ()>) -> DispatchResult {
         Ok(())
     }
-    fn on_event_processed(_event: &EthEvent) -> DispatchResult {
+    fn on_incoming_event_processed(_event: &EthEvent) -> DispatchResult {
         Ok(())
     }
 }
@@ -773,8 +776,8 @@ impl BridgeInterfaceNotification for Tuple {
         Ok(())
     }
 
-    fn on_event_processed(_event: &EthEvent) -> DispatchResult {
-        for_tuples!( #( Tuple::on_event_processed(_event)?; )* );
+    fn on_incoming_event_processed(_event: &EthEvent) -> DispatchResult {
+        for_tuples!( #( Tuple::on_incoming_event_processed(_event)?; )* );
         Ok(())
     }
 }

pallets/eth-bridge/src/lib.rs

@@ -69,7 +69,7 @@ use frame_system::{
 };
 use pallet_avn::{
     self as avn, BridgeInterface, BridgeInterfaceNotification, Error as avn_error, LowerParams,
-    MAX_VALIDATOR_ACCOUNTS,
+    ProcessedEventsChecker, MAX_VALIDATOR_ACCOUNTS,
 };
 
 use pallet_session::historical::IdentificationTuple;
@@ -79,7 +79,7 @@ use sp_application_crypto::RuntimeAppPublic;
 use sp_avn_common::{
     bounds::MaximumValidatorsBound,
     event_discovery::*,
-    event_types::{ValidEvents, Validator},
+    event_types::{Validator},
 };
 use sp_core::{ecdsa, ConstU32, H160, H256};
 use sp_io::hashing::keccak_256;
@@ -101,6 +101,9 @@ mod benchmarking;
 #[path = "tests/event_listener_tests.rs"]
 mod event_listener_tests;
 #[cfg(test)]
+#[path = "tests/incoming_events_tests.rs"]
+mod incoming_events_tests;
+#[cfg(test)]
 #[path = "tests/lower_proof_tests.rs"]
 mod lower_proof_tests;
 #[cfg(test)]
@@ -183,6 +186,7 @@ pub mod pallet {
             IdentificationTuple<Self>,
             CorroborationOffence<IdentificationTuple<Self>>,
         >;
+        type ProcessedEventsChecker: ProcessedEventsChecker;
         type EthereumEventsFilter: EthereumEventsFilterTrait;
     }
 
@@ -219,9 +223,13 @@ pub mod pallet {
                 BoundedVec<(BoundedVec<u8, TypeLimit>, BoundedVec<u8, ValueLimit>), ParamsLimit>,
             caller_id: BoundedVec<u8, CallerIdLimit>,
         },
-        /// EventProcessed(bool, EthEventId)
-        EventProcessed {
-            accepted: bool,
+        EventAccepted {
+            eth_event_id: EthEventId,
+        },
+        EventRejected {
+            eth_event_id: EthEventId,
+        },
+        DuplicateEventSubmission {
             eth_event_id: EthEventId,
         },
     }
@@ -297,13 +305,16 @@ pub mod pallet {
     pub enum Error<T> {
         CorroborateCallFailed,
         DuplicateConfirmation,
+        DuplicateEventSubmission,
         EmptyFunctionName,
         ErrorAssigningSender,
         EthTxHashAlreadySet,
         EthTxHashMustBeSetBySender,
         ExceedsConfirmationLimit,
         ExceedsCorroborationLimit,
         ExceedsFunctionNameLimit,
+        EventAlreadyProcessed,
+        EventNotProcessed,
         FunctionEncodingError,
         FunctionNameError,
         HandlePublishingResultFailed,
@@ -566,6 +577,7 @@ pub mod pallet {
             );
 
             let mut votes = EthereumEvents::<T>::get(&events_partition);
+
             votes.try_insert(author.account_id).map_err(|_| Error::<T>::EventVotesFull)?;
 
             if votes.len() < AVN::<T>::quorum() as usize {
@@ -802,15 +814,22 @@ pub mod pallet {
             match ValidEvents::try_from(&discovered_event.event.event_id.signature) {
                 Some(valid_event) =>
                     if active_range.event_types_filter.contains(&valid_event) {
-                        process_ethereum_event::<T>(&discovered_event.event);
+                        if process_ethereum_event::<T>(&discovered_event.event).is_err() {
+                                log::error!("💔 Duplicate Event Submission");
+                                <Pallet<T>>::deposit_event(Event::<T>::DuplicateEventSubmission {
+                                    eth_event_id: discovered_event.event.event_id.clone(),
+                                });
+                        }
                     } else {
                         log::warn!("Ethereum event signature ({:?}) included in approved range ({:?}), but not part of the expected ones {:?}", &discovered_event.event.event_id.signature, active_range.range, active_range.event_types_filter);
                     },
-                None => log::warn!(
-                    "Unknown Ethereum event signature in range {:?}",
-                    &discovered_event.event.event_id.signature
-                ),
-            };
+                None => {
+                    log::warn!(
+                        "Unknown Ethereum event signature in range {:?}",
+                        &discovered_event.event.event_id.signature
+                    );
+                },
+            }
         }
 
         // Cleanup
@@ -820,23 +839,33 @@ pub mod pallet {
         }
     }
 
-    fn process_ethereum_event<T: Config>(event: &EthEvent) {
-        // TODO before processing ensure that the event has not already been processed
-        match T::BridgeInterfaceNotification::on_event_processed(&event) {
+    fn process_ethereum_event<T: Config>(event: &EthEvent) -> Result<(), DispatchError> {
+        ensure!(
+            false == T::ProcessedEventsChecker::processed_event_exists(&event.event_id.clone()),
+            Error::<T>::EventAlreadyProcessed
+        );
+
+        let mut event_accepted = false;
+
+        match T::BridgeInterfaceNotification::on_incoming_event_processed(&event) {
             Ok(_) => {
-                <Pallet<T>>::deposit_event(Event::<T>::EventProcessed {
-                    accepted: true,
+                event_accepted = true;
+                <Pallet<T>>::deposit_event(Event::<T>::EventAccepted {
                     eth_event_id: event.event_id.clone(),
                 });
             },
             Err(err) => {
                 log::error!("💔 Processing ethereum event failed: {:?}", err);
-                <Pallet<T>>::deposit_event(Event::<T>::EventProcessed {
-                    accepted: false,
+                <Pallet<T>>::deposit_event(Event::<T>::EventRejected {
                     eth_event_id: event.event_id.clone(),
                 });
             },
         };
+
+        // Add record of succesful processing via ProcessedEventsChecker
+        T::ProcessedEventsChecker::add_processed_event(&event.event_id.clone(), event_accepted);
+
+        Ok(())
     }
 
     #[pallet::validate_unsigned]

pallets/eth-bridge/src/tests/incoming_events_tests.rs

@@ -0,0 +1,139 @@
+// Copyright 2023 Aventus Network Systems (UK) Ltd.
+
+#![cfg(test)]
+use crate::{eth::generate_send_calldata, mock::*, request::*, *};
+use frame_support::{
+    assert_err, assert_noop, assert_ok, dispatch::DispatchResultWithPostInfo, error::BadOrigin,
+};
+use sp_runtime::{testing::UintAuthorityId, DispatchError};
+pub extern crate alloc;
+use alloc::collections::BTreeSet;
+use sp_avn_common::{event_discovery::EthBridgeEventsFilter, event_types::ValidEvents};
+
+const ROOT_HASH: &str = "30b83f0d722d1d4308ab4660a72dbaf0a7392d5674eca3cd21d57256d42df7a0";
+const REWARDS: &[u8] = b"15043665996000000000";
+const AVG_STAKED: &[u8] = b"9034532443555111110000";
+const PERIOD: &[u8] = b"3";
+const T2_PUB_KEY: &str = "14aeac90dbd3573458f9e029eb2de122ee94f2f0bc5ee4b6c6c5839894f1a547";
+const T1_PUB_KEY: &str = "23d79f6492dddecb436333a5e7a4cfcc969f568e01283fa2964aae15327fb8a3b685a4d0f3ef9b3c2adb20f681dbc74b7f82c1cf8438d37f2c10e9c79591e9ea";
+
+// Added this function as in event_listener_tests to initialize the active event range
+fn init_active_range() {
+    ActiveEthereumRange::<TestRuntime>::put(ActiveEthRange {
+        range: EthBlockRange { start_block: 1, length: 1000 },
+        partition: 0,
+        event_types_filter: EthBridgeEventsFilter::try_from(
+            vec![
+                ValidEvents::AddedValidator,
+                ValidEvents::Lifted,
+                ValidEvents::AvtGrowthLifted,
+                ValidEvents::AvtLowerClaimed,
+            ]
+            .into_iter()
+            .collect::<BTreeSet<ValidEvents>>(),
+        )
+        .unwrap(),
+    });
+}
+
+mod process_events {
+    use super::*;
+    use sp_avn_common::event_types::EthEventId;
+
+    // succesfully process the specified ethereum_event
+    #[test]
+    fn succesful_event_processing_accepted() {
+        let mut ext = ExtBuilder::build_default().with_validators().as_externality();
+        ext.execute_with(|| {
+            let context = setup_context();
+            init_active_range();
+
+            // Two calls needed as upon the first there are not enough votes to pass the condition
+            // in lib.rs line 563, to reach the call of process_ethereum_events_partition()
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author.clone(),
+                context.mock_event_partition.clone(),
+                context.test_signature.clone()
+            ));
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author_two.clone(),
+                context.mock_event_partition.clone(),
+                context.test_signature_two.clone()
+            ));
+
+            assert!(System::events().iter().any(|record| record.event ==
+                mock::RuntimeEvent::EthBridge(Event::<TestRuntime>::EventAccepted {
+                    eth_event_id: context.eth_event_id.clone(),
+                })));
+        });
+    }
+
+    // This test should fail processing the ethereum_event and emit the specified event
+    #[test]
+    fn succesful_event_processing_not_accepted() {
+        let mut ext = ExtBuilder::build_default().with_validators().as_externality();
+        ext.execute_with(|| {
+            let context = setup_context();
+            init_active_range();
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author.clone(),
+                context.bad_mock_event_partition.clone(),
+                context.test_signature.clone()
+            ));
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author_two.clone(),
+                context.bad_mock_event_partition.clone(),
+                context.test_signature.clone()
+            ));
+
+            assert!(System::events().iter().any(|record| record.event ==
+                mock::RuntimeEvent::EthBridge(Event::<TestRuntime>::EventRejected {
+                    eth_event_id: context.bad_eth_event_id.clone(),
+                })));
+        });
+    }
+
+    // This test should fail on the check
+    // T::ProcessedEventsChecker::processed_event_exists(&event.event_id.clone()), if the event is
+    // already in the system
+    #[test]
+    fn event_already_processed() {
+        let mut ext = ExtBuilder::build_default().with_validators().as_externality();
+        ext.execute_with(|| {
+            let context = setup_context();
+            init_active_range();
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author.clone(),
+                context.mock_event_partition.clone(),
+                context.test_signature.clone()
+            ));
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author_two.clone(),
+                context.mock_event_partition.clone(),
+                context.test_signature_two.clone()
+            ));
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author.clone(),
+                context.second_mock_event_partition.clone(),
+                context.test_signature.clone()
+            ));
+            assert_ok!(EthBridge::submit_ethereum_events(
+                RuntimeOrigin::none(),
+                context.author_two.clone(),
+                context.second_mock_event_partition.clone(),
+                context.test_signature_two.clone()
+            ));
+            assert!(System::events().iter().any(|record| record.event ==
+                mock::RuntimeEvent::EthBridge(Event::<TestRuntime>::DuplicateEventSubmission {
+                    eth_event_id: context.eth_event_id.clone(),
+                })));
+        });
+    }
+}