stashing to handle out-of-sequence revisions

client/lib/patch-db.ts

@@ -2,7 +2,7 @@ import { merge, Observable, of } from 'rxjs'
 import { concatMap, finalize, tap } from 'rxjs/operators'
 import { Source } from './source/source'
 import { Store } from './store'
-import { DBCache } from './types'
+import { DBCache, Http } from './types'
 
 export { Operation } from 'fast-json-patch'
 
@@ -11,9 +11,10 @@ export class PatchDB<T> {
 
   constructor (
     private readonly sources: Source<T>[],
+    private readonly http: Http<T>,
     private readonly initialCache: DBCache<T>,
   ) {
-    this.store = new Store(this.initialCache)
+    this.store = new Store(this.http, this.initialCache)
   }
 
   sync$ (): Observable<DBCache<T>> {

client/lib/store.ts

@@ -1,17 +1,20 @@
-import { DBCache, Dump, Revision, Update } from './types'
+import { DBCache, Dump, Http, Revision, Update } from './types'
 import { applyPatch, getValueByPointer } from 'fast-json-patch'
 import { BehaviorSubject, Observable } from 'rxjs'
 import { finalize } from 'rxjs/operators'
+import BTree from 'sorted-btree'
 
 export class Store<T> {
   cache: DBCache<T>
   sequence$: BehaviorSubject<number>
   private nodes: { [path: string]: BehaviorSubject<any> } = { }
+  private stashed = new BTree<number, Revision>()
 
   constructor (
-    readonly initialCache: DBCache<T>,
+    private readonly http: Http<T>,
+    private readonly initialCache: DBCache<T>,
   ) {
-    this.cache = initialCache
+    this.cache = this.initialCache
     this.sequence$ = new BehaviorSubject(initialCache.sequence)
   }
 
@@ -34,22 +37,57 @@ export class Store<T> {
   }
 
   update (update: Update<T>): void {
-    if ((update as Revision).patch) {
+    // if stale, return
-      if (this.cache.sequence + 1 !== update.id) throw new Error(`Outdated sequence: current: ${this.cache.sequence}, new: ${update.id}`)
+    if (update.id <= this.cache.sequence) return
-      applyPatch(this.cache.data, (update as Revision).patch, true, true);
-      (update as Revision).patch.forEach(op => {
-        this.updateNodesByPath(op.path)
-      })
 
+    if (this.isRevision(update)) {
+      this.handleRevision(update)
     } else {
-      this.cache.data = (update as Dump<T>).value
+      this.handleDump(update)
-      this.updateNodesByPath('')
     }
-    this.cache.sequence = update.id
-    this.sequence$.next(this.cache.sequence)
   }
 
-  updateNodesByPath (revisionPath: string) {
+  reset (): void {
+    Object.values(this.nodes).forEach(node => node.complete())
+    this.stashed.clear()
+  }
+
+  private handleRevision (revision: Revision): void {
+    // stash the revision
+    this.stashed.set(revision.id, revision)
+    // if revision is futuristic, fetch missing revisions and return
+    if (revision.id > this.cache.sequence + 1) {
+      this.http.getRevisions(this.cache.sequence)
+      return
+    // if revision is next in line, apply contiguous stashed
+    } else {
+      this.processStashed(revision.id)
+    }
+  }
+
+  private handleDump (dump: Dump<T>): void {
+    this.cache.data = dump.value
+    this.stashed.deleteRange(this.cache.sequence, dump.id, false)
+    this.updateNodesByPath('')
+    this.updateSequence(dump.id)
+    this.processStashed(dump.id + 1)
+  }
+
+  private processStashed (id: number): void {
+    while (true) {
+      const revision = this.stashed.get(id)
+      if (!revision) break
+      applyPatch(this.cache.data, revision.patch, true, true)
+      revision.patch.map(op => {
+        this.updateNodesByPath(op.path)
+      })
+      this.updateSequence(id)
+      id++
+    }
+    this.stashed.deleteRange(0, id, false)
+  }
+
+  private updateNodesByPath (revisionPath: string) {
     Object.keys(this.nodes).forEach(nodePath => {
       if (!this.nodes[nodePath]) return
       if (nodePath.includes(revisionPath) || revisionPath.includes(nodePath)) {
@@ -63,7 +101,12 @@ export class Store<T> {
     })
   }
 
-  reset (): void {
+  private updateSequence (sequence: number): void {
-    Object.values(this.nodes).forEach(node => node.complete())
+    this.cache.sequence = sequence
+    this.sequence$.next(sequence)
+  }
+
+  private isRevision (update: Update<T>): update is Revision {
+    return !!(update as Revision).patch
   }
 }

client/package-lock.json

@@ -14,6 +14,7 @@
         "mobx": "^6.1.4",
         "mobx-utils": "^6.0.3",
         "rxjs": "^6.6.3",
+        "sorted-btree": "^1.5.0",
         "uuid": "^8.3.2"
       },
       "devDependencies": {
@@ -1178,6 +1179,11 @@
         "randombytes": "^2.1.0"
       }
     },
+    "node_modules/sorted-btree": {
+      "version": "1.5.0",
+      "resolved": "https://registry.npmjs.org/sorted-btree/-/sorted-btree-1.5.0.tgz",
+      "integrity": "sha512-1KzY80r3VpwGLGN/9oWjReUml3czxKfLz4iMV8Ro9KAHCg9xt0HwTkcb20JR+sHCiR5WUJ6uMAbe/HB3gy1qYA=="
+    },
     "node_modules/source-map": {
       "version": "0.6.1",
       "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.6.1.tgz",
@@ -2562,6 +2568,11 @@
         "randombytes": "^2.1.0"
       }
     },
+    "sorted-btree": {
+      "version": "1.5.0",
+      "resolved": "https://registry.npmjs.org/sorted-btree/-/sorted-btree-1.5.0.tgz",
+      "integrity": "sha512-1KzY80r3VpwGLGN/9oWjReUml3czxKfLz4iMV8Ro9KAHCg9xt0HwTkcb20JR+sHCiR5WUJ6uMAbe/HB3gy1qYA=="
+    },
     "source-map": {
       "version": "0.6.1",
       "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.6.1.tgz",

client/package.json

@@ -16,6 +16,7 @@
     "mobx": "^6.1.4",
     "mobx-utils": "^6.0.3",
     "rxjs": "^6.6.3",
+    "sorted-btree": "^1.5.0",
     "uuid": "^8.3.2"
   },
   "devDependencies": {