From 9244e5d2203b63f6515a7ca5db2cc774c2646c36 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sun, 27 Sep 2026 21:56:44 +0100 Subject: [PATCH] Small clean up --- .forgejo/workflows/check.yml | 2 +- .forgejo/workflows/e2e.yml | 2 +- .forgejo/workflows/publish-plugin.yml | 2 +- frontend/sync-client/src/consts.ts | 7 ++- .../metadata-persistence-provider.ts | 2 +- .../sync-client/src/persistence/requests.ts | 15 ++++++ .../src/services/websocket-manager.test.ts | 23 ++++----- .../debugging/slow-web-socket-factory.ts | 48 +++++++------------ 8 files changed, 54 insertions(+), 47 deletions(-) create mode 100644 frontend/sync-client/src/persistence/requests.ts diff --git a/.forgejo/workflows/check.yml b/.forgejo/workflows/check.yml index 40e01dea..8602030e 100644 --- a/.forgejo/workflows/check.yml +++ b/.forgejo/workflows/check.yml @@ -28,7 +28,7 @@ jobs: - name: Setup Rust toolchain uses: dtolnay/rust-toolchain@stable with: - toolchain: "1.92.0" + toolchain: "1.94.0" components: clippy, rustfmt - name: Lint & test diff --git a/.forgejo/workflows/e2e.yml b/.forgejo/workflows/e2e.yml index c6e46ea4..9a8a7354 100644 --- a/.forgejo/workflows/e2e.yml +++ b/.forgejo/workflows/e2e.yml @@ -39,7 +39,7 @@ jobs: - name: Setup Rust toolchain uses: dtolnay/rust-toolchain@stable with: - toolchain: "1.92.0" + toolchain: "1.94.0" components: clippy, rustfmt - name: Expand scheduled fuzz campaigns diff --git a/.forgejo/workflows/publish-plugin.yml b/.forgejo/workflows/publish-plugin.yml index 25a652aa..899742c9 100644 --- a/.forgejo/workflows/publish-plugin.yml +++ b/.forgejo/workflows/publish-plugin.yml @@ -30,7 +30,7 @@ jobs: - name: Setup Rust toolchain uses: dtolnay/rust-toolchain@stable with: - toolchain: "1.92.0" + toolchain: "1.94.0" components: clippy, rustfmt - name: Install cross-compilation tools diff --git a/frontend/sync-client/src/consts.ts b/frontend/sync-client/src/consts.ts index edb133fb..1b62f437 100644 --- a/frontend/sync-client/src/consts.ts +++ b/frontend/sync-client/src/consts.ts @@ -1,6 +1,9 @@ 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 WEBSOCKET_DISCONNECT_TIMEOUT_IN_S = 10; +export const CURSOR_HEARTBEAT_INTERVAL_MS = 15_000; +export const WEBSOCKET_RECEIVE_TIMEOUT_MS = CURSOR_HEARTBEAT_INTERVAL_MS * 3; + +export const HISTORY_HEADER = "x-vault-link-history"; +export const HISTORY_MISMATCH_HEADER = "x-vault-link-history-mismatch"; diff --git a/frontend/sync-client/src/persistence/metadata-persistence-provider.ts b/frontend/sync-client/src/persistence/metadata-persistence-provider.ts index b0f9b974..b001bc1f 100644 --- a/frontend/sync-client/src/persistence/metadata-persistence-provider.ts +++ b/frontend/sync-client/src/persistence/metadata-persistence-provider.ts @@ -1,4 +1,4 @@ -/** Replace the complete metadata value atomically. */ +// Replace the complete metadata value atomically. export interface MetadataPersistenceProvider { load: () => Promise; save: (data: T) => Promise; diff --git a/frontend/sync-client/src/persistence/requests.ts b/frontend/sync-client/src/persistence/requests.ts new file mode 100644 index 00000000..a0201ff8 --- /dev/null +++ b/frontend/sync-client/src/persistence/requests.ts @@ -0,0 +1,15 @@ +import type { SyncEventType } from "../services/protocol-types"; +import type { PutFileContent } from "../services/types/PutFileContent"; +import type { PushFileManifest } from "../services/types/PushFileManifest"; +import type { DocumentId } from "./database"; + +// Exact wire payload retained until acknowledgement, including after rejection. +export type PendingRequest = { rejection?: string } & ( + | { + type: SyncEventType.Content; + documentId: DocumentId; + request: PutFileContent; + hash: string; + } + | { type: SyncEventType.FileManifest; request: PushFileManifest } +); diff --git a/frontend/sync-client/src/services/websocket-manager.test.ts b/frontend/sync-client/src/services/websocket-manager.test.ts index 587c7da0..bf8b02fb 100644 --- a/frontend/sync-client/src/services/websocket-manager.test.ts +++ b/frontend/sync-client/src/services/websocket-manager.test.ts @@ -1,3 +1,4 @@ +import { WebSocketMessageType } from "./protocol-types"; /* eslint-disable @typescript-eslint/no-unsafe-type-assertion */ import { describe, it, beforeEach } from "node:test"; import assert from "node:assert"; @@ -139,13 +140,13 @@ describe("WebSocketManager", () => { .webSocket; mockWs.simulateMessage({ - type: "vaultChanged" + type: WebSocketMessageType.VaultChanged }); mockWs.simulateMessage({ - type: "vaultChanged" + type: WebSocketMessageType.VaultChanged }); mockWs.simulateMessage({ - type: "vaultChanged" + type: WebSocketMessageType.VaultChanged }); await new Promise((resolve) => setTimeout(resolve, 100)); @@ -175,7 +176,7 @@ describe("WebSocketManager", () => { .webSocket; mockWs.simulateMessage({ - type: "cursorPositions", + type: WebSocketMessageType.CursorPositions, clients: [{ deviceId: "other-device", cursors: [] }] }); @@ -317,7 +318,7 @@ describe("WebSocketManager", () => { const mockWs = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; mockWs.simulateMessage({ - type: "vaultChanged" + type: WebSocketMessageType.VaultChanged }); await new Promise((resolve) => setTimeout(resolve, 10)); @@ -479,11 +480,11 @@ for (const closeBehavior of ["throws", "stays open"]) { assert.equal(manager.isWebSocketConnected, false); assert.equal(connected, false); socket.simulateMessage({ - type: "vaultChanged" + type: WebSocketMessageType.VaultChanged }); delayedMessage?.( new MockMessageEvent("message", { - data: JSON.stringify({ type: "vaultChanged" }) + data: JSON.stringify({ type: WebSocketMessageType.VaultChanged }) }) ); await manager.waitUntilFinished(); @@ -514,12 +515,12 @@ it("stop retires the socket immediately but drains messages already being handle const socket = (manager as unknown as { webSocket: MockWebSocket }) .webSocket; assert.deepEqual(JSON.parse(socket.sentMessages[0] ?? ""), { - type: "handshake", + type: WebSocketMessageType.Handshake, token: "test-token", deviceId: "device" }); manager.onRemoteVaultUpdateReceived.add(async () => release.promise); - socket.simulateMessage({ type: "vaultChanged" }); + socket.simulateMessage({ type: WebSocketMessageType.VaultChanged }); let finished = false; const stopping = manager.stop().then(() => { finished = true; @@ -654,9 +655,9 @@ it("does not reconnect while messages keep arriving before the deadline", async assert.equal(attempts, 1); context.mock.timers.tick(10_000); - socket.simulateMessage({ type: "vaultChanged" }); + socket.simulateMessage({ type: WebSocketMessageType.VaultChanged }); context.mock.timers.tick(10_000); - socket.simulateMessage({ type: "cursorPositions", clients: [] }); + socket.simulateMessage({ type: WebSocketMessageType.CursorPositions, clients: [] }); context.mock.timers.tick(WEBSOCKET_RECEIVE_TIMEOUT_MS - 20_000); assert.equal(manager.isWebSocketConnected, true); assert.equal(disconnected, 0); 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 258db398..97381cab 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 @@ -6,6 +6,12 @@ export function slowWebSocketFactory( jitterScaleInSeconds: number, logger: Logger ): typeof WebSocket { + const delay = async (): Promise => { + if (jitterScaleInSeconds > 0) { + await sleep(Math.random() * jitterScaleInSeconds * 1000); + } + }; + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion return class FlakyWebSocket extends WebSocket { private readonly receiving = new Lock(); @@ -13,10 +19,7 @@ export function slowWebSocketFactory( public set onopen(callback: ((event: Event) => void) | null) { super.onopen = async (event: Event): Promise => { - if (jitterScaleInSeconds > 0) { - await sleep(Math.random() * jitterScaleInSeconds * 1000); - } - + await delay(); callback?.(event); }; } @@ -24,12 +27,7 @@ export function slowWebSocketFactory( public set onmessage(callback: ((event: MessageEvent) => void) | null) { super.onmessage = async (event: MessageEvent): Promise => { await this.receiving.withLock(async () => { - if (jitterScaleInSeconds > 0) { - await sleep( - Math.random() * jitterScaleInSeconds * 1000 - ); - } - + await delay(); callback?.(event); }); }; @@ -37,18 +35,14 @@ export function slowWebSocketFactory( public set onclose(callback: ((event: CloseEvent) => void) | null) { super.onclose = async (event: CloseEvent): Promise => { - if (jitterScaleInSeconds > 0) { - await sleep(Math.random() * jitterScaleInSeconds * 1000); - } + await delay(); callback?.(event); }; } public set onerror(callback: ((event: Event) => void) | null) { super.onerror = async (event: Event): Promise => { - if (jitterScaleInSeconds > 0) { - await sleep(Math.random() * jitterScaleInSeconds * 1000); - } + await delay(); callback?.(event); }; } @@ -56,21 +50,15 @@ export function slowWebSocketFactory( public send( data: string | ArrayBufferLike | Blob | ArrayBufferView ): void { - this.waitingSend(data).catch((error: unknown) => { - logger.error(`Error sending WebSocket message: ${error}`); - }); - } - - private async waitingSend( - data: string | ArrayBufferLike | Blob | ArrayBufferView - ): Promise { // maintain message order - await this.sending.withLock(async () => { - if (jitterScaleInSeconds > 0) { - await sleep(Math.random() * jitterScaleInSeconds * 1000); - } - super.send(data); - }); + void this.sending + .withLock(async () => { + await delay(); + super.send(data); + }) + .catch((error: unknown) => { + logger.error(`Error sending WebSocket message: ${error}`); + }); } } as unknown as typeof WebSocket; }