This commit is contained in:
Andras Schmelczer 2026-09-26 20:22:16 +01:00
commit b07aa589ff
35 changed files with 728 additions and 530 deletions

View file

@ -18,11 +18,22 @@ UUID is rejected.
## Reconciliation ## Reconciliation
A new client reads a coherent vault snapshot, applies it, then replays An absent event cursor marks a new client or a durable history reset. The client
reads a coherent vault snapshot, applies it, then replays
subsequent events over HTTP. Pages are folded in memory to their final document subsequent events over HTTP. Pages are folded in memory to their final document
heads and manifest before changing local files. Interrupted fetching restarts from heads and manifest before changing local files. Interrupted fetching restarts from
the last incorporated cursor without saved pagination state. This avoids applying the last incorporated cursor without saved pagination state. This avoids applying
intermediate remote edits that were already reverted. intermediate remote edits that were already reverted. Bootstrap snapshots are
refetched after interruption rather than persisted. One reconciliation pass
handles the folded events, local scans, and previously deferred downloads.
Each document stores a base version/hash and the last observed disk hash. A remote
head is retained only until it can be applied. The observed hash records local
presence as well as rename evidence; there are no separate bootstrap or
materialization flags. Scans rebuild physical exclusions. Original server paths
remain saved while excluded, so ignoring a file cannot publish a deletion or
forget a deferred rename. Rejected payloads and messages live on the retained
requests that event replay already needs for uncertain acknowledgements.
For text, the client compares the common base, current local bytes and current For text, the client compares the common base, current local bytes and current
remote bytes using `reconcile-text`. Configured extensions and valid UTF-8 without remote bytes using `reconcile-text`. Configured extensions and valid UTF-8 without
@ -75,9 +86,11 @@ The client persists and echoes it; the server rejects a checkpoint from a discar
history before executing its request. Each new event gets a fresh token even if history before executing its request. Each new event gets a fresh token even if
its numeric ID or request UUID was used on the discarded branch. its numeric ID or request UUID was used on the discarded branch.
After a mismatch, the client records reset intent, clears old history-dependent state and bootstraps against the restored After a mismatch, the client clears the event cursor and old history-dependent
server. Clean files reconcile against the restored snapshot. Unsent edits with an state before releasing the transport checkpoint, then bootstraps against the
old base are preserved as separate local documents, with conflict names if needed. restored server. Clean files retain only their hash, with no old version ID or
content copy. Unsent edits, including changes made while recovery is interrupted,
are preserved as separate local documents, with conflict names if needed.
Interrupted recovery saves are replayable. Backups must be restored with the server Interrupted recovery saves are replayable. Backups must be restored with the server
stopped. stopped.

View file

