Merge "Make ssync_sender send valid chunked requests"

swift/obj/ssync_sender.py

@@ -82,7 +82,6 @@ class Sender(object):
                                       set(self.send_list))
                     can_delete_obj = dict((hash_, self.available_map[hash_])
                                           for hash_ in in_sync_hashes)
-                self.disconnect()
                 if not self.failures:
                     return True, can_delete_obj
                 else:
@@ -103,6 +102,8 @@ class Sender(object):
                     self.node.get('replication_ip'),
                     self.node.get('replication_port'),
                     self.node.get('device'), self.job.get('partition'))
+            finally:
+                self.disconnect()
         except Exception:
             # We don't want any exceptions to escape our code and possibly
             # mess up the original replicator code that called us since it
@@ -351,6 +352,8 @@ class Sender(object):
         Closes down the connection to the object server once done
         with the SSYNC request.
         """
+        if not self.connection:
+            return
         try:
             with exceptions.MessageTimeout(
                     self.daemon.node_timeout, 'disconnect'):

test/unit/obj/test_ssync_receiver.py

@@ -31,7 +31,7 @@ from swift.common import utils
 from swift.common.swob import HTTPException
 from swift.obj import diskfile
 from swift.obj import server
-from swift.obj import ssync_receiver
+from swift.obj import ssync_receiver, ssync_sender
 from swift.obj.reconstructor import ObjectReconstructor
 
 from test import unit
@@ -1705,6 +1705,35 @@ class TestSsyncRxServer(unittest.TestCase):
     def tearDown(self):
         shutil.rmtree(self.tmpdir)
 
+    def test_SSYNC_disconnect(self):
+        node = {
+            'replication_ip': '127.0.0.1',
+            'replication_port': self.rx_port,
+            'device': 'sdb1',
+        }
+        job = {
+            'partition': 0,
+            'policy': POLICIES[0],
+            'device': 'sdb1',
+        }
+        sender = ssync_sender.Sender(self.daemon, node, job, ['abc'])
+
+        # kick off the sender and let the error trigger failure
+        with mock.patch('swift.obj.ssync_receiver.Receiver.initialize_request')\
+                as mock_initialize_request:
+            mock_initialize_request.side_effect = \
+                swob.HTTPInternalServerError()
+            success, _ = sender()
+        self.assertFalse(success)
+        stderr = six.StringIO()
+        with mock.patch('sys.stderr', stderr):
+            # let gc and eventlet spin a bit
+            del sender
+            for i in range(3):
+                eventlet.sleep(0)
+        self.assertNotIn('ValueError: invalid literal for int() with base 16',
+                         stderr.getvalue())
+
     def test_SSYNC_device_not_available(self):
         with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check')\
                 as mock_missing_check:

test/unit/obj/test_ssync_sender.py

@@ -70,6 +70,9 @@ class NullBufferedHTTPConnection(object):
     def getresponse(*args, **kwargs):
         pass
 
+    def close(*args, **kwargs):
+        pass
+
 
 class FakeResponse(object):