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..39e292402 --- /dev/null +++ b/packages/indexer-common/src/allocations/__tests__/pending-ravs-order.test.ts @@ -0,0 +1,286 @@ +// 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 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. + await queryFeeModels.receiptAggregateVouchers.bulkCreate( + Array.from({ length: 1000 }, (_, i) => ({ + 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(1000) + expect(ravs[0].valueAggregate).toEqual(10n ** 21n) + }) + + test('TAPv2 ordering surfaces high value RAVs from beyond a full batch', async () => { + // Arrange + await queryFeeModels.receiptAggregateVouchersV2.bulkCreate( + Array.from({ length: 1000 }, (_, i) => ({ + 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(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({ + 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..4861c8f92 100644 --- a/packages/indexer-common/src/allocations/graph-tally-collector.ts +++ b/packages/indexer-common/src/allocations/graph-tally-collector.ts @@ -37,12 +37,23 @@ 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 ravRedeemsFailed: Counter ravsRedeemDuration: Histogram ravCollectedFees: Gauge + ravsBelowThreshold: Gauge + ravsBelowThresholdValueGRT: Gauge } interface TapCollectorOptions { @@ -59,6 +70,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 +176,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 +194,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 +422,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,12 +444,21 @@ 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 + // 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 }, - limit: 100, + order: [['valueAggregate', 'DESC']], + 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( @@ -521,7 +559,6 @@ export class GraphTallyCollector { ravs: ReceiptAggregateVoucherV2[], ): Promise { let meta: GraphTallyMeta | undefined = undefined - let lastId = '' const paymentsEscrowTransactions: GraphTallyTransaction[] = [] const unfinalizedRavsAllocationIds = [ @@ -536,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 { @@ -1107,6 +1156,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..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,12 +36,23 @@ 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 ravRedeemsFailed: Counter ravsRedeemDuration: Histogram ravCollectedFees: Gauge + ravsBelowThreshold: Gauge + ravsBelowThresholdValueGRT: Gauge } interface TapCollectorOptions { @@ -162,14 +174,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,12 +392,21 @@ 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 + // 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 }, - limit: 100, + order: [['valueAggregate', 'DESC']], + 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( @@ -480,7 +504,6 @@ export class TapCollector { ravs: ReceiptAggregateVoucher[], ): Promise { let meta: TapMeta | undefined = undefined - let lastId = '' const transactions: TapTransaction[] = [] const unfinalizedRavsAllocationIds = [ @@ -491,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 { @@ -881,4 +916,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], + }), })