Scheduler

Scheduler (sillo.work.scheduler)

The scheduler runs callables on time-based triggers inside your application process. Unlike the queue system which dispatches work to separate worker processes, the scheduler runs jobs in-process using asyncio — ideal for periodic maintenance, cache warming, data sync, and health checks.

---

Architecture

The scheduler has three layers:

1. Triggers — pure functions that answer “when should this fire next?”
2. ScheduledJob — binds a callable to a trigger with execution tracking
3. SchedulerManager — owns all jobs, runs the ticker loop, integrates with the app lifecycle via app.state["scheduler"]

The ticker never blocks — create_task spawns execution and returns immediately. Resolution is 1 second.

---

Setup

from sillo import silloApp
from sillo.work.scheduler import setup_scheduler

app = silloApp()
scheduler = setup_scheduler(app)
# Auto-starts on app startup. Stores in app.state["scheduler"]

---

Triggers

Every trigger implements exactly one method: next_fire(last_fire: float) -> float | None. Given the timestamp of the last execution, it returns the next fire time (epoch seconds) or None if the trigger is exhausted.

Four triggers ship in the box.

IntervalTrigger

Fires every seconds with optional jitter (random offset). Jitter spreads load when many jobs share the same interval — preventing thundering herds.

from sillo.work.scheduler import IntervalTrigger

# Every 5 minutes, ±30s jitter:
IntervalTrigger(seconds=300, jitter=30)

# Every hour, exactly:
IntervalTrigger(seconds=3600)

# Every 10 seconds, no jitter:
IntervalTrigger(seconds=10)

CronTrigger

Standard 5-field cron expression with timezone support. The parser supports wildcards (*), ranges (1-5), steps (*/15, 1-30/5), lists (1,3,5), L (last of month/weekday), W (nearest weekday), and # (nth weekday, e.g. 2#3 = 3rd Monday).

from sillo.work.scheduler import CronTrigger

# Every weekday at 9 AM Eastern:
CronTrigger("0 9 * * 1-5", timezone="America/New_York")

# Every 15 minutes:
CronTrigger("*/15 * * * *")

# Midnight on the 1st of every month:
CronTrigger("0 0 1 * *")

# 8:30 AM and 5:30 PM daily:
CronTrigger("30 8,17 * * *")

# Every 2 hours between 8 AM and 6 PM, weekdays:
CronTrigger("0 8-18/2 * * 1-5")

Cron field reference:

Position	Field	Range	Special characters
1	Minute	0-59	* , - /
2	Hour	0-23	* , - /
3	Day	1-31	* , - / L W
4	Month	1-12	* , - /
5	Weekday	0-6 (Sun=0)	* , - / L #

DateTrigger

One-shot — fires once at the given epoch timestamp, then never again. next_fire() returns None after the first fire, which causes the job to transition to JobStatus.COMPLETED.

from sillo.work.scheduler import DateTrigger
import time

# 5 minutes from now:
DateTrigger(at=time.time() + 300)

# Specific future date:
from datetime import datetime, timezone
target = datetime(2027, 1, 1, tzinfo=timezone.utc).timestamp()
DateTrigger(at=target)

CompoundTrigger

Combine multiple triggers with AND or OR logic:

• OR — fire whenever ANY child trigger is due. Equivalent to a union of schedules. Example: “weekday mornings OR weekend afternoons.”
• AND — fire only when ALL child triggers are simultaneously due. Rarely used but useful for precise alignment (e.g. “the 1st of the month AND a weekday”).

from sillo.work.scheduler import CompoundTrigger, CompoundLogic, CronTrigger

# Fire at 9 AM weekdays OR 2 PM weekends:
trigger = CompoundTrigger(
    triggers=[CronTrigger("0 9 * * 1-5"), CronTrigger("0 14 * * 0,6")],
    logic=CompoundLogic.OR,
)

---

ScheduledJob

Binds a callable to a trigger with execution tracking and concurrency control.

from sillo.work.scheduler import ScheduledJob

job = ScheduledJob(
    my_async_func,
    CronTrigger("*/30 * * * *"),
    name="sync-api",
    args=(api_key,),
    kwargs={"endpoint": "/contacts"},
    max_instances=1,      # prevent overlapping runs
    coalesce=True,        # skip if previous run still active
    middleware=[timeout_mw, retry_mw],
)

job.compute_next()        # calculate first fire time
await job.run()           # execute manually
print(job.to_dict())      # serialized metadata

Constructor Parameters

