Merge lp:~jameinel/bzr/2.5-client-reconnect-819604 into lp:bzr

[John A Meinel]

Bug #819604, allow clients to reconnect if they get ConnectionReset while trying to send an RPC.

And finally, this is a merge-up of all the previous versions of my client-reconnect patch, targetting bzr.dev.

The main conflicts were just the test_smart_transport.py which had some refactorings because of my other work in the area. Otherwise, pretty much the same as the 2.4 patch.

[Martin Pool (mbp)]

If this is just a merge-up, feel free to just land it. Otherwise I'll
read it later.

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 10/13/2011 7:07 AM, Martin Pool wrote:
> If this is just a merge-up, feel free to just land it. Otherwise
> I'll read it later.
>

This is a merge up (+ resolving conflicts, etc), but the 2.1 code
still needs to be approved.

John
=:->
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk6WuOIACgkQJdeBCYSNAAPN/gCfVVd1FRdM7GDecRGo4jda8k0K
7M8An39FrO40evTwnSn9boZ4jqxXZhhb
=ncmU
-----END PGP SIGNATURE-----

[Martin Pool (mbp)]

I think we should land this feature and give it a good shake out on 2.5 first, before even thinking about putting it back into previous series, partly because it's just a big patch, partly because so many different cases can come up in handling disconnects. So partly because of that, and to see everything that would land, and to avoid reviewing the backport shims right now.

I really like the split out of the request object; that is pretty clear.

[fix] Obviously if we land this only in trunk first, we have to remove those lines from the old news.

I am glad you added a flag to turn it off.

I think this is pretty clear and I don't see any problems. I don't see any unaddressed problems in the other reviews (are there?) I wouldn't be surprised if we see some problems on the real network but the only way to tell is to bring it in.

Does anyone else have objections or see problems?

review: Approve

[Jelmer Vernooij (jelmer)]

mgz and I just discussed this independently as well, before reading poolie's review and came to the same conclusion.

Let's land this as early in the cycle as we can so we can shake out bugs.

Independently, we should look into the issues (spurious failures, etc) with some of the smart server tests, some of which have become more frequent after the prerequisites of this branch landed.

We're a bit worried about this feature hiding issues during tests, e.g. with servers dropping connections after requests. One way to work around that would be to add tests that make sure that commands only open X connections. This would also catch some other bugs, such as command implementations not using possible_transports.

review: Approve

[Jelmer Vernooij (jelmer)]

sent to pqm by email

=== modified file 'bzrlib/help_topics/en/debug-flags.txt'
--- bzrlib/help_topics/en/debug-flags.txt	2011-05-27 17:10:05 +0000
+++ bzrlib/help_topics/en/debug-flags.txt	2011-10-10 15:16:30 +0000
@@ -24,6 +24,8 @@
 -Dindex           Trace major index operations.
 -Dknit            Trace knit operations.
 -Dlock            Trace when lockdir locks are taken or released.
+-Dnoretry         If a connection is reset, fail immediately rather than
+                  retrying the request.
 -Dprogress        Trace progress bar operations.
 -Dmem_dump        Dump memory to a file upon an out of memory error.
 -Dmerge           Emit information for debugging merges.

=== modified file 'bzrlib/smart/client.py'
--- bzrlib/smart/client.py	2011-03-30 11:45:54 +0000
+++ bzrlib/smart/client.py	2011-10-10 15:16:30 +0000
@@ -14,12 +14,18 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
+from bzrlib import lazy_import
+lazy_import.lazy_import(globals(), """
+from bzrlib.smart import request as _mod_request
+""")
+
 import bzrlib
 from bzrlib.smart import message, protocol
-from bzrlib.trace import warning
 from bzrlib import (
+    debug,
     errors,
     hooks,
+    trace,
     )
 
 
@@ -39,93 +45,12 @@
     def __repr__(self):
         return '%s(%r)' % (self.__class__.__name__, self._medium)
 
-    def _send_request(self, protocol_version, method, args, body=None,
-                      readv_body=None, body_stream=None):
-        encoder, response_handler = self._construct_protocol(
-            protocol_version)
-        encoder.set_headers(self._headers)
-        if body is not None:
-            if readv_body is not None:
-                raise AssertionError(
-                    "body and readv_body are mutually exclusive.")
-            if body_stream is not None:
-                raise AssertionError(
-                    "body and body_stream are mutually exclusive.")
-            encoder.call_with_body_bytes((method, ) + args, body)
-        elif readv_body is not None:
-            if body_stream is not None:
-                raise AssertionError(
-                    "readv_body and body_stream are mutually exclusive.")
-            encoder.call_with_body_readv_array((method, ) + args, readv_body)
-        elif body_stream is not None:
-            encoder.call_with_body_stream((method, ) + args, body_stream)
-        else:
-            encoder.call(method, *args)
-        return response_handler
-
-    def _run_call_hooks(self, method, args, body, readv_body):
-        if not _SmartClient.hooks['call']:
-            return
-        params = CallHookParams(method, args, body, readv_body, self._medium)
-        for hook in _SmartClient.hooks['call']:
-            hook(params)
-
     def _call_and_read_response(self, method, args, body=None, readv_body=None,
             body_stream=None, expect_response_body=True):
-        self._run_call_hooks(method, args, body, readv_body)
-        if self._medium._protocol_version is not None:
-            response_handler = self._send_request(
-                self._medium._protocol_version, method, args, body=body,
-                readv_body=readv_body, body_stream=body_stream)
-            return (response_handler.read_response_tuple(
-                        expect_body=expect_response_body),
-                    response_handler)
-        else:
-            for protocol_version in [3, 2]:
-                if protocol_version == 2:
-                    # If v3 doesn't work, the remote side is older than 1.6.
-                    self._medium._remember_remote_is_before((1, 6))
-                response_handler = self._send_request(
-                    protocol_version, method, args, body=body,
-                    readv_body=readv_body, body_stream=body_stream)
-                try:
-                    response_tuple = response_handler.read_response_tuple(
-                        expect_body=expect_response_body)
-                except errors.UnexpectedProtocolVersionMarker, err:
-                    # TODO: We could recover from this without disconnecting if
-                    # we recognise the protocol version.
-                    warning(
-                        'Server does not understand Bazaar network protocol %d,'
-                        ' reconnecting.  (Upgrade the server to avoid this.)'
-                        % (protocol_version,))
-                    self._medium.disconnect()
-                    continue
-                except errors.ErrorFromSmartServer:
-                    # If we received an error reply from the server, then it
-                    # must be ok with this protocol version.
-                    self._medium._protocol_version = protocol_version
-                    raise
-                else:
-                    self._medium._protocol_version = protocol_version
-                    return response_tuple, response_handler
-            raise errors.SmartProtocolError(
-                'Server is not a Bazaar server: ' + str(err))
-
-    def _construct_protocol(self, version):
-        request = self._medium.get_request()
-        if version == 3:
-            request_encoder = protocol.ProtocolThreeRequester(request)
-            response_handler = message.ConventionalResponseHandler()
-            response_proto = protocol.ProtocolThreeDecoder(
-                response_handler, expect_version_marker=True)
-            response_handler.setProtoAndMediumRequest(response_proto, request)
-        elif version == 2:
-            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
-            response_handler = request_encoder
-        else:
-            request_encoder = protocol.SmartClientRequestProtocolOne(request)
-            response_handler = request_encoder
-        return request_encoder, response_handler
+        request = _SmartClientRequest(self, method, args, body=body,
+            readv_body=readv_body, body_stream=body_stream,
+            expect_response_body=expect_response_body)
+        return request.call_and_read_response()
 
     def call(self, method, *args):
         """Call a method on the remote server."""
