chore: add error log on stream fail, retry cancels

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/process/storage/connector/sync/SyncStreamQueueSource.java

@@ -147,7 +147,7 @@ private void observeSyncStream() throws InterruptedException {
                 }
 
                 // inner loop for handling messages
-                while (!shutdown.get() && !Context.current().isCancelled()) {
+                while (!shutdown.get() && !context.isCancelled()) {
                     final StreamResponseModel<SyncFlagsResponse> taken = incomingQueue.take();
                     if (taken.isComplete()) {
                         log.debug("Sync stream completed, will reconnect");
@@ -157,9 +157,9 @@ private void observeSyncStream() throws InterruptedException {
 
                     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 GRPC connection, streamException {}, will reconnect", 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);