Add Async Workflow Nexus Test

features/features.go

@@ -33,6 +33,7 @@ import (
 	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_sync_success "github.com/temporalio/features/features/nexus/sync_success"
+	nexus_workflow_run_success "github.com/temporalio/features/features/nexus/workflow_run_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"
 	query_unexpected_arguments "github.com/temporalio/features/features/query/unexpected_arguments"
@@ -95,6 +96,7 @@ func init() {
 		eager_activity_non_remote_activities_worker.Feature,
 		eager_workflow_successful_start.Feature,
 		nexus_sync_success.Feature,
+		nexus_workflow_run_success.Feature,
 		query_successful_query.Feature,
 		query_timeout_due_to_no_active_workers.Feature,
 		query_unexpected_arguments.Feature,

features/nexus/workflow_run_success/README.md

@@ -0,0 +1,14 @@
+# Nexus async workflow operation succeeds
+
+A workflow invokes a Nexus operation backed by another workflow and receives its result.
+
+# Detailed spec
+
+- A Nexus service with a workflow-run operation is registered on the worker.
+- The caller workflow executes the operation against a Nexus endpoint and awaits the result.
+- The operation starts a handler workflow; the handler workflow's result is returned as the
+  operation result and then as the caller workflow's result.
+- An async operation transitions Scheduled -> Started -> Completed.
+- The caller's NexusOperationStarted event links to the handler workflow's
+  WorkflowExecutionStarted event, and the handler workflow's WorkflowExecutionStarted
+  event links back to the caller's NexusOperationScheduled event.

features/nexus/workflow_run_success/feature.go

@@ -0,0 +1,161 @@
+package workflow_run_success
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	"github.com/temporalio/features/harness/go/harness"
+	commonpb "go.temporal.io/api/common/v1"
+	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 AsyncWorkflowOperation = temporalnexus.NewWorkflowRunOperation(
+	"AsyncWorkflowOperation",
+	HandlerWorkflow,
+	func(ctx context.Context, input string, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
+		// Use the request ID so retried start requests resolve to the same workflow.
+		return client.StartWorkflowOptions{ID: opts.RequestID}, nil
+	},
+)
+
+var Service = func() *nexus.Service {
+	s := nexus.NewService(ServiceName)
+	s.MustRegister(AsyncWorkflowOperation)
+	return s
+}()
+
+func CallerWorkflow(ctx workflow.Context, endpoint string) (string, error) {
+	nc := workflow.NewNexusClient(endpoint, ServiceName)
+	fut := nc.ExecuteOperation(ctx, AsyncWorkflowOperation, "world", workflow.NexusOperationOptions{
+		ScheduleToCloseTimeout: time.Minute,
+	})
+	var result string
+	if err := fut.Get(ctx, &result); err != nil {
+		return "", err
+	}
+	return result, nil
+}
+
+var Feature = harness.Feature{
+	Workflows:       []interface{}{CallerWorkflow, HandlerWorkflow},
+	NexusServices:   Service,
+	ExpectRunResult: "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, CallerWorkflow, runner.NexusEndpoint)
+	},
+	CheckHistory: func(ctx context.Context, runner *harness.Runner, run client.WorkflowRun) error {
+		// Async (workflow-run) Nexus operations should transition Scheduled -> Started -> Completed.
+		findCallerEvent := func(t enumspb.EventType) (*historypb.HistoryEvent, error) {
+			hist := runner.Client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+			return harness.FindEvent(hist, func(ev *historypb.HistoryEvent) bool { return ev.EventType == t })
+		}
+		scheduled, err := findCallerEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED)
+		if err != nil {
+			return err
+		}
+		if scheduled == nil {
+			return fmt.Errorf("did not find NexusOperationScheduled event in history")
+		}
+		started, err := findCallerEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED)
+		if err != nil {
+			return err
+		}
+		if started == nil {
+			return fmt.Errorf("did not find NexusOperationStarted event in history")
+		}
+		if completed, err := findCallerEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED); err != nil {
+			return err
+		} else if completed == nil {
+			return fmt.Errorf("did not find NexusOperationCompleted event in history")
+		}
+
+		// The caller's NexusOperationStarted event must link to the handler workflow's
+		// WorkflowExecutionStarted event.
+		handlerLink := findWorkflowEventLink(started.GetLinks(), enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED)
+		if handlerLink == nil {
+			return fmt.Errorf("NexusOperationStarted is missing a link to the handler WorkflowExecutionStarted event")
+		}
+		if handlerLink.GetNamespace() != runner.Namespace {
+			return fmt.Errorf("handler link namespace = %q, want %q", handlerLink.GetNamespace(), runner.Namespace)
+		}
+		// WorkflowExecutionStarted is always event ID 1.
+		if handlerLink.GetEventRef().GetEventId() != 1 {
+			return fmt.Errorf("handler link eventId = %d, want 1", handlerLink.GetEventRef().GetEventId())
+		}
+		// The handler workflow ID is set to the Nexus operation request ID by the operation impl.
+		wantHandlerWorkflowID := scheduled.GetNexusOperationScheduledEventAttributes().GetRequestId()
+		if handlerLink.GetWorkflowId() != wantHandlerWorkflowID {
+			return fmt.Errorf("handler link workflowId = %q, want %q", handlerLink.GetWorkflowId(), wantHandlerWorkflowID)
+		}
+
+		// The handler workflow's WorkflowExecutionStarted event carries the Nexus completion
+		// callback, whose link points back to the caller's NexusOperationScheduled event.
+		// (Nexus links on the started event itself are deduped against the callback link.)
+		handlerHist := runner.Client.GetWorkflowHistory(ctx, handlerLink.GetWorkflowId(), handlerLink.GetRunId(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+		handlerStarted, err := harness.FindEvent(handlerHist, func(ev *historypb.HistoryEvent) bool {
+			return ev.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED
+		})
+		if err != nil {
+			return err
+		}
+		if handlerStarted == nil {
+			return fmt.Errorf("did not find WorkflowExecutionStarted event in handler history")
+		}
+		handlerAttrs := handlerStarted.GetWorkflowExecutionStartedEventAttributes()
+		// Cross-check the run ID embedded in the caller's link against the handler's own attrs.
+		if handlerLink.GetRunId() != handlerAttrs.GetFirstExecutionRunId() {
+			return fmt.Errorf("handler link runId = %q, want %q (firstExecutionRunId)",
+				handlerLink.GetRunId(), handlerAttrs.GetFirstExecutionRunId())
+		}
+		callbacks := handlerAttrs.GetCompletionCallbacks()
+		if len(callbacks) == 0 {
+			return fmt.Errorf("handler WorkflowExecutionStarted has no completion callbacks")
+		}
+		callerLink := findWorkflowEventLink(callbacks[0].GetLinks(), enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED)
+		if callerLink == nil {
+			return fmt.Errorf("handler completion callback is missing a link to the caller NexusOperationScheduled event")
+		}
+		if callerLink.GetNamespace() != runner.Namespace {
+			return fmt.Errorf("caller link namespace = %q, want %q", callerLink.GetNamespace(), runner.Namespace)
+		}
+		if callerLink.GetWorkflowId() != run.GetID() || callerLink.GetRunId() != run.GetRunID() {
+			return fmt.Errorf("handler callback link references %s/%s, expected caller %s/%s",
+				callerLink.GetWorkflowId(), callerLink.GetRunId(), run.GetID(), run.GetRunID())
+		}
+		if callerLink.GetEventRef().GetEventId() != scheduled.GetEventId() {
+			return fmt.Errorf("caller link eventId = %d, want %d", callerLink.GetEventRef().GetEventId(), scheduled.GetEventId())
+		}
+		return nil
+	},
+}
+
+// findWorkflowEventLink returns the first WorkflowEvent-variant link whose event reference matches
+// the given event type, or nil if none match.
+func findWorkflowEventLink(links []*commonpb.Link, eventType enumspb.EventType) *commonpb.Link_WorkflowEvent {
+	for _, l := range links {
+		we := l.GetWorkflowEvent()
+		if we == nil {
+			continue
+		}
+		if we.GetEventRef().GetEventType() == eventType {
+			return we
+		}
+	}
+	return nil
+}

