Scheduling

See Tasks for the rest of what apply_async accepts alongside delay.

See Enqueue options for the rest of what delayMs composes with.

See Enqueue options for the rest of what delay composes with.

See Enqueue options for the rest of what .delay_ms(..) composes with. Every setter on a TaskCall is emitted from the same macro as the one on EnqueueOptions, so the two lists cannot drift apart.

@queue.periodic() registers the function as a normal task (it can still be enqueued manually) and records the schedule. The schedule itself isn't sent to the scheduler until run_worker() starts — every @queue.periodic()-decorated function is collected and registered as a batch at that point, upserted by name, so restarting a worker with the same decorators just re-applies the same schedule rather than duplicating it.

registerPeriodic(name, taskName, cron, options?) talks to the scheduler immediately — there's no separate worker-startup step — and returns the computed next fire time (Unix ms). Re-registering the same name replaces the schedule (upsert by name).

registerPeriodic(PeriodicTask) talks to the scheduler immediately and returns the next fire time (Unix ms). Re-registering the same name replaces the schedule (upsert by name).

There is no separate registration call. The schedule rides on the task, so register::<digest>() picks it up and spawn() writes the row — before the scheduler starts, so a schedule that is already due fires on the first tick rather than an interval late.

That write is conditional, and the condition is in the database rather than in the worker. What the declaration carries is written; enabled is not part of it and is never written, and next_run is written only when the cron expression or the timezone changed. So a restart cannot resume a task an operator paused, whatever the declaration says, and cannot reset a next_run another worker's scheduler advanced unless the schedule it was computed from is gone.

A namespaced worker writes its schedules into its own namespace, so two tenants can each declare a task called nightly without either one overwriting the other's row.

Several schedules can share one @queue.periodic()-decorated worker file:

@queue.periodic(cron="0 */5 * * * *")
def health_check():
    """Run every 5 minutes."""
    ping_services()

@queue.periodic(cron="0 0 0 * * *")
def daily_cleanup():
    """Run at midnight every day."""
    queue.purge_completed(older_than=86400)

@queue.periodic(cron="0 0 9 * * 1", args=("weekly",))
def weekly_report(report_type):
    """Run every Monday at 9:00 AM."""
    generate_report(report_type)

Several schedules are several tasks, each carrying its own, registered on one worker:

#[flexiq::task(cron = "0 */5 * * * *")] // every 5 minutes
fn health_check() -> flexiq::Outcome<()> {
    ping_services().map_err(flexiq::TaskError::retryable)?;
    Ok(())
}

#[flexiq::task(cron = "0 0 0 * * *")] // midnight, every day
fn daily_cleanup() -> flexiq::Outcome<()> {
    Ok(())
}

let worker = queue
    .worker()
    .register::<health_check>()
    .register::<daily_cleanup>()
    .spawn()?;

A scheduled task takes no parameters, so a schedule cannot carry arguments — and a parameter beside cron is a compile error rather than a fire that decodes nothing, forever. Put the constant in the body, or leave the cron off and enqueue the task by hand when a run genuinely needs an argument.

Coming from Celery: Celery's crontab() only maps to the last 5 fields (minute, hour, day of month, month, day of week) — prepend a seconds field to get flexiq's 6-field format.

Coming from BullMQ? BullMQ attaches a repeat schedule to a job via add(name, data, { repeat: { pattern: cron } }) — the schedule lives inside an enqueue call. flexiq separates the two: registerPeriodic(name, taskName, cron, options?) is its own call, independent of any single enqueue(), and deletePeriodic/pausePeriodic/resumePeriodic manage it by name afterward.

cron is the one attribute whose grammar the macro does not check. Durations and rates are parsed while it expands, but the expression is passed through as a string and read at spawn() — so a malformed one fails the worker's startup with an error naming it, rather than at compile time.

Of the enqueue-default attributes, only queue reaches the job a firing creates. The scheduler builds that job itself, from fixed values, so priority, max_retries, timeout, expires and result_ttl shape a call you enqueue by hand and nothing the schedule mints:

The scheduler-policy half is unaffected, because it is keyed by task name rather than carried on the row: rate_limit, retry_budget, max_concurrent, on_excess, and the retry_backoff_ms / retry_max_delay_ms curve all apply to a fired job exactly as they do to any other.

queue.list_periodic()  # every registered schedule, enabled or paused

queue.pause_periodic("daily-digest")  # stop firing, keep the registration
queue.resume_periodic("daily-digest")
queue.delete_periodic("daily-digest")  # unschedule; False if unknown

Each entry is a PeriodicInfo — name, task_name, cron_expr, queue, enabled, last_run, next_run, timezone. Timestamps are Unix milliseconds, and last_run is None until the schedule first fires:

for schedule in queue.list_periodic():
    state = "on" if schedule.enabled else "paused"
    print(f"{schedule.name} [{state}] next={schedule.next_run}")

Every method also has an a-prefixed async form — alist_periodic, apause_periodic, aresume_periodic, adelete_periodic.

These act on the catalog in storage, so any process pointed at the same backend can read and manage it — including a producer that never starts a worker. What such a process won't see are the schedules it declares: @queue.periodic() only declares one, and writing it to storage is something run_worker() does at startup.

For the same reason, removing a @queue.periodic() decorator does not delete its row: the schedule keeps firing and enqueuing jobs for a task name no running worker handles anymore. Retire one with delete_periodic() rather than by deleting the decorator.

queue.list_periodic()?; // every registered schedule, enabled or paused

queue.pause_periodic("daily-digest")?;  // stop firing, keep the registration
queue.resume_periodic("daily-digest")?;
queue.delete_periodic("daily-digest")?; // unschedule; false if there was none

Each entry is a flexiq::PeriodicTask — name, task_name, cron_expr, queue, enabled, last_run, next_run, timezone. Timestamps are Unix milliseconds and last_run is None until the schedule first fires:

for schedule in queue.list_periodic()? {
    let state = if schedule.enabled { "on" } else { "paused" };
    println!("{} [{}] next={}", schedule.name, state, schedule.next_run);
}

All four act on the catalog in storage, so any process pointed at the same backend can manage it — including a producer that never starts a worker.

Deleting the row is not the same as retiring the schedule. A worker that still has the task in its register::<T>() chain writes it back at its next startup, because the declaration is in the code and the row is only its record. Remove the cron attribute as well, or the schedule comes back with the next deploy.

If overlap is unsafe for a task, cap it with max_concurrent=1 on the underlying task (see per-task concurrency) — the poller enforces that limit by task name, whether the job came from a periodic schedule or a normal enqueue.

If overlap is unsafe for a task, cap it with maxConcurrent: 1 (see concurrency) — the scheduler enforces that limit by task name, whether the job came from a periodic schedule or a normal enqueue.

The Java SDK doesn't expose a per-task concurrency cap in code yet. If overlapping runs would be unsafe, keep the cron interval comfortably longer than the task's worst-case execution time, or make the handler itself safe to run concurrently with itself.

If overlap is unsafe, cap the task at one. max_concurrent is enforced by the scheduler by task name, so it holds across every worker on shared storage and does not care whether the job came from a schedule or a hand-written enqueue:

#[flexiq::task(cron = "0 */5 * * * *", max_concurrent = 1)]
fn slow_sync() -> flexiq::Outcome<()> {
    Ok(())
}

A fire that finds the cap full is never dropped: a concurrency cap rolls the claim back and the job stays pending for a later tick. on_excess has no say here — it answers only for a rate limit.