[Nexus-Chasm] Migrate OperationInvocationTaskExecutor (#9680)

## What changed?
This PR migrates the Nexus operation invocation task handler from HSM
version to Chasm.

## Why?
Migrating from HSM to Chasm


## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)



<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Introduces a new CHASM-based Nexus `StartOperation` execution path
with endpoint lookup, callback URL/token generation, and error
classification; mistakes could cause failed invocations, incorrect
retries/timeouts, or misrouted callbacks. Risk is mitigated somewhat by
strict task validation and added unit coverage, but the change touches
critical workflow/history integration and outbound request handling.
> 
> **Overview**
> Migrates Nexus operation invocation execution to CHASM by implementing
`OperationInvocationTaskHandler.Validate/Execute` end-to-end, including
endpoint resolution (ID with name fallback), callback URL selection
(system vs templated), callback token generation, timeout budgeting,
outbound StartOperation calls (HTTP or internal history service),
metrics/logging, and classification of results into operation state
transitions.
> 
> Adds supporting plumbing:
`OperationStore.NexusOperationInvocationData` and workflow
implementation that loads invocation input/headers from the scheduled
history event, plus a new
`MSPointer.LoadHistoryEvent`/`NodeBackend.LoadHistoryEvent` API.
Configuration is extended to parse `CallbackURLTemplate` into a
`*template.Template`, add `UseSystemCallbackURL`, and pass
`NumHistoryShards` for internal routing; new helper utilities centralize
callback building, error/failure conversion, and internal/HTTP start
logic.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
4b13977. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: gow

chasm/lib/nexusoperation/config.go

@@ -2,9 +2,11 @@ package nexusoperation
 
 import (
 	"strings"
+	"text/template"
 	"time"
 
 	"go.temporal.io/server/common/backoff"
+	"go.temporal.io/server/common/config"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/headers"
 	"go.temporal.io/server/common/rpc/interceptor"
@@ -107,9 +109,19 @@ var MaxOperationScheduleToCloseTimeout = dynamicconfig.NewNamespaceDurationSetti
 or a longer timeout than permitted will have their schedule-to-close timeout capped to this value. 0 implies no limit.`,
 )
 
-var CallbackURLTemplate = dynamicconfig.NewGlobalStringSetting(
+var CallbackURLTemplate = dynamicconfig.NewGlobalTypedSettingWithConverter(
 	"nexusoperation.callback.endpoint.template",
-	"unset",
+	func(in any) (*template.Template, error) {
+		s, err := dynamicconfig.ConvertStructure[string]("")(in)
+		if err != nil {
+			return nil, err
+		}
+		if s == "unset" {
+			return nil, nil
+		}
+		return template.New("NexusCallbackURL").Parse(s)
+	},
+	nil,
 	`Controls the template for generating callback URLs included in Nexus operation requests, which are used to deliver
 asynchronous completion for external endpoint targets. The template can be used to interpolate the {{.NamepaceName}}
 and {{.NamespaceID}} parameters to construct a publicly accessible URL.
@@ -141,6 +153,13 @@ Adding high-cardinality tags (like unique operation names) can significantly inc
 query complexity. Consider the cardinality impact when enabling these tags.`,
 )
 
+var UseSystemCallbackURL = dynamicconfig.NewGlobalBoolSetting(
+	"nexusoperation.useSystemCallbackURL",
+	true,
+	`Controls how the executor generates callback URLs for worker targets in Nexus Operations.
+When true, uses the fixed system callback URL for all worker targets.`,
+)
+
 var UseNewFailureWireFormat = dynamicconfig.NewNamespaceBoolSetting(
 	"nexusoperation.useNewFailureWireFormat",
 	true,
@@ -151,6 +170,7 @@ Added for safety. Defaults to true. Likely to be removed in future server versio
 type Config struct {
 	ChasmEnabled                        dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	ChasmNexusEnabled                   dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	NumHistoryShards                    int32
 	RequestTimeout                      dynamicconfig.DurationPropertyFnWithDestinationFilter
 	MinRequestTimeout                   dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	MaxConcurrentOperations             dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -161,16 +181,18 @@ type Config struct {
 	DisallowedOperationHeaders          dynamicconfig.TypedPropertyFn[[]string]
 	MaxOperationScheduleToCloseTimeout  dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	PayloadSizeLimit                    dynamicconfig.IntPropertyFnWithNamespaceFilter
-	CallbackURLTemplate                 dynamicconfig.StringPropertyFn
+	CallbackURLTemplate                 dynamicconfig.TypedPropertyFn[*template.Template]
+	UseSystemCallbackURL                dynamicconfig.BoolPropertyFn
 	UseNewFailureWireFormat             dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	RecordCancelRequestCompletionEvents dynamicconfig.BoolPropertyFn
 	RetryPolicy                         func() backoff.RetryPolicy
 }
 
-func configProvider(dc *dynamicconfig.Collection) *Config {
+func configProvider(dc *dynamicconfig.Collection, cfg *config.Persistence) *Config {
 	return &Config{
 		ChasmEnabled:                       dynamicconfig.EnableChasm.Get(dc),
 		ChasmNexusEnabled:                  ChasmNexusEnabled.Get(dc),
+		NumHistoryShards:                   cfg.NumHistoryShards,
 		RequestTimeout:                     RequestTimeout.Get(dc),
 		MinRequestTimeout:                  MinRequestTimeout.Get(dc),
 		MaxConcurrentOperations:            MaxConcurrentOperations.Get(dc),
@@ -181,8 +203,9 @@ func configProvider(dc *dynamicconfig.Collection) *Config {
 		DisallowedOperationHeaders:         DisallowedOperationHeaders.Get(dc),
 		MaxOperationScheduleToCloseTimeout: MaxOperationScheduleToCloseTimeout.Get(dc),
 		PayloadSizeLimit:                   dynamicconfig.BlobSizeLimitError.Get(dc),
-		UseNewFailureWireFormat:            UseNewFailureWireFormat.Get(dc),
 		CallbackURLTemplate:                CallbackURLTemplate.Get(dc),
+		UseSystemCallbackURL:               UseSystemCallbackURL.Get(dc),
+		UseNewFailureWireFormat:            UseNewFailureWireFormat.Get(dc),
 		RetryPolicy: func() backoff.RetryPolicy {
 			return backoff.NewExponentialRetryPolicy(
 				RetryPolicyInitialInterval.Get(dc)(),

chasm/lib/nexusoperation/operation.go

@@ -1,11 +1,15 @@
 package nexusoperation
 
 import (
+	"fmt"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
 	commonpb "go.temporal.io/api/common/v1"
 	failurepb "go.temporal.io/api/failure/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/chasm"
 	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
+	queueserrors "go.temporal.io/server/service/history/queues/errors"
 	"google.golang.org/protobuf/types/known/anypb"
 )
 
@@ -18,6 +22,16 @@ var ErrCancellationAlreadyRequested = serviceerror.NewFailedPrecondition("cancel
 // ErrOperationAlreadyCompleted is returned when trying to cancel an operation that has already completed.
 var ErrOperationAlreadyCompleted = serviceerror.NewFailedPrecondition("operation already completed")
 
+// InvocationData contains data needed to invoke a Nexus operation.
+type InvocationData struct {
+	// Input is the operation input payload.
+	Input *commonpb.Payload
+	// Header contains the Nexus headers for the operation.
+	Header map[string]string
+	// NexusLink is the link to the caller that scheduled this operation.
+	NexusLink nexus.Link
+}
+
 // OperationStore defines the interface that must be implemented by any parent component that wants to manage Nexus operations.
 // It's the responsibility of the parrent component to apply the appropriate state transitions to the operation.
 type OperationStore interface {
@@ -26,6 +40,8 @@ type OperationStore interface {
 	OnNexusOperationFailed(ctx chasm.MutableContext, operation *Operation, cause *failurepb.Failure) error
 	OnNexusOperationTimedOut(ctx chasm.MutableContext, operation *Operation, cause *failurepb.Failure) error
 	OnNexusOperationCompleted(ctx chasm.MutableContext, operation *Operation, result *commonpb.Payload, links []*commonpb.Link) error
+	// NexusOperationInvocationData loads invocation data (Input, Header, NexusLink) from the scheduled history event.
+	NexusOperationInvocationData(ctx chasm.Context, operation *Operation) (InvocationData, error)
 }
 
 // Operation is a CHASM component that represents a Nexus operation.
@@ -98,7 +114,7 @@ func (o *Operation) Cancel(ctx chasm.MutableContext, parentData *anypb.Any) erro
 }
 
 // OnStarted applies the started transition or delegates to the store if one is present.
-func (o *Operation) OnStarted(ctx chasm.MutableContext, _ *Operation, operationToken string, links []*commonpb.Link) error {
+func (o *Operation) OnStarted(ctx chasm.MutableContext, operationToken string, links []*commonpb.Link) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
 		return store.OnNexusOperationStarted(ctx, o, operationToken, links)
@@ -109,7 +125,7 @@ func (o *Operation) OnStarted(ctx chasm.MutableContext, _ *Operation, operationT
 }
 
 // OnCompleted applies the succeeded transition or delegates to the store if one is present.
-func (o *Operation) OnCompleted(ctx chasm.MutableContext, _ *Operation, result *commonpb.Payload, links []*commonpb.Link) error {
+func (o *Operation) OnCompleted(ctx chasm.MutableContext, result *commonpb.Payload, links []*commonpb.Link) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
 		return store.OnNexusOperationCompleted(ctx, o, result, links)
@@ -118,7 +134,7 @@ func (o *Operation) OnCompleted(ctx chasm.MutableContext, _ *Operation, result *
 }
 
 // OnFailed applies the failed transition or delegates to the store if one is present.
-func (o *Operation) OnFailed(ctx chasm.MutableContext, _ *Operation, cause *failurepb.Failure) error {
+func (o *Operation) OnFailed(ctx chasm.MutableContext, cause *failurepb.Failure) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
 		return store.OnNexusOperationFailed(ctx, o, cause)
@@ -127,7 +143,7 @@ func (o *Operation) OnFailed(ctx chasm.MutableContext, _ *Operation, cause *fail
 }
 
 // OnCancelled applies the canceled transition or delegates to the store if one is present.
-func (o *Operation) OnCancelled(ctx chasm.MutableContext, _ *Operation, cause *failurepb.Failure) error {
+func (o *Operation) OnCancelled(ctx chasm.MutableContext, cause *failurepb.Failure) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
 		return store.OnNexusOperationCancelled(ctx, o, cause)
@@ -136,10 +152,84 @@ func (o *Operation) OnCancelled(ctx chasm.MutableContext, _ *Operation, cause *f
 }
 
 // OnTimedOut applies the timed out transition or delegates to the store if one is present.
-func (o *Operation) OnTimedOut(ctx chasm.MutableContext, _ *Operation, cause *failurepb.Failure) error {
+func (o *Operation) OnTimedOut(ctx chasm.MutableContext, cause *failurepb.Failure) error {
 	store, ok := o.Store.TryGet(ctx)
 	if ok {
 		return store.OnNexusOperationTimedOut(ctx, o, cause)
 	}
 	return TransitionTimedOut.Apply(o, ctx, EventTimedOut{})
 }
+
+// loadStartArgs is a ReadComponent callback that loads the start arguments from the operation.
+func (o *Operation) loadStartArgs(
+	ctx chasm.Context,
+	_ chasm.NoValue,
+) (startArgs, error) {
+	store, ok := o.Store.TryGet(ctx)
+	if !ok {
+		// TODO: For standalone operations, load invocation data from the operation state.
+		return startArgs{}, serviceerror.NewInternal("no store available to load invocation data")
+	}
+	invocationData, err := store.NexusOperationInvocationData(ctx, o)
+	if err != nil {
+		return startArgs{}, err
+	}
+
+	serializedRef, err := ctx.Ref(o)
+	if err != nil {
+		return startArgs{}, err
+	}
+
+	return startArgs{
+		endpointName:           o.GetEndpoint(),
+		endpointID:             o.GetEndpointId(),
+		service:                o.GetService(),
+		operation:              o.GetOperation(),
+		requestID:              o.GetRequestId(),
+		currentTime:            ctx.Now(o),
+		scheduledTime:          o.GetScheduledTime().AsTime(),
+		scheduleToCloseTimeout: o.GetScheduleToCloseTimeout().AsDuration(),
+		scheduleToStartTimeout: o.GetScheduleToStartTimeout().AsDuration(),
+		startToCloseTimeout:    o.GetStartToCloseTimeout().AsDuration(),
+		payload:                invocationData.Input,
+		header:                 invocationData.Header,
+		nexusLink:              invocationData.NexusLink,
+		serializedRef:          serializedRef,
+	}, nil
+}
+
+// saveResult is an UpdateComponent callback that saves the invocation outcome.
+func (o *Operation) saveResult(
+	ctx chasm.MutableContext,
+	input saveResultInput,
+) (chasm.NoValue, error) {
+	switch r := input.result.(type) {
+	case invocationResultOK:
+		if r.response.Pending != nil {
+			return nil, o.OnStarted(ctx, r.response.Pending.Token, r.links)
+		}
+		return nil, o.OnCompleted(ctx, r.response.Successful, r.links)
+	case invocationResultFail:
+		return nil, o.OnFailed(ctx, r.failure)
+	case invocationResultCanceled:
+		return nil, o.OnCancelled(ctx, r.failure)
+	case invocationResultRetry:
+		return nil, transitionAttemptFailed.Apply(o, ctx, EventAttemptFailed{
+			Failure:     r.failure,
+			RetryPolicy: input.retryPolicy(),
+		})
+	case invocationResultTimeout:
+		return nil, o.OnTimedOut(ctx, &failurepb.Failure{
+			Message: "operation timed out",
+			FailureInfo: &failurepb.Failure_TimeoutFailureInfo{
+				TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{
+					TimeoutType: r.timeoutType,
+				},
+			},
+		})
+	default:
+		return nil, queueserrors.NewUnprocessableTaskError(
+			fmt.Sprintf("unrecognized invocation result %T", input.result),
+		)
+	}
+}