fix: 🐛 pass concurrent construction tests

Author: streamich

src/json-crdt-repo/local/level/LevelLocalRepo.ts

@@ -715,6 +715,7 @@ export class LevelLocalRepo implements LocalRepo {
         } catch (error) {
           if (error instanceof Error && error.message === 'EXISTS')
             // TODO: make sure reset does not happen, if models are the same.
+            // TODO: Check for `req.time` in `_syncRead`.
             return await this._syncRead(id);
           throw error;
         }
@@ -732,6 +733,17 @@ export class LevelLocalRepo implements LocalRepo {
 
   private async _syncMerge(req: LocalRepoSyncRequest): Promise<LocalRepoSyncResponse> {
     const {id, patches} = req;
+    let lastKnownTime: number = 0;
+    const reqTime = req.time
+    if (typeof reqTime === 'number') {
+      lastKnownTime = reqTime;
+      const firstPatch = patches?.[0];
+      if (firstPatch?.getId()?.sid === SESSION.GLOBAL)
+        lastKnownTime = firstPatch.getId()!.time + firstPatch.span() - 1;
+    } else if (patches?.length) {
+      const firstPatchTime = patches?.[0]?.getId()?.time;
+      if (typeof firstPatchTime === 'number') lastKnownTime = firstPatchTime - 1;
+    }
     const keyBase = await this.blockKeyBase(id);
     if (!patches || !patches.length) throw new Error('EMPTY_BATCH');
     const writtenPatches: Uint8Array[] = [];
@@ -740,9 +752,12 @@ export class LevelLocalRepo implements LocalRepo {
     // TODO: Return correct response.
     // TODO: Check that remote state is in sync, too.
     let needsReset = false;
+    let cursorBehind = false;
     const didPush = await this.lockBlock(keyBase, async () => {
       const [tip, meta] = await Promise.all([this.readFrontierTip(keyBase), this.readMeta(keyBase)]);
+      if (meta.seq > -1 && (typeof req.cursor !== 'number' || req.cursor < meta.seq)) cursorBehind = true;
       let nextTick = meta.time + 1;
+      if (lastKnownTime < meta.time) needsReset = true;
       cursor = meta.seq;
       if (tip) {
         const tipTime = tip.getId()?.time ?? 0;
@@ -785,7 +800,7 @@ export class LevelLocalRepo implements LocalRepo {
     });
     if (!didPush && !needsReset) {
       const merge = await this.readFrontier0(keyBase);
-      return {cursor, merge};
+      return {cursor, merge, cursorBehind};
     }
     const remote = this.markDirtyAndSync(keyBase, id)
       .then(() => {})
@@ -795,9 +810,9 @@ export class LevelLocalRepo implements LocalRepo {
       });
     if (needsReset) {
       const {cursor, model} = await this._syncRead0(keyBase);
-      return {cursor, model, remote};
+      return {cursor, model, remote, cursorBehind};
     }
-    return {cursor, remote};
+    return {cursor, remote, cursorBehind};
   }
 
   protected async readLocal0(keyBase: string): Promise<[model: Model, cursor: LevelLocalRepoCursor]> {

src/json-crdt-repo/local/types.ts

@@ -87,6 +87,12 @@ export interface LocalRepoSyncRequest {
    */
   id: BlockId;
 
+  /**
+   * Logical clock time of the local operations which the client has caught up
+   * to.
+   */
+  time?: number;
+
   /**
    * The last known cursor returned in the `.sync()` call response. The cursor
    * should be omitted in the first `.sync()` call, and then set to the value
@@ -115,6 +121,13 @@ export interface LocalRepoSyncResponse {
    */
   cursor: undefined | unknown;
 
+  /**
+   * Set to true if the client is behind the remote. When true, the client
+   * should call `.getIf()` after a short wait period to check if the remote
+   * is indeed ahead.
+   */
+  cursorBehind?: boolean;
+
   /**
    * Model snapshot that the client should reset its "start" state to. The
    * `Model` is sent when the *sync* call detects that the client is behind the

src/json-crdt-repo/session/EditSession.ts

@@ -87,7 +87,14 @@ export class EditSession<N extends JsonNode = JsonNode<any>> {
       const length = patches.length;
       // TODO: After async call check that sync state is still valid. New patches, might have been added.
       if (length || this.cursor === undefined) {
-        const res = await this.repo.sync({id: this.id, patches, cursor: this.cursor, session: this.session});
+        let time = this.start.clock.time - 1;
+        const res = await this.repo.sync({
+          id: this.id,
+          patches,
+          time,
+          cursor: this.cursor,
+          session: this.session
+        });
         if (this._stopped) return null;
         // TODO: After sync call succeeds, remove the patches from the log.
         if (length) {
@@ -96,16 +103,33 @@ export class EditSession<N extends JsonNode = JsonNode<any>> {
           if (lastId) this.log.advanceTo(lastId);
           this.start.applyBatch(patches);
         }
-        if (typeof res.cursor !== undefined) this.cursor = res.cursor;
+        // "cursor" shall not be returned from .sync() call. The cursor shall update
+        // only when remote model changes are received, during local .sync() write
+        // only the local model is updated.
+        if (typeof res.cursor !== undefined) {
+          this.cursor = res.cursor;
+        }
         if (res.model) {
           this._syncRace(() => {
             this.reset(<any>res.model!);
           });
-        } else if (res.merge) {
+        } else if (res.merge && res.merge) {
           this._syncRace(() => {
             this.merge(<any>res.merge!);
           });
         }
+        // if (res.cursorBehind) {
+        //   setTimeout(async () => {
+        //     if (!this._stopped) return;
+        //     const get = await this.repo.getIf({
+        //       id: this.id,
+        //       cursor: this.cursor
+        //     });
+        //     if (!this._stopped) return;
+        //     if (!get) return;
+        //     this.reset(<any>get.model);
+        //   }, 50);
+        // }
         return {remote: res.remote};
       } else {
         const res = await this.repo.getIf({id: this.id, time: this.model.clock.time - 1, cursor: this.cursor});
@@ -209,6 +233,12 @@ export class EditSession<N extends JsonNode = JsonNode<any>> {
 
   private onEvent = (event: LocalRepoEvent): void => {
     if (this._stopped) return;
+    if ((event as LocalRepoMergeEvent).merge) {
+      const cursor = (event as LocalRepoMergeEvent).cursor;
+      if (cursor !== undefined) {
+        this.cursor = cursor;
+      }
+    }
     if ((event as LocalRepoRebaseEvent).rebase) {
       if ((event as LocalRepoRebaseEvent).session === this.session) return;
     }

src/json-crdt-repo/session/__tests__/EditSession.sync.spec.ts

@@ -118,6 +118,38 @@ describe('sync', () => {
     await kit.stop();
   });
 
+  test('sessions created just after the first one, converges in state', async () => {
+    const kit = await setup();
+    const schema = s.obj({a: s.con('a')});
+    const {session: session1} = kit.sessions.make({id: kit.blockId, schema, session: 1});
+    const {session: session2} = kit.sessions.make({id: kit.blockId, schema, session: 2});
+    await session1.sync();
+    await session2.sync();
+    expect(session1.log.patches.size()).toBe(0);
+    session1.model.api.obj([]).set({b: 'b'});
+    session1.model.api.obj([]).set({c: 'c'});
+    await tick(5);
+    session1.model.api.obj([]).set({d: 'd'});
+    const {session: session3} = kit.sessions.make({id: kit.blockId, pull: true, schema, session: 3});
+    await tick(5);
+    session1.model.api.obj([]).set({e: 'e'});
+    await tick(5);
+    session1.model.api.obj([]).set({f: 'f'});
+    await session1.sync();
+    await until(async () => {
+      try {
+        expect(session1.model.view()).toEqual(session3.model.view());
+        return true;
+      } catch {
+        return false;
+      }
+    });
+    await session1.dispose();
+    await session2.dispose();
+    await session3.dispose();
+    await kit.stop();
+  });
+
   test('sessions converge to the same view', async () => {
     const kit = await setup();
     const schema = s.obj({a: s.con('a')});
@@ -141,7 +173,21 @@ describe('sync', () => {
     await until(async () => {
       try {
         expect(session1.model.view()).toEqual(session2.model.view());
+        return true;
+      } catch {
+        return false;
+      }
+    });
+    await until(async () => {
+      try {
         expect(session1.model.view()).toEqual(session3.model.view());
+        return true;
+      } catch {
+        return false;
+      }
+    });
+    await until(async () => {
+      try {
         expect(session1.model.view()).toEqual(session4.model.view());
         return true;
       } catch {
@@ -155,6 +201,7 @@ describe('sync', () => {
     await session1.dispose();
     await session2.dispose();
     await session3.dispose();
+    await session4.dispose();
     await kit.stop();
   });
 });