diff --git a/Cargo.lock b/Cargo.lock
index bc31b047..bceec54e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11715,7 +11715,9 @@ dependencies = [
 name = "pallet-prover-db-indexer"
 version = "0.1.0"
 dependencies = [
+ "commitment-sql",
  "parity-scale-codec",
+ "parking_lot 0.12.3",
  "polkadot-sdk 0.9.0",
  "prost 0.13.5",
  "prost-build 0.13.5",
diff --git a/pallets/prover_db_indexer/Cargo.toml b/pallets/prover_db_indexer/Cargo.toml
index 64fcff73..1a899b76 100644
--- a/pallets/prover_db_indexer/Cargo.toml
+++ b/pallets/prover_db_indexer/Cargo.toml
@@ -21,6 +21,7 @@ polkadot-sdk = { workspace = true, features = [
 ]}
 
 sxt-core.workspace = true
+commitment-sql.workspace = true
 prost = { version = "0.13", default-features = false, features = ["prost-derive"] }
 snafu = { workspace = true }
 url = { workspace = true }
@@ -28,6 +29,12 @@ url = { workspace = true }
 [build-dependencies]
 prost-build = "0.13"
 
+[dev-dependencies]
+# Only direct test deps. `polkadot-sdk` already comes through `[dependencies]`
+# with `std` enabled in test builds, so we don't re-declare it here, and the
+# mock+tests don't reference any of pallet-tables / pallet-indexing / etc.
+parking_lot = "0.12"
+
 [features]
 default = ["std"]
 std = [
@@ -35,6 +42,7 @@ std = [
     "codec/std",
     "scale-info/std",
     "sxt-core/std",
+    "commitment-sql/std",
     "prost/std",
     "snafu/std",
     "url/std",
diff --git a/pallets/prover_db_indexer/src/lib.rs b/pallets/prover_db_indexer/src/lib.rs
index 462dce45..ae70b3b9 100644
--- a/pallets/prover_db_indexer/src/lib.rs
+++ b/pallets/prover_db_indexer/src/lib.rs
@@ -18,11 +18,13 @@
 //!   captures.
 //!
 //! **Consumer** (`offchain_worker`, fires at chain tip only):
-//!   For each block in `cursor+1..=current`, reads the high-water-mark.
-//!   If absent, the block had zero captures and is skipped. Otherwise
-//!   probes `key_for_event(block, 0..=hwm)`, forwards each present
-//!   entry via HTTP+protobuf, checkpoints on the server, deletes
-//!   consumed entries, advances cursor.
+//!   Asks the indexer server for its last checkpoint sequence number,
+//!   walks blocks `cursor+1..=current` (capped at
+//!   [`MAX_BLOCKS_PER_INVOCATION`]). For each block, reads the
+//!   high-water-mark; if absent, the block had zero captures and is
+//!   skipped. Otherwise probes `key_for_event(block, 0..=hwm)`,
+//!   forwards each present payload via HTTP+protobuf, checkpoints on
+//!   the server, and deletes consumed entries from the offchain DB.
 //!
 //! ## Why extrinsic-time capture, not `on_finalize`?
 //!
@@ -41,11 +43,14 @@
 
 extern crate alloc;
 
-// Wire-level HTTP+protobuf client for the indexer service. Unused until
-// the OCW consumer (next PR) calls into it; allow dead_code so the build
-// stays clean in the meantime.
-#[allow(dead_code)]
+#[cfg(test)]
+mod mock;
+
+#[cfg(test)]
+mod tests;
+
 mod http_client;
+mod offchain_consumer;
 
 /// Generated protobuf types for the indexer HTTP adapter wire format.
 #[allow(missing_docs, clippy::missing_docs_in_private_items)]
@@ -59,6 +64,59 @@ pub use pallet::*;
 /// keep working.
 pub use sxt_core::prover_db_indexer::PROVER_DB_URL_KEY;
 
+/// Offchain DB key for the lock that serializes OCW consumer rounds.
+const OCW_LOCK_KEY: &[u8] = b"prover_db_indexer/ocw_lock";
+
+/// Typed errors from the OCW consumer round. Each variant carries the
+/// underlying `http_client::Error` (when applicable) so the operator's
+/// log line names both the failing operation and the wire-level reason.
+///
+/// `source` fields aren't named `source` because `url::ParseError`
+/// doesn't implement `core::error::Error` in this crate's no_std build,
+/// which would break Snafu's source-chain wiring. `http_client::Error`
+/// could use the wired name, but we keep the convention uniform so
+/// every variant looks the same.
+#[derive(Debug, snafu::Snafu)]
+pub enum ConsumerError {
+    /// The configured indexer URL was not valid UTF-8.
+    #[snafu(display("indexer URL was not valid UTF-8"))]
+    InvalidUrlEncoding,
+    /// The configured indexer URL couldn't be parsed.
+    #[snafu(display("indexer URL is not a valid URL: {error}"))]
+    InvalidUrl {
+        /// Underlying parse error from the `url` crate.
+        error: url::ParseError,
+    },
+    /// The runtime's `BlockNumber` didn't fit in `u64` (defensive — in
+    /// practice `BlockNumber` is `u32` so this is unreachable).
+    #[snafu(display("block number does not fit in u64"))]
+    BlockNumberOverflow,
+    /// `create_table` HTTP call failed.
+    #[snafu(display("create_table failed: {error}"))]
+    CreateTable {
+        /// Underlying error from the HTTP client.
+        error: http_client::Error,
+    },
+    /// `drop_table` HTTP call failed.
+    #[snafu(display("drop_table failed: {error}"))]
+    DropTable {
+        /// Underlying error from the HTTP client.
+        error: http_client::Error,
+    },
+    /// `put_batches` HTTP call failed.
+    #[snafu(display("put_batches failed: {error}"))]
+    PutBatches {
+        /// Underlying error from the HTTP client.
+        error: http_client::Error,
+    },
+    /// `checkpoint` HTTP call failed.
+    #[snafu(display("checkpoint failed: {error}"))]
+    Checkpoint {
+        /// Underlying error from the HTTP client.
+        error: http_client::Error,
+    },
+}
+
 #[polkadot_sdk::frame_support::pallet]
 #[allow(
     missing_docs,
@@ -71,6 +129,10 @@ pub mod pallet {
 
     use codec::Encode;
     use polkadot_sdk::frame_support::pallet_prelude::*;
+    use polkadot_sdk::frame_system::pallet_prelude::*;
+    use polkadot_sdk::sp_runtime::offchain::storage::StorageValueRef;
+    use polkadot_sdk::sp_runtime::offchain::storage_lock::{StorageLock, Time};
+    use polkadot_sdk::sp_runtime::offchain::Duration;
     use sxt_core::prover_db_indexer::{
         key_for_event,
         key_for_high_water,
@@ -78,6 +140,8 @@ pub mod pallet {
         EventCapture,
     };
 
+    use crate::ConsumerError;
+
     #[pallet::pallet]
     pub struct Pallet<T>(_);
 
@@ -86,6 +150,22 @@ pub mod pallet {
         /// The runtime's overarching event type.
         type RuntimeEvent: From<Event<Self>>
             + IsType<<Self as polkadot_sdk::frame_system::Config>::RuntimeEvent>;
+
+        /// Maximum number of blocks the OCW will walk in a single
+        /// invocation. Controls catch-up speed: with 6s block time and
+        /// the default of 100, catch-up is ~100x realtime. Lower values
+        /// keep the per-round cost bounded; higher values close a wider
+        /// gap faster after downtime.
+        #[pallet::constant]
+        type MaxBlocksPerInvocation: Get<u64>;
+
+        /// How long (in milliseconds) a held OCW lock stays valid before
+        /// being treated as abandoned (e.g. node crashed mid-round). Must
+        /// comfortably cover one full consumer round under normal
+        /// conditions while still letting the chain recover quickly from
+        /// a crashed validator.
+        #[pallet::constant]
+        type OcwLockDeadlineMs: Get<u64>;
     }
 
     #[pallet::event]
@@ -103,6 +183,21 @@ pub mod pallet {
         },
     }
 
+    #[pallet::hooks]
+    impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
+        // Fires at chain tip only (not during sync). Drains the offchain
+        // DB queue, forwards to the HTTP server, deletes consumed entries.
+        fn offchain_worker(block_number: BlockNumberFor<T>) {
+            if let Err(e) = Self::run_consumer(block_number) {
+                polkadot_sdk::sp_tracing::error!(
+                    target: "prover_db_indexer",
+                    "offchain indexer error: {}",
+                    e,
+                );
+            }
+        }
+    }
+
     impl<T: Config> EventCapture for Pallet<T> {
         fn capture_events(events: Vec<BlockEvent<'_>>) {
             if events.is_empty() {
@@ -130,4 +225,167 @@ pub mod pallet {
             polkadot_sdk::sp_io::offchain_index::set(&key_for_high_water(block), &ext_idx.encode());
         }
     }
+
+    impl<T: Config> Pallet<T> {
+        // ═══════════════════════════════════════════════════════════════
+        //  CONSUMER: drain offchain DB → HTTP → delete (called from OCW)
+        // ═══════════════════════════════════════════════════════════════
+
+        fn run_consumer(block_number: BlockNumberFor<T>) -> Result<(), ConsumerError> {
+            // Serialize concurrent OCW invocations. Substrate spawns
+            // `offchain_worker` for every imported block, and rounds can
+            // overlap if one runs longer than block time. Without a lock,
+            // both rounds would read the same server checkpoint and
+            // submit duplicate `put_batches`/`create_table`/`drop_table`
+            // calls; the second one's `checkpoint()` would then fail with
+            // `failed_precondition`. Holding this lock for the full round
+            // makes overlapping invocations no-ops.
+            let mut lock = StorageLock::<Time>::with_deadline(
+                crate::OCW_LOCK_KEY,
+                Duration::from_millis(T::OcwLockDeadlineMs::get()),
+            );
+            let _guard = match lock.try_lock() {
+                Ok(g) => g,
+                Err(_deadline) => {
+                    polkadot_sdk::sp_tracing::debug!(
+                        target: "prover_db_indexer",
+                        "another OCW round is already in progress; skipping",
+                    );
+                    return Ok(());
+                }
+            };
+
+            // 1. Check if this node is configured as an indexer.
+            let url_ref = StorageValueRef::persistent(crate::PROVER_DB_URL_KEY);
+            let url_bytes: Option<Vec<u8>> = url_ref.get::<Vec<u8>>().ok().flatten();
+
+            let Some(url_bytes) = url_bytes else {
+                return Ok(());
+            };
+
+            let url_str =
+                core::str::from_utf8(&url_bytes).map_err(|_| ConsumerError::InvalidUrlEncoding)?;
+            let url =
+                url::Url::parse(url_str).map_err(|error| ConsumerError::InvalidUrl { error })?;
+
+            polkadot_sdk::sp_tracing::debug!(
+                target: "prover_db_indexer",
+                "consumer round starting; indexer base URL = {}",
+                url,
+            );
+
+            // 2. Current block number — passed in by `offchain_worker`.
+            let current_block: u64 = block_number
+                .try_into()
+                .map_err(|_| ConsumerError::BlockNumberOverflow)?;
+
+            // 3. Ask the server for its last checkpoint. The server is the
+            //    sole source of truth — no local cursor. This costs one HTTP
+            //    round-trip per OCW invocation but guarantees we never
+            //    diverge from what the server has actually committed.
+            //    A failure here is treated as transient: log and skip this
+            //    round; the next OCW will retry.
+            let cursor: u64 = match crate::http_client::get_last_checkpoint(&url) {
+                Ok(Some(seq)) => seq,
+                Ok(None) => 0,
+                Err(e) => {
+                    polkadot_sdk::sp_tracing::warn!(
+                        target: "prover_db_indexer",
+                        "get_last_checkpoint failed for {}: {}; skipping this round",
+                        url, e,
+                    );
+                    return Ok(());
+                }
+            };
+
+            // 4. Walk blocks from cursor+1 to tip, capped.
+            let start = cursor + 1;
+            let end = core::cmp::min(current_block, start + T::MaxBlocksPerInvocation::get() - 1);
+
+            if start > current_block {
+                return Ok(());
+            }
+
+            polkadot_sdk::sp_tracing::debug!(
+                target: "prover_db_indexer",
+                "processing blocks {}..={} (server_checkpoint={}, tip={})",
+                start, end, cursor, current_block,
+            );
+
+            for block_num in start..=end {
+                Self::forward_block(&url, block_num)?;
+
+                // Checkpoint on the server (always, even for empty blocks).
+                crate::http_client::checkpoint(&url, block_num)
+                    .map_err(|error| ConsumerError::Checkpoint { error })?;
+            }
+
+            Ok(())
+        }
+
+        /// Forward a single block's events in extrinsic-index order, then
+        /// clear the offchain entries we consumed. If the block had no
+        /// captures (no high-water-mark key), this is a no-op.
+        fn forward_block(url: &url::Url, block_num: u64) -> Result<(), ConsumerError> {
+            let Some(hwm) = crate::offchain_consumer::read_high_water(block_num) else {
+                return Ok(());
+            };
+
+            polkadot_sdk::sp_tracing::info!(
+                target: "prover_db_indexer",
+                "block {} — high-water-mark {}; probing for captured events",
+                block_num,
+                hwm,
+            );
+
+            for ext_idx in 0..=hwm {
+                let Some(events) = crate::offchain_consumer::read_events(block_num, ext_idx) else {
+                    continue;
+                };
+                Self::forward_events(url, block_num, &events)?;
+                crate::offchain_consumer::clear_events(block_num, ext_idx);
+            }
+
+            crate::offchain_consumer::clear_high_water(block_num);
+            Ok(())
+        }
+
+        /// POST one extrinsic's captured events to the indexer in deposit order.
+        fn forward_events(
+            url: &url::Url,
+            block_num: u64,
+            events: &[BlockEvent<'_>],
+        ) -> Result<(), ConsumerError> {
+            let seq = block_num;
+
+            for event in events {
+                match event {
+                    BlockEvent::Drop(ident) => {
+                        crate::http_client::drop_table(url, seq, ident)
+                            .map_err(|error| ConsumerError::DropTable { error })?;
+                    }
+                    BlockEvent::Create(entry) => {
+                        crate::http_client::create_table(
+                            url,
+                            seq,
+                            &entry.ident,
+                            entry.ddl.to_vec(),
+                            commitment_sql::ROW_NUMBER_COLUMN_NAME.into(),
+                        )
+                        .map_err(|error| ConsumerError::CreateTable { error })?;
+                    }
+                    BlockEvent::Insert(entry) => {
+                        crate::http_client::put_batches(
+                            url,
+                            seq,
+                            alloc::vec![(entry.table.as_ref(), entry.data.to_vec())],
+                        )
+                        .map_err(|error| ConsumerError::PutBatches { error })?;
+                    }
+                }
+            }
+
+            Ok(())
+        }
+    }
 }
diff --git a/pallets/prover_db_indexer/src/mock.rs b/pallets/prover_db_indexer/src/mock.rs
new file mode 100644
index 00000000..f4f17dd9
--- /dev/null
+++ b/pallets/prover_db_indexer/src/mock.rs
@@ -0,0 +1,50 @@
+//! Minimal mock runtime for testing the consumer (OCW HTTP forwarding).
+//!
+//! The pallet's `Config` only requires `frame_system::Config`, so the
+//! mock is intentionally tiny — frame-system + the pallet itself. Tests
+//! pre-populate the offchain DB with payload+high-water keys directly,
+//! then drive `offchain_worker` and assert on outbound HTTP traffic.
+
+use polkadot_sdk::frame_support::traits::ConstU64;
+use polkadot_sdk::frame_support::{construct_runtime, derive_impl};
+use polkadot_sdk::sp_core::H256;
+use polkadot_sdk::sp_runtime::traits::IdentityLookup;
+use polkadot_sdk::sp_runtime::BuildStorage;
+use polkadot_sdk::{frame_system, sp_io, sp_runtime};
+
+use crate as pallet_prover_db_indexer;
+
+type Block = frame_system::mocking::MockBlock<Test>;
+
+construct_runtime!(
+    pub enum Test {
+        System: frame_system,
+        ProverDbIndexer: pallet_prover_db_indexer,
+    }
+);
+
+type AccountId = sp_runtime::AccountId32;
+type Nonce = u32;
+
+#[derive_impl(frame_system::config_preludes::TestDefaultConfig)]
+impl frame_system::Config for Test {
+    type Nonce = Nonce;
+    type AccountId = AccountId;
+    type Block = Block;
+    type Lookup = IdentityLookup<Self::AccountId>;
+    type Hash = H256;
+}
+
+impl pallet_prover_db_indexer::Config for Test {
+    type RuntimeEvent = RuntimeEvent;
+    // Match runtime defaults so tests exercise the production limits.
+    type MaxBlocksPerInvocation = ConstU64<100>;
+    type OcwLockDeadlineMs = ConstU64<120_000>;
+}
+
+pub fn new_test_ext() -> sp_io::TestExternalities {
+    frame_system::GenesisConfig::<Test>::default()
+        .build_storage()
+        .unwrap()
+        .into()
+}
diff --git a/pallets/prover_db_indexer/src/offchain_consumer.rs b/pallets/prover_db_indexer/src/offchain_consumer.rs
new file mode 100644
index 00000000..77eae0a6
--- /dev/null
+++ b/pallets/prover_db_indexer/src/offchain_consumer.rs
@@ -0,0 +1,49 @@
+//! OCW-only helpers for reading and clearing the offchain DB entries
+//! the producer wrote. Producer-side helpers (key construction, types)
+//! live in `sxt_core::prover_db_indexer`; this module wraps the OCW host
+//! functions (`local_storage_get` / `local_storage_clear`) so the
+//! consumer code in `lib.rs` doesn't repeat the boilerplate.
+
+use alloc::vec::Vec;
+
+use codec::Decode;
+use polkadot_sdk::sp_core::offchain::StorageKind;
+use sxt_core::prover_db_indexer::{key_for_event, key_for_high_water, BlockEvent};
+
+/// Read the per-block high-water-mark. `None` means the block had no
+/// captured events.
+pub fn read_high_water(block: u64) -> Option<u32> {
+    let raw = polkadot_sdk::sp_io::offchain::local_storage_get(
+        StorageKind::PERSISTENT,
+        &key_for_high_water(block),
+    )?;
+    u32::decode(&mut &raw[..]).ok()
+}
+
+/// Read the events emitted by a single extrinsic in a given block.
+/// `None` means that extrinsic did not call `EventCapture::capture_events`.
+/// The returned `BlockEvent<'static>` decodes into `Cow::Owned` for every
+/// field, so the lifetime is purely a type-level annotation.
+pub fn read_events(block: u64, extrinsic_index: u32) -> Option<Vec<BlockEvent<'static>>> {
+    let raw = polkadot_sdk::sp_io::offchain::local_storage_get(
+        StorageKind::PERSISTENT,
+        &key_for_event(block, extrinsic_index),
+    )?;
+    Vec::<BlockEvent<'static>>::decode(&mut &raw[..]).ok()
+}
+
+/// Delete the per-block high-water-mark.
+pub fn clear_high_water(block: u64) {
+    polkadot_sdk::sp_io::offchain::local_storage_clear(
+        StorageKind::PERSISTENT,
+        &key_for_high_water(block),
+    );
+}
+
+/// Delete the per-extrinsic event payload.
+pub fn clear_events(block: u64, extrinsic_index: u32) {
+    polkadot_sdk::sp_io::offchain::local_storage_clear(
+        StorageKind::PERSISTENT,
+        &key_for_event(block, extrinsic_index),
+    );
+}
diff --git a/pallets/prover_db_indexer/src/tests.rs b/pallets/prover_db_indexer/src/tests.rs
new file mode 100644
index 00000000..05a8fd3f
--- /dev/null
+++ b/pallets/prover_db_indexer/src/tests.rs
@@ -0,0 +1,397 @@
+//! Integration tests for the prover-db-indexer consumer (OCW).
+//!
+//! Tests pre-populate the offchain DB as if `EventCapture::capture_events`
+//! had run during block execution — that means writing the per-extrinsic
+//! event payload at `key_for_event(block, ext_idx)` and the per-block
+//! high-water-mark at `key_for_high_water(block)` — then drive
+//! `offchain_worker` and verify the OCW reads, forwards in order, and
+//! deletes consumed entries.
+
+use std::borrow::Cow;
+
+use codec::Encode;
+use polkadot_sdk::frame_support::traits::Hooks;
+use polkadot_sdk::sp_core::offchain::testing::{PendingRequest, TestOffchainExt};
+use polkadot_sdk::sp_core::offchain::{OffchainDbExt, OffchainStorage, OffchainWorkerExt};
+use polkadot_sdk::sp_runtime::offchain::storage_lock::{StorageLock, Time};
+use polkadot_sdk::sp_runtime::offchain::Duration;
+use prost::Message;
+use sxt_core::prover_db_indexer::{
+    key_for_event,
+    key_for_high_water,
+    BlockEvent,
+    CreateEntry,
+    InsertEntry,
+};
+use sxt_core::tables::TableIdentifier;
+
+use crate::mock::*;
+use crate::{proto, PROVER_DB_URL_KEY};
+
+type StateArc =
+    std::sync::Arc<parking_lot::RwLock<polkadot_sdk::sp_core::offchain::testing::OffchainState>>;
+
+const MOCK_URL: &str = "http://127.0.0.1:9999";
+
+fn encode_url() -> Vec<u8> {
+    codec::Encode::encode(&MOCK_URL.as_bytes().to_vec())
+}
+
+fn expected_request(path: &str, body: Vec<u8>, response: Vec<u8>) -> PendingRequest {
+    PendingRequest {
+        method: "POST".into(),
+        uri: format!("{}{}", MOCK_URL, path),
+        headers: vec![("Content-Type".into(), "application/x-protobuf".into())],
+        body,
+        sent: true,
+        response: Some(response),
+        ..Default::default()
+    }
+}
+
+fn checkpoint_at_response(seq: u64) -> Vec<u8> {
+    proto::GetLastCheckpointResponse {
+        sequence_number: seq,
+        has_checkpoint: true,
+    }
+    .encode_to_vec()
+}
+
+fn no_checkpoint_response() -> Vec<u8> {
+    proto::GetLastCheckpointResponse {
+        sequence_number: 0,
+        has_checkpoint: false,
+    }
+    .encode_to_vec()
+}
+
+fn setup_with_url() -> (polkadot_sdk::sp_io::TestExternalities, StateArc) {
+    let mut ext = new_test_ext();
+    let (offchain, state) = TestOffchainExt::new();
+    ext.register_extension(OffchainWorkerExt::new(offchain.clone()));
+    ext.register_extension(OffchainDbExt::new(offchain));
+    state
+        .write()
+        .persistent_storage
+        .set(b"", PROVER_DB_URL_KEY, &encode_url());
+    (ext, state)
+}
+
+/// Mirror what the producer would write for a block: one event payload
+/// at `(block, ext_idx)` and the matching high-water-mark.
+fn seed_block_events(state: &StateArc, block: u64, ext_idx: u32, events: Vec<BlockEvent<'static>>) {
+    let mut s = state.write();
+    s.persistent_storage
+        .set(b"", &key_for_event(block, ext_idx), &events.encode());
+    s.persistent_storage
+        .set(b"", &key_for_high_water(block), &Encode::encode(&ext_idx));
+}
+
+// ─── Tests ──────────────────────────────────────────────────────────────
+
+#[test]
+fn ocw_skips_when_not_configured() {
+    let mut ext = new_test_ext();
+    let (offchain, _) = TestOffchainExt::new();
+    ext.register_extension(OffchainWorkerExt::new(offchain.clone()));
+    ext.register_extension(OffchainDbExt::new(offchain));
+    ext.execute_with(|| {
+        System::set_block_number(1);
+        ProverDbIndexer::offchain_worker(1);
+    });
+}
+
+/// If another OCW round is in progress (lock held), this invocation
+/// must do nothing — no HTTP traffic, no state reads beyond the lock
+/// itself. `TestOffchainExt` would panic on an unexpected request, so
+/// the absence of queued expectations is the assertion.
+#[test]
+fn ocw_skips_when_lock_is_held() {
+    let (mut ext, _state) = setup_with_url();
+
+    ext.execute_with(|| {
+        let mut lock = StorageLock::<Time>::with_deadline(
+            b"prover_db_indexer/ocw_lock",
+            Duration::from_millis(120_000),
+        );
+        let _guard = lock.try_lock().expect("first acquisition must succeed");
+
+        System::set_block_number(1);
+        ProverDbIndexer::offchain_worker(1);
+    });
+}
+
+#[test]
+fn ocw_forwards_and_deletes_offchain_entry() {
+    let (mut ext, state) = setup_with_url();
+
+    let ddl = b"CREATE TABLE PUBLIC.USERS (ID BIGINT NOT NULL)";
+    seed_block_events(
+        &state,
+        1,
+        0,
+        vec![BlockEvent::Create(CreateEntry {
+            ident: Cow::Owned(TableIdentifier::from_str_unchecked("USERS", "PUBLIC")),
+            ddl: ddl.to_vec().into(),
+        })],
+    );
+
+    {
+        let mut s = state.write();
+        s.expect_request(expected_request(
+            "/v1/get_last_checkpoint",
+            vec![],
+            no_checkpoint_response(),
+        ));
+        s.expect_request(expected_request(
+            "/v1/create_table",
+            proto::CreateTableRequest {
+                sequence_number: 1,
+                table_name: "PUBLIC.USERS".into(),
+                arrow_schema: ddl.to_vec(),
+                key: "META_ROW_NUMBER".into(),
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(),
+            vec![],
+        ));
+    }
+
+    ext.execute_with(|| {
+        System::set_block_number(1);
+        ProverDbIndexer::offchain_worker(1);
+    });
+
+    let s = state.read();
+    assert!(
+        s.persistent_storage.get(&key_for_event(1, 0)).is_none(),
+        "consumed event payload should be deleted"
+    );
+    assert!(
+        s.persistent_storage.get(&key_for_high_water(1)).is_none(),
+        "high-water-mark should be deleted"
+    );
+}
+
+#[test]
+fn ocw_checkpoints_empty_blocks() {
+    let (mut ext, state) = setup_with_url();
+
+    {
+        let mut s = state.write();
+        s.expect_request(expected_request(
+            "/v1/get_last_checkpoint",
+            vec![],
+            no_checkpoint_response(),
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(),
+            vec![],
+        ));
+    }
+
+    ext.execute_with(|| {
+        System::set_block_number(1);
+        ProverDbIndexer::offchain_worker(1);
+    });
+}
+
+#[test]
+fn ocw_resumes_from_server_checkpoint() {
+    let (mut ext, state) = setup_with_url();
+
+    seed_block_events(
+        &state,
+        6,
+        0,
+        vec![BlockEvent::Drop(Cow::Owned(
+            TableIdentifier::from_str_unchecked("OLD", "NS"),
+        ))],
+    );
+
+    {
+        let mut s = state.write();
+        s.expect_request(expected_request(
+            "/v1/get_last_checkpoint",
+            vec![],
+            checkpoint_at_response(5),
+        ));
+        s.expect_request(expected_request(
+            "/v1/drop_table",
+            proto::DropTableRequest {
+                sequence_number: 6,
+                table_name: "NS.OLD".into(),
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 6 }.encode_to_vec(),
+            vec![],
+        ));
+    }
+
+    ext.execute_with(|| {
+        System::set_block_number(6);
+        ProverDbIndexer::offchain_worker(6);
+    });
+}
+
+#[test]
+fn ocw_processes_multiple_blocks_in_order() {
+    let (mut ext, state) = setup_with_url();
+
+    seed_block_events(
+        &state,
+        1,
+        0,
+        vec![BlockEvent::Drop(Cow::Owned(
+            TableIdentifier::from_str_unchecked("T1", "NS"),
+        ))],
+    );
+    seed_block_events(
+        &state,
+        2,
+        0,
+        vec![BlockEvent::Create(CreateEntry {
+            ident: Cow::Owned(TableIdentifier::from_str_unchecked("T2", "NS")),
+            ddl: b"CREATE TABLE NS.T2 (X INT NOT NULL)".to_vec().into(),
+        })],
+    );
+
+    {
+        let mut s = state.write();
+        s.expect_request(expected_request(
+            "/v1/get_last_checkpoint",
+            vec![],
+            no_checkpoint_response(),
+        ));
+        s.expect_request(expected_request(
+            "/v1/drop_table",
+            proto::DropTableRequest {
+                sequence_number: 1,
+                table_name: "NS.T1".into(),
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/create_table",
+            proto::CreateTableRequest {
+                sequence_number: 2,
+                table_name: "NS.T2".into(),
+                arrow_schema: b"CREATE TABLE NS.T2 (X INT NOT NULL)".to_vec(),
+                key: "META_ROW_NUMBER".into(),
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 2 }.encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 3 }.encode_to_vec(),
+            vec![],
+        ));
+    }
+
+    ext.execute_with(|| {
+        System::set_block_number(3);
+        ProverDbIndexer::offchain_worker(3);
+    });
+
+    let s = state.read();
+    assert!(s.persistent_storage.get(&key_for_event(1, 0)).is_none());
+    assert!(s.persistent_storage.get(&key_for_high_water(1)).is_none());
+    assert!(s.persistent_storage.get(&key_for_event(2, 0)).is_none());
+    assert!(s.persistent_storage.get(&key_for_high_water(2)).is_none());
+}
+
+#[test]
+fn ocw_walks_multiple_extrinsics_in_one_block() {
+    let (mut ext, state) = setup_with_url();
+
+    // Two extrinsics in block 1 fire captures: ext 1 and ext 3. Ext 2
+    // didn't (a sparse block). hwm should be 3; the OCW probes 0..=3
+    // and finds payloads at 1 and 3.
+    let mut s = state.write();
+    s.persistent_storage.set(
+        b"",
+        &key_for_event(1, 1),
+        &vec![BlockEvent::Drop(Cow::Owned(
+            TableIdentifier::from_str_unchecked("T1", "NS"),
+        ))]
+        .encode(),
+    );
+    s.persistent_storage.set(
+        b"",
+        &key_for_event(1, 3),
+        &vec![BlockEvent::Insert(InsertEntry {
+            table: Cow::Owned(TableIdentifier::from_str_unchecked("T2", "NS")),
+            data: b"row-data".to_vec().into(),
+        })]
+        .encode(),
+    );
+    s.persistent_storage
+        .set(b"", &key_for_high_water(1), &Encode::encode(&3u32));
+    drop(s);
+
+    {
+        let mut s = state.write();
+        s.expect_request(expected_request(
+            "/v1/get_last_checkpoint",
+            vec![],
+            no_checkpoint_response(),
+        ));
+        s.expect_request(expected_request(
+            "/v1/drop_table",
+            proto::DropTableRequest {
+                sequence_number: 1,
+                table_name: "NS.T1".into(),
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/put_batches",
+            proto::PutBatchesRequest {
+                sequence_number: 1,
+                batches: vec![proto::TableBatch {
+                    table_name: "NS.T2".into(),
+                    record_batch: b"row-data".to_vec(),
+                }],
+            }
+            .encode_to_vec(),
+            vec![],
+        ));
+        s.expect_request(expected_request(
+            "/v1/checkpoint",
+            proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(),
+            vec![],
+        ));
+    }
+
+    ext.execute_with(|| {
+        System::set_block_number(1);
+        ProverDbIndexer::offchain_worker(1);
+    });
+
+    let s = state.read();
+    assert!(s.persistent_storage.get(&key_for_event(1, 1)).is_none());
+    assert!(s.persistent_storage.get(&key_for_event(1, 3)).is_none());
+    assert!(s.persistent_storage.get(&key_for_high_water(1)).is_none());
+}
diff --git a/proof-of-sql/commitment-sql/src/lib.rs b/proof-of-sql/commitment-sql/src/lib.rs
index a7b7e1f8..5874c48b 100644
--- a/proof-of-sql/commitment-sql/src/lib.rs
+++ b/proof-of-sql/commitment-sql/src/lib.rs
@@ -21,7 +21,7 @@ pub use column_type_conversion::{
 
 /// Row number column definition utilities.
 mod row_number_column;
-pub use row_number_column::row_number_column_def;
+pub use row_number_column::{row_number_column_def, ROW_NUMBER_COLUMN_NAME};
 
 /// Create table statement validation.
 mod validated_create_table;
diff --git a/proof-of-sql/commitment-sql/src/row_number_column.rs b/proof-of-sql/commitment-sql/src/row_number_column.rs
index 29491ad6..f59e85bf 100644
--- a/proof-of-sql/commitment-sql/src/row_number_column.rs
+++ b/proof-of-sql/commitment-sql/src/row_number_column.rs
@@ -6,8 +6,10 @@ use snafu::Snafu;
 use sqlparser::ast::helpers::stmt_create_table::CreateTableBuilder;
 use sqlparser::ast::{ColumnDef, ColumnOption, ColumnOptionDef, DataType, Ident, TableConstraint};
 
-/// Row number column name.
-const ROW_NUMBER_COLUMN_NAME: &str = "META_ROW_NUMBER";
+/// Row number column name. Public because the prover-db indexer pallet
+/// passes the same string as the dedup `key` on every `CreateTable` request
+/// to the indexer service.
+pub const ROW_NUMBER_COLUMN_NAME: &str = "META_ROW_NUMBER";
 
 /// Returns a sqlparser `ColumnDef` for the row number column.
 pub fn row_number_column_def() -> ColumnDef {
diff --git a/runtime/src/lib.rs b/runtime/src/lib.rs
index 884258a2..af3c5f65 100644
--- a/runtime/src/lib.rs
+++ b/runtime/src/lib.rs
@@ -896,6 +896,8 @@ impl pallet_rewards::Config for Runtime {
 
 impl pallet_prover_db_indexer::Config for Runtime {
     type RuntimeEvent = RuntimeEvent;
+    type MaxBlocksPerInvocation = ConstU64<100>;
+    type OcwLockDeadlineMs = ConstU64<120_000>;
 }
 
 #[cfg(feature = "runtime-benchmarks")]