polkadot_node_network_protocol/request_response/
outgoing.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Polkadot.
3
4// Polkadot is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8
9// Polkadot is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12// GNU General Public License for more details.
13
14// You should have received a copy of the GNU General Public License
15// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
16
17use futures::{channel::oneshot, prelude::Future, FutureExt};
18
19use codec::{Decode, Encode, Error as DecodingError};
20use network::ProtocolName;
21
22use sc_network as network;
23use sc_network_types::PeerId;
24
25use polkadot_primitives::AuthorityDiscoveryId;
26
27use super::{v1, v2, v3, IsRequest, Protocol};
28
29/// All requests that can be sent to the network bridge via `NetworkBridgeTxMessage::SendRequest`.
30#[derive(Debug)]
31pub enum Requests {
32	/// Request an availability chunk from a node.
33	ChunkFetching(OutgoingRequest<v2::ChunkFetchingRequest, v1::ChunkFetchingRequest>),
34	/// Fetch a collation from a collator which previously announced it.
35	CollationFetchingV1(OutgoingRequest<v1::CollationFetchingRequest>),
36	/// Fetch a PoV from a validator which previously sent out a seconded statement.
37	PoVFetchingV1(OutgoingRequest<v1::PoVFetchingRequest>),
38	/// Request full available data from a node.
39	AvailableDataFetchingV1(OutgoingRequest<v1::AvailableDataFetchingRequest>),
40	/// Requests for notifying about an ongoing dispute.
41	DisputeSendingV1(OutgoingRequest<v1::DisputeRequest>),
42
43	/// Request a candidate and attestations.
44	AttestedCandidateV2(OutgoingRequest<v2::AttestedCandidateRequest>),
45	/// Fetch a collation from a collator which previously announced it.
46	/// Compared to V1 it requires specifying which candidate is requested by its hash.
47	CollationFetchingV2(OutgoingRequest<v2::CollationFetchingRequest>),
48	/// Fetch a collation from a collator which previously announced it.
49	/// Compared to V2, the collation is keyed by the output head data hash rather
50	/// than candidate hash, and the response always carries parent head data.
51	CollationFetchingV3(OutgoingRequest<v3::CollationFetchingRequest>),
52}
53
54impl Requests {
55	/// Encode the request.
56	///
57	/// The corresponding protocol is returned as well, as we are now leaving typed territory.
58	///
59	/// Note: `Requests` is just an enum collecting all supported requests supported by network
60	/// bridge, it is never sent over the wire. This function just encodes the individual requests
61	/// contained in the `enum`.
62	pub fn encode_request(self) -> (Protocol, OutgoingRequest<Vec<u8>>) {
63		match self {
64			Self::ChunkFetching(r) => r.encode_request(),
65			Self::CollationFetchingV1(r) => r.encode_request(),
66			Self::CollationFetchingV2(r) => r.encode_request(),
67			Self::CollationFetchingV3(r) => r.encode_request(),
68			Self::PoVFetchingV1(r) => r.encode_request(),
69			Self::AvailableDataFetchingV1(r) => r.encode_request(),
70			Self::DisputeSendingV1(r) => r.encode_request(),
71			Self::AttestedCandidateV2(r) => r.encode_request(),
72		}
73	}
74}
75
76/// Used by the network to send us a response to a request.
77pub type ResponseSender = oneshot::Sender<Result<(Vec<u8>, ProtocolName), network::RequestFailure>>;
78
79/// Any error that can occur when sending a request.
80#[derive(Debug, thiserror::Error)]
81pub enum RequestError {
82	/// Response could not be decoded.
83	#[error("Response could not be decoded: {0}")]
84	InvalidResponse(#[from] DecodingError),
85
86	/// Some error in substrate/libp2p happened.
87	#[error("{0}")]
88	NetworkError(#[from] network::RequestFailure),
89
90	/// Response got canceled by networking.
91	#[error("Response channel got canceled")]
92	Canceled(#[from] oneshot::Canceled),
93}
94
95impl RequestError {
96	/// Whether the error represents some kind of timeout condition.
97	pub fn is_timed_out(&self) -> bool {
98		match self {
99			Self::Canceled(_) |
100			Self::NetworkError(network::RequestFailure::Obsolete) |
101			Self::NetworkError(network::RequestFailure::Network(
102				network::OutboundFailure::Timeout,
103			)) => true,
104			_ => false,
105		}
106	}
107}
108
109/// A request to be sent to the network bridge, including a sender for sending responses/failures.
110///
111/// The network implementation will make use of that sender for informing the requesting subsystem
112/// about responses/errors.
113///
114/// When using `Recipient::Peer`, keep in mind that no address (as in IP address and port) might
115/// be known for that specific peer. You are encouraged to use `Peer` for peers that you are
116/// expected to be already connected to.
117/// When using `Recipient::Authority`, the addresses can be found thanks to the authority
118/// discovery system.
119#[derive(Debug)]
120pub struct OutgoingRequest<Req, FallbackReq = Req> {
121	/// Intended recipient of this request.
122	pub peer: Recipient,
123	/// The actual request to send over the wire.
124	pub payload: Req,
125	/// Optional fallback request and protocol.
126	pub fallback_request: Option<(FallbackReq, Protocol)>,
127	/// Sender which is used by networking to get us back a response.
128	pub pending_response: ResponseSender,
129}
130
131/// Potential recipients of an outgoing request.
132#[derive(Debug, Eq, Hash, PartialEq, Clone)]
133pub enum Recipient {
134	/// Recipient is a regular peer and we know its peer id.
135	Peer(PeerId),
136	/// Recipient is a validator, we address it via this `AuthorityDiscoveryId`.
137	Authority(AuthorityDiscoveryId),
138}
139
140/// Responses received for an `OutgoingRequest`.
141pub type OutgoingResult<Res> = Result<Res, RequestError>;
142
143impl<Req, FallbackReq> OutgoingRequest<Req, FallbackReq>
144where
145	Req: IsRequest + Encode,
146	Req::Response: Decode,
147	FallbackReq: IsRequest + Encode,
148	FallbackReq::Response: Decode,
149{
150	/// Create a new `OutgoingRequest`.
151	///
152	/// It will contain a sender that is used by the networking for sending back responses. The
153	/// connected receiver is returned as the second element in the returned tuple.
154	pub fn new(
155		peer: Recipient,
156		payload: Req,
157	) -> (Self, impl Future<Output = OutgoingResult<Req::Response>>) {
158		let (tx, rx) = oneshot::channel();
159		let r = Self { peer, payload, pending_response: tx, fallback_request: None };
160		(r, receive_response::<Req>(rx.map(|r| r.map(|r| r.map(|(resp, _)| resp)))))
161	}
162
163	/// Create a new `OutgoingRequest` with a fallback in case the remote does not support this
164	/// protocol. Useful when adding a new version of a req-response protocol, to achieve
165	/// compatibility with the older version.
166	///
167	/// Returns a raw `Vec<u8>` response over the channel. Use the associated `ProtocolName` to know
168	/// which request was the successful one and appropriately decode the response.
169	pub fn new_with_fallback(
170		peer: Recipient,
171		payload: Req,
172		fallback_request: FallbackReq,
173	) -> (Self, impl Future<Output = OutgoingResult<(Vec<u8>, ProtocolName)>>) {
174		let (tx, rx) = oneshot::channel();
175		let r = Self {
176			peer,
177			payload,
178			pending_response: tx,
179			fallback_request: Some((fallback_request, FallbackReq::PROTOCOL)),
180		};
181		(r, async { Ok(rx.await??) })
182	}
183
184	/// Encode a request into a `Vec<u8>`.
185	///
186	/// As this throws away type information, we also return the `Protocol` this encoded request
187	/// adheres to.
188	pub fn encode_request(self) -> (Protocol, OutgoingRequest<Vec<u8>>) {
189		let OutgoingRequest { peer, payload, pending_response, fallback_request } = self;
190		let encoded = OutgoingRequest {
191			peer,
192			payload: payload.encode(),
193			fallback_request: fallback_request.map(|(r, p)| (r.encode(), p)),
194			pending_response,
195		};
196		(Req::PROTOCOL, encoded)
197	}
198}
199
200/// Future for actually receiving a typed response for an `OutgoingRequest`.
201async fn receive_response<Req>(
202	rec: impl Future<Output = Result<Result<Vec<u8>, network::RequestFailure>, oneshot::Canceled>>,
203) -> OutgoingResult<Req::Response>
204where
205	Req: IsRequest,
206	Req::Response: Decode,
207{
208	let raw = rec.await??;
209	Ok(Decode::decode(&mut raw.as_ref())?)
210}
polkadot_node_network_protocol/request_response/outgoing.rs

polkadot_node_network_protocol/request_response/
outgoing.rs
