[NA] [BE] Prevent nested cache deadlocks (#4404)

Author: andrescrz

apps/opik-backend/pom.xml

@@ -313,6 +313,11 @@
             <artifactId>tika-core</artifactId>
             <version>3.2.0</version>
         </dependency>
+        <dependency>
+            <groupId>io.vavr</groupId>
+            <artifactId>vavr</artifactId>
+            <version>0.10.7</version>
+        </dependency>
 
         <!-- Test -->
 

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/cache/CacheInterceptor.java

@@ -5,6 +5,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.type.TypeFactory;
+import io.vavr.CheckedFunction2;
 import jakarta.inject.Provider;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
@@ -25,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
@@ -49,113 +49,113 @@ public Object invoke(MethodInvocation invocation) throws Throwable {
 
         var cacheable = method.getAnnotation(Cacheable.class);
         if (cacheable != null) {
-            return runCacheAwareAction(invocation, isReactive, cacheable.name(), cacheable.key(),
+            return runCacheAwareAction(invocation, cacheable.name(), cacheable.key(),
                     (key, group) -> processCacheableMethod(invocation, isReactive, key, group, cacheable));
         }
 
         var cachePut = method.getAnnotation(CachePut.class);
         if (cachePut != null) {
-            return runCacheAwareAction(invocation, isReactive, cachePut.name(), cachePut.key(),
+            return runCacheAwareAction(invocation, cachePut.name(), cachePut.key(),
                     (key, group) -> processCachePutMethod(invocation, isReactive, key, group));
         }
 
         var cacheEvict = method.getAnnotation(CacheEvict.class);
         if (cacheEvict != null) {
-            return runCacheAwareAction(invocation, isReactive, cacheEvict.name(), cacheEvict.key(),
+            return runCacheAwareAction(invocation, cacheEvict.name(), cacheEvict.key(),
                     (key, group) -> processCacheEvictMethod(invocation, isReactive, key, cacheEvict));
         }
 
         return invocation.proceed();
     }
 
-    private Object runCacheAwareAction(MethodInvocation invocation, boolean isReactive, String group, String keyAgs,
-            BiFunction<String, String, Object> action) throws Throwable {
-
+    private Object runCacheAwareAction(
+            MethodInvocation invocation, String group, String keyArgs, CheckedFunction2<String, String, Object> action)
+            throws Throwable {
         String key;
-
         try {
-            key = getKeyName(group, keyAgs, invocation);
+            key = getKeyName(group, keyArgs, invocation);
         } catch (Exception e) {
             // If there is an error evaluating the key, proceed without caching
-            log.error("Error evaluating key expression: {}", keyAgs, e);
-            log.warn("Cache will be skipped due to error evaluating key expression");
+            log.warn("Cache will be skipped due to error evaluating key expression '{}'", keyArgs, e);
             return invocation.proceed();
         }
-
-        if (isReactive) {
-            return action.apply(key, group);
-        }
-
-        return ((Mono<?>) action.apply(key, group)).block();
+        return action.apply(key, group);
     }
 
-    private Mono<Object> processCacheEvictMethod(MethodInvocation invocation, boolean isReactive, String key,
-            CacheEvict cacheEvict) {
+    private Object processCacheEvictMethod(
+            MethodInvocation invocation, boolean isReactive, String key, CacheEvict cacheEvict) throws Throwable {
         if (isReactive) {
             try {
                 return ((Mono<?>) invocation.proceed())
                         .flatMap(value -> cacheManager.get().evict(key, cacheEvict.keyUsesPatternMatching())
-                                .thenReturn(value))
+                                .thenReturn(value)
+                                .onErrorResume(exception -> {
+                                    log.error("Error evicting cache", exception);
+                                    return Mono.just(value); // Return value even if evict fails
+                                }))
                         .switchIfEmpty(
-                                cacheManager.get().evict(key, cacheEvict.keyUsesPatternMatching()).then(Mono.empty()))
+                                cacheManager.get().evict(key, cacheEvict.keyUsesPatternMatching())
+                                        .onErrorResume(exception -> {
+                                            log.error("Error evicting cache", exception);
+                                            return Mono.empty();
+                                        })
+                                        .then(Mono.empty()))
                         .map(Function.identity());
             } catch (Throwable e) {
                 return Mono.error(e);
             }
         } else {
+            var value = invocation.proceed();
             try {
-                var value = invocation.proceed();
-                if (value == null) {
-                    return cacheManager.get().evict(key, cacheEvict.keyUsesPatternMatching()).then(Mono.empty());
-                }
-                return cacheManager.get().evict(key, cacheEvict.keyUsesPatternMatching()).thenReturn(value);
-            } catch (Throwable e) {
-                return Mono.error(e);
+                // Evict cache asynchronously to avoid blocking the execution thread. Makes cache eventual consistent
+                cacheManager.get().evictAsync(key, cacheEvict.keyUsesPatternMatching());
+            } catch (RuntimeException exception) {
+                log.error("Error evicting async cache", exception);
             }
+            return value;
         }
     }
 
-    private Mono<Object> processCachePutMethod(MethodInvocation invocation, boolean isReactive, String key,
-            String group) {
+    private Object processCachePutMethod(
+            MethodInvocation invocation, boolean isReactive, String key, String group) throws Throwable {
         if (isReactive) {
             try {
                 return ((Mono<?>) invocation.proceed()).flatMap(value -> cachePut(value, key, group));
             } catch (Throwable e) {
                 return Mono.error(e);
             }
         } else {
-            try {
-                var value = invocation.proceed();
-                return cachePut(value, key, group).thenReturn(value);
-            } catch (Throwable e) {
-                return Mono.error(e);
-            }
+            return processSyncCacheMiss(invocation, key, group);
         }
     }
 
-    private Object processCacheableMethod(MethodInvocation invocation, boolean isReactive, String key,
-            String group, Cacheable cacheable) {
-
+    private Object processCacheableMethod(
+            MethodInvocation invocation, boolean isReactive, String key, String group, Cacheable cacheable)
+            throws Throwable {
         if (isReactive) {
-
             if (invocation.getMethod().getReturnType().isAssignableFrom(Mono.class)) {
                 return handleMono(invocation, key, group, cacheable);
             } else {
                 return handleFlux(invocation, key, group, cacheable);
             }
         } else {
-
-            if (cacheable.wrapperType() != Object.class) {
-                TypeReference typeReference = TypeReferenceUtils.forTypes(cacheable.wrapperType(),
-                        cacheable.returnType());
-
-                return cacheManager.get().get(key, typeReference)
-                        .switchIfEmpty(processSyncCacheMiss(invocation, key, group));
+            Object cachedValue;
+            try {
+                if (cacheable.wrapperType() != Object.class) {
+                    var typeReference = TypeReferenceUtils.forTypes(
+                            cacheable.wrapperType(), cacheable.returnType());
+                    cachedValue = cacheManager.get().getSync(key, typeReference);
+                } else {
+                    cachedValue = cacheManager.get().getSync(key, invocation.getMethod().getReturnType());
+                }
+            } catch (RuntimeException exception) {
+                log.error("Error getting value synchronously from cache", exception);
+                cachedValue = null; // Treat as cache miss
             }
-
-            return cacheManager.get().get(key, invocation.getMethod().getReturnType())
-                    .map(Object.class::cast)
-                    .switchIfEmpty(processSyncCacheMiss(invocation, key, group));
+            if (cachedValue != null) {
+                return cachedValue;
+            }
+            return processSyncCacheMiss(invocation, key, group);
         }
     }
 
@@ -189,6 +189,10 @@ private Flux<Object> getFromCacheOrCallMethod(MethodInvocation invocation, Strin
             TypeReference<List<?>> collectionType) {
         return cacheManager.get()
                 .get(key, collectionType)
+                .onErrorResume(exception -> {
+                    log.error("Error getting value from cache", exception);
+                    return Mono.empty(); // Treat as cache miss
+                })
                 .map(Collection.class::cast)
                 .flatMapMany(Flux::fromIterable)
                 .switchIfEmpty(processFluxCacheMiss(invocation, key, group));
@@ -200,22 +204,26 @@ private Mono<Object> handleMono(MethodInvocation invocation, String key, String
                     cacheable.returnType());
 
             return cacheManager.get().get(key, typeReference)
+                    .onErrorResume(exception -> {
+                        log.error("Error getting value from cache", exception);
+                        return Mono.empty(); // Treat as cache miss
+                    })
                     .switchIfEmpty(processCacheMiss(invocation, key, group));
         }
 
         return cacheManager.get().get(key, cacheable.returnType())
+                .onErrorResume(exception -> {
+                    log.error("Error getting value from cache", exception);
+                    return Mono.empty(); // Treat as cache miss
+                })
                 .map(Object.class::cast)
                 .switchIfEmpty(processCacheMiss(invocation, key, group));
     }
 
-    private Mono<Object> processSyncCacheMiss(MethodInvocation invocation, String key, String group) {
-        return Mono.defer(() -> {
-            try {
-                return Mono.just(invocation.proceed());
-            } catch (Throwable e) {
-                return Mono.error(e);
-            }
-        }).flatMap(value -> cachePut(value, key, group));
+    private Object processSyncCacheMiss(MethodInvocation invocation, String key, String group) throws Throwable {
+        var value = invocation.proceed();
+        cachePutAsync(value, key, group);
+        return value;
     }
 
     private Mono<Object> processCacheMiss(MethodInvocation invocation, String key, String group) {
@@ -239,11 +247,9 @@ private Flux<Object> processFluxCacheMiss(MethodInvocation invocation, String ke
                         .flatMap(value -> cachePut(value, key, group));
 
                 return flux
-                        .doOnSubscribe(subscription -> Schedulers.boundedElastic().schedule(() -> {
-                            cacheable.subscribe(
-                                    __ -> log.info("Flux value put in cache"),
-                                    e -> log.error("Error putting flux value in cache", e));
-                        }));
+                        .doOnSubscribe(subscription -> Schedulers.boundedElastic().schedule(() -> cacheable.subscribe(
+                                __ -> log.debug("Flux value put in cache"),
+                                e -> log.error("Error putting flux value in cache", e))));
             } catch (Throwable e) {
                 return Flux.error(e);
             }
@@ -261,10 +267,25 @@ private Mono<Object> cachePut(Object value, String key, String group) {
                 });
     }
 
+    private void cachePutAsync(Object value, String key, String group) {
+        // Methods returning null values are not cached
+        if (value == null) {
+            return;
+        }
+        try {
+            var ttlDuration = cacheConfiguration.getCaches()
+                    .getOrDefault(group, cacheConfiguration.getDefaultDuration());
+            // Set cache asynchronously to avoid blocking the execution thread. Makes cache eventual consistent
+            cacheManager.get().putAsync(key, value, ttlDuration);
+        } catch (RuntimeException exception) {
+            log.error("Error putting async value in cache", exception);
+        }
+    }
+
     private String getKeyName(String name, String key, MethodInvocation invocation) {
         Map<String, Object> params = new HashMap<>();
 
-        // Use Paranamer to resolve parameter names
+        // Use Parameter to resolve parameter names
         Parameter[] parameters = invocation.getMethod().getParameters();
         Object[] args = invocation.getArguments();
 
@@ -281,5 +302,4 @@ private String getKeyName(String name, String key, MethodInvocation invocation)
         }
         return "%s:-%s".formatted(name, evaluatedKey);
     }
-
 }

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/cache/CacheManager.java

@@ -5,13 +5,35 @@
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
+import java.util.concurrent.CompletionStage;
 
+/**
+ * Cache management interface providing both reactive and non-reactive APIs.
+ *
+ * <p><b>API Types:</b></p>
+ * <ul>
+ *   <li><b>Reactive API</b>: Methods without {@code Sync} or {@code Async} suffix (e.g., {@code get()}, {@code put()})
+ *      are designed for use with reactive methods that return {@code Mono} or {@code Flux}.</li>
+ *   <li><b>Non-Reactive API</b>: Methods with {@code Sync} or {@code Async} suffix are designed for non-reactive methods:
+ *     <ul>
+ *       <li><b>Synchronous</b>: Methods with {@code Sync} suffix (e.g., {@code getSync()}) block until the result is available.</li>
+ *       <li><b>Asynchronous</b>: Methods with {@code Async} suffix (e.g., {@code putAsync()}) return {@link CompletionStage} for non-blocking operations without reactiveness.</li>
+ *     </ul>
+ *   </li>
+ * </ul>
+ */
 public interface CacheManager {
 
+    // Reactive API (for reactive methods returning Mono/Flux)
     Mono<Boolean> evict(@NonNull String key, boolean usePatternMatching);
     Mono<Boolean> put(@NonNull String key, @NonNull Object value, @NonNull Duration ttlDuration);
     <T> Mono<T> get(@NonNull String key, @NonNull Class<T> clazz);
     <T> Mono<T> get(@NonNull String key, @NonNull TypeReference<T> clazz);
     Mono<Boolean> contains(@NonNull String key);
 
+    // Non-reactive API (both synchronous and asynchronous methods without reactiveness)
+    CompletionStage<Boolean> evictAsync(String key, boolean usePatternMatching);
+    CompletionStage<Void> putAsync(String key, Object value, Duration ttlDuration);
+    <T> T getSync(String key, Class<T> clazz);
+    <T> T getSync(String key, TypeReference<T> clazz);
 }

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/redis/RedisCacheManager.java

@@ -6,16 +6,21 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
+import org.redisson.api.RedissonClient;
 import org.redisson.api.RedissonReactiveClient;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 import java.time.Duration;
+import java.util.concurrent.CompletionStage;
 
 @RequiredArgsConstructor
 class RedisCacheManager implements CacheManager {
 
     private final @NonNull RedissonReactiveClient redisClient;
+    private final @NonNull RedissonClient nonReactiveRedisClient;
+
+    // Reactive API (for reactive methods returning Mono/Flux)
 
     public Mono<Boolean> evict(@NonNull String key, boolean usePatternMatching) {
         if (usePatternMatching) {
@@ -51,4 +56,39 @@ public <T> Mono<T> get(@NonNull String key, @NonNull TypeReference<T> clazz) {
     public Mono<Boolean> contains(@NonNull String key) {
         return redisClient.getBucket(key).isExists();
     }
+
+    // Non-reactive API (both synchronous and asynchronous methods without reactiveness)
+
+    @Override
+    public CompletionStage<Boolean> evictAsync(@NonNull String key, boolean usePatternMatching) {
+        if (usePatternMatching) {
+            return nonReactiveRedisClient.getKeys().deleteByPatternAsync(key)
+                    .thenApplyAsync(count -> count > 0);
+        }
+        return nonReactiveRedisClient.getBucket(key).deleteAsync();
+    }
+
+    @Override
+    public CompletionStage<Void> putAsync(@NonNull String key, @NonNull Object value, @NonNull Duration ttlDuration) {
+        var json = JsonUtils.writeValueAsString(value);
+        return nonReactiveRedisClient.getBucket(key).setAsync(json, ttlDuration);
+    }
+
+    @Override
+    public <T> T getSync(@NonNull String key, @NonNull Class<T> clazz) {
+        var json = nonReactiveRedisClient.<String>getBucket(key).get();
+        if (StringUtils.isEmpty(json)) {
+            return null;
+        }
+        return JsonUtils.readValue(json, clazz);
+    }
+
+    @Override
+    public <T> T getSync(@NonNull String key, @NonNull TypeReference<T> typeReference) {
+        var json = nonReactiveRedisClient.<String>getBucket(key).get();
+        if (StringUtils.isEmpty(json)) {
+            return null;
+        }
+        return JsonUtils.readValue(json, typeReference);
+    }
 }