orchestra license headers (#5588)

polkadot/node/orchestra/proc-macro/Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "orchestra-proc-macro"
+version = "0.0.1"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2021"
+description = "Generate an orchestra of subsystems from a single annotated struct definition."
+repository = "https://github.com/paritytech/polkadot"
+license = "MIT OR Apache-2.0"
+
+[package.metadata.docs.rs]
+targets = ["x86_64-unknown-linux-gnu"]
+
+[lib]
+proc-macro = true
+
+[dependencies]
+syn = { version = "1.0.95", features = ["full", "extra-traits"] }
+quote = "1.0.18"
+proc-macro2 = "1.0.37"
+proc-macro-crate = "1.1.3"
+expander = { version = "0.0.6", default-features = false }
+petgraph = "0.6.0"
+
+[dev-dependencies]
+assert_matches = "1.5"
+orchestra = { path = "../" }
+thiserror = "1"
+tracing = "0.1"
+
+[features]
+default = []
+# write the expanded version to a `orchestra-expansion.[a-f0-9]{10}.rs`
+# in the `OUT_DIR` as defined by `cargo` for the `expander` crate.
+expand = []
+# Create directional message consuming / outgoing graph.
+# Generates: `${OUT_DIR}/${orchestra|lowercase}-subsystem-messaging.dot`
+graph = []

polkadot/node/orchestra/proc-macro/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+	// populate OUT_DIR
+}

polkadot/node/orchestra/proc-macro/src/impl_builder.rs

@@ -0,0 +1,758 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use quote::{format_ident, quote};
+use syn::{parse_quote, Path, PathSegment};
+
+use super::*;
+
+fn recollect_without_idx<T: Clone>(x: &[T], idx: usize) -> Vec<T> {
+	let mut v = Vec::<T>::with_capacity(x.len().saturating_sub(1));
+	v.extend(x.iter().take(idx).cloned());
+	v.extend(x.iter().skip(idx + 1).cloned());
+	v
+}
+
+/// Implement a builder pattern for the `Orchestra`-type,
+/// which acts as the gateway to constructing the orchestra.
+///
+/// Elements tagged with `wip` are not covered here.
+pub(crate) fn impl_builder(info: &OrchestraInfo) -> proc_macro2::TokenStream {
+	let orchestra_name = info.orchestra_name.clone();
+	let builder = format_ident!("{}Builder", orchestra_name);
+	let handle = format_ident!("{}Handle", orchestra_name);
+	let connector = format_ident!("{}Connector", orchestra_name);
+	let subsystem_ctx_name = format_ident!("{}SubsystemContext", orchestra_name);
+
+	let subsystem_name = &info.subsystem_names_without_wip();
+	let subsystem_generics = &info.subsystem_generic_types();
+
+	let consumes = &info.consumes_without_wip();
+	let channel_name = &info.channel_names_without_wip("");
+	let channel_name_unbounded = &info.channel_names_without_wip("_unbounded");
+
+	let channel_name_tx = &info.channel_names_without_wip("_tx");
+	let channel_name_unbounded_tx = &info.channel_names_without_wip("_unbounded_tx");
+
+	let channel_name_rx = &info.channel_names_without_wip("_rx");
+	let channel_name_unbounded_rx = &info.channel_names_without_wip("_unbounded_rx");
+
+	let baggage_name = &info.baggage_names();
+	let baggage_generic_ty = &info.baggage_generic_types();
+
+	// State generics that are used to encode each field's status (Init/Missing)
+	let baggage_passthrough_state_generics = baggage_name
+		.iter()
+		.enumerate()
+		.map(|(idx, _)| format_ident!("InitStateBaggage{}", idx))
+		.collect::<Vec<_>>();
+	let subsystem_passthrough_state_generics = subsystem_name
+		.iter()
+		.enumerate()
+		.map(|(idx, _)| format_ident!("InitStateSubsystem{}", idx))
+		.collect::<Vec<_>>();
+
+	let error_ty = &info.extern_error_ty;
+
+	let support_crate = info.support_crate_name();
+
+	let blocking = &info
+		.subsystems()
+		.iter()
+		.map(|x| {
+			if x.blocking {
+				quote! { Blocking }
+			} else {
+				quote! { Regular }
+			}
+		})
+		.collect::<Vec<_>>();
+
+	// Helpers to use within quote! macros
+	let spawner_where_clause: syn::TypeParam = parse_quote! {
+			S: #support_crate ::Spawner
+	};
+
+	// Field names and real types
+	let field_name = subsystem_name.iter().chain(baggage_name.iter()).collect::<Vec<_>>();
+	let field_type = subsystem_generics
+		.iter()
+		.map(|ident| Path::from(PathSegment::from(ident.clone())))
+		.chain(info.baggage().iter().map(|bag| bag.field_ty.clone()))
+		.collect::<Vec<_>>();
+
+	// Setters logic
+
+	// For each setter we need to leave the remaining fields untouched and
+	// remove the field that we are fixing in this setter
+	// For subsystems `*_with` and `replace_*` setters are needed.
+	let subsystem_specific_setters =
+		info.subsystems().iter().filter(|ssf| !ssf.wip).enumerate().map(|(idx, ssf)| {
+			let field_name = &ssf.name;
+			let field_type = &ssf.generic;
+			let subsystem_consumes = &ssf.message_to_consume;
+			// Remove state generic for the item to be replaced. It sufficient to know `field_type` for
+			// that since we always move from `Init<#field_type>` to `Init<NEW>`.
+			let impl_subsystem_state_generics = recollect_without_idx(&subsystem_passthrough_state_generics[..], idx);
+
+			let field_name_with = format_ident!("{}_with", field_name);
+			let field_name_replace = format_ident!("replace_{}", field_name);
+
+			// In a setter we replace `Uninit<T>` with `Init<T>` leaving all other
+			// types as they are, as such they will be free generics.
+			let mut current_state_generics = subsystem_passthrough_state_generics
+				.iter()
+				.map(|subsystem_state_generic_ty| parse_quote!(#subsystem_state_generic_ty))
+				.collect::<Vec<syn::GenericArgument>>();
+			current_state_generics[idx] = parse_quote! { Missing<#field_type> };
+
+			// Generics that will be present after initializing a specific `Missing<_>` field.
+			let mut post_setter_state_generics = current_state_generics.clone();
+			post_setter_state_generics[idx] = parse_quote! { Init<#field_type> };
+
+			let mut post_replace_state_generics = current_state_generics.clone();
+			post_replace_state_generics[idx] = parse_quote! { Init<NEW> };
+
+			// All fields except the one we update with the new argument
+			// see the loop below.
+			let to_keep_subsystem_name = recollect_without_idx(&subsystem_name[..], idx);
+
+			let subsystem_sender_trait = format_ident!("{}SenderTrait", field_type);
+			let _subsystem_ctx_trait = format_ident!("{}ContextTrait", field_type);
+
+			let builder_where_clause = quote!{
+				#field_type : #support_crate::Subsystem< #subsystem_ctx_name< #subsystem_consumes >, #error_ty>,
+				< #subsystem_ctx_name < #subsystem_consumes > as #support_crate :: SubsystemContext>::Sender:
+					#subsystem_sender_trait,
+			};
+
+			// Create the field init `fn`
+			quote! {
+				impl <InitStateSpawner, #field_type, #( #impl_subsystem_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+				#builder <InitStateSpawner, #( #current_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+				where
+					#builder_where_clause
+				{
+					/// Specify the subsystem in the builder directly
+					pub fn #field_name (self, var: #field_type ) ->
+						#builder <InitStateSpawner, #( #post_setter_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+					{
+						#builder {
+							#field_name: Init::< #field_type >::Value(var),
+							#(
+								#to_keep_subsystem_name: self. #to_keep_subsystem_name,
+							)*
+							#(
+								#baggage_name: self. #baggage_name,
+							)*
+							spawner: self.spawner,
+
+							channel_capacity: self.channel_capacity,
+							signal_capacity: self.signal_capacity,
+						}
+					}
+					/// Specify the the initialization function for a subsystem
+					pub fn #field_name_with<'a, F>(self, subsystem_init_fn: F ) ->
+						#builder <InitStateSpawner, #( #post_setter_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+					where
+						F: 'static + FnOnce(#handle) ->
+							::std::result::Result<#field_type, #error_ty>,
+					{
+						let boxed_func = Init::<#field_type>::Fn(
+							Box::new(subsystem_init_fn) as SubsystemInitFn<#field_type>
+						);
+						#builder {
+							#field_name: boxed_func,
+							#(
+								#to_keep_subsystem_name: self. #to_keep_subsystem_name,
+							)*
+							#(
+								#baggage_name: self. #baggage_name,
+							)*
+							spawner: self.spawner,
+
+
+							channel_capacity: self.channel_capacity,
+							signal_capacity: self.signal_capacity,
+						}
+					}
+				}
+
+				impl <InitStateSpawner, #field_type, #( #impl_subsystem_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+				#builder <InitStateSpawner, #( #post_setter_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+				where
+					#builder_where_clause
+				{
+					/// Replace a subsystem by another implementation for the
+					/// consumable message type.
+					pub fn #field_name_replace<NEW, F>(self, gen_replacement_fn: F)
+						-> #builder <InitStateSpawner, #( #post_replace_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+					where
+						#field_type: 'static,
+						F: 'static + FnOnce(#field_type) -> NEW,
+						NEW: #support_crate ::Subsystem<#subsystem_ctx_name< #subsystem_consumes >, #error_ty>,
+					{
+						let replacement: Init<NEW> = match self.#field_name {
+							Init::Fn(fx) =>
+								Init::<NEW>::Fn(Box::new(move |handle: #handle| {
+								let orig = fx(handle)?;
+								Ok(gen_replacement_fn(orig))
+							})),
+							Init::Value(val) =>
+								Init::Value(gen_replacement_fn(val)),
+						};
+						#builder {
+							#field_name: replacement,
+							#(
+								#to_keep_subsystem_name: self. #to_keep_subsystem_name,
+							)*
+							#(
+								#baggage_name: self. #baggage_name,
+							)*
+							spawner: self.spawner,
+
+							channel_capacity: self.channel_capacity,
+							signal_capacity: self.signal_capacity,
+						}
+					}
+				}
+			}
+		});
+
+	// Produce setters for all baggage fields as well
+	let baggage_specific_setters = info.baggage().iter().enumerate().map(|(idx, bag_field)| {
+		// Baggage fields follow subsystems
+		let fname = &bag_field.field_name;
+		let field_type = &bag_field.field_ty;
+		let impl_baggage_state_generics = recollect_without_idx(&baggage_passthrough_state_generics[..], idx);
+		let to_keep_baggage_name = recollect_without_idx(&baggage_name[..], idx);
+
+		let mut pre_setter_generics = baggage_passthrough_state_generics
+			.iter()
+			.map(|gen_ty| parse_quote!(#gen_ty))
+			.collect::<Vec<syn::GenericArgument>>();
+		pre_setter_generics[idx] = parse_quote! { Missing<#field_type> };
+
+		let mut post_setter_generics = pre_setter_generics.clone();
+		post_setter_generics[idx] = parse_quote! { Init<#field_type> };
+
+		// Baggage can also be generic, so we need to include that to a signature
+		let preserved_baggage_generic = if bag_field.generic {
+			quote! {#field_type,}
+		} else {
+			TokenStream::new()
+		};
+
+		quote! {
+			impl <InitStateSpawner, #preserved_baggage_generic #( #subsystem_passthrough_state_generics, )* #( #impl_baggage_state_generics, )* >
+			#builder <InitStateSpawner, #( #subsystem_passthrough_state_generics, )* #( #pre_setter_generics, )* >
+			{
+				/// Specify the baggage in the builder when it was not initialized before
+				pub fn #fname (self, var: #field_type ) ->
+					#builder <InitStateSpawner, #( #subsystem_passthrough_state_generics, )* #( #post_setter_generics, )* >
+				{
+					#builder {
+						#fname: Init::<#field_type>::Value(var),
+						#(
+							#subsystem_name: self. #subsystem_name,
+						)*
+						#(
+							#to_keep_baggage_name: self. #to_keep_baggage_name,
+						)*
+						spawner: self.spawner,
+
+						channel_capacity: self.channel_capacity,
+						signal_capacity: self.signal_capacity,
+					}
+				}
+			}
+			impl <InitStateSpawner, #preserved_baggage_generic #( #subsystem_passthrough_state_generics, )* #( #impl_baggage_state_generics, )* >
+			#builder <InitStateSpawner, #( #subsystem_passthrough_state_generics, )* #( #post_setter_generics, )* > {
+				/// Specify the baggage in the builder when it has been previously initialized
+				pub fn #fname (self, var: #field_type ) ->
+					#builder <InitStateSpawner, #( #subsystem_passthrough_state_generics, )* #( #post_setter_generics, )* >
+				{
+					#builder {
+						#fname: Init::<#field_type>::Value(var),
+						#(
+							#subsystem_name: self. #subsystem_name,
+						)*
+						#(
+							#to_keep_baggage_name: self. #to_keep_baggage_name,
+						)*
+						spawner: self.spawner,
+
+						channel_capacity: self.channel_capacity,
+						signal_capacity: self.signal_capacity,
+					}
+				}
+			}
+		}
+	});
+
+	let event = &info.extern_event_ty;
+	let initialized_builder = format_ident!("Initialized{}", builder);
+	// The direct generics as expected by the `Orchestra<_,_,..>`, without states
+	let initialized_builder_generics = quote! {
+		S, #( #baggage_generic_ty, )* #( #subsystem_generics, )*
+	};
+
+	let builder_where_clause = info
+		.subsystems()
+		.iter()
+		.map(|ssf| {
+			let field_type = &ssf.generic;
+			let consumes = &ssf.message_to_consume;
+			let subsystem_sender_trait = format_ident!("{}SenderTrait", ssf.generic);
+			let subsystem_ctx_trait = format_ident!("{}ContextTrait", ssf.generic);
+			quote! {
+				#field_type:
+					#support_crate::Subsystem< #subsystem_ctx_name < #consumes>, #error_ty>,
+				<#subsystem_ctx_name< #consumes > as #subsystem_ctx_trait>::Sender:
+					#subsystem_sender_trait,
+				#subsystem_ctx_name< #consumes >:
+					#subsystem_ctx_trait,
+			}
+		})
+		.fold(TokenStream::new(), |mut ts, addendum| {
+			ts.extend(addendum);
+			ts
+		});
+
+	let mut ts = quote! {
+		/// Convenience alias.
+		type SubsystemInitFn<T> = Box<dyn FnOnce(#handle) -> ::std::result::Result<T, #error_ty> >;
+
+		/// Type for the initialized field of the orchestra builder
+		pub enum Init<T> {
+			/// Defer initialization to a point where the `handle` is available.
+			Fn(SubsystemInitFn<T>),
+			/// Directly initialize the subsystem with the given subsystem type `T`.
+			/// Also used for baggage fields
+			Value(T),
+		}
+		/// Type marker for the uninitialized field of the orchestra builder.
+		/// `PhantomData` is used for type hinting when creating uninitialized
+		/// builder, e.g. to avoid specifying the generics when instantiating
+		/// the `FooBuilder` when calling `Foo::builder()`
+		#[derive(Debug)]
+		pub struct Missing<T>(::core::marker::PhantomData<T>);
+
+		/// Trait used to mark fields status in a builder
+		trait OrchestraFieldState<T> {}
+
+		impl<T> OrchestraFieldState<T> for Init<T> {}
+		impl<T> OrchestraFieldState<T> for Missing<T> {}
+
+		impl<T> ::std::default::Default for Missing<T> {
+			fn default() -> Self {
+				Missing::<T>(::core::marker::PhantomData::<T>::default())
+			}
+		}
+
+		impl<S #(, #baggage_generic_ty )*> #orchestra_name <S #(, #baggage_generic_ty)*>
+		where
+			#spawner_where_clause,
+		{
+			/// Create a new orchestra utilizing the builder.
+			pub fn builder< #( #subsystem_generics),* >() ->
+				#builder<Missing<S> #(, Missing< #field_type > )* >
+			where
+				#builder_where_clause
+			{
+				#builder :: new()
+			}
+		}
+	};
+
+	ts.extend(quote! {
+		/// Handle for an orchestra.
+		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 orchestra pattern.
+			consumer: #support_crate ::metered::MeteredReceiver < #event >,
+		}
+
+		impl #connector {
+			/// Obtain access to the orchestra handle.
+			pub fn as_handle_mut(&mut self) -> &mut #handle {
+				&mut self.handle
+			}
+			/// Obtain access to the orchestra handle.
+			pub fn as_handle(&self) -> &#handle {
+				&self.handle
+			}
+			/// Obtain a clone of the handle.
+			pub fn handle(&self) -> #handle {
+				self.handle.clone()
+			}
+
+			/// Create a new connector with non-default event channel capacity.
+			pub fn with_event_capacity(event_channel_capacity: usize) -> Self {
+				let (events_tx, events_rx) = #support_crate ::metered::channel::<
+					#event
+					>(event_channel_capacity);
+
+				Self {
+					handle: events_tx,
+					consumer: events_rx,
+				}
+			}
+		}
+
+		impl ::std::default::Default for #connector {
+			fn default() -> Self {
+				Self::with_event_capacity(SIGNAL_CHANNEL_CAPACITY)
+			}
+		}
+	});
+
+	ts.extend(quote!{
+		/// Builder pattern to create compile time safe construction path.
+		pub struct #builder <InitStateSpawner, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+		{
+			#(
+				#subsystem_name: #subsystem_passthrough_state_generics,
+			)*
+			#(
+				#baggage_name: #baggage_passthrough_state_generics,
+			)*
+			spawner: InitStateSpawner,
+			// user provided runtime overrides,
+			// if `None`, the `orchestra(message_capacity=123,..)` is used
+			// or the default value.
+			channel_capacity: Option<usize>,
+			signal_capacity: Option<usize>,
+		}
+	});
+
+	ts.extend(quote! {
+		impl<#initialized_builder_generics> #builder<Missing<S>, #( Missing<#field_type>, )*>
+		{
+			/// Create a new builder pattern, with all fields being uninitialized.
+			fn new() -> Self {
+				// explicitly assure the required traits are implemented
+				fn trait_from_must_be_implemented<E>()
+				where
+					E: ::std::error::Error + Send + Sync + 'static + From<#support_crate ::OrchestraError>
+				{}
+
+				trait_from_must_be_implemented::< #error_ty >();
+
+				Self {
+					#(
+						#field_name: Missing::<#field_type>::default(),
+					)*
+					spawner: Missing::<S>::default(),
+
+					channel_capacity: None,
+					signal_capacity: None,
+				}
+			}
+		}
+	});
+
+	// Spawner setter
+	ts.extend(quote!{
+		impl<S, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+			#builder<Missing<S>, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+		where
+			#spawner_where_clause,
+		{
+			/// The `spawner` to use for spawning tasks.
+			pub fn spawner(self, spawner: S) -> #builder<
+				Init<S>,
+				#( #subsystem_passthrough_state_generics, )*
+				#( #baggage_passthrough_state_generics, )*
+			>
+			{
+				#builder {
+					#(
+						#field_name: self. #field_name,
+					)*
+					spawner: Init::<S>::Value(spawner),
+
+					channel_capacity: self.channel_capacity,
+					signal_capacity: self.signal_capacity,
+				}
+			}
+		}
+	});
+
+	// message and signal channel capacity
+	ts.extend(quote! {
+		impl<S, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+			#builder<Init<S>, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*>
+		where
+			#spawner_where_clause,
+		{
+			/// Set the interconnecting signal channel capacity.
+			pub fn signal_channel_capacity(mut self, capacity: usize) -> Self
+			{
+				self.signal_capacity = Some(capacity);
+				self
+			}
+
+			/// Set the interconnecting message channel capacities.
+			pub fn message_channel_capacity(mut self, capacity: usize) -> Self
+			{
+				self.channel_capacity = Some(capacity);
+				self
+			}
+		}
+	});
+
+	// Create the string literals for spawn.
+	let subsystem_name_str_literal = subsystem_name
+		.iter()
+		.map(|ident| proc_macro2::Literal::string(ident.to_string().replace("_", "-").as_str()))
+		.collect::<Vec<_>>();
+
+	ts.extend(quote! {
+		/// Type used to represent a builder where all fields are initialized and the orchestra could be constructed.
+		pub type #initialized_builder<#initialized_builder_generics> = #builder<Init<S>, #( Init<#field_type>, )*>;
+
+		// A builder specialization where all fields are set
+		impl<#initialized_builder_generics> #initialized_builder<#initialized_builder_generics>
+		where
+			#spawner_where_clause,
+			#builder_where_clause
+		{
+			/// Complete the construction and create the orchestra type.
+			pub fn build(self)
+				-> ::std::result::Result<(#orchestra_name<S, #( #baggage_generic_ty, )*>, #handle), #error_ty> {
+				let connector = #connector ::with_event_capacity(
+					self.signal_capacity.unwrap_or(SIGNAL_CHANNEL_CAPACITY)
+				);
+				self.build_with_connector(connector)
+			}
+
+			/// Complete the construction and create the orchestra type based on an existing `connector`.
+			pub fn build_with_connector(self, connector: #connector)
+				-> ::std::result::Result<(#orchestra_name<S, #( #baggage_generic_ty, )*>, #handle), #error_ty>
+			{
+				let #connector {
+					handle: events_tx,
+					consumer: events_rx,
+				} = connector;
+
+				let handle = events_tx.clone();
+
+				let (to_orchestra_tx, to_orchestra_rx) = #support_crate ::metered::unbounded::<
+					ToOrchestra
+				>();
+
+				#(
+					let (#channel_name_tx, #channel_name_rx)
+					=
+						#support_crate ::metered::channel::<
+							MessagePacket< #consumes >
+						>(
+							self.channel_capacity.unwrap_or(CHANNEL_CAPACITY)
+						);
+				)*
+
+				#(
+					let (#channel_name_unbounded_tx, #channel_name_unbounded_rx) =
+						#support_crate ::metered::unbounded::<
+							MessagePacket< #consumes >
+						>();
+				)*
+
+				let channels_out =
+					ChannelsOut {
+						#(
+							#channel_name: #channel_name_tx .clone(),
+						)*
+						#(
+							#channel_name_unbounded: #channel_name_unbounded_tx,
+						)*
+					};
+
+				let mut spawner = match self.spawner {
+					Init::Value(value) => value,
+					_ => unreachable!("Only ever init spawner as value. qed"),
+				};
+
+				let mut running_subsystems = #support_crate ::FuturesUnordered::<
+						BoxFuture<'static, ::std::result::Result<(), #error_ty > >
+					>::new();
+
+				#(
+					let #subsystem_name = match self. #subsystem_name {
+						Init::Fn(func) => func(handle.clone())?,
+						Init::Value(val) => val,
+					};
+
+					let unbounded_meter = #channel_name_unbounded_rx.meter().clone();
+
+					let message_rx: SubsystemIncomingMessages< #consumes > = #support_crate ::select(
+						#channel_name_rx, #channel_name_unbounded_rx
+					);
+					let (signal_tx, signal_rx) = #support_crate ::metered::channel(
+						self.signal_capacity.unwrap_or(SIGNAL_CHANNEL_CAPACITY)
+					);
+
+					let ctx = #subsystem_ctx_name::< #consumes >::new(
+						signal_rx,
+						message_rx,
+						channels_out.clone(),
+						to_orchestra_tx.clone(),
+						#subsystem_name_str_literal
+					);
+
+					let #subsystem_name: OrchestratedSubsystem< #consumes > =
+						spawn::<_,_, #blocking, _, _, _>(
+							&mut spawner,
+							#channel_name_tx,
+							signal_tx,
+							unbounded_meter,
+							ctx,
+							#subsystem_name,
+							#subsystem_name_str_literal,
+							&mut running_subsystems,
+						)?;
+				)*
+
+				use #support_crate ::StreamExt;
+
+				let to_orchestra_rx = to_orchestra_rx.fuse();
+				let orchestra = #orchestra_name {
+					#(
+						#subsystem_name,
+					)*
+
+					#(
+						#baggage_name: match self. #baggage_name {
+							Init::Value(val) => val,
+							_ => panic!("unexpected baggage initialization, must be value"),
+						},
+					)*
+
+					spawner,
+					running_subsystems,
+					events_rx,
+					to_orchestra_rx,
+				};
+
+				Ok((orchestra, handle))
+			}
+		}
+	});
+
+	ts.extend(baggage_specific_setters);
+	ts.extend(subsystem_specific_setters);
+	ts.extend(impl_task_kind(info));
+	ts
+}
+
+pub(crate) fn impl_task_kind(info: &OrchestraInfo) -> proc_macro2::TokenStream {
+	let signal = &info.extern_signal_ty;
+	let error_ty = &info.extern_error_ty;
+	let support_crate = info.support_crate_name();
+
+	let ts = quote! {
+		/// Task kind to launch.
+		pub trait TaskKind {
+			/// Spawn a task, it depends on the implementer if this is blocking or not.
+			fn launch_task<S: Spawner>(spawner: &mut S, task_name: &'static str, subsystem_name: &'static str, future: BoxFuture<'static, ()>);
+		}
+
+		#[allow(missing_docs)]
+		struct Regular;
+		impl TaskKind for Regular {
+			fn launch_task<S: Spawner>(spawner: &mut S, task_name: &'static str, subsystem_name: &'static str, future: BoxFuture<'static, ()>) {
+				spawner.spawn(task_name, Some(subsystem_name), future)
+			}
+		}
+
+		#[allow(missing_docs)]
+		struct Blocking;
+		impl TaskKind for Blocking {
+			fn launch_task<S: Spawner>(spawner: &mut S, task_name: &'static str, subsystem_name: &'static str, future: BoxFuture<'static, ()>) {
+				spawner.spawn_blocking(task_name, Some(subsystem_name), future)
+			}
+		}
+
+		/// Spawn task of kind `self` using spawner `S`.
+		pub fn spawn<S, M, TK, Ctx, E, SubSys>(
+			spawner: &mut S,
+			message_tx: #support_crate ::metered::MeteredSender<MessagePacket<M>>,
+			signal_tx: #support_crate ::metered::MeteredSender< #signal >,
+			// meter for the unbounded channel
+			unbounded_meter: #support_crate ::metered::Meter,
+			ctx: Ctx,
+			s: SubSys,
+			subsystem_name: &'static str,
+			futures: &mut #support_crate ::FuturesUnordered<BoxFuture<'static, ::std::result::Result<(), #error_ty> >>,
+		) -> ::std::result::Result<OrchestratedSubsystem<M>, #error_ty >
+		where
+			S: #support_crate ::Spawner,
+			M: std::fmt::Debug + Send + 'static,
+			TK: TaskKind,
+			Ctx: #support_crate ::SubsystemContext<Message=M>,
+			E: ::std::error::Error + Send + Sync + 'static + ::std::convert::From<#support_crate ::OrchestraError>,
+			SubSys: #support_crate ::Subsystem<Ctx, E>,
+		{
+			let #support_crate ::SpawnedSubsystem::<E> { future, name } = s.start(ctx);
+
+			let (tx, rx) = #support_crate ::oneshot::channel();
+
+			let fut = Box::pin(async move {
+				if let Err(e) = future.await {
+					#support_crate ::tracing::error!(subsystem=name, err = ?e, "subsystem exited with error");
+				} else {
+					#support_crate ::tracing::debug!(subsystem=name, "subsystem exited without an error");
+				}
+				let _ = tx.send(());
+			});
+
+			<TK as TaskKind>::launch_task(spawner, name, subsystem_name, fut);
+
+			futures.push(Box::pin(
+				rx.map(|e| {
+					#support_crate ::tracing::warn!(err = ?e, "dropping error");
+					Ok(())
+				})
+			));
+
+			let instance = Some(SubsystemInstance {
+				meters: #support_crate ::SubsystemMeters {
+					unbounded: unbounded_meter,
+					bounded: message_tx.meter().clone(),
+					signals: signal_tx.meter().clone(),
+				},
+				tx_signal: signal_tx,
+				tx_bounded: message_tx,
+				signals_received: 0,
+				name,
+			});
+
+			Ok(OrchestratedSubsystem {
+				instance,
+			})
+		}
+	};
+
+	ts
+}

