machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Redis cache for block and mempool data

Browse Source
This commit is contained in:
Mononaut
2023-05-12 16:29:45 -06:00
parent ad08ca8bc8
commit da77fa81c7
10 changed files with 358 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/src/__fixtures__/mempool-config.template.json
View File

@@ -10,6 +10,7 @@
"AUTOMATIC_BLOCK_REINDEXING": false,
"POLL_RATE_MS": 3,
"CACHE_DIR": "__MEMPOOL_CACHE_DIR__",
"CACHE_ENABLED": true,
"CLEAR_PROTECTION_MINUTES": 4,
"RECOMMENDED_FEE_PERCENTILE": 5,
"BLOCK_WEIGHT_UNITS": 6,
@@ -127,5 +128,9 @@
"AUDIT": false,
"AUDIT_START_HEIGHT": 774000,
"SERVERS": []
},
"REDIS": {
"ENABLED": false,
"UNIX_SOCKET_PATH": "/tmp/redis.sock"
}
}

8
backend/src/__tests__/config.test.ts
View File

@@ -23,6 +23,7 @@ describe('Mempool Backend Config', () => {
AUTOMATIC_BLOCK_REINDEXING: false,
POLL_RATE_MS: 2000,
CACHE_DIR: './cache',
CACHE_ENABLED: true,
CLEAR_PROTECTION_MINUTES: 20,
RECOMMENDED_FEE_PERCENTILE: 50,
BLOCK_WEIGHT_UNITS: 4000000,
@@ -127,6 +128,11 @@ describe('Mempool Backend Config', () => {
AUDIT_START_HEIGHT: 774000,
SERVERS: []
});
expect(config.REDIS).toStrictEqual({
ENABLED: false,
UNIX_SOCKET_PATH: ''
});
});
});
@@ -160,6 +166,8 @@ describe('Mempool Backend Config', () => {
expect(config.PRICE_DATA_SERVER).toStrictEqual(fixture.PRICE_DATA_SERVER);
expect(config.EXTERNAL_DATA_SERVER).toStrictEqual(fixture.EXTERNAL_DATA_SERVER);
expect(config.REDIS).toStrictEqual(fixture.REDIS);
});
});

12
backend/src/api/blocks.ts
View File

