Grcov report - dev

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

//! Development Polkadot service. Adapted from `polkadot_service` crate

18

//! and removed un-necessary components which are not required in dev node.

19

//!

20

//! Following major changes are made:

21

//! 1. Removed beefy and grandpa notification service and request response protocols

22

//! 2. Removed support for parachains which also eliminated the need to start overseer and all other subsystems associated with collation + network request/response protocols for the same

23

//! 3. Removed support for hardware benchmarking

24

//! 4. Removed authority discovery service

25

//! 5. Removed spawning of beefy, grandpa and MMR worker

26

//! 6. Removed rpc extensions for beefy, grandpa and babe and added support for manual seal

27

//! 7. Removed beefy and grandpa block import from block import pipeline (Babe remains)

28

//! 8. Using manual seal import queue instead of babe import queue

29

//! 9. Started manual seal worker

30

//! 10. If amount of time passed between two block is less than slot duration, we emulate passing of time babe block import and runtime

31

//!     by incrementing timestamp by slot duration.

32

33

use {

34

    crate::dev_rpcs::{DevApiServer, DevRpc},

35

    async_io::Timer,

36

    babe::{BabeBlockImport, BabeLink},

37

    codec::{Decode, Encode},

38

    consensus_common::SelectChain,

39

    cumulus_primitives_core::ParaId,

40

    dancelight_runtime::RuntimeApi,

41

    futures::{Stream, StreamExt},

42

    jsonrpsee::RpcModule,

43

    manual_container_chains_exclusion_rpc::{

44

        ManualContainerChainsExclusion, ManualContainerChainsExclusionApiServer,

45

},

46

    node_common::service::node_builder::Sealing,

47

    polkadot_core_primitives::{AccountId, Balance, Block, Hash, Nonce},

48

    polkadot_node_core_parachains_inherent::Error as InherentError,

49

    polkadot_overseer::Handle,

50

    polkadot_parachain_primitives::primitives::UpwardMessages,

51

    polkadot_primitives::{

52

        runtime_api::ParachainHost, BackedCandidate, CandidateCommitments, CandidateDescriptor,

53

        CollatorPair, CommittedCandidateReceipt, CompactStatement, EncodeAs,

54

        InherentData as ParachainsInherentData, OccupiedCoreAssumption, SigningContext,

55

        ValidityAttestation,

56

},

57

    polkadot_rpc::RpcExtension,

58

    polkadot_service::{

59

        BlockT, Error, IdentifyVariant, NewFullParams, OverseerGen, SelectRelayChain,

60

},

61

    sc_client_api::{AuxStore, Backend},

62

    sc_consensus_manual_seal::{

63

        consensus::babe::BabeConsensusDataProvider,

64

        rpc::{ManualSeal, ManualSealApiServer},

65

        run_manual_seal, EngineCommand, ManualSealParams,

66

},

67

    sc_executor::{HeapAllocStrategy, WasmExecutor, DEFAULT_HEAP_ALLOC_STRATEGY},

68

    sc_keystore::Keystore,

69

    sc_transaction_pool_api::{OffchainTransactionPoolFactory, TransactionPool},

70

    service::{Configuration, KeystoreContainer, RpcHandlers, TaskManager},

71

    sp_api::ProvideRuntimeApi,

72

    sp_block_builder::BlockBuilder,

73

    sp_blockchain::{HeaderBackend, HeaderMetadata},

74

    sp_consensus_aura::{inherents::InherentType as AuraInherentType, AURA_ENGINE_ID},

75

    sp_consensus_babe::SlotDuration,

76

    sp_core::{ByteArray, Pair, H256},

77

    sp_keystore::KeystorePtr,

78

    sp_runtime::{traits::BlakeTwo256, DigestItem, RuntimeAppPublic},

79

    std::{cmp::max, ops::Add, sync::Arc, time::Duration},

80

    telemetry::{Telemetry, TelemetryWorker, TelemetryWorkerHandle},

81

};

82

83

// We use this key to store whether we want the para inherent mocker to be active

84

const PARA_INHERENT_SELECTOR_AUX_KEY: &[u8] = b"__DEV_PARA_INHERENT_SELECTOR";

85

86

const CONTAINER_CHAINS_EXCLUSION_AUX_KEY: &[u8] = b"__DEV_CONTAINER_CHAINS_EXCLUSION";

87

88

pub type FullBackend = service::TFullBackend<Block>;

89

90

pub type FullClient = service::TFullClient<

91

    Block,

92

    RuntimeApi,

93

    WasmExecutor<(

94

        sp_io::SubstrateHostFunctions,

95

        frame_benchmarking::benchmarking::HostFunctions,

96

)>,

97

>;

98

99

pub struct NewFull {

100

    pub task_manager: TaskManager,

101

    pub client: Arc<FullClient>,

102

    pub overseer_handle: Option<Handle>,

103

    pub network: Arc<dyn sc_network::service::traits::NetworkService>,

104

    pub sync_service: Arc<sc_network_sync::SyncingService<Block>>,

105

    pub rpc_handlers: RpcHandlers,

106

    pub backend: Arc<FullBackend>,

107

108

109

/// Custom Deps for dev Rpc extension

110

struct DevDeps<C, P> {

111

    /// The client instance to use.

112

