Merge pull request #23282 from overleaf/em-async-await-persist-changes

Convert the history changes import code to async/await

GitOrigin-RevId: 6421fcaaf3bac69a3404754f935b4902979b4689

services/history-v1/api/controllers/project_import.js

@@ -1,6 +1,9 @@
+// @ts-check
+
 'use strict'
 
-const BPromise = require('bluebird')
+const { expressify } = require('@overleaf/promise-utils')
+
 const HTTPStatus = require('http-status')
 
 const core = require('overleaf-editor-core')
@@ -22,7 +25,7 @@ const persistChanges = storage.persistChanges
 
 const render = require('./render')
 
-exports.importSnapshot = function importSnapshot(req, res, next) {
+async function importSnapshot(req, res) {
   const projectId = req.swagger.params.project_id.value
   const rawSnapshot = req.swagger.params.snapshot.value
 
@@ -34,21 +37,21 @@ exports.importSnapshot = function importSnapshot(req, res, next) {
     return render.unprocessableEntity(res)
   }
 
-  return chunkStore
-    .initializeProject(projectId, snapshot)
-    .then(function (projectId) {
-      res.status(HTTPStatus.OK).json({ projectId })
-    })
-    .catch(err => {
-      if (err instanceof chunkStore.AlreadyInitialized) {
-        render.conflict(res)
-      } else {
-        next(err)
-      }
-    })
+  let historyId
+  try {
+    historyId = await chunkStore.initializeProject(projectId, snapshot)
+  } catch (err) {
+    if (err instanceof chunkStore.AlreadyInitialized) {
+      return render.conflict(res)
+    } else {
+      throw err
+    }
+  }
+
+  res.status(HTTPStatus.OK).json({ projectId: historyId })
 }
 
-exports.importChanges = function importChanges(req, res, next) {
+async function importChanges(req, res, next) {
   const projectId = req.swagger.params.project_id.value
   const rawChanges = req.swagger.params.changes.value
   const endVersion = req.swagger.params.end_version.value
@@ -59,7 +62,7 @@ exports.importChanges = function importChanges(req, res, next) {
   try {
     changes = rawChanges.map(Change.fromRaw)
   } catch (err) {
-    logger.error({ err, projectId }, err.message)
+    logger.warn({ err, projectId }, 'failed to parse changes')
     return render.unprocessableEntity(res)
   }
 
@@ -76,65 +79,61 @@ exports.importChanges = function importChanges(req, res, next) {
   const batchBlobStore = new BatchBlobStore(blobStore)
   const hashCheckBlobStore = new HashCheckBlobStore(blobStore)
 
-  function loadFiles() {
+  async function loadFiles() {
     const blobHashes = new Set()
-    changes.forEach(function findBlobHashesToPreload(change) {
+    for (const change of changes) {
+      // This populates the set blobHashes with blobs referred to in the change
       change.findBlobHashes(blobHashes)
-    })
-
-    function lazyLoadChangeFiles(change) {
-      return change.loadFiles('lazy', batchBlobStore)
     }
 
-    return batchBlobStore
-      .preload(Array.from(blobHashes))
-      .then(function lazyLoadChangeFilesWithBatching() {
-        return BPromise.each(changes, lazyLoadChangeFiles)
-      })
+    await batchBlobStore.preload(Array.from(blobHashes))
+
+    for (const change of changes) {
+      await change.loadFiles('lazy', batchBlobStore)
+    }
   }
 
-  function buildResultSnapshot(resultChunk) {
-    return BPromise.resolve(
-      resultChunk || chunkStore.loadLatest(projectId)
-    ).then(function (chunk) {
-      const snapshot = chunk.getSnapshot()
-      snapshot.applyAll(chunk.getChanges())
-      return snapshot.store(hashCheckBlobStore)
-    })
+  async function buildResultSnapshot(resultChunk) {
+    const chunk = resultChunk || (await chunkStore.loadLatest(projectId))
+    const snapshot = chunk.getSnapshot()
+    snapshot.applyAll(chunk.getChanges())
+    const rawSnapshot = await snapshot.store(hashCheckBlobStore)
+    return rawSnapshot
   }
 
-  return loadFiles()
-    .then(function () {
-      return persistChanges(projectId, changes, limits, endVersion)
-    })
-    .then(function (result) {
-      if (returnSnapshot === 'none') {
-        res.status(HTTPStatus.CREATED).json({})
-      } else {
-        return buildResultSnapshot(result && result.currentChunk).then(
-          function (rawSnapshot) {
-            res.status(HTTPStatus.CREATED).json(rawSnapshot)
-          }
-        )
-      }
-    })
-    .catch(err => {
-      if (
-        err instanceof Chunk.ConflictingEndVersion ||
-        err instanceof TextOperation.UnprocessableError ||
-        err instanceof File.NotEditableError ||
-        err instanceof FileMap.PathnameError ||
-        err instanceof Snapshot.EditMissingFileError ||
-        err instanceof chunkStore.ChunkVersionConflictError
-      ) {
-        // If we failed to apply operations, that's probably because they were
-        // invalid.
-        logger.error({ err, projectId }, err.message)
-        render.unprocessableEntity(res)
-      } else if (err instanceof Chunk.NotFoundError) {
-        render.notFound(res)
-      } else {
-        next(err)
-      }
-    })
+  await loadFiles()
+
+  let result
+  try {
+    result = await persistChanges(projectId, changes, limits, endVersion)
+  } catch (err) {
+    if (
+      err instanceof Chunk.ConflictingEndVersion ||
+      err instanceof TextOperation.UnprocessableError ||
+      err instanceof File.NotEditableError ||
+      err instanceof FileMap.PathnameError ||
+      err instanceof Snapshot.EditMissingFileError ||
+      err instanceof chunkStore.ChunkVersionConflictError
+    ) {
+      // If we failed to apply operations, that's probably because they were
+      // invalid.
+      logger.warn({ err, projectId, endVersion }, 'changes rejected by history')
+      return render.unprocessableEntity(res)
+    } else if (err instanceof Chunk.NotFoundError) {
+      logger.warn({ err, projectId }, 'chunk not found')
+      return render.notFound(res)
+    } else {
+      throw err
+    }
+  }
+
+  if (returnSnapshot === 'none') {
+    res.status(HTTPStatus.CREATED).json({})
+  } else {
+    const rawSnapshot = await buildResultSnapshot(result && result.currentChunk)
+    res.status(HTTPStatus.CREATED).json(rawSnapshot)
+  }
 }
+
+exports.importSnapshot = expressify(importSnapshot)
+exports.importChanges = expressify(importChanges)

services/history-v1/storage/lib/chunk_store/index.js

@@ -153,7 +153,7 @@ async function loadAtTimestamp(projectId, timestamp) {
 /**
  * Store the chunk and insert corresponding records in the database.
  *
- * @param {number} projectId
+ * @param {string} projectId
  * @param {Chunk} chunk
  * @return {Promise.<number>} for the chunkId of the inserted chunk
  */
@@ -192,7 +192,7 @@ async function uploadChunk(projectId, chunk) {
  * Extend the project's history by replacing the latest chunk with a new
  * chunk.
  *
- * @param {number} projectId
+ * @param {string} projectId
  * @param {number} oldEndVersion
  * @param {Chunk} newChunk
  * @return {Promise}
@@ -212,7 +212,7 @@ async function update(projectId, oldEndVersion, newChunk) {
 /**
  * Find the chunk ID for a given version of a project.
  *
- * @param {number} projectId
+ * @param {string} projectId
  * @param {number} version
  * @return {Promise.<number>}
  */

services/history-v1/storage/lib/persist_changes.js

@@ -1,8 +1,8 @@
-/** @module */
+// @ts-check
+
 'use strict'
 
 const _ = require('lodash')
-const BPromise = require('bluebird')
 
 const core = require('overleaf-editor-core')
 const Chunk = core.Chunk
@@ -48,17 +48,12 @@ Timer.prototype.elapsed = function () {
  *    endVersion may be better suited to the metadata record.
  *
  * @param {string} projectId
- * @param {Array.<Change>} allChanges
+ * @param {core.Change[]} allChanges
  * @param {Object} limits
  * @param {number} clientEndVersion
  * @return {Promise.<Object?>}
  */
-module.exports = function persistChanges(
-  projectId,
-  allChanges,
-  limits,
-  clientEndVersion
-) {
+async function persistChanges(projectId, allChanges, limits, clientEndVersion) {
   assert.projectId(projectId)
   assert.array(allChanges)
   assert.maybe.object(limits)
@@ -104,41 +99,42 @@ module.exports = function persistChanges(
     return changesPushed
   }
 
-  function extendLastChunkIfPossible() {
-    return chunkStore.loadLatest(projectId).then(function (latestChunk) {
-      currentChunk = latestChunk
-      originalEndVersion = latestChunk.getEndVersion()
-      if (originalEndVersion !== clientEndVersion) {
-        throw new Chunk.ConflictingEndVersion(
-          clientEndVersion,
-          originalEndVersion
-        )
-      }
+  async function extendLastChunkIfPossible() {
+    const latestChunk = await chunkStore.loadLatest(projectId)
 
-      currentSnapshot = latestChunk.getSnapshot().clone()
-      const timer = new Timer()
-      currentSnapshot.applyAll(latestChunk.getChanges())
+    currentChunk = latestChunk
+    originalEndVersion = latestChunk.getEndVersion()
+    if (originalEndVersion !== clientEndVersion) {
+      throw new Chunk.ConflictingEndVersion(
+        clientEndVersion,
+        originalEndVersion
+      )
+    }
 
-      if (!fillChunk(currentChunk, changesToPersist)) return
-      checkElapsedTime(timer)
+    currentSnapshot = latestChunk.getSnapshot().clone()
+    const timer = new Timer()
+    currentSnapshot.applyAll(latestChunk.getChanges())
 
-      return chunkStore.update(projectId, originalEndVersion, currentChunk)
-    })
+    if (!fillChunk(currentChunk, changesToPersist)) return
+    checkElapsedTime(timer)
+
+    await chunkStore.update(projectId, originalEndVersion, currentChunk)
   }
 
-  function createNewChunksAsNeeded() {
-    if (changesToPersist.length === 0) return
-
-    const endVersion = currentChunk.getEndVersion()
-    const history = new History(currentSnapshot.clone(), [])
-    const chunk = new Chunk(history, endVersion)
-    const timer = new Timer()
-    if (fillChunk(chunk, changesToPersist)) {
-      checkElapsedTime(timer)
-      currentChunk = chunk
-      return chunkStore.create(projectId, chunk).then(createNewChunksAsNeeded)
+  async function createNewChunksAsNeeded() {
+    while (changesToPersist.length > 0) {
+      const endVersion = currentChunk.getEndVersion()
+      const history = new History(currentSnapshot.clone(), [])
+      const chunk = new Chunk(history, endVersion)
+      const timer = new Timer()
+      if (fillChunk(chunk, changesToPersist)) {
+        checkElapsedTime(timer)
+        currentChunk = chunk
+        await chunkStore.create(projectId, chunk)
+      } else {
+        throw new Error('failed to fill empty chunk')
+      }
     }
-    throw new Error('failed to fill empty chunk')
   }
 
   function isOlderThanMinChangeTimestamp(change) {
@@ -157,15 +153,18 @@ module.exports = function persistChanges(
   if (anyTooOld || tooManyChanges || tooManyBytes) {
     changesToPersist = oldChanges
     const numberOfChangesToPersist = oldChanges.length
-    return extendLastChunkIfPossible()
-      .then(createNewChunksAsNeeded)
-      .then(function () {
-        return {
-          numberOfChangesPersisted: numberOfChangesToPersist,
-          originalEndVersion,
-          currentChunk,
-        }
-      })
+
+    await extendLastChunkIfPossible()
+    await createNewChunksAsNeeded()
+
+    return {
+      numberOfChangesPersisted: numberOfChangesToPersist,
+      originalEndVersion,
+      currentChunk,
+    }
+  } else {
+    return null
   }
-  return BPromise.resolve(null)
 }
+
+module.exports = persistChanges