polkadot/node/orchestra/proc-macro/src/impl_channels_out.rs

@@ -0,0 +1,142 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use quote::quote;
+use syn::Result;
+
+use super::*;
+
+/// Implement the helper type `ChannelsOut` and `MessagePacket<T>`.
+pub(crate) fn impl_channels_out_struct(info: &OrchestraInfo) -> Result<proc_macro2::TokenStream> {
+	let message_wrapper = info.message_wrapper.clone();
+
+	let channel_name = &info.channel_names_without_wip("");
+	let channel_name_unbounded = &info.channel_names_without_wip("_unbounded");
+
+	let consumes = &info.consumes_without_wip();
+
+	let consumes_variant = &info.variant_names_without_wip();
+	let unconsumes_variant = &info.variant_names_only_wip();
+
+	let support_crate = info.support_crate_name();
+
+	let ts = quote! {
+		/// Collection of channels to the individual subsystems.
+		///
+		/// Naming is from the point of view of the orchestra.
+		#[derive(Debug, Clone)]
+		pub struct ChannelsOut {
+			#(
+				/// Bounded channel sender, connected to a subsystem.
+				pub #channel_name:
+					#support_crate ::metered::MeteredSender<
+						MessagePacket< #consumes >
+					>,
+			)*
+
+			#(
+				/// Unbounded channel sender, connected to a subsystem.
+				pub #channel_name_unbounded:
+					#support_crate ::metered::UnboundedMeteredSender<
+						MessagePacket< #consumes >
+					>,
+			)*
+		}
+
+		#[allow(unreachable_code)]
+		// when no defined messages in enum
+		impl ChannelsOut {
+			/// Send a message via a bounded channel.
+			pub async fn send_and_log_error(
+				&mut self,
+				signals_received: usize,
+				message: #message_wrapper,
+			) {
+
+				let res: ::std::result::Result<_, _> = match message {
+				#(
+					#message_wrapper :: #consumes_variant ( inner ) => {
+						self. #channel_name .send(
+							#support_crate ::make_packet(signals_received, inner)
+						).await.map_err(|_| stringify!( #channel_name ))
+					}
+				)*
+					// subsystems that are wip
+				#(
+					#message_wrapper :: #unconsumes_variant ( _ ) => Ok(()),
+				)*
+					// dummy message type
+					#message_wrapper :: Empty => Ok(()),
+
+					#[allow(unreachable_patterns)]
+					// And everything that's not WIP but no subsystem consumes it
+					unused_msg => {
+						#support_crate :: tracing :: warn!("Nothing consumes {:?}", unused_msg);
+						Ok(())
+					}
+				};
+
+				if let Err(subsystem_name) = res {
+					#support_crate ::tracing::debug!(
+						target: LOG_TARGET,
+						"Failed to send (bounded) a message to {} subsystem",
+						subsystem_name
+					);
+				}
+			}
+
+			/// Send a message to another subsystem via an unbounded channel.
+			pub fn send_unbounded_and_log_error(
+				&self,
+				signals_received: usize,
+				message: #message_wrapper,
+			) {
+				let res: ::std::result::Result<_, _> = match message {
+				#(
+					#message_wrapper :: #consumes_variant (inner) => {
+						self. #channel_name_unbounded .unbounded_send(
+							#support_crate ::make_packet(signals_received, inner)
+						)
+						.map_err(|_| stringify!( #channel_name ))
+					},
+				)*
+					// subsystems that are wip
+				#(
+					#message_wrapper :: #unconsumes_variant ( _ ) => Ok(()),
+				)*
+					// dummy message type
+					#message_wrapper :: Empty => Ok(()),
+
+					// And everything that's not WIP but no subsystem consumes it
+					#[allow(unreachable_patterns)]
+					unused_msg => {
+						#support_crate :: tracing :: warn!("Nothing consumes {:?}", unused_msg);
+						Ok(())
+					}
+				};
+
+				if let Err(subsystem_name) = res {
+					#support_crate ::tracing::debug!(
+						target: LOG_TARGET,
+						"Failed to send_unbounded a message to {} subsystem",
+						subsystem_name
+					);
+				}
+			}
+		}
+
+	};
+	Ok(ts)
+}

