From b9d89dbace3fccf30ea9f5e8006e1911d840ac5c Mon Sep 17 00:00:00 2001 From: Maruthi ChandraSekhar Vemuri Date: Fri, 24 Jul 2026 15:02:35 -0700 Subject: [PATCH 1/3] [SDK Ergonomics] NEXUS-519: Support Query-backed Nexus Operations --- common/nexus/links.go | 17 +++++++- go.mod | 2 + go.sum | 4 +- service/history/api/queryworkflow/api.go | 22 +++++++++- tests/query_workflow_test.go | 55 ++++++++++++++++++++++++ 5 files changed, 95 insertions(+), 5 deletions(-) diff --git a/common/nexus/links.go b/common/nexus/links.go index f75cf05cfd3..83f253bb3d2 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/go.mod b/go.mod index 7a9d0965bba..3ccfd1b0533 100644 --- a/go.mod +++ b/go.mod @@ -239,3 +239,5 @@ require ( ) tool golang.org/x/perf/cmd/benchstat + +replace go.temporal.io/api => github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6 diff --git a/go.sum b/go.sum index 95797622c25..ba94b7ac1ff 100644 --- a/go.sum +++ b/go.sum @@ -402,6 +402,8 @@ github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6 h1:bc1PZdBF42J/JEmvdhZ27an+KTC/RGDGnm9weIjykbg= +github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= github.com/temporalio/ringpop-go v0.1.0 h1:VSyNcMOLYOnXwTPp2Yevhl5vqhMDU3M6iqpoL7qkrQI= github.com/temporalio/ringpop-go v0.1.0/go.mod h1:RE+CHmY+kOZQk47AQaVzwrGmxpflnLgTd6EOK0853j4= github.com/temporalio/sqlparser v0.1.0 h1:jHbQyN3bfVLWGYPC/tLHpdjJxuL1YKOd1sf5OQPRM7o= @@ -479,8 +481,6 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= -go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4= -go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab h1:99wXW0317BBi49d6xgMdA0EZtvA+xbBUWV4HsTEGEcg= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw= go.temporal.io/sdk v1.44.0 h1:suitPDukX74rW3/N1FqvEbZTZVJJsxMKhv0KMa/j7pU= diff --git a/service/history/api/queryworkflow/api.go b/service/history/api/queryworkflow/api.go index 3f4ddfd35dc..1de268cc8e6 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,26 @@ 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 as long as there isnt an error. + reason := "Query processed" + if resp.Response.QueryRejected != nil { + reason = "Query rejected" + } + resp.Response.Link = &commonpb.Link{ + Variant: &commonpb.Link_Workflow_{ + Workflow: &commonpb.Link_Workflow{ + Namespace: request.Request.GetNamespace(), + WorkflowId: request.Request.Execution.GetWorkflowId(), + RunId: request.Request.Execution.GetRunId(), + 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/query_workflow_test.go b/tests/query_workflow_test.go index e9c30cb257a..971f3d66c05 100644 --- a/tests/query_workflow_test.go +++ b/tests/query_workflow_test.go @@ -140,6 +140,61 @@ func (s *QueryWorkflowSuite) TestQueryWorkflow_Consistent_PiggybackQuery() { s.Equal("pauseabc", queryResultStr) } +func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() { + env := testcore.NewEnv(s.T()) + workflowFn := func(ctx workflow.Context) error { + _ = workflow.SetQueryHandler(ctx, "test", func() (string, error) { + return "", nil + }) + workflow.GetSignalChannel(ctx, "test").Receive(ctx, nil) + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + env.SdkWorker().RegisterWorkflow(workflowFn) + + wid := "test-query-result-link-wid" + run, err := env.SdkClient().ExecuteWorkflow(ctx, + sdkclient.StartWorkflowOptions{ID: wid, TaskQueue: env.WorkerTaskQueue()}, workflowFn) + s.NoError(err) + + // use frontend client to query workflow so that the resp can be + // inspected directly instead of using the sdkclient + resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, + Query: &querypb.WorkflowQuery{QueryType: "test"}, + QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, // fail queries after workflow completes for TC + }) + 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("Query processed", link.GetReason()) + + s.NoError(env.SdkClient().SignalWorkflow(ctx, wid, "", "test", "")) + s.NoError(run.Get(ctx, nil)) + // run same after workflow closed, should reject due to QueryRejectCondition + resp, err = env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, + Query: &querypb.WorkflowQuery{QueryType: "test"}, + QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, + }) + 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("Query rejected", link.GetReason()) +} + func (s *QueryWorkflowSuite) TestQueryWorkflow_QueryWhileBackoff() { env := testcore.NewEnv(s.T()) tv := testvars.New(s.T()) From 009566a2a14757e403e41eab90f2b6f07a0cf538 Mon Sep 17 00:00:00 2001 From: Maruthi ChandraSekhar Vemuri Date: Mon, 27 Jul 2026 11:56:13 -0700 Subject: [PATCH 2/3] address comments --- common/links/validator.go | 10 +++++ common/links/validator_test.go | 33 +++++++++++++++- common/nexus/links_test.go | 50 ++++++++++++++++++++---- go.mod | 2 - go.sum | 4 +- service/history/api/queryworkflow/api.go | 11 +++--- tests/query_workflow_test.go | 2 +- 7 files changed, 93 insertions(+), 19 deletions(-) diff --git a/common/links/validator.go b/common/links/validator.go index 40d894e3ed4..e05c25b4d8d 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 74c39980620..816de48f597 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_test.go b/common/nexus/links_test.go index ac841d6105b..9bd981d5cef 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/go.mod b/go.mod index 3ccfd1b0533..7a9d0965bba 100644 --- a/go.mod +++ b/go.mod @@ -239,5 +239,3 @@ require ( ) tool golang.org/x/perf/cmd/benchstat - -replace go.temporal.io/api => github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6 diff --git a/go.sum b/go.sum index ba94b7ac1ff..95797622c25 100644 --- a/go.sum +++ b/go.sum @@ -402,8 +402,6 @@ github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6 h1:bc1PZdBF42J/JEmvdhZ27an+KTC/RGDGnm9weIjykbg= -github.com/temporalio/api-go v1.63.4-0.20260724233611-d22c21c3cfe6/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= github.com/temporalio/ringpop-go v0.1.0 h1:VSyNcMOLYOnXwTPp2Yevhl5vqhMDU3M6iqpoL7qkrQI= github.com/temporalio/ringpop-go v0.1.0/go.mod h1:RE+CHmY+kOZQk47AQaVzwrGmxpflnLgTd6EOK0853j4= github.com/temporalio/sqlparser v0.1.0 h1:jHbQyN3bfVLWGYPC/tLHpdjJxuL1YKOd1sf5OQPRM7o= @@ -481,6 +479,8 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= +go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4= +go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab h1:99wXW0317BBi49d6xgMdA0EZtvA+xbBUWV4HsTEGEcg= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw= go.temporal.io/sdk v1.44.0 h1:suitPDukX74rW3/N1FqvEbZTZVJJsxMKhv0KMa/j7pU= diff --git a/service/history/api/queryworkflow/api.go b/service/history/api/queryworkflow/api.go index 1de268cc8e6..b0b85a8daae 100644 --- a/service/history/api/queryworkflow/api.go +++ b/service/history/api/queryworkflow/api.go @@ -83,17 +83,18 @@ func Invoke( if retError != nil || resp.GetResponse() == nil { return } - // Add link to Workflow regardless of query status as long as there isnt an error. + // 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.Response.QueryRejected != nil { + if resp.GetResponse().GetQueryRejected() != nil { reason = "Query rejected" } resp.Response.Link = &commonpb.Link{ Variant: &commonpb.Link_Workflow_{ Workflow: &commonpb.Link_Workflow{ - Namespace: request.Request.GetNamespace(), - WorkflowId: request.Request.Execution.GetWorkflowId(), - RunId: request.Request.Execution.GetRunId(), + Namespace: nsEntry.Name().String(), + WorkflowId: workflowKey.WorkflowID, + RunId: workflowKey.RunID, Reason: reason, }, }, diff --git a/tests/query_workflow_test.go b/tests/query_workflow_test.go index 971f3d66c05..47dca94038a 100644 --- a/tests/query_workflow_test.go +++ b/tests/query_workflow_test.go @@ -165,7 +165,7 @@ func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() { Namespace: env.Namespace().String(), Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, Query: &querypb.WorkflowQuery{QueryType: "test"}, - QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, // fail queries after workflow completes for TC + QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, // set here too so the second query below differs only by the workflow being closed }) s.NoError(err) From 064ae7218c9e6b4a86f96db30d9a060bed52ae94 Mon Sep 17 00:00:00 2001 From: Maruthi ChandraSekhar Vemuri Date: Tue, 18 Aug 2026 20:53:27 -0700 Subject: [PATCH 3/3] address comments, add more tests --- tests/nexus_workflow_query_test.go | 105 +++++++++++++++++++++ tests/query_workflow_test.go | 146 +++++++++++++++++++++-------- 2 files changed, 213 insertions(+), 38 deletions(-) create mode 100644 tests/nexus_workflow_query_test.go diff --git a/tests/nexus_workflow_query_test.go b/tests/nexus_workflow_query_test.go new file mode 100644 index 00000000000..b120a5f7bad --- /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 47dca94038a..1bf7ac98448 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" @@ -142,57 +143,126 @@ func (s *QueryWorkflowSuite) TestQueryWorkflow_Consistent_PiggybackQuery() { func (s *QueryWorkflowSuite) TestQueryWorkflowResult_ContainsWorkflowLink() { env := testcore.NewEnv(s.T()) + queryName := "query" + signalName := "test" workflowFn := func(ctx workflow.Context) error { - _ = workflow.SetQueryHandler(ctx, "test", func() (string, error) { - return "", nil + orderStatus := "initialized" + _ = workflow.SetQueryHandler(ctx, queryName, func(_ string) (string, error) { + return orderStatus, nil }) - workflow.GetSignalChannel(ctx, "test").Receive(ctx, nil) + workflow.GetSignalChannel(ctx, signalName).Receive(ctx, &orderStatus) return nil } - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() + ctx := s.Context() env.SdkWorker().RegisterWorkflow(workflowFn) - wid := "test-query-result-link-wid" - run, err := env.SdkClient().ExecuteWorkflow(ctx, - sdkclient.StartWorkflowOptions{ID: wid, TaskQueue: env.WorkerTaskQueue()}, workflowFn) + wid := "nexus-query-workflow-tests" + run, err := env.SdkClient().ExecuteWorkflow( + ctx, + sdkclient.StartWorkflowOptions{ + ID: wid, + TaskQueue: env.WorkerTaskQueue()}, + workflowFn, + ) s.NoError(err) - // use frontend client to query workflow so that the resp can be - // inspected directly instead of using the sdkclient - resp, err := env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ - Namespace: env.Namespace().String(), - Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, - Query: &querypb.WorkflowQuery{QueryType: "test"}, - QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, // set here too so the second query below differs only by the workflow being closed - }) - 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) + }) + } + } - 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("Query processed", link.GetReason()) + // 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, + }, + } - s.NoError(env.SdkClient().SignalWorkflow(ctx, wid, "", "test", "")) + 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)) - // run same after workflow closed, should reject due to QueryRejectCondition - resp, err = env.FrontendClient().QueryWorkflow(ctx, &workflowservice.QueryWorkflowRequest{ - Namespace: env.Namespace().String(), - Execution: &commonpb.WorkflowExecution{WorkflowId: wid}, - Query: &querypb.WorkflowQuery{QueryType: "test"}, - QueryRejectCondition: enumspb.QUERY_REJECT_CONDITION_NOT_OPEN, - }) - 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("Query rejected", link.GetReason()) + 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() {