feat: workflow delete effect (#2305)

* feat: workflows deletion

* fixing readme files

* Update docs/src/modules/java-protobuf/pages/workflows.adoc

Co-authored-by: Eduardo Pinto <[email protected]>

* runtime bump

---------

Co-authored-by: Eduardo Pinto <[email protected]>

Author: aludwiko

docs/src/modules/java-protobuf/pages/workflows.adoc

@@ -179,6 +179,36 @@ IMPORTANT: For simplicity purposes, we are returning the internal state directly
 
 A full transfer workflow source code is available https://github.com/lightbend/kalix-jvm-sdk/tree/main/samples/java-protobuf-transfer-workflow[here (Java Protobuf), {tab-icon}, window="new"] or https://github.com/lightbend/kalix-jvm-sdk/tree/main/samples/scala-protobuf-transfer-workflow[here (Scala Protobuf), {tab-icon}, window="new"]. Follow the `README` file to run and test it.
 
+
+== Deleting state
+
+If you want to delete the workflow, you can use the `delete` method from the effect API.
+
+[.tabset]
+Java::
++
+[source,java,indent=0]
+.src/main/java/com/example/transfer/api/TransferWorkflow.java
+----
+include::example$java-protobuf-transfer-workflow-compensation/src/main/java/com/example/transfer/api/TransferWorkflow.java[tag=delete-workflow]
+----
+<1> Deletes the workflow state.
+
+Scala::
++
+[source,scala,indent=0]
+.src/main/scala/com/example/transfer/api/TransferWorkflow.scala
+----
+include::example$scala-protobuf-transfer-workflow-compensation/src/main/scala/com/example/transfer/api/TransferWorkflow.scala[tag=delete-workflow]
+----
+<1> Deletes the workflow state.
+
+When you give the instruction to delete a running workflow it's equivalent to ending and deleting a workflow. For already finished workflows, it is possible to delete them in the command handler, but any state changes will be ignored. By default, the existence of the workflow is completely cleaned up after a week.
+
+You can still handle read requests to the workflow until it has been completely removed, but the current state will be empty (or null). To check whether the workflow has been deleted, you can use the inherited method `isDeleted`.
+
+It is best to not reuse the same workflow id after deletion, but if that happens after the workflow has been completely removed it will be instantiated as a completely new workflow without any knowledge of previous state.
+
 == Pausing workflow
 
 A long-running workflow can be paused while waiting for some additional information to continue processing. A special `pause` transition can be used to inform Kalix that the execution of the Workflow should be postponed. By launching a request to a Workflow endpoint, the user can then resume the processing. Additionally, a Kalix Timer can be scheduled to automatically inform the Workflow that the expected time for the additional data has passed.

samples/java-protobuf-transfer-workflow-compensation/README.md

@@ -41,25 +41,25 @@ With both the Kalix Runtime and your service running, any defined endpoints shou
 Create wallet `a` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Create wallet `b` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Get wallet `a` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Get wallet `b` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Start transfer from wallet `a` to wallet `b`

samples/java-protobuf-transfer-workflow-compensation/src/it/java/com/example/transfer/api/TransferWorkflowIntegrationTest.java

@@ -51,12 +51,13 @@ public TransferWorkflowIntegrationTest() {
   public void showTransferFunds() throws Exception {
     String walletAId = randomId();
     String walletBId = randomId();
+    String transferId = "1";
     InitialBalance initWalletA = InitialBalance.newBuilder().setWalletId(walletAId).setBalance(100).build();
     InitialBalance initWalletB = InitialBalance.newBuilder().setWalletId(walletBId).setBalance(100).build();
     walletClient.create(initWalletA).toCompletableFuture().get(5, SECONDS);
     walletClient.create(initWalletB).toCompletableFuture().get(5, SECONDS);
 
-    Transfer transfer = Transfer.newBuilder().setTransferId("1").setFrom(walletAId).setTo(walletBId).setAmount(10).build();
+    Transfer transfer = Transfer.newBuilder().setTransferId(transferId).setFrom(walletAId).setTo(walletBId).setAmount(10).build();
     workflowClient.start(transfer).toCompletableFuture().get(5, SECONDS);
 
     await().atMost(10, SECONDS).untilAsserted(() -> {
@@ -68,6 +69,14 @@ public void showTransferFunds() throws Exception {
       assertEquals(90, walletA.getBalance());
       assertEquals(110, walletB.getBalance());
     });
+    
+    var response = workflowClient.hasBeenDeleted(TransferApi.HasBeenDeletedRequest.newBuilder().setTransferId(transferId).build()).toCompletableFuture().get(5, SECONDS);
+    assertEquals(false, response.getDeleted());
+    
+    workflowClient.delete(TransferApi.DeleteRequest.newBuilder().setTransferId(transferId).build()).toCompletableFuture().get(5, SECONDS);
+
+    var response2 = workflowClient.hasBeenDeleted(TransferApi.HasBeenDeletedRequest.newBuilder().setTransferId(transferId).build()).toCompletableFuture().get(5, SECONDS);
+    assertEquals(true, response2.getDeleted());
   }
 
   @Test

samples/java-protobuf-transfer-workflow-compensation/src/main/java/com/example/transfer/api/TransferWorkflow.java

@@ -260,4 +260,18 @@ public Effect<TransferState> getTransferState(TransferState currentState, Transf
       return effects().reply(currentState);
     }
   }
+
+  // tag::delete-workflow[]
+  @Override
+  public Effect<Empty> delete(TransferState currentState, TransferApi.DeleteRequest deleteRequest) {
+    return effects()
+      .delete() // <1>
+      .thenReply(Empty.getDefaultInstance());
+  }
+  // end::delete-workflow[]
+
+  @Override
+  public Effect<TransferApi.HasBeenDeletedResponse> hasBeenDeleted(TransferState currentState, TransferApi.HasBeenDeletedRequest hasBeenDeletedRequest) {
+    return effects().reply(TransferApi.HasBeenDeletedResponse.newBuilder().setDeleted(isDeleted()).build());
+  }
 }

samples/java-protobuf-transfer-workflow-compensation/src/main/proto/com/example/transfer/transfer_api.proto

@@ -28,6 +28,18 @@ message AcceptationTimeoutRequest {
     string transfer_id = 1 [(kalix.field).id = true];
 }
 
+message DeleteRequest{
+    string transfer_id = 1 [(kalix.field).id = true];
+}
+
+message HasBeenDeletedRequest{
+    string transfer_id = 1 [(kalix.field).id = true];
+}
+
+message HasBeenDeletedResponse{
+    bool deleted = 1;
+}
+
 service TransferWorkflowService {
     option (kalix.codegen) = {
         workflow: {
@@ -40,4 +52,6 @@ service TransferWorkflowService {
     rpc Accept(AcceptRequest) returns (google.protobuf.Empty) {}
     rpc AcceptationTimeout(AcceptationTimeoutRequest) returns (google.protobuf.Empty) {}
     rpc GetTransferState(GetRequest) returns (com.example.transfer.domain.TransferState) {}
+    rpc Delete(DeleteRequest) returns (google.protobuf.Empty) {}
+    rpc HasBeenDeleted(HasBeenDeletedRequest) returns (HasBeenDeletedResponse) {}
 }

samples/java-protobuf-transfer-workflow/README.md

@@ -41,25 +41,25 @@ With both the Kalix Runtime and your service running, any defined endpoints shou
 Create wallet `a` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Create wallet `b` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Get wallet `a` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Get wallet `b` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Start transfer from wallet `a` to wallet `b`

samples/scala-protobuf-transfer-workflow-compensation/README.md

@@ -29,25 +29,25 @@ With both the Kalix Runtime and your service running, any defined endpoints shou
 Create wallet `a` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "a", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Create wallet `b` with an initial balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.TransferWorkflowService/Create
+grpcurl -plaintext -d '{"wallet_id": "b", "balance": 100}' localhost:9000 com.example.wallet.api.WalletService/Create
 ```
 
 Get wallet `a` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "a"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Get wallet `b` current balance
 
 ```shell
-grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.TransferWorkflowService/GetWalletState
+grpcurl -plaintext -d '{"wallet_id": "b"}' localhost:9000 com.example.wallet.api.WalletService/GetWalletState
 ```
 
 Start transfer from wallet `a` to wallet `b`
@@ -62,6 +62,10 @@ Get transfer state
 grpcurl -plaintext -d '{"transfer_id": "1"}' localhost:9000 com.example.transfer.api.TransferWorkflowService/GetTransferState
 ```
 
+```shell
+grpcurl -plaintext -d '{"transfer_id": "1"}' localhost:9000 com.example.transfer.api.TransferWorkflowService/GetTransferState
+```
+
 ## Deploying
 
 To deploy your service, install the `kalix` CLI as documented in

samples/scala-protobuf-transfer-workflow-compensation/src/main/proto/com/example/transfer/transfer_api.proto

@@ -26,6 +26,18 @@ message AcceptationTimeoutRequest {
     string transfer_id = 1 [(kalix.field).id = true];
 }
 
+message DeleteRequest{
+    string transfer_id = 1 [(kalix.field).id = true];
+}
+
+message HasBeenDeletedRequest{
+    string transfer_id = 1 [(kalix.field).id = true];
+}
+
+message HasBeenDeletedResponse{
+    bool deleted = 1;
+}
+
 service TransferWorkflowService {
     option (kalix.codegen) = {
         workflow: {
@@ -38,4 +50,6 @@ service TransferWorkflowService {
     rpc Accept(AcceptRequest) returns (google.protobuf.Empty) {}
     rpc AcceptationTimeout(AcceptationTimeoutRequest) returns (google.protobuf.Empty) {}
     rpc GetTransferState(GetRequest) returns (com.example.transfer.domain.TransferState) {}
+    rpc Delete(DeleteRequest) returns (google.protobuf.Empty) {}
+    rpc HasBeenDeleted(HasBeenDeletedRequest) returns (HasBeenDeletedResponse) {}
 }

samples/scala-protobuf-transfer-workflow-compensation/src/main/scala/com/example/transfer/api/TransferWorkflow.scala

@@ -160,7 +160,7 @@ class TransferWorkflow(context: WorkflowContext) extends AbstractTransferWorkflo
       .addStep(waitForAcceptation)
       .addStep(compensateWithdraw) // <4>
       .addStep(failoverHandler);
-      // end::recover-strategy[]
+    // end::recover-strategy[]
   }
 
   override def start(currentState: TransferState, transfer: Transfer): Effect[Empty] = {
@@ -230,4 +230,16 @@ class TransferWorkflow(context: WorkflowContext) extends AbstractTransferWorkflo
     }
   }
 
+  // tag::delete-workflow[]
+  override def delete(currentState: TransferState, deleteRequest: DeleteRequest): Effect[Empty] = {
+    effects.delete // <1>
+      .thenReply(Empty())
+  }
+  // end::delete-workflow[]
+
+  override def hasBeenDeleted(
+      currentState: TransferState,
+      hasBeenDeletedRequest: HasBeenDeletedRequest): Effect[HasBeenDeletedResponse] =
+    effects.reply(HasBeenDeletedResponse(isDeleted))
+
 }

samples/scala-protobuf-transfer-workflow-compensation/src/test/scala/com/example/transfer/api/TransferWorkflowIntegrationSpec.scala

@@ -47,14 +47,23 @@ class TransferWorkflowIntegrationSpec
       walletClient.create(initWalletA).futureValue
       walletClient.create(initWalletB).futureValue
 
-      transferClient.start(Transfer("1", "a", "b", 10)).futureValue
+      val transferId = "1"
+      transferClient.start(Transfer(transferId, "a", "b", 10)).futureValue
 
       eventually {
         val walletA = walletClient.getWalletState(api.GetRequest("a")).futureValue
         val walletB = walletClient.getWalletState(api.GetRequest("b")).futureValue
         walletA.balance shouldBe 90
         walletB.balance shouldBe 110
       }
+
+      val response = transferClient.hasBeenDeleted(HasBeenDeletedRequest(transferId)).futureValue
+      response.deleted shouldBe false
+
+      transferClient.delete(DeleteRequest(transferId)).futureValue
+
+      val response2 = transferClient.hasBeenDeleted(HasBeenDeletedRequest(transferId)).futureValue
+      response2.deleted shouldBe true
     }
 
     "transfer funds with acceptation" in {