Remove worker heartbeat compatibility code

This reverts commit d6cff30.

Author: mikeroll

docs/source/changelog.rst

@@ -19,12 +19,15 @@ Breaking Changes
 * The ``backend`` argument to the |Results| middleware is now required.
   Previously, not supplying this argument would result in a non-functional |Results| middleware.
   (`#728`_, `@LincolnPuzey`_)
+* The long-deprecated ``requeue_{deadline,interval}`` parameters of |RedisBroker|,
+  as well as the obsolete heartbeat compat code, have been removed. (`TBD`_, `@mikeroll`_)
 
 .. _#95: https://github.com/Bogdanp/dramatiq/issues/95
 .. _#345: https://github.com/Bogdanp/dramatiq/issues/345
 .. _#688: https://github.com/Bogdanp/dramatiq/pull/688
 .. _@azmeuk: https://github.com/azmeuk
 .. _#728: https://github.com/Bogdanp/dramatiq/pull/728
+.. _TBD: https://github.com/Bogdanp/dramatiq/pull/TBD
 
 Fixed
 ^^^^^

dramatiq/brokers/redis.py

@@ -20,7 +20,6 @@
 import glob
 import random
 import time
-import warnings
 from os import path
 from threading import Lock
 from typing import Optional
@@ -82,9 +81,7 @@ class RedisBroker(Broker):
         offline.
       dead_message_ttl(int): The amount of time (in ms) that
         dead-lettered messages are kept in Redis for.
-      requeue_deadline(int): Deprecated.  Does nothing.
-      requeue_interval(int): Deprecated.  Does nothing.
-      client(redis.StrictRedis): A redis client to use.
+      client(redis.Redis): A redis client to use.
       **parameters: Connection parameters are passed directly
         to :class:`redis.Redis`.
 
@@ -100,8 +97,6 @@ def __init__(
         maintenance_chance=DEFAULT_MAINTENANCE_CHANCE,
         heartbeat_timeout=DEFAULT_HEARTBEAT_TIMEOUT,
         dead_message_ttl=DEFAULT_DEAD_MESSAGE_TTL,
-        requeue_deadline=None,
-        requeue_interval=None,
         client=None,
         **parameters,
     ):
@@ -110,10 +105,6 @@ def __init__(
         if url:
             parameters["connection_pool"] = redis.ConnectionPool.from_url(url)
 
-        if requeue_deadline or requeue_interval:
-            message = "requeue_{deadline,interval} have been deprecated and no longer do anything"
-            warnings.warn(message, DeprecationWarning, stacklevel=2)
-
         self.broker_id = str(uuid4())
         self.namespace = namespace
         self.maintenance_chance = maintenance_chance

dramatiq/brokers/redis/dispatch.lua

@@ -138,18 +138,6 @@ if do_maintenance == "1" then
             redis.call("hdel", xqueue_messages, unpack(dead_message_ids_batch))
         end
     end
-
-    -- The following code is required for backwards-compatibility with
-    -- the old way acks used to be implemented.  It hoists any
-    -- existing acks zsets into the per-worker sets.
-    local compat_queue_acks = queue_full_name .. ".acks"
-    local compat_message_ids = redis.call("zrangebyscore", compat_queue_acks, 0, timestamp - 86400000 * 7.5)
-    if next(compat_message_ids) then
-        for compat_message_ids_batch in iter_chunks(compat_message_ids) do
-            redis.call("sadd", queue_acks, unpack(compat_message_ids_batch))
-            redis.call("zrem", compat_queue_acks, unpack(compat_message_ids_batch))
-        end
-    end
 end
 
 

tests/test_redis.py

@@ -332,15 +332,6 @@ def test_redis_broker_can_connect_via_client():
     assert broker.client is client
 
 
-def test_redis_broker_warns_about_deprecated_parameters():
-    # When I pass deprecated params to RedisBroker
-    # Then it should warn me that those params do nothing
-    with pytest.warns(DeprecationWarning) as record:
-        RedisBroker(requeue_deadline=1000)
-
-    assert str(record[0].message) == "requeue_{deadline,interval} have been deprecated and no longer do anything"
-
-
 def test_redis_broker_raises_attribute_error_when_given_an_invalid_attribute(redis_broker):
     # Given that I have a Redis broker
     # When I try to access an attribute that doesn't exist
@@ -349,40 +340,6 @@ def test_redis_broker_raises_attribute_error_when_given_an_invalid_attribute(red
         redis_broker.idontexist
 
 
-def test_redis_broker_maintains_backwards_compat_with_old_acks(redis_broker):
-    # Given that I have an actor
-    @dramatiq.actor
-    def do_work(self):
-        pass
-
-    # And that actor has some old-style unacked messages
-    expired_message_ids = set()
-    valid_message_ids = set()
-    for i in range(LUA_MAX_UNPACK_SIZE * 2):
-        expired_message_id = b"expired-old-school-ack-%r" % i
-        valid_message_id = b"valid-old-school-ack-%r" % i
-        expired_message_ids.add(expired_message_id)
-        valid_message_ids.add(valid_message_id)
-        if redis.__version__.startswith("2."):
-            redis_broker.client.zadd("dramatiq:default.acks", 0, expired_message_id)
-            redis_broker.client.zadd("dramatiq:default.acks", current_millis(), valid_message_id)
-        else:
-            redis_broker.client.zadd("dramatiq:default.acks", {expired_message_id: 0})
-            redis_broker.client.zadd("dramatiq:default.acks", {valid_message_id: current_millis()})
-
-    # When maintenance runs for that actor's queue
-    redis_broker.maintenance_chance = MAINTENANCE_SCALE
-    redis_broker.do_qsize(do_work.queue_name)
-
-    # Then maintenance should move the expired message to the new style acks set
-    unacked = redis_broker.client.smembers("dramatiq:__acks__.%s.default" % redis_broker.broker_id)
-    assert set(unacked) == expired_message_ids
-
-    # And the valid message should stay in that set
-    compat_unacked = redis_broker.client.zrangebyscore("dramatiq:default.acks", 0, "+inf")
-    assert set(compat_unacked) == valid_message_ids
-
-
 def test_redis_consumer_ack_can_retry_on_connection_error(redis_broker, redis_worker):
     # Given that I have an actor
     @dramatiq.actor