[backend] feat: batching injects traces (#2830)

.drone.yml

@@ -15,6 +15,7 @@ steps:
       MINIO_ENDPOINT: minio
       MINIO_PORT: 9000
       ENGINE_URL: http://elastic:9200
+      OPENBAS_RABBITMQ_HOSTNAME: rabbitmq
     commands:
       - mvn spotless:check
       - sleep 60
@@ -53,6 +54,7 @@ steps:
       OPENAEV_ADMIN_PASSWORD: admin
       OPENAEV_ADMIN_TOKEN: 0d17ce9a-f3a8-4c6d-9721-c98dc3dc023f
       SPRING_PROFILES_ACTIVE: ci
+      OPENBAS_RABBITMQ_HOSTNAME: rabbitmq-e2e
     commands:
       - apt update && apt install -y gnupg
       - curl -sS https://dl.yarnpkg.com/debian/pubkey.gpg | apt-key add -
@@ -177,6 +179,13 @@ services:
       POSTGRES_USER: openaev
       POSTGRES_PASSWORD: openaev
       POSTGRES_DB: openaev
+    commands: 
+      - docker-entrypoint.sh -c 'max_connections=500'
+  - name: rabbitmq
+    image: rabbitmq:4.1-management
+    environment:
+      RABBITMQ_DEFAULT_USER: guest
+      RABBITMQ_DEFAULT_PASS: guest
   - name: minio-e2e
     image: minio/minio:RELEASE.2025-06-13T11-33-47Z
     environment:
@@ -189,6 +198,13 @@ services:
       POSTGRES_USER: openaev
       POSTGRES_PASSWORD: openaev
       POSTGRES_DB: openaev
+    commands: 
+      - docker-entrypoint.sh -c 'max_connections=500'
+  - name: rabbitmq-e2e
+    image: rabbitmq:4.1-management
+    environment:
+      RABBITMQ_DEFAULT_USER: guest
+      RABBITMQ_DEFAULT_PASS: guest
   - name: elastic
     image: docker.elastic.co/elasticsearch/elasticsearch:8.18.3
     environment:

openaev-api/src/main/java/io/openaev/config/CachingConfig.java

@@ -2,23 +2,41 @@
 
 import com.github.benmanes.caffeine.cache.Caffeine;
 import java.time.Duration;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.cache.CacheManager;
+import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cache.caffeine.CaffeineCacheManager;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.Scheduled;
 
 @Configuration
 @EnableCaching
+@Slf4j
 public class CachingConfig {
 
   @Bean
   public CacheManager cacheManager() {
-    CaffeineCacheManager cacheManager = new CaffeineCacheManager("license");
+    /**
+     * Creating some cache : - license is for the EE license - global for global settings that do
+     * not need to be fetched from the DB everytime we need it (like features flags) - adminUsers is
+     * a low retention cache for users that are admin. This is useful when receiving a lot of calls.
+     * Execution traces for instance can receive several thousands a sec and not fetching the user
+     * everytime helps for the RBAC
+     */
+    CaffeineCacheManager cacheManager = new CaffeineCacheManager("license", "global", "adminUsers");
 
     cacheManager.setCaffeine(
         Caffeine.newBuilder().expireAfterWrite(Duration.ofDays(1)).maximumSize(100));
 
     return cacheManager;
   }
+
+  /** Emptying the cache every second to avoid old data on the admin users being persisted */
+  @CacheEvict(value = "adminUsers", allEntries = true)
+  @Scheduled(fixedRateString = "1000")
+  public void emptyAdminUsersCache() {
+    log.info("emptying admin users cache");
+  }
 }

openaev-api/src/main/java/io/openaev/config/ThreadPoolTaskSchedulerConfig.java

@@ -1,17 +1,42 @@
 package io.openaev.config;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 @Configuration
+@Slf4j
 public class ThreadPoolTaskSchedulerConfig {
 
   @Bean
   public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
     ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
     threadPoolTaskScheduler.setPoolSize(20);
     threadPoolTaskScheduler.setThreadNamePrefix("ThreadPoolTaskScheduler");
+    threadPoolTaskScheduler.setErrorHandler(
+        t -> log.error("Error during scheduled task : {}", t.getMessage(), t));
     return threadPoolTaskScheduler;
   }
+
+  /** Dedicated executor for stream events */
+  @Bean(name = "streamExecutor")
+  public Executor streamExecutor() {
+    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+    executor.setCorePoolSize(5);
+    executor.setMaxPoolSize(10);
+    executor.setQueueCapacity(100);
+    executor.setThreadNamePrefix("Stream-");
+
+    // If we have more event to deal with than the available size in the waiting queue, we discard
+    // the oldest to prevent overloading the stream. This also helps a little preventing
+    // overloading the tab of a user connected when having a lot of events
+    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
+
+    executor.initialize();
+    return executor;
+  }
 }

openaev-api/src/main/java/io/openaev/migration/V4_53__Convert_expectations_to_jsonb.java