features/nexus/workflow_run_success/feature.java

@@ -0,0 +1,197 @@
+package nexus.workflow_run_success;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+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.api.common.v1.Link;
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.api.enums.v1.EventType;
+import io.temporal.api.history.v1.History;
+import io.temporal.api.history.v1.HistoryEvent;
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.internal.client.WorkflowClientHelper;
+import io.temporal.nexus.Nexus;
+import io.temporal.nexus.WorkflowHandle;
+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.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;
+import java.util.List;
+
+@WorkflowInterface
+public interface feature extends Feature {
+  @WorkflowMethod
+  String workflow(String endpoint);
+
+  @Service
+  interface TestService {
+    @Operation(name = "AsyncWorkflowOperation")
+    String asyncWorkflowOperation(String name);
+  }
+
+  @WorkflowInterface
+  interface HandlerWorkflow {
+    @WorkflowMethod
+    String run(String name);
+  }
+
+  class HandlerWorkflowImpl implements HandlerWorkflow {
+    @Override
+    public String run(String name) {
+      return "Hello, " + name + "!";
+    }
+  }
+
+  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);
+      return stub.asyncWorkflowOperation("world");
+    }
+
+    @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("Hello, world!", result);
+    }
+
+    @Override
+    public void checkHistory(Runner runner, Run run) throws Exception {
+      // Async (workflow-run) Nexus operations should transition Scheduled -> Started -> Completed.
+      var events = runner.getWorkflowHistory(run).getEventsList();
+      var scheduled = findEvent(events, e -> e.hasNexusOperationScheduledEventAttributes());
+      assertNotNull(scheduled, "expected NexusOperationScheduled event in history");
+      var started = findEvent(events, e -> e.hasNexusOperationStartedEventAttributes());
+      assertNotNull(started, "expected NexusOperationStarted event in history");
+      var completed = findEvent(events, e -> e.hasNexusOperationCompletedEventAttributes());
+      assertNotNull(completed, "expected NexusOperationCompleted event in history");
+
+      // The caller's NexusOperationStarted event must link to the handler workflow's
+      // WorkflowExecutionStarted event.
+      var handlerLink =
+          findWorkflowEventLink(
+              started.getLinksList(), EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED);
+      assertNotNull(
+          handlerLink,
+          "NexusOperationStarted is missing a link to the handler WorkflowExecutionStarted event");
+      assertEquals(runner.config.namespace, handlerLink.getNamespace());
+      // WorkflowExecutionStarted is always event ID 1.
+      assertEquals(1L, handlerLink.getEventRef().getEventId());
+      // The handler workflow ID is set to the Nexus operation request ID by the operation impl.
+      assertEquals(
+          scheduled.getNexusOperationScheduledEventAttributes().getRequestId(),
+          handlerLink.getWorkflowId());
+
+      // The handler workflow's WorkflowExecutionStarted event carries the Nexus completion
+      // callback, whose link points back to the caller's NexusOperationScheduled event.
+      // (Nexus links on the started event itself are deduped against the callback link.)
+      var handlerExec =
+          WorkflowExecution.newBuilder()
+              .setWorkflowId(handlerLink.getWorkflowId())
+              .setRunId(handlerLink.getRunId())
+              .build();
+      var handlerEventIter =
+          WorkflowClientHelper.getHistory(
+              runner.service, runner.config.namespace, handlerExec, runner.config.metricsScope);
+      var handlerEvents =
+          History.newBuilder().addAllEvents(() -> handlerEventIter).build().getEventsList();
+      var handlerStarted =
+          findEvent(handlerEvents, e -> e.hasWorkflowExecutionStartedEventAttributes());
+      assertNotNull(handlerStarted, "expected WorkflowExecutionStarted event in handler history");
+      var attrs = handlerStarted.getWorkflowExecutionStartedEventAttributes();
+      // Cross-check the run ID embedded in the caller's link against the handler's own attrs.
+      assertEquals(attrs.getFirstExecutionRunId(), handlerLink.getRunId());
+      assertTrue(
+          attrs.getCompletionCallbacksCount() > 0,
+          "handler WorkflowExecutionStarted has no completion callbacks");
+      var callerLink =
+          findWorkflowEventLink(
+              attrs.getCompletionCallbacks(0).getLinksList(),
+              EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED);
+      assertNotNull(
+          callerLink,
+          "handler completion callback is missing a link to the caller NexusOperationScheduled event");
+      assertEquals(runner.config.namespace, callerLink.getNamespace());
+      assertEquals(run.execution.getWorkflowId(), callerLink.getWorkflowId());
+      assertEquals(run.execution.getRunId(), callerLink.getRunId());
+      assertEquals(scheduled.getEventId(), callerLink.getEventRef().getEventId());
+    }
+
+    private static HistoryEvent findEvent(
+        List<HistoryEvent> events, java.util.function.Predicate<HistoryEvent> cond) {
+      return events.stream().filter(cond).findFirst().orElse(null);
+    }
+
+    private static Link.WorkflowEvent findWorkflowEventLink(List<Link> links, EventType type) {
+      for (Link l : links) {
+        if (!l.hasWorkflowEvent()) {
+          continue;
+        }
+        var we = l.getWorkflowEvent();
+        if (we.getEventRef().getEventType() == type) {
+          return we;
+        }
+      }
+      return null;
+    }
+  }
+
+  @ServiceImpl(service = TestService.class)
+  class TestServiceImpl {
+    @OperationImpl
+    public OperationHandler<String, String> asyncWorkflowOperation() {
+      return WorkflowRunOperation.fromWorkflowHandle(
+          (context, details, name) -> {
+            WorkflowClient client = Nexus.getOperationContext().getWorkflowClient();
+            return WorkflowHandle.fromWorkflowMethod(
+                client.newWorkflowStub(
+                        HandlerWorkflow.class,
+                        WorkflowOptions.newBuilder().setWorkflowId(details.getRequestId()).build())
+                    ::run,
+                name);
+          });
+    }
+  }
+}

harness/java/io/temporal/sdkfeatures/PreparedFeature.java

@@ -23,6 +23,7 @@ public class PreparedFeature {
           data_converter.json_protobuf.feature.Impl.class,
           eager_activity.non_remote_activities_worker.feature.Impl.class,
           nexus.sync_success.feature.Impl.class,
+          nexus.workflow_run_success.feature.Impl.class,
           query.successful_query.feature.Impl.class,
           query.timeout_due_to_no_active_workers.feature.Impl.class,
           query.unexpected_arguments.feature.Impl.class,