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

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,33 +105,33 @@ 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) {
     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) {
+  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});
 };