start working on building the real overseer

cli/src/browser.rs

@@ -20,7 +20,6 @@ use browser_utils::{
 	Client,
 	browser_configuration, set_console_error_panic_hook, init_console_log,
 };
-use std::str::FromStr;
 
 /// Starts the client.
 #[wasm_bindgen]

cli/src/command.rs

@@ -148,6 +148,7 @@ pub fn run() -> Result<()> {
 					_ => service::build_full(
 						config,
 						authority_discovery_disabled,
+						service::IsCollator::No,
 						grandpa_pause,
 					).map(|full| full.task_manager),
 				}

node/core/av-store/Cargo.toml

@@ -19,6 +19,8 @@ polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 polkadot-overseer = { path = "../../overseer" }
 polkadot-primitives = { path = "../../../primitives" }
 
+sc-service = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+
 [dev-dependencies]
 env_logger = "0.7.1"
 assert_matches = "1.3.0"

node/core/av-store/src/lib.rs

@@ -397,6 +397,23 @@ pub struct Config {
 	pub path: PathBuf,
 }
 
+impl std::convert::TryFrom<sc_service::config::DatabaseConfig> for Config {
+	type Error = &'static str;
+
+	fn try_from(config: sc_service::config::DatabaseConfig) -> Result<Self, Self::Error> {
+		let path = config.path().ok_or("custom databases are not supported")?;
+
+		Ok(Self {
+			// substrate cache size is improper here; just use the default
+			cache_size: None,
+			// DB path is a sub-directory of substrate db path to give two properties:
+			// 1: column numbers don't conflict with substrate
+			// 2: commands like purge-chain work without further changes
+			path: path.join("parachains").join("av-store"),
+		})
+	}
+}
+
 impl AvailabilityStoreSubsystem {
 	/// Create a new `AvailabilityStoreSubsystem` with a given config on disk.
 	pub fn new_on_disk(config: Config, metrics: Metrics) -> io::Result<Self> {
@@ -449,7 +466,6 @@ async fn run<Context>(mut subsystem: AvailabilityStoreSubsystem, mut ctx: Contex
 where
 	Context: SubsystemContext<Message=AvailabilityStoreMessage>,
 {
-	let ctx = &mut ctx;
 	loop {
 		// Every time the following two methods are called a read from DB is performed.
 		// But given that these are very small values which are essentially a newtype
@@ -470,16 +486,19 @@ where
 						ActiveLeavesUpdate { activated, .. })
 					)) => {
 						for activated in activated.into_iter() {
-							process_block_activated(ctx, &subsystem.inner, activated).await?;
+							process_block_activated(&mut ctx, &subsystem.inner, activated).await?;
 						}
 					}
 					Ok(FromOverseer::Signal(OverseerSignal::BlockFinalized(hash))) => {
-						process_block_finalized(&subsystem, ctx, &subsystem.inner, hash).await?;
+						process_block_finalized(&subsystem, &mut ctx, &subsystem.inner, hash).await?;
 					}
 					Ok(FromOverseer::Communication { msg }) => {
-						process_message(&mut subsystem, ctx, msg).await?;
+						process_message(&mut subsystem, &mut ctx, msg).await?;
 					}
-					Err(_) => break,
+					Err(e) => {
+						log::error!("AvailabilityStoreSubsystem err: {:#?}", e);
+						break
+					},
 				}
 			}
 			pov_pruning_time = pov_pruning_time => {
@@ -945,15 +964,15 @@ fn query_inner<D: Decode>(db: &Arc<dyn KeyValueDB>, column: u32, key: &[u8]) ->
 }
 
 impl<Context> Subsystem<Context> for AvailabilityStoreSubsystem
