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

Add network and schema versioning to redis cache

Browse Source
This commit is contained in:
Mononaut 2023-06-16 19:00:52 -04:00
parent d65bddd30b
commit a9f8bbbcce
No known key found for this signature in database
GPG Key ID: A3F058E41374C04E
1 changed files with 26 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
backend/src/api/redis-cache.ts
View File

@ -6,9 +6,18 @@ import config from '../config';
import { BlockExtended, BlockSummary, MempoolTransactionExtended } from '../mempool.interfaces'; import { BlockExtended, BlockSummary, MempoolTransactionExtended } from '../mempool.interfaces';
import rbfCache from './rbf-cache'; import rbfCache from './rbf-cache';
enum NetworkDB {
mainnet = 0,
testnet,
signet,
liquid,
liquidtestnet,
}
class RedisCache { class RedisCache {
private client; private client;
private connected = false; private connected = false;
private schemaVersion = 1;
private cacheQueue: MempoolTransactionExtended[] = []; private cacheQueue: MempoolTransactionExtended[] = [];
private txFlushLimit: number = 1000; private txFlushLimit: number = 1000;
@ -18,7 +27,8 @@ class RedisCache {
const redisConfig = { const redisConfig = {
socket: { socket: {
path: config.REDIS.UNIX_SOCKET_PATH path: config.REDIS.UNIX_SOCKET_PATH
} },
database: NetworkDB[config.MEMPOOL.NETWORK],
}; };
this.client = createClient(redisConfig); this.client = createClient(redisConfig);
this.client.on('error', (e) => { this.client.on('error', (e) => {
@ -30,9 +40,16 @@ class RedisCache {
private async $ensureConnected(): Promise<void> { private async $ensureConnected(): Promise<void> {
if (!this.connected && config.REDIS.ENABLED) { if (!this.connected && config.REDIS.ENABLED) {
return this.client.connect().then(() => { return this.client.connect().then(async () => {
this.connected = true; this.connected = true;
logger.info(`Redis client connected`); logger.info(`Redis client connected`);
const version = await this.client.get('schema_version');
if (version !== this.schemaVersion) {
// schema changed
// perform migrations or flush DB if necessary
logger.info(`Redis schema version changed from ${version} to ${this.schemaVersion}`);
await this.client.set('schema_version', this.schemaVersion);
}
}); });
} }
} }
@ -63,18 +80,22 @@ class RedisCache {
} }
async $flushTransactions() { async $flushTransactions() {
await this.$addTransactions(this.cacheQueue); const success = await this.$addTransactions(this.cacheQueue);
this.cacheQueue = []; if (success) {
this.cacheQueue = [];
}
} }
async $addTransactions(newTransactions: MempoolTransactionExtended[]) { async $addTransactions(newTransactions: MempoolTransactionExtended[]): Promise<boolean> {
try { try {
await this.$ensureConnected(); await this.$ensureConnected();
await Promise.all(newTransactions.map(tx => { await Promise.all(newTransactions.map(tx => {
return this.client.json.set('tx:' + tx.txid, '$', tx); return this.client.json.set('tx:' + tx.txid, '$', tx);
})); }));
return true;
} catch (e) { } catch (e) {
logger.warn(`Failed to add ${newTransactions.length} transactions to Redis cache: ${e instanceof Error ? e.message : e}`); logger.warn(`Failed to add ${newTransactions.length} transactions to Redis cache: ${e instanceof Error ? e.message : e}`);
return false;
} }
} }