Repository: pika/pika

Issue:
BlockingChannel.queue_declare hanging on non-string queue parameters
Under Python 3.6.4 and Pika 0.11.2, the `BlockingChannel.queue_declare` method hangs when setting its `queue` parameter to a value that is not of `str` type (e.g., `int`, `bool`, `list`, `dict`, `tuple`, `NoneType`).

Input:

```
$ python3 <<EOF
import pika
with pika.BlockingConnection() as connection:
    channel = connection.channel()
    channel.queue_declare(queue=[1, 2, 3])
EOF
```

Discussion:
vitaly-krugl: @lukebakken, I haven't forgotten - will catch up in the next couple of days.
lukebakken: @vitaly-krugl no hurry at all! Thanks again.
lukebakken: @vitaly-krugl the `queue.declare` method never makes it to RabbitMQ. An `AssertionError` is thrown [here](https://github.com/pika/pika/blob/master/pika/spec.py#L1003-L1004) which gums up the works when the `with` clause tries to exit.
vitaly-krugl: Would this alone punch at the heart of the problem [here](https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/channel.py#L1382)?
```
try:
    self._send_method(method)
except Exception:
    self._blocking = None
    raise
```

lukebakken: I'll try that out.
vitaly-krugl: Don't try it out just yet, I missed something there (and the fix in the PR did, too, I think)
vitaly-krugl: The [if acceptable_replies:](https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/channel.py#L1361) block sets `self._blocking` and also registers a number of callbacks. If `self._send_method(method)` (as in this case), we really don't want any of those registered callbacks to remain there either.

Since a successful `self._send_method(method)` call will ultimately just enqueue some data on the output write buffer, it should be possible to move the (https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/channel.py#L1361) block after [self._send_method(method)](https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/channel.py#L1382).

Furthermore, to ensure that an incomplete message doesn't get placed in the output buffer (due to marshaling failure of one of its subframes), [Connection._send_message()](https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/connection.py#L2273) needs to be modified to pre-marshal all of its parts and then append them to the output frame buffer only after all marshaling is done, updating the stats and finally calling `self._flush_outbound() ` and `self._detect_backpressure()` like [Connection._send_frame()](https://github.com/pika/pika/blob/b7f27983cfbcbaf34a06b6fc9259a7fd50b8838d/pika/connection.py#L2251-L2257). To this end,  `Connection._send_message()` and `Connection._send_frame()` should share a method (e.g.,  `Connection._output_marshaled_frame()` that updates `self.bytes_sent` and `self.frames_sent` and appends the marshaled frame data to `self.outbound_buffer` .
lukebakken: @vitaly-krugl - ready for re-review. Thanks!
lukebakken: @vitaly-krugl - I have merged in the tests you provided and this is ready for another review. Thanks!
vitaly-krugl: @lukebakken, I renamed this PR "Request marshaling error should not corrupt a channel", which reflects issues #990 and #912 more accurately. 
vitaly-krugl: On broker's Channel.Close, the draining is necessary because ANQP says to
ignore all incoming  requests after  channel is closed except
Channel.Close. So, draining in that case helps break the gridlock.

However, in the case the client is closing the channel with some blocking
requests still pending normally, we have a perfectly healthy channel and
nothing special is needed. The normal course of events will see it through.

On Tue, Apr 10, 2018, 5:00 AM Luke Bakken <notifications@github.com> wrote:

> *@lukebakken* commented on this pull request.
> ------------------------------
>
> In pika/channel.py
> <https://github.com/pika/pika/pull/991#discussion_r180392933>:
>
> > @@ -1327,9 +1327,10 @@ def _on_synchronous_complete(self, _method_frame_unused):
>          while self._blocked and self._blocking is None:
>              self._rpc(*self._blocked.popleft())
>
> -    def _drain_blocked_methods_on_remote_close(self):
>
> If we think we need an "emergency channel-close" method that purges
> queued-up requests (I don't think we do
>
> Draining blocked methods on a broker-initiated close was introduced in
> #957 <https://github.com/pika/pika/pull/957> - please check that PR out
> again. I still think it's necessary.
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/pika/pika/pull/991#discussion_r180392933>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/ABX9KigMj1hv6PIavaR70oqFZk9LqB0iks5tnJ7igaJpZM4Sg71J>
> .
>

vitaly-krugl: @lukebakken, is this PR ready for re-review?
vitaly-krugl: I think I might not have ended that review ??

On Mon, Apr 16, 2018, 6:22 AM Luke Bakken <notifications@github.com> wrote:

> @vitaly-krugl <https://github.com/vitaly-krugl> if I re-select your name
> in the "Reviewers" dropdown, the status icon changes back to an orange disk
> ... do you not get a new email saying I re-requested a review? I assumed
> that you did. If you don't get an email, I can @-mention you in a comment.
> Thanks for the re-re-reviews 😄
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/pika/pika/pull/991#issuecomment-381597285>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/ABX9KmC9Xn-wJn2vurcgpPYZe9kJuVEtks5tpJsrgaJpZM4Sg71J>
> .
>

lukebakken: @vitaly-krugl thanks! Sorry I missed the previous comment about that test.
lukebakken: @vitaly-krugl - changes made. I'll merge this once builds complete. Thanks!

Patch:
diff --git a/pika/channel.py b/pika/channel.py
index fb67a0d..282f53c 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -1347,7 +1347,7 @@ class Channel(object):
         sent, and thus its completion callback would never be called.
 
         """
-        LOGGER.debug('Draining %i blocked frames due to remote Channel.Close',
+        LOGGER.debug('Draining %i blocked frames due to broker-requested Channel.Close',
                      len(self._blocked))
         while self._blocked:
             method = self._blocked.popleft()[0]
@@ -1408,6 +1408,12 @@ class Channel(object):
             self._blocked.append([method, callback, acceptable_replies])
             return
 
+        # Note: _send_method can throw exceptions if there are framing errors
+        # or invalid data passed in. Call it here to prevent self._blocking
+        # from being set if an exception is thrown. This also prevents
+        # acceptable_replies registering callbacks when exceptions are thrown
+        self._send_method(method)
+
         # If acceptable replies are set, add callbacks
         if acceptable_replies:
             # Block until a response frame is received for synchronous frames
@@ -1430,8 +1436,6 @@ class Channel(object):
                     self.callbacks.add(self.channel_number, reply, callback,
                                        arguments=arguments)
 
-        self._send_method(method)
-
     def _raise_if_not_open(self):
         """If channel is not in the OPEN state, raises ChannelClosed with
         `reply_code` and `reply_text` corresponding to current state. If channel
diff --git a/pika/connection.py b/pika/connection.py
index be2b1bc..271b198 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -2306,11 +2306,7 @@ class Connection(pika.compat.AbstractBase):
                 'Attempted to send a frame on closed connection.')
 
         marshaled_frame = frame_value.marshal()
-        self.bytes_sent += len(marshaled_frame)
-        self.frames_sent += 1
-        self._adapter_emit_data(marshaled_frame)
-        if self.params.backpressure_detection:
-            self._detect_backpressure()
+        self._output_marshaled_frames([marshaled_frame])
 
     def _send_method(self, channel_number, method, content=None):
         """Constructs a RPC method frame and then sends it to the broker.
@@ -2336,8 +2332,14 @@ class Connection(pika.compat.AbstractBase):
 
         """
         length = len(content[1])
-        self._send_frame(frame.Method(channel_number, method_frame))
-        self._send_frame(frame.Header(channel_number, length, content[0]))
+        marshaled_body_frames = []
+
+        # Note: we construct the Method, Header and Content objects, marshal them
+        # *then* output in case the marshaling operation throws an exception
+        frame_method = frame.Method(channel_number, method_frame)
+        frame_header = frame.Header(channel_number, length, content[0])
+        marshaled_body_frames.append(frame_method.marshal())
+        marshaled_body_frames.append(frame_header.marshal())
 
         if content[1]:
             chunks = int(math.ceil(float(length) / self._body_max_length))
@@ -2346,7 +2348,10 @@ class Connection(pika.compat.AbstractBase):
                 end = start + self._body_max_length
                 if end > length:
                     end = length
-                self._send_frame(frame.Body(channel_number, content[1][start:end]))
+                frame_body = frame.Body(channel_number, content[1][start:end])
+                marshaled_body_frames.append(frame_body.marshal())
+
+        self._output_marshaled_frames(marshaled_body_frames)
 
     def _set_connection_state(self, connection_state):
         """Set the connection state.
@@ -2382,3 +2387,16 @@ class Connection(pika.compat.AbstractBase):
         """
         self._frame_buffer = self._frame_buffer[byte_count:]
         self.bytes_received += byte_count
+
+    def _output_marshaled_frames(self, marshaled_frames):
+        """Output list of marshaled frames to buffer and update stats
+
+        :param list marshaled_frames: A list of frames marshaled to bytes
+
+        """
+        for marshaled_frame in marshaled_frames:
+            self.bytes_sent += len(marshaled_frame)
+            self.frames_sent += 1
+            self._adapter_emit_data(marshaled_frame)
+        if self.params.backpressure_detection:
+            self._detect_backpressure()


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 2a51ae1..967e109 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -625,8 +625,9 @@ class TestExchangeRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
         raise AssertionError("Should not have received an Exchange.DeclareOk")
 
 
-class TestPassiveExchangeDeclareWithConcurrentClose(AsyncTestCase, AsyncAdapters):
-    DESCRIPTION = "should close channel: declare passive exchange with close"
+class TestNoDeadlockWhenClosingChannelWithPendingBlockedRequestsAndConcurrentChannelCloseFromBroker(
+        AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "No deadlock when closing a channel with pending blocked requests and concurrent Channel.Close from broker."
 
     # To observe the behavior that this is testing, comment out this line
     # in pika/channel.py - _on_close:
@@ -636,10 +637,12 @@ class TestPassiveExchangeDeclareWithConcurrentClose(AsyncTestCase, AsyncAdapters
     # With the above line commented out, this test will hang
 
     def begin(self, channel):
-        self.name = self.__class__.__name__ + ':' + uuid.uuid1().hex
+        base_exch_name = self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.channel.add_on_close_callback(self.on_channel_closed)
         for i in range(0, 99):
-            exch_name = self.name + ':' + str(i)
+            # Passively declare a non-existent exchange to force Channel.Close
+            # from broker
+            exch_name = base_exch_name + ':' + str(i)
             cb = functools.partial(self.on_bad_result, exch_name)
             channel.exchange_declare(exch_name,
                                      exchange_type='direct',
@@ -648,15 +651,49 @@ class TestPassiveExchangeDeclareWithConcurrentClose(AsyncTestCase, AsyncAdapters
         channel.close()
 
     def on_channel_closed(self, channel, reply_code, reply_text):
+        # The close is expected because the requested exchange doesn't exist
         self.stop()
 
     def on_bad_result(self, exch_name, frame):
-        self.channel.exchange_delete(exch_name)
-        raise AssertionError("Should not have received an Exchange.DeclareOk")
+        self.fail("Should not have received an Exchange.DeclareOk")
 
 
-class TestQueueDeclareAndDelete(AsyncTestCase, AsyncAdapters):
-    DESCRIPTION = "Create and delete a queue"
+class TestClosingAChannelPermitsBlockedRequestToComplete(AsyncTestCase,
+                                                         AsyncAdapters):
+    DESCRIPTION = "Closing a channel permits blocked requests to complete."
+
+    def begin(self, channel):
+        self._queue_deleted = False
+
+        channel.add_on_close_callback(self.on_channel_closed)
+
+        q_name = self.__class__.__name__ + ':' + uuid.uuid1().hex
+        # NOTE we pass callback to make it a blocking request
+        channel.queue_declare(q_name,
+                              exclusive=True,
+                              callback=lambda _frame: None)
+
+        self.assertIsNotNone(channel._blocking)
+
+        # The Queue.Delete should block on completion of Queue.Declare
+        channel.queue_delete(q_name, callback=self.on_queue_deleted)
+        self.assertTrue(channel._blocked)
+
+        # This Channel.Close should allow the blocked Queue.Delete to complete
+        # Before closing the channel
+        channel.close()
+
+    def on_queue_deleted(self, _frame):
+        # Getting this callback shows that the blocked request was processed
+        self._queue_deleted = True
+
+    def on_channel_closed(self, _channel, _reply_code, _reply_text):
+        self.assertTrue(self._queue_deleted)
+        self.stop()
+
+
+class TestQueueUnnamedDeclareAndDelete(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "Create and delete an unnamed queue"
 
     def begin(self, channel):
         channel.queue_declare(queue='',
@@ -673,11 +710,11 @@ class TestQueueDeclareAndDelete(AsyncTestCase, AsyncAdapters):
 
     def on_queue_delete(self, frame):
         self.assertIsInstance(frame.method, spec.Queue.DeleteOk)
+        # NOTE: with event loops that suppress exceptions from callbacks
         self.stop()
 
 
-
-class TestQueueNameDeclareAndDelete(AsyncTestCase, AsyncAdapters):
+class TestQueueNamedDeclareAndDelete(AsyncTestCase, AsyncAdapters):
     DESCRIPTION = "Create and delete a named queue"
 
     def begin(self, channel):
@@ -701,7 +738,6 @@ class TestQueueNameDeclareAndDelete(AsyncTestCase, AsyncAdapters):
         self.stop()
 
 
-
 class TestQueueRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
     DESCRIPTION = "Should close chan: re-declared queue w/ diff params"
 
@@ -745,7 +781,6 @@ class TestTX1_Select(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
         self.stop()
 
 
-
 class TestTX2_Commit(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Start a transaction, and commit it"
 
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index d79ded2..d0ed48e 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -50,7 +50,6 @@ def setUpModule():
     logging.basicConfig(level=logging.DEBUG)
 
 
-#@unittest.skip('SKIPPING WHILE DEBUGGING SOME CHANGES. DO NOT MERGE LIKE THIS')
 class BlockingTestCaseBase(unittest.TestCase):
 
     TIMEOUT = DEFAULT_TIMEOUT
@@ -355,6 +354,16 @@ class TestCreateAndCloseConnectionWithChannelAndConsumer(BlockingTestCaseBase):
         self.assertFalse(ch._impl._consumers)
 
 
+class TestUsingInvalidQueueArgument(BlockingTestCaseBase):
+    def test(self):
+        """BlockingConnection raises expected exception when invalid queue parameter is used
+        """
+        connection = self._connect()
+        ch = connection.channel()
+        with self.assertRaises(AssertionError):
+            ch.queue_declare(queue=[1, 2, 3])
+
+
 class TestSuddenBrokerDisconnectBeforeChannel(BlockingTestCaseBase):
 
     def test(self):
diff --git a/tests/unit/channel_tests.py b/tests/unit/channel_tests.py
index 10e594e..dc353ef 100644
--- a/tests/unit/channel_tests.py
+++ b/tests/unit/channel_tests.py
@@ -1587,3 +1587,18 @@ class ChannelTests(unittest.TestCase):
         self.assertRaises(TypeError,
                           self.obj._validate_rpc_completion_callback,
                           'foo')
+
+    def test_no_side_effects_from_send_method_error(self):
+        self.obj._set_state(self.obj.OPEN)
+
+        self.assertIsNone(self.obj._blocking)
+
+        with mock.patch.object(self.obj.callbacks, 'add') as cb_add_mock:
+            with mock.patch.object(self.obj, '_send_method',
+                                   side_effect=TypeError) as send_method_mock:
+                with self.assertRaises(TypeError):
+                    self.obj.queue_delete('', callback=lambda _frame: None)
+
+        self.assertEqual(send_method_mock.call_count, 1)
+        self.assertIsNone(self.obj._blocking)
+        self.assertEqual(cb_add_mock.call_count, 0)
diff --git a/tests/unit/connection_tests.py b/tests/unit/connection_tests.py
index 19df873..04fd543 100644
--- a/tests/unit/connection_tests.py
+++ b/tests/unit/connection_tests.py
@@ -983,3 +983,31 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Make sure _detect_backpressure doesn't throw
         self.connection._detect_backpressure()
+
+
+    def test_no_side_effects_from_message_marshal_error(self):
+        # Verify that frame buffer is empty on entry
+        self.assertEqual(b'', self.connection._frame_buffer)
+
+        # Use Basic.Public with invalid body to trigger marshalling error
+        method = spec.Basic.Publish()
+        properties = spec.BasicProperties()
+        # Verify that marshalling of method and header won't trigger error
+        frame.Method(1, method).marshal()
+        frame.Header(1, body_size=10, props=properties).marshal()
+        # Create bogus body that should trigger an error during marshalling
+        body = [1,2,3,4]
+        # Verify that frame body can be created using the bogus body, but
+        # that marshalling will fail
+        frame.Body(1, body)
+        with self.assertRaises(TypeError):
+            frame.Body(1, body).marshal()
+
+        # Now, attempt to send the method with the bogus body
+        with self.assertRaises(TypeError):
+            self.connection._send_method(channel_number=1,
+                                    method=method,
+                                    content=(properties, body))
+
+        # Now make sure that nothing is enqueued on frame buffer
+        self.assertEqual(b'', self.connection._frame_buffer)