Skip to content

馃悰 Reject presence whose version is ahead of the doc #749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
alecgibson wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

馃悰 Reject presence whose version is ahead of the doc #749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 18 additions & 15 deletions lib/agent.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -416,7 +416,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) {
Expand Down Expand Up @@ -826,6 +826,7 @@ Agent.prototype._broadcastPresence = function(presence, callback) {
collection: presence.c
};
var start = Date.now();
var previousDocVersion = util.dig(this.latestDocVersions, presence.c, presence.d);

var subscriptionUpdater = presence.p === null ?
this._unsubscribeDocVersion.bind(this) :
Expand All @@ -838,26 +839,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 = previousDocVersion;

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, transformedPresence) {
if (error) {
restorePreviousPresenceRequest(requests, presence, previousRequest);
return callback(error);
Comment on lines +849 to +851
}
var channel = agent._getPresenceChannel(transformedPresence.ch);
agent.backend.pubsub.publish([channel], transformedPresence, function(error) {
if (error) return callback(error);
backend.emit('timing', 'presence.broadcast', Date.now() - start, context);
callback(null, presence);
callback(null, transformedPresence);
});
});
});
Expand Down Expand Up @@ -1011,6 +1008,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) {
Expand Down
25 changes: 23 additions & 2 deletions lib/backend.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -932,12 +932,33 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca
callback(error, snapshot);
};

Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) {
Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, knownVersion, callback) {
if (!presence.c || !presence.d) return callback(null, presence);
this.getOps(agent, presence.c, presence.d, presence.v, null, function(error, ops) {
if (knownVersion != null) return this._transformPresenceFromVersion(agent, presence, knownVersion, 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._transformPresenceFromVersion(agent, presence, snapshot.v, callback);
});
};

Backend.prototype._transformPresenceFromVersion = function(agent, presence, knownVersion, callback) {
if (presence.v === knownVersion) return callback(null, presence);
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'
));
}
for (var i = 0; i < ops.length; i++) {
var op = ops[i];
if (op.v < presence.v) continue;
Comment thread
alecgibson marked this conversation as resolved.
var isOwnOp = op.src === presence.src;
var transformError = ot.transformPresence(presence, op, isOwnOp);
if (transformError) {
Expand Down
21 changes: 21 additions & 0 deletions test/agent.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
238 changes: 238 additions & 0 deletions test/client/presence/doc-presence.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -584,6 +584,244 @@ 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);
});

[{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 without a snapshot read', function(done) {
var localPresence1 = presence1.create('presence-1');

async.series([
localPresence1.submit.bind(localPresence1, {index: 1}),
function(next) {
sinon.spy(backend.db, 'getSnapshot');
localPresence1.on('error', function(error) {
expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT');
expect(backend.db.getSnapshot).not.to.have.been.called;
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;
localPresence1.submit({index: 12}, errorHandler(done));
presence2.once('receive', function(id, presence) {
expect(doc2.version).to.eql(2);
expect(presence).to.eql({index: 12});
next();
});
}
], 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) {
localPresence1.submit(null, errorHandler(done));
presence2.once('receive', function(id, presence) {
expect(presence).to.be.null;
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
Expand Down
Loading