diff --git a/rust/Cargo.lock b/rust/Cargo.lock
index 409c371e5b7..d9b392e0e12 100644
--- a/rust/Cargo.lock
+++ b/rust/Cargo.lock
@@ -11427,6 +11427,7 @@ dependencies = [
  "reth-tasks",
  "reth-trie",
  "reth-trie-common",
+ "reth-trie-db",
  "secp256k1 0.31.1",
  "serde",
  "serial_test",
diff --git a/rust/op-reth/crates/cli/src/commands/op_proofs/backfill.rs b/rust/op-reth/crates/cli/src/commands/op_proofs/backfill.rs
new file mode 100644
index 00000000000..c35d351279f
--- /dev/null
+++ b/rust/op-reth/crates/cli/src/commands/op_proofs/backfill.rs
@@ -0,0 +1,124 @@
+//! Command that backfills OP proofs storage to an older earliest block.
+
+use clap::Parser;
+use reth_cli::chainspec::ChainSpecParser;
+use reth_cli_commands::common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs};
+use reth_node_core::version::version_metadata;
+use reth_optimism_chainspec::OpChainSpec;
+use reth_optimism_node::args::ProofsStorageVersion;
+use reth_optimism_primitives::OpPrimitives;
+use reth_optimism_trie::{
+    BackfillJob, OpProofsProviderRO, OpProofsStore,
+    db::{MdbxProofsStorage, MdbxProofsStorageV2},
+};
+use reth_provider::{
+    BlockHashReader, BlockNumReader, ChangeSetReader, DBProvider, DatabaseProviderFactory,
+    HeaderProvider, StageCheckpointReader, StorageChangeSetReader, StorageSettingsCache,
+};
+use std::{path::PathBuf, sync::Arc};
+use tracing::info;
+
+/// Backfills the proofs storage to an older earliest block.
+#[derive(Debug, Parser)]
+pub struct BackfillCommand<C: ChainSpecParser> {
+    #[command(flatten)]
+    env: EnvironmentArgs<C>,
+
+    /// The path to the storage DB for proofs history.
+    #[arg(
+        long = "proofs-history.storage-path",
+        value_name = "PROOFS_HISTORY_STORAGE_PATH",
+        required = true
+    )]
+    pub storage_path: PathBuf,
+
+    /// Target earliest block number after backfill.
+    #[arg(long = "proofs-history.target-earliest-block", value_name = "TARGET_EARLIEST_BLOCK")]
+    pub target_earliest_block: u64,
+
+    /// Storage schema version. Must match the version used when starting the node.
+    #[arg(
+        long = "proofs-history.storage-version",
+        value_name = "PROOFS_HISTORY_STORAGE_VERSION",
+        default_value = "v1"
+    )]
+    pub storage_version: ProofsStorageVersion,
+}
+
+impl<C: ChainSpecParser<ChainSpec = OpChainSpec>> BackfillCommand<C> {
+    /// Execute [`BackfillCommand`].
+    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec, Primitives = OpPrimitives>>(
+        self,
+        runtime: reth_tasks::Runtime,
+    ) -> eyre::Result<()> {
+        info!(target: "reth::cli", "reth {} starting", version_metadata().short_version);
+        info!(target: "reth::cli", "Backfilling OP proofs storage at: {:?}", self.storage_path);
+
+        let Environment { provider_factory, .. } =
+            self.env.init::<N>(AccessRights::RO, runtime)?;
+
+        match self.storage_version {
+            ProofsStorageVersion::V1 => {
+                let storage: Arc<MdbxProofsStorage> = Arc::new(
+                    MdbxProofsStorage::new(&self.storage_path)
+                        .map_err(|e| eyre::eyre!("Failed to create MdbxProofsStorage: {e}"))?,
+                );
+                Self::run_backfill(&provider_factory, storage, self.target_earliest_block)?;
+            }
+            ProofsStorageVersion::V2 => {
+                let storage: Arc<MdbxProofsStorageV2> = Arc::new(
+                    MdbxProofsStorageV2::new(&self.storage_path)
+                        .map_err(|e| eyre::eyre!("Failed to create MdbxProofsStorageV2: {e}"))?,
+                );
+                Self::run_backfill(&provider_factory, storage, self.target_earliest_block)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn run_backfill<F, S>(
+        provider_factory: &F,
+        storage: S,
+        target_earliest_block: u64,
+    ) -> eyre::Result<()>
+    where
+        F: DatabaseProviderFactory,
+        F::Provider: DBProvider
+            + StageCheckpointReader
+            + ChangeSetReader
+            + StorageChangeSetReader
+            + BlockNumReader
+            + BlockHashReader
+            + HeaderProvider
+            + StorageSettingsCache
+            + Send,
+        S: OpProofsStore + Send,
+    {
+        let ro = storage.provider_ro()?;
+        let earliest = ro.get_earliest_block_number()?;
+        let latest = ro.get_latest_block_number()?;
+        drop(ro);
+        info!(
+            target: "reth::cli",
+            ?earliest,
+            ?latest,
+            target_earliest_block,
+            "Starting backfill job"
+        );
+
+        let provider = provider_factory
+            .database_provider_ro()
+            .map_err(|e| eyre::eyre!("Failed to open reth DB provider: {e}"))?;
+
+        BackfillJob::new(provider, storage).run(target_earliest_block)?;
+        Ok(())
+    }
+}
+
+impl<C: ChainSpecParser> BackfillCommand<C> {
+    /// Returns the underlying chain being used to run this command
+    pub const fn chain_spec(&self) -> Option<&Arc<C::ChainSpec>> {
+        Some(&self.env.chain)
+    }
+}
diff --git a/rust/op-reth/crates/cli/src/commands/op_proofs/mod.rs b/rust/op-reth/crates/cli/src/commands/op_proofs/mod.rs
index fae3476e494..cd9799ac65a 100644
--- a/rust/op-reth/crates/cli/src/commands/op_proofs/mod.rs
+++ b/rust/op-reth/crates/cli/src/commands/op_proofs/mod.rs
@@ -8,6 +8,7 @@ use reth_optimism_primitives::OpPrimitives;
 use std::sync::Arc;
 
 pub mod init;
+pub mod backfill;
 pub mod prune;
 pub mod unwind;
 
@@ -26,6 +27,7 @@ impl<C: ChainSpecParser<ChainSpec = OpChainSpec>> Command<C> {
     ) -> eyre::Result<()> {
         match self.command {
             Subcommands::Init(cmd) => cmd.execute::<N>(runtime).await,
+            Subcommands::Backfill(cmd) => cmd.execute::<N>(runtime).await,
             Subcommands::Prune(cmd) => cmd.execute::<N>(runtime).await,
             Subcommands::Unwind(cmd) => cmd.execute::<N>(runtime).await,
         }
@@ -37,6 +39,7 @@ impl<C: ChainSpecParser> Command<C> {
     pub const fn chain_spec(&self) -> Option<&Arc<C::ChainSpec>> {
         match &self.command {
             Subcommands::Init(cmd) => cmd.chain_spec(),
+            Subcommands::Backfill(cmd) => cmd.chain_spec(),
             Subcommands::Prune(cmd) => cmd.chain_spec(),
             Subcommands::Unwind(cmd) => cmd.chain_spec(),
         }
@@ -49,6 +52,9 @@ pub enum Subcommands<C: ChainSpecParser> {
     /// Initialize the proofs storage with the current state of the chain
     #[command(name = "init")]
     Init(init::InitCommand<C>),
+    /// Backfill proofs history to an older earliest block
+    #[command(name = "backfill")]
+    Backfill(backfill::BackfillCommand<C>),
     /// Prune old proof history to reclaim space
     #[command(name = "prune")]
     Prune(prune::PruneCommand<C>),
diff --git a/rust/op-reth/crates/trie/Cargo.toml b/rust/op-reth/crates/trie/Cargo.toml
index 0d430993537..22dd853561d 100644
--- a/rust/op-reth/crates/trie/Cargo.toml
+++ b/rust/op-reth/crates/trie/Cargo.toml
@@ -21,6 +21,7 @@ reth-provider.workspace = true
 reth-revm.workspace = true
 reth-trie = { workspace = true, features = ["serde"] }
 reth-trie-common = { workspace = true, features = ["serde"] }
+reth-trie-db.workspace = true
 reth-codecs.workspace = true
 reth-tasks.workspace = true
 
diff --git a/rust/op-reth/crates/trie/src/api.rs b/rust/op-reth/crates/trie/src/api.rs
index 8a17d111455..d11a9219503 100644
--- a/rust/op-reth/crates/trie/src/api.rs
+++ b/rust/op-reth/crates/trie/src/api.rs
@@ -162,6 +162,126 @@ pub trait OpProofsProviderRw: OpProofsProviderRO {
     fn commit(self) -> OpProofsStorageResult<()>;
 }
 
+/// Provider for writing historical records for blocks older than the current window boundary.
+///
+/// Unlike [`OpProofsProviderRw::store_trie_updates`], which is strictly append-only (validates
+/// parent hash against `latest` and advances `latest`), this provider is designed for
+/// **prepend-style** writes that extend the window backward.  It does not touch the `latest`
+/// marker, and it does not enforce parent-hash ordering against `latest`.
+///
+/// The typical call sequence for one backfill step is:
+/// ```ignore
+/// let bp = storage.backfill_provider()?;
+/// bp.prepend_block(block_ref, diff)?;
+/// bp.commit()?;
+/// ```
+pub trait OpProofsBackfillProvider: OpProofsProviderRO {
+    /// Write historical changeset and history-bitmap entries for `block_ref`, and move the
+    /// `earliest` marker to `block_ref.parent`.
+    ///
+    /// `diff` contains:
+    /// - `sorted_trie_updates`: trie node **before-values** for `block_ref.block.number`
+    ///   (i.e. what each changed node looked like *before* the block executed).
+    /// - `sorted_post_state`: account / storage **before-values** for the same block.
+    ///
+    /// The implementation must **not** update the `latest` marker and must **not**
+    /// validate `diff` against the current `latest` block.
+    fn prepend_block(
+        &self,
+        block_ref: BlockWithParent,
+        diff: BlockStateDiff,
+    ) -> OpProofsStorageResult<WriteCounts>;
+
+    /// Commit the transaction. Consumes the provider.
+    fn commit(self) -> OpProofsStorageResult<()>;
+}
+
+/// Blanket impl of [`OpProofsProviderRO`] for shared references.
+///
+/// This allows passing `&bp` (where `bp: OpProofsBackfillProvider + OpProofsProviderRO`)
+/// to APIs that require `P: OpProofsProviderRO + Clone`. Since `&T: Copy`, cloning a
+/// reference is free, enabling `StateRoot::overlay_root(&bp, ...)` to work without
+/// requiring the underlying provider to implement `Clone`.
+impl<'a, T: OpProofsProviderRO + 'a> OpProofsProviderRO for &'a T {
+    type StorageTrieCursor<'tx>
+        = T::StorageTrieCursor<'tx>
+    where
+        Self: 'tx,
+        T: 'tx;
+    type AccountTrieCursor<'tx>
+        = T::AccountTrieCursor<'tx>
+    where
+        Self: 'tx,
+        T: 'tx;
+    type StorageCursor<'tx>
+        = T::StorageCursor<'tx>
+    where
+        Self: 'tx,
+        T: 'tx;
+    type AccountHashedCursor<'tx>
+        = T::AccountHashedCursor<'tx>
+    where
+        Self: 'tx,
+        T: 'tx;
+
+    fn get_earliest_block_number(&self) -> crate::OpProofsStorageResult<Option<(u64, B256)>> {
+        T::get_earliest_block_number(self)
+    }
+
+    fn get_latest_block_number(&self) -> crate::OpProofsStorageResult<Option<(u64, B256)>> {
+        T::get_latest_block_number(self)
+    }
+
+    fn storage_trie_cursor<'tx>(
+        &self,
+        hashed_address: B256,
+        max_block_number: u64,
+    ) -> crate::OpProofsStorageResult<Self::StorageTrieCursor<'tx>>
+    where
+        'a: 'tx,
+    {
+        T::storage_trie_cursor(self, hashed_address, max_block_number)
+    }
+
+    fn account_trie_cursor<'tx>(
+        &self,
+        max_block_number: u64,
+    ) -> crate::OpProofsStorageResult<Self::AccountTrieCursor<'tx>>
+    where
+        'a: 'tx,
+    {
+        T::account_trie_cursor(self, max_block_number)
+    }
+
+    fn storage_hashed_cursor<'tx>(
+        &self,
+        hashed_address: B256,
+        max_block_number: u64,
+    ) -> crate::OpProofsStorageResult<Self::StorageCursor<'tx>>
+    where
+        'a: 'tx,
+    {
+        T::storage_hashed_cursor(self, hashed_address, max_block_number)
+    }
+
+    fn account_hashed_cursor<'tx>(
+        &self,
+        max_block_number: u64,
+    ) -> crate::OpProofsStorageResult<Self::AccountHashedCursor<'tx>>
+    where
+        'a: 'tx,
+    {
+        T::account_hashed_cursor(self, max_block_number)
+    }
+
+    fn fetch_trie_updates(
+        &self,
+        block_number: u64,
+    ) -> crate::OpProofsStorageResult<BlockStateDiff> {
+        T::fetch_trie_updates(self, block_number)
+    }
+}
+
 /// Factory trait for creating providers to interact with the proofs storage.
 #[auto_impl(Arc)]
 pub trait OpProofsStore: Send + Sync + Debug {
@@ -180,6 +300,11 @@ pub trait OpProofsStore: Send + Sync + Debug {
     where
         Self: 'a;
 
+    /// The backfill provider type created by the factory.
+    type BackfillProvider<'a>: OpProofsBackfillProvider + 'a
+    where
+        Self: 'a;
+
     /// Create a read-only provider for interacting with the proofs storage.
     fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>>;
 
@@ -188,6 +313,9 @@ pub trait OpProofsStore: Send + Sync + Debug {
 
     /// Create an initialization provider for interacting with the proofs storage.
     fn initialization_provider<'a>(&'a self) -> OpProofsStorageResult<Self::Initializer<'a>>;
+
+    /// Create a backfill provider for prepend-style writes that extend the window backward.
+    fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>>;
 }
 
 /// Status of the initial state anchor.
diff --git a/rust/op-reth/crates/trie/src/backfill/changesets.rs b/rust/op-reth/crates/trie/src/backfill/changesets.rs
new file mode 100644
index 00000000000..32d296db4ff
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/backfill/changesets.rs
@@ -0,0 +1,105 @@
+//! Per-block backfill diff computation.
+//!
+//! For block N, [`compute_block_backfill_diff`] returns:
+//! - `HashedPostStateSorted` — per-block leaf revert (account & storage values
+//!   before block N ran). Read directly from reth's `AccountChangeSets` /
+//!   `StorageChangeSets` and reused as `BlockStateDiff::sorted_post_state`.
+//! - `TrieUpdatesSorted` — trie-node before-values for paths block N touched.
+//!   Written into the four changeset tables by `prepend_block`.
+//!
+//! # Algorithm
+//!
+//! Conceptually equivalent to
+//! `reth_trie_db::changesets::compute_block_trie_changesets_inner`, but reads
+//! the trie from the **op-reth proofs storage** at `max_block_number = N`
+//! instead of from reth's current-state tables. That swap is what makes the
+//! per-block cost scale with `k_N` (this block's diff) rather than `K_tail`
+//! (every changeset entry between N and the DB tip).
+//!
+//! A single call to `overlay_root_from_nodes_with_updates` does the work:
+//! - **Cursor**: `OpProofsHashedAccountCursorFactory` / `OpProofsTrieCursorFactory`
+//!   at `max=N` — serves state@N.
+//! - **State overlay**: the per-block leaf revert (`individual_state_revert`).
+//!   Walked together with the cursor, this yields state@N-1 for every leaf
+//!   block N touched.
+//! - **Prefix sets**: only the paths block N's leaf revert covers.
+//!
+//! The returned `TrieUpdates` is the difference between trie@N (cursor view)
+//! and trie@N-1 (after applying the overlay) for those paths — which is
+//! exactly the trie changeset we want:
+//! - branch modified at N → `(path, Some(value_at_N-1))`
+//! - branch destroyed at N (existed at N-1, gone at N) → `(path, Some(value_at_N-1))`
+//! - branch created at N (existed at N, gone at N-1) → `(path, None)` via `removed_nodes`
+
+use crate::{
+    OpProofsProviderRO, backfill::error::BackfillError, proof::DatabaseStateRoot,
+};
+use alloy_primitives::BlockNumber;
+use reth_provider::{
+    BlockNumReader, ChangeSetReader, DBProvider, ProviderError, StorageChangeSetReader,
+    StorageSettingsCache,
+};
+use reth_trie::{StateRoot, TrieInput};
+use reth_trie_common::{HashedPostStateSorted, updates::TrieUpdatesSorted};
+use reth_trie_db::from_reverts_auto;
+
+/// Compute the backfill diff for `block_number`: the trie-node before-values
+/// for the changeset table, and the per-block leaf revert reused as
+/// `BlockStateDiff::sorted_post_state`.
+///
+/// `proofs_provider` must reflect the proofs-storage state *at the start of
+/// this iteration* — i.e. `earliest == block_number`. Callers should open a
+/// fresh RO provider per iteration so it sees writes committed by the
+/// previous `prepend_block`.
+pub(super) fn compute_block_backfill_diff<P, R>(
+    reth_provider: &P,
+    proofs_provider: R,
+    block_number: BlockNumber,
+) -> Result<(TrieUpdatesSorted, HashedPostStateSorted), BackfillError>
+where
+    P: ChangeSetReader
+        + StorageChangeSetReader
+        + BlockNumReader
+        + DBProvider
+        + StorageSettingsCache,
+    R: OpProofsProviderRO + Clone,
+{
+    // Per-block leaf revert: doubles as `post_state` for `prepend_block` and
+    // as the state overlay for the trie@N-1 reconstruction below.
+    let individual_state_revert = from_reverts_auto(reth_provider, block_number..=block_number)?;
+    let trie_changesets = compute_trie_changesets_against_proofs(
+        proofs_provider,
+        block_number,
+        &individual_state_revert,
+    )?;
+    Ok((trie_changesets, individual_state_revert))
+}
+
+fn compute_trie_changesets_against_proofs<R>(
+    proofs_provider: R,
+    block_number: BlockNumber,
+    individual_state_revert: &HashedPostStateSorted,
+) -> Result<TrieUpdatesSorted, BackfillError>
+where
+    R: OpProofsProviderRO + Clone,
+{
+    // Apply block N's leaf revert as a state overlay on top of the proofs
+    // cursor at max=N, then walk just the paths block N touched. The returned
+    // `TrieUpdates` describes how trie@N-1 differs from the cursor's view at
+    // max=N — which is exactly the changeset:
+    //   - modified branch  → (path, Some(value_at_N-1))
+    //   - destroyed at N   → (path, Some(value_at_N-1))
+    //   - created at N     → (path, None)   (via `removed_nodes`)
+    let input = TrieInput {
+        nodes: Default::default(),
+        state: individual_state_revert.clone().into(),
+        prefix_sets: individual_state_revert.construct_prefix_sets(),
+    };
+    let (_, trie_updates) = StateRoot::overlay_root_from_nodes_with_updates(
+        proofs_provider,
+        block_number,
+        input,
+    )
+    .map_err(ProviderError::other)?;
+    Ok(trie_updates.into_sorted())
+}
diff --git a/rust/op-reth/crates/trie/src/backfill/error.rs b/rust/op-reth/crates/trie/src/backfill/error.rs
new file mode 100644
index 00000000000..fe60b950e65
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/backfill/error.rs
@@ -0,0 +1,32 @@
+//! Error type for backfill operations.
+
+use crate::OpProofsStorageError;
+use alloy_primitives::B256;
+use reth_execution_errors::StateRootError;
+use reth_provider::ProviderError;
+
+/// Error type for backfill operations.
+#[derive(Debug, thiserror::Error)]
+pub enum BackfillError {
+    /// Error bubbled up from proofs storage operations.
+    #[error(transparent)]
+    Storage(#[from] OpProofsStorageError),
+    /// Error from reth provider operations.
+    #[error(transparent)]
+    Provider(#[from] ProviderError),
+    /// State root computation failed.
+    #[error(transparent)]
+    StateRoot(#[from] StateRootError),
+    /// Computed state root does not match the expected root from the header.
+    #[error(
+        "State root mismatch at block {block_number}: computed {computed:?}, expected {expected:?}"
+    )]
+    StateRootMismatch {
+        /// Block number being validated (the block whose before-state is being checked).
+        block_number: u64,
+        /// Computed root from the proofs storage overlay.
+        computed: B256,
+        /// Expected root from reth's block header.
+        expected: B256,
+    },
+}
diff --git a/rust/op-reth/crates/trie/src/backfill/job.rs b/rust/op-reth/crates/trie/src/backfill/job.rs
new file mode 100644
index 00000000000..f0c65b45da5
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/backfill/job.rs
@@ -0,0 +1,170 @@
+//! [`BackfillJob`] implementation.
+
+use super::{changesets::compute_block_backfill_diff, error::BackfillError};
+use crate::{
+    BlockStateDiff, OpProofsBackfillProvider, OpProofsProviderRO, OpProofsStorageError,
+    OpProofsStore,
+    proof::DatabaseStateRoot,
+};
+use alloy_eips::{BlockNumHash, eip1898::BlockWithParent};
+use alloy_primitives::BlockNumber;
+use derive_more::Constructor;
+use reth_primitives_traits::AlloyBlockHeader;
+use reth_provider::{
+    BlockHashReader, BlockNumReader, ChangeSetReader, DBProvider, HeaderProvider, ProviderError,
+    StageCheckpointReader, StorageChangeSetReader, StorageSettingsCache,
+};
+use reth_trie::StateRoot;
+use reth_trie_common::HashedPostState;
+use std::time::Instant;
+use tracing::info;
+
+/// How often to emit a progress line during a long backfill, measured in
+/// blocks committed.
+const LOG_EVERY: u64 = 1_000;
+
+/// Backfill job for proofs storage.
+#[derive(Debug, Constructor)]
+pub struct BackfillJob<P, S: OpProofsStore + Send> {
+    provider: P,
+    storage: S,
+}
+
+impl<P, S> BackfillJob<P, S>
+where
+    P: DBProvider
+        + StageCheckpointReader
+        + ChangeSetReader
+        + StorageChangeSetReader
+        + BlockNumReader
+        + BlockHashReader
+        + HeaderProvider
+        + StorageSettingsCache
+        + Send,
+    S: OpProofsStore + Send,
+{
+    /// Backfill proofs data down to `target_earliest_block`.
+    ///
+    /// Extends the stored proof window from `[earliest, latest]` backward to
+    /// `[target_earliest_block, latest]`. Each block is committed atomically so
+    /// the job is restart-safe: on crash, resume from the current `earliest`.
+    ///
+    /// Returns immediately if `target_earliest_block >= current earliest`.
+    pub fn run(&self, target_earliest_block: u64) -> Result<(), BackfillError> {
+        let ro = self.storage.provider_ro()?;
+        let Some((current_earliest, _)) = ro.get_earliest_block_number()? else {
+            return Err(BackfillError::Storage(OpProofsStorageError::NoBlocksFound));
+        };
+        drop(ro);
+
+        if target_earliest_block >= current_earliest {
+            return Ok(());
+        }
+
+        let total = current_earliest - target_earliest_block;
+        let start = Instant::now();
+        info!(
+            target: "reth::op-proofs::backfill",
+            from = current_earliest,
+            to = target_earliest_block,
+            total,
+            "Starting proofs backfill"
+        );
+
+        for block_number in (target_earliest_block + 1..=current_earliest).rev() {
+            self.backfill_block(block_number)?;
+
+            let done = current_earliest - block_number + 1;
+            let is_final = block_number == target_earliest_block + 1;
+            if done.is_multiple_of(LOG_EVERY) || is_final {
+                let elapsed_secs = start.elapsed().as_secs_f64();
+                let blocks_per_sec = if elapsed_secs.is_normal() {
+                    done as f64 / elapsed_secs
+                } else {
+                    0.0
+                };
+                let eta_secs = if blocks_per_sec.is_normal() && blocks_per_sec > 0.0 {
+                    (total - done) as f64 / blocks_per_sec
+                } else {
+                    0.0
+                };
+                let progress_pct = (done as f64 / total as f64) * 100.0;
+                info!(
+                    target: "reth::op-proofs::backfill",
+                    done,
+                    total,
+                    "progress: {progress_pct:.2}% ({blocks_per_sec:.1} blk/s, ETA {eta_secs:.0}s)"
+                );
+            }
+        }
+
+        info!(
+            target: "reth::op-proofs::backfill",
+            blocks = total,
+            elapsed = ?start.elapsed(),
+            "Proofs backfill complete"
+        );
+
+        Ok(())
+    }
+
+    /// Backfill a single block `E`: write its historical records and advance `earliest` to `E-1`.
+    fn backfill_block(&self, block_number: BlockNumber) -> Result<(), BackfillError> {
+        let block_hash = self
+            .provider
+            .block_hash(block_number)?
+            .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
+        let parent_hash = self
+            .provider
+            .block_hash(block_number - 1)?
+            .ok_or_else(|| ProviderError::HeaderNotFound((block_number - 1).into()))?;
+
+        // Open a fresh RO proofs provider for this iteration: it sees writes
+        // committed by the previous `prepend_block`, so its cursor at max=N
+        // already reflects state@N. Dropped before opening the RW backfill
+        // provider below to avoid holding two transactions on the same env.
+        let trie_updates;
+        let post_state;
+        {
+            let proofs_ro = self.storage.provider_ro()?;
+            (trie_updates, post_state) =
+                compute_block_backfill_diff(&self.provider, &proofs_ro, block_number)?;
+        }
+
+        let block_ref = BlockWithParent {
+            block: BlockNumHash::new(block_number, block_hash),
+            parent: parent_hash,
+        };
+
+        let bp = self.storage.backfill_provider()?;
+        bp.prepend_block(
+            block_ref,
+            BlockStateDiff {
+                sorted_trie_updates: trie_updates,
+                sorted_post_state: post_state,
+            },
+        )?;
+
+        // Validate the written before-values by computing a full state root at block_number - 1
+        // using the backfill provider (which now includes the prepended data in its transaction).
+        // `&bp` implements `OpProofsProviderRO`, so it reads its own uncommitted writes.
+        let expected_root = self
+            .provider
+            .header_by_number(block_number - 1)?
+            .ok_or_else(|| ProviderError::HeaderNotFound((block_number - 1).into()))?
+            .state_root();
+        let computed_root =
+            StateRoot::overlay_root(&bp, block_number - 1, HashedPostState::default())?;
+        if computed_root != expected_root {
+            return Err(BackfillError::StateRootMismatch {
+                block_number,
+                computed: computed_root,
+                expected: expected_root,
+            });
+        }
+
+        bp.commit()?;
+
+        Ok(())
+    }
+}
diff --git a/rust/op-reth/crates/trie/src/backfill/mod.rs b/rust/op-reth/crates/trie/src/backfill/mod.rs
new file mode 100644
index 00000000000..709fd858146
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/backfill/mod.rs
@@ -0,0 +1,57 @@
+//! Backfill job for extending proofs storage window backward.
+//!
+//! # Backfill plan (v2)
+//!
+//! Goal: extend proofs window from `[earliest, latest]` to
+//! `[target_earliest, latest]` where `target_earliest < earliest`.
+//!
+//! ## Core boundary semantics
+//!
+//! - `earliest` is a **base-state boundary**, not "oldest block with its own changeset rows".
+//! - To move boundary from `E` to `E-1`, we must materialize block `E` historical records
+//!   (changesets + history bitmap entries), then set `earliest = E-1`.
+//! - This mirrors prune behavior in reverse.
+//!
+//! ## Per-step algorithm (descending)
+//!
+//! For each `E` from current earliest down to `target_earliest + 1`:
+//! 1. Build historical trie/state views from reth DB (no block execution):
+//!    - `before_E`: state at end of `E-1` (start of block `E`)
+//!    - `after_E`: state at end of `E` (start of block `E+1`)
+//! 2. Derive block `E` changes:
+//!    - leaf (hashed account/storage) before-values from reth `AccountChangeSets` /
+//!      `StorageChangeSets` tables
+//!    - trie node before-values via [`ChangesetCache::get_or_compute`]
+//! 3. Write block `E` records into proofs history tables.
+//! 4. Atomically move earliest marker to `E-1` and commit.
+//!
+//! ## Data sources
+//!
+//! - Leaf before-values come from reth changesets (`account_block_changeset` /
+//!   `storage_changeset`).
+//! - Trie node before-values come from [`ChangesetCache`], which uses
+//!   `compute_block_trie_changesets` as a DB fallback.
+//! - A shared [`ChangesetCache`] is kept across the whole backfill run so that
+//!   blocks that are warm in cache are not recomputed.
+//!
+//! ## Write invariants
+//!
+//! - Do not rewind proofs current-state tables; they remain at `latest`.
+//! - Backfill writes are prepend-style history inserts for older blocks.
+//! - Each step must be idempotent/restart-safe: after crash, resume from current `earliest`.
+//!
+//! ## Validation per step
+//!
+//! - Window remains contiguous after commit.
+//! - `earliest` decreases by exactly one per successful step.
+//! - Historical reads at new boundary succeed, while reads below boundary fail as expected.
+
+mod changesets;
+mod error;
+mod job;
+
+#[cfg(test)]
+mod tests;
+
+pub use error::BackfillError;
+pub use job::BackfillJob;
diff --git a/rust/op-reth/crates/trie/src/backfill/tests.rs b/rust/op-reth/crates/trie/src/backfill/tests.rs
new file mode 100644
index 00000000000..4b686f3d8bb
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/backfill/tests.rs
@@ -0,0 +1,453 @@
+//! Integration tests for [`BackfillJob`].
+//!
+//! Helpers here mirror `crates/trie/tests/live.rs` because integration-test
+//! helpers in `tests/` are not reachable from `src/`. If this duplication grows,
+//! consider extracting a shared `test_utils` module behind a feature flag.
+
+use super::{BackfillError, BackfillJob};
+use crate::{
+    MdbxProofsStorageV2, OpProofsStore, OpProofsStorageError,
+    RethTrieStorageLayout,
+    api::OpProofsProviderRO,
+    initialize::InitializationJob,
+};
+use alloy_consensus::{BlockHeader, Header, TxEip2930, constants::ETH_TO_WEI};
+use alloy_genesis::{Genesis, GenesisAccount};
+use alloy_primitives::{Address, B256, Bytes, TxKind, U256, keccak256};
+use reth_chainspec::{ChainSpec, ChainSpecBuilder, EthereumHardfork, MAINNET, MIN_TRANSACTION_GAS};
+use reth_db::Database;
+use reth_db_common::init::init_genesis;
+use reth_ethereum_primitives::{Block, BlockBody, Receipt, Transaction, TransactionSigned};
+use reth_evm::{ConfigureEvm, execute::Executor};
+use reth_evm_ethereum::EthEvmConfig;
+use reth_node_api::{NodePrimitives, NodeTypesWithDB};
+use reth_primitives_traits::{Block as _, RecoveredBlock};
+use reth_provider::{
+    BlockWriter as _, DatabaseProviderFactory, ExecutionOutcome, HashedPostStateProvider,
+    LatestStateProviderRef, ProviderFactory, StateRootProvider, StorageSettingsCache,
+    providers::ProviderNodeTypes,
+    test_utils::create_test_provider_factory_with_chain_spec,
+};
+use reth_revm::database::StateProviderDatabase;
+use secp256k1::{Keypair, Secp256k1, rand::rng};
+use serial_test::serial;
+use std::sync::Arc;
+use tempfile::TempDir;
+
+// ============================ Chain construction helpers ============================
+
+fn create_storage() -> Arc<MdbxProofsStorageV2> {
+    let path = TempDir::new().unwrap();
+    Arc::new(MdbxProofsStorageV2::new(path.path()).unwrap())
+}
+
+fn public_key_to_address(pubkey: secp256k1::PublicKey) -> Address {
+    let hash = keccak256(&pubkey.serialize_uncompressed()[1..]);
+    Address::from_slice(&hash[12..])
+}
+
+fn sign_tx_with_key_pair(key_pair: Keypair, tx: Transaction) -> TransactionSigned {
+    use alloy_consensus::SignableTransaction;
+    use reth_primitives_traits::crypto::secp256k1::sign_message;
+    let secret = B256::from_slice(&key_pair.secret_bytes());
+    let sig = sign_message(secret, tx.signature_hash()).unwrap();
+    tx.into_signed(sig).into()
+}
+
+/// Pre-allocated contract address for storage-write tests.
+const STORAGE_CONTRACT: Address = Address::repeat_byte(0xAB);
+
+/// Minimal contract that writes `BLOCKNUMBER` (i.e. current block.number) to
+/// storage slot 0:
+///
+/// ```text
+///   0x43        BLOCKNUMBER     push block.number
+///   0x60 0x00   PUSH1 0x00      push slot 0
+///   0x55        SSTORE          store
+///   0x00        STOP
+/// ```
+const STORAGE_BYTECODE: [u8; 5] = [0x43, 0x60, 0x00, 0x55, 0x00];
+
+fn chain_spec_with_address(address: Address) -> Arc<ChainSpec> {
+    Arc::new(
+        ChainSpecBuilder::default()
+            .chain(MAINNET.chain)
+            .genesis(Genesis {
+                alloc: [
+                    (
+                        address,
+                        GenesisAccount {
+                            balance: U256::from(10 * ETH_TO_WEI),
+                            ..Default::default()
+                        },
+                    ),
+                    (
+                        STORAGE_CONTRACT,
+                        GenesisAccount {
+                            code: Some(Bytes::from_static(&STORAGE_BYTECODE)),
+                            ..Default::default()
+                        },
+                    ),
+                ]
+                .into(),
+                ..MAINNET.genesis.clone()
+            })
+            .paris_activated()
+            .build(),
+    )
+}
+
+/// Construct an unsealed block with a single simple transfer.
+fn build_transfer_block(
+    block_number: u64,
+    parent_hash: B256,
+    chain_spec: &Arc<ChainSpec>,
+    key_pair: Keypair,
+    nonce: u64,
+    recipient: Address,
+) -> RecoveredBlock<Block> {
+    let tx = sign_tx_with_key_pair(
+        key_pair,
+        TxEip2930 {
+            chain_id: chain_spec.chain.id(),
+            nonce,
+            gas_limit: MIN_TRANSACTION_GAS,
+            gas_price: 1_500_000_000,
+            to: TxKind::Call(recipient),
+            value: U256::from(1),
+            ..Default::default()
+        }
+        .into(),
+    );
+    Block {
+        header: Header {
+            parent_hash,
+            receipts_root: alloy_primitives::b256!(
+                "0xd3a6acf9a244d78b33831df95d472c4128ea85bf079a1d41e32ed0b7d2244c9e"
+            ),
+            difficulty: chain_spec.fork(EthereumHardfork::Paris).ttd().expect("Paris TTD"),
+            number: block_number,
+            gas_limit: MIN_TRANSACTION_GAS,
+            gas_used: MIN_TRANSACTION_GAS,
+            state_root: B256::ZERO, // filled in by execute_block
+            ..Default::default()
+        },
+        body: BlockBody { transactions: vec![tx], ..Default::default() },
+    }
+    .try_into_recovered()
+    .unwrap()
+}
+
+fn execute_block<N>(
+    block: &mut RecoveredBlock<Block>,
+    provider_factory: &ProviderFactory<N>,
+    chain_spec: &Arc<ChainSpec>,
+) -> reth_evm::execute::BlockExecutionOutput<Receipt>
+where
+    N: ProviderNodeTypes<
+            Primitives: NodePrimitives<Block = Block, BlockBody = BlockBody, Receipt = Receipt>,
+        > + NodeTypesWithDB,
+{
+    let provider = provider_factory.provider().unwrap();
+    let db = StateProviderDatabase::new(LatestStateProviderRef::new(&provider));
+    let evm_config = EthEvmConfig::ethereum(chain_spec.clone());
+    let block_executor = evm_config.batch_executor(db);
+    let execution_result = block_executor.execute(block).unwrap();
+
+    let hashed_state =
+        LatestStateProviderRef::new(&provider).hashed_post_state(&execution_result.state);
+    let state_root = LatestStateProviderRef::new(&provider).state_root(hashed_state).unwrap();
+    block.set_state_root(state_root);
+    execution_result
+}
+
+fn commit_block_to_database<N>(
+    block: &RecoveredBlock<Block>,
+    execution_output: &reth_evm::execute::BlockExecutionOutput<Receipt>,
+    provider_factory: &ProviderFactory<N>,
+) where
+    N: ProviderNodeTypes<
+            Primitives: NodePrimitives<Block = Block, BlockBody = BlockBody, Receipt = Receipt>,
+        > + NodeTypesWithDB,
+{
+    let execution_outcome = ExecutionOutcome {
+        bundle: execution_output.state.clone(),
+        receipts: vec![execution_output.receipts.clone()],
+        first_block: block.number(),
+        requests: vec![execution_output.requests.clone()],
+    };
+    let state_provider = provider_factory.provider().unwrap();
+    let hashed_state = HashedPostStateProvider::hashed_post_state(
+        &LatestStateProviderRef::new(&state_provider),
+        &execution_output.state,
+    );
+    let provider_rw = provider_factory.provider_rw().unwrap();
+    provider_rw
+        .append_blocks_with_state(vec![block.clone()], &execution_outcome, hashed_state.into_sorted())
+        .unwrap();
+    provider_rw.commit().unwrap();
+}
+
+/// Construct an unsealed block whose sole tx calls [`STORAGE_CONTRACT`],
+/// triggering an SSTORE of `block.number` into slot 0 of the contract's storage.
+///
+/// Gas accounting: the executor recomputes `gas_used` against the actual EVM
+/// trace, so we deliberately set `gas_limit == gas_used` to a value large
+/// enough to cover both the 21 000-gas tx base cost and the worst-case cold
+/// SSTORE (~22 100 gas).
+fn build_storage_call_block(
+    block_number: u64,
+    parent_hash: B256,
+    chain_spec: &Arc<ChainSpec>,
+    key_pair: Keypair,
+    nonce: u64,
+) -> RecoveredBlock<Block> {
+    const CALL_GAS_LIMIT: u64 = 100_000;
+    let tx = sign_tx_with_key_pair(
+        key_pair,
+        TxEip2930 {
+            chain_id: chain_spec.chain.id(),
+            nonce,
+            gas_limit: CALL_GAS_LIMIT,
+            gas_price: 1_500_000_000,
+            to: TxKind::Call(STORAGE_CONTRACT),
+            value: U256::ZERO,
+            ..Default::default()
+        }
+        .into(),
+    );
+    Block {
+        header: Header {
+            parent_hash,
+            receipts_root: alloy_primitives::b256!(
+                "0xd3a6acf9a244d78b33831df95d472c4128ea85bf079a1d41e32ed0b7d2244c9e"
+            ),
+            difficulty: chain_spec.fork(EthereumHardfork::Paris).ttd().expect("Paris TTD"),
+            number: block_number,
+            gas_limit: CALL_GAS_LIMIT,
+            gas_used: CALL_GAS_LIMIT,
+            state_root: B256::ZERO,
+            ..Default::default()
+        },
+        body: BlockBody { transactions: vec![tx], ..Default::default() },
+    }
+    .try_into_recovered()
+    .unwrap()
+}
+
+/// Build a chain of `num_blocks` simple transfer blocks on top of a freshly
+/// initialized genesis, then initialize the v2 proofs storage at the latest
+/// block. Returns the provider factory, the storage, and the latest
+/// (number, hash) pair.
+fn build_chain_and_initialize_storage(
+    num_blocks: u64,
+) -> (
+    ProviderFactory<reth_provider::test_utils::MockNodeTypesWithDB>,
+    Arc<MdbxProofsStorageV2>,
+    u64,
+    B256,
+) {
+    let secp = Secp256k1::new();
+    let key_pair = Keypair::new(&secp, &mut rng());
+    let sender = public_key_to_address(key_pair.public_key());
+
+    let chain_spec = chain_spec_with_address(sender);
+    let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
+    init_genesis(&provider_factory).unwrap();
+
+    let recipient = Address::repeat_byte(0x42);
+    let mut last_hash = chain_spec.genesis_hash();
+    let mut last_number = 0u64;
+    for n in 1..=num_blocks {
+        let mut block =
+            build_transfer_block(n, last_hash, &chain_spec, key_pair, n - 1, recipient);
+        let exec = execute_block(&mut block, &provider_factory, &chain_spec);
+        commit_block_to_database(&block, &exec, &provider_factory);
+        last_hash = block.hash();
+        last_number = n;
+    }
+
+    let storage = create_storage();
+    {
+        let trie_layout = if provider_factory.cached_storage_settings().is_v2() {
+            RethTrieStorageLayout::Packed
+        } else {
+            RethTrieStorageLayout::Legacy
+        };
+        let tx = provider_factory.db_ref().tx().unwrap();
+        InitializationJob::new(storage.clone(), tx, trie_layout)
+            .run(last_number, last_hash)
+            .unwrap();
+    }
+
+    (provider_factory, storage, last_number, last_hash)
+}
+
+/// Like [`build_chain_and_initialize_storage`] but every block calls
+/// [`STORAGE_CONTRACT`], so each block produces hashed-storage changesets in
+/// addition to the account-level ones.
+fn build_chain_with_storage_writes_and_initialize_storage(
+    num_blocks: u64,
+) -> (
+    ProviderFactory<reth_provider::test_utils::MockNodeTypesWithDB>,
+    Arc<MdbxProofsStorageV2>,
+    u64,
+    B256,
+) {
+    let secp = Secp256k1::new();
+    let key_pair = Keypair::new(&secp, &mut rng());
+    let sender = public_key_to_address(key_pair.public_key());
+
+    let chain_spec = chain_spec_with_address(sender);
+    let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
+    init_genesis(&provider_factory).unwrap();
+
+    let mut last_hash = chain_spec.genesis_hash();
+    let mut last_number = 0u64;
+    for n in 1..=num_blocks {
+        let mut block = build_storage_call_block(n, last_hash, &chain_spec, key_pair, n - 1);
+        let exec = execute_block(&mut block, &provider_factory, &chain_spec);
+        commit_block_to_database(&block, &exec, &provider_factory);
+        last_hash = block.hash();
+        last_number = n;
+    }
+
+    let storage = create_storage();
+    {
+        let trie_layout = if provider_factory.cached_storage_settings().is_v2() {
+            RethTrieStorageLayout::Packed
+        } else {
+            RethTrieStorageLayout::Legacy
+        };
+        let tx = provider_factory.db_ref().tx().unwrap();
+        InitializationJob::new(storage.clone(), tx, trie_layout)
+            .run(last_number, last_hash)
+            .unwrap();
+    }
+
+    (provider_factory, storage, last_number, last_hash)
+}
+
+// ============================ Tests ============================
+
+#[test]
+#[serial]
+fn run_is_noop_when_target_at_or_above_earliest() {
+    // Build a chain of 3 blocks; storage initialized at block 3 (earliest = 3).
+    let (provider_factory, storage, latest_num, latest_hash) =
+        build_chain_and_initialize_storage(3);
+
+    // target == earliest: no-op.
+    {
+        let provider = provider_factory.database_provider_ro().unwrap();
+        BackfillJob::new(provider, storage.clone()).run(latest_num).unwrap();
+        let ro = storage.provider_ro().unwrap();
+        assert_eq!(ro.get_earliest_block_number().unwrap(), Some((latest_num, latest_hash)));
+    }
+
+    // target > earliest: also no-op.
+    {
+        let provider = provider_factory.database_provider_ro().unwrap();
+        BackfillJob::new(provider, storage.clone()).run(latest_num + 100).unwrap();
+        let ro = storage.provider_ro().unwrap();
+        assert_eq!(ro.get_earliest_block_number().unwrap(), Some((latest_num, latest_hash)));
+    }
+}
+
+#[test]
+#[serial]
+fn run_errors_when_storage_uninitialized() {
+    let secp = Secp256k1::new();
+    let key_pair = Keypair::new(&secp, &mut rng());
+    let chain_spec = chain_spec_with_address(public_key_to_address(key_pair.public_key()));
+    let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec);
+    init_genesis(&provider_factory).unwrap();
+
+    // Storage created but never initialized — no earliest marker.
+    let storage = create_storage();
+    let provider = provider_factory.database_provider_ro().unwrap();
+    let err = BackfillJob::new(provider, storage).run(0).unwrap_err();
+    assert!(
+        matches!(err, BackfillError::Storage(OpProofsStorageError::NoBlocksFound)),
+        "expected NoBlocksFound, got {err:?}"
+    );
+}
+
+#[test]
+#[serial]
+fn run_extends_window_backward_multi_block() {
+    // 5-block chain — exercises descending iteration across multiple
+    // `BackfillContext::step` calls.
+    let (provider_factory, storage, latest_num, latest_hash) =
+        build_chain_and_initialize_storage(5);
+
+    {
+        let ro = storage.provider_ro().unwrap();
+        assert_eq!(ro.get_earliest_block_number().unwrap(), Some((latest_num, latest_hash)));
+    }
+
+    {
+        let provider = provider_factory.database_provider_ro().unwrap();
+        BackfillJob::new(provider, storage.clone()).run(0).unwrap();
+    }
+
+    let provider = provider_factory.database_provider_ro().unwrap();
+    let genesis_hash = reth_provider::BlockHashReader::block_hash(&provider, 0).unwrap().unwrap();
+    let ro = storage.provider_ro().unwrap();
+    assert_eq!(ro.get_earliest_block_number().unwrap(), Some((0, genesis_hash)));
+}
+
+#[test]
+#[serial]
+fn run_extends_window_backward() {
+    // Smallest possible case: 1-block chain, single backfill step from 1 → 0.
+    let (provider_factory, storage, latest_num, latest_hash) =
+        build_chain_and_initialize_storage(1);
+
+    // Sanity: earliest starts at the latest block.
+    {
+        let ro = storage.provider_ro().unwrap();
+        assert_eq!(ro.get_earliest_block_number().unwrap(), Some((latest_num, latest_hash)));
+    }
+
+    // Backfill all the way down to block 0 (genesis).
+    {
+        let provider = provider_factory.database_provider_ro().unwrap();
+        BackfillJob::new(provider, storage.clone()).run(0).unwrap();
+    }
+
+    // Earliest should now point at block 0 (the genesis hash).
+    let provider = provider_factory.database_provider_ro().unwrap();
+    let genesis_hash = reth_provider::BlockHashReader::block_hash(&provider, 0).unwrap().unwrap();
+    let ro = storage.provider_ro().unwrap();
+    assert_eq!(ro.get_earliest_block_number().unwrap(), Some((0, genesis_hash)));
+}
+
+#[test]
+#[serial]
+fn run_extends_window_backward_with_storage_writes() {
+    // Every block calls `STORAGE_CONTRACT`, writing `block.number` to slot 0.
+    // This exercises the backfill code paths that are silent in plain-transfer
+    // tests:
+    //   - `V2HashedStorageChangeSets` / `V2HashedStoragesHistory` writes during
+    //     `prepend_block` (the slot value changes every block).
+    //   - Storage-side reconstruction via `V2StorageCursor` at each historical
+    //     block during the in-job `StateRoot::overlay_root` validation.
+    let (provider_factory, storage, latest_num, latest_hash) =
+        build_chain_with_storage_writes_and_initialize_storage(5);
+
+    {
+        let ro = storage.provider_ro().unwrap();
+        assert_eq!(ro.get_earliest_block_number().unwrap(), Some((latest_num, latest_hash)));
+    }
+
+    {
+        let provider = provider_factory.database_provider_ro().unwrap();
+        BackfillJob::new(provider, storage.clone()).run(0).unwrap();
+    }
+
+    let provider = provider_factory.database_provider_ro().unwrap();
+    let genesis_hash = reth_provider::BlockHashReader::block_hash(&provider, 0).unwrap().unwrap();
+    let ro = storage.provider_ro().unwrap();
+    assert_eq!(ro.get_earliest_block_number().unwrap(), Some((0, genesis_hash)));
+}
diff --git a/rust/op-reth/crates/trie/src/db/store.rs b/rust/op-reth/crates/trie/src/db/store.rs
index d0df27b7250..b1aaae33ed4 100644
--- a/rust/op-reth/crates/trie/src/db/store.rs
+++ b/rust/op-reth/crates/trie/src/db/store.rs
@@ -4,8 +4,8 @@ use crate::{
     OpProofsStorageError::NoBlocksFound,
     OpProofsStorageResult,
     api::{
-        InitialStateAnchor, InitialStateStatus, OpProofsInitProvider, OpProofsProviderRO,
-        OpProofsProviderRw, OpProofsStore, WriteCounts,
+        InitialStateAnchor, InitialStateStatus, OpProofsBackfillProvider, OpProofsInitProvider,
+        OpProofsProviderRO, OpProofsProviderRw, OpProofsStore, WriteCounts,
     },
     db::{
         MdbxAccountCursor, MdbxStorageCursor, MdbxTrieCursor,
@@ -994,10 +994,28 @@ impl<TX: DbTxMut + DbTx + Send + Sync + Debug + 'static> OpProofsInitProvider
     }
 }
 
+impl<TX: DbTxMut + DbTx + Send + Sync + Debug + 'static> OpProofsBackfillProvider
+    for MdbxProofsProvider<TX>
+{
+    fn prepend_block(
+        &self,
+        _block_ref: BlockWithParent,
+        _diff: BlockStateDiff,
+    ) -> OpProofsStorageResult<WriteCounts> {
+        todo!("OpProofsBackfillProvider::prepend_block for MdbxProofsProvider")
+    }
+
+    fn commit(self) -> OpProofsStorageResult<()> {
+        self.tx.commit()?;
+        Ok(())
+    }
+}
+
 impl OpProofsStore for MdbxProofsStorage {
     type ProviderRO<'a> = Arc<MdbxProofsProvider<<DatabaseEnv as Database>::TX>>;
     type ProviderRw<'a> = MdbxProofsProvider<<DatabaseEnv as Database>::TXMut>;
     type Initializer<'a> = MdbxProofsProvider<<DatabaseEnv as Database>::TXMut>;
+    type BackfillProvider<'a> = MdbxProofsProvider<<DatabaseEnv as Database>::TXMut>;
 
     fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>> {
         Ok(Arc::new(MdbxProofsProvider::new(self.env.tx()?)))
@@ -1010,6 +1028,10 @@ impl OpProofsStore for MdbxProofsStorage {
     fn initialization_provider<'a>(&'a self) -> OpProofsStorageResult<Self::Initializer<'a>> {
         Ok(MdbxProofsProvider::new(self.env.tx_mut()?))
     }
+
+    fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>> {
+        Ok(MdbxProofsProvider::new(self.env.tx_mut()?))
+    }
 }
 
 /// This implementation is copied from the
diff --git a/rust/op-reth/crates/trie/src/db/store_v2/backfill.rs b/rust/op-reth/crates/trie/src/db/store_v2/backfill.rs
new file mode 100644
index 00000000000..f4cc2c0d1c6
--- /dev/null
+++ b/rust/op-reth/crates/trie/src/db/store_v2/backfill.rs
@@ -0,0 +1,362 @@
+//! [`OpProofsBackfillProvider`] implementation for [`MdbxProofsProviderV2`].
+
+use super::{MdbxProofsProviderV2, NUM_OF_INDICES_IN_SHARD, write::HistoryCollector};
+use crate::{
+    BlockStateDiff, OpProofsStorageError, OpProofsStorageResult,
+    api::{OpProofsBackfillProvider, WriteCounts},
+    db::models::{
+        AccountTrieShardedKey, BlockNumberHashedAddress, HashedAccountBeforeTx,
+        HashedAccountShardedKey, HashedStorageShardedKey, StorageTrieShardedKey,
+        TrieChangeSetsEntry, V2AccountTrieChangeSets, V2AccountsTrieHistory,
+        V2HashedAccountChangeSets, V2HashedAccountsHistory, V2HashedStorageChangeSets,
+        V2HashedStoragesHistory, V2StorageTrieChangeSets, V2StoragesTrieHistory,
+    },
+};
+use alloy_eips::eip1898::BlockWithParent;
+use alloy_primitives::{B256, BlockNumber};
+use reth_db::{
+    BlockNumberList,
+    cursor::{DbCursorRO, DbCursorRW},
+    models::sharded_key::ShardedKey,
+    table::Table,
+    transaction::{DbTx, DbTxMut},
+};
+use reth_primitives_traits::StorageEntry;
+use reth_trie::{HashedPostStateSorted, StoredNibbles, StoredNibblesSubKey, updates::TrieUpdatesSorted};
+use std::{collections::BTreeMap, fmt::Debug};
+use tracing::debug;
+
+/// Insert `block_number` at the front of the first history-bitmap shard for a logical key.
+///
+/// Backfill prepends blocks in descending order, so `block_number` is always strictly
+/// less than every value already stored for this key. The existing
+/// [`super::write::append_history_indices_batched`] function only touches the last/sentinel
+/// shard; this function instead seeks the **first** shard and prepends there.
+///
+/// If the first shard would exceed [`NUM_OF_INDICES_IN_SHARD`] entries after the insert,
+/// it is split: the new earlier portion gets a fresh shard key keyed by its maximum
+/// block number, and the remainder stays under the original shard key.
+fn prepend_history_index_for_key<T>(
+    cursor: &mut (impl DbCursorRO<T> + DbCursorRW<T>),
+    block_number: BlockNumber,
+    first_shard_key: T::Key,
+    make_shard_key: impl Fn(BlockNumber) -> T::Key,
+    sentinel_key: T::Key,
+    same_logical_key: impl Fn(&T::Key) -> bool,
+) -> OpProofsStorageResult<()>
+where
+    T: Table<Value = BlockNumberList>,
+    T::Key: Clone,
+{
+    match cursor.seek(first_shard_key)? {
+        Some((old_key, existing)) if same_logical_key(&old_key) => {
+            // Build the merged sequence: [block_number, ...existing...].
+            // block_number < all existing values (prepend invariant), so the result is sorted.
+            let mut all_values: Vec<u64> =
+                std::iter::once(block_number).chain(existing.iter()).collect();
+
+            if all_values.len() <= NUM_OF_INDICES_IN_SHARD {
+                // Fits — update shard in-place (its max value, i.e. key, is unchanged).
+                let new_list = BlockNumberList::new_pre_sorted(all_values);
+                cursor.upsert(old_key, &new_list)?;
+            } else {
+                // Overflow — split into two shards:
+                //   first_chunk: [block_number, ..., a_{N-1}]  →  new key = a_{N-1}
+                //   rest:        [a_N, ..., a_K]               →  keep old_key (max unchanged)
+                let rest: Vec<u64> = all_values.split_off(NUM_OF_INDICES_IN_SHARD);
+                let first_chunk_max = *all_values.last().expect("non-empty");
+                let new_first_key = make_shard_key(first_chunk_max);
+                let first_list = BlockNumberList::new_pre_sorted(all_values);
+                let rest_list = BlockNumberList::new_pre_sorted(rest);
+                // Keep the existing shard key for the upper portion.
+                cursor.upsert(old_key, &rest_list)?;
+                // Insert the new lower shard.
+                cursor.upsert(new_first_key, &first_list)?;
+            }
+        }
+        _ => {
+            // No existing shard for this key — create the sentinel shard.
+            let new_list = BlockNumberList::new_pre_sorted([block_number]);
+            cursor.upsert(sentinel_key, &new_list)?;
+        }
+    }
+    Ok(())
+}
+
+impl<TX: DbTxMut + DbTx + Send + Sync + Debug + 'static> MdbxProofsProviderV2<TX> {
+    /// Returns `true` if any changeset entry already exists for `block_number`.
+    ///
+    /// Uses `V2HashedAccountChangeSets` as the sentinel table: nearly every block
+    /// touches at least one account. For the rare empty block the write loop is a
+    /// no-op regardless, so a false-negative here is harmless.
+    fn changeset_exists_for_block(&self, block_number: BlockNumber) -> OpProofsStorageResult<bool> {
+        let mut cs = self.tx.cursor_read::<V2HashedAccountChangeSets>()?;
+        Ok(cs.seek(block_number)?.is_some_and(|(bn, _)| bn == block_number))
+    }
+
+    /// Write changeset entries for `block_number` directly from `diff` (already before-values)
+    /// without reading or modifying the current-state tables.
+    fn prepend_block_changesets(
+        &self,
+        block_number: BlockNumber,
+        diff: BlockStateDiff,
+        collector: &mut HistoryCollector,
+    ) -> OpProofsStorageResult<WriteCounts> {
+        let BlockStateDiff { sorted_trie_updates, sorted_post_state } = diff;
+        Ok(WriteCounts {
+            account_trie_updates_written_total: self.write_account_trie_cs(
+                block_number,
+                &sorted_trie_updates,
+                collector,
+            )?,
+            storage_trie_updates_written_total: self.write_storage_trie_cs(
+                block_number,
+                &sorted_trie_updates,
+                collector,
+            )?,
+            hashed_accounts_written_total: self.write_hashed_accounts_cs(
+                block_number,
+                &sorted_post_state,
+                collector,
+            )?,
+            hashed_storages_written_total: self.write_hashed_storages_cs(
+                block_number,
+                &sorted_post_state,
+                collector,
+            )?,
+        })
+    }
+
+    fn write_account_trie_cs(
+        &self,
+        block_number: BlockNumber,
+        updates: &TrieUpdatesSorted,
+        collector: &mut HistoryCollector,
+    ) -> OpProofsStorageResult<u64> {
+        let mut cs = self.tx.cursor_dup_write::<V2AccountTrieChangeSets>()?;
+        let mut count = 0u64;
+        for (nibbles, maybe_node) in updates.account_nodes_ref() {
+            let stored = StoredNibbles(*nibbles);
+            cs.upsert(
+                block_number,
+                &TrieChangeSetsEntry {
+                    nibbles: StoredNibblesSubKey(*nibbles),
+                    node: maybe_node.clone(),
+                },
+            )?;
+            collector.account_trie.entry(stored).or_default().push(block_number);
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    fn write_storage_trie_cs(
+        &self,
+        block_number: BlockNumber,
+        updates: &TrieUpdatesSorted,
+        collector: &mut HistoryCollector,
+    ) -> OpProofsStorageResult<u64> {
+        let mut cs = self.tx.cursor_dup_write::<V2StorageTrieChangeSets>()?;
+        let mut count = 0u64;
+        for (hashed_address, nodes) in updates.storage_tries_ref() {
+            let cs_key = BlockNumberHashedAddress((block_number, *hashed_address));
+            for (nibbles, maybe_node) in nodes.storage_nodes_ref() {
+                cs.upsert(
+                    cs_key,
+                    &TrieChangeSetsEntry {
+                        nibbles: StoredNibblesSubKey(*nibbles),
+                        node: maybe_node.clone(),
+                    },
+                )?;
+                collector
+                    .storage_trie
+                    .entry((*hashed_address, StoredNibbles(*nibbles)))
+                    .or_default()
+                    .push(block_number);
+                count += 1;
+            }
+        }
+        Ok(count)
+    }
+
+    fn write_hashed_accounts_cs(
+        &self,
+        block_number: BlockNumber,
+        post_state: &HashedPostStateSorted,
+        collector: &mut HistoryCollector,
+    ) -> OpProofsStorageResult<u64> {
+        let mut cs = self.tx.cursor_dup_write::<V2HashedAccountChangeSets>()?;
+        let mut count = 0u64;
+        for &(hashed_address, maybe_account) in &post_state.accounts {
+            cs.upsert(
+                block_number,
+                &HashedAccountBeforeTx::new(hashed_address, maybe_account),
+            )?;
+            collector.hashed_accounts.entry(hashed_address).or_default().push(block_number);
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    fn write_hashed_storages_cs(
+        &self,
+        block_number: BlockNumber,
+        post_state: &HashedPostStateSorted,
+        collector: &mut HistoryCollector,
+    ) -> OpProofsStorageResult<u64> {
+        let mut cs = self.tx.cursor_dup_write::<V2HashedStorageChangeSets>()?;
+        let mut count = 0u64;
+        for (hashed_address, storage) in &post_state.storages {
+            let cs_key = BlockNumberHashedAddress((block_number, *hashed_address));
+            for &(slot, value) in &storage.storage_slots {
+                cs.upsert(cs_key, &StorageEntry { key: slot, value })?;
+                collector
+                    .hashed_storages
+                    .entry((*hashed_address, slot))
+                    .or_default()
+                    .push(block_number);
+                count += 1;
+            }
+        }
+        Ok(count)
+    }
+
+    /// Flush history-bitmap entries collected during a prepend operation.
+    ///
+    /// Unlike [`Self::flush_collected_history`] (which appends to the sentinel/last shard),
+    /// this inserts into the **first** shard because the new block number is smaller than
+    /// all existing entries.
+    fn prepend_collected_history(&self, collector: HistoryCollector) -> OpProofsStorageResult<()> {
+        self.prepend_account_trie_history(collector.account_trie)?;
+        self.prepend_storage_trie_history(collector.storage_trie)?;
+        self.prepend_hashed_account_history(collector.hashed_accounts)?;
+        self.prepend_hashed_storage_history(collector.hashed_storages)?;
+        Ok(())
+    }
+
+    fn prepend_account_trie_history(
+        &self,
+        entries: BTreeMap<StoredNibbles, Vec<BlockNumber>>,
+    ) -> OpProofsStorageResult<()> {
+        let mut cursor = self.tx.cursor_write::<V2AccountsTrieHistory>()?;
+        for (nibbles, blocks) in entries {
+            for block_number in blocks {
+                prepend_history_index_for_key(
+                    &mut cursor,
+                    block_number,
+                    AccountTrieShardedKey::new(nibbles.clone(), 0),
+                    |h| AccountTrieShardedKey::new(nibbles.clone(), h),
+                    AccountTrieShardedKey::new(nibbles.clone(), u64::MAX),
+                    |k| k.key == nibbles,
+                )?;
+            }
+        }
+        Ok(())
+    }
+
+    fn prepend_storage_trie_history(
+        &self,
+        entries: BTreeMap<(B256, StoredNibbles), Vec<BlockNumber>>,
+    ) -> OpProofsStorageResult<()> {
+        let mut cursor = self.tx.cursor_write::<V2StoragesTrieHistory>()?;
+        for ((addr, nibbles), blocks) in entries {
+            for block_number in blocks {
+                prepend_history_index_for_key(
+                    &mut cursor,
+                    block_number,
+                    StorageTrieShardedKey::new(addr, nibbles.clone(), 0),
+                    |h| StorageTrieShardedKey::new(addr, nibbles.clone(), h),
+                    StorageTrieShardedKey::new(addr, nibbles.clone(), u64::MAX),
+                    |k| k.hashed_address == addr && k.key == nibbles,
+                )?;
+            }
+        }
+        Ok(())
+    }
+
+    fn prepend_hashed_account_history(
+        &self,
+        entries: BTreeMap<B256, Vec<BlockNumber>>,
+    ) -> OpProofsStorageResult<()> {
+        let mut cursor = self.tx.cursor_write::<V2HashedAccountsHistory>()?;
+        for (addr, blocks) in entries {
+            for block_number in blocks {
+                prepend_history_index_for_key(
+                    &mut cursor,
+                    block_number,
+                    HashedAccountShardedKey::new(addr, 0),
+                    |h| HashedAccountShardedKey::new(addr, h),
+                    HashedAccountShardedKey::new(addr, u64::MAX),
+                    |k| k.0.key == addr,
+                )?;
+            }
+        }
+        Ok(())
+    }
+
+    fn prepend_hashed_storage_history(
+        &self,
+        entries: BTreeMap<(B256, B256), Vec<BlockNumber>>,
+    ) -> OpProofsStorageResult<()> {
+        let mut cursor = self.tx.cursor_write::<V2HashedStoragesHistory>()?;
+        for ((addr, slot), blocks) in entries {
+            for block_number in blocks {
+                prepend_history_index_for_key(
+                    &mut cursor,
+                    block_number,
+                    HashedStorageShardedKey {
+                        hashed_address: addr,
+                        sharded_key: ShardedKey::new(slot, 0),
+                    },
+                    |h| HashedStorageShardedKey {
+                        hashed_address: addr,
+                        sharded_key: ShardedKey::new(slot, h),
+                    },
+                    HashedStorageShardedKey {
+                        hashed_address: addr,
+                        sharded_key: ShardedKey::new(slot, u64::MAX),
+                    },
+                    |k| k.hashed_address == addr && k.sharded_key.key == slot,
+                )?;
+            }
+        }
+        Ok(())
+    }
+}
+
+impl<TX: DbTxMut + DbTx + Send + Sync + Debug + 'static> OpProofsBackfillProvider
+    for MdbxProofsProviderV2<TX>
+{
+    fn prepend_block(
+        &self,
+        block_ref: BlockWithParent,
+        diff: BlockStateDiff,
+    ) -> OpProofsStorageResult<WriteCounts> {
+        let block_number = block_ref.block.number;
+        let proof_window = self.get_proof_window_inner()?;
+        if block_ref.block.hash != proof_window.earliest.hash {
+            return Err(OpProofsStorageError::PrependOutOfOrder {
+                block_number,
+                block_hash: block_ref.block.hash,
+                earliest_block_number: proof_window.earliest.number,
+                earliest_block_hash: proof_window.earliest.hash,
+            });
+        }
+
+        if self.changeset_exists_for_block(block_number)? {
+            debug!(target: "op-reth::trie::backfill", block_number, "changeset already exists, skipping prepend");
+            return Ok(WriteCounts::default());
+        }
+
+        let mut collector = HistoryCollector::default();
+        let counts = self.prepend_block_changesets(block_number, diff, &mut collector)?;
+        self.prepend_collected_history(collector)?;
+        self.set_earliest_block_number_inner(block_number - 1, block_ref.parent)?;
+        Ok(counts)
+    }
+
+    fn commit(self) -> OpProofsStorageResult<()> {
+        self.tx.commit()?;
+        Ok(())
+    }
+}
diff --git a/rust/op-reth/crates/trie/src/db/store_v2/init.rs b/rust/op-reth/crates/trie/src/db/store_v2/init.rs
index d9f3ef2f535..3ac6eb47314 100644
--- a/rust/op-reth/crates/trie/src/db/store_v2/init.rs
+++ b/rust/op-reth/crates/trie/src/db/store_v2/init.rs
@@ -145,6 +145,7 @@ impl<TX: DbTxMut + DbTx + Send + Sync + Debug + 'static> OpProofsInitProvider
         let anchor =
             self.get_initial_state_anchor_inner()?.ok_or(OpProofsStorageError::NoBlocksFound)?;
         self.set_earliest_block_number_inner(anchor.number, anchor.hash)?;
+        self.set_latest_block_number_inner(anchor.number, anchor.hash)?;
         Ok(anchor)
     }
 
diff --git a/rust/op-reth/crates/trie/src/db/store_v2/mod.rs b/rust/op-reth/crates/trie/src/db/store_v2/mod.rs
index 1b2d3f99247..bbb75e10ba0 100644
--- a/rust/op-reth/crates/trie/src/db/store_v2/mod.rs
+++ b/rust/op-reth/crates/trie/src/db/store_v2/mod.rs
@@ -10,6 +10,7 @@
 //! | Storage Trie | `V2StoragesTrie` | `V2StorageTrieChangeSets` | `V2StoragesTrieHistory` |
 
 pub(crate) mod cursor;
+mod backfill;
 mod init;
 #[cfg(feature = "metrics")]
 mod metrics;
@@ -57,6 +58,7 @@ impl OpProofsStore for MdbxProofsStorageV2 {
     type ProviderRO<'a> = Arc<MdbxProofsProviderV2<<DatabaseEnv as Database>::TX>>;
     type ProviderRw<'a> = MdbxProofsProviderV2<<DatabaseEnv as Database>::TXMut>;
     type Initializer<'a> = MdbxProofsProviderV2<<DatabaseEnv as Database>::TXMut>;
+    type BackfillProvider<'a> = MdbxProofsProviderV2<<DatabaseEnv as Database>::TXMut>;
 
     fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>> {
         Ok(Arc::new(MdbxProofsProviderV2::new(self.env.tx()?)))
@@ -69,6 +71,10 @@ impl OpProofsStore for MdbxProofsStorageV2 {
     fn initialization_provider<'a>(&'a self) -> OpProofsStorageResult<Self::Initializer<'a>> {
         Ok(MdbxProofsProviderV2::new(self.env.tx_mut()?))
     }
+
+    fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>> {
+        Ok(MdbxProofsProviderV2::new(self.env.tx_mut()?))
+    }
 }
 
 // =============================================================================
diff --git a/rust/op-reth/crates/trie/src/db/store_v2/tests.rs b/rust/op-reth/crates/trie/src/db/store_v2/tests.rs
index 3863d80716e..cf953b038bc 100644
--- a/rust/op-reth/crates/trie/src/db/store_v2/tests.rs
+++ b/rust/op-reth/crates/trie/src/db/store_v2/tests.rs
@@ -17,7 +17,10 @@ use tempfile::TempDir;
 
 use crate::{
     BlockStateDiff, OpProofsStorageError,
-    api::{OpProofsInitProvider, OpProofsProviderRO, OpProofsProviderRw},
+    api::{
+        OpProofsBackfillProvider, OpProofsInitProvider, OpProofsProviderRO, OpProofsProviderRw,
+        WriteCounts,
+    },
     db::{
         ProofWindowKey, V2ProofWindow,
         models::{
@@ -1890,3 +1893,186 @@ fn hashed_storages_batch_no_duplicates() {
     assert_eq!(slots.len(), 1, "batch: exactly 1 entry, no duplicates");
     assert_eq!(slots[0], (slot, U256::from(300u64)));
 }
+
+// ========================== Backfill (prepend_block) tests ==========================
+
+#[test]
+fn prepend_block_basic_advances_earliest_and_writes_changeset() {
+    let db = setup_db();
+
+    let addr = B256::from([0xDD; 32]);
+    let block5_hash = B256::repeat_byte(0x05);
+    let block4_hash = B256::repeat_byte(0x04);
+    let before_account = Account { nonce: 10, ..Default::default() };
+
+    // earliest = (5, block5_hash) — backfill window currently empty at block 5
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        provider.set_earliest_block_number(5, block5_hash).expect("set earliest");
+        OpProofsProviderRw::commit(provider).expect("commit");
+    }
+
+    // Prepend block 5 with addr's before-account = state at end of block 4.
+    let counts = {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        let mut post_state = HashedPostState::default();
+        post_state.accounts.insert(addr, Some(before_account));
+        let diff = BlockStateDiff {
+            sorted_trie_updates: TrieUpdates::default().into_sorted(),
+            sorted_post_state: post_state.into_sorted(),
+        };
+        let block_ref = make_block_ref(5, block5_hash, block4_hash);
+        let counts = provider.prepend_block(block_ref, diff).expect("prepend");
+        OpProofsBackfillProvider::commit(provider).expect("commit");
+        counts
+    };
+    assert_eq!(counts.hashed_accounts_written_total, 1);
+    assert_eq!(counts.hashed_storages_written_total, 0);
+    assert_eq!(counts.account_trie_updates_written_total, 0);
+    assert_eq!(counts.storage_trie_updates_written_total, 0);
+
+    // earliest advanced to (4, block4_hash).
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx().expect("ro"));
+        assert_eq!(
+            provider.get_earliest_block_number().expect("get"),
+            Some((4, block4_hash))
+        );
+    }
+
+    // Changeset entry for block 5 carries the supplied before-value.
+    {
+        let tx = db.tx().expect("ro");
+        let mut cur = tx.cursor_dup_read::<V2HashedAccountChangeSets>().expect("cursor");
+        let entry = cur.seek_by_key_subkey(5u64, addr).expect("seek").expect("exists");
+        assert_eq!(entry.hashed_address, addr);
+        assert_eq!(entry.info.unwrap().nonce, 10);
+    }
+
+    // History bitmap created with sentinel shard holding [5].
+    {
+        let tx = db.tx().expect("ro");
+        let mut cur = tx.cursor_read::<V2HashedAccountsHistory>().expect("cursor");
+        let shard_key = HashedAccountShardedKey::new(addr, u64::MAX);
+        let (_, bitmap) = cur.seek_exact(shard_key).expect("seek").expect("exists");
+        assert_eq!(bitmap.iter().collect::<Vec<_>>(), vec![5]);
+    }
+}
+
+#[test]
+fn prepend_block_hash_mismatch_rejects() {
+    let db = setup_db();
+
+    let block5_hash = B256::repeat_byte(0x05);
+    let wrong_hash = B256::repeat_byte(0xFF);
+
+    // earliest = (5, block5_hash)
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        provider.set_earliest_block_number(5, block5_hash).expect("set earliest");
+        OpProofsProviderRw::commit(provider).expect("commit");
+    }
+
+    // Attempt to prepend with a non-matching hash → PrependOutOfOrder.
+    let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+    let block_ref = make_block_ref(5, wrong_hash, B256::repeat_byte(0x04));
+    let res = provider.prepend_block(block_ref, BlockStateDiff::default());
+    assert!(
+        matches!(res, Err(OpProofsStorageError::PrependOutOfOrder { .. })),
+        "expected PrependOutOfOrder, got {res:?}"
+    );
+}
+
+#[test]
+fn prepend_block_idempotent_when_changeset_exists() {
+    let db = setup_db();
+
+    let addr = B256::from([0xEE; 32]);
+    let block1_hash = B256::repeat_byte(0x01);
+
+    // init at block 0, forward-store block 1 (creates changesets + history for addr).
+    init_state(&db, vec![(addr, Some(Account::default()))]);
+    store_block(&db, make_block_ref(1, block1_hash, B256::ZERO), make_nonce_diff(addr, 1));
+
+    // Simulate a retry scenario: rewind earliest back to (1, block1_hash) and
+    // attempt to prepend block 1 again. The changeset-exists guard should short-circuit.
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        provider.set_earliest_block_number(1, block1_hash).expect("rewind earliest");
+        OpProofsProviderRw::commit(provider).expect("commit");
+    }
+
+    let counts = {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        let block_ref = make_block_ref(1, block1_hash, B256::ZERO);
+        let counts = provider.prepend_block(block_ref, make_nonce_diff(addr, 42)).expect("prepend");
+        OpProofsBackfillProvider::commit(provider).expect("commit");
+        counts
+    };
+
+    // Idempotency: zero write counts AND earliest unchanged.
+    assert_eq!(counts, WriteCounts::default());
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx().expect("ro"));
+        assert_eq!(
+            provider.get_earliest_block_number().expect("get"),
+            Some((1, block1_hash))
+        );
+    }
+
+    // Changeset retains the ORIGINAL forward-write value (nonce: 0), not the
+    // would-be-prepended value (nonce: 42).
+    let tx = db.tx().expect("ro");
+    let mut cur = tx.cursor_dup_read::<V2HashedAccountChangeSets>().expect("cursor");
+    let entry = cur.seek_by_key_subkey(1u64, addr).expect("seek").expect("exists");
+    assert_eq!(entry.info.unwrap_or_default().nonce, 0);
+}
+
+#[test]
+fn prepend_block_descending_chain_accumulates_history() {
+    let db = setup_db();
+
+    let addr = B256::from([0xCD; 32]);
+    let hashes = [
+        B256::repeat_byte(0x00),
+        B256::repeat_byte(0x01),
+        B256::repeat_byte(0x02),
+        B256::repeat_byte(0x03),
+    ];
+
+    // earliest = (3, hash_3). Backfill blocks 3, 2, 1 in descending order.
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        provider.set_earliest_block_number(3, hashes[3]).expect("set earliest");
+        OpProofsProviderRw::commit(provider).expect("commit");
+    }
+
+    for block_num in (1u64..=3).rev() {
+        let provider = MdbxProofsProviderV2::new(db.tx_mut().expect("rw"));
+        let block_ref = make_block_ref(
+            block_num,
+            hashes[block_num as usize],
+            hashes[(block_num - 1) as usize],
+        );
+        provider
+            .prepend_block(block_ref, make_nonce_diff(addr, block_num))
+            .expect("prepend");
+        OpProofsBackfillProvider::commit(provider).expect("commit");
+    }
+
+    // earliest now at (0, hash_0).
+    {
+        let provider = MdbxProofsProviderV2::new(db.tx().expect("ro"));
+        assert_eq!(
+            provider.get_earliest_block_number().expect("get"),
+            Some((0, hashes[0]))
+        );
+    }
+
+    // History bitmap accumulated all three prepended blocks in ascending order.
+    let tx = db.tx().expect("ro");
+    let mut cur = tx.cursor_read::<V2HashedAccountsHistory>().expect("cursor");
+    let shard_key = HashedAccountShardedKey::new(addr, u64::MAX);
+    let (_, bitmap) = cur.seek_exact(shard_key).expect("seek").expect("exists");
+    assert_eq!(bitmap.iter().collect::<Vec<_>>(), vec![1, 2, 3]);
+}
diff --git a/rust/op-reth/crates/trie/src/error.rs b/rust/op-reth/crates/trie/src/error.rs
index 23b3c230492..aeff3e08046 100644
--- a/rust/op-reth/crates/trie/src/error.rs
+++ b/rust/op-reth/crates/trie/src/error.rs
@@ -99,6 +99,21 @@ pub enum OpProofsStorageError {
         /// The latest stored block number
         latest_block_number: u64,
     },
