Merge remote-tracking branch 'origin/master' into alindima/add-systematic-chunks-av-recovery

Author: alindima

.gitlab/pipeline/publish.yml

@@ -129,7 +129,7 @@ trigger_workflow:
         curl -q -X POST \
               -H "Accept: application/vnd.github.v3+json" \
               -H "Authorization: token $GITHUB_TOKEN" \
-              https://api.github.com/repos/paritytech/${CI_PROJECT_NAME}/actions/workflows/subsystem-benchmarks.yml/dispatches \
+              https://api.github.com/repos/paritytech/${CI_PROJECT_NAME}/actions/workflows/publish-subsystem-benchmarks.yml/dispatches \
               -d "{\"ref\":\"refs/heads/master\",\"inputs\":{\"benchmark-data-dir-path\":\"$benchmark_dir\",\"output-file-path\":\"$benchmark_name\"}}";
         sleep 300;
       done

Cargo.lock

@@ -77,7 +77,12 @@ pub fn is_ancient_block<N: From<u32> + PartialOrd + Saturating>(block: N, best:
 }
 
 /// Opaque justifications subscription type.
-pub struct Subscription<T>(pub(crate) Mutex<futures::channel::mpsc::Receiver<Option<T>>>);
+pub struct Subscription<T>(
+	pub(crate) Mutex<futures::channel::mpsc::Receiver<Option<T>>>,
+	// The following field is not explicitly used by the code. But when it is dropped,
+	// the bakground task receives a shutdown signal.
+	#[allow(dead_code)] pub(crate) futures::channel::oneshot::Sender<()>,
+);
 
 /// Opaque GRANDPA authorities set.
 pub type OpaqueGrandpaAuthoritiesSet = Vec<u8>;
@@ -621,6 +626,7 @@ impl<C: Chain> Client<C> {
 				e
 			})??;
 
+		let (cancel_sender, cancel_receiver) = futures::channel::oneshot::channel();
 		let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
 		let (tracker, subscription) = self
 			.jsonrpsee_execute(move |client| async move {
@@ -639,7 +645,7 @@ impl<C: Chain> Client<C> {
 					self_clone,
 					stall_timeout,
 					tx_hash,
-					Subscription(Mutex::new(receiver)),
+					Subscription(Mutex::new(receiver), cancel_sender),
 				);
 				Ok((tracker, subscription))
 			})
@@ -649,6 +655,7 @@ impl<C: Chain> Client<C> {
 			"extrinsic".into(),
 			subscription,
 			sender,
+			cancel_receiver,
 		));
 		Ok(tracker)
 	}
@@ -790,14 +797,16 @@ impl<C: Chain> Client<C> {
 				Ok(FC::subscribe_justifications(&client).await?)
 			})
 			.await?;
+		let (cancel_sender, cancel_receiver) = futures::channel::oneshot::channel();
 		let (sender, receiver) = futures::channel::mpsc::channel(MAX_SUBSCRIPTION_CAPACITY);
 		self.data.read().await.tokio.spawn(Subscription::background_worker(
 			C::NAME.into(),
 			"justification".into(),
 			subscription,
 			sender,
+			cancel_receiver,
 		));
-		Ok(Subscription(Mutex::new(receiver)))
+		Ok(Subscription(Mutex::new(receiver), cancel_sender))
 	}
 
 	/// Generates a proof of key ownership for the given authority in the given set.
