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
//! Development Polkadot service. Adapted from `polkadot_service` crate
18
//! and removed un-necessary components which are not required in dev node.
19
//!
20
//! Following major changes are made:
21
//! 1. Removed beefy and grandpa notification service and request response protocols
22
//! 2. Removed support for parachains which also eliminated the need to start overseer and all other subsystems associated with collation + network request/response protocols for the same
23
//! 3. Removed support for hardware benchmarking
24
//! 4. Removed authority discovery service
25
//! 5. Removed spawning of beefy, grandpa and MMR worker
26
//! 6. Removed rpc extensions for beefy, grandpa and babe and added support for manual seal
27
//! 7. Removed beefy and grandpa block import from block import pipeline (Babe remains)
28
//! 8. Using manual seal import queue instead of babe import queue
29
//! 9. Started manual seal worker
30
//! 10. If amount of time passed between two block is less than slot duration, we emulate passing of time babe block import and runtime
31
//!     by incrementing timestamp by slot duration.
32

            
33
use {
34
    crate::dev_rpcs::{DevApiServer, DevRpc},
35
    async_io::Timer,
36
    babe::{BabeBlockImport, BabeLink},
37
    codec::{Decode, Encode},
38
    consensus_common::SelectChain,
39
    dancelight_runtime::RuntimeApi,
40
    futures::{Stream, StreamExt},
41
    jsonrpsee::RpcModule,
42
    node_common::service::Sealing,
43
    polkadot_core_primitives::{AccountId, Balance, Block, Hash, Nonce},
44
    polkadot_node_core_parachains_inherent::Error as InherentError,
45
    polkadot_overseer::Handle,
46
    polkadot_parachain_primitives::primitives::UpwardMessages,
47
    polkadot_primitives::{
48
        runtime_api::ParachainHost, BackedCandidate, CandidateCommitments, CandidateDescriptor,
49
        CollatorPair, CommittedCandidateReceipt, CompactStatement, EncodeAs,
50
        InherentData as ParachainsInherentData, OccupiedCoreAssumption, SigningContext,
51
        ValidityAttestation,
52
    },
53
    polkadot_rpc::RpcExtension,
54
    polkadot_service::{
55
        BlockT, Error, IdentifyVariant, NewFullParams, OverseerGen, SelectRelayChain,
56
    },
57
    sc_client_api::{AuxStore, Backend},
58
    sc_consensus_manual_seal::{
59
        consensus::babe::BabeConsensusDataProvider,
60
        rpc::{ManualSeal, ManualSealApiServer},
61
        run_manual_seal, EngineCommand, ManualSealParams,
62
    },
63
    sc_executor::{HeapAllocStrategy, WasmExecutor, DEFAULT_HEAP_ALLOC_STRATEGY},
64
    sc_keystore::Keystore,
65
    sc_transaction_pool_api::{OffchainTransactionPoolFactory, TransactionPool},
66
    service::{Configuration, KeystoreContainer, RpcHandlers, TaskManager},
67
    sp_api::ProvideRuntimeApi,
68
    sp_block_builder::BlockBuilder,
69
    sp_blockchain::{HeaderBackend, HeaderMetadata},
70
    sp_consensus_aura::{inherents::InherentType as AuraInherentType, AURA_ENGINE_ID},
71
    sp_consensus_babe::SlotDuration,
72
    sp_core::{ByteArray, Pair, H256},
73
    sp_keystore::KeystorePtr,
74
    sp_runtime::{traits::BlakeTwo256, DigestItem, RuntimeAppPublic},
75
    std::{cmp::max, ops::Add, sync::Arc, time::Duration},
76
    telemetry::{Telemetry, TelemetryWorker, TelemetryWorkerHandle},
77
};
78

            
79
// We use this key to store whether we want the para inherent mocker to be active
80
const PARA_INHERENT_SELECTOR_AUX_KEY: &[u8] = b"__DEV_PARA_INHERENT_SELECTOR";
81

            
82
pub type FullBackend = service::TFullBackend<Block>;
83

            
84
pub type FullClient = service::TFullClient<
85
    Block,
86
    RuntimeApi,
87
    WasmExecutor<(
88
        sp_io::SubstrateHostFunctions,
89
        frame_benchmarking::benchmarking::HostFunctions,
90
    )>,
91
>;
92

            
93
pub struct NewFull {
94
    pub task_manager: TaskManager,
95
    pub client: Arc<FullClient>,
96
    pub overseer_handle: Option<Handle>,
97
    pub network: Arc<dyn sc_network::service::traits::NetworkService>,
98
    pub sync_service: Arc<sc_network_sync::SyncingService<Block>>,
99
    pub rpc_handlers: RpcHandlers,
100
    pub backend: Arc<FullBackend>,
101
}
102

            
103
/// Custom Deps for dev Rpc extension
104
struct DevDeps<C, P> {
105
    /// The client instance to use.
106
    pub client: Arc<C>,
107
    /// Transaction pool instance.
108
    pub pool: Arc<P>,
109
    /// Manual seal command sink
110
    pub command_sink: Option<futures::channel::mpsc::Sender<EngineCommand<Hash>>>,
111
    /// Dev rpcs
112
    pub dev_rpc: Option<DevRpc>,
113
}
114

            
115
980
fn create_dev_rpc_extension<C, P>(
116
980
    DevDeps {
117
980
        client,
118
980
        pool,
119
980
        command_sink: maybe_command_sink,
120
980
        dev_rpc: maybe_dev_rpc,
121
980
    }: DevDeps<C, P>,
122
980
) -> Result<RpcExtension, Box<dyn std::error::Error + Send + Sync>>
123
980
where
124
980
    C: ProvideRuntimeApi<Block>