Parameter	Type	Default	Description
func	Callable	required	Async callable to execute
trigger	Trigger	required	One of the trigger instances above
name	str | None	func.__name__	Display name
args	tuple	()	Positional arguments
kwargs	dict | None	{}	Keyword arguments
max_instances	int	1	Max concurrent runs. 0 = unlimited
coalesce	bool	True	Skip if previous run is still active
middleware	list | None	None	Middleware factories
id	str | None	auto	Explicit job ID

max_instances and coalesce work together: If a job takes 10 minutes but fires every 5 minutes, the second fire would queue up while the first is still running. With max_instances=1, coalesce=True, the second fire is skipped entirely.

---

SchedulerManager

The central coordinator. Owns all jobs, runs a 1-second ticker loop, and integrates with the app lifecycle.

Registration

from sillo.work.scheduler import SchedulerManager

s = SchedulerManager()

# Decorator style — concise and declarative:
@s.every(3600)
async def hourly_cleanup(): ...

@s.cron("0 9 * * 1-5")
async def weekday_report(): ...

# Imperative style — full control:
job = s.schedule(
    my_func,
    IntervalTrigger(60),
    name="refresh-cache",
    args=(arg1,),
    max_instances=1,
    coalesce=True,
    middleware=[retry_middleware],
)

Managing Jobs

s.pause(job.id)           # stop firing, preserve state
s.resume(job.id)          # resume a paused job
s.remove(job.id)          # permanently remove

job = s.get("some-id")    # lookup by ID
print(job.to_dict())

# List with filters:
s.list()                       # all jobs
s.list(JobStatus.ACTIVE)       # active only
s.list(JobStatus.PAUSED)       # paused only

Stats

s.stats.to_dict()
# {"jobs_total": 5, "jobs_active": 3, "jobs_paused": 1,
#  "runs": 1420, "errors": 3, "uptime": 86400}

Real-World: Admin Dashboard

@app.get("/admin/scheduler")
async def scheduler_dashboard(request, response):
    sched = request.app.state["scheduler"]
    return response.json({
        "stats": sched.stats.to_dict(),
        "jobs": [j.to_dict() for j in sched.list()],
    })

@app.post("/admin/scheduler/{job_id}/pause")
async def pause_job(request, response, job_id):
    sched = request.app.state["scheduler"]
    if not sched.pause(job_id):
        return response.json({"error": "Not found"}, status_code=404)
    return response.json({"paused": True})

@app.post("/admin/scheduler/{job_id}/resume")
async def resume_job(request, response, job_id):
    sched = request.app.state["scheduler"]
    if not sched.resume(job_id):
        return response.json({"error": "Not found"}, status_code=404)
    return response.json({"resumed": True})

@app.delete("/admin/scheduler/{job_id}")
async def remove_job(request, response, job_id):
    sched = request.app.state["scheduler"]
    if not sched.remove(job_id):
        return response.json({"error": "Not found"}, status_code=404)
    return response.json({"removed": True})

---

Middleware

Per-job middleware factories receive (handler, job) and return a new handler.

Built-in

from sillo.work.scheduler.middleware import (
    timeout_middleware, retry_middleware, rate_limit_middleware,
)

job = scheduler.schedule(
    fragile_api,
    IntervalTrigger(60),
    middleware=[timeout_middleware, retry_middleware],
)

Custom

async def logging_middleware(handler, job, *, extra=""):
    async def wrapper():
        logger.info("Job %s starting (run #%d)", job.name, job._runs + 1)
        try:
            return await handler()
        finally:
            logger.info("Job %s finished", job.name)
    return wrapper

job = scheduler.schedule(my_task, interval,
    middleware=[logging_middleware])

---

Custom Trigger

Implement next_fire(last_fire: float) -> float | None:

class BusinessHoursTrigger:
    """Fire every interval, but only 9 AM – 5 PM on weekdays."""
    def __init__(self, interval: int = 3600):
        self.interval = interval

    def next_fire(self, last_fire: float) -> float | None:
        from datetime import datetime, timedelta
        now = datetime.now()
        if 9 <= now.hour < 17 and now.weekday() < 5:
            return time.time() + self.interval
        # Skip to next business day at 9 AM:
        target = now.replace(hour=9, minute=0, second=0) + timedelta(days=1)
        while target.weekday() >= 5:
            target += timedelta(days=1)
        return target.timestamp()

scheduler.schedule(work_task, BusinessHoursTrigger(1800))

---

What the scheduler guarantees, and what it does not

The scheduler is a one-second ticker over an in-memory dictionary of jobs. Understanding exactly that much explains every one of its limits.

sillo/work/scheduler/manager.py

