feat: purge stale transactions (#801)

Author: krobi64

README.md

@@ -126,7 +126,7 @@ Once node receives a claim, it stores the claim with some metadata including the
 * The highest block read at the time node stores the claim
 * Placeholders for the actual block that was mined including the claim
 
-This allows the node application to track whether or not the claim actually has been successfully saved to the Bitcoin blockchain. There is a configuration value, `maxBlockHeightDelta`, that determines how far ahead the blockchain will grow before resubmitting the claim. Comparing this value against the delta between the highest block read and the block read at the time of claim creation will determine whether node resubmits the claim.
+This allows the node application to track whether or not the claim actually has been successfully saved to the Bitcoin blockchain. There is a configuration value, `maximumTransactionAgeInBlocks`, that determines how far ahead the blockchain will grow before resubmitting the claim. Comparing this value against the delta between the highest block read and the block read at the time of claim creation will determine whether node resubmits the claim.
 
 
 ### Po.et JS

src/BlockchainWriter/BlockchainWriter.ts

@@ -107,8 +107,8 @@ const createContainer = (
     .toConstantValue(new BitcoinCore(bitcoinRPCConfigurationToBitcoinCoreArguments(configuration)))
   container.bind<ServiceConfiguration>('ServiceConfiguration').toConstantValue({
     anchorIntervalInSeconds: configuration.anchorIntervalInSeconds,
-    purgeStaleTransactionsInSeconds: configuration.purgeStaleTransactionsInSeconds,
-    maxBlockHeightDelta: configuration.maxBlockHeightDelta,
+    purgeStaleTransactionsIntervalInSeconds: configuration.purgeStaleTransactionsIntervalInSeconds,
+    maximumTransactionAgeInBlocks: configuration.maximumTransactionAgeInBlocks,
   })
   container.bind<ControllerConfiguration>('ClaimControllerConfiguration').toConstantValue(configuration)
   container.bind<ExchangeConfiguration>('ExchangeConfiguration').toConstantValue(configuration.exchanges)

src/BlockchainWriter/Controller.ts

@@ -15,6 +15,7 @@ import { ExchangeConfiguration } from './ExchangeConfiguration'
 export interface ControllerConfiguration {
   readonly poetNetwork: string
   readonly poetVersion: number
+  readonly maximumTransactionAgeInBlocks: number
 }
 
 export const convertLightBlockToEntry = (lightBlock: LightBlock): Entry => ({
@@ -91,6 +92,19 @@ export class Controller {
     await this.dao.updateAllByTransactionId(transactionIds, convertLightBlockToEntry(lightBlock))
   }
 
+  async purgeStaleTransactions(): Promise<void> {
+    const logger = this.logger.child({ method: 'purgeStaleTransactions' })
+    const { blocks } = await this.bitcoinCore.getBlockchainInfo()
+    logger.info(
+      {
+        blocks, maximumTransactionAgeInBlocks: this.configuration.maximumTransactionAgeInBlocks,
+      },
+      'Purging stale transactions',
+    )
+
+    await this.dao.purgeStaleTransactions(blocks - this.configuration.maximumTransactionAgeInBlocks)
+  }
+
   private async anchorIPFSDirectoryHash(ipfsDirectoryHash: string): Promise<void> {
     const { dao, messaging, anchorData, ipfsDirectoryHashToPoetAnchor } = this
     const logger = this.logger.child({ method: 'anchorIPFSDirectoryHash' })

src/BlockchainWriter/Router.ts

@@ -35,6 +35,7 @@ export class Router {
     await this.messaging.consume(this.exchange.anchorNextHashRequest, this.onAnchorNextHashRequest)
     await this.messaging.consume(this.exchange.batchWriterCreateNextBatchSuccess, this.onCreateBatchSuccess)
     await this.messaging.consumeBlockDownloaded(this.blockDownloadedConsumer)
+    await this.messaging.consume(this.exchange.purgeStaleTransactions, this.onPurgeStaleTransactions)
   }
 
   onAnchorNextHashRequest = async (message: any): Promise<void> => {
@@ -50,11 +51,23 @@ export class Router {
     }
   }
 
-  blockDownloadedConsumer = async (blockDownloaded: BlockDownloaded): Promise<void> => {
-    await this.claimController.setBlockInformationForTransactions(
+  blockDownloadedConsumer = async (blockDownloaded: BlockDownloaded): Promise<void> =>
+    this.claimController.setBlockInformationForTransactions(
       getTxnIds(blockDownloaded.poetBlockAnchors),
       blockDownloaded.block,
     )
+
+  onPurgeStaleTransactions = async (): Promise<void> => {
+    const logger = this.logger.child({ method: 'onPurgeStaleTransactions' })
+
+    try {
+      await this.claimController.purgeStaleTransactions()
+    } catch (error) {
+      logger.trace(
+        { error },
+        'Error encountered while purging stale transactions',
+      )
+    }
   }
 
   onCreateBatchSuccess = async (message: any): Promise<void> => {

src/BlockchainWriter/Service.ts

@@ -9,8 +9,8 @@ import { ExchangeConfiguration } from './ExchangeConfiguration'
 
 export interface ServiceConfiguration {
   readonly anchorIntervalInSeconds: number
-  readonly purgeStaleTransactionsInSeconds: number
-  readonly maxBlockHeightDelta: number
+  readonly purgeStaleTransactionsIntervalInSeconds: number
+  readonly maximumTransactionAgeInBlocks: number
 }
 
 @injectable()
@@ -33,7 +33,7 @@ export class Service {
     this.anchorNextHashInterval = new Interval(this.anchorNextHash, 1000 * configuration.anchorIntervalInSeconds)
     this.purgeStaleTransactionInterval = new Interval(
       this.purgeStaleTransactions,
-      1000 * configuration.purgeStaleTransactionsInSeconds,
+      1000 * configuration.purgeStaleTransactionsIntervalInSeconds,
     )
   }
 

src/Configuration.ts

@@ -33,8 +33,8 @@ export interface Configuration extends LoggingConfiguration, BitcoinRPCConfigura
 
   readonly enableAnchoring: boolean
   readonly anchorIntervalInSeconds: number
-  readonly purgeStaleTransactionsInSeconds: number
-  readonly maxBlockHeightDelta: number
+  readonly purgeStaleTransactionsIntervalInSeconds: number
+  readonly maximumTransactionAgeInBlocks: number
 
   readonly healthIntervalInSeconds: number
   readonly lowWalletBalanceInBitcoin: number
@@ -109,8 +109,8 @@ const defaultConfiguration: Configuration = {
 
   enableAnchoring: false,
   anchorIntervalInSeconds: 30,
-  purgeStaleTransactionsInSeconds: 600,
-  maxBlockHeightDelta: 25,
+  purgeStaleTransactionsIntervalInSeconds: 600,
+  maximumTransactionAgeInBlocks: 25,
 
   healthIntervalInSeconds: 30,
   lowWalletBalanceInBitcoin: 1,

src/app.ts

@@ -179,8 +179,8 @@ export async function app(localVars: any = {}) {
         poetNetwork: configuration.poetNetwork,
         poetVersion: configuration.poetVersion,
         anchorIntervalInSeconds: configuration.anchorIntervalInSeconds,
-        purgeStaleTransactionsInSeconds: configuration.purgeStaleTransactionsInSeconds,
-        maxBlockHeightDelta: configuration.maxBlockHeightDelta,
+        purgeStaleTransactionsIntervalInSeconds: configuration.purgeStaleTransactionsIntervalInSeconds,
+        maximumTransactionAgeInBlocks: configuration.maximumTransactionAgeInBlocks,
         bitcoinUrl: configuration.bitcoinUrl,
         bitcoinPort: configuration.bitcoinPort,
         bitcoinNetwork: configuration.bitcoinNetwork,

tests/functional/index.ts

@@ -1,3 +1,3 @@
 import './claim_with_data'
 import './endpoints'
-// import './transaction_timeout'
+import './transaction_timeout'

tests/functional/transaction_timeout.ts

@@ -1,12 +1,12 @@
 /* tslint:disable:no-relative-imports */
 import { configureCreateVerifiableClaim, getVerifiableClaimSigner, isSignedVerifiableClaim } from '@po.et/poet-js'
-import { allPass, is, isNil, lensPath, path, pipeP, view } from 'ramda'
+import { allPass, is, isNil, lensPath, not, path, pipe, pipeP, view } from 'ramda'
 import { describe } from 'riteway'
 
 import { app } from '../../src/app'
 import { issuer, privateKey } from '../helpers/Keys'
 import { ensureBitcoinBalance, bitcoindClients, resetBitcoinServers } from '../helpers/bitcoin'
-import { delay, runtimeId, createDatabase } from '../helpers/utils'
+import { delayInSeconds, runtimeId, createDatabase } from '../helpers/utils'
 import { getWork, postWork } from '../helpers/works'
 
 const PREFIX = `test-functional-nodeA-poet-${runtimeId()}`
@@ -21,7 +21,8 @@ const blockchainSettings = {
   BATCH_CREATION_INTERVAL_IN_SECONDS: 5,
   READ_DIRECTORY_INTERVAL_IN_SECONDS: 5,
   UPLOAD_CLAIM_INTERVAL_IN_SECONDS: 5,
-  TRANSACTION_MAX_AGE_IN_SECONDS: 60,
+  MAXIMUM_TRANSACTION_AGE_IN_BLOCKS: 1,
+  PURGE_STALE_TRANSACTIONS_INTERVAL_IN_SECONDS: 30,
 }
 
 const { configureSignVerifiableClaim } = getVerifiableClaimSigner()
@@ -32,7 +33,7 @@ const createClaim = pipeP(
   signVerifiableClaim,
 )
 
-const { btcdClientA }: any = bitcoindClients()
+const { btcdClientA, btcdClientB }: any = bitcoindClients()
 
 const blockHash = lensPath(['anchor', 'blockHash'])
 const blockHeight = lensPath(['anchor', 'blockHeight'])
@@ -41,13 +42,16 @@ const transactionId = lensPath(['anchor', 'transactionId'])
 const getTransactionId = view(transactionId)
 const getBlockHash = view(blockHash)
 const getBlockHeight = view(blockHeight)
+const isNotNil = pipe(isNil, not)
 
 const lengthIsGreaterThan0 = (s: string) => s.length > 0
 const hasValidTxId = allPass([is(String), lengthIsGreaterThan0])
 
 describe('Transaction timout will reset the transaction id for the claim', async assert => {
   await resetBitcoinServers()
-  await delay(5 * 1000)
+  await btcdClientB.addNode(btcdClientA.host, 'add')
+
+  await delayInSeconds(5)
 
   const db = await createDatabase(PREFIX)
   const server = await app({
@@ -62,18 +66,23 @@ describe('Transaction timout will reset the transaction id for the claim', async
 
   // Make sure node A has regtest coins to pay for transactions.
   await ensureBitcoinBalance(btcdClientA)
+  await btcdClientA.setNetworkActive(false)
 
   // Allow everything to finish starting.
-  await delay(5 * 1000)
+  await delayInSeconds(5)
+
   const claim = await createClaim({
     name: 'Author Name',
   })
 
   await postWorkToNode(claim)
 
   // Wait for a claim batch to be submitted to the blockchain.
-  await delay((blockchainSettings.ANCHOR_INTERVAL_IN_SECONDS +
-    blockchainSettings.BATCH_CREATION_INTERVAL_IN_SECONDS + 5) * 1000)
+  await delayInSeconds(
+    blockchainSettings.ANCHOR_INTERVAL_IN_SECONDS +
+    blockchainSettings.BATCH_CREATION_INTERVAL_IN_SECONDS +
+    5,
+  )
 
   const firstResponse = await getWorkFromNode(claim.id)
   const firstGet = await firstResponse.json()
@@ -100,7 +109,10 @@ describe('Transaction timout will reset the transaction id for the claim', async
     expected: true,
   })
 
-  await delay(blockchainSettings.TRANSACTION_MAX_AGE_IN_SECONDS * 1000 + 5)
+  await btcdClientB.generate(blockchainSettings.MAXIMUM_TRANSACTION_AGE_IN_BLOCKS + 1)
+  await btcdClientA.setNetworkActive(true)
+
+  await delayInSeconds((blockchainSettings.PURGE_STALE_TRANSACTIONS_INTERVAL_IN_SECONDS  + 5) * 2)
 
   const secondResponse = await getWorkFromNode(claim.id)
   const secondGet = await secondResponse.json()
@@ -120,6 +132,43 @@ describe('Transaction timout will reset the transaction id for the claim', async
     expected: true,
   })
 
+  await btcdClientA.generate(1)
+  await delayInSeconds(
+    blockchainSettings.BATCH_CREATION_INTERVAL_IN_SECONDS +
+      blockchainSettings.READ_DIRECTORY_INTERVAL_IN_SECONDS,
+  )
+
+  const thirdResponse = await getWorkFromNode(claim.id)
+  const thirdGet = await thirdResponse.json()
+  const thirdTxId = getTransactionId(thirdGet)
+
+  assert({
+    given: 'transaction mined',
+    should: 'store the block height with the claim',
+    actual: isNotNil(getBlockHeight(thirdGet)),
+    expected: true,
+  })
+
+  assert({
+    given: 'transaction mined',
+    should: 'store the block hash with the claim',
+    actual: isNotNil(getBlockHash(thirdGet)),
+    expected: true,
+  })
+
+  await delayInSeconds(blockchainSettings.PURGE_STALE_TRANSACTIONS_INTERVAL_IN_SECONDS + 5)
+
+  const fourthResponse = await getWorkFromNode(claim.id)
+  const fourthGet = await fourthResponse.json()
+  const fourthTxId = getTransactionId(fourthGet)
+
+  assert({
+    given: 'a claim with a block height and hash',
+    should: 'have the same transaction id',
+    actual: fourthTxId,
+    expected: thirdTxId,
+  })
+
   await server.stop()
   await db.teardown()
 })

tests/helpers/utils.ts

@@ -1,12 +1,14 @@
 /* tslint:disable:no-relative-imports */
 import { promisify } from 'util'
-
+import { secondsToMiliseconds } from '../../src/Helpers/Time'
+import { asyncPipe } from '../../src/Helpers/asyncPipe'
 import { app } from '../../src/app'
 import { dbHelper } from './database'
-
 export const runtimeId = () => `${process.pid}-${new Date().getMilliseconds()}-${Math.floor(Math.random() * 10)}`
 export const delay = promisify(setTimeout)
 
+export const delayInSeconds = asyncPipe(secondsToMiliseconds, delay)
+
 export const createDatabase = async (prefix: string) => {
   const db = dbHelper()