From 936da9ecf0e5ea705dfc7576a8cc2ae2db5a04a1 Mon Sep 17 00:00:00 2001 From: ozwaldorf Date: Wed, 12 Aug 2026 15:02:24 -0400 Subject: [PATCH 1/3] fix(poll): do not pin toBlock to a cached head getBlockNumber is served from viem's per-client cache, so the value pinned into getLogs could name a block past the head of a lagging node behind a load balancer, which rejects the range with "block range extends beyond current head block". Omit toBlock on the ERC20 branch so the node resolves its own head. The native branch still needs concrete block numbers, so read the head uncached and advance a per-row cursor instead of re-walking the full range every tick. Absorb transient RPC failures and retry on the next poll: the signal is already submitted, so a single bad response should not fail the transfer. --- src/internal/poll.ts | 155 ++++++++++++++++++++++++++----------------- test/poll.test.ts | 136 +++++++++++++++++++++++++++++++++++-- 2 files changed, 223 insertions(+), 68 deletions(-) diff --git a/src/internal/poll.ts b/src/internal/poll.ts index cefa035..770614c 100644 --- a/src/internal/poll.ts +++ b/src/internal/poll.ts @@ -31,13 +31,22 @@ export async function* pollTransfers(params: { pollIntervalMs?: number; signal?: AbortSignal; }): AsyncGenerator { - const { transfers, publicClient, timeout, signal, pollIntervalMs = 2000 } = params; + const { + transfers, + publicClient, + timeout, + signal, + pollIntervalMs = 2000, + } = params; const startBlock = params.fromBlock ?? (await publicClient.getBlockNumber()); const deadline = Date.now() + timeout; const pending = new Set(transfers.map((_, i) => i)); const claimed = new Set(); + // Per-row scan cursor for the native branch, so each poll resumes where the + // last one stopped instead of re-walking the whole range. + const scanned = new Map(); while (pending.size > 0) { checkAbort(signal); @@ -46,86 +55,110 @@ export async function* pollTransfers(params: { throw new TransferTimeoutError(timeout); } - const currentBlock = await publicClient.getBlockNumber(); const delivered: DeliveredTransfer[] = []; for (const index of pending) { const row = transfers[index]; - if (isNativeToken(row.tokenAddress)) { - // Native ETH arrives as a plain transaction to the recipient. - for (let blockNum = startBlock; blockNum <= currentBlock; blockNum++) { - const block = await publicClient.getBlock({ - blockNumber: blockNum, - includeTransactions: true, + try { + if (isNativeToken(row.tokenAddress)) { + // Native ETH arrives as a plain transaction to the recipient, so the + // blocks must be walked individually. Uncached: a stale head names a + // block a lagging node cannot serve yet. + const currentBlock = await publicClient.getBlockNumber({ + cacheTime: 0, }); - // Exact value, matching the ERC20 branch: a larger unrelated payment - // to the same recipient is not this delivery. - const candidates = block.transactions.filter( - (tx) => - typeof tx !== "string" && - tx.to?.toLowerCase() === row.recipientAddress.toLowerCase() && - tx.value === row.amount && - !claimed.has(tx.hash), - ); + // Blocks already walked are never re-fetched; anything past the head + // this tick is picked up by the next one. + let blockNum = scanned.get(index) ?? startBlock; + for (; blockNum <= currentBlock; blockNum++) { + const block = await publicClient.getBlock({ + blockNumber: blockNum, + includeTransactions: true, + }); - let match: (typeof candidates)[number] | undefined; - for (const candidate of candidates) { - if (typeof candidate === "string") continue; - // A reverted transaction still appears in the block but moved no - // value, so it must not count as a delivery. - const receipt = await publicClient.getTransactionReceipt({ hash: candidate.hash }); - if (receipt.status === "success") { - match = candidate; + // Exact value, matching the ERC20 branch: a larger unrelated payment + // to the same recipient is not this delivery. + const candidates = block.transactions.filter( + (tx) => + typeof tx !== "string" && + tx.to?.toLowerCase() === row.recipientAddress.toLowerCase() && + tx.value === row.amount && + !claimed.has(tx.hash), + ); + + let match: (typeof candidates)[number] | undefined; + for (const candidate of candidates) { + if (typeof candidate === "string") continue; + // A reverted transaction still appears in the block but moved no + // value, so it must not count as a delivery. + const receipt = await publicClient.getTransactionReceipt({ + hash: candidate.hash, + }); + if (receipt.status === "success") { + match = candidate; + break; + } + claimed.add(candidate.hash); + } + + if (match && typeof match !== "string") { + claimed.add(match.hash); + delivered.push({ + index, + row, + transfer: { + transactionHash: match.hash, + blockNumber: block.number, + amount: match.value, + from: match.from, + to: row.recipientAddress, + }, + }); break; } - claimed.add(candidate.hash); } + scanned.set(index, blockNum); + } else { + // No toBlock: the node resolves the upper bound against its own head. + // A pinned number can exceed the head of a lagging node behind a load + // balancer, which rejects the range outright. + const logs = await publicClient.getLogs({ + address: row.tokenAddress, + event: transferEventAbi, + args: { to: row.recipientAddress }, + fromBlock: startBlock, + }); - if (match && typeof match !== "string") { - claimed.add(match.hash); + const match = logs.find( + (log) => + log.args.value === row.amount && + !claimed.has(`${log.transactionHash}:${log.logIndex}`), + ); + + if (match) { + claimed.add(`${match.transactionHash}:${match.logIndex}`); delivered.push({ index, row, transfer: { - transactionHash: match.hash, - blockNumber: block.number, - amount: match.value, - from: match.from, - to: row.recipientAddress, + transactionHash: match.transactionHash!, + blockNumber: match.blockNumber, + amount: match.args.value!, + from: match.args.from!, + to: match.args.to!, }, }); - break; } } - } else { - const logs = await publicClient.getLogs({ - address: row.tokenAddress, - event: transferEventAbi, - args: { to: row.recipientAddress }, - fromBlock: startBlock, - toBlock: currentBlock, - }); - - const match = logs.find( - (log) => log.args.value === row.amount && !claimed.has(`${log.transactionHash}:${log.logIndex}`), - ); - - if (match) { - claimed.add(`${match.transactionHash}:${match.logIndex}`); - delivered.push({ - index, - row, - transfer: { - transactionHash: match.transactionHash!, - blockNumber: match.blockNumber, - amount: match.args.value!, - from: match.args.from!, - to: match.args.to!, - }, - }); - } + } catch (error) { + // A transient RPC failure (stale head behind a load balancer, a node + // dropping a request) must not fail the transfer: the signal is + // already submitted and the node delivers regardless. Retry on the + // next tick, still bounded by the deadline above. + checkAbort(signal); + if (error instanceof TransferTimeoutError) throw error; } } diff --git a/test/poll.test.ts b/test/poll.test.ts index b0c2f5b..1b23219 100644 --- a/test/poll.test.ts +++ b/test/poll.test.ts @@ -9,7 +9,11 @@ const R1 = "0x0000000000000000000000000000000000000001" as const; const R2 = "0x0000000000000000000000000000000000000002" as const; const R3 = "0x0000000000000000000000000000000000000003" as const; -function row(tokenAddress: string, amount: bigint, recipientAddress: string): TransferRow { +function row( + tokenAddress: string, + amount: bigint, + recipientAddress: string, +): TransferRow { return { tokenAddress: tokenAddress as `0x${string}`, recipientAddress: recipientAddress as `0x${string}`, @@ -44,7 +48,11 @@ function mockClient(rounds: any[][]) { describe("pollTransfers", () => { it("yields each recipient as it lands rather than waiting for all", async () => { - const rows = [row(USDC, 100n, R1), row(USDC, 200n, R2), row(USDC, 300n, R3)]; + const rows = [ + row(USDC, 100n, R1), + row(USDC, 200n, R2), + row(USDC, 300n, R3), + ]; const client = mockClient([ [log(R1, 100n, "0xaa")], @@ -59,7 +67,10 @@ describe("pollTransfers", () => { timeout: 5_000, pollIntervalMs: 1, })) { - seen.push({ index: delivered.index, hash: delivered.transfer.transactionHash }); + seen.push({ + index: delivered.index, + hash: delivered.transfer.transactionHash, + }); client.advance(); } @@ -91,7 +102,9 @@ describe("pollTransfers", () => { it("matches identical rows to distinct deliveries", async () => { // Two identical payments to the same recipient must consume two events. const rows = [row(USDC, 100n, R1), row(USDC, 100n, R1)]; - const client = mockClient([[log(R1, 100n, "0xaa", 0), log(R1, 100n, "0xbb", 1)]]); + const client = mockClient([ + [log(R1, 100n, "0xaa", 0), log(R1, 100n, "0xbb", 1)], + ]); const hashes = []; for await (const delivered of pollTransfers({ @@ -174,7 +187,12 @@ describe("pollTransfers", () => { getBlock: vi.fn().mockResolvedValue({ number: 10n, transactions: [ - { hash: "0xdd", to: R1, from: "0x00000000000000000000000000000000000000aa", value: 1_000n }, + { + hash: "0xdd", + to: R1, + from: "0x00000000000000000000000000000000000000aa", + value: 1_000n, + }, ], }), getLogs: vi.fn(), @@ -203,7 +221,12 @@ describe("pollTransfers", () => { getBlock: vi.fn().mockResolvedValue({ number: 10n, transactions: [ - { hash: "0xbad", to: R1, from: "0x00000000000000000000000000000000000000aa", value: 1_000n }, + { + hash: "0xbad", + to: R1, + from: "0x00000000000000000000000000000000000000aa", + value: 1_000n, + }, ], }), getLogs: vi.fn(), @@ -229,6 +252,100 @@ describe("pollTransfers", () => { ).rejects.toMatchObject({ code: "TRANSFER_ABORTED" }); }); + it("leaves toBlock unset so the node resolves its own head", async () => { + // A pinned toBlock can exceed the head of a lagging node behind a load + // balancer, which rejects the range outright. + const rows = [row(USDC, 100n, R1)]; + const client = mockClient([[log(R1, 100n, "0xaa")]]); + + for await (const _ of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + fromBlock: 50n, + pollIntervalMs: 1, + })) { + // single delivery + } + + expect(client.getLogs).toHaveBeenCalledWith( + expect.objectContaining({ fromBlock: 50n }), + ); + expect(client.getLogs.mock.calls[0][0]).not.toHaveProperty("toBlock"); + }); + + it("retries the next tick when the RPC rejects a poll", async () => { + const rows = [row(USDC, 100n, R1)]; + const client = mockClient([[log(R1, 100n, "0xaa")]]); + client.getLogs + .mockRejectedValueOnce( + new Error("block range extends beyond current head block"), + ) + .mockResolvedValueOnce([log(R1, 100n, "0xaa")]); + + const seen = []; + for await (const delivered of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + pollIntervalMs: 1, + })) { + seen.push(delivered.transfer.transactionHash); + } + + // The rejection is absorbed and the delivery still lands. + expect(seen).toEqual(["0xaa"]); + expect(client.getLogs).toHaveBeenCalledTimes(2); + }); + + it("does not re-walk native blocks already scanned", async () => { + const rows = [row(NATIVE_TOKEN_ADDRESS, 1_000n, R1)]; + let head = 10n; + const client = { + getBlockNumber: vi.fn().mockImplementation(() => { + const current = head; + if (head < 12n) head += 1n; + return Promise.resolve(current); + }), + getBlock: vi.fn().mockImplementation(({ blockNumber }: any) => + Promise.resolve({ + number: blockNumber, + transactions: + blockNumber === 12n + ? [ + { + hash: "0xdd", + to: R1, + from: "0x00000000000000000000000000000000000000aa", + value: 1_000n, + }, + ] + : [], + }), + ), + getLogs: vi.fn(), + getTransactionReceipt: vi.fn().mockResolvedValue({ status: "success" }), + } as any; + + // Head advances while polling; blocks 10-12 should each be fetched once. + const seen = []; + for await (const delivered of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + fromBlock: 10n, + pollIntervalMs: 1, + })) { + seen.push(delivered.transfer.transactionHash); + } + + expect(seen).toEqual(["0xdd"]); + const scanned = client.getBlock.mock.calls.map( + (c: any) => c[0].blockNumber, + ); + expect(scanned).toEqual([...new Set(scanned)]); + }); + it("ignores a native transaction whose value does not match exactly", async () => { const rows = [row(NATIVE_TOKEN_ADDRESS, 1_000n, R1)]; const client = { @@ -237,7 +354,12 @@ describe("pollTransfers", () => { number: 10n, // An unrelated, larger payment to the same recipient. transactions: [ - { hash: "0xbig", to: R1, from: "0x00000000000000000000000000000000000000aa", value: 5_000n }, + { + hash: "0xbig", + to: R1, + from: "0x00000000000000000000000000000000000000aa", + value: 5_000n, + }, ], }), getLogs: vi.fn(), From dc105e20629f1f91310ef85798346c69531490c4 Mon Sep 17 00:00:00 2001 From: ozwaldorf Date: Wed, 12 Aug 2026 15:08:38 -0400 Subject: [PATCH 2/3] feat(poll): scan log ranges in windows under the provider cap Providers cap eth_getLogs at ~10k blocks, so a transfer resumed long after its deploy sent a span the provider rejects outright. Walk the range in windows instead, advancing a per-row cursor so scanned windows are not refetched each tick. Only a bounded window pins toBlock; the window reaching the head stays open so the node resolves its own tip. Cap how many blocks the native branch walks per tick, since each costs a getBlock and a far-behind resume would otherwise block the tick on one long sweep. Stop retrying errors the provider will reject identically every time: a range rejection now surfaces at once instead of being absorbed until the poll deadline turns it into a bare timeout. --- src/internal/poll.ts | 109 ++++++++++++++++++++++++----- src/transfer.ts | 160 ++++++++++++++++++++++++++++++++++--------- test/poll.test.ts | 135 ++++++++++++++++++++++++++++++++++++ 3 files changed, 351 insertions(+), 53 deletions(-) diff --git a/src/internal/poll.ts b/src/internal/poll.ts index 770614c..a3ee8c9 100644 --- a/src/internal/poll.ts +++ b/src/internal/poll.ts @@ -15,6 +15,39 @@ export interface DeliveredTransfer { index: number; } +/** + * Providers cap how many blocks one eth_getLogs may span. 10k is the common + * ceiling (Alchemy, Infura); a resumed transfer deployed further back than + * that must be scanned in windows rather than one request. + */ +const DEFAULT_MAX_BLOCK_RANGE = 10_000n; + +/** + * Native deliveries cost one getBlock per block, so a far-behind resume is + * caught up across ticks rather than in one blocking sweep. + */ +const NATIVE_BLOCKS_PER_TICK = 200n; + +/** + * A malformed request fails identically on every retry, so it must surface + * instead of being absorbed until the poll deadline. Range and parameter + * rejections are permanent in that sense; a dropped connection or a node + * briefly behind the head is not. + */ +function isPermanentRpcError(error: unknown): boolean { + const message = error instanceof Error ? error.message.toLowerCase() : ""; + // A stale head resolves itself on the next tick, so it stays retryable even + // though the provider reports it as an invalid range. + if (message.includes("beyond current head")) return false; + return ( + message.includes("exceed") || + message.includes("more than") || + message.includes("too large") || + message.includes("range is too") || + message.includes("query timeout") + ); +} + /** * Watch for each recipient's delivery, yielding as they land. The node sends a * separate transaction per recipient, so a batch completes incrementally @@ -29,6 +62,7 @@ export async function* pollTransfers(params: { timeout: number; fromBlock?: bigint; pollIntervalMs?: number; + maxBlockRange?: bigint; signal?: AbortSignal; }): AsyncGenerator { const { @@ -37,6 +71,7 @@ export async function* pollTransfers(params: { timeout, signal, pollIntervalMs = 2000, + maxBlockRange = DEFAULT_MAX_BLOCK_RANGE, } = params; const startBlock = params.fromBlock ?? (await publicClient.getBlockNumber()); @@ -44,9 +79,11 @@ export async function* pollTransfers(params: { const pending = new Set(transfers.map((_, i) => i)); const claimed = new Set(); - // Per-row scan cursor for the native branch, so each poll resumes where the - // last one stopped instead of re-walking the whole range. + // Per-row scan cursors, so each poll resumes where the last one stopped + // instead of re-walking the whole range. A transfer resumed long after its + // deploy starts far behind the head and catches up across several ticks. const scanned = new Map(); + const scannedLogs = new Map(); while (pending.size > 0) { checkAbort(signal); @@ -70,9 +107,15 @@ export async function* pollTransfers(params: { }); // Blocks already walked are never re-fetched; anything past the head - // this tick is picked up by the next one. + // this tick is picked up by the next one. Each block costs a request, + // so a far-behind resume is capped per tick to stay responsive to + // abort and to the deadline rather than blocking on a long catch-up. let blockNum = scanned.get(index) ?? startBlock; - for (; blockNum <= currentBlock; blockNum++) { + const until = + currentBlock - blockNum > NATIVE_BLOCKS_PER_TICK + ? blockNum + NATIVE_BLOCKS_PER_TICK + : currentBlock; + for (; blockNum <= until; blockNum++) { const block = await publicClient.getBlock({ blockNumber: blockNum, includeTransactions: true, @@ -121,21 +164,46 @@ export async function* pollTransfers(params: { } scanned.set(index, blockNum); } else { - // No toBlock: the node resolves the upper bound against its own head. - // A pinned number can exceed the head of a lagging node behind a load - // balancer, which rejects the range outright. - const logs = await publicClient.getLogs({ - address: row.tokenAddress, - event: transferEventAbi, - args: { to: row.recipientAddress }, - fromBlock: startBlock, - }); + const fetchWindow = (fromBlock: bigint, toBlock?: bigint) => + publicClient.getLogs({ + address: row.tokenAddress, + event: transferEventAbi, + args: { to: row.recipientAddress }, + fromBlock, + ...(toBlock === undefined ? {} : { toBlock }), + }); + + let from = scannedLogs.get(index) ?? startBlock; + let match: + Awaited>[number] | undefined; + + // Walk in windows until the remainder fits under the provider's cap. + // Only a bounded window pins toBlock; the window that reaches the + // head is left open so the node resolves its own tip, since a pinned + // number can exceed the head of a lagging node behind a load + // balancer and be rejected outright. + while (!match) { + const head = await publicClient.getBlockNumber({ cacheTime: 0 }); + const bounded = head - from >= maxBlockRange; + if (!bounded && from > head) break; + + const logs = await fetchWindow( + from, + bounded ? from + maxBlockRange - 1n : undefined, + ); + + match = logs.find( + (log) => + log.args.value === row.amount && + !claimed.has(`${log.transactionHash}:${log.logIndex}`), + ); - const match = logs.find( - (log) => - log.args.value === row.amount && - !claimed.has(`${log.transactionHash}:${log.logIndex}`), - ); + if (!bounded) break; + // Only a fully-scanned window may be skipped on the next pass. + from += maxBlockRange; + scannedLogs.set(index, from); + checkAbort(signal); + } if (match) { claimed.add(`${match.transactionHash}:${match.logIndex}`); @@ -144,7 +212,7 @@ export async function* pollTransfers(params: { row, transfer: { transactionHash: match.transactionHash!, - blockNumber: match.blockNumber, + blockNumber: match.blockNumber!, amount: match.args.value!, from: match.args.from!, to: match.args.to!, @@ -159,6 +227,9 @@ export async function* pollTransfers(params: { // next tick, still bounded by the deadline above. checkAbort(signal); if (error instanceof TransferTimeoutError) throw error; + // A request the provider will reject every time must surface now + // rather than be retried until the deadline yields a bare timeout. + if (isPermanentRpcError(error)) throw error; } } diff --git a/src/transfer.ts b/src/transfer.ts index 644c330..96ed485 100644 --- a/src/transfer.ts +++ b/src/transfer.ts @@ -1,4 +1,9 @@ -import { isAddress, type Address, type PublicClient, type WalletClient } from "viem"; +import { + isAddress, + type Address, + type PublicClient, + type WalletClient, +} from "viem"; import type { ApprovalCheckpoint, AssetRequirement, @@ -69,6 +74,12 @@ export interface TransferParams { gasPrice?: GasPrice; abortSignal?: AbortSignal; pollTimeout?: number; + /** + * Largest block span one eth_getLogs may cover. Lower it for providers with + * a cap below the 10k default; a transfer resumed further back than this is + * scanned in windows. + */ + maxBlockRange?: bigint; } const DEFAULT_POLL_TIMEOUT = 120_000; @@ -81,7 +92,9 @@ function resolveRows(params: { }): TransferRow[] { const rows = params.transfers?.length ? params.transfers - : params.tokenAddress && params.recipientAddress && params.amount !== undefined + : params.tokenAddress && + params.recipientAddress && + params.amount !== undefined ? [ { tokenAddress: params.tokenAddress, @@ -106,8 +119,11 @@ function selectEscrowType(rows: TransferRow[]): EscrowKind { } function resolveSender(params: TransferParams): Address { - const walletSender = params.walletClient ? getAccount(params.walletClient) : undefined; - const sender = params.resume?.senderAddress ?? params.senderAddress ?? walletSender; + const walletSender = params.walletClient + ? getAccount(params.walletClient) + : undefined; + const sender = + params.resume?.senderAddress ?? params.senderAddress ?? walletSender; if (!sender) { throw new MirageError( "SENDER_REQUIRED", @@ -115,12 +131,17 @@ function resolveSender(params: TransferParams): Address { ); } if (walletSender && walletSender.toLowerCase() !== sender.toLowerCase()) { - throw new MirageError("ACCOUNT_CHANGED", "The active wallet does not match the quoted sender"); + throw new MirageError( + "ACCOUNT_CHANGED", + "The active wallet does not match the quoted sender", + ); } return sender; } -function attestationOptions(network: NetworkConfig): { verify: VerifyAttestationOptions | false } { +function attestationOptions(network: NetworkConfig): { + verify: VerifyAttestationOptions | false; +} { const policy = network.attestation; if (policy?.required === false) return { verify: false }; return { @@ -216,7 +237,8 @@ interface TransferContext { function isValidFundingMap(value: unknown): value is Record { if (!value || typeof value !== "object" || Array.isArray(value)) return false; return Object.entries(value).every( - ([asset, amount]) => isAddress(asset) && typeof amount === "bigint" && amount >= 0n, + ([asset, amount]) => + isAddress(asset) && typeof amount === "bigint" && amount >= 0n, ); } @@ -226,8 +248,10 @@ function quoteFromResume(params: TransferParams): PricingQuote { const escrowMatchesRows = (resume.escrowType === "batch" && rows.length > 1) || (rows.length === 1 && - ((resume.escrowType === "native" && isNativeToken(rows[0].tokenAddress)) || - (resume.escrowType === "erc20" && !isNativeToken(rows[0].tokenAddress)))); + ((resume.escrowType === "native" && + isNativeToken(rows[0].tokenAddress)) || + (resume.escrowType === "erc20" && + !isNativeToken(rows[0].tokenAddress)))); if ( !["erc20", "native", "batch"].includes(resume.escrowType) || !escrowMatchesRows || @@ -276,7 +300,10 @@ async function buildContext(params: TransferParams): Promise { params.network.chainId, attestationOptions(params.network), ); - if (networkKey.chainId !== 0 && networkKey.chainId !== params.network.chainId) { + if ( + networkKey.chainId !== 0 && + networkKey.chainId !== params.network.chainId + ) { throw new MirageError( "INVALID_NETWORK_KEY", `Nomad attested chain ${networkKey.chainId}, expected ${params.network.chainId}`, @@ -331,15 +358,26 @@ async function buildContext(params: TransferParams): Promise { blindedSigners: blinded.blindedSigners, blindingScalar: blinded.blindingScalar, quote, - fees: feeEstimate(rows, quote, approvalGasEstimate, obfuscation.deploymentGasEstimate), + fees: feeEstimate( + rows, + quote, + approvalGasEstimate, + obfuscation.deploymentGasEstimate, + ), obfuscation, }; } -function assertQuotedAccount(walletClient: WalletClient, sender: Address): Address { +function assertQuotedAccount( + walletClient: WalletClient, + sender: Address, +): Address { const account = getAccount(walletClient); if (account.toLowerCase() !== sender.toLowerCase()) { - throw new MirageError("ACCOUNT_CHANGED", "The active wallet does not match the quoted sender"); + throw new MirageError( + "ACCOUNT_CHANGED", + "The active wallet does not match the quoted sender", + ); } return account; } @@ -370,7 +408,9 @@ export interface PreviewParams { * Call `prepareTransfer` once the wallet connects for the real sender-bound * quote; the two agree on fees for identical rows. */ -export async function previewTransfer(params: PreviewParams): Promise { +export async function previewTransfer( + params: PreviewParams, +): Promise { const rows = resolveRows(params); const escrowType = selectEscrowType(rows); const [preview, obfuscation] = await Promise.all([ @@ -379,7 +419,9 @@ export async function previewTransfer(params: PreviewParams): Promise undefined), + fetchObfuscation(params.network.apiServer, escrowType).catch( + () => undefined, + ), ]); checkAbort(params.abortSignal); @@ -408,7 +450,9 @@ export async function previewTransfer(params: PreviewParams): Promise { +export async function prepareTransfer( + params: TransferParams, +): Promise { const context = await buildContext(params); let transactionGasPrice = params.gasPrice; let checkpoint: ApprovalCheckpoint | undefined; @@ -420,7 +464,10 @@ export async function prepareTransfer(params: TransferParams): Promise { if (approvalInProgress) { - throw new MirageError("INVALID_STAGE", "An approval sequence is already in progress"); + throw new MirageError( + "INVALID_STAGE", + "An approval sequence is already in progress", + ); } approvalInProgress = true; try { @@ -473,7 +520,10 @@ export async function prepareTransfer(params: TransferParams): Promise { const resume = secrets ?? deployedSecrets ?? params.resume; if (!resume) { - throw new MirageError("INVALID_STAGE", "Deploy the transfer before completing it"); + throw new MirageError( + "INVALID_STAGE", + "Deploy the transfer before completing it", + ); } yield* completeTransfer({ ...params, walletClient, resume }, context); } - async function* execute(walletClient = params.walletClient): AsyncGenerator { + async function* execute( + walletClient = params.walletClient, + ): AsyncGenerator { if (!walletClient) { - throw new MirageError("WALLET_REQUIRED", "A wallet client is required to execute a transfer"); + throw new MirageError( + "WALLET_REQUIRED", + "A wallet client is required to execute a transfer", + ); } yield { step: "fees", fees: context.fees }; if (deployedSecrets) { @@ -560,9 +618,19 @@ export async function prepareTransfer(params: TransferParams): Promise { - if (approvalBroadcast || approvalInProgress || checkpoint || deployedSecrets) { - throw new MirageError("INVALID_STAGE", "The quote is locked once approval has begun"); + async function refreshFees( + overrides: FeeRefreshOverrides = {}, + ): Promise { + if ( + approvalBroadcast || + approvalInProgress || + checkpoint || + deployedSecrets + ) { + throw new MirageError( + "INVALID_STAGE", + "The quote is locked once approval has begun", + ); } if (overrides.gasPrice) transactionGasPrice = overrides.gasPrice; const refreshedQuote = await fetchPricingQuote(params.network.apiServer, { @@ -597,14 +665,23 @@ export async function prepareTransfer(params: TransferParams): Promise { - if (approvalBroadcast || approvalInProgress || checkpoint || deployedSecrets) { - throw new MirageError("INVALID_STAGE", "Transfers are locked once approval has begun"); + if ( + approvalBroadcast || + approvalInProgress || + checkpoint || + deployedSecrets + ) { + throw new MirageError( + "INVALID_STAGE", + "Transfers are locked once approval has begun", + ); } const layoutChanged = transfers.length !== context.rows.length || transfers.some( (row, index) => - row.tokenAddress.toLowerCase() !== context.rows[index].tokenAddress.toLowerCase(), + row.tokenAddress.toLowerCase() !== + context.rows[index].tokenAddress.toLowerCase(), ); if (layoutChanged) { throw new MirageError( @@ -630,7 +707,10 @@ export async function prepareTransfer(params: TransferParams): Promise { const context = cached ?? (await buildContext(params)); @@ -654,17 +734,25 @@ async function* completeTransfer( } catch (error) { const requirement = whitelistRequirementFromError(error); if (requirement) { - throw new WhitelistRequiredError(requirement.amountUsd, requirement.thresholdUsd); + throw new WhitelistRequiredError( + requirement.amountUsd, + requirement.thresholdUsd, + ); } throw error; } if ( executionApproval.chainId !== network.chainId || executionApproval.escrowContract.toLowerCase() !== escrow.toLowerCase() || - executionApproval.deploymentTxHash.toLowerCase() !== resume.deployHash.toLowerCase() || - executionApproval.quoteCommitment.toLowerCase() !== resume.quoteCommitment.toLowerCase() + executionApproval.deploymentTxHash.toLowerCase() !== + resume.deployHash.toLowerCase() || + executionApproval.quoteCommitment.toLowerCase() !== + resume.quoteCommitment.toLowerCase() ) { - throw new MirageError("INVALID_APPROVAL", "API execution approval does not match deployment"); + throw new MirageError( + "INVALID_APPROVAL", + "API execution approval does not match deployment", + ); } yield { step: "compliance", approval: executionApproval }; @@ -689,13 +777,15 @@ async function* completeTransfer( const fromBlock = resume.fromBlock ?? - (await publicClient.getTransactionReceipt({ hash: resume.deployHash })).blockNumber; + (await publicClient.getTransactionReceipt({ hash: resume.deployHash })) + .blockNumber; const completed: TransferEvent[] = []; for await (const delivered of pollTransfers({ transfers: context.rows, publicClient, timeout: params.pollTimeout ?? DEFAULT_POLL_TIMEOUT, fromBlock, + maxBlockRange: params.maxBlockRange, signal: params.abortSignal, })) { completed.push(delivered.transfer); @@ -711,7 +801,9 @@ async function* completeTransfer( } /** Prepare and execute a transfer in one call. */ -export async function* executeTransfer(params: TransferParams): AsyncGenerator { +export async function* executeTransfer( + params: TransferParams, +): AsyncGenerator { const prepared = await prepareTransfer(params); yield* prepared.execute(params.walletClient); } diff --git a/test/poll.test.ts b/test/poll.test.ts index 1b23219..331d78e 100644 --- a/test/poll.test.ts +++ b/test/poll.test.ts @@ -252,6 +252,141 @@ describe("pollTransfers", () => { ).rejects.toMatchObject({ code: "TRANSFER_ABORTED" }); }); + it("scans in windows when the start block predates the provider cap", async () => { + // A transfer resumed long after deploy spans more than one window. + const rows = [row(USDC, 100n, R1)]; + const delivery = { ...log(R1, 100n, "0xaa"), blockNumber: 2_500n }; + const client = { + getBlockNumber: vi.fn().mockResolvedValue(2_600n), + getLogs: vi.fn().mockImplementation(({ fromBlock, toBlock }: any) => { + // A window wider than the cap is what the provider would reject. + if (toBlock !== undefined && toBlock - fromBlock >= 1_000n) { + throw new Error("query returned more than 10000 results"); + } + const upper = toBlock ?? 2_600n; + return Promise.resolve( + delivery.blockNumber >= fromBlock && delivery.blockNumber <= upper + ? [delivery] + : [], + ); + }), + getBlock: vi.fn(), + getTransactionReceipt: vi.fn(), + } as any; + + const seen = []; + for await (const d of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + fromBlock: 0n, + maxBlockRange: 1_000n, + pollIntervalMs: 1, + })) { + seen.push(d.transfer.transactionHash); + } + + expect(seen).toEqual(["0xaa"]); + // Bounded windows 0-999 and 1000-1999, then the remainder to the head + // (2000-2600 is under the cap) goes out unbounded. + const ranges = client.getLogs.mock.calls.map((c: any) => [ + c[0].fromBlock, + c[0].toBlock, + ]); + expect(ranges).toEqual([ + [0n, 999n], + [1_000n, 1_999n], + [2_000n, undefined], + ]); + }); + + it("leaves the window reaching the head unbounded", async () => { + // The last window must not pin toBlock, or a lagging node rejects it. + const rows = [row(USDC, 100n, R1)]; + const client = { + getBlockNumber: vi.fn().mockResolvedValue(500n), + getLogs: vi.fn().mockResolvedValue([log(R1, 100n, "0xaa")]), + getBlock: vi.fn(), + getTransactionReceipt: vi.fn(), + } as any; + + for await (const _ of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + fromBlock: 0n, + maxBlockRange: 1_000n, + pollIntervalMs: 1, + })) { + // single delivery + } + + expect(client.getLogs.mock.calls[0][0]).not.toHaveProperty("toBlock"); + }); + + it("surfaces a range rejection instead of retrying until timeout", async () => { + // A request the provider always rejects must not be absorbed: retrying it + // for the full poll timeout hides the real cause behind a bare timeout. + const rows = [row(USDC, 100n, R1)]; + const client = mockClient([[]]); + client.getLogs.mockRejectedValue( + new Error("query returned more than 10000 results"), + ); + + await expect( + (async () => { + for await (const _ of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + pollIntervalMs: 1, + })) { + // unreachable + } + })(), + ).rejects.toThrow(/more than 10000/); + + // Surfaced on the first attempt rather than retried. + expect(client.getLogs).toHaveBeenCalledTimes(1); + }); + + it("caps how many native blocks one tick walks", async () => { + // Each block costs a request, so a far-behind resume must not block the + // tick on a single long sweep. + const rows = [row(NATIVE_TOKEN_ADDRESS, 1_000n, R1)]; + const controller = new AbortController(); + const client = { + getBlockNumber: vi.fn().mockResolvedValue(5_000n), + getBlock: vi.fn().mockImplementation(({ blockNumber }: any) => { + // Stop once the first capped tick completes, so the count is exact + // rather than a race against a timer. + if (blockNumber >= 200n) controller.abort(); + return Promise.resolve({ number: blockNumber, transactions: [] }); + }), + getLogs: vi.fn(), + getTransactionReceipt: vi.fn(), + } as any; + + await expect( + (async () => { + for await (const _ of pollTransfers({ + transfers: rows, + publicClient: client, + timeout: 5_000, + fromBlock: 0n, + pollIntervalMs: 1, + signal: controller.signal, + })) { + // no deliveries + } + })(), + ).rejects.toMatchObject({ code: "TRANSFER_ABORTED" }); + + // Blocks 0-200 inclusive: the cap bounds the tick well short of the 5000 + // blocks an uncapped sweep would have walked before yielding. + expect(client.getBlock).toHaveBeenCalledTimes(201); + }); + it("leaves toBlock unset so the node resolves its own head", async () => { // A pinned toBlock can exceed the head of a lagging node behind a load // balancer, which rejects the range outright. From fbeebf544db8120b6b1d545161bf55b0fb285829 Mon Sep 17 00:00:00 2001 From: ozwaldorf Date: Wed, 12 Aug 2026 15:27:23 -0400 Subject: [PATCH 3/3] chore: release 0.4.3 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 629ec02..31ca941 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@mirageprivacy/sdk", - "version": "0.4.2", + "version": "0.4.3", "description": "SDK for private transfers on Mirage", "type": "module", "main": "./dist/index.cjs",