TaskProcessor 2.0.0

TaskProcessor

A lightweight .NET 8 library for controlled concurrent task processing with built-in exponential backoff retry, fail-fast error handling, ordered results, progress reporting, and full ASP.NET Core DI support.

🚀 Overview

TaskProcessor helps you process large collections of tasks efficiently while preventing system overload.

It provides:

• Bounded parallel execution via a configurable semaphore
• Exponential backoff retry with full jitter
• Per-task timeout enforcement
• Fail-fast or collect-all-errors modes
• Ordered results guaranteed regardless of completion order
• Progress reporting via IProgress<int>
• Structured logging via ILogger
• ASP.NET Core DI integration via AddTaskProcessor

🎯 Problem It Solves

In real-world backend systems, processing large datasets (thousands of records, API calls, or jobs) can lead to:

• CPU spikes
• Database connection exhaustion
• API throttling
• Unstable performance

Naively running all tasks in parallel (Task.WhenAll) can overwhelm the system.

TaskProcessor solves this by limiting concurrency and adding production-grade reliability controls.

⚙️ Features

📦 Installation

dotnet add package TaskProcessor

🧠 Usage

1. ASP.NET Core — Dependency Injection

// Program.cs
builder.Services.AddTaskProcessor(options =>
{
    options.MaxDegreeOfParallelism = 5;
    options.RetryCount            = 3;
    options.RetryDelay            = TimeSpan.FromSeconds(1);
    options.MaxRetryDelay         = TimeSpan.FromSeconds(30);
    options.UseJitter             = true;
    options.TaskTimeout           = TimeSpan.FromSeconds(10);
    options.ContinueOnError       = true;
    options.ShouldRetry           = ex => ex is HttpRequestException;
    options.OnError               = (item, ex) => Console.WriteLine($"Failed: {ex.Message}");
    options.OnSuccess             = item => Console.WriteLine("Item succeeded");
});

Inject ITaskProcessorService wherever you need it:

public class MyService(ITaskProcessorService processor)
{
    public async Task RunAsync(IEnumerable<Order> orders)
    {
        await processor.ProcessAsync(orders, async order =>
        {
            await SendOrderAsync(order);
        });
    }
}

2. Manual instantiation — Process tasks (no return value)

var processor = new TaskProcessorService(new TaskProcessorOptions
{
    MaxDegreeOfParallelism = 3,
    RetryCount             = 2,
    RetryDelay             = TimeSpan.FromMilliseconds(500)
});

await processor.ProcessAsync(data, async item =>
{
    await ProcessItemAsync(item);
});

3. Transform data (with ordered return values)

Results are returned in the same order as the input collection, regardless of completion order.

var results = await processor.ProcessAsync<Source, Destination>(
    sourceData,
    async item =>
    {
        return await mapper.MapAsync(item);
    });

4. Progress reporting

var progress = new Progress<int>(count =>
    Console.WriteLine($"Completed {count} of {total}"));

await processor.ProcessAsync(items, ProcessItemAsync, progress);

5. Exception filtering — only retry transient errors

var options = new TaskProcessorOptions
{
    RetryCount  = 3,
    ShouldRetry = ex => ex is HttpRequestException or TimeoutException
};

🔁 How It Works

items ──► [SemaphoreSlim] ──► [RetryPolicy] ──► action(item)
              │                    │                  │
         max N tasks          exponential         OnSuccess
         at a time             backoff +          callback
                                jitter
                                  │
                              OnError +
                              ILogger

1. A SemaphoreSlim limits concurrency to MaxDegreeOfParallelism
2. Each item runs inside RetryPolicy.ExecuteAsync with exponential backoff and jitter
3. A linked CancellationTokenSource provides fail-fast cancellation when ContinueOnError = false
4. Results are written into an index-aligned array to preserve input order
5. All exceptions are collected and rethrown as AggregateException

📊 Use Cases

• Processing large datasets (data migration, ETL)
• Calling external APIs with rate limits or throttling
• Background job processing
• Bulk data transformations
• File or message queue processing

🏗️ Architecture

TaskProcessor/
├── ITaskProcessorService.cs       # Public interface (mockable, DI-friendly)
├── TaskProcessorService.cs        # Core implementation
├── TaskProcessorOptions.cs        # Configuration with validation
├── RetryPolicy.cs                 # Unified retry: backoff, jitter, filtering
└── ServiceCollectionExtensions.cs # AddTaskProcessor DI extension

TaskProcessor.Tests/
├── RetryPolicyTests.cs            # 7 unit tests for retry logic
└── TaskProcessorServiceTests.cs   # 11 unit tests for service behaviour

⚠️ Notes

• Results from ProcessAsync<T, TResult> are always returned in input order
• Best suited for I/O-bound, independent tasks
• For CPU-bound workloads, keep MaxDegreeOfParallelism at or below Environment.ProcessorCount
• TaskTimeout applies per individual task, not to the entire batch

🧩 Design Philosophy

You define the work — TaskProcessor manages how it runs.

Business logic stays in your lambdas. Concurrency, retry, cancellation, and observability are handled transparently by the library, making it reusable across different domains without coupling.

📄 License

MIT License