From d5d438ca6ceff48fd526e396006a6a48a19db66f Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Wed, 19 Aug 2026 10:08:26 +0400 Subject: [PATCH 1/3] 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/3] 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")); }); } } From 9c3ba0991ef8b3520b3d0442d6fce28994292413 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Sat, 22 Aug 2026 11:34:30 +0400 Subject: [PATCH 3/3] fix(nexus): address review feedback on the nexus features Keep the default history replay: the CheckHistory and checkHistory overrides now call CheckHistoryDefault and checkCurrentAndPastHistories, which the harness skips entirely when a feature supplies its own checker. Assert the contracts the features claim instead of proxies for them: the java caller looks for CanceledFailure in the cause chain rather than treating any failure as cancellation, both languages assert the NexusOperationCanceled event, and the parallel feature asserts that all three operations are scheduled by a single workflow task. Look up the application error by type rather than by cause-chain depth in both languages, and fail with a diagnostic when it is missing. Raise a retryable application error from the java handler so it matches the go one, and share the type and message through constants. Rename parallel_operations to parallel_sync_operations, align the workflow results and error type between go and java, use CompletablePromise instead of a mutable array in the java caller, and correct the sync_operation_error spec: the handler raises an operation error that carries an application error as its cause, not an application error directly. nexus/sync_success gets the same replay fix so the whole nexus family behaves alike; it is the only change outside the original diff of this PR. --- features/features.go | 4 +-- features/nexus/async_cancellation/README.md | 2 ++ features/nexus/async_cancellation/feature.go | 25 +++++++------ .../nexus/async_cancellation/feature.java | 27 ++++++++++---- features/nexus/async_success/feature.go | 6 ++-- features/nexus/async_success/feature.java | 12 ++++--- .../README.md | 7 ++-- .../feature.go | 32 +++++++++++++++-- .../feature.java | 18 ++++++++-- features/nexus/sync_operation_error/README.md | 11 +++--- .../nexus/sync_operation_error/feature.go | 21 ++++++----- .../nexus/sync_operation_error/feature.java | 35 +++++++++++++------ features/nexus/sync_success/feature.go | 2 +- features/nexus/sync_success/feature.java | 1 + .../temporal/sdkfeatures/PreparedFeature.java | 2 +- 15 files changed, 145 insertions(+), 60 deletions(-) rename features/nexus/{parallel_operations => parallel_sync_operations}/README.md (68%) rename features/nexus/{parallel_operations => parallel_sync_operations}/feature.go (71%) rename features/nexus/{parallel_operations => parallel_sync_operations}/feature.java (83%) diff --git a/features/features.go b/features/features.go index 9829ed7f..113be5ba 100644 --- a/features/features.go +++ b/features/features.go @@ -34,7 +34,7 @@ import ( 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_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" @@ -101,7 +101,7 @@ func init() { eager_workflow_successful_start.Feature, nexus_async_cancellation.Feature, nexus_async_success.Feature, - nexus_parallel_operations.Feature, + nexus_parallel_sync_operations.Feature, nexus_sync_operation_error.Feature, nexus_sync_success.Feature, query_successful_query.Feature, diff --git a/features/nexus/async_cancellation/README.md b/features/nexus/async_cancellation/README.md index 13dd3a79..248bf9bd 100644 --- a/features/nexus/async_cancellation/README.md +++ b/features/nexus/async_cancellation/README.md @@ -11,3 +11,5 @@ A workflow cancels a running asynchronous Nexus operation and observes a cancell - 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 index b2e63dbe..5542cad7 100644 --- a/features/nexus/async_cancellation/feature.go +++ b/features/nexus/async_cancellation/feature.go @@ -53,7 +53,7 @@ func Workflow(ctx workflow.Context, endpoint string) (string, error) { err := fut.Get(ctx, nil) if err == nil { - return "", harness.AppErrorf("expected the cancelled operation to fail") + return "", harness.AppErrorf("expected the canceled operation to fail") } var canceledErr *temporal.CanceledError if !errors.As(err, &canceledErr) { @@ -74,16 +74,19 @@ var Feature = harness.Feature{ 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 + 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) + } } - if ev == nil { - return fmt.Errorf("did not find NexusOperationCancelRequested event in history") - } - return nil + return runner.CheckHistoryDefault(ctx, run) }, } diff --git a/features/nexus/async_cancellation/feature.java b/features/nexus/async_cancellation/feature.java index 117b45bf..04b1cf5c 100644 --- a/features/nexus/async_cancellation/feature.java +++ b/features/nexus/async_cancellation/feature.java @@ -16,7 +16,7 @@ import io.temporal.sdkfeatures.Run; import io.temporal.sdkfeatures.Runner; import io.temporal.worker.Worker; -import io.temporal.workflow.CancellationScope; +import io.temporal.workflow.CompletablePromise; import io.temporal.workflow.NexusOperationOptions; import io.temporal.workflow.NexusServiceOptions; import io.temporal.workflow.Workflow; @@ -62,7 +62,7 @@ public String workflow(String endpoint) { .build(); TestService stub = Workflow.newNexusServiceStub(TestService.class, serviceOptions); - var resultHolder = new String[1]; + CompletablePromise outcome = Workflow.newPromise(); var scope = Workflow.newCancellationScope( () -> { @@ -72,14 +72,25 @@ public String workflow(String endpoint) { .getResult() .handle( (value, failure) -> { - resultHolder[0] = failure == null ? "completed" : "cancelled"; + outcome.complete(describeOutcome(failure)); return null; }); }); scope.run(); scope.cancel(); - Workflow.await(() -> resultHolder[0] != null); - return "operation " + resultHolder[0]; + 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 @@ -105,7 +116,7 @@ public Run execute(Runner runner) throws Exception { @Override public void checkResult(Runner runner, Run run) { var result = runner.waitForRunResult(run, String.class); - assertEquals("operation cancelled", result); + assertEquals("canceled", result); } @Override @@ -114,6 +125,10 @@ public void checkHistory(Runner runner, Run run) throws Exception { 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); } } diff --git a/features/nexus/async_success/feature.go b/features/nexus/async_success/feature.go index ab2de809..e8e391e5 100644 --- a/features/nexus/async_success/feature.go +++ b/features/nexus/async_success/feature.go @@ -50,13 +50,13 @@ func Workflow(ctx workflow.Context, endpoint string) (string, error) { if err := fut.Get(ctx, &result); err != nil { return "", err } - return "token+" + result, nil + return result, nil } var Feature = harness.Feature{ Workflows: []interface{}{Workflow, HandlerWorkflow}, NexusServices: Service, - ExpectRunResult: "token+Hello, world!", + ExpectRunResult: "Hello, world!", Execute: func(ctx context.Context, runner *harness.Runner) (client.WorkflowRun, error) { opts := client.StartWorkflowOptions{ TaskQueue: runner.TaskQueue, @@ -79,6 +79,6 @@ var Feature = harness.Feature{ return fmt.Errorf("did not find %v event in history", t) } } - return nil + return runner.CheckHistoryDefault(ctx, run) }, } diff --git a/features/nexus/async_success/feature.java b/features/nexus/async_success/feature.java index 5aef9a61..b533999d 100644 --- a/features/nexus/async_success/feature.java +++ b/features/nexus/async_success/feature.java @@ -9,6 +9,7 @@ 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; @@ -60,9 +61,11 @@ public String workflow(String endpoint) { 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; + if (execution.getOperationToken().orElse("").isEmpty()) { + throw ApplicationFailure.newNonRetryableFailure( + "expected a non-empty operation token", "AssertionFailure"); + } + return handle.getResult().get(); } @Override @@ -88,7 +91,7 @@ public Run execute(Runner runner) throws Exception { @Override public void checkResult(Runner runner, Run run) { var result = runner.waitForRunResult(run, String.class); - assertEquals("token=true result=Hello, world!", result); + assertEquals("Hello, world!", result); } @Override @@ -103,6 +106,7 @@ public void checkHistory(Runner runner, Run run) throws Exception { assertTrue( events.stream().anyMatch(e -> e.hasNexusOperationCompletedEventAttributes()), "expected NexusOperationCompleted event in history"); + runner.checkCurrentAndPastHistories(run); } } diff --git a/features/nexus/parallel_operations/README.md b/features/nexus/parallel_sync_operations/README.md similarity index 68% rename from features/nexus/parallel_operations/README.md rename to features/nexus/parallel_sync_operations/README.md index 8c7a20d7..a04690db 100644 --- a/features/nexus/parallel_operations/README.md +++ b/features/nexus/parallel_sync_operations/README.md @@ -1,4 +1,4 @@ -# Nexus operations run in parallel +# Nexus sync operations run in parallel A workflow starts three synchronous Nexus operations in a single workflow task and awaits all of their results. @@ -8,5 +8,6 @@ of their results. - 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. +- 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_operations/feature.go b/features/nexus/parallel_sync_operations/feature.go similarity index 71% rename from features/nexus/parallel_operations/feature.go rename to features/nexus/parallel_sync_operations/feature.go index 1315d770..c20866de 100644 --- a/features/nexus/parallel_operations/feature.go +++ b/features/nexus/parallel_sync_operations/feature.go @@ -1,4 +1,4 @@ -package parallel_operations +package parallel_sync_operations import ( "context" @@ -48,6 +48,20 @@ func Workflow(ctx workflow.Context, endpoint string) (string, error) { 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, @@ -65,7 +79,6 @@ var Feature = harness.Feature{ 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, } @@ -78,6 +91,19 @@ var Feature = harness.Feature{ return fmt.Errorf("expected %v %v events, got %v", want, t, got) } } - return nil + 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_operations/feature.java b/features/nexus/parallel_sync_operations/feature.java similarity index 83% rename from features/nexus/parallel_operations/feature.java rename to features/nexus/parallel_sync_operations/feature.java index 42f8635d..c93b4a2b 100644 --- a/features/nexus/parallel_operations/feature.java +++ b/features/nexus/parallel_sync_operations/feature.java @@ -1,4 +1,4 @@ -package nexus.parallel_operations; +package nexus.parallel_sync_operations; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -48,7 +48,7 @@ public String workflow(String endpoint) { 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(); + return one.get() + " " + two.get() + " " + three.get(); } @Override @@ -69,7 +69,7 @@ public Run execute(Runner runner) throws Exception { @Override public void checkResult(Runner runner, Run run) { var result = runner.waitForRunResult(run, String.class); - assertEquals("Hello, one!, Hello, two!, Hello, three!", result); + assertEquals("Hello, one! Hello, two! Hello, three!", result); } @Override @@ -79,6 +79,17 @@ public void checkHistory(Runner runner, Run run) throws Exception { 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(), @@ -86,6 +97,7 @@ public void checkHistory(Runner runner, Run run) throws Exception { assertFalse( events.stream().anyMatch(e -> e.hasNexusOperationStartedEventAttributes()), "unexpected NexusOperationStarted event for sync operations"); + runner.checkCurrentAndPastHistories(run); } } diff --git a/features/nexus/sync_operation_error/README.md b/features/nexus/sync_operation_error/README.md index 5ac72365..6b6e4a11 100644 --- a/features/nexus/sync_operation_error/README.md +++ b/features/nexus/sync_operation_error/README.md @@ -1,13 +1,14 @@ # Nexus sync operation fails -A workflow invokes a synchronous Nexus operation that raises an application failure and +A workflow invokes a synchronous Nexus operation that fails with an operation error 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 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 index 676dc6f3..bbaf0960 100644 --- a/features/nexus/sync_operation_error/feature.go +++ b/features/nexus/sync_operation_error/feature.go @@ -50,17 +50,22 @@ func Workflow(ctx workflow.Context, endpoint string) (string, error) { 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()) + 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, @@ -88,6 +93,6 @@ var Feature = harness.Feature{ } else if ok { return fmt.Errorf("unexpected NexusOperationCompleted event for failed operation") } - return nil + return runner.CheckHistoryDefault(ctx, run) }, } diff --git a/features/nexus/sync_operation_error/feature.java b/features/nexus/sync_operation_error/feature.java index 959689bf..8f0c04d3 100644 --- a/features/nexus/sync_operation_error/feature.java +++ b/features/nexus/sync_operation_error/feature.java @@ -25,6 +25,9 @@ @WorkflowInterface public interface feature extends Feature { + String ERROR_TYPE = "TestFailure"; + String ERROR_MESSAGE = "deliberate failure"; + @WorkflowMethod String workflow(String endpoint); @@ -50,16 +53,27 @@ public String workflow(String endpoint) { stub.failingOperation("world"); return "no error"; } catch (NexusOperationFailure e) { - Throwable cause = e.getCause(); - while (cause != null && cause.getCause() != null) { - cause = cause.getCause(); + 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; } - var applicationFailure = (ApplicationFailure) cause; - return "caught " - + applicationFailure.getType() - + ": " - + applicationFailure.getOriginalMessage(); } + return null; } @Override @@ -80,7 +94,7 @@ public Run execute(Runner runner) throws Exception { @Override public void checkResult(Runner runner, Run run) { var result = runner.waitForRunResult(run, String.class); - assertEquals("caught TestError: deliberate failure", result); + assertEquals(ERROR_TYPE + ": " + ERROR_MESSAGE, result); } @Override @@ -92,6 +106,7 @@ public void checkHistory(Runner runner, Run run) throws Exception { assertFalse( events.stream().anyMatch(e -> e.hasNexusOperationCompletedEventAttributes()), "unexpected NexusOperationCompleted event in history"); + runner.checkCurrentAndPastHistories(run); } } @@ -102,7 +117,7 @@ public OperationHandler failingOperation() { return OperationHandler.sync( (context, details, name) -> { throw OperationException.failure( - ApplicationFailure.newNonRetryableFailure("deliberate failure", "TestError")); + 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 a4bbe6f0..662b5b62 100644 --- a/harness/java/io/temporal/sdkfeatures/PreparedFeature.java +++ b/harness/java/io/temporal/sdkfeatures/PreparedFeature.java @@ -24,7 +24,7 @@ public class PreparedFeature { 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.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,