Add TASKS_REDELIVERED counter to track redelivered tasks

This implements issue #152 by adding a counter that tracks how many tasks
are started that were redelivered from other workers.

The implementation:
- Adds transient `redelivered` field to Execution class (defaults to False)
- Detects redeliveries in worker using Redis XAUTOCLAIM marker
- Increments TASKS_REDELIVERED counter when execution.redelivered is True

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: chrisguidry

src/docket/execution.py

@@ -51,6 +51,7 @@ def __init__(
         key: str,
         attempt: int,
         trace_context: opentelemetry.context.Context | None = None,
+        redelivered: bool = False,
     ) -> None:
         self.function = function
         self.args = args
@@ -59,6 +60,7 @@ def __init__(
         self.key = key
         self.attempt = attempt
         self.trace_context = trace_context
+        self.redelivered = redelivered
 
     def as_message(self) -> Message:
         return {
@@ -80,6 +82,7 @@ def from_message(cls, function: TaskFunction, message: Message) -> Self:
             key=message[b"key"].decode(),
             attempt=int(message[b"attempt"].decode()),
             trace_context=propagate.extract(message, getter=message_getter),
+            redelivered=False,  # Default to False, will be set to True in worker if it's a redelivery
         )
 
     def general_labels(self) -> Mapping[str, str]:

src/docket/instrumentation.py

@@ -40,6 +40,12 @@
     unit="1",
 )
 
+TASKS_REDELIVERED = meter.create_counter(
+    "docket_tasks_redelivered",
+    description="How many tasks started that were redelivered from another worker",
+    unit="1",
+)
+
 TASKS_STRICKEN = meter.create_counter(
     "docket_tasks_stricken",
     description="How many tasks have been stricken from executing",

src/docket/worker.py

@@ -47,6 +47,7 @@
     TASKS_COMPLETED,
     TASKS_FAILED,
     TASKS_PERPETUATED,
+    TASKS_REDELIVERED,
     TASKS_RETRIED,
     TASKS_RUNNING,
     TASKS_STARTED,
@@ -286,7 +287,11 @@ async def get_new_deliveries(redis: Redis) -> RedisReadGroupResponse:
                 count=available_slots,
             )
 
