feat: asynchronous and configurable block sealing

e2e-tests/test/debug-apis.test.ts

@@ -5,6 +5,7 @@ import { Deployer } from "@matterlabs/hardhat-zksync-deploy";
 import { RichAccounts } from "../helpers/constants";
 import { deployContract, expectThrowsAsync, getTestProvider } from "../helpers/utils";
 import { BigNumber } from "ethers";
+import { TransactionResponse } from "@ethersproject/abstract-provider";
 
 const provider = getTestProvider();
 
@@ -88,15 +89,16 @@ describe("debug_traceTransaction", function () {
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
 
-    const txReceipt = await greeter.setGreeting("Luke Skywalker");
-    const trace = await provider.send("debug_traceTransaction", [txReceipt.hash]);
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    const txReceipt = await txResponse.wait();
+    const trace = await provider.send("debug_traceTransaction", [txReceipt.transactionHash]);
 
     // call should be successful
     expect(trace.error).to.equal(null);
     expect(trace.calls.length).to.equal(1);
 
     // gas limit should match
-    expect(BigNumber.from(trace.gas).toNumber()).to.equal(txReceipt.gasLimit.toNumber());
+    expect(BigNumber.from(trace.gas).toNumber()).to.equal(txResponse.gasLimit.toNumber());
   });
 
   it("Should respect only_top_calls option", async function () {
@@ -105,9 +107,10 @@ describe("debug_traceTransaction", function () {
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
 
-    const txReceipt = await greeter.setGreeting("Luke Skywalker");
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    const txReceipt = await txResponse.wait();
     const trace = await provider.send("debug_traceTransaction", [
-      txReceipt.hash,
+      txReceipt.transactionHash,
       { tracer: "callTracer", tracerConfig: { onlyTopCall: true } },
     ]);
 
@@ -124,7 +127,8 @@ describe("debug_traceBlockByHash", function () {
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
 
-    const txReceipt = await greeter.setGreeting("Luke Skywalker");
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    await txResponse.wait();
     const latestBlock = await provider.getBlock("latest");
     const block = await provider.getBlock(latestBlock.number - 1);
 
@@ -135,7 +139,7 @@ describe("debug_traceBlockByHash", function () {
 
     // should contain trace for our tx
     const trace = traces[0].result;
-    expect(trace.input).to.equal(txReceipt.data);
+    expect(trace.input).to.equal(txResponse.data);
   });
 });
 
@@ -146,7 +150,8 @@ describe("debug_traceBlockByNumber", function () {
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
 
-    const txReceipt = await greeter.setGreeting("Luke Skywalker");
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    await txResponse.wait();
 
     // latest block will be empty, check we get no traces for it
     const empty_traces = await provider.send("debug_traceBlockByNumber", ["latest"]);
@@ -161,6 +166,6 @@ describe("debug_traceBlockByNumber", function () {
 
     // should contain trace for our tx
     const trace = traces[0].result;
-    expect(trace.input).to.equal(txReceipt.data);
+    expect(trace.input).to.equal(txResponse.data);
   });
 });

e2e-tests/test/evm-apis.test.ts

@@ -49,10 +49,11 @@ describe("evm_increaseTime", function () {
     // Act
     await provider.send("evm_increaseTime", [timeIncreaseInSeconds]);
 
-    await wallet.sendTransaction({
+    const txResponse = await wallet.sendTransaction({
       to: userWallet.address,
       value: ethers.utils.parseEther("0.1"),
     });
+    await txResponse.wait();
     expectedTimestamp += 2; // New transaction will add two blocks
 
     // Assert
@@ -73,10 +74,11 @@ describe("evm_setNextBlockTimestamp", function () {
     // Act
     await provider.send("evm_setNextBlockTimestamp", [expectedTimestamp.toString(16)]);
 
-    await wallet.sendTransaction({
+    const txResponse = await wallet.sendTransaction({
       to: userWallet.address,
       value: ethers.utils.parseEther("0.1"),
     });
+    await txResponse.wait();
     expectedTimestamp += 1; // After executing a transaction, the node puts it into a block and increases its current timestamp
 
     // Assert
@@ -97,10 +99,11 @@ describe("evm_setTime", function () {
     // Act
     await provider.send("evm_setTime", [expectedTimestamp]);
 
-    await wallet.sendTransaction({
+    const txResponse = await wallet.sendTransaction({
       to: userWallet.address,
       value: ethers.utils.parseEther("0.1"),
     });
+    await txResponse.wait();
     expectedTimestamp += 2; // New transaction will add two blocks
 
     // Assert

e2e-tests/test/main.test.ts

@@ -32,10 +32,11 @@ describe("Greeter Smart Contract", function () {
 
       // setup user wallet with 3 ETH
       const userWallet = Wallet.createRandom().connect(provider);
-      await wallet.sendTransaction({
+      const txResponse = await wallet.sendTransaction({
         to: userWallet.address,
         value: ethers.utils.parseEther("3"),
       });
+      await txResponse.wait();
 
       // deploy Greeter contract
       const artifact = await deployer.loadArtifact("Greeter");

e2e-tests/test/zks-apis.test.ts

@@ -6,6 +6,7 @@ import { BigNumber, ethers } from "ethers";
 import * as hre from "hardhat";
 import { TransactionRequest } from "zksync-web3/build/src/types";
 import { Deployer } from "@matterlabs/hardhat-zksync-deploy";
+import { TransactionResponse } from "@ethersproject/abstract-provider";
 
 const provider = getTestProvider();
 
@@ -66,8 +67,9 @@ describe("zks_getTransactionDetails", function () {
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
 
-    const txReceipt = await greeter.setGreeting("Luke Skywalker");
-    const details = await provider.send("zks_getTransactionDetails", [txReceipt.hash]);
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    const txReceipt = await txResponse.wait();
+    const details = await provider.send("zks_getTransactionDetails", [txReceipt.transactionHash]);
 
     expect(details["status"]).to.equal("included");
     expect(details["initiatorAddress"].toLowerCase()).to.equal(wallet.address.toLowerCase());
@@ -96,7 +98,7 @@ describe("zks_getBlockDetails", function () {
     const deployer = new Deployer(hre, wallet);
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
-    await greeter.setGreeting("Luke Skywalker");
+    await (await greeter.setGreeting("Luke Skywalker")).wait();
 
     const latestBlock = await provider.getBlock("latest");
     const details = await provider.send("zks_getBlockDetails", [latestBlock.number]);
@@ -145,13 +147,14 @@ describe("zks_getRawBlockTransactions", function () {
     const deployer = new Deployer(hre, wallet);
 
     const greeter = await deployContract(deployer, "Greeter", ["Hi"]);
-    const receipt = await greeter.setGreeting("Luke Skywalker");
+    const txResponse: TransactionResponse = await greeter.setGreeting("Luke Skywalker");
+    await txResponse.wait();
 
     const latestBlock = await provider.getBlock("latest");
     const txns = await provider.send("zks_getRawBlockTransactions", [latestBlock.number - 1]);
 
     expect(txns.length).to.equal(1);
-    expect(txns[0]["execute"]["calldata"]).to.equal(receipt.data);
+    expect(txns[0]["execute"]["calldata"]).to.equal(txResponse.data);
   });
 });
 

src/config/cli.rs

@@ -1,4 +1,5 @@
 use clap::{arg, command, Parser, Subcommand};
+use std::time::Duration;
 use zksync_types::H256;
 
 use crate::config::{
@@ -127,6 +128,11 @@ pub struct Cli {
     #[arg(long, help_heading = "Cache Options")]
     /// Cache directory location for disk cache (default: .cache).
     pub cache_dir: Option<String>,
+
+    #[arg(long, value_parser = duration_from_secs_f64, help_heading = "Block Sealing")]
+    /// Block time in seconds for interval sealing.
+    /// If unset, node seals a new block as soon as there is at least one transaction.
+    pub block_time: Option<Duration>,
 }
 
 #[derive(Debug, Subcommand)]
@@ -249,3 +255,11 @@ impl Cli {
         }
     }
 }
+
+fn duration_from_secs_f64(s: &str) -> Result<Duration, String> {
+    let s = s.parse::<f64>().map_err(|e| e.to_string())?;
+    if s == 0.0 {
+        return Err("Duration must be greater than 0".to_string());
+    }
+    Duration::try_from_secs_f64(s).map_err(|e| e.to_string())
+}

src/config/mod.rs

@@ -58,6 +58,8 @@ pub struct TestNodeConfig {
     pub log_file_path: String,
     /// Cache configuration for the test node
     pub cache_config: CacheConfig,
+    /// Maximum number of transactions per block
+    pub max_transactions: usize,
 }
 
 impl Default for TestNodeConfig {
@@ -89,6 +91,9 @@ impl Default for TestNodeConfig {
 
             // Cache configuration default
             cache_config: Default::default(),
+
+            // Block sealing configuration default
+            max_transactions: 1000,
         }
     }
 }

src/main.rs

@@ -56,6 +56,8 @@ use crate::namespaces::{
     EthTestNodeNamespaceT, EvmNamespaceT, HardhatNamespaceT, NetNamespaceT, Web3NamespaceT,
     ZksNamespaceT,
 };
+use crate::node::{BlockProducer, BlockSealer, ImpersonationManager, TimestampManager, TxPool};
+use crate::system_contracts::SystemContracts;
 
 #[allow(clippy::too_many_arguments)]
 async fn build_json_http<
@@ -217,8 +219,17 @@ async fn main() -> anyhow::Result<()> {
         }
     }
 
-    let node: InMemoryNode<HttpForkSource> =
-        InMemoryNode::new(fork_details, Some(observability), &config);
+    let time = TimestampManager::default();
+    let impersonation = ImpersonationManager::default();
+    let pool = TxPool::new(impersonation.clone());
+    let node: InMemoryNode<HttpForkSource> = InMemoryNode::new(
+        fork_details,
+        Some(observability),
+        &config,
+        time.clone(),
+        impersonation,
+        pool.clone(),
+    );
 
     if let Some(bytecodes_dir) = opt.override_bytecodes_dir {
         override_bytecodes(&node, bytecodes_dir).unwrap();
@@ -254,15 +265,32 @@ async fn main() -> anyhow::Result<()> {
     let threads = build_json_http(
         SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), config.port),
         log_level_filter,
-        node,
+        node.clone(),
     )
     .await;
 
+    let block_sealer = if let Some(block_time) = opt.block_time {
+        BlockSealer::fixed_time(config.max_transactions, block_time)
+    } else {
+        BlockSealer::immediate(config.max_transactions)
+    };
+    let system_contracts =
+        SystemContracts::from_options(&config.system_contracts_options, config.use_evm_emulator);
+    let block_producer_handle = tokio::task::spawn(BlockProducer::new(
+        node,
+        pool,
+        block_sealer,
+        system_contracts,
+    ));
+
     tracing::info!("========================================");
     tracing::info!("  Node is ready at 127.0.0.1:{}", config.port);
     tracing::info!("========================================");
 
-    future::select_all(vec![threads]).await.0.unwrap();
+    future::select_all(vec![threads, block_producer_handle])
+        .await
+        .0
+        .unwrap();
 
     Ok(())
 }

src/node/block_producer.rs

@@ -0,0 +1,54 @@
+use crate::fork::ForkSource;
+use crate::node::pool::{TxBatch, TxPool};
+use crate::node::sealer::BlockSealer;
+use crate::node::InMemoryNode;
+use crate::system_contracts::SystemContracts;
+use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use zksync_multivm::interface::TxExecutionMode;
+
+pub struct BlockProducer<S: Clone> {
+    node: InMemoryNode<S>,
+    pool: TxPool,
+    block_sealer: BlockSealer,
+    system_contracts: SystemContracts,
+}
+
+impl<S: Clone> BlockProducer<S> {
+    pub fn new(
+        node: InMemoryNode<S>,
+        pool: TxPool,
+        block_sealer: BlockSealer,
+        system_contracts: SystemContracts,
+    ) -> Self {
+        Self {
+            node,
+            pool,
+            block_sealer,
+            system_contracts,
+        }
+    }
+}
+
+impl<S: ForkSource + Clone + fmt::Debug> Future for BlockProducer<S> {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let pin = self.get_mut();
+        loop {
+            if let Poll::Ready(tx_batch) = pin.block_sealer.poll(&pin.pool, cx) {
+                let TxBatch { impersonating, txs } = tx_batch;
+
+                let base_system_contracts = pin
+                    .system_contracts
+                    .contracts(TxExecutionMode::VerifyExecute, impersonating)
+                    .clone();
+                pin.node
+                    .seal_block(txs, base_system_contracts)
+                    .expect("block sealing failed");
+            }
+        }
+    }
+}