-	where
-		Context: SubsystemContext<Message=AvailabilityStoreMessage>,
+where
+	Context: SubsystemContext<Message = AvailabilityStoreMessage>,
 {
 	fn start(self, ctx: Context) -> SpawnedSubsystem {
-		let future = Box::pin(async move {
-			if let Err(e) = run(self, ctx).await {
-				log::error!(target: LOG_TARGET, "Subsystem exited with an error {:?}", e);
-			}
-		});
+		let future = run(self, ctx)
+			.map(|r| if let Err(e) = r {
+				log::error!(target: "availabilitystore", "Subsystem exited with an error {:?}", e);
+			})
+			.boxed();
 
 		SpawnedSubsystem {
 			name: "availability-store-subsystem",

node/core/candidate-selection/src/lib.rs

@@ -420,10 +420,10 @@ impl metrics::Metrics for Metrics {
 			seconds: prometheus::register(
 				prometheus::CounterVec::new(
 					prometheus::Opts::new(
-						"candidate_selection_invalid_selections_total",
-						"Number of Candidate Selection subsystem seconding selections which proved to be invalid.",
+						"candidate_selection_seconds_total",
+						"Number of Candidate Selection subsystem seconding events.",
 					),
-					&["succeeded", "failed"],
+					&["success"],
 				)?,
 				registry,
 			)?,
@@ -433,7 +433,7 @@ impl metrics::Metrics for Metrics {
 						"candidate_selection_invalid_selections_total",
 						"Number of Candidate Selection subsystem seconding selections which proved to be invalid.",
 					),
-					&["succeeded", "failed"],
+					&["success"],
 				)?,
 				registry,
 			)?,

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

@@ -82,7 +82,7 @@ impl Metrics {
 					metrics.validation_requests.with_label_values(&["invalid"]).inc();
 				},
 				Err(_) => {
-					metrics.validation_requests.with_label_values(&["failed"]).inc();
+					metrics.validation_requests.with_label_values(&["validation failure"]).inc();
 				},
 			}
 		}
@@ -98,7 +98,7 @@ impl metrics::Metrics for Metrics {
 						"parachain_validation_requests_total",
 						"Number of validation requests served.",
 					),
-					&["valid", "invalid", "failed"],
+					&["validity"],
 				)?,
 				registry,
 			)?,

node/core/chain-api/src/lib.rs

@@ -40,18 +40,19 @@ use polkadot_node_subsystem_util::{
 };
 use polkadot_primitives::v1::{Block, BlockId};
 use sp_blockchain::HeaderBackend;
+use std::sync::Arc;
 
 use futures::prelude::*;
 
 /// The Chain API Subsystem implementation.
 pub struct ChainApiSubsystem<Client> {
-	client: Client,
+	client: Arc<Client>,
 	metrics: Metrics,
 }
 
 impl<Client> ChainApiSubsystem<Client> {
 	/// Create a new Chain API subsystem with the given client.
-	pub fn new(client: Client, metrics: Metrics) -> Self {
+	pub fn new(client: Arc<Client>, metrics: Metrics) -> Self {
 		ChainApiSubsystem {
 			client,
 			metrics,
@@ -126,8 +127,13 @@ where
 							// fewer than `k` ancestors are available
 							Ok(None) => None,
 							Ok(Some(header)) => {
-								hash = header.parent_hash;
-								Some(Ok(hash))
+								// stop at the genesis header.
+								if header.number == 1 {
+									None
+								} else {
+									hash = header.parent_hash;
+									Some(Ok(hash))
+								}
 							}
 						}
 					});
@@ -171,7 +177,7 @@ impl metrics::Metrics for Metrics {
 						"parachain_chain_api_requests_total",
 						"Number of Chain API requests served.",
 					),
-					&["succeeded", "failed"],
+					&["success"],
 				)?,
 				registry,
 			)?,
