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 frame_support::__private::sp_tracing::tracing::Instrument;
20
use {
21
    crate::command::solochain::{
22
        build_solochain_config_dir, copy_zombienet_keystore, dummy_config, keystore_config,
23
    },
24
    core::marker::PhantomData,
25
    cumulus_client_cli::CollatorOptions,
26
    cumulus_client_collator::service::CollatorService,
27
    cumulus_client_consensus_proposer::Proposer,
28
    cumulus_client_parachain_inherent::{MockValidationDataInherentDataProvider, MockXcmConfig},
29
    cumulus_client_service::{
30
        prepare_node_config, start_relay_chain_tasks, DARecoveryProfile, StartRelayChainTasksParams,
31
    },
32
    cumulus_primitives_core::{
33
        relay_chain::{well_known_keys as RelayWellKnownKeys, CollatorPair},
34
        ParaId,
35
    },
36
    cumulus_relay_chain_interface::{
37
        call_remote_runtime_function, OverseerHandle, RelayChainInterface,
38
    },
39
    dancebox_runtime::{
40
        opaque::{Block, Hash},
41
        AccountId, RuntimeApi,
42
    },
43
    dc_orchestrator_chain_interface::{
44
        BlockNumber, ContainerChainGenesisData, DataPreserverAssignment, DataPreserverProfileId,
45
        OrchestratorChainError, OrchestratorChainInterface, OrchestratorChainResult, PHash,
46
        PHeader,
47
    },
48
    futures::{Stream, StreamExt},
49
    nimbus_primitives::{NimbusId, NimbusPair},
50
    node_common::service::{ManualSealConfiguration, NodeBuilder, NodeBuilderConfig, Sealing},
51
    pallet_author_noting_runtime_api::AuthorNotingApi,
52
    pallet_data_preservers_runtime_api::DataPreserversApi,
53
    pallet_registrar_runtime_api::RegistrarApi,
54
    parity_scale_codec::{Decode, Encode},
55
    polkadot_cli::ProvideRuntimeApi,
56
    polkadot_parachain_primitives::primitives::HeadData,
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::StorageProof,
70
    sp_consensus::SyncOracle,
71
    sp_consensus_slots::Slot,
72
    sp_core::{traits::SpawnEssentialNamed, H256},
73
    sp_keystore::KeystorePtr,
74
    sp_state_machine::{Backend as StateBackend, StorageValue},
75
    std::{pin::Pin, sync::Arc, time::Duration},
76
    tc_consensus::{
77
        collators::lookahead::{
78
            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,
79
        },
80
        OnDemandBlockProductionApi, OrchestratorAuraWorkerAuxData, TanssiAuthorityAssignmentApi,
81
    },
82
    tc_service_container_chain::{
83
        cli::ContainerChainCli,
84
        monitor,
85
        service::{
86
            DevParachainBlockImport, ParachainBlockImport, ParachainClient, ParachainExecutor,
87
            ParachainProposerFactory,
88
        },
89
        spawner::{self, CcSpawnMsg, ContainerChainSpawnParams, ContainerChainSpawner},
90
    },
91
    tokio::sync::mpsc::{unbounded_channel, UnboundedSender},
92
    tokio_util::sync::CancellationToken,
93
};
94

            
95
mod mocked_relay_keys;
96

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

            
100
type FullBackend = TFullBackend<Block>;
101

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

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

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

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

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

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

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

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

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

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

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

            
212
    Ok(())
213
}
214

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

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

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

            
239
    (block_import, import_queue)
