refactor: add fifo set

Author: alexghr

yarn-project/foundation/package.json

@@ -62,6 +62,7 @@
     "./eth-signature": "./dest/eth-signature/index.js",
     "./queue": "./dest/queue/index.js",
     "./fifo": "./dest/fifo/index.js",
+    "./fifo-set": "./dest/fifo_set/index.js",
     "./fs": "./dest/fs/index.js",
     "./buffer": "./dest/buffer/index.js",
     "./json-rpc": "./dest/json-rpc/index.js",

yarn-project/foundation/src/fifo_set/fifo_set.test.ts

@@ -0,0 +1,62 @@
+import { FifoSet } from './fifo_set.js';
+
+describe('FifoSet', () => {
+  it('keeps entries up to the limit', () => {
+    const set = FifoSet.withLimit<string>(3);
+
+    set.add('a');
+    set.add('b');
+    set.add('c');
+
+    expect([...set]).toEqual(['a', 'b', 'c']);
+    expect(set.size).toBe(3);
+    expect(set.limit).toBe(3);
+  });
+
+  it('evicts the oldest entry when adding past the limit', () => {
+    const set = FifoSet.withLimit<string>(2);
+
+    set.add('a');
+    set.add('b');
+    set.add('c');
+
+    expect([...set]).toEqual(['b', 'c']);
+  });
+
+  it('compacts initial values to the newest entries', () => {
+    const set = FifoSet.withLimit(2, ['a', 'b', 'c']);
+
+    expect([...set]).toEqual(['b', 'c']);
+  });
+
+  it('does not evict when adding a duplicate', () => {
+    const set = FifoSet.withLimit(2, ['a', 'b']);
+
+    set.add('a');
+
+    expect([...set]).toEqual(['a', 'b']);
+  });
+
+  it('adds absent values and reports whether the set changed', () => {
+    const set = FifoSet.withLimit(2, ['a']);
+
+    expect(set.addIfAbsent('a')).toBe(false);
+    expect(set.addIfAbsent('b')).toBe(true);
+    expect(set.addIfAbsent('c')).toBe(true);
+    expect([...set]).toEqual(['b', 'c']);
+  });
+
+  it('can evict undefined values', () => {
+    const set = FifoSet.withLimit<string | undefined>(1, [undefined, 'a']);
+
+    expect([...set]).toEqual(['a']);
+
+    set.add(undefined);
+
+    expect([...set]).toEqual([undefined]);
+  });
+
+  it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY])('throws for invalid limit %s', limit => {
+    expect(() => FifoSet.withLimit(limit)).toThrow('FifoSet limit must be a positive safe integer');
+  });
+});

yarn-project/foundation/src/fifo_set/fifo_set.ts

@@ -0,0 +1,52 @@
+/** A Set capped to a fixed number of entries, evicting the oldest inserted value when full. */
+export class FifoSet<T> extends Set<T> {
+  private _limit: number | undefined;
+
+  private constructor(values?: Iterable<T>) {
+    super(values);
+  }
+
+  /** Creates a bounded set with a positive integer limit. */
+  static withLimit<T>(limit: number, values?: Iterable<T>): FifoSet<T> {
+    if (!Number.isSafeInteger(limit) || limit <= 0) {
+      throw new TypeError(`FifoSet limit must be a positive safe integer: ${limit}`);
+    }
+
+    const set = new FifoSet(values);
+    set._limit = limit;
+    set.compact();
+
+    return set;
+  }
+
+  override add(value: T): this {
+    super.add(value);
+    this.compact();
+    return this;
+  }
+
+  /** Maximum number of entries retained by this set. */
+  public get limit(): number {
+    return this._limit ?? Infinity;
+  }
+
+  /** Evicts oldest entries until the set is within its limit. */
+  public compact(): void {
+    while (this.size > this.limit) {
+      const head = this.values().next();
+      if (head.done) {
+        return;
+      }
+      this.delete(head.value);
+    }
+  }
+
+  /** Adds a value only if it is absent, returning whether the set changed. */
+  public addIfAbsent(value: T): boolean {
+    if (this.has(value)) {
+      return false;
+    }
+    this.add(value);
+    return true;
+  }
+}

yarn-project/foundation/src/fifo_set/index.ts

@@ -0,0 +1 @@
+export * from './fifo_set.js';

yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_impl.ts

@@ -1,4 +1,5 @@
 import { BlockNumber, SlotNumber } from '@aztec/foundation/branded-types';
+import { FifoSet } from '@aztec/foundation/fifo-set';
 import type { Logger } from '@aztec/foundation/log';
 import type { DateProvider } from '@aztec/foundation/timer';
 import type { AztecAsyncKVStore, AztecAsyncMap } from '@aztec/kv-store';
