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
//! Service and ServiceFactory implementation. Specialized wrapper over substrate service.
18

            
19
use {
20
    crate::command::solochain::{
21
        build_solochain_config_dir, copy_zombienet_keystore, dummy_config, keystore_config,
22
    },
23
    core::marker::PhantomData,
24
    cumulus_client_cli::CollatorOptions,
25
    cumulus_client_collator::service::CollatorService,
26
    cumulus_client_consensus_proposer::Proposer,
27
    cumulus_client_parachain_inherent::{MockValidationDataInherentDataProvider, MockXcmConfig},
28
    cumulus_client_service::{
29
        prepare_node_config, start_relay_chain_tasks, DARecoveryProfile, StartRelayChainTasksParams,
30
    },
31
    cumulus_primitives_core::{
32
        relay_chain::{well_known_keys as RelayWellKnownKeys, CollatorPair},
33
        CollectCollationInfo, ParaId,
34
    },
35
    cumulus_relay_chain_interface::{call_runtime_api, OverseerHandle, RelayChainInterface},
36
    dancebox_runtime::{
37
        opaque::{Block, Hash},
38
        AccountId, RuntimeApi,
39
    },
40
    dc_orchestrator_chain_interface::{
41
        BlockNumber, ContainerChainGenesisData, DataPreserverAssignment, DataPreserverProfileId,
42
        OrchestratorChainError, OrchestratorChainInterface, OrchestratorChainResult, PHash,
43
        PHeader,
44
    },
45
    frame_support::__private::sp_tracing::tracing::Instrument,
46
    futures::{Stream, StreamExt},
47
    nimbus_primitives::{NimbusId, NimbusPair},
48
    node_common::service::{ManualSealConfiguration, NodeBuilder, NodeBuilderConfig, Sealing},
49
    pallet_author_noting_runtime_api::AuthorNotingApi,
50
    pallet_collator_assignment_runtime_api::CollatorAssignmentApi,
51
    pallet_data_preservers_runtime_api::DataPreserversApi,
52
    pallet_registrar_runtime_api::RegistrarApi,
53
    parity_scale_codec::{Decode, Encode},
54
    polkadot_cli::ProvideRuntimeApi,
55
    polkadot_parachain_primitives::primitives::HeadData,
56
    polkadot_primitives::UpgradeGoAhead,
57
    polkadot_service::Handle,
58
    sc_cli::CliConfiguration,
59
    sc_client_api::{
60
        AuxStore, Backend as BackendT, BlockchainEvents, HeaderBackend, UsageProvider,
61
    },
62
    sc_consensus::BasicQueue,
63
    sc_network::NetworkBlock,
64
    sc_network_common::role::Role,
65
    sc_network_sync::SyncingService,
66
    sc_service::{Configuration, KeystoreContainer, SpawnTaskHandle, TFullBackend, TaskManager},
67
    sc_telemetry::TelemetryHandle,
68
    sc_transaction_pool::TransactionPoolHandle,
69
    sp_api::ApiExt,
70
    sp_api::StorageProof,
71
    sp_consensus::SyncOracle,
72
    sp_consensus_slots::Slot,
73
    sp_core::{traits::SpawnEssentialNamed, H256},
74
    sp_keystore::KeystorePtr,
75
    sp_state_machine::{Backend as StateBackend, StorageValue},
76
    std::{pin::Pin, sync::Arc, time::Duration},
77
    tc_consensus::{
78
        collators::lookahead::{
79
            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,
80
        },
81
        OnDemandBlockProductionApi, OrchestratorAuraWorkerAuxData, TanssiAuthorityAssignmentApi,
82
    },
83
    tc_service_container_chain::{
84
        cli::ContainerChainCli,
85
        monitor,
86
        service::{
87
            DevParachainBlockImport, ParachainBlockImport, ParachainClient, ParachainExecutor,
88
            ParachainProposerFactory,
89
        },
90
        spawner::{self, CcSpawnMsg, ContainerChainSpawnParams, ContainerChainSpawner},
91
    },
92
    tokio::sync::mpsc::{unbounded_channel, UnboundedSender},
93
    tokio_util::sync::CancellationToken,
94
};
95

            
96
mod mocked_relay_keys;
97

            
98
// We use this to detect whether randomness is activated
99
const RANDOMNESS_ACTIVATED_AUX_KEY: &[u8] = b"__DEV_RANDOMNESS_ACTIVATED";
100

            
101
type FullBackend = TFullBackend<Block>;
102

            
103
pub struct NodeConfig;
104
impl NodeBuilderConfig for NodeConfig {
105
    type Block = Block;
106
    type RuntimeApi = RuntimeApi;
107
    type ParachainExecutor = ParachainExecutor;
108
}
109

            
110
thread_local!(static TIMESTAMP: std::cell::RefCell<u64> = const { std::cell::RefCell::new(0) });
111

            
112
/// Provide a mock duration starting at 0 in millisecond for timestamp inherent.
113
/// Each call will increment timestamp by slot_duration making Aura think time has passed.
114
struct MockTimestampInherentDataProvider;
115
#[async_trait::async_trait]
116
impl sp_inherents::InherentDataProvider for MockTimestampInherentDataProvider {
117
    async fn provide_inherent_data(
118
        &self,
119
        inherent_data: &mut sp_inherents::InherentData,
120
7274
    ) -> Result<(), sp_inherents::Error> {
121
7274
        TIMESTAMP.with(|x| {
122
7274
            *x.borrow_mut() += dancebox_runtime::SLOT_DURATION;
123
7274
            inherent_data.put_data(sp_timestamp::INHERENT_IDENTIFIER, &*x.borrow())
124
7274
        })
125
14548
    }
126

            
127
    async fn try_handle_error(
128
        &self,
129
        _identifier: &sp_inherents::InherentIdentifier,
130
        _error: &[u8],
131
    ) -> Option<Result<(), sp_inherents::Error>> {
132
        // The pallet never reports error.
133
        None
134
    }
135
}
136

            
137
/// Background task used to detect changes to container chain assignment,
138
/// and start/stop container chains on demand. The check runs on every new block.
139
pub fn build_check_assigned_para_id(
140
    client: Arc<dyn OrchestratorChainInterface>,
141
    sync_keystore: KeystorePtr,
142
    cc_spawn_tx: UnboundedSender<CcSpawnMsg>,
143
    spawner: impl SpawnEssentialNamed,
144
) {
145
    let check_assigned_para_id_task = async move {
146
        // Subscribe to new blocks in order to react to para id assignment
147
        // This must be the stream of finalized blocks, otherwise the collators may rotate to a
148
        // different chain before the block is finalized, and that could lead to a stalled chain
149
        let mut import_notifications = client.finality_notification_stream().await.unwrap();
150

            
151
        while let Some(msg) = import_notifications.next().await {
152
            let block_hash = msg.hash();
153
            let client_set_aside_for_cidp = client.clone();
154
            let sync_keystore = sync_keystore.clone();
155
            let cc_spawn_tx = cc_spawn_tx.clone();
156

            
157
            check_assigned_para_id(
158
                cc_spawn_tx,
159
                sync_keystore,
160
                client_set_aside_for_cidp,
161
                block_hash,
162
            )
163
            .await
164
            .unwrap();
165
        }
166
    };
167

            
168
    spawner.spawn_essential(
169
        "check-assigned-para-id",
170
        None,
171
        Box::pin(check_assigned_para_id_task),
172
    );
173
}
174

            
175
/// Check the parachain assignment using the orchestrator chain client, and send a `CcSpawnMsg` to
176
/// start or stop the required container chains.
177
///
178
/// Checks the assignment for the next block, so if there is a session change on block 15, this will
179
/// detect the assignment change after importing block 14.
180
async fn check_assigned_para_id(
181
    cc_spawn_tx: UnboundedSender<CcSpawnMsg>,
182
    sync_keystore: KeystorePtr,
183
    client_set_aside_for_cidp: Arc<dyn OrchestratorChainInterface>,
184
    block_hash: H256,
185
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
186
    // Check current assignment
187
    let current_container_chain_para_id =
188
        tc_consensus::first_eligible_key::<dyn OrchestratorChainInterface, NimbusPair>(
189
            client_set_aside_for_cidp.as_ref(),
190
            &block_hash,
191
            sync_keystore.clone(),
192
        )
193
        .await
194
        .map(|(_nimbus_key, para_id)| para_id);
195

            
196
    // Check assignment in the next session
197
    let next_container_chain_para_id = tc_consensus::first_eligible_key_next_session::<
198
        dyn OrchestratorChainInterface,
199
        NimbusPair,
200
    >(
201
        client_set_aside_for_cidp.as_ref(),
202
        &block_hash,
203
        sync_keystore,
204
    )
205
    .await
206
    .map(|(_nimbus_key, para_id)| para_id);
207

            
208
    cc_spawn_tx.send(CcSpawnMsg::UpdateAssignment {
209
        current: current_container_chain_para_id,
210
        next: next_container_chain_para_id,
211
    })?;
212

            
213
    Ok(())
214
}
215

            
216
pub fn import_queue(
217
    parachain_config: &Configuration,
218
    node_builder: &NodeBuilder<NodeConfig>,
219
) -> (ParachainBlockImport, BasicQueue<Block>) {
220
    // The nimbus import queue ONLY checks the signature correctness
221
    // Any other checks corresponding to the author-correctness should be done
222
    // in the runtime
223
    let block_import =
224
        ParachainBlockImport::new(node_builder.client.clone(), node_builder.backend.clone());
225

            
226
    let import_queue = nimbus_consensus::import_queue(
227
        node_builder.client.clone(),
228
        block_import.clone(),
229
        move |_, _| async move {
230
            let time = sp_timestamp::InherentDataProvider::from_system_time();
231

            
232
            Ok((time,))
233
        },
234
        &node_builder.task_manager.spawn_essential_handle(),
235
        parachain_config.prometheus_registry(),
236
        false,
237
    )
238
    .expect("function never fails");
239

            
240
    (block_import, import_queue)
241
}
242

            
243
/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.
244
///
245
/// This is the actual implementation that is abstract over the executor and the runtime api.
246
async fn start_node_impl(
247
    orchestrator_config: Configuration,
248
    polkadot_config: Configuration,
249
    container_chain_config: Option<(ContainerChainCli, tokio::runtime::Handle)>,
250
    collator_options: CollatorOptions,
251
    para_id: ParaId,
252
    hwbench: Option<sc_sysinfo::HwBench>,
253
) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient>)> {
254
    let parachain_config = prepare_node_config(orchestrator_config);
255
    let chain_type: sc_chain_spec::ChainType = parachain_config.chain_spec.chain_type();
256
    let relay_chain = crate::chain_spec::Extensions::try_get(&*parachain_config.chain_spec)
257
        .map(|e| e.relay_chain.clone())
258
        .ok_or("Could not find relay_chain extension in chain-spec.")?;
259

            
260
    // Channel to send messages to start/stop container chains
261
    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();
262

            
263
    // Create a `NodeBuilder` which helps setup parachain nodes common systems.
264
    let mut node_builder = NodeConfig::new_builder(&parachain_config, hwbench.clone())?;
265

            
266
    let (block_import, import_queue) = import_queue(&parachain_config, &node_builder);
267

            
268
    let (relay_chain_interface, collator_key) = node_builder
269
        .build_relay_chain_interface(&parachain_config, polkadot_config, collator_options.clone())
270
        .await?;
271

            
272
    let validator = parachain_config.role.is_authority();
273
    let force_authoring = parachain_config.force_authoring;
274

            
275
    let node_builder = node_builder
276
        .build_cumulus_network::<_, sc_network::NetworkWorker<_, _>>(
277
            &parachain_config,
278
            para_id,
279
            import_queue,
280
            relay_chain_interface.clone(),
281
        )
282
        .await?;
283

            
284
    let rpc_builder = {
285
        let client = node_builder.client.clone();
286
        let transaction_pool = node_builder.transaction_pool.clone();
287

            
288
        Box::new(move |_| {
289
            let deps = crate::rpc::FullDeps {
290
                client: client.clone(),
291
                pool: transaction_pool.clone(),
292
                command_sink: None,
293
                xcm_senders: None,
294
                randomness_sender: None,
295
            };
296

            
297
            crate::rpc::create_full(deps).map_err(Into::into)
298
        })
299
    };
300

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

            
303
    let relay_chain_slot_duration = Duration::from_secs(6);
304
    let overseer_handle = relay_chain_interface
305
        .overseer_handle()
306
        .map_err(|e| sc_service::Error::Application(Box::new(e)))?;
307
    let sync_keystore = node_builder.keystore_container.keystore();
308
    let mut collate_on_tanssi: Arc<
309
        dyn Fn() -> (CancellationToken, futures::channel::oneshot::Receiver<()>) + Send + Sync,
310
    > = Arc::new(move || {
311
        if validator {
312
            panic!("Called uninitialized collate_on_tanssi");
313
        } else {
314
            panic!("Called collate_on_tanssi when node is not running as a validator");
315
        }
316
    });
317

            
318
    let announce_block = {
319
        let sync_service = node_builder.network.sync_service.clone();
320
        Arc::new(move |hash, data| sync_service.announce_block(hash, data))
321
    };
322

            
323
    let (mut node_builder, import_queue_service) = node_builder.extract_import_queue_service();
324

            
325
    start_relay_chain_tasks(StartRelayChainTasksParams {
326
        client: node_builder.client.clone(),
327
        announce_block: announce_block.clone(),
328
        para_id,
329
        relay_chain_interface: relay_chain_interface.clone(),
330
        task_manager: &mut node_builder.task_manager,
331
        da_recovery_profile: if validator {
332
            DARecoveryProfile::Collator
333
        } else {
334
            DARecoveryProfile::FullNode
335
        },
336
        import_queue: import_queue_service,
337
        relay_chain_slot_duration,
338
        recovery_handle: Box::new(overseer_handle.clone()),
339
        sync_service: node_builder.network.sync_service.clone(),
340
    })?;
341

            
342
    let orchestrator_chain_interface_builder = OrchestratorChainInProcessInterfaceBuilder {
343
        client: node_builder.client.clone(),
344
        backend: node_builder.backend.clone(),
345
        sync_oracle: node_builder.network.sync_service.clone(),
346
        overseer_handle: overseer_handle.clone(),
347
    };
348
    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();
349

            
350
    if validator {
351
        let collator_key = collator_key
352
            .clone()
353
            .expect("Command line arguments do not allow this. qed");
354

            
355
        // Start task which detects para id assignment, and starts/stops container chains.
356
        // Note that if this node was started without a `container_chain_config`, we don't
357
        // support collation on container chains, so there is no need to detect changes to assignment
358
        if container_chain_config.is_some() {
359
            build_check_assigned_para_id(
360
                orchestrator_chain_interface.clone(),
361
                sync_keystore.clone(),
362
                cc_spawn_tx.clone(),
363
                node_builder.task_manager.spawn_essential_handle(),
364
            );
365
        }
366

            
367
        let start_collation = {
368
            // Params for collate_on_tanssi closure
369
            let node_spawn_handle = node_builder.task_manager.spawn_handle().clone();
370
            let node_keystore = node_builder.keystore_container.keystore().clone();
371
            let node_telemetry_handle = node_builder.telemetry.as_ref().map(|t| t.handle()).clone();
372
            let node_client = node_builder.client.clone();
373
            let node_backend = node_builder.backend.clone();
374
            let relay_interface = relay_chain_interface.clone();
375
            let node_sync_service = node_builder.network.sync_service.clone();
376
            let orchestrator_tx_pool = node_builder.transaction_pool.clone();
377
            let overseer = overseer_handle.clone();
378
            let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(
379
                node_spawn_handle.clone(),
380
                node_client.clone(),
381
                node_builder.transaction_pool.clone(),
382
                node_builder.prometheus_registry.as_ref(),
383
                node_telemetry_handle.clone(),
384
            );
385

            
386
            move || {
387
                start_consensus_orchestrator(
388
                    node_client.clone(),
389
                    node_backend.clone(),
390
                    block_import.clone(),
391
                    node_spawn_handle.clone(),
392
                    relay_interface.clone(),
393
                    node_sync_service.clone(),
394
                    node_keystore.clone(),
395
                    force_authoring,
396
                    relay_chain_slot_duration,
397
                    para_id,
398
                    collator_key.clone(),
399
                    overseer.clone(),
400
                    announce_block.clone(),
401
                    proposer_factory.clone(),
402
                    orchestrator_tx_pool.clone(),
403
                )
404
            }
405
        };
406
        // Save callback for later, used when collator rotates from container chain back to orchestrator chain
407
        collate_on_tanssi = Arc::new(start_collation);
408
    }
409

            
410
    node_builder.network.start_network.start_network();
411

            
412
    let sync_keystore = node_builder.keystore_container.keystore();
413

            
414
    if let Some((container_chain_cli, tokio_handle)) = container_chain_config {
415
        // If the orchestrator chain is running as a full-node, we start a full node for the
416
        // container chain immediately, because only collator nodes detect their container chain
417
        // assignment so otherwise it will never start.
418
        if !validator {
419
            if let Some(container_chain_para_id) = container_chain_cli.base.para_id {
420
                // Spawn new container chain node
421
                cc_spawn_tx
422
                    .send(CcSpawnMsg::UpdateAssignment {
423
                        current: Some(container_chain_para_id.into()),
424
                        next: Some(container_chain_para_id.into()),
425
                    })
426
                    .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
427
            }
428
        }
429

            
430
        // Start container chain spawner task. This will start and stop container chains on demand.
431
        let orchestrator_client = node_builder.client.clone();
432
        let orchestrator_tx_pool = node_builder.transaction_pool.clone();
433
        let spawn_handle = node_builder.task_manager.spawn_handle();
434

            
435
        // This considers that the container chains have the same APIs as dancebox, which
436
        // is not the case. However the spawner don't call APIs that are not part of the expected
437
        // common APIs for a container chain.
438
        // TODO: Depend on the simple container chain runtime which should be the minimal api?
439
        let container_chain_spawner = ContainerChainSpawner {
440
            params: ContainerChainSpawnParams {
441
                orchestrator_chain_interface,
442
                container_chain_cli,
443
                tokio_handle,
444
                chain_type,
445
                relay_chain,
446
                relay_chain_interface,
447
                sync_keystore,
448
                orchestrator_para_id: para_id,
449
                data_preserver: false,
450
                collation_params: if validator {
451
                    Some(spawner::CollationParams {
452
                        orchestrator_client: Some(orchestrator_client.clone()),
453
                        orchestrator_tx_pool: Some(orchestrator_tx_pool),
454
                        orchestrator_para_id: para_id,
455
                        collator_key: collator_key
456
                            .expect("there should be a collator key if we're a validator"),
457
                        solochain: false,
458
                    })
459
                } else {
460
                    None
461
                },
462
                spawn_handle,
463
                generate_rpc_builder: tc_service_container_chain::rpc::GenerateSubstrateRpcBuilder::<
464
                    dancebox_runtime::RuntimeApi,
465
                >::new(),
466
                phantom: PhantomData,
467
            },
468
            state: Default::default(),
469
            db_folder_cleanup_done: false,
470
            collate_on_tanssi,
471
            collation_cancellation_constructs: None,
472
        };
473
        let state = container_chain_spawner.state.clone();
474

            
475
        node_builder.task_manager.spawn_essential_handle().spawn(
476
            "container-chain-spawner-rx-loop",
477
            None,
478
            container_chain_spawner.rx_loop(cc_spawn_rx, validator, false),
479
        );
480

            
481
        node_builder.task_manager.spawn_essential_handle().spawn(
482
            "container-chain-spawner-debug-state",
483
            None,
484
            monitor::monitor_task(state),
485
        )
486
    }
487

            
488
    Ok((node_builder.task_manager, node_builder.client))
489
}
490

            
491
/// Build the import queue for the parachain runtime (manual seal).
492
182
fn build_manual_seal_import_queue(
493
182
    _client: Arc<ParachainClient>,
494
182
    block_import: DevParachainBlockImport,
495
182
    config: &Configuration,
496
182
    _telemetry: Option<TelemetryHandle>,
497
182
    task_manager: &TaskManager,
498
182
) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error> {
499
182
    Ok(sc_consensus_manual_seal::import_queue(
500
182
        Box::new(block_import),
501
182
        &task_manager.spawn_essential_handle(),
502
182
        config.prometheus_registry(),
503
182
    ))
504
182
}
505

            
506
/// Start collator task for orchestrator chain.
507
/// Returns a `CancellationToken` that can be used to cancel the collator task,
508
/// and a `oneshot::Receiver<()>` that can be used to wait until the task has ended.
509
fn start_consensus_orchestrator(
510
    client: Arc<ParachainClient>,
511
    backend: Arc<FullBackend>,
512
    block_import: ParachainBlockImport,
513
    spawner: SpawnTaskHandle,
514
    relay_chain_interface: Arc<dyn RelayChainInterface>,
515
    sync_oracle: Arc<SyncingService<Block>>,
516
    keystore: KeystorePtr,
517
    force_authoring: bool,
518
    relay_chain_slot_duration: Duration,
519
    para_id: ParaId,
520
    collator_key: CollatorPair,
521
    overseer_handle: OverseerHandle,
522
    announce_block: Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
523
    proposer_factory: ParachainProposerFactory,
524
    orchestrator_tx_pool: Arc<TransactionPoolHandle<Block, ParachainClient>>,
525
) -> (CancellationToken, futures::channel::oneshot::Receiver<()>) {
526
    let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)
