fix(flagd): no retry for certain error codes, implement test steps

Signed-off-by: lea konvalinka <[email protected]>

Author: leakonvalinka

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java

@@ -10,6 +10,7 @@
 import io.grpc.ClientInterceptor;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
 import lombok.Builder;
@@ -122,6 +123,14 @@ public class FlagdOptions {
     @Builder.Default
     private int retryGracePeriod =
             fallBackToEnvOrDefault(Config.STREAM_RETRY_GRACE_PERIOD, Config.DEFAULT_STREAM_RETRY_GRACE_PERIOD);
+
+    /**
+     * List of grpc response status codes for which failed connections are not retried.
+     * Defaults to empty list
+     */
+    @Builder.Default
+    private List<String> nonRetryableStatusCodes = new ArrayList<>();
+
     /**
      * Selector to be used with flag sync gRPC contract.
      **/

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSource.java

@@ -16,10 +16,12 @@
 import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsRequest;
 import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse;
 import dev.openfeature.sdk.Awaitable;
+import dev.openfeature.sdk.exceptions.FatalError;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -49,6 +51,7 @@ public class SyncStreamQueueSource implements QueueSource {
     private final BlockingQueue<QueuePayload> outgoingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
     private final FlagSyncServiceStub flagSyncStub;
     private final FlagSyncServiceBlockingStub metadataStub;
+    private final List<String> nonRetryableStatusCodes;
 
     /**
      * Creates a new SyncStreamQueueSource responsible for observing the event stream.
@@ -65,6 +68,7 @@ public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderE
                 FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
         metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
                 .withWaitForReady();
+        nonRetryableStatusCodes = options.getNonRetryableStatusCodes();
     }
 
     // internal use only
@@ -82,6 +86,7 @@ protected SyncStreamQueueSource(
         flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
         metadataStub = blockingStubMock;
+        nonRetryableStatusCodes = options.getNonRetryableStatusCodes();
     }
 
     /** Initialize sync stream connector. */
@@ -135,8 +140,11 @@ private void observeSyncStream() {
                 SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue, shouldThrottle);
                 try {
                     observer.metadata = getMetadata();
-                } catch (Exception metaEx) {
-                    // retry if getMetadata fails
+                } catch (StatusRuntimeException metaEx) {
+                    if (nonRetryableStatusCodes.contains(metaEx.getStatus().getCode().name())) {
+                        throw new FatalError("Failed to connect for metadata request, not retrying for error " + metaEx.getStatus());
+                    }
+                    // retry for other status codes
                     String message = metaEx.getMessage();
                     log.debug("Metadata request error: {}, will restart", message, metaEx);
                     enqueueError(String.format("Error in getMetadata request: %s", message));
@@ -146,7 +154,11 @@ private void observeSyncStream() {
 
                 try {
                     syncFlags(observer);
-                } catch (Exception ex) {
+                } catch (StatusRuntimeException ex) {
+                    if (nonRetryableStatusCodes.contains(ex.getStatus().toString())) {
+                        throw new FatalError("Failed to connect to sync stream, not retrying for error " + ex.getStatus());
+                    }
+                    // retry for other status codes
                     log.error("Unexpected sync stream exception, will restart.", ex);
                     enqueueError(String.format("Error in syncStream: %s", ex.getMessage()));
                     shouldThrottle.set(true);

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/steps/ProviderSteps.java

@@ -1,6 +1,7 @@
 package dev.openfeature.contrib.providers.flagd.e2e.steps;
 
 import static io.restassured.RestAssured.when;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import dev.openfeature.contrib.providers.flagd.Config;
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
@@ -9,10 +10,12 @@
 import dev.openfeature.contrib.providers.flagd.e2e.State;
 import dev.openfeature.sdk.FeatureProvider;
 import dev.openfeature.sdk.OpenFeatureAPI;
+import dev.openfeature.sdk.ProviderState;
 import io.cucumber.java.After;
 import io.cucumber.java.AfterAll;
 import io.cucumber.java.BeforeAll;
 import io.cucumber.java.en.Given;
+import io.cucumber.java.en.Then;
 import io.cucumber.java.en.When;
 import java.io.File;
 import java.io.IOException;
@@ -33,6 +36,7 @@
 public class ProviderSteps extends AbstractSteps {
 
     public static final int UNAVAILABLE_PORT = 9999;
+    public static final int FORBIDDEN_PORT = 9212;
     static ComposeContainer container;
 
     static Path sharedTempDir;
@@ -51,6 +55,7 @@ public static void beforeAll() throws IOException {
                 .withExposedService("flagd", 8015, Wait.forListeningPort())
                 .withExposedService("flagd", 8080, Wait.forListeningPort())
                 .withExposedService("envoy", 9211, Wait.forListeningPort())
+                .withExposedService("envoy", 9212, Wait.forListeningPort())
                 .withStartupTimeout(Duration.ofSeconds(45));
         container.start();
     }
@@ -87,6 +92,10 @@ public void setupProvider(String providerType) throws InterruptedException {
                 }
                 wait = false;
                 break;
+            case "forbidden":
+                state.builder.port(container.getServicePort("envoy", FORBIDDEN_PORT));
+                wait = false;
+                break;
             case "socket":
                 this.state.providerType = ProviderType.SOCKET;
                 String socketPath =
@@ -190,4 +199,9 @@ public void the_flag_was_modded() {
                 .then()
                 .statusCode(200);
     }
+
+    @Then("the client is in {} state")
+    public void the_client_is_in_fatal_state(String clientState) {
+        assertThat(state.client.getProviderState()).isEqualTo(ProviderState.FATAL);
+    }
 }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/steps/Utils.java

@@ -4,6 +4,7 @@
 import dev.openfeature.contrib.providers.flagd.resolver.rpc.cache.CacheType;
 import dev.openfeature.sdk.Value;
 import java.io.IOException;
+import java.util.List;
 import java.util.Objects;
 import org.testcontainers.shaded.com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -37,6 +38,8 @@ public static Object convert(String value, String type) throws ClassNotFoundExce
                 }
             case "CacheType":
                 return CacheType.valueOf(value.toUpperCase()).getValue();
+            case "StringList":
+                return List.of(value);
             case "Object":
                 return Value.objectToValue(new ObjectMapper().readValue(value, Object.class));
         }

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/e2e/steps/config/ConfigSteps.java

@@ -121,6 +121,7 @@ private static String mapOptionNames(String option) {
         propertyMapper.put("keepAliveTime", "keepAlive");
         propertyMapper.put("retryBackoffMaxMs", "keepAlive");
         propertyMapper.put("cache", "cacheType");
+        propertyMapper.put("fatalStatusCodes", "nonRetryableStatusCodes");
 
         if (propertyMapper.get(option) != null) {
             option = propertyMapper.get(option);