Log errors from failing Perpetual tasks (#361)

When a Perpetual task fails, `on_complete()` returns `True` which tells
the Worker "I handled it" — skipping the normal error log. The
reschedule works fine, but the exception is completely invisible. Now we
log at ERROR with the full traceback before the reschedule info line.

Also fixes the logger names in the private dependency modules
(`_perpetual`, `_concurrency`, `_retry`) to use `docket.dependencies`
instead of leaking internal names like `docket.dependencies._perpetual`
into log output.

Closes #360

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

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: chrisguidry

src/docket/dependencies/_concurrency.py

@@ -16,7 +16,7 @@
     current_worker,
 )
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 if TYPE_CHECKING:  # pragma: no cover
     from redis.asyncio import Redis

src/docket/dependencies/_perpetual.py

@@ -20,7 +20,7 @@
 
 from ..instrumentation import TASKS_PERPETUATED, TASKS_SUPERSEDED
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 
 class Perpetual(CompletionHandler["Perpetual"]):
@@ -131,6 +131,15 @@ async def on_complete(self, execution: Execution, outcome: TaskOutcome) -> bool:
         )
 
         TASKS_PERPETUATED.add(1, {**worker.labels(), **execution.general_labels()})
+
+        if outcome.exception:
+            logger.error(
+                "↩ [%s] %s",
+                format_duration(outcome.duration.total_seconds()),
+                execution.call_repr(),
+                exc_info=outcome.exception,
+            )
+
         logger.info(
             "↫ [%s] %s",
             format_duration(outcome.duration.total_seconds()),

src/docket/dependencies/_retry.py

@@ -19,7 +19,7 @@
 
 from ..instrumentation import TASKS_RETRIED
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 
 class ForcedRetry(Exception):
@@ -89,6 +89,15 @@ async def handle_failure(self, execution: Execution, outcome: TaskOutcome) -> bo
 
         worker = current_worker.get()
         TASKS_RETRIED.add(1, {**worker.labels(), **execution.general_labels()})
+
+        if outcome.exception:
+            logger.error(
+                "↩ [%s] %s",
+                format_duration(outcome.duration.total_seconds()),
+                execution.call_repr(),
+                exc_info=outcome.exception,
+            )
+
         logger.info(
             "↫ [%s] %s",
             format_duration(outcome.duration.total_seconds()),

tests/fundamentals/test_perpetual.py

@@ -1,8 +1,11 @@
 """Tests for Perpetual dependency (automatically rescheduling tasks)."""
 
+import logging
 from datetime import datetime, timedelta, timezone
 
-from docket import Docket, ExecutionState, Perpetual, Worker
+import pytest
+
+from docket import Depends, Docket, ExecutionState, Perpetual, Worker
 
 
 async def test_perpetual_tasks(docket: Docket, worker: Worker):
@@ -106,6 +109,52 @@ async def perpetual_task(
     assert calls == 3
 
 
+async def test_perpetual_tasks_log_errors(
+    docket: Docket, worker: Worker, caplog: pytest.LogCaptureFixture
+):
+    """Failing Perpetual tasks log the exception instead of swallowing it."""
+
+    async def perpetual_task(
+        perpetual: Perpetual = Perpetual(every=timedelta(milliseconds=50)),
+    ):
+        raise ValueError("something broke")
+
+    execution = await docket.add(perpetual_task)()
+
+    with caplog.at_level(logging.ERROR):
+        await worker.run_at_most({execution.key: 1})
+
+    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
+    assert any(
+        "something broke" in r.getMessage()
+        or (r.exc_info and "something broke" in str(r.exc_info[1]))
+        for r in error_records
+    )
+
+
+async def test_perpetual_tasks_log_dependency_errors(
+    docket: Docket, worker: Worker, caplog: pytest.LogCaptureFixture
+):
+    """Failing dependencies in Perpetual tasks log the error instead of swallowing it."""
+
+    async def broken_dep() -> str:
+        raise RuntimeError("dependency failed")
+
+    async def perpetual_task(
+        value: str = Depends(broken_dep),
+        perpetual: Perpetual = Perpetual(every=timedelta(milliseconds=50)),
+    ):
+        pass
+
+    execution = await docket.add(perpetual_task)()
+
+    with caplog.at_level(logging.ERROR):
+        await worker.run_at_most({execution.key: 1})
+
+    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
+    assert any("dependency failed" in (r.exc_text or "") for r in error_records)
+
+
 async def test_perpetual_tasks_can_be_automatically_scheduled(
     docket: Docket, worker: Worker
 ):

tests/fundamentals/test_retries.py

@@ -1,5 +1,6 @@
 """Tests for error handling and retry strategies."""
 
+import logging
 from datetime import datetime, timedelta
 from typing import Callable
 from unittest.mock import AsyncMock
@@ -28,6 +29,31 @@ async def test_errors_are_logged(
     assert "Faily McFailerson" in caplog.text
 
 
+async def test_retry_errors_are_logged(
+    docket: Docket, worker: Worker, caplog: pytest.LogCaptureFixture
+):
+    """Each retry attempt logs the exception at ERROR level."""
+    calls = 0
+
+    async def the_task(retry: Retry = Retry(attempts=3)):
+        nonlocal calls
+        calls += 1
+        raise ValueError("attempt failed")
+
+    await docket.add(the_task)()
+
+    with caplog.at_level(logging.ERROR):
+        await worker.run_until_finished()
+
+    assert calls == 3
+    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
+    assert any(
+        "attempt failed" in r.getMessage()
+        or (r.exc_info and "attempt failed" in str(r.exc_info[1]))
+        for r in error_records
+    )
+
+
 async def test_supports_simple_linear_retries(
     docket: Docket, worker: Worker, now: Callable[[], datetime]
 ):