From 31e1808dd89b717da096f20051eb2ed90e56e0f5 Mon Sep 17 00:00:00 2001 From: Henry Oswald Date: Mon, 8 Feb 2021 12:19:25 +0000 Subject: [PATCH] shard the pending-updates-list queue --- .../app/js/DocumentUpdaterManager.js | 25 ++++++++-- .../real-time/config/settings.defaults.js | 6 +++ .../unit/js/DocumentUpdaterControllerTests.js | 3 +- .../unit/js/DocumentUpdaterManagerTests.js | 47 +++++++++++++++---- 4 files changed, 65 insertions(+), 16 deletions(-) diff --git a/services/real-time/app/js/DocumentUpdaterManager.js b/services/real-time/app/js/DocumentUpdaterManager.js index 925d9f38d8..aacadd9391 100644 --- a/services/real-time/app/js/DocumentUpdaterManager.js +++ b/services/real-time/app/js/DocumentUpdaterManager.js @@ -88,6 +88,15 @@ const DocumentUpdaterManager = { }) }, + _getPendingUpdateListKey() { + const shard = _.random(0, settings.pendingUpdateListShardCount) + if (shard === 0) { + return 'pending-updates-list' + } else { + return `pending-updates-list-${shard}` + } + }, + queueChange(project_id, doc_id, change, callback) { const allowedKeys = [ 'doc', @@ -123,12 +132,18 @@ const DocumentUpdaterManager = { error = new OError('error pushing update into redis').withCause(error) return callback(error) } - rclient.rpush('pending-updates-list', doc_key, function (error) { - if (error) { - error = new OError('error pushing doc_id into redis').withCause(error) + rclient.rpush( + DocumentUpdaterManager._getPendingUpdateListKey(), + doc_key, + function (error) { + if (error) { + error = new OError('error pushing doc_id into redis').withCause( + error + ) + } + callback(error) } - callback(error) - }) + ) }) } } diff --git a/services/real-time/config/settings.defaults.js b/services/real-time/config/settings.defaults.js index 5c0d501b50..1d7f0d4c93 100644 --- a/services/real-time/config/settings.defaults.js +++ b/services/real-time/config/settings.defaults.js @@ -114,6 +114,12 @@ const settings = { max_doc_length: 2 * 1024 * 1024, // 2mb + // should be set to the same same as dispatcherCount in document updater + pendingUpdateListShardCount: parseInt( + process.env.PENDING_UPDATE_LIST_SHARD_COUNT || 1, + 10 + ), + // combine // max_doc_length (2mb see above) * 2 (delete + insert) // max_ranges_size (3mb see MAX_RANGES_SIZE in document-updater) diff --git a/services/real-time/test/unit/js/DocumentUpdaterControllerTests.js b/services/real-time/test/unit/js/DocumentUpdaterControllerTests.js index 738375b32a..0aabc51de7 100644 --- a/services/real-time/test/unit/js/DocumentUpdaterControllerTests.js +++ b/services/real-time/test/unit/js/DocumentUpdaterControllerTests.js @@ -20,7 +20,7 @@ const modulePath = require('path').join( const MockClient = require('./helpers/MockClient') describe('DocumentUpdaterController', function () { - beforeEach(function (done) { + beforeEach(function () { this.project_id = 'project-id-123' this.doc_id = 'doc-id-123' this.callback = sinon.stub() @@ -69,7 +69,6 @@ describe('DocumentUpdaterController', function () { './ChannelManager': (this.ChannelManager = {}) } }) - done() }) describe('listenForUpdatesFromDocumentUpdater', function () { diff --git a/services/real-time/test/unit/js/DocumentUpdaterManagerTests.js b/services/real-time/test/unit/js/DocumentUpdaterManagerTests.js index d6da97ddab..e3c108b824 100644 --- a/services/real-time/test/unit/js/DocumentUpdaterManagerTests.js +++ b/services/real-time/test/unit/js/DocumentUpdaterManagerTests.js @@ -15,6 +15,7 @@ const sinon = require('sinon') const SandboxedModule = require('sandboxed-module') const path = require('path') const modulePath = '../../../app/js/DocumentUpdaterManager' +const _ = require('underscore') describe('DocumentUpdaterManager', function () { beforeEach(function () { @@ -34,7 +35,8 @@ describe('DocumentUpdaterManager', function () { } } }, - maxUpdateSize: 7 * 1024 * 1024 + maxUpdateSize: 7 * 1024 * 1024, + pendingUpdateListShardCount: 10 } this.rclient = { auth() {} } @@ -256,7 +258,7 @@ describe('DocumentUpdaterManager', function () { }) }) - return describe('queueChange', function () { + describe('queueChange', function () { beforeEach(function () { this.change = { doc: '1234567890', @@ -269,7 +271,12 @@ describe('DocumentUpdaterManager', function () { describe('successfully', function () { beforeEach(function () { - return this.DocumentUpdaterManager.queueChange( + this.pendingUpdateListKey = `pending-updates-list-key-${Math.random()}` + + this.DocumentUpdaterManager._getPendingUpdateListKey = sinon + .stub() + .returns(this.pendingUpdateListKey) + this.DocumentUpdaterManager.queueChange( this.project_id, this.doc_id, this.change, @@ -278,7 +285,7 @@ describe('DocumentUpdaterManager', function () { }) it('should push the change', function () { - return this.rclient.rpush + this.rclient.rpush .calledWith( `PendingUpdates:${this.doc_id}`, JSON.stringify(this.change) @@ -286,10 +293,10 @@ describe('DocumentUpdaterManager', function () { .should.equal(true) }) - return it('should notify the doc updater of the change via the pending-updates-list queue', function () { - return this.rclient.rpush + it('should notify the doc updater of the change via the pending-updates-list queue', function () { + this.rclient.rpush .calledWith( - 'pending-updates-list', + this.pendingUpdateListKey, `${this.project_id}:${this.doc_id}` ) .should.equal(true) @@ -366,7 +373,7 @@ describe('DocumentUpdaterManager', function () { }) }) - return describe('with invalid keys', function () { + describe('with invalid keys', function () { beforeEach(function () { this.change = { op: [{ d: 'test', p: 345 }], @@ -380,7 +387,7 @@ describe('DocumentUpdaterManager', function () { ) }) - return it('should remove the invalid keys from the change', function () { + it('should remove the invalid keys from the change', function () { return this.rclient.rpush .calledWith( `PendingUpdates:${this.doc_id}`, @@ -390,4 +397,26 @@ describe('DocumentUpdaterManager', function () { }) }) }) + + describe('_getPendingUpdateListKey', function () { + beforeEach(function () { + const keys = _.times( + 10000, + this.DocumentUpdaterManager._getPendingUpdateListKey + ) + this.keys = _.unique(keys) + }) + it('should return normal pending updates key', function () { + const key = this.DocumentUpdaterManager._getPendingUpdateListKey() + _.contains(this.keys, 'pending-updates-list').should.equal(true) + }) + + it('should return pending-updates-list-n keys', function () { + _.contains(this.keys, 'pending-updates-list-1').should.equal(true) + _.contains(this.keys, 'pending-updates-list-3').should.equal(true) + }) + it('should not include pending-updates-list-0 key', function () { + _.contains(this.keys, 'pending-updates-list-0').should.equal(false) + }) + }) })