diff --git a/features/features.go b/features/features.go index bb8c976f..113be5ba 100644 --- a/features/features.go +++ b/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_sync_operations "github.com/temporalio/features/features/nexus/parallel_sync_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_sync_operations.Feature, + nexus_sync_operation_error.Feature, nexus_sync_success.Feature, query_successful_query.Feature, query_timeout_due_to_no_active_workers.Feature, diff --git a/features/nexus/async_cancellation/README.md b/features/nexus/async_cancellation/README.md new file mode 100644 index 00000000..248bf9bd --- /dev/null +++ b/features/nexus/async_cancellation/README.md @@ -0,0 +1,15 @@ +# 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. +- The history records both the cancellation request and the resulting cancellation of the + operation. diff --git a/features/nexus/async_cancellation/feature.go b/features/nexus/async_cancellation/feature.go new file mode 100644 index 00000000..5542cad7 --- /dev/null +++ b/features/nexus/async_cancellation/feature.go @@ -0,0 +1,92 @@ +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 canceled 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 { + for _, t := range []enumspb.EventType{ + enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED, + enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, + } { + 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 runner.CheckHistoryDefault(ctx, run) + }, +} diff --git a/features/nexus/async_cancellation/feature.java b/features/nexus/async_cancellation/feature.java new file mode 100644 index 00000000..04b1cf5c --- /dev/null +++ b/features/nexus/async_cancellation/feature.java @@ -0,0 +1,151 @@ +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.CompletablePromise; +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); + + CompletablePromise outcome = Workflow.newPromise(); + var scope = + Workflow.newCancellationScope( + () -> { + var handle = Workflow.startNexusOperation(stub::blockingOperation, "world"); + handle.getExecution().get(); + handle + .getResult() + .handle( + (value, failure) -> { + outcome.complete(describeOutcome(failure)); + return null; + }); + }); + scope.run(); + scope.cancel(); + return outcome.get(); + } + + private static String describeOutcome(RuntimeException failure) { + if (failure == null) { + return "completed"; + } + for (Throwable cause = failure; cause != null; cause = cause.getCause()) { + if (cause instanceof CanceledFailure) { + return "canceled"; + } + } + return "failed with " + failure; + } + + @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("canceled", 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"); + assertTrue( + events.stream().anyMatch(e -> e.hasNexusOperationCanceledEventAttributes()), + "expected NexusOperationCanceled event in history"); + runner.checkCurrentAndPastHistories(run); + } + } + + @ServiceImpl(service = TestService.class) + class TestServiceImpl { + @OperationImpl + public OperationHandler blockingOperation() { + return WorkflowRunOperation.fromWorkflowMethod( + (context, details, name) -> + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + HandlerWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(details.getRequestId()) + .build()) + ::handlerWorkflow); + } + } +} diff --git a/features/nexus/async_success/README.md b/features/nexus/async_success/README.md new file mode 100644 index 00000000..dfa5eb54 --- /dev/null +++ b/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. diff --git a/features/nexus/async_success/feature.go b/features/nexus/async_success/feature.go new file mode 100644 index 00000000..e8e391e5 --- /dev/null +++ b/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 result, nil +} + +var Feature = harness.Feature{ + Workflows: []interface{}{Workflow, 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, 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 runner.CheckHistoryDefault(ctx, run) + }, +} diff --git a/features/nexus/async_success/feature.java b/features/nexus/async_success/feature.java new file mode 100644 index 00000000..b533999d --- /dev/null +++ b/features/nexus/async_success/feature.java @@ -0,0 +1,129 @@ +package nexus.async_success; + +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.ApplicationFailure; +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.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 asyncOperation(String name); + } + + @WorkflowInterface + interface HandlerWorkflow { + @WorkflowMethod + String handlerWorkflow(String name); + } + + class HandlerWorkflowImpl implements HandlerWorkflow { + @Override + public String handlerWorkflow(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); + var handle = Workflow.startNexusOperation(stub::asyncOperation, "world"); + var execution = handle.getExecution().get(); + if (execution.getOperationToken().orElse("").isEmpty()) { + throw ApplicationFailure.newNonRetryableFailure( + "expected a non-empty operation token", "AssertionFailure"); + } + return handle.getResult().get(); + } + + @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 { + var events = runner.getWorkflowHistory(run).getEventsList(); + assertTrue( + events.stream().anyMatch(e -> e.hasNexusOperationScheduledEventAttributes()), + "expected NexusOperationScheduled event in history"); + assertTrue( + events.stream().anyMatch(e -> e.hasNexusOperationStartedEventAttributes()), + "expected NexusOperationStarted event in history"); + assertTrue( + events.stream().anyMatch(e -> e.hasNexusOperationCompletedEventAttributes()), + "expected NexusOperationCompleted event in history"); + runner.checkCurrentAndPastHistories(run); + } + } + + @ServiceImpl(service = TestService.class) + class TestServiceImpl { + @OperationImpl + public OperationHandler asyncOperation() { + return WorkflowRunOperation.fromWorkflowMethod( + (context, details, name) -> + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + HandlerWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(details.getRequestId()) + .build()) + ::handlerWorkflow); + } + } +} diff --git a/features/nexus/parallel_sync_operations/README.md b/features/nexus/parallel_sync_operations/README.md new file mode 100644 index 00000000..a04690db --- /dev/null +++ b/features/nexus/parallel_sync_operations/README.md @@ -0,0 +1,13 @@ +# Nexus sync operations run in parallel + +A workflow starts three synchronous Nexus operations in a single workflow task and awaits all +of their results. + +# Detailed spec + +- All three operations are started before any of them is awaited, so they are scheduled in the + same workflow task. +- The caller awaits the operation futures in order and joins their results. +- The history contains one scheduled and one completed event per operation, all three + scheduled by the same workflow task, and no started events, since sync operations never + enter the started state. diff --git a/features/nexus/parallel_sync_operations/feature.go b/features/nexus/parallel_sync_operations/feature.go new file mode 100644 index 00000000..c20866de --- /dev/null +++ b/features/nexus/parallel_sync_operations/feature.go @@ -0,0 +1,109 @@ +package parallel_sync_operations + +import ( + "context" + "fmt" + "strings" + "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/workflow" +) + +const ServiceName = "test-service" + +var SyncOperation = nexus.NewSyncOperation( + "say-hello", + func(ctx context.Context, name string, options nexus.StartOperationOptions) (string, error) { + return "Hello, " + name + "!", nil + }, +) + +var Service = func() *nexus.Service { + s := nexus.NewService(ServiceName) + s.MustRegister(SyncOperation) + return s +}() + +var names = []string{"one", "two", "three"} + +func Workflow(ctx workflow.Context, endpoint string) (string, error) { + nc := workflow.NewNexusClient(endpoint, ServiceName) + futures := make([]workflow.NexusOperationFuture, len(names)) + for i, name := range names { + futures[i] = nc.ExecuteOperation(ctx, SyncOperation, name, workflow.NexusOperationOptions{ + ScheduleToCloseTimeout: time.Minute, + }) + } + results := make([]string, len(futures)) + for i, fut := range futures { + if err := fut.Get(ctx, &results[i]); err != nil { + return "", err + } + } + return strings.Join(results, " "), nil +} + +func scheduledWorkflowTaskIDs(hist client.HistoryEventIterator) ([]int64, error) { + var taskIDs []int64 + for hist.HasNext() { + ev, err := hist.Next() + if err != nil { + return nil, err + } + if attrs := ev.GetNexusOperationScheduledEventAttributes(); attrs != nil { + taskIDs = append(taskIDs, attrs.WorkflowTaskCompletedEventId) + } + } + return taskIDs, nil +} + +var Feature = harness.Feature{ + Workflows: Workflow, + NexusServices: Service, + ExpectRunResult: "Hello, one! Hello, two! Hello, three!", + 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 { + countEvents := func(t enumspb.EventType) (int, error) { + hist := runner.Client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) + return harness.CountEvents(hist, func(ev *historypb.HistoryEvent) bool { return ev.EventType == t }) + } + expected := map[enumspb.EventType]int{ + enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED: len(names), + enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED: 0, + } + for t, want := range expected { + got, err := countEvents(t) + if err != nil { + return err + } + if got != want { + return fmt.Errorf("expected %v %v events, got %v", want, t, got) + } + } + hist := runner.Client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) + scheduledBy, err := scheduledWorkflowTaskIDs(hist) + if err != nil { + return err + } + if len(scheduledBy) != len(names) { + return fmt.Errorf("expected %v scheduled operations, got %v", len(names), len(scheduledBy)) + } + for _, id := range scheduledBy { + if id != scheduledBy[0] { + return fmt.Errorf("expected all operations to be scheduled by a single workflow task, got tasks %v", scheduledBy) + } + } + return runner.CheckHistoryDefault(ctx, run) + }, +} diff --git a/features/nexus/parallel_sync_operations/feature.java b/features/nexus/parallel_sync_operations/feature.java new file mode 100644 index 00000000..c93b4a2b --- /dev/null +++ b/features/nexus/parallel_sync_operations/feature.java @@ -0,0 +1,111 @@ +package nexus.parallel_sync_operations; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; + +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.sdkfeatures.Feature; +import io.temporal.sdkfeatures.Run; +import io.temporal.sdkfeatures.Runner; +import io.temporal.workflow.Async; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Promise; +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 syncOperation(String 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); + Promise one = Async.function(stub::syncOperation, "one"); + Promise two = Async.function(stub::syncOperation, "two"); + Promise three = Async.function(stub::syncOperation, "three"); + Promise.allOf(one, two, three).get(); + return one.get() + " " + two.get() + " " + three.get(); + } + + @Override + public Object[] nexusServiceImplementations() { + return new Object[] {new TestServiceImpl()}; + } + + @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, one! Hello, two! Hello, three!", result); + } + + @Override + public void checkHistory(Runner runner, Run run) throws Exception { + var events = runner.getWorkflowHistory(run).getEventsList(); + assertEquals( + 3, + events.stream().filter(e -> e.hasNexusOperationScheduledEventAttributes()).count(), + "expected three NexusOperationScheduled events in history"); + assertEquals( + 1, + events.stream() + .filter(e -> e.hasNexusOperationScheduledEventAttributes()) + .map( + e -> + e.getNexusOperationScheduledEventAttributes() + .getWorkflowTaskCompletedEventId()) + .distinct() + .count(), + "expected all operations to be scheduled by a single workflow task"); + assertEquals( + 3, + events.stream().filter(e -> e.hasNexusOperationCompletedEventAttributes()).count(), + "expected three NexusOperationCompleted events in history"); + assertFalse( + events.stream().anyMatch(e -> e.hasNexusOperationStartedEventAttributes()), + "unexpected NexusOperationStarted event for sync operations"); + runner.checkCurrentAndPastHistories(run); + } + } + + @ServiceImpl(service = TestService.class) + class TestServiceImpl { + @OperationImpl + public OperationHandler syncOperation() { + return OperationHandler.sync((context, details, name) -> "Hello, " + name + "!"); + } + } +} diff --git a/features/nexus/sync_operation_error/README.md b/features/nexus/sync_operation_error/README.md new file mode 100644 index 00000000..6b6e4a11 --- /dev/null +++ b/features/nexus/sync_operation_error/README.md @@ -0,0 +1,14 @@ +# Nexus sync operation fails + +A workflow invokes a synchronous Nexus operation that fails with an operation error and +inspects the resulting error. + +# Detailed spec + +- The handler fails the operation by raising an operation error in the failed state, carrying + an application error with a known type and message as its cause. Raising an application + error directly is reported as a handler error instead, which this feature does not cover. +- The caller receives a Nexus operation error, and the application error raised by the handler + is present in its cause chain with the original type and message preserved. +- The caller handles the failure and completes successfully. +- A failed operation produces a failed event and never a completed one. diff --git a/features/nexus/sync_operation_error/feature.go b/features/nexus/sync_operation_error/feature.go new file mode 100644 index 00000000..bbaf0960 --- /dev/null +++ b/features/nexus/sync_operation_error/feature.go @@ -0,0 +1,98 @@ +package sync_operation_error + +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/workflow" +) + +const ( + ServiceName = "test-service" + ErrorType = "TestFailure" + ErrorMessage = "deliberate failure" +) + +var FailingOperation = nexus.NewSyncOperation( + "fail", + func(ctx context.Context, name string, options nexus.StartOperationOptions) (string, error) { + return "", &nexus.OperationError{ + State: nexus.OperationStateFailed, + Cause: temporal.NewApplicationError(ErrorMessage, ErrorType), + } + }, +) + +var Service = func() *nexus.Service { + s := nexus.NewService(ServiceName) + s.MustRegister(FailingOperation) + return s +}() + +func Workflow(ctx workflow.Context, endpoint string) (string, error) { + nc := workflow.NewNexusClient(endpoint, ServiceName) + fut := nc.ExecuteOperation(ctx, FailingOperation, "world", workflow.NexusOperationOptions{ + ScheduleToCloseTimeout: time.Minute, + }) + err := fut.Get(ctx, nil) + if err == nil { + return "", harness.AppErrorf("expected the operation to fail") + } + var opErr *temporal.NexusOperationError + if !errors.As(err, &opErr) { + return "", harness.AppErrorf("expected a nexus operation error, got %v", err) + } + appErr := findApplicationError(err, ErrorType) + if appErr == nil { + return "", harness.AppErrorf("expected an application error of type %v in the cause chain, got %v", ErrorType, err) + } + return appErr.Type() + ": " + appErr.Message(), nil +} + +func findApplicationError(err error, errType string) *temporal.ApplicationError { + for ; err != nil; err = errors.Unwrap(err) { + if appErr, ok := err.(*temporal.ApplicationError); ok && appErr.Type() == errType { + return appErr + } + } + return nil +} + +var Feature = harness.Feature{ + Workflows: Workflow, + NexusServices: Service, + ExpectRunResult: ErrorType + ": " + ErrorMessage, + 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 { + hasEvent := func(t enumspb.EventType) (bool, 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 == t }) + return ev != nil, err + } + if ok, err := hasEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED); err != nil { + return err + } else if !ok { + return fmt.Errorf("did not find NexusOperationFailed event in history") + } + if ok, err := hasEvent(enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED); err != nil { + return err + } else if ok { + return fmt.Errorf("unexpected NexusOperationCompleted event for failed operation") + } + return runner.CheckHistoryDefault(ctx, run) + }, +} diff --git a/features/nexus/sync_operation_error/feature.java b/features/nexus/sync_operation_error/feature.java new file mode 100644 index 00000000..8f0c04d3 --- /dev/null +++ b/features/nexus/sync_operation_error/feature.java @@ -0,0 +1,124 @@ +package nexus.sync_operation_error; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.nexusrpc.Operation; +import io.nexusrpc.OperationException; +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.ApplicationFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.sdkfeatures.Feature; +import io.temporal.sdkfeatures.Run; +import io.temporal.sdkfeatures.Runner; +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 { + String ERROR_TYPE = "TestFailure"; + String ERROR_MESSAGE = "deliberate failure"; + + @WorkflowMethod + String workflow(String endpoint); + + @Service + interface TestService { + @Operation + String failingOperation(String 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); + try { + stub.failingOperation("world"); + return "no error"; + } catch (NexusOperationFailure e) { + var applicationFailure = findApplicationFailure(e, ERROR_TYPE); + if (applicationFailure == null) { + throw ApplicationFailure.newNonRetryableFailure( + "expected an application failure of type " + + ERROR_TYPE + + " in the cause chain, got " + + e, + "AssertionFailure"); + } + return applicationFailure.getType() + ": " + applicationFailure.getOriginalMessage(); + } + } + + private static ApplicationFailure findApplicationFailure(Throwable failure, String type) { + for (Throwable cause = failure; cause != null; cause = cause.getCause()) { + if (cause instanceof ApplicationFailure + && type.equals(((ApplicationFailure) cause).getType())) { + return (ApplicationFailure) cause; + } + } + return null; + } + + @Override + public Object[] nexusServiceImplementations() { + return new Object[] {new TestServiceImpl()}; + } + + @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(ERROR_TYPE + ": " + ERROR_MESSAGE, result); + } + + @Override + public void checkHistory(Runner runner, Run run) throws Exception { + var events = runner.getWorkflowHistory(run).getEventsList(); + assertTrue( + events.stream().anyMatch(e -> e.hasNexusOperationFailedEventAttributes()), + "expected NexusOperationFailed event in history"); + assertFalse( + events.stream().anyMatch(e -> e.hasNexusOperationCompletedEventAttributes()), + "unexpected NexusOperationCompleted event in history"); + runner.checkCurrentAndPastHistories(run); + } + } + + @ServiceImpl(service = TestService.class) + class TestServiceImpl { + @OperationImpl + public OperationHandler failingOperation() { + return OperationHandler.sync( + (context, details, name) -> { + throw OperationException.failure( + ApplicationFailure.newFailure(ERROR_MESSAGE, ERROR_TYPE)); + }); + } + } +} diff --git a/features/nexus/sync_success/feature.go b/features/nexus/sync_success/feature.go index 31b34abc..d97b80b3 100644 --- a/features/nexus/sync_success/feature.go +++ b/features/nexus/sync_success/feature.go @@ -74,6 +74,6 @@ var Feature = harness.Feature{ } else if ok { return fmt.Errorf("unexpected NexusOperationStarted event for sync operation") } - return nil + return runner.CheckHistoryDefault(ctx, run) }, } diff --git a/features/nexus/sync_success/feature.java b/features/nexus/sync_success/feature.java index 497fa415..90c5002c 100644 --- a/features/nexus/sync_success/feature.java +++ b/features/nexus/sync_success/feature.java @@ -81,6 +81,7 @@ public void checkHistory(Runner runner, Run run) throws Exception { assertFalse( events.stream().anyMatch(e -> e.hasNexusOperationStartedEventAttributes()), "unexpected NexusOperationStarted event for sync operation"); + runner.checkCurrentAndPastHistories(run); } } diff --git a/harness/java/io/temporal/sdkfeatures/PreparedFeature.java b/harness/java/io/temporal/sdkfeatures/PreparedFeature.java index 93066a73..662b5b62 100644 --- a/harness/java/io/temporal/sdkfeatures/PreparedFeature.java +++ b/harness/java/io/temporal/sdkfeatures/PreparedFeature.java @@ -22,6 +22,10 @@ public class PreparedFeature { data_converter.json.feature.Impl.class, data_converter.json_protobuf.feature.Impl.class, eager_activity.non_remote_activities_worker.feature.Impl.class, + nexus.async_cancellation.feature.Impl.class, + nexus.async_success.feature.Impl.class, + nexus.parallel_sync_operations.feature.Impl.class, + nexus.sync_operation_error.feature.Impl.class, nexus.sync_success.feature.Impl.class, query.successful_query.feature.Impl.class, query.timeout_due_to_no_active_workers.feature.Impl.class,