Grcov report - collators.rs

1

// Copyright (C) Moondance Labs Ltd.

2

// This file is part of Tanssi.

3

4

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

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

16

17

pub mod basic;

18

pub mod lookahead;

19

20

use {

21

    crate::{find_pre_digest, AuthorityId, OrchestratorAuraWorkerAuxData},

22

    cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface,

23

    cumulus_client_consensus_common::ParachainCandidate,

24

    cumulus_client_consensus_proposer::ProposerInterface,

25

    cumulus_client_parachain_inherent::{ParachainInherentData, ParachainInherentDataProvider},

26

    cumulus_primitives_core::{

27

        relay_chain::Hash as PHash, DigestItem, ParachainBlockData, PersistedValidationData,

28

},

29

    cumulus_relay_chain_interface::RelayChainInterface,

30

    futures::prelude::*,

31

    nimbus_primitives::{CompatibleDigestItem as NimbusCompatibleDigestItem, NIMBUS_KEY_ID},

32

    parity_scale_codec::{Codec, Encode},

33

    polkadot_node_primitives::{Collation, MaybeCompressedPoV},

34

    polkadot_primitives::Id as ParaId,

35

    sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy, StateAction},

36

    sp_application_crypto::{AppCrypto, AppPublic},

37

    sp_consensus::BlockOrigin,

38

    sp_consensus_aura::{digests::CompatibleDigestItem, Slot},

39

    sp_core::crypto::{ByteArray, Pair},

40

    sp_inherents::{CreateInherentDataProviders, InherentData, InherentDataProvider},

41

    sp_keystore::{Keystore, KeystorePtr},

42

    sp_runtime::{

43

        generic::Digest,

44

        traits::{Block as BlockT, HashingFor, Header as HeaderT, Member, Zero},

45

},

46

    sp_state_machine::StorageChanges,

47

    sp_timestamp::Timestamp,

48

    std::{convert::TryFrom, error::Error, time::Duration},

49

};

50

51

/// Parameters for instantiating a [`Collator`].

52

pub struct Params<BI, CIDP, RClient, Proposer, CS> {

53

    /// A builder for inherent data builders.

54

    pub create_inherent_data_providers: CIDP,

55

    /// The block import handle.

56

    pub block_import: BI,

57

    /// An interface to the relay-chain client.

58

    pub relay_client: RClient,

59

    /// The keystore handle used for accessing parachain key material.

60

    pub keystore: KeystorePtr,

61

    /// The identifier of the parachain within the relay-chain.

62

    pub para_id: ParaId,

63

    /// The block proposer used for building blocks.

64

    pub proposer: Proposer,

65

    /// The collator service used for bundling proposals into collations and announcing

66

    /// to the network.

67

    pub collator_service: CS,

68

69

70

/// A utility struct for writing collation logic that makes use of

71

/// Tanssi Aura entirely or in part.

72

pub struct Collator<Block, P, BI, CIDP, RClient, Proposer, CS> {

73

    create_inherent_data_providers: CIDP,

74

    block_import: BI,

75

    relay_client: RClient,

76

    keystore: KeystorePtr,

77

    para_id: ParaId,

78

    proposer: Proposer,

79

    collator_service: CS,

80

