feat(flagd): introduce fatalStatusCodes option

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

@@ -1,7 +1,10 @@
 package dev.openfeature.contrib.providers.flagd;
 
 import dev.openfeature.contrib.providers.flagd.resolver.rpc.cache.CacheType;
+import java.util.Arrays;
+import java.util.List;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 
 /** Helper class to hold configuration default values. */
@@ -37,6 +40,7 @@ public final class Config {
     static final String FLAGD_RETRY_BACKOFF_MAX_MS_VAR_NAME = "FLAGD_RETRY_BACKOFF_MAX_MS";
     static final String STREAM_DEADLINE_MS_ENV_VAR_NAME = "FLAGD_STREAM_DEADLINE_MS";
     static final String SOURCE_SELECTOR_ENV_VAR_NAME = "FLAGD_SOURCE_SELECTOR";
+    static final String FATAL_STATUS_CODES_ENV_VAR_NAME = "FLAGD_FATAL_STATUS_CODES";
     /**
      * Environment variable to fetch Provider id.
      *
@@ -93,6 +97,18 @@ static long fallBackToEnvOrDefault(String key, long defaultValue) {
         }
     }
 
+    static List<String> fallBackToEnvOrDefaultList(String key, List<String> defaultValue) {
+        try {
+            return System.getenv(key) != null
+                    ? Arrays.stream(System.getenv(key).split(","))
+                            .map(String::trim)
+                            .collect(Collectors.toList())
+                    : defaultValue;
+        } catch (Exception e) {
+            return defaultValue;
+        }
+    }
+
     static Resolver fromValueProvider(Function<String, String> provider) {
         final String resolverVar = provider.apply(RESOLVER_ENV_VAR);
         if (resolverVar == null) {

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

@@ -1,6 +1,7 @@
 package dev.openfeature.contrib.providers.flagd;
 
 import static dev.openfeature.contrib.providers.flagd.Config.fallBackToEnvOrDefault;
+import static dev.openfeature.contrib.providers.flagd.Config.fallBackToEnvOrDefaultList;
 import static dev.openfeature.contrib.providers.flagd.Config.fromValueProvider;
 
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueueSource;
@@ -122,6 +123,15 @@ 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 the provider transitions into fatal state upon first connection.
+     * Defaults to empty list
+     */
+    @Builder.Default
+    private List<String> fatalStatusCodes =
+            fallBackToEnvOrDefaultList(Config.FATAL_STATUS_CODES_ENV_VAR_NAME, List.of());
+
     /**
      * Selector to be used with flag sync gRPC contract.
      *

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

@@ -1,10 +1,10 @@
 package dev.openfeature.contrib.providers.flagd;
 
 import dev.openfeature.contrib.providers.flagd.resolver.Resolver;
-import dev.openfeature.contrib.providers.flagd.resolver.common.FlagdProviderEvent;
 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;
@@ -192,8 +192,9 @@ EvaluationContext getEnrichedContext() {
     }
 
     @SuppressWarnings("checkstyle:fallthrough")
-    private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
-        log.debug("FlagdProviderEvent event {} ", flagdProviderEvent.getEvent());
+    private void onProviderEvent(
+            ProviderEvent providerEvent, ProviderEventDetails providerEventDetails, Structure syncMetadata) {
+        log.debug("FlagdProviderEvent event {} ", providerEvent);
         synchronized (syncResources) {
             /*
              * We only use Error and Ready as previous states.
@@ -204,10 +205,10 @@ private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
              * forward a configuration changed to the ready, if we are not in the ready
              * state.
              */
-            switch (flagdProviderEvent.getEvent()) {
+            switch (providerEvent) {
                 case PROVIDER_CONFIGURATION_CHANGED:
                     if (syncResources.getPreviousEvent() == ProviderEvent.PROVIDER_READY) {
-                        onConfigurationChanged(flagdProviderEvent);
+                        emit(providerEvent, providerEventDetails);
                         break;
                     }
                 // intentional fall through
@@ -216,33 +217,30 @@ private void onProviderEvent(FlagdProviderEvent flagdProviderEvent) {
                      * Sync metadata is used to enrich the context, and is immutable in flagd,
                      * so we only need it to be fetched once at READY.
                      */
-                    if (flagdProviderEvent.getSyncMetadata() != null) {
-                        syncResources.setEnrichedContext(contextEnricher.apply(flagdProviderEvent.getSyncMetadata()));
+                    if (syncMetadata != null) {
+                        syncResources.setEnrichedContext(contextEnricher.apply(syncMetadata));
                     }
                     onReady();
                     syncResources.setPreviousEvent(ProviderEvent.PROVIDER_READY);
                     break;
-
                 case PROVIDER_ERROR:
+                    if (providerEventDetails != null
+                            && providerEventDetails.getErrorCode() == ErrorCode.PROVIDER_FATAL) {
+                        onFatal();
+                        break;
+                    }
+
                     if (syncResources.getPreviousEvent() != ProviderEvent.PROVIDER_ERROR) {
                         onError();
                         syncResources.setPreviousEvent(ProviderEvent.PROVIDER_ERROR);
                     }
                     break;
-
                 default:
-                    log.warn("Unknown event {}", flagdProviderEvent.getEvent());
+                    log.warn("Unknown event {}", providerEvent);
             }
         }
     }
 
-    private void onConfigurationChanged(FlagdProviderEvent flagdProviderEvent) {
-        this.emitProviderConfigurationChanged(ProviderEventDetails.builder()
-                .flagsChanged(flagdProviderEvent.getFlagsChanged())
-                .message("configuration changed")
-                .build());
-    }
-
     private void onReady() {
         if (syncResources.initialize()) {
             log.info("Initialized FlagdProvider");
@@ -284,4 +282,17 @@ private void onError() {
                     TimeUnit.SECONDS);
         }
     }
+
+    private void onFatal() {
+        if (errorTask != null && !errorTask.isCancelled()) {
+            errorTask.cancel(false);
+        }
+        this.syncResources.setFatal(true);
+
+        this.emitProviderError(ProviderEventDetails.builder()
+                .errorCode(ErrorCode.PROVIDER_FATAL)
+                .build());
+
+        shutdown();
+    }
 }

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

@@ -3,6 +3,7 @@
 import dev.openfeature.sdk.EvaluationContext;
 import dev.openfeature.sdk.ImmutableContext;
 import dev.openfeature.sdk.ProviderEvent;
+import dev.openfeature.sdk.exceptions.FatalError;
 import dev.openfeature.sdk.exceptions.GeneralError;
 import lombok.Getter;
 import lombok.Setter;
@@ -16,8 +17,11 @@ class FlagdProviderSyncResources {
     @Setter
     private volatile ProviderEvent previousEvent = null;
 
+    @Setter
+    private volatile boolean isFatal;
+
     private volatile EvaluationContext enrichedContext = new ImmutableContext();
-    private volatile boolean initialized;
+    private volatile boolean isInitialized;
     private volatile boolean isShutDown;
 
     public void setEnrichedContext(EvaluationContext context) {
@@ -31,32 +35,40 @@ public void setEnrichedContext(EvaluationContext context) {
      * @return true iff this was the first call to {@code initialize()}
      */
     public synchronized boolean initialize() {
-        if (this.initialized) {
+        if (this.isInitialized) {
             return false;
         }
-        this.initialized = true;
+        this.isInitialized = true;
+        this.isFatal = false;
         this.notifyAll();
         return true;
     }
 
     /**
-     * Blocks the calling thread until either {@link FlagdProviderSyncResources#initialize()} or
-     * {@link FlagdProviderSyncResources#shutdown()} is called or the deadline is exceeded, whatever happens first. If
-     * {@link FlagdProviderSyncResources#initialize()} has been executed before {@code waitForInitialization(long)} is
-     * called, it will return instantly. If the deadline is exceeded, a GeneralError will be thrown.
-     * If {@link FlagdProviderSyncResources#shutdown()} is called in the meantime, an {@link IllegalStateException} will
+     * Blocks the calling thread until either
+     * {@link FlagdProviderSyncResources#initialize()} or
+     * {@link FlagdProviderSyncResources#shutdown()} is called or the deadline is
+     * exceeded, whatever happens first. If
+     * {@link FlagdProviderSyncResources#initialize()} has been executed before
+     * {@code waitForInitialization(long)} is
+     * called, it will return instantly. If the deadline is exceeded, a GeneralError
+     * will be thrown.
+     * If {@link FlagdProviderSyncResources#shutdown()} is called in the meantime,
+     * an {@link IllegalStateException} will
      * be thrown. Otherwise, the method will return cleanly.
      *
      * @param deadline the maximum time in ms to wait
-     * @throws GeneralError          when the deadline is exceeded before
-     *                               {@link FlagdProviderSyncResources#initialize()} is called on this object
-     * @throws IllegalStateException when {@link FlagdProviderSyncResources#shutdown()} is called or has been called on
-     *                               this object
+     * @throws GeneralError when the deadline is exceeded before
+     *                      {@link FlagdProviderSyncResources#initialize()} is
+     *                      called on this object, or when
+     *                      {@link FlagdProviderSyncResources#shutdown()}
+     * @throws FatalError   when the provider has been marked as fatal during
+     *                      shutdown
      */
     public void waitForInitialization(long deadline) {
         long start = System.currentTimeMillis();
         long end = start + deadline;
-        while (!initialized && !isShutDown) {
+        while (!isInitialized && !isShutDown) {
             long now = System.currentTimeMillis();
             // if wait(0) is called, the thread would wait forever, so we abort when this would happen
             if (now >= end) {
@@ -68,7 +80,7 @@ public void waitForInitialization(long deadline) {
                 if (isShutDown) {
                     break;
                 }
-                if (initialized) { // might have changed in the meantime
+                if (isInitialized) { // might have changed in the meantime
                     return;
                 }
                 try {
@@ -80,7 +92,11 @@ public void waitForInitialization(long deadline) {
             }
         }
         if (isShutDown) {
-            throw new IllegalStateException("Already shut down");
+            String msg = "Already shut down due to previous error.";
+            if (isFatal) {
+                throw new FatalError(msg);
+            }
+            throw new GeneralError(msg);
         }
     }
 

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

@@ -4,7 +4,6 @@
 
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.Resolver;
-import dev.openfeature.contrib.providers.flagd.resolver.common.FlagdProviderEvent;
 import dev.openfeature.contrib.providers.flagd.resolver.process.model.FeatureFlag;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.FlagStore;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.Storage;
@@ -20,13 +19,15 @@
 import dev.openfeature.sdk.ImmutableMetadata;
 import dev.openfeature.sdk.ProviderEvaluation;
 import dev.openfeature.sdk.ProviderEvent;
+import dev.openfeature.sdk.ProviderEventDetails;
 import dev.openfeature.sdk.Reason;
+import dev.openfeature.sdk.Structure;
 import dev.openfeature.sdk.Value;
 import dev.openfeature.sdk.exceptions.GeneralError;
 import dev.openfeature.sdk.exceptions.ParseError;
 import dev.openfeature.sdk.exceptions.TypeMismatchError;
+import dev.openfeature.sdk.internal.TriConsumer;
 import java.util.Map;
-import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 
@@ -38,7 +39,7 @@
 @Slf4j
 public class InProcessResolver implements Resolver {
     private final Storage flagStore;
-    private final Consumer<FlagdProviderEvent> onConnectionEvent;
+    private final TriConsumer<ProviderEvent, ProviderEventDetails, Structure> onConnectionEvent;
     private final Operator operator;
     private final String scope;
     private final QueueSource queueSource;
@@ -52,7 +53,8 @@ public class InProcessResolver implements Resolver {
      * @param onConnectionEvent lambda which handles changes in the
      *                          connection/stream
      */
-    public InProcessResolver(FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
+    public InProcessResolver(
+            FlagdOptions options, TriConsumer<ProviderEvent, ProviderEventDetails, Structure> onConnectionEvent) {
         this.queueSource = getQueueSource(options);
         this.flagStore = new FlagStore(queueSource);
         this.onConnectionEvent = onConnectionEvent;
@@ -73,14 +75,29 @@ public void init() throws Exception {
                     switch (storageStateChange.getStorageState()) {
                         case OK:
                             log.debug("onConnectionEvent.accept ProviderEvent.PROVIDER_CONFIGURATION_CHANGED");
-                            onConnectionEvent.accept(new FlagdProviderEvent(
+
+                            var eventDetails = ProviderEventDetails.builder()
+                                    .flagsChanged(storageStateChange.getChangedFlagsKeys())
+                                    .message("configuration changed")
+                                    .build();
+
+                            onConnectionEvent.accept(
                                     ProviderEvent.PROVIDER_CONFIGURATION_CHANGED,
-                                    storageStateChange.getChangedFlagsKeys(),
-                                    storageStateChange.getSyncMetadata()));
+                                    eventDetails,
+                                    storageStateChange.getSyncMetadata());
+
                             log.debug("post onConnectionEvent.accept ProviderEvent.PROVIDER_CONFIGURATION_CHANGED");
                             break;
+                        case STALE:
+                            onConnectionEvent.accept(ProviderEvent.PROVIDER_ERROR, null, null);
+                            break;
                         case ERROR:
-                            onConnectionEvent.accept(new FlagdProviderEvent(ProviderEvent.PROVIDER_ERROR));
+                            onConnectionEvent.accept(
+                                    ProviderEvent.PROVIDER_ERROR,
+                                    ProviderEventDetails.builder()
+                                            .errorCode(ErrorCode.PROVIDER_FATAL)
+                                            .build(),
+                                    null);
                             break;
                         default:
                             log.warn(String.format(

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

@@ -11,6 +11,7 @@
 import dev.openfeature.sdk.ImmutableStructure;
 import dev.openfeature.sdk.Structure;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -109,7 +110,7 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
             switch (payload.getType()) {
                 case DATA:
                     try {
-                        List<String> changedFlagsKeys;
+                        List<String> changedFlagsKeys = Collections.emptyList();
                         ParsingResult parsingResult = FlagParser.parseString(payload.getFlagData(), throwIfInvalid);
                         Map<String, FeatureFlag> flagMap = parsingResult.getFlags();
                         Map<String, Object> flagSetMetadataMap = parsingResult.getFlagSetMetadata();
@@ -133,13 +134,19 @@ private void streamerListener(final QueueSource connector) throws InterruptedExc
                         // catch all exceptions and avoid stream listener interruptions
                         log.warn("Invalid flag sync payload from connector", e);
                         if (!stateBlockingQueue.offer(new StorageStateChange(StorageState.STALE))) {
-                            log.warn("Failed to convey STALE status, queue is full");
+                            log.warn("Failed to convey TRANSIENT_ERROR status, queue is full");
                         }
                     }
                     break;
                 case ERROR:
+                    if (!stateBlockingQueue.offer(new StorageStateChange(StorageState.STALE))) {
+                        log.warn("Failed to convey TRANSIENT_ERROR status, queue is full");
+                    }
+                    break;
+                case SHUTDOWN:
+                    shutdown();
                     if (!stateBlockingQueue.offer(new StorageStateChange(StorageState.ERROR))) {
-                        log.warn("Failed to convey ERROR status, queue is full");
+                        log.warn("Failed to convey FATAL_ERROR status, queue is full");
                     }
                     break;
                 default:

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

@@ -1,10 +1,10 @@
 package dev.openfeature.contrib.providers.flagd.resolver.process.storage;
 
-/** Satus of the storage. */
+/** Status of the storage. */
 public enum StorageState {
     /** Storage is upto date and working as expected. */
     OK,
-    /** Storage has gone stale(most recent sync failed). May get to OK status with next sync. */
+    /** Storage has gone stale (most recent sync failed). May get to OK status with next sync. */
     STALE,
     /** Storage is in an unrecoverable error stage. */
     ERROR,