feat(sequencer): make sequencer pausable (#24475)

Supersedes #24449 and #24450, reimplementing both on a simpler design.
Single PR since the lifecycle work is only exercised by the e2e
primitive.

## Context

Several multi-node e2e tests spend minutes of wall-clock waiting for the
L1 clock to roll while live sequencers sit idle. Warping the shared
clock under a running sequencer interrupts whatever iteration is
mid-build, producing the reorg / "Sequencer was interrupted" failures
behind earlier revert attempts. Pausing sequencers around the warp
requires a lifecycle that can actually stop and resume, which it
couldn't: a second `start()` orphaned the previous poll loop (two loops
racing), and a restarted sequencer never cleared the publishers'
`interrupted` flag, so it would build and propose blocks but silently
never publish to L1.

## Approach

**Idempotent, restartable lifecycle.** `start()`/`stop()` are idempotent
across `SequencerClient`, `Sequencer`, and `PublisherManager`, and
`start()` while `STOPPING` is refused so a mid-stop start cannot orphan
a fresh loop. `PublisherManager.start()` clears the interrupted flag via
the previously-unused `restart()` methods, and `SequencerClient.start()`
starts the publisher manager before the sequencer loop so publishers are
un-interrupted before the first post-restart publish. The funding loop
is created once in the constructor and restarted across cycles, and a
start that failed to load publisher state can be retried.

**`pause()` for restarts.** Restartability is funneled through a
dedicated `Sequencer.pause()` / `SequencerClient.pause()`: it halts the
poll loop and waits for the in-flight `work()` iteration and every
pending L1 submission / fallback send to finish *without interrupting
them* — no interrupt lands mid-build, so no spurious `checkpoint-error`
is emitted and no enqueued checkpoint is dropped. It deliberately does
**not** stop inner services: the validator/HA signer and publishers keep
running, so the slashing-protection store stays open and a later
`start()` cleanly resumes. Draining happens without entering `STOPPING`,
since in that state the iteration's own `setState` calls throw
`SequencerInterruptedError`, which would fail the very iteration being
drained. This replaces the wait-for-IDLE heuristic from #24450, which
raced: sequencers reach IDLE at different times, and any of them could
re-enter `work()` before the stop landed, flaking any test that asserts
no sequencer failure events.

**`stop()` stays a full teardown.** `stop()` keeps the fast interrupting
shutdown used for production teardown, and stopping the validator client
closes its slashing-protection database (LMDB single-node / Postgres
HA), as it did before this line of work. Because restarts now go through
`pause()` — which never closes the store — `stop()` no longer needs to
leave the store open, so there is no DB-ownership bookkeeping to
distinguish "owned" from "shared" databases.

**Single shared request tracker.** The checkpoint proposal jobs'
backgrounded L1 submissions and the sequencer's own fire-and-forget
fallback sends (vote-and-prune when we cannot build, escape-hatch votes)
are tracked in one `RequestsTracker` owned by the sequencer and handed
to each job it creates. `stop()` interrupts and drains that single
tracker in one place; `pause()` awaits it untouched. A sender sleeping
until its target slot therefore cannot survive a `stop()` and publish a
stale-slot tx after a restart clears the pooled publishers' interrupted
flag: interrupting the wrapper publisher is permanent, since wrappers
are never restarted.

**e2e primitive + pilot.** `warpWithSequencersPaused(nodes, cheatCodes,
target, opts)` on `SingleNodeTestContext` (inherited by
`MultiNodeTestContext`) pauses every sequencer, runs the warp with
nobody building, and resumes them (`restart: false` leaves them paused).
Archivers, provers, and the chain monitor keep running, so clock-driven
effects such as an orphan-block prune still fire. Applied to
`pipeline_prune` to collapse its ~2-minute dead wait for the orphan
slot's prune deadline: warp two L1 slots into the slot after the
orphaned one, and resume the sequencers only once the prune is
confirmed. `TX_COUNT` is unchanged, so `assertMultipleBlocksPerSlot` and
`assertProposerPipelining` still hold.

Lifecycle calls are assumed to be serialized by the caller (all current
callers are); this does not add a mutex for truly concurrent
start/stop/pause.

The `pipeline_prune` speedup is validated by this PR's CI run — the
multi-node suite can't be run locally.

Author: spalladino

