From 25ae5109b917d1d5f74fa3d5b728feb0b9254c55 Mon Sep 17 00:00:00 2001 From: rodribp Date: Mon, 10 Aug 2026 01:45:04 -0600 Subject: [PATCH] add: backend indexing + reorg handling --- backend/src/api/blocks.ts | 155 +++++++++++++++++- backend/src/indexer.ts | 2 + .../repositories/BlocksSummariesRepository.ts | 28 ++++ 3 files changed, 184 insertions(+), 1 deletion(-) diff --git a/backend/src/api/blocks.ts b/backend/src/api/blocks.ts index be247f571..97276a922 100644 --- a/backend/src/api/blocks.ts +++ b/backend/src/api/blocks.ts @@ -40,6 +40,7 @@ import CpfpRepository from '../repositories/CpfpRepository'; import { parseDATUMTemplateCreator, parseDMNDTemplateCreator } from '../utils/bitcoin-script'; import database from '../database'; import { getBlockFirstSeenFromLogs, getOldestLogTimestampFromLogs, scanLogsForBlocksFirstSeen } from '../utils/file-read'; +import FlagValueRepository, { INDEXING_PRESETS } from '../repositories/FlagValueRepository'; class Blocks { private blocks: BlockExtended[] = []; @@ -54,6 +55,8 @@ class Blocks { private oldestCoreLogTimestamp: number | undefined | null = undefined; private mainLoopTimeout: number = 120000; + private indexingFlagValues: boolean = false; + private flagValuesDeleteQueue: number[]= []; constructor() { } @@ -689,6 +692,155 @@ class Blocks { } } + /** + * [INDEXING] Index all blocks flag values for the goggles graph rendering + * + * @asyncSafe + */ + public async $generateFlagValuesDatabase(): Promise { + const MAX_BLOCKS_PERQUERY = 144; + if (this.indexingFlagValues) { + return; + } + + if (Common.blocksSummariesIndexingEnabled() === false || Common.isLiquid()) { + return; + } + + this.indexingFlagValues = true; + + const tipOfSummaries = await BlocksSummariesRepository.$getTipIndexed(); + if (!tipOfSummaries) { + this.indexingFlagValues = false; + return; + } + + let newlyIndexedBuckets = 0; + + while (this.flagValuesDeleteQueue.length > 0) { // Deletion of in-queue heights due to reorg + const deletionHeight = this.flagValuesDeleteQueue.shift(); + if (deletionHeight === undefined) { + continue; + } + await FlagValueRepository.$deleteFlagValuesFromHeight(deletionHeight); + } + + for (const preset of INDEXING_PRESETS) { + let seedHeight = preset.retentionSpan > -1 ? tipOfSummaries - preset.retentionSpan : 0; + if (config.MEMPOOL.INDEXING_BLOCKS_AMOUNT > 0) { + seedHeight = Math.max(seedHeight, tipOfSummaries - config.MEMPOOL.INDEXING_BLOCKS_AMOUNT + 1); + } + const firstBucket = Math.floor((tipOfSummaries + 1) / preset.bucketSize) * preset.bucketSize - preset.bucketSize; + const lastBucket = Math.max(0, Math.floor(seedHeight / preset.bucketSize) * preset.bucketSize); + + // Deletion of flag values out of retention span + const tipAndTailOfFlagValues = await FlagValueRepository.$getTipAndTailIndexedByBucketSize(preset.bucketSize); + if (tipAndTailOfFlagValues && lastBucket > tipAndTailOfFlagValues.tail) { // Drop buckets that fell out of block span + logger.debug(`Deleting all the flag values ${preset.name} below height #${lastBucket}`, logger.tags.goggles); + await FlagValueRepository.$deleteFlagValuesBelowHeight(lastBucket, preset.bucketSize); + } + + if (firstBucket < lastBucket) { + continue; // no complete bucket in range + } + + const indexedBuckets = await FlagValueRepository.$getIndexedStartHeights(preset.bucketSize, firstBucket, lastBucket); + const isBucketIndexed = {}; + // We map the buckets that are already indexed to skip them + for (const startHeight of indexedBuckets) { + isBucketIndexed[startHeight] = true; + } + + logger.debug(`Processing and indexing flag values from #${firstBucket} to #${lastBucket} ${preset.name}`, logger.tags.goggles); + + let timer = Date.now() / 1000; + const startedAt = Date.now() / 1000; + let blocksComputedInTotal = 0; + let blocksComputedThisRun = 0; + const blocksToCompute = firstBucket + preset.bucketSize - lastBucket - (indexedBuckets.length * preset.bucketSize); + for (let bucketStart = firstBucket; bucketStart >= lastBucket; bucketStart -= preset.bucketSize) { + if (isBucketIndexed[bucketStart]) { + continue; // already indexed + } + try { + const bucketFirstHeight = bucketStart + preset.bucketSize - 1; + const bucketLastHeight = bucketStart - 1; + + let step = bucketFirstHeight; + + const dataPerFlag: Record> = {}; + let sumTimestamps = 0; + let nBlocks = 0; + let incomplete = false; + + // Incrementalized logic capped by max blocks per query, not bucket size + while (step > bucketLastHeight) { + const blocksPerQuery = Math.min(step - bucketLastHeight, MAX_BLOCKS_PERQUERY); + const cappedLastHeight = step - blocksPerQuery; + + const blocks = await BlocksSummariesRepository.$getSummariesBetweenHeights(step, cappedLastHeight); + await Common.sleep$(250); // Don't query/index flag values too fast + + if (!blocks || blocks.length < blocksPerQuery) { + incomplete = true; + break; // Incomplete bucket + } + + // Flag values processing + for (const block of blocks) { + const txData = JSON.parse(block.transactions).map((tx) => ({flags: tx.flags, vsize: tx.vsize})); + for (const data of txData) { + if (dataPerFlag[data.flags] === undefined || Object.keys(dataPerFlag[data.flags]).length === 0) { + dataPerFlag[data.flags] = { + txCount: 0, + vSizeTotal: 0 + }; + } + dataPerFlag[data.flags].txCount = dataPerFlag[data.flags].txCount + 1; + dataPerFlag[data.flags].vSizeTotal = dataPerFlag[data.flags].vSizeTotal + data.vsize; + } + sumTimestamps += block.timestamp; + blocksComputedInTotal++; + blocksComputedThisRun++; + nBlocks++; + } + + // Logging + const elapsedSeconds = (Date.now() / 1000) - timer; + if (elapsedSeconds > 5) { + const runningFor = (Date.now() / 1000) - startedAt; + const blocksPerSecond = blocksComputedThisRun / elapsedSeconds; + const completion = (blocksComputedInTotal / blocksToCompute) * 100; + logger.debug(`Indexing flag values ${preset.name} | ${blocksComputedInTotal}/${blocksToCompute} (${completion.toFixed(2)}%) | ~${blocksPerSecond.toFixed(2)} blocks/sec | elapsed: ${runningFor.toFixed(2)} seconds`,logger.tags.goggles); + timer = Date.now() / 1000; + blocksComputedThisRun = 0; + } + + step -= blocksPerQuery; + } + + if (incomplete) { + continue; + } + + const avgTimestamp = sumTimestamps / nBlocks; + await FlagValueRepository.$saveBatchFlagValues(preset.bucketSize, bucketStart, dataPerFlag, avgTimestamp); + nBlocks = 0; + newlyIndexedBuckets++; + } catch (e) { + logger.err(`Failed to index flag values between #${bucketStart} and #${bucketStart + preset.bucketSize - 1}. Reason: ${(e instanceof Error ? e.message : e)}`, logger.tags.goggles); + } + } + logger.debug(`Successfully indexed #${blocksComputedInTotal} blocks ${preset.name} in ${((Date.now() / 1000) - startedAt).toFixed(2)} seconds`, logger.tags.goggles); + } + if (newlyIndexedBuckets > 0) { + logger.notice(`Flag values indexing completed: indexed ${newlyIndexedBuckets} buckets`, logger.tags.goggles); + } else { + logger.debug(`Flag values indexing completed: indexed ${newlyIndexedBuckets} buckets`, logger.tags.goggles); + } + this.indexingFlagValues = false; + } + /** @asyncUnsafe */ public async $indexBlockSummary(hash: string, height: number, stale?: boolean): Promise { if (config.MEMPOOL.BACKEND === 'esplora') { @@ -1409,6 +1561,7 @@ class Blocks { await DifficultyAdjustmentsRepository.$deleteAdjustementsFromHeight(forkTail.height); await cpfpRepository.$deleteClustersFrom(forkTail.height); await AccelerationRepository.$deleteAccelerationsFrom(forkTail.height); + this.flagValuesDeleteQueue.push(forkTail.height); chainTips.clearOrphanCacheAboveHeight(forkTail.height); this.updateTimerProgress(timer, `deleted stale block data`); @@ -1769,7 +1922,7 @@ class Blocks { if (transactions?.length != null) { const { cpfpSummary } = await detectTemplateAlgorithm(height, transactions, [], true); - if (!stale) { + if (!stale && Common.cpfpIndexingEnabled() === true) { await this.$saveCpfp(hash, height, cpfpSummary); } diff --git a/backend/src/indexer.ts b/backend/src/indexer.ts index ff10d701e..9d47de2f2 100644 --- a/backend/src/indexer.ts +++ b/backend/src/indexer.ts @@ -226,6 +226,8 @@ class Indexer { await AccelerationRepository.$indexPastAccelerations(); await BlocksAuditsRepository.$migrateAuditsV0toV1(); await BlocksRepository.$migrateBlocks(); + + void blocks.$generateFlagValuesDatabase(); // do not wait for classify blocks to finish void blocks.$classifyBlocks(); runSuccessful = true; diff --git a/backend/src/repositories/BlocksSummariesRepository.ts b/backend/src/repositories/BlocksSummariesRepository.ts index fc2771ce8..2239fa258 100644 --- a/backend/src/repositories/BlocksSummariesRepository.ts +++ b/backend/src/repositories/BlocksSummariesRepository.ts @@ -216,6 +216,34 @@ class BlocksSummariesRepository { } return false; } + + /** @asyncSafe */ + public async $getTipIndexed(): Promise { + if (!Common.blocksSummariesIndexingEnabled()) { + return null; + } + try { + const [row]: any[] = await DB.query('SELECT MAX(height) as tip FROM blocks_summaries WHERE version >= 1'); + + if (row !== null && row.length > 0) { + return row[0].tip; + } + } catch (e) { + logger.err(`Cannot get latest block summary. Reason: ` + (e instanceof Error ? e.message : e)); + } + return null; + } + + public async $getSummariesBetweenHeights(startHeight: number, latestHeight: number): Promise<{height: number, transactions: string, timestamp: number}[]> { + try { + const [rows]: any[] = await DB.query(`SELECT bs.height, bs.transactions, UNIX_TIMESTAMP(b.blockTimestamp) as timestamp FROM blocks_summaries bs JOIN blocks b ON bs.id = b.hash WHERE bs.height <= ? AND bs.height > ? AND b.stale = 0 AND bs.version >= 1 ORDER BY height DESC`, [startHeight, latestHeight]); + + return rows; + } catch (e) { + logger.err(`Cannot get blocks between ${startHeight} and ${latestHeight}. Reason: ` + (e instanceof Error ? e.message : e)); + throw e; + } + } } export default new BlocksSummariesRepository();