[Light] synchronize cht pruning with babe pruning

bin/node-template/node/src/service.rs

@@ -205,7 +205,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 
 /// Builds a new service for a light client.
 pub fn new_light(config: Configuration) -> Result<TaskManager, ServiceError> {
-	let (client, backend, keystore, mut task_manager, on_demand) =
+	let (client, backend, keystore, mut task_manager, on_demand, _shared_pruning_requirements) =
 		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(&config)?;
 
 	let transaction_pool = Arc::new(sc_transaction_pool::BasicPool::new_light(

bin/node/cli/src/service.rs

@@ -80,6 +80,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 		sc_consensus_babe::Config::get_or_compute(&*client)?,
 		grandpa_block_import,
 		client.clone(),
+		None,
 	)?;
 
 	let inherent_data_providers = sp_inherents::InherentDataProviders::new();
@@ -338,7 +339,7 @@ pub fn new_light_base(config: Configuration) -> Result<(
 	Arc<NetworkService<Block, <Block as BlockT>::Hash>>,
 	Arc<sc_transaction_pool::LightPool<Block, LightClient, sc_network::config::OnDemand<Block>>>
 ), ServiceError> {
-	let (client, backend, keystore, mut task_manager, on_demand) =
+	let (client, backend, keystore, mut task_manager, on_demand, shared_pruning_requirements) =
 		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(&config)?;
 
 	let select_chain = sc_consensus::LongestChain::new(backend.clone());
@@ -364,6 +365,7 @@ pub fn new_light_base(config: Configuration) -> Result<(
 		sc_consensus_babe::Config::get_or_compute(&*client)?,
 		grandpa_block_import,
 		client.clone(),
+		Some(&shared_pruning_requirements),
 	)?;
 
 	let inherent_data_providers = sp_inherents::InherentDataProviders::new();

client/api/src/backend.rs

@@ -536,3 +536,148 @@ pub fn changes_tries_state_at_block<'a, Block: BlockT>(
 		None => Ok(None),
 	}
 }
+
+#[derive(Clone)]
+/// Pruning requirement to share between multiple client component.
+///
+/// This allows pruning related synchronisation. For instance in light
+/// client we need to synchronize header pruning from CHT (every N blocks)
+/// with the pruning from consensus used (babe for instance require that
+/// its epoch headers are not pruned which works as long as the slot length
+/// is less than the CHT pruning window.
+/// Each compenent register at a given index (call are done by this order).
+///
+/// Note that this struct could be split in two different struct (provider without
+/// component and Component), depending on future usage of this shared info.
+pub struct SharedPruningRequirements<Block: BlockT> {
+	shared: Arc<RwLock<(Vec<ComponentPruningRequirements<Block>>, usize)>>,
+	component: Option<usize>,
+}
+
+impl<Block: BlockT> Default for SharedPruningRequirements<Block> {
+	fn default() -> Self {
+		SharedPruningRequirements {
+			shared: Arc::new(RwLock::new((Vec::new(), 0))),
+			component: None,
+		}
+	}
+}
+
+impl<Block: BlockT> SharedPruningRequirements<Block> {
+	/// Add a following requirement to apply.
+	/// Returns the shared requirement to use from this component.
+	pub fn next_instance(
+		&self,
+	) -> SharedPruningRequirements<Block> {
+		let req = ComponentPruningRequirements::default();
+		let index = {
+			let mut shared = self.shared.write();
+			let index = shared.0.len();
+			shared.0.push(req);
+			shared.1 += 1;
+			if shared.1 == usize::max_value() {
+				shared.1 = 1;
+				// marking all as changed is the safest
+				for component in shared.0.iter_mut() {
+					component.last_modified_check = 0;
+				}
+			}
+			index
+		};
+		let mut result = self.clone();
+		result.component = Some(index);
+		result
+	}
+
+	/// Check if some content was modified after last check.
+	pub fn modification_check(&self) -> bool {
+		if let Some(index) = self.component {
+			let modified = {
+				let read = self.shared.read();
+				if read.0[index].last_modified_check < read.1 {
+					Some(read.1.clone())	
+				} else {
+					None
+				}
+			};
+			if let Some(modified) = modified {
+				self.shared.write().0[index].last_modified_check = modified;
+				true
+			} else {
+				false
+			}
+		} else {
+			false
+		}
+	}
+
+	/// Resolve a finalized block headers to keep.
+	pub fn finalized_headers_needed(&self) -> PruningLimit<NumberFor<Block>> {
+		let mut result = PruningLimit::None;
+		for req in self.shared.read().0.iter() {
+			match &req.requires_finalized_header_up_to {
+				PruningLimit::Locked => return PruningLimit::Locked,
+				PruningLimit::None => (),
+				PruningLimit::Some(n) => {
+					if let PruningLimit::Some(p) = result {
+						if &p < n {
+							result = PruningLimit::Some(p);
+							continue;
+						}
+					}
+					result = PruningLimit::Some(n.clone());
+				},
+			}
+		}
+		result
+	}
+
+	/// Set new requirement on finalized headers.
+	/// Returns false if we do not have a handle on the shared requirement.
+	pub fn set_finalized_headers_needed(&self, limit: PruningLimit<NumberFor<Block>>) -> bool {
+		if let Some(index) = self.component {
+			let mut shared = self.shared.write();
+			if shared.0[index].requires_finalized_header_up_to != limit { 
+				shared.0[index].requires_finalized_header_up_to = limit;
+				shared.1 += 1;
+				if shared.1 == usize::max_value() {
+					shared.1 = 1;
+					// marking all as changed is the safest
+					for component in shared.0.iter_mut() {
+						component.last_modified_check = 0;
+					}
+				}
+			}
+			true
+		} else {
+			false
+		}
+	}
+}
+
+/// Individual pruning requirement for any substrate component.
+struct ComponentPruningRequirements<Block: BlockT> {
+	requires_finalized_header_up_to: PruningLimit<NumberFor<Block>>,
+	last_modified_check: usize,
+}
+
+impl<Block: BlockT> Default for ComponentPruningRequirements<Block> {
+	fn default() -> Self {
+		ComponentPruningRequirements {
+			requires_finalized_header_up_to: PruningLimit::None,
+			last_modified_check: 0,
+		}
+	}
+}
+
+#[derive(Eq, PartialEq)]
+/// Define a block number limit to apply.
+pub enum PruningLimit<N> {
+	/// Ignore.
+	None,
+	/// The component require at least this number
+	/// of unpruned elements.
+	Some(N),
+	/// We lock all pruning.
+	Locked,
+}

client/consensus/babe/rpc/src/lib.rs

@@ -264,6 +264,7 @@ mod tests {
 			config.clone(),
 			client.clone(),
 			client.clone(),
+			None,
 		).expect("can initialize block-import");
 
 		let epoch_changes = link.epoch_changes().clone();

client/consensus/babe/src/lib.rs

@@ -104,6 +104,7 @@ use sp_consensus::import_queue::{Verifier, BasicQueue, DefaultImportQueue, Cache
 use sc_client_api::{
 	backend::AuxStore,
 	BlockchainEvents, ProvideUncles,
+	SharedPruningRequirements, PruningLimit,
 };
 use sp_block_builder::BlockBuilder as BlockBuilderApi;
 use futures::channel::mpsc::{channel, Sender, Receiver};
@@ -1063,6 +1064,8 @@ pub struct BabeBlockImport<Block: BlockT, Client, I> {
 	client: Arc<Client>,
 	epoch_changes: SharedEpochChanges<Block, Epoch>,
 	config: Config,
+	previous_needed_height: Option<NumberFor<Block>>,
+	shared_pruning_requirements: Option<SharedPruningRequirements<Block>>,
 }
 
 impl<Block: BlockT, I: Clone, Client> Clone for BabeBlockImport<Block, Client, I> {
@@ -1072,6 +1075,8 @@ impl<Block: BlockT, I: Clone, Client> Clone for BabeBlockImport<Block, Client, I
 			client: self.client.clone(),
 			epoch_changes: self.epoch_changes.clone(),
 			config: self.config.clone(),
+			previous_needed_height: self.previous_needed_height.clone(),
+			shared_pruning_requirements: self.shared_pruning_requirements.clone(),
 		}
 	}
 }