@@ -26,6 +26,8 @@ import PricesRepository from '../repositories/PricesRepository';
import priceUpdater from '../tasks/price-updater';
import chainTips from './chain-tips';
import websocketHandler from './websocket-handler';
import redisCache from './redis-cache';
import rbfCache from './rbf-cache';
class Blocks {
private blocks: BlockExtended[] = [];
@@ -804,10 +806,18 @@ class Blocks {
if (this.newBlockCallbacks.length) {
this.newBlockCallbacks.forEach((cb) => cb(blockExtended, txIds, transactions));
}
if (!memPool.hasPriority() && (block.height % config.MEMPOOL.DISK_CACHE_BLOCK_INTERVAL === 0)) {
if (config.MEMPOOL.CACHE_ENABLED && !memPool.hasPriority() && (block.height % config.MEMPOOL.DISK_CACHE_BLOCK_INTERVAL === 0)) {
diskCache.$saveCacheToDisk();
}
// Update Redis cache
if (config.REDIS.ENABLED) {
await redisCache.$updateBlocks(this.blocks);
await redisCache.$updateBlockSummaries(this.blockSummaries);
await redisCache.$removeTransactions(txIds);
await rbfCache.updateCache();
}
handledBlocks++;
}

6
backend/src/api/disk-cache.ts
View File

@@ -29,7 +29,7 @@ class DiskCache {
};
constructor() {
if (!cluster.isPrimary) {
if (!cluster.isPrimary || !config.MEMPOOL.CACHE_ENABLED) {
return;
}
process.on('SIGINT', (e) => {
@@ -39,7 +39,7 @@ class DiskCache {
}
async $saveCacheToDisk(sync: boolean = false): Promise<void> {
if (!cluster.isPrimary) {
if (!cluster.isPrimary || !config.MEMPOOL.CACHE_ENABLED) {
return;
}
if (this.isWritingCache) {
@@ -175,7 +175,7 @@ class DiskCache {
}
async $loadMempoolCache(): Promise<void> {
if (!fs.existsSync(DiskCache.FILE_NAME)) {
if (!config.MEMPOOL.CACHE_ENABLED || !fs.existsSync(DiskCache.FILE_NAME)) {
return;
}
try {

12
backend/src/api/mempool.ts
View File

@@ -9,7 +9,7 @@ import loadingIndicators from './loading-indicators';
import bitcoinClient from './bitcoin/bitcoin-client';
import bitcoinSecondClient from './bitcoin/bitcoin-second-client';
import rbfCache from './rbf-cache';
import { IEsploraApi } from './bitcoin/esplora-api.interface';
import redisCache from './redis-cache';
class Mempool {
private inSync: boolean = false;
@@ -102,6 +102,10 @@ class Mempool {
await this.$asyncMempoolChangedCallback(this.mempoolCache, count, [], []);
}
this.addToSpendMap(Object.values(this.mempoolCache));
if (config.MEMPOOL.CACHE_ENABLED && config.REDIS.ENABLED) {
logger.debug('copying mempool from disk cache into Redis');
await redisCache.$addTransactions(Object.values(mempoolData));
}
}
public async $reloadMempool(expectedCount: number): Promise<MempoolTransactionExtended[]> {
@@ -318,6 +322,12 @@ class Mempool {
loadingIndicators.setProgress('mempool', 100);
}
// Update Redis cache
if (config.REDIS.ENABLED) {
await redisCache.$addTransactions(newTransactions);
await redisCache.$removeTransactions(deletedTransactions.map(tx => tx.txid));
}
const end = new Date().getTime();
const time = end - start;
logger.debug(`Mempool updated in ${time / 1000} seconds. New size: ${Object.keys(this.mempoolCache).length} (${diff > 0 ? '+' + diff : diff})`);

140
backend/src/api/redis-cache.ts Normal file
View File

@@ -0,0 +1,140 @@
import { createClient } from 'redis';
import memPool from './mempool';
import blocks from './blocks';
import logger from '../logger';
import config from '../config';
import { BlockExtended, BlockSummary, TransactionExtended } from '../mempool.interfaces';
class RedisCache {
private client;
private connected = false;
constructor() {
if (config.REDIS.ENABLED) {
const redisConfig = {
socket: {
path: config.REDIS.UNIX_SOCKET_PATH
}
};
this.client = createClient(redisConfig);
this.client.on('error', (e) => {
logger.err(`Error in Redis client: ${e instanceof Error ? e.message : e}`);
});
this.$ensureConnected();
}
}
private async $ensureConnected(): Promise<void> {
if (!this.connected && config.REDIS.ENABLED) {
return this.client.connect().then(() => {
this.connected = true;
logger.info(`Redis client connected`);
});
}
}
async $updateBlocks(blocks: BlockExtended[]) {
try {
await this.$ensureConnected();
await this.client.json.set('blocks', '$', blocks);
} catch (e) {
logger.warn(`Failed to update blocks in Redis cache: ${e instanceof Error ? e.message : e}`);
}
}
async $updateBlockSummaries(summaries: BlockSummary[]) {
try {
await this.$ensureConnected();
await this.client.json.set('block-summaries', '$', summaries);
} catch (e) {
logger.warn(`Failed to update blocks in Redis cache: ${e instanceof Error ? e.message : e}`);
}
}
async $addTransactions(newTransactions: TransactionExtended[]) {
try {
await this.$ensureConnected();
await Promise.all(newTransactions.map(tx => {
return this.client.json.set('tx:' + tx.txid, '$', tx);
}));
} catch (e) {
logger.warn(`Failed to add ${newTransactions.length} transactions to Redis cache: ${e instanceof Error ? e.message : e}`);
}
}
async $removeTransactions(transactions: string[]) {
try {
await this.$ensureConnected();
await Promise.all(transactions.map(txid => {
return this.client.del('tx:' + txid);
}));
} catch (e) {
logger.warn(`Failed to remove ${transactions.length} transactions from Redis cache: ${e instanceof Error ? e.message : e}`);
}
}
async $getBlocks(): Promise<BlockExtended[]> {
try {
await this.$ensureConnected();
return this.client.json.get('blocks');
} catch (e) {
logger.warn(`Failed to retrieve blocks from Redis cache: ${e instanceof Error ? e.message : e}`);
return [];
}
}
async $getBlockSummaries(): Promise<BlockSummary[]> {
try {
await this.$ensureConnected();
return this.client.json.get('block-summaries');
} catch (e) {
logger.warn(`Failed to retrieve blocks from Redis cache: ${e instanceof Error ? e.message : e}`);
return [];
}
}
async $getMempool(): Promise<{ [txid: string]: TransactionExtended }> {
const mempool = {};
try {
await this.$ensureConnected();
const keys = await this.client.keys('tx:*');
const promises: Promise<TransactionExtended[]>[] = [];
for (let i = 0; i < keys.length; i += 10000) {
const keySlice = keys.slice(i, i + 10000);
if (!keySlice.length) {
continue;
}
promises.push(this.client.json.mGet(keySlice, '$').then(chunk => {
for (const txs of chunk) {
for (const tx of txs) {
if (tx) {
mempool[tx.txid] = tx;
}
}
}
}));
}
await Promise.all(promises);
} catch (e) {
logger.warn(`Failed to retrieve mempool from Redis cache: ${e instanceof Error ? e.message : e}`);
}
return mempool;
}
async $loadCache() {
logger.info('Restoring mempool and blocks data from Redis cache');
// Load block data
const loadedBlocks = await this.$getBlocks();
const loadedBlockSummaries = await this.$getBlockSummaries();
// Load mempool
const loadedMempool = await this.$getMempool();
// Set loaded data
blocks.setBlocks(loadedBlocks || []);
blocks.setBlockSummaries(loadedBlockSummaries || []);
await memPool.$setMempool(loadedMempool);
}
}
export default new RedisCache();

16
backend/src/config.ts
View File

@@ -12,6 +12,7 @@ interface IConfig {
API_URL_PREFIX: string;
POLL_RATE_MS: number;
CACHE_DIR: string;
CACHE_ENABLED: boolean;
CLEAR_PROTECTION_MINUTES: number;
RECOMMENDED_FEE_PERCENTILE: number;
BLOCK_WEIGHT_UNITS: number;
@@ -137,7 +138,11 @@ interface IConfig {
AUDIT: boolean;
AUDIT_START_HEIGHT: number;
SERVERS: string[];
}
},
REDIS: {
ENABLED: boolean;
UNIX_SOCKET_PATH: string;
},
}
const defaults: IConfig = {
@@ -150,6 +155,7 @@ const defaults: IConfig = {
'API_URL_PREFIX': '/api/v1/',
'POLL_RATE_MS': 2000,
'CACHE_DIR': './cache',
'CACHE_ENABLED': true,
'CLEAR_PROTECTION_MINUTES': 20,
'RECOMMENDED_FEE_PERCENTILE': 50,
'BLOCK_WEIGHT_UNITS': 4000000,
@@ -275,7 +281,11 @@ const defaults: IConfig = {
'AUDIT': false,
'AUDIT_START_HEIGHT': 774000,
'SERVERS': [],
}
},
'REDIS': {
'ENABLED': false,
'UNIX_SOCKET_PATH': '',
},
};
class Config implements IConfig {
@@ -296,6 +306,7 @@ class Config implements IConfig {
EXTERNAL_DATA_SERVER: IConfig['EXTERNAL_DATA_SERVER'];
MAXMIND: IConfig['MAXMIND'];
REPLICATION: IConfig['REPLICATION'];
REDIS: IConfig['REDIS'];
constructor() {
const configs = this.merge(configFromFile, defaults);
@@ -316,6 +327,7 @@ class Config implements IConfig {
this.EXTERNAL_DATA_SERVER = configs.EXTERNAL_DATA_SERVER;
this.MAXMIND = configs.MAXMIND;
this.REPLICATION = configs.REPLICATION;
this.REDIS = configs.REDIS;
}
merge = (...objects: object[]): IConfig => {

7
backend/src/index.ts
View File

@@ -41,6 +41,7 @@ import chainTips from './api/chain-tips';
import { AxiosError } from 'axios';
import v8 from 'v8';
import { formatBytes, getBytesUnit } from './utils/format';
import redisCache from './api/redis-cache';
class Server {
private wss: WebSocket.Server | undefined;
@@ -122,7 +123,11 @@ class Server {
await poolsUpdater.updatePoolsJson(); // Needs to be done before loading the disk cache because we sometimes wipe it
await syncAssets.syncAssets$();
if (config.MEMPOOL.ENABLED) {
await diskCache.$loadMempoolCache();
if (config.MEMPOOL.CACHE_ENABLED) {
await diskCache.$loadMempoolCache();
} else if (config.REDIS.ENABLED) {
await redisCache.$loadCache();
}
}
if (config.STATISTICS.ENABLED && config.DATABASE.ENABLED && cluster.isPrimary) {