Restructure packages

This commit is contained in:
Andras Schmelczer 2025-11-22 11:06:06 +00:00
parent 812eb7a644
commit 56c1f4d58b
19 changed files with 30 additions and 73 deletions

View file

@ -0,0 +1,275 @@
import { describe, it } from "node:test";
import assert from "node:assert";
import { FixedSizeDocumentCache } from "./fix-sized-cache";
describe("fixedSizeDocumentCache", () => {
it("happyPath", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
cache.put(1, doc1);
assert.equal(cache.get(1), doc1);
cache.put(2, doc2);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), doc2);
cache.put(3, doc3);
assert.equal(cache.get(1), undefined);
assert.equal(cache.get(2), doc2);
assert.equal(cache.get(3), doc3);
});
it("updateExistingEntry", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1_v1 = new Uint8Array([1, 2]);
const doc1_v2 = new Uint8Array([3, 4]);
const doc2 = new Uint8Array([5, 6]);
cache.put(1, doc1_v1);
assert.equal(cache.get(1), doc1_v1);
cache.put(2, doc2);
assert.equal(cache.get(1), doc1_v1);
assert.equal(cache.get(2), doc2);
cache.put(1, doc1_v2); // Update doc1
assert.equal(cache.get(1), doc1_v2);
assert.equal(cache.get(2), doc2);
});
it("evictOldestEntry", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
cache.put(1, doc1);
cache.put(2, doc2);
assert.equal(cache.get(2), doc2);
assert.equal(cache.get(1), doc1);
cache.put(3, doc3);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), undefined);
assert.equal(cache.get(3), doc3);
});
it("tooLargeEntry", async () => {
const cache = new FixedSizeDocumentCache(2);
const doc1 = new Uint8Array([1, 2, 3]);
cache.put(1, doc1);
assert.equal(cache.get(1), undefined);
});
it("multipleEvictionsInSinglePut", async () => {
const cache = new FixedSizeDocumentCache(10);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
const doc4 = new Uint8Array([7, 8, 9, 10, 11, 12, 13, 14]); // 8 bytes
cache.put(1, doc1);
cache.put(2, doc2);
cache.put(3, doc3);
// Cache now has 6 bytes total
cache.put(4, doc4); // Should evict doc1 and doc2 to make room (total: 2+8=10)
assert.equal(cache.get(1), undefined); // Evicted
assert.equal(cache.get(2), undefined); // Evicted
assert.equal(cache.get(3), doc3); // Still present
assert.equal(cache.get(4), doc4);
});
it("clearCache", async () => {
const cache = new FixedSizeDocumentCache(10);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
cache.put(1, doc1);
cache.put(2, doc2);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), doc2);
cache.clear();
assert.equal(cache.get(1), undefined);
assert.equal(cache.get(2), undefined);
// Should be able to add entries after clear
cache.put(3, doc1);
assert.equal(cache.get(3), doc1);
});
it("getNonExistentKey", async () => {
const cache = new FixedSizeDocumentCache(10);
const doc1 = new Uint8Array([1, 2]);
cache.put(1, doc1);
assert.equal(cache.get(999), undefined);
});
it("updateEntryWithDifferentSizeTriggeringEviction", async () => {
const cache = new FixedSizeDocumentCache(6);
const doc1_v1 = new Uint8Array([1, 2]);
const doc1_v2 = new Uint8Array([1, 2, 3, 4]); // Larger version
const doc2 = new Uint8Array([5, 6]);
const doc3 = new Uint8Array([7, 8]);
cache.put(1, doc1_v1);
cache.put(2, doc2);
cache.put(3, doc3);
// Update doc1 with larger version, should evict doc2
cache.put(1, doc1_v2);
assert.equal(cache.get(1), doc1_v2);
assert.equal(cache.get(2), undefined); // Evicted
assert.equal(cache.get(3), doc3);
});
it("singleItemCache", async () => {
const cache = new FixedSizeDocumentCache(2);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
cache.put(1, doc1);
assert.equal(cache.get(1), doc1);
cache.put(2, doc2);
assert.equal(cache.get(1), undefined); // Evicted
assert.equal(cache.get(2), doc2);
});
it("multipleGetsOnSameEntry", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
cache.put(1, doc1);
cache.put(2, doc2);
// Multiple gets on doc1
cache.get(1);
cache.get(1);
cache.get(1);
// Order should be: 2 (LRU), 1 (MRU)
cache.put(3, doc3);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), undefined); // Evicted
assert.equal(cache.get(3), doc3);
});
it("exactlySizedEntry", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2, 3, 4]); // Exactly cache size
cache.put(1, doc1);
assert.equal(cache.get(1), doc1);
const doc2 = new Uint8Array([5, 6]);
cache.put(2, doc2);
// doc1 should be evicted to make room for doc2
assert.equal(cache.get(1), undefined);
assert.equal(cache.get(2), doc2);
});
it("updateEntryMakesItMostRecent", async () => {
const cache = new FixedSizeDocumentCache(6);
const doc1_v1 = new Uint8Array([1, 2]);
const doc1_v2 = new Uint8Array([3, 4]);
const doc2 = new Uint8Array([5, 6]);
const doc3 = new Uint8Array([7, 8]);
const doc4 = new Uint8Array([9, 10]);
cache.put(1, doc1_v1);
cache.put(2, doc2);
cache.put(3, doc3);
// Update doc1 (should move it to most recent)
cache.put(1, doc1_v2);
// Order should be: 2 (LRU), 3, 1 (MRU)
// Adding doc4 should evict doc2
cache.put(4, doc4);
assert.equal(cache.get(1), doc1_v2);
assert.equal(cache.get(2), undefined); // Evicted
assert.equal(cache.get(3), doc3);
assert.equal(cache.get(4), doc4);
});
it("alternatingAccessPattern", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
cache.put(1, doc1);
cache.put(2, doc2);
// Alternate access between doc1 and doc2
cache.get(1);
cache.get(2);
cache.get(1);
cache.get(2);
// Order should be: 1, 2 (MRU)
cache.put(3, doc3);
assert.equal(cache.get(1), undefined); // Evicted
assert.equal(cache.get(2), doc2);
assert.equal(cache.get(3), doc3);
});
it("zeroByteDocs", async () => {
const cache = new FixedSizeDocumentCache(2);
const doc1 = new Uint8Array([]);
const doc2 = new Uint8Array([]);
const doc3 = new Uint8Array([1, 2]);
cache.put(1, doc1);
cache.put(2, doc2);
cache.put(3, doc3);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), doc2);
assert.equal(cache.get(3), doc3);
});
it("resizeToLargerSizeNoEviction", async () => {
const cache = new FixedSizeDocumentCache(4);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
cache.put(1, doc1);
cache.put(2, doc2);
cache.resize(10);
assert.equal(cache.get(1), doc1);
assert.equal(cache.get(2), doc2);
});
it("resizeCausesMultipleEvictions", async () => {
const cache = new FixedSizeDocumentCache(10);
const doc1 = new Uint8Array([1, 2]);
const doc2 = new Uint8Array([3, 4]);
const doc3 = new Uint8Array([5, 6]);
const doc4 = new Uint8Array([7, 8]);
cache.put(1, doc1);
cache.put(2, doc2);
cache.put(3, doc3);
cache.put(4, doc4);
// Cache has 8 bytes total
cache.resize(2);
// Should evict doc1, doc2, doc3 to get down to 2 bytes
assert.equal(cache.get(1), undefined);
assert.equal(cache.get(2), undefined);
assert.equal(cache.get(3), undefined);
assert.equal(cache.get(4), doc4);
});
});

