Introduction

cronoq is an elastic, priority-aware task scheduler. It decides when each piece of work runs, in what order, and how much capacity a pool of workers needs. Where the work runs is up to you: async functions in your Rust process, Python coroutines, or remote workers written in any language and connected over gRPC.

            submit                 dispatch                  execute
 your code ─────────▶  cronoq  ─────────────▶  executor / remote worker
     ▲                (ordering, limits,             │
     └──── result ───  reservations, scaling) ◀──────┘ outcome

What it does

Three ways to use it

1. Rust library (cronoq crate): embed the scheduler in a tokio application. Payloads are plain Rust values; nothing is serialized. See the Rust quickstart.
2. Python package (pip install cronoq): the same engine with asyncio bindings. Payloads are Python objects. See the Python quickstart.
3. Server (cronoq-server): a standalone service with a gRPC and HTTP/JSON API. Clients in any language submit work and remote workers lease it, with SDKs for Rust, Python, Go and TypeScript. See the Server quickstart.

All three share one engine, so the scheduling rules and metrics are the same everywhere. The Rust library exposes a few engine settings, such as the priority and urgency parameters, that the Python package and the server leave at their defaults.

How to read this book

• New to cronoq? Read Why cronoq and Benefits at a glance, then the quickstart for your language.
• Evaluating it for a problem? The use cases show complete setups and the concrete benefit each feature brings.
• Designing a system on it? The concepts chapters explain every scheduling rule precisely, including formulas and defaults.
• Operating the server? See deployment, observability and the configuration reference.

Why cronoq

Most systems start with a plain queue: push jobs, have workers pop them. That works until the workload stops being uniform. Then the same problems appear again and again:

cronoq puts all of these rules in one place, the scheduling engine, so that individual workers stay simple: they receive a task, run it, and report the outcome.

Benefits in practice

• Better latency for important work without separate infrastructure. Interactive and background work can share one pool of workers. Priority and deadlines decide the order, so you do not need a dedicated "fast lane" cluster that is idle most of the time.
• Predictable scheduled jobs. A job with a fixed start time has its capacity booked when it is submitted, and a worker is held free shortly before it starts. Admission fails up front (ReservationConflict) instead of silently starting late.
• Fewer wasted retries and rate-limit errors. Rate limits are enforced before dispatch, and a task that hits a limit waits without occupying a worker.
• Lower cost. Pools scale on queue pressure: arrival rate, backlog, wait-time objectives and upcoming reservations. Scale-down is damped by a stabilization window, and cost per pool is reported directly.
• Operational clarity. Every pool reports utilization, wait percentiles, priority adherence, fixed-start hit rate over a sliding window, and cost accumulated since start, as both OpenTelemetry and Prometheus metrics.
• Speed. The engine handles over a million scheduling decisions per second on one core, so it can sit in the hot path of microtask systems as well as hours-long batch jobs.
• One engine, three surfaces. The same scheduling rules apply whether you embed the Rust library, use the Python package, or run the server for a polyglot fleet.

How it compares

cronoq is a scheduler: it orders and admits work and sizes capacity. It is not a workflow engine with durable step state, and not a general message broker. The table is a rough guide; each of these tools is excellent at what it was built for.

When to use cronoq

• Mixed workloads that share capacity and need ordering by importance or deadline.
• Work against rate-limited or capacity-limited resources (LLM APIs, GPUs, licences).
• Jobs that must start at specific times without being starved by everything else.
• Systems that want autoscaling driven by the queue rather than by CPU.
• An embedded scheduler inside a Rust or Python service, without running extra infrastructure.

When not to use cronoq

• Long-running business workflows with durable state (sagas, human-in-the-loop steps, days of waiting). Use a workflow engine such as Temporal; cronoq can still schedule its activities.
• Pure message passing / pub-sub. Use a broker such as Kafka or NATS.
• Complex data-pipeline orchestration with backfills, lineage and a UI for analysts. Use Airflow or Dagster; cronoq's DAGs are deliberately simple.
• Exactly-once side effects. cronoq executes tasks at least once. Make handlers idempotent, or use task keys to deduplicate submissions.

Benefits at a glance

This page is for engineers deciding whether cronoq fits their system, and where to start. It lists what changes when you adopt it, how to check that it worked, and what it asks of you in return.

What you gain

Pick your starting point

Where does the work run?
│
├─ Inside one Rust service ───────────────▶ cronoq crate (Rust quickstart)
├─ Inside one Python asyncio service ─────▶ pip install cronoq (Python quickstart)
├─ Code written for ElasticPriorityScheduler ▶ cronoq.compat, then the native API (Migrating)
└─ In other processes, machines or languages ▶ cronoq-server + workers (Server quickstart)
                                                 │
     Must queued work survive restarts? ── yes ─▶ add a SQLite or Postgres store (and a standby for HA)

Match your symptom

What it asks of you

cronoq is deliberately small. Adopting it means accepting a few rules:

• At-least-once execution. A task can run again if a worker or the leader fails mid-attempt. Make handlers idempotent, and give submissions a key so retries are deduplicated.
• One active scheduler per shard. HA is active-passive per shard: one leader decides, standbys wait. By default all pools are in one shard; sharding splits pools across shards, each with its own leader.
• In memory by default. Without a store, a restart loses queued work. Add a SQLite or Postgres store when that matters.
• A scheduler, not a workflow engine. It orders, admits and sizes work. Long-running workflows with durable step state, human approvals or replays belong in a workflow engine; cronoq can schedule their activities.
• Payloads are plain Rust values in the library, Python objects in the Python package, and opaque bytes on the server. Large blobs belong in object storage, with a reference in the payload (the server limits messages to 16 MiB by default).

See Why cronoq for a comparison with queues, task frameworks, Kubernetes Jobs and workflow engines.

A typical adoption path

1. Embed the library (Rust or Python) in the service that already runs the work, with one pool per kind of worker and the defaults everywhere else.
2. Classify work: priorities for user-facing versus background tasks, deadlines where an SLA exists, keys on everything that may be retried.
3. Protect dependencies: a token bucket per external API, a semaphore per scarce resource.
4. Observe: export metrics and alert on wait percentiles, dead letters and fixed-start hit rate.
5. Scale out when workers outgrow one process: run cronoq-server, move executors into workers, add Postgres and a standby, and drive the worker fleet from cronoq_pool_desired_workers.

Rust quickstart

Install

[dependencies]
cronoq = "0.2"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

The otel feature (on by default) adds OpenTelemetry metrics and trace propagation. Disable it with default-features = false for a minimal dependency tree.

The minimum supported Rust version is 1.94.

A first scheduler

use std::{sync::Arc, time::Duration};
use cronoq::{Pool, Priority, Scheduler, Task, TaskError};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // A scheduler with one pool. Payloads are `String`s, outputs are `usize`s.
    let sched = Scheduler::<String, usize>::builder()
        .pool(
            Pool::new("default")
                .workers(1, 2, 8) // min, initial, max concurrent tasks
                .executor_fn(|ctx, text: Arc<String>| async move {
                    println!("task {} attempt {}", ctx.id, ctx.attempt);
                    Ok::<_, TaskError>(text.len())
                }),
        )
        .build()
        .await?;

    // Submit and await the output.
    let handle = sched
        .submit(Task::new("hello".to_string()).priority(Priority::High))
        .await?;
    let len = handle.await?; // Arc<usize>
    assert_eq!(*len, 5);

    sched.shutdown(Duration::from_secs(5)).await?;
    Ok(())
}

What happened:

1. Scheduler::builder() creates a scheduler for payload type String and output type usize.
2. Pool::new("default") is an autoscaled pool of worker slots. In-process, a slot is simply a concurrently running task, so workers(1, 2, 8) means "start with 2 concurrent tasks, scale between 1 and 8 based on queue pressure".
3. executor_fn turns an async closure into an Executor. It receives a TaskContext (id, attempt, cancellation, heartbeat) and the payload.
4. submit returns a TaskHandle that resolves with Result<Arc<O>, TaskFailure> once the task reaches a terminal status.

Adding scheduling rules

Everything else is set on the task or the pool:

use std::time::Duration;
use cronoq::{Dep, Priority, RetryPolicy, Task};

let task = Task::new(payload)
    .pool("default")
    .kind("summarize")                         // used for metrics and duration learning
    .priority(Priority::High)
    .key("summary:doc-42")                     // idempotent submit + dependency handle
    .deadline_in(Duration::from_secs(30))      // urgency rises as the deadline approaches
    .timeout(Duration::from_secs(60))          // per attempt
    .retry(RetryPolicy { max_attempts: 5, ..Default::default() })
    .resource("llm-api", 1)                    // shared rate limit / semaphore
    .after("fetch:doc-42")                     // run after this key succeeds
    .dep(Dep::on_failure("fetch:doc-42"));     // ...or a failure handler edge

Tasks that must start at a given time use start_in / start_at; recurring work uses Scheduler::add_recurring with a cron expression. See Fixed starts and Recurring schedules.

Signalling failures

Return a TaskError from the executor:

Blocking (non-async) code

If your application does not use tokio, build a BlockingScheduler. It owns a private runtime and exposes the same operations as blocking calls:

let sched = cronoq::Scheduler::<u64, u64>::builder()
    .pool(cronoq::Pool::fixed("default", 4).executor_fn(|_ctx, x: std::sync::Arc<u64>| async move {
        Ok::<_, cronoq::TaskError>(*x * 2)
    }))
    .build_blocking()?;
let out = sched.submit(cronoq::Task::new(21))?.wait()?;
assert_eq!(*out, 42);

Do not call build_blocking from inside a tokio runtime; use the async API there.

Examples in the repository

Next: use cases for complete setups, or concepts for the precise rules.

Python quickstart

The cronoq package embeds the same Rust engine in Python, with an asyncio-native API. Payloads and results are ordinary Python objects; nothing is serialized unless you keep state on disk (see Durable state).

Install

pip install cronoq             # embedded scheduler
pip install "cronoq[client]"   # plus the gRPC client and remote worker for cronoq-server

Wheels are published for CPython 3.14+ (abi3) on Linux, macOS and Windows.

A first scheduler

import asyncio
import cronoq

async def summarize(ctx: cronoq.TaskContext, text: str) -> str:
    # ctx.id, ctx.attempt, ctx.kind, ctx.key, ctx.pool, ctx.tags, ctx.timeout, ctx.deadline
    await asyncio.sleep(0.1)          # call a model, an API, a database...
    return text[:20]

async def main() -> None:
    pools = [cronoq.Pool("default", summarize, min=1, max=8)]
    async with cronoq.Scheduler(pools) as sched:
        handle = await sched.submit("a long document ...", priority="high", deadline_in=5.0)
        print(await handle)           # the executor's return value

asyncio.run(main())

• Pool(name, executor, min=, max=) defines a pool of concurrency slots that autoscales between min and max. Pass autoscale=False for a fixed size.
• Executors are async def functions run on the event loop that started the scheduler. Plain functions are accepted too; they run in the loop's default thread pool, so blocking calls do not stall the loop.
• async with Scheduler(...) starts the scheduler and shuts it down on exit (see Lifecycle).
• Tasks without pool= go to the default pool: the one named "default", or the first.

Scheduling options

Every option of the engine is a keyword argument on submit (and on cronoq.Task for batches):

handle = await sched.submit(
    {"doc": 42},
    pool="default",
    kind="summarize",                 # metrics and duration learning
    priority=cronoq.Priority.HIGH,    # or "critical" / "high" / "normal" / "low"
    key="summary:42",                 # idempotent submit; dependency handle
    deadline_in=30.0,                 # seconds
    timeout=60.0,                     # per attempt
    retry=cronoq.RetryPolicy(max_attempts=5, initial_backoff=0.5),
    resources={"llm-api": 1},         # shared rate limit or semaphore
    after=["fetch:42"],               # run after this key succeeds
)

Retries and failures

async def call_api(ctx, payload):
    resp = await http.post(...)
    if resp.status == 429:
        raise cronoq.Retryable("rate limited", retry_after=float(resp.headers["Retry-After"]))
    if resp.status == 400:
        raise cronoq.Permanent("bad request")   # no retry
    return resp.json()

Any other exception is treated as retryable. Awaiting a task that did not succeed raises cronoq.TaskFailed with .status (failed, dead_lettered, cancelled, skipped, missed, shed, or shutdown), .message and .task_id. All cronoq exceptions can be pickled, so they cross multiprocessing boundaries.

Batches and dependencies

handles = await sched.submit_batch([
    cronoq.Task("extract", key="extract"),
    cronoq.Task("transform", key="transform", after="extract"),
    cronoq.Task("load", key="load", after="transform"),
    cronoq.Task("alert", deps=[cronoq.Dep("load", on="failure")]),
])

A batch is admitted atomically: if any task is rejected (for example because of a cycle), none are.

Status events

sched.events() streams status changes, for logs, metrics or notifying users:

async with sched.events() as events:
    async for e in events:            # cronoq.StatusEvent
        print(e.id, e.key, e.status, e.attempt, e.message)

By default only terminal statuses (succeeded, failed, dead_lettered, ...) are delivered; create the scheduler with emit_transitions=True to also see ready, running, retrying and the rest. Each stream buffers event_capacity events (default 1024); a consumer that falls behind loses the oldest ones (counted in events.lagged), or gets cronoq.EventsLagged with sched.events(on_lag="raise").

Lifecycle

• A scheduler belongs to the event loop that starts it. Use async with cronoq.Scheduler(...), or await sched.start() and await sched.shutdown(), inside that loop.
• Leaving async with waits up to shutdown_grace seconds (default 5) for running attempts, then cancels them. Queued tasks are not started.
• If the loop closes without a shutdown, the scheduler stops and later calls raise cronoq.SchedulerError.
• Cancelling a task, a timeout or a lost lease cancels the executor's asyncio task. An executor that swallows CancelledError keeps running and may overlap with its retry.

Durable state

By default the scheduler keeps its state in memory. Give it a store to keep queued work, schedules and results across restarts:

store = cronoq.SqliteStore("/var/lib/myapp/cronoq.db")
async with cronoq.Scheduler(pools, store=store) as sched:
    await sched.enqueue({"doc": 42}, key="summary:42")

• Payloads and results are stored as JSON by default; submitting something JSON cannot hold raises TypeError. Pass codec= (an object with encode and decode) for other types.
• After a restart, queued and interrupted tasks run again on the new scheduler's pools, and await sched.handle("summary:42") returns the stored result of a task that already finished.
• Several processes may open the same file. One leads; the others wait in start() and take over when it stops.

See SqliteStore for the options.

Coming from ElasticPriorityScheduler

cronoq.compat.ElasticPriorityScheduler is a synchronous, thread-safe drop-in with the original API (submit, start, stop, get_metrics, get_cost, completed_count). See Migrating from ElasticPriorityScheduler.

Talking to a cronoq server

With the client extra, cronoq.client.Client submits work to a cronoq-server and cronoq.client.Worker serves a pool:

# worker.py: serves the pool until stopped
import asyncio, os
from cronoq.client import Worker

async def handle(lease):
    return {"answer": lease.json()}

asyncio.run(Worker("localhost:7070", "llm", handle, slots=8,
                   token=os.environ["CRONOQ_WORKER_TOKEN"]).run())

# client.py: submits a task and waits for its result
import asyncio, os
from cronoq.client import Client

async def main():
    async with Client("localhost:7070", token=os.environ["CRONOQ_CLIENT_TOKEN"]) as c:
        sub = await c.submit({"prompt": "hi"}, pool="llm", priority="high", wait=True)
        print(sub.result.json())

asyncio.run(main())

See Writing executors and workers, the server quickstart and the Python API reference.

Server quickstart

cronoq-server runs the scheduler as a service. Clients submit tasks over gRPC or HTTP/JSON; remote workers in any language connect over a gRPC stream, register for a pool with a number of slots, and receive leases. In about ten minutes you will run a server, connect a worker and push work through it.

 clients (any language)         cronoq-server                  workers (any language)
 ──────────────────────▶  ┌───────────────────────┐  ◀── Workers.Work stream ───
  gRPC :7070 / HTTP :7071 │ engine + remote pools  │  ─── Lease / Abort ────────▶
                          └───────────┬───────────┘  ◀── Heartbeat / Complete ──
                                      │ optional
                                  PostgreSQL

What you get compared with a plain queue: priorities with aging, deadlines, fixed start times with reserved capacity, dependencies, retries with backoff, shared rate limits, idempotent keys and an autoscaling signal, behind one small binary.

1. Create tokens

The server authenticates every call with a bearer token per role. Tokens must be at least 16 characters; use random 32-character values:

for role in CLIENT WORKER METRICS ADMIN; do
  export CRONOQ_${role}_TOKEN=$(openssl rand -hex 16)
done

2. Start the server

From source:

cargo run --release -p cronoq-server -- --config deploy/cronoq.example.toml

Or with the released container image, which contains the same example configuration:

docker run --rm -p 7070:7070 -p 7071:7071 \
  -e CRONOQ_CLIENT_TOKEN -e CRONOQ_WORKER_TOKEN -e CRONOQ_METRICS_TOKEN -e CRONOQ_ADMIN_TOKEN \
  ghcr.io/synaptiai/cronoq:0.2.1

The example configuration defines two pools (default and llm), two shared resources (openai, a token bucket, and gpu, a semaphore) and the four tokens above. It listens on all interfaces, which the server only allows with authentication configured.

Without --config the server starts with a single default pool, listens on 127.0.0.1 only and has authentication disabled: fine for a laptop, refused on any other address unless you pass --insecure-no-auth.

Validate a configuration without starting anything:

cronoq-server --config cronoq.toml --check     # "configuration is valid"

Check the server is up:

curl -s localhost:7071/readyz        # "ready"

3. Connect a worker

A pool's capacity is the sum of the slots of its connected workers, so nothing runs until a worker connects. In Python (pip install "cronoq[client]"):

import asyncio, os
from cronoq.client import Worker

async def handle(lease):
    data = lease.json()
    return {"echo": data, "attempt": lease.attempt}

asyncio.run(Worker("localhost:7070", "default", handle, slots=4,
                   token=os.environ["CRONOQ_WORKER_TOKEN"]).run())

Or in Rust, with the worker client in the cronoq-server crate:

use bytes::Bytes;
use cronoq_server::worker::{run_worker, WorkerOptions};

let opts = WorkerOptions::new("http://localhost:7070", "default")
    .slots(4)
    .token(std::env::var("CRONOQ_WORKER_TOKEN")?);
run_worker(opts, |lease| async move { Ok(Bytes::from(lease.payload.to_ascii_uppercase())) },
           std::future::pending()).await?;

Both clients heartbeat running leases, reconnect with backoff, and drain on shutdown. Workers in other languages implement the cronoq.v1.Workers/Work stream; see the protocol rules.

4. Submit work

With curl (HTTP/JSON):

