162
_ chasm.TaskAttributes,
163
_ *schedulerpb.InvokerExecuteTask,
165
>
var result executeResult
166
>
167
>
var invoker *Invoker
168
>
var scheduler *Scheduler
169
>
var lastCompletionState *schedulerpb.LastCompletionResult
170
>
var schedulerRef []byte
171
>
var now time.Time
172
>
173
>
// Read and deep copy returned components, since we'll continue to access them
174
>
// outside of this function (outside of the MS lock).
175
>
_, err := chasm.ReadComponent(
176
>
ctx,
177
>
invokerRef,
178
>
func(i *Invoker, ctx chasm.Context, _ any) (struct{}, error) {
179
>
invoker = &Invoker{
180
>
InvokerState: common.CloneProto(i.InvokerState),
181
>
}
182
>
183
>
s := i.Scheduler.Get(ctx)
184
>
scheduler = &Scheduler{
185
>
SchedulerState: common.CloneProto(s.SchedulerState),
186
>
cacheConflictToken: s.cacheConflictToken,
187
>
compiledSpec: s.compiledSpec,
188
>
}
189
>
190
>
lcs := s.LastCompletionResult.Get(ctx)
191
>
lastCompletionState = common.CloneProto(lcs)
192
>
193
>
// Capture the scheduler's component ref so a per-start completion callback (carrying that
194
>
// start's request ID in its token) can be built outside the MS lock.
195
>
ref, err := ctx.Ref(s)
196
>
if err != nil {
197
return struct{}{}, err
198
}
200
>
201
>
// Captured here for applyBackoff (runs outside this closure): BackoffTime
202
>
// must be framework-clock to match LastProcessedTime and task deadlines.
203
>
now = ctx.Now(i)
204
>
205
>
return struct{}{}, nil
206
},
207
nil,
208
)
210
return fmt.Errorf("failed to read component: %w", err)
211
}
213
return errors.New("scheduler component was nil after read")
214
}
215
217
>
metricsHandler := newTaggedMetricsHandler(h.metricsHandler, scheduler)
218
>
metricsHandler.Counter(metrics.ScheduleInvokerExecuteTask.Name()).Record(1, metrics.OutcomeTag(outcomeFired), metrics.ReasonTag(reasonNone))
219
>
220
>
// Terminate, cancel, and start workflows. The result struct contains the
221
>
// complete outcome of all requests executed in a single batch.
222
>
//
223
>
// Invoker will never have work pending for more than one of these calls (terminate,
224
>
// cancel, start) at a time, so it isn't sensible to run them in parallel. The
225
>
// structure below is simply for code simplicity.
226
>
ictx := h.newInvokerTaskHandlerContext(ctx, scheduler)
227
>
result = result.Append(h.terminateWorkflows(ictx, logger, metricsHandler, scheduler, invoker.GetTerminateWorkflows()))
228
>
result = result.Append(h.cancelWorkflows(ictx, logger, metricsHandler, scheduler, invoker.GetCancelWorkflows()))
229
>
result = result.Append(h.startWorkflows(ictx, logger, metricsHandler, scheduler, invoker, lastCompletionState, schedulerRef, now))
230
>
231
>
// Record action results on the Invoker (internal state), as well as the
232
>
// Scheduler (user-facing metrics).
233
>
_, _, err = chasm.UpdateComponent(
234
>
ctx,
235
>
invokerRef,
236
>
func(i *Invoker, ctx chasm.MutableContext, _ any) (chasm.NoValue, error) {
237
>
s := i.Scheduler.Get(ctx)
238
>
// Use newlyStarted (not len(result.CompletedStarts)) so a concurrent
239
>
// ExecuteTask's duplicate StartWorkflow can't inflate ActionCount.
240
>
newlyStarted, droppedDuplicates := i.recordExecuteResult(ctx, &result)
241
>
s.recordActionResult(&schedulerActionResult{actionCount: int64(newlyStarted)})
242
>
if droppedDuplicates > 0 {
243
h.recordDuplicateExecuteDrops(s, droppedDuplicates)
244
}
246
},
247
nil,
248
)
250
return fmt.Errorf("failed to update component state: %w", err)
251
}
252
254
}
255