[OPIK-4383] [BE] Add ExperimentAggregationPublisher, ExperimentDenormalizationJob and tests

apps/opik-backend/config.yml

@@ -268,6 +268,12 @@ experimentDenormalization:
   # Default: 3
   # Description: Maximum number of retry attempts for failed messages (1-10)
   maxRetries: ${OPIK_REDIS_EXPERIMENT_DENORM_MAX_RETRIES:-3}
+  # Default: 100
+  # Description: Maximum number of ZSET entries to read per page when flushing pending experiments (10-1000)
+  jobBatchSize: ${OPIK_EXPERIMENT_DENORM_JOB_BATCH_SIZE:-100}
+  # Default: 5s
+  # Description: Interval at which the denormalization job runs to flush pending experiments (min 1s)
+  jobInterval: ${OPIK_EXPERIMENT_DENORM_JOB_INTERVAL:-5s}
   # Default: 10000
   # Description: Maximum length of the Redis stream. Older entries are trimmed when this limit is exceeded.
   streamMaxLen: ${OPIK_REDIS_EXPERIMENT_DENORM_STREAM_MAX_LEN:-10000}

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/ExperimentAggregateEventListener.java

@@ -65,7 +65,11 @@ public void onExperimentUpdated(ExperimentUpdated event) {
             return;
         }
         if (FINISHED_STATUSES.contains(event.newStatus())) {
-            publisher.publish(Set.of(event.experimentId()), event.workspaceId(), event.userName());
+            publisher.publish(Set.of(event.experimentId()), event.workspaceId(), event.userName())
+                    .subscribe(null,
+                            e -> log.error(
+                                    "Error triggering aggregation for experiment '{}' in workspace '{}'",
+                                    event.experimentId(), event.workspaceId(), e));
         }
     }
 
@@ -207,8 +211,14 @@ private Mono<Void> triggerAggregation(Set<UUID> entityIds, String workspaceId, S
                 .contextWrite(ctx -> ctx
                         .put(RequestContext.WORKSPACE_ID, workspaceId)
                         .put(RequestContext.USER_NAME, userName))
-                .filter(CollectionUtils::isNotEmpty)
-                .doOnNext(experimentIds -> publisher.publish(experimentIds, workspaceId, userName))
-                .then();
+                .flatMap(experimentIds -> publishIfNotEmpty(experimentIds, workspaceId, userName));
+    }
+
+    private Mono<Void> publishIfNotEmpty(Set<UUID> experimentIds, String workspaceId, String userName) {
+        if (CollectionUtils.isEmpty(experimentIds)) {
+            log.warn("No finished experiments to publish for workspace '{}'", workspaceId);
+            return Mono.empty();
+        }
+        return publisher.publish(experimentIds, workspaceId, userName);
     }
 }

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/jobs/ExperimentDenormalizationJob.java