curl -s localhost:7071/v1/tasks \
  -H "authorization: Bearer $CRONOQ_CLIENT_TOKEN" -H 'content-type: application/json' \
  -d '{"payload": {"hello": "world"}, "priority": "high", "wait": true}'

{"duplicate":false,"id":"1","result":{"id":"1","output":{"attempt":1,"echo":{"hello":"world"}},"output_available":true,"output_b64":"...","status":"succeeded"}}

Task ids are strings in JSON, because they are 64-bit.

With the Python client:

import asyncio, os
from cronoq.client import Client

async def main():
    async with Client("localhost:7070", token=os.environ["CRONOQ_CLIENT_TOKEN"]) as c:
        sub = await c.submit({"hello": "world"}, priority="high", key="hello-1")
        result = await c.wait(sub.id, timeout=30)
        print(result.status, result.json())

asyncio.run(main())

Submitting again with the same key returns the same task (duplicate = true), so retries after a network error are safe.

5. Look inside

curl -s localhost:7071/v1/snapshot -H "authorization: Bearer $CRONOQ_METRICS_TOKEN" | jq
curl -s localhost:7071/metrics -H "authorization: Bearer $CRONOQ_METRICS_TOKEN" | grep cronoq_pool

cronoq_pool_desired_workers is the number of slots the autoscaler wants for each pool; feed it to KEDA or an HPA to scale your worker deployment.

Next steps

• Using the HTTP API for dependencies, batches, scheduled work and errors.
• Writing executors and workers for worker design.
• Deploying the server for Kubernetes, KEDA autoscaling and HA with Postgres.
• Securing the server before exposing it beyond a trusted network.
• Server configuration for every option.

Use cases

Each page in this section starts from a concrete problem, shows a complete cronoq setup, and lists the specific benefits you get compared with a plain queue. Not sure which one applies? Start with Benefits at a glance, which maps symptoms to pages.

Use cases combine freely: an LLM pipeline can use DAGs, rate limits, deadlines and autoscaling at the same time, because all of them are properties of tasks and pools in a single engine.

LLM and agent workloads

The problem

Applications built on large language models send many requests to model APIs: chat turns, tool calls, embeddings, batch summarization, evaluation runs. These requests:

• share provider rate limits (requests or tokens per minute) across every worker;
• fail with HTTP 429 when the limit is exceeded, often with a Retry-After header;
• have mixed urgency: a user is waiting for a chat reply, nobody is waiting for tonight's evaluation run;
• take seconds to minutes, with a long tail;
• run on expensive workers (or are billed per concurrent request).

A plain queue sends requests as fast as workers can pop them. During a burst every worker hits the rate limit, retries immediately, and makes it worse. Meanwhile user-facing requests wait behind batch work.

The setup

use std::{sync::Arc, time::Duration};
use cronoq::*;

let sched = Scheduler::<Prompt, Answer>::builder()
    // The provider allows 50 requests/s with bursts of 100; every pool shares it.
    .resource(ResourceConfig::token_bucket("openai", 50.0, 100))
    // At most 20 requests in flight at once (connection or concurrency limit).
    .resource(ResourceConfig::semaphore("openai-concurrency", 20))
    .pool(
        Pool::new("llm")
            .workers(2, 4, 64)
            .autoscale(AutoscaleConfig {
                // Interactive requests should not wait more than 2 s in the queue.
                slo: [None, Some(Duration::from_secs(2)), None, None],
                cost_per_worker_second: 0.0004,
                ..Default::default()
            })
            .executor_fn(|ctx, p: Arc<Prompt>| async move {
                match call_model(&p).await {
                    Ok(a) => Ok(a),
                    Err(ApiError::RateLimited { retry_after }) => {
                        Err(TaskError::retry_after("429", retry_after))
                    }
                    Err(ApiError::BadRequest(m)) => Err(TaskError::permanent(m)),
                    Err(e) => Err(TaskError::from_err(e)),
                }
            }),
    )
    .build()
    .await?;

// A chat turn: high priority, should finish within 2 seconds.
sched.submit(Task::new(chat_prompt)
    .pool("llm").kind("chat")
    .priority(Priority::High)
    .deadline_in(Duration::from_secs(2))
    .timeout(Duration::from_secs(60))
    .resource("openai", 1).resource("openai-concurrency", 1)).await?;

// An evaluation item: low priority, retried generously.
sched.enqueue(Task::new(eval_prompt)
    .pool("llm").kind("eval")
    .priority(Priority::Low)
    .retry(RetryPolicy { max_attempts: 8, ..Default::default() })
    .resource("openai", 1).resource("openai-concurrency", 1)).await?;

The same setup in Python:

limits = [cronoq.Resource.token_bucket("openai", rate=50.0, burst=100),
          cronoq.Resource.semaphore("openai-concurrency", 20)]
pool = cronoq.Pool("llm", call_model, min=2, max=64,
                   autoscale=cronoq.AutoscaleConfig(slo={"high": 2.0}))
async with cronoq.Scheduler([pool], resources=limits) as sched:
    await sched.submit(chat, pool="llm", priority="high", deadline_in=2.0,
                       resources={"openai": 1, "openai-concurrency": 1})

The repository has a runnable version: cargo run -p cronoq --example llm_agents.

What you get

• No 429 storms. The token bucket is checked before dispatch. A task that cannot get a token is parked on the resource without occupying a worker, and is released in FIFO order as tokens refill. Requests are paced at the provider's rate instead of bursting into it.
• Correct backoff. TaskError::retry_after (Python: Retryable(retry_after=...)) delays the next attempt by at least the server's Retry-After, combined with exponential backoff and jitter from the retry policy.
• Users first. Chat turns are High and carry a deadline, so they overtake queued evaluation work. Low-priority work still progresses: aging lets it overtake a continuous stream of Normal work.
• Capacity that follows demand. The pool scales on arrival rate and backlog, and the per-class SLO adds capacity when predicted queue wait for High work exceeds 2 seconds.
• Bounded attempts. Per-attempt timeouts abort hung calls; tasks that keep failing end as DeadLettered instead of retrying forever.
• Cost visibility. cost_per_worker_second makes each pool report its accumulated cost.

Variations

• Several providers or models: one token bucket per provider; a task can require several resources at once (all are acquired atomically).
• Token-based limits: request resource("openai-tpm", estimated_tokens / 1000) from a bucket sized in thousands of tokens.
• Agent tool calls that depend on each other: submit them as a batch with keys and after edges (Pipelines and DAGs).
• Workers on GPU machines running local models: use the server and remote workers (Polyglot worker fleets).

Interactive and batch work on shared capacity

The problem

A service runs two kinds of work on the same workers: requests a user is waiting for, and background jobs (exports, re-indexing, reports). With one FIFO queue, a large export blocks every user request behind it. The usual fix is separate queues and separate worker fleets, which means paying for interactive capacity that is idle most of the day, while the batch fleet cannot borrow it during quiet hours.

The setup

One pool, four priority classes:

async with cronoq.Scheduler([cronoq.Pool("app", run, min=2, max=32)]) as sched:
    # A user clicked "export": they are waiting.
    await sched.submit(job, priority="high", kind="export-interactive")

    # Nightly re-index of 50,000 documents.
    for doc in docs:
        await sched.enqueue(doc, priority="low", kind="reindex")

    # A payment webhook that must never wait.
    await sched.submit(event, priority="critical", kind="webhook")

Optionally give each class a wait-time objective, so the autoscaler adds capacity when it would be violated:

autoscale = cronoq.AutoscaleConfig(slo={"critical": 0.5, "high": 2.0})
pool = cronoq.Pool("app", run, min=2, max=32, autoscale=autoscale)

How the order is decided

Every waiting task has a score (see Priorities and aging):

score = base[class] + min(2 × seconds_waiting, 800)
base  = critical 10000 · high 1000 · normal 100 · low 10

Dispatch always picks the highest score. In practice:

• A new critical task goes first.
• high always beats normal and low, however long they have waited.
• A low task that has waited about 45 seconds overtakes freshly submitted normal work, so normal traffic cannot starve the batch queue indefinitely.
• Within a class, order is strictly FIFO.

What you get

• One fleet instead of two. Interactive work borrows the whole pool when it needs it; batch work uses whatever is left. No idle "fast lane" capacity.
• Bounded latency for users regardless of how much batch work is queued, because dispatch cost grows only with the logarithm of the queue length and higher classes are always considered first.
• Batch jobs still finish. Aging prevents starvation by normal traffic, and you can tune the aging rate and caps per class.
• Measurable fairness. priority_adherence in each pool snapshot is the fraction of dispatches made while no task of a strictly higher class was waiting; higher is better (with aging, a value slightly below 1.0 is expected and healthy).

Variations

• Protect batch throughput at peak: put batch work in its own pool with min workers and interactive work in another; resources can still be shared between pools.
• Different aging policy: configure PriorityConfig (Rust) per class, for example a higher age_cap for low to let it overtake high after a long wait.

Deadline-driven work

The problem

Some work has a time by which it must be done: a report promised by 09:00, a fraud check that must finish before a payment settles, a transcoding job due before a broadcast. A priority alone does not capture this: a "normal" job due in five minutes is more urgent than a "high" job due tomorrow, but only once its deadline gets close.

The setup

Give tasks a deadline and, when known, a duration hint:

use std::time::{Duration, SystemTime};
use cronoq::{Priority, Task};

// Must be done in 10 minutes; usually takes ~2 minutes.
sched.submit(Task::new(report)
    .priority(Priority::Normal)
    .deadline_in(Duration::from_secs(600))
    .duration_hint(Duration::from_secs(120))).await?;

// Absolute deadline.
sched.submit(Task::new(batch)
    .deadline_at(SystemTime::now() + Duration::from_secs(3600))).await?;

await sched.submit(report, deadline_in=600, duration_hint=120)

Without a duration hint, cronoq uses the median runtime it has observed for the task's kind in that pool, then the pool's statistics, then the pool's default_duration (1 s).

How urgency works

From the deadline D and the estimated duration d, cronoq computes the latest start time LST = D − d. As the remaining slack LST − now shrinks below a horizon H = max(d, 5 s), the task's score gains urgency, rising linearly to 5000 at the latest start time. After that it keeps rising (100 points per second, up to +3000) so that an overdue task always ranks above one that is merely close. Within a priority class, tasks with deadlines are ordered by latest start time (earliest-deadline-first), which is optimal for meeting deadlines on a single resource.

Because 5000 exceeds the gap between high (1000) and low (10), a low task near its deadline overtakes fresh high work, but not critical work (10000).

What you get

• Fewer missed deadlines with the same capacity. Work is reordered as deadlines approach, instead of strictly by priority or arrival.
• Correct ordering among deadline tasks: earliest latest-start-time first.
• Escalation for overdue work that is monotonic, so overdue tasks never fall behind ones that are merely close to their deadline.
• Learned estimates. Duration estimates improve as tasks of the same kind complete.
• Visibility. TaskContext::deadline (Rust) gives executors the deadline so they can choose a cheaper code path when time is short.

Related

• If the work must start at a specific time rather than finish by one, use a fixed start instead.
• To add capacity when deadlines are at risk, combine deadlines with per-class wait SLOs on the pool's autoscaling configuration.

Scheduled jobs with guaranteed capacity

The problem

Cron jobs and "run at 14:00" tasks usually share workers with everything else. When the time comes, every worker may be busy with long jobs, so the scheduled job starts late, sometimes much later. Teams compensate with dedicated cron machines that sit idle 99% of the time.

The setup

Submit work with a fixed start time:

use std::time::Duration;
use cronoq::{MissPolicy, Task};

sched.submit(Task::new(close_books)
    .start_at(at_14_00)                        // SystemTime
    .start_tolerance(Duration::from_secs(5))   // "on time" = within ±5 s
    .on_miss(MissPolicy::Skip)                 // if it still can't start, drop it
    .duration_hint(Duration::from_secs(300))).await?;

Or a recurring schedule (cron with optional seconds, IANA time zones):

use std::time::Duration;
use cronoq::{OverlapPolicy, Recurring, Task};

sched.add_recurring(
    Recurring::new("hourly-rollup", "0 0 * * * *", Task::new(rollup).duration_hint(Duration::from_secs(600)))
        .tz("Europe/Lisbon")
        .overlap(OverlapPolicy::Skip),          // skip an occurrence while the previous one runs
).await?;

await sched.add_recurring("hourly-rollup", "0 0 * * * *", rollup_payload,
                          tz="Europe/Lisbon", overlap="skip", duration_hint=600)

What happens

