From 9e6322e6e623860c271e232c79e53415bd14d652 Mon Sep 17 00:00:00 2001 From: David Salami Date: Mon, 20 Jul 2026 15:19:49 +0000 Subject: [PATCH 1/2] [indexer]: Index BidPlaced events from the intents-coprocessor pallet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a FillerBid entity recording who bid on which order commitment, along with the raw bid payload. The payload is read out of the place_bid extrinsic in the same block rather than from intents_getBidsForOrder: that RPC pins the chain head internally and has no `at` parameter, and the pallet's offchain copy expires, so on any backfill it would answer with today's live bids and silently attribute them to a historical block. The RPC remains as a fallback for when the call cannot be decoded. place_bid may arrive wrapped in batch/proxy/sudo, so the call tree is walked to find it, matching on commitment — one batch can carry bids for several orders. Handler is registered on hyperbridge chains rather than under enablePriceIndexing, which is mainnet-only, since the pallet runs on gargantua too. --- .../templates/substrate-chain.yaml.hbs | 5 ++ .../indexer/src/configs/schema.graphql | 49 ++++++++++++ .../handleBidPlaced.handler.ts | 58 ++++++++++++++ .../indexer/src/mappings/mappingHandlers.ts | 1 + .../utils/__tests__/extrinsic.helpers.test.ts | 78 +++++++++++++++++++ sdk/packages/indexer/src/utils/bid-data.ts | 49 ++++++++++++ .../indexer/src/utils/extrinsic.helpers.ts | 73 +++++++++++++++++ 7 files changed, 313 insertions(+) create mode 100644 sdk/packages/indexer/src/handlers/events/substrateChains/handleBidPlaced.handler.ts create mode 100644 sdk/packages/indexer/src/utils/__tests__/extrinsic.helpers.test.ts create mode 100644 sdk/packages/indexer/src/utils/bid-data.ts create mode 100644 sdk/packages/indexer/src/utils/extrinsic.helpers.ts diff --git a/sdk/packages/indexer/scripts/templates/substrate-chain.yaml.hbs b/sdk/packages/indexer/scripts/templates/substrate-chain.yaml.hbs index 2907954bb..42afa0365 100644 --- a/sdk/packages/indexer/scripts/templates/substrate-chain.yaml.hbs +++ b/sdk/packages/indexer/scripts/templates/substrate-chain.yaml.hbs @@ -67,6 +67,11 @@ dataSources: filter: module: bandwidth method: TierSet + - handler: handleBidPlaced + kind: substrate/EventHandler + filter: + module: intentsCoprocessor + method: BidPlaced - handler: handleBridgeTokenSupplyIndexing kind: substrate/BlockHandler filter: diff --git a/sdk/packages/indexer/src/configs/schema.graphql b/sdk/packages/indexer/src/configs/schema.graphql index 3cc3ab45d..54cdfc132 100644 --- a/sdk/packages/indexer/src/configs/schema.graphql +++ b/sdk/packages/indexer/src/configs/schema.graphql @@ -2238,6 +2238,55 @@ type PhantomOrder @entity { blockTimestamp: Date @index } +""" +A bid placed by a filler on an order commitment, recorded from the intents-coprocessor pallet's +BidPlaced event. + +The bid payload is read out of the place_bid extrinsic in the same block, falling back to the node's +offchain storage via intents_getBidsForOrder — that RPC pins the chain head and the offchain entry +expires, so it only answers for bids still live on the queried node. + +Fillers may re-bid on the same commitment (the pallet unreserves the previous deposit and emits a +fresh BidPlaced), so one row is written per event rather than per (commitment, filler) — the bid +history is preserved. The commitment usually belongs to a PhantomOrder, but place_bid accepts any +commitment, so the field is a plain reference rather than a foreign key. +""" +type FillerBid @entity { + """ + Composite identifier: {commitment}-{filler}-{blockNumber}-{eventIdx}. Unique per BidPlaced event, + so a filler's repeat bids on the same commitment are kept as separate rows. + """ + id: ID! + + """ + The order commitment bid on — keccak256(abi.encode(Order)). Matches PhantomOrder.id when the bid + targets a phantom order. + """ + commitment: String! @index + + """ + SS58 address of the filler that placed the bid. + """ + filler: String! @index + + """ + The raw bid payload — the SCALE-encoded PackedUserOperation the filler submitted, as hex. Stored + undecoded so a bid stays fully re-interpretable as the fill ABI evolves, and because the pallet's + own copy expires. Null when neither the extrinsic nor the RPC yielded it. + """ + bidData: String + + """ + Hash of the extrinsic that placed the bid, when the event is attributable to one. + """ + extrinsicHash: String + + """ + Hyperbridge block number the bid was placed at. + """ + blockNumber: BigInt! @index +} + """ The price snapshot for a phantom order, written once its bid window closes (signalled by the PhantomBidWindowExhausted event). It collects all live bids via intents_getBidsForOrder and keeps diff --git a/sdk/packages/indexer/src/handlers/events/substrateChains/handleBidPlaced.handler.ts b/sdk/packages/indexer/src/handlers/events/substrateChains/handleBidPlaced.handler.ts new file mode 100644 index 000000000..3b984f8f7 --- /dev/null +++ b/sdk/packages/indexer/src/handlers/events/substrateChains/handleBidPlaced.handler.ts @@ -0,0 +1,58 @@ +import { SubstrateEvent } from "@subql/types" + +import { wrap } from "@/utils/event.utils" +import { replaceWebsocketWithHttp } from "@/utils/rpc.helpers" +import { getHostStateMachine } from "@/utils/substrate.helpers" +import { resolveBidData } from "@/utils/bid-data" +import { ENV_CONFIG } from "@/constants" +import { FillerBid } from "@/configs/src/types" + +/** + * `pallet-intents-coprocessor :: BidPlaced` — a filler placed a bid on an order commitment. + * + * Payload order: + * 0. filler: AccountId + * 1. commitment: H256 + * 2. deposit: Balance + * + * The event carries no bid payload, so it is resolved separately by resolveBidData and stored raw. + */ +export const handleBidPlaced = wrap(async (event: SubstrateEvent): Promise => { + const { + event: { data }, + block, + extrinsic, + } = event + + const [fillerData, commitmentData] = data + + const filler = fillerData.toString() + const commitment = commitmentData.toHex() + const blockNumber = block.block.header.number.toBigInt() + + // A filler may re-bid on the same commitment, so the block number and event index are part of the + // key — keyed on (commitment, filler) alone, a repeat bid would overwrite its predecessor. + const id = `${commitment}-${filler}-${blockNumber}-${event.idx}` + if (await FillerBid.get(id)) return + + const host = getHostStateMachine(chainId) + + const bidData = await resolveBidData({ + extrinsic, + commitment, + // The RPC keys bids by the raw AccountId bytes, not the SS58 form stored on the entity. + fillerHex: fillerData.toHex(), + nodeUrl: replaceWebsocketWithHttp(ENV_CONFIG[host] ?? "") || undefined, + }) + + await FillerBid.create({ + id, + commitment, + filler, + bidData, + extrinsicHash: extrinsic?.extrinsic.hash.toString(), + blockNumber, + }).save() + + logger.info({ commitment, filler, blockNumber }, `FillerBid indexed${bidData ? "" : " (no bid data)"}`) +}) diff --git a/sdk/packages/indexer/src/mappings/mappingHandlers.ts b/sdk/packages/indexer/src/mappings/mappingHandlers.ts index a72758748..1105a717a 100644 --- a/sdk/packages/indexer/src/mappings/mappingHandlers.ts +++ b/sdk/packages/indexer/src/mappings/mappingHandlers.ts @@ -23,6 +23,7 @@ export { handleDustSweptEventV3 } from "@/handlers/events/intentGatewayV3/dustSw export { handleIsmpStateMachineUpdatedEvent } from "@/handlers/events/substrateChains/handleIsmpStateMachineUpdatedEvent.handler" export { handlePhantomOrderRegistered } from "@/handlers/events/substrateChains/handlePhantomOrderRegistered.handler" export { handlePhantomOrderPrices } from "@/handlers/events/substrateChains/handlePhantomOrderPrices.handler" +export { handleBidPlaced } from "@/handlers/events/substrateChains/handleBidPlaced.handler" export { handleSubstratePostRequestTimeoutHandledEvent } from "@/handlers/events/substrateChains/handlePostRequestTimeoutHandledEvent.handler" export { handleSubstrateRequestEvent } from "@/handlers/events/substrateChains/handleRequestEvent.handler" export { handleSubstrateResponseEvent } from "@/handlers/events/substrateChains/handleResponseEvent.handler" diff --git a/sdk/packages/indexer/src/utils/__tests__/extrinsic.helpers.test.ts b/sdk/packages/indexer/src/utils/__tests__/extrinsic.helpers.test.ts new file mode 100644 index 000000000..85d9be74f --- /dev/null +++ b/sdk/packages/indexer/src/utils/__tests__/extrinsic.helpers.test.ts @@ -0,0 +1,78 @@ +import { SubstrateExtrinsic } from "@subql/types" +import { extractUserOpFromExtrinsic } from "@/utils/extrinsic.helpers" + +// The user operation is read out of the place_bid extrinsic rather than intents_getBidsForOrder, +// because that RPC pins the chain head internally and the pallet's offchain copy expires — so on any +// backfill the extrinsic is the only source that still has the bid. place_bid can arrive wrapped in +// batch/proxy/sudo, so the walk has to find it at depth, and has to match on commitment: one batch +// may carry bids for several orders, and picking the wrong one would attribute another order's quote +// to this event. + +const COMMITMENT = `0x${"11".repeat(32)}` +const OTHER_COMMITMENT = `0x${"22".repeat(32)}` +const USER_OP = "0xdeadbeef" +const OTHER_USER_OP = "0xfeedface" + +const hex = (value: string) => ({ toHex: () => value }) + +const placeBid = (commitment: string, userOp: string) => ({ + section: "intentsCoprocessor", + method: "placeBid", + args: [hex(commitment), hex(userOp)], +}) + +const batch = (calls: unknown[]) => ({ + section: "utility", + method: "batchAll", + args: [calls], +}) + +const proxy = (call: unknown) => ({ + section: "proxy", + method: "proxy", + args: [hex("0xaaaa"), hex("0x00"), call], +}) + +const asExtrinsic = (method: unknown) => ({ extrinsic: { method } }) as unknown as SubstrateExtrinsic + + +describe("extractUserOpFromExtrinsic", () => { + it("reads the user op from a direct place_bid call", () => { + expect(extractUserOpFromExtrinsic(asExtrinsic(placeBid(COMMITMENT, USER_OP)), COMMITMENT)).toBe(USER_OP) + }) + + it("finds a place_bid nested in a batch", () => { + const call = batch([placeBid(OTHER_COMMITMENT, OTHER_USER_OP), placeBid(COMMITMENT, USER_OP)]) + expect(extractUserOpFromExtrinsic(asExtrinsic(call), COMMITMENT)).toBe(USER_OP) + }) + + it("finds a place_bid nested in a proxied batch", () => { + const call = proxy(batch([placeBid(COMMITMENT, USER_OP)])) + expect(extractUserOpFromExtrinsic(asExtrinsic(call), COMMITMENT)).toBe(USER_OP) + }) + + it("returns the user op of the matching commitment, not the first bid in the batch", () => { + const call = batch([placeBid(OTHER_COMMITMENT, OTHER_USER_OP), placeBid(COMMITMENT, USER_OP)]) + expect(extractUserOpFromExtrinsic(asExtrinsic(call), OTHER_COMMITMENT)).toBe(OTHER_USER_OP) + }) + + it("returns undefined when no place_bid matches the commitment", () => { + const call = batch([placeBid(OTHER_COMMITMENT, OTHER_USER_OP)]) + expect(extractUserOpFromExtrinsic(asExtrinsic(call), COMMITMENT)).toBeUndefined() + }) + + it("returns undefined for an unrelated extrinsic", () => { + const call = { section: "balances", method: "transfer", args: [hex("0xaaaa"), hex("0x01")] } + expect(extractUserOpFromExtrinsic(asExtrinsic(call), COMMITMENT)).toBeUndefined() + }) + + it("returns undefined when the event has no extrinsic", () => { + expect(extractUserOpFromExtrinsic(undefined, COMMITMENT)).toBeUndefined() + }) + + it("does not loop on a self-referential call tree", () => { + const cyclic: any = { section: "utility", method: "batchAll", args: [] } + cyclic.args = [[cyclic]] + expect(extractUserOpFromExtrinsic(asExtrinsic(cyclic), COMMITMENT)).toBeUndefined() + }) +}) diff --git a/sdk/packages/indexer/src/utils/bid-data.ts b/sdk/packages/indexer/src/utils/bid-data.ts new file mode 100644 index 000000000..c3580085c --- /dev/null +++ b/sdk/packages/indexer/src/utils/bid-data.ts @@ -0,0 +1,49 @@ +import { SubstrateExtrinsic } from "@subql/types" +import { fetchBidsForOrder, setAggregationFetch } from "@hyperbridge/sdk/intents-helpers" + +import { safeFetch } from "@/utils/safeFetch" +import { extractUserOpFromExtrinsic } from "@/utils/extrinsic.helpers" + +// The SDK's RPC helpers run inside the SubQuery VM2 sandbox, which has no global `fetch`. +setAggregationFetch(safeFetch) + +/** + * Falls back to the node's offchain storage for a bid's payload. + * + * intents_getBidsForOrder pins the chain head internally and has no `at` parameter, and the + * underlying offchain entry is node-local and expires, so this only returns anything while the bid + * is still live on the node being queried. It is a best-effort backstop for the extrinsic path, not + * a replacement for it. + */ +async function fetchBidDataFromRpc(nodeUrl: string, commitment: string, fillerHex: string): Promise { + try { + const bids = await fetchBidsForOrder(nodeUrl, commitment) + const match = bids.find((bid) => bid.filler?.toLowerCase() === fillerHex.toLowerCase()) + return match?.user_op || undefined + } catch (err) { + logger.warn({ err, commitment }, "intents_getBidsForOrder failed for bid enrichment") + return undefined + } +} + +/** + * Resolves the raw bid payload behind a BidPlaced event, or undefined if neither source has it. + * + * The extrinsic is tried first because it is part of the block being indexed: it is exact (the very + * bid that raised this event), always present on replay, and costs no network call. The RPC is only + * a backstop for the case where the call cannot be decoded. + */ +export async function resolveBidData(params: { + extrinsic?: SubstrateExtrinsic + commitment: string + fillerHex: string + nodeUrl?: string +}): Promise { + const { extrinsic, commitment, fillerHex, nodeUrl } = params + + const fromExtrinsic = extractUserOpFromExtrinsic(extrinsic, commitment) + if (fromExtrinsic) return fromExtrinsic + + if (!nodeUrl) return undefined + return fetchBidDataFromRpc(nodeUrl, commitment, fillerHex) +} diff --git a/sdk/packages/indexer/src/utils/extrinsic.helpers.ts b/sdk/packages/indexer/src/utils/extrinsic.helpers.ts new file mode 100644 index 000000000..4d11c72dc --- /dev/null +++ b/sdk/packages/indexer/src/utils/extrinsic.helpers.ts @@ -0,0 +1,73 @@ +import { SubstrateExtrinsic } from "@subql/types" + +/** + * Finds a call within an extrinsic's call tree and returns one of its arguments. + * + * Dispatches routinely arrive wrapped — utility.batch, proxy.proxy, sudo — so the call that raised + * an event is often not the extrinsic's top-level method. `match` picks the call of interest (it + * receives the call's decoded args, so it can discriminate between sibling calls of the same kind), + * and its return value is what comes back. + * + * Kept free of SDK imports so it stays cheap to unit test. + */ +export function findInCallTree( + extrinsic: SubstrateExtrinsic | undefined, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + match: (call: { section: string; method: string; args: any[] }) => T | undefined, +): T | undefined { + if (!extrinsic) return undefined + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const visit = (call: any, depth: number): T | undefined => { + // Wrapped calls nest at most a few levels; the bound just stops a malformed or cyclic tree + // from looping forever. + if (!call || depth > 6) return undefined + + const args = call.args + if (!Array.isArray(args)) return undefined + + if (typeof call.section === "string" && typeof call.method === "string") { + const matched = match(call) + if (matched !== undefined) return matched + } + + for (const arg of args) { + // A nested call arg carries section/method itself; batch-style args carry an array of them. + if (Array.isArray(arg)) { + for (const inner of arg) { + const found = visit(inner, depth + 1) + if (found !== undefined) return found + } + } else if (arg?.section && arg?.method) { + const found = visit(arg, depth + 1) + if (found !== undefined) return found + } + } + + return undefined + } + + try { + return visit(extrinsic.extrinsic.method, 0) + } catch { + return undefined + } +} + +/** + * Pulls the `user_op` argument out of the place_bid call that raised a BidPlaced event. + * + * Matches on commitment, not just on the call being a place_bid: one batch may carry bids for + * several orders, and taking the first would attribute another order's quote to this event. + */ +export function extractUserOpFromExtrinsic( + extrinsic: SubstrateExtrinsic | undefined, + commitment: string, +): string | undefined { + return findInCallTree(extrinsic, (call) => { + if (call.section !== "intentsCoprocessor" || call.method !== "placeBid") return undefined + const [commitmentArg, userOpArg] = call.args + if (commitmentArg?.toHex?.() !== commitment) return undefined + return userOpArg?.toHex?.() as string | undefined + }) +} From 4b83e4c5ec1e5dc59b7431f609f048ee92bea4c2 Mon Sep 17 00:00:00 2001 From: David Salami Date: Tue, 21 Jul 2026 08:38:45 +0000 Subject: [PATCH 2/2] [sdk, simplex]: Poll for phantom orders instead of subscribing subscribePhantomOrders watched system.events over the WsProvider. When the socket dropped, polkadot-js reconnected the transport but did not reliably re-establish the storage subscription, and every event emitted while disconnected was lost with no error. The only error handling was a catch at startup, so nothing ever checked liveness again. With a bid window measured in a handful of blocks, that meant silently missed bids. pollPhantomOrders reads the head each tick and scans every block between the last one processed and that head, so the block cursor rather than the connection determines what has been seen. This is gap-free rather than merely self-healing: an outage delays orders but cannot drop them, since the cursor only advances past a block whose events were actually read, and recovery replays the backlog. The cursor advances per block rather than per range, so a failure partway through a catch-up re-scans only the unread blocks. maxBlocksPerPoll bounds a single scan so a long outage catches up over several ticks, and lookbackBlocks lets a restarting process pick up orders whose window is still open. Note this tracks the best head, not the finalized head: a reorg can drop a block whose order was already bid on. The deposit is retractable, and waiting for finality would miss the shorter bid windows entirely. --- .../sdk/src/chains/intentsCoprocessor.ts | 117 ++++++++-- .../sdk/src/tests/pollPhantomOrders.test.ts | 203 ++++++++++++++++++ sdk/packages/simplex/src/core/filler.ts | 23 +- 3 files changed, 315 insertions(+), 28 deletions(-) create mode 100644 sdk/packages/sdk/src/tests/pollPhantomOrders.test.ts diff --git a/sdk/packages/sdk/src/chains/intentsCoprocessor.ts b/sdk/packages/sdk/src/chains/intentsCoprocessor.ts index 266de9842..2d322435e 100644 --- a/sdk/packages/sdk/src/chains/intentsCoprocessor.ts +++ b/sdk/packages/sdk/src/chains/intentsCoprocessor.ts @@ -95,6 +95,24 @@ export interface PhantomOrderEvent { standardAmount: bigint } +export interface PollPhantomOrdersOptions { + /** How often to check for a new head. Defaults to 6s, roughly one block. */ + intervalMs?: number + /** + * Most blocks scanned in a single poll, so a long outage catches up over several ticks instead of + * one unbounded scan. Defaults to 500. + */ + maxBlocksPerPoll?: number + /** + * How many blocks before the current head to start from on the first poll. Defaults to 0 (start + * at the head). Set this to the runtime's bid window to have a restarting process pick up orders + * whose window is still open. + */ + lookbackBlocks?: number + /** Notified when a poll fails; polling continues regardless. */ + onError?: (err: unknown) => void +} + /** * Service for interacting with Hyperbridge's pallet-intents coprocessor. * Handles bid submission and retrieval for the IntentGatewayV2 protocol. @@ -607,26 +625,89 @@ export class IntentsCoprocessor { } /** - * Subscribes to PhantomOrderRegistered events from the intents coprocessor pallet. - * Calls the callback for each new phantom order as blocks arrive. - * Returns an unsubscribe function to stop the subscription. + * Reads the PhantomOrderRegistered events emitted in a single block. */ - async subscribePhantomOrders(callback: (event: PhantomOrderEvent) => void): Promise<() => void> { + async getPhantomOrdersInBlock(blockNumber: number): Promise { + const blockHash = await this.api.rpc.chain.getBlockHash(blockNumber) + const apiAt = await this.api.at(blockHash) + const records = await apiAt.query.system.events() + + const orders: PhantomOrderEvent[] = [] // eslint-disable-next-line @typescript-eslint/no-explicit-any - const unsub = await (this.api.query.system.events as any)((records: any[]) => { - for (const { event } of records) { - if (event.section !== "intentsCoprocessor" || event.method !== "PhantomOrderRegistered") continue - const [commitment, chain, createdAt, tokenA, tokenB, standardAmount] = event.data - callback({ - commitment: commitment.toHex() as HexString, - chain: new TextDecoder().decode(hexToU8a(chain.toHex())), - createdAt: createdAt.toNumber(), - tokenA: tokenA.toHex() as HexString, - tokenB: tokenB.toHex() as HexString, - standardAmount: BigInt(standardAmount.toString()), - }) + for (const { event } of records as unknown as Array<{ event: any }>) { + if (event.section !== "intentsCoprocessor" || event.method !== "PhantomOrderRegistered") continue + const [commitment, chain, createdAt, tokenA, tokenB, standardAmount] = event.data + orders.push({ + commitment: commitment.toHex() as HexString, + chain: new TextDecoder().decode(hexToU8a(chain.toHex())), + createdAt: createdAt.toNumber(), + tokenA: tokenA.toHex() as HexString, + tokenB: tokenB.toHex() as HexString, + standardAmount: BigInt(standardAmount.toString()), + }) + } + return orders + } + + /** + * Polls for newly registered phantom orders, invoking the callback once per order. + * + * Each tick reads the current head and scans every block between the last one processed and that + * head, so the block cursor — not the connection — determines what has been seen. This replaced a + * system.events subscription, which was only as reliable as its socket: polkadot-js reconnects + * the transport but does not reliably re-establish storage subscriptions, and anything emitted + * while disconnected was lost silently. With a bid window measured in a handful of blocks, that + * meant silently missed bids. + * + * Scanning a block range is gap-free rather than merely self-healing: an outage delays orders but + * cannot drop them, because the cursor only advances past a block whose events were actually + * read. Recovery replays the backlog. + * + * Returns a function that stops polling. + */ + pollPhantomOrders(callback: (event: PhantomOrderEvent) => void, options: PollPhantomOrdersOptions = {}): () => void { + const { intervalMs = 6_000, maxBlocksPerPoll = 500, lookbackBlocks = 0, onError } = options + + // Last block whose events have been delivered. Null until the first successful head read. + let cursor: number | null = null + let inFlight = false + let stopped = false + + const tick = async (): Promise => { + // A scan slower than the interval must not stack up behind itself. + if (inFlight || stopped) return + inFlight = true + try { + const head = (await this.api.rpc.chain.getHeader()).number.toNumber() + + if (cursor === null) { + // Start just below the head so the head itself is scanned, less any lookback. + cursor = Math.max(head - 1 - lookbackBlocks, -1) + } + if (head <= cursor) return + + const to = Math.min(head, cursor + maxBlocksPerPoll) + for (let blockNumber = cursor + 1; blockNumber <= to; blockNumber++) { + if (stopped) return + const orders = await this.getPhantomOrdersInBlock(blockNumber) + for (const order of orders) callback(order) + // Advance per block, not per range: a failure partway through re-scans only the + // blocks that were never read, and never re-delivers ones that were. + cursor = blockNumber + } + } catch (err) { + onError?.(err) + } finally { + inFlight = false } - }) - return unsub as () => void + } + + void tick() + const timer = setInterval(() => void tick(), intervalMs) + + return () => { + stopped = true + clearInterval(timer) + } } } diff --git a/sdk/packages/sdk/src/tests/pollPhantomOrders.test.ts b/sdk/packages/sdk/src/tests/pollPhantomOrders.test.ts new file mode 100644 index 000000000..58a719f45 --- /dev/null +++ b/sdk/packages/sdk/src/tests/pollPhantomOrders.test.ts @@ -0,0 +1,203 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest" +import { IntentsCoprocessor, type PhantomOrderEvent } from "@/chains/intentsCoprocessor" + +// Polling replaced a system.events subscription because a dropped socket silently stopped delivering +// phantom orders — polkadot-js reconnects the transport but does not reliably re-establish storage +// subscriptions, and anything emitted while disconnected was gone. The property that makes a block +// cursor an actual fix, rather than a different way to lose orders, is that it advances only past +// blocks whose events were really read, so an outage delays orders instead of dropping them. That is +// what most of these assert. + +const COMMITMENT_A = `0x${"aa".repeat(32)}` +const COMMITMENT_B = `0x${"bb".repeat(32)}` +const TOKEN_A = `0x${"11".repeat(20)}` +const TOKEN_B = `0x${"22".repeat(20)}` + +const CHAIN = "EVM-8453" +const CHAIN_HEX = `0x${Buffer.from(CHAIN, "utf8").toString("hex")}` + +/** A PhantomOrderRegistered record shaped the way polkadot-js decodes it. */ +function registeredEvent(commitment: string) { + return { + event: { + section: "intentsCoprocessor", + method: "PhantomOrderRegistered", + data: [ + { toHex: () => commitment }, + { toHex: () => CHAIN_HEX }, + { toNumber: () => 7 }, + { toHex: () => TOKEN_A }, + { toHex: () => TOKEN_B }, + { toString: () => "1000000" }, + ], + }, + } +} + +const unrelatedEvent = { event: { section: "balances", method: "Transfer", data: [] } } + +interface Harness { + coprocessor: IntentsCoprocessor + /** Blocks scanned, in order. */ + scanned: number[] + setHead: (n: number) => void + /** Registers an order in a block; blocks without one still scan clean. */ + putOrder: (blockNumber: number, commitment: string) => void + /** Makes the next `count` head reads throw, simulating a dropped connection. */ + failHeadReads: (count: number) => void +} + +function harness(initialHead: number): Harness { + let head = initialHead + let headFailures = 0 + const ordersByBlock = new Map() + const scanned: number[] = [] + + const getHeader = vi.fn(async () => { + if (headFailures > 0) { + headFailures -= 1 + throw new Error("websocket disconnected") + } + return { number: { toNumber: () => head } } + }) + + const getBlockHash = vi.fn(async (n: number) => `0xblock${n}`) + + const at = vi.fn(async (blockHash: string) => { + const blockNumber = Number(blockHash.replace("0xblock", "")) + scanned.push(blockNumber) + const commitment = ordersByBlock.get(blockNumber) + const records = commitment ? [unrelatedEvent, registeredEvent(commitment)] : [unrelatedEvent] + return { query: { system: { events: async () => records } } } + }) + + const coprocessor = Object.create(IntentsCoprocessor.prototype) as IntentsCoprocessor + Object.assign(coprocessor, { api: { rpc: { chain: { getHeader, getBlockHash } }, at } }) + + return { + coprocessor, + scanned, + setHead: (n) => { + head = n + }, + putOrder: (blockNumber, commitment) => ordersByBlock.set(blockNumber, commitment), + failHeadReads: (count) => { + headFailures = count + }, + } +} + +/** Advances fake timers and lets the awaited scan settle. */ +const tick = (ms: number) => vi.advanceTimersByTimeAsync(ms) + +describe("pollPhantomOrders", () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it("emits an order registered in the head block, fully decoded", async () => { + const h = harness(100) + h.putOrder(100, COMMITMENT_A) + const seen: PhantomOrderEvent[] = [] + + const stop = h.coprocessor.pollPhantomOrders((e) => seen.push(e), { intervalMs: 1000 }) + await tick(0) + stop() + + expect(seen).toEqual([ + { + commitment: COMMITMENT_A, + chain: CHAIN, + createdAt: 7, + tokenA: TOKEN_A, + tokenB: TOKEN_B, + standardAmount: 1000000n, + }, + ]) + }) + + it("scans each block exactly once as the head advances", async () => { + const h = harness(100) + const stop = h.coprocessor.pollPhantomOrders(() => {}, { intervalMs: 1000 }) + + await tick(0) + h.setHead(103) + await tick(1000) + stop() + + expect(h.scanned).toEqual([100, 101, 102, 103]) + }) + + it("does not rescan when the head has not moved", async () => { + const h = harness(100) + const stop = h.coprocessor.pollPhantomOrders(() => {}, { intervalMs: 1000 }) + + await tick(5000) + stop() + + expect(h.scanned).toEqual([100]) + }) + + // The subscription's failure mode: orders registered while disconnected were lost outright. + it("delivers orders registered during an outage once the connection recovers", async () => { + const h = harness(100) + const seen: PhantomOrderEvent[] = [] + const onError = vi.fn() + + const stop = h.coprocessor.pollPhantomOrders((e) => seen.push(e), { intervalMs: 1000, onError }) + await tick(0) + + // Two ticks fail while the chain moves on and registers an order at 102. + h.putOrder(102, COMMITMENT_B) + h.setHead(103) + h.failHeadReads(2) + await tick(2000) + expect(seen).toHaveLength(0) + expect(onError).toHaveBeenCalledTimes(2) + + await tick(1000) + stop() + + expect(seen.map((e) => e.commitment)).toEqual([COMMITMENT_B]) + expect(h.scanned).toEqual([100, 101, 102, 103]) + }) + + it("caps how many blocks a single poll scans and catches up over later ticks", async () => { + const h = harness(100) + const stop = h.coprocessor.pollPhantomOrders(() => {}, { intervalMs: 1000, maxBlocksPerPoll: 2 }) + + await tick(0) + h.setHead(106) + await tick(1000) + expect(h.scanned).toEqual([100, 101, 102]) + + await tick(1000) + stop() + + expect(h.scanned).toEqual([100, 101, 102, 103, 104]) + }) + + it("starts lookbackBlocks behind the head so a restart mid-window still bids", async () => { + const h = harness(100) + h.putOrder(98, COMMITMENT_A) + const seen: PhantomOrderEvent[] = [] + + const stop = h.coprocessor.pollPhantomOrders((e) => seen.push(e), { intervalMs: 1000, lookbackBlocks: 3 }) + await tick(0) + stop() + + expect(h.scanned).toEqual([97, 98, 99, 100]) + expect(seen.map((e) => e.commitment)).toEqual([COMMITMENT_A]) + }) + + it("stops scanning once stopped", async () => { + const h = harness(100) + const stop = h.coprocessor.pollPhantomOrders(() => {}, { intervalMs: 1000 }) + + await tick(0) + stop() + h.setHead(110) + await tick(5000) + + expect(h.scanned).toEqual([100]) + }) +}) diff --git a/sdk/packages/simplex/src/core/filler.ts b/sdk/packages/simplex/src/core/filler.ts index 70660a67f..0dd07fe58 100644 --- a/sdk/packages/simplex/src/core/filler.ts +++ b/sdk/packages/simplex/src/core/filler.ts @@ -43,7 +43,7 @@ export class IntentFiller { private pendingRetractions = new Set() private rebalancingInterval?: NodeJS.Timeout private retractionSweepInterval?: NodeJS.Timeout - private phantomUnsubscribe: (() => void) | null = null + private stopPhantomPolling: (() => void) | null = null // Last phantom bid commitment per phantom-order series — keyed by chain + the directed token // pair, NOT by chain alone. The pallet generates one phantom order per configured token pair, so // several are live on the same chain at once; a new interval's bid must only retract the previous @@ -285,9 +285,9 @@ export class IntentFiller { public async stop(): Promise { this.monitor.stopListening() - if (this.phantomUnsubscribe) { - this.phantomUnsubscribe() - this.phantomUnsubscribe = null + if (this.stopPhantomPolling) { + this.stopPhantomPolling() + this.stopPhantomPolling = null } // Stop rebalancing interval @@ -755,14 +755,17 @@ export class IntentFiller { private startPhantomBidding(): void { if (!this.hyperbridge) return this.hyperbridge - .then(async (coprocessor) => { - this.phantomUnsubscribe = await coprocessor.subscribePhantomOrders((event) => { - this.globalQueue.add(() => this.handlePhantomOrder(event, coprocessor)) - }) - this.logger.info("Phantom order subscription active") + .then((coprocessor) => { + this.stopPhantomPolling = coprocessor.pollPhantomOrders( + (order) => { + this.globalQueue.add(() => this.handlePhantomOrder(order, coprocessor)) + }, + { onError: (err) => this.logger.warn({ err }, "Phantom order poll failed, will retry") }, + ) + this.logger.info("Phantom order polling active") }) .catch((err) => { - this.logger.error({ err }, "Failed to start phantom order subscription") + this.logger.error({ err }, "Failed to start phantom order polling") }) }