+    /// Attempted to prepend a block whose hash does not match the current earliest
+    #[error(
+        "Cannot prepend block {block_number} (hash {block_hash}): \
+         expected earliest block {earliest_block_number} (hash {earliest_block_hash})"
+    )]
+    PrependOutOfOrder {
+        /// The block number being prepended
+        block_number: u64,
+        /// The hash of the block being prepended
+        block_hash: B256,
+        /// The current earliest block number
+        earliest_block_number: u64,
+        /// The current earliest block hash
+        earliest_block_hash: B256,
+    },
     /// Attempted to prune to a block at or before the earliest stored block
     #[error(
         "Attempted to prune to block {target_block_number} but earliest stored block is already {earliest_block_number}"
diff --git a/rust/op-reth/crates/trie/src/in_memory.rs b/rust/op-reth/crates/trie/src/in_memory.rs
index 07f4e99516c..93022d3edd5 100644
--- a/rust/op-reth/crates/trie/src/in_memory.rs
+++ b/rust/op-reth/crates/trie/src/in_memory.rs
@@ -3,8 +3,8 @@
 use crate::{
     BlockStateDiff, OpProofsStorageError, OpProofsStorageResult, OpProofsStore,
     api::{
-        InitialStateAnchor, InitialStateStatus, OpProofsInitProvider, OpProofsProviderRO,
-        OpProofsProviderRw, WriteCounts,
+        InitialStateAnchor, InitialStateStatus, OpProofsBackfillProvider, OpProofsInitProvider,
+        OpProofsProviderRO, OpProofsProviderRw, WriteCounts,
     },
     db::{HashedStorageKey, StorageTrieKey},
 };