@@ -191,6 +116,203 @@
         return self._medium.remote_path_from_transport(transport)
 
 
+class _SmartClientRequest(object):
+    """Encapsulate the logic for a single request.
+
+    This class handles things like reconnecting and sending the request a
+    second time when the connection is reset in the middle. It also handles the
+    multiple requests that get made if we don't know what protocol the server
+    supports yet.
+
+    Generally, you build up one of these objects, passing in the arguments that
+    you want to send to the server, and then use 'call_and_read_response' to
+    get the response from the server.
+    """
+
+    def __init__(self, client, method, args, body=None, readv_body=None,
+                 body_stream=None, expect_response_body=True):
+        self.client = client
+        self.method = method
+        self.args = args
+        self.body = body
+        self.readv_body = readv_body
+        self.body_stream = body_stream
+        self.expect_response_body = expect_response_body
+
+    def call_and_read_response(self):
+        """Send the request to the server, and read the initial response.
+
+        This doesn't read all of the body content of the response, instead it
+        returns (response_tuple, response_handler). response_tuple is the 'ok',
+        or 'error' information, and 'response_handler' can be used to get the
+        content stream out.
+        """
+        self._run_call_hooks()
+        protocol_version = self.client._medium._protocol_version
+        if protocol_version is None:
+            return self._call_determining_protocol_version()
+        else:
+            return self._call(protocol_version)
+
+    def _is_safe_to_send_twice(self):
+        """Check if the current method is re-entrant safe."""
+        if self.body_stream is not None or 'noretry' in debug.debug_flags:
+            # We can't restart a body stream that has already been consumed.
+            return False
+        request_type = _mod_request.request_handlers.get_info(self.method)
+        if request_type in ('read', 'idem', 'semi'):
+            return True
+        # If we have gotten this far, 'stream' cannot be retried, because we
+        # already consumed the local stream.
+        if request_type in ('semivfs', 'mutate', 'stream'):
+            return False
+        trace.mutter('Unknown request type: %s for method %s'
+                     % (request_type, self.method))
+        return False
+
+    def _run_call_hooks(self):
+        if not _SmartClient.hooks['call']:
+            return
+        params = CallHookParams(self.method, self.args, self.body,
+                                self.readv_body, self.client._medium)
+        for hook in _SmartClient.hooks['call']:
+            hook(params)
+
+    def _call(self, protocol_version):
+        """We know the protocol version.
+
+        So this just sends the request, and then reads the response. This is
+        where the code will be to retry requests if the connection is closed.
+        """
+        response_handler = self._send(protocol_version)
+        try:
+            response_tuple = response_handler.read_response_tuple(
+                expect_body=self.expect_response_body)
+        except errors.ConnectionReset, e:
+            self.client._medium.reset()
+            if not self._is_safe_to_send_twice():
+                raise
+            trace.warning('ConnectionReset reading response for %r, retrying'
+                          % (self.method,))
+            trace.log_exception_quietly()
+            encoder, response_handler = self._construct_protocol(
+                protocol_version)
+            self._send_no_retry(encoder)
+            response_tuple = response_handler.read_response_tuple(
+                expect_body=self.expect_response_body)
+        return (response_tuple, response_handler)
+
+    def _call_determining_protocol_version(self):
+        """Determine what protocol the remote server supports.
+
+        We do this by placing a request in the most recent protocol, and
+        handling the UnexpectedProtocolVersionMarker from the server.
+        """
+        for protocol_version in [3, 2]:
+            if protocol_version == 2:
+                # If v3 doesn't work, the remote side is older than 1.6.
+                self.client._medium._remember_remote_is_before((1, 6))
+            try:
+                response_tuple, response_handler = self._call(protocol_version)
+            except errors.UnexpectedProtocolVersionMarker, err:
+                # TODO: We could recover from this without disconnecting if
+                # we recognise the protocol version.
+                trace.warning(
+                    'Server does not understand Bazaar network protocol %d,'
+                    ' reconnecting.  (Upgrade the server to avoid this.)'
+                    % (protocol_version,))
+                self.client._medium.disconnect()
+                continue
+            except errors.ErrorFromSmartServer:
+                # If we received an error reply from the server, then it
+                # must be ok with this protocol version.
+                self.client._medium._protocol_version = protocol_version
+                raise
+            else:
+                self.client._medium._protocol_version = protocol_version
+                return response_tuple, response_handler
+        raise errors.SmartProtocolError(
+            'Server is not a Bazaar server: ' + str(err))
+
+    def _construct_protocol(self, version):
+        """Build the encoding stack for a given protocol version."""
+        request = self.client._medium.get_request()
+        if version == 3:
+            request_encoder = protocol.ProtocolThreeRequester(request)
+            response_handler = message.ConventionalResponseHandler()
+            response_proto = protocol.ProtocolThreeDecoder(
+                response_handler, expect_version_marker=True)
+            response_handler.setProtoAndMediumRequest(response_proto, request)
+        elif version == 2:
+            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
+            response_handler = request_encoder
+        else:
+            request_encoder = protocol.SmartClientRequestProtocolOne(request)
+            response_handler = request_encoder
+        return request_encoder, response_handler
+
+    def _send(self, protocol_version):
+        """Encode the request, and send it to the server.
+
+        This will retry a request if we get a ConnectionReset while sending the
+        request to the server. (Unless we have a body_stream that we have
+        already started consuming, since we can't restart body_streams)
+
+        :return: response_handler as defined by _construct_protocol
+        """
+        encoder, response_handler = self._construct_protocol(protocol_version)
+        try:
+            self._send_no_retry(encoder)
+        except errors.ConnectionReset, e:
+            # If we fail during the _send_no_retry phase, then we can
+            # be confident that the server did not get our request, because we
+            # haven't started waiting for the reply yet. So try the request
+            # again. We only issue a single retry, because if the connection
+            # really is down, there is no reason to loop endlessly.
+
+            # Connection is dead, so close our end of it.
+            self.client._medium.reset()
+            if (('noretry' in debug.debug_flags)
+                or (self.body_stream is not None
+                    and encoder.body_stream_started)):
+                # We can't restart a body_stream that has been partially
+                # consumed, so we don't retry.
+                # Note: We don't have to worry about
+                #   SmartClientRequestProtocolOne or Two, because they don't
+                #   support client-side body streams.
+                raise
+            trace.warning('ConnectionReset calling %r, retrying'
+                          % (self.method,))
+            trace.log_exception_quietly()
+            encoder, response_handler = self._construct_protocol(
+                protocol_version)
+            self._send_no_retry(encoder)
+        return response_handler
+
+    def _send_no_retry(self, encoder):
+        """Just encode the request and try to send it."""
+        encoder.set_headers(self.client._headers)
+        if self.body is not None:
+            if self.readv_body is not None:
+                raise AssertionError(
+                    "body and readv_body are mutually exclusive.")
+            if self.body_stream is not None:
+                raise AssertionError(
+                    "body and body_stream are mutually exclusive.")
+            encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
+        elif self.readv_body is not None:
+            if self.body_stream is not None:
+                raise AssertionError(
+                    "readv_body and body_stream are mutually exclusive.")
+            encoder.call_with_body_readv_array((self.method, ) + self.args,
+                                               self.readv_body)
+        elif self.body_stream is not None:
+            encoder.call_with_body_stream((self.method, ) + self.args,
+                                          self.body_stream)
+        else:
+            encoder.call(self.method, *self.args)
+
+
 class SmartClientHooks(hooks.Hooks):
 
     def __init__(self):

