From 2d193c6bd2d424d7809df9b09b9aec4c519e302a Mon Sep 17 00:00:00 2001 From: Chris Troutner Date: Fri, 28 Aug 2026 10:05:24 -0700 Subject: [PATCH] Implement ZMQ-mode DB backups every epoch blocks By coder. --- psf-memo-indexer/acceptance/lib/handlers.js | 42 +++++++++++- psf-memo-indexer/config/index.js | 4 ++ psf-memo-indexer/psf-memo-block-indexer.js | 8 +-- psf-memo-indexer/src/use-cases/backup-db.js | 27 ++++++++ .../src/use-cases/use-cases-index.js | 2 + .../test/unit/use-cases/backup-db.unit.js | 68 +++++++++++++++++++ 6 files changed, 146 insertions(+), 5 deletions(-) create mode 100644 psf-memo-indexer/src/use-cases/backup-db.js create mode 100644 psf-memo-indexer/test/unit/use-cases/backup-db.unit.js diff --git a/psf-memo-indexer/acceptance/lib/handlers.js b/psf-memo-indexer/acceptance/lib/handlers.js index 209d9f4..2784c35 100644 --- a/psf-memo-indexer/acceptance/lib/handlers.js +++ b/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 (.+)$/, diff --git a/psf-memo-indexer/config/index.js b/psf-memo-indexer/config/index.js index 34c52b1..7ec1618 100644 --- a/psf-memo-indexer/config/index.js +++ b/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 diff --git a/psf-memo-indexer/psf-memo-block-indexer.js b/psf-memo-indexer/psf-memo-block-indexer.js index f60d1b6..9a0fc35 100644 --- a/psf-memo-indexer/psf-memo-block-indexer.js +++ b/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++ diff --git a/psf-memo-indexer/src/use-cases/backup-db.js b/psf-memo-indexer/src/use-cases/backup-db.js new file mode 100644 index 0000000..830d689 --- /dev/null +++ b/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 diff --git a/psf-memo-indexer/src/use-cases/use-cases-index.js b/psf-memo-indexer/src/use-cases/use-cases-index.js index 1a615a6..ea5fe4d 100644 --- a/psf-memo-indexer/src/use-cases/use-cases-index.js +++ b/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() diff --git a/psf-memo-indexer/test/unit/use-cases/backup-db.unit.js b/psf-memo-indexer/test/unit/use-cases/backup-db.unit.js new file mode 100644 index 0000000..ba8780a --- /dev/null +++ b/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) + }) + }) +})