@@ -0,0 +1,57 @@
+package io.openaev.migration;
+
+import java.sql.Statement;
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+import org.springframework.stereotype.Component;
+
+@Component
+public class V4_53__Convert_expectations_to_jsonb extends BaseJavaMigration {
+
+  @Override
+  public boolean canExecuteInTransaction() {
+    return false;
+  }
+
+  @Override
+  public void migrate(Context context) throws Exception {
+    try (Statement select = context.getConnection().createStatement()) {
+      select.execute(
+          """
+              ALTER TABLE injects_expectations
+                  ALTER COLUMN inject_expectation_signatures
+                  TYPE jsonb
+                  USING inject_expectation_signatures::jsonb;
+              -- Transform Foreign Key to deferrable key
+              ALTER TABLE execution_traces
+                DROP CONSTRAINT execution_traces_execution_agent_id_fkey,
+                DROP CONSTRAINT execution_traces_execution_inject_status_id_fkey,
+                DROP CONSTRAINT execution_traces_execution_inject_test_status_id_fkey;
+
+              ALTER TABLE execution_traces
+                ADD CONSTRAINT execution_traces_execution_inject_status_id_fkey
+                FOREIGN KEY (execution_inject_status_id)
+                REFERENCES injects_statuses(status_id)
+                ON DELETE CASCADE
+                DEFERRABLE INITIALLY DEFERRED,
+
+                ADD CONSTRAINT execution_traces_execution_inject_test_status_id_fkey
+                FOREIGN KEY (execution_inject_test_status_id)
+                REFERENCES injects_tests_statuses(status_id)
+                ON DELETE CASCADE
+                DEFERRABLE INITIALLY DEFERRED,
+
+                ADD CONSTRAINT execution_traces_execution_agent_id_fkey
+                FOREIGN KEY (execution_agent_id)
+                REFERENCES agents(agent_id)
+                ON DELETE CASCADE
+                DEFERRABLE INITIALLY DEFERRED;
+              """);
+      select.execute(
+          """
+              CREATE INDEX CONCURRENTLY idx_injects_expectations_inject_agent
+                  ON injects_expectations(inject_id, agent_id);
+              """);
+    }
+  }
+}

openaev-api/src/main/java/io/openaev/rest/finding/FindingService.java

@@ -20,7 +20,6 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
-import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -91,71 +90,25 @@ public void deleteFinding(@NotNull final String id) {
    */
   public void buildFinding(
       Inject inject, Asset asset, ContractOutputElement contractOutputElement, String finalValue) {
-    try {
-      Optional<Finding> optionalFinding =
-          findingRepository.findByInjectIdAndValueAndTypeAndKey(
-              inject.getId(),
-              finalValue,
-              contractOutputElement.getType(),
-              contractOutputElement.getKey());
-
-      Finding finding =
-          optionalFinding.orElseGet(
-              () -> {
-                Finding newFinding = new Finding();
-                newFinding.setInject(inject);
-                newFinding.setField(contractOutputElement.getKey());
-                newFinding.setType(contractOutputElement.getType());
-                newFinding.setValue(finalValue);
-                newFinding.setName(contractOutputElement.getName());
-                newFinding.setTags(new HashSet<>(contractOutputElement.getTags()));
-                return newFinding;
-              });
-
-      boolean isNewAsset =
-          finding.getAssets().stream().noneMatch(a -> a.getId().equals(asset.getId()));
-
-      if (isNewAsset) {
-        finding.getAssets().add(asset);
-      }
-
-      if (optionalFinding.isEmpty() || isNewAsset) {
-        findingRepository.save(finding);
-      }
-
-    } catch (DataIntegrityViolationException ex) {
-      log.info(
-          String.format(
-              "Race condition: finding already exists. Retrying ... %s ", ex.getMessage()),
-          ex);
-      // Re-fetch and try to add the asset
-      handleRaceCondition(inject, asset, contractOutputElement, finalValue);
-    }
-  }
-
-  private void handleRaceCondition(
-      Inject inject, Asset asset, ContractOutputElement contractOutputElement, String finalValue) {
-    Optional<Finding> retryFinding =
-        findingRepository.findByInjectIdAndValueAndTypeAndKey(
-            inject.getId(),
-            finalValue,
-            contractOutputElement.getType(),
-            contractOutputElement.getKey());
-
-    if (retryFinding.isPresent()) {
-      Finding existingFinding = retryFinding.get();
-      boolean isNewAsset =
-          existingFinding.getAssets().stream().noneMatch(a -> a.getId().equals(asset.getId()));
-      if (isNewAsset) {
-        existingFinding.getAssets().add(asset);
-        findingRepository.save(existingFinding);
-      }
-    } else {
-      log.warn("Retry failed: Finding still not found after race condition.");
-    }
+    String[] tagIds =
+        contractOutputElement.getTags().isEmpty()
+            ? new String[0]
+            : contractOutputElement.getTags().stream().map(Tag::getId).toArray(String[]::new);
+
+    // Save or update the finding and add or update the list of assets and/or tags
+    findingRepository.saveCompleteFinding(
+        contractOutputElement.getKey(),
+        contractOutputElement.getType().name(),
+        finalValue,
+        new String[0],
+        inject.getId(),
+        contractOutputElement.getName(),
+        asset.getId(),
+        tagIds);
   }
 
-  // -- Extract findings from strctured output : Here we compute the findings from structured output
+  // -- Extract findings from structured output : Here we compute the findings from structured
+  // output
   // from ExecutionInjectInput sent by injectors
   // This structured output is generated based on injectorcontract where we can find the node
   // Outputs and with that the injector generate this structure output--