From 7c1a081219fe7d903ded6cf69c380036d24c180f 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 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. 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 | 2 + lib/agent.js | 102 +++++++---- test/agent.js | 71 ++++++++ test/client/presence/doc-presence.js | 55 ++++++ test/client/presence/presence.js | 242 ++++++++++++++++++++++++++- 6 files changed, 436 insertions(+), 38 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 48e7f427c..d00e5a3c9 100644 --- a/docs/middleware/actions.md +++ b/docs/middleware/actions.md @@ -78,6 +78,8 @@ 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. If its client leaves while a presence is still in this middleware, the presence passes through it a second time with `p: null`. ## `'sendPresence'` diff --git a/lib/agent.js b/lib/agent.js index 9060741f1..b7b736490 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -54,6 +54,10 @@ function Agent(backend, stream) { this.latestDocVersionStreams = Object.create(null); this.latestDocVersions = Object.create(null); + this._streams = Object.create(null); + this._nextStreamKey = 1; + 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; @@ -94,20 +98,13 @@ 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.subscribedPresences = Object.create(null); // Clean up query subscription streams for (var id in this.subscribedQueries) { @@ -115,12 +112,15 @@ 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) { + var streams = this._streams; + var key = this._nextStreamKey++; + streams[key] = stream; + stream.once('close', function() { + delete streams[key]; + }); }; /** @@ -143,6 +143,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)); @@ -164,8 +165,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); @@ -174,22 +180,30 @@ 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]; + if (request.p === null) continue; + request.p = null; + this._broadcastPresence(request, function(error) { + if (error) logger.error('Error broadcasting disconnect presence', channel, error); + }); + } + delete this.presenceRequests[channel]; +}; + Agent.prototype._subscribeToQuery = function(emitter, queryId, collection, query) { var previous = this.subscribedQueries[queryId]; if (previous) previous.destroy(); @@ -820,25 +834,34 @@ Agent.prototype._src = function() { Agent.prototype._broadcastPresence = function(presence, callback) { var agent = this; var backend = this.backend; - var presenceRequests = this.presenceRequests; + var leaving = presence.p === null; + var pendingBroadcast = this._addPendingPresenceBroadcast(presence.ch); 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)); + if ((agent.closed || pendingBroadcast.retracted) && !leaving) { + presence.p = null; + return agent._broadcastPresence(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; + requests[presence.id] = presence; } var transformer = function(agent, presence, callback) { @@ -864,6 +887,17 @@ Agent.prototype._broadcastPresence = function(presence, callback) { }); }; +Agent.prototype._addPendingPresenceBroadcast = function(channel) { + var broadcast = {channel: channel, retracted: 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._subscribeDocVersion = function(collection, id, callback) { if (!collection || !id) return callback(); @@ -931,11 +965,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); @@ -948,6 +977,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 5ee3f25a9..2b2e8221a 100644 --- a/test/agent.js +++ b/test/agent.js @@ -7,6 +7,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; @@ -187,4 +191,71 @@ 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('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 f29d5e830..541982df8 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -792,6 +792,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'); @@ -993,6 +1008,46 @@ 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('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..79cab8715 100644 --- a/test/client/presence/presence.js +++ b/test/client/presence/presence.js @@ -718,6 +718,220 @@ 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('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 +946,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 +981,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); });