diff --git a/frontend/sync-client/src/utils/assert-set-contains-exactly.ts b/frontend/sync-client/src/utils/assert-set-contains-exactly.ts deleted file mode 100644 index 5baf965e..00000000 --- a/frontend/sync-client/src/utils/assert-set-contains-exactly.ts +++ /dev/null @@ -1,13 +0,0 @@ -import assert from "node:assert"; - -export function assertSetContainsExactly(set: Set, ...values: T[]): void { - assert.ok( - set.size === values.length && - Array.from(set).every((value) => values.includes(value)), - `Expected set to contain only ${values.map((v) => '"' + v + '"').join(", ")}, but it contained ${Array.from( - set - ) - .map((v) => '"' + v + '"') - .join(", ")}` - ); -} diff --git a/frontend/sync-client/src/utils/data-structures/event-listeners.test.ts b/frontend/sync-client/src/utils/data-structures/event-listeners.test.ts index a5f0cc7c..47c850f9 100644 --- a/frontend/sync-client/src/utils/data-structures/event-listeners.test.ts +++ b/frontend/sync-client/src/utils/data-structures/event-listeners.test.ts @@ -3,6 +3,30 @@ import assert from "node:assert"; import { EventListeners } from "./event-listeners"; describe("EventListeners", () => { + it("isolates throwing and rejecting observers and still notifies later listeners", async () => { + const failures: unknown[] = []; + const listeners = new EventListeners<() => unknown>((error) => + failures.push(error) + ); + const syncError = new Error("sync observer"); + const asyncError = new Error("async observer"); + let called = 0; + listeners.add(() => { + throw syncError; + }); + listeners.add(async () => { + throw asyncError; + }); + listeners.add(() => { + called++; + }); + assert.doesNotThrow(() => { + listeners.trigger(); + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(called, 1); + assert.deepEqual(failures, [syncError, asyncError]); + }); it("should add & remove listeners", () => { const listeners = new EventListeners<() => void>(); // eslint-disable-next-line @typescript-eslint/no-empty-function diff --git a/frontend/sync-client/src/utils/data-structures/event-listeners.ts b/frontend/sync-client/src/utils/data-structures/event-listeners.ts index e08ca65e..42a13eac 100644 --- a/frontend/sync-client/src/utils/data-structures/event-listeners.ts +++ b/frontend/sync-client/src/utils/data-structures/event-listeners.ts @@ -8,50 +8,65 @@ import { awaitAll } from "../await-all"; export class EventListeners any> { private readonly listeners: TListener[] = []; + public constructor( + private readonly onError: (error: unknown) => void = (error) => { + console.error("Event observer failed", error); + } + ) { } + public get count(): number { return this.listeners.length; } /** - * Adds a new listener to the collection. - * - * @param listener The listener callback to add - * @returns An unsubscribe function that removes this listener when called - */ + * Adds a new listener to the collection. + * + * @param listener The listener callback to add + * @returns An unsubscribe function that removes this listener when called + */ public add(listener: TListener): () => void { this.listeners.push(listener); return () => this.remove(listener); } /** - * Removes a listener from the collection. - * - * @param listener The listener callback to remove - * @returns true if the listener was found and removed, false otherwise - */ + * Removes a listener from the collection. + * + * @param listener The listener callback to remove + * @returns true if the listener was found and removed, false otherwise + */ public remove(listener: TListener): boolean { return removeFromArray(this.listeners, listener); } /** - * Triggers all listeners synchronously with the provided arguments. - * Any returned promises are ignored. Use triggerAsync() to await them. - * - * @param args The arguments to pass to each listener - */ + * Triggers all listeners synchronously with the provided arguments. + * Observer failures are reported without interrupting delivery or background + * work. Returned promises are observed; use triggerAsync() to await them. + * + * @param args The arguments to pass to each listener + */ public trigger(...args: Parameters): void { - this.listeners.forEach((listener) => { - listener(...args); - }); + for (const listener of [...this.listeners]) { + try { + void Promise.resolve(listener(...args)).catch( + (error: unknown) => { + this.reportFailure(error); + } + ); + } catch (error) { + this.reportFailure(error); + } + } } /** - * Triggers all listeners and awaits any promises they return. - * Synchronous listeners are called immediately, and any async listeners - * are awaited in parallel. - * - * @param args The arguments to pass to each listener - */ + * Triggers all listeners and awaits any promises they return. + * Synchronous listeners are called immediately, and any async listeners + * are awaited in parallel. + * + * @param args The arguments to pass to each listener + */ public async triggerAsync(...args: Parameters): Promise { await awaitAll( this.listeners @@ -68,4 +83,12 @@ export class EventListeners any> { public clear(): void { this.listeners.length = 0; } + + private reportFailure(error: unknown): void { + try { + this.onError(error); + } catch { + // A diagnostic callback must not break background work either. + } + } } diff --git a/frontend/sync-client/src/utils/data-structures/fix-sized-cache.ts b/frontend/sync-client/src/utils/data-structures/fix-sized-cache.ts index 51ad41c1..02954a83 100644 --- a/frontend/sync-client/src/utils/data-structures/fix-sized-cache.ts +++ b/frontend/sync-client/src/utils/data-structures/fix-sized-cache.ts @@ -1,66 +1,36 @@ -// Implements an in-memory fixed-size cache for document contents, - import type { VaultUpdateId } from "../../persistence/database"; -// Doubly-linked list node for O(1) LRU operations -class LRUNode { - public constructor( - public key: VaultUpdateId, - public value: Uint8Array, - public prev: LRUNode | null = null, - public next: LRUNode | null = null - ) {} -} - -// evicting the least recently used documents when the size limit is exceeded. +/** Map insertion order tracks the least recently used document first. */ export class FixedSizeDocumentCache { - private currentSizeInBytes: number; - private readonly cache: Map; - private head: LRUNode | null; // Least recently used - private tail: LRUNode | null; // Most recently used + private currentSizeInBytes = 0; + private readonly cache = new Map(); - public constructor(private maxSizeInBytes: number) { - this.currentSizeInBytes = 0; - this.cache = new Map(); - this.head = null; - this.tail = null; - } + public constructor(private maxSizeInBytes: number) { } public get(updateId: VaultUpdateId): Uint8Array | undefined { - const node = this.cache.get(updateId); - if (node) { - this.moveToTail(node); - return node.value; + const content = this.cache.get(updateId); + + if (content !== undefined) { + this.cache.delete(updateId); + this.cache.set(updateId, content); } - return undefined; + return content; } public put(updateId: VaultUpdateId, content: Uint8Array): void { - if (content.byteLength > this.maxSizeInBytes) { - // Document is too large to fit in the cache - return; - } + if (content.byteLength > this.maxSizeInBytes) return; - // If the document is already in the cache, update it - const existingNode = this.cache.get(updateId); - if (existingNode != null) { - this.currentSizeInBytes -= existingNode.value.byteLength; - this.removeNode(existingNode); - this.cache.delete(updateId); - } + this.currentSizeInBytes -= this.cache.get(updateId)?.byteLength ?? 0; + this.cache.delete(updateId); - const newNode = new LRUNode(updateId, content); - this.cache.set(updateId, newNode); - this.addToTail(newNode); + this.cache.set(updateId, content); this.currentSizeInBytes += content.byteLength; this.fitBelowMaxSize(); } public reset(): void { this.cache.clear(); - this.head = null; - this.tail = null; this.currentSizeInBytes = 0; } @@ -70,50 +40,10 @@ export class FixedSizeDocumentCache { } private fitBelowMaxSize(): void { - // Evict least recently used documents if over size limit - while (this.currentSizeInBytes > this.maxSizeInBytes && this.head) { - const lruNode = this.head; - this.removeNode(lruNode); - this.cache.delete(lruNode.key); - this.currentSizeInBytes -= lruNode.value.byteLength; + for (const [id, content] of this.cache) { + if (this.currentSizeInBytes <= this.maxSizeInBytes) break; + this.cache.delete(id); + this.currentSizeInBytes -= content.byteLength; } } - - private removeNode(node: LRUNode): void { - if (node.prev) { - node.prev.next = node.next; - } else { - this.head = node.next; - } - - if (node.next) { - node.next.prev = node.prev; - } else { - this.tail = node.prev; - } - - node.prev = null; - node.next = null; - } - - private addToTail(node: LRUNode): void { - node.prev = this.tail; - node.next = null; - - if (this.tail) { - this.tail.next = node; - } - - this.tail = node; - - this.head ??= node; - } - - private moveToTail(node: LRUNode): void { - if (node === this.tail) { - return; - } - this.removeNode(node); - this.addToTail(node); - } } diff --git a/frontend/sync-client/src/utils/data-structures/locks.test.ts b/frontend/sync-client/src/utils/data-structures/locks.test.ts index f7f0acf2..e325496b 100644 --- a/frontend/sync-client/src/utils/data-structures/locks.test.ts +++ b/frontend/sync-client/src/utils/data-structures/locks.test.ts @@ -1,292 +1,46 @@ -import { describe, it, beforeEach } from "node:test"; -import assert from "node:assert"; -import { Logger } from "../../tracing/logger"; -import type { RelativePath } from "../../persistence/database"; -import { Locks } from "./locks"; -import { awaitAll } from "../await-all"; -import { sleep } from "../sleep"; -import { SyncResetError } from "../../errors/errors"; +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { Lock } from "./locks"; -describe("withLock", () => { - const testPath: RelativePath = "test/document/path"; - const testPath2: RelativePath = "test/document/path2"; - const logger = new Logger(); - - // eslint-disable-next-line @typescript-eslint/init-declarations - let locks: Locks; - - beforeEach(() => { - locks = new Locks(logger); +test("queued operations run in order and retain their results", async () => { + const lock = new Lock(); + const held = Promise.withResolvers(); + const started = Promise.withResolvers(); + const calls: number[] = []; + const first = lock.withLock(async () => { + calls.push(1); + started.resolve(); + await held.promise; + calls.push(2); + return "first"; }); - - it("should execute function with single key lock", async () => { - let executionCount = 0; - const result = await locks.withLock(testPath, () => { - executionCount++; - return "success"; - }); - - assert.strictEqual(result, "success"); - assert.strictEqual(executionCount, 1); + const second = lock.withLock(() => { + calls.push(3); + return "second"; }); - - it("should execute async function with single key lock", async () => { - let executionCount = 0; - const result = await locks.withLock(testPath, async () => { - executionCount++; - await sleep(10); - return "async-success"; - }); - - assert.strictEqual(result, "async-success"); - assert.strictEqual(executionCount, 1); - }); - - it("should execute function with multiple key locks", async () => { - let executionCount = 0; - const result = await locks.withLock([testPath, testPath2], () => { - executionCount++; - return "multi-success"; - }); - - assert.strictEqual(result, "multi-success"); - assert.strictEqual(executionCount, 1); - }); - - it("should sort multiple keys to prevent deadlocks", async () => { - const executionOrder: string[] = []; - - // Start two concurrent operations with keys in different orders - const promise1 = locks.withLock([testPath2, testPath], async () => { - executionOrder.push("operation1-start"); - await sleep(50); - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock([testPath, testPath2], async () => { - executionOrder.push("operation2-start"); - await sleep(50); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - // One operation should complete entirely before the other starts - assert.deepStrictEqual(executionOrder, [ - "operation1-start", - "operation1-end", - "operation2-start", - "operation2-end" - ]); - }); - - it("should serialize access to same key", async () => { - const executionOrder: string[] = []; - - const promise1 = locks.withLock(testPath, async () => { - executionOrder.push("operation1-start"); - await sleep(50); - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock(testPath, async () => { - executionOrder.push("operation2-start"); - await sleep(30); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - assert.deepStrictEqual(executionOrder, [ - "operation1-start", - "operation1-end", - "operation2-start", - "operation2-end" - ]); - }); - - it("should allow concurrent access to different keys", async () => { - const executionOrder: string[] = []; - - const promise1 = locks.withLock(testPath, async () => { - executionOrder.push("operation1-start"); - await sleep(50); - - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock(testPath2, async () => { - executionOrder.push("operation2-start"); - await sleep(30); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - // Both operations should run concurrently - assert.strictEqual(executionOrder[0], "operation1-start"); - assert.strictEqual(executionOrder[1], "operation2-start"); - }); - - it("should release locks even if function throws", async () => { - const error = new Error("test error"); - - await assert.rejects( - locks.withLock(testPath, () => { - throw error; - }), - { message: "test error" } - ); - - // Lock should be released, allowing another operation - const result = await locks.withLock( - testPath, - () => "success-after-error" - ); - assert.strictEqual(result, "success-after-error"); - }); - - it("should release locks even if async function throws", async () => { - const error = new Error("async test error"); - - await assert.rejects( - locks.withLock(testPath, async () => { - await sleep(10); - - throw error; - }), - { message: "async test error" } - ); - - // Lock should be released, allowing another operation - const result = await locks.withLock( - testPath, - () => "success-after-async-error" - ); - assert.strictEqual(result, "success-after-async-error"); - }); - - it("should handle empty array of keys", async () => { - const result = await locks.withLock([], () => "empty-keys"); - assert.strictEqual(result, "empty-keys"); - }); - - it("should maintain FIFO order for multiple waiters", async () => { - const executionOrder: string[] = []; - - // Start first operation that holds the lock - const firstPromise = locks.withLock(testPath, async () => { - executionOrder.push("first-start"); - await sleep(100); - executionOrder.push("first-end"); - return "first"; - }); - - // Small delay to ensure first operation starts - await sleep(10); - - // Queue second and third operations - const secondPromise = locks.withLock(testPath, async () => { - executionOrder.push("second-start"); - await sleep(50); - executionOrder.push("second-end"); - return "second"; - }); - - const thirdPromise = locks.withLock(testPath, async () => { - executionOrder.push("third-start"); - await sleep(20); - executionOrder.push("third-end"); - return "third"; - }); - - const [first, second, third] = await awaitAll([ - firstPromise, - secondPromise, - thirdPromise - ]); - - assert.strictEqual(first, "first"); - assert.strictEqual(second, "second"); - assert.strictEqual(third, "third"); - assert.deepStrictEqual(executionOrder, [ - "first-start", - "first-end", - "second-start", - "second-end", - "third-start", - "third-end" - ]); + const third = lock.withLock(() => { + calls.push(4); + return "third"; }); + await started.promise; + assert.deepEqual(calls, [1]); + held.resolve(); + assert.equal(await first, "first"); + assert.equal(await second, "second"); + assert.equal(await third, "third"); + assert.deepEqual(calls, [1, 2, 3, 4]); }); -describe("reset", () => { - const testPath: RelativePath = "test/document/path"; - const logger = new Logger(); - - // eslint-disable-next-line @typescript-eslint/init-declarations - let locks: Locks; - - beforeEach(() => { - locks = new Locks(logger); - }); - - it("should reject pending waiters with SyncResetError while running operation completes", async () => { - const firstPromise = locks.withLock(testPath, async () => { - await sleep(2); - return "first"; - }); - - await sleep(1); - - const secondPromise = locks.withLock(testPath, async () => "second"); - void secondPromise.catch(() => {}); // eslint-disable-line @typescript-eslint/no-empty-function - - locks.reset(); - - assert.strictEqual(await firstPromise, "first"); - - await assert.rejects(secondPromise, (err: Error) => { - assert.ok(err instanceof SyncResetError); - return true; +for (const asynchronous of [false, true]) { + test(`a ${asynchronous ? "rejected promise" : "thrown error"} releases the lock`, async () => { + const lock = new Lock(); + const error = new Error("failed operation"); + const failed = lock.withLock(() => { + if (asynchronous) return Promise.reject(error); + throw error; }); + const next = lock.withLock(() => "continued"); + await assert.rejects(failed, (caught) => caught === error); + assert.equal(await next, "continued"); }); - - it("should allow locks to work normally after reset", async () => { - const firstPromise = locks.withLock(testPath, async () => { - await sleep(1); - return "first"; - }); - - await sleep(1); - - const secondPromise = locks.withLock(testPath, async () => "second"); - void secondPromise.catch(() => {}); // eslint-disable-line @typescript-eslint/no-empty-function - - locks.reset(); - - await firstPromise; - - const result = await locks.withLock(testPath, () => "after-reset"); - assert.strictEqual(result, "after-reset"); - }); - - it("should handle reset with no pending operations", async () => { - locks.reset(); - - const result = await locks.withLock(testPath, () => "success"); - assert.strictEqual(result, "success"); - }); -}); +} diff --git a/frontend/sync-client/src/utils/data-structures/locks.ts b/frontend/sync-client/src/utils/data-structures/locks.ts index 096cacdf..da5962a8 100644 --- a/frontend/sync-client/src/utils/data-structures/locks.ts +++ b/frontend/sync-client/src/utils/data-structures/locks.ts @@ -1,166 +1,10 @@ -import { SyncResetError } from "../../errors/errors"; -import type { Logger } from "../../tracing/logger"; -import { awaitAll } from "../await-all"; - -/** - * Manages exclusive locks on items to prevent concurrent modifications. - * Locks are granted in FIFO order. - * - * @template T The type of the key used for locking - */ -export class Locks { - /** Currently locked keys */ - private readonly locked = new Set(); - - /** Queue of resolve functions waiting for each key */ - private readonly waiters = new Map< - T, - [() => unknown, (err: unknown) => unknown][] - >(); - - public constructor(private readonly logger?: Logger) {} - - /** - * Executes a function while holding exclusive locks on one or more keys. - * - * This method ensures that the provided function runs with exclusive access to the - * specified key(s). Multiple keys are sorted to prevent deadlocks when different - * operations request the same keys in different orders. - * - * @template R The return type of the function to execute - * @param keyOrKeys A single key or array of keys to lock during function execution - * @param fn The function to execute while holding the lock(s). Can be sync or async. - * @returns A Promise that resolves to the return value of the executed function - * - * @example - * ```typescript - * // Lock a single key - * const result = await locks.withLock('file1', () => { - * // Critical section - only one operation can access 'file1' at a time - * return processFile('file1'); - * }); - * - * // Lock multiple keys (prevents deadlocks through consistent ordering) - * await locks.withLock(['file1', 'file2'], async () => { - * // Critical section - exclusive access to both files - * await moveFile('file1', 'file2'); - * }); - * ``` - * - * @throws Any error thrown by the provided function will be propagated after locks are released - */ - public async withLock( - keyOrKeys: T | T[], - fn: () => R | Promise - ): Promise { - const keys = Array.isArray(keyOrKeys) ? keyOrKeys : [keyOrKeys]; - - // Deduplicate keys to prevent deadlock from acquiring same lock twice - const uniqueKeys = Array.from(new Set(keys)); - uniqueKeys.sort((a, b) => String(a).localeCompare(String(b))); // Ensure consistent order to prevent deadlocks - - await awaitAll(uniqueKeys.map(async (key) => this.waitForLock(key))); - - try { - return await fn(); - } finally { - uniqueKeys.forEach((key) => { - this.unlock(key); - }); - } - } - - public reset(): void { - // Resolve all waiting promises before clearing to prevent deadlock - // Any operation waiting for a lock will be granted access immediately - for (const waiting of this.waiters.values()) { - for (const [_, reject] of waiting) { - reject(new SyncResetError()); - } - } - this.locked.clear(); - this.waiters.clear(); - } - - /** - * Attempts to acquire a lock immediately without waiting. - * Must call `unlock()` if successful. - * - * @param key The key to lock - * @returns `true` if lock acquired, `false` if already locked - */ - public tryLock(key: T): boolean { - if (this.locked.has(key)) { - return false; - } - - this.locked.add(key); - - return true; - } - - /** - * Waits to acquire a lock, blocking until available. - * Operations are queued in FIFO order. Must call `unlock()` when done. - * - * @param key The key to wait for and lock - * @returns Promise that resolves when lock is acquired - */ - public async waitForLock(key: T): Promise { - if (this.tryLock(key)) { - return Promise.resolve(); - } - - this.logger?.debug(`Waiting for lock on ${key}`); - - return new Promise((resolve, reject) => { - // DefaultDict behavior - let waiting = this.waiters.get(key); - if (!waiting) { - waiting = []; - this.waiters.set(key, waiting); - } - - waiting.push([resolve, reject]); - }); - } - - /** - * Releases a lock and grants access to the next waiting operation in FIFO order. - * Removes the key from locked set if no waiters. - * - * @param key The key to unlock - * @throws {Error} If key is not currently locked - */ - public unlock(key: T): void { - if (!this.locked.has(key)) { - return; - } - - // Remove first waiter to ensure FIFO order - const [resolveNextWaiting, _] = this.waiters.get(key)?.shift() ?? []; - - if (resolveNextWaiting) { - this.logger?.debug(`Granted lock on ${key}`); - resolveNextWaiting(); - } else { - this.locked.delete(key); - } - } -} - export class Lock { - private readonly locks: Locks; + private tail: Promise = Promise.resolve(); - public constructor(logger?: Logger) { - this.locks = new Locks(logger); - } - - public async withLock(fn: () => R | Promise): Promise { - return this.locks.withLock(true, fn); - } - - public reset(): void { - this.locks.reset(); + // eslint-disable-next-line @typescript-eslint/promise-function-async -- Preserve the queued operation promise without an extra async wrapper. + public withLock(fn: () => R | Promise): Promise { + const operation = this.tail.then(fn); + this.tail = operation.catch(() => undefined); + return operation; } } diff --git a/frontend/sync-client/src/utils/data-structures/min-covered.test.ts b/frontend/sync-client/src/utils/data-structures/min-covered.test.ts deleted file mode 100644 index 7b7271d7..00000000 --- a/frontend/sync-client/src/utils/data-structures/min-covered.test.ts +++ /dev/null @@ -1,76 +0,0 @@ -import { describe, it } from "node:test"; -import assert from "node:assert"; -import { CoveredValues } from "./min-covered"; - -describe("CoveredValues", () => { - it("should initialize with the given min value", () => { - const covered = new CoveredValues(5); - assert.strictEqual(covered.min, 5); - }); - - it("should add values greater than min", () => { - const covered = new CoveredValues(0); - covered.add(3); - assert.strictEqual(covered.min, 0); - covered.add(1); - assert.strictEqual(covered.min, 1); - covered.add(4); - assert.strictEqual(covered.min, 1); - covered.add(2); - assert.strictEqual(covered.min, 4); - }); - - it("should ignore duplicate values", () => { - const covered = new CoveredValues(0); - covered.add(3); - covered.add(3); - covered.add(3); - assert.strictEqual(covered.min, 0); - covered.add(1); - covered.add(2); - assert.strictEqual(covered.min, 3); - }); - - it("should handle multiple consecutive values", () => { - const covered = new CoveredValues(132); - for (let i = 250; i > 132; i--) { - assert.strictEqual(covered.min, 132); - covered.add(i); - } - assert.strictEqual(covered.min, 250); - }); - - it("should handle adding values lower than current min", () => { - const covered = new CoveredValues(5); - covered.add(3); - assert.strictEqual(covered.min, 5); - covered.add(6); - assert.strictEqual(covered.min, 6); - }); - - it("should auto-advance when setting min value", () => { - const covered = new CoveredValues(5); - covered.add(7); - covered.add(8); - covered.add(9); - assert.strictEqual(covered.min, 5); - // Setting min to 6 should auto-advance through 7, 8, 9 - covered.min = 6; - assert.strictEqual(covered.min, 9); - covered.add(10); - assert.strictEqual(covered.min, 10); - }); - - it("should handle setting min value with no consecutive values", () => { - const covered = new CoveredValues(5); - covered.add(10); - covered.add(15); - assert.strictEqual(covered.min, 5); - // Setting min to 8 should not auto-advance (no consecutive values) - covered.min = 8; - assert.strictEqual(covered.min, 8); - // Add 9 to trigger auto-advance to 10 - covered.add(9); - assert.strictEqual(covered.min, 10); - }); -}); diff --git a/frontend/sync-client/src/utils/data-structures/min-covered.ts b/frontend/sync-client/src/utils/data-structures/min-covered.ts deleted file mode 100644 index 8b38822f..00000000 --- a/frontend/sync-client/src/utils/data-structures/min-covered.ts +++ /dev/null @@ -1,61 +0,0 @@ -/** - * A class that tracks the minimum covered value in a sequence of numbers. - * It keeps track of a minimum value based on the seen values. - * - * It expects integers slightly out of order and makes sure that the value of `min` is - * always the minimum of the seen values. This is done with bounded memory usage. - * - * @example - * ```typescript - * const covered = new CoveredValues(0); - * covered.add(2); // seenValues = [2], min = 0 - * covered.add(1); // seenValues = [], min = 2 - * covered.min; // returns 2 - * ``` - */ -export class CoveredValues { - private seenValues: number[] = []; - - public constructor(private minValue: number) {} - - public get min(): number { - return this.minValue; - } - - public set min(value: number) { - this.minValue = Math.max(value, this.minValue); - this.seenValues = this.seenValues.filter((v) => v > this.minValue); - this.advanceMinWhilePossible(); - } - - public add(value: number | undefined): void { - if (value === undefined || value < this.minValue) { - return; - } - - let i = 0; - while (i < this.seenValues.length && this.seenValues[i] < value) { - i++; - } - - if (i === this.seenValues.length) { - this.seenValues.push(value); - } else if (this.seenValues[i] === value) { - return; - } else { - this.seenValues.splice(i, 0, value); - } - - this.advanceMinWhilePossible(); - } - - private advanceMinWhilePossible(): void { - while ( - this.seenValues.length > 0 && - this.seenValues[0] === this.minValue + 1 - ) { - this.seenValues.shift(); - this.minValue++; - } - } -} diff --git a/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts b/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts index c64bff18..258db398 100644 --- a/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts +++ b/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts @@ -1,5 +1,5 @@ import { sleep } from "../sleep"; -import { Locks } from "../data-structures/locks"; +import { Lock } from "../data-structures/locks"; import type { Logger } from "../../tracing/logger"; export function slowWebSocketFactory( @@ -8,10 +8,8 @@ export function slowWebSocketFactory( ): typeof WebSocket { // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion return class FlakyWebSocket extends WebSocket { - private static readonly RECEIVE_KEY = "websocket-receive"; - private static readonly SEND_KEY = "websocket-send"; - - private readonly locks = new Locks(logger); + private readonly receiving = new Lock(); + private readonly sending = new Lock(); public set onopen(callback: ((event: Event) => void) | null) { super.onopen = async (event: Event): Promise => { @@ -25,18 +23,15 @@ export function slowWebSocketFactory( public set onmessage(callback: ((event: MessageEvent) => void) | null) { super.onmessage = async (event: MessageEvent): Promise => { - await this.locks.withLock( - FlakyWebSocket.RECEIVE_KEY, - async () => { - if (jitterScaleInSeconds > 0) { - await sleep( - Math.random() * jitterScaleInSeconds * 1000 - ); - } - - callback?.(event); + await this.receiving.withLock(async () => { + if (jitterScaleInSeconds > 0) { + await sleep( + Math.random() * jitterScaleInSeconds * 1000 + ); } - ); + + callback?.(event); + }); }; } @@ -70,7 +65,7 @@ export function slowWebSocketFactory( data: string | ArrayBufferLike | Blob | ArrayBufferView ): Promise { // maintain message order - await this.locks.withLock(FlakyWebSocket.SEND_KEY, async () => { + await this.sending.withLock(async () => { if (jitterScaleInSeconds > 0) { await sleep(Math.random() * jitterScaleInSeconds * 1000); }