diff --git a/src/executor/executorManager.test.ts b/src/executor/executorManager.test.ts new file mode 100644 index 00000000..77759df3 --- /dev/null +++ b/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) | 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 + }) => { + 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() + }) +}) diff --git a/src/executor/executorManager.ts b/src/executor/executorManager.ts index ccbeff33..58cf0dab 100644 --- a/src/executor/executorManager.ts +++ b/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 { 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