From 0e050b9a56c6c5cd8b6ce8de7aae0c020fadf383 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 10 Jul 2026 22:26:22 +1000 Subject: [PATCH 1/3] fix(tap): collect high-value RAVs before dust and surface deferrals Query fee vouchers are collected 100 rows at a time, previously in no defined order, so more than 100 vouchers below the 1 GRT collection threshold could hide a newly closed high-value one forever. Take the highest value first, and expose the deferred count and value as metrics. --- .../__tests__/pending-ravs-order.test.ts | 243 ++++++++++++++++++ .../src/allocations/graph-tally-collector.ts | 38 ++- .../src/allocations/tap-collector.ts | 33 ++- 3 files changed, 305 insertions(+), 9 deletions(-) create mode 100644 packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts diff --git a/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts new file mode 100644 index 000000000..c81b71743 --- /dev/null +++ b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts @@ -0,0 +1,243 @@ +// Imported from source rather than from '@graphprotocol/indexer-common', whose main +// entry point is the compiled ./dist bundle. Going through the package would test +// whatever was last built instead of the code in this working tree. +import { defineQueryFeeModels, QueryFeeModels } from '../../query-fees/models' +import { TapCollector } from '../tap-collector' +import { GraphTallyCollector } from '../graph-tally-collector' +import { connectDatabase, createLogger, Logger, toAddress } from '@graphprotocol/common-ts' +import { Sequelize } from 'sequelize' + +// 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 + +const SIGNATURE = Uint8Array.from(Buffer.alloc(65, 1)) +const PAYER = toAddress('deadbeefcafebabedeadbeefcafebabedeadbeef') +const DATA_SERVICE = toAddress('0000000000000000000000000000000000000001') +const SERVICE_PROVIDER = toAddress('0000000000000000000000000000000000000002') + +// Values are deliberately inserted out of order so a passing assertion cannot be +// explained by Postgres happening to return rows in insertion order. +const V1_RAVS = [ + { allocationId: toAddress('1111111111111111111111111111111111111111'), value: 5n }, + { allocationId: toAddress('2222222222222222222222222222222222222222'), value: 100n }, + { allocationId: toAddress('3333333333333333333333333333333333333333'), value: 1n }, + { allocationId: toAddress('4444444444444444444444444444444444444444'), value: 50n }, +] + +const V2_RAVS = [ + { collectionId: `0x${'a'.repeat(64)}`, value: 7n }, + { collectionId: `0x${'b'.repeat(64)}`, value: 900n }, + { collectionId: `0x${'c'.repeat(64)}`, value: 3n }, + { collectionId: `0x${'d'.repeat(64)}`, value: 42n }, +] + +const setup = async () => { + logger = createLogger({ + name: 'pending-ravs-order', + async: false, + level: __LOG_LEVEL__ ?? 'error', + }) + sequelize = await connectDatabase(__DATABASE__) + queryFeeModels = defineQueryFeeModels(sequelize) + sequelize = await sequelize.sync({ force: true }) +} + +const teardown = async () => { + await sequelize.drop({}) + await sequelize.close() +} + +beforeAll(setup, 30000) +afterAll(teardown, 30000) +beforeEach(async () => { + sequelize = await sequelize.sync({ force: true }) +}) + +describe('pendingRAVs ordering', () => { + test('TAPv1 returns pending RAVs ordered by value, highest first', async () => { + // Arrange + for (const { allocationId, value } of V1_RAVS) { + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId, + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: value, + last: true, + final: false, + redeemedAt: null, + }) + } + const collector: TapCollector = Object.assign( + Object.create(TapCollector.prototype), + { logger, models: queryFeeModels }, + ) + + // Act + const ravs = await collector['pendingRAVs']() + + // Assert + expect(ravs.map((rav) => rav.valueAggregate)).toEqual([100n, 50n, 5n, 1n]) + }) + + test('TAPv2 returns pending RAVs ordered by value, highest first', async () => { + // Arrange + for (const { collectionId, value } of V2_RAVS) { + await queryFeeModels.receiptAggregateVouchersV2.create({ + collectionId, + payer: PAYER, + dataService: DATA_SERVICE, + serviceProvider: SERVICE_PROVIDER, + signature: SIGNATURE, + metadata: '0x', + timestampNs: 1n, + valueAggregate: value, + last: true, + final: false, + redeemedAt: null, + }) + } + const collector: GraphTallyCollector = Object.assign( + Object.create(GraphTallyCollector.prototype), + { logger, models: queryFeeModels }, + ) + + // Act + const ravs = await collector['pendingRAVs']() + + // Assert + expect(ravs.map((rav) => rav.valueAggregate)).toEqual([900n, 42n, 7n, 3n]) + }) + + test('TAPv1 ordering surfaces high value RAVs from beyond the 100 row limit', async () => { + // Arrange: 100 dust RAVs that would fill the batch on their own, plus one + // valuable RAV inserted last so insertion order alone would exclude it. + for (let i = 0; i < 100; i++) { + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId: toAddress(i.toString(16).padStart(40, '0')), + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: 1n, + last: true, + final: false, + redeemedAt: null, + }) + } + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId: toAddress('ffffffffffffffffffffffffffffffffffffffff'), + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: 10n ** 21n, // 1,000 GRT + last: true, + final: false, + redeemedAt: null, + }) + const collector: TapCollector = Object.assign( + Object.create(TapCollector.prototype), + { logger, models: queryFeeModels }, + ) + + // Act + const ravs = await collector['pendingRAVs']() + + // Assert + expect(ravs).toHaveLength(100) + expect(ravs[0].valueAggregate).toEqual(10n ** 21n) + }) + + test('TAPv2 ordering surfaces high value RAVs from beyond the 100 row limit', async () => { + // Arrange + for (let i = 0; i < 100; i++) { + await queryFeeModels.receiptAggregateVouchersV2.create({ + collectionId: `0x${i.toString(16).padStart(64, '0')}`, + payer: PAYER, + dataService: DATA_SERVICE, + serviceProvider: SERVICE_PROVIDER, + signature: SIGNATURE, + metadata: '0x', + timestampNs: 1n, + valueAggregate: 1n, + last: true, + final: false, + redeemedAt: null, + }) + } + await queryFeeModels.receiptAggregateVouchersV2.create({ + collectionId: `0x${'f'.repeat(64)}`, + payer: PAYER, + dataService: DATA_SERVICE, + serviceProvider: SERVICE_PROVIDER, + signature: SIGNATURE, + metadata: '0x', + timestampNs: 1n, + valueAggregate: 10n ** 21n, // 1,000 GRT + last: true, + final: false, + redeemedAt: null, + }) + const collector: GraphTallyCollector = Object.assign( + Object.create(GraphTallyCollector.prototype), + { logger, models: queryFeeModels }, + ) + + // Act + const ravs = await collector['pendingRAVs']() + + // Assert + expect(ravs).toHaveLength(100) + expect(ravs[0].valueAggregate).toEqual(10n ** 21n) + }) + + test('pendingRAVs excludes final and non-last RAVs regardless of value', async () => { + // Arrange + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId: V1_RAVS[0].allocationId, + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: 10n ** 21n, + last: true, + final: true, // already finalized + redeemedAt: null, + }) + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId: V1_RAVS[1].allocationId, + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: 10n ** 21n, + last: false, // superseded by a newer RAV + final: false, + redeemedAt: null, + }) + await queryFeeModels.receiptAggregateVouchers.create({ + allocationId: V1_RAVS[2].allocationId, + senderAddress: PAYER, + signature: SIGNATURE, + timestampNs: 1n, + valueAggregate: 5n, + last: true, + final: false, + redeemedAt: null, + }) + const collector: TapCollector = Object.assign( + Object.create(TapCollector.prototype), + { logger, models: queryFeeModels }, + ) + + // Act + const ravs = await collector['pendingRAVs']() + + // Assert + expect(ravs).toHaveLength(1) + expect(ravs[0].valueAggregate).toEqual(5n) + }) +}) diff --git a/packages/indexer-common/src/allocations/graph-tally-collector.ts b/packages/indexer-common/src/allocations/graph-tally-collector.ts index 23712e048..ca0cbbaae 100644 --- a/packages/indexer-common/src/allocations/graph-tally-collector.ts +++ b/packages/indexer-common/src/allocations/graph-tally-collector.ts @@ -43,6 +43,8 @@ interface RavMetrics { ravRedeemsFailed: Counter ravsRedeemDuration: Histogram ravCollectedFees: Gauge + ravsBelowThreshold: Gauge + ravsBelowThresholdValueGRT: Gauge } interface TapCollectorOptions { @@ -59,6 +61,9 @@ interface TapCollectorOptions { interface ValidRavs { belowThreshold: RavWithAllocation[] eligible: RavWithAllocation[] + // Sum of what the below threshold RAVs would still pay out, which for TAPv2 is the + // aggregate value minus whatever the payer has already collected against it. + belowThresholdRemaining: bigint } export interface RavWithAllocation { @@ -162,6 +167,12 @@ export class GraphTallyCollector { const notifyAndMapEligible = (signedRavs: ValidRavs) => { const logger = this.logger.child({ function: 'startRAVProcessingV2()' }) + // Set every pass, including to 0, so the gauges decay once deferrals clear + this.metrics.ravsBelowThreshold.set(signedRavs.belowThreshold.length) + this.metrics.ravsBelowThresholdValueGRT.set( + parseFloat(formatGRT(signedRavs.belowThresholdRemaining)), + ) + if (signedRavs.belowThreshold.length > 0) { const totalValueGRT = formatGRT( signedRavs.belowThreshold.reduce( @@ -174,6 +185,9 @@ export class GraphTallyCollector { ravRedemptionThreshold: formatGRT(this.ravRedemptionThreshold), belowThresholdCount: signedRavs.belowThreshold.length, totalValueGRT, + // What is still collectible on those RAVs, once already collected tokens are + // subtracted. This, not totalValueGRT, is the revenue being left on the table. + remainingValueGRT: formatGRT(signedRavs.belowThresholdRemaining), allocations: signedRavs.belowThreshold.map((signedRav) => collectionIdToAllocationId(signedRav.rav.rav.collectionId), ), @@ -399,12 +413,18 @@ export class GraphTallyCollector { }) if (belowThreshold) { results.belowThreshold.push(rav) + results.belowThresholdRemaining += + BigInt(rav.rav.rav.valueAggregate) - tokensCollected } else { results.eligible.push(rav) } return results }, - { belowThreshold: [], eligible: [] }, + { + belowThreshold: [], + eligible: [], + belowThresholdRemaining: 0n, + }, ) }, { @@ -415,10 +435,12 @@ export class GraphTallyCollector { } // redeem only if last is true - // Later can add order and limit + // Highest value first, so that RAVs worth collecting are never crowded out of the + // 100 row batch by dust that sits below the redemption threshold indefinitely. private async pendingRAVs(): Promise { return await this.models.receiptAggregateVouchersV2.findAll({ where: { last: true, final: false }, + order: [['valueAggregate', 'DESC']], limit: 100, }) } @@ -1107,6 +1129,18 @@ const registerReceiptMetrics = (metrics: Metrics, networkIdentifier: string) => registers: [metrics.registry], labelNames: ['collection'], }), + + ravsBelowThreshold: new metrics.client.Gauge({ + name: `indexer_agent_rav_v2_ravs_below_threshold_${networkIdentifier}`, + help: 'Number of pending rav v2s deferred because their collectible value is below the redemption threshold', + registers: [metrics.registry], + }), + + ravsBelowThresholdValueGRT: new metrics.client.Gauge({ + name: `indexer_agent_rav_v2_ravs_below_threshold_value_grt_${networkIdentifier}`, + help: 'Total GRT still collectible on pending rav v2s deferred below the redemption threshold', + registers: [metrics.registry], + }), }) function collectionIdToAllocationId(collectionId: string): string { diff --git a/packages/indexer-common/src/allocations/tap-collector.ts b/packages/indexer-common/src/allocations/tap-collector.ts index 8bf80be04..521549aac 100644 --- a/packages/indexer-common/src/allocations/tap-collector.ts +++ b/packages/indexer-common/src/allocations/tap-collector.ts @@ -41,6 +41,8 @@ interface RavMetrics { ravRedeemsFailed: Counter ravsRedeemDuration: Histogram ravCollectedFees: Gauge + ravsBelowThreshold: Gauge + ravsBelowThresholdValueGRT: Gauge } interface TapCollectorOptions { @@ -162,14 +164,17 @@ export class TapCollector { startRAVProcessing() { const notifyAndMapEligible = (signedRavs: ValidRavs) => { + const totalValue = signedRavs.belowThreshold.reduce( + (total, signedRav) => total + BigInt(signedRav.rav.rav.valueAggregate), + 0n, + ) + // Set every pass, including to 0, so the gauges decay once deferrals clear + this.metrics.ravsBelowThreshold.set(signedRavs.belowThreshold.length) + this.metrics.ravsBelowThresholdValueGRT.set(parseFloat(formatGRT(totalValue))) + if (signedRavs.belowThreshold.length > 0) { const logger = this.logger.child({ function: 'startRAVProcessing()' }) - const totalValueGRT = formatGRT( - signedRavs.belowThreshold.reduce( - (total, signedRav) => total + BigInt(signedRav.rav.rav.valueAggregate), - 0n, - ), - ) + const totalValueGRT = formatGRT(totalValue) logger.info(`[TAPv1] Query RAVs below the redemption threshold`, { hint: 'If you would like to redeem RAVs like this, reduce the voucher redemption threshold', ravRedemptionThreshold: formatGRT(this.ravRedemptionThreshold), @@ -377,10 +382,12 @@ export class TapCollector { } // redeem only if last is true - // Later can add order and limit + // Highest value first, so that RAVs worth collecting are never crowded out of the + // 100 row batch by dust that sits below the redemption threshold indefinitely. private async pendingRAVs(): Promise { return await this.models.receiptAggregateVouchers.findAll({ where: { last: true, final: false }, + order: [['valueAggregate', 'DESC']], limit: 100, }) } @@ -881,4 +888,16 @@ const registerReceiptMetrics = (metrics: Metrics, networkIdentifier: string) => registers: [metrics.registry], labelNames: ['allocation'], }), + + ravsBelowThreshold: new metrics.client.Gauge({ + name: `indexer_agent_ravs_below_threshold_${networkIdentifier}`, + help: 'Number of pending ravs deferred because their value is below the redemption threshold', + registers: [metrics.registry], + }), + + ravsBelowThresholdValueGRT: new metrics.client.Gauge({ + name: `indexer_agent_ravs_below_threshold_value_grt_${networkIdentifier}`, + help: 'Total GRT value of pending ravs deferred below the redemption threshold', + registers: [metrics.registry], + }), }) From 5d85cc2d497605a3a4c96d17d8e1d7cd333aa520 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 10 Jul 2026 22:44:47 +1000 Subject: [PATCH 2/3] style(tests): format pending RAVs ordering test file CI checks that prettier and eslint produce no diff, and this file was committed with 4 statements formatted wider than the repo's prettier settings allow. --- .../__tests__/pending-ravs-order.test.ts | 31 +++++++++++-------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts index c81b71743..9160a93cd 100644 --- a/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts +++ b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts @@ -4,7 +4,12 @@ import { defineQueryFeeModels, QueryFeeModels } from '../../query-fees/models' import { TapCollector } from '../tap-collector' import { GraphTallyCollector } from '../graph-tally-collector' -import { connectDatabase, createLogger, Logger, toAddress } from '@graphprotocol/common-ts' +import { + connectDatabase, + createLogger, + Logger, + toAddress, +} from '@graphprotocol/common-ts' import { Sequelize } from 'sequelize' // Make global Jest variables available @@ -74,10 +79,10 @@ describe('pendingRAVs ordering', () => { redeemedAt: null, }) } - const collector: TapCollector = Object.assign( - Object.create(TapCollector.prototype), - { logger, models: queryFeeModels }, - ) + const collector: TapCollector = Object.assign(Object.create(TapCollector.prototype), { + logger, + models: queryFeeModels, + }) // Act const ravs = await collector['pendingRAVs']() @@ -140,10 +145,10 @@ describe('pendingRAVs ordering', () => { final: false, redeemedAt: null, }) - const collector: TapCollector = Object.assign( - Object.create(TapCollector.prototype), - { logger, models: queryFeeModels }, - ) + const collector: TapCollector = Object.assign(Object.create(TapCollector.prototype), { + logger, + models: queryFeeModels, + }) // Act const ravs = await collector['pendingRAVs']() @@ -228,10 +233,10 @@ describe('pendingRAVs ordering', () => { final: false, redeemedAt: null, }) - const collector: TapCollector = Object.assign( - Object.create(TapCollector.prototype), - { logger, models: queryFeeModels }, - ) + const collector: TapCollector = Object.assign(Object.create(TapCollector.prototype), { + logger, + models: queryFeeModels, + }) // Act const ravs = await collector['pendingRAVs']() From c971b28e72cf3a44453da4d249c7ba6a7ab36f1b Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Sat, 11 Jul 2026 00:17:31 +1000 Subject: [PATCH 3/3] fix(tap): raise pending RAV batch to 1,000 and chunk subgraph filters Each pass reconciles a capped batch of pending payment vouchers, and rows that can never settle could permanently fill a 100 row batch. Raising the cap to 1,000 requires chunking the allocation id filter sent to the subgraph, since larger request bodies have caused http 413. --- .../__tests__/pending-ravs-order.test.ts | 64 +++++-- .../src/allocations/graph-tally-collector.ts | 143 ++++++++------- .../src/allocations/tap-collector.ts | 164 ++++++++++-------- 3 files changed, 232 insertions(+), 139 deletions(-) diff --git a/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts index 9160a93cd..39e292402 100644 --- a/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts +++ b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts @@ -120,11 +120,11 @@ describe('pendingRAVs ordering', () => { expect(ravs.map((rav) => rav.valueAggregate)).toEqual([900n, 42n, 7n, 3n]) }) - test('TAPv1 ordering surfaces high value RAVs from beyond the 100 row limit', async () => { - // Arrange: 100 dust RAVs that would fill the batch on their own, plus one + test('TAPv1 ordering surfaces high value RAVs from beyond a full batch', async () => { + // Arrange: 1,000 dust RAVs that would fill the batch on their own, plus one // valuable RAV inserted last so insertion order alone would exclude it. - for (let i = 0; i < 100; i++) { - await queryFeeModels.receiptAggregateVouchers.create({ + await queryFeeModels.receiptAggregateVouchers.bulkCreate( + Array.from({ length: 1000 }, (_, i) => ({ allocationId: toAddress(i.toString(16).padStart(40, '0')), senderAddress: PAYER, signature: SIGNATURE, @@ -133,8 +133,8 @@ describe('pendingRAVs ordering', () => { last: true, final: false, redeemedAt: null, - }) - } + })), + ) await queryFeeModels.receiptAggregateVouchers.create({ allocationId: toAddress('ffffffffffffffffffffffffffffffffffffffff'), senderAddress: PAYER, @@ -154,14 +154,14 @@ describe('pendingRAVs ordering', () => { const ravs = await collector['pendingRAVs']() // Assert - expect(ravs).toHaveLength(100) + expect(ravs).toHaveLength(1000) expect(ravs[0].valueAggregate).toEqual(10n ** 21n) }) - test('TAPv2 ordering surfaces high value RAVs from beyond the 100 row limit', async () => { + test('TAPv2 ordering surfaces high value RAVs from beyond a full batch', async () => { // Arrange - for (let i = 0; i < 100; i++) { - await queryFeeModels.receiptAggregateVouchersV2.create({ + await queryFeeModels.receiptAggregateVouchersV2.bulkCreate( + Array.from({ length: 1000 }, (_, i) => ({ collectionId: `0x${i.toString(16).padStart(64, '0')}`, payer: PAYER, dataService: DATA_SERVICE, @@ -173,8 +173,8 @@ describe('pendingRAVs ordering', () => { last: true, final: false, redeemedAt: null, - }) - } + })), + ) await queryFeeModels.receiptAggregateVouchersV2.create({ collectionId: `0x${'f'.repeat(64)}`, payer: PAYER, @@ -197,10 +197,48 @@ describe('pendingRAVs ordering', () => { const ravs = await collector['pendingRAVs']() // Assert - expect(ravs).toHaveLength(100) + expect(ravs).toHaveLength(1000) expect(ravs[0].valueAggregate).toEqual(10n ** 21n) }) + test('findTransactionsForRavs chunks the allocation id filter and pins one block', async () => { + // Arrange: 250 pending RAVs with distinct allocations, and a fake subgraph client + // that records each request it receives. + const ravs = Array.from({ length: 250 }, (_, i) => ({ + collectionId: `0x${i.toString(16).padStart(64, '0')}`, + payer: PAYER, + redeemedAt: null, + })) as unknown as Parameters[0] + const query = jest.fn().mockResolvedValue({ + data: { + paymentsEscrowTransactions: [], + _meta: { block: { hash: 'pinned-block', timestamp: 1 } }, + }, + }) + const collector: GraphTallyCollector = Object.assign( + Object.create(GraphTallyCollector.prototype), + { logger, networkSubgraph: { query } }, + ) + + // Act + const response = await collector.findTransactionsForRavs(ravs) + + // Assert: 250 ids split into chunks of at most 100, so 3 requests + expect(query).toHaveBeenCalledTimes(3) + const variables = query.mock.calls.map((call) => call[1]) + for (const vars of variables) { + expect(vars.unfinalizedRavsAllocationIds.length).toBeLessThanOrEqual(100) + } + expect(variables.flatMap((vars) => vars.unfinalizedRavsAllocationIds)).toHaveLength( + 250, + ) + // The first request floats to the chain head; all later ones are pinned to it + expect(variables[0].block).toBeUndefined() + expect(variables[1].block).toEqual({ hash: 'pinned-block' }) + expect(variables[2].block).toEqual({ hash: 'pinned-block' }) + expect(response._meta.block.hash).toEqual('pinned-block') + }) + test('pendingRAVs excludes final and non-last RAVs regardless of value', async () => { // Arrange await queryFeeModels.receiptAggregateVouchers.create({ diff --git a/packages/indexer-common/src/allocations/graph-tally-collector.ts b/packages/indexer-common/src/allocations/graph-tally-collector.ts index ca0cbbaae..4861c8f92 100644 --- a/packages/indexer-common/src/allocations/graph-tally-collector.ts +++ b/packages/indexer-common/src/allocations/graph-tally-collector.ts @@ -37,6 +37,15 @@ const RAV_CHECK_INTERVAL_MS = 900_000 // 1000 here was leading to http 413 request entity too large const PAGE_SIZE = 200 +// How many pending RAVs are reconciled per pass. Rows beyond this stay pending and are +// picked up on later passes once higher value rows settle and leave the set. +const PENDING_RAV_BATCH_SIZE = 1_000 + +// How many allocation ids go into a single subgraph request. The id filter travels in +// every request body, so it is chunked to keep requests the size they had when the +// batch itself was capped at 100 rows (see the http 413 note above). +const FILTER_CHUNK_SIZE = 100 + interface RavMetrics { ravRedeemsSuccess: Counter ravRedeemsInvalid: Counter @@ -436,13 +445,20 @@ export class GraphTallyCollector { // redeem only if last is true // Highest value first, so that RAVs worth collecting are never crowded out of the - // 100 row batch by dust that sits below the redemption threshold indefinitely. + // batch by dust that sits below the redemption threshold indefinitely. private async pendingRAVs(): Promise { - return await this.models.receiptAggregateVouchersV2.findAll({ + const ravs = await this.models.receiptAggregateVouchersV2.findAll({ where: { last: true, final: false }, order: [['valueAggregate', 'DESC']], - limit: 100, + limit: PENDING_RAV_BATCH_SIZE, }) + if (ravs.length === PENDING_RAV_BATCH_SIZE) { + this.logger.warn( + '[TAPv2] Pending RAV batch is full, RAVs below the value cutoff are not reconciled this pass', + { batchSize: PENDING_RAV_BATCH_SIZE }, + ) + } + return ravs } private async filterAndUpdateRavs( @@ -543,7 +559,6 @@ export class GraphTallyCollector { ravs: ReceiptAggregateVoucherV2[], ): Promise { let meta: GraphTallyMeta | undefined = undefined - let lastId = '' const paymentsEscrowTransactions: GraphTallyTransaction[] = [] const unfinalizedRavsAllocationIds = [ @@ -558,69 +573,81 @@ export class GraphTallyCollector { ...new Set(ravs.map((value) => toAddress(value.payer).toLowerCase())), ] - for (;;) { - let block: { hash: string } | undefined = undefined - if (meta?.block?.hash) { - block = { - hash: meta?.block?.hash, + // chunk() yields nothing for an empty input, but a query must still run in that + // case so the caller gets subgraph block metadata back, hence the explicit [[]]. + const allocationIdChunks = + unfinalizedRavsAllocationIds.length === 0 + ? [[] as string[]] + : chunk(unfinalizedRavsAllocationIds, FILTER_CHUNK_SIZE) + + for (const allocationIdsChunk of allocationIdChunks) { + let lastId = '' + for (;;) { + // After the first response, every request (across pages and chunks) is pinned + // to that block, so the whole pass sees one consistent snapshot of the chain. + let block: { hash: string } | undefined = undefined + if (meta?.block?.hash) { + block = { + hash: meta?.block?.hash, + } } - } - const result: QueryResult = - await this.networkSubgraph.query( - gql` - query paymentsEscrowTransactions( - $lastId: String! - $pageSize: Int! - $block: Block_height - $unfinalizedRavsAllocationIds: [String!]! - $payerAddresses: [String!]! - ) { - paymentsEscrowTransactions( - first: $pageSize - block: $block - orderBy: id - orderDirection: asc - where: { - id_gt: $lastId - type: "redeem" - allocationId_in: $unfinalizedRavsAllocationIds - payer_: { id_in: $payerAddresses } - } + const result: QueryResult = + await this.networkSubgraph.query( + gql` + query paymentsEscrowTransactions( + $lastId: String! + $pageSize: Int! + $block: Block_height + $unfinalizedRavsAllocationIds: [String!]! + $payerAddresses: [String!]! ) { - id - allocationId - timestamp - payer { + paymentsEscrowTransactions( + first: $pageSize + block: $block + orderBy: id + orderDirection: asc + where: { + id_gt: $lastId + type: "redeem" + allocationId_in: $unfinalizedRavsAllocationIds + payer_: { id_in: $payerAddresses } + } + ) { id - } - } - _meta { - block { - hash + allocationId timestamp + payer { + id + } + } + _meta { + block { + hash + timestamp + } } } - } - `, - { - lastId, - pageSize: PAGE_SIZE, - block, - unfinalizedRavsAllocationIds, - payerAddresses, - }, - ) + `, + { + lastId, + pageSize: PAGE_SIZE, + block, + unfinalizedRavsAllocationIds: allocationIdsChunk, + payerAddresses, + }, + ) - if (!result.data) { - throw `[TAPv2] There was an error while querying Network Subgraph. Errors: ${result.error}` - } - meta = result.data._meta - paymentsEscrowTransactions.push(...result.data.paymentsEscrowTransactions) - if (result.data.paymentsEscrowTransactions.length < PAGE_SIZE) { - break + if (!result.data) { + throw `[TAPv2] There was an error while querying Network Subgraph. Errors: ${result.error}` + } + meta = result.data._meta + paymentsEscrowTransactions.push(...result.data.paymentsEscrowTransactions) + if (result.data.paymentsEscrowTransactions.length < PAGE_SIZE) { + break + } + lastId = result.data.paymentsEscrowTransactions.slice(-1)[0].id } - lastId = result.data.paymentsEscrowTransactions.slice(-1)[0].id } return { diff --git a/packages/indexer-common/src/allocations/tap-collector.ts b/packages/indexer-common/src/allocations/tap-collector.ts index 521549aac..d37211062 100644 --- a/packages/indexer-common/src/allocations/tap-collector.ts +++ b/packages/indexer-common/src/allocations/tap-collector.ts @@ -22,6 +22,7 @@ import { tapAllocationIdProof, parseGraphQLAllocation, sequentialTimerMap, + chunk, } from '..' import pReduce from 'p-reduce' import { SubgraphClient, QueryResult } from '../subgraph-client' @@ -35,6 +36,15 @@ const RAV_CHECK_INTERVAL_MS = 900_000 // 1000 here was leading to http 413 request entity too large const PAGE_SIZE = 200 +// How many pending RAVs are reconciled per pass. Rows beyond this stay pending and are +// picked up on later passes once higher value rows settle and leave the set. +const PENDING_RAV_BATCH_SIZE = 1_000 + +// How many allocation ids go into a single subgraph request. The id filter travels in +// every request body, so it is chunked to keep requests the size they had when the +// batch itself was capped at 100 rows (see the http 413 note above). +const FILTER_CHUNK_SIZE = 100 + interface RavMetrics { ravRedeemsSuccess: Counter ravRedeemsInvalid: Counter @@ -383,13 +393,20 @@ export class TapCollector { // redeem only if last is true // Highest value first, so that RAVs worth collecting are never crowded out of the - // 100 row batch by dust that sits below the redemption threshold indefinitely. + // batch by dust that sits below the redemption threshold indefinitely. private async pendingRAVs(): Promise { - return await this.models.receiptAggregateVouchers.findAll({ + const ravs = await this.models.receiptAggregateVouchers.findAll({ where: { last: true, final: false }, order: [['valueAggregate', 'DESC']], - limit: 100, + limit: PENDING_RAV_BATCH_SIZE, }) + if (ravs.length === PENDING_RAV_BATCH_SIZE) { + this.logger.warn( + '[TAPv1] Pending RAV batch is full, RAVs below the value cutoff are not reconciled this pass', + { batchSize: PENDING_RAV_BATCH_SIZE }, + ) + } + return ravs } private async filterAndUpdateRavs( @@ -487,7 +504,6 @@ export class TapCollector { ravs: ReceiptAggregateVoucher[], ): Promise { let meta: TapMeta | undefined = undefined - let lastId = '' const transactions: TapTransaction[] = [] const unfinalizedRavsAllocationIds = [ @@ -498,79 +514,91 @@ export class TapCollector { ...new Set(ravs.map((value) => toAddress(value.senderAddress).toLowerCase())), ] - for (;;) { - let block: { hash: string } | undefined = undefined - if (meta?.block?.hash) { - block = { - hash: meta?.block?.hash, + // chunk() yields nothing for an empty input, but a query must still run in that + // case so the caller gets subgraph block metadata back, hence the explicit [[]]. + const allocationIdChunks = + unfinalizedRavsAllocationIds.length === 0 + ? [[] as string[]] + : chunk(unfinalizedRavsAllocationIds, FILTER_CHUNK_SIZE) + + for (const allocationIdsChunk of allocationIdChunks) { + let lastId = '' + for (;;) { + // After the first response, every request (across pages and chunks) is pinned + // to that block, so the whole pass sees one consistent snapshot of the chain. + let block: { hash: string } | undefined = undefined + if (meta?.block?.hash) { + block = { + hash: meta?.block?.hash, + } } - } - this.logger.trace('[TAPv1] Querying Tap Subgraph for RAVs', { - lastId, - pageSize: PAGE_SIZE, - block, - unfinalizedRavsAllocationIds, - senderAddresses, - }) - const result: QueryResult = - await this.tapSubgraph.query( - gql` - query transactions( - $lastId: String! - $pageSize: Int! - $block: Block_height - $unfinalizedRavsAllocationIds: [String!]! - $senderAddresses: [String!]! - ) { - transactions( - first: $pageSize - block: $block - orderBy: id - orderDirection: asc - where: { - id_gt: $lastId - type: "redeem" - allocationID_in: $unfinalizedRavsAllocationIds - sender_: { id_in: $senderAddresses } - } + this.logger.trace('[TAPv1] Querying Tap Subgraph for RAVs', { + lastId, + pageSize: PAGE_SIZE, + block, + allocationIdsChunk, + senderAddresses, + }) + const result: QueryResult = + await this.tapSubgraph.query( + gql` + query transactions( + $lastId: String! + $pageSize: Int! + $block: Block_height + $unfinalizedRavsAllocationIds: [String!]! + $senderAddresses: [String!]! ) { - id - allocationID - timestamp - sender { + transactions( + first: $pageSize + block: $block + orderBy: id + orderDirection: asc + where: { + id_gt: $lastId + type: "redeem" + allocationID_in: $unfinalizedRavsAllocationIds + sender_: { id_in: $senderAddresses } + } + ) { id - } - } - _meta { - block { - hash + allocationID timestamp + sender { + id + } + } + _meta { + block { + hash + timestamp + } } } - } - `, - { - lastId, - pageSize: PAGE_SIZE, - block, - unfinalizedRavsAllocationIds, - senderAddresses, - }, - ) + `, + { + lastId, + pageSize: PAGE_SIZE, + block, + unfinalizedRavsAllocationIds: allocationIdsChunk, + senderAddresses, + }, + ) - if (!result.data) { - this.logger.error('[TAPv1] There was an error while querying Tap Subgraph', { - result, - }) - throw `[TAPv1] There was an error while querying Tap Subgraph. Errors: ${result.error}` - } - meta = result.data._meta - transactions.push(...result.data.transactions) - if (result.data.transactions.length < PAGE_SIZE) { - break + if (!result.data) { + this.logger.error('[TAPv1] There was an error while querying Tap Subgraph', { + result, + }) + throw `[TAPv1] There was an error while querying Tap Subgraph. Errors: ${result.error}` + } + meta = result.data._meta + transactions.push(...result.data.transactions) + if (result.data.transactions.length < PAGE_SIZE) { + break + } + lastId = result.data.transactions.slice(-1)[0].id } - lastId = result.data.transactions.slice(-1)[0].id } return {