[scope] Add typed outcomes and make taskvisor's purpose clear

Cargo.toml

@@ -4,13 +4,13 @@ version      = "0.6.0"
 edition      = "2024"
 rust-version = "1.90.0"
 
-description   = "Task supervisor for Tokio: restarts background tasks on failure with exponential backoff and jitter, graceful shutdown, and lifecycle events"
+description   = "In-process Tokio task supervisor with retries, reliable outcomes, and per-key queue/replace/reject admission"
 repository    = "https://github.com/soltiHQ/taskvisor"
 documentation = "https://docs.rs/taskvisor"
 license       = "Apache-2.0"
 readme        = "README.md"
 
-keywords   = ["supervisor", "tokio", "retry", "restart", "backoff"]
+keywords   = ["tokio", "supervisor", "keyed", "admission-control", "retry"]
 categories = ["asynchronous", "concurrency"]
 
 exclude = [
@@ -23,7 +23,7 @@ exclude = [
 
 [features]
 controller         = ["dep:dashmap"]
-default            = []
+default            = ["controller"]
 logging            = []
 test-util          = []
 tracing            = ["dep:tracing"]
@@ -56,6 +56,10 @@ required-features = ["controller"]
 name              = "admission"
 required-features = ["controller"]
 
+[[example]]
+name              = "tenant_sync"
+required-features = ["controller"]
+
 [[example]]
 name              = "tracing"
 required-features = ["tracing"]

examples/admission.rs

@@ -6,31 +6,26 @@
 //! `submit_and_watch` also returns a `TaskWaiter`:
 //!
 //! - an admitted task resolves to its final runtime outcome;
-//! - a task that never starts resolves to `TaskOutcome::Rejected` with a reason.
+//! - a submission rejected before registry admission resolves to `TaskOutcome::Rejected` with a typed `RejectionKind` and diagnostic reason; its task body never starts.
 //!
 //! This example shows both paths and reads a live controller snapshot.
 //! Use the waiter when rejection affects application logic.
 //! Events are best-effort and are better suited to logs and metrics.
 //!
-//! Run with
-//! `cargo run --example admission --features controller`.
-
-#[cfg(not(feature = "controller"))]
-compile_error!(
-    "This example requires the `controller` feature: cargo run --example admission --features controller"
-);
+//! Run with `cargo run --example admission`.
 
 use std::sync::Arc;
 use std::time::Duration;
 
-use taskvisor::ControllerSpec;
 use taskvisor::prelude::*;
+use taskvisor::{ControllerConfig, ControllerSpec, RejectionKind};
 use tokio::sync::Notify;
 
-/// A job that runs for `dur`, observing cancellation.
-fn job(name: &'static str, dur: Duration) -> TaskSpec {
+/// A job that runs for `duration`, observing cancellation.
+fn job(name: &'static str, duration: Duration) -> TaskSpec {
     let task: TaskRef = TaskFn::arc(name, move |ctx| async move {
-        ctx.run_until_cancelled(tokio::time::sleep(dur)).await?;
+        ctx.run_until_cancelled(tokio::time::sleep(duration))
+            .await?;
         Ok(())
     });
     TaskSpec::once(task)
@@ -52,10 +47,10 @@ fn gated_job(name: &'static str, started: Arc<Notify>, release: Arc<Notify>) ->
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let sup = Supervisor::builder(SupervisorConfig::default())
-        .with_controller(taskvisor::ControllerConfig::default())
+    let supervisor = Supervisor::builder(SupervisorConfig::default())
+        .with_controller(ControllerConfig::default())
         .build();
-    let handle = sup.serve();
+    let handle = supervisor.serve();
 
     println!("Slot 'deploy' admits at most one task at a time.\n");
 
@@ -101,7 +96,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .await?;
     match v2.wait().await? {
-        TaskOutcome::Rejected { reason, .. } => {
+        TaskOutcome::Rejected {
+            kind: RejectionKind::SlotBusy,
+            reason,
+            ..
+        } => {
             println!("    deploy-v2 -> Rejected ({reason}) - never ran\n");
         }
         other => println!("    deploy-v2 -> {other:?} (unexpected)\n"),
@@ -113,6 +112,5 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("    deploy-v1 -> {:?}", v1.wait().await?);
 
     handle.shutdown().await?;
-    println!("\nDone.");
     Ok(())
 }

examples/basic.rs

@@ -21,9 +21,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Ok(())
     });
 
-    let sup = Supervisor::new(SupervisorConfig::default(), vec![]);
-    sup.run(vec![TaskSpec::once(task)]).await?;
+    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
+    supervisor.run(vec![TaskSpec::once(task)]).await?;
 
-    println!("Done.");
     Ok(())
 }

examples/cpu_job.rs

@@ -25,8 +25,8 @@ fn sum_of_primes(limit: u64) -> u64 {
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let sup = Supervisor::new(SupervisorConfig::default(), vec![]);
-    let handle = sup.serve();
+    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
+    let handle = supervisor.serve();
 
     let attempts = Arc::new(AtomicU32::new(0));
     let job: TaskRef = TaskFn::arc("prime-sum", {

examples/dynamic.rs

@@ -44,16 +44,17 @@ fn make_worker(name: &'static str) -> TaskSpec {
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let sup = Supervisor::new(SupervisorConfig::default(), vec![]);
+    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
 
     // serve() starts listeners and returns a handle for dynamic management.
-    let handle = sup.serve();
+    let handle = supervisor.serve();
 
     // Add workers dynamically
     println!("Adding worker-a and worker-b...");
     let id_a = handle.add(make_worker("worker-a")).await?;
     let id_b = handle.add(make_worker("worker-b")).await?;
 
+    // Demo pacing only: add() has already confirmed registration.
     tokio::time::sleep(Duration::from_secs(1)).await;
     println!("Active: {:?}", handle.list().await);
 
@@ -71,6 +72,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Add worker-c
     println!("\nAdding worker-c...");
     handle.add(make_worker("worker-c")).await?;
+    // Demo pacing only: let the terminal show a worker-c tick.
     tokio::time::sleep(Duration::from_millis(500)).await;
 
     // Cancel worker-b
@@ -82,12 +84,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         handle.is_alive("worker-b").await
     );
 
+    // Demo pacing only: let worker-c keep ticking before the final snapshot.
     tokio::time::sleep(Duration::from_millis(500)).await;
     println!("\nActive: {:?}", handle.list().await);
 
     // Graceful shutdown (consumes the handle)
     println!("\nShutting down...");
     handle.shutdown().await?;
-    println!("Done.");
     Ok(())
 }

examples/metrics.rs

@@ -1,13 +1,14 @@
 //! # Metrics from lifecycle events
 //!
 //! This example implements `Subscribe` and maps each event to a Prometheus counter.
-//! `EventKind::as_label` provides stable, machine-readable values such as `task_failed` and `backoff_scheduled`.
+//! `EventKind::as_label` provides stable, machine-readable values such as `attempt_failed` and `backoff_scheduled`.
 //!
 //! A real service would expose the registry on its metrics endpoint.
 //! This example prints the Prometheus text format when it exits.
 //!
 //! The `task` label uses task names. Keep these names bounded and stable.
 //! Do not put request IDs, user IDs, or other unbounded values in metric labels.
+//! Terminal classification comes from `TaskOutcomeKind`; diagnostic `reason` text is deliberately not used as a label.
 //!
 //! Run with `cargo run --example metrics`.
 
@@ -25,10 +26,14 @@ struct PromSubscriber {
 }
 
 impl Subscribe for PromSubscriber {
-    fn on_event(&self, e: &Event) {
-        let task = e.task.as_deref().unwrap_or("none");
+    fn on_event(&self, event: &Event) {
+        let task = event.task.as_deref().unwrap_or("none");
+        let outcome = event
+            .outcome_kind
+            .map(TaskOutcomeKind::as_label)
+            .unwrap_or("none");
         self.events
-            .with_label_values(&[e.kind.as_label(), task])
+            .with_label_values(&[event.kind.as_label(), outcome, task])
             .inc();
     }
 
@@ -46,7 +51,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let registry = Registry::new();
     let events = IntCounterVec::new(
         Opts::new("taskvisor_events_total", "Supervisor lifecycle events"),
-        &["event", "task"],
+        &["event", "outcome", "task"],
     )?;
     registry.register(Box::new(events.clone()))?;
 
@@ -67,11 +72,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let spec = TaskSpec::restartable(flaky)
         .with_backoff(BackoffPolicy::constant(Duration::from_millis(100)));
 
-    let subs: Vec<Arc<dyn Subscribe>> = vec![Arc::new(PromSubscriber {
+    let subscribers: Vec<Arc<dyn Subscribe>> = vec![Arc::new(PromSubscriber {
         events: events.clone(),
     })];
-    let sup = Supervisor::new(SupervisorConfig::default(), subs);
-    sup.run(vec![spec]).await?;
+    let supervisor = Supervisor::new(SupervisorConfig::default(), subscribers);
+    supervisor.run(vec![spec]).await?;
 
     // In a real service: serve this string at GET /metrics.
     let mut buf = Vec::new();

examples/multiple.rs

@@ -6,7 +6,7 @@
 //! |-------------|--------------------|---------------------------------------|
 //! | `one-shot`  | never restart      | run once                              |
 //! | `resilient` | restart on failure | fail twice, then succeed              |
-//! | `always-on` | always restart     | wait 500 ms after each successful run |
+//! | `recurring` | periodic           | wait 500 ms after each successful run |
 //!
 //! The resilient task uses exponential backoff and a retry budget.
 //! Only failure-driven restarts consume that budget.
@@ -52,21 +52,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     });
 
-    // Always-on: repeats every 500ms until Ctrl+C
+    // Recurring: repeats every 500ms until Ctrl+C
     let cycle = Arc::new(AtomicU32::new(0));
-    let always_on: TaskRef = TaskFn::arc("always-on", move |_ctx| {
+    let recurring: TaskRef = TaskFn::arc("recurring", move |_ctx| {
         let cycle = Arc::clone(&cycle);
         async move {
             let n = cycle.fetch_add(1, Ordering::Relaxed) + 1;
-            println!("[always-on] cycle #{n}");
+            println!("[recurring] cycle #{n}");
             tokio::time::sleep(Duration::from_millis(300)).await;
             Ok(())
         }
     });
 
     let specs = vec![
         TaskSpec::once(one_shot),
-        TaskSpec::periodic(always_on, Duration::from_millis(500)),
+        TaskSpec::periodic(recurring, Duration::from_millis(500)),
         TaskSpec::restartable(resilient)
             .with_backoff(
                 BackoffPolicy::exponential(Duration::from_millis(200))
@@ -75,10 +75,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .with_max_retries(NonZeroU32::new(3).unwrap()),
     ];
 
-    let cfg = SupervisorConfig::default().with_grace(Duration::from_secs(5));
-    let sup = Supervisor::new(cfg, vec![]);
-    sup.run(specs).await?;
+    let config = SupervisorConfig::default().with_grace(Duration::from_secs(5));
+    let supervisor = Supervisor::new(config, vec![]);
+    supervisor.run(specs).await?;
 
-    println!("All tasks finished.");
     Ok(())
 }

examples/outcomes.rs

@@ -1,7 +1,7 @@
 //! # Outcomes: wait for the final result
 //!
 //! `add_and_watch` returns a `TaskWaiter`.
-//! The waiter resolves after the task entry has stopped, including all restarts allowed by its policy.
+//! The waiter resolves after all allowed attempts end, the managed runner is joined, and registry membership is removed.
 //!
 //! Taskvisor has two result paths:
 //!
@@ -10,8 +10,8 @@
 //! | lifecycle events | logs, metrics, live progress | bounded and best-effort    |
 //! | `TaskOutcome`    | final business decision      | dedicated terminal channel |
 //!
-//! This example handles successful, failed, and canceled tasks.
-//! Other outcomes cover fatal errors, force-abort, actor panic, and controller rejection.
+//! This example handles successful, retry-exhausted, timed-out, and canceled tasks.
+//! Other outcomes cover fatal errors, force-abort, task-runner panic, and controller rejection.
 //! `TaskWaiter::wait` can still return an error if the runtime closes the terminal channel unexpectedly.
 //!
 //! Run with `cargo run --example outcomes`.
@@ -22,11 +22,12 @@ use std::sync::atomic::{AtomicU32, Ordering};
 use std::time::Duration;
 
 use taskvisor::prelude::*;
+use tokio::sync::Notify;
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let sup = Supervisor::new(SupervisorConfig::default(), vec![]);
-    let handle = sup.serve();
+    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
+    let handle = supervisor.serve();
 
     // 1) A one-shot job that succeeds -> Completed.
     println!("=== Completed ===");
@@ -38,7 +39,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("  import -> {:?}\n", waiter.wait().await?);
 
     // 2) A task that always fails, with a bounded retry budget -> Failed.
-    //    Note the outcome's reason/exit_code are identical to the ActorExhausted event.
+    //    Its reason/exit_code are identical to the typed TaskFinished event.
     println!("=== Failed (retries exhausted) ===");
     let attempts = Arc::new(AtomicU32::new(0));
     let flaky: TaskRef = TaskFn::arc("sync", move |_ctx| {
@@ -61,19 +62,41 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         other => println!("  sync -> {other:?}\n"),
     }
 
-    // 3) A long-running worker we cancel -> Canceled.
+    // 3) A one-shot task that exceeds its per-attempt deadline -> Failed.
+    println!("=== Failed (attempt timed out) ===");
+    let slow: TaskRef = TaskFn::arc("slow-report", |_ctx| async {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        Ok(())
+    });
+    let timed = TaskSpec::once(slow).with_timeout(Duration::from_millis(20));
+    match handle.add_and_watch(timed).await?.1.wait().await? {
+        TaskOutcome::Failed { reason, .. } => {
+            println!("  slow-report -> Failed: {reason}\n");
+        }
+        other => println!("  slow-report -> {other:?}\n"),
+    }
+
+    // 4) A long-running worker we cancel -> Canceled.
     println!("=== Canceled ===");
-    let worker: TaskRef = TaskFn::arc("worker", |ctx| async move {
-        ctx.cancelled().await;
-        Err(TaskError::Canceled)
+    let started = Arc::new(Notify::new());
+    let worker: TaskRef = TaskFn::arc("worker", {
+        let started = Arc::clone(&started);
+        move |ctx| {
+            let started = Arc::clone(&started);
+            async move {
+                started.notify_one();
+                ctx.cancelled().await;
+                Err(TaskError::Canceled)
+            }
+        }
     });
     let (id, waiter) = handle.add_and_watch(TaskSpec::restartable(worker)).await?;
-    tokio::time::sleep(Duration::from_millis(100)).await;
+    // The task body, rather than a timer, confirms that the worker started.
+    started.notified().await;
     println!("  cancelling worker...");
     handle.cancel(id).await?;
     println!("  worker -> {:?}\n", waiter.wait().await?);
 
     handle.shutdown().await?;
-    println!("Done.");
     Ok(())
 }

examples/periodic.rs

@@ -27,8 +27,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let spec = TaskSpec::periodic(heartbeat, Duration::from_secs(2));
 
-    let sup = Supervisor::new(SupervisorConfig::default(), vec![]);
-    sup.run(vec![spec]).await?;
+    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);
+    supervisor.run(vec![spec]).await?;
 
     Ok(())
 }