feat: cleanup aborted broadcasts (#5301)

* feat: cleanup aborted broadcasts

* refactor: use housekeeping migration

* refactor: remove stale from perseverance as well

* chore: remove left over structs

state-chain/pallets/cf-broadcast/src/lib.rs

@@ -688,6 +688,9 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 			.ok_or(Error::<T, I>::InvalidPayload)?;
 
 		Self::remove_pending_broadcast(&broadcast_id);
+		AbortedBroadcasts::<T, I>::mutate(|aborted| {
+			aborted.remove(&broadcast_id);
+		});
 
 		if IncomingKeyAndBroadcastId::<T, I>::exists() {
 			let (incoming_key, rotation_broadcast_id) =
@@ -1089,7 +1092,7 @@ impl<T: Config<I>, I: 'static> Broadcaster<T::TargetChain> for Pallet<T, I> {
 	}
 
 	fn expire_broadcast(broadcast_id: BroadcastId) {
-		// These would otherwise be cleaned up when the broadacst succeeds or aborts.
+		// These would otherwise be cleaned up when the broadcast succeeds or aborts.
 		RequestSuccessCallbacks::<T, I>::remove(broadcast_id);
 		RequestFailureCallbacks::<T, I>::remove(broadcast_id);
 		Self::clean_up_broadcast_storage(broadcast_id);

state-chain/pallets/cf-broadcast/src/migrations.rs

@@ -3,9 +3,11 @@ use cf_runtime_upgrade_utilities::{PlaceholderMigration, VersionedMigration};
 
 mod initialize_broadcast_timeout_storage;
 mod migrate_timeouts;
+pub mod remove_aborted_broadcasts;
 
 pub type PalletMigration<T, I> = (
 	VersionedMigration<Pallet<T, I>, initialize_broadcast_timeout_storage::Migration<T, I>, 6, 7>,
 	VersionedMigration<Pallet<T, I>, migrate_timeouts::Migration<T, I>, 7, 8>,
 	PlaceholderMigration<Pallet<T, I>, 8>,
+	// Migration 8->9 is SerializeSolanaBroadcastMigration in runtime lib.
 );

state-chain/pallets/cf-broadcast/src/migrations/remove_aborted_broadcasts.rs

@@ -0,0 +1,24 @@
+use crate::*;
+
+// Highest stale aborted broadcasts as of 3/10/2024:
+// Mainnet
+pub const ETHEREUM_MAX_ABORTED_BROADCAST_BERGHAIN: BroadcastId = 11592;
+pub const ARBITRUM_MAX_ABORTED_BROADCAST_BERGHAIN: BroadcastId = 426;
+// Perseverance testnet
+pub const ETHEREUM_MAX_ABORTED_BROADCAST_PERSEVERANCE: BroadcastId = 1609;
+pub const ARBITRUM_MAX_ABORTED_BROADCAST_PERSEVERANCE: BroadcastId = 665;
+pub const POLKADOT_MAX_ABORTED_BROADCAST_PERSEVERANCE: BroadcastId = 634;
+
+pub fn remove_stale_and_all_older<T: Config<I>, I: 'static>(latest_stale_broadcast: BroadcastId) {
+	AbortedBroadcasts::<T, I>::mutate(|aborted| {
+		aborted.retain(|id| id > &latest_stale_broadcast);
+	});
+}
+
+#[cfg(feature = "try-runtime")]
+pub fn assert_removed<T: Config<I>, I: 'static>(latest_stale_broadcast: BroadcastId) {
+	let aborted_broadcasts = AbortedBroadcasts::<T, I>::get();
+	if let Some(first) = aborted_broadcasts.first() {
+		assert!(*first > latest_stale_broadcast, "Aborted broadcast {first} was not removed");
+	}
+}

state-chain/pallets/cf-broadcast/src/tests.rs

@@ -1562,3 +1562,27 @@ fn changing_broadcast_timeout() {
 		);
 	});
 }
+
+#[test]
+fn aborted_broadcast_is_cleaned_up_on_success() {
+	new_test_ext().execute_with(|| {
+		// Abort a broadcast
+		let (broadcast_id, tx_id) = start_mock_broadcast();
+		let nominee = ready_to_abort_broadcast(broadcast_id);
+		assert_ok!(Broadcaster::transaction_failed(RuntimeOrigin::signed(nominee), broadcast_id));
+		assert!(AbortedBroadcasts::<Test, Instance1>::get().contains(&broadcast_id));
+
+		// Witness a successful broadcast as if it was manually broadcast
+		assert_ok!(Broadcaster::transaction_succeeded(
+			RuntimeOrigin::root(),
+			tx_id,
+			Default::default(),
+			ETH_TX_FEE,
+			MOCK_TX_METADATA,
+			Default::default(),
+		));
+
+		// Storage should be cleaned, event emitted.
+		assert!(AbortedBroadcasts::<Test, Instance1>::get().is_empty());
+	});
+}

