diff --git a/yarn-project/aztec-node/src/aztec-node/server.ts b/yarn-project/aztec-node/src/aztec-node/server.ts
index 8fa83268f59..931e33dc032 100644
--- a/yarn-project/aztec-node/src/aztec-node/server.ts
+++ b/yarn-project/aztec-node/src/aztec-node/server.ts
@@ -169,7 +169,7 @@ export class AztecNodeService implements AztecNode {
 
     const simulationProvider = await createSimulationProvider(config, log);
 
-    const prover = await createProverClient(config, worldStateSynchronizer, archiver, telemetry);
+    const prover = await createProverClient(config, telemetry);
 
     if (!prover && !config.disableSequencer) {
       throw new Error("Can't start a sequencer without a prover");
@@ -742,6 +742,7 @@ export class AztecNodeService implements AztecNode {
       this.telemetry,
     );
     const processor = publicProcessorFactory.create(prevHeader, newGlobalVariables);
+
     // REFACTOR: Consider merging ProcessReturnValues into ProcessedTx
     const [processedTxs, failedTxs, returns] = await processor.process([tx]);
     // REFACTOR: Consider returning the error/revert rather than throwing
diff --git a/yarn-project/circuit-types/src/interfaces/index.ts b/yarn-project/circuit-types/src/interfaces/index.ts
index c898f123b13..1c12714c1a3 100644
--- a/yarn-project/circuit-types/src/interfaces/index.ts
+++ b/yarn-project/circuit-types/src/interfaces/index.ts
@@ -10,3 +10,4 @@ export * from './block-prover.js';
 export * from './server_circuit_prover.js';
 export * from './private_kernel_prover.js';
 export * from './tx-provider.js';
+export * from './merkle_tree_operations.js';
diff --git a/yarn-project/world-state/src/world-state-db/merkle_tree_operations.ts b/yarn-project/circuit-types/src/interfaces/merkle_tree_operations.ts
similarity index 82%
rename from yarn-project/world-state/src/world-state-db/merkle_tree_operations.ts
rename to yarn-project/circuit-types/src/interfaces/merkle_tree_operations.ts
index ecddc43dcc6..58b29323712 100644
--- a/yarn-project/world-state/src/world-state-db/merkle_tree_operations.ts
+++ b/yarn-project/circuit-types/src/interfaces/merkle_tree_operations.ts
@@ -1,14 +1,56 @@
-import { type L2Block, type MerkleTreeId, type SiblingPath } from '@aztec/circuit-types';
 import { type Fr, type Header, type NullifierLeafPreimage, type StateReference } from '@aztec/circuits.js';
 import { createDebugLogger } from '@aztec/foundation/log';
 import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
-import { type AppendOnlyTree, type BatchInsertionResult, type IndexedTree } from '@aztec/merkle-tree';
+
+import { type L2Block } from '../l2_block.js';
+import { type MerkleTreeId } from '../merkle_tree_id.js';
+import { type SiblingPath } from '../sibling_path/sibling_path.js';
 
 /**
  * Type alias for the nullifier tree ID.
  */
 export type IndexedTreeId = MerkleTreeId.NULLIFIER_TREE | MerkleTreeId.PUBLIC_DATA_TREE;
 
+/**
+ * All of the data to be return during batch insertion.
+ */
+export interface LowLeafWitnessData<N extends number> {
+  /**
+   * Preimage of the low nullifier that proves non membership.
+   */
+  leafPreimage: IndexedTreeLeafPreimage;
+  /**
+   * Sibling path to prove membership of low nullifier.
+   */
+  siblingPath: SiblingPath<N>;
+  /**
+   * The index of low nullifier.
+   */
+  index: bigint;
+}
+
+/**
+ * The result of a batch insertion in an indexed merkle tree.
+ */
+export interface BatchInsertionResult<TreeHeight extends number, SubtreeSiblingPathHeight extends number> {
+  /**
+   * Data for the leaves to be updated when inserting the new ones.
+   */
+  lowLeavesWitnessData?: LowLeafWitnessData<TreeHeight>[];
+  /**
+   * Sibling path "pointing to" where the new subtree should be inserted into the tree.
+   */
+  newSubtreeSiblingPath: SiblingPath<SubtreeSiblingPathHeight>;
+  /**
+   * The new leaves being inserted in high to low order. This order corresponds with the order of the low leaves witness.
+   */
+  sortedNewLeaves: Buffer[];
+  /**
+   * The indexes of the sorted new leaves to the original ones.
+   */
+  sortedNewLeavesIndexes: number[];
+}
+
 /**
  *  Defines tree information.
  */
@@ -32,14 +74,6 @@ export interface TreeInfo {
   depth: number;
 }
 
-export type MerkleTreeMap = {
-  [MerkleTreeId.NULLIFIER_TREE]: IndexedTree;
-  [MerkleTreeId.NOTE_HASH_TREE]: AppendOnlyTree<Fr>;
-  [MerkleTreeId.PUBLIC_DATA_TREE]: IndexedTree;
-  [MerkleTreeId.L1_TO_L2_MESSAGE_TREE]: AppendOnlyTree<Fr>;
-  [MerkleTreeId.ARCHIVE]: AppendOnlyTree<Fr>;
-};
-
 type LeafTypes = {
   [MerkleTreeId.NULLIFIER_TREE]: Buffer;
   [MerkleTreeId.NOTE_HASH_TREE]: Fr;
diff --git a/yarn-project/circuit-types/src/interfaces/prover-client.ts b/yarn-project/circuit-types/src/interfaces/prover-client.ts
index bdf92e94e2c..e25875b67bd 100644
--- a/yarn-project/circuit-types/src/interfaces/prover-client.ts
+++ b/yarn-project/circuit-types/src/interfaces/prover-client.ts
@@ -2,6 +2,7 @@ import { type TxHash } from '@aztec/circuit-types';
 import { type Fr } from '@aztec/circuits.js';
 
 import { type BlockProver } from './block-prover.js';
+import { type MerkleTreeOperations } from './merkle_tree_operations.js';
 import { type ProvingJobSource } from './proving-job.js';
 
 /**
@@ -29,8 +30,11 @@ export type ProverConfig = {
 /**
  * The interface to the prover client.
  * Provides the ability to generate proofs and build rollups.
+ * TODO(palla/prover-node): Rename this interface
  */
-export interface ProverClient extends BlockProver {
+export interface ProverClient {
+  createBlockProver(db: MerkleTreeOperations): BlockProver;
+
   start(): Promise<void>;
 
   stop(): Promise<void>;
diff --git a/yarn-project/circuit-types/src/l2_block_downloader/l2_block_downloader.ts b/yarn-project/circuit-types/src/l2_block_downloader/l2_block_downloader.ts
index f4d412426a5..8c43ce23925 100644
--- a/yarn-project/circuit-types/src/l2_block_downloader/l2_block_downloader.ts
+++ b/yarn-project/circuit-types/src/l2_block_downloader/l2_block_downloader.ts
@@ -68,15 +68,36 @@ export class L2BlockDownloader {
   /**
    * Repeatedly queries the block source and adds the received blocks to the block queue.
    * Stops when no further blocks are received.
+   * @param targetBlockNumber - Optional block number to stop at.
+   * @param proven - Optional override of the default "proven" setting.
    * @returns The total number of blocks added to the block queue.
    */
-  private async collectBlocks() {
+  private async collectBlocks(targetBlockNumber?: number, onlyProven?: boolean) {
     let totalBlocks = 0;
     while (true) {
-      const blocks = await this.l2BlockSource.getBlocks(this.from, 10, this.proven);
+      // If we have a target and have reached it, return
+      if (targetBlockNumber !== undefined && this.from > targetBlockNumber) {
+        log.verbose(`Reached target block number ${targetBlockNumber}`);
+        return totalBlocks;
+      }
+
+      // If we have a target, then request at most the number of blocks to get to it
+      const limit = targetBlockNumber !== undefined ? Math.min(targetBlockNumber - this.from + 1, 10) : 10;
+      const proven = onlyProven === undefined ? this.proven : onlyProven;
+
+      // Hit the archiver for blocks
+      const blocks = await this.l2BlockSource.getBlocks(this.from, limit, proven);
+
+      // If there are no more blocks, return
       if (!blocks.length) {
         return totalBlocks;
       }
+
+      log.verbose(
+        `Received ${blocks.length} blocks from archiver after querying from ${this.from} limit ${limit} (proven ${proven})`,
+      );
+
+      // Push new blocks into the queue and loop
       await this.semaphore.acquire();
       this.blockQueue.put(blocks);
       this.from += blocks.length;
@@ -116,9 +137,13 @@ export class L2BlockDownloader {
 
   /**
    * Forces an immediate request for blocks.
+   * Repeatedly queries the block source and adds the received blocks to the block queue.
+   * Stops when no further blocks are received.
+   * @param targetBlockNumber - Optional block number to stop at.
+   * @param proven - Optional override of the default "proven" setting.
    * @returns A promise that fulfills once the poll is complete
    */
-  public pollImmediate(): Promise<number> {
-    return this.jobQueue.put(() => this.collectBlocks());
+  public pollImmediate(targetBlockNumber?: number, onlyProven?: boolean): Promise<number> {
+    return this.jobQueue.put(() => this.collectBlocks(targetBlockNumber, onlyProven));
   }
 }
diff --git a/yarn-project/end-to-end/src/composed/integration_l1_publisher.test.ts b/yarn-project/end-to-end/src/composed/integration_l1_publisher.test.ts
index 87c5573176d..d638d72a3d3 100644
--- a/yarn-project/end-to-end/src/composed/integration_l1_publisher.test.ts
+++ b/yarn-project/end-to-end/src/composed/integration_l1_publisher.test.ts
@@ -3,6 +3,7 @@ import { getConfigEnvVars } from '@aztec/aztec-node';
 import { AztecAddress, Body, Fr, GlobalVariables, type L2Block, createDebugLogger, mockTx } from '@aztec/aztec.js';
 // eslint-disable-next-line no-restricted-imports
 import {
+  type BlockProver,
   PROVING_STATUS,
   type ProcessedTx,
   makeEmptyProcessedTx as makeEmptyProcessedTxFromHistoricalTreeRoots,
@@ -82,6 +83,7 @@ describe('L1Publisher integration', () => {
 
   let builder: TxProver;
   let builderDb: MerkleTrees;
+  let prover: BlockProver;
 
   // The header of the last block
   let prevHeader: Header;
@@ -138,7 +140,8 @@ describe('L1Publisher integration', () => {
     };
     const worldStateSynchronizer = new ServerWorldStateSynchronizer(tmpStore, builderDb, blockSource, worldStateConfig);
     await worldStateSynchronizer.start();
-    builder = await TxProver.new(config, worldStateSynchronizer, blockSource, new NoopTelemetryClient());
+    builder = await TxProver.new(config, new NoopTelemetryClient());
+    prover = builder.createBlockProver(builderDb.asLatest());
 
     publisher = getL1Publisher(
       {
@@ -285,9 +288,9 @@ describe('L1Publisher integration', () => {
   };
 
   const buildBlock = async (globalVariables: GlobalVariables, txs: ProcessedTx[], l1ToL2Messages: Fr[]) => {
-    const blockTicket = await builder.startNewBlock(txs.length, globalVariables, l1ToL2Messages);
+    const blockTicket = await prover.startNewBlock(txs.length, globalVariables, l1ToL2Messages);
     for (const tx of txs) {
-      await builder.addNewTx(tx);
+      await prover.addNewTx(tx);
     }
     return blockTicket;
   };
@@ -360,7 +363,7 @@ describe('L1Publisher integration', () => {
       const ticket = await buildBlock(globalVariables, txs, currentL1ToL2Messages);
       const result = await ticket.provingPromise;
       expect(result.status).toBe(PROVING_STATUS.SUCCESS);
-      const blockResult = await builder.finaliseBlock();
+      const blockResult = await prover.finaliseBlock();
       const block = blockResult.block;
       prevHeader = block.header;
       blockSource.getL1ToL2Messages.mockResolvedValueOnce(currentL1ToL2Messages);
@@ -450,10 +453,10 @@ describe('L1Publisher integration', () => {
         GasFees.empty(),
       );
       const blockTicket = await buildBlock(globalVariables, txs, l1ToL2Messages);
-      await builder.setBlockCompleted();
+      await prover.setBlockCompleted();
       const result = await blockTicket.provingPromise;
       expect(result.status).toBe(PROVING_STATUS.SUCCESS);
-      const blockResult = await builder.finaliseBlock();
+      const blockResult = await prover.finaliseBlock();
       const block = blockResult.block;
       prevHeader = block.header;
       blockSource.getL1ToL2Messages.mockResolvedValueOnce(l1ToL2Messages);
diff --git a/yarn-project/end-to-end/src/e2e_prover_node.test.ts b/yarn-project/end-to-end/src/e2e_prover_node.test.ts
index 3035ff969f3..df5a6207904 100644
--- a/yarn-project/end-to-end/src/e2e_prover_node.test.ts
+++ b/yarn-project/end-to-end/src/e2e_prover_node.test.ts
@@ -15,7 +15,7 @@ import {
   sleep,
 } from '@aztec/aztec.js';
 import { StatefulTestContract, TestContract } from '@aztec/noir-contracts.js';
-import { type ProverNode, createProverNode } from '@aztec/prover-node';
+import { createProverNode } from '@aztec/prover-node';
 import { type SequencerClientConfig } from '@aztec/sequencer-client';
 
 import { sendL1ToL2Message } from './fixtures/l1_to_l2_messaging.js';
@@ -107,20 +107,12 @@ describe('e2e_prover_node', () => {
     ctx = await snapshotManager.setup();
   });
 
-  const prove = async (proverNode: ProverNode, blockNumber: number) => {
-    logger.info(`Proving block ${blockNumber}`);
-    await proverNode.prove(blockNumber, blockNumber);
-
-    logger.info(`Proof submitted. Awaiting aztec node to sync...`);
-    await retryUntil(async () => (await ctx.aztecNode.getProvenBlockNumber()) === blockNumber, 'block-1', 10, 1);
-    expect(await ctx.aztecNode.getProvenBlockNumber()).toEqual(blockNumber);
-  };
-
   it('submits three blocks, then prover proves the first two', async () => {
     // Check everything went well during setup and txs were mined in two different blocks
     const [txReceipt1, txReceipt2, txReceipt3] = txReceipts;
     const firstBlock = txReceipt1.blockNumber!;
-    expect(txReceipt2.blockNumber).toEqual(firstBlock + 1);
+    const secondBlock = firstBlock + 1;
+    expect(txReceipt2.blockNumber).toEqual(secondBlock);
     expect(txReceipt3.blockNumber).toEqual(firstBlock + 2);
     expect(await contract.methods.get_public_value(recipient).simulate()).toEqual(20n);
     expect(await contract.methods.summed_values(recipient).simulate()).toEqual(10n);
@@ -141,9 +133,18 @@ describe('e2e_prover_node', () => {
     const archiver = ctx.aztecNode.getBlockSource() as Archiver;
     const proverNode = await createProverNode(proverConfig, { aztecNodeTxProvider: ctx.aztecNode, archiver });
 
-    // Prove the first two blocks
-    await prove(proverNode, firstBlock);
-    await prove(proverNode, firstBlock + 1);
+    // Prove the first two blocks simultaneously
+    logger.info(`Starting proof for first block #${firstBlock}`);
+    await proverNode.startProof(firstBlock, firstBlock);
+    logger.info(`Starting proof for second block #${secondBlock}`);
+    await proverNode.startProof(secondBlock, secondBlock);
+
+    // Confirm that we cannot go back to prove an old one
+    await expect(proverNode.startProof(firstBlock, firstBlock)).rejects.toThrow(/behind the current world state/i);
+
+    // Await until proofs get submitted
+    await retryUntil(async () => (await ctx.aztecNode.getProvenBlockNumber()) === secondBlock, 'proven', 10, 1);
+    expect(await ctx.aztecNode.getProvenBlockNumber()).toEqual(secondBlock);
 
     // Check that the prover id made it to the emitted event
     const { publicClient, l1ContractAddresses } = ctx.deployL1ContractsValues;
diff --git a/yarn-project/kv-store/src/interfaces/store.ts b/yarn-project/kv-store/src/interfaces/store.ts
index f13a3241ab5..076d39da1e0 100644
--- a/yarn-project/kv-store/src/interfaces/store.ts
+++ b/yarn-project/kv-store/src/interfaces/store.ts
@@ -58,4 +58,9 @@ export interface AztecKVStore {
    * Clears the store
    */
   clear(): Promise<void>;
+
+  /**
+   * Forks the store.
+   */
+  fork(): Promise<AztecKVStore>;
 }
diff --git a/yarn-project/kv-store/src/lmdb/store.test.ts b/yarn-project/kv-store/src/lmdb/store.test.ts
new file mode 100644
index 00000000000..f6babd0cb67
--- /dev/null
+++ b/yarn-project/kv-store/src/lmdb/store.test.ts
@@ -0,0 +1,29 @@
+import { mkdtemp } from 'fs/promises';
+import { tmpdir } from 'os';
+import { join } from 'path';
+
+import { AztecLmdbStore } from './store.js';
+
+describe('AztecLmdbStore', () => {
+  const itForks = async (store: AztecLmdbStore) => {
+    const singleton = store.openSingleton('singleton');
+    await singleton.set('foo');
+
+    const forkedStore = await store.fork();
+    const forkedSingleton = forkedStore.openSingleton('singleton');
+    expect(forkedSingleton.get()).toEqual('foo');
+    await forkedSingleton.set('bar');
+    expect(singleton.get()).toEqual('foo');
+  };
+
+  it('forks a persistent store', async () => {
+    const path = join(await mkdtemp(join(tmpdir(), 'aztec-store-test-')), 'main.mdb');
+    const store = AztecLmdbStore.open(path, false);
+    await itForks(store);
+  });
+
+  it('forks an ephemeral store', async () => {
+    const store = AztecLmdbStore.open(undefined, true);
+    await itForks(store);
+  });
+});
diff --git a/yarn-project/kv-store/src/lmdb/store.ts b/yarn-project/kv-store/src/lmdb/store.ts
index 4b7a115f2f9..cb6ee87d7d7 100644
--- a/yarn-project/kv-store/src/lmdb/store.ts
+++ b/yarn-project/kv-store/src/lmdb/store.ts
@@ -1,6 +1,9 @@
 import { createDebugLogger } from '@aztec/foundation/log';
 
+import { mkdtemp } from 'fs/promises';
 import { type Database, type Key, type RootDatabase, open } from 'lmdb';
+import { tmpdir } from 'os';
+import { join } from 'path';
 
 import { type AztecArray } from '../interfaces/array.js';
 import { type AztecCounter } from '../interfaces/counter.js';
@@ -22,7 +25,7 @@ export class AztecLmdbStore implements AztecKVStore {
   #data: Database<unknown, Key>;
   #multiMapData: Database<unknown, Key>;
 
-  constructor(rootDb: RootDatabase) {
+  constructor(rootDb: RootDatabase, public readonly isEphemeral: boolean) {
     this.#rootDb = rootDb;
 
     // big bucket to store all the data
@@ -57,11 +60,19 @@ export class AztecLmdbStore implements AztecKVStore {
     log = createDebugLogger('aztec:kv-store:lmdb'),
   ): AztecLmdbStore {
     log.info(`Opening LMDB database at ${path || 'temporary location'}`);
-    const rootDb = open({
-      path,
-      noSync: ephemeral,
-    });
-    return new AztecLmdbStore(rootDb);
+    const rootDb = open({ path, noSync: ephemeral });
+    return new AztecLmdbStore(rootDb, ephemeral);
+  }
+
+  /**
+   * Forks the current DB into a new DB by backing it up to a temporary location and opening a new lmdb db.
+   * @returns A new AztecLmdbStore.
+   */
+  async fork() {
+    const forkPath = join(await mkdtemp(join(tmpdir(), 'aztec-store-fork-')), 'root.mdb');
+    await this.#rootDb.backup(forkPath, false);
+    const forkDb = open(forkPath, { noSync: this.isEphemeral });
+    return new AztecLmdbStore(forkDb, this.isEphemeral);
   }
 
   /**
diff --git a/yarn-project/merkle-tree/src/interfaces/indexed_tree.ts b/yarn-project/merkle-tree/src/interfaces/indexed_tree.ts
index 3e7c1b64c5c..54f30c2a61f 100644
--- a/yarn-project/merkle-tree/src/interfaces/indexed_tree.ts
+++ b/yarn-project/merkle-tree/src/interfaces/indexed_tree.ts
@@ -1,4 +1,4 @@
-import { type SiblingPath } from '@aztec/circuit-types';
+import { type BatchInsertionResult } from '@aztec/circuit-types';
 import { type IndexedTreeLeaf, type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
 
 import {
@@ -36,46 +36,6 @@ export interface PreimageFactory {
   clone(preimage: IndexedTreeLeafPreimage): IndexedTreeLeafPreimage;
 }
 
-/**
- * All of the data to be return during batch insertion.
- */
-export interface LowLeafWitnessData<N extends number> {
-  /**
-   * Preimage of the low nullifier that proves non membership.
-   */
-  leafPreimage: IndexedTreeLeafPreimage;
-  /**
-   * Sibling path to prove membership of low nullifier.
-   */
-  siblingPath: SiblingPath<N>;
-  /**
-   * The index of low nullifier.
-   */
-  index: bigint;
-}
-
-/**
- * The result of a batch insertion in an indexed merkle tree.
- */
-export interface BatchInsertionResult<TreeHeight extends number, SubtreeSiblingPathHeight extends number> {
-  /**
-   * Data for the leaves to be updated when inserting the new ones.
-   */
-  lowLeavesWitnessData?: LowLeafWitnessData<TreeHeight>[];
-  /**
-   * Sibling path "pointing to" where the new subtree should be inserted into the tree.
-   */
-  newSubtreeSiblingPath: SiblingPath<SubtreeSiblingPathHeight>;
-  /**
-   * The new leaves being inserted in high to low order. This order corresponds with the order of the low leaves witness.
-   */
-  sortedNewLeaves: Buffer[];
-  /**
-   * The indexes of the sorted new leaves to the original ones.
-   */
-  sortedNewLeavesIndexes: number[];
-}
-
 /**
  * Indexed merkle tree.
  */
diff --git a/yarn-project/merkle-tree/src/standard_indexed_tree/standard_indexed_tree.ts b/yarn-project/merkle-tree/src/standard_indexed_tree/standard_indexed_tree.ts
index 06d6d91f6c8..bacfeaa2e02 100644
--- a/yarn-project/merkle-tree/src/standard_indexed_tree/standard_indexed_tree.ts
+++ b/yarn-project/merkle-tree/src/standard_indexed_tree/standard_indexed_tree.ts
@@ -1,4 +1,4 @@
-import { SiblingPath } from '@aztec/circuit-types';
+import { type BatchInsertionResult, type LowLeafWitnessData, SiblingPath } from '@aztec/circuit-types';
 import { type TreeInsertionStats } from '@aztec/circuit-types/stats';
 import { toBufferBE } from '@aztec/foundation/bigint-buffer';
 import { type FromBuffer } from '@aztec/foundation/serialize';
@@ -7,12 +7,7 @@ import { type IndexedTreeLeaf, type IndexedTreeLeafPreimage } from '@aztec/found
 import { type AztecKVStore, type AztecMap } from '@aztec/kv-store';
 import { type Hasher } from '@aztec/types/interfaces';
 
-import {
-  type BatchInsertionResult,
-  type IndexedTree,
-  type LowLeafWitnessData,
-  type PreimageFactory,
-} from '../interfaces/indexed_tree.js';
+import { type IndexedTree, type PreimageFactory } from '../interfaces/indexed_tree.js';
 import { IndexedTreeSnapshotBuilder } from '../snapshots/indexed_tree_snapshot.js';
 import { type IndexedTreeSnapshot } from '../snapshots/snapshot_builder.js';
 import { TreeBase } from '../tree_base.js';
diff --git a/yarn-project/prover-client/src/mocks/test_context.ts b/yarn-project/prover-client/src/mocks/test_context.ts
index 99f7738f049..f1a20ccba0d 100644
--- a/yarn-project/prover-client/src/mocks/test_context.ts
+++ b/yarn-project/prover-client/src/mocks/test_context.ts
@@ -1,9 +1,7 @@
 import { type BBProverConfig } from '@aztec/bb-prover';
 import {
   type BlockProver,
-  type BlockResult,
   type ProcessedTx,
-  type ProvingTicket,
   type PublicExecutionRequest,
   type ServerCircuitProver,
   type Tx,
@@ -36,30 +34,7 @@ import { MemoryProvingQueue } from '../prover-agent/memory-proving-queue.js';
 import { ProverAgent } from '../prover-agent/prover-agent.js';
 import { getEnvironmentConfig, getSimulationProvider, makeGlobals } from './fixtures.js';
 
-class DummyProverClient implements BlockProver {
-  constructor(private orchestrator: ProvingOrchestrator) {}
-  startNewBlock(numTxs: number, globalVariables: GlobalVariables, l1ToL2Messages: Fr[]): Promise<ProvingTicket> {
-    return this.orchestrator.startNewBlock(numTxs, globalVariables, l1ToL2Messages);
-  }
-  addNewTx(tx: ProcessedTx): Promise<void> {
-    return this.orchestrator.addNewTx(tx);
-  }
-  cancelBlock(): void {
-    return this.orchestrator.cancelBlock();
-  }
-  finaliseBlock(): Promise<BlockResult> {
-    return this.orchestrator.finaliseBlock();
-  }
-  setBlockCompleted(): Promise<void> {
-    return this.orchestrator.setBlockCompleted();
-  }
-  getProverId(): Fr {
-    return this.orchestrator.proverId;
-  }
-}
-
 export class TestContext {
-  public blockProver: BlockProver;
   constructor(
     public publicExecutor: MockProxy<PublicExecutor>,
     public publicContractsDB: MockProxy<ContractsDataSourcePublicDB>,
@@ -74,8 +49,10 @@ export class TestContext {
     public blockNumber: number,
     public directoriesToCleanup: string[],
     public logger: DebugLogger,
-  ) {
-    this.blockProver = new DummyProverClient(this.orchestrator);
+  ) {}
+
+  public get blockProver() {
+    return this.orchestrator;
   }
 
   static async new(
diff --git a/yarn-project/prover-client/src/orchestrator/orchestrator.ts b/yarn-project/prover-client/src/orchestrator/orchestrator.ts
index be0f21a8460..78760bc7e3c 100644
--- a/yarn-project/prover-client/src/orchestrator/orchestrator.ts
+++ b/yarn-project/prover-client/src/orchestrator/orchestrator.ts
@@ -15,6 +15,7 @@ import {
 } from '@aztec/circuit-types';
 import {
   BlockProofError,
+  type BlockProver,
   type BlockResult,
   PROVING_STATUS,
   type ProvingResult,
@@ -93,7 +94,7 @@ const logger = createDebugLogger('aztec:prover:proving-orchestrator');
 /**
  * The orchestrator, managing the flow of recursive proving operations required to build the rollup proof tree.
  */
-export class ProvingOrchestrator {
+export class ProvingOrchestrator implements BlockProver {
   private provingState: ProvingState | undefined = undefined;
   private pendingProvingJobs: AbortController[] = [];
   private paddingTx: PaddingProcessedTx | undefined = undefined;
@@ -104,7 +105,7 @@ export class ProvingOrchestrator {
     private db: MerkleTreeOperations,
     private prover: ServerCircuitProver,
     telemetryClient: TelemetryClient,
-    public readonly proverId: Fr = Fr.ZERO,
+    private readonly proverId: Fr = Fr.ZERO,
   ) {
     this.metrics = new ProvingOrchestratorMetrics(telemetryClient, 'ProvingOrchestrator');
   }
@@ -113,6 +114,10 @@ export class ProvingOrchestrator {
     return this.metrics.tracer;
   }
 
+  public getProverId(): Fr {
+    return this.proverId;
+  }
+
   /**
    * Resets the orchestrator's cached padding tx.
    */
@@ -140,8 +145,20 @@ export class ProvingOrchestrator {
     if (!Number.isInteger(numTxs) || numTxs < 2) {
       throw new Error(`Length of txs for the block should be at least two (got ${numTxs})`);
     }
+
+    // TODO(palla/prover-node): Store block number in the db itself to make this check more reliable,
+    // and turn this warning into an exception that we throw.
+    const { blockNumber } = globalVariables;
+    const dbBlockNumber = (await this.db.getTreeInfo(MerkleTreeId.ARCHIVE)).size - 1n;
+    if (dbBlockNumber !== blockNumber.toBigInt() - 1n) {
+      logger.warn(
+        `Database is at wrong block number (starting block ${blockNumber.toBigInt()} with db at ${dbBlockNumber})`,
+      );
+    }
+
     // Cancel any currently proving block before starting a new one
     this.cancelBlock();
+
     logger.info(`Starting new block with ${numTxs} transactions`);
     // we start the block by enqueueing all of the base parity circuits
     let baseParityInputs: BaseParityInputs[] = [];
diff --git a/yarn-project/prover-client/src/tx-prover/factory.ts b/yarn-project/prover-client/src/tx-prover/factory.ts
index 6fc40d7f634..742002239a8 100644
--- a/yarn-project/prover-client/src/tx-prover/factory.ts
+++ b/yarn-project/prover-client/src/tx-prover/factory.ts
@@ -1,16 +1,9 @@
-import { type L2BlockSource } from '@aztec/circuit-types';
 import { type TelemetryClient } from '@aztec/telemetry-client';
 import { NoopTelemetryClient } from '@aztec/telemetry-client/noop';
-import { type WorldStateSynchronizer } from '@aztec/world-state';
 
 import { type ProverClientConfig } from '../config.js';
 import { TxProver } from './tx-prover.js';
 
-export function createProverClient(
-  config: ProverClientConfig,
-  worldStateSynchronizer: WorldStateSynchronizer,
-  blockSource: L2BlockSource,
-  telemetry: TelemetryClient = new NoopTelemetryClient(),
-) {
-  return config.disableProver ? undefined : TxProver.new(config, worldStateSynchronizer, blockSource, telemetry);
+export function createProverClient(config: ProverClientConfig, telemetry: TelemetryClient = new NoopTelemetryClient()) {
+  return config.disableProver ? undefined : TxProver.new(config, telemetry);
 }
diff --git a/yarn-project/prover-client/src/tx-prover/tx-prover.ts b/yarn-project/prover-client/src/tx-prover/tx-prover.ts
index f57d475d576..7ea62671292 100644
--- a/yarn-project/prover-client/src/tx-prover/tx-prover.ts
+++ b/yarn-project/prover-client/src/tx-prover/tx-prover.ts
@@ -1,16 +1,14 @@
 import { BBNativeRollupProver, TestCircuitProver } from '@aztec/bb-prover';
-import { type L2BlockSource, type ProcessedTx } from '@aztec/circuit-types';
 import {
-  type BlockResult,
+  type BlockProver,
   type ProverClient,
   type ProvingJobSource,
-  type ProvingTicket,
   type ServerCircuitProver,
 } from '@aztec/circuit-types/interfaces';
-import { Fr, type GlobalVariables } from '@aztec/circuits.js';
+import { Fr } from '@aztec/circuits.js';
 import { NativeACVMSimulator } from '@aztec/simulator';
 import { type TelemetryClient } from '@aztec/telemetry-client';
-import { type WorldStateSynchronizer } from '@aztec/world-state';
+import { type MerkleTreeOperations } from '@aztec/world-state';
 
 import { type ProverClientConfig } from '../config.js';
 import { ProvingOrchestrator } from '../orchestrator/orchestrator.js';
@@ -18,26 +16,25 @@ import { MemoryProvingQueue } from '../prover-agent/memory-proving-queue.js';
 import { ProverAgent } from '../prover-agent/prover-agent.js';
 
 /**
- * A prover accepting individual transaction requests
+ * A prover factory.
+ * TODO(palla/prover-node): Rename this class
  */
 export class TxProver implements ProverClient {
-  private orchestrator: ProvingOrchestrator;
   private queue: MemoryProvingQueue;
   private running = false;
 
   private constructor(
     private config: ProverClientConfig,
-    private worldStateSynchronizer: WorldStateSynchronizer,
     private telemetry: TelemetryClient,
     private agent?: ProverAgent,
   ) {
+    // TODO(palla/prover-node): Cache the paddingTx here, and not in each proving orchestrator,
+    // so it can be reused across multiple ones and not recomputed every time.
     this.queue = new MemoryProvingQueue(telemetry, config.proverJobTimeoutMs, config.proverJobPollIntervalMs);
-    this.orchestrator = new ProvingOrchestrator(
-      worldStateSynchronizer.getLatest(),
-      this.queue,
-      telemetry,
-      config.proverId,
-    );
+  }
+
+  public createBlockProver(db: MerkleTreeOperations): BlockProver {
+    return new ProvingOrchestrator(db, this.queue, this.telemetry, this.config.proverId);
   }
 
   public getProverId(): Fr {
@@ -57,7 +54,7 @@ export class TxProver implements ProverClient {
     }
 
     if (!this.config.realProofs && newConfig.realProofs) {
-      this.orchestrator.reset();
+      // TODO(palla/prover-node): Reset padding tx here once we cache it at this class
     }
 
     this.config = newConfig;
@@ -85,6 +82,8 @@ export class TxProver implements ProverClient {
       return;
     }
     this.running = false;
+
+    // TODO(palla/prover-node): Keep a reference to all proving orchestrators that are alive and stop them?
     await this.agent?.stop();
     await this.queue.stop();
   }
@@ -96,12 +95,7 @@ export class TxProver implements ProverClient {
    * @param worldStateSynchronizer - An instance of the world state
    * @returns An instance of the prover, constructed and started.
    */
-  public static async new(
-    config: ProverClientConfig,
-    worldStateSynchronizer: WorldStateSynchronizer,
-    blockSource: L2BlockSource,
-    telemetry: TelemetryClient,
-  ) {
+  public static async new(config: ProverClientConfig, telemetry: TelemetryClient) {
     const agent = config.proverAgentEnabled
       ? new ProverAgent(
           await TxProver.buildCircuitProver(config, telemetry),
@@ -110,7 +104,7 @@ export class TxProver implements ProverClient {
         )
       : undefined;
 
-    const prover = new TxProver(config, worldStateSynchronizer, telemetry, agent);
+    const prover = new TxProver(config, telemetry, agent);
     await prover.start();
     return prover;
   }
@@ -130,52 +124,6 @@ export class TxProver implements ProverClient {
     return new TestCircuitProver(telemetry, simulationProvider);
   }
 
-  /**
-   * Cancels any block that is currently being built and prepares for a new one to be built
-   * @param numTxs - The complete size of the block, must be a power of 2
-   * @param globalVariables - The global variables for this block
-   * @param l1ToL2Messages - The set of L1 to L2 messages to be included in this block
-   */
-  public async startNewBlock(
-    numTxs: number,
-    globalVariables: GlobalVariables,
-    newL1ToL2Messages: Fr[],
-  ): Promise<ProvingTicket> {
-    const previousBlockNumber = globalVariables.blockNumber.toNumber() - 1;
-    await this.worldStateSynchronizer.syncImmediate(previousBlockNumber);
-    return this.orchestrator.startNewBlock(numTxs, globalVariables, newL1ToL2Messages);
-  }
-
-  /**
-   * Add a processed transaction to the current block
-   * @param tx - The transaction to be added
-   */
-  public addNewTx(tx: ProcessedTx): Promise<void> {
-    return this.orchestrator.addNewTx(tx);
-  }
-
-  /**
-   * Cancels the block currently being proven. Proofs already bring built may continue but further proofs should not be started.
-   */
-  public cancelBlock(): void {
-    this.orchestrator.cancelBlock();
-  }
-
-  /**
-   * Performs the final archive tree insertion for this block and returns the L2Block and Proof instances
-   */
-  public finaliseBlock(): Promise<BlockResult> {
-    return this.orchestrator.finaliseBlock();
-  }
-
-  /**
-   * Mark the block as having all the transactions it is going to contain.
-   * Will pad the block to it's complete size with empty transactions and prove all the way to the root rollup.
-   */
-  public setBlockCompleted(): Promise<void> {
-    return this.orchestrator.setBlockCompleted();
-  }
-
   public getProvingJobSource(): ProvingJobSource {
     return this.queue;
   }
diff --git a/yarn-project/prover-node/src/factory.ts b/yarn-project/prover-node/src/factory.ts
index 26a8bd6cf69..f1e966de091 100644
--- a/yarn-project/prover-node/src/factory.ts
+++ b/yarn-project/prover-node/src/factory.ts
@@ -4,7 +4,7 @@ import { type DebugLogger, createDebugLogger } from '@aztec/foundation/log';
 import { createStore } from '@aztec/kv-store/utils';
 import { createProverClient } from '@aztec/prover-client';
 import { getL1Publisher } from '@aztec/sequencer-client';
-import { PublicProcessorFactory, createSimulationProvider } from '@aztec/simulator';
+import { createSimulationProvider } from '@aztec/simulator';
 import { type TelemetryClient } from '@aztec/telemetry-client';
 import { NoopTelemetryClient } from '@aztec/telemetry-client/noop';
 import { createWorldStateSynchronizer } from '@aztec/world-state';
@@ -40,17 +40,24 @@ export async function createProverNode(
 
   const simulationProvider = await createSimulationProvider(config, log);
 
-  const prover = await createProverClient(config, worldStateSynchronizer, archiver);
+  const prover = await createProverClient(config, telemetry);
 
   // REFACTOR: Move publisher out of sequencer package and into an L1-related package
   const publisher = getL1Publisher(config, telemetry);
 
-  const latestWorldState = worldStateSynchronizer.getLatest();
-  const publicProcessorFactory = new PublicProcessorFactory(latestWorldState, archiver, simulationProvider, telemetry);
-
   const txProvider = deps.aztecNodeTxProvider
     ? new AztecNodeTxProvider(deps.aztecNodeTxProvider)
     : createTxProvider(config);
 
-  return new ProverNode(prover!, publicProcessorFactory, publisher, archiver, archiver, txProvider);
+  return new ProverNode(
+    prover!,
+    publisher,
+    archiver,
+    archiver,
+    archiver,
+    worldStateSynchronizer,
+    txProvider,
+    simulationProvider,
+    telemetry,
+  );
 }
diff --git a/yarn-project/prover-node/src/job/block-proving-job.ts b/yarn-project/prover-node/src/job/block-proving-job.ts
index 37c5989a963..111651868aa 100644
--- a/yarn-project/prover-node/src/job/block-proving-job.ts
+++ b/yarn-project/prover-node/src/job/block-proving-job.ts
@@ -42,66 +42,70 @@ export class BlockProvingJob {
     }
 
     this.log.info(`Starting block proving job`, { fromBlock, toBlock });
-    this.state = 'started';
-
-    // TODO: Fast-forward world state to fromBlock and/or await fromBlock to be published to the unproven chain
-
     this.state = 'processing';
-
-    let historicalHeader = (await this.l2BlockSource.getBlock(fromBlock - 1))?.header;
-    for (let blockNumber = fromBlock; blockNumber <= toBlock; blockNumber++) {
-      const block = await this.getBlock(blockNumber);
-      const globalVariables = block.header.globalVariables;
-      const txHashes = block.body.txEffects.map(tx => tx.txHash);
-      const txCount = block.body.numberOfTxsIncludingPadded;
-      const l1ToL2Messages = await this.getL1ToL2Messages(block);
-
-      this.log.verbose(`Starting block processing`, {
-        number: block.number,
-        blockHash: block.hash().toString(),
-        lastArchive: block.header.lastArchive.root,
-        noteHashTreeRoot: block.header.state.partial.noteHashTree.root,
-        nullifierTreeRoot: block.header.state.partial.nullifierTree.root,
-        publicDataTreeRoot: block.header.state.partial.publicDataTree.root,
-        historicalHeader: historicalHeader?.hash(),
-        ...globalVariables,
-      });
-      const provingTicket = await this.prover.startNewBlock(txCount, globalVariables, l1ToL2Messages);
-      const publicProcessor = this.publicProcessorFactory.create(historicalHeader, globalVariables);
-
-      const txs = await this.getTxs(txHashes);
-      await this.processTxs(publicProcessor, txs, txCount);
-
-      this.log.verbose(`Processed all txs for block`, {
-        blockNumber: block.number,
-        blockHash: block.hash().toString(),
-      });
-
-      await this.prover.setBlockCompleted();
-
-      const result = await provingTicket.provingPromise;
-      if (result.status === PROVING_STATUS.FAILURE) {
-        throw new Error(`Block proving failed: ${result.reason}`);
+    try {
+      let historicalHeader = (await this.l2BlockSource.getBlock(fromBlock - 1))?.header;
+      for (let blockNumber = fromBlock; blockNumber <= toBlock; blockNumber++) {
+        const block = await this.getBlock(blockNumber);
+        const globalVariables = block.header.globalVariables;
+        const txHashes = block.body.txEffects.map(tx => tx.txHash);
+        const txCount = block.body.numberOfTxsIncludingPadded;
+        const l1ToL2Messages = await this.getL1ToL2Messages(block);
+
+        this.log.verbose(`Starting block processing`, {
+          number: block.number,
+          blockHash: block.hash().toString(),
+          lastArchive: block.header.lastArchive.root,
+          noteHashTreeRoot: block.header.state.partial.noteHashTree.root,
+          nullifierTreeRoot: block.header.state.partial.nullifierTree.root,
+          publicDataTreeRoot: block.header.state.partial.publicDataTree.root,
+          historicalHeader: historicalHeader?.hash(),
+          ...globalVariables,
+        });
+
+        // When we move to proving epochs, this should change into a startNewEpoch and be lifted outside the loop.
+        const provingTicket = await this.prover.startNewBlock(txCount, globalVariables, l1ToL2Messages);
+
+        const publicProcessor = this.publicProcessorFactory.create(historicalHeader, globalVariables);
+
+        const txs = await this.getTxs(txHashes);
+        await this.processTxs(publicProcessor, txs, txCount);
+
+        this.log.verbose(`Processed all txs for block`, {
+          blockNumber: block.number,
+          blockHash: block.hash().toString(),
+        });
+
+        await this.prover.setBlockCompleted();
+
+        // This should be moved outside the loop to match the creation of the proving ticket when we move to epochs.
+        this.state = 'awaiting-prover';
+        const result = await provingTicket.provingPromise;
+        if (result.status === PROVING_STATUS.FAILURE) {
+          throw new Error(`Block proving failed: ${result.reason}`);
+        }
+
+        historicalHeader = block.header;
       }
 
-      historicalHeader = block.header;
-    }
+      const { block, aggregationObject, proof } = await this.prover.finaliseBlock();
+      this.log.info(`Finalised proof for block range`, { fromBlock, toBlock });
 
-    this.state = 'awaiting-prover';
-    const { block, aggregationObject, proof } = await this.prover.finaliseBlock();
-    this.log.info(`Finalised proof for block range`, { fromBlock, toBlock });
-
-    this.state = 'publishing-proof';
-    await this.publisher.submitProof(
-      block.header,
-      block.archive.root,
-      this.prover.getProverId(),
-      aggregationObject,
-      proof,
-    );
-    this.log.info(`Submitted proof for block range`, { fromBlock, toBlock });
+      this.state = 'publishing-proof';
+      await this.publisher.submitProof(
+        block.header,
+        block.archive.root,
+        this.prover.getProverId(),
+        aggregationObject,
+        proof,
+      );
+      this.log.info(`Submitted proof for block range`, { fromBlock, toBlock });
 
-    this.state = 'completed';
+      this.state = 'completed';
+    } catch (err) {
+      this.log.error(`Error running block prover job: ${err}`);
+      this.state = 'failed';
+    }
   }
 
   private async getBlock(blockNumber: number): Promise<L2Block> {
@@ -151,8 +155,8 @@ export class BlockProvingJob {
 
 export type BlockProvingJobState =
   | 'initialized'
-  | 'started'
   | 'processing'
   | 'awaiting-prover'
   | 'publishing-proof'
-  | 'completed';
+  | 'completed'
+  | 'failed';
diff --git a/yarn-project/prover-node/src/prover-node.ts b/yarn-project/prover-node/src/prover-node.ts
index 1c435e3cfa3..ca4b3042913 100644
--- a/yarn-project/prover-node/src/prover-node.ts
+++ b/yarn-project/prover-node/src/prover-node.ts
@@ -2,8 +2,11 @@ import { type L1ToL2MessageSource, type L2BlockSource, type ProverClient, type T
 import { createDebugLogger } from '@aztec/foundation/log';
 import { RunningPromise } from '@aztec/foundation/running-promise';
 import { type L1Publisher } from '@aztec/sequencer-client';
-import { type PublicProcessorFactory } from '@aztec/simulator';
+import { PublicProcessorFactory, type SimulationProvider } from '@aztec/simulator';
+import { type TelemetryClient } from '@aztec/telemetry-client';
+import { type WorldStateSynchronizer } from '@aztec/world-state';
 
+import { type ContractDataSource } from '../../types/src/contracts/contract_data_source.js';
 import { BlockProvingJob } from './job/block-proving-job.js';
 
 /**
@@ -14,14 +17,18 @@ import { BlockProvingJob } from './job/block-proving-job.js';
 export class ProverNode {
   private log = createDebugLogger('aztec:prover-node');
   private runningPromise: RunningPromise | undefined;
+  private latestBlockWeAreProving: number | undefined;
 
   constructor(
     private prover: ProverClient,
-    private publicProcessorFactory: PublicProcessorFactory,
     private publisher: L1Publisher,
     private l2BlockSource: L2BlockSource,
     private l1ToL2MessageSource: L1ToL2MessageSource,
+    private contractDataSource: ContractDataSource,
+    private worldState: WorldStateSynchronizer,
     private txProvider: TxProvider,
+    private simulator: SimulationProvider,
+    private telemetryClient: TelemetryClient,
     private options: { pollingIntervalMs: number; disableAutomaticProving: boolean } = {
       pollingIntervalMs: 1_000,
       disableAutomaticProving: false,
@@ -48,12 +55,12 @@ export class ProverNode {
     await this.l2BlockSource.stop();
     this.publisher.interrupt();
     this.log.info('Stopped ProverNode');
+    // TODO(palla/prover-node): Keep a reference to all ongoing ProvingJobs and stop them.
   }
 
   /**
    * Single iteration of recurring work. This method is called periodically by the running promise.
    * Checks whether there are new blocks to prove, proves them, and submits them.
-   * Only proves one block per job and one job at a time (for now).
    */
   protected async work() {
     if (this.options.disableAutomaticProving) {
@@ -65,29 +72,35 @@ export class ProverNode {
       this.l2BlockSource.getProvenBlockNumber(),
     ]);
 
-    if (latestProvenBlockNumber >= latestBlockNumber) {
-      this.log.debug(`No new blocks to prove`, { latestBlockNumber, latestProvenBlockNumber });
+    // Consider both the latest block we are proving and the last block proven on the chain
+    const latestBlockBeingProven = this.latestBlockWeAreProving ?? 0;
+    const latestProven = Math.max(latestBlockBeingProven, latestProvenBlockNumber);
+    if (latestProven >= latestBlockNumber) {
+      this.log.debug(`No new blocks to prove`, { latestBlockNumber, latestProvenBlockNumber, latestBlockBeingProven });
       return;
     }
 
-    const fromBlock = latestProvenBlockNumber + 1;
+    const fromBlock = latestProven + 1;
     const toBlock = fromBlock; // We only prove one block at a time for now
-    await this.prove(fromBlock, toBlock);
+
+    await this.startProof(fromBlock, toBlock);
+    this.latestBlockWeAreProving = toBlock;
   }
 
   /**
    * Creates a proof for a block range. Returns once the proof has been submitted to L1.
    */
-  public prove(fromBlock: number, toBlock: number) {
-    return this.createProvingJob().run(fromBlock, toBlock);
+  public async prove(fromBlock: number, toBlock: number) {
+    const job = await this.createProvingJob(fromBlock);
+    return job.run(fromBlock, toBlock);
   }
 
   /**
    * Starts a proving process and returns immediately.
    */
-  public startProof(fromBlock: number, toBlock: number) {
-    void this.createProvingJob().run(fromBlock, toBlock);
-    return Promise.resolve();
+  public async startProof(fromBlock: number, toBlock: number) {
+    const job = await this.createProvingJob(fromBlock);
+    void job.run(fromBlock, toBlock);
   }
 
   /**
@@ -97,10 +110,25 @@ export class ProverNode {
     return this.prover;
   }
 
-  private createProvingJob() {
+  private async createProvingJob(fromBlock: number) {
+    if ((await this.worldState.status()).syncedToL2Block >= fromBlock) {
+      throw new Error(`Cannot create proving job for block ${fromBlock} as it is behind the current world state`);
+    }
+
+    // Fast forward world state to right before the target block and get a fork
+    const db = await this.worldState.syncImmediateAndFork(fromBlock - 1, true);
+
+    // Create a processor using the forked world state
+    const publicProcessorFactory = new PublicProcessorFactory(
+      db,
+      this.contractDataSource,
+      this.simulator,
+      this.telemetryClient,
+    );
+
     return new BlockProvingJob(
-      this.prover,
-      this.publicProcessorFactory,
+      this.prover.createBlockProver(db),
+      publicProcessorFactory,
       this.publisher,
       this.l2BlockSource,
       this.l1ToL2MessageSource,
diff --git a/yarn-project/sequencer-client/src/client/sequencer-client.ts b/yarn-project/sequencer-client/src/client/sequencer-client.ts
index db9a3704978..dcb72921a15 100644
--- a/yarn-project/sequencer-client/src/client/sequencer-client.ts
+++ b/yarn-project/sequencer-client/src/client/sequencer-client.ts
@@ -1,5 +1,5 @@
 import { type L1ToL2MessageSource, type L2BlockSource } from '@aztec/circuit-types';
-import { type BlockProver } from '@aztec/circuit-types/interfaces';
+import { type ProverClient } from '@aztec/circuit-types/interfaces';
 import { type P2P } from '@aztec/p2p';
 import { PublicProcessorFactory, type SimulationProvider } from '@aztec/simulator';
 import { type TelemetryClient } from '@aztec/telemetry-client';
@@ -37,7 +37,7 @@ export class SequencerClient {
     contractDataSource: ContractDataSource,
     l2BlockSource: L2BlockSource,
     l1ToL2MessageSource: L1ToL2MessageSource,
-    prover: BlockProver,
+    prover: ProverClient,
     simulationProvider: SimulationProvider,
     telemetryClient: TelemetryClient,
   ) {
diff --git a/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts b/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts
index f21cb1eec8e..912d9ebe6b1 100644
--- a/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts
+++ b/yarn-project/sequencer-client/src/sequencer/sequencer.test.ts
@@ -1,4 +1,5 @@
 import {
+  type BlockProver,
   type L1ToL2MessageSource,
   L2Block,
   type L2BlockSource,
@@ -42,6 +43,7 @@ describe('sequencer', () => {
   let globalVariableBuilder: MockProxy<GlobalVariableBuilder>;
   let p2p: MockProxy<P2P>;
   let worldState: MockProxy<WorldStateSynchronizer>;
+  let blockProver: MockProxy<BlockProver>;
   let proverClient: MockProxy<ProverClient>;
   let merkleTreeOps: MockProxy<MerkleTreeOperations>;
   let publicProcessor: MockProxy<PublicProcessor>;
@@ -67,7 +69,11 @@ describe('sequencer', () => {
 
     globalVariableBuilder = mock<GlobalVariableBuilder>();
     merkleTreeOps = mock<MerkleTreeOperations>();
-    proverClient = mock<ProverClient>();
+    blockProver = mock<BlockProver>();
+
+    proverClient = mock<ProverClient>({
+      createBlockProver: () => blockProver,
+    });
 
     p2p = mock<P2P>({
       getStatus: () => Promise.resolve({ state: P2PClientState.IDLE, syncedToL2Block: lastBlockNumber }),
@@ -87,7 +93,7 @@ describe('sequencer', () => {
     });
 
     publicProcessorFactory = mock<PublicProcessorFactory>({
-      create: (_a, _b_) => publicProcessor,
+      create: (_a, _b) => publicProcessor,
     });
 
     l2BlockSource = mock<L2BlockSource>({
@@ -132,8 +138,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce([tx]);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -151,7 +157,7 @@ describe('sequencer', () => {
     await sequencer.initialSync();
     await sequencer.work();
 
-    expect(proverClient.startNewBlock).toHaveBeenCalledWith(
+    expect(blockProver.startNewBlock).toHaveBeenCalledWith(
       2,
       new GlobalVariables(
         chainId,
@@ -166,7 +172,7 @@ describe('sequencer', () => {
       Array(NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP).fill(new Fr(0n)),
     );
     expect(publisher.processL2Block).toHaveBeenCalledWith(block);
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(0);
+    expect(blockProver.cancelBlock).toHaveBeenCalledTimes(0);
   });
 
   it('builds a block when it is their turn', async () => {
@@ -182,8 +188,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce([tx]);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -202,12 +208,12 @@ describe('sequencer', () => {
     publisher.isItMyTurnToSubmit.mockClear().mockResolvedValue(false);
     await sequencer.initialSync();
     await sequencer.work();
-    expect(proverClient.startNewBlock).not.toHaveBeenCalled();
+    expect(blockProver.startNewBlock).not.toHaveBeenCalled();
 
     // Now it is!
     publisher.isItMyTurnToSubmit.mockClear().mockResolvedValue(true);
     await sequencer.work();
-    expect(proverClient.startNewBlock).toHaveBeenCalledWith(
+    expect(blockProver.startNewBlock).toHaveBeenCalledWith(
       2,
       new GlobalVariables(
         chainId,
@@ -222,7 +228,7 @@ describe('sequencer', () => {
       Array(NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP).fill(new Fr(0n)),
     );
     expect(publisher.processL2Block).toHaveBeenCalledWith(block);
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(0);
+    expect(blockProver.cancelBlock).toHaveBeenCalledTimes(0);
   });
 
   it('builds a block out of several txs rejecting double spends', async () => {
@@ -241,8 +247,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce(txs);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -268,7 +274,7 @@ describe('sequencer', () => {
     await sequencer.initialSync();
     await sequencer.work();
 
-    expect(proverClient.startNewBlock).toHaveBeenCalledWith(
+    expect(blockProver.startNewBlock).toHaveBeenCalledWith(
       2,
       new GlobalVariables(
         chainId,
@@ -284,7 +290,7 @@ describe('sequencer', () => {
     );
     expect(publisher.processL2Block).toHaveBeenCalledWith(block);
     expect(p2p.deleteTxs).toHaveBeenCalledWith([doubleSpendTx.getTxHash()]);
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(0);
+    expect(blockProver.cancelBlock).toHaveBeenCalledTimes(0);
   });
 
   it('builds a block out of several txs rejecting incorrect chain ids', async () => {
@@ -303,8 +309,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce(txs);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -325,7 +331,7 @@ describe('sequencer', () => {
     await sequencer.initialSync();
     await sequencer.work();
 
-    expect(proverClient.startNewBlock).toHaveBeenCalledWith(
+    expect(blockProver.startNewBlock).toHaveBeenCalledWith(
       2,
       new GlobalVariables(
         chainId,
@@ -341,7 +347,7 @@ describe('sequencer', () => {
     );
     expect(publisher.processL2Block).toHaveBeenCalledWith(block);
     expect(p2p.deleteTxs).toHaveBeenCalledWith([invalidChainTx.getTxHash()]);
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(0);
+    expect(blockProver.cancelBlock).toHaveBeenCalledTimes(0);
   });
 
   it('builds a block out of several txs dropping the ones that go over max size', async () => {
@@ -359,8 +365,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce(txs);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -382,7 +388,7 @@ describe('sequencer', () => {
     await sequencer.initialSync();
     await sequencer.work();
 
-    expect(proverClient.startNewBlock).toHaveBeenCalledWith(
+    expect(blockProver.startNewBlock).toHaveBeenCalledWith(
       2,
       new GlobalVariables(
         chainId,
@@ -397,7 +403,7 @@ describe('sequencer', () => {
       Array(NUMBER_OF_L1_L2_MESSAGES_PER_ROLLUP).fill(new Fr(0n)),
     );
     expect(publisher.processL2Block).toHaveBeenCalledWith(block);
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(0);
+    expect(blockProver.cancelBlock).toHaveBeenCalledTimes(0);
   });
 
   it('aborts building a block if the chain moves underneath it', async () => {
@@ -413,8 +419,8 @@ describe('sequencer', () => {
     };
 
     p2p.getTxs.mockReturnValueOnce([tx]);
-    proverClient.startNewBlock.mockResolvedValueOnce(ticket);
-    proverClient.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
+    blockProver.startNewBlock.mockResolvedValueOnce(ticket);
+    blockProver.finaliseBlock.mockResolvedValue({ block, aggregationObject: [], proof });
     publisher.processL2Block.mockResolvedValueOnce(true);
     globalVariableBuilder.buildGlobalVariables.mockResolvedValueOnce(
       new GlobalVariables(
@@ -442,7 +448,6 @@ describe('sequencer', () => {
     await sequencer.work();
 
     expect(publisher.processL2Block).not.toHaveBeenCalled();
-    expect(proverClient.cancelBlock).toHaveBeenCalledTimes(1);
   });
 });
 
diff --git a/yarn-project/sequencer-client/src/sequencer/sequencer.ts b/yarn-project/sequencer-client/src/sequencer/sequencer.ts
index 2a08c56dae2..996ca295bca 100644
--- a/yarn-project/sequencer-client/src/sequencer/sequencer.ts
+++ b/yarn-project/sequencer-client/src/sequencer/sequencer.ts
@@ -9,8 +9,8 @@ import {
 import {
   type AllowedElement,
   BlockProofError,
-  type BlockProver,
   PROVING_STATUS,
+  type ProverClient,
 } from '@aztec/circuit-types/interfaces';
 import { type L2BlockBuiltStats } from '@aztec/circuit-types/stats';
 import { AztecAddress, EthAddress, type GlobalVariables, type Header } from '@aztec/circuits.js';
@@ -61,7 +61,7 @@ export class Sequencer {
     private globalsBuilder: GlobalVariableBuilder,
     private p2pClient: P2P,
     private worldState: WorldStateSynchronizer,
-    private prover: BlockProver,
+    private prover: ProverClient,
     private l2BlockSource: L2BlockSource,
     private l1ToL2MessageSource: L1ToL2MessageSource,
     private publicProcessorFactory: PublicProcessorFactory,
@@ -264,8 +264,6 @@ export class Sequencer {
         await this.p2pClient.deleteTxs(txHashes);
       }
       this.log.error(`Rolling back world state DB due to error assembling block`, (err as any).stack);
-      // Cancel any further proving on the block
-      this.prover?.cancelBlock();
       await this.worldState.getLatest().rollback();
     }
   }
@@ -315,10 +313,11 @@ export class Sequencer {
     const blockSize = Math.max(2, numRealTxs);
 
     const blockBuildingTimer = new Timer();
-    const blockTicket = await this.prover.startNewBlock(blockSize, newGlobalVariables, l1ToL2Messages);
+    const prover = this.prover.createBlockProver(this.worldState.getLatest());
+    const blockTicket = await prover.startNewBlock(blockSize, newGlobalVariables, l1ToL2Messages);
 
     const [publicProcessorDuration, [processedTxs, failedTxs]] = await elapsed(() =>
-      processor.process(validTxs, blockSize, this.prover, this.txValidatorFactory.validatorForProcessedTxs()),
+      processor.process(validTxs, blockSize, prover, this.txValidatorFactory.validatorForProcessedTxs()),
     );
     if (failedTxs.length > 0) {
       const failedTxData = failedTxs.map(fail => fail.tx);
@@ -332,14 +331,14 @@ export class Sequencer {
     // we should bail.
     if (processedTxs.length === 0 && !this.skipMinTxsPerBlockCheck(elapsedSinceLastBlock)) {
       this.log.verbose('No txs processed correctly to build block. Exiting');
-      this.prover.cancelBlock();
+      prover.cancelBlock();
       return;
     }
 
     await assertBlockHeight();
 
     // All real transactions have been added, set the block as full and complete the proving.
-    await this.prover.setBlockCompleted();
+    await prover.setBlockCompleted();
 
     // Here we are now waiting for the block to be proven.
     // TODO(@PhilWindle) We should probably periodically check for things like another
@@ -352,7 +351,7 @@ export class Sequencer {
     await assertBlockHeight();
 
     // Block is proven, now finalise and publish!
-    const { block, aggregationObject, proof } = await this.prover.finaliseBlock();
+    const { block, aggregationObject, proof } = await prover.finaliseBlock();
 
     await assertBlockHeight();
 
@@ -381,7 +380,7 @@ export class Sequencer {
       await this.publisher.submitProof(
         block.header,
         block.archive.root,
-        this.prover.getProverId(),
+        prover.getProverId(),
         aggregationObject,
         proof,
       );
diff --git a/yarn-project/simulator/src/public/hints_builder.ts b/yarn-project/simulator/src/public/hints_builder.ts
index 7846f10f152..5b0ad19713e 100644
--- a/yarn-project/simulator/src/public/hints_builder.ts
+++ b/yarn-project/simulator/src/public/hints_builder.ts
@@ -1,4 +1,4 @@
-import { MerkleTreeId } from '@aztec/circuit-types';
+import { type IndexedTreeId, MerkleTreeId } from '@aztec/circuit-types';
 import {
   type Fr,
   type MAX_NULLIFIERS_PER_TX,
@@ -23,7 +23,7 @@ import {
   buildSiloedNullifierReadRequestHints,
 } from '@aztec/circuits.js';
 import { type Tuple } from '@aztec/foundation/serialize';
-import { type IndexedTreeId, type MerkleTreeOperations } from '@aztec/world-state';
+import { type MerkleTreeOperations } from '@aztec/world-state';
 
 export class HintsBuilder {
   constructor(private db: MerkleTreeOperations) {}
diff --git a/yarn-project/simulator/src/public/public_processor.test.ts b/yarn-project/simulator/src/public/public_processor.test.ts
index ec406ecb348..1b79581195b 100644
--- a/yarn-project/simulator/src/public/public_processor.test.ts
+++ b/yarn-project/simulator/src/public/public_processor.test.ts
@@ -4,6 +4,7 @@ import {
   PublicDataWrite,
   PublicKernelType,
   SimulationError,
+  type TreeInfo,
   type TxValidator,
   mockTx,
   toTxEffect,
@@ -43,7 +44,7 @@ import {
   computeFeePayerBalanceLeafSlot,
 } from '@aztec/simulator';
 import { NoopTelemetryClient } from '@aztec/telemetry-client/noop';
-import { type MerkleTreeOperations, type TreeInfo } from '@aztec/world-state';
+import { type MerkleTreeOperations } from '@aztec/world-state';
 
 import { jest } from '@jest/globals';
 import { type MockProxy, mock } from 'jest-mock-extended';
diff --git a/yarn-project/simulator/src/public/public_processor.ts b/yarn-project/simulator/src/public/public_processor.ts
index 995e46ac032..f2fbdf92352 100644
--- a/yarn-project/simulator/src/public/public_processor.ts
+++ b/yarn-project/simulator/src/public/public_processor.ts
@@ -60,24 +60,24 @@ export class PublicProcessorFactory {
    * Creates a new instance of a PublicProcessor.
    * @param historicalHeader - The header of a block previous to the one in which the tx is included.
    * @param globalVariables - The global variables for the block being processed.
-   * @param newContracts - Provides access to contract bytecode for public executions.
    * @returns A new instance of a PublicProcessor.
    */
-  public create(historicalHeader: Header | undefined, globalVariables: GlobalVariables): PublicProcessor {
-    historicalHeader = historicalHeader ?? this.merkleTree.getInitialHeader();
-
+  public create(maybeHistoricalHeader: Header | undefined, globalVariables: GlobalVariables): PublicProcessor {
+    const { merkleTree, telemetryClient } = this;
+    const historicalHeader = maybeHistoricalHeader ?? merkleTree.getInitialHeader();
     const publicContractsDB = new ContractsDataSourcePublicDB(this.contractDataSource);
-    const worldStatePublicDB = new WorldStatePublicDB(this.merkleTree);
-    const worldStateDB = new WorldStateDB(this.merkleTree);
+
+    const worldStatePublicDB = new WorldStatePublicDB(merkleTree);
+    const worldStateDB = new WorldStateDB(merkleTree);
     const publicExecutor = new PublicExecutor(
       worldStatePublicDB,
       publicContractsDB,
       worldStateDB,
       historicalHeader,
-      this.telemetryClient,
+      telemetryClient,
     );
     return new PublicProcessor(
-      this.merkleTree,
+      merkleTree,
       publicExecutor,
       new RealPublicKernelCircuitSimulator(this.simulator),
       globalVariables,
diff --git a/yarn-project/simulator/src/public/setup_phase_manager.test.ts b/yarn-project/simulator/src/public/setup_phase_manager.test.ts
index 41bfa9ea5a7..a765c59e984 100644
--- a/yarn-project/simulator/src/public/setup_phase_manager.test.ts
+++ b/yarn-project/simulator/src/public/setup_phase_manager.test.ts
@@ -1,7 +1,7 @@
-import { mockTx } from '@aztec/circuit-types';
+import { type TreeInfo, mockTx } from '@aztec/circuit-types';
 import { GlobalVariables, Header } from '@aztec/circuits.js';
 import { type PublicExecutor } from '@aztec/simulator';
-import { type MerkleTreeOperations, type TreeInfo } from '@aztec/world-state';
+import { type MerkleTreeOperations } from '@aztec/world-state';
 
 import { it } from '@jest/globals';
 import { type MockProxy, mock } from 'jest-mock-extended';
diff --git a/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts b/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts
index d7ee9504f71..fd99f6ba9ce 100644
--- a/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts
+++ b/yarn-project/world-state/src/synchronizer/server_world_state_synchronizer.ts
@@ -1,19 +1,22 @@
-import { type L1ToL2MessageSource, type L2Block, L2BlockDownloader, type L2BlockSource } from '@aztec/circuit-types';
+import {
+  type HandleL2BlockAndMessagesResult,
+  type L1ToL2MessageSource,
+  type L2Block,
+  L2BlockDownloader,
+  type L2BlockSource,
+} from '@aztec/circuit-types';
 import { type L2BlockHandledStats } from '@aztec/circuit-types/stats';
 import { L1_TO_L2_MSG_SUBTREE_HEIGHT } from '@aztec/circuits.js/constants';
 import { Fr } from '@aztec/foundation/fields';
 import { SerialQueue } from '@aztec/foundation/fifo';
 import { createDebugLogger } from '@aztec/foundation/log';
+import { promiseWithResolvers } from '@aztec/foundation/promise';
 import { elapsed } from '@aztec/foundation/timer';
 import { type AztecKVStore, type AztecSingleton } from '@aztec/kv-store';
 import { openTmpStore } from '@aztec/kv-store/utils';
 import { SHA256Trunc, StandardTree } from '@aztec/merkle-tree';
 
-import {
-  type HandleL2BlockAndMessagesResult,
-  type MerkleTreeOperations,
-  type MerkleTrees,
-} from '../world-state-db/index.js';
+import { type MerkleTreeOperations, type MerkleTrees } from '../world-state-db/index.js';
 import { MerkleTreeOperationsFacade } from '../world-state-db/merkle_tree_operations_facade.js';
 import { MerkleTreeSnapshotOperationsFacade } from '../world-state-db/merkle_tree_snapshot_operations_facade.js';
 import { type WorldStateConfig } from './config.js';
@@ -31,12 +34,16 @@ import {
 export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
   private latestBlockNumberAtStart = 0;
 
+  // TODO(palla/prover-node): JobQueue, stopping, runningPromise, pausedPromise, pausedResolve
+  // should all be hidden under a single abstraction. Also, check if we actually need the jobqueue.
   private l2BlockDownloader: L2BlockDownloader;
   private syncPromise: Promise<void> = Promise.resolve();
   private syncResolve?: () => void = undefined;
   private jobQueue = new SerialQueue();
   private stopping = false;
   private runningPromise: Promise<void> = Promise.resolve();
+  private pausedPromise?: Promise<void> = undefined;
+  private pausedResolve?: () => void = undefined;
   private currentState: WorldStateRunningState = WorldStateRunningState.IDLE;
   private blockNumber: AztecSingleton<number>;
 
@@ -67,6 +74,11 @@ export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
     return new MerkleTreeSnapshotOperationsFacade(this.merkleTreeDb, blockNumber);
   }
 
+  private async getFork(includeUncommitted: boolean): Promise<MerkleTreeOperationsFacade> {
+    this.log.verbose(`Forking world state at ${this.blockNumber.get()}`);
+    return new MerkleTreeOperationsFacade(await this.merkleTreeDb.fork(), includeUncommitted);
+  }
+
   public async start() {
     if (this.currentState === WorldStateRunningState.STOPPED) {
       throw new Error('Synchronizer already stopped');
@@ -102,6 +114,9 @@ export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
     const blockProcess = async () => {
       while (!this.stopping) {
         await this.jobQueue.put(() => this.collectAndProcessBlocks());
+        if (this.pausedPromise) {
+          await this.pausedPromise;
+        }
       }
     };
     this.jobQueue.start();
@@ -129,6 +144,23 @@ export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
     return this.blockNumber.get() ?? 0;
   }
 
+  private async pause() {
+    this.log.debug('Pausing world state synchronizer');
+    ({ promise: this.pausedPromise, resolve: this.pausedResolve } = promiseWithResolvers());
+    await this.jobQueue.syncPoint();
+    this.log.debug('Paused world state synchronizer');
+  }
+
+  private resume() {
+    if (this.pausedResolve) {
+      this.log.debug('Resuming world state synchronizer');
+      this.pausedResolve();
+      this.pausedResolve = undefined;
+      this.pausedPromise = undefined;
+      this.log.debug('Resumed world state synchronizer');
+    }
+  }
+
   public status(): Promise<WorldStateStatus> {
     const status = {
       syncedToL2Block: this.currentL2BlockNum,
@@ -138,30 +170,29 @@ export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
   }
 
   /**
-   * Forces an immediate sync
-   * @param minBlockNumber - The minimum block number that we must sync to
+   * Forces an immediate sync.
+   * @param targetBlockNumber - The target block number that we must sync to. Will download unproven blocks if needed to reach it. Throws if cannot be reached.
    * @returns A promise that resolves with the block number the world state was synced to
    */
-  public async syncImmediate(minBlockNumber?: number): Promise<number> {
+  public async syncImmediate(targetBlockNumber?: number): Promise<number> {
     if (this.currentState !== WorldStateRunningState.RUNNING) {
       throw new Error(`World State is not running, unable to perform sync`);
     }
-    // If we have been given a block number to sync to and we have reached that number
-    // then return.
-    if (minBlockNumber !== undefined && minBlockNumber <= this.currentL2BlockNum) {
+    // If we have been given a block number to sync to and we have reached that number then return.
+    if (targetBlockNumber !== undefined && targetBlockNumber <= this.currentL2BlockNum) {
       return this.currentL2BlockNum;
     }
-    const blockToSyncTo = minBlockNumber === undefined ? 'latest' : `${minBlockNumber}`;
-    this.log.debug(`World State at block ${this.currentL2BlockNum}, told to sync to block ${blockToSyncTo}...`);
-    // ensure any outstanding block updates are completed first.
+    this.log.debug(`World State at ${this.currentL2BlockNum} told to sync to ${targetBlockNumber ?? 'latest'}`);
+    // ensure any outstanding block updates are completed first
     await this.jobQueue.syncPoint();
+
     while (true) {
       // Check the block number again
-      if (minBlockNumber !== undefined && minBlockNumber <= this.currentL2BlockNum) {
+      if (targetBlockNumber !== undefined && targetBlockNumber <= this.currentL2BlockNum) {
         return this.currentL2BlockNum;
       }
-      // Poll for more blocks
-      const numBlocks = await this.l2BlockDownloader.pollImmediate();
+      // Poll for more blocks, requesting even unproven blocks.
+      const numBlocks = await this.l2BlockDownloader.pollImmediate(targetBlockNumber, false);
       this.log.debug(`Block download immediate poll yielded ${numBlocks} blocks`);
       if (numBlocks) {
         // More blocks were received, process them and go round again
@@ -169,15 +200,28 @@ export class ServerWorldStateSynchronizer implements WorldStateSynchronizer {
         continue;
       }
       // No blocks are available, if we have been given a block number then we can't achieve it
-      if (minBlockNumber !== undefined) {
+      if (targetBlockNumber !== undefined) {
         throw new Error(
-          `Unable to sync to block number ${minBlockNumber}, currently synced to block ${this.currentL2BlockNum}`,
+          `Unable to sync to block number ${targetBlockNumber}, currently synced to block ${this.currentL2BlockNum}`,
         );
       }
       return this.currentL2BlockNum;
     }
   }
 
+  public async syncImmediateAndFork(
+    targetBlockNumber: number,
+    forkIncludeUncommitted: boolean,
+  ): Promise<MerkleTreeOperationsFacade> {
+    try {
+      await this.pause();
+      await this.syncImmediate(targetBlockNumber);
+      return await this.getFork(forkIncludeUncommitted);
+    } finally {
+      this.resume();
+    }
+  }
+
   /**
    * Checks for the availability of new blocks and processes them.
    */
diff --git a/yarn-project/world-state/src/synchronizer/world_state_synchronizer.ts b/yarn-project/world-state/src/synchronizer/world_state_synchronizer.ts
index 24dfa22226c..0411827e10d 100644
--- a/yarn-project/world-state/src/synchronizer/world_state_synchronizer.ts
+++ b/yarn-project/world-state/src/synchronizer/world_state_synchronizer.ts
@@ -52,6 +52,14 @@ export interface WorldStateSynchronizer {
    */
   syncImmediate(minBlockNumber?: number): Promise<number>;
 
+  /**
+   * Pauses the synchronizer, syncs to the target block number, forks world state, and resumes.
+   * @param targetBlockNumber - The block number to sync to.
+   * @param forkIncludeUncommitted - Whether to include uncommitted data in the fork.
+   * @returns The db forked at the requested target block number.
+   */
+  syncImmediateAndFork(targetBlockNumber: number, forkIncludeUncommitted: boolean): Promise<MerkleTreeOperations>;
+
   /**
    * Returns an instance of MerkleTreeOperations that will include uncommitted data.
    * @returns An instance of MerkleTreeOperations that will include uncommitted data.
diff --git a/yarn-project/world-state/src/world-state-db/index.ts b/yarn-project/world-state/src/world-state-db/index.ts
index f4c20a567f9..63ec2e7ba65 100644
--- a/yarn-project/world-state/src/world-state-db/index.ts
+++ b/yarn-project/world-state/src/world-state-db/index.ts
@@ -1,5 +1,6 @@
 export * from './merkle_trees.js';
 export * from './merkle_tree_db.js';
-export * from './merkle_tree_operations.js';
 export * from './merkle_tree_operations_facade.js';
 export * from './merkle_tree_snapshot_operations_facade.js';
+
+export { MerkleTreeOperations } from '@aztec/circuit-types/interfaces';
diff --git a/yarn-project/world-state/src/world-state-db/merkle_tree_db.ts b/yarn-project/world-state/src/world-state-db/merkle_tree_db.ts
index 4cee2af4522..d31ba02339c 100644
--- a/yarn-project/world-state/src/world-state-db/merkle_tree_db.ts
+++ b/yarn-project/world-state/src/world-state-db/merkle_tree_db.ts
@@ -1,9 +1,8 @@
 import { type MerkleTreeId } from '@aztec/circuit-types';
+import { type MerkleTreeOperations } from '@aztec/circuit-types/interfaces';
 import { type Fr, MAX_NULLIFIERS_PER_TX, MAX_TOTAL_PUBLIC_DATA_UPDATE_REQUESTS_PER_TX } from '@aztec/circuits.js';
 import { type IndexedTreeSnapshot, type TreeSnapshot } from '@aztec/merkle-tree';
 
-import { type MerkleTreeOperations } from './merkle_tree_operations.js';
-
 /**
  *
  * @remarks Short explanation:
@@ -62,4 +61,9 @@ export type MerkleTreeDb = {
      * @param block - The block number to take the snapshot at.
      */
     getSnapshot(block: number): Promise<TreeSnapshots>;
+
+    /**
+     * Forks the database at its current state.
+     */
+    fork(): Promise<MerkleTreeDb>;
   };
diff --git a/yarn-project/world-state/src/world-state-db/merkle_tree_map.ts b/yarn-project/world-state/src/world-state-db/merkle_tree_map.ts
new file mode 100644
index 00000000000..61c6a91c198
--- /dev/null
+++ b/yarn-project/world-state/src/world-state-db/merkle_tree_map.ts
@@ -0,0 +1,11 @@
+import { type MerkleTreeId } from '@aztec/circuit-types';
+import { type Fr } from '@aztec/circuits.js';
+import { type AppendOnlyTree, type IndexedTree } from '@aztec/merkle-tree';
+
+export type MerkleTreeMap = {
+  [MerkleTreeId.NULLIFIER_TREE]: IndexedTree;
+  [MerkleTreeId.NOTE_HASH_TREE]: AppendOnlyTree<Fr>;
+  [MerkleTreeId.PUBLIC_DATA_TREE]: IndexedTree;
+  [MerkleTreeId.L1_TO_L2_MESSAGE_TREE]: AppendOnlyTree<Fr>;
+  [MerkleTreeId.ARCHIVE]: AppendOnlyTree<Fr>;
+};
diff --git a/yarn-project/world-state/src/world-state-db/merkle_tree_operations_facade.ts b/yarn-project/world-state/src/world-state-db/merkle_tree_operations_facade.ts
index b67636866db..ae014d7d122 100644
--- a/yarn-project/world-state/src/world-state-db/merkle_tree_operations_facade.ts
+++ b/yarn-project/world-state/src/world-state-db/merkle_tree_operations_facade.ts
@@ -1,16 +1,15 @@
-import { type L2Block, type MerkleTreeId, type SiblingPath } from '@aztec/circuit-types';
-import { type Fr, type Header, type NullifierLeafPreimage, type StateReference } from '@aztec/circuits.js';
-import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
-import { type BatchInsertionResult } from '@aztec/merkle-tree';
-
-import { type MerkleTreeDb } from './merkle_tree_db.js';
+import { type BatchInsertionResult, type L2Block, type MerkleTreeId, type SiblingPath } from '@aztec/circuit-types';
 import {
   type HandleL2BlockAndMessagesResult,
   type IndexedTreeId,
   type MerkleTreeLeafType,
   type MerkleTreeOperations,
   type TreeInfo,
-} from './merkle_tree_operations.js';
+} from '@aztec/circuit-types/interfaces';
+import { type Fr, type Header, type NullifierLeafPreimage, type StateReference } from '@aztec/circuits.js';
+import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
+
+import { type MerkleTreeDb } from './merkle_tree_db.js';
 
 /**
  * Wraps a MerkleTreeDbOperations to call all functions with a preset includeUncommitted flag.
diff --git a/yarn-project/world-state/src/world-state-db/merkle_tree_snapshot_operations_facade.ts b/yarn-project/world-state/src/world-state-db/merkle_tree_snapshot_operations_facade.ts
index 3f72c680732..a25037b3d68 100644
--- a/yarn-project/world-state/src/world-state-db/merkle_tree_snapshot_operations_facade.ts
+++ b/yarn-project/world-state/src/world-state-db/merkle_tree_snapshot_operations_facade.ts
@@ -1,16 +1,17 @@
 import { MerkleTreeId, type SiblingPath } from '@aztec/circuit-types';
-import { AppendOnlyTreeSnapshot, Fr, type Header, PartialStateReference, StateReference } from '@aztec/circuits.js';
-import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
-import { type BatchInsertionResult, type IndexedTreeSnapshot } from '@aztec/merkle-tree';
-
-import { type MerkleTreeDb, type TreeSnapshots } from './merkle_tree_db.js';
 import {
+  type BatchInsertionResult,
   type HandleL2BlockAndMessagesResult,
   type IndexedTreeId,
   type MerkleTreeLeafType,
   type MerkleTreeOperations,
   type TreeInfo,
-} from './merkle_tree_operations.js';
+} from '@aztec/circuit-types/interfaces';
+import { AppendOnlyTreeSnapshot, Fr, type Header, PartialStateReference, StateReference } from '@aztec/circuits.js';
+import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
+import { type IndexedTreeSnapshot } from '@aztec/merkle-tree';
+
+import { type MerkleTreeDb, type TreeSnapshots } from './merkle_tree_db.js';
 
 /**
  * Merkle tree operations on readonly tree snapshots.
diff --git a/yarn-project/world-state/src/world-state-db/merkle_trees.ts b/yarn-project/world-state/src/world-state-db/merkle_trees.ts
index 881141caae0..0ebd9937bca 100644
--- a/yarn-project/world-state/src/world-state-db/merkle_trees.ts
+++ b/yarn-project/world-state/src/world-state-db/merkle_trees.ts
@@ -1,4 +1,12 @@
 import { type L2Block, MerkleTreeId, PublicDataWrite, type SiblingPath, TxEffect } from '@aztec/circuit-types';
+import {
+  type BatchInsertionResult,
+  type HandleL2BlockAndMessagesResult,
+  type IndexedTreeId,
+  type MerkleTreeLeafType,
+  type MerkleTreeOperations,
+  type TreeInfo,
+} from '@aztec/circuit-types/interfaces';
 import {
   ARCHIVE_HEIGHT,
   AppendOnlyTreeSnapshot,
@@ -28,7 +36,6 @@ import { type IndexedTreeLeafPreimage } from '@aztec/foundation/trees';
 import { type AztecKVStore, type AztecSingleton } from '@aztec/kv-store';
 import {
   type AppendOnlyTree,
-  type BatchInsertionResult,
   type IndexedTree,
   Poseidon,
   StandardIndexedTree,
@@ -46,14 +53,7 @@ import {
   type MerkleTreeDb,
   type TreeSnapshots,
 } from './merkle_tree_db.js';
-import {
-  type HandleL2BlockAndMessagesResult,
-  type IndexedTreeId,
-  type MerkleTreeLeafType,
-  type MerkleTreeMap,
-  type MerkleTreeOperations,
-  type TreeInfo,
-} from './merkle_tree_operations.js';
+import { type MerkleTreeMap } from './merkle_tree_map.js';
 import { MerkleTreeOperationsFacade } from './merkle_tree_operations_facade.js';
 
 /**
@@ -180,6 +180,15 @@ export class MerkleTrees implements MerkleTreeDb {
     await this.#commit();
   }
 
+  public async fork(): Promise<MerkleTrees> {
+    // TODO(palla/prover-node): If the underlying store is being shared with other components, we're unnecessarily
+    // copying a lot of data unrelated to merkle trees. This may be fine for now, and we may be able to ditch backup-based
+    // forking in favor of a more elegant proposal. But if we see this operation starts taking a lot of time, we may want
+    // to open separate stores for merkle trees and other components.
+    const forked = await this.store.fork();
+    return MerkleTrees.new(forked, this.log);
+  }
+
   public getInitialHeader(): Header {
     return Header.empty({ state: this.#loadInitialStateReference() });
   }