Merge lp:~facundo/ubuntuone-client/aqc_errors into lp:ubuntuone-client

[Facundo Batista]

Make ActionQueueCommand to handle errors by their type, not message.

Make ActionQueueCommand to handle errors by their type, not message.

The same for a couple of specific commands.

And fixed a detail that made ACQ to *never* retry an operation on a retryable error.

Tests included.

[Natalia Bidart (nataliabidart)]

Very nice!

review: Approve

[John Lenton (chipaca)]

Is this a fix for something that is going into Lucid? If so, please link it. If not, I'm afraid it would be a lot easier if we merge it after the freeze.

review: Needs Information

[John Lenton (chipaca)]

OK, brown paper bag for me. I promise I *looked* for the bugs.
Getting more coffee, now.

[John Lenton (chipaca)]

review: Abstain

[John Lenton (chipaca)]

The failure.check thing is much better than what it was doing before. So obviously the right thing to do, now that I see it :)

The change from strings to actual classes is also great.

review: Approve

[dobey (dobey)]

Attempt to merge lp:~facundo/ubuntuone-client/aqc_errors into lp:ubuntuone-client failed due to merge conflicts:

text conflict in tests/syncdaemon/test_action_queue.py
text conflict in ubuntuone/syncdaemon/action_queue.py

[Facundo Batista (facundo)]

Solved the conflicts!

[John Lenton (chipaca)]

The attempt to merge lp:~facundo/ubuntuone-client/aqc_errors into lp:ubuntuone-client failed.Below is the output from the failed tests.

