diff --git a/backend/src/__integration_tests__/pools-repository.test.ts b/backend/src/__integration_tests__/pools-repository.test.ts index 55e503b26..e750c720d 100644 --- a/backend/src/__integration_tests__/pools-repository.test.ts +++ b/backend/src/__integration_tests__/pools-repository.test.ts @@ -1,5 +1,14 @@ -import PoolsRepository from '../repositories/PoolsRepository'; -import { setupTestDatabase, waitForDatabase, cleanupTestData, insertTestPool } from './test-helpers'; +import PoolsRepository, { POOLS_STATS_INTERVALS } from '../repositories/PoolsRepository'; +import { setupTestDatabase, waitForDatabase, cleanupTestData, insertTestPool, insertTestBlock, insertTestBlockAudit } from './test-helpers'; + +function hoursAgo(hours: number): Date { + return new Date(Date.now() - hours * 3600000); +} + +/** Block hashes only need to be unique and 64 hex chars wide for these tests. */ +function blockHash(height: number): string { + return height.toString(16).padStart(64, '0'); +} describe('PoolsRepository Integration Tests', () => { beforeAll(async () => { @@ -118,5 +127,139 @@ describe('PoolsRepository Integration Tests', () => { expect(pool).toBeDefined(); expect(pool!.name).toBe('Updated Pool Name'); }); + + describe('$getPoolsInfoPerInterval', () => { + let poolA: number; + let poolB: number; + + beforeEach(async () => { + poolA = await insertTestPool({ id: 1, name: 'Pool A', slug: 'pool-a' }); + poolB = await insertTestPool({ id: 2, name: 'Pool B', slug: 'pool-b' }); + }); + + test('should return an empty array for every interval on an empty database', async () => { + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + for (const interval of POOLS_STATS_INTERVALS) { + expect(result[interval]).toEqual([]); + } + }); + + test('should only count blocks inside each interval window', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(12), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(48), poolId: poolA }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'][0].blockCount).toBe(1); + expect(result['3d'][0].blockCount).toBe(2); + expect(result['all'][0].blockCount).toBe(2); + }); + + test('should omit pools with no blocks in an interval', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(12), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(48), poolId: poolB }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'].map(p => p.poolId)).toEqual([poolA]); + expect(result['3d'].map(p => p.poolId).sort()).toEqual([poolA, poolB].sort()); + }); + + test('should exclude stale blocks from every interval', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(1), poolId: poolA, stale: true }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'][0].blockCount).toBe(1); + expect(result['all'][0].blockCount).toBe(1); + }); + + test('should count empty blocks only within the interval', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(12), poolId: poolA, tx_count: 1 }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(12), poolId: poolA, tx_count: 500 }); + await insertTestBlock({ height: 3, hash: blockHash(3), blockTimestamp: hoursAgo(48), poolId: poolA, tx_count: 1 }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'][0].emptyBlocks).toBe(1); + expect(result['24h'][0].blockCount).toBe(2); + expect(result['3d'][0].emptyBlocks).toBe(2); + }); + + test('should keep unaudited blocks and leave their averages null', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'][0].blockCount).toBe(1); + expect(result['24h'][0].avgMatchRate).toBeNull(); + expect(result['24h'][0].avgFeeDelta).toBeNull(); + }); + + test('should average match rate over audited blocks only', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(1), poolId: poolA }); + await insertTestBlockAudit({ hash: blockHash(1), height: 1, matchRate: 90 }); + await insertTestBlockAudit({ hash: blockHash(2), height: 2, matchRate: 100 }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'][0].blockCount).toBe(2); + expect(Number(result['24h'][0].avgMatchRate)).toBeCloseTo(95, 5); + }); + + test('should compute the average fee delta against expected fees', async () => { + // actual fees are 10% above expected, so the delta is 0.1 + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA, fees: 1100 }); + await insertTestBlockAudit({ hash: blockHash(1), height: 1, matchRate: 100, expectedFees: 1000 }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(Number(result['24h'][0].avgFeeDelta)).toBeCloseTo(0.1, 5); + }); + + test('should sort pools by descending block count', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(1), poolId: poolB }); + await insertTestBlock({ height: 3, hash: blockHash(3), blockTimestamp: hoursAgo(1), poolId: poolB }); + + const result = await PoolsRepository.$getPoolsInfoPerInterval(); + + expect(result['24h'].map(p => p.poolId)).toEqual([poolB, poolA]); + expect(result['24h'].map(p => p.blockCount)).toEqual([2, 1]); + }); + + test('should break block count ties deterministically', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(1), poolId: poolA }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(1), poolId: poolB }); + + const first = await PoolsRepository.$getPoolsInfoPerInterval(); + const second = await PoolsRepository.$getPoolsInfoPerInterval(); + + // unique_id decides, so the ranks assigned from this order stay stable across rebuilds + expect(first['24h'].map(p => p.poolUniqueId)).toEqual([1, 2]); + expect(second['24h'].map(p => p.poolUniqueId)).toEqual(first['24h'].map(p => p.poolUniqueId)); + }); + + test('should match the legacy per-interval query', async () => { + await insertTestBlock({ height: 1, hash: blockHash(1), blockTimestamp: hoursAgo(2), poolId: poolA, tx_count: 1 }); + await insertTestBlock({ height: 2, hash: blockHash(2), blockTimestamp: hoursAgo(3), poolId: poolA, fees: 1100 }); + await insertTestBlock({ height: 3, hash: blockHash(3), blockTimestamp: hoursAgo(4), poolId: poolB }); + await insertTestBlock({ height: 4, hash: blockHash(4), blockTimestamp: hoursAgo(48), poolId: poolB }); + await insertTestBlock({ height: 5, hash: blockHash(5), blockTimestamp: hoursAgo(2), poolId: poolB, stale: true }); + await insertTestBlockAudit({ hash: blockHash(2), height: 2, matchRate: 80, expectedFees: 1000 }); + + const legacy = await PoolsRepository.$getPoolsInfo('24h'); + const combined = (await PoolsRepository.$getPoolsInfoPerInterval())['24h']; + + expect(combined.map(p => p.poolId)).toEqual(legacy.map(p => p.poolId)); + expect(combined.map(p => p.blockCount)).toEqual(legacy.map(p => p.blockCount)); + expect(combined.map(p => p.emptyBlocks)).toEqual(legacy.map(p => p.emptyBlocks)); + expect(combined.map(p => Number(p.avgMatchRate))).toEqual(legacy.map(p => Number(p.avgMatchRate))); + expect(combined.map(p => Number(p.avgFeeDelta))).toEqual(legacy.map(p => Number(p.avgFeeDelta))); + }); + }); }); diff --git a/backend/src/__integration_tests__/test-helpers.ts b/backend/src/__integration_tests__/test-helpers.ts index 91e47fe2e..8ef3529b1 100644 --- a/backend/src/__integration_tests__/test-helpers.ts +++ b/backend/src/__integration_tests__/test-helpers.ts @@ -138,11 +138,14 @@ export async function insertTestBlock(blockData: { tx_count?: number; difficulty?: number; poolId?: number | null; + stale?: boolean; + fees?: number; }) { const timestamp = blockData.blockTimestamp || new Date(); const size = blockData.size || 1000000; const weight = blockData.weight || 4000000; const txCount = blockData.tx_count || 2000; + const fees = blockData.fees !== undefined ? blockData.fees : 50000000; await DB.query( `INSERT INTO blocks ( @@ -170,9 +173,9 @@ export async function insertTestBlock(blockData: { '0000000000000000000000000000000000000000000000000000000000000000', '0000000000000000000000000000000000000000000000000000000000000000', timestamp, - 0, // stale = false + blockData.stale ? 1 : 0, // Required fields with defaults - 50000000, // fees (in sats) + fees, // fees (in sats) JSON.stringify([0, 0, 0, 0, 0, 0, 0]), // fee_span (JSON array) 10000, // median_fee (in sats) size / txCount, // avg_tx_size @@ -188,3 +191,30 @@ export async function insertTestBlock(blockData: { ); } +/** + * Insert a test block audit into the database + */ +export async function insertTestBlockAudit(auditData: { + hash: string; + height: number; + matchRate: number; + expectedFees?: number | null; + time?: Date; +}) { + await DB.query( + `INSERT INTO blocks_audits ( + hash, height, time, match_rate, expected_fees, missing_txs, added_txs + ) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + [ + auditData.hash, + auditData.height, + auditData.time || new Date(), + auditData.matchRate, + auditData.expectedFees !== undefined ? auditData.expectedFees : null, + '[]', // missing_txs + '[]' // added_txs + ] + ); +} + diff --git a/backend/src/__tests__/api/mining-pools-stats-cache.test.ts b/backend/src/__tests__/api/mining-pools-stats-cache.test.ts new file mode 100644 index 000000000..e7e68bdb8 --- /dev/null +++ b/backend/src/__tests__/api/mining-pools-stats-cache.test.ts @@ -0,0 +1,185 @@ +const mockGetPoolsInfoPerInterval = jest.fn(); +const mockBlockCount = jest.fn(); +const mockGetNetworkHashPs = jest.fn(); + +jest.mock('../../repositories/PoolsRepository', () => ({ + __esModule: true, + default: { $getPoolsInfoPerInterval: mockGetPoolsInfoPerInterval }, + POOLS_STATS_INTERVALS: ['24h', 'all'], +})); + +jest.mock('../../repositories/BlocksRepository', () => ({ + __esModule: true, + default: { $blockCount: mockBlockCount }, +})); + +jest.mock('../../api/bitcoin/bitcoin-client', () => ({ + __esModule: true, + default: { getNetworkHashPs: mockGetNetworkHashPs }, +})); + +// mining.ts pulls these in transitively, and merely importing them opens handles (a polling interval +// in backend-info, a mysql pool) that would keep the test runner alive after the suite finishes +jest.mock('../../repositories/PricesRepository', () => ({ __esModule: true, default: {} })); +jest.mock('../../database', () => ({ __esModule: true, default: { query: jest.fn() } })); +jest.mock('../../api/bitcoin/bitcoin-api-factory', () => ({ __esModule: true, default: {} })); + +import mining from '../../api/mining/mining'; + +/** + * Builds a repository result whose block counts identify which rebuild produced it, so a test can tell + * a superseded snapshot apart from a current one. + */ +function poolsInfo(blockCount: number) { + const pool = { + poolId: 1, + name: 'Pool', + link: '', + slug: 'pool', + poolUniqueId: 1, + blockCount, + emptyBlocks: 0, + avgMatchRate: null, + avgFeeDelta: null, + }; + return { '24h': [pool], 'all': [pool] }; +} + +/** Lets a test hold a rebuild open until it decides to release it. */ +function deferred() { + let resolve!: (value: unknown) => void; + let reject!: (reason: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +function resetCache() { + const state = mining as unknown as Record; + state.poolsStatsCache = null; + state.poolsStatsRebuild = null; + state.poolsStatsGeneration = 0; + state.poolsStatsSyncedGeneration = -1; +} + +describe('Mining pools stats cache', () => { + beforeEach(() => { + jest.clearAllMocks(); + resetCache(); + mockBlockCount.mockResolvedValue(0); + mockGetNetworkHashPs.mockResolvedValue(0); + }); + + test('serves the cached snapshot without querying again', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + + const first = await mining.$getPoolsStats('24h'); + const second = await mining.$getPoolsStats('24h'); + + expect(first.blockCount).toBe(1); + expect(second.blockCount).toBe(1); + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(1); + }); + + test('collapses concurrent cold-start requests into one rebuild', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + + await Promise.all([ + mining.$getPoolsStats('24h'), + mining.$getPoolsStats('all'), + mining.$getPoolsStats('24h'), + ]); + + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(1); + }); + + test('does not rebuild while the cache is clean', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + await mining.$getPoolsStats('24h'); + + await mining.$getPoolsStats('24h'); + await mining.$getPoolsStats('all'); + + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(1); + }); + + test('rebuilds on the next read once marked dirty', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + await mining.$getPoolsStats('24h'); + + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(2)); + mining.markPoolsStatsDirty(); + + // the read is served from the old snapshot, and kicks off the rebuild in the background + expect((await mining.$getPoolsStats('24h')).blockCount).toBe(1); + await new Promise(process.nextTick); + + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(2); + expect((await mining.$getPoolsStats('24h')).blockCount).toBe(2); + }); + + test('an invalidation during a rebuild is not lost', async () => { + const inFlight = deferred(); + mockGetPoolsInfoPerInterval.mockReturnValueOnce(inFlight.promise); + + const rebuilding = mining.$rebuildPoolsStatsCache(); + + // a reorg lands while the first query is still running against pre-reorg rows + mining.markPoolsStatsDirty(); + + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(2)); + inFlight.resolve(poolsInfo(1)); + await rebuilding; + await new Promise(process.nextTick); + + // the superseded result must not be left behind as the current snapshot + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(2); + expect((await mining.$getPoolsStats('24h')).blockCount).toBe(2); + }); + + test('keeps serving the previous snapshot when a rebuild fails', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + await mining.$getPoolsStats('24h'); + + mockGetPoolsInfoPerInterval.mockRejectedValue(new Error('db is down')); + await mining.$rebuildPoolsStatsCache(); + + expect((await mining.$getPoolsStats('24h')).blockCount).toBe(1); + }); + + test('a failed rebuild does not retry itself in a loop', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + await mining.$getPoolsStats('24h'); + + mockGetPoolsInfoPerInterval.mockRejectedValue(new Error('db is down')); + await mining.$rebuildPoolsStatsCache(); + const callsAfterFailure = mockGetPoolsInfoPerInterval.mock.calls.length; + + await new Promise(process.nextTick); + await new Promise(process.nextTick); + + expect(mockGetPoolsInfoPerInterval).toHaveBeenCalledTimes(callsAfterFailure); + }); + + test('a failed rebuild leaves the cache dirty so the next read retries', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + await mining.$getPoolsStats('24h'); + + mockGetPoolsInfoPerInterval.mockRejectedValue(new Error('db is down')); + await mining.$rebuildPoolsStatsCache(); + + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(2)); + await mining.$getPoolsStats('24h'); + await new Promise(process.nextTick); + + expect((await mining.$getPoolsStats('24h')).blockCount).toBe(2); + }); + + test('falls back to the all interval for an unrecognised one', async () => { + mockGetPoolsInfoPerInterval.mockResolvedValue(poolsInfo(1)); + + const stats = await mining.$getPoolsStats('not-an-interval'); + + expect(stats).toBeDefined(); + expect(stats.blockCount).toBe(1); + }); +}); diff --git a/backend/src/api/blocks.ts b/backend/src/api/blocks.ts index be247f571..88c74b2e1 100644 --- a/backend/src/api/blocks.ts +++ b/backend/src/api/blocks.ts @@ -1218,6 +1218,7 @@ class Blocks { if (Common.indexingEnabled()) { await blocksRepository.$saveBlockInDatabase(blockExtended); this.updateTimerProgress(timer, `saved ${this.currentBlockHeight} to database`); + indexer.scheduleSingleTask('poolsStats', 30000); await AccelerationRepository.$indexAccelerationsForBlock( blockExtended, @@ -1410,6 +1411,7 @@ class Blocks { await cpfpRepository.$deleteClustersFrom(forkTail.height); await AccelerationRepository.$deleteAccelerationsFrom(forkTail.height); chainTips.clearOrphanCacheAboveHeight(forkTail.height); + void mining.$rebuildPoolsStatsCache(); this.updateTimerProgress(timer, `deleted stale block data`); this.blocks = newBlocks.reverse(); diff --git a/backend/src/api/mining/mining-routes.ts b/backend/src/api/mining/mining-routes.ts index 47a6b8f0a..b8325a567 100644 --- a/backend/src/api/mining/mining-routes.ts +++ b/backend/src/api/mining/mining-routes.ts @@ -150,7 +150,7 @@ class MiningRoutes { private async $getPoolsHistoricalHashrate(req: Request, res: Response) { try { - const hashrates = await HashratesRepository.$getPoolsWeeklyHashrate(req.params.interval); + const hashrates = await mining.$getPoolsHistoricalHashrate(req.params.interval); const blockCount = await BlocksRepository.$blockCount(null, null); res.header('Pragma', 'public'); res.header('Cache-control', 'public'); diff --git a/backend/src/api/mining/mining.ts b/backend/src/api/mining/mining.ts index ebf8fd40d..0a79d459e 100644 --- a/backend/src/api/mining/mining.ts +++ b/backend/src/api/mining/mining.ts @@ -1,6 +1,6 @@ import { BlockPrice, PoolInfo, PoolStats, RewardStats } from '../../mempool.interfaces'; import BlocksRepository from '../../repositories/BlocksRepository'; -import PoolsRepository from '../../repositories/PoolsRepository'; +import PoolsRepository, { POOLS_STATS_INTERVALS } from '../../repositories/PoolsRepository'; import HashratesRepository from '../../repositories/HashratesRepository'; import bitcoinClient from '../bitcoin/bitcoin-client'; import logger from '../../logger'; @@ -22,6 +22,14 @@ interface DifficultyBlock { difficulty: number, } +interface PoolsStats { + pools: PoolStats[], + blockCount: number, + lastEstimatedHashrate: number, + lastEstimatedHashrate3d: number, + lastEstimatedHashrate1w: number, +} + class Mining { private blocksPriceIndexingRunning = false; public lastHashrateIndexingDate: number | null = null; @@ -30,6 +38,22 @@ class Mining { public reindexHashrateRequested = false; public reindexDifficultyAdjustmentRequested = false; + // Rebuilt in the background on new blocks/reorgs/pool changes; snapshots older than this get a lazy refresh. + // The resolved snapshot and the in-flight rebuild are kept apart so requests never wait on a rebuild, + // and so a failed rebuild leaves the previous snapshot in place instead of emptying the cache. + private static readonly POOLS_STATS_RESYNC_MS = 600000; + private poolsStatsCache: { syncedAt: number, byInterval: Record } | null = null; + private poolsStatsRebuild: Promise> | null = null; + // Bumped by every event that changes the underlying rows, and recorded on the snapshot a rebuild + // publishes. The two drifting apart is what marks the cache dirty: a rebuild that started before an + // invalidation queried stale rows, so its result must not be mistaken for an up to date snapshot. + private poolsStatsGeneration = 0; + private poolsStatsSyncedGeneration = -1; + // Unlike the pools stats snapshot, this one is dropped outright on invalidation: it is rebuilt per + // interval on demand, so there is no single snapshot worth keeping around while it is stale. + private poolsHistoricalHashrateCache: Map = new Map(); + private poolsHistoricalHashrateRebuilds: Map> = new Map(); + private genesisData: { timestamp: number, bits: number, @@ -109,52 +133,221 @@ class Mining { /** * Generate high level overview of the pool ranks and general stats + * + * Only rejects on a cold start with no snapshot to fall back on + * + * @asyncUnsafe */ - public async $getPoolsStats(interval: string | null): Promise { - const poolsStatistics = {}; + public async $getPoolsStats(interval: string | null): Promise { + const cacheKey = (interval && Common.getSqlInterval(interval)) ? interval : 'all'; - const poolsInfo: PoolInfo[] = await PoolsRepository.$getPoolsInfo(interval); + const cached = this.poolsStatsCache; + if (cached) { + // the periodic resync is a backstop for writes that never mark the cache dirty, most notably the + // audit rows that land after a block is indexed (see $rebuildPoolsStatsCache) + if (this.isPoolsStatsDirty() || Date.now() - cached.syncedAt >= Mining.POOLS_STATS_RESYNC_MS) { + // a read never invalidates, so this refreshes against the current generation rather than bumping it + void this.$refreshPoolsStatsQuietly(); + } + // stale data beats waiting on a rebuild, and beats a 500 if that rebuild fails + return cached.byInterval[cacheKey]; + } - const poolsStats: PoolStats[] = []; - let rank = 1; - let blockCount = 0; + // nothing cached yet, so this request has to wait for the first build + return (await this.$refreshPoolsStats())[cacheKey]; + } - poolsInfo.forEach((poolInfo: PoolInfo) => { - const poolStat: PoolStats = { - poolId: poolInfo.poolId, // mysql row id - name: poolInfo.name, - link: poolInfo.link, - blockCount: poolInfo.blockCount, - rank: rank++, - emptyBlocks: poolInfo.emptyBlocks, - slug: poolInfo.slug, - avgMatchRate: poolInfo.avgMatchRate !== null ? Math.round(100 * poolInfo.avgMatchRate) / 100 : null, - avgFeeDelta: poolInfo.avgFeeDelta, - poolUniqueId: poolInfo.poolUniqueId - }; - poolsStats.push(poolStat); - blockCount += poolInfo.blockCount; + private isPoolsStatsDirty(): boolean { + return this.poolsStatsSyncedGeneration !== this.poolsStatsGeneration; + } + + /** + * Marks the cached pools stats as no longer reflecting the database. + * + * Note this deliberately keeps the previous snapshot: dropping it would send the next request down + * the cold path to wait on a full rebuild, which is what the cache exists to avoid. + */ + public markPoolsStatsDirty(): void { + this.poolsStatsGeneration++; + } + + /** + * Marks the cache dirty and rebuilds it. + * + * Block audits are written after the block itself, and their expected_fees later still, so a rebuild + * triggered by a new block is deliberately delayed rather than immediate — see the call in blocks.ts. + * Anything those writes still miss is picked up by the periodic resync. + * + * @asyncSafe + */ + public $rebuildPoolsStatsCache(): Promise { + this.markPoolsStatsDirty(); + return this.$refreshPoolsStatsQuietly(); + } + + /** @asyncSafe */ + private $refreshPoolsStatsQuietly(): Promise { + return this.$refreshPoolsStats().then(() => undefined, (e) => { + logger.err(`Failed to build pools stats cache. Reason: ${(e instanceof Error ? e.message : e)}`, logger.tags.mining); + }); + } + + /** + * Runs one rebuild at a time, and only replaces the cached snapshot once the new one resolves. + * Callers must handle the rejection. + */ + private $refreshPoolsStats(): Promise> { + if (this.poolsStatsRebuild) { + return this.poolsStatsRebuild; + } + + // read before the query runs, so an invalidation raced against it leaves the cache dirty + const generation = this.poolsStatsGeneration; + const rebuild = this.$queryAllPoolsStats().then((byInterval) => { + this.poolsStatsCache = { syncedAt: Date.now(), byInterval }; + this.poolsStatsSyncedGeneration = generation; + return byInterval; + }); + this.poolsStatsRebuild = rebuild; + + // release the slot either way, so a failed rebuild does not block the next attempt + const releaseSlot = (): void => { + if (this.poolsStatsRebuild === rebuild) { + this.poolsStatsRebuild = null; + } + }; + rebuild.then(() => { + releaseSlot(); + // invalidated while this was in flight, so it queried rows that are already superseded + if (this.isPoolsStatsDirty()) { + void this.$refreshPoolsStatsQuietly(); + } + }, () => { + // no retry here: a failed rebuild leaves the cache dirty, and retrying on that would spin against + // an unreachable database. The periodic resync and the next invalidation both still apply. + releaseSlot(); }); - poolsStatistics['pools'] = poolsStats; - poolsStatistics['blockCount'] = blockCount; + return rebuild; + } + private async $queryAllPoolsStats(): Promise> { + const poolsInfoPerInterval: Record = await PoolsRepository.$getPoolsInfoPerInterval(); + const estimatedHashrates = await this.$getEstimatedHashrates(); + + const statsByInterval: Record = {}; + for (const interval of POOLS_STATS_INTERVALS) { + let rank = 1; + let blockCount = 0; + + const poolStats: PoolStats[] = []; + poolsInfoPerInterval[interval].forEach((poolInfo) => { + poolStats.push({ + poolId: poolInfo.poolId, // mysql row id + name: poolInfo.name, + link: poolInfo.link, + blockCount: poolInfo.blockCount, + rank: rank++, + emptyBlocks: poolInfo.emptyBlocks, + slug: poolInfo.slug, + avgMatchRate: poolInfo.avgMatchRate !== null ? Math.round(100 * poolInfo.avgMatchRate) / 100 : null, + avgFeeDelta: poolInfo.avgFeeDelta, + poolUniqueId: poolInfo.poolUniqueId + }); + blockCount += poolInfo.blockCount; + }); + + statsByInterval[interval] = { pools: poolStats, blockCount: blockCount, ...estimatedHashrates }; + } + + return statsByInterval; + } + + /** + * Estimated network hashrate over the last `blockCount` blocks. + * Core rejects a lookup of 0 blocks, and one failed window must not zero the others. + * + * @asyncSafe + */ + private async $getEstimatedHashrate(blockCount: number): Promise { + if (blockCount <= 0) { + return 0; + } + + try { + return await bitcoinClient.getNetworkHashPs(blockCount); + } catch (e) { + logger.debug(`Bitcoin Core is not available, using zeroed value for current hashrate over ${blockCount} blocks`, logger.tags.mining); + return 0; + } + } + + /** @asyncSafe */ + private async $getEstimatedHashrates(): Promise<{ lastEstimatedHashrate: number, lastEstimatedHashrate3d: number, lastEstimatedHashrate1w: number }> { const totalBlock24h: number = await BlocksRepository.$blockCount(null, '24h'); const totalBlock3d: number = await BlocksRepository.$blockCount(null, '3d'); const totalBlock1w: number = await BlocksRepository.$blockCount(null, '1w'); - try { - poolsStatistics['lastEstimatedHashrate'] = await bitcoinClient.getNetworkHashPs(totalBlock24h); - poolsStatistics['lastEstimatedHashrate3d'] = await bitcoinClient.getNetworkHashPs(totalBlock3d); - poolsStatistics['lastEstimatedHashrate1w'] = await bitcoinClient.getNetworkHashPs(totalBlock1w); - } catch (e) { - poolsStatistics['lastEstimatedHashrate'] = 0; - poolsStatistics['lastEstimatedHashrate3d'] = 0; - poolsStatistics['lastEstimatedHashrate1w'] = 0; - logger.debug('Bitcoin Core is not available, using zeroed value for current hashrate', logger.tags.mining); + return { + lastEstimatedHashrate: await this.$getEstimatedHashrate(totalBlock24h), + lastEstimatedHashrate3d: await this.$getEstimatedHashrate(totalBlock3d), + lastEstimatedHashrate1w: await this.$getEstimatedHashrate(totalBlock1w), + }; + } + + /** + * Get weekly hashrate history for all pools + */ + public async $getPoolsHistoricalHashrate(interval: string | null): Promise { + const cacheKey = Common.getSqlInterval(interval) ?? 'all'; + + const cached = this.poolsHistoricalHashrateCache.get(cacheKey); + if (cached) { + if (Date.now() - cached.syncedAt >= Mining.POOLS_STATS_RESYNC_MS) { + this.$refreshPoolsHistoricalHashrate(cacheKey, interval).catch((e) => { + logger.err(`Failed to refresh pools historical hashrate. Reason: ${(e instanceof Error ? e.message : e)}`, logger.tags.mining); + }); + } + return cached.hashrates; } - return poolsStatistics; + return this.$refreshPoolsHistoricalHashrate(cacheKey, interval); + } + + /** + * Runs one rebuild at a time per interval, and only replaces the cached snapshot once it resolves. + * Callers must handle the rejection. + */ + private $refreshPoolsHistoricalHashrate(cacheKey: string, interval: string | null): Promise { + const running = this.poolsHistoricalHashrateRebuilds.get(cacheKey); + if (running) { + return running; + } + + const rebuild = HashratesRepository.$getPoolsWeeklyHashrate(interval).then((hashrates) => { + // an invalidation while this was in flight means the result is already stale, so drop it + if (this.poolsHistoricalHashrateRebuilds.get(cacheKey) === rebuild) { + this.poolsHistoricalHashrateCache.set(cacheKey, { syncedAt: Date.now(), hashrates }); + } + return hashrates; + }); + this.poolsHistoricalHashrateRebuilds.set(cacheKey, rebuild); + + // release the slot either way, so a failed rebuild does not block the next attempt + const releaseSlot = (): void => { + if (this.poolsHistoricalHashrateRebuilds.get(cacheKey) === rebuild) { + this.poolsHistoricalHashrateRebuilds.delete(cacheKey); + } + }; + rebuild.then(releaseSlot, releaseSlot); + + return rebuild; + } + + public invalidatePoolsHistoricalHashrateCache(): void { + this.poolsHistoricalHashrateCache.clear(); + // in-flight rebuilds started before the invalidation must not write their result back + this.poolsHistoricalHashrateRebuilds.clear(); } /** @@ -312,6 +505,7 @@ class Mining { } this.lastWeeklyHashrateIndexingDate = new Date().getUTCDate(); if (newlyIndexed > 0) { + this.invalidatePoolsHistoricalHashrateCache(); logger.info(`Weekly mining pools hashrates indexing completed: indexed ${newlyIndexed} weeks`, logger.tags.mining); } else { logger.debug(`Weekly mining pools hashrates indexing completed: indexed ${newlyIndexed} weeks`, logger.tags.mining); diff --git a/backend/src/api/pools-parser.ts b/backend/src/api/pools-parser.ts index 92c8e919d..cc147b3b7 100644 --- a/backend/src/api/pools-parser.ts +++ b/backend/src/api/pools-parser.ts @@ -129,6 +129,7 @@ class PoolsParser { // update persistent cache with the reindexed data void diskCache.$saveCacheToDisk(); void redisCache.$updateBlocks(blocks.getBlocks()); + void mining.$rebuildPoolsStatsCache(); } } diff --git a/backend/src/index.ts b/backend/src/index.ts index 04690c4bc..cebca0835 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -220,6 +220,7 @@ class Server { if (config.MEMPOOL.ENABLED) { void this.runMainUpdateLoop(); + indexer.scheduleSingleTask('poolsStats', 0); } setInterval(() => { this.healthCheck(); }, 2500); diff --git a/backend/src/indexer.ts b/backend/src/indexer.ts index ff10d701e..7618361aa 100644 --- a/backend/src/indexer.ts +++ b/backend/src/indexer.ts @@ -19,7 +19,7 @@ export interface CoreIndex { best_block_height: number; } -type TaskName = 'blocksPrices' | 'coinStatsIndex'; +type TaskName = 'blocksPrices' | 'coinStatsIndex' | 'poolsStats'; class Indexer { private runIndexer = true; @@ -162,6 +162,15 @@ class Indexer { logger.debug(`failed to index coinstatsindex: ` + (e instanceof Error ? e.message : e)); } } break; + + case 'poolsStats': { + logger.debug('Syncing pools stats'); + try { + await mining.$rebuildPoolsStatsCache(); + } catch (e) { + logger.debug('failed to sync pools stats cache: ' + (e instanceof Error ? e.message : e)); + } + } break; } this.tasksRunning[task] = false; diff --git a/backend/src/repositories/HashratesRepository.ts b/backend/src/repositories/HashratesRepository.ts index b1297bfda..256d6f7a5 100644 --- a/backend/src/repositories/HashratesRepository.ts +++ b/backend/src/repositories/HashratesRepository.ts @@ -211,6 +211,7 @@ class HashratesRepository { // Re-run the hashrate indexing to fill up missing data mining.lastHashrateIndexingDate = null; mining.lastWeeklyHashrateIndexingDate = null; + mining.invalidatePoolsHistoricalHashrateCache(); } catch (e) { logger.err('Cannot delete latest hashrates data points. Reason: ' + (e instanceof Error ? e.message : e), logger.tags.mining); } @@ -227,6 +228,7 @@ class HashratesRepository { // Re-run the hashrate indexing to fill up missing data mining.lastHashrateIndexingDate = null; mining.lastWeeklyHashrateIndexingDate = null; + mining.invalidatePoolsHistoricalHashrateCache(); } catch (e) { logger.err('Cannot delete latest hashrates data points. Reason: ' + (e instanceof Error ? e.message : e), logger.tags.mining); } diff --git a/backend/src/repositories/PoolsRepository.ts b/backend/src/repositories/PoolsRepository.ts index 4fc3742fd..c8826cf8b 100644 --- a/backend/src/repositories/PoolsRepository.ts +++ b/backend/src/repositories/PoolsRepository.ts @@ -5,6 +5,9 @@ import DB from '../database'; import logger from '../logger'; import { PoolInfo, PoolTag } from '../mempool.interfaces'; +// Intervals the pools stats cache is built for. 'all' has no time filter; the rest resolve via Common.getSqlInterval +export const POOLS_STATS_INTERVALS = ['24h', '3d', '1w', '1m', '3m', '6m', '1y', '2y', '3y', '4y', 'all']; + class PoolsRepository { /** * Get all pools tagging info @@ -68,6 +71,72 @@ class PoolsRepository { } } + /** @asyncSafe */ + public async $getPoolsInfoPerInterval(): Promise> { + const feeDelta = `(CAST(blocks.fees as SIGNED) - CAST(blocks_audits.expected_fees as SIGNED)) / NULLIF(CAST(blocks_audits.expected_fees as SIGNED), 0)`; + const columns = POOLS_STATS_INTERVALS.map((label) => { + const sql = Common.getSqlInterval(label); + const inWindow = sql ? `blocks.blockTimestamp BETWEEN DATE_SUB(NOW(), INTERVAL ${sql}) AND NOW()` : '1'; + return ` + COUNT(CASE WHEN ${inWindow} THEN blocks.height END) AS \`blockCount_${label}\`, + COUNT(CASE WHEN ${inWindow} AND blocks.tx_count = 1 THEN 1 END) AS \`emptyBlocks_${label}\`, + AVG(CASE WHEN ${inWindow} THEN blocks_audits.match_rate END) AS \`avgMatchRate_${label}\`, + AVG(CASE WHEN ${inWindow} THEN ${feeDelta} END) AS \`avgFeeDelta_${label}\``; + }).join(','); + + const query = `SELECT + pool_id AS poolId, + pools.name AS name, + pools.link AS link, + pools.slug AS slug, + pools.unique_id AS poolUniqueId, + ${columns} + FROM blocks + JOIN pools on pools.id = pool_id + LEFT JOIN blocks_audits ON blocks_audits.hash = blocks.hash + WHERE blocks.stale = 0 + GROUP BY pool_id`; + + try { + const [rows]: any[] = await DB.query(query); + + // every interval needs an array even with no rows, or callers iterate undefined + const result: Record = {}; + for (const label of POOLS_STATS_INTERVALS) { + result[label] = []; + } + + for (const row of rows) { + for (const label of POOLS_STATS_INTERVALS) { + const blockCount = row[`blockCount_${label}`]; + if (blockCount > 0) { + result[label].push({ + poolId: row.poolId, + name: row.name, + link: row.link, + slug: row.slug, + poolUniqueId: row.poolUniqueId, + blockCount: blockCount, + emptyBlocks: row[`emptyBlocks_${label}`], + avgMatchRate: row[`avgMatchRate_${label}`], + avgFeeDelta: row[`avgFeeDelta_${label}`], + }); + } + } + } + + // rank is assigned from this order, and unique_id breaks ties so it stays stable across rebuilds + for (const pools of Object.values(result)) { + pools.sort((a, b) => b.blockCount - a.blockCount || a.poolUniqueId - b.poolUniqueId); + } + + return result; + } catch(e) { + logger.err(`Cannot generate pools stats per interval. Reason: ` + (e instanceof Error ? e.message : e)); + throw e; + } + } + /** * Get basic pool info and block count between two timestamp * @asyncSafe