527
        .expect("start_consensus_orchestrator: slot duration should exist");
528

            
529
    let proposer = Proposer::new(proposer_factory);
530

            
531
    let collator_service = CollatorService::new(
532
        client.clone(),
533
        Arc::new(spawner.clone()),
534
        announce_block,
535
        client.clone(),
536
    );
537

            
538
    let relay_chain_interace_for_cidp = relay_chain_interface.clone();
539
    let client_set_aside_for_cidp = client.clone();
540
    let client_set_aside_for_orch = client.clone();
541
    let client_for_hash_provider = client.clone();
542
    let client_for_slot_duration_provider = client.clone();
543

            
544
    let code_hash_provider = move |block_hash| {
545
        client_for_hash_provider
546
            .code_at(block_hash)
547
            .ok()
548
            .map(polkadot_primitives::ValidationCode)
549
            .map(|c| c.hash())
550
    };
551

            
552
    let cancellation_token = CancellationToken::new();
553
    let buy_core_params = BuyCoreParams::Orchestrator {
554
        orchestrator_tx_pool,
555
        orchestrator_client: client.clone(),
556
    };
557

            
558
    let params = LookaheadTanssiAuraParams {
559
        get_current_slot_duration: move |block_hash| {
560
            sc_consensus_aura::standalone::slot_duration_at(
561
                &*client_for_slot_duration_provider,
562
                block_hash,
563
            )
564
            .expect("Slot duration should be set")
565
        },
566
        create_inherent_data_providers: move |block_hash, (relay_parent, _validation_data)| {
567
            let relay_chain_interface = relay_chain_interace_for_cidp.clone();
568
            let client_set_aside_for_cidp = client_set_aside_for_cidp.clone();
569
            async move {
570
                // We added a new runtime api that allows to know which parachains have
571
                // some collators assigned to them. We'll now only include those. For older
572
                // runtimes we continue to write all of them.
573
                let para_ids = match client_set_aside_for_cidp
574
                    .runtime_api()
575
                    .api_version::<dyn CollatorAssignmentApi<Block, AccountId, ParaId>>(
576
                    block_hash,
577
                )? {
578
                    Some(version) if version >= 2 => client_set_aside_for_cidp
579
                        .runtime_api()
580
                        .parachains_with_some_collators(block_hash)?,
581
                    _ => client_set_aside_for_cidp
582
                        .runtime_api()
583
                        .registered_paras(block_hash)?,
584
                };
585
                let para_ids: Vec<_> = para_ids.into_iter().collect();
586
                let author_noting_inherent =
587
                    tp_author_noting_inherent::OwnParachainInherentData::create_at(
588
                        relay_parent,
589
                        &relay_chain_interface,
590
                        &para_ids,
591
                    )
592
                    .await;
593

            
594
                // Fetch duration every block to avoid downtime when passing from 12 to 6s
595
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
596
                    &*client_set_aside_for_cidp.clone(),
597
                    block_hash,
598
                )
599
                .expect("Slot duration should be set");
600

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

            
603
                let slot =
604
						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
605
							*timestamp,
606
							slot_duration,
607
						);
608

            
609
                let author_noting_inherent = author_noting_inherent.ok_or_else(|| {
610
                    Box::<dyn std::error::Error + Send + Sync>::from(
611
                        "Failed to create author noting inherent",
612
                    )
613
                })?;
614

            
615
                Ok((slot, timestamp, author_noting_inherent))
616
            }
