Skip to content

Add schedule_to_start_timeout to WorkflowTaskScheduledEvent #9257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saad1998 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add schedule_to_start_timeout to WorkflowTaskScheduledEvent #9257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 10 additions & 5 deletions service/history/historybuilder/event_factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,14 +102,19 @@ func (b *EventFactory) CreateWorkflowTaskScheduledEvent(
startToCloseTimeout *durationpb.Duration,
attempt int32,
scheduleTime time.Time,
scheduleToStartTimeout *durationpb.Duration,
) *historypb.HistoryEvent {
event := b.createHistoryEvent(enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, scheduleTime)
attrs := &historypb.WorkflowTaskScheduledEventAttributes{
TaskQueue: taskQueue,
StartToCloseTimeout: startToCloseTimeout,
Attempt: attempt,
}
if scheduleToStartTimeout != nil {
attrs.ScheduleToStartTimeout = scheduleToStartTimeout
}
event.Attributes = &historypb.HistoryEvent_WorkflowTaskScheduledEventAttributes{
WorkflowTaskScheduledEventAttributes: &historypb.WorkflowTaskScheduledEventAttributes{
TaskQueue: taskQueue,
StartToCloseTimeout: startToCloseTimeout,
Attempt: attempt,
},
WorkflowTaskScheduledEventAttributes: attrs,
}

return event
Expand Down
3 changes: 2 additions & 1 deletion service/history/historybuilder/history_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,8 +182,9 @@ func (b *HistoryBuilder) AddWorkflowTaskScheduledEvent(
startToCloseTimeout *durationpb.Duration,
attempt int32,
scheduleTime time.Time,
scheduleToStartTimeout *durationpb.Duration,
) *historypb.HistoryEvent {
event := b.EventFactory.CreateWorkflowTaskScheduledEvent(taskQueue, startToCloseTimeout, attempt, scheduleTime)
event := b.EventFactory.CreateWorkflowTaskScheduledEvent(taskQueue, startToCloseTimeout, attempt, scheduleTime, scheduleToStartTimeout)
event, _ = b.EventStore.add(event)
return event
}
Expand Down
2 changes: 1 addition & 1 deletion service/history/historybuilder/history_builder_categorization_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1294,7 +1294,7 @@ func (s *sutTestingAdapter) AddWorkflowTaskTimedOutEvent(_ ...eventConfig) *hist
}

func (s *sutTestingAdapter) AddWorkflowTaskScheduledEvent(_ ...eventConfig) *historypb.HistoryEvent {
return s.HistoryBuilder.AddWorkflowTaskScheduledEvent(nil, nil, 1, s.today)
return s.HistoryBuilder.AddWorkflowTaskScheduledEvent(nil, nil, 1, s.today, nil)
}

