hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
## 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
This commit is contained in:
Sean Kane
2025-11-01 00:46:28 +00:00
committed by GitHub
parent c6b1e2e3d2
commit 16df1dc859
4 changed files with 69 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chasm/lib/callback/chasm_invocation.go
+7 -7
View File
@@ -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}
}
chasm/lib/callback/executors.go
+24 -6
View File
@@ -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,
}
}
chasm/lib/callback/executors_test.go
+25 -31
View File
@@ -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)
},
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
},
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
chasm/lib/callback/nexus_invocation.go
+13 -13
View File
@@ -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}
}