ensure document is in redis before consuming ops

services/document-updater/app/coffee/DocumentManager.coffee

@@ -28,6 +28,13 @@ module.exports = DocumentManager =
 			else
 				callback null, lines, version, ranges, true
 
+	ensureDocIsLoaded: (project_id, doc_id, callback = (error) ->) ->
+		RedisManager.checkDocInMemory project_id, doc_id, (error) ->
+			if error?
+				DocumentManager.getDoc project_id, doc_id, callback
+			else
+				callback()
+
 	getDocAndRecentOps: (project_id, doc_id, fromVersion, _callback = (error, lines, version, recentOps, ranges) ->) ->
 		timer = new Metrics.Timer("docManager.getDocAndRecentOps")
 		callback = (args...) ->

services/document-updater/app/coffee/RedisManager.coffee

@@ -30,6 +30,12 @@ historyKeys = Settings.redis.history.key_schema
 module.exports = RedisManager =
 	rclient: rclient
 
+	checkDocInMemory: (project_id, doc_id, callback) ->
+		rclient.exists keys.docLines(doc_id:doc_id), (error, result) ->
+			return callback(error) if error?
+			return callback new Error("doc not in memory") if result isnt 1
+			callback()
+
 	putDocInMemory : (project_id, doc_id, docLines, version, ranges, _callback)->
 		timer = new metrics.Timer("redis.put-doc")
 		callback = (error) ->

services/document-updater/app/coffee/UpdateManager.coffee

@@ -14,10 +14,14 @@ RangesManager = require "./RangesManager"
 module.exports = UpdateManager =
 	processOutstandingUpdates: (project_id, doc_id, callback = (error) ->) ->
 		timer = new Metrics.Timer("updateManager.processOutstandingUpdates")
-		UpdateManager.fetchAndApplyUpdates project_id, doc_id, (error) ->
-			timer.done()
+		DocumentManager.ensureDocIsLoaded project_id, doc_id, (error) ->
+			# an error at this point is safe, because we haven't consumed any ops yet
 			return callback(error) if error?
-			callback()
+			UpdateManager.fetchAndApplyUpdates project_id, doc_id, (error) ->
+				timer.done()
+				# now we have taken ops off the queue so errors here will cause data loss
+				return callback(error) if error?
+				callback()
 
 	processOutstandingUpdatesWithLock: (project_id, doc_id, callback = (error) ->) ->
 		LockManager.tryLock doc_id, (error, gotLock, lockValue) =>

services/document-updater/test/unit/coffee/DocumentManager/DocumentManagerTests.coffee

@@ -188,7 +188,48 @@ describe "DocumentManager", ->
 
 			it "should time the execution", ->
 				@Metrics.Timer::done.called.should.equal true
-	
+
+
+	describe "ensureDocIsLoaded", ->
+		describe "when the doc exists in Redis", ->
+			beforeEach ->
+				@RedisManager.checkDocInMemory = sinon.stub().callsArgWith(2)
+				@DocumentManager.ensureDocIsLoaded @project_id, @doc_id, @callback
+
+			it "should check the doc in Redis", ->
+				@RedisManager.checkDocInMemory
+					.calledWith(@project_id, @doc_id)
+					.should.equal true
+
+			it "should call the callback", ->
+				@callback.called.should.equal true
+
+		describe "when the doc does not exist in Redis", ->
+			beforeEach ->
+				@RedisManager.getDoc = sinon.stub().callsArgWith(2, null, null, null, null, null)
+				@PersistenceManager.getDoc = sinon.stub().callsArgWith(2, null, @lines, @version, @ranges)
+				@RedisManager.putDocInMemory = sinon.stub().yields()
+				@RedisManager.checkDocInMemory = sinon.stub().callsArgWith(2, new Error("doc is not loaded"))
+				@DocumentManager.ensureDocIsLoaded @project_id, @doc_id, @callback
+
+			it "should try to get the doc from Redis", ->
+				@RedisManager.getDoc
+					.calledWith(@project_id, @doc_id)
+					.should.equal true
+
+			it "should get the doc from the PersistenceManager", ->
+				@PersistenceManager.getDoc
+					.calledWith(@project_id, @doc_id)
+					.should.equal true
+
+			it "should set the doc in Redis", ->
+				@RedisManager.putDocInMemory
+					.calledWith(@project_id, @doc_id, @lines, @version, @ranges)
+					.should.equal true
+
+			it "should call the callback", ->
+				@callback.calledWith(null).should.equal true
+
 	describe "setDoc", ->
 		describe "with plain tex lines", ->
 			beforeEach ->
@@ -363,4 +404,4 @@ describe "DocumentManager", ->
 			
 			it "should call the callback with a not found error", ->
 				error = new Errors.NotFoundError("document not found: #{@doc_id}")
-				@callback.calledWith(error).should.equal true
+				@callback.calledWith(error).should.equal true

services/document-updater/test/unit/coffee/UpdateManager/UpdateManagerTests.coffee

@@ -25,6 +25,7 @@ describe "UpdateManager", ->
 
 	describe "processOutstandingUpdates", ->
 		beforeEach ->
+			@DocumentManager.ensureDocIsLoaded = sinon.stub().callsArg(2)
 			@UpdateManager.fetchAndApplyUpdates = sinon.stub().callsArg(2)
 			@UpdateManager.processOutstandingUpdates @project_id, @doc_id, @callback
 
@@ -42,6 +43,7 @@ describe "UpdateManager", ->
 			beforeEach ->
 				@LockManager.tryLock = sinon.stub().callsArgWith(1, null, true, @lockValue = "mock-lock-value")
 				@LockManager.releaseLock = sinon.stub().callsArg(2)
+				@RedisManager.checkDocInMemory = sinon.stub().callsArg(2)
 				@UpdateManager.continueProcessingUpdatesWithLock = sinon.stub().callsArg(2)
 				@UpdateManager.processOutstandingUpdates = sinon.stub().callsArg(2)
 
@@ -90,7 +92,8 @@ describe "UpdateManager", ->
 
 			it "should not process the updates", ->
 				@UpdateManager.processOutstandingUpdates.called.should.equal false
-				
+
+
 	describe "continueProcessingUpdatesWithLock", ->
 		describe "when there are outstanding updates", ->
 			beforeEach ->