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.