125
980
        + HeaderBackend<Block>
126
980
        + AuxStore
127
980
        + HeaderMetadata<Block, Error = sp_blockchain::Error>
128
980
        + Send
129
980
        + Sync
130
980
        + 'static,
131
980
    C::Api: substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>,
132
980
    C::Api: pallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>,
133
980
    C::Api: BlockBuilder<Block>,
134
980
    P: TransactionPool + Sync + Send + 'static,
135
980
{
136
    use {
137
        pallet_transaction_payment_rpc::{TransactionPayment, TransactionPaymentApiServer},
138
        substrate_frame_rpc_system::{System, SystemApiServer},
139
    };
140

            
141
980
    let mut io = RpcModule::new(());
142
980
    io.merge(System::new(client.clone(), pool.clone()).into_rpc())?;
143
980
    io.merge(TransactionPayment::new(client.clone()).into_rpc())?;
144

            
145
980
    if let Some(command_sink) = maybe_command_sink {
146
980
        io.merge(ManualSeal::new(command_sink).into_rpc())?;
147
    }
148

            
149
980
    if let Some(dev_rpc_data) = maybe_dev_rpc {
150
980
        io.merge(dev_rpc_data.into_rpc())?;
151
    }
152

            
153
980
    Ok(io)
154
980
}
155

            
156
/// We use EmptyParachainsInherentDataProvider to insert an empty parachain inherent in the block
157
/// to satisfy runtime
158
struct EmptyParachainsInherentDataProvider;
159

            
160
/// Copied from polkadot service just so that this code retains same structure as
161
/// polkadot_service crate.
162
struct Basics {
163
    task_manager: TaskManager,
164
    client: Arc<FullClient>,
165
    backend: Arc<FullBackend>,
166
    keystore_container: KeystoreContainer,
167
    telemetry: Option<Telemetry>,
168
}
169

            
170
impl EmptyParachainsInherentDataProvider {
171
32487
    pub async fn create<C: HeaderBackend<Block>>(
172
32487
        client: Arc<C>,
173
32487
        parent: Hash,
174
32487
    ) -> Result<ParachainsInherentData, InherentError> {
175
32487
        let parent_header = match client.header(parent) {
176
32487
            Ok(Some(h)) => h,
177
            Ok(None) => return Err(InherentError::ParentHeaderNotFound(parent)),
178
            Err(err) => return Err(InherentError::Blockchain(err)),
179
        };
180

            
181
32487
        Ok(ParachainsInherentData {
182
32487
            bitfields: Vec::new(),
183
32487
            backed_candidates: Vec::new(),
184
32487
            disputes: Vec::new(),
185
32487
            parent_header,
186
32487
        })
187
32487
    }
188
}
189

            
190
/// Creates new development full node with manual seal
191
490
pub fn build_full<OverseerGenerator: OverseerGen>(
192
490
    sealing: Sealing,
193
490
    config: Configuration,
194
490
    mut params: NewFullParams<OverseerGenerator>,
195
490
) -> Result<NewFull, Error> {
196
490
    let is_polkadot = config.chain_spec.is_polkadot();
197
490

            
198
490
    params.overseer_message_channel_capacity_override = params
199
490
        .overseer_message_channel_capacity_override
200
490
        .map(move |capacity| {
201
            if is_polkadot {
202
                gum::warn!("Channel capacity should _never_ be tampered with on polkadot!");
203
            }
204
            capacity
205
490
        });
206
490

            
207
490
    match config.network.network_backend {
208
        sc_network::config::NetworkBackendType::Libp2p => {
209
490
            new_full::<_, sc_network::NetworkWorker<Block, Hash>>(sealing, config, params)
210
        }
211
        sc_network::config::NetworkBackendType::Litep2p => {
212
            new_full::<_, sc_network::Litep2pNetworkBackend>(sealing, config, params)
213
        }
214
    }
215
490
}
216

            
217
/// We use MockParachainsInherentDataProvider to insert an parachain inherent with mocked
218
/// candidates
219
/// We detect whether any of the keys in our keystore is assigned to a core and provide
220
/// a mocked candidate in such core
221
struct MockParachainsInherentDataProvider<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> {
222
    pub client: Arc<C>,
223
    pub parent: Hash,
224
    pub keystore: KeystorePtr,
225
    pub upward_messages_receiver: flume::Receiver<Vec<u8>>,
226
}
227

            
228
impl<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> MockParachainsInherentDataProvider<C>
229
where
230
    C::Api: ParachainHost<Block>,
231
    C: AuxStore,
