fatxpool: fix: remove invalid txs from the dropped stream controller

prdoc/pr_8923.prdoc

@@ -0,0 +1,11 @@
+title: '`fatxpool`: fix: remove invalid txs from the dropped stream controller'
+doc:
+- audience: Node Dev
+  description: |-
+    While testing mortal transaction I encountered exactly the same problem as in #8490.
+    This PR should fix the problem.
+
+    fixes: #8490
+crates:
+- name: sc-transaction-pool
+  bump: minor

substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs

@@ -238,8 +238,12 @@ where
 		let (dropped_stream_controller, dropped_stream) =
 			MultiViewDroppedWatcherController::<ChainApi>::new();
 
-		let view_store =
-			Arc::new(ViewStore::new(pool_api.clone(), listener, dropped_stream_controller));
+		let view_store = Arc::new(ViewStore::new(
+			pool_api.clone(),
+			listener,
+			dropped_stream_controller,
+			import_notification_sink.clone(),
+		));
 
 		let dropped_monitor_task = Self::dropped_monitor_task(
 			dropped_stream,
@@ -372,8 +376,12 @@ where
 		let (dropped_stream_controller, dropped_stream) =
 			MultiViewDroppedWatcherController::<ChainApi>::new();
 
-		let view_store =
-			Arc::new(ViewStore::new(pool_api.clone(), listener, dropped_stream_controller));
+		let view_store = Arc::new(ViewStore::new(
+			pool_api.clone(),
+			listener,
+			dropped_stream_controller,
+			import_notification_sink.clone(),
+		));
 
 		let dropped_monitor_task = Self::dropped_monitor_task(
 			dropped_stream,

substrate/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs

@@ -605,8 +605,18 @@ where
 		let revalidated_invalid_hashes_len = revalidated_invalid_hashes.len();
 		let invalid_hashes_subtrees_len = invalid_hashes_subtrees.len();
 
-		self.listener
-			.transactions_invalidated(&invalid_hashes_subtrees.into_iter().collect::<Vec<_>>());
+		let invalid_hashes_subtrees = invalid_hashes_subtrees.into_iter().collect::<Vec<_>>();
+
+		//note: here the consistency is assumed: it is expected that transaction will be
+		// actually removed from the listener with Invalid event. This means assumption that no view
+		// is referencing tx as ready.
+		self.listener.transactions_invalidated(&invalid_hashes_subtrees);
+		view_store
+			.import_notification_sink
+			.clean_notified_items(&invalid_hashes_subtrees);
+		view_store
+			.dropped_stream_controller
+			.remove_transactions(invalid_hashes_subtrees);
 
 		trace!(
 			target: LOG_TARGET,

substrate/client/transaction-pool/src/fork_aware_txpool/view_store.rs

@@ -19,6 +19,7 @@
 //! Transaction pool view store. Basically block hash to view map with some utility methods.
 
 use super::{
+	import_notification_sink::MultiViewImportNotificationSink,
 	multi_view_listener::{MultiViewListener, TxStatusStream},
 	view::{View, ViewPoolObserver},
 };
@@ -171,6 +172,10 @@ where
 	pub(super) most_recent_view: RwLock<Option<Block::Hash>>,
 	/// The controller of multi view dropped stream.
 	pub(super) dropped_stream_controller: MultiViewDroppedWatcherController<ChainApi>,
+	/// Util providing an aggregated stream of transactions that were imported to ready queue in
+	/// any view. Reference kept here for clean up purposes.
+	pub(super) import_notification_sink:
+		MultiViewImportNotificationSink<Block::Hash, ExtrinsicHash<ChainApi>>,
 	/// The map used to synchronize replacement of transactions between maintain and dropped
 	/// notifcication threads. It is meant to assure that replaced transaction is also removed from
 	/// newly built views in maintain process.
@@ -202,6 +207,10 @@ where
 		api: Arc<ChainApi>,
 		listener: Arc<MultiViewListener<ChainApi>>,
 		dropped_stream_controller: MultiViewDroppedWatcherController<ChainApi>,
+		import_notification_sink: MultiViewImportNotificationSink<
+			Block::Hash,
+			ExtrinsicHash<ChainApi>,
+		>,
 	) -> Self {
 		Self {
 			api,
@@ -210,6 +219,7 @@ where
 			listener,
 			most_recent_view: RwLock::from(None),
 			dropped_stream_controller,
+			import_notification_sink,
 			pending_txs_tasks: Default::default(),
 		}
 	}