EonaCat.LogStack/EonaCat.LogStack/EonaCatLoggerCore/Flows/CircuitBreakerFlow.cs

using EonaCat.LogStack.Core;
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

namespace EonaCat.LogStack.Flows;

// This file is part of the EonaCat project(s) which is released under the Apache License.
// See the LICENSE file or go to https://EonaCat.com/License for full license details.

/// <summary>
/// Circuit-breaker state machine.
/// </summary>
public enum CircuitState
{
    /// <summary>Flow is healthy; all events are forwarded.</summary>
    Closed,
    /// <summary>Too many failures; events are dropped to protect the inner flow.</summary>
    Open,
    /// <summary>Recovery probe in progress; one event is let through per interval.</summary>
    HalfOpen
}

/// <summary>
/// A decorator flow that implements the Circuit Breaker pattern around any inner flow.
///
/// Transitions:
///   Closed  → Open     : after <see cref="FailureThreshold"/> consecutive failures
///   Open    → HalfOpen : after <see cref="RecoveryTimeout"/> has elapsed
///   HalfOpen→ Closed   : on a successful probe write
///   HalfOpen→ Open     : on a failed probe write
///
/// When Open, <see cref="BlastAsync"/> returns <see cref="WriteResult.Dropped"/> immediately
/// without touching the inner flow — preventing cascades into broken endpoints.
///
/// The <see cref="StateChanged"/> event fires on every state transition.
/// </summary>
public sealed class CircuitBreakerFlow : FlowBase
{
    private readonly IFlow _inner;
    private readonly int _failureThreshold;
    private readonly TimeSpan _recoveryTimeout;

    private volatile int _consecutiveFailures;
    private volatile CircuitState _state = CircuitState.Closed;

    // Stopwatch restarted every time we enter Open state
    private readonly Stopwatch _openedAt = new Stopwatch();

    private readonly object _stateLock = new object();

    /// <summary>Fires whenever the circuit state changes (from, to).</summary>
    public event Action<CircuitState, CircuitState> StateChanged;

    public CircuitBreakerFlow(
        IFlow inner,
        int failureThreshold = 5,
        TimeSpan? recoveryTimeout = null,
        LogLevel minimumLevel = LogLevel.Trace)
        : base($"CircuitBreaker({(inner != null ? inner.Name : "?")})", minimumLevel)
    {
        _inner = inner ?? throw new ArgumentNullException(nameof(inner));
        _failureThreshold = failureThreshold > 0 ? failureThreshold : 5;
        _recoveryTimeout = recoveryTimeout ?? TimeSpan.FromSeconds(30);
    }

    /// <summary>Current state of the circuit breaker.</summary>
    public CircuitState State => _state;

    /// <summary>Number of consecutive failures since the last successful write.</summary>
    public int ConsecutiveFailures => _consecutiveFailures;

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public override async Task<WriteResult> BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default)
    {
        if (!IsEnabled || !IsLogLevelEnabled(logEvent))
        {
            return WriteResult.LevelFiltered;
        }

        var currentState = GetEffectiveState();

        if (currentState == CircuitState.Open)
        {
            Interlocked.Increment(ref DroppedCount);
            return WriteResult.Dropped;
        }

        WriteResult result;
        try
        {
            result = await _inner.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception)
        {
            result = WriteResult.Failed;
        }

        if (result == WriteResult.Success)
        {
            OnSuccess(currentState);
            Interlocked.Increment(ref BlastedCount);
        }
        else
        {
            OnFailure();
            Interlocked.Increment(ref DroppedCount);
        }

        return result;
    }

    public override Task FlushAsync(CancellationToken cancellationToken = default)
        => _inner.FlushAsync(cancellationToken);

    public override ValueTask DisposeAsync() => _inner.DisposeAsync();

    private CircuitState GetEffectiveState()
    {
        if (_state != CircuitState.Open)
        {
            return _state;
        }

        if (_openedAt.Elapsed >= _recoveryTimeout)
        {
            Transition(CircuitState.Open, CircuitState.HalfOpen);
            return CircuitState.HalfOpen;
        }

        return CircuitState.Open;
    }

    private void OnSuccess(CircuitState wasState)
    {
        Interlocked.Exchange(ref _consecutiveFailures, 0);
        if (wasState == CircuitState.HalfOpen)
        {
            Transition(CircuitState.HalfOpen, CircuitState.Closed);
        }
    }

    private void OnFailure()
    {
        var failures = Interlocked.Increment(ref _consecutiveFailures);
        if (_state == CircuitState.HalfOpen)
        {
            Transition(CircuitState.HalfOpen, CircuitState.Open);
        }
        else if (_state == CircuitState.Closed && failures >= _failureThreshold)
        {
            Transition(CircuitState.Closed, CircuitState.Open);
        }
    }

    private void Transition(CircuitState from, CircuitState to)
    {
        lock (_stateLock)
        {
            if (_state != from)
            {
                return;
            }

            _state = to;
            if (to == CircuitState.Open)
            {
                _openedAt.Restart();
            }
            else if (to == CircuitState.Closed)
            {
                _openedAt.Reset();
            }
        }
        StateChanged?.Invoke(from, to);
    }
}