chore: add is horizon ready flag to network monitor

Author: Maikol

packages/indexer-common/src/indexer-management/__tests__/horizon-ready.test.ts

@@ -0,0 +1,135 @@
+import { createLogger, Logger } from '@graphprotocol/common-ts'
+import { NetworkMonitor } from '../monitor'
+import { getTestProvider } from '../../utils'
+import { resolveChainId } from '../../indexer-management'
+import { GraphNode } from '../../graph-node'
+import { SubgraphClient } from '../../subgraph-client'
+import { SubgraphFreshnessChecker } from '../../subgraphs'
+import { mockLogger, mockProvider } from '../../__tests__/subgraph.test'
+import { specification as spec } from '../../index'
+import { connectGraphHorizon, connectSubgraphService } from '@graphprotocol/toolshed/deployments'
+import { Provider } from 'ethers'
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+declare const __LOG_LEVEL__: any
+
+describe('Horizon readiness', () => {
+  let logger: Logger
+  let ethereum: Provider
+  let graphNode: GraphNode
+  let networkSubgraph: SubgraphClient
+  let epochSubgraph: SubgraphClient
+  let networkMonitor: NetworkMonitor
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  let mockHorizonStaking: any
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  let contracts: any
+
+  const setupMonitor = async () => {
+    logger = createLogger({
+      name: 'Horizon readiness tests',
+      async: false,
+      level: __LOG_LEVEL__ ?? 'error',
+    })
+    ethereum = getTestProvider('sepolia')
+    
+    mockHorizonStaking = {
+      getMaxThawingPeriod: jest.fn(),
+      connect: jest.fn(),
+      waitForDeployment: jest.fn(),
+      interface: {},
+      queryFilter: jest.fn(),
+    }
+    
+    const horizonContracts = {
+      ...connectGraphHorizon(5, ethereum),
+      HorizonStaking: mockHorizonStaking,
+    }
+    contracts = {
+      ...horizonContracts,
+      ...connectSubgraphService(5, ethereum),
+    }
+
+    const subgraphFreshnessChecker = new SubgraphFreshnessChecker(
+      'Test Subgraph',
+      mockProvider,
+      10,
+      10,
+      mockLogger,
+      1,
+    )
+
+    networkSubgraph = await SubgraphClient.create({
+      name: 'NetworkSubgraph',
+      logger,
+      endpoint: 'http://test-endpoint.xyz',
+      deployment: undefined,
+      subgraphFreshnessChecker,
+    })
+
+    epochSubgraph = await SubgraphClient.create({
+      name: 'EpochSubgraph',
+      logger,
+      endpoint: 'http://test-endpoint.xyz',
+      subgraphFreshnessChecker,
+    })
+
+    graphNode = new GraphNode(
+      logger,
+      'http://test-admin-endpoint.xyz',
+      'https://test-query-endpoint.xyz',
+      'http://test-status-endpoint.xyz',
+      'https://test-ipfs-endpoint.xyz',
+    )
+
+    const indexerOptions = spec.IndexerOptions.parse({
+      address: '0xc61127cdfb5380df4214b0200b9a07c7c49d34f9',
+      mnemonic: 'word ivory whale diesel slab pelican voyage oxygen chat find tobacco sport',
+      url: 'http://test-url.xyz',
+    })
+
+    networkMonitor = new NetworkMonitor(
+      resolveChainId('sepolia'),
+      contracts,
+      indexerOptions,
+      logger,
+      graphNode,
+      networkSubgraph,
+      ethereum,
+      epochSubgraph,
+    )
+  }
+
+  beforeEach(setupMonitor)
+
+  describe('monitorIsHorizon', () => {
+    beforeEach(() => {
+      mockHorizonStaking.getMaxThawingPeriod.mockResolvedValue(0)
+    })
+
+    test('should return false when getMaxThawingPeriod returns 0', async () => {
+      const isHorizon = await networkMonitor.monitorIsHorizon(logger, { ...contracts, HorizonStaking: mockHorizonStaking })
+      const value = await isHorizon.value()
+      expect(value).toBe(false)
+    })
+
+    test('should return true when getMaxThawingPeriod returns > 0', async () => {
+      mockHorizonStaking.getMaxThawingPeriod.mockResolvedValue(1000)
+
+      const isHorizon = await networkMonitor.monitorIsHorizon(logger, { ...contracts, HorizonStaking: mockHorizonStaking })
+      
+      const value = await isHorizon.value()
+      console.log('Final value:', value)
+      expect(value).toBe(true)
+    })
+
+    test('should handle errors and maintain previous state', async () => {
+      mockHorizonStaking.getMaxThawingPeriod.mockRejectedValue(new Error('Contract error'))
+
+      const isHorizon = await networkMonitor.monitorIsHorizon(logger, { ...contracts, HorizonStaking: mockHorizonStaking })
+      
+      const value = await isHorizon.value()
+      expect(value).toBe(false)
+    })
+  })
+})

packages/indexer-common/src/indexer-management/monitor.ts

@@ -1086,6 +1086,47 @@ Please submit an issue at https://github.com/graphprotocol/block-oracle/issues/n
     })
   }
 
