Merge pull request #1930 from tyrielv/tyrielv/oom-circuit-breaker-timeout

Add circuit breaker, connection pool timeout, and contention telemetry

Author: tyrielv

GVFS/GVFS.Common/Http/HttpRequestor.cs

@@ -17,6 +17,9 @@ namespace GVFS.Common.Http
 {
     public abstract class HttpRequestor : IDisposable
     {
+        private const int ConnectionPoolWaitTimeoutMs = 30_000;
+        private const int ConnectionPoolContentionThresholdMs = 100;
+
         private static long requestCount = 0;
         private static SemaphoreSlim availableConnections;
 
@@ -126,8 +129,30 @@ protected GitEndPointResponseData SendRequest(
             responseMetadata.Add("availableConnections", availableConnections.CurrentCount);
 
             Stopwatch requestStopwatch = Stopwatch.StartNew();
-            availableConnections.Wait(cancellationToken);
-            TimeSpan connectionWaitTime = requestStopwatch.Elapsed;
+
+            if (!availableConnections.Wait(ConnectionPoolWaitTimeoutMs, cancellationToken))
+            {
+                TimeSpan connectionWaitTime = requestStopwatch.Elapsed;
+                responseMetadata.Add("connectionWaitTimeMS", $"{connectionWaitTime.TotalMilliseconds:F4}");
+                this.Tracer.RelatedWarning(responseMetadata, "SendRequest: Connection pool exhausted, all connections busy");
+
+                return new GitEndPointResponseData(
+                    HttpStatusCode.ServiceUnavailable,
+                    new GitObjectsHttpException(HttpStatusCode.ServiceUnavailable, "Connection pool exhausted - all connections busy"),
+                    shouldRetry: true,
+                    message: null,
+                    onResponseDisposed: null);
+            }
+
+            TimeSpan connectionWaitTimeElapsed = requestStopwatch.Elapsed;
+            if (connectionWaitTimeElapsed.TotalMilliseconds > ConnectionPoolContentionThresholdMs)
+            {
+                EventMetadata contentionMetadata = new EventMetadata();
+                contentionMetadata.Add("RequestId", requestId);
+                contentionMetadata.Add("availableConnections", availableConnections.CurrentCount);
+                contentionMetadata.Add("connectionWaitTimeMS", $"{connectionWaitTimeElapsed.TotalMilliseconds:F4}");
+                this.Tracer.RelatedWarning(contentionMetadata, "SendRequest: Connection pool contention detected");
+            }
 
             TimeSpan responseWaitTime = default(TimeSpan);
             GitEndPointResponseData gitEndPointResponseData = null;
@@ -248,7 +273,7 @@ protected GitEndPointResponseData SendRequest(
             }
             finally
             {
-                responseMetadata.Add("connectionWaitTimeMS", $"{connectionWaitTime.TotalMilliseconds:F4}");
+                responseMetadata.Add("connectionWaitTimeMS", $"{connectionWaitTimeElapsed.TotalMilliseconds:F4}");
                 responseMetadata.Add("responseWaitTimeMS", $"{responseWaitTime.TotalMilliseconds:F4}");
 
                 this.Tracer.RelatedEvent(EventLevel.Informational, "NetworkResponse", responseMetadata);

GVFS/GVFS.Common/RetryCircuitBreaker.cs

@@ -0,0 +1,72 @@
+using System;
+using System.Threading;
+
+namespace GVFS.Common
+{
+    /// <summary>
+    /// Global circuit breaker for retry operations. When too many consecutive failures
+    /// occur (e.g., during system-wide resource exhaustion), the circuit opens and
+    /// subsequent retry attempts fail fast instead of consuming connections and adding
+    /// backoff delays that worsen the resource pressure.
+    /// </summary>
+    public static class RetryCircuitBreaker
+    {
+        public const int DefaultFailureThreshold = 15;
+        public const int DefaultCooldownMs = 30_000;
+
+        private static int failureThreshold = DefaultFailureThreshold;
+        private static int cooldownMs = DefaultCooldownMs;
+        private static int consecutiveFailures = 0;
+        private static long circuitOpenedAtUtcTicks = 0;
+
+        public static bool IsOpen
+        {
+            get
+            {
+                if (Volatile.Read(ref consecutiveFailures) < failureThreshold)
+                {
+                    return false;
+                }
+
+                long openedAt = Volatile.Read(ref circuitOpenedAtUtcTicks);
+                return (DateTime.UtcNow.Ticks - openedAt) < TimeSpan.FromMilliseconds(cooldownMs).Ticks;
+            }
+        }
+
+        public static int ConsecutiveFailures => Volatile.Read(ref consecutiveFailures);
+
+        public static void RecordSuccess()
+        {
+            Interlocked.Exchange(ref consecutiveFailures, 0);
+        }
+
+        public static void RecordFailure()
+        {
+            int failures = Interlocked.Increment(ref consecutiveFailures);
+            if (failures >= failureThreshold)
+            {
+                Volatile.Write(ref circuitOpenedAtUtcTicks, DateTime.UtcNow.Ticks);
+            }
+        }
+
+        /// <summary>
+        /// Resets the circuit breaker to its initial state. Intended for testing.
+        /// </summary>
+        public static void Reset()
+        {
+            Volatile.Write(ref consecutiveFailures, 0);
+            Volatile.Write(ref circuitOpenedAtUtcTicks, 0);
+            Volatile.Write(ref failureThreshold, DefaultFailureThreshold);
+            Volatile.Write(ref cooldownMs, DefaultCooldownMs);
+        }
+
+        /// <summary>
+        /// Configures the circuit breaker thresholds. Intended for testing.
+        /// </summary>
+        public static void Configure(int threshold, int cooldownMilliseconds)
+        {
+            Volatile.Write(ref failureThreshold, threshold);
+            Volatile.Write(ref cooldownMs, cooldownMilliseconds);
+        }
+    }
+}

GVFS/GVFS.Common/RetryWrapper.cs

@@ -60,6 +60,24 @@ public static Action<ErrorEventArgs> StandardErrorHandler(ITracer tracer, long r
 
         public InvocationResult Invoke(Func<int, CallbackResult> toInvoke)
         {
+            // NOTE: Cascade risk — connection pool timeouts (HttpRequestor returns
+            // ServiceUnavailable when the semaphore wait expires) flow through here
+            // as callback errors with shouldRetry=true and count toward the circuit
+            // breaker. Under sustained pool exhaustion, 15 timeouts can trip the
+            // breaker and fail-fast ALL retry operations for 30 seconds — including
+            // requests that might have succeeded. In practice, request coalescing
+            // (GVFSGitObjects) and the larger pool size drastically reduce the
+            // likelihood of sustained pool exhaustion. If telemetry shows this
+            // cascade occurring, consider excluding local resource pressure (pool
+            // timeouts) from circuit breaker failure counts.
+            if (RetryCircuitBreaker.IsOpen)
+            {
+                RetryableException circuitOpenError = new RetryableException(
+                    "Circuit breaker is open - too many consecutive failures. Fast-failing to prevent resource exhaustion.");
+                this.OnFailure(new ErrorEventArgs(circuitOpenError, tryCount: 1, willRetry: false));
+                return new InvocationResult(1, circuitOpenError);
+            }
+
             // Use 1-based counting. This makes reporting look a lot nicer and saves a lot of +1s
             for (int tryCount = 1; tryCount <= this.maxAttempts; ++tryCount)
             {
@@ -70,13 +88,19 @@ public InvocationResult Invoke(Func<int, CallbackResult> toInvoke)
                     CallbackResult result = toInvoke(tryCount);
                     if (result.HasErrors)
                     {
+                        if (result.ShouldRetry)
+                        {
+                            RetryCircuitBreaker.RecordFailure();
+                        }
+
                         if (!this.ShouldRetry(tryCount, null, result))
                         {
                             return new InvocationResult(tryCount, result.Error, result.Result);
                         }
                     }
                     else
                     {
+                        RetryCircuitBreaker.RecordSuccess();
                         return new InvocationResult(tryCount, true, result.Result);
                     }
                 }
@@ -92,6 +116,7 @@ e is AggregateException
                         throw;
                     }
 
+                    RetryCircuitBreaker.RecordFailure();
                     if (!this.ShouldRetry(tryCount, exceptionToReport, null))
                     {
                         return new InvocationResult(tryCount, exceptionToReport);

GVFS/GVFS.UnitTests/Common/RetryWrapperTests.cs

@@ -12,6 +12,12 @@ namespace GVFS.UnitTests.Common
     [TestFixture]
     public class RetryWrapperTests
     {
+        [SetUp]
+        public void SetUp()
+        {
+            RetryCircuitBreaker.Reset();
+        }
+
         [TestCase]
         [Category(CategoryConstants.ExceptionExpected)]
         public void WillRetryOnIOException()
@@ -233,5 +239,118 @@ public void WillRetryWhenRequested()
             actualTries.ShouldEqual(ExpectedTries);
             actualFailures.ShouldEqual(ExpectedFailures);
         }
+
+        [TestCase]
+        [Category(CategoryConstants.ExceptionExpected)]
+        public void CircuitBreakerOpensAfterConsecutiveFailures()
+        {
+            const int Threshold = 5;
+            const int CooldownMs = 5000;
+            RetryCircuitBreaker.Configure(Threshold, CooldownMs);
+
+            // Generate enough failures to trip the circuit breaker
+            for (int i = 0; i < Threshold; i++)
+            {
+                RetryWrapper<bool> wrapper = new RetryWrapper<bool>(1, CancellationToken.None, exponentialBackoffBase: 0);
+                wrapper.Invoke(tryCount => throw new IOException("simulated failure"));
+            }
+
+            RetryCircuitBreaker.IsOpen.ShouldBeTrue("Circuit breaker should be open after threshold failures");
+
+            // Next invocation should fail fast without calling the callback
+            int callbackInvocations = 0;
+            RetryWrapper<bool> dut = new RetryWrapper<bool>(5, CancellationToken.None, exponentialBackoffBase: 0);
+            RetryWrapper<bool>.InvocationResult result = dut.Invoke(
+                tryCount =>
+                {
+                    callbackInvocations++;
+                    return new RetryWrapper<bool>.CallbackResult(true);
+                });
+
+            result.Succeeded.ShouldEqual(false);
+            callbackInvocations.ShouldEqual(0);
+        }
+
+        [TestCase]
+        public void CircuitBreakerResetsOnSuccess()
+        {
+            const int Threshold = 3;
+            RetryCircuitBreaker.Configure(Threshold, 30_000);
+
+            // Record failures just below threshold
+            for (int i = 0; i < Threshold - 1; i++)
+            {
+                RetryCircuitBreaker.RecordFailure();
+            }
+
+            RetryCircuitBreaker.IsOpen.ShouldBeFalse("Circuit should still be closed below threshold");
+
+            // A successful invocation resets the counter
+            RetryWrapper<bool> dut = new RetryWrapper<bool>(1, CancellationToken.None, exponentialBackoffBase: 0);
+            dut.Invoke(tryCount => new RetryWrapper<bool>.CallbackResult(true));
+
+            RetryCircuitBreaker.ConsecutiveFailures.ShouldEqual(0);
+
+            // Now threshold more failures are needed to trip it again
+            for (int i = 0; i < Threshold - 1; i++)
+            {
+                RetryCircuitBreaker.RecordFailure();
+            }
+
+            RetryCircuitBreaker.IsOpen.ShouldBeFalse("Circuit should still be closed after reset");
+        }
+
+        [TestCase]
+        public void CircuitBreakerIgnoresNonRetryableErrors()
+        {
+            const int Threshold = 3;
+            RetryCircuitBreaker.Configure(Threshold, 30_000);
+
+            // Generate non-retryable failures (e.g., 404/400) — these should NOT count
+            for (int i = 0; i < Threshold + 5; i++)
+            {
+                RetryWrapper<bool> wrapper = new RetryWrapper<bool>(1, CancellationToken.None, exponentialBackoffBase: 0);
+                wrapper.Invoke(tryCount => new RetryWrapper<bool>.CallbackResult(new Exception("404 Not Found"), shouldRetry: false));
+            }
+
+            RetryCircuitBreaker.IsOpen.ShouldBeFalse("Non-retryable errors should not trip the circuit breaker");
+            RetryCircuitBreaker.ConsecutiveFailures.ShouldEqual(0);
+        }
+
+        [TestCase]
+        [Category(CategoryConstants.ExceptionExpected)]
+        public void CircuitBreakerClosesAfterCooldown()
+        {
+            const int Threshold = 3;
+            const int CooldownMs = 100; // Very short cooldown for testing
+            RetryCircuitBreaker.Configure(Threshold, CooldownMs);
+
+            // Trip the circuit breaker
+            for (int i = 0; i < Threshold; i++)
+            {
+                RetryWrapper<bool> wrapper = new RetryWrapper<bool>(1, CancellationToken.None, exponentialBackoffBase: 0);
+                wrapper.Invoke(tryCount => throw new IOException("simulated failure"));
+            }
+
+            RetryCircuitBreaker.IsOpen.ShouldBeTrue("Circuit should be open");
+
+            // Wait for cooldown to expire
+            Thread.Sleep(CooldownMs + 50);
+
+            RetryCircuitBreaker.IsOpen.ShouldBeFalse("Circuit should be closed after cooldown");
+
+            // Should be able to invoke successfully now
+            int callbackInvocations = 0;
+            RetryWrapper<bool> dut = new RetryWrapper<bool>(1, CancellationToken.None, exponentialBackoffBase: 0);
+            RetryWrapper<bool>.InvocationResult result = dut.Invoke(
+                tryCount =>
+                {
+                    callbackInvocations++;
+                    return new RetryWrapper<bool>.CallbackResult(true);
+                });
+
+            result.Succeeded.ShouldEqual(true);
+            callbackInvocations.ShouldEqual(1);
+        }
     }
 }