Struct TaskTracker 

pub struct TaskTracker(/* private fields */);

Hierarchical task tracker with pluggable scheduling and error policies

TaskTracker provides a composable system for managing background tasks with:

• Configurable scheduling via TaskScheduler implementations
• Flexible error handling via OnErrorPolicy implementations
• Parent-child relationships with independent metrics
• Cancellation propagation and isolation
• Built-in cancellation token support

Built on top of tokio_util::task::TaskTracker for robust task lifecycle management.

Example

// Create a task tracker with semaphore-based scheduling
let scheduler = SemaphoreScheduler::with_permits(3);
let policy = LogOnlyPolicy::new();
let root = TaskTracker::builder()
    .scheduler(scheduler)
    .error_policy(policy)
    .build()?;

// Spawn some tasks
let handle1 = root.spawn(async { Ok(1) });
let handle2 = root.spawn(async { Ok(2) });

// Get results and join all tasks
let result1 = handle1.await.unwrap().unwrap();
let result2 = handle2.await.unwrap().unwrap();
assert_eq!(result1, 1);
assert_eq!(result2, 2);

Implementations

impl TaskTracker

pub fn builder() -> TaskTrackerBuilder

Create a new root task tracker using the builder pattern

This is the preferred way to create new task trackers.

Example

let scheduler = SemaphoreScheduler::with_permits(10);
let error_policy = LogOnlyPolicy::new();
let tracker = TaskTracker::builder()
    .scheduler(scheduler)
    .error_policy(error_policy)
    .build()?;

pub fn new( scheduler: Arc<dyn TaskScheduler>, error_policy: Arc<dyn OnErrorPolicy>, ) -> Result<Self>

Create a new root task tracker with simple parameters (legacy)

This method is kept for backward compatibility. Use builder() for new code. Uses default metrics (no Prometheus integration).

Arguments

• scheduler - Scheduling policy to use for all tasks
• error_policy - Error handling policy for this tracker

Example

let scheduler = SemaphoreScheduler::with_permits(10);
let error_policy = LogOnlyPolicy::new();
let tracker = TaskTracker::new(scheduler, error_policy)?;

pub fn new_with_prometheus<R: MetricsRegistry + ?Sized>( scheduler: Arc<dyn TaskScheduler>, error_policy: Arc<dyn OnErrorPolicy>, registry: &R, component_name: &str, ) -> Result<Self>

Create a new root task tracker with Prometheus metrics integration

Arguments

• scheduler - Scheduling policy to use for all tasks
• error_policy - Error handling policy for this tracker
• registry - MetricsRegistry for Prometheus integration
• component_name - Name for this tracker component

Example

let scheduler = SemaphoreScheduler::with_permits(10);
let error_policy = LogOnlyPolicy::new();
let tracker = TaskTracker::new_with_prometheus(
    scheduler,
    error_policy,
    registry.as_ref(),
    "main_tracker"
)?;

pub fn child_tracker(&self) -> Result<TaskTracker>

Create a child tracker that inherits scheduling policy

The child tracker:

• Gets its own independent tokio TaskTracker
• Inherits the parent’s scheduler
• Gets a child error policy via create_child()
• Has hierarchical metrics that chain to parent
• Gets a child cancellation token from the parent
• Is independent for cancellation (child cancellation doesn’t affect parent)

Errors

Returns an error if the parent tracker is already closed

Example

let child_tracker = root_tracker.child_tracker()?;
// Child inherits parent's policies but has separate metrics and lifecycle

pub fn spawn<F, T>(&self, future: F) -> TaskHandle<T>

where F: Future<Output = Result<T>> + Send + 'static, T: Send + 'static,

Create a child tracker builder for flexible customization

The builder allows you to customize scheduling and error policies for the child tracker. If not specified, policies are inherited from the parent.

Example

// Custom scheduler, inherit error policy
let child1 = root_tracker.child_tracker_builder()
    .scheduler(SemaphoreScheduler::with_permits(5))
    .build().unwrap();

// Custom error policy, inherit scheduler
let child2 = root_tracker.child_tracker_builder()
    .error_policy(LogOnlyPolicy::new())
    .build().unwrap();

// Both custom
let child3 = root_tracker.child_tracker_builder()
    .scheduler(SemaphoreScheduler::with_permits(3))
    .error_policy(LogOnlyPolicy::new())
    .build().unwrap();

Spawn a new task

The task will be wrapped with scheduling and error handling logic, then executed according to the configured policies. For tasks that need to inspect cancellation tokens, use [spawn_cancellable] instead.

