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

            
94
mod mocked_relay_keys;
95

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

            
99
const CONTAINER_CHAINS_EXCLUSION_AUX_KEY: &[u8] = b"__DEV_CONTAINER_CHAINS_EXCLUSION";
100

            
101
type FullBackend = TFullBackend<Block>;
102

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

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

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

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

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

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

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

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

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

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

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

            
213
    Ok(())
214
}
215

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

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

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

            
241
    (block_import, import_queue)
242
}
243

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

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

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

            
268
    let (block_import, import_queue) = import_queue(&parachain_config, &node_builder);
269

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

            
274
    let validator = parachain_config.role.is_authority();
275
    let force_authoring = parachain_config.force_authoring;
276

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

            
286
    let rpc_builder = {
287
        let client = node_builder.client.clone();
288
        let transaction_pool = node_builder.transaction_pool.clone();
289

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

            
300
            crate::rpc::create_full(deps).map_err(Into::into)
301
        })
302
    };
303

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

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

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

            
326
    let (mut node_builder, import_queue_service) = node_builder.extract_import_queue_service();
327

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

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

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

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

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

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

            
414
    let sync_keystore = node_builder.keystore_container.keystore();
415

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

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

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

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

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

            
490
    Ok((node_builder.task_manager, node_builder.client))
491
}
492

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

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

            
532
    let proposer = Proposer::new(proposer_factory);
533

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
679
    (cancellation_token, exit_notification_receiver)
680
}
681

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

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

            
718
    let chain_type = polkadot_config.chain_spec.chain_type().clone();
719
    let relay_chain = polkadot_config.chain_spec.id().to_string();
720

            
721
    // We use the relaychain keystore config for collators
722
    // Ensure that the user did not provide any custom keystore path for collators
723
    if container_chain_cli
724
        .base
725
        .base
726
        .keystore_params
727
        .keystore_path
728
        .is_some()
729
    {
730
        panic!(
731
            "--keystore-path not allowed here, must be set in relaychain args, after the first --"
732
        )
733
    }
734
    let keystore = &polkadot_config.keystore;
735

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

            
745
    let keystore_container = KeystoreContainer::new(keystore)?;
746

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

            
751
    // Each container chain will spawn its own telemetry
752
    let telemetry_worker_handle = None;
753

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

            
778
    log::info!("start_solochain_node: is validator? {}", validator);
779

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

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

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

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

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

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

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

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

            
878
    Ok(task_manager)
879
}
880

            
881
pub const SOFT_DEADLINE_PERCENT: sp_runtime::Percent = sp_runtime::Percent::from_percent(100);
882

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

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

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

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

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

            
936
        xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));
937
        randomness_sender = Some(mock_randomness_sender);
938
        container_chains_exclusion_sender = Some(mock_container_chains_exclusion_sender);
939

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

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

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

            
970
7890
                let para_header = client
971
7890
                    .expect_header(hash)
972
7890
                    .expect("Expected parachain header should exist")
973
7890
                    .encode();
974
7890

            
975
7890
                let para_head_data = HeadData(para_header).encode();
976
7890
                let para_head_key = RelayWellKnownKeys::para_head(para_id);
977
7890
                let relay_slot_key = RelayWellKnownKeys::CURRENT_SLOT.to_vec();
978
7890

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

            
984
7890
                let mut timestamp = 0u64;
985
7890
                TIMESTAMP.with(|x| {
986
7890
                    timestamp = x.clone().take();
987
7890
                });
988
7890

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

            
996
7890
                let downward_xcm_receiver = downward_xcm_receiver.clone();
997
7890
                let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
998
7890

            
999
7890
                let randomness_enabler_messages: Vec<(bool, Option<[u8; 32]>)> = mock_randomness_receiver.drain().collect();
                // If there is a value to be updated, we update it
7890
                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");
7886
                }
                // We read the value
                // If error when reading, we simply put false
7890
                let value = client
7890
                    .get_aux(RANDOMNESS_ACTIVATED_AUX_KEY)
7890
                    .expect("Should be able to query aux storage; qed").unwrap_or((false, Option::<[u8; 32]>::None).encode());
7890
                let (mock_additional_randomness, mock_randomness_seed): (bool, Option<[u8; 32]>) = Decode::decode(&mut value.as_slice()).expect("Boolean non-decodable");
7890

            
7890
                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
7890
                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");
7888
                }
7890
                let new_excluded_container_chains_value = client
7890
                    .get_aux(CONTAINER_CHAINS_EXCLUSION_AUX_KEY)
7890
                    .expect("Should be able to query aux storage; qed").unwrap_or(Vec::<ParaId>::new().encode());
7890
                let mock_excluded_container_chains: Vec<ParaId> = Decode::decode(&mut new_excluded_container_chains_value.as_slice()).expect("Vector non-decodable");
15554
                para_ids.retain(|x| !mock_excluded_container_chains.contains(x));
7890
                let client_set_aside_for_cidp = client.clone();
7890
                let client_for_xcm = client.clone();
7890
                async move {
7890
                    let mocked_author_noting =
7890
                        tp_author_noting_inherent::MockAuthorNotingInherentDataProvider {
7890
                            current_para_block,
7890
                            relay_offset: 1000,
7890
                            relay_blocks_per_para_block: 2,
7890
                            para_ids,
7890
                            slots_per_para_block: 1,
7890
                        };
7890
                    let mut additional_keys = mocked_author_noting.get_key_values();
7890
                    // Mock only chain 2002 in relay.
7890
                    // This will allow any signed origin to deregister chains 2000 and 2001, and register 2002.
7890
                    let (registrar_paras_key_2002, para_info_2002) = mocked_relay_keys::get_mocked_registrar_paras(2002.into());
7890
                    additional_keys.extend([(para_head_key, para_head_data), (relay_slot_key, Slot::from(relay_slot).encode()), (registrar_paras_key_2002, para_info_2002)]);
7890

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

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

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