chainHead: Remove all active subscriptions instead of suspending time

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

substrate/client/rpc-spec-v2/src/chain_head/chain_head.rs

@@ -67,11 +67,6 @@ pub struct ChainHeadConfig {
 	///
 	/// Subscriptions are suspended for the `suspended_duration`.
 	pub suspend_on_lagging_distance: usize,
-	/// The amount of time for which the subscriptions are suspended.
-	///
-	/// Subscriptions are suspended when the distance between any leaf
-	/// and the finalized block is too large.
-	pub suspended_duration: Duration,
 	/// The maximum number of items reported by the `chainHead_storage` before
 	/// pagination is required.
 	pub operation_max_storage_items: usize,
@@ -102,19 +97,12 @@ const MAX_STORAGE_ITER_ITEMS: usize = 5;
 /// Subscriptions are suspended for the `suspended_duration`.
 const SUSPEND_ON_LAGGING_DISTANCE: usize = 128;
 
-/// The amount of time for which the subscriptions are suspended.
-///
-/// Subscriptions are suspended when the distance between any leaf
-/// and the finalized block is too large.
-const SUSPENDED_DURATION: Duration = Duration::from_secs(30);
-
 impl Default for ChainHeadConfig {
 	fn default() -> Self {
 		ChainHeadConfig {
 			global_max_pinned_blocks: MAX_PINNED_BLOCKS,
 			subscription_max_pinned_duration: MAX_PINNED_DURATION,
 			subscription_max_ongoing_operations: MAX_ONGOING_OPERATIONS,
-			suspended_duration: SUSPENDED_DURATION,
 			suspend_on_lagging_distance: SUSPEND_ON_LAGGING_DISTANCE,
 			operation_max_storage_items: MAX_STORAGE_ITER_ITEMS,
 		}
@@ -157,7 +145,6 @@ impl<BE: Backend<Block>, Block: BlockT, Client> ChainHead<BE, Block, Client> {
 				config.global_max_pinned_blocks,
 				config.subscription_max_pinned_duration,
 				config.subscription_max_ongoing_operations,
-				config.suspended_duration,
 				backend,
 			)),
 			operation_max_storage_items: config.operation_max_storage_items,
@@ -240,7 +227,7 @@ where
 
 			if let Err(SubscriptionManagementError::BlockDistanceTooLarge) = result {
 				debug!(target: LOG_TARGET, "[follow][id={:?}] All subscriptions are suspended", sub_id);
-				subscriptions.suspend_subscriptions();
+				subscriptions.stop_all_subscriptions();
 			}
 
 			subscriptions.remove_subscription(&sub_id);

substrate/client/rpc-spec-v2/src/chain_head/subscription/inner.rs

@@ -27,11 +27,7 @@ use std::{
 	time::{Duration, Instant},
 };
 
-use crate::chain_head::{
-	chain_head::LOG_TARGET,
-	subscription::{suspend::SuspendSubscriptions, SubscriptionManagementError},
-	FollowEvent,
-};
+use crate::chain_head::{subscription::SubscriptionManagementError, FollowEvent};
 
 /// The queue size after which the `sc_utils::mpsc::tracing_unbounded` would produce warnings.
 const QUEUE_SIZE_WARNING: usize = 512;
@@ -564,8 +560,6 @@ pub struct SubscriptionsInner<Block: BlockT, BE: Backend<Block>> {
 	max_ongoing_operations: usize,
 	/// Map the subscription ID to internal details of the subscription.
 	subs: HashMap<String, SubscriptionState<Block>>,
-	/// Suspend subscriptions for a given amount of time.
-	suspend: SuspendSubscriptions,
 
 	/// Backend pinning / unpinning blocks.
 	///
@@ -579,7 +573,6 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionsInner<Block, BE> {
 		global_max_pinned_blocks: usize,
 		local_max_pin_duration: Duration,
 		max_ongoing_operations: usize,
-		suspend_duration: Duration,
 		backend: Arc<BE>,
 	) -> Self {
 		SubscriptionsInner {
@@ -588,7 +581,6 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionsInner<Block, BE> {
 			local_max_pin_duration,
 			max_ongoing_operations,
 			subs: Default::default(),
-			suspend: SuspendSubscriptions::new(suspend_duration),
 			backend,
 		}
 	}
@@ -599,11 +591,6 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionsInner<Block, BE> {
 		sub_id: String,
 		with_runtime: bool,
 	) -> Option<InsertedSubscriptionData<Block>> {
-		if self.suspend.is_suspended() {
-			log::trace!(target: LOG_TARGET, "[id={:?}] Subscription already suspended", sub_id);
-			return None
-		}
-
 		if let Entry::Vacant(entry) = self.subs.entry(sub_id) {
 			let (tx_stop, rx_stop) = oneshot::channel();
 			let (response_sender, response_receiver) =
@@ -637,12 +624,8 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionsInner<Block, BE> {
 		}
 	}
 
-	/// Suspends all subscriptions for the given duration.
-	///
 	/// All active subscriptions are removed.
-	pub fn suspend_subscriptions(&mut self) {
-		self.suspend.suspend_subscriptions();
-
+	pub fn stop_all_subscriptions(&mut self) {
 		let to_remove: Vec<_> = self.subs.keys().map(|sub_id| sub_id.clone()).collect();
 
 		for sub_id in to_remove {

substrate/client/rpc-spec-v2/src/chain_head/subscription/mod.rs

@@ -23,7 +23,6 @@ use std::{sync::Arc, time::Duration};
 
 mod error;
 mod inner;
-mod suspend;
 
 use self::inner::SubscriptionsInner;
 
@@ -44,15 +43,13 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionManagement<Block, BE> {
 		global_max_pinned_blocks: usize,
 		local_max_pin_duration: Duration,
 		max_ongoing_operations: usize,
-		suspend_duration: Duration,
 		backend: Arc<BE>,
 	) -> Self {
 		SubscriptionManagement {
 			inner: RwLock::new(SubscriptionsInner::new(
 				global_max_pinned_blocks,
 				local_max_pin_duration,
 				max_ongoing_operations,
-				suspend_duration,
 				backend,
 			)),
 		}
@@ -78,15 +75,13 @@ impl<Block: BlockT, BE: Backend<Block>> SubscriptionManagement<Block, BE> {
 		inner.remove_subscription(sub_id)
 	}
 
-	/// Suspends all subscriptions for the given duration.
+	/// Stop all active subscriptions.
 	///
 	/// For all active subscriptions, the internal data is discarded, blocks are unpinned and the
 	/// `Stop` event will be generated.
-	///
-	/// For incoming subscriptions, only the `Stop` event is delivered.
-	pub fn suspend_subscriptions(&self) {
+	pub fn stop_all_subscriptions(&self) {
 		let mut inner = self.inner.write();
-		inner.suspend_subscriptions()
+		inner.stop_all_subscriptions()
 	}
 
 	/// The block is pinned in the backend only once when the block's hash is first encountered.