Coordinate work by key

Use a controller when work that shares a key must take turns. Each controller slot has at most one owner, including work still waiting for admission or physical completion.

Enable the controller

This section requires the controller feature. It is enabled by default, but each supervisor must install a controller explicitly:

use taskvisor::{ControllerConfig, Supervisor, SupervisorConfig};

let _supervisor = Supervisor::builder(SupervisorConfig::default())
    .with_controller(ControllerConfig::default())
    .build();

Separate IDs, names, and slots

Direct add operations bypass controller admission. submit accepts a ControllerSpec. The controller applies its slot policy before asking the runtime registry to admit the task.

Identity	Scope
TaskId	One process-local registration or controller submission.
Task name	Registry key inside one supervisor.
Controller slot	Admission key inside one supervisor controller.

Different task names can share a slot. Without with_slot, the task name is also the slot. A queued submission has an ID, but does not own a registered task name yet.

A busy slot may be waiting for admission, running work, or waiting for physical completion. Busy does not always mean that a task body is currently polling.

Choose a busy-slot policy

Policy	Busy-slot behavior
Queue	Append to the bounded FIFO queue. A later Replace can still displace the queue head.
Replace	Request the owner to stop; create or replace the queue head.
DropIfRunning	Reject the incoming submission without changing the owner or queue.

A replacement is not guaranteed to become the next owner. A newer Replace can displace it before admission. The registry can also reject it later. Replace changes only the queue head and preserves the FIFO tail. It does not use the per-slot max_slot_queue limit, but creating a new head can still reach max_total_pending.

use taskvisor::{ControllerSpec, TaskFn, TaskRef, TaskSpec};

let task: TaskRef = TaskFn::arc(|_ctx| async { Ok(()) });
let request = ControllerSpec::queue(TaskSpec::once("customer-42-job", task))
    .with_slot("customer-42");

assert_eq!(request.task_spec().name(), "customer-42-job");
assert_eq!(request.slot_name(), "customer-42");

Watch admission and completion

submit creates an operation whose execute() confirms command intake only. Its watch() modifier makes execute() return a waiter. Read the submission identity through waiter.id(). The waiter resolves to Rejected if admission fails or to the registered task's final outcome if admission succeeds.

The call first waits for cleanup ownership and controller-command capacity. After the command is queued, returning to the caller and processing the command can race. The return from submit(request).watch().execute() is not an admission reply or proof that the task has started.

This diagram shows successful admission. Slot or registry rejection resolves the waiter to Rejected instead.

sequenceDiagram
accTitle: Controller intake, admission, and slot release
accDescr: Caller return and command processing can race, and slot release waits for logical and physical completion.
participant Caller
participant Controller
participant Registry
participant Waiter as TaskWaiter
Caller->>Caller: Wait for ownership and command capacity
Caller-)Controller: Queue submission
par Return from the call
    Caller-->>Caller: Return waiter (intake only)
and Process the command
    Controller->>Controller: Apply slot policy and wait if queued
    Controller->>Controller: Wait for registry-command capacity
    Controller-)Registry: Request admission
    Registry->>Registry: Admit task and run attempts
    alt Actor finishes without force-abort
        Registry->>Registry: Join actor
        Registry-)Waiter: Final outcome
    else Force-abort with physical exit still pending
        Registry-)Waiter: ForceAborted
        Registry->>Registry: Wait for physical exit
    end
    Registry-)Controller: Physical completion signal
    Controller->>Controller: Release slot and consider queued work
end

Each attempt waits for a concurrency permit if a limit is configured. The slot becomes reusable only after the registered task physically completes and terminal reporting is done. Deferred destruction of retained user values can continue after slot release. See task management for waiting and fail-fast calls.

This example starts an owner, rejects a competing DropIfRunning submission, and then stops the owner:

