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 31

wip

Browse source
This commit is contained in:
Andras Schmelczer 2026-09-23 08:23:17 +01:00
commit 4c002f9acd
20 changed files with 14543 additions and 4367 deletions
Download patch file Download diff file Expand all files Collapse all files

1
frontend/sync-client/src/consts.ts
View file

@ -1,5 +1,4 @@
export const TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS = 60;
export const DIFF_CACHE_SIZE_MB = 2;
export const MAX_LOG_MESSAGE_COUNT = 100000;
export const MAX_HISTORY_ENTRY_COUNT = 5000;
export const SUPPORTED_API_VERSION = 4;

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

@ -1,291 +1,277 @@
import type { Logger } from "../tracing/logger";
import type { FileSystemOperations } from "./filesystem-operations";
import type { Database, RelativePath } from "../persistence/database";
import { SafeFileSystemOperations } from "./safe-filesystem-operations";
import type { TextWithCursors } from "reconcile-text";
import { reconcile } from "reconcile-text";
import { isFileTypeMergable } from "../utils/is-file-type-mergable";
import type {
FileSnapshot,
FileSystemOperations
} from "./filesystem-operations";
import type {
Database,
EngineState,
StoredSnapshot
} from "../persistence/database";
import { isInternalPath } from "../utils/portable-path";
import { FilePaths } from "./file-paths";
import {
fromStoredSnapshot,
toStoredSnapshot,
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 type { ServerConfig } from "../services/server-config";
import {
applyLocalChange,
unappliedChanges,
type LocalChange
} from "../sync-operations/local-changes";
import {
FileNotFoundError,
LocalChangesDuringReconciliation
} from "../errors/errors";
export interface FileWrite {
expected?: StoredSnapshot;
replacement: StoredSnapshot;
}
type Mutation = (action: () => Promise<void>) => Promise<void>;
/** Apply files first, then save metadata. Repeating a content merge is idempotent: reconcile(p, a, a) == a */
export class FileOperations {
private static readonly PARENTHESES_REGEX = / \((?<count>\d+)\)$/;
private readonly fs: SafeFileSystemOperations;
private applying?: EngineState;
private readonly paths: FilePaths;
public constructor(
private readonly logger: Logger,
public readonly fs: FileSystemOperations,
private readonly database: Database,
fs: FileSystemOperations,
private readonly serverConfig: ServerConfig,
private readonly nativeLineEndings = "\n"
private readonly filesApplied: (
paths: readonly string[]
) => void = () => undefined,
private readonly localChanges: {
entries: () => readonly LocalChange[];
flush: () => Promise<void>;
} = { entries: () => [], flush: async () => undefined },
private readonly isProtectedFile: (
path: string,
size: number
) => boolean = () => false
) {
this.fs = new SafeFileSystemOperations(fs, logger);
this.paths = new FilePaths(fs, isProtectedFile);
}
private static getParentDirAndFile(
path: RelativePath
): [RelativePath, RelativePath] {
const pathParts = path.split("/");
const fileName = pathParts.pop();
if (fileName == null || fileName === "") {
throw new Error(`Path '${path}' cannot be empty`);
}
public async read(path: string): Promise<Uint8Array> {
const snapshot = await this.fs.read(path);
return [pathParts.join("/"), fileName];
if (!snapshot)
throw new FileNotFoundError(`File not found: ${path}`, path);
return snapshot.content;
}
public async listFilesRecursively(
root: RelativePath | undefined = undefined
): Promise<RelativePath[]> {
return this.fs.listFilesRecursively(root);
}
public async read(path: RelativePath): Promise<Uint8Array> {
return this.fromNativeLineEndings(await this.fs.read(path));
}
/**
* Create a file at the specified path.
*
* If a file with the same name already exists, it is moved before creating the new one.
* Parent directories are created if necessary.
*/
public async create(
path: RelativePath,
newContent: Uint8Array
): Promise<void> {
await this.ensureClearPath(path);
return this.fs.write(path, this.toNativeLineEndings(newContent));
}
public async ensureClearPath(path: RelativePath): Promise<void> {
if (await this.fs.exists(path)) {
const deconflictedPath = await this.deconflictPath(path);
try {
this.logger.debug(
`Didn't expect ${path} to exist, deconflicting by moving it to '${deconflictedPath}'`
);
this.database.move(path, deconflictedPath);
await this.fs.rename(path, deconflictedPath, true);
} finally {
this.fs.unlock(deconflictedPath);
}
} else {
await this.createParentDirectories(path);
}
}
/**
* Update the file at the given path.
*
* Performs a 3-way merge before writing if the file's content differs from `expectedContent`.
* Does not recreate the file if it no longer exists, returning an empty array instead.
*/
public async write(
path: RelativePath,
expectedContent: Uint8Array,
newContent: Uint8Array
): Promise<void> {
if (!(await this.fs.exists(path))) {
this.logger.debug(
`The caller assumed ${path} exists, but it no longer, so we wont recreate it`
);
return;
}
if (
!isFileTypeMergable(
path,
(await this.serverConfig.getConfig()).mergeableFileExtensions
) ||
isBinary(expectedContent) ||
isBinary(newContent)
) {
this.logger.debug(
`The expected content is not mergable, so we won't perform a 3-way merge, just overwrite it`
);
await this.fs.write(
path,
// `newContent` might not be binary so we still have to ensure the line endings are correct
this.toNativeLineEndings(newContent)
);
return;
}
const expectedText = new TextDecoder().decode(expectedContent); // this comes from a previous read which must only have \n line endings
const newText = new TextDecoder().decode(newContent); // this comes from the server which stores text with \n line endings
await this.fs.atomicUpdateText(
path,
({ text, cursors }: TextWithCursors): TextWithCursors => {
this.logger.debug(
`Performing a 3-way merge for ${path} with the expected content`
);
text = text.replaceAll(this.nativeLineEndings, "\n");
const merged = reconcile(
expectedText,
{ text, cursors },
newText
);
const resultText = merged.text.replaceAll(
"\n",
this.nativeLineEndings
);
return {
text: resultText,
cursors: merged.cursors
};
}
public async listFilesRecursively(): Promise<string[]> {
return (await this.fs.listFilesRecursively()).filter(
(path) => !isInternalPath(path)
);
}
public async delete(path: RelativePath): Promise<void> {
if (await this.exists(path)) {
await this.fs.delete(path);
await this.deletingEmptyParentDirectoriesOfDeletedFile(path);
} else {
this.logger.debug(`No need to delete '${path}', it doesn't exist`);
}
public async snapshot(path: string): Promise<StoredSnapshot | undefined> {
const snapshot = await this.fs.read(path);
return snapshot && toStoredSnapshot(snapshot);
}
public async getFileSize(path: RelativePath): Promise<number> {
return this.fs.getFileSize(path);
}
public async exists(path: RelativePath): Promise<boolean> {
return this.fs.exists(path);
}
public async move(
oldPath: RelativePath,
newPath: RelativePath
public async captureChange(
change: LocalChange,
previous: readonly LocalChange[]
): Promise<void> {
if (oldPath === newPath) {
return;
}
const state = structuredClone(this.applying ?? this.database.state);
await this.ensureClearPath(newPath);
if (!state.initialized) return;
this.database.move(oldPath, newPath);
await this.fs.rename(oldPath, newPath);
await this.deletingEmptyParentDirectoriesOfDeletedFile(oldPath);
for (const pending of unappliedChanges(state, previous))
applyLocalChange(state, pending, () => false);
applyLocalChange(state, change, () => false);
}
public reset(): void {
this.fs.reset();
}
private async deletingEmptyParentDirectoriesOfDeletedFile(
path: RelativePath
public async apply(
next: EngineState,
writes: Partial<Record<string, FileWrite>> = {},
validatePlan: () => void = () => undefined
): Promise<void> {
let directory = path;
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
while (true) {
[directory] = FileOperations.getParentDirAndFile(directory);
if (directory.length === 0) {
break;
}
const remainingContent =
await this.fs.listFilesRecursively(directory);
if (remainingContent.length === 0) {
this.logger.debug(
`Folder (${directory}) is now empty, deleting`
);
await this.fs.delete(directory);
} else {
break;
}
}
}
private fromNativeLineEndings(content: Uint8Array): Uint8Array {
if (isBinary(content)) {
return content;
}
const decoder = new TextDecoder("utf-8");
let text = decoder.decode(content);
text = text.replaceAll(this.nativeLineEndings, "\n");
return new TextEncoder().encode(text);
}
private toNativeLineEndings(content: Uint8Array): Uint8Array {
if (isBinary(content)) {
return content;
}
const decoder = new TextDecoder("utf-8");
let text = decoder.decode(content);
text = text.replaceAll("\n", this.nativeLineEndings);
return new TextEncoder().encode(text);
}
private async createParentDirectories(path: string): Promise<void> {
const components = path.split("/");
if (components.length === 1) {
return;
}
for (let i = 1; i < components.length; i++) {
const parentDir = components.slice(0, i).join("/");
if (!(await this.fs.exists(parentDir))) {
await this.fs.createDirectory(parentDir);
}
}
}
/**
* Deconflicts the given path by appending (1), (2), etc. before the file extension until a non-existent path is found.
* The returned path has a lock acquired on it; it must be released by the caller when no longer needed.
*
* @param path The starting path to deconflict
* @returns a non-existent path with a lock acquired on it
*/
private async deconflictPath(path: RelativePath): Promise<RelativePath> {
// eslint-disable-next-line prefer-const
let [directory, fileName] = FileOperations.getParentDirAndFile(path);
if (directory) {
directory += "/";
}
const nameParts = fileName.split(".");
// Handle dotfiles: ".gitignore" should have no extension, ".config.json" should have ".json"
const isDotfile = fileName.startsWith(".") && nameParts[0] === "";
const extension =
nameParts.length > 1 && !(isDotfile && nameParts.length === 2)
? "." + nameParts[nameParts.length - 1]
: "";
let stem = extension ? nameParts.slice(0, -1).join(".") : fileName;
let currentCount = Number.parseInt(
FileOperations.PARENTHESES_REGEX.exec(stem)?.groups?.count ?? "0"
);
stem = stem.replace(FileOperations.PARENTHESES_REGEX, "");
let newName = path;
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
while (true) {
currentCount++;
newName = `${directory}${stem} (${currentCount})${extension}`;
// Avoid multiple deconflictPath calls returning the same path
if (this.fs.tryLock(newName)) {
const newDocument =
this.database.getLatestDocumentByRelativePath(newName);
if (
newDocument?.isDeleted === false || // the document might have been confirmed by the server at a new path but haven't yet moved there locally
(await this.fs.exists(newName, true))
) {
this.fs.unlock(newName);
} else {
return newName;
await this.localChanges.flush();
const changes = this.localChanges.entries().length;
let started = false;
const guard = (): void => {
if (started) return;
validatePlan();
if (this.localChanges.entries().length > changes)
throw new LocalChangesDuringReconciliation();
};
const mutate: Mutation = async (action) => {
guard();
started = true;
await action();
};
guard();
const current = structuredClone(this.database.state);
current.initialized = next.initialized;
this.applying = current;
const paths = new Set<string>();
const observeChanges = async (): Promise<void> => {
await this.localChanges.flush();
guard();
for (const change of unappliedChanges(
next,
this.localChanges.entries()
)) {
for (const state of [current, next]) {
applyLocalChange(
state,
change,
(path) =>
isInternalPath(path) || this.isProtectedFile(path, 0)
);
state.lastAppliedLocalChangeId = change.changeId;
}
}
};
try {
for (const id of new Set([
...Object.keys(current.local),
...Object.keys(next.local)
])) {
await observeChanges();
let from = current.documents[id]?.materialized
? current.local[id]
: undefined;
let to = next.local[id];
const write = writes[id];
if (!write && (from === undefined || from === to)) continue;
if (from !== undefined) {
paths.add(from);
const info = await this.fs.stat(from);
guard();
if (info?.kind !== "file") {
delete next.local[id];
continue;
}
if (this.isProtectedFile(from, info.size)) {
next.local[id] = from;
next.documents[id] = structuredClone(
current.documents[id]
);
(next.excluded ??= {})[id] =
current.fileManifest.entries[id] ?? null;
(next.protectedPaths ??= []).push(from);
continue;
}
}
if (to === undefined) {
if (from !== undefined) {
await mutate(() => this.fs.deleteFile(from!));
delete current.local[id];
await this.paths.removeEmptyParents(from);
}
continue;
}
if (from !== to) {
to = await this.paths.prepareDestination(
next,
current,
id,
to,
mutate
);
next.local[id] = to;
from = current.documents[id]?.materialized
? current.local[id]
: undefined;
if (from !== undefined) {
await mutate(async () => {
current.local[id] = to;
await this.fs.rename(from!, to);
});
await this.paths.removeEmptyParents(from);
}
}
if (write) {
const snapshot =
from === undefined
? undefined
: await this.fs.read(to);
if (from !== undefined && !snapshot) {
delete next.local[id];
continue;
}
const extensions = (await this.serverConfig.getConfig())
.mergeableFileExtensions;
await mutate(async () => {
current.local[id] = to;
current.documents[id] = { materialized: true };
next.documents[id].observedHash =
await this.updateContent(
to,
snapshot,
write,
extensions
);
});
}
paths.add(to);
next.documents[id].materialized = true;
current.local[id] = to;
current.documents[id] = structuredClone(next.documents[id]);
}
await observeChanges();
await this.database.commit(next);
} finally {
this.applying = undefined;
this.filesApplied([...paths]);
}
}
private async updateContent(
path: string,
current: FileSnapshot | undefined,
write: FileWrite,
extensions: string[]
): Promise<string> {
const replacement = fromStoredSnapshot(write.replacement);
const base = write.expected && fromStoredSnapshot(write.expected);
if (
current &&
base &&
isFileTypeMergeable(path, extensions) &&
![current, base, replacement].some((value) =>
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 text = await this.fs.atomicUpdateText(path, (latest) =>
reconcile(parent, latest, remote)
);
return hash(new TextEncoder().encode(text));
}
const result =
current && write.expected
? await mergeContent(
path,
write.expected,
await toStoredSnapshot(current),
write.replacement,
extensions
)
: write.replacement;
await this.fs.write(path, fromStoredSnapshot(result));
return result.hash;
}
}

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

@ -0,0 +1,144 @@
import { v4 as uuid } from "uuid";
import type { FileSystemOperations } from "./filesystem-operations";
import type { EngineState } 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: string, size: number) => boolean
) { }
public async prepareDestination(
next: EngineState,
current: EngineState,
id: string,
path: string,
mutate: (action: () => Promise<void>) => Promise<void>
): Promise<string> {
try {
for (; ;) {
const obstruction = await this.obstruction(path);
if (!obstruction) break;
const children =
obstruction.kind === "file"
? [obstruction.path]
: await this.fs.listFilesRecursively(obstruction.path);
const occupants = [];
for (const child of children) {
const occupant = Object.keys(current.local).find(
(key) =>
current.documents[key]?.materialized &&
current.local[key] === child
);
const info = await this.fs.stat(child);
if (
!info ||
next.protectedPaths?.includes(child) ||
this.isProtectedFile(child, info.size) ||
(obstruction.kind === "directory" &&
(occupant === undefined ||
next.local[occupant] === child))
)
break;
occupants.push({ id: occupant ?? uuid(), path: child });
}
if (occupants.length !== children.length) {
path = await this.conflictPath(next, id, path);
continue;
}
for (const occupant of occupants) {
const displaced = await this.conflictPath(
next,
occupant.id,
occupant.path.split("/").at(-1)!
);
await mutate(async () => {
current.local[occupant.id] = displaced;
await this.fs.rename(occupant.path, displaced);
});
if (!current.documents[occupant.id]) {
current.documents[occupant.id] = { materialized: true };
next.documents[occupant.id] = { materialized: true };
next.local[occupant.id] = displaced;
} else if (next.local[occupant.id] === occupant.path)
next.local[occupant.id] = displaced;
}
if (obstruction.kind === "directory")
await mutate(() => this.fs.deleteDirectory(obstruction.path));
}
await this.ensureParents(path);
} catch (error) {
const code =
typeof error === "object" && error !== null && "code" in error
? error.code
: undefined;
if (
![
"ENAMETOOLONG",
"EINVAL",
"ENOTSUP",
"EOPNOTSUPP",
"EACCES",
"EPERM"
].includes(String(code))
)
throw error;
const extension =
/\.[^.]*$/u.exec(path.split("/").at(-1) ?? "")?.[0] ?? "";
path = await this.conflictPath(
next,
id,
`Recovered ${id}${extension}`
);
await this.ensureParents(path);
}
return path;
}
private async obstruction(
path: string
): Promise<{ path: string; kind: "file" | "directory" } | undefined> {
const parts = path.split("/");
for (let i = 0; i < parts.length; i++) {
const prefix = parts.slice(0, i + 1).join("/");
const info = await this.fs.stat(prefix);
if (info && (info.kind === "file" || i === parts.length - 1))
return { path: prefix, kind: info.kind };
}
}
private async conflictPath(
next: EngineState,
id: string,
path: string
): Promise<string> {
const occupied = { ...next.local };
delete occupied[id];
const files = (await this.fs.listFilesRecursively()).filter(
(path) => !isInternalPath(path)
);
for (const [i, diskPath] of files.entries())
occupied[`disk-${i}`] = diskPath;
return allocatePortablePath(path, id, occupied);
}
private async ensureParents(path: string): Promise<void> {
const parent = path.split("/").slice(0, -1).join("/");
if (parent) await this.fs.createDirectory(parent);
}
public async removeEmptyParents(path: string): Promise<void> {
let parent = path.split("/").slice(0, -1).join("/");
while (parent && !isInternalPath(parent)) {
if ((await this.fs.listFilesRecursively(parent)).length) break;
try {
await this.fs.deleteDirectory(parent);
} catch {
break;
}
parent = parent.split("/").slice(0, -1).join("/");
}
}
}

54
frontend/sync-client/src/file-operations/filesystem-operations.ts
View file

@ -1,18 +1,36 @@
import type { CursorPosition, TextWithCursors } from "reconcile-text";
import type { RelativePath } from "../persistence/database";
import type { TextWithCursors } from "reconcile-text";
export interface FileSnapshot {
content: Uint8Array;
/** Optional editor selections, moved with the file and applied with its bytes. */
cursors?: CursorPosition[];
}
/** Filesystem operations act on the current user-owned namespace.
* Paths are vault-relative. Never follow symlinks (including ancestor symlinks).
* Only independent regular files are supported; reject hardlinks/special files.
* A successful scan is complete: unreadable directories must reject, not vanish.
*/
export interface FileSystemOperations {
// List all files under root that should be synced. If root is undefined, return every file.
listFilesRecursively: (
root: RelativePath | undefined
) => Promise<RelativePath[]>;
listFilesRecursively: (root?: string) => Promise<string[]>;
// Read the content of a file.
read: (path: RelativePath) => Promise<Uint8Array>;
/** Return undefined on missing file */
read: (path: string) => Promise<FileSnapshot | undefined>;
// Create or overwrite a file with the given content.
write: (path: RelativePath, content: Uint8Array) => Promise<void>;
/** Reject symlinks and special files rather than following them and return undefined. */
stat: (
path: string
) => Promise<{ kind: "file" | "directory"; size: number } | undefined>;
/** Includes directories. */
exists: (path: string) => Promise<boolean>;
/** Recursive, idempotent directory creation. */
createDirectory: (path: string) => Promise<void>;
/** Create or overwrite a file, applying optional editor cursor metadata. */
write: (path: string, snapshot: FileSnapshot) => Promise<void>;
// Atomically update the content of a text file.
atomicUpdateText: (
@ -20,18 +38,12 @@ export interface FileSystemOperations {
updater: (current: TextWithCursors) => TextWithCursors
) => Promise<string>;
// Get the size of a file in bytes.
getFileSize: (path: RelativePath) => Promise<number>;
/** Move a file without replacing an existing destination. */
rename: (from: string, to: string) => Promise<void>;
// Check if a file exists.
exists: (path: RelativePath) => Promise<boolean>;
/** Idempotently unlink one regular file. */
deleteFile: (path: string) => Promise<void>;
// Create a directory at the specified path. All parent directories must already exist.
createDirectory: (path: RelativePath) => Promise<void>;
// Delete a file. It is expected that the path points to an existing file.
delete: (path: RelativePath) => Promise<void>;
// Rename a file. It is expected that the oldPath points to an existing file and the newPath does not exist.
rename: (oldPath: RelativePath, newPath: RelativePath) => Promise<void>;
/** Prune directories using rmdir only. Reject files and nonempty directories. */
deleteDirectory: (path: string) => Promise<void>;
}

4
frontend/sync-client/src/index.ts
View file

@ -26,7 +26,7 @@ export type {
FileSystemOperations,
FileSnapshot
} from "./file-operations/filesystem-operations";
export type { PersistenceProvider } from "./persistence/persistence";
export type { MetadataPersistenceProvider as PersistenceProvider } from "./persistence/metadata-persistence-provider";
export type { CursorSpan } from "./services/types/CursorSpan";
export type { ClientCursors } from "./services/types/ClientCursors";
export type { NetworkConnectionStatus } from "./types/network-connection-status";
@ -36,7 +36,7 @@ export type {
} from "./errors/errors";
export type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
export { DocumentSyncStatus } from "./types/document-sync-status";
export { SyncClient } from "./sync-client";
export { SyncClient, type StoredClient } from "./sync-client";
export type { TextWithCursors, CursorPosition } from "reconcile-text";
export const debugging = {

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

@ -1,374 +1,191 @@
import type { Logger } from "../tracing/logger";
import { EMPTY_HASH } from "../utils/hash";
import { CoveredValues } from "../utils/data-structures/min-covered";
import { awaitAll } from "../utils/await-all";
import { removeFromArray } from "../utils/remove-from-array";
import type { CursorPosition } from "reconcile-text";
import type { DocumentUpdateResponse } from "../services/types/DocumentUpdateResponse";
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
import type { FileManifest } from "../services/types/FileManifest";
import type { FileManifestUpdateResponse } from "../services/types/FileManifestUpdateResponse";
import type { PushFileManifest } from "../services/types/PushFileManifest";
import type { PutFileContent } from "../services/types/PutFileContent";
import type { VaultSnapshot } from "../services/types/VaultSnapshot";
import type { FileManifestEntries } from "../types/file-manifest-entries";
export type VaultUpdateId = number;
export type DocumentId = string;
export type RelativePath = string;
export interface StoredSnapshot {
contentBase64: string;
hash: string;
cursors?: CursorPosition[];
}
export interface DocumentMetadata {
parentVersionId: VaultUpdateId;
hash: string;
remoteRelativePath?: RelativePath;
}
export interface StoredDocumentMetadata {
relativePath: RelativePath;
documentId: DocumentId;
parentVersionId: VaultUpdateId;
remoteRelativePath?: RelativePath;
parentVersionId: number;
hash: string;
}
export interface StoredDatabase {
documents: StoredDocumentMetadata[];
lastSeenUpdateId: VaultUpdateId | undefined;
hasInitialSyncCompleted: boolean;
export interface DocumentState {
/** Local clean base retained while rebasing after a server restore. */
recoveryBase?: StoredSnapshot;
base?: DocumentVersionWithoutContent & { hash: string };
observedHash?: string;
materialized: boolean;
/** First contact with existing local content: there is no common ancestor. */
bootstrap?: boolean;
/** Do not resubmit an unchanged payload rejected by the server. */
rejected?: { hash: string; message: string };
}
/**
* Represents a document in the database.
*
* It is mutable and its content should always represent the latest
* state of the document on disk based on the update events we have seen.
*/
export type PendingRequest = { rejection?: string } & (
| {
type: "content";
documentId: string;
request: PutFileContent;
snapshot: StoredSnapshot;
response?: DocumentUpdateResponse;
}
| {
type: "fileManifest";
request: PushFileManifest;
response?: FileManifestUpdateResponse;
}
);
export interface EngineState {
/** Persisted reset intent; clear the transport checkpoint before bootstrap. */
historyRecovery?: boolean;
vaultKey: string;
initialized: boolean;
fileManifest: FileManifest;
local: FileManifestEntries;
documents: Record<string, DocumentState>;
lastSeenUpdateId: number;
remoteHeads: Record<string, DocumentVersionWithoutContent>;
bootstrap?: VaultSnapshot;
pending?: PendingRequest;
/** A rejected retry does not prove an earlier attempt failed to commit. */
unconfirmed?: PendingRequest[];
/** Commits consumption of the saved notification queue with the identity map. */
lastAppliedLocalChangeId?: string;
/** Physical exclusions, including files with no sync identity. */
protectedPaths?: string[];
/** Original server paths while local files are excluded; null means absent. */
excluded?: Record<string, string | null>;
rejectedManifest?: { entries: FileManifestEntries; message: string };
}
export type StoredDatabase = EngineState;
// Read-only view used by cursor tracking and status reporting.
export interface DocumentRecord {
relativePath: RelativePath;
documentId: DocumentId;
metadata: DocumentMetadata | undefined;
isDeleted: boolean;
updates: Promise<unknown>[];
parallelVersion: number;
documentId: string;
relativePath: string;
metadata?: DocumentMetadata;
}
export function emptyState(vaultKey: string): StoredDatabase {
return {
vaultKey,
initialized: false,
fileManifest: { fileManifestId: 0, entries: {} },
local: {},
documents: {},
remoteHeads: {},
lastSeenUpdateId: 0
};
}
function canRebindEmpty(state: Partial<StoredDatabase> | undefined): boolean {
return (
state !== undefined &&
!state.initialized &&
!state.pending &&
!state.bootstrap &&
(state.unconfirmed?.length ?? 0) === 0 &&
Object.keys(state.local ?? {}).length === 0
);
}
export class Database {
private documents: DocumentRecord[];
private lastSeenUpdateIds: CoveredValues;
private hasInitialSyncCompleted: boolean;
public state: StoredDatabase;
private needsReload = false;
public constructor(
private readonly logger: Logger,
initialState: Partial<StoredDatabase> | undefined,
private readonly saveData: (data: StoredDatabase) => Promise<void>
initial: Partial<StoredDatabase> | undefined,
private readonly saveData: (data: StoredDatabase) => Promise<void>,
vaultKey: string,
private readonly loadData: () => Promise<StoredDatabase | undefined>
) {
initialState ??= {};
this.documents =
initialState.documents?.map(
({ relativePath, documentId, ...metadata }) => ({
relativePath,
documentId,
metadata,
isDeleted: false,
updates: [],
parallelVersion: 0
})
) ?? [];
this.ensureConsistency();
this.logger.debug(`Loaded ${this.documents.length} documents`);
const { lastSeenUpdateId } = initialState;
this.logger.debug(`Loaded last seen update id: ${lastSeenUpdateId}`);
this.lastSeenUpdateIds = new CoveredValues(
Math.max(0, lastSeenUpdateId ?? 0) // the first updateId will be 1 which is the first integer after -1
);
this.documents.forEach((doc) => {
this.lastSeenUpdateIds.add(doc.metadata?.parentVersionId);
});
this.hasInitialSyncCompleted =
initialState.hasInitialSyncCompleted ?? false;
this.logger.debug(
`Loaded hasInitialSyncCompleted: ${this.hasInitialSyncCompleted}`
);
}
public get length(): number {
return this.documents.length;
}
public get resolvedDocuments(): DocumentRecord[] {
const paths = new Map<string, DocumentRecord[]>();
this.documents
// eslint-disable-next-line no-restricted-syntax -- Type narrowing, not removing a specific item
.filter(({ metadata }) => metadata !== undefined)
.forEach((record) =>
paths.set(record.relativePath, [
record,
...(paths.get(record.relativePath) ?? [])
])
);
return Array.from(paths.values()).map((records) => {
records.sort(
(a, b) => b.parallelVersion - a.parallelVersion // descending
);
if (
records.length > 1 &&
records.some((current, i) =>
i === 0
? false
: records[i - 1].parallelVersion ===
current.parallelVersion
)
) {
throw new Error(
`Multiple documents with the same parallel version and path at ${records[0].relativePath}`
);
}
return records[0];
});
}
public updateDocumentMetadata(
metadata: {
parentVersionId: VaultUpdateId;
hash: string;
remoteRelativePath: RelativePath;
},
toUpdate: DocumentRecord
): void {
if (!this.documents.includes(toUpdate)) {
throw new Error("Document not found in database");
}
toUpdate.metadata = metadata;
this.saveInTheBackground();
}
public removeDocumentPromise(promise: Promise<unknown>): void {
const entry = this.documents.find(({ updates }) =>
updates.includes(promise)
);
if (entry === undefined) {
// This method should be idempotent and tolerant of
// stragglers calling it after the databse has been reset.
return;
}
removeFromArray(entry.updates, promise);
// No need to save as Promises don't get serialized
}
public removeDocument(find: DocumentRecord): void {
removeFromArray(this.documents, find);
this.saveInTheBackground();
}
public getLatestDocumentByRelativePath(
find: RelativePath
): DocumentRecord | undefined {
const candidates = this.documents.filter(
({ relativePath }) => relativePath === find
);
candidates.sort((a, b) => b.parallelVersion - a.parallelVersion); // descending
return candidates[0];
}
public async getResolvedDocumentByRelativePath(
relativePath: RelativePath,
promise: Promise<unknown>
): Promise<DocumentRecord> {
const entry = this.getLatestDocumentByRelativePath(relativePath);
if (entry === undefined) {
const hasState = initial && Object.keys(initial).length > 0;
if (
hasState &&
initial.vaultKey !== vaultKey &&
!canRebindEmpty(initial)
) {
throw new Error(
`Document not found by relative path: ${relativePath}, ${JSON.stringify(
this.documents,
null,
2
)}`
"Incompatible vault identity. Preserve it for recovery and initialize a separate state store."
);
}
const currentPromises = entry.updates;
entry.updates = [...currentPromises, promise];
await awaitAll(currentPromises);
return entry;
this.state = hasState
? { ...(initial as StoredDatabase), vaultKey }
: emptyState(vaultKey);
}
public createNewPendingDocument(
documentId: DocumentId,
relativePath: RelativePath,
promise: Promise<unknown>
): DocumentRecord {
this.logger.debug(
`Creating new pending document: ${relativePath} (${documentId})`
);
const previousEntry =
this.getLatestDocumentByRelativePath(relativePath);
const entry = {
relativePath,
documentId,
metadata: undefined,
isDeleted: false,
updates: [promise],
parallelVersion:
previousEntry?.parallelVersion === undefined
? 0
: previousEntry.parallelVersion + 1
};
this.documents.push(entry);
this.saveInTheBackground();
return entry;
}
public createNewEmptyDocument(
documentId: DocumentId,
parentVersionId: VaultUpdateId,
relativePath: RelativePath
): DocumentRecord {
const entry = {
relativePath,
documentId,
metadata: {
parentVersionId,
hash: EMPTY_HASH,
remoteRelativePath: relativePath
},
isDeleted: false,
updates: [],
parallelVersion: 0
};
this.documents.push(entry);
this.saveInTheBackground();
return entry;
}
public getDocumentByDocumentId(
find: DocumentId
): DocumentRecord | undefined {
return this.documents.find(({ documentId }) => documentId === find);
}
public move(
oldRelativePath: RelativePath,
newRelativePath: RelativePath
): void {
const oldDocument =
this.getLatestDocumentByRelativePath(oldRelativePath);
if (oldDocument === undefined) {
return;
}
const newDocument =
this.getLatestDocumentByRelativePath(newRelativePath);
if (newDocument?.isDeleted === false) {
/** Bind an empty database to the configured vault. */
public async bindVault(vaultKey: string): Promise<void> {
if (this.state.vaultKey === vaultKey) return;
if (!canRebindEmpty(this.state))
throw new Error(
`Document already exists at new location: ${newRelativePath}`
"Incompatible vault identity. Preserve it for recovery and initialize a separate state store."
);
}
oldDocument.relativePath = newRelativePath;
// We're in a strange state where the target of the move has just got deleted,
// however, its metadata might already have a bunch of updates queued up for
// the document at the new location. We need to keep these updates.
oldDocument.parallelVersion =
newDocument !== undefined ? newDocument.parallelVersion + 1 : 0;
this.saveInTheBackground();
await this.commit({ ...this.state, vaultKey });
}
public delete(relativePath: RelativePath): void {
const candidate = this.getLatestDocumentByRelativePath(relativePath);
if (candidate === undefined) {
public async commit(next: StoredDatabase): Promise<void> {
const snapshot = structuredClone(next);
if (this.needsReload)
throw new Error(
`Document not found by relative path: ${relativePath}`
"Reload saved state before retrying an uncertain save"
);
}
candidate.isDeleted = true;
this.needsReload = true;
await this.saveData(snapshot);
this.state = snapshot;
this.needsReload = false;
}
public getHasInitialSyncCompleted(): boolean {
return this.hasInitialSyncCompleted;
}
public setHasInitialSyncCompleted(value: boolean): void {
this.hasInitialSyncCompleted = value;
this.saveInTheBackground();
}
public getLastSeenUpdateId(): VaultUpdateId {
return this.lastSeenUpdateIds.min;
}
public addSeenUpdateId(value: number): void {
const previousMin = this.lastSeenUpdateIds.min;
this.lastSeenUpdateIds.add(value);
if (previousMin !== this.lastSeenUpdateIds.min) {
this.saveInTheBackground();
}
}
public setLastSeenUpdateId(value: number): void {
this.lastSeenUpdateIds.min = value;
this.saveInTheBackground();
}
public reset(): void {
this.documents = [];
this.lastSeenUpdateIds = new CoveredValues(
0 // the first updateId will be 1 which is the first integer after -1
);
this.hasInitialSyncCompleted = false;
this.saveInTheBackground();
public async recoverPersistence(): Promise<void> {
if (!this.needsReload) return;
const saved = await this.loadData();
if (saved) this.state = saved;
this.needsReload = false;
}
public async save(): Promise<void> {
return this.saveData({
documents: this.resolvedDocuments.map(
({ relativePath, documentId, metadata }) => ({
documentId,
relativePath,
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
...metadata! // `resolvedDocuments` only returns docs with metadata set
})
),
lastSeenUpdateId: this.lastSeenUpdateIds.min,
hasInitialSyncCompleted: this.hasInitialSyncCompleted
});
await this.commit(this.state);
}
private ensureConsistency(): void {
const idToPath = new Map<string, string[]>();
this.resolvedDocuments.forEach(({ relativePath, documentId }) => {
idToPath.set(documentId, [
...(idToPath.get(documentId) ?? []),
relativePath
]);
});
const duplicates = Array.from(idToPath.entries())
.filter(([_, paths]) => paths.length > 1)
.map(([id, paths]) => `${id} (${paths.join(", ")})`);
if (duplicates.length > 0) {
throw new Error(
"Document IDs are not unique, found duplicates: " +
duplicates.join("; ")
);
}
public get length(): number {
return Object.keys(this.state.local).length;
}
private saveInTheBackground(): void {
this.ensureConsistency();
void this.save().catch((error: unknown) => {
this.logger.error(`Error saving data: ${error}`);
});
public getDocumentByDocumentId(id: string): 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 && {
parentVersionId: base.vaultUpdateId,
hash: base.hash
}
};
}
public getLatestDocumentByRelativePath(
path: string
): DocumentRecord | undefined {
const id = Object.keys(this.state.local).find(
(id) => this.state.local[id] === path
);
return id === undefined ? undefined : this.getDocumentByDocumentId(id);
}
}

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

@ -1,9 +1,10 @@
/* eslint-disable @typescript-eslint/no-unsafe-type-assertion */
import { describe, it, beforeEach } from "node:test";
import assert from "node:assert";
import { createServer } from "node:net";
import { WebSocketManager } from "./websocket-manager";
import type { Logger } from "../tracing/logger";
import type { Settings } from "../persistence/settings";
import { Logger } from "../tracing/logger";
import { Settings } from "../persistence/settings";
// eslint-disable-next-line @typescript-eslint/no-require-imports
const WebSocket = require("ws") as typeof globalThis.WebSocket;
@ -135,19 +136,13 @@ describe("WebSocketManager", () => {
.webSocket;
mockWs.simulateMessage({
type: "vaultEvents",
headEventId: 0,
events: []
type: "vaultChanged"
});
mockWs.simulateMessage({
type: "vaultEvents",
headEventId: 0,
events: []
type: "vaultChanged"
});
mockWs.simulateMessage({
type: "vaultEvents",
headEventId: 0,
events: []
type: "vaultChanged"
});
await new Promise((resolve) => setTimeout(resolve, 100));
@ -207,8 +202,7 @@ describe("WebSocketManager", () => {
manager.sendHandshakeMessage({
type: "handshake",
token: "test",
deviceId: "test",
lastSeenVaultUpdateId: null
deviceId: "test"
});
});
@ -316,9 +310,7 @@ describe("WebSocketManager", () => {
const mockWs = (manager as unknown as { webSocket: MockWebSocket })
.webSocket;
mockWs.simulateMessage({
type: "vaultEvents",
headEventId: 0,
events: []
type: "vaultChanged"
});
await new Promise((resolve) => setTimeout(resolve, 10));
@ -337,3 +329,176 @@ describe("WebSocketManager", () => {
await manager.stop();
});
});
it(
"refused WebSocket connections and replacing a connecting socket do not emit unhandled errors",
{ timeout: 5_000 },
async () => {
const server = createServer();
await new Promise<void>((resolve) =>
server.listen(0, "127.0.0.1", resolve)
);
const address = server.address();
assert(address && typeof address !== "string");
await new Promise<void>((resolve, reject) =>
server.close((error) => (error ? reject(error) : resolve()))
);
const logger = new Logger();
const settings = new Settings(
logger,
{
remoteUri: `http://127.0.0.1:${address.port}`,
webSocketRetryIntervalMs: 10
},
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
WebSocket
);
let closes = 0;
const disconnected = Promise.withResolvers<void>();
manager.onWebSocketStatusChanged.add((connected) => {
if (!connected && ++closes >= 2) disconnected.resolve();
});
try {
manager.start();
manager.start();
await disconnected.promise;
assert.equal(manager.isWebSocketConnected, false);
} finally {
await manager.stop();
}
}
);
for (const failedAttempt of [1, 2]) {
it(`recovers from a WebSocket constructor failure on attempt ${failedAttempt}`, async (context) => {
context.mock.timers.enable({ apis: ["setTimeout"] });
let attempts = 0;
class FailingSocket extends MockWebSocket {
static OPEN = 1;
constructor(url: string) {
if (++attempts === failedAttempt)
throw new Error("socket allocation failed");
super(url);
}
}
const logger = new Logger();
const settings = new Settings(
logger,
{ remoteUri: "http://test", webSocketRetryIntervalMs: 1000 },
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
FailingSocket as unknown as typeof WebSocket
);
try {
assert.doesNotThrow(() => manager.start());
context.mock.timers.tick(0);
if (failedAttempt === 2) {
const socket = (
manager as unknown as { webSocket: MockWebSocket }
).webSocket;
socket.close();
}
assert.doesNotThrow(() => context.mock.timers.tick(1000));
context.mock.timers.tick(1000);
assert.equal(attempts, failedAttempt + 1);
assert.equal(manager.isWebSocketConnected, true);
} finally {
await manager.stop();
}
});
}
for (const closeBehavior of ["throws", "stays open"]) {
it(`stop retires a socket even when close ${closeBehavior}`, async (context) => {
context.mock.timers.enable({ apis: ["setTimeout"] });
class BrokenCloseSocket extends MockWebSocket {
static OPEN = 1;
close() {
if (closeBehavior === "throws") throw new Error("close failed");
}
}
const logger = new Logger();
const settings = new Settings(
logger,
{ remoteUri: "http://test" },
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
BrokenCloseSocket as unknown as typeof WebSocket
);
let connected = false,
delivered = 0;
manager.onWebSocketStatusChanged.add((status) => {
connected = status;
});
manager.onRemoteVaultUpdateReceived.add(async () => {
delivered++;
});
manager.start();
context.mock.timers.tick(0);
const socket = (manager as unknown as { webSocket: MockWebSocket })
.webSocket;
const stopping = manager.stop();
context.mock.timers.tick(10_000);
await stopping;
assert.equal(manager.isWebSocketConnected, false);
assert.equal(connected, false);
socket.simulateMessage({
type: "vaultChanged"
});
await manager.waitUntilFinished();
assert.equal(delivered, 0);
});
}
it("a handshake send failure retires the connection without escaping the open callback", async (context) => {
context.mock.timers.enable({ apis: ["setTimeout"] });
class BrokenSendSocket extends MockWebSocket {
static OPEN = 1;
send() {
throw new Error("transport closed during open");
}
}
const logger = new Logger();
const settings = new Settings(
logger,
{ remoteUri: "http://test" },
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
BrokenSendSocket as unknown as typeof WebSocket
);
manager.onWebSocketStatusChanged.add((connected) => {
if (connected)
manager.sendHandshakeMessage({
type: "handshake",
token: "test",
deviceId: "test"
});
});
const statuses: boolean[] = [];
manager.onWebSocketStatusChanged.add((status) => statuses.push(status));
try {
manager.start();
assert.doesNotThrow(() => context.mock.timers.tick(0));
assert.equal(manager.isWebSocketConnected, false);
assert.equal(statuses.at(-1), false);
} finally {
await manager.stop();
}
});

126
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 { createPromise } from "../utils/create-promise";
import type { EventBatch } from "./types/EventBatch";
import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts";
import { removeFromArray } from "../utils/remove-from-array";
import { EventListeners } from "../utils/data-structures/event-listeners";
@ -17,7 +16,7 @@ export class WebSocketManager {
>();
public readonly onRemoteVaultUpdateReceived = new EventListeners<
(update: EventBatch) => Promise<void>
() => Promise<void>
>();
public readonly onRemoteCursorsUpdateReceived = new EventListeners<
@ -26,6 +25,7 @@ export class WebSocketManager {
private isStopped = true;
private resolveDisconnectingPromise: null | (() => unknown) = null;
private receiveTimeoutId: ReturnType<typeof setTimeout> | undefined;
private reconnectTimeoutId: ReturnType<typeof setTimeout> | undefined;
private readonly outstandingPromises: Promise<unknown>[] = [];
@ -56,12 +56,14 @@ export class WebSocketManager {
public get isWebSocketConnected(): boolean {
return (
this.webSocket?.readyState ===
this.webSocketFactoryImplementation.OPEN
this.webSocket !== undefined &&
this.webSocket.readyState ===
this.webSocketFactoryImplementation.OPEN
);
}
public start(): void {
if (!this.isStopped) return;
this.isStopped = false;
this.initializeWebSocket();
}
@ -71,13 +73,20 @@ export class WebSocketManager {
this.resolveDisconnectingPromise = resolve;
this.isStopped = true;
clearTimeout(this.receiveTimeoutId);
if (this.reconnectTimeoutId !== undefined) {
clearTimeout(this.reconnectTimeoutId);
this.reconnectTimeoutId = undefined;
}
this.webSocket?.close(1000, "WebSocketManager has been stopped");
const socket = this.webSocket;
try {
socket?.close(1000, "WebSocketManager has been stopped");
} catch (error) {
this.logger.warn(`Failed to close WebSocket: ${String(error)}`);
if (socket) this.reconnect(socket);
}
// eslint-disable-next-line @typescript-eslint/init-declarations
let timeoutId: ReturnType<typeof setTimeout> | undefined;
@ -107,6 +116,9 @@ export class WebSocketManager {
if (timeoutId !== undefined) {
clearTimeout(timeoutId);
}
// A throwing, silent or timed-out close must still fence callbacks
// before a stopped client can release its state or restart.
if (socket) this.reconnect(socket);
}
await this.waitUntilFinished();
@ -132,6 +144,7 @@ export class WebSocketManager {
this.logger.error(
`Failed to send handshake message: ${String(error)}`
);
this.reconnect(webSocket);
throw error;
}
}
@ -162,7 +175,66 @@ export class WebSocketManager {
}
}
private expectResponse(socket: WebSocket): void {
clearTimeout(this.receiveTimeoutId);
// Cursor heartbeats are acknowledged every 15 seconds, including when
// there are no open editors. An OPEN TCP socket alone proves nothing.
this.receiveTimeoutId = setTimeout(() => {
if (this.webSocket !== socket || this.isStopped) return;
this.logger.warn("WebSocket receive deadline expired");
this.reconnect(socket);
}, 45_000);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Browser timers have no unref method.
this.receiveTimeoutId.unref?.();
}
private reconnect(socket: WebSocket): void {
if (this.webSocket !== socket) return;
clearTimeout(this.receiveTimeoutId);
socket.onopen = null;
socket.onclose = null;
socket.onmessage = null;
socket.onerror = (): void => {
/* Retired transports may still emit errors. */
};
this.webSocket = undefined;
this.onWebSocketStatusChanged.trigger(false);
try {
socket.close();
} catch {
/* The transport may already be gone. */
}
if (this.isStopped) {
this.resolveDisconnectingPromise?.();
this.resolveDisconnectingPromise = null;
} else {
this.scheduleReconnect();
}
}
private scheduleReconnect(): void {
if (this.isStopped) return;
clearTimeout(this.reconnectTimeoutId);
this.reconnectTimeoutId = setTimeout(() => {
this.reconnectTimeoutId = undefined;
this.initializeWebSocket();
}, this.settings.getSettings().webSocketRetryIntervalMs);
}
private initializeWebSocket(): void {
if (this.isStopped) return;
try {
this.openWebSocket();
} catch (error) {
this.logger.warn(`Cannot initialize WebSocket: ${String(error)}`);
this.webSocket = undefined;
clearTimeout(this.receiveTimeoutId);
this.onWebSocketStatusChanged.trigger(false);
this.scheduleReconnect();
}
}
private openWebSocket(): void {
// Clean up old WebSocket handlers to prevent race conditions
if (this.webSocket) {
try {
@ -170,7 +242,9 @@ export class WebSocketManager {
this.webSocket.onopen = null;
this.webSocket.onclose = null;
this.webSocket.onmessage = null;
this.webSocket.onerror = null;
this.webSocket.onerror = (): void => {
/* Retired transports may still emit errors. */
};
this.webSocket.close();
} catch (e) {
this.logger.error(
@ -188,7 +262,14 @@ export class WebSocketManager {
this.logger.info(`Connecting to WebSocket at ${wsUri.toString()}`);
this.webSocket = new this.webSocketFactoryImplementation(wsUri);
const socket = new this.webSocketFactoryImplementation(wsUri);
this.webSocket = socket;
this.expectResponse(socket);
this.webSocket.onerror = (): void => {
this.logger.warn("WebSocket transport error");
this.reconnect(socket);
};
this.webSocket.onopen = (): void => {
// Check if we've been stopped while connecting
@ -200,7 +281,18 @@ export class WebSocketManager {
return;
}
this.logger.info("WebSocket connection opened");
this.onWebSocketStatusChanged.trigger(true);
try {
this.onWebSocketStatusChanged.trigger(true);
// A handshake failure can retire this socket from inside an
// observer. Finish delivery with its actual connection state.
if (this.webSocket !== socket)
this.onWebSocketStatusChanged.trigger(false);
} catch (error) {
this.logger.warn(
`WebSocket handshake failed: ${String(error)}`
);
this.reconnect(socket);
}
};
this.webSocket.onmessage = (event): void => {
@ -210,6 +302,8 @@ export class WebSocketManager {
event.data
) as WebSocketServerMessage;
if (["vaultChanged", "cursorPositions"].includes(message.type))
this.expectResponse(socket);
// Track the message handling promise
const messageHandlingPromise = this.handleWebSocketMessage(
message
@ -238,25 +332,15 @@ export class WebSocketManager {
this.logger.warn(
`WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})`
);
this.onWebSocketStatusChanged.trigger(false);
if (this.isStopped) {
this.resolveDisconnectingPromise?.();
this.resolveDisconnectingPromise = null;
} else {
this.reconnectTimeoutId = setTimeout(() => {
this.reconnectTimeoutId = undefined;
this.initializeWebSocket();
}, this.settings.getSettings().webSocketRetryIntervalMs);
}
this.reconnect(socket);
};
}
private async handleWebSocketMessage(
message: WebSocketServerMessage
): Promise<void> {
if (message.type === "vaultEvents") {
await this.onRemoteVaultUpdateReceived.triggerAsync(message);
if (message.type === "vaultChanged") {
await this.onRemoteVaultUpdateReceived.triggerAsync();
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
} else if (message.type === "cursorPositions") {

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

@ -1,62 +1,235 @@
import type { PersistenceProvider } from "./persistence/persistence";
import type { HistoryEntry, HistoryStats } from "./tracing/sync-history";
import { SyncHistory } from "./tracing/sync-history";
import { Logger, LogLevel, LogLine } from "./tracing/logger";
import type { RelativePath, StoredDatabase } from "./persistence/database";
import { Database } from "./persistence/database";
import * as Sentry from "@sentry/browser";
import type { SyncSettings } from "./persistence/settings";
import { DEFAULT_SETTINGS, Settings } from "./persistence/settings";
import { SyncService } from "./services/sync-service";
import { Syncer } from "./sync-operations/syncer";
import { setUpTelemetry } from "./utils/set-up-telemetry";
import type { MetadataPersistenceProvider } from "./persistence/metadata-persistence-provider";
import { Database, type StoredDatabase } from "./persistence/database";
import {
Settings,
DEFAULT_SETTINGS,
type SyncSettings
} from "./persistence/settings";
import type { FileSystemOperations } from "./file-operations/filesystem-operations";
import { FileOperations } from "./file-operations/file-operations";
import { FetchController } from "./services/fetch-controller";
import { UnrestrictedSyncer } from "./sync-operations/unrestricted-syncer";
import { rateLimit } from "./utils/rate-limit";
import type { NetworkConnectionStatus } from "./types/network-connection-status";
import { DocumentSyncStatus } from "./types/document-sync-status";
import { WebSocketManager } from "./services/websocket-manager";
import { createClientId } from "./utils/create-client-id";
import { CursorTracker } from "./sync-operations/cursor-tracker";
import type { CursorSpan } from "./services/types/CursorSpan";
import type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
import { FileChangeNotifier } from "./sync-operations/file-change-notifier";
import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache";
import { setUpTelemetry } from "./utils/set-up-telemetry";
import { DIFF_CACHE_SIZE_MB } from "./consts";
import { Logger, LogLevel } from "./tracing/logger";
import { SyncHistory } from "./tracing/sync-history";
import { SyncService } from "./services/sync-service";
import { ServerConfig } from "./services/server-config";
import type { EventListeners } from "./utils/data-structures/event-listeners";
import { WebSocketManager } from "./services/websocket-manager";
import { Syncer } from "./sync-operations/syncer";
import { CursorTracker } from "./sync-operations/cursor-tracker";
import { FileChangeNotifier } from "./sync-operations/file-change-notifier";
import { createClientId } from "./utils/create-client-id";
import { DocumentSyncStatus } from "./types/document-sync-status";
import type { CursorSpan } from "./services/types/CursorSpan";
import { isInternalPath } from "./utils/portable-path";
import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache";
import { Lock } from "./utils/data-structures/locks";
import {
unappliedChanges,
type LocalChange
} from "./sync-operations/local-changes";
export type StoredClient = Partial<{
settings: Partial<SyncSettings>;
database: Partial<StoredDatabase>;
localChanges: (
| LocalChange
| { type: "update"; path: string; changeId?: string }
)[];
localChangesVaultKey: string;
historyCheckpoint: { vaultKey: string; checkpoint?: string };
}>;
const vaultKey = (settings: SyncSettings): string =>
JSON.stringify([
settings.remoteUri.replace(/\/$/u, ""),
settings.vaultName
]);
export class SyncClient {
private hasStartedOfflineSync = false;
private hasFinishedOfflineSync = false;
private hasStarted = false;
private hasBeenDestroyed = false;
private readonly lifecycle = new Lock();
private destroying?: Promise<void>;
private started = false;
private destroyed = false;
private unloadTelemetry?: () => void;
private isDestroying = false;
private readonly eventUnsubscribers: (() => void)[] = [];
private readonly unsubscribeLog: () => void;
private constructor(
public readonly logger: Logger,
private readonly history: SyncHistory,
private readonly settings: Settings,
private readonly database: Database,
private readonly syncer: Syncer,
private readonly webSocketManager: WebSocketManager,
public readonly logger: Logger,
private readonly fetchController: FetchController,
private readonly service: SyncService,
private readonly serverConfig: ServerConfig,
private readonly cursorTracker: CursorTracker,
private readonly fileChangeNotifier: FileChangeNotifier,
private readonly contentCache: FixedSizeDocumentCache,
private readonly fileOperations: FileOperations,
private readonly serverConfig: ServerConfig,
private readonly persistence: PersistenceProvider<
Partial<{
settings: Partial<SyncSettings>;
database: Partial<StoredDatabase>;
}>
>
) {}
private readonly persistence: MetadataPersistenceProvider<StoredClient>
) {
syncer.onServerHistoryChanged.add(() => {
cursorTracker.reset();
});
syncer.onReadyForEvents.add(() => {
if (
this.started &&
!this.destroyed &&
settings.getSettings().isSyncEnabled
)
webSocketManager.start();
});
this.unsubscribeLog = logger.onLogEmitted.add((log) => {
if (log.level === LogLevel.ERROR && Sentry.isInitialized())
Sentry.captureMessage(log.message);
});
}
public static async create({
fs,
persistence,
fetch,
webSocket
}: {
fs: FileSystemOperations;
persistence: MetadataPersistenceProvider<StoredClient>;
fetch?: typeof globalThis.fetch;
webSocket?: typeof globalThis.WebSocket;
}): Promise<SyncClient> {
const logger = new Logger();
let stored = (await persistence.load()) ?? {};
let saving: Promise<void> = Promise.resolve();
let reloadBeforeSave = false;
// Settings and engine state share a store. Serialize complete metadata
// replacements so one cannot overwrite a newer save from the other.
const save = async (update: StoredClient): Promise<void> => {
const operation = saving.then(async () => {
if (reloadBeforeSave) {
stored = (await persistence.load()) ?? {};
reloadBeforeSave = false;
}
const next = { ...stored, ...structuredClone(update) };
try {
await persistence.save(next);
stored = next;
} catch (error) {
// A failed save may have committed. Do not let a queued
// settings/notification save roll back that engine state.
reloadBeforeSave = true;
throw error;
}
});
saving = operation.catch(() => { });
return operation;
};
const settings = new Settings(logger, stored.settings, async (data) =>
save({ settings: data })
);
const database = new Database(
stored.database,
async (data) => save({ database: data }),
vaultKey(settings.getSettings()),
async () => {
await saving;
stored = (await persistence.load()) ?? {};
return stored.database as StoredDatabase | undefined;
}
);
// Older stores also queued content updates. Trim the acknowledged prefix
// before discarding them, since its marker may belong to an update.
const localChanges = unappliedChanges(
database.state,
stored.localChanges ?? []
// eslint-disable-next-line no-restricted-syntax -- Migration needs a narrowed copy, not removal of a specific object.
).filter((change) => change.type !== "update");
if (
localChanges.length &&
stored.localChangesVaultKey !== vaultKey(settings.getSettings())
) {
throw new Error(
"Offline notifications belong to another vault; use a separate state store"
);
}
const deviceId = createClientId();
const service = new SyncService(deviceId, settings, fetch, {
get: () =>
stored.historyCheckpoint?.vaultKey ===
vaultKey(settings.getSettings())
? stored.historyCheckpoint.checkpoint
: undefined,
save: async (checkpoint) =>
save({
historyCheckpoint: {
vaultKey: vaultKey(settings.getSettings()),
checkpoint
}
})
});
const serverConfig = new ServerConfig(service);
const notifier = new FileChangeNotifier();
let syncer: Syncer;
const files = new FileOperations(
fs,
database,
serverConfig,
(paths) => {
for (const path of paths) notifier.notifyOfFileChange(path);
},
{
entries: () => localChanges,
flush: async () => syncer?.flushLocalChanges()
},
(path, size) =>
settings.isIgnored(path) || settings.isOversized(size)
);
const websocket = new WebSocketManager(
deviceId,
logger,
settings,
webSocket
);
const history = new SyncHistory(logger);
const contentCache = new FixedSizeDocumentCache(
settings.getSettings().diffCacheSizeMB * 1024 * 1024
);
syncer = new Syncer(
deviceId,
logger,
database,
settings,
service,
websocket,
files,
serverConfig,
history,
contentCache,
{
entries: localChanges,
save: async (entries) =>
save({
localChanges: entries,
localChangesVaultKey: vaultKey(settings.getSettings())
})
}
);
const cursors = new CursorTracker(database, websocket, files, notifier);
return new SyncClient(
logger,
history,
settings,
database,
syncer,
websocket,
service,
serverConfig,
cursors,
notifier,
contentCache,
persistence
);
}
private check(): void {
if (this.destroyed) throw new Error("SyncClient has been destroyed");
}
public get documentCount(): number {
return this.database.length;
@ -66,478 +239,227 @@ export class SyncClient {
return this.webSocketManager.isWebSocketConnected;
}
public get onSyncHistoryUpdated(): EventListeners<
(stats: HistoryStats) => unknown
> {
this.checkIfDestroyed("onSyncHistoryUpdated getter");
public get onSyncHistoryUpdated() {
return this.history.onHistoryUpdated;
}
public get onSettingsChanged(): EventListeners<
(newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
> {
this.checkIfDestroyed("onSettingsChanged getter");
public get onSettingsChanged() {
return this.settings.onSettingsChanged;
}
public get onRemainingOperationsCountChanged(): EventListeners<
(remainingOperationsCount: number) => unknown
> {
this.checkIfDestroyed("onRemainingOperationsCountChanged getter");
public get onRemainingOperationsCountChanged() {
return this.syncer.onRemainingOperationsCountChanged;
}
public get onWebSocketStatusChanged(): EventListeners<
(isConnected: boolean) => unknown
> {
this.checkIfDestroyed("onWebSocketStatusChanged getter");
public get onWebSocketStatusChanged() {
return this.webSocketManager.onWebSocketStatusChanged;
}
public get onRemoteCursorsUpdated(): EventListeners<
(cursors: MaybeOutdatedClientCursors[]) => unknown
> {
this.checkIfDestroyed("onRemoteCursorsUpdated getter");
public get onRemoteCursorsUpdated() {
return this.cursorTracker.onRemoteCursorsUpdated;
}
public static async create({
fs,
persistence,
fetch,
webSocket,
nativeLineEndings = "\n"
}: {
fs: FileSystemOperations;
persistence: PersistenceProvider<
Partial<{
settings: Partial<SyncSettings>;
database: Partial<StoredDatabase>;
}>
>;
fetch?: typeof globalThis.fetch;
webSocket?: typeof globalThis.WebSocket;
nativeLineEndings?: string;
}): Promise<SyncClient> {
const logger = new Logger();
const deviceId = createClientId();
logger.info(`Creating SyncClient with client id ${deviceId}`);
const history = new SyncHistory(logger);
let state = (await persistence.load()) ?? {
settings: undefined,
database: undefined
};
const settings = new Settings(
logger,
state.settings,
async (data): Promise<void> => {
state = { ...state, settings: data };
// we're not rate-limiting settings saves as (1) we need to initialise the settings to know the rate limit
// and (2) settings changes are infrequent enough that rate-limiting is not necessary
await persistence.save(state);
}
);
const rateLimitedSave = rateLimit(
persistence.save,
() => settings.getSettings().minimumSaveIntervalMs
);
const database = new Database(
logger,
state.database,
async (data): Promise<void> => {
state = { ...state, database: data };
await rateLimitedSave(state);
}
);
const fetchController = new FetchController(
settings.getSettings().isSyncEnabled,
logger
);
const syncService = new SyncService(
deviceId,
fetchController,
settings,
logger,
fetch
);
const serverConfig = new ServerConfig(syncService);
const fileOperations = new FileOperations(
logger,
database,
fs,
serverConfig,
nativeLineEndings
);
const contentCache = new FixedSizeDocumentCache(
1024 * 1024 * DIFF_CACHE_SIZE_MB
);
const unrestrictedSyncer = new UnrestrictedSyncer(
logger,
database,
settings,
syncService,
fileOperations,
history,
contentCache,
serverConfig
);
const webSocketManager = new WebSocketManager(
deviceId,
logger,
settings,
webSocket
);
const syncer = new Syncer(
deviceId,
logger,
database,
settings,
syncService,
webSocketManager,
fileOperations,
unrestrictedSyncer
);
const fileChangeNotifier = new FileChangeNotifier();
const cursorTracker = new CursorTracker(
database,
webSocketManager,
fileOperations,
fileChangeNotifier
);
const client = new SyncClient(
history,
settings,
database,
syncer,
webSocketManager,
logger,
fetchController,
cursorTracker,
fileChangeNotifier,
contentCache,
fileOperations,
serverConfig,
persistence
);
logger.info("SyncClient created successfully");
return client;
}
public async start(): Promise<void> {
this.checkIfDestroyed("start");
if (this.hasStarted) {
throw new Error("SyncClient has already been started");
}
this.hasStarted = true;
if (
!this.unloadTelemetry &&
this.settings.getSettings().enableTelemetry
) {
this.unloadTelemetry = setUpTelemetry();
}
this.eventUnsubscribers.push(
this.settings.onSettingsChanged.add((newSettings, oldSettings) => {
if (oldSettings.isSyncEnabled != newSettings.isSyncEnabled) {
this.fetchController.canFetch = newSettings.isSyncEnabled;
}
})
);
this.eventUnsubscribers.push(
this.logger.onLogEmitted.add((log): void => {
if (log.level === LogLevel.ERROR && Sentry.isInitialized()) {
Sentry.captureMessage(log.message);
}
})
);
this.eventUnsubscribers.push(
this.settings.onSettingsChanged.add(
this.onSettingsChange.bind(this)
)
);
if (this.settings.getSettings().isSyncEnabled) {
this.logger.info("Starting SyncClient");
await this.startSyncing();
this.logger.info("SyncClient has successfully started");
}
}
/**
* Reload settings from disk overriding current in-memory settings.
* Missing values will be filled in from DEFAULT_SETTINGS rather than
* retaining current in-memory settings.
*/
public async reloadSettings(): Promise<void> {
this.checkIfDestroyed("reloadSettings");
const state = (await this.persistence.load()) ?? {
settings: undefined
};
const settings = {
...DEFAULT_SETTINGS,
...(state.settings ?? {})
};
await this.setSettings(settings);
}
public async checkConnection(): Promise<NetworkConnectionStatus> {
this.checkIfDestroyed("checkConnection");
const server = await this.serverConfig.checkConnection(true);
return {
isSuccessful: server.isSuccessful,
serverMessage: server.message,
isWebSocketConnected: this.webSocketManager.isWebSocketConnected
};
}
public getHistoryEntries(): readonly HistoryEntry[] {
public getHistoryEntries() {
return this.history.entries;
}
/**
* Wait for the in-flight operations to finish, reset all tracking,
* and the local database but retain the settings.
* The SyncClient can be used again after calling this method.
*/
public async reset(): Promise<void> {
this.checkIfDestroyed("reset");
this.logger.info(
"Stopping SyncClient to apply changed connection settings"
);
await this.pause();
// clear all local state
this.logger.info("Resetting SyncClient's local state");
this.database.reset();
await this.database.save(); // ensure the new database reads as empty
this.resetInMemoryState();
this.hasStartedOfflineSync = false;
this.hasFinishedOfflineSync = false;
this.serverConfig.reset();
await this.startSyncing();
}
public getSettings(): SyncSettings {
return this.settings.getSettings();
}
public async setSetting<T extends keyof SyncSettings>(
key: T,
value: SyncSettings[T]
): Promise<void> {
this.checkIfDestroyed("setSetting");
await this.settings.setSetting(key, value);
}
public async setSettings(value: Partial<SyncSettings>): Promise<void> {
this.checkIfDestroyed("setSettings");
await this.settings.setSettings(value);
}
public async syncLocallyCreatedFile(
relativePath: RelativePath
): Promise<void> {
this.checkIfDestroyed("syncLocallyCreatedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyCreatedFile(relativePath);
}
public async syncLocallyDeletedFile(
relativePath: RelativePath
): Promise<void> {
this.checkIfDestroyed("syncLocallyDeletedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyDeletedFile(relativePath);
}
public async syncLocallyUpdatedFile({
oldPath,
relativePath
}: {
oldPath?: RelativePath;
relativePath: RelativePath;
}): Promise<void> {
this.checkIfDestroyed("syncLocallyUpdatedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyUpdatedFile({
oldPath,
relativePath
});
}
public getDocumentSyncingStatus(
relativePath: RelativePath
): DocumentSyncStatus {
this.checkIfDestroyed("getDocumentSyncingStatus");
if (!this.settings.getSettings().isSyncEnabled) {
return DocumentSyncStatus.SYNCING_IS_DISABLED;
}
if (!this.syncer.isFirstSyncComplete || !this.hasFinishedOfflineSync) {
return DocumentSyncStatus.SYNCING;
}
const document =
this.database.getLatestDocumentByRelativePath(relativePath);
if (document === undefined) {
return DocumentSyncStatus.SYNCING;
}
return document.updates.length > 0
? DocumentSyncStatus.SYNCING
: DocumentSyncStatus.UP_TO_DATE;
}
public async updateLocalCursors(
documentToCursors: Record<RelativePath, CursorSpan[]>
): Promise<void> {
this.checkIfDestroyed("updateLocalCursors");
await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
}
public async waitUntilFinished(): Promise<void> {
this.checkIfDestroyed("waitUntilIdle");
await this.syncer.waitUntilFinished();
await this.webSocketManager.waitUntilFinished();
await this.database.save(); // flush all changes to disk
}
/**
* Completely destroy the SyncClient, cancelling all in-progress operations.
* After calling this method, the SyncClient cannot be used again.
*/
public async destroy(): Promise<void> {
this.checkIfDestroyed("destroy");
// Prevent concurrent destroy calls
if (this.isDestroying) {
this.logger.warn(
"destroy() called while already destroying, ignoring"
public async start(): Promise<void> {
await this.lifecycle.withLock(async () => {
this.check();
if (this.started)
throw new Error("SyncClient has already been started");
await this.syncer.reloadLocalState();
await this.database.bindVault(
vaultKey(this.settings.getSettings())
);
return;
}
this.isDestroying = true;
// cancel everything that's in progress
await this.pause();
this.hasBeenDestroyed = true;
this.resetInMemoryState();
// Clean up event listeners to prevent memory leaks
this.eventUnsubscribers.forEach((unsubscribe) => {
unsubscribe();
this.started = true;
this.configureTelemetry();
if (this.settings.getSettings().isSyncEnabled) await this.resume();
});
this.eventUnsubscribers.length = 0;
this.logger.info("SyncClient has been successfully disposed");
this.unloadTelemetry?.();
}
private async startSyncing(): Promise<void> {
this.checkIfDestroyed("startSyncing");
this.fetchController.finishReset();
await this.serverConfig.initialize();
this.webSocketManager.start();
if (!this.hasStartedOfflineSync) {
this.hasStartedOfflineSync = true;
await this.syncer.scheduleSyncForOfflineChanges();
}
this.hasFinishedOfflineSync = true;
private async resume(): Promise<void> {
if (this.destroyed) return;
if (!this.settings.getSettings().isSyncEnabled) return;
await this.database.bindVault(vaultKey(this.settings.getSettings()));
this.service.resume();
this.syncer.start();
await this.syncer.waitUntilFinished();
}
private async pause(): Promise<void> {
this.fetchController.startReset();
const stopped = this.syncer.stop();
this.service.pause();
await this.webSocketManager.stop();
await this.waitUntilFinished();
await stopped;
}
private resetInMemoryState(): void {
this.history.reset();
this.contentCache.reset();
// don't reset the logger
this.cursorTracker.reset();
this.syncer.reset();
this.fileOperations.reset();
/** Restart transports; never discard pending requests. */
public async reset(): Promise<void> {
await this.lifecycle.withLock(async () => {
this.check();
await this.pause();
this.serverConfig.reset();
this.cursorTracker.reset();
await this.syncer.retryRejectedRequests();
if (this.started && this.settings.getSettings().isSyncEnabled)
await this.resume();
});
}
private async onSettingsChange(
newSettings: SyncSettings,
oldSettings: SyncSettings
): Promise<void> {
this.checkIfDestroyed("onSettingsChange");
public async setSettings(change: Partial<SyncSettings>): Promise<void> {
const update = structuredClone(change);
const notification = await this.lifecycle.withLock(async () => {
this.check();
const old = this.settings.getSettings();
const next = { ...old, ...update };
if (
vaultKey(next) !== vaultKey(old) &&
(this.started ||
this.database.state.initialized ||
this.database.state.bootstrap ||
this.database.state.pending ||
this.syncer.hasLocalChanges ||
Object.keys(this.database.state.local).length)
)
throw new Error(
"Changing vaults requires a separate state store and client; existing sync state must not be reset"
);
if (
newSettings.vaultName !== oldSettings.vaultName ||
newSettings.remoteUri !== oldSettings.remoteUri
) {
await this.reset();
}
if (newSettings.isSyncEnabled !== oldSettings.isSyncEnabled) {
if (newSettings.isSyncEnabled) {
await this.startSyncing();
} else {
await this.pause();
try {
if (this.started) await this.pause();
await this.database.recoverPersistence();
await this.settings.setSettings(update, false);
await this.syncer.retryRejectedRequests();
this.contentCache.resize(next.diffCacheSizeMB * 1024 * 1024);
await this.database.bindVault(vaultKey(next));
} catch (error) {
this.serverConfig.reset();
this.contentCache.resize(
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)}`
);
});
throw error;
}
}
this.serverConfig.reset();
this.configureTelemetry();
if (this.started && next.isSyncEnabled) await this.resume();
return [this.settings.getSettings(), old] as const;
});
await this.settings.onSettingsChanged.triggerAsync(...notification);
}
if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
this.contentCache.resize(newSettings.diffCacheSizeMB * 1024 * 1024);
}
public async setSetting<K extends keyof SyncSettings>(
key: K,
value: SyncSettings[K]
): Promise<void> {
await this.setSettings({ [key]: value });
}
if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
if (newSettings.enableTelemetry) {
this.unloadTelemetry = setUpTelemetry();
} else {
public async reloadSettings(): Promise<void> {
const stored = await this.persistence.load();
await this.setSettings({ ...DEFAULT_SETTINGS, ...stored?.settings });
}
public async checkConnection() {
const result = await this.serverConfig.checkConnection();
return {
isSuccessful: result.isSuccessful,
serverMessage: result.message,
isWebSocketConnected: this.isWebSocketConnected
};
}
/** Notifications enqueue work. Use waitUntilFinished() to await convergence. */
public async syncLocallyCreatedFile(path: string): Promise<void> {
this.check();
if (isInternalPath(path)) return;
this.fileChangeNotifier.notifyOfFileChange(path);
await this.syncer.syncLocallyCreatedFile(path);
}
public async syncLocallyDeletedFile(path: string): Promise<void> {
this.check();
if (isInternalPath(path)) return;
this.fileChangeNotifier.notifyOfFileChange(path);
await this.syncer.syncLocallyDeletedFile(path);
}
public async syncLocallyUpdatedFile(change: {
oldPath?: string;
relativePath: string;
}): Promise<void> {
this.check();
if (
isInternalPath(change.relativePath) ||
(change.oldPath && isInternalPath(change.oldPath))
)
return;
this.fileChangeNotifier.notifyOfFileChange(change.relativePath);
await this.syncer.syncLocallyUpdatedFile(change);
}
public async updateLocalCursors(
cursors: Record<string, CursorSpan[]>
): Promise<void> {
this.check();
await this.cursorTracker.sendLocalCursorsToServer(cursors);
}
public getDocumentSyncingStatus(path: string): DocumentSyncStatus {
this.check();
if (!this.settings.getSettings().isSyncEnabled)
return DocumentSyncStatus.SYNCING_IS_DISABLED;
return this.syncer.isDocumentUpToDate(path)
? DocumentSyncStatus.UP_TO_DATE
: DocumentSyncStatus.SYNCING;
}
public async waitUntilFinished(): Promise<void> {
this.check();
await this.syncer.waitUntilFinished();
}
private configureTelemetry(): void {
if (
this.settings.getSettings().enableTelemetry &&
typeof window !== "undefined"
) {
this.unloadTelemetry ??= setUpTelemetry();
} else {
this.unloadTelemetry?.();
this.unloadTelemetry = undefined;
}
}
public async destroy(): Promise<void> {
if (this.destroying) return this.destroying;
this.destroyed = true;
// Interrupt a start/reset awaiting network progress before waiting for
// its lifecycle operation to finish.
this.service.pause();
const stopped = this.syncer.stop();
this.destroying = this.lifecycle.withLock(async () => {
try {
await this.webSocketManager.stop();
await stopped;
await this.syncer.flushLocalChanges();
} finally {
this.cursorTracker.reset();
this.unsubscribeLog();
this.unloadTelemetry?.();
}
}
}
private checkIfDestroyed(origin: string): void {
if (this.hasBeenDestroyed) {
throw new Error(
`SyncClient has been destroyed and can no longer be used; called from ${origin}`
);
}
});
return this.destroying;
}
}

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

@ -4,7 +4,7 @@ import type { FileSnapshot } from "../file-operations/filesystem-operations";
import type { StoredSnapshot } from "../persistence/database";
import { hash } from "../utils/hash";
import { isBinary } from "../utils/is-binary";
import { isFileTypeMergable } from "../utils/is-file-type-mergable";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
export async function toStoredSnapshot(
snapshot: FileSnapshot
@ -47,9 +47,8 @@ export async function mergeContent(
remote.contentBase64
].map(base64ToBytes);
// Concurrent unmergeable edits keep the server's content. The filesystem
// journal retains displaced local bytes for recovery.
if (!isFileTypeMergable(path, extensions) || inputs.some(isBinary))
// Concurrent unmergeable edits keep the server's content.
if (!isFileTypeMergeable(path, extensions) || inputs.some(isBinary))
return remote;
const [parentText, localText, remoteText] = inputs.map((bytes) =>

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

@ -18,11 +18,13 @@ it("retains future cursors and publishes them when content catches up", async ()
getDocumentByDocumentId: () => ({
documentId: "document",
relativePath: "note.md",
isDeleted: false,
metadata: { parentVersionId: 2, hash: contentHash }
})
} as unknown as Database;
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteCursorsUpdateReceived: new EventListeners<
(cursors: ClientCursors[]) => Promise<void>
>(),
@ -65,3 +67,209 @@ it("retains future cursors and publishes them when content catches up", async ()
"note.md"
);
});
it("refreshes unchanged cursors on reconnect and heartbeat, and follows the document through renames", async (context) => {
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: string) =>
candidate === path ? record() : undefined,
getDocumentByDocumentId: () => record()
} as unknown as Database;
const sent: {
documentsWithCursors: {
relative_path: string;
vault_update_id: number | null;
}[];
}[] = [];
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteCursorsUpdateReceived: new EventListeners(),
updateLocalCursors: (value: (typeof sent)[number]) => {
sent.push(value);
}
} as unknown as WebSocketManager;
const notifier = new FileChangeNotifier();
const tracker = new CursorTracker(
database,
websocket,
{ read: async () => content } as unknown as FileOperations,
notifier
);
const drain = async () => {
for (let i = 0; i < 20; i++)
await new Promise((resolve) => setImmediate(resolve));
};
try {
await tracker.sendLocalCursorsToServer({
"note.md": [{ start: 1, end: 2 }]
});
assert.equal(sent.length, 1);
await tracker.sendLocalCursorsToServer({
"note.md": [{ start: 1, end: 2 }]
});
assert.equal(sent.length, 1);
websocket.onWebSocketStatusChanged.trigger(true);
await drain();
assert.equal(sent.length, 2);
context.mock.timers.tick(15_000);
await drain();
assert.equal(sent.length, 3);
path = "renamed.md";
await notifier.onFileChanged.triggerAsync(path);
assert.equal(sent.at(-1)!.documentsWithCursors[0].relative_path, path);
websocket.onWebSocketStatusChanged.trigger(false);
const count = sent.length;
context.mock.timers.tick(30_000);
await drain();
assert.equal(sent.length, count);
websocket.onWebSocketStatusChanged.trigger(true);
await drain();
assert.equal(sent.length, count + 1);
} finally {
tracker.reset();
context.mock.timers.reset();
}
});
for (const interruption of ["disconnect", "reset"] as const) {
it(`an in-flight cursor read cannot resurrect presence after ${interruption}`, async () => {
const content = new TextEncoder().encode("current");
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 websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteCursorsUpdateReceived: new EventListeners<
(cursors: ClientCursors[]) => Promise<void>
>(),
updateLocalCursors: () => {}
} as unknown as WebSocketManager;
const tracker = new CursorTracker(
database,
websocket,
{
read: async () => {
entered.resolve();
await release.promise;
return content;
}
} as unknown as FileOperations,
new FileChangeNotifier()
);
let received: ClientCursors[] = [];
tracker.onRemoteCursorsUpdated.add((cursors) => {
received = cursors;
});
const cursor: ClientCursors = {
userName: "Other",
deviceId: "device",
documentsWithCursors: [
{
document_id: "document",
relative_path: "note.md",
vault_update_id: 1,
cursors: [{ start: 0, end: 1 }]
}
]
};
const first = websocket.onRemoteCursorsUpdateReceived.triggerAsync([
cursor
]);
await entered.promise;
const queued = websocket.onRemoteCursorsUpdateReceived.triggerAsync([
cursor
]);
if (interruption === "reset") tracker.reset();
else websocket.onWebSocketStatusChanged.trigger(false);
assert.deepEqual(received, []);
release.resolve();
await first;
await queued;
assert.deepEqual(
received,
[],
"a retired connection must not restore its cursors"
);
tracker.reset();
});
}
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 entered = Promise.withResolvers<void>(),
release = Promise.withResolvers<void>();
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteCursorsUpdateReceived: new EventListeners<
(cursors: ClientCursors[]) => Promise<void>
>(),
updateLocalCursors: () => {}
} as unknown as WebSocketManager;
const tracker = new CursorTracker(
{
getDocumentByDocumentId: () => ({
documentId: "document",
relativePath: "note.md",
metadata
})
} as unknown as Database,
websocket,
{
read: async () => {
entered.resolve();
await release.promise;
return content;
}
} as unknown as FileOperations,
new FileChangeNotifier()
);
let outdated: boolean | undefined;
tracker.onRemoteCursorsUpdated.add((cursors) => {
outdated = cursors[0]?.isOutdated;
});
const update = websocket.onRemoteCursorsUpdateReceived.triggerAsync([
{
userName: "Other",
deviceId: "device",
documentsWithCursors: [
{
document_id: "document",
relative_path: "note.md",
vault_update_id: 1,
cursors: [{ start: 0, end: 1 }]
}
]
}
]);
await entered.promise;
metadata = {
parentVersionId: 2,
hash: await hash(new TextEncoder().encode("new content"))
};
release.resolve();
await update;
assert.equal(outdated, true);
tracker.reset();
});

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

@ -23,6 +23,7 @@ export class CursorTracker {
>();
private readonly updateLock = new Lock();
private sessionGeneration = 0;
private knownRemoteCursors: (ClientCursors & {
upToDateness: DocumentUpToDateness;
@ -31,6 +32,10 @@ export class CursorTracker {
private lastLocalCursorState: DocumentWithCursors[] = [];
private lastLocalCursorStateWithoutDirtyDocuments: DocumentWithCursors[] =
[];
private localCursors: Record<string, CursorSpan[]> = {};
private localCursorDocuments: Record<string, string> = {};
private cursorTimer?: ReturnType<typeof setInterval>;
private readonly sendLock = new Lock();
public constructor(
private readonly database: Database,
@ -38,9 +43,31 @@ export class CursorTracker {
private readonly fileOperations: FileOperations,
private readonly fileChangeNotifier: FileChangeNotifier
) {
this.webSocketManager.onWebSocketStatusChanged.add((connected) => {
this.sessionGeneration++;
clearInterval(this.cursorTimer);
this.lastLocalCursorState = [];
this.lastLocalCursorStateWithoutDirtyDocuments = [];
if (!connected) {
this.knownRemoteCursors = [];
this.onRemoteCursorsUpdated.trigger([]);
return;
}
const refresh = () => {
void this.publishLocalCursors(true).catch(() => {
// A concurrent move/deletion can invalidate an editor view;
// the next notification or heartbeat retries the snapshot.
});
};
refresh();
this.cursorTimer = setInterval(refresh, 15_000);
this.cursorTimer.unref?.();
});
this.webSocketManager.onRemoteCursorsUpdateReceived.add(
async (clientCursors) => {
const generation = this.sessionGeneration;
await this.updateLock.withLock(async () => {
if (generation !== this.sessionGeneration) return;
// The latest message will contain all active clients, so we can delete the ones
// from the local list which are no longer active.
const allIds = new Set(
@ -51,11 +78,7 @@ export class CursorTracker {
allIds.has(c.deviceId)
);
for (const cursor of clientCursors.filter((client) =>
client.documentsWithCursors.every(
(doc) => doc.vault_update_id != null
)
)) {
for (const cursor of clientCursors) {
updatedKnownRemoteCursors.push({
...cursor,
upToDateness:
@ -63,17 +86,20 @@ export class CursorTracker {
});
}
if (generation !== this.sessionGeneration) return;
this.knownRemoteCursors = updatedKnownRemoteCursors;
this.onRemoteCursorsUpdated.trigger(
this.getRelevantAndPruneKnownClientCursors()
);
});
this.onRemoteCursorsUpdated.trigger(
this.getRelevantAndPruneKnownClientCursors()
);
}
);
this.fileChangeNotifier.onFileChanged.add(async (relativePath) => {
const generation = this.sessionGeneration;
await this.publishLocalCursors(true);
const cursors = await this.updateLock.withLock(async () => {
if (generation !== this.sessionGeneration) return [];
for (const clientCursor of this.knownRemoteCursors) {
if (
clientCursor.documentsWithCursors.some(
@ -88,10 +114,13 @@ export class CursorTracker {
await this.getDocumentsUpToDateness(clientCursor);
}
}
return this.getRelevantAndPruneKnownClientCursors();
return generation === this.sessionGeneration
? this.getRelevantAndPruneKnownClientCursors()
: [];
});
this.onRemoteCursorsUpdated.trigger(cursors);
if (generation === this.sessionGeneration)
this.onRemoteCursorsUpdated.trigger(cursors);
});
}
@ -101,71 +130,106 @@ export class CursorTracker {
public async sendLocalCursorsToServer(
documentToCursors: Record<RelativePath, CursorSpan[]>
): Promise<void> {
const documentsWithCursors: DocumentWithCursors[] = [];
this.localCursors = structuredClone(documentToCursors);
this.localCursorDocuments = {};
await this.publishLocalCursors(false);
}
for (const [relativePath, cursors] of Object.entries(
documentToCursors
)) {
const record =
this.database.getLatestDocumentByRelativePath(relativePath);
private async publishLocalCursors(force: boolean): Promise<void> {
const generation = this.sessionGeneration;
await this.sendLock.withLock(async () => {
if (generation !== this.sessionGeneration) return;
const documentsWithCursors: DocumentWithCursors[] = [];
if (!record) {
continue; // Let's wait for the file to be created before sending cursors
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
}
documentsWithCursors.push({
relative_path: relativePath,
document_id: record.documentId,
vault_update_id: record.metadata.parentVersionId,
cursors: cursors.map(({ start, end }) => ({
start: Math.min(start, end),
end: Math.max(start, end)
})) // the client might send directional selections
});
}
if (!record.metadata) {
continue; // this is a new document, no need to sync the cursors
if (
!force &&
JSON.stringify(this.lastLocalCursorState) ===
JSON.stringify(documentsWithCursors)
) {
// Caching step to avoid reading the edited files all the time
return;
}
this.lastLocalCursorState = structuredClone(documentsWithCursors);
for (const doc of documentsWithCursors) {
let readContent: Uint8Array;
try {
readContent = await this.fileOperations.read(
doc.relative_path
);
} catch {
doc.vault_update_id = null;
continue;
}
const digest = await hash(readContent);
const record = this.database.getLatestDocumentByRelativePath(
doc.relative_path
);
if (
record?.documentId !== doc.document_id ||
record.relativePath !== doc.relative_path ||
record.metadata?.parentVersionId !== doc.vault_update_id ||
record.metadata.hash !== digest
) {
doc.vault_update_id = null;
}
}
documentsWithCursors.push({
relative_path: relativePath,
document_id: record.documentId,
vault_update_id: record.metadata.parentVersionId,
cursors: cursors.map(({ start, end }) => ({
start: Math.min(start, end),
end: Math.max(start, end)
})) // the client might send directional selections
});
}
if (
JSON.stringify(this.lastLocalCursorState) ===
JSON.stringify(documentsWithCursors)
) {
// Caching step to avoid reading the edited files all the time
return;
}
this.lastLocalCursorState = documentsWithCursors;
for (const doc of documentsWithCursors) {
const readContent = await this.fileOperations.read(
doc.relative_path
);
const record = this.database.getLatestDocumentByRelativePath(
doc.relative_path
);
if (record?.metadata?.hash !== (await hash(readContent))) {
doc.vault_update_id = null;
if (generation !== this.sessionGeneration) return;
if (
!force &&
JSON.stringify(
this.lastLocalCursorStateWithoutDirtyDocuments
) === JSON.stringify(documentsWithCursors)
) {
return;
}
}
if (
JSON.stringify(this.lastLocalCursorStateWithoutDirtyDocuments) ===
JSON.stringify(documentsWithCursors)
) {
return;
}
this.lastLocalCursorStateWithoutDirtyDocuments =
documentsWithCursors;
this.lastLocalCursorStateWithoutDirtyDocuments = documentsWithCursors;
this.webSocketManager.updateLocalCursors({ documentsWithCursors });
this.webSocketManager.updateLocalCursors({ documentsWithCursors });
});
}
public reset(): void {
this.sessionGeneration++;
clearInterval(this.cursorTimer);
this.knownRemoteCursors = [];
this.onRemoteCursorsUpdated.trigger([]);
this.lastLocalCursorState = [];
this.lastLocalCursorStateWithoutDirtyDocuments = [];
this.updateLock.reset();
// Keep serialization intact while old reads drain. A generation fence
// discards both active and queued work from the retired connection.
}
private getRelevantAndPruneKnownClientCursors(): MaybeOutdatedClientCursors[] {
@ -245,6 +309,8 @@ export class CursorTracker {
private async getDocumentUpToDateness(
document: DocumentWithCursors
): Promise<DocumentUpToDateness> {
if (document.vault_update_id === null)
return DocumentUpToDateness.Prior;
const record = this.database.getDocumentByDocumentId(
document.document_id
);
@ -255,13 +321,11 @@ export class CursorTracker {
}
if (
(record.metadata?.parentVersionId ?? 0) <
(document.vault_update_id ?? 0)
(record.metadata?.parentVersionId ?? 0) < document.vault_update_id
) {
return DocumentUpToDateness.Later;
} else if (
(document.vault_update_id ?? 0) <
(record.metadata?.parentVersionId ?? 0)
document.vault_update_id < (record.metadata?.parentVersionId ?? 0)
) {
// the document of the cursor must be from the past
return DocumentUpToDateness.Prior;
@ -276,7 +340,14 @@ export class CursorTracker {
return DocumentUpToDateness.Prior;
}
return record.metadata?.hash === (await hash(currentContent))
const contentHash = await hash(currentContent);
const current = this.database.getDocumentByDocumentId(
document.document_id
);
return current?.metadata?.parentVersionId ===
document.vault_update_id &&
current.relativePath === record.relativePath &&
current.metadata.hash === contentHash
? DocumentUpToDateness.UpToDate
: DocumentUpToDateness.Prior;
}

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

@ -0,0 +1,69 @@
import type { FileManifestEntries } from "../types/file-manifest-entries";
import {
allocatePortablePath,
validatePortablePaths
} from "../utils/portable-path";
export function validateFileManifest(entries: FileManifestEntries): void {
validatePortablePaths(Object.values(entries));
}
export function sameFileManifest(
a: FileManifestEntries,
b: FileManifestEntries
): boolean {
return (
Object.keys(a).length === Object.keys(b).length &&
Object.keys(a).every((id) => a[id] === b[id])
);
}
export function resolvePaths(
proposed: FileManifestEntries,
remote: FileManifestEntries,
protectedIds: readonly string[] = [],
reservedPaths: readonly string[] = []
): FileManifestEntries {
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 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
)) {
result[id] = allocatePortablePath(proposed[id], id, occupied);
occupied[id] = result[id];
}
return result;
}
export function mergeFileManifests(
base: FileManifestEntries,
local: FileManifestEntries,
remote: FileManifestEntries
): FileManifestEntries {
const proposed: FileManifestEntries = {};
for (const id of new Set([
...Object.keys(base),
...Object.keys(local),
...Object.keys(remote)
])) {
const path =
local[id] === base[id]
? remote[id]
: remote[id] === base[id]
? local[id]
: remote[id];
if (path !== undefined) proposed[id] = path;
}
return resolvePaths(proposed, remote);
}

80
frontend/sync-client/src/sync-operations/local-changes.ts Normal file
View file

@ -0,0 +1,80 @@
import { v4 as uuid } from "uuid";
import type { EngineState } from "../persistence/database";
import type { VaultSnapshot } from "../services/types/VaultSnapshot";
import { arePathAliases } from "../utils/portable-path";
export type LocalChange = {
changeId?: string;
/** Identities at notification time, before a remote application can reuse paths. */
identities?: Record<string, string>;
} & (
| { type: "move"; oldPath: string; relativePath: string }
| { type: "create"; path: string; documentId?: string }
| { type: "delete"; path: string }
);
export function unappliedChanges<T extends { changeId?: string }>(
state: EngineState,
changes: readonly T[]
): T[] {
for (const change of changes) change.changeId ??= uuid();
const applied = changes.findIndex(
(change) => change.changeId === state.lastAppliedLocalChangeId
);
return changes.slice(applied + 1);
}
export function applyLocalChange(
next: EngineState,
change: LocalChange,
ignored: (path: string) => boolean,
initial?: VaultSnapshot
): void {
if (change.type === "create") {
if (
ignored(change.path) ||
Object.values(next.local).includes(change.path)
)
return;
const id = (change.documentId ??=
(initial &&
Object.keys(initial.fileManifest.entries).find(
(id) =>
!next.local[id] &&
arePathAliases(
initial.fileManifest.entries[id],
change.path
)
)) ||
uuid());
next.local[id] = change.path;
next.documents[id] ??= {
materialized: true,
bootstrap: initial?.fileManifest.entries[id] !== undefined
};
return;
}
const from = change.type === "move" ? change.oldPath : change.path;
change.identities ??= Object.fromEntries(
Object.entries(next.local).filter(
([id, path]) =>
next.documents[id]?.materialized &&
(path === from || path.startsWith(from + "/"))
)
);
for (const [id, path] of Object.entries(change.identities)) {
if (change.type === "delete") {
if (ignored(path)) {
next.excluded ??= {};
if (next.excluded[id] === undefined)
next.excluded[id] = next.fileManifest.entries[id] ?? null;
}
delete next.local[id];
} else {
const to = change.relativePath + path.slice(from.length);
for (const [other, otherPath] of Object.entries(next.local))
if (other !== id && otherPath === to) delete next.local[other];
next.local[id] = to;
}
}
}

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

@ -0,0 +1,58 @@
import assert from "node:assert/strict";
import { test } from "node:test";
import { CursorTracker } from "./cursor-tracker";
import { FileChangeNotifier } from "./file-change-notifier";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { hash } from "../utils/hash";
import type { Database, DocumentRecord } from "../persistence/database";
import type { FileOperations } from "../file-operations/file-operations";
import type { WebSocketManager } from "../services/websocket-manager";
import type { DocumentWithCursors } from "../services/types/DocumentWithCursors";
for (const change of ["version", "identity", "path", "delete"] as const)
test(`outgoing cursor is dirty when ${change} changes during its snapshot read`, async () => {
// 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 sent: { documentsWithCursors: DocumentWithCursors[] }[] = [];
const websocket = {
onWebSocketStatusChanged: new EventListeners(),
onRemoteCursorsUpdateReceived: new EventListeners(),
updateLocalCursors: (data: {
documentsWithCursors: DocumentWithCursors[];
}) => sent.push(data)
};
const files = {
read: async () => {
if (change === "delete") current = undefined;
else if (change === "version")
current!.metadata!.parentVersionId = 2;
else if (change === "identity")
current!.documentId = "replacement";
else current!.relativePath = "moved.md";
return content;
}
};
const tracker = new CursorTracker(
{
getDocumentByDocumentId: (id: string) =>
current?.documentId === id ? current : undefined,
getLatestDocumentByRelativePath: (path: string) =>
current?.relativePath === path ? current : undefined
} as unknown as Database,
websocket as unknown as WebSocketManager,
files as unknown as FileOperations,
new FileChangeNotifier()
);
await tracker.sendLocalCursorsToServer({
"a.md": [{ start: 4, end: 4 }]
});
assert.equal(sent.length, 1);
assert.equal(sent[0].documentsWithCursors[0].vault_update_id, null);
tracker.reset();
});

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

@ -0,0 +1,152 @@
import { v4 as uuid } from "uuid";
import type { FileOperations } from "../file-operations/file-operations";
import type { EngineState } from "../persistence/database";
import type { VaultSnapshot } from "../services/types/VaultSnapshot";
import { EMPTY_HASH, hash } from "../utils/hash";
import {
arePathAliases,
findPathWithSameSpelling
} from "../utils/portable-path";
import {
applyLocalChange,
unappliedChanges,
type LocalChange
} from "./local-changes";
export type { LocalChange } from "./local-changes";
interface ScanOptions {
next: EngineState;
guard: () => void;
changes: LocalChange[];
files: FileOperations;
ignored: (path: string) => 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,
guard,
changes: queuedChanges,
files,
ignored,
oversized,
commit
}: ScanOptions,
initial?: VaultSnapshot
): Promise<Set<string>> {
// Leave the prefix queued until its resulting state is saved.
// A persisted change ID makes replay safe after an uncertain save.
const changes = [...queuedChanges];
const unsyncablePaths = new Set<string>();
for (const change of unappliedChanges(next, changes)) {
applyLocalChange(next, change, ignored, initial);
next.lastAppliedLocalChangeId = change.changeId;
}
const paths: string[] = [];
const reportedPaths = await files.listFilesRecursively();
for (const path of reportedPaths) {
// Some filesystems may report a different Unicode spelling after a
// rename. Keep an already-accessible spelling when possible.
const alias = findPathWithSameSpelling(Object.values(next.local), path);
paths.push(
alias &&
!reportedPaths.includes(alias) &&
(await files.fs.exists(alias))
? alias
: path
);
}
const present = new Set(paths);
const snapshots = new Map<string, { hash: string }>();
for (const path of paths) {
if (ignored(path)) {
unsyncablePaths.add(path);
continue;
}
const info = await files.fs.stat(path);
if (!info || info.kind !== "file")
throw new Error(`File changed during scan: ${path}`);
if (oversized(info.size)) {
unsyncablePaths.add(path);
continue;
}
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) });
} else {
unsyncablePaths.add(path);
}
}
const missing = new Set(
Object.keys(next.local).filter(
(id) =>
next.documents[id]?.materialized &&
!ignored(next.local[id]) &&
!present.has(next.local[id])
)
);
const knownPaths = new Set(Object.values(next.local));
const unknown = [...snapshots.keys()].filter(
(path) => !knownPaths.has(path)
);
for (const path of unknown) {
const snapshot = snapshots.get(path)!;
const matches = [...missing].filter(
(id) => next.documents[id]?.observedHash === snapshot.hash
);
const sameHash = unknown.filter(
(other) => snapshots.get(other)?.hash === snapshot.hash
);
let id: string | undefined;
if (
snapshot.hash !== EMPTY_HASH &&
matches.length === 1 &&
sameHash.length === 1
) {
id = matches[0];
missing.delete(id);
} else if (initial) {
id = Object.keys(initial.fileManifest.entries).find(
(candidate) =>
!next.local[candidate] &&
arePathAliases(
initial.fileManifest.entries[candidate],
path
)
);
}
id ??= uuid();
next.local[id] = path;
next.documents[id] ??= {
materialized: true,
bootstrap: initial?.fileManifest.entries[id] !== undefined
};
}
for (const id of missing) delete next.local[id];
for (const [id, path] of Object.entries(next.local)) {
if (unsyncablePaths.has(path) || ignored(path)) {
next.excluded ??= {};
if (next.excluded[id] === undefined)
next.excluded[id] = next.fileManifest.entries[id] ?? null;
}
const snapshot = snapshots.get(path);
if (snapshot) {
next.documents[id] ??= { materialized: true };
if (!next.documents[id].materialized && !next.documents[id].base)
next.documents[id].bootstrap = true;
next.documents[id].materialized = true;
next.documents[id].observedHash = snapshot.hash;
}
}
next.protectedPaths = [...unsyncablePaths];
guard();
await commit(next);
queuedChanges.splice(0, changes.length);
guard();
return unsyncablePaths;
}

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

@ -0,0 +1,327 @@
/* eslint-disable @typescript-eslint/no-unsafe-type-assertion */
import assert from "node:assert";
import { afterEach, beforeEach, describe, it, mock } from "node:test";
import type { Database } from "../persistence/database";
import {
DEFAULT_SETTINGS,
Settings,
type SyncSettings
} from "../persistence/settings";
import type { FileOperations } from "../file-operations/file-operations";
import type { SyncService } from "../services/sync-service";
import type { ServerConfig } from "../services/server-config";
import type { WebSocketManager } from "../services/websocket-manager";
import { Logger } from "../tracing/logger";
import type { SyncHistory } from "../tracing/sync-history";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { Syncer } from "./syncer";
import { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache";
import { toStoredSnapshot } from "./content";
import { undiff } from "reconcile-text";
function createSyncer(overrides: Partial<SyncSettings> = {}): {
syncer: Syncer;
runCount: () => number;
} {
const settings = new Settings(
new Logger(),
overrides,
async () => undefined
);
const database = {
state: {}
} as unknown as Database;
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteVaultUpdateReceived: new EventListeners<() => unknown>()
} as unknown as WebSocketManager;
const syncer = new Syncer(
"device",
{} as Logger,
database,
settings,
{} as SyncService,
websocket,
{} as FileOperations,
{} as ServerConfig,
{} as SyncHistory,
new FixedSizeDocumentCache(1024)
);
let runs = 0;
const internals = syncer as unknown as {
dirty: boolean;
run: () => Promise<void>;
};
internals.run = async (): Promise<void> => {
runs++;
internals.dirty = false;
};
return { syncer, runCount: () => runs };
}
describe("Syncer interval", () => {
beforeEach(() => {
mock.timers.enable({ apis: ["setTimeout"] });
});
afterEach(() => {
mock.timers.reset();
});
it("does not schedule a timer when syncIntervalMs is unset", async () => {
const { syncer, runCount } = createSyncer({
syncIntervalMs: undefined
});
syncer.start();
await syncer.waitUntilFinished();
mock.timers.tick(60_000);
assert.strictEqual(runCount(), 1);
await syncer.stop();
});
it("keeps periodic reconciliation opt-in by default", async () => {
assert.strictEqual(DEFAULT_SETTINGS.syncIntervalMs, undefined);
const { syncer, runCount } = createSyncer();
syncer.start();
await syncer.waitUntilFinished();
mock.timers.tick(60_000);
assert.strictEqual(runCount(), 1);
await syncer.stop();
});
it("does not poll when explicitly disabled with zero", async () => {
const { syncer, runCount } = createSyncer({ syncIntervalMs: 0 });
syncer.start();
await syncer.waitUntilFinished();
mock.timers.tick(60_000);
assert.strictEqual(runCount(), 1);
await syncer.stop();
});
it("wakes after the configured sync interval", async () => {
const { syncer, runCount } = createSyncer({ syncIntervalMs: 25 });
syncer.start();
await syncer.waitUntilFinished();
mock.timers.tick(24);
assert.strictEqual(runCount(), 1);
mock.timers.tick(1);
assert.strictEqual(runCount(), 2);
await syncer.waitUntilFinished();
await syncer.stop();
});
});
describe("Syncer per-file state and event delivery", () => {
const head = {
vaultUpdateId: 1,
documentId: "document",
updatedDate: "2026-01-01T00:00:00Z",
userId: "user",
deviceId: "device",
contentSize: 4
};
function fixture() {
const database = {
state: {
initialized: true,
local: { document: "note.md" },
documents: {
document: {
materialized: true,
observedHash: "hash",
base: { ...head, hash: "hash" }
}
},
fileManifest: {
fileManifestId: 1,
entries: { document: "note.md" }
},
remoteHeads: { document: head },
lastSeenUpdateId: 0
},
getLatestDocumentByRelativePath: (path: string) =>
path === "note.md"
? {
documentId: "document",
relativePath: path
}
: undefined
} as unknown as Database;
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteVaultUpdateReceived: new EventListeners<
() => Promise<void>
>()
} as unknown as WebSocketManager;
const service = {
events: async () => {
return { headEventId: 0, events: [] };
}
} as unknown as SyncService;
const syncer = new Syncer(
"device",
{} as Logger,
database,
new Settings(
new Logger(),
{ syncIntervalMs: 0 },
async () => undefined
),
service,
websocket,
{} as FileOperations,
{} as ServerConfig,
{} as SyncHistory,
new FixedSizeDocumentCache(1024)
);
(syncer as unknown as { hasScanned: boolean }).hasScanned = true;
return {
database,
websocket,
syncer
};
}
it("reports only a genuinely current path as up to date", () => {
const { syncer } = fixture();
assert.strictEqual(syncer.isDocumentUpToDate("note.md"), true);
assert.strictEqual(syncer.isDocumentUpToDate("missing.md"), false);
const internals = syncer as unknown as {
dirty: boolean;
unsyncablePaths: Set<string>;
};
internals.dirty = true;
assert.strictEqual(syncer.isDocumentUpToDate("note.md"), true);
internals.unsyncablePaths.add("note.md");
assert.strictEqual(syncer.isDocumentUpToDate("note.md"), false);
});
it("projects content responses to metadata before persistence", () => {
const { syncer } = fixture();
const projected = (
syncer as unknown as {
documentHead: (value: typeof head) => Record<string, unknown>;
}
).documentHead({ ...head, contentBase64: "secret" } as typeof head);
assert.deepEqual(projected, head);
assert.strictEqual("contentBase64" in projected, false);
});
});
describe("Syncer diff uploads", () => {
const encoder = new TextEncoder();
function fixture(cacheSize = 1024): {
syncer: Syncer;
parent: Uint8Array;
} {
const cache = new FixedSizeDocumentCache(cacheSize);
const parent = encoder.encode("one two three\n");
const websocket = {
onWebSocketStatusChanged: new EventListeners<
(connected: boolean) => unknown
>(),
onRemoteVaultUpdateReceived: new EventListeners<() => unknown>()
} as unknown as WebSocketManager;
const syncer = new Syncer(
"device",
{} as Logger,
{ state: {} } as Database,
new Settings(
new Logger(),
{ syncIntervalMs: 0 },
async () => undefined
),
{
getDocumentVersionContent: async () => parent
} as unknown as SyncService,
websocket,
{} as FileOperations,
{
getConfig: async () => ({ mergeableFileExtensions: ["md"] })
} as ServerConfig,
{} as SyncHistory,
cache
);
return { syncer, parent };
}
it("sends a diff when the fetched parent is still cached", async () => {
const { syncer, parent } = fixture();
const internals = syncer as unknown as {
remoteSnapshot: (
head: {
vaultUpdateId: number;
documentId: string;
},
path: string
) => Promise<unknown>;
pushContent: (
path: string,
parentVersionId: number,
snapshot: Awaited<ReturnType<typeof toStoredSnapshot>>
) => Promise<
| { type: "Diff"; value: (number | string)[] }
| { type: "Snapshot"; value: string }
>;
};
await internals.remoteSnapshot(
{ vaultUpdateId: 7, documentId: "document" },
"note.md"
);
const changed = "one changed three\n";
const payload = await internals.pushContent(
"note.md",
7,
await toStoredSnapshot({ content: encoder.encode(changed) })
);
assert.equal(payload.type, "Diff");
if (payload.type === "Diff") {
assert.equal(
undiff(new TextDecoder().decode(parent), payload.value),
changed
);
}
});
it("falls back to a snapshot when the parent cannot fit in the cache", async () => {
const { syncer } = fixture(0);
const internals = syncer as unknown as {
remoteSnapshot: (
head: {
vaultUpdateId: number;
documentId: string;
},
path: string
) => Promise<unknown>;
pushContent: (
path: string,
parentVersionId: number,
snapshot: Awaited<ReturnType<typeof toStoredSnapshot>>
) => Promise<{ type: string; value: unknown }>;
};
await internals.remoteSnapshot(
{ vaultUpdateId: 7, documentId: "document" },
"note.md"
);
const payload = await internals.pushContent(
"note.md",
7,
await toStoredSnapshot({ content: encoder.encode("changed") })
);
assert.equal(payload.type, "Snapshot");
});
});

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

File diff suppressed because it is too large Load diff

39
frontend/sync-client/src/utils/portable-path.test.ts Normal file
View file

@ -0,0 +1,39 @@
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { resolve } from "node:path";
import { pathToFileURL } from "node:url";
import { test } from "node:test";
import { validatePortablePaths } from "./portable-path";
test("a small deeply nested manifest fits within a bounded validation heap", () => {
const source = pathToFileURL(resolve(__dirname, "portable-path.ts")).href;
const result = spawnSync(
process.execPath,
[
"--max-old-space-size=64",
"--import",
require.resolve("tsx"),
"--input-type=module",
"--eval",
`
import { validatePortablePaths } from ${JSON.stringify(source)};
const directory = Array(20000).fill("a").join("/");
validatePortablePaths([directory + "/one.md", directory + "/two.md"]);
`
],
{ encoding: "utf8", timeout: 10000 }
);
assert.equal(result.status, 0, result.stderr || String(result.error));
});
test("shared directories remain distinct from aliases and file ancestors", () => {
validatePortablePaths(["Notes/one.md", "Notes/two.md", "Other/one.md"]);
for (const paths of [
["Notes/one.md", "notes/two.md"],
["Straße/one.md", "STRASSE/two.md"],
["Notes", "Notes/one.md"],
["Notes/one.md", "Notes"],
["Notes/one.md", "Notes/one.md"]
])
assert.throws(() => validatePortablePaths(paths), /Conflicting path/);
});

190
frontend/sync-client/src/utils/portable-path.ts Normal file
View file

@ -0,0 +1,190 @@
export const INTERNAL_DIRECTORY = ".vault-link-sync";
const MAX_COMPONENT_BYTES = 255;
const byteLength = (value: string): number =>
new TextEncoder().encode(value).length;
function truncate(value: string, limit: number): string {
let result = "";
for (const character of value) {
if (byteLength(result + character) > limit) break;
result += character;
}
return result;
}
const reservedDeviceName =
/^(CON|PRN|AUX|NUL|CONIN\$|CONOUT\$|COM[1-9¹²³]|LPT[1-9¹²³])$/u;
// NFC -> uppercase -> NFC matches the server's locale-independent alias rules.
const pathKey = (path: string): string =>
path.normalize("NFC").toUpperCase().normalize("NFC");
export const arePathAliases = (left: string, right: string): boolean =>
pathKey(left) === pathKey(right);
export function findPathWithSameSpelling(
paths: Iterable<string>,
wanted: string
): string | undefined {
const normalized = wanted.normalize("NFC");
for (const path of paths) {
if (path !== wanted && path.normalize("NFC") === normalized)
return path;
}
return undefined;
}
export const isInternalPath = (path: string): boolean =>
arePathAliases(path.split("/")[0], INTERNAL_DIRECTORY);
function validatePath(path: string): void {
if (!path || path !== path.normalize("NFC") || isInternalPath(path))
throw new Error(`Invalid portable path: ${path}`);
for (const part of path.split("/")) {
if (
!part ||
part === "." ||
part === ".." ||
/[<>:"\\|?*\p{Cc}]|[. ]$/u.test(part) ||
reservedDeviceName.test(
part.split(".")[0].replace(/ +$/u, "").toUpperCase()
)
)
throw new Error(`Invalid portable path: ${path}`);
}
}
export function validatePortablePaths(paths: Iterable<string>): void {
// Component keys keep storage proportional to the input, even when a
// manifest contains thousands of directory levels.
const nodes = new Map<
string,
{ id: number; spelling: string; file: boolean }
>();
for (const path of paths) {
validatePath(path);
const parts = path.split("/");
let parent = 0;
parts.forEach((spelling, index) => {
const file = index === parts.length - 1;
const key = `${parent}/${pathKey(spelling)}`;
const previous = nodes.get(key);
if (
previous &&
(previous.spelling !== spelling || previous.file || file)
)
throw new Error(`Conflicting path: ${path}`);
const node = previous ?? { id: nodes.size + 1, spelling, file };
nodes.set(key, node);
parent = node.id;
});
}
}
function portableName(part: string): string {
let result = part
.normalize("NFC")
.replace(/[<>:"\\|?*\p{Cc}]/gu, "_")
.replace(/[. ]+$/u, "_");
if (!result || result === "." || result === "..") result = "_";
if (
reservedDeviceName.test(
result.split(".")[0].replace(/ +$/u, "").toUpperCase()
)
)
result = "_" + result;
return result;
}
function fitName(name: string, suffix: string, file: boolean): string {
const dot = file ? name.lastIndexOf(".") : -1;
const extension = dot > 0 ? truncate(name.slice(dot), 64) : "";
const stem = dot > 0 ? name.slice(0, dot) : name;
return portableName(
truncate(stem, MAX_COMPONENT_BYTES - byteLength(suffix + extension)) +
suffix +
extension
);
}
function pathsConflict(left: string, right: string): boolean {
const a = left.split("/"),
b = right.split("/");
for (let i = 0; i < Math.min(a.length, b.length); i++) {
if (!arePathAliases(a[i], b[i])) return false;
if (a[i] !== b[i]) return true;
}
return true; // Equal names, or a file occupies an ancestor directory.
}
function isAvailablePath(
candidate: string,
id: string,
occupied: Readonly<Record<string, string>>
): boolean {
validatePath(candidate);
return Object.entries(occupied).every(
([otherId, path]) => otherId === id || !pathsConflict(candidate, path)
);
}
/** Deterministic allocation also handles a file blocking an ancestor directory. */
export function allocatePortablePath(
wanted: string,
id: string,
occupied: Readonly<Record<string, string>>
): string {
id = portableName(truncate(id, 64));
const originals = wanted.split("/");
const parts = originals.map((part, index) => {
const portable = portableName(part);
return byteLength(portable) > MAX_COMPONENT_BYTES
? fitName(
portable,
` (conflict ${id})`,
index === originals.length - 1
)
: portable;
});
if (isInternalPath(parts.join("/"))) parts[0] = "_" + parts[0];
let candidate = parts.join("/");
const suffix = ` (conflict ${id})`;
const limit = (Object.keys(occupied).length + 1) * (parts.length + 1);
for (let attempt = 0; attempt <= limit; attempt++) {
try {
if (isAvailablePath(candidate, id, occupied)) return candidate;
} catch {
/* allocate below */
}
const current = candidate.split("/");
let conflictAt = current.length - 1;
for (let i = 0; i < current.length; i++) {
const prefix = current.slice(0, i + 1).join("/");
if (
Object.values(occupied).some((path) => {
const other = path
.split("/")
.slice(0, i + 1)
.join("/");
return (
arePathAliases(prefix, other) &&
(prefix !== other ||
path.split("/").length === i + 1 ||
i === current.length - 1)
);
})
) {
conflictAt = i;
break;
}
}
const original = parts[conflictAt] ?? current[conflictAt];
current[conflictAt] = fitName(
original,
suffix + (attempt ? ` (${attempt})` : ""),
conflictAt === current.length - 1
);
candidate = current.join("/");
}
throw new Error(`Unable to allocate a portable path: ${wanted}`);
}