232
{
233
41608
    pub fn new(
234
41608
        client: Arc<C>,
235
41608
        parent: Hash,
236
41608
        keystore: KeystorePtr,
237
41608
        upward_messages_receiver: flume::Receiver<Vec<u8>>,
238
41608
    ) -> Self {
239
41608
        MockParachainsInherentDataProvider {
240
41608
            client,
241
41608
            parent,
242
41608
            keystore,
243
41608
            upward_messages_receiver,
244
41608
        }
245
41608
    }
246

            
247
9121
    pub async fn create(
248
9121
        client: Arc<C>,
249
9121
        parent: Hash,
250
9121
        keystore: KeystorePtr,
251
9121
        upward_messages_receiver: flume::Receiver<Vec<u8>>,
252
9121
    ) -> Result<ParachainsInherentData, InherentError> {
253
9121
        let parent_header = match client.header(parent) {
254
9121
            Ok(Some(h)) => h,
255
            Ok(None) => return Err(InherentError::ParentHeaderNotFound(parent)),
256
            Err(err) => return Err(InherentError::Blockchain(err)),
257
        };
258

            
259
        // Strategy:
260
        // we usually have 1 validator per core, and we usually run with --alice
261
        // the idea is that at least alice will be assigned to one core
262
        // if we find in the keystore the validator attached to a particular core,
263
        // we generate a signature for the parachain assigned to that core
264
        // To retrieve the validator keys, cal runtime api:
265

            
266
        // this following piece of code predicts whether the validator is assigned to a particular
267
        // core where a candidate for a parachain needs to be created
268
9121
        let runtime_api = client.runtime_api();
269
9121

            
270
9121
        // we get all validators
271
9121

            
272
9121
        // we get the current claim queue to know core availability
273
9121
        let claim_queue = runtime_api.claim_queue(parent).unwrap();
274
9121

            
275
9121
        // we get the validator groups
276
9121
        let (groups, rotation_info) = runtime_api.validator_groups(parent).unwrap();
277
9121

            
278
9121
        // we calculate rotation since start, which will define the core assignation
279
9121
        // to validators
280
9121
        let rotations_since_session_start = (parent_header.number
281
9121
            - rotation_info.session_start_block)
282
9121
            / rotation_info.group_rotation_frequency;
283
9121

            
284
9121
        // Get all the available keys in the keystore
285
9121
        let available_keys = keystore
286
9121
            .keys(polkadot_primitives::PARACHAIN_KEY_TYPE_ID)
287
9121
            .unwrap();
288
9121

            
289
9121
        // create a slot number identical to the parent block num
290
9121
        let slot_number = AuraInherentType::from(u64::from(parent_header.number));
291
9121

            
292
9121
        // create a mocked header
293
9121
        let parachain_mocked_header = sp_runtime::generic::Header::<u32, BlakeTwo256> {
294
9121
            parent_hash: Default::default(),
295
9121
            number: parent_header.number,
296
9121
            state_root: Default::default(),
297
9121
            extrinsics_root: Default::default(),
298
9121
            digest: sp_runtime::generic::Digest {
299
9121
                logs: vec![DigestItem::PreRuntime(AURA_ENGINE_ID, slot_number.encode())],
300
9121
            },
301
9121
        };
302
9121

            
303
9121
        // retrieve availability cores
304
9121
        let availability_cores = runtime_api.availability_cores(parent).unwrap();
305
9121

            
306
9121
        // retrieve current session_idx
307
9121
        let session_idx = runtime_api.session_index_for_child(parent).unwrap();
308
9121

            
309
9121
        // retrieve all validators
310
9121
        let all_validators = runtime_api.validators(parent).unwrap();
311
9121

            
312
9121
        // construct full availability bitvec
313
9121
        let availability_bitvec = availability_bitvec(1, availability_cores.len());
314
9121

            
315
9121
        let signature_ctx = SigningContext {
316
9121
            parent_hash: parent,
317
9121
            session_index: session_idx,
318
9121
        };
319
9121

            
320
9121
        // we generate the availability bitfield sigs
321
9121
        // TODO: here we assume all validator keys are able to sign with our keystore
322
9121
        // we need to make sure the key is there before we try to sign
323
9121
        // this is mostly to indicate that the erasure coding chunks where received by all val
324
9121
        let bitfields: Vec<UncheckedSigned<AvailabilityBitfield>> = all_validators
325
9121
            .iter()
326
9121
            .enumerate()
327
9121
            .map(|(i, public)| {
328
9121
                keystore_sign(
329
9121
                    &keystore,
330
9121
                    availability_bitvec.clone(),
331
9121
                    &signature_ctx,
332
9121
                    ValidatorIndex(i as u32),
333
9121
                    &public,
334
9121
                )
335
9121
                .unwrap()
336
9121
                .unwrap()
337
9121
            })
338
9121
            .collect();
339
9121

            
340
9121
        // generate a random collator pair
341
9121
        let collator_pair = CollatorPair::generate().0;
342
9121
        let mut backed_cand: Vec<BackedCandidate<H256>> = vec![];
343

            
344
        // iterate over every core|para pair
345
25501
        for (core, para) in claim_queue {
346
            // check which group is assigned to each core
347
16380
            let group_assigned_to_core =
348
16380
                core.0 + rotations_since_session_start % groups.len() as u32;
349
16380
            // check validator indices associated to the core
350
16380
            let indices_associated_to_core = groups.get(group_assigned_to_core as usize).unwrap();
351
24668
            for index in indices_associated_to_core {
352
                // fetch validator keys
353
8288
                let validator_keys_to_find = all_validators.get(index.0 as usize).unwrap();
354
                // Iterate keys until we find an eligible one, or run out of candidates.
355
16576
                for type_public_pair in &available_keys {
356
8288
                    if let Ok(validator) =
357
8288
                        polkadot_primitives::ValidatorId::from_slice(&type_public_pair)
358
                    {
359
                        // if we find the validator in keystore, we try to create a backed cand
360
8288
                        if validator_keys_to_find == &validator {
361
8288
                            // we work with the previous included data
362
8288
                            let mut persisted_validation_data = runtime_api
363
8288
                                .persisted_validation_data(
364
8288
                                    parent,
365
8288
                                    para[0],
366
8288
                                    OccupiedCoreAssumption::Included,
367
8288
                                )
368
8288
                                .unwrap()
369
8288
                                .unwrap();
370
8288

            
371
8288
                            // if we dont do this we have a backed candidate every 2 blocks
372
8288
                            // TODO: figure out why
373
8288
                            persisted_validation_data.relay_parent_storage_root =
374
8288
                                parent_header.state_root;
375
8288

            
376
8288
                            let persisted_validation_data_hash = persisted_validation_data.hash();
377
8288
                            // retrieve the validation code hash
378
8288
                            let validation_code_hash = runtime_api
379
8288
                                .validation_code_hash(
380
8288
                                    parent,
381
8288
                                    para[0],
382
8288
                                    OccupiedCoreAssumption::Included,
383
8288
                                )
384
8288
                                .unwrap()
385
8288
                                .unwrap();
386
8288
                            let pov_hash = Default::default();
387
8288
                            // generate a fake collator signature
388
8288
                            let payload = polkadot_primitives::collator_signature_payload(
389
8288
                                &parent,
390
8288
                                &para[0],
391
8288
                                &persisted_validation_data_hash,
392
8288
                                &pov_hash,
393
8288
                                &validation_code_hash,
394
8288
                            );
395
8288
                            let collator_signature = collator_pair.sign(&payload);
396
8288

            
397
8288
                            let upward_messages = UpwardMessages::try_from(
398
8288
                                upward_messages_receiver.drain().collect::<Vec<_>>(),
399
8288
                            )
400
8288
                            .expect("create upward messages from raw messages");
401
8288

            
402
8288
                            // generate a candidate with most of the values mocked
403
8288
                            let candidate = CommittedCandidateReceipt::<H256> {
404
8288
                                descriptor: CandidateDescriptor::<H256> {
405
8288
                                    para_id: para[0],
406
8288
                                    relay_parent: parent,
407
8288
                                    collator: collator_pair.public(),
408
8288
                                    persisted_validation_data_hash,
409
8288
                                    pov_hash,
410
8288
                                    erasure_root: Default::default(),
411
8288
                                    signature: collator_signature,
412
8288
                                    para_head: parachain_mocked_header.clone().hash(),
413
8288
                                    validation_code_hash,
414
8288
                                },
415
8288
                                commitments: CandidateCommitments::<u32> {
416
8288
                                    upward_messages,
417
8288
                                    horizontal_messages: Default::default(),
418
8288
                                    new_validation_code: None,
419
8288
                                    head_data: parachain_mocked_header.clone().encode().into(),
420
8288
                                    processed_downward_messages: 0,
421
8288
                                    hrmp_watermark: parent_header.number,
422
8288
                                },
423
8288
                            };
424
8288
                            let candidate_hash = candidate.hash();
425
8288
                            let payload = CompactStatement::Valid(candidate_hash);
426
8288

            
427
8288
                            let signature_ctx = SigningContext {
428
8288
                                parent_hash: parent,
429
8288
                                session_index: session_idx,
430
8288
                            };
431
8288

            
432
8288
                            // sign the candidate with the validator key
433
8288
                            let signature = keystore_sign(
434
8288
                                &keystore,
435
8288
                                payload,
436
8288
                                &signature_ctx,
437
8288
                                *index,
438
8288
                                &validator,
439
8288
                            )
440
8288
                            .unwrap()
441
8288
                            .unwrap()
442
8288
                            .benchmark_signature();
443
8288

            
444
8288
                            // construct a validity vote
445
8288
                            let validity_votes = vec![ValidityAttestation::Explicit(signature)];
446
8288

            
447
8288
                            // push the candidate
448
8288
                            backed_cand.push(BackedCandidate::<H256>::new(
449
8288
                                candidate,
450
8288
                                validity_votes.clone(),
451
8288
                                bitvec::bitvec![u8, bitvec::order::Lsb0; 1; indices_associated_to_core.len()],
452
8288
                                Some(core),
453
8288
                            ));
454
8288
                        }
455
                    }
456
                }
457
            }
458
        }
459

            
460
9121
        Ok(ParachainsInherentData {
461
9121
            bitfields: bitfields,
462
9121
            backed_candidates: backed_cand,
463
9121
            disputes: Vec::new(),
464
9121
            parent_header,
465
9121
        })
466
9121
    }
467
}
468

            
469
#[async_trait::async_trait]
470
impl<C: HeaderBackend<Block> + ProvideRuntimeApi<Block>> sp_inherents::InherentDataProvider
471
    for MockParachainsInherentDataProvider<C>
