diff --git a/common/links/validator.go b/common/links/validator.go index 40d894e3ed..e05c25b4d8 100644 --- a/common/links/validator.go +++ b/common/links/validator.go @@ -72,6 +72,16 @@ func Validate(links []*commonpb.Link, maxAllowedLinks, maxSize int) error { if t.Activity.GetRunId() == "" { return serviceerror.NewInvalidArgument("activity link must not have an empty run ID field") } + case *commonpb.Link_Workflow_: + if t.Workflow.GetNamespace() == "" { + return serviceerror.NewInvalidArgument("workflow link must not have an empty namespace field") + } + if t.Workflow.GetWorkflowId() == "" { + return serviceerror.NewInvalidArgument("workflow link must not have an empty workflow ID field") + } + if t.Workflow.GetRunId() == "" { + return serviceerror.NewInvalidArgument("workflow link must not have an empty run ID field") + } default: return serviceerror.NewInvalidArgument("unsupported link variant") } diff --git a/common/links/validator_test.go b/common/links/validator_test.go index 74c3998062..816de48f59 100644 --- a/common/links/validator_test.go +++ b/common/links/validator_test.go @@ -52,6 +52,15 @@ func TestValidate(t *testing.T) { }, }, } + validWorkflow := &commonpb.Link{ + Variant: &commonpb.Link_Workflow_{ + Workflow: &commonpb.Link_Workflow{ + Namespace: "ns", + WorkflowId: "wid", + RunId: "rid", + }, + }, + } t.Run("HappyPath", func(t *testing.T) { err := links.Validate([]*commonpb.Link{ @@ -59,7 +68,8 @@ func TestValidate(t *testing.T) { validBatchJob, validNexusOperation, validActivity, - }, maxLinks+1, maxSize) + validWorkflow, + }, maxLinks+2, maxSize) require.NoError(t, err) }) @@ -155,4 +165,25 @@ func TestValidate(t *testing.T) { err := links.Validate([]*commonpb.Link{{}}, maxLinks, maxSize) require.ErrorContains(t, err, "unsupported link variant") }) + + t.Run("Workflow/EmptyNamespace", func(t *testing.T) { + l := proto.Clone(validWorkflow).(*commonpb.Link) + l.GetWorkflow().Namespace = "" + err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize) + require.EqualError(t, err, "workflow link must not have an empty namespace field") + }) + + t.Run("Workflow/EmptyWorkflowID", func(t *testing.T) { + l := proto.Clone(validWorkflow).(*commonpb.Link) + l.GetWorkflow().WorkflowId = "" + err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize) + require.EqualError(t, err, "workflow link must not have an empty workflow ID field") + }) + + t.Run("Workflow/EmptyRunID", func(t *testing.T) { + l := proto.Clone(validWorkflow).(*commonpb.Link) + l.GetWorkflow().RunId = "" + err := links.Validate([]*commonpb.Link{l}, maxLinks, maxSize) + require.EqualError(t, err, "workflow link must not have an empty run ID field") + }) } diff --git a/common/nexus/links.go b/common/nexus/links.go index f75cf05cfd..83f253bb3d 100644 --- a/common/nexus/links.go +++ b/common/nexus/links.go @@ -5,13 +5,14 @@ import ( "github.com/nexus-rpc/sdk-go/nexus" commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/api/temporalnexus" "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" ) // ConvertNexusLinksToProtoLinks converts a slice of Nexus SDK links into Temporal proto links, -// supporting Link_WorkflowEvent and Link_Activity variants. Unsupported or malformed entries -// are skipped with a warning since links are non-essential to execution. +// supporting Link_Workflow, Link_WorkflowEvent, and Link_Activity variants. Unsupported or +// malformed entries are skipped with a warning since links are non-essential to execution. func ConvertNexusLinksToProtoLinks(nexusLinks []nexus.Link, logger log.Logger) []*commonpb.Link { var out []*commonpb.Link for _, nexusLink := range nexusLinks { @@ -40,6 +41,18 @@ func ConvertNexusLinksToProtoLinks(nexusLinks []nexus.Link, logger log.Logger) [ out = append(out, &commonpb.Link{ Variant: &commonpb.Link_Activity_{Activity: link}, }) + case string((&commonpb.Link_Workflow{}).ProtoReflect().Descriptor().FullName()): + link, err := temporalnexus.ConvertNexusLinkToLinkWorkflow(nexusLink) + if err != nil { + logger.Warn( + fmt.Sprintf("failed to parse link to %q: %s", nexusLink.Type, nexusLink.URL), + tag.Error(err), + ) + continue + } + out = append(out, &commonpb.Link{ + Variant: &commonpb.Link_Workflow_{Workflow: link}, + }) default: logger.Warn(fmt.Sprintf("invalid link data type: %q", nexusLink.Type)) } diff --git a/common/nexus/links_test.go b/common/nexus/links_test.go index ac841d6105..9bd981d5ce 100644 --- a/common/nexus/links_test.go +++ b/common/nexus/links_test.go @@ -13,12 +13,10 @@ import ( "go.temporal.io/server/common/testing/protorequire" ) -// TestConvertNexusLinksToProtoLinks_ActivityVariant verifies that the shared -// converter handles both WorkflowEvent and Activity link variants, drops -// unsupported types, and skips malformed entries — exercised by the Nexus task -// handler's start-response flow so a SAA invoked from a Nexus operation can -// surface its Activity link back to the caller. -func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) { +// TestConvertNexusLinksToProtoLinks verifies that the converter handles the +// Workflow, WorkflowEvent, and Activity link variants, drops unsupported types, +// and skips malformed entries. +func TestConvertNexusLinksToProtoLinks(t *testing.T) { logger := log.NewTestLogger() workflowEvent := nexus.Link{ @@ -36,6 +34,13 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) { }, Type: "temporal.api.common.v1.Link.Activity", } + workflow := nexus.Link{ + URL: &url.URL{ + Scheme: "temporal", + Path: "/namespaces/ns/workflows/wf-id/run-id", + }, + Type: "temporal.api.common.v1.Link.Workflow", + } unsupported := nexus.Link{ URL: &url.URL{Scheme: "temporal", Path: "/foo"}, Type: "unknown.Type", @@ -44,9 +49,29 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) { URL: &url.URL{Scheme: "temporal", Path: "/namespaces/ns/foo/act-id"}, Type: "temporal.api.common.v1.Link.Activity", } + malformedWorkflows := []nexus.Link{ + { + URL: &url.URL{Scheme: "temporal", Path: "/namespaces//workflows/wid/rid"}, // missing ns + Type: "temporal.api.common.v1.Link.Workflow", + }, + { + URL: &url.URL{Scheme: "temporal", Path: "/namespaces/ns/workflows//rid"}, // missing wid + Type: "temporal.api.common.v1.Link.Workflow", + }, + { + URL: &url.URL{Scheme: "temporal", Path: "/namespaces/ns/workflows/wid/"}, // missing rid + Type: "temporal.api.common.v1.Link.Workflow", + }, + { + URL: &url.URL{Scheme: "temporal", Path: "/foo"}, // incorrect path + Type: "temporal.api.common.v1.Link.Workflow", + }, + } + nexusLinks := []nexus.Link{workflowEvent, activity, workflow, unsupported, malformedActivity} + nexusLinks = append(nexusLinks, malformedWorkflows...) - out := commonnexus.ConvertNexusLinksToProtoLinks([]nexus.Link{workflowEvent, activity, unsupported, malformedActivity}, logger) - require.Len(t, out, 2, "workflow-event and activity links must round-trip; unsupported and malformed entries must be dropped") + out := commonnexus.ConvertNexusLinksToProtoLinks(nexusLinks, logger) + require.Len(t, out, 3, "workflow, workflow-event, and activity links must round-trip; unsupported and malformed entries must be dropped") expected := []*commonpb.Link{ { @@ -73,6 +98,15 @@ func TestConvertNexusLinksToProtoLinks_ActivityVariant(t *testing.T) { }, }, }, + { + Variant: &commonpb.Link_Workflow_{ + Workflow: &commonpb.Link_Workflow{ + Namespace: "ns", + WorkflowId: "wf-id", + RunId: "run-id", + }, + }, + }, } protorequire.ProtoSliceEqual(t, expected, out) } diff --git a/service/history/api/queryworkflow/api.go b/service/history/api/queryworkflow/api.go index 3f4ddfd35d..b0b85a8daa 100644 --- a/service/history/api/queryworkflow/api.go +++ b/service/history/api/queryworkflow/api.go @@ -37,7 +37,7 @@ func Invoke( workflowConsistencyChecker api.WorkflowConsistencyChecker, rawMatchingClient matchingservice.MatchingServiceClient, matchingClient matchingservice.MatchingServiceClient, -) (_ *historyservice.QueryWorkflowResponse, retError error) { +) (resp *historyservice.QueryWorkflowResponse, retError error) { scope := shardContext.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.HistoryQueryWorkflowScope)) namespaceID := namespace.ID(request.GetNamespaceId()) err := api.ValidateNamespaceUUID(namespaceID) @@ -79,6 +79,27 @@ func Invoke( // Note: QueryWorkflow should not alter mutable state, so it is safe to ignore error and not clear ms. workflowLease.GetReleaseFn()(nil) }() + defer func() { + if retError != nil || resp.GetResponse() == nil { + return + } + // Add link to Workflow regardless of query status. A rejection on the query is not an RPC error, + // so it gets the same link that a processed query would get - only the reason differs. + reason := "Query processed" + if resp.GetResponse().GetQueryRejected() != nil { + reason = "Query rejected" + } + resp.Response.Link = &commonpb.Link{ + Variant: &commonpb.Link_Workflow_{ + Workflow: &commonpb.Link_Workflow{ + Namespace: nsEntry.Name().String(), + WorkflowId: workflowKey.WorkflowID, + RunId: workflowKey.RunID, + Reason: reason, + }, + }, + } + }() // Context metadata is automatically set during mutable state transaction close for operations that mutate state. // Since QueryWorkflow is readonly and never closes the transaction, we explicitly call SetContextMetadata diff --git a/tests/nexus_workflow_query_test.go b/tests/nexus_workflow_query_test.go new file mode 100644 index 0000000000..b120a5f7ba --- /dev/null +++ b/tests/nexus_workflow_query_test.go @@ -0,0 +1,105 @@ +package tests + +import ( + "context" + + "github.com/nexus-rpc/sdk-go/nexus" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + querypb "go.temporal.io/api/query/v1" + apitemporalnexus "go.temporal.io/api/temporalnexus" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/worker" + "go.temporal.io/sdk/workflow" + "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/payloads" + "go.temporal.io/server/tests/testcore" +) + +func (s *NexusWorkflowTestSuite) TestNexusOperationBackedByQuery(chasmEnabled bool) { + env := s.newTestEnv(chasmEnabled) + ctx := s.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + + queryType := "status" + signalName := "done" + handlerWorkflowID := testcore.RandomizeStr(s.T().Name() + "-handler") + handlerWf := func(ctx workflow.Context) error { + _ = workflow.SetQueryHandler(ctx, queryType, func() (string, error) { + return "handler-status", nil + }) + workflow.GetSignalChannel(ctx, signalName).Receive(ctx, nil) + return nil + } + + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: handlerWorkflowID}, + Query: &querypb.WorkflowQuery{QueryType: queryType}, + }) + if err != nil { + return nil, err + } + // simulate the stitching that will be done by the SDKs eventually + nexus.AddHandlerLinks(ctx, apitemporalnexus.ConvertLinkWorkflowToNexusLink(resp.GetLink().GetWorkflow())) + + var result string + if err := payloads.Decode(resp.GetQueryResult(), &result); err != nil { + return nil, err + } + return &nexus.HandlerStartOperationResultSync[any]{Value: result}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + var result string + err := fut.Get(ctx, &result) + return result, err + } + + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + w.RegisterWorkflow(handlerWf) + s.NoError(w.Start()) + defer w.Stop() + + handlerRun, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + ID: handlerWorkflowID, + TaskQueue: taskQueue, + }, handlerWf) + s.NoError(err) + + callerRun, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + TaskQueue: taskQueue, + }, callerWF) + s.NoError(err) + + var result string + s.NoError(callerRun.Get(ctx, &result)) + s.Equal("handler-status", result) + + // verify the nexus operation completed event carries a link to the handler's workflow + hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: callerRun.GetID()}) + completedEvent := s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) + s.Len(completedEvent.GetLinks(), 1) + workflowLink := completedEvent.GetLinks()[0].GetWorkflow() + s.NotNil(workflowLink, "completed event must carry a link of type workflow") + s.Equal(env.Namespace().String(), workflowLink.GetNamespace()) + s.Equal(handlerWorkflowID, workflowLink.GetWorkflowId()) + s.Equal(handlerRun.GetRunID(), workflowLink.GetRunId()) + s.Equal("Query processed", workflowLink.GetReason()) + + s.NoError(env.SdkClient().SignalWorkflow(ctx, handlerWorkflowID, "", signalName, nil)) + s.NoError(handlerRun.Get(ctx, nil)) +} diff --git a/tests/query_workflow_test.go b/tests/query_workflow_test.go index e9c30cb257..1bf7ac9844 100644 --- a/tests/query_workflow_test.go +++ b/tests/query_workflow_test.go @@ -21,6 +21,7 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/client" sdkclient "go.temporal.io/sdk/client" + "go.temporal.io/sdk/converter" "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" "go.temporal.io/server/common/dynamicconfig" @@ -140,6 +141,130 @@ func (s *QueryWorkflowSuite) TestQueryWorkflow_Consistent_PiggybackQuery() { s.Equal("pauseabc", queryResultStr) } +func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() { + env := testcore.NewEnv(s.T()) + queryName := "query" + signalName := "test" + workflowFn := func(ctx workflow.Context) error { + orderStatus := "initialized" + _ = workflow.SetQueryHandler(ctx, queryName, func(_ string) (string, error) { + return orderStatus, nil + }) + workflow.GetSignalChannel(ctx, signalName).Receive(ctx, &orderStatus) + return nil + } + ctx := s.Context() + + env.SdkWorker().RegisterWorkflow(workflowFn) + + wid := "nexus-query-workflow-tests" + run, err := env.SdkClient().ExecuteWorkflow( + ctx, + sdkclient.StartWorkflowOptions{ + ID: wid, + TaskQueue: env.WorkerTaskQueue()}, + workflowFn, + ) + s.NoError(err) + + type testCase struct { + name string + query *querypb.WorkflowQuery + err bool + reason string + result string + queryRejectCondition enumspb.QueryRejectCondition + } + + runTestCase := func(tc testCase) { + resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, + Query: tc.query, + QueryRejectCondition: tc.queryRejectCondition, + }) + if tc.err { + s.Error(err) + return + } + s.NoError(err) + link := resp.GetLink().GetWorkflow() + s.NotNil(link, "query must carry a link of type workflow") + s.Equal(env.Namespace().String(), link.GetNamespace()) + s.Equal(wid, link.GetWorkflowId()) + s.Equal(run.GetRunID(), link.GetRunId()) + s.Equal(tc.reason, link.GetReason()) + if tc.result != "" { + s.Equal(tc.result, testcore.DecodeString(s.T(), resp.QueryResult)) + } + } + + runCases := func(groupName string, cases []testCase) { + for _, tc := range cases { + s.T().Run(groupName+"/"+tc.name, func(t *testing.T) { //nolint:testifylint // subtests need serialized execution(running, completed) + runTestCase(tc) + }) + } + } + + // TCs that are expected to fail on both running and completed workflows identically + invalidTestCases := []testCase{ + { + name: "malformed args", + query: &querypb.WorkflowQuery{ + QueryType: queryName, + QueryArgs: &commonpb.Payloads{ + Payloads: []*commonpb.Payload{{ + Metadata: map[string][]byte{ + converter.MetadataEncoding: []byte(converter.MetadataEncodingJSON), + }, + Data: []byte("dummy data"), // invalid JSON, fails to decode + }}, + }, + }, + err: true, + }, + { + name: "unknown query type", + query: &querypb.WorkflowQuery{QueryType: "some unknown query"}, + err: true, + }, + } + + runningWorkflowTestCases := []testCase{ + { + name: "well-formed query", + query: &querypb.WorkflowQuery{QueryType: queryName}, + reason: "Query processed", + result: "initialized", + }, + } + runCases("running-workflow", runningWorkflowTestCases) + runCases("running-workflow", invalidTestCases) + + // set the query result via signal so that query "reload" on completed workflow is verified inline + s.NoError(env.SdkClient().SignalWorkflow(ctx, wid, "", signalName, "order-delivered")) + s.NoError(run.Get(ctx, nil)) + + completedWorkflowTestCases := []testCase{ + { + name: "well-formed replayable query succeeds", + query: &querypb.WorkflowQuery{QueryType: queryName}, + reason: "Query processed", + result: "order-delivered", + }, + { + name: "well-formed non-replayable query fails", + query: &querypb.WorkflowQuery{QueryType: queryName}, + reason: "Query rejected", + queryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, + }, + } + + runCases("completed-workflow", completedWorkflowTestCases) + runCases("completed-workflow", invalidTestCases) +} + func (s *QueryWorkflowSuite) TestQueryWorkflow_QueryWhileBackoff() { env := testcore.NewEnv(s.T()) tv := testvars.New(s.T())