fix(engine): fix ES indexing cursor skipping expectations with same timestamp (#6490)

.github/instructions/performance.instructions.md

@@ -101,4 +101,5 @@ Repositories that are used with `ReferenceResolver` must expose a `countByIdIn(S
 - ❌ Iterating a list to call `repository.findById()` in a loop — use `ReferenceResolver` or `findAllById()` instead
 - ❌ Opening a transaction for read-only operations without `readOnly = true`
 - ❌ Returning JPA entities with LAZY collections from `@RestController` (triggers proxy outside session)
+- ❌ Duplicate native `@Query` methods that differ only by an optional parameter — merge into one method using SQL null-guards (e.g. `OR (:lastId IS NOT NULL AND ...)`) to avoid maintaining two copies of large query blocks
 

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

@@ -0,0 +1,23 @@
+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 V5_35__Add_indexing_last_id extends BaseJavaMigration {
+
+  @Override
+  public void migrate(Context context) throws Exception {
+    try (Statement statement = context.getConnection().createStatement()) {
+      statement.execute(
+          "ALTER TABLE indexing_status ADD COLUMN IF NOT EXISTS indexing_last_id TEXT NULL");
+      // Force a one-time full reindex for expectations so that any rows that were previously
+      // skipped (same-timestamp batch overflow before this fix) are picked up on next cycle.
+      statement.execute(
+          "UPDATE indexing_status SET indexing_last_id = NULL, indexing_status_indexing_date = '1970-01-01 00:00:00'"
+              + " WHERE indexing_status_type = 'expectation-inject'");
+    }
+  }
+}

openaev-api/src/main/java/io/openaev/migration/V5_34__Add_inject_expectation_signatures_initialized.java → openaev-api/src/main/java/io/openaev/migration/V6_20260630122959088__Add_inject_expectation_signatures_initialized.java

@@ -6,7 +6,8 @@
 import org.springframework.stereotype.Component;
 
 @Component
-public class V5_34__Add_inject_expectation_signatures_initialized extends BaseJavaMigration {
+public class V6_20260630122959088__Add_inject_expectation_signatures_initialized
+    extends BaseJavaMigration {
   @Override
   public void migrate(Context context) throws Exception {
     try (Statement statement = context.getConnection().createStatement()) {

openaev-api/src/test/java/io/openaev/engine/IndexingRegressionTest.java

@@ -19,6 +19,7 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
+import java.util.stream.Stream;
 import org.junit.jupiter.api.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.TestExecutionListeners;
@@ -282,5 +283,73 @@ void expectation_reindexed_when_linked_inject_updated() {
       // Assert — expectation must appear because its linked inject is recent
       assertThat(results).anyMatch(es -> es.getBase_id().equals(expectation.getId()));
     }
+
+    @Test
+    @DisplayName(
+        "Compound cursor: all expectations sharing the same timestamp are returned across"
+            + " batches")
+    void compound_cursor_returns_all_expectations_at_same_timestamp() {
+      // Arrange — 3 expectations in the same inject, all forced to identical updated_at
+      // (simulates bulkComputeTechnicalExpectations saving many rows in a single saveAll())
+      Instant sharedTimestamp = FROM.plus(30, ChronoUnit.MINUTES);
+
+      InjectExpectation exp1 = InjectExpectationFixture.createDefaultDetectionInjectExpectation();
+      InjectExpectation exp2 = InjectExpectationFixture.createDefaultDetectionInjectExpectation();
+      InjectExpectation exp3 = InjectExpectationFixture.createDefaultDetectionInjectExpectation();
+
+      scenarioComposer
+          .forScenario(ScenarioFixture.createDefaultIncidentResponseScenario())
+          .withInject(
+              injectComposer
+                  .forInject(InjectFixture.getDefaultInject())
+                  .withExpectation(
+                      injectExpectationComposer
+                          .forExpectation(exp1)
+                          .withEndpoint(
+                              endpointComposer.forEndpoint(EndpointFixture.createEndpoint())))
+                  .withExpectation(
+                      injectExpectationComposer
+                          .forExpectation(exp2)
+                          .withEndpoint(
+                              endpointComposer.forEndpoint(EndpointFixture.createEndpoint())))
+                  .withExpectation(
+                      injectExpectationComposer
+                          .forExpectation(exp3)
+                          .withEndpoint(
+                              endpointComposer.forEndpoint(EndpointFixture.createEndpoint()))))
+          .persist();
+      entityManager.flush();
+
+      // Force all 3 expectations to the exact same timestamp
+      for (InjectExpectation exp : List.of(exp1, exp2, exp3)) {
+        entityManager
+            .createNativeQuery(
+                "UPDATE injects_expectations SET inject_expectation_updated_at = :ts"
+                    + " WHERE inject_expectation_id = :id")
+            .setParameter("ts", sharedTimestamp)
+            .setParameter("id", exp.getId())
+            .executeUpdate();
+      }
+      entityManager.flush();
+      entityManager.clear();
+
+      // Act — first batch (limit=2): returns 2 expectations ordered by (updated_at, id)
+      List<EsInjectExpectation> batch1 = injectExpectationHandler.fetch(FROM, 2);
+      assertThat(batch1).hasSize(2);
+
+      // Act — compound cursor: from = updated_at of last item (simulates lastIndexing after full
+      // batch), lastId = id of last item. Must return the 3rd expectation only.
+      String lastId = batch1.getLast().getBase_id();
+      Instant lastUpdatedAt = batch1.getLast().getBase_updated_at();
+      List<EsInjectExpectation> batch2 = injectExpectationHandler.fetch(lastUpdatedAt, lastId, 2);
+
+      // Assert — all 3 expectations are covered across both batches (none skipped)
+      List<String> allReturnedIds =
+          Stream.concat(batch1.stream(), batch2.stream())
+              .map(EsInjectExpectation::getBase_id)
+              .toList();
+      assertThat(allReturnedIds)
+          .containsExactlyInAnyOrder(exp1.getId(), exp2.getId(), exp3.getId());
+    }
   }
 }

openaev-model/src/main/java/io/openaev/database/model/IndexingStatus.java

@@ -23,4 +23,10 @@ public class IndexingStatus {
   @Column(name = "indexing_status_indexing_date")
   @JsonProperty("indexing_status_indexing_date")
   private Instant lastIndexing;
+
+  /** Last processed entity ID within the current indexing timestamp window (compound cursor). */
+  @Getter
+  @Column(name = "indexing_last_id")
+  @JsonProperty("indexing_last_id")
+  private String lastId;
 }

openaev-model/src/main/java/io/openaev/database/repository/InjectExpectationRepository.java

@@ -337,21 +337,33 @@ void insertSignature(
 
   // -- INDEXING --
 
+  /**
+   * Fetches inject expectations updated after {@code from}, using a compound keyset cursor when
+   * {@code lastId} is non-null to avoid skipping items that share the same {@code updated_at}
+   * timestamp across batch boundaries.
+   *
+   * <p>When {@code lastId} is {@code null} the query degrades to a simple {@code > :from} cursor
+   * (first-batch behaviour). When non-null it additionally returns rows at exactly {@code from}
+   * whose ID is strictly greater than {@code lastId}.
+   */
   @Query(
       value =
           """
     WITH changed_expectations AS (
-        SELECT ie.inject_expectation_id FROM injects_expectations ie
-          WHERE ie.inject_expectation_updated_at > :from
-        UNION
-        SELECT ie.inject_expectation_id FROM injects_expectations ie
-          JOIN injects i ON i.inject_id = ie.inject_id
-          WHERE i.inject_updated_at > :from
-        UNION
-        SELECT ie.inject_expectation_id FROM injects_expectations ie
-          JOIN injects i ON i.inject_id = ie.inject_id
-          JOIN injectors_contracts ic ON ic.injector_contract_id = i.inject_injector_contract
-          WHERE ic.injector_contract_updated_at > :from
+      SELECT ie.inject_expectation_id FROM injects_expectations ie
+        WHERE ie.inject_expectation_updated_at > :from
+        OR (:lastId IS NOT NULL AND ie.inject_expectation_updated_at = :from AND ie.inject_expectation_id > :lastId)
+      UNION
+      SELECT ie.inject_expectation_id FROM injects_expectations ie
+        JOIN injects i ON i.inject_id = ie.inject_id
+        WHERE i.inject_updated_at > :from
+        OR (:lastId IS NOT NULL AND i.inject_updated_at = :from AND ie.inject_expectation_id > :lastId)
+      UNION
+      SELECT ie.inject_expectation_id FROM injects_expectations ie
+        JOIN injects i ON i.inject_id = ie.inject_id
+        JOIN injectors_contracts ic ON ic.injector_contract_id = i.inject_injector_contract
+        WHERE ic.injector_contract_updated_at > :from
+        OR (:lastId IS NOT NULL AND ic.injector_contract_updated_at = :from AND ie.inject_expectation_id > :lastId)
     ),
     inject_expectation_data AS (
       SELECT
@@ -402,16 +414,16 @@ LEFT JOIN LATERAL jsonb_array_elements(ie.inject_expectation_results::jsonb) AS
       ie.inject_expectation_id,
       ic.injector_contract_id,
       i.inject_title,
-        i.tenant_id
+      i.tenant_id
     )
     SELECT * FROM inject_expectation_data ied
     WHERE ied.agent_id IS NULL
-    ORDER BY ied.inject_expectation_updated_at ASC
+    ORDER BY ied.inject_expectation_updated_at ASC, ied.inject_expectation_id ASC
     LIMIT :limit
     """,
       nativeQuery = true)
-  List<RawInjectExpectationIndexing> findForIndexing(
-      @Param("from") Instant from, @Param("limit") int limit);
+  List<RawInjectExpectationIndexing> findForIndexingAfter(
+      @Param("from") Instant from, @Param("lastId") String lastId, @Param("limit") int limit);
 
   /**
    * Retrieves a set of distinct inject IDs associated with the specified inject expectation IDs.

openaev-model/src/main/java/io/openaev/engine/Handler.java

@@ -21,4 +21,25 @@ public interface Handler<T extends EsBase> {
    * @return list data to index
    */
   List<T> fetch(Instant from, int limit);
+
+  /**
+   * Variant of {@link #fetch(Instant, int)} using a compound keyset cursor (timestamp + last entity
+   * ID) to avoid missing items that share the same {@code updated_at} timestamp when the previous
+   * batch was full.
+   *
+   * <p>The default implementation ignores {@code lastId} and falls back to the basic cursor,
+   * providing backward-compatible behaviour for handlers that do not need compound pagination.
+   * Override in handlers where the same-millisecond duplicate issue is observable (e.g.
+   * InjectExpectationHandler).
+   *
+   * @param from lower-bound timestamp (exclusive unless lastId is provided)
+   * @param lastId ID of the last successfully indexed entity at {@code from}; when non-null the
+   *     query returns items at {@code from} with ID strictly greater than this value, plus all
+   *     items strictly after {@code from}
+   * @param limit maximum number of records to fetch per batch
+   * @return list data to index
+   */
+  default List<T> fetch(Instant from, String lastId, int limit) {
+    return fetch(from, limit);
+  }
 }