1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

// 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/>.

use sp_runtime::traits::Block as BlockT;

use polkadot_node_primitives::AvailableData;
use polkadot_node_subsystem::messages::AvailabilityRecoveryMessage;
use polkadot_overseer::Handle as OverseerHandle;

use futures::{channel::oneshot, stream::FuturesUnordered, Future, FutureExt, StreamExt};

use std::{collections::HashSet, pin::Pin};

/// The active candidate recovery.
///
/// This handles the candidate recovery and tracks the activate recoveries.
pub(crate) struct ActiveCandidateRecovery<Block: BlockT> {
	/// The recoveries that are currently being executed.
	recoveries: FuturesUnordered<
		Pin<Box<dyn Future<Output = (Block::Hash, Option<AvailableData>)> + Send>>,
	>,
	/// The block hashes of the candidates currently being recovered.
	candidates: HashSet<Block::Hash>,
	overseer_handle: OverseerHandle,
}

impl<Block: BlockT> ActiveCandidateRecovery<Block> {
	pub fn new(overseer_handle: OverseerHandle) -> Self {
		Self { recoveries: Default::default(), candidates: Default::default(), overseer_handle }
	}

	/// Recover the given `pending_candidate`.
	pub async fn recover_candidate(
		&mut self,
		block_hash: Block::Hash,
		pending_candidate: crate::PendingCandidate<Block>,
	) {
		let (tx, rx) = oneshot::channel();

		self.overseer_handle
			.send_msg(
				AvailabilityRecoveryMessage::RecoverAvailableData(
					pending_candidate.receipt,
					pending_candidate.session_index,
					None,
					tx,
				),
				"ActiveCandidateRecovery",
			)
			.await;

		self.candidates.insert(block_hash);

		self.recoveries.push(
			async move {
				match rx.await {
					Ok(Ok(res)) => (block_hash, Some(res)),
					Ok(Err(error)) => {
						tracing::debug!(
							target: crate::LOG_TARGET,
							?error,
							?block_hash,
							"Availability recovery failed",
						);
						(block_hash, None)
					},
					Err(_) => {
						tracing::debug!(
							target: crate::LOG_TARGET,
							"Availability recovery oneshot channel closed",
						);
						(block_hash, None)
					},
				}
			}
			.boxed(),
		);
	}

	/// Returns if the given `candidate` is being recovered.
	pub fn is_being_recovered(&self, candidate: &Block::Hash) -> bool {
		self.candidates.contains(candidate)
	}

	/// Waits for the next recovery.
	///
	/// If the returned [`AvailableData`] is `None`, it means that the recovery failed.
	pub async fn wait_for_recovery(&mut self) -> (Block::Hash, Option<AvailableData>) {
		loop {
			if let Some(res) = self.recoveries.next().await {
				self.candidates.remove(&res.0);
				return res
			} else {
				futures::pending!()
			}
		}
	}
}