fix(session): pin channel open/top-up read-back to receipt block

.changeset/session-open-readback-block-pin.md

@@ -0,0 +1,5 @@
+---
+'mppx': patch
+---
+
+Fixed a read-after-write race in Tempo session channel open/top-up verification. The on-chain read-back now pins to the block that included the management transaction (instead of reading `latest`) and retries transient failures, so a lagging load-balanced RPC replica can no longer reject a valid open with `on-chain channel state does not match open receipt`.

src/tempo/session/precompile/Chain.test.ts

@@ -51,6 +51,7 @@
     channel?: { descriptor: Channel.ChannelDescriptor; state: Chain.ChannelState } | undefined
     receipt?: Record<string, unknown> | null | undefined
     rpcMethods?: string[] | undefined
+    onRequest?: ((method: string, params: unknown) => void) | undefined
   } = {},
 ) {
   return createClient({
@@ -59,6 +60,7 @@
     transport: custom({
       async request(args) {
         parameters.rpcMethods?.push(args.method)
+        parameters.onRequest?.(args.method, args.params)
         if (args.method === 'eth_chainId') return `0x${chainId.toString(16)}`
         if (args.method === 'eth_sendRawTransaction') return txHash
         if (args.method === 'eth_sendRawTransactionSync') return parameters.receipt ?? receipt([])
@@ -232,7 +234,7 @@
     abi: escrowAbi,
     functionName: 'topUp',
     args: [
       parameters.descriptor_ ?? descriptor,
       Types.uint96(parameters.additionalDeposit ?? deposit),
     ],
   })
@@ -641,6 +643,44 @@
     })
   })
 
+  test('pins the read-back to the block that included the open transaction', async () => {
+    const serializedTransaction = await createOpenTransaction()
+    const expiringNonceHash = expectedExpiringNonceHash(serializedTransaction)
+    const expectedDescriptor = { ...descriptor, expiringNonceHash }
+    const channelId = Channel.computeId({
+      ...expectedDescriptor,
+      chainId,
+      escrow: tip20ChannelEscrow,
+    })
+    const state = { settled: 0n, deposit, closeRequestedAt: 0 }
+    // The open tx is mined in block 0x1 (see `receipt`). The read-back must be
+    // pinned to that block, not `latest`, so a lagging RPC replica cannot
+    // answer with stale/empty state.
+    const ethCallBlockTags: unknown[] = []
+
+    await Chain.broadcastOpenTransaction({
+      chainId,
+      client: createMockClient({
+        channel: { descriptor: expectedDescriptor, state },
+        receipt: receipt([openedLog({ channelId, expiringNonceHash })]),
+        onRequest: (method, params) => {
+          if (method === 'eth_call') ethCallBlockTags.push((params as unknown[])[1])
+        },
+      }),
+      escrowContract: tip20ChannelEscrow,
+      expectedAuthorizedSigner: descriptor.authorizedSigner,
+      expectedChannelId: channelId,
+      expectedCurrency: descriptor.token,
+      expectedExpiringNonceHash: expiringNonceHash,
+      expectedOperator: descriptor.operator,
+      expectedPayee: descriptor.payee,
+      expectedPayer: descriptor.payer,
+      serializedTransaction,
+    })
+
+    expect(ethCallBlockTags).toEqual(['0x1'])
+  })
+
   test('rejects ChannelOpened receipt deposit mismatches', async () => {
     const serializedTransaction = await createOpenTransaction()
     const expiringNonceHash = expectedExpiringNonceHash(serializedTransaction)
@@ -1256,3 +1296,47 @@
     }
   })
 })
+
+describe('Chain.readbackWithRetry', () => {
+  test('returns immediately when the read succeeds on the first attempt', async () => {
+    let calls = 0
+    const result = await Chain.readbackWithRetry(
+      async () => {
+        calls++
+        return 'ok'
+      },
+      { delayMs: 0 },
+    )
+    expect(result).toBe('ok')
+    expect(calls).toBe(1)
+  })
+
+  test('retries a lagging read until it resolves', async () => {
+    let calls = 0
+    const result = await Chain.readbackWithRetry(
+      async () => {
+        calls++
+        if (calls < 3) throw new Error('header not found')
+        return 'ok'
+      },
+      { retries: 5, delayMs: 0 },
+    )
+    expect(result).toBe('ok')
+    expect(calls).toBe(3)
+  })
+
+  test('rethrows the last error after exhausting retries', async () => {
+    let calls = 0
+    await expect(
+      Chain.readbackWithRetry(
+        async () => {
+          calls++
+          throw new Error(`attempt ${calls}`)
+        },
+        { retries: 2, delayMs: 0 },
+      ),
+    ).rejects.toThrow('attempt 3')
+    // 1 initial attempt + 2 retries.
+    expect(calls).toBe(3)
+  })
+})

