Round #313

Implement an async circuit breaker that tracks consecutive failures

Below is a **complete, production‑ready implementation** of an **asynchronous circuit‑breaker** that tracks *consecutive* failures. It is written for .NET 6+ (but works on any recent .NET version) and does **not** depend on any external NuGet packages – only the BCL. --- ## 1. Design Overview | State | Meaning | Transition | |-------|---------|------------| | **Closed** | Calls are allowed. The breaker counts consecutive failures. When the count reaches `failureThreshold` the breaker **opens**. | `Closed → Open` when `consecutiveFailures >= failureThreshold` | | **Open** | Calls are rejected immediately (or wait for a fallback). After `openTimeout` the breaker moves to **HalfOpen**. | `Open → HalfOpen` after `openTimeout` expires | | **HalfOpen** | A limited number of “test” calls are allowed (by default 1). If a test call succeeds the breaker **closes**; if it fails the breaker **re‑opens**. | `HalfOpen → Closed` on first success, `HalfOpen → Open` on first failure | *Consecutive* failures are tracked – a successful call resets the counter to zero. The breaker is **thread‑safe** and fully asynchronous (`Task<T>` / `Task`). It also supports an optional **fallback** delegate that is invoked when the circuit is open. --- ## 2. Public API ```csharp public class AsyncCircuitBreaker { // Construction public AsyncCircuitBreaker( int failureThreshold, TimeSpan openTimeout, int halfOpenSuccessThreshold = 1, Action? onOpen = null, Action? onClose = null, Action? onHalfOpen = null); // Execute a delegate under the protection of the breaker public Task<T> ExecuteAsync<T>(Func<CancellationToken, Task<T>> action, CancellationToken ct = default, Func<CancellationToken, Task<T>>? fallback = null); public Task ExecuteAsync(Func<CancellationToken, Task> action, CancellationToken ct = default, Func<CancellationToken, Task>? fallback = null); // Current state (read‑only) public CircuitState State { get; } // Statistics (optional, for monitoring) public int ConsecutiveFailureCount { get; } public DateTimeOffset? OpenedAt { get; } } ``` `CircuitState` is an enum (`Closed`, `Open`, `HalfOpen`). --- ## 3. Full Implementation ```csharp using System; using System.Threading; using System.Threading.Tasks; /// <summary> /// Represents the three possible states of a circuit breaker. /// </summary> public enum CircuitState { Closed, // normal operation Open, // short‑circuit, reject calls HalfOpen // test period } /// <summary> /// An asynchronous circuit breaker that tracks *consecutive* failures. /// </summary> public sealed class AsyncCircuitBreaker { // ----- Configuration ---------------------------------------------------- private readonly int _failureThreshold; // how many consecutive failures open the circuit private readonly TimeSpan _openTimeout; // how long to stay open before trying half‑open private readonly int _halfOpenSuccessThreshold; // how many successful calls are needed to close again (usually 1) private readonly Action? _onOpen; private readonly Action? _onClose; private readonly Action? _onHalfOpen; // ----- Runtime state (protected by _sync) ------------------------------- private readonly SemaphoreSlim _sync = new(1, 1); // async lock private CircuitState _state = CircuitState.Closed; private int _consecutiveFailures; private DateTimeOffset? _openedAt; // when we entered Open state private int _halfOpenSuccesses; // successes observed while in HalfOpen // ----------------------------------------------------------------------- public AsyncCircuitBreaker( int failureThreshold, TimeSpan openTimeout, int halfOpenSuccessThreshold = 1, Action? onOpen = null, Action? onClose = null, Action? onHalfOpen = null) { if (failureThreshold <= 0) throw new ArgumentOutOfRangeException(nameof(failureThreshold)); if (openTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(openTimeout)); if (halfOpenSuccessThreshold <= 0) throw new ArgumentOutOfRangeException(nameof(halfOpenSuccessThreshold)); _failureThreshold = failureThreshold; _openTimeout = openTimeout; _halfOpenSuccessThreshold = halfOpenSuccessThreshold; _onOpen = onOpen; _onClose = onClose; _onHalfOpen = onHalfOpen; } // ----------------------------------------------------------------------- /// <summary> /// Current state (read‑only, thread‑safe). /// </summary> public CircuitState State { get { // No lock needed – reading a reference‑type enum is atomic. return _state; } } /// <summary> /// Number of consecutive failures observed (only meaningful in Closed state). /// </summary> public int ConsecutiveFailureCount => Volatile.Read(ref _consecutiveFailures); /// <summary> /// Timestamp when the circuit entered the Open state, or null if not open. /// </summary> public DateTimeOffset? OpenedAt => _openedAt; // ----------------------------------------------------------------------- // Public entry point for functions returning a value public async Task<T> ExecuteAsync<T>( Func<CancellationToken, Task<T>> action, CancellationToken ct = default, Func<CancellationToken, Task<T>>? fallback = null) { if (action is null) throw new ArgumentNullException(nameof(action)); // 1️⃣ Check circuit state and possibly short‑circuit if (!await IsCallAllowedAsync(ct).ConfigureAwait(false)) { if (fallback != null) return await fallback(ct).ConfigureAwait(false); throw new CircuitBreakerOpenException(); } // 2️⃣ Execute the real work try { T result = await action(ct).ConfigureAwait(false); await OnSuccessAsync().ConfigureAwait(false); return result; } catch (Exception ex) when (!(ex is OperationCanceledException && ct.IsCancellationRequested)) { await OnFailureAsync().ConfigureAwait(false); throw; } } // Overload for void‑returning async methods public async Task ExecuteAsync( Func<CancellationToken, Task> action, CancellationToken ct = default, Func<CancellationToken, Task>? fallback = null) { if (action is null) throw new ArgumentNullException(nameof(action)); if (!await IsCallAllowedAsync(ct).ConfigureAwait(false)) { if (fallback != null) { await fallback(ct).ConfigureAwait(false); return; } throw new CircuitBreakerOpenException(); } try { await action(ct).ConfigureAwait(false); await OnSuccessAsync().ConfigureAwait(false); } catch (Exception ex) when (!(ex is OperationCanceledException && ct.IsCancellationRequested)) { await OnFailureAsync().ConfigureAwait(false); throw; } } // ----------------------------------------------------------------------- // Determines whether a call should be allowed or short‑circuited. private async Task<bool> IsCallAllowedAsync(CancellationToken ct) { // Fast‑path: if we are Closed we can always proceed. if (_state == CircuitState.Closed) return true; // If we are Open we need to see whether the timeout elapsed. if (_state == CircuitState.Open) { // No lock yet – just read the timestamp. var opened = _openedAt; if (opened.HasValue && DateTimeOffset.UtcNow - opened.Value >= _openTimeout) { // Timeout elapsed → try to move to HalfOpen. await TransitionToHalfOpenAsync(ct).ConfigureAwait(false); return _state == CircuitState.HalfOpen; // should be true } // Still open → reject. return false; } // HalfOpen – allow a limited number of test calls. // We allow the call; the success/failure logic will decide the next state. return true; } // ----------------------------------------------------------------------- // Called when the protected operation succeeded. private async Task OnSuccessAsync() { // Acquire lock because we modify shared state. await _sync.WaitAsync().ConfigureAwait(false); try { switch (_state) { case CircuitState.Closed: // Reset consecutive failures on any success. _consecutiveFailures = 0; break; case CircuitState.HalfOpen: _halfOpenSuccesses++; if (_halfOpenSuccesses >= _halfOpenSuccessThreshold) { // Enough successes → close the circuit. await TransitionToClosedAsync().ConfigureAwait(false); } break; case CircuitState.Open: // Should never happen because calls are blocked while Open. break; } } finally { _sync.Release(); } } // ----------------------------------------------------------------------- // Called when the protected operation threw. private async Task OnFailureAsync() { await _sync.WaitAsync().ConfigureAwait(false); try { switch (_state) { case CircuitState.Closed: _consecutiveFailures++; if (_consecutiveFailures >= _failureThreshold) { await TransitionToOpenAsync().ConfigureAwait(false); } break; case CircuitState.HalfOpen: // Any failure while half‑open forces an immediate re‑open. await TransitionToOpenAsync().ConfigureAwait(false); break; case CircuitState.Open: // No action – we are already open. break; } } finally { _sync.Release(); } } // ----------------------------------------------------------------------- // State transition helpers (all called while holding _sync, except the // Open→HalfOpen check which acquires the lock itself). private async Task TransitionToOpenAsync() { // Assume lock already held. _state = CircuitState.Open; _openedAt = DateTimeOffset.UtcNow; _consecutiveFailures = 0; // reset counter _halfOpenSuccesses = 0; _onOpen?.Invoke(); await Task.CompletedTask; } private async Task TransitionToHalfOpenAsync(CancellationToken ct) { // Ensure only one thread performs the transition. await _sync.WaitAsync(ct).ConfigureAwait(false); try { if (_state != CircuitState.Open) // another thread may have already changed it return; _state = CircuitState.HalfOpen; _halfOpenSuccesses = 0; _openedAt = null; // no longer "open" _onHalfOpen?.Invoke(); } finally { _sync.Release(); } } private async Task TransitionToClosedAsync() { // Assume lock already held. _state = CircuitState.Closed; _consecutiveFailures = 0; _halfOpenSuccesses = 0; _openedAt = null; _onClose?.Invoke(); await Task.CompletedTask; } } // --------------------------------------------------------------------------- // Custom exception – thrown when a call is attempted while the circuit is open. public sealed class CircuitBreakerOpenException : Exception { public CircuitBreakerOpenException() : base("The circuit breaker is open; calls are not allowed.") { } } ``` ### How It Works 1. **Entry point** (`ExecuteAsync`) first checks the current state via `IsCallAllowedAsync`. * If **Closed** → proceed. * If **Open** → if the timeout has elapsed, atomically move to **HalfOpen**; otherwise reject (or run fallback). * If **HalfOpen** → allow the call (the success/failure logic will decide the next state). 2. **After the delegate finishes** * On success → `OnSuccessAsync` resets the failure counter (Closed) or, in HalfOpen, counts successes and may close the circuit. * On failure → `OnFailureAsync` increments the consecutive‑failure counter (Closed) and may open the circuit; any failure in HalfOpen immediately re‑opens it. 3. **State transitions** are performed under a `SemaphoreSlim` lock (`_sync`) to guarantee thread‑safety without blocking threads (async‑compatible). 4. **Fallback** – optional `fallback` delegate is executed when the circuit is open. If no fallback is supplied, a `CircuitBreakerOpenException` is thrown. --- ## 4. Usage Example ```csharp // 1️⃣ Create a breaker: 3 consecutive failures → open for 10 seconds. var breaker = new AsyncCircuitBreaker( failureThreshold: 3, openTimeout: TimeSpan.FromSeconds(10), halfOpenSuccessThreshold: 1, onOpen: () => Console.WriteLine("Circuit opened!"), onClose: () => Console.WriteLine("Circuit closed."), onHalfOpen: () => Console.WriteLine("Circuit half‑open, testing...")); // 2️⃣ The operation we want to protect async Task<string> UnreliableCall(CancellationToken ct) { // Simulate flaky external service await Task.Delay(100, ct); if (Random.Shared.NextDouble() < 0.7) // 70 % chance to fail throw new InvalidOperationException("Remote service failed"); return "Success!"; } // 3️⃣ Optional fallback Task<string> Fallback(CancellationToken ct) => Task.FromResult("fallback result"); // 4️⃣ Run many calls for (int i = 0; i < 20; i++) { try { var result = await breaker.ExecuteAsync(UnreliableCall, fallback: Fallback); Console.WriteLine($"Call {i,2}: {result} (state={breaker.State})"); } catch (CircuitBreakerOpenException) { Console.WriteLine($"Call {i,2}: blocked (circuit open)"); } await Task.Delay(500); // pause a bit between calls } ``` **What you’ll see** * After three consecutive failures the circuit prints “Circuit opened!” and subsequent calls are blocked (or use the fallback). * After 10 seconds the breaker moves to HalfOpen, prints “Circuit half‑open…”, lets one test call through. * If that test succeeds, you get “Circuit closed.” and normal operation resumes; otherwise it re‑opens. --- ## 5. Extending / Instrumenting * **Metrics** – expose `ConsecutiveFailureCount`, `OpenedAt`, or add `long TotalSuccesses/Failures`. * **Logging** – replace the `Action` callbacks with `ILogger` calls. * **Bulkhead** – combine with a semaphore/queue to limit concurrent calls. * **Policy composition** – you can wrap this breaker inside a retry policy, or vice‑versa. --- ### TL;DR *The code above gives you a fully asynchronous, thread‑safe circuit breaker that tracks **consecutive** failures, automatically moves between **Closed → Open → HalfOpen → Closed**, supports a timeout, optional fallback, and simple callbacks for observability.*

