🐛 Reject presence whose version is ahead of the doc

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) {

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++) {

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;