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

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

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

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

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

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

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

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

            
207
948
    match config.network.network_backend {
208
        sc_network::config::NetworkBackendType::Libp2p => {
209
948
            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
948
}
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
55956
    pub fn new(
234
55956
        client: Arc<C>,
235
55956
        parent: Hash,
236
55956
        keystore: KeystorePtr,
237
55956
        upward_messages_receiver: flume::Receiver<Vec<u8>>,
238
55956
    ) -> Self {
239
55956
        MockParachainsInherentDataProvider {
240
55956
            client,
241
55956
            parent,
242
55956
            keystore,
243
55956
            upward_messages_receiver,
244
55956
        }
245
55956
    }
246

            
247
8382
    pub async fn create(
248
8382
        client: Arc<C>,
249
8382
        parent: Hash,
250
8382
        keystore: KeystorePtr,
251
8382
        upward_messages_receiver: flume::Receiver<Vec<u8>>,
252
8382
    ) -> Result<ParachainsInherentData, InherentError> {
253
8382
        let parent_header = match client.header(parent) {
254
8382
            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
8382
        let runtime_api = client.runtime_api();
269
8382

            
270
8382
        // we get all validators
271
8382

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

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

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

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

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

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

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

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

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

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

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

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

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

            
344
        // iterate over every core|para pair
345
22866
        for (core, para) in claim_queue {
346
            // check which group is assigned to each core
347
14484
            let group_assigned_to_core =
348
14484
                core.0 + rotations_since_session_start % groups.len() as u32;
349
14484
            // check validator indices associated to the core
350
14484
            let indices_associated_to_core = groups.get(group_assigned_to_core as usize).unwrap();
351
21810
            for index in indices_associated_to_core {
352
                // fetch validator keys
353
7326
                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
14652
                for type_public_pair in &available_keys {
356
7326
                    if let Ok(validator) =
357
7326
                        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
7326
                        if validator_keys_to_find == &validator {
361
7326
                            // we work with the previous included data
362
7326
                            let mut persisted_validation_data = runtime_api
363
7326
                                .persisted_validation_data(
364
7326
                                    parent,
365
7326
                                    para[0],
366
7326
                                    OccupiedCoreAssumption::Included,
367
7326
                                )
368
7326
                                .unwrap()
369
7326
                                .unwrap();
370
7326

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

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

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

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

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

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

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

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

            
460
8382
        Ok(ParachainsInherentData {
461
8382
            bitfields: bitfields,
462
8382
            backed_candidates: backed_cand,
463
8382
            disputes: Vec::new(),
464
8382
            parent_header,
465
8382
        })
466
8382
    }
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
55956
    ) -> Result<(), sp_inherents::Error> {
480
        // fetch whether the para inherent selector has been set
481
55956
        let maybe_para_selector = self
482
55956
            .client
483
55956
            .get_aux(PARA_INHERENT_SELECTOR_AUX_KEY)
484
55956
            .expect("Should be able to query aux storage; qed");
485

            
486
55956
        let inherent_data = {
487
55956
            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
8382
                if aux == true.encode() {
491
8382
                    MockParachainsInherentDataProvider::create(
492
8382
                        self.client.clone(),
493
8382
                        self.parent,
494
8382
                        self.keystore.clone(),
495
8382
                        self.upward_messages_receiver.clone(),
496
8382
                    )
497
                    .await
498
8382
                    .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
47574
                EmptyParachainsInherentDataProvider::create(self.client.clone(), self.parent)
506
                    .await
507
47574
                    .map_err(|e| sp_inherents::Error::Application(Box::new(e)))?
508
            }
509
        };
510

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
648
948
    let mut command_sink = None;
649
948

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
805
948
    network_starter.start_network();
806
948

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

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

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

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

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

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

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

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

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

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

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

            
967
15708
fn payload_data<H: Encode, Payload: Encode>(
968
15708
    payload: &Payload,
969
15708
    context: &SigningContext<H>,
970
15708
) -> Vec<u8> {
971
15708
    // equivalent to (`real_payload`, context).encode()
972
15708
    let mut out = payload.encode_as();
973
15708
    out.extend(context.encode());
974
15708
    out
975
15708
}
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
8382
fn availability_bitvec(used_cores: usize, total_cores: usize) -> AvailabilityBitfield {
980
8382
    let mut bitfields = bitvec::bitvec![u8, bitvec::order::Lsb0; 0; 0];
981
31656
    for i in 0..total_cores {
982
31656
        if i < used_cores {
983
7914
            bitfields.push(true);
984
7914
        } else {
985
23742
            bitfields.push(false)
986
        }
987
    }
988

            
989
8382
    bitfields.into()
990
8382
}