func (s *sutTestingAdapter) AddActivityTaskStartedEvent(optionalConfig ...eventConfig) *historypb.HistoryEvent {
Expand Down
1 change: 1 addition & 0 deletions service/history/historybuilder/history_builder_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -628,6 +628,7 @@ func (s *historyBuilderSuite) TestWorkflowTaskScheduled() {
durationpb.New(startToCloseTimeout),
attempt,
s.now,
nil,
)
s.Equal(event, s.flush())
s.Equal(&historypb.HistoryEvent{
Expand Down
38 changes: 33 additions & 5 deletions service/history/workflow/workflow_task_state_machine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,11 +281,13 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskScheduleToStartTimeoutEvent(
m.ms.RemoveSpeculativeWorkflowTaskTimeoutTask()

// Create corresponding WorkflowTaskScheduled event for speculative WT.
_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
m.ms.CurrentTaskQueue(),
durationpb.New(workflowTask.WorkflowTaskTimeout),
workflowTask.Attempt,
workflowTask.ScheduledTime.UTC(),
scheduleToStartTimeout,
)
workflowTask.ScheduledEventID = scheduledEvent.GetEventId()
}
Expand Down Expand Up @@ -361,11 +363,16 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskScheduledEventAsHeartbeat(
var scheduledEventID int64

if createWorkflowTaskScheduledEvent {
var scheduleToStartTimeout *durationpb.Duration
if taskQueue.GetKind() == enumspb.TASK_QUEUE_KIND_STICKY || workflowTaskType == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
_, scheduleToStartTimeout = m.ms.GetWorkflowTaskScheduleToStartTimeout()
}
scheduledEvent = m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
taskQueue,
startToCloseTimeout,
attempt,
scheduleTime,
scheduleToStartTimeout,
)
scheduledEventID = scheduledEvent.GetEventId()
} else {
Expand Down Expand Up @@ -528,6 +535,7 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskStartedEvent(
if !workflowTaskScheduledEventCreated &&
(workflowTask.ScheduledEventID != m.ms.GetNextEventID() || workflowTask.Version != m.ms.GetCurrentVersion()) {

_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
workflowTask.Attempt = 1
workflowTask.Type = enumsspb.WORKFLOW_TASK_TYPE_NORMAL
workflowTaskScheduledEventCreated = true
Expand All @@ -538,6 +546,7 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskStartedEvent(
durationpb.New(workflowTask.WorkflowTaskTimeout),
workflowTask.Attempt,
startTime,
scheduleToStartTimeout,
)
scheduledEventID = scheduledEvent.GetEventId()
}
Expand Down Expand Up @@ -618,13 +627,15 @@ func (m *workflowTaskStateMachine) processBuildIdRedirectInfo(
if m.ms.IsTransientWorkflowTask() && m.ms.GetExecutionInfo().GetWorkflowTaskBuildId() != buildId {
// we're retrying a workflow task and this attempt is on a different build ID, converting the transient wf task
// to a normal wf task by creating a scheduled event for it and setting its attempt to 1.
_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
m.ms.CurrentTaskQueue(),
durationpb.New(workflowTask.WorkflowTaskTimeout),
// Preserving the number of previous attempts. This value shows the number of attempts made on the last
// build ID + 1 (because it's being reset to 1 for the next build ID. See bellow.)
workflowTask.Attempt,
workflowTask.ScheduledTime,
scheduleToStartTimeout,
)
newWorkflowTask = m.getWorkflowTaskInfo()
newWorkflowTask.ScheduledEventID = scheduledEvent.GetEventId()
Expand Down Expand Up @@ -731,11 +742,16 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskCompletedEvent(

if !workflowTaskScheduledStartedEventsCreated {
// Create corresponding WorkflowTaskScheduled and WorkflowTaskStarted events for transient/speculative workflow tasks.
var scheduleToStartTimeout *durationpb.Duration
if workflowTask.TaskQueue.GetKind() == enumspb.TASK_QUEUE_KIND_STICKY || workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
_, scheduleToStartTimeout = m.ms.GetWorkflowTaskScheduleToStartTimeout()
}
scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
m.ms.CurrentTaskQueue(),
durationpb.New(workflowTask.WorkflowTaskTimeout),
workflowTask.Attempt,
workflowTask.ScheduledTime.UTC(),
scheduleToStartTimeout,
)

workflowTask.ScheduledEventID = scheduledEvent.GetEventId()
Expand Down Expand Up @@ -822,11 +838,13 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskFailedEvent(
m.ms.RemoveSpeculativeWorkflowTaskTimeoutTask()

// Create corresponding WorkflowTaskScheduled and WorkflowTaskStarted events for speculative WT.
_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
m.ms.CurrentTaskQueue(),
durationpb.New(workflowTask.WorkflowTaskTimeout),
workflowTask.Attempt,
workflowTask.ScheduledTime.UTC(),
scheduleToStartTimeout,
)
workflowTask.ScheduledEventID = scheduledEvent.GetEventId()
startedEvent := m.ms.hBuilder.AddWorkflowTaskStartedEvent(
Expand Down Expand Up @@ -894,11 +912,13 @@ func (m *workflowTaskStateMachine) AddWorkflowTaskTimedOutEvent(
m.ms.RemoveSpeculativeWorkflowTaskTimeoutTask()

// Create corresponding WorkflowTaskScheduled and WorkflowTaskStarted events for speculative WT.
_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
m.ms.CurrentTaskQueue(),
durationpb.New(workflowTask.WorkflowTaskTimeout),
workflowTask.Attempt,
workflowTask.ScheduledTime.UTC(),
scheduleToStartTimeout,
)
workflowTask.ScheduledEventID = scheduledEvent.GetEventId()
startedEvent := m.ms.hBuilder.AddWorkflowTaskStartedEvent(
Expand Down Expand Up @@ -1145,17 +1165,23 @@ func (m *workflowTaskStateMachine) GetTransientWorkflowTaskInfo(
) *historyspb.TransientWorkflowTaskInfo {

// Create scheduled and started events which are not written to the history yet.
attrs := &historypb.WorkflowTaskScheduledEventAttributes{
TaskQueue: m.ms.CurrentTaskQueue(),
StartToCloseTimeout: durationpb.New(workflowTask.WorkflowTaskTimeout),
Attempt: workflowTask.Attempt,
}
if workflowTask.TaskQueue.GetKind() == enumspb.TASK_QUEUE_KIND_STICKY || workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
if _, sst := m.ms.GetWorkflowTaskScheduleToStartTimeout(); sst != nil {
attrs.ScheduleToStartTimeout = sst
}
}
scheduledEvent := &historypb.HistoryEvent{
EventId: workflowTask.ScheduledEventID,
EventTime: timestamppb.New(workflowTask.ScheduledTime),
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED,
Version: m.ms.currentVersion,
Attributes: &historypb.HistoryEvent_WorkflowTaskScheduledEventAttributes{
WorkflowTaskScheduledEventAttributes: &historypb.WorkflowTaskScheduledEventAttributes{
TaskQueue: m.ms.CurrentTaskQueue(),
StartToCloseTimeout: durationpb.New(workflowTask.WorkflowTaskTimeout),
Attempt: workflowTask.Attempt,
},
WorkflowTaskScheduledEventAttributes: attrs,
},
}

Expand Down Expand Up @@ -1425,6 +1451,7 @@ func (m *workflowTaskStateMachine) convertSpeculativeWorkflowTaskToNormal() erro
m.ms.workflowTaskUpdated = true
}

_, scheduleToStartTimeout := m.ms.GetWorkflowTaskScheduleToStartTimeout()
m.ms.executionInfo.WorkflowTaskType = enumsspb.WORKFLOW_TASK_TYPE_NORMAL
metrics.SpeculativeWorkflowTaskCommits.With(m.metricsHandler).Record(1,
metrics.ReasonTag("close_transaction"))
Expand All @@ -1436,6 +1463,7 @@ func (m *workflowTaskStateMachine) convertSpeculativeWorkflowTaskToNormal() erro
durationpb.New(wt.WorkflowTaskTimeout),
wt.Attempt,
wt.ScheduledTime,
scheduleToStartTimeout,
)

if scheduledEvent.EventId != wt.ScheduledEventID {
Expand Down