472
where
473
    C::Api: ParachainHost<Block>,
474
    C: AuxStore,
475
{
476
    async fn provide_inherent_data(
477
        &self,
478
        dst_inherent_data: &mut sp_inherents::InherentData,
479
41608
    ) -> Result<(), sp_inherents::Error> {
480
        // fetch whether the para inherent selector has been set
481
41608
        let maybe_para_selector = self
482
41608
            .client
483
41608
            .get_aux(PARA_INHERENT_SELECTOR_AUX_KEY)
484
41608
            .expect("Should be able to query aux storage; qed");
485

            
486
41608
        let inherent_data = {
487
41608
            if let Some(aux) = maybe_para_selector {
488
                // if it is true, the candidates need to be mocked
489
                // else, we output the empty parachain inherent data provider
490
9121
                if aux == true.encode() {
491
9121
                    MockParachainsInherentDataProvider::create(
492
9121
                        self.client.clone(),
493
9121
                        self.parent,
494
9121
                        self.keystore.clone(),
495
9121
                        self.upward_messages_receiver.clone(),
496
9121
                    )
497
                    .await
498
9121
                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?
499
                } else {
500
                    EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)
501
                        .await
502
                        .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?
503
                }
504
            } else {
505
32487
                EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)
506
                    .await
507
32487
                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?
508
            }
509
        };
