From 93dbcb092b3f2ec3d930128aab3f33ddf6cf936b Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 1 Jun 2026 17:42:44 +1200 Subject: [PATCH 1/4] feat(dips): track accepted agreements in the local proposal store The agent already stores a row per indexing-agreement proposal and flips it to 'accepted' once accepted on-chain, but nothing read those accepted rows back. Add a lookup for them and a way to retire them, and expose each row's last-updated time. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../__tests__/pending-rca-consumer.test.ts | 48 +++++++++++++++++++ .../src/indexing-fees/pending-rca-consumer.ts | 23 ++++++++- .../indexer-common/src/indexing-fees/types.ts | 4 ++ 3 files changed, 73 insertions(+), 2 deletions(-) diff --git a/packages/indexer-common/src/indexing-fees/__tests__/pending-rca-consumer.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/pending-rca-consumer.test.ts index d0c59e9f3..88e0d64ea 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/pending-rca-consumer.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/pending-rca-consumer.test.ts @@ -258,6 +258,40 @@ describe('PendingRcaConsumer', () => { }) }) + describe('getAcceptedProposals', () => { + test('queries only accepted rows', async () => { + const model = createMockModel([]) + const consumer = new PendingRcaConsumer(logger, model) + + await consumer.getAcceptedProposals() + + expect(model.findAll).toHaveBeenCalledWith({ + where: { status: 'accepted' }, + }) + }) + + test('exposes the row updatedAt as the acceptance time', async () => { + const acceptedAt = new Date('2024-02-02T03:04:05Z') + const model = createMockModel([ + { + id: 'accepted-uuid', + signed_payload: encodeTestPayload(), + version: 2, + status: 'accepted', + created_at: new Date('2024-01-01'), + updated_at: acceptedAt, + }, + ]) + const consumer = new PendingRcaConsumer(logger, model) + + const proposals = await consumer.getAcceptedProposals() + + expect(proposals).toHaveLength(1) + expect(proposals[0].status).toBe('accepted') + expect(proposals[0].updatedAt).toEqual(acceptedAt) + }) + }) + describe('markAccepted', () => { test('updates status to accepted', async () => { const model = createMockModel() @@ -272,6 +306,20 @@ describe('PendingRcaConsumer', () => { }) }) + describe('markCompleted', () => { + test('updates status to completed', async () => { + const model = createMockModel() + const consumer = new PendingRcaConsumer(logger, model) + + await consumer.markCompleted('test-uuid') + + expect(model.update).toHaveBeenCalledWith( + { status: 'completed' }, + { where: { id: 'test-uuid' } }, + ) + }) + }) + describe('markRejected', () => { test('updates status to rejected', async () => { const model = createMockModel() diff --git a/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts b/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts index daf9e28ec..bc51d9cf5 100644 --- a/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts +++ b/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts @@ -15,8 +15,19 @@ export class PendingRcaConsumer { ) {} async getPendingProposals(): Promise { + return this.getProposalsByStatus('pending') + } + + // Proposals accepted on-chain but not yet retired. The rule reaper keeps these + // deployments' rules alive across the window where the agreement is accepted + // on-chain but the indexing-payments subgraph hasn't indexed it yet. + async getAcceptedProposals(): Promise { + return this.getProposalsByStatus('accepted') + } + + private async getProposalsByStatus(status: string): Promise { const rows = await this.model.findAll({ - where: { status: 'pending' }, + where: { status }, }) const decoded: DecodedRcaProposal[] = [] @@ -29,7 +40,7 @@ export class PendingRcaConsumer { } decoded.push(proposal) } catch (error) { - this.logger.warn(`Failed to decode pending RCA proposal ${row.id}, skipping`, { + this.logger.warn(`Failed to decode ${status} RCA proposal ${row.id}, skipping`, { error, }) } @@ -48,6 +59,13 @@ export class PendingRcaConsumer { await this.model.update({ status: 'accepted' }, { where: { id } }) } + // Retires an accepted row once the indexing-payments subgraph has indexed the + // agreement and become its source of truth. After this the row no longer keeps + // the deployment's rule alive; the subgraph does. + async markCompleted(id: string): Promise { + await this.model.update({ status: 'completed' }, { where: { id } }) + } + async markRejected(id: string, reason?: string): Promise { await this.model.update({ status: 'rejected' }, { where: { id } }) if (reason) { @@ -99,6 +117,7 @@ export class PendingRcaConsumer { id: row.id, status: row.status, createdAt: row.created_at, + updatedAt: row.updated_at, agreementId, diff --git a/packages/indexer-common/src/indexing-fees/types.ts b/packages/indexer-common/src/indexing-fees/types.ts index 8a0d1d60f..baffae00d 100644 --- a/packages/indexer-common/src/indexing-fees/types.ts +++ b/packages/indexer-common/src/indexing-fees/types.ts @@ -5,6 +5,10 @@ export interface DecodedRcaProposal { id: string status: string createdAt: Date + // Last time the row changed status. For an 'accepted' row this is when the + // agreement was accepted on-chain; the rule reaper uses it to decide when the + // indexing-payments subgraph has had time to index the acceptance. + updatedAt: Date // Locally derived bytes16 on-chain agreement id (0x-prefixed lowercase). // Derived from (payer, dataService, serviceProvider, deadline, nonce). From 7b4ab86f69c7175a153f1b84bb5384f9a3f44d81 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 1 Jun 2026 17:42:50 +1200 Subject: [PATCH 2/4] fix(dips): protect freshly accepted rules with durable local state A freshly accepted agreement's rule was shielded from cleanup by an in-memory, 300-second grace window that didn't survive a restart. Keep these rules alive from the durable 'accepted' proposal rows instead, retiring each row once the subgraph has indexed past its acceptance. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../__tests__/accept-proposals.test.ts | 1 + .../src/indexing-fees/__tests__/dips.test.ts | 91 +++++++++----- .../indexer-common/src/indexing-fees/dips.ts | 119 +++++++++--------- 3 files changed, 124 insertions(+), 87 deletions(-) diff --git a/packages/indexer-common/src/indexing-fees/__tests__/accept-proposals.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/accept-proposals.test.ts index dfbcde343..bc8d5f87c 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/accept-proposals.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/accept-proposals.test.ts @@ -35,6 +35,7 @@ function createMockProposal( id: 'proposal-1', status: 'pending', createdAt: new Date(), + updatedAt: new Date(), agreementId: '0xabcd1234567890abcdef1234567890ab', payer: '0x1111111111111111111111111111111111111111', serviceProvider: '0x3333333333333333333333333333333333333333', diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index 3b5808a3d..f655675c1 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -16,10 +16,7 @@ import { MultiNetworks, } from '@graphprotocol/indexer-common' import type { SubgraphIndexingAgreement } from '../agreement-monitor' -import { - DIPS_RULE_GRACE_SECONDS, - DIPS_SUBGRAPH_STALENESS_THRESHOLD_SECONDS, -} from '../dips' +import { DIPS_SUBGRAPH_STALENESS_THRESHOLD_SECONDS } from '../dips' import { definePendingRcaProposalModel } from '../../indexer-management/models/pending-rca-proposal' import { connectDatabase, @@ -514,7 +511,7 @@ describe('DipsManager', () => { expect(rule).toBeNull() }) - test('keeps a freshly accepted DIPS rule within the grace window despite no backing agreement yet', async () => { + test('keeps a freshly accepted DIPS rule while the subgraph has not yet indexed the agreement', async () => { await managementModels.IndexingRule.create({ identifier: testDeploymentId, identifierType: SubgraphIdentifierType.DEPLOYMENT, @@ -525,17 +522,26 @@ describe('DipsManager', () => { jest .spyOn(dipsManager.pendingRcaConsumer!, 'getPendingProposals') .mockResolvedValue([]) - setCollectableAgreements([]) - // Simulate the accept loop having just accepted this deployment on-chain; - // the indexing-payments subgraph has not indexed the agreement yet, so the - // deployment is in neither the pending set nor the on-chain-accepted set. - const internal = dipsManager as unknown as { - recentlyAcceptedDeployments: Map - } - internal.recentlyAcceptedDeployments.set( - new SubgraphDeploymentID(testDeploymentId).bytes32.toLowerCase(), - Math.floor(Date.now() / 1000), - ) + // Accepted on-chain just now; its durable accepted row keeps the rule alive + // even though the subgraph has not indexed the agreement into the active set. + jest + .spyOn(dipsManager.pendingRcaConsumer!, 'getAcceptedProposals') + .mockResolvedValue([ + { + id: 'accepted-1', + status: 'accepted', + createdAt: new Date(), + updatedAt: new Date(), + subgraphDeploymentId: new SubgraphDeploymentID(testDeploymentId), + minSecondsPerCollection: 60, + maxSecondsPerCollection: 3600, + } as never, + ]) + const markCompleted = jest.spyOn(dipsManager.pendingRcaConsumer!, 'markCompleted') + // Subgraph is fresh but its head is behind the just-now acceptance. + setCollectableAgreements([], { + subgraphTimestamp: Math.floor(Date.now() / 1000) - 30, + }) await dipsManager.ensureAgreementRules() @@ -546,9 +552,11 @@ describe('DipsManager', () => { }, }) expect(rule).not.toBeNull() + // Head hasn't reached the acceptance time, so the row is not retired yet. + expect(markCompleted).not.toHaveBeenCalled() }) - test('reaps a DIPS rule and prunes the grace entry once the grace window expires', async () => { + test('retires the accepted row and reaps its rule once the subgraph catches up but the agreement is gone', async () => { await managementModels.IndexingRule.create({ identifier: testDeploymentId, identifierType: SubgraphIdentifierType.DEPLOYMENT, @@ -559,29 +567,52 @@ describe('DipsManager', () => { jest .spyOn(dipsManager.pendingRcaConsumer!, 'getPendingProposals') .mockResolvedValue([]) - setCollectableAgreements([]) - const internal = dipsManager as unknown as { - recentlyAcceptedDeployments: Map - } - const deploymentKey = new SubgraphDeploymentID( - testDeploymentId, - ).bytes32.toLowerCase() - // Accepted longer ago than the grace window: no longer shielded. - internal.recentlyAcceptedDeployments.set( - deploymentKey, - Math.floor(Date.now() / 1000) - (DIPS_RULE_GRACE_SECONDS + 1), + const markCompleted = jest + .spyOn(dipsManager.pendingRcaConsumer!, 'markCompleted') + .mockResolvedValue(undefined) + // Accepted a while ago; the subgraph head (now) has indexed past it, yet + // the agreement is not in the active set — it is gone (or never active). + const acceptedProposals = jest.spyOn( + dipsManager.pendingRcaConsumer!, + 'getAcceptedProposals', ) + acceptedProposals.mockResolvedValue([ + { + id: 'accepted-1', + status: 'accepted', + createdAt: new Date(), + updatedAt: new Date(Date.now() - 100_000), + subgraphDeploymentId: new SubgraphDeploymentID(testDeploymentId), + minSecondsPerCollection: 60, + maxSecondsPerCollection: 3600, + } as never, + ]) + setCollectableAgreements([], { subgraphTimestamp: Math.floor(Date.now() / 1000) }) + // First pass: the accepted row still keeps the rule this tick, but the + // subgraph has caught up to the acceptance, so the row is retired. await dipsManager.ensureAgreementRules() + expect(markCompleted).toHaveBeenCalledWith('accepted-1') + let rule = await managementModels.IndexingRule.findOne({ + where: { + identifier: testDeploymentId, + decisionBasis: IndexingDecisionBasis.DIPS, + }, + }) + expect(rule).not.toBeNull() - const rule = await managementModels.IndexingRule.findOne({ + // Second pass: with the row retired, nothing backs the deployment, so the + // rule is reaped. + acceptedProposals.mockResolvedValue([]) + setCollectableAgreements([], { subgraphTimestamp: Math.floor(Date.now() / 1000) }) + await dipsManager.ensureAgreementRules() + rule = await managementModels.IndexingRule.findOne({ where: { identifier: testDeploymentId, decisionBasis: IndexingDecisionBasis.DIPS, }, }) expect(rule).toBeNull() - expect(internal.recentlyAcceptedDeployments.has(deploymentKey)).toBe(false) }) test('skips rule cleanup when the indexing-payments subgraph is stale', async () => { diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index c554ea66a..6da5449fa 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -42,15 +42,6 @@ const RECENT_BLOCK_OFFSET = 10 // a stuck call from head-of-lining the rest. const DIPS_ACCEPT_CONCURRENCY = 4 -// Window during which a just-accepted deployment's DIPS rule is shielded from -// the ensureAgreementRules reaper. The accept loop creates the rule and clears -// the pending row the moment the accept tx confirms, but the indexing-payments -// subgraph needs a few blocks to index the on-chain agreement. Without this -// shield the reconcile tick sees the deployment in neither the pending set nor -// the on-chain-accepted set and reaps the freshly-paid rule, churning the -// deployment and its allocation until the subgraph catches up. -export const DIPS_RULE_GRACE_SECONDS = 300 - // Reaping rules trusts the indexing-payments subgraph's list of backed // agreements. If that subgraph is lagging, the list is incomplete and reaping // would delete rules for agreements that are actually live. Skip the cleanup @@ -66,13 +57,6 @@ export class DipsManager { declare pendingRcaConsumer: PendingRcaConsumer declare collectionTracker: CollectionTracker declare offerVerifier: OfferVerifier | null - // bytes32 deployment id (lowercased) -> epoch seconds when the accept loop last - // accepted an agreement for it. ensureAgreementRules consults this to shield a - // rule from the reaper during the subgraph-indexing lag. In-memory by design: - // the race is intra-process (accept loop vs reconcile tick on the same - // instance), and a restart gives the subgraph ample time to index before the - // next reap, so the entry need not survive it. - private recentlyAcceptedDeployments = new Map() constructor( private logger: Logger, private models: IndexerManagementModels, @@ -100,12 +84,17 @@ export class DipsManager { return } - const { fromPendingProposals, fromActiveAgreements, deployments } = - await this.getDipsTargetDeployments() + const { + fromPendingProposals, + fromAcceptedProposals, + fromActiveAgreements, + deployments, + } = await this.getDipsTargetDeployments() this.logger.debug( `Ensuring DIPS indexing rules: ${fromPendingProposals.length} pending, ` + - `${fromActiveAgreements.length} active accepted, ${deployments.length} unique deployments`, + `${fromAcceptedProposals.length} locally accepted, ` + + `${fromActiveAgreements.length} active on subgraph, ${deployments.length} unique deployments`, ) const allDeploymentRules = await this.models.IndexingRule.findAll({ @@ -156,14 +145,37 @@ export class DipsManager { }) } + // Locally accepted agreements the subgraph hasn't picked up yet: ensure the + // rule survives the indexing gap. Already accepted on-chain, so a blocklist + // can't undo it here (cancelBlocklistedAgreements handles that separately). + for (const proposal of fromAcceptedProposals) { + const deploymentId = proposal.subgraphDeploymentId + const blocklisted = allDeploymentRules.find((r) => + this.isOnChainOptOutRule(r, deploymentId), + ) + if (blocklisted) { + this.logger.debug( + `Blocklisted accepted deployment ${deploymentId.toString()}; rule creation skipped`, + ) + continue + } + await this.upsertDipsRuleFor(deploymentId, { + allocationLifetime: Math.max( + Number(proposal.minSecondsPerCollection), + Number(proposal.maxSecondsPerCollection), + ), + }) + } + // Skip the reaper when the indexing-payments subgraph is lagging or its // freshness can't be read: its agreement list would be incomplete and the // reaper would delete rules for agreements that are actually live. Only // applies when a subgraph is configured; without one the reaper falls back - // to its pending-proposal basis as before. Fails safe — an unreadable + // to local pending and accepted rows as before. Fails safe — an unreadable // subgraph never drives deletion. + let subgraphHeadTimestamp: number | null = null if (this.network.indexingPaymentsSubgraph) { - const subgraphHeadTimestamp = await this.indexingPaymentsSubgraphHeadTimestamp() + subgraphHeadTimestamp = await this.indexingPaymentsSubgraphHeadTimestamp() const nowSeconds = Math.floor(Date.now() / 1000) const lagSeconds = subgraphHeadTimestamp === null ? null : nowSeconds - subgraphHeadTimestamp @@ -181,18 +193,11 @@ export class DipsManager { } } - // Drop DIPS rules whose deployment is no longer in the target set, unless - // the accept loop accepted it within the grace window: the on-chain - // agreement may not have been indexed by the subgraph yet, so it is absent - // from the target set for a benign reason. Prune grace entries that have - // aged out so the map stays bounded and an aged deployment becomes reapable. + // Drop DIPS rules whose deployment is in none of the target sets: pending + // proposal, local accepted row, or active agreement on the subgraph. A + // freshly accepted agreement stays in the set via its local accepted row + // until the subgraph indexes it, so its rule is never reaped during the gap. const targetSet = new Set(deployments.map((d) => d.bytes32)) - const graceCutoff = Math.floor(Date.now() / 1000) - DIPS_RULE_GRACE_SECONDS - for (const [deployment, acceptedAt] of this.recentlyAcceptedDeployments) { - if (acceptedAt <= graceCutoff) { - this.recentlyAcceptedDeployments.delete(deployment) - } - } const dipsRules = await this.models.IndexingRule.findAll({ where: { identifierType: SubgraphIdentifierType.DEPLOYMENT, @@ -202,15 +207,6 @@ export class DipsManager { for (const rule of dipsRules) { const ruleDeploymentId = new SubgraphDeploymentID(rule.identifier) if (targetSet.has(ruleDeploymentId.bytes32)) { - // Durably backed by the subgraph now; no longer needs the grace shield. - this.recentlyAcceptedDeployments.delete(ruleDeploymentId.bytes32.toLowerCase()) - continue - } - if (this.recentlyAcceptedDeployments.has(ruleDeploymentId.bytes32.toLowerCase())) { - this.logger.debug( - `Keeping recently-accepted DIPS rule for deployment ${ruleDeploymentId.toString()} ` + - 'while the indexing-payments subgraph catches up', - ) continue } this.logger.info( @@ -218,6 +214,20 @@ export class DipsManager { ) await this.models.IndexingRule.destroy({ where: { id: rule.id } }) } + + // Retire accepted rows the subgraph has caught up to. Once its head has + // indexed past when we accepted, the subgraph reflects the agreement's true + // state — active, or gone — and becomes the source of truth, so the local + // row no longer needs to keep the rule alive. Guarded on a configured, fresh + // subgraph (stale returns above), so the head timestamp is trustworthy. + if (subgraphHeadTimestamp !== null) { + for (const accepted of fromAcceptedProposals) { + const acceptedAtSeconds = Math.floor(accepted.updatedAt.getTime() / 1000) + if (subgraphHeadTimestamp >= acceptedAtSeconds) { + await this.pendingRcaConsumer.markCompleted(accepted.id) + } + } + } } private async upsertDipsRuleFor( @@ -256,10 +266,14 @@ export class DipsManager { private async getDipsTargetDeployments(): Promise<{ fromPendingProposals: DecodedRcaProposal[] + fromAcceptedProposals: DecodedRcaProposal[] fromActiveAgreements: SubgraphIndexingAgreement[] deployments: SubgraphDeploymentID[] }> { const fromPendingProposals = await this.pendingRcaConsumer.getPendingProposals() + // Locally accepted but not yet retired: keeps a deployment's rule alive after + // acceptance until the subgraph indexes the agreement and takes over. + const fromAcceptedProposals = await this.pendingRcaConsumer.getAcceptedProposals() let fromActiveAgreements: SubgraphIndexingAgreement[] = [] if (this.network.indexingPaymentsSubgraph) { @@ -282,7 +296,7 @@ export class DipsManager { const seen = new Set() const deployments: SubgraphDeploymentID[] = [] - for (const p of fromPendingProposals) { + for (const p of [...fromPendingProposals, ...fromAcceptedProposals]) { const key = p.subgraphDeploymentId.bytes32 if (!seen.has(key)) { seen.add(key) @@ -297,7 +311,12 @@ export class DipsManager { } } - return { fromPendingProposals, fromActiveAgreements, deployments } + return { + fromPendingProposals, + fromAcceptedProposals, + fromActiveAgreements, + deployments, + } } // Returns the indexing-payments subgraph's latest indexed block timestamp, or @@ -587,13 +606,6 @@ export class DipsManager { } await consumer.markAccepted(proposal.id) - // markAccepted clears the pending row, so the deployment drops out of the - // reconcile target set until the subgraph indexes the agreement. Shield - // its rule from the reaper meanwhile. - this.recentlyAcceptedDeployments.set( - proposal.subgraphDeploymentId.bytes32.toLowerCase(), - Math.floor(Date.now() / 1000), - ) this.logger.info('Proposal accepted on-chain', { proposalId: proposal.id, allocationId: allocation.id, @@ -720,13 +732,6 @@ export class DipsManager { } await consumer.markAccepted(proposal.id) - // markAccepted clears the pending row, so the deployment drops out of the - // reconcile target set until the subgraph indexes the agreement. Shield - // its rule from the reaper meanwhile. - this.recentlyAcceptedDeployments.set( - proposal.subgraphDeploymentId.bytes32.toLowerCase(), - Math.floor(Date.now() / 1000), - ) this.logger.info('Proposal accepted on-chain with new allocation', { proposalId: proposal.id, allocationId, From 122d1e163bb58f59e068f38ee93a3fdcbba92696 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 1 Jun 2026 21:43:45 +1200 Subject: [PATCH 3/4] fix(dips): retire accepted rows on subgraph presence, not a time race A freshly accepted agreement keeps its rule alive via a local row until the network data lists it. Retiring that row by comparing two separately-read timestamps could retire it too early and delete a live rule; retire once the data lists the agreement, and isolate a failed retire. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/indexing-fees/__tests__/dips.test.ts | 64 ++++++++++++++++++- .../indexer-common/src/indexing-fees/dips.ts | 40 ++++++++---- 2 files changed, 91 insertions(+), 13 deletions(-) diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index f655675c1..848a72477 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -529,6 +529,7 @@ describe('DipsManager', () => { .mockResolvedValue([ { id: 'accepted-1', + agreementId: testAgreementId, status: 'accepted', createdAt: new Date(), updatedAt: new Date(), @@ -552,10 +553,70 @@ describe('DipsManager', () => { }, }) expect(rule).not.toBeNull() - // Head hasn't reached the acceptance time, so the row is not retired yet. + // Not in the active set and the head is behind acceptance, so neither + // presence nor the time backstop retires it yet. expect(markCompleted).not.toHaveBeenCalled() }) + test('retires the accepted row by presence once its agreement appears in the active set, even before the head reaches acceptance', async () => { + await managementModels.IndexingRule.create({ + identifier: testDeploymentId, + identifierType: SubgraphIdentifierType.DEPLOYMENT, + decisionBasis: IndexingDecisionBasis.DIPS, + protocolNetwork: 'eip155:421614', + allocationLifetime: 3600, + }) + jest + .spyOn(dipsManager.pendingRcaConsumer!, 'getPendingProposals') + .mockResolvedValue([]) + const markCompleted = jest + .spyOn(dipsManager.pendingRcaConsumer!, 'markCompleted') + .mockResolvedValue(undefined) + // Accepted just now; the durable row's agreementId matches an agreement the + // subgraph now reports, proving the subgraph indexed the acceptance. + jest + .spyOn(dipsManager.pendingRcaConsumer!, 'getAcceptedProposals') + .mockResolvedValue([ + { + id: 'accepted-1', + agreementId: testAgreementId, + status: 'accepted', + createdAt: new Date(), + updatedAt: new Date(), + subgraphDeploymentId: new SubgraphDeploymentID(testDeploymentId), + minSecondsPerCollection: 60, + maxSecondsPerCollection: 3600, + } as never, + ]) + const farFuture = String(Math.floor(Date.now() / 1000) + 7 * 24 * 3600) + // Head is behind the just-now acceptance, so the time backstop cannot fire; + // only presence in the active set can retire the row here. + setCollectableAgreements( + [ + { + id: testAgreementId, + allocationId: testAllocationId, + subgraphDeploymentId: testDeploymentId, + state: 'Accepted', + lastCollectionAt: '0', + endsAt: farFuture, + maxInitialTokens: '0', + maxOngoingTokensPerSecond: '0', + tokensPerSecond: '0', + tokensPerEntityPerSecond: '0', + minSecondsPerCollection: 60, + maxSecondsPerCollection: 1800, + canceledAt: '0', + }, + ], + { subgraphTimestamp: Math.floor(Date.now() / 1000) - 30 }, + ) + + await dipsManager.ensureAgreementRules() + + expect(markCompleted).toHaveBeenCalledWith('accepted-1') + }) + test('retires the accepted row and reaps its rule once the subgraph catches up but the agreement is gone', async () => { await managementModels.IndexingRule.create({ identifier: testDeploymentId, @@ -579,6 +640,7 @@ describe('DipsManager', () => { acceptedProposals.mockResolvedValue([ { id: 'accepted-1', + agreementId: testAgreementId, status: 'accepted', createdAt: new Date(), updatedAt: new Date(Date.now() - 100_000), diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index 6da5449fa..48392802c 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -88,6 +88,7 @@ export class DipsManager { fromPendingProposals, fromAcceptedProposals, fromActiveAgreements, + subgraphAgreementIds, deployments, } = await this.getDipsTargetDeployments() @@ -193,10 +194,9 @@ export class DipsManager { } } - // Drop DIPS rules whose deployment is in none of the target sets: pending - // proposal, local accepted row, or active agreement on the subgraph. A - // freshly accepted agreement stays in the set via its local accepted row - // until the subgraph indexes it, so its rule is never reaped during the gap. + // Drop DIPS rules whose deployment is in none of the target sets (pending, + // local accepted, or active on the subgraph). A freshly accepted agreement + // stays via its accepted row until the subgraph indexes it, so it's not reaped. const targetSet = new Set(deployments.map((d) => d.bytes32)) const dipsRules = await this.models.IndexingRule.findAll({ where: { @@ -215,16 +215,24 @@ export class DipsManager { await this.models.IndexingRule.destroy({ where: { id: rule.id } }) } - // Retire accepted rows the subgraph has caught up to. Once its head has - // indexed past when we accepted, the subgraph reflects the agreement's true - // state — active, or gone — and becomes the source of truth, so the local - // row no longer needs to keep the rule alive. Guarded on a configured, fresh - // subgraph (stale returns above), so the head timestamp is trustworthy. + // Retire accepted rows once the subgraph indexes the agreement (presence, + // the primary signal) or the head-time backstop fires for ones it never + // listed. Per-row failures are logged and retried next tick, never aborting. if (subgraphHeadTimestamp !== null) { for (const accepted of fromAcceptedProposals) { - const acceptedAtSeconds = Math.floor(accepted.updatedAt.getTime() / 1000) - if (subgraphHeadTimestamp >= acceptedAtSeconds) { - await this.pendingRcaConsumer.markCompleted(accepted.id) + try { + const seenBySubgraph = subgraphAgreementIds.has( + accepted.agreementId.toLowerCase(), + ) + const acceptedAtSeconds = Math.floor(accepted.updatedAt.getTime() / 1000) + if (seenBySubgraph || subgraphHeadTimestamp >= acceptedAtSeconds) { + await this.pendingRcaConsumer.markCompleted(accepted.id) + } + } catch (err) { + this.logger.warn('Failed to retire accepted DIPS row; will retry next tick', { + acceptedId: accepted.id, + err, + }) } } } @@ -268,6 +276,7 @@ export class DipsManager { fromPendingProposals: DecodedRcaProposal[] fromAcceptedProposals: DecodedRcaProposal[] fromActiveAgreements: SubgraphIndexingAgreement[] + subgraphAgreementIds: Set deployments: SubgraphDeploymentID[] }> { const fromPendingProposals = await this.pendingRcaConsumer.getPendingProposals() @@ -276,12 +285,18 @@ export class DipsManager { const fromAcceptedProposals = await this.pendingRcaConsumer.getAcceptedProposals() let fromActiveAgreements: SubgraphIndexingAgreement[] = [] + // Agreement ids the subgraph reports as collectable (Accepted or CanceledByPayer). + // Lets us retire accepted rows by presence rather than a head-timestamp race. + const subgraphAgreementIds = new Set() if (this.network.indexingPaymentsSubgraph) { const indexerAddress = this.network.specification.indexerOptions.address const all = await fetchCollectableAgreements( this.network.indexingPaymentsSubgraph, indexerAddress, ) + for (const a of all) { + subgraphAgreementIds.add(a.id.toLowerCase()) + } const nowSeconds = Math.floor(Date.now() / 1000) fromActiveAgreements = all.filter( (a) => @@ -315,6 +330,7 @@ export class DipsManager { fromPendingProposals, fromAcceptedProposals, fromActiveAgreements, + subgraphAgreementIds, deployments, } } From ac722ca14edca930f75ac9618b536646f575e3c9 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 1 Jun 2026 21:46:01 +1200 Subject: [PATCH 4/4] fix(dips): warn when accepted rows can't be retired without the subgraph Accepting a proposal needs the indexing-payments subgraph, so the local accepted rows can only appear without one if it was unset after acceptance. The retirement path then can't run and the rules are kept silently; warn with the stuck deployments so the misconfiguration is visible. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/indexing-fees/__tests__/dips.test.ts | 44 +++++++++++++++++++ .../indexer-common/src/indexing-fees/dips.ts | 13 ++++++ 2 files changed, 57 insertions(+) diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts index 848a72477..084c0941d 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -617,6 +617,50 @@ describe('DipsManager', () => { expect(markCompleted).toHaveBeenCalledWith('accepted-1') }) + test('keeps accepted rows and warns when no indexing-payments subgraph is configured', async () => { + await managementModels.IndexingRule.create({ + identifier: testDeploymentId, + identifierType: SubgraphIdentifierType.DEPLOYMENT, + decisionBasis: IndexingDecisionBasis.DIPS, + protocolNetwork: 'eip155:421614', + allocationLifetime: 3600, + }) + jest + .spyOn(dipsManager.pendingRcaConsumer!, 'getPendingProposals') + .mockResolvedValue([]) + jest + .spyOn(dipsManager.pendingRcaConsumer!, 'getAcceptedProposals') + .mockResolvedValue([ + { + id: 'accepted-1', + agreementId: testAgreementId, + status: 'accepted', + createdAt: new Date(), + updatedAt: new Date(), + subgraphDeploymentId: new SubgraphDeploymentID(testDeploymentId), + minSecondsPerCollection: 60, + maxSecondsPerCollection: 3600, + } as never, + ]) + const markCompleted = jest.spyOn(dipsManager.pendingRcaConsumer!, 'markCompleted') + const warn = jest.spyOn(logger, 'warn') + // No subgraph configured: it can't drive retirement, so the rule is kept and + // the stuck state is surfaced rather than lingering silently forever. + network.indexingPaymentsSubgraph = undefined + + await dipsManager.ensureAgreementRules() + + expect(markCompleted).not.toHaveBeenCalled() + expect(warn).toHaveBeenCalledWith( + expect.stringContaining( + 'cannot be retired without the indexing-payments subgraph', + ), + expect.objectContaining({ + stuckDeployments: [new SubgraphDeploymentID(testDeploymentId).toString()], + }), + ) + }) + test('retires the accepted row and reaps its rule once the subgraph catches up but the agreement is gone', async () => { await managementModels.IndexingRule.create({ identifier: testDeploymentId, diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index 48392802c..72b4f24da 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -235,6 +235,19 @@ export class DipsManager { }) } } + } else if (fromAcceptedProposals.length > 0) { + // No subgraph configured but accepted rows exist (it was unset after they + // were accepted): we can't tell if their agreements still live, so surface + // the stuck rows instead of silently keeping their rules forever. + this.logger.warn( + 'DIPS accepted rows cannot be retired without the indexing-payments subgraph; ' + + 'their rules will be kept until it is configured', + { + stuckDeployments: fromAcceptedProposals.map((p) => + p.subgraphDeploymentId.toString(), + ), + }, + ) } }