=== modified file 'bzrlib/smart/medium.py'
--- bzrlib/smart/medium.py	2011-10-03 08:15:39 +0000
+++ bzrlib/smart/medium.py	2011-10-10 15:16:30 +0000
@@ -881,6 +881,14 @@
         """
         return SmartClientStreamMediumRequest(self)
 
+    def reset(self):
+        """We have been disconnected, reset current state.
+
+        This resets things like _current_request and connected state.
+        """
+        self.disconnect()
+        self._current_request = None
+
 
 class SmartSimplePipesClientMedium(SmartClientStreamMedium):
     """A client medium using simple pipes.
@@ -895,11 +903,20 @@
 
     def _accept_bytes(self, bytes):
         """See SmartClientStreamMedium.accept_bytes."""
-        self._writeable_pipe.write(bytes)
+        try:
+            self._writeable_pipe.write(bytes)
+        except IOError, e:
+            if e.errno in (errno.EINVAL, errno.EPIPE):
+                raise errors.ConnectionReset(
+                    "Error trying to write to subprocess:\n%s" % (e,))
+            raise
         self._report_activity(len(bytes), 'write')
 
     def _flush(self):
         """See SmartClientStreamMedium._flush()."""
+        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
+        #       However, testing shows that even when the child process is
+        #       gone, this doesn't error.
         self._writeable_pipe.flush()
 
     def _read_bytes(self, count):
@@ -924,8 +941,8 @@
 
 class SmartSSHClientMedium(SmartClientStreamMedium):
     """A client medium using SSH.
-    
-    It delegates IO to a SmartClientSocketMedium or
+
+    It delegates IO to a SmartSimplePipesClientMedium or
     SmartClientAlreadyConnectedSocketMedium (depending on platform).
     """
 
