fix: move flush out of wasm resolver api (#288)

move flush out

Author: vahidlazio

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

@@ -46,6 +46,8 @@ class LocalResolverServiceFactory implements ResolverServiceFactory {
       Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder().setDaemon(true).build());
   private static final Duration RESOLVE_INFO_LOG_INTERVAL = Duration.ofMinutes(1);
   private final StickyResolveStrategy stickyResolveStrategy;
+  private static final ScheduledExecutorService logPollExecutor =
+      Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder().setDaemon(true).build());
 
   private static ManagedChannel createConfidenceChannel() {
     final String confidenceDomain =
@@ -111,14 +113,19 @@ private static FlagResolverService createFlagResolverService(
               sidecarFlagsAdminFetcher.accountId,
               stickyResolveStrategy);
       flagsFetcherExecutor.scheduleAtFixedRate(
+          sidecarFlagsAdminFetcher::reload,
+          pollIntervalSeconds,
+          pollIntervalSeconds,
+          TimeUnit.SECONDS);
+
+      logPollExecutor.scheduleAtFixedRate(
           () -> {
-            sidecarFlagsAdminFetcher.reload();
-            wasmResolverApi.updateState(
+            wasmResolverApi.updateStateAndFlushLogs(
                 sidecarFlagsAdminFetcher.rawStateHolder().get().toByteArray(),
                 sidecarFlagsAdminFetcher.accountId);
           },
-          pollIntervalSeconds,
-          pollIntervalSeconds,
+          10,
+          10,
           TimeUnit.SECONDS);
 
       return new WasmFlagResolverService(wasmResolverApi, stickyResolveStrategy);
@@ -176,7 +183,7 @@ private static FlagResolverService createFlagResolverService(
             flagLogger, resolverStateProtobuf, accountId, stickyResolveStrategy);
     flagsFetcherExecutor.scheduleAtFixedRate(
         () -> {
-          wasmResolverApi.updateState(accountStateProvider.provide(), accountId);
+          wasmResolverApi.updateStateAndFlushLogs(accountStateProvider.provide(), accountId);
         },
         pollIntervalSeconds,
         pollIntervalSeconds,
@@ -236,7 +243,7 @@ private static FlagResolverService createFlagResolverService(
   @VisibleForTesting
   public void setState(byte[] state, String accountId) {
     if (this.wasmResolveApi != null) {
-      wasmResolveApi.updateState(state, accountId);
+      wasmResolveApi.updateStateAndFlushLogs(state, accountId);
     }
   }
 

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

@@ -34,21 +34,20 @@ public SwapWasmResolverApi(
     this.wasmResolverApiRef.set(primaryWasmResolverApi);
   }
 
-  public void updateState(byte[] state, String accountId) {
+  public void updateStateAndFlushLogs(byte[] state, String accountId) {
     if (isPrimary) {
       this.secondaryWasmResolverApi.setResolverState(state, accountId);
       this.wasmResolverApiRef.set(secondaryWasmResolverApi);
+      this.primaryWasmResolverApi.flushLogs();
     } else {
       this.primaryWasmResolverApi.setResolverState(state, accountId);
       this.wasmResolverApiRef.set(primaryWasmResolverApi);
+      this.secondaryWasmResolverApi.flushLogs();
     }
     isPrimary = !isPrimary;
   }
 
-  public void close() {
-    primaryWasmResolverApi.close();
-    secondaryWasmResolverApi.close();
-  }
+  public void close() {}
 
   private final ReentrantLock logResolveLock = new ReentrantLock();
 

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

@@ -10,7 +10,6 @@
 import com.dylibso.chicory.wasm.WasmModule;
 import com.dylibso.chicory.wasm.types.FunctionType;
 import com.dylibso.chicory.wasm.types.ValType;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.BytesValue;
 import com.google.protobuf.GeneratedMessageV3;
@@ -29,8 +28,6 @@
 import java.security.SecureRandom;
 import java.time.Instant;
 import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
 import javax.crypto.Cipher;
 import javax.crypto.spec.IvParameterSpec;
@@ -57,14 +54,10 @@ class WasmResolveApi {
   private final ExportFunction wasmMsgGuestSetResolverState;
   private final ExportFunction wasmMsgFlushLogs;
   private final ExportFunction wasmMsgGuestResolve;
-  private static final ScheduledExecutorService logPollExecutor =
-      Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder().setDaemon(true).build());
   private final ExportFunction wasmMsgGuestResolveWithSticky;
 
   public WasmResolveApi(WasmFlagLogger flagLogger) {
     this.writeFlagLogs = flagLogger;
-    logPollExecutor.scheduleAtFixedRate(
-        this::flushLogs, 10, 10, java.util.concurrent.TimeUnit.SECONDS);
     try (InputStream wasmStream =
         getClass().getClassLoader().getResourceAsStream("wasm/confidence_resolver.wasm")) {
       if (wasmStream == null) {
@@ -145,10 +138,6 @@ private Timestamp currentTime(Messages.Void unused) {
     return Timestamp.newBuilder().setSeconds(Instant.now().getEpochSecond()).build();
   }
 
-  public void close() {
-    logPollExecutor.shutdownNow();
-  }
-
   public void setResolverState(byte[] state, String accountId) {
     final var resolverStateRequest =
         Messages.SetResolverStateRequest.newBuilder()
@@ -165,7 +154,7 @@ public void setResolverState(byte[] state, String accountId) {
     consumeResponse(respPtr, Messages.Void::parseFrom);
   }
 
-  private void flushLogs() {
+  public void flushLogs() {
     final var voidRequest = Messages.Void.getDefaultInstance();
     final var reqPtr = transferRequest(voidRequest);
     final var respPtr = (int) wasmMsgFlushLogs.apply(reqPtr)[0];