From e0e38597c33830d4a59a4a94250fcd32844c941e Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Tue, 22 Sep 2026 21:49:24 +0100 Subject: [PATCH] Pick up vault history id and refactor aborting --- .../src/services/fetch-controller.test.ts | 177 -------------- .../src/services/fetch-controller.ts | 149 ------------ .../src/services/sync-service.test.ts | 181 +++++++++++++++ .../sync-client/src/services/sync-service.ts | 215 +++++++++++------- .../src/types/vault-history-status.ts | 4 + frontend/sync-client/src/utils/abortable.ts | 23 ++ sync-server/src/consts.rs | 2 + sync-server/src/server/endpoints/websocket.rs | 4 +- 8 files changed, 342 insertions(+), 413 deletions(-) delete mode 100644 frontend/sync-client/src/services/fetch-controller.test.ts delete mode 100644 frontend/sync-client/src/services/fetch-controller.ts create mode 100644 frontend/sync-client/src/services/sync-service.test.ts create mode 100644 frontend/sync-client/src/types/vault-history-status.ts create mode 100644 frontend/sync-client/src/utils/abortable.ts 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..be8ae94c --- /dev/null +++ b/frontend/sync-client/src/services/sync-service.test.ts @@ -0,0 +1,181 @@ +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: [] }); + }, { get: () => undefined, save: async () => { } }); + await assert.rejects(service.getEvents(0), SyncResetError); + assert.equal(requests, 0); + await service.ping(); + assert.equal(requests, 1); + service.resume(); + assert.equal((await service.getEvents(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.getEvents(0); + service.pause(); + await assert.rejects(first, SyncResetError); + assert.equal(signals[0].aborted, true); + service.resume(); + await service.getEvents(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.getEvents(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.getEvents(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"); +}); + +test("history saves only new checkpoints, including a changed token at the same event ID", async () => { + let checkpoint: string | undefined; + let responseCheckpoint = "1:first"; + const saved: (string | undefined)[] = []; + const service = new SyncService( + "device", + settings(), + async () => + Response.json( + {}, + { headers: { "X-Vault-Link-History": responseCheckpoint } } + ), + { + get: () => checkpoint, + save: async (value) => { + checkpoint = value; + saved.push(value); + } + } + ); + service.resume(); + for (const value of [ + "1:first", + "1:first", + "2:next", + "1:first", + "2:replacement", + "2:replacement" + ]) { + responseCheckpoint = value; + await service.getEvents(0); + } + assert.deepEqual(saved, ["1:first", "2:next", "2:replacement"]); +}); diff --git a/frontend/sync-client/src/services/sync-service.ts b/frontend/sync-client/src/services/sync-service.ts index 3c5f604c..561c6928 100644 --- a/frontend/sync-client/src/services/sync-service.ts +++ b/frontend/sync-client/src/services/sync-service.ts @@ -1,105 +1,66 @@ import type { Settings } from "../persistence/settings"; -import type { FetchController } from "./fetch-controller"; -import type { Logger } from "../tracing/logger"; +import { abortable } from "../utils/abortable"; +import { VaultHistoryStatus } from "../types/vault-history-status"; 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"; import type { VaultSnapshot } from "./types/VaultSnapshot"; import type { PingResponse } from "./types/PingResponse"; -import { AuthenticationError, PermanentSyncError } from "../errors/errors"; +import { + AuthenticationError, + PermanentSyncError, + ServerHistoryChangedError, + SyncResetError +} from "../errors/errors"; 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 getUrl(path: string): string { - const { remoteUri, vaultName } = this.settings.getSettings(); - return `${remoteUri.replace(/\/$/u, "")}/vaults/${encodeURIComponent(vaultName)}${path}`; + public get vaultHistoryStatus(): VaultHistoryStatus { + return this.history.get() !== undefined + ? VaultHistoryStatus.STORED + : VaultHistoryStatus.MISSING; } - private async request( - path: string, - body?: unknown, - raw = false - ): Promise { - const response = await (raw ? this.rawFetch : this.client)( - this.getUrl(path), - { - method: body === undefined ? "GET" : "PUT", - body: body === undefined ? undefined : JSON.stringify(body), - signal: AbortSignal.timeout( - this.settings.getSettings().requestTimeoutMs - ), - headers: { - Authorization: `Bearer ${this.settings.getSettings().token}`, - "Device-Id": this.deviceId, - "Content-Type": "application/json" - } - } - ); - - if (!response.ok) { - this.throwForErrorResponse( - response, - `HTTP ${response.status}: ${await response.text()}` - ); - } - - return (await response.json()) as T; + public pause(): void { + this.session.abort(new SyncResetError()); } - private throwForErrorResponse( - response: Response, - message: string - ): never { - if (response.status === 401 || response.status === 403) { - throw new AuthenticationError(message); - } - if ( - response.status >= 400 && - response.status < 500 && - response.status !== 408 && - response.status !== 429 - ) { - throw new PermanentSyncError(message); - } - throw new Error(message); + public resume(): void { + this.session = new AbortController(); + } + + public async resetHistory(): Promise { + await this.history.save(undefined); } public async ping(): Promise { return this.request("/ping", undefined, true); } - public async vaultSnapshot(): Promise { + public async getVaultSnapshot(): Promise { return this.request("/vault-snapshot"); } - public async events(after: number): Promise { + public async getEvents(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 { @@ -113,8 +74,8 @@ export class SyncService { return this.request(`/documents/${id}`, request); } - public async get(id: string): Promise { - return this.request(`/documents/${id}`); + public async getDocumentMetadata(id: string): Promise { + return this.request(`/documents/${id}/metadata`); } public async getDocumentVersionContent({ @@ -124,27 +85,111 @@ export class SyncService { 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: { - Authorization: `Bearer ${this.settings.getSettings().token}` - } - } + return this.request( + `/documents/${documentId}/versions/${vaultUpdateId}/content`, + undefined, + false, + async (response) => new Uint8Array(await response.arrayBuffer()) ); + } + + private getHistoryHeaders(): Record { + const checkpoint = this.history.get(); + 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 !== null && + checkpoint !== "" + ) { + const previous = this.history.get(); + if (checkpoint === previous) return; + if ( + previous === undefined || + previous === "" || + Number(checkpoint.split(":")[0]) >= + Number(previous.split(":")[0]) + ) + await this.history.save(checkpoint); + } + } + + private checkHistory(response: Response): void { + if (response.headers.get("x-vault-link-history-mismatch") === "1") + throw new ServerHistoryChangedError( + "Server history changed; recovering local work" + ); + } + + private getUrl(path: string): string { + const { remoteUri, vaultName } = this.settings.getSettings(); + return `${remoteUri.replace(/\/$/u, "")}/vaults/${encodeURIComponent(vaultName)}${path}`; + } + + private async request( + path: string, + body?: unknown, + 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 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, + headers: { + ...(raw ? {} : this.getHistoryHeaders()), + 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, - `Cannot fetch version ${vaultUpdateId}: HTTP ${response.status}` + `HTTP ${response.status}: ${await abortable(signal, async () => response.text())}` ); } - return new Uint8Array(await response.arrayBuffer()); + // Finish metadata saves even if the transport is paused meanwhile. + if (!raw) await this.recordHistory(response); + + return abortable(signal, async () => decode(response)); + } + + private throwForErrorResponse(response: Response, message: string): never { + if (response.status === 401 || response.status === 403) { + throw new AuthenticationError(message); + } + + if ( + response.status >= 400 && + response.status < 500 && + response.status !== 408 && + response.status !== 429 + ) { + throw new PermanentSyncError(message); + } + + throw new Error(message); } } diff --git a/frontend/sync-client/src/types/vault-history-status.ts b/frontend/sync-client/src/types/vault-history-status.ts new file mode 100644 index 00000000..74478822 --- /dev/null +++ b/frontend/sync-client/src/types/vault-history-status.ts @@ -0,0 +1,4 @@ +export enum VaultHistoryStatus { + MISSING = "missing", + STORED = "stored" +} \ No newline at end of file diff --git a/frontend/sync-client/src/utils/abortable.ts b/frontend/sync-client/src/utils/abortable.ts new file mode 100644 index 00000000..96044429 --- /dev/null +++ b/frontend/sync-client/src/utils/abortable.ts @@ -0,0 +1,23 @@ +export 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); + } +} \ No newline at end of file diff --git a/sync-server/src/consts.rs b/sync-server/src/consts.rs index dc5f08b9..341fd9eb 100644 --- a/sync-server/src/consts.rs +++ b/sync-server/src/consts.rs @@ -17,7 +17,9 @@ pub const DEFAULT_MAX_CLIENTS_PER_VAULT: usize = 256; pub const DEFAULT_LOG_DIRECTORY: &str = "logs"; pub const DEFAULT_LOG_ROTATION_INTERVAL: Duration = Duration::from_secs(60 * 60 * 24); // 1 day pub const DEFAULT_LOG_LEVEL: LogLevel = LogLevel::Info; + pub const WEBSOCKET_SEND_TIMEOUT: Duration = Duration::from_secs(10); +pub const WEBSOCKET_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(5); pub const DEFAULT_MERGEABLE_FILE_EXTENSIONS: &[&str] = &["md", "txt"]; diff --git a/sync-server/src/server/endpoints/websocket.rs b/sync-server/src/server/endpoints/websocket.rs index e490cf90..0ef24919 100644 --- a/sync-server/src/server/endpoints/websocket.rs +++ b/sync-server/src/server/endpoints/websocket.rs @@ -8,7 +8,7 @@ use crate::{ utils::{get_authenticated_handshake, send_update_over_websocket}, }, }, - consts::WEBSOCKET_SEND_TIMEOUT, + consts::WEBSOCKET_HANDSHAKE_TIMEOUT, errors::{SyncServerError, client_error, server_error}, }; use axum::{ @@ -59,7 +59,7 @@ async fn websocket( let authenticated = get_authenticated_handshake( &state, &vault, - tokio::time::timeout(WEBSOCKET_SEND_TIMEOUT, receiver.next()) + tokio::time::timeout(WEBSOCKET_HANDSHAKE_TIMEOUT, receiver.next()) .await .map_err(|_| client_error(anyhow::anyhow!("WebSocket handshake deadline expired")))? .transpose()