From 3d70f9126e5f44ae94a3c81bd19323a907ae1721 Mon Sep 17 00:00:00 2001 From: James Allen Date: Fri, 28 Feb 2014 18:29:05 +0000 Subject: [PATCH] Flush track changes api every 50 updates --- .../app/coffee/DocOpsManager.coffee | 3 +- .../app/coffee/DocumentManager.coffee | 5 --- .../app/coffee/TrackChangesManager.coffee | 15 +++++++- .../coffee/ApplyingUpdatesToADocTests.coffee | 28 ++++++++++++++- .../coffee/FlushingAProjectTests.coffee | 13 ------- .../coffee/FlushingDocsTests.coffee | 10 ------ .../DocOpsManager/DocOpsManagerTests.coffee | 5 +-- .../DocumentManager/flushDocTests.coffee | 8 ----- .../pushUncompressedHistoryOpTests.coffee | 6 ++-- .../TrackChangesManagerTests.coffee | 35 +++++++++++++++++++ 10 files changed, 84 insertions(+), 44 deletions(-) diff --git a/services/document-updater/app/coffee/DocOpsManager.coffee b/services/document-updater/app/coffee/DocOpsManager.coffee index 971db83358..180f1e564b 100644 --- a/services/document-updater/app/coffee/DocOpsManager.coffee +++ b/services/document-updater/app/coffee/DocOpsManager.coffee @@ -5,6 +5,7 @@ ObjectId = mongojs.ObjectId logger = require "logger-sharelatex" async = require "async" Metrics = require("./Metrics") +TrackChangesManager = require "./TrackChangesManager" module.exports = DocOpsManager = flushDocOpsToMongo: (project_id, doc_id, _callback = (error) ->) -> @@ -46,7 +47,7 @@ module.exports = DocOpsManager = pushDocOp: (project_id, doc_id, op, callback = (error) ->) -> RedisManager.pushDocOp doc_id, op, (error, version) -> return callback(error) if error? - RedisManager.pushUncompressedHistoryOp doc_id, op, (error) -> + TrackChangesManager.pushUncompressedHistoryOp doc_id, op, (error) -> return callback(error) if error? callback null, version diff --git a/services/document-updater/app/coffee/DocumentManager.coffee b/services/document-updater/app/coffee/DocumentManager.coffee index 2f3cfb8d79..aa64ac3d7f 100644 --- a/services/document-updater/app/coffee/DocumentManager.coffee +++ b/services/document-updater/app/coffee/DocumentManager.coffee @@ -2,7 +2,6 @@ RedisManager = require "./RedisManager" PersistenceManager = require "./PersistenceManager" DocOpsManager = require "./DocOpsManager" DiffCodec = require "./DiffCodec" -TrackChangesManager = require "./TrackChangesManager" logger = require "logger-sharelatex" Metrics = require "./Metrics" @@ -82,10 +81,6 @@ module.exports = DocumentManager = timer.done() _callback(args...) - TrackChangesManager.flushDocChanges doc_id, (error) -> - if error? - logger.error err: error, project_id: project_id, doc_id: doc_id, "error flushing doc to track changes api" - RedisManager.getDoc doc_id, (error, lines, version) -> return callback(error) if error? if !lines? or !version? diff --git a/services/document-updater/app/coffee/TrackChangesManager.coffee b/services/document-updater/app/coffee/TrackChangesManager.coffee index c0b40331dc..489694fbf4 100644 --- a/services/document-updater/app/coffee/TrackChangesManager.coffee +++ b/services/document-updater/app/coffee/TrackChangesManager.coffee @@ -1,8 +1,9 @@ settings = require "settings-sharelatex" request = require "request" logger = require "logger-sharelatex" +RedisManager = require "./RedisManager" -module.exports = +module.exports = TrackChangesManager = flushDocChanges: (doc_id, callback = (error) ->) -> if !settings.apis?.trackchanges? logger.warn doc_id: doc_id, "track changes API is not configured, so not flushing" @@ -18,3 +19,15 @@ module.exports = else error = new Error("track changes api returned a failure status code: #{res.statusCode}") return callback(error) + + FLUSH_EVERY_N_OPS: 50 + pushUncompressedHistoryOp: (doc_id, op, callback = (error) ->) -> + RedisManager.pushUncompressedHistoryOp doc_id, op, (error, length) -> + if length > 0 and length % TrackChangesManager.FLUSH_EVERY_N_OPS == 0 + # Do this in the background since it uses HTTP and so may be too + # slow to wait for when processing a doc update. + logger.log length: length, doc_id: doc_id, "flushing track changes api" + TrackChangesManager.flushDocChanges doc_id, (error) -> + if error? + logger.error err: error, project_id: project_id, doc_id: doc_id, "error flushing doc to track changes api" + callback() diff --git a/services/document-updater/test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee b/services/document-updater/test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee index ac1a24223a..5b867b5807 100644 --- a/services/document-updater/test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee +++ b/services/document-updater/test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee @@ -7,6 +7,7 @@ db = mongojs.db ObjectId = mongojs.ObjectId rclient = require("redis").createClient() +MockTrackChangesApi = require "./helpers/MockTrackChangesApi" MockWebApi = require "./helpers/MockWebApi" DocUpdaterClient = require "./helpers/DocUpdaterClient" @@ -230,4 +231,29 @@ describe "Applying updates to a doc", -> DocUpdaterClient.getDoc @project_id, @doc_id, (error, res, doc) => doc.lines.should.deep.equal @lines done() - + + describe "with enough updates to flush to the track changes api", -> + before (done) -> + [@project_id, @doc_id] = [DocUpdaterClient.randomId(), DocUpdaterClient.randomId()] + MockWebApi.insertDoc @project_id, @doc_id, { + lines: @lines + } + @updates = [] + for v in [0..99] # Should flush after 50 ops + @updates.push + doc_id: @doc_id, + op: [i: v.toString(), p: 0] + v: v + + sinon.spy MockTrackChangesApi, "flushDoc" + + DocUpdaterClient.sendUpdates @project_id, @doc_id, @updates, (error) => + throw error if error? + setTimeout done, 200 + + after -> + MockTrackChangesApi.flushDoc.restore() + + it "should flush the doc twice", -> + console.log MockTrackChangesApi.flushDoc.args + MockTrackChangesApi.flushDoc.calledTwice.should.equal true diff --git a/services/document-updater/test/acceptance/coffee/FlushingAProjectTests.coffee b/services/document-updater/test/acceptance/coffee/FlushingAProjectTests.coffee index 9adbc2458c..b78decd5a9 100644 --- a/services/document-updater/test/acceptance/coffee/FlushingAProjectTests.coffee +++ b/services/document-updater/test/acceptance/coffee/FlushingAProjectTests.coffee @@ -4,7 +4,6 @@ chai.should() async = require "async" MockWebApi = require "./helpers/MockWebApi" -MockTrackChangesApi = require "./helpers/MockTrackChangesApi" DocUpdaterClient = require "./helpers/DocUpdaterClient" describe "Flushing a project", -> @@ -41,7 +40,6 @@ describe "Flushing a project", -> describe "with documents which have been updated", -> before (done) -> sinon.spy MockWebApi, "setDocumentLines" - sinon.spy MockTrackChangesApi, "flushDoc" async.series @docs.map((doc) => (callback) => @@ -59,7 +57,6 @@ describe "Flushing a project", -> after -> MockWebApi.setDocumentLines.restore() - MockTrackChangesApi.flushDoc.restore() it "should return a 204 status code", -> @statusCode.should.equal 204 @@ -78,13 +75,3 @@ describe "Flushing a project", -> callback() ), done - it "should flush the docs in the track changes api", (done) -> - # This is done in the background, so wait a little while to ensure it has happened - setTimeout () => - async.series @docs.map((doc) => - (callback) => - MockTrackChangesApi.flushDoc.calledWith(doc.id).should.equal true - ), done - done() - , 100 - diff --git a/services/document-updater/test/acceptance/coffee/FlushingDocsTests.coffee b/services/document-updater/test/acceptance/coffee/FlushingDocsTests.coffee index 6d67dd68a0..da0036bd02 100644 --- a/services/document-updater/test/acceptance/coffee/FlushingDocsTests.coffee +++ b/services/document-updater/test/acceptance/coffee/FlushingDocsTests.coffee @@ -4,7 +4,6 @@ chai.should() async = require "async" MockWebApi = require "./helpers/MockWebApi" -MockTrackChangesApi = require "./helpers/MockTrackChangesApi" DocUpdaterClient = require "./helpers/DocUpdaterClient" mongojs = require "../../../app/js/mongojs" db = mongojs.db @@ -32,7 +31,6 @@ describe "Flushing a doc to Mongo", -> lines: @lines } sinon.spy MockWebApi, "setDocumentLines" - sinon.spy MockTrackChangesApi, "flushDoc" DocUpdaterClient.sendUpdates @project_id, @doc_id, [@update], (error) => throw error if error? @@ -42,7 +40,6 @@ describe "Flushing a doc to Mongo", -> after -> MockWebApi.setDocumentLines.restore() - MockTrackChangesApi.flushDoc.restore() it "should flush the updated document to the web api", -> MockWebApi.setDocumentLines @@ -55,13 +52,6 @@ describe "Flushing a doc to Mongo", -> doc.docOps[0].op.should.deep.equal @update.op done() - it "should flush the doc in the track changes api", (done) -> - # This is done in the background, so wait a little while to ensure it has happened - setTimeout () => - MockTrackChangesApi.flushDoc.calledWith(@doc_id).should.equal true - done() - , 100 - describe "when the doc has a large number of ops to be flushed", -> before (done) -> [@project_id, @doc_id] = [DocUpdaterClient.randomId(), DocUpdaterClient.randomId()] diff --git a/services/document-updater/test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee b/services/document-updater/test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee index df2a546480..fe13d1cd55 100644 --- a/services/document-updater/test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee +++ b/services/document-updater/test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee @@ -19,6 +19,7 @@ describe "DocOpsManager", -> "./Metrics": @Metrics = Timer: class Timer done: sinon.stub() + "./TrackChangesManager": @TrackChangesManager = {} describe "flushDocOpsToMongo", -> describe "when versions are consistent", -> @@ -310,7 +311,7 @@ describe "DocOpsManager", -> beforeEach -> @op = "mock-op" @RedisManager.pushDocOp = sinon.stub().callsArgWith(2, null, @version = 42) - @RedisManager.pushUncompressedHistoryOp = sinon.stub().callsArg(2) + @TrackChangesManager.pushUncompressedHistoryOp = sinon.stub().callsArg(2) @DocOpsManager.pushDocOp @project_id, @doc_id, @op, @callback it "should push the op in to the docOps list", -> @@ -319,7 +320,7 @@ describe "DocOpsManager", -> .should.equal true it "should push the op into the pushUncompressedHistoryOp", -> - @RedisManager.pushUncompressedHistoryOp + @TrackChangesManager.pushUncompressedHistoryOp .calledWith(@doc_id, @op) .should.equal true diff --git a/services/document-updater/test/unit/coffee/DocumentManager/flushDocTests.coffee b/services/document-updater/test/unit/coffee/DocumentManager/flushDocTests.coffee index 5a4adc4a36..079341a536 100644 --- a/services/document-updater/test/unit/coffee/DocumentManager/flushDocTests.coffee +++ b/services/document-updater/test/unit/coffee/DocumentManager/flushDocTests.coffee @@ -10,7 +10,6 @@ describe "DocumentUpdater - flushDocIfLoaded", -> "./RedisManager": @RedisManager = {} "./PersistenceManager": @PersistenceManager = {} "./DocOpsManager": @DocOpsManager = {} - "./TrackChangesManager": @TrackChangesManager = {} "logger-sharelatex": @logger = {log: sinon.stub()} "./Metrics": @Metrics = Timer: class Timer @@ -26,7 +25,6 @@ describe "DocumentUpdater - flushDocIfLoaded", -> @RedisManager.getDoc = sinon.stub().callsArgWith(1, null, @lines, @version) @PersistenceManager.setDoc = sinon.stub().callsArgWith(3) @DocOpsManager.flushDocOpsToMongo = sinon.stub().callsArgWith(2) - @TrackChangesManager.flushDocChanges = sinon.stub().callsArg(1) @DocumentManager.flushDocIfLoaded @project_id, @doc_id, @callback it "should get the doc from redis", -> @@ -50,17 +48,11 @@ describe "DocumentUpdater - flushDocIfLoaded", -> it "should time the execution", -> @Metrics.Timer::done.called.should.equal true - it "should flush the doc in the track changes api", -> - @TrackChangesManager.flushDocChanges - .calledWith(@doc_id) - .should.equal true - describe "when the document is not in Redis", -> beforeEach -> @RedisManager.getDoc = sinon.stub().callsArgWith(1, null, null, null) @PersistenceManager.setDoc = sinon.stub().callsArgWith(3) @DocOpsManager.flushDocOpsToMongo = sinon.stub().callsArgWith(2) - @TrackChangesManager.flushDocChanges = sinon.stub().callsArg(1) @DocumentManager.flushDocIfLoaded @project_id, @doc_id, @callback it "should get the doc from redis", -> diff --git a/services/document-updater/test/unit/coffee/RedisManager/pushUncompressedHistoryOpTests.coffee b/services/document-updater/test/unit/coffee/RedisManager/pushUncompressedHistoryOpTests.coffee index 3b743db6e4..415f8e4572 100644 --- a/services/document-updater/test/unit/coffee/RedisManager/pushUncompressedHistoryOpTests.coffee +++ b/services/document-updater/test/unit/coffee/RedisManager/pushUncompressedHistoryOpTests.coffee @@ -19,7 +19,7 @@ describe "RedisManager.pushUncompressedHistoryOp", -> describe "successfully", -> beforeEach -> @op = { op: [{ i: "foo", p: 4 }] } - @rclient.rpush = sinon.stub().callsArg(2) + @rclient.rpush = sinon.stub().callsArgWith(2, null, @length = 42) @RedisManager.pushUncompressedHistoryOp @doc_id, @op, @callback it "should push the doc op into the doc ops list", -> @@ -27,8 +27,8 @@ describe "RedisManager.pushUncompressedHistoryOp", -> .calledWith("UncompressedHistoryOps:#{@doc_id}", JSON.stringify(@op)) .should.equal true - it "should call the callback", -> - @callback.called.should.equal true + it "should call the callback with the length", -> + @callback.calledWith(null, @length).should.equal true diff --git a/services/document-updater/test/unit/coffee/TrackChangesManager/TrackChangesManagerTests.coffee b/services/document-updater/test/unit/coffee/TrackChangesManager/TrackChangesManagerTests.coffee index 672cdcfaa4..6ff83d7414 100644 --- a/services/document-updater/test/unit/coffee/TrackChangesManager/TrackChangesManagerTests.coffee +++ b/services/document-updater/test/unit/coffee/TrackChangesManager/TrackChangesManagerTests.coffee @@ -8,6 +8,8 @@ describe "TrackChangesManager", -> @TrackChangesManager = SandboxedModule.require modulePath, requires: "request": @request = {} "settings-sharelatex": @Settings = {} + "logger-sharelatex": @logger = { log: sinon.stub() } + "./RedisManager": @RedisManager = {} @doc_id = "mock-doc-id" @callback = sinon.stub() @@ -36,3 +38,36 @@ describe "TrackChangesManager", -> it "should return the callback with an error", -> @callback.calledWith(new Error("track changes api return non-success code: 500")).should.equal true + + describe "pushUncompressedHistoryOp", -> + beforeEach -> + @op = "mock-op" + @TrackChangesManager.flushDocChanges = sinon.stub().callsArg(1) + + describe "pushing the op", -> + beforeEach -> + @RedisManager.pushUncompressedHistoryOp = sinon.stub().callsArgWith(2, null, 1) + @TrackChangesManager.pushUncompressedHistoryOp @doc_id, @op, @callback + + it "should push the op into redis", -> + @RedisManager.pushUncompressedHistoryOp + .calledWith(@doc_id, @op) + .should.equal true + + it "should call the callback", -> + @callback.called.should.equal true + + it "should not try to flush the op", -> + @TrackChangesManager.flushDocChanges.called.should.equal false + + describe "when there are a multiple of FLUSH_EVERY_N_OPS ops", -> + beforeEach -> + @RedisManager.pushUncompressedHistoryOp = + sinon.stub().callsArgWith(2, null, 2 * @TrackChangesManager.FLUSH_EVERY_N_OPS) + @TrackChangesManager.pushUncompressedHistoryOp @doc_id, @op, @callback + + it "should tell the track changes api to flush", -> + @TrackChangesManager.flushDocChanges + .calledWith(@doc_id) + .should.equal true +