Skip to content

Commit bab7e4c

Browse files
toddbaerttoddbaert
committed
fixup: add sleep delay to prevent busy loop
Signed-off-by: Todd Baert <[email protected]>
1 parent 1eccc02 commit bab7e4c

File tree

3 files changed

+94
-87
lines changed

3 files changed

+94
-87
lines changed

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

Lines changed: 15 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -26,25 +26,11 @@
2626
/** gRPC channel builder helper. */
2727
public class ChannelBuilder {
2828

29-
@SuppressWarnings({"unchecked", "rawtypes"})
30-
private static final Map<String, ?> DEFAULT_RETRY_POLICY = new HashMap() {
31-
{
32-
// 1s + 2s + 4s
33-
put("maxAttempts", 3.0); // types used here are important, need to be doubles
34-
put("initialBackoff", "1s");
35-
put("maxBackoff", "5s");
36-
put("backoffMultiplier", 2.0);
37-
// status codes to retry on:
38-
put(
39-
"retryableStatusCodes",
40-
Arrays.asList(
41-
/*
42-
* Only states UNAVAILABLE and UNKNOWN should be retried. All
43-
* other failure states will probably not be resolved with a simple retry.
44-
*/
45-
Code.UNAVAILABLE.toString(), Code.UNKNOWN.toString()));
46-
}
47-
};
29+
/**
30+
* Maximum retry backoff in seconds.
31+
* Also used by SyncStreamQueueSource to control outer loop retry delay.
32+
*/
33+
public static final int MAX_RETRY_BACKOFF_SECONDS = 5;
4834

4935
/**
5036
* Controls retry (not-reconnection) policy for failed RPCs.
@@ -67,46 +53,23 @@ public class ChannelBuilder {
6753
put("service", "flagd.evaluation.v1.Service");
6854
}
6955
}));
70-
put("retryPolicy", DEFAULT_RETRY_POLICY);
71-
}
72-
73-
{
74-
put("name", Arrays.asList(new HashMap() {
75-
{
76-
put("service", "flagd.sync.v1.FlagSyncService");
77-
put("method", "SyncFlags");
78-
}
79-
}));
80-
put("retryPolicy", new HashMap(DEFAULT_RETRY_POLICY) {
56+
put("retryPolicy", new HashMap() {
8157
{
82-
// 1s + 2s + 4s + 5s + 5s + 5s + 5s + 5s + 5s + 5s
83-
put("maxAttempts", 12.0);
84-
// for streaming Retry on more status codes
58+
// 1 + 2 + 4
59+
put("maxAttempts", 3.0); // types used here are important, need to be doubles
60+
put("initialBackoff", "1s");
61+
put("maxBackoff", String.format("%ds", MAX_RETRY_BACKOFF_SECONDS));
62+
put("backoffMultiplier", 2.0);
63+
// status codes to retry on:
8564
put(
8665
"retryableStatusCodes",
8766
Arrays.asList(
8867
/*
89-
* All codes are retryable except OK and DEADLINE_EXCEEDED since
90-
* any others not listed here cause a very tight loop of retries.
91-
* DEADLINE_EXCEEDED is typically a result of a client specified
92-
* deadline,and definitionally should not result in a tight loop
93-
* (it's a timeout).
68+
* As per gRPC spec, the following status codes are safe to retry:
69+
* UNAVAILABLE, UNKNOWN,
9470
*/
95-
Code.CANCELLED.toString(),
9671
Code.UNKNOWN.toString(),
97-
Code.INVALID_ARGUMENT.toString(),
98-
Code.NOT_FOUND.toString(),
99-
Code.ALREADY_EXISTS.toString(),
100-
Code.PERMISSION_DENIED.toString(),
101-
Code.RESOURCE_EXHAUSTED.toString(),
102-
Code.FAILED_PRECONDITION.toString(),
103-
Code.ABORTED.toString(),
104-
Code.OUT_OF_RANGE.toString(),
105-
Code.UNIMPLEMENTED.toString(),
106-
Code.INTERNAL.toString(),
107-
Code.UNAVAILABLE.toString(),
108-
Code.DATA_LOSS.toString(),
109-
Code.UNAUTHENTICATED.toString()));
72+
Code.UNAVAILABLE.toString()));
11073
}
11174
});
11275
}

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