View file

@ -0,0 +1,119 @@
// 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.
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
public constructor(private maxSizeInBytes: number) {
this.currentSizeInBytes = 0;
this.cache = new Map();
this.head = null;
this.tail = null;
}
public get(updateId: VaultUpdateId): Uint8Array | undefined {
const node = this.cache.get(updateId);
if (node) {
this.moveToTail(node);
return node.value;
}
return undefined;
}
public put(updateId: VaultUpdateId, content: Uint8Array): void {
if (content.byteLength > this.maxSizeInBytes) {
// Document is too large to fit in the cache
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);
}
const newNode = new LRUNode(updateId, content);
this.cache.set(updateId, newNode);
this.addToTail(newNode);
this.currentSizeInBytes += content.byteLength;
this.fitBelowMaxSize();
}
public clear(): void {
this.cache.clear();
this.head = null;
this.tail = null;
this.currentSizeInBytes = 0;
}
public resize(newMaxSizeInBytes: number): void {
this.maxSizeInBytes = newMaxSizeInBytes;
this.fitBelowMaxSize();
}
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;
}
}
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);
}
}

View file

@ -0,0 +1,228 @@
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";
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);
});
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);
});
it("should execute async function with single key lock", async () => {
let executionCount = 0;
const result = await locks.withLock(testPath, async () => {
executionCount++;
await new Promise((resolve) => setTimeout(resolve, 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 new Promise((resolve) => setTimeout(resolve, 50));
executionOrder.push("operation1-end");
return "result1";
});
const promise2 = locks.withLock([testPath, testPath2], async () => {
executionOrder.push("operation2-start");
await new Promise((resolve) => setTimeout(resolve, 50));
executionOrder.push("operation2-end");
return "result2";
});
const [result1, result2] = await Promise.all([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 new Promise((resolve) => setTimeout(resolve, 50));
executionOrder.push("operation1-end");
return "result1";
});
const promise2 = locks.withLock(testPath, async () => {
executionOrder.push("operation2-start");
await new Promise((resolve) => setTimeout(resolve, 30));
executionOrder.push("operation2-end");
return "result2";
});
const [result1, result2] = await Promise.all([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 new Promise((resolve) => setTimeout(resolve, 50));
executionOrder.push("operation1-end");
return "result1";
});
const promise2 = locks.withLock(testPath2, async () => {
executionOrder.push("operation2-start");
await new Promise((resolve) => setTimeout(resolve, 30));
executionOrder.push("operation2-end");
return "result2";
});
const [result1, result2] = await Promise.all([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 new Promise((resolve) => setTimeout(resolve, 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 new Promise((resolve) => setTimeout(resolve, 100));
executionOrder.push("first-end");
return "first";
});
// Small delay to ensure first operation starts
await new Promise((resolve) => setTimeout(resolve, 10));
// Queue second and third operations
const secondPromise = locks.withLock(testPath, async () => {
executionOrder.push("second-start");
await new Promise((resolve) => setTimeout(resolve, 30));
executionOrder.push("second-end");
return "second";
});
const thirdPromise = locks.withLock(testPath, async () => {
executionOrder.push("third-start");
await new Promise((resolve) => setTimeout(resolve, 20));
executionOrder.push("third-end");
return "third";
});
const [first, second, third] = await Promise.all([
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"
]);
});
});

View file

@ -0,0 +1,142 @@
import type { Logger } from "../../tracing/logger";
/**
* 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)[]>();
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];
keys.sort((a, b) => String(a).localeCompare(String(b))); // Ensure consistent order to prevent deadlocks
await Promise.all(keys.map(async (key) => this.waitForLock(key)));
try {
return await fn();
} finally {
keys.forEach((key) => {
this.unlock(key);
});
}
}
/**
* 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
*/
private 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
*/
private 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) => {
// DefaultDict behavior
let waiting = this.waiters.get(key);
if (!waiting) {
waiting = [];
this.waiters.set(key, waiting);
}
waiting.push(resolve);
});
}
/**
* 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
*/
private unlock(key: T): void {
if (!this.locked.has(key)) {
throw new Error(`Key '${key}' is not locked, cannot unlock`);
}
// Remove first waiter to ensure FIFO order
const nextWaiting = this.waiters.get(key)?.shift();
if (nextWaiting) {
this.logger?.debug(`Granted lock on ${key}`);
nextWaiting();
} else {
this.locked.delete(key);
}
}
}
export class Lock {
private readonly locks: Locks<boolean>;
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);
}
}

View file

@ -0,0 +1,62 @@
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 handle force setting min value", () => {
const covered = new CoveredValues(5);
covered.add(7);
covered.add(8);
covered.add(9);
assert.strictEqual(covered.min, 5);
covered.min = 6;
assert.strictEqual(covered.min, 6);
covered.add(10);
assert.strictEqual(covered.min, 10);
});
});

View file

@ -0,0 +1,56 @@
/**
* 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 > value);
}
public add(value: number): void {
if (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);
}
while (
this.seenValues.length > 0 &&
this.seenValues[0] === this.minValue + 1
) {
this.seenValues.shift();
this.minValue++;
}
}
}