yarn-project/end-to-end/src/multi-node/recovery/pipeline_prune.test.ts

@@ -6,6 +6,7 @@ import { BlockNumber, CheckpointNumber, SlotNumber } from '@aztec/foundation/bra
 import { executeTimeout } from '@aztec/foundation/timer';
 import type { SequencerEvents } from '@aztec/sequencer-client';
 import { L2BlockSourceEvents } from '@aztec/stdlib/block';
+import { getTimestampForSlot } from '@aztec/stdlib/epoch-helpers';
 
 import { proveAndSendTxs } from '../../test-wallet/utils.js';
 import {
@@ -102,6 +103,29 @@ describe('multi-node/recovery/pipeline_prune', () => {
     // The sequencer keeps building blocks and broadcasting via P2P, but won't submit the checkpoint to L1
     targetSequencer.updateConfig({ skipPublishingCheckpointsPercent: 100 });
 
+    // Wait for the orphan blocks to actually exist before warping: the target proposer builds them during
+    // slot proposerSlotToNotPublish - 1 and broadcasts them via P2P carrying that submission slot, but never
+    // publishes the enclosing checkpoint. Only once node[0]'s archiver holds them as a proposed (uncheckpointed)
+    // tip is there anything for pruneOrphanProposedBlocks to prune — warping before they arrive would fire no prune.
+    await test.waitForAllNodesToReachBlockAtSlot(
+      proposerSlotToNotPublish,
+      'proposed',
+      block => block.header.globalVariables.slotNumber >= proposerSlotToNotPublish,
+      { nodes: [nodes[0]], timeout: test.L2_SLOT_DURATION_IN_S * 3 },
+    );
+    logger.warn(`Orphan blocks for slot ${proposerSlotToNotPublish} are present; warping past the prune deadline`);
+
+    // Collapse the ~2-minute dead gap where the chain just waits for the L1 clock to roll past the orphan
+    // slot's checkpoint-proposal-received deadline so pruneOrphanProposedBlocks fires. The archiver reads the
+    // shared TestDateProvider that eth.warp advances, so jumping the clock into the slot after the orphan one
+    // takes us safely past that deadline and the next archiver sync prunes. The sequencers are kept stopped
+    // (restart: false) until the prune is confirmed, so no proposer builds against the still-unpruned tip;
+    // they are restarted for recovery below.
+    const pruneWarpTarget =
+      getTimestampForSlot(SlotNumber(proposerSlotToNotPublish + 1), test.constants) +
+      BigInt(2 * test.constants.ethereumSlotDuration);
+    await test.warpWithSequencersPaused(nodes, test.context.cheatCodes, pruneWarpTarget, { restart: false });
+
     const pruneTimeout = test.L2_SLOT_DURATION_IN_S * 5 * 1000;
     logger.warn(`Waiting for uncheckpointed blocks to be pruned (timeout=${pruneTimeout}ms)`);
     await executeTimeout(() => prunePromise, pruneTimeout);
@@ -119,9 +143,13 @@ describe('multi-node/recovery/pipeline_prune', () => {
     }
     logger.warn(`Pruning detected, block number now ${await archiver.getBlockNumber()}`);
 
-    // Re-enable checkpoint publishing
+    // Re-enable checkpoint publishing, then restart the sequencers to build the recovery checkpoint.
+    // Restarting only now (after the prune and after listeners are attached) keeps recovery from racing the
+    // prune and ensures every recovery block is captured for the pipelining assertion.
     logger.warn(`Re-enabling checkpoint publishing for validator ${proposerIndex}`);
     targetSequencer.updateConfig({ skipPublishingCheckpointsPercent: 0 });
+    await test.startSequencers(nodes);
+    logger.warn(`Restarted all sequencers for recovery`);
 
     // Wait for a new checkpoint (recovery) - where all txs end up mined
     const timeout = test.L2_SLOT_DURATION_IN_S * 5;

yarn-project/end-to-end/src/single-node/single_node_test_context.ts

@@ -9,6 +9,7 @@ import { Fr } from '@aztec/aztec.js/fields';
 import type { Logger } from '@aztec/aztec.js/log';
 import { MerkleTreeId } from '@aztec/aztec.js/trees';
 import type { Wallet } from '@aztec/aztec.js/wallet';
+import type { CheatCodes } from '@aztec/aztec/testing';
 import { EpochCache } from '@aztec/epoch-cache';
 import { createExtendedL1Client } from '@aztec/ethereum/client';
 import { DefaultL1ContractsConfig } from '@aztec/ethereum/config';
@@ -991,4 +992,50 @@ export class SingleNodeTestContext {
     }
     expect(failEvents).toEqual([]);
   }
+
+  /**
+   * Warps the L1 clock to `target` with every given node's sequencer paused, then resumes them
+   * (pass `restart: false` to leave them paused, e.g. until some clock-driven effect is confirmed).
+   *
+   * Warping the shared date provider under live sequencers interrupts whatever iteration is mid-build,
+   * producing spurious `block-build-failed` / `checkpoint-error` events and dropped checkpoints. The
+   * sequencers are therefore paused first: the poll loop halts and the in-flight iteration, its pending L1
+   * submission, and any pending fallback vote all finish untouched before the warp, so nothing fires with a
+   * stale slot afterwards. Pausing leaves the validator clients (and their slashing-protection stores) and
+   * publishers running, so a later {@link SequencerClient.start} cleanly resumes; archivers, provers, and
+   * the chain monitor keep running throughout, so clock-driven effects of the warp (e.g. an orphan-block
+   * prune) still fire.
+   *
+   * The warp is performed here (rather than via a caller-supplied callback) so it happens only after the
+   * pause has drained. Draining can take several slots, so a `target` computed before the pause may already
+   * lie in the past by the time the sequencers are down. The warp is therefore skipped when the L1 clock has
+   * already reached or passed `target` — `evm_setNextBlockTimestamp` rejects a non-advancing timestamp, so
+   * warping there would throw "timestamp in the past".
+   */
+  public async warpWithSequencersPaused(
+    nodes: AztecNodeService[],
+    cheatCodes: CheatCodes,
+    target: bigint,
+    opts: { restart?: boolean } = {},
+  ): Promise<void> {
+    const sequencers = this.getSequencers(nodes);
+    await testSpan('warp:sequencers-paused', async () => {
+      this.logger.warn(`Pausing ${sequencers.length} sequencers before warp`);
+      await Promise.all(sequencers.map(sequencer => sequencer.pause()));
+      const currentTs = BigInt(await cheatCodes.eth.lastBlockTimestamp());
+      if (currentTs < target) {
+        this.logger.warn(`Warping L1 to ${target} with all sequencers paused`, { currentTs, target });
+        await cheatCodes.eth.warp(Number(target), { resetBlockInterval: true });
+      } else {
+        this.logger.verbose(`Skipping warp: L1 clock ${currentTs} already at or past target ${target}`, {
+          currentTs,
+          target,
+        });
+      }
+      if (opts.restart ?? true) {
+        this.logger.warn(`Resuming ${sequencers.length} sequencers after warp`);
+        await Promise.all(sequencers.map(sequencer => sequencer.start()));
+      }
+    });
+  }
 }

yarn-project/ethereum/src/publisher_manager.test.ts

@@ -418,6 +418,105 @@ describe('PublisherManager', () => {
     });
   });
 
