WIP

Author: Joshua T Corbin

service-proxy.js

@@ -700,10 +700,8 @@ ServiceDispatchHandler.prototype.ensurePeerConnected =
 function ensurePeerConnected(serviceName, peer, reason, now) {
     var self = this;
 
-    if (self.partialAffinityEnabled) {
-        addIndexEntry(self.connectedServicePeers, serviceName, peer.hostPort, now);
-        addIndexEntry(self.connectedPeerServices, peer.hostPort, serviceName, now);
-    }
+    addIndexEntry(self.connectedServicePeers, serviceName, peer.hostPort, now);
+    addIndexEntry(self.connectedPeerServices, peer.hostPort, serviceName, now);
     delete self.peersToPrune[peer.hostPort];
 
     if (peer.isConnected('out') && !peer.draining) {
@@ -837,7 +835,7 @@ function addNewPartialPeer(serviceChannel, hostPort, now) {
         serviceChannel, serviceName, hostPort,
         'advertise', now);
 
-    if (result && result.noop) {
+    if (!result || result.noop) {
         // if ensurePartialConnections did no work, we need to freshen the
         // secondary indices since neither ensurePeerConnected nor
         // ensurePeerDisconnected were called for the advertising peer
@@ -943,8 +941,10 @@ function ensurePartialConnections(serviceChannel, serviceName, hostPort, reason,
                 reason: reason,
                 causingWorker: hostPort,
                 numToConnect: result.toConnect.length,
-                numToDisconnect: result.toDisconnect.length
-            }))
+                numToDisconnect: result.toDisconnect.length,
+                DEBUGToConnect: result.toConnect,
+                DEBUGToDisconnect: result.toDisconnect
+            })
         );
         result.implement();
     }
@@ -955,10 +955,8 @@ ServiceDispatchHandler.prototype.ensurePeerDisconnected =
 function ensurePeerDisconnected(serviceName, peer, reason, now) {
     var self = this;
 
-    if (self.partialAffinityEnabled) {
-        deleteIndexEntry(self.connectedServicePeers, serviceName, peer.hostPort);
-        deleteIndexEntry(self.connectedPeerServices, peer.hostPort, serviceName);
-    }
+    deleteIndexEntry(self.connectedServicePeers, serviceName, peer.hostPort);
+    deleteIndexEntry(self.connectedPeerServices, peer.hostPort, serviceName);
 
     var peerServices = self.connectedPeerServices[peer.hostPort];
     if (!peerServices || isObjectEmpty(peerServices)) {
@@ -984,26 +982,22 @@ function removeServicePeer(serviceName, hostPort) {
     if (!peer) {
         return;
     }
-    serviceChannel.peers.delete(hostPort);
 
     if (self.partialAffinityEnabled) {
         var partialRange = self.partialRanges[serviceName];
         if (partialRange) {
             partialRange.removeWorker(hostPort, now);
         }
-
-        var result = self.ensurePartialConnections(
+        self.ensurePartialConnections(
             serviceChannel, serviceName, hostPort,
             'unadvertise', now);
-        if (result && result.noop) {
-            // if ensurePartialConnections did no work, we need to celar the
-            // secondary indices since neither ensurePeerDisconnected was called
-            // for the unadvertising peer
-            deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
-            deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
-        }
     }
 
+    serviceChannel.peers.delete(hostPort);
+    deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
+    deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
+    deleteIndexEntry(self.knownPeers, hostPort, serviceName); // XXX missing before
+
     var subChanKeys = Object.keys(self.channel.subChannels);
     var remain = [];
     for (var i = 0; i < subChanKeys; i++) {
@@ -1012,6 +1006,7 @@ function removeServicePeer(serviceName, hostPort) {
             remain.push(subChanKeys[i]);
         }
     }
+    // TODO: remaiv vs connectedPeerServices
 
     if (remain.length) {
         self.logger.info(
@@ -1068,7 +1063,13 @@ function removeServicePeer(serviceName, hostPort) {
                 serviceName: serviceName
             })
         );
+
+        // TODO: why double delete
         self.channel.peers.delete(hostPort);
+        deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
+        deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
+        deleteIndexEntry(self.knownPeers, hostPort, serviceName);
+
     }
 };
 
@@ -1419,10 +1420,8 @@ function reapSinglePeer(hostPort, serviceNames, now) {
             serviceChannel.peers.delete(hostPort);
         }
 
-        if (self.partialAffinityEnabled) {
-            deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
-            deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
-        }
+        deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
+        deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
         deleteIndexEntry(self.knownPeers, hostPort, serviceName);
 
         var partialRange = self.partialRanges[serviceName];
@@ -1459,7 +1458,13 @@ function reapSinglePeer(hostPort, serviceNames, now) {
                 )
             );
         }
+
+        // TODO: why double delete
         self.channel.peers.delete(hostPort);
+        deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort);
+        deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName);
+        deleteIndexEntry(self.knownPeers, hostPort, serviceName);
+
     }
 };