src/tempo/session/precompile/Chain.ts

@@ -293,12 +293,14 @@ export async function getChannel(
   client: Client,
   descriptor: ChannelDescriptor,
   escrow: Address = tip20ChannelEscrow,
+  blockNumber?: bigint,
 ): Promise<Channel> {
   const channel = await readContract(client, {
     address: escrow,
     abi: escrowAbi,
     functionName: 'getChannel',
     args: [descriptorTuple(descriptor)],
+    ...(blockNumber !== undefined ? { blockNumber } : {}),
   })
   return {
     descriptor: channel.descriptor,
@@ -313,12 +315,14 @@ export async function getChannelState(
   client: Client,
   channelId: Hex,
   escrow: Address = tip20ChannelEscrow,
+  blockNumber?: bigint,
 ): Promise<ChannelState> {
   const state = await readContract(client, {
     address: escrow,
     abi: escrowAbi,
     functionName: 'getChannelState',
     args: [channelId],
+    ...(blockNumber !== undefined ? { blockNumber } : {}),
   })
   return stateFromTuple(state)
 }
@@ -340,6 +344,47 @@ export async function getChannelStatesBatch(
   return states.map(stateFromTuple)
 }
 
+/** Tuning for {@link readbackWithRetry}. */
+export type ReadbackRetryOptions = {
+  /** Additional attempts after the first, before giving up. @default 5 */
+  retries?: number | undefined
+  /** Delay between attempts, in milliseconds. @default 250 */
+  delayMs?: number | undefined
+}
+
+/**
+ * Retry an on-chain readback that is pinned to the block containing a just-sent
+ * transaction.
+ *
+ * The escrow state read is served by a load-balanced RPC whose replicas can lag
+ * behind the block that produced the transaction receipt. Callers pin the read
+ * to that block number (via the `blockNumber` arg on {@link getChannel} /
+ * {@link getChannelState}) so a node that has imported the block returns
+ * authoritative state; replicas that have not yet imported it throw (e.g.
+ * "header not found"), so we retry with a short backoff until one catches up.
+ *
+ * This closes the read-after-write race behind
+ * `on-chain channel state does not match open receipt` — without it, a stale
+ * `latest` read on a lagging replica returns an empty channel and fails
+ * verification even though the open transaction succeeded.
+ */
+export async function readbackWithRetry<T>(
+  read: () => Promise<T>,
+  options: ReadbackRetryOptions = {},
+): Promise<T> {
+  const { retries = 5, delayMs = 250 } = options
+  let lastError: unknown
+  for (let attempt = 0; attempt <= retries; attempt++) {
+    try {
+      return await read()
+    } catch (error) {
+      lastError = error
+      if (attempt < retries) await new Promise((resolve) => setTimeout(resolve, delayMs))
+    }
+  }
+  throw lastError
+}
+
 /** Options accepted by low-level TIP-1034 on-chain management helpers. */
 export type ChannelTransactionOptions = {
   /** Account used to send the transaction when the viem client has no default account. */
@@ -808,7 +853,9 @@ export async function broadcastOpenTransaction(
     expectedChannelId: parameters.expectedChannelId,
     openDeposit: open.deposit,
   })
-  const chainChannel = await getChannel(parameters.client, descriptor, parameters.escrowContract)
+  const chainChannel = await readbackWithRetry(() =>
+    getChannel(parameters.client, descriptor, parameters.escrowContract, receipt.blockNumber),
+  )
   const state = chainChannel.state
   validateOpenReadbackState({ emittedDeposit: opened.deposit, state })
   return {
@@ -896,10 +943,13 @@ export async function broadcastTopUpTransaction(
     emittedChannelId: toppedUp.channelId,
     expectedChannelId: parameters.expectedChannelId,
   })
-  const state = await getChannelState(
-    parameters.client,
-    toppedUp.channelId,
-    parameters.escrowContract,
+  const state = await readbackWithRetry(() =>
+    getChannelState(
+      parameters.client,
+      toppedUp.channelId,
+      parameters.escrowContract,
+      receipt.blockNumber,
+    ),
   )
   validateTopUpReadbackState({ newDeposit: toppedUp.newDeposit, state })
   return { txHash: receipt.transactionHash, newDeposit: toppedUp.newDeposit, state }