-
Notifications
You must be signed in to change notification settings - Fork 50
/
runtime_service.rs
3259 lines (2989 loc) · 138 KB
/
runtime_service.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Smoldot
// Copyright (C) 2019-2022 Parity Technologies (UK) Ltd.
// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Background runtime download service.
//!
//! This service plugs on top of a [`sync_service`], listens for new best blocks and checks
//! whether the runtime has changed in any way. Its objective is to always provide an up-to-date
//! [`executor::host::HostVmPrototype`] ready to be called by other services.
//!
//! # Usage
//!
//! The runtime service lets user subscribe to block updates, similar to the [`sync_service`].
//! These subscriptions are implemented by subscribing to the underlying [`sync_service`] and,
//! for each notification, checking whether the runtime has changed (thanks to the presence or
//! absence of a header digest item), and downloading the runtime code if necessary. Therefore,
//! these notifications might come with a delay compared to directly using the [`sync_service`].
//!
//! If it isn't possible to download the runtime code of a block (for example because peers refuse
//! to answer or have already pruned the block) or if the runtime service already has too many
//! pending downloads, this block is simply not reported on the subscriptions. The download will
//! be repeatedly tried until it succeeds.
//!
//! Consequently, you are strongly encouraged to not use both the [`sync_service`] *and* the
//! [`RuntimeService`] of the same chain. They each provide a consistent view of the chain, but
//! this view isn't necessarily the same on both services.
//!
//! The main service offered by the runtime service is [`RuntimeService::subscribe_all`], that
//! notifies about new blocks once their runtime is known.
//!
//! # Blocks pinning
//!
//! Blocks that are reported through [`RuntimeService::subscribe_all`] are automatically *pinned*.
//! If multiple subscriptions exist, each block is pinned once per subscription.
//!
//! As long as a block is pinned, the [`RuntimeService`] is guaranteed to keep in its internal
//! state the runtime of this block and its properties.
//!
//! Blocks must be manually unpinned by calling [`Subscription::unpin_block`].
//! Failing to do so is effectively a memory leak. If the number of pinned blocks becomes too
//! large, the subscription is force-killed by the [`RuntimeService`].
//!
use crate::{log, network_service, platform::PlatformRef, sync_service};
use alloc::{
borrow::{Cow, ToOwned as _},
boxed::Box,
collections::{BTreeMap, VecDeque},
format,
string::{String, ToString as _},
sync::{Arc, Weak},
vec::Vec,
};
use async_lock::Mutex;
use core::{
cmp, iter, mem,
num::{NonZeroU32, NonZeroUsize},
ops,
pin::Pin,
time::Duration,
};
use futures_channel::oneshot;
use futures_lite::FutureExt as _;
use futures_util::{future, stream, Stream, StreamExt as _};
use itertools::Itertools as _;
use rand::seq::IteratorRandom as _;
use rand_chacha::rand_core::SeedableRng as _;
use smoldot::{
chain::async_tree,
executor, header,
informant::{BytesDisplay, HashDisplay},
trie::{self, proof_decode, Nibble},
};
/// Configuration for a runtime service.
pub struct Config<TPlat: PlatformRef> {
/// Name of the chain, for logging purposes.
///
/// > **Note**: This name will be directly printed out. Any special character should already
/// > have been filtered out from this name.
pub log_name: String,
/// Access to the platform's capabilities.
pub platform: TPlat,
/// Service responsible for synchronizing the chain.
pub sync_service: Arc<sync_service::SyncService<TPlat>>,
/// Service responsible for accessing the networking of the chain.
pub network_service: Arc<network_service::NetworkServiceChain<TPlat>>,
/// Header of the genesis block of the chain, in SCALE encoding.
pub genesis_block_scale_encoded_header: Vec<u8>,
}
/// Runtime currently pinned within a [`RuntimeService`].
///
/// Destroying this object automatically unpins the runtime.
#[derive(Clone)]
pub struct PinnedRuntime(Arc<Runtime>);
/// See [the module-level documentation](..).
pub struct RuntimeService<TPlat: PlatformRef> {
/// Configuration of the background task. Used to restart the background task if necessary.
background_task_config: BackgroundTaskConfig<TPlat>,
/// Sender to send messages to the background task.
to_background: Mutex<async_channel::Sender<ToBackground<TPlat>>>,
}
impl<TPlat: PlatformRef> RuntimeService<TPlat> {
/// Initializes a new runtime service.
pub fn new(config: Config<TPlat>) -> Self {
// Target to use for all the logs of this service.
let log_target = format!("runtime-{}", config.log_name);
let background_task_config = BackgroundTaskConfig {
log_target: log_target.clone(),
platform: config.platform.clone(),
sync_service: config.sync_service,
network_service: config.network_service,
genesis_block_scale_encoded_header: config.genesis_block_scale_encoded_header,
};
// Spawns a task that runs in the background and updates the content of the mutex.
let to_background;
config.platform.spawn_task(log_target.clone().into(), {
let (tx, rx) = async_channel::bounded(16);
let tx_weak = tx.downgrade();
to_background = tx;
let background_task_config = background_task_config.clone();
run_background(background_task_config, rx, tx_weak)
});
RuntimeService {
background_task_config,
to_background: Mutex::new(to_background),
}
}
/// Calls [`sync_service::SyncService::block_number_bytes`] on the sync service associated to
/// this runtime service.
pub fn block_number_bytes(&self) -> usize {
self.background_task_config
.sync_service
.block_number_bytes()
}
/// Subscribes to the state of the chain: the current state and the new blocks.
///
/// This function only returns once the runtime of the current finalized block is known. This
/// might take a long time.
///
/// Only up to `buffer_size` block notifications are buffered in the channel. If the channel
/// is full when a new notification is attempted to be pushed, the channel gets closed.
///
/// A maximum number of finalized or non-canonical (i.e. not part of the finalized chain)
/// pinned blocks must be passed, indicating the maximum number of blocks that are finalized
/// or non-canonical that the runtime service will pin at the same time for this subscription.
/// If this maximum is reached, the channel will get closed. In situations where the subscriber
/// is guaranteed to always properly unpin blocks, a value of `usize::max_value()` can be
/// passed in order to ignore this maximum.
///
/// The channel also gets closed if a gap in the finality happens, such as after a Grandpa
/// warp syncing.
///
/// See [`SubscribeAll`] for information about the return value.
pub async fn subscribe_all(
&self,
buffer_size: usize,
max_pinned_blocks: NonZeroUsize,
) -> SubscribeAll<TPlat> {
loop {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.send_message_or_restart_service(ToBackground::SubscribeAll(
ToBackgroundSubscribeAll {
result_tx,
buffer_size,
max_pinned_blocks,
},
))
.await;
if let Ok(subscribe_all) = result_rx.await {
break subscribe_all;
}
}
}
/// Unpins a block after it has been reported by a subscription.
///
/// Has no effect if the [`SubscriptionId`] is not or no longer valid (as the runtime service
/// can kill any subscription at any moment).
///
/// # Panic
///
/// Panics if the block hash has not been reported or has already been unpinned.
///
// TODO: add #[track_caller] once possible, see https://github.com/rust-lang/rust/issues/87417
pub async fn unpin_block(&self, subscription_id: SubscriptionId, block_hash: [u8; 32]) {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.to_background
.lock()
.await
.send(ToBackground::UnpinBlock {
result_tx,
subscription_id,
block_hash,
})
.await;
match result_rx.await {
Ok(Ok(())) => {
// Background task has indicated success.
}
Err(_) => {
// Background task has crashed. Subscription is stale. Function has no effect.
}
Ok(Err(_)) => {
// Background task has indicated that the block has already been unpinned.
panic!()
}
}
}
/// Returns the storage value and Merkle value of the `:code` key of the finalized block.
///
/// Returns `None` if the runtime of the current finalized block is not known yet.
// TODO: this function has a bad API but is hopefully temporary
pub async fn finalized_runtime_storage_merkle_values(
&self,
) -> Option<(Option<Vec<u8>>, Option<Vec<u8>>, Option<Vec<Nibble>>)> {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.to_background
.lock()
.await
.send(ToBackground::FinalizedRuntimeStorageMerkleValues { result_tx })
.await;
result_rx.await.unwrap_or(None)
}
/// Pins the runtime of a pinned block.
///
/// The hash of the block passed as parameter corresponds to the block whose runtime is to
/// be pinned. The block must be currently pinned in the context of the provided
/// [`SubscriptionId`].
///
/// Returns the pinned runtime, plus the state trie root hash and height of the block.
///
/// Returns an error if the subscription is stale, meaning that it has been reset by the
/// runtime service.
pub async fn pin_pinned_block_runtime(
&self,
subscription_id: SubscriptionId,
block_hash: [u8; 32],
) -> Result<(PinnedRuntime, [u8; 32], u64), PinPinnedBlockRuntimeError> {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.to_background
.lock()
.await
.send(ToBackground::PinPinnedBlockRuntime {
result_tx,
subscription_id,
block_hash,
})
.await;
match result_rx.await {
Ok(result) => result.map(|(r, v, n)| (PinnedRuntime(r), v, n)),
Err(_) => {
// Background service has crashed. This means that the subscription is obsolete.
Err(PinPinnedBlockRuntimeError::ObsoleteSubscription)
}
}
}
/// Performs a runtime call.
///
/// The hash of the block passed as parameter corresponds to the block whose runtime to use
/// to make the call. The block must be currently pinned in the context of the provided
/// [`SubscriptionId`].
///
/// Returns an error if the subscription is stale, meaning that it has been reset by the
/// runtime service.
pub async fn runtime_call(
&self,
pinned_runtime: PinnedRuntime,
block_hash: [u8; 32],
block_number: u64,
block_state_trie_root_hash: [u8; 32],
function_name: String,
required_api_version: Option<(String, ops::RangeInclusive<u32>)>,
parameters_vectored: Vec<u8>,
total_attempts: u32,
timeout_per_request: Duration,
max_parallel: NonZeroU32,
) -> Result<RuntimeCallSuccess, RuntimeCallError> {
let (result_tx, result_rx) = oneshot::channel();
self.send_message_or_restart_service(ToBackground::RuntimeCall {
result_tx,
pinned_runtime: pinned_runtime.0,
block_hash,
block_number,
block_state_trie_root_hash,
function_name,
required_api_version,
parameters_vectored,
total_attempts,
timeout_per_request,
_max_parallel: max_parallel,
})
.await;
match result_rx.await {
Ok(result) => result,
Err(_) => {
// Background service has crashed.
Err(RuntimeCallError::Crash)
}
}
}
/// Tries to find a runtime within the [`RuntimeService`] that has the given storage code and
/// heap pages. If none is found, compiles the runtime and stores it within the
/// [`RuntimeService`].
pub async fn compile_and_pin_runtime(
&self,
storage_code: Option<Vec<u8>>,
storage_heap_pages: Option<Vec<u8>>,
code_merkle_value: Option<Vec<u8>>,
closest_ancestor_excluding: Option<Vec<Nibble>>,
) -> Result<PinnedRuntime, CompileAndPinRuntimeError> {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.send_message_or_restart_service(ToBackground::CompileAndPinRuntime {
result_tx,
storage_code,
storage_heap_pages,
code_merkle_value,
closest_ancestor_excluding,
})
.await;
Ok(PinnedRuntime(
result_rx
.await
.map_err(|_| CompileAndPinRuntimeError::Crash)?,
))
}
/// Returns the runtime specification of the given runtime.
pub async fn pinned_runtime_specification(
&self,
pinned_runtime: PinnedRuntime,
) -> Result<executor::CoreVersion, PinnedRuntimeSpecificationError> {
match &pinned_runtime.0.runtime {
Ok(rt) => Ok(rt.runtime_version().clone()),
Err(error) => Err(PinnedRuntimeSpecificationError::InvalidRuntime(
error.clone(),
)),
}
}
/// Returns true if it is believed that we are near the head of the chain.
///
/// The way this method is implemented is opaque and cannot be relied on. The return value
/// should only ever be shown to the user and not used for any meaningful logic.
pub async fn is_near_head_of_chain_heuristic(&self) -> bool {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.to_background
.lock()
.await
.send(ToBackground::IsNearHeadOfChainHeuristic { result_tx })
.await;
result_rx.await.unwrap_or(false)
}
/// Sends a message to the background task. Restarts the background task if it has crashed.
async fn send_message_or_restart_service(&self, message: ToBackground<TPlat>) {
let mut lock = self.to_background.lock().await;
if lock.is_closed() {
let (tx, rx) = async_channel::bounded(16);
let tx_weak = tx.downgrade();
*lock = tx;
self.background_task_config.platform.spawn_task(
self.background_task_config.log_target.clone().into(),
{
let background_task_config = self.background_task_config.clone();
let platform = background_task_config.platform.clone();
async move {
// Sleep for a bit in order to avoid infinite loops of repeated crashes.
background_task_config
.platform
.sleep(Duration::from_secs(2))
.await;
let log_target = background_task_config.log_target.clone();
log!(&platform, Debug, &log_target, "restart");
run_background(background_task_config, rx, tx_weak).await;
log!(&platform, Debug, &log_target, "shutdown");
}
},
);
}
// Note that the background task might have crashed again at this point already, and thus
// errors are not impossible.
let _ = lock.send(message).await;
}
}
/// Return value of [`RuntimeService::subscribe_all`].
pub struct SubscribeAll<TPlat: PlatformRef> {
/// SCALE-encoded header of the finalized block at the time of the subscription.
pub finalized_block_scale_encoded_header: Vec<u8>,
/// If the runtime of the finalized block is known, contains the information about it.
pub finalized_block_runtime: Result<executor::CoreVersion, RuntimeError>,
/// List of all known non-finalized blocks at the time of subscription.
///
/// Only one element in this list has [`BlockNotification::is_new_best`] equal to true.
///
/// The blocks are guaranteed to be ordered so that parents are always found before their
/// children.
pub non_finalized_blocks_ancestry_order: Vec<BlockNotification>,
/// Channel onto which new blocks are sent. The channel gets closed if it is full when a new
/// block needs to be reported.
pub new_blocks: Subscription<TPlat>,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct SubscriptionId(u64);
pub struct Subscription<TPlat: PlatformRef> {
subscription_id: u64,
channel: Pin<Box<async_channel::Receiver<Notification>>>,
to_background: async_channel::Sender<ToBackground<TPlat>>,
}
impl<TPlat: PlatformRef> Subscription<TPlat> {
pub async fn next(&mut self) -> Option<Notification> {
self.channel.next().await
}
/// Returns an opaque identifier that can be used to call [`RuntimeService::unpin_block`].
pub fn id(&self) -> SubscriptionId {
SubscriptionId(self.subscription_id)
}
/// Unpins a block after it has been reported.
///
/// # Panic
///
/// Panics if the block hash has not been reported or has already been unpinned.
///
pub async fn unpin_block(&self, block_hash: [u8; 32]) {
let (result_tx, result_rx) = oneshot::channel();
let _ = self
.to_background
.send(ToBackground::UnpinBlock {
result_tx,
subscription_id: SubscriptionId(self.subscription_id),
block_hash,
})
.await;
result_rx.await.unwrap().unwrap()
}
}
/// Notification about a new block or a new finalized block.
///
/// See [`RuntimeService::subscribe_all`].
#[derive(Debug, Clone)]
pub enum Notification {
/// A non-finalized block has been finalized.
Finalized {
/// BLAKE2 hash of the header of the block that has been finalized.
///
/// A block with this hash is guaranteed to have earlier been reported in a
/// [`BlockNotification`], either in [`SubscribeAll::non_finalized_blocks_ancestry_order`]
/// or in a [`Notification::Block`].
///
/// It is also guaranteed that this block is a child of the previously-finalized block. In
/// other words, if multiple blocks are finalized at the same time, only one
/// [`Notification::Finalized`] is generated and contains the highest finalized block.
///
/// If it is not possible for the [`RuntimeService`] to avoid a gap in the list of
/// finalized blocks, then the [`SubscribeAll::new_blocks`] channel is force-closed.
hash: [u8; 32],
/// Hash of the header of the best block after the finalization.
///
/// If the newly-finalized block is an ancestor of the current best block, then this field
/// contains the hash of this current best block. Otherwise, the best block is now
/// the non-finalized block with the given hash.
///
/// A block with this hash is guaranteed to have earlier been reported in a
/// [`BlockNotification`], either in [`SubscribeAll::non_finalized_blocks_ancestry_order`]
/// or in a [`Notification::Block`].
best_block_hash: [u8; 32],
/// List of BLAKE2 hashes of the headers of the blocks that have been discarded because
/// they're not descendants of the newly-finalized block.
///
/// This list contains all the siblings of the newly-finalized block and all their
/// descendants.
pruned_blocks: Vec<[u8; 32]>,
},
/// A new block has been added to the list of unfinalized blocks.
Block(BlockNotification),
/// The best block has changed to a different one.
BestBlockChanged {
/// Hash of the new best block.
///
/// This can be either the hash of the latest finalized block or the hash of a
/// non-finalized block.
hash: [u8; 32],
},
}
/// Notification about a new block.
///
/// See [`RuntimeService::subscribe_all`].
#[derive(Debug, Clone)]
pub struct BlockNotification {
/// True if this block is considered as the best block of the chain.
pub is_new_best: bool,
/// SCALE-encoded header of the block.
pub scale_encoded_header: Vec<u8>,
/// BLAKE2 hash of the header of the parent of this block.
///
///
/// A block with this hash is guaranteed to have earlier been reported in a
/// [`BlockNotification`], either in [`SubscribeAll::non_finalized_blocks_ancestry_order`] or
/// in a [`Notification::Block`].
///
/// > **Note**: The header of a block contains the hash of its parent. When it comes to
/// > consensus algorithms such as Babe or Aura, the syncing code verifies that this
/// > hash, stored in the header, actually corresponds to a valid block. However,
/// > when it comes to parachain consensus, no such verification is performed.
/// > Contrary to the hash stored in the header, the value of this field is
/// > guaranteed to refer to a block that is known by the syncing service. This
/// > allows a subscriber of the state of the chain to precisely track the hierarchy
/// > of blocks, without risking to run into a problem in case of a block with an
/// > invalid header.
pub parent_hash: [u8; 32],
/// If the runtime of the block is different from its parent, contains the information about
/// the new runtime.
pub new_runtime: Option<Result<executor::CoreVersion, RuntimeError>>,
}
/// Successful runtime call.
#[derive(Debug)]
pub struct RuntimeCallSuccess {
/// Output of the runtime call.
pub output: Vec<u8>,
/// Version of the API that was found. `Some` if and only if an API requirement was passed.
pub api_version: Option<u32>,
}
/// See [`RuntimeService::pin_pinned_block_runtime`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum PinPinnedBlockRuntimeError {
/// Subscription is dead.
ObsoleteSubscription,
/// Requested block isn't pinned by the subscription.
BlockNotPinned,
}
/// See [`RuntimeService::pinned_runtime_specification`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum PinnedRuntimeSpecificationError {
/// The runtime is invalid.
InvalidRuntime(RuntimeError),
}
/// See [`RuntimeService::runtime_call`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum RuntimeCallError {
/// The runtime of the requested block is invalid.
InvalidRuntime(RuntimeError),
/// API version required for the call isn't fulfilled.
ApiVersionRequirementUnfulfilled,
/// Runtime service has crashed while the call was in progress.
///
/// This doesn't necessarily indicate that the call was responsible for this crash.
Crash,
/// Error during the execution of the runtime.
///
/// There is no point in trying the same call again, as it would result in the same error.
#[display(fmt = "Error during the execution of the runtime: {_0}")]
Execution(RuntimeCallExecutionError),
/// Error trying to access the storage required for the runtime call.
///
/// Because these errors are non-fatal, the operation is attempted multiple times, and as such
/// there can be multiple errors.
///
/// Trying the same call again might succeed.
#[display(fmt = "Error trying to access the storage required for the runtime call")]
// TODO: better display?
Inaccessible(Vec<RuntimeCallInaccessibleError>),
}
/// See [`RuntimeCallError::Execution`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum RuntimeCallExecutionError {
/// Failed to initialize the virtual machine.
Start(executor::host::StartErr),
/// Error during the execution of the virtual machine.
Execution(executor::runtime_call::ErrorDetail),
/// Virtual machine has called a host function that it is not allowed to call.
ForbiddenHostFunction,
}
/// See [`RuntimeCallError::Inaccessible`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum RuntimeCallInaccessibleError {
/// Failed to download the call proof from the network.
Network(network_service::CallProofRequestError),
/// Call proof downloaded from the network has an invalid format.
InvalidCallProof(proof_decode::Error),
/// One or more entries are missing from the downloaded call proof.
MissingProofEntry,
}
/// Error when analyzing the runtime.
#[derive(Debug, derive_more::Display, Clone)]
pub enum RuntimeError {
/// The `:code` key of the storage is empty.
CodeNotFound,
/// Error while parsing the `:heappages` storage value.
#[display(fmt = "Failed to parse `:heappages` storage value: {_0}")]
InvalidHeapPages(executor::InvalidHeapPagesError),
/// Error while compiling the runtime.
#[display(fmt = "{_0}")]
Build(executor::host::NewErr),
}
/// Error potentially returned by [`RuntimeService::compile_and_pin_runtime`].
#[derive(Debug, derive_more::Display, Clone)]
pub enum CompileAndPinRuntimeError {
/// Background service has crashed while compiling this runtime. The crash might however not
/// necessarily be caused by the runtime compilation.
Crash,
}
/// Message towards the background task.
enum ToBackground<TPlat: PlatformRef> {
SubscribeAll(ToBackgroundSubscribeAll<TPlat>),
CompileAndPinRuntime {
result_tx: oneshot::Sender<Arc<Runtime>>,
storage_code: Option<Vec<u8>>,
storage_heap_pages: Option<Vec<u8>>,
code_merkle_value: Option<Vec<u8>>,
closest_ancestor_excluding: Option<Vec<Nibble>>,
},
FinalizedRuntimeStorageMerkleValues {
// TODO: overcomplicated
result_tx: oneshot::Sender<Option<(Option<Vec<u8>>, Option<Vec<u8>>, Option<Vec<Nibble>>)>>,
},
IsNearHeadOfChainHeuristic {
result_tx: oneshot::Sender<bool>,
},
UnpinBlock {
result_tx: oneshot::Sender<Result<(), ()>>,
subscription_id: SubscriptionId,
block_hash: [u8; 32],
},
PinPinnedBlockRuntime {
result_tx:
oneshot::Sender<Result<(Arc<Runtime>, [u8; 32], u64), PinPinnedBlockRuntimeError>>,
subscription_id: SubscriptionId,
block_hash: [u8; 32],
},
RuntimeCall {
result_tx: oneshot::Sender<Result<RuntimeCallSuccess, RuntimeCallError>>,
pinned_runtime: Arc<Runtime>,
block_hash: [u8; 32],
block_number: u64,
block_state_trie_root_hash: [u8; 32],
function_name: String,
required_api_version: Option<(String, ops::RangeInclusive<u32>)>,
parameters_vectored: Vec<u8>,
total_attempts: u32,
timeout_per_request: Duration,
_max_parallel: NonZeroU32,
},
}
struct ToBackgroundSubscribeAll<TPlat: PlatformRef> {
result_tx: oneshot::Sender<SubscribeAll<TPlat>>,
buffer_size: usize,
max_pinned_blocks: NonZeroUsize,
}
#[derive(Clone)]
struct PinnedBlock {
/// Reference-counted runtime of the pinned block.
runtime: Arc<Runtime>,
/// Hash of the trie root of the pinned block.
state_trie_root_hash: [u8; 32],
/// Height of the pinned block.
block_number: u64,
/// `true` if the block is non-finalized and part of the canonical chain.
/// If `true`, then the block doesn't count towards the maximum number of pinned blocks of
/// the subscription.
block_ignores_limit: bool,
}
#[derive(Clone)]
struct Block {
/// Hash of the block in question. Redundant with `header`, but the hash is so often needed
/// that it makes sense to cache it.
hash: [u8; 32],
/// Height of the block.
height: u64,
/// Header of the block in question.
/// Guaranteed to always be valid for the output best and finalized blocks. Otherwise,
/// not guaranteed to be valid.
scale_encoded_header: Vec<u8>,
}
#[derive(Clone)]
struct BackgroundTaskConfig<TPlat: PlatformRef> {
log_target: String,
platform: TPlat,
sync_service: Arc<sync_service::SyncService<TPlat>>,
network_service: Arc<network_service::NetworkServiceChain<TPlat>>,
genesis_block_scale_encoded_header: Vec<u8>,
}
async fn run_background<TPlat: PlatformRef>(
config: BackgroundTaskConfig<TPlat>,
to_background: async_channel::Receiver<ToBackground<TPlat>>,
to_background_tx: async_channel::WeakSender<ToBackground<TPlat>>,
) {
log!(
&config.platform,
Trace,
&config.log_target,
"start",
genesis_block_hash = HashDisplay(&header::hash_from_scale_encoded_header(
&config.genesis_block_scale_encoded_header
))
);
// State machine containing all the state that will be manipulated below.
let mut background = {
let tree = {
let mut tree = async_tree::AsyncTree::new(async_tree::Config {
finalized_async_user_data: None,
retry_after_failed: Duration::from_secs(10),
blocks_capacity: 32,
});
let node_index = tree.input_insert_block(
Block {
hash: header::hash_from_scale_encoded_header(
&config.genesis_block_scale_encoded_header,
),
height: 0,
scale_encoded_header: config.genesis_block_scale_encoded_header,
},
None,
false,
true,
);
tree.input_finalize(node_index, node_index);
Tree::FinalizedBlockRuntimeUnknown { tree }
};
Background {
log_target: config.log_target.clone(),
platform: config.platform.clone(),
sync_service: config.sync_service.clone(),
network_service: config.network_service.clone(),
to_background: Box::pin(to_background.clone()),
to_background_tx: to_background_tx.clone(),
next_subscription_id: 0,
tree,
runtimes: slab::Slab::with_capacity(2),
pending_subscriptions: VecDeque::with_capacity(8),
blocks_stream: None,
runtime_downloads: stream::FuturesUnordered::new(),
progress_runtime_call_requests: stream::FuturesUnordered::new(),
}
};
// Inner loop. Process incoming events.
loop {
// Yield at every loop in order to provide better tasks granularity.
futures_lite::future::yield_now().await;
enum WakeUpReason<TPlat: PlatformRef> {
MustSubscribe,
StartDownload(async_tree::AsyncOpId, async_tree::NodeIndex),
TreeAdvanceFinalizedKnown(async_tree::OutputUpdate<Block, Arc<Runtime>>),
TreeAdvanceFinalizedUnknown(async_tree::OutputUpdate<Block, Option<Arc<Runtime>>>),
StartPendingSubscribeAll(ToBackgroundSubscribeAll<TPlat>),
Notification(Option<sync_service::Notification>),
ToBackground(ToBackground<TPlat>),
ForegroundClosed,
RuntimeDownloadFinished(
async_tree::AsyncOpId,
Result<
(
Option<Vec<u8>>,
Option<Vec<u8>>,
Option<Vec<u8>>,
Option<Vec<Nibble>>,
),
RuntimeDownloadError,
>,
),
ProgressRuntimeCallRequest(ProgressRuntimeCallRequest),
}
// Wait for something to happen or for some processing to be necessary.
let wake_up_reason: WakeUpReason<_> = {
let finalized_block_known =
matches!(background.tree, Tree::FinalizedBlockRuntimeKnown { .. });
let num_runtime_downloads = background.runtime_downloads.len();
async {
if finalized_block_known {
if let Some(pending_subscription) = background.pending_subscriptions.pop_front()
{
WakeUpReason::StartPendingSubscribeAll(pending_subscription)
} else {
future::pending().await
}
} else {
future::pending().await
}
}
.or(async {
if let Some(blocks_stream) = background.blocks_stream.as_mut() {
WakeUpReason::Notification(blocks_stream.next().await)
} else {
WakeUpReason::MustSubscribe
}
})
.or(async {
background
.to_background
.next()
.await
.map_or(WakeUpReason::ForegroundClosed, WakeUpReason::ToBackground)
})
.or(async {
if !background.runtime_downloads.is_empty() {
let (async_op_id, download_result) =
background.runtime_downloads.select_next_some().await;
WakeUpReason::RuntimeDownloadFinished(async_op_id, download_result)
} else {
future::pending().await
}
})
.or(async {
if !background.progress_runtime_call_requests.is_empty() {
let result = background
.progress_runtime_call_requests
.select_next_some()
.await;
WakeUpReason::ProgressRuntimeCallRequest(result)
} else {
future::pending().await
}
})
.or(async {
loop {
// There might be a new runtime download to start.
// Don't download more than 2 runtimes at a time.
let wait = if num_runtime_downloads < 2 {
// Grab what to download. If there's nothing more to download, do nothing.
let async_op = match &mut background.tree {
Tree::FinalizedBlockRuntimeKnown { tree, .. } => {
tree.next_necessary_async_op(&background.platform.now())
}
Tree::FinalizedBlockRuntimeUnknown { tree, .. } => {
tree.next_necessary_async_op(&background.platform.now())
}
};
match async_op {
async_tree::NextNecessaryAsyncOp::Ready(dl) => {
break WakeUpReason::StartDownload(dl.id, dl.block_index)
}
async_tree::NextNecessaryAsyncOp::NotReady { when } => {
if let Some(when) = when {
either::Left(background.platform.sleep_until(when))
} else {
either::Right(future::pending())
}
}
}
} else {
either::Right(future::pending())
};
match &mut background.tree {
Tree::FinalizedBlockRuntimeKnown { tree, .. } => {
match tree.try_advance_output() {
Some(update) => {
break WakeUpReason::TreeAdvanceFinalizedKnown(update)
}
None => wait.await,
}
}
Tree::FinalizedBlockRuntimeUnknown { tree, .. } => {
match tree.try_advance_output() {
Some(update) => {
break WakeUpReason::TreeAdvanceFinalizedUnknown(update)
}
None => wait.await,
}
}
}
}
})
.await
};
match wake_up_reason {
WakeUpReason::StartDownload(download_id, block_index) => {
let block = match &mut background.tree {
Tree::FinalizedBlockRuntimeKnown { tree, .. } => &tree[block_index],
Tree::FinalizedBlockRuntimeUnknown { tree, .. } => &tree[block_index],
};
log!(
&background.platform,
Debug,
&background.log_target,
"block-runtime-download-start",
block_hash = HashDisplay(&block.hash)
);
// Dispatches a runtime download task to `runtime_downloads`.
background.runtime_downloads.push(Box::pin({
let future = download_runtime(
background.sync_service.clone(),
block.hash,
&block.scale_encoded_header,
);
async move { (download_id, future.await) }
}));
}
WakeUpReason::TreeAdvanceFinalizedKnown(async_tree::OutputUpdate::Finalized {
user_data: new_finalized,
best_block_index,
pruned_blocks,
former_finalized_async_op_user_data: former_finalized_runtime,
..
}) => {
let Tree::FinalizedBlockRuntimeKnown {
tree,