617
        },
618
        get_orchestrator_aux_data: move |block_hash: H256, (_relay_parent, _validation_data)| {
619
            let client_set_aside_for_orch = client_set_aside_for_orch.clone();
620

            
621
            async move {
622
                let authorities = tc_consensus::authorities::<Block, ParachainClient, NimbusPair>(
623
                    client_set_aside_for_orch.as_ref(),
624
                    &block_hash,
625
                    para_id,
626
                );
627

            
628
                let authorities = authorities.ok_or_else(|| {
629
                    Box::<dyn std::error::Error + Send + Sync>::from(
630
                        "Failed to fetch authorities with error",
631
                    )
632
                })?;
633

            
634
                log::info!(
635
                    "Authorities {:?} found for header {:?}",
636
                    authorities,
637
                    block_hash
638
                );
639

            
640
                let aux_data = OrchestratorAuraWorkerAuxData {
641
                    authorities,
642
                    // This is the orchestrator consensus, it does not have a slot frequency
643
                    slot_freq: None,
644
                };
645

            
646
                Ok(aux_data)
647
            }
648
        },
649
        block_import,
650
        para_client: client,
651
        relay_client: relay_chain_interface,
652
        sync_oracle,
653
        keystore,
654
        collator_key,
655
        para_id,
656
        overseer_handle,
