Merge lp:~chipaca/ubuntuone-client/multicon into lp:ubuntuone-client

[John Lenton]

detect and avoid multiple connections that break the single-minded state machine

[John Lenton (chipaca)]

This should avoid the multiple simultaneous live connections issue that is breaking things for a lot of people.

[lp:~chipaca/ubuntuone-client/multicon]

202. By John Lenton on 2009-09-01

moved server rescan to action queue too, for symmetry

[Nicola Larosa (teknico)]

Code looks good, tests pass.

review: Approve

[dobey (dobey)]

review: Approve

[Rick McBride (rmcbride)]

Looks good

review: Approve

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2009-08-31 15:41:20 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2009-09-01 18:51:00 +0000
@@ -28,6 +28,7 @@
 import os
 import random
 import tempfile
+import traceback
 import zlib
 
 from zope.interface import implements
@@ -213,7 +214,7 @@
         might not be described in failure).
         """
         logger.warning('connection lost: %s' % reason.getErrorMessage())
-        self.factory.event_queue.push('SYS_CONNECTION_LOST')
+        self.factory.connectionLost(self, reason)
         LoggingStorageClient.connectionLost(self, reason)
 
 
@@ -721,6 +722,14 @@
 
         event_queue.subscribe(self)
 
+    def connectionLost(self, protocol, reason):
+        """
+        The connection went down, for some reason (which might or
+        might not be described in reason).
+        """
+        if protocol is self.client:
+            self.event_queue.push('SYS_CONNECTION_LOST')
+
     def check_conditions(self):
         """Poll conditions on which running actions may be waiting."""
         self.content_queue.check_conditions()
@@ -837,6 +846,7 @@
         """
         Start the circus going.
         """
+        self.client = None
         self.deferred = defer.Deferred()
         d = self._lookup_srv()
         def _connect(result):
@@ -850,12 +860,116 @@
         d.addCallback(_connect)
         return self.deferred
 
+    def disconnect(self):
+        """
+        Shut down the client, if it isn't already.
+        """
+        if self.client is not None:
+            self.client.disconnect()
+            self.client = None
+
     def conectionFailed(self, reason=None):
         """
         Called when the connect() call fails
         """
         self.deferred.errback(reason)
 
+    @defer.inlineCallbacks
+    def check_version(self):
+        """
+        Check the client protocol version matches that of the
+        server. Call callback on success, errback on failure.
+        """
+        # if the client changes while we're waiting, this message is
+        # old news and should be discarded (the message would
+        # typically be a failure: timeout or disconnect). So keep the
+        # original client around for comparison.
+        client = self.client
+        try:
+            yield client.protocol_version()
+        except Exception, e:
+            if client is self.client:
+                logger.error("Protocol version error")
+                logger.debug('traceback follows:\n\n' + traceback.format_exc())
+                if e.message == 'UNSUPPORTED_VERSION':
+                    self.event_queue.push('SYS_PROTOCOL_VERSION_ERROR',
+                                          error=e.message)
+                else:
+                    self.event_queue.push('SYS_UNKNOWN_ERROR')
+                # it looks like we won't be authenticating, so hook up the
+                # for-testing deferred now
+                self.deferred.callback(Failure(e))
+        else:
+            if client is self.client:
+                logger.info("Protocol version OK")
+                self.event_queue.push('SYS_PROTOCOL_VERSION_OK')
+
+    @defer.inlineCallbacks
+    def set_capabilities(self, caps):
+        """Set the capabilities with the server"""
+        client = self.client
+        is_failed = None
+        try:
+            req = (yield client.query_caps(caps))
+        except Exception, e:
+            is_failed = e
+        else:
+            if not req.accepted:
+                is_failed = StandardError("The server doesn't have"
+                                          " the requested capabilities")
+
+        if client is self.client:
+            if is_failed is not None:
+                logger.error("Capabilities query failed: %s" % is_failed)
+                # Push the error to the event queue,
+                self.event_queue.push('SYS_SET_CAPABILITIES_ERROR',
+                                      error=is_failed.message)
+                # it looks like we won't be authenticating, so hook up the
+                # for-testing deferred now
+                self.deferred.callback(Failure(is_failed))
+                return
+
+            is_failed = None
+            try:
+                req = (yield client.set_caps(caps))
+            except Exception, e:
+                if client is self.client:
+                    is_failed = e
+            else:
+                if client is self.client:
+                    if req.accepted:
+                        self.event_queue.push('SYS_SET_CAPABILITIES_OK')
+                        defer.returnValue(self.client)
+                    else:
+                        is_failed = StandardError("The server denied setting"
+                                                  " %s capabilities" % req.caps)
+            if is_failed is not None:
+                logger.error("Capabilities set failed: %s" % is_failed)
+                # Push the error to the event queue,
+                self.event_queue.push('SYS_SET_CAPABILITIES_ERROR',
+                                      error=is_failed.message)
+                # it looks like we won't be authenticating, so hook up the
+                # for-testing deferred now
+                self.deferred.callback(Failure(is_failed))
+
+    @defer.inlineCallbacks
+    def authenticate(self, oauth_consumer):
+        client = self.client
+        try:
+            yield client.oauth_authenticate(oauth_consumer, self.token)
+        except Exception, e:
+            if client is not self.client:
+                return
+            logger.error("OAuth failed: %s", e.message)
+            self.event_queue.push('SYS_OAUTH_ERROR', error=e.message)
+            self.deferred.callback(Failure(e))
+        else:
+            if client is not self.client:
+                return
+            logger.info("Oauth OK")
+            self.event_queue.push('SYS_OAUTH_OK')
+            self.deferred.callback(client)
+
     def get_root(self, marker):
         """
         Get the user's root uuid. Use the uuid_map, so the caller can

