fixup: add sleep delay to prevent busy loop

toddbaert · toddbaert · commit b4c62f71c41b · 2025-10-13T14:47:12.000-04:00
Signed-off-by: Todd Baert &lt;todd.baert@dynatrace.com&gt;
diff --git a/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/Config.java b/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/Config.java
@@ -14,6 +14,7 @@ public final class Config {
     static final String DEFAULT_HOST = "localhost";
 
     static final int DEFAULT_DEADLINE = 500;
+    static final int DEFAULT_MAX_RETRY_BACKOFF = 5;
     static final int DEFAULT_STREAM_DEADLINE_MS = 10 * 60 * 1000;
     static final int DEFAULT_STREAM_RETRY_GRACE_PERIOD = 5;
     static final int DEFAULT_MAX_CACHE_SIZE = 1000;
@@ -32,6 +33,7 @@ public final class Config {
     static final String MAX_EVENT_STREAM_RETRIES_ENV_VAR_NAME = "FLAGD_MAX_EVENT_STREAM_RETRIES";
     static final String BASE_EVENT_STREAM_RETRY_BACKOFF_MS_ENV_VAR_NAME = "FLAGD_RETRY_BACKOFF_MS";
     static final String DEADLINE_MS_ENV_VAR_NAME = "FLAGD_DEADLINE_MS";
+    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 SOURCE_PROVIDER_ID_ENV_VAR_NAME = "FLAGD_SOURCE_PROVIDER_ID";
diff --git a/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java b/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/FlagdOptions.java
@@ -97,10 +97,18 @@ public class FlagdOptions {
     @Builder.Default
     private int deadline = fallBackToEnvOrDefault(Config.DEADLINE_MS_ENV_VAR_NAME, Config.DEFAULT_DEADLINE);
 
+    /**
+     * Max stream retry backoff in milliseconds.
+     */
+    @Builder.Default
+    private int retryBackoffMaxMs =
+            fallBackToEnvOrDefault(Config.FLAGD_RETRY_BACKOFF_MAX_MS_VAR_NAME, Config.DEFAULT_MAX_RETRY_BACKOFF);
+
     /**
      * Streaming connection deadline in milliseconds.
      * Set to 0 to disable the deadline.
-     * Defaults to 600000 (10 minutes); recommended to prevent infrastructure from killing idle connections.
+     * Defaults to 600000 (10 minutes); recommended to prevent infrastructure from
+     * killing idle connections.
      */
     @Builder.Default
     private int streamDeadlineMs =
diff --git a/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/common/ChannelBuilder.java b/providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/common/ChannelBuilder.java
@@ -26,25 +26,11 @@
 /** gRPC channel builder helper. */
 public class ChannelBuilder {
 
-    @SuppressWarnings({"unchecked", "rawtypes"})
-    private static final Map<String, ?> DEFAULT_RETRY_POLICY = new HashMap() {
-        {
-            // 1s + 2s + 4s
-            put("maxAttempts", 3.0); // types used here are important, need to be doubles
-            put("initialBackoff", "1s");
-            put("maxBackoff", "5s");
-            put("backoffMultiplier", 2.0);
-            // status codes to retry on:
-            put(
-                    "retryableStatusCodes",
-                    Arrays.asList(
-                            /*
-                             * Only states UNAVAILABLE and UNKNOWN should be retried. All
-                             * other failure states will probably not be resolved with a simple retry.
-                             */
-                            Code.UNAVAILABLE.toString(), Code.UNKNOWN.toString()));
-        }
-    };
+    /**
+     * Maximum retry backoff in seconds.
+     * Also used by SyncStreamQueueSource to control outer loop retry delay.
+     */
+    public static final int MAX_RETRY_BACKOFF_SECONDS = 5;
 
     /**
      * Controls retry (not-reconnection) policy for failed RPCs.
@@ -67,46 +53,22 @@ public class ChannelBuilder {
                                             put("service", "flagd.evaluation.v1.Service");
                                         }
                                     }));
-                    put("retryPolicy", DEFAULT_RETRY_POLICY);
-                }
-
-                {
-                    put("name", Arrays.asList(new HashMap() {
-                        {
-                            put("service", "flagd.sync.v1.FlagSyncService");
-                            put("method", "SyncFlags");
-                        }
-                    }));
-                    put("retryPolicy", new HashMap(DEFAULT_RETRY_POLICY) {
+                    put("retryPolicy", new HashMap() {
                         {
-                            // 1s + 2s + 4s + 5s + 5s + 5s + 5s + 5s + 5s + 5s
-                            put("maxAttempts", 12.0);
-                            // for streaming Retry on more status codes
+                            // 1 + 2 + 4
+                            put("maxAttempts", 3.0); // types used here are important, need to be doubles
+                            put("initialBackoff", "1s");
+                            put("maxBackoff", String.format("%ds", MAX_RETRY_BACKOFF_SECONDS));
+                            put("backoffMultiplier", 2.0);
+                            // status codes to retry on:
                             put(
                                     "retryableStatusCodes",
                                     Arrays.asList(
                                             /*
-                                             * All codes are retryable except OK and DEADLINE_EXCEEDED since
-                                             * any others not listed here cause a very tight loop of retries.
-                                             * DEADLINE_EXCEEDED is typically a result of a client specified
-                                             * deadline,and definitionally should not result in a tight loop
-                                             * (it's a timeout).
+                                             * As per gRPC spec, the following status codes are safe to retry:
+                                             * UNAVAILABLE, UNKNOWN,
                                              */
-                                            Code.CANCELLED.toString(),
-                                            Code.UNKNOWN.toString(),
-                                            Code.INVALID_ARGUMENT.toString(),
-                                            Code.NOT_FOUND.toString(),
-                                            Code.ALREADY_EXISTS.toString(),
-                                            Code.PERMISSION_DENIED.toString(),
-                                            Code.RESOURCE_EXHAUSTED.toString(),
-                                            Code.FAILED_PRECONDITION.toString(),
-                                            Code.ABORTED.toString(),
-                                            Code.OUT_OF_RANGE.toString(),
-                                            Code.UNIMPLEMENTED.toString(),
-                                            Code.INTERNAL.toString(),
-                                            Code.UNAVAILABLE.toString(),
-                                            Code.DATA_LOSS.toString(),
-                                            Code.UNAUTHENTICATED.toString()));
+                                            Code.UNKNOWN.toString(), Code.UNAVAILABLE.toString()));
                         }
                     });
                 }
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
@@ -35,11 +35,17 @@
         value = {"PREDICTABLE_RANDOM", "EI_EXPOSE_REP"},
         justification = "Random is used to generate a variation & flag configurations require exposing")
 public class SyncStreamQueueSource implements QueueSource {
+    /**
+     * Delay between retries of the outer loop, in milliseconds. This is derived from the channel retry (5s)
+     */
+    public static final int RETRY_LOOP_DELAY_MS = ChannelBuilder.MAX_RETRY_BACKOFF_SECONDS * 1000;
+
     private static final int QUEUE_SIZE = 5;
 
     private final AtomicBoolean shutdown = new AtomicBoolean(false);
     private final int streamDeadline;
     private final int deadline;
+    private final int maxBackoffMs;
     private final String selector;
     private final String providerId;
     private final boolean syncMetadataDisabled;
@@ -56,6 +62,7 @@ public SyncStreamQueueSource(final FlagdOptions options, Consumer<FlagdProviderE
         deadline = options.getDeadline();
         selector = options.getSelector();
         providerId = options.getProviderId();
+        maxBackoffMs = options.getRetryBackoffMaxMs();
         syncMetadataDisabled = options.isSyncMetadataDisabled();
         channelConnector = new ChannelConnector(options, onConnectionEvent, ChannelBuilder.nettyChannel(options));
         flagSyncStub =
@@ -75,6 +82,7 @@ protected SyncStreamQueueSource(
         selector = options.getSelector();
         providerId = options.getProviderId();
         channelConnector = connectorMock;
+        maxBackoffMs = options.getRetryBackoffMaxMs();
         flagSyncStub = stubMock;
         syncMetadataDisabled = options.isSyncMetadataDisabled();
         metadataStub = blockingStubMock;
@@ -110,26 +118,32 @@ private void observeSyncStream() {
         log.info("Initializing sync stream observer");
 
         // outer loop for re-issuing the stream request
-        // "waitForReady" on the channel, plus our retry policy slow this loop down in error conditions
+        // "waitForReady" on the channel, plus our retry policy slow this loop down in
+        // error conditions
         while (!shutdown.get()) {
-            log.debug("Initializing sync stream request");
-            SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue);
-
-            try {
-                observer.metadata = getMetadata();
-            } catch (Exception metaEx) {
-                // retry if getMetadata fails
-                String message = metaEx.getMessage();
-                log.debug("Metadata request error: {}, will restart", message, metaEx);
-                enqueueError(String.format("Error in getMetadata request: %s", message));
-                continue;
-            }
-
             try {
-                syncFlags(observer);
-            } catch (Exception ex) {
-                log.error("Unexpected sync stream exception, will restart.", ex);
-                enqueueError(String.format("Error in syncStream: %s", ex.getMessage()));
+                log.debug("Initializing sync stream request");
+                SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue);
+                try {
+                    observer.metadata = getMetadata();
+                } catch (Exception metaEx) {
+                    // retry if getMetadata fails
+                    String message = metaEx.getMessage();
+                    log.debug("Metadata request error: {}, will restart", message, metaEx);
+                    enqueueError(String.format("Error in getMetadata request: %s", message));
+                    Thread.sleep(this.maxBackoffMs);
+                    continue;
+                }
+
+                try {
+                    syncFlags(observer);
+                } catch (Exception ex) {
+                    log.error("Unexpected sync stream exception, will restart.", ex);
+                    enqueueError(String.format("Error in syncStream: %s", ex.getMessage()));
+                    Thread.sleep(this.maxBackoffMs);
+                }
+            } catch (InterruptedException ie) {
+                log.debug("Stream loop interrupted, most likely shutdown was invoked", ie);
             }
         }
 
diff --git a/providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java b/providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java
@@ -23,6 +23,7 @@
 import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
 import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsRequest;
 import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse;
+import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -35,11 +36,12 @@ class SyncStreamQueueSourceTest {
     private ChannelConnector mockConnector;
     private FlagSyncServiceBlockingStub blockingStub;
     private FlagSyncServiceStub stub;
+    private FlagSyncServiceStub errorStub;
     private StreamObserver<SyncFlagsResponse> observer;
     private CountDownLatch latch; // used to wait for observer to be initialized
 
     @BeforeEach
-    public void init() throws Exception {
+    public void setup() throws Exception {
         blockingStub = mock(FlagSyncServiceBlockingStub.class);
         when(blockingStub.withDeadlineAfter(anyLong(), any())).thenReturn(blockingStub);
         when(blockingStub.getMetadata(any())).thenReturn(GetMetadataResponse.getDefaultInstance());
@@ -57,22 +59,54 @@ public void init() throws Exception {
                 })
                 .when(stub)
                 .syncFlags(any(SyncFlagsRequest.class), any(StreamObserver.class)); // Mock the initialize
-        // method
+
+        errorStub = mock(FlagSyncServiceStub.class);
+        when(errorStub.withDeadlineAfter(anyLong(), any())).thenReturn(errorStub);
+        doAnswer((Answer<Void>) invocation -> {
+                    Object[] args = invocation.getArguments();
+                    observer = (StreamObserver<SyncFlagsResponse>) args[1];
+                    latch.countDown();
+                    throw new StatusRuntimeException(io.grpc.Status.NOT_FOUND);
+                })
+                .when(errorStub)
+                .syncFlags(any(SyncFlagsRequest.class), any(StreamObserver.class)); // Mock the initialize
+    }
+
+    @Test
+    void initError_DoesNotBusyWait() throws Exception {
+        // make sure we do not spin in a busy loop on errors
+
+        int maxBackoffMs = 1000;
+        SyncStreamQueueSource queueSource = new SyncStreamQueueSource(
+                FlagdOptions.builder().retryBackoffMaxMs(maxBackoffMs).build(), mockConnector, errorStub, blockingStub);
+        latch = new CountDownLatch(1);
+        queueSource.init();
+        latch.await();
+
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
+        QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
+        assertNotNull(payload);
+        assertEquals(QueuePayloadType.ERROR, payload.getType());
+        Thread.sleep(maxBackoffMs + (maxBackoffMs / 2)); // wait 1.5x our delay for reties
+
+        // should have retried the stream (2 calls); initial + 1 retry
+        // it's very important that the retry count is low, to confirm no busy-loop
+        verify(errorStub, times(2)).syncFlags(any(), any());
     }
 
     @Test
     void onNextEnqueuesDataPayload() throws Exception {
-        SyncStreamQueueSource connector =
+        SyncStreamQueueSource queueSource =
                 new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
         latch = new CountDownLatch(1);
-        connector.init();
+        queueSource.init();
         latch.await();
 
         // fire onNext (data) event
         observer.onNext(SyncFlagsResponse.newBuilder().build());
 
         // should enqueue data payload
-        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
         assertNotNull(payload.getSyncContext());
@@ -84,17 +118,17 @@ void onNextEnqueuesDataPayload() throws Exception {
     @Test
     void onNextEnqueuesDataPayloadMetadataDisabled() throws Exception {
         // disable GetMetadata call
-        SyncStreamQueueSource connector = new SyncStreamQueueSource(
+        SyncStreamQueueSource queueSource = new SyncStreamQueueSource(
                 FlagdOptions.builder().syncMetadataDisabled(true).build(), mockConnector, stub, blockingStub);
         latch = new CountDownLatch(1);
-        connector.init();
+        queueSource.init();
         latch.await();
 
         // fire onNext (data) event
         observer.onNext(SyncFlagsResponse.newBuilder().build());
 
         // should enqueue data payload
-        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
         assertNull(payload.getSyncContext());
@@ -108,10 +142,10 @@ void onNextEnqueuesDataPayloadMetadataDisabled() throws Exception {
     @Test
     void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
         // disable GetMetadata call
-        SyncStreamQueueSource connector =
+        SyncStreamQueueSource queueSource =
                 new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
         latch = new CountDownLatch(1);
-        connector.init();
+        queueSource.init();
         latch.await();
 
         // fire onNext (data) event
@@ -120,7 +154,7 @@ void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
                 SyncFlagsResponse.newBuilder().setSyncContext(syncContext).build());
 
         // should enqueue data payload
-        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
         assertEquals(syncContext, payload.getSyncContext());
@@ -131,18 +165,18 @@ void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
 
     @Test
     void onErrorEnqueuesDataPayload() throws Exception {
-        SyncStreamQueueSource connector =
+        SyncStreamQueueSource queueSource =
                 new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
         latch = new CountDownLatch(1);
-        connector.init();
+        queueSource.init();
         latch.await();
 
         // fire onError event and reset latch
         latch = new CountDownLatch(1);
         observer.onError(new Exception("fake exception"));
 
         // should enqueue error payload
-        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
         QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
         assertNotNull(payload);
         assertEquals(QueuePayloadType.ERROR, payload.getType());
@@ -153,18 +187,18 @@ void onErrorEnqueuesDataPayload() throws Exception {
 
     @Test
     void onCompletedEnqueuesDataPayload() throws Exception {
-        SyncStreamQueueSource connector =
+        SyncStreamQueueSource queueSource =
                 new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
         latch = new CountDownLatch(1);
-        connector.init();
+        queueSource.init();
         latch.await();
 
         // fire onCompleted event (graceful stream end) and reset latch
         latch = new CountDownLatch(1);
         observer.onCompleted();
 
         // should enqueue error payload
-        BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
+        BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
         assertTrue(streamQueue.isEmpty());
         // should have restarted the stream (2 calls)
         latch.await();
