From 0978367f923242ac7c0bc7ac942f12f4c51157b1 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 10 Jul 2026 22:08:09 +1000 Subject: [PATCH] fix(tap): collect remaining fees after partial RAV collection A query fee voucher can be paid out more than once as it grows, but the agent treated any past payout as full settlement, so a voucher collected mid-allocation was frozen and the fees earned afterwards were never claimed. Settlement is now judged against the amount actually paid out. --- .../src/allocations/__tests__/tap-v2.test.ts | 458 ++++++++++++++++++ .../src/allocations/graph-tally-collector.ts | 259 ++++++---- 2 files changed, 623 insertions(+), 94 deletions(-) create mode 100644 packages/indexer-common/src/allocations/__tests__/tap-v2.test.ts diff --git a/packages/indexer-common/src/allocations/__tests__/tap-v2.test.ts b/packages/indexer-common/src/allocations/__tests__/tap-v2.test.ts new file mode 100644 index 000000000..d0d1497c8 --- /dev/null +++ b/packages/indexer-common/src/allocations/__tests__/tap-v2.test.ts @@ -0,0 +1,458 @@ +import { + connectDatabase, + createLogger, + Logger, + toAddress, +} from '@graphprotocol/common-ts' +import { Sequelize } from 'sequelize' +import { defineQueryFeeModels, QueryFeeModels } from '../../query-fees/models' +import { GraphTallyCollector, SubgraphResponse } from '../graph-tally-collector' +import { PaymentsEscrowAccounts } from '../horizon-escrow-accounts' + +jest.mock('../horizon-escrow-accounts', () => { + const actual = jest.requireActual('../horizon-escrow-accounts') + return { ...actual, getEscrowAccounts: jest.fn() } +}) +// eslint-disable-next-line @typescript-eslint/no-var-requires +const { getEscrowAccounts } = require('../horizon-escrow-accounts') + +// Make global Jest variables available +// eslint-disable-next-line @typescript-eslint/no-explicit-any +declare const __DATABASE__: any +declare const __LOG_LEVEL__: never + +let logger: Logger +let sequelize: Sequelize +let queryFeeModels: QueryFeeModels +let graphTallyCollector: GraphTallyCollector + +const timeout = 30000 +const FINALITY_TIME = 3600 + +const INDEXER_ADDRESS = toAddress('0x6aea8894b5ab5a36cdc2d8be9290046801dd5fed') +const COLLECTOR_ADDRESS = toAddress('0x5aeef48fe943f91c39a7609049f8968f5b84414e') + +const ALLOCATION_ID_1 = toAddress('0xedde47df40c29949a75a6693c77834c00b8ad626') +const ALLOCATION_ID_2 = toAddress('0xdead47df40c29949a75a6693c77834c00b8ad624') + +const PAYER_1 = toAddress('0xffcf8fdee72ac11b5c542428b35eef5769c409f0') +const PAYER_2 = toAddress('0x9fbda871d559710256a2502a2517b794b482db40') + +const DATA_SERVICE = toAddress('0x0355b7b8cb128fa5692729ab3aaa199c1753f726') +const SERVICE_PROVIDER = INDEXER_ADDRESS + +const SIGNATURE = Buffer.from( + 'ede3f7ca5ace3629009f190bb51271f30c1aeaf565f82c25c447c7c9501f3ff31b628efcaf69138bf12960dd663924a692ee91f401785901848d8d7a639003ad1b', + 'hex', +) + +const GRT = 10n ** 18n +// A RAV that grew to 15 GRT after 5 GRT of it had already been collected on chain. +const RAV_VALUE = 15n * GRT +const PARTIALLY_COLLECTED = 5n * GRT + +// A collection id is the 20 byte allocation id right-aligned in 32 bytes. +const collectionIdFor = (allocationId: string): string => + `0x${'0'.repeat(24)}${allocationId.toLowerCase().replace(/^0x/, '')}` + +const COLLECTION_ID_1 = collectionIdFor(ALLOCATION_ID_1) +const COLLECTION_ID_2 = collectionIdFor(ALLOCATION_ID_2) + +const createRav = ( + collectionId: string, + payer: string, + redeemedAt: Date | null, + valueAggregate: bigint = RAV_VALUE, +) => ({ + collectionId, + payer, + dataService: DATA_SERVICE, + serviceProvider: SERVICE_PROVIDER, + timestampNs: 1709067401177959664n, + valueAggregate, + metadata: '', + signature: SIGNATURE, + last: true, + final: false, + redeemedAt, +}) + +// Builds the escrow accounts exactly as the subgraph would: 0x prefixed, cumulative +// tokensCollected per payer and collection. +const escrowAccountsWith = ( + collected: { payer: string; collectionId: string; tokens: bigint }[], +): PaymentsEscrowAccounts => + PaymentsEscrowAccounts.fromResponse(logger, { + paymentsEscrowAccounts: [], + graphTallyTokensCollecteds: collected.map((entry) => ({ + tokens: entry.tokens.toString(), + collectionId: entry.collectionId, + payer: { id: entry.payer }, + })), + }) + +const mockEscrow = ( + collected: { payer: string; collectionId: string; tokens: bigint }[], +) => { + getEscrowAccounts.mockResolvedValue(escrowAccountsWith(collected)) +} + +const mockTransactions = ( + transactions: { allocationId: string; payer: string; timestamp: number }[], + blockTimestamp: number, +) => { + jest + .spyOn(graphTallyCollector, 'findTransactionsForRavs') + .mockImplementation(async (): Promise => { + return { + paymentsEscrowTransactions: transactions.map((tx, index) => ({ + id: `tx-${index}`, + allocationId: tx.allocationId.toLowerCase(), + timestamp: tx.timestamp, + payer: { id: tx.payer.toLowerCase() }, + })), + _meta: { block: { timestamp: blockTimestamp, hash: 'block-hash' } }, + } + }) +} + +const setup = async () => { + logger = createLogger({ + name: 'GraphTallyCollector tests', + async: false, + level: __LOG_LEVEL__ ?? 'error', + }) + sequelize = await connectDatabase(__DATABASE__) + queryFeeModels = defineQueryFeeModels(sequelize) + sequelize = await sequelize.sync({ force: true }) + + // Instantiating through Network.create would require a live chain connection for the + // contracts, so we build the collector directly and keep the database models real. + graphTallyCollector = Object.create(GraphTallyCollector.prototype) + Object.assign(graphTallyCollector, { + logger, + models: queryFeeModels, + contracts: { GraphTallyCollector: { target: COLLECTOR_ADDRESS } }, + networkSubgraph: {}, + protocolNetwork: 'eip155:1337', + indexerAddress: INDEXER_ADDRESS, + finalityTime: FINALITY_TIME, + }) +} + +const teardownEach = async () => { + jest.restoreAllMocks() + getEscrowAccounts.mockReset() + await queryFeeModels.receiptAggregateVouchersV2.truncate({ cascade: true }) +} + +const teardownAll = async () => { + await sequelize.drop({}) + await sequelize.close() +} + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +const filterAndUpdateRavs = async (): Promise => { + const pending = await graphTallyCollector['pendingRAVs']() + return await graphTallyCollector['filterAndUpdateRavs'](pending) +} + +// Sequelize does not run the column setters on `where` clauses, so we match on the +// getters instead of asking the database for a 0x prefixed collection id. +const findRav = async (collectionId: string, payer: string) => { + const ravs = await queryFeeModels.receiptAggregateVouchersV2.findAll() + const rav = ravs.find( + (candidate) => + candidate.collectionId.toLowerCase() === collectionId.toLowerCase() && + candidate.payer.toLowerCase() === payer.toLowerCase(), + ) + if (!rav) { + throw new Error(`RAV not found for ${payer} ${collectionId}`) + } + return rav +} + +describe('GraphTallyCollector partial collection', () => { + beforeAll(setup, timeout) + afterEach(teardownEach, timeout) + afterAll(teardownAll, timeout) + + test( + 'a partially collected RAV is un-redeemed and offered for collection again', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 26 * 24 * 60 * 60 + + // The RAV was collected for 5 GRT mid allocation, then grew to 15 GRT. + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(collectedAtSecs * 1000)), + ) + mockEscrow([ + { payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: PARTIALLY_COLLECTED }, + ]) + mockTransactions( + [{ allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: collectedAtSecs }], + nowSecs, + ) + + const collectable = await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.final).toBe(false) + expect(rav.redeemedAt).toBeNull() + expect(collectable).toEqual([ + expect.objectContaining({ collectionId: COLLECTION_ID_1, payer: PAYER_1 }), + ]) + }, + timeout, + ) + + test( + 'a fully settled RAV older than the finality time becomes final', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 2 * FINALITY_TIME + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(collectedAtSecs * 1000)), + ) + mockEscrow([{ payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }]) + mockTransactions( + [{ allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: collectedAtSecs }], + nowSecs, + ) + + const collectable = await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.final).toBe(true) + expect(rav.redeemedAt).not.toBeNull() + expect(collectable).toEqual([]) + }, + timeout, + ) + + test( + 'a fully settled RAV within the finality time stays non final', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 100 + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(collectedAtSecs * 1000)), + ) + mockEscrow([{ payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }]) + mockTransactions( + [{ allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: collectedAtSecs }], + nowSecs, + ) + + const collectable = await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.final).toBe(false) + // Settled with a transaction on chain, so the stamp is kept and it is not resubmitted. + expect(rav.redeemedAt).not.toBeNull() + expect(collectable).toEqual([]) + }, + timeout, + ) + + test( + 'a submission the subgraph has not indexed yet keeps its redeemed_at stamp', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + // The subgraph is behind: its block predates the submission we just made. + const blockTimestamp = nowSecs - 600 + const submittedAtSecs = nowSecs + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(submittedAtSecs * 1000)), + ) + // On chain the collection happened, but neither the escrow accounts nor the + // transaction list reflect it yet. + mockEscrow([ + { payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: PARTIALLY_COLLECTED }, + ]) + mockTransactions([], blockTimestamp) + + await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.redeemedAt).not.toBeNull() + expect(rav.final).toBe(false) + }, + timeout, + ) + + test( + 'a settled RAV is recognised despite the database storing collection ids without a 0x prefix', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 2 * FINALITY_TIME + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(collectedAtSecs * 1000)), + ) + + // Guard the premise of this test: the raw column really has no 0x prefix. + const [rawRows] = await sequelize.query( + 'SELECT collection_id, payer FROM tap_horizon_ravs', + ) + const raw = rawRows[0] as { collection_id: string; payer: string } + expect(raw.collection_id).toBe(COLLECTION_ID_1.replace(/^0x/, '')) + expect(raw.payer).toBe(PAYER_1.toLowerCase().replace(/^0x/, '')) + + // The subgraph, by contrast, keys everything 0x prefixed. + mockEscrow([{ payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }]) + mockTransactions( + [{ allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: collectedAtSecs }], + nowSecs, + ) + + await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.final).toBe(true) + }, + timeout, + ) + + test( + 'a RAV that was never collected is left unredeemed and offered for collection', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, null), + ) + // No entry at all for this payer and collection: tokensCollected reads as 0. + mockEscrow([]) + mockTransactions([], nowSecs) + + const collectable = await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.redeemedAt).toBeNull() + expect(rav.final).toBe(false) + expect(collectable).toEqual([ + expect.objectContaining({ collectionId: COLLECTION_ID_1, payer: PAYER_1 }), + ]) + }, + timeout, + ) + + test( + 'a settled RAV is stamped with the newest collection transaction, not the oldest', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const oldestSecs = nowSecs - 26 * 24 * 60 * 60 + const newestSecs = nowSecs - 2 * FINALITY_TIME + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, null), + ) + mockEscrow([{ payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }]) + mockTransactions( + [ + { allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: oldestSecs }, + { allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: newestSecs }, + ], + nowSecs, + ) + + await graphTallyCollector['markRavsInTransactionsAsRedeemed']( + await graphTallyCollector['findTransactionsForRavs']([]), + await graphTallyCollector['pendingRAVs'](), + new Set([`${PAYER_1.toLowerCase()}-${COLLECTION_ID_1}`]), + ) + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.redeemedAt).toEqual(new Date(newestSecs * 1000)) + }, + timeout, + ) + + test( + 'marking a RAV redeemed without a timestamp stamps it with the current time', + async () => { + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, null), + ) + const beforeMs = Date.now() - 1000 + + // This is the path taken right after we submit a collection on chain. + await graphTallyCollector['markRavAsRedeemed'](COLLECTION_ID_1, PAYER_1) + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.redeemedAt).not.toBeNull() + expect(rav.redeemedAt!.getTime()).toBeGreaterThanOrEqual(beforeMs) + }, + timeout, + ) + + test( + 'a reorg that removed the collection transaction clears redeemed_at', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 2 * FINALITY_TIME + + await queryFeeModels.receiptAggregateVouchersV2.create( + createRav(COLLECTION_ID_1, PAYER_1, new Date(collectedAtSecs * 1000)), + ) + // The escrow says settled, but the transaction has vanished from the subgraph. + mockEscrow([{ payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }]) + mockTransactions([], nowSecs) + + const collectable = await filterAndUpdateRavs() + + const rav = await findRav(COLLECTION_ID_1, PAYER_1) + expect(rav.redeemedAt).toBeNull() + expect(rav.final).toBe(false) + expect(collectable).toEqual([ + expect.objectContaining({ collectionId: COLLECTION_ID_1, payer: PAYER_1 }), + ]) + }, + timeout, + ) + + test( + 'settlement is tracked per payer, so one payer finalizing does not finalize another', + async () => { + const nowSecs = Math.floor(Date.now() / 1000) + const collectedAtSecs = nowSecs - 2 * FINALITY_TIME + const redeemedAt = new Date(collectedAtSecs * 1000) + + await queryFeeModels.receiptAggregateVouchersV2.bulkCreate([ + createRav(COLLECTION_ID_1, PAYER_1, redeemedAt), + createRav(COLLECTION_ID_2, PAYER_2, redeemedAt), + ]) + // Payer 1 settled in full, payer 2 only partially. + mockEscrow([ + { payer: PAYER_1, collectionId: COLLECTION_ID_1, tokens: RAV_VALUE }, + { payer: PAYER_2, collectionId: COLLECTION_ID_2, tokens: PARTIALLY_COLLECTED }, + ]) + mockTransactions( + [ + { allocationId: ALLOCATION_ID_1, payer: PAYER_1, timestamp: collectedAtSecs }, + { allocationId: ALLOCATION_ID_2, payer: PAYER_2, timestamp: collectedAtSecs }, + ], + nowSecs, + ) + + const collectable = await filterAndUpdateRavs() + + const settled = await findRav(COLLECTION_ID_1, PAYER_1) + expect(settled.final).toBe(true) + + const partial = await findRav(COLLECTION_ID_2, PAYER_2) + expect(partial.final).toBe(false) + expect(partial.redeemedAt).toBeNull() + + expect(collectable).toEqual([ + expect.objectContaining({ collectionId: COLLECTION_ID_2, payer: PAYER_2 }), + ]) + }, + timeout, + ) +}) diff --git a/packages/indexer-common/src/allocations/graph-tally-collector.ts b/packages/indexer-common/src/allocations/graph-tally-collector.ts index 23712e048..e5eae3e40 100644 --- a/packages/indexer-common/src/allocations/graph-tally-collector.ts +++ b/packages/indexer-common/src/allocations/graph-tally-collector.ts @@ -29,7 +29,7 @@ import { SubgraphServiceContracts, } from '@graphprotocol/toolshed/deployments' import { encodeCollectQueryFeesData, PaymentTypes } from '@graphprotocol/toolshed' -import { dataSlice, hexlify, zeroPadValue, TransactionReceipt } from 'ethers' +import { dataSlice, hexlify, TransactionReceipt } from 'ethers' // every 15 minutes const RAV_CHECK_INTERVAL_MS = 900_000 @@ -429,8 +429,38 @@ export class GraphTallyCollector { // look for all transactions for that includes senderaddress[] and allocations[] const subgraphResponse = await this.findTransactionsForRavs(ravsLastNotFinal) + // The collector contract pays out (value_aggregate - tokensCollected) each time, and + // value_aggregate keeps growing while the allocation earns, so one RAV can be + // collected many times. It is settled only once tokensCollected covers its value. + const escrowAccounts = await getEscrowAccounts( + this.logger, + this.networkSubgraph, + this.indexerAddress, + this.contracts.GraphTallyCollector.target.toString(), + ) + + const settledRavKeys = new Set() + for (const rav of ravsLastNotFinal) { + const tokensCollected = escrowAccounts.getTokensCollectedForReceiver( + hexPrefixed(rav.payer), + hexPrefixed(rav.collectionId), + ) + const settled = tokensCollected >= BigInt(rav.valueAggregate) + if (settled) { + settledRavKeys.add(ravKey(rav.payer, rav.collectionId)) + } + this.logger.trace('[TAPv2] RAV settlement check', { + collectionId: rav.collectionId, + payer: rav.payer, + valueAggregate: formatGRT(rav.valueAggregate), + tokensCollected: formatGRT(tokensCollected), + settled, + }) + } + this.logger.trace('[TAPv2] Cross checking RAVs indexer database with subgraph', { subgraphResponse, + settledCount: settledRavKeys.size, ravsLastNotFinal: ravsLastNotFinal.map((rav) => ({ collectionId: rav.collectionId, payer: rav.payer, @@ -439,37 +469,39 @@ export class GraphTallyCollector { })), }) - // check for redeemed ravs in tx list but not marked as redeemed in our database - this.markRavsInTransactionsAsRedeemed(subgraphResponse, ravsLastNotFinal) + // check for settled ravs in tx list but not marked as redeemed in our database + await this.markRavsInTransactionsAsRedeemed( + subgraphResponse, + ravsLastNotFinal, + settledRavKeys, + ) - // Filter unfinalized RAVS fetched from DB, keeping RAVs that have not yet been redeemed on-chain - const nonRedeemedRavs = ravsLastNotFinal - // get all ravs that were marked as redeemed in our database - .filter((rav) => !!rav.redeemedAt) - // get all ravs that wasn't possible to find the transaction - .filter( - (rav) => + // Rows carrying a redeemed_at stamp they no longer deserve: either a chain reorg + // undid the collection transaction, or the RAV has grown since it was collected and + // is not settled. Clearing redeemed_at puts the row back on the submission list. + const ravsToClear = ravsLastNotFinal.filter( + (rav) => + !!rav.redeemedAt && + (!settledRavKeys.has(ravKey(rav.payer, rav.collectionId)) || !subgraphResponse.paymentsEscrowTransactions.find( (tx) => toAddress(rav.payer) === toAddress(tx.payer.id) && toAddress(collectionIdToAllocationId(rav.collectionId)) === toAddress(tx.allocationId), - ), - ) + )), + ) // we use the subgraph timestamp to make decisions // block timestamp minus 1 minute (because of blockchain timestamp uncertainty) const ONE_MINUTE = 60 const blockTimestampSecs = subgraphResponse._meta.block.timestamp - ONE_MINUTE - // Mark RAVs as unredeemed in DB if the TAP subgraph couldn't find the redeem Tx. - // To handle a chain reorg that "unredeemed" the RAVs. - if (nonRedeemedRavs.length > 0) { - await this.revertRavsRedeemed(nonRedeemedRavs, blockTimestampSecs) + if (ravsToClear.length > 0) { + await this.clearRavsRedeemedAt(ravsToClear, blockTimestampSecs) } - // For all RAVs that passed finality time, we mark it as final - await this.markRavsAsFinal(blockTimestampSecs) + // For all settled RAVs that passed finality time, we mark it as final + await this.markRavsAsFinal(blockTimestampSecs, settledRavKeys) return await this.models.receiptAggregateVouchersV2.findAll({ where: { redeemedAt: null, final: false, last: true }, @@ -479,41 +511,43 @@ export class GraphTallyCollector { public async markRavsInTransactionsAsRedeemed( subgraphResponse: SubgraphResponse, ravsLastNotFinal: ReceiptAggregateVoucherV2[], + settledRavKeys: Set, ) { - // get a list of transactions for ravs marked as not redeemed in our database - const redeemedRavsNotOnOurDatabase = subgraphResponse.paymentsEscrowTransactions - // get only the transactions that exists, this prevents errors marking as redeemed - // transactions for different senders with the same allocation id - .filter((tx) => { - // check if exists in the ravsLastNotFinal list - return !!ravsLastNotFinal.find( - (rav) => - // rav has the same sender address as tx - toAddress(rav.payer) === toAddress(tx.payer.id) && - // rav has the same allocation id as tx - toAddress(collectionIdToAllocationId(rav.collectionId)) === - toAddress(tx.allocationId) && - // rav was marked as not redeemed in the db - !rav.redeemedAt, - ) - }) + // The newest transaction per payer and allocation is the one that completed the + // settlement, so its timestamp is the one that starts the finality countdown. + const newestTransactionTimestamps = new Map() + for (const tx of subgraphResponse.paymentsEscrowTransactions) { + const key = `${toAddress(tx.payer.id)}-${toAddress(tx.allocationId)}` + const newest = newestTransactionTimestamps.get(key) + if (newest === undefined || tx.timestamp > newest) { + newestTransactionTimestamps.set(key, tx.timestamp) + } + } - // for each transaction that is not redeemed on our database - // but was redeemed on the blockchain, update it to redeemed - if (redeemedRavsNotOnOurDatabase.length > 0) { - for (const rav of redeemedRavsNotOnOurDatabase) { - this.logger.trace( - '[TAPv2] Found transaction for RAV that was redeemed on the blockchain but not on our database, marking it as redeemed', - { - rav, - }, - ) - await this.markRavAsRedeemed( - zeroPadValue(rav.allocationId, 32), - rav.payer.id, - rav.timestamp, - ) + // Only stamp redeemed_at on a RAV that has a collection transaction on chain AND + // whose full value has been paid out. A partially collected RAV stays unredeemed so + // the rest of its value still gets collected. + for (const rav of ravsLastNotFinal) { + if (rav.redeemedAt || !settledRavKeys.has(ravKey(rav.payer, rav.collectionId))) { + continue + } + const timestamp = newestTransactionTimestamps.get( + `${toAddress(rav.payer)}-${toAddress( + collectionIdToAllocationId(rav.collectionId), + )}`, + ) + if (timestamp === undefined) { + continue } + this.logger.trace( + '[TAPv2] Found transaction for RAV that was fully collected on the blockchain but not marked as redeemed on our database, marking it as redeemed', + { + collectionId: rav.collectionId, + payer: rav.payer, + timestamp, + }, + ) + await this.markRavAsRedeemed(rav.collectionId, rav.payer, timestamp) } } @@ -607,67 +641,84 @@ export class GraphTallyCollector { } } - // for every allocation_id of this list that contains the redeemedAt less than the current - // subgraph timestamp - private async revertRavsRedeemed( - ravsNotRedeemed: { collectionId: string; payer: string }[], + // The redeemed_at guard leaves a RAV submitted moments ago alone: the subgraph has not + // indexed its transaction yet, so we would otherwise clear a stamp that is still valid. + private async clearRavsRedeemedAt( + ravsToClear: { collectionId: string; payer: string }[], blockTimestampSecs: number, ) { - if (ravsNotRedeemed.length == 0) { + if (ravsToClear.length == 0) { return } - this.logger.trace( - '[TAPv2] Could not find transaction for RAV that was redeemed on the database, unsetting redeemed_at', - { - ravsNotRedeemed, - }, - ) + this.logger.trace('[TAPv2] Unsetting redeemed_at for RAVs that are not settled', { + ravsToClear: ravsToClear.map((rav) => ({ + collectionId: rav.collectionId, + payer: rav.payer, + })), + }) - // WE use sql directly due to a bug in sequelize update: - // https://github.com/sequelize/sequelize/issues/7664 (bug been open for 7 years no fix yet or ever) + // We use raw SQL because of a bug in sequelize update: + // https://github.com/sequelize/sequelize/issues/7664 (open for 7 years, no fix yet) const query = ` UPDATE tap_horizon_ravs SET redeemed_at = NULL - WHERE (collection_id::char(64), payer::char(40)) IN (VALUES ${ravsNotRedeemed - .map( - (rav) => - `('${rav.collectionId - .toString() - .toLowerCase() - .replace('0x', '')}'::char(64), '${rav.payer - .toString() - .toLowerCase() - .replace('0x', '')}'::char(40))`, - ) - .join(', ')}) - AND redeemed_at < to_timestamp(${blockTimestampSecs}) + WHERE (collection_id, payer) IN ( + SELECT * FROM unnest($1::char(64)[], $2::char(40)[]) + ) + AND redeemed_at < to_timestamp($3) ` - await this.models.receiptAggregateVouchersV2.sequelize?.query(query) + await this.models.receiptAggregateVouchersV2.sequelize?.query(query, { + bind: [ + ravsToClear.map((rav) => dbCollectionId(rav.collectionId)), + ravsToClear.map((rav) => dbPayer(rav.payer)), + blockTimestampSecs, + ], + }) this.logger.warn( - `[TAPv2] Reverted Redeemed RAVs: ${ravsNotRedeemed + `[TAPv2] Cleared redeemed_at for RAVs: ${ravsToClear .map((rav) => `(${rav.payer},${rav.collectionId})`) .join(', ')}`, ) } - // we use blockTimestamp instead of NOW() because we must be older than - // the subgraph timestamp - private async markRavsAsFinal(blockTimestampSecs: number) { + // Only settled RAVs may be finalized: finalizing one that still has value left to + // collect would hide it from the submission list forever and strand its query fees. + // We use blockTimestamp instead of NOW() because we must be older than the subgraph. + private async markRavsAsFinal(blockTimestampSecs: number, settledRavKeys: Set) { + if (settledRavKeys.size === 0) { + this.logger.debug('[TAPv2] No settled RAVs to mark as final') + return + } + + const settled = [...settledRavKeys].map((key) => { + const [payer, collectionId] = key.split('-') + return { payer, collectionId } + }) const query = ` UPDATE tap_horizon_ravs SET final = TRUE - WHERE last = TRUE - AND final = FALSE + WHERE (collection_id, payer) IN ( + SELECT * FROM unnest($1::char(64)[], $2::char(40)[]) + ) + AND last = TRUE + AND final = FALSE AND redeemed_at IS NOT NULL - AND redeemed_at < to_timestamp(${blockTimestampSecs - this.finalityTime}) + AND redeemed_at < to_timestamp($3) ` - const result = await this.models.receiptAggregateVouchersV2.sequelize?.query(query) + const result = await this.models.receiptAggregateVouchersV2.sequelize?.query(query, { + bind: [ + settled.map((rav) => dbCollectionId(rav.collectionId)), + settled.map((rav) => dbPayer(rav.payer)), + blockTimestampSecs - this.finalityTime, + ], + }) this.logger.debug('[TAPv2] Marked RAVs as final', { result, + settledCount: settledRavKeys.size, blockTimestampSecs, finalityTime: this.finalityTime, threshold: blockTimestampSecs - this.finalityTime, @@ -1056,22 +1107,42 @@ export class GraphTallyCollector { payer: string, timestamp?: number, ) { - // WE use sql directly due to a bug in sequelize update: - // https://github.com/sequelize/sequelize/issues/7664 (bug been open for 7 years no fix yet or ever) + // We use raw SQL because of a bug in sequelize update: + // https://github.com/sequelize/sequelize/issues/7664 (open for 7 years, no fix yet) const query = ` UPDATE tap_horizon_ravs - SET redeemed_at = ${timestamp ? `to_timestamp(${timestamp})` : 'NOW()'} - WHERE collection_id = '${collectionId - .toString() - .toLowerCase() - .replace('0x', '')}' - AND payer = '${payer.toString().toLowerCase().replace('0x', '')}' + SET redeemed_at = COALESCE(to_timestamp($3::double precision), NOW()) + WHERE collection_id = $1 + AND payer = $2 ` - await this.models.receiptAggregateVouchersV2.sequelize?.query(query) + await this.models.receiptAggregateVouchersV2.sequelize?.query(query, { + bind: [dbCollectionId(collectionId), dbPayer(payer), timestamp ?? null], + }) } } +// The database stores collection ids and addresses lowercased and without the 0x prefix, +// while the subgraph, the contracts and the signed RAV objects all keep the prefix. +function dbCollectionId(collectionId: string): string { + return collectionId.toString().toLowerCase().replace(/^0x/, '') +} + +function dbPayer(payer: string): string { + return payer.toString().toLowerCase().replace(/^0x/, '') +} + +function hexPrefixed(value: string): string { + const lowercased = value.toString().toLowerCase() + return lowercased.startsWith('0x') ? lowercased : `0x${lowercased}` +} + +// Keyed the way the escrow accounts key their tokensCollected lookup, so a RAV read from +// the database and one read from the subgraph always agree on identity. +function ravKey(payer: string, collectionId: string): string { + return `${hexPrefixed(payer)}-${hexPrefixed(collectionId)}` +} + const registerReceiptMetrics = (metrics: Metrics, networkIdentifier: string) => ({ ravRedeemsSuccess: new metrics.client.Counter({ name: `indexer_agent_rav_v2_exchanges_ok_${networkIdentifier}`,