Small clean up

.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

.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

.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

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";

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<T> {
     load: () => Promise<T | undefined>;
     save: (data: T) => Promise<void>;

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 }
+);

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);

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<void> => {
+        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<void> => {
-                if (jitterScaleInSeconds > 0) {
+                await delay();
-                    await sleep(Math.random() * jitterScaleInSeconds * 1000);
-                }
-
                 callback?.(event);
             };
         }
@@ -24,12 +27,7 @@ export function slowWebSocketFactory(
         public set onmessage(callback: ((event: MessageEvent) => void) | null) {
             super.onmessage = async (event: MessageEvent): Promise<void> => {
                 await this.receiving.withLock(async () => {
-                    if (jitterScaleInSeconds > 0) {
+                    await delay();
-                        await sleep(
-                            Math.random() * jitterScaleInSeconds * 1000
-                        );
-                    }
-
                     callback?.(event);
                 });
             };
@@ -37,18 +35,14 @@ export function slowWebSocketFactory(
 
         public set onclose(callback: ((event: CloseEvent) => void) | null) {
             super.onclose = async (event: CloseEvent): Promise<void> => {
-                if (jitterScaleInSeconds > 0) {
+                await delay();
-                    await sleep(Math.random() * jitterScaleInSeconds * 1000);
-                }
                 callback?.(event);
             };
         }
 
         public set onerror(callback: ((event: Event) => void) | null) {
             super.onerror = async (event: Event): Promise<void> => {
-                if (jitterScaleInSeconds > 0) {
+                await delay();
-                    await sleep(Math.random() * jitterScaleInSeconds * 1000);
-                }
                 callback?.(event);
             };
         }
@@ -56,20 +50,14 @@ 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<void> {
             // maintain message order
-            await this.sending.withLock(async () => {
+            void this.sending
-                if (jitterScaleInSeconds > 0) {
+                .withLock(async () => {
-                    await sleep(Math.random() * jitterScaleInSeconds * 1000);
+                    await delay();
-                }
                     super.send(data);
+                })
+                .catch((error: unknown) => {
+                    logger.error(`Error sending WebSocket message: ${error}`);
                 });
         }
     } as unknown as typeof WebSocket;