510

            
511
41608
        dst_inherent_data.put_data(
512
41608
            polkadot_primitives::PARACHAINS_INHERENT_IDENTIFIER,
513
41608
            &inherent_data,
514
41608
        )
515
83216
    }
516

            
517
    async fn try_handle_error(
518
        &self,
519
        _identifier: &sp_inherents::InherentIdentifier,
520
        _error: &[u8],
521
    ) -> Option<Result<(), sp_inherents::Error>> {
522
        // Inherent isn't checked and can not return any error
523
        None
524
    }
525
}
526

            
527
/// We store past timestamp we created in the aux storage, which enable us to return timestamp which is increased by
528
/// slot duration from previous timestamp or current timestamp if in reality more time is passed.
529
41608
fn get_next_timestamp(
530
41608
    client: Arc<FullClient>,
531
41608
    slot_duration: SlotDuration,
532
41608
) -> sp_timestamp::InherentDataProvider {
533
    const TIMESTAMP_AUX_KEY: &[u8] = b"__DEV_TIMESTAMP";
534

            
535
41608
    let maybe_last_timestamp = client
536
41608
        .get_aux(TIMESTAMP_AUX_KEY)
537
41608
        .expect("Should be able to query aux storage; qed");
538
41608
    if let Some(last_timestamp) = maybe_last_timestamp {
539
41160
        let last_inherent_data = sp_timestamp::InherentType::decode(&mut last_timestamp.as_slice())
540
41160
            .expect("Timestamp data must be decoded; qed");
541
41160
        let new_inherent_data: sp_timestamp::InherentType = max(
542
41160
            last_inherent_data.add(slot_duration.as_millis()),
543
41160
            sp_timestamp::InherentType::current(),
544
41160
        );
545
41160
        client
546
41160
            .insert_aux(
547
41160
                &[(TIMESTAMP_AUX_KEY, new_inherent_data.encode().as_slice())],
548
41160
                &[],
549
41160
            )
550
41160
            .expect("Should be able to write to aux storage; qed");
551
41160
        sp_timestamp::InherentDataProvider::new(new_inherent_data)
552
    } else {
553
448
        let current_timestamp = sp_timestamp::InherentType::current();
554
448
        client
555
448
            .insert_aux(
556
448
                &[(TIMESTAMP_AUX_KEY, current_timestamp.encode().as_slice())],
557
448
                &[],
558
448
            )
559
448
            .expect("Should be able to write to aux storage; qed");
560
448
        sp_timestamp::InherentDataProvider::new(current_timestamp)
561
    }
562
41608
}
563

            
564
490
fn new_full<
565
490
    OverseerGenerator: OverseerGen,
566
490
    Network: sc_network::NetworkBackend<Block, <Block as BlockT>::Hash>,
