feat(nexus): add async, cancellation, failure and parallel features

features/features.go

@@ -32,6 +32,10 @@ import (
 	deployment_versioning_routing_with_ramp "github.com/temporalio/features/features/deployment_versioning/routing_with_ramp"
 	eager_activity_non_remote_activities_worker "github.com/temporalio/features/features/eager_activity/non_remote_activities_worker"
 	eager_workflow_successful_start "github.com/temporalio/features/features/eager_workflow/successful_start"
+	nexus_async_cancellation "github.com/temporalio/features/features/nexus/async_cancellation"
+	nexus_async_success "github.com/temporalio/features/features/nexus/async_success"
+	nexus_parallel_operations "github.com/temporalio/features/features/nexus/parallel_operations"
+	nexus_sync_operation_error "github.com/temporalio/features/features/nexus/sync_operation_error"
 	nexus_sync_success "github.com/temporalio/features/features/nexus/sync_success"
 	query_successful_query "github.com/temporalio/features/features/query/successful_query"
 	query_timeout_due_to_no_active_workers "github.com/temporalio/features/features/query/timeout_due_to_no_active_workers"
@@ -95,6 +99,10 @@ func init() {
 		deployment_versioning_routing_with_ramp.Feature,
 		eager_activity_non_remote_activities_worker.Feature,
 		eager_workflow_successful_start.Feature,
+		nexus_async_cancellation.Feature,
+		nexus_async_success.Feature,
+		nexus_parallel_operations.Feature,
+		nexus_sync_operation_error.Feature,
 		nexus_sync_success.Feature,
 		query_successful_query.Feature,
 		query_timeout_due_to_no_active_workers.Feature,

features/nexus/async_cancellation/README.md

@@ -0,0 +1,13 @@
+# Nexus async operation is cancelled
+
+A workflow cancels a running asynchronous Nexus operation and observes a cancellation error.
+
+# Detailed spec
+
+- The backing workflow of the operation blocks until it is cancelled, so the cancellation is
+  deterministic and never races with completion.
+- The caller starts the operation in a cancellable scope and waits until the operation has
+  actually started before cancelling that scope.
+- Cancelling the scope requests cancellation of the operation, which cancels the backing
+  workflow, and the operation future resolves with a cancellation error.
+- The caller handles that error and completes successfully.

features/nexus/async_cancellation/feature.go

@@ -0,0 +1,89 @@
+package async_cancellation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	"github.com/temporalio/features/harness/go/harness"
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
+	"go.temporal.io/sdk/temporalnexus"
+	"go.temporal.io/sdk/workflow"
+)
+
+const ServiceName = "test-service"
+
+// BlockingWorkflow never completes on its own - it only ends when cancelled, which makes the
+// cancellation race deterministic.
+func BlockingWorkflow(ctx workflow.Context, name string) (string, error) {
+	ctx.Done().Receive(ctx, nil)
+	return "", ctx.Err()
+}
+
+var AsyncOperation = temporalnexus.NewWorkflowRunOperation(
+	"block-forever",
+	BlockingWorkflow,
+	func(ctx context.Context, name string, options nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
+		return client.StartWorkflowOptions{ID: "async-cancellation-" + name}, nil
+	},
+)
+
+var Service = func() *nexus.Service {
+	s := nexus.NewService(ServiceName)
+	s.MustRegister(AsyncOperation)
+	return s
+}()
+
+func Workflow(ctx workflow.Context, endpoint string) (string, error) {
+	nc := workflow.NewNexusClient(endpoint, ServiceName)
+	opCtx, cancel := workflow.WithCancel(ctx)
+	fut := nc.ExecuteOperation(opCtx, AsyncOperation, "world", workflow.NexusOperationOptions{
+		ScheduleToCloseTimeout: time.Minute,
+	})
+	var exec workflow.NexusOperationExecution
+	if err := fut.GetNexusOperationExecution().Get(ctx, &exec); err != nil {
+		return "", err
+	}
+	cancel()
+
+	err := fut.Get(ctx, nil)
+	if err == nil {
+		return "", harness.AppErrorf("expected the cancelled operation to fail")
+	}
+	var canceledErr *temporal.CanceledError
+	if !errors.As(err, &canceledErr) {
+		return "", harness.AppErrorf("expected a canceled error, got %v", err)
+	}
+	return "canceled", nil
+}
+
+var Feature = harness.Feature{
+	Workflows:       []interface{}{Workflow, BlockingWorkflow},
+	NexusServices:   Service,
+	ExpectRunResult: "canceled",
+	Execute: func(ctx context.Context, runner *harness.Runner) (client.WorkflowRun, error) {
+		opts := client.StartWorkflowOptions{
+			TaskQueue:                runner.TaskQueue,
+			WorkflowExecutionTimeout: time.Minute,
+		}
+		return runner.Client.ExecuteWorkflow(ctx, opts, Workflow, runner.NexusEndpoint)
+	},
+	CheckHistory: func(ctx context.Context, runner *harness.Runner, run client.WorkflowRun) error {
+		hist := runner.Client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+		ev, err := harness.FindEvent(hist, func(ev *historypb.HistoryEvent) bool {
+			return ev.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED
+		})
+		if err != nil {
+			return err
+		}
+		if ev == nil {
+			return fmt.Errorf("did not find NexusOperationCancelRequested event in history")
+		}
+		return nil
+	},
+}

features/nexus/async_cancellation/feature.java

@@ -0,0 +1,136 @@
+package nexus.async_cancellation;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.nexusrpc.Operation;
+import io.nexusrpc.Service;
+import io.nexusrpc.handler.OperationHandler;
+import io.nexusrpc.handler.OperationImpl;
+import io.nexusrpc.handler.ServiceImpl;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.failure.CanceledFailure;
+import io.temporal.nexus.Nexus;
+import io.temporal.nexus.WorkflowRunOperation;
+import io.temporal.sdkfeatures.Feature;
+import io.temporal.sdkfeatures.Run;
+import io.temporal.sdkfeatures.Runner;
+import io.temporal.worker.Worker;
+import io.temporal.workflow.CancellationScope;
+import io.temporal.workflow.NexusOperationOptions;
+import io.temporal.workflow.NexusServiceOptions;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+import java.time.Duration;
+
+@WorkflowInterface
+public interface feature extends Feature {
+  @WorkflowMethod
+  String workflow(String endpoint);
+
+  @Service
+  interface TestService {
+    @Operation
+    String blockingOperation(String name);
+  }
+
+  @WorkflowInterface
+  interface HandlerWorkflow {
+    @WorkflowMethod
+    String handlerWorkflow(String name);
+  }
+
+  class HandlerWorkflowImpl implements HandlerWorkflow {
+    @Override
+    public String handlerWorkflow(String name) {
+      Workflow.await(() -> false);
+      return "unreachable";
+    }
+  }
+
+  class Impl implements feature {
+    @Override
+    public String workflow(String endpoint) {
+      var serviceOptions =
+          NexusServiceOptions.newBuilder()
+              .setEndpoint(endpoint)
+              .setOperationOptions(
+                  NexusOperationOptions.newBuilder()
+                      .setScheduleToCloseTimeout(Duration.ofMinutes(1))
+                      .build())
+              .build();
+      TestService stub = Workflow.newNexusServiceStub(TestService.class, serviceOptions);
+
+      var resultHolder = new String[1];
+      var scope =
+          Workflow.newCancellationScope(
+              () -> {
+                var handle = Workflow.startNexusOperation(stub::blockingOperation, "world");
+                handle.getExecution().get();
+                handle
+                    .getResult()
+                    .handle(
+                        (value, failure) -> {
+                          resultHolder[0] = failure == null ? "completed" : "cancelled";
+                          return null;
+                        });
+              });
+      scope.run();
+      scope.cancel();
+      Workflow.await(() -> resultHolder[0] != null);
+      return "operation " + resultHolder[0];
+    }
+
+    @Override
+    public Object[] nexusServiceImplementations() {
+      return new Object[] {new TestServiceImpl()};
+    }
+
+    @Override
+    public void prepareWorker(Worker worker) {
+      worker.registerWorkflowImplementationTypes(HandlerWorkflowImpl.class);
+    }
+
+    @Override
+    public Run execute(Runner runner) throws Exception {
+      var options =
+          WorkflowOptions.newBuilder()
+              .setTaskQueue(runner.config.taskQueue)
+              .setWorkflowExecutionTimeout(Duration.ofMinutes(1))
+              .build();
+      return runner.executeSingleWorkflow(options, runner.nexusEndpoint);
+    }
+
+    @Override
+    public void checkResult(Runner runner, Run run) {
+      var result = runner.waitForRunResult(run, String.class);
+      assertEquals("operation cancelled", result);
+    }
+
+    @Override
+    public void checkHistory(Runner runner, Run run) throws Exception {
+      var events = runner.getWorkflowHistory(run).getEventsList();
+      assertTrue(
+          events.stream().anyMatch(e -> e.hasNexusOperationCancelRequestedEventAttributes()),
+          "expected NexusOperationCancelRequested event in history");
+    }
+  }
+
+  @ServiceImpl(service = TestService.class)
+  class TestServiceImpl {
+    @OperationImpl
+    public OperationHandler<String, String> blockingOperation() {
+      return WorkflowRunOperation.fromWorkflowMethod(
+          (context, details, name) ->
+              Nexus.getOperationContext()
+                      .getWorkflowClient()
+                      .newWorkflowStub(
+                          HandlerWorkflow.class,
+                          WorkflowOptions.newBuilder()
+                              .setWorkflowId(details.getRequestId())
+                              .build())
+                  ::handlerWorkflow);
+    }
+  }
+}

features/nexus/async_success/README.md

@@ -0,0 +1,14 @@
+# Nexus async operation succeeds
+
+A workflow invokes an asynchronous Nexus operation backed by a workflow run, observes the
+operation token, and then receives the backing workflow's result.
+
+# Detailed spec
+
+- A Nexus service with a workflow-run operation is registered on the worker, along with the
+  backing workflow it starts.
+- The caller workflow executes the operation and first awaits the operation execution, which
+  carries a non-empty operation token.
+- The caller then awaits the operation result, which is the output of the backing workflow.
+- An async operation is scheduled, then started once the backing workflow is running, and
+  completed when the backing workflow completes.

features/nexus/async_success/feature.go

@@ -0,0 +1,84 @@
+package async_success
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	"github.com/temporalio/features/harness/go/harness"
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporalnexus"
+	"go.temporal.io/sdk/workflow"
+)
+
+const ServiceName = "test-service"
+
+func HandlerWorkflow(ctx workflow.Context, name string) (string, error) {
+	return "Hello, " + name + "!", nil
+}
+
+var AsyncOperation = temporalnexus.NewWorkflowRunOperation(
+	"say-hello-async",
+	HandlerWorkflow,
+	func(ctx context.Context, name string, options nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
+		return client.StartWorkflowOptions{ID: "async-success-" + name}, nil
+	},
+)
+
+var Service = func() *nexus.Service {
+	s := nexus.NewService(ServiceName)
+	s.MustRegister(AsyncOperation)
+	return s
+}()
+
+func Workflow(ctx workflow.Context, endpoint string) (string, error) {
+	nc := workflow.NewNexusClient(endpoint, ServiceName)
+	fut := nc.ExecuteOperation(ctx, AsyncOperation, "world", workflow.NexusOperationOptions{
+		ScheduleToCloseTimeout: time.Minute,
+	})
+	var exec workflow.NexusOperationExecution
+	if err := fut.GetNexusOperationExecution().Get(ctx, &exec); err != nil {
+		return "", err
+	}
+	if exec.OperationToken == "" {
+		return "", harness.AppErrorf("expected a non-empty operation token")
+	}
+	var result string
+	if err := fut.Get(ctx, &result); err != nil {
+		return "", err
+	}
+	return "token+" + result, nil
+}
+
+var Feature = harness.Feature{
+	Workflows:       []interface{}{Workflow, HandlerWorkflow},
+	NexusServices:   Service,
+	ExpectRunResult: "token+Hello, world!",
+	Execute: func(ctx context.Context, runner *harness.Runner) (client.WorkflowRun, error) {
+		opts := client.StartWorkflowOptions{
+			TaskQueue:                runner.TaskQueue,
+			WorkflowExecutionTimeout: time.Minute,
+		}
+		return runner.Client.ExecuteWorkflow(ctx, opts, Workflow, runner.NexusEndpoint)
+	},
+	CheckHistory: func(ctx context.Context, runner *harness.Runner, run client.WorkflowRun) error {
+		for _, t := range []enumspb.EventType{
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED,
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED,
+			enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED,
+		} {
+			hist := runner.Client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+			ev, err := harness.FindEvent(hist, func(ev *historypb.HistoryEvent) bool { return ev.EventType == t })
+			if err != nil {
+				return err
+			}
+			if ev == nil {
+				return fmt.Errorf("did not find %v event in history", t)
+			}
+		}
+		return nil
+	},
+}