Skip to content

Commit 09c8a81

Browse files
stephanosstephanos
committed
Workflow Task stamp
1 parent b8b622f commit 09c8a81

22 files changed

Lines changed: 485 additions & 92 deletions

‎api/persistence/v1/executions.pb.go‎

Lines changed: 27 additions & 15 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎common/persistence/serialization/task_serializer.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -473,6 +473,7 @@ func (s *TaskSerializer) transferWorkflowTaskToProto(
473473
Version: workflowTask.Version,
474474
TaskId: workflowTask.TaskID,
475475
VisibilityTime: timestamppb.New(workflowTask.VisibilityTimestamp),
476+
Stamp: workflowTask.Stamp,
476477
}
477478
}
478479

@@ -490,6 +491,7 @@ func (s *TaskSerializer) transferWorkflowTaskFromProto(
490491
TaskQueue: workflowTask.TaskQueue,
491492
ScheduledEventID: workflowTask.ScheduledEventId,
492493
Version: workflowTask.Version,
494+
Stamp: workflowTask.Stamp,
493495
}
494496
}
495497

@@ -736,6 +738,7 @@ func (s *TaskSerializer) timerWorkflowTaskToProto(
736738
EventId: workflowTimer.EventID,
737739
TaskId: workflowTimer.TaskID,
738740
VisibilityTime: timestamppb.New(workflowTimer.VisibilityTimestamp),
741+
Stamp: workflowTimer.Stamp,
739742
}
740743
}
741744

@@ -754,6 +757,7 @@ func (s *TaskSerializer) timerWorkflowTaskFromProto(
754757
ScheduleAttempt: workflowTimer.ScheduleAttempt,
755758
TimeoutType: workflowTimer.TimeoutType,
756759
Version: workflowTimer.Version,
760+
Stamp: workflowTimer.Stamp,
757761
}
758762
}
759763

