Make the Redis ack-emulation restore cadence configurable

[arkadiuszbach]

Summary

The Redis transport re-queues delivered-but-unacked messages whose
visibility_timeout has expired via QoS.restore_visible. How often that
scan actually runs was previously governed by two hard-coded constants:

• the event-loop restore timer fired every 10s (call_repeatedly(10, ...)), and
• restore_visible only performed a real Redis scan on every 10th call
  (a hard-coded interval=10 throttle).

The effective sweep period was therefore fixed at roughly 100 seconds,
independent of the configured visibility_timeout. Users who set a low
visibility_timeout (e.g. 30s) still had to wait up to ~100s for abandoned
messages to be restored, making the setting appear random or ignored.

This PR exposes both constants as transport options so the restore cadence can
be tuned to match the configured visibility_timeout, while keeping the
previous values as defaults (fully backward compatible).

Changes

• Add two transport options to the Redis Channel:
  • unacked_restore_interval (int, default 10) — seconds between periodic
    restore sweeps on the asynchronous (event-loop / prefork) path.
  • unacked_restore_throttle (int, default 10) — only perform an actual
    Redis scan on every N-th restore_visible call. Set to 1 to scan on
    every call.
• Wire unacked_restore_interval into register_with_event_loop
  (call_repeatedly) and unacked_restore_throttle into the
  restore_visible(interval=...) call sites (sync poll path and
  maybe_restore_messages).
• Harden restore_visible against a misconfigured throttle of 0
  (% (interval or 1)), which previously would raise ZeroDivisionError.
• Document the options, the cadence formula, the prefork vs green-pool
  path difference (see below), the shared-visibility_timeout cutoff caveat,
  and a "janitor worker" pattern for green pools in the Redis transport
  reference.

Restore behaviour differs by worker pool ⚠️

The two new options do not behave identically across pools, because the
Redis transport has two distinct code paths that trigger restore_visible:

Asynchronous path — prefork (default on Linux/macOS)

The worker drives Redis via the kombu event loop / hub. Restores are scheduled
by a dedicated timer:

loop.call_repeatedly(unacked_restore_interval, cycle.maybe_restore_messages)

Here both options apply. The effective sweep period is:

unacked_restore_interval * unacked_restore_throttle   (seconds)

So with the defaults (10 * 10) a real scan happens about every 100s — the
previous behaviour. To scan every 5s, set unacked_restore_interval=5 and
unacked_restore_throttle=1:

app.conf.broker_transport_options = {
    "visibility_timeout": 30,
    "unacked_restore_interval": 5,
    "unacked_restore_throttle": 1,
}

Synchronous path — eventlet / gevent (and Windows, or plain drain_events)

Green-pool workers do not use the event-loop restore timer. They reach
restore_visible through MultiChannelPoller.get(), which attempts a restore
on every empty poll (~brpop_timeout, 1s by default). On this path:

• unacked_restore_interval is ignored (there is no call_repeatedly
  timer), and
• unacked_restore_throttle is the only control — a real Redis scan runs
  on every N-th empty poll (default 10 → roughly every ~10s while idle).

There is a further caveat specific to gevent/eventlet: all greenlets share
a single OS thread, so a CPU-bound task that doesn't yield blocks the hub,
which also stalls the consumer loop — restores (and heartbeats) pause until the
task finishes. For deployments that need reliable restores under green pools,
the docs now describe a dedicated prefork "janitor" worker (same Redis db /
global_keyprefix, same visibility_timeout, aggressive restore cadence) that
drives restores independently of the busy green-pool workers.

Practical guidance

• Prefork: tune unacked_restore_interval (× unacked_restore_throttle).
• eventlet/gevent: tune unacked_restore_throttle only; consider a janitor
  worker for CPU-bound workloads.

Backward compatibility

Defaults reproduce the prior behaviour exactly (10s timer × 10 throttle ≈ 100s
sweep on the async path; ~10× brpop_timeout on the sync path). No change
unless the new options are set.

Related issues

• Fixes Tasks longer than the the visibility timeout are not being re-queued with acks_late and redis broker celery#6229 — "Tasks longer than the visibility timeout are not
  being re-queued with acks_late and the Redis broker." The root cause
  identified in that thread is exactly kombu's internal ~100s restore timer /
  hard-coded interval=10 throttle; this PR makes that cadence configurable.
• Addresses several comments in Visibility timeout not functioning as expected celery#9339 — "Visibility timeout not
  functioning as expected." In particular the finding that tweaking the
  kombu.transport.redis.QoS interval resolves the inconsistent redelivery
  delay, the request to expose it as configuration, and the gevent-specific
  observation that restore_visible stops being called while workers are busy
  (now documented, with the janitor-worker workaround).
  • Visibility timeout not functioning as expected celery#9339 (comment)
  • Visibility timeout not functioning as expected celery#9339 (comment)

