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) => {

146

				let block_hash = self.storage.read().hashes.get(&n).cloned();

147

				match block_hash {

148

					None => {

149

						let block_hash =

150

							self.rpc_client.block_hash::<Block>(Some(n)).ok().flatten();

151

152

						block_hash.clone().map(|h| {

153

							self.storage.write().hashes.insert(n, h);

154

});

155

156

						block_hash

157

158

					block_hash => block_hash,

159

160

161

162

163

164

	/// Create new in-memory blockchain storage.

165

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

166

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

167

			blocks: HashMap::new(),

168

			hashes: HashMap::new(),

169

			best_hash: Default::default(),

170

			best_number: Zero::zero(),

171

			finalized_hash: Default::default(),

172

			finalized_number: Zero::zero(),

173

			genesis_hash: Default::default(),

174

			header_cht_roots: HashMap::new(),

175

			leaves: LeafSet::new(),

176

			aux: HashMap::new(),

177

		}));

178

		Blockchain {

179

			rpc_client,

180

			storage,

181

182

183

184

	/// Insert a block header and associated data.

185

	pub fn insert(

186

		&self,

187

		hash: Block::Hash,

188

		header: <Block as BlockT>::Header,

189

		justifications: Option<Justifications>,

190

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

191

		new_state: NewBlockState,

192

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

193

		let number = *header.number();

194

		if new_state.is_best() {

195

			self.apply_head(&header)?;

196

197

198

199

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

200

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

201

			storage

202

				.blocks

203

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

204

205

			if let NewBlockState::Final = new_state {

206

				storage.finalized_hash = hash;

207

				storage.finalized_number = number;

208

209

210

			if number == Zero::zero() {

211

				storage.genesis_hash = hash;

212

213

214

215

		Ok(())

216

217

218

	/// Get total number of blocks.

219

	pub fn blocks_count(&self) -> usize {

220

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

221

222

		log::debug!(

223

			target: super::LAZY_LOADING_LOG_TARGET,

224

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

225

			count

226

);

227

228

		count

229

230

231

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

232

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

233

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

234

		if ptr::eq(self, other) {

235

			return true;

236

237

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

238

239

240

	/// Compare canonical chain to other canonical chain.

241

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

242

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

243

		if ptr::eq(self, other) {

244

			return true;

245

246

		let this = self.storage.read();

247

		let other = other.storage.read();

248

		this.hashes == other.hashes

249

			&& this.best_hash == other.best_hash

250

			&& this.best_number == other.best_number

251

			&& this.genesis_hash == other.genesis_hash

252

253

254

	/// Insert header CHT root.

255

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

256

		self.storage

257

			.write()

258

			.header_cht_roots

259

			.insert(block, cht_root);

260

261

262

	/// Set an existing block as head.

263

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

264

		let header = self

265

			.header(hash)?

266

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

267

268

		self.apply_head(&header)

269

270

271

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

272

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

273

274

		let hash = header.hash();

275

		let number = header.number();

276

277

		storage.best_hash = hash;

278

		storage.best_number = *number;

279

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

280

281

		Ok(())

282

283

284

	fn finalize_header(

285

		&self,

286

		block: Block::Hash,

287

		justification: Option<Justification>,

288

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

289

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

290

		storage.finalized_hash = block;

291

292

		if justification.is_some() {

293

			let block = storage

294

				.blocks

295

				.get_mut(&block)

296

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

297

298

			let block_justifications = match block {

299

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

300

};

301

302

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

303

304

305

		Ok(())

306

307

308

	fn append_justification(

309

		&self,

310

		hash: Block::Hash,

311

		justification: Justification,

312

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

313

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

314

315

		let block = storage

316

			.blocks

317

			.get_mut(&hash)

318

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

319

320

		let block_justifications = match block {

321

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

322

};

323

324

		if let Some(stored_justifications) = block_justifications {

325

			if !stored_justifications.append(justification) {

326

				return Err(sp_blockchain::Error::BadJustification(

327

					"Duplicate consensus engine ID".into(),

328

));

329

330

		} else {

331

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

332

};

333

334

		Ok(())

335

336

337

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

338

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

339

		for (k, v) in ops {

340

			match v {

341

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

342

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

343

};

344

345

346

347

348

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

349

	fn header(

350

		&self,

351

		hash: Block::Hash,

352

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

353

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

354

		if let Some(header) = self

355

			.storage

356

			.read()

357

			.blocks

358

			.get(&hash)

359

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

360

361

			return Ok(Some(header));

362

363

364

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

365

		let header = self

366

			.rpc_client

367

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

368

			.ok()

369

			.flatten()

370

			.map(|full_block| {

371

				// Cache block header

372

				let block = full_block.block.clone();

373

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

374

					hash,

375

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

376

);

377

378

				block.header().clone()

379

});

380

381

		if header.is_none() {

382

			log::warn!(

383

				target: super::LAZY_LOADING_LOG_TARGET,

384

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

385

				&hash

386

);

387

388

389

		Ok(header)

390

391

392

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

393

		let storage = self.storage.read();

394

		blockchain::Info {

395

			best_hash: storage.best_hash,

396

			best_number: storage.best_number,

397

			genesis_hash: storage.genesis_hash,

398

			finalized_hash: storage.finalized_hash,

399

			finalized_number: storage.finalized_number,

400

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

401

			number_leaves: storage.leaves.count(),

402

			block_gap: None,

403

404

405

406

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

407

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

408

			true => Ok(BlockStatus::InChain),

409

			false => Ok(BlockStatus::Unknown),

410

411

412

413

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

414

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

415

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

416

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

417

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

418

				err => {

419

					return Err(sp_blockchain::Error::UnknownBlock(

420

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

421

));

422

423

},

424

};

425

426

		Ok(Some(number))

427

428

429

	fn hash(

430

		&self,

431

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

432

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

433

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

434

435

436

437

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

438

	type Error = sp_blockchain::Error;

439

440

	fn header_metadata(

441

		&self,

442

		hash: Block::Hash,

443

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

444

		self.header(hash)?

445

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

446

			.ok_or_else(|| {

447

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

448

})

449

450

451

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

452

		// No need to implement.

453

		unimplemented!("insert_header_metadata")

454

455

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

456

		// No need to implement.

457

		unimplemented!("remove_header_metadata")

458

459

460

461

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

462

	fn body(

463

		&self,

464

		hash: Block::Hash,

465

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

466

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

467

		if let Some(extrinsics) = self

468

			.storage

469

			.read()

470

			.blocks

471

			.get(&hash)

472

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

473

474

			return Ok(Some(extrinsics));

475

476

		let extrinsics = self

477

			.rpc_client

478

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

479

			.ok()

480

			.flatten()

481

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

482

483

		Ok(extrinsics)

484

485

486

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

487

		Ok(self

488

			.storage

489

			.read()

490

			.blocks

491

			.get(&hash)

492

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

493

494

495

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

496

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

497

498

		Ok(last_finalized)

499

500

501

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

502

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

503

504

505

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

506

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

507

508

509

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

510

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

511

512

513

	fn block_indexed_body(

514

		&self,

515

		_hash: Block::Hash,

516

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

517

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

518

519

520

521

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

522

	fn insert_aux<

523

'a,

524

		'b: 'a,

525

		'c: 'a,

526

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

527

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

528

>(

529

		&self,

530

		insert: I,

531

		delete: D,

532

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

533

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

534

		for (k, v) in insert {

535

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

536

537

		for k in delete {

538

			storage.aux.remove(*k);

539

540

		Ok(())

541

542

543

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

544

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

545

546

547

548

pub struct BlockImportOperation<Block: BlockT> {

549

	pending_block: Option<PendingBlock<Block>>,

550

	old_state: ForkedLazyBackend<Block>,

551

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

552

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

553

	storage_updates: StorageCollection,

554

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

555

	set_head: Option<Block::Hash>,

556

	pub(crate) before_fork: bool,

557

558

559

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

560

	fn apply_storage(

561

		&mut self,

562

		storage: Storage,

563

		commit: bool,

564

		state_version: StateVersion,

565

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

566

		use sp_state_machine::Backend;

567

		check_genesis_storage(&storage)?;

568

569

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

570

571

				&child_content.child_info,

572

				child_content

573

					.data

574

					.iter()

575

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

576

577

});

578

579

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

580

			storage

581

				.top

582

				.iter()

583

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

584

			child_delta,

585

			state_version,

586

);

587

588

		if commit {

589

			self.new_state = Some(transaction);

590

			self.storage_updates = storage

591

				.top

592

				.iter()

593

				.map(|(k, v)| {

594

					if v.is_empty() {

595

						(k.clone(), None)

596

					} else {

597

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

598

599

})

600

				.collect();

601

602

		Ok(root)

603

604

605

606

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

607

	for BlockImportOperation<Block>

608

609

	type State = ForkedLazyBackend<Block>;

610

611

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

612

		Ok(Some(&self.old_state))

613

614

615

	fn set_block_data(

616

		&mut self,

617

		header: <Block as BlockT>::Header,

618

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

619

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

620

		justifications: Option<Justifications>,

621

		state: NewBlockState,

622

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

623

		assert!(

624

			self.pending_block.is_none(),

625

			"Only one block per operation is allowed"

626

);

627

		self.pending_block = Some(PendingBlock {

628

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

629

			state,

630

});

631

		Ok(())

632

633

634

	fn update_db_storage(

635

		&mut self,

636

		update: BackendTransaction<HashingFor<Block>>,

637

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

638

		self.new_state = Some(update);

639

		Ok(())

640

641

642

	fn set_genesis_state(

643

		&mut self,

644

		storage: Storage,

645

		commit: bool,

646

		state_version: StateVersion,

647

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

648

		self.apply_storage(storage, commit, state_version)

649

650

651

	fn reset_storage(

652

		&mut self,

653

		storage: Storage,

654

		state_version: StateVersion,

655

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

656

		self.apply_storage(storage, true, state_version)

657

658

659

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

660

	where

661

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

662

663

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

664

		Ok(())

665

666

667

	fn update_storage(

668

		&mut self,

669

		update: StorageCollection,

670

		_child_update: ChildStorageCollection,

671

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

672

		self.storage_updates = update.clone();

673

		Ok(())

674

675

676

	fn mark_finalized(

677

		&mut self,

678

		hash: Block::Hash,

679

		justification: Option<Justification>,

680

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

681

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

682

		Ok(())

683

684

685

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

686

		assert!(

687

			self.pending_block.is_none(),

688

			"Only one set block per operation is allowed"

689

);

690

		self.set_head = Some(hash);

691

		Ok(())

692

693

694

	fn update_transaction_index(

695

		&mut self,

696

		_index: Vec<IndexOperation>,

697

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

698

		Ok(())

699

700

701

702

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

703

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

704

705

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

706

#[derive(Default)]

707

pub struct RawIterArgs {

708

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

709

	pub prefix: Option<Vec<u8>>,

710

711

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

712

///

713

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

714

	pub start_at: Option<Vec<u8>>,

715

716

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

717

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

718

	pub start_at_exclusive: bool,

719

720

721

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

722

pub struct RawIter<Block: BlockT> {

723

	pub(crate) args: RawIterArgs,

724

	complete: bool,

725

	_phantom: PhantomData<Block>,

726

727

728

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

729

	for RawIter<Block>

730

731

	type Backend = ForkedLazyBackend<Block>;

732

	type Error = String;

733

734

	fn next_key(

735

		&mut self,

736

		backend: &Self::Backend,

737

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

738

		use sp_state_machine::Backend;

739

740

		let remote_fetch =

741

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

742

				let result = backend

743

					.rpc_client

744

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

745

746

				match result {

747

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

748

					Err(err) => {

749

						log::trace!(

750

							target: super::LAZY_LOADING_LOG_TARGET,

751

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

752

err

753

);

754

755

						None

756

757

758

};

759

760

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

761

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

762

763

		let maybe_next_key = if backend.before_fork {

764

			remote_fetch(prefix, start_key, backend.block_hash)

765

		} else {

766

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

767

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

768

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

769

			iter_args.start_at_exclusive = true;

770

			iter_args.stop_on_incomplete_database = true;

771

772

			let readable_db = backend.db.read();

773

			let next_storage_key = readable_db

774

				.raw_iter(iter_args)

775

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

776

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

777

				.ok()

778

				.flatten();

779

780

			// IMPORTANT: free storage read lock

781

			drop(readable_db);

782

783

			let removed_key = start_key

784

				.clone()

785

				.or(prefix.clone())

786

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

787

				.unwrap_or(false);

788

			if next_storage_key.is_none() && !removed_key {

789

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

790

				match maybe_next_key {

791

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

792

					_ => None,

793

794

			} else {

795

				next_storage_key

796

797

};

798

799

		log::trace!(

800

			target: super::LAZY_LOADING_LOG_TARGET,

801

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

802

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

803

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

804

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

805

);

806

807

		if let Some(next_key) = maybe_next_key {

808

			if self

809

				.args

810

				.prefix

811

				.clone()

812

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

813

				.unwrap_or(false)

814

815

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

816

				Some(Ok(next_key))

817

			} else {

818

				self.complete = true;

819

				None

820

821

		} else {

822

			self.complete = true;

823

			None

824

825

826

827

	fn next_pair(

828

		&mut self,

829

		backend: &Self::Backend,

830

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

831

832

		use sp_state_machine::Backend;

833

834

		let remote_fetch =

835

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

836

				let result = backend

837

					.rpc_client

838

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

839

840

				match result {

841

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

842

					Err(err) => {

843

						log::trace!(

844

							target: super::LAZY_LOADING_LOG_TARGET,

845

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

846

err

847

);

848

849

						None

850

851

852

};

853

854

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

855

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

856

857

		let maybe_next_key = if backend.before_fork {

858

			remote_fetch(prefix, start_key, backend.block_hash)

859

		} else {

860

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

861

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

862

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

863

			iter_args.start_at_exclusive = true;

864

			iter_args.stop_on_incomplete_database = true;

865

866

			let readable_db = backend.db.read();

867

			let next_storage_key = readable_db

868

				.raw_iter(iter_args)

869

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

870

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

871

				.ok()

872

				.flatten();

873

874

			// IMPORTANT: free storage read lock

875

			drop(readable_db);

876

877

			let removed_key = start_key

878

				.clone()

879

				.or(prefix.clone())

880

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

881

				.unwrap_or(false);

882

			if next_storage_key.is_none() && !removed_key {

883

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

884

				match maybe_next_key {

885

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

886

					_ => None,

887

888

			} else {

889

				next_storage_key

890

891

};

892

893

		log::trace!(

894

			target: super::LAZY_LOADING_LOG_TARGET,

895

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

896

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

897

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

898

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

899

);

900

901

		let maybe_value = maybe_next_key

902

			.clone()

903

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

904

			.flatten()

905

			.flatten();

906

907

		if let Some(next_key) = maybe_next_key {

908

			if self

909

				.args

910

				.prefix

911

				.clone()

912

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

913

				.unwrap_or(false)

914

915

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

916

917

				match maybe_value {

918

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

919

					_ => None,

920

921

			} else {

922

				self.complete = true;

923

				None

924

925

		} else {

926

			self.complete = true;

927

			None

928

929

930

931

	fn was_complete(&self) -> bool {

932

		self.complete

933

934

935

936

#[derive(Debug, Clone)]

937

pub struct ForkedLazyBackend<Block: BlockT> {

938

	rpc_client: Arc<RPC>,

939

	block_hash: Option<Block::Hash>,

940

	fork_block: Block::Hash,

941

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

942

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

943

	before_fork: bool,

944

945

946

impl<Block: BlockT> ForkedLazyBackend<Block> {

947

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

948

		if let Some(ref val) = value {

949

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

950

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

951

952

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

953

954

955

956

957

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

958

	for ForkedLazyBackend<Block>

959

960

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

961

	type TrieBackendStorage = PrefixedMemoryDB<HashingFor<Block>>;

962

	type RawIter = RawIter<Block>;

963

964

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

965

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

966

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

967

968

			match result {

969

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

970

				Err(err) => {

971

					log::debug!(

972

						target: super::LAZY_LOADING_LOG_TARGET,

973

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

974

err

975

);

976

					None

977

978

979

};

980

981

		if self.before_fork {

982

			return Ok(remote_fetch(self.block_hash));

983

984

985

		let readable_db = self.db.read();

986

		let maybe_storage = readable_db.storage(key);

987

		let value = match maybe_storage {

988

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

989

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

990

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

991

992

				// Cache state

993

				drop(readable_db);

994

				self.update_storage(key, &result);

995

996

				result

997

998

			_ => None,

999

};

1000

1001

		Ok(value)

1002

1003

1004

	fn storage_hash(

1005

		&self,

1006

		key: &[u8],

1007

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

1008

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

1009

			let result = self

1010

				.rpc_client

1011

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

1012

1013

			match result {

1014

				Ok(hash) => Ok(hash),

1015

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

1016

1017

};

1018

1019

		if self.before_fork {

1020

			return remote_fetch(self.block_hash);

1021

1022

1023

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

1024

		match storage_hash {

1025

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

1026

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

1027

			_ => Ok(None),

1028

1029

1030

1031

	fn closest_merkle_value(

1032

		&self,

1033

		_key: &[u8],

1034

	) -> Result<

1035

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

1036

		Self::Error,

1037

> {

1038

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

1039

1040

1041

	fn child_closest_merkle_value(

1042

		&self,

1043

		_child_info: &sp_storage::ChildInfo,

1044

		_key: &[u8],

1045

	) -> Result<

1046

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

1047

		Self::Error,

1048

> {

1049

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

1050

1051

1052

	fn child_storage(

1053

		&self,

1054

		_child_info: &sp_storage::ChildInfo,

1055

		_key: &[u8],

1056

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

1057

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

1058

1059

1060

	fn child_storage_hash(

1061

		&self,

1062

		_child_info: &sp_storage::ChildInfo,

1063

		_key: &[u8],

1064

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

1065

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

1066

1067

1068

	fn next_storage_key(

1069

		&self,

1070

		key: &[u8],

1071

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

1072

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

1073

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

1074

			let result = self

1075

				.rpc_client

1076

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

1077

1078

			match result {

1079

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

1080

				Err(err) => {

1081

					log::trace!(

1082

						target: super::LAZY_LOADING_LOG_TARGET,

1083

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

1084

err

1085

);

1086

1087

					None

1088

1089

1090

};

1091

1092

		let maybe_next_key = if self.before_fork {

1093

			remote_fetch(self.block_hash)

1094

		} else {

1095

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

1096

			match next_storage_key {

1097

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

1098

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

1099

					remote_fetch(Some(self.fork_block))

1100

1101

				_ => None,

1102

1103

};

1104

1105

		log::trace!(

1106

			target: super::LAZY_LOADING_LOG_TARGET,

1107

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

1108

			hex::encode(key),

1109

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

1110

);

1111

1112

		Ok(maybe_next_key)

1113

1114

1115

	fn next_child_storage_key(

1116

		&self,

1117

		_child_info: &sp_storage::ChildInfo,

1118

		_key: &[u8],

1119

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

1120

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

1121

1122

1123

	fn storage_root<'a>(

1124

		&self,

1125

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

1126

		state_version: StateVersion,

1127

	) -> (

1128

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

1129

		BackendTransaction<HashingFor<Block>>,

1130

1131

	where

1132

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

1133

1134

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

1135

1136

1137

	fn child_storage_root<'a>(

1138

		&self,

1139

		_child_info: &sp_storage::ChildInfo,

1140

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

1141

		_state_version: StateVersion,

1142

	) -> (

1143

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

1144

		bool,

1145

		BackendTransaction<HashingFor<Block>>,

1146

1147

	where

1148

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

1149

1150

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

1151

1152

1153

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

1154

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

1155

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

1156

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

1157

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

1158

1159

		Ok(RawIter::<Block> {

1160

			args: clone,

1161

			complete: false,

1162

			_phantom: Default::default(),

1163

})

1164

1165

1166

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

1167

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

1168

1169

1170

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

1171

		self.db.read().usage_info()

1172

1173

1174

1175

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

1176

	type TrieBackendStorage = PrefixedMemoryDB<HashingFor<B>>;

1177

1178

	fn as_trie_backend(

1179

		&self,

1180

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

1181

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

1182

1183

1184

1185

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

1186

pub struct Backend<Block: BlockT> {

1187

	pub(crate) rpc_client: Arc<RPC>,

1188

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

1189

	pub(crate) blockchain: Blockchain<Block>,

1190

	import_lock: parking_lot::RwLock<()>,

1191

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

1192

	pub(crate) fork_checkpoint: Block::Header,

1193

1194

1195

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

1196

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

1197

		Backend {

1198

			rpc_client: rpc_client.clone(),

1199

			states: Default::default(),

1200

			blockchain: Blockchain::new(rpc_client),

1201

			import_lock: Default::default(),

1202

			pinned_blocks: Default::default(),

1203

			fork_checkpoint,

1204

1205

1206

1207

1208

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

1209

	fn insert_aux<

1210

'a,

1211

		'b: 'a,

1212

		'c: 'a,

1213

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

1214

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

1215

>(

1216

		&self,

1217

		_insert: I,

1218

		_delete: D,

1219

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

1220

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

1221

1222

1223

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

1224

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

1225

1226

1227

1228

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

1229

	type BlockImportOperation = BlockImportOperation<Block>;

1230

	type Blockchain = Blockchain<Block>;

1231

	type State = ForkedLazyBackend<Block>;

1232

	type OffchainStorage = InMemOffchainStorage;

1233

1234

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

1235

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

1236

		Ok(BlockImportOperation {

1237

			pending_block: None,

1238

			old_state,

1239

			new_state: None,

1240

			aux: Default::default(),

1241

			storage_updates: Default::default(),

1242

			finalized_blocks: Default::default(),

1243

			set_head: None,

1244

			before_fork: false,

1245

})

1246

1247

1248

	fn begin_state_operation(

1249

		&self,

1250

		operation: &mut Self::BlockImportOperation,

1251

		block: Block::Hash,

1252

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

1253

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

1254

		Ok(())

1255

1256

1257

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

1258

		if !operation.finalized_blocks.is_empty() {

1259

			for (block, justification) in operation.finalized_blocks {

1260

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

1261

1262

1263

1264

		if let Some(pending_block) = operation.pending_block {

1265

			let old_state = &operation.old_state;

1266

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

1267

			let hash = header.hash();

1268

1269

			let new_removed_keys = old_state.removed_keys.clone();

1270

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

1271

				if value.is_some() {

1272

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

1273

				} else {

1274

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

1275

1276

1277

1278

			let new_db = old_state.db.clone();

1279

			new_db.write().insert(

1280

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

1281

				StateVersion::V0,

1282

);

1283

			let new_state = ForkedLazyBackend {

1284

				rpc_client: self.rpc_client.clone(),

1285

				block_hash: Some(hash.clone()),

1286

				fork_block: self.fork_checkpoint.hash(),

1287

				db: new_db,

1288

				removed_keys: new_removed_keys,

1289

				before_fork: operation.before_fork,

1290

};

1291

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

1292

1293

			self.blockchain

1294

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

1295

1296

1297

		if !operation.aux.is_empty() {

1298

			self.blockchain.write_aux(operation.aux);

1299

1300

1301

		if let Some(set_head) = operation.set_head {

1302

			self.blockchain.set_head(set_head)?;

1303

1304

1305

		Ok(())

1306

1307

1308

	fn finalize_block(

1309

		&self,

1310

		hash: Block::Hash,

1311

		justification: Option<Justification>,

1312

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

1313

		self.blockchain.finalize_header(hash, justification)

1314

1315

1316

	fn append_justification(

1317

		&self,

1318

		hash: Block::Hash,

1319

		justification: Justification,

1320

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

1321

		self.blockchain.append_justification(hash, justification)

1322

1323

1324

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

1325

		&self.blockchain

1326

1327

1328

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

1329

		None

1330

1331

1332

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

1333

		None

1334

1335

1336

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

1337

		if hash == Default::default() {

1338

			return Ok(ForkedLazyBackend::<Block> {

1339

				rpc_client: self.rpc_client.clone(),

1340

				block_hash: Some(hash),

1341

				fork_block: self.fork_checkpoint.hash(),

1342

				db: Default::default(),

1343

				removed_keys: Default::default(),

1344

				before_fork: true,

1345

});

1346

1347

1348

		let (backend, should_write) = self

1349

			.states

1350

			.read()

1351

			.get(&hash)

1352

			.cloned()

1353

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

1354

			.unwrap_or_else(|| {

1355

				let header: Block::Header = self

1356

					.rpc_client

1357

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

1358

					.ok()

1359

					.flatten()

1360

					.expect("block header");

1361

1362

				let checkpoint = self.fork_checkpoint.clone();

1363

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

1364

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

1365

1366

					ForkedLazyBackend::<Block> {

1367

						rpc_client: self.rpc_client.clone(),

1368

						block_hash: Some(hash),

1369

						fork_block: checkpoint.hash(),

1370

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

1371

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

1372

						before_fork: false,

1373

1374

				} else {

1375

					ForkedLazyBackend::<Block> {

1376

						rpc_client: self.rpc_client.clone(),

1377

						block_hash: Some(hash),

1378

						fork_block: checkpoint.hash(),

1379

						db: Default::default(),

1380

						removed_keys: Default::default(),

1381

						before_fork: true,

1382

1383

};

1384

1385

				(state, true)

1386

});

1387

1388

		if should_write {

1389

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

1390

1391

1392

		Ok(backend)

1393

1394

1395

	fn revert(

1396

		&self,

1397

		_n: NumberFor<Block>,

1398

		_revert_finalized: bool,

1399

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

1400

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

1401

1402

1403

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

1404

		Ok(())

1405

1406

1407

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

1408

		&self.import_lock

1409

1410

1411

	fn requires_full_sync(&self) -> bool {

1412

		false

1413

1414

1415

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

1416

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

1417

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

1418

		Ok(())

1419

1420

1421

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

1422

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

1423

		blocks

1424

			.entry(hash)

1425

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

1426

			.or_insert(-1);

1427

1428

1429

1430

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

1431

1432

/// Check that genesis storage is valid.

1433

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

1434

	if storage

1435

		.top

1436

		.iter()

1437

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

1438

1439

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

1440

1441

1442

	if storage

1443

		.children_default

1444

		.keys()

1445

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

1446

1447

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

1448

1449

1450

	Ok(())

1451

1452

1453

#[derive(Debug, Clone)]

1454

pub struct RPC {

1455

	http_client: HttpClient,

1456

	delay_between_requests_ms: u64,

1457

	max_retries_per_request: usize,

1458

	counter: Arc<ReadWriteLock<u64>>,

1459

1460

1461

impl RPC {

1462

	pub fn new(

1463

		http_client: HttpClient,

1464

		delay_between_requests_ms: u64,

1465

		max_retries_per_request: usize,

1466

	) -> Self {

1467

		Self {

1468

			http_client,

1469

			delay_between_requests_ms,

1470

			max_retries_per_request,

1471

			counter: Default::default(),

1472

1473

1474

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

1475

		let request = &|| {

1476

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

1477

};

1478

1479

		self.block_on(request)

1480

1481

1482

	pub fn system_properties(

1483

		&self,

1484

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

1485

		let request = &|| {

1486

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

1487

				&self.http_client,

1488

1489

};

1490

1491

		self.block_on(request)

1492

1493

1494

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

1495

		let request = &|| {

1496

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

1497

};

1498

1499

		self.block_on(request)

1500

1501

1502

	pub fn block<Block, Hash: Clone>(

1503

		&self,

1504

		hash: Option<Hash>,

1505

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

1506

	where

1507

		Block: BlockT + DeserializeOwned,

1508

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

1509

1510

		let request = &|| {

1511

			substrate_rpc_client::ChainApi::<

1512

				BlockNumber,

1513

				Hash,

1514

				Block::Header,

1515

				SignedBlock<Block>,

1516

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

1517

};

1518

1519

		self.block_on(request)

1520

1521

1522

	pub fn block_hash<Block: BlockT + DeserializeOwned>(

1523

		&self,

1524

		block_number: Option<<Block::Header as HeaderT>::Number>,

1525

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

1526

		let request = &|| {

1527

			substrate_rpc_client::ChainApi::<

1528

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

1529

				Block::Hash,

1530

				Block::Header,

1531

				SignedBlock<Block>,

1532

			>::block_hash(

1533

				&self.http_client,

1534

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

1535

1536

};

1537

1538

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

1539

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

1540

			ListOrValue::Value(v) => v,

1541

})

1542

1543

1544

	pub fn header<Block: BlockT + DeserializeOwned>(

1545

		&self,

1546

		hash: Option<Block::Hash>,

1547

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

1548

		let request = &|| {

1549

			substrate_rpc_client::ChainApi::<

1550

				BlockNumber,

1551

				Block::Hash,

1552

				Block::Header,

1553

				SignedBlock<Block>,

1554

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

1555

};

1556

1557

		self.block_on(request)

1558

1559

1560

	pub fn storage_hash<

1561

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

1562

>(

1563

		&self,

1564

		key: StorageKey,

1565

		at: Option<Hash>,

1566

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

1567

		let request = &|| {

1568

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

1569

				&self.http_client,

1570

				key.clone(),

1571

				at.clone(),

1572

1573

};

1574

1575

		self.block_on(request)

1576

1577

1578

	pub fn storage<

1579

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

1580

>(

1581

		&self,

1582

		key: StorageKey,

1583

		at: Option<Hash>,

1584

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

1585

		let request = &|| {

1586

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

1587

				&self.http_client,

1588

				key.clone(),

1589

				at.clone(),

1590

1591

};

1592

1593

		self.block_on(request)

1594

1595

1596

	pub fn storage_keys_paged<

1597

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

1598

>(

1599

		&self,

1600

		key: Option<StorageKey>,

1601

		count: u32,

1602

		start_key: Option<StorageKey>,

1603

		at: Option<Hash>,

1604

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

1605

		let request = &|| {

1606

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

1607

				&self.http_client,

1608

				key.clone(),

1609

				count.clone(),

1610

				start_key.clone(),

1611

				at.clone(),

1612

1613

};

1614

		let result = self.block_on(request);

1615

1616

		match result {

1617

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

1618

			Err(err) => Err(err),

1619

1620

1621

1622

	pub fn query_storage_at<

1623

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

1624

>(

1625

		&self,

1626

		keys: Vec<StorageKey>,

1627

		from_block: Option<Hash>,

1628

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

1629

		let request = &|| {

1630

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

1631

				&self.http_client,

1632

				keys.clone(),

1633

				from_block.clone(),

1634

1635

};

1636

		let result = self.block_on(request);

1637

1638

		match result {

1639

			Ok(result) => Ok(result

1640

				.iter()

1641

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

1642

				.collect()),

1643

			Err(err) => Err(err),

1644

1645

1646

1647

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

1648

	where

1649

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

1650

1651

		use tokio::runtime::Handle;

1652

1653

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

1654

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

1655

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

1656

1657

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

1658

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

1659

				log::debug!(

1660

					target: super::LAZY_LOADING_LOG_TARGET,

1661

					"Sending request: {}",

1662

					self.counter.read()

1663

);

1664

1665

				// Explicit request delay, to avoid getting 429 errors

1666

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

1667

1668

				// Retry request in case of failure

1669

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

1670

				let retry_strategy =

1671

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

1672

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

1673

1674

				log::debug!(

1675

					target: super::LAZY_LOADING_LOG_TARGET,

1676

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

1677

					self.counter.read(),

1678

					result.is_ok(),

1679

					start.elapsed()

1680

);

1681

1682

				result

1683

})

1684

})

1685

1686

1687

1688

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

1689

pub fn new_lazy_loading_backend<Block>(

1690

	config: &mut Configuration,

1691

	lazy_loading_config: &LazyLoadingConfig,

1692

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

1693

where

1694

	Block: BlockT + DeserializeOwned,

1695

	Block::Hash: From<H256>,

1696

1697

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

1698

1699

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

1700

		.max_request_size(u32::MAX)

1701

		.max_response_size(u32::MAX)

1702

		.request_timeout(Duration::from_secs(10))

1703

		.build(uri)

1704

		.map_err(|e| {

1705

			sp_blockchain::Error::Backend(

1706

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

1707

1708

		})?;

1709

1710

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

1711

	let block_hash = lazy_loading_config

1712

		.from_block

1713

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

1714

	let checkpoint: Block = rpc

1715

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

1716

		.ok()

1717

		.flatten()

1718

		.expect("Fetching fork checkpoint")

1719

		.block;

1720

1721

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

1722

1723

	let chain_name = backend

1724

		.rpc_client

1725

		.system_chain()

1726

		.expect("Should fetch chain id");

1727

	let chain_properties = backend

1728

		.rpc_client

1729

		.system_properties()

1730

		.expect("Should fetch chain properties");

1731

1732

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

1733

		.with_name(chain_name.as_str())

1734

		.with_properties(chain_properties);

1735

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

1736

1737

	let base_overrides =

1738

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

1739

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

1740

		state_overrides::read(path)?

1741

	} else {

1742

		Default::default()

1743

};

1744

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

1745

		.concat()

1746

		.iter()

1747

		.map(|entry| match entry {

1748

			StateEntry::Concrete(v) => {

1749

				let key = [

1750

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

1751

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

1752

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

1753

1754

				.concat();

1755

1756

				(key, v.value.clone())

1757

1758

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

1759

})

1760

		.collect();

1761

1762

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

1763

1764

	// Produce first block after the fork

1765

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

1766

1767

	Ok(backend)

1768