657
        orchestrator_slot_duration: slot_duration,
658
        relay_chain_slot_duration,
659
        force_authoring,
660
        proposer,
661
        collator_service,
662
        authoring_duration: Duration::from_millis(2000),
663
        code_hash_provider,
664
        para_backend: backend,
665
        cancellation_token: cancellation_token.clone(),
666
        buy_core_params,
667
    };
668

            
669
    let (fut, exit_notification_receiver) =
670
        lookahead_tanssi_aura::run::<_, Block, NimbusPair, _, _, _, _, _, _, _, _, _, _, _, _, _>(
671
            params,
672
        );
673
    spawner.spawn("tanssi-aura", None, fut);
674

            
675
    (cancellation_token, exit_notification_receiver)
676
}
677

            
678
/// Start a parachain node.
679
pub async fn start_parachain_node(
680
    parachain_config: Configuration,
681
    polkadot_config: Configuration,
682
    container_config: Option<(ContainerChainCli, tokio::runtime::Handle)>,
683
    collator_options: CollatorOptions,
684
    para_id: ParaId,
685
    hwbench: Option<sc_sysinfo::HwBench>,
686
) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient>)> {
687
    start_node_impl(
688
        parachain_config,
689
        polkadot_config,
690
        container_config,
691
        collator_options,
692
        para_id,
693
        hwbench,
694
    )
695
    .instrument(sc_tracing::tracing::info_span!(
696
        sc_tracing::logging::PREFIX_LOG_SPAN,
697
        name = "Orchestrator",
698
    ))
699
    .await
700
}
701

            
702
/// Start a solochain node.
703
pub async fn start_solochain_node(
704
    polkadot_config: Configuration,
705
    container_chain_cli: ContainerChainCli,
706
    collator_options: CollatorOptions,
707
    hwbench: Option<sc_sysinfo::HwBench>,
708
) -> sc_service::error::Result<TaskManager> {
709
    let tokio_handle = polkadot_config.tokio_handle.clone();
710
    let orchestrator_para_id = Default::default();
711

            
712
    let chain_type = polkadot_config.chain_spec.chain_type().clone();
713
    let relay_chain = polkadot_config.chain_spec.id().to_string();
714

            
715
    let base_path = container_chain_cli
716
        .base
717
        .base
718
        .shared_params
719
        .base_path
720
        .as_ref()
721
        .expect("base_path is always set");
722
    let config_dir = build_solochain_config_dir(base_path);
723
    let keystore = keystore_config(container_chain_cli.keystore_params(), &config_dir)
724
        .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
725

            
726
    // Instead of putting keystore in
727
    // Collator1000-01/data/chains/simple_container_2000/keystore
728
    // We put it in
729
    // Collator1000-01/data/config/keystore
730
    // And same for "network" folder
731
    // But zombienet will put the keys in the old path, so we need to manually copy it if we
732
    // are running under zombienet
733
    copy_zombienet_keystore(&keystore)?;
734

            
735
    let keystore_container = KeystoreContainer::new(&keystore)?;
736

            
737
    // No metrics so no prometheus registry
738
    let prometheus_registry = None;
739
    let mut task_manager = TaskManager::new(tokio_handle.clone(), prometheus_registry)?;
740

            
741
    // Each container chain will spawn its own telemetry
742
    let telemetry_worker_handle = None;
743

            
744
    // Dummy parachain config only needed because `build_relay_chain_interface` needs to know if we
745
    // are collators or not
746
    let validator = container_chain_cli.base.collator;
747
    let mut dummy_parachain_config = dummy_config(
748
        polkadot_config.tokio_handle.clone(),
749
        polkadot_config.base_path.clone(),
750
    );
751
    dummy_parachain_config.role = if validator {
752
        Role::Authority
753
    } else {
754
        Role::Full
755
    };
756
    let (relay_chain_interface, collator_key) =
757
        cumulus_client_service::build_relay_chain_interface(
758
            polkadot_config,
759
            &dummy_parachain_config,
760
            telemetry_worker_handle.clone(),
761
            &mut task_manager,
762
            collator_options.clone(),
763
            hwbench.clone(),
764
        )
765
        .await
766
        .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
767

            
768
    log::info!("start_solochain_node: is validator? {}", validator);
769

            
770
    let overseer_handle = relay_chain_interface
771
        .overseer_handle()
772
        .map_err(|e| sc_service::Error::Application(Box::new(e)))?;
773
    let sync_keystore = keystore_container.keystore();
774
    let collate_on_tanssi: Arc<
775
        dyn Fn() -> (CancellationToken, futures::channel::oneshot::Receiver<()>) + Send + Sync,
776
    > = Arc::new(move || {
777
        // collate_on_tanssi will not be called in solochains because solochains use a different consensus
778
        // mechanism and need validators instead of collators.
779
        // The runtime enforces this because the orchestrator_chain is never assigned any collators.
780
        panic!("Called collate_on_tanssi on solochain collator. This is unsupported and the runtime shouldn't allow this, it is a bug")
781
    });
782

            
783
    let orchestrator_chain_interface_builder = OrchestratorChainSolochainInterfaceBuilder {
784
        overseer_handle: overseer_handle.clone(),
785
        relay_chain_interface: relay_chain_interface.clone(),
786
    };
787
    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();
788
    // Channel to send messages to start/stop container chains
789
    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();
790

            
791
    if validator {
792
        // Start task which detects para id assignment, and starts/stops container chains.
793
        build_check_assigned_para_id(
794
            orchestrator_chain_interface.clone(),
795
            sync_keystore.clone(),
796
            cc_spawn_tx.clone(),
797
            task_manager.spawn_essential_handle(),
798
        );
799
    }
800

            
801
    // If the orchestrator chain is running as a full-node, we start a full node for the
802
    // container chain immediately, because only collator nodes detect their container chain
803
    // assignment so otherwise it will never start.
804
    if !validator {
805
        if let Some(container_chain_para_id) = container_chain_cli.base.para_id {
806
            // Spawn new container chain node
807
            cc_spawn_tx
808
                .send(CcSpawnMsg::UpdateAssignment {
809
                    current: Some(container_chain_para_id.into()),
810
                    next: Some(container_chain_para_id.into()),
811
                })
812
                .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
813
        }
814
    }
815

            
816
    // Start container chain spawner task. This will start and stop container chains on demand.
817
    let spawn_handle = task_manager.spawn_handle();
818

            
819
    let container_chain_spawner = ContainerChainSpawner {
820
        params: ContainerChainSpawnParams {
821
            orchestrator_chain_interface,
822
            container_chain_cli,
823
            tokio_handle,
824
            chain_type,
825
            relay_chain,
826
            relay_chain_interface,
827
            sync_keystore,
828
            orchestrator_para_id,
829
            collation_params: if validator {
830
                Some(spawner::CollationParams {
831
                    // TODO: all these args must be solochain instead of orchestrator
832
                    orchestrator_client: None,
833
                    orchestrator_tx_pool: None,
834
                    orchestrator_para_id,
835
                    collator_key: collator_key
836
                        .expect("there should be a collator key if we're a validator"),
837
                    solochain: true,
838
                })
839
            } else {
840
                None
841
            },
842
            spawn_handle,
843
            data_preserver: false,
844
            generate_rpc_builder: tc_service_container_chain::rpc::GenerateSubstrateRpcBuilder::<
845
                dancebox_runtime::RuntimeApi,
846
            >::new(),
847
            phantom: PhantomData,
848
        },
849
        state: Default::default(),
850
        db_folder_cleanup_done: false,
851
        collate_on_tanssi,
852
        collation_cancellation_constructs: None,
853
    };
854
    let state = container_chain_spawner.state.clone();
855

            
856
    task_manager.spawn_essential_handle().spawn(
857
        "container-chain-spawner-rx-loop",
858
        None,
859
        container_chain_spawner.rx_loop(cc_spawn_rx, validator, true),
860
    );
861

            
862
    task_manager.spawn_essential_handle().spawn(
863
        "container-chain-spawner-debug-state",
864
        None,
865
        monitor::monitor_task(state),
866
    );
867

            
868
    Ok(task_manager)
869
}
870

            
871
pub const SOFT_DEADLINE_PERCENT: sp_runtime::Percent = sp_runtime::Percent::from_percent(100);
872

            
873
/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.
874
///
875
/// This is the actual implementation that is abstract over the executor and the runtime api.
876
182
#[sc_tracing::logging::prefix_logs_with("Orchestrator Dev Node")]
877
pub fn start_dev_node(
878
    orchestrator_config: Configuration,
879
    sealing: Sealing,
880
    hwbench: Option<sc_sysinfo::HwBench>,
881
    para_id: ParaId,
882
) -> sc_service::error::Result<TaskManager> {
883
    let parachain_config = prepare_node_config(orchestrator_config);
884

            
885
    // Create a `NodeBuilder` which helps setup parachain nodes common systems.
886
    let node_builder = NodeConfig::new_builder(&parachain_config, hwbench)?;
887

            
888
    // This node block import.
889
    let block_import = DevParachainBlockImport::new(node_builder.client.clone());
890
    let import_queue = build_manual_seal_import_queue(
891
        node_builder.client.clone(),
892
        block_import.clone(),
893
        &parachain_config,
894
        node_builder
895
            .telemetry
896
            .as_ref()
897
            .map(|telemetry| telemetry.handle()),
898
        &node_builder.task_manager,
899
    )?;
900

            
901
    // Build a Substrate Network. (not cumulus since it is a dev node, it mocks
902
    // the relaychain)
903
    let mut node_builder = node_builder
904
        .build_substrate_network::<sc_network::NetworkWorker<_, _>>(
905
            &parachain_config,
906
            import_queue,
907
        )?;
908

            
909
    // If we're running a collator dev node we must install manual seal block
910
    // production.
911
    let mut command_sink = None;
912
    let mut xcm_senders = None;
913
    let mut randomness_sender = None;
914
    if parachain_config.role.is_authority() {
915
        let client = node_builder.client.clone();
916
        let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);
917
        let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);