@@ -1164,5 +1181,3 @@
         This invokes self._medium._flush to ensure all bytes are transmitted.
         """
         self._medium._flush()
-
-

=== modified file 'bzrlib/smart/protocol.py'
--- bzrlib/smart/protocol.py	2011-05-19 09:32:38 +0000
+++ bzrlib/smart/protocol.py	2011-10-10 15:16:30 +0000
@@ -1081,9 +1081,6 @@
         self._real_write_func = write_func
 
     def _write_func(self, bytes):
-        # TODO: It is probably more appropriate to use sum(map(len, _buf))
-        #       for total number of bytes to write, rather than buffer based on
-        #       the number of write() calls
         # TODO: Another possibility would be to turn this into an async model.
         #       Where we let another thread know that we have some bytes if
         #       they want it, but we don't actually block for it
@@ -1292,6 +1289,7 @@
         _ProtocolThreeEncoder.__init__(self, medium_request.accept_bytes)
         self._medium_request = medium_request
         self._headers = {}
+        self.body_stream_started = None
 
     def set_headers(self, headers):
         self._headers = headers.copy()
@@ -1357,6 +1355,7 @@
             if path is not None:
                 mutter('                  (to %s)', path)
             self._request_start_time = osutils.timer_func()
+        self.body_stream_started = False
         self._write_protocol_version()
         self._write_headers(self._headers)
         self._write_structure(args)
@@ -1364,6 +1363,9 @@
         #       have finished sending the stream.  We would notice at the end
         #       anyway, but if the medium can deliver it early then it's good
         #       to short-circuit the whole request...
+        # Provoke any ConnectionReset failures before we start the body stream.
+        self.flush()
+        self.body_stream_started = True
         for exc_info, part in _iter_with_errors(stream):
             if exc_info is not None:
                 # Iterating the stream failed.  Cleanly abort the request.

=== modified file 'bzrlib/smart/request.py'
--- bzrlib/smart/request.py	2011-05-19 09:32:38 +0000
+++ bzrlib/smart/request.py	2011-10-10 15:16:30 +0000
@@ -1,4 +1,4 @@
-# Copyright (C) 2006-2010 Canonical Ltd
+# Copyright (C) 2006-2011 Canonical Ltd
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -491,157 +491,202 @@
         return SuccessfulSmartServerResponse((answer,))
 
 
+# In the 'info' attribute, we store whether this request is 'safe' to retry if
+# we get a disconnect while reading the response. It can have the values:
+#   read    This is purely a read request, so retrying it is perfectly ok.
+#   idem    An idempotent write request. Something like 'put' where if you put
+#           the same bytes twice you end up with the same final bytes.
+#   semi    This is a request that isn't strictly idempotent, but doesn't
+#           result in corruption if it is retried. This is for things like
+#           'lock' and 'unlock'. If you call lock, it updates the disk
+#           structure. If you fail to read the response, you won't be able to
+#           use the lock, because you don't have the lock token. Calling lock
+#           again will fail, because the lock is already taken. However, we
+#           can't tell if the server received our request or not. If it didn't,
+#           then retrying the request is fine, as it will actually do what we
+#           want. If it did, we will interrupt the current operation, but we
+#           are no worse off than interrupting the current operation because of
+#           a ConnectionReset.
+#   semivfs Similar to semi, but specific to a Virtual FileSystem request.
+#   stream  This is a request that takes a stream that cannot be restarted if
+#           consumed. This request is 'safe' in that if we determine the
+#           connection is closed before we consume the stream, we can try
+#           again.
+#   mutate  State is updated in a way that replaying that request results in a
+#           different state. For example 'append' writes more bytes to a given
+#           file. If append succeeds, it moves the file pointer.
 request_handlers = registry.Registry()
 request_handlers.register_lazy(
-    'append', 'bzrlib.smart.vfs', 'AppendRequest')
+    'append', 'bzrlib.smart.vfs', 'AppendRequest', info='mutate')
 request_handlers.register_lazy(
     'Branch.get_config_file', 'bzrlib.smart.branch',
-    'SmartServerBranchGetConfigFile')
+    'SmartServerBranchGetConfigFile', info='read')
 request_handlers.register_lazy(
-    'Branch.get_parent', 'bzrlib.smart.branch', 'SmartServerBranchGetParent')
+    'Branch.get_parent', 'bzrlib.smart.branch', 'SmartServerBranchGetParent',
+    info='read')
 request_handlers.register_lazy(
     'Branch.get_tags_bytes', 'bzrlib.smart.branch',
-    'SmartServerBranchGetTagsBytes')
+    'SmartServerBranchGetTagsBytes', info='read')
 request_handlers.register_lazy(
     'Branch.set_tags_bytes', 'bzrlib.smart.branch',
-    'SmartServerBranchSetTagsBytes')
+    'SmartServerBranchSetTagsBytes', info='idem')
 request_handlers.register_lazy(
     'Branch.heads_to_fetch', 'bzrlib.smart.branch',
-    'SmartServerBranchHeadsToFetch')
-request_handlers.register_lazy(
-    'Branch.get_stacked_on_url', 'bzrlib.smart.branch', 'SmartServerBranchRequestGetStackedOnURL')
-request_handlers.register_lazy(
-    'Branch.last_revision_info', 'bzrlib.smart.branch', 'SmartServerBranchRequestLastRevisionInfo')
-request_handlers.register_lazy(
-    'Branch.lock_write', 'bzrlib.smart.branch', 'SmartServerBranchRequestLockWrite')
-request_handlers.register_lazy( 'Branch.revision_history',
-    'bzrlib.smart.branch', 'SmartServerRequestRevisionHistory')
-request_handlers.register_lazy( 'Branch.set_config_option',
-    'bzrlib.smart.branch', 'SmartServerBranchRequestSetConfigOption')
-request_handlers.register_lazy( 'Branch.set_config_option_dict',
-    'bzrlib.smart.branch', 'SmartServerBranchRequestSetConfigOptionDict')
-request_handlers.register_lazy( 'Branch.set_last_revision',
-    'bzrlib.smart.branch', 'SmartServerBranchRequestSetLastRevision')
+    'SmartServerBranchHeadsToFetch', info='read')
+request_handlers.register_lazy(
+    'Branch.get_stacked_on_url', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestGetStackedOnURL', info='read')
+request_handlers.register_lazy(
+    'Branch.last_revision_info', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestLastRevisionInfo', info='read')
+request_handlers.register_lazy(
+    'Branch.lock_write', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestLockWrite', info='semi')
+request_handlers.register_lazy(
+    'Branch.revision_history', 'bzrlib.smart.branch',
+    'SmartServerRequestRevisionHistory', info='read')
+request_handlers.register_lazy(
+    'Branch.set_config_option', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestSetConfigOption', info='idem')
+request_handlers.register_lazy(
+    'Branch.set_config_option_dict', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestSetConfigOptionDict', info='idem')
+request_handlers.register_lazy(
+    'Branch.set_last_revision', 'bzrlib.smart.branch',
+    'SmartServerBranchRequestSetLastRevision', info='idem')
 request_handlers.register_lazy(
     'Branch.set_last_revision_info', 'bzrlib.smart.branch',
-    'SmartServerBranchRequestSetLastRevisionInfo')
+    'SmartServerBranchRequestSetLastRevisionInfo', info='idem')
 request_handlers.register_lazy(
     'Branch.set_last_revision_ex', 'bzrlib.smart.branch',
-    'SmartServerBranchRequestSetLastRevisionEx')
+    'SmartServerBranchRequestSetLastRevisionEx', info='idem')
 request_handlers.register_lazy(
     'Branch.set_parent_location', 'bzrlib.smart.branch',
-    'SmartServerBranchRequestSetParentLocation')
+    'SmartServerBranchRequestSetParentLocation', info='idem')
 request_handlers.register_lazy(
-    'Branch.unlock', 'bzrlib.smart.branch', 'SmartServerBranchRequestUnlock')
+    'Branch.unlock', 'bzrlib.smart.branch', 'SmartServerBranchRequestUnlock',
+    info='semi')
 request_handlers.register_lazy(
     'BzrDir.cloning_metadir', 'bzrlib.smart.bzrdir',
-    'SmartServerBzrDirRequestCloningMetaDir')
+    'SmartServerBzrDirRequestCloningMetaDir', info='read')
 request_handlers.register_lazy(
     'BzrDir.create_branch', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestCreateBranch')
+    'SmartServerRequestCreateBranch', info='semi')
 request_handlers.register_lazy(
     'BzrDir.create_repository', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestCreateRepository')
+    'SmartServerRequestCreateRepository', info='semi')
 request_handlers.register_lazy(
     'BzrDir.find_repository', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestFindRepositoryV1')
+    'SmartServerRequestFindRepositoryV1', info='read')
 request_handlers.register_lazy(
     'BzrDir.find_repositoryV2', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestFindRepositoryV2')
+    'SmartServerRequestFindRepositoryV2', info='read')
 request_handlers.register_lazy(
     'BzrDir.find_repositoryV3', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestFindRepositoryV3')
+    'SmartServerRequestFindRepositoryV3', info='read')
 request_handlers.register_lazy(
     'BzrDir.get_config_file', 'bzrlib.smart.bzrdir',
-    'SmartServerBzrDirRequestConfigFile')
+    'SmartServerBzrDirRequestConfigFile', info='read')
 request_handlers.register_lazy(
     'BzrDirFormat.initialize', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestInitializeBzrDir')
+    'SmartServerRequestInitializeBzrDir', info='semi')
 request_handlers.register_lazy(
     'BzrDirFormat.initialize_ex_1.16', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestBzrDirInitializeEx')
-request_handlers.register_lazy(
-    'BzrDir.open', 'bzrlib.smart.bzrdir', 'SmartServerRequestOpenBzrDir')
-request_handlers.register_lazy(
-    'BzrDir.open_2.1', 'bzrlib.smart.bzrdir', 'SmartServerRequestOpenBzrDir_2_1')
+    'SmartServerRequestBzrDirInitializeEx', info='semi')
+request_handlers.register_lazy(
+    'BzrDir.open', 'bzrlib.smart.bzrdir', 'SmartServerRequestOpenBzrDir',
+    info='read')
+request_handlers.register_lazy(
+    'BzrDir.open_2.1', 'bzrlib.smart.bzrdir',
+    'SmartServerRequestOpenBzrDir_2_1', info='read')
 request_handlers.register_lazy(
     'BzrDir.open_branch', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestOpenBranch')
+    'SmartServerRequestOpenBranch', info='read')
 request_handlers.register_lazy(
     'BzrDir.open_branchV2', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestOpenBranchV2')
+    'SmartServerRequestOpenBranchV2', info='read')
 request_handlers.register_lazy(
     'BzrDir.open_branchV3', 'bzrlib.smart.bzrdir',
-    'SmartServerRequestOpenBranchV3')
-request_handlers.register_lazy(
-    'delete', 'bzrlib.smart.vfs', 'DeleteRequest')
-request_handlers.register_lazy(
-    'get', 'bzrlib.smart.vfs', 'GetRequest')
-request_handlers.register_lazy(
-    'get_bundle', 'bzrlib.smart.request', 'GetBundleRequest')
-request_handlers.register_lazy(
-    'has', 'bzrlib.smart.vfs', 'HasRequest')
-request_handlers.register_lazy(
-    'hello', 'bzrlib.smart.request', 'HelloRequest')
-request_handlers.register_lazy(
-    'iter_files_recursive', 'bzrlib.smart.vfs', 'IterFilesRecursiveRequest')
-request_handlers.register_lazy(
-    'list_dir', 'bzrlib.smart.vfs', 'ListDirRequest')
-request_handlers.register_lazy(
-    'mkdir', 'bzrlib.smart.vfs', 'MkdirRequest')
-request_handlers.register_lazy(
-    'move', 'bzrlib.smart.vfs', 'MoveRequest')
-request_handlers.register_lazy(
-    'put', 'bzrlib.smart.vfs', 'PutRequest')
-request_handlers.register_lazy(
-    'put_non_atomic', 'bzrlib.smart.vfs', 'PutNonAtomicRequest')
-request_handlers.register_lazy(
-    'readv', 'bzrlib.smart.vfs', 'ReadvRequest')
-request_handlers.register_lazy(
-    'rename', 'bzrlib.smart.vfs', 'RenameRequest')
+    'SmartServerRequestOpenBranchV3', info='read')
+request_handlers.register_lazy(
+    'delete', 'bzrlib.smart.vfs', 'DeleteRequest', info='semivfs')
+request_handlers.register_lazy(
+    'get', 'bzrlib.smart.vfs', 'GetRequest', info='read')
+request_handlers.register_lazy(
+    'get_bundle', 'bzrlib.smart.request', 'GetBundleRequest', info='read')
+request_handlers.register_lazy(
+    'has', 'bzrlib.smart.vfs', 'HasRequest', info='read')
+request_handlers.register_lazy(
+    'hello', 'bzrlib.smart.request', 'HelloRequest', info='read')
+request_handlers.register_lazy(
+    'iter_files_recursive', 'bzrlib.smart.vfs', 'IterFilesRecursiveRequest',
+    info='read')
+request_handlers.register_lazy(
+    'list_dir', 'bzrlib.smart.vfs', 'ListDirRequest', info='read')
+request_handlers.register_lazy(
+    'mkdir', 'bzrlib.smart.vfs', 'MkdirRequest', info='semivfs')
+request_handlers.register_lazy(
+    'move', 'bzrlib.smart.vfs', 'MoveRequest', info='semivfs')
+request_handlers.register_lazy(
+    'put', 'bzrlib.smart.vfs', 'PutRequest', info='idem')
+request_handlers.register_lazy(
+    'put_non_atomic', 'bzrlib.smart.vfs', 'PutNonAtomicRequest', info='idem')
+request_handlers.register_lazy(
+    'readv', 'bzrlib.smart.vfs', 'ReadvRequest', info='read')
+request_handlers.register_lazy(
+    'rename', 'bzrlib.smart.vfs', 'RenameRequest', info='semivfs')
 request_handlers.register_lazy(
     'PackRepository.autopack', 'bzrlib.smart.packrepository',
-    'SmartServerPackRepositoryAutopack')
-request_handlers.register_lazy('Repository.gather_stats',
-                               'bzrlib.smart.repository',
-                               'SmartServerRepositoryGatherStats')
-request_handlers.register_lazy('Repository.get_parent_map',
-                               'bzrlib.smart.repository',
-                               'SmartServerRepositoryGetParentMap')
-request_handlers.register_lazy(
-    'Repository.get_revision_graph', 'bzrlib.smart.repository', 'SmartServerRepositoryGetRevisionGraph')
-request_handlers.register_lazy(
-    'Repository.has_revision', 'bzrlib.smart.repository', 'SmartServerRequestHasRevision')
-request_handlers.register_lazy(
-    'Repository.insert_stream', 'bzrlib.smart.repository', 'SmartServerRepositoryInsertStream')
-request_handlers.register_lazy(
-    'Repository.insert_stream_1.19', 'bzrlib.smart.repository', 'SmartServerRepositoryInsertStream_1_19')
-request_handlers.register_lazy(
-    'Repository.insert_stream_locked', 'bzrlib.smart.repository', 'SmartServerRepositoryInsertStreamLocked')
-request_handlers.register_lazy(
-    'Repository.is_shared', 'bzrlib.smart.repository', 'SmartServerRepositoryIsShared')
-request_handlers.register_lazy(
-    'Repository.lock_write', 'bzrlib.smart.repository', 'SmartServerRepositoryLockWrite')
+    'SmartServerPackRepositoryAutopack', info='idem')
+request_handlers.register_lazy(
+    'Repository.gather_stats', 'bzrlib.smart.repository',
+    'SmartServerRepositoryGatherStats', info='read')
+request_handlers.register_lazy(
+    'Repository.get_parent_map', 'bzrlib.smart.repository',
+    'SmartServerRepositoryGetParentMap', info='read')
+request_handlers.register_lazy(
+    'Repository.get_revision_graph', 'bzrlib.smart.repository',
+    'SmartServerRepositoryGetRevisionGraph', info='read')
+request_handlers.register_lazy(
+    'Repository.has_revision', 'bzrlib.smart.repository',
+    'SmartServerRequestHasRevision', info='read')
+request_handlers.register_lazy(
+    'Repository.insert_stream', 'bzrlib.smart.repository',
+    'SmartServerRepositoryInsertStream', info='stream')
+request_handlers.register_lazy(
+    'Repository.insert_stream_1.19', 'bzrlib.smart.repository',
+    'SmartServerRepositoryInsertStream_1_19', info='stream')
+request_handlers.register_lazy(
+    'Repository.insert_stream_locked', 'bzrlib.smart.repository',
+    'SmartServerRepositoryInsertStreamLocked', info='stream')
+request_handlers.register_lazy(
+    'Repository.is_shared', 'bzrlib.smart.repository',
+    'SmartServerRepositoryIsShared', info='read')
+request_handlers.register_lazy(
+    'Repository.lock_write', 'bzrlib.smart.repository',
+    'SmartServerRepositoryLockWrite', info='semi')
 request_handlers.register_lazy(
     'Repository.set_make_working_trees', 'bzrlib.smart.repository',
-    'SmartServerRepositorySetMakeWorkingTrees')
+    'SmartServerRepositorySetMakeWorkingTrees', info='idem')
 request_handlers.register_lazy(
-    'Repository.unlock', 'bzrlib.smart.repository', 'SmartServerRepositoryUnlock')
+    'Repository.unlock', 'bzrlib.smart.repository',
+    'SmartServerRepositoryUnlock', info='semi')
 request_handlers.register_lazy(
     'Repository.get_rev_id_for_revno', 'bzrlib.smart.repository',
-    'SmartServerRepositoryGetRevIdForRevno')
+    'SmartServerRepositoryGetRevIdForRevno', info='read')
 request_handlers.register_lazy(
     'Repository.get_stream', 'bzrlib.smart.repository',
-    'SmartServerRepositoryGetStream')
+    'SmartServerRepositoryGetStream', info='read')
 request_handlers.register_lazy(
     'Repository.get_stream_1.19', 'bzrlib.smart.repository',
-    'SmartServerRepositoryGetStream_1_19')
+    'SmartServerRepositoryGetStream_1_19', info='read')
 request_handlers.register_lazy(
     'Repository.tarball', 'bzrlib.smart.repository',
-    'SmartServerRepositoryTarball')
-request_handlers.register_lazy(
-    'rmdir', 'bzrlib.smart.vfs', 'RmdirRequest')
-request_handlers.register_lazy(
-    'stat', 'bzrlib.smart.vfs', 'StatRequest')
-request_handlers.register_lazy(
-    'Transport.is_readonly', 'bzrlib.smart.request', 'SmartServerIsReadonly')
+    'SmartServerRepositoryTarball', info='read')
+request_handlers.register_lazy(
+    'rmdir', 'bzrlib.smart.vfs', 'RmdirRequest', info='semivfs')
+request_handlers.register_lazy(
+    'stat', 'bzrlib.smart.vfs', 'StatRequest', info='read')
+request_handlers.register_lazy(
+    'Transport.is_readonly', 'bzrlib.smart.request', 'SmartServerIsReadonly',
+    info='read')

=== modified file 'bzrlib/tests/test_smart.py'
--- bzrlib/tests/test_smart.py	2011-08-19 22:34:02 +0000
+++ bzrlib/tests/test_smart.py	2011-10-10 15:16:30 +0000
@@ -1860,8 +1860,11 @@
         """All registered request_handlers can be found."""
         # If there's a typo in a register_lazy call, this loop will fail with
         # an AttributeError.
-        for key, item in smart_req.request_handlers.iteritems():
-            pass
+        for key in smart_req.request_handlers.keys():
+            try:
+                item = smart_req.request_handlers.get(key)
+            except AttributeError, e:
+                raise AttributeError('failed to get %s: %s' % (key, e))
 
     def assertHandlerEqual(self, verb, handler):
         self.assertEqual(smart_req.request_handlers.get(verb), handler)

=== modified file 'bzrlib/tests/test_smart_request.py'
--- bzrlib/tests/test_smart_request.py	2011-08-19 22:34:02 +0000
+++ bzrlib/tests/test_smart_request.py	2011-10-10 15:16:30 +0000
@@ -118,6 +118,16 @@
         self.assertEqual(
             [[transport]] * 3, handler._command.jail_transports_log)
 
+    def test_all_registered_requests_are_safety_qualified(self):
+        unclassified_requests = []
+        allowed_info = ('read', 'idem', 'mutate', 'semivfs', 'semi', 'stream')
+        for key in request.request_handlers.keys():
+            info = request.request_handlers.get_info(key)
+            if info is None or info not in allowed_info:
+                unclassified_requests.append(key)
+        if unclassified_requests:
+            self.fail('These requests were not categorized as safe/unsafe'
+                      ' to retry: %s'  % (unclassified_requests,))
 
 
 class TestSmartRequestHandlerErrorTranslation(TestCase):

=== modified file 'bzrlib/tests/test_smart_transport.py'
--- bzrlib/tests/test_smart_transport.py	2011-09-23 20:24:51 +0000
+++ bzrlib/tests/test_smart_transport.py	2011-10-10 15:16:30 +0000
@@ -19,8 +19,10 @@
 # all of this deals with byte strings so this is safe
 from cStringIO import StringIO
 import doctest
+import errno
 import os
 import socket
+import subprocess
 import sys
 import threading
 import time
@@ -30,6 +32,7 @@
 import bzrlib
 from bzrlib import (
         bzrdir,
+        debug,
         errors,
         osutils,
         tests,
@@ -59,6 +62,14 @@
         )
 
 
+def create_file_pipes():
+    r, w = os.pipe()
+    # These must be opened without buffering, or we get undefined results
+    rf = os.fdopen(r, 'rb', 0)
+    wf = os.fdopen(w, 'wb', 0)
+    return rf, wf
+
+
 def portable_socket_pair():
     """Return a pair of TCP sockets connected to each other.
 
