Repository: pika/pika

Issue:
HeartbeatChecker is confused about heartbeat timeouts
cc @lukebakken, the fix should probably be back-ported to the 0.12 release candidate.

`HeartbeatChecker` constructor presently accepts an interval value and an `idle_count` which defaults to 2. `Connection` class instantiates `HeartbeatChecker` with `interval=hearbeat_timeout` and default `idle_count`.

So, if the connection is configured with a heartbeat timeout of 600 (10 minutes), it will pass 600 as the `interval` arg to `HeartbeatChecker`. So, `HearbeatChecker` will emit heartbeats to the broker only once every 600 seconds. And it will detect heartbeat timeout after 1200 seconds.

So, in the event that receipt of the heartbeat by the broker is slightly delayed (and in absence of any other AMQP frames from the client), the broker can erroneously conclude that connection with the client is lost and prematurely close the connection.

This is clearly not what was intended. `HeartbeatChecker` should be detecting a heartbeat timeout after 600 seconds of inactivity. And it should be sending a heartbeat to the broker more often than just once within the heartbeat timeout window.

I see two problems here:

1. Given `HeartbeatChecker`'s present interface, `Connection` should be instantiating it as`HeartbeatChecker(self, interval=float(self.params.heartbeat) / 2, idle_count=2) or something like that (how often does RabbitMQ broker send heartbeats within one heartbeat timeout interval?)
2. `HeartbeatChecker` is not abstracting the internals of heartbeat processing sufficiently. It's constructor should accept the heartbeat timeout value directly (no interval/idle_count business) and encapsulate the frequency of heartbeats internally without bleeding that detail to the `Connection`.


Discussion:

Patch:
diff --git a/examples/consume.py b/examples/consume.py
index da95d9e..7344149 100644
--- a/examples/consume.py
+++ b/examples/consume.py
@@ -1,17 +1,15 @@
+import functools
+import logging
 import pika
 
-def on_message(channel, method_frame, header_frame, body):
-    channel.queue_declare(queue=body, auto_delete=True)
+LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
+              '-35s %(lineno) -5d: %(message)s')
+LOGGER = logging.getLogger(__name__)
 
-    if body.startswith("queue:"):
-        queue = body.replace("queue:", "")
-        key = body + "_key"
-        print("Declaring queue %s bound with key %s" %(queue, key))
-        channel.queue_declare(queue=queue, auto_delete=True)
-        channel.queue_bind(queue=queue, exchange="test_exchange", routing_key=key)
-    else:
-        print("Message body", body)
+logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
 
+def on_message(channel, method_frame, header_frame, body, userdata=None):
+    LOGGER.info('Userdata: {} Message body: {}'.format(userdata, body))
     channel.basic_ack(delivery_tag=method_frame.delivery_tag)
 
 credentials = pika.PlainCredentials('guest', 'guest')
@@ -24,7 +22,8 @@ channel.queue_declare(queue="standard", auto_delete=True)
 channel.queue_bind(queue="standard", exchange="test_exchange", routing_key="standard_key")
 channel.basic_qos(prefetch_count=1)
 
-channel.basic_consume(on_message, 'standard')
+on_message_callback = functools.partial(on_message, userdata='on_message_userdata')
+channel.basic_consume(on_message_callback, 'standard')
 
 try:
     channel.start_consuming()
diff --git a/pika/heartbeat.py b/pika/heartbeat.py
index c02d5df..8d3d20a 100644
--- a/pika/heartbeat.py
+++ b/pika/heartbeat.py
@@ -23,13 +23,22 @@ class HeartbeatChecker(object):
         :param pika.connection.Connection: Connection object
         :param int interval: Heartbeat check interval. Note: heartbeats will
                              be sent at interval / 2 frequency.
+        :param int idle_count: The number of heartbeat intervals without data
+                               received that will close the current connection.
 
         """
         self._connection = connection
+
         # Note: see the following document:
         # https://www.rabbitmq.com/heartbeats.html#heartbeats-timeout
         self._interval = float(interval / 2)
-        self._max_idle_count = idle_count
+
+        # Note: even though we're sending heartbeats in half the specified
+        # interval, the broker will be sending them to us at the specified
+        # interval. This means we'll be checking for an idle connection
+        # twice as many times as the broker will send heartbeats to us,
+        # so we need to double the max idle count here
+        self._max_idle_count = idle_count * 2
 
         # Initialize counters
         self._bytes_received = 0
@@ -82,9 +91,12 @@ class HeartbeatChecker(object):
         been idle too long.
 
         """
-        LOGGER.debug('Received %i heartbeat frames, sent %i',
+        LOGGER.debug('Received %i heartbeat frames, sent %i, '
+                     'idle intervals %i, max idle count %i',
                      self._heartbeat_frames_received,
-                     self._heartbeat_frames_sent)
+                     self._heartbeat_frames_sent,
+                     self._idle_byte_intervals,
+                     self._max_idle_count)
 
         if self.connection_is_idle:
             return self._close_connection()


Test patch:
diff --git a/tests/unit/heartbeat_tests.py b/tests/unit/heartbeat_tests.py
index fa97338..f0431c2 100644
--- a/tests/unit/heartbeat_tests.py
+++ b/tests/unit/heartbeat_tests.py
@@ -29,7 +29,7 @@ class HeartbeatTests(unittest.TestCase):
         self.assertEqual(self.obj._interval, self.HALF_INTERVAL)
 
     def test_default_initialization_max_idle_count(self):
-        self.assertEqual(self.obj._max_idle_count, self.obj.MAX_IDLE_COUNT)
+        self.assertEqual(self.obj._max_idle_count, self.obj.MAX_IDLE_COUNT * 2)
 
     def test_constructor_assignment_connection(self):
         self.assertIs(self.obj._connection, self.mock_conn)