feat/overseer: introduce closure init (#3775)

* feat/overseer: introduce closure init

Enables removal of the connected/disconnected overseer state.

* feat/overseer: allow replacement logic to access the original

Allows to re-use init-once types, which would otherwise error.

* feat/overseer: introduce external connector

Preparation for removal of `AllSubsystems`
which is another prerequisite for removing
the connect/disconnect state.

* fix/test: replace needs closure

* fixup

* simplify

* mea culpa

* all-subsystems-gen test

polkadot/node/core/candidate-validation/src/lib.rs

@@ -69,8 +69,10 @@ pub struct Config {
 
 /// The candidate validation subsystem.
 pub struct CandidateValidationSubsystem {
-	metrics: Metrics,
-	pvf_metrics: polkadot_node_core_pvf::Metrics,
+	#[allow(missing_docs)]
+	pub metrics: Metrics,
+	#[allow(missing_docs)]
+	pub pvf_metrics: polkadot_node_core_pvf::Metrics,
 	config: Config,
 }
 

polkadot/node/malus/src/variant-a.rs

@@ -34,9 +34,8 @@ use polkadot_cli::{
 
 // Import extra types relevant to the particular
 // subsystem.
-use polkadot_node_core_candidate_validation::{CandidateValidationSubsystem, Metrics};
+use polkadot_node_core_candidate_validation::CandidateValidationSubsystem;
 use polkadot_node_subsystem::messages::CandidateValidationMessage;
-use polkadot_node_subsystem_util::metrics::Metrics as _;
 
 // Filter wrapping related types.
 use malus::*;
@@ -88,14 +87,16 @@ impl OverseerGen for BehaveMaleficient {
 		// modify the subsystem(s) as needed:
 		let all_subsystems = create_default_subsystems(args)?.replace_candidate_validation(
 			// create the filtered subsystem
-			FilteredSubsystem::new(
-				CandidateValidationSubsystem::with_config(
-					candidate_validation_config,
-					Metrics::register(registry)?,
-					polkadot_node_core_pvf::Metrics::register(registry)?,
-				),
-				Skippy::default(),
-			),
+			|orig: CandidateValidationSubsystem| {
+				FilteredSubsystem::new(
+					CandidateValidationSubsystem::with_config(
+						candidate_validation_config,
+						orig.metrics,
+						orig.pvf_metrics,
+					),
+					Skippy::default(),
+				)
+			},
 		);
 
 		Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)

polkadot/node/overseer/all-subsystems-gen/src/lib.rs

@@ -118,10 +118,11 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
 
 			// generate an impl of `fn replace_#name`
 			for NameTyTup { field: replacable_item, ty: replacable_item_ty } in replacable_items {
-				let keeper = all_fields
+				let keeper = &all_fields
 					.iter()
 					.filter(|ntt| ntt.field != replacable_item)
-					.map(|ntt| ntt.field.clone());
+					.map(|ntt| ntt.field.clone())
+					.collect::<Vec<_>>();
 				let strukt_ty = strukt_ty.clone();
 				let fname = Ident::new(&format!("replace_{}", replacable_item), span);
 				// adjust the generics such that the appropriate member type is replaced
@@ -154,11 +155,27 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
 				additive.extend(quote! {
 					impl #orig_generics #strukt_ty #orig_generics {
 						#[doc = #msg]
-						pub fn #fname < NEW > (self, replacement: NEW) -> #strukt_ty #modified_generics {
+						pub fn #fname < NEW, F > (self, gen_replacement_fn: F) -> #strukt_ty #modified_generics
+							where
+								F: FnOnce(#replacable_item_ty) -> NEW,
+						{
+							let Self {
+								// To be replaced field:
+								#replacable_item,
+								// Fields to keep:
+								#(
+									#keeper,
+								)*
+							} = self;
+
+							// Some cases require that parts of the original are copied
+							// over, since they include a one time initialization.
+							let replacement = gen_replacement_fn(#replacable_item);
+
 							#strukt_ty :: #modified_generics {
 								#replacable_item: replacement,
 								#(
-									#keeper: self.#keeper,
+									#keeper,
 								)*
 							}
 						}

polkadot/node/overseer/all-subsystems-gen/tests/ui/err-01-generic-used-twice.rs

@@ -12,5 +12,5 @@ fn main() {
 		a: 0_u16,
 		b: 1_u16,
 	};
-	let _all = all.replace_a(77u8);
+	let _all = all.replace_a(|_| 77u8);
 }

polkadot/node/overseer/all-subsystems-gen/tests/ui/err-01-generic-used-twice.stderr

@@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems<u16>`
 5  | struct AllSubsystems<X> {
    | ----------------------- method `replace_a` not found for this
 ...
-15 |     let _all = all.replace_a(77u8);
+15 |     let _all = all.replace_a(|_| 77u8);
    |                    ^^^^^^^^^ method not found in `AllSubsystems<u16>`

polkadot/node/overseer/all-subsystems-gen/tests/ui/err-01-no-generic.rs

@@ -13,5 +13,5 @@ fn main() {
 		a: 0_f32,
 		b: 1_u16,
 	};
-	let _all = all.replace_a(77u8);
+	let _all = all.replace_a(|_| 77u8);
 }

polkadot/node/overseer/all-subsystems-gen/tests/ui/err-01-no-generic.stderr

@@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems` in th
 6  | struct AllSubsystems {
    | -------------------- method `replace_a` not found for this
 ...
-16 |     let _all = all.replace_a(77u8);
+16 |     let _all = all.replace_a(|_| 77u8);
    |                    ^^^^^^^^^ method not found in `AllSubsystems`

polkadot/node/overseer/all-subsystems-gen/tests/ui/err-01-no-generics.stderr

@@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems<u16>`
 6  | struct AllSubsystems<X> {
    | ----------------------- method `replace_a` not found for this
 ...
-16 |     let _all = all.replace_a(77u8);
+16 |     let _all = all.replace_a(|_| 77u8);
    |                    ^^^^^^^^^ method not found in `AllSubsystems<u16>`

polkadot/node/overseer/all-subsystems-gen/tests/ui/ok-01-w-generics.rs

@@ -13,5 +13,5 @@ fn main() {
 		a: 0u8,
 		b: 1u16,
 	};
-	let _all: AllSubsystems<_,_> = all.replace_a::<u32>(777_777u32);
+	let _all: AllSubsystems<_,_> = all.replace_a::<u32,_>(|_| 777_777u32);
 }

polkadot/node/overseer/examples/minimal-example.rs

@@ -170,8 +170,8 @@ fn main() {
 		});
 
 		let all_subsystems = AllSubsystems::<()>::dummy()
-			.replace_candidate_validation(Subsystem2)
-			.replace_candidate_backing(Subsystem1);
+			.replace_candidate_validation(|_| Subsystem2)
+			.replace_candidate_backing(|orig| orig);
 
 		let (overseer, _handle) =
 			Overseer::new(vec![], all_subsystems, None, AlwaysSupportsParachains, spawner).unwrap();

polkadot/node/overseer/overseer-gen/proc-macro/src/impl_builder.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use quote::quote;
+use quote::{format_ident, quote};
 use syn::Ident;
 
 use super::*;
@@ -27,8 +27,14 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 	let overseer_name = info.overseer_name.clone();
 	let builder = Ident::new(&(overseer_name.to_string() + "Builder"), overseer_name.span());
 	let handle = Ident::new(&(overseer_name.to_string() + "Handle"), overseer_name.span());
+	let connector = Ident::new(&(overseer_name.to_string() + "Connector"), overseer_name.span());
 
 	let subsystem_name = &info.subsystem_names_without_wip();
+	let subsystem_name_init_with = &info
+		.subsystem_names_without_wip()
+		.iter()
+		.map(|subsystem_name| format_ident!("{}_with", subsystem_name))
+		.collect::<Vec<_>>();
 	let builder_generic_ty = &info.builder_generic_types();
 
 	let channel_name = &info.channel_names_without_wip("");
@@ -106,10 +112,65 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 		/// Handle for an overseer.
 		pub type #handle = #support_crate ::metered::MeteredSender< #event >;
 
+		/// External connector.
+		pub struct #connector {
+			/// Publicly accessible handle, to be used for setting up
+			/// components that are _not_ subsystems but access is needed
+			/// due to other limitations.
+			///
+			/// For subsystems, use the `_with` variants of the builder.
+			handle: #handle,
+			/// The side consumed by the `spawned` side of the overseer pattern.
+			consumer: #support_crate ::metered::MeteredReceiver < #event >,
+		}
+
+		impl #connector {
+			/// Obtain access to the overseer handle.
+			pub fn as_handle_mut(&mut self) -> &mut #handle {
+				&mut self.handle
+			}
+			/// Obtain access to the overseer handle.
+			pub fn as_handle(&mut self) -> &#handle {
+				&self.handle
+			}
+		}
+
+		impl ::std::default::Default for #connector {
+			fn default() -> Self {
+				let (events_tx, events_rx) = #support_crate ::metered::channel::<
+					#event
+					>(SIGNAL_CHANNEL_CAPACITY);
+
+				Self {
+					handle: events_tx,
+					consumer: events_rx,
+				}
+			}
+		}
+
+		/// Convenience alias.
+		type SubsystemInitFn<T> = Box<dyn FnOnce(#handle) -> ::std::result::Result<T, #error_ty> >;
+
+		/// Init kind of a field of the overseer.
+		enum FieldInitMethod<T> {
+			/// Defer initialization to a point where the `handle` is available.
+			Fn(SubsystemInitFn<T>),
+			/// Directly initialize the subsystem with the given subsystem type `T`.
+			Value(T),
+			/// Subsystem field does not have value just yet.
+			Uninitialized
+		}
+
+		impl<T> ::std::default::Default for FieldInitMethod<T> {
+			fn default() -> Self {
+				Self::Uninitialized
+			}
+		}
+
 		#[allow(missing_docs)]
 		pub struct #builder #builder_generics {
 			#(
-				#subsystem_name : ::std::option::Option< #builder_generic_ty >,
+				#subsystem_name : FieldInitMethod< #builder_generic_ty >,
 			)*
 			#(
 				#baggage_name : ::std::option::Option< #baggage_ty >,
@@ -129,7 +190,7 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 
 				Self {
 				#(
-					#subsystem_name: None,
+					#subsystem_name: Default::default(),
 				)*
 				#(
 					#baggage_name: None,
@@ -152,7 +213,18 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 			#(
 				/// Specify the particular subsystem implementation.
 				pub fn #subsystem_name (mut self, subsystem: #builder_generic_ty ) -> Self {
-					self. #subsystem_name = Some( subsystem );
+					self. #subsystem_name = FieldInitMethod::Value( subsystem );
+					self
+				}
+
+				/// Specify the particular subsystem by giving a init function.
+				pub fn #subsystem_name_init_with <'a, F> (mut self, subsystem_init_fn: F ) -> Self
+				where
+					F: 'static + FnOnce(#handle) -> ::std::result::Result<#builder_generic_ty, #error_ty>,
+				{
+					self. #subsystem_name = FieldInitMethod::Fn(
+						Box::new(subsystem_init_fn) as SubsystemInitFn<#builder_generic_ty>
+					);
 					self
 				}
 			)*
@@ -166,13 +238,20 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 			)*
 
 			/// Complete the construction and create the overseer type.
-			pub fn build(mut self) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty>
-			{
-				let (events_tx, events_rx) = #support_crate ::metered::channel::<
-					#event
-				>(SIGNAL_CHANNEL_CAPACITY);
+			pub fn build(mut self) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty> {
+				let connector = #connector ::default();
+				self.build_with_connector(connector)
+			}
 
-				let handle: #handle = events_tx.clone();
+			/// Complete the construction and create the overseer type based on an existing `connector`.
+			pub fn build_with_connector(mut self, connector: #connector) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty>
+			{
+				let #connector {
+					handle: events_tx,
+					consumer: events_rx,
+				} = connector;
+
+				let handle = events_tx.clone();
 
 				let (to_overseer_tx, to_overseer_rx) = #support_crate ::metered::unbounded::<
 					ToOverseer
@@ -212,7 +291,12 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
 				#(
 					// TODO generate a builder pattern that ensures this
 					// TODO https://github.com/paritytech/polkadot/issues/3427
-					let #subsystem_name = self. #subsystem_name .expect("All subsystem must exist with the builder pattern.");
+					let #subsystem_name = match self. #subsystem_name {
+						 FieldInitMethod::Fn(func) => func(handle.clone())?,
+						FieldInitMethod::Value(val) => val,
+						FieldInitMethod::Uninitialized =>
+							panic!("All subsystems must exist with the builder pattern."),
+					};
 
 					let unbounded_meter = #channel_name_unbounded_rx.meter().clone();
 

polkadot/node/overseer/src/lib.rs

@@ -543,7 +543,7 @@ where
 	/// }
 	/// let spawner = sp_core::testing::TaskExecutor::new();
 	/// let all_subsystems = AllSubsystems::<()>::dummy()
-	///		.replace_candidate_validation(ValidationSubsystem);
+	///		.replace_candidate_validation(|_| ValidationSubsystem);
 	/// let (overseer, _handle) = Overseer::new(
 	///     vec![],
 	///     all_subsystems,

polkadot/node/overseer/src/tests.rs

@@ -161,8 +161,8 @@ fn overseer_works() {
 		let mut s2_rx = s2_rx.fuse();
 
 		let all_subsystems = AllSubsystems::<()>::dummy()
-			.replace_candidate_validation(TestSubsystem1(s1_tx))
-			.replace_candidate_backing(TestSubsystem2(s2_tx));
+			.replace_candidate_validation(move |_| TestSubsystem1(s1_tx))
+			.replace_candidate_backing(move |_| TestSubsystem2(s2_tx));
 
 		let (overseer, handle) =
 			Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
@@ -278,7 +278,8 @@ fn overseer_ends_on_subsystem_exit() {
 	let spawner = sp_core::testing::TaskExecutor::new();
 
 	executor::block_on(async move {
-		let all_subsystems = AllSubsystems::<()>::dummy().replace_candidate_backing(ReturnOnStart);
+		let all_subsystems =
+			AllSubsystems::<()>::dummy().replace_candidate_backing(|_| ReturnOnStart);
 		let (overseer, _handle) =
 			Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
 
@@ -379,8 +380,8 @@ fn overseer_start_stop_works() {
 		let (tx_5, mut rx_5) = metered::channel(64);
 		let (tx_6, mut rx_6) = metered::channel(64);
 		let all_subsystems = AllSubsystems::<()>::dummy()
-			.replace_candidate_validation(TestSubsystem5(tx_5))
-			.replace_candidate_backing(TestSubsystem6(tx_6));
+			.replace_candidate_validation(move |_| TestSubsystem5(tx_5))
+			.replace_candidate_backing(move |_| TestSubsystem6(tx_6));
 		let (overseer, handle) =
 			Overseer::new(vec![first_block], all_subsystems, None, MockSupportsParachains, spawner)
 				.unwrap();
@@ -475,8 +476,8 @@ fn overseer_finalize_works() {
 		let (tx_6, mut rx_6) = metered::channel(64);
 
 		let all_subsystems = AllSubsystems::<()>::dummy()
-			.replace_candidate_validation(TestSubsystem5(tx_5))
-			.replace_candidate_backing(TestSubsystem6(tx_6));
+			.replace_candidate_validation(move |_| TestSubsystem5(tx_5))
+			.replace_candidate_backing(move |_| TestSubsystem6(tx_6));
 
 		// start with two forks of different height.
 		let (overseer, handle) = Overseer::new(
@@ -570,7 +571,7 @@ fn do_not_send_empty_leaves_update_on_block_finalization() {
 		let (tx_5, mut rx_5) = metered::channel(64);
 
 		let all_subsystems =
-			AllSubsystems::<()>::dummy().replace_candidate_backing(TestSubsystem6(tx_5));
+			AllSubsystems::<()>::dummy().replace_candidate_backing(move |_| TestSubsystem6(tx_5));
 
 		let (overseer, handle) =
 			Overseer::new(Vec::new(), all_subsystems, None, MockSupportsParachains, spawner)

polkadot/node/subsystem-test-helpers/src/lib.rs

@@ -384,7 +384,7 @@ mod tests {
 		let spawner = sp_core::testing::TaskExecutor::new();
 		let (tx, rx) = mpsc::channel(2);
 		let all_subsystems =
-			AllSubsystems::<()>::dummy().replace_collator_protocol(ForwardSubsystem(tx));
+			AllSubsystems::<()>::dummy().replace_collator_protocol(|_| ForwardSubsystem(tx));
 		let (overseer, handle) = Overseer::new(
 			Vec::new(),
 			all_subsystems,