From 74731b7e5f38e49c3cf514cf70006bd717be1199 Mon Sep 17 00:00:00 2001 From: Maruthi ChandraSekhar Vemuri Date: Fri, 24 Jul 2026 15:02:35 -0700 Subject: [PATCH 1/2] [SDK Ergonomics] NEXUS-519: Support Query-backed Nexus Operations --- common/nexus/links.go | 17 +++++++- go.mod | 2 + go.sum | 2 + service/history/api/queryworkflow/api.go | 22 +++++++++- tests/query_workflow_test.go | 55 ++++++++++++++++++++++++ 5 files changed, 95 insertions(+), 3 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 8b154b7a704..3b80505d411 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 dce6989fb82..6ab77850e95 100644 --- a/go.sum +++ b/go.sum @@ -404,6 +404,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= 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/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/sqlparser v0.0.0-20260722001706-17d16cfe1da5 h1:5V5CN6pyeYt219lJF8CXtl3UC2DDzynXLNNZ5gjK2GE= github.com/temporalio/sqlparser v0.0.0-20260722001706-17d16cfe1da5/go.mod h1:143qKdh3G45IgV9p+gbAwp3ikRDI8mxsijFiXDfuxsw= github.com/temporalio/tchannel-go v1.22.1-0.20220818200552-1be8d8cffa5b/go.mod h1:c+V9Z/ZgkzAdyGvHrvC5AsXgN+M9Qwey04cBdKYzV7U= 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 5658029902c..7d7d06529a4 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 203efe9dec54060f977cf925f05b00cbbd93d426 Mon Sep 17 00:00:00 2001 From: Maruthi ChandraSekhar Vemuri Date: Mon, 27 Jul 2026 11:56:13 -0700 Subject: [PATCH 2/2] address comments --- common/links/validator.go | 10 +++++ common/links/validator_test.go | 33 +++++++++++++++- common/nexus/links_test.go | 50 ++++++++++++++++++++---- go.mod | 2 - go.sum | 2 - service/history/api/queryworkflow/api.go | 11 +++--- tests/query_workflow_test.go | 2 +- 7 files changed, 91 insertions(+), 19 deletions(-) diff --git a/common/links/validator.go b/common/links/validator.go index e690a80f350..550b4461360 100644 --- a/common/links/validator.go +++ b/common/links/validator.go @@ -57,6 +57,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 3b80505d411..8b154b7a704 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 6ab77850e95..dce6989fb82 100644 --- a/go.sum +++ b/go.sum @@ -404,8 +404,6 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= 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/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/sqlparser v0.0.0-20260722001706-17d16cfe1da5 h1:5V5CN6pyeYt219lJF8CXtl3UC2DDzynXLNNZ5gjK2GE= github.com/temporalio/sqlparser v0.0.0-20260722001706-17d16cfe1da5/go.mod h1:143qKdh3G45IgV9p+gbAwp3ikRDI8mxsijFiXDfuxsw= github.com/temporalio/tchannel-go v1.22.1-0.20220818200552-1be8d8cffa5b/go.mod h1:c+V9Z/ZgkzAdyGvHrvC5AsXgN+M9Qwey04cBdKYzV7U= 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 7d7d06529a4..29115036f14 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)