Grcov report - backend.rs

1

// Copyright 2024 Moonbeam foundation

2

// This file is part of Moonbeam.

3

4

// Moonbeam 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

// Moonbeam 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 Moonbeam.  If not, see <http://www.gnu.org/licenses/>.

16

17

use sp_blockchain::{CachedHeaderMetadata, HeaderMetadata};

18

use sp_core::storage::well_known_keys;

19

use sp_runtime::{

20

	generic::BlockId,

21

	traits::{Block as BlockT, HashingFor, Header as HeaderT, NumberFor, Zero},

22

	Justification, Justifications, StateVersion, Storage,

23

};

24

use sp_state_machine::{

25

	BackendTransaction, ChildStorageCollection, IndexOperation, StorageCollection, TrieBackend,

26

};

27

use std::future::Future;

28

use std::marker::PhantomData;

29

use std::ops::AddAssign;

30

use std::time::Duration;

31

use std::{

32

	collections::{HashMap, HashSet},

33

	ptr,

34

	sync::Arc,

35

};

36

37

use sc_client_api::{

38

	backend::{self, NewBlockState},

39

	blockchain::{self, BlockStatus, HeaderBackend},

40

	leaves::LeafSet,

41

	UsageInfo,

42

};

43

44

use jsonrpsee::http_client::HttpClient;

45

use sp_runtime::generic::SignedBlock;

46

47

use crate::chain_spec;

48

use crate::lazy_loading::lock::ReadWriteLock;

49

use crate::lazy_loading::state_overrides::StateEntry;

50

use crate::lazy_loading::{helpers, state_overrides};

51

use moonbeam_cli_opt::LazyLoadingConfig;

52

use moonbeam_core_primitives::BlockNumber;

53

use sc_client_api::StorageKey;

54

use sc_service::{Configuration, Error};

55

use serde::de::DeserializeOwned;

56

use sp_core::offchain::storage::InMemOffchainStorage;

57

use sp_core::{twox_128, H256};

58

use sp_rpc::list::ListOrValue;

59

use sp_rpc::number::NumberOrHex;

60

use sp_storage::{ChildInfo, StorageData};

61

use sp_trie::PrefixedMemoryDB;

62

use tokio_retry::strategy::FixedInterval;

63

use tokio_retry::Retry;

64

65

struct PendingBlock<B: BlockT> {

66

	block: StoredBlock<B>,

67

	state: NewBlockState,

68

69

70

#[derive(PartialEq, Eq, Clone)]

71

enum StoredBlock<B: BlockT> {

72

	Header(B::Header, Option<Justifications>),

73

	Full(B, Option<Justifications>),

74

75

76

impl<B: BlockT> StoredBlock<B> {

77

