fix: reconcile pending bundles on a timer when blocks stall (SUP-1552)

src/executor/executorManager.test.ts

@@ -0,0 +1,215 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+import { ExecutorManager } from "./executorManager"
+
+// Chains that only produce a block per transaction deadlock a block-driven
+// reconciler. These cover the timer that breaks that loop.
+
+const RESUBMIT_STUCK_TIMEOUT = 10_000
+const BLOCK_TIME = 1000
+
+const noopLogger = () => {
+    const logger = {
+        trace: vi.fn(),
+        debug: vi.fn(),
+        info: vi.fn(),
+        warn: vi.fn(),
+        error: vi.fn()
+    }
+    return logger
+}
+
+const createHarness = () => {
+    // Captured so a test can emit a block the way watchBlocks would.
+    let onBlock: ((block: unknown) => Promise<void>) | undefined
+    const unwatch = vi.fn()
+
+    const submittedBundle = {
+        uid: "0xbundle",
+        transactionHash: "0xtx",
+        previousTransactionHashes: [],
+        // Matches networkGasPrice, so isGasPriceTooLow stays false.
+        transactionRequest: {
+            maxFeePerGas: 1n,
+            maxPriorityFeePerGas: 1n,
+            nonce: 0
+        },
+        bundle: {
+            entryPoint: "0xentrypoint",
+            version: "0.7",
+            userOps: [{ userOpHash: "0xuserop" }],
+            submissionAttempts: 1
+        },
+        executor: { address: "0xexecutor" },
+        lastReplaced: Date.now()
+    }
+
+    // Mutable: production drops the bundle in processIncludedBundle.
+    let pendingBundles: unknown[] = [submittedBundle]
+
+    const getBundleStatuses = vi.fn().mockResolvedValue([])
+    const processIncludedBundle = vi.fn(() => {
+        pendingBundles = []
+    })
+
+    const bundleManager = {
+        getPendingBundles: vi.fn(() => pendingBundles),
+        getBundleStatuses,
+        processIncludedBundle
+    }
+
+    const config = {
+        bundleMode: "manual",
+        blockTime: BLOCK_TIME,
+        resubmitStuckTimeout: RESUBMIT_STUCK_TIMEOUT,
+        flashblocksPreconfirmationTime: undefined,
+        // Keeps getBaseFee() off the network.
+        legacyTransactions: true,
+        logLevel: "info",
+        executorLogLevel: "info",
+        getLogger: () => noopLogger(),
+        publicClient: {
+            watchBlocks: (args: {
+                onBlock: (block: unknown) => Promise<void>
+            }) => {
+                onBlock = args.onBlock
+                return unwatch
+            },
+            // Only cost metrics reach this, and they swallow failures.
+            getTransactionReceipt: vi
+                .fn()
+                .mockRejectedValue(new Error("no receipt in test"))
+        }
+    }
+
+    const executorManager = new ExecutorManager({
+        // Narrow stubs: only the block-reconcile path is under test.
+        config: config as any,
+        executor: {} as any,
+        mempool: {} as any,
+        metrics: {} as any,
+        gasPriceManager: {
+            tryGetNetworkGasPrice: vi.fn().mockResolvedValue({
+                maxFeePerGas: 1n,
+                maxPriorityFeePerGas: 1n
+            })
+        } as any,
+        senderManager: {} as any,
+        bundleManager: bundleManager as any
+    })
+
+    return {
+        executorManager,
+        getBundleStatuses,
+        processIncludedBundle,
+        submittedBundle,
+        unwatch,
+        getPending: () => pendingBundles,
+        emitBlock: async () => {
+            await onBlock?.({ number: 1n, baseFeePerGas: 1n })
+        }
+    }
+}
+
+const includedStatus = {
+    status: "included",
+    userOpReceipts: {},
+    transactionHash: "0xtx",
+    blockNumber: 2n
+}
+
+describe("ExecutorManager stale block watchdog", () => {
+    beforeEach(() => {
+        vi.useFakeTimers()
+        return () => vi.useRealTimers()
+    })
+
+    it("reconciles pending bundles when no new block arrives", async () => {
+        const { executorManager, getBundleStatuses } = createHarness()
+
+        executorManager.startWatchingBlocks()
+
+        // No block ever emitted. Was a permanent stall before the watchdog.
+        expect(getBundleStatuses).not.toHaveBeenCalled()
+
+        await vi.advanceTimersByTimeAsync(RESUBMIT_STUCK_TIMEOUT + BLOCK_TIME)
+
+        expect(getBundleStatuses).toHaveBeenCalled()
+    })
+
+    it("releases a bundle that lands while blocks are stalled", async () => {
+        const {
+            executorManager,
+            getBundleStatuses,
+            processIncludedBundle,
+            submittedBundle,
+            emitBlock,
+            getPending
+        } = createHarness()
+
+        // Not yet included, and this is the last block the chain produces.
+        getBundleStatuses
+            .mockResolvedValueOnce([{ status: "not_found" }])
+            .mockResolvedValueOnce([includedStatus])
+
+        executorManager.startWatchingBlocks()
+
+        await emitBlock()
+
+        expect(processIncludedBundle).not.toHaveBeenCalled()
+        expect(getPending()).toHaveLength(1)
+
+        // Only the watchdog can see the inclusion now.
+        await vi.advanceTimersByTimeAsync(RESUBMIT_STUCK_TIMEOUT)
+
+        expect(processIncludedBundle).toHaveBeenCalledWith(
+            expect.objectContaining({ submittedBundle })
+        )
+        // Left the pending set -> wallet released.
+        expect(getPending()).toHaveLength(0)
+    })
+
+    it("stays idle while blocks keep arriving", async () => {
+        const { executorManager, getBundleStatuses, emitBlock } =
+            createHarness()
+
+        executorManager.startWatchingBlocks()
+
+        // Fresh lastReconcileAt -> watchdog adds no reconciles of its own.
+        for (let i = 0; i < RESUBMIT_STUCK_TIMEOUT / BLOCK_TIME + 2; i++) {
+            await emitBlock()
+            await vi.advanceTimersByTimeAsync(BLOCK_TIME)
+        }
+
+        expect(getBundleStatuses).toHaveBeenCalledTimes(
+            RESUBMIT_STUCK_TIMEOUT / BLOCK_TIME + 2
+        )
+    })
+
+    it("keeps reconciling after a failed tick", async () => {
+        const { executorManager, getBundleStatuses, emitBlock } =
+            createHarness()
+
+        getBundleStatuses.mockRejectedValueOnce(new Error("store unavailable"))
+
+        executorManager.startWatchingBlocks()
+
+        // A throw used to leave currentlyHandlingBlock set forever.
+        await emitBlock().catch(() => undefined)
+        await emitBlock()
+
+        expect(getBundleStatuses).toHaveBeenCalledTimes(2)
+    })
+
+    it("clears the watchdog when the watcher stops", async () => {
+        const { executorManager, getBundleStatuses, unwatch } = createHarness()
+
+        executorManager.startWatchingBlocks()
+        executorManager.stopWatchingBlocks()
+
+        expect(unwatch).toHaveBeenCalled()
+
+        await vi.advanceTimersByTimeAsync(RESUBMIT_STUCK_TIMEOUT * 3)
+
+        expect(getBundleStatuses).not.toHaveBeenCalled()
+    })
+})