Tests

• Updated test_on_poll_init and
  test_maybe_restore_messages_calls_restore_visible to assert the new
  interval= kwarg is forwarded.
• Added test_qos_restore_visible_interval_throttles — verifies the throttle
  gates scans (scans on the 1st and 4th call for interval=3).
• Added test_qos_restore_visible_zero_interval_no_zerodivision — verifies
  interval=0 does not raise and scans on every call.

Drafted-by: Claude Code (Opus 4.8)

[Copilot]

Pull request overview

This PR makes the Redis transport’s ack-emulation “restore visible” sweep cadence configurable so users can tune abandoned-message recovery to match their visibility_timeout, while preserving current defaults for backward compatibility.

Changes:

• Adds unacked_restore_interval and unacked_restore_throttle as Redis transport options and wires them into the async (hub timer) and sync (poll) restore paths.
• Hardens QoS.restore_visible() against a throttle of 0 to avoid ZeroDivisionError.
• Updates unit tests and extends Redis transport documentation with configuration guidance (including prefork vs green-pool behavior).

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
kombu/transport/redis.py	Adds configurable restore timer interval and scan throttling, and forwards throttle to restore_visible.
t/unit/transport/test_redis.py	Adds tests for restore throttling behavior and asserts new interval= kwarg forwarding.
docs/reference/kombu.transport.redis.rst	Documents the new options, the cadence formula, pool-specific behavior, and a janitor-worker pattern.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.90%. Comparing base (781576a) to head (ebc6335).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2558   +/-   ##
=======================================
  Coverage   82.89%   82.90%           
=======================================
  Files          79       79           
  Lines       10397    10402    +5     
  Branches     1196     1197    +1     
=======================================
+ Hits         8619     8624    +5     
  Misses       1577     1577           
  Partials      201      201           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[auvipy]

should we also consider adding integration tests for this changes?

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

Suppressed comments (5)

docs/reference/kombu.transport.redis.rst:122

• These two indented lines look like a leftover fragment of the earlier broker_transport_options example, but they now sit inside the janitor-worker note block and render as an unrelated literal block. They should be removed (or moved back into the earlier code block).

            "credential_provider": my_streaming_credential_provider,
        }

kombu/transport/redis.py:1739

• unacked_restore_interval is read from transport_options and compared to 0, which will raise TypeError if the option is present but non-numeric (e.g. None or a string). Since this option is explicitly user-configurable now, it should be coerced (or validated) and fall back to the default on invalid values.

        restore_interval = connection.client.transport_options.get(
            'unacked_restore_interval', Channel.unacked_restore_interval
        )
        if restore_interval <= 0:
            restore_interval = Channel.unacked_restore_interval

docs/reference/kombu.transport.redis.rst:53

• The broker_transport_options example dictionary is closed before the credential_provider entry, leaving "credential_provider": ... to appear later as a stray, indented literal line. This breaks the example and the surrounding reST structure.

This issue also appears on line 121 of the same file.

        app.conf.broker_transport_options = {
            "visibility_timeout": 30,
            "unacked_restore_interval": 5,
            "unacked_restore_throttle": 1,
        }

kombu/transport/redis.py:444

• restore_visible() now accepts a user-configurable throttle interval. With the current % (interval or 1) guard, interval=None is handled, but negative values still produce surprising throttling, and non-int values (e.g. strings) will raise TypeError in the modulo. It should coerce to an int and clamp to >= 1 so misconfiguration fails safe (scan every call).

    def restore_visible(self, start=0, num=10, interval=10):
        self._vrestore_count += 1
        # ``interval or 1`` avoids a ZeroDivisionError if the throttle is
        # misconfigured to 0; 1 means "scan on every call".
        if (self._vrestore_count - 1) % (interval or 1):

t/unit/transport/test_redis.py:760

• If restore_visible() is hardened to treat non-positive throttle values as "scan every call", the unit test should cover a negative interval as well (not just 0) to prevent regressions in misconfiguration handling.

    def test_qos_restore_visible_zero_interval_no_zerodivision(self):
        client = self.channel._create_client = Mock(name='client')
        client = client()

        def pipe(*args, **kwargs):
            return Pipeline(client)
        client.pipeline = pipe
        client.zrevrangebyscore.return_value = []
        qos = redis.QoS(self.channel)
        qos.restore_by_tag = Mock(name='restore_by_tag')

        # interval=0 must not raise ZeroDivisionError and scans on every call.
        qos._vrestore_count = 0
        qos.restore_visible(interval=0)
        qos.restore_visible(interval=0)
        assert client.zrevrangebyscore.call_count == 2