240
}
241

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
557
    let params = LookaheadTanssiAuraParams {
558
        get_current_slot_duration: move |block_hash| {
559
            sc_consensus_aura::standalone::slot_duration_at(
560
                &*client_for_slot_duration_provider,
561
                block_hash,
562
            )
563
            .expect("Slot duration should be set")
564
        },
565
        create_inherent_data_providers: move |block_hash, (relay_parent, _validation_data)| {
566
            let relay_chain_interface = relay_chain_interace_for_cidp.clone();
567
            let client_set_aside_for_cidp = client_set_aside_for_cidp.clone();
568
            async move {
569
                let para_ids = client_set_aside_for_cidp
570
                    .runtime_api()
571
                    .registered_paras(block_hash)?;
572
                let para_ids: Vec<_> = para_ids.into_iter().collect();
573
                let author_noting_inherent =
574
                    tp_author_noting_inherent::OwnParachainInherentData::create_at(
575
                        relay_parent,
576
                        &relay_chain_interface,
577
                        &para_ids,
578
                    )
579
                    .await;
580

            
581
                // Fetch duration every block to avoid downtime when passing from 12 to 6s
582
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
583
                    &*client_set_aside_for_cidp.clone(),
584
                    block_hash,
585
                )
586
                .expect("Slot duration should be set");
587

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

            
590
                let slot =
591
						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
592
							*timestamp,
593
							slot_duration,
594
						);
595

            
596
                let author_noting_inherent = author_noting_inherent.ok_or_else(|| {
597
                    Box::<dyn std::error::Error + Send + Sync>::from(
598
                        "Failed to create author noting inherent",
599
                    )
600
                })?;
601

            
602
                Ok((slot, timestamp, author_noting_inherent))
603
            }
604
        },
605
        get_orchestrator_aux_data: move |block_hash: H256, (_relay_parent, _validation_data)| {
606
            let client_set_aside_for_orch = client_set_aside_for_orch.clone();
607

            
608
            async move {
609
                let authorities = tc_consensus::authorities::<Block, ParachainClient, NimbusPair>(
610
                    client_set_aside_for_orch.as_ref(),
611
                    &block_hash,
612
                    para_id,
613
                );
614

            
615
                let authorities = authorities.ok_or_else(|| {
616
                    Box::<dyn std::error::Error + Send + Sync>::from(
617
                        "Failed to fetch authorities with error",
618
                    )
619
                })?;
620

            
621
                log::info!(
622
                    "Authorities {:?} found for header {:?}",
623
                    authorities,
624
                    block_hash
625
                );
626

            
627
                let aux_data = OrchestratorAuraWorkerAuxData {
628
                    authorities,
629
                    // This is the orchestrator consensus, it does not have a slot frequency
630
                    slot_freq: None,
631
                };
632

            
633
                Ok(aux_data)
634
            }
635
        },
636
        block_import,
637
        para_client: client,
638
        relay_client: relay_chain_interface,
639
        sync_oracle,
640
        keystore,
641
        collator_key,
642
        para_id,
643
        overseer_handle,
644
        orchestrator_slot_duration: slot_duration,
645
        relay_chain_slot_duration,
646
        force_authoring,
647
        proposer,
648
        collator_service,
649
        authoring_duration: Duration::from_millis(2000),
650
        code_hash_provider,
651
        para_backend: backend,
652
        cancellation_token: cancellation_token.clone(),
653
        buy_core_params,
654
    };
655

            
656
    let (fut, exit_notification_receiver) =
657
        lookahead_tanssi_aura::run::<_, Block, NimbusPair, _, _, _, _, _, _, _, _, _, _, _, _, _>(
658
            params,
659
        );
660
    spawner.spawn("tanssi-aura", None, fut);
661

            
662
    (cancellation_token, exit_notification_receiver)
