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

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

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

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

            
153
994
    Ok(io)
154
994
}
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
32536
    pub async fn create<C: HeaderBackend<Block>>(
172
32536
        client: Arc<C>,
173
32536
        parent: Hash,
174
32536
    ) -> Result<ParachainsInherentData, InherentError> {
175
32536
        let parent_header = match client.header(parent) {
176
32536
            Ok(Some(h)) => h,
177
            Ok(None) => return Err(InherentError::ParentHeaderNotFound(parent)),
178
            Err(err) => return Err(InherentError::Blockchain(err)),
179
        };
180

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

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

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

            
207
497
    match config.network.network_backend {
208
        sc_network::config::NetworkBackendType::Libp2p => {
209
497
            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
497
}
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
41657
    pub fn new(
234
41657
        client: Arc<C>,
235
41657
        parent: Hash,
236
41657
        keystore: KeystorePtr,
237
41657
        upward_messages_receiver: flume::Receiver<Vec<u8>>,
238
41657
    ) -> Self {
239
41657
        MockParachainsInherentDataProvider {
240
41657
            client,
241
41657
            parent,
242
41657
            keystore,
243
41657
            upward_messages_receiver,
244
41657
        }
245
41657
    }
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
25529
        for (core, para) in claim_queue {
346
            // check which group is assigned to each core
347
16408
            let group_assigned_to_core =
348
16408
                core.0 + rotations_since_session_start % groups.len() as u32;
349
16408
            // check validator indices associated to the core
350
16408
            let indices_associated_to_core = groups.get(group_assigned_to_core as usize).unwrap();
351
24710
            for index in indices_associated_to_core {
352
                // fetch validator keys
353
8302
                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
16604
                for type_public_pair in &available_keys {
356
8302
                    if let Ok(validator) =
357
8302
                        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
8302
                        if validator_keys_to_find == &validator {
361
8302
                            // we work with the previous included data
362
8302
                            let mut persisted_validation_data = runtime_api
363
8302
                                .persisted_validation_data(
364
8302
                                    parent,
365
8302
                                    para[0],
366
8302
                                    OccupiedCoreAssumption::Included,
367
8302
                                )
368
8302
                                .unwrap()
369
8302
                                .unwrap();
370
8302

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

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

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

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

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

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

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

            
447
8302
                            // push the candidate
448
8302
                            backed_cand.push(BackedCandidate::<H256>::new(
449
8302
                                candidate,
450
8302
                                validity_votes.clone(),
451
8302
                                bitvec::bitvec![u8, bitvec::order::Lsb0; 1; indices_associated_to_core.len()],
452
8302
                                Some(core),
453
8302
                            ));
454
8302
                        }
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
41657
    ) -> Result<(), sp_inherents::Error> {
480
        // fetch whether the para inherent selector has been set
481
41657
        let maybe_para_selector = self
482
41657
            .client
483
41657
            .get_aux(PARA_INHERENT_SELECTOR_AUX_KEY)
484
41657
            .expect("Should be able to query aux storage; qed");
485

            
486
41657
        let inherent_data = {
487
41657
            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
32536
                EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)
506
                    .await
507
32536
                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?
508
            }
509
        };
510

            
511
41657
        dst_inherent_data.put_data(
512
41657
            polkadot_primitives::PARACHAINS_INHERENT_IDENTIFIER,
513
41657
            &inherent_data,
514
41657
        )
515
83314
    }
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
41657
fn get_next_timestamp(
530
41657
    client: Arc<FullClient>,
531
41657
    slot_duration: SlotDuration,
532
41657
) -> sp_timestamp::InherentDataProvider {
533
    const TIMESTAMP_AUX_KEY: &[u8] = b"__DEV_TIMESTAMP";
534

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
648
497
    let mut command_sink = None;
649
497

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

            
659
497
        let commands_stream: Box<
660
497
            dyn Stream<Item = EngineCommand<<Block as BlockT>::Hash>> + Send + Sync + Unpin,
661
497
        > = 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
497
                let (sink, stream) = futures::channel::mpsc::channel(1000);
677
497
                // Keep a reference to the other end of the channel. It goes to the RPC.
678
497
                command_sink = Some(sink);
679
497
                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
497
        let keystore_clone = keystore.clone();
692
497

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

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

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

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

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

            
731
                        // If there is a value to be updated, we update it
732
41657
                        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
41622
                        }
740

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

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

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

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

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

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

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

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

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

            
805
497
    network_starter.start_network();
806
497

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

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

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

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

            
882
497
fn new_partial_basics(
883
497
    config: &mut Configuration,
884
497
    telemetry_worker_handle: Option<TelemetryWorkerHandle>,
885
497
) -> Result<Basics, Error> {
886
497
    let telemetry = config
887
497
        .telemetry_endpoints
888
497
        .clone()
889
497
        .filter(|x| !x.is_empty())
890
497
        .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
497
        })
901
497
        .transpose()?;
902

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

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

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

            
929
497
    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
497
    });
939
497

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

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

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

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

            
989
9121
    bitfields.into()
990
9121
}