queue: Don't blow up when a connection closes quickly.

zerver/lib/queue.py

@@ -217,8 +217,13 @@ class TornadoQueueClient(SimpleQueueClient):
         ioloop.IOLoop.instance().call_later(retry_secs, self._reconnect)
 
     def _on_open(self, connection: pika.connection.Connection) -> None:
-        self.connection.channel(
+        try:
-            on_open_callback = self._on_channel_open)
+            self.connection.channel(
+                on_open_callback = self._on_channel_open)
+        except pika.exceptions.ConnectionClosed:
+            # The connection didn't stay open long enough for this code to get to it.
+            # Let _on_connection_closed deal with trying again.
+            self.log.warning("TornadoQueueClient couldn't open channel: connection already closed")
 
     def _on_channel_open(self, channel: BlockingChannel) -> None:
         self.channel = channel

zerver/tests/test_queue.py

@@ -0,0 +1,18 @@
+import mock
+import os
+from typing import Any
+import ujson
+
+from pika.exceptions import ConnectionClosed
+
+from zerver.lib.queue import TornadoQueueClient
+from zerver.lib.test_classes import ZulipTestCase
+
+class TestTornadoQueueClient(ZulipTestCase):
+    @mock.patch('zerver.lib.queue.logging.getLogger', autospec=True)
+    @mock.patch('zerver.lib.queue.ExceptionFreeTornadoConnection', autospec=True)
+    def test_on_open_closed(self, mock_cxn: mock.MagicMock,
+                            mock_get_logger: mock.MagicMock) -> None:
+        connection = TornadoQueueClient()
+        connection.connection.channel.side_effect = ConnectionClosed
+        connection._on_open(mock.MagicMock())