diff --git a/src/eth/follower/importer/importer_supervisor.rs b/src/eth/follower/importer/importer_supervisor.rs
index 44046529e..b680dc8a7 100644
--- a/src/eth/follower/importer/importer_supervisor.rs
+++ b/src/eth/follower/importer/importer_supervisor.rs
@@ -141,10 +141,11 @@ pub async fn start_importer(
                 .run(resume_from, sync_interval, chain, stop_at_block)
                 .await?;
         }
-        ImporterMode::FakeLeader =>
+        ImporterMode::FakeLeader => {
             FakeLeader::new(executor, miner, storage, Arc::clone(&chain))
                 .run(resume_from, sync_interval, chain, stop_at_block)
-                .await?,
+                .await?;
+        }
     }
     Ok(())
 }
diff --git a/src/eth/rpc/blockchain_client/blockchain_client.rs b/src/eth/rpc/blockchain_client/blockchain_client.rs
index 9f4aa6854..5074973ae 100644
--- a/src/eth/rpc/blockchain_client/blockchain_client.rs
+++ b/src/eth/rpc/blockchain_client/blockchain_client.rs
@@ -13,6 +13,7 @@ use jsonrpsee::ws_client::WsClientBuilder;
 use tokio::sync::RwLock;
 use tokio::sync::RwLockReadGuard;
 
+use super::importer_pagination::ImporterPaginationClient;
 use crate::GlobalState;
 use crate::alias::AlloyBytes;
 use crate::alias::AlloyTransaction;
