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
const CONTAINER_CHAINS_EXCLUSION_AUX_KEY: &[u8] = b"__DEV_CONTAINER_CHAINS_EXCLUSION";
102

            
103
type FullBackend = TFullBackend<Block>;
104

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

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

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

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

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

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

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

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

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

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

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

            
215
    Ok(())
216
}
217

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

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

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

            
243
    (block_import, import_queue)
244
}
245

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

            
264
    // Channel to send messages to start/stop container chains
265
    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();
266

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

            
270
    let (block_import, import_queue) = import_queue(&parachain_config, &node_builder);
271

            
272
    let (relay_chain_interface, collator_key) = node_builder
273
        .build_relay_chain_interface(&parachain_config, polkadot_config, collator_options.clone())
274
        .await?;
275

            
276
    let validator = parachain_config.role.is_authority();
277
    let force_authoring = parachain_config.force_authoring;
278

            
279
    let node_builder = node_builder
280
        .build_cumulus_network::<_, sc_network::NetworkWorker<_, _>>(
281
            &parachain_config,
282
            para_id,
283
            import_queue,
284
            relay_chain_interface.clone(),
285
        )
286
        .await?;
287

            
288
    let rpc_builder = {
289
        let client = node_builder.client.clone();
290
        let transaction_pool = node_builder.transaction_pool.clone();
291

            
292
        Box::new(move |_| {
293
            let deps = crate::rpc::FullDeps {
294
                client: client.clone(),
295
                pool: transaction_pool.clone(),
296
                command_sink: None,
297
                xcm_senders: None,
298
                randomness_sender: None,
299
                container_chain_exclusion_sender: None,
300
            };
301

            
302
            crate::rpc::create_full(deps).map_err(Into::into)
303
        })
304
    };
305

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

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

            
323
    let announce_block = {
324
        let sync_service = node_builder.network.sync_service.clone();
325
        Arc::new(move |hash, data| sync_service.announce_block(hash, data))
326
    };
327

            
328
    let (mut node_builder, import_queue_service) = node_builder.extract_import_queue_service();
329

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

            
347
    let orchestrator_chain_interface_builder = OrchestratorChainInProcessInterfaceBuilder {
348
        client: node_builder.client.clone(),
349
        backend: node_builder.backend.clone(),
350
        sync_oracle: node_builder.network.sync_service.clone(),
351
        overseer_handle: overseer_handle.clone(),
352
    };
353
    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();
354

            
355
    if validator {
356
        let collator_key = collator_key
357
            .clone()
358
            .expect("Command line arguments do not allow this. qed");
359

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

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

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

            
416
    let sync_keystore = node_builder.keystore_container.keystore();
417

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

            
434
        // Start container chain spawner task. This will start and stop container chains on demand.
435
        let orchestrator_client = node_builder.client.clone();
436
        let orchestrator_tx_pool = node_builder.transaction_pool.clone();
437
        let spawn_handle = node_builder.task_manager.spawn_handle();
438

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

            
479
        node_builder.task_manager.spawn_essential_handle().spawn(
480
            "container-chain-spawner-rx-loop",
481
            None,
482
            container_chain_spawner.rx_loop(cc_spawn_rx, validator, false),
483
        );
484

            
485
        node_builder.task_manager.spawn_essential_handle().spawn(
486
            "container-chain-spawner-debug-state",
487
            None,
488
            monitor::monitor_task(state),
489
        )
490
    }
491

            
492
    Ok((node_builder.task_manager, node_builder.client))
493
}
494

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

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

            
534
    let proposer = Proposer::new(proposer_factory);
535

            
536
    let collator_service = CollatorService::new(
537
        client.clone(),
538
        Arc::new(spawner.clone()),
539
        announce_block,
540
        client.clone(),
541
    );
542

            
543
    let relay_chain_interace_for_cidp = relay_chain_interface.clone();
544
    let client_set_aside_for_cidp = client.clone();
545
    let client_set_aside_for_orch = client.clone();
546
    let client_for_hash_provider = client.clone();
547
    let client_for_slot_duration_provider = client.clone();
