diff --git a/.github/workflows/e2e-leader-follower.yml b/.github/workflows/e2e-leader-follower.yml
index 636c50f29..7a16dac9a 100644
--- a/.github/workflows/e2e-leader-follower.yml
+++ b/.github/workflows/e2e-leader-follower.yml
@@ -87,7 +87,17 @@ jobs:
       fail-fast: false
       matrix:
         test:
-          [brlc, importer, miner, change, deploy, kafka, health, tx-types]
+          [
+            brlc,
+            importer,
+            miner,
+            change,
+            deploy,
+            kafka,
+            health,
+            tx-types,
+            blockhash,
+          ]
         use_block_changes_replication: [false, true]
     name: "E2E Leader & Follower on ${{ matrix.test }} (Replication: ${{ matrix.use_block_changes_replication }})"
     needs: setup_cache
diff --git a/docs/continuity.md b/docs/continuity.md
new file mode 100644
index 000000000..d2e80f077
--- /dev/null
+++ b/docs/continuity.md
@@ -0,0 +1,150 @@
+# Block continuity
+
+## Progress model
+
+Stratus tracks two process-local progress tips:
+
+- `latest_sealed` is the execution tip. Temporary storage owns its latest in-process state overlay and final hash. Its hash is the parent used to build the next header; after startup, the durable underlying state still comes from RocksDB.
+- `last_saved` is the durable tip. `StratusStorage` stores only its block number and hash because the complete block and state already live in RocksDB.
+
+On a populated database, both are initialized from the latest permanent block. On an empty database, temporary storage keeps the canonical genesis reference, but its pending block remains block 0 and `last_saved` remains empty until genesis is imported or persisted.
+
+Normal leader and follower flows seal and save sequentially, so the tips usually match. The offline importer deliberately pipelines execution and persistence, allowing `latest_sealed` to run ahead.
+
+```mermaid
+flowchart LR
+    subgraph durable [Permanent progress]
+        P0["Block N-1"] --> P1["Block N: last_saved"]
+    end
+
+    subgraph backlog [Offline FIFO backlog]
+        Q1["Block N+1"] --> Q2["Block N+2"]
+    end
+
+    T["Block N+3: latest_sealed"]
+    P1 --> Q1
+    Q2 --> T
+```
+
+These are not independent chains. Permanent progress must always be an ordered prefix of sealed progress.
+
+## Hash schemes
+
+Genesis retains the legacy V1 hash:
+
+```text
+V1 = keccak256(number as 8-byte big-endian)
+```
+
+Locally sealed non-genesis blocks use V2:
+
+```text
+V2 = keccak256(
+    number as 8-byte big-endian
+    || timestamp as 8-byte big-endian
+    || transactions_root
+    || parent_hash
+)
+```
+
+Reexecution validates that the imported hash is V2 or the temporary V1 compatibility hash. Replication receives a prebuilt block and relies on the universal saved-chain continuity checks.
+
+## Guarded sealing
+
+`PendingSession` owns a `PendingBlockGuard`, which wraps temporary storage's `pending_session` mutex colocated with `InMemoryChainState`. Callers use session methods instead of manually pairing lock-acquiring methods with variants that accept an existing guard.
+
+One session spans the complete pending-block lifecycle:
+
+```text
+set pending header when importing
+→ execute and save transactions
+→ snapshot pending state
+→ build the block and determine its final hash
+→ finish pending state
+```
+
+Local sealing and reexecution use the snapshot to build a block without destroying pending state. Local sealing calculates V2; reexecution validates the external V2 or V1 hash. If external validation fails, pending remains unchanged. Replication skips this snapshot-and-build step because it receives a complete block.
+
+Once the block is accepted, `finish_pending_block` uses `std::mem::replace` to move the original pending state into `latest_sealed`, attaches the final hash, and creates the next pending state.
+
+Pending and latest sealed state share one `RwLock<InMemoryChainState>`, so the move, hash update, and next-pending creation are one atomic write. Another pending session cannot start until `PendingSession` is dropped or consumed by sealing.
+
+Local synchronous modes also retain the existing `mine_and_commit` mutex from sealing through persistence. It guarantees that concurrent local triggers cannot seal blocks in one order and race to save them in another order. Offline importer does not use this mutex; its single executor, FIFO channel, and single saver provide ordering.
+
+## Saving and continuity
+
+`save_block` is the authoritative continuity boundary for leader mining, follower reexecution, follower replication, fake leader, and offline import.
+
+Before saving block `N`, it validates:
+
+```text
+N == last_saved.number + 1
+N.parent_hash == last_saved.hash
+```
+
+RocksDB is written only after these checks pass. `last_saved` advances only after the write succeeds, and no permanent read is required during each save.
+
+Online follower modes perform the same check as a read-only preflight before emitting Kafka events. `save_block` repeats it authoritatively before persistence.
+
+External reexecution calculates the block locally and accepts the external V2 hash or the temporary V1 compatibility hash. Replication receives a prebuilt block, but both modes still pass through the universal saved-chain continuity check.
+
+If the process restarts, sealed-but-unsaved work is discarded. The durable tip is loaded from RocksDB, temporary state resumes from it, and the unsaved range is executed again.
+
+## Block-hash lookup
+
+The block-hash cache does not determine chain progress or parent continuity. Its normal role is accelerating the EVM `BLOCKHASH` opcode.
+
+```mermaid
+flowchart LR
+    EVM["EVM BLOCKHASH"] --> TempCheck{"Is it the latest sealed block?"}
+    TempCheck -->|"yes"| Result["Block hash"]
+    TempCheck -->|"no"| Cache["Block-hash cache"]
+    Cache -->|"hit"| Result
+    Cache -->|"miss"| Permanent["Permanent storage"]
+    Permanent --> Result
+
+    Queue["Offline sealed backlog"] -. "temporary workaround" .-> Cache
+```
+
+Lookup order is:
+
+1. The latest sealed block, whether or not it was already saved.
+2. The block-hash cache.
+3. Permanent storage.
+
+The first step does not consult `last_saved`. A sealed block keeps its hash when it is persisted, so the sealed tip is authoritative for its own number either way. Reading it would otherwise put the `BLOCKHASH` opcode behind the whole block saving critical section, which holds `last_saved` across the RocksDB write.
+
+The cache is a fixed ring indexed by `number % capacity`, not a general purpose cache. Block numbers are dense and published in order, so the ring retains exactly the most recent `capacity` blocks, and two numbers share a slot only when they are `capacity` apart. Nothing inside the window can be evicted, including a hash that is published and never read. That matters because a general purpose cache does the opposite: it spreads entries over independently sized shards and evicts the never read ones first, which is exactly what a freshly sealed hash is.
+
+The default is 256 entries, matching the `BLOCKHASH` window. Administrators may shrink it, and a configured zero is raised to a single slot: that slot can only ever hold the sealed tip, which is answered from temporary storage before the ring is consulted, so it is a cache in name only. Importer-offline always adds capacity for its bounded sealed-but-unsaved backlog:
+
+```text
+configured capacity + batch_size × (queue_size + 2)
+```
+
+The extra two batches cover one batch being built by the executor and one being processed by the saver. Sizing the ring this way is what makes the unsaved backlog unevictable rather than merely likely to survive.
+
+Older unsaved offline blocks are temporarily dependent on this cache because permanent storage cannot serve them yet. Remove this workaround when importer-offline is removed.
+
+## Lock roles
+
+- `PendingSession`: exposes pending-header setup, execution append, and sealing under one temporary-storage session lock.
+- `PendingBlockGuard`: private capability held by `PendingSession` and passed only to lower storage layers.
+- `TransactionGuard`: lets fake leader hold the executor transaction mutex before miner locks, matching RPC lock order and preventing deadlock.
+- `mine_and_commit`: preserves seal-to-save ordering for synchronous local modes.
+- `commit`: serializes permanent writes.
+- `last_saved`: serializes durable continuity validation and advancement.
+- `transient_state_lock`: preserves consistency between permanent writes and latest account/slot caches.
+
+## TODO: Temporary storage naming
+
+`InmemoryTransactionTemporaryStorage` and `transaction_storage` are misleading names. This component does not represent one Ethereum transaction or a database transaction. It owns:
+
+- The pending block header and all transaction executions.
+- Aggregated pending account and slot changes.
+- The latest sealed block state and hash.
+- The transition from pending to latest sealed.
+
+For example, an interval-mined pending block can contain many Ethereum transactions. When sealed, the complete pending state becomes `latest_sealed`; an individual transaction does not.
+
+A future focused refactor should rename it. Preferred naming is `InMemoryBlockStateStorage` with a `block_storage` field. `InMemoryExecutionStateStorage` with `execution_storage` is another reasonable option.
\ No newline at end of file
diff --git a/e2e/cloudwalk-contracts/integration/test/leader-follower-blockhash.test.ts b/e2e/cloudwalk-contracts/integration/test/leader-follower-blockhash.test.ts
new file mode 100644
index 000000000..3b1ebe9d0
--- /dev/null
+++ b/e2e/cloudwalk-contracts/integration/test/leader-follower-blockhash.test.ts
@@ -0,0 +1,159 @@
+import { expect } from "chai";
+import { concat, keccak256, toBeHex } from "ethers";
+
+import { ALICE, BOB, CHARLIE } from "./helpers/account";
+import {
+    sendAndGetFullResponse,
+    sendWithRetry,
+    toHex,
+    updateProviderUrl,
+    waitForFollowerToSyncWithLeader,
+} from "./helpers/rpc";
+
+const HASH_ZERO = "0x" + "0".repeat(64);
+
+// Root reported by blocks that mined no transaction.
+const EMPTY_TRANSACTIONS_ROOT = "0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421";
+
+// How many of the most recent blocks are inspected, so that the suite does not walk a chain of arbitrary length.
+const INSPECTED_BLOCKS = 20;
+
+// Recomputes the hash of a mined block from the header fields it reports, mirroring the node.
+//
+// The preimage is the block number and the timestamp, both as 8 byte big endian integers, followed by the
+// 32 byte transactions root and the 32 byte parent hash.
+function calculateBlockHash(block: any): string {
+    const number = toBeHex(BigInt(block.number), 8);
+    const timestamp = toBeHex(BigInt(block.timestamp), 8);
+    return keccak256(concat([number, timestamp, block.transactionsRoot, block.parentHash]));
+}
+
+// The genesis keeps the legacy scheme, which hashes only the block number.
+function calculateGenesisHash(): string {
+    return keccak256(toBeHex(0n, 8));
+}
+
+async function getBlock(node: string, blockNumber: number): Promise<any> {
+    updateProviderUrl(node);
+    return await sendWithRetry("eth_getBlockByNumber", [toHex(blockNumber), false]);
+}
+
+describe("Leader & Follower block hash integration test", function () {
+    // The most recent blocks both nodes agree on, plus the first mined block, which is the oldest one
+    // hashed with the current scheme. Set once the follower catches up with the leader.
+    let inspectedBlocks: number[] = [];
+
+    it("Validate initial Leader and Follower health", async function () {
+        updateProviderUrl("stratus");
+        expect(await sendWithRetry("stratus_health", [])).to.equal(true);
+
+        updateProviderUrl("stratus-follower");
+        expect(await sendWithRetry("stratus_health", [])).to.equal(true);
+    });
+
+    it("Genesis block on the Leader is hashed with the legacy scheme", async function () {
+        const genesis = await getBlock("stratus", 0);
+
+        expect(genesis.number).to.equal("0x0");
+        expect(genesis.parentHash, "genesis has no parent").to.equal(HASH_ZERO);
+        expect(genesis.hash, "genesis hash is the keccak of its number").to.equal(calculateGenesisHash());
+    });
+
+    it("Genesis block on the Follower is identical to the one on the Leader", async function () {
+        const leaderGenesis = await getBlock("stratus", 0);
+        const followerGenesis = await getBlock("stratus-follower", 0);
+
+        expect(followerGenesis, "genesis blocks differ between leader and follower").to.deep.equal(leaderGenesis);
+    });
+
+    it("Send transactions to the Leader so that the inspected blocks carry a transactions root", async function () {
+        updateProviderUrl("stratus");
+
+        for (const [sender, receiver] of [
+            [ALICE, BOB],
+            [BOB, CHARLIE],
+            [CHARLIE, ALICE],
+        ]) {
+            const nonce = parseInt(await sendWithRetry("eth_getTransactionCount", [sender.address, "latest"]), 16);
+            const signedTx = await sender.signWeiTransfer(receiver.address, 0, nonce);
+            const txHash = keccak256(signedTx);
+
+            const response = await sendAndGetFullResponse("eth_sendRawTransaction", [signedTx]);
+            expect(response.data.result).to.equal(txHash);
+
+            await sendWithRetry("eth_getTransactionReceipt", [txHash]);
+        }
+    });
+
+    it("Wait for Follower to sync with Leader", async function () {
+        const { leaderBlock } = await waitForFollowerToSyncWithLeader();
+        const syncedBlock = parseInt(leaderBlock, 16);
+        expect(syncedBlock, "chain should have blocks past the genesis").to.be.greaterThan(0);
+
+        const oldest = Math.max(1, syncedBlock - INSPECTED_BLOCKS + 1);
+        const recent = Array.from({ length: syncedBlock - oldest + 1 }, (_, i) => oldest + i);
+        inspectedBlocks = recent[0] === 1 ? recent : [1, ...recent];
+    });
+
+    it("Leader hashes every block over its number, timestamp, transactions root and parent hash", async function () {
+        let blocksWithTransactions = 0;
+
+        for (const blockNumber of inspectedBlocks) {
+            const block = await getBlock("stratus", blockNumber);
+
+            expect(block.hash, `hash of block ${blockNumber}`).to.equal(calculateBlockHash(block));
+
+            if (block.transactionsRoot !== EMPTY_TRANSACTIONS_ROOT) {
+                blocksWithTransactions++;
+            }
+        }
+
+        // without this the transactions root would be constant and its contribution to the hash unverified
+        expect(blocksWithTransactions, "no inspected block mined a transaction").to.be.greaterThan(0);
+    });
+
+    it("Leader chains every block to the hash of its parent", async function () {
+        for (const blockNumber of inspectedBlocks) {
+            const block = await getBlock("stratus", blockNumber);
+            const parent = await getBlock("stratus", blockNumber - 1);
+
+            expect(block.parentHash, `parent hash of block ${blockNumber}`).to.equal(parent.hash);
+        }
+    });
+
+    it("Follower reports the same hashes as the Leader for every block", async function () {
+        for (const blockNumber of [0, ...inspectedBlocks]) {
+            const leaderBlock = await getBlock("stratus", blockNumber);
+            const followerBlock = await getBlock("stratus-follower", blockNumber);
+
+            expect(followerBlock.hash, `hash of block ${blockNumber}`).to.equal(leaderBlock.hash);
+            expect(followerBlock.parentHash, `parent hash of block ${blockNumber}`).to.equal(leaderBlock.parentHash);
+            expect(followerBlock.timestamp, `timestamp of block ${blockNumber}`).to.equal(leaderBlock.timestamp);
+            expect(followerBlock.transactionsRoot, `transactions root of block ${blockNumber}`).to.equal(
+                leaderBlock.transactionsRoot,
+            );
+        }
+    });
+
+    it("Follower stores hashes that match the fields of the blocks it imported", async function () {
+        for (const blockNumber of inspectedBlocks) {
+            const block = await getBlock("stratus-follower", blockNumber);
+
+            expect(block.hash, `hash of block ${blockNumber}`).to.equal(calculateBlockHash(block));
+        }
+    });
+
+    it("Both nodes resolve blocks by the hash the other one reports", async function () {
+        for (const blockNumber of [0, ...inspectedBlocks.slice(-1)]) {
+            const leaderBlock = await getBlock("stratus", blockNumber);
+
+            updateProviderUrl("stratus-follower");
+            const byHashOnFollower = await sendWithRetry("eth_getBlockByHash", [leaderBlock.hash, false]);
+            expect(byHashOnFollower.number, `block ${blockNumber} by hash on follower`).to.equal(leaderBlock.number);
+
+            updateProviderUrl("stratus");
+            const byHashOnLeader = await sendWithRetry("eth_getBlockByHash", [leaderBlock.hash, false]);
+            expect(byHashOnLeader.number, `block ${blockNumber} by hash on leader`).to.equal(leaderBlock.number);
+        }
+    });
+});
diff --git a/e2e/contracts/TestBlockHash.sol b/e2e/contracts/TestBlockHash.sol
new file mode 100644
index 000000000..f9c8be44f
--- /dev/null
+++ b/e2e/contracts/TestBlockHash.sol
@@ -0,0 +1,47 @@
+// SPDX-License-Identifier: MIT
+pragma solidity ^0.8.0;
+
+contract TestBlockHash {
+    event BlockHashRecorded(uint256 blockNumber, bytes32 blockHash);
+
+    mapping(uint256 => bytes32) public recordedBlockHashes;
+
+    /// @dev Reads the BLOCKHASH opcode for an arbitrary block.
+    /// @return The hash of the block, or zero when it is outside the 256 block window
+    function getBlockHash(uint256 blockNumber) external view returns (bytes32) {
+        return blockhash(blockNumber);
+    }
+
+    /// @dev Reads the BLOCKHASH opcode for the block being executed, which the EVM always answers with zero.
+    /// @return The number of the block being executed and its hash
+    function getCurrentBlockHash() external view returns (uint256, bytes32) {
+        return (block.number, blockhash(block.number));
+    }
+
+    /// @dev Reads the BLOCKHASH opcode for the parent of the block being executed. The parent number is returned
+    ///      along with the hash so that callers can resolve it without racing against newly mined blocks.
+    /// @return The number and the hash of the parent block
+    function getParentBlockHash() external view returns (uint256, bytes32) {
+        uint256 parentNumber = block.number - 1;
+        return (parentNumber, blockhash(parentNumber));
+    }
+
+    /// @dev Same as `getBlockHash`, but executed while mining a block instead of during a call.
+    /// @return The hash of the block
+    function recordBlockHash(uint256 blockNumber) external returns (bytes32) {
+        bytes32 blockHash = blockhash(blockNumber);
+        recordedBlockHashes[blockNumber] = blockHash;
+        emit BlockHashRecorded(blockNumber, blockHash);
+        return blockHash;
+    }
+
+    /// @dev Records the hash of the parent of the block that mines this transaction.
+    /// @return The number and the hash of the parent block
+    function recordParentBlockHash() external returns (uint256, bytes32) {
+        uint256 parentNumber = block.number - 1;
+        bytes32 blockHash = blockhash(parentNumber);
+        recordedBlockHashes[parentNumber] = blockHash;
+        emit BlockHashRecorded(parentNumber, blockHash);
+        return (parentNumber, blockHash);
+    }
+}
diff --git a/e2e/test/automine/e2e-blockhash.test.ts b/e2e/test/automine/e2e-blockhash.test.ts
new file mode 100644
index 000000000..688b69193
--- /dev/null
+++ b/e2e/test/automine/e2e-blockhash.test.ts
@@ -0,0 +1,122 @@
+import { expect } from "chai";
+
+import { TestBlockHash } from "../../typechain-types";
+import { CHARLIE } from "../helpers/account";
+import { isStratus } from "../helpers/network";
+import {
+    ETHERJS,
+    HASH_ZERO,
+    SUCCESS,
+    deployTestBlockHash,
+    pollReceipt,
+    send,
+    sendClearCache,
+    sendEvmMine,
+    sendGetBlockNumber,
+    toHex,
+} from "../helpers/rpc";
+
+// Reads the hash a block reports through the JSON-RPC interface.
+async function blockHashOf(blockNumber: number): Promise<string> {
+    const block = await send("eth_getBlockByNumber", [toHex(blockNumber), false]);
+    expect(block, `block ${blockNumber} should exist`).to.not.be.null;
+    return block.hash;
+}
+
+describe("BLOCKHASH opcode", () => {
+    let contract: TestBlockHash;
+
+    before(async () => {
+        contract = await deployTestBlockHash();
+    });
+
+    it("yields zero for the block being executed", async () => {
+        const [blockNumber, blockHash] = await contract.getCurrentBlockHash();
+
+        expect(blockNumber).to.be.greaterThan(0n);
+        expect(blockHash).eq(HASH_ZERO);
+    });
+
+    it("yields zero for blocks that were not mined yet", async () => {
+        const latest = await sendGetBlockNumber();
+
+        expect(await contract.getBlockHash(latest + 1)).eq(HASH_ZERO);
+        expect(await contract.getBlockHash(latest + 1000)).eq(HASH_ZERO);
+    });
+
+    it("yields the hash the parent block reports", async () => {
+        await sendEvmMine();
+
+        const [parentNumber, parentHash] = await contract.getParentBlockHash();
+
+        expect(parentHash).to.not.eq(HASH_ZERO);
+        expect(parentHash).eq(await blockHashOf(Number(parentNumber)));
+    });
+
+    it("yields the hash of several consecutive blocks", async () => {
+        await sendEvmMine();
+        await sendEvmMine();
+
+        // the parent of the executing block is the most recent one the opcode can reach
+        const [parentNumber] = await contract.getParentBlockHash();
+
+        for (let blockNumber = Number(parentNumber); blockNumber > Number(parentNumber) - 3; blockNumber--) {
+            const blockHash = await contract.getBlockHash(blockNumber);
+            expect(blockHash, `blockhash of block ${blockNumber}`).to.not.eq(HASH_ZERO);
+            expect(blockHash, `blockhash of block ${blockNumber}`).eq(await blockHashOf(blockNumber));
+        }
+    });
+
+    it("yields the parent hash of the block that mines the transaction", async () => {
+        const receipt = await pollReceipt(contract.connect(CHARLIE.signer()).recordParentBlockHash());
+        expect(receipt.status).eq(SUCCESS);
+
+        const parentNumber = receipt.blockNumber - 1;
+        const recorded = await contract.recordedBlockHashes(parentNumber);
+
+        expect(recorded).to.not.eq(HASH_ZERO);
+        expect(recorded).eq(await blockHashOf(parentNumber));
+
+        // the opcode must agree with the parent hash stamped on the block that mined the transaction
+        const minedBlock = await ETHERJS.getBlock(receipt.blockNumber);
+        expect(recorded).eq(minedBlock?.parentHash);
+    });
+
+    it("yields the same hash for a call and for a transaction", async () => {
+        const target = (await sendGetBlockNumber()) - 1;
+
+        const receipt = await pollReceipt(contract.connect(CHARLIE.signer()).recordBlockHash(target));
+        expect(receipt.status).eq(SUCCESS);
+
+        const recorded = await contract.recordedBlockHashes(target);
+        expect(recorded).to.not.eq(HASH_ZERO);
+        expect(recorded).eq(await contract.getBlockHash(target));
+    });
+
+    it("yields blocks hashes read back from the permanent storage", async function () {
+        if (!isStratus) {
+            this.skip();
+            return;
+        }
+
+        await sendEvmMine();
+
+        // drops the hashes this node published while mining, forcing the reads to hit the permanent storage
+        await sendClearCache();
+
+        const [parentNumber, parentHash] = await contract.getParentBlockHash();
+
+        expect(parentHash).to.not.eq(HASH_ZERO);
+        expect(parentHash).eq(await blockHashOf(Number(parentNumber)));
+        expect(await contract.getBlockHash(Number(parentNumber) - 1)).eq(await blockHashOf(Number(parentNumber) - 1));
+    });
+
+    it("chains every mined block to the hash of its parent", async () => {
+        const latest = await sendGetBlockNumber();
+
+        for (let blockNumber = latest; blockNumber > latest - 5; blockNumber--) {
+            const block = await ETHERJS.getBlock(blockNumber);
+            expect(block?.parentHash, `parent hash of block ${blockNumber}`).eq(await blockHashOf(blockNumber - 1));
+        }
+    });
+});
diff --git a/e2e/test/helpers/rpc.ts b/e2e/test/helpers/rpc.ts
index 42e0a9f3d..05b750cb9 100644
--- a/e2e/test/helpers/rpc.ts
+++ b/e2e/test/helpers/rpc.ts
@@ -21,6 +21,7 @@ import { Numbers } from "web3-types";
 import { WebSocket } from "ws";
 
 import {
+    TestBlockHash,
     TestContractBalances,
     TestContractBlockTimestamp,
     TestContractCounter,
@@ -236,6 +237,12 @@ export async function deployTestContractBlockTimestamp(): Promise<TestContractBl
     return await testContractFactory.connect(CHARLIE.signer()).deploy();
 }
 
