From af9c279fd94cf6baf470128134d3618faf7bd935 Mon Sep 17 00:00:00 2001 From: Alec Gibson <12036746+alecgibson@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:57:57 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Destroy=20every=20stream=20an=20?= =?UTF-8?q?`Agent`=20holds=20when=20it=20closes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes https://github.com/share/sharedb/issues/740 Fixes https://github.com/share/sharedb/issues/748 Fixes https://github.com/share/sharedb/issues/763 At the moment, `Agent._cleanup()` only destroys streams that are still in each subscription path's own map, so any bookkeeping slip in one of those paths leaves a stream subscribed after the client disconnects. This change tracks every stream handed to the agent in one map, and destroys them all on cleanup. So that presence streams can be replaced like doc streams, the `p: null` "client left" broadcast moves off the presence stream's `'end'` and into `_unsubscribePresence()` and `_cleanup()`. A broadcast still in middleware at that point goes out as `p: null` instead, and one that a newer presence or that leave has replaced by the time it would be published isn't sent at all. Any presence replaces a stored leave whatever its `pv`, because a presence re-created with the same id starts again at `pv` 0. That also fixes presence outliving its client when it was never subscribed, or still in flight, which `receivePresence` middleware now sees as `p: null`: 1. Client A submits presence on a channel it hasn't subscribed to 2. Client B, subscribed to the channel, receives it 3. Client A disconnects, and B keeps A's presence forever 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- docs/api/presence.md | 2 +- docs/middleware/actions.md | 4 + lib/agent.js | 168 ++++++--- test/agent.js | 124 +++++++ test/client/presence/doc-presence.js | 261 +++++++++++++ test/client/presence/presence.js | 536 ++++++++++++++++++++++++++- 6 files changed, 1048 insertions(+), 47 deletions(-) diff --git a/docs/api/presence.md b/docs/api/presence.md index d8768a7a9..f1ee656d4 100644 --- a/docs/api/presence.md +++ b/docs/api/presence.md @@ -54,7 +54,7 @@ Optional ### unsubscribe() -Unsubscribe from presence updates from remote clients. +Unsubscribe from presence updates from remote clients. If the channel was subscribed, or a subscription was pending, any presence this client has submitted on it is broadcast to remote subscribers as `null`. ```javascript presence.unsubscribe([callback]) diff --git a/docs/middleware/actions.md b/docs/middleware/actions.md index 3d34dc6a3..f44c3f424 100644 --- a/docs/middleware/actions.md +++ b/docs/middleware/actions.md @@ -78,6 +78,10 @@ This action has these additional `context` properties: `presence` -- Object > The presence object that was received. Its shape depends on its [type]({{ site.baseurl }}{% link types/index.md %}) +> +> `presence.p` is `null` when the presence leaves the channel: when it is destroyed, or when its client unsubscribes or disconnects. When its client unsubscribes or disconnects, the leave reuses the client's latest `presence` object for that ID, which passes through this middleware again with `p` set to `null` and any changes from its first pass kept, even if it is still being processed. Middleware with side effects should expect to see the same presence object twice. +> +> A presence that passes this middleware may still not be broadcast. If a newer presence with the same ID from the same client, or its leave, replaces it before it is published, its client gets an acknowledgement but nobody else receives it. ## `'sendPresence'` diff --git a/lib/agent.js b/lib/agent.js index 230928852..83a84263d 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -45,14 +45,26 @@ function Agent(backend, stream) { this.presenceSubscriptionSeq = Object.create(null); // Keep track of the last request that has been sent by each local presence // belonging to this agent. This is used to generate a new disconnection - // request if the client disconnects ungracefully. This is a - // map of channel -> id -> request + // request if the client disconnects ungracefully, and a presence that's + // been replaced here by the time it would be published isn't sent. This + // is a map of channel -> id -> request this.presenceRequests = Object.create(null); // Keep track of the latest known Doc version, so that we can avoid fetching // ops to transform presence if not needed this.latestDocVersionStreams = Object.create(null); this.latestDocVersions = Object.create(null); + // Every doc, doc version and presence stream this agent holds, so that + // _cleanup() destroys them even if they've gone missing from their own map. + // Map of key -> stream + this._streams = Object.create(null); + this._nextStreamKey = 1; + // Presence broadcasts still in their doc version lookup or receivePresence + // middleware. _retractPresences() flags them to go out as a leave, and a + // leave from the client that gets through middleware flags its own older + // ones to be dropped + this._pendingPresenceBroadcasts = []; + // We need to track this manually to make sure we don't reply to messages // after the stream was closed. this.closed = false; @@ -93,20 +105,16 @@ Agent.prototype._cleanup = function() { this.backend.agentsCount--; if (!this.stream.isServer) this.backend.remoteAgentsCount--; - // Clean up doc subscription streams - for (var collection in this.subscribedDocs) { - var docs = this.subscribedDocs[collection]; - for (var id in docs) { - var stream = docs[id]; - stream.destroy(); - } + for (var channel in this.presenceRequests) { + this._retractPresences(channel); } - this.subscribedDocs = Object.create(null); - for (var channel in this.subscribedPresences) { - this.subscribedPresences[channel].destroy(); + for (var key in this._streams) { + this._streams[key].destroy(); } + this.subscribedDocs = Object.create(null); this.subscribedPresences = Object.create(null); + this.latestDocVersionStreams = Object.create(null); // Clean up query subscription streams for (var id in this.subscribedQueries) { @@ -114,12 +122,16 @@ Agent.prototype._cleanup = function() { emitter.destroy(); } this.subscribedQueries = Object.create(null); +}; - for (var collection in this.latestDocVersionStreams) { - var streams = this.latestDocVersionStreams[collection]; - for (var id in streams) streams[id].destroy(); - } - this.latestDocVersionStreams = Object.create(null); +Agent.prototype._trackStream = function(stream) { + if (stream.open === false) return; + var streams = this._streams; + var key = this._nextStreamKey++; + streams[key] = stream; + stream.once('close', function() { + delete streams[key]; + }); }; /** @@ -142,6 +154,7 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) { Agent.prototype._subscribeMapToStream = function(map, collection, id, stream, dataHandler) { if (this.closed) return stream.destroy(); + this._trackStream(stream); var streams = map[collection] || (map[collection] = Object.create(null)); @@ -163,8 +176,13 @@ Agent.prototype._subscribeMapToStream = function(map, collection, id, stream, da Agent.prototype._subscribeToPresenceStream = function(channel, stream) { if (this.closed) return stream.destroy(); + this._trackStream(stream); var agent = this; + var previous = this.subscribedPresences[channel]; + if (previous) previous.destroy(); + this.subscribedPresences[channel] = stream; + stream.on('data', function(data) { if (data.error) { logger.error('Presence subscription stream error', channel, data.error); @@ -173,19 +191,34 @@ Agent.prototype._subscribeToPresenceStream = function(channel, stream) { }); stream.on('end', function() { - var requests = agent.presenceRequests[channel] || {}; - for (var id in requests) { - var request = agent.presenceRequests[channel][id]; - request.seq++; - request.p = null; - agent._broadcastPresence(request, function(error) { - if (error) logger.error('Error broadcasting disconnect presence', channel, error); - }); - } if (agent.subscribedPresences[channel] === stream) { delete agent.subscribedPresences[channel]; } - delete agent.presenceRequests[channel]; + }); +}; + +Agent.prototype._retractPresences = function(channel) { + var pendingBroadcasts = this._pendingPresenceBroadcasts; + for (var i = 0; i < pendingBroadcasts.length; i++) { + if (pendingBroadcasts[i].channel === channel) pendingBroadcasts[i].retracted = true; + } + + var requests = this.presenceRequests[channel]; + for (var id in requests) { + var request = requests[id]; + var alreadyLeft = request.p === null; + if (alreadyLeft) this._unsubscribeDocVersion(request.c, request.d, function() {}); + else this._broadcastLeave(request); + } + delete this.presenceRequests[channel]; +}; + +Agent.prototype._broadcastLeave = function(presence, callback) { + var agent = this; + presence.p = null; + this._broadcastPresence(presence, [], function(error, result) { + if (error) logger.error('Error broadcasting disconnect presence', presence.ch, presence.id, agent._src(), error); + if (callback) callback(error, result); }); }; @@ -522,7 +555,8 @@ Agent.prototype._handleMessage = function(request, callback) { message: 'Type does not support presence: ' + presence.t }); } - return this._broadcastPresence(presence, callback); + var olderBroadcasts = presence.p === null ? this._olderPendingPresenceBroadcasts(presence) : []; + return this._broadcastPresence(presence, olderBroadcasts, callback); case ACTIONS.presenceSubscribe: if (!this.backend.presenceEnabled) return; return this._subscribePresence(request.ch, request.seq, callback); @@ -843,28 +877,43 @@ Agent.prototype._src = function() { return this.src || this.clientId; }; -Agent.prototype._broadcastPresence = function(presence, callback) { +Agent.prototype._broadcastPresence = function(presence, supersededBroadcasts, callback) { var agent = this; var backend = this.backend; - var presenceRequests = this.presenceRequests; + var leaving = presence.p === null; + var pendingBroadcast = this._addPendingPresenceBroadcast(presence); var context = { presence: presence, collection: presence.c }; var start = Date.now(); - var subscriptionUpdater = presence.p === null ? + var subscriptionUpdater = leaving ? this._unsubscribeDocVersion.bind(this) : this._subscribeDocVersion.bind(this); subscriptionUpdater(presence.c, presence.d, function(error) { - if (error) return callback(error); + if (error) { + agent._removePendingPresenceBroadcast(pendingBroadcast); + return callback(error); + } backend.trigger(backend.MIDDLEWARE_ACTIONS.receivePresence, agent, context, function(error) { + agent._removePendingPresenceBroadcast(pendingBroadcast); if (error) return callback(error); - var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); + for (var i = 0; i < supersededBroadcasts.length; i++) { + supersededBroadcasts[i].superseded = true; + } + if (pendingBroadcast.superseded && !leaving) { + return agent._unsubscribeDocVersion(presence.c, presence.d, function() { + callback(null, presence); + }); + } + var clientHasLeft = agent.closed || pendingBroadcast.retracted; + if (clientHasLeft && !leaving) return agent._broadcastLeave(presence, callback); + var requests = agent.presenceRequests[presence.ch] || (agent.presenceRequests[presence.ch] = Object.create(null)); var previousRequest = requests[presence.id]; - if (!previousRequest || previousRequest.pv < presence.pv) { - presenceRequests[presence.ch][presence.id] = presence; + if (!previousRequest || previousRequest.p === null || previousRequest.pv < presence.pv) { + requests[presence.id] = presence; } var transformer = function(agent, presence, callback) { @@ -877,19 +926,52 @@ Agent.prototype._broadcastPresence = function(presence, callback) { transformer = backend.transformPresenceToLatestVersion.bind(backend); } - 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) { + transformer(agent, presence, function(error, transformedPresence) { + if (error) { + agent._withdrawPresenceRequest(requests, presence, previousRequest); + return callback(error); + } + var replaced = util.dig(agent.presenceRequests, presence.ch, presence.id) !== presence; + if (replaced && !leaving) return callback(null, transformedPresence); + 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); }); }); }); }); }; +Agent.prototype._withdrawPresenceRequest = function(requests, presence, previousRequest) { + if (requests[presence.id] !== presence) return; + if (previousRequest) requests[presence.id] = previousRequest; + else presence.p = null; +}; + +Agent.prototype._addPendingPresenceBroadcast = function(presence) { + var broadcast = {channel: presence.ch, id: presence.id, pv: presence.pv, retracted: false, superseded: false}; + this._pendingPresenceBroadcasts.push(broadcast); + return broadcast; +}; + +Agent.prototype._removePendingPresenceBroadcast = function(broadcast) { + var index = this._pendingPresenceBroadcasts.indexOf(broadcast); + if (index !== -1) this._pendingPresenceBroadcasts.splice(index, 1); +}; + +Agent.prototype._olderPendingPresenceBroadcasts = function(presence) { + var olderBroadcasts = []; + var pendingBroadcasts = this._pendingPresenceBroadcasts; + for (var i = 0; i < pendingBroadcasts.length; i++) { + var broadcast = pendingBroadcasts[i]; + var isOlder = broadcast.channel === presence.ch && broadcast.id === presence.id && broadcast.pv < presence.pv; + if (isOlder) olderBroadcasts.push(broadcast); + } + return olderBroadcasts; +}; + Agent.prototype._subscribeDocVersion = function(collection, id, callback) { if (!collection || !id) return callback(); @@ -957,11 +1039,6 @@ Agent.prototype._subscribePresence = function(channel, seq, cb) { return callback(); } agent.presenceSubscriptionSeq[channel] = seq; - if (agent.subscribedPresences[channel]) { - stream.destroy(); - return callback(); - } - agent.subscribedPresences[channel] = stream; agent._subscribeToPresenceStream(channel, stream); agent._requestPresence(channel, function(error) { callback(error); @@ -974,6 +1051,7 @@ Agent.prototype._unsubscribePresence = function(channel, seq, callback) { var stream = this.subscribedPresences[channel]; if (stream) stream.destroy(); delete this.subscribedPresences[channel]; + this._retractPresences(channel); callback(null, {ch: channel, seq: seq}); }; diff --git a/test/agent.js b/test/agent.js index b364b7df8..e2fb1fea8 100644 --- a/test/agent.js +++ b/test/agent.js @@ -8,6 +8,10 @@ var ACTIONS = require('../lib/message-actions').ACTIONS; var Connection = require('../lib/client/connection'); var protocol = require('../lib/protocol'); var LegacyConnection = require('sharedb-legacy/lib/client').Connection; +var async = require('async'); +var types = require('../lib/types'); +var presenceTestType = require('./client/presence/presence-test-type'); +types.register(presenceTestType.type); describe('Agent', function() { var backend; @@ -323,4 +327,124 @@ describe('Agent', function() { }); }); }); + + describe('cleanup', function() { + var connection; + + beforeEach(function(done) { + backend.close(function(error) { + if (error) return done(error); + backend = new Backend({presence: true}); + connection = backend.connect(); + connection.get('books', 'northern-lights').create('North Lights', presenceTestType.type.name, done); + }); + }); + + function closeConnection(callback) { + connection.agent.stream.once('end', callback); + connection.close(); + } + + function opensStreams(count, step) { + return function(next) { + var streamsCount = backend.pubsub.streamsCount; + step(function(error) { + if (error) return next(error); + expect(backend.pubsub.streamsCount - streamsCount).to.equal(count); + next(); + }); + }; + } + + it('destroys a stream that is missing from its subscription map', function(done) { + connection.get('dogs', 'fido').subscribe(function(error) { + if (error) return done(error); + delete connection.agent.subscribedDocs.dogs; + closeConnection(function() { + expect(backend.pubsub.streamsCount).to.equal(0); + done(); + }); + }); + }); + + it('stops tracking streams once they close', function(done) { + var doc = connection.get('dogs', 'fido'); + var presence = connection.getDocPresence('books', 'northern-lights'); + var localPresence = presence.create('presence-1'); + async.series([ + doc.subscribe.bind(doc), + doc.unsubscribe.bind(doc), + presence.subscribe.bind(presence), + presence.unsubscribe.bind(presence), + localPresence.submit.bind(localPresence, {index: 0}), + localPresence.destroy.bind(localPresence), + function(next) { + expect(connection.agent._streams).to.be.empty; + next(); + } + ], done); + }); + + it('empties its stream maps as soon as the connection closes', function(done) { + var presence = connection.getDocPresence('books', 'northern-lights'); + var agent = connection.agent; + async.series([ + function(next) { + connection.get('dogs', 'fido').subscribe(next); + }, + presence.subscribe.bind(presence), + function(next) { + presence.create('presence-1').submit({index: 0}, next); + }, + function(next) { + agent.stream.once('end', function() { + expect(agent.subscribedDocs).to.be.empty; + expect(agent.subscribedPresences).to.be.empty; + expect(agent.latestDocVersionStreams).to.be.empty; + next(); + }); + connection.close(); + } + ], done); + }); + + it('does not track a stream that closed before reaching the agent', function(done) { + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, next) { + backend.pubsub.close(); + next(); + }); + connection.get('dogs', 'fido').subscribe(function(error) { + if (error) return done(error); + expect(connection.agent._streams).to.be.empty; + done(); + }); + }); + + it('destroys every pub/sub stream when the connection closes', function(done) { + var presence = connection.getDocPresence('books', 'northern-lights'); + async.series([ + opensStreams(1, function(next) { + connection.get('dogs', 'fido').subscribe(next); + }), + opensStreams(2, function(next) { + connection.startBulk(); + connection.get('dogs', 'rex').subscribe(); + connection.get('dogs', 'spot').subscribe(next); + connection.endBulk(); + }), + opensStreams(1, presence.subscribe.bind(presence)), + opensStreams(1, function(next) { + presence.create('presence-1').submit({index: 0}, next); + }), + opensStreams(1, function(next) { + connection.createSubscribeQuery('dogs', {}, null, next); + }), + closeConnection, + function(next) { + expect(backend.pubsub.streamsCount).to.equal(0); + next(); + } + ], done); + }); + }); }); diff --git a/test/client/presence/doc-presence.js b/test/client/presence/doc-presence.js index 5cb4b5ef7..7b9b3828a 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -793,6 +793,21 @@ describe('DocPresence', function() { ], done); }); + it('does not keep presence when fetching its doc version fails', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, next) { + next(new Error('denied')); + }); + + localPresence1.submit({index: 1}, function(error) { + expect(error.message).to.contain('denied'); + expect(connection1.agent.presenceRequests).to.be.empty; + expect(connection1.agent._pendingPresenceBroadcasts).to.be.empty; + done(); + }); + }); + it('errors when trying to submit presence on a type that does not support it', function(done) { var jsonDoc = connection1.get('books', 'snuff'); var jsonPresence = connection1.getDocPresence('books', 'snuff'); @@ -1047,6 +1062,252 @@ describe('DocPresence', function() { ], done); }); + it('releases the doc version stream of presence in flight when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var streamsCount; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + next(); + }); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + streamsCount = backend.pubsub.streamsCount; + next(); + }, + presence1.subscribe.bind(presence1), + function(next) { + async.parallel([ + function(cb) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + cb(); + }); + }, + function(cb) { + localPresence1.submit({index: 1}, cb); + } + ], next); + }, + function(next) { + expect(backend.pubsub.streamsCount).to.equal(streamsCount); + next(); + } + ], done); + }); + + it('releases the doc version stream of presence transformed across a delete when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence1.subscribe.bind(presence1), + doc1.unsubscribe.bind(doc1), + doc2.del.bind(doc2), + localPresence1.submit.bind(localPresence1, {index: 1}), + function(next) { + expect(connection1.agent.presenceRequests['books.northern-lights']['presence-1'].p).to.be.null; + expect(connection1.agent.latestDocVersionStreams).not.to.be.empty; + presence1.unsubscribe(next); + }, + function(next) { + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + next(); + } + ], done); + }); + + it('releases the doc version stream of presence that failed to transform when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + localPresence1.submit({badProp: 'foo'}, function(error) { + expect(error.code).to.equal('ERR_PRESENCE_TRANSFORM_FAILED'); + expect(connection1.agent.latestDocVersionStreams).not.to.be.empty; + presence1.unsubscribe(next); + }); + }, + function(next) { + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + } + ], done); + }); + + it('does not broadcast an older presence that finishes transforming after its client unsubscribes', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releaseFirstTransform; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (releaseFirstTransform || context.agent !== connection1.agent) return callback(); + releaseFirstTransform = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + presence1.unsubscribe(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + releaseFirstTransform(); + localPresence2.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.have.all.keys('presence-2'); + next(); + } + ], done); + }); + + it('broadcasts an older presence still transforming when a newer one fails to transform', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseFirstTransform; + + async.series([ + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (releaseFirstTransform || context.agent !== connection1.agent) return callback(); + releaseFirstTransform = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.submit({badProp: 'foo'}, function(error) { + expect(error.code).to.equal('ERR_PRESENCE_TRANSFORM_FAILED'); + next(); + }); + }, + function(next) { + releaseFirstTransform(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + } + ], done); + }); + + it('broadcasts a null presence still being transformed when the connection closes', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (context.agent !== connection1.agent) return callback(); + connection1.agent.stream.once('end', function() { + callback(); + }); + connection1.close(); + }); + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('does not broadcast presence still looking up its doc version when it is destroyed', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releaseDocVersion; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, callback) { + if (releaseDocVersion || context.agent !== connection1.agent) return callback(); + releaseDocVersion = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.destroy(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + connection1.once('receive', function(request) { + expect(request.data).to.include({id: 'presence-1', pv: 0}); + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + next(); + }); + releaseDocVersion(); + }, + function(next) { + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.have.all.keys('presence-2'); + next(); + } + ], done); + }); + it('does not error when destroying presence for a deleted doc', function(done) { var localPresence1 = presence1.create('presence-1'); diff --git a/test/client/presence/presence.js b/test/client/presence/presence.js index f972327e2..cd7d321e5 100644 --- a/test/client/presence/presence.js +++ b/test/client/presence/presence.js @@ -1,4 +1,5 @@ var Backend = require('../../../lib/backend'); +var logger = require('../../../lib/logger'); var PresencePauser = require('./presence-pauser'); var expect = require('chai').expect; var async = require('async'); @@ -718,6 +719,513 @@ describe('Presence', function() { ], done); }); + it('broadcasts a null presence when a connection that never subscribed is disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('broadcasts a null presence for presence in flight when the connection is disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + connection1.agent.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + localPresence1.submit({index: 3}, errorHandler(done)); + } + ], done); + }); + + it('broadcasts a null presence for presence in flight when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + next(); + }); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + async.parallel([ + function(cb) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + cb(); + }); + }, + function(cb) { + localPresence1.submit({index: 3}, cb); + } + ], next); + } + ], done); + }); + + it('broadcasts presence submitted after unsubscribing while earlier presence was in flight', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var received = {}; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.id !== 'presence-1' || context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + localPresence2.submit({index: 2}, function(error) { + if (error) return done(error); + next(); + }); + }); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + presence2.on('receive', function(id, presence) { + received[id] = presence; + if (Object.keys(received).length === 2) next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + expect(received).to.eql({'presence-1': null, 'presence-2': {index: 2}}); + next(); + } + ], done); + }); + + it('broadcasts a null presence for presence held in receive middleware when disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receive, function(context, next) { + if (context.agent !== connection1.agent || context.data.a !== 'p') return next(); + connection1.agent.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + localPresence1.submit({index: 3}, errorHandler(done)); + } + ], done); + }); + + it('does not rewrite a null presence again when middleware fills it in on disconnect', function(done) { + var localPresence1 = presence1.create('presence-1'); + var nullPresenceCount = 0; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, callback) { + if (context.presence.p !== null) return callback(); + if (++nullPresenceCount > 1) return done(new Error('Rewrote a null presence more than once')); + context.presence.p = {index: 0}; + callback(); + }); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-1'); + expect(nullPresenceCount).to.equal(1); + next(); + }); + connection1.close(); + } + ], done); + }); + + it('logs an error when the leave for presence in flight fails', function(done) { + var localPresence1 = presence1.create('presence-1'); + var agent1 = connection1.agent; + + sinon.stub(logger, 'error').callsFake(function(message, channel, id, src, error) { + expect(message).to.equal('Error broadcasting disconnect presence'); + expect([channel, id, src]).to.eql(['test-channel', 'presence-1', agent1._src()]); + expect(error.message).to.equal('bad!'); + done(); + }); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(new Error('bad!')); + agent1.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + localPresence1.submit({index: 3}, errorHandler(done)); + }); + + it('does not broadcast presence that gets through middleware after its own leave', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + var isFirstPresence = context.presence.id === 'presence-1' && context.presence.pv === 0; + if (!isFirstPresence || releasePresence) return next(); + releasePresence = next; + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + releasePresence(); + localPresence2.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.eql({'presence-2': {index: 3}}); + next(); + } + ], done); + }); + + it('broadcasts presence in flight when its leave is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(new Error('bad!')); + releasePresence = next; + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + localPresence1.submit(null, function(error) { + expect(error.message).to.contain('bad!'); + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence that overtakes its older leave in receive middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseLeave; + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receive, function(context, next) { + var isLeave = context.agent === connection1.agent && context.data.a === 'p' && context.data.p === null; + if (!isLeave) return next(); + releaseLeave = next; + }); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + releasePresence = next; + releaseLeave(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + localPresence1.submit(null, errorHandler(done)); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence re-created with the same id while its old leave is in middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseLeave; + var releasePresence; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, callback) { + if (releaseLeave && releasePresence) return callback(); + if (context.presence.p === null) releaseLeave = callback; + else releasePresence = callback; + if (releaseLeave && releasePresence) next(); + }); + localPresence1.submit(null); + presence1.create('presence-1').submit({index: 2}, errorHandler(done)); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + releaseLeave(); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence submitted after unsubscribing while an older one is still in middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var heldPresences = []; + var onHeld; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + heldPresences.push(next); + onHeld(); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + onHeld = next; + localPresence1.submit({index: 1}, errorHandler(done)); + }, + presence1.unsubscribe.bind(presence1), + function(next) { + onHeld = next; + localPresence1.submit({index: 2}, errorHandler(done)); + }, + function(next) { + heldPresences[0](); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + heldPresences[1](); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + } + ], done); + }); + + it('broadcasts presence re-created with the same id, and its leave on disconnect', 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) { + localPresence1.destroy(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + presence1.create('presence-1').submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + connection1.close(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('broadcasts presence re-created with the same id after destroying the channel presence', function(done) { + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + presence1.create('presence-1').submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + var leaveCount = 0; + presence1.destroy(errorHandler(done)); + presence2.on('receive', function receiveLeave(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + leaveCount++; + if (leaveCount < 2) return; + presence2.off('receive', receiveLeave); + next(); + }); + }, + function(next) { + connection1.getPresence('test-channel').create('presence-1').submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + connection1.close(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('broadcasts a null presence when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + presence1.unsubscribe(errorHandler(done)); + } + ], done); + }); + + it('does not broadcast a null presence again when disconnecting after unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var agent1 = connection1.agent; + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function() { + next(); + }); + presence1.unsubscribe(errorHandler(done)); + }, + function(next) { + sinon.spy(agent1, '_broadcastPresence'); + agent1.stream.once('end', next); + connection1.close(); + }, + function(next) { + expect(agent1._broadcastPresence).not.to.have.been.called; + next(); + } + ], done); + }); + describe('middleware', function() { describe('receivePresence', function() { it('provides the presence in the middleware', function(done) { @@ -732,7 +1240,7 @@ describe('Presence', function() { it('can mutate the presence in the middleware', function(done) { backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { - context.presence.p.index++; + if (context.presence.p) context.presence.p.index++; next(); }); @@ -767,8 +1275,34 @@ describe('Presence', function() { localPresence1.submit({index: 5}, function(error) { expect(error.message).to.contain('bad!'); expect(connection1.agent.presenceRequests['test-channel']).not.to.be.ok; + expect(connection1.agent._pendingPresenceBroadcasts).to.be.empty; + next(); + }); + } + ], done); + }); + + it('still broadcasts presence in flight when another presence on the channel is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.id === 'presence-2') return next(new Error('bad!')); + localPresence2.submit({index: 2}, function(error) { + expect(error.message).to.contain('bad!'); + next(); + }); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); next(); }); + localPresence1.submit({index: 1}, errorHandler(done)); } ], done); });