feat: optimize state machines (#1396)

* feat: optimize state machine with index on state and state_timestamp

* chore: deps file

edc-extensions/migrations/control-plane-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/contractnegotiation/V0_0_9__Alter_ContractNegotiation_CreateStateIndex.sql

@@ -0,0 +1,15 @@
+--
+--  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+-- This will help to identify states that need to be transitioned without a table scan when the entries grow
+CREATE INDEX IF NOT EXISTS contract_negotiation_state ON edc_contract_negotiation (state,state_timestamp);

edc-extensions/migrations/control-plane-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/policy-monitor/V0_0_2__Alter_PolicyMonitor_CreateStateIndex.sql

@@ -0,0 +1,15 @@
+--
+--  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+-- This will help to identify states that need to be transitioned without a table scan when the entries grow
+CREATE INDEX IF NOT EXISTS policy_monitor_state ON edc_policy_monitor (state,state_time_stamp);

edc-extensions/migrations/control-plane-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/transferprocess/V0_0_16__Alter_TransferProcess_CreateStateIndex.sql

@@ -0,0 +1,16 @@
+--
+--  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+-- This will help to identify states that need to be transitioned without a table scan when the entries grow
+CREATE INDEX IF NOT EXISTS transfer_process_state ON edc_transfer_process (state,state_time_stamp);
+

edc-extensions/migrations/data-plane-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/dataplane/V0_0_2__Alter_Dataplane_AddTransferTypeDestinationColumn.sql

@@ -0,0 +1,15 @@
+--
+--  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+
+ALTER TABLE edc_data_plane ADD COLUMN transfer_type_destination VARCHAR DEFAULT 'HttpData';

edc-extensions/migrations/data-plane-migration/src/main/resources/org/eclipse/tractusx/edc/postgresql/migration/dataplane/V0_0_3__Alter_Dataplane_CreateStateIndex.sql

@@ -0,0 +1,16 @@
+--
+--  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+--
+--  This program and the accompanying materials are made available under the
+--  terms of the Apache License, Version 2.0 which is available at
+--  https://www.apache.org/licenses/LICENSE-2.0
+--
+--  SPDX-License-Identifier: Apache-2.0
+--
+--  Contributors:
+--       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+--
+
+
+-- This will help to identify states that need to be transitioned without a table scan when the entries grow
+CREATE INDEX IF NOT EXISTS data_plane_state ON edc_data_plane (state,state_time_stamp);

edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/AzureToAzureTest.java

@@ -30,6 +30,7 @@
 import org.eclipse.edc.spi.security.Vault;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -268,6 +269,7 @@ private DataFlowStartMessage createFlowRequest(String blobName) {
                 .sourceDataAddress(blobSourceAddress(blobName))
                 .destinationDataAddress(blobDestinationAddress(blobName))
                 .processId("test-process-id")
+                .flowType(FlowType.PUSH)
                 .build();
     }
 
@@ -283,6 +285,8 @@ private DataFlowStartMessage createMultipleFileFlowRequest(String blobPrefix) {
                         .type("AzureStorage").property("container", AZBLOB_CONSUMER_CONTAINER_NAME)
                         .property("account", AZBLOB_CONSUMER_ACCOUNT_NAME).property("keyName", AZBLOB_CONSUMER_KEY_ALIAS)
                         .build())
-                .processId("test-process-multiple-file-id").build();
+                .processId("test-process-multiple-file-id")
+                .flowType(FlowType.PUSH)
+                .build();
     }
 }

edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/MultiCloudTest.java

@@ -32,6 +32,7 @@
 import org.eclipse.edc.spi.security.Vault;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -142,6 +143,7 @@ void transferFile_azureToS3MultipleFiles(Vault vault) {
                         .property(S3BucketSchema.ENDPOINT_OVERRIDE, s3EndpointOverride)
                         .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-request-multiple")
                 .build();
 
@@ -193,6 +195,7 @@ void transferFile_azureToS3(Vault vault) {
                         .property(S3BucketSchema.ENDPOINT_OVERRIDE, s3EndpointOverride)
                         .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-process-id")
                 .build();
 
@@ -257,6 +260,7 @@ void transferFile_s3ToAzureMultipleFiles(Vault vault) {
                                 .property("keyName", AZBLOB_CONSUMER_KEY_ALIAS)
                                 .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-process-multiple-file-id")
                 .build();
 
@@ -322,6 +326,7 @@ void transferFile_s3ToAzureMultipleFiles_whenConsumerDefinesBloblName_success(Va
                                 .property("blobName", "NOME_TEST")
                                 .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-process-multiple-file-id")
                 .build();
 
@@ -369,6 +374,7 @@ void transferFile_s3ToAzure(Vault vault) {
                 )
                 .destinationDataAddress(blobDestinationAddress(TESTFILE_NAME))
                 .processId("test-process-id")
+                .flowType(FlowType.PUSH)
                 .build();
 
         var url = "http://localhost:%s/control/transfer".formatted(PROVIDER_CONTROL_PORT);

edc-tests/edc-dataplane/cloud-transfer-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/transfer/test/S3ToS3Test.java

@@ -32,6 +32,7 @@
 import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -302,6 +303,7 @@ private DataFlowStartMessage createFlowRequest() {
                         .property(S3BucketSchema.ENDPOINT_OVERRIDE, consumerEndpointOverride)
                         .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-process-id")
                 .build();
     }
@@ -329,6 +331,7 @@ private DataFlowStartMessage createMultipleFileFlowRequest() {
                         .property(S3BucketSchema.ENDPOINT_OVERRIDE, consumerEndpointOverride)
                         .build()
                 )
+                .flowType(FlowType.PUSH)
                 .processId("test-process-multiple-file-id")
                 .build();
     }

edc-tests/edc-dataplane/edc-dataplane-tokenrefresh-tests/src/test/java/org/eclipse/tractusx/edc/dataplane/tokenrefresh/e2e/DataPlaneTokenRefreshEndToEndTest.java

@@ -397,6 +397,7 @@ private DataFlowStartMessage createStartMessage(String processId, String audienc
                 .sourceDataAddress(DataAddress.Builder.newInstance().type("HttpData").property(EDC_NAMESPACE + "baseUrl", "http://foo.bar/").build())
                 .destinationDataAddress(DataAddress.Builder.newInstance().type("HttpData").property(EDC_NAMESPACE + "baseUrl", "http://fizz.buzz").build())
                 .flowType(FlowType.PULL)
+                .transferTypeDestination("HttpData")
                 .participantId("some-participantId")
                 .assetId("test-asset")
                 .callbackAddress(URI.create("https://foo.bar/callback"))

gradle/libs.versions.toml

@@ -2,7 +2,7 @@
 format.version = "1.1"
 
 [versions]
-edc = "0.7.1"
+edc = "0.7.2-20240704-SNAPSHOT"
 assertj = "3.26.0"
 awaitility = "4.2.1"
 aws = "2.26.14"