+// Deploys the "TestBlockHash" contract.
+export async function deployTestBlockHash(): Promise<TestBlockHash> {
+    const testContractFactory = await ethers.getContractFactory("TestBlockHash");
+    return await testContractFactory.connect(CHARLIE.signer()).deploy();
+}
+
 // Deploys the "TestContractCounter" contract.
 export async function deployTestContractCounter(): Promise<TestContractCounter> {
     const testContractFactory = await ethers.getContractFactory("TestContractCounter");
diff --git a/src/bin/importer_offline.rs b/src/bin/importer_offline.rs
index 74949acb5..575d082f1 100644
--- a/src/bin/importer_offline.rs
+++ b/src/bin/importer_offline.rs
@@ -56,9 +56,15 @@ fn main() -> anyhow::Result<()> {
     global_services.runtime.block_on(run(global_services.config))
 }
 
-async fn run(config: ImporterOfflineConfig) -> anyhow::Result<()> {
+async fn run(mut config: ImporterOfflineConfig) -> anyhow::Result<()> {
     let _timer = DropTimer::start("importer-offline");
 
+    // The executor can seal one batch while the bounded channel is full and the saver is processing
+    // another one. Those unsaved hashes must remain available when imported transactions execute
+    // BLOCKHASH. This is a temporary requirement of the pipelined offline importer.
+    let unsaved_hash_capacity = config.block_saver_batch_size.saturating_mul(config.block_saver_queue_size.saturating_add(2));
+    config.storage.cache.block_hash_cache_capacity = config.storage.cache.block_hash_cache_capacity.saturating_add(unsaved_hash_capacity);
+
     // init services
     let rpc_storage = config.rpc_storage.init().await?;
     let storage = config.storage.init()?;
@@ -92,8 +98,9 @@ async fn run(config: ImporterOfflineConfig) -> anyhow::Result<()> {
 
     if block_start.is_zero() && !storage.has_genesis()? {
         let genesis_block = Block::genesis();
+        let genesis_hash = genesis_block.hash();
+        storage.publish_block_hash(BlockNumber::ZERO, genesis_hash);
         storage.save_genesis_block(genesis_block, initial_accounts, ExecutionChanges::default())?;
-        storage.finish_pending_block()?;
         block_start = BlockNumber::from(1);
     }
 
@@ -238,8 +245,9 @@ fn run_external_block_executor(
                     return Ok(());
                 }
 
-                executor.execute_external_block(block.clone(), ExternalReceipts::from(receipts))?;
-                let mined_block = miner.mine_external(block)?;
+                let session = miner.pending_session();
+                executor.execute_external_block(&session, block.clone(), ExternalReceipts::from(receipts))?;
+                let mined_block = session.seal_external(block)?;
                 executed_batch.push(mined_block);
             }
 
diff --git a/src/eth/executor/evm.rs b/src/eth/executor/evm.rs
index 76cc5ede1..34db11a44 100644
--- a/src/eth/executor/evm.rs
+++ b/src/eth/executor/evm.rs
@@ -54,6 +54,7 @@ use crate::eth::executor::ExecutorConfig;
 use crate::eth::primitives::Account;
 use crate::eth::primitives::Address;
 use crate::eth::primitives::BlockFilter;
+use crate::eth::primitives::BlockNumber;
 use crate::eth::primitives::Bytes;
 use crate::eth::primitives::EvmExecution;
 use crate::eth::primitives::EvmExecutionMetrics;
@@ -487,8 +488,8 @@ impl Database for RevmSession {
         Ok(slot.value.into())
     }
 
-    fn block_hash(&mut self, _: u64) -> Result<B256, StratusError> {
-        Err(anyhow!("block hash opcode not implemented").into())
+    fn block_hash(&mut self, number: u64) -> Result<B256, StratusError> {
+        self.block_hash_ref(number)
     }
 }
 
@@ -513,8 +514,17 @@ impl DatabaseRef for RevmSession {
         Ok(slot.value.into())
     }
 
-    fn block_hash_ref(&self, _: u64) -> Result<B256, Self::Error> {
-        Err(anyhow!("block hash opcode not implemented").into())
+    /// Resolves a block hash for the `BLOCKHASH` opcode.
+    ///
+    /// The interpreter only asks for blocks inside the 256 block window that precedes the block
+    /// being executed, so every request here is for a block that was already mined. Failing to
+    /// resolve it means the node lost track of its own chain, which must not be served as zero.
+    fn block_hash_ref(&self, number: u64) -> Result<B256, Self::Error> {
+        let number = BlockNumber::from(number);
+        match self.storage.read_block_hash(number)? {
+            Some(hash) => Ok(hash.into()),
+            None => Err(StorageError::BlockHashMissing { number }.into()),
+        }
     }
 
     fn code_by_hash_ref(&self, _code_hash: B256) -> Result<revm::state::Bytecode, Self::Error> {
diff --git a/src/eth/executor/executor.rs b/src/eth/executor/executor.rs
index d22ac6561..a95f11167 100644
--- a/src/eth/executor/executor.rs
+++ b/src/eth/executor/executor.rs
@@ -11,6 +11,7 @@ use anyhow::anyhow;
 use anyhow::bail;
 use cfg_if::cfg_if;
 use parking_lot::Mutex;
+use parking_lot::MutexGuard;
 use tracing::Span;
 use tracing::debug_span;
 #[cfg(feature = "tracing")]
@@ -26,6 +27,7 @@ use crate::eth::executor::EvmInput;
 use crate::eth::executor::ExecutorConfig;
 use crate::eth::executor::evm::EvmKind;
 use crate::eth::miner::Miner;
+use crate::eth::miner::miner::PendingSession;
 use crate::eth::primitives::BlockNumber;
 use crate::eth::primitives::CallInput;
 use crate::eth::primitives::EvmExecution;
@@ -259,6 +261,10 @@ pub struct ExecutorLocks {
     transaction: Mutex<()>,
 }
 
+pub(crate) struct TransactionGuard<'a> {
+    _guard: MutexGuard<'a, ()>,
+}
+
 pub struct Executor {
     /// Executor inner locks.
     locks: ExecutorLocks,
@@ -285,6 +291,12 @@ impl Executor {
         }
     }
 
+    pub(crate) fn transaction_guard(&self) -> TransactionGuard<'_> {
+        TransactionGuard {
+            _guard: self.locks.transaction.lock(),
+        }
+    }
+
     // -------------------------------------------------------------------------
     // External transactions
     // -------------------------------------------------------------------------
@@ -292,7 +304,7 @@ impl Executor {
     /// Reexecutes an external block locally and imports it to the temporary storage.
     ///
     /// Returns the remaining receipts that were not consumed by the execution.
-    pub fn execute_external_block(&self, mut block: ExternalBlock, mut receipts: ExternalReceipts) -> anyhow::Result<()> {
+    pub fn execute_external_block(&self, session: &PendingSession<'_>, mut block: ExternalBlock, mut receipts: ExternalReceipts) -> anyhow::Result<()> {
         // track
         #[cfg(feature = "metrics")]
         let (start, mut block_metrics) = (metrics::now(), EvmExecutionMetrics::default());
@@ -301,7 +313,7 @@ impl Executor {
         let _span = info_span!("executor::external_block", block_number = %block.number()).entered();
         tracing::info!(block_number = %block.number(), "reexecuting external block");
 
-        self.storage.set_pending_from_external(&block);
+        session.set_pending_from_external(&block);
 
         // track pending block
         let block_number = block.number();
@@ -312,6 +324,7 @@ impl Executor {
         for tx in block_transactions.into_transactions() {
             let receipt = receipts.try_remove(tx.hash())?;
             self.execute_external_transaction(
+                session,
                 tx,
                 receipt,
                 block_number,
@@ -338,6 +351,7 @@ impl Executor {
     /// to facilitate re-execution of parallel transactions that failed
     fn execute_external_transaction(
         &self,
+        session: &PendingSession<'_>,
         tx: ExternalTransaction,
         receipt: ExternalReceipt,
         block_number: BlockNumber,
@@ -426,7 +440,7 @@ impl Executor {
         }
 
         // persist state
-        self.miner.save_execution(tx_execution)?;
+        session.append_execution(tx_execution)?;
 
         // track metrics
         #[cfg(feature = "metrics")]
@@ -475,7 +489,7 @@ impl Executor {
         let _transaction_lock = self.locks.transaction.lock();
 
         // execute transaction
-        let tx_execution = self.execute_local_transaction_attempts(tx, INFINITE_ATTEMPTS);
+        let tx_execution = self.execute_local_transaction_attempts(tx, INFINITE_ATTEMPTS, None);
 
         #[cfg(feature = "metrics")]
         metrics::inc_executor_local_transaction(start.elapsed(), tx_execution.is_ok(), contract, function);
@@ -483,8 +497,23 @@ impl Executor {
         tx_execution
     }
 
+    pub(crate) fn execute_local_transaction_in_session(
+        &self,
+        _transaction_guard: &TransactionGuard<'_>,
+        session: &PendingSession<'_>,
+        tx: TransactionInput,
+    ) -> Result<(), StratusError> {
+        const INFINITE_ATTEMPTS: usize = usize::MAX;
+        self.execute_local_transaction_attempts(tx, INFINITE_ATTEMPTS, Some(session))
+    }
+
     /// Executes a transaction until it reaches the max number of attempts.
-    fn execute_local_transaction_attempts(&self, tx_input: TransactionInput, max_attempts: usize) -> Result<(), StratusError> {
+    fn execute_local_transaction_attempts(
+        &self,
+        tx_input: TransactionInput,
+        max_attempts: usize,
+        session: Option<&PendingSession<'_>>,
+    ) -> Result<(), StratusError> {
         // validate
         if tx_input.signer().is_zero() {
             return Err(TransactionError::FromZeroAddress.into());
@@ -552,7 +581,11 @@ impl Executor {
                 metrics::inc_executor_local_transaction_reverts(contract, function, reason.0.as_ref());
             }
 
-            match self.miner.save_execution(tx_execution) {
+            let save_result = match session {
+                Some(session) => session.append_execution(tx_execution),
+                None => self.miner.save_execution(tx_execution),
+            };
+            match save_result {
                 Ok(_) => {
                     // track metrics
                     #[cfg(feature = "metrics")]
diff --git a/src/eth/follower/importer/importers/execution.rs b/src/eth/follower/importer/importers/execution.rs
index c2b1a53eb..830b2b499 100644
--- a/src/eth/follower/importer/importers/execution.rs
+++ b/src/eth/follower/importer/importers/execution.rs
@@ -35,23 +35,31 @@ impl ImporterWorker for ReexecutionWorker {
         const TASK_NAME: &str = "block-executor";
 
         let receipts_len = receipts.len();
+        let (mined_block, changes) = {
+            let session = self.miner.pending_session();
 
-        if let Err(e) = self.executor.execute_external_block(block.clone(), ExternalReceipts::from(receipts)) {
-            let message = GlobalState::shutdown_from(TASK_NAME, "failed to reexecute external block");
-            return log_and_err!(reason = e, message);
-        };
-
-        let (mined_block, changes) = match self.miner.mine_external(block) {
-            Ok((mined_block, changes)) => {
-                tracing::info!(number = %mined_block.number(), "mined external block");
-                (mined_block, changes)
-            }
-            Err(e) => {
-                let message = GlobalState::shutdown_from(TASK_NAME, "failed to mine external block");
+            if let Err(e) = self.executor.execute_external_block(&session, block.clone(), ExternalReceipts::from(receipts)) {
+                let message = GlobalState::shutdown_from(TASK_NAME, "failed to reexecute external block");
                 return log_and_err!(reason = e, message);
+            };
+
+            match session.seal_external(block) {
+                Ok((mined_block, changes)) => {
+                    tracing::info!(number = %mined_block.number(), "mined external block");
+                    (mined_block, changes)
+                }
+                Err(e) => {
+                    let message = GlobalState::shutdown_from(TASK_NAME, "failed to mine external block");
+                    return log_and_err!(reason = e, message);
+                }
             }
         };
 
+        if let Err(e) = self.miner.validate_next_saved_block(&mined_block) {
+            let message = GlobalState::shutdown_from(TASK_NAME, "external block failed continuity preflight");
+            return log_and_err!(reason = e, message);
+        }
+
         send_block_to_kafka(&self.kafka_connector, &mined_block).await?;
 
         match self.miner.commit(CommitItem::Block(mined_block), changes) {
diff --git a/src/eth/follower/importer/importers/fake_leader.rs b/src/eth/follower/importer/importers/fake_leader.rs
index 43502d403..4cf9126a3 100644
--- a/src/eth/follower/importer/importers/fake_leader.rs
+++ b/src/eth/follower/importer/importers/fake_leader.rs
@@ -11,7 +11,6 @@ use crate::eth::follower::importer::importers::ImportData;
 use crate::eth::follower::importer::importers::ImporterWorker;
 use crate::eth::miner::Miner;
 use crate::eth::miner::miner::interval_miner::commit_retry;
-use crate::eth::miner::miner::interval_miner::mine_local_retry;
 use crate::eth::primitives::Block;
 use crate::eth::primitives::EvmExecutionMetrics;
 use crate::eth::primitives::ExecutionChanges;
@@ -37,10 +36,13 @@ impl ImporterWorker for FakeLeaderWorker {
 
     async fn import(&self, ((block, _), (expected_block, expected_changes)): Self::DataType) -> anyhow::Result<usize> {
         let block_tx_len = block.transactions.len();
-        self.storage.set_pending_from_external(&block);
+        let transaction_guard = self.executor.transaction_guard();
+        let mine_and_commit_guard = self.miner.locks.mine_and_commit.lock();
+        let session = self.miner.pending_session();
+        session.set_pending_from_external(&block);
         for tx in block.0.transactions.into_transactions() {
             tracing::info!(?tx, "executing tx as fake miner");
-            if let Err(e) = self.executor.execute_local_transaction(tx.try_into()?) {
+            if let Err(e) = self.executor.execute_local_transaction_in_session(&transaction_guard, &session, tx.try_into()?) {
                 match e {
                     StratusError::Transaction(TransactionError::Nonce { transaction: _, account: _ }) => {
                         tracing::warn!(reason = ?e, "transaction failed, was this node restarted?");
@@ -53,7 +55,8 @@ impl ImporterWorker for FakeLeaderWorker {
                 }
             }
         }
-        let (mined_block, changes, miner_guard) = mine_local_retry(&self.miner);
+        let (mined_block, changes) = session.seal_local();
+        drop(transaction_guard);
 
         let completed_expected_changes = expected_changes.complete(self.storage.as_ref())?;
         if changes != completed_expected_changes {
@@ -75,7 +78,7 @@ impl ImporterWorker for FakeLeaderWorker {
             bail!("block mismatch between leader and fake leader")
         }
 
-        commit_retry(&self.miner, mined_block, changes, miner_guard);
+        commit_retry(&self.miner, mined_block, changes, mine_and_commit_guard);
         Ok(block_tx_len)
     }
 }
diff --git a/src/eth/follower/importer/importers/replication.rs b/src/eth/follower/importer/importers/replication.rs
index 2b1550da1..76278ea64 100644
--- a/src/eth/follower/importer/importers/replication.rs
+++ b/src/eth/follower/importer/importers/replication.rs
@@ -34,6 +34,7 @@ impl ImporterWorker for ReplicationWorker {
 
         let block_tx_len = block.transactions.len();
 
+        self.storage.validate_next_saved_block(&block)?;
         send_block_to_kafka(&self.kafka_connector, &block).await?;
 
         let completed_changes = changes.complete(self.storage.as_ref())?;
diff --git a/src/eth/follower/importer/mod.rs b/src/eth/follower/importer/mod.rs
index f5db39af9..e7c9ff4fb 100644
--- a/src/eth/follower/importer/mod.rs
+++ b/src/eth/follower/importer/mod.rs
@@ -302,7 +302,18 @@ mod tests {
     use crate::infra::BlockchainClient;
 
     /// Mines a block applying `changes` (mirrors the helper in `stratus_storage` tests).
-    fn mine_block(storage: &StratusStorage, changes: ExecutionChanges) {
+    fn initialize_genesis(storage: &StratusStorage) {
+        if storage.has_genesis().expect("read genesis") {
+            return;
+        }
+
+        let genesis = Block::genesis();
+        storage
+            .save_genesis_block(genesis, Vec::new(), ExecutionChanges::default())
+            .expect("save genesis");
+    }
+
+    fn seal_block(storage: &StratusStorage, miner: &Miner, changes: ExecutionChanges) -> (Block, ExecutionChanges) {
         let (header, _) = storage.read_pending_block_header();
         let evm_input = EvmInput::from_eth_transaction(&TransactionInput::default(), header.number, *header.timestamp);
 
@@ -311,10 +322,15 @@ mod tests {
         result.execution.changes = changes;
 
         let tx = TransactionExecution::new(TransactionInfo::default(), Signature::default(), ExecutionInfo::default(), evm_input, result);
-        storage.save_execution(tx).expect("save execution");
+        let session = miner.pending_session();
+        session.append_execution(tx).expect("save execution");
+        session.seal_local()
+    }
 
-        let (block, block_changes) = storage.finish_pending_block().expect("finish pending block");
-        storage.save_block(block.into(), block_changes).expect("save block");
+    fn mine_block(storage: &StratusStorage, miner: &Miner, changes: ExecutionChanges) -> Block {
+        let (block, block_changes) = seal_block(storage, miner, changes);
+        storage.save_block(block.clone(), block_changes).expect("save block");
+        block
     }
 
     /// Builds `ExecutionChanges` that set `address`'s balance to `balance` (nonce/bytecode untouched).
@@ -373,23 +389,26 @@ mod tests {
         let fetcher = BlockWithChangesFetcher { chain };
 
         let address = Address::new([0xCC; 20]);
+        initialize_genesis(&storage);
 
         // Block 1: B.balance = 100. permanent storage is now at block 1.
-        mine_block(&storage, balance_changes(address, Wei::from(100u64)));
+        mine_block(&storage, &worker.miner, balance_changes(address, Wei::from(100u64)));
 
         // The fetcher post-processes block 3 while the importer is still at block 1 (fetcher ahead).
         // Block 3 changed B's nonce but left its balance untouched (balance entry is `None`).
         // `post_process` returns `ExecutionChanges<Incomplete>` — it does NOT read perm, so the
         // fetcher being ahead does not corrupt the changes.
-        let fetched_3 = (
-            BlockRocksdb::from(Block::new(BlockNumber::from(3u64), UnixTime::from(0u64))),
-            block_changes_nonce_only(address),
-        );
+        // Seal intervening block 2 without saving it yet. This is the correct pre-state for block 3.
+        let (block_2, block_2_changes) = seal_block(&storage, &worker.miner, balance_changes(address, Wei::from(200u64)));
+
+        let mut block_3 = Block::new(BlockNumber::from(3u64), UnixTime::from(0u64));
+        block_3.header.parent_hash = block_2.hash();
+        block_3.apply_default_hash();
+        let fetched_3 = (BlockRocksdb::from(block_3), block_changes_nonce_only(address));
         let (block_3, changes_3) = fetcher.post_process(fetched_3).await.expect("post_process");
 
-        // Intervening block 2: B.balance = 200. This is the correct pre-state for block 3.
-        // permanent storage is now at block 2.
-        mine_block(&storage, balance_changes(address, Wei::from(200u64)));
+        // The saver catches permanent storage up to block 2 after block 3 was already post-processed.
+        storage.save_block(block_2, block_2_changes).expect("save block 2");
 
         // The importer imports block 3. `ReplicationWorker::import` must complete `changes_3`
         // (Incomplete) against perm at import time, when perm is caught up to block 2 (200).
diff --git a/src/eth/genesis.rs b/src/eth/genesis.rs
index 1d11637e5..c6ce08025 100644
--- a/src/eth/genesis.rs
+++ b/src/eth/genesis.rs
@@ -295,10 +295,11 @@ impl GenesisConfig {
         header.nonce = nonce_b64.into();
 
         // Create the block
-        let block = Block {
+        let mut block = Block {
             header,
             transactions: Vec::new(),
         };
+        block.header.hash = block.calculate_hash_v1();
 
         Ok(block)
     }
diff --git a/src/eth/miner/miner.rs b/src/eth/miner/miner.rs
index 174049f9f..c53357218 100644
--- a/src/eth/miner/miner.rs
+++ b/src/eth/miner/miner.rs
@@ -23,6 +23,8 @@ use crate::eth::primitives::LogMessage;
 use crate::eth::primitives::StorageError;
 use crate::eth::primitives::StratusError;
 use crate::eth::primitives::TransactionExecution;
+use crate::eth::storage::BlockReference;
+use crate::eth::storage::PendingBlockGuard;
 use crate::eth::storage::StratusStorage;
 use crate::ext::DisplayExt;
 use crate::ext::not;
@@ -78,12 +80,95 @@ pub struct Miner {
 /// Locks used in operations that mutate state.
 #[derive(Default)]
 pub struct MinerLocks {
-    save_execution: Mutex<()>,
     pub mine_and_commit: Mutex<()>,
-    mine: Mutex<()>,
     commit: Mutex<()>,
 }
 
+pub struct PendingSession<'a> {
+    miner: &'a Miner,
+    guard: PendingBlockGuard<'a>,
+}
+
+impl PendingSession<'_> {
+    pub(crate) fn set_pending_from_external(&self, block: &ExternalBlock) {
+        self.miner.storage.set_pending_from_external(&self.guard, block);
+    }
+
+    pub(crate) fn append_execution(&self, tx_execution: TransactionExecution) -> Result<(), StratusError> {
+        let tx_hash = tx_execution.info.hash;
+
+        #[cfg(feature = "tracing")]
+        let _span = info_span!("miner::save_execution", %tx_hash).entered();
+
+        self.miner.storage.save_execution(&self.guard, tx_execution)?;
+
+        if self.miner.has_pending_tx_subscribers() {
+            self.miner.send_pending_tx_notification(&Some(tx_hash));
+        }
+
+        Ok(())
+    }
+
+    pub fn seal_external(self, external_block: ExternalBlock) -> anyhow::Result<(Block, ExecutionChanges)> {
+        #[cfg(feature = "tracing")]
+        let _span = info_span!("miner::mine_external", block_number = field::Empty).entered();
+
+        let parent_hash = self.miner.storage.read_pending_parent_hash(&self.guard);
+        let (pending_block, changes) = self.miner.storage.pending_block_to_seal(&self.guard);
+        let timestamp = pending_block.header.timestamp.clone();
+        let mut block = Block::from_pending(pending_block, parent_hash);
+
+        Span::with(|s| s.rec_str("block_number", &block.header.number));
+        let external_parent_hash = external_block.parent_hash();
+        block.apply_external(&external_block)?;
+        // Preserve the imported parent so save_block can validate continuity against last_saved.
+        // V2 already commits to this field; the assignment is relevant to the temporary V1 fallback.
+        block.header.parent_hash = external_parent_hash;
+
+        match external_block == block {
+            true => {
+                self.miner.storage.finish_pending_block(&self.guard, BlockReference::from(&block), timestamp);
+                self.miner.storage.publish_block_hash(block.number(), block.hash());
+                Ok((block, changes))
+            }
+            false => Err(anyhow!(
+                "mismatching block info:\n\tlocal:\n\t\tnumber: {:?}\n\t\ttimestamp: {:?}\n\t\thash: {:?}\n\texternal:\n\t\tnumber: {:?}\n\t\ttimestamp: {:?}\n\t\thash: {:?}",
+                block.number(),
+                block.header.timestamp,
+                block.hash(),
+                external_block.number(),
+                external_block.timestamp(),
+                external_block.hash()
+            )),
+        }
+    }
+
+    pub(crate) fn seal_local(self) -> (Block, ExecutionChanges) {
+        let parent_hash = self.miner.storage.read_pending_parent_hash(&self.guard);
+        let (pending_block, changes) = self.miner.storage.pending_block_to_seal(&self.guard);
+        let timestamp = pending_block.header.timestamp.clone();
+        let block = Block::from_pending(pending_block, parent_hash);
+        self.miner.storage.finish_pending_block(&self.guard, BlockReference::from(&block), timestamp);
+        self.miner.storage.publish_block_hash(block.number(), block.hash());
+        Span::with(|s| s.rec_str("block_number", &block.header.number));
+
+        (block, changes)
+    }
+
+    fn seal_replication(self, block: &Block) {
+        self.miner.storage.set_pending_header(&self.guard, block.number(), block.timestamp());
+        self.miner
+            .storage
+            .finish_pending_block(&self.guard, BlockReference::from(block), block.timestamp().into());
+        self.miner.storage.publish_block_hash(block.number(), block.hash());
+    }
+
+    #[cfg(feature = "dev")]
+    fn reset_to_genesis(&self) -> Result<(), StorageError> {
+        self.miner.storage.reset_to_genesis(&self.guard)
+    }
+}
+
 impl Miner {
     pub fn new(storage: Arc<StratusStorage>, mode: MinerMode) -> Self {
         tracing::info!(?mode, "creating block miner");
@@ -100,6 +185,21 @@ impl Miner {
         }
     }
 
+    pub fn pending_session(&self) -> PendingSession<'_> {
+        PendingSession {
+            miner: self,
+            guard: self.storage.pending_block_guard(),
+        }
+    }
+
+    #[cfg(feature = "dev")]
+    pub fn reset_to_genesis(&self) -> Result<(), StorageError> {
+        let _mine_and_commit_guard = self.locks.mine_and_commit.lock();
+        let session = self.pending_session();
+        let _commit_guard = self.locks.commit.lock();
+        session.reset_to_genesis()
+    }
+
     /// Spawns a new thread that keep mining blocks in the specified interval.
     ///
     /// Also unpauses `Miner` if it was paused.
@@ -197,72 +297,27 @@ impl Miner {
 
     /// Persists a transaction execution.
     pub fn save_execution(&self, tx_execution: TransactionExecution) -> Result<(), StratusError> {
-        let tx_hash = tx_execution.info.hash;
-
-        // track
-        #[cfg(feature = "tracing")]
-        let _span = info_span!("miner::save_execution", %tx_hash).entered();
-
         // Check if automine is enabled
         let is_automine = self.mode().is_automine();
 
-        // if automine is enabled, only one transaction can enter the block at a time.
-        let _save_execution_lock = if is_automine { Some(self.locks.save_execution.lock()) } else { None };
-
-        // save execution to temporary storage
-        self.storage.save_execution(tx_execution)?;
-
-        // notify
-        if self.has_pending_tx_subscribers() {
-            self.send_pending_tx_notification(&Some(tx_hash));
-        }
-
-        // if automine is enabled, automatically mines a block
         if is_automine {
-            self.mine_local_and_commit()?;
+            let _mine_and_commit_lock = self.locks.mine_and_commit.lock();
+            let session = self.pending_session();
+            session.append_execution(tx_execution)?;
+            let (block, changes) = session.seal_local();
+            self.commit(CommitItem::Block(block), changes)?;
+        } else {
+            self.pending_session().append_execution(tx_execution)?;
         }
 
         Ok(())
     }
 
-    /// Mines external block and external transactions.
-    ///
-    /// Local transactions are not allowed to be part of the block.
-    pub fn mine_external(&self, external_block: ExternalBlock) -> anyhow::Result<(Block, ExecutionChanges)> {
-        // track
-        #[cfg(feature = "tracing")]
-        let _span = info_span!("miner::mine_external", block_number = field::Empty).entered();
-
-        // lock
-        let _mine_lock = self.locks.mine.lock();
-
-        // mine block
-        let (pending_block, changes) = self.storage.finish_pending_block()?;
-        let mut block: Block = pending_block.into();
-
-        Span::with(|s| s.rec_str("block_number", &block.header.number));
-        block.apply_external(&external_block);
-
-        match external_block == block {
-            true => Ok((block, changes)),
-            false => Err(anyhow!(
-                "mismatching block info:\n\tlocal:\n\t\tnumber: {:?}\n\t\ttimestamp: {:?}\n\t\thash: {:?}\n\texternal:\n\t\tnumber: {:?}\n\t\ttimestamp: {:?}\n\t\thash: {:?}",
-                block.number(),
-                block.header.timestamp,
-                block.hash(),
-                external_block.number(),
-                external_block.timestamp(),
-                external_block.hash()
-            )),
-        }
-    }
-
     /// Same as [`Self::mine_local`], but automatically commits the block instead of returning it.
     /// mainly used when is_automine is enabled.
     pub fn mine_local_and_commit(&self) -> anyhow::Result<(), StorageError> {
         let _mine_and_commit_lock = self.locks.mine_and_commit.lock();
-
-        let (block, changes) = self.mine_local()?;
+        let (block, changes) = self.pending_session().seal_local();
         self.commit(CommitItem::Block(block), changes)
     }
 
@@ -273,22 +328,18 @@ impl Miner {
         #[cfg(feature = "tracing")]
         let _span = info_span!("miner::mine_local", block_number = field::Empty).entered();
 
-        // lock
-        let _mine_lock = self.locks.mine.lock();
-
-        // mine block
-        let (block, changes) = self.storage.finish_pending_block()?;
-        Span::with(|s| s.rec_str("block_number", &block.header.number));
+        Ok(self.pending_session().seal_local())
+    }
 
-        Ok((block.into(), changes))
+    pub(crate) fn validate_next_saved_block(&self, block: &Block) -> Result<(), StorageError> {
+        self.storage.validate_next_saved_block(block)
     }
 
     pub fn commit(&self, item: CommitItem, changes: ExecutionChanges) -> anyhow::Result<(), StorageError> {
         match item {
             CommitItem::Block(block) => self.commit_block(block, changes),
             CommitItem::ReplicationBlock(block) => {
-                self.storage.set_pending_header(block.number(), block.timestamp());
-                self.storage.finish_pending_block()?;
+                self.pending_session().seal_replication(&block);
                 self.commit_block(block, changes)
             }
         }
@@ -488,3 +539,102 @@ mod interval_miner_ticker {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use fake::Fake;
+    use fake::Faker;
+
+    use super::*;
+    use crate::eth::primitives::BlockNumber;
+
+    fn initialize_genesis(storage: &Arc<StratusStorage>) -> Block {
+        if let Some(genesis) = storage.read_block(crate::eth::primitives::BlockFilter::Number(BlockNumber::ZERO)).unwrap() {
+            return genesis;
+        }
+
+        let genesis = Block::genesis();
+        storage
+            .save_genesis_block(genesis.clone(), Vec::new(), ExecutionChanges::default())
+            .expect("save genesis block");
+        genesis
+    }
+
+    #[test]
+    fn local_mining_uses_latest_sealed_hash_when_cache_is_empty() {
+        let storage = Arc::new(StratusStorage::new_test().expect("create test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::Automine);
+        let genesis = initialize_genesis(&storage);
+        storage.clear_cache();
+
+        let (block, _) = miner.mine_local().expect("mine local block");
+
+        assert_eq!(block.number(), BlockNumber::ONE);
+        assert_eq!(block.header.parent_hash, genesis.hash());
+    }
+
+    #[test]
+    fn invalid_external_hash_does_not_advance_pending_block() {
+        let storage = Arc::new(StratusStorage::new_test().expect("create test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::External);
+        initialize_genesis(&storage);
+
+        let mut external_block: ExternalBlock = Faker.fake();
+        external_block.0.header.inner.number = 1;
+        external_block.0.header.hash = alloy_primitives::B256::ZERO;
+
+        let session = miner.pending_session();
+        session.set_pending_from_external(&external_block);
+        session.seal_external(external_block).expect_err("invalid external hash should be rejected");
+
+        assert_eq!(storage.read_pending_block_header().0.number, BlockNumber::ONE);
+    }
+
+    #[test]
+    fn legacy_external_parent_is_validated_when_saved() {
+        let storage = Arc::new(StratusStorage::new_test().expect("create test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::External);
+        let genesis = initialize_genesis(&storage);
+
+        let mut external_block: ExternalBlock = Faker.fake();
+        external_block.0.header.inner.number = 1;
+        external_block.0.header.inner.parent_hash = alloy_primitives::B256::ZERO;
+        external_block.0.header.hash = BlockNumber::ONE.hash().into();
+        external_block.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(Vec::new());
+
+        let session = miner.pending_session();
+        session.set_pending_from_external(&external_block);
+        let (block, changes) = session.seal_external(external_block).expect("legacy hash should be accepted while importing");
+
+        assert!(matches!(
+            miner.validate_next_saved_block(&block),
+            Err(StorageError::ParentHashConflict { number, local, external })
+                if number == BlockNumber::ONE && local == genesis.hash() && external == Hash::ZERO
+        ));
+        let error = storage.save_block(block, changes).expect_err("disconnected external parent should be rejected");
+        assert!(matches!(
+            error,
+            StorageError::ParentHashConflict { number, local, external }
+                if number == BlockNumber::ONE && local == genesis.hash() && external == Hash::ZERO
+        ));
+    }
+
+    #[test]
+    fn pending_session_serializes_pending_writers() {
+        let storage = Arc::new(StratusStorage::new_test().expect("create test storage"));
+        let miner = Arc::new(Miner::new(storage, MinerMode::External));
+        let first_session = miner.pending_session();
+        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
+
+        let other_miner = Arc::clone(&miner);
+        let handle = std::thread::spawn(move || {
+            let _session = other_miner.pending_session();
+            acquired_tx.send(()).expect("notify guard acquisition");
+        });
+
+        assert!(acquired_rx.recv_timeout(Duration::from_millis(20)).is_err());
+        drop(first_session);
+        acquired_rx.recv_timeout(Duration::from_secs(1)).expect("second writer should acquire guard");
+        handle.join().expect("join guard thread");
+    }
+}
diff --git a/src/eth/primitives/block.rs b/src/eth/primitives/block.rs
index e6041eed3..203936b48 100644
--- a/src/eth/primitives/block.rs
+++ b/src/eth/primitives/block.rs
@@ -1,6 +1,8 @@
 use alloy_primitives::B256;
+use alloy_primitives::keccak256;
 use alloy_rpc_types_eth::BlockTransactions;
 use alloy_trie::root::ordered_trie_root;
+use anyhow::bail;
 use display_json::DebugAsJson;
 use itertools::Itertools;
 
@@ -39,7 +41,37 @@ impl Block {
 
     /// Constructs an empty genesis block.
     pub fn genesis() -> Block {
-        Block::new(BlockNumber::ZERO, UnixTime::from(1702568764))
+        let mut block = Block::new(BlockNumber::ZERO, UnixTime::from(1702568764));
+        block.header.hash = block.calculate_hash_v1();
+        block
+    }
+
+    /// Converts a finished pending block into a mined block chained to `parent_hash`.
+    ///
+    /// The resulting block is hashed and the hash is stamped on all of its transactions.
+    pub fn from_pending(pending: PendingBlock, parent_hash: Hash) -> Block {
+        let mut block = Block::new(pending.header.number, *pending.header.timestamp);
+        if !block.number().is_zero() {
+            block.header.parent_hash = parent_hash;
+        }
+
+        let txs: Vec<TransactionExecution> = pending.transactions.into_values().collect();
+        block.transactions.reserve(txs.len());
+        block.header.size = Size::from(txs.len() as u64);
+
+        let mut log_index = Index::ZERO;
+        for (tx_idx, execution) in txs.into_iter().enumerate() {
+            let log_count = execution.result.execution.logs.len() as u64;
+            let transaction_mined = TransactionMined::from_execution(execution, Hash::ZERO, (tx_idx as u64).into(), log_index);
+            block.header.gas_used += transaction_mined.execution.result.execution.gas_used;
+            block.transactions.push(transaction_mined);
+            log_index += Index(log_count);
+        }
+
+        block.calculate_transaction_root();
+        block.apply_default_hash();
+
+        block
     }
 
     /// Serializes itself to JSON-RPC block format with full transactions included.
@@ -92,35 +124,60 @@ impl Block {
         }
     }
 
-    pub fn apply_external(&mut self, external_block: &ExternalBlock) {
-        self.header.hash = external_block.hash();
-        assert!(*self.header.timestamp == external_block.header.timestamp);
+    pub fn calculate_hash_v1(&self) -> Hash {
+        self.number().hash()
+    }
+
+    pub fn calculate_hash_v2(&self) -> Hash {
+        let mut input = [0_u8; 80];
+        input[0..8].copy_from_slice(&self.number().as_u64().to_be_bytes());
+        input[8..16].copy_from_slice(&self.header.timestamp.to_be_bytes());
+        input[16..48].copy_from_slice(self.header.transactions_root.as_ref());
+        input[48..80].copy_from_slice(self.header.parent_hash.as_ref());
+        keccak256(input).into()
+    }
+
+    pub fn calculate_hash_default(&self) -> Hash {
+        if self.number().is_zero() {
+            self.calculate_hash_v1()
+        } else {
+            self.calculate_hash_v2()
+        }
+    }
+
+    pub fn apply_hash(&mut self, hash: Hash) {
+        self.header.hash = hash;
         for transaction in self.transactions.iter_mut() {
-            assert!(transaction.evm_input.block_timestamp == self.header.timestamp);
-            transaction.mined_data.block_hash = external_block.hash();
+            transaction.mined_data.block_hash = hash;
         }
     }
-}
 
-impl From<PendingBlock> for Block {
-    fn from(value: PendingBlock) -> Self {
-        let mut block = Block::new(value.header.number, *value.header.timestamp);
-        let txs: Vec<TransactionExecution> = value.transactions.into_values().collect();
-        block.transactions.reserve(txs.len());
-        block.header.size = Size::from(txs.len() as u64);
+    pub fn apply_default_hash(&mut self) {
+        let hash = self.calculate_hash_default();
+        self.apply_hash(hash);
+    }
 
-        let mut log_index = Index::ZERO;
-        for (tx_idx, execution) in txs.into_iter().enumerate() {
-            let log_count = execution.result.execution.logs.len() as u64;
-            let transaction_mined = TransactionMined::from_execution(execution, block.hash(), (tx_idx as u64).into(), log_index);
-            block.header.gas_used += transaction_mined.execution.result.execution.gas_used;
-            block.transactions.push(transaction_mined);
-            log_index += Index(log_count);
+    pub fn apply_external(&mut self, external_block: &ExternalBlock) -> anyhow::Result<()> {
+        if *self.header.timestamp != external_block.header.timestamp {
+            bail!(
+                "mismatching block timestamp: local={} external={}",
+                *self.header.timestamp,
+                external_block.header.timestamp
+            );
         }
 
-        Self::calculate_transaction_root(&mut block);
+        let external_hash = external_block.hash();
+        let default_hash = self.calculate_hash_default();
+        if external_hash != default_hash {
+            // TODO: Remove the V1 hash arm after every node has been upgraded.
+            let v1_hash = self.calculate_hash_v1();
+            if external_hash != v1_hash {
+                bail!("invalid external block hash: imported={external_hash} default={default_hash} v1={v1_hash}");
+            }
+        }
 
-        block
+        self.apply_hash(external_hash);
+        Ok(())
     }
 }
 
@@ -150,3 +207,93 @@ impl From<Block> for AlloyBlockB256 {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use fake::Fake;
+    use fake::Faker;
+
+    use super::*;
+
+    fn block_with_v2_hash() -> Block {
+        let mut block = Block::new(BlockNumber::ONE, UnixTime::from(1234567890));
+        block.header.transactions_root = Hash::new([1; 32]);
+        block.header.parent_hash = Hash::new([2; 32]);
+        block.apply_default_hash();
+        block
+    }
+
+    fn external_block(block: &Block, hash: Hash) -> ExternalBlock {
+        let mut external: ExternalBlock = Faker.fake();
+        external.0.header.inner.number = block.number().as_u64();
+        external.0.header.inner.timestamp = *block.header.timestamp;
+        external.0.header.inner.parent_hash = block.header.parent_hash.into();
+        external.0.header.hash = hash.into();
+        external.0.transactions = BlockTransactions::Full(Vec::new());
+        external
+    }
+
+    #[test]
+    fn v2_hash_depends_on_finalized_block_fields() {
+        let block = block_with_v2_hash();
+
+        let mut changed = block.clone();
+        changed.header.number = BlockNumber::from(2_u64);
+        changed.apply_default_hash();
+        assert_ne!(block.hash(), changed.hash());
+
+        changed = block.clone();
+        changed.header.timestamp = UnixTime::from(*changed.header.timestamp + 1);
+        changed.apply_default_hash();
+        assert_ne!(block.hash(), changed.hash());
+
+        changed = block.clone();
+        changed.header.transactions_root = Hash::new([3; 32]);
+        changed.apply_default_hash();
+        assert_ne!(block.hash(), changed.hash());
+
+        changed = block.clone();
+        changed.header.parent_hash = Hash::new([4; 32]);
+        changed.apply_default_hash();
+        assert_ne!(block.hash(), changed.hash());
+    }
+
+    #[test]
+    fn external_hash_must_be_v2_or_v1() {
+        let block = block_with_v2_hash();
+
+        let mut v2_block = block.clone();
+        v2_block.header.hash = Hash::ZERO;
+        v2_block
+            .apply_external(&external_block(&block, block.hash()))
+            .expect("V2 hash should be accepted");
+        assert_eq!(v2_block.hash(), block.hash());
+
+        let v1_hash = block.calculate_hash_v1();
+        let mut v1_block = block.clone();
+        v1_block.header.hash = Hash::ZERO;
+        v1_block.apply_external(&external_block(&block, v1_hash)).expect("V1 hash should be accepted");
+        assert_eq!(v1_block.hash(), v1_hash);
+
+        let mut invalid_block = block.clone();
+        let error = invalid_block
+            .apply_external(&external_block(&block, Hash::ZERO))
+            .expect_err("invalid hash should be rejected");
+        assert!(error.to_string().contains("invalid external block hash"));
+    }
+
+    #[test]
+    fn genesis_uses_legacy_hash() {
+        let genesis = Block::genesis();
+        assert_eq!(genesis.hash(), BlockNumber::ZERO.hash());
+    }
+
+    #[test]
+    fn sealed_genesis_uses_legacy_hash() {
+        let pending = PendingBlock::new_at_now(BlockNumber::ZERO);
+        let genesis = Block::from_pending(pending, Hash::new([1; 32]));
+
+        assert_eq!(genesis.hash(), BlockNumber::ZERO.hash());
+        assert_eq!(genesis.header.parent_hash, Hash::ZERO);
+    }
+}
diff --git a/src/eth/primitives/block_header.rs b/src/eth/primitives/block_header.rs
index 8dd26aa93..be6f00424 100644
--- a/src/eth/primitives/block_header.rs
+++ b/src/eth/primitives/block_header.rs
@@ -68,13 +68,13 @@ impl BlockHeader {
     pub fn new(number: BlockNumber, timestamp: UnixTime) -> Self {
         Self {
             number,
-            hash: number.hash(),
+            hash: Hash::ZERO,
             transactions_root: HASH_EMPTY_TRIE,
             gas_used: Gas::ZERO,
             gas_limit: Gas::ZERO,
             bloom: LogsBloom::default(),
             timestamp,
-            parent_hash: number.prev().map(|n| n.hash()).unwrap_or(Hash::ZERO),
+            parent_hash: Hash::ZERO,
             author: Address::default(),
             extra_data: Bytes::default(),
             miner: Address::default(),
@@ -223,18 +223,15 @@ mod tests {
     use crate::eth::primitives::UnixTime;
 
     #[test]
-    fn block_header_hash_calculation() {
+    fn block_header_hash_starts_zero() {
         let header = BlockHeader::new(BlockNumber::ZERO, UnixTime::from(1234567890));
-        assert_eq!(header.hash.to_string(), "0x011b4d03dd8c01f1049143cf9c4c817e4b167f1d1b83e5c6f0f10d89ba1e7bce");
+        assert_eq!(header.hash, Hash::ZERO);
     }
 
     #[test]
-    fn block_header_parent_hash() {
+    fn block_header_parent_hash_starts_zero() {
         let header = BlockHeader::new(BlockNumber::ONE, UnixTime::from(1234567891));
-        assert_eq!(
-            header.parent_hash.to_string(),
-            "0x011b4d03dd8c01f1049143cf9c4c817e4b167f1d1b83e5c6f0f10d89ba1e7bce"
-        );
+        assert_eq!(header.parent_hash, Hash::ZERO);
     }
 
     #[test]
diff --git a/src/eth/primitives/external_block.rs b/src/eth/primitives/external_block.rs
index 94c226299..b1fef81c2 100644
--- a/src/eth/primitives/external_block.rs
+++ b/src/eth/primitives/external_block.rs
@@ -51,6 +51,11 @@ impl ExternalBlock {
         self.0.header.inner.timestamp.into()
     }
 
+    /// Returns the parent block hash.
+    pub fn parent_hash(&self) -> Hash {
+        Hash::from(self.0.header.inner.parent_hash)
+    }
+
     /// Returns the block author.
     pub fn author(&self) -> Address {
         self.0.header.inner.beneficiary.into()
diff --git a/src/eth/primitives/pending_block.rs b/src/eth/primitives/pending_block.rs
index 40f98ee63..43870d6ff 100644
--- a/src/eth/primitives/pending_block.rs
+++ b/src/eth/primitives/pending_block.rs
@@ -5,6 +5,7 @@ use crate::eth::primitives::BlockNumber;
 use crate::eth::primitives::Hash;
 use crate::eth::primitives::PendingBlockHeader;
 use crate::eth::primitives::TransactionExecution;
+use crate::eth::primitives::UnixTime;
 
 /// Block that is being mined and receiving updates.
 #[derive(DebugAsJson, Clone, Default, serde::Serialize)]
@@ -23,6 +24,17 @@ impl PendingBlock {
         }
     }
 
+    /// Creates a new pending block with an explicit timestamp.
+    pub fn new_at(number: BlockNumber, timestamp: UnixTime) -> Self {
+        Self {
+            header: PendingBlockHeader {
+                number,
+                timestamp: timestamp.into(),
+            },
+            transactions: IndexMap::new(),
+        }
+    }
+
     /// Adds a transaction execution to the block.
     pub fn push_transaction(&mut self, tx: TransactionExecution) {
         self.transactions.insert(tx.info.hash, tx);
diff --git a/src/eth/primitives/stratus_error.rs b/src/eth/primitives/stratus_error.rs
index b1bd2edf5..519376dcb 100644
--- a/src/eth/primitives/stratus_error.rs
+++ b/src/eth/primitives/stratus_error.rs
@@ -15,6 +15,7 @@ use crate::eth::primitives::Address;
 use crate::eth::primitives::BlockFilter;
 use crate::eth::primitives::BlockNumber;
 use crate::eth::primitives::Bytes;
+use crate::eth::primitives::Hash;
 use crate::eth::primitives::Nonce;
 use crate::ext::to_json_value;
 
@@ -145,6 +146,14 @@ pub enum StorageError {
     #[error("unexpected storage error: {msg}")]
     #[error_code = 9]
     Unexpected { msg: String },
+
+    #[error("parent hash conflict at block {number} between local ({local}) and external ({external}) chains.")]
+    #[error_code = 10]
+    ParentHashConflict { number: BlockNumber, local: Hash, external: Hash },
+
+    #[error("hash of block {number} is unknown.")]
+    #[error_code = 11]
+    BlockHashMissing { number: BlockNumber },
 }
 
 #[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
diff --git a/src/eth/rpc/rpc_server.rs b/src/eth/rpc/rpc_server.rs
index 2afa1724c..494c22fad 100644
--- a/src/eth/rpc/rpc_server.rs
+++ b/src/eth/rpc/rpc_server.rs
@@ -426,7 +426,7 @@ async fn stratus_health(_: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> R
 
 #[cfg(feature = "dev")]
 fn stratus_reset(_: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
-    ctx.server.storage.reset_to_genesis()?;
+    ctx.server.miner.reset_to_genesis()?;
     Ok(to_json_value(true))
 }
 
diff --git a/src/eth/storage/block_hash_ring.rs b/src/eth/storage/block_hash_ring.rs
new file mode 100644
index 000000000..fdb8a5e79
--- /dev/null
+++ b/src/eth/storage/block_hash_ring.rs
@@ -0,0 +1,135 @@
+//! Retention of recent block hashes for the EVM `BLOCKHASH` opcode.
+
+use std::num::NonZeroUsize;
+
+use parking_lot::RwLock;
+
+use crate::eth::primitives::BlockNumber;
+use crate::eth::primitives::Hash;
+use crate::eth::storage::BlockReference;
+
+/// A ring position, empty until a block whose number maps to it is published.
+type RingSlot = Option<BlockReference>;
+
+/// Fixed size ring of block hashes, indexed by `number % capacity`.
+///
+/// Block numbers are dense and published in order, so the ring holds exactly the most recent
+/// `capacity` blocks. A general purpose cache cannot promise that: it spreads entries over shards
+/// sized independently of each other, and within a shard it evicts the never read entries first,
+/// which is precisely what a freshly sealed hash is. Importer-offline depends on the retention
+/// being exact, because the permanent storage cannot answer for blocks it has not saved yet.
+///
+/// Two numbers share a slot only when they are `capacity` apart, so reading an older hash back
+/// from the permanent storage cannot displace a block that is still inside the window. Callers are
+/// expected to keep the capacity at or above the range they read back, which the default does for
+/// the 256 block `BLOCKHASH` window.
+pub struct BlockHashRing {
+    capacity: NonZeroUsize,
+    slots: RwLock<Box<[RingSlot]>>,
+}
+
+impl BlockHashRing {
+    pub fn new(capacity: NonZeroUsize) -> Self {
+        Self {
+            capacity,
+            slots: RwLock::new(vec![RingSlot::None; capacity.get()].into_boxed_slice()),
+        }
+    }
+
+    /// The hash of a block never changes, so this overwrites whatever occupied the slot.
+    pub fn insert(&self, number: BlockNumber, hash: Hash) {
+        let slot = self.slot_of(number);
+        self.slots.write()[slot] = Some(BlockReference { number, hash });
+    }
+
+    pub fn get(&self, number: BlockNumber) -> Option<Hash> {
+        let slot = self.slot_of(number);
+        match self.slots.read()[slot] {
+            Some(occupant) if occupant.number == number => Some(occupant.hash),
+            _ => None,
+        }
+    }
+
+    pub fn clear(&self) {
+        self.slots.write().fill(RingSlot::None);
+    }
+
+    fn slot_of(&self, number: BlockNumber) -> usize {
+        (number.as_u64() % self.capacity.get() as u64) as usize
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn new_ring(capacity: usize) -> BlockHashRing {
+        BlockHashRing::new(NonZeroUsize::new(capacity).unwrap())
+    }
+
+    fn block_hash(number: u64) -> Hash {
+        let mut bytes = [0_u8; 32];
+        bytes[..8].copy_from_slice(&number.to_be_bytes());
+        Hash::new(bytes)
+    }
+
+    fn publish(ring: &BlockHashRing, numbers: impl IntoIterator<Item = u64>) {
+        for number in numbers {
+            ring.insert(BlockNumber::from(number), block_hash(number));
+        }
+    }
+
+    fn read(ring: &BlockHashRing, number: u64) -> Option<Hash> {
+        ring.get(BlockNumber::from(number))
+    }
+
+    /// The whole point of the ring: a full window is retained, with no entry lost to an eviction
+    /// policy or to an unlucky shard.
+    #[test]
+    fn every_block_in_the_configured_window_is_retained() {
+        let ring = new_ring(8);
+
+        publish(&ring, 1..=8);
+
+        for number in 1..=8 {
+            assert_eq!(read(&ring, number), Some(block_hash(number)), "block {number}");
+        }
+    }
+
+    #[test]
+    fn only_blocks_that_left_the_window_are_dropped() {
+        let ring = new_ring(4);
+
+        publish(&ring, 1..=6);
+
+        assert_eq!(read(&ring, 1), None);
+        assert_eq!(read(&ring, 2), None);
+        for number in 3..=6 {
+            assert_eq!(read(&ring, number), Some(block_hash(number)), "block {number}");
+        }
+    }
+
+    /// Reading an older hash back from the permanent storage must not cost the window a block,
+    /// which is what made the sealed-but-unsaved hashes of importer-offline evictable before.
+    #[test]
+    fn reading_an_older_hash_back_does_not_displace_the_window() {
+        let ring = new_ring(4);
+        publish(&ring, 6..=8);
+
+        publish(&ring, [5]);
+
+        for number in 5..=8 {
+            assert_eq!(read(&ring, number), Some(block_hash(number)), "block {number}");
+        }
+    }
+
+    #[test]
+    fn clearing_drops_published_block_hashes() {
+        let ring = new_ring(4);
+        publish(&ring, 1..=4);
+
+        ring.clear();
+
+        assert_eq!(read(&ring, 4), None);
+    }
+}
diff --git a/src/eth/storage/cache.rs b/src/eth/storage/cache.rs
index 29d27c32f..f24601e7c 100644
--- a/src/eth/storage/cache.rs
+++ b/src/eth/storage/cache.rs
@@ -1,4 +1,4 @@
-use std::hash::Hash;
+use std::num::NonZeroUsize;
 
 use clap::Parser;
 use display_json::DebugAsJson;
@@ -11,16 +11,23 @@ use rustc_hash::FxBuildHasher;
 
 use crate::eth::primitives::Account;
 use crate::eth::primitives::Address;
+use crate::eth::primitives::BlockNumber;
 use crate::eth::primitives::ExecutionChanges;
+use crate::eth::primitives::Hash;
 use crate::eth::primitives::Slot;
 use crate::eth::primitives::SlotIndex;
 use crate::eth::primitives::SlotValue;
+use crate::eth::storage::block_hash_ring::BlockHashRing;
+
+/// Default capacity covers the complete history window reachable by `BLOCKHASH`.
+pub const DEFAULT_BLOCK_HASH_CACHE_CAPACITY: usize = 256;
 
 pub struct StorageCache {
     slot_cache: Cache<(Address, SlotIndex), SlotValue, UnitWeighter, FxBuildHasher>,
     account_cache: Cache<Address, Account, UnitWeighter, FxBuildHasher>,
     account_latest_cache: Cache<Address, Account, UnitWeighter, FxBuildHasher>,
     slot_latest_cache: Cache<(Address, SlotIndex), SlotValue, UnitWeighter, FxBuildHasher>,
+    block_hashes: BlockHashRing,
 }
 
 #[derive(DebugAsJson, Clone, Parser, serde::Serialize)]
@@ -40,6 +47,15 @@ pub struct CacheConfig {
     /// Capacity of slot history cache
     #[arg(long = "slot-history-cache-capacity", env = "SLOT_HISTORY_CACHE_CAPACITY", default_value = "100000")]
     pub slot_history_cache_capacity: usize,
+
+    /// Number of the most recent block hashes kept in memory.
+    ///
+    /// Zero is raised to one, which keeps nothing worth having: the only block a single slot can
+    /// hold is the sealed tip, and `BLOCKHASH` already reads that one from temporary storage.
+    ///
+    /// The offline importer raises this value to cover its sealed-but-unsaved backlog.
+    #[arg(long = "block-hash-cache-capacity", env = "BLOCK_HASH_CACHE_CAPACITY", default_value_t = DEFAULT_BLOCK_HASH_CACHE_CAPACITY)]
+    pub block_hash_cache_capacity: usize,
 }
 
 impl CacheConfig {
@@ -79,6 +95,7 @@ impl StorageCache {
                 FxBuildHasher,
                 DefaultLifecycle::default(),
             ),
+            block_hashes: BlockHashRing::new(NonZeroUsize::new(config.block_hash_cache_capacity).unwrap_or(NonZeroUsize::MIN)),
         }
     }
 
@@ -87,6 +104,7 @@ impl StorageCache {
         self.account_cache.clear();
         self.account_latest_cache.clear();
         self.slot_latest_cache.clear();
+        self.block_hashes.clear();
     }
 
     pub fn cache_slot_if_missing(&self, address: Address, slot: Slot) {
@@ -145,6 +163,14 @@ impl StorageCache {
     pub fn get_slot_latest(&self, address: Address, index: SlotIndex) -> Option<Slot> {
         self.slot_latest_cache.get(&(address, index)).map(|value| Slot { value, index })
     }
+
+    pub fn cache_block_hash(&self, number: BlockNumber, hash: Hash) {
+        self.block_hashes.insert(number, hash);
+    }
+
+    pub fn get_block_hash(&self, number: BlockNumber) -> Option<Hash> {
+        self.block_hashes.get(number)
+    }
 }
 
 trait CacheExt<Key, Val> {
@@ -153,7 +179,7 @@ trait CacheExt<Key, Val> {
 
 impl<Key, Val, We, B, L> CacheExt<Key, Val> for Cache<Key, Val, We, B, L>
 where
-    Key: Hash + Equivalent<Key> + ToOwned<Owned = Key> + std::cmp::Eq,
+    Key: std::hash::Hash + Equivalent<Key> + ToOwned<Owned = Key> + std::cmp::Eq,
     Val: Clone,
     We: quick_cache::Weighter<Key, Val> + Clone,
     B: std::hash::BuildHasher + Clone,
@@ -170,3 +196,67 @@ where
         }
     }
 }
+
+/// Retention itself is covered by the block-hash ring. These only check that the configuration
+/// reaches it and that clearing the cache reaches it too.
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn cache_holding_block_hashes(capacity: usize) -> StorageCache {
+        CacheConfig {
+            slot_cache_capacity: 1,
+            account_cache_capacity: 1,
+            account_history_cache_capacity: 1,
+            slot_history_cache_capacity: 1,
+            block_hash_cache_capacity: capacity,
+        }
+        .init()
+    }
+
+    fn publish(cache: &StorageCache, number: u64) -> Hash {
+        let hash = Hash::new([number as u8; 32]);
+        cache.cache_block_hash(BlockNumber::from(number), hash);
+        hash
+    }
+
+    fn read(cache: &StorageCache, number: u64) -> Option<Hash> {
+        cache.get_block_hash(BlockNumber::from(number))
+    }
+
+    #[test]
+    fn block_hashes_are_retained_up_to_the_configured_capacity() {
+        let cache = cache_holding_block_hashes(2);
+
+        publish(&cache, 1);
+        let second = publish(&cache, 2);
+        let third = publish(&cache, 3);
+
+        assert_eq!(read(&cache, 1), None);
+        assert_eq!(read(&cache, 2), Some(second));
+        assert_eq!(read(&cache, 3), Some(third));
+    }
+
+    #[test]
+    fn clearing_the_cache_drops_published_block_hashes() {
+        let cache = cache_holding_block_hashes(2);
+        publish(&cache, 1);
+
+        cache.clear();
+
+        assert_eq!(read(&cache, 1), None);
+    }
+
+    /// A single slot only ever answers for the sealed tip, which the temporary storage already
+    /// serves, so raising zero to one costs an operator asking for no cache nothing but 48 bytes.
+    #[test]
+    fn a_zero_capacity_is_raised_to_a_single_slot() {
+        let cache = cache_holding_block_hashes(0);
+
+        publish(&cache, 1);
+        let second = publish(&cache, 2);
+
+        assert_eq!(read(&cache, 1), None);
+        assert_eq!(read(&cache, 2), Some(second));
+    }
+}
diff --git a/src/eth/storage/mod.rs b/src/eth/storage/mod.rs
index 980980230..fe481ff93 100644
--- a/src/eth/storage/mod.rs
+++ b/src/eth/storage/mod.rs
@@ -8,8 +8,10 @@ pub use permanent::RocksPermanentStorage;
 pub use stratus_storage::MinedPointInTime;
 pub use stratus_storage::StratusStorage;
 pub use temporary::InMemoryTemporaryStorage;
+pub use temporary::PendingBlockGuard;
 pub use temporary::TemporaryStorageConfig;
 
+mod block_hash_ring;
 mod cache;
 pub mod permanent;
 mod resolve_pending;
@@ -22,10 +24,33 @@ use clap::Parser;
 use display_json::DebugAsJson;
 pub use temporary::compute_pending_block_number;
 
+use crate::eth::primitives::Block;
 use crate::eth::primitives::BlockNumber;
+use crate::eth::primitives::Hash;
 use crate::eth::primitives::Index;
 use crate::eth::primitives::StratusError;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct BlockReference {
+    pub number: BlockNumber,
+    pub hash: Hash,
+}
+
+impl BlockReference {
+    pub fn genesis() -> Self {
+        Self::from(&Block::genesis())
+    }
+}
+
+impl From<&Block> for BlockReference {
+    fn from(block: &Block) -> Self {
+        Self {
+            number: block.number(),
+            hash: block.hash(),
+        }
+    }
+}
+
 // -----------------------------------------------------------------------------
 // Config
 // -----------------------------------------------------------------------------
diff --git a/src/eth/storage/permanent/rocks/rocks_permanent.rs b/src/eth/storage/permanent/rocks/rocks_permanent.rs
index 2894dcf54..d1cfba718 100644
--- a/src/eth/storage/permanent/rocks/rocks_permanent.rs
+++ b/src/eth/storage/permanent/rocks/rocks_permanent.rs
@@ -29,6 +29,7 @@ use crate::eth::primitives::StorageError;
 use crate::eth::primitives::TransactionMined;
 #[cfg(feature = "dev")]
 use crate::eth::primitives::Wei;
+use crate::eth::storage::BlockReference;
 use crate::eth::storage::MinedPointInTime;
 use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
 use crate::ext::SleepReason;
@@ -150,6 +151,10 @@ impl RocksPermanentStorage {
         Ok(genesis.is_some())
     }
 
+    pub(crate) fn read_chain_tip(&self) -> Result<Option<BlockReference>, StorageError> {
+        Ok(self.read_block(BlockFilter::Latest)?.as_ref().map(BlockReference::from))
+    }
+
     // -------------------------------------------------------------------------
     // State operations
     // -------------------------------------------------------------------------
diff --git a/src/eth/storage/stratus_storage.rs b/src/eth/storage/stratus_storage.rs
index c30863392..ad603ab31 100644
--- a/src/eth/storage/stratus_storage.rs
+++ b/src/eth/storage/stratus_storage.rs
@@ -28,11 +28,14 @@ use crate::eth::primitives::StorageError;
 use crate::eth::primitives::TransactionExecution;
 use crate::eth::primitives::TransactionStage;
 use crate::eth::primitives::UnixTime;
+use crate::eth::primitives::UnixTimeNow;
 #[cfg(feature = "dev")]
 use crate::eth::primitives::Wei;
 #[cfg(feature = "dev")]
 use crate::eth::primitives::test_accounts;
+use crate::eth::storage::BlockReference;
 use crate::eth::storage::InMemoryTemporaryStorage;
+use crate::eth::storage::PendingBlockGuard;
 use crate::eth::storage::ReadKind;
 use crate::eth::storage::RocksPermanentStorage;
 use crate::eth::storage::StorageCache;
@@ -40,7 +43,6 @@ use crate::eth::storage::TxCount;
 use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
 use crate::eth::storage::permanent::rocks::types::BlockRocksdb;
 use crate::eth::storage::resolve_pending;
-use crate::ext::not;
 use crate::infra::metrics;
 use crate::infra::metrics::timed;
 use crate::infra::tracing::SpanExt;
@@ -58,6 +60,7 @@ pub struct StratusStorage {
     temp: InMemoryTemporaryStorage,
     cache: StorageCache,
     pub perm: RocksPermanentStorage,
+    last_saved: parking_lot::Mutex<Option<BlockReference>>,
     // CONTRACT: Always acquire a lock when reading slots or accounts from latest (cache OR perm) and when saving a block
     pub(super) transient_state_lock: parking_lot::RwLock<()>,
     #[cfg(feature = "dev")]
@@ -240,6 +243,35 @@ impl EntityRead for Slot {
 }
 
 impl StratusStorage {
+    fn validate_saved_continuity(last_saved: Option<BlockReference>, block: &Block) -> Result<(), StorageError> {
+        let block_number = block.number();
+        let (expected_number, expected_parent_hash) = match last_saved {
+            None => (BlockNumber::ZERO, Hash::ZERO),
+            Some(parent) => (parent.number.next_block_number(), parent.hash),
+        };
+
+        if block_number != expected_number {
+            return Err(StorageError::MinedNumberConflict {
+                new: block_number,
+                mined: expected_number.prev().unwrap_or_default(),
+            });
+        }
+
+        if block.header.parent_hash != expected_parent_hash {
+            return Err(StorageError::ParentHashConflict {
+                number: block_number,
+                local: expected_parent_hash,
+                external: block.header.parent_hash,
+            });
+        }
+
+        Ok(())
+    }
+
+    pub fn validate_next_saved_block(&self, block: &Block) -> Result<(), StorageError> {
+        Self::validate_saved_continuity(*self.last_saved.lock(), block)
+    }
+
     /// Creates a new storage with the specified temporary and permanent implementations.
     pub fn new(
         temp: InMemoryTemporaryStorage,
@@ -247,10 +279,13 @@ impl StratusStorage {
         cache: StorageCache,
         #[cfg(feature = "dev")] perm_config: crate::eth::storage::permanent::PermanentStorageConfig,
     ) -> Result<Self, StorageError> {
+        let last_saved = perm.read_chain_tip()?;
+
         let this = Self {
             temp,
             cache,
             perm,
+            last_saved: parking_lot::Mutex::new(last_saved),
             transient_state_lock: parking_lot::RwLock::new(()),
             #[cfg(feature = "dev")]
             perm_config,
@@ -259,7 +294,7 @@ impl StratusStorage {
         // create genesis block and accounts if necessary
         #[cfg(feature = "dev")]
         if !this.has_genesis()? {
-            this.reset_to_genesis()?;
+            this.reset_to_genesis_inner()?;
         }
 
         Ok(this)
@@ -282,7 +317,7 @@ impl StratusStorage {
 
         use crate::eth::storage::cache::CacheConfig;
 
-        let temp = InMemoryTemporaryStorage::new(0.into());
+        let temp = InMemoryTemporaryStorage::new(Some(BlockReference::genesis()));
 
         // Create a temporary directory for RocksDB
         let rocks_dir = tempdir().expect("Failed to create temporary directory for tests");
@@ -303,6 +338,7 @@ impl StratusStorage {
             account_cache_capacity: 20000,
             account_history_cache_capacity: 20000,
             slot_history_cache_capacity: 100000,
+            block_hash_cache_capacity: super::cache::DEFAULT_BLOCK_HASH_CACHE_CAPACITY,
         }
         .init();
 
@@ -353,14 +389,59 @@ impl StratusStorage {
         })
     }
 
-    pub fn set_pending_from_external(&self, block: &ExternalBlock) {
-        self.temp.set_pending_header(block.number(), block.timestamp());
+    /// Prepares the guarded pending state to receive an external block.
+    pub fn set_pending_from_external(&self, guard: &PendingBlockGuard<'_>, block: &ExternalBlock) {
+        self.set_pending_header(guard, block.number(), block.timestamp());
     }
 
-    pub fn set_pending_header(&self, number: BlockNumber, timestamp: UnixTime) {
+    pub fn pending_block_guard(&self) -> PendingBlockGuard<'_> {
+        self.temp.pending_block_guard()
+    }
+
+    pub fn set_pending_header(&self, _guard: &PendingBlockGuard<'_>, number: BlockNumber, timestamp: UnixTime) {
         self.temp.set_pending_header(number, timestamp);
     }
 
+    pub fn read_pending_parent_hash(&self, _guard: &PendingBlockGuard<'_>) -> Hash {
+        self.temp.read_latest_sealed().hash
+    }
+
+    /// Publishes the identity of a block that was just sealed.
+    ///
+    /// This must happen at seal time rather than at save time: mining and saving can run in separate
+    /// threads, so the next block may be sealed while this one is still on its way to the permanent
+    /// storage, and it would find no parent to chain to.
+    pub fn publish_block_hash(&self, number: BlockNumber, hash: Hash) {
+        self.cache.cache_block_hash(number, hash);
+    }
+
+    /// Reads the hash of a mined block from the latest sealed block, the cache or the permanent storage.
+    ///
+    /// The latest sealed block answers for its own number whether or not it already reached the
+    /// permanent storage, so the saved tip is never consulted here and this stays off the block
+    /// saving critical section. Cache misses are expected for blocks mined before this process
+    /// started, since sealing a block is what publishes its hash.
+    pub fn read_block_hash(&self, number: BlockNumber) -> Result<Option<Hash>, StorageError> {
+        let latest = self.temp.read_latest_sealed();
+        if latest.number == number {
+            tracing::debug!(storage = %label::TEMP, %number, "block hash found in temporary storage");
+            return Ok(Some(latest.hash));
+        }
+
+        if let Some(hash) = self.cache.get_block_hash(number) {
+            tracing::debug!(storage = %label::CACHE, %number, "block hash found in cache");
+            return Ok(Some(hash));
+        }
+
+        let Some(block) = self.read_block(BlockFilter::Number(number))? else {
+            return Ok(None);
+        };
+
+        let hash = block.hash();
+        self.cache.cache_block_hash(number, hash);
+        Ok(Some(hash))
+    }
+
     pub fn set_mined_block_number(&self, block_number: BlockNumber) {
         #[cfg(feature = "tracing")]
         let _span = tracing::info_span!("storage::set_mined_block_number", %block_number).entered();
@@ -453,7 +534,7 @@ impl StratusStorage {
     // Blocks
     // -------------------------------------------------------------------------
 
-    pub fn save_execution(&self, tx: TransactionExecution) -> Result<(), StorageError> {
+    pub fn save_execution(&self, _guard: &PendingBlockGuard<'_>, tx: TransactionExecution) -> Result<(), StorageError> {
         let changes = tx.result.execution.changes.clone();
 
         #[cfg(feature = "tracing")]
@@ -488,27 +569,27 @@ impl StratusStorage {
         self.temp.read_pending_executions()
     }
 
-    pub fn finish_pending_block(&self) -> Result<(PendingBlock, ExecutionChanges), StorageError> {
+    pub fn pending_block_to_seal(&self, _guard: &PendingBlockGuard<'_>) -> (PendingBlock, ExecutionChanges) {
+        self.temp.pending_block_to_seal()
+    }
+
+    pub(crate) fn finish_pending_block(&self, _guard: &PendingBlockGuard<'_>, block: BlockReference, timestamp: UnixTimeNow) {
         #[cfg(feature = "tracing")]
-        let _span = tracing::info_span!("storage::finish_pending_block", block_number = tracing::field::Empty).entered();
-        tracing::debug!(storage = %label::TEMP, "finishing pending block");
+        let _span = tracing::info_span!("storage::finish_pending_block", block_number = %block.number).entered();
+        tracing::debug!(storage = %label::TEMP, block_number = %block.number, "finishing pending block");
 
-        let result = timed(|| self.temp.finish_pending_block()).with(|m| {
-            metrics::inc_storage_finish_pending_block(m.elapsed, label::TEMP, m.result.is_ok());
-            if let Err(ref e) = m.result {
-                tracing::error!(reason = ?e, "failed to finish pending block");
-            }
+        timed(|| self.temp.finish_pending_block(block, timestamp)).with(|m| {
+            metrics::inc_storage_finish_pending_block(m.elapsed, label::TEMP, true);
         });
 
-        if let Ok((ref block, _)) = result {
-            Span::with(|s| s.rec_str("block_number", &block.header.number));
-        }
-
-        result
+        Span::with(|s| s.rec_str("block_number", &block.number));
     }
 
     pub fn save_genesis_block(&self, block: Block, accounts: Vec<Account>, changes: ExecutionChanges) -> Result<(), StorageError> {
         let block_number = block.number();
+        let block_reference = BlockReference::from(&block);
+        let mut last_saved = self.last_saved.lock();
+        Self::validate_saved_continuity(*last_saved, &block)?;
 
         #[cfg(feature = "tracing")]
         let _span = tracing::info_span!("storage::save_genesis_block", block_number = %block_number).entered();
@@ -520,25 +601,23 @@ impl StratusStorage {
             if let Err(ref e) = m.result {
                 tracing::error!(reason = ?e, "failed to save genesis block");
             }
-        })
+        })?;
+
+        *last_saved = Some(block_reference);
+        self.set_mined_block_number(block_number);
+        Ok(())
     }
 
     pub fn save_block(&self, block: Block, changes: ExecutionChanges) -> Result<(), StorageError> {
         let block_number = block.number();
+        let block_reference = BlockReference::from(&block);
+        let mut last_saved = self.last_saved.lock();
 
         #[cfg(feature = "tracing")]
         let _span = tracing::info_span!("storage::save_block", block_number = %block.number()).entered();
         tracing::debug!(storage = %label::PERM, block_number = %block_number, transactions_len = %block.transactions.len(), ?changes, "saving block");
 
-        // check mined number
-        let mined_number = self.read_mined_block_number();
-        if not(block_number.is_zero()) && block_number != mined_number.next_block_number() {
-            tracing::error!(%block_number, %mined_number, "failed to save block because mismatch with mined block number");
-            return Err(StorageError::MinedNumberConflict {
-                new: block_number,
-                mined: mined_number,
-            });
-        }
+        Self::validate_saved_continuity(*last_saved, &block)?;
 
         // check pending number
         let pending_header = self.read_pending_block_header();
@@ -550,13 +629,6 @@ impl StratusStorage {
             });
         }
 
-        // check mined block
-        let existing_block = self.read_block(BlockFilter::Number(block_number))?;
-        if existing_block.is_some() {
-            tracing::error!(%block_number, %mined_number, "failed to save block because block with the same number already exists in the permanent storage");
-            return Err(StorageError::BlockConflict { number: block_number });
-        }
-
         let tens_of_millions_gas_used = block.header.gas_used.as_u64() / 10_000_000;
 
         timed(|| {
@@ -573,6 +645,7 @@ impl StratusStorage {
             }
         })?;
 
+        *last_saved = Some(block_reference);
         self.set_mined_block_number(block_number);
 
         Ok(())
@@ -705,11 +778,16 @@ impl StratusStorage {
     // General state
     // -------------------------------------------------------------------------
 
+    #[cfg(feature = "dev")]
+    pub(crate) fn reset_to_genesis(&self, _guard: &PendingBlockGuard<'_>) -> Result<(), StorageError> {
+        self.reset_to_genesis_inner()
+    }
+
     #[cfg(feature = "dev")]
     /// Resets the storage to the genesis state.
     /// If a genesis.json file is available, it will be used.
     /// Otherwise, it will use the default genesis configuration.
-    pub fn reset_to_genesis(&self) -> Result<(), StorageError> {
+    fn reset_to_genesis_inner(&self) -> Result<(), StorageError> {
         tracing::info!("resetting storage to genesis state");
 
         self.cache.clear();
@@ -725,6 +803,7 @@ impl StratusStorage {
                 tracing::error!(reason = ?e, "failed to reset permanent storage");
             }
         })?;
+        *self.last_saved.lock() = None;
 
         // reset temp
         tracing::debug!(storage = %label::TEMP, "reseting temporary storage");
@@ -762,6 +841,8 @@ impl StratusStorage {
             tracing::info!("using default genesis block");
             Block::genesis()
         };
+        let genesis_hash = genesis_block.hash();
+        self.publish_block_hash(BlockNumber::ZERO, genesis_hash);
         // Try to load genesis.json from the path specified in GenesisFileConfig
         // or use default genesis configuration
         let (genesis_accounts, genesis_slots) = if let Some(genesis_path) = &self.perm_config.genesis_file.genesis_path {
@@ -848,9 +929,13 @@ impl StratusStorage {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
     use super::*;
     use crate::eth::executor::EvmExecutionResult;
     use crate::eth::executor::EvmInput;
+    use crate::eth::miner::Miner;
+    use crate::eth::miner::MinerMode;
     use crate::eth::primitives::ExecutionAccountChanges;
     use crate::eth::primitives::ExecutionInfo;
     use crate::eth::primitives::ExecutionResult;
@@ -860,8 +945,22 @@ mod tests {
     use crate::eth::primitives::TransactionInput;
     use crate::eth::primitives::Wei;
 
+    fn initialize_genesis(storage: &Arc<StratusStorage>) -> Block {
+        if let Some(genesis) = storage.read_block(BlockFilter::Number(BlockNumber::ZERO)).unwrap() {
+            return genesis;
+        }
+
+        let genesis = Block::genesis();
+        storage
+            .save_genesis_block(genesis.clone(), Vec::new(), ExecutionChanges::default())
+            .expect("save genesis block");
+        genesis
+    }
+
     /// Mines a block applying `changes`
-    fn mine_block(storage: &StratusStorage, changes: ExecutionChanges) -> BlockNumber {
+    fn mine_block(storage: &Arc<StratusStorage>, changes: ExecutionChanges) -> BlockNumber {
+        initialize_genesis(storage);
+        let miner = Miner::new(Arc::clone(storage), MinerMode::Automine);
         let (header, _) = storage.read_pending_block_header();
         let evm_input = EvmInput::from_eth_transaction(&TransactionInput::default(), header.number, *header.timestamp);
 
@@ -870,19 +969,211 @@ mod tests {
         result.execution.changes = changes;
 
         let tx = TransactionExecution::new(TransactionInfo::default(), Signature::default(), ExecutionInfo::default(), evm_input, result);
-        storage.save_execution(tx).expect("save execution");
-
-        let (block, block_changes) = storage.finish_pending_block().expect("finish pending block");
-        storage.save_block(block.into(), block_changes).expect("save block");
+        let session = miner.pending_session();
+        session.append_execution(tx).expect("save execution");
+        let (block, block_changes) = session.seal_local();
+        storage.save_block(block, block_changes).expect("save block");
 
         storage.read_mined_block_number()
     }
 
+    /// Mining and saving can run in separate threads, so a block must be chainable as soon as it is
+    /// sealed, before it reaches the permanent storage.
+    #[test]
+    fn block_hash_is_readable_before_the_block_is_saved() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+
+        let number = BlockNumber::from(7_u64);
+        let hash = Hash::new([7; 32]);
+        storage.publish_block_hash(number, hash);
+
+        assert_eq!(storage.read_block_hash(number).expect("read block hash"), Some(hash));
+        assert!(storage.read_block(BlockFilter::Number(number)).expect("read block").is_none());
+    }
+
+    #[test]
+    fn latest_unsaved_hash_survives_cache_clear() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::External);
+        initialize_genesis(&storage);
+        let (block, _) = miner.mine_local().expect("seal block");
+        storage.clear_cache();
+
+        assert_eq!(storage.read_block_hash(block.number()).expect("read unsaved hash"), Some(block.hash()));
+        assert!(storage.read_block(BlockFilter::Number(block.number())).unwrap().is_none());
+    }
+
+    #[test]
+    fn block_hash_falls_back_to_permanent_storage_when_the_cache_is_cold() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+
+        // the second block keeps the first one behind the sealed tip, so it cannot be answered from temporary storage
+        let number = mine_block(&storage, ExecutionChanges::default());
+        mine_block(&storage, ExecutionChanges::default());
+        let hash = storage
+            .read_block(BlockFilter::Number(number))
+            .expect("read block")
+            .expect("mined block should exist")
+            .hash();
+
+        // simulates a restart, where nothing was published by this process
+        storage.cache.clear();
+
+        assert_eq!(storage.read_block_hash(number).expect("read block hash"), Some(hash));
+    }
+
+    /// The sealed tip keeps answering for its own number after it is saved, so the block saving
+    /// critical section never has to be consulted to resolve it.
+    #[test]
+    fn latest_saved_hash_is_served_from_temporary_storage() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+
+        let number = mine_block(&storage, ExecutionChanges::default());
+        let hash = storage
+            .read_block(BlockFilter::Number(number))
+            .expect("read block")
+            .expect("mined block should exist")
+            .hash();
+
+        storage.cache.clear();
+
+        assert_eq!(storage.temp.read_latest_sealed(), BlockReference { number, hash });
+        assert_eq!(storage.read_block_hash(number).expect("read block hash"), Some(hash));
+    }
+
+    #[test]
+    fn mined_blocks_are_chained_to_their_parent() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+
+        let first = mine_block(&storage, ExecutionChanges::default());
+        let second = mine_block(&storage, ExecutionChanges::default());
+        assert_ne!(first, second);
+
+        let read = |number| {
+            storage
+                .read_block(BlockFilter::Number(number))
+                .expect("read block")
+                .expect("block should exist")
+        };
+
+        assert_eq!(read(second).header.parent_hash, read(first).hash());
+    }
+
+    #[test]
+    fn save_block_rejects_wrong_parent_without_advancing_last_saved() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::Automine);
+        let genesis = initialize_genesis(&storage);
+        let (block, changes) = miner.mine_local().expect("seal block");
+
+        let mut invalid = block.clone();
+        invalid.header.parent_hash = Hash::ZERO;
+        invalid.apply_default_hash();
+        let error = storage.save_block(invalid, changes.clone()).expect_err("wrong parent should be rejected");
+        assert!(matches!(
+            error,
+            StorageError::ParentHashConflict { number, local, external }
+                if number == BlockNumber::ONE && local == genesis.hash() && external == Hash::ZERO
+        ));
+
+        storage.save_block(block, changes).expect("valid block should still save");
+        assert_eq!(
+            *storage.last_saved.lock(),
+            Some(BlockReference {
+                number: BlockNumber::ONE,
+                hash: storage.read_block(BlockFilter::Number(BlockNumber::ONE)).unwrap().unwrap().hash(),
+            })
+        );
+    }
+
+    #[test]
+    fn sealed_tip_can_run_ahead_of_saved_tip() {
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::External);
+        initialize_genesis(&storage);
+
+        let (first, first_changes) = miner.mine_local().expect("seal first block");
+        let (second, second_changes) = miner.mine_local().expect("seal second block");
+
+        assert_eq!(second.header.parent_hash, first.hash());
+        assert_eq!(storage.temp.read_latest_sealed(), BlockReference::from(&second));
+        assert_eq!(
+            *storage.last_saved.lock(),
+            Some(BlockReference {
+                number: BlockNumber::ZERO,
+                hash: Block::genesis().hash(),
+            })
+        );
+
+        storage.save_block(first, first_changes).expect("save first block");
+        storage.save_block(second, second_changes).expect("save second block");
+    }
+
+    #[test]
+    fn startup_preloads_legacy_saved_tip_for_next_parent() {
+        use crate::eth::storage::cache::CacheConfig;
+
+        let rocks_dir = tempfile::tempdir().expect("create rocks directory");
+        let rocks_prefix = rocks_dir.path().join("preloaded-tip").to_string_lossy().into_owned();
+        let perm = RocksPermanentStorage::new(
+            Some(rocks_prefix.clone()),
+            std::time::Duration::from_secs(240),
+            super::super::permanent::RocksCfCacheConfig::default(),
+            true,
+            None,
+            1024,
+        )
+        .expect("create permanent storage");
+
+        let genesis = Block::genesis();
+        perm.save_genesis_block(genesis.clone(), Vec::new(), ExecutionChanges::default())
+            .expect("save genesis");
+        let mut legacy = Block::new(BlockNumber::ONE, UnixTime::from(1_u64));
+        legacy.header.parent_hash = genesis.hash();
+        legacy.apply_hash(legacy.calculate_hash_v1());
+        perm.save_block(legacy.clone(), ExecutionChanges::default()).expect("save legacy block");
+        perm.set_mined_block_number(BlockNumber::ONE);
+
+        let temp = InMemoryTemporaryStorage::new(Some(BlockReference::from(&legacy)));
+        let cache = CacheConfig {
+            slot_cache_capacity: 1,
+            account_cache_capacity: 1,
+            account_history_cache_capacity: 1,
+            slot_history_cache_capacity: 1,
+            block_hash_cache_capacity: 256,
+        }
+        .init();
+        let storage = Arc::new(
+            StratusStorage::new(
+                temp,
+                perm,
+                cache,
+                #[cfg(feature = "dev")]
+                super::super::permanent::PermanentStorageConfig {
+                    rocks_path_prefix: Some(rocks_prefix),
+                    rocks_shutdown_timeout: std::time::Duration::from_secs(240),
+                    rocks_cf_cache: super::super::permanent::RocksCfCacheConfig::default(),
+                    rocks_disable_sync_write: false,
+                    rocks_cf_size_metrics_interval: None,
+                    genesis_file: crate::config::GenesisFileConfig::default(),
+                    rocks_file_descriptors_limit: 1024,
+                },
+            )
+            .expect("create storage"),
+        );
+        let miner = Miner::new(Arc::clone(&storage), MinerMode::Automine);
+
+        let (block, _) = miner.mine_local().expect("seal next block");
+
+        assert_eq!(block.number(), BlockNumber::from(2_u64));
+        assert_eq!(block.header.parent_hash, legacy.hash());
+    }
+
     /// An `eth_call` pinned to a block that is no longer the latest must read the historical
     /// state at its captured block, not the current latest state.
     #[test]
     fn read_slot_for_call_pinned_to_older_block_must_not_read_latest_state() {
-        let storage = StratusStorage::new_test().expect("failed to build test storage");
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
 
         let address = Address::new([0xAA; 20]);
         let index = SlotIndex::ZERO;
@@ -909,7 +1200,7 @@ mod tests {
 
     #[test]
     fn read_account_for_call_pinned_to_older_block_must_not_read_latest_state() {
-        let storage = StratusStorage::new_test().expect("failed to build test storage");
+        let storage = Arc::new(StratusStorage::new_test().expect("failed to build test storage"));
 
         let address = Address::new([0xBB; 20]);
 
diff --git a/src/eth/storage/temporary/inmemory/mod.rs b/src/eth/storage/temporary/inmemory/mod.rs
index ffd34f221..6bf3f0a9b 100644
--- a/src/eth/storage/temporary/inmemory/mod.rs
+++ b/src/eth/storage/temporary/inmemory/mod.rs
@@ -1,5 +1,6 @@
 //! In-memory storage implementations.
 
+pub use self::transaction::PendingBlockGuard;
 use crate::eth::primitives::Account;
 use crate::eth::primitives::Address;
 use crate::eth::primitives::BlockNumber;
@@ -16,8 +17,10 @@ use crate::eth::primitives::SlotIndex;
 use crate::eth::primitives::StorageError;
 use crate::eth::primitives::TransactionExecution;
 use crate::eth::primitives::UnixTime;
+use crate::eth::primitives::UnixTimeNow;
 #[cfg(feature = "dev")]
 use crate::eth::primitives::Wei;
+use crate::eth::storage::BlockReference;
 use crate::eth::storage::ReadKind;
 use crate::eth::storage::TxCount;
 use crate::eth::storage::temporary::inmemory::call::InMemoryCallTemporaryStorage;
@@ -28,14 +31,14 @@ mod transaction;
 
 #[derive(Debug)]
 pub struct InMemoryTemporaryStorage {
-    pub transaction_storage: InmemoryTransactionTemporaryStorage,
+    transaction_storage: InmemoryTransactionTemporaryStorage,
     pub call_storage: InMemoryCallTemporaryStorage,
 }
 
 impl InMemoryTemporaryStorage {
-    pub fn new(block_number: BlockNumber) -> Self {
+    pub(crate) fn new(saved_tip: Option<BlockReference>) -> Self {
         Self {
-            transaction_storage: InmemoryTransactionTemporaryStorage::new(block_number),
+            transaction_storage: InmemoryTransactionTemporaryStorage::new(saved_tip),
             call_storage: InMemoryCallTemporaryStorage::new(),
         }
     }
@@ -44,6 +47,14 @@ impl InMemoryTemporaryStorage {
         self.transaction_storage.read_pending_block_header()
     }
 
+    pub(crate) fn read_latest_sealed(&self) -> BlockReference {
+        self.transaction_storage.read_latest_sealed()
+    }
+
+    pub fn pending_block_guard(&self) -> PendingBlockGuard<'_> {
+        self.transaction_storage.pending_block_guard()
+    }
+
     #[cfg(feature = "dev")]
     pub fn set_pending_block_header(&self, block_number: BlockNumber) -> anyhow::Result<(), StorageError> {
         self.transaction_storage.set_pending_block_header(block_number)
@@ -62,9 +73,13 @@ impl InMemoryTemporaryStorage {
         self.transaction_storage.read_pending_executions()
     }
 
-    pub fn finish_pending_block(&self) -> anyhow::Result<(PendingBlock, ExecutionChanges), StorageError> {
+    pub fn pending_block_to_seal(&self) -> (PendingBlock, ExecutionChanges) {
+        self.transaction_storage.pending_block_to_seal()
+    }
+
+    pub(crate) fn finish_pending_block(&self, block: BlockReference, timestamp: UnixTimeNow) {
+        self.transaction_storage.finish_pending_block(block, timestamp);
         self.call_storage.retain_recent_blocks();
-        self.transaction_storage.finish_pending_block()
     }
 
     pub fn read_pending_execution(&self, hash: Hash) -> anyhow::Result<Option<TransactionExecution>, StorageError> {
@@ -132,8 +147,37 @@ impl InMemoryTemporaryStorageState {
         }
     }
 
+    /// Creates state for a block that is already sealed without advancing the development clock.
+    pub fn new_sealed(block_number: BlockNumber) -> Self {
+        Self {
+            block: PendingBlock::new_at(block_number, UnixTime::ZERO),
+            block_changes: ExecutionChanges::default(),
+        }
+    }
+
     pub fn reset(&mut self) {
         self.block = PendingBlock::new_at_now(1.into());
         self.block_changes = ExecutionChanges::default();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_chain_starts_with_genesis_pending() {
+        let storage = InMemoryTemporaryStorage::new(None);
+
+        assert_eq!(storage.read_pending_block_header().0.number, BlockNumber::ZERO);
+    }
+
+    #[test]
+    fn saved_tip_starts_with_its_successor_pending() {
+        let genesis = BlockReference::genesis();
+        let storage = InMemoryTemporaryStorage::new(Some(genesis));
+
+        assert_eq!(storage.read_latest_sealed(), genesis);
+        assert_eq!(storage.read_pending_block_header().0.number, BlockNumber::ONE);
+    }
+}
diff --git a/src/eth/storage/temporary/inmemory/transaction.rs b/src/eth/storage/temporary/inmemory/transaction.rs
index b9fedd2a3..a14bdadf8 100644
--- a/src/eth/storage/temporary/inmemory/transaction.rs
+++ b/src/eth/storage/temporary/inmemory/transaction.rs
@@ -1,9 +1,9 @@
 //! In-memory storage implementations.
 
+use parking_lot::Mutex;
+use parking_lot::MutexGuard;
 use parking_lot::RwLock;
 use parking_lot::RwLockUpgradableReadGuard;
-#[cfg(not(feature = "dev"))]
-use parking_lot::RwLockWriteGuard;
 
 use crate::eth::executor::EvmInput;
 use crate::eth::primitives::Account;
@@ -23,34 +23,75 @@ use crate::eth::primitives::StorageError;
 use crate::eth::primitives::TransactionExecution;
 use crate::eth::primitives::TransactionInput;
 use crate::eth::primitives::UnixTime;
-#[cfg(feature = "dev")]
 use crate::eth::primitives::UnixTimeNow;
 #[cfg(feature = "dev")]
 use crate::eth::primitives::Wei;
+use crate::eth::storage::BlockReference;
 use crate::eth::storage::TxCount;
 use crate::eth::storage::temporary::inmemory::InMemoryTemporaryStorageState;
 
+#[derive(Debug, Clone)]
+pub struct InMemorySealedBlock {
+    pub state: InMemoryTemporaryStorageState,
+    pub hash: Hash,
+}
+
+#[derive(Debug)]
+pub struct InMemoryChainState {
+    pub pending_block: InMemoryTemporaryStorageState,
+    pub latest_sealed: InMemorySealedBlock,
+}
+
+pub struct PendingBlockGuard<'a> {
+    _guard: MutexGuard<'a, ()>,
+}
+
 #[derive(Debug)]
 pub struct InmemoryTransactionTemporaryStorage {
-    pub pending_block: RwLock<InMemoryTemporaryStorageState>,
-    pub latest_block: RwLock<Option<InMemoryTemporaryStorageState>>,
+    pending_session: Mutex<()>,
+    pub state: RwLock<InMemoryChainState>,
 }
 
 impl InmemoryTransactionTemporaryStorage {
-    pub fn new(block_number: BlockNumber) -> Self {
+    pub fn new(saved_tip: Option<BlockReference>) -> Self {
+        let pending_block_number = saved_tip.map_or(BlockNumber::ZERO, |tip| tip.number.next_block_number());
+        // Keep the canonical genesis reference available for flows that persist genesis directly,
+        // but do not advance the pending block until genesis actually exists in permanent storage.
+        let latest_sealed = saved_tip.unwrap_or_else(BlockReference::genesis);
+
         Self {
-            pending_block: RwLock::new(InMemoryTemporaryStorageState {
-                block: PendingBlock::new_at_now(block_number),
-                block_changes: ExecutionChanges::default(),
+            pending_session: Mutex::new(()),
+            state: RwLock::new(InMemoryChainState {
+                pending_block: InMemoryTemporaryStorageState {
+                    block: PendingBlock::new_at_now(pending_block_number),
+                    block_changes: ExecutionChanges::default(),
+                },
+                latest_sealed: InMemorySealedBlock {
+                    state: InMemoryTemporaryStorageState::new_sealed(latest_sealed.number),
+                    hash: latest_sealed.hash,
+                },
             }),
-            latest_block: RwLock::new(None),
+        }
+    }
+
+    pub(super) fn pending_block_guard(&self) -> PendingBlockGuard<'_> {
+        PendingBlockGuard {
+            _guard: self.pending_session.lock(),
+        }
+    }
+
+    pub(super) fn read_latest_sealed(&self) -> BlockReference {
+        let state = self.state.read();
+        BlockReference {
+            number: state.latest_sealed.state.block.header.number,
+            hash: state.latest_sealed.hash,
         }
     }
 
     pub fn set_pending_header(&self, number: BlockNumber, timestamp: UnixTime) {
-        let mut pending_block = self.pending_block.write();
-        pending_block.block.header.number = number;
-        pending_block.block.header.timestamp = timestamp.into();
+        let mut state = self.state.write();
+        state.pending_block.block.header.number = number;
+        state.pending_block.block.header.timestamp = timestamp.into();
     }
 
     // -------------------------------------------------------------------------
@@ -59,13 +100,16 @@ impl InmemoryTransactionTemporaryStorage {
 
     // Uneeded clone here, return Cow
     pub fn read_pending_block_header(&self) -> (PendingBlockHeader, TxCount) {
-        let pending_block = self.pending_block.read();
-        (pending_block.block.header.clone(), (pending_block.block.transactions.len() as u64).into())
+        let state = self.state.read();
+        (
+            state.pending_block.block.header.clone(),
+            (state.pending_block.block.transactions.len() as u64).into(),
+        )
     }
 
     #[cfg(feature = "dev")]
     pub fn set_pending_block_header(&self, block_number: BlockNumber) -> anyhow::Result<(), StorageError> {
-        self.pending_block.write().block.header.number = block_number;
+        self.state.write().pending_block.block.header.number = block_number;
         Ok(())
     }
 
@@ -75,75 +119,66 @@ impl InmemoryTransactionTemporaryStorage {
 
     pub fn save_pending_execution(&self, tx: TransactionExecution) -> Result<(), StorageError> {
         // check conflicts
-        let pending_block = self.pending_block.upgradable_read();
-        if tx.evm_input != &pending_block.block.header {
+        let state = self.state.upgradable_read();
+        if tx.evm_input != &state.pending_block.block.header {
             let actual_input = tx.evm_input.clone();
             let tx_input: TransactionInput = tx.into();
-            let expected_input = EvmInput::from_eth_transaction(&tx_input, pending_block.block.header.number, *pending_block.block.header.timestamp);
+            let expected_input =
+                EvmInput::from_eth_transaction(&tx_input, state.pending_block.block.header.number, *state.pending_block.block.header.timestamp);
             return Err(StorageError::EvmInputMismatch {
                 expected: Box::new(expected_input),
                 actual: Box::new(actual_input),
             });
         }
 
-        let mut pending_block = RwLockUpgradableReadGuard::<InMemoryTemporaryStorageState>::upgrade(pending_block);
+        let mut state = RwLockUpgradableReadGuard::<InMemoryChainState>::upgrade(state);
 
-        pending_block.block_changes.merge(tx.result.execution.changes.clone()); // TODO: This clone can be removed by reworking the primitives
+        state.pending_block.block_changes.merge(tx.result.execution.changes.clone()); // TODO: This clone can be removed by reworking the primitives
 
         // save execution
-        pending_block.block.push_transaction(tx);
+        state.pending_block.block.push_transaction(tx);
 
         Ok(())
     }
 
     pub fn read_pending_executions(&self) -> Vec<TransactionExecution> {
-        self.pending_block.read().block.transactions.iter().map(|(_, tx)| tx.clone()).collect()
+        self.state.read().pending_block.block.transactions.iter().map(|(_, tx)| tx.clone()).collect()
     }
 
-    pub fn clone_pending_state(&self) -> InMemoryTemporaryStorageState {
-        let pending_block = self.pending_block.read();
-        (*pending_block).clone()
-    }
-
-    pub fn finish_pending_block(&self) -> anyhow::Result<(PendingBlock, ExecutionChanges), StorageError> {
-        let pending_block = self.pending_block.upgradable_read();
-        let changes = pending_block.block_changes.clone();
+    pub fn pending_block_to_seal(&self) -> (PendingBlock, ExecutionChanges) {
+        let state = self.state.read();
+        let block = state.pending_block.block.clone();
 
-        // This has to happen BEFORE creating the new state, because UnixTimeNow::default() may change the offset.
+        // This has to happen before creating the next state because UnixTimeNow::default() may change the offset.
         #[cfg(feature = "dev")]
-        let finished_block = {
-            let mut finished_block = pending_block.block.clone();
-            // Update block timestamp only if evm_setNextBlockTimestamp was called,
-            // otherwise keep the original timestamp from pending block creation
+        let block = {
+            let mut block = block;
+            // Update the timestamp only if evm_setNextBlockTimestamp was called.
             if UnixTime::evm_set_next_block_timestamp_was_called() {
-                finished_block.header.timestamp = UnixTimeNow::default();
+                block.header.timestamp = UnixTimeNow::default();
             }
-            finished_block
+            block
         };
 
-        let next_state = InMemoryTemporaryStorageState::new(pending_block.block.header.number.next_block_number());
-
-        let mut pending_block = RwLockUpgradableReadGuard::<InMemoryTemporaryStorageState>::upgrade(pending_block);
-        let mut latest = self.latest_block.write();
-
-        *latest = Some(std::mem::replace(&mut *pending_block, next_state));
-
-        drop(pending_block);
+        (block, state.pending_block.block_changes.clone())
+    }
 
-        #[cfg(not(feature = "dev"))]
-        let finished_block = {
-            let latest = RwLockWriteGuard::<Option<InMemoryTemporaryStorageState>>::downgrade(latest);
+    pub(super) fn finish_pending_block(&self, block: BlockReference, timestamp: UnixTimeNow) {
+        let next_state = InMemoryTemporaryStorageState::new(block.number.next_block_number());
+        let mut state = self.state.write();
 
-            #[allow(clippy::expect_used)]
-            latest.as_ref().expect("latest should be Some after finishing the pending block").block.clone()
+        debug_assert_eq!(state.pending_block.block.header.number, block.number);
+        let mut finished_state = std::mem::replace(&mut state.pending_block, next_state);
+        finished_state.block.header.timestamp = timestamp;
+        state.latest_sealed = InMemorySealedBlock {
+            state: finished_state,
+            hash: block.hash,
         };
-
-        Ok((finished_block, changes))
     }
 
     pub fn read_pending_execution(&self, hash: Hash) -> anyhow::Result<Option<TransactionExecution>, StorageError> {
-        let pending_block = self.pending_block.read();
-        match pending_block.block.transactions.get(&hash) {
+        let state = self.state.read();
+        match state.pending_block.block.transactions.get(&hash) {
             Some(tx) => Ok(Some(tx.clone())),
             None => Ok(None),
         }
@@ -154,25 +189,30 @@ impl InmemoryTransactionTemporaryStorage {
     // -------------------------------------------------------------------------
 
     pub fn read_account(&self, address: Address) -> anyhow::Result<Option<Account>, StorageError> {
-        Ok(match self.pending_block.read().block_changes.accounts.get(&address) {
+        let state = self.state.read();
+        Ok(match state.pending_block.block_changes.accounts.get(&address) {
             Some(pending_account) => Some(pending_account.clone().to_account(address)),
-            None => self
-                .latest_block
-                .read()
-                .as_ref()
-                .and_then(|latest| latest.block_changes.accounts.get(&address))
+            None => state
+                .latest_sealed
+                .state
+                .block_changes
+                .accounts
+                .get(&address)
                 .map(|account| account.clone().to_account(address)),
         })
     }
 
     pub fn read_slot(&self, address: Address, index: SlotIndex) -> anyhow::Result<Option<Slot>, StorageError> {
-        Ok(match self.pending_block.read().block_changes.slots.get(&(address, index)) {
+        let state = self.state.read();
+        Ok(match state.pending_block.block_changes.slots.get(&(address, index)) {
             Some(pending_value) => Some(Slot::new(index, *pending_value)),
-            None => self
-                .latest_block
-                .read()
-                .as_ref()
-                .and_then(|latest| latest.block_changes.slots.get(&(address, index)).map(|value| Slot::new(index, *value))),
+            None => state
+                .latest_sealed
+                .state
+                .block_changes
+                .slots
+                .get(&(address, index))
+                .map(|value| Slot::new(index, *value)),
         })
     }
 
@@ -182,17 +222,17 @@ impl InmemoryTransactionTemporaryStorage {
 
     #[cfg(feature = "dev")]
     pub fn save_slot(&self, address: Address, slot: Slot) -> anyhow::Result<(), StorageError> {
-        let mut pending_block = self.pending_block.write();
-        pending_block.block_changes.slots.insert((address, slot.index), slot.value);
+        let mut state = self.state.write();
+        state.pending_block.block_changes.slots.insert((address, slot.index), slot.value);
         Ok(())
     }
 
     #[cfg(feature = "dev")]
     pub fn save_account_nonce(&self, address: Address, nonce: Nonce) -> anyhow::Result<(), StorageError> {
-        let mut pending_block = self.pending_block.write();
+        let mut state = self.state.write();
 
         // Only update if the account exists
-        if let Some(account) = pending_block.block_changes.accounts.get_mut(&address) {
+        if let Some(account) = state.pending_block.block_changes.accounts.get_mut(&address) {
             account.nonce.apply(nonce);
         }
 
@@ -201,10 +241,10 @@ impl InmemoryTransactionTemporaryStorage {
 
     #[cfg(feature = "dev")]
     pub fn save_account_balance(&self, address: Address, balance: Wei) -> anyhow::Result<(), StorageError> {
-        let mut pending_block = self.pending_block.write();
+        let mut state = self.state.write();
 
         // Only update if the account exists
-        if let Some(account) = pending_block.block_changes.accounts.get_mut(&address) {
+        if let Some(account) = state.pending_block.block_changes.accounts.get_mut(&address) {
             account.balance.apply(balance);
         }
 
@@ -215,10 +255,10 @@ impl InmemoryTransactionTemporaryStorage {
     pub fn save_account_code(&self, address: Address, code: Bytes) -> anyhow::Result<(), StorageError> {
         use crate::alias::RevmBytecode;
 
-        let mut pending_block = self.pending_block.write();
+        let mut state = self.state.write();
 
         // Only update if the account exists
-        if let Some(account) = pending_block.block_changes.accounts.get_mut(&address) {
+        if let Some(account) = state.pending_block.block_changes.accounts.get_mut(&address) {
             account.bytecode.apply(if code.0.is_empty() {
                 None
             } else {
@@ -233,8 +273,13 @@ impl InmemoryTransactionTemporaryStorage {
     // Global state
     // -------------------------------------------------------------------------
     pub fn reset(&self) -> anyhow::Result<(), StorageError> {
-        self.pending_block.write().reset();
-        *self.latest_block.write() = None;
+        let genesis = BlockReference::genesis();
+        let mut state = self.state.write();
+        state.pending_block.reset();
+        state.latest_sealed = InMemorySealedBlock {
+            state: InMemoryTemporaryStorageState::new_sealed(genesis.number),
+            hash: genesis.hash,
+        };
         Ok(())
     }
 }
diff --git a/src/eth/storage/temporary/mod.rs b/src/eth/storage/temporary/mod.rs
index 9dd3d9d97..c6539da6d 100644
--- a/src/eth/storage/temporary/mod.rs
+++ b/src/eth/storage/temporary/mod.rs
@@ -1,4 +1,5 @@
 pub use inmemory::InMemoryTemporaryStorage;
+pub use inmemory::PendingBlockGuard;
 
 mod inmemory;
 
@@ -22,16 +23,36 @@ impl TemporaryStorageConfig {
     /// Initializes temporary storage implementation.
     pub fn init(&self, perm_storage: &RocksPermanentStorage) -> anyhow::Result<InMemoryTemporaryStorage> {
         tracing::info!(config = ?self, "creating temporary storage");
-        let pending_block_number = compute_pending_block_number(perm_storage)?;
-        Ok(InMemoryTemporaryStorage::new(pending_block_number))
+        Ok(InMemoryTemporaryStorage::new(perm_storage.read_chain_tip()?))
     }
 }
 
 pub fn compute_pending_block_number(perm_storage: &RocksPermanentStorage) -> anyhow::Result<BlockNumber> {
-    let mined_block_number = perm_storage.read_mined_block_number();
-    Ok(if !perm_storage.has_genesis()? && mined_block_number == BlockNumber::ZERO {
-        BlockNumber::ZERO
-    } else {
-        mined_block_number + 1
-    })
+    Ok(perm_storage
+        .read_chain_tip()?
+        .map_or(BlockNumber::ZERO, |saved_tip| saved_tip.number.next_block_number()))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use super::*;
+    use crate::eth::storage::permanent::RocksCfCacheConfig;
+
+    #[test]
+    fn empty_permanent_storage_initializes_pending_genesis() {
+        let rocks_dir = tempfile::tempdir().expect("create rocks directory");
+        let rocks_prefix = rocks_dir.path().join("empty-chain").to_string_lossy().into_owned();
+        let permanent = RocksPermanentStorage::new(Some(rocks_prefix), Duration::from_secs(240), RocksCfCacheConfig::default(), true, None, 1024)
+            .expect("create permanent storage");
+
+        let temporary = TemporaryStorageConfig {}.init(&permanent).expect("create temporary storage");
+
+        assert_eq!(temporary.read_pending_block_header().0.number, BlockNumber::ZERO);
+        assert_eq!(
+            compute_pending_block_number(&permanent).expect("compute pending block number"),
+            BlockNumber::ZERO
+        );
+    }
 }