-        def start_task(message_id: RedisMessageID, message: RedisMessage) -> bool:
+        def start_task(
+            message_id: RedisMessageID,
+            message: RedisMessage,
+            is_redelivery: bool = False,
+        ) -> bool:
             function_name = message[b"function"].decode()
             if not (function := self.docket.tasks.get(function_name)):
                 logger.warning(
@@ -297,6 +302,7 @@ def start_task(message_id: RedisMessageID, message: RedisMessage) -> bool:
                 return False
 
             execution = Execution.from_message(function, message)
+            execution.redelivered = is_redelivery
 
             task = asyncio.create_task(self._execute(execution), name=execution.key)
             active_tasks[task] = message_id
@@ -342,12 +348,15 @@ async def ack_message(redis: Redis, message_id: RedisMessageID) -> None:
                     continue
 
                 for source in [get_redeliveries, get_new_deliveries]:
-                    for _, messages in await source(redis):
+                    for stream_key, messages in await source(redis):
+                        is_redelivery = stream_key == b"__redelivery__"
                         for message_id, message in messages:
                             if not message:  # pragma: no cover
                                 continue
 
-                            task_started = start_task(message_id, message)
+                            task_started = start_task(
+                                message_id, message, is_redelivery
+                            )
                             if not task_started:
                                 # Other errors - delete and ack
                                 await self._delete_known_task(redis, message)
@@ -521,6 +530,8 @@ async def _execute(self, execution: Execution) -> None:
         duration = 0.0
 
         TASKS_STARTED.add(1, counter_labels)
+        if execution.redelivered:
+            TASKS_REDELIVERED.add(1, counter_labels)
         TASKS_RUNNING.add(1, counter_labels)
         TASK_PUNCTUALITY.record(punctuality, counter_labels)
 

tests/test_instrumentation.py

@@ -376,6 +376,14 @@ def TASKS_RETRIED(monkeypatch: pytest.MonkeyPatch) -> Mock:
     return mock
 
 
+@pytest.fixture
+def TASKS_REDELIVERED(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Mock for the TASKS_REDELIVERED counter."""
+    mock = Mock(spec=Counter.add)
+    monkeypatch.setattr("docket.instrumentation.TASKS_REDELIVERED.add", mock)
+    return mock
+
+
 async def test_worker_execution_increments_task_counters(
     docket: Docket,
     worker: Worker,
@@ -386,6 +394,7 @@ async def test_worker_execution_increments_task_counters(
     TASKS_SUCCEEDED: Mock,
     TASKS_FAILED: Mock,
     TASKS_RETRIED: Mock,
+    TASKS_REDELIVERED: Mock,
 ):
     """Should increment the appropriate task counters when a worker executes a task."""
     await docket.add(the_task)()
@@ -397,6 +406,7 @@ async def test_worker_execution_increments_task_counters(
     TASKS_SUCCEEDED.assert_called_once_with(1, worker_labels)
     TASKS_FAILED.assert_not_called()
     TASKS_RETRIED.assert_not_called()
+    TASKS_REDELIVERED.assert_not_called()
 
 
 async def test_failed_task_increments_failure_counter(
@@ -409,6 +419,7 @@ async def test_failed_task_increments_failure_counter(
     TASKS_SUCCEEDED: Mock,
     TASKS_FAILED: Mock,
     TASKS_RETRIED: Mock,
+    TASKS_REDELIVERED: Mock,
 ):
     """Should increment the TASKS_FAILED counter when a task fails."""
     the_task.side_effect = ValueError("Womp")
@@ -422,6 +433,7 @@ async def test_failed_task_increments_failure_counter(
     TASKS_FAILED.assert_called_once_with(1, worker_labels)
     TASKS_SUCCEEDED.assert_not_called()
     TASKS_RETRIED.assert_not_called()
+    TASKS_REDELIVERED.assert_not_called()
 
 
 async def test_retried_task_increments_retry_counter(
@@ -433,6 +445,7 @@ async def test_retried_task_increments_retry_counter(
     TASKS_SUCCEEDED: Mock,
     TASKS_FAILED: Mock,
     TASKS_RETRIED: Mock,
+    TASKS_REDELIVERED: Mock,
 ):
     """Should increment the TASKS_RETRIED counter when a task is retried."""
 
@@ -448,6 +461,7 @@ async def the_task(retry: Retry = Retry(attempts=2)):
     assert TASKS_FAILED.call_count == 2
     assert TASKS_RETRIED.call_count == 1
     TASKS_SUCCEEDED.assert_not_called()
+    TASKS_REDELIVERED.assert_not_called()
 
 
 async def test_exhausted_retried_task_increments_retry_counter(
@@ -459,6 +473,7 @@ async def test_exhausted_retried_task_increments_retry_counter(
     TASKS_SUCCEEDED: Mock,
     TASKS_FAILED: Mock,
     TASKS_RETRIED: Mock,
+    TASKS_REDELIVERED: Mock,
 ):
     """Should increment the appropriate counters when retries are exhausted."""
 
@@ -474,6 +489,41 @@ async def the_task(retry: Retry = Retry(attempts=1)):
     TASKS_FAILED.assert_called_once_with(1, worker_labels)
     TASKS_RETRIED.assert_not_called()
     TASKS_SUCCEEDED.assert_not_called()
+    TASKS_REDELIVERED.assert_not_called()
+
+
+async def test_redelivered_tasks_increment_redelivered_counter(
+    docket: Docket,
+    worker_labels: dict[str, str],
+    TASKS_STARTED: Mock,
+    TASKS_COMPLETED: Mock,
+    TASKS_SUCCEEDED: Mock,
+    TASKS_FAILED: Mock,
+    TASKS_RETRIED: Mock,
+    TASKS_REDELIVERED: Mock,
+):
+    """Should increment the TASKS_REDELIVERED counter for redelivered tasks."""
+
+    async def test_task():
+        await asyncio.sleep(0.01)
+
+    await docket.add(test_task)()
+
+    worker = Worker(docket, redelivery_timeout=timedelta(milliseconds=50))
+
+    async with worker:
+        worker._execute = AsyncMock(side_effect=Exception("Simulated worker failure"))  # type: ignore[assignment]
+
+        with pytest.raises(Exception, match="Simulated worker failure"):
+            await worker.run_until_finished()
+
+    await asyncio.sleep(0.075)
+
+    worker2 = Worker(docket, redelivery_timeout=timedelta(milliseconds=100))
+    async with worker2:
+        await worker2.run_until_finished()
+
+    assert TASKS_REDELIVERED.call_count >= 1
 
 
 @pytest.fixture