feat: cleanup aborted broadcasts

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

@@ -60,7 +60,7 @@ pub enum PalletConfigUpdate {
 	BroadcastTimeout { blocks: u32 },
 }
 
-pub const PALLET_VERSION: StorageVersion = StorageVersion::new(9);
+pub const PALLET_VERSION: StorageVersion = StorageVersion::new(10);
 
 #[frame_support::pallet]
 pub mod pallet {
@@ -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

@@ -8,4 +8,6 @@ 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.
+	// Migration 9->10 is remove_aborted_broadcasts in runtime lib.
 );

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/lib.rs

@@ -1211,6 +1211,7 @@ type AllMigrations = (
 	MigrationsForV1_7,
 	migrations::housekeeping::Migration,
 	migrations::reap_old_accounts::Migration,
+	migrations::remove_aborted_broadcasts::AllInstancesMigration,
 );
 
 /// All the pallet-specific migrations and migrations that depend on pallet migration order. Do not

state-chain/runtime/src/migrations.rs

@@ -2,4 +2,5 @@
 
 pub mod housekeeping;
 pub mod reap_old_accounts;
+pub mod remove_aborted_broadcasts;
 pub mod serialize_solana_broadcast;

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

@@ -0,0 +1,85 @@
+use cf_chains::instances::{
+	ArbitrumInstance, BitcoinInstance, EthereumInstance, PolkadotInstance, SolanaInstance,
+};
+use cf_runtime_upgrade_utilities::VersionedMigration;
+use frame_support::{traits::OnRuntimeUpgrade, weights::Weight};
+use pallet_cf_broadcast::AbortedBroadcasts;
+use sp_runtime::DispatchError;
+
+use crate::*;
+
+pub type AllInstancesMigration = (
+	// Only Eth and Arb have stale aborted broadcasts
+	VersionedMigration<
+		pallet_cf_broadcast::Pallet<Runtime, EthereumInstance>,
+		EthereumMigration,
+		9,
+		10,
+	>,
+	VersionedMigration<
+		pallet_cf_broadcast::Pallet<Runtime, ArbitrumInstance>,
+		ArbitrumMigration,
+		9,
+		10,
+	>,
+	VersionedMigration<pallet_cf_broadcast::Pallet<Runtime, SolanaInstance>, NoopUpgrade, 9, 10>,
+	VersionedMigration<pallet_cf_broadcast::Pallet<Runtime, PolkadotInstance>, NoopUpgrade, 9, 10>,
+	VersionedMigration<pallet_cf_broadcast::Pallet<Runtime, BitcoinInstance>, NoopUpgrade, 9, 10>,
+);
+
+// Stale aborted broadcasts on mainnet as of 23/09/2024
+const ETHEREUM_ABORTED_BROADCASTS: [BroadcastId; 5] = [3026, 3684, 3686, 11350, 11592];
+const ARBITRUM_ABORTED_BROADCASTS: [BroadcastId; 5] = [238, 239, 345, 423, 426];
+
+pub struct EthereumMigration;
+pub struct ArbitrumMigration;
+
+impl OnRuntimeUpgrade for EthereumMigration {
+	fn on_runtime_upgrade() -> Weight {
+		AbortedBroadcasts::<Runtime, EthereumInstance>::mutate(|aborted| {
+			ETHEREUM_ABORTED_BROADCASTS.iter().for_each(|broadcast_id| {
+				if aborted.remove(broadcast_id) {
+					log::info!("Removed Ethereum aborted broadcast {}", broadcast_id);
+				}
+			});
+		});
+		Weight::zero()
+	}
+
+	#[cfg(feature = "try-runtime")]
+	fn post_upgrade(_state: Vec<u8>) -> Result<(), DispatchError> {
+		let aborted_broadcasts = AbortedBroadcasts::<Runtime, EthereumInstance>::get();
+		ETHEREUM_ABORTED_BROADCASTS.iter().for_each(|broadcast_id| {
+			assert!(
+				!aborted_broadcasts.contains(broadcast_id),
+				"Aborted broadcast {broadcast_id} still exists"
+			);
+		});
+		Ok(())
+	}
+}
+
+impl OnRuntimeUpgrade for ArbitrumMigration {
+	fn on_runtime_upgrade() -> Weight {
+		AbortedBroadcasts::<Runtime, ArbitrumInstance>::mutate(|aborted| {
+			ARBITRUM_ABORTED_BROADCASTS.iter().for_each(|broadcast_id| {
+				if aborted.remove(broadcast_id) {
+					log::info!("Removed Arbitrum aborted broadcast {}", broadcast_id);
+				}
+			});
+		});
+		Weight::zero()
+	}
+
+	#[cfg(feature = "try-runtime")]
+	fn post_upgrade(_state: Vec<u8>) -> Result<(), DispatchError> {
+		let aborted_broadcasts = AbortedBroadcasts::<Runtime, ArbitrumInstance>::get();
+		ARBITRUM_ABORTED_BROADCASTS.iter().for_each(|broadcast_id| {
+			assert!(
+				!aborted_broadcasts.contains(broadcast_id),
+				"Aborted broadcast {broadcast_id} still exists"
+			);
+		});
+		Ok(())
+	}
+}