diff --git a/packages/indexer-agent/.eslintrc.js b/packages/indexer-agent/.eslintrc.js index 60edb6c8b..d74df7526 100644 --- a/packages/indexer-agent/.eslintrc.js +++ b/packages/indexer-agent/.eslintrc.js @@ -2,5 +2,18 @@ module.exports = { root: true, parser: '@typescript-eslint/parser', plugins: ['@typescript-eslint'], - extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'] + extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'], + rules: { + // Pino emits a per-call field whose key matches a logger binding as a SECOND JSON + // key; strict parsers keep only the last. Forbid pino reserved keys as per-call fields. + 'no-restricted-syntax': [ + 'error', + { + selector: + "CallExpression[callee.property.name=/^(trace|debug|info|warn|error|fatal)$/] > ObjectExpression > Property[key.name=/^(name|level|time|pid|hostname|msg|v)$/]", + message: + 'Do not pass a pino reserved key (name, level, time, pid, hostname, msg, v) as a per-call log field: it collides with the logger bindings and emits a duplicate JSON key that strict parsers silently drop. Use a distinct key (e.g. subgraphName) or set bindings via logger.child({ ... }).', + }, + ], + }, } diff --git a/packages/indexer-agent/src/__tests__/agent.ts b/packages/indexer-agent/src/__tests__/agent.ts index 7c2ac17f5..e58ab8716 100644 --- a/packages/indexer-agent/src/__tests__/agent.ts +++ b/packages/indexer-agent/src/__tests__/agent.ts @@ -524,14 +524,27 @@ describe('reconcileDeployments indexing-payments carve-out wiring', () => { trace: jest.fn(), } - // eslint-disable-next-line @typescript-eslint/no-explicit-any - function createAgentUnderTest(network: any) { + function createAgentUnderTest( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + network: any, + dipsDeployments: SubgraphDeploymentID[] = [], + dipsManagerPresent = true, + ) { + const operator = { + dipsManager: dipsManagerPresent + ? { + getActiveDipsDeployments: jest + .fn() + .mockResolvedValue(dipsDeployments), + } + : null, + } const agent = Object.create(Agent.prototype) agent.logger = mockLogger agent.offchainSubgraphs = [] agent.multiNetworks = { // eslint-disable-next-line @typescript-eslint/no-explicit-any - map: async (fn: any) => Promise.all([fn({ network })]), + map: async (fn: any) => Promise.all([fn({ network, operator })]), } agent.graphNode = { subgraphDeploymentsAssignments: jest.fn().mockResolvedValue([]), @@ -581,4 +594,68 @@ describe('reconcileDeployments indexing-payments carve-out wiring', () => { expect(agent.graphNode.ensure).not.toHaveBeenCalled() }) + + it('does not pause an active deployment that has an active DIPS agreement', async () => { + const dipsDeployment = new SubgraphDeploymentID( + 'QmWTbiUJQPEYDdUxt7sS8EWGwkxWmWTApJPaeGa4NXMqHQ', + ) + const agent = createAgentUnderTest( + { + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: true } }, + indexingPaymentsSubgraph: { deployment: undefined }, + }, + [dipsDeployment], + ) + + // Active (being indexed) but absent from the target and eligible-allocation sets. + await agent.reconcileDeployments([dipsDeployment], [], []) + + const pausedDeployments = agent.graphNode.pause.mock.calls.map( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (call: any[]) => (call[0] as SubgraphDeploymentID).bytes32, + ) + expect(pausedDeployments).not.toContain(dipsDeployment.bytes32) + }) + + it('pauses an active deployment with no DIPS agreement that is not targeted', async () => { + const orphan = new SubgraphDeploymentID( + 'QmNYBVzrWYrhmNF7srCs9qNUUnK1urXA1dNACrRp9xVPrH', + ) + const agent = createAgentUnderTest( + { + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: true } }, + indexingPaymentsSubgraph: { deployment: undefined }, + }, + [], + ) + + await agent.reconcileDeployments([orphan], [], []) + + const pausedDeployments = agent.graphNode.pause.mock.calls.map( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (call: any[]) => (call[0] as SubgraphDeploymentID).bytes32, + ) + expect(pausedDeployments).toContain(orphan.bytes32) + }) + + it('skips DIPS protection without throwing when the DipsManager is not yet available', async () => { + const orphan = new SubgraphDeploymentID( + 'QmNYBVzrWYrhmNF7srCs9qNUUnK1urXA1dNACrRp9xVPrH', + ) + const agent = createAgentUnderTest( + { + networkSubgraph: { deployment: undefined }, + specification: { indexerOptions: { enableDips: true } }, + indexingPaymentsSubgraph: { deployment: undefined }, + }, + [], + false, // DIPS enabled, but the DipsManager has not been registered yet. + ) + + await expect( + agent.reconcileDeployments([orphan], [], []), + ).resolves.not.toThrow() + }) }) diff --git a/packages/indexer-agent/src/agent.ts b/packages/indexer-agent/src/agent.ts index bd7be1323..c7629d540 100644 --- a/packages/indexer-agent/src/agent.ts +++ b/packages/indexer-agent/src/agent.ts @@ -296,10 +296,13 @@ export class Agent { async () => { return this.multiNetworks.map(async ({ network, operator }) => { if (network.specification.indexerOptions.enableDips) { - logger.debug('Ensuring indexing rules for DIPs', { - protocolNetwork: network.specification.networkIdentifier, - }) - await operator.dipsManager!.ensureAgreementRules() + const dipsManager = this.readyDipsManager(network, operator) + if (dipsManager) { + logger.debug('Ensuring indexing rules for DIPs', { + protocolNetwork: network.specification.networkIdentifier, + }) + await dipsManager.ensureAgreementRules() + } } else { logger.debug( 'DIPs is disabled, skipping indexing rule enforcement', @@ -384,14 +387,14 @@ export class Agent { protocolNetwork: network.specification.networkIdentifier, }) const deployments = network.networkMonitor.subgraphDeployments() - if (network.specification.indexerOptions.enableDips) { + const dipsManager = network.specification.indexerOptions.enableDips + ? this.readyDipsManager(network, operator) + : null + if (dipsManager) { const resolvedDeployments = await deployments const dipsDeployments = await Promise.all( - (await operator.dipsManager!.getActiveDipsDeployments()).map( - deployment => - network.networkMonitor.subgraphDeployment( - deployment.ipfsHash, - ), + (await dipsManager.getActiveDipsDeployments()).map(deployment => + network.networkMonitor.subgraphDeployment(deployment.ipfsHash), ), ) for (const deployment of dipsDeployments) { @@ -580,7 +583,6 @@ export class Agent { disputableAllocations, }).pipe( async ({ - currentEpochNumber, maxAllocationDuration, activeDeployments, targetDeployments, @@ -589,8 +591,22 @@ export class Agent { recentlyClosedAllocations, disputableAllocations, }) => { + // Read the epoch fresh once per pass instead of the separately-timed Eventual, which + // can lag the chain by an epoch and make a just-created allocation look stale. + const currentEpochNumberWithProvenance = await this.multiNetworks.map( + async ({ network }) => + network.networkMonitor.currentEpochNumberWithProvenance(), + ) + const currentEpochNumber = await this.multiNetworks.mapNetworkMapped( + currentEpochNumberWithProvenance, + async ( + _: NetworkAndOperator, + provenance: { epoch: number; readAtBlock: number }, + ) => provenance.epoch, + ) logger.info(`Reconcile with the network`, { currentEpochNumber, + epochProvenance: currentEpochNumberWithProvenance, }) try { @@ -649,22 +665,18 @@ export class Agent { } break case DeploymentManagementMode.MANUAL: - await this.multiNetworks.map(async ({ network, operator }) => { + await this.multiNetworks.map(async ({ network }) => { if (network.specification.indexerOptions.enableDips) { - // Reconcile DIPs deployments anyways + // Manual mode normally leaves deployments untouched, but reconcileDeployments + // keeps deployments with an active DIPS agreement out of the pause path, so + // still run it here; it resolves the active DIPS deployments itself. this.logger.warn( `Deployment management is manual, but DIPs is enabled. Reconciling DIPs deployments anyways.`, ) - const dipsDeployments = - await operator.dipsManager!.getActiveDipsDeployments() - const newTargetDeployments = new Set([ - ...activeDeployments, - ...dipsDeployments, - ]) try { await this.reconcileDeployments( activeDeployments, - Array.from(newTargetDeployments), + [...activeDeployments], eligibleAllocations, ) } catch (err) { @@ -863,6 +875,22 @@ export class Agent { }) } + // The DipsManager is built lazily once the action manager registers its allocation + // manager, so it can briefly be absent at startup even with DIPS enabled. Return it when + // ready, otherwise null with a debug log so callers skip their DIPS step rather than crash. + private readyDipsManager( + network: Network, + operator: Operator, + ): Operator['dipsManager'] { + if (!operator.dipsManager) { + this.logger.debug( + 'DIPS enabled but DipsManager not ready; skipping DIPS work this pass', + { protocolNetwork: network.specification.networkIdentifier }, + ) + } + return operator.dipsManager + } + // This function assumes that allocations and deployments passed to it have already // been retrieved from multiple networks. async reconcileDeployments( @@ -897,6 +925,25 @@ export class Agent { ) }) + // Keep deployments with an active DIPS agreement indexed even when a lagging or removed + // DIPS rule has dropped them from the target set; otherwise the pause path below would + // stop indexing a deployment the agreement is still paying for. + await this.multiNetworks.map(async ({ network, operator }) => { + if (!network.specification.indexerOptions.enableDips) { + return + } + const dipsManager = this.readyDipsManager(network, operator) + if (!dipsManager) { + return + } + const dipsDeployments = await dipsManager.getActiveDipsDeployments() + for (const deployment of dipsDeployments) { + if (!deploymentInList(targetDeployments, deployment)) { + targetDeployments.push(deployment) + } + } + }) + // ---------------------------------------------------------------------------------------- // Inspect Deployments and Networks // ---------------------------------------------------------------------------------------- @@ -954,7 +1001,7 @@ export class Agent { const name = `indexer-agent/${deployment.ipfsHash.slice(-10)}` logger.info(`Index subgraph deployment`, { - name, + subgraphName: name, deployment: deployment.display, }) diff --git a/packages/indexer-agent/src/db/cli/umzug.ts b/packages/indexer-agent/src/db/cli/umzug.ts index 71654a44f..c117fa6f0 100644 --- a/packages/indexer-agent/src/db/cli/umzug.ts +++ b/packages/indexer-agent/src/db/cli/umzug.ts @@ -50,7 +50,7 @@ const sequelize = new Sequelize({ logging: false, }) -logger.debug('Successfully connected to DB', { name: database }) +logger.debug('Successfully connected to DB', { database }) export const migrator = new Umzug({ migrations: { diff --git a/packages/indexer-cli/.eslintrc.js b/packages/indexer-cli/.eslintrc.js index 437698c81..9189b3077 100644 --- a/packages/indexer-cli/.eslintrc.js +++ b/packages/indexer-cli/.eslintrc.js @@ -2,5 +2,18 @@ module.exports = { root: false, parser: '@typescript-eslint/parser', plugins: ['@typescript-eslint'], - extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'] + extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'], + rules: { + // Pino emits a per-call field whose key matches a logger binding as a SECOND JSON + // key; strict parsers keep only the last. Forbid pino reserved keys as per-call fields. + 'no-restricted-syntax': [ + 'error', + { + selector: + "CallExpression[callee.property.name=/^(trace|debug|info|warn|error|fatal)$/] > ObjectExpression > Property[key.name=/^(name|level|time|pid|hostname|msg|v)$/]", + message: + 'Do not pass a pino reserved key (name, level, time, pid, hostname, msg, v) as a per-call log field: it collides with the logger bindings and emits a duplicate JSON key that strict parsers silently drop. Use a distinct key (e.g. subgraphName) or set bindings via logger.child({ ... }).', + }, + ], + }, } diff --git a/packages/indexer-common/.eslintrc.js b/packages/indexer-common/.eslintrc.js index 437698c81..9189b3077 100644 --- a/packages/indexer-common/.eslintrc.js +++ b/packages/indexer-common/.eslintrc.js @@ -2,5 +2,18 @@ module.exports = { root: false, parser: '@typescript-eslint/parser', plugins: ['@typescript-eslint'], - extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'] + extends: ['eslint:recommended', 'plugin:@typescript-eslint/recommended', 'prettier'], + rules: { + // Pino emits a per-call field whose key matches a logger binding as a SECOND JSON + // key; strict parsers keep only the last. Forbid pino reserved keys as per-call fields. + 'no-restricted-syntax': [ + 'error', + { + selector: + "CallExpression[callee.property.name=/^(trace|debug|info|warn|error|fatal)$/] > ObjectExpression > Property[key.name=/^(name|level|time|pid|hostname|msg|v)$/]", + message: + 'Do not pass a pino reserved key (name, level, time, pid, hostname, msg, v) as a per-call log field: it collides with the logger bindings and emits a duplicate JSON key that strict parsers silently drop. Use a distinct key (e.g. subgraphName) or set bindings via logger.child({ ... }).', + }, + ], + }, } diff --git a/packages/indexer-common/src/graph-node.ts b/packages/indexer-common/src/graph-node.ts index 836985514..b238a40cf 100644 --- a/packages/indexer-common/src/graph-node.ts +++ b/packages/indexer-common/src/graph-node.ts @@ -536,16 +536,16 @@ export class GraphNode { async create(name: string): Promise { try { - this.logger.info(`Create subgraph name`, { name }) + this.logger.info(`Create subgraph name`, { subgraphName: name }) const response = await this.admin.request('subgraph_create', { name }) if (response.error) { throw response.error } - this.logger.info(`Successfully created subgraph name`, { name }) + this.logger.info(`Successfully created subgraph name`, { subgraphName: name }) } catch (error) { if (error.message.includes('already exists')) { this.logger.debug(`Subgraph name already exists, will deploy to existing name`, { - name, + subgraphName: name, }) return } @@ -556,7 +556,7 @@ export class GraphNode { async deploy(name: string, deployment: SubgraphDeploymentID): Promise { try { this.logger.info(`Deploy subgraph deployment`, { - name, + subgraphName: name, deployment: deployment.display, }) const response = await this.admin.request('subgraph_deploy', { @@ -566,7 +566,7 @@ export class GraphNode { this.logger.trace(`Response from 'subgraph_deploy' call`, { deployment: deployment.display, - name, + subgraphName: name, response, }) @@ -574,7 +574,7 @@ export class GraphNode { throw response.error } this.logger.info(`Successfully deployed subgraph deployment`, { - name, + subgraphName: name, deployment: deployment.display, }) } catch (error) { @@ -587,7 +587,7 @@ export class GraphNode { const err = indexerError(errorCode, error) this.logger.error(INDEXER_ERROR_MESSAGES[errorCode], { - name, + subgraphName: name, deployment: deployment.display, err, }) @@ -678,7 +678,7 @@ export class GraphNode { currentAssignments?: SubgraphDeploymentAssignment[], ): Promise { this.logger.debug('Ensure subgraph deployment is syncing', { - name, + subgraphName: name, deployment: deployment.ipfsHash, }) try { @@ -693,12 +693,12 @@ export class GraphNode { if (matchingAssignment?.paused == false) { this.logger.debug('Subgraph deployment already syncing, ensure() is a no-op', { - name, + subgraphName: name, deployment: deployment.ipfsHash, }) } else if (matchingAssignment?.paused == true) { this.logger.debug('Subgraph deployment paused, resuming', { - name, + subgraphName: name, deployment: deployment.ipfsHash, }) await this.resume(deployment) @@ -714,7 +714,7 @@ export class GraphNode { this.logger.debug( 'Subgraph deployment not found, creating subgraph name and deploying...', { - name, + subgraphName: name, deployment: deployment.ipfsHash, }, ) @@ -725,7 +725,7 @@ export class GraphNode { if (!(error instanceof IndexerError)) { const errorCode = IndexerErrorCode.IE020 this.logger.error(INDEXER_ERROR_MESSAGES[errorCode], { - name, + subgraphName: name, deployment: deployment.display, error: indexerError(errorCode, error), }) @@ -753,7 +753,7 @@ export class GraphNode { // Safety check - should not happen if called correctly from ensure() if (!this.manifestResolver) { this.logger.error('Auto-graft called but manifest resolver not initialized', { - name, + subgraphName: name, deployment: deployment.display, }) return @@ -763,7 +763,7 @@ export class GraphNode { const dependencies = await this.manifestResolver.resolveWithDependencies(deployment) if (dependencies.dependencies.length == 0) { this.logger.debug('No subgraph dependencies found', { - name, + subgraphName: name, deployment: deployment.display, }) } else { @@ -788,7 +788,7 @@ export class GraphNode { if (dependencyAssignment) { this.logger.info("Dependency subgraph found, checking if it's healthy", { - name, + subgraphName: name, deployment: dependency.base.display, block_required: dependency.block, }) @@ -940,11 +940,9 @@ export class GraphNode { throw new Error(`Chain not found in indexing status for deployment`) } - // NOTES: - // - latestBlock is the latest block that has been indexed - // - earliestBlock and chainHeadBlock are the earliest and latest blocks on the chain, respectively - // if the deployment is paused and latestBlock is null or lower than we need, unpause it, - // otherwise, if it's paused, we can't unpause it, so just wait + // Unpause a paused deployment only when its indexed head (latestBlock) is null + // or below the block we need; once paused past that point it can't be resumed, + // so wait instead. if ( deployed[0].paused && (!chain.latestBlock || chain.latestBlock.number < blockHeight) diff --git a/packages/indexer-common/src/indexer-management/monitor.ts b/packages/indexer-common/src/indexer-management/monitor.ts index c4771d89e..32daaf1cb 100644 --- a/packages/indexer-common/src/indexer-management/monitor.ts +++ b/packages/indexer-common/src/indexer-management/monitor.ts @@ -118,13 +118,30 @@ export class NetworkMonitor { return Number(await this.contracts.EpochManager.currentEpoch()) } + // Read the chain block alongside the epoch so a reconcile pass can log where its epoch + // came from; the block is provenance only, the epoch is what every decision then uses. + async currentEpochNumberWithProvenance(): Promise<{ + epoch: number + readAtBlock: number + }> { + const readAtBlock = await this.ethereum.getBlockNumber() + const epoch = Number(await this.contracts.EpochManager.currentEpoch()) + return { epoch, readAtBlock } + } + + // One epoch in seconds, the single place that turns the on-chain epoch length (in + // blocks) into wall-clock time for the seconds<->epochs conversions callers need. + // TODO HORIZON: assumes a 12s block time, true for the current protocol chain but not always. + async epochLengthInSeconds(): Promise { + const BLOCK_IN_SECONDS = 12n + const epochLengthInBlocks = await this.contracts.EpochManager.epochLength() + return Number(epochLengthInBlocks * BLOCK_IN_SECONDS) + } + // Maximum allocation duration is measured in seconds, determined by maxPOIStaleness. // This function converts the value to epochs. async maxAllocationDuration(): Promise { - // TODO HORIZON: this assumes a block time of 12 seconds which is true for current protocol chain but not always - const BLOCK_IN_SECONDS = 12n - const epochLengthInBlocks = await this.contracts.EpochManager.epochLength() - const epochLengthInSeconds = Number(epochLengthInBlocks * BLOCK_IN_SECONDS) + const epochLengthInSeconds = await this.epochLengthInSeconds() // When converting to epochs we give it a bit of leeway since missing the allocation expiration in horizon // incurs in a severe penalty (missing out on indexing rewards) @@ -164,10 +181,8 @@ export class NetworkMonitor { * @returns network `alias` if the network is supported, `null` otherwise */ async allocationNetworkAlias(allocation: Allocation): Promise { - // TODO: - // resolveChainId will throw an Error when we can't resolve the chainId in - // the future, let's get this from the epoch subgraph (perhaps at startup) - // and then resolve it here. + // TODO: resolveChainId will throw when we can't resolve the chainId; in the future + // get this from the epoch subgraph (perhaps at startup) and resolve it here. try { const { network: allocationNetworkAlias } = await this.graphNode.subgraphFeatures( allocation.subgraphDeployment.id, @@ -947,7 +962,7 @@ Please submit an issue at https://github.com/graphprotocol/block-oracle/issues/n } else { this.logger.error(`Failed to query latest epoch number`, { err, - msg: err.message, + errorMessage: err.message, networkID, networkAlias, }) @@ -1388,10 +1403,9 @@ Please submit an issue at https://github.com/graphprotocol/block-oracle/issues/n return [hexlify(new Uint8Array(32).fill(0)), 0] } - // poi = undefined, force=true -- submit even if poi is 0x0 - // poi = defined, force=true -- no generatedPOI needed, just submit the POI supplied (with some sanitation?) - // poi = undefined, force=false -- submit with generated POI if one available - // poi = defined, force=false -- submit user defined POI only if generated POI matches + // force=true: poi undefined -> submit even if 0x0; poi defined -> submit the supplied POI + // force=false: poi undefined -> submit a generated POI if available; poi defined -> submit the + // user POI only if it matches the generated POI switch (force) { case true: switch (!!poi) { 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 bc8d5f87c..6153efa2e 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 @@ -154,6 +154,8 @@ function createMockNetwork() { }, networkMonitor: { currentEpoch: jest.fn().mockResolvedValue(100n), + // 5 blocks x 12s = 60s/epoch for the seconds->epochs allocation-lifetime conversion. + epochLengthInSeconds: jest.fn().mockResolvedValue(60), }, specification: { indexerOptions: { 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 406dd1c05..3c4871de7 100644 --- a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts +++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts @@ -213,6 +213,13 @@ describe('DipsManager', () => { // Clear mock calls between tests jest.clearAllMocks() + // Deterministic epoch length for the seconds→epochs conversion: 5 blocks × 12s = 60s/epoch. + network.contracts.EpochManager.epochLength = jest + .fn() + .mockResolvedValue( + 5n, + ) as unknown as typeof network.contracts.EpochManager.epochLength + const allocationManager = new AllocationManager( logger, managementModels, @@ -264,7 +271,7 @@ describe('DipsManager', () => { identifierType: SubgraphIdentifierType.DEPLOYMENT, decisionBasis: IndexingDecisionBasis.DIPS, autoRenewal: true, - allocationLifetime: 3600, // max(min, max) seconds + allocationLifetime: 60, // ceil(max(min,max)=3600s / 60s-per-epoch) }) }) @@ -302,7 +309,7 @@ describe('DipsManager', () => { }) expect(rules).toHaveLength(1) expect(rules[0].decisionBasis).toBe(IndexingDecisionBasis.DIPS) - expect(rules[0].allocationLifetime).toBe(1800) + expect(rules[0].allocationLifetime).toBe(30) }) test('deduplicates when a deployment has both a pending proposal and an accepted agreement', async () => { 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 88e0d64ea..7dd323da9 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 @@ -220,22 +220,16 @@ describe('PendingRcaConsumer', () => { expect(proposals[1].tokensPerSecond).toBe(200n) }) - test('rejects rows with non-empty signature (producer regression)', async () => { - const errorSpy = jest.fn() - const testLogger = { - ...logger, - error: errorSpy, - info: jest.fn(), - warn: jest.fn(), - child: () => testLogger, - } as unknown as Logger - - const badPayload = encodeTestPayload({ signature: '0xdeadbeef' }) + test('ignores a non-empty signer signature and decodes the proposal', async () => { + const signedPayload = encodeTestPayload({ + signature: '0xdeadbeef', + tokensPerSecond: 150n, + }) const model = createMockModel([ { - id: 'bad-sig-uuid', - signed_payload: badPayload, + id: 'signed-uuid', + signed_payload: signedPayload, version: 2, status: 'pending', created_at: new Date(), @@ -243,18 +237,13 @@ describe('PendingRcaConsumer', () => { }, ]) - const consumer = new PendingRcaConsumer(testLogger, model) + const consumer = new PendingRcaConsumer(logger, model) const proposals = await consumer.getPendingProposals() - expect(proposals).toHaveLength(0) - expect(model.update).toHaveBeenCalledWith( - { status: 'rejected' }, - { where: { id: 'bad-sig-uuid' } }, - ) - expect(errorSpy).toHaveBeenCalledWith( - expect.stringContaining('non-empty signature'), - expect.any(Object), - ) + expect(proposals).toHaveLength(1) + expect(proposals[0].id).toBe('signed-uuid') + expect(proposals[0].tokensPerSecond).toBe(150n) + expect(model.update).not.toHaveBeenCalled() }) }) diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts index 92c65910d..ed5dafe97 100644 --- a/packages/indexer-common/src/indexing-fees/dips.ts +++ b/packages/indexer-common/src/indexing-fees/dips.ts @@ -118,7 +118,7 @@ export class DipsManager { continue } await this.upsertDipsRuleFor(deploymentId, { - allocationLifetime: Math.max( + maxCollectionSeconds: Math.max( Number(proposal.minSecondsPerCollection), Number(proposal.maxSecondsPerCollection), ), @@ -139,7 +139,7 @@ export class DipsManager { continue } await this.upsertDipsRuleFor(deploymentId, { - allocationLifetime: Math.max( + maxCollectionSeconds: Math.max( Number(agreement.minSecondsPerCollection), Number(agreement.maxSecondsPerCollection), ), @@ -161,7 +161,7 @@ export class DipsManager { continue } await this.upsertDipsRuleFor(deploymentId, { - allocationLifetime: Math.max( + maxCollectionSeconds: Math.max( Number(proposal.minSecondsPerCollection), Number(proposal.maxSecondsPerCollection), ), @@ -253,7 +253,7 @@ export class DipsManager { private async upsertDipsRuleFor( deploymentId: SubgraphDeploymentID, - opts: { allocationLifetime: number }, + opts: { maxCollectionSeconds: number }, ): Promise { const ruleExists = await this.parent!.matchingRuleExists(this.logger, deploymentId) if (ruleExists) { @@ -270,6 +270,7 @@ export class DipsManager { ) const { amount } = await this.getDipsAllocationAmount(deploymentId) + const allocationLifetime = await this.secondsToEpochs(opts.maxCollectionSeconds) this.logger.info( `Creating DIPS indexing rule for deployment ${deploymentId.toString()}`, ) @@ -280,11 +281,21 @@ export class DipsManager { decisionBasis: IndexingDecisionBasis.DIPS, protocolNetwork: this.network.specification.networkIdentifier, autoRenewal: true, - allocationLifetime: opts.allocationLifetime, + allocationLifetime, requireSupported: false, } as Partial) } + // A DIPS agreement's collection window is in seconds, but allocationLifetime is consumed + // in epochs (agent expires at createdAtEpoch + allocationLifetime). Convert so a seconds + // value like 86400 isn't read as epochs, which would make the allocation never expire. + private async secondsToEpochs(seconds: number): Promise { + const epochLengthInSeconds = await this.network.networkMonitor.epochLengthInSeconds() + // Round up so the allocation outlives the window; floor at 1 epoch so a sub-epoch + // window still yields a valid non-zero lifetime (0 would expire the allocation each tick). + return Math.max(1, Math.ceil(seconds / epochLengthInSeconds)) + } + private async getDipsTargetDeployments(): Promise<{ fromPendingProposals: DecodedRcaProposal[] fromAcceptedProposals: DecodedRcaProposal[] @@ -514,7 +525,7 @@ export class DipsManager { return } await this.upsertDipsRuleFor(proposal.subgraphDeploymentId, { - allocationLifetime: Math.max( + maxCollectionSeconds: Math.max( Number(proposal.minSecondsPerCollection), Number(proposal.maxSecondsPerCollection), ), 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 bc51d9cf5..b5ab52f28 100644 --- a/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts +++ b/packages/indexer-common/src/indexing-fees/pending-rca-consumer.ts @@ -73,34 +73,13 @@ export class PendingRcaConsumer { } } - // Returns the decoded proposal, or null if the row was rejected here - // (non-empty signature — producer regression, marked rejected in the DB - // before returning). - // - // Decode failures from toolshed (malformed payload, bad metadata, etc.) - // propagate as throws; the caller skip-logs them, leaving the row pending - // for the next cycle. + // Returns the decoded proposal, or throws on a malformed payload (the caller + // skip-logs those, leaving the row pending). An embedded signer signature is + // ignored — acceptance is offer-based, so the agent has no use for it. private async decodeRow(row: PendingRcaProposal): Promise { const signedPayload = new Uint8Array(row.signed_payload) const signedRca = decodeSignedRCA(signedPayload) - const { rca, signature } = signedRca - - if (signature && signature !== '0x') { - const sigByteLength = Math.max(0, (signature.length - 2) / 2) - this.logger.error( - `Pending RCA proposal ${row.id} has non-empty signature (producer regression); rejecting`, - { id: row.id, signatureLength: sigByteLength }, - ) - try { - await this.markRejected(row.id, 'non_empty_signature') - } catch (err) { - this.logger.error('Failed to mark non-empty-signature proposal as rejected', { - id: row.id, - error: err instanceof Error ? err.message : String(err), - }) - } - return null - } + const { rca } = signedRca const metadata = decodeAcceptIndexingAgreementMetadata(rca.metadata) const terms = decodeIndexingAgreementTermsV1(metadata.terms) diff --git a/packages/indexer-common/src/network.ts b/packages/indexer-common/src/network.ts index 3aa87f90a..1eb900dec 100644 --- a/packages/indexer-common/src/network.ts +++ b/packages/indexer-common/src/network.ts @@ -125,7 +125,7 @@ export class Network { networkProvider, specification.subgraphs.maxBlockDistance, specification.subgraphs.freshnessSleepMilliseconds, - logger.child({ component: 'FreshnessChecker' }), + logger.child({ subComponent: 'FreshnessChecker' }), Infinity, ) @@ -162,7 +162,7 @@ export class Network { networkProvider, specification.subgraphs.maxBlockDistance, specification.subgraphs.freshnessSleepMilliseconds, - logger.child({ component: 'FreshnessChecker' }), + logger.child({ subComponent: 'FreshnessChecker' }), Infinity, ) indexingPaymentsSubgraph = await SubgraphClient.create({ @@ -214,7 +214,7 @@ export class Network { networkProvider, specification.subgraphs.maxBlockDistance, specification.subgraphs.freshnessSleepMilliseconds, - logger.child({ component: 'FreshnessChecker' }), + logger.child({ subComponent: 'FreshnessChecker' }), Infinity, ) @@ -243,8 +243,7 @@ export class Network { contracts, specification.indexerOptions, logger.child({ - component: 'NetworkMonitor', - protocolNetwork: specification.networkIdentifier, + subComponent: 'NetworkMonitor', }), graphNode, networkSubgraph, diff --git a/packages/indexer-common/src/utils.ts b/packages/indexer-common/src/utils.ts index e66b678ef..08b91b2c6 100644 --- a/packages/indexer-common/src/utils.ts +++ b/packages/indexer-common/src/utils.ts @@ -46,7 +46,7 @@ export async function monitorEthBalance( metrics: Metrics, networkIdentifier: string, ): Promise { - logger = logger.child({ component: 'ETHBalanceMonitor' }) + logger = logger.child({ subComponent: 'ETHBalanceMonitor' }) logger.info('Monitor operator ETH balance (refreshes every 120s)')