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::rpc::generate_rpc_builder::{GenerateRpcBuilder, GenerateRpcBuilderParams},

19

    cumulus_client_consensus_common::{

20

        ParachainBlockImport as TParachainBlockImport, ParachainBlockImportMarker,

21

},

22

    cumulus_client_service::{

23

        prepare_node_config, start_relay_chain_tasks, DARecoveryProfile, ParachainHostFunctions,

24

        StartRelayChainTasksParams,

25

},

26

    cumulus_primitives_core::ParaId,

27

    cumulus_relay_chain_interface::{call_runtime_api, OverseerHandle, RelayChainInterface},

28

    dancebox_runtime::{

29

        opaque::{Block, Hash},

30

        RuntimeApi,

31

},

32

    dc_orchestrator_chain_interface::OrchestratorChainInterface,

33

    dp_slot_duration_runtime_api::TanssiSlotDurationApi,

34

    nimbus_primitives::{NimbusId, NimbusPair},

35

    node_common::service::{MinimalCumulusRuntimeApi, NodeBuilder, NodeBuilderConfig},

36

    sc_basic_authorship::ProposerFactory,

37

    sc_consensus::{BasicQueue, BlockImport},

38

    sc_executor::WasmExecutor,

39

    sc_network::NetworkBlock,

40

    sc_network_sync::SyncingService,

41

    sc_service::{

42

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

43

},

44

    sc_telemetry::TelemetryHandle,

45

    sc_tracing::tracing::Instrument,

46

    sc_transaction_pool::TransactionPoolHandle,

47

    sp_api::ProvideRuntimeApi,

48

    sp_consensus::EnableProofRecording,

49

    sp_consensus_aura::SlotDuration,

50

    sp_keystore::KeystorePtr,

51

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

52

    substrate_prometheus_endpoint::Registry,

53

    tc_consensus::{

54

        collators::lookahead::{

55

            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,

56

},

57

        OrchestratorAuraWorkerAuxData,

58

},

59

    tokio_util::sync::CancellationToken,

60

};

61

62

use crate::cli::ContainerChainCli;

63

#[allow(deprecated)]

64

use sc_executor::NativeElseWasmExecutor;

65

66

type FullBackend = TFullBackend<Block>;

67

68

/// Native executor type.

69

pub struct ParachainNativeExecutor;

70

71

impl sc_executor::NativeExecutionDispatch for ParachainNativeExecutor {

72

    type ExtendHostFunctions = ParachainHostFunctions;

73

74

297786

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

75

297786

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

76

297786

77

78

2940

