feat(redis): lease renewal for in-flight stream entries

[sebastianwessel]

Description

Problem. With Redis Streams consumer groups and SDK-managed retries (retry_on_idle_ms), the reclaimer treats any PEL entry idle for longer than retry_on_idle_ms as abandoned. Redis resets an entry's idle time only when it is (re)delivered, not while a handler is working on it. So a handler that runs longer than retry_on_idle_ms (multi-minute LLM calls, in our case) has its entry reclaimed and redelivered through the .retry stream while the first run is still going: the same message gets processed twice, in parallel. The only workaround today is a very long retry_on_idle_ms, which delays crash recovery by the same amount.

Fix. An opt-in in-flight lease: subscribe(..., renew_lease=True) keeps renewing the entries a subscription is working on, so the reclaimer skips them. A crashed consumer stops renewing, and its entries are still reclaimed retry_on_idle_ms after the last renewal.

Also in this PR (see FastStream compatibility): consuming resumes after a lost consumer group on FastStream ≥ 0.7.6, the minimum FastStream is raised to 0.6.4 (older 0.6 releases never worked), and CI now tests the lowest supported FastStream.

Type of Change

• New feature (non-breaking change which adds functionality)
• Bug fix (non-breaking change which fixes an issue)
• Documentation update

Design

• Renewal primitive. Every renew_lease_interval_ms, one Lua script runs per chunk of up to 100 ids. For each entry it checks that the entry is still pending for this consumer (XPENDING <id> <id> 1) and still in the stream (XRANGE), then runs XCLAIM <stream> <group> <same consumer> 0 <id> JUSTID. The owner stays the same, the delivery count doesn't move (JUSTID), and the idle time resets. The reclaimer's own XCLAIM passes min-idle-time = retry_on_idle_ms, so a renewed entry is skipped even if it was already on the reclaimer's candidate list. The ownership check sits in the same script because a bare XCLAIM ... 0 would take the entry back from the reclaimer. Per-id result: renewed / lost / trimmed.
• Schedule. Renewal runs at a fixed rate: a slow round (several chunks, a timed-out call) shortens the next wait instead of pushing every later renewal back. A round slower than the whole interval still gets a short pause (10 % of the interval) rather than back-to-back rounds, and the overrun is logged once. Each Redis call is bounded by interval / 2.
• Both streams. A redelivery is a new entry on {channel}.{handler}.retry, read by its own group (<handler>-retry) and a per-process consumer. The recursive retry subscribe gets its own keeper, with that group and consumer.
• Prefetch (verified in FastStream 0.7.5). StreamSubscriber._get_msgs runs XREADGROUP COUNT=max_records and then calls consume_one for each message in turn. With max_records=None no COUNT is sent, so the whole backlog is read at once. Every entry read is already in this consumer's PEL, and the queued ones go idle behind a slow handler. This PR handles that in two ways:
  • The keeper also renews entries that are pending for this consumer and not older than the oldest run in flight (one XPENDING page of 1000 per round in practice). This covers max_records > 1 and max_workers > 1. Entries whose handler already finished (NACKed failures) are excluded, so retries still happen. Entries older than the oldest run in flight (left by an earlier process with the same consumer name, e.g. a restarted container with the same host and pid) are left for the reclaimer. For batch=True, every id in the batch is renewed while the batch handler runs.
  • renew_lease=True defaults max_records to 1 (unless it is set explicitly or batch=True). A leased entry stays with the worker that read it, so with long handlers you want an idle worker to take the next entry rather than a busy one holding it.
