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::{

28

        call_remote_runtime_function, OverseerHandle, RelayChainInterface,

29

},

30

    dancebox_runtime::{

31

        opaque::{Block, Hash},

32

        RuntimeApi,

33

},

34

    dc_orchestrator_chain_interface::OrchestratorChainInterface,

35

    dp_slot_duration_runtime_api::TanssiSlotDurationApi,

36

    nimbus_primitives::{NimbusId, NimbusPair},

37

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

38

    sc_basic_authorship::ProposerFactory,

39

    sc_consensus::{BasicQueue, BlockImport},

40

    sc_executor::WasmExecutor,

41

    sc_network::NetworkBlock,

42

    sc_network_sync::SyncingService,

43

    sc_service::{

44

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

45

},

46

    sc_telemetry::TelemetryHandle,

47

    sc_tracing::tracing::Instrument,

48

    sc_transaction_pool::TransactionPoolHandle,

49

    sp_api::ProvideRuntimeApi,

50

    sp_consensus::EnableProofRecording,

51

    sp_consensus_aura::SlotDuration,

52

    sp_keystore::KeystorePtr,

53

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

54

    substrate_prometheus_endpoint::Registry,

55

    tc_consensus::{

56

        collators::lookahead::{

57

            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,

58

},

59

        OrchestratorAuraWorkerAuxData,

60

},

61

    tokio_util::sync::CancellationToken,

62

};

63

64

#[allow(deprecated)]

65

use sc_executor::NativeElseWasmExecutor;

66

67

type FullBackend = TFullBackend<Block>;

68

69

/// Native executor type.

70

pub struct ParachainNativeExecutor;

71

72

impl sc_executor::NativeExecutionDispatch for ParachainNativeExecutor {

73

    type ExtendHostFunctions = ParachainHostFunctions;

74

75

256056

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

76

256056

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

77

256056

78

79

2760

