Updated packages

EonaCat.LogStack/EonaCatLoggerCore/Flows/FailoverFlow.cs

@@ -15,24 +15,65 @@ namespace EonaCat.LogStack.Flows
     {
         private readonly IFlow _primary;
         private readonly IFlow _secondary;
+        private readonly TimeSpan _recoveryCheckInterval;
+        private readonly int _failureThreshold;
 
-        public FailoverFlow(IFlow primary, IFlow secondary)
+        private IFlow _activeFlow;
+        private long _consecutiveFailures;
+        private DateTime _lastRecoveryCheck;
+        private CancellationTokenSource _recoveryCancellation;
+        private Task _recoveryTask;
+        private readonly object _syncLock = new object();
+
+        public FailoverFlow(
+            IFlow primary, 
+            IFlow secondary,
+            TimeSpan? recoveryCheckInterval = null,
+            int failureThreshold = 5)
             : base($"Failover({primary.Name})", primary.MinimumLevel)
         {
             _primary = primary ?? throw new ArgumentNullException(nameof(primary));
             _secondary = secondary ?? throw new ArgumentNullException(nameof(secondary));
+            _recoveryCheckInterval = recoveryCheckInterval ?? TimeSpan.FromSeconds(10);
+            _failureThreshold = failureThreshold;
+            _activeFlow = _primary;
+            _consecutiveFailures = 0;
+            _lastRecoveryCheck = DateTime.UtcNow;
+            _recoveryCancellation = new CancellationTokenSource();
+            _recoveryTask = StartRecoveryDetectionAsync();
         }
 
         public override async Task<WriteResult> BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default)
         {
-            var result = await _primary.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false);
+            var result = await _activeFlow.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false);
 
             if (result == WriteResult.Success)
             {
+                // Reset failure counter on success
+                Interlocked.Exchange(ref _consecutiveFailures, 0);
                 return result;
             }
 
-            return await _secondary.BlastAsync(logEvent, cancellationToken).ConfigureAwait(false);
+            // Only failover to secondary if using primary
+            if (_activeFlow == _primary)
+            {
+                var failures = Interlocked.Increment(ref _consecutiveFailures);
+
+                if (failures >= _failureThreshold)
+                {
+                    lock (_syncLock)
+                    {
+                        // Double-check inside lock
+                        if (_activeFlow == _primary && Interlocked.Read(ref _consecutiveFailures) >= _failureThreshold)
+                        {
+                            _activeFlow = _secondary;
+                            Interlocked.Exchange(ref _consecutiveFailures, 0);
+                        }
+                    }
+                }
+            }
+
+            return result;
         }
 
         public override async Task FlushAsync(CancellationToken cancellationToken = default)
@@ -43,8 +84,80 @@ namespace EonaCat.LogStack.Flows
 
         public override async ValueTask DisposeAsync()
         {
+            _recoveryCancellation?.Cancel();
+
+            if (_recoveryTask != null)
+            {
+                try
+                {
+                    await _recoveryTask.ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    // Expected when cancelling the recovery task
+                }
+            }
+
+            _recoveryCancellation?.Dispose();
             await _primary.DisposeAsync();
             await _secondary.DisposeAsync();
         }
+
+        private async Task StartRecoveryDetectionAsync()
+        {
+            try
+            {
+                while (!_recoveryCancellation.Token.IsCancellationRequested)
+                {
+                    await Task.Delay(_recoveryCheckInterval, _recoveryCancellation.Token).ConfigureAwait(false);
+
+                    // Only check recovery if we've switched to secondary
+                    if (_activeFlow == _secondary)
+                    {
+                        await CheckPrimaryRecoveryAsync().ConfigureAwait(false);
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                // Expected when shutting down
+            }
+        }
+
+        private async Task CheckPrimaryRecoveryAsync()
+        {
+            try
+            {
+                // Try to write a minimal test log to check if primary is healthy
+                var testEvent = new LogEvent
+                {
+                    Level = LogLevel.Trace,
+                    Message = new ReadOnlyMemory<char>("recovery_check".ToCharArray()),
+                    Timestamp = DateTime.UtcNow.Ticks,
+                    Category = "RecoveryDetection",
+                    ThreadId = Thread.CurrentThread.ManagedThreadId
+                };
+
+                var result = await _primary.BlastAsync(testEvent, _recoveryCancellation.Token).ConfigureAwait(false);
+
+                if (result == WriteResult.Success)
+                {
+                    lock (_syncLock)
+                    {
+                        if (_activeFlow == _secondary)
+                        {
+                            _activeFlow = _primary;
+                            Interlocked.Exchange(ref _consecutiveFailures, 0);
+                            _lastRecoveryCheck = DateTime.UtcNow;
+                        }
+                    }
+                }
+            }
+            catch
+            {
+                // Silently catch exceptions during recovery check
+                // The primary is still unhealthy
+            }
+        }
     }
 }