@@ -36,7 +37,7 @@ use crate::log_and_err;
 
 #[derive(Debug)]
 pub struct BlockchainClient {
-    http: HttpClient,
+    pub(super) http: HttpClient,
     pub http_url: String,
     ws: Option<RwLock<WsClient>>,
     ws_url: Option<String>,
@@ -159,33 +160,13 @@ impl BlockchainClient {
     /// Fetches a block by number with receipts.
     pub async fn fetch_block_and_receipts(&self, block_number: BlockNumber) -> anyhow::Result<Option<ExternalBlockWithReceipts>> {
         tracing::debug!(%block_number, "fetching block");
-
-        let number = to_json_value(block_number);
-        let result = self
-            .http
-            .request::<Option<ExternalBlockWithReceipts>, _>("stratus_getBlockAndReceipts", [number])
-            .await;
-
-        match result {
-            Ok(block) => Ok(block),
-            Err(e) => log_and_err!(reason = e, "failed to fetch block with receipts"),
-        }
+        ImporterPaginationClient::new(self).fetch_block_and_receipts(block_number).await
     }
 
     /// Fetches a block by number with changes.
     pub async fn fetch_block_with_changes(&self, block_number: BlockNumber) -> anyhow::Result<Option<(BlockRocksdb, BlockChangesRocksdb)>> {
         tracing::debug!(%block_number, "fetching block with changes");
-
-        let number = to_json_value(block_number);
-        let result = self
-            .http
-            .request::<Option<(BlockRocksdb, BlockChangesRocksdb)>, _>("stratus_getBlockWithChanges", [number])
-            .await;
-
-        match result {
-            Ok(block) => Ok(block),
-            Err(e) => log_and_err!(reason = e, "failed to fetch block with changes"),
-        }
+        ImporterPaginationClient::new(self).fetch_block_with_changes(block_number).await
     }
 
     /// Fetches a block by number.
diff --git a/src/eth/rpc/blockchain_client/importer_pagination.rs b/src/eth/rpc/blockchain_client/importer_pagination.rs
new file mode 100644
index 000000000..1dc737a1f
--- /dev/null
+++ b/src/eth/rpc/blockchain_client/importer_pagination.rs
@@ -0,0 +1,615 @@
+use anyhow::bail;
+use jsonrpsee::core::client::ClientT;
+use serde::de::DeserializeOwned;
+
+use super::blockchain_client::BlockchainClient;
+use crate::eth::rpc::types::BlockAndReceiptsPageResponse;
+use crate::eth::rpc::types::BlockWithChangesPageResponse;
+use crate::eth::rpc::types::IMPORTER_PAGE_LIMIT_DEFAULT;
+use crate::eth::rpc::types::ImporterPageInfo;
+use crate::eth::rpc::types::ImporterPageRequest;
+use crate::eth::rpc::types::PageReducer;
+use crate::eth::rpc::types::PaginatedPageFetcher;
+use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
+use crate::eth::storage::permanent::rocks::types::BlockRocksdb;
+use crate::eth::types::BlockNumber;
+use crate::eth::types::ExternalBlock;
+use crate::eth::types::ExternalBlockWithReceipts;
+use crate::eth::types::ExternalReceipt;
+use crate::ext::to_json_value;
+use crate::log_and_err;
+
+const GET_BLOCK_AND_RECEIPTS: &str = "stratus_getBlockAndReceipts";
+const GET_BLOCK_WITH_CHANGES: &str = "stratus_getBlockWithChanges";
+
+pub(super) struct ImporterPaginationClient<'a> {
+    client: &'a BlockchainClient,
+}
+
+impl<'a> ImporterPaginationClient<'a> {
+    pub(super) fn new(client: &'a BlockchainClient) -> Self {
+        Self { client }
+    }
+
+    pub(super) async fn fetch_block_and_receipts(&self, block_number: BlockNumber) -> anyhow::Result<Option<ExternalBlockWithReceipts>> {
+        PaginatedPageFetcher::new(BlockAndReceiptsPages::new(block_number))
+            .collect(|cursor| self.fetch_page(GET_BLOCK_AND_RECEIPTS, block_number, cursor, "failed to fetch block with receipts"))
+            .await
+    }
+
+    pub(super) async fn fetch_block_with_changes(&self, block_number: BlockNumber) -> anyhow::Result<Option<(BlockRocksdb, BlockChangesRocksdb)>> {
+        PaginatedPageFetcher::new(BlockWithChangesPages::new(block_number))
+            .collect(|cursor| self.fetch_page(GET_BLOCK_WITH_CHANGES, block_number, cursor, "failed to fetch block with changes"))
+            .await
+    }
+
+    async fn fetch_page<T>(&self, method: &str, block_number: BlockNumber, cursor: Option<String>, error_message: &'static str) -> anyhow::Result<Option<T>>
+    where
+        T: DeserializeOwned,
+    {
+        let params = [
+            to_json_value(block_number),
+            to_json_value(ImporterPageRequest {
+                cursor,
+                limit: Some(IMPORTER_PAGE_LIMIT_DEFAULT),
+            }),
+        ];
+
+        match self.client.http.request::<Option<T>, _>(method, params).await {
+            Ok(page) => Ok(page),
+            Err(e) => log_and_err!(reason = e, error_message),
+        }
+    }
+}
+
+fn validate_progress(page: &ImporterPageInfo, expected_total: &mut Option<usize>, context: &str) -> anyhow::Result<Option<String>> {
+    if page.returned == 0 && page.next_cursor.is_some() {
+        bail!("paginated {context} returned no items but provided a next cursor");
+    }
+
+    match expected_total {
+        Some(expected_total) if *expected_total != page.total => {
+            bail!("paginated {context} changed total from {expected_total} to {}", page.total);
+        }
+        Some(_) => {}
+        None => *expected_total = Some(page.total),
+    }
+
+    Ok(page.next_cursor.clone())
+}
+
+struct BlockAndReceiptsPages {
+    block_number: BlockNumber,
+    block: Option<ExternalBlock>,
+    receipts: Vec<ExternalReceipt>,
+    expected_total: Option<usize>,
+}
+
+impl BlockAndReceiptsPages {
+    fn new(block_number: BlockNumber) -> Self {
+        Self {
+            block_number,
+            block: None,
+            receipts: Vec::new(),
+            expected_total: None,
+        }
+    }
+
+    fn push_block(&mut self, page_block: ExternalBlock) -> anyhow::Result<()> {
+        if page_block.number() != self.block_number {
+            bail!(
+                "paginated block with receipts returned unexpected block number {} instead of {}",
+                page_block.number(),
+                self.block_number
+            );
+        }
+
+        match &mut self.block {
+            Some(block) => block.extend_full_transactions_from(page_block),
+            None => {
+                self.block = Some(page_block);
+                Ok(())
+            }
+        }
+    }
+}
+
+impl PageReducer<BlockAndReceiptsPageResponse> for BlockAndReceiptsPages {
+    type Output = ExternalBlockWithReceipts;
+    type NextPage = String;
+
+    fn reduce(&mut self, page: BlockAndReceiptsPageResponse) -> anyhow::Result<Option<String>> {
+        let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with receipts")?;
+        let page_block = ExternalBlock::try_from(page.block)?;
+        self.push_block(page_block)?;
+        self.receipts.extend(page.receipts);
+        Ok(cursor)
+    }
+
+    fn finish_after_not_found(self) -> anyhow::Result<Option<Self::Output>> {
+        if self.block.is_none() {
+            Ok(None)
+        } else {
+            bail!("block disappeared while fetching paginated block with receipts");
+        }
+    }
+
+    fn finish(self) -> anyhow::Result<Option<Self::Output>> {
+        let Some(block) = self.block else {
+            return Ok(None);
+        };
+
+        let expected_total = self.expected_total.unwrap_or_default();
+        let transactions_len = block.full_transactions_len()?;
+        if transactions_len != expected_total {
+            bail!("paginated block with receipts assembled {transactions_len} transactions but expected {expected_total}");
+        }
+        if transactions_len != self.receipts.len() {
+            bail!(
+                "paginated block with receipts assembled {} transactions but {} receipts",
+                transactions_len,
+                self.receipts.len()
+            );
+        }
+
+        Ok(Some(ExternalBlockWithReceipts {
+            block,
+            receipts: self.receipts,
+        }))
+    }
+}
+
+struct BlockWithChangesPages {
+    block_number: BlockNumber,
+    block: Option<BlockRocksdb>,
+    changes: BlockChangesRocksdb,
+    expected_total: Option<usize>,
+}
+
+impl BlockWithChangesPages {
+    fn new(block_number: BlockNumber) -> Self {
+        Self {
+            block_number,
+            block: None,
+            changes: BlockChangesRocksdb::default(),
+            expected_total: None,
+        }
+    }
+
+    fn push_block(&mut self, page_block: BlockRocksdb) -> anyhow::Result<()> {
+        let page_block_number = BlockNumber::from(page_block.header.number);
+        if page_block_number != self.block_number {
+            bail!(
+                "paginated block with changes returned unexpected block number {page_block_number} instead of {}",
+                self.block_number
+            );
+        }
+
+        match &mut self.block {
+            Some(block) => {
+                if block.header.hash != page_block.header.hash {
+                    bail!("paginated block with changes changed block hash");
+                }
+                block.transactions.extend(page_block.transactions);
+            }
+            None => self.block = Some(page_block),
+        }
+
+        Ok(())
+    }
+
+    fn push_changes(&mut self, page_changes: BlockChangesRocksdb) -> anyhow::Result<()> {
+        for (address, change) in page_changes.account_changes {
+            if self.changes.account_changes.insert(address, change).is_some() {
+                bail!("paginated block with changes returned duplicate account change");
+            }
+        }
+        for (slot, value) in page_changes.slot_changes {
+            if self.changes.slot_changes.insert(slot, value).is_some() {
+                bail!("paginated block with changes returned duplicate slot change");
+            }
+        }
+        Ok(())
+    }
+}
+
+impl PageReducer<BlockWithChangesPageResponse> for BlockWithChangesPages {
+    type Output = (BlockRocksdb, BlockChangesRocksdb);
+    type NextPage = String;
+
+    fn reduce(&mut self, page: BlockWithChangesPageResponse) -> anyhow::Result<Option<String>> {
+        let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with changes")?;
+        self.push_block(page.block)?;
+        self.push_changes(page.changes)?;
+        Ok(cursor)
+    }
+
+    fn finish_after_not_found(self) -> anyhow::Result<Option<Self::Output>> {
+        if self.block.is_none() {
+            Ok(None)
+        } else {
+            bail!("block disappeared while fetching paginated block with changes");
+        }
+    }
+
+    fn finish(self) -> anyhow::Result<Option<Self::Output>> {
+        let Some(block) = self.block else {
+            return Ok(None);
+        };
+
+        let expected_total = self.expected_total.unwrap_or_default();
+        let total = block.transactions.len() + self.changes.account_changes.len() + self.changes.slot_changes.len();
+        if total != expected_total {
+            bail!("paginated block with changes assembled {total} items but expected {expected_total}");
+        }
+
+        Ok(Some((block, self.changes)))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use fake::Fake;
+    use fake::Faker;
+    use hash_hasher::HashBuildHasher;
+
+    use super::BlockAndReceiptsPages;
+    use super::BlockWithChangesPages;
+    use super::validate_progress;
+    use crate::eth::rpc::types::BlockAndReceiptsPageResponse;
+    use crate::eth::rpc::types::BlockWithChangesPageResponse;
+    use crate::eth::rpc::types::ImporterPageInfo;
+    use crate::eth::rpc::types::PageReducer;
+    use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb;
+    use crate::eth::storage::permanent::rocks::types::AddressRocksdb;
+    use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
+    use crate::eth::storage::permanent::rocks::types::BlockRocksdb;
+    use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb;
+    use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb;
+    use crate::eth::types::Block;
+    use crate::eth::types::BlockNumber;
+    use crate::eth::types::ExternalBlock;
+    use crate::eth::types::ExternalReceipt;
+    use crate::eth::types::SlotIndex;
+    use crate::eth::types::SlotValue;
+    use crate::eth::types::UnixTime;
+    use crate::ext::to_json_value;
+
+    // helpers
+
+    fn page_info(returned: usize, total: usize, next_cursor: Option<&str>) -> ImporterPageInfo {
+        ImporterPageInfo {
+            limit: 256,
+            returned,
+            total,
+            next_cursor: next_cursor.map(String::from),
+        }
+    }
+
+    fn external_block_with_txs(count: usize) -> ExternalBlock {
+        let mut block: ExternalBlock = Faker.fake();
+        let txs: Vec<_> = std::iter::repeat_with(|| Faker.fake()).take(count).collect();
+        block.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs);
+        block
+    }
+
+    fn split_external_block(block: &ExternalBlock, at: usize) -> (ExternalBlock, ExternalBlock) {
+        let txs = match &block.0.transactions {
+            alloy_rpc_types_eth::BlockTransactions::Full(txs) => txs.clone(),
+            _ => unreachable!(),
+        };
+        let mut page1 = block.clone();
+        let mut page2 = block.clone();
+        page1.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs[..at].to_vec());
+        page2.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs[at..].to_vec());
+        (page1, page2)
+    }
+
+    fn receipts_page(block: ExternalBlock, receipts: Vec<ExternalReceipt>, pagination: ImporterPageInfo) -> BlockAndReceiptsPageResponse {
+        BlockAndReceiptsPageResponse {
+            block: to_json_value(block),
+            receipts,
+            pagination,
+        }
+    }
+
+    fn block_rocksdb_with_txs(number: u64, count: usize) -> BlockRocksdb {
+        let mut block = Block::new(BlockNumber::from(number), UnixTime::from(0u64));
+        block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect();
+        BlockRocksdb::from(block)
+    }
+
+    fn changes_with_accounts(addresses: &[[u8; 20]]) -> BlockChangesRocksdb {
+        let mut account_changes = HashMap::with_hasher(HashBuildHasher::default());
+        for addr in addresses {
+            account_changes.insert(AddressRocksdb(*addr), AccountChangesRocksdb::default());
+        }
+        BlockChangesRocksdb {
+            account_changes,
+            slot_changes: HashMap::with_hasher(HashBuildHasher::default()),
+        }
+    }
+
+    fn changes_with_slots(slots: &[(AddressRocksdb, SlotIndexRocksdb)]) -> BlockChangesRocksdb {
+        let mut slot_changes = HashMap::with_hasher(HashBuildHasher::default());
+        for (addr, idx) in slots {
+            slot_changes.insert((*addr, *idx), SlotValueRocksdb::default());
+        }
+        BlockChangesRocksdb {
+            account_changes: HashMap::with_hasher(HashBuildHasher::default()),
+            slot_changes,
+        }
+    }
+
+    fn slot_idx(a: u64, b: u64, c: u64, d: u64) -> SlotIndexRocksdb {
+        SlotIndexRocksdb::from(SlotIndex::from([a, b, c, d]))
+    }
+
+    fn slot_val(a: u64, b: u64, c: u64, d: u64) -> SlotValueRocksdb {
+        SlotValueRocksdb::from(SlotValue::from([a, b, c, d]))
+    }
+
+    // validate_progress
+
+    #[test]
+    fn validate_progress_first_page_sets_expected_total() {
+        let mut expected_total = None;
+        let info = page_info(3, 10, Some("cursor"));
+        let cursor = validate_progress(&info, &mut expected_total, "test").expect("ok");
+        assert_eq!(expected_total, Some(10));
+        assert_eq!(cursor, Some("cursor".to_string()));
+    }
+
+    #[test]
+    fn validate_progress_same_total_ok() {
+        let mut expected_total = Some(10);
+        let info = page_info(3, 10, Some("cursor"));
+        let cursor = validate_progress(&info, &mut expected_total, "test").expect("ok");
+        assert_eq!(cursor, Some("cursor".to_string()));
+    }
+
+    #[test]
+    fn validate_progress_different_total_errors() {
+        let mut expected_total = Some(10);
+        let info = page_info(3, 20, Some("cursor"));
+        assert!(validate_progress(&info, &mut expected_total, "test").is_err());
+    }
+
+    #[test]
+    fn validate_progress_zero_returned_with_cursor_errors() {
+        let mut expected_total = None;
+        let info = page_info(0, 10, Some("cursor"));
+        assert!(validate_progress(&info, &mut expected_total, "test").is_err());
+    }
+
+    // BlockAndReceiptsPages reducer
+
+    #[test]
+    fn receipts_reducer_two_pages_reassemble() {
+        let full_block = external_block_with_txs(5);
+        let (page1_block, page2_block) = split_external_block(&full_block, 3);
+        let block_number = full_block.number();
+
+        let mut reducer = BlockAndReceiptsPages::new(block_number);
+
+        let page1 = receipts_page(page1_block, vec![Faker.fake(); 3], page_info(3, 5, Some("cursor")));
+        let cursor = reducer.reduce(page1).expect("ok");
+        assert_eq!(cursor, Some("cursor".to_string()));
+
+        let page2 = receipts_page(page2_block, vec![Faker.fake(); 2], page_info(2, 5, None));
+        let cursor = reducer.reduce(page2).expect("ok");
+        assert!(cursor.is_none());
+
+        let result = reducer.finish().expect("ok").expect("some output");
+        assert_eq!(result.block.full_transactions_len().unwrap(), 5);
+        assert_eq!(result.receipts.len(), 5);
+    }
+
+    #[test]
+    fn receipts_reducer_wrong_block_number_errors() {
+        let block = external_block_with_txs(2);
+        let block_number = BlockNumber::from(999u64);
+
+        let mut reducer = BlockAndReceiptsPages::new(block_number);
+        let page = receipts_page(block, vec![], page_info(0, 0, None));
+        assert!(reducer.reduce(page).is_err());
+    }
+
+    #[test]
+    fn receipts_reducer_count_mismatch_errors() {
+        let block = external_block_with_txs(3);
+        let block_number = block.number();
+
+        let mut reducer = BlockAndReceiptsPages::new(block_number);
+        let page = receipts_page(block, vec![Faker.fake(); 2], page_info(3, 3, None));
+        reducer.reduce(page).expect("ok");
+
+        assert!(reducer.finish().is_err());
+    }
+
+    #[test]
+    fn receipts_reducer_finish_after_not_found_no_block_returns_none() {
+        let reducer = BlockAndReceiptsPages::new(BlockNumber::from(1u64));
+        assert!(reducer.finish_after_not_found().expect("ok").is_none());
+    }
+
+    #[test]
+    fn receipts_reducer_finish_after_not_found_with_partial_block_errors() {
+        let block = external_block_with_txs(1);
+        let mut reducer = BlockAndReceiptsPages::new(block.number());
+        let page = receipts_page(block, vec![Faker.fake()], page_info(1, 1, Some("cursor")));
+        reducer.reduce(page).expect("ok");
+
+        assert!(reducer.finish_after_not_found().is_err());
+    }
+
+    // BlockWithChangesPages reducer
+
+    #[test]
+    fn changes_reducer_two_pages_reassemble() {
+        let block_number = BlockNumber::from(1u64);
+        let full_block = block_rocksdb_with_txs(1, 3);
+        let (page1_block, page2_block) = {
+            let txs = full_block.transactions.clone();
+            let mut p1 = full_block.clone();
+            let mut p2 = full_block.clone();
+            p1.transactions = txs[..1].to_vec();
+            p2.transactions = txs[1..].to_vec();
+            (p1, p2)
+        };
+
+        let addr_a = AddressRocksdb([0x01; 20]);
+        let addr_b = AddressRocksdb([0x02; 20]);
+        let changes1 = changes_with_accounts(&[addr_a.0]);
+        let changes2 = changes_with_accounts(&[addr_b.0]);
+
+        let mut reducer = BlockWithChangesPages::new(block_number);
+
+        let page1 = BlockWithChangesPageResponse {
+            block: page1_block,
+            changes: changes1,
+            pagination: page_info(2, 5, Some("cursor")),
+        };
+        let cursor = reducer.reduce(page1).expect("ok");
+        assert_eq!(cursor, Some("cursor".to_string()));
+
+        let page2 = BlockWithChangesPageResponse {
+            block: page2_block,
+            changes: changes2,
+            pagination: page_info(3, 5, None),
+        };
+        let cursor = reducer.reduce(page2).expect("ok");
+        assert!(cursor.is_none());
+
+        let (block, changes) = reducer.finish().expect("ok").expect("some output");
+        assert_eq!(block.transactions.len(), 3);
+        assert_eq!(changes.account_changes.len(), 2);
+    }
+
+    #[test]
+    fn changes_reducer_block_number_mismatch_errors() {
+        let block = block_rocksdb_with_txs(1, 1);
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(999u64));
+        let page = BlockWithChangesPageResponse {
+            block,
+            changes: BlockChangesRocksdb::default(),
+            pagination: page_info(1, 1, None),
+        };
+        assert!(reducer.reduce(page).is_err());
+    }
+
+    #[test]
+    fn changes_reducer_hash_changed_between_pages_errors() {
+        let block1 = block_rocksdb_with_txs(1, 1);
+        let mut block2 = block_rocksdb_with_txs(1, 1);
+        block2.header.hash = Faker.fake();
+
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+
+        let page1 = BlockWithChangesPageResponse {
+            block: block1,
+            changes: BlockChangesRocksdb::default(),
+            pagination: page_info(1, 2, Some("cursor")),
+        };
+        reducer.reduce(page1).expect("ok");
+
+        let page2 = BlockWithChangesPageResponse {
+            block: block2,
+            changes: BlockChangesRocksdb::default(),
+            pagination: page_info(1, 2, None),
+        };
+        assert!(reducer.reduce(page2).is_err());
+    }
+
+    #[test]
+    fn changes_reducer_duplicate_account_change_errors() {
+        let block = block_rocksdb_with_txs(1, 0);
+        let addr = [0x01; 20];
+        let changes = changes_with_accounts(&[addr]);
+
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+
+        let page1 = BlockWithChangesPageResponse {
+            block: block.clone(),
+            changes: changes.clone(),
+            pagination: page_info(1, 2, Some("cursor")),
+        };
+        reducer.reduce(page1).expect("ok");
+
+        let page2 = BlockWithChangesPageResponse {
+            block,
+            changes,
+            pagination: page_info(1, 2, None),
+        };
+        assert!(reducer.reduce(page2).is_err());
+    }
+
+    #[test]
+    fn changes_reducer_duplicate_slot_change_errors() {
+        let block = block_rocksdb_with_txs(1, 0);
+        let addr = AddressRocksdb([0x01; 20]);
+        let idx = slot_idx(0, 0, 0, 1);
+        let changes = changes_with_slots(&[(addr, idx)]);
+
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+
+        let page1 = BlockWithChangesPageResponse {
+            block: block.clone(),
+            changes: changes.clone(),
+            pagination: page_info(1, 2, Some("cursor")),
+        };
+        reducer.reduce(page1).expect("ok");
+
+        let page2 = BlockWithChangesPageResponse {
+            block,
+            changes,
+            pagination: page_info(1, 2, None),
+        };
+        assert!(reducer.reduce(page2).is_err());
+    }
+
+    #[test]
+    fn changes_reducer_total_mismatch_errors() {
+        let block = block_rocksdb_with_txs(1, 2);
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+
+        let page = BlockWithChangesPageResponse {
+            block,
+            changes: BlockChangesRocksdb::default(),
+            pagination: page_info(2, 10, None),
+        };
+        reducer.reduce(page).expect("ok");
+
+        assert!(reducer.finish().is_err());
+    }
+
+    #[test]
+    fn changes_reducer_finish_after_not_found_no_block_returns_none() {
+        let reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+        assert!(reducer.finish_after_not_found().expect("ok").is_none());
+    }
+
+    #[test]
+    fn changes_reducer_finish_after_not_found_with_partial_block_errors() {
+        let block = block_rocksdb_with_txs(1, 1);
+        let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64));
+        let page = BlockWithChangesPageResponse {
+            block,
+            changes: BlockChangesRocksdb::default(),
+            pagination: page_info(1, 1, Some("cursor")),
+        };
+        reducer.reduce(page).expect("ok");
+
+        assert!(reducer.finish_after_not_found().is_err());
+    }
+
+    // Slot helpers for completeness
+
+    #[test]
+    fn slot_helpers_construct_valid_values() {
+        let _ = slot_val(1, 2, 3, 4);
+        let _ = slot_idx(1, 2, 3, 4);
+    }
+}
diff --git a/src/eth/rpc/blockchain_client/mod.rs b/src/eth/rpc/blockchain_client/mod.rs
index 721ae25ac..566e98a09 100644
--- a/src/eth/rpc/blockchain_client/mod.rs
+++ b/src/eth/rpc/blockchain_client/mod.rs
@@ -1,4 +1,5 @@
 #[allow(clippy::module_inception)]
 pub mod blockchain_client;
