fix: mempool seen txids cache

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "groundcontrol",
-  "version": "3.1.2",
+  "version": "3.1.3",
   "lockfileVersion": 2,
   "requires": true,
   "packages": {
     "": {
       "name": "groundcontrol",
-      "version": "3.1.2",
+      "version": "3.1.3",
       "dependencies": {
         "body-parser": "^1.20.5",
         "cors": "^2.8.5",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "groundcontrol",
-  "version": "3.1.2",
+  "version": "3.1.3",
   "description": "GroundControl push server API",
   "devDependencies": {
     "@types/node": "18.7.16",

src/lru-cache.test.ts

@@ -0,0 +1,61 @@
+import { describe, expect, it } from "vitest";
+import { LruCache } from "./lru-cache";
+
+describe("LruCache", () => {
+  it("evicts the oldest txid when the cache exceeds its maximum size", () => {
+    const cache = new LruCache(2);
+
+    cache.add("tx-1");
+    cache.add("tx-2");
+    cache.add("tx-3");
+
+    expect(cache.size).toBe(2);
+    expect(cache.has("tx-1")).toBe(false);
+    expect(cache.has("tx-2")).toBe(true);
+    expect(cache.has("tx-3")).toBe(true);
+  });
+
+  it("refreshes recently seen txids before evicting", () => {
+    const cache = new LruCache(2);
+
+    cache.add("tx-1");
+    cache.add("tx-2");
+    expect(cache.has("tx-1")).toBe(true);
+    cache.add("tx-3");
+
+    expect(cache.has("tx-1")).toBe(true);
+    expect(cache.has("tx-2")).toBe(false);
+    expect(cache.has("tx-3")).toBe(true);
+  });
+
+  it("evicts in least-recently-used order across many interleaved adds and reads", () => {
+    const cache = new LruCache(3);
+
+    cache.add("a");
+    cache.add("b");
+    cache.add("c");
+
+    cache.has("a"); // refresh a -> order: b, c, a
+    cache.add("d"); // evict oldest (b)
+    expect(cache.has("b")).toBe(false);
+
+    // order now: c, a, d
+    cache.add("e"); // evict oldest (c)
+    expect(cache.has("c")).toBe(false);
+    expect(cache.has("a")).toBe(true);
+    expect(cache.has("d")).toBe(true);
+    expect(cache.has("e")).toBe(true);
+    expect(cache.size).toBe(3);
+  });
+
+  it("stays bounded and correct under heavy churn", () => {
+    const max = 1000;
+    const cache = new LruCache(max);
+    for (let i = 0; i < 50000; i++) cache.add("tx-" + i);
+
+    expect(cache.size).toBe(max);
+    expect(cache.has("tx-0")).toBe(false);
+    expect(cache.has("tx-49999")).toBe(true);
+    expect(cache.has("tx-49000")).toBe(true);
+  });
+});

src/lru-cache.ts

@@ -0,0 +1,46 @@
+export class LruCache {
+  private readonly entries = new Map<string, true>();
+  // Persistent forward iterator used for eviction. Creating a fresh
+  // `entries.keys().next()` on every eviction is O(n) in the number of
+  // tombstones left at the head of the Map by prior deletes, which degrades
+  // to ~O(n^2) under churn. Advancing a single long-lived iterator instead
+  // keeps eviction amortized O(1) and is the only thing that scaled in the
+  // mempool-sized benchmark (~270ms vs ~20s for 300k evictions).
+  private evictionCursor = this.entries.keys();
+
+  constructor(private readonly maxSize: number) {
+    if (maxSize < 1) throw new Error("maxSize must be greater than zero");
+  }
+
+  get size() {
+    return this.entries.size;
+  }
+
+  has(key: string) {
+    if (!this.entries.has(key)) return false;
+
+    this.entries.delete(key);
+    this.entries.set(key, true);
+    return true;
+  }
+
+  add(key: string) {
+    if (!key) return;
+
+    if (this.entries.has(key)) this.entries.delete(key);
+    this.entries.set(key, true);
+
+    while (this.entries.size > this.maxSize) this.evictOldest();
+  }
+
+  private evictOldest() {
+    let next = this.evictionCursor.next();
+    if (next.done) {
+      // Cursor caught up to the tail (or was created on an empty map); restart
+      // it from the current oldest entry.
+      this.evictionCursor = this.entries.keys();
+      next = this.evictionCursor.next();
+    }
+    if (!next.done) this.entries.delete(next.value);
+  }
+}

src/worker-processmempool.ts

@@ -4,13 +4,14 @@ import { TokenToAddress } from "./entity/TokenToAddress";
 import { SendQueue } from "./entity/SendQueue";
 import dataSource from "./data-source";
 import { components } from "./openapi/api";
+import { LruCache } from "./lru-cache";
 require("dotenv").config();
 const url = require("url");
 let jayson = require("jayson/promise");
 let rpc = url.parse(process.env.BITCOIN_RPC);
 let client = jayson.client.http(rpc);
 
-let processedTxids = {};
+const processedTxids = new LruCache(250000);
 if (!process.env.BITCOIN_RPC) {
   console.error("not all env variables set");
   process.exit();
@@ -29,7 +30,7 @@ process
 let sendQueueRepository: Repository<SendQueue>;
 
 async function processMempool() {
-  process.env.VERBOSE && console.log("cached txids=", Object.keys(processedTxids).length);
+  process.env.VERBOSE && console.log("cached txids=", processedTxids.size);
   const responseGetrawmempool = await client.request("getrawmempool", []);
   process.env.VERBOSE && console.log(responseGetrawmempool.result.length, "txs in mempool");
 
@@ -42,7 +43,7 @@ async function processMempool() {
   for (const txid of responseGetrawmempool.result) {
     countTxidsProcessed++;
     if (!txid) continue;
-    if (!processedTxids[txid]) rpcBatch.push(client.request("getrawtransaction", [txid, true], undefined, false));
+    if (!processedTxids.has(txid)) rpcBatch.push(client.request("getrawtransaction", [txid, true], undefined, false));
     if (rpcBatch.length >= batchSize || countTxidsProcessed === responseGetrawmempool.result.length) {
       const startBatch = +new Date();
       // got enough txids lets batch fetch them from bitcoind rpc
@@ -53,7 +54,7 @@ async function processMempool() {
             if (output.scriptPubKey && (output.scriptPubKey.addresses || output.scriptPubKey.address)) {
               for (const address of output.scriptPubKey?.addresses ?? (output.scriptPubKey?.address ? [output.scriptPubKey?.address] : [])) {
                 addresses.push(address);
-                processedTxids[response.result.txid] = true;
+                processedTxids.add(response.result.txid);
                 const payload: components["schemas"]["PushNotificationOnchainAddressGotUnconfirmedTransaction"] = {
                   address,
                   txid: response.result.txid,