src/executor/executorManager.ts

@@ -57,6 +57,11 @@ export class ExecutorManager {
 
     private currentlyHandlingBlock = false
 
+    // When handleBlock last reconciled pending bundles, from either trigger
+    // (a new block, or the stale-block watchdog below). Drives the watchdog.
+    private lastReconcileAt = Date.now()
+    private staleBlockTimer: NodeJS.Timeout | undefined
+
     // Executor wallets rotated away from a stuck bundle whose cancel did not
     // confirm. Held out of the sender pool (markWalletProcessed deferred) until
     // their on-chain nonce advances past the stuck nonce, so no other instance
@@ -219,12 +224,57 @@ export class ExecutorManager {
                     includeTransactions: false,
                     emitMissed: false
                 })
+                this.startStaleBlockWatchdog()
             }
 
             this.logger.debug("started watching blocks")
         })
     }
 
+    // handleBlock is the only path that frees an executor wallet, and
+    // watchBlocks only fires it on a new block. On a chain that produces blocks
+    // only when it receives a transaction, that deadlocks: no submissions -> no
+    // blocks -> no wallets freed. Re-arms the time-based stuck check the block
+    // gate makes unreachable; no RPC on a healthy chain.
+    private startStaleBlockWatchdog(): void {
+        if (this.staleBlockTimer) {
+            return
+        }
+
+        this.lastReconcileAt = Date.now()
+        this.staleBlockTimer = setInterval(() => {
+            const msSinceReconcile = Date.now() - this.lastReconcileAt
+
+            if (msSinceReconcile < this.config.resubmitStuckTimeout) {
+                return
+            }
+
+            const pendingBundles = this.bundleManager.getPendingBundles().length
+            if (pendingBundles === 0) {
+                // Deliberately not stopWatchingBlocks(): re-arming a fresh
+                // watcher can miss the block the next submission triggers, so
+                // that bundle never reconciles and its wallet never frees.
+                return
+            }
+
+            this.logger.warn(
+                {
+                    event: "staleBlockWatchdogFired",
+                    msSinceReconcile,
+                    pendingBundles
+                },
+                "no new block within resubmitStuckTimeout, reconciling pending bundles on a timer"
+            )
+
+            this.handleBlock().catch((err) =>
+                this.logger.error(
+                    { err },
+                    "stale block watchdog failed to reconcile pending bundles"
+                )
+            )
+        }, this.config.blockTime)
+    }
+
     async getBaseFee(): Promise<bigint> {
         if (this.config.legacyTransactions) {
             return 0n
@@ -479,6 +529,10 @@ export class ExecutorManager {
             this.unWatch()
             this.unWatch = undefined
         }
+        if (this.staleBlockTimer) {
+            clearInterval(this.staleBlockTimer)
+            this.staleBlockTimer = undefined
+        }
     }
 
     private async updateTransactionCostMetrics(
@@ -537,29 +591,36 @@ export class ExecutorManager {
             return
         }
 
+        // try/finally, else a throw in handleBlockInner leaves this set and
+        // every later tick bails at the guard above.
+        this.currentlyHandlingBlock = true
+
         // startWatchingBlocks() registers its timers inside whichever flow
         // first started the watcher, so those timers inherit that flow's log
         // context on every future tick. Open a fresh context here so block
         // handling is never attributed to one arbitrary bundle.
-        await runWithLogContext({ flow: "block" }, () =>
-            timed(
-                this.logger,
-                "handleBlock",
-                { blockNumber: block ? Number(block.number) : undefined },
-                () => this.handleBlockInner(block)
+        try {
+            await runWithLogContext({ flow: "block" }, () =>
+                timed(
+                    this.logger,
+                    "handleBlock",
+                    { blockNumber: block ? Number(block.number) : undefined },
+                    () => this.handleBlockInner(block)
+                )
             )
-        )
+        } finally {
+            this.currentlyHandlingBlock = false
+        }
     }
 
     private async handleBlockInner(block?: Block) {
-        this.currentlyHandlingBlock = true
         const blockReceivedTimestamp = Date.now()
+        this.lastReconcileAt = blockReceivedTimestamp
 
         const pendingBundles = this.bundleManager.getPendingBundles()
 
         if (pendingBundles.length === 0) {
             this.stopWatchingBlocks()
-            this.currentlyHandlingBlock = false
             return
         }
 
@@ -623,8 +684,6 @@ export class ExecutorManager {
                 }
             })
         )
-
-        this.currentlyHandlingBlock = false
     }
 
     potentiallyResubmitBundle({
@@ -649,7 +708,7 @@ export class ExecutorManager {
             maxPriorityFeePerGas < networkGasPrice.maxPriorityFeePerGas
 
         const isStuck =
-            Date.now() - lastReplaced > this.config.resubmitStuckTimeout
+            Date.now() - lastReplaced >= this.config.resubmitStuckTimeout
 
         if (!(isGasPriceTooLow || isStuck)) {
             return