chore: extract deal observer loop

backend/bin/deal-observer-backend.js

@@ -1,12 +1,10 @@
 import { createPgPool } from '@filecoin-station/deal-observer-db'
 import * as Sentry from '@sentry/node'
-import timers from 'node:timers/promises'
-import slug from 'slug'
 import '../lib/instrument.js'
 import { createInflux } from '../lib/telemetry.js'
-import { getChainHead, rpcRequest } from '../lib/rpc-service/service.js'
-import { fetchDealWithHighestActivatedEpoch, observeBuiltinActorEvents } from '../lib/deal-observer.js'
+import { rpcRequest } from '../lib/rpc-service/service.js'
 import assert from 'node:assert'
+import { dealObserverLoop } from '../lib/deal-observer-loop.js'
 
 const { INFLUXDB_TOKEN } = process.env
 if (!INFLUXDB_TOKEN) {
@@ -19,42 +17,18 @@ const finalityEpochs = 940
 const maxPastEpochs = 1999
 assert(finalityEpochs <= maxPastEpochs)
 const pgPool = await createPgPool()
-
-const LOOP_NAME = 'Built-in actor events'
 const { recordTelemetry } = createInflux(INFLUXDB_TOKEN)
-
-const dealObserverLoop = async (makeRpcRequest, pgPool) => {
-  while (true) {
-    const start = Date.now()
-    try {
-      const currentChainHead = await getChainHead(makeRpcRequest)
-      const currentFinalizedChainHead = currentChainHead.Height - finalityEpochs
-      // If the storage is empty we start 2000 blocks into the past as that is the furthest we can go with the public glif rpc endpoints.
-      const lastInsertedDeal = await fetchDealWithHighestActivatedEpoch(pgPool)
-      const lastEpochStored = lastInsertedDeal ? lastInsertedDeal.activated_at_epoch : currentChainHead.Height - maxPastEpochs
-      for (let epoch = lastEpochStored + 1; epoch <= currentFinalizedChainHead; epoch++) {
-        await observeBuiltinActorEvents(epoch, pgPool, makeRpcRequest)
-      }
-    } catch (e) {
-      console.error(e)
-      Sentry.captureException(e)
-    }
-    const dt = Date.now() - start
-    console.log(`Loop "${LOOP_NAME}" took ${dt}ms`)
-
-    if (INFLUXDB_TOKEN) {
-      recordTelemetry(`loop_${slug(LOOP_NAME, '_')}`, point => {
-        point.intField('interval_ms', LOOP_INTERVAL)
-        point.intField('duration_ms', dt)
-      })
-    }
-    if (dt < LOOP_INTERVAL) {
-      await timers.setTimeout(LOOP_INTERVAL - dt)
-    }
-  }
-}
+const controller = new AbortController()
+const { signal } = controller
 
 await dealObserverLoop(
   rpcRequest,
-  pgPool
+  pgPool,
+  recordTelemetry,
+  Sentry,
+  maxPastEpochs,
+  finalityEpochs,
+  LOOP_INTERVAL,
+  INFLUXDB_TOKEN,
+  signal
 )

backend/lib/deal-observer-loop.js

@@ -0,0 +1,50 @@
+import { fetchDealWithHighestActivatedEpoch, observeBuiltinActorEvents } from './deal-observer.js'
+import { getChainHead } from './rpc-service/service.js'
+import timers from 'node:timers/promises'
+import slug from 'slug'
+/** @import {Queryable} from '@filecoin-station/deal-observer-db' */
+/** @import {Point} from '@influxdata/influxdb-client' */
+
+/**
+ * @param {(method:string,params:object) => object} makeRpcRequest
+ * @param {Queryable} pgPool
+ * @param {(name: string, fn: (p: Point) => void) => void} recordTelemetry
+ * @param {import("@sentry/node")} Sentry
+ * @param {number} maxPastEpochs
+ * @param {number} finalityEpochs
+ * @param {number} loopInterval
+ * @param {string | undefined} influxToken
+ * @returns {Promise<void>}
+ * */
+export const dealObserverLoop = async (makeRpcRequest, pgPool, recordTelemetry, Sentry, maxPastEpochs, finalityEpochs, loopInterval, influxToken, signal) => {
+  const LOOP_NAME = 'Built-in actor events'
+
+  while (!signal?.aborted) {
+    const start = Date.now()
+    try {
+      const currentChainHead = await getChainHead(makeRpcRequest)
+      const currentFinalizedChainHead = currentChainHead.Height - finalityEpochs
+      // If the storage is empty we start 2000 blocks into the past as that is the furthest we can go with the public glif rpc endpoints.
+      const lastInsertedDeal = await fetchDealWithHighestActivatedEpoch(pgPool)
+      const lastEpochStored = lastInsertedDeal ? lastInsertedDeal.activated_at_epoch : currentChainHead.Height - maxPastEpochs
+      for (let epoch = lastEpochStored + 1; epoch <= currentFinalizedChainHead; epoch++) {
+        await observeBuiltinActorEvents(epoch, pgPool, makeRpcRequest)
+      }
+    } catch (e) {
+      console.error(e)
+      Sentry.captureException(e)
+    }
+    const dt = Date.now() - start
+    console.log(`Loop "${LOOP_NAME}" took ${dt}ms`)
+
+    if (influxToken) {
+      recordTelemetry(`loop_${slug(LOOP_NAME, '_')}`, point => {
+        point.intField('interval_ms', loopInterval)
+        point.intField('duration_ms', dt)
+      })
+    }
+    if (dt < loopInterval) {
+      await timers.setTimeout(loopInterval - dt)
+    }
+  }
+}

backend/test/deal-observer-loop.test.js

@@ -0,0 +1,59 @@
+import { createPgPool, migrateWithPgClient } from '@filecoin-station/deal-observer-db'
+import { dealObserverLoop } from '../lib/deal-observer-loop.js'
+import { before, beforeEach, it, describe, after } from 'node:test'
+import { rawActorEventTestData } from './test_data/rawActorEvent.js'
+import { chainHeadTestData } from './test_data/chainHead.js'
+import { parse } from '@ipld/dag-json'
+
+describe('dealObserverLoop', () => {
+  let pgPool
+  const makeRpcRequest = async (method, params) => {
+    switch (method) {
+      case 'Filecoin.ChainHead':
+        return parse(JSON.stringify(chainHeadTestData))
+      case 'Filecoin.GetActorEventsRaw':
+        return parse(JSON.stringify(rawActorEventTestData)).filter(e => e.height >= params[0].fromHeight && e.height <= params[0].toHeight)
+      default:
+        throw new Error(`Unsupported RPC API method: "${method}"`)
+    }
+  }
+  before(async () => {
+    pgPool = await createPgPool()
+    await migrateWithPgClient(pgPool)
+  })
+
+  after(async () => {
+    await pgPool.end()
+  })
+
+  beforeEach(async () => {
+    await pgPool.query('DELETE FROM active_deals')
+  })
+
+  it('then deal observer loop fetches new active deals and stores them in storage', async (t) => {
+    const controller = new AbortController()
+    const { signal } = controller
+    const waitForDealCount = async (targetCount) => {
+      while (true) {
+        const { rows } = (await pgPool.query('SELECT COUNT(*) FROM active_deals'))
+        console.log('Current deal count:', rows[0].count)
+        if (parseInt(rows[0].count) === targetCount) break
+      }
+      controller.abort()
+    }
+    const failOnTimeout = async () => { await setTimeout(() => { if (!signal.aborted) { throw new Error('Test timed out') } }, 2000) }
+    await Promise.all([failOnTimeout(), waitForDealCount(360),
+      dealObserverLoop(
+        makeRpcRequest,
+        pgPool,
+        undefined,
+        undefined,
+        // The testdata has a total amount of 11 blocks
+        11,
+        0,
+        100,
+        undefined,
+        signal
+      )])
+  })
+})

backend/test/deal-observer.test.js

@@ -1,5 +1,5 @@
 import assert from 'node:assert'
-import { after, before, beforeEach, describe, it } from 'node:test'
+import { after, before, beforeEach, it, describe } from 'node:test'
 import { createPgPool, migrateWithPgClient } from '@filecoin-station/deal-observer-db'
 import { fetchDealWithHighestActivatedEpoch, storeActiveDeals } from '../lib/deal-observer.js'
 import { ActiveDealDbEntry } from '@filecoin-station/deal-observer-db/lib/types.js'

backend/test/rpc-client.test.js

@@ -1,11 +1,11 @@
 import assert from 'node:assert'
 import { describe, it } from 'node:test'
 import { chainHeadTestData } from './test_data/chainHead.js'
-import { rawActorEventTestData } from './test_data/rawActorEvent.js'
 import { parse } from '@ipld/dag-json'
 import { getActorEvents, getActorEventsFilter, getChainHead, rpcRequest } from '../lib/rpc-service/service.js'
 import { ClaimEvent } from '../lib/rpc-service/data-types.js'
 import { Value } from '@sinclair/typebox/value'
+import { rawActorEventTestData } from './test_data/rawActorEvent.js'
 
 describe('RpcApiClient', () => {
   const makeRpcRequest = async (method, params) => {
@@ -15,7 +15,7 @@ describe('RpcApiClient', () => {
       case 'Filecoin.GetActorEventsRaw':
         return parse(JSON.stringify(rawActorEventTestData)).filter(e => e.height >= params[0].fromHeight && e.height <= params[0].toHeight)
       default:
-        console.error('Unknown method')
+        throw new Error(`Unsupported RPC API method: "${method}"`)
     }
   }