use std::sync::Arc;
use tokio::sync::Notify;
use taskvisor::prelude::*;

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let supervisor = Supervisor::builder(SupervisorConfig::default())
        .with_controller(ControllerConfig::default())
        .try_build()?;
    let handle = supervisor.serve()?;

    let owner_started = Arc::new(Notify::new());
    let owner: TaskRef = {
        let owner_started = Arc::clone(&owner_started);
        TaskFn::arc(move |ctx| {
            let owner_started = Arc::clone(&owner_started);
            async move {
                owner_started.notify_one();
                ctx.cancelled().await;
                Err(TaskError::Canceled)
            }
        })
    };

    let owner_request = ControllerSpec::queue(TaskSpec::once("tenant-42/owner", owner))
        .with_slot("tenant-42");
    let owner_waiter = handle.submit(owner_request).watch().execute().await?;
    let owner_id = owner_waiter.id();
    owner_started.notified().await;

    let contender: TaskRef = TaskFn::arc(|_ctx| async { Ok(()) });
    let contender_request = ControllerSpec::drop_if_running(TaskSpec::once(
        "tenant-42/contender",
        contender,
    ))
    .with_slot("tenant-42");
    let contender_waiter = handle
        .submit(contender_request)
        .watch()
        .execute()
        .await?;

    assert!(matches!(
        contender_waiter.wait().await?,
        TaskOutcome::Rejected {
            kind: RejectionKind::SlotBusy,
            ..
        }
    ));

    assert!(handle.cancel(owner_id).execute().await?);
    assert!(matches!(
        owner_waiter.wait().await?,
        TaskOutcome::Canceled
    ));

    handle.shutdown().await?;
    Ok(())
}

The notification proves that the first task has started before the contender is submitted. That owner waits for cancellation and keeps the slot occupied. The two waiters deliver the contender's rejection and the owner's final outcome directly.

prepare_submission allocates a task ID before intake. It does not reserve a name, slot, queue position, or runtime capacity.

Read diagnostic state

controller_snapshot reads each slot separately. Its ControllerSnapshot can already be stale when returned. Do not treat it as a transaction boundary.

Know timeout and cancellation scope

Attempt timeout starts only after registry admission and after Task::spawn returns the attempt future. It does not limit time spent in a controller queue. Controller submission has no built-in end-to-end deadline.

The ownership_timeout(duration) submission modifier limits only the wait for cleanup ownership before command intake. The timer stops when Taskvisor acquires the ownership permit. It does not cover controller-command capacity, a busy-slot queue, slot admission, registry-command capacity, task execution, or final outcome delivery. The operation created by PreparedSubmission::submit() uses the same boundary. A timeout consumes the prepared value, but produces no command or lifecycle event for its reserved ID.

Slots govern admission, not cancellation. There is no slot-wide cancel or remove operation. Stop queued work by task ID and registered work by task ID or task name. Removing or canceling work that is still queued or waiting for registry-command capacity removes it before it runs. Its waiter resolves to Rejected with RejectionKind::RemovedFromQueue, not Canceled.

Bound controller resources

ControllerConfig bounds separate stages:

Setting	What it bounds	When full
queue_capacity	Ordered controller commands.	execute() waits; try_intake() returns ControllerError::Full.
max_slot_queue	Pending depth behind one owner, including a replacement head.	New Queue submissions get watched Rejected with QueueFull.
max_total_pending	Slot queues and registry-capacity waits.	New pending entries get watched Rejected with ResourceLimit.
max_controller_slots	Tracked slots.	New slots get watched Rejected with ResourceLimit.
admission_capacity	Waits for registry-command capacity.	New waits get watched Rejected with ResourceLimit.
identity_operation_capacity	Concurrent registry-backed ID remove/cancel operations.	execute().await returns RuntimeError::ResourceLimitReached.

Queued controller removal happens before the identity-operation limit is checked. Buffered controller commands and work already handed to the registry do not count toward max_total_pending. Replacing an existing queue head does not need another pending unit. See the API reference for defaults and configuration for runtime and ownership limits.

Runnable controller examples:

• controller_slots.rs compares all three policies;
• controller_admission.rs watches admission and rejection;
• tenant_sync.rs keeps the newest waiting revision per tenant.

Source: submission intake, slot placement, queue policies, slot lifecycle, and completion signals.