🐛 Broadcast a leave before presence re-created with its ID

docs/middleware/actions.md

@@ -82,7 +82,7 @@ This action has these additional `context` properties:
 
 > 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.
+> `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 is a shallow copy of the client's latest `presence` object for that ID, which passes through this middleware with `p` set to `null` and any changes from its first pass kept, even if the original is still being processed. Middleware with side effects should expect to see that presence twice, as two different objects.
 >
 > 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.
 

lib/agent.js

@@ -66,6 +66,11 @@ function Agent(backend, stream) {
   // leave from the client that gets through middleware flags its own older
   // ones to be dropped
   this._pendingPresenceBroadcasts = [];
+  // The newest leave of each presence that's through middleware but not yet
+  // handed to pubsub. Later presence with its ID waits for it, since
+  // subscribers ignore a re-created presence's restarted presence version
+  // until they've applied the leave. Map of channel -> id -> leave
+  this._leavesInFlight = Object.create(null);
 
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
@@ -217,9 +222,10 @@ Agent.prototype._retractPresences = function(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);
+  var leave = shallowCopy(presence);
+  leave.p = null;
+  this._broadcastPresence(leave, [], function(error, result) {
+    if (error) logger.error('Error broadcasting disconnect presence', leave.ch, leave.id, agent._src(), error);
     if (callback) callback(error, result);
   });
 };
@@ -944,34 +950,56 @@ Agent.prototype._broadcastPresence = function(presence, supersededBroadcasts, ca
         requests[presence.id] = presence;
       }
 
-      var transformer = function(agent, presence, callback) {
-        callback(null, presence);
-      };
-
-      var latestDocVersion = util.dig(agent.latestDocVersions, presence.c, presence.d);
-      var presenceIsUpToDate = presence.v === latestDocVersion;
-      if (!presenceIsUpToDate) {
-        transformer = backend.transformPresenceToLatestVersion.bind(backend);
-      }
-
-      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, transformedPresence);
+      var previousLeave = util.dig(agent._leavesInFlight, presence.ch, presence.id);
+      var leaveInFlight = leaving ? agent._addLeaveInFlight(presence) : null;
+      agent._afterLeaveHandedOver(previousLeave, function() {
+        agent._bringPresenceUpToDate(presence, function(error, transformedPresence) {
+          if (error) {
+            agent._withdrawPresenceRequest(requests, presence, previousRequest);
+            agent._handOverLeaveInFlight(presence, leaveInFlight);
+            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, transformedPresence);
+          });
+          agent._handOverLeaveInFlight(presence, leaveInFlight);
         });
       });
     });
   });
 };
 
+Agent.prototype._bringPresenceUpToDate = function(presence, callback) {
+  var latestDocVersion = util.dig(this.latestDocVersions, presence.c, presence.d);
+  if (presence.v === latestDocVersion) return callback(null, presence);
+  this.backend.transformPresenceToLatestVersion(this, presence, callback);
+};
+
+Agent.prototype._addLeaveInFlight = function(presence) {
+  var leave = {waiters: []};
+  var leaves = this._leavesInFlight[presence.ch] || (this._leavesInFlight[presence.ch] = Object.create(null));
+  leaves[presence.id] = leave;
+  return leave;
+};
+
+Agent.prototype._afterLeaveHandedOver = function(leave, callback) {
+  if (leave) leave.waiters.push(callback);
+  else callback();
+};
+
+Agent.prototype._handOverLeaveInFlight = function(presence, leave) {
+  if (!leave) return;
+  if (util.dig(this._leavesInFlight, presence.ch, presence.id) === leave) {
+    util.digAndRemove(this._leavesInFlight, presence.ch, presence.id);
+  }
+  for (var i = 0; i < leave.waiters.length; i++) util.nextTick(leave.waiters[i]);
+};
+
 Agent.prototype._withdrawPresenceRequest = function(requests, presence, previousRequest) {
   if (requests[presence.id] !== presence) return;
   if (previousRequest) requests[presence.id] = previousRequest;