diff --git a/packages/indexer-common/src/__tests__/sequential-timer.test.ts b/packages/indexer-common/src/__tests__/sequential-timer.test.ts new file mode 100644 index 000000000..8a76f7221 --- /dev/null +++ b/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(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((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) + }) +}) diff --git a/packages/indexer-common/src/__tests__/subgraph.test.ts b/packages/indexer-common/src/__tests__/subgraph.test.ts index a5585f54a..a25927ae4 100644 --- a/packages/indexer-common/src/__tests__/subgraph.test.ts +++ b/packages/indexer-common/src/__tests__/subgraph.test.ts @@ -1,3 +1,5 @@ +import axios, { AxiosInstance } from 'axios' +import { createLogger, SubgraphDeploymentID } from '@graphprotocol/common-ts' import { DocumentNode, print } from 'graphql' import { SubgraphFreshnessChecker, @@ -5,7 +7,8 @@ import { 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' }) + }) +}) diff --git a/packages/indexer-common/src/sequential-timer.ts b/packages/indexer-common/src/sequential-timer.ts index 30e5f4583..2971ebfc9 100644 --- a/packages/indexer-common/src/sequential-timer.ts +++ b/packages/indexer-common/src/sequential-timer.ts @@ -40,14 +40,17 @@ function logWorkTime( } /** - * Create an eventual that performs the work in the Reducer 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 * @param initial U * @returns Eventual */ +// Keep T for callers that explicitly supply both type arguments. +// eslint-disable-next-line @typescript-eslint/no-unused-vars export function sequentialTimerReduce( { logger, milliseconds }: TimerTaskContext, reducer: Reducer, @@ -59,22 +62,17 @@ export function sequentialTimerReduce( 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( }, ) } else { - outputReduce(promiseOrT) + outputReduce(promiseOrU) logWorkTime(workStarted, logger, caller, milliseconds) setTimeout(work, milliseconds) }