fix(indexer-common): publish sequential timer reduction results

packages/indexer-common/src/__tests__/sequential-timer.test.ts

@@ -0,0 +1,89 @@
+import { createLogger } from '@graphprotocol/common-ts'
+import { sequentialTimerReduce } from '../sequential-timer'
+
+describe('sequentialTimerReduce', () => {
+  const logger = createLogger({
+    name: 'Sequential timer tests',
+    async: false,
+    level: 'error',
+  })
+  const options = { logger, milliseconds: 20 }
+
+  beforeEach(() => jest.useFakeTimers())
+  afterEach(() => {
+    jest.useRealTimers()
+    jest.restoreAllMocks()
+  })
+
+  it('publishes synchronous results and passes the latest value to the reducer', async () => {
+    const reducer = jest.fn((value: number) => value + 1)
+    const result = sequentialTimerReduce<number, number>(options, reducer, 0)
+    const observed: number[] = []
+    result.subscribe((value) => observed.push(value))
+
+    await expect(result.value()).resolves.toBe(1)
+    expect(observed).toEqual([1])
+
+    await jest.advanceTimersByTimeAsync(20)
+
+    await expect(result.value()).resolves.toBe(2)
+    expect(observed).toEqual([1, 2])
+    expect(reducer).toHaveBeenNthCalledWith(2, 1, expect.any(Number))
+  })
+
+  it('publishes changed asynchronous results without repeating equal values', async () => {
+    const initial = { synced: false, health: 'healthy' }
+    const reducer = jest
+      .fn()
+      .mockResolvedValueOnce({ synced: true, health: 'healthy' })
+      .mockResolvedValueOnce({ synced: true, health: 'healthy' })
+      .mockResolvedValueOnce({ synced: false, health: 'unhealthy' })
+    const result = sequentialTimerReduce(options, reducer, initial)
+    const observed: (typeof initial)[] = []
+    result.subscribe((value) => observed.push(value))
+
+    expect(observed).toEqual([initial])
+    await jest.advanceTimersByTimeAsync(0)
+    await expect(result.value()).resolves.toEqual({ synced: true, health: 'healthy' })
+
+    await jest.advanceTimersByTimeAsync(20)
+    expect(observed).toHaveLength(2)
+
+    await jest.advanceTimersByTimeAsync(20)
+    expect(observed).toEqual([
+      initial,
+      { synced: true, health: 'healthy' },
+      { synced: false, health: 'unhealthy' },
+    ])
+    await expect(result.value()).resolves.toEqual({ synced: false, health: 'unhealthy' })
+  })
+
+  it('waits for each result and retries after a rejected reduction', async () => {
+    let resolveFirst: (value: number) => void = () => undefined
+    const first = new Promise<number>((resolve) => {
+      resolveFirst = resolve
+    })
+    const reducer = jest
+      .fn()
+      .mockReturnValueOnce(first)
+      .mockRejectedValueOnce(new Error('temporary failure'))
+      .mockResolvedValueOnce(2)
+    const logError = jest.spyOn(console, 'error').mockImplementation(() => undefined)
+    const result = sequentialTimerReduce(options, reducer, 0)
+
+    await jest.advanceTimersByTimeAsync(40)
+    expect(reducer).toHaveBeenCalledTimes(1)
+
+    resolveFirst(1)
+    await jest.advanceTimersByTimeAsync(0)
+    await expect(result.value()).resolves.toBe(1)
+
+    await jest.advanceTimersByTimeAsync(20)
+    expect(logError).toHaveBeenCalledTimes(1)
+    await expect(result.value()).resolves.toBe(1)
+
+    await jest.advanceTimersByTimeAsync(20)
+    expect(reducer).toHaveBeenNthCalledWith(3, 1, expect.any(Number))
+    await expect(result.value()).resolves.toBe(2)
+  })
+})

packages/indexer-common/src/__tests__/subgraph.test.ts

@@ -1,11 +1,14 @@
+import axios, { AxiosInstance } from 'axios'
+import { createLogger, SubgraphDeploymentID } from '@graphprotocol/common-ts'
 import { DocumentNode, print } from 'graphql'
 import {
   SubgraphFreshnessChecker,
   LoggerInterface,
   ProviderInterface,
   SubgraphQueryInterface,
 } from '../subgraphs'
-import { QueryResult } from '../subgraph-client'
+import { QueryResult, SubgraphClient } from '../subgraph-client'
+import { GraphNode } from '../graph-node'
 import gql from 'graphql-tag'
 import { mergeSelectionSets } from '../utils'
 
@@ -256,3 +259,91 @@ describe('SubgraphFreshnessChecker', () => {
     })
   })
 })
