Round #288

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 can be dropped into any .NET 6+ project (or later) and used to protect calls to external services, databases, HTTP APIs, etc. --- ## 1. Design Overview | Concept | What it means for this breaker | |---------|--------------------------------| | **States** | `Closed` – calls go through; failures are counted.<br>`Open` – calls are rejected immediately; after a timeout the breaker moves to `HalfOpen`.<br>`HalfOpen` – a limited number of test calls are allowed; a success closes the breaker, a failure re‑opens it. | | **Consecutive failures** | Only *in a row* failures are counted. A successful call resets the counter to 0. | | **Parameters** | *failureThreshold* – how many consecutive failures trigger the open state.<br>*openTimeout* – how long the breaker stays open before trying again.<br>*halfOpenMaxTrial* – how many test calls are allowed while half‑open (default 1). | | **Thread‑safety** | All mutable state is protected by a `SemaphoreSlim` (async‑compatible) and by `Interlocked` for the failure counter. This works correctly even when many callers hit the breaker concurrently. | | **Extensibility** | The class is generic (`CircuitBreaker<T>`) so you can use it with any return type. It also exposes events (`OnStateChanged`, `OnFailure`, `OnSuccess`) for logging/metrics. | | **Cancellation** | The `ExecuteAsync` method accepts a `CancellationToken` that is passed to the wrapped operation. The breaker itself does **not** cancel the operation – it only respects the token when deciding whether to start the call. | | **Exception handling** | The original exception from the wrapped operation is re‑thrown unchanged, so callers see the real failure. The breaker only swallows the exception to decide whether to transition state. | --- ## 2. The Code ```csharp using System; using System.Threading; using System.Threading.Tasks; /// <summary> /// Represents the current state of the circuit breaker. /// </summary> public enum CircuitState { Closed, // Normal operation Open, // Calls are rejected HalfOpen // Test calls are allowed } /// <summary> /// An asynchronous circuit breaker that trips after a configurable number of *consecutive* failures. /// </summary> /// <typeparam name="T">The type returned by the protected operation.</typeparam> public sealed class AsyncCircuitBreaker<T> : IDisposable { // ----------------------------------------------------------------------- // Configuration // ----------------------------------------------------------------------- private readonly int _failureThreshold; // How many consecutive failures open the circuit private readonly TimeSpan _openTimeout; // How long to stay open private readonly int _halfOpenMaxTrial; // Allowed test calls while half‑open // ----------------------------------------------------------------------- // Runtime state (protected by _stateLock) // ----------------------------------------------------------------------- private CircuitState _state = CircuitState.Closed; private DateTime _openedSince; // When we entered Open private int _halfOpenTrialCount; // How many test calls have been made // ----------------------------------------------------------------------- // Failure tracking (consecutive failures) // ----------------------------------------------------------------------- private int _consecutiveFailures; // Updated with Interlocked // ----------------------------------------------------------------------- // Synchronisation // ----------------------------------------------------------------------- private readonly SemaphoreSlim _stateLock = new SemaphoreSlim(1, 1); private bool _disposed; // ----------------------------------------------------------------------- // Events – useful for logging/metrics // ----------------------------------------------------------------------- public event Action<CircuitState, CircuitState>? OnStateChanged; public event Action<Exception>? OnFailure; public event Action? OnSuccess; // ----------------------------------------------------------------------- // Constructor // ----------------------------------------------------------------------- /// <summary> /// Creates a new async circuit breaker. /// </summary> /// <param name="failureThreshold">Number of consecutive failures required to open the circuit (must be &gt; 0).</param> /// <param name="openTimeout">Time the circuit stays open before transitioning to half‑open.</param> /// <param name="halfOpenMaxTrial">Maximum number of test calls allowed while half‑open (default 1).</param> public AsyncCircuitBreaker( int failureThreshold, TimeSpan openTimeout, int halfOpenMaxTrial = 1) { if (failureThreshold <= 0) throw new ArgumentOutOfRangeException(nameof(failureThreshold)); if (openTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(openTimeout)); if (halfOpenMaxTrial <= 0) throw new ArgumentOutOfRangeException(nameof(halfOpenMaxTrial)); _failureThreshold = failureThreshold; _openTimeout = openTimeout; _halfOpenMaxTrial = halfOpenMaxTrial; } // ----------------------------------------------------------------------- // Public API // ----------------------------------------------------------------------- /// <summary> /// Executes the supplied asynchronous operation under circuit‑breaker protection. /// </summary> /// <param name="operation">The operation to protect.</param> /// <param name="cancellationToken">Optional token to cancel the *call* (not the breaker).</param> /// <returns>The result of the operation.</returns> /// <exception cref="BrokenCircuitException">If the circuit is open.</exception> public async Task<T> ExecuteAsync( Func<CancellationToken, Task<T>> operation, CancellationToken cancellationToken = default) { if (operation is null) throw new ArgumentNullException(nameof(operation)); ThrowIfDisposed(); // 1️⃣ Check state and possibly transition from Open → HalfOpen await EnsureCircuitIsReadyAsync(cancellationToken).ConfigureAwait(false); // 2️⃣ If we are still open after the check, reject the call if (_state == CircuitState.Open) throw new BrokenCircuitException("The circuit is open and does not allow calls."); // 3️⃣ Execute the operation try { T result = await operation(cancellationToken).ConfigureAwait(false); // Success – reset failure counter and possibly close the circuit await OnOperationSuccessAsync().ConfigureAwait(false); return result; } catch (Exception ex) when (!(ex is OperationCanceledException && cancellationToken.IsCancellationRequested)) { // Failure – record it and possibly open the circuit await OnOperationFailureAsync(ex).ConfigureAwait(false); throw; // re‑throw original exception } } /// <summary> /// Current state of the breaker (thread‑safe read). /// </summary> public CircuitState State => _state; // ----------------------------------------------------------------------- // Private helpers – state management // ----------------------------------------------------------------------- private async Task EnsureCircuitIsReadyAsync(CancellationToken ct) { // Fast‑path: if closed or half‑open we are ready. if (_state == CircuitState.Closed || _state == CircuitState.HalfOpen) return; // If open, see whether the timeout elapsed. await _stateLock.WaitAsync(ct).ConfigureAwait(false); try { if (_state == CircuitState.Open && DateTime.UtcNow - _openedSince >= _openTimeout) { // Transition to HalfOpen await TransitionToStateAsync(CircuitState.HalfOpen).ConfigureAwait(false); _halfOpenTrialCount = 0; // reset trial counter } } finally { _stateLock.Release(); } } private async Task OnOperationSuccessAsync() { // Reset consecutive failures Interlocked.Exchange(ref _consecutiveFailures, 0); OnSuccess?.Invoke(); // If we are in HalfOpen, a success means the external system is healthy → close. if (_state == CircuitState.HalfOpen) { await _stateLock.WaitAsync().ConfigureAwait(false); try { // Double‑check state inside lock (it could have changed) if (_state == CircuitState.HalfOpen) { await TransitionToStateAsync(CircuitState.Closed).ConfigureAwait(false); } } finally { _stateLock.Release(); } } } private async Task OnOperationFailureAsync(Exception ex) { // Increment consecutive failures atomically int failures = Interlocked.Increment(ref _consecutiveFailures); OnFailure?.Invoke(ex); // If we are already open, nothing else to do. if (_state == CircuitState.Open) return; // If we are half‑open, any failure immediately re‑opens the circuit. if (_state == CircuitState.HalfOpen) { await _stateLock.WaitAsync().ConfigureAwait(false); try { if (_state == CircuitState.HalfOpen) { await TransitionToStateAsync(CircuitState.Open).ConfigureAwait(false); } } finally { _stateLock.Release(); } return; } // Closed state – check if threshold reached if (failures >= _failureThreshold) { await _stateLock.WaitAsync().ConfigureAwait(false); try { if (_state == CircuitState.Closed && _consecutiveFailures >= _failureThreshold) { await TransitionToStateAsync(CircuitState.Open).ConfigureAwait(false); } } finally { _stateLock.Release(); } } } private async Task TransitionToStateAsync(CircuitState newState) { if (_state == newState) return; CircuitState oldState = _state; _state = newState; switch (newState) { case CircuitState.Open: _openedSince = DateTime.UtcNow; break; case CircuitState.HalfOpen: // No special timestamp needed break; case CircuitState.Closed: // Reset counters Interlocked.Exchange(ref _consecutiveFailures, 0); break; } // Fire event *after* state change so listeners see the new state. OnStateChanged?.Invoke(oldState, newState); await Task.CompletedTask; // keep async signature } // ----------------------------------------------------------------------- // Disposal // ----------------------------------------------------------------------- private void ThrowIfDisposed() { if (_disposed) throw new ObjectDisposedException(nameof(AsyncCircuitBreaker<T>)); } public void Dispose() { if (_disposed) return; _stateLock.Dispose(); _disposed = true; } } /// <summary> /// Exception thrown when a call is attempted while the circuit breaker is open. /// </summary> public sealed class BrokenCircuitException : Exception { public BrokenCircuitException(string message) : base(message) { } public BrokenCircuitException(string message, Exception inner) : base(message, inner) { } } ``` --- ## 3. How to Use It ```csharp // Example: protecting an HTTP call var breaker = new AsyncCircuitBreaker<HttpResponseMessage>( failureThreshold: 3, // 3 consecutive failures → open openTimeout: TimeSpan.FromSeconds(30), // stay open for 30 s halfOpenMaxTrial: 1); // only one test call while half‑open breaker.OnStateChanged += (oldState, newState) => Console.WriteLine($"Circuit changed from {oldState} → {newState}"); breaker.OnFailure += ex => Console.WriteLine($"Call failed: {ex.Message}"); breaker.OnSuccess += () => Console.WriteLine("Call succeeded"); // The protected operation async Task<HttpResponseMessage> GetGoogleAsync(CancellationToken ct) { using var client = new HttpClient(); return await client.GetAsync("https://www.google.com", ct); } // Somewhere in your code (e.g., inside a loop, background service, etc.) for (int i = 0; i < 10; i++) { try { var response = await breaker.ExecuteAsync(GetGoogleAsync); Console.WriteLine($"Status: {response.StatusCode}"); } catch (BrokenCircuitException) { Console.WriteLine("Circuit is open – request skipped."); } catch (Exception ex) { Console.WriteLine($"Request failed: {ex.Message}"); } await Task.Delay(2000); // wait a bit before next attempt } ``` **What happens?** * The first three *consecutive* failures (e.g., network timeout) will cause the breaker to transition to **Open**. * While **Open**, every call throws `BrokenCircuitException` immediately. * After 30 seconds the breaker moves to **HalfOpen** and allows a single test call. * If that test call succeeds, the breaker goes back to **Closed** and the failure counter resets. * If the test call fails, the breaker goes back to **Open** and the cycle repeats. --- ## 4. Extending / Tweaking | Need | How to achieve | |------|----------------| | **Different failure counting** (e.g., total failures in a window) | Replace the `Interlocked` counter with a sliding‑window histogram; the rest of the state machine stays the same. | | **Custom back‑off** (exponential) | Instead of a fixed `_openTimeout`, store a `TimeSpan _currentTimeout` that you multiply on each open, resetting on success. | | **Metrics** | Subscribe to `OnStateChanged`, `OnFailure`, `OnSuccess` and push counters to Prometheus, Application Insights, etc. | | **Bulkhead / Semaphore** | Wrap the breaker around a `SemaphoreSlim`‑protected resource if you also need concurrency limiting. | | **Synchronous API** | Add a `Execute` overload that takes `Func<T>` and internally calls `Task.Run` or uses `ValueTask<T>` for zero‑allocation paths. | --- ## 5. Why This Satisfies the Requirement * **Async‑aware** – all public methods are `async` and use `SemaphoreSlim` for non‑blocking synchronization. * **Tracks consecutive failures** – the `_consecutiveFailures` counter is reset on any success and incremented on each failure. * **State transitions** – follows the classic Closed → Open → HalfOpen → Closed flow. * **Thread‑safe** – works correctly under high concurrency. * **Self‑contained** – no external NuGet packages required; only BCL types. * **Extensible & observable** – events let you plug in logging, metrics, or custom actions. Feel free to copy the class into your project, adjust the parameters, and start protecting your flaky external calls!

