Add Workflow API e2e test (#264)

contracts/build.gradle.kts

@@ -17,6 +17,8 @@ plugins {
 }
 
 dependencies {
+  annotationProcessor(libs.restate.sdk.api.gen)
+
   compileOnly(libs.javax.annotation.api)
 
   api(libs.protobuf.java)
@@ -26,6 +28,9 @@ dependencies {
   api(libs.grpc.kotlin.stub) { exclude("javax.annotation", "javax.annotation-api") }
 
   protobuf(libs.restate.sdk.common)
+
+  api(libs.restate.sdk.workflow.api)
+  api(libs.restate.sdk.jackson)
 }
 
 protobuf {

contracts/src/main/java/my/restate/e2e/services/WorkflowAPIBlockAndWait.java

@@ -0,0 +1,30 @@
+// Copyright (c) 2024 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate e2e tests,
+// which are released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/e2e/blob/main/LICENSE
+
+package my.restate.e2e.services;
+
+import dev.restate.sdk.annotation.Service;
+import dev.restate.sdk.annotation.ServiceType;
+import dev.restate.sdk.annotation.Shared;
+import dev.restate.sdk.annotation.Workflow;
+import dev.restate.sdk.common.StateKey;
+import dev.restate.sdk.workflow.WorkflowContext;
+import dev.restate.sdk.workflow.WorkflowSharedContext;
+
+@Service(ServiceType.WORKFLOW)
+public interface WorkflowAPIBlockAndWait {
+
+  @Workflow
+  String blockAndWait(WorkflowContext context, String input);
+
+  @Shared
+  void unblock(WorkflowSharedContext context, String output);
+
+  StateKey<String> MY_STATE = StateKey.string("my-state");
+}

services/java-services/src/main/java/dev/restate/e2e/services/Main.java

@@ -39,6 +39,8 @@
 import dev.restate.e2e.services.upgradetest.UpgradeTestServiceGrpc;
 import dev.restate.sdk.http.vertx.RestateHttpEndpointBuilder;
 import java.util.Objects;
+import my.restate.e2e.services.WorkflowAPIBlockAndWaitImpl;
+import my.restate.e2e.services.WorkflowAPIBlockAndWaitServiceAdapter;
 
 public class Main {
 
@@ -49,8 +51,9 @@ public static void main(String[] args) {
             "SERVICES env variable needs to specify which service to run.");
 
     RestateHttpEndpointBuilder restateHttpEndpointBuilder = RestateHttpEndpointBuilder.builder();
-    for (String service : env.split(",")) {
-      switch (service.trim()) {
+    for (String svc : env.split(",")) {
+      String fqsn = svc.trim();
+      switch (fqsn) {
         case ListServiceGrpc.SERVICE_NAME:
           restateHttpEndpointBuilder.withService(new ListService());
           break;
@@ -95,6 +98,10 @@ public static void main(String[] args) {
         case MapServiceGrpc.SERVICE_NAME:
           restateHttpEndpointBuilder.withService(new MapService());
           break;
+        case WorkflowAPIBlockAndWaitServiceAdapter.SERVICE_NAME:
+          restateHttpEndpointBuilder.with(
+              new WorkflowAPIBlockAndWaitImpl(), new WorkflowAPIBlockAndWaitServiceAdapter());
+          break;
       }
     }
 

services/java-services/src/main/java/my/restate/e2e/services/WorkflowAPIBlockAndWaitImpl.java

@@ -0,0 +1,43 @@
+// Copyright (c) 2024 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate e2e tests,
+// which are released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/e2e/blob/main/LICENSE
+
+package my.restate.e2e.services;
+
+import dev.restate.sdk.common.TerminalException;
+import dev.restate.sdk.workflow.DurablePromiseKey;
+import dev.restate.sdk.workflow.WorkflowContext;
+import dev.restate.sdk.workflow.WorkflowSharedContext;
+
+public class WorkflowAPIBlockAndWaitImpl implements WorkflowAPIBlockAndWait {
+
+  private static final DurablePromiseKey<String> MY_DURABLE_PROMISE =
+      DurablePromiseKey.string("durable-promise");
+
+  @Override
+  public String blockAndWait(WorkflowContext context, String input) {
+    context.set(MY_STATE, input);
+
+    // Wait on unblock
+    String output = context.durablePromise(MY_DURABLE_PROMISE).awaitable().await();
+
+    if (!context.durablePromise(MY_DURABLE_PROMISE).isCompleted()) {
+      throw new TerminalException("Durable promise should be completed");
+    }
+    if (context.durablePromise(MY_DURABLE_PROMISE).peek().isEmpty()) {
+      throw new TerminalException("Durable promise should be completed");
+    }
+
+    return output;
+  }
+
+  @Override
+  public void unblock(WorkflowSharedContext context, String output) {
+    context.durablePromiseHandle(MY_DURABLE_PROMISE).resolve(output);
+  }
+}

services/node-services/src/app.ts

@@ -71,6 +71,10 @@ import {
   CancelTestServiceFQN,
   BlockingService,
 } from "./cancel_test";
+import {
+  WorkflowAPIBlockAndWait,
+  WorkflowAPIBlockAndWaitFQN,
+} from "./workflow";
 
 let serverBuilder;
 export let handler: (event: any) => Promise<any>;
@@ -82,7 +86,7 @@ if (process.env.AWS_LAMBDA_FUNCTION_NAME) {
 
 const services = new Map<
   string,
-  restate.ServiceOpts | { router: any } | { keyedRouter: any }
+  restate.ServiceOpts | { router: any } | { keyedRouter: any } | { bundle: any }
 >([
   [
     CounterServiceFQN,
@@ -248,6 +252,12 @@ const services = new Map<
       instance: new MapService(),
     },
   ],
+  [
+    WorkflowAPIBlockAndWaitFQN,
+    {
+      bundle: WorkflowAPIBlockAndWait,
+    },
+  ],
 ]);
 console.log("Known services: " + services.keys());
 
@@ -269,20 +279,25 @@ if (process.env.SERVICES) {
       serverBuilder = serverBuilder.bindService(
         foundService as restate.ServiceOpts
       );
-    } else if (
-      (foundService as restate.UnKeyedRouter<any>).router !== undefined
-    ) {
+    } else if ((foundService as { router: any }).router !== undefined) {
       console.log("Mounting router " + service);
       serverBuilder = serverBuilder.bindRouter(
         service,
         (foundService as { router: any }).router
       );
-    } else {
+    } else if (
+      (foundService as { keyedRouter: any }).keyedRouter !== undefined
+    ) {
       console.log("Mounting keyed router " + service);
       serverBuilder = serverBuilder.bindKeyedRouter(
         service,
         (foundService as { keyedRouter: any }).keyedRouter
       );
+    } else {
+      console.log("Mounting bundle " + service);
+      serverBuilder = serverBuilder.bind(
+        (foundService as { bundle: any }).bundle
+      );
     }
   }
 

services/node-services/src/workflow.ts

@@ -0,0 +1,47 @@
+// Copyright (c) 2024 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate e2e tests,
+// which are released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/e2e/blob/main/LICENSE
+
+import * as restate from "@restatedev/restate-sdk";
+
+const MY_STATE = "my-state";
+const MY_DURABLE_PROMISE = "durable-promise";
+
+export const WorkflowAPIBlockAndWaitFQN = "WorkflowAPIBlockAndWait";
+
+export const WorkflowAPIBlockAndWait = restate.workflow.workflow(
+  WorkflowAPIBlockAndWaitFQN,
+  {
+    run: async (ctx: restate.workflow.WfContext, params: { input: string }) => {
+      ctx.console.log("input: " + JSON.stringify(params));
+      ctx.set(MY_STATE, params.input);
+
+      // Wait on unblock
+      const p = ctx.promise<string>(MY_DURABLE_PROMISE);
+      const output = await p.promise();
+
+      // Check peek works
+      ctx.console.assert((await p.peek()) == output);
+
+      return output;
+    },
+
+    unblock: async (
+      ctx: restate.workflow.SharedWfContext,
+      params: { output: string }
+    ) => {
+      ctx.promise<string>(MY_DURABLE_PROMISE).resolve(params.output);
+    },
+
+    getState: async (
+      ctx: restate.workflow.SharedWfContext
+    ): Promise<string> => {
+      return (await ctx.get(MY_STATE)) ?? "(not yet set)";
+    },
+  }
+);

settings.gradle.kts

@@ -45,8 +45,10 @@ dependencyResolutionManagement {
       library("restate-sdk-common", "dev.restate", "sdk-common").versionRef("restate")
       library("restate-admin", "dev.restate", "admin-client").versionRef("restate")
       library("restate-sdk-api", "dev.restate", "sdk-api").versionRef("restate")
+      library("restate-sdk-api-gen", "dev.restate", "sdk-api-gen").versionRef("restate")
       library("restate-sdk-jackson", "dev.restate", "sdk-serde-jackson").versionRef("restate")
       library("restate-sdk-http-vertx", "dev.restate", "sdk-http-vertx").versionRef("restate")
+      library("restate-sdk-workflow-api", "dev.restate", "sdk-workflow-api").versionRef("restate")
 
       library("protoc", "com.google.protobuf", "protoc").versionRef("protobuf")
       library("protobuf-java", "com.google.protobuf", "protobuf-java").versionRef("protobuf")
@@ -93,6 +95,8 @@ if (!System.getenv("JAVA_SDK_LOCAL_BUILD").isNullOrEmpty()) {
       substitute(module("dev.restate:admin-client")).using(project(":admin-client"))
       substitute(module("dev.restate:sdk-common")).using(project(":sdk-common"))
       substitute(module("dev.restate:sdk-api")).using(project(":sdk-api"))
+      substitute(module("dev.restate:sdk-api-gen")).using(project(":sdk-api-gen"))
+      substitute(module("dev.restate:sdk-workflow-api")).using(project(":sdk-workflow-api"))
       substitute(module("dev.restate:sdk-http-vertx")).using(project(":sdk-http-vertx"))
       substitute(module("dev.restate:sdk-serde-jackson")).using(project(":sdk-serde-jackson"))
     }

tests/src/test/kotlin/dev/restate/e2e/AwaitTimeoutTest.kt

@@ -47,8 +47,6 @@ class NodeAwaitTimeoutTest : BaseAwaitTimeoutTest() {
   }
 }
 
-// Only a Java test because the typescript SDK is still lacking this feature:
-// https://github.com/restatedev/sdk-typescript/issues/21
 @Tag("always-suspending")
 abstract class BaseAwaitTimeoutTest {
 

tests/src/test/kotlin/dev/restate/e2e/Containers.kt

@@ -21,6 +21,7 @@ import dev.restate.e2e.services.singletoncounter.SingletonCounterGrpc
 import dev.restate.e2e.services.verification.interpreter.CommandInterpreterGrpc
 import dev.restate.e2e.services.verification.verifier.CommandVerifierGrpc
 import dev.restate.e2e.utils.ServiceSpec
+import my.restate.e2e.services.WorkflowAPIBlockAndWaitServiceAdapter
 import org.testcontainers.containers.GenericContainer
 
 object Containers {
@@ -72,6 +73,10 @@ object Containers {
               "HTTP_SERVER_ADDRESS", "http://${INT_SORTER_HTTP_SERVER_CONTAINER_SPEC.first}:8080")
           .build()
 
+  val JAVA_WORKFLOW_SERVICE_SPEC =
+      javaServicesContainer("java-workflow", WorkflowAPIBlockAndWaitServiceAdapter.SERVICE_NAME)
+          .build()
+
   // -- Node containers
 
   fun nodeServicesContainer(hostName: String, vararg services: String): ServiceSpec.Builder {
@@ -112,6 +117,10 @@ object Containers {
   val NODE_HANDLER_API_ECHO_TEST_SERVICE_SPEC =
       nodeServicesContainer("node-proxy", HANDLER_API_ECHO_TEST_SERVICE_NAME).build()
 
+  const val WORKFLOW_API_BLOCK_AND_WAIT_SERVICE_NAME = "WorkflowAPIBlockAndWait"
+  val NODE_WORKFLOW_SERVICE_SPEC =
+      nodeServicesContainer("node-workflow", WORKFLOW_API_BLOCK_AND_WAIT_SERVICE_NAME).build()
+
   const val EMBEDDED_HANDLER_SERVER_HOSTNAME = "node-embedded-handler"
   const val EMBEDDED_HANDLER_SERVER_PORT = 8080
 

tests/src/test/kotlin/dev/restate/e2e/Utils.kt

@@ -16,6 +16,7 @@ import java.net.http.HttpClient
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
 import java.nio.charset.StandardCharsets
+import org.assertj.core.api.Assertions.assertThat
 
 object Utils {
 
@@ -41,4 +42,19 @@ object Utils {
             .build()
     return httpClient.send(req, jacksonBodyHandler())
   }
+
+  fun doJsonRequestToService(
+      restateEndpoint: String,
+      service: String,
+      method: String,
+      reqBody: Any
+  ): JsonNode {
+    val res = postJsonRequest("${restateEndpoint}${service}/${method}", reqBody)
+    assertThat(res.statusCode()).isEqualTo(200)
+    assertThat(res.headers().firstValue("content-type"))
+        .get()
+        .asString()
+        .contains("application/json")
+    return res.body()
+  }
 }