add cancel-stuck-transactions

bin/dry-run.js

@@ -142,7 +142,8 @@ const { ignoredErrors } = await evaluate({
   ieContractWithSigner,
   logger: console,
   recordTelemetry,
-  createPgClient
+  createPgClient,
+  stuckTransactionsCanceller: { addPending: () => {} }
 })
 
 console.log('Duration: %sms', Date.now() - started)

bin/fetch-recent-miner-measurements.js

@@ -216,7 +216,8 @@ async function processRound (roundIndex, measurementCids, resultCounts) {
     fetchRoundDetails,
     recordTelemetry,
     logger: { log: debug, error: debug },
-    ieContractWithSigner
+    ieContractWithSigner,
+    stuckTransactionsCanceller: { addPending: () => {} }
   })
 
   for (const m of round.measurements) {

bin/spark-evaluate.js

@@ -10,7 +10,7 @@ import { fetchMeasurements } from '../lib/preprocess.js'
 import { migrateWithPgConfig } from '../lib/migrate.js'
 import pg from 'pg'
 import { createMeridianContract } from '../lib/ie-contract.js'
-import { startCancelStuckTxs } from '../lib/cancel-stuck-txs.js'
+import { createStuckTransactionsCanceller, startCancelStuckTransactions } from '../lib/cancel-stuck-transactions.js'
 
 const {
   SENTRY_ENVIRONMENT = 'development',
@@ -43,6 +43,13 @@ const createPgClient = async () => {
   return pgClient
 }
 
+const pgClient = await createPgClient()
+const stuckTransactionsCanceller = createStuckTransactionsCanceller({
+  pgClient,
+  // Bypass NonceManager as we need to cancel transactions with the same nonce
+  signer: wallet
+})
+
 await Promise.all([
   startEvaluate({
     ieContract,
@@ -51,12 +58,8 @@ await Promise.all([
     fetchRoundDetails,
     recordTelemetry,
     createPgClient,
-    logger: console
+    logger: console,
+    stuckTransactionsCanceller
   }),
-  startCancelStuckTxs({
-    walletDelegatedAddress,
-    address: wallet.address,
-    // Bypass NonceManager as we need to cancel transactions with the same nonce
-    signer: wallet
-  })
+  startCancelStuckTransactions(stuckTransactionsCanceller)
 ])

index.js

@@ -19,7 +19,8 @@ export const startEvaluate = async ({
   fetchRoundDetails,
   recordTelemetry,
   createPgClient,
-  logger
+  logger,
+  stuckTransactionsCanceller
 }) => {
   assert(typeof createPgClient === 'function', 'createPgClient must be a function')
 
@@ -115,7 +116,8 @@ export const startEvaluate = async ({
       fetchRoundDetails,
       recordTelemetry,
       createPgClient,
-      logger
+      logger,
+      stuckTransactionsCanceller
     }).catch(err => {
       console.error('CANNOT EVALUATE ROUND %s:', evaluatedRoundIndex, err)
       Sentry.captureException(err, {

lib/cancel-stuck-transactions.js

@@ -0,0 +1,68 @@
+import * as Sentry from '@sentry/node'
+import { StuckTransactionsCanceller } from 'cancel-stuck-transactions'
+import ms from 'ms'
+import timers from 'node:timers/promises'
+
+const ROUND_LENGTH_MS = ms('20 minutes')
+const CHECK_STUCK_TXS_DELAY = ms('1 minute')
+
+export const createStuckTransactionsCanceller = ({ pgClient, signer }) => {
+  return new StuckTransactionsCanceller({
+    store: {
+      async set ({ hash, timestamp, from, maxPriorityFeePerGas, nonce }) {
+        await pgClient.query(`
+          INSERT INTO transactions_pending (hash, timestamp, from_address, max_priority_fee_per_gas, nonce)
+          VALUES ($1, $2, $3, $4, $5)
+          `,
+        [hash, timestamp, from, maxPriorityFeePerGas, nonce]
+        )
+      },
+      async list () {
+        const { rows } = await pgClient.query(`
+          SELECT
+            hash,
+            timestamp,
+            from_address as "from",
+            max_priority_fee_per_gas as "maxPriorityFeePerGas",
+            nonce
+          FROM transactions_pending
+        `)
+        return rows
+      },
+      async remove (hash) {
+        await pgClient.query(
+          'DELETE FROM transactions_pending WHERE hash = $1',
+          [hash]
+        )
+      }
+    },
+    log (str) {
+      console.log(str)
+    },
+    sendTransaction (tx) {
+      return signer.sendTransaction(tx)
+    }
+  })
+}
+
+export const startCancelStuckTransactions = async stuckTransactionsCanceller => {
+  while (true) {
+    try {
+      const res = await stuckTransactionsCanceller.cancelOlderThan(
+        2 * ROUND_LENGTH_MS
+      )
+      if (res !== undefined) {
+        for (const { status, reason } of res) {
+          if (status === 'rejected') {
+            console.error('Failed to cancel transaction:', reason)
+            Sentry.captureException(reason)
+          }
+        }
+      }
+    } catch (err) {
+      console.error(err)
+      Sentry.captureException(err)
+    }
+    await timers.setTimeout(CHECK_STUCK_TXS_DELAY)
+  }
+}

lib/evaluate.js

@@ -55,6 +55,7 @@ export const createSetScoresBuckets = participants => {
  * @param {import('./typings.js').RecordTelemetryFn} args.recordTelemetry
  * @param {import('./typings.js').CreatePgClient} [args.createPgClient]
  * @param {Pick<Console, 'log' | 'error'>} args.logger
+ * @param {import('cancel-stuck-transactions').StuckTransactionsCanceller} args.stuckTransactionsCanceller
  */
 export const evaluate = async ({
   round,
@@ -64,7 +65,8 @@ export const evaluate = async ({
   fetchRoundDetails,
   recordTelemetry,
   createPgClient,
-  logger
+  logger,
+  stuckTransactionsCanceller
 }) => {
   requiredCommitteeSize ??= REQUIRED_COMMITTEE_SIZE
 
@@ -161,6 +163,11 @@ export const evaluate = async ({
         Number(bucketIndex) + 1,
         buckets.length
       )
+      await stuckTransactionsCanceller.addPending(tx)
+      ;(async () => {
+        await tx.wait()
+        stuckTransactionsCanceller.removeConfirmed(tx)
+      })().catch(console.error)
     } catch (err) {
       logger.error('CANNOT SUBMIT SCORES FOR ROUND %s (CALL %s/%s): %s',
         roundIndex,

migrations/017.do.cancel-stuck-transactions.sql

@@ -0,0 +1,7 @@
+CREATE TABLE transactions_pending (
+  hash TEXT NOT NULL PRIMARY KEY,
+  timestamp TIMESTAMPTZ NOT NULL,
+  from_address TEXT NOT NULL,
+  max_priority_fee_per_gas BIGINT NOT NULL,
+  nonce INT NOT NULL
+);