diff --git a/.github/workflows/build-samples.yml b/.github/workflows/build-samples.yml index 210f670..5e793a7 100644 --- a/.github/workflows/build-samples.yml +++ b/.github/workflows/build-samples.yml @@ -23,6 +23,7 @@ on: - "csharp-language/csharp-12-features/**" - "csharp-language/high-performance-memory-management/**" - "csharp-language/modern-patterns-result-pipeline/**" + - "dotnet-8-essentials/background-jobs-hostedservice-queues/**" - ".github/workflows/build-samples.yml" pull_request: @@ -46,6 +47,7 @@ on: - "csharp-language/csharp-12-features/**" - "csharp-language/high-performance-memory-management/**" - "csharp-language/modern-patterns-result-pipeline/**" + - "dotnet-8-essentials/background-jobs-hostedservice-queues/**" - ".github/workflows/build-samples.yml" workflow_dispatch: @@ -993,3 +995,108 @@ jobs: )" test "${line_count}" = "5" + + test-background-job-queue: + name: Test bounded background job queue + runs-on: ubuntu-latest + + permissions: + contents: read + + steps: + - name: Check out repository + uses: actions/checkout@v5 + + - name: Install .NET 10 SDK + uses: actions/setup-dotnet@v5 + with: + dotnet-version: "10.0.x" + + - name: Restore + run: > + dotnet restore + dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx + + - name: Build + run: > + dotnet build + dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx + --configuration Release + --no-restore + + - name: Test + run: > + dotnet test + dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx + --configuration Release + --no-build + + - name: Smoke-test API + shell: bash + run: | + app_log="${RUNNER_TEMP}/background-job-queue.log" + + ASPNETCORE_URLS="http://127.0.0.1:5096" \ + dotnet run \ + --project dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/BackgroundJobQueueMinimal.csproj \ + --configuration Release \ + --no-build \ + >"${app_log}" 2>&1 & + + app_pid="$!" + + cleanup() { + kill "${app_pid}" 2>/dev/null || true + wait "${app_pid}" 2>/dev/null || true + } + + trap cleanup EXIT + + for attempt in $(seq 1 40); do + if curl \ + --fail \ + --silent \ + http://127.0.0.1:5096/ \ + >"${RUNNER_TEMP}/background-job-root.json"; then + break + fi + + if ! kill -0 "${app_pid}" 2>/dev/null; then + cat "${app_log}" + exit 1 + fi + + sleep 0.25 + done + + root="$( + cat "${RUNNER_TEMP}/background-job-root.json" + )" + + expected_root='{"sample":"bounded-background-job-queue","durability":"in-memory","capacity":4}' + + test "${root}" = "${expected_root}" + + queue="$( + curl \ + --fail \ + --silent \ + http://127.0.0.1:5096/jobs/queue + )" + + expected_queue='{"capacity":4,"depth":0,"accepting":true}' + + test "${queue}" = "${expected_queue}" + + status="$( + curl \ + --silent \ + --output "${RUNNER_TEMP}/invalid-job.json" \ + --write-out '%{http_code}' \ + --request POST \ + --header 'Content-Type: application/json' \ + --data '{"to":"","subject":""}' \ + http://127.0.0.1:5096/jobs/email + )" + + test "${status}" = "400" diff --git a/README.md b/README.md index 2d3995c..443d6f0 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ Each sample folder contains a focused implementation of one tutorial topic. The | [`csharp-language/csharp-12-features`](csharp-language/csharp-12-features/) | Focused .NET 10 console lab locked to C# 12.0, demonstrating primary constructors with explicit properties, collection expressions and spreads, default lambda parameters, tuple-type aliases, deterministic output, and unit tests | [C# 12 Language Features: Primary Constructors, Collections & More](https://www.dotnet-guide.com/tutorials/csharp-language/csharp-12-features/) | | [`csharp-language/high-performance-memory-management`](csharp-language/high-performance-memory-management/) | Focused .NET 10 UTF-8 ingestion sample demonstrating PipeReader framing, segmented ReadOnlySequence handling, span-based numeric parsing, bounded MemoryPool ownership, strict input validation, deterministic output, and tests | [High-Performance C#: Span, Memory, SIMD & Pipelines](https://www.dotnet-guide.com/tutorials/csharp-language/high-performance-memory-management/) | | [`csharp-language/modern-patterns-result-pipeline`](csharp-language/modern-patterns-result-pipeline/) | Focused .NET 10 console companion locked to C# 12.0, demonstrating a custom Result type, safe and diagnostic errors, Map/Bind/BindAsync composition, invariant text-import validation, short-circuit persistence, cancellation, deterministic output, and tests | [C# 12 Functional Patterns: Result Type, Error Handling & Composable Pipeline Testing](https://www.dotnet-guide.com/tutorials/csharp-language/modern-patterns-result-pipeline/) | +| [`dotnet-8-essentials/background-jobs-hostedservice-queues`](dotnet-8-essentials/background-jobs-hostedservice-queues/) | Focused ASP.NET Core Minimal API demonstrating a bounded Channel queue, asynchronous backpressure, a BackgroundService consumer, fresh scoped handlers, safe in-memory job status, exception isolation, cancellation, and integration tests | [.NET 8 Background Jobs: IBackgroundTaskQueue, BackgroundService & Production-Ready Patterns](https://www.dotnet-guide.com/tutorials/dotnet-8-essentials/background-jobs-hostedservice-queues/) | ## Companion articles - [Common Microsoft.Extensions.AI mistakes](https://www.dotnet-guide.com/articles/dotnet-ai/microsoft-extensions-ai-common-mistakes/) @@ -131,6 +132,28 @@ tutorials/ | `-- TransactionalOutboxMinimal.Tests/ | |-- TransactionalOutboxMinimal.Tests.csproj | `-- OutboxFlowTests.cs +|-- dotnet-8-essentials/ +| `-- background-jobs-hostedservice-queues/ +| |-- BackgroundJobQueueMinimal.slnx +| |-- README.md +| |-- src/ +| | `-- BackgroundJobQueueMinimal/ +| | |-- BackgroundJobQueueMinimal.csproj +| | |-- Program.cs +| | |-- Jobs/ +| | | |-- BackgroundJobModels.cs +| | | |-- IBackgroundJobQueue.cs +| | | |-- BoundedBackgroundJobQueue.cs +| | | |-- IJobTracker.cs +| | | |-- InMemoryJobTracker.cs +| | | `-- QueuedEmailWorker.cs +| | `-- Services/ +| | |-- IEmailJobHandler.cs +| | `-- FakeEmailJobHandler.cs +| `-- tests/ +| `-- BackgroundJobQueueMinimal.Tests/ +| |-- BackgroundJobQueueMinimal.Tests.csproj +| `-- BackgroundJobQueueTests.cs |-- blazor/ | |-- create-interactive-ui-csharp-12/ | | |-- BlazorTodoMinimal.slnx diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx b/dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx new file mode 100644 index 0000000..878e48d --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/BackgroundJobQueueMinimal.slnx @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/README.md b/dotnet-8-essentials/background-jobs-hostedservice-queues/README.md new file mode 100644 index 0000000..dd2db2b --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/README.md @@ -0,0 +1,221 @@ +# Bounded Background Job Queue + +A focused ASP.NET Core Minimal API companion demonstrating a bounded +`Channel`, asynchronous producer backpressure, a `BackgroundService` +consumer, fresh dependency-injection scopes per job, safe in-memory status +tracking, exception isolation, and cooperative cancellation. + +## Full tutorial + +[.NET 8 Background Jobs: IBackgroundTaskQueue, BackgroundService & Production-Ready Patterns](https://www.dotnet-guide.com/tutorials/dotnet-8-essentials/background-jobs-hostedservice-queues/) + +## Framework note + +The full tutorial is written for ASP.NET Core 8. + +This repository companion targets .NET 10 because that is the current DOTNET +GUIDE sample SDK. + +The core APIs demonstrated here are the same hosted-service and +`System.Threading.Channels` patterns. + +## Flow + +```text +POST /jobs/email + -> validate + -> create typed EmailJob + -> register Queued state + -> bounded Channel + -> BackgroundService + -> fresh async DI scope + -> scoped handler + -> terminal status +``` + +## What this sample proves + +- the queue is bounded; +- producers wait asynchronously when it is full; +- jobs are consumed in FIFO order; +- one consumer processes jobs sequentially; +- every job gets a fresh dependency-injection scope; +- one failed job does not stop the next job; +- host cancellation reaches the in-flight handler; +- status responses do not expose exception details; +- queue admission closes during worker shutdown. + +## Non-durability boundary + +The queue and status tracker live only in process memory. + +A process restart loses: + +- queued jobs; +- running state; +- completed status history. + +`202 Accepted` means the job entered this process's in-memory queue. + +It does not mean the job is durably stored or guaranteed to finish. + +Use a durable broker or job framework when work must survive restarts. + +## Typed jobs + +The queue stores an `EmailJob` record rather than a delegate. + +This keeps job payloads inspectable and avoids implying that delegates can be +serialized into Redis or a cloud queue. + +A durable implementation would still need: + +- a message contract; +- serialization; +- delivery acknowledgements; +- visibility or lease handling; +- retry classification; +- poison-message storage; +- idempotency. + +## Shutdown model + +This sample uses cancellation-first shutdown. + +When stopping begins: + +1. queue admission closes; +2. the worker token is canceled; +3. the in-flight handler receives cancellation; +4. queued items can remain unprocessed. + +This is not a full drain or durable recovery implementation. + +## API + +### Enqueue + +```http +POST /jobs/email +Content-Type: application/json + +{ + "to": "reader@example.com", + "subject": "Queue sample" +} +``` + +Returns: + +```text +202 Accepted +Location: /jobs/{jobId} +``` + +### Job status + +```http +GET /jobs/{jobId} +``` + +### Local queue status + +```http +GET /jobs/queue +``` + +The queue-status endpoint is for demonstration and testing. + +It is not an authenticated production admin API. + +### Cancellation and concurrency + +A producer waiting for bounded capacity uses the HTTP request cancellation +token. If the request is canceled before the channel accepts the job, the +temporary tracker registration is removed and the cancellation continues to +the caller. + +Status snapshots returned by `GET /jobs/{jobId}` are synchronized +independently from the `ConcurrentDictionary` that stores tracker entries. +Each snapshot read acquires the per-entry lock, ensuring a consistent view of +terminal state after the worker transitions the job. + +## Prerequisite + +- .NET 10 SDK + +## Restore, build, and test + +```powershell +dotnet restore ` + .\BackgroundJobQueueMinimal.slnx + +dotnet build ` + .\BackgroundJobQueueMinimal.slnx ` + --configuration Release ` + --no-restore + +dotnet test ` + .\BackgroundJobQueueMinimal.slnx ` + --configuration Release ` + --no-build +``` + +## Run + +```powershell +dotnet run ` + --project .\src\BackgroundJobQueueMinimal\BackgroundJobQueueMinimal.csproj ` + --configuration Release ` + --no-build ` + --urls http://127.0.0.1:5096 +``` + +## Project structure + +```text +BackgroundJobQueueMinimal.slnx +README.md +src/ +`-- BackgroundJobQueueMinimal/ + |-- BackgroundJobQueueMinimal.csproj + |-- Program.cs + |-- Jobs/ + | |-- BackgroundJobModels.cs + | |-- IBackgroundJobQueue.cs + | |-- BoundedBackgroundJobQueue.cs + | |-- IJobTracker.cs + | |-- InMemoryJobTracker.cs + | `-- QueuedEmailWorker.cs + `-- Services/ + |-- IEmailJobHandler.cs + `-- FakeEmailJobHandler.cs +tests/ +`-- BackgroundJobQueueMinimal.Tests/ + |-- BackgroundJobQueueMinimal.Tests.csproj + `-- BackgroundJobQueueTests.cs +``` + +## Deliberately omitted + +- retries; +- Polly; +- dead-letter storage; +- schedules; +- metrics exporters; +- health probes; +- durable brokers; +- databases; +- real email delivery; +- multiple workers; +- parallel processing. + +## Verification + +- Target framework: .NET 10 +- Application NuGet dependencies: none +- Test count: 8 +- External services: none +- Queue capacity: 4 by default +- Durability: in-memory only +- Last reviewed: 2026-08-06 \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/BackgroundJobQueueMinimal.csproj b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/BackgroundJobQueueMinimal.csproj new file mode 100644 index 0000000..b9e3a83 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/BackgroundJobQueueMinimal.csproj @@ -0,0 +1,10 @@ + + + + net10.0 + enable + enable + true + + + \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BackgroundJobModels.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BackgroundJobModels.cs new file mode 100644 index 0000000..8ee0adc --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BackgroundJobModels.cs @@ -0,0 +1,45 @@ +namespace BackgroundJobQueueMinimal.Jobs; + +public sealed class BackgroundJobQueueOptions +{ + public const string SectionName = + "BackgroundJobs"; + + public int Capacity + { + get; + set; + } = 4; +} + +public sealed record EmailJobRequest( + string? To, + string? Subject); + +public sealed record EmailJob( + Guid JobId, + string To, + string Subject, + DateTimeOffset EnqueuedAt); + +public enum JobState +{ + Queued, + Running, + Succeeded, + Failed, + Canceled +} + +public sealed record JobSnapshot( + Guid JobId, + string JobName, + JobState State, + DateTimeOffset EnqueuedAt, + string? FailureCode = null, + string? FailureMessage = null); + +public sealed record QueueSnapshot( + int Capacity, + int Depth, + bool Accepting); \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BoundedBackgroundJobQueue.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BoundedBackgroundJobQueue.cs new file mode 100644 index 0000000..d25c65e --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/BoundedBackgroundJobQueue.cs @@ -0,0 +1,124 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Microsoft.Extensions.Options; + +namespace BackgroundJobQueueMinimal.Jobs; + +public sealed class BoundedBackgroundJobQueue : + IBackgroundJobQueue +{ + private readonly + Channel + _channel; + + private int _accepting = + 1; + + public BoundedBackgroundJobQueue( + IOptions< + BackgroundJobQueueOptions> + options) + { + ArgumentNullException + .ThrowIfNull( + options); + + int capacity = + options.Value.Capacity; + + if (capacity + <= 0) + { + throw new + ArgumentOutOfRangeException( + nameof(options), + capacity, + "Queue capacity must be greater than zero."); + } + + Capacity = + capacity; + + _channel = + Channel.CreateBounded< + EmailJob>( + new + BoundedChannelOptions( + capacity) + { + FullMode = + BoundedChannelFullMode + .Wait, + + SingleReader = + true, + + SingleWriter = + false, + + AllowSynchronousContinuations = + false + }); + } + + public int Capacity + { + get; + } + + public int Depth => + _channel.Reader.Count; + + public bool IsAccepting => + Volatile.Read( + ref _accepting) + == 1; + + public ValueTask EnqueueAsync( + EmailJob job, + CancellationToken + cancellationToken = + default) + { + ArgumentNullException + .ThrowIfNull( + job); + + return _channel.Writer + .WriteAsync( + job, + cancellationToken); + } + + public async + IAsyncEnumerable + ReadAllAsync( + [EnumeratorCancellation] + CancellationToken + cancellationToken = + default) + { + await foreach ( + EmailJob job + in _channel.Reader + .ReadAllAsync( + cancellationToken)) + { + yield return job; + } + } + + public void Complete() + { + if (Interlocked.Exchange( + ref _accepting, + 0) + == 0) + { + return; + } + + _channel.Writer + .TryComplete(); + } +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IBackgroundJobQueue.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IBackgroundJobQueue.cs new file mode 100644 index 0000000..8973742 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IBackgroundJobQueue.cs @@ -0,0 +1,33 @@ +namespace BackgroundJobQueueMinimal.Jobs; + +public interface IBackgroundJobQueue +{ + int Capacity + { + get; + } + + int Depth + { + get; + } + + bool IsAccepting + { + get; + } + + ValueTask EnqueueAsync( + EmailJob job, + CancellationToken + cancellationToken = + default); + + IAsyncEnumerable + ReadAllAsync( + CancellationToken + cancellationToken = + default); + + void Complete(); +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IJobTracker.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IJobTracker.cs new file mode 100644 index 0000000..3822f82 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/IJobTracker.cs @@ -0,0 +1,36 @@ +namespace BackgroundJobQueueMinimal.Jobs; + +public interface IJobTracker +{ + void Register( + EmailJob job); + + void MarkRunning( + Guid jobId); + + void MarkSucceeded( + Guid jobId); + + void MarkFailed( + Guid jobId, + string failureCode, + string failureMessage); + + void MarkCanceled( + Guid jobId); + + bool TryGet( + Guid jobId, + out JobSnapshot? + snapshot); + + bool Remove( + Guid jobId); + + Task + WaitForTerminalAsync( + Guid jobId, + CancellationToken + cancellationToken = + default); +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/InMemoryJobTracker.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/InMemoryJobTracker.cs new file mode 100644 index 0000000..1905394 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/InMemoryJobTracker.cs @@ -0,0 +1,283 @@ +using System.Collections.Concurrent; + +namespace BackgroundJobQueueMinimal.Jobs; + +public sealed class InMemoryJobTracker : + IJobTracker +{ + private readonly + ConcurrentDictionary< + Guid, + Entry> + _entries = + new(); + + public void Register( + EmailJob job) + { + ArgumentNullException + .ThrowIfNull( + job); + + var snapshot = + new JobSnapshot( + JobId: + job.JobId, + + JobName: + "send-email", + + State: + JobState.Queued, + + EnqueuedAt: + job.EnqueuedAt); + + if (!_entries.TryAdd( + job.JobId, + new Entry( + snapshot))) + { + throw new + InvalidOperationException( + $"Job '{job.JobId}' is already registered."); + } + } + + public void MarkRunning( + Guid jobId) => + Update( + jobId, + current => + current + with + { + State = + JobState.Running + }); + + public void MarkSucceeded( + Guid jobId) => + Complete( + jobId, + current => + current + with + { + State = + JobState.Succeeded + }); + + public void MarkFailed( + Guid jobId, + string failureCode, + string failureMessage) + { + ArgumentException + .ThrowIfNullOrWhiteSpace( + failureCode); + + ArgumentException + .ThrowIfNullOrWhiteSpace( + failureMessage); + + Complete( + jobId, + current => + current + with + { + State = + JobState.Failed, + + FailureCode = + failureCode, + + FailureMessage = + failureMessage + }); + } + + public void MarkCanceled( + Guid jobId) => + Complete( + jobId, + current => + current + with + { + State = + JobState.Canceled, + + FailureCode = + "JOB_CANCELED", + + FailureMessage = + "Job processing was canceled." + }); + + public bool TryGet( + Guid jobId, + out JobSnapshot? + snapshot) + { + if (_entries.TryGetValue( + jobId, + out Entry? + entry)) + { + lock (entry.SyncRoot) + { + snapshot = + entry.Snapshot; + } + + return true; + } + + snapshot = + null; + + return false; + } + + public bool Remove( + Guid jobId) => + _entries.TryRemove( + jobId, + out _); + + public async Task + WaitForTerminalAsync( + Guid jobId, + CancellationToken + cancellationToken = + default) + { + if (!_entries.TryGetValue( + jobId, + out Entry? + entry)) + { + throw new + KeyNotFoundException( + $"Job '{jobId}' was not found."); + } + + Task + completionTask; + + lock (entry.SyncRoot) + { + if (IsTerminal( + entry.Snapshot.State)) + { + return entry.Snapshot; + } + + completionTask = + entry.Completion.Task; + } + + return await completionTask + .WaitAsync( + cancellationToken); + } + + private void Update( + Guid jobId, + Func< + JobSnapshot, + JobSnapshot> + update) + { + if (!_entries.TryGetValue( + jobId, + out Entry? + entry)) + { + throw new + KeyNotFoundException( + $"Job '{jobId}' was not found."); + } + + lock (entry.SyncRoot) + { + entry.Snapshot = + update( + entry.Snapshot); + } + } + + private void Complete( + Guid jobId, + Func< + JobSnapshot, + JobSnapshot> + update) + { + if (!_entries.TryGetValue( + jobId, + out Entry? + entry)) + { + throw new + KeyNotFoundException( + $"Job '{jobId}' was not found."); + } + + JobSnapshot completed; + + lock (entry.SyncRoot) + { + if (IsTerminal( + entry.Snapshot.State)) + { + return; + } + + completed = + update( + entry.Snapshot); + + entry.Snapshot = + completed; + } + + entry.Completion + .TrySetResult( + completed); + } + + private static bool IsTerminal( + JobState state) => + state + is JobState.Succeeded + or JobState.Failed + or JobState.Canceled; + + private sealed class Entry( + JobSnapshot snapshot) + { + public object SyncRoot + { + get; + } = new(); + + public JobSnapshot Snapshot + { + get; + set; + } = snapshot; + + public TaskCompletionSource< + JobSnapshot> + Completion + { + get; + } = + new( + TaskCreationOptions + .RunContinuationsAsynchronously); + } +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/QueuedEmailWorker.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/QueuedEmailWorker.cs new file mode 100644 index 0000000..8561ee0 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Jobs/QueuedEmailWorker.cs @@ -0,0 +1,92 @@ +using BackgroundJobQueueMinimal.Services; + +namespace BackgroundJobQueueMinimal.Jobs; + +public sealed class QueuedEmailWorker( + IBackgroundJobQueue queue, + IJobTracker tracker, + IServiceScopeFactory + scopeFactory, + ILogger< + QueuedEmailWorker> + logger) : + BackgroundService +{ + protected override async Task + ExecuteAsync( + CancellationToken + stoppingToken) + { + logger.LogInformation( + "Background email worker started"); + + await foreach ( + EmailJob job + in queue.ReadAllAsync( + stoppingToken)) + { + tracker.MarkRunning( + job.JobId); + + try + { + await using + AsyncServiceScope scope = + scopeFactory + .CreateAsyncScope(); + + IEmailJobHandler handler = + scope.ServiceProvider + .GetRequiredService< + IEmailJobHandler>(); + + await handler.HandleAsync( + job, + stoppingToken); + + tracker.MarkSucceeded( + job.JobId); + + logger.LogInformation( + "Background job {JobId} succeeded", + job.JobId); + } + catch ( + OperationCanceledException) + when ( + stoppingToken + .IsCancellationRequested) + { + tracker.MarkCanceled( + job.JobId); + + throw; + } + catch (Exception exception) + { + tracker.MarkFailed( + job.JobId, + "JOB_PROCESSING_FAILED", + "Job processing failed."); + + logger.LogError( + exception, + "Background job {JobId} failed", + job.JobId); + } + } + + logger.LogInformation( + "Background email worker stopped"); + } + + public override async Task StopAsync( + CancellationToken + cancellationToken) + { + queue.Complete(); + + await base.StopAsync( + cancellationToken); + } +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Program.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Program.cs new file mode 100644 index 0000000..13e2f5b --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Program.cs @@ -0,0 +1,223 @@ +using System.Threading.Channels; +using BackgroundJobQueueMinimal.Jobs; +using BackgroundJobQueueMinimal.Services; + +WebApplicationBuilder builder = + WebApplication.CreateBuilder( + args); + +builder.Services + .Configure< + BackgroundJobQueueOptions>( + builder.Configuration + .GetSection( + BackgroundJobQueueOptions + .SectionName)); + +builder.Services + .AddSingleton< + IBackgroundJobQueue, + BoundedBackgroundJobQueue>(); + +builder.Services + .AddSingleton< + IJobTracker, + InMemoryJobTracker>(); + +builder.Services + .AddScoped< + IEmailJobHandler, + FakeEmailJobHandler>(); + +builder.Services + .AddHostedService< + QueuedEmailWorker>(); + +builder.Services + .AddSingleton( + TimeProvider.System); + +WebApplication app = + builder.Build(); + +app.MapGet( + "/", + ( + IBackgroundJobQueue + queue) => + Results.Ok( + new + { + Sample = + "bounded-background-job-queue", + + Durability = + "in-memory", + + queue.Capacity + })); + +app.MapPost( + "/jobs/email", + async ( + EmailJobRequest request, + IBackgroundJobQueue queue, + IJobTracker tracker, + TimeProvider timeProvider, + CancellationToken + cancellationToken) => + { + Dictionary + errors = + Validate( + request); + + if (errors.Count + > 0) + { + return Results + .ValidationProblem( + errors); + } + + var job = + new EmailJob( + JobId: + Guid.NewGuid(), + + To: + request.To!.Trim(), + + Subject: + request.Subject!.Trim(), + + EnqueuedAt: + timeProvider + .GetUtcNow()); + + tracker.Register( + job); + + try + { + await queue.EnqueueAsync( + job, + cancellationToken); + } + catch ( + OperationCanceledException) + when ( + cancellationToken + .IsCancellationRequested) + { + tracker.Remove( + job.JobId); + + throw; + } + catch ( + ChannelClosedException) + { + tracker.Remove( + job.JobId); + + return Results.Problem( + statusCode: + StatusCodes + .Status503ServiceUnavailable, + + title: + "Queue admission is closed.", + + detail: + "The application is stopping and is not accepting new jobs."); + } + + return Results.Accepted( + $"/jobs/{job.JobId}", + new + { + job.JobId, + State = + JobState.Queued + .ToString() + }); + }); + +app.MapGet( + "/jobs/{jobId:guid}", + ( + Guid jobId, + IJobTracker tracker) => + tracker.TryGet( + jobId, + out JobSnapshot? + snapshot) + ? Results.Ok( + snapshot) + : Results.NotFound()); + +app.MapGet( + "/jobs/queue", + ( + IBackgroundJobQueue + queue) => + Results.Ok( + new QueueSnapshot( + Capacity: + queue.Capacity, + + Depth: + queue.Depth, + + Accepting: + queue.IsAccepting))); + +app.Run(); + +static Dictionary< + string, + string[]> + Validate( + EmailJobRequest request) +{ + var errors = + new Dictionary< + string, + string[]>( + StringComparer + .Ordinal); + + if (string.IsNullOrWhiteSpace( + request.To) + || !request.To.Contains( + '@', + StringComparison.Ordinal)) + { + errors["to"] = + [ + "A recipient email address is required." + ]; + } + + if (string.IsNullOrWhiteSpace( + request.Subject)) + { + errors["subject"] = + [ + "A subject is required." + ]; + } + else if (request.Subject.Length + > 100) + { + errors["subject"] = + [ + "The subject must not exceed 100 characters." + ]; + } + + return errors; +} + +public partial class Program; \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/FakeEmailJobHandler.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/FakeEmailJobHandler.cs new file mode 100644 index 0000000..990f000 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/FakeEmailJobHandler.cs @@ -0,0 +1,30 @@ +using BackgroundJobQueueMinimal.Jobs; + +namespace BackgroundJobQueueMinimal.Services; + +public sealed class FakeEmailJobHandler( + ILogger< + FakeEmailJobHandler> + logger) : + IEmailJobHandler +{ + public async Task HandleAsync( + EmailJob job, + CancellationToken + cancellationToken) + { + ArgumentNullException + .ThrowIfNull( + job); + + await Task.Delay( + TimeSpan + .FromMilliseconds( + 25), + cancellationToken); + + logger.LogInformation( + "Processed background email job {JobId}", + job.JobId); + } +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/IEmailJobHandler.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/IEmailJobHandler.cs new file mode 100644 index 0000000..bde33d2 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/src/BackgroundJobQueueMinimal/Services/IEmailJobHandler.cs @@ -0,0 +1,11 @@ +using BackgroundJobQueueMinimal.Jobs; + +namespace BackgroundJobQueueMinimal.Services; + +public interface IEmailJobHandler +{ + Task HandleAsync( + EmailJob job, + CancellationToken + cancellationToken); +} \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueMinimal.Tests.csproj b/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueMinimal.Tests.csproj new file mode 100644 index 0000000..a3332a4 --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueMinimal.Tests.csproj @@ -0,0 +1,50 @@ + + + + net10.0 + enable + enable + true + false + true + Exe + + + + + + + + + + + all + + runtime; + build; + native; + contentfiles; + analyzers; + buildtransitive + + + + + + + + + + + + + \ No newline at end of file diff --git a/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueTests.cs b/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueTests.cs new file mode 100644 index 0000000..b3d49bd --- /dev/null +++ b/dotnet-8-essentials/background-jobs-hostedservice-queues/tests/BackgroundJobQueueMinimal.Tests/BackgroundJobQueueTests.cs @@ -0,0 +1,801 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Http.Json; +using System.Threading.Channels; +using BackgroundJobQueueMinimal.Jobs; +using BackgroundJobQueueMinimal.Services; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Mvc.Testing; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; + +namespace BackgroundJobQueueMinimal.Tests; + +public sealed class BackgroundJobQueueTests +{ + [Fact] + public void + Queue_rejects_nonpositive_capacity() + { + Assert.Throws< + ArgumentOutOfRangeException>( + () => + CreateQueue( + 0)); + + Assert.Throws< + ArgumentOutOfRangeException>( + () => + CreateQueue( + -1)); + } + + [Fact] + public async Task + Queue_preserves_fifo_and_reports_depth() + { + BoundedBackgroundJobQueue queue = + CreateQueue( + 3); + + EmailJob first = + CreateJob( + "first"); + + EmailJob second = + CreateJob( + "second"); + + await queue.EnqueueAsync( + first, + TestContext.Current + .CancellationToken); + + await queue.EnqueueAsync( + second, + TestContext.Current + .CancellationToken); + + Assert.Equal( + 2, + queue.Depth); + + await using + IAsyncEnumerator< + EmailJob> + reader = + queue + .ReadAllAsync( + TestContext + .Current + .CancellationToken) + .GetAsyncEnumerator( + TestContext + .Current + .CancellationToken); + + Assert.True( + await reader + .MoveNextAsync()); + + Assert.Equal( + first.JobId, + reader.Current.JobId); + + Assert.True( + await reader + .MoveNextAsync()); + + Assert.Equal( + second.JobId, + reader.Current.JobId); + + Assert.Equal( + 0, + queue.Depth); + } + + [Fact] + public async Task + Bounded_queue_waits_until_capacity_is_available() + { + BoundedBackgroundJobQueue queue = + CreateQueue( + 1); + + EmailJob first = + CreateJob( + "first"); + + EmailJob second = + CreateJob( + "second"); + + await queue.EnqueueAsync( + first, + TestContext.Current + .CancellationToken); + + Task secondWrite = + queue.EnqueueAsync( + second, + TestContext.Current + .CancellationToken) + .AsTask(); + + Assert.False( + secondWrite.IsCompleted); + + await using + IAsyncEnumerator< + EmailJob> + reader = + queue + .ReadAllAsync( + TestContext + .Current + .CancellationToken) + .GetAsyncEnumerator( + TestContext + .Current + .CancellationToken); + + Assert.True( + await reader + .MoveNextAsync()); + + Assert.Equal( + first.JobId, + reader.Current.JobId); + + await secondWrite + .WaitAsync( + TestContext.Current + .CancellationToken); + + Assert.True( + await reader + .MoveNextAsync()); + + Assert.Equal( + second.JobId, + reader.Current.JobId); + + // Second scenario: cancellation of a pending bounded write + { + BoundedBackgroundJobQueue cancelQueue = + CreateQueue( + 1); + + EmailJob blocking = + CreateJob( + "blocking"); + + EmailJob canceled = + CreateJob( + "canceled"); + + await cancelQueue.EnqueueAsync( + blocking, + TestContext.Current + .CancellationToken); + + using var cancelSource = + new CancellationTokenSource(); + + Task canceledWrite = + cancelQueue.EnqueueAsync( + canceled, + cancelSource.Token) + .AsTask(); + + Assert.False( + canceledWrite.IsCompleted); + + cancelSource.Cancel(); + + await Assert.ThrowsAsync< + OperationCanceledException>( + async () => + await canceledWrite); + + Assert.Equal( + 1, + cancelQueue.Depth); + + await using + IAsyncEnumerator< + EmailJob> + cancelReader = + cancelQueue + .ReadAllAsync( + TestContext + .Current + .CancellationToken) + .GetAsyncEnumerator( + TestContext + .Current + .CancellationToken); + + Assert.True( + await cancelReader + .MoveNextAsync()); + + Assert.Equal( + blocking.JobId, + cancelReader.Current.JobId); + + cancelQueue.Complete(); + } + } + + [Fact] + public async Task + Completion_drains_buffer_and_rejects_new_writes() + { + BoundedBackgroundJobQueue queue = + CreateQueue( + 2); + + EmailJob first = + CreateJob( + "first"); + + EmailJob second = + CreateJob( + "second"); + + await queue.EnqueueAsync( + first, + TestContext.Current + .CancellationToken); + + await queue.EnqueueAsync( + second, + TestContext.Current + .CancellationToken); + + queue.Complete(); + queue.Complete(); + + Assert.False( + queue.IsAccepting); + + var received = + new List(); + + await foreach ( + EmailJob job + in queue.ReadAllAsync( + TestContext.Current + .CancellationToken)) + { + received.Add( + job.JobId); + } + + Assert.Equal( + [ + first.JobId, + second.JobId + ], + received); + + await Assert.ThrowsAsync< + ChannelClosedException>( + async () => + await queue.EnqueueAsync( + CreateJob( + "late"), + TestContext + .Current + .CancellationToken)); + } + + [Fact] + public async Task + Worker_uses_fresh_scopes_and_isolates_job_failure() + { + BoundedBackgroundJobQueue queue = + CreateQueue( + 4); + + var tracker = + new InMemoryJobTracker(); + + var recorder = + new HandlerRecorder(); + + var services = + new ServiceCollection(); + + services.AddSingleton( + recorder); + + services.AddScoped< + ScopedMarker>(); + + services.AddScoped< + IEmailJobHandler, + RecordingEmailJobHandler>(); + + await using + ServiceProvider provider = + services + .BuildServiceProvider(); + + var worker = + new QueuedEmailWorker( + queue, + tracker, + provider + .GetRequiredService< + IServiceScopeFactory>(), + NullLogger< + QueuedEmailWorker> + .Instance); + + EmailJob failing = + CreateJob( + "fail"); + + EmailJob succeeding = + CreateJob( + "succeed"); + + tracker.Register( + failing); + + tracker.Register( + succeeding); + + await worker.StartAsync( + TestContext.Current + .CancellationToken); + + await queue.EnqueueAsync( + failing, + TestContext.Current + .CancellationToken); + + await queue.EnqueueAsync( + succeeding, + TestContext.Current + .CancellationToken); + + JobSnapshot first = + await tracker + .WaitForTerminalAsync( + failing.JobId, + TestContext + .Current + .CancellationToken); + + JobSnapshot second = + await tracker + .WaitForTerminalAsync( + succeeding.JobId, + TestContext + .Current + .CancellationToken); + + Assert.Equal( + JobState.Failed, + first.State); + + Assert.Equal( + "JOB_PROCESSING_FAILED", + first.FailureCode); + + Assert.Equal( + "Job processing failed.", + first.FailureMessage); + + Assert.Equal( + JobState.Succeeded, + second.State); + + Assert.Equal( + 2, + recorder.ScopeIds + .Distinct() + .Count()); + + // Verify TryGet returns the same terminal state + Assert.True( + tracker.TryGet( + failing.JobId, + out JobSnapshot? + tryGetFirst)); + + Assert.NotNull( + tryGetFirst); + + Assert.Equal( + first.State, + tryGetFirst.State); + + Assert.Equal( + first.FailureCode, + tryGetFirst.FailureCode); + + Assert.Equal( + first.FailureMessage, + tryGetFirst.FailureMessage); + + Assert.True( + tracker.TryGet( + succeeding.JobId, + out JobSnapshot? + tryGetSecond)); + + Assert.NotNull( + tryGetSecond); + + Assert.Equal( + second.State, + tryGetSecond.State); + + Assert.Null( + tryGetSecond.FailureCode); + + Assert.Null( + tryGetSecond.FailureMessage); + + using var stop = + new CancellationTokenSource( + TimeSpan.FromSeconds( + 2)); + + await worker.StopAsync( + stop.Token); + } + + [Fact] + public async Task + Worker_cancellation_marks_inflight_job_canceled() + { + BoundedBackgroundJobQueue queue = + CreateQueue( + 1); + + var tracker = + new InMemoryJobTracker(); + + var services = + new ServiceCollection(); + + services.AddScoped< + IEmailJobHandler, + BlockingEmailJobHandler>(); + + await using + ServiceProvider provider = + services + .BuildServiceProvider(); + + var worker = + new QueuedEmailWorker( + queue, + tracker, + provider + .GetRequiredService< + IServiceScopeFactory>(), + NullLogger< + QueuedEmailWorker> + .Instance); + + EmailJob job = + CreateJob( + "block"); + + tracker.Register( + job); + + await worker.StartAsync( + TestContext.Current + .CancellationToken); + + await queue.EnqueueAsync( + job, + TestContext.Current + .CancellationToken); + + await WaitForStateAsync( + tracker, + job.JobId, + JobState.Running, + TestContext.Current + .CancellationToken); + + using var stop = + new CancellationTokenSource( + TimeSpan.FromSeconds( + 2)); + + await worker.StopAsync( + stop.Token); + + JobSnapshot snapshot = + await tracker + .WaitForTerminalAsync( + job.JobId, + TestContext + .Current + .CancellationToken); + + Assert.Equal( + JobState.Canceled, + snapshot.State); + + Assert.False( + queue.IsAccepting); + } + + [Fact] + public async Task + Api_accepts_job_and_reports_terminal_status() + { + await using var factory = + new WebApplicationFactory< + Program>(); + + HttpClient client = + factory.CreateClient(); + + HttpResponseMessage response = + await client.PostAsJsonAsync( + "/jobs/email", + new EmailJobRequest( + "reader@example.com", + "Queue sample"), + TestContext.Current + .CancellationToken); + + Assert.Equal( + HttpStatusCode.Accepted, + response.StatusCode); + + Assert.NotNull( + response.Headers.Location); + + AcceptedJob? accepted = + await response.Content + .ReadFromJsonAsync< + AcceptedJob>( + TestContext.Current + .CancellationToken); + + Assert.NotNull( + accepted); + + Assert.NotEqual( + Guid.Empty, + accepted.JobId); + + JobSnapshot snapshot = + await PollTerminalStatusAsync( + client, + accepted.JobId, + TestContext.Current + .CancellationToken); + + Assert.Equal( + JobState.Succeeded, + snapshot.State); + + Assert.Null( + snapshot.FailureMessage); + } + + [Fact] + public async Task + Api_rejects_invalid_request_and_unknown_job() + { + await using var factory = + new WebApplicationFactory< + Program>(); + + HttpClient client = + factory.CreateClient(); + + HttpResponseMessage invalid = + await client.PostAsJsonAsync( + "/jobs/email", + new EmailJobRequest( + "", + ""), + TestContext.Current + .CancellationToken); + + Assert.Equal( + HttpStatusCode.BadRequest, + invalid.StatusCode); + + HttpResponseMessage missing = + await client.GetAsync( + $"/jobs/{Guid.NewGuid()}", + TestContext.Current + .CancellationToken); + + Assert.Equal( + HttpStatusCode.NotFound, + missing.StatusCode); + + QueueSnapshot? queue = + await client + .GetFromJsonAsync< + QueueSnapshot>( + "/jobs/queue", + TestContext.Current + .CancellationToken); + + Assert.NotNull( + queue); + + Assert.True( + queue.Accepting); + + Assert.Equal( + 0, + queue.Depth); + } + + private static + BoundedBackgroundJobQueue + CreateQueue( + int capacity) => + new( + Options.Create( + new + BackgroundJobQueueOptions + { + Capacity = + capacity + })); + + private static EmailJob CreateJob( + string subject) => + new( + JobId: + Guid.NewGuid(), + + To: + "reader@example.com", + + Subject: + subject, + + EnqueuedAt: + DateTimeOffset + .UnixEpoch); + + private static async Task + WaitForStateAsync( + IJobTracker tracker, + Guid jobId, + JobState expected, + CancellationToken + cancellationToken) + { + while (true) + { + cancellationToken + .ThrowIfCancellationRequested(); + + if (tracker.TryGet( + jobId, + out JobSnapshot? + snapshot) + && snapshot! + .State + == expected) + { + return; + } + + await Task.Delay( + 10, + cancellationToken); + } + } + + private static async Task< + JobSnapshot> + PollTerminalStatusAsync( + HttpClient client, + Guid jobId, + CancellationToken + cancellationToken) + { + while (true) + { + JobSnapshot? snapshot = + await client + .GetFromJsonAsync< + JobSnapshot>( + $"/jobs/{jobId}", + cancellationToken); + + Assert.NotNull( + snapshot); + + if (snapshot.State + is JobState.Succeeded + or JobState.Failed + or JobState.Canceled) + { + return snapshot; + } + + await Task.Delay( + 10, + cancellationToken); + } + } + + private sealed record AcceptedJob( + Guid JobId, + string State); + + private sealed class HandlerRecorder + { + public ConcurrentBag + ScopeIds + { + get; + } = new(); + } + + private sealed class ScopedMarker + { + public Guid Id + { + get; + } = + Guid.NewGuid(); + } + + private sealed class + RecordingEmailJobHandler( + ScopedMarker marker, + HandlerRecorder recorder) : + IEmailJobHandler + { + public Task HandleAsync( + EmailJob job, + CancellationToken + cancellationToken) + { + cancellationToken + .ThrowIfCancellationRequested(); + + recorder.ScopeIds.Add( + marker.Id); + + if (job.Subject + == "fail") + { + throw new + InvalidOperationException( + "Deterministic test failure."); + } + + return Task.CompletedTask; + } + } + + private sealed class + BlockingEmailJobHandler : + IEmailJobHandler + { + public Task HandleAsync( + EmailJob job, + CancellationToken + cancellationToken) => + Task.Delay( + Timeout.InfiniteTimeSpan, + cancellationToken); + } +} \ No newline at end of file