@@ -88,6 +99,27 @@
         return StringIOSSHConnection(self)
 
 
+class FirstRejectedStringIOSSHVendor(StringIOSSHVendor):
+    """The first connection will be considered closed.
+
+    The second connection will succeed normally.
+    """
+
+    def __init__(self, read_from, write_to, fail_at_write=True):
+        super(FirstRejectedStringIOSSHVendor, self).__init__(read_from,
+            write_to)
+        self.fail_at_write = fail_at_write
+        self._first = True
+
+    def connect_ssh(self, username, password, host, port, command):
+        self.calls.append(('connect_ssh', username, password, host, port,
+            command))
+        if self._first:
+            self._first = False
+            return ClosedSSHConnection(self)
+        return StringIOSSHConnection(self)
+
+
 class StringIOSSHConnection(ssh.SSHConnection):
     """A SSH connection that uses StringIO to buffer writes and answer reads."""
 
@@ -103,6 +135,29 @@
         return 'pipes', (self.vendor.read_from, self.vendor.write_to)
 
 
+class ClosedSSHConnection(ssh.SSHConnection):
+    """An SSH connection that just has closed channels."""
+
+    def __init__(self, vendor):
+        self.vendor = vendor
+
+    def close(self):
+        self.vendor.calls.append(('close', ))
+
+    def get_sock_or_pipes(self):
+        # We create matching pipes, and then close the ssh side
+        bzr_read, ssh_write = create_file_pipes()
+        # We always fail when bzr goes to read
+        ssh_write.close()
+        if self.vendor.fail_at_write:
+            # If set, we'll also fail when bzr goes to write
+            ssh_read, bzr_write = create_file_pipes()
+            ssh_read.close()
+        else:
+            bzr_write = self.vendor.write_to
+        return 'pipes', (bzr_read, bzr_write)
+
+
 class _InvalidHostnameFeature(features.Feature):
     """Does 'non_existent.invalid' fail to resolve?
 
@@ -198,6 +253,91 @@
         client_medium._accept_bytes('abc')
         self.assertEqual('abc', output.getvalue())
 
+    def test_simple_pipes__accept_bytes_subprocess_closed(self):
+        # It is unfortunate that we have to use Popen for this. However,
+        # os.pipe() does not behave the same as subprocess.Popen().
+        # On Windows, if you use os.pipe() and close the write side,
+        # read.read() hangs. On Linux, read.read() returns the empty string.
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        self.assertEqual('abc', client_medium._read_bytes(3))
+        p.wait()
+        # While writing to the underlying pipe,
+        #   Windows py2.6.6 we get IOError(EINVAL)
+        #   Lucid py2.6.5, we get IOError(EPIPE)
+        # In both cases, it should be wrapped to ConnectionReset
+        self.assertRaises(errors.ConnectionReset,
+                          client_medium._accept_bytes, 'more')
+
+    def test_simple_pipes__accept_bytes_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            None, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        self.assertEqual('abc\n', child_read.read(4))
+        # While writing to the underlying pipe,
+        #   Windows py2.6.6 we get IOError(EINVAL)
+        #   Lucid py2.6.5, we get IOError(EPIPE)
+        # In both cases, it should be wrapped to ConnectionReset
+        child_read.close()
+        self.assertRaises(errors.ConnectionReset,
+                          client_medium._accept_bytes, 'more')
+
+    def test_simple_pipes__flush_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            None, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        child_read.close()
+        # Even though the pipe is closed, flush on the write side seems to be a
+        # no-op, rather than a failure.
+        client_medium._flush()
+
+    def test_simple_pipes__flush_subprocess_closed(self):
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        p.wait()
+        # Even though the child process is dead, flush seems to be a no-op.
+        client_medium._flush()
+
+    def test_simple_pipes__read_bytes_pipe_closed(self):
+        child_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(
+            child_read, client_write, 'base')
+        client_medium._accept_bytes('abc\n')
+        client_write.close()
+        self.assertEqual('abc\n', client_medium._read_bytes(4))
+        self.assertEqual('', client_medium._read_bytes(4))
+
+    def test_simple_pipes__read_bytes_subprocess_closed(self):
+        p = subprocess.Popen([sys.executable, '-c',
+            'import sys\n'
+            'if sys.platform == "win32":\n'
+            '    import msvcrt, os\n'
+            '    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'
+            '    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)\n'
+            'sys.stdout.write(sys.stdin.read(4))\n'
+            'sys.stdout.close()\n'],
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        client_medium = medium.SmartSimplePipesClientMedium(
+            p.stdout, p.stdin, 'base')
+        client_medium._accept_bytes('abc\n')
+        p.wait()
+        self.assertEqual('abc\n', client_medium._read_bytes(4))
+        self.assertEqual('', client_medium._read_bytes(4))
+
     def test_simple_pipes_client_disconnect_does_nothing(self):
         # calling disconnect does nothing.
         input = StringIO()
@@ -585,6 +725,28 @@
         request.finished_reading()
         self.assertRaises(errors.ReadingCompleted, request.read_bytes, None)
 
+    def test_reset(self):
+        server_sock, client_sock = portable_socket_pair()
+        # TODO: Use SmartClientAlreadyConnectedSocketMedium for the versions of
+        #       bzr where it exists.
+        client_medium = medium.SmartTCPClientMedium(None, None, None)
+        client_medium._socket = client_sock
+        client_medium._connected = True
+        req = client_medium.get_request()
+        self.assertRaises(errors.TooManyConcurrentRequests,
+            client_medium.get_request)
+        client_medium.reset()
+        # The stream should be reset, marked as disconnected, though ready for
+        # us to make a new request
+        self.assertFalse(client_medium._connected)
+        self.assertIs(None, client_medium._socket)
+        try:
+            self.assertEqual('', client_sock.recv(1))
+        except socket.error, e:
+            if e.errno not in (errno.EBADF,):
+                raise
+        req = client_medium.get_request()
+
 
 class RemoteTransportTests(test_smart.TestCaseWithSmartMedium):
 
@@ -3101,6 +3263,33 @@
             'e', # end
             output.getvalue())
 
+    def test_records_start_of_body_stream(self):
+        requester, output = self.make_client_encoder_and_output()
+        requester.set_headers({})
+        in_stream = [False]
+        def stream_checker():
+            self.assertTrue(requester.body_stream_started)
+            in_stream[0] = True
+            yield 'content'
+        flush_called = []
+        orig_flush = requester.flush
+        def tracked_flush():
+            flush_called.append(in_stream[0])
+            if in_stream[0]:
+                self.assertTrue(requester.body_stream_started)
+            else:
+                self.assertFalse(requester.body_stream_started)
+            return orig_flush()
+        requester.flush = tracked_flush
+        requester.call_with_body_stream(('one arg',), stream_checker())
+        self.assertEqual(
+            'bzr message 3 (bzr 1.6)\n' # protocol version
+            '\x00\x00\x00\x02de' # headers
+            's\x00\x00\x00\x0bl7:one arge' # args
+            'b\x00\x00\x00\x07content' # body
+            'e', output.getvalue())
+        self.assertEqual([False, True, True], flush_called)
+
 
 class StubMediumRequest(object):
     """A stub medium request that tracks the number of times accept_bytes is