polkadot/node/orchestra/proc-macro/src/impl_message_wrapper.rs

@@ -0,0 +1,85 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use quote::quote;
+use syn::{spanned::Spanned, Result};
+
+use super::*;
+
+/// Generates the wrapper type enum.
+pub(crate) fn impl_message_wrapper_enum(info: &OrchestraInfo) -> Result<proc_macro2::TokenStream> {
+	let consumes = info.any_message();
+	let consumes_variant = info.variant_names();
+
+	let outgoing = &info.outgoing_ty;
+
+	let message_wrapper = &info.message_wrapper;
+
+	let (outgoing_from_impl, outgoing_decl) = if let Some(outgoing) = outgoing {
+		let outgoing_variant = outgoing.get_ident().ok_or_else(|| {
+			syn::Error::new(
+				outgoing.span(),
+				"Missing identifier to use as enum variant for outgoing.",
+			)
+		})?;
+		(
+			quote! {
+				impl ::std::convert::From< #outgoing > for #message_wrapper {
+					fn from(message: #outgoing) -> Self {
+						#message_wrapper :: #outgoing_variant ( message )
+					}
+				}
+			},
+			quote! {
+				#outgoing_variant ( #outgoing ) ,
+			},
+		)
+	} else {
+		(TokenStream::new(), TokenStream::new())
+	};
+
+	let ts = quote! {
+		/// Generated message type wrapper over all possible messages
+		/// used by any subsystem.
+		#[allow(missing_docs)]
+		#[derive(Debug)]
+		pub enum #message_wrapper {
+			#(
+				#consumes_variant ( #consumes ),
+			)*
+			#outgoing_decl
+			// dummy message type
+			Empty,
+		}
+
+		impl ::std::convert::From< () > for #message_wrapper {
+			fn from(_: ()) -> Self {
+				#message_wrapper :: Empty
+			}
+		}
+
+		#(
+			impl ::std::convert::From< #consumes > for #message_wrapper {
+				fn from(message: #consumes) -> Self {
+					#message_wrapper :: #consumes_variant ( message )
+				}
+			}
+		)*
+
+		#outgoing_from_impl
+	};
+
+	Ok(ts)
+}

polkadot/node/orchestra/proc-macro/src/impl_orchestra.rs

@@ -0,0 +1,265 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use quote::quote;
+
+use super::*;
+
+pub(crate) fn impl_orchestra_struct(info: &OrchestraInfo) -> proc_macro2::TokenStream {
+	let message_wrapper = &info.message_wrapper.clone();
+	let orchestra_name = info.orchestra_name.clone();
+	let subsystem_name = &info.subsystem_names_without_wip();
+	let support_crate = info.support_crate_name();
+
+	let baggage_decl = &info.baggage_decl();
+
+	let baggage_generic_ty = &info.baggage_generic_types();
+
+	let generics = quote! {
+		< S, #( #baggage_generic_ty, )* >
+	};
+
+	let where_clause = quote! {
+		where
+			S: #support_crate ::Spawner,
+	};
+	// TODO add `where ..` clauses for baggage types
+	// TODO <https://github.com/paritytech/polkadot/issues/3427>
+
+	let consumes = &info.consumes_without_wip();
+	let consumes_variant = &info.variant_names_without_wip();
+	let unconsumes_variant = &info.variant_names_only_wip();
+
+	let signal_ty = &info.extern_signal_ty;
+
+	let error_ty = &info.extern_error_ty;
+
+	let event_ty = &info.extern_event_ty;
+
+	let message_channel_capacity = info.message_channel_capacity;
+	let signal_channel_capacity = info.signal_channel_capacity;
+
+	let log_target =
+		syn::LitStr::new(orchestra_name.to_string().to_lowercase().as_str(), orchestra_name.span());
+
+	let ts = quote! {
+		/// Capacity of a bounded message channel between orchestra and subsystem
+		/// but also for bounded channels between two subsystems.
+		const CHANNEL_CAPACITY: usize = #message_channel_capacity;
+
+		/// Capacity of a signal channel between a subsystem and the orchestra.
+		const SIGNAL_CHANNEL_CAPACITY: usize = #signal_channel_capacity;
+
+		/// The log target tag.
+		const LOG_TARGET: &'static str = #log_target;
+
+		/// The orchestra.
+		pub struct #orchestra_name #generics {
+
+			#(
+				/// A subsystem instance.
+				#subsystem_name: OrchestratedSubsystem< #consumes >,
+			)*
+
+			#(
+				/// A user specified addendum field.
+				#baggage_decl ,
+			)*
+
+			/// Responsible for driving the subsystem futures.
+			spawner: S,
+
+			/// The set of running subsystems.
+			running_subsystems: #support_crate ::FuturesUnordered<
+				BoxFuture<'static, ::std::result::Result<(), #error_ty>>
+			>,
+
+			/// Gather running subsystems' outbound streams into one.
+			to_orchestra_rx: #support_crate ::stream::Fuse<
+				#support_crate ::metered::UnboundedMeteredReceiver< #support_crate ::ToOrchestra >
+			>,
+
+			/// Events that are sent to the orchestra from the outside world.
+			events_rx: #support_crate ::metered::MeteredReceiver< #event_ty >,
+		}
+
+		impl #generics #orchestra_name #generics #where_clause {
+			/// Send the given signal, a termination signal, to all subsystems
+			/// and wait for all subsystems to go down.
+			///
+			/// The definition of a termination signal is up to the user and
+			/// implementation specific.
+			pub async fn wait_terminate(&mut self, signal: #signal_ty, timeout: ::std::time::Duration) -> ::std::result::Result<(), #error_ty > {
+				#(
+					::std::mem::drop(self. #subsystem_name .send_signal(signal.clone()).await);
+				)*
+				let _ = signal;
+
+				let mut timeout_fut = #support_crate ::Delay::new(
+						timeout
+					).fuse();
+
+				loop {
+					select! {
+						_ = self.running_subsystems.next() =>
+						if self.running_subsystems.is_empty() {
+							break;
+						},
+						_ = timeout_fut => break,
+						complete => break,
+					}
+				}
+
+				Ok(())
+			}
+
+			/// Broadcast a signal to all subsystems.
+			pub async fn broadcast_signal(&mut self, signal: #signal_ty) -> ::std::result::Result<(), #error_ty > {
+				#(
+					let _ = self. #subsystem_name .send_signal(signal.clone()).await;
+				)*
+				let _ = signal;
+
+				Ok(())
+			}
+
+			/// Route a particular message to a subsystem that consumes the message.
+			pub async fn route_message(&mut self, message: #message_wrapper, origin: &'static str) -> ::std::result::Result<(), #error_ty > {
+				match message {
+					#(
+						#message_wrapper :: #consumes_variant ( inner ) =>
+							OrchestratedSubsystem::< #consumes >::send_message2(&mut self. #subsystem_name, inner, origin ).await?,
+					)*
+					// subsystems that are still work in progress
+					#(
+						#message_wrapper :: #unconsumes_variant ( _ ) => {}
+					)*
+					#message_wrapper :: Empty => {}
+
+					// And everything that's not WIP but no subsystem consumes it
+					#[allow(unreachable_patterns)]
+					unused_msg => {
+						#support_crate :: tracing :: warn!("Nothing consumes {:?}", unused_msg);
+					}
+				}
+				Ok(())
+			}
+
+			/// Extract information from each subsystem.
+			pub fn map_subsystems<'a, Mapper, Output>(&'a self, mapper: Mapper)
+			-> Vec<Output>
+				where
+				#(
+					Mapper: MapSubsystem<&'a OrchestratedSubsystem< #consumes >, Output=Output>,
+				)*
+			{
+				vec![
+				#(
+					mapper.map_subsystem( & self. #subsystem_name ),
+				)*
+				]
+			}
+
+			/// Get access to internal task spawner.
+			pub fn spawner<'a> (&'a mut self) -> &'a mut S {
+				&mut self.spawner
+			}
+		}
+
+	};
+
+	ts
+}
+
+pub(crate) fn impl_orchestrated_subsystem(info: &OrchestraInfo) -> proc_macro2::TokenStream {
+	let signal = &info.extern_signal_ty;
+	let error_ty = &info.extern_error_ty;
+	let support_crate = info.support_crate_name();
+
+	let ts = quote::quote! {
+		/// A subsystem that the orchestrator orchestrates.
+		///
+		/// Ties together the [`Subsystem`] itself and it's running instance
+		/// (which may be missing if the [`Subsystem`] is not running at the moment
+		/// for whatever reason).
+		///
+		/// [`Subsystem`]: trait.Subsystem.html
+		pub struct OrchestratedSubsystem<M> {
+			/// The instance.
+			pub instance: std::option::Option<
+				#support_crate ::SubsystemInstance<M, #signal>
+			>,
+		}
+
+		impl<M> OrchestratedSubsystem<M> {
+			/// Send a message to the wrapped subsystem.
+			///
+			/// If the inner `instance` is `None`, nothing is happening.
+			pub async fn send_message2(&mut self, message: M, origin: &'static str) -> ::std::result::Result<(), #error_ty > {
+				const MESSAGE_TIMEOUT: Duration = Duration::from_secs(10);
+
+				if let Some(ref mut instance) = self.instance {
+					match instance.tx_bounded.send(MessagePacket {
+						signals_received: instance.signals_received,
+						message: message.into(),
+					}).timeout(MESSAGE_TIMEOUT).await
+					{
+						None => {
+							#support_crate ::tracing::error!(
+								target: LOG_TARGET,
+								%origin,
+								"Subsystem {} appears unresponsive.",
+								instance.name,
+							);
+							Err(#error_ty :: from(
+								#support_crate ::OrchestraError::SubsystemStalled(instance.name)
+							))
+						}
+						Some(res) => res.map_err(Into::into),
+					}
+				} else {
+					Ok(())
+				}
+			}
+
+			/// Send a signal to the wrapped subsystem.
+			///
+			/// If the inner `instance` is `None`, nothing is happening.
+			pub async fn send_signal(&mut self, signal: #signal) -> ::std::result::Result<(), #error_ty > {
+				const SIGNAL_TIMEOUT: ::std::time::Duration = ::std::time::Duration::from_secs(10);
+
+				if let Some(ref mut instance) = self.instance {
+					match instance.tx_signal.send(signal).timeout(SIGNAL_TIMEOUT).await {
+						None => {
+							Err(#error_ty :: from(
+								#support_crate ::OrchestraError::SubsystemStalled(instance.name)
+							))
+						}
+						Some(res) => {
+							let res = res.map_err(Into::into);
+							if res.is_ok() {
+								instance.signals_received += 1;
+							}
+							res
+						}
+					}
+				} else {
+					Ok(())
+				}
+			}
+		}
+	};
+	ts
+}

polkadot/node/orchestra/proc-macro/src/impl_subsystem_ctx_sender.rs

