Fix web sync fallback during reconnect (#350)

* Fix web sync fallback during reconnect

* Fix lint in reconnect sync fallback

Author: appflowy

src/application/services/js-services/__tests__/sync.test.ts

@@ -96,6 +96,113 @@ const mockSync = (clientCount: number, tracer = defaultTracer): SyncContext[] =>
   return clients;
 };
 
+interface QueuedSyncBus {
+  clients: SyncContext[];
+  online: boolean[];
+  texts: Y.Text[];
+  disconnect: (index: number) => void;
+  reconnect: (index: number) => void;
+  publishManifest: (index: number) => void;
+  publishManifests: () => void;
+  drain: () => void;
+}
+
+const createQueuedSyncBus = (clientCount: number): QueuedSyncBus => {
+  outboxMock.__clearTestClients();
+
+  const guid = random.uuidv4();
+  const clients: SyncContext[] = [];
+  const online = Array.from({ length: clientCount }, () => true);
+  const queue: Array<{ sender: number; message: messages.IMessage }> = [];
+
+  const enqueue = (sender: number, message: messages.IMessage) => {
+    if (!online[sender]) return;
+    queue.push({ sender, message });
+  };
+
+  for (let i = 0; i < clientCount; i += 1) {
+    const doc = new Y.Doc({ guid });
+
+    clients.push({
+      doc,
+      awareness: new awarenessProtocol.Awareness(doc),
+      emit: (message) => enqueue(i, message),
+      collabType: Types.Document,
+    });
+  }
+
+  clients.forEach((client) => initSync(client));
+  queue.length = 0;
+
+  const publishManifest = (index: number) => {
+    const client = clients[index];
+
+    client.emit({
+      collabMessage: {
+        objectId: client.doc.guid,
+        collabType: client.collabType,
+        syncRequest: {
+          stateVector: Y.encodeStateVector(client.doc),
+          lastMessageId: { timestamp: 0, counter: 0 },
+          version: client.doc.version,
+        },
+      },
+    });
+  };
+
+  const publishManifests = () => {
+    clients.forEach((_, index) => publishManifest(index));
+  };
+
+  const drain = () => {
+    let guard = 0;
+
+    while (queue.length > 0) {
+      if (guard > 1_000) {
+        throw new Error('sync queue did not settle');
+      }
+
+      guard += 1;
+      const { sender, message } = queue.shift()!;
+
+      if (!online[sender] || !message.collabMessage) continue;
+
+      clients.forEach((client, index) => {
+        if (index !== sender && online[index]) {
+          handleMessage(client, message.collabMessage!);
+        }
+      });
+    }
+  };
+
+  const disconnect = (index: number) => {
+    online[index] = false;
+  };
+
+  const reconnect = (index: number) => {
+    online[index] = true;
+    publishManifests();
+  };
+
+  return {
+    clients,
+    online,
+    texts: clients.map((client) => client.doc.getText('test')),
+    disconnect,
+    reconnect,
+    publishManifest,
+    publishManifests,
+    drain,
+  };
+};
+
+const expectTextConvergence = (texts: Y.Text[], expectedChars: string) => {
+  const values = texts.map((text) => text.toString());
+
+  expect(new Set(values).size).toBe(1);
+  expect([...values[0]].sort().join('')).toBe(expectedChars);
+};
+
 describe('sync protocol', () => {
   it('should exchange updates between client and server', () => {
     const [local, remote] = mockSync(2);
@@ -114,4 +221,146 @@ describe('sync protocol', () => {
     txt2.insert(5, ' World');
     expect(txt1.toString()).toEqual('Hello World');
   });
+
+  it('converges three online clients after concurrent same-position edits', () => {
+    const { texts, publishManifests, drain } = createQueuedSyncBus(3);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    publishManifests();
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
+
+  it('converges when one client edits offline and then reconnects', () => {
+    const { texts, disconnect, reconnect, publishManifest, drain } = createQueuedSyncBus(3);
+
+    disconnect(2);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    publishManifest(0);
+    publishManifest(1);
+    drain();
+
+    expect(texts[0].toString()).toEqual(texts[1].toString());
+    expect([...texts[0].toString()].sort().join('')).toBe('AB');
+    expect(texts[2].toString()).toBe('C');
+
+    reconnect(2);
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
+
+  it('converges when two offline clients reconnect in ascending order', () => {
+    const { texts, disconnect, reconnect, publishManifest, drain } = createQueuedSyncBus(3);
+
+    disconnect(0);
+    disconnect(1);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    publishManifest(2);
+    drain();
+    expect(texts.map((text) => text.toString())).toEqual(['A', 'B', 'C']);
+
+    reconnect(0);
+    drain();
+    expect(texts[0].toString()).toEqual(texts[2].toString());
+    expect([...texts[0].toString()].sort().join('')).toBe('AC');
+    expect(texts[1].toString()).toBe('B');
+
+    reconnect(1);
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
+
+  it('converges when two offline clients reconnect in reverse order', () => {
+    const { texts, disconnect, reconnect, publishManifest, drain } = createQueuedSyncBus(3);
+
+    disconnect(0);
+    disconnect(1);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    publishManifest(2);
+    drain();
+    expect(texts.map((text) => text.toString())).toEqual(['A', 'B', 'C']);
+
+    reconnect(1);
+    drain();
+    expect(texts[1].toString()).toEqual(texts[2].toString());
+    expect([...texts[1].toString()].sort().join('')).toBe('BC');
+    expect(texts[0].toString()).toBe('A');
+
+    reconnect(0);
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
+
+  it('converges when all clients edit offline and reconnect left to right', () => {
+    const { texts, disconnect, reconnect, drain } = createQueuedSyncBus(3);
+
+    disconnect(0);
+    disconnect(1);
+    disconnect(2);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    reconnect(0);
+    drain();
+    expect(texts.map((text) => text.toString())).toEqual(['A', 'B', 'C']);
+
+    reconnect(1);
+    drain();
+    expect(texts[0].toString()).toEqual(texts[1].toString());
+    expect([...texts[0].toString()].sort().join('')).toBe('AB');
+    expect(texts[2].toString()).toBe('C');
+
+    reconnect(2);
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
+
+  it('converges when all clients edit offline and reconnect right to left', () => {
+    const { texts, disconnect, reconnect, drain } = createQueuedSyncBus(3);
+
+    disconnect(0);
+    disconnect(1);
+    disconnect(2);
+
+    texts[0].insert(0, 'A');
+    texts[1].insert(0, 'B');
+    texts[2].insert(0, 'C');
+
+    reconnect(2);
+    drain();
+    expect(texts.map((text) => text.toString())).toEqual(['A', 'B', 'C']);
+
+    reconnect(1);
+    drain();
+    expect(texts[1].toString()).toEqual(texts[2].toString());
+    expect([...texts[1].toString()].sort().join('')).toBe('BC');
+    expect(texts[0].toString()).toBe('A');
+
+    reconnect(0);
+    drain();
+
+    expectTextConvergence(texts, 'ABC');
+  });
 });

src/application/services/js-services/http/collab-api.ts

@@ -13,6 +13,14 @@ import { Log } from '@/utils/log';
 
 import { APIResponse, APIError, executeAPIRequest, executeAPIVoidRequest, getAxios, parseRetryAfterSecs } from './core';
 
+export interface CollabFullSyncBatchResult {
+  objectId: string;
+  collabType: Types;
+  missingUpdate: Uint8Array;
+  serverStateVector: Uint8Array;
+  error?: string;
+}
+
 export async function updateCollab(
   workspaceId: string,
   objectId: string,
@@ -61,7 +69,7 @@ export async function collabFullSyncBatch(
     stateVector: Uint8Array;
     docState: Uint8Array;
   }>
-): Promise<void> {
+): Promise<CollabFullSyncBatchResult[]> {
   const url = `/api/workspace/v1/${workspaceId}/collab/full-sync/batch`;
 
   // Build the protobuf request
@@ -113,6 +121,7 @@ export async function collabFullSyncBatch(
   // Decode and check the response for errors
   const responseData = new Uint8Array(response.data);
   const batchResponse = collab.CollabBatchSyncResponse.decode(responseData);
+  const results: CollabFullSyncBatchResult[] = [];
 
   // Check for any errors in the results
   for (const result of batchResponse.results) {
@@ -123,7 +132,17 @@ export async function collabFullSyncBatch(
         error: result.error,
       });
     }
+
+    results.push({
+      objectId: result.objectId ?? '',
+      collabType: result.collabType as Types,
+      missingUpdate: result.missingUpdate ?? new Uint8Array(),
+      serverStateVector: result.serverStateVector ?? new Uint8Array(),
+      error: result.error || undefined,
+    });
   }
+
+  return results;
 }
 
 export async function getCollab(workspaceId: string, objectId: string, collabType: Types) {

src/application/services/js-services/sync-protocol.ts

@@ -40,6 +40,19 @@ export interface SyncContext {
    * Used by version reset flows where pending updates are stale and must be discarded.
    */
   discardPendingUpdates?: () => Promise<void>;
+  /**
+   * Called after a local Yjs update is observed. The WebSocket path still owns
+   * immediate delivery; this hook lets the app schedule a debounced HTTP full
+   * sync when the WebSocket is reconnecting.
+   */
+  onLocalUpdate?: (objectId: string) => void;
+  /**
+   * Called after this context participates in a WebSocket state-vector sync
+   * exchange. A completed manifest-style exchange reconciles the full Yjs
+   * state, so HTTP fallback dirty markers for the object can be cleared when
+   * the socket is open.
+   */
+  onManifestSync?: (objectId: string) => void;
   /**
    * Cleanup function to remove update/awareness observers and cancel debounced sends.
    * Set by initSync, called during deferred sync context cleanup.
@@ -83,6 +96,7 @@ const handleSyncRequest = (ctx: SyncContext, message: collab.ISyncRequest): void
       },
     },
   });
+  ctx.onManifestSync?.(doc.guid);
 };
 
 const handleAccessChanged = (ctx: SyncContext, message: collab.IAccessChanged): void => {
@@ -190,6 +204,7 @@ export const initSync = (ctx: SyncContext) => {
       version: doc.version ?? null,
       payload: update,
     });
+    ctx.onLocalUpdate?.(doc.guid);
   };
 
   const onDestroy = () => {