[OPIK-8202] [BE] fix: harden ClickHouseAppender against dropped user logs

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java

@@ -8,6 +8,7 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -27,6 +28,10 @@
 @Slf4j
 class ClickHouseAppender extends AppenderBase<ILoggingEvent> {
 
+    private static final long INSERT_RETRY_ATTEMPTS = 3;
+    private static final Duration INSERT_RETRY_MIN_BACKOFF = Duration.ofMillis(100);
+    private static final Retry INSERT_RETRY = Retry.backoff(INSERT_RETRY_ATTEMPTS, INSERT_RETRY_MIN_BACKOFF);
+
     private static ClickHouseAppender instance;
 
     public static synchronized ClickHouseAppender init(@NonNull UserLogTableFactory userLogTableFactory, int batchSize,
@@ -58,13 +63,23 @@ private static void setInstance(ClickHouseAppender instance) {
 
     @Override
     public void start() {
-        // Background flush thread
-        scheduler.get().scheduleAtFixedRate(this::flushLogs, flushIntervalDuration.toMillis(),
+        // Background flush thread. Wrapped in safeFlushLogs because scheduleAtFixedRate silently
+        // cancels the task forever if it ever throws, which would stop persisting user logs for the
+        // rest of the JVM's life.
+        scheduler.get().scheduleAtFixedRate(this::safeFlushLogs, flushIntervalDuration.toMillis(),
                 flushIntervalDuration.toMillis(), TimeUnit.MILLISECONDS);
 
         super.start();
     }
 
+    private void safeFlushLogs() {
+        try {
+            flushLogs();
+        } catch (Exception e) {
+            log.error("Failed to flush logs", e);
+        }
+    }
+
     private void flushLogs() {
         if (logQueue.isEmpty()) return;
 
@@ -87,16 +102,43 @@ private void flushLogs() {
                         UserLogTableFactory.UserLogTableDAO tableDAO = userLogTableFactory
                                 .getDAO(UserLog.valueOf(userLog));
 
-                        tableDAO
-                                .saveAll(events)
-                                .subscribe(
-                                        noop -> {
-                                        },
-                                        e -> log.error("Failed to insert logs", e));
+                        // The batch is already drained out of the queue, so a failed insert would lose
+                        // these events outright. Retry transient failures, and put the events back on
+                        // the queue if the retries are exhausted so a later flush can pick them up.
+                        try {
+                            tableDAO
+                                    .saveAll(events)
+                                    .retryWhen(INSERT_RETRY)
+                                    .subscribe(
+                                            noop -> {
+                                            },
+                                            e -> {
+                                                log.error("Failed to insert logs", e);
+                                                requeue(events);
+                                            });
+                        } catch (Exception e) {
+                            // A synchronous throw from saveAll means the batch itself is rejected before
+                            // any I/O (e.g. an event missing workspace_id or rule_id). Requeueing would
+                            // fail identically on every later flush, so drop it instead of looping.
+                            log.error("Dropping '{}' logs rejected before insert", events.size(), e);
+                        }
                     }
                 });
     }
 
+    private void requeue(List<ILoggingEvent> events) {
+        if (!running) {
+            log.warn("ClickHouseAppender is stopped, dropping '{}' logs after failed insert", events.size());
+            return;
+        }
+
+        for (ILoggingEvent event : events) {
+            if (!logQueue.offer(event)) {
+                log.warn("Log queue is full, dropping log: {}", event.getFormattedMessage());
+            }
+        }
+    }
+
     @Override
     protected void append(ILoggingEvent event) {
         if (!running) {
@@ -110,7 +152,7 @@ protected void append(ILoggingEvent event) {
         }
 
         if (logQueue.size() >= batchSize) {
-            scheduler.get().execute(this::flushLogs);
+            scheduler.get().execute(this::safeFlushLogs);
         }
     }
 

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/AutomationRuleEvaluatorsResourceTest.java

@@ -99,6 +99,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -244,6 +245,11 @@ def score(
 
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
+    // The evaluator log rows these tests wait for are written by the async ClickHouseAppender, which
+    // flushes every 500ms. Stated explicitly rather than relying on Awaitility's implicit 10s default,
+    // which is easy to miss when reading the waits below.
+    private static final int AWAIT_TIMEOUT_SECONDS = 30;
+
     private final RedisContainer redis = RedisContainerUtils.newRedisContainer();
     private final MySQLContainer mysql = MySQLContainerUtils.newMySQLContainer();
     private final GenericContainer<?> zookeeper = ClickHouseContainerUtils.newZookeeperContainer();
@@ -641,7 +647,7 @@ void getLogsPerRuleEvaluators__whenSessionTokenIsPresent__thenReturnProperRespon
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 try (var actualResponse = evaluatorsResourceClient.getLogsWithSessionToken(
                         id, sessionToken, workspaceName)) {
                     if (isAuthorized) {
@@ -1435,7 +1441,7 @@ void getLogsLlmAsJudgeScorer(LlmProviderFactory llmProviderFactory) throws JsonP
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 var logPage = evaluatorsResourceClient.getLogs(id, WORKSPACE_NAME, API_KEY);
                 assertTraceLogResponse(logPage, id, trace);
             });
@@ -1472,7 +1478,7 @@ void getLogsTraceThreadLlmAsJudgeScorer(LlmProviderFactory llmProviderFactory) t
             Instant createdAt = trace.createdAt();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
 
                 TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY,
                         WORKSPACE_NAME);
@@ -1531,7 +1537,7 @@ void getLogsUserDefinedMetricPythonScorer() throws JsonProcessingException {
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 var logPage = evaluatorsResourceClient.getLogs(id, WORKSPACE_NAME, API_KEY);
                 assertTraceLogResponse(logPage, id, trace);
             });
@@ -1590,7 +1596,7 @@ void getLogsTraceThreadUserDefinedMetricPythonScorer() throws JsonProcessingExce
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
             // Then
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY,
                         WORKSPACE_NAME);
 
@@ -1641,7 +1647,7 @@ void getLogsTraceSkipped() {
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 var logPagePython = evaluatorsResourceClient.getLogs(idPython, WORKSPACE_NAME, API_KEY);
                 assertLogResponse(logPagePython, idPython, evaluatorPython, trace);
                 var logPageLlm = evaluatorsResourceClient.getLogs(idLlm, WORKSPACE_NAME, API_KEY);
@@ -1693,7 +1699,7 @@ void getLogsTraceThreadSkipped() {
             Instant createdAt = trace.createdAt();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY,
                         WORKSPACE_NAME);
 
@@ -1783,7 +1789,7 @@ void getLogsTraceSkippedDueToDisabledRule() {
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 var logPagePython = evaluatorsResourceClient.getLogs(idPython, WORKSPACE_NAME, API_KEY);
                 assertDisabledRuleLogResponse(logPagePython, idPython, evaluatorPython, trace);
                 var logPageLlm = evaluatorsResourceClient.getLogs(idLlm, WORKSPACE_NAME, API_KEY);
@@ -1835,7 +1841,7 @@ void getLogsTraceThreadSkippedDueToDisabledRule() {
             Instant createdAt = trace.createdAt();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY,
                         WORKSPACE_NAME);
 
@@ -1894,7 +1900,7 @@ void mixedEnabledAndDisabledRules() {
                     .build();
             traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME);
 
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 // Enabled rule should generate sampling rate message (skipped due to 0% rate)
                 var enabledLogPage = evaluatorsResourceClient.getLogs(enabledId, WORKSPACE_NAME, API_KEY);
                 assertLogResponse(enabledLogPage, enabledId, enabledRule, trace);
@@ -1933,7 +1939,7 @@ void disabledRuleDoesNotConsumeResources() {
             }
 
             // All traces should be skipped with "disabled" message, none with sampling rate message
-            Awaitility.await().untilAsserted(() -> {
+            Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> {
                 var logPage = evaluatorsResourceClient.getLogs(disabledId, WORKSPACE_NAME, API_KEY);
                 assertLogPage(logPage, 5); // Should have 5 log entries
 

apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java

@@ -0,0 +1,136 @@
+package com.comet.opik.infrastructure.log;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEvent;
+import com.comet.opik.domain.evaluators.UserLog;
+import com.comet.opik.infrastructure.log.tables.UserLogTableFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@DisplayName("ClickHouse Appender")
+class ClickHouseAppenderTest {
+
+    private static final Duration FLUSH_INTERVAL = Duration.ofMillis(50);
+    private static final int BATCH_SIZE = 1000;
+    private static final int AWAIT_TIMEOUT_SECONDS = 10;
+
+    private ClickHouseAppender appender;
+
+    @AfterEach
+    void tearDown() {
+        if (appender != null) {
+            appender.stop();
+        }
+    }
+
+    @Test
+    @DisplayName("when the insert fails transiently, then the events are retried and still persisted")
+    void whenInsertFailsTransiently__thenEventsAreRetriedAndPersisted() {
+        var attempts = new AtomicInteger();
+        var persisted = new ConcurrentLinkedQueue<ILoggingEvent>();
+
+        // Fail the first attempt, then accept. Without the retry, the drained batch would be lost.
+        startAppender(events -> Mono.defer(() -> {
+            if (attempts.incrementAndGet() == 1) {
+                return Mono.error(new IllegalStateException("transient failure"));
+            }
+            persisted.addAll(events);
+            return Mono.empty();
+        }));
+
+        appender.doAppend(event("a message"));
+
+        // The attempt count is what pins Retry.backoff: without it a batch requeued and picked up by a
+        // later flush would also persist eventually, so hasSize(1) alone would not prove a retry ran.
+        await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    assertThat(persisted).hasSize(1);
+                    assertThat(attempts).hasValue(2);
+                });
+    }
+
+    @Test
+    @DisplayName("when the batch is rejected before insert, then it is dropped rather than requeued")
+    void whenBatchRejectedBeforeInsert__thenItIsDroppedRatherThanRequeued() {
+        var attempts = new AtomicInteger();
+
+        // A synchronous throw means the batch is malformed (e.g. missing workspace_id), so it would
+        // fail identically forever. Requeueing it would spin the flush thread on a poison batch.
+        startAppender(events -> {
+            attempts.incrementAndGet();
+            throw new IllegalStateException("workspace_id is not set");
+        });
+
+        appender.doAppend(event("a malformed message"));
+
+        await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertThat(attempts).hasValue(1));
+
+        // Give several further flush intervals a chance to re-attempt the dropped batch.
+        await().during(FLUSH_INTERVAL.multipliedBy(10))
+                .atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertThat(attempts).hasValue(1));
+    }
+
+    @Test
+    @DisplayName("when a flush throws, then subsequent flushes still run")
+    void whenFlushThrows__thenSubsequentFlushesStillRun() {
+        var persisted = new ConcurrentLinkedQueue<ILoggingEvent>();
+        var failNext = new AtomicInteger(1);
+
+        // A throw escaping into scheduleAtFixedRate would cancel the periodic flush permanently,
+        // silently stopping log persistence for the rest of the JVM's life.
+        startAppender(events -> {
+            if (failNext.getAndSet(0) == 1) {
+                throw new IllegalStateException("failure inside flush");
+            }
+            persisted.addAll(events);
+            return Mono.empty();
+        });
+
+        appender.doAppend(event("first"));
+
+        // The first batch is dropped (it was rejected before insert), so the surviving scheduler is
+        // proven by a later event still being persisted rather than by the failed batch reappearing.
+        await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertThat(failNext).hasValue(0));
+
+        appender.doAppend(event("second"));
+
+        await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertThat(persisted).hasSize(1));
+    }
+
+    private void startAppender(SaveAll saveAll) {
+        UserLogTableFactory tableFactory = userLog -> saveAll::apply;
+
+        appender = ClickHouseAppender.init(tableFactory, BATCH_SIZE, FLUSH_INTERVAL, new LoggerContext());
+    }
+
+    private ILoggingEvent event(String message) {
+        var event = new LoggingEvent();
+        event.setLevel(Level.INFO);
+        event.setMessage(message);
+        event.setMDCPropertyMap(Map.of(UserLog.MARKER, UserLog.AUTOMATION_RULE_EVALUATOR.name()));
+        return event;
+    }
+
+    @FunctionalInterface
+    private interface SaveAll {
+        Mono<Void> apply(List<ILoggingEvent> events);
+    }
+}