From b07aa589ff5e48e03a6a90b50f200933a5c53c13 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sat, 26 Sep 2026 20:22:16 +0100 Subject: [PATCH] . --- docs/architecture/sync-algorithm.md | 23 +- .../deterministic-tests/src/crash-worker.ts | 2 +- frontend/sync-client/README.md | 15 +- frontend/sync-client/src/consts.ts | 2 + .../src/file-operations/file-operations.ts | 33 +- .../src/file-operations/file-paths.ts | 7 +- .../file-operations/filesystem-operations.ts | 9 +- frontend/sync-client/src/index.ts | 6 +- .../src/persistence/database.test.ts | 94 +++ .../sync-client/src/persistence/database.ts | 70 +-- .../src/persistence/upgrade-state.ts | 57 ++ .../src/services/websocket-manager.test.ts | 94 +++ .../src/services/websocket-manager.ts | 10 +- frontend/sync-client/src/snapshot.ts | 32 ++ frontend/sync-client/src/sync-client.ts | 3 +- .../src/sync-operations/content.test.ts | 3 +- .../src/sync-operations/content.ts | 23 +- .../src/sync-operations/cursor-tracker.ts | 6 +- .../src/sync-operations/file-manifest.ts | 18 +- .../src/sync-operations/local-changes.ts | 7 +- .../sync-client/src/sync-operations/scan.ts | 25 +- .../src/sync-operations/syncer.test.ts | 7 +- .../sync-client/src/sync-operations/syncer.ts | 535 ++++++++---------- frontend/test-support/canonical.ts | 4 +- frontend/test-support/content-limits.test.ts | 4 +- .../test-support/engine-regressions.test.ts | 93 ++- frontend/test-support/event-delivery.test.ts | 2 +- .../test-support/file-application.test.ts | 2 +- .../test-support/oracle-regressions.test.ts | 6 +- frontend/test-support/polling.test.ts | 6 +- frontend/test-support/protocol.test.ts | 2 +- frontend/test-support/real-disk.test.ts | 4 +- frontend/test-support/recovery.test.ts | 10 +- frontend/test-support/sync-audit.test.ts | 35 +- frontend/test-support/sync-fixture.ts | 9 +- 35 files changed, 728 insertions(+), 530 deletions(-) create mode 100644 frontend/sync-client/src/persistence/database.test.ts create mode 100644 frontend/sync-client/src/persistence/upgrade-state.ts create mode 100644 frontend/sync-client/src/snapshot.ts diff --git a/docs/architecture/sync-algorithm.md b/docs/architecture/sync-algorithm.md index a58aa0a3..2df745a8 100644 --- a/docs/architecture/sync-algorithm.md +++ b/docs/architecture/sync-algorithm.md @@ -18,11 +18,22 @@ UUID is rejected. ## 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 heads and manifest before changing local files. Interrupted fetching restarts from 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 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 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 -server. Clean files reconcile against the restored snapshot. Unsent edits with an -old base are preserved as separate local documents, with conflict names if needed. +After a mismatch, the client clears the event cursor and old history-dependent +state before releasing the transport checkpoint, then bootstraps against the +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 stopped. diff --git a/frontend/deterministic-tests/src/crash-worker.ts b/frontend/deterministic-tests/src/crash-worker.ts index c015fac6..af3945dc 100644 --- a/frontend/deterministic-tests/src/crash-worker.ts +++ b/frontend/deterministic-tests/src/crash-worker.ts @@ -72,7 +72,7 @@ async function main() { assert(documentId, "Edit must target an established document"); editedBase = { documentId, - vaultUpdateId: state.documents![documentId].base!.vaultUpdateId + vaultUpdateId: state.documents![documentId].base!.vaultUpdateId! }; await disk.userWrite( input.editAfterSync.path, diff --git a/frontend/sync-client/README.md b/frontend/sync-client/README.md index b3849c67..40de1fdf 100644 --- a/frontend/sync-client/README.md +++ b/frontend/sync-client/README.md @@ -21,7 +21,14 @@ resuming starts a fresh session. Connection checks can run while sync is disable ## State and policy 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 submissions advance the base to the submitted content snapshot, preserving later 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 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 conflict names instead of applying a diff to an unrelated restored version. 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. 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 -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 before mutations invalidate the plan. Changes during application are incorporated diff --git a/frontend/sync-client/src/consts.ts b/frontend/sync-client/src/consts.ts index fd0642b8..8d266eb8 100644 --- a/frontend/sync-client/src/consts.ts +++ b/frontend/sync-client/src/consts.ts @@ -3,3 +3,5 @@ export const MAX_LOG_MESSAGE_COUNT = 100000; export const MAX_HISTORY_ENTRY_COUNT = 5000; export const SUPPORTED_API_VERSION = 4; 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; diff --git a/frontend/sync-client/src/file-operations/file-operations.ts b/frontend/sync-client/src/file-operations/file-operations.ts index 2e22e688..8625fde0 100644 --- a/frontend/sync-client/src/file-operations/file-operations.ts +++ b/frontend/sync-client/src/file-operations/file-operations.ts @@ -1,20 +1,15 @@ import { reconcile } from "reconcile-text"; -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 type { FileSystemOperations } from "./filesystem-operations"; +import type { Database, EngineState } from "../persistence/database"; import { fromStoredSnapshot, toStoredSnapshot, - mergeContent -} from "../sync-operations/content"; + type FileSnapshot, + 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 { isBinary } from "../utils/is-binary"; import { isFileTypeMergeable } from "../utils/is-file-type-mergeable"; @@ -89,7 +84,7 @@ export class FileOperations { ): Promise { const state = structuredClone(this.applying ?? this.database.state); - if (!state.initialized) return; + if (state.lastSeenUpdateId === undefined) return; for (const pending of unappliedChanges(state, previous)) applyLocalChange(state, pending, () => false); @@ -118,7 +113,7 @@ export class FileOperations { }; guard(); const current = structuredClone(this.database.state); - current.initialized = next.initialized; + current.lastSeenUpdateId = next.lastSeenUpdateId; this.applying = current; const paths = new Set(); const observeChanges = async (): Promise => { @@ -145,7 +140,7 @@ export class FileOperations { ...Object.keys(next.local) ])) { await observeChanges(); - let from = current.documents[id]?.materialized + let from = current.documents[id]?.observedHash !== undefined ? current.local[id] : undefined; let to = next.local[id]; @@ -166,7 +161,6 @@ export class FileOperations { ); (next.excluded ??= {})[id] = current.fileManifest.entries[id] ?? null; - (next.protectedPaths ??= []).push(from); continue; } } @@ -187,7 +181,7 @@ export class FileOperations { mutate ); next.local[id] = to; - from = current.documents[id]?.materialized + from = current.documents[id]?.observedHash !== undefined ? current.local[id] : undefined; if (from !== undefined) { @@ -211,7 +205,7 @@ export class FileOperations { .mergeableFileExtensions; await mutate(async () => { current.local[id] = to; - current.documents[id] = { materialized: true }; + current.documents[id] = { observedHash: "" }; next.documents[id].observedHash = await this.updateContent( to, @@ -222,7 +216,6 @@ export class FileOperations { }); } paths.add(to); - next.documents[id].materialized = true; current.local[id] = to; current.documents[id] = structuredClone(next.documents[id]); } diff --git a/frontend/sync-client/src/file-operations/file-paths.ts b/frontend/sync-client/src/file-operations/file-paths.ts index fe9908d5..09474403 100644 --- a/frontend/sync-client/src/file-operations/file-paths.ts +++ b/frontend/sync-client/src/file-operations/file-paths.ts @@ -29,13 +29,12 @@ export class FilePaths { for (const child of children) { const occupant = Object.keys(current.local).find( (key) => - current.documents[key]?.materialized && + current.documents[key]?.observedHash !== undefined && 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 || @@ -59,8 +58,8 @@ export class FilePaths { await this.fs.rename(occupant.path, displaced); }); if (!current.documents[occupant.id]) { - current.documents[occupant.id] = { materialized: true }; - next.documents[occupant.id] = { materialized: true }; + current.documents[occupant.id] = { observedHash: "" }; + next.documents[occupant.id] = { observedHash: "" }; next.local[occupant.id] = displaced; } else if (next.local[occupant.id] === occupant.path) next.local[occupant.id] = displaced; diff --git a/frontend/sync-client/src/file-operations/filesystem-operations.ts b/frontend/sync-client/src/file-operations/filesystem-operations.ts index 9187d67b..02dd7cec 100644 --- a/frontend/sync-client/src/file-operations/filesystem-operations.ts +++ b/frontend/sync-client/src/file-operations/filesystem-operations.ts @@ -1,11 +1,6 @@ -import type { CursorPosition, TextWithCursors } from "reconcile-text"; +import type { TextWithCursors } from "reconcile-text"; import type { RelativePath } from "../persistence/database"; - -export interface FileSnapshot { - content: Uint8Array; - /** Optional editor selections, moved with the file and applied with its bytes. */ - cursors?: CursorPosition[]; -} +import type { FileSnapshot } from "../snapshot"; /** Filesystem operations act on the current user-owned namespace. * Paths are vault-relative. Never follow symlinks (including ancestor symlinks). diff --git a/frontend/sync-client/src/index.ts b/frontend/sync-client/src/index.ts index 8d9fd7bd..ff901a7e 100644 --- a/frontend/sync-client/src/index.ts +++ b/frontend/sync-client/src/index.ts @@ -22,10 +22,8 @@ export { Logger, LogLevel, LogLine } from "./tracing/logger"; export { type SyncSettings, DEFAULT_SETTINGS } from "./persistence/settings"; export { rateLimit } from "./utils/rate-limit"; export type { RelativePath, StoredDatabase } from "./persistence/database"; -export type { - FileSystemOperations, - FileSnapshot -} from "./file-operations/filesystem-operations"; +export type { FileSystemOperations } from "./file-operations/filesystem-operations"; +export type { FileSnapshot } from "./snapshot"; export type { MetadataPersistenceProvider as PersistenceProvider } from "./persistence/metadata-persistence-provider"; export type { CursorSpan } from "./services/types/CursorSpan"; export type { ClientCursors } from "./services/types/ClientCursors"; diff --git a/frontend/sync-client/src/persistence/database.test.ts b/frontend/sync-client/src/persistence/database.test.ts new file mode 100644 index 00000000..d0cad230 --- /dev/null +++ b/frontend/sync-client/src/persistence/database.test.ts @@ -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)); +}); diff --git a/frontend/sync-client/src/persistence/database.ts b/frontend/sync-client/src/persistence/database.ts index ec47430d..217b31c0 100644 --- a/frontend/sync-client/src/persistence/database.ts +++ b/frontend/sync-client/src/persistence/database.ts @@ -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 { 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: number; hash: string; } export interface DocumentState { - /** Local clean base retained while rebasing after a server restore. */ - recoveryBase?: StoredSnapshot; - base?: DocumentVersionWithoutContent & { hash: string }; + /** A null version retains only the clean hash after a server history reset. */ + base?: { vaultUpdateId: number | null; hash: string }; + /** Last observed disk hash; an empty string means a notification awaits scanning. */ 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 }; + /** A remote version still to be incorporated, including excluded downloads. */ + remote?: DocumentVersionWithoutContent; } export type PendingRequest = { rejection?: string } & ( @@ -51,26 +42,19 @@ export type PendingRequest = { rejection?: string } & ( ); export interface EngineState { - /** Persisted reset intent; clear the transport checkpoint before bootstrap. */ - historyRecovery?: boolean; vaultKey: string; - initialized: boolean; fileManifest: FileManifest; local: FileManifestEntries; documents: Record; - lastSeenUpdateId: number; - remoteHeads: Record; - bootstrap?: VaultSnapshot; + /** Absent until a vault snapshot is applied; also marks a durable history reset. */ + lastSeenUpdateId?: number; 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; - rejectedManifest?: { entries: FileManifestEntries; message: string }; } export type StoredDatabase = EngineState; @@ -85,21 +69,17 @@ export interface DocumentRecord { export function emptyState(vaultKey: string): StoredDatabase { return { vaultKey, - initialized: false, fileManifest: { fileManifestId: 0, entries: {} }, local: {}, - documents: {}, - remoteHeads: {}, - lastSeenUpdateId: 0 + documents: {} }; } function canRebindEmpty(state: Partial | undefined): boolean { return ( state !== undefined && - !state.initialized && + state.lastSeenUpdateId === undefined && !state.pending && - !state.bootstrap && (state.unconfirmed?.length ?? 0) === 0 && Object.keys(state.local ?? {}).length === 0 ); @@ -114,19 +94,16 @@ export class Database { vaultKey: string, private readonly loadData: () => Promise ) { - const hasState = initial && Object.keys(initial).length > 0; - if ( - hasState && - initial.vaultKey !== vaultKey && - !canRebindEmpty(initial) - ) { + const state = + initial !== undefined && Object.keys(initial).length > 0 + ? upgradeState(initial as StoredDatabase) + : emptyState(vaultKey); + if (state.vaultKey !== vaultKey && !canRebindEmpty(state)) { throw new Error( "Incompatible vault identity. Preserve it for recovery and initialize a separate state store." ); } - this.state = hasState - ? { ...(initial as StoredDatabase), vaultKey } - : emptyState(vaultKey); + this.state = { ...state, vaultKey }; } /** Bind an empty database to the configured vault. */ @@ -154,7 +131,7 @@ export class Database { public async recoverPersistence(): Promise { if (!this.needsReload) return; const saved = await this.loadData(); - if (saved) this.state = saved; + if (saved) this.state = upgradeState(saved); this.needsReload = false; } @@ -173,10 +150,13 @@ export class Database { return { documentId: id, relativePath, - metadata: base && { - parentVersionId: base.vaultUpdateId, - hash: base.hash - } + metadata: + base?.vaultUpdateId != null + ? { + parentVersionId: base.vaultUpdateId, + hash: base.hash + } + : undefined }; } diff --git a/frontend/sync-client/src/persistence/upgrade-state.ts b/frontend/sync-client/src/persistence/upgrade-state.ts new file mode 100644 index 00000000..e1eb7f4f --- /dev/null +++ b/frontend/sync-client/src/persistence/upgrade-state.ts @@ -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; + 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; +} diff --git a/frontend/sync-client/src/services/websocket-manager.test.ts b/frontend/sync-client/src/services/websocket-manager.test.ts index 645fa627..4d2119cb 100644 --- a/frontend/sync-client/src/services/websocket-manager.test.ts +++ b/frontend/sync-client/src/services/websocket-manager.test.ts @@ -3,6 +3,7 @@ import { describe, it, beforeEach } from "node:test"; import assert from "node:assert"; import { createServer } from "node:net"; import { WebSocketManager } from "./websocket-manager"; +import { WEBSOCKET_RECEIVE_TIMEOUT_MS } from "../consts"; import { Logger } from "../tracing/logger"; import { Settings } from "../persistence/settings"; // 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(); } }); + +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(); + } +}); diff --git a/frontend/sync-client/src/services/websocket-manager.ts b/frontend/sync-client/src/services/websocket-manager.ts index 8d90fbbd..24107454 100644 --- a/frontend/sync-client/src/services/websocket-manager.ts +++ b/frontend/sync-client/src/services/websocket-manager.ts @@ -5,7 +5,7 @@ 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 { 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 { EventListeners } from "../utils/data-structures/event-listeners"; import { awaitAll } from "../utils/await-all"; @@ -177,13 +177,15 @@ 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. + // Cursor heartbeats are acknowledged every CURSOR_HEARTBEAT_INTERVAL_MS, + // 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(() => { if (this.webSocket !== socket || this.isStopped) return; this.logger.warn("WebSocket receive deadline expired"); this.reconnect(socket); - }, 45_000); + }, WEBSOCKET_RECEIVE_TIMEOUT_MS); // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Browser timers have no unref method. this.receiveTimeoutId.unref?.(); } diff --git a/frontend/sync-client/src/snapshot.ts b/frontend/sync-client/src/snapshot.ts new file mode 100644 index 00000000..9a44a903 --- /dev/null +++ b/frontend/sync-client/src/snapshot.ts @@ -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 { + 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 + }; +} diff --git a/frontend/sync-client/src/sync-client.ts b/frontend/sync-client/src/sync-client.ts index 6eabdb06..41c7e5da 100644 --- a/frontend/sync-client/src/sync-client.ts +++ b/frontend/sync-client/src/sync-client.ts @@ -320,8 +320,7 @@ export class SyncClient { if ( vaultKey(next) !== vaultKey(old) && (this.started || - this.database.state.initialized || - this.database.state.bootstrap || + this.database.state.lastSeenUpdateId !== undefined || this.database.state.pending || this.syncer.hasLocalChanges || Object.keys(this.database.state.local).length) diff --git a/frontend/sync-client/src/sync-operations/content.test.ts b/frontend/sync-client/src/sync-operations/content.test.ts index db446097..e33c935e 100644 --- a/frontend/sync-client/src/sync-operations/content.test.ts +++ b/frontend/sync-client/src/sync-operations/content.test.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; 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 snapshot = async (text: string) => diff --git a/frontend/sync-client/src/sync-operations/content.ts b/frontend/sync-client/src/sync-operations/content.ts index 4f305ddb..756615fa 100644 --- a/frontend/sync-client/src/sync-operations/content.ts +++ b/frontend/sync-client/src/sync-operations/content.ts @@ -1,28 +1,9 @@ -import { base64ToBytes, bytesToBase64 } from "byte-base64"; +import { base64ToBytes } from "byte-base64"; import { reconcile } from "reconcile-text"; -import type { FileSnapshot } from "../file-operations/filesystem-operations"; -import type { StoredSnapshot } from "../persistence/database"; -import { hash } from "../utils/hash"; +import { toStoredSnapshot, type StoredSnapshot } from "../snapshot"; import { isBinary } from "../utils/is-binary"; import { isFileTypeMergeable } from "../utils/is-file-type-mergeable"; -export async function toStoredSnapshot( - snapshot: FileSnapshot -): Promise { - 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( path: string, base: StoredSnapshot | undefined, diff --git a/frontend/sync-client/src/sync-operations/cursor-tracker.ts b/frontend/sync-client/src/sync-operations/cursor-tracker.ts index a6170195..02399c2a 100644 --- a/frontend/sync-client/src/sync-operations/cursor-tracker.ts +++ b/frontend/sync-client/src/sync-operations/cursor-tracker.ts @@ -10,6 +10,7 @@ import { hash } from "../utils/hash"; import type { FileChangeNotifier } from "./file-change-notifier"; import { Lock } from "../utils/data-structures/locks"; import { EventListeners } from "../utils/data-structures/event-listeners"; +import { CURSOR_HEARTBEAT_INTERVAL_MS } from "../consts"; // Cursor positions are updated separately from documents. However, a given cursor position is only // valid within a certain version of the document it belongs to. This class tracks previous and the latest @@ -60,7 +61,10 @@ export class CursorTracker { }); }; refresh(); - this.cursorTimer = setInterval(refresh, 15_000); + this.cursorTimer = setInterval( + refresh, + CURSOR_HEARTBEAT_INTERVAL_MS + ); this.cursorTimer.unref?.(); }); this.webSocketManager.onRemoteCursorsUpdateReceived.add( diff --git a/frontend/sync-client/src/sync-operations/file-manifest.ts b/frontend/sync-client/src/sync-operations/file-manifest.ts index ad2d393a..fa4cc1b5 100644 --- a/frontend/sync-client/src/sync-operations/file-manifest.ts +++ b/frontend/sync-client/src/sync-operations/file-manifest.ts @@ -34,7 +34,7 @@ export function resolvePaths( const ids = Object.keys(proposed).sort( (a, b) => Number(remote[b] === proposed[b]) - - Number(remote[a] === proposed[a]) || + Number(remote[a] === proposed[a]) || (a < b ? -1 : a > b ? 1 : 0) ); for (const id of ids.filter( @@ -57,12 +57,16 @@ export function mergeFileManifests( ...Object.keys(local), ...Object.keys(remote) ])) { - const path = - local[id] === base[id] - ? remote[id] - : remote[id] === base[id] - ? local[id] - : remote[id]; + let path: string | undefined; + + if (local[id] === base[id]) { + path = remote[id]; + } else if (remote[id] === base[id]) { + path = local[id]; + } else { + path = remote[id]; + } + if (path !== undefined) proposed[id] = path; } return resolvePaths(proposed, remote); diff --git a/frontend/sync-client/src/sync-operations/local-changes.ts b/frontend/sync-client/src/sync-operations/local-changes.ts index 209c6853..641e7061 100644 --- a/frontend/sync-client/src/sync-operations/local-changes.ts +++ b/frontend/sync-client/src/sync-operations/local-changes.ts @@ -48,17 +48,14 @@ export function applyLocalChange( )) || uuid()); next.local[id] = change.path; - next.documents[id] ??= { - materialized: true, - bootstrap: initial?.fileManifest.entries[id] !== undefined - }; + next.documents[id] ??= { observedHash: "" }; 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 && + next.documents[id]?.observedHash !== undefined && (path === from || path.startsWith(from + "/")) ) ); diff --git a/frontend/sync-client/src/sync-operations/scan.ts b/frontend/sync-client/src/sync-operations/scan.ts index 5dbb29cf..0ded18ed 100644 --- a/frontend/sync-client/src/sync-operations/scan.ts +++ b/frontend/sync-client/src/sync-operations/scan.ts @@ -85,7 +85,7 @@ export async function scanLocalFiles( const missing = new Set( Object.keys(next.local).filter( (id) => - next.documents[id]?.materialized && + next.documents[id]?.observedHash !== undefined && !ignored(next.local[id]) && !present.has(next.local[id]) ) @@ -122,10 +122,7 @@ export async function scanLocalFiles( } id ??= uuid(); next.local[id] = path; - next.documents[id] ??= { - materialized: true, - bootstrap: initial?.fileManifest.entries[id] !== undefined - }; + next.documents[id] ??= { observedHash: "" }; } for (const id of missing) delete next.local[id]; for (const [id, path] of Object.entries(next.local)) { @@ -136,14 +133,20 @@ export async function scanLocalFiles( } 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; + const doc = (next.documents[id] ??= {}); + // The old history no longer contains this base. Preserve any edit + // made during recovery as a new file instead of overwriting it. + if ( + doc.base?.vaultUpdateId === null && + doc.base.hash !== snapshot.hash + ) { + const recoveredId = uuid(); + next.local[recoveredId] = path; + next.documents[recoveredId] = { observedHash: snapshot.hash }; + delete next.local[id]; + } else doc.observedHash = snapshot.hash; } } - next.protectedPaths = [...unsyncablePaths]; guard(); await commit(next); queuedChanges.splice(0, changes.length); diff --git a/frontend/sync-client/src/sync-operations/syncer.test.ts b/frontend/sync-client/src/sync-operations/syncer.test.ts index 524fbabb..c4c3b289 100644 --- a/frontend/sync-client/src/sync-operations/syncer.test.ts +++ b/frontend/sync-client/src/sync-operations/syncer.test.ts @@ -16,7 +16,7 @@ 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 { toStoredSnapshot } from "../snapshot"; import { undiff } from "reconcile-text"; function createSyncer(overrides: Partial = {}): { @@ -131,11 +131,11 @@ describe("Syncer per-file state and event delivery", () => { function fixture() { const database = { state: { - initialized: true, + local: { document: "note.md" }, documents: { document: { - materialized: true, + observedHash: "hash", base: { ...head, hash: "hash" } } @@ -144,7 +144,6 @@ describe("Syncer per-file state and event delivery", () => { fileManifestId: 1, entries: { document: "note.md" } }, - remoteHeads: { document: head }, lastSeenUpdateId: 0 }, getLatestDocumentByRelativePath: (path: string) => diff --git a/frontend/sync-client/src/sync-operations/syncer.ts b/frontend/sync-client/src/sync-operations/syncer.ts index 7884efda..10379437 100644 --- a/frontend/sync-client/src/sync-operations/syncer.ts +++ b/frontend/sync-client/src/sync-operations/syncer.ts @@ -6,9 +6,10 @@ import { diff } from "reconcile-text"; import type { Database, EngineState, - StoredSnapshot + PendingRequest } from "../persistence/database"; import { emptyState } from "../persistence/database"; +import { toStoredSnapshot, type StoredSnapshot } from "../snapshot"; import type { FileOperations, FileWrite @@ -17,7 +18,6 @@ import type { Settings } from "../persistence/settings"; import type { SyncService } from "../services/sync-service"; import type { ServerConfig } from "../services/server-config"; import type { WebSocketManager } from "../services/websocket-manager"; -import type { DocumentVersion } from "../services/types/DocumentVersion"; import type { DocumentVersionWithoutContent } from "../services/types/DocumentVersionWithoutContent"; import type { EventBatch } from "../services/types/EventBatch"; import type { EventRecord } from "../services/types/EventRecord"; @@ -27,7 +27,7 @@ import type { PushContent } from "../services/types/PushContent"; import type { Logger } from "../tracing/logger"; import { EventListeners } from "../utils/data-structures/event-listeners"; import { VaultHistoryStatus } from "../types/vault-history-status"; -import { mergeContent, toStoredSnapshot } from "./content"; +import { mergeContent } from "./content"; import { scanLocalFiles, type LocalChange } from "./scan"; import { mergeFileManifests, @@ -83,7 +83,7 @@ export class Syncer { private readonly localChangeStorage: { entries: LocalChange[]; save: (entries: LocalChange[]) => Promise; - } = { entries: [], save: async () => { } } + } = { entries: [], save: async () => {} } ) { this.changes = localChangeStorage.entries; websocket.onWebSocketStatusChanged.add((connected) => { @@ -104,7 +104,7 @@ export class Syncer { } public get isFirstSyncComplete(): boolean { - return this.database.state.initialized; + return this.database.state.lastSeenUpdateId !== undefined; } public get hasLocalChanges(): boolean { @@ -123,7 +123,7 @@ export class Syncer { public isDocumentUpToDate(path: string): boolean { if ( - !this.database.state.initialized || + this.database.state.lastSeenUpdateId === undefined || !this.hasScanned || this.pendingPaths.has(path) || this.unsyncablePaths.has(path) || @@ -136,17 +136,16 @@ export class Syncer { if (!record) return false; const document = this.database.state.documents[record.documentId]; if ( - !document?.materialized || + document?.observedHash === undefined || !document.base || - document.bootstrap || document.observedHash !== document.base.hash || this.database.state.fileManifest.entries[record.documentId] !== path ) { return false; } - const remote = this.database.state.remoteHeads[record.documentId]; - if (remote && remote.vaultUpdateId > document.base.vaultUpdateId) + const remote = document.remote; + if (remote && remote.vaultUpdateId > (document.base.vaultUpdateId ?? 0)) return false; const { pending } = this.database.state; @@ -159,7 +158,7 @@ export class Syncer { if ( pending?.type === "fileManifest" && pending.request.entries[record.documentId] !== - this.database.state.fileManifest.entries[record.documentId] + this.database.state.fileManifest.entries[record.documentId] ) { return false; } @@ -323,15 +322,13 @@ export class Syncer { try { await this.reloadLocalState(); if ( - this.database.state.historyRecovery || - (this.database.state.initialized && - this.service.vaultHistoryStatus === VaultHistoryStatus.MISSING) + this.isFirstSyncComplete && + this.service.vaultHistoryStatus === VaultHistoryStatus.MISSING ) await this.recoverServerHistory(); + if (!this.isFirstSyncComplete) await this.service.resetHistory(); await this.serverConfig.initialize(); - if (!this.database.state.initialized) { - await this.bootstrap(); - } + if (!this.isFirstSyncComplete) await this.bootstrap(); if (!this.stopped) this.onReadyForEvents.trigger(); do { this.dirty = false; @@ -341,31 +338,9 @@ export class Syncer { await this.finishPending(); } const batch = await this.service.getEvents( - this.database.state.lastSeenUpdateId + this.database.state.lastSeenUpdateId! ); await this.incorporateEventBatch(batch); - // Previously excluded content becomes eligible after a setting - // change. Its saved obligation survives advancing the cursor. - if (Object.keys(this.database.state.excluded ?? {}).length) - await this.incorporateFileManifest( - this.database.state.fileManifest - ); - else await this.scan(); - for (const head of Object.values( - this.database.state.remoteHeads - )) { - const doc = this.database.state.documents[head.documentId]; - if ( - this.database.state.local[head.documentId] !== - undefined && - (!doc?.materialized || - !doc.base || - doc.bootstrap || - doc.base.vaultUpdateId < head.vaultUpdateId) - ) { - await this.incorporateContent(head); - } - } if (await this.preparePush()) { await this.finishPending(); this.dirty = true; @@ -399,52 +374,40 @@ export class Syncer { this.onServerHistoryChanged.trigger(); this.contentCache.reset(); this.hasScanned = false; - if (!this.database.state.historyRecovery) { - await this.scan(); - const guard = this.localGuard(); - const previous = this.database.state; - const next = emptyState(previous.vaultKey); - next.historyRecovery = true; - next.lastAppliedLocalChangeId = previous.lastAppliedLocalChangeId; - next.protectedPaths = previous.protectedPaths; - for (const [id, path] of Object.entries(previous.local)) { - const info = await this.files.fs.stat(path); - if (info?.kind !== "file") continue; - const doc = previous.documents[id]; - if ( - this.settings.isIgnored(path) || - this.settings.isOversized(info.size) - ) { - next.local[id] = path; - next.documents[id] = { - materialized: true, - bootstrap: true - }; - (next.excluded ??= {})[id] = - previous.fileManifest.entries[id] ?? null; - continue; - } - const snapshot = await this.files.snapshot(path); - if (!snapshot) throw new LocalChangesDuringReconciliation(); - // An unsent edit from the abandoned history is a separate local - // document. It must never be diffed against a reused version ID. - const clean = doc?.base?.hash === snapshot.hash; - const localId = doc?.base && !clean ? uuid() : id; - next.local[localId] = path; - next.documents[localId] = { - materialized: true, - observedHash: snapshot.hash, - recoveryBase: clean ? snapshot : doc?.recoveryBase - }; + await this.scan(); + const guard = this.localGuard(); + const previous = this.database.state; + const next = emptyState(previous.vaultKey); + next.lastAppliedLocalChangeId = previous.lastAppliedLocalChangeId; + for (const [id, path] of Object.entries(previous.local)) { + const info = await this.files.fs.stat(path); + if (info?.kind !== "file") continue; + const doc = previous.documents[id]; + if ( + this.settings.isIgnored(path) || + this.settings.isOversized(info.size) + ) { + next.local[id] = path; + next.documents[id] = { observedHash: doc?.observedHash ?? "" }; + (next.excluded ??= {})[id] = + previous.fileManifest.entries[id] ?? null; + continue; } - guard(); - // Persist the reset BEFORE releasing the old checkpoint. After a - // crash no old request can be retried against the replacement log. - await this.database.commit(next); + const snapshot = await this.files.snapshot(path); + if (!snapshot) throw new LocalChangesDuringReconciliation(); + const clean = doc?.base?.hash === snapshot.hash; + const localId = doc?.base && !clean ? uuid() : id; + next.local[localId] = path; + next.documents[localId] = { + observedHash: snapshot.hash, + base: clean + ? { vaultUpdateId: null, hash: snapshot.hash } + : undefined + }; } - await this.service.resetHistory(); - const next = this.next(); - delete next.historyRecovery; + guard(); + // An absent cursor durably abandons all old requests and version IDs. + // The next bootstrap clears the transport checkpoint before any request. await this.database.commit(next); } @@ -511,20 +474,11 @@ export class Syncer { } private async bootstrap(): Promise { - let vaultSnapshot = this.database.state.bootstrap; - if (!vaultSnapshot) { - vaultSnapshot = await this.service.getVaultSnapshot(); - validateFileManifest(vaultSnapshot.fileManifest.entries); - await this.database.commit({ - ...this.next(), - bootstrap: vaultSnapshot - }); - } - await this.incorporateFileManifest( - vaultSnapshot.fileManifest, - vaultSnapshot.headEventId, - vaultSnapshot - ); + const vaultSnapshot = await this.service.getVaultSnapshot(); + await this.reconcile(vaultSnapshot.fileManifest, { + cursor: vaultSnapshot.headEventId, + initial: vaultSnapshot + }); } private async scan(initial?: VaultSnapshot): Promise { @@ -548,13 +502,15 @@ export class Syncer { } private async incorporateEventBatch(batch: EventBatch): Promise { - let cursor = this.database.state.lastSeenUpdateId; + let cursor = this.database.state.lastSeenUpdateId ?? 0; let manifest = this.database.state.fileManifest; - const heads = { ...this.database.state.remoteHeads }; + const heads: Record = {}; + for (const doc of Object.values(this.database.state.documents)) + if (doc.remote) heads[doc.remote.documentId] = doc.remote; const receipts: EventRecord[] = []; // Fold every page before touching local files. If fetching is interrupted, // discard this work and retry from the last committed cursor. - for (; ;) { + for (;;) { const after = cursor; for (const event of batch.events) { if (event.eventId <= cursor) continue; @@ -579,7 +535,7 @@ export class Syncer { } else if ( !heads[event.document.documentId] || heads[event.document.documentId].vaultUpdateId < - event.document.vaultUpdateId + event.document.vaultUpdateId ) { heads[event.document.documentId] = this.documentHead( event.document @@ -596,7 +552,6 @@ export class Syncer { if (end === batch.headEventId) break; batch = await this.service.getEvents(cursor); } - if (cursor === this.database.state.lastSeenUpdateId) return; // Recover our submitted bases before folding later remote edits. A // permanent retry failure cannot cancel an earlier in-flight attempt. const next = this.next(); @@ -608,13 +563,15 @@ export class Syncer { if (!pending) continue; if (event.type === "content" && pending.type === "content") { const doc = next.documents[pending.documentId]; - if (!doc.base || doc.base.vaultUpdateId < event.eventId) { + if ( + !doc.base || + (doc.base.vaultUpdateId ?? 0) < event.eventId + ) { doc.base = { - ...event.document, + vaultUpdateId: event.eventId, hash: pending.snapshot.hash }; - doc.bootstrap = false; - delete doc.rejected; + this.clearRejections(next, pending.documentId); await this.cacheContent( event.eventId, next.local[pending.documentId], @@ -627,7 +584,7 @@ export class Syncer { ) { if (next.fileManifest.fileManifestId < event.eventId) { next.fileManifest = event.fileManifest; - delete next.rejectedManifest; + this.clearRejections(next, "manifest"); } } else throw new PermanentSyncError( @@ -639,13 +596,7 @@ export class Syncer { if (acknowledged) await this.database.commit(next); // Three-way merging compares snapshots. Replaying intermediate remote // edits can discard a local edit even when the remote ends at its base. - await this.incorporateFileManifest( - manifest, - cursor, - undefined, - false, - heads - ); + await this.reconcile(manifest, { cursor, heads }); } private documentHead( @@ -679,34 +630,30 @@ export class Syncer { details: from && to && from !== to ? { - type: SyncType.MOVE, - relativePath: to, - movedFrom: from - } + type: SyncType.MOVE, + relativePath: to, + movedFrom: from + } : !from && to - ? { type: SyncType.CREATE, relativePath: to } - : from && !to - ? { type: SyncType.DELETE, relativePath: from } - : { - type: SyncType.UPDATE, - relativePath: to ?? from - } + ? { type: SyncType.CREATE, relativePath: to } + : from && !to + ? { type: SyncType.DELETE, relativePath: from } + : { + type: SyncType.UPDATE, + relativePath: to ?? from + } }); } } private async remoteSnapshot( - head: DocumentVersionWithoutContent, - path: string, - supplied?: DocumentVersion + head: Pick< + DocumentVersionWithoutContent, + "documentId" | "vaultUpdateId" + >, + path: string ): Promise { - const content = - supplied !== undefined - ? base64ToBytes(supplied.contentBase64) - : await this.service.getDocumentVersionContent({ - documentId: head.documentId, - vaultUpdateId: head.vaultUpdateId - }); + const content = await this.service.getDocumentVersionContent(head); await this.cacheContent(head.vaultUpdateId, path, content); return toStoredSnapshot({ content }); } @@ -714,12 +661,9 @@ export class Syncer { private async contentWrite( next: EngineState, head: DocumentVersionWithoutContent, - path: string, - supplied?: DocumentVersion + path: string ): Promise { - const document = (next.documents[head.documentId] ??= { - materialized: false - }); + const document = (next.documents[head.documentId] ??= {}); if ( next.excluded?.[head.documentId] !== undefined || this.settings.isIgnored(path) || @@ -730,7 +674,7 @@ export class Syncer { const currentPath = this.database.state.local[head.documentId]; const local = currentPath && - (await this.files.fs.stat(currentPath))?.kind === "file" + (await this.files.fs.stat(currentPath))?.kind === "file" ? await this.files.snapshot(currentPath) : undefined; if ( @@ -739,16 +683,27 @@ export class Syncer { ) { return undefined; } + if ( + document.base?.vaultUpdateId === null && + local && + local.hash !== document.base.hash + ) + throw new LocalChangesDuringReconciliation(); const base = - document.recoveryBase ?? - (local !== undefined && - document.base !== undefined && - document.bootstrap !== true - ? await this.remoteSnapshot(document.base, path) - : undefined); + local && document.base + ? document.base.vaultUpdateId === null + ? local + : await this.remoteSnapshot( + { + documentId: head.documentId, + vaultUpdateId: document.base.vaultUpdateId + }, + path + ) + : undefined; // Cache the new head after the older merge base so it remains the most // recently used version and can back the follow-up local diff. - const remote = await this.remoteSnapshot(head, path, supplied); + const remote = await this.remoteSnapshot(head, path); let result = remote; if (local !== undefined) { result = await mergeContent( @@ -759,74 +714,30 @@ export class Syncer { (await this.serverConfig.getConfig()).mergeableFileExtensions ); } - document.base = { ...this.documentHead(head), hash: remote.hash }; - delete document.recoveryBase; - document.bootstrap = false; - document.materialized = true; + document.base = { + vaultUpdateId: head.vaultUpdateId, + hash: remote.hash + }; + delete document.remote; document.observedHash = result.hash; return local?.contentBase64 === result.contentBase64 ? undefined : { expected: local, replacement: result }; } - private async incorporateContent( - head: DocumentVersionWithoutContent, - cursor?: number, - supplied?: DocumentVersion, - clearPending = false - ): Promise { - await this.scan(); - const guard = this.localGuard(); - const next = this.next(); - if ( - !next.remoteHeads[head.documentId] || - next.remoteHeads[head.documentId].vaultUpdateId < head.vaultUpdateId - ) { - next.remoteHeads[head.documentId] = this.documentHead(head); - } - if (cursor !== undefined) { - next.lastSeenUpdateId = cursor; - } - if (clearPending) { - delete next.pending; - } - const path = next.local[head.documentId]; - const document = next.documents[head.documentId]; - const wasMaterialized = document?.materialized; - const writes: Record = {}; - if ( - path !== undefined && - (!document?.base || - !document.materialized || - document.bootstrap || - document.base.vaultUpdateId < head.vaultUpdateId) - ) { - const write = await this.contentWrite(next, head, path, supplied); - if (write) { - writes[head.documentId] = write; - } - } - await this.files.apply(next, writes, guard); - if (path !== undefined && writes[head.documentId]) { - this.history.addHistoryEntry({ - status: SyncStatus.SUCCESS, - message: "Remote content applied", - details: { - type: wasMaterialized ? SyncType.UPDATE : SyncType.CREATE, - relativePath: path - }, - author: head.userId, - timestamp: new Date(head.updatedDate) - }); - } - } - - private async incorporateFileManifest( + private async reconcile( remote: FileManifest, - cursor?: number, - initial?: VaultSnapshot, - clearPending = false, - heads?: EngineState["remoteHeads"] + { + cursor, + initial, + clearPending = false, + heads + }: { + cursor?: number; + initial?: VaultSnapshot; + clearPending?: boolean; + heads?: Record; + } = {} ): Promise { validateFileManifest(remote.entries); await this.scan(initial); @@ -852,7 +763,8 @@ export class Syncer { const previousManifest = next.fileManifest.entries; next.local = mergeFileManifests(base, next.local, remote.entries); next.fileManifest = remote; - if (heads) next.remoteHeads = heads; + for (const head of Object.values(heads ?? {})) + (next.documents[head.documentId] ??= {}).remote = head; for (const [id, oldPath] of Object.entries(before)) { const path = next.local[id]; if ( @@ -877,41 +789,43 @@ export class Syncer { next.local, remote.entries, Object.keys(next.excluded ?? {}), - next.protectedPaths + [...this.unsyncablePaths] ); if (initial) { - next.initialized = true; - delete next.bootstrap; - for (const head of initial.documents) { - next.remoteHeads[head.documentId] = this.documentHead(head); - } + for (const head of initial.documents) + (next.documents[head.documentId] ??= {}).remote = + this.documentHead(head); } const writes: Record = {}; for (const [id, path] of Object.entries(next.local)) { - next.documents[id] ??= { materialized: false }; - if (before[id] === undefined) - next.documents[id].materialized = false; + const doc = (next.documents[id] ??= {}); + if (before[id] === undefined) delete doc.observedHash; + if ( + doc.observedHash !== undefined && + doc.remote && + doc.remote.vaultUpdateId <= (doc.base?.vaultUpdateId ?? 0) + ) + delete doc.remote; if ( next.excluded?.[id] === undefined && !this.settings.isIgnored(path) && - remote.entries[id] !== undefined && + (remote.entries[id] !== undefined || + doc.remote !== undefined) && (!this.database.state.local[id] || - !next.documents[id].base || - next.documents[id].bootstrap || - !next.documents[id].materialized || - (next.remoteHeads[id]?.vaultUpdateId ?? 0) > - (next.documents[id].base?.vaultUpdateId ?? 0)) + !doc.base || + doc.observedHash === undefined || + (doc.remote?.vaultUpdateId ?? 0) > + (doc.base.vaultUpdateId ?? 0)) ) { - const knownHead = - next.remoteHeads[id] ?? - initial?.documents.find((doc) => doc.documentId === id); - const head = knownHead ?? (await this.service.getDocumentMetadata(id)); + const knownHead = doc.remote; + const head = + knownHead ?? (await this.service.getDocumentMetadata(id)); if (!head) { throw new PermanentSyncError( `File manifest references missing content: ${id}` ); } - next.remoteHeads[id] = this.documentHead(head); + doc.remote = this.documentHead(head); const write = await this.contentWrite(next, head, path); if (write) { writes[id] = write; @@ -932,7 +846,7 @@ export class Syncer { this.database.state.local, this.database.state.fileManifest.entries, Object.keys(this.database.state.excluded ?? {}), - this.database.state.protectedPaths + [...this.unsyncablePaths] ); if (!sameFileManifest(canonical, this.database.state.local)) { await this.files.apply( @@ -960,11 +874,11 @@ export class Syncer { if ( this.settings.isIgnored(path) || this.database.state.excluded?.[id] !== undefined || - !this.database.state.documents[id]?.materialized || + this.database.state.documents[id]?.observedHash === undefined || this.unsyncablePaths.has(path) || this.settings.isOversized( // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- Document heads are a sparse ID map at runtime. - this.database.state.remoteHeads[id]?.contentSize ?? 0 + this.database.state.documents[id]?.remote?.contentSize ?? 0 ) ) { continue; @@ -978,10 +892,15 @@ export class Syncer { ) { continue; } - const { base, rejected } = this.database.state.documents[id]; + const { base } = this.database.state.documents[id]; if ( base?.hash === snapshot.hash || - rejected?.hash === snapshot.hash + this.rejections.some( + (request) => + request.type === "content" && + request.documentId === id && + request.snapshot.hash === snapshot.hash + ) ) { continue; } @@ -997,7 +916,7 @@ export class Syncer { parentVersionId: base?.vaultUpdateId ?? null, content: await this.pushContent( path, - base?.vaultUpdateId, + base?.vaultUpdateId ?? undefined, snapshot ) } @@ -1013,11 +932,11 @@ export class Syncer { entries, this.database.state.fileManifest.entries ) && - (!this.database.state.rejectedManifest || - !sameFileManifest( - entries, - this.database.state.rejectedManifest.entries - )) + !this.rejections.some( + (request) => + request.type === "fileManifest" && + sameFileManifest(entries, request.request.entries) + ) ) { guard(); await this.database.commit({ @@ -1058,7 +977,7 @@ export class Syncer { this.database.state.fileManifest.entries, Object.keys(this.database.state.excluded ?? {}), [ - ...(this.database.state.protectedPaths ?? []), + ...this.unsyncablePaths, ...Object.keys(this.database.state.excluded ?? {}).flatMap( (id) => this.database.state.local[id] @@ -1069,50 +988,59 @@ export class Syncer { ); } + private get rejections(): PendingRequest[] { + return (this.database.state.unconfirmed ?? []).filter( + (request) => request.rejection !== undefined + ); + } + private get rejectionMessage(): string | undefined { const { state } = this.database; - for (const [id, path] of Object.entries(state.local ?? {})) { - const doc = state.documents[id]; - if ( - doc?.rejected && - doc.observedHash === doc.rejected.hash && - doc.base?.hash !== doc.rejected.hash && - !this.settings.isIgnored(path) && - state.excluded?.[id] === undefined + for (const request of this.rejections) { + if (request.type === "content") { + const id = request.documentId; + const path = state.local[id]; + const doc = state.documents[id]; + if ( + path !== undefined && + doc?.observedHash === request.snapshot.hash && + doc.base?.hash !== request.snapshot.hash && + !this.settings.isIgnored(path) && + state.excluded?.[id] === undefined + ) + return request.rejection; + } else if ( + sameFileManifest( + this.manifestToSubmit(), + request.request.entries + ) && + !sameFileManifest( + state.fileManifest.entries, + request.request.entries + ) ) - return doc.rejected.message; + return request.rejection; } - if ( - state.rejectedManifest && - sameFileManifest( - this.manifestToSubmit(), - state.rejectedManifest.entries - ) && - !sameFileManifest( - state.fileManifest.entries, - state.rejectedManifest.entries + } + + private clearRejections(state: EngineState, id?: string): void { + for (const request of state.unconfirmed ?? []) + if ( + id === undefined || + (request.type === "content" + ? request.documentId + : "manifest") === id ) - ) - return state.rejectedManifest.message; - return undefined; + delete request.rejection; } /** An explicit reset or settings change permits fresh attempts after rejection. */ public async retryRejectedRequests(): Promise { await this.database.recoverPersistence(); - const next = structuredClone(this.database.state); - let changed = false; - if (next.rejectedManifest) { - delete next.rejectedManifest; - changed = true; - } - for (const doc of Object.values(next.documents)) { - if (doc.rejected) { - delete doc.rejected; - changed = true; - } - } - if (changed) await this.database.commit(next); + if (!this.rejections.length) return; + const next = this.next(); + this.clearRejections(next); + await this.database.commit(next); } private async finishPending(): Promise { @@ -1144,16 +1072,6 @@ export class Syncer { if (pending.rejection && !pending.response) { const next = this.next(); - if (pending.type === "content") { - next.documents[pending.documentId].rejected = { - hash: pending.snapshot.hash, - message: pending.rejection - }; - } else - next.rejectedManifest = { - entries: pending.request.entries, - message: pending.rejection - }; (next.unconfirmed ??= []).push(pending); delete next.pending; await this.database.commit(next); @@ -1165,7 +1083,7 @@ export class Syncer { relativePath: pending.type === "content" ? (next.local[pending.documentId] ?? - pending.documentId) + pending.documentId) : "File manifest" } }); @@ -1175,15 +1093,10 @@ export class Syncer { if (pending.type === "fileManifest") { const response = pending.response!; if (response.type === "StaleBase") { - await this.incorporateFileManifest( - response, - undefined, - undefined, - true - ); + await this.reconcile(response, { clearPending: true }); } else { const next = this.next(); - delete next.rejectedManifest; + this.clearRejections(next, "manifest"); next.fileManifest = { fileManifestId: response.fileManifestId, entries: pending.request.entries @@ -1206,15 +1119,15 @@ export class Syncer { details: after === undefined ? { - type: SyncType.DELETE, - relativePath: before - } + type: SyncType.DELETE, + relativePath: before + } : before === undefined - ? { + ? { type: SyncType.CREATE, relativePath: after } - : { + : { type: SyncType.MOVE, relativePath: after, movedFrom: before @@ -1225,25 +1138,19 @@ export class Syncer { } else { const response = pending.response!; if (response.type === "StaleBase") { - await this.incorporateContent( - response, - undefined, - undefined, - true - ); + await this.reconcile(this.database.state.fileManifest, { + clearPending: true, + heads: { [pending.documentId]: this.documentHead(response) } + }); } else { const next = this.next(); - const doc = (next.documents[pending.documentId] ??= { - materialized: false - }); + const doc = (next.documents[pending.documentId] ??= {}); doc.base = { - ...this.documentHead(response), + vaultUpdateId: response.vaultUpdateId, hash: pending.snapshot.hash }; - doc.bootstrap = false; - delete doc.rejected; - next.remoteHeads[pending.documentId] = - this.documentHead(response); + this.clearRejections(next, pending.documentId); + delete doc.remote; delete next.pending; await this.database.commit(next); const path = next.local[pending.documentId]; diff --git a/frontend/test-support/canonical.ts b/frontend/test-support/canonical.ts index 296413a2..4fb1c751 100644 --- a/frontend/test-support/canonical.ts +++ b/frontend/test-support/canonical.ts @@ -68,7 +68,7 @@ export async function assertCanonical( "Snapshot heads differ from membership" ); for (const [i, state] of states.entries()) { - assert(state?.initialized, `Client ${i} never initialized`); + assert(state?.lastSeenUpdateId !== undefined, `Client ${i} never initialized`); assertQuiescent(state, entries, snapshot.headEventId); assert.equal( state.fileManifest?.fileManifestId, @@ -82,7 +82,7 @@ export async function assertCanonical( ); for (const id of Object.keys(entries)) { assert( - state.documents?.[id]?.materialized, + state.documents?.[id]?.observedHash !== undefined, `Unmaterialized document: ${id}` ); assert.equal( diff --git a/frontend/test-support/content-limits.test.ts b/frontend/test-support/content-limits.test.ts index baf510a4..09627b58 100644 --- a/frontend/test-support/content-limits.test.ts +++ b/frontend/test-support/content-limits.test.ts @@ -75,9 +75,7 @@ test("stale content acknowledgements contain metadata without the excluded paylo test("highly fragmented local edits use a snapshot within the server operation budget", async () => { const { fixture } = await import("./sync-fixture"); - const { toStoredSnapshot } = await import( - "../sync-client/src/sync-operations/content" - ); + const { toStoredSnapshot } = await import("../sync-client/src/snapshot"); const f = await fixture({}); f.contentCache.resize(1024 * 1024); const original = Array.from({ length: 12_000 }, (_, i) => `word${i}`).join( diff --git a/frontend/test-support/engine-regressions.test.ts b/frontend/test-support/engine-regressions.test.ts index 35beadc4..2f99bfd5 100644 --- a/frontend/test-support/engine-regressions.test.ts +++ b/frontend/test-support/engine-regressions.test.ts @@ -4,7 +4,7 @@ import { test } from "node:test"; import { Database, emptyState } from "../sync-client/src/persistence/database"; import { Syncer } from "../sync-client/src/sync-operations/syncer"; import { SyncClient, type StoredClient } from "../sync-client/src/sync-client"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import { allocatePortablePath } from "../sync-client/src/utils/portable-path"; import { MemoryDisk, MemoryPersistence } from "./storage"; @@ -14,10 +14,10 @@ test("offline content notifications need no saves and restart reads the current const disk = new MemoryDisk(); await disk.userWrite("a.md", bytes("original")); const initial = emptyState(JSON.stringify(["http://offline.test", "test"])); - initial.initialized = true; + initial.lastSeenUpdateId = 0; initial.local = { a: "a.md" }; initial.documents.a = { - materialized: true, + observedHash: (await toStoredSnapshot({ content: bytes("original") })) .hash }; @@ -70,7 +70,7 @@ test("legacy content notifications preserve the acknowledged namespace prefix on await disk.userWrite("d.md", bytes("A")); await disk.userWrite("b.md", bytes("B")); const initial = emptyState(JSON.stringify(["http://offline.test", "test"])); - initial.initialized = true; + initial.lastSeenUpdateId = 0; initial.local = { a: "c.md", b: "b.md" }; initial.fileManifest.entries = { ...initial.local }; for (const [id, content] of [ @@ -78,7 +78,7 @@ test("legacy content notifications preserve the acknowledged namespace prefix on ["b", "B"] ]) initial.documents[id] = { - materialized: true, + observedHash: (await toStoredSnapshot({ content: bytes(content) })) .hash }; @@ -134,7 +134,7 @@ test("remote binary replacement and deletion do not create backup artifacts", as }); assert.deepEqual(await f.disk.listFilesRecursively(), ["a.bin"]); assert.deepEqual(f.disk.userFiles(), new Map([["a.bin", bytes("remote")]])); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: {} }); @@ -202,10 +202,11 @@ test("an unmaterialized file with a directory at its path is deconflicted", asyn ); await f.disk.userWrite("folder/note.md", bytes("local")); f.database.state.local.a = "folder"; - f.database.state.documents.a = { materialized: false }; + f.database.state.documents.a = {}; f.database.state.fileManifest.entries.a = "folder"; await f.database.save(); - await f.internals.incorporateContent(head("a", 2, "remote")); + f.database.state.documents.a.remote = head("a", 2, "remote"); + await f.internals.reconcile(f.database.state.fileManifest); assert( [...f.disk.userFiles().values()].some( (value) => Buffer.from(value).toString() === "remote" @@ -221,7 +222,7 @@ test("an unmaterialized file with a directory at its path is deconflicted", asyn test("ignored materialized content survives remote deletion", async () => { const f = await fixture(); await f.settings.setSettings({ ignorePatterns: ["a.md"] }); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: {} }); @@ -231,11 +232,9 @@ test("ignored materialized content survives remote deletion", async () => { test("unfinished bootstrap cannot be rebound to another vault", async () => { const database = emptyState(JSON.stringify(["http://example.test", "old"])); - database.bootstrap = { - headEventId: 1, - fileManifest: { fileManifestId: 1, entries: {} }, - documents: [] - }; + // A scan may commit local identities before the first remote application. + database.local = { a: "a.md" }; + database.documents.a = { observedHash: "" }; const persistence = new MemoryPersistence({ settings: { remoteUri: "http://example.test", vaultName: "old" }, database @@ -276,7 +275,7 @@ test("a rejected upload is retained locally and does not block another file or a assert(await f.internals.preparePush()); await f.internals.finishPending(); assert.equal(f.database.state.pending, undefined); - assert.match(f.database.state.documents.a.rejected!.message, /413/); + assert.match(f.database.state.unconfirmed![0].rejection!, /413/); assert(await f.internals.preparePush()); await f.internals.finishPending(); assert.deepEqual(sent, ["a", "b"]); @@ -286,7 +285,7 @@ test("a rejected upload is retained locally and does not block another file or a assert(await f.internals.preparePush()); await f.internals.finishPending(); assert.deepEqual(sent, ["a", "b", "a"]); - assert.equal(f.database.state.documents.a.rejected, undefined); + assert.equal(f.database.state.unconfirmed?.find((request) => request.rejection), undefined); }); test("a transient upload failure retries the exact request and recovers its acknowledgement", async () => { @@ -306,7 +305,7 @@ test("a transient upload failure retries the exact request and recovers its ackn assert.equal(requests.length, 2); assert.deepEqual(requests[1], requests[0]); assert.equal(f.database.state.documents.a.base!.vaultUpdateId, 2); - assert.equal(f.database.state.documents.a.rejected, undefined); + assert.equal(f.database.state.unconfirmed?.find((request) => request.rejection), undefined); assert.equal(f.database.state.pending, undefined); }); @@ -327,7 +326,7 @@ test("a persisted upload rejection finishes without filesystem writes", async () await f.internals.finishPending(); assert.equal(uploads, 1); assert.equal(f.database.state.pending, undefined); - assert.equal(f.database.state.documents.a.rejected?.message, "HTTP 413"); + assert.equal(f.database.state.unconfirmed![0].rejection, "HTTP 413"); assert.deepEqual(f.disk.userFiles(), new Map([["a.md", bytes("local")]])); }); @@ -388,7 +387,6 @@ test("rejected manifests can be replaced after editing the local namespace", asy }; f.database.state.local = { a: "new.md" }; f.database.state.documents.a = { - materialized: false, base: { ...head("a", 1, "x"), hash: "hash" } }; await f.database.save(); @@ -405,9 +403,9 @@ test("rejected manifests can be replaced after editing the local namespace", asy test("offline delete/recreate identities survive restarting before any network connection", async () => { const key = JSON.stringify(["http://offline.test", "test"]); const initial = emptyState(key); - initial.initialized = true; + initial.lastSeenUpdateId = 0; initial.local.a = "a.md"; - initial.documents.a = { materialized: true }; + initial.documents.a = { observedHash: "" }; initial.fileManifest = { fileManifestId: 1, entries: { a: "a.md" } }; const disk = new MemoryDisk(); await disk.userWrite("a.md", bytes("same")); @@ -465,12 +463,12 @@ test("offline notifications bind an otherwise empty client to its vault", async test("unignoring a deleted file resumes the deferred remote deletion without resurrection", async () => { const f = await fixture(); await f.settings.setSettings({ ignorePatterns: ["a.md"] }); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: {} }); await f.settings.setSettings({ ignorePatterns: [] }); - await f.internals.incorporateFileManifest(f.database.state.fileManifest); + await f.internals.reconcile(f.database.state.fileManifest); assert.equal(f.disk.userFiles().size, 0); assert.deepEqual(f.database.state.local, {}); }); @@ -478,7 +476,7 @@ test("unignoring a deleted file resumes the deferred remote deletion without res test("an ignored file does not move when a remote rename targets its path", async () => { const f = await fixture({ "a.md": "A", "b.md": "B" }); await f.settings.setSettings({ ignorePatterns: ["a.md"] }); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: { a: "c.md", b: "a.md" } }); @@ -623,12 +621,12 @@ test("ignoring a locally renamed file defers its move without forgetting the loc await f.internals.scan(); f.internals.stopped = false; assert.equal(await f.internals.preparePush(), false); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: { a: "a.md" } }); await f.settings.setSettings({ ignorePatterns: [] }); - await f.internals.incorporateFileManifest(f.database.state.fileManifest); + await f.internals.reconcile(f.database.state.fileManifest); assert.equal(f.database.state.local.a, "ignored.md"); assert.equal( Buffer.from( @@ -648,7 +646,7 @@ test("a notified deletion of an ignored file does not delete its server membersh assert.equal(await f.internals.preparePush(), false); assert.equal(f.database.state.local.a, undefined); await f.settings.setSettings({ ignorePatterns: [] }); - await f.internals.incorporateFileManifest(f.database.state.fileManifest); + await f.internals.reconcile(f.database.state.fileManifest); assert(await f.internals.preparePush()); assert.equal(f.database.state.pending?.type, "fileManifest"); }); @@ -656,7 +654,7 @@ test("a notified deletion of an ignored file does not delete its server membersh test("an incoming ignored document cannot force an existing ignored file to move", async () => { const f = await fixture(); await f.settings.setSettings({ ignorePatterns: ["a.md", "*conflict*"] }); - await f.internals.incorporateFileManifest({ + await f.internals.reconcile({ fileManifestId: 2, entries: { a: "b.md", other: "a.md" } }); @@ -687,7 +685,7 @@ test("rejected content is reported to callers and an explicit retry clears the r test("notification saves cannot roll back an uncertain engine commit, even if the first reload fails", async () => { const key = JSON.stringify(["http://offline.test", "test"]); const initial = emptyState(key); - initial.initialized = true; + initial.lastSeenUpdateId = 0; const disk = new MemoryDisk(); const store = new MemoryPersistence({ database: initial, @@ -755,3 +753,40 @@ test( assert(notified); } ); + +for (const edited of [false, true]) { + test(`hash-only history recovery preserves ${edited ? "an edit made after reset" : "a clean file"} across restart`, async () => { + const remote = head("a", 2, "restored"); + const service = { + getVaultSnapshot: async () => ({ + headEventId: 3, + fileManifest: { fileManifestId: 3, entries: { a: "a.md" } }, + documents: [remote] + }), + getDocumentVersionContent: async ({ vaultUpdateId }: { vaultUpdateId: number }) => { + assert.equal(vaultUpdateId, 2, "never read a version from the abandoned history"); + return bytes("restored"); + } + }; + let f = await fixture({ "a.md": "old branch" }, service); + const hash = f.database.state.documents.a.observedHash!; + f.database.state.documents.a.base = { vaultUpdateId: 50, hash }; + await (f.syncer as unknown as { recoverServerHistory(): Promise }).recoverServerHistory(); + assert.equal(f.database.state.lastSeenUpdateId, undefined); + assert.deepEqual(f.database.state.documents.a, { + observedHash: hash, + base: { vaultUpdateId: null, hash } + }); + if (edited) await f.disk.userWrite("a.md", bytes("my new edit")); + f = await fixture({}, service, f); + await (f.syncer as unknown as { bootstrap(): Promise }).bootstrap(); + assert.equal(f.database.state.lastSeenUpdateId, 3); + assert.deepEqual((await f.disk.read("a.md"))?.content, bytes("restored")); + const other = Object.keys(f.database.state.local).filter((id) => id !== "a"); + assert.equal(other.length, edited ? 1 : 0); + if (edited) { + assert.deepEqual((await f.disk.read(f.database.state.local[other[0]]))?.content, bytes("my new edit")); + assert.equal(f.database.state.documents[other[0]].base, undefined); + } + }); +} diff --git a/frontend/test-support/event-delivery.test.ts b/frontend/test-support/event-delivery.test.ts index 7cf4ff0b..d01649f2 100644 --- a/frontend/test-support/event-delivery.test.ts +++ b/frontend/test-support/event-delivery.test.ts @@ -30,7 +30,7 @@ for (const [count, delivery] of [ async () => { const logger = new Logger(); const initial = emptyState("event-delivery"); - initial.initialized = true; + initial.lastSeenUpdateId = 0; const persistence = new MemoryPersistence({ database: initial }); const database = new Database( initial, diff --git a/frontend/test-support/file-application.test.ts b/frontend/test-support/file-application.test.ts index 250047f0..ce53eba7 100644 --- a/frontend/test-support/file-application.test.ts +++ b/frontend/test-support/file-application.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { fixture, bytes } from "./sync-fixture"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import { LocalChangesDuringReconciliation } from "../sync-client/src/errors/errors"; for (const equal of [false, true]) diff --git a/frontend/test-support/oracle-regressions.test.ts b/frontend/test-support/oracle-regressions.test.ts index 3d378086..dbd21b40 100644 --- a/frontend/test-support/oracle-regressions.test.ts +++ b/frontend/test-support/oracle-regressions.test.ts @@ -34,13 +34,13 @@ test("canonical checks independently verify both persisted content hashes", asyn }; const manifest = { fileManifestId: 2, entries: { [id]: "a.md" } }; const state: Partial = { - initialized: true, + local: manifest.entries, fileManifest: manifest, lastSeenUpdateId: 2, documents: { [id]: { - materialized: true, + observedHash: hash, base: { ...document, hash } } @@ -297,7 +297,7 @@ test("canonical checks follow pages without weakening gap or duplicate-request d const client = () => ({ files: () => new Map(), database: () => ({ - initialized: true, + local: {}, documents: {}, fileManifest: manifest, diff --git a/frontend/test-support/polling.test.ts b/frontend/test-support/polling.test.ts index a6314bea..dabc49c3 100644 --- a/frontend/test-support/polling.test.ts +++ b/frontend/test-support/polling.test.ts @@ -8,7 +8,7 @@ import { import { Settings } from "../sync-client/src/persistence/settings"; import { FileOperations } from "../sync-client/src/file-operations/file-operations"; import { Syncer } from "../sync-client/src/sync-operations/syncer"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import { Logger } from "../sync-client/src/tracing/logger"; import { SyncHistory } from "../sync-client/src/tracing/sync-history"; import { FixedSizeDocumentCache } from "../sync-client/src/utils/data-structures/fix-sized-cache"; @@ -87,14 +87,12 @@ async function fixture(syncIntervalMs: number | undefined) { } }; const initial = emptyState("polling-test"); - initial.initialized = true; initial.lastSeenUpdateId = 2; initial.local = { [documentId]: "note.bin" }; initial.fileManifest = { fileManifestId: 2, entries: { ...initial.local } }; - initial.remoteHeads = { [documentId]: latest }; const hash = (await toStoredSnapshot({ content: bytes("base") })).hash; initial.documents[documentId] = { - materialized: true, + remote: latest, observedHash: hash, base: { ...latest, hash } }; diff --git a/frontend/test-support/protocol.test.ts b/frontend/test-support/protocol.test.ts index 6e96bda6..6658ed53 100644 --- a/frontend/test-support/protocol.test.ts +++ b/frontend/test-support/protocol.test.ts @@ -148,7 +148,7 @@ for (const [kind, format, originalPath] of [ )[0]; originalParent = initialized.stored.database!.documents![originalId] - .base!.vaultUpdateId; + .base!.vaultUpdateId!; disk.restore(initialized.disk); uploaded += "local-update\n"; await disk.userWrite( diff --git a/frontend/test-support/real-disk.test.ts b/frontend/test-support/real-disk.test.ts index 283d7afb..ad25f9d6 100644 --- a/frontend/test-support/real-disk.test.ts +++ b/frontend/test-support/real-disk.test.ts @@ -7,7 +7,7 @@ import { NodeFileSystemOperations } from "../local-client-cli/src/node-filesyste import { Database, emptyState } from "../sync-client/src/persistence/database"; import { FileOperations } from "../sync-client/src/file-operations/file-operations"; import type { ServerConfig } from "../sync-client/src/services/server-config"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; test( "real filesystem: direct updates, unsafe entries, and file swaps", @@ -101,7 +101,7 @@ test( initial.local = { a: "a.md", b: "b.md" }; for (const [id, relative] of Object.entries(initial.local)) initial.documents[id] = { - materialized: true, + observedHash: ( await toStoredSnapshot( (await disk.read(relative))! diff --git a/frontend/test-support/recovery.test.ts b/frontend/test-support/recovery.test.ts index 24af52f6..70975c35 100644 --- a/frontend/test-support/recovery.test.ts +++ b/frontend/test-support/recovery.test.ts @@ -7,7 +7,7 @@ import { type StoredDatabase } from "../sync-client/src/persistence/database"; import { FileOperations } from "../sync-client/src/file-operations/file-operations"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import type { ServerConfig } from "../sync-client/src/services/server-config"; import { mergeFileManifests } from "../sync-client/src/sync-operations/file-manifest"; import { assertManifest, pathDecision } from "./oracles"; @@ -35,7 +35,7 @@ async function fixture( } for (const [id, path] of Object.entries(initial.local)) initial.documents[id] = { - materialized: true, + observedHash: ( await toStoredSnapshot((await disk.read(path))!) ).hash @@ -73,7 +73,7 @@ async function swapFixture() { const current = await fixture(); const next = structuredClone(current.initial); next.local = { a: "b.md", b: "a.md", c: "nested/c.md" }; - next.documents.c = { materialized: false }; + next.documents.c = {}; const writes = { c: { replacement: await toStoredSnapshot({ content: bytes("C") }) } }; @@ -274,7 +274,7 @@ for (const [name, create] of [ )) { assert( runtime.database.state.documents[id] - .materialized + .observedHash !== undefined ); assert.equal( runtime.database.state.documents[id] @@ -560,7 +560,7 @@ test("normalization-sensitive disk does not lose a distinct decomposed filename" const runtime = current.restart(); const next = structuredClone(runtime.database.state); next.local.composed = "é.md"; - next.documents.composed = { materialized: true }; + next.documents.composed = { observedHash: "" }; await runtime.database.commit(next); await current.disk.userWrite("e\u0301.md", bytes("decomposed")); await scanLocalFiles({ diff --git a/frontend/test-support/sync-audit.test.ts b/frontend/test-support/sync-audit.test.ts index 49ae8330..a5773b60 100644 --- a/frontend/test-support/sync-audit.test.ts +++ b/frontend/test-support/sync-audit.test.ts @@ -5,7 +5,7 @@ import { test } from "node:test"; import type { Database } from "../sync-client/src/persistence/database"; import { emptyState } from "../sync-client/src/persistence/database"; import { SyncClient } from "../sync-client/src/sync-client"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import { EventListeners } from "../sync-client/src/utils/data-structures/event-listeners"; import type { SyncService } from "../sync-client/src/services/sync-service"; import type { WebSocketManager } from "../sync-client/src/services/websocket-manager"; @@ -50,8 +50,8 @@ test("ignored untracked ancestor must remain in place", async () => { ); await f.disk.userWrite("private", bytes("LOCAL PRIVATE")); await f.settings.setSettings({ ignorePatterns: ["private"] }); - f.database.state.remoteHeads.a = head("a", 2, "remote"); - await f.internals.incorporateFileManifest({ + (f.database.state.documents.a ??= {}).remote = head("a", 2, "remote"); + await f.internals.reconcile({ fileManifestId: 3, entries: { a: "private/note.md" } }); @@ -69,8 +69,8 @@ test("oversized untracked file must remain in place", async () => { ); await f.disk.userWrite("a.md", bytes("LOCAL OVERSIZED")); await f.settings.setSettings({ maxFileSizeMB: 5 / (1024 * 1024) }); - f.database.state.remoteHeads.a = head("a", 2, "R"); - await f.internals.incorporateFileManifest({ + (f.database.state.documents.a ??= {}).remote = head("a", 2, "R"); + await f.internals.reconcile({ fileManifestId: 3, entries: { a: "a.md" } }); @@ -133,7 +133,7 @@ test("rejected unknown success must use submitted content as merge base", async test("bootstrap adopts same-path identity even after multiple notifications", async () => { const f = await fixture({}); - f.database.state.initialized = false; + delete f.database.state.lastSeenUpdateId; f.database.state.fileManifest = { fileManifestId: 0, entries: {} }; await f.disk.userWrite("a.md", bytes("local a")); await f.syncer.syncLocallyCreatedFile("a.md"); @@ -168,7 +168,7 @@ test("oversized remote head must not cause identical stale uploads forever", asy ...head("a", 1, "base"), hash: (await toStoredSnapshot({ content: bytes("base") })).hash }; - f.database.state.remoteHeads.a = head("a", 2, remoteText); + (f.database.state.documents.a ??= {}).remote = head("a", 2, remoteText); await f.settings.setSettings({ maxFileSizeMB: 10 / (1024 * 1024) }); f.internals.stopped = false; // Even a small local edit must wait until its required remote base is readable. @@ -480,13 +480,13 @@ for (const interrupted of [false, true]) { ...head("a", 1, "base"), hash: (await toStoredSnapshot({ content: bytes("base") })).hash }; - f.database.state.remoteHeads.a = head("a", 1, "base"); + (f.database.state.documents.a ??= {}).remote = head("a", 1, "base"); f.database.state.lastSeenUpdateId = 1; await f.database.save(); const saved = f.persistence.snapshot(); const catchUp = async () => f.internals.incorporateEventBatch( - await service.getEvents!(f.database.state.lastSeenUpdateId) + await service.getEvents!(f.database.state.lastSeenUpdateId!) ); if (interrupted) { await assert.rejects(catchUp(), /interrupted page fetch/); @@ -515,8 +515,8 @@ test("an ignored occupant arriving after the scan keeps its path and privacy", a } ); await f.settings.setSettings({ ignorePatterns: ["private"] }); - f.database.state.remoteHeads.a = head("a", 2, "remote"); - await f.internals.incorporateFileManifest({ + (f.database.state.documents.a ??= {}).remote = head("a", 2, "remote"); + await f.internals.reconcile({ fileManifestId: 3, entries: { a: "private/note.md" } }); @@ -818,10 +818,15 @@ for (const boundary of ["before:save", "durable:save"]) { const { database } = f.client as unknown as { database: Database }; await database.commit({ ...database.state, - rejectedManifest: { - entries: { retired: "old.md" }, - message: "retired rejection" - } + unconfirmed: [{ + type: "fileManifest", + request: { + requestId: "rejected", + parentFileManifestId: 0, + entries: { retired: "old.md" } + }, + rejection: "retired rejection" + }] }); let saves = 0, failed = false; diff --git a/frontend/test-support/sync-fixture.ts b/frontend/test-support/sync-fixture.ts index 79047f5d..7128bb6d 100644 --- a/frontend/test-support/sync-fixture.ts +++ b/frontend/test-support/sync-fixture.ts @@ -6,7 +6,7 @@ import { import { Settings } from "../sync-client/src/persistence/settings"; import { FileOperations } from "../sync-client/src/file-operations/file-operations"; import { Syncer } from "../sync-client/src/sync-operations/syncer"; -import { toStoredSnapshot } from "../sync-client/src/sync-operations/content"; +import { toStoredSnapshot } from "../sync-client/src/snapshot"; import { Logger } from "../sync-client/src/tracing/logger"; import { SyncHistory } from "../sync-client/src/tracing/sync-history"; import { FixedSizeDocumentCache } from "../sync-client/src/utils/data-structures/fix-sized-cache"; @@ -37,13 +37,13 @@ export async function fixture( ? (recovered.persistence.snapshot().database as StoredDatabase) : emptyState("test"); if (!recovered) { - initial.initialized = true; + initial.lastSeenUpdateId = 0; for (const [path, content] of Object.entries(contents)) { await disk.userWrite(path, bytes(content)); const id = path.split(".")[0]; initial.local[id] = path; initial.documents[id] = { - materialized: true, + observedHash: ( await toStoredSnapshot({ content: bytes(content) }) ).hash @@ -110,11 +110,10 @@ export async function fixture( batch: import("../sync-client/src/services/types/EventBatch").EventBatch ) => Promise; stopped: boolean; - incorporateFileManifest: (remote: { + reconcile: (remote: { fileManifestId: number; entries: Record; }) => Promise; - incorporateContent: (remote: ReturnType) => Promise; }; return { disk,