Scheduled tasks¶

@app.scheduled_task registers a task that a scheduler publishes on a cron schedule. The scheduler only publishes; a worker runs the tasks, as for any other message.

"""Scheduled tasks published by a Scheduler and run by a Worker.

A virtual clock makes a few minutes of schedule pass instantly; a real
scheduler uses the system clock.
"""

import asyncio
from datetime import UTC, datetime, timedelta

from asynk import App, MemoryStore, Scheduler, Worker
from asynk.transport.memory import memory_transport

producer, consumer = memory_transport()
app = App("scheduled", producer=producer, consumer=consumer, store=MemoryStore())

ran: list[str] = []
all_ran = asyncio.Event()


def record(line: str) -> None:
    ran.append(line)
    if len(ran) == 3:
        all_ran.set()


@app.scheduled_task(cron="30 8 * * mon-fri", tz="America/Sao_Paulo")
async def morning_report() -> None:
    record("morning_report")


@app.scheduled_task(cron="*/5 * * * *", args=["eu"], kwargs={"limit": 100})
async def sync(region: str, *, limit: int) -> None:
    record(f"sync {region} limit={limit}")


class VirtualClock:
    """A clock whose `sleep` advances time instead of waiting.

    Attributes:
        finished: Set once the clock reaches its stop time.
    """

    def __init__(self, start: datetime, stop: datetime) -> None:
        self._now = start
        self._stop = stop
        self.finished = asyncio.Event()

    def now(self) -> datetime:
        return self._now

    async def sleep(self, seconds: float) -> None:
        self._now += timedelta(seconds=seconds)
        if self._now >= self._stop:
            self.finished.set()


async def main() -> None:
    # Monday 11:24:30 to 11:31 UTC: `sync` fires at 11:25 and 11:30, and
    # `morning_report` at 11:30 UTC, which is 08:30 in Sao Paulo.
    clock = VirtualClock(
        start=datetime(2026, 1, 5, 11, 24, 30, tzinfo=UTC),
        stop=datetime(2026, 1, 5, 11, 31, tzinfo=UTC),
    )
    scheduler = Scheduler(app, clock=clock)
    run_scheduler = asyncio.create_task(scheduler.run())
    await clock.finished.wait()
    scheduler.request_shutdown()
    await run_scheduler

    # The scheduler only publishes; a worker runs what it published.
    worker = Worker(app, concurrency=1)
    run_worker = asyncio.create_task(worker.run())
    await all_ran.wait()
    worker.request_shutdown()
    await run_worker

    for line in ran:
        print(line)


if __name__ == "__main__":
    asyncio.run(main())

A scheduled task is an ordinary task with fixed args and kwargs; it also accepts name and retry. tz defaults to the app's timezone, App(name, tz="UTC"). In production, run the scheduler from the command line:

asynk scheduler tasks:app --heartbeat-interval 5

Cron syntax¶

Expressions have five fields (minute, hour, day of month, month, day of week), with a resolution of one minute. Month and weekday names such as jan and mon-fri are accepted. Macros such as @daily, L, W, #, ? and a seconds field are not.

When both day of month and day of week are restricted, a slot matches if either does (the classic Vixie cron rule); a field starting with * counts as unrestricted.

Validation¶

The cron expression is checked when the decorator is called, and so is the timezone when name is given; otherwise the timezone is checked when the function is decorated. The fixed arguments are checked with the app's codec when a Scheduler is created.

Timezones¶

Only non-UTC timezones need a time zone database. On systems without one (slim containers, Windows) install the tzdata package; "UTC" works without it.

Daylight saving time. A local time that does not exist (the spring-forward gap) is skipped for that day. A local time that occurs twice fires only on its first occurrence, so an every-minute schedule does not fire during the repeated hour.

Startup, missed runs and clock changes¶

Nothing fires at startup: each scheduled task's first slot is the first one after the scheduler starts. Shutdown interrupts the wait and does not wait for slots that are not yet due; a publish in flight is cancelled.

The scheduler keeps no state and never replays the past. A due slot fires if it is at most grace seconds late (default 30 s); an older one, after a stall, is skipped with a warning and the scheduler resumes at the next slot. Slots that passed while the scheduler was down are not published. Set grace with Scheduler(app, grace=...); asynk scheduler uses the default.

Fire times are decided with the wall clock, re-read after every wake. If the clock steps backwards, a scheduler never publishes a slot twice: it waits for the next slot after the last one it handled.

Running one scheduler, and dedupe¶

Each publish carries two metadata keys:

idempotency_key: "<task name>:<slot as a UTC ISO instant>";
scheduled_at: the slot, as a UTC ISO instant.

With a store, schedulers claim each slot in the store before publishing it, so overlapping schedulers publish each slot once. Give them one shared store: MemoryStore covers one process, and several machines need a network store, such as the Redis store. A scheduler releases its claim when its publish fails or is cancelled, so another scheduler can publish that slot. If the store cannot be reached, the scheduler publishes anyway rather than stop the schedule. A hard crash between the claim and the publish loses that slot.

Without a store there is no locking or coordination: run exactly one scheduler, since a second one publishes every slot twice. Duplicates are also possible around restarts and failovers. Deduping on the consuming side with metadata["idempotency_key"] needs a record shared by every worker: a set in one worker's memory only covers that worker.

There is no guard: every slot is published, and a scheduled task decides for itself whether there is work to do. A guard inside the scheduler could stall the other scheduled tasks.

Hooks and probes¶

The scheduler publishes with the app's producer and runs the app's lifecycle hooks, with runner.role set to "scheduler". Keep those hooks quick: the scheduler does not refresh its liveness probe while one runs, so a hook slower than your probe's staleness limit gets the process killed, and a slow hook delays shutdown by its full duration.

The scheduler writes the liveness probe file, never readiness; use the liveness probe from Probes with the same --probe-dir. Its main loop refreshes the probe, so a scheduler stuck in a hung publish (for example on a half-open connection) stops refreshing it and asynk health liveness fails. That is intended: run the scheduler under a supervisor, such as a Kubernetes liveness probe, that restarts it.