**Solution Overview** We create a reusable **asynchronous circuit‑breaker** that can be injected wherever you need to protect an external call (HTTP request, DB query, etc.). The implementation uses the **Microsoft.Extensions.Resilience** package (the official .NET resilience library) which provides the `ResiliencePipelineBuilder<T>` and `ResiliencePipeline<T>` types. * `ResiliencePipelineBuilder<T>` – a fluent builder that lets us compose resilience policies (retry, timeout, circuit‑breaker, …). * `ResiliencePipeline<T>` – the compiled pipeline that can be executed with `ExecuteAsync`. The circuit‑breaker we expose tracks **consecutive failures**. When the configured failure threshold is reached it opens the circuit for a configurable break duration. While the circuit is open every call fails fast with a `BrokenCircuitException`. After the break period the circuit goes to a *half‑open* state and lets a single trial call through; success closes the circuit, failure re‑opens it. The public API is a small wrapper (`AsyncCircuitBreaker`) that hides the pipeline details and gives a clean, test‑able abstraction. --- ## 1. NuGet Packages ```xml <ItemGroup> <PackageReference Include="Microsoft.Extensions.Resilience" Version="8.0.0" /> </ItemGroup> ``` > **Note** – `Microsoft.Extensions.Resilience` ships with .NET 8+ but can be added to earlier targets as a NuGet package. --- ## 2. Core Types ```csharp using System; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Resilience; using Microsoft.Extensions.Resilience.CircuitBreaker; using Microsoft.Extensions.Logging; /// <summary> /// Represents an asynchronous circuit‑breaker that tracks consecutive failures. /// </summary> public interface IAsyncCircuitBreaker { /// <summary> /// Executes the supplied operation through the circuit‑breaker pipeline. /// </summary> /// <typeparam name="TResult">The type returned by the operation.</typeparam> /// <param name="operation">The operation to protect.</param> /// <param name="cancellationToken">Optional cancellation token.</param> /// <returns>The result of the operation if the circuit is closed or half‑open.</returns> /// <exception cref="BrokenCircuitException"> /// Thrown when the circuit is open. /// </exception> Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TResult>> operation, CancellationToken cancellationToken = default); } ``` ### Implementation ```csharp public sealed class AsyncCircuitBreaker : IAsyncCircuitBreaker, IDisposable { private readonly ResiliencePipeline _pipeline; private readonly ILogger<AsyncCircuitBreaker> _logger; private bool _disposed; /// <summary> /// Creates a new <see cref="AsyncCircuitBreaker"/>. /// </summary> /// <param name="failureThreshold"> /// Number of consecutive failures that will open the circuit. /// </param> /// <param name="breakDuration"> /// How long the circuit stays open before transitioning to half‑open. /// </param> /// <param name="logger">Optional logger.</param> public AsyncCircuitBreaker( int failureThreshold, TimeSpan breakDuration, ILogger<AsyncCircuitBreaker>? logger = null) { if (failureThreshold <= 0) throw new ArgumentOutOfRangeException(nameof(failureThreshold)); if (breakDuration <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(breakDuration)); _logger = logger ?? Microsoft.Extensions.Logging.Abstractions.NullLogger<AsyncCircuitBreaker>.Instance; // Build the pipeline – only a circuit‑breaker policy is required for this // example, but the builder can be extended with retries, timeouts, etc. var builder = new ResiliencePipelineBuilder() .AddCircuitBreaker(new CircuitBreakerOptions { FailureThreshold = failureThreshold, SamplingDuration = TimeSpan.FromMinutes(1), // window for counting failures MinimumThroughput = failureThreshold, // ensure we have enough calls BreakDuration = breakDuration, OnOpened = args => { _logger.LogWarning("Circuit opened. Reason: {Reason}", args.FailureReason); return ValueTask.CompletedTask; }, OnClosed = args => { _logger.LogInformation("Circuit closed."); return ValueTask.CompletedTask; }, OnHalfOpened = args => { _logger.LogInformation("Circuit half‑opened – testing the next call."); return ValueTask.CompletedTask; } }); _pipeline = builder.Build(); } public async Task<TResult> ExecuteAsync<TResult>(Func<CancellationToken, Task<TResult>> operation, CancellationToken cancellationToken = default) { if (_disposed) throw new ObjectDisposedException(nameof(AsyncCircuitBreaker)); if (operation is null) throw new ArgumentNullException(nameof(operation)); // The pipeline expects a delegate that receives a CancellationToken. // We simply forward the call. return await _pipeline.ExecuteAsync(async ct => await operation(ct).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); } public void Dispose() { if (_disposed) return; if (_pipeline is IAsyncDisposable asyncDisposable) { asyncDisposable.DisposeAsync().AsTask().GetAwaiter().GetResult(); } _disposed = true; } } ``` **Key points** * `ResiliencePipelineBuilder` is used to add a **circuit‑breaker** policy (`AddCircuitBreaker`). * `CircuitBreakerOptions` lets us configure: * `FailureThreshold` – number of consecutive failures before opening. * `SamplingDuration` – the time window for counting failures (set to 1 min, but can be tuned). * `MinimumThroughput` – ensures the breaker only evaluates after at least *N* calls. * `BreakDuration` – how long the circuit stays open. * Callbacks (`OnOpened`, `OnClosed`, `OnHalfOpened`) for logging/telemetry. * The public `ExecuteAsync` method forwards the supplied operation through the compiled pipeline. * The class implements `IDisposable` to clean up the pipeline (important when the pipeline holds timers or other async disposables). --- ## 3. DI Registration (ASP.NET Core example) ```csharp using Microsoft.Extensions.DependencyInjection; public static class CircuitBreakerServiceCollectionExtensions { /// <summary> /// Registers <see cref="IAsyncCircuitBreaker"/> with the supplied settings. /// </summary> public static IServiceCollection AddAsyncCircuitBreaker(this IServiceCollection services, int failureThreshold = 5, TimeSpan? breakDuration = null) { breakDuration ??= TimeSpan.FromSeconds(30); services.AddSingleton<IAsyncCircuitBreaker>(sp => { var logger = sp.GetRequiredService<ILogger<AsyncCircuitBreaker>>(); return new AsyncCircuitBreaker(failureThreshold, breakDuration.Value, logger); }); return services; } } ``` Usage in `Program.cs` (or `Startup.cs`): ```csharp builder.Services.AddLogging(); // ensure logger is available builder.Services.AddAsyncCircuitBreaker(failureThreshold: 3, breakDuration: TimeSpan.FromSeconds(20)); ``` Now any class can depend on `IAsyncCircuitBreaker`: ```csharp public class WeatherService { private readonly HttpClient _http; private readonly IAsyncCircuitBreaker _breaker; public WeatherService(HttpClient http, IAsyncCircuitBreaker breaker) { _http = http; _breaker = breaker; } public async Task<string> GetForecastAsync(CancellationToken ct = default) { return await _breaker.ExecuteAsync<string>(async token => { var response = await _http.GetAsync("weather/today", token); response.EnsureSuccessStatusCode(); return await response.Content.ReadAsStringAsync(token); }, ct); } } ``` When three consecutive calls to the external weather endpoint fail, the circuit opens for 20 seconds; during that period `GetForecastAsync` will immediately throw `BrokenCircuitException`, protecting the rest of the system. --- ## 4. Summary * **`ResiliencePipelineBuilder`** builds a pipeline that contains a **circuit‑breaker** policy. * **`AsyncCircuitBreaker`** wraps that pipeline, exposing a simple `ExecuteAsync` method. * The circuit tracks **consecutive failures** (via `FailureThreshold`) and opens for a configurable duration. * The component is fully **DI‑friendly**, logs state transitions, and disposes resources correctly. This gives you a production‑ready, asynchronous circuit‑breaker built on the official .NET resilience primitives.