From ab9d4d4c9c8e17d3ab274f33465af524b88b9964 Mon Sep 17 00:00:00 2001 From: Eric Mc Sween <5454374+emcsween@users.noreply.github.com> Date: Thu, 16 Jan 2025 08:25:27 -0500 Subject: [PATCH] Merge pull request #22845 from overleaf/em-project-snapshot-queue-refreshes Concurrency control in ProjectSnapshot GitOrigin-RevId: b62b886b59a67f2c694ef7cefcff5c32da3e4457 --- .../js/infrastructure/project-snapshot.ts | 212 ++++++++++++++++-- 1 file changed, 191 insertions(+), 21 deletions(-) diff --git a/services/web/frontend/js/infrastructure/project-snapshot.ts b/services/web/frontend/js/infrastructure/project-snapshot.ts index d16f7b81ea..b6300e505b 100644 --- a/services/web/frontend/js/infrastructure/project-snapshot.ts +++ b/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 + private queuedRefreshPromise: Promise + 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 + + case 'ready': + this.refreshPromise = this.loadChanges() + await this.refreshPromise + break + + case 'refreshing': + this.queuedRefreshPromise = this.queueRefresh() + await this.queuedRefreshPromise + break + + case 'queued-ready': + case 'queued-waiting': + await this.queuedRefreshPromise + break + + default: + throw new OError('Unknown state for project snapshot', { + state: this.state.getState(), + }) } - - await flushHistory(this.projectId) - - 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 - } - - this.state = 'ready' - await this.loadDocs() } + /** + * 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 */