1. Admission. The task reserves an interval [start − tolerance, start + duration + tolerance] in its pool. Up to half of the pool's max_workers (at least one) may be reserved at any moment; if the new reservation would exceed that, the submission fails immediately with ReservationConflict. You learn about an impossible schedule when you create it, not when it runs late.
2. Pre-warming. Reservations within the autoscaler's horizon (60 s by default) count towards the pool's desired capacity, so workers are added before they are needed.
3. Holding. Shortly before the start (by the pool's p90 task duration, at least the tolerance, at most 5 minutes), the next free worker is held for the task: regular work cannot take it.
4. Start. At the start time the task runs on its held worker. The recorded start time is the actual dispatch time, and the pool's fixed_start_hit_rate metric counts it as on time if it started within the tolerance.
5. Miss. If it still could not start within the tolerance, the miss policy applies: RunLate (default: run as soon as possible, ahead of everything else), Skip (status Missed), or Fail.

What you get

• On-time starts without a dedicated fleet. Capacity is reserved and a worker is held only around the start time; the rest of the time it serves regular work.
• Early feedback. Over-subscribed schedules are rejected at submission time.
• Explicit miss semantics instead of silently late jobs.
• Measurable punctuality via fixed_start_hit_rate.
• Durable schedules. With a SQLite or Postgres store, recurring definitions and their next occurrence survive restarts and failovers. After downtime, at most one missed occurrence runs late (with RunLate); the schedule then resumes from the present instead of replaying every missed occurrence.

Pipelines and DAGs

The problem

Multi-step work (extract → transform → load, or crawl → embed → index) is often chained by hand: each step enqueues the next one when it finishes. Failures leave half-finished pipelines, downstream steps run on missing inputs, and there is nowhere to hang a "notify on failure" step.

The setup

Give tasks keys and declare edges. A batch is admitted atomically and may reference keys in any order:

use cronoq::{Dep, Task};

let mut tasks = vec![Task::new(Step::Extract).key("extract")];
for shard in 0..4 {
    tasks.push(Task::new(Step::Transform(shard)).key(format!("t{shard}")).after("extract"));
}
let mut load = Task::new(Step::Load).key("load");
for shard in 0..4 {
    load = load.after(format!("t{shard}"));            // fan-in
}
tasks.push(load);
tasks.push(Task::new(Step::Alert).dep(Dep::on_failure("load")));   // runs only if load fails
tasks.push(Task::new(Step::Cleanup).dep(Dep::on_complete("load"))); // runs either way

let handles = sched.submit_batch(tasks).await?;

await sched.submit_batch([
    cronoq.Task("extract", key="extract"),
    *[cronoq.Task(("transform", i), key=f"t{i}", after="extract") for i in range(4)],
    cronoq.Task("load", key="load", after=[f"t{i}" for i in range(4)]),
    cronoq.Task("alert", deps=[cronoq.Dep("load", on="failure")]),
    cronoq.Task("cleanup", deps=[cronoq.Dep("load", on="complete")]),
])

Runnable: cargo run -p cronoq --example dag_pipeline.

Edge semantics

Skipped propagates: everything downstream of a skipped task is skipped too, except children attached with on_complete. A task that waits on dependencies has status Blocked; it enters the queue only when every edge is satisfied, so blocked tasks never occupy workers.

Dependencies can reference task ids or keys. A key that has never been seen is rejected by default; configure UnknownDependency::Await { timeout } on the pool to let a child wait for a task that will be submitted later (useful when producers are independent).

What you get

• Correctness by construction. Children never start before their parents finish. This is checked continuously by cronoq's randomized simulation tests.
• Failure handling as a first-class step with on_failure and on_complete edges.
• Atomic submission. A batch with a cycle, an unknown dependency, a reservation conflict or a full queue is rejected as a whole, never half-admitted.
• Idempotent pipelines. Re-submitting a batch whose keys already exist returns the existing tasks (duplicate = true) instead of running them twice. Keys stay reserved while the task is live and for the retention window after it finishes (1 hour by default).
• Late dependents still work. A child submitted after its parent finished is resolved from the retained terminal status.

Limits

cronoq's DAGs are a scheduling primitive, not a workflow engine: there are no data passing between steps (pass references in payloads), no conditional branching beyond edge conditions, and no long-term workflow history. For those, pair cronoq with a workflow engine.

GPUs and other scarce resources

The problem

Some work needs something there is little of: GPU slots, database connections, a licensed tool, a partner API with a concurrency cap. Workers usually enforce such limits locally (a semaphore per process), which over-subscribes the resource as soon as you run more than one worker, or under-uses it when the limit is split statically.

Worse, a FIFO queue blocks: if the task at the head needs the GPU and the GPU is busy, the CPU tasks behind it wait too.

The setup

Declare the resource once on the scheduler; tasks request units of it:

use cronoq::{Pool, ResourceConfig, Scheduler, Task};

let sched = Scheduler::<Job, Out>::builder()
    .resource(ResourceConfig::semaphore("gpu", 4))          // 4 GPUs in total
    .resource(ResourceConfig::semaphore("warehouse-db", 10)) // 10 connections
    .pool(Pool::new("ml").workers(1, 4, 16).executor_fn(run_ml))
    .pool(Pool::new("etl").workers(1, 4, 32).executor_fn(run_etl))
    .build().await?;

sched.submit(Task::new(train).pool("ml").resource("gpu", 2)).await?;        // needs 2 GPUs
sched.submit(Task::new(export).pool("etl").resource("warehouse-db", 1)).await?;
sched.submit(Task::new(features).pool("ml").resource("gpu", 1).resource("warehouse-db", 1)).await?;

In the server configuration:

[[resources]]
kind = "semaphore"
name = "gpu"
permits = 4

How it works

• At dispatch, every resource a task requests is acquired atomically: all units or none.
• If they are not available, the task is parked on that resource and removed from the dispatch queue, so tasks behind it (needing other resources or none) keep flowing. There is no head-of-line blocking.
• Semaphore units are released at the end of each attempt (success, failure, timeout, cancel). Parked tasks are re-offered in FIFO order as units become free, and only when their full request fits.
• A request larger than the resource's capacity is rejected at submission (ResourceExceedsCapacity), instead of waiting forever.

What you get

• Global limits across all pools, workers and machines (with the server), enforced in one place.
• Full utilization of the scarce resource with no static splitting between workers.
• No head-of-line blocking for work that does not need the resource.
• Fair access in FIFO order among tasks waiting for the same resource.
• Multi-resource requests (for example GPU + database connection) without deadlock, because acquisition is all-or-nothing.

High-throughput microtasks

The problem

When tasks take milliseconds (cache refreshes, notifications, small transforms, fan-out of events), the scheduler's own overhead can dominate. Broker-based queues spend more time on network round trips than on the work, and many schedulers rescan or re-sort their queues periodically, so their cost grows with the backlog.

The setup

Embed the scheduler and submit in bulk:

use std::sync::Arc;
use cronoq::{Pool, Priority, Scheduler, Task, TaskError};

let sched = Scheduler::<Event, ()>::builder()
    .pool(Pool::fixed("default", 256).executor_fn(|_ctx, e: Arc<Event>| async move {
        handle(&e).await;
        Ok::<_, TaskError>(())
    }))
    .queue_capacity(65_536)
    .build()
    .await?;

// One round trip for up to thousands of tasks.
let batch: Vec<_> = events.into_iter().map(|e| Task::new(e).priority(Priority::Normal)).collect();
sched.enqueue_many(batch).await?;

Runnable: cargo run --release -p cronoq --example firehose -- 1000000.

Why it is fast

• Constant-time dispatch. Each priority class keeps a FIFO lane. Because every task in a class ages at the same rate, the oldest task is always the best one, so dispatch only compares the heads of at most eight queues: no periodic rescoring, no full-queue scans.
• Single-threaded engine, no locks. The engine runs on one task and receives commands in batches over channels.
• No serialization in-process. Payloads are shared via Arc.
• enqueue and enqueue_many skip result handles when you do not need outputs.

Measured on a shared 4-vCPU VM:

What you get

• Scheduling overhead in the microseconds: choosing the next task looks only at queue heads, so a deep backlog barely slows it down.
• All the scheduling rules still apply (priorities, deadlines, resources, retries) at this rate.
• Backpressure via bounded queues instead of unbounded memory growth during spikes.

Tips

• Prefer enqueue_many over many single enqueue calls; each call is a channel round trip.
• Size queue_capacity to your burst size.
• Disable the otel feature if you do not export metrics.
• See Performance tuning.

Polyglot worker fleets

The problem

The work runs somewhere else: Python model servers on GPU machines, a Go service with access to a private network, a Node.js renderer. Each team reinvents leasing, heartbeats, retries and autoscaling on top of a broker.

The setup

Run cronoq-server and connect workers to it. Producers submit over gRPC or HTTP; workers open one bidirectional stream per process:

producer ──HTTP/gRPC──▶ cronoq-server ◀──Workers.Work stream── worker (Python, GPU box)
                              ▲      ◀──Workers.Work stream── worker (Rust)
                              └──────◀──Workers.Work stream── worker (any gRPC language)

A Python worker serving the gpu pool with two slots:

from cronoq.client import Worker
import cronoq

async def infer(lease):
    if lease.cancelled.is_set():
        return None
    try:
        return run_model(lease.json())
    except OutOfMemory:
        raise cronoq.Retryable("OOM", retry_after=10)

await Worker("cronoq:7070", "gpu", infer, slots=2, token=WORKER_TOKEN).run()

A producer in any language, over HTTP:

curl -X POST https://cronoq.internal/v1/tasks -H "authorization: Bearer $TOKEN" \
  -H 'content-type: application/json' \
  -d '{"pool":"gpu","payload":{"image":"s3://..."},"priority":"high","timeout_s":120}'

The worker protocol

1. The worker opens Workers/Work and sends Register{pool, slots, protocol_version}.
2. The server answers Registered{worker_id, lease_ttl_s, slots}. The pool's capacity grows by the granted slots.
3. The server sends Lease messages, at most slots at a time.
4. For each lease the worker sends Heartbeat well within lease_ttl_s (the bundled clients send one every third of the TTL), then Complete{output}, Fail{message, permanent, retry_after_s}, or Nack to hand the lease to another worker.
5. The server may send Abort{lease} (timeout, cancellation, lease expiry). The worker stops that attempt.
6. When the stream closes, its in-flight leases are retried elsewhere and capacity shrinks. When the server restarts or leadership moves, it sends Shutdown and workers reconnect.

See Writing executors and workers for a full description and the gRPC reference.

What you get

• Language freedom. Any gRPC-capable language can be a worker; producers can use plain HTTP.
• At-least-once execution across machines. Crashed or partitioned workers lose their lease (missed heartbeats) and the attempt is retried on another worker.
• Central policies. Priorities, deadlines, rate limits and retries are enforced by the server, not re-implemented in each worker.
• Elastic fleets. The server computes the desired number of workers per pool and exports it (cronoq_pool_desired_workers) for KEDA or HPA, or calls your webhook. See Cost-aware autoscaling.
• High availability. With a durable store (Postgres, or SQLite on one host), standby servers take over; workers reconnect and accepted work resumes. See Durability and HA.

Cost-aware autoscaling

The problem

Fixed fleets are over-provisioned most of the day and saturated at peak. CPU-based autoscaling reacts to the wrong signal for I/O-bound work: a worker waiting on a model API uses no CPU while hundreds of tasks queue up behind it. And nothing scales ahead of a known event such as a scheduled job.

The setup

Each pool computes a desired capacity from queue pressure every second:

use std::time::Duration;
use cronoq::{AutoscaleConfig, Pool};

let pool = Pool::new("workers")
    .workers(1, 2, 50)                                 // min, initial, max
    .autoscale(AutoscaleConfig {
        target_utilization: 0.75,                      // steady-state headroom
        drain_time: Duration::from_secs(30),           // clear a backlog within 30 s
        slo: [Some(Duration::from_millis(500)),        // critical may wait ≤ 0.5 s
              Some(Duration::from_secs(5)), None, None], // high ≤ 5 s
        stabilization_window: Duration::from_secs(120),// scale down slowly
        cooldown: Duration::from_secs(30),
        cost_per_worker_second: 0.0011,                // e.g. $4/h per worker
        ..Default::default()
    });

• In-process (Rust/Python library), capacity is a number of concurrent task slots and the built-in SlotScaler applies it immediately.
• With the server, capacity is the slots of connected workers. The desired count is exported as cronoq_pool_desired_workers{pool} for KEDA/HPA, or posted to a webhook (scaler = { kind = "webhook", url = "..." }), which then starts or stops workers.

A KEDA ScaledObject for a worker Deployment:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata: { name: llm-workers }
spec:
  scaleTargetRef: { name: llm-workers }      # Deployment of workers with slots=1
  minReplicaCount: 0
  maxReplicaCount: 50
  triggers:
    - type: prometheus
      metadata:
        serverAddress: http://prometheus:9090
        query: max(cronoq_pool_desired_workers{pool="llm"})
        threshold: "1"

The policy

base   = ceil(arrival_rate × mean_service_time / target_utilization)
drain  = ceil(backlog × mean_service_time / drain_time)
slo    = smallest capacity keeping predicted wait ≤ SLO for every class with backlog
desired = clamp(max(base + drain, slo, busy) + reservations_in_horizon, min, max)

Scale-up happens immediately, limited per step to +100% or +4 workers (whichever is larger). Scale-down only goes to the highest desired value seen over the stabilization window, by at most one worker per cooldown. Details: Pools and autoscaling.

What you get

• Scaling on the right signal: queue pressure and wait-time objectives, not CPU.
• Scale-to-zero friendly: with min = 0, idle pools cost nothing; the first arrival raises the desired count.
• Pre-warming for scheduled work: reservations starting within the horizon (60 s by default) add capacity before they start.
• Stable fleets: stabilization and cooldown prevent flapping during bursty traffic.
• Cost reporting: each pool reports worker_seconds and cost, so you can compare policies in money, not just utilization.

Migrating from ElasticPriorityScheduler

cronoq ships a compatibility layer for code written against the Python ElasticPriorityScheduler API (Task, Priority, submit, start, stop, get_metrics, get_cost). It runs on the cronoq engine, so existing code gets the engine's behaviour and performance with a one-line import change.

Step 1: swap the import

# before
from elastic_scheduler import ElasticPriorityScheduler, Priority, Task
# after
from cronoq.compat import ElasticPriorityScheduler, Priority, Task

import time

scheduler = ElasticPriorityScheduler(initial_workers=1, max_workers=4)
scheduler.start()
scheduler.submit(Task(id="report", duration=2.0, priority=Priority.HIGH))
scheduler.submit(Task(id="nightly", duration=0.5, fixed_start=time.time() + 60))
scheduler.submit(Task(id="email", duration=0.2, dependencies=["report"]))
...
print(scheduler.get_metrics())   # QualityMetrics(schedule_quality=..., priority_adherence=...)
scheduler.stop()

The shim is synchronous and thread-safe; it runs the native scheduler on a private event loop in a background thread. As in the original, a task "runs" by sleeping for its duration.

Behavioural differences

Step 2: move to the native API (recommended)

The native API gives you everything the shim hides: real executors instead of sleeps, results, retries, timeouts, cancellation, resources and per-pool configuration.

What you get

• A one-line migration. Existing code keeps its API, threads and call sites; the scheduling decisions move to the cronoq engine.
• Correct semantics where the prototype was approximate. Dependencies are resolved from retained terminal statuses (1 hour by default) instead of the last 1000 completions, deadline urgency is monotonic, and fixed starts are admitted against real capacity instead of conflicting whenever two of them overlap.
• No polling. Dispatch is event-driven: a task starts as soon as a worker is free instead of on the next 100 ms poll, and an idle scheduler uses no CPU.
• A path forward. Once on the native API, the same code gains real executors, results, retries, timeouts, cancellation, shared rate limits, cron schedules, status events and, with cronoq-server, durable state and remote workers.

Tasks and their lifecycle

A task is one unit of work: a payload plus the rules for scheduling it. A pool is a set of worker slots that runs tasks. An executor (in-process) or a remote worker (server mode) runs one attempt of a task at a time; a task may need several attempts.

Task fields

Every task gets a numeric id assigned by the engine. Ids increase monotonically; in server mode the top bits carry the leader's fencing epoch, so ids stay unique across failovers.

Lifecycle

                 ┌── dependencies unmet ──┐
  submit ──▶  Blocked                      │
     │           │ all edges satisfied      │
     ├──────▶ Scheduled ── start time ──┐   │
     │        (fixed start, in future)   ▼   ▼
     └────────────────────────────▶  Ready ◀──────── backoff elapsed ──── Retrying
                                       │ dispatched                        ▲
                                       ▼                                   │ retryable failure,
                                    Running ──────────────────────────────┘ timeout, lease lost
                                       │
      ┌──────────────┬────────────────┼──────────────┬───────────────────────┐
  Succeeded        Failed       DeadLettered     Cancelled        Skipped · Missed · Shed
                 (permanent)   (retries spent)  (by request)   (edge unmet · start missed · overload)

Terminal statuses are retained for the retention window (1 hour, up to 1,000,000 entries by default). During that window the task can still be looked up, keys stay reserved, and new tasks may depend on it.

Results

• Rust: submit returns a TaskHandle<O>; awaiting it yields Ok(Arc<O>) on success or Err(TaskFailure) with the terminal status and message.
• Python: awaiting a TaskHandle returns the executor's return value or raises cronoq.TaskFailed.
• Server: Wait / GET /v1/tasks/{id}/result return the output bytes (and JSON, when the bytes are JSON). Outputs are kept in a bounded in-memory cache and, with a durable store (SQLite or Postgres), in the store.

Handles are not required: enqueue submits without one, which is cheaper.

Events

Every terminal transition is published as a StatusEvent { id, pool, key, status, attempt, at, message }. With emit_transitions enabled, every intermediate transition is published as well. Subscribe with Scheduler::events() in Rust, sched.events() in Python, or Tasks/Watch over gRPC (with all_transitions for intermediate ones). Slow subscribers lose the oldest events rather than slowing the scheduler down.

Priorities and aging

Classes

Effective score

When a worker slot is free, the pool dispatches the ready task with the highest effective score:

score = base[class] + min(aging_rate[class] × wait_seconds, age_cap[class]) + urgency

with defaults aging_rate = 2.0 points per second and age_cap = 800 for every class. urgency is zero unless the task has a deadline. Ties are broken by class, then by submission order.

Aging

Aging prevents starvation: a task that keeps waiting slowly gains score until it overtakes newer work of a higher class.

With the defaults:

• A Low task reaches 10 + 800 = 810 after 400 seconds. Against a continuous stream of fresh Normal work (score about 100), it wins once 10 + 2w > 100, that is after about 45 seconds of waiting.
• Normal can reach 900 and High can reach 1800. Neither crosses the next class's base score (1000 and 10000), so fresh High work always beats aged Normal work, and Critical always beats everything except deadline urgency.

This is deliberate: aging rescues background work from floods of ordinary work, but never lets it delay urgent work. If you want stronger mixing, raise age_cap (Rust: Builder::priority(PriorityConfig { ... })).

The time is measured from submission. A task released by its dependencies, or retried after a failure, keeps its original place in its class.

Why dispatch never rescans the backlog

Within one class, every task ages at the same rate from its own submission time, so an older task always has a score at least as high as a younger one. A FIFO queue per class therefore always has its best task at the head. Deadline tasks live in a second per-class queue ordered by latest start time. In pools with fair sharing, tasks without deadlines live in a per-class fair-share queue ordered by start tag instead of the FIFO lane. Dispatch compares at most eight heads (four classes × two queues), computing their scores lazily. There is no periodic rescoring and no full-queue scan, so choosing the next task costs at most the logarithm of the backlog. Fresh tasks join the back of their lane; a retried or woken task goes back to its place by submission order, which in a very long lane moves the entries behind it.

Measuring adherence

Each pool counts priority inversions: dispatches made while a ready task of a strictly higher class was waiting in the same pool (tasks blocked on resources do not count). priority_adherence = 1 − inversions / dispatches over the metrics window. With aging and deadlines some inversions are intended; a sudden drop usually means an aging or deadline policy is overriding priorities more often than expected.

Configuration

All three are arrays indexed by class (critical, high, normal, low).

Deadlines and urgency

A deadline is a soft "should be done by" time. It does not cancel the task when it passes; it raises the task's score as the deadline approaches so that it is dispatched in time.

Latest start time

d   = duration estimate (duration_hint, else learned median for (pool, kind),
      else the pool's learned median, else the pool's default_duration = 1 s)
LST = deadline − d          latest start time
slack = LST − now
H   = max(d, urgency_window)            urgency_window = 5 s

Urgency curve

urgency = 0                                      if slack ≥ H
        = U × (1 − slack / H)                    if 0 ≤ slack < H
        = U + min(overdue_rate × (−slack), cap)  if slack < 0

with U = 5000, overdue_rate = 100 per second and cap = 3000.

 urgency
 8000 ┤                                  ┌──────────── cap reached
      │                               ╱
 5000 ┤                        ●───╱       overdue: keeps rising
      │                     ╱
      │                  ╱      rising linearly over the horizon H
    0 ┼───────────────╱
      └──────────────┬──────────┬──────────────────▶ time
                  LST − H      LST

The curve is monotonic: a task that is further past its latest start always has at least the urgency of one that is less late.

Ordering

• Within a class, tasks with deadlines are kept in a queue ordered by latest start time (earliest-deadline-first), and the head competes with the class's FIFO lane by full score.
• Across classes, urgency is added to the base score, so a Low task near its deadline (10 + 5000) overtakes fresh High work (1000) but not Critical work (10000).

Choosing values

• Give a duration_hint when you know it; otherwise the first tasks of a kind use the pool's learned typical run time (or default_duration if nothing in the pool has completed yet) until cronoq has observed completions of that kind.
• Deadlines are about finishing. If something must start at a given time, use a fixed start.
• To get more capacity when deadlines are at risk, set per-class wait SLOs in the pool's autoscaling configuration.

Configuration

Fixed starts and reservations

A task with a fixed start should begin at a given time. cronoq treats this as a capacity commitment: it books capacity when the task is submitted and holds a worker shortly before the start.

Task::new(job)
    .start_at(when)                          // or .start_in(Duration)
    .start_tolerance(Duration::from_secs(5)) // default 1 s
    .on_miss(MissPolicy::Skip)               // default RunLate

1. Admission: reservations

The task occupies a reservation interval in its pool:

[start − tolerance,  start + estimated_duration + tolerance]

At any instant, at most floor(max_workers × reservable_fraction) reservations may overlap (reservable_fraction defaults to 0.5, and at least one reservation is always allowed). If the new interval would exceed that, submission fails with ReservationConflict. When max_workers is 2 or more, a fraction below 1 keeps part of the pool free for ad-hoc work. A pool with max_workers = 1 can be fully reserved, because one reservation is always allowed.

2. Holding a worker

Shortly before the start, the task asks for a hold: the next free slot of its pool is kept idle for it instead of taking regular work. How early depends on the pool's own behaviour:

hold_lead = min(max(tolerance, p90 run time of regular tasks in the pool), max_hold)

The idea is to hold a slot just early enough for the regular work that would otherwise take it to finish. The lead is computed from the run times of the pool's tasks without a fixed start, and it is re-evaluated as the start approaches (the first check happens at start − max_hold, max_hold = 5 minutes by default), so a pool whose tasks get slower starts holding earlier without any configuration. If no slot is free when the hold is requested, the first one to free up is held, before any queued task can take it.

Reservations starting within the autoscaler's horizon also add to the pool's desired capacity, so an autoscaled pool grows before the hold is needed.

3. Start

At the start time (and once its dependencies are satisfied) the task runs on its held slot. The recorded start time is the actual dispatch time; the pool's fixed_start_hit_rate counts the task as a hit if |actual − start| ≤ tolerance.

A tolerance of zero is allowed: the task must start exactly at its time (in practice, on the first scheduling step at or after it), and anything later is a miss.

4. Missing the start

If the task has not started by start + tolerance, its miss policy applies:

Typical reasons for a miss: all held capacity is in use by other fixed tasks, the pool has no workers (server mode with no connected workers), or a dependency finished late.

Interaction with other features

• Dependencies: a fixed-start task with unmet dependencies waits for both its time and its dependencies.
• Resources: a fixed-start task also needs its resources; if they are not available at the start time it waits for them while keeping its slot.
• Retries: a retried attempt has no held slot and no miss policy. It waits in the due queue and takes the next free slot ahead of regular work.
• Overload: fixed-start tasks are never shed by the shed_lowest / shed_oldest overflow policies; their capacity was promised at admission.
• Recurring schedules create fixed-start tasks; see Recurring schedules.

Configuration

Recurring schedules

A recurring schedule turns a cron expression into a sequence of fixed-start tasks.

sched.add_recurring(
    Recurring::new("nightly-export", "0 30 2 * * *", Task::new(export).pool("batch"))
        .tz("America/New_York")
        .overlap(OverlapPolicy::Skip)
        .tolerance(Duration::from_secs(60))
        .on_miss(MissPolicy::RunLate),
).await?;
sched.remove_recurring("nightly-export").await?;

await sched.add_recurring("nightly-export", "0 30 2 * * *", payload, pool="batch",
                          tz="America/New_York", overlap="queue", tolerance=60,
                          key="nightly-export", deadline_in=3600)

Over gRPC, use Admin.AddRecurring with an admin token (for example the Python client's Client.add_recurring).

Cron syntax

Five fields (minute hour day month weekday) or six with leading seconds (second minute hour day month weekday). Names (MON, JAN) and nicknames (@hourly, @daily) are accepted. Occurrences fall on whole seconds.

Time zones are IANA names and default to UTC. Daylight-saving transitions follow the zone's rules:

Use UTC for schedules that must run at exact intervals.

How occurrences are created

• Each occurrence is created as a task shortly before it starts: hold_lead + tolerance ahead (capped by the pool's max_hold). Only one occurrence exists ahead of time; the next one is created when the current one is due.
• Each occurrence is a normal fixed-start task: it goes through reservation, holding and the miss policy, carries the tag cronoq.recurring = <name>, and appears in metrics and status events like any other task.
• The template is checked when the schedule is added, like a task submission: its pool and resources must exist, resource amounts must fit their capacity, and max_attempts must be at least 1. A schedule recovered from the store whose template the configuration no longer supports (a removed pool or resource) is not armed; the recurring_invalid counter records it, and it runs again once the configuration supports it.
• If an occurrence cannot be reserved (the pool's reservable capacity is full at that time), it is not created; the snapshot's recurring_conflicts counter records it.

Overlap

A queued occurrence that waits past its start tolerance follows the miss policy, so keep on_miss at its default, RunLate, unless a late run should be dropped.

The overlap guard and the chain survive failover: with a durable store, the occurrence a queued schedule depends on is stored with the schedule, and the occurrence that was running is recognised by its cronoq.recurring tag.

The template

The template's payload, pool, kind, priority, duration hint, timeout, retry policy, resources, tags, labels and tenant are copied to every occurrence. Two options apply per occurrence:

A template may not set an absolute deadline, a fixed start or dependencies: they describe a single task, so add_recurring rejects them (Error::Config in Rust, SchedulerError in Python, INVALID_ARGUMENT on the server). Use Queue to make occurrences depend on each other, and the schedule's tolerance and on_miss (or the template's start tolerance and miss policy) to control how late an occurrence may start.

Downtime and failover

With a durable store, schedules and their next occurrence time are persisted. After downtime:

• if the missed occurrence's policy is RunLate, one late occurrence runs;
• otherwise the missed occurrence is counted as missed and not run;
• the schedule then resumes from the present; missed occurrences are not replayed one by one.

Re-adding a schedule with the same name replaces it without duplicating or overlapping an occurrence that was already created. Removing a schedule does not cancel an occurrence that was already created.

Dependencies and keys

Keys

A key is an optional unique name for a task.

• Deduplication: submitting a task whose key belongs to a live task, or to a terminal task still in the retention window, does not create a new task. It returns the existing id with duplicate = true. This makes producers safe to retry.
• References: dependencies, cancellation, status lookups and result waits accept a key wherever they accept an id.

Keys stay reserved while the task is live and for the retention window after it ends (limits.key_retention_s on the server, CoreConfig::retention in the library; 1 hour by default). Separately, the server deletes finished rows from a Postgres or SQLite store after store.keep_finished_s (7 days by default).

Edges

A dependency is a reference to a parent (id or key) plus a condition:

A task with unmet dependencies is Blocked and does not enter any queue. When a parent ends:

• every satisfied edge decrements the child's count of unmet dependencies; at zero the child becomes eligible;
• an edge that can no longer be satisfied makes the child Skipped, and this propagates to its own dependents.

If the parent had already finished when the child was submitted, the edge is resolved immediately from the retained status.

After a restart with a durable store, a parent that finished so long ago that its status is no longer retained counts as satisfied: it was accepted, and only work that has not finished is recovered.

Unknown parents

A dependency on a key that does not exist yet is handled per pool:

In Python: Pool(unknown_dependency="await", await_timeout=...); on the server: await_unknown_dependencies_s on the pool. Waiting children are tracked per key and cleaned up when they are cancelled, time out or are resolved, so abandoned awaits do not accumulate.

A dependency on an id that does not exist is always rejected.

Cycles

Cycles are rejected with Cycle: a task depending on its own key, cycles within a batch, and cycles closed through a task that is awaiting a key the new submission defines.

Batches

submit_batch admits several tasks atomically. Dependencies may reference keys defined anywhere in the batch, in any order. If any task is invalid, would close a cycle, conflicts with a reservation or exceeds a bound, nothing is admitted and the error names the offending index.

Complexity

Dependencies are tracked with in-degree counters and reverse edges, so a parent finishing costs O(number of its children). Completed parents are never scanned.

Retries, timeouts and leases

cronoq executes tasks at least once: an attempt that may have been lost is retried. Make handlers idempotent, or make their side effects exactly-once with fencing tokens, and deduplicate submissions with keys.

Attempts and outcomes

Each dispatch starts an attempt identified by a lease. An attempt ends with:

Exactly-once side effects

No scheduler can stop a handler from running twice: a worker can lose its connection after charging a card but before reporting success. What cronoq guarantees is enough to make the effect happen once:

• Attempt numbers never repeat for a task, also across failovers: an attempt is dispatched only after it is durably recorded, and a new leader continues from the recorded number. So (task id, attempt) is a fencing token: a later attempt always has a higher number.
• Stale reports change nothing. A completion or failure for an attempt that timed out, was aborted, was already reported or came from a previous leader is ignored, and counted in stale_completions (metric cronoq_stale_completions_total). Lease ids carry the leader epoch, so they never repeat across leaders either.
• Resubmitting a finished keyed task returns it. Within the key retention window (limits.key_retention_s, default 1 hour; CoreConfig::retention in the library), a submission with the key of a finished task is a duplicate: it returns that task and its stored output instead of running again.

Record the token with the side effect, in the same transaction, and let the write succeed only for the newest attempt:

INSERT INTO charges (task_id, attempt, amount) VALUES ($1, $2, $3)
ON CONFLICT (task_id) DO UPDATE SET attempt = excluded.attempt, amount = excluded.amount
WHERE charges.attempt < excluded.attempt;   -- an older attempt arriving late changes nothing

For effects that must happen once at all (not "last attempt wins"), insert on task_id alone and skip when the row exists. The token is ctx.id and ctx.attempt in Rust and Python executors, and Lease.task_id and Lease.attempt for remote workers.

Retry policy

RetryPolicy {
    max_attempts: 3,                          // total, including the first
    initial_backoff: Duration::from_secs(1),
    multiplier: 2.0,
    max_backoff: Duration::from_secs(60),
    jitter: true,                             // full jitter
}

The delay before attempt n (n ≥ 2) is min(initial × multiplier^(n−2), max_backoff); with jitter, a uniformly random fraction of it. A retry_after hint raises the delay to at least that value. The policy is set per pool and can be overridden per task; RetryPolicy::none() (or Task::no_retry()) disables retries.

A retried task keeps its original submission time, so it keeps its place in its priority class.

Timeouts

timeout limits each attempt. When it expires the engine aborts the attempt (the executor's future is cancelled, or the remote worker receives Abort) and records a retryable failure.

Leases and heartbeats

Pools with a lease TTL (always the case for server pools; lease_ttl_s, default 30 s) require the executor or worker to heartbeat each lease:

• Rust: ctx.heartbeat(); Python bindings: ctx.heartbeat(); Python worker client and Rust worker client: automatic, every TTL / 3.
• If no heartbeat arrives within the TTL, the lease expires, the attempt is aborted and treated as a retryable failure. This catches crashed workers and network partitions.

In-process pools have no TTL by default because the engine knows when the attempt's future ends.

Cancellation

cancel(id or key) ends a task immediately with status Cancelled:

• a queued, blocked or retrying task is removed;
• a running attempt is aborted (the cancellation token fires and the future is dropped; remote workers get Abort);
• dependents attached with an on-success or on-failure edge are Skipped; for dependents attached with an on-complete edge, that edge is satisfied.

Shutdown and dropped schedulers

shutdown(grace) stops admitting work, lets running attempts finish for grace, then aborts the rest; queued tasks are not started (with a durable store they stay queued for the next leader). If every Scheduler handle is dropped without shutdown, the runtime shuts down by itself with drop_grace (default 5 s). Handles awaiting unfinished tasks then resolve with TaskFailure::Shutdown (Python TaskFailed(status="shutdown")).

Outputs

A TaskHandle resolves with the executor's output. Handles attached after the task finished (handle(key), a duplicate submission) get the output from a bounded cache of recent outputs (output_cache, 1024 by default) or from the store. If neither still holds it, the handle resolves with TaskFailure::OutputUnavailable (Python: TaskFailed with status succeeded; server: output_available = false): the task did succeed, but its result is gone.

Dead letters

Tasks that end as DeadLettered or Failed keep their final message, available through status events, Wait results and (with a SQLite or Postgres store) the store. Monitor cronoq_tasks_total{status="dead_lettered"} to catch systematic failures.

Recovery after a crash

With a durable store, a task that was running when the leader died counts that attempt as used. If attempts remain, it is retried after a backoff; otherwise it is dead-lettered. See Durability and HA.

Resources: semaphores and rate limits

A resource is a named, shared limit that tasks request units of. Resources are global to a scheduler: tasks in any pool can use the same resource.

Request units on the task: .resource("gpu", 1) (Rust), resources={"gpu": 1} (Python), "resources": {"gpu": 1} (HTTP).

Dispatch rules

1. When a task reaches the front of the queue, all its resources are checked together. If every one has enough units, they are all taken at once (no partial acquisition, so no deadlocks between tasks needing several resources).
2. If any is short, the task is parked on that resource and removed from the dispatch queue. The next task in the pool is considered immediately, so work that does not need the resource is not blocked.
3. When units return (semaphore) or refill (token bucket), parked tasks are re-queued in FIFO order, as many as the available units can satisfy. A token bucket schedules a timer for the moment enough tokens will have accumulated.
4. A request larger than the resource's capacity (permits or burst) is rejected at submission with ResourceExceedsCapacity. Requests for unknown resources fail with UnknownResource.

Choosing between them

• Use a semaphore for things that are held while working: GPUs, connections, licences, concurrent requests to a partner.
• Use a token bucket for rates: requests per second or minute, tokens per minute (request one unit per thousand tokens, for example).
• Combine both for APIs with a rate limit and a concurrency limit.

Interaction with priorities

Parking happens after priority selection, so the most urgent task that can run is dispatched. A parked high-priority task gets the resource as soon as it is available, ahead of later waiters for that resource. Parked tasks are not counted as priority inversions.

Resources and shards

With sharding, each shard counts its resources on its own, so every resource belongs to one shard (shard = "...") and only that shard's tasks may use it. To share an outside limit between shards, define one resource per shard with its share of the limit.

Worker labels

One pool can hold workers that differ: some have a GPU, some run in another region. Labels describe them, and a task can require labels to run only on matching workers.

// The task needs an A100.
sched.submit(Task::new(payload).pool("inference").label("gpu", "a100")).await?;

await sched.submit(payload, pool="inference", labels={"gpu": "a100"})

Worker groups

The workers of a pool that carry exactly the same labels form a group. The scheduler keeps the capacity of each group:

• Remote workers send their labels when they register (Register.labels; WorkerOptions::label in Rust, Worker(labels=...) in Python). The server adds each worker's slots to its group.
• Embedded schedulers report groups with Scheduler::set_group_capacity(pool, labels, n) (Python set_group_capacity). set_capacity sets the unlabeled group.
• A pool's capacity is the sum over its groups; snapshots list each group's capacity and busy slots (PoolSnapshot.groups, /v1/snapshot → pools[].groups).

How tasks are placed

A task whose required labels no free group has waits without blocking the pool: other tasks keep running in the free slots, and the waiting task starts as soon as a matching slot frees or a matching worker connects, in submission order among the tasks waiting for that group. A task that requires labels no worker has waits until one connects (or until it is cancelled; deadlines raise urgency but never cancel).

Each attempt is assigned to a group: remote leases go only to a worker of that group, and embedded executors see it as ctx.group.

Limits

• Fixed-start reservations are counted per pool, not per group: a fixed-start task that requires labels starts when a matching slot is free, so give it enough tolerance if matching workers are scarce.
• Autoscaling targets a pool's total capacity. Scale labeled worker fleets separately, for example one deployment per label set.

Security

Workers declare their own labels, so a worker credential could claim labels it should not have and receive tasks meant for other machines. Restrict what a credential may claim:

[[auth.tokens]]
token_env = "CRONOQ_GPU_WORKER_TOKEN"
role = "worker"
pools = ["inference"]
labels = ["gpu=a100", "zone=eu"]   # a worker with this token may claim only these

A worker that claims a label its credential does not allow is refused with PERMISSION_DENIED.

Fair sharing between tenants

When several teams or customers share a pool, one of them submitting a burst of work should not hold everyone else back. A pool with fair sharing splits its dispatches between tenants by weight.

[[pools]]
name = "llm"
[pools.tenant_weights]   # implies fair sharing; tenants not listed weigh 1
acme = 3
globex = 1

let mut pool = PoolConfig::new("llm");
pool.fairness = Some(Fairness::weights([("acme", 3.0), ("globex", 1.0)]));
sched.submit(Task::new(payload).pool("llm").tenant("acme")).await?;

cronoq.Pool("llm", run, tenant_weights={"acme": 3, "globex": 1})
await sched.submit(payload, pool="llm", tenant="acme")

Set fair_share = true (Python fair_share=True) for equal shares without listing weights.

How it works

• Priority comes first. Fair sharing orders tasks within a priority class: a critical task still runs before any normal one, whatever its tenant.
• Shares follow weights. While several tenants have waiting work in a class, each gets dispatches in proportion to its weight. With weights 3 and 1, acme gets three dispatches for every one of globex's, even if globex queued all of its work first.
• No banked credit. A tenant that had nothing waiting does not catch up later: when it submits again, it shares from that moment on. A tenant alone in the pool gets everything.
• Tasks without a tenant share as one more tenant with the default weight.
• Deadlines keep their order. Tasks with a deadline are ordered by urgency, as in any pool, and are not part of the fair share.
• Shares count dispatches, not time: give long and short tasks different pools if their run times differ widely.

The scheduler uses start-time fair queuing: each task gets a virtual start tag, the later of the class's virtual time and its tenant's previous finish tag, and tasks run in tag order. Dispatch stays logarithmic in the backlog.

Tenants and credentials

The caller names a task's tenant (tenant in the API). To stop a client from spending another tenant's share, pin the tenant on its credentials:

[[auth.tokens]]
token_env = "CRONOQ_ACME_TOKEN"
role = "client"
tenant = "acme"   # tasks from this token are acme's; naming another tenant is refused

A task submitted with such a token gets its tenant filled in; a task that names another tenant is refused with PERMISSION_DENIED.

Pools and autoscaling

Pools

A pool is a group of worker slots with one executor (or one set of remote workers), one autoscaling policy and its own queue bounds. A slot runs one attempt at a time.

The minimum bounds the autoscaler's desired capacity: desired is never below it. It does not set capacity itself. In the library, initial_workers is the capacity at start (at most the maximum, and not raised to the minimum); in the server, capacity comes only from connected workers.

Use several pools to isolate workloads with different executors, hardware or scaling needs; use resources to share limits between them.

Capacity

• In-process (Rust or Python library): a slot is a concurrently running task. The default SlotScaler sets capacity to whatever the autoscaler asks for, immediately.
• Server: capacity is the sum of slots of the workers connected to the pool. The scaler only requests capacity (External: export a metric; Webhook: POST the number), and capacity changes when workers connect or disconnect.
• You can report capacity yourself with Scheduler::set_capacity (Rust) or set_capacity (Python) for externally managed pools.

The queue-pressure policy

Every interval (1 s) each autoscaled pool computes:

λ        arrival rate (EWMA, α = 0.3) of tasks becoming ready
S        mean service time (EWMA of observed run times; default_duration before any)
B        backlog per class: ready tasks plus tasks waiting for a retry
         (fixed-start tasks that are due count as critical)

base     = ceil(λ × S / target_utilization)             target_utilization = 0.75
             (0 when below 0.01: the smoothed rate decays toward zero but never reaches it)
drain    = ceil(ΣB × S / drain_time)                    drain_time = 30 s
slo      = max over classes c with backlog and an SLO:
             ceil(B_≤c × S / slo[c])                    B_≤c = backlog in classes c and higher
reserved = peak concurrent reservations in [now, now + reservation_horizon]   horizon = 60 s

desired  = clamp(max(base + drain, slo, busy) + reserved, min_workers, max_workers)

Then:

• Scale up immediately, by at most max(ceil(current × max_step_up_ratio), max_step_up_abs) per evaluation (defaults: +100% or +4).
• Scale down only to the highest desired value seen over the stabilization_window (60 s), by at most max_step_down (1) per cooldown (30 s).
• Running tasks are never interrupted: when capacity shrinks below the busy count, slots retire as their attempts finish.

Cost

Set cost_per_worker_second and each pool reports worker_seconds and cost, accumulated since start, in snapshots. Cost is also exported as a metric (cronoq_pool_cost_total in Prometheus, cronoq.pool.cost in OpenTelemetry); worker_seconds is in snapshots only.

Configuration reference

The policy is implemented in cronoq-core and is fully deterministic, which makes it testable with simulated time.

Backpressure

Queues in cronoq can be bounded, so that overload turns into explicit backpressure instead of unbounded memory growth.

Bounds

A task counts as pending from submission until it starts running (and again while it waits for a retry). Bounds limit pending tasks:

The library leaves queues unbounded unless you set a bound; the server always bounds them.

Overflow policies

What happens when a submission would exceed a bound is set per pool:

Shedding rules:

• Victims are tasks waiting to run: queued tasks, including those parked waiting for a shared resource. Running tasks are never shed.
• Fixed-start tasks are never shed: their capacity was promised when they were admitted.
• Only when it helps. A bound on the incoming task's own class is never relieved by evicting tasks of other classes, so such submissions are rejected without shedding.
• Only for admissible tasks. All other admission checks (unknown pool or dependency, cycle, reservation conflict) run first, so a submission that would be rejected anyway never evicts anything.
• Not for batches. A batch that does not fit is rejected (or waits, with Wait).
• A shed task ends with status Shed; its dependents are resolved like for any other terminal status.

Choosing a policy

• In-process producers (Wait): natural flow control; a fast producer slows to the pool's pace.
• Network clients (Reject): clients should not hold connections open; they retry with backoff on 429 / RESOURCE_EXHAUSTED.
• Best-effort background work (ShedLowest): important work is always admitted; low-value work is dropped under pressure and you see it in cronoq_tasks_total{status="shed"}.

A waiting submitter whose caller gives up (dropped future, timed-out request) is removed from the wait queue without being admitted.

Metrics and quality signals

cronoq reports two kinds of telemetry:

1. Snapshots: per-pool quality metrics over a sliding window (60 s by default), returned by Scheduler::snapshot() (Rust), await sched.snapshot() (Python), GET /v1/snapshot and Admin/Snapshot.
2. Exported metrics: OpenTelemetry instruments (library, otel feature) and Prometheus text at /metrics (server). See the metrics reference.

Pool snapshot fields

Snapshots also include totals since start (submitted, succeeded, failed, dead_lettered, cancelled, skipped, missed, shed, retries, timeouts, lease_expiries, stale_completions, recurring_conflicts, recurring_invalid).

Reading the signals

Implementation notes

Window metrics use a ring of 12 buckets, and wait/run percentiles come from a log-linear histogram (about 9% resolution). Utilization integrates busy and capacity over time, so it is exact for the window, not sampled.

Durability and high availability

By default the scheduler keeps its state in memory: fast, but a process restart loses queued work. The Store trait makes state durable. Two stores implement it, both with active-passive high availability:

Both pass the same conformance suite, so the behaviour below holds for either.

Model

The engine is the system of record while it runs. The store is a write-ahead log of the engine's state transitions:

• task submissions (full spec and payload), status changes (running, retrying, terminal, with attempt number and message), outputs of successful tasks;
• cancellations;
• recurring schedules, their next occurrence and the last task each one created.

Persisted mutations are group-committed: everything produced in one engine step is written in one transaction, so durability costs one round trip per step, not per task.

Durable acknowledgements

With a persistent store, a call returns only after its effect is committed:

The engine also dispatches an attempt only after the mutations that led to it are durable, so a worker never runs a task that a crash could "unsubmit".

Time

Engine time is Unix time in nanoseconds, so persisted timestamps (fixed starts, deadlines, retry times, next cron occurrences) mean the same thing to every replica and survive restarts without translation.

Recovery

When a scheduler starts with a durable store it:

1. becomes leader (see below) and obtains a new fencing epoch;
2. loads non-terminal tasks, terminal tasks within the store's retention window (for dependencies and key deduplication), terminal tasks that live tasks depend on, and recurring schedules;
3. re-inserts tasks with their original ids, submission order and attempt counts.

Tasks that were running when the previous leader died count that attempt as used: they are retried after a backoff, or dead-lettered if no attempts remain. Fixed-start tasks whose start has already passed apply their miss policy. A dependency on a parent that finished before the retention window counts as satisfied. Recurring schedules keep their overlap guard: an occurrence still running from before the failover is not overlapped by the next one. This is what makes execution at-least-once.

Store failures

The driver distinguishes three kinds of store errors:

A commit that takes longer than commit_timeout (default 30 s) counts as Unavailable. Callers whose operation could not be made durable get Error::Store (Python SchedulerError, server UNAVAILABLE / 503); retry them, ideally with a key so the retry is idempotent.

Leader election and fencing (Postgres)

Several cronoq-server replicas (or embedding processes) can point at the same database:

• Each replica tries to take a Postgres advisory lock on a dedicated session, every lock_poll (0.5 s). The holder is the leader; it increments the epoch and serves. The others wait and report GET /readyz = 503.
• Every commit checks the epoch in the same transaction. A leader that lost its lock (for example after a network partition) can no longer write: its commit fails with Fenced and it stops.
• The leader checks its lock session every keepalive (1 s) and steps down if the session is unresponsive for session_timeout (10 s); the same timeout configures TCP keepalives on the database side, so Postgres releases the lock of a vanished leader promptly.
• When the leader stops, a standby acquires the lock, recovers and starts serving. Task ids carry the epoch in their upper bits, and the store keeps a high-water mark of issued ids, so ids are never reused, even after old rows are pruned.

Clients and workers reach the leader either through a load balancer that routes on /readyz (only the leader is ready), or through one that sends connections to every replica. In the second case a standby answers each call with UNAVAILABLE, and closes every connection about a second after accepting it (and a leader that steps down closes its connections the same way), so a client that retries opens a new connection and the balancer picks a replica again. Workers reconnect on their own. During a graceful shutdown connections stay open, so running attempts can still report. The Helm chart routes to every replica, so that rolling updates can proceed.

Leader election and fencing (SQLite)

Several processes on one host can open the same database file:

• Leadership is a lease row in the file: the holder's identity and an expiry time. A process that finds the lease free or expired takes it, increments the epoch and serves. The others check every lock_poll (0.5 s) and report GET /readyz = 503.
• The leader renews the lease every lease_renew (2 s); it lasts lease_ttl (10 s). If the leader dies or hangs, a standby takes over once the lease runs out. A leader that shuts down cleanly releases the lease, so the standby takes over at once.
• SQLite lets one writer at a time into the file. Every commit checks the epoch and lease holder inside its write transaction, so a leader that was paused past its lease and replaced can no longer write: its commit fails with Fenced and it stops. Its renewal also notices the takeover.
• Lease expiry uses the wall clock, which processes on one host share. Keep the file on a local filesystem: SQLite's WAL journal does not work over network filesystems, which is also why this store does not span hosts.
• The file is created readable and writable by its owner only (on Unix), and SQLite gives its journal files the same permissions. By default every commit syncs the journal to disk (full_sync).

            ┌──────────────┐   advisory lock    ┌──────────────┐
 clients ──▶│ replica A    │◀──────────────────▶│              │
 workers ──▶│ (leader)     │   commits + epoch  │  PostgreSQL  │
            └──────────────┘                    │              │
            ┌──────────────┐  waiting for lock  │              │
            │ replica B    │◀──────────────────▶│              │
            │ (standby,503)│                    └──────────────┘
            └──────────────┘

Schema and migrations

• PostgreSQL tables live in one schema, public by default (PgStoreConfig::schema when you embed the store, store.schema in the server). Instances configured with the same schema form one cluster; different schemas are fully independent, so one database can host several clusters. With SQLite, each file is one cluster.
• Migrations are versioned in cronoq_schema_version and applied automatically on connect, under a lock, by whichever instance starts first. An older build refuses to run against a newer schema instead of corrupting it.
• PostgreSQL text cannot hold NUL characters. The APIs reject them in keys, kinds, tags and names; NUL in messages reported by executors or workers is stored as U+FFFD (in both stores, so they behave alike).
• A row that cannot be decoded during recovery (for example after manual edits) is quarantined: marked failed with an explanatory message, instead of preventing the leader from starting.

Guarantees and limits

Retention: the server prunes finished rows after keep_finished_s (7 days by default). Recovery loads finished tasks from the last retention (1 hour by default) for dependency resolution and key deduplication.

Custom stores

Implement cronoq::Store<P, O> (acquire_leadership, commit, recover, optionally release_leadership and load_output) to persist to another database. Return StoreError::Unavailable for transient failures and StoreError::Fenced from commit when the caller's epoch is stale. Apply the TimeBase you are given when converting timestamps; the runtime always passes Unix-epoch time (unix_nanos_at_zero = 0).

Sharding

One server leads all pools by default, with standbys taking over when it stops (Durability and high availability). When one leader is not enough, or pools should fail independently, split the pools into shards: each shard has its own leader, standbys and store, and shards keep serving when another one is down.

[[pools]]
name = "default"
[[pools]]
name = "llm"

[[shards]]
name = "cpu"
number = 1
pools = ["default"]
endpoint = "https://cronoq-cpu.internal:7070"

[[shards]]
name = "gpu"
number = 2
pools = ["llm"]
endpoint = "https://cronoq-gpu.internal:7070"

[store]
kind = "postgres"
url_env = "DATABASE_URL"   # each shard uses the schema cronoq_<shard>

All servers read the same file and are started for one shard each:

cronoq-server --config cronoq.toml --shard cpu   # 2+ replicas for failover
cronoq-server --config cronoq.toml --shard gpu

What a shard is

• Its own pools. A server runs only its shard's pools.
• Its own resources. Shards do not share counts, so each [[resources]] entry names the shard that owns it (shard = "gpu"), and only that shard's tasks may use it; a task of another shard asking for it is refused. To share an outside limit between shards, split it into one resource per shard with its share: api_cpu with 6 calls per second in cpu, api_gpu with 4 in gpu. [limits], such as max_pending_total, apply to each shard separately.
• Its own store. Postgres keeps each shard in its own schema (store.schema must contain {shard} and defaults to cronoq_{shard}); SQLite in its own file (store.path must contain {shard}). A server refuses to start on a store that holds another shard's tasks. Start a sharded deployment on new stores: an unsharded server's tasks are not always recognised, and moving them into a shard is not supported.
• Its own ids. Task ids carry the shard's number in their top 8 bits, so ids never collide between shards, and any server can tell which shard a task belongs to.
• Its own keys. Keys deduplicate within a shard; two shards can each have a task with the same key.

Calling any server

Clients can call the server of any shard. It answers from every shard that the call concerns:

• Work for another shard is forwarded to it: submissions and batches for its pools, lookups, waits and cancellations of its tasks, and schedules for its pools. Lookups by key ask every shard, so the answer does not depend on which server receives the call: a key that names tasks in two shards is refused (FAILED_PRECONDITION, use the task id). While a shard cannot be asked, Get and Wait by key still find a task that one reachable shard holds and name the missing shards (cronoq-unavailable-shards metadata; unavailable_shards over HTTP), but Cancel by key is refused with UNAVAILABLE, since the missing shard might hold the same key. Keys are unique only within a shard, so prefer ids for anything that changes a task, or keep each key's tasks in one shard's pools.
• Reads cover every shard: task listings (paged across shards in id order), Watch, worker and schedule listings, and snapshots. When a shard cannot be reached, the others still answer, and the response names the missing shard in unavailable_shards. A Watch ends when any shard's stream ends; reconnect. Prometheus /metrics stays per server: scrape each shard.

A forwarded call carries the caller's own bearer token, and the other shard authenticates and authorizes it again, so forwarding never grants more than the caller's own credentials. Each shard's server sees the tokens of the callers it forwards; protect every shard's servers as you would one server. Forwarded calls are marked (cronoq-forwarded) and answered by the receiving shard only, so they never travel further.

Configure how servers reach each other's https:// endpoints with [server.peer_tls] (ca, and cert / key for shards that require mTLS); the system roots are the default.

When a call is not forwarded

Some calls are refused instead, with FAILED_PRECONDITION (HTTP 409) naming the shard and its endpoint in the message and in the response metadata (cronoq-shard, cronoq-shard-endpoint), so the caller can go there directly:

• Client certificates. A caller that authenticated with a certificate has no token to forward. Its reads cover only the receiving server's shard (the others are listed as unavailable), and its Watch streams only that shard's events.
• Plain http:// to another host. Tokens are not sent unencrypted across the network: calls for a shard whose endpoint is http:// on another host are refused with its endpoint, unless server.insecure_peer_http = true (trusted networks only). Loopback endpoints and https:// endpoints are always used.
• Workers register with the shard of their pool. A worker that connects to another shard's server is refused with the endpoint to use, and the Rust, Python, Go and TypeScript workers connect there on their own (up to three redirects in a row). They keep their TLS settings, never follow a redirect from https:// to http://, and send their token over plain http:// only to loopback unless the worker allows it (insecure_redirect_http, trusted networks only). When that shard cannot be reached, the worker asks its configured endpoint again, so a load balancer in front of all shards works as the worker's endpoint.

Credentials are checked first: a caller that may not use a pool learns nothing about where it is served.

What cannot span shards

• Dependencies on another shard's task (INVALID_ARGUMENT): a task can only wait for tasks of its own shard. Dependencies by key refer to keys of the task's own shard.
• Batches with tasks of several shards (INVALID_ARGUMENT): a batch is admitted atomically by one shard.
• Resources of another shard (INVALID_ARGUMENT).

Writing executors and workers

An executor runs attempts in-process (library mode). A worker runs attempts in another process and talks to cronoq-server (server mode). The rules are the same: run the payload, report an outcome, respect cancellation, and heartbeat long attempts when a lease TTL applies.

Rust executors

Most executors are closures:

Pool::new("default").executor_fn(|ctx: TaskContext, job: Arc<Job>| async move {
    tokio::select! {
        r = do_work(&job) => r.map_err(TaskError::from_err),
        _ = ctx.cancelled() => Err(TaskError::Cancelled),
    }
})

For more control, implement the trait:

use std::sync::Arc;
use cronoq::{BoxFuture, Executor, LeaseId, TaskContext, TaskError};

struct HttpExecutor { client: reqwest::Client }

impl Executor<Request, Response> for HttpExecutor {
    fn execute(&self, ctx: TaskContext, req: Arc<Request>) -> BoxFuture<Result<Response, TaskError>> {
        let client = self.client.clone();
        Box::pin(async move { /* ... */ })
    }

    // Optional: release per-lease resources synchronously when the engine abandons an attempt.
    fn abort(&self, _lease: LeaseId) {}
}

TaskContext provides id, lease, attempt, pool, kind, key, tags, deadline, timeout, is_cancelled(), cancelled(), cancellation_token(), heartbeat(), tag(name) and traceparent().

Guidelines:

• Be idempotent. Execution is at-least-once; record (ctx.id, ctx.attempt) with side effects that must happen once (fencing tokens).
• Classify errors. Use permanent for errors a retry cannot fix (validation, 4xx), and retry_after when the remote side says when to come back.
• Do not block the runtime. Use tokio::task::spawn_blocking for CPU-heavy or blocking work.
• Check ctx.deadline to pick a faster path when time is short.
• Panics are contained. A panicking executor fails its task permanently; the scheduler and other attempts keep running.
• Heartbeat when the pool has a lease TTL. Long attempts call ctx.heartbeat(), or they are failed as retryable when the TTL expires.

Python executors

async def run(ctx: cronoq.TaskContext, payload):
    if ctx.attempt > 1:
        log.info("retrying %s", ctx.id)
    for chunk in work(payload):
        await process(chunk)
        ctx.heartbeat()                 # only needed with lease_ttl
    return result

Plain (non-async) functions run in the event loop's default thread pool. When the engine cancels an attempt (cancellation, timeout, lease loss) the executor's asyncio task is cancelled, so asyncio.CancelledError is raised at the next await.

Remote workers: Python

import os
import cronoq
from cronoq.client import CancelTask, Reject, Worker

async def handle(lease):
    # lease.id, .task_id, .attempt, .payload (bytes), .json(), .text(), .content_type, .kind,
    # .key, .tags, .timeout, .deadline, .pool, .lease_ttl, .cancelled (asyncio.Event)
    if overloaded():
        raise Reject("local GPU busy")              # hand it to another worker
    try:
        return await process(lease.json())          # bytes, str, JSON-serialisable or None
    except TransientError as e:
        raise cronoq.Retryable(str(e), retry_after=5)
    except ValidationError as e:
        raise cronoq.Permanent(str(e))

worker = Worker("cronoq:7070", "default", handle, slots=8,
                token=os.environ["CRONOQ_WORKER_TOKEN"], tls=True, grace=30)
await worker.run()          # until worker.stop()

The worker:

• runs at most slots handlers at once, or fewer if the server grants fewer (worker.granted_slots);
• heartbeats every running lease every lease_ttl / 3;
• cancels the handler's task and sets lease.cancelled when the server aborts an attempt;
• reconnects with exponential backoff when the stream drops, and right away when the server announces a leadership change;
• connects to the right shard when the pool belongs to another one (Sharding); pass insecure_redirect_http=True only on trusted networks where shards use plain http://;
• on stop(), stops taking leases, lets running ones finish for up to grace seconds, then cancels the rest.

Pass labels={"gpu": "a100"} to register the worker with labels: tasks submitted with labels= run only on workers that have them (Worker labels).

Raise Reject to give a lease back unrun: the server offers it to another worker of the pool with the same labels without counting an attempt, or, if no such worker can take it, fails the attempt as retryable. Raise CancelTask to end the task as cancelled. For mTLS, pass credentials=grpc.ssl_channel_credentials(root, key, cert).

Remote workers: Rust

use bytes::Bytes;
use cronoq::TaskError;
use cronoq_server::worker::{run_worker, WorkerOptions, WorkerTls};

let opts = WorkerOptions::new("https://cronoq.internal:7070", "default")
    .slots(8)
    .token(token)
    .name("w1")
    .tls(WorkerTls { ca_pem: Some(ca), identity: None, domain: None })
    .drain_timeout(Duration::from_secs(60));
run_worker(opts, |lease| async move {
    // lease.lease, task_id, attempt, pool, payload, content_type, kind, key, tags,
    // timeout, deadline, cancel (a CancellationToken)
    tokio::select! {
        out = process(&lease.payload) => out.map(Bytes::from).map_err(TaskError::from_err),
        _ = lease.cancel.cancelled() => Err(TaskError::Cancelled),
    }
}, shutdown_signal()).await?;

run_worker enforces the granted slots, heartbeats each lease, cancels lease.cancel on Abort, turns a panicking handler into a retryable failure, reconnects with backoff (initial_backoff 200 ms up to max_backoff 10 s), and when shutdown resolves drains running handlers for up to drain_timeout (30 s). It returns an error only for problems retrying cannot fix, such as a rejected token or an unknown pool. Use identity: Some((cert_pem, key_pem)) for mTLS. TLS settings need an https:// endpoint: with http:// the worker returns an error rather than connecting in plaintext. .label("gpu", "a100") registers the worker with a label. When another shard serves the pool, the worker connects to that shard's endpoint (.insecure_redirect_http(true) allows plain http:// to other hosts).

Go workers

The Go SDK runs workers with cronoq.RunWorker:

err := cronoq.RunWorker(ctx, cronoq.WorkerOptions{
	Target: "http://127.0.0.1:7070", Pool: "default", Slots: 8, Token: token,
}, func(ctx context.Context, l *cronoq.Lease) ([]byte, error) {
	return process(ctx, l.Payload)
})

It enforces the granted slots, heartbeats each lease, ends the handler's context on Abort, turns a panic into a retryable failure, reconnects with backoff, follows its pool to the right shard, and when ctx ends drains running handlers for up to DrainTimeout (30 s).

TypeScript workers

The TypeScript SDK runs workers in Node with runWorker from @synaptiai/cronoq/node:

await runWorker(
  { target: "http://127.0.0.1:7070", pool: "default", slots: 8, token, signal: stop.signal },
  async (lease) => handle(lease.json()),
);

It enforces the granted slots, heartbeats each lease, aborts lease.signal on Abort, reconnects with backoff, follows its pool to the right shard, and when signal aborts drains running handlers for up to drainTimeout (30 s).

Any language over gRPC

Workers in other languages implement one bidirectional stream, cronoq.v1.Workers/Work, from cronoq.proto. Print it with cronoq-server --print-proto and generate stubs with your language's protobuf tooling.

1. Open the stream with metadata authorization: Bearer <worker token> (or a client certificate).
2. Send register{pool, slots, worker_name, protocol_version: 3, labels} first. If the stream fails with FAILED_PRECONDITION and cronoq-shard-endpoint metadata, another shard serves the pool: connect there instead, without moving from https:// to http://.
3. Receive registered{worker_id, lease_ttl_s, slots, protocol_version, max_message_bytes}; run at most the granted slots leases at once.
4. For each lease:
   • run the task (payload, content_type, kind, key, tags, timeout_s, deadline_unix_ns);
   • send heartbeat{lease} well within lease_ttl_s while it runs;
   • finish with exactly one of complete{lease, output}, fail{lease, message, permanent, retry_after_s}, nack{lease, reason} (give it back) or cancelled{lease}.
5. On abort{lease, reason}, stop that attempt; its slot is already free.
6. Send set_slots{slots} to change concurrency; 0 drains the worker.
7. On shutdown{reason, not_leader}, or when the stream ends, reconnect with backoff.

In-flight leases of a dropped stream are failed as retryable and retried elsewhere. The full rules are in the gRPC reference.

Using the HTTP API

The HTTP gateway exposes the client side of the gRPC API as JSON, so any language with an HTTP client can submit and inspect work. It listens on server.http_addr (default 127.0.0.1:7071). When authentication is enabled, every /v1 route needs authorization: Bearer <token>: the /v1/tasks routes need the client role, and /v1/snapshot, /v1/workers and /v1/recurring need the metrics role (admin works for all of them). Request bodies need content-type: application/json.

export CRONOQ=http://localhost:7071
export AUTH="authorization: Bearer $CRONOQ_CLIENT_TOKEN"
export JSON="content-type: application/json"

Submit

curl -s $CRONOQ/v1/tasks -H "$AUTH" -H "$JSON" -d '{
  "payload": {"prompt": "summarize ..."},
  "pool": "llm",
  "kind": "summarize",
  "priority": "high",
  "key": "summary-42",
  "deadline_in_s": 30,
  "timeout_s": 60,
  "retry": {"max_attempts": 5, "initial_backoff_s": 0.5},
  "resources": {"openai": 1},
  "tags": {"tenant": "acme"}
}'
# {"duplicate":false,"id":"17"}

• payload is any JSON value (sent to workers as application/json); use payload_b64 for binary data.
• key makes the submission idempotent: sending it again returns the same id with "duplicate": true. Use it for every request you might retry.
• Add "wait": true (and optionally "wait_timeout_s") to block until the task finishes and include its result.
• Unknown fields are rejected, so typos fail loudly instead of being ignored.

Dependencies and batches

curl -s "$CRONOQ/v1/tasks:batch" -H "$AUTH" -H "$JSON" -d '{"tasks": [
  {"payload": "extract", "key": "extract"},
  {"payload": "load", "key": "load", "after": ["extract"]},
  {"payload": "alert", "deps": [{"key": "load", "cond": "on_failure"}]}
]}'
# {"tasks":[{"duplicate":false,"id":"18"},{"duplicate":false,"id":"19"},{"duplicate":false,"id":"20"}]}

after accepts keys (strings) and ids ({"id": "18"} or a number). A batch is all-or-nothing, and its tasks may reference each other in any order. alert runs only if load fails; if load succeeds, alert ends as skipped.

Status, results, cancellation

Ids and keys have separate routes:

curl -s $CRONOQ/v1/tasks/17 -H "$AUTH"
curl -s $CRONOQ/v1/tasks/by-key/summary-42 -H "$AUTH"
# {"attempt":1,...,"id":"17","key":"summary-42","kind":"summarize",...,"pool":"llm",...,"status":"running"}

curl -s "$CRONOQ/v1/tasks/17/result?timeout_s=60" -H "$AUTH"
# {"id":"17","output":{...},"output_available":true,"output_b64":"...","status":"succeeded"}

curl -s -X DELETE $CRONOQ/v1/tasks/by-key/summary-42 -H "$AUTH"
# {"cancelled":true}

Percent-encode keys that contain / (report%2F2024). A result request that times out returns 504 while the task keeps running: ask again.

Scheduled work

# Start in 10 minutes; skip it if it cannot start within 30 s of that time.
curl -s $CRONOQ/v1/tasks -H "$AUTH" -H "$JSON" -d '{
  "payload": {"report": "daily"}, "start_in_s": 600, "tolerance_s": 30, "on_miss": "skip"}'

