SYS-3965: added first stage migration of processed events mechanism to eth-bridge

pallets/avn/src/lib.rs

@@ -703,13 +703,17 @@ 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> {
@@ -749,7 +753,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(())
  }
 }
@@ -770,8 +774,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

@@ -68,7 +68,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;
@@ -103,6 +103,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,
  },
  }
@@ -296,13 +304,16 @@ pub mod pallet {
  pub enum Error<T> {
  CorroborateCallFailed,
  DuplicateConfirmation,
+ DuplicateEventSubmission,
  EmptyFunctionName,
  ErrorAssigningSender,
  EthTxHashAlreadySet,
  EthTxHashMustBeSetBySender,
  ExceedsConfirmationLimit,
  ExceedsCorroborationLimit,
  ExceedsFunctionNameLimit,
+ EventAlreadyProcessed,
+ EventNotProcessed,
  FunctionEncodingError,
  FunctionNameError,
  HandlePublishingResultFailed,
@@ -565,12 +576,13 @@ 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 {
  EthereumEvents::<T>::insert(&events_partition, votes);
  } else {
- process_ethereum_events_partition::<T>(&active_range, &events_partition);
+ process_ethereum_events_partition::<T>(&active_range, &events_partition)?;
  advance_partition::<T>(&active_range, &events_partition);
  }
 
@@ -794,48 +806,62 @@ pub mod pallet {
  fn process_ethereum_events_partition<T: Config>(
  active_range: &ActiveEthRange,
  partition: &EthereumEventsPartition,
- ) {
+ ) -> Result<(), DispatchError> {
  // Remove entry from storage. Ignore votes.
  let _ = EthereumEvents::<T>::take(partition);
  for discovered_event in partition.events().iter() {
  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);
+ process_ethereum_event::<T>(&discovered_event.event)?;
  } 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
  for (partition, votes) in EthereumEvents::<T>::drain() {
  // TODO raise offences
  log::info!("Collators with invalid votes on ethereum events (range: {:?}, partition: {}): {:?}", partition.range(), partition.partition(), votes);
  }
+
+ Ok(())
  }
 
- 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,138 @@
+// 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;
+
+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_noop!(
+ EthBridge::submit_ethereum_events(
+ RuntimeOrigin::none(),
+ context.author_two.clone(),
+ context.second_mock_event_partition.clone(),
+ context.test_signature_two.clone()
+ ),
+ Error::<TestRuntime>::EventAlreadyProcessed
+ );
+ });
+ }
+}