663
}
664

            
665
/// Start a parachain node.
666
pub async fn start_parachain_node(
667
    parachain_config: Configuration,
668
    polkadot_config: Configuration,
669
    container_config: Option<(ContainerChainCli, tokio::runtime::Handle)>,
670
    collator_options: CollatorOptions,
671
    para_id: ParaId,
672
    hwbench: Option<sc_sysinfo::HwBench>,
673
) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient>)> {
674
    start_node_impl(
675
        parachain_config,
676
        polkadot_config,
677
        container_config,
678
        collator_options,
679
        para_id,
680
        hwbench,
681
    )
682
    .instrument(sc_tracing::tracing::info_span!(
683
        sc_tracing::logging::PREFIX_LOG_SPAN,
684
        name = "Orchestrator",
685
    ))
686
    .await
687
}
688

            
689
/// Start a solochain node.
690
pub async fn start_solochain_node(
691
    polkadot_config: Configuration,
692
    container_chain_cli: ContainerChainCli,
693
    collator_options: CollatorOptions,
694
    hwbench: Option<sc_sysinfo::HwBench>,
695
) -> sc_service::error::Result<TaskManager> {
696
    let tokio_handle = polkadot_config.tokio_handle.clone();
697
    let orchestrator_para_id = Default::default();
698

            
699
    let chain_type = polkadot_config.chain_spec.chain_type().clone();
700
    let relay_chain = polkadot_config.chain_spec.id().to_string();
701

            
702
    let base_path = container_chain_cli
703
        .base
704
        .base
705
        .shared_params
706
        .base_path
707
        .as_ref()
708
        .expect("base_path is always set");
709
    let config_dir = build_solochain_config_dir(base_path);
710
    let keystore = keystore_config(container_chain_cli.keystore_params(), &config_dir)
711
        .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
712

            
713
    // Instead of putting keystore in
714
    // Collator1000-01/data/chains/simple_container_2000/keystore
715
    // We put it in
716
    // Collator1000-01/data/config/keystore
717
    // And same for "network" folder
718
    // But zombienet will put the keys in the old path, so we need to manually copy it if we
719
    // are running under zombienet
720
    copy_zombienet_keystore(&keystore)?;
721

            
722
    let keystore_container = KeystoreContainer::new(&keystore)?;
723

            
724
    // No metrics so no prometheus registry
725
    let prometheus_registry = None;
726
    let mut task_manager = TaskManager::new(tokio_handle.clone(), prometheus_registry)?;
727

            
728
    // Each container chain will spawn its own telemetry
729
    let telemetry_worker_handle = None;
730

            
731
    // Dummy parachain config only needed because `build_relay_chain_interface` needs to know if we
732
    // are collators or not
733
    let validator = container_chain_cli.base.collator;
734
    let mut dummy_parachain_config = dummy_config(
735
        polkadot_config.tokio_handle.clone(),
736
        polkadot_config.base_path.clone(),
737
    );
738
    dummy_parachain_config.role = if validator {
739
        Role::Authority
740
    } else {
741
        Role::Full
742
    };
743
    let (relay_chain_interface, collator_key) =
744
        cumulus_client_service::build_relay_chain_interface(
745
            polkadot_config,
746
            &dummy_parachain_config,
747
            telemetry_worker_handle.clone(),
748
            &mut task_manager,
749
            collator_options.clone(),
750
            hwbench.clone(),
751
        )
752
        .await
753
        .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
754

            
755
    log::info!("start_solochain_node: is validator? {}", validator);
756

            
757
    let overseer_handle = relay_chain_interface
758
        .overseer_handle()
759
        .map_err(|e| sc_service::Error::Application(Box::new(e)))?;
760
    let sync_keystore = keystore_container.keystore();
761
    let collate_on_tanssi: Arc<
762
        dyn Fn() -> (CancellationToken, futures::channel::oneshot::Receiver<()>) + Send + Sync,
763
    > = Arc::new(move || {
764
        // collate_on_tanssi will not be called in solochains because solochains use a different consensus
765
        // mechanism and need validators instead of collators.
766
        // The runtime enforces this because the orchestrator_chain is never assigned any collators.
767
        panic!("Called collate_on_tanssi on solochain collator. This is unsupported and the runtime shouldn't allow this, it is a bug")
768
    });
769

            
770
    let orchestrator_chain_interface_builder = OrchestratorChainSolochainInterfaceBuilder {
771
        overseer_handle: overseer_handle.clone(),
772
        relay_chain_interface: relay_chain_interface.clone(),
773
    };
774
    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();
775
    // Channel to send messages to start/stop container chains
776
    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();
777

            
778
    if validator {
779
        // Start task which detects para id assignment, and starts/stops container chains.
780
        build_check_assigned_para_id(
781
            orchestrator_chain_interface.clone(),
782
            sync_keystore.clone(),
783
            cc_spawn_tx.clone(),
784
            task_manager.spawn_essential_handle(),
785
        );
786
    }
787

            
788
    // If the orchestrator chain is running as a full-node, we start a full node for the
789
    // container chain immediately, because only collator nodes detect their container chain
790
    // assignment so otherwise it will never start.
791
    if !validator {
792
        if let Some(container_chain_para_id) = container_chain_cli.base.para_id {
793
            // Spawn new container chain node
794
            cc_spawn_tx
795
                .send(CcSpawnMsg::UpdateAssignment {
796
                    current: Some(container_chain_para_id.into()),
797
                    next: Some(container_chain_para_id.into()),
798
                })
799
                .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
800
        }
801
    }
802

            
803
    // Start container chain spawner task. This will start and stop container chains on demand.
804
    let spawn_handle = task_manager.spawn_handle();
805

            
806
    let container_chain_spawner = ContainerChainSpawner {
807
        params: ContainerChainSpawnParams {
808
            orchestrator_chain_interface,
809
            container_chain_cli,
810
            tokio_handle,
811
            chain_type,
812
            relay_chain,
813
            relay_chain_interface,
814
            sync_keystore,
815
            orchestrator_para_id,
816
            collation_params: if validator {
817
                Some(spawner::CollationParams {
818
                    // TODO: all these args must be solochain instead of orchestrator
819
                    orchestrator_client: None,
820
                    orchestrator_tx_pool: None,
821
                    orchestrator_para_id,
822
                    collator_key: collator_key
823
                        .expect("there should be a collator key if we're a validator"),
824
                    solochain: true,
825
                })
826
            } else {
827
                None
828
            },
829
            spawn_handle,
830
            data_preserver: false,
831
            generate_rpc_builder: tc_service_container_chain::rpc::GenerateSubstrateRpcBuilder::<
832
                dancebox_runtime::RuntimeApi,
833
            >::new(),
834
            phantom: PhantomData,
835
        },
836
        state: Default::default(),
837
        db_folder_cleanup_done: false,
838
        collate_on_tanssi,
839
        collation_cancellation_constructs: None,
840
    };
841
    let state = container_chain_spawner.state.clone();
842

            
843
    task_manager.spawn_essential_handle().spawn(
844
        "container-chain-spawner-rx-loop",
845
        None,
846
        container_chain_spawner.rx_loop(cc_spawn_rx, validator, true),
847
    );
848

            
849
    task_manager.spawn_essential_handle().spawn(
850
        "container-chain-spawner-debug-state",
851
        None,
852
        monitor::monitor_task(state),
853
    );
854

            
855
    Ok(task_manager)
856
}
857

            
858
pub const SOFT_DEADLINE_PERCENT: sp_runtime::Percent = sp_runtime::Percent::from_percent(100);
859

            
860
/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.
861
///
862
/// This is the actual implementation that is abstract over the executor and the runtime api.
863
182
#[sc_tracing::logging::prefix_logs_with("Orchestrator Dev Node")]
864
pub fn start_dev_node(
865
    orchestrator_config: Configuration,
866
    sealing: Sealing,
867
    hwbench: Option<sc_sysinfo::HwBench>,
868
    para_id: ParaId,
869
) -> sc_service::error::Result<TaskManager> {
870
    let parachain_config = prepare_node_config(orchestrator_config);
871

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

            
875
    // This node block import.
876
    let block_import = DevParachainBlockImport::new(node_builder.client.clone());
877
    let import_queue = build_manual_seal_import_queue(
878
        node_builder.client.clone(),
879
        block_import.clone(),
880
        &parachain_config,
881
        node_builder
882
            .telemetry
883
            .as_ref()
884
            .map(|telemetry| telemetry.handle()),
885
        &node_builder.task_manager,
886
    )?;
887

            
888
    // Build a Substrate Network. (not cumulus since it is a dev node, it mocks
889
    // the relaychain)
890
    let mut node_builder = node_builder
891
        .build_substrate_network::<sc_network::NetworkWorker<_, _>>(
892
            &parachain_config,
893
            import_queue,
894
        )?;
895

            
896
    // If we're running a collator dev node we must install manual seal block
897
    // production.
898
    let mut command_sink = None;
899
    let mut xcm_senders = None;
900
    let mut randomness_sender = None;
901
    if parachain_config.role.is_authority() {
902
        let client = node_builder.client.clone();
903
        let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);
904
        let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);