@@ -82,7 +83,7 @@ export class TxPoolV2Impl {
   #evictionManager: EvictionManager;
   #dateProvider: DateProvider;
   #instrumentation: TxPoolV2Instrumentation;
-  #evictedTxHashes: Set<string> = new Set();
+  #evictedTxHashes: FifoSet<string>;
   #log: Logger;
   #callbacks: TxPoolV2Callbacks;
 
@@ -105,6 +106,7 @@ export class TxPoolV2Impl {
     this.#checkAllowedSetupCalls = deps.checkAllowedSetupCalls;
 
     this.#config = { ...DEFAULT_TX_POOL_V2_CONFIG, ...config };
+    this.#evictedTxHashes = FifoSet.withLimit<string>(this.#config.evictedTxCacheSize);
     this.#archive = new TxArchive(archiveStore, this.#config.archivedTxLimit, log);
     this.#deletedPool = new DeletedPool(store, this.#txsDB, log);
     this.#dateProvider = dateProvider;
@@ -903,21 +905,11 @@ export class TxPoolV2Impl {
     this.#instrumentation.recordEvictions(txHashes.length, reason);
     for (const txHashStr of txHashes) {
       this.#log.debug(`Evicting tx ${txHashStr}`, { txHash: txHashStr, reason });
-      this.#addToEvictedCache(txHashStr);
+      this.#evictedTxHashes.add(txHashStr);
     }
     await this.#deleteTxsBatch(txHashes);
   }
 
-  /** Adds a tx hash to the bounded evicted cache, evicting the oldest entry if at capacity. */
-  #addToEvictedCache(txHashStr: string): void {
-    if (this.#evictedTxHashes.size >= this.#config.evictedTxCacheSize) {
-      // FIFO eviction: remove the first (oldest) entry
-      const oldest = this.#evictedTxHashes.values().next().value!;
-      this.#evictedTxHashes.delete(oldest);
-    }
-    this.#evictedTxHashes.add(txHashStr);
-  }
-
   // ============================================================================
   // PRIVATE HELPERS - Validation & Conflict Resolution
   // ============================================================================

yarn-project/p2p/src/services/tx_file_store/tx_file_store.ts

@@ -1,3 +1,4 @@
+import { FifoSet } from '@aztec/foundation/fifo-set';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import { RunningPromise } from '@aztec/foundation/promise';
 import { makeBackoff, retry } from '@aztec/foundation/retry';
@@ -10,6 +11,8 @@ import type { TxPoolV2 } from '../../mem_pools/index.js';
 import type { TxFileStoreConfig } from './config.js';
 import { TxFileStoreInstrumentation } from './instrumentation.js';
 
+const MAX_RECENT_UPLOADS = 1000;
+
 /**
  * Uploads validated transactions to a file store as a fallback retrieval mechanism.
  * Listens to TxPool txs-added events and uploads txs asynchronously with bounded concurrency.
@@ -21,9 +24,7 @@ export class TxFileStore {
   private readonly handleTxsAdded: (args: { txs: Tx[]; source?: string }) => void;
 
   /** Recently uploaded tx hashes for deduplication. */
-  private recentUploads: Set<string> = new Set();
-  private recentUploadsOrder: string[] = [];
-  private readonly maxRecentUploads = 1000;
+  private recentUploads = FifoSet.withLimit<string>(MAX_RECENT_UPLOADS);
 
   private constructor(
     private readonly fileStore: FileStore,
@@ -127,24 +128,11 @@ export class TxFileStore {
     const path = `${this.basePath}/txs/${txHash}.bin`;
     const timer = new Timer();
 
-    if (this.recentUploads.has(txHash)) {
+    if (!this.recentUploads.addIfAbsent(txHash)) {
       return;
     }
 
     try {
-      this.recentUploads.add(txHash);
-      this.recentUploadsOrder.push(txHash);
-
-      if (this.recentUploadsOrder.length > this.maxRecentUploads) {
-        // delete old entries in recentUploads
-        for (const txHashToRemove of this.recentUploadsOrder.splice(
-          0,
-          this.recentUploadsOrder.length - this.maxRecentUploads,
-        )) {
-          this.recentUploads.delete(txHashToRemove);
-        }
-      }
-
       await retry(
         () => this.fileStore.save(path, tx.toBuffer(), { compress: true }),
         `Uploading tx ${txHash}`,

yarn-project/slasher/src/watchers/attestations_block_watcher.ts

@@ -1,14 +1,14 @@
 import { EpochCache } from '@aztec/epoch-cache';
 import { SlotNumber } from '@aztec/foundation/branded-types';
 import { merge, pick } from '@aztec/foundation/collection';
+import { FifoSet } from '@aztec/foundation/fifo-set';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import {
   type InvalidCheckpointDetectedEvent,
   type L2BlockSourceEventEmitter,
   L2BlockSourceEvents,
   type ValidateCheckpointNegativeResult,
 } from '@aztec/stdlib/block';
-import type { CheckpointInfo } from '@aztec/stdlib/checkpoint';
 import { OffenseType } from '@aztec/stdlib/slashing';
 
 import EventEmitter from 'node:events';
@@ -20,6 +20,7 @@ const AttestationsBlockWatcherConfigKeys = [
   'slashAttestDescendantOfInvalidPenalty',
   'slashProposeInvalidAttestationsPenalty',
 ] as const;
+const MAX_INVALID_CHECKPOINTS = 100;
 
 type AttestationsBlockWatcherConfig = Pick<SlasherConfig, (typeof AttestationsBlockWatcherConfigKeys)[number]>;
 
@@ -32,11 +33,8 @@ type AttestationsBlockWatcherConfig = Pick<SlasherConfig, (typeof AttestationsBl
 export class AttestationsBlockWatcher extends (EventEmitter as new () => WatcherEmitter) implements Watcher {
   private log: Logger = createLogger('attestations-block-watcher');
 
-  // Only keep track of the last N invalid checkpoints
-  private maxInvalidCheckpoints = 100;
-
-  // All invalid archive roots seen
-  private invalidArchiveRoots: Set<string> = new Set();
+  // Recently seen invalid archive roots.
+  private invalidArchiveRoots = FifoSet.withLimit<string>(MAX_INVALID_CHECKPOINTS);
 
   private config: AttestationsBlockWatcherConfig;
 
@@ -98,8 +96,7 @@ export class AttestationsBlockWatcher extends (EventEmitter as new () => Watcher
       reason: validationResult.valid === false ? validationResult.reason : 'unknown',
     });
 
-    // Store the invalid checkpoint
-    this.addInvalidCheckpoint(event.validationResult.checkpoint);
+    this.invalidArchiveRoots.add(checkpoint.archive.toString());
 
     // Slash the proposer of the invalid checkpoint
     this.slashProposer(event.validationResult);
@@ -181,14 +178,4 @@ export class AttestationsBlockWatcher extends (EventEmitter as new () => Watcher
       }
     }
   }
-
-  private addInvalidCheckpoint(checkpoint: CheckpointInfo) {
-    this.invalidArchiveRoots.add(checkpoint.archive.toString());
-
-    // Prune old entries if we exceed the maximum
-    if (this.invalidArchiveRoots.size > this.maxInvalidCheckpoints) {
-      const oldestKey = this.invalidArchiveRoots.keys().next().value!;
-      this.invalidArchiveRoots.delete(oldestKey);
-    }
-  }
 }

yarn-project/slasher/src/watchers/broadcasted_invalid_checkpoint_proposal_watcher.ts

@@ -2,6 +2,7 @@ import type { EpochCacheInterface } from '@aztec/epoch-cache';
 import { SlotNumber } from '@aztec/foundation/branded-types';
 import { merge, pick } from '@aztec/foundation/collection';
 import type { EthAddress } from '@aztec/foundation/eth-address';
+import { FifoSet } from '@aztec/foundation/fifo-set';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import { RunningPromise } from '@aztec/foundation/running-promise';
 import type { P2PClient, SlasherConfig } from '@aztec/stdlib/interfaces/server';
@@ -39,7 +40,7 @@ export class BroadcastedInvalidCheckpointProposalWatcher
 {
   private readonly log: Logger = createLogger('broadcasted-invalid-checkpoint-proposal-watcher');
   private readonly runningPromise: RunningPromise;
-  private readonly emittedOffenses = new Set<string>();
+  private readonly emittedOffenses: FifoSet<string>;
   private readonly scanSlotLookback: number;
   private config: BroadcastedInvalidCheckpointProposalWatcherConfig;
   private lastScannedSlot: SlotNumber | undefined;
@@ -54,6 +55,12 @@ export class BroadcastedInvalidCheckpointProposalWatcher
     const constants = epochCache.getL1Constants();
     this.config = pick(config, ...BroadcastedInvalidCheckpointProposalWatcherConfigKeys);
     this.scanSlotLookback = Math.max(1, scanSlotLookback);
+
+    // Bound emitted offenses to the number of slots we rescan. This watcher currently tracks one offense type,
+    // and at most one offense of that type can be emitted per slot.
+    const offenseTypes = 1;
+    this.emittedOffenses = FifoSet.withLimit<string>(offenseTypes * this.scanSlotLookback);
+
     const intervalMs = Math.max(1000, (constants.ethereumSlotDuration * 1000) / 4);
     this.runningPromise = new RunningPromise(() => this.scan(), this.log, intervalMs);
     this.log.info('BroadcastedInvalidCheckpointProposalWatcher initialized', {
@@ -182,10 +189,6 @@ export class BroadcastedInvalidCheckpointProposalWatcher
 
   private markAsNewOffense(args: WantToSlashArgs): boolean {
     const key = `${args.validator.toString()}-${args.offenseType}-${args.epochOrSlot}`;
-    if (this.emittedOffenses.has(key)) {
-      return false;
-    }
-    this.emittedOffenses.add(key);
-    return true;
+    return this.emittedOffenses.addIfAbsent(key);
   }
 }