Taskvisor

In-process task supervision for Tokio - from long-running workers to one-shot jobs.

Write ordinary async code. Taskvisor adds restart and backoff, cooperative shutdown, dynamic task management, typed lifecycle events, reliable final outcomes, and optional per-slot admission control.

Quick start - Examples - Production limits

The loop you stop writing

Long-running services often grow a loop like this:

tokio::spawn(async move {
    loop {
        match run_worker().await {
            Ok(()) => break,
            Err(error) => {
                eprintln!("worker failed: {error}; retrying in 1s");
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
});

It has a fixed retry delay, no jitter, no graceful shutdown, and no typed lifecycle observability.

Once the task is defined, its supervision policy becomes one declaration:

supervisor
    .run(vec![TaskSpec::restartable(worker)])
    .await?;

Taskvisor owns the lifecycle machinery. Your task keeps the application logic.

Quick start

[dependencies]
taskvisor = "0.6"
tokio = { version = "1", features = ["full"] }

Put this in src/main.rs, then run cargo run. The task fails twice, Taskvisor retries it with backoff, and the third attempt succeeds.

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use taskvisor::prelude::*;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let attempts = Arc::new(AtomicU32::new(0));
    let flaky = TaskFn::arc("flaky", move |_ctx| {
        let attempts = Arc::clone(&attempts);
        async move {
            let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1;
            println!("attempt {attempt}");

            if attempt < 3 {
                Err(TaskError::fail(format!("temporary failure #{attempt}")))
            } else {
                Ok(())
            }
        }
    });

    let spec = TaskSpec::restartable(flaky)
        .with_backoff(BackoffPolicy::constant(Duration::from_millis(50)));

    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
    supervisor.run(vec![spec]).await?;
    Ok(())
}

attempt 1
attempt 2
attempt 3

Every attempt gets a fresh future. A retryable failure follows the configured backoff; a successful restartable task stops. Supervisor::run returns only after lifecycle cleanup finishes.

For a resident worker that runs until Ctrl+C, see worker.rs. For a reconnecting queue consumer, see queue_consumer.rs.

Why Taskvisor?

Restart and backoff are the baseline. Taskvisor also provides:

• Cooperative shutdown with a deadline. Tasks observe TaskContext; tasks that miss the grace period are force-aborted.
• Reliable final outcomes. TaskWaiter reports how watched work ended even when best-effort events are dropped.
• Typed lifecycle events. Logs, metrics, traces, and live status consume one structured event model.
• Dynamic management. Add, list, cancel, remove, and watch tasks through SupervisorHandle.
• Admission control. The optional controller applies Queue, Replace, or DropIfRunning per named slot.
• Explicit limits. Configure per-attempt timeout, retry budget, global concurrency, and bounded queues.

JoinSet and TaskTracker help own and join spawned futures. Taskvisor owns the restart policy and the task's complete lifecycle contract.

It is a good fit for queue consumers, pollers, sync loops, connection keepers, periodic work, and one-shot in-process jobs that need admission or a reliable outcome.

When the primary requirement is different, use a more specialized tool:

Core model

Four types form the main API:

Retries for one task run in sequence. Two attempts for the same TaskId never run at the same time. Active task names must be unique; the name can be reused after terminal cleanup, with a new TaskId.

There are two runtime modes:

Supervisor::run(...).await == Ok(()) means the supervisor lifecycle and cleanup completed successfully. It does not mean that every task succeeded. Use a watched outcome when application logic needs that answer.

Choose task behavior

The named constructors cover the common cases:

Fatal errors and cancellation always stop the task. A periodic interval begins after a successful attempt finishes; it is not a wall-clock or cron schedule.

A retry limit counts retries after the first failed attempt. max_retries = 3 therefore allows at most four attempts when every attempt fails.

use std::num::NonZeroU32;
use std::time::Duration;
use taskvisor::{BackoffPolicy, JitterPolicy, TaskRef, TaskSpec};

fn supervised(task: TaskRef) -> TaskSpec {
    TaskSpec::restartable(task)
        .with_backoff(
            BackoffPolicy::exponential(Duration::from_millis(200))
                .with_max(Duration::from_secs(30))
                .with_jitter(JitterPolicy::Equal),
        )
        .with_timeout(Duration::from_secs(10))
        .with_max_retries(NonZeroU32::new(3).unwrap())
}

Equal jitter chooses a real delay between half of the current base delay and the full base delay, spreading simultaneous retries over time. Per-task settings override values inherited from TaskDefaults.

Cancellation and shutdown

Cancellation is cooperative first. A long-running task must observe TaskContext:

use taskvisor::{TaskContext, TaskError};

async fn do_work() -> Result<(), TaskError> {
    // Application work goes here.
    Ok(())
}

async fn run_one_operation(ctx: &TaskContext) -> Result<(), TaskError> {
    ctx.run_until_cancelled(do_work()).await?
}

async fn run_with_more_branches(ctx: &TaskContext) -> Result<(), TaskError> {
    tokio::select! {
        _ = ctx.cancelled() => Err(TaskError::Canceled),
        result = do_work() => result,
    }
}

The joined shutdown path:

1. Closes admission for new work.
2. Sends cancellation to active tasks.
3. Waits for the configured grace period.
4. Force-aborts tasks that did not stop.
5. Drains subscriber queues for their separate shutdown timeout.

Call handle.shutdown().await to wait for cleanup and receive its result. Dropping the last public owner starts non-blocking cancellation but cannot report cleanup errors.

Dynamic management uses TaskId:

let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
let handle = supervisor.serve();

let id = handle.add(TaskSpec::restartable(worker)).await?;
let registered = handle.list().await;
let stopped = handle.cancel(id).await?;

println!("registered={}, stopped={stopped}", registered.len());
handle.shutdown().await?;

add().await? means the registry accepted the task, not that the task completed. Regular management methods wait for bounded queue capacity; their try_* forms fail fast when a queue is full. See dynamic.rs for a complete program.

Events and outcomes

Taskvisor has two result paths with different contracts:

Use add_and_watch when the final result matters:

use taskvisor::{RuntimeError, SupervisorHandle, TaskOutcome, TaskRef, TaskSpec};

async fn wait_for_task(
    handle: &SupervisorHandle,
    job: TaskRef,
) -> Result<(), RuntimeError> {
    let (id, waiter) = handle
        .add_and_watch(TaskSpec::once(job))
        .await?;

    match waiter.wait().await? {
        TaskOutcome::Completed => println!("{id} completed"),
        TaskOutcome::Failed { reason, .. } => eprintln!("{id} failed: {reason}"),
        TaskOutcome::Canceled => eprintln!("{id} was canceled"),
        other => eprintln!("{id} ended with {other:?}"),
    }
    Ok(())
}

Events carry a process-local sequence number and, where relevant, task identity, attempt, duration, reason, timeout, delay, and exit code. Stable string labels are available for telemetry.

Each subscriber has its own bounded FIFO queue. Its synchronous callback runs on Tokio's blocking pool. A slow subscriber cannot block publishers, but its queue may fill and lose events. Keep callbacks short and forward async work to another channel.

See subscriber.rs, the TracingBridge in tracing.rs, and the Prometheus counters in metrics.rs.

Admission control (feature: controller)

Enable the controller on the dependency:

taskvisor = { version = "0.6", features = ["controller"] }

The controller groups submissions into named slots. At most one task can occupy a slot; different slots can run concurrently.

The slot defaults to the task name. Use with_slot to place differently named tasks in the same lane:

use taskvisor::prelude::*;

async fn submit_to_slot(
    handle: &SupervisorHandle,
    job: TaskRef,
) -> Result<TaskOutcome, Box<dyn std::error::Error>> {
    let request = ControllerSpec::queue(TaskSpec::once(job))
        .with_slot("customer-42");
    let (_id, waiter) = handle.submit_and_watch(request).await?;
    Ok(waiter.wait().await?)
}

submit().await? confirms controller intake; admission happens later. submit_and_watch returns a final outcome. Work that never starts resolves to TaskOutcome::Rejected.

Queue depth is bounded per slot. Replace changes only the queue head; FIFO items behind it remain queued. controller_snapshot() returns a best-effort, non-transactional view of slot status and queue depth.

See slots.rs and admission.rs for complete programs.

Configuration

Runtime limits and task defaults are separate:

use std::num::{NonZeroU32, NonZeroUsize};
use std::sync::Arc;
use std::time::Duration;
use taskvisor::{Supervisor, SupervisorConfig, TaskDefaults};

fn configured_supervisor() -> Arc<Supervisor> {
    let runtime = SupervisorConfig::default()
        .with_grace(Duration::from_secs(30))
        .with_subscriber_shutdown_timeout(Duration::from_secs(5))
        .with_max_concurrent(NonZeroUsize::new(16));

    let tasks = TaskDefaults::default()
        .with_timeout(Duration::from_secs(20))
        .with_max_retries(NonZeroU32::new(5).unwrap());

    Supervisor::builder(runtime)
        .with_task_defaults(tasks)
        .build()
}

Main defaults:

Capacity types are non-zero where zero would make the runtime unusable. Checked try_with_* setters accept raw values.

Production limits

Taskvisor defines an in-process lifecycle. Keep these boundaries explicit:

• Events are best-effort. Do not use them as a durable audit log.
• Watched outcomes are not durable after the process exits.
• Cancellation depends on the task reaching an await point that observes TaskContext. Force-abort cannot stop synchronous code that blocks a runtime thread.
• Subscriber callbacks may still run on Tokio's blocking pool after their drain deadline. Tokio runtime shutdown may wait for such callbacks.
• Periodic tasks use an interval after completion. They do not provide calendar scheduling or missed-run recovery.
• The controller coordinates tasks inside one supervisor.
• With panic = "unwind", Taskvisor catches task-future panics. It cannot recover from panic = "abort", process aborts, memory exhaustion, or failures outside the process.

For a service deployment, call the joined shutdown path, make resident tasks cancellation-aware, set finite timeouts and retry limits where endless retry is unsafe, monitor lifecycle failures and overflow, and use watched outcomes for decisions that depend on completion.

The crate forbids unsafe Rust with #![forbid(unsafe_code)].

Feature flags

Taskvisor has no default features. The core depends on tokio, tokio-util, thiserror, and fastrand.

taskvisor = { version = "0.6", features = ["controller", "tracing"] }

Examples

From a cloned repository checkout, run the smallest example with:

cargo run --example basic

Benchmarks

The repository includes Criterion suites for lifecycle, throughput, subscriber fan-out, dynamic management, and controller paths:

cargo bench
cargo bench --bench controller --features controller

Contributing

Issues and pull requests are welcome. Read the contributing guide before a large change.

If Taskvisor earns a place in your stack, a GitHub star helps other Rust developers find it.