xtr-scheduler

Recurring messages on xtr-messenger: cron and periodic triggers, catch-up after downtime, locks and saved state across restarts.

Why?

"Every night at three" and "every ten minutes" sound simple until a worker restarts at 02:59, two replicas both think it is their turn, or the machine was off for an hour and nobody knows which runs were missed.

A schedule here is a set of recurring messages — a message and when to send it. A scheduler transport turns the schedule into messages as they fall due, and a messenger worker consumes it like any other transport: each message is handled there, or sent on to wherever routing puts it — a RabbitMQ queue and its worker pool, say.

• ⏱️ At least once, in order — every message is recorded as sent before the next is produced, so in the ordinary course nothing goes twice and a restarted worker resumes where the last one stopped; a crash between sending a run and recording it sends that run again.
• 🔁 Catch-up — after downtime every missed run is sent, oldest first, or only the latest of each when the schedule asks.
• 🔒 One sender — give a schedule an xtr-lock lock and only the process holding it sends; give it an xtr-cache pool and its progress survives restarts.
• 🧭 Cron, intervals, or your own — hashed cron (#daily) spreads tasks across the hour; jitter spreads a fleet; triggers are a one-method protocol.
• 🧪 Testable in no time — time comes from xtr-clock, so a test freezes it and steps through a day instantly.

Install

uv add "xtr-scheduler[cron]"            # + cron expressions
uv add "xtr-scheduler[cron,di]"         # + a SchedulerBundle for xtr-dependency-injection
uv add "xtr-scheduler[cron,console]"    # + the debug:scheduler console command

Requires Python 3.11+.

Quick start

from xtr_messenger import MessageBusConfig, TransportConfig, WorkerFactory
from xtr_scheduler import RecurringMessage, Schedule
from xtr_scheduler.messenger import SchedulerTransportFactory

schedule = Schedule(
    RecurringMessage.cron("0 3 * * *", PurgeSessions(), timezone="Europe/Bucharest"),
    RecurringMessage.every("10 minutes", RefreshRates()),
)

config = MessageBusConfig(transports={"scheduler": TransportConfig("schedule://default")})
factories = [SchedulerTransportFactory({"default": schedule})]

await WorkerFactory(config, factories).worker(["scheduler"]).run()

The worker handles each message as it falls due, with whatever @as_message_handler declares for its type. schedule://default names the schedule.

Sending runs on

A worker consuming a schedule is one process. To have the runs handled on your worker pool instead, wrap the message in a RedispatchMessage — it is dispatched again, through routing:

from xtr_messenger import RedispatchMessage

RecurringMessage.every(
    "1 hour", RedispatchMessage(BuildReport())
)  # where routing sends BuildReport
RecurringMessage.every("1 hour", RedispatchMessage(BuildReport(), "reports"))  # to "reports" only

schedule://default?use_messenger_routing=true does this for every message of the schedule. The message arrives carrying a ScheduledStamp — which schedule, which run — on whichever worker handles it.

Recurring messages

Seconds, minutes and hours are exact: runs fall exactly that far apart. Days, weeks, months and years follow the calendar — "every day" at 09:00 stays at 09:00 across a clock change, and "every month" from the 31st runs on the last day of shorter months. every(..., from_=..., until=...) bounds an interval; left without from_, it counts from when the schedule first ran.

Hashed cron puts # where a value would go: "# # * * *" is once a day at a minute and hour picked for that message, from its string form — the same pick in every process, a different one per message, so a hundred daily tasks do not all start at midnight. #(0-5) picks within a range, and #hourly, #daily, #weekly, #monthly, #yearly and #midnight are shorthands.

A message may be a MessageProviderInterface instead, deciding the messages of each run — CallbackMessageProvider(lambda context: [Invoice(t) for t in active_tenants()]).

Triggers are strict: the next run is always strictly after the one asked about, or None to stop. CallbackTrigger, ExcludeTimeTrigger (skip a window) and JitterTrigger compose with the rest.

A schedule's behaviour

schedule = (
    Schedule(...)
    .lock(lock_factory.create_lock("schedule-default"))  # only the holder sends
    .stateful(cache_pool)  # progress survives restarts
    .process_only_last_missed_run()  # after downtime, the latest run only
)

Between runs the lock is kept until the next one is due. A worker that stops gives it back, so the next process takes over at once; one that crashes leaves it to run out.

A pool that fails to save stops the schedule with SchedulerRuntimeError rather than carrying on: going on would send every run again after a restart, or nothing at all.

Adding or removing messages while the schedule runs is safe — what already ran is not sent again.

Events

schedule.before(lambda event: event.should_cancel(not maintenance_window_open()))
schedule.after(lambda event: metrics.record(event.context.id, event.result))
schedule.on_failure(lambda event: alert(event.message, event.error))

PreRunEvent (a listener may cancel the run — the worker then skips the message), PostRunEvent (with what handling returned) and FailureEvent. They come from the worker that handles the message, through DispatchSchedulerEventListener subscribed to the messenger's worker events — the application's dispatcher hears them first, then the schedule's own listeners. A run redispatched through routing is announced once, by the worker that handles it where it lands: one sent to sync://, or to a transport no worker with the listener consumes, is not announced.

Declaring tasks

A class or function can be a task without writing a schedule:

from xtr_scheduler.decorator import as_cron_task, as_periodic_task, as_schedule


@as_cron_task("0 3 * * *", timezone="Europe/Bucharest")
async def purge_sessions() -> None: ...


@as_periodic_task("10 minutes", jitter=30, transports="jobs")
class RefreshRates:
    async def __call__(self) -> None: ...


class Maintenance:
    @as_cron_task("#daily")
    async def vacuum(self) -> None: ...

    @as_cron_task("#weekly", arguments=["full"])
    async def reindex(self, mode: str) -> None: ...

Each run sends a ServiceCallMessage naming the target, handled by calling it. schedule= picks the schedule ("default" otherwise), method= the method of a class, arguments= JSON values to call it with, and transports= sends the runs there to be handled rather than where the schedule is consumed. Stack the decorators for several runs.

@as_schedule("name") declares a class providing a whole schedule; tasks on the same name join it.

Without a container, schedule://<name> still serves the declared schedules as discovery finds the transport, each provider built with no arguments. Handling a run is separate: the ServiceCallMessageHandler takes its task targets and its per-target method allow-list as required arguments, so without a container the caller wires it — ServiceCallMessageHandler( declared_task_targets(), TaskMethods(declared_task_methods())) — rather than it building bare.

Use in an application

Everything adding this package to an application on xtr-dependency-injection takes — and, read backwards, what removing it undoes.

• Install — uv add "xtr-scheduler[cron,di,console]"; add cache for the scheduler cache pool a schedule can keep its state in.
• Recipe — uv run xtr-recipes recipes:sync does the Activate step below: it lists SchedulerBundle, which brings the messenger bundle with it. There is no config file, environment or ignore line to write; it prints the steps to declare your recurring work and copy the starter schedule, which a recipe cannot make for you.
• Activate — SchedulerBundle: {"all": True} in BUNDLES in <app>/bundles.py, imported from xtr_scheduler.bundle.
• Brings along — the messenger bundle always, because schedules are sent through it; the event dispatcher and console bundles, when those packages are installed.
• Configure — optional: with no configuration every task joins a plain default schedule, which keeps no state and takes no lock. For one that resumes after a restart and sends each run once, copy a starter schedule into <app>/schedule.py and activate the cache and lock bundles it needs.
• Environment — nothing.
• Ignore — nothing.
• Run — a worker is <script> messenger:consume scheduler_default.
• Remove — drop the BUNDLES entry, delete <app>/schedule.py and every @as_cron_task / @as_periodic_task, then uv remove xtr-scheduler.
• Check — debug:scheduler lists each schedule with every message's next run.

Kernel / bundle

# app/bundles.py
from xtr_scheduler.bundle import SchedulerBundle

BUNDLES = {SchedulerBundle: {"all": True}}

The bundle brings the messenger bundle with it. Every schedule and task its scan finds is built by the container with its dependencies — and a function task may ask for services with Injected[...]. Each schedule becomes consumable as scheduler_<name> without configuring a transport:

uv run python -m app.console messenger:consume scheduler_default

Configure scheduler_<name> yourself — TransportConfig("schedule://default?use_messenger_routing=true") — and yours is used instead. SchedulerConfig(use_messenger_routing=True) sets it for all of them. The transport asks for due runs once a second, so a run starts up to a second after it falls due; nothing is skipped, as a late run is still sent.

A starter schedule

Tasks alone need no schedule class: the bundle gathers them into a plain Schedule per name. A plain schedule keeps no state and takes no lock, though — a restarted worker starts from now, forgetting what it missed, and two workers both send every run. To choose, write the schedule the tasks join; this is a good first one to copy into the application:

# app/schedule.py
from typing import Annotated

from typing_extensions import override
from xtr_cache_contracts import CacheInterface
from xtr_dependency_injection import Target
from xtr_lock import LockFactory
from xtr_scheduler import Schedule, ScheduleProviderInterface
from xtr_scheduler.decorator import as_schedule


@as_schedule()  # "default" — every task not naming a schedule joins this one
class AppSchedule(ScheduleProviderInterface):
    def __init__(
        self,
        cache: Annotated[CacheInterface, Target("scheduler")],
        locks: LockFactory,
    ) -> None:
        self._schedule = (
            Schedule()  # add recurring messages of your own here
            .stateful(cache)  # a restart resumes, sending what was missed
            .process_only_last_missed_run()  # ...only the latest run of each, though
            .lock(locks.create_lock("scheduler-default"))  # one worker sends, however many run
        )

    @override
    def get_schedule(self) -> Schedule:
        return self._schedule

The scheduler pool it asks for exists when the cache bundle is active: the scheduler bundle adds it on the app pool's adapter, under a namespace of its own, so checkpoints never mix with the application's values and cache:pool:clear scheduler clears them alone. Configure a scheduler pool yourself — CacheConfig(pools={"scheduler": "redis://…"}) — and yours is used instead. Nothing uses the pool on its own. LockFactory comes from the lock bundle: its default resource keeps file locks, enough for workers on one machine; give it a Redis store for several.

env="prod" on a task keeps it to those environments, the way @when does for a service; its schedule stays, empty if nothing else is on it, so a worker can be started for it anywhere. With the event dispatcher bundle active, runs are announced; with the console bundle, debug:scheduler lists each schedule with every message's next run:

┌───────────────────────────────────┬─────────────────┬───────────────────────────┬─────────────┐
│ Trigger                           │ Provider        │ Next Run On               │ Next Run In │
├───────────────────────────────────┼─────────────────┼───────────────────────────┼─────────────┤
│ every 60 seconds                  │ Named (ping)    │ 2026-01-01T00:01:00+00:00 │ 1 min       │
│ 0 3 * * * with 0-30 second jitter │ Named (nightly) │ 2026-01-01T03:00:27+00:00 │ 3 h, 27 s   │
└───────────────────────────────────┴─────────────────┴───────────────────────────┴─────────────┘

--date computes from another date, --all shows messages that never run again, --sort orders by next run. A stateful schedule is listed from where its saved state says it got to.

Without a message bus

Scheduler runs schedules in-process, handing each message to a handler for its exact type:

from xtr_scheduler.scheduler import Scheduler

scheduler = Scheduler({PurgeSessions: purge}, [schedule], event_dispatcher=events)
await scheduler.run()  # until scheduler.stop()

With a dispatcher, a FailureEvent listener may ignore an error rather than let it stop the run. When run() returns, however it returns, every schedule's lock is handed back, so another process takes over at once; await scheduler.close() does the same for a scheduler that was never run.

Known limitations

• Cron expressions are read by croniter 6: a clock change that is not a whole hour can shift a match, and stepped ranges that wrap around are read as croniter reads them.
• Interval strings take one unit — "90 minutes", not "1 hour 30 minutes".
• A RedispatchMessage holds an envelope and lives in the process that creates it; the message inside is what crosses a transport.

Errors

Everything the library raises derives from SchedulerError: InvalidArgumentError (a value it cannot use — an interval, a date, an unknown schedule), SchedulerLogicError (a trigger that does not move forward, a message added twice, a transport asked to send) and SchedulerRuntimeError (state that could not be saved).

Development

Developed in the python-xtr monorepo, under packages/xtr-scheduler; run the commands below from there. The python-xtr-scheduler repository is a read-only copy, so send issues and pull requests to the monorepo.

uv sync
uv run ruff check && uv run ruff format --check && uv run basedpyright && uv run ty check && uv run pytest

License

MIT — see LICENSE.