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

Merge pull request #4632 from mempool/mononaut/redis-error-handling

Browse Source 
Handle Redis errors and disconnects
This commit is contained in:
softsimon 2024-02-19 22:18:24 +07:00 committed by GitHub
commit 035068a72e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 196 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -19,45 +19,90 @@ class RedisCache {
private client; private client;
private connected = false; private connected = false;
private schemaVersion = 1; private schemaVersion = 1;
private redisConfig: any;
private pauseFlush: boolean = false;
private cacheQueue: MempoolTransactionExtended[] = []; private cacheQueue: MempoolTransactionExtended[] = [];
private removeQueue: string[] = [];
private rbfCacheQueue: { type: string, txid: string, value: any }[] = [];
private rbfRemoveQueue: { type: string, txid: string }[] = [];
private txFlushLimit: number = 10000; private txFlushLimit: number = 10000;
constructor() { constructor() {
if (config.REDIS.ENABLED) { if (config.REDIS.ENABLED) {
const redisConfig = { this.redisConfig = {
socket: { socket: {
path: config.REDIS.UNIX_SOCKET_PATH path: config.REDIS.UNIX_SOCKET_PATH
}, },
database: NetworkDB[config.MEMPOOL.NETWORK], database: NetworkDB[config.MEMPOOL.NETWORK],
}; };
this.client = createClient(redisConfig);
this.client.on('error', (e) => {
logger.err(`Error in Redis client: ${e instanceof Error ? e.message : e}`);
});
this.$ensureConnected(); this.$ensureConnected();
setInterval(() => { this.$ensureConnected(); }, 10000);
} }
} }
private async $ensureConnected(): Promise<void> { private async $ensureConnected(): Promise<boolean> {
if (!this.connected && config.REDIS.ENABLED) { if (!this.connected && config.REDIS.ENABLED) {
return this.client.connect().then(async () => { try {
this.connected = true; this.client = createClient(this.redisConfig);
logger.info(`Redis client connected`); this.client.on('error', async (e) => {
const version = await this.client.get('schema_version'); logger.err(`Error in Redis client: ${e instanceof Error ? e.message : e}`);
if (version !== this.schemaVersion) { this.connected = false;
// schema changed await this.client.disconnect();
// perform migrations or flush DB if necessary });
logger.info(`Redis schema version changed from ${version} to ${this.schemaVersion}`); await this.client.connect().then(async () => {
await this.client.set('schema_version', this.schemaVersion); try {
} const version = await this.client.get('schema_version');
}); this.connected = true;
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);
}
logger.info(`Redis client connected`);
return true;
} catch (e) {
this.connected = false;
logger.warn('Failed to connect to Redis');
return false;
}
});
await this.$onConnected();
return true;
} catch (e) {
logger.warn('Error connecting to Redis: ' + (e instanceof Error ? e.message : e));
return false;
}
} else {
try {
// test connection
await this.client.get('schema_version');
return true;
} catch (e) {
logger.warn('Lost connection to Redis: ' + (e instanceof Error ? e.message : e));
logger.warn('Attempting to reconnect in 10 seconds');
this.connected = false;
return false;
}
} }
} }
async $updateBlocks(blocks: BlockExtended[]) { private async $onConnected(): Promise<void> {
await this.$flushTransactions();
await this.$removeTransactions([]);
await this.$flushRbfQueues();
}
async $updateBlocks(blocks: BlockExtended[]): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
if (!this.connected) {
logger.warn(`Failed to update blocks in Redis cache: Redis is not connected`);
return;
}
try { try {
await this.$ensureConnected();
await this.client.set('blocks', JSON.stringify(blocks)); await this.client.set('blocks', JSON.stringify(blocks));
logger.debug(`Saved latest blocks to Redis cache`); logger.debug(`Saved latest blocks to Redis cache`);
} catch (e) { } catch (e) {
@ -65,9 +110,15 @@ class RedisCache {
} }
} }
async $updateBlockSummaries(summaries: BlockSummary[]) { async $updateBlockSummaries(summaries: BlockSummary[]): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
if (!this.connected) {
logger.warn(`Failed to update block summaries in Redis cache: Redis is not connected`);
return;
}
try { try {
await this.$ensureConnected();
await this.client.set('block-summaries', JSON.stringify(summaries)); await this.client.set('block-summaries', JSON.stringify(summaries));
logger.debug(`Saved latest block summaries to Redis cache`); logger.debug(`Saved latest block summaries to Redis cache`);
} catch (e) { } catch (e) {
@ -75,30 +126,35 @@ class RedisCache {
} }
} }
async $addTransaction(tx: MempoolTransactionExtended) { async $addTransaction(tx: MempoolTransactionExtended): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
this.cacheQueue.push(tx); this.cacheQueue.push(tx);
if (this.cacheQueue.length >= this.txFlushLimit) { if (this.cacheQueue.length >= this.txFlushLimit) {
await this.$flushTransactions(); if (!this.pauseFlush) {
await this.$flushTransactions();
}
} }
} }
async $flushTransactions() { async $flushTransactions(): Promise<void> {
const success = await this.$addTransactions(this.cacheQueue); if (!config.REDIS.ENABLED) {
if (success) { return;
logger.debug(`Saved ${this.cacheQueue.length} transactions to Redis cache`); }
this.cacheQueue = []; if (!this.cacheQueue.length) {
} else { return;
logger.err(`Failed to save ${this.cacheQueue.length} transactions to Redis cache`); }
if (!this.connected) {
logger.warn(`Failed to add ${this.cacheQueue.length} transactions to Redis cache: Redis not connected`);
return;
} }
}
private async $addTransactions(newTransactions: MempoolTransactionExtended[]): Promise<boolean> { this.pauseFlush = false;
if (!newTransactions.length) {
return true; const toAdd = this.cacheQueue.slice(0, this.txFlushLimit);
}
try { try {
await this.$ensureConnected(); const msetData = toAdd.map(tx => {
const msetData = newTransactions.map(tx => {
const minified: any = { ...tx }; const minified: any = { ...tx };
delete minified.hex; delete minified.hex;
for (const vin of minified.vin) { for (const vin of minified.vin) {
@ -112,30 +168,53 @@ class RedisCache {
return [`mempool:tx:${tx.txid}`, JSON.stringify(minified)]; return [`mempool:tx:${tx.txid}`, JSON.stringify(minified)];
}); });
await this.client.MSET(msetData); await this.client.MSET(msetData);
return true; // successful, remove transactions from cache queue
this.cacheQueue = this.cacheQueue.slice(toAdd.length);
logger.debug(`Saved ${toAdd.length} transactions to Redis cache, ${this.cacheQueue.length} left in queue`);
} 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 ${toAdd.length} transactions to Redis cache: ${e instanceof Error ? e.message : e}`);
return false; this.pauseFlush = true;
} }
} }
async $removeTransactions(transactions: string[]) { async $removeTransactions(transactions: string[]): Promise<void> {
try { if (!config.REDIS.ENABLED) {
await this.$ensureConnected(); return;
}
const toRemove = this.removeQueue.concat(transactions);
this.removeQueue = [];
let failed: string[] = [];
let numRemoved = 0;
if (this.connected) {
const sliceLength = config.REDIS.BATCH_QUERY_BASE_SIZE; const sliceLength = config.REDIS.BATCH_QUERY_BASE_SIZE;
for (let i = 0; i < Math.ceil(transactions.length / sliceLength); i++) { for (let i = 0; i < Math.ceil(toRemove.length / sliceLength); i++) {
const slice = transactions.slice(i * sliceLength, (i + 1) * sliceLength); const slice = toRemove.slice(i * sliceLength, (i + 1) * sliceLength);
await this.client.unlink(slice.map(txid => `mempool:tx:${txid}`)); try {
logger.debug(`Deleted ${slice.length} transactions from the Redis cache`); await this.client.unlink(slice.map(txid => `mempool:tx:${txid}`));
numRemoved+= sliceLength;
logger.debug(`Deleted ${slice.length} transactions from the Redis cache`);
} catch (e) {
logger.warn(`Failed to remove ${slice.length} transactions from Redis cache: ${e instanceof Error ? e.message : e}`);
failed = failed.concat(slice);
}
} }
} catch (e) { // concat instead of replace, in case more txs have been added in the meantime
logger.warn(`Failed to remove ${transactions.length} transactions from Redis cache: ${e instanceof Error ? e.message : e}`); this.removeQueue = this.removeQueue.concat(failed);
} else {
this.removeQueue = this.removeQueue.concat(toRemove);
} }
} }
async $setRbfEntry(type: string, txid: string, value: any): Promise<void> { async $setRbfEntry(type: string, txid: string, value: any): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
if (!this.connected) {
this.rbfCacheQueue.push({ type, txid, value });
logger.warn(`Failed to set RBF ${type} in Redis cache: Redis is not connected`);
return;
}
try { try {
await this.$ensureConnected();
await this.client.set(`rbf:${type}:${txid}`, JSON.stringify(value)); await this.client.set(`rbf:${type}:${txid}`, JSON.stringify(value));
} catch (e) { } catch (e) {
logger.warn(`Failed to set RBF ${type} in Redis cache: ${e instanceof Error ? e.message : e}`); logger.warn(`Failed to set RBF ${type} in Redis cache: ${e instanceof Error ? e.message : e}`);
@ -143,17 +222,55 @@ class RedisCache {
} }
async $removeRbfEntry(type: string, txid: string): Promise<void> { async $removeRbfEntry(type: string, txid: string): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
if (!this.connected) {
this.rbfRemoveQueue.push({ type, txid });
logger.warn(`Failed to remove RBF ${type} from Redis cache: Redis is not connected`);
return;
}
try { try {
await this.$ensureConnected();
await this.client.unlink(`rbf:${type}:${txid}`); await this.client.unlink(`rbf:${type}:${txid}`);
} catch (e) { } catch (e) {
logger.warn(`Failed to remove RBF ${type} from Redis cache: ${e instanceof Error ? e.message : e}`); logger.warn(`Failed to remove RBF ${type} from Redis cache: ${e instanceof Error ? e.message : e}`);
} }
} }
async $getBlocks(): Promise<BlockExtended[]> { private async $flushRbfQueues(): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
if (!this.connected) {
return;
}
try {
const toAdd = this.rbfCacheQueue;
this.rbfCacheQueue = [];
for (const { type, txid, value } of toAdd) {
await this.$setRbfEntry(type, txid, value);
}
logger.debug(`Saved ${toAdd.length} queued RBF entries to the Redis cache`);
const toRemove = this.rbfRemoveQueue;
this.rbfRemoveQueue = [];
for (const { type, txid } of toRemove) {
await this.$removeRbfEntry(type, txid);
}
logger.debug(`Removed ${toRemove.length} queued RBF entries from the Redis cache`);
} catch (e) {
logger.warn(`Failed to flush RBF cache event queues after reconnecting to Redis: ${e instanceof Error ? e.message : e}`);
}
}
async $getBlocks(): Promise<BlockExtended[]> {
if (!config.REDIS.ENABLED) {
return [];
}
if (!this.connected) {
logger.warn(`Failed to retrieve blocks from Redis cache: Redis is not connected`);
return [];
}
try { try {
await this.$ensureConnected();
const json = await this.client.get('blocks'); const json = await this.client.get('blocks');
return JSON.parse(json); return JSON.parse(json);
} catch (e) { } catch (e) {
@ -163,8 +280,14 @@ class RedisCache {
} }
async $getBlockSummaries(): Promise<BlockSummary[]> { async $getBlockSummaries(): Promise<BlockSummary[]> {
if (!config.REDIS.ENABLED) {
return [];
}
if (!this.connected) {
logger.warn(`Failed to retrieve blocks from Redis cache: Redis is not connected`);
return [];
}
try { try {
await this.$ensureConnected();
const json = await this.client.get('block-summaries'); const json = await this.client.get('block-summaries');
return JSON.parse(json); return JSON.parse(json);
} catch (e) { } catch (e) {
@ -174,10 +297,16 @@ class RedisCache {
} }
async $getMempool(): Promise<{ [txid: string]: MempoolTransactionExtended }> { async $getMempool(): Promise<{ [txid: string]: MempoolTransactionExtended }> {
if (!config.REDIS.ENABLED) {
return {};
}
if (!this.connected) {
logger.warn(`Failed to retrieve mempool from Redis cache: Redis is not connected`);
return {};
}
const start = Date.now(); const start = Date.now();
const mempool = {}; const mempool = {};
try { try {
await this.$ensureConnected();
const mempoolList = await this.scanKeys<MempoolTransactionExtended>('mempool:tx:*'); const mempoolList = await this.scanKeys<MempoolTransactionExtended>('mempool:tx:*');
for (const tx of mempoolList) { for (const tx of mempoolList) {
mempool[tx.key] = tx.value; mempool[tx.key] = tx.value;
@ -191,8 +320,14 @@ class RedisCache {
} }
async $getRbfEntries(type: string): Promise<any[]> { async $getRbfEntries(type: string): Promise<any[]> {
if (!config.REDIS.ENABLED) {
return [];
}
if (!this.connected) {
logger.warn(`Failed to retrieve Rbf ${type}s from Redis cache: Redis is not connected`);
return [];
}
try { try {
await this.$ensureConnected();
const rbfEntries = await this.scanKeys<MempoolTransactionExtended[]>(`rbf:${type}:*`); const rbfEntries = await this.scanKeys<MempoolTransactionExtended[]>(`rbf:${type}:*`);
return rbfEntries; return rbfEntries;
} catch (e) { } catch (e) {
@ -201,7 +336,10 @@ class RedisCache {
} }
} }
async $loadCache() { async $loadCache(): Promise<void> {
if (!config.REDIS.ENABLED) {
return;
}
logger.info('Restoring mempool and blocks data from Redis cache'); logger.info('Restoring mempool and blocks data from Redis cache');
// Load block data // Load block data
const loadedBlocks = await this.$getBlocks(); const loadedBlocks = await this.$getBlocks();
@ -226,7 +364,7 @@ class RedisCache {
}); });
} }
private inflateLoadedTxs(mempool: { [txid: string]: MempoolTransactionExtended }) { private inflateLoadedTxs(mempool: { [txid: string]: MempoolTransactionExtended }): void {
for (const tx of Object.values(mempool)) { for (const tx of Object.values(mempool)) {
for (const vin of tx.vin) { for (const vin of tx.vin) {
if (vin.scriptsig) { if (vin.scriptsig) {