Fixed fx.In embedding in output types (#8584)

spkane31 · web-flow · commit 4eafd4a5b5cb · 2025-11-01T00:46:28.000Z
## What changed?
`fx.In` shouldn't be embedded in the `{Invocation,Backoff}TaskExecutor`
because these are output parameters.

## Why?
These will cause tests to fail once mutable state is integrated.

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

## Potential risks
No
diff --git a/chasm/lib/callback/chasm_invocation.go b/chasm/lib/callback/chasm_invocation.go
@@ -59,32 +59,32 @@ func (c chasmInvocation) Invoke(
 	// Get back the base64-encoded ComponentRef from the header.
 	encodedRef, ok := c.nexus.GetHeader()[commonnexus.CallbackTokenHeader]
 	if !ok {
-		return invocationResultFail{logInternalError(e.Logger, "callback missing token", nil)}
+		return invocationResultFail{logInternalError(e.logger, "callback missing token", nil)}
 	}
 
 	decodedRef, err := base64.RawURLEncoding.DecodeString(encodedRef)
 	if err != nil {
-		return invocationResultFail{logInternalError(e.Logger, "failed to decode CHASM ComponentRef", err)}
+		return invocationResultFail{logInternalError(e.logger, "failed to decode CHASM ComponentRef", err)}
 	}
 
 	// Validate that the bytes are a valid ChasmComponentRef
 	ref := &persistencespb.ChasmComponentRef{}
 	err = proto.Unmarshal(decodedRef, ref)
 	if err != nil {
-		return invocationResultFail{logInternalError(e.Logger, "failed to unmarshal CHASM ComponentRef", err)}
+		return invocationResultFail{logInternalError(e.logger, "failed to unmarshal CHASM ComponentRef", err)}
 	}
 
 	request, err := c.getHistoryRequest(decodedRef)
 	if err != nil {
-		return invocationResultFail{logInternalError(e.Logger, "failed to build history request: %v", err)}
+		return invocationResultFail{logInternalError(e.logger, "failed to build history request: %v", err)}
 	}
 
 	// RPC to History for cross-shard completion delivery.
-	_, err = e.HistoryClient.CompleteNexusOperationChasm(ctx, request)
+	_, err = e.historyClient.CompleteNexusOperationChasm(ctx, request)
 	if err != nil {
-		msg := logInternalError(e.Logger, "failed to complete Nexus operation: %v", err)
+		msg := logInternalError(e.logger, "failed to complete Nexus operation: %v", err)
 		if isRetryableRPCResponse(err) {
-			return invocationResultRetry{err: msg, retryPolicy: e.Config.RetryPolicy()}
+			return invocationResultRetry{err: msg, retryPolicy: e.config.RetryPolicy()}
 		}
 		return invocationResultFail{msg}
 	}
diff --git a/chasm/lib/callback/executors.go b/chasm/lib/callback/executors.go
@@ -23,7 +23,14 @@ type HTTPCallerProvider func(queues.NamespaceIDAndDestination) HTTPCaller
 
 func NewInvocationTaskExecutor(opts InvocationTaskExecutorOptions) *InvocationTaskExecutor {
 	return &InvocationTaskExecutor{
-		InvocationTaskExecutorOptions: opts,
+		config:             opts.Config,
+		namespaceRegistry:  opts.NamespaceRegistry,
+		metricsHandler:     opts.MetricsHandler,
+		logger:             opts.Logger,
+		httpCallerProvider: opts.HTTPCallerProvider,
+		httpTraceProvider:  opts.HTTPTraceProvider,
+		historyClient:      opts.HistoryClient,
+		chasmEngine:        opts.ChasmEngine,
 	}
 }
 
@@ -41,7 +48,14 @@ type InvocationTaskExecutorOptions struct {
 }
 
 type InvocationTaskExecutor struct {
-	InvocationTaskExecutorOptions
+	config             *Config
+	namespaceRegistry  namespace.Registry
+	metricsHandler     metrics.Handler
+	logger             log.Logger
+	httpCallerProvider HTTPCallerProvider
+	httpTraceProvider  commonnexus.HTTPClientTraceProvider
+	historyClient      resource.HistoryClient
+	chasmEngine        chasm.Engine
 }
 
 func (e InvocationTaskExecutor) Execute(ctx context.Context, ref chasm.ComponentRef, attrs chasm.TaskAttributes, task *callbackspb.InvocationTask) error {
@@ -106,7 +120,7 @@ func (e InvocationTaskExecutor) Invoke(
 	taskAttr chasm.TaskAttributes,
 	task *callbackspb.InvocationTask,
 ) error {
-	ns, err := e.NamespaceRegistry.GetNamespaceByID(namespace.ID(ref.NamespaceID))
+	ns, err := e.namespaceRegistry.GetNamespaceByID(namespace.ID(ref.NamespaceID))
 	if err != nil {
 		return fmt.Errorf("failed to get namespace by ID: %w", err)
 	}
@@ -123,7 +137,7 @@ func (e InvocationTaskExecutor) Invoke(
 
 	callCtx, cancel := context.WithTimeout(
 		ctx,
-		e.Config.RequestTimeout(ns.Name().String(), taskAttr.Destination),
+		e.config.RequestTimeout(ns.Name().String(), taskAttr.Destination),
 	)
 	defer cancel()
 
@@ -138,7 +152,9 @@ func (e InvocationTaskExecutor) Invoke(
 }
 
 type BackoffTaskExecutor struct {
-	BackoffTaskExecutorOptions
+	config         *Config
+	metricsHandler metrics.Handler
+	logger         log.Logger
 }
 
 type BackoffTaskExecutorOptions struct {
@@ -151,7 +167,9 @@ type BackoffTaskExecutorOptions struct {
 
 func NewBackoffTaskExecutor(opts BackoffTaskExecutorOptions) *BackoffTaskExecutor {
 	return &BackoffTaskExecutor{
-		BackoffTaskExecutorOptions: opts,
+		config:         opts.Config,
+		metricsHandler: opts.MetricsHandler,
+		logger:         opts.Logger,
 	}
 }
 
diff --git a/chasm/lib/callback/executors_test.go b/chasm/lib/callback/executors_test.go
@@ -229,21 +229,19 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 			})
 
 			executor := InvocationTaskExecutor{
-				InvocationTaskExecutorOptions: InvocationTaskExecutorOptions{
-					Config: &Config{
-						RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
-						RetryPolicy: func() backoff.RetryPolicy {
-							return backoff.NewExponentialRetryPolicy(time.Second)
-						},
-					},
-					NamespaceRegistry: nsRegistry,
-					MetricsHandler:    metricsHandler,
-					Logger:            logger,
-					HTTPCallerProvider: func(nid queues.NamespaceIDAndDestination) HTTPCaller {
-						return tc.caller
+				config: &Config{
+					RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
+					RetryPolicy: func() backoff.RetryPolicy {
+						return backoff.NewExponentialRetryPolicy(time.Second)
 					},
-					ChasmEngine: mockEngine,
 				},
+				namespaceRegistry: nsRegistry,
+				metricsHandler:    metricsHandler,
+				logger:            logger,
+				httpCallerProvider: func(nid queues.NamespaceIDAndDestination) HTTPCaller {
+					return tc.caller
+				},
+				chasmEngine: mockEngine,
 			}
 
 			// Create ComponentRef
@@ -318,15 +316,13 @@ func TestProcessBackoffTask(t *testing.T) {
 	}
 
 	executor := BackoffTaskExecutor{
-		BackoffTaskExecutorOptions: BackoffTaskExecutorOptions{
-			Config: &Config{
-				RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
-				RetryPolicy: func() backoff.RetryPolicy {
-					return backoff.NewExponentialRetryPolicy(time.Second)
-				},
+		config: &Config{
+			RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
+			RetryPolicy: func() backoff.RetryPolicy {
+				return backoff.NewExponentialRetryPolicy(time.Second)
 			},
-			Logger: logger,
 		},
+		logger: logger,
 	}
 
 	// Execute the backoff task
@@ -644,19 +640,17 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 			})
 
 			executor := InvocationTaskExecutor{
-				InvocationTaskExecutorOptions: InvocationTaskExecutorOptions{
-					Config: &Config{
-						RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
-						RetryPolicy: func() backoff.RetryPolicy {
-							return backoff.NewExponentialRetryPolicy(time.Second)
-						},
+				config: &Config{
+					RequestTimeout: dynamicconfig.GetDurationPropertyFnFilteredByDestination(time.Second),
+					RetryPolicy: func() backoff.RetryPolicy {
+						return backoff.NewExponentialRetryPolicy(time.Second)
 					},
-					NamespaceRegistry: nsRegistry,
-					MetricsHandler:    metrics.NoopMetricsHandler,
-					Logger:            logger,
-					HistoryClient:     historyClient,
-					ChasmEngine:       mockEngine,
 				},
+				namespaceRegistry: nsRegistry,
+				metricsHandler:    metrics.NoopMetricsHandler,
+				logger:            logger,
+				historyClient:     historyClient,
+				chasmEngine:       mockEngine,
 			}
 
 			// Create ComponentRef
diff --git a/chasm/lib/callback/nexus_invocation.go b/chasm/lib/callback/nexus_invocation.go
@@ -68,8 +68,8 @@ func (n nexusInvocation) Invoke(
 	task *callbackspb.InvocationTask,
 	taskAttr chasm.TaskAttributes,
 ) invocationResult {
-	if e.HTTPTraceProvider != nil {
-		traceLogger := log.With(e.Logger,
+	if e.httpTraceProvider != nil {
+		traceLogger := log.With(e.logger,
 			tag.WorkflowNamespace(ns.Name().String()),
 			tag.Operation("CompleteNexusOperation"),
 			tag.NewStringTag("destination", taskAttr.Destination),
@@ -78,7 +78,7 @@ func (n nexusInvocation) Invoke(
 			tag.AttemptStart(time.Now().UTC()),
 			tag.Attempt(n.attempt),
 		)
-		if trace := e.HTTPTraceProvider.NewTrace(n.attempt, traceLogger); trace != nil {
+		if trace := e.httpTraceProvider.NewTrace(n.attempt, traceLogger); trace != nil {
 			ctx = httptrace.WithClientTrace(ctx, trace)
 		}
 	}
@@ -96,7 +96,7 @@ func (n nexusInvocation) Invoke(
 		request.Header.Set(k, v)
 	}
 
-	caller := e.HTTPCallerProvider(queues.NamespaceIDAndDestination{
+	caller := e.httpCallerProvider(queues.NamespaceIDAndDestination{
 		NamespaceID: ns.ID().String(),
 		Destination: taskAttr.Destination,
 	})
@@ -107,12 +107,12 @@ func (n nexusInvocation) Invoke(
 	namespaceTag := metrics.NamespaceTag(ns.Name().String())
 	destTag := metrics.DestinationTag(taskAttr.Destination)
 	statusCodeTag := metrics.OutcomeTag(outcomeTag(ctx, response, err))
-	e.MetricsHandler.Counter(RequestCounter.Name()).Record(1, namespaceTag, destTag, statusCodeTag)
-	e.MetricsHandler.Timer(RequestLatencyHistogram.Name()).Record(time.Since(startTime), namespaceTag, destTag, statusCodeTag)
+	e.metricsHandler.Counter(RequestCounter.Name()).Record(1, namespaceTag, destTag, statusCodeTag)
+	e.metricsHandler.Timer(RequestLatencyHistogram.Name()).Record(time.Since(startTime), namespaceTag, destTag, statusCodeTag)
 
 	if err != nil {
-		e.Logger.Error("Callback request failed with error", tag.Error(err))
-		return invocationResultRetry{err: err, retryPolicy: e.Config.RetryPolicy()}
+		e.logger.Error("Callback request failed with error", tag.Error(err))
+		return invocationResultRetry{err: err, retryPolicy: e.config.RetryPolicy()}
 	}
 
 	if response.StatusCode >= 200 && response.StatusCode < 300 {
@@ -121,18 +121,18 @@ func (n nexusInvocation) Invoke(
 		// propagate errors to the machine.
 		if _, err = io.Copy(io.Discard, response.Body); err == nil {
 			if err = response.Body.Close(); err != nil {
-				e.Logger.Error("Callback request failed with error", tag.Error(err))
-				return invocationResultRetry{err: err, retryPolicy: e.Config.RetryPolicy()}
+				e.logger.Error("Callback request failed with error", tag.Error(err))
+				return invocationResultRetry{err: err, retryPolicy: e.config.RetryPolicy()}
 			}
 		}
 		return invocationResultOK{}
 	}
 
 	retryable := isRetryableHTTPResponse(response)
-	err = readHandlerErrFromResponse(response, e.Logger)
-	e.Logger.Error("Callback request failed", tag.Error(err), tag.NewStringTag("status", response.Status), tag.NewBoolTag("retryable", retryable))
+	err = readHandlerErrFromResponse(response, e.logger)
+	e.logger.Error("Callback request failed", tag.Error(err), tag.NewStringTag("status", response.Status), tag.NewBoolTag("retryable", retryable))
 	if retryable {
-		return invocationResultRetry{err: err, retryPolicy: e.Config.RetryPolicy()}
+		return invocationResultRetry{err: err, retryPolicy: e.config.RetryPolicy()}
 	}
 	return invocationResultFail{err}
 }