• Lost lease (the entry left this consumer's PEL, normally because it was reclaimed after renewals failed for longer than retry_on_idle_ms, e.g. a Redis outage): by default the handler (running in a child task) is cancelled and eggai.transport.LeaseLostError is raised in place of its result, so it can't overlap its redelivery. The entry is not acked. A batch is one unit: if any of its entries is lost the batch handler is cancelled and none is acked (the entries still owned stay pending and are retried); LeaseLostError.lost_ids names the lost entries, message_ids the whole batch. A prefetched entry found lost before its handler started raises without running the handler. With cancel_on_lease_lost=False the SDK logs an error and the handler finishes.
• Trimmed entry (still pending for this consumer but deleted from the stream by MAXLEN/XTRIM/XDEL): not a lost lease. Nothing can redeliver it, so the handler finishes, a warning is logged, and renewal stops for that id. Only an entry this consumer still owns counts as trimmed. An entry that left the PEL counts as lost even if it is gone from the stream, because with delete_on_ack the reclaimer XDELs the original when it moves it to .retry.
• Cancellation. The handler task is only ever awaited through asyncio.wait, which never raises the child's cancellation, so the wrapper tells its own cancellation (agent stop, forwarded to the handler) from an SDK-initiated one (lost lease, deadline) without Task.cancelling(), which Python 3.10 doesn't have. A handler that cancels itself still propagates CancelledError.
• Robustness. Renewal failures (timeout, connection error, NOGROUP) are logged with stream, group, consumer and ids, and retried at the next interval; they never reach the consumer loop. A message without stream ids runs unleased, with a warning. Renewal starts in connect() before the consumers start (and right away for a lease subscription added after connect()), and stops in disconnect() after broker.stop(), so handlers finishing during a graceful shutdown keep their leases. If the broker fails to start, the renewal tasks are stopped again. A subscribe that fails rolls back exactly the keepers it (and its retry subscribe) created; a second registration of the same stream/group/consumer with different lease settings is refused instead of silently replacing a keeper that is in use.
• max_processing_ms. The SDK had no handler deadline, and a lease makes one necessary: a hung handler would otherwise hold its lease forever. A handler still running at the deadline is cancelled, ProcessingTimeoutError (a TimeoutError) is raised, and the entry is NACKed. It then goes through the normal retry path and counts towards max_retries. It also works without renew_lease.
• Wrapper order. The order is delete_on_ack( lease( tracing( filters( handler ) ) ) ), so renewal stops before the entry is acked or deleted. The sync-handler path (call_handler: worker thread, awaitable results awaited) is now shared with delete_on_ack instead of being duplicated; is_async_callable also recognises functools.partial of async callables.

Options

Naming follows the SDK's existing conventions (review feedback): boolean behaviour flags are verb phrases like delete_on_ack / retry_on_error, hence renew_lease and cancel_on_lease_lost; a feature's options share its prefix like retry_on_idle_ms / retry_reclaim_interval_s / retry_backoff_*, hence renew_lease_interval_ms; every SDK-defined duration carries its unit, in ms when it is measured against retry_on_idle_ms (like retry_backoff_max_ms): the interval is bounded by and defaults to a third of it; limits are max_*.

Option	Default
renew_lease	False	Enable. Requires retry_on_idle_ms and a consumer name. Rejected with no_ack=True and AckPolicy.MANUAL (an entry acked mid-run would look like a lost lease).
renew_lease_interval_ms	retry_on_idle_ms // 3	Must be a positive int < retry_on_idle_ms. The default tolerates two failed renewals in a row. Requires renew_lease=True.
cancel_on_lease_lost	True	Must be a bool. Requires renew_lease=True.
max_processing_ms	None	Handler deadline, positive int. Requires retry_on_idle_ms.

These are validated in Agent.subscribe (at decoration time, same checks, via a shared resolve_lease_options) and in RedisTransport.subscribe, before anything is registered on the broker. True is not accepted as a millisecond value. LeaseLostError and ProcessingTimeoutError are exported from eggai.transport.

FastStream compatibility

Testing across the allowed faststream>=0.6,<0.8 range (CI only ever installed the locked version) turned up two problems at either end, fixed here:

• Lost consumer group on FastStream ≥ 0.7.6. FastStream 0.7.6 changed the stream subscriber's NOGROUP handling: it used to raise, and the task supervisor restarted the read, so reading resumed once the SDK's group monitor recreated the group. Now it stops the subscriber for good ("restart the application"). The monitor still recreated the group, but nothing read from it (test_nogroup_recovery_via_monitor and test_partial_group_loss_redelivers_existing_messages failed on 0.7.7, on main too). connect() now remembers the consumer-group stream subscribers it started, and each monitor pass, right after ensuring the groups, restarts those that are stopped and whose stop() has finished. The group exists by then, so FastStream's own XGROUP CREATE is a BUSYGROUP no-op, and a partial loss still redelivers from id 0. A failed restart is logged and retried; nothing restarts once disconnect() has begun. No-op on FastStream ≤ 0.7.5. The lock moves to FastStream 0.7.7 (fast-depends 3.0.9, nothing else) so CI covers this path; requirements.txt follows.
• Minimum FastStream is now 0.6.4 (>=0.6.4,<0.8, was >=0.6). The SDK never worked on the older 0.6 releases: on 0.6.0–0.6.2 every Redis stream subscribe raised TypeError (StreamSub has no min_idle_time before 0.6.3, and the SDK always passes it), and on 0.6.3 a min_idle_time subscription never claimed pending entries. No setup that works today breaks; the failure moves from runtime to install time. The two backoff tests that failed on 0.6.x were a test-helper issue (BinaryMessageFormatV1.encode() is sync in 0.6, async in 0.7); the SDK only uses parse(), which is sync in both.
• anyio is now a declared dependency (>=4.0,<5, the range FastStream 0.6.4 already requires, so no new constraint): the SDK imports it directly to run sync handlers in a worker thread (redis.py already did on main).
• CI: new job test-min-faststream runs the whole suite on Python 3.10 with FastStream 0.6.4 (installed over the lock) and is part of the all-checks-passed gate.

Compatibility

Lease renewal is fully opt-in. Without these options, subscriptions are unchanged: no lease client, no renewal task, no extra wrapper (a test asserts transport._lease_manager is None). No change to the reclaimer, stream keys or message shape. Requirements: Lua scripting (EVALSHA) on the server; XPENDING/XCLAIM JUSTID are Redis ≥ 5. Tested against Redis 7 (OSS).

The FastStream changes don't alter behaviour on versions where the SDK already worked (≤ 0.7.5); the subscriber restart only fires for subscribers FastStream stopped.

Limits (documented): a sync handler runs in a worker thread and can't be interrupted, so a lost lease or a missed deadline takes effect when the thread returns. A handler that blocks the event loop also blocks renewal; that includes a sync handler combined with data_type / filter_by_message, which the (pre-existing) filter wrapper calls on the event loop. Delivery stays at-least-once: the lease removes the parallel duplicate, not the redelivery after a crash.

Review findings (QualOps)

All findings from both QualOps runs were checked against the code.

First run:

• Fixed: Python 3.10 outer-cancel ambiguity (ISSUE-006), keepers added after connect() never started (007), unbounded sequential XPENDING paging (008), pruning against a stale lower bound (001), empty message ids silently unleased (002), is_async_callable missing functools.partial (003), a rollback that re-derived the retry consumer name (wrong with a custom consumer=), add() replacing a keeper whose task kept running the old one, fragile raw_message["message_ids"] access, bool/int option validation gaps.
• Not bugs: "timeout never cancels the handler" (it does, via the invocation's task), "resolve_lease_options mutates kwargs" (read-only; now documented), "min() can raise ValueError" (no await between the empty check and min()), "retry subscribe reuses the main stream's coordinates" (only the options are passed), "Redis URL stored in plaintext" (same as the rest of the transport).
• Out of scope: ISSUE-004 (the filter wrapper runs sync handlers on the event loop) predates this PR and affects all transports; documented under the lease limits.

Second run (after the first round of fixes):

• Fixed: the assert guarding group / consumer / interval (stripped by python -O) is now an explicit ValueError; a renewal round slower than the interval no longer triggers back-to-back rounds; anyio is declared as a dependency; LeaseLostError names the lost entries of a batch (lost_ids).
• Hardened, not a bug: "race between begin() and attaching the handler task" (high). Nothing awaits in that window, so the keeper can't run there; the wrapper now cancels the task right away if the invocation was already cancelled, so the window stays closed if an await is ever added.
• By design: "one lost entry fails the whole batch". A batch handler processes its entries together, so running it would duplicate the lost entry; now documented and reflected in lost_ids.

CodeQL: no alerts ("No new alerts in code changed by this pull request").

Testing

All CI jobs reproduced locally in containers, each against its own throwaway redis:7-alpine:

• lint: ruff check, ruff format --check, mypy clean (FastStream 0.7.7).
• test on Python 3.10, 3.11, 3.12, 3.13 with the lock (FastStream 0.7.7): 193 passed, 12 skipped each (Kafka, not available).
• test-min-faststream, Python 3.10 with FastStream 0.6.4: 193 passed, 12 skipped.
• build: wheel and sdist pass twine check; a clean install imports eggai, the transports and the new exceptions.
• Also checked: the suite on FastStream 0.7.5 and 0.6.7; the lease and delete_on_ack suites on Python 3.13 with FastStream 0.7.7.
• sdk/tests/test_redis_lease_renewal.py (61 tests), against real Redis:
  • A 2 s handler with retry_on_idle_ms=500 runs exactly once with the lease. While it runs, the PEL entry keeps its owner, times_delivered == 1, and idle stays below 500 ms. Without the lease the same handler runs twice, in parallel (the contrast test reproduces the bug).
  • A crashed consumer's entry (read by another consumer, never acked) is still reclaimed and handled via .retry. Stopping renewal mid-run lets the entry be reclaimed. A stale entry left under the same consumer name by an earlier process is not renewed.
  • Lost lease (the entry is claimed away mid-run): the handler is cancelled and LeaseLostError is logged. With cancel_on_lease_lost=False the handler finishes. An XDEL'd entry lets the handler finish, with a warning.
  • A retry-stream delivery (fast failure, then a slow retry) is renewed under the retry group and consumer, and runs exactly once.
  • Prefetch: max_records=5 and max_workers=2 each have a with/without contrast (redelivered without the lease, exactly once with it). The default max_records=1 is checked. Batch: every id in the batch is renewed.
  • max_processing_ms cancels a hung handler, and the entry is retried via .retry.
  • agent.stop() cancels a running handler, stops the renewal tasks and closes the client. A lease subscription added after connect() gets its renewal task immediately.
  • Rollback: a failed subscribe removes the main and retry keepers it created (also with a custom consumer=) and keeps a keeper an earlier subscribe registered.
  • The Lua script against real Redis: renewed / other owner (not stolen back) / trimmed / not pending; idle reset; delivery count unchanged.
  • Unit tests: failed and hung renewals are logged and bounded; renewal runs at a fixed rate despite slow rounds, and pauses (warning once) when a round overruns the interval; a lost entry in a batch is named in lost_ids; a run cancelled before its task exists never starts; the prefetch scan pages through a large PEL; pruning uses the state after the round; a lost status cancels only its own invocation; finished (NACKed) entries are never renewed; a queued entry lost before start raises without running; results, errors, outer cancellation (also while a timed-out handler unwinds) and a self-cancelling handler pass through the wrapper; missing message ids run unleased with a warning; is_async_callable with partials, async __call__ and AsyncMock; ProcessingTimeoutError; keeper reuse / conflict / discard; option validation on both Agent.subscribe and RedisTransport.subscribe.
• sdk/tests/test_redis.py: new unit test for the group monitor's subscriber restart (restarts stopped subscribers, leaves running and still-stopping ones alone, retries a failed restart, never restarts after disconnect() began).
• Docs: built mkdocs locally for the Redis transport page; the new section and API docstring render.

Changelog

• I have updated sdk/CHANGELOG.md under [Unreleased] section (Added: lease renewal, max_processing_ms; Changed: minimum FastStream 0.6.4, anyio declared; Fixed: NOGROUP recovery on FastStream ≥ 0.7.6)

Checklist

• My code follows the project's code style (make lint passes)
• My code is properly formatted (make format applied)
• I have added/updated tests that prove my fix/feature works
• I have added/updated documentation as needed: new "Long-Running Handlers: Lease Renewal" section in docs/docs/sdk/redis-transport.md, SDK README, and the RedisTransport.subscribe docstring (the SDK's typed option reference)
• All tests pass locally
• I have reviewed my own code
• My changes generate no new warnings
• My commit messages follow the Conventional Commits standard

Additional Notes

There is no retry example in this repository (examples live in eggai-examples), so the docs section carries the usage example.

🤖 Generated with Claude Code

[github-actions]

QualOps Code Quality Analysis

Status: ⚠️ FAILED - Critical or high severity issues found

Summary

• Total Issues: 15
• Critical: 0 🔴
• High: 5 🟠
• Medium: 6 🟡
• Low: 4 🟢
• Files Analyzed: 14

🟠 High Issues (5)

• sdk/eggai/transport/middleware_utils.py:72 - bug
  anyio.to_thread.run_sync does not accept positional arguments beyond the callable; wrapping with functools.partial is correct, but cancellable keyword is not forwarded, so long-running sync handlers cannot be cancelled by anyio's cancel scopes.

• sdk/eggai/transport/redis.py:286 - bug
  In connect(), when broker.start() raises a BaseException, the lease manager is stopped (line 289), but the _delete_client that was opened earlier (lines 260-263) is NOT closed in the same except block — it is only cleaned up in disconnect(). If broker.start() raises, disconnect() may never be called, leaking the Redis connection.

• sdk/eggai/transport/lease.py:396 - bug
  The zip(..., strict=True) call requires Python 3.10+, which may break compatibility with older supported Python versions.

• sdk/eggai/transport/lease.py:68 - bug
  XCLAIM JUSTID requires Redis ≥ 6.2, but no version check is performed; users on Redis 5.x/6.0/6.1 get opaque renewal failures instead of a clear startup error

• .github/workflows/ci.yaml:1 - bug
  _current_message_ids() depends on FastStream's private internal message representation; a FastStream minor-version change silently disables all lease tracking

🟡 Medium Issues (6)

• sdk/eggai/transport/redis.py:1163 - bug
  In _restart_stopped_subscribers (line 1163), the check getattr(sub, 'running', True) uses True as the default, meaning subscribers that lack a running attribute are treated as still running and never restarted — but the same attribute is checked with False as default in connect() (line 284), creating an inconsistency.

• .github/workflows/ci.yaml:1 - bug
  asyncio.wait_for inside renew_once() swallows CancelledError on Python < 3.12, causing stop() to block for up to one renewal interval on Python 3.10/3.11

• sdk/eggai/transport/lease.py:359 - performance
  The prefetch scan issues one xpending_range call per _SCAN_PAGE entries in a sequential loop, blocking the renewal round for the full round-trip time of each page.

...and 3 more medium issues

📊 Full Report

View detailed report

---

Powered by QualOps

[valdis]

maybe we should use a verb form for the flags name, like auto_renew_lease the same style as we have with delete_on_ack ?

[sebastianwessel]

@valdis good call, done:

• lease_renewal → renew_lease (be9ce62)
• lease_renewal_interval_ms → renew_lease_interval_ms (eb65982), so the options share a prefix like retry_* does

I skipped the auto_ prefix: the existing flags (delete_on_ack, retry_on_error) are verb forms without it, and they're just as automatic.

Final set: renew_lease, renew_lease_interval_ms, cancel_on_lease_lost, max_processing_ms.

[sebastianwessel]

Went through the QualOps findings (plus an internal review) and pushed fixes in eb65982..b28d656.

Fixed

• lease.py:570, sync handlers can't be interrupted. Real, and worse than reported: cancelling the task that awaits to_thread.run_sync returned immediately while the thread kept running, so the entry was released and redelivered in parallel. Sync handlers are now never cancelled. The wrapper waits for the thread (still leased), then raises ProcessingTimeoutError / LeaseLostError (d1cc197).
• lease.py:262, in_flight exposes the mutable dict. It now returns a read-only view (31f04f6).
• redis.py:85, unguarded stream_sub.name. Now uses getattr (31f04f6).

Not changed

• middleware_utils.py:72 (partial / run_sync kwargs): functools.partial carries the kwargs. cancellable=True would make thread overlap worse.
• lease.py:262 (lost_ids race): asyncio is single-threaded, and begin/end/_apply don't await.
• pyproject.toml:40 (<0.8): unchanged from main, only the floor moved. FastStream 0.8 doesn't exist.
• redis.py:1163 (TOCTOU on _running): disconnect() cancels and awaits the monitor before broker.stop().
• agent.py:135 (kwargs mutation): it only reads.
• redis.py:1215 (no try/except): the caller _read_ids handles errors.
• redis.py:1188 (XINFO for disabled delete_on_ack streams): pre-existing from feat(redis): opt-in delete_on_ack to XDEL entries when they are acked #297, out of scope here.

Other fixes in the same push:

• CHANGELOG entries moved to [Unreleased] after merging main (v0.6.0).
• Bounded _finished set.
• No spurious queued-lease-lost log.
• SCRIPT LOAD check on start.
• Separate CI cache key for the min-FastStream job.
• Accurate restart log.
• Lease tests now handle numeric _retry_count (fix(redis): write _retry_count and _dlq_retries as numbers #293).
• Trimmed docs/README/CHANGELOG.

203 passed / 12 skipped, also on FastStream 0.6.4.