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

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> fatalStatusCodes = new ArrayList<>();
+
     /**
      * Selector to be used with flag sync gRPC contract.
      **/

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

@@ -5,6 +5,7 @@
 import dev.openfeature.contrib.providers.flagd.resolver.process.InProcessResolver;
 import dev.openfeature.contrib.providers.flagd.resolver.rpc.RpcResolver;
 import dev.openfeature.contrib.providers.flagd.resolver.rpc.cache.Cache;
+import dev.openfeature.sdk.ErrorCode;
 import dev.openfeature.sdk.EvaluationContext;
 import dev.openfeature.sdk.EventProvider;
 import dev.openfeature.sdk.Hook;
@@ -135,7 +136,7 @@ public void initialize(EvaluationContext evaluationContext) throws Exception {
     public void shutdown() {
         synchronized (syncResources) {
             try {
-                if (!syncResources.isInitialized() || syncResources.isShutDown()) {
+                if (syncResources.isShutDown()) {
                     return;
                 }
 
@@ -193,7 +194,7 @@ EvaluationContext getEnrichedContext() {
 
     @SuppressWarnings("checkstyle:fallthrough")
     private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
-        log.debug("FlagdProviderEvent event {} ", flagdProviderEvent.getEvent());
+        log.info("FlagdProviderEvent event {} ", flagdProviderEvent.getEvent());
         synchronized (syncResources) {
             /*
              * We only use Error and Ready as previous states.
@@ -222,20 +223,26 @@ private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
                     onReady();
                     syncResources.setPreviousEvent(ProviderEvent.PROVIDER_READY);
                     break;
-
-                case PROVIDER_ERROR:
-                    if (syncResources.getPreviousEvent() != ProviderEvent.PROVIDER_ERROR) {
-                        onError();
-                        syncResources.setPreviousEvent(ProviderEvent.PROVIDER_ERROR);
+                case PROVIDER_STALE:
+                    if (syncResources.getPreviousEvent() != ProviderEvent.PROVIDER_STALE) {
+                        onStale();
+                        syncResources.setPreviousEvent(ProviderEvent.PROVIDER_STALE);
                     }
                     break;
-
+                case PROVIDER_ERROR:
+                    onError();
+                    break;
                 default:
                     log.warn("Unknown event {}", flagdProviderEvent.getEvent());
             }
         }
     }
 
+    private void onError() {
+        this.emitProviderError(ProviderEventDetails.builder().errorCode(ErrorCode.PROVIDER_FATAL).build());
+        shutdown();
+    }
+
     private void onConfigurationChanged(FlagdProviderEvent flagdProviderEvent) {
         this.emitProviderConfigurationChanged(ProviderEventDetails.builder()
                 .flagsChanged(flagdProviderEvent.getFlagsChanged())
@@ -255,7 +262,7 @@ private void onReady() {
                 ProviderEventDetails.builder().message("connected to flagd").build());
     }
 
-    private void onError() {
+    private void onStale() {
         log.debug(
                 "Stream error. Emitting STALE, scheduling ERROR, and waiting {}s for connection to become available.",
                 gracePeriod);
@@ -270,7 +277,7 @@ private void onError() {
         if (!errorExecutor.isShutdown()) {
             errorTask = errorExecutor.schedule(
                     () -> {
-                        if (syncResources.getPreviousEvent() == ProviderEvent.PROVIDER_ERROR) {
+                        if (syncResources.getPreviousEvent() == ProviderEvent.PROVIDER_STALE) {
                             log.error(
                                     "Provider did not reconnect successfully within {}s. Emitting ERROR event...",
                                     gracePeriod);

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessResolver.java

@@ -77,6 +77,9 @@ public void init() throws Exception {
                                     storageStateChange.getSyncMetadata()));
                             log.debug("post onConnectionEvent.accept ProviderEvent.PROVIDER_CONFIGURATION_CHANGED");
                             break;
+                        case STALE:
+                            onConnectionEvent.accept(new FlagdProviderEvent(ProviderEvent.PROVIDER_STALE));
+                            break;
                         case ERROR:
                             onConnectionEvent.accept(new FlagdProviderEvent(ProviderEvent.PROVIDER_ERROR));
                             break;

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/FlagStore.java

@@ -138,6 +138,11 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
                     }
                     break;
                 case ERROR:
+                    if (!stateBlockingQueue.offer(new StorageStateChange(StorageState.STALE))) {
+                        log.warn("Failed to convey STALE status, queue is full");
+                    }
+                    break;
+                case FATAL:
                     if (!stateBlockingQueue.offer(new StorageStateChange(StorageState.ERROR))) {
                         log.warn("Failed to convey ERROR status, queue is full");
                     }

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

@@ -3,5 +3,6 @@
 /** Payload type emitted by {@link QueueSource}. */
 public enum QueuePayloadType {
     DATA,
-    ERROR
+    ERROR,
+    FATAL
 }

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

@@ -20,6 +20,7 @@
 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 +50,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> fatalStatusCodes;
 
     /**
      * Creates a new SyncStreamQueueSource responsible for observing the event stream.
@@ -65,6 +67,7 @@ public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderE
                 FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
         metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
                 .withWaitForReady();
+        fatalStatusCodes = options.getFatalStatusCodes();
     }
 
     // internal use only
@@ -82,6 +85,7 @@ protected SyncStreamQueueSource(
         flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
         metadataStub = blockingStubMock;
+        fatalStatusCodes = options.getFatalStatusCodes();
     }
 
     /** Initialize sync stream connector. */
@@ -132,11 +136,16 @@ private void observeSyncStream() {
                 }
 
                 log.debug("Initializing sync stream request");
-                SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue, shouldThrottle);
+                SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue, shouldThrottle, fatalStatusCodes);
                 try {
                     observer.metadata = getMetadata();
-                } catch (Exception metaEx) {
-                    // retry if getMetadata fails
+                } catch (StatusRuntimeException metaEx) {
+                    if (fatalStatusCodes.contains(metaEx.getStatus().getCode().name())) {
+                        //throw new FatalError("Failed to connect for metadata request, not retrying for error " + metaEx.getStatus());
+                        enqueueFatal("Fatal: Failed to connect for metadata request, not retrying for error " + metaEx.getStatus().getCode());
+                        return;
+                    }
+                    // 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 +155,13 @@ private void observeSyncStream() {
 
                 try {
                     syncFlags(observer);
-                } catch (Exception ex) {
+                } catch (StatusRuntimeException ex) {
+                    if (fatalStatusCodes.contains(ex.getStatus().getCode().toString())) {
+                        //throw new FatalError("Failed to connect for metadata request, not retrying for error " + ex.getStatus().getCode());
+                        enqueueFatal("Fatal: Failed to connect for metadata request, not retrying for error " + ex.getStatus().getCode());
+                        return;
+                    }
+                    // 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);
@@ -215,22 +230,34 @@ private void enqueueError(String message) {
         enqueueError(outgoingQueue, message);
     }
 
+    private void enqueueFatal(String message) {
+        enqueueFatal(outgoingQueue, message);
+    }
+
     private static void enqueueError(BlockingQueue<QueuePayload> queue, String message) {
         if (!queue.offer(new QueuePayload(QueuePayloadType.ERROR, message, null))) {
             log.error("Failed to convey ERROR status, queue is full");
         }
     }
 
+    private static void enqueueFatal(BlockingQueue<QueuePayload> queue, String message) {
+        if (!queue.offer(new QueuePayload(QueuePayloadType.FATAL, message, null))) {
+            log.error("Failed to convey FATAL status, queue is full");
+        }
+    }
+
     private static class SyncStreamObserver implements StreamObserver<SyncFlagsResponse> {
         private final BlockingQueue<QueuePayload> outgoingQueue;
         private final AtomicBoolean shouldThrottle;
         private final Awaitable done = new Awaitable();
+        private final List<String> fatalStatusCodes;
 
         private Struct metadata;
 
-        public SyncStreamObserver(BlockingQueue<QueuePayload> outgoingQueue, AtomicBoolean shouldThrottle) {
+        public SyncStreamObserver(BlockingQueue<QueuePayload> outgoingQueue, AtomicBoolean shouldThrottle, List<String> fatalStatusCodes) {
             this.outgoingQueue = outgoingQueue;
             this.shouldThrottle = shouldThrottle;
+            this.fatalStatusCodes = fatalStatusCodes;
         }
 
         @Override
@@ -248,9 +275,14 @@ public void onNext(SyncFlagsResponse syncFlagsResponse) {
         @Override
         public void onError(Throwable throwable) {
             try {
+                Status status = Status.fromThrowable(throwable);
                 String message = throwable != null ? throwable.getMessage() : "unknown";
                 log.debug("Stream error: {}, will restart", message, throwable);
-                enqueueError(outgoingQueue, String.format("Error from stream: %s", message));
+                if (fatalStatusCodes.contains(status.getCode())) {
+                    enqueueFatal(outgoingQueue, String.format("Error from stream: %s", message));
+                } else {
+                    enqueueError(outgoingQueue, String.format("Error from stream: %s", message));
+                }
 
                 // Set throttling flag to ensure backoff before retry
                 this.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.valueOf(clientState.toUpperCase()));
+    }
 }

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));
         }