    fn native_version() -> sc_executor::NativeVersion {

79

2940

        dancebox_runtime::native_version()

80

2940

81

82

83

#[derive(Default, Copy, Clone)]

84

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

85

impl<RuntimeApi> NodeBuilderConfig for ContainerChainNodeConfig<RuntimeApi> {

86

    type Block = Block;

87

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

88

    /// runtime api features.

89

    type RuntimeApi = RuntimeApi;

90

    type ParachainExecutor = ContainerChainExecutor;

91

92

93

impl<RuntimeApi> ContainerChainNodeConfig<RuntimeApi> {

94

    pub fn new() -> Self {

95

        Self(PhantomData)

96

97

98

99

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

100

/// chain selection rule

101

#[derive(Clone)]

102

pub struct OrchestratorParachainBlockImport<BI> {

103

    inner: BI,

104

105

106

impl<BI> OrchestratorParachainBlockImport<BI> {

107

    /// Create a new instance.

108

194

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

109

194

        Self { inner }

110

194

111

112

113

/// We simply rely on the inner

114

#[async_trait::async_trait]

115

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

116

where

117

    BI: BlockImport<Block> + Send + Sync,

118

119

    type Error = BI::Error;

120

121

    async fn check_block(

122

        &self,

123

        block: sc_consensus::BlockCheckParams<Block>,

124

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

125

        self.inner.check_block(block).await

126

127

128

    async fn import_block(

129

        &self,

130

        params: sc_consensus::BlockImportParams<Block>,

131

7810

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

132

7810

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

133

134

7810

        Ok(res)

135

15620

136

137

138

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

139

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

140

141

// Orchestrator chain types

142

#[allow(deprecated)]

143

pub type ParachainExecutor = NativeElseWasmExecutor<ParachainNativeExecutor>;

144

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

145

pub type ParachainBackend = TFullBackend<Block>;

146

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

147

pub type ParachainBlockImport =

148

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

149

pub type ParachainProposerFactory = ProposerFactory<

150

    TransactionPoolHandle<Block, ParachainClient>,

151

    ParachainClient,

152

    EnableProofRecording,

153

>;

154

155

// Container chains types

156

type ContainerChainExecutor = WasmExecutor<ParachainHostFunctions>;

157

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

158

pub type ContainerChainBackend = TFullBackend<Block>;

159

type ContainerChainBlockImport<RuntimeApi> =

160

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

161

162

tp_traits::alias!(

163

    pub trait MinimalContainerRuntimeApi:

164

        MinimalCumulusRuntimeApi<Block, ContainerChainClient<Self>>

165

        + sp_api::ConstructRuntimeApi<

166

            Block,

167

            ContainerChainClient<Self>,

168

            RuntimeApi:

169

                TanssiSlotDurationApi<Block>

170

                + async_backing_primitives::UnincludedSegmentApi<Block>,

171

172

        + Sized

173

);

174

175

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

176

///

177

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

178

pub fn start_node_impl_container<

179

'a,

180

    RuntimeApi: MinimalContainerRuntimeApi + 'a,

181

    TGenerateRpcBuilder: GenerateRpcBuilder<RuntimeApi> + 'a,

182

>(

183

    parachain_config: Configuration,

184

    relay_chain_interface: Arc<dyn RelayChainInterface>,

185

    orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>,

186

    keystore: KeystorePtr,

187

    para_id: ParaId,

188

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

189

    generate_rpc_builder: TGenerateRpcBuilder,

190

    container_chain_cli: &'a ContainerChainCli,

191

    data_preserver: bool,

192

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

193

    Output = sc_service::error::Result<(

194

        TaskManager,

195

        Arc<ContainerChainClient<RuntimeApi>>,

196

        Arc<ParachainBackend>,

197

)>,

198

> + 'a {

199

    async move {

200

        let parachain_config = prepare_node_config(parachain_config);

201

202

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

203

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

204

205

        let (block_import, import_queue) = container_chain_import_queue(

206

            &parachain_config,

207

            &node_builder,

208

            container_chain_cli,

209

            data_preserver,

210

);

211

        let import_queue_service = import_queue.service();

212

213

        let node_builder = node_builder

214

            .build_cumulus_network::<_, sc_network::NetworkWorker<_, _>>(

215

                &parachain_config,

216

                para_id,

217

                import_queue,

218

                relay_chain_interface.clone(),

219

220

            .await?;

221

222

        let force_authoring = parachain_config.force_authoring;

223

224

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

225

226

        let rpc_builder = generate_rpc_builder.generate(GenerateRpcBuilderParams {

227

            task_manager: &node_builder.task_manager,

228

            container_chain_config: &parachain_config,

229

            client: node_builder.client.clone(),

230

            backend: node_builder.backend.clone(),

231

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

232

            transaction_pool: node_builder.transaction_pool.clone(),

233

            prometheus_registry: node_builder.prometheus_registry.clone(),

234

            command_sink: None,

235

            xcm_senders: None,

236

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

237

        })?;

238

239

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

240

241

        let announce_block = {

242

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

243

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

244

};

245

246

        let relay_chain_slot_duration = Duration::from_secs(6);

247

248

        let overseer_handle = relay_chain_interface

249

            .overseer_handle()

250

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

251

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

252

253

        start_relay_chain_tasks(StartRelayChainTasksParams {

254

            client: node_builder.client.clone(),

255

            announce_block: announce_block.clone(),

256

            para_id,

257

            relay_chain_interface: relay_chain_interface.clone(),

258

            task_manager: &mut node_builder.task_manager,

259

            da_recovery_profile: if collation_params.is_some() {

260

                DARecoveryProfile::Collator

261

            } else {

262

                DARecoveryProfile::FullNode

263

},

264

            import_queue: import_queue_service,

265

            relay_chain_slot_duration,

266

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

267

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

268

        })?;

269

270

        if let Some(collation_params) = collation_params {

271

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

272

            let node_client = node_builder.client.clone();

273

            let node_backend = node_builder.backend.clone();

274

275

            start_consensus_container(

276

                node_client.clone(),

277

                node_backend.clone(),

278

                collation_params,

279

                block_import.clone(),

280

                prometheus_registry.clone(),

281

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

282

                node_spawn_handle.clone(),

283

                relay_chain_interface.clone(),

284

                orchestrator_chain_interface.clone(),

285

                node_builder.transaction_pool.clone(),

286

                node_builder.network.sync_service.clone(),

287

                keystore.clone(),

288

                force_authoring,

289

                relay_chain_slot_duration,

290

                para_id,

291

                overseer_handle.clone(),

292

                announce_block.clone(),

293

                container_chain_cli.base.experimental_max_pov_percentage,

294

);

295

296

297

        node_builder.network.start_network.start_network();

298

        Ok((

299

            node_builder.task_manager,

300

            node_builder.client,

301

            node_builder.backend,

302

))

303

304

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

305

        sc_tracing::logging::PREFIX_LOG_SPAN,

306

        name = container_log_str(para_id),

307

))

308

309

310

pub fn container_chain_import_queue<RuntimeApi: MinimalContainerRuntimeApi>(

311

    parachain_config: &Configuration,

312

    node_builder: &NodeBuilder<ContainerChainNodeConfig<RuntimeApi>>,

313

    container_chain_cli: &ContainerChainCli,

314

    data_preserver: bool,

315

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

316

    // The nimbus import queue ONLY checks the signature correctness

317

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

318

    // in the runtime

319

    let block_import =

320

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

321

322

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

323

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

324

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

325

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

326

        // Default value for download_block_history:

327

        // false if running a collator

328

        // true if running a data preserver node

329

        data_preserver,

330

);

331

332

    let import_queue = nimbus_consensus::import_queue(

333

        node_builder.client.clone(),

334

        block_import.clone(),

335

        move |_, _| async move {

336

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

337

338

            Ok((time,))

339

},

340

        &node_builder.task_manager.spawn_essential_handle(),

341

        parachain_config.prometheus_registry(),

342

        false,

343

        dont_create_gap,

344

345

    .expect("function never fails");

346

347

    (block_import, import_queue)

348

349

350

fn start_consensus_container<RuntimeApi: MinimalContainerRuntimeApi>(

351

    client: Arc<ContainerChainClient<RuntimeApi>>,

352

    backend: Arc<FullBackend>,

353

    collation_params: crate::spawner::CollationParams,

354

    block_import: ContainerChainBlockImport<RuntimeApi>,

355

    prometheus_registry: Option<Registry>,

356

    telemetry: Option<TelemetryHandle>,

357

    spawner: SpawnTaskHandle,

358

    relay_chain_interface: Arc<dyn RelayChainInterface>,

359

    orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>,

360

    transaction_pool: Arc<

361

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

362

>,

363

    sync_oracle: Arc<SyncingService<Block>>,

364

    keystore: KeystorePtr,

365

    force_authoring: bool,

366

    relay_chain_slot_duration: Duration,

367

    para_id: ParaId,

368

    overseer_handle: OverseerHandle,

369

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

370

    max_pov_percentage: Option<u32>,

371

) {

372

    let crate::spawner::CollationParams {

373

        collator_key,

374

        orchestrator_tx_pool,

375

        orchestrator_client,

376

        orchestrator_para_id,

377

        solochain,

378

    } = collation_params;

379

    let slot_duration = if solochain {

380

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

381

        let relay_slot_ms = relay_chain_slot_duration.as_millis();

382

        SlotDuration::from_millis(

383

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

384

385

    } else {

386

        cumulus_client_consensus_aura::slot_duration(

387

            orchestrator_client

388

                .as_deref()

389

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

390

391

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

392

};

393

394

    let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(

395

        spawner.clone(),

396

        client.clone(),

397

        transaction_pool,

398

        prometheus_registry.as_ref(),

399

        telemetry.clone(),

400

);

401

402

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

403

404

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

405

        client.clone(),

406

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

407

        announce_block,

408

        client.clone(),

409

);

410

411

    let relay_chain_interace_for_cidp = relay_chain_interface.clone();

412

    let relay_chain_interace_for_orch = relay_chain_interface.clone();

413

    let orchestrator_client_for_cidp = orchestrator_client.clone();

414

    let client_for_cidp = client.clone();

415

    let client_for_hash_provider = client.clone();

416

    let client_for_slot_duration = client.clone();

417

418

    let code_hash_provider = move |block_hash| {

419

        client_for_hash_provider

420

            .code_at(block_hash)

421

            .ok()

422

            .map(polkadot_primitives::ValidationCode)

423

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

424

};

425

    let buy_core_params = if solochain {

426

        BuyCoreParams::Solochain {}

427

    } else {

428

        BuyCoreParams::Orchestrator {

429

            orchestrator_tx_pool: orchestrator_tx_pool

430

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

431

            orchestrator_client: orchestrator_client

432

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

433

434

};

435

436

    let params = LookaheadTanssiAuraParams {

437

        max_pov_percentage,

438

        get_current_slot_duration: move |block_hash| {

439

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

440

            let slot_duration_ms = client_for_slot_duration

441

                .runtime_api()

442

                .slot_duration(block_hash)

443

                .unwrap_or(12_000);

444

445

            SlotDuration::from_millis(slot_duration_ms)

446

},

447

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

448

            let relay_chain_interface = relay_chain_interace_for_cidp.clone();

449

            let orchestrator_chain_interface = orchestrator_chain_interface.clone();

450

            let client = client_for_cidp.clone();

451

452

            async move {

453

                let authorities_noting_inherent = if solochain {

454

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at_solochain(

455

                        relay_parent,

456

                        &relay_chain_interface,

457

458

                        .await

459

                } else {

460

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at(

461

                        relay_parent,

462

                        &relay_chain_interface,

463

                        &orchestrator_chain_interface,

464

                        orchestrator_para_id,

465

466

                        .await

467

};

468

469

                let slot_duration = {

470

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

471

                    let slot_duration_ms = client

472

                        .runtime_api()

473

                        .slot_duration(block_hash)

474

                        .unwrap_or(12_000);

475

476

                    SlotDuration::from_millis(slot_duration_ms)

477

};

478

479

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

480

481

                let slot =

482

						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(

483

							*timestamp,

484

							slot_duration,

485

);

486

487

                let authorities_noting_inherent = authorities_noting_inherent.ok_or_else(|| {

488

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

489

                        "Failed to create authoritiesnoting inherent",

490

491

                })?;

492

493

                Ok((slot, timestamp, authorities_noting_inherent))

494

495

},

496

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

497

            let relay_chain_interace_for_orch = relay_chain_interace_for_orch.clone();

498

            let orchestrator_client_for_cidp = orchestrator_client_for_cidp.clone();

499

500

            async move {

501

                if solochain {

502

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

503

                        &relay_chain_interace_for_orch,

504

                        "TanssiAuthorityAssignmentApi_para_id_authorities",

505

                        relay_parent,

506

                        &para_id,

507

508

                    .await?;

509

510

                    let authorities = authorities.ok_or_else(|| {

511

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

512

                            "Failed to fetch authorities with error",

513

514

                    })?;

515

516

                    log::info!(

517

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

518

                        authorities,

519

                        relay_parent

520

);

521

522

                    let slot_freq: Option<_> = call_runtime_api(

523

                        &relay_chain_interace_for_orch,

524

                        "OnDemandBlockProductionApi_parathread_slot_frequency",

525

                        relay_parent,

526

                        &para_id,

527

528

                    .await?;

529

530

                    let aux_data = OrchestratorAuraWorkerAuxData {

531

                        authorities,

532

                        slot_freq,

533

};

534

535

                    Ok(aux_data)

536

                } else {

537

                    let latest_header =

538

                        ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::get_latest_orchestrator_head_info(

539

                            relay_parent,

540

                            &relay_chain_interace_for_orch,

541

                            orchestrator_para_id,

542

543

                            .await;

544

545

                    let latest_header = latest_header.ok_or_else(|| {

546

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

547

                            "Failed to fetch latest header",

548

549

                    })?;

550

551

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

552

                        orchestrator_client_for_cidp

553

                            .as_ref()

554

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

555

                        &latest_header.hash(),

556

                        para_id,

557

);

558

559

                    let authorities = authorities.ok_or_else(|| {

560

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

561

                            "Failed to fetch authorities with error",

562

563

                    })?;

564

565

                    log::info!(

566

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

567

                        authorities,

568

                        latest_header

569

);

570

571

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

572

                        orchestrator_client_for_cidp

573

                            .as_ref()

574

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

575

                        &latest_header.hash(),

576

                        para_id,

577

);

578

579

                    let aux_data = OrchestratorAuraWorkerAuxData {

580

                        authorities,

581

                        slot_freq,

582

};

583

584

                    Ok(aux_data)

585

586

587

},

588

        block_import,

589

        para_client: client,

590

        relay_client: relay_chain_interface,

591

        sync_oracle,

592

        keystore,

593

        collator_key,

594

        para_id,

595

        overseer_handle,

596

        orchestrator_slot_duration: slot_duration,

597

        force_authoring,

598

        relay_chain_slot_duration,

599

        proposer,

600

        collator_service,

601

        authoring_duration: Duration::from_millis(2000),

602

        para_backend: backend,

603

        code_hash_provider,

604

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

605

        cancellation_token: CancellationToken::new(),

606

        buy_core_params,

607

};

608

609

    let (fut, _exit_notification_receiver) =

610

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

611

            params,

612

);

613

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

614

615

616

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

617

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

618

// has trouble parsing expressions.

619

fn container_log_str(para_id: ParaId) -> String {

620

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

621