dropped_watcher: explicit view removal

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

@@ -142,12 +142,7 @@ where
 					self.transaction_states.entry(tx_hash)
 				{
 					views_keeping_tx_valid.get_mut().remove(&block_hash);
-					if views_keeping_tx_valid.get().is_empty() ||
-						views_keeping_tx_valid
-							.get()
-							.iter()
-							.all(|h| !self.stream_map.contains_key(h))
-					{
+					if views_keeping_tx_valid.get().is_empty() {
 						return Some(tx_hash)
 					}
 				} else {
@@ -373,6 +368,7 @@ mod dropped_watcher_tests {
 
 		watcher.add_view(block_hash0, view_stream0);
 		assert!(output_stream.next().now_or_never().is_none());
+		watcher.remove_view(block_hash0);
 
 		watcher.add_view(block_hash1, view_stream1);
 		let handle = tokio::spawn(async move { output_stream.take(1).collect::<Vec<_>>().await });

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

@@ -438,27 +438,44 @@ where
 		let finalized_xts = self.finalize_route(finalized_hash, tree_route).await;
 		let finalized_number = self.api.block_id_to_number(&BlockId::Hash(finalized_hash));
 
+		let mut dropped_views = vec![];
 		//clean up older then finalized
 		{
 			let mut active_views = self.active_views.write();
-			active_views.retain(|hash, v| match finalized_number {
-				Err(_) | Ok(None) => *hash == finalized_hash,
-				Ok(Some(n)) if v.at.number == n => *hash == finalized_hash,
-				Ok(Some(n)) => v.at.number > n,
+			active_views.retain(|hash, v| {
+				let retain = match finalized_number {
+					Err(_) | Ok(None) => *hash == finalized_hash,
+					Ok(Some(n)) if v.at.number == n => *hash == finalized_hash,
+					Ok(Some(n)) => v.at.number > n,
+				};
+				if !retain {
+					dropped_views.push(*hash);
+				}
+				retain
 			});
 		}
 
 		{
 			let mut inactive_views = self.inactive_views.write();
-			inactive_views.retain(|_, v| match finalized_number {
-				Err(_) | Ok(None) => false,
-				Ok(Some(n)) => v.at.number >= n,
+			inactive_views.retain(|hash, v| {
+				let retain = match finalized_number {
+					Err(_) | Ok(None) => false,
+					Ok(Some(n)) => v.at.number >= n,
+				};
+				if !retain {
+					dropped_views.push(*hash);
+				}
+				retain
 			});
 
 			log::trace!(target:LOG_TARGET,"handle_finalized: inactive_views: {:?}", inactive_views.keys());
 		}
 
 		self.listener.remove_view(finalized_hash);
+		for view in dropped_views {
+			self.listener.remove_view(view);
+			self.dropped_stream_controller.remove_view(view);
+		}
 		self.listener.remove_stale_controllers();
 		self.dropped_stream_controller.remove_finalized_txs(finalized_xts.clone());