Loop error handling and transition to single-table approach

Author: PatrickNercessian

migrations/001.do.daily-reward-transfers.sql

@@ -0,0 +1 @@
+SELECT now();

migrations/001.do.sql

@@ -0,0 +1,10 @@
+CREATE TABLE daily_reward_transfers (
+    day DATE NOT NULL,
+    to_address TEXT NOT NULL,
+    amount NUMERIC NOT NULL,
+    last_checked_block INTEGER NOT NULL,
+    PRIMARY KEY (day, to_address)
+);
+
+CREATE INDEX daily_reward_transfers_day ON daily_reward_transfers (day);
+CREATE INDEX daily_reward_transfers_last_block ON daily_reward_transfers (last_checked_block DESC);

migrations/002.do.daily-reward-transfers.sql

@@ -15,9 +15,6 @@ const provider = new ethers.JsonRpcProvider(fetchRequest, null, { polling: true
 const ieContract = new ethers.Contract(SparkImpactEvaluator.ADDRESS, SparkImpactEvaluator.ABI, provider)
 
 await pgPool.query('DELETE FROM daily_reward_transfers')
-await pgPool.query('DELETE FROM reward_transfer_last_block')
-// Set the last block to -800 to simulate the observer starting from the beginning
-await pgPool.query('INSERT INTO reward_transfer_last_block (last_block) VALUES (-800)')
 
 await observeTransferEvents(pgPool, ieContract, provider)
 

observer/bin/dry-run.js

@@ -1,5 +1,6 @@
 import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
 import { ethers } from 'ethers'
+import * as Sentry from '@sentry/node'
 
 import { RPC_URL, rpcHeaders, OBSERVATION_INTERVAL_MS } from '../lib/config.js'
 import { getPgPool } from '../lib/db.js'
@@ -15,6 +16,11 @@ const ieContract = new ethers.Contract(SparkImpactEvaluator.ADDRESS, SparkImpact
 
 // Listen for Transfer events from the IE contract
 while (true) {
-  observeTransferEvents(pgPool, ieContract, provider)
+  try {
+    await observeTransferEvents(pgPool, ieContract, provider)
+  } catch (e) {
+    console.error(e)
+    Sentry.captureException(e)
+  }
   await new Promise(resolve => setTimeout(resolve, OBSERVATION_INTERVAL_MS))
 }

observer/bin/spark-observer.js

@@ -7,10 +7,9 @@ import { updateDailyTransferStats } from './platform-stats.js'
  * @param {import('ethers').Provider} provider
  */
 export const observeTransferEvents = async (pgPool, ieContract, provider) => {
-  // Get the last checked block. Even though there should be only one row, use MAX just to be safe
   const lastCheckedBlock = await pgPool.query(
-    'SELECT MAX(last_block) AS last_block FROM reward_transfer_last_block'
-  ).then(res => res.rows[0].last_block)
+    'SELECT MAX(last_checked_block) FROM daily_reward_transfers'
+  ).then(res => res.rows[0].last_checked_block)
 
   console.log('Querying impact evaluator Transfer events after block', lastCheckedBlock)
   let events
@@ -19,29 +18,21 @@ export const observeTransferEvents = async (pgPool, ieContract, provider) => {
   } catch (error) {
     console.error('Error querying impact evaluator Transfer events', error)
     if (error.message.includes('bad tipset height')) {
-      console.log('Block number too old, GLIF only provides last 2000 blocks, querying from there')
-      events = await ieContract.queryFilter(ieContract.filters.Transfer(), -1999)
+      console.log('Block number too old, GLIF only provides last 2000 blocks, querying from -1900')
+      events = await ieContract.queryFilter(ieContract.filters.Transfer(), -1900)
     } else {
       throw error
     }
   }
+  const currentBlockNumber = await provider.getBlockNumber()
+  console.log('Current block number:', currentBlockNumber)
   console.log(`Found ${events.length} Transfer events`)
   for (const event of events) {
     const transferEvent = {
       to_address: event.args.to,
       amount: event.args.amount
     }
     console.log('Transfer event:', transferEvent)
-    await updateDailyTransferStats(pgPool, transferEvent)
+    await updateDailyTransferStats(pgPool, transferEvent, currentBlockNumber)
   }
-
-  // Get the current block number and update the last_block in reward_transfer_last_block table
-  // For safety, only update if the new block number is greater than the existing one
-  const blockNumber = await provider.getBlockNumber()
-  console.log('Current block number:', blockNumber)
-  await pgPool.query(`
-    UPDATE reward_transfer_last_block
-    SET last_block = $1
-    WHERE $1 > last_block
-  `, [blockNumber])
 }

observer/lib/observer.js

@@ -3,13 +3,14 @@
  * @param {Object} transferEvent
  * @param {string} transferEvent.to_address
  * @param {number} transferEvent.amount
- * @param {number} transferEvent.blockNumber
+ * @param {number} currentBlockNumber
  */
-export const updateDailyTransferStats = async (pgClient, transferEvent) => {
+export const updateDailyTransferStats = async (pgClient, transferEvent, currentBlockNumber) => {
   await pgClient.query(`
-    INSERT INTO daily_reward_transfers (day, to_address, amount)
-    VALUES (now(), $1, $2)
-    ON CONFLICT (day, to_address) DO UPDATE
-    SET amount = daily_reward_transfers.amount + EXCLUDED.amount
-  `, [transferEvent.to_address, transferEvent.amount])
+    INSERT INTO daily_reward_transfers (day, to_address, amount, last_checked_block)
+    VALUES (now(), $1, $2, $3)
+    ON CONFLICT (day, to_address) DO UPDATE SET
+      amount = daily_reward_transfers.amount + EXCLUDED.amount,
+      last_checked_block = EXCLUDED.last_checked_block
+  `, [transferEvent.to_address, transferEvent.amount, currentBlockNumber])
 }

observer/lib/platform-stats.js

@@ -36,8 +36,8 @@ describe('platform-stats-generator', () => {
 
   describe('updateDailyTransferStats', () => {
     it('should correctly update daily Transfer stats with new transfer events', async () => {
-      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 100, blockNumber: 1 })
-      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 200, blockNumber: 1 })
+      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 100 }, 1)
+      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 200 }, 1)
 
       const { rows } = await pgPool.query(`
         SELECT day::TEXT, to_address, amount FROM daily_reward_transfers
@@ -48,18 +48,18 @@ describe('platform-stats-generator', () => {
     })
 
     it('should handle multiple addresses in daily Transfer stats', async () => {
-      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 50, blockNumber: 1 })
-      await updateDailyTransferStats(pgPool, { to_address: 'address2', amount: 150, blockNumber: 1 })
+      await updateDailyTransferStats(pgPool, { to_address: 'address1', amount: 50 }, 1)
+      await updateDailyTransferStats(pgPool, { to_address: 'address2', amount: 150 }, 1)
 
       const { rows } = await pgPool.query(`
-        SELECT day::TEXT, to_address, amount FROM daily_reward_transfers
+        SELECT day::TEXT, to_address, amount, last_checked_block FROM daily_reward_transfers
         ORDER BY to_address
       `)
       assert.strictEqual(rows.length, 2)
 
       assert.deepStrictEqual(rows, [
-        { day: today, to_address: 'address1', amount: '50' },
-        { day: today, to_address: 'address2', amount: '150' }
+        { day: today, to_address: 'address1', amount: '50', last_checked_block: 1 },
+        { day: today, to_address: 'address2', amount: '150', last_checked_block: 1 }
       ])
     })
   })

