Add Cron dependency for cron-style task scheduling

pyproject.toml

@@ -24,6 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "cloudpickle>=3.1.1",
+    "croniter>=6",
     "exceptiongroup>=1.2.0; python_version < '3.11'",
     "taskgroup>=0.2.2; python_version < '3.11'",
     "fakeredis[lua]>=2.32.1",
@@ -69,6 +70,7 @@ dev = [
     "pytest-timeout>=2.4.0",
     "pytest-xdist>=3.6.1",
     "ruff>=0.14.14",
+    "types-croniter>=6",
 ]
 
 docs = [

src/docket/__init__.py

@@ -12,6 +12,7 @@
 from .annotations import Logged
 from .dependencies import (
     ConcurrencyLimit,
+    Cron,
     CurrentDocket,
     CurrentExecution,
     CurrentWorker,
@@ -36,6 +37,7 @@
     "__version__",
     "Agenda",
     "ConcurrencyLimit",
+    "Cron",
     "CurrentDocket",
     "CurrentExecution",
     "CurrentWorker",

src/docket/dependencies/__init__.py

@@ -16,6 +16,7 @@
     format_duration,
 )
 from ._concurrency import ConcurrencyBlocked, ConcurrencyLimit
+from ._cron import Cron
 from ._contextual import (
     CurrentDocket,
     CurrentExecution,
@@ -74,6 +75,7 @@
     "AdmissionBlocked",
     "ConcurrencyBlocked",
     "ConcurrencyLimit",
+    "Cron",
     "Perpetual",
     "Progress",
     "Timeout",

src/docket/dependencies/_cron.py

@@ -0,0 +1,88 @@
+"""Cron-style scheduling dependency."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from croniter import croniter
+
+from ._perpetual import Perpetual
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ._base import TaskOutcome
+    from ..execution import Execution
+
+
+class Cron(Perpetual):
+    """Declare a task that should run on a cron schedule. Cron tasks are automatically
+    rescheduled for the next matching time after they finish (whether they succeed or
+    fail). By default, a cron task is scheduled at worker startup with `automatic=True`.
+
+    Unlike `Perpetual` which schedules based on intervals from the current time, `Cron`
+    schedules based on wall-clock time, ensuring tasks run at consistent times regardless
+    of execution duration or delays.
+
+    Supports standard cron expressions and Vixie cron-style keywords (@daily, @hourly, etc.)
+    via the croniter library.
+
+    Example:
+
+    ```python
+    @task
+    async def weekly_report(cron: Cron = Cron("0 9 * * 1")) -> None:
+        # Runs every Monday at 9:00 AM UTC
+        ...
+
+    @task
+    async def daily_cleanup(cron: Cron = Cron("@daily")) -> None:
+        # Runs every day at midnight UTC
+        ...
+    ```
+    """
+
+    expression: str
+
+    _croniter: croniter[datetime]
+
+    def __init__(
+        self,
+        expression: str,
+        automatic: bool = True,
+    ) -> None:
+        """
+        Args:
+            expression: A cron expression string. Supports:
+                - Standard 5-field syntax: "minute hour day month weekday"
+                  (e.g., "0 9 * * 1" for Mondays at 9 AM)
+                - Vixie cron keywords: @yearly, @annually, @monthly, @weekly,
+                  @daily, @midnight, @hourly
+            automatic: If set, this task will be automatically scheduled during worker
+                startup and continually through the worker's lifespan. This ensures
+                that the task will always be scheduled despite crashes and other
+                adverse conditions. Automatic tasks must not require any arguments.
+        """
+        super().__init__(automatic=automatic)
+        self.expression = expression
+        self._croniter = croniter(expression, datetime.now(timezone.utc), datetime)
+
+    async def __aenter__(self) -> Cron:
+        execution = self.execution.get()
+        cron = Cron(expression=self.expression, automatic=self.automatic)
+        cron.args = execution.args
+        cron.kwargs = execution.kwargs
+        return cron
+
+    @property
+    def initial_when(self) -> datetime:
+        """Return the next cron time for initial scheduling."""
+        return self._croniter.get_next()
+
+    async def on_complete(self, execution: Execution, outcome: TaskOutcome) -> bool:
+        """Handle completion by scheduling the next execution at the exact cron time.
+
+        This overrides Perpetual's on_complete to ensure we hit the exact wall-clock
+        time rather than adjusting for task duration.
+        """
+        self.at(self._croniter.get_next())
+        return await super().on_complete(execution, outcome)

src/docket/dependencies/_perpetual.py

@@ -61,11 +61,16 @@ def __init__(
 
     async def __aenter__(self) -> Perpetual:
         execution = self.execution.get()
-        perpetual = Perpetual(every=self.every)
+        perpetual = Perpetual(every=self.every, automatic=self.automatic)
         perpetual.args = execution.args
         perpetual.kwargs = execution.kwargs
         return perpetual
 
+    @property
+    def initial_when(self) -> datetime | None:
+        """Return None to schedule for immediate execution at worker startup."""
+        return None
+
     def cancel(self) -> None:
         self.cancelled = True
 

src/docket/worker.py

@@ -763,15 +763,12 @@ async def _schedule_all_automatic_perpetual_tasks(self) -> None:
                         perpetual = get_single_dependency_parameter_of_type(
                             task_function, Perpetual
                         )
-                        if perpetual is None:
-                            continue
 
-                        if not perpetual.automatic:
-                            continue
-
-                        key = task_function.__name__
-
-                        await self.docket.add(task_function, key=key)()
+                        if perpetual is not None and perpetual.automatic:
+                            key = task_function.__name__
+                            await self.docket.add(
+                                task_function, when=perpetual.initial_when, key=key
+                            )()
             except LockError:  # pragma: no cover
                 return
 

tests/fundamentals/test_cron.py

@@ -0,0 +1,136 @@
+"""Tests for Cron dependency (cron-style scheduled tasks)."""
+
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from croniter import croniter
+
+from docket import Docket, Worker
+from docket.dependencies import Cron
+
+
+async def test_cron_task_reschedules_itself(docket: Docket, worker: Worker):
+    """Cron tasks automatically reschedule after each execution."""
+    runs = 0
+
+    async def my_cron_task(cron: Cron = Cron("0 9 * * *", automatic=False)):
+        nonlocal runs
+        runs += 1
+
+    # Patch croniter.get_next to return a time 10ms in the future
+    with patch.object(
+        croniter,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(my_cron_task)()
+        await worker.run_at_most({execution.key: 3})
+
+    assert runs == 3
+
+
+async def test_cron_tasks_are_automatically_scheduled(docket: Docket, worker: Worker):
+    """Cron tasks with automatic=True are scheduled at worker startup."""
+    calls = 0
+
+    async def my_automatic_cron(
+        cron: Cron = Cron("0 0 * * *"),
+    ):  # automatic=True is default
+        nonlocal calls
+        calls += 1
+
+    docket.register(my_automatic_cron)
+
+    with patch.object(
+        croniter,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        await worker.run_at_most({"my_automatic_cron": 2})
+
+    assert calls == 2
+
+
+async def test_cron_tasks_continue_after_errors(docket: Docket, worker: Worker):
+    """Cron tasks keep rescheduling even when they raise exceptions."""
+    calls = 0
+
+    async def flaky_cron_task(cron: Cron = Cron("0 * * * *", automatic=False)):
+        nonlocal calls
+        calls += 1
+        raise ValueError("Task failed!")
+
+    with patch.object(
+        croniter,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(flaky_cron_task)()
+        await worker.run_at_most({execution.key: 3})
+
+    assert calls == 3
+
+
+async def test_cron_tasks_can_cancel_themselves(docket: Docket, worker: Worker):
+    """A cron task can stop rescheduling by calling cron.cancel()."""
+    calls = 0
+
+    async def limited_cron_task(cron: Cron = Cron("0 * * * *", automatic=False)):
+        nonlocal calls
+        calls += 1
+        if calls >= 3:
+            cron.cancel()
+
+    with patch.object(
+        croniter,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        await docket.add(limited_cron_task)()
+        await worker.run_until_finished()
+
+    assert calls == 3
+
+
+async def test_cron_supports_vixie_keywords(docket: Docket, worker: Worker):
+    """Cron supports Vixie cron keywords like @daily, @weekly, @hourly."""
+    runs = 0
+
+    # @daily is equivalent to "0 0 * * *" (midnight every day)
+    async def daily_task(cron: Cron = Cron("@daily", automatic=False)):
+        nonlocal runs
+        runs += 1
+
+    with patch.object(
+        croniter,
+        "get_next",
+        return_value=datetime.now(timezone.utc) + timedelta(milliseconds=10),
+    ):
+        execution = await docket.add(daily_task)()
+        await worker.run_at_most({execution.key: 1})
+
+    assert runs == 1
+
+
+async def test_automatic_cron_waits_for_scheduled_time(docket: Docket, worker: Worker):
+    """Automatic cron tasks wait for their next scheduled time instead of running immediately.
+
+    Unlike Perpetual tasks which run immediately at worker startup, Cron tasks
+    schedule themselves for the next matching cron time. This ensures a Monday 9 AM
+    cron doesn't accidentally run on a Wednesday startup.
+    """
+    calls: list[datetime] = []
+
+    async def scheduled_task(cron: Cron = Cron("0 9 * * 1")):  # Mondays at 9 AM
+        calls.append(datetime.now(timezone.utc))
+
+    docket.register(scheduled_task)
+
+    # Schedule for 100ms in the future (simulating next Monday 9 AM)
+    future_time = datetime.now(timezone.utc) + timedelta(milliseconds=100)
+    with patch.object(croniter, "get_next", return_value=future_time):
+        await worker.run_at_most({"scheduled_task": 1})
+
+    assert len(calls) == 1
+    # The task ran at or after the scheduled time, not immediately
+    assert calls[0] >= future_time - timedelta(milliseconds=50)