Arguments

• future - The async task to execute

Returns

A TaskHandle that can be used to await completion and access the task’s cancellation token

Panics

Panics if the tracker has been closed. This indicates a programming error where tasks are being spawned after the tracker lifecycle has ended.

Example

let handle = tracker.spawn(async {
    // Your async work here
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    Ok(42)
});

// Access the task's cancellation token
let cancel_token = handle.cancellation_token();

let result = handle.await?;

pub fn spawn_cancellable<F, Fut, T>(&self, task_fn: F) -> TaskHandle<T>

where F: FnMut(CancellationToken) -> Fut + Send + 'static, Fut: Future<Output = CancellableTaskResult<T>> + Send + 'static, T: Send + 'static,

Spawn a cancellable task that receives a cancellation token

This is useful for tasks that need to inspect the cancellation token and gracefully handle cancellation within their logic. The task function must return a CancellableTaskResult to properly track cancellation vs errors.

Arguments

• task_fn - Function that takes a cancellation token and returns a future that resolves to CancellableTaskResult<T>

Returns

A TaskHandle that can be used to await completion and access the task’s cancellation token

Panics

Panics if the tracker has been closed. This indicates a programming error where tasks are being spawned after the tracker lifecycle has ended.

Example

let handle = tracker.spawn_cancellable(|cancel_token| async move {
    tokio::select! {
        _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
            CancellableTaskResult::Ok(42)
        },
        _ = cancel_token.cancelled() => CancellableTaskResult::Cancelled,
    }
});

// Access the task's individual cancellation token
let task_cancel_token = handle.cancellation_token();

let result = handle.await?;

pub fn metrics(&self) -> &dyn HierarchicalTaskMetrics

Get metrics for this tracker

Metrics are specific to this tracker and do not include metrics from parent or child trackers.

Example

let metrics = tracker.metrics();
println!("Success: {}, Failed: {}", metrics.success(), metrics.failed());

pub fn cancel(&self)

Cancel this tracker and all its tasks

This will signal cancellation to all currently running tasks and prevent new tasks from being spawned. The cancellation is immediate and forceful.

Example

// Spawn a long-running task
let handle = tracker.spawn_cancellable(|cancel_token| async move {
    tokio::select! {
        _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
            dynamo_runtime::utils::tasks::tracker::CancellableTaskResult::Ok(42)
        }
        _ = cancel_token.cancelled() => {
            dynamo_runtime::utils::tasks::tracker::CancellableTaskResult::Cancelled
        }
    }
}).await?;

// Cancel the tracker (and thus the task)
tracker.cancel();

pub fn is_closed(&self) -> bool

Check if this tracker is closed

pub fn cancellation_token(&self) -> CancellationToken

Get the cancellation token for this tracker

This allows external code to observe or trigger cancellation of this tracker.

Example

let token = tracker.cancellation_token();
// Can check cancellation state or cancel manually
if !token.is_cancelled() {
    token.cancel();
}

pub fn child_count(&self) -> usize

Get the number of active child trackers

This counts only child trackers that are still alive (not dropped). Dropped child trackers are automatically cleaned up.

Example

let child_count = tracker.child_count();
println!("This tracker has {} active children", child_count);

pub fn child_tracker_builder(&self) -> ChildTrackerBuilder<'_>

Create a child tracker builder with custom configuration

This provides fine-grained control over child tracker creation, allowing you to override the scheduler or error policy while maintaining the parent-child relationship.

Example

// Custom scheduler, inherit error policy
let child1 = parent.child_tracker_builder()
    .scheduler(SemaphoreScheduler::with_permits(5))
    .build().unwrap();

// Custom error policy, inherit scheduler
let child2 = parent.child_tracker_builder()
    .error_policy(LogOnlyPolicy::new())
    .build().unwrap();

// Inherit both policies from parent
let child3 = parent.child_tracker_builder()
    .build().unwrap();

pub async fn join(&self)

Join this tracker and all child trackers

This method gracefully shuts down the entire tracker hierarchy by:

1. Closing all trackers (preventing new task spawning)
2. Waiting for all existing tasks to complete

Uses stack-safe traversal to prevent stack overflow in deep hierarchies. Children are processed before parents to ensure proper shutdown order.

Hierarchical Behavior:

• Processes children before parents to ensure proper shutdown order
• Each tracker is closed before waiting (Tokio requirement)
• Leaf trackers simply close and wait for their own tasks

Example

tracker.join().await;

Trait Implementations

impl Clone for TaskTracker

fn clone(&self) -> TaskTracker

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.