fix: interrupt publisher send-at-slot sleep on sequencer stop (#23990)

## Summary

- Propagate `CheckpointProposalJob.interrupt()` to its
`SequencerPublisher` so the publisher's `sendRequestsAt` slot-deadline
sleep is cancelled on sequencer stop.
- Check `interrupted` *before* sleeping in `sendRequestsAt`, since
`InterruptibleSleep.interrupt()` only resolves sleeps already in flight.

Completes #23930, which made the job's own polling waits interruptible
but not the publisher's. In [this `e2e_ha_full.test.ts`
flake](http://ci.aztec-labs.com/1c46f3d4a226073d) a node became proposer
165ms before teardown and took the no-broadcast votes path, leaving
`pendingL1Submission = publisher.sendRequestsAt(targetSlot)` sleeping
until the target slot — ~22 wall-clock minutes away under the warped
test clock. `Sequencer.stop()` blocked on it ("Awaiting pending L1
payload submission"), node stops were abandoned, and Jest hung on leaked
handles until CI killed the run. Nothing in the shutdown path interrupts
the per-job `SequencerPublisher`: `publisherFactory.stopAll()` only
interrupts the underlying `L1TxUtils`.

## Tests

- `checkpoint_proposal_job.test.ts`: pending L1 submission blocked in
the publisher resolves promptly on `job.interrupt()` (red without the
fix).
- `sequencer-publisher.test.ts`: `sendRequestsAt` returns immediately
when interrupted before the sleep starts (red without the fix).

Author: spypsy

yarn-project/end-to-end/src/composed/ha/e2e_ha_full.test.ts

@@ -304,24 +304,39 @@ describe('HA Full Setup', () => {
   });
 
   afterAll(async () => {
+    const cleanupErrors: string[] = [];
+
+    dateProvider?.reset();
+
     // Stop all HA peer nodes in parallel with a per-node deadline. A single stuck node can otherwise
-    // block the serial loop long enough to blow the jest hook timeout — e.g. a sequencer.stop() that
-    // awaits an L1 publish whose tx-timeout was computed on a test-warped clock and never fires.
+    // block the serial loop long enough to blow the jest hook timeout, so report the stuck node directly
+    // instead of letting the suite pass and later fail with Jest open handles.
     if (haNodeServices) {
       const STOP_DEADLINE_MS = 30_000;
-      await Promise.allSettled(
+      const stopResults = await Promise.allSettled(
         haNodeServices.map((service, i) => {
           logger.info(`Stopping HA peer node ${i}`);
           return Promise.race([
             service.stop().catch(error => {
-              logger.error(`Failed to stop HA peer node ${i}: ${error}`);
+              const message = `Failed to stop HA peer node ${i}: ${error}`;
+              logger.error(message);
+              return message;
             }),
             sleep(STOP_DEADLINE_MS).then(() => {
-              logger.error(`HA peer node ${i} stop did not return within ${STOP_DEADLINE_MS}ms; abandoning`);
+              const message = `HA peer node ${i} stop did not return within ${STOP_DEADLINE_MS}ms; abandoning`;
+              logger.error(message);
+              return message;
             }),
           ]);
         }),
       );
+      for (const result of stopResults) {
+        if (result.status === 'rejected') {
+          cleanupErrors.push(`Unexpected HA node stop error: ${result.reason}`);
+        } else if (result.value) {
+          cleanupErrors.push(result.value);
+        }
+      }
     }
 
     // Cleanup HA keystore temp directories
@@ -350,6 +365,10 @@ describe('HA Full Setup', () => {
 
     // Cleanup bootstrap node and test infrastructure (this cleans up the shared data directory)
     await teardown();
+
+    if (cleanupErrors.length > 0) {
+      throw new Error(cleanupErrors.join('\n'));
+    }
   });
 
   afterEach(async () => {

yarn-project/sequencer-client/src/publisher/sequencer-publisher.test.ts

@@ -720,6 +720,28 @@ describe('SequencerPublisher', () => {
     expect((publisher as any).requests.length).toEqual(0);
   });
 
+  it('does not sleep in sendRequestsAt if interrupted beforehand', async () => {
+    // A target slot far enough in the future that sendRequestsAt would sleep for ~1 hour
+    // (EmptyL1RollupConstants has slotDuration 1s and l1GenesisTime 0, so slot N starts at N seconds).
+    const targetSlot = SlotNumber(Math.ceil(Date.now() / 1000) + 3600);
+    publisher.interrupt();
+
+    let timeout: NodeJS.Timeout | undefined;
+    try {
+      const result = await Promise.race([
+        publisher.sendRequestsAt(targetSlot),
+        new Promise<'timed-out'>(resolve => {
+          timeout = setTimeout(() => resolve('timed-out'), 1000);
+        }),
+      ]);
+      expect(result).toBeUndefined();
+    } finally {
+      if (timeout) {
+        clearTimeout(timeout);
+      }
+    }
+  });
+
   it('does not send requests if no valid requests are found', async () => {
     publisher.addRequest({
       action: 'propose',

yarn-project/sequencer-client/src/publisher/sequencer-publisher.ts

@@ -630,6 +630,9 @@ export class SequencerPublisher {
     // Aim to be in the mempool one L1 slot before the L2 slot starts, so we have a chance of
     // being picked up by the first L1 block of the L2 slot.
     const submitAfterMs = startOfTargetSlotMs - Number(this.ethereumSlotDuration) * 1000;
+    if (this.interrupted) {
+      return undefined;
+    }
     const sleepMs = submitAfterMs - this.dateProvider.now();
     if (sleepMs > 0) {
       this.log.debug(`Sleeping ${sleepMs}ms before sending requests`, {

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.test.ts

@@ -14,6 +14,7 @@ import { TimeoutError } from '@aztec/foundation/error';
 import { EthAddress } from '@aztec/foundation/eth-address';
 import { Signature } from '@aztec/foundation/eth-signature';
 import { createLogger } from '@aztec/foundation/log';
+import { promiseWithResolvers } from '@aztec/foundation/promise';
 import { TestDateProvider } from '@aztec/foundation/timer';
 import type { TypedEventEmitter } from '@aztec/foundation/types';
 import { type P2P, P2PClientState } from '@aztec/p2p';
@@ -1719,6 +1720,39 @@ describe('CheckpointProposalJob', () => {
       }
     });
 
+    it('interrupts a pending L1 submission sleeping in the publisher', async () => {
+      const { txs, block } = await setupTxsAndBlock(p2p, globalVariables, 1, chainId);
+      checkpointBuilder.seedBlocks([block], [txs]);
+      validatorClient.collectAttestations.mockResolvedValue(getAttestations(block));
+
+      // Simulate sendRequestsAt sleeping until the target slot: the promise only resolves once
+      // the publisher itself is interrupted.
+      const sendDeferred = promiseWithResolvers<undefined>();
+      publisher.sendRequestsAt.mockReturnValue(sendDeferred.promise);
+      publisher.interrupt.mockImplementation(() => sendDeferred.resolve(undefined));
+
+      const checkpoint = await job.execute();
+      expect(checkpoint).toBeDefined();
+
+      const pendingSubmission = job.awaitPendingSubmission().then(() => 'stopped' as const);
+      job.interrupt();
+
+      let timeout: NodeJS.Timeout | undefined;
+      try {
+        const result = await Promise.race([
+          pendingSubmission,
+          new Promise<'timed-out'>(resolve => {
+            timeout = setTimeout(() => resolve('timed-out'), 1000);
+          }),
+        ]);
+        expect(result).toBe('stopped');
+      } finally {
+        if (timeout) {
+          clearTimeout(timeout);
+        }
+      }
+    });
+
     it('aborts checkpoint when syncing proposed block to archiver fails', async () => {
       const { txs, block } = await setupTxsAndBlock(p2p, globalVariables, 1, chainId);
       checkpointBuilder.seedBlocks([block], [txs]);

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts

@@ -189,10 +189,11 @@ export class CheckpointProposalJob implements Traceable {
     await this.pendingL1Submission;
   }
 
-  /** Interrupts job-owned waits so shutdown can finish. */
+  /** Interrupts job-owned waits, including the publisher's send-at-slot sleep, so shutdown can finish. */
   public interrupt(): void {
     this.interrupted = true;
     this.interruptibleSleep.interrupt(true);
+    this.publisher.interrupt();
   }
 
   private async awaitInterruptibleSleep(ms: number): Promise<void> {