Add registry decorators for deadlines and service-keyed circuit breakers

.gitignore

@@ -5,3 +5,6 @@
 build
 
 .idea
+
+.superpowers/
+docs/superpowers/

grpc-circuitbreaker-utils/build.gradle.kts

@@ -7,11 +7,12 @@ plugins {
 
 dependencies {
   api(commonLibs.grpc.api)
+  api(commonLibs.typesafe.config)
   api(projects.grpcContextUtils)
+  api(projects.grpcClientUtils)
 
   implementation(commonLibs.slf4j2.api)
   implementation(localLibs.resilience4j.circuitbreaker)
-  implementation(commonLibs.typesafe.config)
   implementation(commonLibs.guava)
   implementation(commonLibs.jakarta.inject.api)
 

grpc-circuitbreaker-utils/gradle.lockfile

@@ -20,13 +20,16 @@ com.google.guava:guava:33.3.1-android=runtimeClasspath,testRuntimeClasspath
 com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 com.google.j2objc:j2objc-annotations:2.8=compileClasspath,testCompileClasspath
 com.google.j2objc:j2objc-annotations:3.0.0=runtimeClasspath,testRuntimeClasspath
+com.google.protobuf:protobuf-java-util:3.25.8=testRuntimeClasspath
+com.google.protobuf:protobuf-java:3.25.8=testRuntimeClasspath
 com.typesafe:config:1.4.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.github.resilience4j:resilience4j-circuitbreaker:1.7.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.github.resilience4j:resilience4j-core:1.7.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.grpc:grpc-api:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.grpc:grpc-bom:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
-io.grpc:grpc-context:1.75.0=runtimeClasspath,testRuntimeClasspath
+io.grpc:grpc-context:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.grpc:grpc-core:1.75.0=runtimeClasspath,testRuntimeClasspath
+io.grpc:grpc-inprocess:1.75.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.netty:netty-bom:4.1.132.Final=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 io.perfmark:perfmark-api:0.27.0=runtimeClasspath,testRuntimeClasspath
 io.vavr:vavr-match:0.10.2=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerChannelRegistry.java

@@ -0,0 +1,31 @@
+package org.hypertrace.circuitbreaker.grpcutils;
+
+import com.typesafe.config.Config;
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry;
+import org.hypertrace.core.grpcutils.client.InterceptingChannelRegistry;
+
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class CircuitBreakerChannelRegistry {
+
+  static final String CIRCUIT_BREAKER_CONFIG = CircuitBreakerConfigParser.CIRCUIT_BREAKER_CONFIG;
+
+  public static GrpcChannelRegistry wrap(final GrpcChannelRegistry registry, final Config config) {
+    return wrap(registry, CircuitBreakerConfigParser.parseServiceThresholds(config));
+  }
+
+  public static GrpcChannelRegistry wrap(
+      final GrpcChannelRegistry registry,
+      final Map<String, CircuitBreakerThresholds> thresholdsByService) {
+    if (registry == null || thresholdsByService == null || thresholdsByService.isEmpty()) {
+      log.debug("No enabled service circuit breakers; leaving the channel registry unchanged");
+      return registry;
+    }
+    return InterceptingChannelRegistry.wrap(
+        registry, CircuitBreakerInterceptorFactory.forService(thresholdsByService));
+  }
+}

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerConfigParser.java

@@ -1,6 +1,12 @@
 package org.hypertrace.circuitbreaker.grpcutils;
 
 import com.typesafe.config.Config;
+import com.typesafe.config.ConfigException;
+import com.typesafe.config.ConfigUtil;
+import io.grpc.Status;
+import java.util.EnumSet;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -26,6 +32,8 @@ public class CircuitBreakerConfigParser {
   private static final String PERMITTED_NUMBER_OF_CALLS_IN_HALF_OPEN_STATE =
       "permittedNumberOfCallsInHalfOpenState";
   private static final String SLIDING_WINDOW_TYPE = "slidingWindowType";
+  private static final String FAILURE_STATUS_CODES = "failureStatusCodes";
+  static final String CIRCUIT_BREAKER_CONFIG = "circuit.breaker.config";
   public static final String ENABLED = "enabled";
   public static final String DEFAULT_THRESHOLDS = "defaultThresholds";
   private static final Set<String> NON_THRESHOLD_KEYS = Set.of(ENABLED, DEFAULT_THRESHOLDS);
@@ -56,9 +64,54 @@ public static <T> CircuitBreakerConfiguration.CircuitBreakerConfigurationBuilder
     return builder;
   }
 
+  /**
+   * Service-keyed thresholds for {@code circuit.breaker.config}. An absent or disabled block is an
+   * empty map. Defaults come from {@link CircuitBreakerThresholds}; config only overlays fields
+   * that are present.
+   */
+  public static Map<String, CircuitBreakerThresholds> parseServiceThresholds(
+      final Config rootConfig) {
+    if (rootConfig == null || !rootConfig.hasPath(CIRCUIT_BREAKER_CONFIG)) {
+      log.debug("circuit.breaker.config is absent; no service circuit breakers installed");
+      return Map.of();
+    }
+    final Config config = rootConfig.getConfig(CIRCUIT_BREAKER_CONFIG);
+    if (!config.hasPath(ENABLED) || !config.getBoolean(ENABLED)) {
+      log.debug("circuit.breaker.config is disabled; no service circuit breakers installed");
+      return Map.of();
+    }
+    final CircuitBreakerThresholds defaults =
+        config.hasPath(DEFAULT_THRESHOLDS)
+            ? overlay(buildCircuitBreakerDefaultThresholds(), config.getConfig(DEFAULT_THRESHOLDS))
+            : buildCircuitBreakerDefaultThresholds();
+    final Map<String, CircuitBreakerThresholds> thresholdsByService = new LinkedHashMap<>();
+    for (final String serviceName : config.root().keySet()) {
+      if (ENABLED.equals(serviceName) || DEFAULT_THRESHOLDS.equals(serviceName)) {
+        continue;
+      }
+      // getConfig(key) treats dots as path separators, so service names are read with joinPath.
+      final CircuitBreakerThresholds thresholds =
+          overlay(defaults, config.getConfig(ConfigUtil.joinPath(serviceName)));
+      if (!thresholds.isEnabled()) {
+        log.debug("Service circuit breaker disabled for {}", serviceName);
+        continue;
+      }
+      thresholdsByService.put(serviceName, thresholds);
+    }
+    return Map.copyOf(thresholdsByService);
+  }
+
+  public static CircuitBreakerThresholds overlay(
+      final CircuitBreakerThresholds base, final Config overrides) {
+    return applyOverrides(base.toBuilder(), overrides).build();
+  }
+
   private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config config) {
-    CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder =
-        CircuitBreakerThresholds.builder();
+    return overlay(buildCircuitBreakerDefaultThresholds(), config);
+  }
+
+  private static CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder applyOverrides(
+      final CircuitBreakerThresholds.CircuitBreakerThresholdsBuilder builder, final Config config) {
 
     if (config.hasPath(FAILURE_RATE_THRESHOLD)) {
       builder.failureRateThreshold((float) config.getDouble(FAILURE_RATE_THRESHOLD));
@@ -97,7 +150,25 @@ private static CircuitBreakerThresholds buildCircuitBreakerThresholds(Config con
       builder.enabled(config.getBoolean(ENABLED));
     }
 
-    return builder.build();
+    if (config.hasPath(FAILURE_STATUS_CODES)) {
+      builder.failureStatusCodes(
+          parseFailureStatusCodes(config.getStringList(FAILURE_STATUS_CODES)));
+    }
+
+    return builder;
+  }
+
+  // An unknown name must fail startup. Ignoring it would protect the wrong calls.
+  private static Set<Status.Code> parseFailureStatusCodes(List<String> names) {
+    Set<Status.Code> codes = EnumSet.noneOf(Status.Code.class);
+    for (String name : names) {
+      try {
+        codes.add(Status.Code.valueOf(name));
+      } catch (IllegalArgumentException exception) {
+        throw new ConfigException.BadValue(FAILURE_STATUS_CODES, name);
+      }
+    }
+    return Set.copyOf(codes);
   }
 
   public static CircuitBreakerThresholds buildCircuitBreakerDefaultThresholds() {

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerInterceptorFactory.java

@@ -1,14 +1,26 @@
 package org.hypertrace.circuitbreaker.grpcutils;
 
+import com.typesafe.config.Config;
+import io.grpc.ClientInterceptor;
 import jakarta.inject.Inject;
 import java.time.Clock;
+import java.util.Map;
 import lombok.AllArgsConstructor;
 import org.hypertrace.circuitbreaker.grpcutils.resilience.ResilienceCircuitBreakerFactory;
 
 @AllArgsConstructor(onConstructor_ = @Inject)
 public class CircuitBreakerInterceptorFactory {
   private final Clock clock;
 
+  public static ClientInterceptor forService(final Config rootConfig) {
+    return forService(CircuitBreakerConfigParser.parseServiceThresholds(rootConfig));
+  }
+
+  public static ClientInterceptor forService(
+      final Map<String, CircuitBreakerThresholds> thresholdsByService) {
+    return ServiceCircuitBreakerInterceptor.create(thresholdsByService);
+  }
+
   public CircuitBreakerInterceptor buildInterceptor(CircuitBreakerConfiguration<?> configuration) {
     return ResilienceCircuitBreakerFactory.getCircuitBreakerInterceptor(configuration, clock);
   }

grpc-circuitbreaker-utils/src/main/java/org/hypertrace/circuitbreaker/grpcutils/CircuitBreakerThresholds.java

@@ -1,11 +1,13 @@
 package org.hypertrace.circuitbreaker.grpcutils;
 
+import io.grpc.Status;
 import java.time.Duration;
+import java.util.Set;
 import lombok.Builder;
 import lombok.Value;
 
 @Value
-@Builder
+@Builder(toBuilder = true)
 public class CircuitBreakerThresholds {
   // Percentage of failures to trigger OPEN state
   @Builder.Default float failureRateThreshold = 50f;
@@ -25,6 +27,11 @@ public class CircuitBreakerThresholds {
   @Builder.Default int permittedNumberOfCallsInHalfOpenState = 5;
   @Builder.Default boolean enabled = true;
 
+  @Builder.Default
+  Set<Status.Code> failureStatusCodes =
+      Set.of(
+          Status.Code.DEADLINE_EXCEEDED, Status.Code.UNAVAILABLE, Status.Code.RESOURCE_EXHAUSTED);
+
   public enum SlidingWindowType {
     COUNT_BASED,
     TIME_BASED