Dispute distribution implementation

Cargo.toml

@@ -62,6 +62,7 @@ members = [
 	"node/network/availability-recovery",
 	"node/network/collator-protocol",
 	"node/network/gossip-support",
+	"node/network/dispute-distribution",
 	"node/overseer",
 	"node/primitives",
 	"node/service",

node/network/bridge/src/lib.rs

@@ -24,6 +24,7 @@ use parity_scale_codec::{Encode, Decode};
 use parking_lot::Mutex;
 use futures::prelude::*;
 use futures::stream::BoxStream;
+use polkadot_subsystem::messages::DisputeDistributionMessage;
 use sc_network::Event as NetworkEvent;
 use sp_consensus::SyncOracle;
 
@@ -852,6 +853,10 @@ where
 		.get_statement_fetching()
 		.expect("Gets initialized, must be `Some` on startup. qed.");
 
+	let dispute_receiver = request_multiplexer
+		.get_dispute_sending()
+		.expect("Gets initialized, must be `Some` on startup. qed.");
+
 	let (remote, network_event_handler) = handle_network_messages(
 		ctx.sender().clone(),
 		network_service.clone(),
@@ -867,6 +872,9 @@ where
 	ctx.send_message(AllMessages::StatementDistribution(
 		StatementDistributionMessage::StatementFetchingReceiver(statement_receiver)
 	)).await;
+	ctx.send_message(AllMessages::DisputeDistribution(
+		DisputeDistributionMessage::DisputeSendingReceiver(dispute_receiver)
+	)).await;
 
 	let subsystem_event_handler = handle_subsystem_messages(
 		ctx,

node/network/bridge/src/multiplexer.rs

@@ -42,6 +42,7 @@ use polkadot_subsystem::messages::AllMessages;
 pub struct RequestMultiplexer {
 	receivers: Vec<(Protocol, mpsc::Receiver<network::IncomingRequest>)>,
 	statement_fetching: Option<mpsc::Receiver<network::IncomingRequest>>,
+	dispute_sending: Option<mpsc::Receiver<network::IncomingRequest>>,
 	next_poll: usize,
 }
 
@@ -68,19 +69,31 @@ impl RequestMultiplexer {
 			})
 			.unzip();
 
+		// Ok this code is ugly as hell, it is also a hack, see #2842. But it works and is executed
+		// on startup so, if anything is wrong here it will be noticed immediately.
 		let index = receivers.iter().enumerate().find_map(|(i, (p, _))|
 			if let Protocol::StatementFetching = p {
 				Some(i)
 			} else {
 				None
 			}
-		).expect("Statement fetching must be registered. qed.");
+		).expect("Statement fetching must be registered.");
 		let statement_fetching = Some(receivers.remove(index).1);
 
+		let index = receivers.iter().enumerate().find_map(|(i, (p, _))|
+			if let Protocol::DisputeSending = p {
+				Some(i)
+			} else {
+				None
+			}
+		).expect("Dispute sending must be registered.");
+		let dispute_sending = Some(receivers.remove(index).1);
+
 		(
 			Self {
 				receivers,
 				statement_fetching,
+                dispute_sending,
 				next_poll: 0,
 			},
 			cfgs,
@@ -93,6 +106,13 @@ impl RequestMultiplexer {
 	pub fn get_statement_fetching(&mut self) -> Option<mpsc::Receiver<network::IncomingRequest>> {
 		std::mem::take(&mut self.statement_fetching)
 	}
+
+	/// Get the receiver for handling dispute sending requests.
+	///
+	/// This function will only return `Some` once.
+	pub fn get_dispute_sending(&mut self) -> Option<mpsc::Receiver<network::IncomingRequest>> {
+		std::mem::take(&mut self.dispute_sending)
+	}
 }
 
 impl Stream for RequestMultiplexer {
@@ -174,6 +194,9 @@ fn multiplex_single(
 		Protocol::StatementFetching => {
 			panic!("Statement fetching requests are handled directly. qed.");
 		}
+		Protocol::DisputeSending => {
+			panic!("Dispute sending request are handled directly. qed.");
+		}
 	};
 	Ok(r)
 }

node/network/dispute-distribution/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "polkadot-dispute-distribution"
+version = "0.1.0"
+authors = ["Parity Technologies <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+futures = "0.3.15"
+tracing = "0.1.26"
+parity-scale-codec = { version = "2.0.0", features = ["std"]  }
+polkadot-primitives = { path = "../../../primitives" }
+polkadot-erasure-coding = { path = "../../../erasure-coding" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+polkadot-node-network-protocol = { path = "../../network/protocol" }
+polkadot-node-subsystem-util = { path = "../../subsystem-util" }
+polkadot-node-primitives = { path = "../../primitives" }
+polkadot-node-core-runtime-api = { path = "../../core/runtime-api" }
+sp-application-crypto = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"]  }
+sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
+thiserror = "1.0.23"
+rand = "0.8.3"
+lru = "0.6.5"
+
+[dev-dependencies]
+polkadot-subsystem-testhelpers = { package = "polkadot-node-subsystem-test-helpers", path = "../../subsystem-test-helpers" }
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"] }
+sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-tracing = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-network = {git = "https://github.com/paritytech/substrate", branch = "master" }
+futures-timer = "3.0.2"
+assert_matches = "1.4.0"
+maplit = "1.0"
+smallvec = "1.6.1"

node/network/dispute-distribution/src/error.rs

@@ -0,0 +1,104 @@
+// Copyright 2021 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+//
+
+//! Error handling related code and Error/Result definitions.
+
+use thiserror::Error;
+
+use polkadot_node_subsystem_util::{Fault, runtime, unwrap_non_fatal};
+use polkadot_subsystem::SubsystemError;
+
+use crate::LOG_TARGET;
+use crate::sender;
+
+#[derive(Debug, Error)]
+#[error(transparent)]
+pub struct Error(pub Fault<NonFatal, Fatal>);
+
+impl From<NonFatal> for Error {
+	fn from(e: NonFatal) -> Self {
+		Self(Fault::from_non_fatal(e))
+	}
+}
+
+impl From<Fatal> for Error {
+	fn from(f: Fatal) -> Self {
+		Self(Fault::from_fatal(f))
+	}
+}
+
+impl From<sender::Error> for Error {
+	fn from(e: sender::Error) -> Self {
+		match e.0 {
+			Fault::Fatal(f) => Self(Fault::Fatal(Fatal::Sender(f))),
+			Fault::Err(nf) => Self(Fault::Err(NonFatal::Sender(nf))),
+		}
+	}
+}
+
+/// Fatal errors of this subsystem.
+#[derive(Debug, Error)]
+pub enum Fatal {
+
+	/// Receiving subsystem message from overseer failed.
+	#[error("Receiving message from overseer failed")]
+	SubsystemReceive(#[source] SubsystemError),
+
+	/// Spawning a running task failed.
+	#[error("Spawning subsystem task failed")]
+	SpawnTask(#[source] SubsystemError),
+
+	/// DisputeSender mpsc receiver exhausted.
+	#[error("Erasure chunk requester stream exhausted")]
+	SenderExhausted,
+
+	#[error("Receive channel closed")]
+	IncomingMessageChannel(#[source] SubsystemError),
+
+	/// Errors coming from runtime::Runtime.
+	#[error("Error while accessing runtime information")]
+	Runtime(#[from] #[source] runtime::Fatal),
+
+	/// Errors coming from DisputeSender
+	#[error("Error while accessing runtime information")]
+	Sender(#[from] #[source] sender::Fatal),
+}
+
+/// Non-fatal errors of this subsystem.
+#[derive(Debug, Error)]
+pub enum NonFatal {
+	/// Errors coming from DisputeSender
+	#[error("Error while accessing runtime information")]
+	Sender(#[from] #[source] sender::NonFatal),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+pub type FatalResult<T> = std::result::Result<T, Fatal>;
+
+/// Utility for eating top level errors and log them.
+///
+/// We basically always want to try and continue on error. This utility function is meant to
+/// consume top-level errors by simply logging them
+pub fn log_error(result: Result<()>, ctx: &'static str)
+	-> std::result::Result<(), Fatal>
+{
+	if let Some(error) = unwrap_non_fatal(result.map_err(|e| e.0))? {
+		tracing::warn!(target: LOG_TARGET, error = ?error, ctx);
+	}
+	Ok(())
+}