Add TASKS_SUPERSEDED metric and move claim() before all metrics/timing

The supersession check now happens right after the strikelist check,
before execution counts, timing, or any lifecycle metrics. This means
superseded tasks don't affect any counters at all.

A new TASKS_SUPERSEDED counter tracks how often this happens, with a
docket.where label to distinguish the two check points:
- "worker": head check in claim() before the task runs
- "on_complete": tail check in Perpetual.on_complete() after it runs

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

Author: chrisguidry

src/docket/dependencies/_perpetual.py

@@ -11,7 +11,7 @@
 if TYPE_CHECKING:  # pragma: no cover
     from ..execution import Execution
 
-from ..instrumentation import TASKS_PERPETUATED
+from ..instrumentation import TASKS_PERPETUATED, TASKS_SUPERSEDED
 
 logger = logging.getLogger(__name__)
 
@@ -95,6 +95,15 @@ async def on_complete(self, execution: Execution, outcome: TaskOutcome) -> bool:
             return False
 
         if await execution.is_superseded():
+            worker = self.worker.get()
+            TASKS_SUPERSEDED.add(
+                1,
+                {
+                    **worker.labels(),
+                    **execution.general_labels(),
+                    "docket.where": "on_complete",
+                },
+            )
             logger.info(
                 "↬ [%s] %s (superseded)",
                 format_duration(outcome.duration.total_seconds()),

src/docket/instrumentation.py

@@ -50,6 +50,12 @@
     unit="1",
 )
 
+TASKS_SUPERSEDED = meter.create_counter(
+    "docket_tasks_superseded",
+    description="How many tasks were superseded by a newer schedule before execution",
+    unit="1",
+)
+
 TASKS_COMPLETED = meter.create_counter(
     "docket_tasks_completed",
     description="How many tasks that have completed in any state",

src/docket/worker.py

@@ -77,6 +77,7 @@
     TASKS_STARTED,
     TASKS_STRICKEN,
     TASKS_SUCCEEDED,
+    TASKS_SUPERSEDED,
     healthcheck_server,
     metrics_server,
 )
@@ -799,6 +800,12 @@ async def _execute(self, execution: Execution) -> None:
             TASKS_STRICKEN.add(1, counter_labels | {"docket.where": "worker"})
             return
 
+        # Atomically check supersession and claim task in a single round-trip
+        if not await execution.claim(self.name):
+            logger.info("↬ %s (superseded)", call, extra=log_context)
+            TASKS_SUPERSEDED.add(1, counter_labels | {"docket.where": "worker"})
+            return
+
         if execution.key in self._execution_counts:
             self._execution_counts[execution.key] += 1
 
@@ -818,11 +825,6 @@ async def _execute(self, execution: Execution) -> None:
             "%s [%s] %s", arrow, format_duration(punctuality), call, extra=log_context
         )
 
-        # Atomically check supersession and claim task in a single round-trip
-        if not await execution.claim(self.name):
-            logger.info("↬ %s (superseded)", call, extra=log_context)
-            return
-
         dependencies: dict[str, Dependency] = {}
 
         with tracer.start_as_current_span(

tests/instrumentation/test_counters.py

@@ -2,14 +2,14 @@
 
 import asyncio
 import sys
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from unittest.mock import AsyncMock, Mock
 
 import pytest
 
 if sys.version_info < (3, 11):  # pragma: no cover
     from exceptiongroup import ExceptionGroup
-from opentelemetry.metrics import Counter
+from opentelemetry.metrics import Counter, UpDownCounter
 
 from docket import Docket, Worker
 from docket.dependencies import Perpetual, Retry
@@ -361,3 +361,97 @@ async def test_task():
         await worker2.run_until_finished()
 
     assert TASKS_REDELIVERED.call_count >= 1
+
+
+@pytest.fixture
+def TASKS_RUNNING(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Mock for the TASKS_RUNNING up-down counter."""
+    mock_obj = Mock(spec=UpDownCounter.add)
+    monkeypatch.setattr("docket.instrumentation.TASKS_RUNNING.add", mock_obj)
+    return mock_obj
+
+
+@pytest.fixture
+def TASKS_SUPERSEDED(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Mock for the TASKS_SUPERSEDED counter."""
+    mock_obj = Mock(spec=Counter.add)
+    monkeypatch.setattr("docket.instrumentation.TASKS_SUPERSEDED.add", mock_obj)
+    return mock_obj
+
+
+async def test_superseded_task_increments_superseded_counter(
+    docket: Docket,
+    worker: Worker,
+    TASKS_STARTED: Mock,
+    TASKS_COMPLETED: Mock,
+    TASKS_RUNNING: Mock,
+    TASKS_SUPERSEDED: Mock,
+):
+    """Superseded tasks increment TASKS_SUPERSEDED but not lifecycle metrics.
+
+    When claim() detects that a task has been superseded by a newer generation,
+    the worker records TASKS_SUPERSEDED with docket.where=worker, but doesn't
+    touch TASKS_STARTED, TASKS_RUNNING, or TASKS_COMPLETED.
+    """
+
+    async def superseded_task():
+        pass  # pragma: no cover
+
+    await docket.add(superseded_task, key="metrics-superseded")()
+
+    # Bump the generation so the worker sees the message as superseded
+    async with docket.redis() as redis:
+        await redis.hincrby(  # type: ignore[misc]
+            docket.key("runs:metrics-superseded"), "generation", 1
+        )
+
+    await worker.run_until_finished()
+
+    TASKS_SUPERSEDED.assert_called_once_with(
+        1,
+        {
+            "docket.name": docket.name,
+            "docket.worker": worker.name,
+            "docket.task": "superseded_task",
+            "docket.where": "worker",
+        },
+    )
+    TASKS_STARTED.assert_not_called()
+    TASKS_COMPLETED.assert_not_called()
+    TASKS_RUNNING.assert_not_called()
+
+
+async def test_replaced_task_only_counts_replacement(
+    docket: Docket,
+    worker: Worker,
+    TASKS_STARTED: Mock,
+    TASKS_COMPLETED: Mock,
+    TASKS_RUNNING: Mock,
+    TASKS_SUCCEEDED: Mock,
+    TASKS_SUPERSEDED: Mock,
+):
+    """When a task is replaced before execution, only the replacement runs.
+
+    In the normal case, replace() successfully deletes the old stream message
+    via XDEL, so the worker only sees the replacement. No supersession occurs
+    because the stale message is already gone.
+    """
+
+    async def replaceable_task():
+        pass
+
+    await docket.add(replaceable_task, key="metrics-replace")()
+    await docket.replace(
+        replaceable_task, datetime.now(timezone.utc), "metrics-replace"
+    )()
+
+    await worker.run_until_finished()
+
+    TASKS_SUPERSEDED.assert_not_called()
+    TASKS_STARTED.assert_called_once()
+    TASKS_COMPLETED.assert_called_once()
+    TASKS_SUCCEEDED.assert_called_once()
+    # TASKS_RUNNING: +1 then -1
+    assert TASKS_RUNNING.call_count == 2
+    increments = [c.args[0] for c in TASKS_RUNNING.call_args_list]
+    assert sum(increments) == 0, "running gauge should be balanced"