Implement ZMQ-mode DB backups every epoch blocks

By coder.

psf-memo-indexer/acceptance/lib/handlers.js

@@ -10,6 +10,7 @@ import crypto from 'node:crypto'
 import { handlePost } from '../../src/use-cases/action-types/post.js'
 import { handleReply } from '../../src/use-cases/action-types/reply.js'
 import { handleLike } from '../../src/use-cases/action-types/like.js'
+import BackupDb from '../../src/use-cases/backup-db.js'
 
 function makeInMemoryDb () {
   const store = new Map()
@@ -75,6 +76,7 @@ async function createWorld () {
   const postChildrenDb = makeInMemoryDb()
   const likesDb = makeInMemoryDb()
   const postLikesDb = makeInMemoryDb()
+  const backupRequestsDb = makeInMemoryDb()
 
   const adapters = {
     postDb: postsDb,
@@ -84,7 +86,13 @@ async function createWorld () {
     postChildDb: postChildrenDb,
     likeDb: likesDb,
     postLikeDb: postLikesDb,
-    processErrorDb: makeInMemoryDb()
+    processErrorDb: makeInMemoryDb(),
+    dbCtrl: {
+      backupDb: async (height, epoch) => {
+        await backupRequestsDb.create(`${height}:${epoch}`, { height, epoch })
+        return true
+      }
+    }
   }
 
   return {
@@ -96,6 +104,7 @@ async function createWorld () {
     postChildrenDb,
     likesDb,
     postLikesDb,
+    backupRequestsDb,
     txidMap: new Map(),
     lastTxid: null,
     lastHeight: null,
@@ -111,6 +120,13 @@ const handlers = [
       // World is already created with both stores.
     }
   },
+  {
+    name: 'db instance that records backup requests',
+    pattern: /^a psf-memo-db instance that records backup requests$/,
+    async run () {
+      // World is already created with a backup request store.
+    }
+  },
   {
     name: 'db instance with new indexes',
     pattern: /^a psf-memo-db instance with posts, postHeights, addrPostHeights, likes, and postLikes stores$/,
@@ -244,6 +260,30 @@ const handlers = [
       })
     }
   },
+  {
+    name: 'block indexer in ZMQ mode processes a block',
+    pattern: /^the block indexer in ZMQ mode processes a block at height (.+) with epoch (.+)$/,
+    async run (m, example, world) {
+      const height = parseInt(resolveParam(m[1], example), 10)
+      const epoch = parseInt(resolveParam(m[2], example), 10)
+      const backupDb = new BackupDb({ adapters: world.adapters })
+      await backupDb.maybeBackupDb(height, epoch)
+    }
+  },
+  {
+    name: 'db receives backup request',
+    pattern: /^the psf-memo-db receives (.+) backup request for block (.+) with epoch (.+)$/,
+    run (m, example, world) {
+      const expectedCount = parseInt(resolveParam(m[1], example), 10)
+      const height = parseInt(resolveParam(m[2], example), 10)
+      const epoch = parseInt(resolveParam(m[3], example), 10)
+      const key = `${height}:${epoch}`
+      const matching = world.backupRequestsDb.entries().filter(([k]) => k === key)
+      if (matching.length !== expectedCount) {
+        throw new Error(`Expected ${expectedCount} backup request(s) for block ${height} epoch ${epoch}, got ${matching.length}`)
+      }
+    }
+  },
   {
     name: 'posts store contains post document',
     pattern: /^the posts store contains (.+) post document for (.+)$/,

psf-memo-indexer/config/index.js

@@ -29,6 +29,10 @@ export default {
 
   exitOnMissingBackup: process.env.EXIT_ON_MISSING_BACKUP === 'true',
 
+  dbBackupEpoch: process.env.DB_BACKUP_EPOCH
+    ? parseInt(process.env.DB_BACKUP_EPOCH, 10)
+    : 1000,
+
   debugLevel: process.env.DEBUG_LEVEL !== undefined
     ? parseInt(process.env.DEBUG_LEVEL, 10)
     : 0

psf-memo-indexer/psf-memo-block-indexer.js

@@ -5,12 +5,11 @@
 import RetryQueue from '@chris.troutner/retry-queue'
 import 'dotenv/config'
 
+import config from './config/index.js'
 import Adapters from './src/adapters/adapters-index.js'
 import UseCases from './src/use-cases/use-cases-index.js'
 import Controllers from './src/controllers/controllers-index.js'
 
-const EPOCH = 1000
-
 async function start () {
   try {
     const adapters = new Adapters()
@@ -49,9 +48,9 @@ async function start () {
           process.exit(1)
         }
 
-        if (nextBlockHeight % EPOCH === 0) {
+        if (nextBlockHeight % config.dbBackupEpoch === 0) {
           console.log(`Creating DB backup at block ${nextBlockHeight}`)
-          await adapters.dbCtrl.backupDb(nextBlockHeight, EPOCH)
+          await useCases.backupDb.maybeBackupDb(nextBlockHeight, config.dbBackupEpoch)
         }
 
         biggestBlockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
@@ -86,6 +85,7 @@ async function start () {
         await adapters.statusDb.updateStatus(liveStatus)
 
         await useCases.indexBlocks.processBlock(blockHeight)
+        await useCases.backupDb.maybeBackupDb(blockHeight, config.dbBackupEpoch)
       }
 
       loopCnt++

psf-memo-indexer/src/use-cases/backup-db.js

@@ -0,0 +1,27 @@
+/*
+  Decide when the indexer should ask psf-memo-db to back up its LevelDB.
+*/
+
+class BackupDb {
+  constructor (localConfig = {}) {
+    if (!localConfig.adapters) {
+      throw new Error('Adapters required for backup-db.js')
+    }
+    this.adapters = localConfig.adapters
+    this.maybeBackupDb = this.maybeBackupDb.bind(this)
+  }
+
+  async maybeBackupDb (blockHeight, epoch) {
+    const height = parseInt(blockHeight, 10)
+    const backupEpoch = parseInt(epoch, 10)
+
+    if (height > 0 && height % backupEpoch === 0) {
+      await this.adapters.dbCtrl.backupDb(height, backupEpoch)
+      return true
+    }
+
+    return false
+  }
+}
+
+export default BackupDb

psf-memo-indexer/src/use-cases/use-cases-index.js

@@ -1,4 +1,5 @@
 import RetryQueue from '@chris.troutner/retry-queue'
+import BackupDb from './backup-db.js'
 import IndexBlocks from './index-blocks.js'
 import State from './state.js'
 import Utils from './utils.js'
@@ -10,6 +11,7 @@ class UseCases {
     }
     this.adapters = localConfig.adapters
     this.indexBlocks = new IndexBlocks({ adapters: this.adapters })
+    this.backupDb = new BackupDb({ adapters: this.adapters })
     this.state = new State({ adapters: this.adapters })
     this.utils = new Utils()
     this.retryQueue = new RetryQueue()

psf-memo-indexer/test/unit/use-cases/backup-db.unit.js

@@ -0,0 +1,68 @@
+import { assert } from 'chai'
+import sinon from 'sinon'
+import BackupDb from '../../../src/use-cases/backup-db.js'
+
+describe('#BackupDb', () => {
+  let uut
+  let sandbox
+  let adapters
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox()
+    adapters = {
+      dbCtrl: {
+        backupDb: sandbox.stub().resolves(true)
+      }
+    }
+    uut = new BackupDb({ adapters })
+  })
+
+  afterEach(() => sandbox.restore())
+
+  describe('#maybeBackupDb', () => {
+    it('should request a backup at an epoch boundary', async () => {
+      const result = await uut.maybeBackupDb(1000, 1000)
+
+      assert.equal(result, true)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 1)
+      assert.deepEqual(adapters.dbCtrl.backupDb.firstCall.args, [1000, 1000])
+    })
+
+    it('should request a backup at a multiple of the epoch', async () => {
+      const result = await uut.maybeBackupDb(2000, 1000)
+
+      assert.equal(result, true)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 1)
+      assert.deepEqual(adapters.dbCtrl.backupDb.firstCall.args, [2000, 1000])
+    })
+
+    it('should not request a backup between epoch boundaries', async () => {
+      const result = await uut.maybeBackupDb(1001, 1000)
+
+      assert.equal(result, false)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 0)
+    })
+
+    it('should not request a backup at height zero', async () => {
+      const result = await uut.maybeBackupDb(0, 1000)
+
+      assert.equal(result, false)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 0)
+    })
+
+    it('should support a configurable epoch smaller than 1000', async () => {
+      const result = await uut.maybeBackupDb(500, 500)
+
+      assert.equal(result, true)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 1)
+      assert.deepEqual(adapters.dbCtrl.backupDb.firstCall.args, [500, 500])
+    })
+
+    it('should not request a backup just past a non-1000 epoch boundary', async () => {
+      const result = await uut.maybeBackupDb(1001, 500)
+
+      assert.equal(result, false)
+      assert.equal(adapters.dbCtrl.backupDb.callCount, 0)
+    })
+  })
+})