feat: automatic deprovisioning for provider transfers

eclipse-edc · Oct 4, 2023 · 93a2f88 · 93a2f88
1 parent ffcdfb6
commit 93a2f88
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 8 deletions.
diff --git a/.../src/main/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImpl.java b/.../src/main/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImpl.java
@@ -358,7 +358,12 @@ private boolean processCompleting(TransferProcess process) {
         var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());
         return entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
                 .entityRetrieve(id -> store.findById(id))
-                .onSuccess((t, content) -> transitionToCompleted(t))
+                .onSuccess((t, content) -> {
+                    transitionToCompleted(t);
+                    if (t.getType() == PROVIDER) {
+                        transitionToDeprovisioning(t);
+                    }
+                })
                 .onFailure((t, throwable) -> transitionToCompleting(t))
                 .onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
                 .onRetryExhausted((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
@@ -389,7 +394,12 @@ private boolean processTerminating(TransferProcess process) {
         var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());
         return entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
                 .entityRetrieve(id -> store.findById(id))
-                .onSuccess((t, content) -> transitionToTerminated(t))
+                .onSuccess((t, content) -> {
+                    transitionToTerminated(t);
+                    if (t.getType() == PROVIDER) {
+                        transitionToDeprovisioning(t);
+                    }
+                })
                 .onFailure((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
                 .onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
                 .onRetryExhausted(this::transitionToTerminated)

diff --git a/.../test/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImplTest.java b/.../test/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImplTest.java
@@ -558,8 +558,8 @@ void starting_whenShouldWait_updatesStateCount() {
     }
 
     @Test
-    void completing_shouldTransitionToCompleted_whenSendingMessageSucceed() {
-        var process = createTransferProcessBuilder(COMPLETING).dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
+    void completing_provider_shouldTransitionToDeprovisioning_whenSendingMessageSucceed() {
+        var process = createTransferProcessBuilder(COMPLETING).type(PROVIDER).dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
         when(transferProcessStore.nextNotLeased(anyInt(), stateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
         when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(COMPLETING.code()).build());
         when(dispatcherRegistry.dispatch(any(), isA(TransferCompletionMessage.class))).thenReturn(completedFuture(StatusResult.success("any")));
@@ -571,13 +571,32 @@ void completing_shouldTransitionToCompleted_whenSendingMessageSucceed() {
             verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
             var message = captor.getValue();
             assertThat(message.getProcessId()).isEqualTo("correlationId");
-            verify(transferProcessStore, times(RETRY_LIMIT)).save(argThat(p -> p.getState() == COMPLETED.code()));
+            verify(transferProcessStore, atLeastOnce()).save(argThat(p -> p.getState() == DEPROVISIONING.code()));
             verify(listener).completed(process);
         });
     }
 
     @Test
-    void terminating_shouldTransitionToTerminated_whenMessageSentCorrectly() {
+    void completing_consumer_shouldTransitionToCompleted_whenSendingMessageSucceed() {
+        var process = createTransferProcessBuilder(COMPLETING).type(CONSUMER).dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
+        when(transferProcessStore.nextNotLeased(anyInt(), stateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
+        when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(COMPLETING.code()).build());
+        when(dispatcherRegistry.dispatch(any(), isA(TransferCompletionMessage.class))).thenReturn(completedFuture(StatusResult.success("any")));
+
+        manager.start();
+
+        await().untilAsserted(() -> {
+            var captor = ArgumentCaptor.forClass(TransferCompletionMessage.class);
+            verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
+            var message = captor.getValue();
+            assertThat(message.getProcessId()).isEqualTo("correlationId");
+            verify(transferProcessStore).save(argThat(p -> p.getState() == COMPLETED.code()));
+            verify(listener).completed(process);
+        });
+    }
+
+    @Test
+    void terminating_provider_shouldTransitionToDeprovisioning_whenMessageSentCorrectly() {
         var process = createTransferProcessBuilder(TERMINATING).type(PROVIDER)
                 .dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
         when(transferProcessStore.nextNotLeased(anyInt(), stateIs(TERMINATING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
@@ -591,7 +610,27 @@ void terminating_shouldTransitionToTerminated_whenMessageSentCorrectly() {
             verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
             var message = captor.getValue();
             assertThat(message.getProcessId()).isEqualTo("correlationId");
-            verify(transferProcessStore, times(RETRY_LIMIT)).save(argThat(p -> p.getState() == TERMINATED.code()));
+            verify(transferProcessStore, atLeastOnce()).save(argThat(p -> p.getState() == DEPROVISIONING.code()));
+            verify(listener).terminated(process);
+        });
+    }
+
+    @Test
+    void terminating_consumer_shouldTransitionToTerminated_whenMessageSentCorrectly() {
+        var process = createTransferProcessBuilder(TERMINATING).type(CONSUMER)
+                .dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
+        when(transferProcessStore.nextNotLeased(anyInt(), stateIs(TERMINATING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
+        when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(TERMINATING.code()).build());
+        when(dispatcherRegistry.dispatch(any(), isA(TransferTerminationMessage.class))).thenReturn(completedFuture(StatusResult.success("any")));
+
+        manager.start();
+
+        await().untilAsserted(() -> {
+            var captor = ArgumentCaptor.forClass(TransferTerminationMessage.class);
+            verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
+            var message = captor.getValue();
+            assertThat(message.getProcessId()).isEqualTo("correlationId");
+            verify(transferProcessStore).save(argThat(p -> p.getState() == TERMINATED.code()));
             verify(listener).terminated(process);
         });
     }

diff --git a/...nsfer-spi/src/main/java/org/eclipse/edc/connector/transfer/spi/types/TransferProcess.java b/...nsfer-spi/src/main/java/org/eclipse/edc/connector/transfer/spi/types/TransferProcess.java
@@ -287,7 +287,7 @@ public void transitionCompleted() {
     }
 
     public boolean canBeDeprovisioned() {
-        return currentStateIsOneOf(COMPLETED, TERMINATED, DEPROVISIONING);
+        return currentStateIsOneOf(COMPLETING, TERMINATING, COMPLETED, TERMINATED, DEPROVISIONING);
     }
 
     public void transitionDeprovisioning() {