fix: adds order by timestamp

core/policy-monitor/policy-monitor-core/src/main/java/org/eclipse/edc/connector/policy/monitor/manager/PolicyMonitorManagerImpl.java

@@ -52,12 +52,6 @@ private PolicyMonitorManagerImpl() {
 
     }
 
-    @Override
-    protected StateMachineManager.Builder configureStateMachineManager(StateMachineManager.Builder builder) {
-        return builder
-                .processor(processEntriesInState(STARTED, this::processMonitoring));
-    }
-
     @Override
     public void startMonitoring(String transferProcessId, String contractId) {
         var entry = PolicyMonitorEntry.Builder.newInstance()
@@ -71,6 +65,12 @@ public void startMonitoring(String transferProcessId, String contractId) {
         update(entry);
     }
 
+    @Override
+    protected StateMachineManager.Builder configureStateMachineManager(StateMachineManager.Builder builder) {
+        return builder
+                .processor(processEntriesInState(STARTED, this::processMonitoring));
+    }
+
     private boolean processMonitoring(PolicyMonitorEntry entry) {
         var transferProcess = transferProcessService.findById(entry.getId());
         if (transferProcess == null) {
@@ -110,8 +110,10 @@ private boolean processMonitoring(PolicyMonitorEntry entry) {
             }
         }
 
-        breakLease(entry);
-        return true;
+        // we update the state timestamp ensure fairness on polling on  `STARTED` state
+        // the lease will be broken in `onNotProcessed`
+        entry.updateStateTimestamp();
+        return false;
     }
 
     private Processor processEntriesInState(PolicyMonitorEntryStates state, Function<PolicyMonitorEntry, Boolean> function) {
@@ -125,14 +127,14 @@ private Processor processEntriesInState(PolicyMonitorEntryStates state, Function
     public static class Builder
             extends AbstractStateEntityManager.Builder<PolicyMonitorEntry, PolicyMonitorStore, PolicyMonitorManagerImpl, Builder> {
 
-        public static Builder newInstance() {
-            return new Builder();
-        }
-
         private Builder() {
             super(new PolicyMonitorManagerImpl());
         }
 
+        public static Builder newInstance() {
+            return new Builder();
+        }
+
         public Builder contractAgreementService(ContractAgreementService contractAgreementService) {
             manager.contractAgreementService = contractAgreementService;
             return this;

core/policy-monitor/policy-monitor-core/src/test/java/org/eclipse/edc/connector/policy/monitor/manager/PolicyMonitorManagerImplTest.java

@@ -149,6 +149,8 @@ void started_shouldDoNothing_whenPolicyIsValid() {
                 .state(STARTED.code())
                 .build();
         var policy = Policy.Builder.newInstance().build();
+
+        var stateTimestamp = entry.getStateTimestamp();
         when(store.nextNotLeased(anyInt(), stateIs(STARTED.code()))).thenReturn(List.of(entry)).thenReturn(emptyList());
         when(transferProcessService.findById(entry.getId()))
                 .thenReturn(TransferProcess.Builder.newInstance().state(TransferProcessStates.STARTED.code()).build());
@@ -159,7 +161,7 @@ void started_shouldDoNothing_whenPolicyIsValid() {
 
         await().untilAsserted(() -> {
             verify(transferProcessService, never()).terminate(any());
-            verify(store).save(argThat(it -> it.getState() == STARTED.code()));
+            verify(store).save(argThat(it -> it.getState() == STARTED.code() && stateTimestamp < it.getStateTimestamp()));
         });
     }
 

extensions/control-plane/store/sql/contract-negotiation-store-sql/docs/schema.sql

@@ -80,3 +80,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS contract_negotiation_id_uindex
 
 CREATE UNIQUE INDEX IF NOT EXISTS contract_agreement_id_uindex
     ON edc_contract_agreement (agr_id);
+
+
+-- This will help to identify states that needs to be transition without table scan when the entries start to grow
+CREATE INDEX IF NOT EXISTS contract_negotiation_state ON edc_contract_negotiation (state,state_timestamp);

extensions/control-plane/store/sql/contract-negotiation-store-sql/src/main/java/org/eclipse/edc/connector/controlplane/store/sql/contractnegotiation/store/SqlContractNegotiationStore.java

@@ -79,21 +79,39 @@ public SqlContractNegotiationStore(DataSourceRegistry dataSourceRegistry, String
     }
 
     @Override
-    public @Nullable ContractNegotiation findForCorrelationId(String correlationId) {
+    public @NotNull List<ContractNegotiation> nextNotLeased(int max, Criterion... criteria) {
         return transactionContext.execute(() -> {
-            // utilize the generic query api
-            var query = correlationIdQuerySpec(correlationId);
-            try (var stream = queryNegotiations(query)) {
-                return single(stream.collect(toList()));
+            var filter = Arrays.stream(criteria).toList();
+            var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
+            var statement = statements.createNegotiationsQuery(querySpec)
+                    .addWhereClause(statements.getNotLeasedFilter(), clock.millis());
+
+            try (
+                    var connection = getConnection();
+                    var stream = queryExecutor.query(getConnection(), true, contractNegotiationWithAgreementMapper(connection), statement.getQueryAsString(), statement.getParameters())
+            ) {
+                var negotiations = stream.collect(toList());
+                negotiations.forEach(cn -> leaseContext.withConnection(connection).acquireLease(cn.getId()));
+                return negotiations;
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
             }
         });
     }
 
     @Override
-    public @Nullable ContractAgreement findContractAgreement(String contractId) {
+    public StoreResult<ContractNegotiation> findByIdAndLease(String id) {
         return transactionContext.execute(() -> {
             try (var connection = getConnection()) {
-                return findContractAgreementInternal(connection, contractId);
+                var entity = findInternal(connection, id);
+                if (entity == null) {
+                    return StoreResult.notFound(format("ContractNegotiation %s not found", id));
+                }
+
+                leaseContext.withConnection(connection).acquireLease(id);
+                return StoreResult.success(entity);
+            } catch (IllegalStateException e) {
+                return StoreResult.alreadyLeased(format("ContractNegotiation %s is already leased", id));
             } catch (SQLException e) {
                 throw new EdcPersistenceException(e);
             }
@@ -119,6 +137,28 @@ public void save(ContractNegotiation negotiation) {
 
     }
 
+    @Override
+    public @Nullable ContractNegotiation findForCorrelationId(String correlationId) {
+        return transactionContext.execute(() -> {
+            // utilize the generic query api
+            var query = correlationIdQuerySpec(correlationId);
+            try (var stream = queryNegotiations(query)) {
+                return single(stream.collect(toList()));
+            }
+        });
+    }
+
+    @Override
+    public @Nullable ContractAgreement findContractAgreement(String contractId) {
+        return transactionContext.execute(() -> {
+            try (var connection = getConnection()) {
+                return findContractAgreementInternal(connection, contractId);
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
+            }
+        });
+    }
+
     @Override
     public void delete(String negotiationId) {
         transactionContext.execute(() -> {
@@ -171,46 +211,6 @@ public void delete(String negotiationId) {
         });
     }
 
-    @Override
-    public @NotNull List<ContractNegotiation> nextNotLeased(int max, Criterion... criteria) {
-        return transactionContext.execute(() -> {
-            var filter = Arrays.stream(criteria).toList();
-            var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
-            var statement = statements.createNegotiationsQuery(querySpec)
-                    .addWhereClause(statements.getNotLeasedFilter(), clock.millis());
-
-            try (
-                    var connection = getConnection();
-                    var stream = queryExecutor.query(getConnection(), true, contractNegotiationWithAgreementMapper(connection), statement.getQueryAsString(), statement.getParameters())
-            ) {
-                var negotiations = stream.collect(toList());
-                negotiations.forEach(cn -> leaseContext.withConnection(connection).acquireLease(cn.getId()));
-                return negotiations;
-            } catch (SQLException e) {
-                throw new EdcPersistenceException(e);
-            }
-        });
-    }
-
-    @Override
-    public StoreResult<ContractNegotiation> findByIdAndLease(String id) {
-        return transactionContext.execute(() -> {
-            try (var connection = getConnection()) {
-                var entity = findInternal(connection, id);
-                if (entity == null) {
-                    return StoreResult.notFound(format("ContractNegotiation %s not found", id));
-                }
-
-                leaseContext.withConnection(connection).acquireLease(id);
-                return StoreResult.success(entity);
-            } catch (IllegalStateException e) {
-                return StoreResult.alreadyLeased(format("ContractNegotiation %s is already leased", id));
-            } catch (SQLException e) {
-                throw new EdcPersistenceException(e);
-            }
-        });
-    }
-
     @Override
     public StoreResult<ContractNegotiation> findByCorrelationIdAndLease(String correlationId) {
         return transactionContext.execute(() -> {

extensions/control-plane/store/sql/transfer-process-store-sql/docs/schema.sql

@@ -65,3 +65,6 @@ CREATE UNIQUE INDEX IF NOT EXISTS transfer_process_id_uindex
 
 CREATE UNIQUE INDEX IF NOT EXISTS lease_lease_id_uindex
     ON edc_lease (lease_id);
+
+-- This will help to identify states that needs to be transition without table scan when the entries start to grow
+CREATE INDEX IF NOT EXISTS transfer_process_state ON edc_transfer_process (state,state_time_stamp);

extensions/control-plane/store/sql/transfer-process-store-sql/src/main/java/org/eclipse/edc/connector/controlplane/store/sql/transferprocess/store/SqlTransferProcessStore.java

@@ -70,11 +70,22 @@ public SqlTransferProcessStore(DataSourceRegistry dataSourceRegistry, String dat
         leaseContext = SqlLeaseContextBuilder.with(transactionContext, leaseHolderName, statements, clock, queryExecutor);
     }
 
+    @Override
+    public @Nullable TransferProcess findById(String id) {
+        return transactionContext.execute(() -> {
+            try (var connection = getConnection()) {
+                return findByIdInternal(connection, id);
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
+            }
+        });
+    }
+
     @Override
     public @NotNull List<TransferProcess> nextNotLeased(int max, Criterion... criteria) {
         return transactionContext.execute(() -> {
             var filter = Arrays.stream(criteria).collect(toList());
-            var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
+            var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
             var statement = statements.createQuery(querySpec)
                     .addWhereClause(statements.getNotLeasedFilter(), clock.millis());
 
@@ -110,30 +121,6 @@ public StoreResult<TransferProcess> findByIdAndLease(String id) {
         });
     }
 
-    @Override
-    public StoreResult<TransferProcess> findByCorrelationIdAndLease(String correlationId) {
-        return transactionContext.execute(() -> {
-            var query = correlationIdQuerySpec(correlationId);
-
-            try (
-                    var connection = getConnection();
-                    var stream = executeQuery(connection, query)
-            ) {
-                var entity = stream.findFirst().orElse(null);
-                if (entity == null) {
-                    return StoreResult.notFound(format("TransferProcess with correlationId %s not found", correlationId));
-                }
-
-                leaseContext.withConnection(connection).acquireLease(entity.getId());
-                return StoreResult.success(entity);
-            } catch (IllegalStateException e) {
-                return StoreResult.alreadyLeased(format("TransferProcess with correlationId %s is already leased", correlationId));
-            } catch (SQLException e) {
-                throw new EdcPersistenceException(e);
-            }
-        });
-    }
-
     @Override
     public void save(TransferProcess entity) {
         Objects.requireNonNull(entity.getId(), "TransferProcesses must have an ID!");
@@ -152,17 +139,6 @@ public void save(TransferProcess entity) {
         });
     }
 
-    @Override
-    public @Nullable TransferProcess findById(String id) {
-        return transactionContext.execute(() -> {
-            try (var connection = getConnection()) {
-                return findByIdInternal(connection, id);
-            } catch (SQLException e) {
-                throw new EdcPersistenceException(e);
-            }
-        });
-    }
-
     @Override
     public @Nullable TransferProcess findForCorrelationId(String correlationId) {
         return transactionContext.execute(() -> {
@@ -207,6 +183,30 @@ public Stream<TransferProcess> findAll(QuerySpec querySpec) {
         });
     }
 
+    @Override
+    public StoreResult<TransferProcess> findByCorrelationIdAndLease(String correlationId) {
+        return transactionContext.execute(() -> {
+            var query = correlationIdQuerySpec(correlationId);
+
+            try (
+                    var connection = getConnection();
+                    var stream = executeQuery(connection, query)
+            ) {
+                var entity = stream.findFirst().orElse(null);
+                if (entity == null) {
+                    return StoreResult.notFound(format("TransferProcess with correlationId %s not found", correlationId));
+                }
+
+                leaseContext.withConnection(connection).acquireLease(entity.getId());
+                return StoreResult.success(entity);
+            } catch (IllegalStateException e) {
+                return StoreResult.alreadyLeased(format("TransferProcess with correlationId %s is already leased", correlationId));
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
+            }
+        });
+    }
+
     private QuerySpec correlationIdQuerySpec(String correlationId) {
         var criterion = criterion("correlationId", "=", correlationId);
         return QuerySpec.Builder.newInstance().filter(criterion).build();

extensions/data-plane/store/sql/data-plane-store-sql/docs/schema.sql

@@ -38,3 +38,6 @@ COMMENT ON COLUMN edc_data_plane.trace_context IS 'Java Map serialized as JSON';
 COMMENT ON COLUMN edc_data_plane.source IS 'DataAddress serialized as JSON';
 COMMENT ON COLUMN edc_data_plane.destination IS 'DataAddress serialized as JSON';
 COMMENT ON COLUMN edc_data_plane.properties IS 'Java Map serialized as JSON';
+
+-- This will help to identify states that needs to be transition without table scan when the entries start to grow
+CREATE INDEX IF NOT EXISTS data_plane_state ON edc_data_plane (state,state_time_stamp);

extensions/data-plane/store/sql/data-plane-store-sql/src/main/java/org/eclipse/edc/connector/dataplane/store/sql/SqlDataPlaneStore.java

@@ -81,7 +81,7 @@ public SqlDataPlaneStore(DataSourceRegistry dataSourceRegistry, String dataSourc
     public @NotNull List<DataFlow> nextNotLeased(int max, Criterion... criteria) {
         return transactionContext.execute(() -> {
             var filter = Arrays.stream(criteria).collect(toList());
-            var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
+            var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
             var statement = statements.createQuery(querySpec)
                     .addWhereClause(statements.getNotLeasedFilter(), clock.millis());
 

extensions/policy-monitor/store/sql/policy-monitor-store-sql/docs/schema.sql

@@ -30,3 +30,7 @@ CREATE TABLE IF NOT EXISTS edc_policy_monitor
     properties           JSON,
     contract_id          VARCHAR
 );
+
+
+-- This will help to identify states that needs to be transition without table scan when the entries start to grow
+CREATE INDEX IF NOT EXISTS policy_monitor_state ON edc_policy_monitor (state,state_time_stamp);

extensions/policy-monitor/store/sql/policy-monitor-store-sql/src/main/java/org/eclipse/edc/connector/policy/monitor/store/sql/SqlPolicyMonitorStore.java

@@ -74,7 +74,7 @@ public SqlPolicyMonitorStore(DataSourceRegistry dataSourceRegistry, String dataS
     public @NotNull List<PolicyMonitorEntry> nextNotLeased(int max, Criterion... criteria) {
         return transactionContext.execute(() -> {
             var filter = Arrays.stream(criteria).collect(toList());
-            var querySpec = QuerySpec.Builder.newInstance().filter(filter).limit(max).build();
+            var querySpec = QuerySpec.Builder.newInstance().filter(filter).sortField("stateTimestamp").limit(max).build();
             var statement = statements.createQuery(querySpec)
                     .addWhereClause(statements.getNotLeasedFilter(), clock.millis());