mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
Reconstruct failure for forwarded Nexus completion requests (#8198)
## What changed? When forwarding a `CompleteNexusOperation` HTTP request that contains a failure, the completion will be reconstructed instead of reusing the original request body. ## Why? The Nexus SDK reads and closes the HTTP request body when the operation state is `failed` or `canceled` so we cannot reuse it for the forwarded request. For `successful` operations, the SDK just passes on the result content in the form of a `nexus.LazyValue` which we can forward directly since it is not read or closed. ## How did you test it? new functional xdc tests
This commit is contained in:
pdoerner
@@ -264,10 +264,41 @@ func (h *completionHandler) forwardCompleteOperation(ctx context.Context, r *nex
|
||||
}
|
||||
}
|
||||
|
||||
forwardReq, err := http.NewRequestWithContext(ctx, r.HTTPRequest.Method, forwardURL, r.HTTPRequest.Body)
|
||||
if err != nil {
|
||||
h.Logger.Error("failed to construct forwarding HTTP request", tag.Operation(apiName), tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err))
|
||||
return nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
|
||||
var forwardReq *http.Request
|
||||
switch r.State {
|
||||
case nexus.OperationStateSucceeded:
|
||||
// For successful operations, the Nexus framework streams the result as a LazyValue, so we can reuse the
|
||||
// incoming request body.
|
||||
forwardReq, err = http.NewRequestWithContext(ctx, r.HTTPRequest.Method, forwardURL, r.HTTPRequest.Body)
|
||||
if err != nil {
|
||||
h.Logger.Error("failed to construct forwarding HTTP request", tag.Operation(apiName), tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err))
|
||||
return nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
|
||||
}
|
||||
case nexus.OperationStateFailed, nexus.OperationStateCanceled:
|
||||
// For unsuccessful operations, the Nexus framework reads and closes the original request body to deserialize
|
||||
// the failure, so we must construct a new completion to forward.
|
||||
var failureErr *nexus.FailureError
|
||||
if !errors.As(r.Error, &failureErr) {
|
||||
// This shouldn't happen as the Nexus SDK is always expected to convert Failures from the wire to
|
||||
// FailureErrors.
|
||||
h.Logger.Error("received unexpected error type when trying to forward Nexus operation completion", tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err))
|
||||
return nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
|
||||
}
|
||||
c := &nexus.OperationCompletionUnsuccessful{
|
||||
Header: httpHeaderToNexusHeader(r.HTTPRequest.Header),
|
||||
State: r.State,
|
||||
OperationToken: r.OperationToken,
|
||||
StartTime: r.StartTime,
|
||||
Links: r.Links,
|
||||
Failure: failureErr.Failure,
|
||||
}
|
||||
forwardReq, err = nexus.NewCompletionHTTPRequest(ctx, forwardURL, c)
|
||||
if err != nil {
|
||||
h.Logger.Error("failed to construct forwarding HTTP request", tag.Operation(apiName), tag.WorkflowNamespace(rCtx.namespace.Name().String()), tag.Error(err))
|
||||
return nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
|
||||
}
|
||||
default:
|
||||
return nexus.HandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid operation state: %q", r.State)
|
||||
}
|
||||
|
||||
if r.HTTPRequest.Header != nil {
|
||||
@@ -337,6 +368,19 @@ func isMediaTypeJSON(contentType string) bool {
|
||||
return err == nil && mediaType == "application/json"
|
||||
}
|
||||
|
||||
// Copies HTTP request headers to Nexus headers except those starting with content- since those will be added by the client.
|
||||
func httpHeaderToNexusHeader(httpHeader http.Header) nexus.Header {
|
||||
header := nexus.Header{}
|
||||
for k, v := range httpHeader {
|
||||
lowerK := strings.ToLower(k)
|
||||
if !strings.HasPrefix(lowerK, "content-") {
|
||||
// Nexus headers can only have single values, ignore multiple values.
|
||||
header[lowerK] = v[0]
|
||||
}
|
||||
}
|
||||
return header
|
||||
}
|
||||
|
||||
type requestContext struct {
|
||||
*completionHandler
|
||||
logger log.Logger
|
||||
|
||||
@@ -30,6 +30,7 @@ import (
|
||||
"go.temporal.io/server/common/metrics/metricstest"
|
||||
cnexus "go.temporal.io/server/common/nexus"
|
||||
"go.temporal.io/server/common/nexus/nexustest"
|
||||
"go.temporal.io/server/common/payload"
|
||||
"go.temporal.io/server/components/callbacks"
|
||||
"go.temporal.io/server/components/nexusoperations"
|
||||
"go.temporal.io/server/tests/testcore"
|
||||
@@ -319,192 +320,264 @@ func (s *NexusRequestForwardingSuite) TestCancelOperationForwardedFromStandbyToA
|
||||
}
|
||||
}
|
||||
|
||||
func (s *NexusRequestForwardingSuite) TestCompleteOperationForwardedFromStandbyToActive() {
|
||||
// Override templates to always return passive cluster in callback URL
|
||||
s.clusters[0].OverrideDynamicConfig(
|
||||
s.T(),
|
||||
nexusoperations.CallbackURLTemplate,
|
||||
"http://"+s.clusters[1].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback")
|
||||
s.clusters[1].OverrideDynamicConfig(
|
||||
s.T(),
|
||||
nexusoperations.CallbackURLTemplate,
|
||||
"http://"+s.clusters[1].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback")
|
||||
|
||||
ctx := testcore.NewContext()
|
||||
ns := s.createGlobalNamespace()
|
||||
taskQueue := fmt.Sprintf("%v-%v", "test-task-queue", uuid.New())
|
||||
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(cnexus.CallbackTokenHeader)
|
||||
publicCallbackUrl = options.CallbackURL
|
||||
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
|
||||
func (s *NexusRequestForwardingSuite) TestOperationCompletionForwardedFromStandbyToActive() {
|
||||
testCases := []struct {
|
||||
name string
|
||||
getCompletionFn func() (nexus.OperationCompletion, error)
|
||||
assertHistoryAndGetCompleteWF func(*testing.T, []*historypb.HistoryEvent) *workflowservice.RespondWorkflowTaskCompletedRequest
|
||||
assertResult func(*testing.T, string)
|
||||
}{
|
||||
{
|
||||
name: "success",
|
||||
getCompletionFn: func() (nexus.OperationCompletion, error) {
|
||||
return nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
|
||||
Serializer: cnexus.PayloadSerializer,
|
||||
})
|
||||
},
|
||||
assertHistoryAndGetCompleteWF: func(t *testing.T, events []*historypb.HistoryEvent) *workflowservice.RespondWorkflowTaskCompletedRequest {
|
||||
completedEventIdx := slices.IndexFunc(events, func(e *historypb.HistoryEvent) bool {
|
||||
return e.GetNexusOperationCompletedEventAttributes() != nil
|
||||
})
|
||||
require.Greater(t, completedEventIdx, 0)
|
||||
return &workflowservice.RespondWorkflowTaskCompletedRequest{
|
||||
Identity: "test",
|
||||
Commands: []*commandpb.Command{{CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
|
||||
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
|
||||
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
|
||||
Result: &commonpb.Payloads{
|
||||
Payloads: []*commonpb.Payload{
|
||||
events[completedEventIdx].GetNexusOperationCompletedEventAttributes().Result,
|
||||
}}}}}},
|
||||
}
|
||||
},
|
||||
assertResult: func(t *testing.T, result string) {
|
||||
require.Equal(t, "result", result)
|
||||
},
|
||||
},
|
||||
}
|
||||
listenAddr := nexustest.AllocListenAddress()
|
||||
nexustest.NewNexusServer(s.T(), listenAddr, h)
|
||||
|
||||
createEndpointReq := &operatorservice.CreateNexusEndpointRequest{
|
||||
Spec: &nexuspb.EndpointSpec{
|
||||
Name: endpointName,
|
||||
Target: &nexuspb.EndpointTarget{
|
||||
Variant: &nexuspb.EndpointTarget_External_{
|
||||
External: &nexuspb.EndpointTarget_External{
|
||||
Url: "http://" + listenAddr,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "operation error",
|
||||
getCompletionFn: func() (nexus.OperationCompletion, error) {
|
||||
f := nexus.Failure{Message: "intentional operation failure"}
|
||||
return nexus.NewOperationCompletionUnsuccessful(
|
||||
&nexus.OperationError{State: nexus.OperationStateFailed, Cause: &nexus.FailureError{Failure: f}},
|
||||
nexus.OperationCompletionUnsuccessfulOptions{})
|
||||
},
|
||||
assertHistoryAndGetCompleteWF: func(t *testing.T, events []*historypb.HistoryEvent) *workflowservice.RespondWorkflowTaskCompletedRequest {
|
||||
failedEventIdx := slices.IndexFunc(events, func(e *historypb.HistoryEvent) bool {
|
||||
return e.GetNexusOperationFailedEventAttributes() != nil
|
||||
})
|
||||
require.Greater(t, failedEventIdx, 0)
|
||||
return &workflowservice.RespondWorkflowTaskCompletedRequest{
|
||||
Identity: "test",
|
||||
Commands: []*commandpb.Command{{CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
|
||||
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
|
||||
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
|
||||
Result: &commonpb.Payloads{
|
||||
Payloads: []*commonpb.Payload{
|
||||
payload.EncodeString(events[failedEventIdx].GetNexusOperationFailedEventAttributes().GetFailure().Message),
|
||||
}}}}}},
|
||||
}
|
||||
},
|
||||
assertResult: func(t *testing.T, result string) {
|
||||
require.Equal(t, "intentional operation failure", result)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "canceled",
|
||||
getCompletionFn: func() (nexus.OperationCompletion, error) {
|
||||
f := nexus.Failure{Message: "operation canceled"}
|
||||
return nexus.NewOperationCompletionUnsuccessful(
|
||||
&nexus.OperationError{State: nexus.OperationStateCanceled, Cause: &nexus.FailureError{Failure: f}},
|
||||
nexus.OperationCompletionUnsuccessfulOptions{})
|
||||
},
|
||||
assertHistoryAndGetCompleteWF: func(t *testing.T, events []*historypb.HistoryEvent) *workflowservice.RespondWorkflowTaskCompletedRequest {
|
||||
canceledEventIdx := slices.IndexFunc(events, func(e *historypb.HistoryEvent) bool {
|
||||
return e.GetNexusOperationCanceledEventAttributes() != nil
|
||||
})
|
||||
require.Greater(t, canceledEventIdx, 0)
|
||||
return &workflowservice.RespondWorkflowTaskCompletedRequest{
|
||||
Identity: "test",
|
||||
Commands: []*commandpb.Command{{CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
|
||||
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
|
||||
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
|
||||
Result: &commonpb.Payloads{
|
||||
Payloads: []*commonpb.Payload{
|
||||
payload.EncodeString(events[canceledEventIdx].GetNexusOperationCanceledEventAttributes().GetFailure().Message),
|
||||
}}}}}},
|
||||
}
|
||||
},
|
||||
assertResult: func(t *testing.T, result string) {
|
||||
require.Equal(t, "operation canceled", result)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
_, err := s.clusters[0].OperatorClient().CreateNexusEndpoint(ctx, createEndpointReq)
|
||||
s.NoError(err)
|
||||
for _, tc := range testCases {
|
||||
tc := tc
|
||||
s.T().Run(tc.name, func(t *testing.T) {
|
||||
// Override templates to always return passive cluster in callback URL
|
||||
s.clusters[0].OverrideDynamicConfig(
|
||||
s.T(),
|
||||
nexusoperations.CallbackURLTemplate,
|
||||
"http://"+s.clusters[1].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback")
|
||||
s.clusters[1].OverrideDynamicConfig(
|
||||
s.T(),
|
||||
nexusoperations.CallbackURLTemplate,
|
||||
"http://"+s.clusters[1].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback")
|
||||
|
||||
_, err = s.clusters[1].OperatorClient().CreateNexusEndpoint(ctx, createEndpointReq)
|
||||
s.NoError(err)
|
||||
ctx := testcore.NewContext()
|
||||
ns := s.createGlobalNamespace()
|
||||
taskQueue := fmt.Sprintf("%v-%v", "test-task-queue", uuid.New())
|
||||
endpointName := testcore.RandomizedNexusEndpoint(s.T().Name())
|
||||
|
||||
activeSDKClient, err := client.Dial(client.Options{
|
||||
HostPort: s.clusters[0].Host().FrontendGRPCAddress(),
|
||||
Namespace: ns,
|
||||
Logger: log.NewSdkLogger(s.logger),
|
||||
})
|
||||
s.NoError(err)
|
||||
var callbackToken, publicCallbackUrl string
|
||||
|
||||
run, err := activeSDKClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
||||
TaskQueue: taskQueue,
|
||||
}, "workflow")
|
||||
s.NoError(err)
|
||||
|
||||
feClient0 := s.clusters[0].FrontendClient()
|
||||
feClient1 := s.clusters[1].FrontendClient()
|
||||
|
||||
pollResp, err := feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
TaskQueue: &taskqueuepb.TaskQueue{
|
||||
Name: taskQueue,
|
||||
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
||||
},
|
||||
Identity: "test",
|
||||
})
|
||||
s.NoError(err)
|
||||
_, err = feClient0.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"),
|
||||
},
|
||||
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(cnexus.CallbackTokenHeader)
|
||||
publicCallbackUrl = options.CallbackURL
|
||||
return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
// Poll and verify that the "started" event was recorded.
|
||||
pollResp, err = feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
TaskQueue: &taskqueuepb.TaskQueue{
|
||||
Name: taskQueue,
|
||||
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
||||
},
|
||||
Identity: "test",
|
||||
})
|
||||
s.NoError(err)
|
||||
_, err = feClient0.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)
|
||||
|
||||
// Wait for Nexus operation to be replicated
|
||||
s.Eventually(func() bool {
|
||||
resp, err := feClient1.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
|
||||
Namespace: ns,
|
||||
Execution: &commonpb.WorkflowExecution{
|
||||
WorkflowId: run.GetID(),
|
||||
RunId: run.GetRunID(),
|
||||
},
|
||||
})
|
||||
return err == nil && len(resp.PendingNexusOperations) > 0
|
||||
}, 5*time.Second, 500*time.Millisecond)
|
||||
|
||||
// Send a valid - successful completion request to standby cluster.
|
||||
completion, err := nexus.NewOperationCompletionSuccessful(s.mustToPayload("result"), nexus.OperationCompletionSuccessfulOptions{
|
||||
Serializer: cnexus.PayloadSerializer,
|
||||
})
|
||||
s.NoError(err)
|
||||
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), s.clusters[1], 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": ns, "outcome": "request_forwarded"})
|
||||
|
||||
// Ensure that CompleteOperation request is tracked as part of normal service telemetry metrics
|
||||
s.Condition(func() bool {
|
||||
for _, m := range snap["service_requests"] {
|
||||
if opTag, ok := m.Tags["operation"]; ok && opTag == "CompleteNexusOperation" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
listenAddr := nexustest.AllocListenAddress()
|
||||
nexustest.NewNexusServer(s.T(), listenAddr, h)
|
||||
|
||||
// Resend the request and verify we get a not found error since the operation has already completed.
|
||||
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), s.clusters[0], 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": ns, "outcome": "error_not_found"})
|
||||
|
||||
// Poll active cluster and verify the completion is recorded and triggers workflow progress.
|
||||
pollResp, err = feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
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 = feClient0.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,
|
||||
createEndpointReq := &operatorservice.CreateNexusEndpointRequest{
|
||||
Spec: &nexuspb.EndpointSpec{
|
||||
Name: endpointName,
|
||||
Target: &nexuspb.EndpointTarget{
|
||||
Variant: &nexuspb.EndpointTarget_External_{
|
||||
External: &nexuspb.EndpointTarget_External{
|
||||
Url: "http://" + listenAddr,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
s.NoError(err)
|
||||
var result string
|
||||
s.NoError(run.Get(ctx, &result))
|
||||
s.Equal("result", result)
|
||||
}
|
||||
|
||||
_, err := s.clusters[0].OperatorClient().CreateNexusEndpoint(ctx, createEndpointReq)
|
||||
s.NoError(err)
|
||||
|
||||
_, err = s.clusters[1].OperatorClient().CreateNexusEndpoint(ctx, createEndpointReq)
|
||||
s.NoError(err)
|
||||
|
||||
activeSDKClient, err := client.Dial(client.Options{
|
||||
HostPort: s.clusters[0].Host().FrontendGRPCAddress(),
|
||||
Namespace: ns,
|
||||
Logger: log.NewSdkLogger(s.logger),
|
||||
})
|
||||
s.NoError(err)
|
||||
|
||||
run, err := activeSDKClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
||||
TaskQueue: taskQueue,
|
||||
}, "workflow")
|
||||
s.NoError(err)
|
||||
|
||||
feClient0 := s.clusters[0].FrontendClient()
|
||||
feClient1 := s.clusters[1].FrontendClient()
|
||||
|
||||
pollResp, err := feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
TaskQueue: &taskqueuepb.TaskQueue{
|
||||
Name: taskQueue,
|
||||
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
||||
},
|
||||
Identity: "test",
|
||||
})
|
||||
s.NoError(err)
|
||||
_, err = feClient0.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 = feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
TaskQueue: &taskqueuepb.TaskQueue{
|
||||
Name: taskQueue,
|
||||
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
||||
},
|
||||
Identity: "test",
|
||||
})
|
||||
s.NoError(err)
|
||||
_, err = feClient0.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)
|
||||
|
||||
// Wait for Nexus operation to be replicated
|
||||
s.Eventually(func() bool {
|
||||
resp, err := feClient1.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
|
||||
Namespace: ns,
|
||||
Execution: &commonpb.WorkflowExecution{
|
||||
WorkflowId: run.GetID(),
|
||||
RunId: run.GetRunID(),
|
||||
},
|
||||
})
|
||||
return err == nil && len(resp.PendingNexusOperations) > 0
|
||||
}, 5*time.Second, 500*time.Millisecond)
|
||||
|
||||
completion, err := tc.getCompletionFn()
|
||||
s.NoError(err)
|
||||
res, snap := s.sendNexusCompletionRequest(ctx, s.T(), s.clusters[1], 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": ns, "outcome": "request_forwarded"})
|
||||
|
||||
// Ensure that CompleteOperation request is tracked as part of normal service telemetry metrics
|
||||
s.Condition(func() bool {
|
||||
for _, m := range snap["service_requests"] {
|
||||
if opTag, ok := m.Tags["operation"]; ok && opTag == "CompleteNexusOperation" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
|
||||
// Resend the request and verify we get a not found error since the operation has already completed.
|
||||
res, snap = s.sendNexusCompletionRequest(ctx, s.T(), s.clusters[0], 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": ns, "outcome": "error_not_found"})
|
||||
|
||||
// Poll active cluster and verify the completion is recorded and triggers workflow progress.
|
||||
pollResp, err = feClient0.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{
|
||||
Namespace: ns,
|
||||
TaskQueue: &taskqueuepb.TaskQueue{
|
||||
Name: taskQueue,
|
||||
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
|
||||
},
|
||||
Identity: "test",
|
||||
})
|
||||
s.NoError(err)
|
||||
completeWfReq := tc.assertHistoryAndGetCompleteWF(t, pollResp.History.Events)
|
||||
completeWfReq.TaskToken = pollResp.TaskToken
|
||||
_, err = feClient0.RespondWorkflowTaskCompleted(ctx, completeWfReq)
|
||||
s.NoError(err)
|
||||
var result string
|
||||
s.NoError(run.Get(ctx, &result))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *NexusRequestForwardingSuite) nexusTaskPoller(ctx context.Context, frontendClient workflowservice.WorkflowServiceClient, ns string, taskQueue string, handler func(*workflowservice.PollNexusTaskQueueResponse) (*nexuspb.Response, *nexuspb.HandlerError)) {
|
||||
|
||||
Reference in New Issue
Block a user