Lines changed: 28 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@
3535
value = {"PREDICTABLE_RANDOM", "EI_EXPOSE_REP"},
3636
justification = "Random is used to generate a variation & flag configurations require exposing")
3737
public class SyncStreamQueueSource implements QueueSource {
38+
/**
39+
* Delay between retries of the outer loop, in milliseconds. This is derived from the channel retry (5s)
40+
*/
41+
public static final int RETRY_LOOP_DELAY_MS = ChannelBuilder.MAX_RETRY_BACKOFF_SECONDS * 1000;
3842
private static final int QUEUE_SIZE = 5;
3943

4044
private final AtomicBoolean shutdown = new AtomicBoolean(false);
@@ -110,26 +114,32 @@ private void observeSyncStream() {
110114
log.info("Initializing sync stream observer");
111115

112116
// outer loop for re-issuing the stream request
113-
// "waitForReady" on the channel, plus our retry policy slow this loop down in error conditions
117+
// "waitForReady" on the channel, plus our retry policy slow this loop down in
118+
// error conditions
114119
while (!shutdown.get()) {
115-
log.debug("Initializing sync stream request");
116-
SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue);
117-
118-
try {
119-
observer.metadata = getMetadata();
120-
} catch (Exception metaEx) {
121-
// retry if getMetadata fails
122-
String message = metaEx.getMessage();
123-
log.debug("Metadata request error: {}, will restart", message, metaEx);
124-
enqueueError(String.format("Error in getMetadata request: %s", message));
125-
continue;
126-
}
127-
128120
try {
129-
syncFlags(observer);
130-
} catch (Exception ex) {
131-
log.error("Unexpected sync stream exception, will restart.", ex);
132-
enqueueError(String.format("Error in syncStream: %s", ex.getMessage()));
121+
log.debug("Initializing sync stream request");
122+
SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue);
123+
try {
124+
observer.metadata = getMetadata();
125+
} catch (Exception metaEx) {
126+
// retry if getMetadata fails
127+
String message = metaEx.getMessage();
128+
log.debug("Metadata request error: {}, will restart", message, metaEx);
129+
enqueueError(String.format("Error in getMetadata request: %s", message));
130+
Thread.sleep(1000);
131+
continue;
132+
}
133+
134+
try {
135+
syncFlags(observer);
136+
} catch (Exception ex) {
137+
log.error("Unexpected sync stream exception, will restart.", ex);
138+
enqueueError(String.format("Error in syncStream: %s", ex.getMessage()));
139+
Thread.sleep(1000);
140+
}
141+
} catch (InterruptedException ie) {
142+
log.debug("Stream loop interrupted, most likely shutdown was invoked", ie);
133143
}
134144
}
135145

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

