Bugfix/picking changes to allow for failover polling

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

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=

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())
+}

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())
 }

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
 }

common/types/mapper/proto/history.go

@@ -21,6 +21,7 @@
 package proto
 
 import (
+	adminv1 "github.com/uber/cadence-idl/go/proto/admin/v1"
 	apiv1 "github.com/uber/cadence-idl/go/proto/api/v1"
 
 	historyv1 "github.com/uber/cadence/.gen/proto/history/v1"
@@ -292,23 +293,41 @@ func FromHistoryGetMutableStateRequest(t *types.GetMutableStateRequest) *history
 	if t == nil {
 		return nil
 	}
+
+	var versionHistoryItem *adminv1.VersionHistoryItem
+	if t.VersionHistoryItem != nil {
+		versionHistoryItem = &adminv1.VersionHistoryItem{
+			EventId: t.VersionHistoryItem.EventID,
+			Version: t.VersionHistoryItem.Version,
+		}
+	}
+
 	return &historyv1.GetMutableStateRequest{
 		DomainId:            t.DomainUUID,
 		WorkflowExecution:   FromWorkflowExecution(t.Execution),
 		ExpectedNextEventId: t.ExpectedNextEventID,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  versionHistoryItem,
 	}
 }
 
 func ToHistoryGetMutableStateRequest(t *historyv1.GetMutableStateRequest) *types.GetMutableStateRequest {
 	if t == nil {
 		return nil
 	}
+	var versionHistoryItem *types.VersionHistoryItem
+	if t.VersionHistoryItem != nil {
+		versionHistoryItem = &types.VersionHistoryItem{
+			EventID: t.VersionHistoryItem.EventId,
+			Version: t.VersionHistoryItem.Version,
+		}
+	}
 	return &types.GetMutableStateRequest{
 		DomainUUID:          t.DomainId,
 		Execution:           ToWorkflowExecution(t.WorkflowExecution),
 		ExpectedNextEventID: t.ExpectedNextEventId,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  versionHistoryItem,
 	}
 }
 

common/types/mapper/thrift/history.go

@@ -22,6 +22,7 @@ package thrift
 
 import (
 	"github.com/uber/cadence/.gen/go/history"
+	"github.com/uber/cadence/.gen/go/shared"
 	"github.com/uber/cadence/common/types"
 )
 
@@ -227,11 +228,21 @@ func FromHistoryGetMutableStateRequest(t *types.GetMutableStateRequest) *history
 	if t == nil {
 		return nil
 	}
+
+	var versionHistoryItem *shared.VersionHistoryItem
+	if t.VersionHistoryItem != nil {
+		versionHistoryItem = &shared.VersionHistoryItem{
+			EventID: &t.VersionHistoryItem.EventID,
+			Version: &t.VersionHistoryItem.Version,
+		}
+	}
+
 	return &history.GetMutableStateRequest{
 		DomainUUID:          &t.DomainUUID,
 		Execution:           FromWorkflowExecution(t.Execution),
 		ExpectedNextEventId: &t.ExpectedNextEventID,
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  versionHistoryItem,
 	}
 }
 
@@ -240,11 +251,24 @@ func ToHistoryGetMutableStateRequest(t *history.GetMutableStateRequest) *types.G
 	if t == nil {
 		return nil
 	}
+
+	var versionHistoryItem *types.VersionHistoryItem
+	if t.VersionHistoryItem != nil {
+		versionHistoryItem = &types.VersionHistoryItem{}
+		if t.VersionHistoryItem.EventID != nil {
+			versionHistoryItem.EventID = *t.VersionHistoryItem.EventID
+		}
+		if t.VersionHistoryItem.Version != nil {
+			versionHistoryItem.Version = *t.VersionHistoryItem.Version
+		}
+	}
+
 	return &types.GetMutableStateRequest{
 		DomainUUID:          t.GetDomainUUID(),
 		Execution:           ToWorkflowExecution(t.Execution),
 		ExpectedNextEventID: t.GetExpectedNextEventId(),
 		CurrentBranchToken:  t.CurrentBranchToken,
+		VersionHistoryItem:  versionHistoryItem,
 	}
 }
 

common/types/testdata/service_history.go

@@ -51,6 +51,7 @@ var (
 		Execution:           &WorkflowExecution,
 		ExpectedNextEventID: EventID1,
 		CurrentBranchToken:  BranchToken,
+		VersionHistoryItem:  &VersionHistoryItem,
 	}
 	HistoryGetMutableStateResponse = types.GetMutableStateResponse{
 		Execution:                            &WorkflowExecution,

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

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=

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 {

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
 		}