diff --git a/docs/architecture/sync-algorithm.md b/docs/architecture/sync-algorithm.md index 229e9879..4f3dc8bb 100644 --- a/docs/architecture/sync-algorithm.md +++ b/docs/architecture/sync-algorithm.md @@ -76,8 +76,9 @@ old base are preserved as separate local documents, with conflict names if neede Interrupted recovery saves are replayable. Backups must be restored with the server stopped. -WebSocket messages wake HTTP replay; missed notifications are recovered by replay -and transport liveness checks. Authentication and permanent errors remain visible; +WebSocket `vaultChanged` hints wake HTTP replay. The server periodically compares +its history checkpoint to detect missed broadcasts; transport liveness checks +recover disconnected clients. Authentication and permanent errors remain visible; transient errors retry according to client settings. Observer callback failures are isolated from the background sync loop. diff --git a/frontend/local-client-cli/README.md b/frontend/local-client-cli/README.md index e91322f9..f0cc6388 100644 --- a/frontend/local-client-cli/README.md +++ b/frontend/local-client-cli/README.md @@ -62,7 +62,6 @@ vaultlink \ | `--ignore-pattern ` | - | Glob pattern to ignore (repeatable) | | `--websocket-retry-interval-ms ` | `3500` | WebSocket reconnection interval | | `--log-level ` | `INFO` | Log level: DEBUG, INFO, WARNING, ERROR | -| `--line-endings ` | `auto` | Line ending style: auto, lf, crlf | | `-q, --quiet` | - | Suppress startup banner for non-interactive use | | `-h, --help` | - | Show help | | `-V, --version` | - | Show version | @@ -96,12 +95,7 @@ vaultlink -l ./vault -r wss://sync.example.com -t token123 -v default \ --log-level DEBUG --quiet ``` -Force LF line endings (useful for cross-platform vaults): - -```bash -vaultlink -l ./vault -r wss://sync.example.com -t token123 -v default \ - --line-endings lf -``` +File bytes, including line endings, are preserved on every platform. ## Docker Deployment diff --git a/frontend/local-client-cli/src/args.ts b/frontend/local-client-cli/src/args.ts index 34d839b1..cb6ce421 100644 --- a/frontend/local-client-cli/src/args.ts +++ b/frontend/local-client-cli/src/args.ts @@ -2,9 +2,6 @@ import { Command, Option } from "commander"; import packageJson from "../package.json"; import { LogLevel } from "sync-client"; -export const LINE_ENDING_MODES = ["auto", "lf", "crlf"] as const; -export type LineEndingMode = (typeof LINE_ENDING_MODES)[number]; - interface CliArgs { remoteUri: string; token: string; @@ -17,7 +14,6 @@ interface CliArgs { health?: string; enableTelemetry?: boolean; quiet: boolean; - lineEndings: LineEndingMode; } const VALID_PROTOCOLS = ["http://", "https://", "ws://", "wss://"]; @@ -131,15 +127,6 @@ export function parseArgs(argv: string[]): CliArgs { "[OPTIONAL] Suppress startup banner for non-interactive use" ).env("VAULTLINK_QUIET") ) - .addOption( - new Option( - "--line-endings ", - "[OPTIONAL] Line ending style: auto (platform default), lf, crlf" - ) - .default("auto") - .choices([...LINE_ENDING_MODES]) - .env("VAULTLINK_LINE_ENDINGS") - ) .addHelpText( "after", ` @@ -173,7 +160,6 @@ Environment variables: const health = opts.health as string | undefined; const enableTelemetry = opts.enableTelemetry as boolean | undefined; const quiet = (opts.quiet as boolean | undefined) ?? false; - const lineEndingsStr = (opts.lineEndings as string | undefined) ?? "auto"; /* eslint-enable @typescript-eslint/no-unsafe-type-assertion */ const requiredLocalPath = requireOption(localPath, "localPath"); @@ -203,15 +189,6 @@ Environment variables: } const logLevel = logLevelUpper; - const isLineEndingMode = (value: string): value is LineEndingMode => - (LINE_ENDING_MODES as readonly string[]).includes(value); - if (!isLineEndingMode(lineEndingsStr)) { - throw new Error( - `Invalid line endings mode '${lineEndingsStr}'. Valid values are: ${LINE_ENDING_MODES.join(", ")}` - ); - } - const lineEndings = lineEndingsStr; - return { localPath: requiredLocalPath, remoteUri: requiredRemoteUri, @@ -223,7 +200,6 @@ Environment variables: logLevel, health, enableTelemetry, - quiet, - lineEndings + quiet }; } diff --git a/frontend/local-client-cli/src/cli.ts b/frontend/local-client-cli/src/cli.ts index e6aabc68..0a40d9fb 100644 --- a/frontend/local-client-cli/src/cli.ts +++ b/frontend/local-client-cli/src/cli.ts @@ -1,8 +1,7 @@ import * as path from "path"; import * as fs from "fs/promises"; import * as fsSync from "fs"; -import type { NetworkConnectionStatus , - Logger} from "sync-client"; +import type { NetworkConnectionStatus, Logger } from "sync-client"; import { SyncClient, DEFAULT_SETTINGS, @@ -11,7 +10,7 @@ import { type SyncSettings, type StoredDatabase } from "sync-client"; -import { parseArgs, type LineEndingMode } from "./args"; +import { parseArgs } from "./args"; import { NodeFileSystemOperations, VAULTLINK_DIR } from "./node-filesystem"; import { FileWatcher } from "./file-watcher"; import { formatLogLine } from "./logger-formatter"; @@ -41,7 +40,6 @@ const LOG_LEVEL_ORDER = { function createLogHandler(minLevel: LogLevel): (logLine: LogLine) => void { return (logLine: LogLine): void => { if (LOG_LEVEL_ORDER[logLine.level] >= LOG_LEVEL_ORDER[minLevel]) { - console.log(formatLogLine(logLine)); } }; @@ -50,17 +48,6 @@ function createLogHandler(minLevel: LogLevel): (logLine: LogLine) => void { const HEALTH_CHECK_INTERVAL_MS = 30 * 1000; const PROGRESS_LOG_INTERVAL_MS = 2000; -function resolveLineEndings(mode: LineEndingMode): string { - switch (mode) { - case "lf": - return "\n"; - case "crlf": - return "\r\n"; - case "auto": - return process.platform === "win32" ? "\r\n" : "\n"; - } -} - async function main(): Promise { const args = parseArgs(process.argv); const absolutePath = path.resolve(args.localPath); @@ -96,12 +83,6 @@ async function main(): Promise { emitBoot(LogLevel.INFO, `Local path: ${absolutePath}`); emitBoot(LogLevel.INFO, `Remote URI: ${args.remoteUri}`); emitBoot(LogLevel.INFO, `Vault name: ${args.vaultName}`); - if (args.lineEndings !== "auto") { - emitBoot( - LogLevel.INFO, - `Line endings: ${args.lineEndings.toUpperCase()}` - ); - } } const dataDir = path.join(absolutePath, VAULTLINK_DIR); @@ -158,8 +139,7 @@ async function main(): Promise { JSON.stringify(persistedDatabase, null, 2) ); } - }, - nativeLineEndings: resolveLineEndings(args.lineEndings) + } }); if (args.health !== undefined) { @@ -279,7 +259,6 @@ async function main(): Promise { } main().catch((error: unknown) => { - console.error( `Unexpected error: ${error instanceof Error ? error.message : String(error)}` ); diff --git a/frontend/obsidian-plugin/src/vault-link-plugin.ts b/frontend/obsidian-plugin/src/vault-link-plugin.ts index e222796b..2c456b02 100644 --- a/frontend/obsidian-plugin/src/vault-link-plugin.ts +++ b/frontend/obsidian-plugin/src/vault-link-plugin.ts @@ -5,7 +5,7 @@ import type { TAbstractFile, WorkspaceLeaf } from "obsidian"; -import { Notice, Platform, Plugin, TFile } from "obsidian"; +import { Notice, Plugin, TFile } from "obsidian"; import "../manifest.json"; import { HistoryView } from "./views/history/history-view"; import { StatusBar } from "./views/status-bar/status-bar"; @@ -132,7 +132,6 @@ export default class VaultLinkPlugin extends Plugin { load: this.loadData.bind(this), save: this.saveData.bind(this) }, - nativeLineEndings: Platform.isWin ? "\r\n" : "\n", ...(IS_DEBUG_BUILD ? { fetch: debugging.slowFetchFactory(1), diff --git a/frontend/sync-client/README.md b/frontend/sync-client/README.md index 0a5bc956..c92671db 100644 --- a/frontend/sync-client/README.md +++ b/frontend/sync-client/README.md @@ -1,7 +1,7 @@ # Sync client (API v5) The engine uses one serialized loop per vault and requires Web Crypto (SHA-256). Local notifications and WebSocket -messages wake it; complete disk scans and the durable server event log are the +`vaultChanged` hints wake it; complete disk scans and the durable server event log are the sources of truth. Periodic reconciliation is opt-in: set `syncIntervalMs` to a positive interval to repair missed notifications. Unset or `0` disables polling. Create, delete and move notifications persist logical identity changes before @@ -13,6 +13,8 @@ as content updates. This distinguishes a deliberate delete/recreate from an edit Use `waitUntilFinished()` to await the active attempt; errors leave pending requests saved and are reported to the caller. Transient failures retry; authentication and protocol errors wait for an explicit wake or settings change. +Pausing or destroying the client aborts its HTTP session, including body reads; +resuming starts a fresh session. Connection checks can run while sync is disabled. ## State and policy diff --git a/frontend/sync-client/src/consts.ts b/frontend/sync-client/src/consts.ts index edb133fb..a649d067 100644 --- a/frontend/sync-client/src/consts.ts +++ b/frontend/sync-client/src/consts.ts @@ -2,5 +2,5 @@ export const TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS = 60; export const DIFF_CACHE_SIZE_MB = 2; export const MAX_LOG_MESSAGE_COUNT = 100000; export const MAX_HISTORY_ENTRY_COUNT = 5000; -export const SUPPORTED_API_VERSION = 4; +export const SUPPORTED_API_VERSION = 5; export const WEBSOCKET_DISCONNECT_TIMEOUT_IN_S = 10; diff --git a/frontend/sync-client/src/persistence/settings.test.ts b/frontend/sync-client/src/persistence/settings.test.ts index b17df261..cff24895 100644 --- a/frontend/sync-client/src/persistence/settings.test.ts +++ b/frontend/sync-client/src/persistence/settings.test.ts @@ -57,3 +57,33 @@ it("owns settings arrays instead of sharing them with callers and listeners", as await settings.setSettings({ maxFileSizeMB: 12 }); assert.deepEqual(settings.getSettings().ignorePatterns, ["updated"]); }); + +it("file eligibility follows successfully saved settings", async () => { + let fail = false; + const settings = new Settings( + new Logger(), + { + ignorePatterns: ["private/**"], + maxFileSizeMB: 1 + }, + async () => { + if (fail) throw new Error("save failed"); + } + ); + assert(settings.isIgnored("private/note.md")); + assert(settings.isIgnored(".vault-link-sync/state.json")); + assert(!settings.isIgnored("public/note.md")); + assert(!settings.isOversized(1024 * 1024)); + assert(settings.isOversized(1024 * 1024 + 1)); + fail = true; + await assert.rejects( + settings.setSettings({ ignorePatterns: [], maxFileSizeMB: 2 }), + /save failed/ + ); + assert(settings.isIgnored("private/note.md")); + assert(settings.isOversized(1024 * 1024 + 1)); + fail = false; + await settings.setSettings({ ignorePatterns: [], maxFileSizeMB: 2 }); + assert(!settings.isIgnored("private/note.md")); + assert(!settings.isOversized(1024 * 1024 + 1)); +}); diff --git a/frontend/sync-client/src/persistence/settings.ts b/frontend/sync-client/src/persistence/settings.ts index 9cc3bbe3..9a93e0dd 100644 --- a/frontend/sync-client/src/persistence/settings.ts +++ b/frontend/sync-client/src/persistence/settings.ts @@ -1,4 +1,6 @@ import type { Logger } from "../tracing/logger"; +import { globsToRegexes } from "../utils/globs-to-regexes"; +import { isInternalPath } from "../utils/portable-path"; import { Lock } from "../utils/data-structures/locks"; import { EventListeners } from "../utils/data-structures/event-listeners"; @@ -43,6 +45,7 @@ export class Settings { >(); private settings: SyncSettings; + private ignorePatterns: RegExp[]; private readonly lock: Lock = new Lock(); public constructor( @@ -55,11 +58,27 @@ export class Settings { ...structuredClone(initialState ?? {}) }; + this.ignorePatterns = globsToRegexes( + this.settings.ignorePatterns, + logger + ); + this.logger.debug( `Loaded settings: ${JSON.stringify(redactSettings(this.settings), null, 2)}` ); } + public isIgnored(path: string): boolean { + return ( + isInternalPath(path) || + this.ignorePatterns.some((pattern) => pattern.test(path)) + ); + } + + public isOversized(size: number): boolean { + return size > this.settings.maxFileSizeMB * 1024 * 1024; + } + public getSettings(): SyncSettings { return { ...this.settings, @@ -91,6 +110,11 @@ export class Settings { const next = { ...this.settings, ...update }; await this.saveData(next); this.settings = next; + if (update.ignorePatterns !== undefined) + this.ignorePatterns = globsToRegexes( + next.ignorePatterns, + this.logger + ); return [this.getSettings(), oldSettings] as const; }); diff --git a/frontend/sync-client/src/services/fetch-controller.test.ts b/frontend/sync-client/src/services/fetch-controller.test.ts deleted file mode 100644 index a0a70d80..00000000 --- a/frontend/sync-client/src/services/fetch-controller.test.ts +++ /dev/null @@ -1,177 +0,0 @@ -import type { Mock } from "node:test"; -import { describe, it, mock, beforeEach, afterEach } from "node:test"; -import assert from "node:assert"; -import { FetchController } from "./fetch-controller"; -import { Logger } from "../tracing/logger"; -import { SyncResetError } from "../errors/errors"; -import { sleep } from "../utils/sleep"; - -describe("FetchController", () => { - const createMockFetch = ( - shouldSleep: boolean - ): Mock<() => Promise> => - mock.fn(async () => { - if (shouldSleep) { - await sleep(30); - } - return Promise.resolve(new Response("OK", { status: 200 })); - }); - - beforeEach(() => { - mock.timers.enable({ apis: ["setTimeout"] }); - }); - - afterEach(() => { - mock.timers.reset(); - }); - - it("should allow fetch when canFetch is true", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - const mockFetch = createMockFetch(false); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - const response = await controlledFetch("http://example.com"); - - assert.strictEqual(await response.text(), "OK"); - assert.strictEqual(mockFetch.mock.calls.length, 1); - }); - - it("should block fetch until canFetch becomes true", async () => { - const logger = new Logger(); - const controller = new FetchController(false, logger); - const mockFetch = createMockFetch(true); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - const fetchPromise = controlledFetch("http://example.com"); - assert.strictEqual(mockFetch.mock.calls.length, 0); - - controller.canFetch = true; - await Promise.resolve(); - mock.timers.tick(30); - - const response = await fetchPromise; - assert.strictEqual(await response.text(), "OK"); - assert.strictEqual(mockFetch.mock.calls.length, 1); - }); - - it("should reject during reset", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - const mockFetch = createMockFetch(true); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - const firstRequest = controlledFetch("http://example.com"); - assert.strictEqual(mockFetch.mock.calls.length, 1); - - controller.startReset(); - - const secondRequest = controlledFetch("http://example.com"); - - await assert.rejects( - firstRequest, - (error: unknown) => error instanceof SyncResetError - ); - await assert.rejects( - secondRequest, - (error: unknown) => error instanceof SyncResetError - ); - assert.strictEqual(mockFetch.mock.calls.length, 1); - }); - - it("should allow fetch after reset finishes", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - const mockFetch = createMockFetch(false); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - controller.startReset(); - controller.finishReset(); - - const response = await controlledFetch("http://example.com"); - assert.strictEqual(await response.text(), "OK"); - }); - - it("should defer canFetch changes during reset", async () => { - const logger = new Logger(); - const controller = new FetchController(false, logger); - const mockFetch = createMockFetch(true); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - controller.startReset(); - controller.canFetch = true; - - await assert.rejects( - async () => controlledFetch("http://example.com"), - (error: unknown) => error instanceof SyncResetError - ); - - controller.finishReset(); - - const fetchPromise = controlledFetch("http://example.com"); - mock.timers.tick(30); - - const response = await fetchPromise; - assert.strictEqual(await response.text(), "OK"); - }); - - it("should handle different input types", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - const mockFetch = createMockFetch(false); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - await controlledFetch("http://example.com"); - await controlledFetch(new URL("http://example.com")); - await controlledFetch( - new Request("http://example.com", { method: "POST" }) - ); - - assert.strictEqual(mockFetch.mock.calls.length, 3); - }); - - it("should handle fetch errors", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - const mockFetch = mock.fn(async () => { - throw new Error("Network error"); - }); - const controlledFetch = controller.getControlledFetchImplementation( - logger, - mockFetch - ); - - await assert.rejects( - async () => controlledFetch("http://example.com"), - (error: unknown) => - error instanceof Error && error.message === "Network error" - ); - }); - - it("should not create unhandled rejection on reset with no waiting fetches", async () => { - const logger = new Logger(); - const controller = new FetchController(true, logger); - - controller.startReset(); - mock.timers.tick(10); - controller.finishReset(); - }); -}); diff --git a/frontend/sync-client/src/services/fetch-controller.ts b/frontend/sync-client/src/services/fetch-controller.ts deleted file mode 100644 index c1bd29c9..00000000 --- a/frontend/sync-client/src/services/fetch-controller.ts +++ /dev/null @@ -1,149 +0,0 @@ -import type { Logger } from "../tracing/logger"; -import { createPromise } from "../utils/create-promise"; -import { SyncResetError } from "../errors/errors"; - -/** - * Offers a resettable fetch implementation that waits until syncing is enabled - * and aborts outstanding requests when a reset is started. - */ -export class FetchController { - private static readonly UNTIL_RESOLUTION = Symbol(); - - private isResetting = false; - - // Promise resolves on the next state change: sync enabled/disabled or reset started/ended - private until: Promise; - private resolveUntil: (result: symbol) => unknown; - private rejectUntil: (reason: unknown) => unknown; - - public constructor( - private _canFetch: boolean, - private readonly logger: Logger - ) { - [this.until, this.resolveUntil, this.rejectUntil] = - createPromise(); - } - - /** - * Whether the fetch implementation can immediately send requests once outside of a reset. - */ - public get canFetch(): boolean { - return this._canFetch; - } - - /** - * Allow or disallow fetching. The changes only take effect if not resetting. - * When called during a reset, its effect is deferred until the reset is finished. - * - * @param canFetch Whether fetching is enabled - */ - public set canFetch(canFetch: boolean) { - this._canFetch = canFetch; - - if (!this.isResetting) { - const previousResolve = this.resolveUntil; - [this.until, this.resolveUntil, this.rejectUntil] = - createPromise(); - previousResolve(FetchController.UNTIL_RESOLUTION); - } - } - - private static getUrlFromInput(input: RequestInfo | URL): string { - if (input instanceof URL) { - return input.href; - } - if (typeof input === "string") { - return input; - } - return input.url; - } - - /** - * Starts a reset, causing all ongoing and future fetches to be rejected - * with a SyncResetError until finishReset is called. - */ - public startReset(): void { - this.isResetting = true; - this.rejectUntil(new SyncResetError()); - // Catch unhandled rejection if no fetches are waiting - this.until.catch(() => { - // Intentionally ignore - this rejection is handled by waiting fetches - }); - } - - /** - * Finishes a reset, allowing fetches to proceed or wait again depending on - * the current sync settings. - */ - public finishReset(): void { - if (!this.isResetting) { - return; - } - - this.isResetting = false; - [this.until, this.resolveUntil, this.rejectUntil] = createPromise(); - } - - /** - * - * |------------------|---------------|-----------------------------------------------------| - * | | Sync enabled | Sync disabled | - * |------------------|-------------- |-----------------------------------------------------| - * | During reset | Rejects with SyncResetError without sending request | - * |------------------|-------------- |-----------------------------------------------------| - * | Outside of reset | Same as fetch | Blocks until sync is enabled and then same as fetch | - * |------------------|---------------|-----------------------------------------------------| - * - * @param logger for errors - * @param fetch to wrap - * @returns a wrapped fetch implementation affected by the FetchController state - */ - public getControlledFetchImplementation( - logger: Logger, - fetch: typeof globalThis.fetch = globalThis.fetch - ): typeof globalThis.fetch { - return async ( - input: RequestInfo | URL, - init?: RequestInit - ): Promise => { - while (!this.canFetch || this.isResetting) { - await this.until; - } - - try { - // https://github.com/jonbern/fetch-retry/blob/8684ef4e688375f623bd76f13add76dbc1d67cfb/index.js#L67C1-L70C21 - const _input = - typeof Request !== "undefined" && input instanceof Request - ? input.clone() - : input; - - const fetchPromise = fetch(_input, init); - - // We only want to catch rejections from `this.until` - let result: symbol | Response | undefined = undefined; - do { - result = await Promise.race([this.until, fetchPromise]); - } while (result === FetchController.UNTIL_RESOLUTION); - - const fetchResult: Response = result as Response; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion - - if (!fetchResult.ok) { - this.logger.warn( - `Fetch for ${FetchController.getUrlFromInput( - input - )}, got status ${fetchResult.status}` - ); - } - - return fetchResult; - } catch (error) { - logger.warn( - `Fetch for ${FetchController.getUrlFromInput( - input - )}, got error: ${error}` - ); - throw error; - } - }; - } -} diff --git a/frontend/sync-client/src/services/sync-service.test.ts b/frontend/sync-client/src/services/sync-service.test.ts new file mode 100644 index 00000000..d4f72b47 --- /dev/null +++ b/frontend/sync-client/src/services/sync-service.test.ts @@ -0,0 +1,146 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { Settings } from "../persistence/settings"; +import { Logger } from "../tracing/logger"; +import { SyncResetError } from "../errors/errors"; +import { SyncService } from "./sync-service"; + +const settings = () => + new Settings( + new Logger(), + { + remoteUri: "http://test", + vaultName: "test", + requestTimeoutMs: 1000 + }, + async () => {} + ); + +test("paused sync sends no requests, while connection checks still work", async () => { + let requests = 0; + const service = new SyncService("device", settings(), async () => { + requests++; + return Response.json({ headEventId: 0, events: [] }); + }); + await assert.rejects(service.events(0), SyncResetError); + assert.equal(requests, 0); + await service.ping(); + assert.equal(requests, 1); + service.resume(); + assert.equal((await service.events(0)).headEventId, 0); + assert.equal(requests, 2); +}); + +test("pause aborts requests and fences late responses from an injected fetch", async () => { + const response = Promise.withResolvers(); + const signals: AbortSignal[] = []; + const saved: (string | undefined)[] = []; + const service = new SyncService( + "device", + settings(), + async (_, init) => { + signals.push(init!.signal!); + return signals.length === 1 + ? response.promise + : Response.json({ headEventId: 0, events: [] }); + }, + { + get: () => undefined, + save: async (value) => { + saved.push(value); + } + } + ); + service.resume(); + const first = service.events(0); + service.pause(); + await assert.rejects(first, SyncResetError); + assert.equal(signals[0].aborted, true); + service.resume(); + await service.events(0); + assert.equal(signals[1].aborted, false); + response.resolve( + Response.json( + {}, + { headers: { "X-Vault-Link-History": "99:abandoned" } } + ) + ); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(saved, []); +}); + +for (const binary of [false, true]) { + test(`pause interrupts a stalled ${binary ? "binary" : "JSON"} response body`, async () => { + const headersSaved = Promise.withResolvers(); + let body!: ReadableStreamDefaultController; + const service = new SyncService( + "device", + settings(), + async () => + new Response( + new ReadableStream({ + start(controller) { + body = controller; + } + }), + { headers: { "X-Vault-Link-History": "1:head" } } + ), + { + get: () => undefined, + save: async () => { + headersSaved.resolve(); + } + } + ); + service.resume(); + const request = binary + ? service.getDocumentVersionContent({ + documentId: "a", + vaultUpdateId: 1 + }) + : service.events(0); + await headersSaved.promise; + await new Promise((resolve) => setImmediate(resolve)); + service.pause(); + await assert.rejects(request, SyncResetError); + 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(); + let settled = false; + let checkpoint: string | undefined; + const service = new SyncService( + "device", + settings(), + async () => + Response.json( + {}, + { + headers: { "X-Vault-Link-History": "1:head" } + } + ), + { + get: () => checkpoint, + save: async (value) => { + saving.resolve(); + await release.promise; + checkpoint = value; + } + } + ); + service.resume(); + const request = service.events(0).finally(() => { + settled = true; + }); + await saving.promise; + service.pause(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(settled, false); + release.resolve(); + await assert.rejects(request, SyncResetError); + assert.equal(checkpoint, "1:head"); +}); diff --git a/frontend/sync-client/src/services/sync-service.ts b/frontend/sync-client/src/services/sync-service.ts index 48629d49..cf918151 100644 --- a/frontend/sync-client/src/services/sync-service.ts +++ b/frontend/sync-client/src/services/sync-service.ts @@ -1,11 +1,7 @@ import type { Settings } from "../persistence/settings"; -import type { FetchController } from "./fetch-controller"; -import type { Logger } from "../tracing/logger"; import type { DocumentUpdateResponse } from "./types/DocumentUpdateResponse"; -import type { DocumentVersion } from "./types/DocumentVersion"; import type { DocumentVersionWithoutContent } from "./types/DocumentVersionWithoutContent"; import type { EventBatch } from "./types/EventBatch"; -import type { FileManifest } from "./types/FileManifest"; import type { FileManifestUpdateResponse } from "./types/FileManifestUpdateResponse"; import type { PutFileContent } from "./types/PutFileContent"; import type { PushFileManifest } from "./types/PushFileManifest"; @@ -14,29 +10,47 @@ import type { PingResponse } from "./types/PingResponse"; import { AuthenticationError, PermanentSyncError, - ServerHistoryChangedError + ServerHistoryChangedError, + SyncResetError } from "../errors/errors"; +// Injected fetch implementations may ignore cancellation. Stop waiting anyway, +// and never let their late responses reach checkpoint or engine persistence. +async function abortable( + signal: AbortSignal, + operation: () => Promise +): Promise { + signal.throwIfAborted(); + let abort = (): void => undefined; + try { + return await Promise.race([ + new Promise((_, reject) => { + abort = (): void => { + // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- Only reset errors and timeout errors reach this helper. + reject(signal.reason); + }; + signal.addEventListener("abort", abort, { once: true }); + }), + operation() + ]); + } finally { + signal.removeEventListener("abort", abort); + } +} + export class SyncService { - private readonly client: typeof fetch; - private readonly rawFetch: typeof fetch; + private session = new AbortController(); public constructor( private readonly deviceId: string, - fetchController: FetchController, private readonly settings: Settings, - logger: Logger, - fetchImplementation: typeof fetch = globalThis.fetch, + private readonly fetchImplementation: typeof fetch = globalThis.fetch, private readonly history?: { get: () => string | undefined; save: (checkpoint: string | undefined) => Promise; } ) { - this.rawFetch = async (...args) => fetchImplementation(...args); - this.client = fetchController.getControlledFetchImplementation( - logger, - this.rawFetch - ); + this.pause(); } public get verifiesHistory(): boolean { @@ -46,21 +60,80 @@ export class SyncService { return this.history?.get() !== undefined; } + public resume(): void { + this.session = new AbortController(); + } + + public pause(): void { + this.session.abort(new SyncResetError()); + } + public async resetHistory(): Promise { await this.history?.save(undefined); } + public async ping(): Promise { + return this.request("/ping", undefined, true); + } + + public async vaultSnapshot(): Promise { + return this.request("/vault-snapshot"); + } + + public async events(after: number): Promise { + return this.request(`/events-since?after=${after}`); + } + + public async pushFileManifest( + request: PushFileManifest + ): Promise { + return this.request("/file-manifest", request); + } + + public async putFileContent( + id: string, + request: PutFileContent + ): Promise { + return this.request(`/documents/${id}`, request); + } + + public async metadata(id: string): Promise { + return this.request(`/documents/${id}/metadata`); + } + + public async getDocumentVersionContent({ + documentId, + vaultUpdateId + }: { + documentId: string; + vaultUpdateId: number; + }): Promise { + return this.request( + `/documents/${documentId}/versions/${vaultUpdateId}/content`, + undefined, + false, + async (response) => new Uint8Array(await response.arrayBuffer()) + ); + } + private historyHeaders(): Record { const checkpoint = this.history?.get(); - return checkpoint ? { "X-Vault-Link-History": checkpoint } : {}; + return checkpoint !== undefined && checkpoint !== "" + ? { "X-Vault-Link-History": checkpoint } + : {}; } private async recordHistory(response: Response): Promise { const checkpoint = response.headers.get("x-vault-link-history"); - if (checkpoint && this.history) { + if ( + checkpoint !== null && + checkpoint !== "" && + this.history !== undefined + ) { const previous = this.history.get(); if ( - !previous || + previous === undefined || + previous === "" || Number(checkpoint.split(":")[0]) >= Number(previous.split(":")[0]) ) @@ -75,7 +148,7 @@ export class SyncService { ); } - public getUrl(path: string): string { + private getUrl(path: string): string { const { remoteUri, vaultName } = this.settings.getSettings(); return `${remoteUri.replace(/\/$/u, "")}/vaults/${encodeURIComponent(vaultName)}${path}`; } @@ -83,35 +156,42 @@ export class SyncService { private async request( path: string, body?: unknown, - raw = false + raw = false, + decode: (response: Response) => Promise = async (response) => + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- Endpoint methods supply the generated protocol response type. + response.json() as Promise ): Promise { - const response = await (raw ? this.rawFetch : this.client)( - this.getUrl(path), - { + const timeout = AbortSignal.timeout( + this.settings.getSettings().requestTimeoutMs + ); + const signal = raw + ? timeout + : AbortSignal.any([timeout, this.session.signal]); + const response = await abortable(signal, async () => + this.fetchImplementation(this.getUrl(path), { method: body === undefined ? "GET" : "PUT", body: body === undefined ? undefined : JSON.stringify(body), - signal: AbortSignal.timeout( - this.settings.getSettings().requestTimeoutMs - ), + signal, headers: { ...(raw ? {} : this.historyHeaders()), Authorization: `Bearer ${this.settings.getSettings().token}`, "Device-Id": this.deviceId, "Content-Type": "application/json" } - } + }) ); - + signal.throwIfAborted(); this.checkHistory(response); if (!response.ok) { this.throwForErrorResponse( response, - `HTTP ${response.status}: ${await response.text()}` + `HTTP ${response.status}: ${await abortable(signal, async () => response.text())}` ); } + // Finish metadata saves even if the transport is paused meanwhile. if (!raw) await this.recordHistory(response); - return (await response.json()) as T; + return abortable(signal, async () => decode(response)); } private throwForErrorResponse(response: Response, message: string): never { @@ -128,75 +208,4 @@ export class SyncService { } throw new Error(message); } - - public async ping(): Promise { - return this.request("/ping", undefined, true); - } - - public async vaultSnapshot(): Promise { - return this.request("/vault-snapshot"); - } - - public async events(after: number): Promise { - return this.request(`/events-since?after=${after}`); - } - - public async fileManifest(): Promise { - return this.request("/file-manifest"); - } - - public async pushFileManifest( - request: PushFileManifest - ): Promise { - return this.request("/file-manifest", request); - } - - public async putFileContent( - id: string, - request: PutFileContent - ): Promise { - return this.request(`/documents/${id}`, request); - } - - public async get(id: string): Promise { - return this.request(`/documents/${id}`); - } - - public async metadata(id: string): Promise { - return this.request(`/documents/${id}/metadata`); - } - - public async getDocumentVersionContent({ - documentId, - vaultUpdateId - }: { - documentId: string; - vaultUpdateId: number; - }): Promise { - const response = await this.client( - this.getUrl( - `/documents/${documentId}/versions/${vaultUpdateId}/content` - ), - { - signal: AbortSignal.timeout( - this.settings.getSettings().requestTimeoutMs - ), - headers: { - ...this.historyHeaders(), - Authorization: `Bearer ${this.settings.getSettings().token}` - } - } - ); - - this.checkHistory(response); - if (!response.ok) { - this.throwForErrorResponse( - response, - `Cannot fetch version ${vaultUpdateId}: HTTP ${response.status}` - ); - } - - await this.recordHistory(response); - return new Uint8Array(await response.arrayBuffer()); - } } diff --git a/frontend/sync-client/src/services/types/WebSocketHandshake.ts b/frontend/sync-client/src/services/types/WebSocketHandshake.ts index 85c2cf0d..310c0bb8 100644 --- a/frontend/sync-client/src/services/types/WebSocketHandshake.ts +++ b/frontend/sync-client/src/services/types/WebSocketHandshake.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type WebSocketHandshake = { token: string, deviceId: string, lastSeenVaultUpdateId: number | null, }; +export type WebSocketHandshake = { token: string, deviceId: string, }; diff --git a/frontend/sync-client/src/services/types/WebSocketServerMessage.ts b/frontend/sync-client/src/services/types/WebSocketServerMessage.ts index 2a7a7813..c9759e04 100644 --- a/frontend/sync-client/src/services/types/WebSocketServerMessage.ts +++ b/frontend/sync-client/src/services/types/WebSocketServerMessage.ts @@ -1,5 +1,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { CursorPositionFromServer } from "./CursorPositionFromServer"; -import type { EventBatch } from "./EventBatch"; -export type WebSocketServerMessage = { "type": "vaultEvents" } & EventBatch | { "type": "cursorPositions" } & CursorPositionFromServer; +export type WebSocketServerMessage = { "type": "vaultChanged" } | { "type": "cursorPositions" } & CursorPositionFromServer; diff --git a/frontend/sync-client/src/services/websocket-manager.test.ts b/frontend/sync-client/src/services/websocket-manager.test.ts index d5223b6a..645fa627 100644 --- a/frontend/sync-client/src/services/websocket-manager.test.ts +++ b/frontend/sync-client/src/services/websocket-manager.test.ts @@ -136,19 +136,13 @@ describe("WebSocketManager", () => { .webSocket; mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); await new Promise((resolve) => setTimeout(resolve, 100)); @@ -208,8 +202,7 @@ describe("WebSocketManager", () => { manager.sendHandshakeMessage({ type: "handshake", token: "test", - deviceId: "test", - lastSeenVaultUpdateId: null + deviceId: "test" }); }); @@ -317,9 +310,7 @@ describe("WebSocketManager", () => { const mockWs = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; mockWs.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); await new Promise((resolve) => setTimeout(resolve, 10)); @@ -465,9 +456,7 @@ for (const closeBehavior of ["throws", "stays open"]) { assert.equal(manager.isWebSocketConnected, false); assert.equal(connected, false); socket.simulateMessage({ - type: "vaultEvents", - headEventId: 0, - events: [] + type: "vaultChanged" }); await manager.waitUntilFinished(); assert.equal(delivered, 0); @@ -499,8 +488,7 @@ it("a handshake send failure retires the connection without escaping the open ca manager.sendHandshakeMessage({ type: "handshake", token: "test", - deviceId: "test", - lastSeenVaultUpdateId: 0 + deviceId: "test" }); }); const statuses: boolean[] = []; diff --git a/frontend/sync-client/src/services/websocket-manager.ts b/frontend/sync-client/src/services/websocket-manager.ts index 3fb80947..8d90fbbd 100644 --- a/frontend/sync-client/src/services/websocket-manager.ts +++ b/frontend/sync-client/src/services/websocket-manager.ts @@ -5,7 +5,6 @@ import type { WebSocketClientMessage } from "./types/WebSocketClientMessage"; import type { CursorPositionFromClient } from "./types/CursorPositionFromClient"; import type { ClientCursors } from "./types/ClientCursors"; import { createPromise } from "../utils/create-promise"; -import type { EventBatch } from "./types/EventBatch"; import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts"; import { removeFromArray } from "../utils/remove-from-array"; import { EventListeners } from "../utils/data-structures/event-listeners"; @@ -17,7 +16,7 @@ export class WebSocketManager { >(); public readonly onRemoteVaultUpdateReceived = new EventListeners< - (update: EventBatch) => Promise + () => Promise >(); public readonly onRemoteCursorsUpdateReceived = new EventListeners< @@ -303,7 +302,7 @@ export class WebSocketManager { event.data ) as WebSocketServerMessage; - if (["vaultEvents", "cursorPositions"].includes(message.type)) + if (["vaultChanged", "cursorPositions"].includes(message.type)) this.expectResponse(socket); // Track the message handling promise const messageHandlingPromise = this.handleWebSocketMessage( @@ -340,8 +339,8 @@ export class WebSocketManager { private async handleWebSocketMessage( message: WebSocketServerMessage ): Promise { - if (message.type === "vaultEvents") { - await this.onRemoteVaultUpdateReceived.triggerAsync(message); + if (message.type === "vaultChanged") { + await this.onRemoteVaultUpdateReceived.triggerAsync(); // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition } else if (message.type === "cursorPositions") { diff --git a/frontend/sync-client/src/sync-client.ts b/frontend/sync-client/src/sync-client.ts index 51b92dfa..8660c0b1 100644 --- a/frontend/sync-client/src/sync-client.ts +++ b/frontend/sync-client/src/sync-client.ts @@ -11,7 +11,6 @@ import type { FileSystemOperations } from "./file-operations/filesystem-operatio import { FileOperations } from "./file-operations/file-operations"; import { Logger, LogLevel } from "./tracing/logger"; import { SyncHistory } from "./tracing/sync-history"; -import { FetchController } from "./services/fetch-controller"; import { SyncService } from "./services/sync-service"; import { ServerConfig } from "./services/server-config"; import { WebSocketManager } from "./services/websocket-manager"; @@ -23,7 +22,6 @@ import { DocumentSyncStatus } from "./types/document-sync-status"; import type { CursorSpan } from "./services/types/CursorSpan"; import { isInternalPath } from "./utils/portable-path"; import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache"; -import { globsToRegexes } from "./utils/globs-to-regexes"; import { Lock } from "./utils/data-structures/locks"; import { unappliedChanges, @@ -61,7 +59,7 @@ export class SyncClient { private readonly database: Database, private readonly syncer: Syncer, private readonly webSocketManager: WebSocketManager, - private readonly fetchController: FetchController, + private readonly service: SyncService, private readonly serverConfig: ServerConfig, private readonly cursorTracker: CursorTracker, private readonly fileChangeNotifier: FileChangeNotifier, @@ -95,8 +93,6 @@ export class SyncClient { persistence: PersistenceProvider; fetch?: typeof globalThis.fetch; webSocket?: typeof globalThis.WebSocket; - /** V4 preserves bytes and line endings as stored, on every platform. */ - nativeLineEndings?: string; }): Promise { const logger = new Logger(); let stored = (await persistence.load()) ?? {}; @@ -152,32 +148,21 @@ export class SyncClient { "Offline notifications belong to another vault; use a separate state store" ); } - const fetchController = new FetchController( - settings.getSettings().isSyncEnabled, - logger - ); const deviceId = createClientId(); - const service = new SyncService( - deviceId, - fetchController, - settings, - logger, - fetch, - { - get: () => - stored.historyCheckpoint?.vaultKey === - vaultKey(settings.getSettings()) - ? stored.historyCheckpoint.checkpoint - : undefined, - save: async (checkpoint) => - save({ - historyCheckpoint: { - vaultKey: vaultKey(settings.getSettings()), - checkpoint - } - }) - } - ); + const service = new SyncService(deviceId, settings, fetch, { + get: () => + stored.historyCheckpoint?.vaultKey === + vaultKey(settings.getSettings()) + ? stored.historyCheckpoint.checkpoint + : undefined, + save: async (checkpoint) => + save({ + historyCheckpoint: { + vaultKey: vaultKey(settings.getSettings()), + checkpoint + } + }) + }); const serverConfig = new ServerConfig(service); const notifier = new FileChangeNotifier(); let syncer: Syncer; @@ -192,15 +177,8 @@ export class SyncClient { entries: () => localChanges, flush: async () => syncer?.flushLocalChanges() }, - (path, size) => { - const current = settings.getSettings(); - return ( - size > current.maxFileSizeMB * 1024 * 1024 || - globsToRegexes(current.ignorePatterns, logger).some( - (pattern) => pattern.test(path) - ) - ); - } + (path, size) => + settings.isIgnored(path) || settings.isOversized(size) ); const websocket = new WebSocketManager( deviceId, @@ -240,7 +218,7 @@ export class SyncClient { database, syncer, websocket, - fetchController, + service, serverConfig, cursors, notifier, @@ -308,15 +286,14 @@ export class SyncClient { if (this.destroyed) return; if (!this.settings.getSettings().isSyncEnabled) return; await this.database.bindVault(vaultKey(this.settings.getSettings())); - this.fetchController.finishReset(); - this.fetchController.canFetch = true; + this.service.resume(); this.syncer.start(); await this.syncer.waitUntilFinished(); } private async pause(): Promise { const stopped = this.syncer.stop(); - this.fetchController.startReset(); + this.service.pause(); await this.webSocketManager.stop(); await stopped; } @@ -470,7 +447,7 @@ export class SyncClient { this.destroyed = true; // Interrupt a start/reset awaiting network progress before waiting for // its lifecycle operation to finish. - this.fetchController.startReset(); + this.service.pause(); const stopped = this.syncer.stop(); this.destroying = this.lifecycle.withLock(async () => { try { diff --git a/frontend/sync-client/src/sync-operations/syncer.test.ts b/frontend/sync-client/src/sync-operations/syncer.test.ts index 862c56e1..524fbabb 100644 --- a/frontend/sync-client/src/sync-operations/syncer.test.ts +++ b/frontend/sync-client/src/sync-operations/syncer.test.ts @@ -4,16 +4,15 @@ import { afterEach, beforeEach, describe, it, mock } from "node:test"; import type { Database } from "../persistence/database"; import { DEFAULT_SETTINGS, - type Settings, + Settings, type SyncSettings } from "../persistence/settings"; import type { FileOperations } from "../file-operations/file-operations"; import type { SyncService } from "../services/sync-service"; import type { ServerConfig } from "../services/server-config"; import type { WebSocketManager } from "../services/websocket-manager"; -import type { Logger } from "../tracing/logger"; +import { Logger } from "../tracing/logger"; import type { SyncHistory } from "../tracing/sync-history"; -import type { EventBatch } from "../services/types/EventBatch"; import { EventListeners } from "../utils/data-structures/event-listeners"; import { Syncer } from "./syncer"; import { FixedSizeDocumentCache } from "../utils/data-structures/fix-sized-cache"; @@ -24,12 +23,11 @@ function createSyncer(overrides: Partial = {}): { syncer: Syncer; runCount: () => number; } { - const settings = { - getSettings: () => ({ - ...DEFAULT_SETTINGS, - ...overrides - }) - } as unknown as Settings; + const settings = new Settings( + new Logger(), + overrides, + async () => undefined + ); const database = { state: {} } as unknown as Database; @@ -162,7 +160,7 @@ describe("Syncer per-file state and event delivery", () => { (connected: boolean) => unknown >(), onRemoteVaultUpdateReceived: new EventListeners< - (batch: EventBatch) => Promise + () => Promise >() } as unknown as WebSocketManager; const service = { @@ -174,9 +172,11 @@ describe("Syncer per-file state and event delivery", () => { "device", {} as Logger, database, - { - getSettings: () => ({ ...DEFAULT_SETTINGS, syncIntervalMs: 0 }) - } as Settings, + new Settings( + new Logger(), + { syncIntervalMs: 0 }, + async () => undefined + ), service, websocket, {} as FileOperations, @@ -238,9 +238,11 @@ describe("Syncer diff uploads", () => { "device", {} as Logger, { state: {} } as Database, - { - getSettings: () => ({ ...DEFAULT_SETTINGS, syncIntervalMs: 0 }) - } as Settings, + new Settings( + new Logger(), + { syncIntervalMs: 0 }, + async () => undefined + ), { getDocumentVersionContent: async () => parent } as unknown as SyncService, diff --git a/frontend/sync-client/src/sync-operations/syncer.ts b/frontend/sync-client/src/sync-operations/syncer.ts index 9753c7c5..8a123c04 100644 --- a/frontend/sync-client/src/sync-operations/syncer.ts +++ b/frontend/sync-client/src/sync-operations/syncer.ts @@ -26,7 +26,6 @@ import type { VaultSnapshot } from "../services/types/VaultSnapshot"; import type { PushContent } from "../services/types/PushContent"; import type { Logger } from "../tracing/logger"; import { EventListeners } from "../utils/data-structures/event-listeners"; -import { globsToRegexes } from "../utils/globs-to-regexes"; import { mergeContent, toStoredSnapshot } from "./content"; import { scanLocalFiles, type LocalChange } from "./scan"; import { @@ -91,8 +90,7 @@ export class Syncer { websocket.sendHandshakeMessage({ type: "handshake", token: settings.getSettings().token, - deviceId, - lastSeenVaultUpdateId: database.state.lastSeenUpdateId + deviceId }); } // Failed upgrades and lost transports still need HTTP catchup, @@ -128,7 +126,7 @@ export class Syncer { !this.hasScanned || this.pendingPaths.has(path) || this.unsyncablePaths.has(path) || - this.ignored(path) + this.settings.isIgnored(path) ) { return false; } @@ -414,7 +412,10 @@ export class Syncer { const info = await this.files.fs.stat(path); if (info?.kind !== "file") continue; const doc = previous.documents[id]; - if (this.ignored(path) || this.oversized(info.size)) { + if ( + this.settings.isIgnored(path) || + this.settings.isOversized(info.size) + ) { next.local[id] = path; next.documents[id] = { materialized: true, @@ -457,20 +458,6 @@ export class Syncer { }; } - private ignored(path: string): boolean { - return ( - isInternalPath(path) || - globsToRegexes( - this.settings.getSettings().ignorePatterns, - this.logger - ).some((pattern) => pattern.test(path)) - ); - } - - private oversized(size: number): boolean { - return size > this.settings.getSettings().maxFileSizeMB * 1024 * 1024; - } - private async cacheContent( updateId: number, path: string | undefined, @@ -550,8 +537,8 @@ export class Syncer { guard: this.localGuard(), changes: this.changes, files: this.files, - ignored: (path) => this.ignored(path), - oversized: (size) => this.oversized(size), + ignored: (path) => this.settings.isIgnored(path), + oversized: (size) => this.settings.isOversized(size), commit: async (next) => this.database.commit(next) }, initial @@ -735,8 +722,8 @@ export class Syncer { }); if ( next.excluded?.[head.documentId] !== undefined || - this.ignored(path) || - this.oversized(head.contentSize) + this.settings.isIgnored(path) || + this.settings.isOversized(head.contentSize) ) { return undefined; } @@ -748,7 +735,7 @@ export class Syncer { : undefined; if ( local !== undefined && - this.oversized(base64ToBytes(local.contentBase64).length) + this.settings.isOversized(base64ToBytes(local.contentBase64).length) ) { return undefined; } @@ -869,9 +856,9 @@ export class Syncer { for (const [id, oldPath] of Object.entries(before)) { const path = next.local[id]; if ( - this.ignored(oldPath) || + this.settings.isIgnored(oldPath) || this.unsyncablePaths.has(oldPath) || - (path !== undefined && this.ignored(path)) + (path !== undefined && this.settings.isIgnored(path)) ) { next.excluded ??= {}; if (next.excluded[id] === undefined) @@ -882,7 +869,7 @@ export class Syncer { for (const [id, path] of Object.entries(next.excluded ?? {})) { if ( before[id] === undefined && - (path === null || !this.ignored(path)) + (path === null || !this.settings.isIgnored(path)) ) delete next.excluded![id]; } @@ -906,7 +893,7 @@ export class Syncer { next.documents[id].materialized = false; if ( next.excluded?.[id] === undefined && - !this.ignored(path) && + !this.settings.isIgnored(path) && remote.entries[id] !== undefined && (!this.database.state.local[id] || !next.documents[id].base || @@ -971,11 +958,11 @@ export class Syncer { } for (const [id, path] of Object.entries(this.database.state.local)) { if ( - this.ignored(path) || + this.settings.isIgnored(path) || this.database.state.excluded?.[id] !== undefined || !this.database.state.documents[id]?.materialized || this.unsyncablePaths.has(path) || - this.oversized( + 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 ) @@ -985,7 +972,9 @@ export class Syncer { const snapshot = await this.files.snapshot(path); if ( !snapshot || - this.oversized(base64ToBytes(snapshot.contentBase64).length) + this.settings.isOversized( + base64ToBytes(snapshot.contentBase64).length + ) ) { continue; } @@ -1088,7 +1077,7 @@ export class Syncer { doc?.rejected && doc.observedHash === doc.rejected.hash && doc.base?.hash !== doc.rejected.hash && - !this.ignored(path) && + !this.settings.isIgnored(path) && state.excluded?.[id] === undefined ) return doc.rejected.message; diff --git a/frontend/sync-client/src/utils/data-structures/locks.test.ts b/frontend/sync-client/src/utils/data-structures/locks.test.ts index f7f0acf2..e325496b 100644 --- a/frontend/sync-client/src/utils/data-structures/locks.test.ts +++ b/frontend/sync-client/src/utils/data-structures/locks.test.ts @@ -1,292 +1,46 @@ -import { describe, it, beforeEach } from "node:test"; -import assert from "node:assert"; -import { Logger } from "../../tracing/logger"; -import type { RelativePath } from "../../persistence/database"; -import { Locks } from "./locks"; -import { awaitAll } from "../await-all"; -import { sleep } from "../sleep"; -import { SyncResetError } from "../../errors/errors"; +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { Lock } from "./locks"; -describe("withLock", () => { - const testPath: RelativePath = "test/document/path"; - const testPath2: RelativePath = "test/document/path2"; - const logger = new Logger(); - - // eslint-disable-next-line @typescript-eslint/init-declarations - let locks: Locks; - - beforeEach(() => { - locks = new Locks(logger); +test("queued operations run in order and retain their results", async () => { + const lock = new Lock(); + const held = Promise.withResolvers(); + const started = Promise.withResolvers(); + const calls: number[] = []; + const first = lock.withLock(async () => { + calls.push(1); + started.resolve(); + await held.promise; + calls.push(2); + return "first"; }); - - it("should execute function with single key lock", async () => { - let executionCount = 0; - const result = await locks.withLock(testPath, () => { - executionCount++; - return "success"; - }); - - assert.strictEqual(result, "success"); - assert.strictEqual(executionCount, 1); + const second = lock.withLock(() => { + calls.push(3); + return "second"; }); - - it("should execute async function with single key lock", async () => { - let executionCount = 0; - const result = await locks.withLock(testPath, async () => { - executionCount++; - await sleep(10); - return "async-success"; - }); - - assert.strictEqual(result, "async-success"); - assert.strictEqual(executionCount, 1); - }); - - it("should execute function with multiple key locks", async () => { - let executionCount = 0; - const result = await locks.withLock([testPath, testPath2], () => { - executionCount++; - return "multi-success"; - }); - - assert.strictEqual(result, "multi-success"); - assert.strictEqual(executionCount, 1); - }); - - it("should sort multiple keys to prevent deadlocks", async () => { - const executionOrder: string[] = []; - - // Start two concurrent operations with keys in different orders - const promise1 = locks.withLock([testPath2, testPath], async () => { - executionOrder.push("operation1-start"); - await sleep(50); - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock([testPath, testPath2], async () => { - executionOrder.push("operation2-start"); - await sleep(50); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - // One operation should complete entirely before the other starts - assert.deepStrictEqual(executionOrder, [ - "operation1-start", - "operation1-end", - "operation2-start", - "operation2-end" - ]); - }); - - it("should serialize access to same key", async () => { - const executionOrder: string[] = []; - - const promise1 = locks.withLock(testPath, async () => { - executionOrder.push("operation1-start"); - await sleep(50); - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock(testPath, async () => { - executionOrder.push("operation2-start"); - await sleep(30); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - assert.deepStrictEqual(executionOrder, [ - "operation1-start", - "operation1-end", - "operation2-start", - "operation2-end" - ]); - }); - - it("should allow concurrent access to different keys", async () => { - const executionOrder: string[] = []; - - const promise1 = locks.withLock(testPath, async () => { - executionOrder.push("operation1-start"); - await sleep(50); - - executionOrder.push("operation1-end"); - return "result1"; - }); - - const promise2 = locks.withLock(testPath2, async () => { - executionOrder.push("operation2-start"); - await sleep(30); - executionOrder.push("operation2-end"); - return "result2"; - }); - - const [result1, result2] = await awaitAll([promise1, promise2]); - - assert.strictEqual(result1, "result1"); - assert.strictEqual(result2, "result2"); - // Both operations should run concurrently - assert.strictEqual(executionOrder[0], "operation1-start"); - assert.strictEqual(executionOrder[1], "operation2-start"); - }); - - it("should release locks even if function throws", async () => { - const error = new Error("test error"); - - await assert.rejects( - locks.withLock(testPath, () => { - throw error; - }), - { message: "test error" } - ); - - // Lock should be released, allowing another operation - const result = await locks.withLock( - testPath, - () => "success-after-error" - ); - assert.strictEqual(result, "success-after-error"); - }); - - it("should release locks even if async function throws", async () => { - const error = new Error("async test error"); - - await assert.rejects( - locks.withLock(testPath, async () => { - await sleep(10); - - throw error; - }), - { message: "async test error" } - ); - - // Lock should be released, allowing another operation - const result = await locks.withLock( - testPath, - () => "success-after-async-error" - ); - assert.strictEqual(result, "success-after-async-error"); - }); - - it("should handle empty array of keys", async () => { - const result = await locks.withLock([], () => "empty-keys"); - assert.strictEqual(result, "empty-keys"); - }); - - it("should maintain FIFO order for multiple waiters", async () => { - const executionOrder: string[] = []; - - // Start first operation that holds the lock - const firstPromise = locks.withLock(testPath, async () => { - executionOrder.push("first-start"); - await sleep(100); - executionOrder.push("first-end"); - return "first"; - }); - - // Small delay to ensure first operation starts - await sleep(10); - - // Queue second and third operations - const secondPromise = locks.withLock(testPath, async () => { - executionOrder.push("second-start"); - await sleep(50); - executionOrder.push("second-end"); - return "second"; - }); - - const thirdPromise = locks.withLock(testPath, async () => { - executionOrder.push("third-start"); - await sleep(20); - executionOrder.push("third-end"); - return "third"; - }); - - const [first, second, third] = await awaitAll([ - firstPromise, - secondPromise, - thirdPromise - ]); - - assert.strictEqual(first, "first"); - assert.strictEqual(second, "second"); - assert.strictEqual(third, "third"); - assert.deepStrictEqual(executionOrder, [ - "first-start", - "first-end", - "second-start", - "second-end", - "third-start", - "third-end" - ]); + const third = lock.withLock(() => { + calls.push(4); + return "third"; }); + await started.promise; + assert.deepEqual(calls, [1]); + held.resolve(); + assert.equal(await first, "first"); + assert.equal(await second, "second"); + assert.equal(await third, "third"); + assert.deepEqual(calls, [1, 2, 3, 4]); }); -describe("reset", () => { - const testPath: RelativePath = "test/document/path"; - const logger = new Logger(); - - // eslint-disable-next-line @typescript-eslint/init-declarations - let locks: Locks; - - beforeEach(() => { - locks = new Locks(logger); - }); - - it("should reject pending waiters with SyncResetError while running operation completes", async () => { - const firstPromise = locks.withLock(testPath, async () => { - await sleep(2); - return "first"; - }); - - await sleep(1); - - const secondPromise = locks.withLock(testPath, async () => "second"); - void secondPromise.catch(() => {}); // eslint-disable-line @typescript-eslint/no-empty-function - - locks.reset(); - - assert.strictEqual(await firstPromise, "first"); - - await assert.rejects(secondPromise, (err: Error) => { - assert.ok(err instanceof SyncResetError); - return true; +for (const asynchronous of [false, true]) { + test(`a ${asynchronous ? "rejected promise" : "thrown error"} releases the lock`, async () => { + const lock = new Lock(); + const error = new Error("failed operation"); + const failed = lock.withLock(() => { + if (asynchronous) return Promise.reject(error); + throw error; }); + const next = lock.withLock(() => "continued"); + await assert.rejects(failed, (caught) => caught === error); + assert.equal(await next, "continued"); }); - - it("should allow locks to work normally after reset", async () => { - const firstPromise = locks.withLock(testPath, async () => { - await sleep(1); - return "first"; - }); - - await sleep(1); - - const secondPromise = locks.withLock(testPath, async () => "second"); - void secondPromise.catch(() => {}); // eslint-disable-line @typescript-eslint/no-empty-function - - locks.reset(); - - await firstPromise; - - const result = await locks.withLock(testPath, () => "after-reset"); - assert.strictEqual(result, "after-reset"); - }); - - it("should handle reset with no pending operations", async () => { - locks.reset(); - - const result = await locks.withLock(testPath, () => "success"); - assert.strictEqual(result, "success"); - }); -}); +} diff --git a/frontend/sync-client/src/utils/data-structures/locks.ts b/frontend/sync-client/src/utils/data-structures/locks.ts index 096cacdf..34bf2e60 100644 --- a/frontend/sync-client/src/utils/data-structures/locks.ts +++ b/frontend/sync-client/src/utils/data-structures/locks.ts @@ -1,166 +1,11 @@ -import { SyncResetError } from "../../errors/errors"; -import type { Logger } from "../../tracing/logger"; -import { awaitAll } from "../await-all"; - -/** - * Manages exclusive locks on items to prevent concurrent modifications. - * Locks are granted in FIFO order. - * - * @template T The type of the key used for locking - */ -export class Locks { - /** Currently locked keys */ - private readonly locked = new Set(); - - /** Queue of resolve functions waiting for each key */ - private readonly waiters = new Map< - T, - [() => unknown, (err: unknown) => unknown][] - >(); - - public constructor(private readonly logger?: Logger) {} - - /** - * Executes a function while holding exclusive locks on one or more keys. - * - * This method ensures that the provided function runs with exclusive access to the - * specified key(s). Multiple keys are sorted to prevent deadlocks when different - * operations request the same keys in different orders. - * - * @template R The return type of the function to execute - * @param keyOrKeys A single key or array of keys to lock during function execution - * @param fn The function to execute while holding the lock(s). Can be sync or async. - * @returns A Promise that resolves to the return value of the executed function - * - * @example - * ```typescript - * // Lock a single key - * const result = await locks.withLock('file1', () => { - * // Critical section - only one operation can access 'file1' at a time - * return processFile('file1'); - * }); - * - * // Lock multiple keys (prevents deadlocks through consistent ordering) - * await locks.withLock(['file1', 'file2'], async () => { - * // Critical section - exclusive access to both files - * await moveFile('file1', 'file2'); - * }); - * ``` - * - * @throws Any error thrown by the provided function will be propagated after locks are released - */ - public async withLock( - keyOrKeys: T | T[], - fn: () => R | Promise - ): Promise { - const keys = Array.isArray(keyOrKeys) ? keyOrKeys : [keyOrKeys]; - - // Deduplicate keys to prevent deadlock from acquiring same lock twice - const uniqueKeys = Array.from(new Set(keys)); - uniqueKeys.sort((a, b) => String(a).localeCompare(String(b))); // Ensure consistent order to prevent deadlocks - - await awaitAll(uniqueKeys.map(async (key) => this.waitForLock(key))); - - try { - return await fn(); - } finally { - uniqueKeys.forEach((key) => { - this.unlock(key); - }); - } - } - - public reset(): void { - // Resolve all waiting promises before clearing to prevent deadlock - // Any operation waiting for a lock will be granted access immediately - for (const waiting of this.waiters.values()) { - for (const [_, reject] of waiting) { - reject(new SyncResetError()); - } - } - this.locked.clear(); - this.waiters.clear(); - } - - /** - * Attempts to acquire a lock immediately without waiting. - * Must call `unlock()` if successful. - * - * @param key The key to lock - * @returns `true` if lock acquired, `false` if already locked - */ - public tryLock(key: T): boolean { - if (this.locked.has(key)) { - return false; - } - - this.locked.add(key); - - return true; - } - - /** - * Waits to acquire a lock, blocking until available. - * Operations are queued in FIFO order. Must call `unlock()` when done. - * - * @param key The key to wait for and lock - * @returns Promise that resolves when lock is acquired - */ - public async waitForLock(key: T): Promise { - if (this.tryLock(key)) { - return Promise.resolve(); - } - - this.logger?.debug(`Waiting for lock on ${key}`); - - return new Promise((resolve, reject) => { - // DefaultDict behavior - let waiting = this.waiters.get(key); - if (!waiting) { - waiting = []; - this.waiters.set(key, waiting); - } - - waiting.push([resolve, reject]); - }); - } - - /** - * Releases a lock and grants access to the next waiting operation in FIFO order. - * Removes the key from locked set if no waiters. - * - * @param key The key to unlock - * @throws {Error} If key is not currently locked - */ - public unlock(key: T): void { - if (!this.locked.has(key)) { - return; - } - - // Remove first waiter to ensure FIFO order - const [resolveNextWaiting, _] = this.waiters.get(key)?.shift() ?? []; - - if (resolveNextWaiting) { - this.logger?.debug(`Granted lock on ${key}`); - resolveNextWaiting(); - } else { - this.locked.delete(key); - } - } -} - +/** Serializes operations in arrival order; failures do not block the queue. */ export class Lock { - private readonly locks: Locks; + private tail: Promise = Promise.resolve(); - public constructor(logger?: Logger) { - this.locks = new Locks(logger); - } - - public async withLock(fn: () => R | Promise): Promise { - return this.locks.withLock(true, fn); - } - - public reset(): void { - this.locks.reset(); + // eslint-disable-next-line @typescript-eslint/promise-function-async -- Preserve the queued operation promise without an extra async wrapper. + public withLock(fn: () => R | Promise): Promise { + const operation = this.tail.then(fn); + this.tail = operation.catch(() => undefined); + return operation; } } diff --git a/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts b/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts index c64bff18..258db398 100644 --- a/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts +++ b/frontend/sync-client/src/utils/debugging/slow-web-socket-factory.ts @@ -1,5 +1,5 @@ import { sleep } from "../sleep"; -import { Locks } from "../data-structures/locks"; +import { Lock } from "../data-structures/locks"; import type { Logger } from "../../tracing/logger"; export function slowWebSocketFactory( @@ -8,10 +8,8 @@ export function slowWebSocketFactory( ): typeof WebSocket { // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion return class FlakyWebSocket extends WebSocket { - private static readonly RECEIVE_KEY = "websocket-receive"; - private static readonly SEND_KEY = "websocket-send"; - - private readonly locks = new Locks(logger); + private readonly receiving = new Lock(); + private readonly sending = new Lock(); public set onopen(callback: ((event: Event) => void) | null) { super.onopen = async (event: Event): Promise => { @@ -25,18 +23,15 @@ export function slowWebSocketFactory( public set onmessage(callback: ((event: MessageEvent) => void) | null) { super.onmessage = async (event: MessageEvent): Promise => { - await this.locks.withLock( - FlakyWebSocket.RECEIVE_KEY, - async () => { - if (jitterScaleInSeconds > 0) { - await sleep( - Math.random() * jitterScaleInSeconds * 1000 - ); - } - - callback?.(event); + await this.receiving.withLock(async () => { + if (jitterScaleInSeconds > 0) { + await sleep( + Math.random() * jitterScaleInSeconds * 1000 + ); } - ); + + callback?.(event); + }); }; } @@ -70,7 +65,7 @@ export function slowWebSocketFactory( data: string | ArrayBufferLike | Blob | ArrayBufferView ): Promise { // maintain message order - await this.locks.withLock(FlakyWebSocket.SEND_KEY, async () => { + await this.sending.withLock(async () => { if (jitterScaleInSeconds > 0) { await sleep(Math.random() * jitterScaleInSeconds * 1000); } diff --git a/frontend/test-support/protocol-audit.test.ts b/frontend/test-support/protocol-audit.test.ts index 2cb7d782..bf888cad 100644 --- a/frontend/test-support/protocol-audit.test.ts +++ b/frontend/test-support/protocol-audit.test.ts @@ -63,8 +63,7 @@ test( JSON.stringify({ type: "handshake", token, - deviceId: randomUUID(), - lastSeenVaultUpdateId: 0 + deviceId: randomUUID() }) ); ws.onmessage = () => resolve(true); @@ -302,8 +301,7 @@ test( JSON.stringify({ type: "handshake", token, - deviceId, - lastSeenVaultUpdateId: 0 + deviceId }) ); ws.onmessage = () => resolve(); diff --git a/frontend/test-support/protocol.test.ts b/frontend/test-support/protocol.test.ts index 50b0a38d..6e96bda6 100644 --- a/frontend/test-support/protocol.test.ts +++ b/frontend/test-support/protocol.test.ts @@ -282,7 +282,7 @@ for (const [kind, format, originalPath] of [ } test( - "concurrent CAS commits have identical HTTP/event-log/WebSocket total order and reconnect catchup", + "WebSocket hints trigger HTTP catchup in commit order, including after reconnect", { timeout: 20_000 }, async () => { const server = makeServer(); @@ -291,17 +291,29 @@ test( await server.start(); const url = `${server.remoteUri}/vaults/order-${randomUUID()}`; const received: number[] = []; - const connect = async (after: number) => { + let catchup = Promise.resolve(); + let catchupError: unknown; + const connect = async () => { ws = new WebSocket(`${url.replace(/^http/, "ws")}/ws`); ws.addEventListener("message", (message) => { - const batch = JSON.parse(String(message.data)) as { - type: string; - events: { eventId: number }[]; - }; - if (batch.type === "vaultEvents") - received.push( - ...batch.events.map((event) => event.eventId) - ); + const hint = JSON.parse(String(message.data)); + if (hint.type !== "vaultChanged") return; + catchup = catchup + .then(async () => { + assert.deepEqual(hint, { type: "vaultChanged" }); + const batch = await getJson<{ + events: { eventId: number }[]; + }>( + `${url}/events-since?after=${received.at(-1) ?? 0}`, + token + ); + received.push( + ...batch.events.map((event) => event.eventId) + ); + }) + .catch((error) => { + catchupError = error; + }); }); await new Promise((resolve, reject) => { ws!.onopen = () => resolve(); @@ -311,12 +323,11 @@ test( JSON.stringify({ type: "handshake", token, - deviceId: "order-observer", - lastSeenVaultUpdateId: after + deviceId: "order-observer" }) ); }; - await connect(0); + await connect(); const ids = Array.from({ length: 12 }, () => randomUUID()); const requests = ids.map((id, i) => ({ id, @@ -360,10 +371,11 @@ test( const deadline = Date.now() + 5_000; while (received.length < head && Date.now() < deadline) await new Promise((resolve) => setTimeout(resolve, 10)); + if (catchupError) throw catchupError; assert.deepEqual( received, Array.from({ length: head }, (_, i) => i + 1), - "WebSocket reordered, lost or duplicated an event" + "HTTP catchup after WebSocket hints lost or duplicated an event" ); }; await waitForEvents(13); @@ -375,7 +387,7 @@ test( `${url}/documents/${ids[0]}`, content("next", replies[0].vaultUpdateId) ); - await connect(13); + await connect(); await waitForEvents(14); const log = await getJson<{ events: { eventId: number }[] }>( `${url}/events-since?after=0`, diff --git a/frontend/test-support/sync-fixture.ts b/frontend/test-support/sync-fixture.ts index 30bf6a85..79047f5d 100644 --- a/frontend/test-support/sync-fixture.ts +++ b/frontend/test-support/sync-fixture.ts @@ -11,7 +11,6 @@ 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"; import { EventListeners } from "../sync-client/src/utils/data-structures/event-listeners"; -import { globsToRegexes } from "../sync-client/src/utils/globs-to-regexes"; import type { ServerConfig } from "../sync-client/src/services/server-config"; import type { SyncService } from "../sync-client/src/services/sync-service"; import type { WebSocketManager } from "../sync-client/src/services/websocket-manager"; @@ -81,11 +80,7 @@ export async function fixture( config, undefined, { entries: () => changes, flush: async () => {} }, - (path, size) => - size > settings.getSettings().maxFileSizeMB * 1024 * 1024 || - globsToRegexes(settings.getSettings().ignorePatterns, logger).some( - (pattern) => pattern.test(path) - ) + (path, size) => settings.isIgnored(path) || settings.isOversized(size) ); const websocket = { onWebSocketStatusChanged: new EventListeners(), diff --git a/sync-server/README.md b/sync-server/README.md index 27e68c20..a5c40318 100644 --- a/sync-server/README.md +++ b/sync-server/README.md @@ -86,13 +86,11 @@ content/file-manifest version IDs are the event IDs that created them. Rejected requests and retries of accepted requests append no events. Timestamps never determine event ordering. -The WebSocket handshake retains `lastSeenVaultUpdateId`. The server sends -`{type: "vaultEvents", headEventId, events}` in increasing order, including events -originated by that client. Each connection drains the database through one sender. -In-memory notifications only wake it; polling covers missed wakes and requests -interrupted after commit. Reconnect with the last durably processed event ID. The -HTTP replay endpoint uses the same history, so losing the final notification is -recoverable. +WebSocket messages are `{type: "vaultChanged"}` hints. Clients fetch ordered +history through HTTP from their last incorporated event ID. The server compares +its current history checkpoint after broadcasts and periodically, so missed wakes +and requests interrupted after commit still trigger catchup. Reconnecting sends a +fresh hint without replaying events over the socket. Cursor messages are unchanged. A fresh client reads `/vault-snapshot`, applies it durably, then replays after its watermark. Successful HTTP responses include `X-Vault-Link-History`, containing an event ID and a random event-incarnation token. Clients persist and echo this diff --git a/sync-server/src/app_state/websocket/models.rs b/sync-server/src/app_state/websocket/models.rs index 32d97966..4eb1fa10 100644 --- a/sync-server/src/app_state/websocket/models.rs +++ b/sync-server/src/app_state/websocket/models.rs @@ -1,16 +1,13 @@ use serde::{Deserialize, Serialize}; use ts_rs::TS; -use crate::app_state::database::models::{DeviceId, DocumentId, EventBatch, VaultUpdateId}; +use crate::app_state::database::models::{DeviceId, DocumentId, VaultUpdateId}; #[derive(TS, Deserialize, Clone, Debug)] #[serde(rename_all = "camelCase")] pub struct WebSocketHandshake { pub token: String, pub device_id: DeviceId, - - #[ts(as = "Option")] - pub last_seen_vault_update_id: Option, } #[derive(TS, Deserialize, Clone, Debug)] @@ -67,6 +64,6 @@ pub enum WebSocketClientMessage { #[serde(rename_all = "camelCase", tag = "type")] #[ts(export)] pub enum WebSocketServerMessage { - VaultEvents(EventBatch), + VaultChanged, CursorPositions(CursorPositionFromServer), } diff --git a/sync-server/src/consts.rs b/sync-server/src/consts.rs index 83d99a82..e1fee209 100644 --- a/sync-server/src/consts.rs +++ b/sync-server/src/consts.rs @@ -20,4 +20,4 @@ pub const DEFAULT_LOG_LEVEL: LogLevel = LogLevel::Info; pub const DEFAULT_MERGEABLE_FILE_EXTENSIONS: &[&str] = &["md", "txt"]; -pub const SUPPORTED_API_VERSION: u32 = 4; +pub const SUPPORTED_API_VERSION: u32 = 5; diff --git a/sync-server/src/server/endpoints/websocket.rs b/sync-server/src/server/endpoints/websocket.rs index 3a97911b..a4b779de 100644 --- a/sync-server/src/server/endpoints/websocket.rs +++ b/sync-server/src/server/endpoints/websocket.rs @@ -69,26 +69,21 @@ async fn websocket( let device = authenticated.handshake.device_id; let session = state.cursors.register_connection(&vault, &device).await; - let mut after = authenticated - .handshake - .last_seen_vault_update_id - .unwrap_or(0); + let mut last_checkpoint = None; let mut notifications = state.broadcasts.get_receiver(vault.clone()).await; let mut timer = tokio::time::interval(Duration::from_secs(2)); timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); info!("WebSocket connected to vault {vault}"); - // Exactly one sender drains durable events in commit order. In-memory - // notifications are only wakeups, so lag and commit/notify crashes are safe. + // HTTP owns event replay. Poll the cheap checkpoint to recover missed + // broadcasts, including a commit followed by an interrupted notification. let result = async { loop { - let batch = state.database.events_after(&vault, after).await.map_err(server_error)?; - - if !batch.events.is_empty() { - let head = batch.end_event_id.unwrap_or(batch.head_event_id); - send_update_over_websocket(&WebSocketServerMessage::VaultEvents(batch), &mut sender).await?; - after = head; + let checkpoint = state.database.history_checkpoint(&vault).await.map_err(server_error)?; + if last_checkpoint.as_ref() != Some(&checkpoint) { + send_update_over_websocket(&WebSocketServerMessage::VaultChanged, &mut sender).await?; + last_checkpoint = Some(checkpoint); } tokio::select! { @@ -99,7 +94,7 @@ async fn websocket( send_update_over_websocket(&WebSocketServerMessage::CursorPositions(positions), &mut sender).await?; }, - Err(RecvError::Lagged(_)) => {}, // Drain the durable log on the next iteration. + Err(RecvError::Lagged(_)) => {}, // Check the durable head on the next iteration. Err(RecvError::Closed) => break, },