state-chain/runtime/src/migrations/housekeeping.rs

@@ -1,17 +1,38 @@
 use crate::Runtime;
+use cf_chains::instances::{ArbitrumInstance, EthereumInstance, PolkadotInstance};
+use cf_runtime_upgrade_utilities::genesis_hashes;
 use frame_support::{traits::OnRuntimeUpgrade, weights::Weight};
+use pallet_cf_broadcast::migrations::remove_aborted_broadcasts;
+#[cfg(feature = "try-runtime")]
+use sp_runtime::DispatchError;
+#[cfg(feature = "try-runtime")]
+use sp_std::vec::Vec;
 
 pub struct Migration;
 
 impl OnRuntimeUpgrade for Migration {
 	fn on_runtime_upgrade() -> Weight {
-		use cf_runtime_upgrade_utilities::genesis_hashes;
 		match genesis_hashes::genesis_hash::<Runtime>() {
 			genesis_hashes::BERGHAIN => {
-				log::info!("🧹 No housekeeping required for Berghain.");
+				log::info!("🧹 Housekeeping, removing stale aborted broadcasts");
+				remove_aborted_broadcasts::remove_stale_and_all_older::<Runtime, EthereumInstance>(
+					remove_aborted_broadcasts::ETHEREUM_MAX_ABORTED_BROADCAST_BERGHAIN,
+				);
+				remove_aborted_broadcasts::remove_stale_and_all_older::<Runtime, ArbitrumInstance>(
+					remove_aborted_broadcasts::ARBITRUM_MAX_ABORTED_BROADCAST_BERGHAIN,
+				);
 			},
 			genesis_hashes::PERSEVERANCE => {
-				log::info!("🧹 No housekeeping required for Perseverance.");
+				log::info!("🧹 Housekeeping, removing stale aborted broadcasts");
+				remove_aborted_broadcasts::remove_stale_and_all_older::<Runtime, EthereumInstance>(
+					remove_aborted_broadcasts::ETHEREUM_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
+				remove_aborted_broadcasts::remove_stale_and_all_older::<Runtime, ArbitrumInstance>(
+					remove_aborted_broadcasts::ARBITRUM_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
+				remove_aborted_broadcasts::remove_stale_and_all_older::<Runtime, PolkadotInstance>(
+					remove_aborted_broadcasts::POLKADOT_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
 			},
 			genesis_hashes::SISYPHOS => {
 				log::info!("🧹 No housekeeping required for Sisyphos.");
@@ -21,4 +42,40 @@ impl OnRuntimeUpgrade for Migration {
 
 		Weight::zero()
 	}
+
+	#[cfg(feature = "try-runtime")]
+	fn post_upgrade(_state: Vec<u8>) -> Result<(), DispatchError> {
+		match genesis_hashes::genesis_hash::<Runtime>() {
+			genesis_hashes::BERGHAIN => {
+				log::info!(
+					"Housekeeping post_upgrade, checking stale aborted broadcasts are removed."
+				);
+				remove_aborted_broadcasts::assert_removed::<Runtime, EthereumInstance>(
+					remove_aborted_broadcasts::ETHEREUM_MAX_ABORTED_BROADCAST_BERGHAIN,
+				);
+				remove_aborted_broadcasts::assert_removed::<Runtime, ArbitrumInstance>(
+					remove_aborted_broadcasts::ARBITRUM_MAX_ABORTED_BROADCAST_BERGHAIN,
+				);
+			},
+			genesis_hashes::PERSEVERANCE => {
+				log::info!(
+					"Housekeeping post_upgrade, checking stale aborted broadcasts are removed."
+				);
+				remove_aborted_broadcasts::assert_removed::<Runtime, EthereumInstance>(
+					remove_aborted_broadcasts::ETHEREUM_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
+				remove_aborted_broadcasts::assert_removed::<Runtime, ArbitrumInstance>(
+					remove_aborted_broadcasts::ARBITRUM_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
+				remove_aborted_broadcasts::assert_removed::<Runtime, PolkadotInstance>(
+					remove_aborted_broadcasts::POLKADOT_MAX_ABORTED_BROADCAST_PERSEVERANCE,
+				);
+			},
+			genesis_hashes::SISYPHOS => {
+				log::info!("Skipping housekeeping post_upgrade for Sisyphos.");
+			},
+			_ => {},
+		}
+		Ok(())
+	}
 }