567
490
>(
568
490
    sealing: Sealing,
569
490
    mut config: Configuration,
570
490
    NewFullParams {
571
490
        telemetry_worker_handle,
572
490
        ..
573
490
    }: NewFullParams<OverseerGenerator>,
574
490
) -> Result<NewFull, Error> {
575
490
    let role = config.role;
576

            
577
490
    let basics = new_partial_basics(&mut config, telemetry_worker_handle)?;
578

            
579
490
    let prometheus_registry = config.prometheus_registry().cloned();
580
490

            
581
490
    let keystore = basics.keystore_container.local_keystore();
582
490

            
583
490
    let select_chain = SelectRelayChain::new_longest_chain(basics.backend.clone());
584

            
585
490
    let service::PartialComponents::<_, _, SelectRelayChain<_>, _, _, _> {
586
490
        client,
587
490
        backend,
588
490
        mut task_manager,
589
490
        keystore_container,
590
490
        select_chain,
591
490
        import_queue,
592
490
        transaction_pool,
593
490
        other: (block_import, babe_link, slot_duration, mut telemetry),
594
490
    } = new_partial::<SelectRelayChain<_>>(&mut config, basics, select_chain)?;
595

            
596
490
    let metrics = Network::register_notification_metrics(
597
490
        config.prometheus_config.as_ref().map(|cfg| &cfg.registry),
598
490
    );
599
490

            
600
490
    let net_config = sc_network::config::FullNetworkConfiguration::<_, _, Network>::new(
601
490
        &config.network,
602
490
        prometheus_registry.clone(),
603
490
    );
604
490

            
605
490
    // Create channels for mocked parachain candidates.
606
490
    let (downward_mock_para_inherent_sender, downward_mock_para_inherent_receiver) =
607
490
        flume::bounded::<Vec<u8>>(100);
608
490

            
609
490
    let (upward_mock_sender, upward_mock_receiver) = flume::bounded::<Vec<u8>>(100);
610

            
611
490
    let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =
612
490
        service::build_network(service::BuildNetworkParams {
613
490
            config: &config,
614
490
            net_config,
615
490
            client: client.clone(),
616
490
            transaction_pool: transaction_pool.clone(),
617
490
            spawn_handle: task_manager.spawn_handle(),
618
490
            import_queue,
619
490
            block_announce_validator_builder: None,
620
490
            warp_sync_config: None,
621
490
            block_relay: None,
622
490
            metrics,
623
490
        })?;
624

            
625
490
    if config.offchain_worker.enabled {
626
490
        use futures::FutureExt;
627
490

            
628
490
        task_manager.spawn_handle().spawn(
629
490
            "offchain-workers-runner",
630
490
            "offchain-work",
631
490
            sc_offchain::OffchainWorkers::new(sc_offchain::OffchainWorkerOptions {
632
490
                runtime_api_provider: client.clone(),
633
490
                keystore: Some(keystore_container.keystore()),
634
490
                offchain_db: backend.offchain_storage(),
635
490
                transaction_pool: Some(OffchainTransactionPoolFactory::new(
636
490
                    transaction_pool.clone(),
637
490
                )),
638
490
                network_provider: Arc::new(network.clone()),
639
490
                is_validator: role.is_authority(),
640
490
                enable_http_requests: false,
641
41608
                custom_extensions: move |_| vec![],
642
490
            })
643
490
            .run(client.clone(), task_manager.spawn_handle())
644
490
            .boxed(),
645
490
        );
646
490
    }
647

            
648
490
    let mut command_sink = None;
649
490

            
650
490
    if role.is_authority() {
651
490
        let proposer = sc_basic_authorship::ProposerFactory::with_proof_recording(
652
490
            task_manager.spawn_handle(),
653
490
            client.clone(),
654
490
            transaction_pool.clone(),
655
490
            prometheus_registry.as_ref(),
656
490
            telemetry.as_ref().map(|x| x.handle()),
657
490
        );
658

            
659
490
        let commands_stream: Box<
660
490
            dyn Stream<Item = EngineCommand<<Block as BlockT>::Hash>> + Send + Sync + Unpin,
661
490
        > = match sealing {
662
            Sealing::Instant => {
663
                Box::new(
664
                    // This bit cribbed from the implementation of instant seal.
665
                    transaction_pool.import_notification_stream().map(|_| {
666
                        EngineCommand::SealNewBlock {
667
                            create_empty: false,
668
                            finalize: false,
669
                            parent_hash: None,
670
                            sender: None,
671
                        }
672
                    }),
673
                )
674
            }
675
            Sealing::Manual => {
676
490
                let (sink, stream) = futures::channel::mpsc::channel(1000);
677
490
                // Keep a reference to the other end of the channel. It goes to the RPC.
678
490
                command_sink = Some(sink);
679
490
                Box::new(stream)
680
            }
681
            Sealing::Interval(millis) => Box::new(StreamExt::map(
682
                Timer::interval(Duration::from_millis(millis)),
683
                |_| EngineCommand::SealNewBlock {
684
                    create_empty: true,
685
                    finalize: true,
686
                    parent_hash: None,
687
                    sender: None,
688
                },
689
            )),
690
        };
691
490
        let keystore_clone = keystore.clone();
692
490

            
693
490
        let babe_config = babe_link.config();
694
490
        let babe_consensus_provider = BabeConsensusDataProvider::new(
695
490
            client.clone(),
696
490
            keystore,
697
490
            babe_link.epoch_changes().clone(),
698
490
            babe_config.authorities.clone(),
699
490
        )
700
490
        .map_err(|babe_error| {
701
            Error::Consensus(consensus_common::Error::Other(babe_error.into()))
702
490
        })?;
703

            
704
        // Need to clone it and store here to avoid moving of `client`
705
        // variable in closure below.
706
490
        let client_clone = client.clone();
707
490

            
708
490
        task_manager.spawn_essential_handle().spawn_blocking(
709
490
            "authorship_task",
710
490
            Some("block-authoring"),
711
490
            run_manual_seal(ManualSealParams {
712
490
                block_import,
713
490
                env: proposer,
714
490
                client: client.clone(),
715
490
                pool: transaction_pool.clone(),
716
490
                commands_stream,
717
490
                select_chain,
718
41608
                create_inherent_data_providers: move |parent, ()| {
719
41608
                    let client_clone = client_clone.clone();
720
41608
                    let keystore = keystore_clone.clone();
721
41608
                    let downward_mock_para_inherent_receiver = downward_mock_para_inherent_receiver.clone();
722
41608
                    let upward_mock_receiver = upward_mock_receiver.clone();
723
41608
                    async move {
724
41608

            
725
41608
                        let downward_mock_para_inherent_receiver = downward_mock_para_inherent_receiver.clone();
726
41608
                        // here we only take the last one
727
41608
                        let para_inherent_decider_messages: Vec<Vec<u8>> = downward_mock_para_inherent_receiver.drain().collect();
728
41608

            
729
41608
                        let upward_messages_receiver = upward_mock_receiver.clone();
730

            
731
                        // If there is a value to be updated, we update it
732
41608
                        if let Some(value) = para_inherent_decider_messages.last() {
733
35
                            client_clone
734
35
                            .insert_aux(
735
35
                                &[(PARA_INHERENT_SELECTOR_AUX_KEY, value.as_slice())],
736
35
                                &[],
737
35
                            )
738
35
                            .expect("Should be able to write to aux storage; qed");
739
41573
                        }
740

            
741
41608
                        let parachain = MockParachainsInherentDataProvider::new(
742
41608
                            client_clone.clone(),
743
41608
                            parent,
744
41608
                            keystore,
745
41608
                            upward_messages_receiver,
746
41608
                        );
747
41608

            
748
41608
                        let timestamp = get_next_timestamp(client_clone, slot_duration);
749
41608

            
750
41608
                        let slot =
751
41608
                            sp_consensus_babe::inherents::InherentDataProvider::from_timestamp_and_slot_duration(
752
41608
                                *timestamp,
753
41608
                                slot_duration,
754
41608
                            );
755
41608

            
756
41608
                        Ok((slot, timestamp, parachain))
757
41608
                    }
758
41608
                },
759
490
                consensus_data_provider: Some(Box::new(babe_consensus_provider)),
760
490
            }),
761
490
        );
762
490
    }
763

            
764
490
    let dev_rpc = if role.clone().is_authority() {
765
490
        Some(DevRpc {
766
490
            mock_para_inherent_channel: downward_mock_para_inherent_sender,
767
490
            upward_message_channel: upward_mock_sender,
768
490
        })
769
    } else {
770
        None
771
    };
772

            
773
490
    let rpc_extensions_builder = {
774
490
        let client = client.clone();
775
490
        let transaction_pool = transaction_pool.clone();
776

            
777
        move |_subscription_executor: polkadot_rpc::SubscriptionTaskExecutor|
778
980
            -> Result<RpcExtension, service::Error> {
779
980
            let deps = DevDeps {
780
980
                client: client.clone(),
781
980
                pool: transaction_pool.clone(),
782
980
                command_sink: command_sink.clone(),
783
980
                dev_rpc: dev_rpc.clone(),
784
980
            };
785
980

            
786
980
            create_dev_rpc_extension(deps).map_err(Into::into)
787
980
        }
788
    };
789

            
790
490
    let rpc_handlers = service::spawn_tasks(service::SpawnTasksParams {
791
490
        config,
792
490
        backend: backend.clone(),
793
490
        client: client.clone(),
794
490
        keystore: keystore_container.keystore(),
795
490
        network: network.clone(),
796
490
        sync_service: sync_service.clone(),
797
490
        rpc_builder: Box::new(rpc_extensions_builder),
798
490
        transaction_pool: transaction_pool.clone(),
799
490
        task_manager: &mut task_manager,
800
490
        system_rpc_tx,
801
490
        tx_handler_controller,
802
490
        telemetry: telemetry.as_mut(),
803
490
    })?;
804

            
805
490
    network_starter.start_network();
806
490

            
807
490
    Ok(NewFull {
808
490
        task_manager,
809
490
        client,
810
490
        overseer_handle: None,
811
490
        network,
812
490
        sync_service,
813
490
        rpc_handlers,
814
490
        backend,
815
490
    })
816
490
}
817

            
818
490
fn new_partial<ChainSelection>(
819
490
    config: &mut Configuration,
820
490
    Basics {
821
490
        task_manager,
822
490
        backend,
823
490
        client,
824
490
        keystore_container,
825
490
        telemetry,
826
490
    }: Basics,
827
490
    select_chain: ChainSelection,
828
490
) -> Result<
829
490
    service::PartialComponents<