918
        // Create channels for mocked parachain candidates.
919
        let (mock_randomness_sender, mock_randomness_receiver) =
920
            flume::bounded::<(bool, Option<[u8; 32]>)>(100);
921

            
922
        xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));
923
        randomness_sender = Some(mock_randomness_sender);
924

            
925
        command_sink = node_builder.install_manual_seal(ManualSealConfiguration {
926
            block_import,
927
            sealing,
928
            soft_deadline: Some(SOFT_DEADLINE_PERCENT),
929
            select_chain: sc_consensus::LongestChain::new(node_builder.backend.clone()),
930
            consensus_data_provider: Some(Box::new(
931
                tc_consensus::OrchestratorManualSealAuraConsensusDataProvider::new(
932
                    node_builder.client.clone(),
933
                    node_builder.keystore_container.keystore(),
934
                    para_id,
935
                ),
936
            )),
937
7274
            create_inherent_data_providers: move |block: H256, ()| {
938
7274
                let current_para_block = client
939
7274
                    .number(block)
940
7274
                    .expect("Header lookup should succeed")
941
7274
                    .expect("Header passed in as parent should be present in backend.");
942
7274

            
943
7274
                let para_ids = client
944
7274
                    .runtime_api()
945
7274
                    .registered_paras(block)
946
7274
                    .expect("registered_paras runtime API should exist")
947
7274
                    .into_iter()
948
7274
                    .collect();
949
7274

            
950
7274
                let hash = client
951
7274
                    .hash(current_para_block.saturating_sub(1))
952
7274
                    .expect("Hash of the desired block must be present")
953
7274
                    .expect("Hash of the desired block should exist");
954
7274

            
955
7274
                let para_header = client
956
7274
                    .expect_header(hash)
957
7274
                    .expect("Expected parachain header should exist")
958
7274
                    .encode();
959
7274

            
960
7274
                let para_head_data = HeadData(para_header).encode();
961
7274
                let para_head_key = RelayWellKnownKeys::para_head(para_id);
962
7274
                let relay_slot_key = RelayWellKnownKeys::CURRENT_SLOT.to_vec();
963
7274

            
964
7274
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
965
7274
                    &*client.clone(),
966
7274
                    block,
967
7274
                ).expect("Slot duration should be set");
968
7274

            
969
7274
                let mut timestamp = 0u64;
970
7274
                TIMESTAMP.with(|x| {
971
7274
                    timestamp = x.clone().take();
972
7274
                });
973
7274

            
974
7274
                timestamp += dancebox_runtime::SLOT_DURATION;
975
7274
                let relay_slot = sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
976
7274
						timestamp.into(),
977
7274
						slot_duration,
978
7274
                    );
979
7274
                let relay_slot = u64::from(*relay_slot);
980
7274

            
981
7274
                let downward_xcm_receiver = downward_xcm_receiver.clone();
982
7274
                let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
983
7274

            
984
7274
                let randomness_enabler_messages: Vec<(bool, Option<[u8; 32]>)> = mock_randomness_receiver.drain().collect();
985

            
986
                // If there is a value to be updated, we update it
987
7274
                if let Some((enable_randomness, new_seed)) = randomness_enabler_messages.last() {
988
4
                    let value = client
989
4
                        .get_aux(RANDOMNESS_ACTIVATED_AUX_KEY)
990
4
                        .expect("Should be able to query aux storage; qed").unwrap_or((false, Option::<[u8; 32]>::None).encode());
991
4
                    let (_mock_additional_randomness, mut mock_randomness_seed): (bool, Option<[u8; 32]>) = Decode::decode(&mut value.as_slice()).expect("Boolean non-decodable");
992

            
993
4
                    if let Some(new_seed) = new_seed {
994
2
                        mock_randomness_seed = Some(*new_seed);
995
2
                    }
996

            
997
4
                    client
998
4
                    .insert_aux(
999
4
                        &[(RANDOMNESS_ACTIVATED_AUX_KEY, (enable_randomness, mock_randomness_seed).encode().as_slice())],
4
                        &[],
4
                    )
4
                    .expect("Should be able to write to aux storage; qed");
7270
                }
                // We read the value
                // If error when reading, we simply put false
7274
                let value = client
7274
                    .get_aux(RANDOMNESS_ACTIVATED_AUX_KEY)
7274
                    .expect("Should be able to query aux storage; qed").unwrap_or((false, Option::<[u8; 32]>::None).encode());
7274
                let (mock_additional_randomness, mock_randomness_seed): (bool, Option<[u8; 32]>) = Decode::decode(&mut value.as_slice()).expect("Boolean non-decodable");
7274

            
7274
                let client_set_aside_for_cidp = client.clone();
7274
                let client_for_xcm = client.clone();
7274
                async move {
7274
                    let mocked_author_noting =
7274
                        tp_author_noting_inherent::MockAuthorNotingInherentDataProvider {
7274
                            current_para_block,
7274
                            relay_offset: 1000,
7274
                            relay_blocks_per_para_block: 2,
7274
                            para_ids,
7274
                            slots_per_para_block: 1,
7274
                        };
7274
                    let mut additional_keys = mocked_author_noting.get_key_values();
7274
                    // Mock only chain 2002 in relay.
7274
                    // This will allow any signed origin to deregister chains 2000 and 2001, and register 2002.
7274
                    let (registrar_paras_key_2002, para_info_2002) = mocked_relay_keys::get_mocked_registrar_paras(2002.into());
7274
                    additional_keys.extend([(para_head_key, para_head_data), (relay_slot_key, Slot::from(relay_slot).encode()), (registrar_paras_key_2002, para_info_2002)]);
7274

            
7274
                    if mock_additional_randomness {
200
                        let mut mock_randomness: [u8; 32] = [0u8; 32];
200
                        mock_randomness[..4].copy_from_slice(&current_para_block.to_be_bytes());
200
                        if let Some(seed) = mock_randomness_seed {
3300
                            for i in 0..32 {
3200
                                mock_randomness[i] ^= seed[i];
3200
                            }
100
                        }
200
                        additional_keys.extend([(RelayWellKnownKeys::CURRENT_BLOCK_RANDOMNESS.to_vec(), Some(mock_randomness).encode())]);
200
                        log::info!("mokcing randomnessss!!! {}", current_para_block);
7074
                    }
7274
                    let current_para_head = client_set_aside_for_cidp
7274
                            .header(block)
7274
                            .expect("Header lookup should succeed")
7274
                            .expect("Header passed in as parent should be present in backend.");
7274
                    let should_send_go_ahead = match client_set_aside_for_cidp
7274
                            .runtime_api()
7274
                            .collect_collation_info(block, &current_para_head)
                    {
7274
                            Ok(info) => info.new_validation_code.is_some(),
                            Err(e) => {
                                    log::error!("Failed to collect collation info: {:?}", e);
                                    false
                            },
                    };
7274
                    let time = MockTimestampInherentDataProvider;
7274
                    let mocked_parachain = MockValidationDataInherentDataProvider {
7274
                        current_para_block,
7274
                        current_para_block_head: None,
7274
                        relay_offset: 1000,
7274
                        relay_blocks_per_para_block: 2,
7274
                        // TODO: Recheck
7274
                        para_blocks_per_relay_epoch: 10,
7274
                        relay_randomness_config: (),
7274
                        xcm_config: MockXcmConfig::new(
7274
                            &*client_for_xcm,
7274
                            block,
7274
                            Default::default(),
7274
                        ),
7274
                        raw_downward_messages: downward_xcm_receiver.drain().collect(),
7274
                        raw_horizontal_messages: hrmp_xcm_receiver.drain().collect(),
7274
                        additional_key_values: Some(additional_keys),
7274
                        para_id,
7274
                        upgrade_go_ahead: should_send_go_ahead.then(|| {
                            log::info!(
                                "Detected pending validation code, sending go-ahead signal."
                            );
                            UpgradeGoAhead::GoAhead
7274
                        }),
7274
                    };
7274

            
7274
                    Ok((time, mocked_parachain, mocked_author_noting))
7274
                }
7274
            },
        })?;
    }
    // This node RPC builder.
    let rpc_builder = {
        let client = node_builder.client.clone();
        let transaction_pool = node_builder.transaction_pool.clone();
364
        Box::new(move |_| {
364
            let deps = crate::rpc::FullDeps {
364
                client: client.clone(),
364
                pool: transaction_pool.clone(),
364
                command_sink: command_sink.clone(),
364
                xcm_senders: xcm_senders.clone(),
364
                randomness_sender: randomness_sender.clone(),
364
            };
364

            
364
            crate::rpc::create_full(deps).map_err(Into::into)
364
        })
    };
    // We spawn all the common substrate tasks to properly run a node.
    let node_builder = node_builder.spawn_common_tasks(parachain_config, rpc_builder)?;
    log::info!("Development Service Ready");
    // We start the networking part.
    node_builder.network.start_network.start_network();
    Ok(node_builder.task_manager)
}
/// Can be called for a `Configuration` to check if it is a configuration for
/// the orchestrator network.
pub trait IdentifyVariant {
    /// Returns `true` if this is a configuration for a dev network.
    fn is_dev(&self) -> bool;
}
impl IdentifyVariant for Box<dyn sc_service::ChainSpec> {
182
    fn is_dev(&self) -> bool {
182
        self.chain_type() == sc_chain_spec::ChainType::Development
182
    }
}
/// Builder for a concrete relay chain interface, created from a full node. Builds
/// a [`RelayChainInProcessInterface`] to access relay chain data necessary for parachain operation.
///
/// The builder takes a [`polkadot_client::Client`]
/// that wraps a concrete instance. By using [`polkadot_client::ExecuteWithClient`]
/// the builder gets access to this concrete instance and instantiates a [`RelayChainInProcessInterface`] with it.
struct OrchestratorChainInProcessInterfaceBuilder {
    client: Arc<ParachainClient>,
    backend: Arc<FullBackend>,
    sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
    overseer_handle: Handle,
}
impl OrchestratorChainInProcessInterfaceBuilder {
    pub fn build(self) -> Arc<dyn OrchestratorChainInterface> {
        Arc::new(OrchestratorChainInProcessInterface::new(
            self.client,
            self.backend,
            self.sync_oracle,
            self.overseer_handle,
        ))
    }
}
/// Builder for a concrete relay chain interface, created from a full node. Builds
/// a [`RelayChainInProcessInterface`] to access relay chain data necessary for parachain operation.
///
/// The builder takes a [`polkadot_client::Client`]
/// that wraps a concrete instance. By using [`polkadot_client::ExecuteWithClient`]
/// the builder gets access to this concrete instance and instantiates a [`RelayChainInProcessInterface`] with it.
struct OrchestratorChainSolochainInterfaceBuilder {
    overseer_handle: Handle,
    relay_chain_interface: Arc<dyn RelayChainInterface>,
}
impl OrchestratorChainSolochainInterfaceBuilder {
    pub fn build(self) -> Arc<dyn OrchestratorChainInterface> {
        Arc::new(OrchestratorChainSolochainInterface::new(
            self.overseer_handle,
            self.relay_chain_interface,
        ))
    }
}
/// Provides an implementation of the [`RelayChainInterface`] using a local in-process relay chain node.
pub struct OrchestratorChainInProcessInterface<Client> {
    pub full_client: Arc<Client>,
    pub backend: Arc<FullBackend>,
    pub sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
    pub overseer_handle: Handle,
}
impl<Client> OrchestratorChainInProcessInterface<Client> {
    /// Create a new instance of [`RelayChainInProcessInterface`]
    pub fn new(
        full_client: Arc<Client>,
        backend: Arc<FullBackend>,
        sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
        overseer_handle: Handle,
    ) -> Self {
        Self {
            full_client,
            backend,
            sync_oracle,
            overseer_handle,
        }
    }
}
impl<T> Clone for OrchestratorChainInProcessInterface<T> {
    fn clone(&self) -> Self {
        Self {
            full_client: self.full_client.clone(),
            backend: self.backend.clone(),
            sync_oracle: self.sync_oracle.clone(),
            overseer_handle: self.overseer_handle.clone(),
        }
    }
}
#[async_trait::async_trait]
impl<Client> OrchestratorChainInterface for OrchestratorChainInProcessInterface<Client>
where
    Client: ProvideRuntimeApi<Block>
        + BlockchainEvents<Block>
        + AuxStore
        + UsageProvider<Block>
        + Sync
        + Send,
    Client::Api: TanssiAuthorityAssignmentApi<Block, NimbusId>
        + OnDemandBlockProductionApi<Block, ParaId, Slot>
        + RegistrarApi<Block, ParaId>
        + AuthorNotingApi<Block, AccountId, BlockNumber, ParaId>
        + DataPreserversApi<Block, DataPreserverProfileId, ParaId>,
{
    async fn get_storage_by_key(
        &self,
        orchestrator_parent: PHash,
        key: &[u8],
    ) -> OrchestratorChainResult<Option<StorageValue>> {
        let state = self.backend.state_at(orchestrator_parent)?;
        state
            .storage(key)
            .map_err(OrchestratorChainError::GenericError)
    }
    async fn prove_read(
        &self,
        orchestrator_parent: PHash,
        relevant_keys: &Vec<Vec<u8>>,
    ) -> OrchestratorChainResult<StorageProof> {
        let state_backend = self.backend.state_at(orchestrator_parent)?;
        sp_state_machine::prove_read(state_backend, relevant_keys)
            .map_err(OrchestratorChainError::StateMachineError)
    }
    fn overseer_handle(&self) -> OrchestratorChainResult<Handle> {
        Ok(self.overseer_handle.clone())
    }
    /// Get a stream of import block notifications.
    async fn import_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        let notification_stream = self
            .full_client
            .import_notification_stream()
            .map(|notification| notification.header);
        Ok(Box::pin(notification_stream))
    }
    /// Get a stream of new best block notifications.
    async fn new_best_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        let notifications_stream =
            self.full_client
                .import_notification_stream()
                .filter_map(|notification| async move {
                    notification.is_new_best.then_some(notification.header)
                });
        Ok(Box::pin(notifications_stream))
    }
    /// Get a stream of finality notifications.
    async fn finality_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        let notification_stream = self
            .full_client
            .finality_notification_stream()
            .map(|notification| notification.header);
        Ok(Box::pin(notification_stream))
    }
    async fn genesis_data(
        &self,
        orchestrator_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Option<ContainerChainGenesisData>> {
        let runtime_api = self.full_client.runtime_api();
        Ok(runtime_api.genesis_data(orchestrator_parent, para_id)?)
    }
    async fn boot_nodes(
        &self,
        orchestrator_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Vec<Vec<u8>>> {
        let runtime_api = self.full_client.runtime_api();
        Ok(runtime_api.boot_nodes(orchestrator_parent, para_id)?)
    }
    async fn latest_block_number(
        &self,
        orchestrator_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Option<BlockNumber>> {
        let runtime_api = self.full_client.runtime_api();
        Ok(runtime_api.latest_block_number(orchestrator_parent, para_id)?)
    }
    async fn best_block_hash(&self) -> OrchestratorChainResult<PHash> {
        Ok(self.backend.blockchain().info().best_hash)
    }
    async fn finalized_block_hash(&self) -> OrchestratorChainResult<PHash> {
        Ok(self.backend.blockchain().info().finalized_hash)
    }
    async fn data_preserver_active_assignment(
        &self,
        orchestrator_parent: PHash,
        profile_id: DataPreserverProfileId,
    ) -> OrchestratorChainResult<DataPreserverAssignment<ParaId>> {
        let runtime_api = self.full_client.runtime_api();
        use {
            dc_orchestrator_chain_interface::DataPreserverAssignment as InterfaceAssignment,
            pallet_data_preservers_runtime_api::Assignment as RuntimeAssignment,
        };
        Ok(
            match runtime_api.get_active_assignment(orchestrator_parent, profile_id)? {
                RuntimeAssignment::NotAssigned => InterfaceAssignment::NotAssigned,
                RuntimeAssignment::Active(para_id) => InterfaceAssignment::Active(para_id),
                RuntimeAssignment::Inactive(para_id) => InterfaceAssignment::Inactive(para_id),
            },
        )
    }
    async fn check_para_id_assignment(
        &self,
        orchestrator_parent: PHash,
        authority: NimbusId,
    ) -> OrchestratorChainResult<Option<ParaId>> {
        let runtime_api = self.full_client.runtime_api();
        Ok(runtime_api.check_para_id_assignment(orchestrator_parent, authority)?)
    }
    async fn check_para_id_assignment_next_session(
        &self,
        orchestrator_parent: PHash,
        authority: NimbusId,
    ) -> OrchestratorChainResult<Option<ParaId>> {
        let runtime_api = self.full_client.runtime_api();
        Ok(runtime_api.check_para_id_assignment_next_session(orchestrator_parent, authority)?)
    }
}
/// Provides an implementation of the [`RelayChainInterface`] using a local in-process relay chain node.
pub struct OrchestratorChainSolochainInterface {
    pub overseer_handle: Handle,
    pub relay_chain_interface: Arc<dyn RelayChainInterface>,
}
impl OrchestratorChainSolochainInterface {
    /// Create a new instance of [`RelayChainInProcessInterface`]
    pub fn new(
        overseer_handle: Handle,
        relay_chain_interface: Arc<dyn RelayChainInterface>,
    ) -> Self {
        Self {
            overseer_handle,
            relay_chain_interface,
        }
    }
}
#[async_trait::async_trait]
impl OrchestratorChainInterface for OrchestratorChainSolochainInterface {
    async fn get_storage_by_key(
        &self,
        relay_parent: PHash,
        key: &[u8],
    ) -> OrchestratorChainResult<Option<StorageValue>> {
        self.relay_chain_interface
            .get_storage_by_key(relay_parent, key)
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    async fn prove_read(
        &self,
        relay_parent: PHash,
        relevant_keys: &Vec<Vec<u8>>,
    ) -> OrchestratorChainResult<StorageProof> {
        self.relay_chain_interface
            .prove_read(relay_parent, relevant_keys)
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    fn overseer_handle(&self) -> OrchestratorChainResult<Handle> {
        Ok(self.overseer_handle.clone())
    }
    /// Get a stream of import block notifications.
    async fn import_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        self.relay_chain_interface
            .import_notification_stream()
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    /// Get a stream of new best block notifications.
    async fn new_best_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        self.relay_chain_interface
            .new_best_notification_stream()
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    /// Get a stream of finality notifications.
    async fn finality_notification_stream(
        &self,
    ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
        self.relay_chain_interface
            .finality_notification_stream()
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    async fn genesis_data(
        &self,
        relay_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Option<ContainerChainGenesisData>> {
        let res: Option<ContainerChainGenesisData> = call_runtime_api(
            &self.relay_chain_interface,
            "RegistrarApi_genesis_data",
            relay_parent,
            &para_id,
        )
        .await
        .map_err(|e| OrchestratorChainError::Application(Box::new(e)))?;
        Ok(res)
    }
    async fn boot_nodes(
        &self,
        relay_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Vec<Vec<u8>>> {
        let res: Vec<Vec<u8>> = call_runtime_api(
            &self.relay_chain_interface,
            "RegistrarApi_boot_nodes",
            relay_parent,
            &para_id,
        )
        .await
        .map_err(|e| OrchestratorChainError::Application(Box::new(e)))?;
        Ok(res)
    }
    async fn latest_block_number(
        &self,
        relay_parent: PHash,
        para_id: ParaId,
    ) -> OrchestratorChainResult<Option<BlockNumber>> {
        let res: Option<BlockNumber> = call_runtime_api(
            &self.relay_chain_interface,
            "AuthorNotingApi_latest_block_number",
            relay_parent,
            &para_id,
        )
        .await
        .map_err(|e| OrchestratorChainError::Application(Box::new(e)))?;
        Ok(res)
    }
    async fn best_block_hash(&self) -> OrchestratorChainResult<PHash> {
        self.relay_chain_interface
            .best_block_hash()
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    async fn finalized_block_hash(&self) -> OrchestratorChainResult<PHash> {
        self.relay_chain_interface
            .finalized_block_hash()
            .await
            .map_err(|e| OrchestratorChainError::Application(Box::new(e)))
    }
    async fn data_preserver_active_assignment(
        &self,
        _orchestrator_parent: PHash,
        _profile_id: DataPreserverProfileId,
    ) -> OrchestratorChainResult<DataPreserverAssignment<ParaId>> {
        unimplemented!("Data preserver node does not support Dancelight yet")
    }
    async fn check_para_id_assignment(
        &self,
        relay_parent: PHash,
        authority: NimbusId,
    ) -> OrchestratorChainResult<Option<ParaId>> {
        let res: Option<ParaId> = call_runtime_api(
            &self.relay_chain_interface,
            "TanssiAuthorityAssignmentApi_check_para_id_assignment",
            relay_parent,
            &authority,
        )
        .await
        .map_err(|e| OrchestratorChainError::Application(Box::new(e)))?;
        Ok(res)
    }
    async fn check_para_id_assignment_next_session(
        &self,
        relay_parent: PHash,
        authority: NimbusId,
    ) -> OrchestratorChainResult<Option<ParaId>> {
        let res: Option<ParaId> = call_runtime_api(
            &self.relay_chain_interface,
            "TanssiAuthorityAssignmentApi_check_para_id_assignment_next_session",
            relay_parent,
            &authority,
        )
        .await
        .map_err(|e| OrchestratorChainError::Application(Box::new(e)))?;
        Ok(res)
    }
}