Skip to content

Commit 2bc01da

Browse files
dwernerdwerner
committed
common, agent: switch most timer calls to sequential impl
1 parent 2e336b2 commit 2bc01da

File tree

12 files changed

+413
-209
lines changed

12 files changed

+413
-209
lines changed

‎packages/indexer-agent/src/agent.ts

Lines changed: 69 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import {
3737
networkIsL1,
3838
DeploymentManagementMode,
3939
SubgraphStatus,
40+
sequentialTimerMap,
4041
} from '@graphprotocol/indexer-common'
4142

4243
import PQueue from 'p-queue'
@@ -253,40 +254,41 @@ export class Agent {
253254
const requestIntervalSmall = this.pollingInterval
254255
const requestIntervalLarge = this.pollingInterval * 5
255256
const logger = this.logger.child({ component: 'ReconciliationLoop' })
256-
const currentEpochNumber: Eventual<NetworkMapped<number>> = timer(
257-
requestIntervalLarge,
258-
).tryMap(
259-
async () =>
260-
await this.multiNetworks.map(({ network }) => {
261-
logger.trace('Fetching current epoch number', {
262-
protocolNetwork: network.specification.networkIdentifier,
263-
})
264-
return network.networkMonitor.currentEpochNumber()
265-
}),
266-
{
267-
onError: error =>
268-
logger.warn(`Failed to fetch current epoch`, { error }),
269-
},
270-
)
257+
const currentEpochNumber: Eventual<NetworkMapped<number>> =
258+
sequentialTimerMap(
259+
{ logger, milliseconds: requestIntervalLarge },
260+
async () =>
261+
await this.multiNetworks.map(({ network }) => {
262+
logger.trace('Fetching current epoch number', {
263+
protocolNetwork: network.specification.networkIdentifier,
264+
})
265+
return network.networkMonitor.currentEpochNumber()
266+
}),
267+
{
268+
onError: error =>
269+
logger.warn(`Failed to fetch current epoch`, { error }),
270+
},
271+
)
271272

272-
const maxAllocationEpochs: Eventual<NetworkMapped<number>> = timer(
273-
requestIntervalLarge,
274-
).tryMap(
275-
() =>
276-
this.multiNetworks.map(({ network }) => {
277-
logger.trace('Fetching max allocation epochs', {
278-
protocolNetwork: network.specification.networkIdentifier,
279-
})
280-
return network.contracts.staking.maxAllocationEpochs()
281-
}),
282-
{
283-
onError: error =>
284-
logger.warn(`Failed to fetch max allocation epochs`, { error }),
285-
},
286-
)
273+
const maxAllocationEpochs: Eventual<NetworkMapped<number>> =
274+
sequentialTimerMap(
275+
{ logger, milliseconds: requestIntervalLarge },
276+
() =>
277+
this.multiNetworks.map(({ network }) => {
278+
logger.trace('Fetching max allocation epochs', {
279+
protocolNetwork: network.specification.networkIdentifier,
280+
})
281+
return network.contracts.staking.maxAllocationEpochs()
282+
}),
283+
{
284+
onError: error =>
285+
logger.warn(`Failed to fetch max allocation epochs`, { error }),
286+
},
287+
)
287288

288289
const indexingRules: Eventual<NetworkMapped<IndexingRuleAttributes[]>> =
289-
timer(requestIntervalSmall).tryMap(
290+
sequentialTimerMap(
291+
{ logger, milliseconds: requestIntervalSmall },
290292
async () => {
291293
return this.multiNetworks.map(async ({ network, operator }) => {
292294
logger.trace('Fetching indexing rules', {
@@ -322,24 +324,25 @@ export class Agent {
322324
},
323325
)
324326

325-
const activeDeployments: Eventual<SubgraphDeploymentID[]> = timer(
326-
requestIntervalSmall,
327-
).tryMap(
328-
() => {
329-
logger.trace('Fetching active deployments')
330-
return this.graphNode.subgraphDeployments()
331-
},
332-
{
333-
onError: error =>
334-
logger.warn(
335-
`Failed to obtain active deployments, trying again later`,
336-
{ error },
337-
),
338-
},
339-
)
327+
const activeDeployments: Eventual<SubgraphDeploymentID[]> =
328+
sequentialTimerMap(
329+
{ logger, milliseconds: requestIntervalSmall },
330+
() => {
331+
logger.trace('Fetching active deployments')
332+
return this.graphNode.subgraphDeployments()
333+
},
334+
{
335+
onError: error =>
336+
logger.warn(
337+
`Failed to obtain active deployments, trying again later`,
338+
{ error },
339+
),
340+
},
341+
)
340342

341343
const networkDeployments: Eventual<NetworkMapped<SubgraphDeployment[]>> =
342-
timer(requestIntervalSmall).tryMap(
344+
sequentialTimerMap(
345+
{ logger, milliseconds: requestIntervalSmall },
343346
async () =>
344347
await this.multiNetworks.map(({ network }) => {
345348
logger.trace('Fetching network deployments', {
@@ -358,7 +361,8 @@ export class Agent {
358361

359362
const eligibleTransferDeployments: Eventual<
360363
NetworkMapped<TransferredSubgraphDeployment[]>
361-
> = timer(requestIntervalLarge).tryMap(
364+
> = sequentialTimerMap(
365+
{ logger, milliseconds: requestIntervalLarge },
362366
async () => {
363367
// Return early if the auto migration feature is disabled.
364368
if (!this.autoMigrationSupport) {
@@ -558,23 +562,23 @@ export class Agent {
558562
},
559563
)
560564

561-
const activeAllocations: Eventual<NetworkMapped<Allocation[]>> = timer(
562-
requestIntervalSmall,
563-
).tryMap(
564-
() =>
565-
this.multiNetworks.map(({ network }) => {
566-
logger.trace('Fetching active allocations', {
567-
protocolNetwork: network.specification.networkIdentifier,
568-
})
569-
return network.networkMonitor.allocations(AllocationStatus.ACTIVE)
570-
}),
571-
{
572-
onError: () =>
573-
logger.warn(
574-
`Failed to obtain active allocations, trying again later`,
575-
),
576-
},
577-
)
565+
const activeAllocations: Eventual<NetworkMapped<Allocation[]>> =
566+
sequentialTimerMap(
567+
{ logger, milliseconds: requestIntervalSmall },
568+
() =>
569+
this.multiNetworks.map(({ network }) => {
570+
logger.trace('Fetching active allocations', {
571+
protocolNetwork: network.specification.networkIdentifier,
572+
})
573+
return network.networkMonitor.allocations(AllocationStatus.ACTIVE)
574+
}),
575+
{
576+
onError: () =>
577+
logger.warn(
578+
`Failed to obtain active allocations, trying again later`,
579+
),
580+
},
581+
)
578582

579583
// `activeAllocations` is used to trigger this Eventual, but not really needed
580584
// inside.

‎packages/indexer-common/src/allocations/monitor.ts

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,13 @@ import {
22
indexerError,
33
IndexerErrorCode,
44
parseGraphQLAllocation,
5+
sequentialTimerReduce,
56
} from '@graphprotocol/indexer-common'
67
import { Allocation, MonitorEligibleAllocationsOptions } from './types'
78

89
import gql from 'graphql-tag'
910

10-
import { Eventual, timer } from '@graphprotocol/common-ts'
11+
import { Eventual } from '@graphprotocol/common-ts'
1112

1213
export const monitorEligibleAllocations = ({
1314
indexer,
@@ -168,7 +169,14 @@ export const monitorEligibleAllocations = ({
168169
}
169170
}
170171

171-
const allocations = timer(interval).reduce(refreshAllocations, [])
172+
const allocations = sequentialTimerReduce(
173+
{
174+
logger,
175+
milliseconds: interval,
176+
},
177+
refreshAllocations,
178+
[],
179+
)
172180

173181
allocations.pipe((allocations) => {
174182
logger.info(`Eligible allocations`, {

‎packages/indexer-common/src/allocations/query-fees.ts

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ import { Counter, Gauge, Histogram } from 'prom-client'
22
import axios from 'axios'
33
import {
44
Logger,
5-
timer,
65
BytesWriter,
76
toAddress,
87
formatGRT,
@@ -20,6 +19,7 @@ import {
2019
ensureAllocationSummary,
2120
TransactionManager,
2221
specification as spec,
22+
sequentialTimerMap,
2323
} from '..'
2424
import { DHeap } from '@thi.ng/heaps'
2525
import { BigNumber, BigNumberish, Contract } from 'ethers'
@@ -264,7 +264,7 @@ export class AllocationReceiptCollector implements ReceiptCollector {
264264
}
265265

266266
// Check if there's another batch of receipts to collect every 10s
267-
timer(10_000).pipe(async () => {
267+
sequentialTimerMap({ logger: this.logger, milliseconds: 10_000 }, async () => {
268268
while (hasReceiptsReadyForCollecting()) {
269269
// Remove the batch from the processing queue
270270
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -283,7 +283,7 @@ export class AllocationReceiptCollector implements ReceiptCollector {
283283
}
284284

285285
private startVoucherProcessing() {
286-
timer(30_000).pipe(async () => {
286+
sequentialTimerMap({ logger: this.logger, milliseconds: 30_000 }, async () => {
287287
let pendingVouchers: Voucher[] = []
288288
try {
289289
pendingVouchers = await this.pendingVouchers() // Ordered by value

‎packages/indexer-common/src/allocations/tap-collector.ts

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
11
import { Counter, Gauge, Histogram } from 'prom-client'
22
import {
33
Logger,
4-
timer,
54
toAddress,
65
formatGRT,
76
Address,
87
Metrics,
98
Eventual,
10-
join as joinEventual,
119
} from '@graphprotocol/common-ts'
1210
import { NetworkContracts as TapContracts } from '@semiotic-labs/tap-contracts-bindings'
1311
import {
@@ -23,6 +21,7 @@ import {
2321
allocationSigner,
2422
tapAllocationIdProof,
2523
parseGraphQLAllocation,
24+
sequentialTimerMap,
2625
} from '..'
2726
import { BigNumber } from 'ethers'
2827
import pReduce from 'p-reduce'
@@ -184,9 +183,11 @@ export class TapCollector {
184183
}
185184

186185
private getPendingRAVs(): Eventual<RavWithAllocation[]> {
187-
return joinEventual({
188-
timer: timer(RAV_CHECK_INTERVAL_MS),
189-
}).tryMap(
186+
return sequentialTimerMap(
187+
{
188+
logger: this.logger,
189+
milliseconds: RAV_CHECK_INTERVAL_MS,
190+
},
190191
async () => {
191192
let ravs = await this.pendingRAVs()
192193
if (ravs.length === 0) {

‎packages/indexer-common/src/index.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,3 +17,4 @@ export * from './utils'
1717
export * from './parsers'
1818
export * as specification from './network-specification'
1919
export * from './multi-networks'
20+
export * from './sequential-timer'

‎packages/indexer-common/src/indexer-management/actions.ts

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,11 @@ import {
1818
Network,
1919
OrderDirection,
2020
GraphNode,
21+
sequentialTimerMap,
2122
} from '@graphprotocol/indexer-common'
2223

2324
import { Order, Transaction } from 'sequelize'
24-
import { Eventual, join, Logger, timer } from '@graphprotocol/common-ts'
25+
import { Eventual, join, Logger } from '@graphprotocol/common-ts'
2526
import groupBy from 'lodash.groupby'
2627

2728
export class ActionManager {
@@ -116,7 +117,11 @@ export class ActionManager {
116117

117118
async monitorQueue(): Promise<void> {
118119
const logger = this.logger.child({ component: 'QueueMonitor' })
119-
const approvedActions: Eventual<Action[]> = timer(30_000).tryMap(
120+
const approvedActions: Eventual<Action[]> = sequentialTimerMap(
121+
{
122+
logger,
123+
milliseconds: 30_000,
124+
},
120125
async () => {
121126
logger.trace('Fetching approved actions')
122127
let actions: Action[] = []

0 commit comments

Comments
 (0)