/usr/bin/gnome-autogen.sh
checking for autoconf >= 2.53...
(B testing autoconf2.50... not found.
  testing autoconf... found 2.65
checking for automake >= 1.10...
(B testing automake-1.11... found 1.11.1
checking for libtool >= 1.5...
(B testing libtoolize... found 2.2.6b
checking for intltool >= 0.30...
(B testing intltoolize... found 0.41.0
checking for pkg-config >= 0.14.0...
(B testing pkg-config... found 0.22
Checking for required M4 macros...
(BChecking for forbidden M4 macros...
(BProcessing ./configure.ac
(BRunning libtoolize...
(Blibtoolize: putting auxiliary files in `.'.
libtoolize: copying file `./ltmain.sh'
libtoolize: putting macros in AC_CONFIG_MACRO_DIR, `m4'.
libtoolize: copying file `m4/libtool.m4'
libtoolize: copying file `m4/ltoptions.m4'
libtoolize: copying file `m4/ltsugar.m4'
libtoolize: copying file `m4/ltversion.m4'
libtoolize: copying file `m4/lt~obsolete.m4'
Running intltoolize...
(BRunning aclocal-1.11...
(BRunning autoconf...
(BRunning autoheader...
(BRunning automake-1.11...
(BRunning ./configure --prefix=/usr ...
(Bchecking for a BSD-compatible install... /usr/bin/install -c
checking whether build environment is sane... yes
checking for a thread-safe mkdir -p... /bin/mkdir -p
checking for gawk... gawk
checking whether make sets $(MAKE)... yes
checking for style of include used by make... GNU
checking for gcc... gcc
checking whether the C compiler works... yes
checking for C compiler default output file name... a.out
checking for suffix of executables...
checking whether we are cross compiling... no
checking for suffix of object files... o
checking whether we are using the GNU C compiler... yes
checking whether gcc accepts -g... yes
checking for gcc option to accept ISO C89... none needed
checking dependency style of gcc... gcc3
checking for library containing strerror... none required
checking for gcc... (cached) gcc
checking whether we are using the GNU C compiler... (cached) yes
checking whether gcc accepts -g... (cached) yes
checking for gcc option to accept ISO C89... (cached) none needed
checking dependency style of gcc... (cached) gcc3
checking build system type... x86_64-unknown-linux-gnu
checking host system type... x86_64-unknown-linux-gnu
checking for a sed that does not truncate output... /bin/sed
checking for grep that handles long lines and -e... /bin/grep
checking for egrep... /bin/grep -E
checking for fgrep... /bin/grep -F
checking for ld used by gcc... /usr/bin/ld
checking if the linker (/usr/bin/ld) is GNU ld... yes
checking for BSD- or MS-compatible name lister (nm)... /usr/bin/nm -B
checking the name lister (/usr/bin/nm -B) interface... BSD nm
checking whether ln -s works... yes
checking the maximum length of command line arguments... 1572864
checking whether the shell understands some XSI constructs... yes
checking whether the shell understands "+="... yes
checking for /usr/bin/ld option to reload object files... -r
checking for objdu...

[Facundo Batista (facundo)]

It seems there's a problem in your set up, there, as errors is there.

I'm setting this back to approved, to run by someone else. Chipaca, if you see this message, let's talk, :)

[John Lenton (chipaca)]

On Mon, Apr 12, 2010 at 05:05:46PM -0000, Facundo Batista wrote:
> It seems there's a problem in your set up, there, as errors is there.
>
> I'm setting this back to approved, to run by someone else. Chipaca, if you see this message, let's talk, :)

sure thing

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2010-04-12 13:19:40 +0000
+++ tests/syncdaemon/test_action_queue.py	2010-04-12 16:06:36 +0000
@@ -33,6 +33,7 @@
 from functools import wraps
 from StringIO import StringIO
 from twisted.internet import defer, threads, reactor
+from twisted.internet import error as twisted_error
 from twisted.python.failure import DefaultException, Failure
 from twisted.web import server
 
@@ -784,6 +785,113 @@
         self.assertTrue(res is None)
 
 
+class ActionQueueCommandErrors(ConnectedBaseTestCase):
+    """Test the error handling in ActionQueueCommand."""
+
+    def setUp(self):
+        res = super(ActionQueueCommandErrors, self).setUp()
+
+        self.deferred = defer.Deferred()
+
+        class MyLogger(object):
+            """Fake logger that just stores error and warning calls."""
+            def __init__(self):
+                self.logged = None
+
+            def error(self, *a):
+                """Mark that this method was called."""
+                self.logged = "error"
+
+            def warn(self, *a):
+                """Mark that this method was called."""
+                self.logged = "warn"
+
+            def debug(self, *a):
+                """Nothing."""
+
+        class MyCommand(ActionQueueCommand):
+            """Inherit ACQ to provide a retry signaller and a custom log."""
+            # class-closure, cannot use self, pylint: disable-msg=E0213
+            def __init__(innerself, request_queue):
+                super(MyCommand, innerself).__init__(request_queue)
+                innerself.log = MyLogger()
+
+            def retry(innerself):
+                """Signal the retry."""
+                self.deferred.callback(True)
+
+        self.rq = RequestQueue(name='foo', action_queue=self.action_queue)
+        self.command = MyCommand(self.rq)
+        return res
+
+    def test_suppressed_yes_knownerrors(self):
+        """Check that the log is in warning for the known errors."""
+        def send_failure_and_check(errnum, exception_class):
+            """Send the failure."""
+            # prepare what to send
+            protocol_msg = protocol_pb2.Message()
+            protocol_msg.type = protocol_pb2.Message.ERROR
+            protocol_msg.error.type = errnum
+            err = exception_class("request", protocol_msg)
+
+            # set up and test
+            self.command.log.logged = None
+            self.command.end_errback(failure=Failure(err))
+            self.assertEqual(self.command.log.logged, "warn",
+                             "Bad log in exception %s" % (exception_class,))
+
+        known_errors = [x for x in errors._error_mapping.items()
+                        if x[1] != errors.InternalError]
+        for errnum, exception_class in known_errors:
+            send_failure_and_check(errnum, exception_class)
+
+    def test_suppressed_no_internalerror(self):
+        """Check that the log is in error for InternalError."""
+        # prepare what to send
+        protocol_msg = protocol_pb2.Message()
+        protocol_msg.type = protocol_pb2.Message.ERROR
+        protocol_msg.error.type = protocol_pb2.Error.INTERNAL_ERROR
+        err = errors.InternalError("request", protocol_msg)
+
+        # set up and test
+        self.command.end_errback(failure=Failure(err))
+        self.assertEqual(self.command.log.logged, "error")
+
+    def test_suppressed_yes_cancelled(self):
+        """Check that the log is in warning for Cancelled."""
+        err = errors.RequestCancelledError("CANCELLED")
+        self.command.end_errback(failure=Failure(err))
+        self.assertEqual(self.command.log.logged, "warn")
+
+    def test_suppressed_yes_and_retry_when_connectiondone(self):
+        """Check that the log is in warning and retries for ConnectionDone."""
+        self.command.running = True
+        err = twisted_error.ConnectionDone()
+        self.command.end_errback(failure=Failure(err))
+        self.assertEqual(self.command.log.logged, "warn")
+        return self.deferred
+
+    def test_retry_connectionlost(self):
+        """Check that it retries when ConnectionLost."""
+        self.command.running = True
+        err = twisted_error.ConnectionLost()
+        self.command.end_errback(failure=Failure(err))
+        return self.deferred
+
+    def test_retry_tryagain(self):
+        """Check that it retries when TryAgain."""
+        # prepare what to send
+        self.command.running = True
+        protocol_msg = protocol_pb2.Message()
+        protocol_msg.type = protocol_pb2.Message.ERROR
+        protocol_msg.error.type = protocol_pb2.Error.TRY_AGAIN
+        err = errors.TryAgainError("request", protocol_msg)
+
+        # set up and test
+        self.command.end_errback(failure=Failure(err))
+        return self.deferred
+
+
 class ListVolumesTestCase(ConnectedBaseTestCase):
     """Test for ListVolumes ActionQueueCommand."""
 
@@ -1171,9 +1279,8 @@
 
     def test_failure_with_CANCELLED(self):
         """AQ_DOWNLOAD_CANCELLED is pushed."""
-        msg = 'CANCELLED'
-        failure = Failure(DefaultException(msg))
-        res = self.command.handle_failure(failure=failure)
+        err = errors.RequestCancelledError("CANCELLED")
+        res = self.command.handle_failure(failure=Failure(err))
         kwargs = dict(share_id='a_share_id', node_id='a_node_id',
                       server_hash='a_server_hash')
         events = [('AQ_DOWNLOAD_CANCELLED', (), kwargs)]
@@ -1193,9 +1300,11 @@
 
     def test_failure_with_DOES_NOT_EXIST(self):
         """AQ_DOWNLOAD_DOES_NOT_EXIST is pushed."""
-        msg = 'DOES_NOT_EXIST'
-        failure = Failure(DefaultException(msg))
-        res = self.command.handle_failure(failure=failure)
+        protocol_msg = protocol_pb2.Message()
+        protocol_msg.type = protocol_pb2.Message.ERROR
+        protocol_msg.error.type = protocol_pb2.Error.DOES_NOT_EXIST
+        err = errors.DoesNotExistError("request", protocol_msg)
+        res = self.command.handle_failure(failure=Failure(err))
         kwargs = dict(share_id='a_share_id', node_id='a_node_id')
         events = [('AQ_DOWNLOAD_DOES_NOT_EXIST', (), kwargs)]
         self.assertEquals(events, self.command.action_queue.event_queue.events)
@@ -1224,9 +1333,10 @@
 
     def setUp(self):
         """Init."""
-        res = super(UploadTestCase, self).setUp()
+        super(UploadTestCase, self).setUp()
 
-        request_queue = RequestQueue(name='FOO', action_queue=self.action_queue)
+        self.rq = request_queue = RequestQueue(name='FOO',
+                                               action_queue=self.action_queue)
         self.command = Upload(request_queue, share_id='a_share_id',
                               node_id='a_node_id', previous_hash='prev_hash',
                               hash='yadda', crc32=0, size=0,
@@ -1234,7 +1344,29 @@
                               tempfile_factory=lambda: None)
         self.command.start_unqueued() # create the logger
 
-        return res
+    def test_upload_in_progress(self):
+        """Test Upload retries on UploadInProgress."""
+        # monkeypatching is not allowed, let's do inheritance
+        d = defer.Deferred()
+        class MyUpload(Upload):
+            """Just to redefine retry."""
+            def retry(self):
+                """Detect retry was called."""
+                d.callback(True)
+
+        # set up the command
+        command = MyUpload(self.rq, 'share', 'bode', 'prvhash', 'currhash',
+                           0, 0, lambda: None, lambda: None)
+        command.start_unqueued() # create log in the instance
+        command.running = True
+
+        # send the failure
+        protocol_msg = protocol_pb2.Message()
+        protocol_msg.type = protocol_pb2.Message.ERROR
+        protocol_msg.error.type = protocol_pb2.Error.UPLOAD_IN_PROGRESS
+        err = errors.UploadInProgressError("request", protocol_msg)
+        command.end_errback(failure=Failure(err))
+        return d
 
     def test_handle_success_push_event(self):
         """Test AQ_UPLOAD_FINISHED is pushed on success."""

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2010-04-12 13:19:40 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2010-04-12 16:06:36 +0000
@@ -44,11 +44,13 @@
 
 from zope.interface import implements
 from twisted.internet import reactor, defer, threads
+from twisted.internet import error as twisted_errors
 from twisted.names import client as dns_client
 from twisted.python.failure import Failure, DefaultException
 
 from oauth import oauth
-from ubuntuone.storageprotocol import errors, protocol_pb2
+from ubuntuone.storageprotocol import protocol_pb2
+from ubuntuone.storageprotocol import errors as protocol_errors
 from ubuntuone.storageprotocol.client import (
     ThrottlingStorageClient, ThrottlingStorageClientFactory
 )
@@ -913,12 +915,12 @@
                     return
         except request_error, failure:
             self.event_queue.push(event_error, error=str(failure))
-        except errors.AuthenticationRequiredError, failure:
+        except protocol_errors.AuthenticationRequiredError, failure:
             # we need to separate this case from the rest because an
             # AuthenticationRequiredError is an StorageRequestError,
             # and we treat it differently.
             self.event_queue.push('SYS_UNKNOWN_ERROR')
-        except errors.StorageRequestError, failure:
+        except protocol_errors.StorageRequestError, failure:
             self.event_queue.push('SYS_SERVER_ERROR', error=str(failure))
         except Exception, failure:
             self.event_queue.push('SYS_UNKNOWN_ERROR')
@@ -940,7 +942,7 @@
         """Check if the client protocol version matches that of the server."""
         check_version_d = self._send_request_and_handle_errors(
             request=self.client.protocol_version,
-            request_error=errors.UnsupportedVersionError,
+            request_error=protocol_errors.UnsupportedVersionError,
             event_error='SYS_PROTOCOL_VERSION_ERROR',
             event_ok='SYS_PROTOCOL_VERSION_OK'
         )
@@ -986,7 +988,7 @@
         """Authenticate the client against the server using oauth_consumer."""
         authenticate_d = self._send_request_and_handle_errors(
             request=self.client.oauth_authenticate,
-            request_error=errors.AuthenticationFailedError,
+            request_error=protocol_errors.AuthenticationFailedError,
             event_error='SYS_AUTH_ERROR', event_ok='SYS_AUTH_OK',
             args=(oauth_consumer, self.token)
         )
@@ -1231,21 +1233,20 @@
 class ActionQueueCommand(object):
     """Base of all the action queue commands."""
 
-    # protobuf doesn't seem to have much introspectionable stuff
-    # without going into private attributes
+    # the info used in the protocol errors is hidden, but very useful!
     # pylint: disable-msg=W0212
-    known_error_messages = (set(protocol_pb2._ERROR_ERRORTYPE.values_by_name)
-                            | set(['CANCELLED']))
-    suppressed_error_messages = (known_error_messages
-                                 - set(['INTERNAL_ERROR'])
-                                 | set(['Cleaned up',
-                                        'Connection was closed cleanly.']))
-    retryable_errors = set([
-            'Cleaned up',
-            'TRY_AGAIN',
-            'Connection was closed cleanly.',
-            'Connection to the other side was lost in a non-clean fashion.',
-            ])
+    suppressed_error_messages = (
+        [x for x in protocol_errors._error_mapping.values()
+         if x is not protocol_errors.InternalError] +
+        [protocol_errors.RequestCancelledError,
+         twisted_errors.ConnectionDone, RequestCleanedUp]
+    )
+
+    retryable_errors = (
+        protocol_errors.TryAgainError,
+        twisted_errors.ConnectionDone,
+        twisted_errors.ConnectionLost,
+    )
 
     logged_attrs = ()
 
@@ -1277,12 +1278,9 @@
 
     def check_conditions(self):
         """Check conditions on which the command may be waiting."""
-        pass
 
     def demark(self, *maybe_markers):
-        """
-        Arrange to have maybe_markers realized
-        """
+        """Arrange to have maybe_markers realized."""
         l = []
         for marker in maybe_markers:
             if IMarker.providedBy(marker):
@@ -1303,9 +1301,9 @@
 
     @staticmethod
     def unwrap(results):
-        """
-        Unpack the values from the result of a DeferredList. If
-        there's a failure, return it instead.
+        """Unpack the values from the result of a DeferredList.
+
+        If there's a failure, return it instead.
         """
         values = []
         for result in results:
@@ -1321,9 +1319,7 @@
         return values
 
     def end_callback(self, arg):
-        """
-        It worked!
-        """
+        """It worked!"""
         if self.running:
             self.log.debug('success')
             return self.handle_success(arg)
@@ -1331,18 +1327,17 @@
             self.log.debug('not running, so no success')
 
     def end_errback(self, failure):
-        """
-        It failed!
-        """
+        """It failed!"""
         error_message = failure.getErrorMessage()
-        if error_message not in self.suppressed_error_messages:
+        if failure.check(*self.suppressed_error_messages):
+            self.log.warn('failure', error_message)
+        else:
             self.log.error('failure', error_message)
             self.log.debug('traceback follows:\n\n' + failure.getTraceback())
-        else:
-            self.log.warn('failure', error_message)
+        was_running = self.running
         self.cleanup()
-        if error_message in self.retryable_errors:
-            if self.running:
+        if failure.check(*self.retryable_errors):
+            if was_running:
                 reactor.callLater(0.1, self.retry)
         else:
             return self.handle_failure(failure)
@@ -1352,37 +1347,29 @@
         self.log = self.make_logger()
 
     def start(self, _=None):
-        """
-        Queue the command.
-        """
+        """Queue the command."""
         self.start_unqueued()
         self.log.debug('queueing in the %s', self._queue.name)
         self._queue.queue(self)
 
     def cleanup(self):
-        """
-        Do whatever is needed to clean up from a failure (such as stop
-        producers and other such that aren't cleaned up appropriately
-        on their own)
+        """Do whatever is needed to clean up from a failure.
+
+        For example, stop producers and others that aren't cleaned up
+        appropriately on their own.
         """
         self.running = False
         self.log.debug('cleanup')
 
     def _start(self):
-        """
-        Do the specialized pre-run setup
-        """
+        """Do the specialized pre-run setup."""
         return defer.succeed(None)
 
     def store_marker_result(self, _):
-        """
-        Called when all the markers are realized.
-        """
+        """Called when all the markers are realized."""
 
     def run(self):
-        """
-        Do the deed.
-        """
+        """Do the deed."""
         self.running = True
         if self.start_done:
             self.log.debug('retrying')
@@ -1409,37 +1396,31 @@
         return d
 
     def handle_success(self, success):
-        """
-        Do anthing that's needed to handle success of the operation.
-        """
+        """Do anthing that's needed to handle success of the operation."""
         return success
 
     def handle_failure(self, failure):
-        """
-        Do anthing that's needed to handle failure of the operation.
+        """Do anthing that's needed to handle failure of the operation.
+
         Note that cancellation and TRY_AGAIN are already handled.
         """
         return failure
 
     def cleanup_and_retry(self):
-        """
-        First, cleanup; then, retry :)
-        """
+        """First, cleanup; then, retry :)"""
         self.log.debug('cleanup and retry')
         self.cleanup()
         return self.retry()
 
     def retry(self):
-        """
-        Request cancelled or TRY_AGAIN. Well then, try again!
-        """
+        """Request cancelled or TRY_AGAIN. Well then, try again!"""
         self.running = False
         self.log.debug('will retry')
         return self._queue.queue_top(self)
 
     @property
     def action_queue(self):
-        """Returns the action queue."""
+        """Return the action queue."""
         return self._queue.action_queue
 
     def __str__(self, str_attrs=None):
@@ -2214,33 +2195,25 @@
         self.action_queue.cancel_download(self.share_id, self.node_id)
 
     def _start(self):
-        """
-        Do the specialized pre-run setup
-        """
+        """Do the specialized pre-run setup."""
         return self.demark(self.node_id)
 
     def cancel(self):
-        """
-        Cancel the download.
-        """
+        """Cancel the download."""
         self.cancelled = True
         if self.download_req is not None:
             self.download_req.cancel()
         self.cleanup()
 
     def store_marker_result(self, (node_id,)):
-        """
-        Called when all the markers are realized.
-        """
+        """Called when all the markers are realized."""
         self.node_id = node_id
         self.start_done = True
 
     def _run(self):
-        """
-        Do the actual running
-        """
+        """Do the actual running."""
         if self.cancelled:
-            return defer.fail(RuntimeError('CANCELLED'))
+            return defer.fail(RequestCleanedUp('CANCELLED'))
         if self.fileobj is None:
             try:
                 self.fileobj = self.fileobj_factory()
@@ -2269,16 +2242,14 @@
         self.download_req = req
         downloading[self.share_id, self.node_id]['req'] = req
         d = req.deferred
-        d.addBoth(lambda x: defer.fail(RuntimeError('CANCELLED'))
+        d.addBoth(lambda x: defer.fail(RequestCleanedUp('CANCELLED'))
                   if self.cancelled else x)
         d.addCallback(passit(
                 lambda _: downloading.pop((self.share_id, self.node_id))))
         return d
 
     def handle_success(self, _):
-        """
-        It worked! Push the event.
-        """
+        """It worked! Push the event."""
         self.sync()
         # for directories, a FINISHED; for files, a COMMIT (the Nanny
         # will issue the FINISHED if it's ok)
@@ -2292,27 +2263,25 @@
                                            server_hash=self.server_hash)
 
     def handle_failure(self, failure):
-        """
-        It didn't work! Push the event.
-        """
+        """It didn't work! Push the event."""
         downloading = self.action_queue.downloading
         if (self.share_id, self.node_id) in downloading and \
             downloading[self.share_id, self.node_id]['command'] is self:
             del downloading[self.share_id, self.node_id]
         self.reset_fileobj()
-        error_msg = failure.getErrorMessage()
-        if error_msg == 'CANCELLED':
+        if failure.check(protocol_errors.RequestCancelledError,
+                         RequestCleanedUp):
             self.action_queue.event_queue.push('AQ_DOWNLOAD_CANCELLED',
                                                share_id=self.share_id,
                                                node_id=self.node_id,
                                                server_hash=self.server_hash)
-        elif error_msg == 'DOES_NOT_EXIST':
+        elif failure.check(protocol_errors.DoesNotExistError):
             self.action_queue.event_queue.push('AQ_DOWNLOAD_DOES_NOT_EXIST',
                                                share_id=self.share_id,
                                                node_id=self.node_id)
         else:
             self.action_queue.event_queue.push('AQ_DOWNLOAD_ERROR',
-                                               error=error_msg,
+                                               error=failure.getErrorMessage(),
                                                share_id=self.share_id,
                                                node_id=self.node_id,
                                                server_hash=self.server_hash)
@@ -2328,9 +2297,7 @@
             self.fileobj.truncate(0)
 
     def cb(self, bytes):
-        """
-        A streaming decompressor
-        """
+        """A streaming decompressor."""
         dloading = self.action_queue.downloading[self.share_id,
                                                  self.node_id]
         dloading['n_bytes_read'] += len(bytes)
@@ -2339,16 +2306,12 @@
                                  # see the downloaded size
 
     def nacb(self, **kwargs):
-        """
-        set the node attrs in the 'currently downloading' dict
-        """
+        """Set the node attrs in the 'currently downloading' dict."""
         self.action_queue.downloading[self.share_id,
                                       self.node_id].update(kwargs)
 
     def sync(self):
-        """
-        Flush the buffers and sync them to disk if possible
-        """
+        """Flush the buffers and sync them to disk if possible."""
         remains = self.gunzip.flush()
         if remains:
             self.fileobj.write(remains)
@@ -2391,8 +2354,8 @@
 
     logged_attrs = ('share_id', 'node_id', 'previous_hash', 'hash', 'crc32',
                     'size', 'fileobj_factory')
-    retryable_errors = (ActionQueueCommand.retryable_errors
-                        | set(['UPLOAD_IN_PROGRESS']))
+    retryable_errors = ActionQueueCommand.retryable_errors + (
+                                        protocol_errors.UploadInProgressError,)
 
     def __init__(self, request_queue, share_id, node_id, previous_hash, hash,
                  crc32, size, fileobj_factory, tempfile_factory):
@@ -2436,9 +2399,7 @@
             self.upload_req.producer.stopProducing()
 
     def _start(self):
-        """
-        Do the specialized pre-run setup
-        """
+        """Do the specialized pre-run setup."""
         d = defer.Deferred()
 
         uploading = {"hash": self.hash, "req": self}
@@ -2446,7 +2407,7 @@
 
         d = self.action_queue.zip_queue.zip(self)
         d.addCallback(lambda _: self.demark(self.node_id))
-        d.addBoth(lambda x: defer.fail(RuntimeError('CANCELLED'))
+        d.addBoth(lambda x: defer.fail(RequestCleanedUp('CANCELLED'))
                   if self.cancelled else x)
         return d
 
@@ -2462,11 +2423,9 @@
         self.start_done = True
 
     def _run(self):
-        """
-        Do the actual running
-        """
+        """Do the actual running."""
         if self.cancelled:
-            return defer.fail(RuntimeError('CANCELLED'))
+            return defer.fail(RequestCleanedUp('CANCELLED'))
         uploading = {"hash": self.hash, "deflated_size": self.deflated_size,
                      "req": self}
         self.action_queue.uploading[self.share_id, self.node_id] = uploading
@@ -2484,7 +2443,7 @@
             self.crc32, self.size, self.deflated_size, f)
         self.upload_req = req
         d = req.deferred
-        d.addBoth(lambda x: defer.fail(RuntimeError('CANCELLED'))
+        d.addBoth(lambda x: defer.fail(RequestCleanedUp('CANCELLED'))
                   if self.cancelled else x)
         d.addBoth(passit(lambda _:
                              self.action_queue.uploading.pop((self.share_id,