Clean up helpers
This commit is contained in:
parent
2d0b7b2db1
commit
5c7ba128ad
9 changed files with 149 additions and 729 deletions
|
|
@ -1,13 +0,0 @@
|
|||
import assert from "node:assert";
|
||||
|
||||
export function assertSetContainsExactly<T>(set: Set<T>, ...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(", ")}`
|
||||
);
|
||||
}
|
||||
|
|
@ -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<void>((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
|
||||
|
|
|
|||
|
|
@ -8,50 +8,65 @@ import { awaitAll } from "../await-all";
|
|||
export class EventListeners<TListener extends (...args: any[]) => 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<TListener>): 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<TListener>): Promise<void> {
|
||||
await awaitAll(
|
||||
this.listeners
|
||||
|
|
@ -68,4 +83,12 @@ export class EventListeners<TListener extends (...args: any[]) => 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.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<VaultUpdateId, LRUNode>;
|
||||
private head: LRUNode | null; // Least recently used
|
||||
private tail: LRUNode | null; // Most recently used
|
||||
private currentSizeInBytes = 0;
|
||||
private readonly cache = new Map<VaultUpdateId, Uint8Array>();
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<RelativePath>;
|
||||
|
||||
beforeEach(() => {
|
||||
locks = new Locks<RelativePath>(logger);
|
||||
test("queued operations run in order and retain their results", async () => {
|
||||
const lock = new Lock();
|
||||
const held = Promise.withResolvers<void>();
|
||||
const started = Promise.withResolvers<void>();
|
||||
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<RelativePath>;
|
||||
|
||||
beforeEach(() => {
|
||||
locks = new Locks<RelativePath>(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");
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<T> {
|
||||
/** Currently locked keys */
|
||||
private readonly locked = new Set<T>();
|
||||
|
||||
/** 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<R>(
|
||||
keyOrKeys: T | T[],
|
||||
fn: () => R | Promise<R>
|
||||
): Promise<R> {
|
||||
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<void> {
|
||||
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<boolean>;
|
||||
private tail: Promise<unknown> = Promise.resolve();
|
||||
|
||||
public constructor(logger?: Logger) {
|
||||
this.locks = new Locks(logger);
|
||||
}
|
||||
|
||||
public async withLock<R>(fn: () => R | Promise<R>): Promise<R> {
|
||||
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<R>(fn: () => R | Promise<R>): Promise<R> {
|
||||
const operation = this.tail.then(fn);
|
||||
this.tail = operation.catch(() => undefined);
|
||||
return operation;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<void> => {
|
||||
|
|
@ -25,18 +23,15 @@ export function slowWebSocketFactory(
|
|||
|
||||
public set onmessage(callback: ((event: MessageEvent) => void) | null) {
|
||||
super.onmessage = async (event: MessageEvent): Promise<void> => {
|
||||
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<void> {
|
||||
// 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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue