Expose Nexus Endpoint in Nexus Info (temporalio#2291)

Author: Quinn-With-Two-Ns

internal/internal_nexus_task_handler.go

@@ -450,6 +450,7 @@ func (h *nexusTaskHandler) newNexusOperationContext(response *workflowservice.Po
 
 	return &NexusOperationContext{
 		client:         h.client,
+		Endpoint:       response.GetRequest().GetEndpoint(),
 		Namespace:      h.namespace,
 		TaskQueue:      h.taskQueueName,
 		metricsHandler: metricsHandler,

internal/internal_workflow_testsuite.go

@@ -3568,6 +3568,7 @@ func (h *testNexusOperationHandle) newStartTask() *workflowservice.PollNexusTask
 		Request: &nexuspb.Request{
 			ScheduledTime: timestamppb.Now(),
 			Header:        h.params.nexusHeader,
+			Endpoint:      h.params.client.Endpoint(),
 			Capabilities: &nexuspb.Request_Capabilities{
 				TemporalFailureResponses: true,
 			},
@@ -3595,6 +3596,7 @@ func (h *testNexusOperationHandle) newCancelTask() *workflowservice.PollNexusTas
 		Request: &nexuspb.Request{
 			ScheduledTime: timestamppb.Now(),
 			Header:        h.params.nexusHeader,
+			Endpoint:      h.params.client.Endpoint(),
 			Capabilities: &nexuspb.Request_Capabilities{
 				TemporalFailureResponses: true,
 			},

internal/nexus_operations.go

@@ -28,13 +28,17 @@ type NexusOperationInfo struct {
 	Namespace string
 	// The task queue of the worker handling this Nexus operation.
 	TaskQueue string
+	// The endpoint this request was addressed to before forwarding to the worker.
+	// Supported from server version 1.30.0.
+	Endpoint string
 }
 
 // NexusOperationContext is an internal only struct that holds fields used by the temporalnexus functions.
 type NexusOperationContext struct {
 	client         Client
 	Namespace      string
 	TaskQueue      string
+	Endpoint       string
 	metricsHandler metrics.Handler
 	log            log.Logger
 	registry       *registry
@@ -54,6 +58,7 @@ func (nc *nexusOperationEnvironment) GetOperationInfo(ctx context.Context) Nexus
 		panic("temporalnexus GetInfo: Not a valid Nexus context")
 	}
 	return NexusOperationInfo{
+		Endpoint:  nctx.Endpoint,
 		Namespace: nctx.Namespace,
 		TaskQueue: nctx.TaskQueue,
 	}

test/nexus_test.go

@@ -548,7 +548,7 @@ func TestOperationInfo(t *testing.T) {
 
 	op := nexus.NewSyncOperation("op", func(ctx context.Context, outcome string, o nexus.StartOperationOptions) (string, error) {
 		info := temporalnexus.GetOperationInfo(ctx)
-		return info.Namespace + ":" + info.TaskQueue, nil
+		return info.Endpoint + ":" + info.Namespace + ":" + info.TaskQueue, nil
 	})
 
 	wf := func(ctx workflow.Context, outcome string) (string, error) {
@@ -586,7 +586,7 @@ func TestOperationInfo(t *testing.T) {
 	require.NoError(t, err)
 	var result string
 	require.NoError(t, run.Get(ctx, &result))
-	require.Equal(t, tc.testConfig.Namespace+":"+tc.taskQueue, result)
+	require.Equal(t, tc.endpoint+":"+tc.testConfig.Namespace+":"+tc.taskQueue, result)
 }
 
 func TestSyncOperationFromWorkflow(t *testing.T) {