The server reserves capacity for the start time, and holds a worker slot free shortly before it. Recurring schedules are managed through the gRPC Admin service with an admin token, for example with the Python client's add_recurring.

Operations

curl -s $CRONOQ/v1/snapshot -H "authorization: Bearer $CRONOQ_METRICS_TOKEN"  # pools, waits, adherence, cost
curl -s $CRONOQ/metrics -H "authorization: Bearer $CRONOQ_METRICS_TOKEN"      # Prometheus text
curl -s $CRONOQ/healthz                  # process is up
curl -s $CRONOQ/readyz                   # 200 on the leader, 503 on standbys

Set server.metrics_addr to serve /metrics without a token on a separate, internal port.

Errors

Errors are JSON {"error": "...", "code": "..."}:

Full details: HTTP API reference.

Command-line tool

cronoq talks to a running cronoq-server over gRPC. It is in the release archives next to cronoq-server, in the container image (/usr/local/bin/cronoq), and installs with cargo install cronoq-cli.

Connecting

There is no --token flag: values on a command line are visible to other users of the machine. Use https:// with TLS; the TLS flags are refused with an http:// server rather than silently ignored, and a token sent over plain http:// to a host other than loopback prints a warning.

Each command needs a role (Security hardening): task commands need client, workers, schedules and snapshot need metrics, and schedule add / schedule remove need admin. An admin token works everywhere.