830
490
        FullClient,
831
490
        FullBackend,
832
490
        ChainSelection,
833
490
        sc_consensus::DefaultImportQueue<Block>,
834
490
        sc_transaction_pool::TransactionPoolHandle<Block, FullClient>,
835
490
        (
836
490
            BabeBlockImport<Block, FullClient, Arc<FullClient>>,
837
490
            BabeLink<Block>,
838
490
            SlotDuration,
839
490
            Option<Telemetry>,
840
490
        ),
841
490
    >,
842
490
    Error,
843
490
>
844
490
where
845
490
    ChainSelection: 'static + SelectChain<Block>,
846
490
{
847
490
    let transaction_pool = sc_transaction_pool::Builder::new(
848
490
        task_manager.spawn_essential_handle(),
849
490
        client.clone(),
850
490
        config.role.is_authority().into(),
851
490
    )
852
490
    .with_options(config.transaction_pool.clone())
853
490
    .with_prometheus(config.prometheus_registry())
854
490
    .build();
855

            
856
    // Create babe block import queue; this is required to have correct epoch data
857
    // available for manual seal to produce block
858
490
    let babe_config = babe::configuration(&*client)?;
859
490
    let (babe_block_import, babe_link) =
860
490
        babe::block_import(babe_config.clone(), client.clone(), client.clone())?;
861
490
    let slot_duration = babe_link.config().slot_duration();
862
490

            
863
490
    // Create manual seal block import with manual seal block import queue
864
490
    let import_queue = sc_consensus_manual_seal::import_queue(
865
490
        Box::new(babe_block_import.clone()),
866
490
        &task_manager.spawn_essential_handle(),
867
490
        config.prometheus_registry(),
868
490
    );
869
490

            
870
490
    Ok(service::PartialComponents {
871
490
        client,
872
490
        backend,
873
490
        task_manager,
874
490
        keystore_container,
875
490
        select_chain,
876
490
        import_queue,
877
490
        transaction_pool: transaction_pool.into(),
878
490
        other: (babe_block_import, babe_link, slot_duration, telemetry),
879
490
    })
880
490
}
881

            
882
490
fn new_partial_basics(
883
490
    config: &mut Configuration,
884
490
    telemetry_worker_handle: Option<TelemetryWorkerHandle>,
885
490
) -> Result<Basics, Error> {
886
490
    let telemetry = config
887
490
        .telemetry_endpoints
888
490
        .clone()
889
490
        .filter(|x| !x.is_empty())
890
490
        .map(move |endpoints| -> Result<_, telemetry::Error> {
891
            let (worker, mut worker_handle) = if let Some(worker_handle) = telemetry_worker_handle {
892
                (None, worker_handle)
893
            } else {
894
                let worker = TelemetryWorker::new(16)?;
895
                let worker_handle = worker.handle();
896
                (Some(worker), worker_handle)
897
            };
898
            let telemetry = worker_handle.new_telemetry(endpoints);
899
            Ok((worker, telemetry))
900
490
        })
901
490
        .transpose()?;
902

            
903
490
    let heap_pages = config
904
490
        .executor
905
490
        .default_heap_pages
906
490
        .map_or(DEFAULT_HEAP_ALLOC_STRATEGY, |h| HeapAllocStrategy::Static {
907
            extra_pages: h as u32,
908
490
        });
909
490

            
910
490
    let mut wasm_builder = WasmExecutor::builder()
911
490
        .with_execution_method(config.executor.wasm_method)
912
490
        .with_onchain_heap_alloc_strategy(heap_pages)
913
490
        .with_offchain_heap_alloc_strategy(heap_pages)
914
490
        .with_max_runtime_instances(config.executor.max_runtime_instances)
915
490
        .with_runtime_cache_size(config.executor.runtime_cache_size);
916
490
    if let Some(ref wasmtime_precompiled_path) = config.executor.wasmtime_precompiled {
917
462
        wasm_builder = wasm_builder.with_wasmtime_precompiled_path(wasmtime_precompiled_path);
918
462
    }
919
490
    let executor = wasm_builder.build();
920

            
921
490
    let (client, backend, keystore_container, task_manager) =
922
490
        service::new_full_parts::<Block, RuntimeApi, _>(
923
490
            config,
924
490
            telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),
925
490
            executor,
926
490
        )?;
