From d5d438ca6ceff48fd526e396006a6a48a19db66f Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 10:08:26 +0400 Subject: [PATCH 1/2] feat(nexus): add async, cancellation, failure and parallel features Four Nexus scenarios implemented in Go and Java: an async workflow-run operation, cancelling a running async operation, a sync operation that fails with an application error, and three parallel sync operations. --- features/features.go | 8 ++ features/nexus/async_cancellation/README.md | 13 ++ features/nexus/async_cancellation/feature.go | 89 ++++++++++++ .../nexus/async_cancellation/feature.java | 136 ++++++++++++++++++ features/nexus/async_success/README.md | 14 ++ features/nexus/async_success/feature.go | 84 +++++++++++ features/nexus/async_success/feature.java | 125 ++++++++++++++++ features/nexus/parallel_operations/README.md | 12 ++ features/nexus/parallel_operations/feature.go | 83 +++++++++++ .../nexus/parallel_operations/feature.java | 99 +++++++++++++ features/nexus/sync_operation_error/README.md | 13 ++ .../nexus/sync_operation_error/feature.go | 93 ++++++++++++ .../nexus/sync_operation_error/feature.java | 107 ++++++++++++++ .../temporal/sdkfeatures/PreparedFeature.java | 4 + 14 files changed, 880 insertions(+) create mode 100644 features/nexus/async_cancellation/README.md create mode 100644 features/nexus/async_cancellation/feature.go create mode 100644 features/nexus/async_cancellation/feature.java create mode 100644 features/nexus/async_success/README.md create mode 100644 features/nexus/async_success/feature.go create mode 100644 features/nexus/async_success/feature.java create mode 100644 features/nexus/parallel_operations/README.md create mode 100644 features/nexus/parallel_operations/feature.go create mode 100644 features/nexus/parallel_operations/feature.java create mode 100644 features/nexus/sync_operation_error/README.md create mode 100644 features/nexus/sync_operation_error/feature.go create mode 100644 features/nexus/sync_operation_error/feature.java diff --git a/features/features.go b/features/features.go index bb8c976f..9829ed7f 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_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, diff --git a/features/nexus/async_cancellation/README.md b/features/nexus/async_cancellation/README.md new file mode 100644 index 00000000..13dd3a79 --- /dev/null +++ b/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. diff --git a/features/nexus/async_cancellation/feature.go b/features/nexus/async_cancellation/feature.go new file mode 100644 index 00000000..b2e63dbe --- /dev/null +++ b/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 + }, +} diff --git a/features/nexus/async_cancellation/feature.java b/features/nexus/async_cancellation/feature.java new file mode 100644 index 00000000..117b45bf --- /dev/null +++ b/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 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..ab2de809 --- /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 "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 + }, +} diff --git a/features/nexus/async_success/feature.java b/features/nexus/async_success/feature.java new file mode 100644 index 00000000..5aef9a61 --- /dev/null +++ b/features/nexus/async_success/feature.java @@ -0,0 +1,125 @@ +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.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(); + var token = execution.getOperationToken().orElse(""); + var result = handle.getResult().get(); + return "token=" + !token.isEmpty() + " result=" + result; + } + + @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("token=true result=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"); + } + } + + @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_operations/README.md b/features/nexus/parallel_operations/README.md new file mode 100644 index 00000000..8c7a20d7 --- /dev/null +++ b/features/nexus/parallel_operations/README.md @@ -0,0 +1,12 @@ +# Nexus 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 and no started + events, since sync operations never enter the started state. diff --git a/features/nexus/parallel_operations/feature.go b/features/nexus/parallel_operations/feature.go new file mode 100644 index 00000000..1315d770 --- /dev/null +++ b/features/nexus/parallel_operations/feature.go @@ -0,0 +1,83 @@ +package parallel_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 +} + +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_SCHEDULED: len(names), + 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) + } + } + return nil + }, +} diff --git a/features/nexus/parallel_operations/feature.java b/features/nexus/parallel_operations/feature.java new file mode 100644 index 00000000..42f8635d --- /dev/null +++ b/features/nexus/parallel_operations/feature.java @@ -0,0 +1,99 @@ +package nexus.parallel_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( + 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"); + } + } + + @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..5ac72365 --- /dev/null +++ b/features/nexus/sync_operation_error/README.md @@ -0,0 +1,13 @@ +# Nexus sync operation fails + +A workflow invokes a synchronous Nexus operation that raises an application failure and +inspects the resulting error. + +# Detailed spec + +- The sync operation returns an operation error in the failed state whose cause is an + application error with a known type and message. +- The caller receives a Nexus operation error whose cause is that application error, with the + original type and message preserved across the operation boundary. +- 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..676dc6f3 --- /dev/null +++ b/features/nexus/sync_operation_error/feature.go @@ -0,0 +1,93 @@ +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) + } + var operationErr *temporal.ApplicationError + if !errors.As(opErr.Unwrap(), &operationErr) { + return "", harness.AppErrorf("expected an application error cause, got %v", opErr.Unwrap()) + } + var appErr *temporal.ApplicationError + if !errors.As(operationErr.Unwrap(), &appErr) { + return "", harness.AppErrorf("expected the original application error, got %v", operationErr.Unwrap()) + } + return appErr.Type() + ": " + appErr.Message(), 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 nil + }, +} diff --git a/features/nexus/sync_operation_error/feature.java b/features/nexus/sync_operation_error/feature.java new file mode 100644 index 00000000..3849640d --- /dev/null +++ b/features/nexus/sync_operation_error/feature.java @@ -0,0 +1,107 @@ +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.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 { + @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) { + Throwable cause = e.getCause(); + while (cause != null && !(cause instanceof ApplicationFailure)) { + cause = cause.getCause(); + } + var applicationFailure = (ApplicationFailure) cause; + return "caught " + + applicationFailure.getType() + + ": " + + applicationFailure.getOriginalMessage(); + } + } + + @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("caught TestError: deliberate failure", 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"); + } + } + + @ServiceImpl(service = TestService.class) + class TestServiceImpl { + @OperationImpl + public OperationHandler failingOperation() { + return OperationHandler.sync( + (context, details, name) -> { + throw ApplicationFailure.newNonRetryableFailure("deliberate failure", "TestError"); + }); + } + } +} diff --git a/harness/java/io/temporal/sdkfeatures/PreparedFeature.java b/harness/java/io/temporal/sdkfeatures/PreparedFeature.java index 93066a73..a4bbe6f0 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_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, From 8e0324e37cdd46f1825d565aabd54130caeea4b1 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 10:26:10 +0400 Subject: [PATCH 2/2] fix(nexus): raise an operation error in the java failure feature The java handler threw a bare application failure, which the SDK reports as a handler error, so the two languages were asserting different contracts. --- features/nexus/sync_operation_error/feature.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/features/nexus/sync_operation_error/feature.java b/features/nexus/sync_operation_error/feature.java index 3849640d..959689bf 100644 --- a/features/nexus/sync_operation_error/feature.java +++ b/features/nexus/sync_operation_error/feature.java @@ -5,6 +5,7 @@ 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; @@ -50,7 +51,7 @@ public String workflow(String endpoint) { return "no error"; } catch (NexusOperationFailure e) { Throwable cause = e.getCause(); - while (cause != null && !(cause instanceof ApplicationFailure)) { + while (cause != null && cause.getCause() != null) { cause = cause.getCause(); } var applicationFailure = (ApplicationFailure) cause; @@ -100,7 +101,8 @@ class TestServiceImpl { public OperationHandler failingOperation() { return OperationHandler.sync( (context, details, name) -> { - throw ApplicationFailure.newNonRetryableFailure("deliberate failure", "TestError"); + throw OperationException.failure( + ApplicationFailure.newNonRetryableFailure("deliberate failure", "TestError")); }); } }