Commands

cronoq submit "hello"                       # prints the task id
cronoq submit '{"doc": 42}' --json --pool llm --priority high --tenant acme \
    --label gpu=a100 --tag team=search --deadline-in 30 --key doc-42
cronoq submit --file report.pdf --kind ocr --wait > text.txt
cronoq get 17                               # details; --key to look up by key
cronoq get --key doc-42
cronoq wait 17 --timeout 60                 # prints the output
cronoq cancel 17
cronoq tasks --pool llm --status running --status failed --limit 50
cronoq tasks --all                          # follow every page
cronoq workers
cronoq schedules
cronoq schedule add nightly "0 0 3 * * *" '{"report": "daily"}' --json \
    --tz Europe/Berlin --overlap queue --pool reports --key nightly
cronoq schedule remove nightly
cronoq snapshot
cronoq watch --pool llm --all               # every transition, until Ctrl-C

submit sends a string argument as UTF-8 text, --json as JSON (checked first), and --file PATH or - (stdin) as bytes; --content-type overrides the type the worker sees. --after TASK (repeatable, an id or a key) makes the task wait for another to succeed.

wait and submit --wait write the task's output to standard output unchanged when it is redirected to a file or a pipe. Keys, messages and other text come from other clients of the server, so on a terminal, control characters in them (and in outputs) are shown as escapes such as \u{1b} instead of being sent to the terminal.

