Bugfix/picking changes to allow for failover polling (cadence-workflo…

…w#6523) With thanks to my colleague @Shaddoll for noticing, this ports across a Temporal fix temporalio/temporal#4943 to the cadence server repo. The problem is that for sync workflows, or workflows using the Execute API in go, the branch token itself is not sufficiently good enough information about the state of history polling. It can change, such as during a failover. This wraps that change in version-history so that it's able to handle this more gracefully. Why? To fix a problem for customers during failover
sankari165 · Nov 26, 2024 · f799577 · f799577
1 parent 33f755a
commit f799577
Show file tree

Hide file tree

Showing 27 changed files with 1,939 additions and 1,115 deletions.
diff --git a/.gen/go/history/history.go b/.gen/go/history/history.go
diff --git a/.gen/go/matching/matching.go b/.gen/go/matching/matching.go
diff --git a/.gen/go/shared/shared.go b/.gen/go/shared/shared.go
diff --git a/.gen/proto/history/v1/service.pb.go b/.gen/proto/history/v1/service.pb.go
diff --git a/.gen/proto/history/v1/service.pb.yarpc.go b/.gen/proto/history/v1/service.pb.yarpc.go
diff --git a/.gen/proto/matching/v1/service.pb.yarpc.go b/.gen/proto/matching/v1/service.pb.yarpc.go
diff --git a/cmd/server/go.mod b/cmd/server/go.mod
@@ -40,7 +40,7 @@ require (
 	github.com/startreedata/pinot-client-go v0.2.0 // latest release supports pinot v0.12.0 which is also internal version
 	github.com/stretchr/testify v1.8.3
 	github.com/uber-go/tally v3.3.15+incompatible // indirect
-	github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c
+	github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f
 	github.com/uber/ringpop-go v0.8.5 // indirect
 	github.com/uber/tchannel-go v1.22.2 // indirect
 	github.com/valyala/fastjson v1.4.1 // indirect

diff --git a/cmd/server/go.sum b/cmd/server/go.sum
@@ -407,8 +407,8 @@ github.com/uber-go/tally v3.3.12+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyu
 github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9PcvAkXO4nNMwg=
 github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
 github.com/uber/cadence-idl v0.0.0-20211111101836-d6b70b60eb8c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
-github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c h1:sagx8l5XOlJWlwwflrxsxlYXgsgyr1Jpe2eXl7q5Vic=
-github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
+github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f h1:U2nI6IKh80rrueDb2G3wuhCkCHYCsLp9EFBazeTs7Dk=
+github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
 github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
 github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

diff --git a/common/persistence/data_manager_interfaces_test.go b/common/persistence/data_manager_interfaces_test.go
@@ -547,3 +547,30 @@ func TestTaskListPartitionConfigToInternalType(t *testing.T) {
 		})
 	}
 }
