perf(evm): read commits in batches (#1334)

* conditional storage compression

* add tests

* rename CompressedBincode to CompactBincode

* add TransactionKey wrapper

* tests

* add `getCommitsByBlockRange`

* use new evm  to read commits

* pass maxBytes

* fix

* ensure no negative start

---------

Co-authored-by: oXtxNt9U <120286271+oXtxNt9U@users.noreply.github.com>

Author: sebastijankuzner

packages/api-sync/source/restore.ts

@@ -324,7 +324,7 @@ export class Restore {
 			const fromBlockNumber = Math.min(currentBlockNumber, mostRecentCommit.block.number);
 			const toBlockNumber = Math.min(currentBlockNumber + BATCH_SIZE - 1, mostRecentCommit.block.number);
 
-			const commits = this.databaseService.readCommits(fromBlockNumber, toBlockNumber);
+			const commits = this.databaseService.readCommits(fromBlockNumber, toBlockNumber, Number.MAX_SAFE_INTEGER);
 
 			const blocks: Models.Block[] = [];
 			const transactions: Models.Transaction[] = [];

packages/contracts/source/contracts/database.ts

@@ -14,8 +14,8 @@ export interface DatabaseService extends CommitHandler {
 
 	getLastCommit(): Promise<Commit>;
 	hasCommitByHash(blockHash: string): Promise<boolean>;
-	findCommitBuffers(start: number, end: number): Promise<Buffer[]>;
-	readCommits(start: number, end: number): AsyncGenerator<Commit>;
+	findCommitBuffers(start: number, end: number, maxBytes: number): Promise<Buffer[]>;
+	readCommits(start: number, end: number, maxBytes: number): AsyncGenerator<Commit>;
 
 	getBlock(blockNumber: number): Promise<Block | undefined>;
 	getBlockByHash(blockHash: string): Promise<Block | undefined>;

packages/contracts/source/contracts/evm/storage.ts

@@ -55,6 +55,11 @@ export interface Storage {
 	getBlockHeaderData(blockNumber: number): Promise<BlockHeaderStorageData | undefined>;
 	getBlockNumberByHash(blockHash: string): Promise<number | undefined>;
 	getCommitData(blockNumber: number): Promise<CommitStorageData | undefined>;
+	getCommitsByBlockRange(
+		fromBlockNumber: number,
+		toBlockNumber: number,
+		maxBytes: number,
+	): Promise<CommitStorageData[]>;
 	getTransactionData(key: string): Promise<TransactionStorageData | undefined>;
 	getTransactionKeyByHash(txHash: string): Promise<string | undefined>;
 	isEmpty(): Promise<boolean>;

packages/database/source/database-service.test.ts

@@ -83,13 +83,13 @@ describe<{
 	});
 
 	it("findCommitBuffers - should be ok", async ({ databaseService }) => {
-		const commits = await databaseService.findCommitBuffers(1, 2);
+		const commits = await databaseService.findCommitBuffers(1, 2, Number.MAX_SAFE_INTEGER);
 		assert.empty(commits);
 	});
 
 	it("readCommits - should be ok", async ({ databaseService }) => {
 		const commits = [];
-		for await (const commit of databaseService.readCommits(1, 2)) {
+		for await (const commit of databaseService.readCommits(1, 2, Number.MAX_SAFE_INTEGER)) {
 			commits.push(commit);
 		}
 
@@ -197,7 +197,6 @@ describe<{
 				from: transaction.from,
 				gasLimit: BigInt(transaction.gasLimit),
 				gasPrice: BigInt(transaction.gasPrice),
-				index: transaction.transactionIndex,
 				nonce: transaction.nonce,
 				specId: Enums.Evm.SpecId.LATEST,
 				to: transaction.to,
@@ -237,7 +236,9 @@ describe<{
 	});
 
 	it("#findCommitBuffers - should return commit buffer", async ({ databaseService, genesisCommit }) => {
-		assert.equal(await databaseService.findCommitBuffers(0, 1), [Buffer.from(genesisCommit.serialized, "hex")]);
+		assert.equal(await databaseService.findCommitBuffers(0, 1, Number.MAX_SAFE_INTEGER), [
+			Buffer.from(genesisCommit.serialized, "hex"),
+		]);
 	});
 
 	it("#getBlock - should return block", async ({ databaseService, genesisCommit }) => {
@@ -269,7 +270,7 @@ describe<{
 
 	it("#readCommits - should return commits", async ({ databaseService, genesisCommit }) => {
 		const commits = [];
-		for await (const commit of databaseService.readCommits(0, 1)) {
+		for await (const commit of databaseService.readCommits(0, 1, Number.MAX_SAFE_INTEGER)) {
 			commits.push(commit);
 		}
 		assert.equal(

packages/database/source/database-service.ts

@@ -38,26 +38,14 @@ export class DatabaseService implements Contracts.Database.DatabaseService {
 		return blockNumber !== undefined;
 	}
 
-	public async findCommitBuffers(start: number, end: number): Promise<Buffer[]> {
-		const blockNumbers: number[] = [];
+	public async findCommitBuffers(start: number, end: number, maxBytes: number): Promise<Buffer[]> {
+		const buffers: Buffer[] = [];
 
-		for (const blockNumber of this.#range(start, end)) {
-			blockNumbers.push(blockNumber);
+		for await (const commit of this.readCommits(start, end, maxBytes)) {
+			buffers.push(Buffer.from(commit.serialized, "hex"));
 		}
 
-		const buffers = await Promise.all(
-			blockNumbers.map(async (blockNumber: number) => {
-				const commitStorage = await this.#readCommitStorage(blockNumber);
-				if (!commitStorage) {
-					return;
-				}
-
-				const commit = await this.commitFactory.fromStorage(commitStorage);
-				return Buffer.from(commit.serialized, "hex");
-			}),
-		);
-
-		return buffers.filter((commit) => !!commit);
+		return buffers;
 	}
 
 	public async getBlock(blockNumber: number): Promise<Contracts.Crypto.Block | undefined> {
@@ -106,12 +94,13 @@ export class DatabaseService implements Contracts.Database.DatabaseService {
 	}
 
 	public async findBlocks(start: number, end: number): Promise<Contracts.Crypto.Block[]> {
-		const commitBuffers = await this.findCommitBuffers(start, end);
+		const blocks: Contracts.Crypto.Block[] = [];
+
+		for await (const commit of this.readCommits(start, end, Number.MAX_SAFE_INTEGER)) {
+			blocks.push(commit.block);
+		}
 
-		return await this.#map(
-			commitBuffers,
-			async (buffer: Buffer) => (await this.commitFactory.fromBytes(buffer)).block,
-		);
+		return blocks;
 	}
 
 	public async getTransactionByHash(transactionHash: string): Promise<Contracts.Crypto.BlockTransaction | undefined> {
@@ -143,16 +132,29 @@ export class DatabaseService implements Contracts.Database.DatabaseService {
 		return this.#readTransaction(`${blockNumber}-${index}`);
 	}
 
-	public async *readCommits(start: number, end: number): AsyncGenerator<Contracts.Crypto.Commit> {
-		for (let blockNumber = start; blockNumber <= end; blockNumber++) {
-			const data = await this.#readCommitStorage(blockNumber);
+	public async *readCommits(start: number, end: number, maxBytes: number): AsyncGenerator<Contracts.Crypto.Commit> {
+		let from = Math.max(0, start);
+		let remainingBytes = maxBytes;
 
-			if (!data) {
+		while (from <= end) {
+			const commitsData = await this.storage.getCommitsByBlockRange(from, end, remainingBytes);
+			if (commitsData.length === 0) {
 				return;
 			}
 
-			const commit = await this.commitFactory.fromStorage(data);
-			yield commit;
+			let lastBlockNumber = from;
+			for (const data of commitsData) {
+				const commit = await this.commitFactory.fromStorage(data);
+				lastBlockNumber = commit.block.number;
+				yield commit;
+
+				remainingBytes -= commit.serialized.length / 2;
+				if (remainingBytes <= 0) {
+					return;
+				}
+			}
+
+			from = lastBlockNumber + 1;
 		}
 	}
 
@@ -201,19 +203,4 @@ export class DatabaseService implements Contracts.Database.DatabaseService {
 
 		return this.transactionFactory.fromStorage({ ...transactionStorageData, blockHash: blockHeaderData.hash });
 	}
-
-	async #map<T, U>(data: U[], callback: (...arguments_: U[]) => Promise<T>): Promise<T[]> {
-		const result: T[] = [];
-		for (const [index, datum] of data.entries()) {
-			result[index] = await callback(datum);
-		}
-
-		return result;
-	}
-
-	*#range(start: number, end: number): Generator<number> {
-		for (let index = start; index <= end; index++) {
-			yield index;
-		}
-	}
 }

packages/evm-service/source/instances/evm.ts

@@ -208,6 +208,14 @@ export class EvmInstance implements Contracts.Evm.Instance, Contracts.Evm.Storag
 		return result;
 	}
 
+	public async getCommitsByBlockRange(
+		fromBlockNumber: number,
+		toBlockNumber: number,
+		maxBytes: number,
+	): Promise<Contracts.Evm.CommitStorageData[]> {
+		return this.#evm.getCommitsByBlockRange(BigInt(fromBlockNumber), BigInt(toBlockNumber), BigInt(maxBytes));
+	}
+
 	public async getTransactionData(key: string): Promise<Contracts.Evm.TransactionStorageData | undefined> {
 		const result = await this.#evm.getTransactionData(key);
 		if (result === null || result === undefined) {

packages/evm/bindings/src/lib.rs

@@ -915,6 +915,27 @@ impl EvmInner {
         Ok(Some((proof, header, txs)))
     }
 
+    pub fn get_commits_by_block_range(
+        &mut self,
+        from_block_number: u64,
+        to_block_number: u64,
+        max_bytes: u64,
+    ) -> std::result::Result<
+        Vec<(ProofData, BlockHeaderData, Vec<TransactionData>)>,
+        EVMError<String>,
+    > {
+        match self.persistent_db.get_commits_by_block_range(
+            from_block_number,
+            to_block_number,
+            max_bytes,
+        ) {
+            Ok(commits) => Ok(commits),
+            Err(err) => Err(EVMError::Database(
+                format!("failed reading commits by block range: {}", err).into(),
+            )),
+        }
+    }
+
     pub fn get_transaction_data(
         &mut self,
         key: String,
@@ -1625,6 +1646,33 @@ impl JsEvmWrapper {
         )
     }
 
+    #[napi]
+    pub fn get_commits_by_block_range<'env>(
+        &mut self,
+        env: &'env Env,
+        from_block_number: BigInt,
+        to_block_number: BigInt,
+        max_bytes: BigInt,
+    ) -> Result<PromiseRaw<'env, Vec<JsCommitData>>> {
+        let from_block_number = from_block_number.get_u64().1;
+        let to_block_number = to_block_number.get_u64().1;
+        let max_bytes = max_bytes.get_u64().1;
+        env.spawn_future_with_callback(
+            Self::get_commits_by_block_range_async(
+                self.evm.clone(),
+                from_block_number,
+                to_block_number,
+                max_bytes,
+            ),
+            |_, result| {
+                Ok(result
+                    .into_iter()
+                    .map(|(proof, header, txs)| JsCommitData::new(proof, header, txs))
+                    .collect())
+            },
+        )
+    }
+
     #[napi]
     pub fn get_transaction_data<'env>(
         &mut self,
@@ -2050,6 +2098,21 @@ impl JsEvmWrapper {
         }
     }
 
+    async fn get_commits_by_block_range_async(
+        evm: Arc<tokio::sync::Mutex<EvmInner>>,
+        from_block_number: u64,
+        to_block_number: u64,
+        max_bytes: u64,
+    ) -> Result<Vec<(ProofData, BlockHeaderData, Vec<TransactionData>)>> {
+        let mut lock = evm.lock().await;
+        let result = lock.get_commits_by_block_range(from_block_number, to_block_number, max_bytes);
+
+        match result {
+            Ok(result) => Result::Ok(result),
+            Err(err) => Result::Err(serde::de::Error::custom(err)),
+        }
+    }
+
     async fn get_transaction_data_async(
         evm: Arc<tokio::sync::Mutex<EvmInner>>,
         key: String,