-o json prints JSON shaped like the HTTP API, with ids and nanosecond times as strings so that tools reading numbers as doubles do not round them.

Exit codes:

cronoq top

cronoq top refreshes a full-screen view every second (--interval):

• Pools: capacity, busy slots, pending tasks, waits at the median and 90th percentile, utilization, and tasks succeeded and attempts failed over the metrics window (the last minute).
• Workers: name, pool, busy and total slots, labels, and when each one connected.
• Schedules: cron expression, pool and next occurrence.
• Live tasks: running tasks first, then waiting ones, up to --tasks (200).
• Recent events: the latest 200 status transitions, newest first, with the task's key and message. The stream reconnects by itself after errors or a leader change.

--pool limits the task and event panes to some pools.

Press q to quit and r to refresh now. The pool, worker and schedule panes need the metrics role, the task and event panes the client role. With a token that has only one of them, the other panes say which role they need; use an admin token to see everything. When the server is a standby or restarting, the panes show the error and recover on the next refresh.

Deploying the server

Container image

Releases publish a multi-architecture image (linux/amd64, linux/arm64) to ghcr.io/synaptiai/cronoq, tagged with the version (0.2.1) and the minor series (0.2). The image is distroless: it contains the cronoq-server binary, the cronoq CLI and the example configuration, runs as a non-root user, and has no shell or curl. Run it with your own configuration:

docker run -p 7070:7070 -p 7071:7071 \
  -v $PWD/cronoq.toml:/etc/cronoq/cronoq.toml:ro \
  -e CRONOQ_CLIENT_TOKEN -e CRONOQ_WORKER_TOKEN -e CRONOQ_METRICS_TOKEN -e CRONOQ_ADMIN_TOKEN \
  ghcr.io/synaptiai/cronoq:0.2.1

To build the image from a checkout instead, run docker build -t cronoq ..

The configuration path comes from --config or CRONOQ_CONFIG (the image sets /etc/cronoq/cronoq.toml). Every token_env the configuration names must be set, or the server refuses to start. Log verbosity follows RUST_LOG (default info). The server stops gracefully on SIGTERM or SIGINT: it stops admitting work, gives running attempts shutdown_grace_s (default 10 s) to finish, then tells workers to reconnect elsewhere.

Validate configuration changes before rolling them out:

docker run --rm -v $PWD/cronoq.toml:/c.toml:ro ghcr.io/synaptiai/cronoq:0.2.1 --config /c.toml --check

Kubernetes

The Helm chart in deploy/helm/cronoq deploys the server with standbys, a store, tokens, probes and, if you use them, shards. Releases publish it, and the operator's chart, to oci://ghcr.io/synaptiai/charts; the examples below install from a checkout (deploy/helm/cronoq), and oci://ghcr.io/synaptiai/charts/cronoq --version 0.2.1 works the same way.