927
490
    let client = Arc::new(client);
928
490

            
929
490
    let telemetry = telemetry.map(|(worker, telemetry)| {
930
        if let Some(worker) = worker {
931
            task_manager.spawn_handle().spawn(
932
                "telemetry",
933
                Some("telemetry"),
934
                Box::pin(worker.run()),
935
            );
936
        }
937
        telemetry
938
490
    });
939
490

            
940
490
    Ok(Basics {
941
490
        task_manager,
942
490
        client,
943
490
        backend,
944
490
        keystore_container,
945
490
        telemetry,
946
490
    })
947
490
}
948

            
949
use polkadot_primitives::{AvailabilityBitfield, UncheckedSigned, ValidatorId, ValidatorIndex};
950
use sp_keystore::Error as KeystoreError;
951
17409
fn keystore_sign<H: Encode, Payload: Encode>(
952
17409
    keystore: &KeystorePtr,
953
17409
    payload: Payload,
954
17409
    context: &SigningContext<H>,
955
17409
    validator_index: ValidatorIndex,
956
17409
    key: &ValidatorId,
957
17409
) -> Result<Option<UncheckedSigned<Payload>>, KeystoreError> {
958
17409
    let data = payload_data(&payload, context);
959
17409
    let signature = keystore
960
17409
        .sr25519_sign(ValidatorId::ID, key.as_ref(), &data)?
961
17409
        .map(|sig| UncheckedSigned::new(payload, validator_index, sig.into()));
962
17409
    Ok(signature)
963
17409
}
964

            
965
17409
fn payload_data<H: Encode, Payload: Encode>(
966
17409
    payload: &Payload,
967
17409
    context: &SigningContext<H>,
968
17409
) -> Vec<u8> {
969
17409
    // equivalent to (`real_payload`, context).encode()
970
17409
    let mut out = payload.encode_as();
971
17409
    out.extend(context.encode());
972
17409
    out
973
17409
}
974

            
975
/// Create an `AvailabilityBitfield` with size `total_cores`. The first `used_cores` set to true (occupied),
976
/// and the remaining to false (available).
977
9121
fn availability_bitvec(used_cores: usize, total_cores: usize) -> AvailabilityBitfield {
978
9121
    let mut bitfields = bitvec::bitvec![u8, bitvec::order::Lsb0; 0; 0];
979
35112
    for i in 0..total_cores {
980
35112
        if i < used_cores {
981
8778
            bitfields.push(true);
982
8778
        } else {
983
26334
            bitfields.push(false)
984
        }
985
    }
986

            
987
9121
    bitfields.into()
988
9121
}