From a7c9e3b20fe275f96d4319966638e283855d105b Mon Sep 17 00:00:00 2001 From: Eric Mc Sween Date: Thu, 16 Feb 2023 07:47:22 -0500 Subject: [PATCH] Merge pull request #11790 from overleaf/em-promisify-redis-manager Clean up and promisify RedisManager in project-history GitOrigin-RevId: 8bd8bb7d51a0a68f7b1a97ffa310a674086714ba --- .../project-history/app/js/RedisManager.js | 649 +++++++-------- .../app/js/UpdatesProcessor.js | 59 +- services/project-history/test/setup.js | 2 + .../unit/js/RedisManager/RedisManagerTests.js | 787 ++++++------------ 4 files changed, 566 insertions(+), 931 deletions(-) diff --git a/services/project-history/app/js/RedisManager.js b/services/project-history/app/js/RedisManager.js index 97c1c31ea2..2f3b55263e 100644 --- a/services/project-history/app/js/RedisManager.js +++ b/services/project-history/app/js/RedisManager.js @@ -1,20 +1,7 @@ -/* eslint-disable - camelcase, -*/ -// TODO: This file was created by bulk-decaffeinate. -// Fix any style issues and re-enable lint. -/* - * decaffeinate suggestions: - * DS101: Remove unnecessary use of Array.from - * DS102: Remove unnecessary code created because of implicit returns - * DS205: Consider reworking code to avoid use of IIFEs - * DS207: Consider shorter variations of null checks - * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md - */ -import { promisify } from 'util' +import { callbackify, promisify } from 'node:util' +import { setTimeout } from 'node:timers/promises' import logger from '@overleaf/logger' import Settings from '@overleaf/settings' -import async from 'async' import redis from '@overleaf/redis-wrapper' import metrics from '@overleaf/metrics' import OError from '@overleaf/o-error' @@ -39,190 +26,137 @@ const CACHE_TTL_IN_SECONDS = 3600 const Keys = Settings.redis.project_history.key_schema const rclient = redis.createClient(Settings.redis.project_history) -/** - * Container for functions that need to be mocked in tests - * - * TODO: Rewrite tests in terms of exported functions only - */ -export const _mocks = {} - -export function countUnprocessedUpdates(project_id, callback) { - const key = Keys.projectHistoryOps({ project_id }) - return rclient.llen(key, callback) +async function countUnprocessedUpdates(projectId) { + const key = Keys.projectHistoryOps({ project_id: projectId }) + const updates = await rclient.llen(key) + return updates } -_mocks.getOldestDocUpdates = (project_id, batch_size, callback) => { - if (callback == null) { - callback = function () {} - } - const key = Keys.projectHistoryOps({ project_id }) - rclient.lrange(key, 0, batch_size - 1, callback) +async function getOldestDocUpdates(projectId, batchSize) { + const key = Keys.projectHistoryOps({ project_id: projectId }) + const updates = await rclient.lrange(key, 0, batchSize - 1) + return updates } -export function getOldestDocUpdates(...args) { - _mocks.getOldestDocUpdates(...args) +export function parseDocUpdates(jsonUpdates) { + return jsonUpdates.map(update => JSON.parse(update)) } -_mocks.parseDocUpdates = (json_updates, callback) => { - let parsed_updates - if (callback == null) { - callback = function () {} - } - try { - parsed_updates = Array.from(json_updates || []).map(update => - JSON.parse(update) - ) - } catch (e) { - return callback(e) - } - callback(null, parsed_updates) -} - -export function parseDocUpdates(...args) { - _mocks.parseDocUpdates(...args) -} - -export function getUpdatesInBatches(project_id, batch_size, runner, callback) { +async function getUpdatesInBatches(projectId, batchSize, runner) { let moreBatches = true - let lastResults = [] - const processBatch = cb => - getOldestDocUpdates(project_id, batch_size, function (error, raw_updates) { - let raw_update - if (error != null) { - return cb(OError.tag(error)) - } - moreBatches = raw_updates.length === batch_size - if (raw_updates.length === 0) { - return cb() - } - // don't process any more batches if we are single stepping - if (batch_size === 1) { - moreBatches = false + while (moreBatches) { + let rawUpdates = await getOldestDocUpdates(projectId, batchSize) + + moreBatches = rawUpdates.length === batchSize + + if (rawUpdates.length === 0) { + return + } + + // don't process any more batches if we are single stepping + if (batchSize === 1) { + moreBatches = false + } + + // consume the updates up to a maximum total number of bytes + // ensuring that at least one update will be processed (we may + // exceed RAW_UPDATE_SIZE_THRESHOLD is the first update is bigger + // than that). + let totalRawUpdatesSize = 0 + const updatesToProcess = [] + for (const rawUpdate of rawUpdates) { + const nextTotalSize = totalRawUpdatesSize + rawUpdate.length + if ( + updatesToProcess.length > 0 && + nextTotalSize > RAW_UPDATE_SIZE_THRESHOLD + ) { + // stop consuming updates if we have at least one and the + // next update would exceed the size threshold + break + } else { + updatesToProcess.push(rawUpdate) + totalRawUpdatesSize += rawUpdate.length } + } - // consume the updates up to a maximum total number of bytes - // ensuring that at least one update will be processed (we may - // exceed RAW_UPDATE_SIZE_THRESHOLD is the first update is bigger - // than that). - let total_raw_updates_size = 0 - const updates_to_process = [] - for (raw_update of Array.from(raw_updates)) { - const next_total_size = total_raw_updates_size + raw_update.length - if ( - updates_to_process.length > 0 && - next_total_size > RAW_UPDATE_SIZE_THRESHOLD - ) { - // stop consuming updates if we have at least one and the - // next update would exceed the size threshold - break - } else { - updates_to_process.push(raw_update) - total_raw_updates_size += raw_update.length - } - } + // if we hit the size limit above, only process the updates up to that point + if (updatesToProcess.length < rawUpdates.length) { + moreBatches = true // process remaining raw updates in the next iteration + rawUpdates = updatesToProcess + } - // if we hit the size limit above, only process the updates up to that point - if (updates_to_process.length < raw_updates.length) { - moreBatches = true // process remaining raw updates in the next iteration - raw_updates = updates_to_process - } + metrics.timing('redis.incoming.bytes', totalRawUpdatesSize, 1) + if (totalRawUpdatesSize > WARN_RAW_UPDATE_SIZE) { + const rawUpdateSizes = rawUpdates.map(rawUpdate => rawUpdate.length) + logger.warn( + { projectId, totalRawUpdatesSize, rawUpdateSizes }, + 'large raw update size' + ) + } - metrics.timing('redis.incoming.bytes', total_raw_updates_size, 1) - if (total_raw_updates_size > WARN_RAW_UPDATE_SIZE) { - const raw_update_sizes = (() => { - const result = [] - for (raw_update of Array.from(raw_updates)) { - result.push(raw_update.length) - } - return result - })() - logger.warn( - { project_id, total_raw_updates_size, raw_update_sizes }, - 'large raw update size' - ) - } - - return parseDocUpdates(raw_updates, function (error, updates) { - if (error != null) { - OError.tag(error, 'failed to parse updates', { - project_id, - updates, - }) - return cb(error) - } - - // consume the updates up to a maximum number of ops (insertions and deletions) - let total_op_length = 0 - let updates_to_process_count = 0 - let total_doc_content_count = 0 - for (const parsed_update of Array.from(updates)) { - if (parsed_update.resyncDocContent) { - total_doc_content_count++ - } - if (total_doc_content_count > MAX_NEW_DOC_CONTENT_COUNT) { - break - } - const next_total_op_length = - total_op_length + (parsed_update?.op?.length || 1) - if ( - updates_to_process_count > 0 && - next_total_op_length > MAX_UPDATE_OP_LENGTH - ) { - break - } else { - total_op_length = next_total_op_length - updates_to_process_count++ - } - } - - // if we hit the op limit above, only process the updates up to that point - if (updates_to_process_count < updates.length) { - logger.debug( - { - project_id, - updates_to_process_count, - updates_count: updates.length, - total_op_length, - }, - 'restricting number of ops to be processed' - ) - moreBatches = true - // there is a 1:1 mapping between raw_updates and updates - // which we need to preserve here to ensure we only - // delete the updates that are actually processed - raw_updates = raw_updates.slice(0, updates_to_process_count) - updates = updates.slice(0, updates_to_process_count) - } - - logger.debug({ project_id }, 'retrieved raw updates from redis') - return runner(updates, function (error, ...args) { - lastResults = args - if (error != null) { - return cb(OError.tag(error)) - } - return deleteAppliedDocUpdates(project_id, raw_updates, cb) - }) + let updates + try { + updates = parseDocUpdates(rawUpdates) + } catch (error) { + throw OError.tag(error, 'failed to parse updates', { + projectId, + updates, }) - }) + } - const hasMoreBatches = (...args) => { - const cb = args[args.length - 1] - return cb(null, moreBatches) + // consume the updates up to a maximum number of ops (insertions and deletions) + let totalOpLength = 0 + let updatesToProcessCount = 0 + let totalDocContentCount = 0 + for (const parsedUpdate of updates) { + if (parsedUpdate.resyncDocContent) { + totalDocContentCount++ + } + if (totalDocContentCount > MAX_NEW_DOC_CONTENT_COUNT) { + break + } + const nextTotalOpLength = totalOpLength + (parsedUpdate?.op?.length || 1) + if ( + updatesToProcessCount > 0 && + nextTotalOpLength > MAX_UPDATE_OP_LENGTH + ) { + break + } else { + totalOpLength = nextTotalOpLength + updatesToProcessCount++ + } + } + + // if we hit the op limit above, only process the updates up to that point + if (updatesToProcessCount < updates.length) { + logger.debug( + { + projectId, + updatesToProcessCount, + updates_count: updates.length, + totalOpLength, + }, + 'restricting number of ops to be processed' + ) + moreBatches = true + // there is a 1:1 mapping between rawUpdates and updates + // which we need to preserve here to ensure we only + // delete the updates that are actually processed + rawUpdates = rawUpdates.slice(0, updatesToProcessCount) + updates = updates.slice(0, updatesToProcessCount) + } + + logger.debug({ projectId }, 'retrieved raw updates from redis') + await runner(updates) + await deleteAppliedDocUpdates(projectId, rawUpdates) } - - return async.doWhilst(processBatch, hasMoreBatches, error => - callback(error, ...Array.from(lastResults)) - ) } -_mocks.deleteAppliedDocUpdates = (project_id, updates, callback) => { - if (callback == null) { - callback = function () {} - } +async function deleteAppliedDocUpdates(projectId, updates) { const multi = rclient.multi() // Delete all the updates which have been applied (exact match) - for (const update of Array.from(updates || [])) { + for (const update of updates) { // Delete the first occurrence of the update with LREM KEY COUNT // VALUE by setting COUNT to 1 which 'removes COUNT elements equal to // value moving from head to tail.' @@ -233,209 +167,161 @@ _mocks.deleteAppliedDocUpdates = (project_id, updates, callback) => { metrics.summary('redis.projectHistoryOps', update.length, { status: 'lrem', }) - multi.lrem(Keys.projectHistoryOps({ project_id }), 1, update) - multi.del(Keys.projectHistoryFirstOpTimestamp({ project_id })) + multi.lrem(Keys.projectHistoryOps({ project_id: projectId }), 1, update) } - multi.exec(callback) + if (updates.length > 0) { + multi.del(Keys.projectHistoryFirstOpTimestamp({ project_id: projectId })) + } + await multi.exec() } -export function deleteAppliedDocUpdates(...args) { - _mocks.deleteAppliedDocUpdates(...args) -} - -export function destroyDocUpdatesQueue(project_id, callback) { - // deletes the entire queue - use with caution - if (callback == null) { - callback = function () {} - } - return rclient.del( - Keys.projectHistoryOps({ project_id }), - Keys.projectHistoryFirstOpTimestamp({ project_id }), - callback +/** + * Deletes the entire queue - use with caution + */ +async function destroyDocUpdatesQueue(projectId) { + await rclient.del( + Keys.projectHistoryOps({ project_id: projectId }), + Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }) ) } -// iterate over keys asynchronously using redis scan (non-blocking) -// handle all the cluster nodes or single redis server -function _getKeys(pattern, limit, callback) { - const nodes = (typeof rclient.nodes === 'function' - ? rclient.nodes('master') - : undefined) || [rclient] - const doKeyLookupForNode = (node, cb) => - _getKeysFromNode(node, pattern, limit, cb) - return async.concatSeries(nodes, doKeyLookupForNode, callback) +/** + * Iterate over keys asynchronously using redis scan (non-blocking) + * + * handle all the cluster nodes or single redis server + */ +async function _getKeys(pattern, limit) { + const nodes = rclient.nodes?.('master') || [rclient] + const keysByNode = [] + for (const node of nodes) { + const keys = await _getKeysFromNode(node, pattern, limit) + keysByNode.push(keys) + } + return [].concat(...keysByNode) } -function _getKeysFromNode(node, pattern, limit, callback) { +async function _getKeysFromNode(node, pattern, limit) { let cursor = 0 // redis iterator - const keySet = {} // use hash to avoid duplicate results + const keySet = new Set() // avoid duplicate results const batchSize = limit != null ? Math.min(limit, 1000) : 1000 + // scan over all keys looking for pattern - const doIteration = ( - cb // avoid hitting redis too hard - ) => - node.scan( - cursor, - 'MATCH', - pattern, - 'COUNT', - batchSize, - function (error, reply) { - let keys - if (error != null) { - return callback(OError.tag(error)) - } - ;[cursor, keys] = Array.from(reply) - for (const key of Array.from(keys)) { - keySet[key] = true - } - keys = Object.keys(keySet) - const noResults = cursor === '0' // redis returns string results not numeric - const limitReached = limit != null && keys.length >= limit - if (noResults || limitReached) { - return callback(null, keys) - } else { - return setTimeout(doIteration, 10) - } - } - ) - return doIteration() + while (true) { + const reply = await node.scan(cursor, 'MATCH', pattern, 'COUNT', batchSize) + const [newCursor, keys] = reply + cursor = newCursor + + for (const key of keys) { + keySet.add(key) + } + + const noResults = cursor === '0' // redis returns string results not numeric + const limitReached = limit != null && keySet.size >= limit + if (noResults || limitReached) { + return Array.from(keySet) + } + + // avoid hitting redis too hard + await setTimeout(10) + } } -// extract ids from keys like DocsWithHistoryOps:57fd0b1f53a8396d22b2c24b -// or DocsWithHistoryOps:{57fd0b1f53a8396d22b2c24b} (for redis cluster) +/** + * Extract ids from keys like DocsWithHistoryOps:57fd0b1f53a8396d22b2c24b + * or DocsWithHistoryOps:{57fd0b1f53a8396d22b2c24b} (for redis cluster) + */ function _extractIds(keyList) { - const ids = (() => { - const result = [] - for (const key of Array.from(keyList)) { - const m = key.match(/:\{?([0-9a-f]{24})\}?/) // extract object id - result.push(m[1]) - } - return result - })() - return ids + return keyList.map(key => { + const m = key.match(/:\{?([0-9a-f]{24})\}?/) // extract object id + return m[1] + }) } -export function getProjectIdsWithHistoryOps(limit, callback) { - if (callback == null) { - callback = function () {} - } - return _getKeys( +async function getProjectIdsWithHistoryOps(limit) { + const projectKeys = await _getKeys( Keys.projectHistoryOps({ project_id: '*' }), - limit, - function (error, project_keys) { - if (error != null) { - return callback(OError.tag(error)) - } - const project_ids = _extractIds(project_keys) - return callback(error, project_ids) - } + limit ) + const projectIds = _extractIds(projectKeys) + return projectIds } -export function getProjectIdsWithHistoryOpsCount(callback) { - if (callback == null) { - callback = function () {} - } - return getProjectIdsWithHistoryOps(null, function (error, projectIds) { - if (error != null) { - return callback(OError.tag(error)) - } - const queuedProjectsCount = projectIds.length - metrics.globalGauge('queued-projects', queuedProjectsCount) - return callback(null, queuedProjectsCount) - }) +async function getProjectIdsWithHistoryOpsCount() { + const projectIds = await getProjectIdsWithHistoryOps() + const queuedProjectsCount = projectIds.length + metrics.globalGauge('queued-projects', queuedProjectsCount) + return queuedProjectsCount } -export function setFirstOpTimestamp(project_id, callback) { - if (callback == null) { - callback = function () {} - } - const key = Keys.projectHistoryFirstOpTimestamp({ project_id }) +async function setFirstOpTimestamp(projectId) { + const key = Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }) // store current time as an integer (string) - return rclient.setnx(key, Date.now(), callback) + await rclient.setnx(key, Date.now()) } -export function getFirstOpTimestamp(project_id, callback) { - if (callback == null) { - callback = function () {} +async function getFirstOpTimestamp(projectId) { + const key = Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }) + const result = await rclient.get(key) + + // convert stored time back to a numeric timestamp + const timestamp = parseInt(result, 10) + + // check for invalid timestamp + if (isNaN(timestamp)) { + return null } - const key = Keys.projectHistoryFirstOpTimestamp({ project_id }) - return rclient.get(key, function (err, result) { - if (err != null) { - return callback(OError.tag(err)) - } - // convert stored time back to a numeric timestamp - const timestamp = parseInt(result, 10) - // check for invalid timestamp - if (isNaN(timestamp)) { - return callback() - } - // convert numeric timestamp to a date object - const firstOpTimestamp = new Date(timestamp) - return callback(null, firstOpTimestamp) - }) + + // convert numeric timestamp to a date object + const firstOpTimestamp = new Date(timestamp) + + return firstOpTimestamp } -export function clearFirstOpTimestamp(project_id, callback) { - if (callback == null) { - callback = function () {} - } - const key = Keys.projectHistoryFirstOpTimestamp({ project_id }) - return rclient.del(key, callback) +async function clearFirstOpTimestamp(projectId) { + const key = Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }) + await rclient.del(key) } -export function getProjectIdsWithFirstOpTimestamps(limit, callback) { - return _getKeys( +async function getProjectIdsWithFirstOpTimestamps(limit) { + const projectKeys = await _getKeys( Keys.projectHistoryFirstOpTimestamp({ project_id: '*' }), - limit, - function (error, project_keys) { - if (error != null) { - return callback(OError.tag(error)) - } - const project_ids = _extractIds(project_keys) - return callback(error, project_ids) - } + limit ) + const projectIds = _extractIds(projectKeys) + return projectIds } -export function clearDanglingFirstOpTimestamp(project_id, callback) { - rclient.exists( - Keys.projectHistoryFirstOpTimestamp({ project_id }), - Keys.projectHistoryOps({ project_id }), - function (error, count) { - if (error) { - return callback(error) - } - if (count === 2 || count === 0) { - // both (or neither) keys are present, so don't delete the timestamp - return callback(null, 0) - } - // only one key is present, which makes this a dangling record, - // so delete the timestamp - rclient.del(Keys.projectHistoryFirstOpTimestamp({ project_id }), callback) - } +async function clearDanglingFirstOpTimestamp(projectId) { + const count = await rclient.exists( + Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }), + Keys.projectHistoryOps({ project_id: projectId }) ) + if (count === 2 || count === 0) { + // both (or neither) keys are present, so don't delete the timestamp + return 0 + } + // only one key is present, which makes this a dangling record, + // so delete the timestamp + const cleared = await rclient.del( + Keys.projectHistoryFirstOpTimestamp({ project_id: projectId }) + ) + return cleared } -export function getCachedHistoryId(project_id, callback) { - const key = Keys.projectHistoryCachedHistoryId({ project_id }) - rclient.get(key, function (err, historyId) { - if (err) { - return callback(OError.tag(err)) - } - callback(null, historyId) - }) +async function getCachedHistoryId(projectId) { + const key = Keys.projectHistoryCachedHistoryId({ project_id: projectId }) + const historyId = await rclient.get(key) + return historyId } -export function setCachedHistoryId(project_id, historyId, callback) { - const key = Keys.projectHistoryCachedHistoryId({ project_id }) - rclient.setex(key, CACHE_TTL_IN_SECONDS, historyId, callback) +async function setCachedHistoryId(projectId, historyId) { + const key = Keys.projectHistoryCachedHistoryId({ project_id: projectId }) + await rclient.setex(key, CACHE_TTL_IN_SECONDS, historyId) } -export function clearCachedHistoryId(project_id, callback) { - const key = Keys.projectHistoryCachedHistoryId({ project_id }) - rclient.del(key, callback) +async function clearCachedHistoryId(projectId) { + const key = Keys.projectHistoryCachedHistoryId({ project_id: projectId }) + await rclient.del(key) } // for tests @@ -451,10 +337,77 @@ export function setMaxNewDocContentCount(value) { MAX_NEW_DOC_CONTENT_COUNT = value } -export const promises = { - countUnprocessedUpdates: promisify(countUnprocessedUpdates), - getProjectIdsWithFirstOpTimestamps: promisify( - getProjectIdsWithFirstOpTimestamps - ), - clearDanglingFirstOpTimestamp: promisify(clearDanglingFirstOpTimestamp), +// EXPORTS + +const countUnprocessedUpdatesCb = callbackify(countUnprocessedUpdates) +const getOldestDocUpdatesCb = callbackify(getOldestDocUpdates) +const deleteAppliedDocUpdatesCb = callbackify(deleteAppliedDocUpdates) +const destroyDocUpdatesQueueCb = callbackify(destroyDocUpdatesQueue) +const getProjectIdsWithHistoryOpsCb = callbackify(getProjectIdsWithHistoryOps) +const getProjectIdsWithHistoryOpsCountCb = callbackify( + getProjectIdsWithHistoryOpsCount +) +const setFirstOpTimestampCb = callbackify(setFirstOpTimestamp) +const getFirstOpTimestampCb = callbackify(getFirstOpTimestamp) +const clearFirstOpTimestampCb = callbackify(clearFirstOpTimestamp) +const getProjectIdsWithFirstOpTimestampsCb = callbackify( + getProjectIdsWithFirstOpTimestamps +) +const clearDanglingFirstOpTimestampCb = callbackify( + clearDanglingFirstOpTimestamp +) +const getCachedHistoryIdCb = callbackify(getCachedHistoryId) +const setCachedHistoryIdCb = callbackify(setCachedHistoryId) +const clearCachedHistoryIdCb = callbackify(clearCachedHistoryId) + +const getUpdatesInBatchesCb = function ( + projectId, + batchSize, + runner, + callback +) { + const runnerPromises = promisify(runner) + getUpdatesInBatches(projectId, batchSize, runnerPromises) + .then(result => { + callback(null, result) + }) + .catch(err => { + callback(err) + }) +} + +export { + countUnprocessedUpdatesCb as countUnprocessedUpdates, + getOldestDocUpdatesCb as getOldestDocUpdates, + deleteAppliedDocUpdatesCb as deleteAppliedDocUpdates, + destroyDocUpdatesQueueCb as destroyDocUpdatesQueue, + getUpdatesInBatchesCb as getUpdatesInBatches, + getProjectIdsWithHistoryOpsCb as getProjectIdsWithHistoryOps, + getProjectIdsWithHistoryOpsCountCb as getProjectIdsWithHistoryOpsCount, + setFirstOpTimestampCb as setFirstOpTimestamp, + getFirstOpTimestampCb as getFirstOpTimestamp, + clearFirstOpTimestampCb as clearFirstOpTimestamp, + getProjectIdsWithFirstOpTimestampsCb as getProjectIdsWithFirstOpTimestamps, + clearDanglingFirstOpTimestampCb as clearDanglingFirstOpTimestamp, + getCachedHistoryIdCb as getCachedHistoryId, + setCachedHistoryIdCb as setCachedHistoryId, + clearCachedHistoryIdCb as clearCachedHistoryId, +} + +export const promises = { + countUnprocessedUpdates, + getOldestDocUpdates, + deleteAppliedDocUpdates, + destroyDocUpdatesQueue, + getUpdatesInBatches, + getProjectIdsWithHistoryOps, + getProjectIdsWithHistoryOpsCount, + setFirstOpTimestamp, + getFirstOpTimestamp, + clearFirstOpTimestamp, + getProjectIdsWithFirstOpTimestamps, + clearDanglingFirstOpTimestamp, + getCachedHistoryId, + setCachedHistoryId, + clearCachedHistoryId, } diff --git a/services/project-history/app/js/UpdatesProcessor.js b/services/project-history/app/js/UpdatesProcessor.js index 503e0b1fd6..6bd2fea21d 100644 --- a/services/project-history/app/js/UpdatesProcessor.js +++ b/services/project-history/app/js/UpdatesProcessor.js @@ -19,11 +19,6 @@ import { Profiler } from './Profiler.js' const keys = Settings.redis.lock.key_schema -const PROJECT_HISTORY = { - ENABLED: 'enabled', - NOT_ENABLED: 'not-enabled', -} - export const REDIS_READ_BATCH_SIZE = 500 /** @@ -41,25 +36,28 @@ export function getRawUpdates(projectId, batchSize, callback) { if (error != null) { return callback(OError.tag(error)) } - RedisManager.parseDocUpdates(rawUpdates, (error, updates) => { + + let updates + try { + updates = RedisManager.parseDocUpdates(rawUpdates) + } catch (error) { + return callback(OError.tag(error)) + } + + _getHistoryId(projectId, updates, (error, historyId) => { if (error != null) { return callback(OError.tag(error)) } - _getHistoryId(projectId, updates, (error, historyId) => { - if (error != null) { - return callback(OError.tag(error)) - } - HistoryStoreManager.getMostRecentChunk( - projectId, - historyId, - (error, chunk) => { - if (error != null) { - return callback(OError.tag(error)) - } - callback(null, { project_id: projectId, chunk, updates }) + HistoryStoreManager.getMostRecentChunk( + projectId, + historyId, + (error, chunk) => { + if (error != null) { + return callback(OError.tag(error)) } - ) - }) + callback(null, { project_id: projectId, chunk, updates }) + } + ) }) } ) @@ -175,16 +173,11 @@ _mocks._countAndProcessUpdates = ( (updates, cb) => { _processUpdatesBatch(projectId, updates, extendLock, cb) }, - (error, isProjectHistoryEnabled) => { - // We can error before it is known whether project history is enabled - // for the project, so this key has 3 values. - const enabled = isProjectHistoryEnabled || 'unknown' - // This metrics key tries to convet that processing is not atomic. - // Some updates may have been processed even if there was an error. - const success = error != null ? 'with-error' : 'without-error' - metrics.gauge(`updates.${enabled}.${success}`, queueSize) - metrics.count(`updates.${enabled}.${success}`, queueSize) - callback(error, queueSize) + error => { + if (error) { + return callback(error) + } + callback(null, queueSize) } ) } else { @@ -210,14 +203,14 @@ function _processUpdatesBatch(projectId, updates, extendLock, callback) { { projectId }, 'discarding updates as project does not use history' ) - return callback(null, PROJECT_HISTORY.NOT_ENABLED) + return callback() } _processUpdates(projectId, historyId, updates, extendLock, error => { if (error != null) { - return callback(OError.tag(error), PROJECT_HISTORY.ENABLED) + return callback(OError.tag(error)) } - callback(null, PROJECT_HISTORY.ENABLED) + callback() }) }) } diff --git a/services/project-history/test/setup.js b/services/project-history/test/setup.js index f011481be3..86309460ca 100644 --- a/services/project-history/test/setup.js +++ b/services/project-history/test/setup.js @@ -1,6 +1,8 @@ import chai from 'chai' import sinonChai from 'sinon-chai' +import chaiAsPromised from 'chai-as-promised' // Chai configuration chai.should() chai.use(sinonChai) +chai.use(chaiAsPromised) diff --git a/services/project-history/test/unit/js/RedisManager/RedisManagerTests.js b/services/project-history/test/unit/js/RedisManager/RedisManagerTests.js index b63165baec..7652f7935a 100644 --- a/services/project-history/test/unit/js/RedisManager/RedisManagerTests.js +++ b/services/project-history/test/unit/js/RedisManager/RedisManagerTests.js @@ -1,20 +1,5 @@ -/* eslint-disable - camelcase, - mocha/no-identical-title, - no-return-assign, - no-undef, - no-unused-vars, -*/ -// TODO: This file was created by bulk-decaffeinate. -// Fix any style issues and re-enable lint. -/* - * decaffeinate suggestions: - * DS101: Remove unnecessary use of Array.from - * DS102: Remove unnecessary code created because of implicit returns - * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md - */ -import sinon from 'sinon' import { expect } from 'chai' +import sinon from 'sinon' import { strict as esmock } from 'esmock' const MODULE_PATH = '../../../../app/js/RedisManager.js' @@ -23,7 +8,7 @@ describe('RedisManager', function () { beforeEach(async function () { this.rclient = { auth: sinon.stub(), - exec: sinon.stub().yields(), + exec: sinon.stub().resolves(), lrange: sinon.stub(), lrem: sinon.stub(), srem: sinon.stub(), @@ -37,16 +22,17 @@ describe('RedisManager', function () { redis: { project_history: { key_schema: { - projectHistoryOps({ project_id }) { - return `Project:HistoryOps:${project_id}` + projectHistoryOps({ project_id: projectId }) { + return `Project:HistoryOps:{${projectId}}` }, - projectHistoryFirstOpTimestamp({ project_id }) { - return `ProjectHistory:FirstOpTimestamp:{${project_id}}` + projectHistoryFirstOpTimestamp({ project_id: projectId }) { + return `ProjectHistory:FirstOpTimestamp:{${projectId}}` }, }, }, }, } + this.Metrics = { timing: sinon.stub(), summary: sinon.stub(), @@ -59,769 +45,470 @@ describe('RedisManager', function () { }) this.project_id = 'project-id-123' - this.batch_size = 100 + this.batchSize = 100 + this.historyOpsKey = `Project:HistoryOps:{${this.project_id}}` + this.firstOpTimestampKey = `ProjectHistory:FirstOpTimestamp:{${this.project_id}}` this.updates = [ - { v: 42, op: 'mock-op-42' }, - { v: 45, op: 'mock-op-45' }, + { v: 42, op: ['a', 'b', 'c', 'd'] }, + { v: 45, op: ['e', 'f', 'g', 'h'] }, ] - this.json_updates = Array.from(this.updates).map(update => + this.extraUpdates = [{ v: 100, op: ['i', 'j', 'k'] }] + this.rawUpdates = this.updates.map(update => JSON.stringify(update)) + this.extraRawUpdates = this.extraUpdates.map(update => JSON.stringify(update) ) - - return (this.callback = sinon.stub()) }) describe('getOldestDocUpdates', function () { - beforeEach(function () { - this.rclient.lrange.yields(null, this.json_updates) - return this.RedisManager.getOldestDocUpdates( + beforeEach(async function () { + this.rclient.lrange.resolves(this.rawUpdates) + this.batchSize = 3 + this.result = await this.RedisManager.promises.getOldestDocUpdates( this.project_id, - this.batch_size, - this.callback + this.batchSize ) }) it('should read the updates from redis', function () { - return this.rclient.lrange - .calledWith( - `Project:HistoryOps:${this.project_id}`, - 0, - this.batch_size - 1 - ) + this.rclient.lrange + .calledWith(this.historyOpsKey, 0, this.batchSize - 1) .should.equal(true) }) - return it('should call the callback with the unparsed ops', function () { - return this.callback - .calledWith(null, this.json_updates) - .should.equal(true) + it('should call the callback with the unparsed ops', function () { + this.result.should.equal(this.rawUpdates) }) }) describe('parseDocUpdates', function () { - beforeEach(function () { - return this.RedisManager.parseDocUpdates(this.json_updates, this.callback) - }) - - return it('should call the callback with the parsed ops', function () { - return this.callback.calledWith(null, this.updates).should.equal(true) + it('should return the parsed ops', function () { + this.RedisManager.parseDocUpdates(this.rawUpdates).should.deep.equal( + this.updates + ) }) }) - describe('deleteAppliedDocUpdates', function () { + describe('getUpdatesInBatches', function () { beforeEach(function () { - return this.RedisManager.deleteAppliedDocUpdates( - this.project_id, - this.json_updates, - this.callback - ) - }) - - it('should delete the first update from redis', function () { - this.rclient.lrem.should.have.been.calledWith( - `Project:HistoryOps:${this.project_id}`, - 1, - this.json_updates[0] - ) - }) - - it('should delete the second update from redis', function () { - return this.rclient.lrem - .calledWith( - `Project:HistoryOps:${this.project_id}`, - 1, - this.json_updates[1] - ) - .should.equal(true) - }) - - it('should clear the first op timestamp', function () { - return this.rclient.del - .calledWith(`ProjectHistory:FirstOpTimestamp:{${this.project_id}}`) - .should.equal(true) - }) - - return it('should call the callback ', function () { - return this.callback.called.should.equal(true) - }) - }) - - return describe('getUpdatesInBatches', function () { - beforeEach(function () { - this.rawUpdates = ['raw-update-1', 'raw-update-2'] - this.expandedUpdates = ['expanded-update-1', 'expanded-update-2'] - this.RedisManager._mocks.deleteAppliedDocUpdates = sinon.stub().yields() - - this.isProjectHistoryEnabled = true - return (this.runner = sinon - .stub() - .yields(null, this.isProjectHistoryEnabled)) + this.runner = sinon.stub().resolves() }) describe('single batch smaller than batch size', function () { - beforeEach(function (done) { - this.RedisManager._mocks.getOldestDocUpdates = sinon - .stub() - .yields(null, this.rawUpdates) - this.RedisManager._mocks.parseDocUpdates = sinon - .stub() - .yields(null, this.expandedUpdates) - return this.RedisManager.getUpdatesInBatches( + beforeEach(async function () { + this.rclient.lrange.resolves(this.rawUpdates) + this.batchSize = 3 + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, - 3, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.batchSize, + this.runner ) }) it('requests a single batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 1 + this.rclient.lrange.should.have.been.calledOnce + this.rclient.lrange.should.have.been.calledWith( + this.historyOpsKey, + 0, + this.batchSize - 1 ) }) it('calls the runner once', function () { - return this.runner.callCount.should.equal(1) + this.runner.callCount.should.equal(1) }) it('calls the runner with the updates', function () { - return this.runner.calledWith(this.expandedUpdates).should.equal(true) + this.runner.calledWith(this.updates).should.equal(true) }) it('deletes the applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals(this.rawUpdates) + for (const update of this.rawUpdates) { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + update ) - .should.equal(true) + } }) - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + it('deletes the first op timestamp', function () { + expect(this.rclient.del).to.have.been.calledWith( + this.firstOpTimestampKey + ) }) }) describe('single batch at batch size', function () { - beforeEach(function (done) { - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates) - this.RedisManager._mocks.getOldestDocUpdates.onCall(1).yields(null, []) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates) - - return this.RedisManager.getUpdatesInBatches( + beforeEach(async function () { + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves([]) + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner once', function () { - return this.runner.callCount.should.equal(1) + this.runner.callCount.should.equal(1) }) it('calls the runner with the updates', function () { - return this.runner.calledWith(this.expandedUpdates).should.equal(true) + this.runner.calledWith(this.updates).should.equal(true) }) it('deletes the applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals(this.rawUpdates) + for (const update of this.rawUpdates) { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + update ) - .should.equal(true) + } }) - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + it('deletes the first op timestamp', function () { + expect(this.rclient.del).to.have.been.calledWith( + this.firstOpTimestampKey + ) }) }) describe('single batch exceeding size limit on updates', function () { - beforeEach(function (done) { - this.rawUpdates0 = ['raw-update-1-12345678', 'raw-update-2-12345678'] - this.rawUpdates1 = ['raw-update-2-12345678'] - this.expandedUpdates0 = ['expanded-update-1'] - this.expandedUpdates1 = ['expanded-update-2'] + beforeEach(async function () { // set the threshold below the size of the first update this.RedisManager.setRawUpdateSizeThreshold( - this.rawUpdates0[0].length - 1 + this.rawUpdates[0].length - 1 ) - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates0) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, this.rawUpdates1) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates0) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, this.expandedUpdates1) + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.rawUpdates.slice(1)) - return this.RedisManager.getUpdatesInBatches( + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) - it('calls the runner with the first updates', function () { - return this.runner.calledWith(this.expandedUpdates0).should.equal(true) + it('calls the runner with the first update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(0, 1)) }) - it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[0]]) - ) - .should.equal(true) + it('deletes the first update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[0] + ) }) - it('calls the runner with the second updates', function () { - return this.runner.calledWith(this.expandedUpdates1).should.equal(true) + it('calls the runner with the second update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(1)) }) it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[1]]) - ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[1] + ) }) }) describe('two batches with first update below and second update above the size limit on updates', function () { - beforeEach(function (done) { - this.rawUpdates0 = ['raw-update-1', 'raw-update-2-12345678'] - this.rawUpdates1 = ['raw-update-2-12345678'] - this.expandedUpdates0 = ['expanded-update-1'] - this.expandedUpdates1 = ['expanded-update-2'] + beforeEach(async function () { // set the threshold above the size of the first update, but below the total size this.RedisManager.setRawUpdateSizeThreshold( - this.rawUpdates0[0].length + 1 + this.rawUpdates[0].length + 1 ) - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates0) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, this.rawUpdates1) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates0) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, this.expandedUpdates1) - - return this.RedisManager.getUpdatesInBatches( + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.rawUpdates.slice(1)) + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) it('calls the runner with the first update', function () { - return this.runner.calledWith(this.expandedUpdates0).should.equal(true) + this.runner.calledWith(this.updates.slice(0, 1)).should.equal(true) }) - it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[0]]) - ) - .should.equal(true) + it('deletes the first set applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[0] + ) }) it('calls the runner with the second update', function () { - return this.runner.calledWith(this.expandedUpdates1).should.equal(true) + this.runner.calledWith(this.updates.slice(1)).should.equal(true) }) - it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[1]]) - ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + it('deletes the second applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[1] + ) }) }) describe('single batch exceeding op count limit on updates', function () { - beforeEach(function (done) { - this.rawUpdates0 = [ - "{op: ['a', 'b', 'c', 'd']}", - "{op:['e', 'f', 'g', 'h']}", - ] - this.rawUpdates1 = ["{op:['e', 'f', 'g', 'h']}"] - this.expandedUpdates0 = [ - { op: ['a', 'b', 'c', 'd'] }, - { op: ['e', 'f', 'g', 'h'] }, - ] - this.expandedUpdates1 = [{ op: ['e', 'f', 'g', 'h'] }] + beforeEach(async function () { // set the threshold below the size of the first update - this.RedisManager.setMaxUpdateOpLength( - this.expandedUpdates0[0].op.length - 1 - ) - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates0) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, this.rawUpdates1) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates0) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, this.expandedUpdates1) + this.RedisManager.setMaxUpdateOpLength(this.updates[0].op.length - 1) + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.rawUpdates.slice(1)) - return this.RedisManager.getUpdatesInBatches( + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) it('calls the runner with the first updates', function () { - return this.runner - .calledWith([this.expandedUpdates0[0]]) - .should.equal(true) + this.runner.calledWith(this.updates.slice(0, 1)).should.equal(true) }) - it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[0]]) - ) - .should.equal(true) + it('deletes the first applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[0] + ) }) it('calls the runner with the second updates', function () { - return this.runner.calledWith(this.expandedUpdates1).should.equal(true) + this.runner.calledWith(this.updates.slice(1)).should.equal(true) }) - it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[1]]) - ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + it('deletes the second applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[1] + ) }) }) describe('single batch exceeding doc content count', function () { - beforeEach(function (done) { - this.rawUpdates0 = [ - '{resyncDocContent: 123}', - '{resyncDocContent: 456}', - ] - this.rawUpdates1 = ['{resyncDocContent: 456}'] - this.expandedUpdates0 = [ - { resyncDocContent: 123 }, - { resyncDocContent: 456 }, - ] - this.expandedUpdates1 = [{ resyncDocContent: 456 }] + beforeEach(async function () { + this.updates = [{ resyncDocContent: 123 }, { resyncDocContent: 456 }] + this.rawUpdates = this.updates.map(update => JSON.stringify(update)) // set the threshold below the size of the first update - this.RedisManager.setMaxNewDocContentCount( - this.expandedUpdates0.length - 1 - ) - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates0) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, this.rawUpdates1) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates0) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, this.expandedUpdates1) + this.RedisManager.setMaxNewDocContentCount(this.updates.length - 1) + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.rawUpdates.slice(1)) - return this.RedisManager.getUpdatesInBatches( + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) - it('calls the runner with the first updates', function () { - return this.runner - .calledWith([this.expandedUpdates0[0]]) - .should.equal(true) + it('calls the runner with the first update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(0, 1)) }) - it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[0]]) - ) - .should.equal(true) + it('deletes the first applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[0] + ) }) - it('calls the runner with the second updates', function () { - return this.runner.calledWith(this.expandedUpdates1).should.equal(true) + it('calls the runner with the second update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(1)) }) it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[1]]) - ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[1] + ) }) }) - describe('two batches with first update below and second update above the size limit on updates', function () { - beforeEach(function (done) { - this.rawUpdates0 = [ - "{op: ['a', 'b', 'c', 'd']}", - "{op:['e', 'f', 'g', 'h']}", - ] - this.rawUpdates1 = ["{op:['e', 'f', 'g', 'h']}"] - this.expandedUpdates0 = [ - { op: ['a', 'b', 'c', 'd'] }, - { op: ['e', 'f', 'g', 'h'] }, - ] - this.expandedUpdates1 = [{ op: ['e', 'f', 'g', 'h'] }] + describe('two batches with first update below and second update above the ops length limit on updates', function () { + beforeEach(async function () { // set the threshold below the size of the first update - this.RedisManager.setMaxUpdateOpLength( - this.expandedUpdates0[0].op.length + 1 - ) - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates0) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, this.rawUpdates1) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates0) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, this.expandedUpdates1) + this.RedisManager.setMaxUpdateOpLength(this.updates[0].op.length + 1) + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.rawUpdates.slice(1)) - return this.RedisManager.getUpdatesInBatches( + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) - it('calls the runner with the first updates', function () { - return this.runner - .calledWith([this.expandedUpdates0[0]]) - .should.equal(true) + it('calls the runner with the first update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(0, 1)) }) - it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[0]]) - ) - .should.equal(true) + it('deletes the first applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[0] + ) }) - it('calls the runner with the second updates', function () { - return this.runner.calledWith(this.expandedUpdates1).should.equal(true) + it('calls the runner with the second update', function () { + this.runner.should.have.been.calledWith(this.updates.slice(1)) }) - it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals([this.rawUpdates0[1]]) - ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + it('deletes the second applied update', function () { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + this.rawUpdates[1] + ) }) }) describe('two batches', function () { - beforeEach(function (done) { - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(null, ['raw-update-3']) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates) - this.RedisManager._mocks.parseDocUpdates - .onCall(1) - .yields(null, ['expanded-update-3']) - - return this.RedisManager.getUpdatesInBatches( + beforeEach(async function () { + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).resolves(this.extraRawUpdates) + await this.RedisManager.promises.getUpdatesInBatches( this.project_id, 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } + this.runner ) }) it('requests a second batch of updates', function () { - return this.RedisManager._mocks.getOldestDocUpdates.callCount.should.equal( - 2 - ) + this.rclient.lrange.should.have.been.calledTwice }) it('calls the runner twice', function () { - return this.runner.callCount.should.equal(2) + this.runner.callCount.should.equal(2) }) it('calls the runner with the updates', function () { - return this.runner.calledWith(this.expandedUpdates).should.equal(true) + this.runner.should.have.been.calledWith(this.updates) + this.runner.should.have.been.calledWith(this.extraUpdates) }) it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals(this.rawUpdates) + for (const update of this.rawUpdates) { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + update ) - .should.equal(true) + } }) it('deletes the second set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals(['raw-update-3']) + for (const update of this.extraRawUpdates) { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + update ) - .should.equal(true) - }) - - return it('calls the callback with the result of the runner', function () { - return this.callback - .calledWith(null, this.isProjectHistoryEnabled) - .should.equal(true) + } }) }) describe('error when first reading updates', function () { - beforeEach(function (done) { + beforeEach(async function () { this.error = new Error('error') - this.RedisManager._mocks.getOldestDocUpdates = sinon - .stub() - .yields(this.error) - return this.RedisManager.getUpdatesInBatches( - this.project_id, - 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } - ) + this.rclient.lrange.rejects(this.error) + await expect( + this.RedisManager.promises.getUpdatesInBatches( + this.project_id, + 2, + this.runner + ) + ).to.be.rejected }) it('does not delete any updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates.called.should.equal( - false - ) - }) - - return it('calls the callback with the error', function () { - return this.callback - .calledWith(this.error, undefined) - .should.equal(true) + expect(this.rclient.lrem).not.to.have.been.called }) }) - return describe('error when reading updates for a second batch', function () { - beforeEach(function (done) { + describe('error when reading updates for a second batch', function () { + beforeEach(async function () { this.error = new Error('error') - this.RedisManager._mocks.getOldestDocUpdates = sinon.stub() - this.RedisManager._mocks.getOldestDocUpdates - .onCall(0) - .yields(null, this.rawUpdates) - this.RedisManager._mocks.getOldestDocUpdates - .onCall(1) - .yields(this.error) - this.RedisManager._mocks.parseDocUpdates = sinon.stub() - this.RedisManager._mocks.parseDocUpdates - .onCall(0) - .yields(null, this.expandedUpdates) + this.rclient.lrange.onCall(0).resolves(this.rawUpdates) + this.rclient.lrange.onCall(1).rejects(this.error) - return this.RedisManager.getUpdatesInBatches( - this.project_id, - 2, - this.runner, - (error, isProjectHistoryEnabled) => { - this.callback(error, isProjectHistoryEnabled) - return done() - } - ) + await expect( + this.RedisManager.promises.getUpdatesInBatches( + this.project_id, + 2, + this.runner + ) + ).to.be.rejected }) it('deletes the first set of applied updates', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates - .calledWith( - this.project_id, - sinon.match.array.deepEquals(this.rawUpdates) + for (const update of this.rawUpdates) { + expect(this.rclient.lrem).to.have.been.calledWith( + this.historyOpsKey, + 1, + update ) - .should.equal(true) + } }) it('deletes applied updates only once', function () { - return this.RedisManager._mocks.deleteAppliedDocUpdates.callCount.should.equal( - 1 - ) - }) - - return it('calls the callback with the error and the first result of the runner', function () { - return this.callback - .calledWith(this.error, this.isProjectHistoryEnabled) - .should.equal(true) + expect(this.rclient.lrem.callCount).to.equal(this.rawUpdates.length) }) }) })