referrerpolicy=no-referrer-when-downgrade

cumulus_client_pov_recovery/
active_candidate_recovery.rs

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
// Copyright (C) Parity Technologies (UK) Ltd.
// This file is part of Cumulus.
// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0

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

// Cumulus 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 Cumulus. If not, see <https://www.gnu.org/licenses/>.

use sp_runtime::traits::Block as BlockT;

use polkadot_node_primitives::PoV;
use polkadot_node_subsystem::messages::AvailabilityRecoveryMessage;

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

use std::{pin::Pin, sync::Arc};

use crate::RecoveryHandle;

/// 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<Arc<PoV>>)> + Send>>>,
	recovery_handle: Box<dyn RecoveryHandle>,
}

impl<Block: BlockT> ActiveCandidateRecovery<Block> {
	pub fn new(recovery_handle: Box<dyn RecoveryHandle>) -> Self {
		Self { recoveries: Default::default(), recovery_handle }
	}

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

		self.recovery_handle
			.send_recovery_msg(
				AvailabilityRecoveryMessage::RecoverAvailableData(
					candidate.receipt.clone(),
					candidate.session_index,
					None,
					None,
					tx,
				),
				"ActiveCandidateRecovery",
			)
			.await;

		self.recoveries.push(
			async move {
				match rx.await {
					Ok(Ok(res)) => (block_hash, Some(res.pov)),
					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(),
		);
	}

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