hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
release/v1.27.x
tasks/tests/nexus_workflow_test.go
T
pdoernerandStephan Behnke 2ffaae839b Deserialize failure details from CompleteOperation HTTP response (#7497) 
## What changed?
Added logic to read failure details from the response to
CompleteOperation calls in the Nexus callback task executor.
Adjusted the frontend logic for the CompleteOperation handler to return
more details for serviceerror.NotFound errors

## Why?
To provide a better debugging experience, especially when callbacks are
failing because the caller workflow has already completed.

## How did you test it?
New functional test
2025-03-25 15:43:26 -07:00

2296 lines
81 KiB
Go
Raw Permalink Blame History

// The MIT License
//
// Copyright (c) 2024 Temporal Technologies Inc. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package tests
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"slices"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/nexus-rpc/sdk-go/nexus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
commandpb "go.temporal.io/api/command/v1"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
failurepb "go.temporal.io/api/failure/v1"
historypb "go.temporal.io/api/history/v1"
nexuspb "go.temporal.io/api/nexus/v1"
"go.temporal.io/api/operatorservice/v1"
taskqueuepb "go.temporal.io/api/taskqueue/v1"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/temporalnexus"
"go.temporal.io/sdk/worker"
"go.temporal.io/sdk/workflow"
"go.temporal.io/server/api/adminservice/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/authorization"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/metrics/metricstest"
commonnexus "go.temporal.io/server/common/nexus"
"go.temporal.io/server/common/nexus/nexustest"
"go.temporal.io/server/common/testing/protorequire"
"go.temporal.io/server/components/nexusoperations"
"go.temporal.io/server/service/frontend/configs"
"go.temporal.io/server/tests/testcore"
"google.golang.org/protobuf/types/known/durationpb"
)
type NexusWorkflowTestSuite struct {
NexusTestBaseSuite
}
func TestNexusWorkflowTestSuite(t *testing.T) {
t.Parallel()
suite.Run(t, new(NexusWorkflowTestSuite))
}
func (s *NexusWorkflowTestSuite) TestNexusOperationCancelation() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
if service != "service" {
return nil, nexus.HandlerErrorf(nexus.HandlerErrorTypeBadRequest, `expected service to equal "service"`)
}
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
},
OnCancelOperation: func(ctx context.Context, service, operation, token string, options nexus.CancelOperationOptions) error {
return nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
WorkflowTaskTimeout: time.Second,
}, "workflow")
s.NoError(err)
s.EventuallyWithT(func(t *assert.CollectT) {
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
assert.NoError(t, err)
if err != nil {
return
}
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
assert.NoError(t, err)
}, time.Second*20, time.Millisecond*200)
// Poll and wait for the "started" event to be recorded.
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Greater(startedEventIdx, 0)
// Get the scheduleEventId to issue the cancel command.
scheduledEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationScheduledEventAttributes() != nil
})
s.Greater(scheduledEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION,
Attributes: &commandpb.Command_RequestCancelNexusOperationCommandAttributes{
RequestCancelNexusOperationCommandAttributes: &commandpb.RequestCancelNexusOperationCommandAttributes{
ScheduledEventId: pollResp.History.Events[scheduledEventIdx].EventId,
},
},
},
},
})
s.NoError(err)
// Poll and wait for the cancelation request to go through.
s.EventuallyWithT(func(t *assert.CollectT) {
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, run.GetID(), run.GetRunID())
assert.NoError(t, err)
if err != nil {
return
}
assert.Equal(t, 1, len(desc.PendingNexusOperations))
if len(desc.PendingNexusOperations) < 1 {
return
}
op := desc.PendingNexusOperations[0]
assert.Equal(t, pollResp.History.Events[scheduledEventIdx].EventId, op.ScheduledEventId)
assert.Equal(t, endpointName, op.Endpoint)
assert.Equal(t, "service", op.Service)
assert.Equal(t, "operation", op.Operation)
assert.Equal(t, enumspb.PENDING_NEXUS_OPERATION_STATE_STARTED, op.State)
assert.Equal(t, enumspb.NEXUS_OPERATION_CANCELLATION_STATE_SUCCEEDED, op.CancellationInfo.State)
}, time.Second*10, time.Millisecond*30)
err = s.SdkClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "test")
s.NoError(err)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationSyncCompletion() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
handlerLink := &commonpb.Link_WorkflowEvent{
Namespace: "handler-ns",
WorkflowId: "handler-wf-id",
RunId: "handler-run-id",
Reference: &commonpb.Link_WorkflowEvent_EventRef{
EventRef: &commonpb.Link_WorkflowEvent_EventReference{
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
},
},
}
handlerNexusLink := temporalnexus.ConvertLinkWorkflowEventToNexusLink(handlerLink)
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
return &nexus.HandlerStartOperationResultSync[any]{Value: "result", Links: []nexus.Link{handlerNexusLink}}, nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
s.EventuallyWithT(func(t *assert.CollectT) {
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
assert.NoError(t, err)
if err != nil {
return
}
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
assert.NoError(t, err)
}, time.Second*20, time.Millisecond*200)
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationCompletedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
s.Len(pollResp.History.Events[completedEventIdx].GetLinks(), 1)
protorequire.ProtoEqual(s.T(), handlerLink, pollResp.History.Events[completedEventIdx].GetLinks()[0].GetWorkflowEvent())
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
},
},
},
},
},
},
})
s.NoError(err)
var result string
s.NoError(run.Get(ctx, &result))
s.Equal("result", result)
// Use this test case to verify that the state machine is actually deleted, the workflowservice
// DescribeWorkflowExecution API filters out operations in terminal state in case they complete in a server version
// without state machine deletion enabled, hence the use of the adminservice API here.
desc, err := s.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{
Namespace: s.Namespace().String(),
Execution: &commonpb.WorkflowExecution{
WorkflowId: run.GetID(),
},
})
s.NoError(err)
s.Len(desc.DatabaseMutableState.GetExecutionInfo().SubStateMachinesByType, 0)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationSyncCompletion_LargePayload() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
// Use -10 to avoid hitting MaxNexusAPIRequestBodyBytes. Actual payload will still exceed limit because of
// additional Content headers. See common/rpc/grpc.go:66
return &nexus.HandlerStartOperationResultSync[any]{Value: strings.Repeat("a", (2*1024*1024)-10)}, nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
s.EventuallyWithT(func(t *assert.CollectT) {
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
assert.NoError(t, err)
if err != nil {
return
}
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
assert.NoError(t, err)
}, time.Second*20, time.Millisecond*200)
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
failedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationFailedEventAttributes() != nil
})
s.Greater(failedEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
s.mustToPayload(pollResp.History.Events[failedEventIdx].GetNexusOperationFailedEventAttributes().Failure.Cause.Message),
},
},
},
},
},
},
})
s.NoError(err)
var result string
s.NoError(run.Get(ctx, &result))
s.Equal("http: response body too large", result)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletion() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
testClusterInfo, err := s.FrontendClient().GetClusterInfo(ctx, &workflowservice.GetClusterInfoRequest{})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
var callbackToken, publicCallbackUrl string
handlerLink := &commonpb.Link_WorkflowEvent{
Namespace: "handler-ns",
WorkflowId: "handler-wf-id",
RunId: "handler-run-id",
Reference: &commonpb.Link_WorkflowEvent_EventRef{
EventRef: &commonpb.Link_WorkflowEvent_EventReference{
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
},
},
}
handlerNexusLink := temporalnexus.ConvertLinkWorkflowEventToNexusLink(handlerLink)
h := nexustest.Handler{
OnStartOperation: func(
ctx context.Context,
service, operation string,
input *nexus.LazyValue,
options nexus.StartOperationOptions,
) (nexus.HandlerStartOperationResult[any], error) {
s.Equal(testClusterInfo.GetClusterId(), options.CallbackHeader.Get("source"))
s.Len(options.Links, 1)
var links []*commonpb.Link
for _, nexusLink := range options.Links {
link, err := temporalnexus.ConvertNexusLinkToLinkWorkflowEvent(nexusLink)
s.NoError(err)
links = append(links, &commonpb.Link{
Variant: &commonpb.Link_WorkflowEvent_{
WorkflowEvent: link,
},
})
}
s.NotNil(links[0].GetWorkflowEvent())
protorequire.ProtoEqual(s.T(), &commonpb.Link_WorkflowEvent{
Namespace: s.Namespace().String(),
WorkflowId: run.GetID(),
RunId: run.GetRunID(),
Reference: &commonpb.Link_WorkflowEvent_EventRef{
EventRef: &commonpb.Link_WorkflowEvent_EventReference{
// Event history:
// 1 WORKFLOW_EXECUTION_STARTED
// 2 WORKFLOW_TASK_SCHEDULED
// 3 WORKFLOW_TASK_STARTED
// 4 WORKFLOW_TASK_COMPLETED
// 5 NEXUS_OPERATION_SCHEDULED
EventId: 5,
EventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED,
},
},
}, links[0].GetWorkflowEvent())
callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader)
publicCallbackUrl = options.CallbackURL
return &nexus.HandlerStartOperationResultAsync{
OperationToken: "test",
Links: []nexus.Link{handlerNexusLink},
}, nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err = s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
s.NoError(err)
// Poll and verify that the "started" event was recorded.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
})
s.NoError(err)
// Remember the workflow task completed event ID (next after the last WFT started), we'll use it to test reset
// below.
wftCompletedEventID := int64(len(pollResp.History.Events))
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Greater(startedEventIdx, 0)
s.Len(pollResp.History.Events[startedEventIdx].Links, 1)
l := pollResp.History.Events[startedEventIdx].Links[0].GetWorkflowEvent()
protorequire.ProtoEqual(s.T(), handlerLink, l)
// Completion request fails if the result payload is too large.
largeCompletion, err := nexus.NewOperationCompletionSuccessful(
// Use -10 to avoid hitting MaxNexusAPIRequestBodyBytes. Actual payload will still exceed limit because of
// additional Content headers. See common/rpc/grpc.go:66
s.mustToPayload(strings.Repeat("a", (2*1024*1024)-10)),
nexus.OperationCompletionSuccessfulOptions{Serializer: commonnexus.PayloadSerializer},
)
s.NoError(err)
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, largeCompletion, callbackToken)
s.Equal(http.StatusBadRequest, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_bad_request"})
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload(nil), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
})
s.NoError(err)
invalidNamespace := testcore.RandomizeStr("ns")
_, err = s.FrontendClient().RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{
Namespace: invalidNamespace,
WorkflowExecutionRetentionPeriod: durationpb.New(time.Hour * 24),
})
s.NoError(err)
// Send an invalid completion request and verify that we get an error that the namespace in the URL doesn't match the namespace in the token.
invalidCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(invalidNamespace)
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), invalidCallbackUrl, completion, callbackToken)
s.Equal(http.StatusBadRequest, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": invalidNamespace, "outcome": "error_bad_request"})
// Manipulate the token to verify we get the expected errors in the API.
gen := &commonnexus.CallbackTokenGenerator{}
decodedToken, err := commonnexus.DecodeCallbackToken(callbackToken)
s.NoError(err)
completionToken, err := gen.DecodeCompletion(decodedToken)
s.NoError(err)
// Request fails if the workflow is not found.
workflowNotFoundToken := common.CloneProto(completionToken)
workflowNotFoundToken.WorkflowId = "not-found"
callbackToken, err = gen.Tokenize(workflowNotFoundToken)
s.NoError(err)
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusNotFound, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_not_found"})
// Request fails if the state machine reference is stale.
staleToken := common.CloneProto(completionToken)
staleToken.Ref.MachineInitialVersionedTransition.NamespaceFailoverVersion++
callbackToken, err = gen.Tokenize(staleToken)
s.NoError(err)
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusNotFound, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_not_found"})
// Send a valid - successful completion request.
completion, err = nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
})
s.NoError(err)
callbackToken, err = gen.Tokenize(completionToken)
s.NoError(err)
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusOK, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "success"})
// Ensure that CompleteOperation request is tracked as part of normal service telemetry metrics
idx := slices.IndexFunc(snap["service_requests"], func(m *metricstest.CapturedRecording) bool {
opTag, ok := m.Tags["operation"]
return ok && opTag == "CompleteNexusOperation"
})
s.Greater(idx, -1)
// Resend the request and verify we get a not found error since the operation has already completed.
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusNotFound, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_not_found"})
// Poll again and verify the completion is recorded and triggers workflow progress.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationCompletedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
},
},
},
},
},
},
})
s.NoError(err)
var result string
s.NoError(run.Get(ctx, &result))
s.Equal("result", result)
// Reset the workflow and check that the completion event has been reapplied.
resp, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowExecution: pollResp.WorkflowExecution,
Reason: "test",
RequestId: uuid.NewString(),
WorkflowTaskFinishEventId: wftCompletedEventID,
})
s.NoError(err)
hist := s.SdkClient().GetWorkflowHistory(ctx, run.GetID(), resp.RunId, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
seenCompletedEvent := false
for hist.HasNext() {
event, err := hist.Next()
s.NoError(err)
if event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED {
seenCompletedEvent = true
break
}
}
s.True(seenCompletedEvent)
// Reset the workflow again to the same point with enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS option
// and verify that the completion event has been excluded.
resp, err = s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowExecution: pollResp.WorkflowExecution,
Reason: "test",
RequestId: uuid.NewString(),
WorkflowTaskFinishEventId: wftCompletedEventID,
ResetReapplyExcludeTypes: []enumspb.ResetReapplyExcludeType{enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS},
})
s.NoError(err)
hist = s.SdkClient().GetWorkflowHistory(ctx, run.GetID(), resp.RunId, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
seenCompletedEvent = false
for hist.HasNext() {
event, err := hist.Next()
s.NoError(err)
if event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED {
seenCompletedEvent = true
break
}
}
s.False(seenCompletedEvent)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: taskQueue,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
completionWFType := "completion_wf"
completionWFID := testcore.RandomizeStr(s.T().Name())
completionWFTaskQueue := testcore.RandomizeStr(s.T().Name())
completionWFStartReq := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: completionWFID,
WorkflowType: &commonpb.WorkflowType{Name: completionWFType},
TaskQueue: &taskqueuepb.TaskQueue{Name: completionWFTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
Input: nil,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
Identity: "test",
}
startLink := &commonpb.Link_WorkflowEvent{
Namespace: s.Namespace().String(),
WorkflowId: completionWFID,
Reference: &commonpb.Link_WorkflowEvent_EventRef{
EventRef: &commonpb.Link_WorkflowEvent_EventReference{
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
},
},
}
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "test-service",
Operation: "my-operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
s.NoError(err)
// Poll for the Nexus task
res, err := s.FrontendClient().PollNexusTaskQueue(ctx, &workflowservice.PollNexusTaskQueueRequest{
Namespace: s.Namespace().String(),
Identity: uuid.NewString(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
})
s.NoError(err)
start := res.Request.Variant.(*nexuspb.Request_StartOperation).StartOperation
s.Equal(op.Name(), start.Operation)
start.CallbackHeader[nexus.HeaderOperationToken] = completionWFID
completionWFStartReq.CompletionCallbacks = []*commonpb.Callback{
{
Variant: &commonpb.Callback_Nexus_{
Nexus: &commonpb.Callback_Nexus{
Url: start.Callback,
Header: start.CallbackHeader,
},
},
},
}
completionRun, err := s.FrontendClient().StartWorkflowExecution(ctx, completionWFStartReq)
s.NoError(err)
startLink.RunId = completionRun.RunId
// Complete workflow containing callback
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: completionWFTaskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
s.mustToPayload("result"),
},
},
},
},
},
},
})
s.NoError(err)
// Poll and verify the fabricated start event and completion event are recorded and triggers workflow progress.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Len(pollResp.History.Events[startedEventIdx].Links, 1)
startedEvent := pollResp.History.Events[startedEventIdx].GetNexusOperationStartedEventAttributes()
s.Equal(completionWFID, startedEvent.OperationToken)
l := pollResp.History.Events[startedEventIdx].Links[0].GetWorkflowEvent()
protorequire.ProtoEqual(s.T(), startLink, l)
s.Greater(startedEventIdx, 0)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationCompletedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
// Complete start request to verify response is ignored.
_, err = s.FrontendClient().RespondNexusTaskCompleted(ctx, &workflowservice.RespondNexusTaskCompletedRequest{
Namespace: s.Namespace().String(),
Identity: uuid.NewString(),
TaskToken: res.TaskToken,
Response: &nexuspb.Response{
Variant: &nexuspb.Response_StartOperation{
StartOperation: &nexuspb.StartOperationResponse{
Variant: &nexuspb.StartOperationResponse_AsyncSuccess{
AsyncSuccess: &nexuspb.StartOperationResponse_Async{
OperationToken: completionWFID,
},
},
},
},
},
})
s.NoErrorf(err, "Duplicate start response should be ignored.")
// Complete caller workflow
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
},
},
},
},
},
},
})
s.NoError(err)
var result string
s.NoError(run.Get(ctx, &result))
s.Equal("result", result)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncFailure() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
var callbackToken, publicCallbackUrl string
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader)
publicCallbackUrl = options.CallbackURL
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
s.EventuallyWithT(func(t *assert.CollectT) {
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
assert.NoError(t, err)
if err != nil {
return
}
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
assert.NoError(t, err)
}, time.Second*20, time.Millisecond*200)
// Poll and verify that the "started" event was recorded.
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
})
s.NoError(err)
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Greater(startedEventIdx, 0)
// Send a valid - failed completion request.
completion, err := nexus.NewOperationCompletionUnsuccessful(nexus.NewFailedOperationError(errors.New("test operation failed")), nexus.OperationCompletionUnsuccessfulOptions{})
s.NoError(err)
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusOK, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "success"})
// Poll again and verify the completion is recorded and triggers workflow progress.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationFailedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{
FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
Failure: pollResp.History.Events[completedEventIdx].GetNexusOperationFailedEventAttributes().Failure,
},
},
},
},
})
s.NoError(err)
var result string
err = run.Get(ctx, &result)
var wee *temporal.WorkflowExecutionError
s.ErrorAs(err, &wee)
s.True(strings.HasPrefix(wee.Unwrap().Error(), "nexus operation completed unsuccessfully"))
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionErrors() {
ctx := testcore.NewContext()
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
})
s.NoError(err)
s.Run("ConfigDisabled", func() {
s.OverrideDynamicConfig(dynamicconfig.EnableNexus, false)
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(s.Namespace().String())
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, "")
s.Equal(http.StatusNotFound, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_request_preprocess_errors"]))
})
s.Run("NamespaceNotFound", func() {
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path("namespace-doesnt-exist")
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, "")
s.Equal(http.StatusNotFound, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_request_preprocess_errors"]))
})
s.Run("OperationTokenTooLong", func() {
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(s.Namespace().String())
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
OperationToken: strings.Repeat("long", 2000),
})
s.NoError(err)
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, "")
s.Equal(http.StatusBadRequest, res.StatusCode)
s.Equal(0, len(snap["nexus_completion_request_preprocess_errors"]))
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_bad_request"})
})
s.Run("InvalidCallbackToken", func() {
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(s.Namespace().String())
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, "")
s.Equal(http.StatusBadRequest, res.StatusCode)
s.Equal(0, len(snap["nexus_completion_request_preprocess_errors"]))
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_bad_request"})
})
s.Run("InvalidClientVersion", func() {
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(s.Namespace().String())
capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture()
defer s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
req, err := nexus.NewCompletionHTTPRequest(ctx, publicCallbackUrl, completion)
s.NoError(err)
req.Header.Set("User-Agent", "Nexus-go-sdk/v99.0.0")
res, err := http.DefaultClient.Do(req)
s.NoError(err)
_, err = io.ReadAll(res.Body)
s.NoError(err)
defer res.Body.Close()
snap := capture.Snapshot()
s.Equal(http.StatusBadRequest, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "unsupported_client"})
})
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionAuthErrors() {
ctx := testcore.NewContext()
onAuthorize := func(ctx context.Context, c *authorization.Claims, ct *authorization.CallTarget) (authorization.Result, error) {
if ct.APIName == configs.CompleteNexusOperation {
return authorization.Result{Decision: authorization.DecisionDeny, Reason: "unauthorized in test"}, nil
}
return authorization.Result{Decision: authorization.DecisionAllow}, nil
}
s.GetTestCluster().Host().SetOnAuthorize(onAuthorize)
defer s.GetTestCluster().Host().SetOnAuthorize(nil)
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
})
s.NoError(err)
publicCallbackUrl := "http://" + s.HttpAPIAddress() + "/" + commonnexus.RouteCompletionCallback.Path(s.Namespace().String())
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, "")
s.Equal(http.StatusForbidden, res.StatusCode)
s.Equal(1, len(snap["nexus_completion_requests"]))
s.Subset(snap["nexus_completion_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "outcome": "error_unauthorized"})
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionInternalAuth() {
// Set URL template with invalid host
s.OverrideDynamicConfig(
nexusoperations.CallbackURLTemplate,
"http://INTERNAL/namespaces/{{.NamespaceName}}/nexus/callback")
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: taskQueue,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
completionWFType := "completion_wf"
completionWFTaskQueue := testcore.RandomizeStr(s.T().Name())
completionWFStartReq := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: testcore.RandomizeStr(s.T().Name()),
WorkflowType: &commonpb.WorkflowType{Name: completionWFType},
TaskQueue: &taskqueuepb.TaskQueue{Name: completionWFTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
Input: nil,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
Identity: "test",
}
go s.nexusTaskPoller(ctx, taskQueue, func(res *workflowservice.PollNexusTaskQueueResponse) (*nexuspb.Response, *nexuspb.HandlerError) {
start := res.Request.Variant.(*nexuspb.Request_StartOperation).StartOperation
s.Equal(op.Name(), start.Operation)
completionWFStartReq.CompletionCallbacks = []*commonpb.Callback{
{
Variant: &commonpb.Callback_Nexus_{
Nexus: &commonpb.Callback_Nexus{
Url: start.Callback,
Header: start.CallbackHeader,
},
},
},
}
_, err := s.FrontendClient().StartWorkflowExecution(ctx, completionWFStartReq)
s.NoError(err)
return &nexuspb.Response{
Variant: &nexuspb.Response_StartOperation{
StartOperation: &nexuspb.StartOperationResponse{
Variant: &nexuspb.StartOperationResponse_AsyncSuccess{
AsyncSuccess: &nexuspb.StartOperationResponse_Async{
OperationToken: "test-token",
},
},
},
},
}, nil
})
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "test-service",
Operation: "my-operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
s.NoError(err)
// Poll and verify that the "started" event was recorded.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
})
s.NoError(err)
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Greater(startedEventIdx, 0)
// Complete workflow containing callback
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: completionWFTaskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
s.mustToPayload("result"),
},
},
},
},
},
},
})
s.NoError(err)
// Poll again and verify the completion is recorded and triggers workflow progress.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationCompletedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
},
},
},
},
},
},
})
s.NoError(err)
var result string
s.NoError(run.Get(ctx, &result))
s.Equal("result", result)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationCancelBeforeStarted_CancelationEventuallyDelivered() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
canStartCh := make(chan struct{})
cancelSentCh := make(chan struct{})
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
select {
case <-canStartCh:
case <-ctx.Done():
return nil, ctx.Err()
}
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
},
OnCancelOperation: func(ctx context.Context, service, operation, token string, options nexus.CancelOperationOptions) error {
cancelSentCh <- struct{}{}
return nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
// Also wake the workflow up so it can cancel the operation.
{
CommandType: enumspb.COMMAND_TYPE_START_TIMER,
Attributes: &commandpb.Command_StartTimerCommandAttributes{
StartTimerCommandAttributes: &commandpb.StartTimerCommandAttributes{
TimerId: "1",
StartToFireTimeout: durationpb.New(time.Millisecond),
},
},
},
},
})
s.NoError(err)
// Poll and cancel the operation.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
// Get the scheduleEventId to issue the cancel command.
scheduledEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationScheduledEventAttributes() != nil
})
s.Greater(scheduledEventIdx, 0)
scheduledEventID := pollResp.History.Events[scheduledEventIdx].EventId
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION,
Attributes: &commandpb.Command_RequestCancelNexusOperationCommandAttributes{
RequestCancelNexusOperationCommandAttributes: &commandpb.RequestCancelNexusOperationCommandAttributes{
ScheduledEventId: scheduledEventID,
},
},
},
},
})
s.NoError(err)
canStartCh <- struct{}{}
s.WaitForChannel(ctx, cancelSentCh)
// Terminate the workflow for good measure.
err = s.SdkClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "test")
s.NoError(err)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionAfterReset() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
var callbackToken, publicCallbackUrl string
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader)
publicCallbackUrl = options.CallbackURL
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, "workflow")
s.NoError(err)
pollResp, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
Endpoint: endpointName,
Service: "service",
Operation: "operation",
Input: s.mustToPayload("input"),
},
},
},
},
})
s.NoError(err)
// Poll and verify that the "started" event was recorded.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
})
s.NoError(err)
startedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationStartedEventAttributes() != nil
})
s.Greater(startedEventIdx, 0)
// Remember the workflow task completed event ID (next after the last WFT started), we'll use it to test reset
// below.
wftCompletedEventID := int64(len(pollResp.History.Events))
// Reset the workflow and check that the started event has been reapplied.
resetResp, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowExecution: pollResp.WorkflowExecution,
Reason: "test",
RequestId: uuid.NewString(),
WorkflowTaskFinishEventId: wftCompletedEventID,
})
s.NoError(err)
hist := s.SdkClient().GetWorkflowHistory(ctx, run.GetID(), resetResp.RunId, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
seenStartedEvent := false
for hist.HasNext() {
event, err := hist.Next()
s.NoError(err)
if event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED {
seenStartedEvent = true
}
}
s.True(seenStartedEvent)
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
Serializer: commonnexus.PayloadSerializer,
})
s.NoError(err)
res, _ := s.sendNexusCompletionRequest(ctx, s.T(), publicCallbackUrl, completion, callbackToken)
s.Equal(http.StatusOK, res.StatusCode)
// Poll again and verify the completion is recorded and triggers workflow progress.
pollResp, err = s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: taskQueue,
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
},
Identity: "test",
})
s.NoError(err)
completedEventIdx := slices.IndexFunc(pollResp.History.Events, func(e *historypb.HistoryEvent) bool {
return e.GetNexusOperationCompletedEventAttributes() != nil
})
s.Greater(completedEventIdx, 0)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Identity: "test",
TaskToken: pollResp.TaskToken,
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: &commonpb.Payloads{
Payloads: []*commonpb.Payload{
pollResp.History.Events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
},
},
},
},
},
},
})
s.NoError(err)
var result string
run = s.SdkClient().GetWorkflow(ctx, run.GetID(), resetResp.RunId)
s.NoError(run.Get(ctx, &result))
s.Equal("result", result)
}
func (s *NexusWorkflowTestSuite) TestNexusAsyncOperationWithNilIO() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
defer cancel()
callerTaskQueue := testcore.RandomizeStr("caller_" + s.T().Name())
handlerWorkflowTaskQueue := testcore.RandomizeStr("handler_" + s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
handlerWorkflowID := testcore.RandomizeStr(s.T().Name())
_, err := s.SdkClient().OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: callerTaskQueue,
},
},
},
},
})
s.NoError(err)
w := worker.New(
s.SdkClient(),
callerTaskQueue,
worker.Options{},
)
svc := nexus.NewService("test")
wf := func(ctx workflow.Context, input nexus.NoValue) (nexus.NoValue, error) {
return nil, nil
}
op := temporalnexus.NewWorkflowRunOperation("op", wf, func(ctx context.Context, nv nexus.NoValue, opts nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
return client.StartWorkflowOptions{
ID: handlerWorkflowID,
TaskQueue: handlerWorkflowTaskQueue,
}, nil
})
svc.Register(op)
callerWF := func(ctx workflow.Context, input nexus.NoValue) (nexus.NoValue, error) {
c := workflow.NewNexusClient(endpointName, svc.Name)
fut := c.ExecuteOperation(ctx, op, nil, workflow.NexusOperationOptions{})
var opExec workflow.NexusOperationExecution
err := fut.GetNexusOperationExecution().Get(ctx, &opExec)
s.NoError(err)
s.NotEmpty(opExec.OperationToken)
return nil, fut.Get(ctx, nil)
}
w.RegisterNexusService(svc)
w.RegisterWorkflow(wf)
w.RegisterWorkflow(callerWF)
w.Start()
defer w.Stop()
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: callerTaskQueue,
}, callerWF, nil)
s.NoError(err)
pollRes, err := s.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
Namespace: s.Namespace().String(),
TaskQueue: &taskqueuepb.TaskQueue{
Name: handlerWorkflowTaskQueue,
},
Identity: "test",
})
s.NoError(err)
_, err = s.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{
Namespace: s.Namespace().String(),
TaskToken: pollRes.TaskToken,
Identity: "test",
Commands: []*commandpb.Command{
{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: nil, // Return nil here to verify that the conversion to nexus content works as expected.
},
},
},
},
})
s.NoError(err)
s.NoError(run.Get(ctx, nil))
history := s.SdkClient().GetWorkflowHistory(ctx, run.GetID(), "", false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
for history.HasNext() {
ev, err := history.Next()
s.NoError(err)
if attr := ev.GetNexusOperationCompletedEventAttributes(); attr != nil {
protorequire.ProtoEqual(s.T(), s.mustToPayload(nil), attr.GetResult())
break
}
}
}
func (s *NexusWorkflowTestSuite) TestNexusSyncOperationErrorRehydration() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
defer cancel()
taskQueue := testcore.RandomizeStr("caller_" + s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
converter := temporal.NewDefaultFailureConverter(temporal.DefaultFailureConverterOptions{})
_, err := s.SdkClient().OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: taskQueue,
},
},
},
},
})
s.NoError(err)
w := worker.New(
s.SdkClient(),
taskQueue,
worker.Options{},
)
svc := nexus.NewService("test")
op := temporalnexus.NewSyncOperation("op", func(ctx context.Context, c client.Client, outcome string, soo nexus.StartOperationOptions) (nexus.NoValue, error) {
switch outcome {
case "fail-handler-internal":
return nil, nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "intentional internal error")
case "fail-handler-app-error":
return nil, temporal.NewApplicationError("app error", "TestError", "details")
case "fail-handler-bad-request":
return nil, nexus.HandlerErrorf(nexus.HandlerErrorTypeBadRequest, "bad request")
case "fail-operation":
return nil, nexus.NewOperationFailedError("some error")
case "fail-operation-app-error":
return nil, temporal.NewNonRetryableApplicationError("app error", "TestError", nil, "details")
}
return nil, nexus.HandlerErrorf(nexus.HandlerErrorTypeBadRequest, "unexpected outcome: %s", outcome)
})
s.NoError(svc.Register(op))
callerWF := func(ctx workflow.Context, outcome string) (nexus.NoValue, error) {
c := workflow.NewNexusClient(endpointName, svc.Name)
fut := c.ExecuteOperation(ctx, op, outcome, workflow.NexusOperationOptions{})
return nil, fut.Get(ctx, nil)
}
w.RegisterNexusService(svc)
w.RegisterWorkflow(callerWF)
s.NoError(w.Start())
s.T().Cleanup(w.Stop)
cases := []struct {
outcome string
metricsOutcome string
checkPendingError func(t *testing.T, pendingErr error)
checkWorkflowError func(t *testing.T, wfErr error)
}{
{
outcome: "fail-handler-internal",
metricsOutcome: "handler-error:INTERNAL",
checkPendingError: func(t *testing.T, pendingErr error) {
var handlerErr *nexus.HandlerError
require.ErrorAs(t, pendingErr, &handlerErr)
require.Equal(t, nexus.HandlerErrorTypeInternal, handlerErr.Type)
var appErr *temporal.ApplicationError
require.ErrorAs(t, handlerErr.Cause, &appErr)
require.Equal(t, "intentional internal error", appErr.Message())
},
},
{
outcome: "fail-handler-app-error",
metricsOutcome: "handler-error:INTERNAL",
checkPendingError: func(t *testing.T, pendingErr error) {
var handlerErr *nexus.HandlerError
require.ErrorAs(t, pendingErr, &handlerErr)
require.Equal(t, nexus.HandlerErrorTypeInternal, handlerErr.Type)
var appErr *temporal.ApplicationError
require.ErrorAs(t, handlerErr.Cause, &appErr)
require.Equal(t, "app error", appErr.Message())
require.Equal(t, "TestError", appErr.Type())
var details string
require.NoError(t, appErr.Details(&details))
require.Equal(t, "details", details)
},
},
{
outcome: "fail-handler-bad-request",
metricsOutcome: "handler-error:BAD_REQUEST",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var handlerErr *nexus.HandlerError
require.ErrorAs(t, opErr, &handlerErr)
require.Equal(t, nexus.HandlerErrorTypeBadRequest, handlerErr.Type)
var appErr *temporal.ApplicationError
require.ErrorAs(t, handlerErr.Cause, &appErr)
require.Equal(t, "bad request", appErr.Message())
},
},
{
outcome: "fail-operation",
metricsOutcome: "operation-unsuccessful:failed",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var appErr *temporal.ApplicationError
require.ErrorAs(t, opErr, &appErr)
require.Equal(t, "some error", appErr.Message())
},
},
{
outcome: "fail-operation-app-error",
metricsOutcome: "handler-error:INTERNAL",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var appErr *temporal.ApplicationError
require.ErrorAs(t, opErr, &appErr)
require.Equal(t, "app error", appErr.Message())
require.Equal(t, "TestError", appErr.Type())
var details string
require.NoError(t, appErr.Details(&details))
require.Equal(t, "details", details)
},
},
}
for _, tc := range cases {
s.T().Run(tc.outcome, func(t *testing.T) {
capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture()
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, callerWF, tc.outcome)
s.NoError(err)
if tc.checkPendingError != nil {
var f *failurepb.Failure
require.EventuallyWithT(t, func(t *assert.CollectT) {
desc, err := s.SdkClient().DescribeWorkflowExecution(ctx, run.GetID(), run.GetRunID())
require.NoError(t, err)
if err != nil {
return
}
assert.Len(t, desc.PendingNexusOperations, 1)
if len(desc.PendingNexusOperations) != 1 {
return
}
f = desc.PendingNexusOperations[0].LastAttemptFailure
assert.NotNil(t, f)
}, 10*time.Second, 100*time.Millisecond)
s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
tc.checkPendingError(t, converter.FailureToError(f))
s.NoError(s.SdkClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "test cleanup"))
} else {
wfErr := run.Get(ctx, nil)
s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
tc.checkWorkflowError(t, wfErr)
}
snap := capture.Snapshot()
require.Len(t, snap["nexus_outbound_requests"], 1)
require.Subset(
t,
snap["nexus_outbound_requests"][0].Tags,
map[string]string{
"namespace": s.Namespace().String(),
"method": "StartOperation",
"failure_source": "worker",
"outcome": tc.metricsOutcome,
},
)
})
}
}
func (s *NexusWorkflowTestSuite) TestNexusAsyncOperationErrorRehydration() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
defer cancel()
testCtx := ctx
taskQueue := testcore.RandomizeStr("caller_" + s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
handlerWorkflowID := testcore.RandomizeStr(s.T().Name())
_, err := s.SdkClient().OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: taskQueue,
},
},
},
},
})
s.NoError(err)
w := worker.New(
s.SdkClient(),
taskQueue,
worker.Options{},
)
svc := nexus.NewService("test")
handlerWF := func(ctx workflow.Context, outcome string) (nexus.NoValue, error) {
switch outcome {
case "wait", "timeout":
// Wait for the workflow to be canceled.
return nil, workflow.Await(ctx, func() bool { return false })
case "fail":
return nil, temporal.NewApplicationError("app error", "TestError", "details")
}
return nil, fmt.Errorf("unexpected outcome: %s", outcome)
}
op := temporalnexus.NewWorkflowRunOperation("op", handlerWF, func(ctx context.Context, outcome string, soo nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
var workflowExecutionTimeout time.Duration
if outcome == "timeout" {
workflowExecutionTimeout = time.Second
}
return client.StartWorkflowOptions{ID: handlerWorkflowID, WorkflowExecutionTimeout: workflowExecutionTimeout}, nil
})
s.NoError(svc.Register(op))
callerWF := func(ctx workflow.Context, outcome, action string) (nexus.NoValue, error) {
opCtx, cancel := workflow.WithCancel(ctx)
defer cancel()
c := workflow.NewNexusClient(endpointName, svc.Name)
fut := c.ExecuteOperation(opCtx, op, outcome, workflow.NexusOperationOptions{})
var exec workflow.NexusOperationExecution
if err := fut.GetNexusOperationExecution().Get(ctx, &exec); err != nil {
return nil, err
}
switch action {
case "terminate":
// Lazy man's version of a local activity, don't try this at home.
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
err := s.SdkClient().TerminateWorkflow(testCtx, handlerWorkflowID, "", "")
if err != nil {
panic(err)
}
return nil
})
case "cancel":
cancel()
err := fut.Get(ctx, nil)
// The Go SDK unwraps CanceledErrors when an error is returned from the workflow, assert in-workflow.
var opErr *temporal.NexusOperationError
if !errors.As(err, &opErr) {
return nil, fmt.Errorf("expected NexusOperationError, got %w", err)
}
var canceledErr *temporal.CanceledError
if !errors.As(opErr, &canceledErr) {
return nil, fmt.Errorf("expected CanceledError, got %w", err)
}
}
return nil, fut.Get(ctx, nil)
}
w.RegisterNexusService(svc)
w.RegisterWorkflow(callerWF)
w.RegisterWorkflow(handlerWF)
s.NoError(w.Start())
s.T().Cleanup(w.Stop)
cases := []struct {
outcome, action string
checkWorkflowError func(t *testing.T, wfErr error)
}{
{
outcome: "fail",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var appErr *temporal.ApplicationError
require.ErrorAs(t, opErr, &appErr)
require.Equal(t, "app error", appErr.Message())
require.Equal(t, "TestError", appErr.Type())
var details string
require.NoError(t, appErr.Details(&details))
require.Equal(t, "details", details)
},
},
{
outcome: "wait",
action: "terminate",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var termErr *temporal.TerminatedError
require.ErrorAs(t, opErr, &termErr)
},
},
{
outcome: "wait",
action: "cancel",
checkWorkflowError: func(t *testing.T, wfErr error) {
// The Go SDK loses the NexusOperationError (as well as any other error if it wraps a CanceledError),
// assertions done in workflow.
var canceledErr *temporal.CanceledError
require.ErrorAs(t, wfErr, &canceledErr)
},
},
{
outcome: "timeout",
checkWorkflowError: func(t *testing.T, wfErr error) {
var opErr *temporal.NexusOperationError
require.ErrorAs(t, wfErr, &opErr)
var timeoutErr *temporal.TimeoutError
require.ErrorAs(t, opErr, &timeoutErr)
},
},
}
for _, tc := range cases {
s.T().Run(tc.outcome+"-"+tc.action, func(t *testing.T) {
capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture()
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, callerWF, tc.outcome, tc.action)
s.NoError(err)
wfErr := run.Get(ctx, nil)
s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
tc.checkWorkflowError(t, wfErr)
snap := capture.Snapshot()
require.GreaterOrEqual(t, len(snap["nexus_outbound_requests"]), 1)
require.Subset(t, snap["nexus_outbound_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "method": "StartOperation", "failure_source": "_unknown_", "outcome": "pending"})
})
}
}
func (s *NexusWorkflowTestSuite) TestNexusCallbackAfterCallerComplete() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
defer cancel()
taskQueue := testcore.RandomizeStr("caller_" + s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
handlerWorkflowID := testcore.RandomizeStr(s.T().Name())
_, err := s.SdkClient().OperatorService().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_Worker_{
Worker: &nexuspb.EndpointTarget_Worker{
Namespace: s.Namespace().String(),
TaskQueue: taskQueue,
},
},
},
},
})
s.NoError(err)
w := worker.New(
s.SdkClient(),
taskQueue,
worker.Options{},
)
svc := nexus.NewService("test")
handlerWF := func(ctx workflow.Context, _ nexus.NoValue) (nexus.NoValue, error) {
return nil, workflow.Sleep(ctx, 1*time.Second)
}
op := temporalnexus.NewWorkflowRunOperation("op", handlerWF, func(ctx context.Context, _ nexus.NoValue, soo nexus.StartOperationOptions) (client.StartWorkflowOptions, error) {
return client.StartWorkflowOptions{ID: handlerWorkflowID}, nil
})
s.NoError(svc.Register(op))
callerWF := func(ctx workflow.Context) error {
c := workflow.NewNexusClient(endpointName, svc.Name)
fut := c.ExecuteOperation(ctx, op, nil, workflow.NexusOperationOptions{})
return fut.Get(ctx, nil)
}
w.RegisterNexusService(svc)
w.RegisterWorkflow(callerWF)
w.RegisterWorkflow(handlerWF)
s.NoError(w.Start())
s.T().Cleanup(w.Stop)
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
WorkflowRunTimeout: 200 * time.Millisecond,
}, callerWF)
s.NoError(err)
wfErr := run.Get(ctx, nil)
s.Error(wfErr)
s.EventuallyWithT(func(ct *assert.CollectT) {
resp, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
Execution: &commonpb.WorkflowExecution{
WorkflowId: handlerWorkflowID,
},
})
require.NoError(ct, err)
require.Len(ct, resp.Callbacks, 1)
require.Equal(ct, resp.Callbacks[0].State, enumspb.CALLBACK_STATE_FAILED)
require.NotNil(ct, resp.Callbacks[0].LastAttemptFailure)
require.Equal(ct, resp.Callbacks[0].LastAttemptFailure.Message, "handler error (NOT_FOUND): workflow execution already completed")
}, 3*time.Second, 200*time.Millisecond)
}
func (s *NexusWorkflowTestSuite) TestNexusOperationSyncNexusFailure() {
ctx := testcore.NewContext()
taskQueue := testcore.RandomizeStr(s.T().Name())
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
h := nexustest.Handler{
OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) {
return nil, &nexus.HandlerError{
Type: nexus.HandlerErrorTypeBadRequest,
Cause: &nexus.FailureError{
Failure: nexus.Failure{
Message: "fail me",
Metadata: map[string]string{"key": "val"},
Details: []byte(`"details"`),
},
},
}
},
}
listenAddr := nexustest.AllocListenAddress()
nexustest.NewNexusServer(s.T(), listenAddr, h)
_, err := s.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{
Spec: &nexuspb.EndpointSpec{
Name: endpointName,
Target: &nexuspb.EndpointTarget{
Variant: &nexuspb.EndpointTarget_External_{
External: &nexuspb.EndpointTarget_External{
Url: "http://" + listenAddr,
},
},
},
},
})
s.NoError(err)
w := worker.New(
s.SdkClient(),
taskQueue,
worker.Options{},
)
callerWF := func(ctx workflow.Context) (nexus.NoValue, error) {
c := workflow.NewNexusClient(endpointName, "dont-care")
fut := c.ExecuteOperation(ctx, "operation", nil, workflow.NexusOperationOptions{})
return nil, fut.Get(ctx, nil)
}
w.RegisterWorkflow(callerWF)
s.NoError(w.Start())
s.T().Cleanup(w.Stop)
capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture()
run, err := s.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{
TaskQueue: taskQueue,
}, callerWF)
s.NoError(err)
wfErr := run.Get(ctx, nil)
s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
var handlerErr *nexus.HandlerError
s.ErrorAs(wfErr, &handlerErr)
s.Equal(nexus.HandlerErrorTypeBadRequest, handlerErr.Type)
var appErr *temporal.ApplicationError
s.ErrorAs(handlerErr.Cause, &appErr)
s.Equal(appErr.Message(), "fail me")
var failure nexus.Failure
s.NoError(appErr.Details(&failure))
s.Equal(map[string]string{"key": "val"}, failure.Metadata)
var details string
s.NoError(json.Unmarshal(failure.Details, &details))
s.Equal("details", details)
snap := capture.Snapshot()
s.Len(snap["nexus_outbound_requests"], 1)
// Confirming that requests which do not go through our frontend are not tagged with `failure_source`
s.Subset(snap["nexus_outbound_requests"][0].Tags, map[string]string{"namespace": s.Namespace().String(), "method": "StartOperation", "failure_source": "_unknown_", "outcome": "handler-error:BAD_REQUEST"})
}
func (s *NexusWorkflowTestSuite) sendNexusCompletionRequest(
ctx context.Context,
t *testing.T,
url string,
completion nexus.OperationCompletion,
callbackToken string,
) (*http.Response, map[string][]*metricstest.CapturedRecording) {
capture := s.GetTestCluster().Host().CaptureMetricsHandler().StartCapture()
defer s.GetTestCluster().Host().CaptureMetricsHandler().StopCapture(capture)
req, err := nexus.NewCompletionHTTPRequest(ctx, url, completion)
require.NoError(t, err)
if callbackToken != "" {
req.Header.Add(commonnexus.CallbackTokenHeader, callbackToken)
}
res, err := http.DefaultClient.Do(req)
require.NoError(t, err)
_, err = io.ReadAll(res.Body)
require.NoError(t, err)
defer res.Body.Close()
return res, capture.Snapshot()
}
Reference in New Issue View Git Blame Copy Permalink