@@ -493,6 +493,7 @@ impl OpProofsStore for InMemoryProofsStorage {
     type ProviderRO<'a> = InMemoryProofsProvider;
     type ProviderRw<'a> = InMemoryProofsProvider;
     type Initializer<'a> = InMemoryProofsProvider;
+    type BackfillProvider<'a> = InMemoryProofsProvider;
 
     fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>> {
         Ok(InMemoryProofsProvider { inner: self.inner.clone() })
@@ -505,6 +506,10 @@ impl OpProofsStore for InMemoryProofsStorage {
     fn initialization_provider<'a>(&'a self) -> OpProofsStorageResult<Self::Initializer<'a>> {
         Ok(InMemoryProofsProvider { inner: self.inner.clone() })
     }
+
+    fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>> {
+        Ok(InMemoryProofsProvider { inner: self.inner.clone() })
+    }
 }
 
 impl InMemoryProofsProvider {
@@ -873,6 +878,23 @@ impl OpProofsInitProvider for InMemoryProofsProvider {
     }
 }
 
+impl OpProofsBackfillProvider for InMemoryProofsProvider {
+    fn prepend_block(
+        &self,
+        block_ref: BlockWithParent,
+        diff: BlockStateDiff,
+    ) -> OpProofsStorageResult<WriteCounts> {
+        let mut inner = self.inner.write();
+        let counts = inner.store_trie_updates(block_ref.block.number, diff);
+        inner.earliest_block = Some((block_ref.block.number - 1, block_ref.parent));
+        Ok(counts)
+    }
+
+    fn commit(self) -> OpProofsStorageResult<()> {
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/rust/op-reth/crates/trie/src/initialize.rs b/rust/op-reth/crates/trie/src/initialize.rs
index d0405f8873c..36c5ba5a034 100644
--- a/rust/op-reth/crates/trie/src/initialize.rs
+++ b/rust/op-reth/crates/trie/src/initialize.rs
@@ -1523,6 +1523,7 @@ mod tests {
         type ProviderRw<'a> = <MdbxProofsStorage as OpProofsStore>::ProviderRw<'a>;
         type Initializer<'a> =
             RecordingInitProvider<<MdbxProofsStorage as OpProofsStore>::Initializer<'a>>;
+        type BackfillProvider<'a> = <MdbxProofsStorage as OpProofsStore>::BackfillProvider<'a>;
 
         fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>> {
             self.inner.provider_ro()
@@ -1539,6 +1540,10 @@ mod tests {
                 storage_branch_addresses: self.storage_branch_addresses.clone(),
             })
         }
+
+        fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>> {
+            self.inner.backfill_provider()
+        }
     }
 
     // ── Regression tests: address order must be preserved ──────────────
diff --git a/rust/op-reth/crates/trie/src/lib.rs b/rust/op-reth/crates/trie/src/lib.rs
index 7ef1ea48fa4..972980ec3b7 100644
--- a/rust/op-reth/crates/trie/src/lib.rs
+++ b/rust/op-reth/crates/trie/src/lib.rs
@@ -18,12 +18,16 @@ use reth_ethereum_primitives as _;
 
 pub mod api;
 pub use api::{
-    BlockStateDiff, OpProofsInitProvider, OpProofsProviderRO, OpProofsProviderRw, OpProofsStore,
+    BlockStateDiff, OpProofsBackfillProvider, OpProofsInitProvider, OpProofsProviderRO,
+    OpProofsProviderRw, OpProofsStore,
 };
 
 pub mod initialize;
 pub use initialize::{InitializationJob, RethTrieStorageLayout};
 
+pub mod backfill;
+pub use backfill::{BackfillError, BackfillJob};
+
 pub mod in_memory;
 pub use in_memory::{
     InMemoryAccountCursor, InMemoryProofsStorage, InMemoryStorageCursor, InMemoryTrieCursor,
diff --git a/rust/op-reth/crates/trie/src/metrics.rs b/rust/op-reth/crates/trie/src/metrics.rs
index 27f71ba0d4d..cc8c63c4baa 100644
--- a/rust/op-reth/crates/trie/src/metrics.rs
+++ b/rust/op-reth/crates/trie/src/metrics.rs
@@ -336,6 +336,11 @@ where
     where
         Self: 'a;
 
+    type BackfillProvider<'a>
+        = S::BackfillProvider<'a>
+    where
+        Self: 'a;
+
     fn provider_ro<'a>(&'a self) -> OpProofsStorageResult<Self::ProviderRO<'a>> {
         Ok(OpProofsProviderROWithMetrics::new(self.storage.provider_ro()?, self.metrics.clone()))
     }
@@ -350,6 +355,10 @@ where
             self.metrics.clone(),
         ))
     }
+
+    fn backfill_provider<'a>(&'a self) -> OpProofsStorageResult<Self::BackfillProvider<'a>> {
+        self.storage.backfill_provider()
+    }
 }
 
 /// Wrapper for [`OpProofsProviderRO`] that records metrics.