Lines changed: 51 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import dev.openfeature.flagd.grpc.sync.Sync.GetMetadataResponse;
2424
import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsRequest;
2525
import dev.openfeature.flagd.grpc.sync.Sync.SyncFlagsResponse;
26+
import io.grpc.StatusRuntimeException;
2627
import io.grpc.stub.StreamObserver;
2728
import java.util.concurrent.BlockingQueue;
2829
import java.util.concurrent.CountDownLatch;
@@ -35,11 +36,12 @@ class SyncStreamQueueSourceTest {
3536
private ChannelConnector mockConnector;
3637
private FlagSyncServiceBlockingStub blockingStub;
3738
private FlagSyncServiceStub stub;
39+
private FlagSyncServiceStub errorStub;
3840
private StreamObserver<SyncFlagsResponse> observer;
3941
private CountDownLatch latch; // used to wait for observer to be initialized
4042

4143
@BeforeEach
42-
public void init() throws Exception {
44+
public void setup() throws Exception {
4345
blockingStub = mock(FlagSyncServiceBlockingStub.class);
4446
when(blockingStub.withDeadlineAfter(anyLong(), any())).thenReturn(blockingStub);
4547
when(blockingStub.getMetadata(any())).thenReturn(GetMetadataResponse.getDefaultInstance());
@@ -57,22 +59,53 @@ public void init() throws Exception {
5759
})
5860
.when(stub)
5961
.syncFlags(any(SyncFlagsRequest.class), any(StreamObserver.class)); // Mock the initialize
60-
// method
62+
63+
errorStub = mock(FlagSyncServiceStub.class);
64+
when(errorStub.withDeadlineAfter(anyLong(), any())).thenReturn(errorStub);
65+
doAnswer((Answer<Void>) invocation -> {
66+
Object[] args = invocation.getArguments();
67+
observer = (StreamObserver<SyncFlagsResponse>) args[1];
68+
latch.countDown();
69+
throw new StatusRuntimeException(io.grpc.Status.NOT_FOUND);
70+
})
71+
.when(errorStub)
72+
.syncFlags(any(SyncFlagsRequest.class), any(StreamObserver.class)); // Mock the initialize
6173
}
6274

75+
// @Test
76+
// void initError_DoesNotBusyWait() throws Exception {
77+
// // make sure we do not spin in a busy loop on errors
78+
79+
// SyncStreamQueueSource queueSource =
80+
// new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, errorStub, blockingStub);
81+
// latch = new CountDownLatch(1);
82+
// queueSource.init();
83+
// latch.await();
84+
85+
// BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
86+
// QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
87+
// assertNotNull(payload);
88+
// assertEquals(QueuePayloadType.ERROR, payload.getType());
89+
// Thread.sleep(SyncStreamQueueSource.RETRY_LOOP_DELAY_MS + 1000); // wait for retries
90+
91+
// // should have retried the stream (2 calls); initial + 1 retry
92+
// // it's very important that the retry count is low, to confirm no busy-loop
93+
// verify(errorStub, times(2)).syncFlags(any(), any());
94+
// }
95+
6396
@Test
6497
void onNextEnqueuesDataPayload() throws Exception {
65-
SyncStreamQueueSource connector =
98+
SyncStreamQueueSource queueSource =
6699
new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
67100
latch = new CountDownLatch(1);
68-
connector.init();
101+
queueSource.init();
69102
latch.await();
70103

71104
// fire onNext (data) event
72105
observer.onNext(SyncFlagsResponse.newBuilder().build());
73106

74107
// should enqueue data payload
75-
BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
108+
BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
76109
QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
77110
assertNotNull(payload);
78111
assertNotNull(payload.getSyncContext());
@@ -81,20 +114,21 @@ void onNextEnqueuesDataPayload() throws Exception {
81114
verify(stub, times(1)).syncFlags(any(), any());
82115
}
83116

117+
84118
@Test
85119
void onNextEnqueuesDataPayloadMetadataDisabled() throws Exception {
86120
// disable GetMetadata call
87-
SyncStreamQueueSource connector = new SyncStreamQueueSource(
121+
SyncStreamQueueSource queueSource = new SyncStreamQueueSource(
88122
FlagdOptions.builder().syncMetadataDisabled(true).build(), mockConnector, stub, blockingStub);
89123
latch = new CountDownLatch(1);
90-
connector.init();
124+
queueSource.init();
91125
latch.await();
92126

93127
// fire onNext (data) event
94128
observer.onNext(SyncFlagsResponse.newBuilder().build());
95129

96130
// should enqueue data payload
97-
BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
131+
BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
98132
QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
99133
assertNotNull(payload);
100134
assertNull(payload.getSyncContext());
@@ -108,10 +142,10 @@ void onNextEnqueuesDataPayloadMetadataDisabled() throws Exception {
108142
@Test
109143
void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
110144
// disable GetMetadata call
111-
SyncStreamQueueSource connector =
145+
SyncStreamQueueSource queueSource =
112146
new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
113147
latch = new CountDownLatch(1);
114-
connector.init();
148+
queueSource.init();
115149
latch.await();
116150

117151
// fire onNext (data) event
@@ -120,7 +154,7 @@ void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
120154
SyncFlagsResponse.newBuilder().setSyncContext(syncContext).build());
121155

122156
// should enqueue data payload
123-
BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
157+
BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
124158
QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
125159
assertNotNull(payload);
126160
assertEquals(syncContext, payload.getSyncContext());
@@ -131,18 +165,18 @@ void onNextEnqueuesDataPayloadWithSyncContext() throws Exception {
131165

132166
@Test
133167
void onErrorEnqueuesDataPayload() throws Exception {
134-
SyncStreamQueueSource connector =
168+
SyncStreamQueueSource queueSource =
135169
new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
136170
latch = new CountDownLatch(1);
137-
connector.init();
171+
queueSource.init();
138172
latch.await();
139173

140174
// fire onError event and reset latch
141175
latch = new CountDownLatch(1);
142176
observer.onError(new Exception("fake exception"));
143177

144178
// should enqueue error payload
145-
BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
179+
BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
146180
QueuePayload payload = streamQueue.poll(1000, TimeUnit.MILLISECONDS);
147181
assertNotNull(payload);
148182
assertEquals(QueuePayloadType.ERROR, payload.getType());
@@ -153,18 +187,18 @@ void onErrorEnqueuesDataPayload() throws Exception {
153187

154188
@Test
155189
void onCompletedEnqueuesDataPayload() throws Exception {
156-
SyncStreamQueueSource connector =
190+
SyncStreamQueueSource queueSource =
157191
new SyncStreamQueueSource(FlagdOptions.builder().build(), mockConnector, stub, blockingStub);
158192
latch = new CountDownLatch(1);
159-
connector.init();
193+
queueSource.init();
160194
latch.await();
161195

162196
// fire onCompleted event (graceful stream end) and reset latch
163197
latch = new CountDownLatch(1);
164198
observer.onCompleted();
165199

166200
// should enqueue error payload
167-
BlockingQueue<QueuePayload> streamQueue = connector.getStreamQueue();
201+
BlockingQueue<QueuePayload> streamQueue = queueSource.getStreamQueue();
168202
assertTrue(streamQueue.isEmpty());
169203
// should have restarted the stream (2 calls)
170204
latch.await();

0 commit comments

Comments
 (0)