feat: StorageWriter retries (#396)

Author: kennylavender

src/Configuration.ts

@@ -42,6 +42,9 @@ export interface Configuration extends LoggingConfiguration, BitcoinRPCConfigura
   readonly batchCreationIntervalInSeconds: number
 
   readonly readDirectoryIntervalInSeconds: number
+
+  readonly uploadClaimIntervalInSeconds: number
+  readonly uploadClaimMaxAttempts: number
 }
 
 export interface LoggingConfiguration {
@@ -67,6 +70,7 @@ export interface ExchangeConfiguration {
   readonly exchangeIpfsHashTxId: string
   readonly exchangePoetAnchorDownloaded: string
   readonly exchangeClaimsDownloaded: string
+  readonly exchangeStorageWriterStoreNextClaim: string
 }
 
 const defaultConfiguration: Configuration = {
@@ -106,6 +110,9 @@ const defaultConfiguration: Configuration = {
 
   readDirectoryIntervalInSeconds: 30,
 
+  uploadClaimIntervalInSeconds: 30,
+  uploadClaimMaxAttempts: 10,
+
   forceBlockHeight: undefined,
 
   exchangeBatchReaderReadNextDirectoryRequest: 'BATCH_READER::READ_NEXT_DIRECTORY_REQUEST',
@@ -117,6 +124,7 @@ const defaultConfiguration: Configuration = {
   exchangeIpfsHashTxId: 'IPFS_HASH_TX_ID',
   exchangePoetAnchorDownloaded: 'POET_ANCHOR_DOWNLOADED',
   exchangeClaimsDownloaded: 'CLAIMS_DOWNLOADED',
+  exchangeStorageWriterStoreNextClaim: 'STORAGE_WRITER::STORE_NEXT_CLAIM',
 }
 
 export const configurationPath = () => path.join(homedir(), '/.po.et/configuration.json')

src/StorageWriter/ClaimController.test.ts

@@ -1,54 +1,85 @@
 import { Claim } from '@po.et/poet-js'
 import { inject, injectable } from 'inversify'
-import { Collection, Db } from 'mongodb'
 import * as Pino from 'pino'
+import { pipeP } from 'ramda'
 
 import { childWithFileName } from 'Helpers/Logging'
-import { Messaging } from 'Messaging/Messaging'
 
-import { ExchangeConfiguration } from './ExchangeConfiguration'
+import { DAOClaims } from './DAOClaims'
 import { IPFS } from './IPFS'
 
+enum LogTypes {
+  info = 'info',
+  trace = 'trace',
+  error = 'error',
+}
+
+interface StoreNextClaimData {
+  readonly claim: Claim
+  readonly ipfsFileHash?: string
+}
+
 @injectable()
 export class ClaimController {
   private readonly logger: Pino.Logger
-  private readonly db: Db
-  private readonly collection: Collection
-  private readonly messaging: Messaging
+  private readonly daoClaims: DAOClaims
   private readonly ipfs: IPFS
-  private readonly exchange: ExchangeConfiguration
 
   constructor(
     @inject('Logger') logger: Pino.Logger,
-    @inject('DB') db: Db,
-    @inject('Messaging') messaging: Messaging,
-    @inject('IPFS') ipfs: IPFS,
-    @inject('ExchangeConfiguration') exchange: ExchangeConfiguration
+    @inject('DAOClaims') daoClaims: DAOClaims,
+    @inject('IPFS') ipfs: IPFS
   ) {
     this.logger = childWithFileName(logger, __filename)
-    this.db = db
-    this.collection = this.db.collection('storageWriter')
-    this.messaging = messaging
+    this.daoClaims = daoClaims
     this.ipfs = ipfs
-    this.exchange = exchange
   }
 
-  async create(claim: Claim): Promise<void> {
+  private readonly log = (level: LogTypes) => (message: string) => async (value: any) => {
+    const logger = this.logger
+    logger[level]({ value }, message)
+    return value
+  }
+
+  public readonly create = async (claim: Claim): Promise<void> => {
     const logger = this.logger.child({ method: 'create' })
 
-    logger.trace({ claim }, 'Storing Claim')
+    logger.trace({ claim }, 'Adding Claim')
+
+    await this.daoClaims.addClaim(claim)
 
-    const ipfsFileHash = await this.ipfs.addText(JSON.stringify(claim))
+    logger.trace({ claim }, 'Added Claim')
+  }
+
+  private readonly findNextClaim = async (): Promise<StoreNextClaimData> => {
+    const claim = await this.daoClaims.findNextClaim()
+    return { claim }
+  }
 
-    logger.info({ claim, ipfsFileHash }, 'Claim Stored')
+  private readonly uploadClaim = (claim: Claim) => this.ipfs.addText(JSON.stringify(claim))
 
-    await this.collection.insertOne({
-      claimId: claim.id,
-      ipfsFileHash,
-    })
-    await this.messaging.publish(this.exchange.claimIpfsHash, {
-      claimId: claim.id,
+  private readonly storeClaim = async (data: StoreNextClaimData): Promise<StoreNextClaimData> => {
+    const { claim } = data
+    const ipfsFileHash = await this.uploadClaim(claim)
+    return {
+      ...data,
       ipfsFileHash,
-    })
+    }
   }
+
+  private readonly addIPFSHashToClaim = async (data: StoreNextClaimData): Promise<StoreNextClaimData> => {
+    const { claim, ipfsFileHash } = data
+    await this.daoClaims.addClaimHash(claim.id, ipfsFileHash)
+    return data
+  }
+
+  public storeNextClaim = pipeP(
+    this.log(LogTypes.trace)('Finding Claim'),
+    this.findNextClaim,
+    this.log(LogTypes.trace)('Storing Claim'),
+    this.storeClaim,
+    this.log(LogTypes.trace)('Adding IPFS hash to Claim Entry'),
+    this.addIPFSHashToClaim,
+    this.log(LogTypes.trace)('Finished Storing Claim')
+  )
 }

src/StorageWriter/ClaimController.ts

@@ -0,0 +1,69 @@
+import { Claim } from '@po.et/poet-js'
+import { FindAndModifyWriteOpResultObject } from 'mongodb'
+import { describe, Try } from 'riteway'
+
+import { getClaimFromFindAndUpdateResponse, throwIfClaimNotFound } from './DAOClaims'
+
+describe('DOAClaims.getClaimFromFindAndUpdateResponse', async should => {
+  const { assert } = should('return the correct value')
+
+  {
+    const response: FindAndModifyWriteOpResultObject = {}
+    assert({
+      given: 'a response that does not contain a value',
+      actual: getClaimFromFindAndUpdateResponse(response),
+      expected: undefined,
+    })
+  }
+
+  {
+    const response: FindAndModifyWriteOpResultObject = { value: {} }
+    assert({
+      given: 'a response that does not contain a claim',
+      actual: getClaimFromFindAndUpdateResponse(response),
+      expected: undefined,
+    })
+  }
+
+  {
+    const claim = { id: 'bar' } as Claim
+    const response: FindAndModifyWriteOpResultObject = {
+      value: {
+        claim,
+      },
+    }
+    assert({
+      given: 'a response that contians a claim',
+      actual: getClaimFromFindAndUpdateResponse(response),
+      expected: claim,
+    })
+  }
+})
+
+describe('DOAClaims.throwIfClaimNotFound', async should => {
+  const { assert } = should()
+
+  assert({
+    given: 'null',
+    should: 'throw',
+    actual: Try(throwIfClaimNotFound, null),
+    expected: new Error(),
+  })
+
+  assert({
+    given: 'undefined',
+    should: 'throw',
+    actual: Try(throwIfClaimNotFound, undefined),
+    expected: new Error(),
+  })
+
+  {
+    const claim: Claim = { id: 'bar' } as Claim
+    assert({
+      given: 'a claim',
+      should: 'return the claim',
+      actual: throwIfClaimNotFound(claim),
+      expected: claim,
+    })
+  }
+})

src/StorageWriter/DAOClaims.test.ts

@@ -0,0 +1,62 @@
+import { Claim } from '@po.et/poet-js'
+import { inject, injectable } from 'inversify'
+import { Collection, Db, FindAndModifyWriteOpResultObject } from 'mongodb'
+import { isNil, pipeP, lensPath, view } from 'ramda'
+
+import { NoMoreEntriesException } from './Exceptions'
+
+export interface DAOClaimsConfiguration {
+  readonly maxStorageAttempts: number
+}
+
+const L = {
+  valueClaim: lensPath(['value', 'claim']),
+}
+
+export const getClaimFromFindAndUpdateResponse = (response: FindAndModifyWriteOpResultObject): Claim | undefined =>
+  view(L.valueClaim, response)
+
+export const throwIfClaimNotFound = (claim: Claim): Claim => {
+  if (isNil(claim)) throw new NoMoreEntriesException('No claims found')
+  return claim
+}
+
+@injectable()
+export class DAOClaims {
+  private readonly collection: Collection
+  private readonly maxStorageAttempts: number
+
+  constructor(@inject('DB') db: Db, @inject('DAOClaimsConfiguration') configuration: DAOClaimsConfiguration) {
+    this.collection = db.collection('storageWriterClaims')
+    this.maxStorageAttempts = configuration.maxStorageAttempts
+  }
+
+  public readonly start = async () => {
+    await this.collection.createIndex({ 'claim.id': 1 }, { unique: true })
+  }
+
+  public readonly addClaim = async (claim: Claim) => {
+    await this.collection.insertOne({ claim, storageAttempts: 0, ipfsFileHash: null })
+  }
+
+  public readonly addClaimHash = async (claimId: string, ipfsFileHash: string) => {
+    await this.collection.updateOne({ 'claim.id': claimId }, { $set: { ipfsFileHash } })
+  }
+
+  private readonly findClaimToStore = () =>
+    this.collection.findOneAndUpdate(
+      {
+        $and: [{ ipfsFileHash: null }, { storageAttempts: { $lt: this.maxStorageAttempts } }],
+      },
+      {
+        $inc: { storageAttempts: 1 },
+        $set: { lastStorageAttemptTime: new Date().getTime() },
+      }
+    )
+
+  public readonly findNextClaim = pipeP(
+    this.findClaimToStore,
+    getClaimFromFindAndUpdateResponse,
+    throwIfClaimNotFound
+  )
+}

src/StorageWriter/DAOClaims.ts

@@ -0,0 +1,17 @@
+import { describe } from 'riteway'
+
+import { NoMoreEntriesException, isNoMoreEntriesException } from './Exceptions'
+
+describe('Exceptions.isNoMoreEntriesException', async should => {
+  const { assert } = should('return the correct boolean')
+  assert({
+    given: 'a normal error',
+    actual: isNoMoreEntriesException(new Error()),
+    expected: false,
+  })
+  assert({
+    given: 'a NoMoreEntriesException error',
+    actual: isNoMoreEntriesException(new NoMoreEntriesException()),
+    expected: true,
+  })
+})

src/StorageWriter/Exceptions.test.ts

@@ -0,0 +1,7 @@
+export class NoMoreEntriesException extends Error {
+  constructor(message?: string) {
+    super(message)
+  }
+}
+
+export const isNoMoreEntriesException = (err: any) => err instanceof NoMoreEntriesException

src/StorageWriter/Exceptions.ts

@@ -3,4 +3,5 @@ export interface ExchangeConfiguration {
   readonly newClaim?: string
   readonly poetAnchorDownloaded?: string
   readonly claimsDownloaded?: string
+  readonly storageWriterStoreNextClaim: string
 }

src/StorageWriter/ExchangeConfiguration.ts

@@ -3,7 +3,9 @@ import { inject, injectable } from 'inversify'
 import fetch from 'node-fetch'
 import * as str from 'string-to-stream'
 
-import { IPFSConfiguration } from './IPFSConfiguration'
+export interface IPFSConfiguration {
+  readonly ipfsUrl: string
+}
 
 /**
  * Wrapper around IPFS' RPC