+mod importer_pagination;
 
 pub use blockchain_client::BlockchainClient;
diff --git a/src/eth/rpc/server.rs b/src/eth/rpc/server.rs
index 5c5d476a3..58e049f34 100644
--- a/src/eth/rpc/server.rs
+++ b/src/eth/rpc/server.rs
@@ -72,6 +72,7 @@ use crate::eth::rpc::next_rpc_param;
 use crate::eth::rpc::next_rpc_param_or_default;
 use crate::eth::rpc::parser::RpcExtensionsExt;
 use crate::eth::rpc::subscriptions::RpcSubscriptionsHandles;
+use crate::eth::rpc::types::ImporterPagination;
 use crate::eth::storage::ExecutionKind;
 use crate::eth::storage::StorageError;
 use crate::eth::storage::StratusStorage;
@@ -897,11 +898,23 @@ fn stratus_get_block_and_receipts(params: Params<'_>, ctx: Arc<RpcContext>, ext:
     let _method_enter = info_span!("rpc::stratus_getBlockAndReceipts").entered();
 
     // parse params
-    let (_, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;
+    let (params, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;
+    let pagination = ImporterPagination::from_params(params, filter)?;
 
     // track
     tracing::info!(%filter, "reading block and receipts");
 
+    if let Some((filter, pagination)) = pagination {
+        let Some(block) = ctx.server.storage.read_block(filter)? else {
+            tracing::info!(%filter, "block not found");
+            return Ok(JsonValue::Null);
+        };
+
+        let response = pagination.block_and_receipts_response(block)?;
+        tracing::info!(%filter, returned = response.pagination.returned, total = response.pagination.total, "block with transactions page found");
+        return Ok(json!(response));
+    }
+
     let Some(block) = ctx.server.storage.read_block(filter)? else {
         tracing::info!(%filter, "block not found");
         return Ok(JsonValue::Null);
@@ -922,11 +935,23 @@ fn stratus_get_block_with_changes(params: Params<'_>, ctx: Arc<RpcContext>, ext:
     let _method_enter = info_span!("rpc::stratus_getBlockWithChanges").entered();
 
     // parse params
-    let (_, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;
+    let (params, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;
+    let pagination = ImporterPagination::from_params(params, filter)?;
 
     // track
     tracing::info!(%filter, "reading block and changes");
 
+    if let Some((filter, pagination)) = pagination {
+        let Some((block, changes)) = ctx.server.storage.read_block_with_changes(filter)? else {
+            tracing::info!(%filter, "block not found");
+            return Ok(JsonValue::Null);
+        };
+
+        let response = pagination.block_with_changes_response(block, changes)?;
+        tracing::info!(%filter, returned = response.pagination.returned, total = response.pagination.total, "block with changes page found");
+        return Ok(json!(response));
+    }
+
     let Some(block) = ctx.server.storage.read_block_with_changes(filter)? else {
         tracing::info!(%filter, "block not found");
         return Ok(JsonValue::Null);
diff --git a/src/eth/rpc/types/importer_pagination.rs b/src/eth/rpc/types/importer_pagination.rs
new file mode 100644
index 000000000..325916169
--- /dev/null
+++ b/src/eth/rpc/types/importer_pagination.rs
@@ -0,0 +1,488 @@
+use jsonrpsee::types::ParamsSequence;
+
+use super::BlockFilter;
+use super::RpcError;
+use super::pagination::CursorCodec;
+use super::pagination::CursorPageInfo;
+use super::pagination::CursorPaginator;
+use super::pagination::Paginator;
+use crate::alias::AlloyReceipt;
+use crate::alias::JsonValue;
+use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb;
+use crate::eth::storage::permanent::rocks::types::AddressRocksdb;
+use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
+use crate::eth::storage::permanent::rocks::types::BlockRocksdb;
+use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb;
+use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb;
+use crate::eth::types::Block;
+use crate::eth::types::ExternalReceipt;
+use crate::eth::types::Hash;
+
+pub const IMPORTER_PAGE_LIMIT_DEFAULT: usize = 256;
+pub const IMPORTER_PAGE_LIMIT_MAX: usize = 5_000;
+
+const IMPORTER_CURSOR_VERSION: &str = "v1";
+
+#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ImporterPageRequest {
+    pub cursor: Option<String>,
+    pub limit: Option<usize>,
+}
+
+impl ImporterPageRequest {
+    fn parse_next(mut params: ParamsSequence<'_>) -> Result<Option<Self>, RpcError> {
+        match params.optional_next::<Self>() {
+            Ok(page_request) => Ok(page_request),
+            Err(e) => Err(RpcError::ParameterDecodeError {
+                rust_type: "ImporterPageRequest",
+                decode_error: e.data().map(|x| x.to_string()).unwrap_or_default(),
+            }),
+        }
+    }
+
+    pub(crate) fn limit(&self) -> usize {
+        match self.limit {
+            Some(0) | None => IMPORTER_PAGE_LIMIT_DEFAULT,
+            Some(limit) => limit.min(IMPORTER_PAGE_LIMIT_MAX),
+        }
+    }
+}
+
+pub struct ImporterPagination {
+    request: ImporterPageRequest,
+    start: usize,
+}
+
+impl ImporterPagination {
+    pub fn from_params(params: ParamsSequence<'_>, filter: BlockFilter) -> Result<Option<(BlockFilter, Self)>, RpcError> {
+        let Some(request) = ImporterPageRequest::parse_next(params)? else {
+            return Ok(None);
+        };
+        let (filter, start) = Self::resolve_filter(filter, request.cursor.as_deref())?;
+        Ok(Some((filter, Self { request, start })))
+    }
+
+    /// Test-only constructor that builds a pagination with the given start index and limit.
+    #[cfg(test)]
+    pub(crate) fn for_test(start: usize, limit: usize) -> Self {
+        Self {
+            request: ImporterPageRequest {
+                cursor: None,
+                limit: Some(limit),
+            },
+            start,
+        }
+    }
+
+    pub fn block_and_receipts_response(&self, block: Block) -> Result<BlockAndReceiptsPageResponse, RpcError> {
+        let mut block = block;
+        let mut paginator = self.cursor_paginator(block.transactions.len(), block.hash())?;
+        let tx_range = paginator.take(block.transactions.len());
+        let transactions = block.transactions[tx_range].to_vec();
+        let receipts = transactions.iter().cloned().map(AlloyReceipt::from).map(ExternalReceipt).collect::<Vec<_>>();
+
+        block.transactions = transactions;
+
+        Ok(BlockAndReceiptsPageResponse {
+            block: block.to_json_rpc_with_full_transactions(),
+            receipts,
+            pagination: paginator.finish(),
+        })
+    }
+
+    pub fn block_with_changes_response(&self, block: BlockRocksdb, changes: BlockChangesRocksdb) -> Result<BlockWithChangesPageResponse, RpcError> {
+        let BlockRocksdb { header, transactions } = block;
+        let total = transactions.len() + changes.account_changes.len() + changes.slot_changes.len();
+        let mut paginator = self.cursor_paginator(total, header.hash.into())?;
+
+        let tx_range = paginator.take(transactions.len());
+        let account_entries = sorted_account_changes(&changes);
+        let account_range = paginator.take(account_entries.len());
+        let slot_entries = sorted_slot_changes(&changes);
+        let slot_range = paginator.take(slot_entries.len());
+
+        let mut page_changes = BlockChangesRocksdb::with_capacity(account_range.len());
+        for (address, change) in account_entries[account_range].iter().cloned() {
+            page_changes.account_changes.insert(address, change);
+        }
+        for ((address, slot), value) in slot_entries[slot_range].iter().copied() {
+            page_changes.slot_changes.insert((address, slot), value);
+        }
+
+        Ok(BlockWithChangesPageResponse {
+            block: BlockRocksdb {
+                header,
+                transactions: transactions[tx_range].to_vec(),
+            },
+            changes: page_changes,
+            pagination: paginator.finish(),
+        })
+    }
+
+    fn resolve_filter(filter: BlockFilter, cursor: Option<&str>) -> Result<(BlockFilter, usize), RpcError> {
+        match cursor {
+            Some(cursor) => {
+                let (cursor, next_index) = BlockHashCursor::decode_cursor(cursor)?;
+                Ok((BlockFilter::Hash(cursor.block_hash), next_index))
+            }
+            None => Ok((filter, 0)),
+        }
+    }
+
+    fn cursor_paginator(&self, total: usize, block_hash: Hash) -> Result<ImporterCursorPaginator, RpcError> {
+        ImporterCursorPaginator::new(total, self.start, self.request.limit(), BlockHashCursor { block_hash })
+    }
+}
+
+pub type ImporterPageInfo = CursorPageInfo;
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BlockAndReceiptsPageResponse {
+    pub block: JsonValue,
+    pub receipts: Vec<ExternalReceipt>,
+    pub pagination: ImporterPageInfo,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BlockWithChangesPageResponse {
+    pub block: BlockRocksdb,
+    pub changes: BlockChangesRocksdb,
+    pub pagination: ImporterPageInfo,
+}
+
+fn sorted_account_changes(changes: &BlockChangesRocksdb) -> Vec<(AddressRocksdb, AccountChangesRocksdb)> {
+    let mut entries = changes
+        .account_changes
+        .iter()
+        .map(|(address, change)| (*address, change.clone()))
+        .collect::<Vec<_>>();
+    entries.sort_by_key(|(address, _)| *address);
+    entries
+}
+
+fn sorted_slot_changes(changes: &BlockChangesRocksdb) -> Vec<((AddressRocksdb, SlotIndexRocksdb), SlotValueRocksdb)> {
+    let mut entries = changes.slot_changes.iter().map(|(key, value)| (*key, *value)).collect::<Vec<_>>();
+    entries.sort_by_key(|(key, _)| *key);
+    entries
+}
+
+type ImporterCursorPaginator = CursorPaginator<BlockHashCursor>;
+
+pub(crate) struct BlockHashCursor {
+    block_hash: Hash,
+}
+
+impl CursorCodec for BlockHashCursor {
+    type Error = RpcError;
+
+    fn invalid_start_error() -> Self::Error {
+        RpcError::ParameterInvalid
+    }
+
+    fn encode_cursor(&self, next_index: usize) -> String {
+        format!("{IMPORTER_CURSOR_VERSION}:{}:{next_index}", self.block_hash)
+    }
+
+    fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error> {
+        let mut parts = cursor.split(':');
+        let version = parts.next().ok_or(RpcError::ParameterInvalid)?;
+        let block_hash = parts.next().ok_or(RpcError::ParameterInvalid)?;
+        let next_index = parts.next().ok_or(RpcError::ParameterInvalid)?;
+
+        if version != IMPORTER_CURSOR_VERSION || parts.next().is_some() {
+            return Err(RpcError::ParameterInvalid);
+        }
+
+        Ok((
+            Self {
+                block_hash: block_hash.parse().map_err(|_| RpcError::ParameterInvalid)?,
+            },
+            next_index.parse().map_err(|_| RpcError::ParameterInvalid)?,
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use fake::Fake;
+    use fake::Faker;
+    use hash_hasher::HashBuildHasher;
+    use jsonrpsee::types::Params;
+
+    use super::BlockHashCursor;
+    use super::CursorCodec;
+    use super::IMPORTER_PAGE_LIMIT_DEFAULT;
+    use super::IMPORTER_PAGE_LIMIT_MAX;
+    use super::ImporterPageRequest;
+    use super::ImporterPagination;
+    use super::RpcError;
+    use crate::eth::rpc::BlockFilter;
+    use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb;
+    use crate::eth::storage::permanent::rocks::types::AddressRocksdb;
+    use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb;
+    use crate::eth::storage::permanent::rocks::types::BlockRocksdb;
+    use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb;
+    use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb;
+    use crate::eth::types::Block;
+    use crate::eth::types::BlockNumber;
+    use crate::eth::types::Hash;
+    use crate::eth::types::SlotIndex;
+    use crate::eth::types::SlotValue;
+    use crate::eth::types::UnixTime;
+
+    fn codec(block_hash: &str) -> BlockHashCursor {
+        BlockHashCursor {
+            block_hash: block_hash.parse().unwrap(),
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // Cursor codec tests
+    // -------------------------------------------------------------------------
+
+    #[test]
+    fn cursor_roundtrip_preserves_hash_and_index() {
+        let original = codec("0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c");
+        let encoded = original.encode_cursor(42);
+        let (decoded, next_index) = BlockHashCursor::decode_cursor(&encoded).expect("decode succeeds");
+
+        assert_eq!(decoded.block_hash, original.block_hash);
+        assert_eq!(next_index, 42);
+    }
+
+    #[test]
+    fn cursor_decode_rejects_wrong_version() {
+        let bad = "v2:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:0";
+        let err = BlockHashCursor::decode_cursor(bad).err();
+        assert!(matches!(err, Some(RpcError::ParameterInvalid)));
+    }
+
+    #[test]
+    fn cursor_decode_rejects_missing_parts() {
+        assert!(BlockHashCursor::decode_cursor("v1:0xabc").is_err());
+        assert!(BlockHashCursor::decode_cursor("v1").is_err());
+    }
+
+    #[test]
+    fn cursor_decode_rejects_extra_parts() {
+        let extra = "v1:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:0:extra";
+        assert!(BlockHashCursor::decode_cursor(extra).is_err());
+    }
+
+    #[test]
+    fn cursor_decode_rejects_non_numeric_index() {
+        let bad = "v1:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:notanumber";
+        assert!(BlockHashCursor::decode_cursor(bad).is_err());
+    }
+
+    #[test]
+    fn cursor_decode_rejects_invalid_hash() {
+        let bad = "v1:0xnotahash:0";
+        assert!(BlockHashCursor::decode_cursor(bad).is_err());
+    }
+
+    #[test]
+    fn encode_uses_v1_format() {
+        let c = codec("0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c");
+        let encoded = c.encode_cursor(7);
+        assert!(encoded.starts_with("v1:"));
+        assert_eq!(encoded.split(':').count(), 3);
+    }
+
+    // ensure Hash parses from a 0x-prefixed hex string in tests
+    #[test]
+    fn hash_parses_for_test_fixture() {
+        let _: Hash = "0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c".parse().unwrap();
+    }
+
+    // ImporterPageRequest::limit()
+
+    #[test]
+    fn limit_none_returns_default() {
+        let req = ImporterPageRequest { cursor: None, limit: None };
+        assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_DEFAULT);
+    }
+
+    #[test]
+    fn limit_zero_returns_default() {
+        let req = ImporterPageRequest { cursor: None, limit: Some(0) };
+        assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_DEFAULT);
+    }
+
+    #[test]
+    fn limit_small_value_passed_through() {
+        let req = ImporterPageRequest { cursor: None, limit: Some(50) };
+        assert_eq!(req.limit(), 50);
+    }
+
+    #[test]
+    fn limit_large_value_clamped_to_max() {
+        let req = ImporterPageRequest {
+            cursor: None,
+            limit: Some(10_000),
+        };
+        assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_MAX);
+    }
+
+    // ImporterPagination::from_params
+
+    #[test]
+    fn from_params_no_pagination_param_returns_none() {
+        let params = Params::new(Some("[]"));
+        let result = ImporterPagination::from_params(params.sequence(), BlockFilter::Latest).expect("ok");
+        assert!(result.is_none());
+    }
+
+    #[test]
+    fn from_params_with_cursor_resolves_hash_filter_and_start() {
+        let block_hash = "0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c";
+        let cursor = codec(block_hash).encode_cursor(5);
+        let json = format!(r#"[{{"cursor":"{cursor}","limit":10}}]"#);
+
+        let params = Params::new(Some(&json));
+        let (filter, pagination) = ImporterPagination::from_params(params.sequence(), BlockFilter::Latest)
+            .expect("ok")
+            .expect("pagination present");
+
+        assert!(matches!(filter, BlockFilter::Hash(h) if h == block_hash.parse::<Hash>().unwrap()));
+        assert_eq!(pagination.start, 5);
+    }
+
+    // block_and_receipts_response (server slicing)
+
+    fn block_with_txs(count: usize) -> Block {
+        let mut block = Block::new(BlockNumber::from(1u64), UnixTime::from(0u64));
+        block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect();
+        block
+    }
+
+    #[test]
+    fn block_and_receipts_single_page_returns_all() {
+        let block = block_with_txs(3);
+        let pagination = ImporterPagination::for_test(0, 10);
+        let response = pagination.block_and_receipts_response(block).expect("ok");
+
+        assert_eq!(response.pagination.returned, 3);
+        assert_eq!(response.pagination.total, 3);
+        assert_eq!(response.receipts.len(), 3);
+        assert!(response.pagination.next_cursor.is_none());
+    }
+
+    #[test]
+    fn block_and_receipts_multi_page_slices_correctly() {
+        let block = block_with_txs(5);
+
+        // page 1: start=0, limit=3
+        let pagination = ImporterPagination::for_test(0, 3);
+        let response = pagination.block_and_receipts_response(block.clone()).expect("ok");
+
+        assert_eq!(response.pagination.returned, 3);
+        assert_eq!(response.pagination.total, 5);
+        assert_eq!(response.receipts.len(), 3);
+        let cursor = response.pagination.next_cursor.expect("more pages");
+
+        // decode cursor -> start=3
+        let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor");
+        assert_eq!(start, 3);
+
+        // page 2: start=3, limit=3
+        let pagination = ImporterPagination::for_test(start, 3);
+        let response = pagination.block_and_receipts_response(block).expect("ok");
+
+        assert_eq!(response.pagination.returned, 2);
+        assert_eq!(response.pagination.total, 5);
+        assert_eq!(response.receipts.len(), 2);
+        assert!(response.pagination.next_cursor.is_none());
+    }
+
+    // block_with_changes_response (3-section slicing)
+
+    fn changes_fixture() -> BlockChangesRocksdb {
+        let mut account_changes = HashMap::with_hasher(HashBuildHasher::default());
+        account_changes.insert(AddressRocksdb([0x01; 20]), AccountChangesRocksdb::default());
+        account_changes.insert(AddressRocksdb([0x02; 20]), AccountChangesRocksdb::default());
+        account_changes.insert(AddressRocksdb([0x03; 20]), AccountChangesRocksdb::default());
+
+        let mut slot_changes = HashMap::with_hasher(HashBuildHasher::default());
+        slot_changes.insert(
+            (AddressRocksdb([0x01; 20]), SlotIndexRocksdb::from(SlotIndex::from([0u64, 0, 0, 1]))),
+            SlotValueRocksdb::from(SlotValue::from([0u64, 0, 0, 1])),
+        );
+        slot_changes.insert(
+            (AddressRocksdb([0x01; 20]), SlotIndexRocksdb::from(SlotIndex::from([0u64, 0, 0, 2]))),
+            SlotValueRocksdb::from(SlotValue::from([0u64, 0, 0, 2])),
+        );
+
+        BlockChangesRocksdb { account_changes, slot_changes }
+    }
+
+    fn block_rocksdb_with_txs(count: usize) -> BlockRocksdb {
+        let mut block = Block::new(BlockNumber::from(1u64), UnixTime::from(0u64));
+        block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect();
+        BlockRocksdb::from(block)
+    }
+
+    #[test]
+    fn block_with_changes_multi_page_slices_three_sections() {
+        let block = block_rocksdb_with_txs(2);
+        let changes = changes_fixture();
+        // total = 2 txs + 3 accounts + 2 slots = 7
+
+        // page 1: start=0, limit=3 -> 2 txs + 1 account
+        let pagination = ImporterPagination::for_test(0, 3);
+        let response = pagination.block_with_changes_response(block.clone(), changes.clone()).expect("ok");
+
+        assert_eq!(response.pagination.returned, 3);
+        assert_eq!(response.pagination.total, 7);
+        assert_eq!(response.block.transactions.len(), 2);
+        assert_eq!(response.changes.account_changes.len(), 1);
+        assert_eq!(response.changes.slot_changes.len(), 0);
+        let cursor = response.pagination.next_cursor.expect("more pages");
+        let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor");
+        assert_eq!(start, 3);
+
+        // page 2: start=3, limit=3 -> 2 accounts + 1 slot
+        let pagination = ImporterPagination::for_test(3, 3);
+        let response = pagination.block_with_changes_response(block.clone(), changes.clone()).expect("ok");
+
+        assert_eq!(response.pagination.returned, 3);
+        assert_eq!(response.block.transactions.len(), 0);
+        assert_eq!(response.changes.account_changes.len(), 2);
+        assert_eq!(response.changes.slot_changes.len(), 1);
+        let cursor = response.pagination.next_cursor.expect("more pages");
+        let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor");
+        assert_eq!(start, 6);
+
+        // page 3: start=6, limit=3 -> 1 slot
+        let pagination = ImporterPagination::for_test(6, 3);
+        let response = pagination.block_with_changes_response(block, changes).expect("ok");
+
+        assert_eq!(response.pagination.returned, 1);
+        assert_eq!(response.block.transactions.len(), 0);
+        assert_eq!(response.changes.account_changes.len(), 0);
+        assert_eq!(response.changes.slot_changes.len(), 1);
+        assert!(response.pagination.next_cursor.is_none());
+    }
+
+    #[test]
+    fn block_with_changes_single_page_returns_all() {
+        let block = block_rocksdb_with_txs(1);
+        let changes = changes_fixture();
+        // total = 1 + 3 + 2 = 6
+
+        let pagination = ImporterPagination::for_test(0, 10);
+        let response = pagination.block_with_changes_response(block, changes).expect("ok");
+
+        assert_eq!(response.pagination.returned, 6);
+        assert_eq!(response.pagination.total, 6);
+        assert_eq!(response.block.transactions.len(), 1);
+        assert_eq!(response.changes.account_changes.len(), 3);
+        assert_eq!(response.changes.slot_changes.len(), 2);
+        assert!(response.pagination.next_cursor.is_none());
+    }
+}
diff --git a/src/eth/rpc/types/mod.rs b/src/eth/rpc/types/mod.rs
index 52bbeac38..84a730e7f 100644
--- a/src/eth/rpc/types/mod.rs
+++ b/src/eth/rpc/types/mod.rs
@@ -1,16 +1,26 @@
 mod block_filter;
 mod error;
+mod importer_pagination;
 mod log_filter;
 mod log_filter_input;
+mod pagination;
 mod rpc_client_app;
 mod timestamp_filter;
 
 pub use block_filter::BlockFilter;
 pub use error::MulticallError;
 pub use error::RpcError;
+pub use importer_pagination::BlockAndReceiptsPageResponse;
+pub use importer_pagination::BlockWithChangesPageResponse;
+pub use importer_pagination::IMPORTER_PAGE_LIMIT_DEFAULT;
+pub use importer_pagination::ImporterPageInfo;
+pub use importer_pagination::ImporterPageRequest;
+pub use importer_pagination::ImporterPagination;
 pub use log_filter::LogFilter;
 pub use log_filter_input::LogFilterInput;
 pub use log_filter_input::LogFilterInputTopic;
+pub use pagination::PageReducer;
+pub use pagination::PaginatedPageFetcher;
 pub use rpc_client_app::RpcClientApp;
 pub use timestamp_filter::BlockTimestampFilter;
 pub use timestamp_filter::BlockTimestampSeekMode;
diff --git a/src/eth/rpc/types/pagination.rs b/src/eth/rpc/types/pagination.rs
new file mode 100644
index 000000000..f76310626
--- /dev/null
+++ b/src/eth/rpc/types/pagination.rs
@@ -0,0 +1,406 @@
+use std::future::Future;
+use std::ops::Range;
+
+/// Generic interface exposed by pagination engines.
+///
+/// Kept as a trait so alternative slicing strategies (limit/offset, page number, etc.)
+/// can be slotted in later without touching call sites.
+pub trait Paginator {
+    type Error;
+    type NextPage;
+    type PageInfo;
+
+    fn take(&mut self, section_len: usize) -> Range<usize>;
+    fn finish(self) -> Self::PageInfo;
+}
+
+/// Converts a domain cursor value into the generic cursor-pagination behavior.
+///
+/// Domains implement only this codec; the paginator behavior is shared.
+pub trait CursorCodec: Sized {
+    type Error;
+
+    fn invalid_start_error() -> Self::Error;
+    fn encode_cursor(&self, next_index: usize) -> String;
+    fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error>;
+}
+
+/// Cursor-based paginator.
+///
+/// Knows only how to slice a logical stream. The [`CursorCodec`] decides how
+/// cursors are encoded/decoded; the paginator reports the final page info via
+/// [`CursorPageInfo`].
+pub struct CursorPaginator<C> {
+    codec: C,
+    start: usize,
+    limit: usize,
+    total: usize,
+    returned: usize,
+    skipped: usize,
+    remaining: usize,
+}
+
+impl<C> CursorPaginator<C>
+where
+    C: CursorCodec,
+{
+    pub fn new(total: usize, start: usize, limit: usize, codec: C) -> Result<Self, C::Error> {
+        if start > total || (start == total && total != 0) {
+            return Err(C::invalid_start_error());
+        }
+
+        Ok(Self {
+            codec,
+            start,
+            limit,
+            total,
+            returned: 0,
+            skipped: start,
+            remaining: limit,
+        })
+    }
+
+    fn next_index(&self) -> Option<usize> {
+        let next_index = self.start.saturating_add(self.returned);
+        (next_index < self.total).then_some(next_index)
+    }
+}
+
+impl<C> Paginator for CursorPaginator<C>
+where
+    C: CursorCodec,
+{
+    type Error = C::Error;
+    type NextPage = String;
+    type PageInfo = CursorPageInfo;
+
+    fn take(&mut self, section_len: usize) -> Range<usize> {
+        // skip an entire section
+        if self.skipped >= section_len {
+            self.skipped -= section_len;
+            return section_len..section_len;
+        }
+
+        let start = self.skipped;
+        self.skipped = 0;
+
+        let end = (start + self.remaining).min(section_len);
+        let returned = end - start;
+
+        self.returned += returned;
+        self.remaining -= returned;
+
+        start..end
+    }
+
+    fn finish(self) -> Self::PageInfo {
+        let next_cursor = self.next_index().map(|next_index| self.codec.encode_cursor(next_index));
+        CursorPageInfo {
+            limit: self.limit,
+            returned: self.returned,
+            total: self.total,
+            next_cursor,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CursorPageInfo {
+    pub limit: usize,
+    pub returned: usize,
+    pub total: usize,
+    pub next_cursor: Option<String>,
+}
+
+/// Reducer for client-side paginated fetches.
+///
+/// Every page is folded into an accumulator until the paginator reports
+/// no next page, then the accumulator is finalized.
+pub trait PageReducer<Page> {
+    type Output;
+    type NextPage;
+
+    fn reduce(&mut self, page: Page) -> anyhow::Result<Option<Self::NextPage>>;
+    fn finish_after_not_found(self) -> anyhow::Result<Option<Self::Output>>;
+    fn finish(self) -> anyhow::Result<Option<Self::Output>>;
+}
+
+/// Client-side fetcher that repeatedly fetches pages and reduces them into a final output.
+pub struct PaginatedPageFetcher<Reducer> {
+    reducer: Reducer,
+}
+
+impl<Reducer> PaginatedPageFetcher<Reducer> {
+    pub fn new(reducer: Reducer) -> Self {
+        Self { reducer }
+    }
+
+    pub async fn collect<Page, FetchPage, FetchFuture>(mut self, mut fetch_page: FetchPage) -> anyhow::Result<Option<Reducer::Output>>
+    where
+        Reducer: PageReducer<Page>,
+        FetchPage: FnMut(Option<Reducer::NextPage>) -> FetchFuture,
+        FetchFuture: Future<Output = anyhow::Result<Option<Page>>>,
+    {
+        let mut next_page = None;
+
+        loop {
+            let Some(page) = fetch_page(next_page).await? else {
+                return self.reducer.finish_after_not_found();
+            };
+
+            next_page = self.reducer.reduce(page)?;
+            if next_page.is_none() {
+                return self.reducer.finish();
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::ops::Range;
+
+    use super::CursorCodec;
+    use super::CursorPageInfo;
+    use super::CursorPaginator;
+    use super::Paginator;
+
+    // A minimal codec for testing the paginator independently of any domain type.
+    struct IndexCodec;
+
+    impl CursorCodec for IndexCodec {
+        type Error = &'static str;
+
+        fn invalid_start_error() -> Self::Error {
+            "invalid start"
+        }
+
+        fn encode_cursor(&self, next_index: usize) -> String {
+            next_index.to_string()
+        }
+
+        fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error> {
+            Ok((IndexCodec, cursor.parse().map_err(|_| "bad cursor")?))
+        }
+    }
+
+    fn paginator(total: usize, start: usize, limit: usize) -> CursorPaginator<IndexCodec> {
+        CursorPaginator::new(total, start, limit, IndexCodec).expect("valid paginator")
+    }
+
+    fn take_all(mut p: CursorPaginator<IndexCodec>, sections: &[usize]) -> (Vec<Range<usize>>, CursorPageInfo) {
+        let mut ranges = Vec::new();
+        for &section_len in sections {
+            ranges.push(p.take(section_len));
+        }
+        let info = p.finish();
+        (ranges, info)
+    }
+
+    #[test]
+    fn empty_total_is_valid() {
+        let p = paginator(0, 0, 10);
+        let (_, info) = take_all(p, &[]);
+        assert_eq!(
+            info,
+            CursorPageInfo {
+                limit: 10,
+                returned: 0,
+                total: 0,
+                next_cursor: None
+            }
+        );
+    }
+
+    #[test]
+    fn start_equal_to_total_is_invalid() {
+        assert!(CursorPaginator::new(5, 5, 10, IndexCodec).is_err());
+    }
+
+    #[test]
+    fn start_greater_than_total_is_invalid() {
+        assert!(CursorPaginator::new(5, 6, 10, IndexCodec).is_err());
+    }
+
+    #[test]
+    fn limit_smaller_than_one_section() {
+        let (ranges, info) = take_all(paginator(100, 0, 3), &[10]);
+        assert_eq!(ranges, vec![0..3]);
+        assert_eq!(info.returned, 3);
+        assert_eq!(info.next_cursor, Some("3".to_string()));
+    }
+
+    #[test]
+    fn limit_spanning_multiple_sections() {
+        let (ranges, info) = take_all(paginator(100, 0, 7), &[3, 3, 3]);
+        // first section fully taken (3), second fully taken (3), third partial (1)
+        assert_eq!(ranges, vec![0..3, 0..3, 0..1]);
+        assert_eq!(info.returned, 7);
+        assert_eq!(info.next_cursor, Some("7".to_string()));
+    }
+
+    #[test]
+    fn skip_across_section_boundary() {
+        // start=5, sections of length 3, 3, 3 -> skip 3 (section 1), skip 2 (section 2),
+        // then take 1 from section 2 and the remaining 3 from section 3.
+        let (ranges, info) = take_all(paginator(100, 5, 4), &[3, 3, 3]);
+        assert_eq!(ranges, vec![3..3, 2..3, 0..3]);
+        assert_eq!(info.returned, 4);
+        assert_eq!(info.next_cursor, Some("9".to_string()));
+    }
+
+    #[test]
+    fn fully_consumed_has_no_cursor() {
+        let (ranges, info) = take_all(paginator(5, 0, 5), &[3, 2]);
+        assert_eq!(ranges, vec![0..3, 0..2]);
+        assert_eq!(info.returned, 5);
+        assert_eq!(info.next_cursor, None);
+    }
+
+    #[test]
+    fn limit_larger_than_total_clamps_to_total() {
+        let (ranges, info) = take_all(paginator(4, 0, 100), &[2, 2]);
+        assert_eq!(ranges, vec![0..2, 0..2]);
+        assert_eq!(info.returned, 4);
+        assert_eq!(info.next_cursor, None);
+        assert_eq!(info.total, 4);
+    }
+
+    #[test]
+    fn section_larger_than_remaining_returns_partial() {
+        let (ranges, info) = take_all(paginator(100, 0, 2), &[10]);
+        assert_eq!(ranges, vec![0..2]);
+        assert_eq!(info.returned, 2);
+        assert_eq!(info.next_cursor, Some("2".to_string()));
+    }
+
+    #[test]
+    fn extra_sections_after_limit_are_empty() {
+        let (ranges, info) = take_all(paginator(100, 0, 2), &[2, 2, 2]);
+        assert_eq!(ranges, vec![0..2, 0..0, 0..0]);
+        assert_eq!(info.returned, 2);
+        assert_eq!(info.next_cursor, Some("2".to_string()));
+    }
+
+    // PaginatedPageFetcher::collect tests
+
+    use super::PageReducer;
+    use super::PaginatedPageFetcher;
+
+    /// Stub page carrying data and the next cursor (None = last page).
+    #[derive(Clone)]
+    struct StubPage {
+        data: Vec<u8>,
+        next_cursor: Option<String>,
+    }
+
+    /// Stub reducer that accumulates `StubPage` data into a single concatenated vec.
+    struct CountingReducer {
+        pages: Vec<Vec<u8>>,
+    }
+
+    impl PageReducer<StubPage> for CountingReducer {
+        type Output = Vec<u8>;
+        type NextPage = String;
+
+        fn reduce(&mut self, page: StubPage) -> anyhow::Result<Option<String>> {
+            self.pages.push(page.data);
+            Ok(page.next_cursor)
+        }
+
+        fn finish_after_not_found(self) -> anyhow::Result<Option<Self::Output>> {
+            if self.pages.is_empty() {
+                Ok(None)
+            } else {
+                Err(anyhow::anyhow!("block disappeared while fetching paginated pages"))
+            }
+        }
+
+        fn finish(self) -> anyhow::Result<Option<Self::Output>> {
+            let merged = self.pages.into_iter().flatten().collect();
+            Ok(Some(merged))
+        }
+    }
+
+    #[tokio::test]
+    async fn collect_happy_path_two_pages() {
+        let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() });
+        let pages = [
+            StubPage {
+                data: vec![1, 2, 3],
+                next_cursor: Some("page-2".to_string()),
+            },
+            StubPage {
+                data: vec![4, 5],
+                next_cursor: None,
+            },
+        ];
+        let mut call_idx = 0;
+        let result = fetcher
+            .collect(|_cursor: Option<String>| {
+                let idx = call_idx;
+                call_idx += 1;
+                let page = pages.get(idx).cloned();
+                async move { Ok(page) }
+            })
+            .await
+            .expect("ok");
+
+        let merged = result.expect("some output");
+        assert_eq!(merged, vec![1, 2, 3, 4, 5]);
+    }
+
+    #[tokio::test]
+    async fn collect_not_found_immediately_returns_none() {
+        let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() });
+        let result = fetcher.collect(|_cursor: Option<String>| async { Ok(None) }).await.expect("ok");
+        assert!(result.is_none());
+    }
+
+    #[tokio::test]
+    async fn collect_not_found_after_partial_errors() {
+        let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() });
+        let mut call_idx = 0;
+        let result = fetcher
+            .collect(|_cursor: Option<String>| {
+                let idx = call_idx;
+                call_idx += 1;
+                async move {
+                    if idx == 0 {
+                        Ok(Some(StubPage {
+                            data: vec![1, 2, 3],
+                            next_cursor: Some("page-2".to_string()),
+                        }))
+                    } else {
+                        Ok(None)
+                    }
+                }
+            })
+            .await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn collect_error_propagates() {
+        let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() });
+        let result = fetcher.collect(|_cursor: Option<String>| async { Err(anyhow::anyhow!("network error")) }).await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn collect_single_page_no_cursor() {
+        let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() });
+        let result = fetcher
+            .collect(|_cursor: Option<String>| async {
+                Ok(Some(StubPage {
+                    data: vec![42],
+                    next_cursor: None,
+                }))
+            })
+            .await
+            .expect("ok");
+        let merged = result.expect("some output");
+        assert_eq!(merged, vec![42]);
+    }
+}
diff --git a/src/eth/types/external/external_block.rs b/src/eth/types/external/external_block.rs
index eb4be6198..3c37e75c0 100644
--- a/src/eth/types/external/external_block.rs
+++ b/src/eth/types/external/external_block.rs
@@ -10,6 +10,8 @@ use alloy_primitives::Bloom;
 use alloy_primitives::Bytes;
 #[cfg(test)]
 use alloy_primitives::U256;
+use alloy_rpc_types_eth::BlockTransactions;
+use anyhow::bail;
 #[cfg(test)]
 use fake::Dummy;
 #[cfg(test)]
@@ -55,6 +57,36 @@ impl ExternalBlock {
     pub fn author(&self) -> Address {
         self.0.header.inner.beneficiary.into()
     }
+
+    /// Returns the number of full transactions in the block.
+    pub fn full_transactions_len(&self) -> anyhow::Result<usize> {
+        let BlockTransactions::Full(transactions) = &self.0.transactions else {
+            bail!("expected full transactions, got hashes or uncle");
+        };
+
+        Ok(transactions.len())
+    }
+
+    /// Appends full transactions from another page of the same block.
+    pub fn extend_full_transactions_from(&mut self, other: Self) -> anyhow::Result<()> {
+        if self.hash() != other.hash() {
+            bail!(
+                "cannot extend external block transactions from block {} into block {}",
+                other.hash(),
+                self.hash()
+            );
+        }
+
+        let BlockTransactions::Full(other_transactions) = other.0.transactions else {
+            bail!("expected full transactions, got hashes or uncle");
+        };
+        let BlockTransactions::Full(transactions) = &mut self.0.transactions else {
+            bail!("expected full transactions, got hashes or uncle");
+        };
+
+        transactions.extend(other_transactions);
+        Ok(())
+    }
 }
 
 impl PartialEq<Block> for ExternalBlock {
@@ -129,3 +161,107 @@ impl TryFrom<JsonValue> for ExternalBlock {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use alloy_primitives::B256;
+    use alloy_rpc_types_eth::BlockTransactions;
+    use fake::Fake;
+    use fake::Faker;
+
+    use super::ExternalBlock;
+    use crate::eth::types::ExternalTransaction;
+
+    // Builds an ExternalBlock with a fixed hash and `count` random full transactions.
+    fn block_with_txs(count: usize) -> ExternalBlock {
+        let mut block: ExternalBlock = Faker.fake();
+        block.0.header.hash = fixed_hash();
+        let txs: Vec<ExternalTransaction> = std::iter::repeat_with(|| Faker.fake()).take(count).collect();
+        block.0.transactions = BlockTransactions::Full(txs);
+        block
+    }
+
+    fn fixed_hash() -> B256 {
+        B256::from_slice(&[0xAA; 32])
+    }
+
+    fn as_full(block: &ExternalBlock) -> &Vec<ExternalTransaction> {
+        let BlockTransactions::Full(txs) = &block.0.transactions else {
+            unreachable!("expected full transactions");
+        };
+        txs
+    }
+
+    #[test]
+    fn full_transactions_len_with_full() {
+        let block = block_with_txs(3);
+        assert_eq!(block.full_transactions_len().expect("full transactions"), 3);
+    }
+
+    #[test]
+    fn full_transactions_len_with_hashes() {
+        let mut block = block_with_txs(0);
+        block.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]);
+        assert!(block.full_transactions_len().is_err());
+    }
+
+    #[test]
+    fn full_transactions_len_with_uncle() {
+        let mut block = block_with_txs(0);
+        block.0.transactions = BlockTransactions::Uncle;
+        assert!(block.full_transactions_len().is_err());
+    }
+
+    #[test]
+    fn extend_full_transactions_from_same_hash_merges() {
+        let mut target = block_with_txs(2);
+        let other = block_with_txs(1);
+
+        target.extend_full_transactions_from(other).expect("same hash merges");
+
+        assert_eq!(target.full_transactions_len().expect("full transactions"), 3);
+    }
+
+    #[test]
+    fn extend_full_transactions_from_different_hash_errors() {
+        let mut target = block_with_txs(1);
+        let mut other = block_with_txs(1);
+        other.0.header.hash = B256::from_slice(&[0xBB; 32]);
+
+        assert!(target.extend_full_transactions_from(other).is_err());
+    }
+
+    #[test]
+    fn extend_full_transactions_from_non_full_source_errors() {
+        let mut target = block_with_txs(1);
+        let mut other = block_with_txs(0);
+        other.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]);
+
+        assert!(target.extend_full_transactions_from(other).is_err());
+    }
+
+    #[test]
+    fn extend_full_transactions_into_non_full_target_errors() {
+        let mut target = block_with_txs(0);
+        target.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]);
+        let other = block_with_txs(1);
+
+        assert!(target.extend_full_transactions_from(other).is_err());
+    }
+
+    #[test]
+    fn extend_full_transactions_preserves_order() {
+        let mut target = block_with_txs(1);
+        let original = as_full(&target).clone();
+        let other = block_with_txs(2);
+        let other_txs = as_full(&other).clone();
+
+        target.extend_full_transactions_from(other).expect("same hash merges");
+
+        let merged = as_full(&target);
+        assert_eq!(merged.len(), 3);
+        assert_eq!(merged[0], original[0]);
+        assert_eq!(merged[1], other_txs[0]);
+        assert_eq!(merged[2], other_txs[1]);
+    }
+}