Create the tokens (one key per role) and a Secret with the database URL (postgres://USER:PASSWORD@HOST:5432/DB?sslmode=verify-full, key url), then install:

kubectl create secret generic cronoq-tokens \
  --from-literal=client="$(openssl rand -hex 24)" --from-literal=worker="$(openssl rand -hex 24)" \
  --from-literal=metrics="$(openssl rand -hex 24)" --from-literal=admin="$(openssl rand -hex 24)"
kubectl create secret generic cronoq-db --from-file=url=./database-url
helm install cronoq deploy/helm/cronoq \
  --set auth.existingSecret=cronoq-tokens --set store.postgres.urlSecret.name=cronoq-db

Pools, resources and limits go in config, as the TOML that follows the chart's own tables:

config: |
  [[pools]]
  name = "default"
  max = 50
  lease_ttl_s = 30
  [[pools]]
  name = "llm"
  max = 200
  lease_ttl_s = 120

Every pod is ready as soon as it is up (readiness uses /healthz), so the Service routes to the leader and the standbys, and rolling updates proceed. A standby answers UNAVAILABLE (503 over HTTP) and closes each connection about a second after accepting it, so a client that retries reconnects and the Service picks a pod again until it reaches the leader; workers reconnect on their own. /readyz still answers 200 only on the leader, for your own checks. The pods run as non-root with a read-only root filesystem and no service-account token.

gRPC streams are long-lived. If an ingress or L7 load balancer sits in front, enable HTTP/2 (gRPC) and raise idle timeouts above your longest wait; the server sends HTTP/2 keepalive pings every 20 s (limits.keepalive_interval_s).

Shards in the chart

shards:
  - { name: cpu, number: 1, pools: [default] }
  - { name: gpu, number: 2, pools: [llm] }
store:
  postgres:
    schema: "cronoq_{shard}"   # each shard gets its own tables

Each shard gets its own Deployment, Service (<release>-cronoq-<shard>, or <release>-<shard> when the release name contains cronoq; the shard's endpoint) and disruption budget, from one shared configuration. The main Service reaches every shard's servers, and any server answers for every shard (Sharding).

Servers forward callers' tokens to other shards, and workers follow redirects to them, so shard endpoints must be encrypted or explicitly trusted:

• with tls.secretName, the endpoints use https:// and the servers verify each other with ca.crt; the certificate must name every shard's Service;
• without TLS, set insecurePeerHttp: true (trusted cluster networks only); workers then need their own insecure-redirect option (insecure_redirect_http, InsecureRedirectHTTP, insecureRedirectHttp). The chart refuses shards with neither.

Without the chart

Write the configuration yourself (bind to 0.0.0.0 inside the pod, which requires credentials) and run the image with probes on the HTTP port: /healthz for liveness, and for readiness either /healthz (all pods ready, as in the chart) or /readyz (only the leader; rolling updates of a Deployment with standbys then stall, so use strategy: Recreate). With [server.tls], set scheme: HTTPS on the probes.

Durable state on one host with SQLite

For a single machine (a VM, a bare-metal host, a container with a persistent volume), keep state in a file:

[store]
kind = "sqlite"
path = "/var/lib/cronoq/state.db"   # a local filesystem, not NFS
keep_finished_s = 604800            # prune finished tasks after 7 days

A second server process pointed at the same file stands by and takes over when the first stops (at once on a clean shutdown, after the 10 s lease otherwise). Back the file up with SQLite's online backup (sqlite3 state.db ".backup backup.db"), not by copying it while it is in use.

High availability with Postgres

Add a store and run two or more replicas:

[store]
kind = "postgres"
url_env = "DATABASE_URL"   # postgres://...?sslmode=verify-full
keep_finished_s = 604800   # prune finished tasks after 7 days

spec:
  replicas: 2

Standbys poll for the leader lock every 0.5 s. The leader checks its lock session every second and steps down if the session stays unresponsive for 10 s; when its session ends (process exit, pod deletion, a connection terminated by Postgres, or that timeout), a standby takes the lock, recovers all accepted work and becomes ready. Clients retry on UNAVAILABLE / 503; workers receive Shutdown{not_leader} and reconnect. See Durability and HA.

The database role needs permission to create tables in store.schema on first start (public by default; the chart sets cronoq), and to create that schema if it does not exist. Migrations are versioned and applied automatically, under a lock, by whichever replica starts first; replicas of an older version refuse to run against a newer schema.

Scaling workers with the operator

Run workers as their own Deployment (one worker process per pod, with several slots each). The operator in deploy/helm/cronoq-operator sets each Deployment's replicas from its pool's desired capacity. Install it once per cluster, naming the servers it may read and a token with the metrics role for each:

helm install cronoq-operator deploy/helm/cronoq-operator \
  --set 'servers[0].name=main' --set 'servers[0].url=http://cronoq.cronoq:7070' \
  --set 'servers[0].tokenSecret.name=cronoq-tokens' --set 'servers[0].tokenSecret.key=metrics'

Then add a scaler next to each worker Deployment:

apiVersion: cronoq.io/v1alpha1
kind: CronoqWorkerScaler
metadata: { name: default-workers }
spec:
  server: main              # a server from the operator's configuration
  pool: default
  deployment: default-workers
  slotsPerReplica: 4        # the slots each worker pod registers
  minReplicas: 0
  maxReplicas: 100

Every intervalSeconds (15 by default) the operator reads the pool's desired capacity in slots and sets the Deployment's replicas to ceil(desired / slotsPerReplica), clamped to [minReplicas, maxReplicas], through its scale subresource. When the server cannot be reached, or does not report the pool (for example while its shard is unreachable), the operator leaves the Deployment alone; kubectl get cronoqworkerscalers shows the desired slots, the replicas set and why. Give each Deployment one scaler, and no other autoscaler.

The server URLs and tokens live in the operator's configuration, so a scaler can only name a configured server; the operator has no access to Secrets through the API. Its ClusterRole allows reading scalers, writing their status, and reading and changing deployments/scale. Anyone who may create a scaler in a namespace can therefore have the operator scale any Deployment in that namespace, within the scaler's bounds: grant cronoqworkerscalers only to those who may scale the Deployments there. A scaler's status also shows the desired slots of the pool it names, read with the operator's metrics token, so such users can read that counter for any pool on the configured servers.

Autoscaling workers with KEDA

Without the operator, scale a worker Deployment from the server's desired-capacity metric:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata: { name: cronoq-default-workers }
spec:
  scaleTargetRef: { name: default-workers }
  minReplicaCount: 0
  maxReplicaCount: 100
  pollingInterval: 5
  triggers:
    - type: prometheus
      metadata:
        serverAddress: http://prometheus.monitoring:9090
        # desired slots ÷ slots per pod
        query: ceil(max(cronoq_pool_desired_workers{pool="default"}) / 4)
        threshold: "1"

cronoq_pool_desired_workers already accounts for queue pressure, wait-time objectives and upcoming fixed-start reservations, and is smoothed against flapping. Alternatively, set scaler = { kind = "webhook", url = "...", token_env = "..." } on the pool and have your own service create workers when it receives {"pool": ..., "desired": n}.

Give worker pods a terminationGracePeriodSeconds longer than their drain timeout, so running leases finish before the pod is killed.

Sizing

The scheduler itself is light: one core comfortably handles hundreds of thousands of task transitions per second. Memory is dominated by pending tasks (a few hundred bytes plus the payload each), the output cache ([outputs], 64 MiB by default) and the task-info cache. With a durable store, the leader writes the mutations queued since its last commit in one transaction; replies wait for that commit.

Checklist

• One token per role and consumer, delivered through secrets (token_env), scoped to pools.
• TLS on both listeners (or a service mesh), and mTLS for workers if required.
• /healthz for liveness; for readiness /healthz with rolling updates (standbys answer UNAVAILABLE, which clients retry) or /readyz (only the leader) with strategy: Recreate.
• Prometheus scraping /metrics with a metrics token or on metrics_addr; alerts on dead letters, wait times and missing workers.
• max_pending and overflow set on pools fed by untrusted producers.
• sslmode=verify-full in the database URL.
• Workers drain on shutdown, with a termination grace period to match.
• cronoq-server --check in CI for configuration changes.

Observability

cronoq tells you not only how busy it is, but how well it is scheduling: whether priorities are respected, whether fixed starts are on time, and how long work waits. This guide shows how to collect those signals; Metrics and quality signals explains how to read them.

Prometheus (server)

GET /metrics serves Prometheus text built from the scheduler snapshot at scrape time. On the main HTTP port it requires a token with the metrics (or admin) role:

scrape_configs:
  - job_name: cronoq
    authorization: { credentials_file: /etc/prometheus/cronoq-metrics-token }
    static_configs: [{ targets: ["cronoq:7071"] }]

If your scraper cannot send a token, set server.metrics_addr = "0.0.0.0:9090" and scrape that port instead; it serves /metrics, /healthz and /readyz without authentication or TLS, so keep it on an internal network.

Standbys answer 503; scrape every replica and the leader's series are the live ones. Every metric is listed in the metrics reference.

OpenTelemetry (library)

With the default otel feature, the cronoq crate records metrics through the globally installed OpenTelemetry meter provider and propagates trace context through the global text-map propagator. Configure both once at startup:

use opentelemetry::global;
use opentelemetry_sdk::{metrics::SdkMeterProvider, propagation::TraceContextPropagator};

global::set_text_map_propagator(TraceContextPropagator::new());
let meter_provider = SdkMeterProvider::builder()
    .with_periodic_exporter(opentelemetry_otlp::MetricExporter::builder().with_tonic().build()?)
    .build();
global::set_meter_provider(meter_provider);
// Build the scheduler after this, so its instruments use the provider.

Without a provider the instruments cost almost nothing. Disable the feature (default-features = false) to drop the OpenTelemetry dependencies entirely.

Traces

Each attempt runs inside a tracing span named cronoq.task with fields task.id, task.pool, task.kind and task.attempt. When a task is submitted inside a span and a propagator is installed, its W3C traceparent is stored as a tag; each attempt's span is then parented to the submitter's trace (with tracing-opentelemetry installed), so retries show up under the request that caused them. Remote workers receive the tag in lease.tags["traceparent"] and can continue the trace.

Snapshots and events

For dashboards inside your application, read snapshots directly:

snap = await sched.snapshot()
llm = snap.pool("llm")
print(llm.capacity, llm.busy, llm.wait_p99, llm.priority_adherence, snap.totals["dead_lettered"])

and subscribe to status events for audit logs or user notifications:

async with sched.events() as events:
    async for e in events:
        if e.status == "dead_lettered":
            await alert(e.id, e.message)

The server equivalents are GET /v1/snapshot / Admin.Snapshot and the Tasks.Watch stream.

Key alerts

Logs

The server logs with tracing to stdout; set RUST_LOG (for example RUST_LOG=info,cronoq_server=debug). Notable events: listening addresses, leadership acquired or lost, worker registration and disconnection (with granted slots), store outages and retries, failed scale requests, and configuration warnings such as short tokens or a database URL without sslmode=verify-full.

Security hardening

This guide is the checklist for running cronoq-server outside a laptop. The threat model (what the server protects and what it does not) is in SECURITY.md.

Roles

Every token or certificate mapping can be limited to pools = [...]. Scoped credentials cannot submit into, serve, watch or look up tasks of other pools; those tasks look like they do not exist. Keys, however, are one namespace across pools (see the threat model).

Checklist

1. Configure credentials. Without [[auth.tokens]] or [[auth.certs]], authentication is off: the server then refuses non-loopback addresses unless you pass --insecure-no-auth, and its HTTP API answers only requests whose host is localhost, *.localhost, 127.0.0.1 or [::1], which stops web pages that rebind their name to 127.0.0.1 (the gRPC listener does the same when gRPC-Web is on). Create one token per role and per consumer, scoped to the pools it needs:

   [[auth.tokens]]
   token_env = "CRONOQ_LLM_WORKER_TOKEN"
   role = "worker"
   pools = ["llm"]
   

2. Keep tokens out of files. Use token_env and inject secrets from your platform. Tokens need at least 16 characters and must be unique; use 32+ random characters (openssl rand -hex 16). The server keeps only SHA-256 digests and never logs tokens.

3. Encrypt traffic. Configure [server.tls] with cert and key; it covers both the gRPC and the HTTP listener. Bearer tokens over plaintext are acceptable only on a trusted local network or inside a service mesh that encrypts. Workers given TLS settings must use an https:// endpoint; they refuse an http:// one instead of connecting in plaintext.

4. Use mTLS where you can. With client_ca, every client must present a certificate signed by that CA. Map certificates to roles by DNS subject alternative name, so workers need no bearer token at all:

   [server.tls]
   cert = "/etc/cronoq/tls/server.pem"
   key = "/etc/cronoq/tls/server.key"
   client_ca = "/etc/cronoq/tls/ca.pem"
   
   [[auth.certs]]
   dns_name = "llm-worker.internal.example.com"
   role = "worker"
   pools = ["llm"]
   

   A bearer token, when present, takes precedence over the certificate's mapping.

5. Keep metrics internal. /metrics on http_addr requires the metrics role. If your scraper cannot send a token, set metrics_addr and bind it to an internal interface: that listener has no TLS and no authentication.

6. Bind deliberately. The defaults are 127.0.0.1:7070 and 127.0.0.1:7071. Listen on 0.0.0.0 only behind a firewall, security group or network policy.

7. Bound everything. Keep the [limits] defaults or lower them, and set max_pending and overflow on pools fed by less trusted producers, so one client cannot exhaust memory. max_slots_per_worker caps how many leases a single worker can hold.

8. Protect the database. Read the URL from url_env (default DATABASE_URL), use sslmode=verify-full for a remote database, and grant the server's role privileges on its own schema only. Put the tables in a dedicated schema (store.schema) that only the server's role can create objects in; if you keep public on a database created before PostgreSQL 15, run REVOKE CREATE ON SCHEMA public FROM PUBLIC. The server refuses existing cronoq tables owned by another role. PostgreSQL holds payloads and outputs in clear text: use encryption at rest or encrypt sensitive payloads in your application.

9. Run the provided image. It is distroless, runs as a non-root user and has no shell. Validate configuration changes in CI with cronoq-server --config cronoq.toml --check.

10. Treat workers as trusted with their pools. A worker sees every payload of the pools it serves and decides their outputs. Give each worker fleet credentials for its own pools only, and, where workers carry labels, a labels = [...] allow-list so a credential cannot claim labels of another fleet.

11. Give browser dashboards narrow tokens. Enable [server.grpc_web] only for the exact origins of your dashboards. A token held by a page can be read by any script on that page, so give it the least role it needs (often metrics), scoped to the pools it shows, and serve the page over HTTPS with TLS on the server.

Reporting vulnerabilities

See SECURITY.md. Please report privately through GitHub Security Advisories.

Performance tuning

What to expect

Choosing the next task looks only at the heads of a few sorted queues, so its cost grows at most with the logarithm of the backlog, with no rescans (see Architecture). Re-queuing a retried or woken task into a very long lane also moves the entries behind it.

Tuning the library

• Submit in bulk. enqueue_many(Vec<Task>) admits thousands of tasks in one channel round trip; individual submit calls each wait for a reply.
• Skip handles you do not need. enqueue is cheaper than submit (no result waiter).
• Size the command queue. Builder::queue_capacity(n) (default 8192) should cover bursts.
• Disable telemetry you do not export. Build with default-features = false to remove OpenTelemetry instrumentation.
• Keep executors async. A blocking executor holds a tokio worker thread; move CPU-heavy work to spawn_blocking or a dedicated pool.
• Prefer fewer, larger pools. Each pool adds queues and an autoscaling timer; resources can express most sharing needs.
• Retention. Terminal statuses are kept for an hour (up to 1 M entries) for dependencies and deduplication. Lower CoreConfig::retention / retention_max for very high-rate workloads that do not need them.

Tuning the server

• Give workers enough slots that per-lease round trips are amortized; heartbeats are cheap.
• With a durable store (SQLite or Postgres), commit latency bounds how fast the engine can acknowledge submissions in a single step. Batched submissions (SubmitBatch, /v1/tasks:batch) share one commit.
• Keep the database close (same region/zone) to the server.

Measuring

cargo bench -p cronoq-core --bench core      # engine
cargo bench -p cronoq --bench runtime        # end to end
cargo run --release -p cronoq --example firehose -- 1000000

Shared or virtualized machines are noisy; compare instruction counts (valgrind --tool=callgrind on crates/cronoq-core/examples/prof.rs) when optimizing.

Testing your executors

Rust: paused time

The runtime uses tokio's clock, so tests with paused time run schedules, timeouts and backoffs instantly and deterministically:

#[tokio::test(start_paused = true)]
async fn retries_then_succeeds() {
    let s = Scheduler::<u64, u64>::builder()
        .pool(Pool::fixed("default", 1).executor_fn(|ctx, x: Arc<u64>| async move {
            if ctx.attempt < 3 { Err(TaskError::retryable("flaky")) } else { Ok(*x) }
        }))
        .build()
        .await
        .unwrap();
    let h = s.submit(Task::new(7)).await.unwrap();
    assert_eq!(*h.await.unwrap(), 7);   // backoffs elapse in virtual time
}

Fixed starts and cron schedules work the same way: tokio::time::sleep(Duration::from_secs(60)) inside a paused test advances the scheduler's clock by a minute without waiting.

Rust: the engine directly

cronoq-core is a pure state machine. Drive it with explicit times and inspect the actions it emits; this is how cronoq's own deterministic simulation tests work:

use cronoq_core::*;

let mut core: Core<&str> = Core::new(CoreConfig::default());
core.submit(Time::ZERO, TaskSpec::new("a").priority(Priority::High)).unwrap();
let Some(Action::Dispatch(d)) = core.poll_action() else { panic!() };
core.complete(Time::from_secs_f64(0.5), d.lease, Outcome::Success);
core.check_invariants();

Python

Run executors through a real scheduler inside an asyncio test (pytest-asyncio):

import cronoq, pytest

@pytest.mark.asyncio
async def test_executor_retries():
    calls = 0
    async def run(ctx, x):
        nonlocal calls
        calls += 1
        if ctx.attempt < 2:
            raise cronoq.Retryable("flaky")
        return x * 2
    retry = cronoq.RetryPolicy(max_attempts=3, initial_backoff=0.01, jitter=False)
    async with cronoq.Scheduler([cronoq.Pool("default", run, retry=retry)]) as s:
        assert await (await s.submit(21)) == 42
    assert calls == 2

Against a server

Start cronoq-server with a temporary config on free ports (grpc_addr = "127.0.0.1:0" works when you embed it with cronoq_server::run), connect a test worker, and submit through the client. The repository's crates/cronoq-server/tests/server.rs and crates/cronoq-py/python/tests/test_client.py are complete examples.

run reports the addresses it bound once the server is the leader. When a test needs them earlier, for example to reach a standby, bind the listeners on port 0 yourself and pass them to cronoq_server::run_with_listeners. Choosing a free port, closing it, and putting its number in the config is not safe when tests run in parallel: another process can take the port before the server binds it. A server started as a separate process has to bind its ports itself, so retry the start on new ports if it exits with "Address already in use".

Server configuration

cronoq-server reads one TOML file, given with --config <path> (or the CRONOQ_CONFIG environment variable). Without a file it starts with defaults: one default pool, in-memory state, no auth tokens and listeners on 127.0.0.1 only. Unknown keys are rejected and every value is validated at startup, so typos and out-of-range numbers fail fast instead of at runtime.

cronoq-server --config cronoq.toml --check   # validate and exit
cronoq-server --config cronoq.toml           # run
cronoq-server --print-proto                  # print the gRPC API definition

The log level comes from RUST_LOG (default info).

An annotated example lives in deploy/cronoq.example.toml; it is the configuration baked into the container image.

Durations are seconds (floats). They must be finite and between their minimum and one year.

[server]

[server.tls]

Optional. Enables TLS on both the gRPC and the HTTP listener (the optional metrics listener stays plain HTTP). Files are read once at startup; a missing file is reported with its path.

[server.peer_tls]

Optional. How this server verifies other shards' https:// endpoints when it forwards calls (Sharding).

[server.grpc_web]

Optional. Serves gRPC-Web for browser clients on the gRPC listener, alongside native gRPC. Without this table the listener speaks only HTTP/2 gRPC.

The origin list only controls which pages a browser lets call the server; every call still needs credentials with the right role, as over native gRPC.

[auth]

Callers authenticate with authorization: Bearer <token> (gRPC metadata or HTTP header), or with a client certificate mapped to a role. Roles:

/healthz and /readyz never need credentials.

[[auth.tokens]]

[[auth.tokens]]
token_env = "CRONOQ_CLIENT_TOKEN"
role = "client"
pools = ["llm"]            # optional

Exactly one of token / token_env is required. Tokens must be at least 16 characters (the server warns below 32; use 32+ random characters, e.g. openssl rand -hex 24) and unique. Tokens are stored as SHA-256 digests and compared in constant time; Debug output and logs never contain them.

[[auth.certs]]

Map mTLS client certificates to roles by an exact DNS subject alternative name. Requires server.tls.client_ca.

[[auth.certs]]
dns_name = "worker-1.internal"
role = "worker"
pools = ["gpu"]            # optional

A bearer token, when present, takes precedence over the certificate.

[store]

[store]
kind = "postgres"
url_env = "DATABASE_URL"
keep_finished_s = 604800

The server refuses to start if the schema already holds cronoq tables owned by a role other than the one it connects as (or a superuser). Prefer a dedicated schema that only the server's role can create objects in (see Security hardening).

[limits]

Request and resource limits. The defaults are safe for a shared deployment; every key is optional.

Strings in task specs (keys, kinds, tags, resource names, worker messages) must not contain NUL characters; worker failure messages are sanitized (NUL replaced, cut at 8 KiB) instead of rejected.

[[pools]]

Server pools are served by remote workers: a pool's capacity is the number of slots its connected workers registered. Defaults to a single pool named default.

[[pools]]
name = "llm"
max = 200
lease_ttl_s = 120
overflow = "reject"
retry = { max_attempts = 5, initial_backoff_s = 2, max_backoff_s = 120 }

scaler

retry

[[resources]]

Shared limits that tasks request by name (resources = {"openai": 1}).

[[resources]]
kind = "token_bucket"
name = "openai"
rate = 50
burst = 100

[[resources]]
kind = "semaphore"
name = "gpu"
permits = 8

[[shards]]

Optional. Splits the pools between shards, each served by its own leader and standbys with a store of its own (Sharding). Every shard's servers use the same list and pick their own with server.shard.

With shards, each shard keeps its own state: a Postgres store.schema must contain {shard} (default cronoq_{shard}), and a SQLite store.path must contain {shard}; the server fills in its shard's name. Each resource belongs to one shard (its shard key), because shards do not share counts: a resource shared by two shards would have its full capacity in each. [limits] apply to each shard separately.

[outputs]

Recent outputs and task records, used by Wait/Get after a task finished. With a durable store (postgres or sqlite), outputs that are no longer cached are read from the database.

Validation summary

--check (and every startup) rejects:

• unknown keys, duplicate pool or resource names, invalid names;
• non-finite, negative or out-of-range numbers (durations, fractions, multipliers, rates, tenant weights), and a pool min above its max;
• [[auth.tokens]] entries without exactly one of token/token_env, duplicate literal token values, pool allow-lists naming unknown pools;
• [[auth.certs]] without server.tls.client_ca;
• server.grpc_web.allowed_origins that is empty, has more than 64 entries, or holds anything other than exact origins;
• both store.url and store.url_env;
• [[shards]] with duplicate names, numbers or endpoints, unknown pools, pools in no shard or in two, a missing or unknown server.shard, a store namespace without {shard}, or a resource without a known shard.

Startup also rejects the following, which --check does not test because they depend on the environment:

• a token_env variable that is not set;
• tokens shorter than 16 characters, and duplicate tokens (including tokens read from token_env);
• non-loopback listeners with no credentials, unless insecure_no_auth is set.

HTTP API

The HTTP gateway exposes the Tasks service of the gRPC API as JSON, except the Watch stream, plus the read-only Admin calls, metrics and health probes. It listens on server.http_addr (default 127.0.0.1:7071) and serves HTTP/1.1 and HTTP/2 (h2c without TLS, ALPN with TLS). With [server.tls] configured it serves HTTPS and accepts client certificates on the same terms as the gRPC listener.

Remote workers cannot use HTTP; they need the gRPC worker stream. Recurring schedules are managed over gRPC (Admin.AddRecurring) or in the embedding runtime.

Conventions

• Authentication. Send authorization: Bearer <token>, or connect with a client certificate mapped in [[auth.certs]]. The admin role passes every check. Without any [[auth.tokens]] or [[auth.certs]], authentication is off and every request is treated as admin; the server only starts that way on loopback addresses unless --insecure-no-auth is given. In that mode, unless --insecure-no-auth is given, a request whose Host is not a loopback name or address (localhost, *.localhost, any 127.x.x.x, [::1]) gets 403.
• Pool scope. A token or certificate with pools = [...] may only submit into those pools. Tasks in other pools look like they do not exist (404), so their ids and keys do not leak. Submitting a key that is already used by a task in a pool outside the scope returns 409.
• Tenant. Credentials with a tenant submit for that tenant: tasks without a tenant get it, and naming another tenant returns 403.
• Request bodies must be JSON with content-type: application/json (otherwise 415) and at most limits.max_http_body_bytes (default 16 MiB, otherwise 413). Unknown fields are rejected with 400, so a typo such as "priorty" never passes silently.
• Task ids are strings in every JSON response ("id": "17"), because they are 64-bit and JavaScript numbers lose precision above 2^53. Requests accept ids as strings or numbers.
• Timestamps (*_unix_ns) are Unix nanoseconds, returned as strings for the same reason. Durations (*_s) are seconds as JSON numbers and must be finite and non-negative.
• Every request has a time limit of limits.request_timeout_s (default 30 s). Every request to /v1/tasks (POST and GET) and to every path ending in /result gets limits.max_wait_s (default 300 s) on top. A request that runs out of time returns 504.
• Only the leader serves. A standby (or a leader that is shutting down) answers every /v1 route and /metrics with 503 not the leader (standby or shutting down). Put the servers behind a load balancer that routes on /readyz.

Routes

{id} must be a decimal task id; anything else returns 400 and points to the by-key route. Keys are percent-decoded path segments: encode / as %2F (/v1/tasks/by-key/report%2F42).

When server.metrics_addr is set, a second listener serves /metrics, /healthz and /readyz without authentication, for a Prometheus scraper or kubelet on a private network. The authenticated /metrics on http_addr stays available.

Task object

The body of POST /v1/tasks and each element of a batch's tasks. Every field is optional; {} submits an empty payload to the default pool.

Use either the relative or the absolute form of a time, not both. Strings must not contain NUL characters. Dependencies (after plus deps) are limited to max_deps.

POST /v1/tasks

curl -s localhost:7071/v1/tasks -H "authorization: Bearer $CLIENT_TOKEN" \
  -H 'content-type: application/json' \
  -d '{"pool": "default", "payload": {"prompt": "hi"}, "priority": "high", "key": "job-1"}'

{"id": "17", "duplicate": false}

duplicate is true when a task with the same key is already known; id is then the existing task's id and nothing new is enqueued. With a durable store (Postgres or SQLite), the response is sent after the task is stored.

With "wait": true the response also carries the result:

{"id": "17", "duplicate": false,
 "result": {"id": "17", "status": "succeeded", "output_available": true,
            "output": {"answer": 42}, "output_b64": "eyJhbnN3ZXIiOjQyfQ=="}}

If the task is still running when wait_timeout_s expires, the request fails with 504; the task keeps running and its result can be fetched with GET /v1/tasks/{id}/result.

POST /v1/tasks:batch

{"tasks": [
  {"key": "extract", "payload": {"doc": 1}},
  {"key": "summarise", "payload": {"doc": 1}, "after": ["extract"]}
]}

{"tasks": [{"id": "18", "duplicate": false}, {"id": "19", "duplicate": false}]}

The batch is admitted atomically: either every task is accepted, or none is and the error names the offending task. Dependencies may reference keys defined anywhere in the same batch, in any order. Results are in request order. At most limits.max_tasks_per_batch tasks (default 1000).

GET /v1/tasks/{id} and GET /v1/tasks/by-key/{key}

{"id": "17", "status": "running", "pool": "default", "key": "job-1", "kind": "default",
 "attempt": 1, "max_attempts": 3, "output_available": false,
 "created_unix_ns": "1760000000000000000", "started_unix_ns": "1760000000120000000"}

id and status are authoritative; the other fields are best effort and come from the serving leader's bounded info cache (outputs.info_entries) or, with a durable store, from the stored task. Unknown tasks, tasks past retention, and tasks in pools the caller may not use return 404.

GET /v1/tasks

Lists tasks in id order, oldest first, as task objects like those of GET /v1/tasks/{id}:

GET /v1/tasks?pool=default,reports&status=running,failed&limit=50

{"tasks": [{"id": "17", "status": "running", "pool": "default", "attempt": 1, ...}],
 "next_after": "17"}

next_after is present when more tasks follow. With shards, the list covers every shard, and unavailable_shards names shards that could not be reached. It is present only when a shard was missing, on this route and on /v1/workers, /v1/recurring, the task-detail routes and the result routes; /v1/snapshot always includes it (empty when every shard answered). Live tasks come from the scheduler. Finished tasks come from the store, or, without one, from the bounded info cache, so only the most recent ones are listed. A page is not an atomic snapshot: a task that finishes while the page is built can be missing from it.

GET /v1/workers

{"workers": [{"id": "3", "name": "gpu-1", "pool": "default", "labels": {"gpu": "a100"},
              "slots": 4, "busy": 1, "connected_unix_ns": "1760000000000000000"}]}

Connected remote workers of the pools the caller may use, in connection order. id is assigned by the server per connection; name is what the worker registered with.

GET /v1/recurring

{"schedules": [{"name": "nightly", "cron": "0 0 3 * * *", "tz": "Europe/Berlin",
                "pool": "default", "kind": "report", "overlap": "queue",
                "next_unix_ns": "1760061600000000000", "last_task_id": "41"}]}

Recurring schedules of the pools the caller may use, by name. tz is absent for UTC, next_unix_ns when the schedule has no further occurrence, and last_task_id before the first occurrence.

DELETE /v1/tasks/{id} and DELETE /v1/tasks/by-key/{key}

{"cancelled": true}

false when the task already finished. Cancelling a running task aborts the worker's lease and cancels its handler. Dependents on an on_success or on_failure edge end as skipped; on_complete dependents proceed. With a durable store (Postgres or SQLite), the cancellation is stored before the response is sent.

GET /v1/tasks/{id}/result and GET /v1/tasks/by-key/{key}/result

Waits until the task finishes, at most timeout_s seconds (query parameter, default and maximum limits.max_wait_s). ?timeout_s=0 returns immediately.

curl -s "localhost:7071/v1/tasks/17/result?timeout_s=60" -H "authorization: Bearer $CLIENT_TOKEN"

Returns a result object, or 504 task is still running; wait again or poll. Results of finished tasks remain available while the leader's output cache (outputs.cache_entries, outputs.cache_bytes) or the durable store (Postgres or SQLite, store.keep_finished_s) still holds them.

Result object

{"id": "17", "status": "failed", "output_available": false, "message": "asked to fail"}

GET /v1/snapshot

{"live_tasks": 12, "running": 4, "pending": 8, "workers_connected": 2, "unavailable_shards": [],
 "pools": [{"name": "default", "capacity": 4, "busy": 4, "held": 0, "desired": 6,
            "pending": 8, "pending_by_class": [0, 2, 6, 0], "reservations": 0,
            "dispatched": 120, "succeeded": 118, "failed": 1,
            "priority_adherence": 1.0, "fixed_start_hit_rate": 1.0,
            "utilization": 0.83, "schedule_efficiency": 0.80,
            "wait_p50_s": 0.2, "wait_p90_s": 1.4, "wait_p99_s": 2.9, "cost": 0.0,
            "workers_connected": 2,
            "groups": [{"labels": {}, "capacity": 3, "busy": 3},
                       {"labels": {"gpu": "a100"}, "capacity": 1, "busy": 1}]}],
 "totals": {"submitted": 130, "dispatched": 124, "succeeded": 118, "failed": 1,
            "dead_lettered": 0, "cancelled": 2, "skipped": 0, "missed": 0, "shed": 0,
            "retries": 4, "timeouts": 1, "lease_expiries": 0,
            "stale_completions": 0}}

Field meanings: Metrics and quality signals.

Statuses

Errors

Errors have the body {"error": "<message>", "code": "<reason>"}, where code is the lower-case HTTP reason phrase (for example too_many_requests). 401 responses carry www-authenticate: Bearer.

See also Errors for retry guidance.

gRPC API

Package cronoq.v1, defined in crates/cronoq-proto/proto/cronoq/v1/cronoq.proto. Print it from a running binary with cronoq-server --print-proto, depend on the cronoq-proto crate in Rust, or use the generated stubs shipped in the Python package (cronoq._proto). The server listens on server.grpc_addr (default 127.0.0.1:7070), with TLS when [server.tls] is configured.

This page describes worker protocol version 3.

Conventions

• Authentication. Send the metadata entry authorization: Bearer <token>, or present a client certificate mapped in [[auth.certs]]. Each service requires a role (below); admin passes every check. A token or certificate scoped to pools = [...] may only submit into, watch and serve those pools; tasks in other pools are reported as NOT_FOUND.
• Times are relative seconds (*_in_s, *_s, double) or absolute Unix nanoseconds (*_unix_ns, int64). Durations must be finite and non-negative. Set at most one of the relative and absolute forms.
• Enum zero values (*_UNSPECIFIED) mean "default": PRIORITY_NORMAL, EDGE_COND_ON_SUCCESS, MISS_POLICY_RUN_LATE, OVERLAP_SKIP.
• Limits. Messages are limited to limits.max_message_bytes (default 16 MiB), strings and collections to the [limits] table in the configuration reference. Strings must not contain NUL characters.
• Only the leader serves. A standby or a stopping leader answers every RPC with UNAVAILABLE; retry with backoff. A standby also closes each connection about a second after accepting it, so behind a load balancer that sends connections to every replica, a retry on a new connection can reach the leader. A load balancer that routes on /readyz sends connections to the leader only.
• Deadlines. Every unary call is bounded by limits.max_wait_s + limits.request_timeout_s; a call that runs longer fails with CANCELLED. Submit with wait and Wait also stop waiting after limits.max_wait_s (or the shorter timeout the call asks for).

Tasks (role: client)

TaskSpec

The fields match the HTTP task object, which documents them in detail. TaskRef is a oneof of id (uint64) and key (string).

Submit and batches

SubmitRequest{task, wait, wait_timeout_s} returns SubmitResponse{id, duplicate, result}. duplicate is true when the key already exists (then id is the existing task). With wait, result is set once the task finished; if wait_timeout_s (capped at limits.max_wait_s) expires first the call fails with DEADLINE_EXCEEDED and the task keeps running.

SubmitBatchRequest{tasks} is admitted atomically: all tasks or none. Dependencies may name keys defined anywhere in the batch. The response lists one SubmitResponse per task, in request order, without result.

With a durable store (Postgres or SQLite), Submit, SubmitBatch and Cancel return after the change is stored.

TaskInfo

Fields other than id and status are best effort: they come from the serving leader's info cache (outputs.info_entries) or, with a durable store, from the stored task.

ListTasks

ListTasksRequest has pools (empty: every pool the caller may use), statuses (empty: any; TASK_STATUS_UNSPECIFIED is INVALID_ARGUMENT), after (only larger ids; 0 starts from the beginning) and limit (0 means 100; capped at limits.max_list). The response's next_after is the value to pass as after for the next page, and 0 on the last page. Pools are checked as for Watch: unknown ones are NOT_FOUND, pools the caller may not use PERMISSION_DENIED.

Live tasks come from the scheduler. Finished tasks come from the store, or, without one, from the bounded info cache, which holds only the most recent ones. A page is not an atomic snapshot: a task that finishes while the page is built can be missing from it.

TaskResult

{id, status, output, message, output_available}. status is always terminal. output is set when the task succeeded and its output is still retained (output_available); a succeeded task whose output was evicted from the cache and the store has output_available = false. message explains failures.

Wait without timeout_s waits up to limits.max_wait_s; on timeout it fails with DEADLINE_EXCEEDED (task is still running; wait again or poll). It fails with UNAVAILABLE if the server stops serving while waiting: wait again on the new leader.

Watch

WatchRequest fields:

Each StatusEvent carries {id, pool, key, status, attempt, at_unix_ns, message}. The stream ends with UNAVAILABLE when the server shuts down or loses leadership, and with DATA_LOSS when the client reads too slowly and events were dropped. Events are live only: reconnect and use Get / Wait to reconcile anything that happened while disconnected.

Workers (role: worker)

Messages

Protocol rules

1. Register first. A stream whose first message is not Register fails with INVALID_ARGUMENT; one that sends nothing for 10 s fails with DEADLINE_EXCEEDED. A second Register on the same stream ends it. Unknown pools are NOT_FOUND.
2. Capacity is slots. A pool's capacity is the sum of the granted slots of its connected workers. The server never sends more concurrent leases to a worker than its slots. Use the slots value from Registered, not the one you requested.
3. Heartbeat each running lease more often than lease_ttl_s (every lease_ttl_s / 3 is a good rule). A lease without a heartbeat for lease_ttl_s expires: the attempt fails as retryable and the worker receives Abort.
4. Report exactly one outcome per lease: Complete, Fail, Nack or Cancelled. Reports for leases the worker does not hold (already aborted, or unknown) are ignored.
5. Nack offers the lease to another worker of the pool, without counting an attempt. If no other worker has a free slot, the attempt fails as retryable.
6. Abort means the server has already given up the attempt and freed the slot. Stop the handler; sending Cancelled afterwards is allowed and ignored.
7. Read the stream promptly. Each worker has a bounded outbound queue (about one lease and one abort per slot). A worker that stops reading is disconnected rather than allowed to hold leases it cannot see.
8. Disconnects fail running leases as retryable (worker disconnected); they are retried according to the task's retry policy.
9. Shutdown. On a graceful stop the server first gives running attempts server.shutdown_grace_s to finish (workers keep reporting outcomes), then sends Shutdown{not_leader: false}. When it loses leadership it sends Shutdown{not_leader: true} immediately. Either way, leases still running are failed as retryable and the stream ends about a second later: stop running work and reconnect with backoff.
10. Messages in both directions are limited to max_message_bytes from Registered.

The Rust (cronoq_server::worker::run_worker) and Python (cronoq.client.Worker) workers implement these rules; see Writing executors and workers.

Admin

ListWorkers and ListRecurring list only the pools the caller may use. A WorkerInfo has the connection id the server assigned, the name, pool and labels the worker registered with, its slots and busy slots, and connected_unix_ns. A RecurringInfo has the schedule's name, cron, tz (empty for UTC), pool, kind and overlap, next_unix_ns (unset when no further occurrence is due) and last_task_id (unset before the first occurrence); schedules are sorted by name.

SnapshotResponse carries pools (one PoolSnapshot per pool: name, capacity, busy, held, desired, pending_by_class, reservations, dispatched, succeeded, failed, priority_adherence, fixed_start_hit_rate, utilization, schedule_efficiency, wait_p50_s, wait_p90_s, wait_p99_s, cost, workers_connected, and groups: one WorkerGroup{labels, capacity, busy} per set of worker labels, unlabeled first), live_tasks, running, pending, workers_connected, unavailable_shards and totals (submitted, dispatched, succeeded, failed, dead_lettered, cancelled, skipped, missed, shed, retries, timeouts, lease_expiries, stale_completions). The HTTP snapshot documents each field.

AddRecurring registers (or replaces) a cron schedule named name. cron has six fields (seconds first) or five (seconds = 0); tz is an IANA time zone (default UTC). The template is a TaskSpec without deadline_unix_ns, start_* or deps (INVALID_ARGUMENT otherwise). Each occurrence gets its own fixed start at the cron time, with the template's tolerance_s (default 1 s) and on_miss (default RUN_LATE). The template's deadline_in_s is relative to each occurrence's start, and its key is a prefix: each occurrence's key is <key>@<start in UTC, RFC 3339>. overlap is OVERLAP_SKIP (the default: an occurrence whose previous run is still going ends as SKIPPED), OVERLAP_ALLOW, OVERLAP_QUEUE (wait for the previous run) or OVERLAP_QUEUE_ON_SUCCESS (wait, and run only if it succeeded). Schedules are durable with a durable store; AddRecurring returns once the schedule is stored. See Recurring schedules.

Shards

With [[shards]], any server answers for every shard (Sharding): calls for another shard are forwarded to it with the caller's token, and ListTasks, Watch, ListWorkers, ListRecurring and Snapshot cover all shards. The listing responses and SnapshotResponse name shards that could not be reached in unavailable_shards; Get and Wait by key name them in the cronoq-unavailable-shards response metadata, and Cancel by key is refused while any shard cannot be asked. A call that is not forwarded (client-certificate callers, plain-HTTP endpoints on other hosts, workers) gets FAILED_PRECONDITION with the shard's name and endpoint in the cronoq-shard and cronoq-shard-endpoint metadata. Workers reconnect to that endpoint (see Sharding).

gRPC-Web

With [server.grpc_web], the gRPC listener also answers gRPC-Web requests (application/grpc-web and application/grpc-web+proto, binary framing) for browser clients such as the TypeScript SDK's @synaptiai/cronoq/web, or those generated by protoc-gen-grpc-web or Connect-ES's gRPC-Web transport, over HTTP/1.1 or HTTP/2.

• Calls. Unary calls and the server stream Watch work. Workers.Work is a bidirectional stream, which gRPC-Web cannot carry; workers use native gRPC.
• Authentication is the same as for native gRPC: send authorization: Bearer <token>. Browsers do not attach credentials themselves, and the server does not allow credentialed CORS requests, so cookies and browser-held client certificates are not used.
• CORS. Preflights from the configured allowed_origins are answered for POST with the headers authorization, content-type, x-grpc-web, x-user-agent and grpc-timeout; the response headers grpc-status, grpc-message and grpc-status-details-bin are exposed. Other origins get no CORS headers.
• HTTP/1.1 carries only gRPC-Web: a plain gRPC request over HTTP/1.1 gets UNIMPLEMENTED.
• Without credentials (loopback-only mode), every request on the gRPC listener must name a loopback host (localhost, *.localhost, 127.0.0.1, [::1]), or it gets PERMISSION_DENIED, as on the HTTP API. With --insecure-no-auth this rule is off, as it is for the HTTP API, so a page on a machine that can reach the server could call every service, Admin included: do not enable gRPC-Web without credentials outside a trusted network.

Status codes

Enumerations

Status meanings: HTTP statuses.

Protocol version

Workers send protocol_version 3; a server refuses older versions with FAILED_PRECONDITION and a message naming the version it needs.

Python API

pip install cronoq            # embedded scheduler (module: cronoq)
pip install "cronoq[client]"  # + cronoq.client: gRPC client and remote worker for cronoq-server

Wheels use the stable ABI (abi3, CPython 3.14+) for Linux, macOS and Windows, on x86_64 and arm64. The client extra needs grpcio, which has no prebuilt wheel for Windows on arm64, so there pip tries to build it from source. Type stubs ship with the package (py.typed). Durations are seconds (float); absolute times accept a datetime or Unix seconds. Invalid durations or configuration (negative, NaN, out of range) raise ValueError.

cronoq.Scheduler

Scheduler(pools: Sequence[Pool], resources: Sequence[Resource] | None = None,
          emit_transitions: bool = False, *, shutdown_grace: float = 5.0,
          event_capacity: int = 1024, store: SqliteStore | None = None,
          codec: Codec | None = None)

A scheduler is bound to the event loop that starts it. If that loop closes without a shutdown, the scheduler stops and further calls raise SchedulerError; calls before start() raise RuntimeError.

Task options

Accepted by submit, enqueue and Task(...).

Strings must not contain NUL characters.

Status events

sched.events() returns an EventStream, an async iterator of StatusEvent (id, pool, key, status, attempt, at in Unix seconds, message).

async with sched.events() as events:
    async for e in events:
        print(e.id, e.status, e.message)

• Each stream buffers up to event_capacity events. When the consumer falls behind, the oldest are dropped and counted in events.lagged. With on_lag="raise" the next iteration raises EventsLagged (.skipped); iterating again continues with the buffered events.
• The stream ends (StopAsyncIteration) when the scheduler shuts down or the stream is closed (close(), await aclose(), or leaving async with). closed tells whether it ended.

cronoq.SqliteStore

SqliteStore(path, *, lease_ttl=10.0, lease_renew=2.0, lock_poll=0.5, retention=3600.0,
            full_sync=True)

Durable state in a local SQLite file, created with owner-only permissions. Submissions, cancellations and schedule changes return only once they are stored. After a restart, queued and interrupted tasks run again (an attempt that was running counts as used), recurring schedules resume, and results of finished tasks stay available through handle().

With a store, payloads and results are encoded by the scheduler's codec:

• cronoq.JsonCodec (the default) stores JSON. Payloads it cannot encode (bytes, sets, arbitrary objects) are refused at submit with TypeError, before anything is stored. Decoding JSON never runs code, so a tampered file cannot execute anything in your process.
• Pass your own codec for other types, for example msgpack. A pickle codec works, but anyone who can write the file can then run code in your process.
• A result the codec cannot encode is not stored (a warning is logged on the cronoq logger); the task still succeeds.

cronoq.Pool

Pool(name, executor, *, min=1, max=10, initial=None, autoscale=None, lease_ttl=None,
     max_pending=None, overflow="wait", retry=None, unknown_dependency="reject",
     await_timeout=None, default_duration=1.0, reservable_fraction=0.5,
     tenant_weights=None, fair_share=False)

A sync executor cannot be interrupted: on timeout or cancellation the attempt is aborted, but the function keeps running until it returns. Check ctx.is_cancelled() in long-running sync executors.

Other types

cronoq.__version__ is the package version as a string.

Exceptions

All live in cronoq (and cronoq.errors), derive from CronoqError and can be pickled.

Any other exception from an executor is a retryable failure. See Errors and status codes.

cronoq.client

The gRPC client and remote worker for cronoq-server, speaking worker protocol version 3 (cronoq.client.PROTOCOL_VERSION). Both can be created outside a running loop; they connect on first use.

Client(target, *, token=None, tls=False, credentials=None, options=None)

target is host:port. tls=True uses system roots; pass grpc.ssl_channel_credentials(...) as credentials for a private CA or mTLS. Use it as async with Client(...) as c or call await c.close().

Result types (frozen dataclasses):