**Solution Overview** We will create a reusable **asynchronous circuit‑breaker** that can be injected wherever it is needed. The implementation uses the **Microsoft.Extensions.Resilience** package (the official .NET resilience library that ships with the `ResiliencePipeline` and `ResiliencePipelineBuilder` types). The pipeline is built with a **circuit‑breaker policy** that: * Counts consecutive failures (exceptions or results that are considered failures). * Trips the circuit after a configurable number of failures (`failureThreshold`). * Keeps the circuit open for a configurable `breakDuration`. * After the break period, it goes to a **half‑open** state and allows a single trial call. * If the trial succeeds the circuit closes, otherwise it re‑opens. The pipeline is exposed through an interface `ICircuitBreakerService` that hides the underlying `ResiliencePipeline`. Consumers only need to call `ExecuteAsync<T>`. The code includes: 1. **Package reference** – `Microsoft.Extensions.Resilience` (and its dependency `Microsoft.Extensions.DependencyInjection`). 2. **Configuration POCO** – `CircuitBreakerOptions`. 3. **Service interface** – `ICircuitBreakerService`. 4. **Implementation** – `CircuitBreakerService` that builds the pipeline with `ResiliencePipelineBuilder`. 5. **Extension method** – `AddCircuitBreaker` for clean DI registration. 6. **Example usage** – a console app that demonstrates the breaker in action. All code is fully async, thread‑safe and follows idiomatic .NET patterns. --- ## 1. Project file (csproj) ```xml <Project Sdk="Microsoft.NET.Sdk"> <PropertyGroup> <TargetFramework>net8.0</TargetFramework> <ImplicitUsings>enable</ImplicitUsings> <Nullable>enable</Nullable> </PropertyGroup> <ItemGroup> <!-- Resilience pipeline library --> <PackageReference Include="Microsoft.Extensions.Resilience" Version="8.0.0" /> <!-- For DI --> <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.0" /> </ItemGroup> </Project> ``` --- ## 2. Configuration POCO ```csharp namespace ResilienceDemo; /// <summary> /// Options that control the behaviour of the circuit‑breaker. /// </summary> public sealed class CircuitBreakerOptions { /// <summary> /// Number of consecutive failures required to open the circuit. /// </summary> public int FailureThreshold { get; set; } = 5; /// <summary> /// How long the circuit stays open before transitioning to half‑open. /// </summary> public TimeSpan BreakDuration { get; set; } = TimeSpan.FromSeconds(30); /// <summary> /// Optional: how many trial calls are allowed in half‑open state. /// The default (0) means a single trial call. /// </summary> public int? AllowedTrialCalls { get; set; } = null; } ``` --- ## 3. Service contract ```csharp using System.Threading.Tasks; namespace ResilienceDemo; /// <summary> /// Executes asynchronous operations through a circuit‑breaker pipeline. /// </summary> public interface ICircuitBreakerService { /// <summary> /// Executes the supplied <paramref name="operation"/> inside the circuit‑breaker. /// </summary> /// <typeparam name="TResult">Result type of the operation.</typeparam> /// <param name="operation">The async delegate to execute.</param> /// <returns>The result of the operation if the circuit is closed or the call succeeds.</returns> /// <exception cref="BrokenCircuitException"> /// Thrown when the circuit is open and the call is short‑circuited. /// </exception> Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TResult>> operation); } ``` --- ## 4. Implementation – `CircuitBreakerService` ```csharp using Microsoft.Extensions.Options; using Microsoft.Extensions.Resilience; using Microsoft.Extensions.Resilience.CircuitBreaker; using System; using System.Threading; using System.Threading.Tasks; namespace ResilienceDemo; /// <summary> /// Concrete implementation that builds a <see cref="ResiliencePipeline{TResult}"/> /// containing a circuit‑breaker policy. /// </summary> internal sealed class CircuitBreakerService : ICircuitBreakerService, IDisposable { private readonly ResiliencePipeline _pipeline; private readonly ResiliencePipelineBuilder _builder; private readonly CircuitBreakerOptions _options; private bool _disposed; public CircuitBreakerService(IOptions<CircuitBreakerOptions> options) { _options = options?.Value ?? throw new ArgumentNullException(nameof(options)); // Build a pipeline that works for any TResult (non‑generic pipeline) _builder = new ResiliencePipelineBuilder(); // Add the circuit‑breaker policy _builder.AddCircuitBreaker(new CircuitBreakerStrategyOptions { FailureRatio = 1.0, // we care about *consecutive* failures, not ratio SamplingDuration = TimeSpan.MaxValue, // effectively disables time‑window sampling MinimumThroughput = _options.FailureThreshold, BreakDuration = _options.BreakDuration, // The built‑in circuit‑breaker works on *any* exception. // To treat specific results as failures you could add a ResultPredicate. // Here we keep it simple and rely on exceptions. // AllowedTrialCalls is only available in .NET 9 preview; we emulate with // the default behaviour (single trial call) when null. }); // Build the pipeline (non‑generic, can be used for any TResult) _pipeline = _builder.Build(); } public async Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TResult>> operation) { if (_disposed) throw new ObjectDisposedException(nameof(CircuitBreakerService)); // The generic overload of ExecuteAsync creates a typed pipeline on‑the‑fly. // This avoids having to create a separate generic pipeline for each TResult. return await _pipeline.ExecuteAsync(operation); } public void Dispose() { if (_disposed) return; _pipeline.Dispose(); // disposes any inner resources _disposed = true; } } ``` **Key points** * `ResiliencePipelineBuilder` is used to add a **circuit‑breaker** strategy. * `FailureRatio = 1.0` together with `MinimumThroughput = FailureThreshold` makes the breaker trigger after *exactly* `FailureThreshold` consecutive failures. * `BreakDuration` controls how long the circuit stays open. * The pipeline is non‑generic; `ExecuteAsync<TResult>` creates a typed execution context internally, so the same pipeline can be reused for any result type. * The service implements `IDisposable` to clean up the pipeline. --- ## 5. DI registration extension ```csharp using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; namespace ResilienceDemo; /// <summary> /// Extension methods to register the circuit‑breaker service in an IServiceCollection. /// </summary> public static class CircuitBreakerServiceCollectionExtensions { /// <summary> /// Registers <see cref="ICircuitBreakerService"/> with the supplied options. /// </summary> /// <param name="services">The service collection.</param> /// <param name="configure">Configuration delegate for <see cref="CircuitBreakerOptions"/>.</param> /// <returns>The same service collection for chaining.</returns> public static IServiceCollection AddCircuitBreaker(this IServiceCollection services, Action<CircuitBreakerOptions> configure) { services.Configure(configure); services.AddSingleton<ICircuitBreakerService, CircuitBreakerService>(); return services; } } ``` --- ## 6. Example console program ```csharp using Microsoft.Extensions.DependencyInjection; using ResilienceDemo; using System; using System.Net.Http; using System.Threading; using System.Threading.Tasks; class Program { static async Task Main() { // ------------------------------------------------- // 1️⃣ Set up DI container // ------------------------------------------------- var services = new ServiceCollection(); services.AddHttpClient(); // for demo purposes services.AddCircuitBreaker(opts => { opts.FailureThreshold = 3; // trip after 3 failures opts.BreakDuration = TimeSpan.FromSeconds(10); }); var provider = services.BuildServiceProvider(); var breaker = provider.GetRequiredService<ICircuitBreakerService>(); var http = provider.GetRequiredService<IHttpClientFactory>().CreateClient(); // ------------------------------------------------- // 2️⃣ Simulate a flaky endpoint // ------------------------------------------------- var rnd = new Random(); async Task<string> UnreliableCall(CancellationToken ct) { // 50 % chance of failure if (rnd.NextDouble() < 0.5) throw new HttpRequestException("Simulated network failure"); // otherwise succeed await Task.Delay(100, ct); return "Success!"; } // ------------------------------------------------- // 3️⃣ Run a loop to see the breaker in action // ------------------------------------------------- for (int i = 1; i <= 12; i++) { try { var result = await breaker.ExecuteAsync<string>(UnreliableCall); Console.WriteLine($"{i,2}: Call succeeded – {result}"); } catch (BrokenCircuitException) { Console.WriteLine($"{i,2}: Call short‑circuited – circuit is OPEN"); } catch (Exception ex) { Console.WriteLine($"{i,2}: Call failed – {ex.GetType().Name}: {ex.Message}"); } await Task.Delay(500); // small pause between attempts } // Keep console open Console.WriteLine("Demo finished. Press any key..."); Console.ReadKey(); } } ``` **What the demo shows** * The first three failures cause the circuit to open. * Subsequent calls within the 10‑second break period immediately throw `BrokenCircuitException`. * After the break duration expires, the next call is allowed (half‑open). If it succeeds, the circuit closes; if it fails, the circuit re‑opens. --- ## 7. Summary * **`ResiliencePipelineBuilder`** is used to configure a circuit‑breaker that tracks *consecutive* failures. * The **`CircuitBreakerService`** wraps the pipeline behind a clean, async‑friendly interface. * **DI registration** (`AddCircuitBreaker`) makes the service easy to consume throughout an application. * The example console app demonstrates the behaviour and can be swapped for any real operation (database calls, HTTP requests, etc.). This implementation satisfies the request for an **asynchronous circuit breaker that tracks consecutive failures** while leveraging the modern .NET `ResiliencePipeline` infrastructure.