chore: error log on stream fail, retry cancels, 5000ms default deadline

Signed-off-by: Todd Baert <[email protected]>

Author: toddbaert

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

@@ -61,7 +61,8 @@ public void waitForInitialization(long deadline) {
             // if wait(0) is called, the thread would wait forever, so we abort when this would happen
             if (now >= end) {
                 throw new GeneralError(String.format(
-                        "Deadline exceeded. Condition did not complete within the %d ms deadline", deadline));
+                        "Startup timeout exceeded. Initialization did not complete within the %d ms deadline",
+                        deadline));
             }
             long remaining = end - now;
             synchronized (this) {

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/common/ChannelBuilder.java

@@ -58,12 +58,12 @@ public class ChannelBuilder {
                                     "retryableStatusCodes",
                                     Arrays.asList(
                                             /*
-                                             * All codes are retryable except OK, CANCELLED and DEADLINE_EXCEEDED since
+                                             * All codes are retryable except OK and DEADLINE_EXCEEDED since
                                              * any others not listed here cause a very tight loop of retries.
-                                             * CANCELLED is not retryable because it is a client-side termination.
                                              * DEADLINE_EXCEEDED is typically a result of a client specified deadline,
                                              * and definitionally should not result in a tight loop (it's a timeout).
                                              */
+                                            Code.CANCELLED.toString(),
                                             Code.UNKNOWN.toString(),
                                             Code.INVALID_ARGUMENT.toString(),
                                             Code.NOT_FOUND.toString(),

providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/common/QueueingStreamObserver.java

@@ -17,6 +17,7 @@ public class QueueingStreamObserver<T> implements StreamObserver<T> {
 
     public QueueingStreamObserver(final BlockingQueue<StreamResponseModel<T>> queue) {
         blockingQueue = queue;
+        queue.clear();
     }
 
     @Override

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

@@ -128,7 +128,7 @@ private void observeSyncStream() throws InterruptedException {
             // create a context which exists to track and cancel the stream
             try (CancellableContext context = Context.current().withCancellation()) {
 
-                restart(); // start the stream within the context
+                restart(); // start the stream with the context
 
                 // TODO: remove the metadata call entirely after https://github.com/open-feature/flagd/issues/1584
                 if (!syncMetadataDisabled) {
@@ -150,16 +150,16 @@ private void observeSyncStream() throws InterruptedException {
                 while (!shutdown.get() && !Context.current().isCancelled()) {
                     final StreamResponseModel<SyncFlagsResponse> taken = incomingQueue.take();
                     if (taken.isComplete()) {
-                        log.debug("Sync stream completed, will reconnect");
+                        log.debug("Sync stream completed, will restart");
                         // The stream is complete, we still try to reconnect
                         break;
                     }
 
                     Throwable streamException = taken.getError();
                     if (streamException != null) {
-                        log.debug("Exception in GRPC connection, streamException {}, will reconnect", streamException);
-                        if (!outgoingQueue.offer(new QueuePayload(
-                                QueuePayloadType.ERROR, "Error from stream or metadata", metadataResponse))) {
+                        log.error("Exception in stream RPC, streamException {}, will restart", streamException);
+                        if (!outgoingQueue.offer(
+                                new QueuePayload(QueuePayloadType.ERROR, "Error from stream: ", metadataResponse))) {
                             log.error("Failed to convey ERROR status, queue is full");
                         }
                         break;

providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/sync/SyncStreamQueueSourceTest.java

@@ -40,6 +40,7 @@ class SyncStreamQueueSourceTest {
     @BeforeEach
     public void init() throws Exception {
         blockingStub = mock(FlagSyncServiceBlockingStub.class);
+        when(blockingStub.withDeadlineAfter(anyLong(), any())).thenReturn(blockingStub);
         when(blockingStub.getMetadata(any())).thenReturn(GetMetadataResponse.getDefaultInstance());
 
         mockConnector = mock(ChannelConnector.class);