andras/vault-link
andras/vault-link
Watch
1
0
Fork
You've already forked vault-link
 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 35

....

Browse source
This commit is contained in:
Andras Schmelczer 2026-09-27 09:04:05 +01:00
commit c869f0a880
32 changed files with 461 additions and 578 deletions
Download patch file Download diff file Expand all files Collapse all files

18
frontend/sync-client/src/errors/errors.ts
View file

@ -1,5 +1,3 @@
import type { RelativePath } from "../persistence/database";
export class SyncResetError extends Error {
public constructor() {
super("SyncClient has been reset, cleaning up");
@ -21,19 +19,9 @@ export class AuthenticationError extends Error {
}
}
export class FileNotFoundError extends Error {
public constructor(
message: string,
public readonly filePath: RelativePath
) {
super(message);
this.name = "FileNotFoundError";
}
}
export class PermanentSyncError extends Error {}
export class PermanentSyncError extends Error { }
export class LocalChangesDuringReconciliation extends Error { }
export class LocalChangesDuringReconciliation extends Error {}
/** The server rejected a checkpoint from a discarded/restored history. */
export class ServerHistoryChangedError extends Error { }
export class ServerHistoryChangedError extends Error {}

35
frontend/sync-client/src/file-operations/file-operations.ts
View file

@ -17,16 +17,14 @@ import { mergeContent } from "../sync-operations/content";
import { hash } from "../utils/hash";
import { isBinary } from "../utils/is-binary";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
import { decodeText } from "../utils/decode-text";
import type { ServerConfig } from "../services/server-config";
import {
applyLocalChange,
unappliedChanges,
type LocalChange
} from "../sync-operations/local-changes";
import {
FileNotFoundError,
LocalChangesDuringReconciliation
} from "../errors/errors";
import { LocalChangesDuringReconciliation } from "../errors/errors";
export interface FileWrite {
expected?: HashedSnapshot;
@ -62,15 +60,6 @@ export class FileOperations {
this.paths = new FilePaths(fs, isProtectedFile);
}
public async read(path: RelativePath): Promise<Uint8Array> {
const snapshot = await this.fs.read(path);
if (!snapshot)
throw new FileNotFoundError(`File not found: ${path}`, path);
return snapshot.content;
}
public async snapshot(
path: RelativePath
): Promise<HashedSnapshot | undefined> {
@ -84,10 +73,10 @@ export class FileOperations {
);
}
public async captureChange(
public captureChange(
change: LocalChange,
previous: readonly LocalChange[]
): Promise<void> {
): void {
const state = structuredClone(this.applying ?? this.database.state);
if (state.lastSeenUpdateId === undefined) return;
@ -249,27 +238,23 @@ export class FileOperations {
isBinary(value.content)
)
) {
const decoder = new TextDecoder("utf-8", {
fatal: true,
ignoreBOM: true
});
const parent = decoder.decode(base.content);
const remote = decoder.decode(replacement.content);
const parent = decodeText(base.content);
const remote = decodeText(replacement.content);
const text = await this.fs.atomicUpdateText(path, (latest) =>
reconcile(parent, latest, remote)
);
return hash(new TextEncoder().encode(text));
}
const result =
current && write.expected
current && base
? await mergeContent(
path,
write.expected,
base,
await hashSnapshot(current),
write.replacement,
replacement,
extensions
)
: write.replacement;
: replacement;
await this.fs.write(path, {
content: result.content,
cursors: result.cursors

36
frontend/sync-client/src/file-operations/file-paths.ts
View file

@ -1,14 +1,21 @@
import { v4 as uuid } from "uuid";
import type { FileSystemOperations } from "./filesystem-operations";
import type { DocumentId, EngineState, RelativePath } from "../persistence/database";
import type {
DocumentId,
EngineState,
RelativePath
} from "../persistence/database";
import { allocatePortablePath, isInternalPath } from "../utils/portable-path";
/** Resolve occupied rename destinations without discarding another file. */
export class FilePaths {
public constructor(
private readonly fs: FileSystemOperations,
private readonly isProtectedFile: (path: RelativePath, size: number) => boolean
) { }
private readonly isProtectedFile: (
path: RelativePath,
size: number
) => boolean
) {}
public async prepareDestination(
next: EngineState,
@ -18,7 +25,7 @@ export class FilePaths {
mutate: (action: () => Promise<void>) => Promise<void>
): Promise<RelativePath> {
try {
for (; ;) {
for (;;) {
const obstruction = await this.obstruction(path);
if (!obstruction) break;
const children =
@ -29,8 +36,8 @@ export class FilePaths {
for (const child of children) {
const occupant = Object.keys(current.local).find(
(key) =>
current.documents[key]?.observedHash !== undefined &&
current.local[key] === child
current.documents[key]?.observedHash !==
undefined && current.local[key] === child
);
const info = await this.fs.stat(child);
if (
@ -65,7 +72,9 @@ export class FilePaths {
next.local[occupant.id] = displaced;
}
if (obstruction.kind === "directory")
await mutate(() => this.fs.deleteDirectory(obstruction.path));
await mutate(() =>
this.fs.deleteDirectory(obstruction.path)
);
}
await this.ensureParents(path);
} catch (error) {
@ -113,14 +122,13 @@ export class FilePaths {
id: DocumentId,
path: RelativePath
): Promise<RelativePath> {
const occupied = { ...next.local };
delete occupied[id];
const files = (await this.fs.listFilesRecursively()).filter(
(path) => !isInternalPath(path)
const occupied = Object.entries(next.local).flatMap(
([other, occupiedPath]) => (other === id ? [] : [occupiedPath])
);
for (const [i, diskPath] of files.entries())
occupied[`disk-${i}`] = diskPath;
return allocatePortablePath(path, id, occupied);
const files = (await this.fs.listFilesRecursively()).filter(
(diskPath) => !isInternalPath(diskPath)
);
return allocatePortablePath(path, id, [...occupied, ...files]);
}
private async ensureParents(path: RelativePath): Promise<void> {

36
frontend/sync-client/src/persistence/database.ts
View file

@ -6,11 +6,6 @@ export type VaultUpdateId = number;
export type DocumentId = string;
export type RelativePath = string;
export interface DocumentMetadata {
parentVersionId: VaultUpdateId;
hash: string;
}
export interface ContentHead {
vaultUpdateId: VaultUpdateId;
contentSize: number;
@ -43,13 +38,6 @@ export interface EngineState {
export type StoredDatabase = EngineState;
// Read-only view used by cursor tracking and status reporting.
export interface DocumentRecord {
documentId: DocumentId;
relativePath: RelativePath;
metadata?: DocumentMetadata;
}
export function emptyState(vaultKey: string): StoredDatabase {
return {
vaultKey,
@ -125,29 +113,9 @@ export class Database {
return Object.keys(this.state.local).length;
}
public getDocumentByDocumentId(id: DocumentId): DocumentRecord | undefined {
const relativePath = this.state.local[id];
if (relativePath === undefined) return undefined;
const base = this.state.documents[id]?.base;
return {
documentId: id,
relativePath,
metadata:
base?.vaultUpdateId != null
? {
parentVersionId: base.vaultUpdateId,
hash: base.hash
}
: undefined
};
}
public getLatestDocumentByRelativePath(
path: RelativePath
): DocumentRecord | undefined {
const id = Object.keys(this.state.local).find(
public findDocumentId(path: RelativePath): DocumentId | undefined {
return Object.keys(this.state.local).find(
(id) => this.state.local[id] === path
);
return id === undefined ? undefined : this.getDocumentByDocumentId(id);
}
}

23
frontend/sync-client/src/services/server-config.ts
View file

@ -4,24 +4,20 @@ import {
ServerVersionMismatchError
} from "../errors/errors";
import type { SyncService } from "./sync-service";
export interface ServerConfigData {
mergeableFileExtensions: string[];
supportedApiVersion: number;
isAuthenticated: boolean;
}
import type { PingResponse } from "./types/PingResponse";
export class ServerConfig {
private config: Promise<ServerConfigData> | undefined;
private config?: Promise<PingResponse>;
public constructor(private readonly syncService: SyncService) { }
public constructor(private readonly syncService: SyncService) {}
private static validateConfig(config: ServerConfigData): void {
private static validateConfig(config: PingResponse): void {
if (config.supportedApiVersion !== SUPPORTED_API_VERSION) {
const shouldUpgradeClient =
config.supportedApiVersion > SUPPORTED_API_VERSION;
throw new ServerVersionMismatchError(
`Unsupported API version: ${config.supportedApiVersion}. Consider upgrading the ${shouldUpgradeClient ? "client" : "sync-server"
`Unsupported API version: ${config.supportedApiVersion}. Consider upgrading the ${
shouldUpgradeClient ? "client" : "sync-server"
} to ensure compatibility`
);
}
@ -33,11 +29,6 @@ export class ServerConfig {
}
}
// warm the cache
public async initialize(): Promise<void> {
await this.getConfig();
}
public async checkConnection(): Promise<{
isSuccessful: boolean;
message: string;
@ -59,7 +50,7 @@ export class ServerConfig {
}
}
public async getConfig(): Promise<ServerConfigData> {
public async getConfig(): Promise<PingResponse> {
if (!this.config) {
const pending = this.syncService
.ping()

81
frontend/sync-client/src/services/sync-service.ts
View file

@ -44,7 +44,7 @@ export class SyncService {
}
public async ping(): Promise<PingResponse> {
return this.request("/ping", undefined, true);
return this.request("/ping", { independent: true });
}
public async getVaultSnapshot(): Promise<VaultSnapshot> {
@ -58,17 +58,19 @@ export class SyncService {
public async pushFileManifest(
request: PushFileManifest
): Promise<FileManifestUpdateResponse> {
return this.request("/file-manifest", request);
return this.request("/file-manifest", { body: request });
}
public async putFileContent(
id: DocumentId,
request: PutFileContent
): Promise<DocumentUpdateResponse> {
return this.request(`/documents/${id}`, request);
return this.request(`/documents/${id}`, { body: request });
}
public async getDocumentMetadata(id: DocumentId): Promise<DocumentVersionWithoutContent> {
public async getDocumentMetadata(
id: DocumentId
): Promise<DocumentVersionWithoutContent> {
return this.request(`/documents/${id}/metadata`);
}
@ -81,19 +83,13 @@ export class SyncService {
}): Promise<Uint8Array> {
return this.request(
`/documents/${documentId}/versions/${vaultUpdateId}/content`,
undefined,
false,
async (response) => new Uint8Array(await response.arrayBuffer())
{
decode: async (response) =>
new Uint8Array(await response.arrayBuffer())
}
);
}
private getHistoryHeaders(): Record<string, string> {
const checkpoint = this.history.get();
return checkpoint !== undefined
? { "X-Vault-Link-History": checkpoint }
: {};
}
private async recordHistory(response: Response): Promise<void> {
const checkpoint = response.headers.get("x-vault-link-history");
if (checkpoint === null || checkpoint === "")
@ -102,48 +98,44 @@ export class SyncService {
if (checkpoint === previous) return;
if (
previous === undefined ||
Number(checkpoint.split(":")[0]) >=
Number(previous.split(":")[0])
Number(checkpoint.split(":")[0]) >= Number(previous.split(":")[0])
)
await this.history.save(checkpoint);
}
private checkHistory(response: Response): void {
if (response.headers.get("x-vault-link-history-mismatch") === "1")
throw new ServerHistoryChangedError(
"Server history changed; recovering local work"
);
}
private getUrl(path: string): string {
const { remoteUri, vaultName } = this.settings.getSettings();
return `${remoteUri.replace(/\/$/u, "")}/vaults/${encodeURIComponent(vaultName)}${path}`;
}
private async request<T>(
path: string,
body?: unknown,
raw = false,
decode: (response: Response) => Promise<T> = async (response) =>
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Endpoint methods supply the generated protocol response type.
response.json() as Promise<T>
{
body,
independent = false,
decode = async (response): Promise<T> =>
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Endpoint methods supply the generated protocol response type.
response.json() as Promise<T>
}: {
body?: unknown;
independent?: boolean;
decode?: (response: Response) => Promise<T>;
} = {}
): Promise<T> {
const timeout = AbortSignal.timeout(
this.settings.getSettings().requestTimeoutMs
);
const signal = raw
const { remoteUri, vaultName, token, requestTimeoutMs } =
this.settings.getSettings();
const url = `${remoteUri.replace(/\/$/u, "")}/vaults/${encodeURIComponent(vaultName)}${path}`;
const checkpoint = independent ? undefined : this.history.get();
const timeout = AbortSignal.timeout(requestTimeoutMs);
const signal = independent
? timeout
: AbortSignal.any([timeout, this.session.signal]);
const response = await abortable(signal, async () =>
this.fetchImplementation(this.getUrl(path), {
this.fetchImplementation(url, {
method: body === undefined ? "GET" : "PUT",
body: body === undefined ? undefined : JSON.stringify(body),
signal,
headers: {
...(raw ? {} : this.getHistoryHeaders()),
Authorization: `Bearer ${this.settings.getSettings().token}`,
...(checkpoint === undefined
? {}
: { "X-Vault-Link-History": checkpoint }),
Authorization: `Bearer ${token}`,
"Device-Id": this.deviceId,
"Content-Type": "application/json"
}
@ -151,7 +143,10 @@ export class SyncService {
);
signal.throwIfAborted();
this.checkHistory(response);
if (response.headers.get("x-vault-link-history-mismatch") === "1")
throw new ServerHistoryChangedError(
"Server history changed; recovering local work"
);
if (!response.ok) {
this.throwForErrorResponse(
@ -161,7 +156,7 @@ export class SyncService {
}
// Finish metadata saves even if the transport is paused meanwhile.
if (!raw) await this.recordHistory(response);
if (!independent) await this.recordHistory(response);
return abortable(signal, async () => decode(response));
}

14
frontend/sync-client/src/services/websocket-manager.test.ts
View file

@ -131,7 +131,7 @@ describe("WebSocketManager", () => {
await new Promise((resolve) => setTimeout(resolve, 50));
const { outstandingPromises } = manager as unknown as {
outstandingPromises: Promise<unknown>[];
outstandingPromises: Set<Promise<void>>;
};
const mockWs = (manager as unknown as { webSocket: MockWebSocket })
.webSocket;
@ -148,7 +148,7 @@ describe("WebSocketManager", () => {
await new Promise((resolve) => setTimeout(resolve, 100));
assert.strictEqual(outstandingPromises.length, 0);
assert.strictEqual(outstandingPromises.size, 0);
await manager.stop();
});
@ -167,7 +167,7 @@ describe("WebSocketManager", () => {
await new Promise((resolve) => setTimeout(resolve, 50));
const { outstandingPromises } = manager as unknown as {
outstandingPromises: Promise<unknown>[];
outstandingPromises: Set<Promise<void>>;
};
const mockWs = (manager as unknown as { webSocket: MockWebSocket })
.webSocket;
@ -178,7 +178,7 @@ describe("WebSocketManager", () => {
});
await new Promise((resolve) => setTimeout(resolve, 100));
assert.strictEqual(outstandingPromises.length, 0);
assert.strictEqual(outstandingPromises.size, 0);
await manager.stop();
});
@ -321,16 +321,16 @@ describe("WebSocketManager", () => {
await new Promise((resolve) => setTimeout(resolve, 10));
const { outstandingPromises } = manager as unknown as {
outstandingPromises: Promise<unknown>[];
outstandingPromises: Set<Promise<void>>;
};
assert.ok(outstandingPromises.length > 0);
assert.ok(outstandingPromises.size > 0);
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
resolveListener!();
await new Promise((resolve) => setTimeout(resolve, 50));
assert.strictEqual(outstandingPromises.length, 0);
assert.strictEqual(outstandingPromises.size, 0);
await manager.stop();
});
});

34
frontend/sync-client/src/services/websocket-manager.ts
View file

@ -5,7 +5,6 @@ import type { WebSocketClientMessage } from "./types/WebSocketClientMessage";
import type { CursorPositionFromClient } from "./types/CursorPositionFromClient";
import type { ClientCursors } from "./types/ClientCursors";
import { WEBSOCKET_RECEIVE_TIMEOUT_MS } from "../consts";
import { removeFromArray } from "../utils/remove-from-array";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { awaitAll } from "../utils/await-all";
@ -26,7 +25,7 @@ export class WebSocketManager {
private receiveTimeoutId: ReturnType<typeof setTimeout> | undefined;
private reconnectTimeoutId: ReturnType<typeof setTimeout> | undefined;
private readonly outstandingPromises: Promise<unknown>[] = [];
private readonly outstandingPromises = new Set<Promise<void>>();
private webSocket: WebSocket | undefined;
private readonly webSocketFactoryImplementation: typeof globalThis.WebSocket;
@ -39,16 +38,12 @@ export class WebSocketManager {
) {
if (webSocketImplementation) {
this.webSocketFactoryImplementation = webSocketImplementation;
} else if (typeof globalThis.WebSocket !== "undefined") {
this.webSocketFactoryImplementation = globalThis.WebSocket;
} else {
if (
typeof globalThis !== "undefined" &&
typeof globalThis.WebSocket === "undefined"
) {
// eslint-disable-next-line
this.webSocketFactoryImplementation = require("ws"); // polyfill for WebSocket in Node.js
} else {
this.webSocketFactoryImplementation = WebSocket;
}
this.webSocketFactoryImplementation =
// eslint-disable-next-line @typescript-eslint/no-require-imports, @typescript-eslint/no-unsafe-type-assertion -- Node versions without WebSocket use ws.
require("ws") as typeof globalThis.WebSocket;
}
}
@ -75,7 +70,7 @@ export class WebSocketManager {
}
public async waitUntilFinished(): Promise<void> {
await awaitAll(this.outstandingPromises);
await awaitAll([...this.outstandingPromises]);
}
public updateLocalCursors(cursorPositions: CursorPositionFromClient): void {
@ -182,12 +177,12 @@ export class WebSocketManager {
this.webSocket = socket;
this.expectResponse(socket);
this.webSocket.onerror = (): void => {
socket.onerror = (): void => {
this.logger.warn("WebSocket transport error");
this.reconnect(socket);
};
this.webSocket.onopen = (): void => {
socket.onopen = (): void => {
if (this.webSocket !== socket || this.isStopped) return;
try {
const handshake: WebSocketClientMessage = {
@ -209,7 +204,7 @@ export class WebSocketManager {
}
};
this.webSocket.onmessage = (event): void => {
socket.onmessage = (event): void => {
if (this.webSocket !== socket || this.isStopped) return;
try {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
@ -229,13 +224,10 @@ export class WebSocketManager {
);
})
.finally(() => {
removeFromArray(
this.outstandingPromises,
messageHandlingPromise
);
this.outstandingPromises.delete(messageHandlingPromise);
});
this.outstandingPromises.push(messageHandlingPromise);
this.outstandingPromises.add(messageHandlingPromise);
} catch (error) {
this.logger.error(
`Error parsing WebSocket message: ${String(error)}`
@ -243,7 +235,7 @@ export class WebSocketManager {
}
};
this.webSocket.onclose = (event): void => {
socket.onclose = (event): void => {
this.logger.warn(
`WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})`
);

28
frontend/sync-client/src/sync-client.ts
View file

@ -206,7 +206,7 @@ export class SyncClient {
})
}
);
const cursors = new CursorTracker(database, websocket, files, notifier);
const cursors = new CursorTracker(database, websocket, fs, notifier);
return new SyncClient(
logger,
history,
@ -274,13 +274,17 @@ export class SyncClient {
);
this.started = true;
this.configureTelemetry();
if (this.settings.getSettings().isSyncEnabled) await this.resume();
await this.resume();
});
}
private async resume(): Promise<void> {
if (this.destroyed) return;
if (!this.settings.getSettings().isSyncEnabled) return;
if (
!this.started ||
this.destroyed ||
!this.settings.getSettings().isSyncEnabled
)
return;
await this.database.bindVault(vaultKey(this.settings.getSettings()));
this.service.resume();
this.syncer.start();
@ -302,8 +306,7 @@ export class SyncClient {
this.serverConfig.reset();
this.cursorTracker.reset();
await this.syncer.retryRejectedRequests();
if (this.started && this.settings.getSettings().isSyncEnabled)
await this.resume();
await this.resume();
});
}
@ -338,17 +341,16 @@ export class SyncClient {
this.settings.getSettings().diffCacheSizeMB * 1024 * 1024
);
this.configureTelemetry();
if (this.started && this.settings.getSettings().isSyncEnabled)
await this.resume().catch((resumeError: unknown) => {
this.logger.error(
`Settings recovery could not resume sync: ${String(resumeError)}`
);
});
await this.resume().catch((resumeError: unknown) => {
this.logger.error(
`Settings recovery could not resume sync: ${String(resumeError)}`
);
});
throw error;
}
this.serverConfig.reset();
this.configureTelemetry();
if (this.started && next.isSyncEnabled) await this.resume();
await this.resume();
return [this.settings.getSettings(), old] as const;
});
await this.settings.onSettingsChanged.triggerAsync(...notification);

5
frontend/sync-client/src/sync-operations/content.ts
View file

@ -3,6 +3,7 @@ import { reconcile } from "reconcile-text";
import { hashSnapshot, type HashedSnapshot } from "../snapshot";
import { isBinary } from "../utils/is-binary";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
import { decodeText } from "../utils/decode-text";
export async function mergeContent(
path: RelativePath,
@ -31,9 +32,7 @@ export async function mergeContent(
if (!isFileTypeMergeable(path, extensions) || inputs.some(isBinary))
return remote;
const [parentText, localText, remoteText] = inputs.map((bytes) =>
new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode(bytes)
);
const [parentText, localText, remoteText] = inputs.map(decodeText);
const merged = reconcile(
parentText,

87
frontend/sync-client/src/sync-operations/cursor-tracker.test.ts
View file

@ -1,29 +1,35 @@
/* eslint-disable @typescript-eslint/no-unsafe-type-assertion */
import assert from "node:assert/strict";
import { it } from "node:test";
import type {
import {
Database,
RelativePath,
VaultUpdateId
emptyState,
type RelativePath,
type VaultUpdateId
} from "../persistence/database";
import type { FileOperations } from "../file-operations/file-operations";
import type { FileSystemOperations } from "../file-operations/filesystem-operations";
import type { WebSocketManager } from "../services/websocket-manager";
import type { ClientCursors } from "../services/types/ClientCursors";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { hash } from "../utils/hash";
import { CursorTracker } from "./cursor-tracker";
function createDatabase(digest: string): Database {
const state = emptyState("test");
state.local.document = "note.md";
state.documents.document = { base: { vaultUpdateId: 1, hash: digest } };
return new Database(
state,
async () => undefined,
"test",
async () => undefined
);
}
it("keeps the last usable cursor while retaining only the latest future update", async (context) => {
const content = new TextEncoder().encode("current");
const contentHash = await hash(content);
let version = 1;
const database = {
getDocumentByDocumentId: () => ({
documentId: "document",
relativePath: "note.md",
metadata: { parentVersionId: version, hash: contentHash }
})
} as unknown as Database;
const database = createDatabase(contentHash);
const websocket = {
onWebSocketStatusChanged: new EventListeners(),
onRemoteCursorsUpdateReceived: new EventListeners<
@ -35,7 +41,7 @@ it("keeps the last usable cursor while retaining only the latest future update",
const tracker = new CursorTracker(
database,
websocket,
{ read: async () => content } as unknown as FileOperations,
{ read: async () => ({ content }) } as unknown as FileSystemOperations,
notifier
);
context.after(() => tracker.reset());
@ -61,14 +67,20 @@ it("keeps the last usable cursor while retaining only the latest future update",
]);
assert.equal(received[0].documentsWithCursors[0].vault_update_id, 1);
}
version = 2;
database.state.documents.document.base = {
vaultUpdateId: 2,
hash: contentHash
};
await notifier.triggerAsync("note.md");
assert.equal(
received[0].documentsWithCursors[0].vault_update_id,
1,
"superseded future cursors must not reappear"
);
version = 3;
database.state.documents.document.base = {
vaultUpdateId: 3,
hash: contentHash
};
await notifier.triggerAsync("note.md");
assert.equal(received[0].documentsWithCursors[0].vault_update_id, 3);
assert.equal(received[0].documentsWithCursors[0].relative_path, "note.md");
@ -80,17 +92,7 @@ it("refreshes unchanged cursors on reconnect and heartbeat, and follows the docu
context.mock.timers.enable({ apis: ["setInterval"] });
const content = new TextEncoder().encode("hello");
let path = "note.md";
const record = () => ({
documentId: "document",
relativePath: path,
metadata: { parentVersionId: 1, hash: contentHash }
});
const contentHash = await hash(content);
const database = {
getLatestDocumentByRelativePath: (candidate: RelativePath) =>
candidate === path ? record() : undefined,
getDocumentByDocumentId: () => record()
} as unknown as Database;
const database = createDatabase(await hash(content));
const sent: {
documentsWithCursors: {
relative_path: RelativePath;
@ -110,7 +112,7 @@ it("refreshes unchanged cursors on reconnect and heartbeat, and follows the docu
const tracker = new CursorTracker(
database,
websocket,
{ read: async () => content } as unknown as FileOperations,
{ read: async () => ({ content }) } as unknown as FileSystemOperations,
notifier
);
const drain = async () => {
@ -133,6 +135,7 @@ it("refreshes unchanged cursors on reconnect and heartbeat, and follows the docu
await drain();
assert.equal(sent.length, 3);
path = "renamed.md";
database.state.local.document = path;
await notifier.triggerAsync(path);
assert.equal(sent.at(-1)!.documentsWithCursors[0].relative_path, path);
websocket.onWebSocketStatusChanged.trigger(false);
@ -155,13 +158,7 @@ for (const interruption of ["disconnect", "reset"] as const) {
const contentHash = await hash(content);
const entered = Promise.withResolvers<void>();
const release = Promise.withResolvers<void>();
const database = {
getDocumentByDocumentId: () => ({
documentId: "document",
relativePath: "note.md",
metadata: { parentVersionId: 1, hash: contentHash }
})
} as unknown as Database;
const database = createDatabase(contentHash);
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
@ -178,9 +175,9 @@ for (const interruption of ["disconnect", "reset"] as const) {
read: async () => {
entered.resolve();
await release.promise;
return content;
return { content };
}
} as unknown as FileOperations,
} as unknown as FileSystemOperations,
new EventListeners<(path: RelativePath) => unknown>()
);
let received: ClientCursors[] = [];
@ -223,7 +220,7 @@ for (const interruption of ["disconnect", "reset"] as const) {
it("a remote cursor becomes outdated if the document advances during its snapshot read", async () => {
const content = new TextEncoder().encode("old");
let metadata = { parentVersionId: 1, hash: await hash(content) };
const database = createDatabase(await hash(content));
const entered = Promise.withResolvers<void>(),
release = Promise.withResolvers<void>();
const websocket = {
@ -236,21 +233,15 @@ it("a remote cursor becomes outdated if the document advances during its snapsho
updateLocalCursors: () => {}
} as unknown as WebSocketManager;
const tracker = new CursorTracker(
{
getDocumentByDocumentId: () => ({
documentId: "document",
relativePath: "note.md",
metadata
})
} as unknown as Database,
database,
websocket,
{
read: async () => {
entered.resolve();
await release.promise;
return content;
return { content };
}
} as unknown as FileOperations,
} as unknown as FileSystemOperations,
new EventListeners<(path: RelativePath) => unknown>()
);
let outdated: boolean | undefined;
@ -272,8 +263,8 @@ it("a remote cursor becomes outdated if the document advances during its snapsho
}
]);
await entered.promise;
metadata = {
parentVersionId: 2,
database.state.documents.document.base = {
vaultUpdateId: 2,
hash: await hash(new TextEncoder().encode("new content"))
};
release.resolve();

112
frontend/sync-client/src/sync-operations/cursor-tracker.ts
View file

@ -1,4 +1,5 @@
import type { FileOperations } from "../file-operations/file-operations";
/* eslint-disable @typescript-eslint/no-unnecessary-condition -- Persisted document and path maps are sparse at runtime. */
import type { FileSystemOperations } from "../file-operations/filesystem-operations";
import type {
Database,
DocumentId,
@ -15,13 +16,8 @@ import { Lock } from "../utils/data-structures/locks";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { CURSOR_HEARTBEAT_INTERVAL_MS } from "../consts";
// Cursor positions are updated separately from documents. However, a given cursor position is only
// valid within a certain version of the document it belongs to. This class tracks previous and the latest
// known remote cursor positions, and for each document, tries to return the latest cursor positions that are
// not from the future.
/** Retain each device's last usable cursors and latest update awaiting content. */
export class CursorTracker {
// The returned position may be accurate, if it matches the document version, or outdated, in which case
// the client has to heuristically guess it's current position based on the local edits.
public readonly onRemoteCursorsUpdated = new EventListeners<
(cursors: MaybeOutdatedClientCursors[]) => unknown
>();
@ -43,19 +39,15 @@ export class CursorTracker {
public constructor(
private readonly database: Database,
private readonly webSocketManager: WebSocketManager,
private readonly fileOperations: FileOperations,
private readonly fs: FileSystemOperations,
fileChanges: EventListeners<(path: RelativePath) => unknown>
) {
this.webSocketManager.onWebSocketStatusChanged.add((connected) => {
this.sessionGeneration++;
clearInterval(this.cursorTimer);
this.lastCursorInput = "[]";
this.lastCursorMessage = "[]";
if (!connected) {
this.knownRemoteCursors = [];
this.onRemoteCursorsUpdated.trigger([]);
this.reset();
return;
}
this.resetSession();
const refresh = () => {
void this.publishLocalCursors(true).catch(() => {
// A concurrent move/deletion can invalidate an editor view;
@ -111,9 +103,9 @@ export class CursorTracker {
clientCursor.documentsWithCursors.some(
(document) =>
document.relative_path === relativePath ||
this.database.getDocumentByDocumentId(
this.database.state.local[
document.document_id
)?.relativePath === relativePath
] === relativePath
)
) {
clientCursor.upToDateness =
@ -130,9 +122,7 @@ export class CursorTracker {
});
}
/// Update the local cursors for the given documents.
/// Can be called frequently as it only emits an event
/// if the state has actually changed.
/** Publish changed editor selections; heartbeats also refresh unchanged ones. */
public async sendLocalCursorsToServer(
documentToCursors: Record<RelativePath, CursorSpan[]>
): Promise<void> {
@ -150,25 +140,21 @@ export class CursorTracker {
for (const [editorPath, cursors] of Object.entries(
this.localCursors
)) {
const id = this.localCursorDocuments[editorPath];
const record = id
? this.database.getDocumentByDocumentId(id)
: this.database.getLatestDocumentByRelativePath(editorPath);
if (!record) {
continue; // Let's wait for the file to be created before sending cursors
}
this.localCursorDocuments[editorPath] = record.documentId;
const { relativePath } = record;
if (!record.metadata) {
continue; // this is a new document, no need to sync the cursors
}
const id =
this.localCursorDocuments[editorPath] ??
this.database.findDocumentId(editorPath);
if (id === undefined) continue;
this.localCursorDocuments[editorPath] = id;
const relativePath = this.database.state.local[id];
const base = this.database.state.documents[id]?.base;
// Wait until the document has a path and an acknowledged version.
if (relativePath === undefined || base?.vaultUpdateId == null)
continue;
documentsWithCursors.push({
relative_path: relativePath,
document_id: record.documentId,
vault_update_id: record.metadata.parentVersionId,
document_id: id,
vault_update_id: base.vaultUpdateId,
cursors: cursors.map(({ start, end }) => ({
start: Math.min(start, end),
end: Math.max(start, end)
@ -195,10 +181,14 @@ export class CursorTracker {
}
public reset(): void {
this.sessionGeneration++;
clearInterval(this.cursorTimer);
this.resetSession();
this.knownRemoteCursors = [];
this.onRemoteCursorsUpdated.trigger([]);
}
private resetSession(): void {
this.sessionGeneration++;
clearInterval(this.cursorTimer);
this.lastCursorInput = "[]";
this.lastCursorMessage = "[]";
// Keep serialization intact while old reads drain. A generation fence
@ -232,9 +222,8 @@ export class CursorTracker {
(document) => ({
...document,
relative_path:
this.database.getDocumentByDocumentId(
document.document_id
)?.relativePath ?? document.relative_path
this.database.state.local[document.document_id] ??
document.relative_path
})
),
isOutdated:
@ -250,8 +239,6 @@ export class CursorTracker {
return result;
}
// We store up-to-dateness on a per-client basis to simplify the implementation.
// An individual client won't have too many documents open at once, so this is a reasonable trade-off.
private async getDocumentsUpToDateness(
clientCursor: ClientCursors
): Promise<DocumentUpToDateness> {
@ -269,29 +256,19 @@ export class CursorTracker {
): Promise<DocumentUpToDateness> {
if (document.vault_update_id === null)
return DocumentUpToDateness.Prior;
const record = this.database.getDocumentByDocumentId(
document.document_id
);
if (!record) {
// the document of the cursor must be from the future
const path = this.database.state.local[document.document_id];
if (path === undefined) return DocumentUpToDateness.Later;
const version =
this.database.state.documents[document.document_id]?.base
?.vaultUpdateId ?? 0;
if (version < document.vault_update_id)
return DocumentUpToDateness.Later;
}
if (
(record.metadata?.parentVersionId ?? 0) < document.vault_update_id
) {
return DocumentUpToDateness.Later;
} else if (
document.vault_update_id < (record.metadata?.parentVersionId ?? 0)
) {
// the document of the cursor must be from the past
if (version > document.vault_update_id)
return DocumentUpToDateness.Prior;
}
return (await this.matchesLocalVersion({
...document,
relative_path: record.relativePath
relative_path: path
}))
? DocumentUpToDateness.UpToDate
: DocumentUpToDateness.Prior;
@ -303,16 +280,17 @@ export class CursorTracker {
relative_path,
vault_update_id
}: DocumentWithCursors): Promise<boolean> {
const content = await this.fileOperations
const snapshot = await this.fs
.read(relative_path)
.catch(() => undefined);
if (!content) return false;
const digest = await hash(content);
const current = this.database.getDocumentByDocumentId(document_id);
if (!snapshot) return false;
const digest = await hash(snapshot.content);
const { local, documents } = this.database.state;
const base = documents[document_id]?.base;
return (
current?.metadata?.parentVersionId === vault_update_id &&
current.relativePath === relative_path &&
current.metadata.hash === digest
base?.vaultUpdateId === vault_update_id &&
local[document_id] === relative_path &&
base.hash === digest
);
}
}

12
frontend/sync-client/src/sync-operations/file-manifest.ts
View file

@ -33,21 +33,17 @@ export function resolvePaths(
const result: FileManifestEntries = {};
for (const id of protectedIds)
if (proposed[id] !== undefined) result[id] = proposed[id];
const occupied = { ...result };
for (const [index, path] of reservedPaths.entries())
if (!Object.values(result).includes(path))
occupied[`reserved-${index}`] = path;
const occupied = [...Object.values(result), ...reservedPaths];
const ids = Object.keys(proposed).sort(
(a, b) =>
Number(remote[b] === proposed[b]) -
Number(remote[a] === proposed[a]) ||
(a < b ? -1 : a > b ? 1 : 0)
);
for (const id of ids.filter(
(candidate) => result[candidate] === undefined
)) {
for (const id of ids) {
if (result[id] !== undefined) continue;
result[id] = allocatePortablePath(proposed[id], id, occupied);
occupied[id] = result[id];
occupied.push(result[id]);
}
return result;
}

47
frontend/sync-client/src/sync-operations/outgoing-cursor.test.ts
View file

@ -3,13 +3,12 @@ import { test } from "node:test";
import { CursorTracker } from "./cursor-tracker";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { hash } from "../utils/hash";
import type {
import {
Database,
DocumentId,
DocumentRecord,
RelativePath
emptyState,
type RelativePath
} from "../persistence/database";
import type { FileOperations } from "../file-operations/file-operations";
import type { FileSystemOperations } from "../file-operations/filesystem-operations";
import type { WebSocketManager } from "../services/websocket-manager";
import type { DocumentWithCursors } from "../services/types/DocumentWithCursors";
@ -18,11 +17,15 @@ for (const change of ["version", "identity", "path", "delete"] as const)
// Equal bytes deliberately defeat a freshness check that compares only hashes.
const content = new TextEncoder().encode("same bytes");
const digest = await hash(content);
let current: DocumentRecord | undefined = {
documentId: "a",
relativePath: "a.md",
metadata: { parentVersionId: 1, hash: digest }
};
const state = emptyState("test");
state.local.a = "a.md";
state.documents.a = { base: { vaultUpdateId: 1, hash: digest } };
const database = new Database(
state,
async () => undefined,
"test",
async () => undefined
);
const sent: { documentsWithCursors: DocumentWithCursors[] }[] = [];
const websocket = {
onWebSocketStatusChanged: new EventListeners(),
@ -33,24 +36,22 @@ for (const change of ["version", "identity", "path", "delete"] as const)
};
const files = {
read: async () => {
if (change === "delete") current = undefined;
const current = database.state;
if (change === "delete") delete current.local.a;
else if (change === "version")
current!.metadata!.parentVersionId = 2;
else if (change === "identity")
current!.documentId = "replacement";
else current!.relativePath = "moved.md";
return content;
current.documents.a.base!.vaultUpdateId = 2;
else if (change === "identity") {
current.local.replacement = current.local.a;
current.documents.replacement = current.documents.a;
delete current.local.a;
} else current.local.a = "moved.md";
return { content };
}
};
const tracker = new CursorTracker(
{
getDocumentByDocumentId: (id: DocumentId) =>
current?.documentId === id ? current : undefined,
getLatestDocumentByRelativePath: (path: RelativePath) =>
current?.relativePath === path ? current : undefined
} as unknown as Database,
database,
websocket as unknown as WebSocketManager,
files as unknown as FileOperations,
files as unknown as FileSystemOperations,
new EventListeners<(path: RelativePath) => unknown>()
);
await tracker.sendLocalCursorsToServer({

46
frontend/sync-client/src/sync-operations/scan.ts
View file

@ -1,6 +1,10 @@
import { v4 as uuid } from "uuid";
import type { FileOperations } from "../file-operations/file-operations";
import type { DocumentId, EngineState, RelativePath } from "../persistence/database";
import type {
Database,
DocumentId,
RelativePath
} from "../persistence/database";
import type { VaultSnapshot } from "../services/types/VaultSnapshot";
import { EMPTY_HASH, hash } from "../utils/hash";
import {
@ -16,28 +20,27 @@ import {
export type { LocalChange } from "./local-changes";
interface ScanOptions {
next: EngineState;
database: Database;
guard: () => void;
changes: LocalChange[];
files: FileOperations;
ignored: (path: RelativePath) => boolean;
oversized: (size: number) => boolean;
commit: (next: EngineState) => Promise<void>;
}
/** Scans are complete or fail. Missing unmaterialized/ignored entries are not deletions. */
export async function scanLocalFiles(
{
next,
database,
guard,
changes: queuedChanges,
files,
ignored,
oversized,
commit
oversized
}: ScanOptions,
initial?: VaultSnapshot
): Promise<Set<RelativePath>> {
const next = structuredClone(database.state);
// Leave the prefix queued until its resulting state is saved.
// A persisted change ID makes replay safe after an uncertain save.
const changes = [...queuedChanges];
@ -61,7 +64,7 @@ export async function scanLocalFiles(
);
}
const present = new Set(paths);
const snapshots = new Map<RelativePath, { hash: string }>();
const hashes = new Map<RelativePath, string>();
for (const path of paths) {
if (ignored(path)) {
unsyncablePaths.add(path);
@ -77,7 +80,7 @@ export async function scanLocalFiles(
const snapshot = await files.fs.read(path);
if (!snapshot) throw new Error(`File changed during scan: ${path}`);
if (!oversized(snapshot.content.length)) {
snapshots.set(path, { hash: await hash(snapshot.content) });
hashes.set(path, await hash(snapshot.content));
} else {
unsyncablePaths.add(path);
}
@ -91,20 +94,18 @@ export async function scanLocalFiles(
)
);
const knownPaths = new Set(Object.values(next.local));
const unknown = [...snapshots.keys()].filter(
(path) => !knownPaths.has(path)
);
const unknown = [...hashes.keys()].filter((path) => !knownPaths.has(path));
for (const path of unknown) {
const snapshot = snapshots.get(path)!;
const digest = hashes.get(path)!;
const matches = [...missing].filter(
(id) => next.documents[id]?.observedHash === snapshot.hash
(id) => next.documents[id]?.observedHash === digest
);
const sameHash = unknown.filter(
(other) => snapshots.get(other)?.hash === snapshot.hash
(other) => hashes.get(other) === digest
);
let id: DocumentId | undefined;
if (
snapshot.hash !== EMPTY_HASH &&
digest !== EMPTY_HASH &&
matches.length === 1 &&
sameHash.length === 1
) {
@ -131,24 +132,21 @@ export async function scanLocalFiles(
if (next.excluded[id] === undefined)
next.excluded[id] = next.fileManifest.entries[id] ?? null;
}
const snapshot = snapshots.get(path);
if (snapshot) {
const digest = hashes.get(path);
if (digest !== undefined) {
const doc = (next.documents[id] ??= {});
// The old history no longer contains this base. Preserve any edit
// made during recovery as a new file instead of overwriting it.
if (
doc.base?.vaultUpdateId === null &&
doc.base.hash !== snapshot.hash
) {
if (doc.base?.vaultUpdateId === null && doc.base.hash !== digest) {
const recoveredId = uuid();
next.local[recoveredId] = path;
next.documents[recoveredId] = { observedHash: snapshot.hash };
next.documents[recoveredId] = { observedHash: digest };
delete next.local[id];
} else doc.observedHash = snapshot.hash;
} else doc.observedHash = digest;
}
}
guard();
await commit(next);
await database.commit(next);
queuedChanges.splice(0, changes.length);
guard();
return unsyncablePaths;

16
frontend/sync-client/src/sync-operations/syncer.test.ts
View file

@ -135,11 +135,9 @@ describe("Syncer per-file state and event delivery", () => {
function fixture() {
const database = {
state: {
local: { document: "note.md" },
documents: {
document: {
observedHash: "hash",
base: { ...head, hash: "hash" }
}
@ -150,13 +148,8 @@ describe("Syncer per-file state and event delivery", () => {
},
lastSeenUpdateId: 0
},
getLatestDocumentByRelativePath: (path: RelativePath) =>
path === "note.md"
? {
documentId: "document",
relativePath: path
}
: undefined
findDocumentId: (path: RelativePath) =>
path === "note.md" ? "document" : undefined
} as unknown as Database;
const websocket = {
onWebSocketStatusChanged: new EventListeners<
@ -216,7 +209,10 @@ describe("Syncer per-file state and event delivery", () => {
contentHead: (value: typeof head) => Record<string, unknown>;
}
).contentHead({ ...head, contentBase64: "secret" } as typeof head);
assert.deepEqual(projected, { vaultUpdateId: head.vaultUpdateId, contentSize: head.contentSize });
assert.deepEqual(projected, {
vaultUpdateId: head.vaultUpdateId,
contentSize: head.contentSize
});
assert.strictEqual("contentBase64" in projected, false);
});
});

67
frontend/sync-client/src/sync-operations/syncer.ts
View file

@ -42,6 +42,7 @@ import {
import { isInternalPath } from "../utils/portable-path";
import { isBinary } from "../utils/is-binary";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
import { decodeText } from "../utils/decode-text";
import type { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache";
import {
AuthenticationError,
@ -126,14 +127,14 @@ export class Syncer {
return false;
}
const record = this.database.getLatestDocumentByRelativePath(path);
if (!record) return false;
const document = this.database.state.documents[record.documentId];
const id = this.database.findDocumentId(path);
if (id === undefined) return false;
const document = this.database.state.documents[id];
if (
document?.observedHash === undefined ||
!document.base ||
document.observedHash !== document.base.hash ||
this.database.state.fileManifest.entries[record.documentId] !== path
this.database.state.fileManifest.entries[id] !== path
) {
return false;
}
@ -143,16 +144,13 @@ export class Syncer {
return false;
const { pending } = this.database.state;
if (
pending?.type === "content" &&
pending.documentId === record.documentId
) {
if (pending?.type === "content" && pending.documentId === id) {
return false;
}
if (
pending?.type === "fileManifest" &&
pending.request.entries[record.documentId] !==
this.database.state.fileManifest.entries[record.documentId]
pending.request.entries[id] !==
this.database.state.fileManifest.entries[id]
) {
return false;
}
@ -234,22 +232,18 @@ export class Syncer {
if (rejection !== undefined) throw new PermanentSyncError(rejection);
}
public async syncLocallyCreatedFile(_path: RelativePath): Promise<void> {
this.localGeneration++;
this.pendingPaths.add(_path);
public async syncLocallyCreatedFile(path: RelativePath): Promise<void> {
await this.recordLocalChange({
type: "create",
path: _path,
path,
changeId: uuid()
});
}
public async syncLocallyDeletedFile(_path: RelativePath): Promise<void> {
this.localGeneration++;
this.pendingPaths.add(_path);
public async syncLocallyDeletedFile(path: RelativePath): Promise<void> {
await this.recordLocalChange({
type: "delete",
path: _path,
path,
changeId: uuid()
});
}
@ -261,12 +255,6 @@ export class Syncer {
oldPath?: RelativePath;
relativePath: RelativePath;
}): Promise<void> {
// Content snapshots are re-read before writing. Only namespace
// notifications invalidate an identity plan; keystrokes must not starve
// a complete scan of unrelated documents.
if (oldPath !== undefined && oldPath !== "") this.localGeneration++;
this.pendingPaths.add(relativePath);
if (oldPath) this.pendingPaths.add(oldPath);
if (
oldPath &&
!isInternalPath(oldPath) &&
@ -278,8 +266,11 @@ export class Syncer {
relativePath,
changeId: uuid()
});
} else {
// Keystrokes wake reconciliation without invalidating identity plans.
this.pendingPaths.add(relativePath);
this.wake();
}
this.wake();
}
/** Reload an uncertain metadata save before starting another attempt. */
@ -289,10 +280,15 @@ export class Syncer {
}
private async recordLocalChange(change: LocalChange): Promise<void> {
this.localGeneration++;
if (change.type === "move") {
this.pendingPaths.add(change.oldPath);
this.pendingPaths.add(change.relativePath);
} else this.pendingPaths.add(change.path);
const previous = [...this.changes];
this.changes.push(change);
const saving = this.notificationSaving.then(async () => {
await this.files.captureChange(change, previous);
this.files.captureChange(change, previous);
await this.localChangeStorage.save(structuredClone(this.changes));
this.notificationSaveFailed = false;
});
@ -323,7 +319,7 @@ export class Syncer {
if (rejection !== undefined)
throw new PermanentSyncError(rejection);
if (!this.isFirstSyncComplete) await this.service.resetHistory();
await this.serverConfig.initialize();
await this.serverConfig.getConfig();
if (!this.isFirstSyncComplete) await this.bootstrap();
if (!this.stopped) this.onReadyForEvents.trigger();
do {
@ -452,12 +448,7 @@ export class Syncer {
(await this.serverConfig.getConfig()).mergeableFileExtensions
)
) {
const decode = (bytes: Uint8Array): string =>
new TextDecoder("utf-8", {
fatal: true,
ignoreBOM: true
}).decode(bytes);
const value = diff(decode(parent), decode(content));
const value = diff(decodeText(parent), decodeText(content));
// Match the server's work bound; large edit sets remain syncable.
if (value.length <= 10_000) return { type: "Diff", value };
}
@ -482,13 +473,12 @@ export class Syncer {
const hadChanges = this.changes.length > 0;
this.unsyncablePaths = await scanLocalFiles(
{
next: this.next(),
database: this.database,
guard: this.localGuard(),
changes: this.changes,
files: this.files,
ignored: (path) => this.settings.isIgnored(path),
oversized: (size) => this.settings.isOversized(size),
commit: async (next) => this.database.commit(next)
oversized: (size) => this.settings.isOversized(size)
},
initial
);
@ -969,10 +959,7 @@ export class Syncer {
: this.contentCache.get(parentVersionId);
// A restart may have evicted the parent. Future uploads can use snapshots.
if (!parent) return;
const text = new TextDecoder("utf-8", {
fatal: true,
ignoreBOM: true
}).decode(parent);
const text = decodeText(parent);
bytes = new TextEncoder().encode(undiff(text, content.value));
}
await this.cacheContent(

18
frontend/sync-client/src/utils/await-all.test.ts
View file

@ -54,3 +54,21 @@ void test("awaitAll works with async functions", async () => {
assert.strictEqual(results[0], "async");
assert.strictEqual(results[1], 123);
});
void test("awaitAll drains unfinished work before reporting a rejection", async () => {
const pending = Promise.withResolvers<undefined>();
const error = new Error("failed first");
let finished = false;
const checked = assert
.rejects(awaitAll([Promise.reject(error), pending.promise]), error)
.then(() => {
finished = true;
});
try {
await new Promise((resolve) => setImmediate(resolve));
assert.equal(finished, false);
} finally {
pending.resolve(undefined);
}
await checked;
});

27
frontend/sync-client/src/utils/await-all.ts
View file

@ -1,25 +1,18 @@
type PromiseTuple<T extends readonly unknown[]> = readonly [
...{ [K in keyof T]: Promise<T[K]> }
];
type ResolvedTuple<T extends readonly unknown[]> = {
[K in keyof T]: T[K];
-readonly [K in keyof T]: Awaited<T[K]>;
};
export const awaitAll = async <T extends readonly unknown[]>(
promises: PromiseTuple<T>
): Promise<ResolvedTuple<T>> => {
/** Await every operation before returning results or throwing the first error. */
export async function awaitAll<T extends readonly unknown[] | []>(
promises: T
): Promise<ResolvedTuple<T>> {
// eslint-disable-next-line no-restricted-properties
const result = await Promise.allSettled(promises);
for (const res of result) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Array.map preserves tuple order and length.
return result.map((res) => {
if (res.status === "rejected") {
throw res.reason;
}
}
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return result.map(
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
(res) => (res as PromiseFulfilledResult<unknown>).value
) as ResolvedTuple<T>;
};
return res.value;
}) as ResolvedTuple<T>;
}

46
frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts
View file

@ -6,6 +6,10 @@ export function slowWebSocketFactory(
jitterScaleInSeconds: number,
logger: Logger
): typeof WebSocket {
const delay = async (): Promise<void> => {
if (jitterScaleInSeconds > 0)
await sleep(Math.random() * jitterScaleInSeconds * 1000);
};
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return class FlakyWebSocket extends WebSocket {
private readonly receiving = new Lock();
@ -13,10 +17,7 @@ export function slowWebSocketFactory(
public set onopen(callback: ((event: Event) => void) | null) {
super.onopen = async (event: Event): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
await delay();
callback?.(event);
};
}
@ -24,12 +25,7 @@ export function slowWebSocketFactory(
public set onmessage(callback: ((event: MessageEvent) => void) | null) {
super.onmessage = async (event: MessageEvent): Promise<void> => {
await this.receiving.withLock(async () => {
if (jitterScaleInSeconds > 0) {
await sleep(
Math.random() * jitterScaleInSeconds * 1000
);
}
await delay();
callback?.(event);
});
};
@ -37,18 +33,14 @@ export function slowWebSocketFactory(
public set onclose(callback: ((event: CloseEvent) => void) | null) {
super.onclose = async (event: CloseEvent): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
await delay();
callback?.(event);
};
}
public set onerror(callback: ((event: Event) => void) | null) {
super.onerror = async (event: Event): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
await delay();
callback?.(event);
};
}
@ -56,21 +48,15 @@ export function slowWebSocketFactory(
public send(
data: string | ArrayBufferLike | Blob | ArrayBufferView
): void {
this.waitingSend(data).catch((error: unknown) => {
logger.error(`Error sending WebSocket message: ${error}`);
});
}
private async waitingSend(
data: string | ArrayBufferLike | Blob | ArrayBufferView
): Promise<void> {
// maintain message order
await this.sending.withLock(async () => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
super.send(data);
});
void this.sending
.withLock(async () => {
await delay();
super.send(data);
})
.catch((error: unknown) => {
logger.error(`Error sending WebSocket message: ${error}`);
});
}
} as unknown as typeof WebSocket;
}

Some files were not shown because too many files have changed in this diff Show more