fix: chunk down flag assigned to not hit the max grpc limit

openfeature-provider-local/src/main/java/com/spotify/confidence/GrpcWasmFlagLogger.java

@@ -1,5 +1,6 @@
 package com.spotify.confidence;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.spotify.confidence.shaded.flags.resolver.v1.InternalFlagLoggerServiceGrpc;
 import com.spotify.confidence.shaded.flags.resolver.v1.WriteFlagLogsRequest;
 import com.spotify.confidence.shaded.iam.v1.AuthServiceGrpc;
@@ -8,14 +9,41 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@FunctionalInterface
+interface FlagLogWriter {
+  void write(WriteFlagLogsRequest request);
+}
+
 public class GrpcWasmFlagLogger implements WasmFlagLogger {
   private static final String CONFIDENCE_DOMAIN = "edge-grpc.spotify.com";
   private static final Logger logger = LoggerFactory.getLogger(GrpcWasmFlagLogger.class);
+  // Max number of flag_assigned entries per chunk to avoid exceeding gRPC max message size
+  private static final int MAX_FLAG_ASSIGNED_PER_CHUNK = 1000;
   private final InternalFlagLoggerServiceGrpc.InternalFlagLoggerServiceBlockingStub stub;
+  private final ExecutorService executorService;
+  private final FlagLogWriter writer;
+
+  @VisibleForTesting
+  public GrpcWasmFlagLogger(ApiSecret apiSecret, FlagLogWriter writer) {
+    final var channel = createConfidenceChannel();
+    final AuthServiceGrpc.AuthServiceBlockingStub authService =
+        AuthServiceGrpc.newBlockingStub(channel);
+    final TokenHolder tokenHolder =
+        new TokenHolder(apiSecret.clientId(), apiSecret.clientSecret(), authService);
+    final Channel authenticatedChannel =
+        ClientInterceptors.intercept(channel, new JwtAuthClientInterceptor(tokenHolder));
+    this.stub = InternalFlagLoggerServiceGrpc.newBlockingStub(authenticatedChannel);
+    this.executorService = Executors.newCachedThreadPool();
+    this.writer = writer;
+  }
 
   public GrpcWasmFlagLogger(ApiSecret apiSecret) {
     final var channel = createConfidenceChannel();
@@ -26,15 +54,88 @@ public GrpcWasmFlagLogger(ApiSecret apiSecret) {
     final Channel authenticatedChannel =
         ClientInterceptors.intercept(channel, new JwtAuthClientInterceptor(tokenHolder));
     this.stub = InternalFlagLoggerServiceGrpc.newBlockingStub(authenticatedChannel);
+    this.executorService = Executors.newCachedThreadPool();
+    this.writer =
+        request ->
+            executorService.submit(
+                () -> {
+                  try {
+                    final var ignore = stub.writeFlagLogs(request);
+                    logger.debug(
+                        "Successfully sent flag log with {} entries",
+                        request.getFlagAssignedCount());
+                  } catch (Exception e) {
+                    logger.error("Failed to write flag logs", e);
+                  }
+                });
   }
 
   @Override
   public void write(WriteFlagLogsRequest request) {
-    if (request.getClientResolveInfoList().isEmpty() && request.getFlagAssignedList().isEmpty()) {
+    if (request.getClientResolveInfoList().isEmpty()
+        && request.getFlagAssignedList().isEmpty()
+        && request.getFlagResolveInfoList().isEmpty()) {
       logger.debug("Skipping empty flag log request");
       return;
     }
-    final var ignore = stub.writeFlagLogs(request);
+
+    final int flagAssignedCount = request.getFlagAssignedCount();
+
+    // If flag_assigned list is small enough, send everything as-is
+    if (flagAssignedCount <= MAX_FLAG_ASSIGNED_PER_CHUNK) {
+      sendAsync(request);
+      return;
+    }
+
+    // Split flag_assigned into chunks and send each chunk asynchronously
+    logger.debug(
+        "Splitting {} flag_assigned entries into chunks of {}",
+        flagAssignedCount,
+        MAX_FLAG_ASSIGNED_PER_CHUNK);
+
+    final List<WriteFlagLogsRequest> chunks = createFlagAssignedChunks(request);
+    for (WriteFlagLogsRequest chunk : chunks) {
+      sendAsync(chunk);
+    }
+  }
+
+  private List<WriteFlagLogsRequest> createFlagAssignedChunks(WriteFlagLogsRequest request) {
+    final List<WriteFlagLogsRequest> chunks = new ArrayList<>();
+    final int totalFlags = request.getFlagAssignedCount();
+
+    for (int i = 0; i < totalFlags; i += MAX_FLAG_ASSIGNED_PER_CHUNK) {
+      final int end = Math.min(i + MAX_FLAG_ASSIGNED_PER_CHUNK, totalFlags);
+      final WriteFlagLogsRequest.Builder chunkBuilder =
+          WriteFlagLogsRequest.newBuilder()
+              .addAllFlagAssigned(request.getFlagAssignedList().subList(i, end));
+
+      // Include telemetry and resolve info only in the first chunk
+      if (i == 0) {
+        if (request.hasTelemetryData()) {
+          chunkBuilder.setTelemetryData(request.getTelemetryData());
+        }
+        chunkBuilder
+            .addAllClientResolveInfo(request.getClientResolveInfoList())
+            .addAllFlagResolveInfo(request.getFlagResolveInfoList());
+      }
+
+      chunks.add(chunkBuilder.build());
+    }
+
+    return chunks;
+  }
+
+  private void sendAsync(WriteFlagLogsRequest request) {
+    writer.write(request);
+  }
+
+  /**
+   * Shutdown the executor service. This will allow any pending async writes to complete. Call this
+   * when the application is shutting down.
+   */
+  @Override
+  public void shutdown() {
+    executorService.shutdown();
   }
 
   private static ManagedChannel createConfidenceChannel() {

openfeature-provider-local/src/main/java/com/spotify/confidence/LocalResolverServiceFactory.java

@@ -10,7 +10,7 @@
 import com.spotify.confidence.shaded.flags.admin.v1.ResolverStateServiceGrpc.ResolverStateServiceBlockingStub;
 import com.spotify.confidence.shaded.flags.resolver.v1.InternalFlagLoggerServiceGrpc;
 import com.spotify.confidence.shaded.flags.resolver.v1.Sdk;
-import com.spotify.confidence.shaded.flags.resolver.v1.WriteFlagLogsResponse;
+import com.spotify.confidence.shaded.flags.resolver.v1.WriteFlagLogsRequest;
 import com.spotify.confidence.shaded.iam.v1.AuthServiceGrpc;
 import com.spotify.confidence.shaded.iam.v1.AuthServiceGrpc.AuthServiceBlockingStub;
 import com.spotify.confidence.shaded.iam.v1.ClientCredential.ClientSecret;
@@ -179,7 +179,15 @@ private static FlagResolverService createFlagResolverService(
             .orElse(Duration.ofMinutes(5).toSeconds());
     final AtomicReference<byte[]> resolverStateProtobuf =
         new AtomicReference<>(accountStateProvider.provide());
-    final WasmFlagLogger flagLogger = request -> WriteFlagLogsResponse.getDefaultInstance();
+    // No-op logger for wasm mode with AccountStateProvider
+    final WasmFlagLogger flagLogger =
+        new WasmFlagLogger() {
+          @Override
+          public void write(WriteFlagLogsRequest request) {}
+
+          @Override
+          public void shutdown() {}
+        };
     final ResolverApi wasmResolverApi =
         new ThreadLocalSwapWasmResolverApi(
             flagLogger, resolverStateProtobuf.get(), accountId, stickyResolveStrategy);

openfeature-provider-local/src/main/java/com/spotify/confidence/ThreadLocalSwapWasmResolverApi.java

@@ -121,6 +121,7 @@ public int getInstanceCount() {
   @Override
   public void close() {
     resolverInstances.values().forEach(SwapWasmResolverApi::close);
+    flagLogger.shutdown();
     resolverInstances.clear();
   }
 }

openfeature-provider-local/src/main/java/com/spotify/confidence/WasmResolveApi.java

@@ -31,9 +31,10 @@
 
 class IsClosedException extends Exception {}
 
-@FunctionalInterface
 interface WasmFlagLogger {
   void write(WriteFlagLogsRequest request);
+
+  void shutdown();
 }
 
 class WasmResolveApi {