feat: 🎸 start synchronization when connection appears

Author: streamich

src/json-crdt-repo/local/server-crud/ServerCrudLocalHistory.ts

@@ -31,7 +31,7 @@ export class ServerCrudLocalHistory implements LocalHistory {
     const remote = (async () => {
       await this.sync.markDirty(collection, id);
       // TODO: use pushNewBlock instead?
-      const success = await this.sync.push(collection, id);
+      const success = await this.sync.sync(collection, id);
       if (!success) throw new Error('NOT_SYNCED');
     })();
     remote.catch(() => {});

src/json-crdt-repo/local/server-crud/ServerCrudLocalHistorySync.ts

@@ -1,9 +1,11 @@
 import {timeout} from 'thingies/lib/timeout';
+import {SESSION} from 'json-joy/lib/json-crdt-patch/constants';
+import {ts} from 'json-joy/lib/json-crdt-patch';
+import {once} from 'thingies';
 import type {RemoteBlockPatch} from '../../remote/types';
 import type {ServerCrudLocalHistoryCore} from './ServerCrudLocalHistoryCore';
 import type {BlockSyncMetadata} from './types';
-import {SESSION} from 'json-joy/lib/json-crdt-patch/constants';
-import {ts} from 'json-joy/lib/json-crdt-patch';
+import type {Subscription} from 'rxjs';
 
 const SYNC_FILE_NAME = 'sync.cbor';
 
@@ -25,29 +27,38 @@ export interface ServerCrudLocalHistorySyncOpts {
 }
 
 export class ServerCrudLocalHistorySync {
-  private syncLoopTimer: any = 0;
+  // private syncLoopTimer: any = 0;
+  private _conSub: Subscription | undefined = undefined;
 
   constructor(
     protected readonly opts: ServerCrudLocalHistorySyncOpts,
     protected readonly core: ServerCrudLocalHistoryCore,
   ) {}
 
+  @once
   public start(): void {
-    
+    this._conSub = this.core.connected$.subscribe((connected) => {
+      if (connected) {
+        this.syncAll().catch(() => {});
+      } else {
+
+      }
+    });
   }
 
+  @once
   public stop(): void {
-
+    this._conSub?.unsubscribe();
   }
 
   protected remoteTimeout(): number {
     return this.opts.remoteTimeout ?? 5000;
   }
 
-  public async push(collection: string[], id: string): Promise<boolean> {
-    return await this.lock<boolean>({collection, id}, async () => {
-      // TODO: handle case when this times out, but actually succeeds, so on re-sync it handles the case when the block is already synced.
+  public async sync(collection: string[], id: string): Promise<boolean> {
+    return await this.lockItemSync<boolean>({collection, id}, async () => {
       try {
+        // TODO: handle case when this times out, but actually succeeds, so on re-sync it handles the case when the block is already synced.
         return timeout(this.remoteTimeout(), async () => {
           const core = this.core;
           if (!core.connected$.getValue()) return false;
@@ -86,7 +97,10 @@ export class ServerCrudLocalHistorySync {
     });
   }
 
-  public async lock<T>(
+  /**
+   * Locks a specific item for synchronization.
+   */
+  private async lockItemSync<T>(
     {
       collection,
       id,
@@ -141,28 +155,42 @@ export class ServerCrudLocalHistorySync {
     }
   }
 
-  protected async * listDirty(collection: string[] = ['sync', 'dirty']): AsyncIterableIterator<{collection: string[]; id: string}> {
+  protected async * listDirty(collection: string[] = ['sync', 'dirty']): AsyncIterableIterator<ItemId> {
     for await (const entry of this.core.crud.scan(collection)) {
       if (entry.type === 'collection') yield* this.listDirty([...collection, entry.id]);
       else yield {collection, id: entry.id};
     }
   }
 
-  protected async * syncDirty(): AsyncIterableIterator<[block: {collection: string[]; id: string}, success: boolean]> {
+  protected async * syncDirty(): AsyncIterableIterator<SyncResult> {
     for await (const block of this.listDirty()) {
-      const {collection, id} = block;
-      const success = await this.push(collection, id);
-      yield [block, success];
+      const {collection: [_sync, _dirty, ...collection], id} = block;
+      try {
+        const success = await this.sync(collection, id);
+        yield [block, success];
+      } catch (error) {
+        yield [block, false, error];      
+      }
     }
   }
 
-  protected async syncAllDirty(): Promise<SyncResultList> {
+  public async syncAll(): Promise<SyncResultList> {
+    const locks = this.core.locks;
+    if (locks.isLocked('sync')) return [];
     const list: SyncResultList = [];
-    for await (const result of this.syncDirty()) list.push(result);
-    return list;
+    const duration = 30000;
+    const start = Date.now();
+    return await locks.lock('sync', duration, 3000)(async () => {
+      for await (const result of this.syncDirty()) {
+        list.push(result);
+        const now = Date.now();
+        if (now - start + 100 > duration) break;
+      }
+      return list;
+    });
   }
 }
 
 export type ItemId = {collection: string[], id: string};
-export type SyncResult = [block: ItemId, success: boolean];
+export type SyncResult = [block: ItemId, success: boolean, err?: Error | unknown];
 export type SyncResultList = SyncResult[];

src/json-crdt-repo/local/server-crud/__tests__/ServerCrudLocalHistory.spec.ts

@@ -7,7 +7,7 @@ import {Model, nodes, s} from 'json-joy/lib/json-crdt';
 import {Log} from 'json-joy/lib/json-crdt/log/Log';
 import {BehaviorSubject} from 'rxjs';
 import {setup as remoteSetup} from '../../../remote/__tests__/setup';
-import {tick} from 'thingies';
+import {tick, until} from 'thingies';
 import {SESSION} from 'json-joy/lib/json-crdt-patch/constants';
 
 const setup = async (
@@ -30,11 +30,15 @@ const setup = async (
     connected$: new BehaviorSubject(true),
     ...opts.local,
   });
+  local.sync.start();
   const log = Log.fromNewModel(Model.withLogicalClock(sid));
   log.end.api.root({foo: 'bar'});
   log.end.api.flush();
   const genId = () => Date.now().toString(36) + Math.random().toString(36).slice(2);
   const id = genId();
+  const stop = () => {
+    local.sync.stop();
+  };
   return {
     remote,
     fs,
@@ -47,6 +51,7 @@ const setup = async (
     log,
     genId,
     id,
+    stop,
   };
 };
 
@@ -171,6 +176,41 @@ describe('.create()', () => {
     });
   });
 
+  describe('when not connected, but connection resumes', () => {
+    const setupNotConnected = async () => {
+      const connected$ = new BehaviorSubject(false);
+      const deps = await setup({
+        local: {
+          connected$,
+          sync: {
+            remoteTimeout: 100,
+          },
+        },
+      });
+      return {
+        ...deps,
+        connected$,
+      };
+    };
+
+    test('does not store the block on remote at first, synchronizes it when connection resumes', async () => {
+      const kit = await setupNotConnected();
+      const res = await kit.local.create(['my', 'col'], kit.log, kit.id);
+      expect(kit.remote.services.blocks.stats().blocks).toBe(0);
+      try {
+        await res.remote;
+        throw 'not this error';
+      } catch (error) {
+        expect(error).toEqual(new Error('NOT_SYNCED'));
+      }
+      await tick(50);
+      expect(kit.remote.services.blocks.stats().blocks).toBe(0);
+      kit.connected$.next(true);
+      await until(() => kit.remote.services.blocks.stats().blocks === 1);
+      expect(kit.remote.services.blocks.stats().blocks).toBe(1);
+    });
+  });
+
   describe('when remote call fails', () => {
     const setupFaultyConnection = () => {
       const remote = remoteSetup();
@@ -364,7 +404,8 @@ describe('.create()', () => {
       const isDirty = await kit.local.sync.isDirty(['my', 'col'], kit.id);
       expect(isDirty).toBe(true);
       await tick(200);
-      expect(isDirty).toBe(false);
+      const isDirty2 = await kit.local.sync.isDirty(['my', 'col'], kit.id);
+      expect(isDirty2).toBe(false);
     });
   });
 });