…tion

When cancelling asyncio tasks, we need to distinguish between cancellations
we initiated (timeout, cleanup) vs external cancellations (shutdown, docket.cancel()).
This uses task.cancel(msg=...) to pass sentinel messages.

Changes:
- Add _cancellation.py with CANCEL_MSG_TIMEOUT, CANCEL_MSG_CLEANUP, and
  is_our_cancellation() helper
- worker.py: Use sentinel in _run_function_with_timeout (only convert OUR
  CancelledError to TimeoutError, propagate external cancellations)
- worker.py: Use sentinel for heartbeat and cancellation listener cleanup
- strikelist.py: Use sentinel for monitor task cleanup
- _concurrency.py: Use sentinel for lease renewal task cleanup
- Add regression test for cancelling running task with Timeout dependency

Co-Authored-By: Claude Opus 4.5 <[email protected]>

The shields were added to ensure cleanup completes even under cancellation, but
they swallow CancelledErrors which can mask issues. Cleanup code now uses plain
try/except Exception blocks, allowing CancelledError to propagate if cleanup is
interrupted.

Files changed:
- _redis.py: Remove shields from __aexit__ and _cluster_pubsub
- _result_store.py: Remove shield from __aexit__
- docket.py: Remove shields from __aexit__, add try/except around strike_list

Co-Authored-By: Claude Opus 4.5 <[email protected]>

These code paths handle external cancellation during cleanup operations.
They're defensive branches that are difficult to trigger in tests but ensure
we don't accidentally swallow external CancelledErrors.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Uses AsyncExitStack for managing cleanup of Redis connections, result
storage, and strike lists. This ensures all cleanups run even if one
fails, with exceptions properly propagated.

Changes:
- RedisConnection, ResultStorage, Docket use AsyncExitStack for cleanup
- Added AsyncCloseable protocol and close_resource helper in _redis.py
- Replaced if-checks with asserts for context manager entry validation
- Removed tests for suppressed cleanup errors (behavior changed to propagate)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Refactors the worker loop to use event-based synchronization instead of
cancellation for clean shutdown. Internal tasks (`_cancellation_listener`,
`_scheduler_loop`, `_renew_leases`) now check `_worker_stopping` periodically
and use interruptible waits so they respond quickly to shutdown signals.

Key changes:
- Replaced blocking `pubsub.listen()` iteration with polling via
  `pubsub.get_message(timeout=0.1)` in the cancellation listener
- Internal loops use `asyncio.wait_for(_worker_stopping.wait(), timeout=...)`
  for interruptible sleeps
- Manual task management with `asyncio.create_task()` instead of TaskGroup
  to preserve exception propagation behavior
- Finally block cancels unresponsive infra tasks before gathering

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Switches from manual asyncio.create_task() management to TaskGroup for
the worker's internal infrastructure tasks. This provides proper structured
concurrency where tasks are tied to a scope.

When exceptions occur in the main work loop, they're now wrapped in
ExceptionGroup (the modern Python 3.11+ pattern for concurrent exceptions).
Tests updated to check for the wrapped exception using ExceptionGroup.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

These paths handle race conditions during worker shutdown where
exceptions occur simultaneously with the stopping event. They're
defensive fallbacks that are difficult to test reliably.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Plan has been fully implemented.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Registers cleanup callbacks immediately where resources are initialized
rather than grouping them in __aexit__. This follows the pattern of
registering the delattr/setattr callback BEFORE the async cleanup so
LIFO ordering ensures attributes are cleared AFTER resources close.

- _redis.py: Uses setattr(..., None) for conditional resources (cluster
  vs standalone mode)
- _result_store.py: Uses delattr for pool/client, removes | None types
- docket.py: Uses del self._stack pattern
- worker.py: Uses delattr with correct LIFO ordering
- strikelist.py: Always creates _stack in __aenter__, removes hasattr
  check since __aexit__ without __aenter__ is invalid
- _cancellation.py: Adds cancel_task() helper and sentinel constants
  for distinguishing internal vs external cancellation

Co-Authored-By: Claude Opus 4.5 <[email protected]>

- _cancellation.py line 63: only hit on 3.11+ when message propagates
- worker.py line 1181: branch for task not found in cancellation handler

Co-Authored-By: Claude Opus 4.5 <[email protected]>

When a perpetual task called `perpetual.cancel()`, `_perpetuate_if_requested`
would call `docket.cancel(execution.key)` which publishes a pub/sub signal.
The cancellation listener (running in the same worker) would receive this
signal and cancel the very task that sent it - a self-cancellation race.

This showed up as flaky coverage: line 1047 (`return False`) was covered with
the memory backend but not with real Redis, because the in-memory pub/sub
doesn't give the listener time to process the message before the call returns.

The fix uses `_cancel()` directly for Redis cleanup without the pub/sub
signal. When a perpetual task says "don't reschedule me", we're completing
normally, not cancelling a running task - we ARE that task.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

RedisConnection and StrikeList were silently returning early if
__aenter__ was called twice, but __aexit__ would then try to clean
up resources from the wrong context. Instead of supporting re-entrancy,
we now assert to fail fast on misuse.

Sequential reuse (enter, exit, enter again) is still supported.

Co-Authored-By: Claude Opus 4.5 <[email protected]>