diff --git a/openfeature-provider/go/confidence/internal/flag_logger/grpc_test.go b/openfeature-provider/go/confidence/internal/flag_logger/grpc_test.go index b91e7f0b..9e70a267 100644 --- a/openfeature-provider/go/confidence/internal/flag_logger/grpc_test.go +++ b/openfeature-provider/go/confidence/internal/flag_logger/grpc_test.go @@ -241,6 +241,34 @@ func TestGrpcWasmFlagLogger_FailureStats_NoLogBeforeWindow(t *testing.T) { } } +func TestGrpcWasmFlagLogger_FailureStats_AllFail(t *testing.T) { + var buf bytes.Buffer + testLogger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})) + + mockStub := &mockInternalFlagLoggerServiceClient{ + writeFlagLogsFunc: func(ctx context.Context, req *resolverv1.WriteFlagLogsRequest) (*resolverv1.WriteFlagLogsResponse, error) { + return nil, errors.New("unavailable") + }, + } + + logger := NewGrpcWasmFlagLogger(mockStub, "test-secret", testLogger) + + for i := 0; i < 10; i++ { + logger.Write(&resolverv1.WriteFlagLogsRequest{ + FlagAssigned: make([]*resolverv1.FlagAssigned, 1), + }) + } + logger.Shutdown() + + output := buf.String() + if !strings.Contains(output, "Flag log write failures") { + t.Error("Expected failure log after window with all failures") + } + if !strings.Contains(output, "failures=10") { + t.Errorf("Expected failures=10 in output, got: %s", output) + } +} + func TestNoOpWasmFlagLogger(t *testing.T) { logger := NewNoOpWasmFlagLogger() diff --git a/openfeature-provider/go/confidence/retry_test.go b/openfeature-provider/go/confidence/retry_test.go index 4d91e515..87e6a5a3 100644 --- a/openfeature-provider/go/confidence/retry_test.go +++ b/openfeature-provider/go/confidence/retry_test.go @@ -7,6 +7,7 @@ import ( "os" "sync/atomic" "testing" + "time" "github.com/spotify/confidence-resolver/openfeature-provider/go/confidence" fl "github.com/spotify/confidence-resolver/openfeature-provider/go/confidence/internal/flag_logger" @@ -91,6 +92,168 @@ func TestNoRetryOnPermissionDenied(t *testing.T) { } } +func TestAllRetriesExhausted(t *testing.T) { + var callCount atomic.Int32 + srv := grpc.NewServer() + resolverv1.RegisterInternalFlagLoggerServiceServer(srv, &retryTestService{ + callCount: &callCount, + failUntil: 100, + }) + + lis := bufconn.Listen(1024 * 1024) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + conn, err := grpc.NewClient("passthrough:///bufconn", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultServiceConfig(confidence.RetryServiceConfig), + ) + if err != nil { + t.Fatalf("Failed to dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + + stub := resolverv1.NewInternalFlagLoggerServiceClient(conn) + logger := fl.NewGrpcWasmFlagLogger(stub, "test-secret", slog.New(slog.NewTextHandler(os.Stderr, nil))) + + logger.Write(&resolverv1.WriteFlagLogsRequest{ + FlagAssigned: []*resolverv1.FlagAssigned{{ResolveId: "r1"}}, + }) + logger.Shutdown() + + if got := callCount.Load(); got != 3 { + t.Errorf("Expected 3 calls (max retries exhausted), got %d", got) + } +} + +func TestShutdownDuringRetries(t *testing.T) { + var callCount atomic.Int32 + srv := grpc.NewServer() + resolverv1.RegisterInternalFlagLoggerServiceServer(srv, &retryTestService{ + callCount: &callCount, + failUntil: 100, + }) + + lis := bufconn.Listen(1024 * 1024) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + conn, err := grpc.NewClient("passthrough:///bufconn", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultServiceConfig(confidence.RetryServiceConfig), + ) + if err != nil { + t.Fatalf("Failed to dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + + stub := resolverv1.NewInternalFlagLoggerServiceClient(conn) + logger := fl.NewGrpcWasmFlagLogger(stub, "test-secret", slog.New(slog.NewTextHandler(os.Stderr, nil))) + + logger.Write(&resolverv1.WriteFlagLogsRequest{ + FlagAssigned: []*resolverv1.FlagAssigned{{ResolveId: "r1"}}, + }) + + done := make(chan struct{}) + go func() { + logger.Shutdown() + close(done) + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + t.Fatal("Shutdown did not complete within 30 seconds") + } +} + +func TestTransportHooksPreserveRetry(t *testing.T) { + var callCount atomic.Int32 + srv := grpc.NewServer() + resolverv1.RegisterInternalFlagLoggerServiceServer(srv, &retryTestService{ + callCount: &callCount, + failUntil: 1, + }) + + lis := bufconn.Listen(1024 * 1024) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + baseOpts := []grpc.DialOption{ + grpc.WithDefaultServiceConfig(confidence.RetryServiceConfig), + } + opts := append([]grpc.DialOption{}, baseOpts...) + opts = append(opts, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + }), + ) + + conn, err := grpc.NewClient("passthrough:///bufconn", opts...) + if err != nil { + t.Fatalf("Failed to dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + + stub := resolverv1.NewInternalFlagLoggerServiceClient(conn) + logger := fl.NewGrpcWasmFlagLogger(stub, "test-secret", slog.New(slog.NewTextHandler(os.Stderr, nil))) + + logger.Write(&resolverv1.WriteFlagLogsRequest{ + FlagAssigned: []*resolverv1.FlagAssigned{{ResolveId: "r1"}}, + }) + logger.Shutdown() + + if got := callCount.Load(); got != 2 { + t.Errorf("Expected 2 calls (retry should work with custom hooks), got %d", got) + } +} + +func TestMultipleWritesWithRetryConfig(t *testing.T) { + var callCount atomic.Int32 + srv := grpc.NewServer() + resolverv1.RegisterInternalFlagLoggerServiceServer(srv, &retryTestService{ + callCount: &callCount, + failUntil: 0, + }) + + lis := bufconn.Listen(1024 * 1024) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + conn, err := grpc.NewClient("passthrough:///bufconn", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultServiceConfig(confidence.RetryServiceConfig), + ) + if err != nil { + t.Fatalf("Failed to dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + + stub := resolverv1.NewInternalFlagLoggerServiceClient(conn) + logger := fl.NewGrpcWasmFlagLogger(stub, "test-secret", slog.New(slog.NewTextHandler(os.Stderr, nil))) + + for i := 0; i < 5; i++ { + logger.Write(&resolverv1.WriteFlagLogsRequest{ + FlagAssigned: []*resolverv1.FlagAssigned{{ResolveId: "r1"}}, + }) + } + logger.Shutdown() + + if got := callCount.Load(); got != 5 { + t.Errorf("Expected 5 calls (all successful, no retries), got %d", got) + } +} + type retryTestService struct { resolverv1.UnimplementedInternalFlagLoggerServiceServer callCount *atomic.Int32 diff --git a/openfeature-provider/java/src/test/java/com/spotify/confidence/sdk/GrpcRetryTest.java b/openfeature-provider/java/src/test/java/com/spotify/confidence/sdk/GrpcRetryTest.java index 5ce3b589..798aef4b 100644 --- a/openfeature-provider/java/src/test/java/com/spotify/confidence/sdk/GrpcRetryTest.java +++ b/openfeature-provider/java/src/test/java/com/spotify/confidence/sdk/GrpcRetryTest.java @@ -133,4 +133,236 @@ public void clientWriteFlagLogs( server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); } } + + @Test + void allRetriesExhausted() throws Exception { + final var serverName = InProcessServerBuilder.generateName(); + final var callCount = new AtomicInteger(0); + + final var service = + new InternalFlagLoggerServiceGrpc.InternalFlagLoggerServiceImplBase() { + @Override + public void clientWriteFlagLogs( + WriteFlagLogsRequest request, + StreamObserver responseObserver) { + callCount.incrementAndGet(); + responseObserver.onError( + Status.UNAVAILABLE.withDescription("always fails").asRuntimeException()); + } + }; + + Server server = + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(service) + .build() + .start(); + + try { + final ChannelFactory retryChannelFactory = + (target, interceptors) -> { + InProcessChannelBuilder builder = + InProcessChannelBuilder.forName(serverName) + .directExecutor() + .defaultServiceConfig(DefaultChannelFactory.RETRY_SERVICE_CONFIG) + .enableRetry(); + if (!interceptors.isEmpty()) { + builder.intercept(interceptors.toArray(new ClientInterceptor[0])); + } + return builder.build(); + }; + + final var logger = new GrpcWasmFlagLogger("test-secret", retryChannelFactory); + + final var request = + WriteFlagLogsRequest.newBuilder() + .addFlagAssigned(FlagAssigned.newBuilder().setResolveId("r1").build()) + .build(); + + logger.write(request); + logger.shutdown(); + + assertEquals(3, callCount.get(), "Should have retried exactly 3 times (maxAttempts)"); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void shutdownCompletesWhenRetriesExhausted() throws Exception { + final var serverName = InProcessServerBuilder.generateName(); + final var callCount = new AtomicInteger(0); + + final var service = + new InternalFlagLoggerServiceGrpc.InternalFlagLoggerServiceImplBase() { + @Override + public void clientWriteFlagLogs( + WriteFlagLogsRequest request, + StreamObserver responseObserver) { + callCount.incrementAndGet(); + responseObserver.onError( + Status.UNAVAILABLE.withDescription("always fails").asRuntimeException()); + } + }; + + Server server = + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(service) + .build() + .start(); + + try { + final ChannelFactory retryChannelFactory = + (target, interceptors) -> { + InProcessChannelBuilder builder = + InProcessChannelBuilder.forName(serverName) + .directExecutor() + .defaultServiceConfig(DefaultChannelFactory.RETRY_SERVICE_CONFIG) + .enableRetry(); + if (!interceptors.isEmpty()) { + builder.intercept(interceptors.toArray(new ClientInterceptor[0])); + } + return builder.build(); + }; + + final var logger = new GrpcWasmFlagLogger("test-secret", retryChannelFactory); + + final var request = + WriteFlagLogsRequest.newBuilder() + .addFlagAssigned(FlagAssigned.newBuilder().setResolveId("r1").build()) + .build(); + + logger.write(request); + + long start = System.currentTimeMillis(); + logger.shutdown(); + long duration = System.currentTimeMillis() - start; + + assertTrue(duration < 30_000, "Shutdown should complete within 30s, took " + duration + "ms"); + assertTrue(callCount.get() > 0, "Server should have received at least one call"); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void customChannelFactoryWithRetryConfig() throws Exception { + final var serverName = InProcessServerBuilder.generateName(); + final var callCount = new AtomicInteger(0); + final var successLatch = new CountDownLatch(1); + + final var service = + new InternalFlagLoggerServiceGrpc.InternalFlagLoggerServiceImplBase() { + @Override + public void clientWriteFlagLogs( + WriteFlagLogsRequest request, + StreamObserver responseObserver) { + int attempt = callCount.incrementAndGet(); + if (attempt == 1) { + responseObserver.onError( + Status.UNAVAILABLE.withDescription("simulated").asRuntimeException()); + } else { + successLatch.countDown(); + responseObserver.onNext(WriteFlagLogsResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } + } + }; + + Server server = + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(service) + .build() + .start(); + + try { + final ChannelFactory customFactory = + (target, interceptors) -> { + InProcessChannelBuilder builder = + InProcessChannelBuilder.forName(serverName) + .directExecutor() + .defaultServiceConfig(DefaultChannelFactory.RETRY_SERVICE_CONFIG) + .enableRetry(); + if (!interceptors.isEmpty()) { + builder.intercept(interceptors.toArray(new ClientInterceptor[0])); + } + return builder.build(); + }; + + final var logger = new GrpcWasmFlagLogger("test-secret", customFactory); + + final var request = + WriteFlagLogsRequest.newBuilder() + .addFlagAssigned(FlagAssigned.newBuilder().setResolveId("r1").build()) + .build(); + + logger.write(request); + logger.shutdown(); + + assertTrue( + successLatch.await(5, TimeUnit.SECONDS), + "Custom factory with retry config should enable retries"); + assertEquals(2, callCount.get(), "Should have retried once (2 total calls)"); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void multipleWritesThenShutdown() throws Exception { + final var serverName = InProcessServerBuilder.generateName(); + final var callCount = new AtomicInteger(0); + + final var service = + new InternalFlagLoggerServiceGrpc.InternalFlagLoggerServiceImplBase() { + @Override + public void clientWriteFlagLogs( + WriteFlagLogsRequest request, + StreamObserver responseObserver) { + callCount.incrementAndGet(); + responseObserver.onNext(WriteFlagLogsResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } + }; + + Server server = + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(service) + .build() + .start(); + + try { + final ChannelFactory retryChannelFactory = + (target, interceptors) -> { + InProcessChannelBuilder builder = + InProcessChannelBuilder.forName(serverName) + .directExecutor() + .defaultServiceConfig(DefaultChannelFactory.RETRY_SERVICE_CONFIG) + .enableRetry(); + if (!interceptors.isEmpty()) { + builder.intercept(interceptors.toArray(new ClientInterceptor[0])); + } + return builder.build(); + }; + + final var logger = new GrpcWasmFlagLogger("test-secret", retryChannelFactory); + + for (int i = 0; i < 5; i++) { + final var request = + WriteFlagLogsRequest.newBuilder() + .addFlagAssigned(FlagAssigned.newBuilder().setResolveId("r" + i).build()) + .build(); + logger.write(request); + } + + logger.shutdown(); + + assertEquals(5, callCount.get(), "All 5 writes should complete with retry config active"); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } } diff --git a/openfeature-provider/python/tests/test_flag_logger.py b/openfeature-provider/python/tests/test_flag_logger.py index e8c25dfe..e8e651a6 100644 --- a/openfeature-provider/python/tests/test_flag_logger.py +++ b/openfeature-provider/python/tests/test_flag_logger.py @@ -310,6 +310,197 @@ def test_no_log_before_window_boundary(self) -> None: mock_logger.warning.assert_not_called() +class TestGrpcFlagLoggerRetryExhausted: + """Test behavior when all writes fail (simulating exhausted retries).""" + + def test_all_writes_fail_stats_correct(self) -> None: + """Failure stats should report 10/10 failures when all writes fail.""" + mock_channel = MagicMock() + mock_stub = MagicMock() + mock_stub.ClientWriteFlagLogs = MagicMock(side_effect=Exception("unavailable")) + + stub_path = ( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ) + with ( + patch(stub_path, return_value=mock_stub), + patch("confidence.flag_logger.logger") as mock_logger, + ): + flag_logger = GrpcFlagLogger( + client_secret="test-secret", channel=mock_channel + ) + + request = internal_api_pb2.WriteFlagLogsRequest() + request.flag_assigned.add() + request_bytes = request.SerializeToString() + + for _ in range(10): + flag_logger.write(request_bytes) + + flag_logger.shutdown() + + mock_logger.warning.assert_called_once() + call_args = mock_logger.warning.call_args + assert "Flag log write failures" in call_args[0][0] + assert call_args[0][1] == 10 + + +class TestGrpcFlagLoggerShutdownDuringFailures: + """Test shutdown behavior when writes are failing.""" + + def test_shutdown_completes_when_writes_fail(self) -> None: + """shutdown() should complete promptly even when all writes fail.""" + mock_channel = MagicMock() + mock_stub = MagicMock() + mock_stub.ClientWriteFlagLogs = MagicMock(side_effect=Exception("unavailable")) + + stub_path = ( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ) + with patch(stub_path, return_value=mock_stub): + flag_logger = GrpcFlagLogger( + client_secret="test-secret", channel=mock_channel + ) + + request = internal_api_pb2.WriteFlagLogsRequest() + request.flag_assigned.add() + request_bytes = request.SerializeToString() + + for _ in range(5): + flag_logger.write(request_bytes) + + start = time.monotonic() + flag_logger.shutdown() + elapsed = time.monotonic() - start + + assert elapsed < 5.0, f"shutdown() took {elapsed:.2f}s, expected < 5s" + + def test_shutdown_drains_slow_failing_writes(self) -> None: + """shutdown() should wait for slow failing writes to complete.""" + mock_channel = MagicMock() + mock_stub = MagicMock() + + call_count = 0 + + def slow_fail(*args, **kwargs): + nonlocal call_count + call_count += 1 + time.sleep(0.1) + raise Exception("unavailable") + + mock_stub.ClientWriteFlagLogs = slow_fail + + stub_path = ( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ) + with patch(stub_path, return_value=mock_stub): + flag_logger = GrpcFlagLogger( + client_secret="test-secret", channel=mock_channel + ) + + request = internal_api_pb2.WriteFlagLogsRequest() + request.flag_assigned.add() + request_bytes = request.SerializeToString() + + for _ in range(3): + flag_logger.write(request_bytes) + + flag_logger.shutdown() + + assert call_count == 3, f"Expected 3 calls, got {call_count}" + + +class TestGrpcFlagLoggerCustomChannel: + """Test behavior with custom vs default gRPC channels.""" + + def test_custom_channel_used_directly(self) -> None: + """Custom channel should be used as-is, not owned by the logger.""" + mock_channel = MagicMock() + mock_stub = MagicMock() + mock_stub.ClientWriteFlagLogs = MagicMock( + return_value=internal_api_pb2.WriteFlagLogsResponse() + ) + + stub_path = ( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ) + with patch(stub_path, return_value=mock_stub) as mock_stub_class: + flag_logger = GrpcFlagLogger( + client_secret="test-secret", channel=mock_channel + ) + + mock_stub_class.assert_called_once_with(mock_channel) + assert not flag_logger._owns_channel + + flag_logger.shutdown() + + mock_channel.close.assert_not_called() + + def test_default_channel_has_retry_config(self) -> None: + """Default channel should be created with retry service config.""" + with ( + patch("confidence.flag_logger.grpc.secure_channel") as mock_secure_channel, + patch("confidence.flag_logger.grpc.ssl_channel_credentials") as mock_creds, + patch( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ), + ): + mock_secure_channel.return_value = MagicMock() + mock_creds.return_value = MagicMock() + + flag_logger = GrpcFlagLogger(client_secret="test-secret") + + mock_secure_channel.assert_called_once() + call_kwargs = mock_secure_channel.call_args + options = call_kwargs.kwargs.get("options") + assert options is not None, "Options should be passed to secure_channel" + + config_found = False + for key, value in options: + if key == "grpc.service_config": + import json + + config = json.loads(value) + assert "methodConfig" in config + retry_policy = config["methodConfig"][0]["retryPolicy"] + assert retry_policy["maxAttempts"] == 3 + assert "UNAVAILABLE" in retry_policy["retryableStatusCodes"] + config_found = True + break + + assert config_found, "Retry service config not found in channel options" + + assert flag_logger._owns_channel + flag_logger.shutdown() + + def test_multiple_writes_with_custom_channel(self) -> None: + """Multiple writes through custom channel should all complete.""" + mock_channel = MagicMock() + mock_stub = MagicMock() + mock_stub.ClientWriteFlagLogs = MagicMock( + return_value=internal_api_pb2.WriteFlagLogsResponse() + ) + + stub_path = ( + "confidence.flag_logger.internal_api_pb2_grpc.InternalFlagLoggerServiceStub" + ) + with patch(stub_path, return_value=mock_stub): + flag_logger = GrpcFlagLogger( + client_secret="test-secret", channel=mock_channel + ) + + request = internal_api_pb2.WriteFlagLogsRequest() + request.flag_assigned.add() + request_bytes = request.SerializeToString() + + for _ in range(5): + flag_logger.write(request_bytes) + + flag_logger.shutdown() + + assert mock_stub.ClientWriteFlagLogs.call_count == 5 + + class TestNoOpFlagLogger: """Test NoOpFlagLogger behavior."""