fix(flagd): use upstream SyncStreamQueueSource constructor

aepfli · claude · aepfli · commit 0b01b9b02321 · 2025-12-29T11:56:57.000+01:00
Update InProcessWasmResolver to use the single-parameter SyncStreamQueueSource constructor that matches upstream. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
diff --git a/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessWasmResolver.java b/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/InProcessWasmResolver.java
@@ -6,11 +6,8 @@
 import static dev.openfeature.contrib.providers.flagd.resolver.process.FlagdWasmRuntime.getModule;
 
 import com.dylibso.chicory.runtime.ExportFunction;
-import com.dylibso.chicory.runtime.HostFunction;
 import com.dylibso.chicory.runtime.Instance;
 import com.dylibso.chicory.runtime.Memory;
-import com.dylibso.chicory.wasm.types.FunctionType;
-import com.dylibso.chicory.wasm.types.ValType;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -57,7 +54,6 @@ public class InProcessWasmResolver implements Resolver {
     }
 
     private final Consumer<FlagdProviderEvent> onConnectionEvent;
-    private final String scope;
     private final QueueSource connector;
     private Thread stateWatcher;
     private final ExportFunction validationMode;
@@ -78,8 +74,7 @@ public class InProcessWasmResolver implements Resolver {
      */
     public InProcessWasmResolver(FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
         this.onConnectionEvent = onConnectionEvent;
-        this.connector = getConnector(options, onConnectionEvent);
-        this.scope = options.getSelector();
+        this.connector = getConnector(options);
 
         var store = createStoreWithHostFunctions();
 
@@ -210,14 +205,14 @@ public ProviderEvaluation<Value> objectEvaluation(String key, Value defaultValue
                 .build();
     }
 
-    static QueueSource getConnector(final FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
+    static QueueSource getConnector(final FlagdOptions options) {
         if (options.getCustomConnector() != null) {
             return options.getCustomConnector();
         }
         return options.getOfflineFlagSourcePath() != null
                         && !options.getOfflineFlagSourcePath().isEmpty()
                 ? new FileQueueSource(options.getOfflineFlagSourcePath(), options.getOfflinePollIntervalMs())
-                : new SyncStreamQueueSource(options, onConnectionEvent);
+                : new SyncStreamQueueSource(options);
     }
 
     private <T> ProviderEvaluation<T> resolve(Class<T> type, String key, EvaluationContext ctx) {
diff --git a/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSource.java b/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSource.java
@@ -4,7 +4,6 @@
 import dev.openfeature.contrib.providers.flagd.FlagdOptions;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelBuilder;
 import dev.openfeature.contrib.providers.flagd.resolver.common.ChannelConnector;
-import dev.openfeature.contrib.providers.flagd.resolver.common.FlagdProviderEvent;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayload;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueuePayloadType;
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.QueueSource;
@@ -24,16 +23,16 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 
 /**
- * Implements the {@link QueueSource} contract and emit flags obtained from flagd sync gRPC contract.
+ * Implements the {@link QueueSource} contract and emit flags obtained from
+ * flagd sync gRPC contract.
  */
 @Slf4j
 @SuppressFBWarnings(
         value = {"EI_EXPOSE_REP"},
-        justification = "Random is used to generate a variation & flag configurations require exposing")
+        justification = "We need to expose the BlockingQueue to allow consumers to read from it")
 public class SyncStreamQueueSource implements QueueSource {
     private static final int QUEUE_SIZE = 5;
 
@@ -45,26 +44,42 @@ public class SyncStreamQueueSource implements QueueSource {
     private final String selector;
     private final String providerId;
     private final boolean syncMetadataDisabled;
-    private final ChannelConnector channelConnector;
+    private final boolean reinitializeOnError;
+    private final FlagdOptions options;
     private final BlockingQueue<QueuePayload> outgoingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
-    private final FlagSyncServiceStub flagSyncStub;
-    private final FlagSyncServiceBlockingStub metadataStub;
+    private volatile GrpcComponents grpcComponents;
 
     /**
-     * Creates a new SyncStreamQueueSource responsible for observing the event stream.
+     * Container for gRPC components to ensure atomicity during reinitialization.
+     * All three components are updated together to prevent consumers from seeing
+     * an inconsistent state where components are from different channel instances.
      */
-    public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderEvent> onConnectionEvent) {
+    private static class GrpcComponents {
+        final ChannelConnector channelConnector;
+        final FlagSyncServiceStub flagSyncStub;
+        final FlagSyncServiceBlockingStub metadataStub;
+
+        GrpcComponents(ChannelConnector connector, FlagSyncServiceStub stub, FlagSyncServiceBlockingStub blockingStub) {
+            this.channelConnector = connector;
+            this.flagSyncStub = stub;
+            this.metadataStub = blockingStub;
+        }
+    }
+
+    /**
+     * Creates a new SyncStreamQueueSource responsible for observing the event
+     * stream.
+     */
+    public SyncStreamQueueSource(final FlagdOptions options) {
         streamDeadline = options.getStreamDeadlineMs();
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
         maxBackoffMs = options.getRetryBackoffMaxMs();
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        channelConnector = new ChannelConnector(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
-        flagSyncStub =
-                FlagSyncServiceGrpc.newStub(channelConnector.getChannel()).withWaitForReady();
-        metadataStub = FlagSyncServiceGrpc.newBlockingStub(channelConnector.getChannel())
-                .withWaitForReady();
+        reinitializeOnError = options.isReinitializeOnError();
+        this.options = options;
+        initializeChannelComponents();
     }
 
     // internal use only
@@ -77,16 +92,54 @@ protected SyncStreamQueueSource(
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
-        channelConnector = connectorMock;
         maxBackoffMs = options.getRetryBackoffMaxMs();
-        flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
-        metadataStub = blockingStubMock;
+        reinitializeOnError = options.isReinitializeOnError();
+        this.options = options;
+        this.grpcComponents = new GrpcComponents(connectorMock, stubMock, blockingStubMock);
+    }
+
+    /** Initialize channel connector and stubs. */
+    private synchronized void initializeChannelComponents() {
+        ChannelConnector newConnector = new ChannelConnector(options, ChannelBuilder.nettyChannel(options));
+        FlagSyncServiceStub newFlagSyncStub =
+                FlagSyncServiceGrpc.newStub(newConnector.getChannel()).withWaitForReady();
+        FlagSyncServiceBlockingStub newMetadataStub =
+                FlagSyncServiceGrpc.newBlockingStub(newConnector.getChannel()).withWaitForReady();
+
+        // atomic assignment of all components as a single unit
+        grpcComponents = new GrpcComponents(newConnector, newFlagSyncStub, newMetadataStub);
+    }
+
+    /** Reinitialize channel connector and stubs on error. */
+    public synchronized void reinitializeChannelComponents() {
+        if (!reinitializeOnError || shutdown.get()) {
+            return;
+        }
+
+        log.info("Reinitializing channel gRPC components in attempt to restore stream.");
+        GrpcComponents oldComponents = grpcComponents;
+
+        try {
+            // create new channel components first
+            initializeChannelComponents();
+        } catch (Exception e) {
+            log.error("Failed to reinitialize channel components", e);
+            return;
+        }
+
+        // shutdown old connector after successful reinitialization
+        if (oldComponents != null && oldComponents.channelConnector != null) {
+            try {
+                oldComponents.channelConnector.shutdown();
+            } catch (Exception e) {
+                log.debug("Error shutting down old channel connector during reinitialization", e);
+            }
+        }
     }
 
     /** Initialize sync stream connector. */
     public void init() throws Exception {
-        channelConnector.initialize();
         Thread listener = new Thread(this::observeSyncStream);
         listener.setDaemon(true);
         listener.start();
@@ -109,7 +162,7 @@ public void shutdown() throws InterruptedException {
             log.debug("Shutdown already in progress or completed");
             return;
         }
-        this.channelConnector.shutdown();
+        grpcComponents.channelConnector.shutdown();
     }
 
     /** Contains blocking calls, to be used concurrently. */
@@ -159,13 +212,14 @@ private void observeSyncStream() {
         log.info("Shutdown invoked, exiting event stream listener");
     }
 
-    // TODO: remove the metadata call entirely after https://github.com/open-feature/flagd/issues/1584
+    // TODO: remove the metadata call entirely after
+    // https://github.com/open-feature/flagd/issues/1584
     private Struct getMetadata() {
         if (syncMetadataDisabled) {
             return null;
         }
 
-        FlagSyncServiceBlockingStub localStub = metadataStub;
+        FlagSyncServiceBlockingStub localStub = grpcComponents.metadataStub;
 
         if (deadline > 0) {
             localStub = localStub.withDeadlineAfter(deadline, TimeUnit.MILLISECONDS);
@@ -180,7 +234,8 @@ private Struct getMetadata() {
 
             return null;
         } catch (StatusRuntimeException e) {
-            // In newer versions of flagd, metadata is part of the sync stream. If the method is unimplemented, we
+            // In newer versions of flagd, metadata is part of the sync stream. If the
+            // method is unimplemented, we
             // can ignore the error
             if (e.getStatus() != null
                     && Status.Code.UNIMPLEMENTED.equals(e.getStatus().getCode())) {
@@ -192,7 +247,7 @@ private Struct getMetadata() {
     }
 
     private void syncFlags(SyncStreamObserver streamObserver) {
-        FlagSyncServiceStub localStub = flagSyncStub; // don't mutate the stub
+        FlagSyncServiceStub localStub = grpcComponents.flagSyncStub; // don't mutate the stub
         if (streamDeadline > 0) {
             localStub = localStub.withDeadlineAfter(streamDeadline, TimeUnit.MILLISECONDS);
         }
