Ensure that failed unis are not cached

Fixes: quarkusio#39677

extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/UniReturnTypeWithFailureTest.java

@@ -0,0 +1,53 @@
+package io.quarkus.cache.test.runtime;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.cache.CacheResult;
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Uni;
+import io.vertx.core.impl.NoStackTraceException;
+
+public class UniReturnTypeWithFailureTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest TEST = new QuarkusUnitTest().withApplicationRoot((jar) -> jar.addClass(CachedService.class));
+
+    @Inject
+    CachedService cachedService;
+
+    @Test
+    void testCacheResult() {
+        assertThrows(NoStackTraceException.class, () -> cachedService.cacheResult("k1").await().indefinitely());
+        assertEquals(1, cachedService.getCacheResultInvocations());
+        assertEquals("", cachedService.cacheResult("k1").await().indefinitely());
+        assertEquals(2, cachedService.getCacheResultInvocations());
+        assertEquals("", cachedService.cacheResult("k1").await().indefinitely());
+        assertEquals(2, cachedService.getCacheResultInvocations());
+    }
+
+    @ApplicationScoped
+    static class CachedService {
+
+        private volatile int cacheResultInvocations;
+
+        @CacheResult(cacheName = "test-cache")
+        public Uni<String> cacheResult(String key) {
+            cacheResultInvocations++;
+            if (cacheResultInvocations == 1) {
+                return Uni.createFrom().failure(new NoStackTraceException("dummy"));
+            }
+            return Uni.createFrom().item(() -> new String());
+        }
+
+        public int getCacheResultInvocations() {
+            return cacheResultInvocations;
+        }
+    }
+}

extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/CacheResultInterceptor.java

@@ -71,6 +71,11 @@ public Uni<Object> apply(Object key) {
                             throw new CacheException(e);
                         }
                     }
+                }).onFailure().call(new Function<>() {
+                    @Override
+                    public Uni<?> apply(Throwable throwable) {
+                        return cache.invalidate(key).replaceWith(throwable);
+                    }
                 }).emitOn(new Executor() {
                     // We need make sure we go back to the original context when the cache value is computed.
                     // Otherwise, we would always emit on the context having computed the value, which could