mirror of
https://github.com/mempool/mempool.git
synced 2026-08-13 12:33:11 +02:00
mining: decouple pools stats from API traffic
This commit is contained in:
jramos0
parent
29afce710d
commit
d199f0e5ab
11 changed files with 676 additions and 40 deletions
|
|
@ -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)));
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
185
backend/src/__tests__/api/mining-pools-stats-cache.test.ts
Normal file
185
backend/src/__tests__/api/mining-pools-stats-cache.test.ts
Normal file
|
|
@ -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<string, unknown>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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<string, PoolsStats> } | null = null;
|
||||
private poolsStatsRebuild: Promise<Record<string, PoolsStats>> | 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<string, { syncedAt: number, hashrates: any[] }> = new Map();
|
||||
private poolsHistoricalHashrateRebuilds: Map<string, Promise<any[]>> = 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<object> {
|
||||
const poolsStatistics = {};
|
||||
public async $getPoolsStats(interval: string | null): Promise<PoolsStats> {
|
||||
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<void> {
|
||||
this.markPoolsStatsDirty();
|
||||
return this.$refreshPoolsStatsQuietly();
|
||||
}
|
||||
|
||||
/** @asyncSafe */
|
||||
private $refreshPoolsStatsQuietly(): Promise<void> {
|
||||
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<Record<string, PoolsStats>> {
|
||||
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<Record<string, PoolsStats>> {
|
||||
const poolsInfoPerInterval: Record<string, PoolInfo[]> = await PoolsRepository.$getPoolsInfoPerInterval();
|
||||
const estimatedHashrates = await this.$getEstimatedHashrates();
|
||||
|
||||
const statsByInterval: Record<string, PoolsStats> = {};
|
||||
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<number> {
|
||||
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<any[]> {
|
||||
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<any[]> {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -129,6 +129,7 @@ class PoolsParser {
|
|||
// update persistent cache with the reindexed data
|
||||
void diskCache.$saveCacheToDisk();
|
||||
void redisCache.$updateBlocks(blocks.getBlocks());
|
||||
void mining.$rebuildPoolsStatsCache();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -220,6 +220,7 @@ class Server {
|
|||
|
||||
if (config.MEMPOOL.ENABLED) {
|
||||
void this.runMainUpdateLoop();
|
||||
indexer.scheduleSingleTask('poolsStats', 0);
|
||||
}
|
||||
|
||||
setInterval(() => { this.healthCheck(); }, 2500);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Record<string, PoolInfo[]>> {
|
||||
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<string, PoolInfo[]> = {};
|
||||
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue