Merge pull request #28401 from overleaf/em-promisify-document-updater-handler

Promisify DocumentUpdaterHandler

GitOrigin-RevId: 8793f30ef51f5cd5886d0f74773f4b952761c2d4
This commit is contained in:
Eric Mc Sween
2025-09-15 08:05:25 +00:00
committed by Copybot
parent b87812d102
commit ee094a21eb
2 changed files with 793 additions and 1270 deletions
@@ -1,271 +1,210 @@
const request = require('request').defaults({ timeout: 30 * 1000 })
const OError = require('@overleaf/o-error') const OError = require('@overleaf/o-error')
const settings = require('@overleaf/settings') const settings = require('@overleaf/settings')
const {
fetchJson,
fetchNothing,
fetchString,
RequestFailedError,
} = require('@overleaf/fetch-utils')
const _ = require('lodash') const _ = require('lodash')
const async = require('async')
const logger = require('@overleaf/logger') const logger = require('@overleaf/logger')
const metrics = require('@overleaf/metrics') const { callbackifyAll } = require('@overleaf/promise-utils')
const { promisify, callbackify } = require('util')
const { promisifyMultiResult } = require('@overleaf/promise-utils')
const ProjectGetter = require('../Project/ProjectGetter') const ProjectGetter = require('../Project/ProjectGetter')
const Modules = require('../../infrastructure/Modules') const Modules = require('../../infrastructure/Modules')
function getProjectLastUpdatedAt(projectId, callback) { const REQUEST_TIMEOUT_MS = 30 * 1000
_makeRequest( const RESYNC_TIMEOUT_MS = 6 * 60 * 1000
{ const BASE_URL = settings.apis.documentupdater.url
path: `/project/${projectId}/last_updated_at`,
method: 'GET', async function getProjectLastUpdatedAt(projectId) {
json: true, const body = await fetchJson(
}, `${BASE_URL}/project/${projectId}/last_updated_at`,
projectId, { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }
'project.redis.last_updated_at',
(err, body) => {
if (err || !body?.lastUpdatedAt) return callback(err, null)
callback(null, new Date(body.lastUpdatedAt))
}
) )
return body.lastUpdatedAt != null ? new Date(body.lastUpdatedAt) : null
} }
/** /**
* @param {string} projectId * @param {string} projectId
*/ */
function flushProjectToMongo(projectId, callback) { async function flushProjectToMongo(projectId) {
_makeRequest( await fetchNothing(`${BASE_URL}/project/${projectId}/flush`, {
{ method: 'POST',
path: `/project/${projectId}/flush`, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
method: 'POST',
},
projectId,
'flushing.mongo.project',
callback
)
}
function flushMultipleProjectsToMongo(projectIds, callback) {
const jobs = projectIds.map(projectId => callback => {
flushProjectToMongo(projectId, callback)
}) })
async.series(jobs, callback) }
async function flushMultipleProjectsToMongo(projectIds) {
for (const projectId of projectIds) {
await flushProjectToMongo(projectId)
}
} }
/** /**
* @param {string} projectId * @param {string} projectId
*/ */
function flushProjectToMongoAndDelete(projectId, callback) { async function flushProjectToMongoAndDelete(projectId) {
_makeRequest( await fetchNothing(`${BASE_URL}/project/${projectId}`, {
{ method: 'DELETE',
path: `/project/${projectId}`, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
method: 'DELETE', })
},
projectId,
'flushing.mongo.project',
callback
)
} }
/** /**
* @param {string} projectId * @param {string} projectId
* @param {string} docId * @param {string} docId
* @param {Callback} callback
*/ */
function flushDocToMongo(projectId, docId, callback) { async function flushDocToMongo(projectId, docId) {
_makeRequest( await fetchNothing(`${BASE_URL}/project/${projectId}/doc/${docId}/flush`, {
{ method: 'POST',
path: `/project/${projectId}/doc/${docId}/flush`, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
method: 'POST', })
},
projectId,
'flushing.mongo.doc',
callback
)
} }
function deleteDoc(projectId, docId, ignoreFlushErrors, callback) { async function deleteDoc(projectId, docId, ignoreFlushErrors = false) {
if (typeof ignoreFlushErrors === 'function') { const url = new URL(`${BASE_URL}/project/${projectId}/doc/${docId}`)
callback = ignoreFlushErrors
ignoreFlushErrors = false
}
let path = `/project/${projectId}/doc/${docId}`
if (ignoreFlushErrors) { if (ignoreFlushErrors) {
path += '?ignore_flush_errors=true' url.searchParams.set('ignore_flush_errors', 'true')
} }
const method = 'DELETE' await fetchNothing(url, {
_makeRequest( method: 'DELETE',
{ signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
path, })
method,
},
projectId,
'delete.mongo.doc',
callback
)
} }
function getComment(projectId, docId, commentId, callback) { async function getComment(projectId, docId, commentId) {
_makeRequest( const comment = await fetchJson(
{ `${BASE_URL}/project/${projectId}/doc/${docId}/comment/${commentId}`,
path: `/project/${projectId}/doc/${docId}/comment/${commentId}`, { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }
json: true,
},
projectId,
'get-comment',
function (error, comment) {
if (error) {
return callback(error)
}
callback(null, comment)
}
) )
return comment
} }
function getDocument(projectId, docId, fromVersion, callback) { async function getDocument(projectId, docId, fromVersion) {
_makeRequest( const url = new URL(`${BASE_URL}/project/${projectId}/doc/${docId}`)
{ url.searchParams.set('fromVersion', fromVersion)
path: `/project/${projectId}/doc/${docId}?fromVersion=${fromVersion}`, const doc = await fetchJson(url, {
json: true, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
}, })
projectId, return {
'get-document', lines: doc.lines,
function (error, doc) { version: doc.version,
if (error) { ranges: doc.ranges,
return callback(error) ops: doc.ops,
} }
callback(null, doc.lines, doc.version, doc.ranges, doc.ops)
}
)
} }
/** /**
* Get a document with its history ranges * Get a document with its history ranges
* @param {string} projectId * @param {string} projectId
* @param {string} docId * @param {string} docId
* @param {Callback} callback
*/ */
function getDocumentWithHistoryRanges(projectId, docId, callback) { async function getDocumentWithHistoryRanges(projectId, docId) {
_makeRequest( const doc = await fetchJson(
{ `${BASE_URL}/project/${projectId}/doc/${docId}?historyRanges=true`,
path: `/project/${projectId}/doc/${docId}?historyRanges=true`, { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }
json: true,
},
projectId,
'get-document-with-history-ranges',
function (error, doc) {
if (error) {
return callback(error)
}
callback(null, doc)
}
) )
return doc
} }
function setDocument(projectId, docId, userId, docLines, source, callback) { async function setDocument(projectId, docId, userId, docLines, source) {
_makeRequest( const maybeJson = await fetchString(
`${BASE_URL}/project/${projectId}/doc/${docId}`,
{ {
path: `/project/${projectId}/doc/${docId}`,
method: 'POST', method: 'POST',
json: { json: {
lines: docLines, lines: docLines,
source, source,
user_id: userId, user_id: userId,
}, },
}, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
projectId, }
'set-document',
callback
) )
// The set document endpoint sometimes returns json and sometimes just an
// empty response
try {
const { rev, modified } = JSON.parse(maybeJson)
return { rev, modified }
} catch (err) {
return undefined
}
} }
function appendToDocument(projectId, docId, userId, lines, source, callback) { async function appendToDocument(projectId, docId, userId, lines, source) {
_makeRequest( const maybeJson = await fetchString(
`${BASE_URL}/project/${projectId}/doc/${docId}/append`,
{ {
path: `/project/${projectId}/doc/${docId}/append`,
method: 'POST', method: 'POST',
json: { json: {
lines, lines,
source, source,
user_id: userId, user_id: userId,
}, },
}, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
projectId, }
'append-to-document',
callback
) )
// The append to document endpoint sometimes returns json and sometimes just an
// empty response
try {
const { rev, modified } = JSON.parse(maybeJson)
return { rev, modified }
} catch (err) {
return undefined
}
} }
function getProjectDocsIfMatch(projectId, projectStateHash, callback) { async function getProjectDocsIfMatch(projectId, projectStateHash) {
// If the project state hasn't changed, we can get all the latest // If the project state hasn't changed, we can get all the latest
// docs from redis via the docupdater. Otherwise we will need to // docs from redis via the docupdater. Otherwise we will need to
// fall back to getting them from mongo. // fall back to getting them from mongo.
const timer = new metrics.Timer('get-project-docs') const url = new URL(`${BASE_URL}/project/${projectId}/get_and_flush_if_old`)
const url = `${settings.apis.documentupdater.url}/project/${projectId}/get_and_flush_if_old?state=${projectStateHash}` url.searchParams.set('state', projectStateHash)
request.post(url, function (error, res, body) {
timer.done() let docs
if (error) { try {
OError.tag(error, 'error getting project docs from doc updater', { docs = await fetchJson(url, {
url, method: 'POST',
projectId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
}) })
return callback(error) } catch (err) {
} if (err instanceof RequestFailedError && err.response.status === 409) {
if (res.statusCode === 409) {
// HTTP response code "409 Conflict" // HTTP response code "409 Conflict"
// Docupdater has checked the projectStateHash and found that // Docupdater has checked the projectStateHash and found that
// it has changed. This means that the docs currently in redis // it has changed. This means that the docs currently in redis
// aren't the only change to the project and the full set of // aren't the only change to the project and the full set of
// docs/files should be retreived from docstore/filestore // docs/files should be retreived from docstore/filestore
// instead. // instead.
callback() return undefined
} else if (res.statusCode >= 200 && res.statusCode < 300) {
let docs
try {
docs = JSON.parse(body)
} catch (error1) {
return callback(OError.tag(error1))
}
callback(null, docs)
} else { } else {
callback( throw err
new OError(
`doc updater returned a non-success status code: ${res.statusCode}`,
{
projectId,
url,
}
)
)
} }
}) }
return docs
} }
function clearProjectState(projectId, callback) { async function clearProjectState(projectId) {
_makeRequest( await fetchNothing(`${BASE_URL}/project/${projectId}/clearState`, {
{ method: 'POST',
path: `/project/${projectId}/clearState`, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
method: 'POST', })
},
projectId,
'clear-project-state',
callback
)
} }
/** /**
* @param {string} projectId * @param {string} projectId
* @param {string} docId * @param {string} docId
* @param {string[]} changeIds * @param {string[]} changeIds
* @param {Callback} callback
*/ */
async function acceptChanges(projectId, docId, changeIds) { async function acceptChanges(projectId, docId, changeIds) {
await _makeRequestAsync( await fetchNothing(
`${BASE_URL}/project/${projectId}/doc/${docId}/change/accept`,
{ {
path: `/project/${projectId}/doc/${docId}/change/accept`,
json: { change_ids: changeIds },
method: 'POST', method: 'POST',
}, json: { change_ids: changeIds },
projectId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
'accept-changes' }
) )
await Modules.promises.hooks.fire('changesAccepted', projectId, docId) await Modules.promises.hooks.fire('changesAccepted', projectId, docId)
} }
@@ -273,18 +212,33 @@ async function acceptChanges(projectId, docId, changeIds) {
* @param {string} projectId * @param {string} projectId
* @param {string} docId * @param {string} docId
* @param {string[]} changeIds * @param {string[]} changeIds
* @param {Callback} callback
*/ */
function rejectChanges(projectId, docId, changeIds, userId, callback) { async function rejectChanges(projectId, docId, changeIds, userId) {
_makeRequest( const { rejectedChangeIds } = await fetchJson(
`${BASE_URL}/project/${projectId}/doc/${docId}/change/reject`,
{ {
path: `/project/${projectId}/doc/${docId}/change/reject`, method: 'POST',
json: { change_ids: changeIds, user_id: userId }, json: { change_ids: changeIds, user_id: userId },
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
}
)
return { rejectedChangeIds }
}
/**
* @param {string} projectId
* @param {string} docId
* @param {string} threadId
* @param {string} userId
*/
async function resolveThread(projectId, docId, threadId, userId) {
await fetchNothing(
`${BASE_URL}/project/${projectId}/doc/${docId}/comment/${threadId}/resolve`,
{
method: 'POST', method: 'POST',
}, json: { user_id: userId },
projectId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
'reject-changes', }
callback
) )
} }
@@ -293,67 +247,35 @@ function rejectChanges(projectId, docId, changeIds, userId, callback) {
* @param {string} docId * @param {string} docId
* @param {string} threadId * @param {string} threadId
* @param {string} userId * @param {string} userId
* @param {Callback} callback
*/ */
function resolveThread(projectId, docId, threadId, userId, callback) { async function reopenThread(projectId, docId, threadId, userId) {
_makeRequest( await fetchNothing(
`${BASE_URL}/project/${projectId}/doc/${docId}/comment/${threadId}/reopen`,
{ {
path: `/project/${projectId}/doc/${docId}/comment/${threadId}/resolve`,
method: 'POST', method: 'POST',
json: { json: { user_id: userId },
user_id: userId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
}, }
},
projectId,
'resolve-thread',
callback
) )
} }
/** async function deleteThread(projectId, docId, threadId, userId) {
* @param {string} projectId await fetchNothing(
* @param {string} docId `${BASE_URL}/project/${projectId}/doc/${docId}/comment/${threadId}`,
* @param {string} threadId
* @param {string} userId
* @param {Callback} callback
*/
function reopenThread(projectId, docId, threadId, userId, callback) {
_makeRequest(
{ {
path: `/project/${projectId}/doc/${docId}/comment/${threadId}/reopen`,
method: 'POST',
json: {
user_id: userId,
},
},
projectId,
'reopen-thread',
callback
)
}
function deleteThread(projectId, docId, threadId, userId, callback) {
_makeRequest(
{
path: `/project/${projectId}/doc/${docId}/comment/${threadId}`,
method: 'DELETE', method: 'DELETE',
json: { json: { user_id: userId },
user_id: userId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
}, }
},
projectId,
'delete-thread',
callback
) )
} }
function resyncProjectHistory( async function resyncProjectHistory(
projectId, projectId,
projectHistoryId, projectHistoryId,
docs, docs,
files, files,
opts, opts
callback
) { ) {
docs = docs.map(doc => ({ docs = docs.map(doc => ({
doc: doc.doc._id, doc: doc.doc._id,
@@ -362,9 +284,7 @@ function resyncProjectHistory(
// Files without a hash likely do not have a blob. Abort. // Files without a hash likely do not have a blob. Abort.
for (const { file } of files) { for (const { file } of files) {
if (!file.hash) { if (!file.hash) {
return callback( throw new OError('found file with missing hash', { projectId, file })
new OError('found file with missing hash', { projectId, file })
)
} }
} }
files = files.map(file => ({ files = files.map(file => ({
@@ -382,192 +302,119 @@ function resyncProjectHistory(
if (opts.resyncProjectStructureOnly) { if (opts.resyncProjectStructureOnly) {
body.resyncProjectStructureOnly = opts.resyncProjectStructureOnly body.resyncProjectStructureOnly = opts.resyncProjectStructureOnly
} }
_makeRequest( await fetchNothing(`${BASE_URL}/project/${projectId}/history/resync`, {
{ json: body,
path: `/project/${projectId}/history/resync`, method: 'POST',
json: body, signal: AbortSignal.timeout(RESYNC_TIMEOUT_MS), // allow 6 minutes for resync
method: 'POST', })
timeout: 6 * 60 * 1000, // allow 6 minutes for resync
},
projectId,
'resync-project-history',
callback
)
} }
/** /**
* Block a project from being loaded in docupdater * Block a project from being loaded in docupdater
* *
* @param {string} projectId * @param {string} projectId
* @param {Callback} callback
*/ */
function blockProject(projectId, callback) { async function blockProject(projectId) {
_makeRequest( const body = await fetchJson(`${BASE_URL}/project/${projectId}/block`, {
{ path: `/project/${projectId}/block`, method: 'POST', json: true }, method: 'POST',
projectId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
'block-project', })
(err, body) => { return body.blocked
if (err) {
return callback(err)
}
callback(null, body.blocked)
}
)
} }
/** /**
* Unblock a previously blocked project * Unblock a previously blocked project
* *
* @param {string} projectId * @param {string} projectId
* @param {Callback} callback
*/ */
function unblockProject(projectId, callback) { async function unblockProject(projectId) {
_makeRequest( const body = await fetchJson(`${BASE_URL}/project/${projectId}/unblock`, {
{ path: `/project/${projectId}/unblock`, method: 'POST', json: true }, method: 'POST',
projectId, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
'unblock-project', })
(err, body) => { return body.wasBlocked
if (err) {
return callback(err)
}
callback(null, body.wasBlocked)
}
)
} }
function updateProjectStructure( async function updateProjectStructure(
projectId, projectId,
projectHistoryId, projectHistoryId,
userId, userId,
changes, changes,
source, source
callback
) { ) {
if ( if (
settings.apis.project_history == null || settings.apis.project_history == null ||
!settings.apis.project_history.sendProjectStructureOps !settings.apis.project_history.sendProjectStructureOps
) { ) {
return callback() return
} }
ProjectGetter.getProjectWithoutLock( const project = await ProjectGetter.promises.getProjectWithoutLock(
projectId, projectId,
{ overleaf: true }, { overleaf: true }
(err, project) => {
if (err) {
return callback(err)
}
const historyRangesSupport = _.get(
project,
'overleaf.history.rangesSupportEnabled',
false
)
const {
deletes: docDeletes,
adds: docAdds,
renames: docRenames,
} = _getUpdates(
'doc',
changes.oldDocs,
changes.newDocs,
historyRangesSupport
)
for (const newEntity of changes.newFiles || []) {
if (!newEntity.file.hash) {
// Files without a hash likely do not have a blob. Abort.
return callback(
new OError('found file with missing hash', { newEntity })
)
}
}
const {
deletes: fileDeletes,
adds: fileAdds,
renames: fileRenames,
} = _getUpdates(
'file',
changes.oldFiles,
changes.newFiles,
historyRangesSupport
)
const updates = [].concat(
docDeletes,
fileDeletes,
docAdds,
fileAdds,
docRenames,
fileRenames
)
const projectVersion =
changes && changes.newProject && changes.newProject.version
if (updates.length < 1) {
return callback()
}
if (projectVersion == null) {
logger.warn(
{ projectId, changes, projectVersion },
'did not receive project version in changes'
)
return callback(new Error('did not receive project version in changes'))
}
_makeRequest(
{
path: `/project/${projectId}`,
json: {
updates,
userId,
version: projectVersion,
projectHistoryId,
source,
},
method: 'POST',
},
projectId,
'update-project-structure',
callback
)
}
) )
} const historyRangesSupport = _.get(
project,
'overleaf.history.rangesSupportEnabled',
false
)
const {
deletes: docDeletes,
adds: docAdds,
renames: docRenames,
} = _getUpdates('doc', changes.oldDocs, changes.newDocs, historyRangesSupport)
for (const newEntity of changes.newFiles || []) {
if (!newEntity.file.hash) {
// Files without a hash likely do not have a blob. Abort.
throw new OError('found file with missing hash', { newEntity })
}
}
const {
deletes: fileDeletes,
adds: fileAdds,
renames: fileRenames,
} = _getUpdates(
'file',
changes.oldFiles,
changes.newFiles,
historyRangesSupport
)
const updates = [].concat(
docDeletes,
fileDeletes,
docAdds,
fileAdds,
docRenames,
fileRenames
)
const projectVersion =
changes && changes.newProject && changes.newProject.version
function _makeRequest(options, projectId, metricsKey, callback) { if (updates.length < 1) {
const timer = new metrics.Timer(metricsKey) return
request( }
{
url: `${settings.apis.documentupdater.url}${options.path}`, if (projectVersion == null) {
json: options.json, logger.warn(
method: options.method || 'GET', { projectId, changes, projectVersion },
timeout: options.timeout || 30 * 1000, 'did not receive project version in changes'
)
throw new Error('did not receive project version in changes')
}
await fetchNothing(`${BASE_URL}/project/${projectId}`, {
method: 'POST',
json: {
updates,
userId,
version: projectVersion,
projectHistoryId,
source,
}, },
function (error, res, body) { signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
timer.done() })
if (error) {
logger.warn(
{ error, projectId },
'error making request to document updater'
)
callback(error)
} else if (res.statusCode >= 200 && res.statusCode < 300) {
callback(null, body)
} else {
error = new Error(
`document updater returned a failure status code: ${res.statusCode}`
)
logger.warn(
{ error, projectId },
`document updater returned failure status code: ${res.statusCode}`
)
callback(error)
}
}
)
} }
const _makeRequestAsync = promisify(_makeRequest)
function _getUpdates( function _getUpdates(
entityType, entityType,
oldEntities, oldEntities,
@@ -662,7 +509,7 @@ function buildFileMetadataForHistory(file) {
return metadata return metadata
} }
module.exports = { const DocumentUpdaterHandler = {
flushProjectToMongo, flushProjectToMongo,
flushMultipleProjectsToMongo, flushMultipleProjectsToMongo,
flushProjectToMongoAndDelete, flushProjectToMongoAndDelete,
@@ -675,7 +522,7 @@ module.exports = {
appendToDocument, appendToDocument,
getProjectDocsIfMatch, getProjectDocsIfMatch,
clearProjectState, clearProjectState,
acceptChanges: callbackify(acceptChanges), acceptChanges,
rejectChanges, rejectChanges,
resolveThread, resolveThread,
reopenThread, reopenThread,
@@ -685,33 +532,13 @@ module.exports = {
unblockProject, unblockProject,
updateProjectStructure, updateProjectStructure,
getDocumentWithHistoryRanges, getDocumentWithHistoryRanges,
promises: { }
flushProjectToMongo: promisify(flushProjectToMongo),
flushMultipleProjectsToMongo: promisify(flushMultipleProjectsToMongo), module.exports = {
flushProjectToMongoAndDelete: promisify(flushProjectToMongoAndDelete), ...callbackifyAll(DocumentUpdaterHandler, {
flushDocToMongo: promisify(flushDocToMongo), multiResult: {
deleteDoc: promisify(deleteDoc), getDocument: ['lines', 'version', 'ranges', 'ops'],
getComment: promisify(getComment), },
getDocument: promisifyMultiResult(getDocument, [ }),
'lines', promises: DocumentUpdaterHandler,
'version',
'ranges',
'ops',
]),
setDocument: promisify(setDocument),
getProjectDocsIfMatch: promisify(getProjectDocsIfMatch),
getProjectLastUpdatedAt: promisify(getProjectLastUpdatedAt),
clearProjectState: promisify(clearProjectState),
acceptChanges,
rejectChanges: promisify(rejectChanges),
resolveThread: promisify(resolveThread),
reopenThread: promisify(reopenThread),
deleteThread: promisify(deleteThread),
resyncProjectHistory: promisify(resyncProjectHistory),
blockProject: promisify(blockProject),
unblockProject: promisify(unblockProject),
updateProjectStructure: promisify(updateProjectStructure),
appendToDocument: promisify(appendToDocument),
getDocumentWithHistoryRanges: promisify(getDocumentWithHistoryRanges),
},
} }
File diff suppressed because it is too large Load Diff