    fn native_version() -> sc_executor::NativeVersion {

80

2760

        dancebox_runtime::native_version()

81

2760

82

83

84

#[derive(Default, Copy, Clone)]

85

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

86

impl<RuntimeApi> NodeBuilderConfig for ContainerChainNodeConfig<RuntimeApi> {

87

    type Block = Block;

88

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

89

    /// runtime api features.

90

    type RuntimeApi = RuntimeApi;

91

    type ParachainExecutor = ContainerChainExecutor;

92

93

94

impl<RuntimeApi> ContainerChainNodeConfig<RuntimeApi> {

95

    pub fn new() -> Self {

96

        Self(PhantomData)

97

98

99

100

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

101

/// chain selection rule

102

#[derive(Clone)]

103

pub struct OrchestratorParachainBlockImport<BI> {

104

    inner: BI,

105

106

107

impl<BI> OrchestratorParachainBlockImport<BI> {

108

    /// Create a new instance.

109

182

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

110

182

        Self { inner }

111

182

112

113

114

/// We simply rely on the inner

115

#[async_trait::async_trait]

116

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

117

where

118

    BI: BlockImport<Block> + Send + Sync,

119

120

    type Error = BI::Error;

121

122

    async fn check_block(

123

        &self,

124

        block: sc_consensus::BlockCheckParams<Block>,

125

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

126

        self.inner.check_block(block).await

127

128

129

    async fn import_block(

130

        &self,

131

        params: sc_consensus::BlockImportParams<Block>,

132

7274

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

133

7274

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

134

135

7274

        Ok(res)

136

14548

137

138

139

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

140

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

141

142

// Orchestrator chain types

143

#[allow(deprecated)]

144

pub type ParachainExecutor = NativeElseWasmExecutor<ParachainNativeExecutor>;

145

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

146

pub type ParachainBackend = TFullBackend<Block>;

147

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

148

pub type ParachainBlockImport =

149

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

150

pub type ParachainProposerFactory = ProposerFactory<

151

    TransactionPoolHandle<Block, ParachainClient>,

152

    ParachainClient,

153

    EnableProofRecording,

154

>;

155

156

// Container chains types

157

type ContainerChainExecutor = WasmExecutor<ParachainHostFunctions>;

158

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

159

pub type ContainerChainBackend = TFullBackend<Block>;

160

type ContainerChainBlockImport<RuntimeApi> =

161

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

162

163

tp_traits::alias!(

164

    pub trait MinimalContainerRuntimeApi:

165

        MinimalCumulusRuntimeApi<Block, ContainerChainClient<Self>>

166

        + sp_api::ConstructRuntimeApi<

167

            Block,

168

            ContainerChainClient<Self>,

169

            RuntimeApi:

170

                TanssiSlotDurationApi<Block>

171

                + async_backing_primitives::UnincludedSegmentApi<Block>,

172

173

        + Sized

174

);

175

176

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

177

///

178

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

179

pub fn start_node_impl_container<

180

    RuntimeApi: MinimalContainerRuntimeApi,

181

    TGenerateRpcBuilder: GenerateRpcBuilder<RuntimeApi>,

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

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

191

    Output = sc_service::error::Result<(

192

        TaskManager,

193

        Arc<ContainerChainClient<RuntimeApi>>,

194

        Arc<ParachainBackend>,

195

)>,

196

> {

197

    async move {

198

        let parachain_config = prepare_node_config(parachain_config);

199

200

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

201

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

202

203

        let (block_import, import_queue) =

204

            container_chain_import_queue(&parachain_config, &node_builder);

205

        let import_queue_service = import_queue.service();

206

207

        let node_builder = node_builder

208

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

209

                &parachain_config,

210

                para_id,

211

                import_queue,

212

                relay_chain_interface.clone(),

213

214

            .await?;

215

216

        let force_authoring = parachain_config.force_authoring;

217

218

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

219

220

        let rpc_builder = generate_rpc_builder.generate(GenerateRpcBuilderParams {

221

            task_manager: &node_builder.task_manager,

222

            container_chain_config: &parachain_config,

223

            client: node_builder.client.clone(),

224

            backend: node_builder.backend.clone(),

225

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

226

            transaction_pool: node_builder.transaction_pool.clone(),

227

            prometheus_registry: node_builder.prometheus_registry.clone(),

228

            command_sink: None,

229

            xcm_senders: None,

230

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

231

        })?;

232

233

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

234

235

        let announce_block = {

236

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

237

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

238

};

239

240

        let relay_chain_slot_duration = Duration::from_secs(6);

241

242

        let overseer_handle = relay_chain_interface

243

            .overseer_handle()

244

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

245

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

246

247

        start_relay_chain_tasks(StartRelayChainTasksParams {

248

            client: node_builder.client.clone(),

249

            announce_block: announce_block.clone(),

250

            para_id,

251

            relay_chain_interface: relay_chain_interface.clone(),

252

            task_manager: &mut node_builder.task_manager,

253

            da_recovery_profile: if collation_params.is_some() {

254

                DARecoveryProfile::Collator

255

            } else {

256

                DARecoveryProfile::FullNode

257

},

258

            import_queue: import_queue_service,

259

            relay_chain_slot_duration,

260

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

261

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

262

        })?;

263

264

        if let Some(collation_params) = collation_params {

265

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

266

            let node_client = node_builder.client.clone();

267

            let node_backend = node_builder.backend.clone();

268

269

            start_consensus_container(

270

                node_client.clone(),

271

                node_backend.clone(),

272

                collation_params,

273

                block_import.clone(),

274

                prometheus_registry.clone(),

275

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

276

                node_spawn_handle.clone(),

277

                relay_chain_interface.clone(),

278

                orchestrator_chain_interface.clone(),

279

                node_builder.transaction_pool.clone(),

280

                node_builder.network.sync_service.clone(),

281

                keystore.clone(),

282

                force_authoring,

283

                relay_chain_slot_duration,

284

                para_id,

285

                overseer_handle.clone(),

286

                announce_block.clone(),

287

);

288

289

290

        node_builder.network.start_network.start_network();

291

        Ok((

292

            node_builder.task_manager,

293

            node_builder.client,

294

            node_builder.backend,

295

))

296

297

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

298

        sc_tracing::logging::PREFIX_LOG_SPAN,

299

        name = container_log_str(para_id),

300

))

301

302

303

pub fn container_chain_import_queue<RuntimeApi: MinimalContainerRuntimeApi>(

304

    parachain_config: &Configuration,

305

    node_builder: &NodeBuilder<ContainerChainNodeConfig<RuntimeApi>>,

306

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

307

    // The nimbus import queue ONLY checks the signature correctness

308

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

309

    // in the runtime

310

    let block_import =

311

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

312

313

    let import_queue = nimbus_consensus::import_queue(

314

        node_builder.client.clone(),

315

        block_import.clone(),

316

        move |_, _| async move {

317

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

318

319

            Ok((time,))

320

},

321

        &node_builder.task_manager.spawn_essential_handle(),

322

        parachain_config.prometheus_registry(),

323

        false,

324

325

    .expect("function never fails");

326

327

    (block_import, import_queue)

328

329

330

fn start_consensus_container<RuntimeApi: MinimalContainerRuntimeApi>(

331

    client: Arc<ContainerChainClient<RuntimeApi>>,

332

    backend: Arc<FullBackend>,

333

    collation_params: crate::spawner::CollationParams,

334

    block_import: ContainerChainBlockImport<RuntimeApi>,

335

    prometheus_registry: Option<Registry>,

336

    telemetry: Option<TelemetryHandle>,

337

    spawner: SpawnTaskHandle,

338

    relay_chain_interface: Arc<dyn RelayChainInterface>,

339

    orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>,

340

    transaction_pool: Arc<

341

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

342

>,

343

    sync_oracle: Arc<SyncingService<Block>>,

344

    keystore: KeystorePtr,

345

    force_authoring: bool,

346

    relay_chain_slot_duration: Duration,

347

    para_id: ParaId,

348

    overseer_handle: OverseerHandle,

349

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

350

) {

351

    let crate::spawner::CollationParams {

352

        collator_key,

353

        orchestrator_tx_pool,

354

        orchestrator_client,

355

        orchestrator_para_id,

356

        solochain,

357

    } = collation_params;

358

    let slot_duration = if solochain {

359

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

360

        let relay_slot_ms = relay_chain_slot_duration.as_millis();

361

        SlotDuration::from_millis(

362

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

363

364

    } else {

365

        cumulus_client_consensus_aura::slot_duration(

366

            orchestrator_client

367

                .as_deref()

368

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

369

370

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

371

};

372

373

    let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(

374

        spawner.clone(),

375

        client.clone(),

376

        transaction_pool,

377

        prometheus_registry.as_ref(),

378

        telemetry.clone(),

379

);

380

381

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

382

383

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

384

        client.clone(),

385

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

386

        announce_block,

387

        client.clone(),

388

);

389

390

    let relay_chain_interace_for_cidp = relay_chain_interface.clone();

391

    let relay_chain_interace_for_orch = relay_chain_interface.clone();

392

    let orchestrator_client_for_cidp = orchestrator_client.clone();

393

    let client_for_cidp = client.clone();

394

    let client_for_hash_provider = client.clone();

395

    let client_for_slot_duration = client.clone();

396

397

    let code_hash_provider = move |block_hash| {

398

        client_for_hash_provider

399

            .code_at(block_hash)

400

            .ok()

401

            .map(polkadot_primitives::ValidationCode)

402

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

403

};

404

    let buy_core_params = if solochain {

405

        BuyCoreParams::Solochain {}

406

    } else {

407

        BuyCoreParams::Orchestrator {

408

            orchestrator_tx_pool: orchestrator_tx_pool

409

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

410

            orchestrator_client: orchestrator_client

411

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

412

413

};

414

415

    let params = LookaheadTanssiAuraParams {

416

        get_current_slot_duration: move |block_hash| {

417

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

418

            let slot_duration_ms = client_for_slot_duration

419

                .runtime_api()

420

                .slot_duration(block_hash)

421

                .unwrap_or(12_000);

422

423

            SlotDuration::from_millis(slot_duration_ms)

424

},

425

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

426

            let relay_chain_interface = relay_chain_interace_for_cidp.clone();

427

            let orchestrator_chain_interface = orchestrator_chain_interface.clone();

428

            let client = client_for_cidp.clone();

429

430

            async move {

431

                let authorities_noting_inherent = if solochain {

432

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at_solochain(

433

                        relay_parent,

434

                        &relay_chain_interface,

435

436

                        .await

437

                } else {

438

                    ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::create_at(

439

                        relay_parent,

440

                        &relay_chain_interface,

441

                        &orchestrator_chain_interface,

442

                        orchestrator_para_id,

443

444

                        .await

445

};

446

447

                let slot_duration = {

448

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

449

                    let slot_duration_ms = client

450

                        .runtime_api()

451

                        .slot_duration(block_hash)

452

                        .unwrap_or(12_000);

453

454

                    SlotDuration::from_millis(slot_duration_ms)

455

};

456

457

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

458

459

                let slot =

460

						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(

461

							*timestamp,

462

							slot_duration,

463

);

464

465

                let authorities_noting_inherent = authorities_noting_inherent.ok_or_else(|| {

466

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

467

                        "Failed to create authoritiesnoting inherent",

468

469

                })?;

470

471

                Ok((slot, timestamp, authorities_noting_inherent))

472

473

},

474

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

475

            let relay_chain_interace_for_orch = relay_chain_interace_for_orch.clone();

476

            let orchestrator_client_for_cidp = orchestrator_client_for_cidp.clone();

477

478

            async move {

479

                if solochain {

480

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

481

                        &relay_chain_interace_for_orch,

482

                        "TanssiAuthorityAssignmentApi_para_id_authorities",

483

                        relay_parent,

484

                        &para_id,

485

486

                    .await?;

487

488

                    let authorities = authorities.ok_or_else(|| {

489

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

490

                            "Failed to fetch authorities with error",

491

492

                    })?;

493

494

                    log::info!(

495

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

496

                        authorities,

497

                        relay_parent

498

);

499

500

                    let slot_freq: Option<_> = call_remote_runtime_function(

501

                        &relay_chain_interace_for_orch,

502

                        "OnDemandBlockProductionApi_parathread_slot_frequency",

503

                        relay_parent,

504

                        &para_id,

505

506

                    .await?;

507

508

                    let aux_data = OrchestratorAuraWorkerAuxData {

509

                        authorities,

510

                        slot_freq,

511

};

512

513

                    Ok(aux_data)

514

                } else {

515

                    let latest_header =

516

                        ccp_authorities_noting_inherent::ContainerChainAuthoritiesInherentData::get_latest_orchestrator_head_info(

517

                            relay_parent,

518

                            &relay_chain_interace_for_orch,

519

                            orchestrator_para_id,

520

521

                            .await;

522

523

                    let latest_header = latest_header.ok_or_else(|| {

524

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

525

                            "Failed to fetch latest header",

526

527

                    })?;

528

529

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

530

                        orchestrator_client_for_cidp

531

                            .as_ref()

532

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

533

                        &latest_header.hash(),

534

                        para_id,

535

);

536

537

                    let authorities = authorities.ok_or_else(|| {

538

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

539

                            "Failed to fetch authorities with error",

540

541

                    })?;

542

543

                    log::info!(

544

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

545

                        authorities,

546

                        latest_header

547

);

548

549

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

550

                        orchestrator_client_for_cidp

551

                            .as_ref()

552

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

553

                        &latest_header.hash(),

554

                        para_id,

555

);

556

557

                    let aux_data = OrchestratorAuraWorkerAuxData {

558

                        authorities,

559

                        slot_freq,

560

};

561

562

                    Ok(aux_data)

563

564

565

},

566

        block_import,

567

        para_client: client,

568

        relay_client: relay_chain_interface,

569

        sync_oracle,

570

        keystore,

571

        collator_key,

572

        para_id,

573

        overseer_handle,

574

        orchestrator_slot_duration: slot_duration,

575

        force_authoring,

576

        relay_chain_slot_duration,

577

        proposer,

578

        collator_service,

579

        authoring_duration: Duration::from_millis(2000),

580

        para_backend: backend,

581

        code_hash_provider,

582

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

583

        cancellation_token: CancellationToken::new(),

584

        buy_core_params,

585

};

586

587

    let (fut, _exit_notification_receiver) =

588

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

589

            params,

590

);

591

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

592

593

594

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

595

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

596

// has trouble parsing expressions.

597

fn container_log_str(para_id: ParaId) -> String {

598

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

599