Add circuit breaker configuration support

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Circuit breaker configuration (#474)
 - Support additional commands (#435):
   - `BGREWRITEAOF` (#444)
   - `BGSAVE CANCEL` (#436)
@@ -34,7 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `RESET` (#435)
   - `SAVE` (#440)
 - Custom socket address resolution support via callback (#392)
-- `NodeDiscoveryMode` configuration option for standalone clients, with `Standard`, `Static` (proxy-compatible, e.g. Envoy), and `DiscoverAll` modes (#131)
+- `NodeDiscoveryMode` configuration option for standalone clients (#131)
 
 ## 1.1.0
 

rust/src/ffi.rs

@@ -105,6 +105,18 @@ pub struct ClientSideCacheConfig {
     pub server_assisted: bool,
 }
 
+/// A mirror of [`glide_core::client::ClientCircuitBreakerConfig`] adopted for FFI.
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub struct CircuitBreakerConfig {
+    pub window_size_ms: u32,
+    pub failure_rate_threshold: f32,
+    pub min_errors: u32,
+    pub open_timeout_ms: u32,
+    pub count_timeouts: bool,
+    pub consecutive_successes: u32,
+}
+
 /// A mirror of [`ConnectionRequest`] adopted for FFI.
 #[repr(C)]
 #[derive(Clone, Copy)]
@@ -144,10 +156,16 @@ pub struct ConnectionConfig {
 
     pub has_compression_config: bool,
     pub compression_config: CompressionConfig,
+
     pub read_only: bool,
+
     pub node_discovery_mode: NodeDiscoveryMode,
+
     pub has_client_side_cache_config: bool,
     pub client_side_cache_config: ClientSideCacheConfig,
+
+    pub has_circuit_breaker_config: bool,
+    pub circuit_breaker_config: CircuitBreakerConfig,
 }
 
 #[repr(C)]
@@ -269,14 +287,10 @@ pub(crate) unsafe fn create_connection_request(
                         ServiceType::ElastiCache => glide_core::iam::ServiceType::ElastiCache,
                         ServiceType::MemoryDB => glide_core::iam::ServiceType::MemoryDB,
                     },
-                    refresh_interval_seconds: if auth_info
+                    refresh_interval_seconds: auth_info
                         .iam_credentials
                         .has_refresh_interval_seconds
-                    {
-                        Some(auth_info.iam_credentials.refresh_interval_seconds)
-                    } else {
-                        None
-                    },
+                        .then_some(auth_info.iam_credentials.refresh_interval_seconds),
                 })
             } else {
                 None
@@ -335,8 +349,16 @@ pub(crate) unsafe fn create_connection_request(
                     .then_some(config.compression_config.max_decompressed_size as usize),
             },
         ),
+
         read_only: config.read_only,
 
+        // Node discovery mode
+        node_discovery_mode: match config.node_discovery_mode {
+            NodeDiscoveryMode::Standard => glide_core::client::NodeDiscoveryMode::Standard,
+            NodeDiscoveryMode::Static => glide_core::client::NodeDiscoveryMode::Static,
+            NodeDiscoveryMode::DiscoverAll => glide_core::client::NodeDiscoveryMode::DiscoverAll,
+        },
+
         // Client-side cache configuration
         client_side_cache: if config.has_client_side_cache_config {
             use redis::cache::EvictionPolicy as CoreEvictionPolicy;
@@ -345,26 +367,30 @@ pub(crate) unsafe fn create_connection_request(
                 cache_id: unsafe { ptr_to_str(csc.cache_id) }?,
                 max_cache_kb: csc.max_cache_kb,
                 entry_ttl_ms: csc.entry_ttl_ms,
-                eviction_policy: if csc.has_eviction_policy {
-                    Some(match csc.eviction_policy {
+                eviction_policy: csc
+                    .has_eviction_policy
+                    .then_some(match csc.eviction_policy {
                         EvictionPolicy::Lru => CoreEvictionPolicy::Lru,
                         EvictionPolicy::Lfu => CoreEvictionPolicy::Lfu,
-                    })
-                } else {
-                    None
-                },
+                    }),
                 enable_metrics: csc.enable_metrics,
                 server_assisted: csc.server_assisted,
             })
         } else {
             None
         },
 
-        node_discovery_mode: match config.node_discovery_mode {
-            NodeDiscoveryMode::Standard => glide_core::client::NodeDiscoveryMode::Standard,
-            NodeDiscoveryMode::Static => glide_core::client::NodeDiscoveryMode::Static,
-            NodeDiscoveryMode::DiscoverAll => glide_core::client::NodeDiscoveryMode::DiscoverAll,
-        },
+        // Circuit breaker configuration
+        client_circuit_breaker: config.has_circuit_breaker_config.then_some(
+            glide_core::client::ClientCircuitBreakerConfig {
+                window_size_ms: config.circuit_breaker_config.window_size_ms,
+                failure_rate_threshold: config.circuit_breaker_config.failure_rate_threshold,
+                min_errors: config.circuit_breaker_config.min_errors,
+                open_timeout_ms: config.circuit_breaker_config.open_timeout_ms,
+                count_timeouts: config.circuit_breaker_config.count_timeouts,
+                consecutive_successes: config.circuit_breaker_config.consecutive_successes,
+            },
+        ),
 
         // Unimplemented configuration options.
         client_cert: Vec::new(),
@@ -376,7 +402,6 @@ pub(crate) unsafe fn create_connection_request(
         periodic_checks: None, // TODO #485: Expose periodic_checks in ClusterClientConfiguration.
         inflight_requests_limit: None, // TODO #484: Expose inflight_requests_limit in ConnectionConfiguration.
         address_resolver: None,
-        client_circuit_breaker: None,
     })
 }
 

sources/Valkey.Glide/Abstract/Database.SetCommands.cs

@@ -147,7 +147,7 @@ public IAsyncEnumerable<ValkeyValue> SetScanAsync(
         CommandFlags flags = CommandFlags.None)
     {
         GuardClauses.ThrowIfCommandFlags(flags);
-        GuardClauses.ThrowIfNotPositive(pageSize, nameof(pageSize));
+        ArgumentOutOfRangeException.ThrowIfLessThan(pageSize, 0, nameof(pageSize));
 
         var options = new ScanOptions { MatchPattern = pattern, Count = pageSize };
         return SetScanAsync(key, cursor, options).SkipAsync(pageOffset);

sources/Valkey.Glide/Abstract/Database.SortedSetCommands.cs

@@ -300,7 +300,7 @@ public IAsyncEnumerable<SortedSetEntry> SortedSetScanAsync(
         CommandFlags flags = CommandFlags.None)
     {
         GuardClauses.ThrowIfCommandFlags(flags);
-        GuardClauses.ThrowIfNotPositive(pageSize, nameof(pageSize));
+        ArgumentOutOfRangeException.ThrowIfLessThan(pageSize, 0, nameof(pageSize));
 
         var options = new ScanOptions { MatchPattern = pattern, Count = pageSize };
         return SortedSetScanAsync(key, cursor, options).SkipAsync(pageOffset);

sources/Valkey.Glide/Abstract/Database.StreamCommands.cs

@@ -202,7 +202,7 @@ public Task<StreamAutoClaimJustIdResult> StreamAutoClaimIdsOnlyAsync(ValkeyKey k
     public Task<long> StreamTrimAsync(ValkeyKey key, int maxLength, bool useApproximateMaxLength = false, CommandFlags flags = CommandFlags.None)
     {
         GuardClauses.ThrowIfCommandFlags(flags);
-        GuardClauses.ThrowIfNegative(maxLength, nameof(maxLength));
+        ArgumentOutOfRangeException.ThrowIfLessThan(maxLength, 0, nameof(maxLength));
 
         return StreamTrimAsync(key, new StreamTrimOptions.MaxLen
         {

sources/Valkey.Glide/Abstract/ValkeyServer.cs

@@ -290,7 +290,7 @@ public IAsyncEnumerable<ValkeyKey> KeysAsync(
         CommandFlags flags = CommandFlags.None)
     {
         GuardClauses.ThrowIfCommandFlags(flags);
-        GuardClauses.ThrowIfNotPositive(pageSize, nameof(pageSize));
+        ArgumentOutOfRangeException.ThrowIfLessThan(pageSize, 0, nameof(pageSize));
 
         var options = new ScanOptions { MatchPattern = pattern, Count = pageSize };
         return ScanAsync(cursor.ToString(), options).SkipAsync(pageOffset);

sources/Valkey.Glide/CircuitBreakerConfig.cs

@@ -0,0 +1,200 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+using Valkey.Glide.Internals;
+
+namespace Valkey.Glide;
+
+/// <summary>
+/// Configuration for the client-wide circuit breaker.
+/// </summary>
+/// <seealso href="https://glide.valkey.io/how-to/connections/circuit-breaker/">Valkey GLIDE – Configure a Circuit Breaker</seealso>
+public sealed class CircuitBreakerConfig
+{
+    #region Constants
+
+    /// <summary>
+    /// TimeSpan parameters are marshalled to the Rust core as <c>u32</c> milliseconds,
+    /// so values exceeding this limit will throw an exception during configuration.
+    /// </summary>
+    public static readonly TimeSpan MaxTimeSpan = TimeSpan.FromMilliseconds(uint.MaxValue);
+
+    /// <summary>
+    /// Default sliding window duration for error rate calculation (10 seconds).
+    /// </summary>
+    public static readonly TimeSpan DefaultWindowSize = TimeSpan.FromSeconds(10);
+
+    /// <summary>
+    /// Default failure rate threshold within the window to trip the breaker (50%).
+    /// </summary>
+    public const float DefaultFailureRateThreshold = 0.5f;
+
+    /// <summary>
+    /// Default minimum number of errors within the window before the rate is evaluated (50).
+    /// </summary>
+    public const uint DefaultMinErrors = 50;
+
+    /// <summary>
+    /// Default time in Open state before allowing a probe request (5 seconds).
+    /// </summary>
+    public static readonly TimeSpan DefaultOpenTimeout = TimeSpan.FromSeconds(5);
+
+    /// <summary>
+    /// Default number of consecutive successful probe requests needed before closing the breaker (3).
+    /// </summary>
+    public const uint DefaultConsecutiveSuccesses = 3;
+
+    #endregion
+    #region Public Properties
+
+    /// <summary>
+    /// Sliding window duration for error rate calculation.
+    /// </summary>
+    public TimeSpan WindowSize { get; private set; } = DefaultWindowSize;
+
+    /// <summary>
+    /// Failure rate threshold (0.0, 1.0] within the window to trip the breaker.
+    /// </summary>
+    public float FailureRateThreshold { get; private set; } = DefaultFailureRateThreshold;
+
+    /// <summary>
+    /// Minimum number of errors within the window before the rate is evaluated.
+    /// Prevents tripping on small sample sizes.
+    /// </summary>
+    public uint MinErrors { get; private set; } = DefaultMinErrors;
+
+    /// <summary>
+    /// Time in Open state before allowing a probe request.
+    /// </summary>
+    public TimeSpan OpenTimeout { get; private set; } = DefaultOpenTimeout;
+
+    /// <summary>
+    /// Whether command timeouts count toward tripping the breaker. Set to true only if
+    /// timeouts reliably indicate server-side issues rather than client-side thread pool starvation.
+    /// </summary>
+    public bool CountTimeouts { get; private set; } = false;
+
+    /// <summary>
+    /// Number of consecutive successful probe requests needed before closing the breaker.
+    /// Provides a grace period to prevent flapping.
+    /// </summary>
+    public uint ConsecutiveSuccesses { get; private set; } = DefaultConsecutiveSuccesses;
+
+    #endregion
+    #region Public Methods
+
+    /// <summary>
+    /// Sets the sliding window duration for error rate calculation.
+    /// </summary>
+    /// <param name="windowSize">The window size.</param>
+    /// <returns>This instance for method chaining.</returns>
+    /// <exception cref="ArgumentOutOfRangeException">
+    /// Thrown if <paramref name="windowSize"/> is not positive or exceeds <see cref="MaxTimeSpan"/>.
+    /// </exception>
+    public CircuitBreakerConfig WithWindowSize(TimeSpan windowSize)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(windowSize, TimeSpan.Zero, nameof(windowSize));
+        ArgumentOutOfRangeException.ThrowIfGreaterThan(windowSize, MaxTimeSpan, nameof(windowSize));
+
+        WindowSize = windowSize;
+        return this;
+    }
+
+    /// <summary>
+    /// Sets the error rate threshold (0.0, 1.0] within the window to trip the breaker.
+    /// </summary>
+    /// <param name="threshold">The threshold value.</param>
+    /// <returns>This instance for method chaining.</returns>
+    /// <exception cref="ArgumentOutOfRangeException">
+    /// Thrown when <paramref name="threshold"/> is zero or negative.
+    /// </exception>
+    public CircuitBreakerConfig WithFailureRateThreshold(float threshold)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(threshold, 0.0f, nameof(threshold));
+
+        FailureRateThreshold = threshold;
+        return this;
+    }
+
+    /// <summary>
+    /// Sets the minimum number of errors within the window before the rate is evaluated.
+    /// </summary>
+    /// <param name="minErrors">The minimum error count.</param>
+    /// <returns>This instance for method chaining.</returns>
+    /// <exception cref="ArgumentOutOfRangeException">
+    /// Thrown when <paramref name="minErrors"/> is zero.
+    /// </exception>
+    public CircuitBreakerConfig WithMinErrors(uint minErrors)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(minErrors, 0u, nameof(minErrors));
+
+        MinErrors = minErrors;
+        return this;
+    }
+
+    /// <summary>
+    /// Sets the time in Open state before allowing a probe request.
+    /// </summary>
+    /// <param name="openTimeout">The open timeout duration.</param>
+    /// <returns>This instance for method chaining.</returns>
+    /// <exception cref="ArgumentOutOfRangeException">
+    /// Thrown if <paramref name="openTimeout"/> is not positive or exceeds <see cref="MaxTimeSpan"/>.
+    /// </exception>
+    public CircuitBreakerConfig WithOpenTimeout(TimeSpan openTimeout)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(openTimeout, TimeSpan.Zero, nameof(openTimeout));
+        ArgumentOutOfRangeException.ThrowIfGreaterThan(openTimeout, MaxTimeSpan, nameof(openTimeout));
+
+        OpenTimeout = openTimeout;
+        return this;
+    }
+
+    /// <summary>
+    /// Sets whether command timeouts count toward tripping the breaker.
+    /// </summary>
+    /// <param name="countTimeouts">Whether to count timeouts.</param>
+    /// <returns>This instance for method chaining.</returns>
+    public CircuitBreakerConfig WithCountTimeouts(bool countTimeouts = true)
+    {
+        CountTimeouts = countTimeouts;
+        return this;
+    }
+
+    /// <summary>
+    /// Sets the number of consecutive successful probe requests needed before closing the breaker.
+    /// </summary>
+    /// <param name="consecutiveSuccesses">The number of successes.</param>
+    /// <returns>This instance for method chaining.</returns>
+    /// <exception cref="ArgumentOutOfRangeException">
+    /// Thrown when <paramref name="consecutiveSuccesses"/> is zero.
+    /// </exception>
+    public CircuitBreakerConfig WithConsecutiveSuccesses(uint consecutiveSuccesses)
+    {
+        ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(consecutiveSuccesses, 0u, nameof(consecutiveSuccesses));
+
+        ConsecutiveSuccesses = consecutiveSuccesses;
+        return this;
+    }
+
+    #endregion
+    #region Internal Methods
+
+    /// <summary>
+    /// Converts to the FFI representation for marshalling to Rust core.
+    /// </summary>
+    internal FFI.CircuitBreakerConfig ToFfi()
+    {
+        // Casts to uint are safe: validated by WithWindowSize and WithOpenTimeout.
+        var windowSize = (uint)WindowSize.TotalMilliseconds;
+        var openTimeout = (uint)OpenTimeout.TotalMilliseconds;
+
+        return new(
+            windowSize,
+            FailureRateThreshold,
+            MinErrors,
+            openTimeout,
+            CountTimeouts,
+            ConsecutiveSuccesses);
+    }
+
+    #endregion
+}

sources/Valkey.Glide/ClientSideCacheConfig.cs

@@ -96,7 +96,7 @@ public sealed class ClientSideCacheConfig
     /// <param name="maxCacheKb">Maximum size of the cache in kilobytes (KB). Must be positive.</param>
     /// <param name="entryTtl">Time-To-Live for cached entries. Use <see cref="TimeSpan.Zero"/> to disable expiration.</param>
     /// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="maxCacheKb"/> is zero.</exception>
-    /// <exception cref="ArgumentException">Thrown when <paramref name="entryTtl"/> is negative.</exception>
+    /// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="entryTtl"/> is negative.</exception>
     /// <example>
     /// <code>
     /// var cache = new ClientSideCacheConfig(1024, TimeSpan.FromMinutes(1))
@@ -111,12 +111,8 @@ public sealed class ClientSideCacheConfig
     /// </example>
     public ClientSideCacheConfig(ulong maxCacheKb, TimeSpan entryTtl)
     {
-        if (maxCacheKb == 0)
-        {
-            throw new ArgumentOutOfRangeException(nameof(maxCacheKb), "maxCacheKb must be positive.");
-        }
-
-        GuardClauses.ThrowIfNegative(entryTtl);
+        ArgumentOutOfRangeException.ThrowIfEqual(maxCacheKb, 0u, nameof(maxCacheKb));
+        ArgumentOutOfRangeException.ThrowIfLessThan(entryTtl, TimeSpan.Zero, nameof(entryTtl));
 
         MaxCacheKb = maxCacheKb;
         EntryTtl = entryTtl;