@@ -3526,6 +3715,193 @@
         # encoder.
 
 
+class Test_SmartClientRequest(tests.TestCase):
+
+    def make_client_with_failing_medium(self, fail_at_write=True, response=''):
+        response_io = StringIO(response)
+        output = StringIO()
+        vendor = FirstRejectedStringIOSSHVendor(response_io, output,
+                    fail_at_write=fail_at_write)
+        ssh_params = medium.SSHParams('a host', 'a port', 'a user', 'a pass')
+        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)
+        smart_client = client._SmartClient(client_medium, headers={})
+        return output, vendor, smart_client
+
+    def make_response(self, args, body=None, body_stream=None):
+        response_io = StringIO()
+        response = _mod_request.SuccessfulSmartServerResponse(args, body=body,
+            body_stream=body_stream)
+        responder = protocol.ProtocolThreeResponder(response_io.write)
+        responder.send_response(response)
+        return response_io.getvalue()
+
+    def test__call_doesnt_retry_append(self):
+        response = self.make_response(('appended', '8'))
+        output, vendor, smart_client = self.make_client_with_failing_medium(
+            fail_at_write=False, response=response)
+        smart_request = client._SmartClientRequest(smart_client, 'append',
+            ('foo', ''), body='content\n')
+        self.assertRaises(errors.ConnectionReset, smart_request._call, 3)
+
+    def test__call_retries_get_bytes(self):
+        response = self.make_response(('ok',), 'content\n')
+        output, vendor, smart_client = self.make_client_with_failing_medium(
+            fail_at_write=False, response=response)
+        smart_request = client._SmartClientRequest(smart_client, 'get',
+            ('foo',))
+        response, response_handler = smart_request._call(3)
+        self.assertEqual(('ok',), response)
+        self.assertEqual('content\n', response_handler.read_body_bytes())
+
+    def test__call_noretry_get_bytes(self):
+        debug.debug_flags.add('noretry')
+        response = self.make_response(('ok',), 'content\n')
+        output, vendor, smart_client = self.make_client_with_failing_medium(
+            fail_at_write=False, response=response)
+        smart_request = client._SmartClientRequest(smart_client, 'get',
+            ('foo',))
+        self.assertRaises(errors.ConnectionReset, smart_request._call, 3)
+
+    def test__send_no_retry_pipes(self):
+        client_read, server_write = create_file_pipes()
+        server_read, client_write = create_file_pipes()
+        client_medium = medium.SmartSimplePipesClientMedium(client_read,
+            client_write, base='/')
+        smart_client = client._SmartClient(client_medium)
+        smart_request = client._SmartClientRequest(smart_client,
+            'hello', ())
+        # Close the server side
+        server_read.close()
+        encoder, response_handler = smart_request._construct_protocol(3)
+        self.assertRaises(errors.ConnectionReset,
+            smart_request._send_no_retry, encoder)
+
+    def test__send_read_response_sockets(self):
+        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        listen_sock.bind(('127.0.0.1', 0))
+        listen_sock.listen(1)
+        host, port = listen_sock.getsockname()
+        client_medium = medium.SmartTCPClientMedium(host, port, '/')
+        client_medium._ensure_connection()
+        smart_client = client._SmartClient(client_medium)
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        # Accept the connection, but don't actually talk to the client.
+        server_sock, _ = listen_sock.accept()
+        server_sock.close()
+        # Sockets buffer and don't really notice that the server has closed the
+        # connection until we try to read again.
+        handler = smart_request._send(3)
+        self.assertRaises(errors.ConnectionReset,
+            handler.read_response_tuple, expect_body=False)
+
+    def test__send_retries_on_write(self):
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        handler = smart_request._send(3)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloee',
+                         output.getvalue())
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+             ('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+            ],
+            vendor.calls)
+
+    def test__send_doesnt_retry_read_failure(self):
+        output, vendor, smart_client = self.make_client_with_failing_medium(
+            fail_at_write=False)
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        handler = smart_request._send(3)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloee',
+                         output.getvalue())
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+            ],
+            vendor.calls)
+        self.assertRaises(errors.ConnectionReset, handler.read_response_tuple)
+
+    def test__send_request_retries_body_stream_if_not_started(self):
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', (),
+            body_stream=['a', 'b'])
+        response_handler = smart_request._send(3)
+        # We connect, get disconnected, and notice before consuming the stream,
+        # so we try again one time and succeed.
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+             ('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+            ],
+            vendor.calls)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloe'
+                         'b\x00\x00\x00\x01a'
+                         'b\x00\x00\x00\x01b'
+                         'e',
+                         output.getvalue())
+
+    def test__send_request_stops_if_body_started(self):
+        # We intentionally use the python StringIO so that we can subclass it.
+        from StringIO import StringIO
+        response = StringIO()
+
+        class FailAfterFirstWrite(StringIO):
+            """Allow one 'write' call to pass, fail the rest"""
+            def __init__(self):
+                StringIO.__init__(self)
+                self._first = True
+
+            def write(self, s):
+                if self._first:
+                    self._first = False
+                    return StringIO.write(self, s)
+                raise IOError(errno.EINVAL, 'invalid file handle')
+        output = FailAfterFirstWrite()
+
+        vendor = FirstRejectedStringIOSSHVendor(response, output,
+            fail_at_write=False)
+        ssh_params = medium.SSHParams('a host', 'a port', 'a user', 'a pass')
+        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)
+        smart_client = client._SmartClient(client_medium, headers={})
+        smart_request = client._SmartClientRequest(smart_client, 'hello', (),
+            body_stream=['a', 'b'])
+        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
+        # We connect, and manage to get to the point that we start consuming
+        # the body stream. The next write fails, so we just stop.
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+            ],
+            vendor.calls)
+        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
+                         '\x00\x00\x00\x02de'   # empty headers
+                         's\x00\x00\x00\tl5:helloe',
+                         output.getvalue())
+
+    def test__send_disabled_retry(self):
+        debug.debug_flags.add('noretry')
+        output, vendor, smart_client = self.make_client_with_failing_medium()
+        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
+        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
+        self.assertEqual(
+            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
+              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
+             ('close',),
+            ],
+            vendor.calls)
+
+
 class LengthPrefixedBodyDecoder(tests.TestCase):
 
     # XXX: TODO: make accept_reading_trailer invoke translate_response or