@ -72,7 +72,7 @@ async function main() {
assert(documentId, "Edit must target an established document"); assert(documentId, "Edit must target an established document");
editedBase = { editedBase = {
documentId, documentId,
vaultUpdateId: state.documents![documentId].base!.vaultUpdateId vaultUpdateId: state.documents![documentId].base!.vaultUpdateId!
}; };
await disk.userWrite( await disk.userWrite(
input.editAfterSync.path, input.editAfterSync.path,

View file

@ -21,7 +21,14 @@ resuming starts a fresh session. Connection checks can run while sync is disable
## State and policy ## State and policy
The persisted state contains the incorporated server file manifest, the current The persisted state contains the incorporated server file manifest, the current
local identity map, and the exact pending submission. A received server file local identity map, an optional event cursor, and exact pending submissions. An
absent cursor means the client must fetch a fresh vault snapshot; there are no
separate initialization or recovery flags and no saved bootstrap snapshot. Each
document stores its base version/hash, its last observed disk hash, and a remote
head only while that content is deferred. The observed hash also records local
file presence; missing remote downloads are not local deletions. Physical path
exclusions are rebuilt by scans. Rejection details live on the retained request,
without duplicate per-document or manifest rejection records. A received server file
manifest becomes the merge base when its application metadata is saved. Successful manifest becomes the merge base when its application metadata is saved. Successful
submissions advance the base to the submitted content snapshot, preserving later submissions advance the base to the submitted content snapshot, preserving later
local edits. Unknown outcomes are retried unchanged before incorporating newer local edits. Unknown outcomes are retried unchanged before incorporating newer
@ -72,7 +79,9 @@ event incarnation returns a history mismatch before executing the request.
On mismatch the client scans local work and records a reset before clearing the On mismatch the client scans local work and records a reset before clearing the
old checkpoint. It clears old requests, receipts, cursors, bases and diff-cache old checkpoint. It clears old requests, receipts, cursors, bases and diff-cache
entries. Clean files reconcile against the restored snapshot. Unsent edits with entries. Clean files retain only their hash and reconcile against the restored
snapshot. A file edited after this reset is also preserved as a separate local
document; recovery never persists another copy of its content. Unsent edits with
an old base become separate local documents; path conflicts preserve them under an old base become separate local documents; path conflicts preserve them under
conflict names instead of applying a diff to an unrelated restored version. conflict names instead of applying a diff to an unrelated restored version.
Recovery resumes after interrupted state saves. Existing API v4 client state Recovery resumes after interrupted state saves. Existing API v4 client state
@ -112,7 +121,7 @@ object. A failed save is reloaded before retrying. No ordering between metadata
user-file durability is required. Settings and engine saves are serialized. user-file durability is required. Settings and engine saves are serialized.
Keep metadata outside the user-file namespace or in `.vault-link-sync`, which is Keep metadata outside the user-file namespace or in `.vault-link-sync`, which is
excluded from scans. Changing a client's vault once it holds identities, offline excluded from scans. Changing a client's vault once it holds identities, offline
notifications, pending requests or bootstrap state requires a separate state store. notifications, pending requests or an incorporated event cursor requires a separate state store.
Filesystem application saves metadata once when it finishes. Namespace changes detected Filesystem application saves metadata once when it finishes. Namespace changes detected
before mutations invalidate the plan. Changes during application are incorporated before mutations invalidate the plan. Changes during application are incorporated

View file

@ -3,3 +3,5 @@ export const MAX_LOG_MESSAGE_COUNT = 100000;
export const MAX_HISTORY_ENTRY_COUNT = 5000; export const MAX_HISTORY_ENTRY_COUNT = 5000;
export const SUPPORTED_API_VERSION = 4; export const SUPPORTED_API_VERSION = 4;
export const WEBSOCKET_DISCONNECT_TIMEOUT_IN_S = 10; export const WEBSOCKET_DISCONNECT_TIMEOUT_IN_S = 10;
export const CURSOR_HEARTBEAT_INTERVAL_MS = 15_000;
export const WEBSOCKET_RECEIVE_TIMEOUT_MS = CURSOR_HEARTBEAT_INTERVAL_MS * 3;

View file

@ -1,20 +1,15 @@
import { reconcile } from "reconcile-text"; import { reconcile } from "reconcile-text";
import type { import type { FileSystemOperations } from "./filesystem-operations";
FileSnapshot, import type { Database, EngineState } from "../persistence/database";
FileSystemOperations
} from "./filesystem-operations";
import type {
Database,
EngineState,
StoredSnapshot
} from "../persistence/database";
import { isInternalPath } from "../utils/portable-path";
import { FilePaths } from "./file-paths";
import { import {
fromStoredSnapshot, fromStoredSnapshot,
toStoredSnapshot, toStoredSnapshot,
mergeContent type FileSnapshot,
} from "../sync-operations/content"; type StoredSnapshot
} from "../snapshot";
import { isInternalPath } from "../utils/portable-path";
import { FilePaths } from "./file-paths";
import { mergeContent } from "../sync-operations/content";
import { hash } from "../utils/hash"; import { hash } from "../utils/hash";
import { isBinary } from "../utils/is-binary"; import { isBinary } from "../utils/is-binary";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable"; import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
@ -89,7 +84,7 @@ export class FileOperations {
): Promise<void> { ): Promise<void> {
const state = structuredClone(this.applying ?? this.database.state); const state = structuredClone(this.applying ?? this.database.state);
if (!state.initialized) return; if (state.lastSeenUpdateId === undefined) return;
for (const pending of unappliedChanges(state, previous)) for (const pending of unappliedChanges(state, previous))
applyLocalChange(state, pending, () => false); applyLocalChange(state, pending, () => false);
@ -118,7 +113,7 @@ export class FileOperations {
}; };
guard(); guard();
const current = structuredClone(this.database.state); const current = structuredClone(this.database.state);
current.initialized = next.initialized; current.lastSeenUpdateId = next.lastSeenUpdateId;
this.applying = current; this.applying = current;
const paths = new Set<string>(); const paths = new Set<string>();
const observeChanges = async (): Promise<void> => { const observeChanges = async (): Promise<void> => {
@ -145,7 +140,7 @@ export class FileOperations {
...Object.keys(next.local) ...Object.keys(next.local)
])) { ])) {
await observeChanges(); await observeChanges();
let from = current.documents[id]?.materialized let from = current.documents[id]?.observedHash !== undefined
? current.local[id] ? current.local[id]
: undefined; : undefined;
let to = next.local[id]; let to = next.local[id];
@ -166,7 +161,6 @@ export class FileOperations {
); );
(next.excluded ??= {})[id] = (next.excluded ??= {})[id] =
current.fileManifest.entries[id] ?? null; current.fileManifest.entries[id] ?? null;
(next.protectedPaths ??= []).push(from);
continue; continue;
} }
} }
@ -187,7 +181,7 @@ export class FileOperations {
mutate mutate
); );
next.local[id] = to; next.local[id] = to;
from = current.documents[id]?.materialized from = current.documents[id]?.observedHash !== undefined
? current.local[id] ? current.local[id]
: undefined; : undefined;
if (from !== undefined) { if (from !== undefined) {
@ -211,7 +205,7 @@ export class FileOperations {
.mergeableFileExtensions; .mergeableFileExtensions;
await mutate(async () => { await mutate(async () => {
current.local[id] = to; current.local[id] = to;
current.documents[id] = { materialized: true }; current.documents[id] = { observedHash: "" };
next.documents[id].observedHash = next.documents[id].observedHash =
await this.updateContent( await this.updateContent(
to, to,
@ -222,7 +216,6 @@ export class FileOperations {
}); });
} }
paths.add(to); paths.add(to);
next.documents[id].materialized = true;
current.local[id] = to; current.local[id] = to;
current.documents[id] = structuredClone(next.documents[id]); current.documents[id] = structuredClone(next.documents[id]);
} }

View file

@ -29,13 +29,12 @@ export class FilePaths {
for (const child of children) { for (const child of children) {
const occupant = Object.keys(current.local).find( const occupant = Object.keys(current.local).find(
(key) => (key) =>
current.documents[key]?.materialized && current.documents[key]?.observedHash !== undefined &&
current.local[key] === child current.local[key] === child
); );
const info = await this.fs.stat(child); const info = await this.fs.stat(child);
if ( if (
!info || !info ||
next.protectedPaths?.includes(child) ||
this.isProtectedFile(child, info.size) || this.isProtectedFile(child, info.size) ||
(obstruction.kind === "directory" && (obstruction.kind === "directory" &&
(occupant === undefined || (occupant === undefined ||
@ -59,8 +58,8 @@ export class FilePaths {
await this.fs.rename(occupant.path, displaced); await this.fs.rename(occupant.path, displaced);
}); });
if (!current.documents[occupant.id]) { if (!current.documents[occupant.id]) {
current.documents[occupant.id] = { materialized: true }; current.documents[occupant.id] = { observedHash: "" };
next.documents[occupant.id] = { materialized: true }; next.documents[occupant.id] = { observedHash: "" };
next.local[occupant.id] = displaced; next.local[occupant.id] = displaced;
} else if (next.local[occupant.id] === occupant.path) } else if (next.local[occupant.id] === occupant.path)
next.local[occupant.id] = displaced; next.local[occupant.id] = displaced;

View file

@ -1,11 +1,6 @@
import type { CursorPosition, TextWithCursors } from "reconcile-text"; import type { TextWithCursors } from "reconcile-text";
import type { RelativePath } from "../persistence/database"; import type { RelativePath } from "../persistence/database";
import type { FileSnapshot } from "../snapshot";
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. /** Filesystem operations act on the current user-owned namespace.
* Paths are vault-relative. Never follow symlinks (including ancestor symlinks). * Paths are vault-relative. Never follow symlinks (including ancestor symlinks).

View file

@ -22,10 +22,8 @@ export { Logger, LogLevel, LogLine } from "./tracing/logger";
export { type SyncSettings, DEFAULT_SETTINGS } from "./persistence/settings"; export { type SyncSettings, DEFAULT_SETTINGS } from "./persistence/settings";
export { rateLimit } from "./utils/rate-limit"; export { rateLimit } from "./utils/rate-limit";
export type { RelativePath, StoredDatabase } from "./persistence/database"; export type { RelativePath, StoredDatabase } from "./persistence/database";
export type { export type { FileSystemOperations } from "./file-operations/filesystem-operations";
FileSystemOperations, export type { FileSnapshot } from "./snapshot";
FileSnapshot
} from "./file-operations/filesystem-operations";
export type { MetadataPersistenceProvider as PersistenceProvider } from "./persistence/metadata-persistence-provider"; export type { MetadataPersistenceProvider as PersistenceProvider } from "./persistence/metadata-persistence-provider";
export type { CursorSpan } from "./services/types/CursorSpan"; export type { CursorSpan } from "./services/types/CursorSpan";
export type { ClientCursors } from "./services/types/ClientCursors"; export type { ClientCursors } from "./services/types/ClientCursors";

View file

@ -0,0 +1,94 @@
import assert from "node:assert/strict";
import { test } from "node:test";
import { Database, emptyState, type StoredDatabase } from "./database";
test("old state loads without duplicate flags, full recovery content, or acknowledged heads", async () => {
const head = {
documentId: "clean",
vaultUpdateId: 1,
contentSize: 3,
updatedDate: "2026-09-23",
userId: "user",
deviceId: "device"
};
const legacy = {
...emptyState("vault"),
initialized: true,
lastSeenUpdateId: 2,
protectedPaths: ["ignored.md"],
local: {
clean: "clean.md",
recovering: "recovering.md",
deferred: "deferred.md"
},
documents: {
clean: {
materialized: true,
observedHash: "clean",
base: { ...head, hash: "clean" }
},
recovering: {
materialized: true,
observedHash: "edited",
recoveryBase: { contentBase64: "T0xE", hash: "old" }
},
deferred: { materialized: false, observedHash: undefined }
},
remoteHeads: {
clean: head,
deferred: { ...head, documentId: "deferred" }
}
};
const saves: StoredDatabase[] = [];
const db = new Database(
legacy,
async (next) => {
saves.push(next);
},
"vault",
async () => saves.at(-1)
);
await db.save();
const saved = saves.at(-1);
assert(saved);
assert.deepEqual(saved.documents, {
clean: {
base: { vaultUpdateId: 1, hash: "clean" },
observedHash: "clean"
},
recovering: {
base: { vaultUpdateId: null, hash: "old" },
observedHash: "edited"
},
deferred: {
base: undefined,
observedHash: undefined,
remote: { ...head, documentId: "deferred" }
}
});
assert.equal(saved.lastSeenUpdateId, 2);
for (const field of ["initialized", "remoteHeads", "protectedPaths"])
assert(!(field in saved));
assert(!JSON.stringify(saved).includes("contentBase64"));
});
test("old reset intent leaves the cursor absent after an uncertain save reload", async () => {
const legacy = {
...emptyState("vault"),
initialized: false,
historyRecovery: true,
lastSeenUpdateId: 0
};
const db = new Database(
undefined,
async () => {
throw new Error("interrupted");
},
"vault",
async () => legacy
);
await assert.rejects(db.save(), /interrupted/);
await db.recoverPersistence();
assert.equal(db.state.lastSeenUpdateId, undefined);
assert(!("historyRecovery" in db.state));
});

View file

@ -1,38 +1,29 @@
import type { CursorPosition } from "reconcile-text"; import { upgradeState } from "./upgrade-state";
import type { StoredSnapshot } from "../snapshot";
import type { DocumentUpdateResponse } from "../services/types/DocumentUpdateResponse"; import type { DocumentUpdateResponse } from "../services/types/DocumentUpdateResponse";
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent"; import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
import type { FileManifest } from "../services/types/FileManifest"; import type { FileManifest } from "../services/types/FileManifest";
import type { FileManifestUpdateResponse } from "../services/types/FileManifestUpdateResponse"; import type { FileManifestUpdateResponse } from "../services/types/FileManifestUpdateResponse";
import type { PushFileManifest } from "../services/types/PushFileManifest"; import type { PushFileManifest } from "../services/types/PushFileManifest";
import type { PutFileContent } from "../services/types/PutFileContent"; import type { PutFileContent } from "../services/types/PutFileContent";
import type { VaultSnapshot } from "../services/types/VaultSnapshot";
import type { FileManifestEntries } from "../types/file-manifest-entries"; import type { FileManifestEntries } from "../types/file-manifest-entries";
export type VaultUpdateId = number; export type VaultUpdateId = number;
export type DocumentId = string; export type DocumentId = string;
export type RelativePath = string; export type RelativePath = string;
export interface StoredSnapshot {
contentBase64: string;
hash: string;
cursors?: CursorPosition[];
}
export interface DocumentMetadata { export interface DocumentMetadata {
parentVersionId: number; parentVersionId: number;
hash: string; hash: string;
} }
export interface DocumentState { export interface DocumentState {
/** Local clean base retained while rebasing after a server restore. */ /** A null version retains only the clean hash after a server history reset. */
recoveryBase?: StoredSnapshot; base?: { vaultUpdateId: number | null; hash: string };
base?: DocumentVersionWithoutContent & { hash: string }; /** Last observed disk hash; an empty string means a notification awaits scanning. */
observedHash?: string; observedHash?: string;
materialized: boolean; /** A remote version still to be incorporated, including excluded downloads. */
/** First contact with existing local content: there is no common ancestor. */ remote?: DocumentVersionWithoutContent;
bootstrap?: boolean;
/** Do not resubmit an unchanged payload rejected by the server. */
rejected?: { hash: string; message: string };
} }
export type PendingRequest = { rejection?: string } & ( export type PendingRequest = { rejection?: string } & (
@ -51,26 +42,19 @@ export type PendingRequest = { rejection?: string } & (
); );
export interface EngineState { export interface EngineState {
/** Persisted reset intent; clear the transport checkpoint before bootstrap. */
historyRecovery?: boolean;
vaultKey: string; vaultKey: string;
initialized: boolean;
fileManifest: FileManifest; fileManifest: FileManifest;
local: FileManifestEntries; local: FileManifestEntries;
documents: Record<string, DocumentState>; documents: Record<string, DocumentState>;
lastSeenUpdateId: number; /** Absent until a vault snapshot is applied; also marks a durable history reset. */
remoteHeads: Record<string, DocumentVersionWithoutContent>; lastSeenUpdateId?: number;
bootstrap?: VaultSnapshot;
pending?: PendingRequest; pending?: PendingRequest;
/** A rejected retry does not prove an earlier attempt failed to commit. */ /** A rejected retry does not prove an earlier attempt failed to commit. */
unconfirmed?: PendingRequest[]; unconfirmed?: PendingRequest[];
/** Commits consumption of the saved notification queue with the identity map. */ /** Commits consumption of the saved notification queue with the identity map. */
lastAppliedLocalChangeId?: string; lastAppliedLocalChangeId?: string;
/** Physical exclusions, including files with no sync identity. */
protectedPaths?: string[];
/** Original server paths while local files are excluded; null means absent. */ /** Original server paths while local files are excluded; null means absent. */
excluded?: Record<string, string | null>; excluded?: Record<string, string | null>;
rejectedManifest?: { entries: FileManifestEntries; message: string };
} }
export type StoredDatabase = EngineState; export type StoredDatabase = EngineState;
@ -85,21 +69,17 @@ export interface DocumentRecord {
export function emptyState(vaultKey: string): StoredDatabase { export function emptyState(vaultKey: string): StoredDatabase {
return { return {
vaultKey, vaultKey,
initialized: false,
fileManifest: { fileManifestId: 0, entries: {} }, fileManifest: { fileManifestId: 0, entries: {} },
local: {}, local: {},
documents: {}, documents: {}
remoteHeads: {},
lastSeenUpdateId: 0
}; };
} }
function canRebindEmpty(state: Partial<StoredDatabase> | undefined): boolean { function canRebindEmpty(state: Partial<StoredDatabase> | undefined): boolean {
return ( return (
state !== undefined && state !== undefined &&
!state.initialized && state.lastSeenUpdateId === undefined &&
!state.pending && !state.pending &&
!state.bootstrap &&
(state.unconfirmed?.length ?? 0) === 0 && (state.unconfirmed?.length ?? 0) === 0 &&
Object.keys(state.local ?? {}).length === 0 Object.keys(state.local ?? {}).length === 0
); );
@ -114,19 +94,16 @@ export class Database {
vaultKey: string, vaultKey: string,
private readonly loadData: () => Promise<StoredDatabase | undefined> private readonly loadData: () => Promise<StoredDatabase | undefined>
) { ) {
const hasState = initial && Object.keys(initial).length > 0; const state =
if ( initial !== undefined && Object.keys(initial).length > 0
hasState && ? upgradeState(initial as StoredDatabase)
initial.vaultKey !== vaultKey && : emptyState(vaultKey);
!canRebindEmpty(initial) if (state.vaultKey !== vaultKey && !canRebindEmpty(state)) {
) {
throw new Error( throw new Error(
"Incompatible vault identity. Preserve it for recovery and initialize a separate state store." "Incompatible vault identity. Preserve it for recovery and initialize a separate state store."
); );
} }
this.state = hasState this.state = { ...state, vaultKey };
? { ...(initial as StoredDatabase), vaultKey }
: emptyState(vaultKey);
} }
/** Bind an empty database to the configured vault. */ /** Bind an empty database to the configured vault. */
@ -154,7 +131,7 @@ export class Database {
public async recoverPersistence(): Promise<void> { public async recoverPersistence(): Promise<void> {
if (!this.needsReload) return; if (!this.needsReload) return;
const saved = await this.loadData(); const saved = await this.loadData();
if (saved) this.state = saved; if (saved) this.state = upgradeState(saved);
this.needsReload = false; this.needsReload = false;
} }
@ -173,10 +150,13 @@ export class Database {
return { return {
documentId: id, documentId: id,
relativePath, relativePath,
metadata: base && { metadata:
parentVersionId: base.vaultUpdateId, base?.vaultUpdateId != null
hash: base.hash ? {
} parentVersionId: base.vaultUpdateId,
hash: base.hash
}
: undefined
}; };
} }

View file

@ -0,0 +1,57 @@
import type { DocumentState, EngineState } from "./database";
import type { StoredSnapshot } from "../snapshot";
import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent";
/** Read the previous redundant state once; all subsequent saves use the smaller format. */
export function upgradeState(state: EngineState): EngineState {
if (!("initialized" in state)) return state;
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Read the previous persisted schema at the migration boundary.
const legacy = state as EngineState & {
initialized: boolean;
historyRecovery?: boolean;
remoteHeads?: Record<string, DocumentVersionWithoutContent>;
documents: Record<
string,
DocumentState & {
materialized?: boolean;
bootstrap?: boolean;
recoveryBase?: StoredSnapshot;
}
>;
};
const next: EngineState = {
vaultKey: state.vaultKey,
fileManifest: state.fileManifest,
local: state.local,
documents: {},
lastSeenUpdateId:
legacy.initialized && legacy.historyRecovery !== true
? state.lastSeenUpdateId
: undefined,
pending: state.pending,
unconfirmed: state.unconfirmed,
lastAppliedLocalChangeId: state.lastAppliedLocalChangeId,
excluded: state.excluded
};
for (const [id, doc] of Object.entries(legacy.documents)) {
const base = doc.recoveryBase
? { vaultUpdateId: null, hash: doc.recoveryBase.hash }
: doc.base && doc.bootstrap !== true
? { vaultUpdateId: doc.base.vaultUpdateId, hash: doc.base.hash }
: undefined;
next.documents[id] = {
base,
observedHash:
doc.materialized === true ? (doc.observedHash ?? "") : undefined
};
}
for (const [id, remote] of Object.entries(legacy.remoteHeads ?? {})) {
const doc = (next.documents[id] ??= {});
if (
doc.observedHash === undefined ||
remote.vaultUpdateId > (doc.base?.vaultUpdateId ?? 0)
)
doc.remote = remote;
}
return next;
}

View file

@ -3,6 +3,7 @@ import { describe, it, beforeEach } from "node:test";
import assert from "node:assert"; import assert from "node:assert";
import { createServer } from "node:net"; import { createServer } from "node:net";
import { WebSocketManager } from "./websocket-manager"; import { WebSocketManager } from "./websocket-manager";
import { WEBSOCKET_RECEIVE_TIMEOUT_MS } from "../consts";
import { Logger } from "../tracing/logger"; import { Logger } from "../tracing/logger";
import { Settings } from "../persistence/settings"; import { Settings } from "../persistence/settings";
// eslint-disable-next-line @typescript-eslint/no-require-imports // eslint-disable-next-line @typescript-eslint/no-require-imports
@ -502,3 +503,96 @@ it("a handshake send failure retires the connection without escaping the open ca
await manager.stop(); await manager.stop();
} }
}); });
it("reconnects a connection that stops delivering messages", async (context) => {
context.mock.timers.enable({ apis: ["setTimeout"] });
let attempts = 0;
class CountingSocket extends MockWebSocket {
static OPEN = 1;
constructor(url: string) {
super(url);
attempts++;
}
}
const logger = new Logger();
const settings = new Settings(
logger,
{ remoteUri: "http://test", webSocketRetryIntervalMs: 1000 },
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
CountingSocket as unknown as typeof WebSocket
);
const statuses: boolean[] = [];
manager.onWebSocketStatusChanged.add((status) => statuses.push(status));
try {
manager.start();
context.mock.timers.tick(0);
assert.equal(manager.isWebSocketConnected, true);
assert.equal(attempts, 1);
const silentSocket = (
manager as unknown as { webSocket: MockWebSocket }
).webSocket;
context.mock.timers.tick(WEBSOCKET_RECEIVE_TIMEOUT_MS);
assert.equal(manager.isWebSocketConnected, false);
assert.equal(statuses.at(-1), false);
assert.equal(silentSocket.onmessage, null);
context.mock.timers.tick(1000);
context.mock.timers.tick(0);
assert.equal(attempts, 2);
assert.equal(manager.isWebSocketConnected, true);
} finally {
await manager.stop();
}
});
it("does not reconnect while messages keep arriving before the deadline", async (context) => {
context.mock.timers.enable({ apis: ["setTimeout"] });
let attempts = 0;
class CountingSocket extends MockWebSocket {
static OPEN = 1;
constructor(url: string) {
super(url);
attempts++;
}
}
const logger = new Logger();
const settings = new Settings(
logger,
{ remoteUri: "http://test", webSocketRetryIntervalMs: 1000 },
async () => {}
);
const manager = new WebSocketManager(
"test",
logger,
settings,
CountingSocket as unknown as typeof WebSocket
);
let disconnected = 0;
manager.onWebSocketStatusChanged.add((status) => {
if (!status) disconnected++;
});
try {
manager.start();
context.mock.timers.tick(0);
const socket = (manager as unknown as { webSocket: MockWebSocket })
.webSocket;
assert.equal(attempts, 1);
context.mock.timers.tick(10_000);
socket.simulateMessage({ type: "vaultChanged" });
context.mock.timers.tick(10_000);
socket.simulateMessage({ type: "cursorPositions", clients: [] });
context.mock.timers.tick(WEBSOCKET_RECEIVE_TIMEOUT_MS - 20_000);
assert.equal(manager.isWebSocketConnected, true);
assert.equal(disconnected, 0);
assert.equal(attempts, 1);
} finally {
await manager.stop();
}
});

View file

@ -5,7 +5,7 @@ import type { WebSocketClientMessage } from "./types/WebSocketClientMessage";
import type { CursorPositionFromClient } from "./types/CursorPositionFromClient"; import type { CursorPositionFromClient } from "./types/CursorPositionFromClient";
import type { ClientCursors } from "./types/ClientCursors"; import type { ClientCursors } from "./types/ClientCursors";
import { createPromise } from "../utils/create-promise"; import { createPromise } from "../utils/create-promise";
import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts"; import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S, WEBSOCKET_RECEIVE_TIMEOUT_MS } from "../consts";
import { removeFromArray } from "../utils/remove-from-array"; import { removeFromArray } from "../utils/remove-from-array";
import { EventListeners } from "../utils/data-structures/event-listeners"; import { EventListeners } from "../utils/data-structures/event-listeners";
import { awaitAll } from "../utils/await-all"; import { awaitAll } from "../utils/await-all";
@ -177,13 +177,15 @@ export class WebSocketManager {
private expectResponse(socket: WebSocket): void { private expectResponse(socket: WebSocket): void {
clearTimeout(this.receiveTimeoutId); clearTimeout(this.receiveTimeoutId);
// Cursor heartbeats are acknowledged every 15 seconds, including when // Cursor heartbeats are acknowledged every CURSOR_HEARTBEAT_INTERVAL_MS,
// there are no open editors. An OPEN TCP socket alone proves nothing. // including when there are no open editors. An OPEN TCP socket alone
// proves nothing: silence past the heartbeat deadline means a
// half-open connection.
this.receiveTimeoutId = setTimeout(() => { this.receiveTimeoutId = setTimeout(() => {
if (this.webSocket !== socket || this.isStopped) return; if (this.webSocket !== socket || this.isStopped) return;
this.logger.warn("WebSocket receive deadline expired"); this.logger.warn("WebSocket receive deadline expired");
this.reconnect(socket); this.reconnect(socket);
}, 45_000); }, WEBSOCKET_RECEIVE_TIMEOUT_MS);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Browser timers have no unref method. // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Browser timers have no unref method.
this.receiveTimeoutId.unref?.(); this.receiveTimeoutId.unref?.();
} }

View file

@ -0,0 +1,32 @@
import { base64ToBytes, bytesToBase64 } from "byte-base64";
import type { CursorPosition } from "reconcile-text";
import { hash } from "./utils/hash";
export interface FileSnapshot {
content: Uint8Array;
/** Optional editor selections, moved with the file and applied with its bytes. */
cursors?: CursorPosition[];
}
export interface StoredSnapshot {
contentBase64: string;
hash: string;
cursors?: CursorPosition[];
}
export async function toStoredSnapshot(
snapshot: FileSnapshot
): Promise<StoredSnapshot> {
return {
contentBase64: bytesToBase64(snapshot.content),
cursors: snapshot.cursors,
hash: await hash(snapshot.content)
};
}
export function fromStoredSnapshot(snapshot: StoredSnapshot): FileSnapshot {
return {
content: base64ToBytes(snapshot.contentBase64),
cursors: snapshot.cursors
};
}

View file

@ -320,8 +320,7 @@ export class SyncClient {
if ( if (
vaultKey(next) !== vaultKey(old) && vaultKey(next) !== vaultKey(old) &&
(this.started || (this.started ||
this.database.state.initialized || this.database.state.lastSeenUpdateId !== undefined ||
this.database.state.bootstrap ||
this.database.state.pending || this.database.state.pending ||
this.syncer.hasLocalChanges || this.syncer.hasLocalChanges ||
Object.keys(this.database.state.local).length) Object.keys(this.database.state.local).length)

View file

@ -1,6 +1,7 @@
import assert from "node:assert/strict"; import assert from "node:assert/strict";
import { describe, it } from "node:test"; import { describe, it } from "node:test";
import { mergeContent, fromStoredSnapshot, toStoredSnapshot } from "./content"; import { fromStoredSnapshot, toStoredSnapshot } from "../snapshot";
import { mergeContent } from "./content";
const extensions = ["md"]; const extensions = ["md"];
const snapshot = async (text: string) => const snapshot = async (text: string) =>

View file

@ -1,28 +1,9 @@
import { base64ToBytes, bytesToBase64 } from "byte-base64"; import { base64ToBytes } from "byte-base64";
import { reconcile } from "reconcile-text"; import { reconcile } from "reconcile-text";
import type { FileSnapshot } from "../file-operations/filesystem-operations"; import { toStoredSnapshot, type StoredSnapshot } from "../snapshot";
import type { StoredSnapshot } from "../persistence/database";
import { hash } from "../utils/hash";
import { isBinary } from "../utils/is-binary"; import { isBinary } from "../utils/is-binary";
import { isFileTypeMergeable } from "../utils/is-file-type-mergeable"; import { isFileTypeMergeable } from "../utils/is-file-type-mergeable";
export async function toStoredSnapshot(
snapshot: FileSnapshot
): Promise<StoredSnapshot> {
return {
contentBase64: bytesToBase64(snapshot.content),
cursors: snapshot.cursors,
hash: await hash(snapshot.content)
};
}
export function fromStoredSnapshot(snapshot: StoredSnapshot): FileSnapshot {
return {
content: base64ToBytes(snapshot.contentBase64),
cursors: snapshot.cursors
};
}
export async function mergeContent( export async function mergeContent(
path: string, path: string,
base: StoredSnapshot | undefined, base: StoredSnapshot | undefined,

View file

@ -10,6 +10,7 @@ import { hash } from "../utils/hash";
import type { FileChangeNotifier } from "./file-change-notifier"; import type { FileChangeNotifier } from "./file-change-notifier";
import { Lock } from "../utils/data-structures/locks"; import { Lock } from "../utils/data-structures/locks";
import { EventListeners } from "../utils/data-structures/event-listeners"; import { EventListeners } from "../utils/data-structures/event-listeners";
import { CURSOR_HEARTBEAT_INTERVAL_MS } from "../consts";
// Cursor positions are updated separately from documents. However, a given cursor position is only // Cursor positions are updated separately from documents. However, a given cursor position is only
// valid within a certain version of the document it belongs to. This class tracks previous and the latest // valid within a certain version of the document it belongs to. This class tracks previous and the latest
@ -60,7 +61,10 @@ export class CursorTracker {
}); });
}; };
refresh(); refresh();
this.cursorTimer = setInterval(refresh, 15_000); this.cursorTimer = setInterval(
refresh,
CURSOR_HEARTBEAT_INTERVAL_MS
);
this.cursorTimer.unref?.(); this.cursorTimer.unref?.();
}); });
this.webSocketManager.onRemoteCursorsUpdateReceived.add( this.webSocketManager.onRemoteCursorsUpdateReceived.add(

View file

@ -34,7 +34,7 @@ export function resolvePaths(
const ids = Object.keys(proposed).sort( const ids = Object.keys(proposed).sort(
(a, b) => (a, b) =>
Number(remote[b] === proposed[b]) - Number(remote[b] === proposed[b]) -
Number(remote[a] === proposed[a]) || Number(remote[a] === proposed[a]) ||
(a < b ? -1 : a > b ? 1 : 0) (a < b ? -1 : a > b ? 1 : 0)
); );
for (const id of ids.filter( for (const id of ids.filter(
@ -57,12 +57,16 @@ export function mergeFileManifests(
...Object.keys(local), ...Object.keys(local),
...Object.keys(remote) ...Object.keys(remote)
])) { ])) {
const path = let path: string | undefined;
local[id] === base[id]
? remote[id] if (local[id] === base[id]) {
: remote[id] === base[id] path = remote[id];
? local[id] } else if (remote[id] === base[id]) {
: remote[id]; path = local[id];
} else {
path = remote[id];
}
if (path !== undefined) proposed[id] = path; if (path !== undefined) proposed[id] = path;
} }
return resolvePaths(proposed, remote); return resolvePaths(proposed, remote);

View file

@ -48,17 +48,14 @@ export function applyLocalChange(
)) || )) ||
uuid()); uuid());
next.local[id] = change.path; next.local[id] = change.path;
next.documents[id] ??= { next.documents[id] ??= { observedHash: "" };
materialized: true,
bootstrap: initial?.fileManifest.entries[id] !== undefined
};
return; return;
} }
const from = change.type === "move" ? change.oldPath : change.path; const from = change.type === "move" ? change.oldPath : change.path;
change.identities ??= Object.fromEntries( change.identities ??= Object.fromEntries(
Object.entries(next.local).filter( Object.entries(next.local).filter(
([id, path]) => ([id, path]) =>
next.documents[id]?.materialized && next.documents[id]?.observedHash !== undefined &&
(path === from || path.startsWith(from + "/")) (path === from || path.startsWith(from + "/"))
) )
); );

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