feat: tx file store source for tx collector

yarn-project/foundation/src/array/sorted_array.test.ts

@@ -2,6 +2,7 @@ import {
   dedupeSortedArray,
   findInSortedArray,
   findIndexInSortedArray,
+  findInsertionIndexInSortedArray,
   insertIntoSortedArray,
   merge,
   removeAnyOf,
@@ -125,6 +126,55 @@ describe('sorted_array', () => {
     }
   });
 
+  describe('findInsertionIndexInSortedArray', () => {
+    it('returns 0 for empty array', () => {
+      expect(findInsertionIndexInSortedArray([], 1, cmp)).toBe(0);
+    });
+
+    it('returns count of elements <= needle', () => {
+      const tests: [number[], number, number][] = [
+        [[5], 3, 0],
+        [[5], 5, 1],
+        [[5], 7, 1],
+
+        [[1, 3, 5, 7], 0, 0],
+        [[1, 3, 5, 7], 1, 1],
+        [[1, 3, 5, 7], 2, 1],
+        [[1, 3, 5, 7], 3, 2],
+        [[1, 3, 5, 7], 4, 2],
+        [[1, 3, 5, 7], 5, 3],
+        [[1, 3, 5, 7], 6, 3],
+        [[1, 3, 5, 7], 7, 4],
+        [[1, 3, 5, 7], 8, 4],
+      ];
+      for (const [arr, needle, expected] of tests) {
+        expect(findInsertionIndexInSortedArray(arr, needle, cmp)).toBe(expected);
+      }
+    });
+
+    it('handles duplicates by returning index after all equal elements', () => {
+      expect(findInsertionIndexInSortedArray([1, 2, 2, 2, 3], 2, cmp)).toBe(4);
+      expect(findInsertionIndexInSortedArray([2, 2, 2], 2, cmp)).toBe(3);
+      expect(findInsertionIndexInSortedArray([1, 1, 1, 2], 1, cmp)).toBe(3);
+    });
+
+    it('works with heterogeneous types', () => {
+      type Timer = { deadline: number; callback: () => void };
+      const arr: Timer[] = [
+        { deadline: 100, callback: () => {} },
+        { deadline: 300, callback: () => {} },
+        { deadline: 500, callback: () => {} },
+      ];
+      const cmpByDeadline = (timer: Timer, needle: { deadline: number }) => cmp(timer.deadline, needle.deadline);
+
+      expect(findInsertionIndexInSortedArray(arr, { deadline: 0 }, cmpByDeadline)).toBe(0);
+      expect(findInsertionIndexInSortedArray(arr, { deadline: 100 }, cmpByDeadline)).toBe(1);
+      expect(findInsertionIndexInSortedArray(arr, { deadline: 200 }, cmpByDeadline)).toBe(1);
+      expect(findInsertionIndexInSortedArray(arr, { deadline: 300 }, cmpByDeadline)).toBe(2);
+      expect(findInsertionIndexInSortedArray(arr, { deadline: 600 }, cmpByDeadline)).toBe(3);
+    });
+  });
+
   it('findIndexInSortedArray with duplicates returns any valid occurrence', () => {
     // Binary search doesn't guarantee first occurrence, just any valid occurrence
     const arr = [1, 2, 2, 2, 3];

yarn-project/foundation/src/array/sorted_array.ts

@@ -21,34 +21,39 @@ export function dedupeSortedArray<T>(arr: T[], cmp: Cmp<T>): void {
 }
 
 export function insertIntoSortedArray<T>(arr: T[], item: T, cmp: Cmp<T>, allowDuplicates = true): boolean {
+  const index = findInsertionIndexInSortedArray(arr, item, cmp);
+
+  if (!allowDuplicates) {
+    // Check element before insertion point (upper bound returns index after equal elements)
+    if (index > 0 && cmp(arr[index - 1], item) === 0) {
+      return false;
+    }
+  }
+
+  arr.splice(index, 0, item);
+  return true;
+}
+
+/**
+ * Finds the index where needle would be inserted to maintain sorted order.
+ * Returns the count of elements less than or equal to needle.
+ */
+export function findInsertionIndexInSortedArray<T, N>(values: T[], needle: N, cmp: (a: T, b: N) => number): number {
   let start = 0;
-  let end = arr.length;
+  let end = values.length;
 
   while (start < end) {
     const mid = start + (((end - start) / 2) | 0);
-    const comparison = cmp(arr[mid], item);
+    const comparison = cmp(values[mid], needle);
 
-    if (comparison < 0) {
+    if (comparison <= 0) {
       start = mid + 1;
     } else {
       end = mid;
     }
   }
 
-  if (!allowDuplicates) {
-    // Check element at insertion point
-    if (start < arr.length && cmp(arr[start], item) === 0) {
-      return false;
-    }
-
-    // Check element before insertion point (in case we landed after duplicates)
-    if (start > 0 && cmp(arr[start - 1], item) === 0) {
-      return false;
-    }
-  }
-
-  arr.splice(start, 0, item);
-  return true;
+  return start;
 }
 
 export function findIndexInSortedArray<T, N>(values: T[], needle: N, cmp: (a: T, b: N) => number): number {

yarn-project/foundation/src/config/env_var.ts

@@ -246,8 +246,10 @@ export type EnvVar =
   | 'TX_COLLECTION_NODE_RPC_MAX_BATCH_SIZE'
   | 'TX_COLLECTION_NODE_RPC_URLS'
   | 'TX_COLLECTION_MISSING_TXS_COLLECTOR_TYPE'
+  | 'TX_COLLECTION_FILE_STORE_URLS'
+  | 'TX_COLLECTION_FILE_STORE_SLOW_DELAY_MS'
+  | 'TX_COLLECTION_FILE_STORE_FAST_DELAY_MS'
   | 'TX_FILE_STORE_URL'
-  | 'TX_FILE_STORE_DOWNLOAD_URL'
   | 'TX_FILE_STORE_UPLOAD_CONCURRENCY'
   | 'TX_FILE_STORE_MAX_QUEUE_SIZE'
   | 'TX_FILE_STORE_ENABLED'

yarn-project/foundation/src/queue/base_memory_queue.ts

@@ -122,7 +122,7 @@ export abstract class BaseMemoryQueue<T> {
    * @param handler - A function that takes an item of type T and returns a Promise<void> after processing the item.
    * @returns A Promise<void> that resolves when the queue is finished processing.
    */
-  public async process(handler: (item: T) => Promise<void>) {
+  public async process(handler: (item: T) => Promise<void> | void) {
     try {
       while (true) {
         const item = await this.get();

yarn-project/foundation/src/timer/date.test.ts

@@ -1,33 +1,186 @@
+import { retryUntil } from '../retry/index.js';
 import { sleep } from '../sleep/index.js';
 import { TestDateProvider } from './date.js';
 
 describe('TestDateProvider', () => {
   let dateProvider: TestDateProvider;
+
   beforeEach(() => {
     dateProvider = new TestDateProvider();
   });
 
-  it('should return the current datetime', () => {
-    const currentTime = Date.now();
-    const result = dateProvider.now();
-    expect(result).toBeGreaterThanOrEqual(currentTime);
-    expect(result).toBeLessThan(currentTime + 100);
+  afterEach(() => {
+    dateProvider.clearPendingTimeouts();
+  });
+
+  describe('now', () => {
+    it('should return the current datetime', () => {
+      const currentTime = Date.now();
+      const result = dateProvider.now();
+      expect(result).toBeGreaterThanOrEqual(currentTime);
+      expect(result).toBeLessThan(currentTime + 100);
+    });
+
+    it('should return the overridden datetime', () => {
+      const overriddenTime = Date.now() + 1000;
+      dateProvider.setTime(overriddenTime);
+      const result = dateProvider.now();
+      expect(result).toBeGreaterThanOrEqual(overriddenTime);
+      expect(result).toBeLessThan(overriddenTime + 100);
+    });
+
+    it('should keep ticking after overriding', async () => {
+      const overriddenTime = Date.now() + 1000;
+      dateProvider.setTime(overriddenTime);
+      await sleep(510);
+      const result = dateProvider.now();
+      expect(result).toBeGreaterThanOrEqual(overriddenTime + 500);
+      expect(result).toBeLessThan(overriddenTime + 600);
+    });
   });
 
-  it('should return the overridden datetime', () => {
-    const overriddenTime = Date.now() + 1000;
-    dateProvider.setTime(overriddenTime);
-    const result = dateProvider.now();
-    expect(result).toBeGreaterThanOrEqual(overriddenTime);
-    expect(result).toBeLessThan(overriddenTime + 100);
+  describe('createTimeoutSignal', () => {
+    it('should not abort signal before deadline', () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const signal = dateProvider.createTimeoutSignal(1000);
+
+      expect(signal.aborted).toBe(false);
+    });
+
+    it('should abort signal when setTime advances past deadline', () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const signal = dateProvider.createTimeoutSignal(1000);
+      expect(signal.aborted).toBe(false);
+
+      // Advance time past the deadline
+      dateProvider.setTime(baseTime + 1001);
+
+      expect(signal.aborted).toBe(true);
+      expect(signal.reason).toBeInstanceOf(DOMException);
+      expect(signal.reason.name).toBe('TimeoutError');
+    });
+
+    it('should abort immediately when ms <= 0', () => {
+      const signal = dateProvider.createTimeoutSignal(0);
+
+      expect(signal.aborted).toBe(true);
+      expect(signal.reason.name).toBe('TimeoutError');
+    });
+
+    it('should abort multiple signals in deadline order when time advances', () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const signal1 = dateProvider.createTimeoutSignal(1000);
+      const signal2 = dateProvider.createTimeoutSignal(500);
+      const signal3 = dateProvider.createTimeoutSignal(2000);
+
+      expect(signal1.aborted).toBe(false);
+      expect(signal2.aborted).toBe(false);
+      expect(signal3.aborted).toBe(false);
+
+      // Advance past signal2's deadline only
+      dateProvider.setTime(baseTime + 600);
+
+      expect(signal1.aborted).toBe(false);
+      expect(signal2.aborted).toBe(true);
+      expect(signal3.aborted).toBe(false);
+
+      // Advance past signal1's deadline
+      dateProvider.setTime(baseTime + 1500);
+
+      expect(signal1.aborted).toBe(true);
+      expect(signal3.aborted).toBe(false);
+
+      // Advance past signal3's deadline
+      dateProvider.setTime(baseTime + 2500);
+
+      expect(signal3.aborted).toBe(true);
+    });
   });
 
-  it('should keep ticking after overriding', async () => {
-    const overriddenTime = Date.now() + 1000;
-    dateProvider.setTime(overriddenTime);
-    await sleep(510);
-    const result = dateProvider.now();
-    expect(result).toBeGreaterThanOrEqual(overriddenTime + 500);
-    expect(result).toBeLessThan(overriddenTime + 600);
+  describe('sleep', () => {
+    it('should resolve immediately when ms <= 0', async () => {
+      await expect(dateProvider.sleep(0)).resolves.toBeUndefined();
+    });
+
+    it('should resolve when setTime advances past deadline', async () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const sleepPromise = dateProvider.sleep(1000);
+
+      // Advance time past the deadline
+      dateProvider.setTime(baseTime + 1001);
+
+      await expect(sleepPromise).resolves.toBeUndefined();
+    });
+
+    it('should resolve multiple sleeps in deadline order when time advances', async () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const resolveOrder: number[] = [];
+
+      const sleep1 = dateProvider.sleep(1000).then(() => resolveOrder.push(1));
+      const sleep2 = dateProvider.sleep(500).then(() => resolveOrder.push(2));
+      const sleep3 = dateProvider.sleep(2000).then(() => resolveOrder.push(3));
+
+      // Advance past all deadlines at once
+      dateProvider.setTime(baseTime + 3000);
+
+      await Promise.all([sleep1, sleep2, sleep3]);
+
+      // Should resolve in deadline order: sleep2 (500ms), sleep1 (1000ms), sleep3 (2000ms)
+      expect(resolveOrder).toEqual([2, 1, 3]);
+    });
+  });
+
+  describe('clearPendingTimeouts', () => {
+    it('should clear pending timeouts so they never abort', async () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const signal = dateProvider.createTimeoutSignal(1000);
+
+      expect(signal.aborted).toBe(false);
+
+      dateProvider.clearPendingTimeouts();
+
+      const aborted = await retryUntil(() => signal.aborted, 'wait for abort', 0.1, 0.01);
+      expect(aborted).toBe(true);
+    });
+  });
+
+  describe('combined timeout and sleep behavior', () => {
+    it('should handle interleaved timeouts and sleeps', async () => {
+      const baseTime = Date.now();
+      dateProvider.setTime(baseTime);
+
+      const signal1 = dateProvider.createTimeoutSignal(500);
+      const sleep1Promise = dateProvider.sleep(750);
+      const signal2 = dateProvider.createTimeoutSignal(1000);
+
+      // Advance to 600ms - only signal1 should abort
+      dateProvider.setTime(baseTime + 600);
+
+      expect(signal1.aborted).toBe(true);
+      expect(signal2.aborted).toBe(false);
+
+      // Advance to 800ms - sleep1 should resolve
+      dateProvider.setTime(baseTime + 800);
+      await sleep1Promise;
+
+      expect(signal2.aborted).toBe(false);
+
+      // Advance to 1100ms - signal2 should abort
+      dateProvider.setTime(baseTime + 1100);
+
+      expect(signal2.aborted).toBe(true);
+    });
   });
 });