+
+describe('SubgraphClient deployment monitoring', () => {
+  const logger = createLogger({
+    name: 'Subgraph client tests',
+    async: false,
+    level: 'error',
+  })
+  const deployment = new SubgraphDeploymentID(
+    'Qmd9nZKCH8UZU1pBzk7G8ECJr3jX3a2vAf3vowuTwFvrQg',
+  )
+  const unsynced = { synced: false, health: 'healthy', chains: [] }
+  const synced = { synced: true, health: 'healthy', chains: [] }
+  const unhealthy = { synced: true, health: 'unhealthy', chains: [] }
+  let indexingStatus: jest.Mock
+  let graphNode: GraphNode
+  let remotePost: jest.Mock
+  let localPost: jest.Mock
+
+  beforeEach(() => {
+    jest.useFakeTimers()
+    indexingStatus = jest.fn()
+    remotePost = jest.fn().mockResolvedValue({ data: 'remote' })
+    localPost = jest.fn().mockResolvedValue({ data: 'local' })
+    jest
+      .spyOn(axios, 'create')
+      .mockReturnValue({ post: remotePost } as unknown as AxiosInstance)
+    graphNode = {
+      indexingStatus,
+      getQueryClient: jest.fn().mockReturnValue({ post: localPost }),
+      getQueryEndpoint: jest.fn().mockReturnValue('http://local'),
+    } as unknown as GraphNode
+  })
+
+  afterEach(() => {
+    jest.useRealTimers()
+    jest.restoreAllMocks()
+  })
+
+  it('selects the local deployment when it syncs and retains its status on a failed poll', async () => {
+    indexingStatus
+      .mockResolvedValueOnce([unsynced])
+      .mockResolvedValueOnce([synced])
+      .mockRejectedValueOnce(new Error('status unavailable'))
+      .mockResolvedValueOnce([unhealthy])
+
+    const client = await SubgraphClient.create({
+      logger,
+      name: 'Test Subgraph',
+      endpoint: 'http://remote',
+      deployment: { graphNode, deployment },
+    })
+    await jest.advanceTimersByTimeAsync(0)
+
+    await expect(client.queryRaw('{}')).resolves.toEqual({ data: 'remote' })
+    await jest.advanceTimersByTimeAsync(60_000)
+    await expect(client.queryRaw('{}')).resolves.toEqual({ data: 'local' })
+
+    await jest.advanceTimersByTimeAsync(60_000)
+    await expect(client.queryRaw('{}')).resolves.toEqual({ data: 'local' })
+
+    await jest.advanceTimersByTimeAsync(60_000)
+    await expect(client.queryRaw('{}')).resolves.toEqual({ data: 'remote' })
+    expect(remotePost).toHaveBeenCalledTimes(2)
+    expect(localPost).toHaveBeenCalledTimes(2)
+  })
+
+  it('waits for a deployment-only client to sync', async () => {
+    indexingStatus.mockResolvedValueOnce([unsynced]).mockResolvedValueOnce([synced])
+
+    let created = false
+    const clientPromise = SubgraphClient.create({
+      logger,
+      name: 'Test Subgraph',
+      deployment: { graphNode, deployment },
+    }).then((client) => {
+      created = true
+      return client
+    })
+
+    await jest.advanceTimersByTimeAsync(0)
+    expect(created).toBe(false)
+
+    await jest.advanceTimersByTimeAsync(60_000)
+    const client = await clientPromise
+    expect(created).toBe(true)
+    await expect(client.queryRaw('{}')).resolves.toEqual({ data: 'local' })
+  })
+})

packages/indexer-common/src/sequential-timer.ts

@@ -40,14 +40,17 @@ function logWorkTime(
 }
 
 /**
- * Create an eventual that performs the work in the Reducer<number, U> function every `milliseconds` milliseconds.
- * The main difference between this and `timer(...).reduce(...)` is that this function will wait for the previous work to complete before starting the next one.
+ * Create an eventual that runs the reducer immediately, then waits `milliseconds` after each
+ * completed run before starting the next one. The eventual publishes changed results; reducers
+ * should treat the accumulator as immutable so changes can be detected.
  *
  * @param milliseconds number
  * @param reducer Reducer<number, U>
  * @param initial U
  * @returns Eventual<U>
  */
+// Keep T for callers that explicitly supply both type arguments.
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
 export function sequentialTimerReduce<T, U>(
   { logger, milliseconds }: TimerTaskContext,
   reducer: Reducer<number, U>,
@@ -59,22 +62,17 @@ export function sequentialTimerReduce<T, U>(
   const caller = stack?.split('\n')[2].trim()
 
   let acc: U = initial
-  let previousT: T | undefined
-  let latestT: T | undefined
 
   function outputReduce(value: U) {
-    previousT = latestT
     acc = value
-    if (!equal(latestT, previousT)) {
-      output.push(value)
-    }
+    output.push(value)
   }
 
   function work() {
     const workStarted = Date.now()
-    const promiseOrT = reducer(acc, workStarted)
-    if (isPromiseLike(promiseOrT)) {
-      promiseOrT.then(
+    const promiseOrU = reducer(acc, workStarted)
+    if (isPromiseLike(promiseOrU)) {
+      promiseOrU.then(
         function onfulfilled(value) {
           outputReduce(value)
           logWorkTime(workStarted, logger, caller, milliseconds)
@@ -87,7 +85,7 @@ export function sequentialTimerReduce<T, U>(
         },
       )
     } else {
-      outputReduce(promiseOrT)
+      outputReduce(promiseOrU)
       logWorkTime(workStarted, logger, caller, milliseconds)
       setTimeout(work, milliseconds)
     }