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

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

@@ -8,14 +8,21 @@
 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;
 
 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;
 
   public GrpcWasmFlagLogger(ApiSecret apiSecret) {
     final var channel = createConfidenceChannel();
@@ -26,15 +33,84 @@ public GrpcWasmFlagLogger(ApiSecret apiSecret) {
     final Channel authenticatedChannel =
         ClientInterceptors.intercept(channel, new JwtAuthClientInterceptor(tokenHolder));
     this.stub = InternalFlagLoggerServiceGrpc.newBlockingStub(authenticatedChannel);
+    this.executorService = Executors.newCachedThreadPool();
   }
 
   @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 metadata 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) {
+    executorService.submit(
+        () -> {
+          try {
+            stub.writeFlagLogs(request);
+            logger.debug(
+                "Successfully sent flag log with {} entries", request.getFlagAssignedCount());
+          } catch (Exception e) {
+            logger.error("Failed to write flag logs", e);
+          }
+        });
+  }
+
+  /**
+   * 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 {

openfeature-provider-local/src/test/java/com/spotify/confidence/TestBase.java

@@ -7,6 +7,7 @@
 import com.google.protobuf.util.Values;
 import com.spotify.confidence.shaded.flags.resolver.v1.ResolveFlagsRequest;
 import com.spotify.confidence.shaded.flags.resolver.v1.ResolveFlagsResponse;
+import com.spotify.confidence.shaded.flags.resolver.v1.WriteFlagLogsRequest;
 import com.spotify.confidence.shaded.iam.v1.Client;
 import com.spotify.confidence.shaded.iam.v1.ClientCredential;
 import java.util.BitSet;
@@ -45,7 +46,16 @@ protected TestBase(ResolverState state, boolean isWasm) {
     if (isWasm) {
       final var wasmResolverApi =
           new SwapWasmResolverApi(
-              request -> {}, desiredState.toProto().toByteArray(), "", mockFallback);
+              new WasmFlagLogger() {
+                @Override
+                public void write(WriteFlagLogsRequest request) {}
+
+                @Override
+                public void shutdown() {}
+              },
+              desiredState.toProto().toByteArray(),
+              "",
+              mockFallback);
       resolverServiceFactory =
           new LocalResolverServiceFactory(
               wasmResolverApi, resolverState, resolveTokenConverter, mock(), mockFallback);