=== modified file 'ubuntuone/syncdaemon/main.py'
--- ubuntuone/syncdaemon/main.py	2009-08-27 18:44:53 +0000
+++ ubuntuone/syncdaemon/main.py	2009-09-01 18:41:31 +0000
@@ -252,106 +252,28 @@
         """
         Check the client protocol version matches that of the server.
         """
-        d = self.action_q.client.protocol_version()
-        def protocol_callback(_):
-            """protocol check was OK"""
-            self.logger.info("Protocol version OK")
-            self.event_q.push('SYS_PROTOCOL_VERSION_OK')
-        def protocol_errback(failure):
-            """protocol check was *not* OK"""
-            self.logger.error("Protocol version error")
-            self.logger.debug('traceback follows:\n\n' + failure.getTraceback())
-            if failure.value.message == 'UNSUPPORTED_VERSION':
-                self.event_q.push('SYS_PROTOCOL_VERSION_ERROR',
-                                  error=failure.getErrorMessage())
-            else:
-                self.event_q.push('SYS_UNKNOWN_ERROR')
-            # it looks like we won't be authenticating, so hook up the
-            # for-testing action_queue deferred now
-            d.chainDeferred(self.action_q.deferred)
-            return failure
-        d.addCallbacks(protocol_callback, protocol_errback)
+        self.action_q.check_version()
 
     def authenticate(self):
         """
         Do the OAuth dance.
         """
-        d = self.action_q.client.oauth_authenticate(self.oauth_client.consumer,
-                                                    self.action_q.token)
-        def oauth_errback(failure):
-            "OAuth failed"
-            self.logger.error("OAuth failed: %s", failure)
-            self.event_q.push('SYS_OAUTH_ERROR', error=failure)
-            return failure
-        def oauth_callback(a):
-            "OAuth succeeded"
-            self.logger.info("Oauth OK")
-            self.event_q.push('SYS_OAUTH_OK')
-            return self.action_q.client
-        d.addCallbacks(oauth_callback, oauth_errback)
-        d.chainDeferred(self.action_q.deferred)
+        self.action_q.authenticate(self.oauth_client.consumer)
 
     def set_capabilities(self):
         """Set the capabilities with the server"""
         self.logger.debug("capabilities query: %r", syncdaemon.REQUIRED_CAPS)
-        d = self.action_q.client.query_caps(syncdaemon.REQUIRED_CAPS)
-
-        def set_caps_callback(req):
-            "Caps query succeeded"
-            if req.accepted:
-                self.event_q.push('SYS_SET_CAPABILITIES_OK')
-                return self.action_q.client
-            else:
-                de = defer.fail("The server denied setting %s capabilities" % \
-                                req.caps)
-                de.addErrback(set_caps_errback)
-                return de
-
-        def set_caps_errback(failure):
-            "Caps set failed"
-            self.logger.error("Capabilities set failed: %s", failure)
-            self.event_q.push('SYS_SET_CAPABILITIES_ERROR',
-                                  error=failure.getErrorMessage())
-            # it looks like we won't be authenticating, so hook up the
-            # for-testing action_queue deferred now
-            d.chainDeferred(self.action_q.deferred)
-            return failure
-
-        def query_caps_callback(req):
-            "Caps query succeeded"
-            self.logger.debug("Capabilities query accepted: %r", req.accepted)
-            if req.accepted:
-                set_d = self.action_q.client.set_caps(syncdaemon.REQUIRED_CAPS)
-                set_d.addCallbacks(set_caps_callback, set_caps_errback)
-                return set_d
-            else:
-                # the server don't have the requested capabilities.
-                # Add the query_errback and return a failure,
-                # in the future we might want to reconnect to another server
-                de = defer.fail("The server don't have the requested"
-                                  " capabilities")
-                de.addErrback(query_caps_errback)
-                return de
-
-        def query_caps_errback(failure):
-            "Caps query failed"
-            self.logger.error("Capabilities query failed: %s", failure)
-            # Push the error to the event queue,
-            self.event_q.push('SYS_SET_CAPABILITIES_ERROR',
-                                  error=failure.getErrorMessage())
-            # it looks like we won't be authenticating, so hook up the
-            # for-testing action_queue deferred now
-            d.chainDeferred(self.action_q.deferred)
-            return failure
-
-        d.addCallbacks(query_caps_callback, query_caps_errback)
+        self.action_q.set_capabilities(syncdaemon.REQUIRED_CAPS)
 
     @defer.inlineCallbacks
     def server_rescan(self):
         """
         Do the server rescan
         """
+        client = self.action_q.client
         yield self.get_root(object())
+        if client is not self.action_q.client:
+            return
         self.event_q.push('SYS_SERVER_RESCAN_STARTING')
         data = self.fs.get_data_for_server_rescan()
         self.logger.info("server rescan: will query %d objects" % len(data))
@@ -362,6 +284,8 @@
                                   % (share or '/root/', node, hash))
             self.logger.debug("server rescan: all data shown")
         yield self.action_q.client.query(data)
+        if client is not self.action_q.client:
+            return
         self.logger.info("server rescan: done")
         self.event_q.push('SYS_SERVER_RESCAN_DONE')
 

=== modified file 'ubuntuone/syncdaemon/state.py'
--- ubuntuone/syncdaemon/state.py	2009-08-21 13:44:34 +0000
+++ ubuntuone/syncdaemon/state.py	2009-08-28 20:46:35 +0000
@@ -95,9 +95,7 @@
         """
         self.watchdog = None
         self.num_timeouts += 1
-        client = self.main.action_q.client
-        if client is not None:
-            client.disconnect()
+        self.main.action_q.disconnect()
         self.main.event_q.push('SYS_HANDSHAKE_TIMEOUT')
 
     @property