feat: Implement Redis storage (#4438)

Signed-off-by: Konstantina Blazhukova <[email protected]>

Author: konstantinabl

packages/relay/src/lib/services/transactionPoolService/RedisPendingTransactionStorage.ts

@@ -0,0 +1,107 @@
+// SPDX-License-Identifier: Apache-2.0
+import { RedisClientType } from 'redis';
+
+import { AddToListResult, PendingTransactionStorage } from '../../types/transactionPool';
+
+export class RedisPendingTransactionStorage implements PendingTransactionStorage {
+  /**
+   * Number of elements to scan per step.
+   */
+  private elementsPerStep = 500;
+
+  /**
+   * Prefix used to namespace all keys managed by this storage.
+   *
+   * @remarks
+   * Using a prefix allows efficient scanning and cleanup of related keys
+   */
+  private readonly keyPrefix = 'pending:';
+
+  /**
+   * Creates a new Redis-backed pending transaction storage.
+   *
+   * @param redisClient - A connected {@link RedisClientType} instance.
+   */
+  constructor(private readonly redisClient: RedisClientType) {}
+
+  /**
+   * Resolves the Redis key for a given address.
+   *
+   * @param addr - Account address whose pending list key should be derived.
+   * @returns The Redis key (e.g., `pending:<address>`).
+   */
+  private keyFor(address: string): string {
+    return `${this.keyPrefix}${address}`;
+  }
+
+  /**
+   * Appends a transaction hash to the pending list for the provided address.
+   *
+   * @remarks
+   * This uses Redis `RPUSH`, which is atomic. The integer result from Redis is
+   * the new length of the list after the append.
+   *
+   * Duplicate values are not prevented by this method.
+   *
+   * @param addr - Account address whose pending list will be appended to.
+   * @param txHash - Transaction hash to append.
+   * @returns Result indicating success and the new list length.
+   */
+  async addToList(address: string, txHash: string): Promise<AddToListResult> {
+    const key = this.keyFor(address);
+    await this.redisClient.sAdd(key, txHash);
+    const newLen = await this.redisClient.sCard(key);
+
+    return { ok: true, newValue: newLen };
+  }
+
+  /**
+   * Removes a transaction hash from the pending list of the given address.
+   *
+   * @param address - Account address whose pending list should be modified.
+   * @param txHash - Transaction hash to remove from the list.
+   * @returns The updated number of pending transactions for the address.
+   */
+  async removeFromList(address: string, txHash: string): Promise<number> {
+    const key = this.keyFor(address);
+    await this.redisClient.sRem(key, txHash);
+
+    return await this.redisClient.sCard(key);
+  }
+
+  /**
+   * Removes all keys managed by this storage (all `pending:*`).
+   *
+   * @remarks
+   * Iterates keys using `SCAN` via `scanIterator` to avoid blocking Redis, and
+   * batches deletions using a pipeline for efficiency.
+   */
+  async removeAll(): Promise<void> {
+    let pipeline = this.redisClient.multi();
+    let batched = 0;
+    for await (const key of this.redisClient.scanIterator({
+      MATCH: `${this.keyPrefix}*`,
+      COUNT: this.elementsPerStep,
+    })) {
+      pipeline.del(key);
+      batched++;
+      if (batched % 100 === 0) {
+        await pipeline.execAsPipeline();
+        pipeline = this.redisClient.multi();
+      }
+    }
+    await pipeline.execAsPipeline();
+  }
+
+  /**
+   * Retrieves the number of pending transactions for a given address.
+   *
+   * @param addr - Account address to query.
+   * @returns The current pending count (0 if the list does not exist).
+   */
+  async getList(address: string): Promise<number> {
+    const key = this.keyFor(address);
+
+    return await this.redisClient.sCard(key);
+  }
+}

packages/relay/tests/lib/services/nonceManagement/RedisPendingTransactionStorage.spec.ts

@@ -0,0 +1,118 @@
+// SPDX-License-Identifier: Apache-2.0
+
+import chai, { expect } from 'chai';
+import chaiAsPromised from 'chai-as-promised';
+import { pino } from 'pino';
+import { createClient, RedisClientType } from 'redis';
+
+import { RedisPendingTransactionStorage } from '../../../../src/lib/services/transactionPoolService/RedisPendingTransactionStorage';
+import { useInMemoryRedisServer } from '../../../helpers';
+
+chai.use(chaiAsPromised);
+
+describe('RedisPendingTransactionStorage Test Suite', function () {
+  this.timeout(10000);
+
+  const logger = pino({ level: 'silent' });
+
+  let redisClient: RedisClientType;
+  let storage: RedisPendingTransactionStorage;
+
+  useInMemoryRedisServer(logger, 6390);
+
+  before(async () => {
+    redisClient = createClient({ url: 'redis://127.0.0.1:6390' });
+    await redisClient.connect();
+    storage = new RedisPendingTransactionStorage(redisClient);
+  });
+
+  beforeEach(async () => {
+    await redisClient.flushAll();
+  });
+
+  const addr1 = '0x1111111111111111111111111111111111111111';
+  const addr2 = '0x2222222222222222222222222222222222222222';
+  const tx1 = '0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';
+  const tx2 = '0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb';
+  const tx3 = '0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc';
+
+  describe('addToList (Set-based)', () => {
+    it('adds first transaction and returns size 1', async () => {
+      const res = await storage.addToList(addr1, tx1);
+      expect(res.ok).to.equal(true);
+      if (res.ok) expect(res.newValue).to.equal(1);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(1);
+    });
+
+    it('deduplicates the same transaction hash', async () => {
+      await storage.addToList(addr1, tx1);
+      const res = await storage.addToList(addr1, tx1);
+      expect(res.ok).to.equal(true);
+      if (res.ok) expect(res.newValue).to.equal(1);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(1);
+    });
+
+    it('adds multiple distinct tx hashes and returns correct size', async () => {
+      await storage.addToList(addr1, tx1);
+      const r2 = await storage.addToList(addr1, tx2);
+      expect(r2.ok).to.equal(true);
+      if (r2.ok) expect(r2.newValue).to.equal(2);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(2);
+    });
+  });
+
+  describe('getList (Set-based)', () => {
+    it('returns 0 for empty/non-existent key', async () => {
+      const count = await storage.getList(addr2);
+      expect(count).to.equal(0);
+    });
+
+    it('returns size after multiple adds', async () => {
+      await storage.addToList(addr1, tx1);
+      await storage.addToList(addr1, tx2);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(2);
+    });
+  });
+
+  describe('removeFromList (Set-based)', () => {
+    it('removes existing tx and returns new size', async () => {
+      await storage.addToList(addr1, tx1);
+      await storage.addToList(addr1, tx2);
+      const remaining = await storage.removeFromList(addr1, tx1);
+      expect(remaining).to.equal(1);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(1);
+    });
+
+    it('is idempotent when removing non-existent tx', async () => {
+      await storage.addToList(addr1, tx1);
+      const remaining = await storage.removeFromList(addr1, tx2);
+      expect(remaining).to.equal(1);
+      const count = await storage.getList(addr1);
+      expect(count).to.equal(1);
+    });
+  });
+
+  describe('removeAll', () => {
+    after(async () => {
+      await redisClient.quit();
+    });
+
+    it('deletes all pending:* keys', async () => {
+      await storage.addToList(addr1, tx1);
+      await storage.addToList(addr1, tx2);
+      await storage.addToList(addr2, tx3);
+
+      await storage.removeAll();
+
+      const c1 = await storage.getList(addr1);
+      const c2 = await storage.getList(addr2);
+      expect(c1).to.equal(0);
+      expect(c2).to.equal(0);
+    });
+  });
+});