    _marker: std::marker::PhantomData<(Block, Box<dyn Fn(P) + Send + Sync + 'static>)>,

81

82

83

impl<Block, P, BI, CIDP, RClient, Proposer, CS> Collator<Block, P, BI, CIDP, RClient, Proposer, CS>

84

where

85

    Block: BlockT,

86

    RClient: RelayChainInterface,

87

    CIDP: CreateInherentDataProviders<Block, (PHash, PersistedValidationData)> + 'static,

88

    BI: BlockImport<Block> + Send + Sync + 'static,

89

    Proposer: ProposerInterface<Block>,

90

    CS: CollatorServiceInterface<Block>,

91

    P: Pair + Send + Sync + 'static,

92

    P::Public: AppPublic + Member,

93

    P::Signature: TryFrom<Vec<u8>> + Member + Codec,

94

95

    /// Instantiate a new instance of the `Tanssi Aura` manager.

96

    pub fn new(params: Params<BI, CIDP, RClient, Proposer, CS>) -> Self {

97

        Collator {

98

            create_inherent_data_providers: params.create_inherent_data_providers,

99

            block_import: params.block_import,

100

            relay_client: params.relay_client,

101

            keystore: params.keystore,

102

            para_id: params.para_id,

103

            proposer: params.proposer,

104

            collator_service: params.collator_service,

105

            _marker: std::marker::PhantomData,

106

107

108

109

    /// Explicitly creates the inherent data for parachain block authoring.

110

    pub async fn create_inherent_data(

111

        &self,

112

        relay_parent: PHash,

113

        validation_data: &PersistedValidationData,

114

        parent_hash: <Block as BlockT>::Hash,

115

        _timestamp: impl Into<Option<Timestamp>>,

116

    ) -> Result<(ParachainInherentData, InherentData), Box<dyn Error + Send + Sync + 'static>> {

117

        let paras_inherent_data = ParachainInherentDataProvider::create_at(

118

            relay_parent,

119

            &self.relay_client,

120

            validation_data,

121

            self.para_id,

122

123

        .await;

124

125

        let paras_inherent_data = match paras_inherent_data {

126

            Some(p) => p,

127

            None => {

128

                return Err(

129

                    format!("Could not create paras inherent data at {:?}", relay_parent).into(),

130

131

132

};

133

134

        let other_inherent_data = self

135

            .create_inherent_data_providers

136

            .create_inherent_data_providers(parent_hash, (relay_parent, validation_data.clone()))

137

            .map_err(|e| e as Box<dyn Error + Send + Sync + 'static>)

138

            .await?

139

            .create_inherent_data()

140

            .await

141

            .map_err(Box::new)?;

142

143

        Ok((paras_inherent_data, other_inherent_data))

144

145

146

    /// Propose, seal, and import a block, packaging it into a collation.

147

///

148

    /// Provide the slot to build at as well as any other necessary pre-digest logs,

149

    /// the inherent data, and the proposal duration and PoV size limits.

150

///

151

    /// The Tanssi Aura pre-digest is set internally.

152

///

153

    /// This does not announce the collation to the parachain network or the relay chain.

154

    #[allow(clippy::cast_precision_loss)]

155

    pub async fn collate(

156

        &mut self,

157

        parent_header: &Block::Header,

158

        slot_claim: &mut SlotClaim<P::Public>,

159

        additional_pre_digest: impl Into<Option<Vec<DigestItem>>>,

160

        inherent_data: (ParachainInherentData, InherentData),

161

        proposal_duration: Duration,

162

        max_pov_size: usize,

163

    ) -> Result<

164

        Option<(

165

            Collation,

166

            ParachainBlockData<Block>,

167

            <Block as BlockT>::Hash,

168

)>,

169

        Box<dyn Error + Send + 'static>,

170

> {

171

        let mut digest = additional_pre_digest.into().unwrap_or_default();

172

        digest.append(&mut slot_claim.pre_digest);

173

174

        let maybe_proposal = self

175

            .proposer

176

            .propose(

177

                parent_header,

178

                &inherent_data.0,

179

                inherent_data.1,

180

                Digest { logs: digest },

181

                proposal_duration,

182

                Some(max_pov_size),

183

184

            .await

185

            .map_err(|e| Box::new(e) as Box<dyn Error + Send>)?;

186

187

        let proposal = match maybe_proposal {

188

            None => return Ok(None),

189

            Some(p) => p,

190

};

191

192

        let sealed_importable = seal_tanssi::<_, P>(

193

            proposal.block,

194

            proposal.storage_changes,

195

            &slot_claim.author_pub,

196

            &self.keystore,

197

198

        .map_err(|e| e as Box<dyn Error + Send>)?;

199

200

        let post_hash = sealed_importable.post_hash();

201

        let block = Block::new(

202

            sealed_importable.post_header(),

203

            sealed_importable

204

                .body

205

                .as_ref()

206

                .expect("body always created with this `propose` fn; qed")

207

                .clone(),

208

);

209

210

        self.block_import

211

            .import_block(sealed_importable)

212

            .map_err(|e| Box::new(e) as Box<dyn Error + Send>)

213

            .await?;

214

215

        if let Some((collation, block_data)) = self.collator_service.build_collation(

216

            parent_header,

217

            post_hash,

218

            ParachainCandidate {

219

                block,

220

                proof: proposal.proof,

221

},

222

) {

223

            tracing::info!(

224

                target: crate::LOG_TARGET,

225

                "PoV size {{ header: {}kb, extrinsics: {}kb, storage_proof: {}kb }}",

226

                block_data.header().encoded_size() as f64 / 1024f64,

227

                block_data.extrinsics().encoded_size() as f64 / 1024f64,

228

                block_data.storage_proof().encoded_size() as f64 / 1024f64,

229

);

230

231

            if let MaybeCompressedPoV::Compressed(ref pov) = collation.proof_of_validity {

232

                tracing::info!(

233

                    target: crate::LOG_TARGET,

234

                    "Compressed PoV size: {}kb",

235

                    pov.block_data.0.len() as f64 / 1024f64,

236

);

237

238

239

            Ok(Some((collation, block_data, post_hash)))

240

        } else {

241

            Err(

242

                Box::<dyn Error + Send + Sync>::from("Unable to produce collation")

243

                    as Box<dyn Error + Send>,

244

245

246

247

248

    /// Get the underlying collator service.

249

    pub fn collator_service(&self) -> &CS {

250

        &self.collator_service

251

252

253

254

fn pre_digest_data<P: Pair>(slot: Slot, claim: P::Public) -> Vec<sp_runtime::DigestItem>

255

where

256

    P::Public: Codec,

257

    P::Signature: Codec,

258

259

    vec![

260

        <DigestItem as CompatibleDigestItem<P::Signature>>::aura_pre_digest(slot),

261

        // We inject the nimbus digest as well. Crutial to be able to verify signatures

262

        <DigestItem as NimbusCompatibleDigestItem>::nimbus_pre_digest(

263

            // TODO remove this unwrap through trait reqs

264

            nimbus_primitives::NimbusId::from_slice(claim.as_ref()).unwrap(),

265

),

266

267

268

269

#[derive(Debug)]

270

pub struct SlotClaim<Pub> {

271

    author_pub: Pub,

272

    pre_digest: Vec<DigestItem>,

273

    slot: Slot,

274

275

276

impl<Pub: Clone> SlotClaim<Pub> {

277

    pub fn unchecked<P>(author_pub: Pub, slot: Slot) -> Self

278

    where

279

        P: Pair<Public = Pub>,

280

        P::Public: Codec,

281

        P::Signature: Codec,

282

283

        SlotClaim {

284

            author_pub: author_pub.clone(),

285

            pre_digest: pre_digest_data::<P>(slot, author_pub),

286

            slot,

287

288

289

290

    /// Get the author's public key.

291

    pub fn author_pub(&self) -> &Pub {

292

        &self.author_pub

293

294

295

    /// Get the pre-digest.

296

    pub fn pre_digest(&self) -> &Vec<DigestItem> {

297

        &self.pre_digest

298

299

300

    /// Get the slot assigned to this claim.

301

    pub fn slot(&self) -> Slot {

302

        self.slot

303

304

305

306

#[derive(Debug, PartialEq, Eq, Copy, Clone)]

307

pub enum ClaimMode {

308

    ForceAuthoring,

309

    NormalAuthoring,

310

    ParathreadCoreBuying { drift_permitted: Slot },

311

312

313

/// Attempt to claim a slot locally.

314

pub fn tanssi_claim_slot<P, B>(

315

    aux_data: OrchestratorAuraWorkerAuxData<P>,

316

    chain_head: &B::Header,

317

    slot: Slot,

318

    claim_mode: ClaimMode,

319

    keystore: &KeystorePtr,

320

) -> Option<SlotClaim<P::Public>>

321

where

322

    P: Pair + Send + Sync + 'static,

323

    P::Public: Codec + std::fmt::Debug,

324

    P::Signature: Codec,

325

    B: BlockT,

326

327

    let author_pub = claim_slot_inner::<P>(slot, &aux_data.authorities, keystore, claim_mode)?;

328

329

    if is_parathread_and_should_skip_slot::<P, B>(&aux_data, chain_head, slot, claim_mode) {

330

        return None;

331

332

333

    Some(SlotClaim::unchecked::<P>(author_pub, slot))

334

335

336

/// Returns true if this container chain is a parathread and the collator should skip this slot and not produce a block

337

pub fn is_parathread_and_should_skip_slot<P, B>(

338

    aux_data: &OrchestratorAuraWorkerAuxData<P>,

339

    chain_head: &B::Header,

340

    slot: Slot,

341

    claim_mode: ClaimMode,

342

) -> bool

343

where

344

    P: Pair + Send + Sync + 'static,

345

    P::Public: Codec + std::fmt::Debug,

346

    P::Signature: Codec,

347

    B: BlockT,

348

349

    if slot.is_zero() {

350

        // Always produce on slot 0 (for tests)

351

        return false;

352

353

    if let Some(slot_freq) = &aux_data.slot_freq {

354

        if let Ok(chain_head_slot) = find_pre_digest::<B, P::Signature>(chain_head) {

355

            // TODO: this doesn't take into account force authoring.

356

            // So a node with `force_authoring = true` will not propose a block for a parathread until the

357

            // `min_slot_freq` has elapsed.

358

            match claim_mode {

359

                ClaimMode::NormalAuthoring | ClaimMode::ForceAuthoring => {

360

                    !slot_freq.should_parathread_author_block(slot, chain_head_slot)

361

362

                ClaimMode::ParathreadCoreBuying { drift_permitted } => {

363

                    !slot_freq.should_parathread_buy_core(slot, drift_permitted, chain_head_slot)

364

365

366

        } else {

367

            // In case of error always propose

368

            false

369

370

    } else {

371

        // Not a parathread: always propose

372

        false

373

374

375

376

/// Attempt to claim a slot using a keystore.

377

pub fn claim_slot_inner<P>(

378

    slot: Slot,

379

    authorities: &Vec<AuthorityId<P>>,

380

    keystore: &KeystorePtr,

381

    claim_mode: ClaimMode,

382

) -> Option<P::Public>

383

where

384

    P: Pair,

385

    P::Public: Codec + std::fmt::Debug,

386

    P::Signature: Codec,

387

388

    let expected_author = crate::slot_author::<P>(slot, authorities.as_slice());

389

    // if running with force-authoring, as long as you are in the authority set, propose

390

    if claim_mode == ClaimMode::ForceAuthoring {

391

        authorities

392

            .iter()

393

            .find(|key| keystore.has_keys(&[(key.to_raw_vec(), NIMBUS_KEY_ID)]))

394

            .cloned()

395

396

    // if not running with force-authoring, just do the usual slot check

397

    else {

398

        expected_author.and_then(|p| {

399

            if keystore.has_keys(&[(p.to_raw_vec(), NIMBUS_KEY_ID)]) {

400

                Some(p.clone())

401

            } else {

402

                None

403

404

})

405

406

407

408

/// Seal a block with a signature in the header.

409

/// This is a copy of [`cumulus_client_consensus_aura::collator::seal`] but using Nimbus instead of

410

/// Aura for the signature.

411

pub fn seal_tanssi<B: BlockT, P>(

412

    pre_sealed: B,

413

    storage_changes: StorageChanges<HashingFor<B>>,

414

    author_pub: &P::Public,

415

    keystore: &KeystorePtr,

416

) -> Result<BlockImportParams<B>, Box<dyn Error + Send + Sync + 'static>>

417

where

418

    P: Pair,

419

    P::Signature: Codec + TryFrom<Vec<u8>>,

420

    P::Public: AppPublic,

421

422

    let (pre_header, body) = pre_sealed.deconstruct();

423

    let pre_hash = pre_header.hash();

424

    let block_number = *pre_header.number();

425

426

    // sign the pre-sealed hash of the block and then

427

    // add it to a digest item.

428

    let signature = Keystore::sign_with(

429

        keystore,

430

        <AuthorityId<P> as AppCrypto>::ID,

431

        <AuthorityId<P> as AppCrypto>::CRYPTO_ID,

432

        author_pub.as_slice(),

433

        pre_hash.as_ref(),

434

435

    .map_err(|e| sp_consensus::Error::CannotSign(format!("{}. Key: {:?}", e, author_pub)))?

436

    .ok_or_else(|| {

437

        sp_consensus::Error::CannotSign(format!(

438

            "Could not find key in keystore. Key: {:?}",

439

            author_pub

440

))

441

    })?;

442

    let signature = signature

443

        .as_slice()

444

        .try_into()

445

        .map_err(|_| sp_consensus::Error::InvalidSignature(signature, author_pub.to_raw_vec()))?;

446

447

    let signature_digest_item = <DigestItem as NimbusCompatibleDigestItem>::nimbus_seal(signature);

448

449

    // seal the block.

450

    let block_import_params = {

451

        let mut block_import_params = BlockImportParams::new(BlockOrigin::Own, pre_header);

452

        block_import_params.post_digests.push(signature_digest_item);

453

        block_import_params.body = Some(body);

454

        block_import_params.state_action =

455

            StateAction::ApplyChanges(sc_consensus::StorageChanges::Changes(storage_changes));

456

        block_import_params.fork_choice = Some(ForkChoiceStrategy::LongestChain);

457

        block_import_params

458

};

459

    let post_hash = block_import_params.post_hash();

460

461

    tracing::info!(

462

        target: crate::LOG_TARGET,

463

        "🔖 Pre-sealed block for proposal at {}. Hash now {:?}, previously {:?}.",

464

        block_number,

465

        post_hash,

466

        pre_hash,

467

);

468

469

    Ok(block_import_params)

470