‎common/serviceerror/obsolete_matching_task.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ type (
1313
// is no longer valid. Some examples:
1414
// - WFT belongs to a sticky queue but the workflow is not on that sticky queue anymore.
1515
// - WFT belongs to the normal queue but no the workflow is on a sticky queue.
16+
// - WFT options were updated and the task was rescheduled
1617
// - WF or activity task wants to start but their schedule time directive deployment no longer
1718
// matched the workflows effective deployment.
1819
// When Matching receives this error it can safely drop the task because History has already

‎proto/internal/temporal/server/api/persistence/v1/executions.proto‎

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,10 @@ message WorkflowExecutionInfo {
9999
// for a transient/speculative WFT.
100100
// Deprecated. Clean up with versioning-2. [cleanup-old-wv]
101101
int64 workflow_task_build_id_redirect_counter = 89;
102+
// Stamp represents the "version" of the workflow's internal state.
103+
// It increases monotonically when the workflow's options are modified.
104+
// It is used to check if a workflow task is still relevant to the corresponding workflow state machine.
105+
int32 workflow_task_stamp = 107;
102106

103107
bool cancel_requested = 29;
104108
string cancel_request_id = 32;
@@ -331,9 +335,9 @@ message TransferTaskInfo {
331335
// If the task addresses a CHASM component, this field will be set.
332336
ChasmTaskInfo chasm_task_info = 18;
333337
}
334-
// Stamp represents the version of the activity internal state, for which the timer task was created.
335-
// It monotonically increments when the activity options are changed.
336-
// It is used to check if activity related tasks are still relevant to their corresponding state machine.
338+
// Stamp represents the "version" of the entity's internal state for which the transfer task was created.
339+
// It increases monotonically when the entity's options are modified.
340+
// It is used to check if a task is still relevant to the entity's corresponding state machine.
337341
int32 stamp = 17;
338342
}
339343

@@ -414,9 +418,9 @@ message TimerTaskInfo {
414418
// A workflow chain is identified by the run_id of the first workflow in the chain.
415419
string first_run_id = 15;
416420

417-
// Stamp represents the version of the activity internal state, for which the timer task was created.
418-
// It monotonically increments when the activity options are changed.
419-
// It is used to check if activity related tasks are still relevant to their corresponding state machine.
421+
// Stamp represents the "version" of the entity's internal state for which the timer task was created.
422+
// It increases monotonically when the entity's options are modified.
423+
// It is used to check if a task is still relevant to the entity's corresponding state machine.
420424
int32 stamp = 16;
421425

422426
oneof task_details {
@@ -545,10 +549,9 @@ message ActivityInfo {
545549
// The last time an activity attempt completion was recorded by the server.
546550
google.protobuf.Timestamp last_attempt_complete_time = 40;
547551

548-
549-
// Stamp represents the “version” of the activity internal state (options) and can/will be changed with Activity API.
550-
// It monotonically increments when the activity options are changed.
551-
// It is used to check if activity related tasks are still relevant to their corresponding state machine.
552+
// Stamp represents the “version” of the activity's internal state and can/will be changed with Activity API.
553+
// It increases monotonically when the activity's options are modified.
554+
// It is used to check if an activity task is still relevant to the corresponding activity state machine.
552555
int32 stamp = 41;
553556

554557
// Paused state. When activity is paused it will not advance until unpaused.

‎service/history/api/recordworkflowtaskstarted/api.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,6 @@ func Invoke(
6262
),
6363
func(workflowLease api.WorkflowLease) (res *api.UpdateWorkflowAction, retErr error) {
6464
mutableState := workflowLease.GetMutableState()
65-
updateRegistry := workflowLease.GetContext().UpdateRegistry(ctx)
6665
if !mutableState.IsWorkflowExecutionRunning() {
6766
return nil, consts.ErrWorkflowCompleted
6867
}
@@ -74,6 +73,10 @@ func Invoke(
7473
// - Speculative WFT is lost (ScheduleToStart timeout for speculative WFT will recreate it).
7574
return nil, serviceerror.NewNotFound("Workflow task not found.")
7675
}
76+
if workflowTask.Stamp != mutableState.GetExecutionInfo().GetWorkflowTaskStamp() {
77+
// This happens when the workflow task was rescheduled.
78+
return nil, serviceerrors.NewObsoleteMatchingTask("Workflow task stamp mismatch")
79+
}
7780

7881
metricsScope := shardContext.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.HistoryRecordWorkflowTaskStartedScope))
7982

@@ -90,6 +93,7 @@ func Invoke(
9093

9194
workflowKey = mutableState.GetWorkflowKey()
9295
updateAction := &api.UpdateWorkflowAction{}
96+
updateRegistry := workflowLease.GetContext().UpdateRegistry(ctx)
9397

9498
if workflowTask.StartedEventID != common.EmptyEventID {
9599
// If workflow task is started as part of the current request scope then return a positive response

‎service/history/api/recordworkflowtaskstarted/api_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
package recordworkflowtaskstarted
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
"github.com/google/uuid"
8+
"github.com/stretchr/testify/require"
9+
commonpb "go.temporal.io/api/common/v1"
10+
enumspb "go.temporal.io/api/enums/v1"
11+
"go.temporal.io/api/serviceerror"
12+
taskqueuepb "go.temporal.io/api/taskqueue/v1"
13+
"go.temporal.io/api/workflowservice/v1"
14+
"go.temporal.io/server/api/historyservice/v1"
15+
persistencespb "go.temporal.io/server/api/persistence/v1"
16+
"go.temporal.io/server/common/cluster"
17+
"go.temporal.io/server/common/namespace"
18+
serviceerrors "go.temporal.io/server/common/serviceerror"
19+
"go.temporal.io/server/service/history/api"
20+
"go.temporal.io/server/service/history/configs"
21+
"go.temporal.io/server/service/history/consts"
22+
historyi "go.temporal.io/server/service/history/interfaces"
23+
"go.temporal.io/server/service/history/tests"
24+
"go.uber.org/mock/gomock"
25+
)
26+
27+
type mutableStateModifier func(*historyi.MockMutableState)
28+
29+
func TestRecordWorkflowTaskStarted_Errors(t *testing.T) {
30+
t.Run("Stamp Mismatch", func(t *testing.T) {
31+
resp, err := invoke(t, func(mutableState *historyi.MockMutableState) {
32+
executionInfo := &persistencespb.WorkflowExecutionInfo{WorkflowTaskStamp: 1}
33+
mutableState.EXPECT().GetExecutionInfo().Return(executionInfo).Times(1)
34+
})()
35+
36+
require.Nil(t, resp)
37+
require.Error(t, err)
38+
var obsoleteErr *serviceerrors.ObsoleteMatchingTask
39+
require.ErrorAs(t, err, &obsoleteErr)
40+
require.Contains(t, err.Error(), "Workflow task stamp mismatch")
41+
})
42+
43+
t.Run("Task Not Found", func(t *testing.T) {
44+
resp, err := invoke(t, func(mutableState *historyi.MockMutableState) {
45+
mutableState.EXPECT().GetWorkflowTaskByID(gomock.Any()).Return(nil).Times(1)
46+
})()
47+
48+
require.Nil(t, resp)
49+
require.Error(t, err)
50+
var notFoundErr *serviceerror.NotFound
51+
require.ErrorAs(t, err, &notFoundErr)
52+
require.Contains(t, err.Error(), "Workflow task not found")
53+
})
54+
55+
t.Run("Workflow Completed", func(t *testing.T) {
56+
resp, err := invoke(t, func(mutableState *historyi.MockMutableState) {
57+
mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).Times(1)
58+
})()
59+
60+
require.Nil(t, resp)
61+
require.Error(t, err)
62+
require.Equal(t, consts.ErrWorkflowCompleted, err)
63+
})
64+
}
65+
66+
func invoke(t *testing.T, modifyMutableState mutableStateModifier) func() (*historyservice.RecordWorkflowTaskStartedResponseWithRawHistory, error) {
67+
ctx := context.Background()
68+
ctrl := gomock.NewController(t)
69+
t.Cleanup(ctrl.Finish)
70+
71+
testNamespaceID := tests.NamespaceID
72+
workflowID := uuid.NewString()
73+
runID := uuid.NewString()
74+
scheduledEventID := int64(42)
75+
76+
request := &historyservice.RecordWorkflowTaskStartedRequest{
77+
NamespaceId: testNamespaceID.String(),
78+
WorkflowExecution: &commonpb.WorkflowExecution{
79+
WorkflowId: workflowID,
80+
RunId: runID,
81+
},
82+
ScheduledEventId: scheduledEventID,
83+
RequestId: "test-request",
84+
PollRequest: &workflowservice.PollWorkflowTaskQueueRequest{
85+
Identity: "test-worker",
86+
TaskQueue: &taskqueuepb.TaskQueue{
87+
Name: "test-tq",
88+
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
89+
},
90+
},
91+
}
92+
93+
mockNamespaceRegistry := namespace.NewMockRegistry(ctrl)
94+
mockNamespaceRegistry.EXPECT().GetNamespaceByID(testNamespaceID).Return(tests.GlobalNamespaceEntry, nil).AnyTimes()
95+
mockClusterMetadata := cluster.NewMockMetadata(ctrl)
96+
mockClusterMetadata.EXPECT().GetCurrentClusterName().Return("active").AnyTimes()
97+
shardContext := historyi.NewMockShardContext(ctrl)
98+
shardContext.EXPECT().GetNamespaceRegistry().Return(mockNamespaceRegistry).AnyTimes()
99+
shardContext.EXPECT().GetClusterMetadata().Return(mockClusterMetadata).AnyTimes()
100+
101+
mutableState := historyi.NewMockMutableState(ctrl)
102+
modifyMutableState(mutableState) // must come before setting defaults
103+
mutableState.EXPECT().GetWorkflowTaskByID(scheduledEventID).Return(&historyi.WorkflowTaskInfo{
104+
ScheduledEventID: scheduledEventID,
105+
Stamp: 0,
106+
}).AnyTimes()
107+
mutableState.EXPECT().IsWorkflowExecutionRunning().Return(true).AnyTimes()
108+
mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{}).AnyTimes()
109+
110+
mockWorkflowContext := historyi.NewMockWorkflowContext(ctrl)
111+
workflowLease := api.NewWorkflowLease(mockWorkflowContext, func(_ error) {}, mutableState)
112+
113+
consistencyChecker := api.NewMockWorkflowConsistencyChecker(ctrl)
114+
consistencyChecker.EXPECT().GetWorkflowLease(
115+
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
116+
).Return(workflowLease, nil)
117+
118+
config := &configs.Config{}
119+
120+
return func() (*historyservice.RecordWorkflowTaskStartedResponseWithRawHistory, error) {
121+
return Invoke(ctx, request, shardContext, config, nil, nil, consistencyChecker)
122+
}
123+
}

‎service/history/interfaces/workflow_task_info.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,11 @@ type WorkflowTaskInfo struct {
4848
// - when persisting transient/speculative WFT, the right Build ID is used in the WFT started event
4949
// Deprecated.
5050
BuildId string
51+
52+
// Stamp represents the "version" of the workflow's internal state.
53+
// It increases monotonically when the workflow's options are modified.
54+
// It is used to check if a workflow task is still relevant to the corresponding workflow state machine.
55+
Stamp int32
5156
}
5257

5358
type WorkflowTaskCompletionLimits struct {

‎service/history/tasks/workflow_task.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ type (
1818
TaskQueue string
1919
ScheduledEventID int64
2020
Version int64
21+
Stamp int32
2122
}
2223
)
2324

‎service/history/tasks/workflow_task_timer.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ type (
2929
ScheduleAttempt int32
3030
TimeoutType enumspb.TimeoutType
3131
Version int64
32+
Stamp int32
3233

3334
// InMemory field is not persisted in the database.
3435
InMemory bool

‎service/history/timer_queue_active_task_executor.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -374,6 +374,10 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTaskTimeoutTask(
374374
if workflowTask == nil {
375375
return nil
376376
}
377+
if task.Stamp != workflowTask.Stamp {
378+
release(nil) // release(nil) so that the mutable state is not unloaded from cache
379+
return consts.ErrStaleReference
380+
}
377381

378382
var operationMetricsTag string
379383
if workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {

0 commit comments

Comments
 (0)