    pub client: Arc<C>,

113

    /// Transaction pool instance.

114

    pub pool: Arc<P>,

115

    /// Manual seal command sink

116

    pub command_sink: Option<futures::channel::mpsc::Sender<EngineCommand<Hash>>>,

117

    /// Dev rpcs

118

    pub dev_rpc: Option<DevRpc>,

119

    /// Channels for manually excluding container chains from producing blocks

120

    pub container_chain_exclusion_sender: Option<flume::Sender<Vec<ParaId>>>,

121

122

123

120

fn create_dev_rpc_extension<C, P>(

124

120

    DevDeps {

125

120

        client,

126

120

        pool,

127

120

        command_sink: maybe_command_sink,

128

120

        dev_rpc: maybe_dev_rpc,

129

120

        container_chain_exclusion_sender: maybe_container_chain_exclusion_sender,

130

120

    }: DevDeps<C, P>,

131

120

) -> Result<RpcExtension, Box<dyn std::error::Error + Send + Sync>>

132

120

where

133

120

    C: ProvideRuntimeApi<Block>

134

120

        + HeaderBackend<Block>

135

120

        + AuxStore

136

120

        + HeaderMetadata<Block, Error = sp_blockchain::Error>

137

120

        + Send

138

120

        + Sync

139

120

        + 'static,

140

120

    C::Api: substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>,

141

120

    C::Api: pallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>,

142

120

    C::Api: BlockBuilder<Block>,

143

120

    P: TransactionPool + Sync + Send + 'static,

144

145

    use {

146

        pallet_transaction_payment_rpc::{TransactionPayment, TransactionPaymentApiServer},

147

        substrate_frame_rpc_system::{System, SystemApiServer},

148

};

149

150

120

    let mut io = RpcModule::new(());

151

120

    io.merge(System::new(client.clone(), pool.clone()).into_rpc())?;

152

120

    io.merge(TransactionPayment::new(client.clone()).into_rpc())?;

153

154

120