@@ -1082,12 +1087,20 @@ impl<Block: BlockT, Client, I> BabeBlockImport<Block, Client, I> {
 		epoch_changes: SharedEpochChanges<Block, Epoch>,
 		block_import: I,
 		config: Config,
+		shared_pruning_requirements: Option<&SharedPruningRequirements<Block>>,
 	) -> Self {
+		let shared_pruning_requirements = shared_pruning_requirements.map(|shared| {
+			let req = shared.next_instance();
+			assert!(req.set_finalized_headers_needed(PruningLimit::Locked));
+			req
+		});
 		BabeBlockImport {
 			client,
 			inner: block_import,
 			epoch_changes,
 			config,
+			previous_needed_height: Some(Zero::zero()),
+			shared_pruning_requirements,
 		}
 	}
 }
@@ -1286,6 +1299,20 @@ impl<Block, Client, Inner> BlockImport<Block> for BabeBlockImport<Block, Client,
 					insert.iter().map(|(k, v)| (k.to_vec(), Some(v.to_vec())))
 				)
 			);
+
+			// new epoch limits, update shared pruning limit if needed
+			if let Some(shared_pruning_requirements) = self.shared_pruning_requirements.as_ref() {
+				let needed_height = epoch_changes.needed_parent_relation();
+				debug!(target: "babe", "Using prune limit {:?}", needed_height);
+				if needed_height != self.previous_needed_height {
+					self.previous_needed_height = needed_height.clone();
+					if let Some(height) = needed_height {
+						assert!(shared_pruning_requirements.set_finalized_headers_needed(PruningLimit::Some(height)));
+					} else {
+						assert!(shared_pruning_requirements.set_finalized_headers_needed(PruningLimit::Locked));
+					}
+				}
+			}
 		}
 
 		aux_schema::write_block_weight(
@@ -1385,6 +1412,7 @@ pub fn block_import<Client, Block: BlockT, I>(
 	config: Config,
 	wrapped_block_import: I,
 	client: Arc<Client>,
+	shared_pruning_requirements: Option<&SharedPruningRequirements<Block>>,
 ) -> ClientResult<(BabeBlockImport<Block, Client, I>, BabeLink<Block>)> where
 	Client: AuxStore + HeaderBackend<Block> + HeaderMetadata<Block, Error = sp_blockchain::Error>,
 {
@@ -1408,6 +1436,7 @@ pub fn block_import<Client, Block: BlockT, I>(
 		epoch_changes,
 		wrapped_block_import,
 		config,
+		shared_pruning_requirements,
 	);
 
 	Ok((import, link))

client/consensus/babe/src/tests.rs

@@ -278,6 +278,7 @@ impl TestNetFactory for BabeTestNet {
 			config,
 			client.clone(),
 			client.clone(),
+			None,
 		).expect("can initialize block-import");
 
 		let block_import = PanickingBlockImport(block_import);

client/consensus/epochs/src/lib.rs

@@ -18,7 +18,7 @@
 
 pub mod migration;
 
-use std::{sync::Arc, ops::Add, collections::BTreeMap, borrow::{Borrow, BorrowMut}};
+use std::{sync::Arc, ops::{Add, Sub}, collections::BTreeMap, borrow::{Borrow, BorrowMut}};
 use parking_lot::Mutex;
 use codec::{Encode, Decode};
 use fork_tree::ForkTree;
@@ -322,7 +322,7 @@ impl<Hash, Number, E: Epoch> Default for EpochChanges<Hash, Number, E> where
 
 impl<Hash, Number, E: Epoch> EpochChanges<Hash, Number, E> where
 	Hash: PartialEq + Ord + AsRef<[u8]> + AsMut<[u8]> + Copy,
-	Number: Ord + One + Zero + Add<Output=Number> + Copy,
+	Number: Ord + One + Zero + Sub<Output=Number> + Add<Output=Number> + Copy,
 {
 	/// Create a new epoch change.
 	pub fn new() -> Self {
@@ -640,6 +640,14 @@ impl<Hash, Number, E: Epoch> EpochChanges<Hash, Number, E> where
 	pub fn tree(&self) -> &ForkTree<Hash, Number, PersistedEpochHeader<E>> {
 		&self.inner
 	}
+
+	/// Get the minimal block height where we need to be able
+	/// to do `is_descendent_of` queries.
+	pub fn needed_parent_relation(&self) -> Option<Number> {
+		// TODO consider using tree accessor and remove this function
+		self.inner.lowest_node_number()
+			.map(|number| number - One::one())
+	}
 }
 
 /// Type alias to produce the epoch-changes tree from a block type.