From 0a26a7464b4efcd68e99c90d587f92e051f39fe2 Mon Sep 17 00:00:00 2001 From: Alec Gibson <12036746+alecgibson@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:28:43 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Reject=20presence=20whose=20vers?= =?UTF-8?q?ion=20is=20ahead=20of=20the=20doc?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes https://github.com/share/sharedb/issues/735 At the moment, the server relays a doc presence whose `v` is ahead of the doc unchanged, which lets one client make every subscriber fetch: 1. Submit presence with `v: 1e9` to a doc with 20 subscribers 2. Each subscriber calls `doc.fetch()`, since `v > doc.version` 3. The presence stays pending, so every later op fetches again This change has the `Agent` pass its tracked doc version through to the presence transform, which reads ops from the lower of that and `v`. The tracker can lag the database but never lead it, so a `v` past the last op read is ahead of the doc. Ops in that lag may have been deleted, though, so we check the snapshot version before rejecting. Doc presence must now carry an integer `v`, and so must ops that send one. A rejected presence restores the `Agent`'s previously cached request, so the leave it replays on disconnect isn't rejected too. Presence sent without `c`/`d` skips all of this, since the server can't tell it from generic presence, but clients ignore it since #752. We deliberately don't bound presence that is far *behind*: a `fetch` with `v: 0` already makes the server read the whole op history. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- lib/agent.js | 34 +-- lib/backend.js | 45 +++- test/agent.js | 21 ++ test/client/presence/doc-presence.js | 364 +++++++++++++++++++++++++++ 4 files changed, 448 insertions(+), 16 deletions(-) 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