905
        // Create channels for mocked parachain candidates.
906
        let (mock_randomness_sender, mock_randomness_receiver) =
907
            flume::bounded::<(bool, Option<[u8; 32]>)>(100);
908

            
909
        xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));
910
        randomness_sender = Some(mock_randomness_sender);
911

            
912
        command_sink = node_builder.install_manual_seal(ManualSealConfiguration {
913
            block_import,
914
            sealing,
915
            soft_deadline: Some(SOFT_DEADLINE_PERCENT),
916
            select_chain: sc_consensus::LongestChain::new(node_builder.backend.clone()),
917
            consensus_data_provider: Some(Box::new(
918
                tc_consensus::OrchestratorManualSealAuraConsensusDataProvider::new(
919
                    node_builder.client.clone(),
920
                    node_builder.keystore_container.keystore(),
921
                    para_id,
922
                ),
923
            )),
924
7274
            create_inherent_data_providers: move |block: H256, ()| {
925
7274
                let current_para_block = client
926
7274
                    .number(block)
927
7274
                    .expect("Header lookup should succeed")
928
7274
                    .expect("Header passed in as parent should be present in backend.");
929
7274

            
930
7274
                let para_ids = client
931
7274
                    .runtime_api()
932
7274
                    .registered_paras(block)
933
7274
                    .expect("registered_paras runtime API should exist")
934
7274
                    .into_iter()
935
7274
                    .collect();
936
7274

            
937
7274
                let hash = client
938
7274
                    .hash(current_para_block.saturating_sub(1))
939
7274
                    .expect("Hash of the desired block must be present")
940
7274
                    .expect("Hash of the desired block should exist");
941
7274

            
942
7274
                let para_header = client
943
7274
                    .expect_header(hash)
944
7274
                    .expect("Expected parachain header should exist")
945
7274
                    .encode();
946
7274

            
947
7274
                let para_head_data = HeadData(para_header).encode();
948
7274
                let para_head_key = RelayWellKnownKeys::para_head(para_id);
949
7274
                let relay_slot_key = RelayWellKnownKeys::CURRENT_SLOT.to_vec();
950
7274

            
951
7274
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
952
7274
                    &*client.clone(),
953
7274
                    block,
954
7274
                ).expect("Slot duration should be set");
955
7274

            
956
7274
                let mut timestamp = 0u64;
957
7274
                TIMESTAMP.with(|x| {
958
7274
                    timestamp = x.clone().take();
959
7274
                });
960
7274

            
961
7274
                timestamp += dancebox_runtime::SLOT_DURATION;
962
7274
                let relay_slot = sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
963
7274
						timestamp.into(),
964
7274
						slot_duration,
965
7274
                    );
966
7274
                let relay_slot = u64::from(*relay_slot);
967
7274

            
968
7274
                let downward_xcm_receiver = downward_xcm_receiver.clone();
969
7274
                let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
970
7274

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

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

            
980
4
                    if let Some(new_seed) = new_seed {
981
2
                        mock_randomness_seed = Some(*new_seed);
982
2
                    }
983

            
984
4
                    client
985
4
                    .insert_aux(
986
4
                        &[(RANDOMNESS_ACTIVATED_AUX_KEY, (enable_randomness, mock_randomness_seed).encode().as_slice())],
987
4
                        &[],
988
4
                    )
989
4
                    .expect("Should be able to write to aux storage; qed");
990
7270
                }
991

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

            
999
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 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
                    };
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_remote_runtime_function(
            &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_remote_runtime_function(
            &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_remote_runtime_function(
            &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_remote_runtime_function(
            &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_remote_runtime_function(
            &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)
    }
}