From 63e7d29d752d6b58a98af345cf37ccdcf25fca62 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 8 Apr 2026 09:44:52 -0700 Subject: [PATCH 01/12] Add support for Stand Alone Nexus Operations --- client/client.go | 110 ++++ interceptor/interceptor.go | 48 ++ internal/client.go | 24 + internal/interceptor.go | 125 ++++ internal/interceptor_base.go | 59 ++ internal/internal_nexus_client.go | 865 +++++++++++++++++++++++++ internal/internal_nexus_client_test.go | 90 +++ internal/nexus_operations.go | 16 + mocks/Client.go | 98 +++ test/integration_test.go | 195 ++++++ 10 files changed, 1630 insertions(+) create mode 100644 internal/internal_nexus_client.go create mode 100644 internal/internal_nexus_client_test.go diff --git a/client/client.go b/client/client.go index 1b1749aa1..0c39aed5f 100644 --- a/client/client.go +++ b/client/client.go @@ -990,6 +990,92 @@ type ( // NOTE: Experimental TerminateActivityOptions = internal.ClientTerminateActivityOptions + // StartNexusOperationOptions contains configuration parameters for starting a Nexus operation execution. + // + // NOTE: Experimental + StartNexusOperationOptions = internal.ClientStartNexusOperationOptions + + // NexusClientOptions contains options for creating a NexusClient. + // + // NOTE: Experimental + NexusClientOptions = internal.ClientNexusClientOptions + + // NexusClient is the client for starting Nexus operations bound to a specific endpoint and service. + // This is for standalone Nexus operations outside of workflow context. + // For Nexus operations within workflows, use workflow.NexusClient. + // + // NOTE: Experimental + NexusClient = internal.ClientNexusClient + + // NexusOperationHandle represents a running or completed standalone Nexus operation execution. + // It can be used to get the result, describe, cancel, or terminate the operation. + // + // NOTE: Experimental + NexusOperationHandle = internal.ClientNexusOperationHandle + + // NexusOperationMetadata contains information about a Nexus operation execution. + // This is returned by ListNexusOperations and embedded in NexusOperationExecutionDescription. + // + // NOTE: Experimental + NexusOperationMetadata = internal.ClientNexusOperationMetadata + + // NexusOperationExecutionDescription contains detailed information about a Nexus operation execution. + // This is returned by NexusOperationHandle.Describe. + // + // NOTE: Experimental + NexusOperationExecutionDescription = internal.ClientNexusOperationExecutionDescription + + // NexusOperationCancellationInfo contains cancellation information for a Nexus operation. + // + // NOTE: Experimental + NexusOperationCancellationInfo = internal.ClientNexusOperationCancellationInfo + + // DescribeNexusOperationOptions contains options for NexusOperationHandle.Describe call. + // + // NOTE: Experimental + DescribeNexusOperationOptions = internal.ClientDescribeNexusOperationOptions + + // CancelNexusOperationOptions contains options for NexusOperationHandle.Cancel call. + // + // NOTE: Experimental + CancelNexusOperationOptions = internal.ClientCancelNexusOperationOptions + + // TerminateNexusOperationOptions contains options for NexusOperationHandle.Terminate call. + // + // NOTE: Experimental + TerminateNexusOperationOptions = internal.ClientTerminateNexusOperationOptions + + // ListNexusOperationsOptions contains input for ListNexusOperations call. + // + // NOTE: Experimental + ListNexusOperationsOptions = internal.ClientListNexusOperationsOptions + + // CountNexusOperationsOptions contains input for CountNexusOperations call. + // + // NOTE: Experimental + CountNexusOperationsOptions = internal.ClientCountNexusOperationsOptions + + // CountNexusOperationsResult contains the result of the CountNexusOperations call. + // + // NOTE: Experimental + CountNexusOperationsResult = internal.ClientCountNexusOperationsResult + + // CountNexusOperationsAggregationGroup contains groups of Nexus operations if + // CountNexusOperationExecutions is grouped by a field. + // + // NOTE: Experimental + CountNexusOperationsAggregationGroup = internal.ClientCountNexusOperationsAggregationGroup + + // ListNexusOperationsResult contains the result of the ListNexusOperations call. + // + // NOTE: Experimental + ListNexusOperationsResult = internal.ClientListNexusOperationsResult + + // GetNexusOperationHandleOptions contains input for GetNexusOperationHandle call. + // + // NOTE: Experimental + GetNexusOperationHandleOptions = internal.ClientGetNexusOperationHandleOptions + // Client is the client for starting and getting information about a workflow executions as well as // completing activities asynchronously. Client interface { @@ -1504,6 +1590,30 @@ type ( // NOTE: Experimental CountActivities(ctx context.Context, options CountActivitiesOptions) (*CountActivitiesResult, error) + // NexusClient creates a new Nexus client bound to the given endpoint and service. + // This is for standalone Nexus operations outside of workflow context. + // For Nexus operations within workflows, use workflow.NexusClient instead. + // + // NOTE: Experimental + NewNexusClient(options NexusClientOptions) (NexusClient, error) + + // GetNexusOperationHandle creates a handle to the referenced Nexus operation. + // No network call is made. The handle can be used to poll, describe, cancel, or terminate. + // + // NOTE: Experimental + GetNexusOperationHandle(options GetNexusOperationHandleOptions) NexusOperationHandle + + // ListNexusOperations lists Nexus operation executions based on query. + // Currently, all errors are returned in the iterator and not the base level error. + // + // NOTE: Experimental + ListNexusOperations(ctx context.Context, options ListNexusOperationsOptions) (ListNexusOperationsResult, error) + + // CountNexusOperations counts Nexus operation executions based on query. + // + // NOTE: Experimental + CountNexusOperations(ctx context.Context, options CountNexusOperationsOptions) (*CountNexusOperationsResult, error) + // WorkflowService provides access to the underlying gRPC service. This should only be used for advanced use cases // that cannot be accomplished via other Client methods. Unlike calls to other Client methods, calls directly to the // service are not configured with internal semantics such as automatic retries. diff --git a/interceptor/interceptor.go b/interceptor/interceptor.go index 55f497f67..318aceca7 100644 --- a/interceptor/interceptor.go +++ b/interceptor/interceptor.go @@ -244,6 +244,54 @@ type ClientPollActivityResultInput = internal.ClientPollActivityResultInput // NOTE: Experimental type ClientPollActivityResultOutput = internal.ClientPollActivityResultOutput +// ClientExecuteNexusOperationInput is the input to +// ClientOutboundInterceptor.ExecuteNexusOperation. +// +// NOTE: Experimental +type ClientExecuteNexusOperationInput = internal.ClientExecuteNexusOperationInput + +// ClientGetNexusOperationHandleInput is the input to +// ClientOutboundInterceptor.GetNexusOperationHandle. +// +// NOTE: Experimental +type ClientGetNexusOperationHandleInput = internal.ClientGetNexusOperationHandleInput + +// ClientCancelNexusOperationInput is the input to +// ClientOutboundInterceptor.CancelNexusOperation. +// +// NOTE: Experimental +type ClientCancelNexusOperationInput = internal.ClientCancelNexusOperationInput + +// ClientTerminateNexusOperationInput is the input to +// ClientOutboundInterceptor.TerminateNexusOperation. +// +// NOTE: Experimental +type ClientTerminateNexusOperationInput = internal.ClientTerminateNexusOperationInput + +// ClientDescribeNexusOperationInput is the input to +// ClientOutboundInterceptor.DescribeNexusOperation. +// +// NOTE: Experimental +type ClientDescribeNexusOperationInput = internal.ClientDescribeNexusOperationInput + +// ClientDescribeNexusOperationOutput is the output of +// ClientOutboundInterceptor.DescribeNexusOperation. +// +// NOTE: Experimental +type ClientDescribeNexusOperationOutput = internal.ClientDescribeNexusOperationOutput + +// ClientPollNexusOperationResultInput is the input to +// ClientOutboundInterceptor.PollNexusOperationResult. +// +// NOTE: Experimental +type ClientPollNexusOperationResultInput = internal.ClientPollNexusOperationResultInput + +// ClientPollNexusOperationResultOutput is the output of +// ClientOutboundInterceptor.PollNexusOperationResult. +// +// NOTE: Experimental +type ClientPollNexusOperationResultOutput = internal.ClientPollNexusOperationResultOutput + // ScheduleClientCreateInput is input for // ScheduleClientInterceptor.CreateSchedule. type ScheduleClientCreateInput = internal.ScheduleClientCreateInput diff --git a/internal/client.go b/internal/client.go index 8f81dd0f2..0a4234366 100644 --- a/internal/client.go +++ b/internal/client.go @@ -534,6 +534,30 @@ type ( // NOTE: Experimental CountActivities(ctx context.Context, options ClientCountActivitiesOptions) (*ClientCountActivitiesResult, error) + // NewNexusClient creates a new Nexus client bound to the given endpoint and service. + // This is for standalone Nexus operations outside of workflow context. + // For Nexus operations within workflows, use workflow.NewNexusClient instead. + // + // NOTE: Experimental + NewNexusClient(options ClientNexusClientOptions) (ClientNexusClient, error) + + // GetNexusOperationHandle creates a handle to the referenced Nexus operation. + // No network call is made. The handle can be used to poll, describe, cancel, or terminate. + // + // NOTE: Experimental + GetNexusOperationHandle(options ClientGetNexusOperationHandleOptions) ClientNexusOperationHandle + + // ListNexusOperations lists Nexus operation executions based on query. + // Currently, all errors are returned in the iterator and not the base level error. + // + // NOTE: Experimental + ListNexusOperations(ctx context.Context, options ClientListNexusOperationsOptions) (ClientListNexusOperationsResult, error) + + // CountNexusOperations counts Nexus operation executions based on query. + // + // NOTE: Experimental + CountNexusOperations(ctx context.Context, options ClientCountNexusOperationsOptions) (*ClientCountNexusOperationsResult, error) + // WorkflowService provides access to the underlying gRPC service. This should only be used for advanced use cases // that cannot be accomplished via other Client methods. Unlike calls to other Client methods, calls directly to the // service are not configured with internal semantics such as automatic retries. diff --git a/internal/interceptor.go b/internal/interceptor.go index dbccfb945..8a530d387 100644 --- a/internal/interceptor.go +++ b/internal/interceptor.go @@ -450,6 +450,36 @@ type ClientOutboundInterceptor interface { // NOTE: Experimental PollActivityResult(context.Context, *ClientPollActivityResultInput) (*ClientPollActivityResultOutput, error) + // ExecuteNexusOperation intercepts NexusClient.ExecuteOperation. + // + // NOTE: Experimental + ExecuteNexusOperation(context.Context, *ClientExecuteNexusOperationInput) (ClientNexusOperationHandle, error) + + // GetNexusOperationHandle intercepts client.Client.GetNexusOperationHandle. + // + // NOTE: Experimental + GetNexusOperationHandle(*ClientGetNexusOperationHandleInput) ClientNexusOperationHandle + + // CancelNexusOperation intercepts NexusOperationHandle.Cancel. + // + // NOTE: Experimental + CancelNexusOperation(context.Context, *ClientCancelNexusOperationInput) error + + // TerminateNexusOperation intercepts NexusOperationHandle.Terminate. + // + // NOTE: Experimental + TerminateNexusOperation(context.Context, *ClientTerminateNexusOperationInput) error + + // DescribeNexusOperation intercepts NexusOperationHandle.Describe. + // + // NOTE: Experimental + DescribeNexusOperation(context.Context, *ClientDescribeNexusOperationInput) (*ClientDescribeNexusOperationOutput, error) + + // PollNexusOperationResult intercepts NexusOperationHandle.Get. + // + // NOTE: Experimental + PollNexusOperationResult(context.Context, *ClientPollNexusOperationResultInput) (*ClientPollNexusOperationResultOutput, error) + mustEmbedClientOutboundInterceptorBase() } @@ -670,6 +700,101 @@ type ClientPollActivityResultOutput struct { Error error } +// ClientExecuteNexusOperationInput is the input to +// ClientOutboundInterceptor.ExecuteNexusOperation. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientExecuteNexusOperationInput] +type ClientExecuteNexusOperationInput struct { + Options *ClientStartNexusOperationOptions + Endpoint string + Service string + OperationType string + Input interface{} // single value, NOT Args []interface{} +} + +// ClientGetNexusOperationHandleInput is the input to +// ClientOutboundInterceptor.GetNexusOperationHandle. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientGetNexusOperationHandleInput] +type ClientGetNexusOperationHandleInput struct { + OperationID string + RunID string +} + +// ClientCancelNexusOperationInput is the input to +// ClientOutboundInterceptor.CancelNexusOperation. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientCancelNexusOperationInput] +type ClientCancelNexusOperationInput struct { + OperationID string + RunID string + Reason string +} + +// ClientTerminateNexusOperationInput is the input to +// ClientOutboundInterceptor.TerminateNexusOperation. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientTerminateNexusOperationInput] +type ClientTerminateNexusOperationInput struct { + OperationID string + RunID string + Reason string +} + +// ClientDescribeNexusOperationInput is the input to +// ClientOutboundInterceptor.DescribeNexusOperation. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientDescribeNexusOperationInput] +type ClientDescribeNexusOperationInput struct { + OperationID string + RunID string +} + +// ClientDescribeNexusOperationOutput is the output of +// ClientOutboundInterceptor.DescribeNexusOperation. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientDescribeNexusOperationOutput] +type ClientDescribeNexusOperationOutput struct { + Description *ClientNexusOperationExecutionDescription +} + +// ClientPollNexusOperationResultInput is the input to +// ClientOutboundInterceptor.PollNexusOperationResult. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientPollNexusOperationResultInput] +type ClientPollNexusOperationResultInput struct { + OperationID string + RunID string +} + +// ClientPollNexusOperationResultOutput is the output of +// ClientOutboundInterceptor.PollNexusOperationResult. +// +// NOTE: Experimental +// +// Exposed as: [go.temporal.io/sdk/interceptor.ClientPollNexusOperationResultOutput] +type ClientPollNexusOperationResultOutput struct { + // Result is the result of the operation, if it has completed successfully. + Result converter.EncodedValue + // Error is the result of a failed operation. + Error error +} + + // NexusOutboundInterceptor intercepts Nexus operation method invocations. See documentation in the interceptor package // for more details. // diff --git a/internal/interceptor_base.go b/internal/interceptor_base.go index c1ae31044..487760093 100644 --- a/internal/interceptor_base.go +++ b/internal/interceptor_base.go @@ -635,6 +635,65 @@ func (c *ClientOutboundInterceptorBase) PollActivityResult( return c.Next.PollActivityResult(ctx, in) } +// ExecuteNexusOperation implements ClientOutboundInterceptor.ExecuteNexusOperation. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) ExecuteNexusOperation( + ctx context.Context, + in *ClientExecuteNexusOperationInput, +) (ClientNexusOperationHandle, error) { + return c.Next.ExecuteNexusOperation(ctx, in) +} + +// GetNexusOperationHandle implements ClientOutboundInterceptor.GetNexusOperationHandle. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) GetNexusOperationHandle( + in *ClientGetNexusOperationHandleInput, +) ClientNexusOperationHandle { + return c.Next.GetNexusOperationHandle(in) +} + +// CancelNexusOperation implements ClientOutboundInterceptor.CancelNexusOperation. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) CancelNexusOperation( + ctx context.Context, + in *ClientCancelNexusOperationInput, +) error { + return c.Next.CancelNexusOperation(ctx, in) +} + +// TerminateNexusOperation implements ClientOutboundInterceptor.TerminateNexusOperation. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) TerminateNexusOperation( + ctx context.Context, + in *ClientTerminateNexusOperationInput, +) error { + return c.Next.TerminateNexusOperation(ctx, in) +} + +// DescribeNexusOperation implements ClientOutboundInterceptor.DescribeNexusOperation. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) DescribeNexusOperation( + ctx context.Context, + in *ClientDescribeNexusOperationInput, +) (*ClientDescribeNexusOperationOutput, error) { + return c.Next.DescribeNexusOperation(ctx, in) +} + +// PollNexusOperationResult implements ClientOutboundInterceptor.PollNexusOperationResult. +// +// NOTE: Experimental +func (c *ClientOutboundInterceptorBase) PollNexusOperationResult( + ctx context.Context, + in *ClientPollNexusOperationResultInput, +) (*ClientPollNexusOperationResultOutput, error) { + return c.Next.PollNexusOperationResult(ctx, in) +} + func (*ClientOutboundInterceptorBase) mustEmbedClientOutboundInterceptorBase() {} // NexusOperationInboundInterceptorBase is a default implementation of [NexusOperationInboundInterceptor] that diff --git a/internal/internal_nexus_client.go b/internal/internal_nexus_client.go new file mode 100644 index 000000000..33677c9aa --- /dev/null +++ b/internal/internal_nexus_client.go @@ -0,0 +1,865 @@ +package internal + +import ( + "context" + "errors" + "fmt" + "iter" + "reflect" + "time" + + "github.com/google/uuid" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/converter" + "google.golang.org/protobuf/types/known/durationpb" +) + +const pollNexusOperationTimeout = 60 * time.Second + +type ( + // ClientStartNexusOperationOptions contains configuration parameters for starting a Nexus operation execution. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.StartNexusOperationOptions] + ClientStartNexusOperationOptions struct { + // OperationID - The business identifier of the operation. + // + // Mandatory: No default. + OperationID string + // ScheduleToCloseTimeout - Total time that the operation is allowed to run. + // + // Optional: Defaults to unlimited. + ScheduleToCloseTimeout time.Duration + // IDConflictPolicy - Defines how to resolve an operation id conflict with a running operation. + // + // Optional: Defaults to NEXUS_OPERATION_ID_CONFLICT_POLICY_FAIL. + IDConflictPolicy enumspb.NexusOperationIdConflictPolicy + // IDReusePolicy - Defines whether to allow re-using an operation ID from a previously closed operation. + // + // Optional: Defaults to NEXUS_OPERATION_ID_REUSE_POLICY_ALLOW_DUPLICATE. + IDReusePolicy enumspb.NexusOperationIdReusePolicy + // SearchAttributes - Specifies Search Attributes that will be attached to the operation. + // + // Optional: default to none. + SearchAttributes SearchAttributes + // Summary is a single-line summary for this operation that will appear in UI/CLI. + // + // Optional: defaults to none/empty. + Summary string + } + + // ClientNexusClientOptions contains options for creating a NexusClient. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusClientOptions] + ClientNexusClientOptions struct { + // Endpoint - The Nexus endpoint name. + // + // Mandatory: No default. + Endpoint string + // Service - The Nexus service name. + // + // Mandatory: No default. + Service string + } + + // ClientGetNexusOperationHandleOptions contains input for GetNexusOperationHandle call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.GetNexusOperationHandleOptions] + ClientGetNexusOperationHandleOptions struct { + // OperationID - The operation ID. + // + // Mandatory: No default. + OperationID string + // RunID - The run ID. Can be empty to target the latest run. + // + // Optional: defaults to empty. + RunID string + } + + // ClientDescribeNexusOperationOptions contains options for ClientNexusOperationHandle.Describe call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.DescribeNexusOperationOptions] + ClientDescribeNexusOperationOptions struct { + } + + // ClientCancelNexusOperationOptions contains options for ClientNexusOperationHandle.Cancel call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.CancelNexusOperationOptions] + ClientCancelNexusOperationOptions struct { + // Reason is optional description of the reason for cancellation. + Reason string + } + + // ClientTerminateNexusOperationOptions contains options for ClientNexusOperationHandle.Terminate call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.TerminateNexusOperationOptions] + ClientTerminateNexusOperationOptions struct { + // Reason is optional description of the reason for termination. + Reason string + } + + // ClientListNexusOperationsOptions contains input for ListNexusOperations call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.ListNexusOperationsOptions] + ClientListNexusOperationsOptions struct { + // Query is a visibility query for listing Nexus operations. + // See https://docs.temporal.io/list-filter for the syntax. + Query string + } + + // ClientCountNexusOperationsOptions contains input for CountNexusOperations call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.CountNexusOperationsOptions] + ClientCountNexusOperationsOptions struct { + // Query is a visibility query for counting Nexus operations. + // See https://docs.temporal.io/list-filter for the syntax. + Query string + } + + // ClientNexusOperationMetadata contains information about a Nexus operation execution. + // This is returned by ListNexusOperations and embedded in ClientNexusOperationExecutionDescription. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusOperationMetadata] + ClientNexusOperationMetadata struct { + // RawExecutionListInfo is the raw PB message this struct was built from. This field is nil + // in the result of ClientNexusOperationHandle.Describe call - use + // ClientNexusOperationExecutionDescription.RawInfo instead. + RawExecutionListInfo *nexuspb.NexusOperationExecutionListInfo + // OperationID is the unique identifier of this operation within its namespace. + OperationID string + // OperationRunID is the run ID of the operation. + OperationRunID string + // Endpoint is the Nexus endpoint name. + Endpoint string + // Service is the Nexus service name. + Service string + // Operation is the Nexus operation name. + Operation string + // ScheduledTime is the time when the operation was originally scheduled. + ScheduledTime time.Time + // CloseTime is the time when the operation transitioned to a terminal state. + CloseTime time.Time + // Status is the current execution status of the operation. + Status enumspb.NexusOperationExecutionStatus + // SearchAttributes are the search attributes attached to this operation. + SearchAttributes SearchAttributes + // StateTransitionCount is incremented each time the operation state is mutated. + StateTransitionCount int64 + // ExecutionDuration is the difference between close time and scheduled time. + // Only populated if the operation is closed. + ExecutionDuration time.Duration + } + + // ClientNexusOperationExecutionDescription contains detailed information about a Nexus operation execution. + // This is returned by ClientNexusOperationHandle.Describe. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusOperationExecutionDescription] + ClientNexusOperationExecutionDescription struct { + ClientNexusOperationMetadata + // RawInfo is the raw PB message this struct was built from. + RawInfo *nexuspb.NexusOperationExecutionInfo + // State is a more detailed breakdown of the running status. + State enumspb.PendingNexusOperationState + // ScheduleToCloseTimeout is the schedule-to-close timeout for this operation. + ScheduleToCloseTimeout time.Duration + // ScheduleToStartTimeout is the schedule-to-start timeout for this operation. + // May not be populated by all server versions. + ScheduleToStartTimeout time.Duration + // StartToCloseTimeout is the start-to-close timeout for this operation. + // May not be populated by all server versions. + StartToCloseTimeout time.Duration + // Attempt is the number of attempts made to start/deliver the operation request. + Attempt int32 + // ExpirationTime is the scheduled time plus schedule-to-close timeout. + ExpirationTime time.Time + // LastAttemptCompleteTime is the time when the last attempt completed. + LastAttemptCompleteTime time.Time + // NextAttemptScheduleTime is the time when the next attempt is scheduled. + NextAttemptScheduleTime time.Time + // LastAttemptFailure is the last attempt's failure, if any. + LastAttemptFailure *failurepb.Failure + // BlockedReason provides additional information if the state is BLOCKED. + BlockedReason string + // OperationToken is only set for asynchronous operations after a successful StartOperation call. + OperationToken string + // Identity is the identity of the client who started this operation. + Identity string + // CancellationInfo contains cancellation information if cancellation has been requested. + CancellationInfo *ClientNexusOperationCancellationInfo + dc converter.DataConverter + failureConverter converter.FailureConverter + inboundPayloadVisitor PayloadVisitor + } + + // ClientNexusOperationCancellationInfo contains cancellation information for a Nexus operation. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusOperationCancellationInfo] + ClientNexusOperationCancellationInfo struct { + // RawInfo is the raw PB message this struct was built from. + RawInfo *nexuspb.NexusOperationExecutionCancellationInfo + // RequestedTime is the time when cancellation was requested. + RequestedTime time.Time + // State is the current state of the cancellation request. + State enumspb.NexusOperationCancellationState + // Attempt is the number of attempts made to deliver the cancel operation request. + Attempt int32 + // LastAttemptCompleteTime is the time when the last cancellation attempt completed. + LastAttemptCompleteTime time.Time + // NextAttemptScheduleTime is the time when the next cancellation attempt is scheduled. + NextAttemptScheduleTime time.Time + // BlockedReason provides additional information if the cancellation state is BLOCKED. + BlockedReason string + // Reason is the reason specified in the cancellation request. + Reason string + lastAttemptFailure *failurepb.Failure + failureConverter converter.FailureConverter + inboundPayloadVisitor PayloadVisitor + } + + // ClientCountNexusOperationsResult contains the result of the CountNexusOperations call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.CountNexusOperationsResult] + ClientCountNexusOperationsResult struct { + // Count is the approximate number of operations matching the query. + Count int64 + // Groups contains aggregation groups if the query includes a GROUP BY clause. + Groups []ClientCountNexusOperationsAggregationGroup + } + + // ClientCountNexusOperationsAggregationGroup contains groups of Nexus operations if + // CountNexusOperationExecutions is grouped by a field. + // The list might not be complete, and the counts of each group is approximate. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.CountNexusOperationsAggregationGroup] + ClientCountNexusOperationsAggregationGroup struct { + // GroupValues contains the group-by field values for this group. + GroupValues []any + // Count is the approximate number of operations in this group. + Count int64 + } + + // ClientListNexusOperationsResult contains the result of the ListNexusOperations call. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.ListNexusOperationsResult] + ClientListNexusOperationsResult struct { + // Results is an iterator over Nexus operation metadata entries. + Results iter.Seq2[*ClientNexusOperationMetadata, error] + } + + // ClientNexusClient is the client for starting Nexus operations bound to a specific endpoint and service. + // This is for standalone Nexus operations outside of workflow context. + // For Nexus operations within workflows, use workflow.NexusClient instead. + // + // Methods may be added to this interface; implementing it directly is discouraged. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusClient] + ClientNexusClient interface { + // ExecuteOperation starts a Nexus operation and returns a handle to it. + // + // NOTE: Experimental + ExecuteOperation(ctx context.Context, operation any, input any, options ClientStartNexusOperationOptions) (ClientNexusOperationHandle, error) + } + + // ClientNexusOperationHandle represents a running or completed standalone Nexus operation execution. + // It can be used to get the result, describe, cancel, or terminate the operation. + // + // Methods may be added to this interface; implementing it directly is discouraged. + // + // NOTE: Experimental + // + // Exposed as: [go.temporal.io/sdk/client.NexusOperationHandle] + ClientNexusOperationHandle interface { + // GetID returns the ID of the operation this handle points to. + GetID() string + // GetRunID returns the run ID that this handle was created with. + GetRunID() string + // Get waits until the operation finishes and gets its result. If the operation completes + // successfully, the result is written to valuePtr and nil is returned. If the operation + // failed, the failure is returned as an error. + Get(ctx context.Context, valuePtr any) error + // Describe returns detailed information about current state of the operation execution. + Describe(ctx context.Context, options ClientDescribeNexusOperationOptions) (*ClientNexusOperationExecutionDescription, error) + // Cancel requests cancellation of the operation. + Cancel(ctx context.Context, options ClientCancelNexusOperationOptions) error + // Terminate terminates the operation. + Terminate(ctx context.Context, options ClientTerminateNexusOperationOptions) error + } + + // nexusClientImpl is the default implementation of ClientNexusClient. + nexusClientImpl struct { + client *WorkflowClient + endpoint string + service string + } + + // clientNexusOperationHandleImpl is the default implementation of ClientNexusOperationHandle. + clientNexusOperationHandleImpl struct { + client *WorkflowClient + id string + runID string + result *ClientPollNexusOperationResultOutput + } +) + +// GetSummary returns summary of the operation. See ClientStartNexusOperationOptions.Summary. +// Returns empty string if there is no summary. +// Uses the data converter of the client used to make the Describe call. Returns error if data conversion fails. +// +// NOTE: Experimental +func (d *ClientNexusOperationExecutionDescription) GetSummary() (string, error) { + payload := d.RawInfo.GetUserMetadata().GetSummary() + if payload == nil { + return "", nil + } + var err error + if payload, err = visitPayload(context.Background(), d.inboundPayloadVisitor, payload); err != nil { + return "", err + } + var summary string + err = d.dc.FromPayload(payload, &summary) + if err != nil { + return "", err + } + return summary, nil +} + +// GetLastAttemptFailure returns the last attempt failure of the operation, using the failure +// converter of the client used to make the Describe call. Returns nil if there was no failure. +// +// NOTE: Experimental +func (d *ClientNexusOperationExecutionDescription) GetLastAttemptFailure() error { + failure := d.LastAttemptFailure + if failure == nil { + return nil + } + if err := visitProtoPayloads(context.Background(), d.inboundPayloadVisitor, failure); err != nil { + return err + } + return d.failureConverter.FailureToError(failure) +} + +// GetLastAttemptFailure returns the last attempt failure of the cancellation info. +// Returns nil if there was no failure. +// +// NOTE: Experimental +func (c *ClientNexusOperationCancellationInfo) GetLastAttemptFailure() error { + if c.lastAttemptFailure == nil { + return nil + } + if err := visitProtoPayloads(context.Background(), c.inboundPayloadVisitor, c.lastAttemptFailure); err != nil { + return err + } + return c.failureConverter.FailureToError(c.lastAttemptFailure) +} + +func (nc *nexusClientImpl) ExecuteOperation(ctx context.Context, operation any, input any, options ClientStartNexusOperationOptions) (ClientNexusOperationHandle, error) { + if err := nc.client.ensureInitialized(ctx); err != nil { + return nil, err + } + + // Resolve operation name from the operation parameter + operationName, err := resolveNexusOperationName(operation, input) + if err != nil { + return nil, err + } + + // Set header before interceptor run so interceptors can access it + ctx = contextWithNewHeader(ctx) + + return nc.client.interceptor.ExecuteNexusOperation(ctx, &ClientExecuteNexusOperationInput{ + Options: &options, + Endpoint: nc.endpoint, + Service: nc.service, + OperationType: operationName, + Input: input, + }) +} + +// resolveNexusOperationName resolves a Nexus operation name from the given value. +// It accepts a string name or a typed operation reference (with Name() and InputType() methods). +// This matches the resolution logic used in workflow context (see prepareNexusOperationParams). +func resolveNexusOperationName(operation any, input any) (string, error) { + if name, ok := operation.(string); ok { + if name == "" { + return "", fmt.Errorf("operation name must not be empty") + } + return name, nil + } + if regOp, ok := operation.(interface { + Name() string + InputType() reflect.Type + }); ok { + operationName := regOp.Name() + inputType := reflect.TypeOf(input) + if inputType != nil && !inputType.AssignableTo(regOp.InputType()) { + return "", fmt.Errorf("cannot assign argument of type %q to type %q for operation %q", inputType, regOp.InputType(), operationName) + } + return operationName, nil + } + return "", fmt.Errorf("invalid 'operation' parameter, must be an OperationReference or a string") +} + +func (h *clientNexusOperationHandleImpl) GetID() string { + return h.id +} + +func (h *clientNexusOperationHandleImpl) GetRunID() string { + return h.runID +} + +func (h *clientNexusOperationHandleImpl) Get(ctx context.Context, valuePtr any) error { + if h.result != nil { + if h.result.Error != nil { + return h.result.Error + } + if h.result.Result != nil { + if valuePtr == nil { + return nil + } + return h.result.Result.Get(valuePtr) + } + } + if err := h.client.ensureInitialized(ctx); err != nil { + return err + } + + // repeatedly poll, the loop repeats until there's an outcome + for { + resp, err := h.client.interceptor.PollNexusOperationResult(ctx, &ClientPollNexusOperationResultInput{ + OperationID: h.id, + RunID: h.runID, + }) + if err != nil { + return err + } + if resp.Error != nil { + h.result = &ClientPollNexusOperationResultOutput{Error: resp.Error} + return resp.Error + } + if resp.Result != nil { + if valuePtr == nil { + return nil + } + h.result = &ClientPollNexusOperationResultOutput{Result: resp.Result} + return resp.Result.Get(valuePtr) + } + } +} + +func (h *clientNexusOperationHandleImpl) Describe(ctx context.Context, options ClientDescribeNexusOperationOptions) (*ClientNexusOperationExecutionDescription, error) { + if err := h.client.ensureInitialized(ctx); err != nil { + return nil, err + } + out, err := h.client.interceptor.DescribeNexusOperation(ctx, &ClientDescribeNexusOperationInput{ + OperationID: h.id, + RunID: h.runID, + }) + if err != nil { + return nil, err + } + return out.Description, nil +} + +func (h *clientNexusOperationHandleImpl) Cancel(ctx context.Context, options ClientCancelNexusOperationOptions) error { + if err := h.client.ensureInitialized(ctx); err != nil { + return err + } + return h.client.interceptor.CancelNexusOperation(ctx, &ClientCancelNexusOperationInput{ + OperationID: h.id, + RunID: h.runID, + Reason: options.Reason, + }) +} + +func (h *clientNexusOperationHandleImpl) Terminate(ctx context.Context, options ClientTerminateNexusOperationOptions) error { + if err := h.client.ensureInitialized(ctx); err != nil { + return err + } + return h.client.interceptor.TerminateNexusOperation(ctx, &ClientTerminateNexusOperationInput{ + OperationID: h.id, + RunID: h.runID, + Reason: options.Reason, + }) +} + +// WorkflowClient methods for Nexus operations + +func (wc *WorkflowClient) NewNexusClient(options ClientNexusClientOptions) (ClientNexusClient, error) { + if options.Endpoint == "" { + return nil, errors.New("endpoint is required") + } + if options.Service == "" { + return nil, errors.New("service is required") + } + return &nexusClientImpl{client: wc, endpoint: options.Endpoint, service: options.Service}, nil +} + +func (wc *WorkflowClient) GetNexusOperationHandle(options ClientGetNexusOperationHandleOptions) ClientNexusOperationHandle { + return wc.interceptor.GetNexusOperationHandle(&ClientGetNexusOperationHandleInput{ + OperationID: options.OperationID, + RunID: options.RunID, + }) +} + +// ListNexusOperations does not go through the interceptor chain, consistent with ListActivities. +func (wc *WorkflowClient) ListNexusOperations(ctx context.Context, options ClientListNexusOperationsOptions) (ClientListNexusOperationsResult, error) { + return ClientListNexusOperationsResult{ + Results: func(yield func(*ClientNexusOperationMetadata, error) bool) { + if err := wc.ensureInitialized(ctx); err != nil { + yield(nil, err) + return + } + + request := &workflowservice.ListNexusOperationExecutionsRequest{ + Namespace: wc.namespace, + Query: options.Query, + } + + for { + resp, err := wc.getListNexusOperationsPage(ctx, request) + if err != nil { + yield(nil, err) + return + } + + for _, op := range resp.Operations { + if !yield(&ClientNexusOperationMetadata{ + RawExecutionListInfo: op, + OperationID: op.OperationId, + OperationRunID: op.RunId, + Endpoint: op.Endpoint, + Service: op.Service, + Operation: op.Operation, + ScheduledTime: op.ScheduleTime.AsTime(), + CloseTime: op.CloseTime.AsTime(), + Status: op.Status, + SearchAttributes: convertToTypedSearchAttributes(wc.logger, op.SearchAttributes.GetIndexedFields()), + StateTransitionCount: op.StateTransitionCount, + ExecutionDuration: op.ExecutionDuration.AsDuration(), + }, nil) { + return + } + } + + if resp.NextPageToken != nil { + request.NextPageToken = resp.NextPageToken + } else { + return + } + } + }, + }, nil +} + +func (wc *WorkflowClient) getListNexusOperationsPage(ctx context.Context, request *workflowservice.ListNexusOperationExecutionsRequest) (*workflowservice.ListNexusOperationExecutionsResponse, error) { + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + return wc.WorkflowService().ListNexusOperationExecutions(grpcCtx, request) +} + +// CountNexusOperations does not go through the interceptor chain, consistent with CountActivities. +func (wc *WorkflowClient) CountNexusOperations(ctx context.Context, options ClientCountNexusOperationsOptions) (*ClientCountNexusOperationsResult, error) { + if err := wc.ensureInitialized(ctx); err != nil { + return nil, err + } + + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + request := &workflowservice.CountNexusOperationExecutionsRequest{ + Namespace: wc.namespace, + Query: options.Query, + } + resp, err := wc.WorkflowService().CountNexusOperationExecutions(grpcCtx, request) + if err != nil { + return nil, err + } + + groups := make([]ClientCountNexusOperationsAggregationGroup, len(resp.Groups)) + for i, group := range resp.Groups { + groupValues := make([]any, len(group.GroupValues)) + for j, groupValue := range group.GroupValues { + // should never fail, and if it does, leaving nil behind + _ = converter.GetDefaultDataConverter().FromPayload(groupValue, &groupValues[j]) + } + groups[i] = ClientCountNexusOperationsAggregationGroup{ + GroupValues: groupValues, + Count: group.Count, + } + } + + return &ClientCountNexusOperationsResult{ + Count: resp.Count, + Groups: groups, + }, nil +} + +// workflowClientInterceptor implementations for Nexus operations + +func (w *workflowClientInterceptor) ExecuteNexusOperation( + ctx context.Context, + in *ClientExecuteNexusOperationInput, +) (ClientNexusOperationHandle, error) { + dataConverter := WithContext(ctx, w.client.dataConverter) + if dataConverter == nil { + dataConverter = converter.GetDefaultDataConverter() + } + + if in.Options.OperationID == "" { + return nil, errors.New("operation ID is required") + } + if in.Options.ScheduleToCloseTimeout < 0 { + return nil, errors.New("ScheduleToCloseTimeout must not be negative") + } + + // Encode input as a single Payload (not Payloads) + var inputPayload *commonpb.Payload + if in.Input != nil { + var err error + inputPayload, err = dataConverter.ToPayload(in.Input) + if err != nil { + return nil, err + } + } + + searchAttrs, err := serializeTypedSearchAttributes(in.Options.SearchAttributes.GetUntypedValues()) + if err != nil { + return nil, err + } + + userMetadata, err := buildUserMetadata(in.Options.Summary, "", dataConverter) + if err != nil { + return nil, err + } + + request := &workflowservice.StartNexusOperationExecutionRequest{ + Namespace: w.client.namespace, + Identity: w.client.identity, + RequestId: uuid.NewString(), + OperationId: in.Options.OperationID, + Endpoint: in.Endpoint, + Service: in.Service, + Operation: in.OperationType, + Input: inputPayload, + IdReusePolicy: in.Options.IDReusePolicy, + IdConflictPolicy: in.Options.IDConflictPolicy, + SearchAttributes: searchAttrs, + UserMetadata: userMetadata, + } + if in.Options.ScheduleToCloseTimeout > 0 { + request.ScheduleToCloseTimeout = durationpb.New(in.Options.ScheduleToCloseTimeout) + } + if err := visitProtoPayloads(ctx, w.client.outboundPayloadVisitor, request); err != nil { + return nil, err + } + + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + resp, err := w.client.WorkflowService().StartNexusOperationExecution(grpcCtx, request) + if err != nil { + return nil, err + } + + return &clientNexusOperationHandleImpl{ + client: w.client, + id: in.Options.OperationID, + runID: resp.RunId, + }, nil +} + +func (w *workflowClientInterceptor) GetNexusOperationHandle( + in *ClientGetNexusOperationHandleInput, +) ClientNexusOperationHandle { + return &clientNexusOperationHandleImpl{ + client: w.client, + id: in.OperationID, + runID: in.RunID, + } +} + +func (w *workflowClientInterceptor) PollNexusOperationResult( + ctx context.Context, + in *ClientPollNexusOperationResultInput, +) (*ClientPollNexusOperationResultOutput, error) { + request := &workflowservice.PollNexusOperationExecutionRequest{ + Namespace: w.client.namespace, + OperationId: in.OperationID, + RunId: in.RunID, + WaitStage: enumspb.NEXUS_OPERATION_WAIT_STAGE_CLOSED, + } + + var resp *workflowservice.PollNexusOperationExecutionResponse + for resp.GetOutcome() == nil { + grpcCtx, cancel := newGRPCContext(ctx, grpcLongPoll(true), grpcTimeout(pollNexusOperationTimeout), defaultGrpcRetryParameters(ctx)) + var err error + resp, err = w.client.WorkflowService().PollNexusOperationExecution(grpcCtx, request) + cancel() + if err != nil { + return nil, err + } + } + + if err := visitProtoPayloads(ctx, w.client.inboundPayloadVisitor, resp); err != nil { + return nil, err + } + + switch v := resp.GetOutcome().(type) { + case *workflowservice.PollNexusOperationExecutionResponse_Result: + // Wrap single Payload in Payloads for EncodedValue compatibility + payloads := &commonpb.Payloads{Payloads: []*commonpb.Payload{v.Result}} + return &ClientPollNexusOperationResultOutput{Result: newEncodedValue(payloads, w.client.dataConverter)}, nil + case *workflowservice.PollNexusOperationExecutionResponse_Failure: + return &ClientPollNexusOperationResultOutput{Error: w.client.failureConverter.FailureToError(v.Failure)}, nil + default: + return nil, fmt.Errorf("unexpected nexus operation outcome type: %T", v) + } +} + +func (w *workflowClientInterceptor) DescribeNexusOperation( + ctx context.Context, + in *ClientDescribeNexusOperationInput, +) (*ClientDescribeNexusOperationOutput, error) { + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + request := &workflowservice.DescribeNexusOperationExecutionRequest{ + Namespace: w.client.namespace, + OperationId: in.OperationID, + RunId: in.RunID, + } + resp, err := w.client.WorkflowService().DescribeNexusOperationExecution(grpcCtx, request) + if err != nil { + return nil, err + } + info := resp.GetInfo() + if info == nil { + return nil, errors.New("DescribeNexusOperationExecution response doesn't contain info") + } + + var cancellationInfo *ClientNexusOperationCancellationInfo + if info.CancellationInfo != nil { + cancellationInfo = &ClientNexusOperationCancellationInfo{ + RawInfo: info.CancellationInfo, + RequestedTime: info.CancellationInfo.RequestedTime.AsTime(), + State: info.CancellationInfo.State, + Attempt: info.CancellationInfo.Attempt, + LastAttemptCompleteTime: info.CancellationInfo.LastAttemptCompleteTime.AsTime(), + NextAttemptScheduleTime: info.CancellationInfo.NextAttemptScheduleTime.AsTime(), + BlockedReason: info.CancellationInfo.BlockedReason, + Reason: info.CancellationInfo.Reason, + lastAttemptFailure: info.CancellationInfo.LastAttemptFailure, + failureConverter: w.client.failureConverter, + inboundPayloadVisitor: w.client.inboundPayloadVisitor, + } + } + + return &ClientDescribeNexusOperationOutput{ + Description: &ClientNexusOperationExecutionDescription{ + ClientNexusOperationMetadata: ClientNexusOperationMetadata{ + RawExecutionListInfo: nil, + OperationID: info.OperationId, + OperationRunID: info.RunId, + Endpoint: info.Endpoint, + Service: info.Service, + Operation: info.Operation, + ScheduledTime: info.ScheduleTime.AsTime(), + CloseTime: info.CloseTime.AsTime(), + Status: info.Status, + SearchAttributes: convertToTypedSearchAttributes(w.client.logger, info.SearchAttributes.GetIndexedFields()), + StateTransitionCount: info.StateTransitionCount, + ExecutionDuration: info.ExecutionDuration.AsDuration(), + }, + RawInfo: info, + State: info.State, + ScheduleToCloseTimeout: info.ScheduleToCloseTimeout.AsDuration(), + ScheduleToStartTimeout: info.ScheduleToStartTimeout.AsDuration(), + StartToCloseTimeout: info.StartToCloseTimeout.AsDuration(), + Attempt: info.Attempt, + ExpirationTime: info.ExpirationTime.AsTime(), + LastAttemptCompleteTime: info.LastAttemptCompleteTime.AsTime(), + NextAttemptScheduleTime: info.NextAttemptScheduleTime.AsTime(), + LastAttemptFailure: info.LastAttemptFailure, + BlockedReason: info.BlockedReason, + OperationToken: info.OperationToken, + Identity: info.Identity, + CancellationInfo: cancellationInfo, + dc: WithContext(ctx, w.client.dataConverter), + failureConverter: w.client.failureConverter, + inboundPayloadVisitor: w.client.inboundPayloadVisitor, + }, + }, nil +} + +func (w *workflowClientInterceptor) CancelNexusOperation( + ctx context.Context, + in *ClientCancelNexusOperationInput, +) error { + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + request := &workflowservice.RequestCancelNexusOperationExecutionRequest{ + Namespace: w.client.namespace, + OperationId: in.OperationID, + RunId: in.RunID, + Identity: w.client.identity, + RequestId: uuid.NewString(), + Reason: in.Reason, + } + _, err := w.client.WorkflowService().RequestCancelNexusOperationExecution(grpcCtx, request) + return err +} + +func (w *workflowClientInterceptor) TerminateNexusOperation( + ctx context.Context, + in *ClientTerminateNexusOperationInput, +) error { + grpcCtx, cancel := newGRPCContext(ctx, defaultGrpcRetryParameters(ctx)) + defer cancel() + + request := &workflowservice.TerminateNexusOperationExecutionRequest{ + Namespace: w.client.namespace, + OperationId: in.OperationID, + RunId: in.RunID, + Identity: w.client.identity, + RequestId: uuid.NewString(), + Reason: in.Reason, + } + _, err := w.client.WorkflowService().TerminateNexusOperationExecution(grpcCtx, request) + return err +} + diff --git a/internal/internal_nexus_client_test.go b/internal/internal_nexus_client_test.go new file mode 100644 index 000000000..abde26e5d --- /dev/null +++ b/internal/internal_nexus_client_test.go @@ -0,0 +1,90 @@ +package internal + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/workflowservice/v1" +) + +// nexusHeaderCheckInterceptor is a ClientInterceptor that verifies the header is +// present on the context when ExecuteNexusOperation is called. This ensures that +// contextWithNewHeader is called before the interceptor chain runs, so +// interceptors (like the tracing interceptor) can read/write headers. +type nexusHeaderCheckInterceptor struct { + ClientInterceptorBase + headerWasPresent bool +} + +func (h *nexusHeaderCheckInterceptor) InterceptClient(next ClientOutboundInterceptor) ClientOutboundInterceptor { + return &nexusHeaderCheckOutbound{ + ClientOutboundInterceptorBase: ClientOutboundInterceptorBase{Next: next}, + parent: h, + } +} + +type nexusHeaderCheckOutbound struct { + ClientOutboundInterceptorBase + parent *nexusHeaderCheckInterceptor +} + +func (h *nexusHeaderCheckOutbound) ExecuteNexusOperation( + ctx context.Context, + in *ClientExecuteNexusOperationInput, +) (ClientNexusOperationHandle, error) { + h.parent.headerWasPresent = Header(ctx) != nil + // Return an error to short-circuit the rest of the chain (avoids needing a + // real gRPC connection for the base interceptor). + return nil, fmt.Errorf("short-circuit") +} + +func TestExecuteNexusOperationHeaderAvailableToInterceptors(t *testing.T) { + interceptor := &nexusHeaderCheckInterceptor{} + + client := NewServiceClient(nil, nil, ClientOptions{ + Interceptors: []ClientInterceptor{interceptor}, + }) + // Pre-set capabilities so ensureInitialized doesn't make a gRPC call. + client.capabilities = &workflowservice.GetSystemInfoResponse_Capabilities{} + + nexusClient, err := client.NewNexusClient(ClientNexusClientOptions{ + Endpoint: "test-endpoint", + Service: "test-service", + }) + require.NoError(t, err) + + _, err = nexusClient.ExecuteOperation(context.Background(), "test-op", "test-input", ClientStartNexusOperationOptions{ + OperationID: "test-op-id", + }) + // We expect the short-circuit error from our interceptor. + require.ErrorContains(t, err, "short-circuit") + require.True(t, interceptor.headerWasPresent, + "Header should be set on context before interceptor chain runs") +} + +func TestNexusClientValidation(t *testing.T) { + client := NewServiceClient(nil, nil, ClientOptions{}) + + _, err := client.NewNexusClient(ClientNexusClientOptions{}) + require.ErrorContains(t, err, "endpoint is required") + + _, err = client.NewNexusClient(ClientNexusClientOptions{Endpoint: "ep"}) + require.ErrorContains(t, err, "service is required") + + nc, err := client.NewNexusClient(ClientNexusClientOptions{Endpoint: "ep", Service: "svc"}) + require.NoError(t, err) + require.NotNil(t, nc) +} + +func TestResolveNexusOperationName(t *testing.T) { + // String name + name, err := resolveNexusOperationName("my-op", nil) + require.NoError(t, err) + require.Equal(t, "my-op", name) + + // Invalid type + _, err = resolveNexusOperationName(123, nil) + require.ErrorContains(t, err, "invalid 'operation' parameter") +} diff --git a/internal/nexus_operations.go b/internal/nexus_operations.go index 1bbb94e86..a3a72c24c 100644 --- a/internal/nexus_operations.go +++ b/internal/nexus_operations.go @@ -797,6 +797,22 @@ func (t *testSuiteClientForNexusOperations) UpdateWorkflowExecutionOptions(ctx c panic("not implemented in the test environment") } +func (t *testSuiteClientForNexusOperations) NewNexusClient(options ClientNexusClientOptions) (ClientNexusClient, error) { + panic("not implemented in the test environment") +} + +func (t *testSuiteClientForNexusOperations) GetNexusOperationHandle(options ClientGetNexusOperationHandleOptions) ClientNexusOperationHandle { + panic("not implemented in the test environment") +} + +func (t *testSuiteClientForNexusOperations) ListNexusOperations(ctx context.Context, options ClientListNexusOperationsOptions) (ClientListNexusOperationsResult, error) { + panic("not implemented in the test environment") +} + +func (t *testSuiteClientForNexusOperations) CountNexusOperations(ctx context.Context, options ClientCountNexusOperationsOptions) (*ClientCountNexusOperationsResult, error) { + panic("not implemented in the test environment") +} + var _ Client = &testSuiteClientForNexusOperations{} // testEnvWorkflowRunForNexusOperations is a partial [WorkflowRun] implementation for the test workflow environment used diff --git a/mocks/Client.go b/mocks/Client.go index 48c441fe7..2fac5691e 100644 --- a/mocks/Client.go +++ b/mocks/Client.go @@ -1294,6 +1294,104 @@ func (_m *Client) CountActivities(ctx context.Context, options client.CountActiv return r0, r1 } +// NewNexusClient provides a mock function with given fields: options +func (_m *Client) NewNexusClient(options client.NexusClientOptions) (client.NexusClient, error) { + ret := _m.Called(options) + + if len(ret) == 0 { + panic("no return value specified for NexusClient") + } + + var r0 client.NexusClient + var r1 error + if rf, ok := ret.Get(0).(func(client.NexusClientOptions) (client.NexusClient, error)); ok { + return rf(options) + } + if rf, ok := ret.Get(0).(func(client.NexusClientOptions) client.NexusClient); ok { + r0 = rf(options) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(client.NexusClient) + } + } + + if rf, ok := ret.Get(1).(func(client.NexusClientOptions) error); ok { + r1 = rf(options) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetNexusOperationHandle provides a mock function with given fields: options +func (_m *Client) GetNexusOperationHandle(options client.GetNexusOperationHandleOptions) client.NexusOperationHandle { + ret := _m.Called(options) + + if len(ret) == 0 { + panic("no return value specified for GetNexusOperationHandle") + } + + var r0 client.NexusOperationHandle + if rf, ok := ret.Get(0).(func(client.GetNexusOperationHandleOptions) client.NexusOperationHandle); ok { + r0 = rf(options) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(client.NexusOperationHandle) + } + } + + return r0 +} + +// ListNexusOperations provides a mock function with given fields: ctx, options +func (_m *Client) ListNexusOperations(ctx context.Context, options client.ListNexusOperationsOptions) (client.ListNexusOperationsResult, error) { + ret := _m.Called(ctx, options) + + if len(ret) == 0 { + panic("no return value specified for ListNexusOperations") + } + + var r0 client.ListNexusOperationsResult + if rf, ok := ret.Get(0).(func(context.Context, client.ListNexusOperationsOptions) client.ListNexusOperationsResult); ok { + r0 = rf(ctx, options) + } else { + r0 = ret.Get(0).(client.ListNexusOperationsResult) + } + + return r0, nil +} + +// CountNexusOperations provides a mock function with given fields: ctx, options +func (_m *Client) CountNexusOperations(ctx context.Context, options client.CountNexusOperationsOptions) (*client.CountNexusOperationsResult, error) { + ret := _m.Called(ctx, options) + + if len(ret) == 0 { + panic("no return value specified for CountNexusOperations") + } + + var r0 *client.CountNexusOperationsResult + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, client.CountNexusOperationsOptions) (*client.CountNexusOperationsResult, error)); ok { + return rf(ctx, options) + } + if rf, ok := ret.Get(0).(func(context.Context, client.CountNexusOperationsOptions) *client.CountNexusOperationsResult); ok { + r0 = rf(ctx, options) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*client.CountNexusOperationsResult) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, client.CountNexusOperationsOptions) error); ok { + r1 = rf(ctx, options) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // WorkerDeploymentClient provides a mock function with given fields: func (_m *Client) WorkerDeploymentClient() client.WorkerDeploymentClient { ret := _m.Called() diff --git a/test/integration_test.go b/test/integration_test.go index 7669b93be..425211cf1 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -26,8 +26,11 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "go.opentelemetry.io/otel/trace" + "github.com/nexus-rpc/sdk-go/nexus" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/operatorservice/v1" "go.temporal.io/api/serviceerror" workflowpb "go.temporal.io/api/workflow/v1" "go.temporal.io/api/workflowservice/v1" @@ -53,6 +56,7 @@ import ( "go.temporal.io/sdk/internal/interceptortest" ilog "go.temporal.io/sdk/internal/log" "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/temporalnexus" "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" ) @@ -9148,3 +9152,194 @@ func (ts *IntegrationTestSuite) TestPayloadSizeWarningDefaultSize() { return strings.HasPrefix(line, "WARN [TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit.") })) } + +func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { + if os.Getenv("DISABLE_STANDALONE_NEXUS_TESTS") != "" { + ts.T().SkipNow() + } + + ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout) + defer cancel() + + // Create a Nexus endpoint targeting our task queue. + endpoint := "sdk-go-nexus-standalone-test-ep-" + uuid.NewString() + _, err := ts.client.OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{ + Spec: &nexuspb.EndpointSpec{ + Name: endpoint, + Target: &nexuspb.EndpointTarget{ + Variant: &nexuspb.EndpointTarget_Worker_{ + Worker: &nexuspb.EndpointTarget_Worker{ + Namespace: ts.config.Namespace, + TaskQueue: ts.taskQueueName, + }, + }, + }, + }, + }) + ts.NoError(err) + + // Register Nexus operations on the worker. + service := nexus.NewService("test-standalone-service") + syncOp := nexus.NewSyncOperation("echo-op", func(ctx context.Context, input string, opts nexus.StartOperationOptions) (string, error) { + return input, nil + }) + blockForeverWf := func(ctx workflow.Context, input string) (string, error) { + return "", workflow.Await(ctx, func() bool { return false }) + } + ts.worker.RegisterWorkflowWithOptions(blockForeverWf, workflow.RegisterOptions{Name: "block-forever-wf"}) + asyncOp := temporalnexus.NewWorkflowRunOperation( + "async-op", + blockForeverWf, + func(ctx context.Context, input string, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) { + return client.StartWorkflowOptions{ID: "nexus-async-" + uuid.NewString()}, nil + }, + ) + ts.NoError(service.Register(syncOp, asyncOp)) + ts.worker.RegisterNexusService(service) + + nexusClient, err := ts.client.NewNexusClient(client.NexusClientOptions{ + Endpoint: endpoint, + Service: "test-standalone-service", + }) + ts.NoError(err) + + // executeNexusOpWithRetry retries ExecuteOperation until the endpoint has propagated. + // The endpoint registry is eventually consistent and may take a few attempts. + executeNexusOpWithRetry := func( + opName string, + input string, + options client.StartNexusOperationOptions, + ) client.NexusOperationHandle { + ts.T().Helper() + var handle client.NexusOperationHandle + require.Eventually(ts.T(), func() bool { + var execErr error + handle, execErr = nexusClient.ExecuteOperation(ctx, opName, input, options) + return execErr == nil + }, 10*time.Second, 100*time.Millisecond, "timed out waiting for endpoint to propagate") + return handle + } + + ts.Run("Execute and Get result", func() { + input := "hello-nexus" + handle := executeNexusOpWithRetry("echo-op", input, client.StartNexusOperationOptions{ + OperationID: uuid.NewString(), + ScheduleToCloseTimeout: 10 * time.Second, + }) + ts.NotEmpty(handle.GetID()) + + var result string + err := handle.Get(ctx, &result) + ts.NoError(err) + ts.Equal(input, result) + }) + + ts.Run("Describe operation", func() { + handle := executeNexusOpWithRetry("echo-op", "describe-test", client.StartNexusOperationOptions{ + OperationID: uuid.NewString(), + ScheduleToCloseTimeout: 10 * time.Second, + }) + + // Wait for operation to complete. + err := handle.Get(ctx, nil) + ts.NoError(err) + + description, err := handle.Describe(ctx, client.DescribeNexusOperationOptions{}) + ts.NoError(err) + ts.Equal(handle.GetID(), description.OperationID) + ts.NotNil(description.RawInfo) + }) + + ts.Run("GetNexusOperationHandle", func() { + operationID := uuid.NewString() + handle := executeNexusOpWithRetry("echo-op", "handle-test", client.StartNexusOperationOptions{ + OperationID: operationID, + ScheduleToCloseTimeout: 10 * time.Second, + }) + + // Wait for operation to complete. + err := handle.Get(ctx, nil) + ts.NoError(err) + + // Get a handle to the same operation. + handle2 := ts.client.GetNexusOperationHandle(client.GetNexusOperationHandleOptions{ + OperationID: operationID, + RunID: handle.GetRunID(), + }) + ts.Equal(operationID, handle2.GetID()) + + var result string + err = handle2.Get(ctx, &result) + ts.NoError(err) + ts.Equal("handle-test", result) + }) + + ts.Run("Cancel operation", func() { + handle := executeNexusOpWithRetry("async-op", "cancel-test", client.StartNexusOperationOptions{ + OperationID: uuid.NewString(), + ScheduleToCloseTimeout: 30 * time.Second, + }) + ts.NotEmpty(handle.GetID()) + + // Operation record exists on the server after ExecuteOperation returns successfully; + // cancel targets the record by ID so no waiting is needed. + err := handle.Cancel(ctx, client.CancelNexusOperationOptions{Reason: "test cancellation"}) + ts.NoError(err) + }) + + ts.Run("Terminate operation", func() { + handle := executeNexusOpWithRetry("async-op", "terminate-test", client.StartNexusOperationOptions{ + OperationID: uuid.NewString(), + ScheduleToCloseTimeout: 30 * time.Second, + }) + ts.NotEmpty(handle.GetID()) + + // Operation record exists on the server after ExecuteOperation returns successfully; + // terminate targets the record by ID so no waiting is needed. + err := handle.Terminate(ctx, client.TerminateNexusOperationOptions{Reason: "test termination"}) + ts.NoError(err) + }) + + ts.Run("Count operations", func() { + // Visibility is eventually consistent; poll until operations appear. + require.Eventually(ts.T(), func() bool { + result, err := ts.client.CountNexusOperations(ctx, client.CountNexusOperationsOptions{ + Query: "Endpoint = '" + endpoint + "'", + }) + return err == nil && result.Count > 0 + }, 10*time.Second, 200*time.Millisecond, "timed out waiting for operations to appear in count") + }) + + ts.Run("List operations", func() { + // Visibility is eventually consistent; poll until operations appear. + require.Eventually(ts.T(), func() bool { + listResult, err := ts.client.ListNexusOperations(ctx, client.ListNexusOperationsOptions{ + Query: "Endpoint = '" + endpoint + "'", + }) + if err != nil { + return false + } + count := 0 + for metadata, iterErr := range listResult.Results { + if iterErr != nil { + return false + } + if metadata.OperationID == "" || metadata.Endpoint != endpoint { + return false + } + count++ + } + return count > 0 + }, 10*time.Second, 200*time.Millisecond, "timed out waiting for operations to appear in list") + }) + + ts.Run("NexusClient creation validation", func() { + _, err := ts.client.NewNexusClient(client.NexusClientOptions{}) + ts.Error(err) + ts.Contains(err.Error(), "endpoint is required") + + _, err = ts.client.NewNexusClient(client.NexusClientOptions{Endpoint: "ep"}) + ts.Error(err) + ts.Contains(err.Error(), "service is required") + }) +} From 619f5feffd7ed1f00418acb70c583f2fd7cea043 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 8 Apr 2026 10:15:45 -0700 Subject: [PATCH 02/12] Fix some naming issues --- internal/interceptor.go | 3 +-- internal/internal_nexus_client.go | 23 +++++++++++------------ internal/internal_nexus_client_test.go | 2 +- 3 files changed, 13 insertions(+), 15 deletions(-) diff --git a/internal/interceptor.go b/internal/interceptor.go index 8a530d387..2a05d117a 100644 --- a/internal/interceptor.go +++ b/internal/interceptor.go @@ -711,7 +711,7 @@ type ClientExecuteNexusOperationInput struct { Endpoint string Service string OperationType string - Input interface{} // single value, NOT Args []interface{} + Input interface{} } // ClientGetNexusOperationHandleInput is the input to @@ -794,7 +794,6 @@ type ClientPollNexusOperationResultOutput struct { Error error } - // NexusOutboundInterceptor intercepts Nexus operation method invocations. See documentation in the interceptor package // for more details. // diff --git a/internal/internal_nexus_client.go b/internal/internal_nexus_client.go index 33677c9aa..9a8cee953 100644 --- a/internal/internal_nexus_client.go +++ b/internal/internal_nexus_client.go @@ -27,10 +27,10 @@ type ( // // Exposed as: [go.temporal.io/sdk/client.StartNexusOperationOptions] ClientStartNexusOperationOptions struct { - // OperationID - The business identifier of the operation. + // ID - The business identifier of the operation. // // Mandatory: No default. - OperationID string + ID string // ScheduleToCloseTimeout - Total time that the operation is allowed to run. // // Optional: Defaults to unlimited. @@ -208,10 +208,10 @@ type ( // Identity is the identity of the client who started this operation. Identity string // CancellationInfo contains cancellation information if cancellation has been requested. - CancellationInfo *ClientNexusOperationCancellationInfo - dc converter.DataConverter - failureConverter converter.FailureConverter - inboundPayloadVisitor PayloadVisitor + CancellationInfo *ClientNexusOperationCancellationInfo + dc converter.DataConverter + failureConverter converter.FailureConverter + inboundPayloadVisitor PayloadVisitor } // ClientNexusOperationCancellationInfo contains cancellation information for a Nexus operation. @@ -557,7 +557,7 @@ func (wc *WorkflowClient) ListNexusOperations(ctx context.Context, options Clien for _, op := range resp.Operations { if !yield(&ClientNexusOperationMetadata{ - RawExecutionListInfo: op, + RawExecutionListInfo: op, OperationID: op.OperationId, OperationRunID: op.RunId, Endpoint: op.Endpoint, @@ -639,7 +639,7 @@ func (w *workflowClientInterceptor) ExecuteNexusOperation( dataConverter = converter.GetDefaultDataConverter() } - if in.Options.OperationID == "" { + if in.Options.ID == "" { return nil, errors.New("operation ID is required") } if in.Options.ScheduleToCloseTimeout < 0 { @@ -670,7 +670,7 @@ func (w *workflowClientInterceptor) ExecuteNexusOperation( Namespace: w.client.namespace, Identity: w.client.identity, RequestId: uuid.NewString(), - OperationId: in.Options.OperationID, + OperationId: in.Options.ID, Endpoint: in.Endpoint, Service: in.Service, Operation: in.OperationType, @@ -697,7 +697,7 @@ func (w *workflowClientInterceptor) ExecuteNexusOperation( return &clientNexusOperationHandleImpl{ client: w.client, - id: in.Options.OperationID, + id: in.Options.ID, runID: resp.RunId, }, nil } @@ -791,7 +791,7 @@ func (w *workflowClientInterceptor) DescribeNexusOperation( return &ClientDescribeNexusOperationOutput{ Description: &ClientNexusOperationExecutionDescription{ ClientNexusOperationMetadata: ClientNexusOperationMetadata{ - RawExecutionListInfo: nil, + RawExecutionListInfo: nil, OperationID: info.OperationId, OperationRunID: info.RunId, Endpoint: info.Endpoint, @@ -862,4 +862,3 @@ func (w *workflowClientInterceptor) TerminateNexusOperation( _, err := w.client.WorkflowService().TerminateNexusOperationExecution(grpcCtx, request) return err } - diff --git a/internal/internal_nexus_client_test.go b/internal/internal_nexus_client_test.go index abde26e5d..602921738 100644 --- a/internal/internal_nexus_client_test.go +++ b/internal/internal_nexus_client_test.go @@ -56,7 +56,7 @@ func TestExecuteNexusOperationHeaderAvailableToInterceptors(t *testing.T) { require.NoError(t, err) _, err = nexusClient.ExecuteOperation(context.Background(), "test-op", "test-input", ClientStartNexusOperationOptions{ - OperationID: "test-op-id", + ID: "test-op-id", }) // We expect the short-circuit error from our interceptor. require.ErrorContains(t, err, "short-circuit") From a1286f9b7bdf171f155d477714f298d5d14fea96 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 8 Apr 2026 10:18:01 -0700 Subject: [PATCH 03/12] Fix ID in test --- test/integration_test.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/test/integration_test.go b/test/integration_test.go index 425211cf1..488f884b7 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -17,6 +17,7 @@ import ( "time" "github.com/google/uuid" + "github.com/nexus-rpc/sdk-go/nexus" "github.com/opentracing/opentracing-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,7 +27,6 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "go.opentelemetry.io/otel/trace" - "github.com/nexus-rpc/sdk-go/nexus" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" nexuspb "go.temporal.io/api/nexus/v1" @@ -9223,7 +9223,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Run("Execute and Get result", func() { input := "hello-nexus" handle := executeNexusOpWithRetry("echo-op", input, client.StartNexusOperationOptions{ - OperationID: uuid.NewString(), + ID: uuid.NewString(), ScheduleToCloseTimeout: 10 * time.Second, }) ts.NotEmpty(handle.GetID()) @@ -9236,7 +9236,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Run("Describe operation", func() { handle := executeNexusOpWithRetry("echo-op", "describe-test", client.StartNexusOperationOptions{ - OperationID: uuid.NewString(), + ID: uuid.NewString(), ScheduleToCloseTimeout: 10 * time.Second, }) @@ -9253,7 +9253,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Run("GetNexusOperationHandle", func() { operationID := uuid.NewString() handle := executeNexusOpWithRetry("echo-op", "handle-test", client.StartNexusOperationOptions{ - OperationID: operationID, + ID: operationID, ScheduleToCloseTimeout: 10 * time.Second, }) @@ -9276,7 +9276,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Run("Cancel operation", func() { handle := executeNexusOpWithRetry("async-op", "cancel-test", client.StartNexusOperationOptions{ - OperationID: uuid.NewString(), + ID: uuid.NewString(), ScheduleToCloseTimeout: 30 * time.Second, }) ts.NotEmpty(handle.GetID()) @@ -9289,7 +9289,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Run("Terminate operation", func() { handle := executeNexusOpWithRetry("async-op", "terminate-test", client.StartNexusOperationOptions{ - OperationID: uuid.NewString(), + ID: uuid.NewString(), ScheduleToCloseTimeout: 30 * time.Second, }) ts.NotEmpty(handle.GetID()) From ebc15e2f1444fb851fb21fd0bdea59bb547d2437 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Fri, 10 Apr 2026 16:56:53 -0700 Subject: [PATCH 04/12] Respond to PR comments --- .github/workflows/ci.yml | 4 ++++ client/client.go | 2 +- internal/internal_nexus_client.go | 17 +++++++++++++++++ internal/internal_nexus_client_test.go | 20 ++++++++++++++++++++ mocks/Client.go | 2 +- 5 files changed, 43 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e1c06b119..90a7ef30a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,6 +78,7 @@ jobs: working-directory: ./internal/cmd/build env: WORKFLOW_CACHE_SIZE: "0" + DISABLE_STANDALONE_NEXUS_TESTS: "1" integration-test-with-cache: strategy: @@ -109,6 +110,8 @@ jobs: - name: Integration tests (with cache) run: go run . integration-test -dev-server working-directory: ./internal/cmd/build + env: + DISABLE_STANDALONE_NEXUS_TESTS: "1" docker-compose-test: runs-on: ubuntu-latest @@ -147,6 +150,7 @@ jobs: DISABLE_SERVER_1_27_TESTS: "1" DISABLE_PRIORITY_TESTS: "1" DISABLE_STANDALONE_ACTIVITY_TESTS: "1" + DISABLE_STANDALONE_NEXUS_TESTS: "1" DISABLE_NEXUS_CALLER_TIMEOUT_TESTS: "1" DISABLE_NEW_NEXUS_ERROR_FORMAT_TESTS: "1" working-directory: ./internal/cmd/build diff --git a/client/client.go b/client/client.go index 0c39aed5f..08ac35ea0 100644 --- a/client/client.go +++ b/client/client.go @@ -1590,7 +1590,7 @@ type ( // NOTE: Experimental CountActivities(ctx context.Context, options CountActivitiesOptions) (*CountActivitiesResult, error) - // NexusClient creates a new Nexus client bound to the given endpoint and service. + // NewNexusClient creates a new Nexus client bound to the given endpoint and service. // This is for standalone Nexus operations outside of workflow context. // For Nexus operations within workflows, use workflow.NexusClient instead. // diff --git a/internal/internal_nexus_client.go b/internal/internal_nexus_client.go index 9a8cee953..5bc78d565 100644 --- a/internal/internal_nexus_client.go +++ b/internal/internal_nexus_client.go @@ -303,18 +303,30 @@ type ( // Exposed as: [go.temporal.io/sdk/client.NexusOperationHandle] ClientNexusOperationHandle interface { // GetID returns the ID of the operation this handle points to. + // + // NOTE: Experimental GetID() string // GetRunID returns the run ID that this handle was created with. + // + // NOTE: Experimental GetRunID() string // Get waits until the operation finishes and gets its result. If the operation completes // successfully, the result is written to valuePtr and nil is returned. If the operation // failed, the failure is returned as an error. + // + // NOTE: Experimental Get(ctx context.Context, valuePtr any) error // Describe returns detailed information about current state of the operation execution. + // + // NOTE: Experimental Describe(ctx context.Context, options ClientDescribeNexusOperationOptions) (*ClientNexusOperationExecutionDescription, error) // Cancel requests cancellation of the operation. + // + // NOTE: Experimental Cancel(ctx context.Context, options ClientCancelNexusOperationOptions) error // Terminate terminates the operation. + // + // NOTE: Experimental Terminate(ctx context.Context, options ClientTerminateNexusOperationOptions) error } @@ -334,6 +346,11 @@ type ( } ) +var ( + _ ClientNexusClient = &nexusClientImpl{} + _ ClientNexusOperationHandle = &clientNexusOperationHandleImpl{} +) + // GetSummary returns summary of the operation. See ClientStartNexusOperationOptions.Summary. // Returns empty string if there is no summary. // Uses the data converter of the client used to make the Describe call. Returns error if data conversion fails. diff --git a/internal/internal_nexus_client_test.go b/internal/internal_nexus_client_test.go index 602921738..e4fd9df3c 100644 --- a/internal/internal_nexus_client_test.go +++ b/internal/internal_nexus_client_test.go @@ -3,6 +3,7 @@ package internal import ( "context" "fmt" + "reflect" "testing" "github.com/stretchr/testify/require" @@ -78,12 +79,31 @@ func TestNexusClientValidation(t *testing.T) { require.NotNil(t, nc) } +// mockOperationReference implements the Name()/InputType() interface used by resolveNexusOperationName. +type mockOperationReference struct { + name string + inputType reflect.Type +} + +func (m mockOperationReference) Name() string { return m.name } +func (m mockOperationReference) InputType() reflect.Type { return m.inputType } + func TestResolveNexusOperationName(t *testing.T) { // String name name, err := resolveNexusOperationName("my-op", nil) require.NoError(t, err) require.Equal(t, "my-op", name) + // Typed operation reference with correct input type + op := mockOperationReference{name: "typed-op", inputType: reflect.TypeOf("")} + name, err = resolveNexusOperationName(op, "hello") + require.NoError(t, err) + require.Equal(t, "typed-op", name) + + // Typed operation reference with wrong input type + _, err = resolveNexusOperationName(op, 123) + require.ErrorContains(t, err, "cannot assign argument of type") + // Invalid type _, err = resolveNexusOperationName(123, nil) require.ErrorContains(t, err, "invalid 'operation' parameter") diff --git a/mocks/Client.go b/mocks/Client.go index 2fac5691e..5149b0149 100644 --- a/mocks/Client.go +++ b/mocks/Client.go @@ -1299,7 +1299,7 @@ func (_m *Client) NewNexusClient(options client.NexusClientOptions) (client.Nexu ret := _m.Called(options) if len(ret) == 0 { - panic("no return value specified for NexusClient") + panic("no return value specified for NewNexusClient") } var r0 client.NexusClient From c828f3e6f3e8a98b9056e54bf3536d90ec1c141d Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 08:01:51 -0700 Subject: [PATCH 05/12] Update API --- contrib/tools/workflowcheck/go.mod | 1 + contrib/tools/workflowcheck/go.sum | 2 ++ 2 files changed, 3 insertions(+) diff --git a/contrib/tools/workflowcheck/go.mod b/contrib/tools/workflowcheck/go.mod index 3712b2a5b..c01a7ff07 100644 --- a/contrib/tools/workflowcheck/go.mod +++ b/contrib/tools/workflowcheck/go.mod @@ -11,6 +11,7 @@ require ( github.com/google/go-cmp v0.7.0 // indirect github.com/kr/pretty v0.1.0 // indirect github.com/kr/text v0.2.0 // indirect + go.temporal.io/api v1.62.11 // indirect golang.org/x/mod v0.30.0 // indirect golang.org/x/sync v0.19.0 // indirect gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect diff --git a/contrib/tools/workflowcheck/go.sum b/contrib/tools/workflowcheck/go.sum index 29a097436..1a5106624 100644 --- a/contrib/tools/workflowcheck/go.sum +++ b/contrib/tools/workflowcheck/go.sum @@ -7,6 +7,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +go.temporal.io/api v1.62.11 h1:MWDaooDvOJCIRb1atqeZX2ErDPNTsNc3/mMEVEvvaVU= +go.temporal.io/api v1.62.11/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM= golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= From 3a79139fbb43bc7b5441dfb84151fc36b499e3a2 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 08:34:46 -0700 Subject: [PATCH 06/12] Use real test version --- internal/cmd/build/main.go | 3 +- internal/internal_nexus_client.go | 34 +++++++++--- test/integration_test.go | 91 +++++++++++++++++++++++-------- 3 files changed, 95 insertions(+), 33 deletions(-) diff --git a/internal/cmd/build/main.go b/internal/cmd/build/main.go index 46525de79..ac202cee4 100644 --- a/internal/cmd/build/main.go +++ b/internal/cmd/build/main.go @@ -121,7 +121,7 @@ func (b *builder) integrationTest() error { if *devServerFlag { devServer, err := testsuite.StartDevServer(context.Background(), testsuite.DevServerOptions{ CachedDownload: testsuite.CachedDownload{ - Version: "v1.7.0", + Version: "v1.7.1-standalone-nexus-operations", }, ClientOptions: &client.Options{ HostPort: "127.0.0.1:7233", @@ -160,6 +160,7 @@ func (b *builder) integrationTest() error { "--dynamic-config-value", "history.enableTransitionHistory=true", "--dynamic-config-value", `component.nexusoperations.useSystemCallbackURL=false`, "--dynamic-config-value", `component.nexusoperations.callback.endpoint.template="http://localhost:7243/namespaces/{{.NamespaceName}}/nexus/callback"`, + "--dynamic-config-value", "nexusoperation.enableStandalone=true", "--dynamic-config-value", "frontend.ListWorkersEnabled=true", }, }) diff --git a/internal/internal_nexus_client.go b/internal/internal_nexus_client.go index 5bc78d565..32d228090 100644 --- a/internal/internal_nexus_client.go +++ b/internal/internal_nexus_client.go @@ -31,10 +31,20 @@ type ( // // Mandatory: No default. ID string - // ScheduleToCloseTimeout - Total time that the operation is allowed to run. + // ScheduleToCloseTimeout - The end to end timeout for the Nexus Operation. // - // Optional: Defaults to unlimited. + // Optional: defaults to the maximum allowed by the Temporal server. ScheduleToCloseTimeout time.Duration + // ScheduleToStartTimeout - Maximum time to wait for an operation to be started (or completed + // if synchronous) by the handler. + // + // Optional: If not set or zero, no schedule-to-start timeout is enforced. + ScheduleToStartTimeout time.Duration + // StartToCloseTimeout - Maximum time to wait for an asynchronous operation to complete after + // it has been started. Only applies to asynchronous operations. Ignored for synchronous operations. + // + // Optional: If not set or zero, no start-to-close timeout is enforced. + StartToCloseTimeout time.Duration // IDConflictPolicy - Defines how to resolve an operation id conflict with a running operation. // // Optional: Defaults to NEXUS_OPERATION_ID_CONFLICT_POLICY_FAIL. @@ -382,7 +392,7 @@ func (d *ClientNexusOperationExecutionDescription) GetLastAttemptFailure() error if failure == nil { return nil } - if err := visitProtoPayloads(context.Background(), d.inboundPayloadVisitor, failure); err != nil { + if err := visitProtoPayloads(context.Background(), d.inboundPayloadVisitor, failure, 0); err != nil { return err } return d.failureConverter.FailureToError(failure) @@ -396,7 +406,7 @@ func (c *ClientNexusOperationCancellationInfo) GetLastAttemptFailure() error { if c.lastAttemptFailure == nil { return nil } - if err := visitProtoPayloads(context.Background(), c.inboundPayloadVisitor, c.lastAttemptFailure); err != nil { + if err := visitProtoPayloads(context.Background(), c.inboundPayloadVisitor, c.lastAttemptFailure, 0); err != nil { return err } return c.failureConverter.FailureToError(c.lastAttemptFailure) @@ -487,10 +497,10 @@ func (h *clientNexusOperationHandleImpl) Get(ctx context.Context, valuePtr any) return resp.Error } if resp.Result != nil { + h.result = &ClientPollNexusOperationResultOutput{Result: resp.Result} if valuePtr == nil { return nil } - h.result = &ClientPollNexusOperationResultOutput{Result: resp.Result} return resp.Result.Get(valuePtr) } } @@ -700,7 +710,13 @@ func (w *workflowClientInterceptor) ExecuteNexusOperation( if in.Options.ScheduleToCloseTimeout > 0 { request.ScheduleToCloseTimeout = durationpb.New(in.Options.ScheduleToCloseTimeout) } - if err := visitProtoPayloads(ctx, w.client.outboundPayloadVisitor, request); err != nil { + if in.Options.ScheduleToStartTimeout > 0 { + request.ScheduleToStartTimeout = durationpb.New(in.Options.ScheduleToStartTimeout) + } + if in.Options.StartToCloseTimeout > 0 { + request.StartToCloseTimeout = durationpb.New(in.Options.StartToCloseTimeout) + } + if err := visitProtoPayloads(ctx, w.outboundPayloadVisitor, request, 0); err != nil { return nil, err } @@ -751,7 +767,7 @@ func (w *workflowClientInterceptor) PollNexusOperationResult( } } - if err := visitProtoPayloads(ctx, w.client.inboundPayloadVisitor, resp); err != nil { + if err := visitProtoPayloads(ctx, w.inboundPayloadVisitor, resp, 0); err != nil { return nil, err } @@ -801,7 +817,7 @@ func (w *workflowClientInterceptor) DescribeNexusOperation( Reason: info.CancellationInfo.Reason, lastAttemptFailure: info.CancellationInfo.LastAttemptFailure, failureConverter: w.client.failureConverter, - inboundPayloadVisitor: w.client.inboundPayloadVisitor, + inboundPayloadVisitor: w.inboundPayloadVisitor, } } @@ -837,7 +853,7 @@ func (w *workflowClientInterceptor) DescribeNexusOperation( CancellationInfo: cancellationInfo, dc: WithContext(ctx, w.client.dataConverter), failureConverter: w.client.failureConverter, - inboundPayloadVisitor: w.client.inboundPayloadVisitor, + inboundPayloadVisitor: w.inboundPayloadVisitor, }, }, nil } diff --git a/test/integration_test.go b/test/integration_test.go index 488f884b7..c8a401fbd 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -267,6 +267,9 @@ func (ts *IntegrationTestSuite) SetupTest() { ts.worker = worker.New(ts.client, ts.taskQueueName, options) ts.workerStopped = false ts.registerWorkflowsAndActivities(ts.worker) + if strings.Contains(ts.T().Name(), "TestExecuteNexusOperationSuite") { + ts.registerStandaloneNexusOperations(ts.worker) + } if strings.Contains(ts.T().Name(), "NoWorker") { // Don't even start the worker ts.workerStopped = true @@ -7158,6 +7161,26 @@ func (ts *IntegrationTestSuite) registerWorkflowsAndActivities(w worker.Worker) ts.activities.register(w) } +func (ts *IntegrationTestSuite) registerStandaloneNexusOperations(w worker.Worker) { + service := nexus.NewService("test-standalone-service") + syncOp := nexus.NewSyncOperation("echo-op", func(ctx context.Context, input string, opts nexus.StartOperationOptions) (string, error) { + return input, nil + }) + blockForeverWf := func(ctx workflow.Context, input string) (string, error) { + return "", workflow.Await(ctx, func() bool { return false }) + } + w.RegisterWorkflowWithOptions(blockForeverWf, workflow.RegisterOptions{Name: "block-forever-wf"}) + asyncOp := temporalnexus.NewWorkflowRunOperation( + "async-op", + blockForeverWf, + func(ctx context.Context, input string, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) { + return client.StartWorkflowOptions{ID: "nexus-async-" + uuid.NewString()}, nil + }, + ) + ts.NoError(service.Register(syncOp, asyncOp)) + w.RegisterNexusService(service) +} + var ( _ interceptor.WorkerInterceptor = (*tracingInterceptor)(nil) _ interceptor.WorkflowInboundInterceptor = (*tracingWorkflowInboundInterceptor)(nil) @@ -9163,7 +9186,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { // Create a Nexus endpoint targeting our task queue. endpoint := "sdk-go-nexus-standalone-test-ep-" + uuid.NewString() - _, err := ts.client.OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{ + createResp, err := ts.client.OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{ Spec: &nexuspb.EndpointSpec{ Name: endpoint, Target: &nexuspb.EndpointTarget{ @@ -9177,25 +9200,12 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { }, }) ts.NoError(err) - - // Register Nexus operations on the worker. - service := nexus.NewService("test-standalone-service") - syncOp := nexus.NewSyncOperation("echo-op", func(ctx context.Context, input string, opts nexus.StartOperationOptions) (string, error) { - return input, nil - }) - blockForeverWf := func(ctx workflow.Context, input string) (string, error) { - return "", workflow.Await(ctx, func() bool { return false }) - } - ts.worker.RegisterWorkflowWithOptions(blockForeverWf, workflow.RegisterOptions{Name: "block-forever-wf"}) - asyncOp := temporalnexus.NewWorkflowRunOperation( - "async-op", - blockForeverWf, - func(ctx context.Context, input string, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) { - return client.StartWorkflowOptions{ID: "nexus-async-" + uuid.NewString()}, nil - }, - ) - ts.NoError(service.Register(syncOp, asyncOp)) - ts.worker.RegisterNexusService(service) + defer func() { + _, _ = ts.client.OperatorService().DeleteNexusEndpoint(ctx, &operatorservice.DeleteNexusEndpointRequest{ + Id: createResp.Endpoint.Id, + Version: createResp.Endpoint.Version, + }) + }() nexusClient, err := ts.client.NewNexusClient(client.NexusClientOptions{ Endpoint: endpoint, @@ -9220,7 +9230,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { return handle } - ts.Run("Execute and Get result", func() { + ts.Run("Execute and get result", func() { input := "hello-nexus" handle := executeNexusOpWithRetry("echo-op", input, client.StartNexusOperationOptions{ ID: uuid.NewString(), @@ -9250,7 +9260,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.NotNil(description.RawInfo) }) - ts.Run("GetNexusOperationHandle", func() { + ts.Run("Get operation handle", func() { operationID := uuid.NewString() handle := executeNexusOpWithRetry("echo-op", "handle-test", client.StartNexusOperationOptions{ ID: operationID, @@ -9274,6 +9284,41 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Equal("handle-test", result) }) + ts.Run("Get operation handle without run ID gets latest", func() { + operationID := uuid.NewString() + + // Start the first operation and wait for it to complete. + handle1 := executeNexusOpWithRetry("echo-op", "first", client.StartNexusOperationOptions{ + ID: operationID, + ScheduleToCloseTimeout: 10 * time.Second, + }) + err := handle1.Get(ctx, nil) + ts.NoError(err) + + // Start a second operation with the same ID (allowed by ALLOW_DUPLICATE). + handle2 := executeNexusOpWithRetry("echo-op", "second", client.StartNexusOperationOptions{ + ID: operationID, + IDReusePolicy: enumspb.NEXUS_OPERATION_ID_REUSE_POLICY_ALLOW_DUPLICATE, + ScheduleToCloseTimeout: 10 * time.Second, + }) + err = handle2.Get(ctx, nil) + ts.NoError(err) + + // The two operations should have different run IDs. + ts.NotEqual(handle1.GetRunID(), handle2.GetRunID()) + + // Get a handle without specifying RunID — should resolve to the latest. + handle3 := ts.client.GetNexusOperationHandle(client.GetNexusOperationHandleOptions{ + OperationID: operationID, + }) + ts.Equal(operationID, handle3.GetID()) + + var result string + err = handle3.Get(ctx, &result) + ts.NoError(err) + ts.Equal("second", result) + }) + ts.Run("Cancel operation", func() { handle := executeNexusOpWithRetry("async-op", "cancel-test", client.StartNexusOperationOptions{ ID: uuid.NewString(), @@ -9333,7 +9378,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { }, 10*time.Second, 200*time.Millisecond, "timed out waiting for operations to appear in list") }) - ts.Run("NexusClient creation validation", func() { + ts.Run("Client creation validation", func() { _, err := ts.client.NewNexusClient(client.NexusClientOptions{}) ts.Error(err) ts.Contains(err.Error(), "endpoint is required") From 53295570616a10e05480a361b08970688d159ab2 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 08:37:22 -0700 Subject: [PATCH 07/12] Remove DISABLE_STANDALONE_NEXUS_TESTS for dev server test --- .github/workflows/ci.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 90a7ef30a..7153716ec 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,7 +78,6 @@ jobs: working-directory: ./internal/cmd/build env: WORKFLOW_CACHE_SIZE: "0" - DISABLE_STANDALONE_NEXUS_TESTS: "1" integration-test-with-cache: strategy: @@ -110,8 +109,6 @@ jobs: - name: Integration tests (with cache) run: go run . integration-test -dev-server working-directory: ./internal/cmd/build - env: - DISABLE_STANDALONE_NEXUS_TESTS: "1" docker-compose-test: runs-on: ubuntu-latest From 61324bbff0ae754e7a32109e92643396a2fd5475 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 09:03:56 -0700 Subject: [PATCH 08/12] Fix test failure --- test/nexus_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/nexus_test.go b/test/nexus_test.go index df1a1d6bc..7ab52c4a3 100644 --- a/test/nexus_test.go +++ b/test/nexus_test.go @@ -395,7 +395,7 @@ func TestNexusSyncOperation(t *testing.T) { t.Run("timeout", func(t *testing.T) { _, err := nexusclient.ExecuteOperation(ctx, nc, syncOp, "timeout", nexus.StartOperationOptions{ // Force shorter timeout to speed up the test and get a response back. - Header: nexus.Header{nexus.HeaderRequestTimeout: "300ms"}, + Header: nexus.Header{nexus.HeaderRequestTimeout: "2s"}, }) var handlerErr *nexus.HandlerError require.ErrorAs(t, err, &handlerErr) From 3e1b7fc2c5ba4580bb0aecb53671af5ff8b97441 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 12:40:13 -0700 Subject: [PATCH 09/12] Add async test --- test/integration_test.go | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/test/integration_test.go b/test/integration_test.go index c8a401fbd..c76c90067 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -7177,7 +7177,18 @@ func (ts *IntegrationTestSuite) registerStandaloneNexusOperations(w worker.Worke return client.StartWorkflowOptions{ID: "nexus-async-" + uuid.NewString()}, nil }, ) - ts.NoError(service.Register(syncOp, asyncOp)) + echoWf := func(ctx workflow.Context, input string) (string, error) { + return input, nil + } + w.RegisterWorkflowWithOptions(echoWf, workflow.RegisterOptions{Name: "echo-wf"}) + asyncEchoOp := temporalnexus.NewWorkflowRunOperation( + "async-echo-op", + echoWf, + func(ctx context.Context, input string, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) { + return client.StartWorkflowOptions{ID: "nexus-async-echo-" + uuid.NewString()}, nil + }, + ) + ts.NoError(service.Register(syncOp, asyncOp, asyncEchoOp)) w.RegisterNexusService(service) } @@ -9319,6 +9330,26 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Equal("second", result) }) + ts.Run("Async operation start and describe", func() { + input := "async-hello" + handle := executeNexusOpWithRetry("async-echo-op", input, client.StartNexusOperationOptions{ + ID: uuid.NewString(), + ScheduleToCloseTimeout: 30 * time.Second, + }) + ts.NotEmpty(handle.GetID()) + ts.NotEmpty(handle.GetRunID()) + + // Verify the operation can be described. + desc, err := handle.Describe(ctx, client.DescribeNexusOperationOptions{}) + ts.NoError(err) + ts.Equal(handle.GetID(), desc.OperationID) + ts.NotNil(desc.RawInfo) + + // TODO(nexus): Test Get() for async operations once the server supports + // completion callbacks for standalone nexus operations. Currently the server + // fails with "NamespaceID is empty" when processing the callback. + }) + ts.Run("Cancel operation", func() { handle := executeNexusOpWithRetry("async-op", "cancel-test", client.StartNexusOperationOptions{ ID: uuid.NewString(), From 5ec68ba9aad18c1b65167948e20ea5f57e5e9181 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Thu, 7 May 2026 13:18:46 -0700 Subject: [PATCH 10/12] Fix DC --- internal/cmd/build/main.go | 2 ++ test/integration_test.go | 13 ++++--------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/internal/cmd/build/main.go b/internal/cmd/build/main.go index ac202cee4..b01695c9c 100644 --- a/internal/cmd/build/main.go +++ b/internal/cmd/build/main.go @@ -161,6 +161,8 @@ func (b *builder) integrationTest() error { "--dynamic-config-value", `component.nexusoperations.useSystemCallbackURL=false`, "--dynamic-config-value", `component.nexusoperations.callback.endpoint.template="http://localhost:7243/namespaces/{{.NamespaceName}}/nexus/callback"`, "--dynamic-config-value", "nexusoperation.enableStandalone=true", + "--dynamic-config-value", "nexusoperation.enableChasm=true", + "--dynamic-config-value", "history.enableChasmCallbacks=true", "--dynamic-config-value", "frontend.ListWorkersEnabled=true", }, }) diff --git a/test/integration_test.go b/test/integration_test.go index c76c90067..6e17eba9b 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -9330,7 +9330,7 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.Equal("second", result) }) - ts.Run("Async operation start and describe", func() { + ts.Run("Async operation completes and returns result", func() { input := "async-hello" handle := executeNexusOpWithRetry("async-echo-op", input, client.StartNexusOperationOptions{ ID: uuid.NewString(), @@ -9339,15 +9339,10 @@ func (ts *IntegrationTestSuite) TestExecuteNexusOperationSuite() { ts.NotEmpty(handle.GetID()) ts.NotEmpty(handle.GetRunID()) - // Verify the operation can be described. - desc, err := handle.Describe(ctx, client.DescribeNexusOperationOptions{}) + var result string + err := handle.Get(ctx, &result) ts.NoError(err) - ts.Equal(handle.GetID(), desc.OperationID) - ts.NotNil(desc.RawInfo) - - // TODO(nexus): Test Get() for async operations once the server supports - // completion callbacks for standalone nexus operations. Currently the server - // fails with "NamespaceID is empty" when processing the callback. + ts.Equal(input, result) }) ts.Run("Cancel operation", func() { From 88e2e039a75dea92320ad85e26a06744f57d093e Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Fri, 8 May 2026 13:44:58 -0700 Subject: [PATCH 11/12] Fix CI --- internal/cmd/build/main.go | 1 - test/nexus_test.go | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/internal/cmd/build/main.go b/internal/cmd/build/main.go index b01695c9c..021f023c9 100644 --- a/internal/cmd/build/main.go +++ b/internal/cmd/build/main.go @@ -161,7 +161,6 @@ func (b *builder) integrationTest() error { "--dynamic-config-value", `component.nexusoperations.useSystemCallbackURL=false`, "--dynamic-config-value", `component.nexusoperations.callback.endpoint.template="http://localhost:7243/namespaces/{{.NamespaceName}}/nexus/callback"`, "--dynamic-config-value", "nexusoperation.enableStandalone=true", - "--dynamic-config-value", "nexusoperation.enableChasm=true", "--dynamic-config-value", "history.enableChasmCallbacks=true", "--dynamic-config-value", "frontend.ListWorkersEnabled=true", }, diff --git a/test/nexus_test.go b/test/nexus_test.go index 7ab52c4a3..daad144d3 100644 --- a/test/nexus_test.go +++ b/test/nexus_test.go @@ -395,7 +395,7 @@ func TestNexusSyncOperation(t *testing.T) { t.Run("timeout", func(t *testing.T) { _, err := nexusclient.ExecuteOperation(ctx, nc, syncOp, "timeout", nexus.StartOperationOptions{ // Force shorter timeout to speed up the test and get a response back. - Header: nexus.Header{nexus.HeaderRequestTimeout: "2s"}, + Header: nexus.Header{nexus.HeaderRequestTimeout: "5s"}, }) var handlerErr *nexus.HandlerError require.ErrorAs(t, err, &handlerErr) From 8408739bd0308bcf9479163acccc9c847b0f3597 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Tue, 19 May 2026 09:07:52 -0700 Subject: [PATCH 12/12] Respond to PR comments --- internal/internal_activity_client.go | 4 ++-- internal/internal_nexus_client.go | 22 +++++++++------------- workflow/workflow.go | 1 + 3 files changed, 12 insertions(+), 15 deletions(-) diff --git a/internal/internal_activity_client.go b/internal/internal_activity_client.go index 46667f608..215ab19f8 100644 --- a/internal/internal_activity_client.go +++ b/internal/internal_activity_client.go @@ -31,11 +31,11 @@ type ( ClientStartActivityOptions struct { // ID - The business identifier of the activity. // - // Mandatory: No default. + // Required ID string // TaskQueue - The task queue to schedule the activity on. // - // Mandatory: No default. + // Required TaskQueue string // ScheduleToCloseTimeout - Total time that a workflow is willing to wait for an Activity to complete. // ScheduleToCloseTimeout limits the total time of an Activity's execution including retries diff --git a/internal/internal_nexus_client.go b/internal/internal_nexus_client.go index 32d228090..3efe60195 100644 --- a/internal/internal_nexus_client.go +++ b/internal/internal_nexus_client.go @@ -29,7 +29,7 @@ type ( ClientStartNexusOperationOptions struct { // ID - The business identifier of the operation. // - // Mandatory: No default. + // Required ID string // ScheduleToCloseTimeout - The end to end timeout for the Nexus Operation. // @@ -71,11 +71,11 @@ type ( ClientNexusClientOptions struct { // Endpoint - The Nexus endpoint name. // - // Mandatory: No default. + // Required Endpoint string // Service - The Nexus service name. // - // Mandatory: No default. + // Required Service string } @@ -87,7 +87,7 @@ type ( ClientGetNexusOperationHandleOptions struct { // OperationID - The operation ID. // - // Mandatory: No default. + // Required OperationID string // RunID - The run ID. Can be empty to target the latest run. // @@ -357,8 +357,8 @@ type ( ) var ( - _ ClientNexusClient = &nexusClientImpl{} - _ ClientNexusOperationHandle = &clientNexusOperationHandleImpl{} + _ ClientNexusClient = (*nexusClientImpl)(nil) + _ ClientNexusOperationHandle = (*clientNexusOperationHandleImpl)(nil) ) // GetSummary returns summary of the operation. See ClientStartNexusOperationOptions.Summary. @@ -674,13 +674,9 @@ func (w *workflowClientInterceptor) ExecuteNexusOperation( } // Encode input as a single Payload (not Payloads) - var inputPayload *commonpb.Payload - if in.Input != nil { - var err error - inputPayload, err = dataConverter.ToPayload(in.Input) - if err != nil { - return nil, err - } + inputPayload, err := dataConverter.ToPayload(in.Input) + if err != nil { + return nil, err } searchAttrs, err := serializeTypedSearchAttributes(in.Options.SearchAttributes.GetUntypedValues()) diff --git a/workflow/workflow.go b/workflow/workflow.go index 575ff1594..bd14a194d 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -186,6 +186,7 @@ type ( // NOTE to maintainers, this interface definition is duplicated in the internal package to provide a better UX. // NexusClient is a client for executing Nexus Operations from a workflow. + // For Nexus operations outside workflows, use client.NexusClient instead. NexusClient interface { // The endpoint name this client uses. Endpoint() string