@@ -0,0 +1,711 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use proc_macro2::TokenStream;
+use quote::quote;
+use syn::{Ident, Path, Result, Type};
+
+use petgraph::{
+	dot::{self, Dot},
+	graph::NodeIndex,
+	visit::EdgeRef,
+	Direction,
+};
+use std::collections::HashMap;
+
+use super::*;
+
+/// Render a graphviz (aka dot graph) to a file.
+fn graphviz(
+	graph: &petgraph::Graph<Ident, Path>,
+	dest: &mut impl std::io::Write,
+) -> std::io::Result<()> {
+	let config = &[dot::Config::EdgeNoLabel, dot::Config::NodeNoLabel][..];
+	let dot = Dot::with_attr_getters(
+		graph,
+		config,
+		&|_graph, edge| -> String {
+			format!(
+				r#"label="{}""#,
+				edge.weight().get_ident().expect("Must have a trailing identifier. qed")
+			)
+		},
+		&|_graph, (_node_index, subsystem_name)| -> String {
+			format!(r#"label="{}""#, subsystem_name,)
+		},
+	);
+	dest.write_all(format!("{:?}", &dot).as_bytes())?;
+	Ok(())
+}
+
+/// Generates all subsystem types and related accumulation traits.
+pub(crate) fn impl_subsystem_types_all(info: &OrchestraInfo) -> Result<TokenStream> {
+	let mut ts = TokenStream::new();
+
+	let orchestra_name = &info.orchestra_name;
+	let span = orchestra_name.span();
+	let all_messages_wrapper = &info.message_wrapper;
+	let support_crate = info.support_crate_name();
+	let signal_ty = &info.extern_signal_ty;
+	let error_ty = &info.extern_error_ty;
+
+	// create a directed graph with all the subsystems as nodes and the messages as edges
+	// key is always the message path, values are node indices in the graph and the subsystem generic identifier
+	// store the message path and the source sender, both in the graph as well as identifier
+	let mut outgoing_lut = HashMap::<&Path, Vec<(Ident, NodeIndex)>>::with_capacity(128);
+	// same for consuming the incoming messages
+	let mut consuming_lut = HashMap::<&Path, (Ident, NodeIndex)>::with_capacity(128);
+
+	// Ident = Node = subsystem generic names
+	// Path = Edge = messages
+	let mut graph = petgraph::Graph::<Ident, Path>::new();
+
+	// prepare the full index of outgoing and source subsystems
+	for ssf in info.subsystems() {
+		let node_index = graph.add_node(ssf.generic.clone());
+		for outgoing in ssf.messages_to_send.iter() {
+			outgoing_lut
+				.entry(outgoing)
+				.or_default()
+				.push((ssf.generic.clone(), node_index));
+		}
+		consuming_lut.insert(&ssf.message_to_consume, (ssf.generic.clone(), node_index));
+	}
+
+	for (message_ty, (_consuming_subsystem_ident, consuming_node_index)) in consuming_lut.iter() {
+		// match the outgoing ones that were registered above with the consumed message
+		if let Some(origin_subsystems) = outgoing_lut.get(message_ty) {
+			for (_origin_subsystem_ident, sending_node_index) in origin_subsystems.iter() {
+				graph.add_edge(*sending_node_index, *consuming_node_index, (*message_ty).clone());
+			}
+		}
+	}
+
+	// All outgoing edges are now usable to derive everything we need
+	for node_index in graph.node_indices() {
+		let subsystem_name = graph[node_index].to_string();
+		let outgoing_wrapper = Ident::new(&(subsystem_name + "OutgoingMessages"), span);
+
+		// cannot be a hashmap, duplicate keys and sorting required
+		// maps outgoing messages to the subsystem that consumes it
+		let outgoing_to_consumer = graph
+			.edges_directed(node_index, Direction::Outgoing)
+			.map(|edge| {
+				let message_ty = edge.weight();
+				let subsystem_generic_consumer = graph[edge.target()].clone();
+				Ok((to_variant(message_ty, span.clone())?, subsystem_generic_consumer))
+			})
+			.collect::<Result<Vec<(Ident, Ident)>>>()?;
+
+		// Split it for usage with quote
+		let outgoing_variant = outgoing_to_consumer.iter().map(|x| x.0.clone()).collect::<Vec<_>>();
+		let subsystem_generic = outgoing_to_consumer.into_iter().map(|x| x.1).collect::<Vec<_>>();
+
+		ts.extend(quote! {
+			impl ::std::convert::From< #outgoing_wrapper > for #all_messages_wrapper {
+				fn from(message: #outgoing_wrapper) -> Self {
+					match message {
+					#(
+						#outgoing_wrapper :: #outgoing_variant ( msg ) => #all_messages_wrapper :: #subsystem_generic ( msg ),
+					)*
+						#outgoing_wrapper :: Empty => #all_messages_wrapper :: Empty,
+						// And everything that's not WIP but no subsystem consumes it
+						#[allow(unreachable_patterns)]
+						unused_msg => {
+							#support_crate :: tracing :: warn!("Nothing consumes {:?}", unused_msg);
+							#all_messages_wrapper :: Empty
+						}
+					}
+				}
+			}
+		})
+	}
+
+	// Dump the graph to file.
+	if cfg!(feature = "graph") || true {
+		let path = std::path::PathBuf::from(env!("OUT_DIR"))
+			.join(orchestra_name.to_string().to_lowercase() + "-subsystem-messaging.dot");
+		if let Err(e) = std::fs::OpenOptions::new()
+			.truncate(true)
+			.create(true)
+			.write(true)
+			.open(&path)
+			.and_then(|mut f| graphviz(&graph, &mut f))
+		{
+			eprintln!("Failed to write dot graph to {}: {:?}", path.display(), e);
+		} else {
+			println!("Wrote dot graph to {}", path.display());
+		}
+	}
+
+	let subsystem_sender_name = &Ident::new(&(orchestra_name.to_string() + "Sender"), span);
+	let subsystem_ctx_name = &Ident::new(&(orchestra_name.to_string() + "SubsystemContext"), span);
+	ts.extend(impl_subsystem_context(info, &subsystem_sender_name, &subsystem_ctx_name));
+
+	ts.extend(impl_associate_outgoing_messages_trait(&all_messages_wrapper));
+
+	ts.extend(impl_subsystem_sender(
+		support_crate,
+		info.subsystems().iter().map(|ssf| {
+			let outgoing_wrapper =
+				Ident::new(&(ssf.generic.to_string() + "OutgoingMessages"), span);
+			outgoing_wrapper
+		}),
+		&all_messages_wrapper,
+		&subsystem_sender_name,
+	));
+
+	// Create all subsystem specific types, one by one
+	for ssf in info.subsystems() {
+		let subsystem_name = ssf.generic.to_string();
+		let outgoing_wrapper = &Ident::new(&(subsystem_name.clone() + "OutgoingMessages"), span);
+
+		let subsystem_ctx_trait = &Ident::new(&(subsystem_name.clone() + "ContextTrait"), span);
+		let subsystem_sender_trait = &Ident::new(&(subsystem_name.clone() + "SenderTrait"), span);
+
+		ts.extend(impl_per_subsystem_helper_traits(
+			info,
+			subsystem_ctx_name,
+			subsystem_ctx_trait,
+			subsystem_sender_name,
+			subsystem_sender_trait,
+			&ssf.message_to_consume,
+			&ssf.messages_to_send,
+			outgoing_wrapper,
+		));
+
+		ts.extend(impl_associate_outgoing_messages(&ssf.message_to_consume, &outgoing_wrapper));
+
+		ts.extend(impl_wrapper_enum(&outgoing_wrapper, ssf.messages_to_send.as_slice())?);
+	}
+
+	// impl the emtpy tuple handling for tests
+	let empty_tuple: Type = parse_quote! { () };
+	ts.extend(impl_subsystem_context_trait_for(
+		empty_tuple.clone(),
+		&[],
+		empty_tuple,
+		all_messages_wrapper,
+		subsystem_ctx_name,
+		subsystem_sender_name,
+		support_crate,
+		signal_ty,
+		error_ty,
+	));
+
+	Ok(ts)
+}
+
+/// Extract the final component of the message type path as used in the `#[subsystem(consumes: path::to::Foo)]` annotation.
+fn to_variant(path: &Path, span: Span) -> Result<Ident> {
+	let ident = path
+		.segments
+		.last()
+		.ok_or_else(|| syn::Error::new(span, "Path is empty, but it must end with an identifier"))
+		.map(|segment| segment.ident.clone())?;
+	Ok(ident)
+}
+
+/// Converts the outgoing message types to variants.
+///
+/// Note: Commonly this is `${X}Message` becomes `${X}OutgoingMessages::${X}Message`
+/// where for `AllMessages` it would be `AllMessages::${X}`.
+fn to_variants(message_types: &[Path], span: Span) -> Result<Vec<Ident>> {
+	let variants: Vec<_> =
+		Result::from_iter(message_types.into_iter().map(|path| to_variant(path, span.clone())))?;
+	Ok(variants)
+}
+
+/// Generates the wrapper type enum, no bells or whistles.
+pub(crate) fn impl_wrapper_enum(wrapper: &Ident, message_types: &[Path]) -> Result<TokenStream> {
+	// The message types are path based, each of them must finish with a type
+	// and as such we do this upfront.
+	let variants = to_variants(message_types, wrapper.span())?;
+
+	let ts = quote! {
+		#[allow(missing_docs)]
+		#[derive(Debug)]
+		pub enum #wrapper {
+			#(
+				#variants ( #message_types ),
+			)*
+			Empty,
+		}
+
+		#(
+			impl ::std::convert::From< #message_types > for #wrapper {
+				fn from(message: #message_types) -> Self {
+					#wrapper :: #variants ( message )
+				}
+			}
+		)*
+
+		// Useful for unit and integration tests:
+		impl ::std::convert::From< () > for #wrapper {
+			fn from(_message: ()) -> Self {
+				#wrapper :: Empty
+			}
+		}
+	};
+	Ok(ts)
+}
+
+/// Create the subsystem sender type and implements `trait SubsystemSender`
+/// for the `#outgoing_wrappers: From<OutgoingMessage>` with the proper associated types.
+pub(crate) fn impl_subsystem_sender(
+	support_crate: &Path,
+	outgoing_wrappers: impl IntoIterator<Item = Ident>,
+	all_messages_wrapper: &Ident,
+	subsystem_sender_name: &Ident,
+) -> TokenStream {
+	let mut ts = quote! {
+		/// Connector to send messages towards all subsystems,
+		/// while tracking the which signals where already received.
+		#[derive(Debug)]
+		pub struct #subsystem_sender_name < OutgoingWrapper > {
+			/// Collection of channels to all subsystems.
+			channels: ChannelsOut,
+			/// Systemwide tick for which signals were received by all subsystems.
+			signals_received: SignalsReceived,
+			/// Keep that marker around.
+			_phantom: ::core::marker::PhantomData< OutgoingWrapper >,
+		}
+
+		// can't derive due to `PhantomData` and `OutgoingWrapper` not being
+		// bounded by `Clone`.
+		impl<OutgoingWrapper> std::clone::Clone for #subsystem_sender_name < OutgoingWrapper > {
+			fn clone(&self) -> Self {
+				Self {
+					channels: self.channels.clone(),
+					signals_received: self.signals_received.clone(),
+					_phantom: ::core::marker::PhantomData::default(),
+				}
+			}
+		}
+	};
+
+	// Create the same for a wrapping enum:
+	//
+	// 1. subsystem specific `*OutgoingMessages`-type
+	// 2. orchestra-global-`AllMessages`-type
+	let wrapped = |outgoing_wrapper: &TokenStream| {
+		quote! {
+			#[#support_crate ::async_trait]
+			impl<OutgoingMessage> SubsystemSender< OutgoingMessage > for #subsystem_sender_name < #outgoing_wrapper >
+			where
+				OutgoingMessage: Send + 'static,
+				#outgoing_wrapper: ::std::convert::From<OutgoingMessage> + Send,
+				#all_messages_wrapper: ::std::convert::From< #outgoing_wrapper > + Send,
+			{
+				async fn send_message(&mut self, msg: OutgoingMessage)
+				{
+					self.channels.send_and_log_error(
+						self.signals_received.load(),
+						<#all_messages_wrapper as ::std::convert::From<_>> ::from (
+							<#outgoing_wrapper as ::std::convert::From<_>> :: from ( msg )
+						)
+					).await;
+				}
+
+				async fn send_messages<I>(&mut self, msgs: I)
+				where
+					I: IntoIterator<Item=OutgoingMessage> + Send,
+					I::IntoIter: Iterator<Item=OutgoingMessage> + Send,
+				{
+					for msg in msgs {
+						self.send_message( msg ).await;
+					}
+				}
+
+				fn send_unbounded_message(&mut self, msg: OutgoingMessage)
+				{
+					self.channels.send_unbounded_and_log_error(
+						self.signals_received.load(),
+						<#all_messages_wrapper as ::std::convert::From<_>> ::from (
+							<#outgoing_wrapper as ::std::convert::From<_>> :: from ( msg )
+						)
+					);
+				}
+			}
+		}
+	};
+
+	for outgoing_wrapper in outgoing_wrappers {
+		ts.extend(wrapped(&quote! {
+			#outgoing_wrapper
+		}));
+	}
+
+	ts.extend(wrapped(&quote! {
+		()
+	}));
+
+	ts
+}
+
+/// Define the `trait AssociateOutgoing` and implement it for  `#all_messages_wrapper` and `()`.
+pub(crate) fn impl_associate_outgoing_messages_trait(all_messages_wrapper: &Ident) -> TokenStream {
+	quote! {
+		/// Binds a generated type covering all declared outgoing messages,
+		/// which implements `#generated_outgoing: From<M>` for all annotated types
+		/// of a particular subsystem.
+		///
+		/// Note: This works because there is a 1?:1 relation between consumed messages and subsystems.
+		pub trait AssociateOutgoing: ::std::fmt::Debug + Send {
+			/// The associated _outgoing_ messages for a subsystem that _consumes_ the message `Self`.
+			type OutgoingMessages: Into< #all_messages_wrapper > + ::std::fmt::Debug + Send;
+		}
+
+		// Helper for tests, where nothing is ever sent.
+		impl AssociateOutgoing for () {
+			type OutgoingMessages = ();
+		}
+
+		// Helper for tests, allows sending of arbitrary messages give
+		// an test context.
+		impl AssociateOutgoing for #all_messages_wrapper {
+			type OutgoingMessages = #all_messages_wrapper ;
+		}
+	}
+}
+
+/// Implement `AssociateOutgoing` for `#consumes` being handled by a particular subsystem.
+///
+/// Binds the outgoing messages to the inbound message.
+///
+/// Note: Works, since there is a 1:1 relation between inbound message type and subsystem declarations.
+/// Note: A workaround until default associated types work in `rustc`.
+pub(crate) fn impl_associate_outgoing_messages(
+	consumes: &Path,
+	outgoing_wrapper: &Ident,
+) -> TokenStream {
+	quote! {
+		impl AssociateOutgoing for #outgoing_wrapper {
+			type OutgoingMessages = #outgoing_wrapper;
+		}
+
+		impl AssociateOutgoing for #consumes {
+			type OutgoingMessages = #outgoing_wrapper;
+		}
+	}
+}
+
+/// Implement `trait SubsystemContext` for a particular subsystem context,
+/// that is generated by the proc-macro too.
+pub(crate) fn impl_subsystem_context_trait_for(
+	consumes: Type,
+	outgoing: &[Type],
+	outgoing_wrapper: Type,
+	all_messages_wrapper: &Ident,
+	subsystem_ctx_name: &Ident,
+	subsystem_sender_name: &Ident,
+	support_crate: &Path,
+	signal: &Path,
+	error_ty: &Path,
+) -> TokenStream {
+	// impl the subsystem context trait
+	let where_clause = quote! {
+		#consumes: AssociateOutgoing + ::std::fmt::Debug + Send + 'static,
+		#all_messages_wrapper: From< #outgoing_wrapper >,
+		#all_messages_wrapper: From< #consumes >,
+		#outgoing_wrapper: #( From< #outgoing > )+*,
+	};
+
+	quote! {
+		#[#support_crate ::async_trait]
+		impl #support_crate ::SubsystemContext for #subsystem_ctx_name < #consumes >
+		where
+			#where_clause
+		{
+			type Message = #consumes;
+			type Signal = #signal;
+			type OutgoingMessages = #outgoing_wrapper;
+			type Sender = #subsystem_sender_name < #outgoing_wrapper >;
+			type Error = #error_ty;
+
+			async fn try_recv(&mut self) -> ::std::result::Result<Option<FromOrchestra< Self::Message, #signal>>, ()> {
+				match #support_crate ::poll!(self.recv()) {
+					#support_crate ::Poll::Ready(msg) => Ok(Some(msg.map_err(|_| ())?)),
+					#support_crate ::Poll::Pending => Ok(None),
+				}
+			}
+
+			async fn recv(&mut self) -> ::std::result::Result<FromOrchestra<Self::Message, #signal>, #error_ty> {
+				loop {
+					// If we have a message pending an orchestra signal, we only poll for signals
+					// in the meantime.
+					if let Some((needs_signals_received, msg)) = self.pending_incoming.take() {
+						if needs_signals_received <= self.signals_received.load() {
+							return Ok( #support_crate ::FromOrchestra::Communication { msg });
+						} else {
+							self.pending_incoming = Some((needs_signals_received, msg));
+
+							// wait for next signal.
+							let signal = self.signals.next().await
+								.ok_or(#support_crate ::OrchestraError::Context(
+									"Signal channel is terminated and empty."
+									.to_owned()
+								))?;
+
+							self.signals_received.inc();
+							return Ok( #support_crate ::FromOrchestra::Signal(signal))
+						}
+					}
+
+					let mut await_message = self.messages.next().fuse();
+					let mut await_signal = self.signals.next().fuse();
+					let signals_received = self.signals_received.load();
+					let pending_incoming = &mut self.pending_incoming;
+
+					// Otherwise, wait for the next signal or incoming message.
+					let from_orchestra = #support_crate ::futures::select_biased! {
+						signal = await_signal => {
+							let signal = signal
+								.ok_or( #support_crate ::OrchestraError::Context(
+									"Signal channel is terminated and empty."
+									.to_owned()
+								))?;
+
+							#support_crate ::FromOrchestra::Signal(signal)
+						}
+						msg = await_message => {
+							let packet = msg
+								.ok_or( #support_crate ::OrchestraError::Context(
+									"Message channel is terminated and empty."
+									.to_owned()
+								))?;
+
+							if packet.signals_received > signals_received {
+								// wait until we've received enough signals to return this message.
+								*pending_incoming = Some((packet.signals_received, packet.message));
+								continue;
+							} else {
+								// we know enough to return this message.
+								#support_crate ::FromOrchestra::Communication { msg: packet.message}
+							}
+						}
+					};
+
+					if let #support_crate ::FromOrchestra::Signal(_) = from_orchestra {
+						self.signals_received.inc();
+					}
+
+					return Ok(from_orchestra);
+				}
+			}
+
+			fn sender(&mut self) -> &mut Self::Sender {
+				&mut self.to_subsystems
+			}
+
+			fn spawn(&mut self, name: &'static str, s: Pin<Box<dyn Future<Output = ()> + Send>>)
+				-> ::std::result::Result<(), #error_ty>
+			{
+				self.to_orchestra.unbounded_send(#support_crate ::ToOrchestra::SpawnJob {
+					name,
+					subsystem: Some(self.name()),
+					s,
+				}).map_err(|_| #support_crate ::OrchestraError::TaskSpawn(name))?;
+				Ok(())
+			}
+
+			fn spawn_blocking(&mut self, name: &'static str, s: Pin<Box<dyn Future<Output = ()> + Send>>)
+				-> ::std::result::Result<(), #error_ty>
+			{
+				self.to_orchestra.unbounded_send(#support_crate ::ToOrchestra::SpawnBlockingJob {
+					name,
+					subsystem: Some(self.name()),
+					s,
+				}).map_err(|_| #support_crate ::OrchestraError::TaskSpawn(name))?;
+				Ok(())
+			}
+		}
+	}
+}
+
+/// Implement the additional subsystem accumulation traits, for simplified usage,
+/// i.e. `${Subsystem}SenderTrait` and `${Subsystem}ContextTrait`.
+pub(crate) fn impl_per_subsystem_helper_traits(
+	info: &OrchestraInfo,
+	subsystem_ctx_name: &Ident,
+	subsystem_ctx_trait: &Ident,
+	subsystem_sender_name: &Ident,
+	subsystem_sender_trait: &Ident,
+	consumes: &Path,
+	outgoing: &[Path],
+	outgoing_wrapper: &Ident,
+) -> TokenStream {
+	let all_messages_wrapper = &info.message_wrapper;
+	let signal_ty = &info.extern_signal_ty;
+	let error_ty = &info.extern_error_ty;
+	let support_crate = info.support_crate_name();
+
+	let mut ts = TokenStream::new();
+
+	// Create a helper trait bound of all outgoing messages, and the generated wrapper type
+	// for ease of use within subsystems:
+	let acc_sender_trait_bounds = quote! {
+		#support_crate ::SubsystemSender< #outgoing_wrapper >
+		#(
+			+ #support_crate ::SubsystemSender< #outgoing >
+		)*
+			+ #support_crate ::SubsystemSender< () >
+			+ Send
+			+ 'static
+	};
+
+	ts.extend(quote! {
+		/// A abstracting trait for usage with subsystems.
+		pub trait #subsystem_sender_trait : #acc_sender_trait_bounds
+		{}
+
+		impl<T> #subsystem_sender_trait for T
+		where
+			T: #acc_sender_trait_bounds
+		{}
+	});
+
+	// Create a helper accumulated per subsystem trait bound:
+	let where_clause = quote! {
+		#consumes: AssociateOutgoing + ::std::fmt::Debug + Send + 'static,
+		#all_messages_wrapper: From< #outgoing_wrapper >,
+		#all_messages_wrapper: From< #consumes >,
+		#all_messages_wrapper: From< () >,
+		#outgoing_wrapper: #( From< #outgoing > )+*,
+		#outgoing_wrapper: From< () >,
+	};
+
+	ts.extend(quote! {
+		/// Accumulative trait for a particular subsystem wrapper.
+		pub trait #subsystem_ctx_trait : SubsystemContext <
+			Message = #consumes,
+			Signal = #signal_ty,
+			OutgoingMessages = #outgoing_wrapper,
+			// Sender,
+			Error = #error_ty,
+		>
+		where
+			#where_clause
+			<Self as SubsystemContext>::Sender:
+				#subsystem_sender_trait
+				+ #acc_sender_trait_bounds,
+		{
+			/// Sender.
+			type Sender: #subsystem_sender_trait;
+		}
+
+		impl<T> #subsystem_ctx_trait for T
+		where
+			T: SubsystemContext <
+				Message = #consumes,
+				Signal = #signal_ty,
+				OutgoingMessages = #outgoing_wrapper,
+				// Sender
+				Error = #error_ty,
+			>,
+			#where_clause
+			<T as SubsystemContext>::Sender:
+				#subsystem_sender_trait
+				+ #acc_sender_trait_bounds,
+		{
+			type Sender = <T as SubsystemContext>::Sender;
+		}
+	});
+
+	ts.extend(impl_subsystem_context_trait_for(
+		parse_quote! { #consumes },
+		&Vec::from_iter(outgoing.iter().map(|path| {
+			parse_quote! { #path }
+		})),
+		parse_quote! { #outgoing_wrapper },
+		all_messages_wrapper,
+		subsystem_ctx_name,
+		subsystem_sender_name,
+		support_crate,
+		signal_ty,
+		error_ty,
+	));
+	ts
+}
+
+/// Generate the subsystem context type and provide `fn new` on it.
+///
+/// Note: The generated `fn new` is used by the [builder pattern](../impl_builder.rs).
+pub(crate) fn impl_subsystem_context(
+	info: &OrchestraInfo,
+	subsystem_sender_name: &Ident,
+	subsystem_ctx_name: &Ident,
+) -> TokenStream {
+	let signal_ty = &info.extern_signal_ty;
+	let support_crate = info.support_crate_name();
+
+	let ts = quote! {
+		/// A context type that is given to the [`Subsystem`] upon spawning.
+		/// It can be used by [`Subsystem`] to communicate with other [`Subsystem`]s
+		/// or to spawn it's [`SubsystemJob`]s.
+		///
+		/// [`Orchestra`]: struct.Orchestra.html
+		/// [`Subsystem`]: trait.Subsystem.html
+		/// [`SubsystemJob`]: trait.SubsystemJob.html
+		#[derive(Debug)]
+		#[allow(missing_docs)]
+		pub struct #subsystem_ctx_name<M: AssociateOutgoing + Send + 'static> {
+			signals: #support_crate ::metered::MeteredReceiver< #signal_ty >,
+			messages: SubsystemIncomingMessages< M >,
+			to_subsystems: #subsystem_sender_name < <M as AssociateOutgoing>::OutgoingMessages >,
+			to_orchestra: #support_crate ::metered::UnboundedMeteredSender<
+				#support_crate ::ToOrchestra
+				>,
+			signals_received: SignalsReceived,
+			pending_incoming: Option<(usize, M)>,
+			name: &'static str
+		}
+
+		impl<M> #subsystem_ctx_name <M>
+		where
+			M: AssociateOutgoing + Send + 'static,
+		{
+			/// Create a new context.
+			fn new(
+				signals: #support_crate ::metered::MeteredReceiver< #signal_ty >,
+				messages: SubsystemIncomingMessages< M >,
+				to_subsystems: ChannelsOut,
+				to_orchestra: #support_crate ::metered::UnboundedMeteredSender<#support_crate:: ToOrchestra>,
+				name: &'static str
+			) -> Self {
+				let signals_received = SignalsReceived::default();
+				#subsystem_ctx_name :: <M> {
+					signals,
+					messages,
+					to_subsystems: #subsystem_sender_name :: < <M as AssociateOutgoing>::OutgoingMessages > {
+						channels: to_subsystems,
+						signals_received: signals_received.clone(),
+						_phantom: ::core::marker::PhantomData::default(),
+					},
+					to_orchestra,
+					signals_received,
+					pending_incoming: None,
+					name
+				}
+			}
+
+			fn name(&self) -> &'static str {
+				self.name
+			}
+		}
+	};
+
+	ts
+}

