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..e193801d --- /dev/null +++ b/frontend/sync-client/src/persistence/database.test.ts @@ -0,0 +1,48 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { Database, createEmptyDatabase } from "./database"; + +test("current state keeps absent optional fields absent across save and reload", async () => { + const initial = createEmptyDatabase("vault"); + initial.local = { local: "local.md", remote: "remote.md" }; + initial.documents = { + local: { observedHash: "local" }, + remote: { remote: { vaultUpdateId: 1, contentSize: 3 } }, + empty: {} + }; + let saved = structuredClone(initial); + const restart = (): Database => + new Database( + saved, + async (next) => { + saved = structuredClone(next); + }, + "vault", + async () => structuredClone(saved) + ); + const db = restart(); + assert.deepEqual(db.state, initial); + await db.save(); + assert.deepEqual(saved, initial); + assert.deepEqual(restart().state, initial); +}); + +test("current reset state is restored after an uncertain save", async () => { + const saved = createEmptyDatabase("vault"); + saved.local = { note: "note.md" }; + saved.documents.note = { + base: { vaultUpdateId: null, hash: "clean" }, + observedHash: "clean" + }; + const db = new Database( + undefined, + async () => { + throw new Error("interrupted"); + }, + "vault", + async () => structuredClone(saved) + ); + await assert.rejects(db.save(), /interrupted/); + await db.recoverPersistence(); + assert.deepEqual(db.state, saved); +}); diff --git a/frontend/sync-client/src/persistence/settings.test.ts b/frontend/sync-client/src/persistence/settings.test.ts index cff24895..915ce77e 100644 --- a/frontend/sync-client/src/persistence/settings.test.ts +++ b/frontend/sync-client/src/persistence/settings.test.ts @@ -9,7 +9,7 @@ describe("Settings logging", () => { const settings = new Settings( logger, { token: "loaded-secret" }, - async () => {} + async () => undefined ); await settings.setSettings({ token: "updated-secret" }); @@ -43,7 +43,7 @@ it("owns settings arrays instead of sharing them with callers and listeners", as const settings = new Settings( new Logger(), { ignorePatterns: original }, - async () => {} + async () => undefined ); original.push("external mutation"); assert.deepEqual(settings.getSettings().ignorePatterns, ["original"]); @@ -67,7 +67,9 @@ it("file eligibility follows successfully saved settings", async () => { maxFileSizeMB: 1 }, async () => { - if (fail) throw new Error("save failed"); + if (fail) { + throw new Error("save failed"); + } } ); assert(settings.isIgnored("private/note.md")); diff --git a/frontend/sync-client/src/services/server-config.test.ts b/frontend/sync-client/src/services/server-config.test.ts index 735d798f..410b2b67 100644 --- a/frontend/sync-client/src/services/server-config.test.ts +++ b/frontend/sync-client/src/services/server-config.test.ts @@ -1,3 +1,4 @@ +import { awaitAll } from "../utils/await-all"; import assert from "node:assert/strict"; import { test } from "node:test"; import { ServerConfig } from "./server-config"; @@ -12,28 +13,31 @@ const config = { mergeableFileExtensions: ["md"], serverVersion: "test" }; -const make = (fetch: typeof globalThis.fetch) => +const make = (fetch: typeof globalThis.fetch): ServerConfig => new ServerConfig( new SyncService( "device", new Settings( new Logger(), { remoteUri: "http://test" }, - async () => {} + async () => undefined ), fetch, - { get: () => undefined, save: async () => {} } + { get: () => undefined, save: async () => undefined } ) ); test("configuration shares one validated request and retries failures", async () => { let calls = 0; const server = make(async () => { - if (++calls === 1) throw new Error("offline"); + if (++calls === 1) { + throw new Error("offline"); + } + return Response.json(config); }); await assert.rejects(server.getConfig(), /offline/); - const [a, b] = await Promise.all([server.getConfig(), server.getConfig()]); + const [a, b] = await awaitAll([server.getConfig(), server.getConfig()]); assert.equal(a, b); assert.equal(calls, 2); await server.getConfig(); @@ -74,6 +78,7 @@ for (const fail of [false, true]) { ); await old; } + assert.equal(await server.getConfig(), current); assert.equal(calls, 2); }); diff --git a/frontend/sync-client/src/services/sync-service.test.ts b/frontend/sync-client/src/services/sync-service.test.ts index be8ae94c..f3603974 100644 --- a/frontend/sync-client/src/services/sync-service.test.ts +++ b/frontend/sync-client/src/services/sync-service.test.ts @@ -5,7 +5,7 @@ import { Logger } from "../tracing/logger"; import { SyncResetError } from "../errors/errors"; import { SyncService } from "./sync-service"; -const settings = () => +const settings = (): Settings => new Settings( new Logger(), { @@ -13,15 +13,26 @@ const settings = () => vaultName: "test", requestTimeoutMs: 1000 }, - async () => { } + async () => undefined ); -test("paused sync sends no requests, while connection checks still work", async () => { +test("paused sync sends no requests, while connection checks still work", async (): Promise => { let requests = 0; - const service = new SyncService("device", settings(), async () => { - requests++; - return Response.json({ headEventId: 0, events: [] }); - }, { get: () => undefined, save: async () => { } }); + const service = new SyncService( + "device", + settings(), + async () => { + requests++; + return Response.json( + { headEventId: 0, endEventId: 0, events: [] }, + { headers: { "X-Vault-Link-History": "0:empty" } } + ); + }, + { + get: (): undefined => undefined, + save: async (): Promise => undefined + } + ); await assert.rejects(service.getEvents(0), SyncResetError); assert.equal(requests, 0); await service.ping(); @@ -31,7 +42,7 @@ test("paused sync sends no requests, while connection checks still work", async assert.equal(requests, 2); }); -test("pause aborts requests and fences late responses from an injected fetch", async () => { +test("pause aborts requests and fences late responses from an injected fetch", async (): Promise => { const response = Promise.withResolvers(); const signals: AbortSignal[] = []; const saved: (string | undefined)[] = []; @@ -39,14 +50,18 @@ test("pause aborts requests and fences late responses from an injected fetch", a "device", settings(), async (_, init) => { - signals.push(init!.signal!); + assert.ok(init?.signal); + signals.push(init.signal); return signals.length === 1 ? response.promise - : Response.json({ headEventId: 0, events: [] }); + : Response.json( + { headEventId: 0, endEventId: 0, events: [] }, + { headers: { "X-Vault-Link-History": "0:empty" } } + ); }, { - get: () => undefined, - save: async (value) => { + get: (): undefined => undefined, + save: async (value): Promise => { saved.push(value); } } @@ -55,10 +70,10 @@ test("pause aborts requests and fences late responses from an injected fetch", a const first = service.getEvents(0); service.pause(); await assert.rejects(first, SyncResetError); - assert.equal(signals[0].aborted, true); + assert.equal(signals[0]?.aborted, true); service.resume(); await service.getEvents(0); - assert.equal(signals[1].aborted, false); + assert.equal(signals[1]?.aborted, false); response.resolve( Response.json( {}, @@ -66,53 +81,57 @@ test("pause aborts requests and fences late responses from an injected fetch", a ) ); await new Promise((resolve) => setImmediate(resolve)); - assert.deepEqual(saved, []); + assert.deepEqual(saved, ["0:empty"]); }); for (const binary of [false, true]) { - test(`pause interrupts a stalled ${binary ? "binary" : "JSON"} response body`, async () => { - const headersSaved = Promise.withResolvers(); - let body!: ReadableStreamDefaultController; + test(`pause interrupts a stalled ${binary ? "binary" : "JSON"} response body`, async (): Promise => { + const headersSaved = Promise.withResolvers(); + const bodyReady = + Promise.withResolvers< + ReadableStreamDefaultController + >(); const service = new SyncService( "device", settings(), async () => new Response( - new ReadableStream({ - start(controller) { - body = controller; + new ReadableStream({ + start(controller): void { + bodyReady.resolve(controller); } }), { headers: { "X-Vault-Link-History": "1:head" } } ), { - get: () => undefined, - save: async () => { - headersSaved.resolve(); + get: (): undefined => undefined, + save: async (): Promise => { + headersSaved.resolve(undefined); } } ); service.resume(); const request = binary ? service.getDocumentVersionContent({ - documentId: "a", - vaultUpdateId: 1 - }) + documentId: "a", + vaultUpdateId: 1 + }) : service.getEvents(0); await headersSaved.promise; await new Promise((resolve) => setImmediate(resolve)); service.pause(); await assert.rejects(request, SyncResetError); + const body = await bodyReady.promise; body.enqueue(new TextEncoder().encode("{}")); body.close(); }); } -test("pause waits for a checkpoint save already in progress", async () => { - const saving = Promise.withResolvers(); - const release = Promise.withResolvers(); +test("pause waits for a checkpoint save already in progress", async (): Promise => { + const saving = Promise.withResolvers(); + const release = Promise.withResolvers(); let settled = false; - let checkpoint: string | undefined; + let checkpoint: string | undefined = undefined; const service = new SyncService( "device", settings(), @@ -124,9 +143,9 @@ test("pause waits for a checkpoint save already in progress", async () => { } ), { - get: () => checkpoint, - save: async (value) => { - saving.resolve(); + get: (): string | undefined => checkpoint, + save: async (value): Promise => { + saving.resolve(undefined); await release.promise; checkpoint = value; } @@ -140,13 +159,13 @@ test("pause waits for a checkpoint save already in progress", async () => { service.pause(); await new Promise((resolve) => setImmediate(resolve)); assert.equal(settled, false); - release.resolve(); + release.resolve(undefined); await assert.rejects(request, SyncResetError); assert.equal(checkpoint, "1:head"); }); -test("history saves only new checkpoints, including a changed token at the same event ID", async () => { - let checkpoint: string | undefined; +test("history saves only new checkpoints, including a changed token at the same event ID", async (): Promise => { + let checkpoint: string | undefined = undefined; let responseCheckpoint = "1:first"; const saved: (string | undefined)[] = []; const service = new SyncService( @@ -158,8 +177,8 @@ test("history saves only new checkpoints, including a changed token at the same { headers: { "X-Vault-Link-History": responseCheckpoint } } ), { - get: () => checkpoint, - save: async (value) => { + get: (): string | undefined => checkpoint, + save: async (value): Promise => { checkpoint = value; saved.push(value); } @@ -177,5 +196,36 @@ test("history saves only new checkpoints, including a changed token at the same responseCheckpoint = value; await service.getEvents(0); } + assert.deepEqual(saved, ["1:first", "2:next", "2:replacement"]); }); + +for (const checkpoint of [undefined, ""]) { + test(`successful sync responses reject ${checkpoint === undefined ? "missing" : "empty"} history checkpoints`, async () => { + const service = new SyncService( + "device", + settings(), + async () => + Response.json( + { headEventId: 0, endEventId: 0, events: [] }, + { + headers: + checkpoint === undefined + ? {} + : { "X-Vault-Link-History": checkpoint } + } + ), + { + get: (): undefined => undefined, + save: async (): Promise => { + assert.fail("must not save a missing checkpoint"); + } + } + ); + service.resume(); + await assert.rejects( + service.getEvents(0), + /Missing server history checkpoint/ + ); + }); +} diff --git a/frontend/sync-client/src/services/websocket-manager.test.ts b/frontend/sync-client/src/services/websocket-manager.test.ts index 969a02eb..587c7da0 100644 --- a/frontend/sync-client/src/services/websocket-manager.test.ts +++ b/frontend/sync-client/src/services/websocket-manager.test.ts @@ -1,9 +1,11 @@ /* eslint-disable @typescript-eslint/no-unsafe-type-assertion */ import { describe, it, beforeEach } from "node:test"; import assert from "node:assert"; +import { createServer } from "node:net"; import { WebSocketManager } from "./websocket-manager"; -import type { Logger } from "../tracing/logger"; -import type { Settings } from "../persistence/settings"; +import { 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 const WebSocket = require("ws") as typeof globalThis.WebSocket; @@ -39,7 +41,7 @@ class MockWebSocket { public sentMessages: string[] = []; - public constructor(public url: string) { + public constructor(public url: string | URL) { setTimeout(() => { if (this.readyState === WebSocket.CONNECTING) { this.readyState = WebSocket.OPEN; @@ -52,6 +54,7 @@ class MockWebSocket { if (this.readyState !== WebSocket.OPEN) { throw new Error("WebSocket is not open"); } + this.sentMessages.push(data); } @@ -98,6 +101,7 @@ describe("WebSocketManager", () => { const noop = (): void => { // Intentionally empty for mock }; + mockLogger = { info: createMockFn(noop), warn: createMockFn(noop), @@ -129,30 +133,24 @@ describe("WebSocketManager", () => { await new Promise((resolve) => setTimeout(resolve, 50)); const { outstandingPromises } = manager as unknown as { - outstandingPromises: Promise[]; + outstandingPromises: Set>; }; const mockWs = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); await new Promise((resolve) => setTimeout(resolve, 100)); - assert.strictEqual(outstandingPromises.length, 0); + assert.strictEqual(outstandingPromises.size, 0); await manager.stop(); }); @@ -171,7 +169,7 @@ describe("WebSocketManager", () => { await new Promise((resolve) => setTimeout(resolve, 50)); const { outstandingPromises } = manager as unknown as { - outstandingPromises: Promise[]; + outstandingPromises: Set>; }; const mockWs = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; @@ -182,7 +180,7 @@ describe("WebSocketManager", () => { }); await new Promise((resolve) => setTimeout(resolve, 100)); - assert.strictEqual(outstandingPromises.length, 0); + assert.strictEqual(outstandingPromises.size, 0); await manager.stop(); }); @@ -203,19 +201,22 @@ describe("WebSocketManager", () => { throw new Error("Buffer full"); }; - assert.throws(() => { - manager.sendHandshakeMessage({ - type: "handshake", - token: "test", - deviceId: "test", - lastSeenVaultUpdateId: null - }); - }); + assert.doesNotThrow(() => mockWs.onopen?.(new Event("open"))); + assert.equal(manager.isWebSocketConnected, false); + assert( + ( + mockLogger.warn as typeof mockLogger.warn & { + calls: [string][]; + } + ).calls.some(([message]) => + message.includes("WebSocket handshake failed") + ) + ); await manager.stop(); }); - it("completes stop with timeout protection", async () => { + it("completes stop after retiring the socket", async () => { const manager = new WebSocketManager( deviceId, mockLogger, @@ -227,7 +228,7 @@ describe("WebSocketManager", () => { await new Promise((resolve) => setTimeout(resolve, 50)); await manager.stop(); - assert.ok(true); + assert.equal(manager.isWebSocketConnected, false); }); it("uses a secure URL, preserves the configured prefix, and encodes the vault", async () => { @@ -316,24 +317,351 @@ describe("WebSocketManager", () => { const mockWs = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); await new Promise((resolve) => setTimeout(resolve, 10)); const { outstandingPromises } = manager as unknown as { - outstandingPromises: Promise[]; + outstandingPromises: Set>; }; - assert.ok(outstandingPromises.length > 0); + assert.ok(outstandingPromises.size > 0); // eslint-disable-next-line @typescript-eslint/no-non-null-assertion resolveListener!(); await new Promise((resolve) => setTimeout(resolve, 50)); - assert.strictEqual(outstandingPromises.length, 0); + assert.strictEqual(outstandingPromises.size, 0); await manager.stop(); }); }); + +it( + "refused WebSocket connections and replacing a connecting socket do not emit unhandled errors", + { timeout: 5_000 }, + async () => { + const server = createServer(); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", resolve) + ); + const address = server.address(); + assert(address !== null && typeof address !== "string"); + await new Promise((resolve, reject) => + server.close((error) => { + if (error !== undefined) { + reject(error); + } else { + resolve(); + } + }) + ); + const logger = new Logger(); + const settings = new Settings( + logger, + { + remoteUri: `http://127.0.0.1:${address.port}`, + webSocketRetryIntervalMs: 10 + }, + async () => undefined + ); + const manager = new WebSocketManager( + "test", + logger, + settings, + WebSocket + ); + let closes = 0; + const disconnected = Promise.withResolvers(); + manager.onWebSocketStatusChanged.add((connected) => { + if (!connected && ++closes >= 2) { + disconnected.resolve(undefined); + } + }); + try { + manager.start(); + manager.start(); + await disconnected.promise; + assert.equal(manager.isWebSocketConnected, false); + } finally { + await manager.stop(); + } + } +); + +for (const failedAttempt of [1, 2]) { + it(`recovers from a WebSocket constructor failure on attempt ${failedAttempt}`, async (context) => { + context.mock.timers.enable({ apis: ["setTimeout"] }); + let attempts = 0; + class FailingSocket extends MockWebSocket { + public static OPEN = 1; + public constructor(url: string) { + if (++attempts === failedAttempt) { + throw new Error("socket allocation failed"); + } + + super(url); + } + } + const logger = new Logger(); + const settings = new Settings( + logger, + { remoteUri: "http://test", webSocketRetryIntervalMs: 1000 }, + async () => undefined + ); + const manager = new WebSocketManager( + "test", + logger, + settings, + FailingSocket as unknown as typeof WebSocket + ); + try { + assert.doesNotThrow(() => { + manager.start(); + }); + context.mock.timers.tick(0); + if (failedAttempt === 2) { + const socket = ( + manager as unknown as { webSocket: MockWebSocket } + ).webSocket; + socket.close(); + } + + assert.doesNotThrow(() => { + context.mock.timers.tick(1000); + }); + context.mock.timers.tick(1000); + assert.equal(attempts, failedAttempt + 1); + assert.equal(manager.isWebSocketConnected, true); + } finally { + await manager.stop(); + } + }); +} + +for (const closeBehavior of ["throws", "stays open"]) { + it(`stop retires a socket even when close ${closeBehavior}`, async (context) => { + context.mock.timers.enable({ apis: ["setTimeout"] }); + class BrokenCloseSocket extends MockWebSocket { + public static OPEN = 1; + public close(): void { + if (closeBehavior === "throws") { + throw new Error("close failed"); + } + } + } + const logger = new Logger(); + const settings = new Settings( + logger, + { remoteUri: "http://test" }, + async () => undefined + ); + const manager = new WebSocketManager( + "test", + logger, + settings, + BrokenCloseSocket as unknown as typeof WebSocket + ); + let connected = false, + delivered = 0; + manager.onWebSocketStatusChanged.add((status) => { + connected = status; + }); + manager.onRemoteVaultUpdateReceived.add(async () => { + delivered++; + }); + manager.start(); + context.mock.timers.tick(0); + const socket = (manager as unknown as { webSocket: MockWebSocket }) + .webSocket; + const delayedMessage = socket.onmessage; + await manager.stop(); // No close event or timer advance is needed. + assert.equal(manager.isWebSocketConnected, false); + assert.equal(connected, false); + socket.simulateMessage({ + type: "vaultChanged" + }); + delayedMessage?.( + new MockMessageEvent("message", { + data: JSON.stringify({ type: "vaultChanged" }) + }) + ); + await manager.waitUntilFinished(); + assert.equal(delivered, 0); + }); +} + +it("stop retires the socket immediately but drains messages already being handled", async (context) => { + context.mock.timers.enable({ apis: ["setTimeout"] }); + const logger = new Logger(); + const manager = new WebSocketManager( + "device", + logger, + new Settings( + logger, + { remoteUri: "http://test", token: "test-token" }, + async () => undefined + ), + MockWebSocket as unknown as typeof WebSocket + ); + const release = Promise.withResolvers(); + context.after(async () => { + release.resolve(undefined); + await manager.stop(); + }); + manager.start(); + context.mock.timers.tick(0); + const socket = (manager as unknown as { webSocket: MockWebSocket }) + .webSocket; + assert.deepEqual(JSON.parse(socket.sentMessages[0] ?? ""), { + type: "handshake", + token: "test-token", + deviceId: "device" + }); + manager.onRemoteVaultUpdateReceived.add(async () => release.promise); + socket.simulateMessage({ type: "vaultChanged" }); + let finished = false; + const stopping = manager.stop().then(() => { + finished = true; + }); + assert.equal(manager.isWebSocketConnected, false); + assert.equal(socket.onmessage, null); + await Promise.resolve(); + assert(!finished); + release.resolve(undefined); + await stopping; + assert(finished); +}); + +it("a handshake send failure retires the connection without escaping the open callback", async (context) => { + context.mock.timers.enable({ apis: ["setTimeout"] }); + class BrokenSendSocket extends MockWebSocket { + public static OPEN = 1; + public send(): void { + throw new Error("transport closed during open"); + } + } + const logger = new Logger(); + const settings = new Settings( + logger, + { remoteUri: "http://test" }, + async () => undefined + ); + const manager = new WebSocketManager( + "test", + logger, + settings, + BrokenSendSocket as unknown as typeof WebSocket + ); + const statuses: boolean[] = []; + manager.onWebSocketStatusChanged.add((status) => statuses.push(status)); + try { + manager.start(); + assert.doesNotThrow(() => { + context.mock.timers.tick(0); + }); + assert.equal(manager.isWebSocketConnected, false); + assert.equal(statuses.at(-1), false); + assert( + !statuses.includes(true), + "a failed handshake must never announce a connection" + ); + } finally { + 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 { + public static OPEN = 1; + public constructor(url: string) { + super(url); + attempts++; + } + } + const logger = new Logger(); + const settings = new Settings( + logger, + { remoteUri: "http://test", webSocketRetryIntervalMs: 1000 }, + async () => undefined + ); + 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 { + public static OPEN = 1; + public constructor(url: string) { + super(url); + attempts++; + } + } + const logger = new Logger(); + const settings = new Settings( + logger, + { remoteUri: "http://test", webSocketRetryIntervalMs: 1000 }, + async () => undefined + ); + 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/sync-client.test.ts b/frontend/sync-client/src/sync-client.test.ts new file mode 100644 index 00000000..14ffc4db --- /dev/null +++ b/frontend/sync-client/src/sync-client.test.ts @@ -0,0 +1,41 @@ +/* eslint-disable @typescript-eslint/no-unsafe-type-assertion -- No filesystem operation is reached before initialization. */ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { SyncClient } from "./sync-client"; +import type { FileSystemOperations } from "./file-operations/filesystem-operations"; +import { awaitAll } from "./utils/await-all"; + +test("destroy during startup metadata work cannot restart the client", async () => { + const saving = Promise.withResolvers(); + const release = Promise.withResolvers(); + const client = await SyncClient.create({ + fs: {} as FileSystemOperations, + persistence: { + load: async () => ({ + settings: { + isSyncEnabled: true, + enableTelemetry: false, + remoteUri: "http://test" + } + }), + save: async () => { + saving.resolve(undefined); + await release.promise; + } + }, + fetch: async () => { + assert.fail("a destroyed client must not start network requests"); + } + }); + + const notification = client.syncLocallyCreatedFile("note.md"); + await saving.promise; + const starting = client.start(); + // Let start acquire the lifecycle lock and begin draining notification saves. + await Promise.resolve(); + const destroying = client.destroy(); + release.resolve(undefined); + await awaitAll([notification, starting, destroying]); + + await assert.rejects(client.start(), /destroyed/u); +}); diff --git a/frontend/sync-client/src/sync-operations/content.test.ts b/frontend/sync-client/src/sync-operations/content.test.ts index db446097..6a194e3a 100644 --- a/frontend/sync-client/src/sync-operations/content.test.ts +++ b/frontend/sync-client/src/sync-operations/content.test.ts @@ -1,26 +1,28 @@ import assert from "node:assert/strict"; import { describe, it } from "node:test"; -import { mergeContent, fromStoredSnapshot, toStoredSnapshot } from "./content"; +import { hashSnapshot, type HashedSnapshot } from "../snapshot"; +import { mergeContent } from "./content"; const extensions = ["md"]; -const snapshot = async (text: string) => - toStoredSnapshot({ content: new TextEncoder().encode(text) }); +const createSnapshot = async (text: string): Promise => + hashSnapshot({ content: new TextEncoder().encode(text) }); describe("content reconciliation without file timestamps", () => { - it("round-trips content and editor selections without timestamps", async () => { + it("hashes snapshots without changing bytes or editor selections", async () => { const original = { content: new Uint8Array([0, 255, 13, 10]), cursors: [] }; - assert.deepEqual( - fromStoredSnapshot(await toStoredSnapshot(original)), - original - ); + const snapshot = await hashSnapshot(original); + assert.deepEqual(snapshot.content, original.content); + assert.deepEqual(snapshot.cursors, original.cursors); + original.content.fill(1); + assert.deepEqual(snapshot.content, new Uint8Array([0, 255, 13, 10])); }); it("keeps local binary edits when the server content is unchanged", async () => { - const base = await snapshot("\0base"); - const local = await snapshot("\0local"); + const base = await createSnapshot("\0base"); + const local = await createSnapshot("\0local"); const merged = await mergeContent( "file.bin", base, @@ -28,12 +30,12 @@ describe("content reconciliation without file timestamps", () => { base, extensions ); - assert.equal(merged.contentBase64, local.contentBase64); + assert.deepEqual(merged.content, local.content); }); it("incorporates server binary edits when local content is unchanged", async () => { - const base = await snapshot("\0base"); - const remote = await snapshot("\0remote"); + const base = await createSnapshot("\0base"); + const remote = await createSnapshot("\0remote"); const merged = await mergeContent( "file.bin", base, @@ -41,7 +43,7 @@ describe("content reconciliation without file timestamps", () => { remote, extensions ); - assert.equal(merged.contentBase64, remote.contentBase64); + assert.deepEqual(merged.content, remote.content); }); it("keeps the server on concurrent binary or unmergeable text edits", async () => { @@ -49,10 +51,10 @@ describe("content reconciliation without file timestamps", () => { ["file.bin", "\0"], ["file.md", "\0"], ["file.json", ""] - ]) { - const base = await snapshot(`${prefix}base`); - const local = await snapshot(`${prefix}local`); - const remote = await snapshot(`${prefix}remote`); + ] as const) { + const base = await createSnapshot(`${prefix}base`); + const local = await createSnapshot(`${prefix}local`); + const remote = await createSnapshot(`${prefix}remote`); const merged = await mergeContent( path, base, @@ -60,13 +62,13 @@ describe("content reconciliation without file timestamps", () => { remote, extensions ); - assert.equal(merged.contentBase64, remote.contentBase64, path); + assert.deepEqual(merged.content, remote.content, path); } }); it("keeps the server for differing binary content without a common base", async () => { - const local = await snapshot("\0local"); - const remote = await snapshot("\0remote"); + const local = await createSnapshot("\0local"); + const remote = await createSnapshot("\0remote"); const merged = await mergeContent( "file.bin", undefined, @@ -74,19 +76,23 @@ describe("content reconciliation without file timestamps", () => { remote, extensions ); - assert.equal(merged.contentBase64, remote.contentBase64); + assert.deepEqual(merged.content, remote.content); }); it("still merges independent text edits from both sides", async () => { const merged = await mergeContent( "note.md", - await snapshot("First paragraph.\n\nLast paragraph.\n"), - await snapshot("Updated first paragraph.\n\nLast paragraph.\n"), - await snapshot("First paragraph.\n\nUpdated last paragraph.\n"), + await createSnapshot("First paragraph.\n\nLast paragraph.\n"), + await createSnapshot( + "Updated first paragraph.\n\nLast paragraph.\n" + ), + await createSnapshot( + "First paragraph.\n\nUpdated last paragraph.\n" + ), extensions ); assert.equal( - new TextDecoder().decode(fromStoredSnapshot(merged).content), + new TextDecoder().decode(merged.content), "Updated first paragraph.\n\nUpdated last paragraph.\n" ); }); diff --git a/frontend/sync-client/src/sync-operations/cursor-tracker.test.ts b/frontend/sync-client/src/sync-operations/cursor-tracker.test.ts index 01eac85a..23377cf8 100644 --- a/frontend/sync-client/src/sync-operations/cursor-tracker.test.ts +++ b/frontend/sync-client/src/sync-operations/cursor-tracker.test.ts @@ -1,67 +1,304 @@ /* eslint-disable @typescript-eslint/no-unsafe-type-assertion */ import assert from "node:assert/strict"; import { it } from "node:test"; -import type { Database } from "../persistence/database"; -import type { FileOperations } from "../file-operations/file-operations"; +import { + Database, + createEmptyDatabase, + type RelativePath, + type VaultUpdateId +} from "../persistence/database"; +import type { FileSystemOperations } from "../file-operations/filesystem-operations"; import type { WebSocketManager } from "../services/websocket-manager"; import type { ClientCursors } from "../services/types/ClientCursors"; -import { DocumentUpToDateness } from "../types/document-up-to-dateness"; import { EventListeners } from "../utils/data-structures/event-listeners"; import { hash } from "../utils/hash"; import { CursorTracker } from "./cursor-tracker"; -import { FileChangeNotifier } from "./file-change-notifier"; -it("retains future cursors and publishes them when content catches up", async () => { +function createDatabase(digest: string): Database { + const state = createEmptyDatabase("test"); + state.local.document = "note.md"; + state.documents.document = { base: { vaultUpdateId: 1, hash: digest } }; + return new Database( + state, + async () => undefined, + "test", + async () => undefined + ); +} + +it("keeps the last usable cursor while retaining only the latest future update", async (context) => { const content = new TextEncoder().encode("current"); const contentHash = await hash(content); - const database = { - getDocumentByDocumentId: () => ({ - documentId: "document", - relativePath: "note.md", - isDeleted: false, - metadata: { parentVersionId: 2, hash: contentHash } - }) - } as unknown as Database; + const database = createDatabase(contentHash); const websocket = { + onWebSocketStatusChanged: new EventListeners(), onRemoteCursorsUpdateReceived: new EventListeners< (cursors: ClientCursors[]) => Promise >(), - updateLocalCursors: () => {} + updateLocalCursors: () => undefined } as unknown as WebSocketManager; - const files = { read: async () => content } as unknown as FileOperations; - const notifier = new FileChangeNotifier(); - const tracker = new CursorTracker(database, websocket, files, notifier); - const future = { + const notifier = new EventListeners<(path: RelativePath) => unknown>(); + const tracker = new CursorTracker( + database, + websocket, + { read: async () => ({ content }) } as unknown as FileSystemOperations, + notifier + ); + context.after(() => { + tracker.reset(); + }); + let received: ClientCursors[] = []; + tracker.onRemoteCursorsUpdated.add((cursors) => { + received = cursors; + }); + const cursor = (version: number): ClientCursors => ({ userName: "Other", deviceId: "device", documentsWithCursors: [ { document_id: "document", relative_path: "old-name.md", - vault_update_id: 2, - cursors: [{ start: 0, end: 1 }] + vault_update_id: version, + cursors: [{ start: version, end: version }] } - ], - upToDateness: DocumentUpToDateness.Later - }; - const internals = tracker as unknown as { - knownRemoteCursors: (typeof future)[]; - getRelevantAndPruneKnownClientCursors: () => ClientCursors[]; - }; - internals.knownRemoteCursors = [future]; - - assert.deepEqual(internals.getRelevantAndPruneKnownClientCursors(), []); - assert.strictEqual(internals.knownRemoteCursors.length, 1); - - let received: ClientCursors[] = []; - tracker.onRemoteCursorsUpdated.add((cursors) => { - received = cursors; + ] }); - await notifier.onFileChanged.triggerAsync("note.md"); + for (const remoteVersion of [1, 2, 3]) { + await websocket.onRemoteCursorsUpdateReceived.triggerAsync([ + cursor(remoteVersion) + ]); + assert.equal(received[0]?.documentsWithCursors[0]?.vault_update_id, 1); + } - assert.strictEqual(received.length, 1); - assert.strictEqual( - received[0].documentsWithCursors[0].relative_path, - "note.md" + assert.ok(database.state.documents.document); + database.state.documents.document.base = { + vaultUpdateId: 2, + hash: contentHash + }; + await notifier.triggerAsync("note.md"); + assert.equal( + received[0]?.documentsWithCursors[0]?.vault_update_id, + 1, + "superseded future cursors must not reappear" ); + database.state.documents.document.base = { + vaultUpdateId: 3, + hash: contentHash + }; + await notifier.triggerAsync("note.md"); + assert.equal(received[0].documentsWithCursors[0].vault_update_id, 3); + assert.equal(received[0].documentsWithCursors[0].relative_path, "note.md"); + await websocket.onRemoteCursorsUpdateReceived.triggerAsync([]); + assert.deepEqual(received, []); +}); + +it( + "refreshes unchanged cursors on reconnect and heartbeat, and follows the document through renames", + { timeout: 5_000 }, + async (context) => { + context.mock.timers.enable({ apis: ["setInterval"] }); + const content = new TextEncoder().encode("hello"); + let path = "note.md"; + const database = createDatabase(await hash(content)); + const sent: { + documentsWithCursors: { + relative_path: RelativePath; + vault_update_id: VaultUpdateId | null; + }[]; + }[] = []; + let published = Promise.withResolvers(); + const websocket = { + onWebSocketStatusChanged: new EventListeners< + (connected: boolean) => unknown + >(), + onRemoteCursorsUpdateReceived: new EventListeners(), + updateLocalCursors: (value: (typeof sent)[number]) => { + sent.push(value); + published.resolve(undefined); + } + } as unknown as WebSocketManager; + const notifier = new EventListeners<(path: RelativePath) => unknown>(); + const read = context.mock.fn(async () => ({ content })); + const tracker = new CursorTracker( + database, + websocket, + { read } as unknown as FileSystemOperations, + notifier + ); + const waitForSend = async (trigger: () => void): Promise => { + published = Promise.withResolvers(); + trigger(); + // Hashing finishes outside the event loop; a fixed number of turns + // cannot guarantee that the cursor snapshot has been published. + await published.promise; + }; + + try { + await tracker.sendLocalCursorsToServer({ + "note.md": [{ start: 1, end: 2 }] + }); + assert.equal(sent.length, 1); + await tracker.sendLocalCursorsToServer({ + "note.md": [{ start: 1, end: 2 }] + }); + assert.equal(sent.length, 1); + await waitForSend(() => { + websocket.onWebSocketStatusChanged.trigger(true); + }); + assert.equal(sent.length, 2); + await waitForSend(() => { + context.mock.timers.tick(15_000); + }); + assert.equal(sent.length, 3); + path = "renamed.md"; + database.state.local.document = path; + await notifier.triggerAsync(path); + assert.equal( + sent.at(-1)?.documentsWithCursors[0]?.relative_path, + path + ); + websocket.onWebSocketStatusChanged.trigger(false); + const count = Number(sent.length); + const readCount = read.mock.callCount(); + context.mock.timers.tick(30_000); + // A surviving heartbeat would start a snapshot read in the send lock's + // next microtask, even if hashing that snapshot has not finished yet. + await Promise.resolve(); + assert.equal(read.mock.callCount(), readCount); + assert.equal(sent.length, count); + await waitForSend(() => { + websocket.onWebSocketStatusChanged.trigger(true); + }); + assert.equal(sent.length, count + 1); + } finally { + tracker.reset(); + context.mock.timers.reset(); + } + } +); + +for (const interruption of ["disconnect", "reset"] as const) { + it(`an in-flight cursor read cannot resurrect presence after ${interruption}`, async () => { + const content = new TextEncoder().encode("current"); + const contentHash = await hash(content); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const database = createDatabase(contentHash); + const websocket = { + onWebSocketStatusChanged: new EventListeners< + (connected: boolean) => unknown + >(), + onRemoteCursorsUpdateReceived: new EventListeners< + (cursors: ClientCursors[]) => Promise + >(), + updateLocalCursors: () => undefined + } as unknown as WebSocketManager; + const tracker = new CursorTracker( + database, + websocket, + { + read: async () => { + entered.resolve(undefined); + await release.promise; + return { content }; + } + } as unknown as FileSystemOperations, + new EventListeners<(path: RelativePath) => unknown>() + ); + let received: ClientCursors[] = []; + tracker.onRemoteCursorsUpdated.add((cursors) => { + received = cursors; + }); + const cursor: ClientCursors = { + userName: "Other", + deviceId: "device", + documentsWithCursors: [ + { + document_id: "document", + relative_path: "note.md", + vault_update_id: 1, + cursors: [{ start: 0, end: 1 }] + } + ] + }; + const first = websocket.onRemoteCursorsUpdateReceived.triggerAsync([ + cursor + ]); + await entered.promise; + const queued = websocket.onRemoteCursorsUpdateReceived.triggerAsync([ + cursor + ]); + if (interruption === "reset") { + tracker.reset(); + } else { + websocket.onWebSocketStatusChanged.trigger(false); + } + + assert.deepEqual(received, []); + release.resolve(undefined); + await first; + await queued; + assert.deepEqual( + received, + [], + "a retired connection must not restore its cursors" + ); + tracker.reset(); + }); +} + +it("a remote cursor becomes outdated if the document advances during its snapshot read", async () => { + const content = new TextEncoder().encode("old"); + const database = createDatabase(await hash(content)); + const entered = Promise.withResolvers(), + release = Promise.withResolvers(); + const websocket = { + onWebSocketStatusChanged: new EventListeners< + (connected: boolean) => unknown + >(), + onRemoteCursorsUpdateReceived: new EventListeners< + (cursors: ClientCursors[]) => Promise + >(), + updateLocalCursors: () => undefined + } as unknown as WebSocketManager; + const tracker = new CursorTracker( + database, + websocket, + { + read: async () => { + entered.resolve(undefined); + await release.promise; + return { content }; + } + } as unknown as FileSystemOperations, + new EventListeners<(path: RelativePath) => unknown>() + ); + let outdated: boolean | undefined = undefined; + tracker.onRemoteCursorsUpdated.add((cursors) => { + outdated = cursors[0]?.isOutdated; + }); + const update = websocket.onRemoteCursorsUpdateReceived.triggerAsync([ + { + userName: "Other", + deviceId: "device", + documentsWithCursors: [ + { + document_id: "document", + relative_path: "note.md", + vault_update_id: 1, + cursors: [{ start: 0, end: 1 }] + } + ] + } + ]); + await entered.promise; + assert.ok(database.state.documents.document); + database.state.documents.document.base = { + vaultUpdateId: 2, + hash: await hash(new TextEncoder().encode("new content")) + }; + release.resolve(undefined); + await update; + assert.equal(outdated, true); + tracker.reset(); }); diff --git a/frontend/sync-client/src/sync-operations/outgoing-cursor.test.ts b/frontend/sync-client/src/sync-operations/outgoing-cursor.test.ts new file mode 100644 index 00000000..90119c8f --- /dev/null +++ b/frontend/sync-client/src/sync-operations/outgoing-cursor.test.ts @@ -0,0 +1,71 @@ +/* eslint-disable @typescript-eslint/no-unsafe-type-assertion -- Test doubles implement only the exercised adapter methods. */ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { CursorTracker } from "./cursor-tracker"; +import { EventListeners } from "../utils/data-structures/event-listeners"; +import { hash } from "../utils/hash"; +import { + Database, + createEmptyDatabase, + type RelativePath +} from "../persistence/database"; +import type { FileSystemOperations } from "../file-operations/filesystem-operations"; +import type { WebSocketManager } from "../services/websocket-manager"; +import type { DocumentWithCursors } from "../services/types/DocumentWithCursors"; + +for (const change of ["version", "identity", "path", "delete"] as const) { + test(`outgoing cursor is dirty when ${change} changes during its snapshot read`, async (): Promise => { + // Equal bytes deliberately defeat a freshness check that compares only hashes. + const content = new TextEncoder().encode("same bytes"); + const digest = await hash(content); + const state = createEmptyDatabase("test"); + state.local.a = "a.md"; + state.documents.a = { base: { vaultUpdateId: 1, hash: digest } }; + const database = new Database( + state, + async () => undefined, + "test", + async () => undefined + ); + const sent: { documentsWithCursors: DocumentWithCursors[] }[] = []; + const websocket = { + onWebSocketStatusChanged: new EventListeners(), + onRemoteCursorsUpdateReceived: new EventListeners(), + updateLocalCursors: (data: { + documentsWithCursors: DocumentWithCursors[]; + }): number => sent.push(data) + }; + const files = { + read: async (): Promise<{ content: Uint8Array }> => { + const current = database.state; + assert.ok(current.documents.a?.base); + assert.ok(current.local.a !== undefined); + if (change === "delete") { + delete current.local.a; + } else if (change === "version") { + current.documents.a.base.vaultUpdateId = 2; + } else if (change === "identity") { + current.local.replacement = current.local.a; + current.documents.replacement = current.documents.a; + delete current.local.a; + } else { + current.local.a = "moved.md"; + } + + return { content }; + } + }; + const tracker = new CursorTracker( + database, + websocket as unknown as WebSocketManager, + files as unknown as FileSystemOperations, + new EventListeners<(path: RelativePath) => unknown>() + ); + await tracker.sendLocalCursorsToServer({ + "a.md": [{ start: 4, end: 4 }] + }); + assert.equal(sent.length, 1); + assert.equal(sent[0]?.documentsWithCursors[0]?.vault_update_id, null); + tracker.reset(); + }); +} diff --git a/frontend/sync-client/src/sync-operations/sync-scheduler.test.ts b/frontend/sync-client/src/sync-operations/sync-scheduler.test.ts new file mode 100644 index 00000000..2103956d --- /dev/null +++ b/frontend/sync-client/src/sync-operations/sync-scheduler.test.ts @@ -0,0 +1,138 @@ +import assert from "node:assert/strict"; +import { afterEach, beforeEach, describe, it, mock } from "node:test"; +import { Settings, type SyncSettings } from "../persistence/settings"; +import { Logger } from "../tracing/logger"; +import { PermanentSyncError } from "../errors/errors"; +import { SyncPhase, SyncScheduler } from "./sync-scheduler"; + +function createScheduler( + overrides: Partial = {}, + pass: () => Promise = async () => undefined +): { scheduler: SyncScheduler; runCount: () => number } { + let runs = 0; + const scheduler = new SyncScheduler( + new Settings(new Logger(), overrides, async () => undefined), + async () => { + do { + scheduler.beginPass(); + runs++; + await pass(); + } while (scheduler.needsAnotherPass); + }, + () => undefined, + () => false + ); + return { scheduler, runCount: () => runs }; +} + +describe("Sync scheduler transitions", () => { + beforeEach(() => { + mock.timers.enable({ apis: ["setTimeout"] }); + }); + afterEach(() => { + mock.timers.reset(); + }); + + for (const interval of [undefined, 0]) { + it(`leaves polling disabled for interval ${interval}`, async () => { + const { scheduler, runCount } = createScheduler({ + syncIntervalMs: interval + }); + scheduler.start(); + await scheduler.waitUntilFinished(); + assert.equal(scheduler.currentPhase, SyncPhase.Idle); + mock.timers.tick(60_000); + assert.equal(runCount(), 1); + await scheduler.stop(); + assert.equal(scheduler.currentPhase, SyncPhase.Stopped); + }); + } + + it("polls after the configured interval and cancels the timer on stop", async () => { + const { scheduler, runCount } = createScheduler({ syncIntervalMs: 25 }); + scheduler.start(); + await scheduler.waitUntilFinished(); + mock.timers.tick(24); + assert.equal(runCount(), 1); + mock.timers.tick(1); + assert.equal(runCount(), 2); + await scheduler.waitUntilFinished(); + await scheduler.stop(); + mock.timers.tick(25); + assert.equal(runCount(), 2); + }); + + it("coalesces notifications during a pass without running concurrently", async () => { + const gate = Promise.withResolvers(); + const { scheduler, runCount } = createScheduler( + {}, + async () => gate.promise + ); + scheduler.start(); + assert.equal(scheduler.currentPhase, SyncPhase.Running); + scheduler.requestSync(); + scheduler.requestSync(); + assert.equal(scheduler.currentPhase, SyncPhase.RunRequested); + assert.equal(runCount(), 1); + gate.resolve(undefined); + await scheduler.waitUntilFinished(); + assert.equal(runCount(), 2); + assert.equal(scheduler.currentPhase, SyncPhase.Idle); + await scheduler.stop(); + }); + + it("drains an active pass on stop and discards queued notifications", async () => { + const gate = Promise.withResolvers(); + const { scheduler, runCount } = createScheduler( + {}, + async () => gate.promise + ); + scheduler.start(); + scheduler.requestSync(); + const stopped = scheduler.stop(); + assert.equal(scheduler.currentPhase, SyncPhase.Stopping); + scheduler.requestSync(); + gate.resolve(undefined); + await stopped; + assert.equal(scheduler.currentPhase, SyncPhase.Stopped); + assert.equal(runCount(), 1); + }); + + it("retries transient failures and returns to idle on success", async () => { + let attempts = 0; + const { scheduler } = createScheduler( + { networkRetryIntervalMs: 10 }, + async () => { + if (++attempts === 1) { + throw new Error("offline"); + } + } + ); + scheduler.start(); + await assert.rejects(scheduler.waitUntilFinished(), /offline/u); + assert.equal(scheduler.currentPhase, SyncPhase.WaitingToRetry); + mock.timers.tick(10); + await scheduler.waitUntilFinished(); + assert.equal(scheduler.currentPhase, SyncPhase.Idle); + assert.equal(attempts, 2); + await scheduler.stop(); + }); + + it("blocks automatic retries of permanent errors but accepts an explicit request", async () => { + const { scheduler, runCount } = createScheduler( + { networkRetryIntervalMs: 10 }, + async () => { + throw new PermanentSyncError("rejected"); + } + ); + scheduler.start(); + await assert.rejects(scheduler.waitUntilFinished(), /rejected/u); + assert.equal(scheduler.currentPhase, SyncPhase.Blocked); + mock.timers.tick(100); + assert.equal(runCount(), 1); + scheduler.requestSync(); + await assert.rejects(scheduler.waitUntilFinished(), /rejected/u); + assert.equal(runCount(), 2); + await scheduler.stop(); + }); +}); diff --git a/frontend/sync-client/src/sync-operations/syncer.test.ts b/frontend/sync-client/src/sync-operations/syncer.test.ts new file mode 100644 index 00000000..5b6b53f2 --- /dev/null +++ b/frontend/sync-client/src/sync-operations/syncer.test.ts @@ -0,0 +1,177 @@ +import { ContentSync, selectContentHead } from "./content-sync"; +/* eslint-disable @typescript-eslint/no-unsafe-type-assertion */ +import assert from "node:assert"; +import { describe, it } from "node:test"; +import type { Database, RelativePath } from "../persistence/database"; +import { Settings } from "../persistence/settings"; +import type { FileOperations } from "../file-operations/file-operations"; +import type { SyncService } from "../services/sync-service"; +import type { ServerConfig } from "../services/server-config"; +import type { WebSocketManager } from "../services/websocket-manager"; +import { Logger } from "../tracing/logger"; +import type { SyncHistory } from "../tracing/sync-history"; +import { EventListeners } from "../utils/data-structures/event-listeners"; +import { Syncer } from "./syncer"; +import { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache"; +import { hashSnapshot } from "../snapshot"; +import { undiff } from "reconcile-text"; + +describe("Syncer per-file state and event delivery", () => { + const head = { + vaultUpdateId: 1, + documentId: "document", + updatedDate: "2026-01-01T00:00:00Z", + userId: "user", + deviceId: "device", + contentSize: 4 + }; + + function fixture(): { + database: Database; + websocket: WebSocketManager; + syncer: Syncer; + } { + const database = { + state: { + local: { document: "note.md" }, + documents: { + document: { + observedHash: "hash", + base: { ...head, hash: "hash" } + } + }, + fileManifest: { + fileManifestId: 1, + entries: { document: "note.md" } + }, + lastSeenUpdateId: 0 + }, + findDocumentId: (path: RelativePath) => + path === "note.md" ? "document" : undefined + } as unknown as Database; + const websocket = { + onWebSocketStatusChanged: new EventListeners< + (connected: boolean) => unknown + >(), + onRemoteVaultUpdateReceived: new EventListeners< + () => Promise + >() + } as unknown as WebSocketManager; + const service = { + events: async () => { + return { headEventId: 0, endEventId: 0, events: [] }; + } + } as unknown as SyncService; + const syncer = new Syncer( + {} as Logger, + database, + new Settings( + new Logger(), + { syncIntervalMs: 0 }, + async () => undefined + ), + service, + websocket, + {} as FileOperations, + {} as ServerConfig, + {} as SyncHistory, + new FixedSizeDocumentCache(1024) + ); + (syncer as unknown as { hasScanned: boolean }).hasScanned = true; + return { + database, + websocket, + syncer + }; + } + + it("reports only a genuinely current path as up to date", () => { + const { syncer } = fixture(); + assert.strictEqual(syncer.isDocumentUpToDate("note.md"), true); + assert.strictEqual(syncer.isDocumentUpToDate("missing.md"), false); + + const internals = syncer as unknown as { + unsyncablePaths: Set; + }; + internals.unsyncablePaths.add("note.md"); + assert.strictEqual(syncer.isDocumentUpToDate("note.md"), false); + }); + + it("projects content responses to metadata before persistence", () => { + const projected = selectContentHead({ + ...head, + contentBase64: "secret" + } as typeof head); + assert.deepEqual(projected, { + vaultUpdateId: head.vaultUpdateId, + contentSize: head.contentSize + }); + assert.strictEqual("contentBase64" in projected, false); + }); +}); + +describe("Syncer diff uploads", () => { + const encoder = new TextEncoder(); + + function fixture(cacheSize = 1024): { + content: ContentSync; + parent: Uint8Array; + } { + const cache = new FixedSizeDocumentCache(cacheSize); + const parent = encoder.encode("one two three\n"); + const content = new ContentSync( + { state: {} } as Database, + new Settings( + new Logger(), + { syncIntervalMs: 0 }, + async () => undefined + ), + { + getDocumentVersionContent: async () => parent + } as unknown as SyncService, + {} as FileOperations, + { + getConfig: async () => ({ mergeableFileExtensions: ["md"] }) + } as ServerConfig, + cache + ); + return { content, parent }; + } + + it("sends a diff when the fetched parent is still cached", async () => { + const { content, parent } = fixture(); + await content.fetchSnapshot( + { vaultUpdateId: 7, documentId: "document" }, + "note.md" + ); + const changed = "one changed three\n"; + const payload = await content.createUploadPayload( + "note.md", + 7, + await hashSnapshot({ content: encoder.encode(changed) }) + ); + + assert.equal(payload.type, "Diff"); + if (payload.type === "Diff") { + assert.equal( + undiff(new TextDecoder().decode(parent), payload.value), + changed + ); + } + }); + + it("falls back to a snapshot when the parent cannot fit in the cache", async () => { + const { content } = fixture(0); + await content.fetchSnapshot( + { vaultUpdateId: 7, documentId: "document" }, + "note.md" + ); + const payload = await content.createUploadPayload( + "note.md", + 7, + await hashSnapshot({ content: encoder.encode("changed") }) + ); + + assert.equal(payload.type, "Snapshot"); + }); +}); diff --git a/frontend/sync-client/src/utils/await-all.test.ts b/frontend/sync-client/src/utils/await-all.test.ts index 09a22d89..0467b7c1 100644 --- a/frontend/sync-client/src/utils/await-all.test.ts +++ b/frontend/sync-client/src/utils/await-all.test.ts @@ -54,3 +54,22 @@ void test("awaitAll works with async functions", async () => { assert.strictEqual(results[0], "async"); assert.strictEqual(results[1], 123); }); + +void test("awaitAll drains unfinished work before reporting a rejection", async () => { + const pending = Promise.withResolvers(); + const error = new Error("failed first"); + let finished = false; + const checked = assert + .rejects(awaitAll([Promise.reject(error), pending.promise]), error) + .then(() => { + finished = true; + }); + try { + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(finished, false); + } finally { + pending.resolve(undefined); + } + + await checked; +}); diff --git a/frontend/sync-client/src/utils/decode-text.test.ts b/frontend/sync-client/src/utils/decode-text.test.ts new file mode 100644 index 00000000..3cfc1dae --- /dev/null +++ b/frontend/sync-client/src/utils/decode-text.test.ts @@ -0,0 +1,20 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { decodeText } from "./decode-text"; +import { isBinary } from "./is-binary"; + +test("text decoding preserves BOMs and line endings across successive files", () => { + for (const text of ["\uFEFFfirst\r\n", "plain\n", "\uFEFFé🙂\r\n"]) { + const bytes = new TextEncoder().encode(text); + assert.equal(isBinary(bytes), false); + assert.equal(decodeText(bytes), text); + } +}); + +test("invalid UTF-8 is binary and does not affect subsequent text decoding", () => { + const invalid = new Uint8Array([0xc3]); + assert.equal(isBinary(invalid), true); + assert.throws(() => decodeText(invalid)); + assert.equal(decodeText(new TextEncoder().encode("valid")), "valid"); + assert.equal(isBinary(new Uint8Array([0])), true); +}); diff --git a/frontend/sync-client/src/utils/globs-to-regexes.test.ts b/frontend/sync-client/src/utils/globs-to-regexes.test.ts index f2b11787..2eed4911 100644 --- a/frontend/sync-client/src/utils/globs-to-regexes.test.ts +++ b/frontend/sync-client/src/utils/globs-to-regexes.test.ts @@ -7,6 +7,7 @@ describe("globsToRegexes", () => { it("basicExample", async () => { const [regex] = globsToRegexes([".git/**"], new Logger()); + assert.ok(regex); assert.ok(regex.test(".git/objects/object")); assert.ok(regex.test(".git/objects/.object")); }); diff --git a/frontend/sync-client/src/utils/is-file-type-mergeable.test.ts b/frontend/sync-client/src/utils/is-file-type-mergeable.test.ts index 78173b54..4c418268 100644 --- a/frontend/sync-client/src/utils/is-file-type-mergeable.test.ts +++ b/frontend/sync-client/src/utils/is-file-type-mergeable.test.ts @@ -5,13 +5,19 @@ import { isFileTypeMergeable } from "./is-file-type-mergeable"; const mergeableExtensions = ["md", "txt"]; describe("isFileTypeMergeable", () => { it("should return true for .md files", () => { - assert.strictEqual(isFileTypeMergeable(".md", mergeableExtensions), true); + assert.strictEqual( + isFileTypeMergeable(".md", mergeableExtensions), + true + ); assert.strictEqual( isFileTypeMergeable("hi.md", mergeableExtensions), true ); assert.strictEqual( - isFileTypeMergeable("my/path/to/my/document.md", mergeableExtensions), + isFileTypeMergeable( + "my/path/to/my/document.md", + mergeableExtensions + ), true ); }); @@ -40,7 +46,10 @@ describe("isFileTypeMergeable", () => { true ); assert.strictEqual( - isFileTypeMergeable("my/path/to/my/DOCUMENT.MD", mergeableExtensions), + isFileTypeMergeable( + "my/path/to/my/DOCUMENT.MD", + mergeableExtensions + ), true ); assert.strictEqual( diff --git a/frontend/sync-client/src/utils/portable-path.test.ts b/frontend/sync-client/src/utils/portable-path.test.ts new file mode 100644 index 00000000..bc79be8a --- /dev/null +++ b/frontend/sync-client/src/utils/portable-path.test.ts @@ -0,0 +1,86 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { resolve } from "node:path"; +import { pathToFileURL } from "node:url"; +import { test } from "node:test"; +import { allocatePortablePath, validatePortablePaths } from "./portable-path"; + +test("a small deeply nested manifest fits within a bounded validation heap", () => { + const source = pathToFileURL(resolve(__dirname, "portable-path.ts")).href; + const result = spawnSync( + process.execPath, + [ + "--max-old-space-size=64", + "--import", + require.resolve("tsx"), + "--input-type=module", + "--eval", + ` + import { validatePortablePaths } from ${JSON.stringify(source)}; + const directory = Array(20000).fill("a").join("/"); + validatePortablePaths([directory + "/one.md", directory + "/two.md"]); + ` + ], + { encoding: "utf8", timeout: 10000 } + ); + assert.equal(result.status, 0, result.stderr || String(result.error)); +}); + +test("conflict allocation renames the first obstructed component deterministically", () => { + const cases: { wanted: string; occupied: string[]; expected: string }[] = [ + { + wanted: "Notes/a.md", + occupied: ["Notes/b.md"], + expected: "Notes/a.md" + }, + { + wanted: "Notes/a.md", + occupied: ["notes/b.md"], + expected: "Notes (conflict new)/a.md" + }, + { + wanted: "Notes/a.md", + occupied: ["Notes"], + expected: "Notes (conflict new)/a.md" + }, + { + wanted: "Notes", + occupied: ["Notes/a.md"], + expected: "Notes (conflict new)" + }, + { + wanted: "note.md", + occupied: ["note.md", "note (conflict new).md"], + expected: "note (conflict new) (1).md" + }, + { + wanted: "Straße/a.md", + occupied: ["STRASSE/b.md"], + expected: "Straße (conflict new)/a.md" + } + ]; + for (const { wanted, occupied, expected } of cases) { + const actual = allocatePortablePath(wanted, "new", occupied); + assert.equal(actual, expected); + assert.equal( + allocatePortablePath(wanted, "new", occupied.toReversed()), + actual + ); + validatePortablePaths([...occupied, actual]); + } +}); + +test("shared directories remain distinct from aliases and file ancestors", () => { + validatePortablePaths(["Notes/one.md", "Notes/two.md", "Other/one.md"]); + for (const paths of [ + ["Notes/one.md", "notes/two.md"], + ["Straße/one.md", "STRASSE/two.md"], + ["Notes", "Notes/one.md"], + ["Notes/one.md", "Notes"], + ["Notes/one.md", "Notes/one.md"] + ]) { + assert.throws(() => { + validatePortablePaths(paths); + }, /Conflicting path/); + } +}); diff --git a/frontend/sync-client/src/utils/rate-limit.test.ts b/frontend/sync-client/src/utils/rate-limit.test.ts index 6c7fb434..d68edd38 100644 --- a/frontend/sync-client/src/utils/rate-limit.test.ts +++ b/frontend/sync-client/src/utils/rate-limit.test.ts @@ -37,7 +37,7 @@ describe("rateLimit", () => { await promise2; assert.strictEqual(mockFn.mock.callCount(), 2); - assert.deepStrictEqual(mockFn.mock.calls[1].arguments, [2]); + assert.deepStrictEqual(mockFn.mock.calls[1]?.arguments, [2]); }); it("should use the most recent arguments if multiple calls are made within interval", async () => { @@ -58,7 +58,7 @@ describe("rateLimit", () => { assert.strictEqual(await promise3, undefined); assert.strictEqual(mockFn.mock.callCount(), 2); - assert.deepStrictEqual(mockFn.mock.calls[0].arguments, ["first"]); - assert.deepStrictEqual(mockFn.mock.calls[1].arguments, ["third"]); + assert.deepStrictEqual(mockFn.mock.calls[0]?.arguments, ["first"]); + assert.deepStrictEqual(mockFn.mock.calls[1]?.arguments, ["third"]); }); });