while self._running:
    now = time.time()
    for job in list(self._jobs.values()):
        if job.status != JobStatus.ACTIVE:
            continue
        if job.next_run_time and job.next_run_time <= now:
            ...
            job.compute_next(now)
            asyncio.create_task(self._execute(job))
    await asyncio.sleep(1)

Resolution is one second, and firing is “at or after”

A job due at 09:00:00.100 fires on the next tick, up to a second late. An interval of less than a second is not achievable — every(0.1) fires once per second, not ten times.

The tick also drifts. asyncio.sleep(1) plus the loop body is slightly more than a second, so over a day the ticker accumulates lag. For interval jobs that does not matter, because the next run is computed from the actual firing time. For cron jobs it does not matter either, because the expression is absolute. It matters only if you were counting ticks.

The schedule is in memory and is rebuilt every start

_jobs is a dictionary on the manager instance. Nothing is persisted. Two consequences:

Restarting the process reloads the schedule from your source code, which is usually what you want — your code is the source of truth.

A job due while the process was down is not caught up. Missing a nightly run because a deploy took four minutes at 02:00 is silent. If missed runs matter, record the last successful run and reconcile at startup:

catching up a missed run

@app.on_startup
async def catch_up_billing():
    last = await Setting.get_or_none(key="billing.last_run")
    if last is None or last.value < yesterday_at_2am():
        await run_billing()

Jobs added at runtime through schedule() are also lost on restart — they exist only in that process’s memory. Anything a user can create from a UI needs its definition in the database and a startup hook that re-registers it.

Every process runs its own scheduler

setup_work starts a SchedulerManager in whatever process calls it. Run uvicorn --workers 4 and you have four tickers, four copies of the schedule, and every “nightly” job running four times a night.

The failure is not obvious: four identical report emails look like a mail problem, four billing runs look like a billing bug. Nothing in the logs says “this ran on four workers”.

Three ways out:

A dedicated scheduler process. Run one instance with the scheduler enabled and no HTTP traffic, and leave it out of the web deployment. The simplest option and the easiest to reason about.

A distributed lock per run. Each tick tries to acquire a lock keyed by job name and the scheduled minute; the loser skips. Requires Redis or the database, and the lock must outlive the job.

async def run_once(job_name: str, when: str, coro):
    key = f"lock:{job_name}:{when}"
    if not await redis.set(key, "1", nx=True, ex=3600):
        return          # another process got there first
    await coro()

Leader election. One process holds a renewable lease and runs the scheduler; the others idle. More moving parts than most applications need.

Whichever you choose, make it explicit in your deployment configuration rather than relying on “we only run one worker” staying true.

Overlap control

Two settings decide what happens when a run is still going at the next scheduled time.

max_instances caps concurrent runs of the same job. At the cap, the tick skips — it does not queue.

coalesce=True skips the run entirely if any instance is still going, collapsing a backlog into one run rather than a burst.

a job that must never overlap

job = scheduler.schedule(
    sync_upstream,
    IntervalTrigger(60),
    name="sync-upstream",
    max_instances=1,
    coalesce=True,
)

Without these, a sync job that starts taking ninety seconds on a sixty-second interval accumulates overlapping runs until something breaks — usually the database connection pool. Set them on anything whose runtime is variable.

every() and cron() return a ScheduledJob

The decorators do not return the function:

@scheduler.every(300)
async def poll_upstream():
    ...

poll_upstream              # a ScheduledJob, not the function
await poll_upstream()      # TypeError: 'ScheduledJob' object is not callable

That is deliberate — it gives you poll_upstream.id for pausing and removing — but it means the decorated name can no longer be called directly, including from your tests. Keep the callable separate when you need both:

testable scheduled work

async def poll_upstream():
    ...

scheduler.schedule(poll_upstream, IntervalTrigger(300), name="poll-upstream")

Now the function is importable and awaitable in a test, and the schedule is a separate line you can leave out of the test configuration entirely.

Interval seconds are positional

every() takes a float in seconds as its only positional argument. There is no minutes=, hours=, or days= keyword:

@scheduler.every(300)              # every five minutes
async def poll_upstream(): ...


@scheduler.every(60 * 60 * 24)     # daily-ish
async def rotate_logs(): ...

For anything expressed naturally in wall-clock terms — “9am on weekdays”, “midnight on the first” — use cron(). An interval of 86400 seconds drifts relative to the clock, because it counts from the last run rather than from midnight.

Exceptions do not stop the schedule

_execute catches failures per run and records them in stats.errors. A job that raises every time keeps being scheduled, keeps failing, and keeps incrementing the counter. Alert on stats.to_dict()["errors"] rather than assuming a broken job announces itself.