	fn new(

78

		header: B::Header,

79

		body: Option<Vec<B::Extrinsic>>,

80

		just: Option<Justifications>,

81

	) -> Self {

82

		match body {

83

			Some(body) => StoredBlock::Full(B::new(header, body), just),

84

			None => StoredBlock::Header(header, just),

85

86

87

88

	fn header(&self) -> &B::Header {

89

		match *self {

90

			StoredBlock::Header(ref h, _) => h,

91

			StoredBlock::Full(ref b, _) => b.header(),

92

93

94

95

	fn justifications(&self) -> Option<&Justifications> {

96

		match *self {

97

			StoredBlock::Header(_, ref j) | StoredBlock::Full(_, ref j) => j.as_ref(),

98

99

100

101

	fn extrinsics(&self) -> Option<&[B::Extrinsic]> {

102

		match *self {

103

			StoredBlock::Header(_, _) => None,

104

			StoredBlock::Full(ref b, _) => Some(b.extrinsics()),

105

106

107

108

	fn into_inner(self) -> (B::Header, Option<Vec<B::Extrinsic>>, Option<Justifications>) {

109

		match self {

110

			StoredBlock::Header(header, just) => (header, None, just),

111

			StoredBlock::Full(block, just) => {

112

				let (header, body) = block.deconstruct();

113

				(header, Some(body), just)

114

115

116

117

118

119

#[derive(Clone)]

120

struct BlockchainStorage<Block: BlockT> {

121

	blocks: HashMap<Block::Hash, StoredBlock<Block>>,

122

	hashes: HashMap<NumberFor<Block>, Block::Hash>,

123

	best_hash: Block::Hash,

124

	best_number: NumberFor<Block>,

125

	finalized_hash: Block::Hash,

126

	finalized_number: NumberFor<Block>,

127

	genesis_hash: Block::Hash,

128

	header_cht_roots: HashMap<NumberFor<Block>, Block::Hash>,

129

	leaves: LeafSet<Block::Hash, NumberFor<Block>>,

130

	aux: HashMap<Vec<u8>, Vec<u8>>,

131

132

133

/// In-memory blockchain. Supports concurrent reads.

134

#[derive(Clone)]

135

pub struct Blockchain<Block: BlockT> {

136

	rpc_client: Arc<RPC>,

137

	storage: Arc<ReadWriteLock<BlockchainStorage<Block>>>,

138

139

140

impl<Block: BlockT + DeserializeOwned> Blockchain<Block> {

141

	/// Get header hash of given block.

142

	pub fn id(&self, id: BlockId<Block>) -> Option<Block::Hash> {

143

		match id {

144

			BlockId::Hash(h) => Some(h),

145

			BlockId::Number(n) => self.storage.read().hashes.get(&n).cloned(),

146

147

148

149

	/// Create new in-memory blockchain storage.

150

	fn new(rpc_client: Arc<RPC>) -> Blockchain<Block> {

151

		let storage = Arc::new(ReadWriteLock::new(BlockchainStorage {

152

			blocks: HashMap::new(),

153

			hashes: HashMap::new(),

154

			best_hash: Default::default(),

155

			best_number: Zero::zero(),

156

			finalized_hash: Default::default(),

157

			finalized_number: Zero::zero(),

158

			genesis_hash: Default::default(),

159

			header_cht_roots: HashMap::new(),

160

			leaves: LeafSet::new(),

161

			aux: HashMap::new(),

162

		}));

163

		Blockchain {

164

			rpc_client,

165

			storage,

166

167

168

169

	/// Insert a block header and associated data.

170

	pub fn insert(

171

		&self,

172

		hash: Block::Hash,

173

		header: <Block as BlockT>::Header,

174

		justifications: Option<Justifications>,

175

		body: Option<Vec<<Block as BlockT>::Extrinsic>>,

176

		new_state: NewBlockState,

177

	) -> sp_blockchain::Result<()> {

178

		let number = *header.number();

179

		if new_state.is_best() {

180

			self.apply_head(&header)?;

181

182

183

184

			let mut storage = self.storage.write();

185

			storage.leaves.import(hash, number, *header.parent_hash());

186

			storage

187

				.blocks

188

				.insert(hash, StoredBlock::new(header, body, justifications));

189

190

			if let NewBlockState::Final = new_state {

191

				storage.finalized_hash = hash;

192

				storage.finalized_number = number;

193

194

195

			if number == Zero::zero() {

196

				storage.genesis_hash = hash;

197

198

199

200

		Ok(())

201

202

203

	/// Get total number of blocks.

204

	pub fn blocks_count(&self) -> usize {

205

		let count = self.storage.read().blocks.len();

206

207

		log::debug!(

208

			target: super::LAZY_LOADING_LOG_TARGET,

209

			"Total number of blocks: {:?}",

210

			count

211

);

212

213

		count

214

215

216

	/// Compare this blockchain with another in-mem blockchain

217

	pub fn equals_to(&self, other: &Self) -> bool {

218

		// Check ptr equality first to avoid double read locks.

219

		if ptr::eq(self, other) {

220

			return true;

221

222

		self.canon_equals_to(other) && self.storage.read().blocks == other.storage.read().blocks

223

224

225

	/// Compare canonical chain to other canonical chain.

226

	pub fn canon_equals_to(&self, other: &Self) -> bool {

227

		// Check ptr equality first to avoid double read locks.

228

		if ptr::eq(self, other) {

229

			return true;

230

231

		let this = self.storage.read();

232

		let other = other.storage.read();

233

		this.hashes == other.hashes

234

			&& this.best_hash == other.best_hash

235

			&& this.best_number == other.best_number

236

			&& this.genesis_hash == other.genesis_hash

237

238

239

	/// Insert header CHT root.

240

	pub fn insert_cht_root(&self, block: NumberFor<Block>, cht_root: Block::Hash) {

241

		self.storage

242

			.write()

243

			.header_cht_roots

244

			.insert(block, cht_root);

245

246

247

	/// Set an existing block as head.

248

	pub fn set_head(&self, hash: Block::Hash) -> sp_blockchain::Result<()> {

249

		let header = self

250

			.header(hash)?

251

			.ok_or_else(|| sp_blockchain::Error::UnknownBlock(format!("{}", hash)))?;

252

253

		self.apply_head(&header)

254

255

256

	fn apply_head(&self, header: &<Block as BlockT>::Header) -> sp_blockchain::Result<()> {

257

		let mut storage = self.storage.write();

258

259

		let hash = header.hash();

260

		let number = header.number();

261

262

		storage.best_hash = hash;

263

		storage.best_number = *number;

264

		storage.hashes.insert(*number, hash);

265

266

		Ok(())

267

268

269

	fn finalize_header(

270

		&self,

271

		block: Block::Hash,

272

		justification: Option<Justification>,

273

	) -> sp_blockchain::Result<()> {

274

		let mut storage = self.storage.write();

275

		storage.finalized_hash = block;

276

277

		if justification.is_some() {

278

			let block = storage

279

				.blocks

280

				.get_mut(&block)

281

				.expect("hash was fetched from a block in the db; qed");

282

283

			let block_justifications = match block {

284

				StoredBlock::Header(_, ref mut j) | StoredBlock::Full(_, ref mut j) => j,

285

};

286

287

			*block_justifications = justification.map(Justifications::from);

288

289

290

		Ok(())

291

292

293

	fn append_justification(

294

		&self,

295

		hash: Block::Hash,

296

		justification: Justification,

297

	) -> sp_blockchain::Result<()> {

298

		let mut storage = self.storage.write();

299

300

		let block = storage

301

			.blocks

302

			.get_mut(&hash)

303

			.expect("hash was fetched from a block in the db; qed");

304

305

		let block_justifications = match block {

306

			StoredBlock::Header(_, ref mut j) | StoredBlock::Full(_, ref mut j) => j,

307

};

308

309

		if let Some(stored_justifications) = block_justifications {

310

			if !stored_justifications.append(justification) {

311

				return Err(sp_blockchain::Error::BadJustification(

312

					"Duplicate consensus engine ID".into(),

313

));

314

315

		} else {

316

			*block_justifications = Some(Justifications::from(justification));

317

};

318

319

		Ok(())

320

321

322

	fn write_aux(&self, ops: Vec<(Vec<u8>, Option<Vec<u8>>)>) {

323

		let mut storage = self.storage.write();

324

		for (k, v) in ops {

325

			match v {

326

				Some(v) => storage.aux.insert(k, v),

327

				None => storage.aux.remove(&k),

328

};

329

330

331

332

333

impl<Block: BlockT + DeserializeOwned> HeaderBackend<Block> for Blockchain<Block> {

334

	fn header(

335

		&self,

336

		hash: Block::Hash,

337

	) -> sp_blockchain::Result<Option<<Block as BlockT>::Header>> {

338

		// First, try to get the header from local storage

339

		if let Some(header) = self

340

			.storage

341

			.read()

342

			.blocks

343

			.get(&hash)

344

			.map(|b| b.header().clone())

345

346

			return Ok(Some(header));

347

348

349

		// If not found in local storage, fetch from RPC client

350

		let header = self

351

			.rpc_client

352

			.block::<Block, _>(Some(hash))

353

			.ok()

354

			.flatten()

355

			.map(|full_block| {

356

				// Cache block header

357

				let block = full_block.block.clone();

358

				self.storage.write().blocks.insert(

359

					hash,

360

					StoredBlock::Full(block.clone(), full_block.justifications),

361

);

362

363

				block.header().clone()

364

});

365

366

		if header.is_none() {

367

			log::warn!(

368

				target: super::LAZY_LOADING_LOG_TARGET,

369

				"Expected block {:x?} to exist.",

370

				&hash

371

);

372

373

374

		Ok(header)

375

376

377

	fn info(&self) -> blockchain::Info<Block> {

378

		let storage = self.storage.read();

379

		blockchain::Info {

380

			best_hash: storage.best_hash,

381

			best_number: storage.best_number,

382

			genesis_hash: storage.genesis_hash,

383

			finalized_hash: storage.finalized_hash,

384

			finalized_number: storage.finalized_number,

385

			finalized_state: Some((storage.finalized_hash, storage.finalized_number)),

386

			number_leaves: storage.leaves.count(),

387

			block_gap: None,

388

389

390

391

	fn status(&self, hash: Block::Hash) -> sp_blockchain::Result<BlockStatus> {

392

		match self.storage.read().blocks.contains_key(&hash) {

393

			true => Ok(BlockStatus::InChain),

394

			false => Ok(BlockStatus::Unknown),

395

396

397

398

	fn number(&self, hash: Block::Hash) -> sp_blockchain::Result<Option<NumberFor<Block>>> {

399

		let number = match self.storage.read().blocks.get(&hash) {

400

			Some(block) => *block.header().number(),

401

			_ => match self.rpc_client.block::<Block, _>(Some(hash)) {

402

				Ok(Some(block)) => *block.block.header().number(),

403

				err => {

404

					return Err(sp_blockchain::Error::UnknownBlock(

405

						format!("Failed to fetch block number from RPC: {:?}", err).into(),

406

));

407

408

},

409

};

410

411

		Ok(Some(number))

412

413

414

	fn hash(

415

		&self,

416

		number: <<Block as BlockT>::Header as HeaderT>::Number,

417

	) -> sp_blockchain::Result<Option<Block::Hash>> {

418

		Ok(self.id(BlockId::Number(number)))

419

420

421

422

impl<Block: BlockT + DeserializeOwned> HeaderMetadata<Block> for Blockchain<Block> {

423

	type Error = sp_blockchain::Error;

424

425

	fn header_metadata(

426

		&self,

427

		hash: Block::Hash,

428

	) -> Result<CachedHeaderMetadata<Block>, Self::Error> {

429

		self.header(hash)?

430

			.map(|header| CachedHeaderMetadata::from(&header))

431

			.ok_or_else(|| {

432

				sp_blockchain::Error::UnknownBlock(format!("header not found: {}", hash))

433

})

434

435

436

	fn insert_header_metadata(&self, _hash: Block::Hash, _metadata: CachedHeaderMetadata<Block>) {

437

		// No need to implement.

438

		unimplemented!("insert_header_metadata")

439

440

	fn remove_header_metadata(&self, _hash: Block::Hash) {

441

		// No need to implement.

442

		unimplemented!("remove_header_metadata")

443

444

445

446

impl<Block: BlockT + DeserializeOwned> blockchain::Backend<Block> for Blockchain<Block> {

447

	fn body(

448

		&self,

449

		hash: Block::Hash,

450

	) -> sp_blockchain::Result<Option<Vec<<Block as BlockT>::Extrinsic>>> {

451

		// First, try to get the header from local storage

452

		if let Some(extrinsics) = self

453

			.storage

454

			.read()

455

			.blocks

456

			.get(&hash)

457

			.and_then(|b| b.extrinsics().map(|x| x.to_vec()))

458

459

			return Ok(Some(extrinsics));

460

461

		let extrinsics = self

462

			.rpc_client

463

			.block::<Block, Block::Hash>(Some(hash))

464

			.ok()

465

			.flatten()

466

			.map(|b| b.block.extrinsics().to_vec());

467

468

		Ok(extrinsics)

469

470

471

	fn justifications(&self, hash: Block::Hash) -> sp_blockchain::Result<Option<Justifications>> {

472

		Ok(self

473

			.storage

474

			.read()

475

			.blocks

476

			.get(&hash)

477

			.and_then(|b| b.justifications().cloned()))

478

479

480

	fn last_finalized(&self) -> sp_blockchain::Result<Block::Hash> {

481

		let last_finalized = self.storage.read().finalized_hash;

482

483

		Ok(last_finalized)

484

485

486

	fn leaves(&self) -> sp_blockchain::Result<Vec<Block::Hash>> {

487

		Ok(self.storage.read().leaves.hashes())

488

489

490

	fn children(&self, _parent_hash: Block::Hash) -> sp_blockchain::Result<Vec<Block::Hash>> {

491

		unimplemented!("Not supported by the `lazy-loading` backend.")

492

493

494

	fn indexed_transaction(&self, _hash: Block::Hash) -> sp_blockchain::Result<Option<Vec<u8>>> {

495

		unimplemented!("Not supported by the `lazy-loading` backend.")

496

497

498

	fn block_indexed_body(

499

		&self,

500

		_hash: Block::Hash,

501

	) -> sp_blockchain::Result<Option<Vec<Vec<u8>>>> {

502

		unimplemented!("Not supported by the `lazy-loading` backend.")

503

504

505

506

impl<Block: BlockT + DeserializeOwned> backend::AuxStore for Blockchain<Block> {

507

	fn insert_aux<

508

'a,

509

		'b: 'a,

510

		'c: 'a,

511

		I: IntoIterator<Item = &'a (&'c [u8], &'c [u8])>,

512

		D: IntoIterator<Item = &'a &'b [u8]>,

513

>(

514

		&self,

515

		insert: I,

516

		delete: D,

517

	) -> sp_blockchain::Result<()> {

518

		let mut storage = self.storage.write();

519

		for (k, v) in insert {

520

			storage.aux.insert(k.to_vec(), v.to_vec());

521

522

		for k in delete {

523

			storage.aux.remove(*k);

524

525

		Ok(())

526

527

528

	fn get_aux(&self, key: &[u8]) -> sp_blockchain::Result<Option<Vec<u8>>> {

529

		Ok(self.storage.read().aux.get(key).cloned())

530

531

532

533

pub struct BlockImportOperation<Block: BlockT> {

534

	pending_block: Option<PendingBlock<Block>>,

535

	old_state: ForkedLazyBackend<Block>,

536

	new_state: Option<BackendTransaction<HashingFor<Block>>>,

537

	aux: Vec<(Vec<u8>, Option<Vec<u8>>)>,

538

	storage_updates: StorageCollection,

539

	finalized_blocks: Vec<(Block::Hash, Option<Justification>)>,

540

	set_head: Option<Block::Hash>,

541

	pub(crate) before_fork: bool,

542

543

544

impl<Block: BlockT + DeserializeOwned> BlockImportOperation<Block> {

545

	fn apply_storage(

546

		&mut self,

547

		storage: Storage,

548

		commit: bool,

549

		state_version: StateVersion,

550

	) -> sp_blockchain::Result<Block::Hash> {

551

		use sp_state_machine::Backend;

552

		check_genesis_storage(&storage)?;

553

554

		let child_delta = storage.children_default.values().map(|child_content| {

555

556

				&child_content.child_info,

557

				child_content

558

					.data

559

					.iter()

560

					.map(|(k, v)| (k.as_ref(), Some(v.as_ref()))),

561

562

});

563

564

		let (root, transaction) = self.old_state.full_storage_root(

565

			storage

566

				.top

567

				.iter()

568

				.map(|(k, v)| (k.as_ref(), Some(v.as_ref()))),

569

			child_delta,

570

			state_version,

571

);

572

573

		if commit {

574

			self.new_state = Some(transaction);

575

			self.storage_updates = storage

576

				.top

577

				.iter()

578

				.map(|(k, v)| {

579

					if v.is_empty() {

580

						(k.clone(), None)

581

					} else {

582

						(k.clone(), Some(v.clone()))

583

584

})

585

				.collect();

586

587

		Ok(root)

588

589

590

591

impl<Block: BlockT + DeserializeOwned> backend::BlockImportOperation<Block>

592

	for BlockImportOperation<Block>

593

594

	type State = ForkedLazyBackend<Block>;

595

596

	fn state(&self) -> sp_blockchain::Result<Option<&Self::State>> {

597

		Ok(Some(&self.old_state))

598

599

600

	fn set_block_data(

601

		&mut self,

602

		header: <Block as BlockT>::Header,

603

		body: Option<Vec<<Block as BlockT>::Extrinsic>>,

604

		_indexed_body: Option<Vec<Vec<u8>>>,

605

		justifications: Option<Justifications>,

606

		state: NewBlockState,

607

	) -> sp_blockchain::Result<()> {

608

		assert!(

609

			self.pending_block.is_none(),

610

			"Only one block per operation is allowed"

611

);

612

		self.pending_block = Some(PendingBlock {

613

			block: StoredBlock::new(header, body, justifications),

614

			state,

615

});

616

		Ok(())

617

618

619

	fn update_db_storage(

620

		&mut self,

621

		update: BackendTransaction<HashingFor<Block>>,

622

	) -> sp_blockchain::Result<()> {

623

		self.new_state = Some(update);

624

		Ok(())

625

626

627

	fn set_genesis_state(

628

		&mut self,

629

		storage: Storage,

630

		commit: bool,

631

		state_version: StateVersion,

632

	) -> sp_blockchain::Result<Block::Hash> {

633

		self.apply_storage(storage, commit, state_version)

634

635

636

	fn reset_storage(

637

		&mut self,

638

		storage: Storage,

639

		state_version: StateVersion,

640

	) -> sp_blockchain::Result<Block::Hash> {

641

		self.apply_storage(storage, true, state_version)

642

643

644

	fn insert_aux<I>(&mut self, ops: I) -> sp_blockchain::Result<()>

645

	where

646

		I: IntoIterator<Item = (Vec<u8>, Option<Vec<u8>>)>,

647

648

		self.aux.append(&mut ops.into_iter().collect());

649

		Ok(())

650

651

652

	fn update_storage(

653

		&mut self,

654

		update: StorageCollection,

655

		_child_update: ChildStorageCollection,

656

	) -> sp_blockchain::Result<()> {

657

		self.storage_updates = update.clone();

658

		Ok(())

659

660

661

	fn mark_finalized(

662

		&mut self,

663

		hash: Block::Hash,

664

		justification: Option<Justification>,

665

	) -> sp_blockchain::Result<()> {

666

		self.finalized_blocks.push((hash, justification));

667

		Ok(())

668

669

670

	fn mark_head(&mut self, hash: Block::Hash) -> sp_blockchain::Result<()> {

671

		assert!(

672

			self.pending_block.is_none(),

673

			"Only one set block per operation is allowed"

674

);

675

		self.set_head = Some(hash);

676

		Ok(())

677

678

679

	fn update_transaction_index(

680

		&mut self,

681

		_index: Vec<IndexOperation>,

682

	) -> sp_blockchain::Result<()> {

683

		Ok(())

684

685

686

687

/// DB-backed patricia trie state, transaction type is an overlay of changes to commit.

688

pub type DbState<B> = TrieBackend<Arc<dyn sp_state_machine::Storage<HashingFor<B>>>, HashingFor<B>>;

689

690

/// A struct containing arguments for iterating over the storage.

691

#[derive(Default)]

692

pub struct RawIterArgs {

693

	/// The prefix of the keys over which to iterate.

694

	pub prefix: Option<Vec<u8>>,

695

696

	/// The prefix from which to start the iteration from.

697

///

698

	/// This is inclusive and the iteration will include the key which is specified here.

699

	pub start_at: Option<Vec<u8>>,

700

701

	/// If this is `true` then the iteration will *not* include

702

	/// the key specified in `start_at`, if there is such a key.

703

	pub start_at_exclusive: bool,

704

705

706

/// A raw iterator over the `BenchmarkingState`.

707

pub struct RawIter<Block: BlockT> {

708

	pub(crate) args: RawIterArgs,

709

	complete: bool,

710

	_phantom: PhantomData<Block>,

711

712

713

impl<Block: BlockT + DeserializeOwned> sp_state_machine::StorageIterator<HashingFor<Block>>

714

	for RawIter<Block>

715

716

	type Backend = ForkedLazyBackend<Block>;

717

	type Error = String;

718

719

	fn next_key(

720

		&mut self,

721

		backend: &Self::Backend,

722

	) -> Option<Result<sp_state_machine::StorageKey, Self::Error>> {

723

		use sp_state_machine::Backend;

724

725

		let remote_fetch =

726

			|key: Option<StorageKey>, start_key: Option<StorageKey>, block: Option<Block::Hash>| {

727

				let result = backend

728

					.rpc_client

729

					.storage_keys_paged(key, 5, start_key, block);

730

731

				match result {

732

					Ok(keys) => keys.first().map(|key| key.clone()),

733

					Err(err) => {

734

						log::trace!(

735

							target: super::LAZY_LOADING_LOG_TARGET,

736

							"Failed to fetch `next key` from RPC: {:?}",

737

err

738

);

739

740

						None

741

742

743

};

744

745

		let prefix = self.args.prefix.clone().map(|k| StorageKey(k));

746

		let start_key = self.args.start_at.clone().map(|k| StorageKey(k));

747

748

		let maybe_next_key = if backend.before_fork {

749

			remote_fetch(prefix, start_key, backend.block_hash)

750

		} else {

751

			let mut iter_args = sp_state_machine::backend::IterArgs::default();

752

			iter_args.prefix = self.args.prefix.as_ref().map(|b| b.as_slice());

753

			iter_args.start_at = self.args.start_at.as_ref().map(|b| b.as_slice());

754

			iter_args.start_at_exclusive = true;

755

			iter_args.stop_on_incomplete_database = true;

756

757

			let readable_db = backend.db.read();

758

			let next_storage_key = readable_db

759

				.raw_iter(iter_args)

760

				.map(|mut iter| iter.next_key(&readable_db))

761

				.map(|op| op.map(|result| result.ok()).flatten())

762

				.ok()

763

				.flatten();

764

765

			// IMPORTANT: free storage read lock

766

			drop(readable_db);

767

768

			let removed_key = start_key

769

				.clone()

770

				.or(prefix.clone())

771

				.map(|key| backend.removed_keys.read().contains_key(&key.0))

772

				.unwrap_or(false);

773

			if next_storage_key.is_none() && !removed_key {

774

				let maybe_next_key = remote_fetch(prefix, start_key, Some(backend.fork_block));

775

				match maybe_next_key {

776

					Some(key) if !backend.removed_keys.read().contains_key(&key) => Some(key),

777

					_ => None,

778

779

			} else {

780

				next_storage_key

781

782

};

783

784

		log::trace!(

785

			target: super::LAZY_LOADING_LOG_TARGET,

786

			"next_key: (prefix: {:?}, start_at: {:?}, next_key: {:?})",

787

			self.args.prefix.clone().map(|key| hex::encode(key)),

788

			self.args.start_at.clone().map(|key| hex::encode(key)),

789

			maybe_next_key.clone().map(|key| hex::encode(key))

790

);

791

792

		if let Some(next_key) = maybe_next_key {

793

			if self

794

				.args

795

				.prefix

796

				.clone()

797

				.map(|filter_key| next_key.starts_with(&filter_key))

798

				.unwrap_or(false)

799

800

				self.args.start_at = Some(next_key.clone());

801

				Some(Ok(next_key))

802

			} else {

803

				self.complete = true;

804

				None

805

806

		} else {

807

			self.complete = true;

808

			None

809

810

811

812

	fn next_pair(

813

		&mut self,

814

		backend: &Self::Backend,

815

	) -> Option<Result<(sp_state_machine::StorageKey, sp_state_machine::StorageValue), Self::Error>>

816

817

		use sp_state_machine::Backend;

818

819

		let remote_fetch =

820

			|key: Option<StorageKey>, start_key: Option<StorageKey>, block: Option<Block::Hash>| {

821

				let result = backend

822

					.rpc_client

823

					.storage_keys_paged(key, 5, start_key, block);

824

825

				match result {

826

					Ok(keys) => keys.first().map(|key| key.clone()),

827

					Err(err) => {

828

						log::trace!(

829

							target: super::LAZY_LOADING_LOG_TARGET,

830

							"Failed to fetch `next key` from RPC: {:?}",

831

err

832

);

833

834

						None

835

836

837

};

838

839

		let prefix = self.args.prefix.clone().map(|k| StorageKey(k));

840

		let start_key = self.args.start_at.clone().map(|k| StorageKey(k));

841

842

		let maybe_next_key = if backend.before_fork {

843

			remote_fetch(prefix, start_key, backend.block_hash)

844

		} else {

845

			let mut iter_args = sp_state_machine::backend::IterArgs::default();

846

			iter_args.prefix = self.args.prefix.as_ref().map(|b| b.as_slice());

847

			iter_args.start_at = self.args.start_at.as_ref().map(|b| b.as_slice());

848

			iter_args.start_at_exclusive = true;

849

			iter_args.stop_on_incomplete_database = true;

850

851

			let readable_db = backend.db.read();

852

			let next_storage_key = readable_db

853

				.raw_iter(iter_args)

854

				.map(|mut iter| iter.next_key(&readable_db))

855

				.map(|op| op.map(|result| result.ok()).flatten())

856

				.ok()

857

				.flatten();

858

859

			// IMPORTANT: free storage read lock

860

			drop(readable_db);

861

862

			let removed_key = start_key

863

				.clone()

864

				.or(prefix.clone())

865

				.map(|key| backend.removed_keys.read().contains_key(&key.0))

866

				.unwrap_or(false);

867

			if next_storage_key.is_none() && !removed_key {

868

				let maybe_next_key = remote_fetch(prefix, start_key, Some(backend.fork_block));

869

				match maybe_next_key {

870

					Some(key) if !backend.removed_keys.read().contains_key(&key) => Some(key),

871

					_ => None,

872

873

			} else {

874

				next_storage_key

875

876

};

877

878

		log::trace!(

879

			target: super::LAZY_LOADING_LOG_TARGET,

880

			"next_pair: (prefix: {:?}, start_at: {:?}, next_key: {:?})",

881

			self.args.prefix.clone().map(|key| hex::encode(key)),

882

			self.args.start_at.clone().map(|key| hex::encode(key)),

883

			maybe_next_key.clone().map(|key| hex::encode(key))

884

);

885

886

		let maybe_value = maybe_next_key

887

			.clone()

888

			.map(|key| (*backend).storage(key.as_slice()).ok())

889

			.flatten()

890

			.flatten();

891

892

		if let Some(next_key) = maybe_next_key {

893

			if self

894

				.args

895

				.prefix

896

				.clone()

897

				.map(|filter_key| next_key.starts_with(&filter_key))

898

				.unwrap_or(false)

899

900

				self.args.start_at = Some(next_key.clone());

901

902

				match maybe_value {

903

					Some(value) => Some(Ok((next_key, value))),

904

					_ => None,

905

906

			} else {

907

				self.complete = true;

908

				None

909

910

		} else {

911

			self.complete = true;

912

			None

913

914

915

916

	fn was_complete(&self) -> bool {

917

		self.complete

918

919

920

921

#[derive(Debug, Clone)]

922

pub struct ForkedLazyBackend<Block: BlockT> {

923

	rpc_client: Arc<RPC>,

924

	block_hash: Option<Block::Hash>,

925

	fork_block: Block::Hash,

926

	pub(crate) db: Arc<ReadWriteLock<sp_state_machine::InMemoryBackend<HashingFor<Block>>>>,

927

	pub(crate) removed_keys: Arc<ReadWriteLock<HashMap<Vec<u8>, ()>>>,

928

	before_fork: bool,

929

930

931

impl<Block: BlockT> ForkedLazyBackend<Block> {

932

	fn update_storage(&self, key: &[u8], value: &Option<Vec<u8>>) {

933

		if let Some(ref val) = value {

934

			let mut entries: HashMap<Option<ChildInfo>, StorageCollection> = Default::default();

935

			entries.insert(None, vec![(key.to_vec(), Some(val.clone()))]);

936

937

			self.db.write().insert(entries, StateVersion::V0);

938

939

940

941

942

impl<Block: BlockT + DeserializeOwned> sp_state_machine::Backend<HashingFor<Block>>

943

	for ForkedLazyBackend<Block>

944

945

	type Error = <DbState<Block> as sp_state_machine::Backend<HashingFor<Block>>>::Error;

946

	type TrieBackendStorage = PrefixedMemoryDB<HashingFor<Block>>;

947

	type RawIter = RawIter<Block>;

948

949

	fn storage(&self, key: &[u8]) -> Result<Option<sp_state_machine::StorageValue>, Self::Error> {

950

		let remote_fetch = |block: Option<Block::Hash>| {

951

			let result = self.rpc_client.storage(StorageKey(key.to_vec()), block);

952

953

			match result {

954

				Ok(data) => data.map(|v| v.0),

955

				Err(err) => {

956

					log::debug!(

957

						target: super::LAZY_LOADING_LOG_TARGET,

958

						"Failed to fetch storage from live network: {:?}",

959

err

960

);

961

					None

962

963

964

};

965

966

		if self.before_fork {

967

			return Ok(remote_fetch(self.block_hash));

968

969

970

		let readable_db = self.db.read();

971

		let maybe_storage = readable_db.storage(key);

972

		let value = match maybe_storage {

973

			Ok(Some(data)) => Some(data),

974

			_ if !self.removed_keys.read().contains_key(key) => {

975

				let result = remote_fetch(Some(self.fork_block));

976

977

				// Cache state

978

				drop(readable_db);

979

				self.update_storage(key, &result);

980

981

				result

982

983

			_ => None,

984

};

985

986

		Ok(value)

987

988

989

	fn storage_hash(

990

		&self,

991

		key: &[u8],

992

	) -> Result<Option<<HashingFor<Block> as sp_core::Hasher>::Out>, Self::Error> {

993

		let remote_fetch = |block: Option<Block::Hash>| {

994

			let result = self

995

				.rpc_client

996

				.storage_hash(StorageKey(key.to_vec()), block);

997

998

			match result {

999

				Ok(hash) => Ok(hash),

1000

				Err(err) => Err(format!("Failed to fetch storage hash from RPC: {:?}", err).into()),

1001

1002

};

1003

1004

		if self.before_fork {

1005

			return remote_fetch(self.block_hash);

1006

1007

1008

		let storage_hash = self.db.read().storage_hash(key);

1009

		match storage_hash {

1010

			Ok(Some(hash)) => Ok(Some(hash)),

1011

			_ if !self.removed_keys.read().contains_key(key) => remote_fetch(Some(self.fork_block)),

1012

			_ => Ok(None),

1013

1014

1015

1016

	fn closest_merkle_value(

1017

		&self,

1018

		_key: &[u8],

1019

	) -> Result<

1020

		Option<sp_trie::MerkleValue<<HashingFor<Block> as sp_core::Hasher>::Out>>,

1021

		Self::Error,

1022

> {

1023

		unimplemented!("closest_merkle_value: unsupported feature for lazy loading")

1024

1025

1026

	fn child_closest_merkle_value(

1027

		&self,

1028

		_child_info: &sp_storage::ChildInfo,

1029

		_key: &[u8],

1030

	) -> Result<

1031

		Option<sp_trie::MerkleValue<<HashingFor<Block> as sp_core::Hasher>::Out>>,

1032

		Self::Error,

1033

> {

1034

		unimplemented!("child_closest_merkle_value: unsupported feature for lazy loading")

1035

1036

1037

	fn child_storage(

1038

		&self,

1039

		_child_info: &sp_storage::ChildInfo,

1040

		_key: &[u8],

1041

	) -> Result<Option<sp_state_machine::StorageValue>, Self::Error> {

1042

		unimplemented!("child_storage: unsupported feature for lazy loading");

1043

1044

1045

	fn child_storage_hash(

1046

		&self,

1047

		_child_info: &sp_storage::ChildInfo,

1048

		_key: &[u8],

1049

	) -> Result<Option<<HashingFor<Block> as sp_core::Hasher>::Out>, Self::Error> {

1050

		unimplemented!("child_storage_hash: unsupported feature for lazy loading");

1051

1052

1053

	fn next_storage_key(

1054

		&self,

1055

		key: &[u8],

1056

	) -> Result<Option<sp_state_machine::StorageKey>, Self::Error> {

1057

		let remote_fetch = |block: Option<Block::Hash>| {

1058

			let start_key = Some(StorageKey(key.to_vec()));

1059

			let result = self

1060

				.rpc_client

1061

				.storage_keys_paged(start_key.clone(), 2, None, block);

1062

1063

			match result {

1064

				Ok(keys) => keys.last().cloned(),

1065

				Err(err) => {

1066

					log::trace!(

1067

						target: super::LAZY_LOADING_LOG_TARGET,

1068

						"Failed to fetch `next storage key` from RPC: {:?}",

1069

err

1070

);

1071

1072

					None

1073

1074

1075

};

1076

1077

		let maybe_next_key = if self.before_fork {

1078

			remote_fetch(self.block_hash)

1079

		} else {

1080

			let next_storage_key = self.db.read().next_storage_key(key);

1081

			match next_storage_key {

1082

				Ok(Some(key)) => Some(key),

1083

				_ if !self.removed_keys.read().contains_key(key) => {

1084

					remote_fetch(Some(self.fork_block))

1085

1086

				_ => None,

1087

1088

};

1089

1090

		log::trace!(

1091

			target: super::LAZY_LOADING_LOG_TARGET,

1092

			"next_storage_key: (key: {:?}, next_key: {:?})",

1093

			hex::encode(key),

1094

			maybe_next_key.clone().map(|key| hex::encode(key))

1095

);

1096

1097

		Ok(maybe_next_key)

1098

1099

1100

	fn next_child_storage_key(

1101

		&self,

1102

		_child_info: &sp_storage::ChildInfo,

1103

		_key: &[u8],

1104

	) -> Result<Option<sp_state_machine::StorageKey>, Self::Error> {

1105

		unimplemented!("next_child_storage_key: unsupported feature for lazy loading");

1106

1107

1108

	fn storage_root<'a>(

1109

		&self,

1110

		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,

1111

		state_version: StateVersion,

1112

	) -> (

1113

		<HashingFor<Block> as sp_core::Hasher>::Out,

1114

		BackendTransaction<HashingFor<Block>>,

1115

1116

	where

1117

		<HashingFor<Block> as sp_core::Hasher>::Out: Ord,

1118

1119

		self.db.read().storage_root(delta, state_version)

1120

1121

1122

	fn child_storage_root<'a>(

1123

		&self,

1124

		_child_info: &sp_storage::ChildInfo,

1125

		_delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,

1126

		_state_version: StateVersion,

1127

	) -> (

1128

		<HashingFor<Block> as sp_core::Hasher>::Out,

1129

		bool,

1130

		BackendTransaction<HashingFor<Block>>,

1131

1132

	where

1133

		<HashingFor<Block> as sp_core::Hasher>::Out: Ord,

1134

1135

		unimplemented!("child_storage_root: unsupported in lazy loading")

1136

1137

1138

	fn raw_iter(&self, args: sp_state_machine::IterArgs) -> Result<Self::RawIter, Self::Error> {

1139

		let mut clone: RawIterArgs = Default::default();

1140

		clone.start_at_exclusive = args.start_at_exclusive.clone();

1141

		clone.prefix = args.prefix.map(|v| v.to_vec());

1142

		clone.start_at = args.start_at.map(|v| v.to_vec());

1143

1144

		Ok(RawIter::<Block> {

1145

			args: clone,

1146

			complete: false,

1147

			_phantom: Default::default(),

1148

})

1149

1150

1151

	fn register_overlay_stats(&self, stats: &sp_state_machine::StateMachineStats) {

1152

		self.db.read().register_overlay_stats(stats)

1153

1154

1155

	fn usage_info(&self) -> sp_state_machine::UsageInfo {

1156

		self.db.read().usage_info()

1157

1158

1159

1160

impl<B: BlockT> sp_state_machine::backend::AsTrieBackend<HashingFor<B>> for ForkedLazyBackend<B> {

1161

	type TrieBackendStorage = PrefixedMemoryDB<HashingFor<B>>;

1162

1163

	fn as_trie_backend(

1164

		&self,

1165

	) -> &sp_state_machine::TrieBackend<Self::TrieBackendStorage, HashingFor<B>> {

1166

		unimplemented!("`as_trie_backend` is not supported in lazy loading mode.")

1167

1168

1169

1170

/// Lazy loading (In-memory) backend. Keeps all states and blocks in memory.

1171

pub struct Backend<Block: BlockT> {

1172

	pub(crate) rpc_client: Arc<RPC>,

1173

	states: ReadWriteLock<HashMap<Block::Hash, ForkedLazyBackend<Block>>>,

1174

	pub(crate) blockchain: Blockchain<Block>,

1175

	import_lock: parking_lot::RwLock<()>,

1176

	pinned_blocks: ReadWriteLock<HashMap<Block::Hash, i64>>,

1177

	pub(crate) fork_checkpoint: Block::Header,

1178

1179

1180

impl<Block: BlockT + DeserializeOwned> Backend<Block> {

1181

	fn new(rpc_client: Arc<RPC>, fork_checkpoint: Block::Header) -> Self {

1182

		Backend {

1183

			rpc_client: rpc_client.clone(),

1184

			states: Default::default(),

1185

			blockchain: Blockchain::new(rpc_client),

1186

			import_lock: Default::default(),

1187

			pinned_blocks: Default::default(),

1188

			fork_checkpoint,

1189

1190

1191

1192

1193

impl<Block: BlockT + DeserializeOwned> backend::AuxStore for Backend<Block> {

1194

	fn insert_aux<

1195

'a,

1196

		'b: 'a,

1197

		'c: 'a,

1198

		I: IntoIterator<Item = &'a (&'c [u8], &'c [u8])>,

1199

		D: IntoIterator<Item = &'a &'b [u8]>,

1200

>(

1201

		&self,

1202

		_insert: I,

1203

		_delete: D,

1204

	) -> sp_blockchain::Result<()> {

1205

		unimplemented!("`insert_aux` is not supported in lazy loading mode.")

1206

1207

1208

	fn get_aux(&self, _key: &[u8]) -> sp_blockchain::Result<Option<Vec<u8>>> {

1209

		unimplemented!("`get_aux` is not supported in lazy loading mode.")

1210

1211

1212

1213

impl<Block: BlockT + DeserializeOwned> backend::Backend<Block> for Backend<Block> {

1214

	type BlockImportOperation = BlockImportOperation<Block>;

1215

	type Blockchain = Blockchain<Block>;

1216

	type State = ForkedLazyBackend<Block>;

1217

	type OffchainStorage = InMemOffchainStorage;

1218

1219

	fn begin_operation(&self) -> sp_blockchain::Result<Self::BlockImportOperation> {

1220

		let old_state = self.state_at(Default::default())?;

1221

		Ok(BlockImportOperation {

1222

			pending_block: None,

1223

			old_state,

1224

			new_state: None,

1225

			aux: Default::default(),

1226

			storage_updates: Default::default(),

1227

			finalized_blocks: Default::default(),

1228

			set_head: None,

1229

			before_fork: false,

1230

})

1231

1232

1233

	fn begin_state_operation(

1234

		&self,

1235

		operation: &mut Self::BlockImportOperation,

1236

		block: Block::Hash,

1237

	) -> sp_blockchain::Result<()> {

1238

		operation.old_state = self.state_at(block)?;

1239

		Ok(())

1240

1241

1242

	fn commit_operation(&self, operation: Self::BlockImportOperation) -> sp_blockchain::Result<()> {

1243

		if !operation.finalized_blocks.is_empty() {

1244

			for (block, justification) in operation.finalized_blocks {

1245

				self.blockchain.finalize_header(block, justification)?;

1246

1247

1248

1249

		if let Some(pending_block) = operation.pending_block {

1250

			let old_state = &operation.old_state;

1251

			let (header, body, justification) = pending_block.block.into_inner();

1252

			let hash = header.hash();

1253

1254

			let new_removed_keys = old_state.removed_keys.clone();

1255

			for (key, value) in operation.storage_updates.clone() {

1256

				if value.is_some() {

1257

					new_removed_keys.write().remove(&key.clone());

1258

				} else {

1259

					new_removed_keys.write().insert(key.clone(), ());

1260

1261

1262

1263

			let new_db = old_state.db.clone();

1264

			new_db.write().insert(

1265

				vec![(None::<ChildInfo>, operation.storage_updates)],

1266

				StateVersion::V0,

1267

);

1268

			let new_state = ForkedLazyBackend {

1269

				rpc_client: self.rpc_client.clone(),

1270

				block_hash: Some(hash.clone()),

1271

				fork_block: self.fork_checkpoint.hash(),

1272

				db: new_db,

1273

				removed_keys: new_removed_keys,

1274

				before_fork: operation.before_fork,

1275

};

1276

			self.states.write().insert(hash, new_state);

1277

1278

			self.blockchain

1279

				.insert(hash, header, justification, body, pending_block.state)?;

1280

1281

1282

		if !operation.aux.is_empty() {

1283

			self.blockchain.write_aux(operation.aux);

1284

1285

1286

		if let Some(set_head) = operation.set_head {

1287

			self.blockchain.set_head(set_head)?;

1288

1289

1290

		Ok(())

1291

1292

1293

	fn finalize_block(

1294

		&self,

1295

		hash: Block::Hash,

1296

		justification: Option<Justification>,

1297

	) -> sp_blockchain::Result<()> {

1298

		self.blockchain.finalize_header(hash, justification)

1299

1300

1301

	fn append_justification(

1302

		&self,

1303

		hash: Block::Hash,

1304

		justification: Justification,

1305

	) -> sp_blockchain::Result<()> {

1306

		self.blockchain.append_justification(hash, justification)

1307

1308

1309

	fn blockchain(&self) -> &Self::Blockchain {

1310

		&self.blockchain

1311

1312

1313

	fn usage_info(&self) -> Option<UsageInfo> {

1314

		None

1315

1316

1317

	fn offchain_storage(&self) -> Option<Self::OffchainStorage> {

1318

		None

1319

1320

1321

	fn state_at(&self, hash: Block::Hash) -> sp_blockchain::Result<Self::State> {

1322

		if hash == Default::default() {

1323

			return Ok(ForkedLazyBackend::<Block> {

1324

				rpc_client: self.rpc_client.clone(),

1325

				block_hash: Some(hash),

1326

				fork_block: self.fork_checkpoint.hash(),

1327

				db: Default::default(),

1328

				removed_keys: Default::default(),

1329

				before_fork: true,

1330

});

1331

1332

1333

		let (backend, should_write) = self

1334

			.states

1335

			.read()

1336

			.get(&hash)

1337

			.cloned()

1338

			.map(|state| (state, false))

1339

			.unwrap_or_else(|| {

1340

				let header: Block::Header = self

1341

					.rpc_client

1342

					.header::<Block>(Some(hash))

1343

					.ok()

1344

					.flatten()

1345

					.expect("block header");

1346

1347

				let checkpoint = self.fork_checkpoint.clone();

1348

				let state = if header.number().gt(checkpoint.number()) {

1349

					let parent = self.state_at(*header.parent_hash()).ok();

1350

1351

					ForkedLazyBackend::<Block> {

1352

						rpc_client: self.rpc_client.clone(),

1353

						block_hash: Some(hash),

1354

						fork_block: checkpoint.hash(),

1355

						db: parent.clone().map_or(Default::default(), |p| p.db),

1356

						removed_keys: parent.map_or(Default::default(), |p| p.removed_keys),

1357

						before_fork: false,

1358

1359

				} else {

1360

					ForkedLazyBackend::<Block> {

1361

						rpc_client: self.rpc_client.clone(),

1362

						block_hash: Some(hash),

1363

						fork_block: checkpoint.hash(),

1364

						db: Default::default(),

1365

						removed_keys: Default::default(),

1366

						before_fork: true,

1367

1368

};

1369

1370

				(state, true)

1371

});

1372

1373

		if should_write {

1374

			self.states.write().insert(hash, backend.clone());

1375

1376

1377

		Ok(backend)

1378

1379

1380

	fn revert(

1381

		&self,

1382

		_n: NumberFor<Block>,

1383

		_revert_finalized: bool,

1384

	) -> sp_blockchain::Result<(NumberFor<Block>, HashSet<Block::Hash>)> {

1385

		Ok((Zero::zero(), HashSet::new()))

1386

1387

1388

	fn remove_leaf_block(&self, _hash: Block::Hash) -> sp_blockchain::Result<()> {

1389

		Ok(())

1390

1391

1392

	fn get_import_lock(&self) -> &parking_lot::RwLock<()> {

1393

		&self.import_lock

1394

1395

1396

	fn requires_full_sync(&self) -> bool {

1397

		false

1398

1399

1400

	fn pin_block(&self, hash: <Block as BlockT>::Hash) -> blockchain::Result<()> {

1401

		let mut blocks = self.pinned_blocks.write();

1402

		*blocks.entry(hash).or_default() += 1;

1403

		Ok(())

1404

1405

1406

	fn unpin_block(&self, hash: <Block as BlockT>::Hash) {

1407

		let mut blocks = self.pinned_blocks.write();

1408

		blocks

1409

			.entry(hash)

1410

			.and_modify(|counter| *counter -= 1)

1411

			.or_insert(-1);

1412

1413

1414

1415

impl<Block: BlockT + DeserializeOwned> backend::LocalBackend<Block> for Backend<Block> {}

1416

1417

/// Check that genesis storage is valid.

1418

pub fn check_genesis_storage(storage: &Storage) -> sp_blockchain::Result<()> {

1419

	if storage

1420

		.top

1421

		.iter()

1422

		.any(|(k, _)| well_known_keys::is_child_storage_key(k))

1423

1424

		return Err(sp_blockchain::Error::InvalidState);

1425

1426

1427

	if storage

1428

		.children_default

1429

		.keys()

1430

		.any(|child_key| !well_known_keys::is_child_storage_key(child_key))

1431

1432

		return Err(sp_blockchain::Error::InvalidState);

1433

1434

1435

	Ok(())

1436

1437

1438

#[derive(Debug, Clone)]

1439

pub struct RPC {

1440

	http_client: HttpClient,

1441

	delay_between_requests_ms: u64,

1442

	max_retries_per_request: usize,

1443

	counter: Arc<ReadWriteLock<u64>>,

1444

1445

1446

impl RPC {

1447

	pub fn new(

1448

		http_client: HttpClient,

1449

		delay_between_requests_ms: u64,

1450

		max_retries_per_request: usize,

1451

	) -> Self {

1452

		Self {

1453

			http_client,

1454

			delay_between_requests_ms,

1455

			max_retries_per_request,

1456

			counter: Default::default(),

1457

1458

1459

	pub fn system_chain(&self) -> Result<String, jsonrpsee::core::ClientError> {

1460

		let request = &|| {

1461

			substrate_rpc_client::SystemApi::<H256, BlockNumber>::system_chain(&self.http_client)

1462

};

1463

1464

		self.block_on(request)

1465

1466

1467

	pub fn system_properties(

1468

		&self,

1469

	) -> Result<sc_chain_spec::Properties, jsonrpsee::core::ClientError> {

1470

		let request = &|| {

1471

			substrate_rpc_client::SystemApi::<H256, BlockNumber>::system_properties(

1472

				&self.http_client,

1473

1474

};

1475

1476

		self.block_on(request)

1477

1478

1479

	pub fn system_name(&self) -> Result<String, jsonrpsee::core::ClientError> {

1480

		let request = &|| {

1481

			substrate_rpc_client::SystemApi::<H256, BlockNumber>::system_name(&self.http_client)

1482

};

1483

1484

		self.block_on(request)

1485

1486

1487

	pub fn block<Block, Hash: Clone>(

1488

		&self,

1489

		hash: Option<Hash>,

1490

	) -> Result<Option<SignedBlock<Block>>, jsonrpsee::core::ClientError>

1491

	where

1492

		Block: BlockT + DeserializeOwned,

1493

		Hash: 'static + Send + Sync + sp_runtime::Serialize + DeserializeOwned,

1494

1495

		let request = &|| {

1496

			substrate_rpc_client::ChainApi::<

1497

				BlockNumber,

1498

				Hash,

1499

				Block::Header,

1500

				SignedBlock<Block>,

1501

			>::block(&self.http_client, hash.clone())

1502

};

1503

1504

		self.block_on(request)

1505

1506

1507

	pub fn block_hash<Block: BlockT + DeserializeOwned>(

1508

		&self,

1509

		block_number: Option<BlockNumber>,

1510

	) -> Result<Option<Block::Hash>, jsonrpsee::core::ClientError> {

1511

		let request = &|| {

1512

			substrate_rpc_client::ChainApi::<

1513

				BlockNumber,

1514

				Block::Hash,

1515

				Block::Header,

1516

				SignedBlock<Block>,

1517

			>::block_hash(

1518

				&self.http_client,

1519

				block_number.map(|n| ListOrValue::Value(NumberOrHex::Number(n.into()))),

1520

1521

};

1522

1523

		self.block_on(request).map(|ok| match ok {

1524

			ListOrValue::List(v) => v.get(0).map_or(None, |some| *some),

1525

			ListOrValue::Value(v) => v,

1526

})

1527

1528

1529

	pub fn header<Block: BlockT + DeserializeOwned>(

1530

		&self,

1531

		hash: Option<Block::Hash>,

1532

	) -> Result<Option<Block::Header>, jsonrpsee::core::ClientError> {

1533

		let request = &|| {

1534

			substrate_rpc_client::ChainApi::<

1535

				BlockNumber,

1536

				Block::Hash,

1537

				Block::Header,

1538

				SignedBlock<Block>,

1539

			>::header(&self.http_client, hash)

1540

};

1541

1542

		self.block_on(request)

1543

1544

1545

	pub fn storage_hash<

1546

		Hash: 'static + Clone + Sync + Send + DeserializeOwned + sp_runtime::Serialize,

1547

>(

1548

		&self,

1549

		key: StorageKey,

1550

		at: Option<Hash>,

1551

	) -> Result<Option<Hash>, jsonrpsee::core::ClientError> {

1552

		let request = &|| {

1553

			substrate_rpc_client::StateApi::<Hash>::storage_hash(

1554

				&self.http_client,

1555

				key.clone(),

1556

				at.clone(),

1557

1558

};

1559

1560

		self.block_on(request)

1561

1562

1563

	pub fn storage<

1564

		Hash: 'static + Clone + Sync + Send + DeserializeOwned + sp_runtime::Serialize + core::fmt::Debug,

1565

>(

1566

		&self,

1567

		key: StorageKey,

1568

		at: Option<Hash>,

1569

	) -> Result<Option<StorageData>, jsonrpsee::core::ClientError> {

1570

		let request = &|| {

1571

			substrate_rpc_client::StateApi::<Hash>::storage(

1572

				&self.http_client,

1573

				key.clone(),

1574

				at.clone(),

1575

1576

};

1577

1578

		self.block_on(request)

1579

1580

1581

	pub fn storage_keys_paged<

1582

		Hash: 'static + Clone + Sync + Send + DeserializeOwned + sp_runtime::Serialize,

1583

>(

1584

		&self,

1585

		key: Option<StorageKey>,

1586

		count: u32,

1587

		start_key: Option<StorageKey>,

1588

		at: Option<Hash>,

1589

	) -> Result<Vec<sp_state_machine::StorageKey>, jsonrpsee::core::ClientError> {

1590

		let request = &|| {

1591

			substrate_rpc_client::StateApi::<Hash>::storage_keys_paged(

1592

				&self.http_client,

1593

				key.clone(),

1594

				count.clone(),

1595

				start_key.clone(),

1596

				at.clone(),

1597

1598

};

1599

		let result = self.block_on(request);

1600

1601

		match result {

1602

			Ok(result) => Ok(result.iter().map(|item| item.0.clone()).collect()),

1603

			Err(err) => Err(err),

1604

1605

1606

1607

	pub fn query_storage_at<

1608

		Hash: 'static + Clone + Sync + Send + DeserializeOwned + sp_runtime::Serialize,

1609

>(

1610

		&self,

1611

		keys: Vec<StorageKey>,

1612

		from_block: Option<Hash>,

1613

	) -> Result<Vec<(StorageKey, Option<StorageData>)>, jsonrpsee::core::ClientError> {

1614

		let request = &|| {

1615

			substrate_rpc_client::StateApi::<Hash>::query_storage_at(

1616

				&self.http_client,

1617

				keys.clone(),

1618

				from_block.clone(),

1619

1620

};

1621

		let result = self.block_on(request);

1622

1623

		match result {

1624

			Ok(result) => Ok(result

1625

				.iter()

1626

				.flat_map(|item| item.changes.clone())

1627

				.collect()),

1628

			Err(err) => Err(err),

1629

1630

1631

1632

	fn block_on<F, T, E>(&self, f: &dyn Fn() -> F) -> Result<T, E>

1633

	where

1634

		F: Future<Output = Result<T, E>>,

1635

1636

		use tokio::runtime::Handle;

1637

1638

		tokio::task::block_in_place(move || {

1639

			Handle::current().block_on(async move {

1640

				let delay_between_requests = Duration::from_millis(self.delay_between_requests_ms);

1641

1642

				let start = std::time::Instant::now();

1643

				self.counter.write().add_assign(1);

1644

				log::debug!(

1645

					target: super::LAZY_LOADING_LOG_TARGET,

1646

					"Sending request: {}",

1647

					self.counter.read()

1648

);

1649

1650

				// Explicit request delay, to avoid getting 429 errors

1651

				let _ = tokio::time::sleep(delay_between_requests).await;

1652

1653

				// Retry request in case of failure

1654

				// The maximum number of retries is specified by `self.max_retries_per_request`

1655

				let retry_strategy =

1656

					FixedInterval::new(delay_between_requests).take(self.max_retries_per_request);

1657

				let result = Retry::spawn(retry_strategy, f).await;

1658

1659

				log::debug!(

1660

					target: super::LAZY_LOADING_LOG_TARGET,

1661

					"Completed request (id: {}, successful: {}, elapsed_time: {:?})",

1662

					self.counter.read(),

1663

					result.is_ok(),

1664

					start.elapsed()

1665

);

1666

1667

				result

1668

})

1669

})

1670

1671

1672

1673

/// Create an instance of a lazy loading memory backend.

1674

pub fn new_lazy_loading_backend<Block>(

1675

	config: &mut Configuration,

1676

	lazy_loading_config: &LazyLoadingConfig,

1677

) -> Result<Arc<Backend<Block>>, Error>

1678

where

1679

	Block: BlockT + DeserializeOwned,

1680

	Block::Hash: From<H256>,

1681

1682

	let uri: String = lazy_loading_config.state_rpc.clone().into();

1683

1684

	let http_client = jsonrpsee::http_client::HttpClientBuilder::default()

1685

		.max_request_size(u32::MAX)

1686

		.max_response_size(u32::MAX)

1687

		.request_timeout(Duration::from_secs(10))

1688

		.build(uri)

1689

		.map_err(|e| {

1690

			sp_blockchain::Error::Backend(

1691

				format!("failed to build http client: {:?}", e).to_string(),

1692

1693

		})?;

1694

1695

	let rpc = RPC::new(http_client, 100, 10);

1696

	let block_hash = lazy_loading_config

1697

		.from_block

1698

		.map(|block| Into::<Block::Hash>::into(block));

1699

	let checkpoint: Block = rpc

1700

		.block::<Block, _>(block_hash)

1701

		.ok()

1702

		.flatten()

1703

		.expect("Fetching fork checkpoint")

1704

		.block;

1705

1706

	let backend = Arc::new(Backend::new(Arc::new(rpc), checkpoint.header().clone()));

1707

1708

	let chain_name = backend

1709

		.rpc_client

1710

		.system_chain()

1711

		.expect("Should fetch chain id");

1712

	let chain_properties = backend

1713

		.rpc_client

1714

		.system_properties()

1715

		.expect("Should fetch chain properties");

1716

1717

	let spec_builder = chain_spec::test_spec::lazy_loading_spec_builder(Default::default())

1718

		.with_name(chain_name.as_str())

1719

		.with_properties(chain_properties);

1720

	config.chain_spec = Box::new(spec_builder.build());

1721

1722

	let base_overrides =

1723

		state_overrides::base_state_overrides(lazy_loading_config.runtime_override.clone());

1724

	let custom_overrides = if let Some(path) = lazy_loading_config.state_overrides_path.clone() {

1725

		state_overrides::read(path)?

1726

	} else {

1727

		Default::default()

1728

};

1729

	let state_overrides: Vec<(Vec<u8>, Vec<u8>)> = [base_overrides, custom_overrides]

1730

		.concat()

1731

		.iter()

1732

		.map(|entry| match entry {

1733

			StateEntry::Concrete(v) => {

1734

				let key = [

1735

					&twox_128(v.pallet.as_bytes()),

1736

					&twox_128(v.storage.as_bytes()),

1737

					v.key.clone().unwrap_or(Vec::new()).as_slice(),

1738

1739

				.concat();

1740

1741

				(key, v.value.clone())

1742

1743

			StateEntry::Raw(raw) => (raw.key.clone(), raw.value.clone()),

1744

})

1745

		.collect();

1746

1747

	let _ = helpers::produce_genesis_block(backend.clone());

1748

1749

	// Produce first block after the fork

1750

	let _ = helpers::produce_first_block(backend.clone(), checkpoint, state_overrides)?;

1751

1752

	Ok(backend)

1753