fix: guard terminal FILLED status from downgrade; reaper re-checks DB status before resolving (PROTO-1201) (#688)

* fix: guard terminal FILLED status from downgrade; reaper re-checks DB status before resolving

Follow-up hardening to #683/#687 (PROTO-1201). Order 0x33e09be0... was filled
on mainnet and recorded FILLED (txHash + settledAmounts) by check-order-status,
but a concurrent GS Reaper run -- whose fill scan predated the fill -- read the
order's used nonce as a cancellation and blindly overwrote the record to
CANCELLED. Two gaps remained:

1. GenericOrdersRepository.updateOrderStatus wrote orderStatus unconditionally,
   so any writer could downgrade a terminal FILLED order. Now a non-FILLED
   write carries a DynamoDB ConditionExpression (orderStatus <> filled), making
   the guard atomic with the write. A ConditionalCheckFailedException is logged
   and skipped rather than thrown, so callers don't enter retry loops.
   Legitimate transitions (OPEN/INSUFFICIENT_FUNDS -> FILLED, re-writing FILLED
   with fill details) are unaffected.

2. The reaper's CHECK_CANCELLED stage validated orders from the run's
   GET_OPEN_ORDERS snapshot without re-checking their current DB status. It now
   skips any order whose status no longer matches the run's unresolved-status
   snapshot (e.g. resolved to FILLED mid-run by the status state machine).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EpJxMZvZepV198FGXBNUwk

* refactor: drop reaper status param for terminal-set guard; simplify repository condition per review

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: claude[bot]

lib/crons/gs-reaper/gs-reaper.ts

@@ -8,7 +8,7 @@ import { BLOCK_RANGE, REAPER_MAX_ATTEMPTS, DYNAMO_BATCH_WRITE_MAX, OLDEST_BLOCK_
 import { ethers } from 'ethers'
 import { CosignedPriorityOrder, CosignedV2DutchOrder, CosignedV3DutchOrder, DutchOrder, FillInfo, CosignedHybridOrder, OrderType, OrderValidation, OrderValidator, REACTOR_ADDRESS_MAPPING, UniswapXEventWatcher, UniswapXOrder } from '@uniswap/uniswapx-sdk'
 import { parseOrder } from '../../handlers/OrderParser'
-import { AVERAGE_BLOCK_TIME, getSettledAmounts } from '../../handlers/check-order-status/util'
+import { AVERAGE_BLOCK_TIME, getSettledAmounts, IS_TERMINAL_STATE } from '../../handlers/check-order-status/util'
 import { ChainId } from '../../util/chain'
 import { getRpcUrl } from '../../Config'
 import { LimitOrdersRepository } from '../../repositories/limit-orders-repository'
@@ -412,7 +412,7 @@ async function checkCancelledOrders(
   provider: ethers.providers.StaticJsonRpcProvider,
   chainId: number,
   log: Logger,
-  failedFillScanRanges: BlockRange[],
+  failedFillScanRanges: BlockRange[]
 ): Promise<Record<string, OrderUpdate>> {
   const orderUpdates = { ...existingUpdates }
   const quoter = new OrderValidator(provider, chainId)
@@ -461,6 +461,12 @@ async function checkCancelledOrders(
     if (!orderUpdates[orderHash]) {
       try {
         const { order, signature, entity } = await getOrderByHash(repo, orderHash)
+        // Another writer (e.g. check-order-status) may have resolved this order since
+        // the run's GET_OPEN_ORDERS snapshot; never re-resolve a terminal order.
+        if (IS_TERMINAL_STATE(entity.orderStatus)) {
+          log.info(`Order ${orderHash} is already terminal (${entity.orderStatus}); skipping resolution`)
+          continue
+        }
         // We only check for nonce used and expired for permissioned tokens
         // since the order quoter can't move input tokens
         // For v4 orders like Hybrid, input is at a different level

lib/repositories/generic-orders-repository.ts

@@ -14,6 +14,12 @@ import { BaseOrdersRepository, OrderEntityType, QueryResult } from './base'
 import { IndexMapper } from './IndexMappers/IndexMapper'
 
 export const MAX_ORDERS = 50
+
+// aws-sdk v2 (used by dynamodb-toolbox here) sets both code and name to the error code
+function isConditionalCheckFailed(e: unknown): boolean {
+  return (e as { code?: string } | null)?.code === 'ConditionalCheckFailedException'
+}
+
 // Shared implementation for Dutch and Limit orders
 // will work for orders with the same GSIs
 export abstract class GenericOrdersRepository<
@@ -114,14 +120,27 @@ export abstract class GenericOrdersRepository<
         `cannot find order by hash when updating order status, hash: ${orderHash}`
       )
 
-      await this.entity.update({
-        [TABLE_KEY.ORDER_HASH]: orderHash,
-        ...this.indexMapper.getIndexFieldsForStatusUpdate(order, status),
-        ...(txHash && { txHash }),
-        ...(fillBlock && { fillBlock }),
-        ...(settledAmounts && { settledAmounts })
-      })
+      // FILLED is terminal; the DynamoDB condition makes the no-downgrade check atomic with the write
+      const conditions =
+        status === ORDER_STATUS.FILLED
+          ? {}
+          : { conditions: [{ attr: TABLE_KEY.ORDER_STATUS, ne: ORDER_STATUS.FILLED }] }
+      await this.entity.update(
+        {
+          [TABLE_KEY.ORDER_HASH]: orderHash,
+          ...this.indexMapper.getIndexFieldsForStatusUpdate(order, status),
+          ...(txHash && { txHash }),
+          ...(fillBlock && { fillBlock }),
+          ...(settledAmounts && { settledAmounts })
+        },
+        conditions
+      )
     } catch (e) {
+      if (isConditionalCheckFailed(e)) {
+        // the order was FILLED by another writer since our read; skip the downgrade rather than throw
+        log.warn('skipping updateOrderStatus: order is already in terminal status FILLED', { orderHash, status })
+        return
+      }
       log.error('updateOrderStatus error', { error: e })
       throw e
     }

test/integ/repositories/dynamo-repository.test.ts

@@ -723,6 +723,51 @@ describe('OrdersRepository update status test', () => {
       new Error('cannot find order by hash when updating order status, hash: nonexistent')
     )
   })
+
+  it('should not downgrade a FILLED (terminal) order to another status', async () => {
+    // Regression (PROTO-1201): a reaper run racing the check-order-status
+    // state machine must not clobber a recorded fill with CANCELLED/EXPIRED.
+    await ordersRepository.updateOrderStatus('0x1', ORDER_STATUS.FILLED, 'txHash', 1, [
+      { tokenOut: '0x1', amountOut: '1' } as SettledAmount,
+    ])
+
+    // The downgrade is skipped (not thrown), so callers don't enter retry loops.
+    await expect(ordersRepository.updateOrderStatus('0x1', ORDER_STATUS.CANCELLED)).resolves.toBeUndefined()
+    await expect(ordersRepository.updateOrderStatus('0x1', ORDER_STATUS.EXPIRED)).resolves.toBeUndefined()
+
+    await expect(ordersRepository.getByHash('0x1')).resolves.toMatchObject({
+      orderStatus: ORDER_STATUS.FILLED,
+      offerer_orderStatus: `${MOCK_ORDER_1.offerer}_${ORDER_STATUS.FILLED}`,
+      chainId_orderStatus: `${MOCK_ORDER_1.chainId}_${ORDER_STATUS.FILLED}`,
+      txHash: 'txHash',
+      fillBlock: 1,
+      settledAmounts: [{ tokenOut: '0x1', amountOut: '1' }],
+    })
+  })
+
+  it('should allow re-writing FILLED with updated fill details', async () => {
+    await ordersRepository.updateOrderStatus('0x1', ORDER_STATUS.FILLED, 'txHash2', 2)
+    await expect(ordersRepository.getByHash('0x1')).resolves.toMatchObject({
+      orderStatus: ORDER_STATUS.FILLED,
+      txHash: 'txHash2',
+      fillBlock: 2,
+    })
+  })
+
+  it('should allow non-terminal transitions and upgrades to FILLED', async () => {
+    // e.g. INSUFFICIENT_FUNDS -> FILLED (the insufficient-funds reaper finding
+    // a fill) must keep working.
+    await ordersRepository.updateOrderStatus('0x2', ORDER_STATUS.INSUFFICIENT_FUNDS)
+    await expect(ordersRepository.getByHash('0x2')).resolves.toMatchObject({
+      orderStatus: ORDER_STATUS.INSUFFICIENT_FUNDS,
+    })
+    await ordersRepository.updateOrderStatus('0x2', ORDER_STATUS.FILLED, 'txHash', 3)
+    await expect(ordersRepository.getByHash('0x2')).resolves.toMatchObject({
+      orderStatus: ORDER_STATUS.FILLED,
+      txHash: 'txHash',
+      fillBlock: 3,
+    })
+  })
 })
 
 describe('OrdersRepository delete test', () => {

test/unit/handlers/gs-reaper/gs-reaper.test.ts

@@ -352,6 +352,39 @@ describe('GSReaper', () => {
       expect(result?.orderUpdates[MOCK_ORDER_ENTITY.orderHash].status).toBe(ORDER_STATUS.CANCELLED)
     })
 
+    it('does NOT resolve an order whose DB status is already terminal', async () => {
+      // Regression (PROTO-1201): another writer (e.g. the check-order-status
+      // state machine) can resolve the order to FILLED between the run's
+      // GET_OPEN_ORDERS snapshot and CHECK_CANCELLED. A used nonce is
+      // consistent with that fill, so the reaper must skip already-terminal
+      // orders instead of clobbering FILLED with CANCELLED.
+      await mockOrdersRepository.addOrder({
+        ...MOCK_ORDER_ENTITY,
+        orderStatus: ORDER_STATUS.FILLED,
+      })
+
+      const state = {
+        chainId: ChainId.MAINNET,
+        currentBlock: OLDEST_BLOCK_BY_CHAIN[ChainId.MAINNET],
+        earliestBlock: OLDEST_BLOCK_BY_CHAIN[ChainId.MAINNET],
+        orderUpdates: {},
+        orderHashes: [MOCK_ORDER_ENTITY.orderHash],
+        failedFillScanRanges: [],
+        stage: ReaperStage.CHECK_CANCELLED
+      }
+
+      const { OrderValidation } = jest.requireActual('@uniswap/uniswapx-sdk')
+      const mockOrderValidator = jest.requireMock('@uniswap/uniswapx-sdk').OrderValidator
+      mockOrderValidator.mockImplementation(() => ({
+        validate: jest.fn().mockResolvedValue(OrderValidation.NonceUsed)
+      }))
+
+      const result = await reaper.processChainState(state)
+
+      expect(result?.stage).toBe(ReaperStage.UPDATE_DB)
+      expect(result?.orderUpdates[MOCK_ORDER_ENTITY.orderHash]).toBeUndefined()
+    })
+
     it('does NOT mark a used-nonce order CANCELLED when a failed scan range may hide its fill', async () => {
       // Regression: a used nonce is consistent with both a fill and a cancel.
       // MOCK_ORDER_ENTITY has no createdAt, so its fill window cannot be