@@ -843,9 +852,17 @@ impl<C: Chain> Client<C> {
 impl<T: DeserializeOwned> Subscription<T> {
 	/// Consumes subscription and returns future statuses stream.
 	pub fn into_stream(self) -> impl futures::Stream<Item = T> {
-		futures::stream::unfold(self, |this| async {
+		futures::stream::unfold(Some(self), |mut this| async move {
+			let Some(this) = this.take() else { return None };
 			let item = this.0.lock().await.next().await.unwrap_or(None);
-			item.map(|i| (i, this))
+			match item {
+				Some(item) => Some((item, Some(this))),
+				None => {
+					// let's make it explicit here
+					let _ = this.1.send(());
+					None
+				},
+			}
 		})
 	}
 
@@ -860,36 +877,61 @@ impl<T: DeserializeOwned> Subscription<T> {
 	async fn background_worker(
 		chain_name: String,
 		item_type: String,
-		mut subscription: jsonrpsee::core::client::Subscription<T>,
+		subscription: jsonrpsee::core::client::Subscription<T>,
 		mut sender: futures::channel::mpsc::Sender<Option<T>>,
+		cancel_receiver: futures::channel::oneshot::Receiver<()>,
 	) {
+		log::trace!(
+			target: "bridge",
+			"Starting background worker for {} {} subscription stream.",
+			chain_name,
+			item_type,
+		);
+
+		futures::pin_mut!(subscription, cancel_receiver);
 		loop {
-			match subscription.next().await {
-				Some(Ok(item)) =>
+			match futures::future::select(subscription.next(), &mut cancel_receiver).await {
+				futures::future::Either::Left((Some(Ok(item)), _)) =>
 					if sender.send(Some(item)).await.is_err() {
+						log::trace!(
+							target: "bridge",
+							"{} {} subscription stream: no listener. Stopping background worker.",
+							chain_name,
+							item_type,
+						);
+
 						break
 					},
-				Some(Err(e)) => {
+				futures::future::Either::Left((Some(Err(e)), _)) => {
 					log::trace!(
 						target: "bridge",
-						"{} {} subscription stream has returned '{:?}'. Stream needs to be restarted.",
+						"{} {} subscription stream has returned '{:?}'. Stream needs to be restarted. Stopping background worker.",
 						chain_name,
 						item_type,
 						e,
 					);
 					let _ = sender.send(None).await;
 					break
 				},
-				None => {
+				futures::future::Either::Left((None, _)) => {
 					log::trace!(
 						target: "bridge",
-						"{} {} subscription stream has returned None. Stream needs to be restarted.",
+						"{} {} subscription stream has returned None. Stream needs to be restarted. Stopping background worker.",
 						chain_name,
 						item_type,
 					);
 					let _ = sender.send(None).await;
 					break
 				},
+				futures::future::Either::Right((_, _)) => {
+					log::trace!(
+						target: "bridge",
+						"{} {} subscription stream: listener has been dropped. Stopping background worker.",
+						chain_name,
+						item_type,
+					);
+					break;
+				},
 			}
 		}
 	}

bridges/relays/client-substrate/src/client.rs

@@ -306,12 +306,13 @@ mod tests {
 		TrackedTransactionStatus<HeaderIdOf<TestChain>>,
 		InvalidationStatus<HeaderIdOf<TestChain>>,
 	)> {
+		let (cancel_sender, _cancel_receiver) = futures::channel::oneshot::channel();
 		let (mut sender, receiver) = futures::channel::mpsc::channel(1);
 		let tx_tracker = TransactionTracker::<TestChain, TestEnvironment>::new(
 			TestEnvironment(Ok(HeaderId(0, Default::default()))),
 			Duration::from_secs(0),
 			Default::default(),
-			Subscription(async_std::sync::Mutex::new(receiver)),
+			Subscription(async_std::sync::Mutex::new(receiver), cancel_sender),
 		);
 
 		let wait_for_stall_timeout = futures::future::pending();
@@ -428,12 +429,13 @@ mod tests {
 
 	#[async_std::test]
 	async fn lost_on_timeout_when_waiting_for_invalidation_status() {
+		let (cancel_sender, _cancel_receiver) = futures::channel::oneshot::channel();
 		let (_sender, receiver) = futures::channel::mpsc::channel(1);
 		let tx_tracker = TransactionTracker::<TestChain, TestEnvironment>::new(
 			TestEnvironment(Ok(HeaderId(0, Default::default()))),
 			Duration::from_secs(0),
 			Default::default(),
-			Subscription(async_std::sync::Mutex::new(receiver)),
+			Subscription(async_std::sync::Mutex::new(receiver), cancel_sender),
 		);
 
 		let wait_for_stall_timeout = futures::future::ready(()).shared();

bridges/relays/client-substrate/src/transaction_tracker.rs

@@ -125,14 +125,13 @@ impl PrometheusParams {
 			None
 		};
 
-		let relay_version = option_env!("CARGO_PKG_VERSION").unwrap_or("unknown");
+		let relay_version = relay_utils::initialize::RELAYER_VERSION
+			.lock()
+			.clone()
+			.unwrap_or_else(|| "unknown".to_string());
 		let relay_commit = SubstrateRelayBuildInfo::get_git_commit();
-		relay_utils::metrics::MetricsParams::new(
-			metrics_address,
-			relay_version.into(),
-			relay_commit,
-		)
-		.map_err(|e| anyhow::format_err!("{:?}", e))
+		relay_utils::metrics::MetricsParams::new(metrics_address, relay_version, relay_commit)
+			.map_err(|e| anyhow::format_err!("{:?}", e))
 	}
 }
 

bridges/relays/lib-substrate-relay/src/cli/mod.rs

@@ -22,6 +22,7 @@ futures = "0.3.30"
 jsonpath_lib = "0.3"
 log = { workspace = true }
 num-traits = "0.2"
+parking_lot = "0.12.1"
 serde_json = { workspace = true, default-features = true }
 sysinfo = "0.30"
 time = { version = "0.3", features = ["formatting", "local-offset", "std"] }

bridges/relays/utils/Cargo.toml

@@ -16,8 +16,13 @@
 
 //! Relayer initialization functions.
 
+use parking_lot::Mutex;
 use std::{cell::RefCell, fmt::Display, io::Write};
 
+/// Relayer version that is provided as metric. Must be set by a binary
+/// (get it with `option_env!("CARGO_PKG_VERSION")` from a binary package code).
+pub static RELAYER_VERSION: Mutex<Option<String>> = Mutex::new(None);
+
 async_std::task_local! {
 	pub(crate) static LOOP_NAME: RefCell<String> = RefCell::new(String::default());
 }

bridges/relays/utils/src/initialize.rs

@@ -104,6 +104,7 @@ pub mod pallet {
 	#[pallet::error]
 	pub enum Error<T> {
 		SkippedSyncCommitteePeriod,
+		SyncCommitteeUpdateRequired,
 		/// Attested header is older than latest finalized header.
 		IrrelevantUpdate,
 		NotBootstrapped,
@@ -138,41 +139,39 @@ pub mod pallet {
 	/// Latest imported checkpoint root
 	#[pallet::storage]
 	#[pallet::getter(fn initial_checkpoint_root)]
-	pub(super) type InitialCheckpointRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
+	pub type InitialCheckpointRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
 
 	/// Latest imported finalized block root
 	#[pallet::storage]
 	#[pallet::getter(fn latest_finalized_block_root)]
-	pub(super) type LatestFinalizedBlockRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
+	pub type LatestFinalizedBlockRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
 
 	/// Beacon state by finalized block root
 	#[pallet::storage]
 	#[pallet::getter(fn finalized_beacon_state)]
-	pub(super) type FinalizedBeaconState<T: Config> =
+	pub type FinalizedBeaconState<T: Config> =
 		StorageMap<_, Identity, H256, CompactBeaconState, OptionQuery>;
 
 	/// Finalized Headers: Current position in ring buffer
 	#[pallet::storage]
-	pub(crate) type FinalizedBeaconStateIndex<T: Config> = StorageValue<_, u32, ValueQuery>;
+	pub type FinalizedBeaconStateIndex<T: Config> = StorageValue<_, u32, ValueQuery>;
 
 	/// Finalized Headers: Mapping of ring buffer index to a pruning candidate
 	#[pallet::storage]
-	pub(crate) type FinalizedBeaconStateMapping<T: Config> =
+	pub type FinalizedBeaconStateMapping<T: Config> =
 		StorageMap<_, Identity, u32, H256, ValueQuery>;
 
 	#[pallet::storage]
 	#[pallet::getter(fn validators_root)]
-	pub(super) type ValidatorsRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
+	pub type ValidatorsRoot<T: Config> = StorageValue<_, H256, ValueQuery>;
 
 	/// Sync committee for current period
 	#[pallet::storage]
-	pub(super) type CurrentSyncCommittee<T: Config> =
-		StorageValue<_, SyncCommitteePrepared, ValueQuery>;
+	pub type CurrentSyncCommittee<T: Config> = StorageValue<_, SyncCommitteePrepared, ValueQuery>;
 
 	/// Sync committee for next period
 	#[pallet::storage]
-	pub(super) type NextSyncCommittee<T: Config> =
-		StorageValue<_, SyncCommitteePrepared, ValueQuery>;
+	pub type NextSyncCommittee<T: Config> = StorageValue<_, SyncCommitteePrepared, ValueQuery>;
 
 	/// The current operating mode of the pallet.
 	#[pallet::storage]
@@ -320,6 +319,7 @@ pub mod pallet {
 
 			// Verify update is relevant.
 			let update_attested_period = compute_period(update.attested_header.slot);
+			let update_finalized_period = compute_period(update.finalized_header.slot);
 			let update_has_next_sync_committee = !<NextSyncCommittee<T>>::exists() &&
 				(update.next_sync_committee_update.is_some() &&
 					update_attested_period == store_period);
@@ -395,6 +395,11 @@ pub mod pallet {
 					),
 					Error::<T>::InvalidSyncCommitteeMerkleProof
 				);
+			} else {
+				ensure!(
+					update_finalized_period == store_period,
+					Error::<T>::SyncCommitteeUpdateRequired
+				);
 			}
 
 			// Verify sync committee aggregate signature.

bridges/snowbridge/pallets/ethereum-client/src/lib.rs

@@ -362,20 +362,32 @@ fn submit_update_with_sync_committee_in_current_period() {
 }
 
 #[test]
-fn submit_update_in_next_period() {
+fn reject_submit_update_in_next_period() {
 	let checkpoint = Box::new(load_checkpoint_update_fixture());
 	let sync_committee_update = Box::new(load_sync_committee_update_fixture());
 	let update = Box::new(load_next_finalized_header_update_fixture());
 	let sync_committee_period = compute_period(sync_committee_update.finalized_header.slot);
 	let next_sync_committee_period = compute_period(update.finalized_header.slot);
 	assert_eq!(sync_committee_period + 1, next_sync_committee_period);
+	let next_sync_committee_update = Box::new(load_next_sync_committee_update_fixture());
 
 	new_tester().execute_with(|| {
 		assert_ok!(EthereumBeaconClient::process_checkpoint_update(&checkpoint));
 		assert_ok!(EthereumBeaconClient::submit(
 			RuntimeOrigin::signed(1),
 			sync_committee_update.clone()
 		));
+		// check an update in the next period is rejected
+		assert_err!(
+			EthereumBeaconClient::submit(RuntimeOrigin::signed(1), update.clone()),
+			Error::<Test>::SyncCommitteeUpdateRequired
+		);
+		// submit update with next sync committee
+		assert_ok!(EthereumBeaconClient::submit(
+			RuntimeOrigin::signed(1),
+			next_sync_committee_update
+		));
+		// check same header in the next period can now be submitted successfully
 		assert_ok!(EthereumBeaconClient::submit(RuntimeOrigin::signed(1), update.clone()));
 		let block_root: H256 = update.finalized_header.clone().hash_tree_root().unwrap();
 		assert!(<FinalizedBeaconState<Test>>::contains_key(block_root));

bridges/snowbridge/pallets/ethereum-client/src/tests.rs

@@ -18,7 +18,7 @@ pub type NextSyncCommitteeUpdate = primitives::NextSyncCommitteeUpdate<SC_SIZE>;
 pub use primitives::{AncestryProof, ExecutionProof};
 
 /// FinalizedState ring buffer implementation
-pub(crate) type FinalizedBeaconStateBuffer<T> = RingBufferMapImpl<
+pub type FinalizedBeaconStateBuffer<T> = RingBufferMapImpl<
 	u32,
 	crate::MaxFinalizedHeadersToKeep<T>,
 	crate::FinalizedBeaconStateIndex<T>,