+  describe('lifecycle', () => {
+    let funder: TestL1TxUtils & L1TxUtils;
+
+    beforeEach(() => {
+      funder = new TestL1TxUtils(EthAddress.random()) as TestL1TxUtils & L1TxUtils;
+      funder.balance = 5000n;
+    });
+
+    it('stop interrupts all publishers and the funder', async () => {
+      mockPublishers = createMockPublishers(3);
+      publisherManager = new PublisherManager(mockPublishers, {}, { funder });
+
+      await publisherManager.start();
+      await publisherManager.stop();
+
+      expect(mockPublishers.every(p => p.interrupted)).toBe(true);
+      expect(funder.interrupted).toBe(true);
+    });
+
+    it('start after stop clears the interrupted flag so publishing works again', async () => {
+      mockPublishers = createMockPublishers(3);
+      publisherManager = new PublisherManager(mockPublishers, {}, { funder });
+
+      await publisherManager.start();
+      await publisherManager.stop();
+      expect(mockPublishers.every(p => p.interrupted)).toBe(true);
+
+      // Restart: interrupted must be cleared, otherwise sendTransaction would throw InterruptError.
+      await publisherManager.start();
+
+      expect(mockPublishers.every(p => !p.interrupted)).toBe(true);
+      expect(funder.interrupted).toBe(false);
+    });
+
+    it('a second start does not reload state, which would duplicate background monitors', async () => {
+      mockPublishers = createMockPublishers(1);
+      publisherManager = new PublisherManager(mockPublishers, {});
+
+      await publisherManager.start();
+      await publisherManager.start();
+
+      expect(mockPublishers[0].loadCount).toBe(1);
+    });
+
+    it('a start after a stop reloads state so in-flight txs resume monitoring', async () => {
+      mockPublishers = createMockPublishers(1);
+      publisherManager = new PublisherManager(mockPublishers, {});
+
+      await publisherManager.start();
+      await publisherManager.stop();
+      await publisherManager.start();
+
+      expect(mockPublishers[0].loadCount).toBe(2);
+    });
+
+    it('a failed start can be retried', async () => {
+      mockPublishers = createMockPublishers(1);
+      publisherManager = new PublisherManager(mockPublishers, {});
+
+      mockPublishers[0].failNextLoad = true;
+      await expect(publisherManager.start()).rejects.toThrow('load failed');
+
+      await publisherManager.start();
+      expect(mockPublishers[0].loadCount).toBe(1);
+    });
+
+    it('is idempotent on double stop', async () => {
+      mockPublishers = createMockPublishers(2);
+      publisherManager = new PublisherManager(mockPublishers, {});
+
+      await publisherManager.start();
+      await publisherManager.stop();
+      await expect(publisherManager.stop()).resolves.not.toThrow();
+
+      expect(mockPublishers.every(p => p.interrupted)).toBe(true);
+    });
+
+    it('resumes periodic funding checks after a restart', async () => {
+      mockPublishers = createMockPublishers(1);
+      mockPublishers[0].balance = 50n; // stays below threshold, so every funding check funds it
+      publisherManager = new PublisherManager(
+        mockPublishers,
+        { publisherFundingThreshold: 100n, publisherFundingAmount: 50n },
+        { funder },
+      );
+
+      await publisherManager.start();
+      await new Promise(resolve => setTimeout(resolve, 10));
+      await publisherManager.stop();
+      expect(funder.sendAndMonitorTransaction).toHaveBeenCalledTimes(1);
+
+      // The funding loop must be re-armed by the restart, triggering another immediate check.
+      await publisherManager.start();
+      await new Promise(resolve => setTimeout(resolve, 10));
+      await publisherManager.stop();
+      expect(funder.sendAndMonitorTransaction).toHaveBeenCalledTimes(2);
+    });
+  });
+
   function createMockPublishers(count: number, addresses: EthAddress[] = []): (TestL1TxUtils & L1TxUtils)[] {
     const tempAddress = [...addresses];
     return times(
@@ -431,6 +530,10 @@ class TestL1TxUtils {
   public state: TxUtilsState = TxUtilsState.IDLE;
   public lastMinedAtBlockNumber: bigint | undefined = undefined;
   public balance: bigint = 1000n;
+  /** Mirrors the real ReadOnlyL1TxUtils.interrupted flag so tests can assert publishing is re-enabled. */
+  public interrupted = false;
+  public loadCount = 0;
+  public failNextLoad = false;
   public sendAndMonitorTransaction = jest.fn<() => Promise<any>>().mockResolvedValue({
     receipt: { transactionHash: '0xabc', status: 'success' },
     state: {},
@@ -446,7 +549,20 @@ class TestL1TxUtils {
     return this.senderAddress;
   }
 
-  public async loadStateAndResumeMonitoring() {}
+  public loadStateAndResumeMonitoring() {
+    if (this.failNextLoad) {
+      this.failNextLoad = false;
+      return Promise.reject(new Error('load failed'));
+    }
+    this.loadCount++;
+    return Promise.resolve();
+  }
 
-  public interrupt() {}
+  public interrupt() {
+    this.interrupted = true;
+  }
+
+  public restart() {
+    this.interrupted = false;
+  }
 }

yarn-project/ethereum/src/publisher_manager.ts

@@ -41,7 +41,8 @@ export class PublisherManager<UtilsType extends L1TxUtils = L1TxUtils> {
   private config: PublisherManagerConfig;
   private static readonly FUNDING_CHECK_INTERVAL_MS = 2 * 60 * 1000;
   protected funder?: UtilsType;
-  protected fundingPromise?: RunningPromise;
+  protected readonly fundingPromise?: RunningPromise;
+  private started = false;
 
   constructor(
     protected publishers: UtilsType[],
@@ -67,31 +68,50 @@ export class PublisherManager<UtilsType extends L1TxUtils = L1TxUtils> {
         this.funder = undefined;
       }
     }
+
+    if (this.funder && hasThreshold && hasAmount) {
+      this.fundingPromise = new RunningPromise(
+        () => this.triggerFundingIfNeeded(),
+        this.log,
+        PublisherManager.FUNDING_CHECK_INTERVAL_MS,
+      );
+    }
   }
 
-  /** Loads the state of all publishers and the funder, and starts periodic funding checks. */
+  /**
+   * Clears any interrupted flag left by a previous {@link stop} so publishing works again after a restart,
+   * loads the state of all publishers and the funder, and starts periodic funding checks. Idempotent: a
+   * start while already started is a no-op, so it never re-runs `loadStateAndResumeMonitoring` (which
+   * would spawn a duplicate background monitor per pending nonce). Lifecycle calls are expected to be
+   * serialized by the caller.
+   */
   public async start(): Promise<void> {
+    if (this.started) {
+      this.log.debug('PublisherManager already started, ignoring start');
+      return;
+    }
+
+    // Clear the interrupted flag set by a previous stop() so a restarted manager can publish again.
+    // On a first start this is a no-op (the flag is already clear).
+    this.publishers.forEach(pub => pub.restart());
+    this.funder?.restart();
+
     await Promise.all([
       ...this.publishers.map(pub => pub.loadStateAndResumeMonitoring()),
       this.funder?.loadStateAndResumeMonitoring(),
     ]);
 
-    if (
-      this.funder &&
-      this.config.publisherFundingThreshold !== undefined &&
-      this.config.publisherFundingAmount !== undefined
-    ) {
-      this.fundingPromise = new RunningPromise(
-        () => this.triggerFundingIfNeeded(),
-        this.log,
-        PublisherManager.FUNDING_CHECK_INTERVAL_MS,
-      );
-      this.fundingPromise.start();
-    }
+    this.fundingPromise?.start();
+    // Marked started only once fully up, so a start that failed to load state can be retried.
+    this.started = true;
   }
 
-  /** Stops the funding loop and interrupts all publishers. */
+  /**
+   * Stops the funding loop and interrupts all publishers so no further L1 txs are sent. Idempotent, and
+   * the manager may be restarted afterwards via {@link start}, which clears the interrupted flag.
+   */
   public async stop(): Promise<void> {
+    this.started = false;
     await this.fundingPromise?.stop();
     this.publishers.forEach(pub => pub.interrupt());
     this.funder?.interrupt();

yarn-project/foundation/src/promise/running-promise.test.ts

@@ -52,6 +52,44 @@ describe('RunningPromise', () => {
     });
   });
 
+  describe('lifecycle', () => {
+    it('a second start does not spawn a second poll loop', async () => {
+      runningPromise.start();
+      expect(counter).toEqual(1);
+
+      // Starting again while already running must be a no-op, not a second concurrent loop.
+      runningPromise.start();
+      expect(counter).toEqual(1);
+
+      await jest.advanceTimersToNextTimerAsync();
+      // Exactly one loop advanced the counter, not two.
+      expect(counter).toEqual(2);
+    });
+
+    it('stop is safe to call when never started and when already stopped', async () => {
+      await expect(runningPromise.stop()).resolves.toBeUndefined();
+
+      runningPromise.start();
+      await runningPromise.stop();
+      expect(runningPromise.isRunning()).toBe(false);
+
+      await expect(runningPromise.stop()).resolves.toBeUndefined();
+    });
+
+    it('can be restarted after a stop', async () => {
+      runningPromise.start();
+      expect(counter).toEqual(1);
+      await runningPromise.stop();
+
+      runningPromise.start();
+      expect(runningPromise.isRunning()).toBe(true);
+      expect(counter).toEqual(2);
+
+      await jest.advanceTimersToNextTimerAsync();
+      expect(counter).toEqual(3);
+    });
+  });
+
   describe('handles errors', () => {
     beforeEach(() => {
       fn.mockImplementation(() => {