🐛 Destroy every stream an Agent holds when it closes

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

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'`
 

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,33 +98,29 @@ 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) {
     var emitter = this.subscribedQueries[id];
     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});
 };
 

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);
+    });
+  });
 });

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');