548

            
549
    let code_hash_provider = move |block_hash| {
550
        client_for_hash_provider
551
            .code_at(block_hash)
552
            .ok()
553
            .map(polkadot_primitives::ValidationCode)
554
            .map(|c| c.hash())
555
    };
556

            
557
    let cancellation_token = CancellationToken::new();
558
    let buy_core_params = BuyCoreParams::Orchestrator {
559
        orchestrator_tx_pool,
560
        orchestrator_client: client.clone(),
561
    };
562

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

            
600
                // Fetch duration every block to avoid downtime when passing from 12 to 6s
601
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
602
                    &*client_set_aside_for_cidp.clone(),
603
                    block_hash,
604
                )
605
                .expect("Slot duration should be set");
606

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

            
609
                let slot =
610
						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
611
							*timestamp,
612
							slot_duration,
613
						);
614

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

            
621
                Ok((slot, timestamp, author_noting_inherent))
622
            }
623
        },
624
        get_orchestrator_aux_data: move |block_hash: H256, (_relay_parent, _validation_data)| {
625
            let client_set_aside_for_orch = client_set_aside_for_orch.clone();
626

            
627
            async move {
628
                let authorities = tc_consensus::authorities::<Block, ParachainClient, NimbusPair>(
629
                    client_set_aside_for_orch.as_ref(),
630
                    &block_hash,
631
                    para_id,
632
                );
633

            
634
                let authorities = authorities.ok_or_else(|| {
635
                    Box::<dyn std::error::Error + Send + Sync>::from(
636
                        "Failed to fetch authorities with error",
637
                    )
638
                })?;
639

            
640
                log::info!(
641
                    "Authorities {:?} found for header {:?}",
642
                    authorities,
643
                    block_hash
644
                );
645

            
646
                let aux_data = OrchestratorAuraWorkerAuxData {
647
                    authorities,
648
                    // This is the orchestrator consensus, it does not have a slot frequency
649
                    slot_freq: None,
650
                };
651

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

            
675
    let (fut, exit_notification_receiver) =
676
        lookahead_tanssi_aura::run::<_, Block, NimbusPair, _, _, _, _, _, _, _, _, _, _, _, _, _>(
677
            params,
678
        );
679
    spawner.spawn("tanssi-aura", None, fut);
680

            
681
    (cancellation_token, exit_notification_receiver)
682
}
683

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

            
710
/// Start a solochain node.
711
pub async fn start_solochain_node(
712
    polkadot_config: Configuration,
713
    container_chain_cli: ContainerChainCli,
714
    collator_options: CollatorOptions,
715
    hwbench: Option<sc_sysinfo::HwBench>,
716
) -> sc_service::error::Result<TaskManager> {
717
    let tokio_handle = polkadot_config.tokio_handle.clone();
718
    let orchestrator_para_id = Default::default();
719

            
720
    let chain_type = polkadot_config.chain_spec.chain_type().clone();
721
    let relay_chain = polkadot_config.chain_spec.id().to_string();
722

            
723
    let base_path = container_chain_cli
724
        .base
725
        .base
726
        .shared_params
727
        .base_path
728
        .as_ref()
729
        .expect("base_path is always set");
730
    let config_dir = build_solochain_config_dir(base_path);
731
    let keystore = keystore_config(container_chain_cli.keystore_params(), &config_dir)
732
        .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
733

            
734
    // Instead of putting keystore in
735
    // Collator1000-01/data/chains/simple_container_2000/keystore
736
    // We put it in
737
    // Collator1000-01/data/config/keystore
738
    // And same for "network" folder
739
    // But zombienet will put the keys in the old path, so we need to manually copy it if we
740
    // are running under zombienet
741
    copy_zombienet_keystore(&keystore)?;
742

            
743
    let keystore_container = KeystoreContainer::new(&keystore)?;
744

            
745
    // No metrics so no prometheus registry
746
    let prometheus_registry = None;
747
    let mut task_manager = TaskManager::new(tokio_handle.clone(), prometheus_registry)?;
748

            
749
    // Each container chain will spawn its own telemetry
750
    let telemetry_worker_handle = None;
751

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

            
776
    log::info!("start_solochain_node: is validator? {}", validator);
777

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

            
791
    let orchestrator_chain_interface_builder = OrchestratorChainSolochainInterfaceBuilder {
792
        overseer_handle: overseer_handle.clone(),
793
        relay_chain_interface: relay_chain_interface.clone(),
794
    };
795
    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();
796
    // Channel to send messages to start/stop container chains
797
    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();
798

            
799
    if validator {
800
        // Start task which detects para id assignment, and starts/stops container chains.
801
        build_check_assigned_para_id(
802
            orchestrator_chain_interface.clone(),
803
            sync_keystore.clone(),
804
            cc_spawn_tx.clone(),
805
            task_manager.spawn_essential_handle(),
806
        );
807
    }
808

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

            
824
    // Start container chain spawner task. This will start and stop container chains on demand.
825
    let spawn_handle = task_manager.spawn_handle();
826

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

            
864
    task_manager.spawn_essential_handle().spawn(
865
        "container-chain-spawner-rx-loop",
866
        None,
867
        container_chain_spawner.rx_loop(cc_spawn_rx, validator, true),
868
    );
869

            
870
    task_manager.spawn_essential_handle().spawn(
871
        "container-chain-spawner-debug-state",
872
        None,
873
        monitor::monitor_task(state),
874
    );
875

            
876
    Ok(task_manager)
877
}
878

            
879
pub const SOFT_DEADLINE_PERCENT: sp_runtime::Percent = sp_runtime::Percent::from_percent(100);
880

            
881
/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.
882
///
883
/// This is the actual implementation that is abstract over the executor and the runtime api.
884
#[sc_tracing::logging::prefix_logs_with("Orchestrator Dev Node")]
885
pub fn start_dev_node(
886
    orchestrator_config: Configuration,
887
    sealing: Sealing,
888
    hwbench: Option<sc_sysinfo::HwBench>,
889
    para_id: ParaId,
890
) -> sc_service::error::Result<TaskManager> {
891
    let parachain_config = prepare_node_config(orchestrator_config);
892

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

            
896
    // This node block import.
897
    let block_import = DevParachainBlockImport::new(node_builder.client.clone());
898
    let import_queue = build_manual_seal_import_queue(
899
        node_builder.client.clone(),
900
        block_import.clone(),
901
        &parachain_config,
902
        node_builder
903
            .telemetry
904
            .as_ref()
905
            .map(|telemetry| telemetry.handle()),
906
        &node_builder.task_manager,
907
    )?;
908

            
909
    // Build a Substrate Network. (not cumulus since it is a dev node, it mocks
910
    // the relaychain)
911
    let mut node_builder = node_builder
912
        .build_substrate_network::<sc_network::NetworkWorker<_, _>>(
913
            &parachain_config,
914
            import_queue,
915
        )?;
916

            
917
    // If we're running a collator dev node we must install manual seal block
918
    // production.
919
    let mut command_sink = None;
920
    let mut xcm_senders = None;
921
    let mut randomness_sender = None;
922
    let mut container_chains_exclusion_sender = None;
923
    if parachain_config.role.is_authority() {
924
        let client = node_builder.client.clone();
925
        let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);
926
        let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);
927
        // Create channels for mocked parachain candidates.
928
        let (mock_randomness_sender, mock_randomness_receiver) =
929
            flume::bounded::<(bool, Option<[u8; 32]>)>(100);
930
        // Create channels for mocked exclusion of parachains from producing blocks
931
        let (mock_container_chains_exclusion_sender, mock_container_chains_exclusion_receiver) =
932
            flume::bounded::<Vec<ParaId>>(100);
933

            
934
        xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));
