storage: fix crash when a client tries to "steal" the UUID of another client

neo/lib/node.py

@@ -409,6 +409,7 @@ class NodeManager(object):
             # XXX: It's probably a bug to include connecting nodes but there's
             #      no API yet to update manager when connection is established.
             if node.isConnected(connecting=True):
+                assert node in self._node_set, node
                 self._identified_dict[uuid] = node
             else:
                 self._identified_dict.pop(uuid, None)

neo/storage/handlers/identification.py

@@ -52,9 +52,11 @@ class IdentificationHandler(EventHandler):
                 if node is None:
                     node = app.nm.createClient(uuid=uuid)
                 elif node.isConnected():
-                    # cut previous connection
-                    node.getConnection().close()
-                    assert not node.isConnected()
+                    # This can happen if we haven't processed yet a notification
+                    # from the master, telling us the existing node is not
+                    # running anymore. If we accept the new client, we won't
+                    # know what to do with this late notification.
+                    raise NotReadyError('uuid conflict: retry later')
                 node.setRunning()
             elif node_type == NodeTypes.STORAGE:
                 if node is None:

neo/tests/threaded/test.py

@@ -800,5 +800,32 @@ class Test(NEOThreadedTest):
         finally:
             cluster.stop()
 
+    def testRecycledClientUUID(self):
+        def delayNotifyInformation(conn, packet):
+            return isinstance(packet, Packets.NotifyNodeInformation)
+        def notReady(orig, *args):
+            m2s.remove(delayNotifyInformation)
+            return orig(*args)
+        cluster = NEOCluster()
+        try:
+            cluster.start()
+            cluster.getTransaction()
+            with cluster.master.filterConnection(cluster.storage) as m2s:
+                m2s.add(delayNotifyInformation)
+                cluster.client.master_conn.close()
+                cluster.client.setPoll(0)
+                client = ClientApplication(name=cluster.name,
+                                           master_nodes=cluster.master_nodes)
+                p = Patch(client.storage_bootstrap_handler, notReady=notReady)
+                try:
+                    client.setPoll(1)
+                    x = client.load(ZERO_TID)
+                finally:
+                    del p
+                    client.close()
+                self.assertNotIn(delayNotifyInformation, m2s)
+        finally:
+            cluster.stop()
+
 if __name__ == "__main__":
     unittest.main()