Grcov report - service.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

use {

18

    crate::cli::ContainerChainCli,

19

    crate::rpc::generate_rpc_builder::{GenerateRpcBuilder, GenerateRpcBuilderParams},

20

    cumulus_client_consensus_common::{

21

        ParachainBlockImport as TParachainBlockImport, ParachainBlockImportMarker,

22

},

23

    cumulus_client_service::{

24

        prepare_node_config, start_relay_chain_tasks, DARecoveryProfile, ParachainHostFunctions,

25

        StartRelayChainTasksParams,

26

},

27

    cumulus_primitives_core::ParaId,

28

    cumulus_relay_chain_interface::{call_runtime_api, OverseerHandle, RelayChainInterface},

29

    dancebox_runtime::{

30

        opaque::{Block, Hash},

31

        RuntimeApi,

32

},

33

    dc_orchestrator_chain_interface::OrchestratorChainInterface,

34

    dp_slot_duration_runtime_api::TanssiSlotDurationApi,

35

    nimbus_primitives::{NimbusId, NimbusPair},

36

    node_common::service::node_builder::{

37

        MinimalCumulusRuntimeApi, NodeBuilder, NodeBuilderConfig,

38

},

39

    sc_basic_authorship::ProposerFactory,

40

    sc_consensus::{BasicQueue, BlockImport},

41

    sc_executor::WasmExecutor,

42

    sc_network::NetworkBackend,

43

    sc_network::NetworkBlock,

44

    sc_network_sync::SyncingService,

45

    sc_service::{

46

        Configuration, ImportQueue, SpawnTaskHandle, TFullBackend, TFullClient, TaskManager,

47

},

48

    sc_telemetry::TelemetryHandle,

49

    sc_tracing::tracing::Instrument,

50

    sc_transaction_pool::TransactionPoolHandle,

51

    sp_api::ProvideRuntimeApi,

52

    sp_consensus::EnableProofRecording,

53

    sp_consensus_aura::SlotDuration,

54

    sp_keystore::KeystorePtr,

55

    std::{marker::PhantomData, sync::Arc, time::Duration},

56

    substrate_prometheus_endpoint::Registry,

57

    tc_consensus::{

58

        collators::lookahead::{

59

            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,

60

},

61

        OrchestratorAuraWorkerAuxData,

62

},

63

    tokio_util::sync::CancellationToken,

64

};

65

66

#[allow(deprecated)]

67

use sc_executor::NativeElseWasmExecutor;

68

69

type FullBackend = TFullBackend<Block>;

70

71

/// Native executor type.

72

pub struct ParachainNativeExecutor;

73

74

impl sc_executor::NativeExecutionDispatch for ParachainNativeExecutor {

75

    type ExtendHostFunctions = ParachainHostFunctions;

76

77

300918

    fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {

78

300918

        dancebox_runtime::api::dispatch(method, data)

79

300918

80

81

2970

    fn native_version() -> sc_executor::NativeVersion {

82

2970

        dancebox_runtime::native_version()

83

2970

84

85

86

#[derive(Default, Copy, Clone)]

87

pub struct ContainerChainNodeConfig<RuntimeApi>(PhantomData<RuntimeApi>);

88

impl<RuntimeApi> NodeBuilderConfig for ContainerChainNodeConfig<RuntimeApi> {

89

    type Block = Block;

90

    /// RuntimeApi is customizable to allow supporting more features than the common subset of

91

    /// runtime api features.

92

    type RuntimeApi = RuntimeApi;

93

    type ParachainExecutor = ContainerChainExecutor;

94

95

96

impl<RuntimeApi> ContainerChainNodeConfig<RuntimeApi> {

97

    pub fn new() -> Self {

98

        Self(PhantomData)

99

100

101

102

/// Orchestrator Parachain Block import. We cannot use the one in cumulus as it overrides the best

103

/// chain selection rule

104

#[derive(Clone)]

105

pub struct OrchestratorParachainBlockImport<BI> {

106

    inner: BI,

107

108

109

impl<BI> OrchestratorParachainBlockImport<BI> {

110

    /// Create a new instance.

111

196

    pub fn new(inner: BI) -> Self {

112

196

        Self { inner }

113

196

114

115

116

/// We simply rely on the inner

117

#[async_trait::async_trait]

118

impl<BI> BlockImport<Block> for OrchestratorParachainBlockImport<BI>

119

where

120

    BI: BlockImport<Block> + Send + Sync,

121

122

    type Error = BI::Error;

123

124

    async fn check_block(

125

        &self,

126

        block: sc_consensus::BlockCheckParams<Block>,

127

    ) -> Result<sc_consensus::ImportResult, Self::Error> {

128

        self.inner.check_block(block).await

129

130

131

    async fn import_block(

132

        &self,

133

        params: sc_consensus::BlockImportParams<Block>,

134

7890

    ) -> Result<sc_consensus::ImportResult, Self::Error> {

135

7890

        let res = self.inner.import_block(params).await?;

136

137

7890

        Ok(res)

138

15780

139

140

141

/// But we need to implement the ParachainBlockImportMarker trait to fullfil

142

impl<BI> ParachainBlockImportMarker for OrchestratorParachainBlockImport<BI> {}

143

144

// Orchestrator chain types

145

#[allow(deprecated)]

146

pub type ParachainExecutor = NativeElseWasmExecutor<ParachainNativeExecutor>;

147

pub type ParachainClient = TFullClient<Block, RuntimeApi, ParachainExecutor>;

148

pub type ParachainBackend = TFullBackend<Block>;

149

pub type DevParachainBlockImport = OrchestratorParachainBlockImport<Arc<ParachainClient>>;

150

pub type ParachainBlockImport =

151

    TParachainBlockImport<Block, Arc<ParachainClient>, ParachainBackend>;

152

pub type ParachainProposerFactory = ProposerFactory<

153

    TransactionPoolHandle<Block, ParachainClient>,

154

    ParachainClient,

155

    EnableProofRecording,

156

>;

157

158

// Container chains types

159

type ContainerChainExecutor = WasmExecutor<ParachainHostFunctions>;

160

pub type ContainerChainClient<RuntimeApi> = TFullClient<Block, RuntimeApi, ContainerChainExecutor>;

161

pub type ContainerChainBackend = TFullBackend<Block>;

162

type ContainerChainBlockImport<RuntimeApi> =

163

    TParachainBlockImport<Block, Arc<ContainerChainClient<RuntimeApi>>, ContainerChainBackend>;

164

165

tp_traits::alias!(

166

    pub trait MinimalContainerRuntimeApi:

167

        MinimalCumulusRuntimeApi<Block, ContainerChainClient<Self>>

168

        + sp_api::ConstructRuntimeApi<

169

            Block,

170

            ContainerChainClient<Self>,

171

            RuntimeApi:

172

                TanssiSlotDurationApi<Block>

173

                + async_backing_primitives::UnincludedSegmentApi<Block>,

174

175

        + Sized

176

);

177

178

/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.

179

///

180

/// This is the actual implementation that is abstract over the executor and the runtime api.

181

pub fn start_node_impl_container<

182

'a,

183

    RuntimeApi: MinimalContainerRuntimeApi + 'a,

184

    TGenerateRpcBuilder: GenerateRpcBuilder<RuntimeApi> + 'a,

185

    Net: NetworkBackend<Block, Hash>,

186

>(

187

    parachain_config: Configuration,

188

    relay_chain_interface: Arc<dyn RelayChainInterface>,

189

    orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>,

190

    keystore: KeystorePtr,

191

    para_id: ParaId,

192

    collation_params: Option<crate::spawner::CollationParams>,

193

    generate_rpc_builder: TGenerateRpcBuilder,

194

    container_chain_cli: &'a ContainerChainCli,

195

    data_preserver: bool,

196

) -> impl std::future::Future<

197

    Output = sc_service::error::Result<(

198

        TaskManager,

199

        Arc<ContainerChainClient<RuntimeApi>>,

200

        Arc<ParachainBackend>,

201

)>,

202

> + 'a {

203

    async move {

204

        let parachain_config = prepare_node_config(parachain_config);

205

206

        // Create a `NodeBuilder` which helps setup parachain nodes common systems.

207

        let node_builder = ContainerChainNodeConfig::new_builder(&parachain_config, None)?;

208

209

        let (block_import, import_queue) = container_chain_import_queue(

210

            &parachain_config,

211

            &node_builder,

212

            container_chain_cli,

213

            data_preserver,

214

);

215

        let import_queue_service = import_queue.service();

216

217

        let node_builder = node_builder

218

            .build_cumulus_network::<_, Net>(

219

                &parachain_config,

220

                para_id,

221

                import_queue,

222

                relay_chain_interface.clone(),

223

224

            .await?;

225

226

        let force_authoring = parachain_config.force_authoring;

227

228

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

229

230

        let rpc_builder = generate_rpc_builder.generate(GenerateRpcBuilderParams {

231

            task_manager: &node_builder.task_manager,

232

            container_chain_config: &parachain_config,

233

            client: node_builder.client.clone(),

234

            backend: node_builder.backend.clone(),

235

            sync_service: node_builder.network.sync_service.clone(),

236

            transaction_pool: node_builder.transaction_pool.clone(),

237

            prometheus_registry: node_builder.prometheus_registry.clone(),

238

            command_sink: None,

239

            xcm_senders: None,

240

            network: node_builder.network.network.clone(),

241

        })?;

242

243

        let node_builder = node_builder.spawn_common_tasks(parachain_config, rpc_builder)?;

244

245

        let announce_block = {

246

            let sync_service = node_builder.network.sync_service.clone();

247

            Arc::new(move |hash, data| sync_service.announce_block(hash, data))

248

};

249

250

        let relay_chain_slot_duration = Duration::from_secs(6);

251

252

        let overseer_handle = relay_chain_interface

253

            .overseer_handle()

254

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

255

        let (mut node_builder, _) = node_builder.extract_import_queue_service();

256

257

        start_relay_chain_tasks(StartRelayChainTasksParams {

258

            client: node_builder.client.clone(),

259

            announce_block: announce_block.clone(),

260

            para_id,

261

            relay_chain_interface: relay_chain_interface.clone(),

262

            task_manager: &mut node_builder.task_manager,

263

            da_recovery_profile: if collation_params.is_some() {

264

                DARecoveryProfile::Collator

265

            } else {

266

                DARecoveryProfile::FullNode

267

},

268

            import_queue: import_queue_service,

269

            relay_chain_slot_duration,

270

            recovery_handle: Box::new(overseer_handle.clone()),

271

            sync_service: node_builder.network.sync_service.clone(),

272

        })?;

273

274

        if let Some(collation_params) = collation_params {

275

            let node_spawn_handle = node_builder.task_manager.spawn_handle().clone();

276

            let node_client = node_builder.client.clone();

277

            let node_backend = node_builder.backend.clone();

278

279

            start_consensus_container(

280

                node_client.clone(),

281

                node_backend.clone(),

282

                collation_params,

283

                block_import.clone(),

284

                prometheus_registry.clone(),

285

                node_builder.telemetry.as_ref().map(|t| t.handle()).clone(),

286

                node_spawn_handle.clone(),

287

                relay_chain_interface.clone(),

288

                orchestrator_chain_interface.clone(),

289

                node_builder.transaction_pool.clone(),

290

                node_builder.network.sync_service.clone(),

291

                keystore.clone(),

292

                force_authoring,

293

                relay_chain_slot_duration,

294

                para_id,

295

                overseer_handle.clone(),

296

                announce_block.clone(),

297

                container_chain_cli.base.experimental_max_pov_percentage,

298

);

299

300

301

        Ok((

302

            node_builder.task_manager,

303

            node_builder.client,

304

            node_builder.backend,

305

))

306

307

    .instrument(sc_tracing::tracing::info_span!(

308

        sc_tracing::logging::PREFIX_LOG_SPAN,

309

        name = container_log_str(para_id),

310

))

311

312

313

pub fn container_chain_import_queue<RuntimeApi: MinimalContainerRuntimeApi>(

314

    parachain_config: &Configuration,

315

    node_builder: &NodeBuilder<ContainerChainNodeConfig<RuntimeApi>>,

316

    container_chain_cli: &ContainerChainCli,

317

    data_preserver: bool,

318

) -> (ContainerChainBlockImport<RuntimeApi>, BasicQueue<Block>) {

319

    // The nimbus import queue ONLY checks the signature correctness

320

    // Any other checks corresponding to the author-correctness should be done

321

    // in the runtime

322

    let block_import =

323

        ContainerChainBlockImport::new(node_builder.client.clone(), node_builder.backend.clone());

324

325

    // Disable gap creation to check if that avoids block history download in warp sync.

326

    // Create gap means download block history. If the user passes `--download-block-history`, we

327

    // set dont_create_gap=false, so create_gap=true, which is the default behavior in polkadot.

328

    let dont_create_gap = !container_chain_cli.base.download_block_history.unwrap_or(

329

        // Default value for download_block_history:

330

        // false if running a collator

331

        // true if running a data preserver node

332

        data_preserver,

333

);

334

335

    let import_queue = nimbus_consensus::import_queue(

336

        node_builder.client.clone(),

337

        block_import.clone(),

338

        move |_, _| async move {

339

            let time = sp_timestamp::InherentDataProvider::from_system_time();

340

341

            Ok((time,))

342

},

343

        &node_builder.task_manager.spawn_essential_handle(),

344

        parachain_config.prometheus_registry(),

345

        false,

346

        dont_create_gap,

347

348

    .expect("function never fails");

349

350

    (block_import, import_queue)

351

352

353

fn start_consensus_container<RuntimeApi: MinimalContainerRuntimeApi>(

354

    client: Arc<ContainerChainClient<RuntimeApi>>,

355

    backend: Arc<FullBackend>,

356

    collation_params: crate::spawner::CollationParams,

357

    block_import: ContainerChainBlockImport<RuntimeApi>,

358

    prometheus_registry: Option<Registry>,

359

    telemetry: Option<TelemetryHandle>,

360

    spawner: SpawnTaskHandle,

361

    relay_chain_interface: Arc<dyn RelayChainInterface>,

362

    orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>,

363

    transaction_pool: Arc<

364

        sc_transaction_pool::TransactionPoolHandle<Block, ContainerChainClient<RuntimeApi>>,

365

>,

366

    sync_oracle: Arc<SyncingService<Block>>,

367

    keystore: KeystorePtr,

368

    force_authoring: bool,

369

    relay_chain_slot_duration: Duration,

370

    para_id: ParaId,

371

    overseer_handle: OverseerHandle,

372

    announce_block: Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,

373

    max_pov_percentage: Option<u32>,

374

) {

375

    let crate::spawner::CollationParams {

376

        collator_key,

377

        orchestrator_tx_pool,

378

        orchestrator_client,

379

        orchestrator_para_id,

380

        solochain,

381

    } = collation_params;

382

    let slot_duration = if solochain {

383

        // Solochains use Babe instead of Aura, which has 6s slot duration

384

        let relay_slot_ms = relay_chain_slot_duration.as_millis();

385

        SlotDuration::from_millis(

386

            u64::try_from(relay_slot_ms).expect("relay chain slot duration overflows u64"),

387

388

    } else {

389

        cumulus_client_consensus_aura::slot_duration(

390

            orchestrator_client

391

                .as_deref()

392

                .expect("solochain is false, orchestrator_client must be Some"),

393

394

        .expect("start_consensus_container: slot duration should exist")

395

};

396

397

    let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(

398

        spawner.clone(),

399

        client.clone(),

400

        transaction_pool,

401

        prometheus_registry.as_ref(),

402

        telemetry.clone(),

403

);

404

405

    let proposer = cumulus_client_consensus_proposer::Proposer::new(proposer_factory);

406

407

    let collator_service = cumulus_client_collator::service::CollatorService::new(

408

        client.clone(),

409

        Arc::new(spawner.clone()),

410

        announce_block,

411

        client.clone(),

412

);

413

414

    let relay_chain_interace_for_cidp = relay_chain_interface.clone();

415

    let relay_chain_interace_for_orch = relay_chain_interface.clone();

416

    let orchestrator_client_for_cidp = orchestrator_client.clone();

417

    let client_for_cidp = client.clone();

418

    let client_for_hash_provider = client.clone();

419

    let client_for_slot_duration = client.clone();

420

421

    let code_hash_provider = move |block_hash| {

422

        client_for_hash_provider

423

            .code_at(block_hash)

424

            .ok()

425

            .map(polkadot_primitives::ValidationCode)

426

            .map(|c| c.hash())

427

};

428

    let buy_core_params = if solochain {

429

        BuyCoreParams::Solochain {}

430

    } else {

431

        BuyCoreParams::Orchestrator {

432

            orchestrator_tx_pool: orchestrator_tx_pool

433

                .expect("solochain is false, orchestrator_tx_pool must be Some"),

434

            orchestrator_client: orchestrator_client

435

                .expect("solochain is false, orchestrator_client must be Some"),

436

437

};

438

439

    let params = LookaheadTanssiAuraParams {

440

        max_pov_percentage,

441

        get_current_slot_duration: move |block_hash| {

442

            // Default to 12s if runtime API does not exist

443

            let slot_duration_ms = client_for_slot_duration

444

                .runtime_api()

445

                .slot_duration(block_hash)

446

                .unwrap_or(12_000);

447

448

            SlotDuration::from_millis(slot_duration_ms)

449

},

450

        create_inherent_data_providers: move |block_hash, (relay_parent, _validation_data)| {

451

            let relay_chain_interface = relay_chain_interace_for_cidp.clone();

452

            let orchestrator_chain_interface = orchestrator_chain_interface.clone();

453

            let client = client_for_cidp.clone();

454

455

            async move {

456

                let authorities_noting_inherent = if solochain {

457

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at_solochain(

458

                        relay_parent,

459

                        &relay_chain_interface,

460

461

                        .await

462

                } else {

463

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at(

464

                        relay_parent,

465

                        &relay_chain_interface,

466

                        &orchestrator_chain_interface,

467

                        orchestrator_para_id,

468

469

                        .await

470

};

471

472

                let slot_duration = {

473

                    // Default to 12s if runtime API does not exist

474

                    let slot_duration_ms = client

475

                        .runtime_api()

476

                        .slot_duration(block_hash)

477

                        .unwrap_or(12_000);

478

479

                    SlotDuration::from_millis(slot_duration_ms)

480

};

481

482

                let timestamp = sp_timestamp::InherentDataProvider::from_system_time();

483

484

                let slot =

485

						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(

486

							*timestamp,

487

							slot_duration,

488

);

489

490

                let authorities_noting_inherent = authorities_noting_inherent.ok_or_else(|| {

491

                    Box::<dyn std::error::Error + Send + Sync>::from(

492

                        "Failed to create authoritiesnoting inherent",

493

494

                })?;

495

496

                Ok((slot, timestamp, authorities_noting_inherent))

497

498

},

499

        get_orchestrator_aux_data: move |_block_hash, (relay_parent, _validation_data)| {

500

            let relay_chain_interace_for_orch = relay_chain_interace_for_orch.clone();

501

            let orchestrator_client_for_cidp = orchestrator_client_for_cidp.clone();

502

503

            async move {

504

                if solochain {

505

                    let authorities: Option<Vec<NimbusId>> = call_runtime_api(

506

                        &relay_chain_interace_for_orch,

507

                        "TanssiAuthorityAssignmentApi_para_id_authorities",

508

                        relay_parent,

509

                        &para_id,

510

511

                    .await?;

512

513

                    let authorities = authorities.ok_or_else(|| {

514

                        Box::<dyn std::error::Error + Send + Sync>::from(

515

                            "Failed to fetch authorities with error",

516

517

                    })?;

518

519

                    log::info!(

520

                        "Authorities {:?} found for header {:?}",

521

                        authorities,

522

                        relay_parent

523

);

524

525

                    let slot_freq: Option<_> = call_runtime_api(

526

                        &relay_chain_interace_for_orch,

527

                        "OnDemandBlockProductionApi_parathread_slot_frequency",

528

                        relay_parent,

529

                        &para_id,

530

531

                    .await?;

532

533

                    let aux_data = OrchestratorAuraWorkerAuxData {

534

                        authorities,

535

                        slot_freq,

536

};

537

538

                    Ok(aux_data)

539

                } else {

540

                    let latest_header =

541

                        ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::get_latest_orchestrator_head_info(

542

                            relay_parent,

543

                            &relay_chain_interace_for_orch,

544

                            orchestrator_para_id,

545

546

                            .await;

547

548

                    let latest_header = latest_header.ok_or_else(|| {

549

                        Box::<dyn std::error::Error + Send + Sync>::from(

550

                            "Failed to fetch latest header",

551

552

                    })?;

553

554

                    let authorities = tc_consensus::authorities::<Block, ParachainClient, NimbusPair>(

555

                        orchestrator_client_for_cidp

556

                            .as_ref()

557

                            .expect("solochain is false, orchestrator_client must be Some"),

558

                        &latest_header.hash(),

559

                        para_id,

560

);

561

562

                    let authorities = authorities.ok_or_else(|| {

563

                        Box::<dyn std::error::Error + Send + Sync>::from(

564

                            "Failed to fetch authorities with error",

565

566

                    })?;

567

568

                    log::info!(

569

                        "Authorities {:?} found for header {:?}",

570

                        authorities,

571

                        latest_header

572

);

573

574

                    let slot_freq = tc_consensus::min_slot_freq::<Block, ParachainClient, NimbusPair>(

575

                        orchestrator_client_for_cidp

576

                            .as_ref()

577

                            .expect("solochain is false, orchestrator_client must be Some"),

578

                        &latest_header.hash(),

579

                        para_id,

580

);

581

582

                    let aux_data = OrchestratorAuraWorkerAuxData {

583

                        authorities,

584

                        slot_freq,

585

};

586

587

                    Ok(aux_data)

588

589

590

},

591

        block_import,

592

        para_client: client,

593

        relay_client: relay_chain_interface,

594

        sync_oracle,

595

        keystore,

596

        collator_key,

597

        para_id,

598

        overseer_handle,

599

        orchestrator_slot_duration: slot_duration,

600

        force_authoring,

601

        relay_chain_slot_duration,

602

        proposer,

603

        collator_service,

604

        authoring_duration: Duration::from_millis(2000),

605

        para_backend: backend,

606

        code_hash_provider,

607

        // This cancellation token is no-op as it is not shared outside.

608

        cancellation_token: CancellationToken::new(),

609

        buy_core_params,

610

};

611

612

    let (fut, _exit_notification_receiver) =

613

        lookahead_tanssi_aura::run::<_, Block, NimbusPair, _, _, _, _, _, _, _, _, _, _, _, _, _>(

614

            params,

615

);

616

    spawner.spawn("tanssi-aura-container", None, fut);

617

618

619

// Log string that will be shown for the container chain: `[Container-2000]`.

620

// This needs to be a separate function because the `prefix_logs_with` macro

621

// has trouble parsing expressions.

622

fn container_log_str(para_id: ParaId) -> String {

623

    format!("Container-{}", para_id)

624