103
}
104
105
>
callbackURL, err := buildCallbackURL(h.config.UseSystemCallbackURL(), h.config.CallbackURLTemplate(), ns, endpoint)
operation_tasks.go
106
>
if err != nil {
107
return fmt.Errorf("failed to build callback URL: %w", err)
108
}
109
110
>
token, err := h.generateCallbackToken(args.serializedRef, args.requestID)
operation_tasks.go
111
>
if err != nil {
112
return err
113
}
114
116
>
callTimeout := h.config.RequestTimeout(ns.Name().String(), attrs.Destination)
117
>
var timeoutType enumspb.TimeoutType
118
>
// Adjust timeout based on remaining operation timeouts.
119
>
// ScheduleToStart takes precedence over ScheduleToClose since it is already capped by it.
120
>
if args.scheduleToStartTimeout > 0 {
121
callTimeout = min(callTimeout, args.scheduleToStartTimeout-elapsed)
122
timeoutType = enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START
124
>
callTimeout = min(callTimeout, args.scheduleToCloseTimeout-elapsed)
operation_tasks.go
125
>
timeoutType = enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE
126
>
}
127
128
// Inform the handler of the operation timeout via header.
129
// StartToClose takes precedence over ScheduleToClose since it is already capped by it.
131
>
if args.startToCloseTimeout > 0 {
132
opTimeout = args.startToCloseTimeout
133
}
136
>
}
138
>
// Set the operation timeout header if not already set.
139
>
if opTimeoutHeader := header.Get(nexus.HeaderOperationTimeout); opTimeout != maxDuration && opTimeoutHeader == "" {
140
>
header.Set(nexus.HeaderOperationTimeout, commonnexus.FormatDuration(opTimeout))
operation_tasks.go
141
>
}
142
// If this request is handled by a newer server that supports Nexus failure serialization, trigger that behavior.
144
>
header.Set(nexusrpc.HeaderTemporalNexusFailureSupport, "true")
145
>
}
146
148
>
defer cancel()
149
>
// Set this value on the parent context so that our custom HTTP caller can mutate it since we cannot
150
>
// access response headers directly.
151
>
callCtx = context.WithValue(callCtx, commonnexus.FailureSourceContextKey, &atomic.Value{})
152
>
153
>
options := nexus.StartOperationOptions{
154
>
Header: header,
155
>
CallbackURL: callbackURL,
156
>
RequestID: args.requestID,
157
>
CallbackHeader: nexus.Header{
158
>
commonnexus.CallbackTokenHeader: token,
159
>
},
160
>
Links: args.nexusLinks,
161
>
}
162
>
163
>
traceCtx := invocationTraceContext{
164
>
operationTag: "StartOperation",
165
>
namespaceName: ns.Name().String(),
166
>
targetNamespaceID: endpoint.GetEndpoint().GetSpec().GetTarget().GetWorker().GetNamespaceId(),
167
>
requestID: args.requestID,
168
>
operation: args.operation,
169
>
endpointName: args.endpointName,
170
>
workflowID: opRef.BusinessID,
171
>
runID: opRef.RunID,
172
>
attemptStart: args.currentTime.UTC(),
173
>
attempt: task.GetAttempt(),
174
>
}
175
>
176
>
invocation, err := h.newInvocation(
177
>
callCtx,
178
>
ns,
179
>
endpoint,
180
>
traceCtx.endpointName,
181
>
args.service,
182
>
callTimeout,
183
>
timeoutType,
184
>
traceCtx,
185
>
)
186
>
if err != nil {
187
return fmt.Errorf("failed to construct invocation: %w", err)
188
}
189
>
startTime := time.Now() // nolint:forbidigo // Time can be used for timing metrics.
operation_tasks.go
190
>
response, callErr := invocation.Start(callCtx, args, options)
191
>
callDuration := time.Since(startTime)
192
>
if validationErr := h.validateStartResult(ns, response); validationErr != nil {
193
callErr = validationErr
194
}
196
>
197
>
h.recordCallOutcome(endpoint, startCallOutcomeTag(callCtx, response, callErr), callErr, callDuration, failureSource, traceCtx)
198
>
199
>
result, err := newInvocationResult(response, callErr)
200
>
if err != nil {
201
return fmt.Errorf("failed to construct invocation result: %w", err)
202
}
203
>
_, _, saveErr := chasm.UpdateComponent(ctx, opRef, (*Operation).saveInvocationResult, saveInvocationResultInput{
operation_tasks.go
204
>
result: result,
205
>
retryPolicy: h.config.RetryPolicy(),
206
>
})
207
>
208
>
if callErr != nil && isDestinationDown(callErr) {
209
saveErr = queueserrors.NewDestinationDownError(callErr.Error(), saveErr)
210
}
211
213
}
214
215
// buildRequestHeader returns a copy of the supplied header, or a new map if nil.
217
>
if header == nil {
218
>
return make(nexus.Header, 2) // To set the failure support and timeout headers.
operation_tasks.go
219
>
}
220
return nexus.Header(maps.Clone(header))
221
}