polkadot/node/orchestra/proc-macro/src/lib.rs

@@ -0,0 +1,88 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use proc_macro2::{Ident, Span, TokenStream};
+use syn::{parse_quote, spanned::Spanned, Path};
+
+mod impl_builder;
+mod impl_channels_out;
+mod impl_message_wrapper;
+mod impl_orchestra;
+mod impl_subsystem_ctx_sender;
+mod orchestra;
+mod parse;
+mod subsystem;
+
+#[cfg(test)]
+mod tests;
+
+use impl_builder::*;
+use impl_channels_out::*;
+use impl_message_wrapper::*;
+use impl_orchestra::*;
+use impl_subsystem_ctx_sender::*;
+use parse::*;
+
+use self::{orchestra::*, subsystem::*};
+
+/// Obtain the support crate `Path` as `TokenStream`.
+pub(crate) fn support_crate() -> Result<Path, proc_macro_crate::Error> {
+	Ok(if cfg!(test) {
+		parse_quote! {crate}
+	} else {
+		use proc_macro_crate::{crate_name, FoundCrate};
+		let crate_name = crate_name("orchestra")?;
+		match crate_name {
+			FoundCrate::Itself => parse_quote! {crate},
+			FoundCrate::Name(name) => Ident::new(&name, Span::call_site()).into(),
+		}
+	})
+}
+
+#[proc_macro_attribute]
+pub fn orchestra(
+	attr: proc_macro::TokenStream,
+	item: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+	let attr: TokenStream = attr.into();
+	let item: TokenStream = item.into();
+	impl_orchestra_gen(attr, item)
+		.unwrap_or_else(|err| err.to_compile_error())
+		.into()
+}
+
+#[proc_macro_attribute]
+pub fn subsystem(
+	attr: proc_macro::TokenStream,
+	item: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+	let attr: TokenStream = attr.into();
+	let item: TokenStream = item.into();
+	impl_subsystem_context_trait_bounds(attr, item, MakeSubsystem::ImplSubsystemTrait)
+		.unwrap_or_else(|err| err.to_compile_error())
+		.into()
+}
+
+#[proc_macro_attribute]
+pub fn contextbounds(
+	attr: proc_macro::TokenStream,
+	item: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+	let attr: TokenStream = attr.into();
+	let item: TokenStream = item.into();
+	impl_subsystem_context_trait_bounds(attr, item, MakeSubsystem::AddContextTraitBounds)
+		.unwrap_or_else(|err| err.to_compile_error())
+		.into()
+}

polkadot/node/orchestra/proc-macro/src/orchestra.rs

@@ -0,0 +1,66 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use proc_macro2::TokenStream;
+use syn::{parse2, Result};
+
+use super::{parse::*, *};
+
+pub(crate) fn impl_orchestra_gen(
+	attr: TokenStream,
+	orig: TokenStream,
+) -> Result<proc_macro2::TokenStream> {
+	let args: OrchestraAttrArgs = parse2(attr)?;
+	let message_wrapper = args.message_wrapper;
+
+	let of: OrchestraGuts = parse2(orig)?;
+
+	let support_crate = support_crate().expect("The crate this macro is run for, includes the proc-macro support as dependency, otherwise it could not be run in the first place. qed");
+	let info = OrchestraInfo {
+		support_crate,
+		subsystems: of.subsystems,
+		baggage: of.baggage,
+		orchestra_name: of.name,
+		message_wrapper,
+		message_channel_capacity: args.message_channel_capacity,
+		signal_channel_capacity: args.signal_channel_capacity,
+		extern_event_ty: args.extern_event_ty,
+		extern_signal_ty: args.extern_signal_ty,
+		extern_error_ty: args.extern_error_ty,
+		outgoing_ty: args.outgoing_ty,
+	};
+
+	let mut additive = impl_orchestra_struct(&info);
+	additive.extend(impl_builder(&info));
+
+	additive.extend(impl_orchestrated_subsystem(&info));
+	additive.extend(impl_channels_out_struct(&info));
+	additive.extend(impl_subsystem_types_all(&info)?);
+
+	additive.extend(impl_message_wrapper_enum(&info)?);
+
+	let ts = expander::Expander::new("orchestra-expansion")
+		.add_comment("Generated orchestra code by `#[orchestra(..)]`".to_owned())
+		.dry(!cfg!(feature = "expand"))
+		.verbose(true)
+		// once all our needed format options are available on stable
+		// we should enabled this again, until then too many warnings
+		// are generated
+		// .fmt(expander::Edition::_2021)
+		.write_to_out_dir(additive)
+		.expect("Expander does not fail due to IO in OUT_DIR. qed");
+
+	Ok(ts)
+}

polkadot/node/orchestra/proc-macro/src/parse/mod.rs

@@ -0,0 +1,38 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod kw {
+	syn::custom_keyword!(event);
+	syn::custom_keyword!(signal);
+	syn::custom_keyword!(error);
+	syn::custom_keyword!(outgoing);
+	syn::custom_keyword!(gen);
+	syn::custom_keyword!(signal_capacity);
+	syn::custom_keyword!(message_capacity);
+	syn::custom_keyword!(subsystem);
+	syn::custom_keyword!(prefix);
+}
+
+mod parse_orchestra_attr;
+mod parse_orchestra_struct;
+
+mod parse_subsystem_attr;
+
+#[cfg(test)]
+mod tests;
+
+pub(crate) use self::{parse_orchestra_attr::*, parse_orchestra_struct::*};
+
+pub(crate) use self::parse_subsystem_attr::*;

polkadot/node/orchestra/proc-macro/src/parse/parse_orchestra_attr.rs