@@ -0,0 +1,188 @@
+package com.comet.opik.api.resources.v1.jobs;
+
+import com.comet.opik.api.events.ExperimentAggregationMessage;
+import com.comet.opik.infrastructure.ExperimentDenormalizationConfig;
+import com.comet.opik.infrastructure.lock.LockService;
+import io.dropwizard.jobs.Job;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.InterruptableJob;
+import org.quartz.JobExecutionContext;
+import org.quartz.UnableToInterruptJobException;
+import org.redisson.api.RedissonReactiveClient;
+import org.redisson.api.stream.StreamAddArgs;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import ru.vyarus.dropwizard.guice.module.yaml.bind.Config;
+
+import java.time.Instant;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+
+import static com.comet.opik.infrastructure.lock.LockService.Lock;
+
+/**
+ * Scheduled job responsible for flushing debounced experiment aggregation events to the Redis stream.
+ *
+ * <p>Periodically (default 5s, configurable via {@code experimentDenormalization.jobInterval}) this job:
+ * <ol>
+ *   <li>Queries the Redis ZSET index for members whose debounce window has elapsed.</li>
+ *   <li>Each ZSET member encodes both workspaceId and experimentId as {@code "workspaceId:experimentId"},
+ *       ensuring cross-workspace isolation for experiments that share the same UUID.</li>
+ *   <li>Reads the userName from the associated Redis hash bucket.</li>
+ *   <li>Publishes an {@link ExperimentAggregationMessage} to the Redis stream for each ready experiment.</li>
+ *   <li>Removes the processed entry from both the ZSET index and the hash bucket.</li>
+ *   <li>Handles stale ZSET entries (expired hash) by removing them without publishing.</li>
+ * </ol>
+ *
+ * <p>Uses a ZSET scored by expiry timestamp for O(log(N)+M) index lookups, avoiding full keyspace scans.
+ */
+@Slf4j
+@Singleton
+@DisallowConcurrentExecution
+public class ExperimentDenormalizationJob extends Job implements InterruptableJob {
+
+    private static final Lock SCAN_LOCK_KEY = new Lock("experiment_denormalization_job:scan_lock");
+
+    private final AtomicBoolean interrupted = new AtomicBoolean(false);
+    private final ExperimentDenormalizationConfig config;
+    private final RedissonReactiveClient redisClient;
+    private final LockService lockService;
+
+    @Inject
+    public ExperimentDenormalizationJob(
+            @NonNull @Config("experimentDenormalization") ExperimentDenormalizationConfig config,
+            @NonNull RedissonReactiveClient redisClient,
+            @NonNull LockService lockService) {
+        this.config = config;
+        this.redisClient = redisClient;
+        this.lockService = lockService;
+    }
+
+    @Override
+    public void doJob(JobExecutionContext context) {
+        if (!config.isEnabled()) {
+            log.debug("Experiment denormalization job is disabled, skipping");
+            return;
+        }
+
+        // Check for interruption before starting
+        if (interrupted.get()) {
+            log.info("Experiment denormalization job interrupted before execution, skipping");
+            return;
+        }
+
+        log.info("Starting experiment denormalization job - checking for pending experiments");
+
+        lockService.bestEffortLock(
+                SCAN_LOCK_KEY,
+                Mono.defer(() -> getExperimentsReadyToProcess()
+                        .flatMap(this::processExperiment)
+                        .onErrorContinue((throwable, experimentId) -> log.error(
+                                "Failed to process pending experiment '{}'",
+                                experimentId, throwable))
+                        .doOnComplete(
+                                () -> log.info(
+                                        "Experiment denormalization job finished processing all ready experiments"))
+                        .then()),
+                Mono.defer(() -> {
+                    log.info(
+                            "Could not acquire lock for scanning pending experiments, another job instance is running");
+                    return Mono.empty();
+                }),
+                config.getJobLockTime().toJavaDuration(),
+                config.getJobLockWaitTime().toJavaDuration())
+                .subscribe(
+                        __ -> log.info("Experiment denormalization job execution completed"),
+                        error -> log.error("Experiment denormalization job interrupted while acquiring lock", error));
+    }
+
+    /**
+     * Queries the ZSET index in pages for experiment IDs whose debounce window has elapsed (score &lt;= now).
+     * Uses offset/count pagination to avoid materializing the entire ZSET into memory.
+     * Since each processed experiment is removed from the ZSET, we always query from offset 0.
+     * A safety counter caps the number of expand iterations (using batchSize as the limit) to
+     * prevent infinite loops if entries fail to be removed (e.g., due to errors swallowed by onErrorContinue).
+     */
+    private Flux<String> getExperimentsReadyToProcess() {
+        long nowMillis = Instant.now().toEpochMilli();
+        int batchSize = config.getJobBatchSize();
+        var index = redisClient.<String>getScoredSortedSet(ExperimentDenormalizationConfig.PENDING_SET_KEY);
+        var iterations = new int[]{0};
+
+        log.debug("Checking for experiments ready to process (up to timestamp: '{}', batchSize: '{}')",
+                nowMillis, batchSize);
+
+        if (interrupted.get()) {
+            log.info("Experiment denormalization job interrupted before execution, skipping");
+            return Flux.empty();
+        }
+
+        return index.valueRange(Double.NEGATIVE_INFINITY, true, nowMillis, true, 0, batchSize)
+                .expand(collection -> {
+                    if (collection.size() < batchSize) {
+                        return Mono.empty();
+                    }
+                    iterations[0]++;
+                    if (iterations[0] >= batchSize) {
+                        log.warn("Reached maximum expand iterations '{}', stopping pagination to prevent infinite loop",
+                                batchSize);
+                        return Mono.empty();
+                    }
+                    return index.valueRange(Double.NEGATIVE_INFINITY, true, nowMillis, true, 0, batchSize);
+                })
+                .flatMapIterable(Function.identity())
+                .map(Object::toString);
+    }
+
+    /**
+     * Processes a single pending experiment: publishes a stream message and cleans up the Redis state.
+     * The {@code member} is a compound key of the form {@code "workspaceId:experimentId"}, which
+     * ensures experiments with the same UUID in different workspaces are handled independently.
+     * If the hash bucket has already expired (stale ZSET entry), only the ZSET entry is removed.
+     */
+    private Mono<Void> processExperiment(String member) {
+        int separatorIndex = member.indexOf(ExperimentDenormalizationConfig.MEMBER_SEPARATOR);
+        String workspaceId = member.substring(0, separatorIndex);
+        String experimentIdStr = member.substring(separatorIndex + 1);
+
+        log.info("Processing pending experiment: '{}' for workspace: '{}'", experimentIdStr, workspaceId);
+
+        var bucket = redisClient.<String, String>getMap(ExperimentDenormalizationConfig.EXPERIMENT_KEY_PREFIX + member);
+        var index = redisClient.getScoredSortedSet(ExperimentDenormalizationConfig.PENDING_SET_KEY);
+        var stream = redisClient.getStream(config.getStreamName(), config.getCodec());
+
+        return bucket.get(ExperimentDenormalizationConfig.USER_NAME_FIELD)
+                .flatMap(userName -> {
+                    var message = ExperimentAggregationMessage.builder()
+                            .experimentId(UUID.fromString(experimentIdStr))
+                            .workspaceId(workspaceId)
+                            .userName(userName)
+                            .build();
+
+                    return stream.add(StreamAddArgs.entry(ExperimentDenormalizationConfig.PAYLOAD_FIELD, message))
+                            .doOnNext(id -> log.info(
+                                    "Enqueued aggregation message for experiment '{}' with stream id '{}'",
+                                    experimentIdStr, id))
+                            .then(bucket.delete())
+                            .then(index.remove(member));
+                })
+                .switchIfEmpty(Mono.defer(() -> {
+                    log.warn("Stale index entry found with no bucket data, removing member: '{}'", member);
+                    return index.remove(member);
+                }))
+                .then()
+                .doOnSuccess(__ -> log.info("Successfully processed and removed pending experiment: '{}'",
+                        experimentIdStr));
+    }
+
+    @Override
+    public void interrupt() throws UnableToInterruptJobException {
+        interrupted.set(true);
+        log.info("ExperimentDenormalizationJob reaper job interrupted");
+    }
+}