@@ -300,11 +306,11 @@ mod tests {
 	}
 
 	fn test_harness(
-		test: impl FnOnce(TestClient, TestSubsystemContextHandle<ChainApiMessage>)
+		test: impl FnOnce(Arc<TestClient>, TestSubsystemContextHandle<ChainApiMessage>)
 			-> BoxFuture<'static, ()>,
 	) {
 		let (ctx, ctx_handle) = make_subsystem_context(TaskExecutor::new());
-		let client = TestClient::default();
+		let client = Arc::new(TestClient::default());
 
 		let subsystem = ChainApiSubsystem::new(client.clone(), Metrics(None));
 		let chain_api_task = run(ctx, subsystem).map(|x| x.unwrap());

node/core/proposer/src/lib.rs

@@ -145,7 +145,7 @@ where
 			let (sender, receiver) = futures::channel::oneshot::channel();
 
 			overseer.wait_for_activation(parent_header_hash, sender).await?;
-			receiver.await.map_err(Error::ClosedChannelFromProvisioner)??;
+			receiver.await.map_err(|_| Error::ClosedChannelAwaitingActivation)??;
 
 			let (sender, receiver) = futures::channel::oneshot::channel();
 			// strictly speaking, we don't _have_ to .await this send_msg before opening the
@@ -156,7 +156,7 @@ where
 				ProvisionerMessage::RequestInherentData(parent_header_hash, sender),
 			)).await?;
 
-			receiver.await.map_err(Error::ClosedChannelFromProvisioner)
+			receiver.await.map_err(|_| Error::ClosedChannelAwaitingInherentData)
 		}
 		.boxed()
 		.fuse();
@@ -236,7 +236,8 @@ pub enum Error {
 	Blockchain(sp_blockchain::Error),
 	Inherent(sp_inherents::Error),
 	Timeout,
-	ClosedChannelFromProvisioner(futures::channel::oneshot::Canceled),
+	ClosedChannelAwaitingActivation,
+	ClosedChannelAwaitingInherentData,
 	Subsystem(SubsystemError)
 }
 
@@ -271,7 +272,8 @@ impl fmt::Display for Error {
 			Self::Blockchain(err) => write!(f, "blockchain error: {}", err),
 			Self::Inherent(err) => write!(f, "inherent error: {:?}", err),
 			Self::Timeout => write!(f, "timeout: provisioner did not return inherent data after {:?}", PROPOSE_TIMEOUT),
-			Self::ClosedChannelFromProvisioner(err) => write!(f, "provisioner closed inherent data channel before sending: {}", err),
+			Self::ClosedChannelAwaitingActivation => write!(f, "closed channel from overseer when awaiting activation"),
+			Self::ClosedChannelAwaitingInherentData => write!(f, "closed channel from provisioner when awaiting inherent data"),
 			Self::Subsystem(err) => write!(f, "subsystem error: {:?}", err),
 		}
 	}
@@ -282,7 +284,6 @@ impl std::error::Error for Error {
 		match self {
 			Self::Consensus(err) => Some(err),
 			Self::Blockchain(err) => Some(err),
-			Self::ClosedChannelFromProvisioner(err) => Some(err),
 			Self::Subsystem(err) => Some(err),
 			_ => None
 		}

node/core/provisioner/src/lib.rs

@@ -482,7 +482,7 @@ impl Metrics {
 	fn on_inherent_data_request(&self, response: Result<(), ()>) {
 		if let Some(metrics) = &self.0 {
 			match response {
-				Ok(()) => metrics.inherent_data_requests.with_label_values(&["succeded"]).inc(),
+				Ok(()) => metrics.inherent_data_requests.with_label_values(&["succeeded"]).inc(),
 				Err(()) => metrics.inherent_data_requests.with_label_values(&["failed"]).inc(),
 			}
 		}
@@ -498,7 +498,7 @@ impl metrics::Metrics for Metrics {
 						"parachain_inherent_data_requests_total",
 						"Number of InherentData requests served by provisioner.",
 					),
-					&["succeeded", "failed"],
+					&["success"],
 				)?,
 				registry,
 			)?,