@@ -0,0 +1,190 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::kw;
+use proc_macro2::Span;
+use quote::{quote, ToTokens};
+use std::collections::{hash_map::RandomState, HashMap};
+use syn::{
+	parse::{Parse, ParseBuffer},
+	punctuated::Punctuated,
+	spanned::Spanned,
+	Error, Ident, LitInt, Path, Result, Token,
+};
+
+#[derive(Clone, Debug)]
+enum OrchestraAttrItem {
+	ExternEventType { tag: kw::event, eq_token: Token![=], value: Path },
+	ExternOrchestraSignalType { tag: kw::signal, eq_token: Token![=], value: Path },
+	ExternErrorType { tag: kw::error, eq_token: Token![=], value: Path },
+	OutgoingType { tag: kw::outgoing, eq_token: Token![=], value: Path },
+	MessageWrapperName { tag: kw::gen, eq_token: Token![=], value: Ident },
+	SignalChannelCapacity { tag: kw::signal_capacity, eq_token: Token![=], value: usize },
+	MessageChannelCapacity { tag: kw::message_capacity, eq_token: Token![=], value: usize },
+}
+
+impl ToTokens for OrchestraAttrItem {
+	fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
+		let ts = match self {
+			Self::ExternEventType { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::ExternOrchestraSignalType { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::ExternErrorType { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::OutgoingType { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::MessageWrapperName { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::SignalChannelCapacity { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::MessageChannelCapacity { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+		};
+		tokens.extend(ts.into_iter());
+	}
+}
+
+impl Parse for OrchestraAttrItem {
+	fn parse(input: &ParseBuffer) -> Result<Self> {
+		let lookahead = input.lookahead1();
+		if lookahead.peek(kw::event) {
+			Ok(OrchestraAttrItem::ExternEventType {
+				tag: input.parse::<kw::event>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::signal) {
+			Ok(OrchestraAttrItem::ExternOrchestraSignalType {
+				tag: input.parse::<kw::signal>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::error) {
+			Ok(OrchestraAttrItem::ExternErrorType {
+				tag: input.parse::<kw::error>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::outgoing) {
+			Ok(OrchestraAttrItem::OutgoingType {
+				tag: input.parse::<kw::outgoing>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::gen) {
+			Ok(OrchestraAttrItem::MessageWrapperName {
+				tag: input.parse::<kw::gen>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::signal_capacity) {
+			Ok(OrchestraAttrItem::SignalChannelCapacity {
+				tag: input.parse::<kw::signal_capacity>()?,
+				eq_token: input.parse()?,
+				value: input.parse::<LitInt>()?.base10_parse::<usize>()?,
+			})
+		} else if lookahead.peek(kw::message_capacity) {
+			Ok(OrchestraAttrItem::MessageChannelCapacity {
+				tag: input.parse::<kw::message_capacity>()?,
+				eq_token: input.parse()?,
+				value: input.parse::<LitInt>()?.base10_parse::<usize>()?,
+			})
+		} else {
+			Err(lookahead.error())
+		}
+	}
+}
+
+/// Attribute arguments
+#[derive(Clone, Debug)]
+pub(crate) struct OrchestraAttrArgs {
+	pub(crate) message_wrapper: Ident,
+	pub(crate) extern_event_ty: Path,
+	pub(crate) extern_signal_ty: Path,
+	pub(crate) extern_error_ty: Path,
+	pub(crate) outgoing_ty: Option<Path>,
+	pub(crate) signal_channel_capacity: usize,
+	pub(crate) message_channel_capacity: usize,
+}
+
+macro_rules! extract_variant {
+	($unique:expr, $variant:ident ; default = $fallback:expr) => {
+		extract_variant!($unique, $variant).unwrap_or_else(|| $fallback)
+	};
+	($unique:expr, $variant:ident ; err = $err:expr) => {
+		extract_variant!($unique, $variant).ok_or_else(|| Error::new(Span::call_site(), $err))
+	};
+	($unique:expr, $variant:ident) => {
+		$unique.values().find_map(|item| {
+			if let OrchestraAttrItem::$variant { value, .. } = item {
+				Some(value.clone())
+			} else {
+				None
+			}
+		})
+	};
+}
+
+impl Parse for OrchestraAttrArgs {
+	fn parse(input: &ParseBuffer) -> Result<Self> {
+		let items: Punctuated<OrchestraAttrItem, Token![,]> =
+			input.parse_terminated(OrchestraAttrItem::parse)?;
+
+		let mut unique = HashMap::<
+			std::mem::Discriminant<OrchestraAttrItem>,
+			OrchestraAttrItem,
+			RandomState,
+		>::default();
+		for item in items {
+			if let Some(first) = unique.insert(std::mem::discriminant(&item), item.clone()) {
+				let mut e = Error::new(
+					item.span(),
+					format!("Duplicate definition of orchestra generation type found"),
+				);
+				e.combine(Error::new(first.span(), "previously defined here."));
+				return Err(e)
+			}
+		}
+
+		let signal_channel_capacity =
+			extract_variant!(unique, SignalChannelCapacity; default = 64_usize);
+		let message_channel_capacity =
+			extract_variant!(unique, MessageChannelCapacity; default = 1024_usize);
+
+		let error = extract_variant!(unique, ExternErrorType; err = "Must declare the orchestra error type via `error=..`.")?;
+		let event = extract_variant!(unique, ExternEventType; err = "Must declare the orchestra event type via `event=..`.")?;
+		let signal = extract_variant!(unique, ExternOrchestraSignalType; err = "Must declare the orchestra signal type via `signal=..`.")?;
+		let message_wrapper = extract_variant!(unique, MessageWrapperName; err = "Must declare the orchestra generated wrapping message type via `gen=..`.")?;
+		let outgoing = extract_variant!(unique, OutgoingType);
+
+		Ok(OrchestraAttrArgs {
+			signal_channel_capacity,
+			message_channel_capacity,
+			extern_event_ty: event,
+			extern_signal_ty: signal,
+			extern_error_ty: error,
+			outgoing_ty: outgoing,
+			message_wrapper,
+		})
+	}
+}

polkadot/node/orchestra/proc-macro/src/parse/parse_orchestra_struct.rs

@@ -0,0 +1,540 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use proc_macro2::{Span, TokenStream};
+use std::collections::{hash_map::RandomState, HashMap, HashSet};
+use syn::{
+	parenthesized,
+	parse::{Parse, ParseStream},
+	punctuated::Punctuated,
+	spanned::Spanned,
+	token::Bracket,
+	AttrStyle, Error, Field, FieldsNamed, GenericParam, Ident, ItemStruct, Path, Result, Token,
+	Type, Visibility,
+};
+
+use quote::{quote, ToTokens};
+
+mod kw {
+	syn::custom_keyword!(wip);
+	syn::custom_keyword!(blocking);
+	syn::custom_keyword!(consumes);
+	syn::custom_keyword!(sends);
+}
+
+#[derive(Clone, Debug)]
+pub(crate) enum SubSysAttrItem {
+	/// The subsystem is still a work in progress
+	/// and should not be communicated with.
+	Wip(kw::wip),
+	/// The subsystem is blocking and requires to be
+	/// spawned on an exclusive thread.
+	Blocking(kw::blocking),
+	/// Message to be sent by this subsystem.
+	Sends(Sends),
+	/// Message to be consumed by this subsystem.
+	Consumes(Consumes),
+}
+
+impl Parse for SubSysAttrItem {
+	fn parse(input: ParseStream) -> Result<Self> {
+		let lookahead = input.lookahead1();
+		Ok(if lookahead.peek(kw::wip) {
+			Self::Wip(input.parse::<kw::wip>()?)
+		} else if lookahead.peek(kw::blocking) {
+			Self::Blocking(input.parse::<kw::blocking>()?)
+		} else if lookahead.peek(kw::sends) {
+			Self::Sends(input.parse::<Sends>()?)
+		} else {
+			Self::Consumes(input.parse::<Consumes>()?)
+		})
+	}
+}
+
+impl ToTokens for SubSysAttrItem {
+	fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
+		let ts = match self {
+			Self::Wip(wip) => {
+				quote! { #wip }
+			},
+			Self::Blocking(blocking) => {
+				quote! { #blocking }
+			},
+			Self::Sends(_) => {
+				quote! {}
+			},
+			Self::Consumes(_) => {
+				quote! {}
+			},
+		};
+		tokens.extend(ts.into_iter());
+	}
+}
+
+/// A field of the struct annotated with
+/// `#[subsystem(A, B, C)]`
+#[derive(Clone, Debug)]
+pub(crate) struct SubSysField {
+	/// Name of the field.
+	pub(crate) name: Ident,
+	/// Generate generic type name for the `AllSubsystems` type
+	/// which is also used `#wrapper_message :: #variant` variant
+	/// part.
+	pub(crate) generic: Ident,
+	/// Type of message to be consumed by the subsystem.
+	pub(crate) message_to_consume: Path,
+	/// Types of messages to be sent by the subsystem.
+	pub(crate) messages_to_send: Vec<Path>,
+	/// If the subsystem implementation is blocking execution and hence
+	/// has to be spawned on a separate thread or thread pool.
+	pub(crate) blocking: bool,
+	/// The subsystem is a work in progress.
+	/// Avoids dispatching `Wrapper` type messages, but generates the variants.
+	/// Does not require the subsystem to be instantiated with the builder pattern.
+	pub(crate) wip: bool,
+}
+
+fn try_type_to_path(ty: Type, span: Span) -> Result<Path> {
+	match ty {
+		Type::Path(path) => Ok(path.path),
+		_ => Err(Error::new(span, "Type must be a path expression.")),
+	}
+}
+
+macro_rules! extract_variant {
+	($unique:expr, $variant:ident ; default = $fallback:expr) => {
+		extract_variant!($unique, $variant).unwrap_or_else(|| $fallback)
+	};
+	($unique:expr, $variant:ident ; err = $err:expr) => {
+		extract_variant!($unique, $variant).ok_or_else(|| Error::new(Span::call_site(), $err))
+	};
+	($unique:expr, $variant:ident take) => {
+		$unique.values().find_map(|item| {
+			if let SubSysAttrItem::$variant(value) = item {
+				Some(value.clone())
+			} else {
+				None
+			}
+		})
+	};
+	($unique:expr, $variant:ident) => {
+		$unique.values().find_map(|item| {
+			if let SubSysAttrItem::$variant(_) = item {
+				Some(true)
+			} else {
+				None
+			}
+		})
+	};
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct Sends {
+	#[allow(dead_code)]
+	pub(crate) keyword_sends: kw::sends,
+	#[allow(dead_code)]
+	pub(crate) colon: Token![:],
+	#[allow(dead_code)]
+	pub(crate) bracket: Option<Bracket>,
+	pub(crate) sends: Punctuated<Path, Token![,]>,
+}
+
+impl Parse for Sends {
+	fn parse(input: syn::parse::ParseStream) -> Result<Self> {
+		let content;
+		let keyword_sends = input.parse()?;
+		let colon = input.parse()?;
+		let (bracket, sends) = if !input.peek(syn::token::Bracket) {
+			let mut sends = Punctuated::new();
+			sends.push_value(input.parse::<Path>()?);
+			(None, sends)
+		} else {
+			let bracket = Some(syn::bracketed!(content in input));
+			let sends = Punctuated::parse_terminated(&content)?;
+			(bracket, sends)
+		};
+		Ok(Self { keyword_sends, colon, bracket, sends })
+	}
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct Consumes {
+	#[allow(dead_code)]
+	pub(crate) keyword_consumes: Option<kw::consumes>,
+	#[allow(dead_code)]
+	pub(crate) colon: Option<Token![:]>,
+	pub(crate) consumes: Path,
+}
+
+impl Parse for Consumes {
+	fn parse(input: syn::parse::ParseStream) -> Result<Self> {
+		let lookahead = input.lookahead1();
+		Ok(if lookahead.peek(kw::consumes) {
+			Self {
+				keyword_consumes: Some(input.parse()?),
+				colon: input.parse()?,
+				consumes: input.parse()?,
+			}
+		} else {
+			Self { keyword_consumes: None, colon: None, consumes: input.parse()? }
+		})
+	}
+}
+
+/// Parses `(Foo, sends = [Bar, Baz])`
+/// including the `(` and `)`.
+#[derive(Debug, Clone)]
+pub(crate) struct SubSystemAttrItems {
+	/// The subsystem is in progress, only generate the `Wrapper` variant, but do not forward messages
+	/// and also not include the subsystem in the list of subsystems.
+	pub(crate) wip: bool,
+	/// If there are blocking components in the subsystem and hence it should be
+	/// spawned on a dedicated thread pool for such subssytems.
+	pub(crate) blocking: bool,
+	/// The message type being consumed by the subsystem.
+	pub(crate) consumes: Option<Consumes>,
+	pub(crate) sends: Option<Sends>,
+}
+
+impl Parse for SubSystemAttrItems {
+	fn parse(input: syn::parse::ParseStream) -> Result<Self> {
+		let span = input.span();
+
+		let content;
+		let _paren_token = parenthesized!(content in input);
+
+		let items = content.call(Punctuated::<SubSysAttrItem, Token![,]>::parse_terminated)?;
+
+		let mut unique = HashMap::<
+			std::mem::Discriminant<SubSysAttrItem>,
+			SubSysAttrItem,
+			RandomState,
+		>::default();
+
+		for item in items {
+			if let Some(first) = unique.insert(std::mem::discriminant(&item), item.clone()) {
+				let mut e =
+					Error::new(item.span(), "Duplicate definition of subsystem attribute found");
+				e.combine(Error::new(first.span(), "previously defined here."));
+				return Err(e)
+			}
+		}
+
+		// A subsystem makes no sense if not one of them is provided
+		let sends = extract_variant!(unique, Sends take);
+		let consumes = extract_variant!(unique, Consumes take);
+		if sends.as_ref().map(|sends| sends.sends.is_empty()).unwrap_or(true) && consumes.is_none()
+		{
+			return Err(Error::new(
+				span,
+				"Must have at least one of `consumes: [..]` and `sends: [..]`.",
+			))
+		}
+
+		let blocking = extract_variant!(unique, Blocking; default = false);
+		let wip = extract_variant!(unique, Wip; default = false);
+
+		Ok(Self { blocking, wip, sends, consumes })
+	}
+}
+
+/// Fields that are _not_ subsystems.
+#[derive(Debug, Clone)]
+pub(crate) struct BaggageField {
+	pub(crate) field_name: Ident,
+	pub(crate) field_ty: Path,
+	pub(crate) generic: bool,
+	pub(crate) vis: Visibility,
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct OrchestraInfo {
+	/// Where the support crate `::orchestra` lives.
+	pub(crate) support_crate: Path,
+
+	/// Fields annotated with `#[subsystem(..)]`.
+	pub(crate) subsystems: Vec<SubSysField>,
+	/// Fields that do not define a subsystem,
+	/// but are mere baggage.
+	pub(crate) baggage: Vec<BaggageField>,
+	/// Name of the wrapping enum for all messages, defaults to `AllMessages`.
+	pub(crate) message_wrapper: Ident,
+	/// Name of the orchestra struct, used as a prefix for
+	/// almost all generated types.
+	pub(crate) orchestra_name: Ident,
+
+	/// Size of the bounded channel.
+	pub(crate) message_channel_capacity: usize,
+	/// Size of the bounded signal channel.
+	pub(crate) signal_channel_capacity: usize,
+
+	/// Signals to be sent, sparse information that is used intermittently.
+	pub(crate) extern_signal_ty: Path,
+
+	/// Incoming event type from the outer world, usually an external framework of some sort.
+	pub(crate) extern_event_ty: Path,
+
+	/// Type of messages that are sent to an external subsystem.
+	/// Merely here to be included during generation of `#message_wrapper` type.
+	pub(crate) outgoing_ty: Option<Path>,
+
+	/// Incoming event type from the outer world, commonly from the network.
+	pub(crate) extern_error_ty: Path,
+}
+
+impl OrchestraInfo {
+	pub(crate) fn support_crate_name(&self) -> &Path {
+		&self.support_crate
+	}
+
+	pub(crate) fn variant_names(&self) -> Vec<Ident> {
+		self.subsystems.iter().map(|ssf| ssf.generic.clone()).collect::<Vec<_>>()
+	}
+
+	pub(crate) fn variant_names_without_wip(&self) -> Vec<Ident> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| !ssf.wip)
+			.map(|ssf| ssf.generic.clone())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn variant_names_only_wip(&self) -> Vec<Ident> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| ssf.wip)
+			.map(|ssf| ssf.generic.clone())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn subsystems(&self) -> &[SubSysField] {
+		self.subsystems.as_slice()
+	}
+
+	pub(crate) fn subsystem_names_without_wip(&self) -> Vec<Ident> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| !ssf.wip)
+			.map(|ssf| ssf.name.clone())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn subsystem_generic_types(&self) -> Vec<Ident> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| !ssf.wip)
+			.map(|sff| sff.generic.clone())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn baggage(&self) -> &[BaggageField] {
+		self.baggage.as_slice()
+	}
+
+	pub(crate) fn baggage_names(&self) -> Vec<Ident> {
+		self.baggage.iter().map(|bag| bag.field_name.clone()).collect::<Vec<_>>()
+	}
+	pub(crate) fn baggage_decl(&self) -> Vec<TokenStream> {
+		self.baggage
+			.iter()
+			.map(|bag| {
+				let BaggageField { vis, field_ty, field_name, .. } = bag;
+				quote! { #vis #field_name: #field_ty }
+			})
+			.collect::<Vec<TokenStream>>()
+	}
+
+	pub(crate) fn baggage_generic_types(&self) -> Vec<Ident> {
+		self.baggage
+			.iter()
+			.filter(|bag| bag.generic)
+			.filter_map(|bag| bag.field_ty.get_ident().cloned())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn any_message(&self) -> Vec<Path> {
+		self.subsystems
+			.iter()
+			.map(|ssf| ssf.message_to_consume.clone())
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn channel_names_without_wip(&self, suffix: &'static str) -> Vec<Ident> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| !ssf.wip)
+			.map(|ssf| Ident::new(&(ssf.name.to_string() + suffix), ssf.name.span()))
+			.collect::<Vec<_>>()
+	}
+
+	pub(crate) fn consumes_without_wip(&self) -> Vec<Path> {
+		self.subsystems
+			.iter()
+			.filter(|ssf| !ssf.wip)
+			.map(|ssf| ssf.message_to_consume.clone())
+			.collect::<Vec<_>>()
+	}
+}
+
+/// Internals of the orchestra.
+#[derive(Debug, Clone)]
+pub(crate) struct OrchestraGuts {
+	pub(crate) name: Ident,
+	pub(crate) subsystems: Vec<SubSysField>,
+	pub(crate) baggage: Vec<BaggageField>,
+}
+
+impl OrchestraGuts {
+	pub(crate) fn parse_fields(
+		name: Ident,
+		baggage_generics: HashSet<Ident>,
+		fields: FieldsNamed,
+	) -> Result<Self> {
+		let n = fields.named.len();
+		let mut subsystems = Vec::with_capacity(n);
+		let mut baggage = Vec::with_capacity(n);
+
+		// The types of `#[subsystem(..)]` annotated fields
+		// have to be unique, since they are used as generics
+		// for the builder pattern besides other places.
+		let mut unique_subsystem_idents = HashSet::<Ident>::new();
+		for Field { attrs, vis, ident, ty, .. } in fields.named.into_iter() {
+			// collect all subsystem annotations per field
+			let mut subsystem_attr =
+				attrs.iter().filter(|attr| attr.style == AttrStyle::Outer).filter_map(|attr| {
+					let span = attr.path.span();
+					attr.path.get_ident().filter(|ident| *ident == "subsystem").map(move |_ident| {
+						let attr_tokens = attr.tokens.clone();
+						(attr_tokens, span)
+					})
+				});
+			let ident = ident.ok_or_else(|| {
+				Error::new(
+					ty.span(),
+					"Missing identifier for field, only named fields are expceted.",
+				)
+			})?;
+
+			// a `#[subsystem(..)]` annotation exists
+			if let Some((attr_tokens, span)) = subsystem_attr.next() {
+				if let Some((_attr_tokens2, span2)) = subsystem_attr.next() {
+					return Err({
+						let mut err = Error::new(span, "The first subsystem annotation is at");
+						err.combine(Error::new(span2, "but another here for the same field."));
+						err
+					})
+				}
+
+				let span = attr_tokens.span();
+
+				let attr_tokens = attr_tokens.clone();
+				let subsystem_attrs: SubSystemAttrItems = syn::parse2(attr_tokens.clone())?;
+
+				let field_ty = try_type_to_path(ty, span)?;
+				let generic = field_ty
+					.get_ident()
+					.ok_or_else(|| {
+						Error::new(
+							field_ty.span(),
+							"Must be an identifier, not a path. It will be used as a generic.",
+						)
+					})?
+					.clone();
+				// check for unique subsystem name, otherwise we'd create invalid code:
+				if let Some(previous) = unique_subsystem_idents.get(&generic) {
+					let mut e = Error::new(generic.span(), "Duplicate subsystem names");
+					e.combine(Error::new(previous.span(), "previously defined here."));
+					return Err(e)
+				}
+				unique_subsystem_idents.insert(generic.clone());
+
+				let SubSystemAttrItems { wip, blocking, consumes, sends, .. } = subsystem_attrs;
+
+				// messages to be sent
+				let sends = if let Some(sends) = sends {
+					Vec::from_iter(sends.sends.iter().cloned())
+				} else {
+					vec![]
+				};
+				// messages deemed for consumption
+				let consumes = if let Some(consumes) = consumes {
+					consumes.consumes
+				} else {
+					return Err(Error::new(span, "Must provide exactly one consuming message type"))
+				};
+
+				subsystems.push(SubSysField {
+					name: ident,
+					generic,
+					message_to_consume: consumes,
+					messages_to_send: sends,
+					wip,
+					blocking,
+				});
+			} else {
+				let field_ty = try_type_to_path(ty, ident.span())?;
+				let generic = field_ty
+					.get_ident()
+					.map(|ident| baggage_generics.contains(ident))
+					.unwrap_or(false);
+				baggage.push(BaggageField { field_name: ident, generic, field_ty, vis });
+			}
+		}
+		Ok(Self { name, subsystems, baggage })
+	}
+}
+
+impl Parse for OrchestraGuts {
+	fn parse(input: ParseStream) -> Result<Self> {
+		let ds: ItemStruct = input.parse()?;
+		match ds.fields {
+			syn::Fields::Named(named) => {
+				let name = ds.ident.clone();
+
+				// collect the indepedentent subsystem generics
+				// which need to be carried along, there are the non-generated ones
+				let mut orig_generics = ds.generics;
+
+				// remove defaults from types
+				let mut baggage_generic_idents = HashSet::with_capacity(orig_generics.params.len());
+				orig_generics.params = orig_generics
+					.params
+					.into_iter()
+					.map(|mut generic| {
+						match generic {
+							GenericParam::Type(ref mut param) => {
+								baggage_generic_idents.insert(param.ident.clone());
+								param.eq_token = None;
+								param.default = None;
+							},
+							_ => {},
+						}
+						generic
+					})
+					.collect();
+
+				Self::parse_fields(name, baggage_generic_idents, named)
+			},
+			syn::Fields::Unit => Err(Error::new(
+				ds.fields.span(),
+				"Must be a struct with named fields. Not an unit struct.",
+			)),
+			syn::Fields::Unnamed(unnamed) => Err(Error::new(
+				unnamed.span(),
+				"Must be a struct with named fields. Not an unnamed fields struct.",
+			)),
+		}
+	}
+}

polkadot/node/orchestra/proc-macro/src/parse/parse_subsystem_attr.rs

@@ -0,0 +1,143 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::kw;
+use proc_macro2::Span;
+use quote::{quote, ToTokens};
+use std::collections::{hash_map::RandomState, HashMap};
+use syn::{
+	parse::{Parse, ParseBuffer},
+	punctuated::Punctuated,
+	spanned::Spanned,
+	Error, Ident, Path, Result, Token,
+};
+
+#[derive(Clone, Debug)]
+enum SubsystemAttrItem {
+	/// Error type provided by the user.
+	Error { tag: kw::error, eq_token: Token![=], value: Path },
+	/// For which slot in the orchestra this should be plugged.
+	///
+	/// The subsystem implementation can and should have a different name
+	/// from the declared parameter type in the orchestra.
+	Subsystem { tag: Option<kw::subsystem>, eq_token: Option<Token![=]>, value: Ident },
+	/// The prefix to apply when a subsystem is implemented in a different file/crate
+	/// than the orchestra itself.
+	///
+	/// Important for `#[subsystem(..)]` to reference the traits correctly.
+	TraitPrefix { tag: kw::prefix, eq_token: Token![=], value: Path },
+}
+
+impl ToTokens for SubsystemAttrItem {
+	fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
+		let ts = match self {
+			Self::TraitPrefix { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::Error { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+			Self::Subsystem { tag, eq_token, value } => {
+				quote! { #tag #eq_token, #value }
+			},
+		};
+		tokens.extend(ts.into_iter());
+	}
+}
+
+impl Parse for SubsystemAttrItem {
+	fn parse(input: &ParseBuffer) -> Result<Self> {
+		let lookahead = input.lookahead1();
+		if lookahead.peek(kw::error) {
+			Ok(SubsystemAttrItem::Error {
+				tag: input.parse::<kw::error>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::prefix) {
+			Ok(SubsystemAttrItem::TraitPrefix {
+				tag: input.parse::<kw::prefix>()?,
+				eq_token: input.parse()?,
+				value: input.parse()?,
+			})
+		} else if lookahead.peek(kw::subsystem) {
+			Ok(SubsystemAttrItem::Subsystem {
+				tag: Some(input.parse::<kw::subsystem>()?),
+				eq_token: Some(input.parse()?),
+				value: input.parse()?,
+			})
+		} else {
+			Ok(SubsystemAttrItem::Subsystem { tag: None, eq_token: None, value: input.parse()? })
+		}
+	}
+}
+
+/// Attribute arguments `$args` in `#[subsystem( $args )]`.
+#[derive(Clone, Debug)]
+pub(crate) struct SubsystemAttrArgs {
+	span: Span,
+	pub(crate) error_path: Option<Path>,
+	pub(crate) subsystem_ident: Ident,
+	pub(crate) trait_prefix_path: Option<Path>,
+}
+
+impl Spanned for SubsystemAttrArgs {
+	fn span(&self) -> Span {
+		self.span.clone()
+	}
+}
+
+macro_rules! extract_variant {
+	($unique:expr, $variant:ident ; default = $fallback:expr) => {
+		extract_variant!($unique, $variant).unwrap_or_else(|| $fallback)
+	};
+	($unique:expr, $variant:ident ; err = $err:expr) => {
+		extract_variant!($unique, $variant).ok_or_else(|| Error::new(Span::call_site(), $err))
+	};
+	($unique:expr, $variant:ident) => {
+		$unique.values().find_map(|item| match item {
+			SubsystemAttrItem::$variant { value, .. } => Some(value.clone()),
+			_ => None,
+		})
+	};
+}
+
+impl Parse for SubsystemAttrArgs {
+	fn parse(input: &ParseBuffer) -> Result<Self> {
+		let span = input.span();
+		let items: Punctuated<SubsystemAttrItem, Token![,]> =
+			input.parse_terminated(SubsystemAttrItem::parse)?;
+
+		let mut unique = HashMap::<
+			std::mem::Discriminant<SubsystemAttrItem>,
+			SubsystemAttrItem,
+			RandomState,
+		>::default();
+		for item in items {
+			if let Some(first) = unique.insert(std::mem::discriminant(&item), item.clone()) {
+				let mut e = Error::new(
+					item.span(),
+					format!("Duplicate definition of subsystem generation type found"),
+				);
+				e.combine(Error::new(first.span(), "previously defined here."));
+				return Err(e)
+			}
+		}
+		let error_path = extract_variant!(unique, Error);
+		let subsystem_ident = extract_variant!(unique, Subsystem; err = "Must annotate the identical orchestra error type via `subsystem=..` or plainly as `Subsystem` as specified in the orchestra declaration.")?;
+		let trait_prefix_path = extract_variant!(unique, TraitPrefix);
+		Ok(SubsystemAttrArgs { span, error_path, subsystem_ident, trait_prefix_path })
+	}
+}

polkadot/node/orchestra/proc-macro/src/parse/tests.rs

@@ -0,0 +1,294 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::{SubSysAttrItem, SubSystemAttrItems};
+use assert_matches::assert_matches;
+use quote::quote;
+use syn::parse_quote;
+
+mod attr {
+	use super::*;
+
+	#[test]
+	fn attr_full_works() {
+		let attr: OrchestraAttrArgs = parse_quote! {
+			gen=AllMessage, event=::some::why::ExternEvent, signal=SigSigSig, signal_capacity=111, message_capacity=222,
+			error=OrchestraError,
+		};
+		assert_matches!(attr, OrchestraAttrArgs {
+			message_channel_capacity,
+			signal_channel_capacity,
+			..
+		} => {
+			assert_eq!(message_channel_capacity, 222);
+			assert_eq!(signal_channel_capacity, 111);
+		});
+	}
+
+	#[test]
+	fn attr_partial_works() {
+		let attr: OrchestraAttrArgs = parse_quote! {
+			gen=AllMessage, event=::some::why::ExternEvent, signal=::foo::SigSigSig,
+			error=OrchestraError,
+		};
+		assert_matches!(attr, OrchestraAttrArgs {
+			message_channel_capacity: _,
+			signal_channel_capacity: _,
+			..
+		} => {
+		});
+	}
+}
+
+mod strukt {
+
+	use super::*;
+
+	#[test]
+	fn parse_subsystem_attr_item_works_00_wip() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			wip
+		}), Ok(SubSysAttrItem::Wip(_)) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_02_sends() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			sends: [A, B, C]
+		}), Ok(SubSysAttrItem::Sends(sends)) => {
+			assert_eq!(sends.sends.len(), 3);
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_03_sends() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			sends: [A]
+		}), Ok(SubSysAttrItem::Sends(sends)) => {
+			assert_eq!(sends.sends.len(), 1);
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_04_sends() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			sends: [A,]
+		}), Ok(SubSysAttrItem::Sends(sends)) => {
+			assert_eq!(sends.sends.len(), 1);
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_05_sends() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			sends: []
+		}), Ok(SubSysAttrItem::Sends(sends)) => {
+			assert_eq!(sends.sends.len(), 0);
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_06_consumes() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			consumes: Foo
+		}), Ok(SubSysAttrItem::Consumes(_consumes)) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attr_item_works_07_consumes() {
+		assert_matches!(
+		syn::parse2::<SubSysAttrItem>(quote! {
+			Foo
+		}), Ok(SubSysAttrItem::Consumes(_consumes)) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_00() {
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(wip, blocking, consumes: Foo, sends: [])
+		})
+		.unwrap();
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_01() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(blocking, Foo, sends: [])
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_02() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(consumes: Foo, sends: [Bar])
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_03() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(blocking, consumes: Foo, sends: [Bar])
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_04() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(wip, consumes: Foo, sends: [Bar])
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_05() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(consumes: Foo)
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_06() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(sends: [Foo], consumes: Bar)
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_07_duplicate_send() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(sends: [Foo], Bar, Y)
+		}), Err(e) => {
+			dbg!(e)
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_08() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(sends: [Foo], consumes: Bar)
+		}), Ok(_) => {
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_09_neither_consumes_nor_sends() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(sends: [])
+		}), Err(e) => {
+			// must either consume smth or sends smth, neither is NOK
+			dbg!(e)
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_10_empty_with_braces() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			()
+		}), Err(e) => {
+			dbg!(e)
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_11_empty() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+
+		}), Err(e) => {
+			dbg!(e)
+		});
+	}
+
+	#[test]
+	fn parse_subsystem_attributes_works_12_duplicate_consumes_different_fmt() {
+		assert_matches!(
+		syn::parse2::<SubSystemAttrItems>(quote! {
+			(Foo, consumes = Foo)
+		}), Err(e) => {
+			dbg!(e)
+		});
+	}
+
+	#[test]
+	fn struct_parse_baggage() {
+		let item: OrchestraGuts = parse_quote! {
+			pub struct Ooooh<X = Pffffffft> where X: Secrit {
+				#[subsystem(consumes: Foo, sends: [])]
+				sub0: FooSubsystem,
+
+				metrics: Metrics,
+			}
+		};
+		let _ = dbg!(item);
+	}
+
+	#[test]
+	fn struct_parse_full() {
+		let item: OrchestraGuts = parse_quote! {
+			pub struct Ooooh<X = Pffffffft> where X: Secrit {
+				#[subsystem(consumes: Foo, sends: [])]
+				sub0: FooSubsystem,
+
+				#[subsystem(blocking, consumes: Bar, sends: [])]
+				yyy: BaersBuyBilliardBalls,
+
+				#[subsystem(blocking, consumes: Twain, sends: [])]
+				fff: Beeeeep,
+
+				#[subsystem(consumes: Rope)]
+				mc: MountainCave,
+
+				metrics: Metrics,
+			}
+		};
+		let _ = dbg!(item);
+	}
+
+	#[test]
+	fn struct_parse_basic() {
+		let item: OrchestraGuts = parse_quote! {
+			pub struct Ooooh {
+				#[subsystem(consumes: Foo, sends: [])]
+				sub0: FooSubsystem,
+			}
+		};
+		let _ = dbg!(item);
+	}
+}

polkadot/node/orchestra/proc-macro/src/subsystem.rs

@@ -0,0 +1,309 @@
+// Copyright (C) 2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Generates the bounds for a particular subsystem `Context` and associate `type Sender`.
+//!
+//!
+//! ## Implement `trait Subsystem<Context, Error>` via `subsystem`
+//!
+//! ```ignore
+//! # use orchestra_proc_macro::subsystem;
+//! # mod somewhere {
+//! # use orchestra_proc_macro::orchestra;
+//! # pub use orchestra::*;
+//! #
+//! # #[derive(Debug, thiserror::Error)]
+//! # #[error("Yikes!")]
+//! # pub struct Yikes;
+//! # impl From<OrchestraError> for Yikes {
+//! #   fn from(_: OrchestraError) -> Yikes { Yikes }
+//! # }
+//! # impl From<mpsc::SendError> for Yikes {
+//! #   fn from(_: mpsc::SendError) -> Yikes { Yikes }
+//! # }
+//! #
+//! # #[derive(Debug)]
+//! # pub struct Eve;
+//! #
+//! # #[derive(Debug, Clone)]
+//! # pub struct Sig;
+//! #
+//! # #[derive(Debug, Clone, Copy)]
+//! # pub struct A;
+//! # #[derive(Debug, Clone, Copy)]
+//! # pub struct B;
+//! #
+//! # #[orchestra(signal=Sig, gen=AllOfThem, event=Eve, error=Yikes)]
+//! # pub struct Wonderland {
+//! # 	#[subsystem(A, sends: [B])]
+//! # 	foo: Foo,
+//! # 	#[subsystem(B, sends: [A])]
+//! # 	bar: Bar,
+//! # }
+//! # }
+//! # use somewhere::{Yikes, SpawnedSubsystem};
+//! #
+//! # struct FooSubsystem;
+//! #
+//! #[subsystem(Foo, error = Yikes, prefix = somewhere)]
+//! impl<Context> FooSubsystem {
+//!    fn start(self, context: Context) -> SpawnedSubsystem<Yikes> {
+//!	       // ..
+//!        # let _ = context;
+//!        # unimplemented!()
+//!    }
+//! }
+//! ```
+//!
+//! expands to
+//!
+//! ```ignore
+//! # use orchestra_proc_macro::subsystem;
+//! # mod somewhere {
+//! # use orchestra_proc_macro::orchestra;
+//! # pub use orchestra::*;
+//! #
+//! # #[derive(Debug, thiserror::Error)]
+//! # #[error("Yikes!")]
+//! # pub struct Yikes;
+//! # impl From<OrchestraError> for Yikes {
+//! #   fn from(_: OrchestraError) -> Yikes { Yikes }
+//! # }
+//! # impl From<mpsc::SendError> for Yikes {
+//! #   fn from(_: mpsc::SendError) -> Yikes { Yikes }
+//! # }
+//! #
+//! # #[derive(Debug)]
+//! # pub struct Eve;
+//! #
+//! # #[derive(Debug, Clone)]
+//! # pub struct Sig;
+//! #
+//! # #[derive(Debug, Clone, Copy)]
+//! # pub struct A;
+//! # #[derive(Debug, Clone, Copy)]
+//! # pub struct B;
+//! #
+//! # #[orchestra(signal=Sig, gen=AllOfThem, event=Eve, error=Yikes)]
+//! # pub struct Wonderland {
+//! # 	#[subsystem(A, sends: [B])]
+//! # 	foo: Foo,
+//! # 	#[subsystem(B, sends: [A])]
+//! # 	bar: Bar,
+//! # }
+//! # }
+//! # use somewhere::{Yikes, SpawnedSubsystem};
+//! # use orchestra as support_crate;
+//! #
+//! # struct FooSubsystem;
+//! #
+//! impl<Context> support_crate::Subsystem<Context, Yikes> for FooSubsystem
+//! where
+//! 	Context: somewhere::FooContextTrait,
+//!     Context: support_crate::SubsystemContext,
+//!		<Context as somewhere::FooContextTrait>::Sender: somewhere::FooSenderTrait,
+//!		<Context as support_crate::SubsystemContext>::Sender: somewhere::FooSenderTrait,
+//! {
+//!       fn start(self, context: Context) -> SpawnedSubsystem<Yikes> {
+//!        // ..
+//!        # let _ = context;
+//!        # unimplemented!()
+//!       }
+//! }
+//! ```
+//!
+//! where `support_crate` is either equivalent to `somewhere` or derived from the cargo manifest.
+//!
+//!
+//! ## Add additional trait bounds for a generic `Context` via `contextbounds`
+//!
+//! ### To an `ImplItem`
+//!
+//! ```ignore
+//! #[contextbounds(Foo, prefix = somewhere)]
+//! impl<Context> X {
+//! ..
+//! }
+//! ```
+//!
+//! expands to
+//!
+//! ```ignore
+//! impl<Context> X
+//! where
+//! 	Context: somewhere::FooSubsystemTrait,
+//!     Context: support_crate::SubsystemContext,
+//!		<Context as somewhere::FooContextTrait>::Sender: somewhere::FooSenderTrait,
+//!		<Context as support_crate::SubsystemContext>::Sender: somewhere::FooSenderTrait,
+//! {
+//! }
+//! ```
+//!
+//! ### To a free standing `Fn` (not a method, that's covered by the above)
+//!
+//! ```ignore
+//! #[contextbounds(Foo, prefix = somewhere)]
+//! fn do_smth<Context>(context: &mut Context) {
+//! ..
+//! }
+//! ```
+//!
+//! expands to
+//!
+//! ```ignore
+//! fn do_smth<Context>(context: &mut Context)
+//! where
+//! 	Context: somewhere::FooSubsystemTrait,
+//!     Context: support_crate::SubsystemContext,
+//!		<Context as somewhere::FooContextTrait>::Sender: somewhere::FooSenderTrait,
+//!		<Context as support_crate::SubsystemContext>::Sender: somewhere::FooSenderTrait,
+//! {
+//! }
+//! ```
+use proc_macro2::TokenStream;
+use quote::{format_ident, ToTokens};
+use syn::{parse2, parse_quote, punctuated::Punctuated, Result};
+
+use super::{parse::*, *};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum MakeSubsystem {
+	/// Implements `trait Subsystem` and apply the trait bounds to the `Context` generic.
+	///
+	/// Relevant to `impl Item` only.
+	ImplSubsystemTrait,
+	/// Only apply the trait bounds to the context.
+	AddContextTraitBounds,
+}
+
+pub(crate) fn impl_subsystem_context_trait_bounds(
+	attr: TokenStream,
+	orig: TokenStream,
+	make_subsystem: MakeSubsystem,
+) -> Result<proc_macro2::TokenStream> {
+	let args = parse2::<SubsystemAttrArgs>(attr.clone())?;
+	let span = args.span();
+	let SubsystemAttrArgs { error_path, subsystem_ident, trait_prefix_path, .. } = args;
+
+	let mut item = parse2::<syn::Item>(orig)?;
+
+	// always prefer the direct usage, if it's not there, let's see if there is
+	// a `prefix=*` provided. Either is ok.
+
+	// Technically this is two different things:
+	// The place where the `#[orchestra]` is annotated is where all `trait *SenderTrait` and
+	// `trait *ContextTrait` types exist.
+	// The other usage is the true support crate `orchestra`, where the static ones
+	// are declared.
+	// Right now, if the `support_crate` is not included, it falls back silently to the `trait_prefix_path`.
+	let support_crate = support_crate()
+		.or_else(|_e| {
+			trait_prefix_path.clone().ok_or_else(|| {
+				syn::Error::new(attr.span(), "Couldn't find `orchestra` in manifest, but also missing a `prefix=` to help trait bound resolution")
+			})
+		})?;
+
+	let trait_prefix_path = trait_prefix_path.unwrap_or_else(|| parse_quote! { self });
+	if trait_prefix_path.segments.trailing_punct() {
+		return Err(syn::Error::new(trait_prefix_path.span(), "Must not end with `::`"))
+	}
+
+	let subsystem_ctx_trait = format_ident!("{}ContextTrait", subsystem_ident);
+	let subsystem_sender_trait = format_ident!("{}SenderTrait", subsystem_ident);
+
+	let extra_where_predicates: Punctuated<syn::WherePredicate, syn::Token![,]> = parse_quote! {
+		Context: #trait_prefix_path::#subsystem_ctx_trait,
+		Context: #support_crate::SubsystemContext,
+		<Context as #trait_prefix_path::#subsystem_ctx_trait>::Sender: #trait_prefix_path::#subsystem_sender_trait,
+		<Context as #support_crate::SubsystemContext>::Sender: #trait_prefix_path::#subsystem_sender_trait,
+	};
+
+	let apply_ctx_bound_if_present = move |generics: &mut syn::Generics| -> bool {
+		if generics
+			.params
+			.iter()
+			.find(|generic| match generic {
+				syn::GenericParam::Type(ty) if ty.ident == "Context" => true,
+				_ => false,
+			})
+			.is_some()
+		{
+			let where_clause = generics.make_where_clause();
+			where_clause.predicates.extend(extra_where_predicates.clone());
+			true
+		} else {
+			false
+		}
+	};
+
+	match item {
+		syn::Item::Impl(ref mut struktured_impl) => {
+			if make_subsystem == MakeSubsystem::ImplSubsystemTrait {
+				let error_path = error_path.ok_or_else(|| {
+					syn::Error::new(
+						span,
+						"Must annotate the identical orchestra error type via `error=..`.",
+					)
+				})?;
+				// Only replace the subsystem trait if it's desired.
+				struktured_impl.trait_.replace((
+					None,
+					parse_quote! {
+						#support_crate::Subsystem<Context, #error_path>
+					},
+					syn::token::For::default(),
+				));
+			}
+
+			apply_ctx_bound_if_present(&mut struktured_impl.generics);
+			for item in struktured_impl.items.iter_mut() {
+				match item {
+					syn::ImplItem::Method(method) => {
+						apply_ctx_bound_if_present(&mut method.sig.generics);
+					},
+					_others => {
+						// don't error, just nop
+					},
+				}
+			}
+		},
+		syn::Item::Fn(ref mut struktured_fn) => {
+			if make_subsystem == MakeSubsystem::ImplSubsystemTrait {
+				return Err(syn::Error::new(struktured_fn.span(), "Cannot make a free function a subsystem, did you mean to apply `contextbound` instead?"))
+			}
+			apply_ctx_bound_if_present(&mut struktured_fn.sig.generics);
+		},
+		other =>
+			return Err(syn::Error::new(
+				other.span(),
+				"Macro can only be annotated on functions or struct implementations",
+			)),
+	};
+
+	Ok(item.to_token_stream())
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn is_path() {
+		let _p: Path = parse_quote! { self };
+		let _p: Path = parse_quote! { crate };
+		let _p: Path = parse_quote! { ::foo };
+		let _p: Path = parse_quote! { bar };
+	}
+}

polkadot/node/orchestra/proc-macro/src/tests.rs

@@ -0,0 +1,116 @@
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use assert_matches::assert_matches;
+use quote::quote;
+use syn::parse_quote;
+
+#[test]
+fn print() {
+	let attr = quote! {
+		gen=AllMessage,
+		event=::some::why::ExternEvent,
+		signal=SigSigSig,
+		signal_capacity=111,
+		message_capacity=222,
+		error=OrchestraError,
+	};
+
+	let item = quote! {
+		pub struct Ooooh<X = Pffffffft> where X: Secrit {
+			#[subsystem(Foo)]
+			sub0: FooSubsystem,
+
+			#[subsystem(blocking, Bar)]
+			yyy: BaersBuyBilliardBalls,
+
+			#[subsystem(blocking, Twain)]
+			fff: Beeeeep,
+
+			#[subsystem(Rope)]
+			mc: MountainCave,
+
+			metrics: Metrics,
+		}
+	};
+
+	let output = impl_orchestra_gen(attr, item).expect("Simple example always works. qed");
+	println!("//generated:");
+	println!("{}", output);
+}
+
+#[test]
+fn struct_parse_full() {
+	let item: OrchestraGuts = parse_quote! {
+		pub struct Ooooh<X = Pffffffft> where X: Secrit {
+			#[subsystem(Foo)]
+			sub0: FooSubsystem,
+
+			#[subsystem(blocking, Bar)]
+			yyy: BaersBuyBilliardBalls,
+
+			#[subsystem(blocking, Twain)]
+			fff: Beeeeep,
+
+			#[subsystem(Rope)]
+			mc: MountainCave,
+
+			metrics: Metrics,
+		}
+	};
+	let _ = dbg!(item);
+}
+
+#[test]
+fn struct_parse_basic() {
+	let item: OrchestraGuts = parse_quote! {
+		pub struct Ooooh {
+			#[subsystem(Foo)]
+			sub0: FooSubsystem,
+		}
+	};
+	let _ = dbg!(item);
+}
+
+#[test]
+fn attr_full() {
+	let attr: OrchestraAttrArgs = parse_quote! {
+		gen=AllMessage, event=::some::why::ExternEvent, signal=SigSigSig, signal_capacity=111, message_capacity=222,
+		error=OrchestraError,
+	};
+	assert_matches!(attr, OrchestraAttrArgs {
+		message_channel_capacity,
+		signal_channel_capacity,
+		..
+	} => {
+		assert_eq!(message_channel_capacity, 222);
+		assert_eq!(signal_channel_capacity, 111);
+	});
+}
+
+#[test]
+fn attr_partial() {
+	let attr: OrchestraAttrArgs = parse_quote! {
+		gen=AllMessage, event=::some::why::ExternEvent, signal=::foo::SigSigSig,
+		error=OrchestraError,
+	};
+	assert_matches!(attr, OrchestraAttrArgs {
+		message_channel_capacity: _,
+		signal_channel_capacity: _,
+		..
+	} => {
+	});
+}