469
func (r *TaskGeneratorImpl) GenerateScheduleSpeculativeWorkflowTaskTasks(
470
workflowTask *historyi.WorkflowTaskInfo,
472
>
473
>
var scheduleToStartTimeout time.Duration
474
>
if r.mutableState.IsStickyTaskQueueSet() {
475
scheduleToStartTimeout = timestamp.DurationValue(r.mutableState.GetExecutionInfo().StickyScheduleToStartTimeout)
477
>
// Speculative WT has ScheduleToStart timeout even on normal task queue.
478
>
// Normally WT should be added to matching right after being created
479
>
// (i.e. from UpdateWorkflowExecution API handler), but if this "add" operation failed,
480
>
// there is no good way to handle the error.
481
>
// In this case WT will be timed out (as if it was on sticky task queue),
482
>
// and new normal WT will be created.
483
>
// Note: this timer will also fire if workflow received an update,
484
>
// but there is no workers available. Speculative WT will time out, and normal WT will be created.
485
>
scheduleToStartTimeout = tasks.SpeculativeWorkflowTaskScheduleToStartTimeout
486
>
}
487
488
>
isSpeculative := workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE
task_generator.go
489
>
wttt := &tasks.WorkflowTaskTimeoutTask{
490
>
// TaskID is set by shard
491
>
WorkflowKey: r.mutableState.GetWorkflowKey(),
492
>
VisibilityTimestamp: workflowTask.ScheduledTime.Add(scheduleToStartTimeout),
493
>
TimeoutType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
494
>
EventID: workflowTask.ScheduledEventID,
495
>
ScheduleAttempt: workflowTask.Attempt,
496
>
Version: workflowTask.Version,
497
>
Stamp: workflowTask.Stamp,
498
>
InMemory: isSpeculative,
499
>
}
500
>
501
>
if isSpeculative {
502
>
// If WT is still speculative, create task in in-memory task queue.
503
>
return r.mutableState.SetSpeculativeWorkflowTaskTimeoutTask(wttt)
504
>
}
505
506
// This function can be called for speculative WT which just was converted to normal