chore: add telemetry for piece indexer

backend/bin/deal-observer-backend.js

@@ -8,7 +8,7 @@ import '../lib/instrument.js'
 import { createInflux } from '../lib/telemetry.js'
 import { getChainHead, rpcRequest } from '../lib/rpc-service/service.js'
 import { fetchDealWithHighestActivatedEpoch, countStoredActiveDeals, observeBuiltinActorEvents } from '../lib/deal-observer.js'
-import { indexPieces } from '../lib/piece-indexer.js'
+import { countStoredActiveDealsWithMissingPayloadCid, indexPieces } from '../lib/piece-indexer.js'
 import { findAndSubmitUnsubmittedDeals, submitDealsToSparkApi } from '../lib/spark-api-submit-deals.js'
 import { getDealPayloadCid } from '../lib/piece-indexer-service.js'
 
@@ -133,7 +133,14 @@ export const pieceIndexerLoop = async (makeRpcRequest, getDealPayloadCid, pgPool
     // Maximum number of deals to index in one loop iteration
     const maxDeals = 1000
     try {
-      await indexPieces(makeRpcRequest, getDealPayloadCid, pgPool, maxDeals)
+      const numOfMissingPayloadsCidsResolved = await indexPieces(makeRpcRequest, getDealPayloadCid, pgPool, maxDeals)
+      const numOfMissingPayloadsCids = await countStoredActiveDealsWithMissingPayloadCid(pgPool)
+      if (INFLUXDB_TOKEN) {
+        recordTelemetry('piece_indexer_stats', point => {
+          point.intField('total_missing_payload_cids', numOfMissingPayloadsCids)
+          point.intField('missing_payload_cids_resolved', numOfMissingPayloadsCidsResolved)
+        })
+      }
     } catch (e) {
       console.error(e)
       Sentry.captureException(e)

backend/lib/piece-indexer.js

@@ -12,17 +12,20 @@ import { getMinerPeerId } from './rpc-service/service.js'
  * @param {function} getDealPayloadCid
  * @param {Queryable} pgPool
  * @param {number} maxDeals
- * @returns {Promise<void>}
+ * @returns {Promise<number>}
  */
 export const indexPieces = async (makeRpcRequest, getDealPayloadCid, pgPool, maxDeals) => {
+  let missingPayloadCidsResolved = 0
   for (const deal of await fetchDealsWithNoPayloadCid(pgPool, maxDeals)) {
     const minerPeerId = await getMinerPeerId(deal.miner_id, makeRpcRequest)
     const payloadCid = await getDealPayloadCid(minerPeerId, deal.piece_cid)
     if (payloadCid) {
       deal.payload_cid = payloadCid
       await updatePayloadInActiveDeal(pgPool, deal)
+      missingPayloadCidsResolved++
     }
   }
+  return missingPayloadCidsResolved
 }
 
 /**
@@ -35,6 +38,12 @@ export async function fetchDealsWithNoPayloadCid (pgPool, maxDeals) {
   return await loadDeals(pgPool, query, [maxDeals])
 }
 
+export async function countStoredActiveDealsWithMissingPayloadCid (pgPool) {
+  const query = 'SELECT COUNT(*) FROM active_deals WHERE payload_cid IS NULL'
+  const result = await pgPool.query(query)
+  return result.rows[0].count
+}
+
 /**
  * @param {Queryable} pgPool
  * @param {Static<typeof ActiveDealDbEntry>} deal

backend/test/piece-indexer.test.js

@@ -7,7 +7,7 @@ import { observeBuiltinActorEvents } from '../lib/deal-observer.js'
 import assert from 'assert'
 import { minerPeerIds } from './test_data/minerInfo.js'
 import { payloadCIDs } from './test_data/payloadCIDs.js'
-import { indexPieces } from '../lib/piece-indexer.js'
+import { countStoredActiveDealsWithMissingPayloadCid, indexPieces } from '../lib/piece-indexer.js'
 
 describe('deal-observer-backend piece indexer', () => {
   const makeRpcRequest = async (method, params) => {
@@ -63,4 +63,19 @@ describe('deal-observer-backend piece indexer', () => {
       85 // Not all deals have a payload CID in the test data
     )
   })
+
+  it('piece indexer count number of missing payload CIDs', async () => {
+    let missingPayloadCids = await countStoredActiveDealsWithMissingPayloadCid(pgPool)
+    assert.strictEqual(missingPayloadCids, 336n)
+    const getDealPayloadCidCalls = []
+    const getDealPayloadCid = async (providerId, pieceCid) => {
+      getDealPayloadCidCalls.push({ providerId, pieceCid })
+      const payloadCid = payloadCIDs.get(JSON.stringify({ minerId: providerId, pieceCid }))
+      return payloadCid?.payloadCid
+    }
+
+    await indexPieces(makeRpcRequest, getDealPayloadCid, pgPool, 10000)
+    missingPayloadCids = await countStoredActiveDealsWithMissingPayloadCid(pgPool)
+    assert.strictEqual(missingPayloadCids, 85n)
+  })
 })