+
+func TestVersionHistoryCopy(t *testing.T) {
+	a := VersionHistories{
+		CurrentVersionHistoryIndex: 1,
+		Histories: []*VersionHistory{
+			{
+				BranchToken: []byte("token"),
+				Items: []*VersionHistoryItem{
+					{
+						EventID: 1,
+						Version: 2,
+					},
+				},
+			},
+			{
+				BranchToken: []byte("token"),
+				Items: []*VersionHistoryItem{
+					{
+						EventID: 1,
+						Version: 2,
+					},
+				},
+			},
+		},
+	}
+	assert.Equal(t, &a, a.Duplicate())
+}
diff --git a/common/persistence/versionHistory.go b/common/persistence/versionHistory.go
@@ -65,7 +65,9 @@ func (item *VersionHistoryItem) Duplicate() *VersionHistoryItem {
 
 // ToInternalType return internal format of version history item
 func (item *VersionHistoryItem) ToInternalType() *types.VersionHistoryItem {
-
+	if item == nil {
+		return nil
+	}
 	return &types.VersionHistoryItem{
 		EventID: item.EventID,
 		Version: item.Version,
@@ -229,7 +231,6 @@ func (v *VersionHistory) AddOrUpdateItem(
 func (v *VersionHistory) ContainsItem(
 	item *VersionHistoryItem,
 ) bool {
-
 	prevEventID := common.FirstEventID - 1
 	for _, currentItem := range v.Items {
 		if item.Version == currentItem.Version {
@@ -301,11 +302,9 @@ func (v *VersionHistory) GetFirstItem() (*VersionHistoryItem, error) {
 
 // GetLastItem return the last version history item
 func (v *VersionHistory) GetLastItem() (*VersionHistoryItem, error) {
-
 	if len(v.Items) == 0 {
 		return nil, &types.BadRequestError{Message: "version history is empty"}
 	}
-
 	return v.Items[len(v.Items)-1].Duplicate(), nil
 }
 
@@ -406,6 +405,9 @@ func NewVersionHistoriesFromInternalType(
 
 // Duplicate duplicate VersionHistories
 func (h *VersionHistories) Duplicate() *VersionHistories {
+	if h == nil {
+		return nil
+	}
 
 	currentVersionHistoryIndex := h.CurrentVersionHistoryIndex
 	histories := []*VersionHistory{}
@@ -590,6 +592,5 @@ func (h *VersionHistories) GetCurrentVersionHistoryIndex() int {
 
 // GetCurrentVersionHistory get the current version history
 func (h *VersionHistories) GetCurrentVersionHistory() (*VersionHistory, error) {
-
 	return h.GetVersionHistory(h.GetCurrentVersionHistoryIndex())
 }
diff --git a/common/types/history.go b/common/types/history.go
@@ -100,10 +100,11 @@ func (v *FailoverMarkerToken) GetFailoverMarker() (o *FailoverMarkerAttributes)
 
 // GetMutableStateRequest is an internal type (TBD...)
 type GetMutableStateRequest struct {
-	DomainUUID          string             `json:"domainUUID,omitempty"`
-	Execution           *WorkflowExecution `json:"execution,omitempty"`
-	ExpectedNextEventID int64              `json:"expectedNextEventId,omitempty"`
-	CurrentBranchToken  []byte             `json:"currentBranchToken,omitempty"`
+	DomainUUID          string              `json:"domainUUID,omitempty"`
+	Execution           *WorkflowExecution  `json:"execution,omitempty"`
+	ExpectedNextEventID int64               `json:"expectedNextEventId,omitempty"`
+	CurrentBranchToken  []byte              `json:"currentBranchToken,omitempty"`
+	VersionHistoryItem  *VersionHistoryItem `json:"versionHistoryItem,omitempty"`
 }
 
 // GetDomainUUID is an internal getter (TBD...)
@@ -272,16 +273,24 @@ func (v *ParentExecutionInfo) GetExecution() (o *WorkflowExecution) {
 
 // PollMutableStateRequest is an internal type (TBD...)
 type PollMutableStateRequest struct {
-	DomainUUID          string             `json:"domainUUID,omitempty"`
-	Execution           *WorkflowExecution `json:"execution,omitempty"`
-	ExpectedNextEventID int64              `json:"expectedNextEventId,omitempty"`
-	CurrentBranchToken  []byte             `json:"currentBranchToken,omitempty"`
+	DomainUUID          string              `json:"domainUUID,omitempty"`
+	Execution           *WorkflowExecution  `json:"execution,omitempty"`
+	ExpectedNextEventID int64               `json:"expectedNextEventId,omitempty"`
+	CurrentBranchToken  []byte              `json:"currentBranchToken,omitempty"`
+	VersionHistoryItem  *VersionHistoryItem `json:"versionHistoryItem,omitempty"`
+}
+
+func (p *PollMutableStateRequest) GetVersionHistoryItem() *VersionHistoryItem {
+	if p.VersionHistoryItem == nil {
+		return nil
+	}
+	return p.VersionHistoryItem
 }
 
 // GetDomainUUID is an internal getter (TBD...)
-func (v *PollMutableStateRequest) GetDomainUUID() (o string) {
-	if v != nil {
-		return v.DomainUUID
+func (p *PollMutableStateRequest) GetDomainUUID() (o string) {
+	if p != nil {
+		return p.DomainUUID
 	}
 	return
 }

diff --git a/common/types/mapper/proto/history.go b/common/types/mapper/proto/history.go
@@ -297,6 +297,7 @@ func FromHistoryGetMutableStateRequest(t *types.GetMutableStateRequest) *history
 		WorkflowExecution:   FromWorkflowExecution(t.Execution),
 		ExpectedNextEventId: t.ExpectedNextEventID,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  FromVersionHistoryItem(t.VersionHistoryItem),
 	}
 }
 
@@ -309,6 +310,7 @@ func ToHistoryGetMutableStateRequest(t *historyv1.GetMutableStateRequest) *types
 		Execution:           ToWorkflowExecution(t.WorkflowExecution),
 		ExpectedNextEventID: t.ExpectedNextEventId,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  ToVersionHistoryItem(t.VersionHistoryItem),
 	}
 }
 

diff --git a/common/types/mapper/thrift/history.go b/common/types/mapper/thrift/history.go
@@ -232,6 +232,7 @@ func FromHistoryGetMutableStateRequest(t *types.GetMutableStateRequest) *history
 		Execution:           FromWorkflowExecution(t.Execution),
 		ExpectedNextEventId: &t.ExpectedNextEventID,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  FromVersionHistoryItem(t.VersionHistoryItem),
 	}
 }
 
@@ -245,6 +246,7 @@ func ToHistoryGetMutableStateRequest(t *history.GetMutableStateRequest) *types.G
 		Execution:           ToWorkflowExecution(t.Execution),
 		ExpectedNextEventID: t.GetExpectedNextEventId(),
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  ToVersionHistoryItem(t.VersionHistoryItem),
 	}
 }
 

diff --git a/common/types/testdata/service_history.go b/common/types/testdata/service_history.go
@@ -51,6 +51,7 @@ var (
 		Execution:           &WorkflowExecution,
 		ExpectedNextEventID: EventID1,
 		CurrentBranchToken:  BranchToken,
+		VersionHistoryItem:  &VersionHistoryItem,
 	}
 	HistoryGetMutableStateResponse = types.GetMutableStateResponse{
 		Execution:                            &WorkflowExecution,

diff --git a/go.mod b/go.mod
@@ -40,7 +40,7 @@ require (
 	github.com/startreedata/pinot-client-go v0.2.0 // latest release supports pinot v0.12.0 which is also internal version
 	github.com/stretchr/testify v1.8.3
 	github.com/uber-go/tally v3.3.15+incompatible
-	github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c
+	github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f
 	github.com/uber/ringpop-go v0.8.5
 	github.com/uber/tchannel-go v1.22.2
 	github.com/urfave/cli/v2 v2.27.4

diff --git a/go.sum b/go.sum
@@ -445,8 +445,8 @@ github.com/uber-go/tally v3.3.12+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyu
 github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9PcvAkXO4nNMwg=
 github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
 github.com/uber/cadence-idl v0.0.0-20211111101836-d6b70b60eb8c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
-github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c h1:sagx8l5XOlJWlwwflrxsxlYXgsgyr1Jpe2eXl7q5Vic=
-github.com/uber/cadence-idl v0.0.0-20241118185545-0ff09166fc7c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
+github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f h1:U2nI6IKh80rrueDb2G3wuhCkCHYCsLp9EFBazeTs7Dk=
+github.com/uber/cadence-idl v0.0.0-20241126065313-57bd6876d48f/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
 github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
 github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

diff --git a/idls b/idls
diff --git a/proto/internal/uber/cadence/history/v1/service.proto b/proto/internal/uber/cadence/history/v1/service.proto
@@ -339,6 +339,7 @@ message GetMutableStateRequest {
   api.v1.WorkflowExecution workflow_execution = 2;
   int64 expected_next_event_id = 3;
   bytes current_branch_token = 4;
+  admin.v1.VersionHistoryItem version_history_item = 5;
 }
 
 message GetMutableStateResponse {
@@ -360,6 +361,7 @@ message GetMutableStateResponse {
   shared.v1.VersionHistories version_histories = 16;
   bool is_sticky_task_list_enabled = 17;
   int64 history_size = 18;
+
 }
 
 message PollMutableStateRequest {

diff --git a/service/frontend/api/handler.go b/service/frontend/api/handler.go
@@ -87,13 +87,14 @@ type (
 	}
 
 	getHistoryContinuationToken struct {
-		RunID             string
-		FirstEventID      int64
-		NextEventID       int64
-		IsWorkflowRunning bool
-		PersistenceToken  []byte
-		TransientDecision *types.TransientDecisionInfo
-		BranchToken       []byte
+		RunID              string
+		FirstEventID       int64
+		NextEventID        int64
+		IsWorkflowRunning  bool
+		PersistenceToken   []byte
+		TransientDecision  *types.TransientDecisionInfo
+		BranchToken        []byte
+		VersionHistoryItem *types.VersionHistoryItem
 	}
 
 	domainGetter interface {
@@ -1866,30 +1867,46 @@ func (wh *WorkflowHandler) GetWorkflowExecutionHistory(
 	// 3. the last first event ID (the event ID of the last batch of events in the history)
 	// 4. the next event ID
 	// 5. whether the workflow is closed
-	// 6. error if any
+	// 6 The version history
+	// 7. error if any
 	queryHistory := func(
 		domainUUID string,
 		execution *types.WorkflowExecution,
 		expectedNextEventID int64,
 		currentBranchToken []byte,
-	) ([]byte, string, int64, int64, bool, error) {
+		versionHistoryItem *persistence.VersionHistoryItem,
+	) ([]byte, string, int64, int64, bool, *types.VersionHistoryItem, error) {
+
 		response, err := wh.GetHistoryClient().PollMutableState(ctx, &types.PollMutableStateRequest{
 			DomainUUID:          domainUUID,
 			Execution:           execution,
 			ExpectedNextEventID: expectedNextEventID,
 			CurrentBranchToken:  currentBranchToken,
+			VersionHistoryItem:  versionHistoryItem.ToInternalType(),
 		})
 
 		if err != nil {
-			return nil, "", 0, 0, false, err
+			return nil, "", 0, 0, false, nil, err
 		}
+
 		isWorkflowRunning := response.GetWorkflowCloseState() == persistence.WorkflowCloseStatusNone
+		currentVersionHistory, err := persistence.NewVersionHistoriesFromInternalType(response.VersionHistories).GetCurrentVersionHistory()
+		if err != nil {
+			wh.GetLogger().Error("Failed to get current version history", tag.Dynamic("version-histories", response.VersionHistories))
+			return nil, "", 0, 0, false, nil, fmt.Errorf("failed to get the current version from the response from history: %w", err)
+		}
+
+		lastVersionHistoryItem, err := currentVersionHistory.GetLastItem()
+		if err != nil {
+			return nil, "", 0, 0, false, nil, err
+		}
 
 		return response.CurrentBranchToken,
 			response.Execution.GetRunID(),
 			response.GetLastFirstEventID(),
 			response.GetNextEventID(),
 			isWorkflowRunning,
+			lastVersionHistoryItem.ToInternalType(),
 			nil
 	}
 
@@ -1931,8 +1948,10 @@ func (wh *WorkflowHandler) GetWorkflowExecutionHistory(
 			if !isCloseEventOnly {
 				queryNextEventID = token.NextEventID
 			}
-			token.BranchToken, _, lastFirstEventID, nextEventID, isWorkflowRunning, err =
-				queryHistory(domainID, execution, queryNextEventID, token.BranchToken)
+
+			vh := persistence.NewVersionHistoryItemFromInternalType(token.VersionHistoryItem)
+			token.BranchToken, _, lastFirstEventID, nextEventID, isWorkflowRunning, token.VersionHistoryItem, err =
+				queryHistory(domainID, execution, queryNextEventID, token.BranchToken, vh)
 			if err != nil {
 				return nil, err
 			}
@@ -1944,8 +1963,8 @@ func (wh *WorkflowHandler) GetWorkflowExecutionHistory(
 		if !isCloseEventOnly {
 			queryNextEventID = common.FirstEventID
 		}
-		token.BranchToken, runID, lastFirstEventID, nextEventID, isWorkflowRunning, err =
-			queryHistory(domainID, execution, queryNextEventID, nil)
+		token.BranchToken, runID, lastFirstEventID, nextEventID, isWorkflowRunning, token.VersionHistoryItem, err =
+			queryHistory(domainID, execution, queryNextEventID, nil, nil)
 		if err != nil {
 			return nil, err
 		}

diff --git a/service/frontend/api/handler_test.go b/service/frontend/api/handler_test.go
@@ -1819,6 +1819,20 @@ func (s *workflowHandlerSuite) TestRestartWorkflowExecution__Success() {
 		},
 		LastFirstEventID: 0,
 		NextEventID:      2,
+		VersionHistories: &types.VersionHistories{
+			CurrentVersionHistoryIndex: 0,
+			Histories: []*types.VersionHistory{
+				{
+					BranchToken: []byte("token"),
+					Items: []*types.VersionHistoryItem{
+						{
+							EventID: 1,
+							Version: 1,
+						},
+					},
+				},
+			},
+		},
 	}, nil).AnyTimes()
 	s.mockDomainCache.EXPECT().GetDomainID(gomock.Any()).Return(s.testDomainID, nil).AnyTimes()
 	s.mockVersionChecker.EXPECT().SupportsRawHistoryQuery(gomock.Any(), gomock.Any()).Return(nil).Times(1)

diff --git a/service/history/decision/handler_test.go b/service/history/decision/handler_test.go
@@ -648,7 +648,7 @@ func TestHandleDecisionTaskCompleted(t *testing.T) {
 				engine := engine.NewMockEngine(ctrl)
 				decisionHandler.shard.(*shard.MockContext).EXPECT().GetEngine().Return(engine).Times(3)
 				engine.EXPECT().NotifyNewHistoryEvent(events.NewNotification(constants.TestDomainID, &types.WorkflowExecution{WorkflowID: constants.TestWorkflowID, RunID: constants.TestRunID},
-					0, 5, 0, nil, 1, 0))
+					0, 5, 0, 1, 0, nil))
 				engine.EXPECT().NotifyNewTransferTasks(gomock.Any())
 				engine.EXPECT().NotifyNewTimerTasks(gomock.Any())
 				engine.EXPECT().NotifyNewReplicationTasks(gomock.Any())
@@ -777,7 +777,7 @@ func TestHandleDecisionTaskCompleted(t *testing.T) {
 				engine := engine.NewMockEngine(ctrl)
 				decisionHandler.shard.(*shard.MockContext).EXPECT().GetEngine().Return(engine).Times(3)
 				engine.EXPECT().NotifyNewHistoryEvent(events.NewNotification(constants.TestDomainID, &types.WorkflowExecution{WorkflowID: constants.TestWorkflowID, RunID: constants.TestRunID},
-					0, 1, 0, nil, 1, 0))
+					0, 1, 0, 1, 0, nil))
 				engine.EXPECT().NotifyNewTransferTasks(gomock.Any())
 				engine.EXPECT().NotifyNewTimerTasks(gomock.Any())
 				engine.EXPECT().NotifyNewReplicationTasks(gomock.Any())
@@ -1153,7 +1153,7 @@ func TestHandleDecisionTaskCompleted(t *testing.T) {
 				engine := engine.NewMockEngine(ctrl)
 				decisionHandler.shard.(*shard.MockContext).EXPECT().GetEngine().Return(engine).Times(3)
 				engine.EXPECT().NotifyNewHistoryEvent(events.NewNotification(constants.TestDomainID, &types.WorkflowExecution{WorkflowID: constants.TestWorkflowID, RunID: constants.TestRunID},
-					0, 3, 0, nil, 1, 0))
+					0, 3, 0, 1, 0, nil))
 				engine.EXPECT().NotifyNewTransferTasks(gomock.Any())
 				engine.EXPECT().NotifyNewTimerTasks(gomock.Any())
 				engine.EXPECT().NotifyNewReplicationTasks(gomock.Any())