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", diff --git a/src/internal/poll.ts b/src/internal/poll.ts index cefa035..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,15 +62,28 @@ export async function* pollTransfers(params: { timeout: number; fromBlock?: bigint; pollIntervalMs?: number; + maxBlockRange?: bigint; signal?: AbortSignal; }): AsyncGenerator { - const { transfers, publicClient, timeout, signal, pollIntervalMs = 2000 } = params; + const { + transfers, + publicClient, + timeout, + signal, + pollIntervalMs = 2000, + maxBlockRange = DEFAULT_MAX_BLOCK_RANGE, + } = 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 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); @@ -46,86 +92,144 @@ 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), - ); - - 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; + // Blocks already walked are never re-fetched; anything past the head + // 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; + 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, + }); + + // 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 { + const fetchWindow = (fromBlock: bigint, toBlock?: bigint) => + publicClient.getLogs({ + address: row.tokenAddress, + event: transferEventAbi, + args: { to: row.recipientAddress }, + fromBlock, + ...(toBlock === undefined ? {} : { toBlock }), + }); - if (match && typeof match !== "string") { - claimed.add(match.hash); + 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}`), + ); + + 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}`); 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; + // 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 b0c2f5b..331d78e 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,235 @@ 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. + 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 +489,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(),