+  async monitorIsHorizon(
+    logger: Logger,
+    contracts: GraphHorizonContracts & SubgraphServiceContracts,
+    interval: number = 300_000,
+  ): Promise<Eventual<boolean>> {
+    // Get initial value
+    
+    const initialValue = await contracts.HorizonStaking.getMaxThawingPeriod()
+      .then(maxThawingPeriod => maxThawingPeriod > 0)
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      .catch(_ => false)
+
+    return sequentialTimerReduce(
+      {
+        logger,
+        milliseconds: interval,
+      },
+      async (isHorizon) => {
+        try {
+          logger.debug('Check if network is Horizon ready')
+          const maxThawingPeriod = await contracts.HorizonStaking.getMaxThawingPeriod()
+          return maxThawingPeriod > 0
+        } catch (err) {
+          logger.warn(
+            `Failed to check if network is Horizon ready, assuming it has not changed`,
+            { err: indexerError(IndexerErrorCode.IE008, err), isHorizon },
+          )
+          return isHorizon
+        }
+      },
+      initialValue,
+    ).map((isHorizon) => {
+      logger.info(
+        isHorizon
+          ? `Network is Horizon ready`
+          : `Network is not Horizon ready`,
+      )
+      return isHorizon
+    })
+  }
+
   async claimableAllocations(disputableEpoch: number): Promise<Allocation[]> {
     try {
       this.logger.debug('Fetch claimable allocations', {

packages/indexer-common/src/network.ts

@@ -54,6 +54,7 @@ export class Network {
   specification: spec.NetworkSpecification
   paused: Eventual<boolean>
   isOperator: Eventual<boolean>
+  isHorizon: Eventual<boolean>
 
   private constructor(
     logger: Logger,
@@ -68,6 +69,7 @@ export class Network {
     specification: spec.NetworkSpecification,
     paused: Eventual<boolean>,
     isOperator: Eventual<boolean>,
+    isHorizon: Eventual<boolean>,
   ) {
     this.logger = logger
     this.contracts = contracts
@@ -81,6 +83,7 @@ export class Network {
     this.specification = specification
     this.paused = paused
     this.isOperator = isOperator
+    this.isHorizon = isHorizon
   }
 
   static async create(
@@ -253,11 +256,17 @@ export class Network {
       wallet,
     )
 
+    const isHorizon = await networkMonitor.monitorIsHorizon(
+      logger,
+      contracts,
+    )
+
     const transactionManager = new TransactionManager(
       networkProvider,
       wallet,
       paused,
       isOperator,
+      isHorizon,
       specification.transactionMonitoring,
     )
 
@@ -347,6 +356,7 @@ export class Network {
       specification,
       paused,
       isOperator,
+      isHorizon,
     )
   }
 

packages/indexer-common/src/transactions.ts

@@ -30,6 +30,7 @@ export class TransactionManager {
   wallet: HDNodeWallet
   paused: Eventual<boolean>
   isOperator: Eventual<boolean>
+  isHorizon: Eventual<boolean>
   specification: TransactionMonitoring
   adjustedGasIncreaseFactor: bigint
   adjustedBaseFeePerGasMax: number
@@ -39,12 +40,14 @@ export class TransactionManager {
     wallet: HDNodeWallet,
     paused: Eventual<boolean>,
     isOperator: Eventual<boolean>,
+    isHorizon: Eventual<boolean>,
     specification: TransactionMonitoring,
   ) {
     this.ethereum = ethereum
     this.wallet = wallet
     this.paused = paused
     this.isOperator = isOperator
+    this.isHorizon = isHorizon
     this.specification = specification
     this.adjustedGasIncreaseFactor = parseUnits(
       specification.gasIncreaseFactor.toString(),
@@ -420,6 +423,46 @@ export class TransactionManager {
     })
   }
 
+  async monitorIsHorizon(
+    logger: Logger,
+    contracts: GraphHorizonContracts & SubgraphServiceContracts,
+    interval: number = 300_000,
+  ): Promise<Eventual<boolean>> {
+    // Get initial value
+    const initialValue = await contracts.HorizonStaking.getMaxThawingPeriod()
+      .then(maxThawingPeriod => maxThawingPeriod > 0)
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      .catch(_ => false)
+
+    return sequentialTimerReduce(
+      {
+        logger,
+        milliseconds: interval,
+      },
+      async (isHorizon) => {
+        try {
+          logger.debug('Check if network is Horizon ready')
+          const maxThawingPeriod = await contracts.HorizonStaking.getMaxThawingPeriod()
+          return maxThawingPeriod > 0
+        } catch (err) {
+          logger.warn(
+            `Failed to check if network is Horizon ready, assuming it has not changed`,
+            { err: indexerError(IndexerErrorCode.IE008, err), isHorizon },
+          )
+          return isHorizon
+        }
+      },
+      initialValue,
+    ).map((isHorizon) => {
+      logger.info(
+        isHorizon
+          ? `Network is Horizon ready`
+          : `Network is not Horizon ready`,
+      )
+      return isHorizon
+    })
+  }
+
   findEvent(
     eventType: string,
     contractInterface: Interface,