feat: support cronjob tasks

Author: fubuloubu

setup.py

@@ -69,6 +69,7 @@
         "eth-pydantic-types",  # Use same version as eth-ape
         "exceptiongroup; python_version < '3.11'",  # Used with TaskGroup
         "packaging",  # Use same version as eth-ape
+        "pycron>=3.1,<4",  # Checking/triggering cron tasks
         "pydantic_settings",  # Use same version as eth-ape
         "quattro>=25.2,<26",  # Manage task groups and background tasks
         "taskiq[metrics]>=0.11.16,<0.12",

silverback/main.py

@@ -1,11 +1,12 @@
 import atexit
 import inspect
 from collections import defaultdict
-from datetime import timedelta
+from datetime import datetime, timedelta
 from functools import wraps
 from types import MethodType
 from typing import Any, Awaitable, Callable
 
+import pycron  # type: ignore[import-untyped]
 from ape.api.networks import LOCAL_NETWORK_NAME
 from ape.contracts import ContractEvent, ContractEventWrapper, ContractInstance
 from ape.logging import logger
@@ -300,6 +301,7 @@ def broker_task_decorator(
         self,
         task_type: TaskType,
         container: BlockContainer | ContractEvent | ContractEventWrapper | None = None,
+        cron_schedule: str | None = None,
     ) -> Callable[[Callable], AsyncTaskiqDecoratedTask]:
         """
         Dynamically create a new broker task that handles tasks of ``task_type``.
@@ -365,6 +367,17 @@ def add_taskiq_task(
                 labels["contract_address"] = contract_address
                 labels["event_signature"] = container.abi.signature
 
+            elif task_type is TaskType.CRON_JOB:
+                # NOTE: If cron schedule has never been true over a year timeframe, it's bad
+                if not cron_schedule or not pycron.has_been(
+                    cron_schedule, datetime.now() - timedelta(days=366)
+                ):
+                    raise InvalidContainerTypeError(
+                        f"'{cron_schedule}' is not a valid cron schedule"
+                    )
+
+                labels["cron"] = cron_schedule
+
             self.tasks[task_type].append(TaskData(name=handler.__name__, labels=labels))
 
             if self.use_fork:
@@ -498,3 +511,12 @@ def on_(
         # TODO: Support account transaction polling
         # TODO: Support mempool polling?
         raise InvalidContainerTypeError(container)
+
+    def cron(self, cron_schedule: str) -> Callable:
+        """
+        Create task to run on a schedule.
+
+        Args:
+            cron_schedule (str): A cron-like schedule string.
+        """
+        return self.broker_task_decorator(TaskType.CRON_JOB, cron_schedule=cron_schedule)

silverback/runner.py

@@ -2,8 +2,10 @@
 import signal
 import sys
 from abc import ABC, abstractmethod
+from datetime import timedelta
 from typing import Any, Coroutine, Type
 
+import pycron  # type: ignore[import-untyped]
 import quattro
 from ape import chain
 from ape.logging import logger
@@ -33,7 +35,7 @@
 from .main import SilverbackBot, TaskData
 from .recorder import BaseRecorder, TaskResult
 from .state import Datastore, StateSnapshot
-from .types import TaskType
+from .types import TaskType, utc_now
 from .utils import async_wrap_iter
 
 if sys.version_info < (3, 11):
@@ -123,6 +125,31 @@ async def _checkpoint(
         ):
             await self.datastore.save(snapshot)
 
+    async def _cron_tasks(self, cron_tasks: list[TaskData]):
+        """
+        Handle all cron tasks
+        """
+
+        while True:
+            # NOTE: Sleep until next exact time boundary (every minute)
+            current_time = utc_now()
+            wait_time = timedelta(
+                seconds=60 - 1 - current_time.second,
+                microseconds=int(1e6) - current_time.microsecond,
+            )
+            await asyncio.sleep(wait_time.total_seconds())
+            current_time += wait_time
+
+            for task_data in cron_tasks:
+                if not (cron := task_data.labels.get("cron")):
+                    logger.warning(f"Cron task missing `cron` label: '{task_data.name}'")
+                    continue
+
+                if pycron.is_now(cron, dt=current_time):
+                    self._runtime_task_group.create_task(self.run_task(task_data, current_time))
+
+            # NOTE: TaskGroup waits for all tasks to complete before continuing
+
     @abstractmethod
     async def _block_task(self, task_data: TaskData) -> None:
         """
@@ -209,6 +236,13 @@ async def startup(self) -> list[Coroutine]:
             # NOTE: No need to handle results otherwise
 
         # Create our long-running event listeners
+        cron_tasks_taskdata = (
+            await self.run_system_task(TaskType.SYSTEM_USER_TASKDATA, TaskType.CRON_JOB)
+            if Version(config.sdk_version) >= Version("0.7.15")
+            # NOTE: Not supported in prior versions
+            else []
+        )
+
         new_block_tasks_taskdata = await self.run_system_task(
             TaskType.SYSTEM_USER_TASKDATA, TaskType.NEW_BLOCK
         )
@@ -309,6 +343,9 @@ async def wait_for_graceful_shutdown():
 
         try:
             async with quattro.TaskGroup() as tg:
+                # NOTE: Our runtime tasks can use this to spawn more tasks
+                self._runtime_task_group = tg
+
                 # NOTE: User tasks that should run forever
                 for coro in user_tasks:
                     tg.create_task(coro)

silverback/types.py

@@ -21,6 +21,7 @@ class TaskType(str, Enum):
 
     # User-accessible Tasks
     STARTUP = "user:startup"
+    CRON_JOB = "user:cron-job"
     NEW_BLOCK = "user:new-block"
     EVENT_LOG = "user:event-log"
     SHUTDOWN = "user:shutdown"