diff --git a/lib/agent.js b/lib/agent.js index 4d3d302d..efccc1cf 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -417,7 +417,7 @@ Agent.prototype._checkRequest = function(request) { } if (request.a === ACTIONS.op || request.a === ACTIONS.presence) { - if (request.v != null && (typeof request.v !== 'number' || request.v < 0)) return 'Invalid version'; + if (request.v != null && !util.isValidVersion(request.v)) return 'Invalid version'; } if (request.a === ACTIONS.presence) { @@ -427,6 +427,7 @@ Agent.prototype._checkRequest = function(request) { if (isDocPresence(request) && !matchesDocPresenceChannel(request)) { return 'Invalid presence channel'; } + if (isDocPresence(request) && request.v == null) return 'Invalid version'; } } else if ( request.a === ACTIONS.bulkFetch || @@ -840,6 +841,7 @@ Agent.prototype._broadcastPresence = function(presence, callback) { collection: presence.c }; var start = Date.now(); + var docVersionBeforeUpdate = util.dig(this.latestDocVersions, presence.c, presence.d); var subscriptionUpdater = presence.p === null ? this._unsubscribeDocVersion.bind(this) : @@ -852,26 +854,22 @@ Agent.prototype._broadcastPresence = function(presence, callback) { var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); var previousRequest = requests[presence.id]; if (!previousRequest || previousRequest.pv < presence.pv) { - presenceRequests[presence.ch][presence.id] = presence; + requests[presence.id] = presence; } - var transformer = function(agent, presence, callback) { - callback(null, presence); - }; - var latestDocVersion = util.dig(agent.latestDocVersions, presence.c, presence.d); - var presenceIsUpToDate = presence.v === latestDocVersion; - if (!presenceIsUpToDate) { - transformer = backend.transformPresenceToLatestVersion.bind(backend); - } + if (latestDocVersion == null) latestDocVersion = docVersionBeforeUpdate; - transformer(agent, presence, function(error, presence) { - if (error) return callback(error); - var channel = agent._getPresenceChannel(presence.ch); - agent.backend.pubsub.publish([channel], presence, function(error) { + backend._transformPresenceToLatestVersion(agent, presence, latestDocVersion, function(error, latestPresence) { + if (error) { + restorePreviousPresenceRequest(requests, presence, previousRequest); + return callback(error); + } + var channel = agent._getPresenceChannel(latestPresence.ch); + agent.backend.pubsub.publish([channel], latestPresence, function(error) { if (error) return callback(error); backend.emit('timing', 'presence.broadcast', Date.now() - start, context); - callback(null, presence); + callback(null, latestPresence); }); }); }); @@ -1025,6 +1023,12 @@ function createClientOp(request, clientId) { undefined; } +function restorePreviousPresenceRequest(requests, rejectedRequest, previousRequest) { + if (requests[rejectedRequest.id] !== rejectedRequest) return; + if (previousRequest) requests[rejectedRequest.id] = previousRequest; + else delete requests[rejectedRequest.id]; +} + function shallowCopy(object) { var out = {}; for (var key in object) { diff --git a/lib/backend.js b/lib/backend.js index 17a409df..cdda2173 100644 --- a/lib/backend.js +++ b/lib/backend.js @@ -936,8 +936,12 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca }; Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) { + this._transformPresenceToLatestVersion(agent, presence, null, callback); +}; + +Backend.prototype._transformPresenceToLatestVersion = function(agent, presence, trackedVersion, callback) { if (!presence.c || !presence.d) return callback(null, presence); - this.getOps(agent, presence.c, presence.d, presence.v, null, function(error, ops) { + this._getOpsSincePresence(agent, presence, trackedVersion, function(error, ops) { if (error) return callback(error); for (var i = 0; i < ops.length; i++) { var op = ops[i]; @@ -951,6 +955,45 @@ Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, c }); }; +Backend.prototype._getOpsSincePresence = function(agent, presence, trackedVersion, callback) { + if (trackedVersion == null) return this._getOpsSincePresenceFromSnapshotVersion(agent, presence, callback); + var backend = this; + this._getOpsSincePresenceFromVersion(agent, presence, trackedVersion, function(error, ops) { + if (error && presence.v > trackedVersion) { + return backend._getOpsSincePresenceFromSnapshotVersion(agent, presence, callback); + } + callback(error, ops); + }); +}; + +Backend.prototype._getOpsSincePresenceFromSnapshotVersion = function(agent, presence, callback) { + var backend = this; + var projection = this.projections[presence.c]; + var collection = (projection) ? projection.target : presence.c; + this.db.getSnapshot(collection, presence.d, {}, {agentCustom: agent.custom}, function(error, snapshot) { + if (error) return callback(error); + backend._getOpsSincePresenceFromVersion(agent, presence, snapshot.v, callback); + }); +}; + +Backend.prototype._getOpsSincePresenceFromVersion = function(agent, presence, knownVersion, callback) { + if (presence.v === knownVersion) return callback(null, []); + var from = Math.min(presence.v, knownVersion); + this.getOps(agent, presence.c, presence.d, from, null, function(error, ops) { + if (error) return callback(error); + var versionAfterOps = ops.length ? ops[ops.length - 1].v + 1 : from; + if (presence.v > versionAfterOps) { + return callback(new ShareDBError( + ERROR_CODE.ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT, + 'Presence version newer than current snapshot' + )); + } + callback(null, ops.filter(function(op) { + return op.v >= presence.v; + })); + }); +}; + function pluckIds(snapshots) { var ids = []; for (var i = 0; i < snapshots.length; i++) { diff --git a/test/agent.js b/test/agent.js index 5ee3f25a..eba05553 100644 --- a/test/agent.js +++ b/test/agent.js @@ -84,6 +84,27 @@ describe('Agent', function() { }); }); + describe('message validation', function() { + it('rejects an op with a non-integer version', function(done) { + var connection = backend.connect(); + connection.on('receive', function(request) { + var message = request.data; + if (message.c !== 'dogs' || message.d !== 'fido') return; + request.data = null; + expect(message.error).to.include({code: 'ERR_MESSAGE_BADLY_FORMED', message: 'Invalid version'}); + done(); + }); + connection.send({ + a: ACTIONS.op, + c: 'dogs', + d: 'fido', + v: 1.5, + seq: connection.seq++, + op: [{p: ['age'], na: 1}] + }); + }); + }); + describe('pub/sub channel collisions', function() { var writer; var subscriber; diff --git a/test/client/presence/doc-presence.js b/test/client/presence/doc-presence.js index e2d6ea45..b6bdccf7 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -585,6 +585,370 @@ describe('DocPresence', function() { connection1.send(message); }); + it('rejects a presence message with a non-integer version', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.v = 1.5; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message without a version', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + [ + {description: 'without doc coordinates', c: undefined, d: undefined}, + {description: 'for another doc', c: 'books', d: 'subtle-knife'} + ].forEach(function(test) { + it('ignores remote presence on its channel ' + test.description, function(done) { + sinon.spy(doc2, 'fetch'); + + [doc2.version, doc2.version + 1].forEach(function(version, index) { + var message = presence1.create('presence-' + index)._message(); + message.c = test.c; + message.d = test.d; + message.p = {index: 1}; + message.v = version; + message.t = presenceTestType.type.uri; + connection2.handleMessage(message); + }); + + doc2.whenNothingPending(function() { + expect(presence2.remotePresences).to.be.empty; + expect(presence2._remotePresenceInstances).to.be.empty; + expect(doc2.fetch).not.to.have.been.called; + done(); + }); + }); + }); + + [{index: 1}, null].forEach(function(value) { + it('rejects ' + JSON.stringify(value) + ' presence whose version is ahead of the doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + presence2.on('receive', function() { + done(new Error('Should not broadcast presence from the future')); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.on('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + expect(connection1.agent.presenceRequests[presence1.channel]).not.to.have.property('presence-1'); + next(); + }); + + var message = localPresence1._message(); + message.p = value; + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + }); + + it('rejects ' + JSON.stringify(value) + ' presence ahead of a tracked doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + localPresence1.submit.bind(localPresence1, {index: 1}), + function(next) { + localPresence1.on('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + + var message = localPresence1._message(); + message.p = value; + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + }); + }); + + it('broadcasts null presence on disconnect after rejecting presence ahead of the doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + + var message = localPresence1._message(); + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('keeps newer presence cached when an earlier presence ahead of the doc is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseGetOps; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + var getOps = backend.db.getOps; + sinon.stub(backend.db, 'getOps').callsFake(function() { + var args = arguments; + releaseGetOps = function() { + getOps.apply(backend.db, args); + }; + next(); + }); + + var message = localPresence1._message(); + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }, + function(next) { + backend.db.getOps.restore(); + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + releaseGetOps(); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('accepts up-to-date presence when the agent\'s tracked doc version lags behind', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + sinon.spy(backend.db, 'getSnapshot'); + localPresence1.submit({index: 12}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(doc2.version).to.eql(2); + expect(presence).to.eql({index: 12}); + expect(backend.db.getSnapshot).not.to.have.been.called; + next(); + }); + } + ], done); + }); + + it('accepts up-to-date presence when the agent\'s tracked doc version lags behind deleted ops', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + localPresence1.submit({index: 12}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 12}); + next(); + }); + } + ], done); + }); + + it( + 'transforms presence by surviving ops when the agent\'s tracked doc version lags behind deleted ops', + function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + function(next) { + doc2.submitOp({index: 0, value: 'The '}, errorHandler(done)); + doc1.once('op', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', 0, 2, null), + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 16}); + next(); + }); + + var message = localPresence1._message(); + message.p = {index: 12}; + message.v = 2; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + } + ); + + it('rejects presence ahead of the doc when transformed without a tracked version', function(done) { + var presence = { + c: 'books', + d: 'northern-lights', + p: {index: 1}, + v: doc1.version + 1, + t: presenceTestType.type.uri + }; + backend.transformPresenceToLatestVersion(connection1.agent, presence, function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + done(); + }); + }); + + it('broadcasts null presence on an up-to-date doc whose ops have been deleted', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + sinon.spy(backend.db, 'getSnapshot'); + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + expect(backend.db.getSnapshot).not.to.have.been.called; + next(); + }); + } + ], done); + }); + + it('broadcasts null presence from a second local presence on a doc whose ops have been deleted', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + sinon.spy(backend.db, 'getSnapshot'); + localPresence2.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-2'); + expect(presence).to.be.null; + expect(backend.db.getSnapshot).to.have.been.called; + next(); + }); + } + ], done); + }); + it('rejects a presence message without an ID', function(done) { var localPresence1 = presence1.create('presence-1'); // Have to catch the error on the Presence instance, because obviously