apps/opik-backend/src/main/java/com/comet/opik/domain/experiments/aggregations/ExperimentAggregationPublisher.java

@@ -1,34 +1,69 @@
 package com.comet.opik.domain.experiments.aggregations;
 
+import com.comet.opik.infrastructure.ExperimentDenormalizationConfig;
 import com.google.inject.ImplementedBy;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RMapReactive;
+import org.redisson.api.RedissonReactiveClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import ru.vyarus.dropwizard.guice.module.yaml.bind.Config;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Set;
 import java.util.UUID;
 
 @ImplementedBy(ExperimentAggregationPublisher.ExperimentAggregationPublisherImpl.class)
 public interface ExperimentAggregationPublisher {
 
-    void publish(@NonNull Set<UUID> experimentIds, @NonNull String workspaceId, @NonNull String userName);
+    Mono<Void> publish(@NonNull Set<UUID> experimentIds, @NonNull String workspaceId, @NonNull String userName);
 
     @Singleton
     @Slf4j
     class ExperimentAggregationPublisherImpl implements ExperimentAggregationPublisher {
 
+        private final ExperimentDenormalizationConfig config;
+        private final RedissonReactiveClient redisClient;
+
         @Inject
-        ExperimentAggregationPublisherImpl() {
+        ExperimentAggregationPublisherImpl(
+                @NonNull @Config("experimentDenormalization") ExperimentDenormalizationConfig config,
+                @NonNull RedissonReactiveClient redisClient) {
+            this.config = config;
+            this.redisClient = redisClient;
         }
 
         @Override
-        public void publish(@NonNull Set<UUID> experimentIds, @NonNull String workspaceId,
+        public Mono<Void> publish(@NonNull Set<UUID> experimentIds, @NonNull String workspaceId,
                 @NonNull String userName) {
-            // TODO: implement debounce mechanism before enqueuing to Redis stream
-            log.debug(
-                    "Experiment aggregation publish skipped for experiments '{}': debounce mechanism not yet implemented",
-                    experimentIds);
+            if (!config.isEnabled() || experimentIds.isEmpty()) {
+                log.info("Skipping publish: enabled='{}', experimentIds.size='{}'",
+                        config.isEnabled(), experimentIds.size());
+                return Mono.empty();
+            }
+
+            Instant expiryTimestamp = Instant.now().plusMillis(config.getDebounceDelay().toMilliseconds());
+            var index = redisClient.getScoredSortedSet(ExperimentDenormalizationConfig.PENDING_SET_KEY);
+
+            return Flux.fromIterable(experimentIds)
+                    .flatMap(experimentId -> {
+                        String member = workspaceId + ":" + experimentId;
+                        RMapReactive<String, String> bucket = redisClient
+                                .getMap(ExperimentDenormalizationConfig.EXPERIMENT_KEY_PREFIX + member);
+
+                        return index.add(expiryTimestamp.toEpochMilli(), member)
+                                .then(bucket.put(ExperimentDenormalizationConfig.USER_NAME_FIELD, userName))
+                                .then(bucket.expire(Duration.ofMillis(config.getDebounceDelay().toMilliseconds() * 2)))
+                                .doOnSuccess(__ -> log.info(
+                                        "Enqueued experiment '{}' for workspace '{}' in pending bucket with expiryTimestamp='{}'",
+                                        experimentId, workspaceId, expiryTimestamp));
+                    })
+                    .doOnError(error -> log.error("Error enqueueing experiments in pending bucket", error))
+                    .then();
         }
     }
 }

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

@@ -21,6 +21,9 @@ public class ExperimentDenormalizationConfig implements StreamConfiguration {
 
     public static final String PAYLOAD_FIELD = "message";
     public static final String PENDING_SET_KEY = "experiment:denorm:pending";
+    public static final String EXPERIMENT_KEY_PREFIX = PENDING_SET_KEY + ":";
+    public static final String USER_NAME_FIELD = "userName";
+    public static final String MEMBER_SEPARATOR = ":";
 
     @JsonProperty
     private boolean enabled = false;
@@ -76,6 +79,13 @@ public class ExperimentDenormalizationConfig implements StreamConfiguration {
     @JsonProperty
     @Min(1) @Max(10) private int maxRetries = 3;
 
+    @Valid @JsonProperty
+    @Min(10) @Max(1000) private int jobBatchSize = 100;
+
+    @Valid @JsonProperty
+    @NotNull @MinDuration(value = 1, unit = TimeUnit.SECONDS)
+    private Duration jobInterval = Duration.seconds(5);
+
     @Override
     @JsonIgnore
     public Codec getCodec() {

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/bi/OpikGuiceyLifecycleEventListener.java

@@ -1,8 +1,10 @@
 package com.comet.opik.infrastructure.bi;
 
 import com.comet.opik.api.resources.v1.jobs.DatasetVersionItemsTotalMigrationJob;
+import com.comet.opik.api.resources.v1.jobs.ExperimentDenormalizationJob;
 import com.comet.opik.api.resources.v1.jobs.TraceThreadsClosingJob;
 import com.comet.opik.domain.alerts.MetricsAlertJob;
+import com.comet.opik.infrastructure.ExperimentDenormalizationConfig;
 import com.comet.opik.infrastructure.OpikConfiguration;
 import com.comet.opik.infrastructure.TraceThreadConfig;
 import com.google.inject.Injector;
@@ -42,6 +44,7 @@ public void onEvent(GuiceyLifecycleEvent event) {
                 setupDailyJob();
                 setTraceThreadsClosingJob();
                 setMetricsAlertJob();
+                setExperimentDenormalizationJob();
                 scheduleDatasetVersionItemsTotalMigrationJobIfEnabled();
             }
 
@@ -111,6 +114,41 @@ private void setTraceThreadsClosingJob() {
         }
     }
 
+    // This method sets up a job that periodically flushes debounced experiment aggregation events.
+    private void setExperimentDenormalizationJob() {
+        ExperimentDenormalizationConfig denormConfig = injector.get().getInstance(OpikConfiguration.class)
+                .getExperimentDenormalization();
+
+        if (!denormConfig.isEnabled()) {
+            log.info("Experiment denormalization job is disabled, skipping job setup");
+            return;
+        }
+
+        Duration jobInterval = denormConfig.getJobInterval().toJavaDuration();
+
+        var jobDetail = JobBuilder.newJob(ExperimentDenormalizationJob.class)
+                .storeDurably()
+                .build();
+
+        var trigger = TriggerBuilder.newTrigger()
+                .forJob(jobDetail)
+                .startNow()
+                .withSchedule(
+                        org.quartz.SimpleScheduleBuilder.simpleSchedule()
+                                .withIntervalInMilliseconds(jobInterval.toMillis())
+                                .repeatForever())
+                .build();
+
+        try {
+            var scheduler = getScheduler();
+            scheduler.addJob(jobDetail, false);
+            scheduler.scheduleJob(trigger);
+            log.info("Experiment denormalization job scheduled successfully with interval '{}'", jobInterval);
+        } catch (SchedulerException e) {
+            log.error("Failed to schedule experiment denormalization job", e);
+        }
+    }
+
     // This method sets up a job that periodically evaluates metrics-based alerts for cost and latency thresholds.
     private void setMetricsAlertJob() {
         try {