    if let Some(command_sink) = maybe_command_sink {

155

120

        io.merge(ManualSeal::new(command_sink).into_rpc())?;

156

157

158

120

    if let Some(dev_rpc_data) = maybe_dev_rpc {

159

120

        io.merge(dev_rpc_data.into_rpc())?;

160

161

162

120

    if let Some(container_chain_exclusion_message_channel) = maybe_container_chain_exclusion_sender

163

164

120

        io.merge(

165

120

            ManualContainerChainsExclusion {

166

120

                container_chain_exclusion_message_channel,

167

120

168

120

            .into_rpc(),

169

120

)?;

170

171

172

120

    Ok(io)

173

120

174

175

/// We use EmptyParachainsInherentDataProvider to insert an empty parachain inherent in the block

176

/// to satisfy runtime

177

struct EmptyParachainsInherentDataProvider;

178

179

/// Copied from polkadot service just so that this code retains same structure as

180

/// polkadot_service crate.

181

struct Basics {

182

    task_manager: TaskManager,

183

    client: Arc<FullClient>,

184

    backend: Arc<FullBackend>,

185

    keystore_container: KeystoreContainer,

186

    telemetry: Option<Telemetry>,

187

188

189

impl EmptyParachainsInherentDataProvider {

190

    pub async fn create<C: HeaderBackend<Block>>(

191

        client: Arc<C>,

192

        parent: Hash,

193

    ) -> Result<ParachainsInherentData, InherentError> {

194

        let parent_header = match client.header(parent) {

195

            Ok(Some(h)) => h,

196

            Ok(None) => return Err(InherentError::ParentHeaderNotFound(parent)),

197

            Err(err) => return Err(InherentError::Blockchain(err)),

198

};

199

200

        Ok(ParachainsInherentData {

201

            bitfields: Vec::new(),

202

            backed_candidates: Vec::new(),

203

            disputes: Vec::new(),

204

            parent_header,

205

})

206

207

208

209

/// Creates new development full node with manual seal

210

60

pub fn build_full<OverseerGenerator: OverseerGen>(

211

60

    sealing: Sealing,

212

60

    config: Configuration,

213

60

    mut params: NewFullParams<OverseerGenerator>,

214

60

) -> Result<NewFull, Error> {

215

60

    let is_polkadot = config.chain_spec.is_polkadot();

216

217

60

    params.overseer_message_channel_capacity_override = params

218

60

        .overseer_message_channel_capacity_override

219

60

        .map(move |capacity| {

220

            if is_polkadot {

221

                gum::warn!("Channel capacity should _never_ be tampered with on polkadot!");

222

223

            capacity

224

});

225

226

60

    match config.network.network_backend {

227

        sc_network::config::NetworkBackendType::Libp2p => {

228

            new_full::<_, sc_network::NetworkWorker<Block, Hash>>(sealing, config, params)

229

230

        sc_network::config::NetworkBackendType::Litep2p => {

231

60

            new_full::<_, sc_network::Litep2pNetworkBackend>(sealing, config, params)

232

233

234

60

235

236

/// We use MockParachainsInherentDataProvider to insert an parachain inherent with mocked

237

/// candidates

238

/// We detect whether any of the keys in our keystore is assigned to a core and provide

239

/// a mocked candidate in such core

240

struct MockParachainsInherentDataProvider<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> {

241

    pub client: Arc<C>,

242

    pub parent: Hash,

243

    pub keystore: KeystorePtr,

244

    pub upward_messages_receiver: flume::Receiver<Vec<u8>>,

245

    pub container_chain_exclusion_receiver: flume::Receiver<Vec<ParaId>>,

246

247

248

impl<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> MockParachainsInherentDataProvider<C>

249

where

250

    C::Api: ParachainHost<Block>,

251

    C: AuxStore,

252

253

    pub fn new(

254

        client: Arc<C>,

255

        parent: Hash,

256

        keystore: KeystorePtr,

257

        upward_messages_receiver: flume::Receiver<Vec<u8>>,

258

        container_chain_exclusion_receiver: flume::Receiver<Vec<ParaId>>,

259

    ) -> Self {

260

        MockParachainsInherentDataProvider {

261

            client,

262

            parent,

263

            keystore,

264

            upward_messages_receiver,

265

            container_chain_exclusion_receiver,

266

267

268

269

    pub async fn create(

270

        client: Arc<C>,

271

        parent: Hash,

272

        keystore: KeystorePtr,

273

        upward_messages_receiver: flume::Receiver<Vec<u8>>,

274

        container_chains_exclusion_receiver: flume::Receiver<Vec<ParaId>>,

275

    ) -> Result<ParachainsInherentData, InherentError> {

276

        let parent_header = match client.header(parent) {

277

            Ok(Some(h)) => h,

278

            Ok(None) => return Err(InherentError::ParentHeaderNotFound(parent)),

279

            Err(err) => return Err(InherentError::Blockchain(err)),

280

};

281

282

        // Strategy:

283

        // we usually have 1 validator per core, and we usually run with --alice

284

        // the idea is that at least alice will be assigned to one core

285

        // if we find in the keystore the validator attached to a particular core,

286

        // we generate a signature for the parachain assigned to that core

287

        // To retrieve the validator keys, cal runtime api:

288

289

        // this following piece of code predicts whether the validator is assigned to a particular

290

        // core where a candidate for a parachain needs to be created

291

        let runtime_api = client.runtime_api();

292

293

        // we get all validators

294

295

        // we get the current claim queue to know core availability

296

        let claim_queue = runtime_api.claim_queue(parent).unwrap();

297

298

        // we get the validator groups

299

        let (groups, rotation_info) = runtime_api.validator_groups(parent).unwrap();

300

301

        // we calculate rotation since start, which will define the core assignation

302

        // to validators

303

        let rotations_since_session_start = (parent_header.number

304

            - rotation_info.session_start_block)

305

            / rotation_info.group_rotation_frequency;

306

307

        // Get all the available keys in the keystore

308

        let available_keys = keystore

309

            .keys(polkadot_primitives::PARACHAIN_KEY_TYPE_ID)

310

            .unwrap();

311

312

        // create a slot number identical to the parent block num

313

        let slot_number = AuraInherentType::from(u64::from(parent_header.number));

314

315

        // create a mocked header

316

        let parachain_mocked_header = sp_runtime::generic::Header::<u32, BlakeTwo256> {

317

            parent_hash: Default::default(),

318

            number: parent_header.number,

319

            state_root: Default::default(),

320

            extrinsics_root: Default::default(),

321

            digest: sp_runtime::generic::Digest {

322

                logs: vec![DigestItem::PreRuntime(AURA_ENGINE_ID, slot_number.encode())],

323

},

324

};

325

326

        // retrieve availability cores

327

        let availability_cores = runtime_api.availability_cores(parent).unwrap();

328

329

        // retrieve current session_idx

330

        let session_idx = runtime_api.session_index_for_child(parent).unwrap();

331

332

        // retrieve all validators

333

        let all_validators = runtime_api.validators(parent).unwrap();

334

335

        // construct full availability bitvec

336

        let availability_bitvec = availability_bitvec(1, availability_cores.len());

337

338

        let signature_ctx = SigningContext {

339

            parent_hash: parent,

340

            session_index: session_idx,

341

};

342

343

        // we generate the availability bitfield sigs

344

        // TODO: here we assume all validator keys are able to sign with our keystore

345

        // we need to make sure the key is there before we try to sign

346

        // this is mostly to indicate that the erasure coding chunks where received by all val

347

        let bitfields: Vec<UncheckedSigned<AvailabilityBitfield>> = all_validators

348

            .iter()

349

            .enumerate()

350

            .map(|(i, public)| {

351

                keystore_sign(

352

                    &keystore,

353

                    availability_bitvec.clone(),

354

                    &signature_ctx,

355

                    ValidatorIndex(i as u32),

356

                    public,

357

358

                .unwrap()

359

                .unwrap()

360

})

361

            .collect();

362

363

        // generate a random collator pair

364

        let collator_pair = CollatorPair::generate().0;

365

        let mut backed_cand: Vec<BackedCandidate<H256>> = vec![];

366

367

        let container_chains_exclusion_messages: Vec<Vec<ParaId>> =

368

            container_chains_exclusion_receiver.drain().collect();

369

        // If there is a new set of excluded container chains, we update it

370

        if let Some(mock_excluded_container_chains) = container_chains_exclusion_messages.last() {

371

            client

372

                .insert_aux(

373

&[(

374

                        CONTAINER_CHAINS_EXCLUSION_AUX_KEY,

375

                        mock_excluded_container_chains.encode().as_slice(),

376

)],

377

                    &[],

378

379

                .expect("Should be able to write to aux storage; qed");

380

381

        let new_excluded_container_chains_value = client

382

            .get_aux(CONTAINER_CHAINS_EXCLUSION_AUX_KEY)

383

            .expect("Should be able to query aux storage; qed")

384

            .unwrap_or(Vec::<ParaId>::new().encode());

385

        let mock_excluded_container_chains: Vec<ParaId> =

386

            Decode::decode(&mut new_excluded_container_chains_value.as_slice())

387

                .expect("Vector non-decodable");

388

389

        // iterate over every core|para pair

390

        for (core, para) in claim_queue {

391

            // allows preventing container chains from producing blocks in dev mode

392

            let mut para = para.clone();

393

            para.retain(|x| !mock_excluded_container_chains.contains(x));

394

            // check which group is assigned to each core

395

            let group_assigned_to_core =

396

                core.0 + rotations_since_session_start % groups.len() as u32;

397

            // check validator indices associated to the core

398

            let indices_associated_to_core = groups.get(group_assigned_to_core as usize).unwrap();

399

            for index in indices_associated_to_core {

400

                // fetch validator keys

401

                let validator_keys_to_find = all_validators.get(index.0 as usize).unwrap();

402

                // Iterate keys until we find an eligible one, or run out of candidates.

403

                for type_public_pair in &available_keys {

404

                    if let Ok(validator) =

405

                        polkadot_primitives::ValidatorId::from_slice(type_public_pair)

406

407

                        // if we find the validator in keystore, we try to create a backed cand

408

                        if validator_keys_to_find == &validator {

409

                            // we work with the previous included data

410

                            let mut persisted_validation_data = runtime_api

411

                                .persisted_validation_data(

412

                                    parent,

413

                                    para[0],

414

                                    OccupiedCoreAssumption::Included,

415

416

                                .unwrap()

417

                                .unwrap();

418

419

                            // if we dont do this we have a backed candidate every 2 blocks

420

                            // we want

421

                            persisted_validation_data.relay_parent_storage_root =

422

                                parent_header.state_root;

423

424

                            let persisted_validation_data_hash = persisted_validation_data.hash();

425

                            // retrieve the validation code hash

426

                            let validation_code_hash = runtime_api

427

                                .validation_code_hash(

428

                                    parent,

429

                                    para[0],

430

                                    OccupiedCoreAssumption::Included,

431

432

                                .unwrap()

433

                                .unwrap();

434

                            let pov_hash = Default::default();

435

                            // generate a fake collator signature

436

                            let payload = polkadot_primitives::collator_signature_payload(

437

                                &parent,

438

                                &para[0],

439

                                &persisted_validation_data_hash,

440

                                &pov_hash,

441

                                &validation_code_hash,

442

);

443

                            let collator_signature = collator_pair.sign(&payload);

444

445

                            let upward_messages = UpwardMessages::try_from(

446

                                upward_messages_receiver.drain().collect::<Vec<_>>(),

447

448

                            .expect("create upward messages from raw messages");

449

450

                            // generate a candidate with most of the values mocked

451

                            let candidate = CommittedCandidateReceipt::<H256> {

452

                                descriptor: CandidateDescriptor::<H256> {

453

                                    para_id: para[0],

454

                                    relay_parent: parent,

455

                                    collator: collator_pair.public(),

456

                                    persisted_validation_data_hash,

457

                                    pov_hash,

458

                                    erasure_root: Default::default(),

459

                                    signature: collator_signature,

460

                                    para_head: parachain_mocked_header.clone().hash(),

461

                                    validation_code_hash,

462

},

463

                                commitments: CandidateCommitments::<u32> {

464

                                    upward_messages,

465

                                    horizontal_messages: Default::default(),

466

                                    new_validation_code: None,

467

                                    head_data: parachain_mocked_header.clone().encode().into(),

468

                                    processed_downward_messages: 0,

469

                                    hrmp_watermark: parent_header.number,

470

},

471

};

472

                            let candidate_hash = candidate.hash();

473

                            let payload = CompactStatement::Valid(candidate_hash);

474

475

                            let signature_ctx = SigningContext {

476

                                parent_hash: parent,

477

                                session_index: session_idx,

478

};

479

480

                            // sign the candidate with the validator key

481

                            let signature = keystore_sign(

482

                                &keystore,

483

                                payload,

484

                                &signature_ctx,

485

                                *index,

486

                                &validator,

487

488

                            .unwrap()

489

                            .unwrap()

490

                            .benchmark_signature();

491

492

                            // construct a validity vote

493

                            let validity_votes = vec![ValidityAttestation::Explicit(signature)];

494

495

                            // push the candidate

496

                            backed_cand.push(BackedCandidate::<H256>::new(

497

                                candidate,

498

                                validity_votes.clone(),

499

                                bitvec::bitvec![u8, bitvec::order::Lsb0; 1; indices_associated_to_core.len()],

500

                                core,

501

));

502

503

504

505

506

507

508

        Ok(ParachainsInherentData {

509

            bitfields,

510

            backed_candidates: backed_cand,

511

            disputes: Vec::new(),

512

            parent_header,

513

})

514

515

516

517

#[async_trait::async_trait]

518

impl<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> sp_inherents::InherentDataProvider

519

    for MockParachainsInherentDataProvider<C>

520

where

521

    C::Api: ParachainHost<Block>,

522

    C: AuxStore,

523

524

    async fn provide_inherent_data(

525

        &self,

526

        dst_inherent_data: &mut sp_inherents::InherentData,

527

    ) -> Result<(), sp_inherents::Error> {

528

        // fetch whether the para inherent selector has been set

529

        let maybe_para_selector = self

530

            .client

531

            .get_aux(PARA_INHERENT_SELECTOR_AUX_KEY)

532

            .expect("Should be able to query aux storage; qed");

533

534

        let inherent_data = {

535

            if let Some(aux) = maybe_para_selector {

536

                // if it is true, the candidates need to be mocked

537

                // else, we output the empty parachain inherent data provider

538

                if aux == true.encode() {

539

                    MockParachainsInherentDataProvider::create(

540

                        self.client.clone(),

541

                        self.parent,

542

                        self.keystore.clone(),

543

                        self.upward_messages_receiver.clone(),

544

                        self.container_chain_exclusion_receiver.clone(),

545

546

                    .await

547

                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?

548

                } else {

549

                    EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)

550

                        .await

551

                        .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?

552

553

            } else {

554

                EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)

555

                    .await

556

                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?

557

558

};

559

560

        dst_inherent_data.put_data(

561

            polkadot_primitives::PARACHAINS_INHERENT_IDENTIFIER,

562

            &inherent_data,

563

564

565

566

    async fn try_handle_error(

567

        &self,

568

        _identifier: &sp_inherents::InherentIdentifier,

569

        _error: &[u8],

570

    ) -> Option<Result<(), sp_inherents::Error>> {

571

        // Inherent isn't checked and can not return any error

572

        None

573

574

575

576

/// We store past timestamp we created in the aux storage, which enable us to return timestamp which is increased by

577

/// slot duration from previous timestamp or current timestamp if in reality more time is passed.

578

fn get_next_timestamp(

579

    client: Arc<FullClient>,

580

    slot_duration: SlotDuration,

581

) -> sp_timestamp::InherentDataProvider {

582

    const TIMESTAMP_AUX_KEY: &[u8] = b"__DEV_TIMESTAMP";

583

584

    let maybe_last_timestamp = client

585

        .get_aux(TIMESTAMP_AUX_KEY)

586

        .expect("Should be able to query aux storage; qed");

587

    if let Some(last_timestamp) = maybe_last_timestamp {

588

        let last_inherent_data = sp_timestamp::InherentType::decode(&mut last_timestamp.as_slice())

589

            .expect("Timestamp data must be decoded; qed");

590

        let new_inherent_data: sp_timestamp::InherentType = max(

591

            last_inherent_data.add(slot_duration.as_millis()),

592

            sp_timestamp::InherentType::current(),

593

);

594

        client

595

            .insert_aux(

596

                &[(TIMESTAMP_AUX_KEY, new_inherent_data.encode().as_slice())],

597

                &[],

598

599

            .expect("Should be able to write to aux storage; qed");

600

        sp_timestamp::InherentDataProvider::new(new_inherent_data)

601

    } else {

602

        let current_timestamp = sp_timestamp::InherentType::current();

603

        client

604

            .insert_aux(

605

                &[(TIMESTAMP_AUX_KEY, current_timestamp.encode().as_slice())],

606

                &[],

607

608

            .expect("Should be able to write to aux storage; qed");

609

        sp_timestamp::InherentDataProvider::new(current_timestamp)

610

611

612

613

60

fn new_full<

614

60

    OverseerGenerator: OverseerGen,

615

60

    Network: sc_network::NetworkBackend<Block, <Block as BlockT>::Hash>,

616

60

>(

617

60

    sealing: Sealing,

618

60

    mut config: Configuration,

619

60

    NewFullParams {

620

60

        telemetry_worker_handle,

621

60

..

622

60

    }: NewFullParams<OverseerGenerator>,

623

60

) -> Result<NewFull, Error> {

624

60

    let role = config.role;

625

626

60

    let basics = new_partial_basics(&mut config, telemetry_worker_handle)?;

627

628

60

    let prometheus_registry = config.prometheus_registry().cloned();

629

630

60

    let keystore = basics.keystore_container.local_keystore();

631

632

60

    let select_chain = SelectRelayChain::new_longest_chain(basics.backend.clone());

633

634

60

    let service::PartialComponents::<_, _, SelectRelayChain<_>, _, _, _> {

635

60

        client,

636

60

        backend,

637

60

        mut task_manager,

638

60

        keystore_container,

639

60

        select_chain,

640

60

        import_queue,

641

60

        transaction_pool,

642

60

        other: (block_import, babe_link, slot_duration, mut telemetry),

643

60

    } = new_partial::<SelectRelayChain<_>>(&mut config, basics, select_chain)?;

644

645

60

    let metrics = Network::register_notification_metrics(

646

60

        config.prometheus_config.as_ref().map(|cfg| &cfg.registry),

647

);

648

649

60

    let net_config = sc_network::config::FullNetworkConfiguration::<_, _, Network>::new(

650

60

        &config.network,

651

60

        prometheus_registry.clone(),

652

);

653

654

    // Create channels for mocked parachain candidates.

655

60

    let (downward_mock_para_inherent_sender, downward_mock_para_inherent_receiver) =

656

60

        flume::bounded::<Vec<u8>>(100);

657

658

60

    let (upward_mock_sender, upward_mock_receiver) = flume::bounded::<Vec<u8>>(100);

659

660

60

    let (network, system_rpc_tx, tx_handler_controller, sync_service) =

661

60

        service::build_network(service::BuildNetworkParams {

662

60

            config: &config,

663

60

            net_config,

664

60

            client: client.clone(),

665

60

            transaction_pool: transaction_pool.clone(),

666

60

            spawn_handle: task_manager.spawn_handle(),

667

60

            import_queue,

668

60

            block_announce_validator_builder: None,

669

60

            warp_sync_config: None,

670

60

            block_relay: None,

671

60

            metrics,

672

60

        })?;

673

674

60

    if config.offchain_worker.enabled {

675

        use futures::FutureExt;

676

677

60

        task_manager.spawn_handle().spawn(

678

            "offchain-workers-runner",

679

            "offchain-work",

680

60

            sc_offchain::OffchainWorkers::new(sc_offchain::OffchainWorkerOptions {

681

60

                runtime_api_provider: client.clone(),

682

60

                keystore: Some(keystore_container.keystore()),

683

60

                offchain_db: backend.offchain_storage(),

684

60

                transaction_pool: Some(OffchainTransactionPoolFactory::new(

685

60

                    transaction_pool.clone(),

686

60

)),

687

60

                network_provider: Arc::new(network.clone()),

688

60

                is_validator: role.is_authority(),

689

                enable_http_requests: false,

690

                custom_extensions: move |_| vec![],

691

})?

692

60

            .run(client.clone(), task_manager.spawn_handle())

693

60

            .boxed(),

694

);

695

696

697

60

    let mut command_sink = None;

698

60

    let mut container_chain_exclusion_sender = None;

699

700

60

    if role.is_authority() {

701

60

        let proposer = sc_basic_authorship::ProposerFactory::with_proof_recording(

702

60

            task_manager.spawn_handle(),

703

60

            client.clone(),

704

60

            transaction_pool.clone(),

705

60

            prometheus_registry.as_ref(),

706

60

            telemetry.as_ref().map(|x| x.handle()),

707

);

708

709

60

        let commands_stream: Box<

710

60

            dyn Stream<Item = EngineCommand<<Block as BlockT>::Hash>> + Send + Sync + Unpin,

711

60

        > = match sealing {

712

            Sealing::Instant => {

713

                Box::new(

714

                    // This bit cribbed from the implementation of instant seal.

715

                    transaction_pool.import_notification_stream().map(|_| {

716

                        EngineCommand::SealNewBlock {

717

                            create_empty: false,

718

                            finalize: false,

719

                            parent_hash: None,

720

                            sender: None,

721

722

}),

723

724

725

            Sealing::Manual => {

726

60

                let (sink, stream) = futures::channel::mpsc::channel(1000);

727

                // Keep a reference to the other end of the channel. It goes to the RPC.

728

60

                command_sink = Some(sink);

729

60

                Box::new(stream)

730

731

            Sealing::Interval(millis) => Box::new(StreamExt::map(

732

                Timer::interval(Duration::from_millis(millis)),

733

                |_| EngineCommand::SealNewBlock {

734

                    create_empty: true,

735

                    finalize: true,

736

                    parent_hash: None,

737

                    sender: None,

738

},

739

)),

740

};

741

60

        let keystore_clone = keystore.clone();

742

743

60

        let babe_config = babe_link.config();

744

60

        let babe_consensus_provider = BabeConsensusDataProvider::new(

745

60

            client.clone(),

746

60

            keystore,

747

60

            babe_link.epoch_changes().clone(),

748

60

            babe_config.authorities.clone(),

749

750

60

        .map_err(|babe_error| {

751

            Error::Consensus(consensus_common::Error::Other(babe_error.into()))

752

        })?;

753

754

60

        let (mock_container_chains_exclusion_sender, mock_container_chains_exclusion_receiver) =

755

60

            flume::bounded::<Vec<ParaId>>(100);

756

60

        container_chain_exclusion_sender = Some(mock_container_chains_exclusion_sender);

757

758

        // Need to clone it and store here to avoid moving of `client`

759

        // variable in closure below.

760

60

        let client_clone = client.clone();

761

762

60

        task_manager.spawn_essential_handle().spawn_blocking(

763

            "authorship_task",

764

60

            Some("block-authoring"),

765

60

            run_manual_seal(ManualSealParams {

766

60

                block_import,

767

60

                env: proposer,

768

60

                client: client.clone(),

769

60

                pool: transaction_pool.clone(),

770

60

                commands_stream,

771

60

                select_chain,

772

                create_inherent_data_providers: move |parent, ()| {

773

                    let client_clone = client_clone.clone();

774

                    let keystore = keystore_clone.clone();

775

                    let downward_mock_para_inherent_receiver = downward_mock_para_inherent_receiver.clone();

776

                    let upward_mock_receiver = upward_mock_receiver.clone();

777

                    let mock_container_chains_exclusion_receiver = mock_container_chains_exclusion_receiver.clone();

778

                    async move {

779

780

                        let downward_mock_para_inherent_receiver = downward_mock_para_inherent_receiver.clone();

781

                        // here we only take the last one

782

                        let para_inherent_decider_messages: Vec<Vec<u8>> = downward_mock_para_inherent_receiver.drain().collect();

783

784

                        let upward_messages_receiver = upward_mock_receiver.clone();

785

786

                        // If there is a value to be updated, we update it

787

                        if let Some(value) = para_inherent_decider_messages.last() {

788

                            client_clone

789

                            .insert_aux(

790

                                &[(PARA_INHERENT_SELECTOR_AUX_KEY, value.as_slice())],

791

                                &[],

792

793

                            .expect("Should be able to write to aux storage; qed");

794

795

796

                        let parachain = MockParachainsInherentDataProvider::new(

797

                            client_clone.clone(),

798

                            parent,

799

                            keystore,

800

                            upward_messages_receiver,

801

                            mock_container_chains_exclusion_receiver

802

);

803

804

                        let timestamp = get_next_timestamp(client_clone, slot_duration);

805

806

                        let slot =

807

                            sp_consensus_babe::inherents::InherentDataProvider::from_timestamp_and_slot_duration(

808

                                *timestamp,

809

                                slot_duration,

810

);

811

812

                        Ok((slot, timestamp, parachain))

813

814

},

815

60

                consensus_data_provider: Some(Box::new(babe_consensus_provider)),

816

}),

817

);

818

819

820

60

    let dev_rpc = if role.clone().is_authority() {

821

60

        Some(DevRpc {

822

60

            mock_para_inherent_channel: downward_mock_para_inherent_sender,

823

60

            upward_message_channel: upward_mock_sender,

824

60

})

825

    } else {

826

        None

827

};

828

829

60

    let rpc_extensions_builder = {

830

60

        let client = client.clone();

831

60

        let transaction_pool = transaction_pool.clone();

832

833

        move |_subscription_executor: polkadot_rpc::SubscriptionTaskExecutor|

834

120

            -> Result<RpcExtension, service::Error> {

835

120

            let deps = DevDeps {

836

120

                client: client.clone(),

837

120

                pool: transaction_pool.clone(),

838

120

                command_sink: command_sink.clone(),

839

120

                dev_rpc: dev_rpc.clone(),

840

120

                container_chain_exclusion_sender: container_chain_exclusion_sender.clone(),

841

120

};

842

843

120

            create_dev_rpc_extension(deps).map_err(Into::into)

844

120

845

};

846

847

60

    let rpc_handlers = service::spawn_tasks(service::SpawnTasksParams {

848

60

        config,

849

60

        backend: backend.clone(),

850

60

        client: client.clone(),

851

60

        keystore: keystore_container.keystore(),

852

60

        network: network.clone(),

853

60

        sync_service: sync_service.clone(),

854

60

        rpc_builder: Box::new(rpc_extensions_builder),

855

60

        transaction_pool: transaction_pool.clone(),

856

60

        task_manager: &mut task_manager,

857

60

        system_rpc_tx,

858

60

        tx_handler_controller,

859

60

        telemetry: telemetry.as_mut(),

860

60

    })?;

861

862

60

    Ok(NewFull {

863

60

        task_manager,

864

60

        client,

865

60

        overseer_handle: None,

866

60

        network,

867

60

        sync_service,

868

60

        rpc_handlers,

869

60

        backend,

870

60

})

871

60

872

873

60

fn new_partial<ChainSelection>(

874

60

    config: &mut Configuration,

875

60

    Basics {

876

60

        task_manager,

877

60

        backend,

878

60

        client,

879

60

        keystore_container,

880

60

        telemetry,

881

60

    }: Basics,

882

60

    select_chain: ChainSelection,

883

60

) -> Result<

884

60

    service::PartialComponents<

885

60

        FullClient,

886

60

        FullBackend,

887

60

        ChainSelection,

888

60

        sc_consensus::DefaultImportQueue<Block>,

889

60

        sc_transaction_pool::TransactionPoolHandle<Block, FullClient>,

890

60

891

60

            BabeBlockImport<Block, FullClient, Arc<FullClient>>,

892

60

            BabeLink<Block>,

893

60

            SlotDuration,

894

60

            Option<Telemetry>,

895

60

),

896

60

>,

897

60

    Error,

898

60

899

60

where

900

60

    ChainSelection: 'static + SelectChain<Block>,

901

902

60

    let transaction_pool = sc_transaction_pool::Builder::new(

903

60

        task_manager.spawn_essential_handle(),

904

60

        client.clone(),

905

60

        config.role.is_authority().into(),

906

907

60

    .with_options(config.transaction_pool.clone())

908

60

    .with_prometheus(config.prometheus_registry())

909

60

    .build();

910

911

    // Create babe block import queue; this is required to have correct epoch data

912

    // available for manual seal to produce block

913

60

    let babe_config = babe::configuration(&*client)?;

914

60

    let (babe_block_import, babe_link) =

915

60

        babe::block_import(babe_config.clone(), client.clone(), client.clone())?;

916

60

    let slot_duration = babe_link.config().slot_duration();

917

918

    // Create manual seal block import with manual seal block import queue

919

60

    let import_queue = sc_consensus_manual_seal::import_queue(

920

60

        Box::new(babe_block_import.clone()),

921

60

        &task_manager.spawn_essential_handle(),

922

60

        config.prometheus_registry(),

923

);

924

925

60

    Ok(service::PartialComponents {

926

60

        client,

927

60

        backend,

928

60

        task_manager,

929

60

        keystore_container,

930

60

        select_chain,

931

60

        import_queue,

932

60

        transaction_pool: transaction_pool.into(),

933

60

        other: (babe_block_import, babe_link, slot_duration, telemetry),

934

60

})

935

60

936

937

60

fn new_partial_basics(

938

60

    config: &mut Configuration,

939

60

    telemetry_worker_handle: Option<TelemetryWorkerHandle>,

940

60

) -> Result<Basics, Error> {

941

60

    let telemetry = config

942

60

        .telemetry_endpoints

943

60

        .clone()

944

60

        .filter(|x| !x.is_empty())

945

60

        .map(move |endpoints| -> Result<_, telemetry::Error> {

946

            let (worker, mut worker_handle) = if let Some(worker_handle) = telemetry_worker_handle {

947

                (None, worker_handle)

948

            } else {

949

                let worker = TelemetryWorker::new(16)?;

950

                let worker_handle = worker.handle();

951

                (Some(worker), worker_handle)

952

};

953

            let telemetry = worker_handle.new_telemetry(endpoints);

954

            Ok((worker, telemetry))

955

})

956

60

        .transpose()?;

957

958

60

    let heap_pages = config

959

60

        .executor

960

60

        .default_heap_pages

961

60

        .map_or(DEFAULT_HEAP_ALLOC_STRATEGY, |h| HeapAllocStrategy::Static {

962

            extra_pages: h as u32,

963

});

964

965

60

    let mut wasm_builder = WasmExecutor::builder()

966

60

        .with_execution_method(config.executor.wasm_method)

967

60

        .with_onchain_heap_alloc_strategy(heap_pages)

968

60

        .with_offchain_heap_alloc_strategy(heap_pages)

969

60

        .with_max_runtime_instances(config.executor.max_runtime_instances)

970

60

        .with_runtime_cache_size(config.executor.runtime_cache_size);

971

60

    if let Some(ref wasmtime_precompiled_path) = config.executor.wasmtime_precompiled {

972

        wasm_builder = wasm_builder.with_wasmtime_precompiled_path(wasmtime_precompiled_path);

973

60

974

60

    let executor = wasm_builder.build();

975

976

60

    let (client, backend, keystore_container, task_manager) =

977

60

        service::new_full_parts::<Block, RuntimeApi, _>(

978

60

            config,

979

60

            telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),

980

60

            executor,

981

)?;

982

60

    let client = Arc::new(client);

983

984

60

    let telemetry = telemetry.map(|(worker, telemetry)| {

985

        if let Some(worker) = worker {

986

            task_manager.spawn_handle().spawn(

987

                "telemetry",

988

                Some("telemetry"),

989

                Box::pin(worker.run()),

990

);

991

992

        telemetry

993

});

994

995

60

    Ok(Basics {

996

60

        task_manager,

997

60

        client,

998

60

        backend,

999

60

        keystore_container,

1000

60

        telemetry,

1001

60

})

1002

60

1003

1004

use {

1005

    polkadot_primitives::{AvailabilityBitfield, UncheckedSigned, ValidatorId, ValidatorIndex},

1006

    sp_keystore::Error as KeystoreError,

1007

};

1008

fn keystore_sign<H: Encode, Payload: Encode>(

1009

    keystore: &KeystorePtr,

1010

    payload: Payload,

1011

    context: &SigningContext<H>,

1012

    validator_index: ValidatorIndex,

1013

    key: &ValidatorId,

1014

) -> Result<Option<UncheckedSigned<Payload>>, KeystoreError> {

1015

    let data = payload_data(&payload, context);

1016

    let signature = keystore

1017

        .sr25519_sign(ValidatorId::ID, key.as_ref(), &data)?

1018

        .map(|sig| UncheckedSigned::new(payload, validator_index, sig.into()));

1019

    Ok(signature)

1020

1021

1022

fn payload_data<H: Encode, Payload: Encode>(

1023

    payload: &Payload,

1024

    context: &SigningContext<H>,

1025

) -> Vec<u8> {

1026

    // equivalent to (`real_payload`, context).encode()

1027

    let mut out = payload.encode_as();

1028

    out.extend(context.encode());

1029

out

1030

1031

1032

/// Create an `AvailabilityBitfield` with size `total_cores`. The first `used_cores` set to true (occupied),

1033

/// and the remaining to false (available).

1034

fn availability_bitvec(used_cores: usize, total_cores: usize) -> AvailabilityBitfield {

1035

    let mut bitfields = bitvec::bitvec![u8, bitvec::order::Lsb0; 0; 0];

1036

    for i in 0..total_cores {

1037

        if i < used_cores {

1038

            bitfields.push(true);

1039

        } else {

1040

            bitfields.push(false)

1041

1042

1043

1044

    bitfields.into()

1045