935
        randomness_sender = Some(mock_randomness_sender);
936
        container_chains_exclusion_sender = Some(mock_container_chains_exclusion_sender);
937

            
938
        command_sink = node_builder.install_manual_seal(ManualSealConfiguration {
939
            block_import,
940
            sealing,
941
            soft_deadline: Some(SOFT_DEADLINE_PERCENT),
942
            select_chain: sc_consensus::LongestChain::new(node_builder.backend.clone()),
943
            consensus_data_provider: Some(Box::new(
944
                tc_consensus::OrchestratorManualSealAuraConsensusDataProvider::new(
945
                    node_builder.client.clone(),
946
                    node_builder.keystore_container.keystore(),
947
                    para_id,
948
                ),
949
            )),
950
7810
            create_inherent_data_providers: move |block: H256, ()| {
951
7810
                let current_para_block = client
952
7810
                    .number(block)
953
7810
                    .expect("Header lookup should succeed")
954
7810
                    .expect("Header passed in as parent should be present in backend.");
955
7810

            
956
7810
                let mut para_ids: Vec<ParaId> = client
957
7810
                    .runtime_api()
958
7810
                    .registered_paras(block)
959
7810
                    .expect("registered_paras runtime API should exist")
960
7810
                    .into_iter()
961
7810
                    .collect();
962
7810

            
963
7810
                let hash = client
964
7810
                    .hash(current_para_block.saturating_sub(1))
965
7810
                    .expect("Hash of the desired block must be present")
966
7810
                    .expect("Hash of the desired block should exist");
967
7810

            
968
7810
                let para_header = client
969
7810
                    .expect_header(hash)
970
7810
                    .expect("Expected parachain header should exist")
971
7810
                    .encode();
972
7810

            
973
7810
                let para_head_data = HeadData(para_header).encode();
974
7810
                let para_head_key = RelayWellKnownKeys::para_head(para_id);
975
7810
                let relay_slot_key = RelayWellKnownKeys::CURRENT_SLOT.to_vec();
976
7810

            
977
7810
                let slot_duration = sc_consensus_aura::standalone::slot_duration_at(
978
7810
                    &*client.clone(),
979
7810
                    block,
980
7810
                ).expect("Slot duration should be set");
981
7810

            
982
7810
                let mut timestamp = 0u64;
983
7810
                TIMESTAMP.with(|x| {
984
7810
                    timestamp = x.clone().take();
985
7810
                });
986
7810

            
987
7810
                timestamp += dancebox_runtime::SLOT_DURATION;
988
7810
                let relay_slot = sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
989
7810
						timestamp.into(),
990
7810
						slot_duration,
991
7810
                    );
992
7810
                let relay_slot = u64::from(*relay_slot);
993
7810

            
994
7810
                let downward_xcm_receiver = downward_xcm_receiver.clone();
995
7810
                let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
996
7810

            
997
7810
                let randomness_enabler_messages: Vec<(bool, Option<[u8; 32]>)> = mock_randomness_receiver.drain().collect();
998

            
999
                // If there is a value to be updated, we update it
7810
                if let Some((enable_randomness, new_seed)) = randomness_enabler_messages.last() {
4
                    let value = client
4
                        .get_aux(RANDOMNESS_ACTIVATED_AUX_KEY)
4
                        .expect("Should be able to query aux storage; qed").unwrap_or((false, Option::<[u8; 32]>::None).encode());
4
                    let (_mock_additional_randomness, mut mock_randomness_seed): (bool, Option<[u8; 32]>) = Decode::decode(&mut value.as_slice()).expect("Boolean non-decodable");
4
                    if let Some(new_seed) = new_seed {
2
                        mock_randomness_seed = Some(*new_seed);
2
                    }
4
                    client
4
                    .insert_aux(
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");
7806
                }
                // We read the value
                // If error when reading, we simply put false
7810
                let value = client
7810
                    .get_aux(RANDOMNESS_ACTIVATED_AUX_KEY)
7810
                    .expect("Should be able to query aux storage; qed").unwrap_or((false, Option::<[u8; 32]>::None).encode());
7810
                let (mock_additional_randomness, mock_randomness_seed): (bool, Option<[u8; 32]>) = Decode::decode(&mut value.as_slice()).expect("Boolean non-decodable");
7810

            
7810
                let container_chains_exclusion_messages: Vec<Vec<ParaId>> = mock_container_chains_exclusion_receiver.drain().collect();
                // If there is a new set of excluded container chains, we update it
7810
                if let Some(mock_excluded_container_chains) = container_chains_exclusion_messages.last() {
2
                    client
2
                        .insert_aux(
2
                            &[(CONTAINER_CHAINS_EXCLUSION_AUX_KEY, mock_excluded_container_chains.encode().as_slice())],
2
                            &[],
2
                        )
2
                        .expect("Should be able to write to aux storage; qed");
7808
                }
7810
                let new_excluded_container_chains_value = client
7810
                    .get_aux(CONTAINER_CHAINS_EXCLUSION_AUX_KEY)
7810
                    .expect("Should be able to query aux storage; qed").unwrap_or(Vec::<ParaId>::new().encode());
7810
                let mock_excluded_container_chains: Vec<ParaId> = Decode::decode(&mut new_excluded_container_chains_value.as_slice()).expect("Vector non-decodable");
15394
                para_ids.retain(|x| !mock_excluded_container_chains.contains(x));
7810
                let client_set_aside_for_cidp = client.clone();
7810
                let client_for_xcm = client.clone();
7810
                async move {
7810
                    let mocked_author_noting =
7810
                        tp_author_noting_inherent::MockAuthorNotingInherentDataProvider {
7810
                            current_para_block,
7810
                            relay_offset: 1000,
7810
                            relay_blocks_per_para_block: 2,
7810
                            para_ids,
7810
                            slots_per_para_block: 1,
7810
                        };
7810
                    let mut additional_keys = mocked_author_noting.get_key_values();
7810
                    // Mock only chain 2002 in relay.
7810
                    // This will allow any signed origin to deregister chains 2000 and 2001, and register 2002.
7810
                    let (registrar_paras_key_2002, para_info_2002) = mocked_relay_keys::get_mocked_registrar_paras(2002.into());
7810
                    additional_keys.extend([(para_head_key, para_head_data), (relay_slot_key, Slot::from(relay_slot).encode()), (registrar_paras_key_2002, para_info_2002)]);
7810

            
7810
                    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);
7610
                    }
7810
                    let current_para_head = client_set_aside_for_cidp
7810
                            .header(block)
7810
                            .expect("Header lookup should succeed")
7810
                            .expect("Header passed in as parent should be present in backend.");
7810
                    let should_send_go_ahead = match client_set_aside_for_cidp
7810
                            .runtime_api()
7810
                            .collect_collation_info(block, &current_para_head)
                    {
7810
                            Ok(info) => info.new_validation_code.is_some(),
                            Err(e) => {
                                    log::error!("Failed to collect collation info: {:?}", e);
                                    false
                            },
                    };
7810
                    let time = MockTimestampInherentDataProvider;
7810
                    let mocked_parachain = MockValidationDataInherentDataProvider {
7810
                        current_para_block,
7810
                        current_para_block_head: None,
7810
                        relay_offset: 1000,
7810
                        relay_blocks_per_para_block: 2,
7810
                        para_blocks_per_relay_epoch: 10,
7810
                        relay_randomness_config: (),
7810
                        xcm_config: MockXcmConfig::new(
7810
                            &*client_for_xcm,
7810
                            block,
7810
                            Default::default(),
7810
                        ),
7810
                        raw_downward_messages: downward_xcm_receiver.drain().collect(),
7810
                        raw_horizontal_messages: hrmp_xcm_receiver.drain().collect(),
7810
                        additional_key_values: Some(additional_keys),
7810
                        para_id,
7810
                        upgrade_go_ahead: should_send_go_ahead.then(|| {
2
                            log::info!(
2
                                "Detected pending validation code, sending go-ahead signal."
                            );
2
                            UpgradeGoAhead::GoAhead
7810
                        }),
7810
                    };
7810

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

            
388
            crate::rpc::create_full(deps).map_err(Into::into)
388
        })
    };
    // 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");
    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> {
194
    fn is_dev(&self) -> bool {
194
        self.chain_type() == sc_chain_spec::ChainType::Development
194
    }
}
/// 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)
    }
}