observer/test/platform-stats.test.js

@@ -163,17 +163,17 @@ describe('Platform Routes HTTP request handler', () => {
   describe('GET /transfers/daily', () => {
     it('returns daily total Rewards sent for the given date range', async () => {
       await givenDailyRewardTransferMetrics(pgPoolStatsDb, '2024-01-10', [
-        { to_address: 'to1', amount: 100 }
+        { to_address: 'to1', amount: 100, last_checked_block: 1 }
       ])
       await givenDailyRewardTransferMetrics(pgPoolStatsDb, '2024-01-11', [
-        { to_address: 'to2', amount: 150 }
+        { to_address: 'to2', amount: 150, last_checked_block: 1 }
       ])
       await givenDailyRewardTransferMetrics(pgPoolStatsDb, '2024-01-12', [
-        { to_address: 'to2', amount: 300 },
-        { to_address: 'to3', amount: 250 }
+        { to_address: 'to2', amount: 300, last_checked_block: 1 },
+        { to_address: 'to3', amount: 250, last_checked_block: 1 }
       ])
       await givenDailyRewardTransferMetrics(pgPoolStatsDb, '2024-01-13', [
-        { to_address: 'to1', amount: 100 }
+        { to_address: 'to1', amount: 100, last_checked_block: 1 }
       ])
 
       const res = await fetch(
@@ -208,12 +208,13 @@ const givenDailyStationMetrics = async (pgPoolEvaluateDb, day, stationStats) =>
 
 const givenDailyRewardTransferMetrics = async (pgPoolStatsDb, day, transferStats) => {
   await pgPoolStatsDb.query(`
-    INSERT INTO daily_reward_transfers (day, to_address, amount)
-    SELECT $1 AS day, UNNEST($2::text[]) AS to_address, UNNEST($3::int[]) AS amount
+    INSERT INTO daily_reward_transfers (day, to_address, amount, last_checked_block)
+    SELECT $1 AS day, UNNEST($2::text[]) AS to_address, UNNEST($3::int[]) AS amount, UNNEST($4::int[]) AS last_checked_block
     ON CONFLICT DO NOTHING
     `, [
     day,
     transferStats.map(s => s.to_address),
-    transferStats.map(s => s.amount)
+    transferStats.map(s => s.amount),
+    transferStats.map(s => s.last_checked_block)
   ])
 }