=== modified file 'doc/en/release-notes/bzr-2.1.txt'
--- doc/en/release-notes/bzr-2.1.txt	2011-08-20 09:28:27 +0000
+++ doc/en/release-notes/bzr-2.1.txt	2011-10-10 15:16:30 +0000
@@ -43,6 +43,11 @@
 
   (John Arbash Meinel, #609187, #812928)
 
+* Teach the bzr client how to reconnect if we get ``ConnectionReset``
+  while making an RPC request. This doesn't handle all possible network
+  disconnects, but it should at least handle when the server is asked to
+  shutdown gracefully. (John Arbash Meinel, #819604)
+
 Improvements
 ************
 

=== modified file 'doc/en/release-notes/bzr-2.2.txt'
--- doc/en/release-notes/bzr-2.2.txt	2011-09-09 13:30:12 +0000
+++ doc/en/release-notes/bzr-2.2.txt	2011-10-10 15:16:30 +0000
@@ -19,6 +19,11 @@
 Bug Fixes
 *********
 
+* Teach the bzr client how to reconnect if we get ``ConnectionReset``
+  while making an RPC request. This doesn't handle all possible network
+  disconnects, but it should at least handle when the server is asked to
+  shutdown gracefully. (John Arbash Meinel, #819604)
+
 Improvements
 ************
 

=== modified file 'doc/en/release-notes/bzr-2.3.txt'
--- doc/en/release-notes/bzr-2.3.txt	2011-09-09 13:30:12 +0000
+++ doc/en/release-notes/bzr-2.3.txt	2011-10-10 15:16:30 +0000
@@ -35,6 +35,11 @@
 * Cope cleanly with buggy HTTP proxies that close the socket in the middle
   of a multipart response.  (Martin Pool, #198646).
 
+* Teach the bzr client how to reconnect if we get ``ConnectionReset``
+  while making an RPC request. This doesn't handle all possible network
+  disconnects, but it should at least handle when the server is asked to
+  shutdown gracefully. (John Arbash Meinel, #819604)
+
 Documentation
 *************
 

=== modified file 'doc/en/release-notes/bzr-2.4.txt'
--- doc/en/release-notes/bzr-2.4.txt	2011-10-10 13:59:22 +0000
+++ doc/en/release-notes/bzr-2.4.txt	2011-10-10 15:16:30 +0000
@@ -32,8 +32,8 @@
 Bug Fixes
 *********
 
-* Fixed an infinite loop when creating a repo at the root of the filesystem, 
-  i.e. "/", due to posixpath.normpath() not collapsing 2 leading slashes into 
+* Fixed an infinite loop when creating a repo at the root of the filesystem,
+  i.e. "/", due to posixpath.normpath() not collapsing 2 leading slashes into
   one, thus respecting the POSIX standard, but making relpath() loop infinitely.
   (Florian Vichot, #861008)
 
@@ -50,7 +50,12 @@
 * Return early from create_delta_index_from_delta given tiny inputs. This
   avoids raising a spurious MemoryError on certain platforms such as AIX.
   (John Arbash Meinel, #856731)
-  
+
+* Teach the bzr client how to reconnect if we get ``ConnectionReset``
+  while making an RPC request. This doesn't handle all possible network
+  disconnects, but it should at least handle when the server is asked to
+  shutdown gracefully. (John Arbash Meinel, #819604)
+
 Documentation
 *************
 

=== modified file 'doc/en/release-notes/bzr-2.5.txt'
--- doc/en/release-notes/bzr-2.5.txt	2011-10-09 13:52:06 +0000
+++ doc/en/release-notes/bzr-2.5.txt	2011-10-10 15:16:30 +0000
@@ -46,6 +46,11 @@
 * ``bzr mkdir --quiet`` now does not print a line for every created
   directory. (Martin von Gagern, #869915)
 
+* Teach the bzr client how to reconnect if we get ``ConnectionReset``
+  while making an RPC request. This doesn't handle all possible network
+  disconnects, but it should at least handle when the server is asked to
+  shutdown gracefully. (John Arbash Meinel, #819604)
+
 Documentation
 *************