Merge pull request #22845 from overleaf/em-project-snapshot-queue-refreshes

Concurrency control in ProjectSnapshot

GitOrigin-RevId: b62b886b59a67f2c694ef7cefcff5c32da3e4457

services/web/frontend/js/infrastructure/project-snapshot.ts

@@ -1,4 +1,5 @@
 import pLimit from 'p-limit'
+import OError from '@overleaf/o-error'
 import { Change, Chunk, Snapshot } from 'overleaf-editor-core'
 import { RawChange, RawChunk } from 'overleaf-editor-core/lib/types'
 import { FetchError, getJSON, postJSON } from '@/infrastructure/fetch-json'
@@ -12,45 +13,68 @@ export class ProjectSnapshot {
   private projectId: string
   private snapshot: Snapshot
   private version: number
-  private state: 'init' | 'refreshing' | 'ready'
   private blobStore: SimpleBlobStore
+  private refreshPromise: Promise<void>
+  private queuedRefreshPromise: Promise<void>
+  private state: ProjectSnapshotState
 
   constructor(projectId: string) {
     this.projectId = projectId
     this.snapshot = new Snapshot()
     this.version = 0
-    this.state = 'init'
+    this.refreshPromise = Promise.resolve()
+    this.queuedRefreshPromise = Promise.resolve()
+    this.state = new ProjectSnapshotState()
     this.blobStore = new SimpleBlobStore(this.projectId)
   }
 
+  /**
+   * Request a refresh of the snapshot.
+   *
+   * When the returned promise resolves, the snapshot is guaranteed to have been
+   * updated at least to the version of the document that was current when the
+   * function was called.
+   */
   async refresh() {
-    if (this.state === 'refreshing') {
-      // Prevent concurrent refreshes
-      return
-    }
+    switch (this.state.getState()) {
+      case 'init':
+        this.refreshPromise = this.initialize()
+        await this.refreshPromise
+        break
 
-    await flushHistory(this.projectId)
+      case 'ready':
+        this.refreshPromise = this.loadChanges()
+        await this.refreshPromise
+        break
 
-    if (this.state === 'init') {
-      const chunk = await fetchLatestChunk(this.projectId)
-      this.snapshot = chunk.getSnapshot()
-      this.snapshot.applyAll(chunk.getChanges())
-      this.version = chunk.getEndVersion()
-    } else {
-      const changes = await fetchLatestChanges(this.projectId, this.version)
-      this.snapshot.applyAll(changes)
-      this.version += changes.length
-    }
+      case 'refreshing':
+        this.queuedRefreshPromise = this.queueRefresh()
+        await this.queuedRefreshPromise
+        break
 
-    this.state = 'ready'
-    await this.loadDocs()
+      case 'queued-ready':
+      case 'queued-waiting':
+        await this.queuedRefreshPromise
+        break
+
+      default:
+        throw new OError('Unknown state for project snapshot', {
+          state: this.state.getState(),
+        })
+    }
   }
 
+  /**
+   * Get the list of paths to editable docs.
+   */
   getDocPaths(): string[] {
     const allPaths = this.snapshot.getFilePathnames()
     return allPaths.filter(path => this.snapshot.getFile(path)?.isEditable())
   }
 
+  /**
+   * Get the doc content at the given path.
+   */
   getDocContents(path: string): string | null {
     const file = this.snapshot.getFile(path)
     if (file == null) {
@@ -59,6 +83,52 @@ export class ProjectSnapshot {
     return file.getContent({ filterTrackedDeletes: true }) ?? null
   }
 
+  /**
+   * Initialize the snapshot using the project's latest chunk.
+   *
+   * This is run on the first refresh.
+   */
+  private async initialize() {
+    this.state.startRefresh()
+    await flushHistory(this.projectId)
+    const chunk = await fetchLatestChunk(this.projectId)
+    this.snapshot = chunk.getSnapshot()
+    this.snapshot.applyAll(chunk.getChanges())
+    this.version = chunk.getEndVersion()
+    await this.loadDocs()
+    this.state.endRefresh()
+  }
+
+  /**
+   * Apply changes since the last refresh.
+   *
+   * This is run on the second and subsequent refreshes
+   */
+  private async loadChanges() {
+    this.state.startRefresh()
+    await flushHistory(this.projectId)
+    const changes = await fetchLatestChanges(this.projectId, this.version)
+    this.snapshot.applyAll(changes)
+    this.version += changes.length
+    await this.loadDocs()
+    this.state.endRefresh()
+  }
+
+  /**
+   * Wait for the current refresh to complete, then start a refresh.
+   */
+  private async queueRefresh() {
+    this.state.queueRefresh()
+    await this.refreshPromise
+    await this.loadChanges()
+  }
+
+  /**
+   * Load all editable docs in the snapshot.
+   *
+   * This is done by converting any lazy file data into an "eager" file data. If
+   * a doc is already loaded, the load is a no-op.
+   */
   private async loadDocs() {
     const paths = this.getDocPaths()
     const limit = pLimit(DOWNLOAD_BLOBS_CONCURRENCY)
@@ -73,6 +143,106 @@ export class ProjectSnapshot {
   }
 }
 
+/**
+ * State machine for the project snapshot
+ *
+ * There are 5 states:
+ *
+ * - init: when the snapshot is built
+ * - refreshing: while the snapshot is refreshing
+ * - queued-waiting: while the snapshot is refreshing and another refresh is queued
+ * - queued-ready: when a refresh is queued, but no refresh is running
+ * - ready: when no refresh is running and no refresh is queued
+ *
+ * There are three transitions:
+ *
+ * - start: start a refresh operation
+ * - end: end a refresh operation
+ * - queue: queue a refresh operation
+ *
+ * Valid transitions are as follows:
+ *
+ *                       +------------+
+ *                       |   ready    |
+ *                       +------------+
+ *                         ^       |
+ *                         |       |
+ *                        end    start
+ *                         |       |
+ *                         |       v
+ * +------+              +------------+             +----------------+
+ * | init |----start---->| refreshing |---queue---> | queued-waiting |
+ * +------+              +------------+             +----------------+
+ *                             ^                             |
+ *                             |                             |
+ *                           start                          end
+ *                             |                             |
+ *                             |     +--------------+        |
+ *                             +-----| queued-ready |<-------+
+ *                                   +--------------+
+ *
+ * These transitions ensure that there are never two refreshes running
+ * concurrently. In every path, "start" and "end" transitions always alternate.
+ * You never have two consecutive "start" or two consecutive "end".
+ */
+class ProjectSnapshotState {
+  private state:
+    | 'init'
+    | 'refreshing'
+    | 'ready'
+    | 'queued-waiting'
+    | 'queued-ready' = 'init'
+
+  getState() {
+    return this.state
+  }
+
+  startRefresh() {
+    switch (this.state) {
+      case 'init':
+      case 'ready':
+      case 'queued-ready':
+        this.state = 'refreshing'
+        break
+
+      default:
+        throw new OError("Can't start a snapshot refresh in this state", {
+          state: this.state,
+        })
+    }
+  }
+
+  endRefresh() {
+    switch (this.state) {
+      case 'refreshing':
+        this.state = 'ready'
+        break
+
+      case 'queued-waiting':
+        this.state = 'queued-ready'
+        break
+
+      default:
+        throw new OError("Can't end a snapshot refresh in this state", {
+          state: this.state,
+        })
+    }
+  }
+
+  queueRefresh() {
+    switch (this.state) {
+      case 'refreshing':
+        this.state = 'queued-waiting'
+        break
+
+      default:
+        throw new OError("Can't queue a snapshot refresh in this state", {
+          state: this.state,
+        })
+    }
+  }
+}
+
 /**
  * Blob store that fetches blobs from the history service
  */