Refactor history branch manipulation logic into its own utility

common/metrics/metric_defs.go

@@ -480,12 +480,6 @@ const (
 	PersistenceAppendRawHistoryNodesScope = "AppendRawHistoryNodes"
 	// PersistenceDeleteHistoryNodesScope tracks DeleteHistoryNodes calls made by service to persistence layer
 	PersistenceDeleteHistoryNodesScope = "DeleteHistoryNodes"
-	// PersistenceParseHistoryBranchInfoScope tracks NewHistoryBranch calls made by service to persistence layer
-	PersistenceParseHistoryBranchInfoScope = "ParseHistoryBranchInfo"
-	// PersistenceUpdateHistoryBranchInfoScope tracks NewHistoryBranch calls made by service to persistence layer
-	PersistenceUpdateHistoryBranchInfoScope = "UpdateHistoryBranchInfo"
-	// PersistenceNewHistoryBranchScope tracks NewHistoryBranch calls made by service to persistence layer
-	PersistenceNewHistoryBranchScope = "NewHistoryBranch"
 	// PersistenceReadHistoryBranchScope tracks ReadHistoryBranch calls made by service to persistence layer
 	PersistenceReadHistoryBranchScope = "ReadHistoryBranch"
 	// PersistenceReadHistoryBranchReverseScope tracks ReadHistoryBranchReverse calls made by service to persistence layer

common/persistence/cassandra/history_store.go

@@ -30,7 +30,6 @@ import (
 
 	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/api/serviceerror"
-
 	"go.temporal.io/server/common/log"
 	p "go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/nosql/nosqlplugin/cassandra/gocql"
@@ -72,6 +71,7 @@ type (
 	HistoryStore struct {
 		Session gocql.Session
 		Logger  log.Logger
+		p.HistoryBranchUtilImpl
 	}
 )
 
@@ -162,56 +162,6 @@ func (h *HistoryStore) DeleteHistoryNodes(
 	return nil
 }
 
-// ParseHistoryBranchInfo parses the history branch for branch information
-func (h *HistoryStore) ParseHistoryBranchInfo(
-	ctx context.Context,
-	request *p.ParseHistoryBranchInfoRequest,
-) (*p.ParseHistoryBranchInfoResponse, error) {
-
-	branchInfo, err := p.ParseHistoryBranchToken(request.BranchToken)
-	if err != nil {
-		return nil, err
-	}
-	return &p.ParseHistoryBranchInfoResponse{
-		BranchInfo: branchInfo,
-	}, nil
-}
-
-// UpdateHistoryBranchInfo updates the history branch with branch information
-func (h *HistoryStore) UpdateHistoryBranchInfo(
-	ctx context.Context,
-	request *p.UpdateHistoryBranchInfoRequest,
-) (*p.UpdateHistoryBranchInfoResponse, error) {
-
-	branchToken, err := p.UpdateHistoryBranchToken(request.BranchToken, request.BranchInfo)
-	if err != nil {
-		return nil, err
-	}
-	return &p.UpdateHistoryBranchInfoResponse{
-		BranchToken: branchToken,
-	}, nil
-}
-
-// NewHistoryBranch initializes a new history branch
-func (h *HistoryStore) NewHistoryBranch(
-	ctx context.Context,
-	request *p.NewHistoryBranchRequest,
-) (*p.NewHistoryBranchResponse, error) {
-	var branchID string
-	if request.BranchID == nil {
-		branchID = primitives.NewUUID().String()
-	} else {
-		branchID = *request.BranchID
-	}
-	branchToken, err := p.NewHistoryBranchToken(request.TreeID, branchID, request.Ancestors)
-	if err != nil {
-		return nil, err
-	}
-	return &p.NewHistoryBranchResponse{
-		BranchToken: branchToken,
-	}, nil
-}
-
 // ReadHistoryBranch returns history node data for a branch
 // NOTE: For branch that has ancestors, we need to query Cassandra multiple times, because it doesn't support OR/UNION operator
 func (h *HistoryStore) ReadHistoryBranch(

common/persistence/client/fault_injection.go

@@ -71,6 +71,7 @@ type (
 	}
 
 	FaultInjectionExecutionStore struct {
+		persistence.HistoryBranchUtilImpl
 		baseExecutionStore persistence.ExecutionStore
 		ErrorGenerator     ErrorGenerator
 	}
@@ -650,36 +651,6 @@ func (e *FaultInjectionExecutionStore) DeleteHistoryNodes(
 	return e.baseExecutionStore.DeleteHistoryNodes(ctx, request)
 }
 
-func (e *FaultInjectionExecutionStore) ParseHistoryBranchInfo(
-	ctx context.Context,
-	request *persistence.ParseHistoryBranchInfoRequest,
-) (*persistence.ParseHistoryBranchInfoResponse, error) {
-	if err := e.ErrorGenerator.Generate(); err != nil {
-		return nil, err
-	}
-	return e.baseExecutionStore.ParseHistoryBranchInfo(ctx, request)
-}
-
-func (e *FaultInjectionExecutionStore) UpdateHistoryBranchInfo(
-	ctx context.Context,
-	request *persistence.UpdateHistoryBranchInfoRequest,
-) (*persistence.UpdateHistoryBranchInfoResponse, error) {
-	if err := e.ErrorGenerator.Generate(); err != nil {
-		return nil, err
-	}
-	return e.baseExecutionStore.UpdateHistoryBranchInfo(ctx, request)
-}
-
-func (e *FaultInjectionExecutionStore) NewHistoryBranch(
-	ctx context.Context,
-	request *persistence.NewHistoryBranchRequest,
-) (*persistence.NewHistoryBranchResponse, error) {
-	if err := e.ErrorGenerator.Generate(); err != nil {
-		return nil, err
-	}
-	return e.baseExecutionStore.NewHistoryBranch(ctx, request)
-}
-
 func (e *FaultInjectionExecutionStore) ReadHistoryBranch(
 	ctx context.Context,
 	request *persistence.InternalReadHistoryBranchRequest,

common/persistence/dataInterfaces.go

@@ -40,7 +40,6 @@ import (
 
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
-	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/service/history/tasks"
 )
 
@@ -731,47 +730,6 @@ type (
 		NodeID int64
 	}
 
-	ParseHistoryBranchInfoRequest struct {
-		// The branch token to parse the branch info from
-		BranchToken []byte
-	}
-
-	ParseHistoryBranchInfoResponse struct {
-		// The branch info parsed from the branch token
-		BranchInfo *persistencespb.HistoryBranch
-	}
-
-	UpdateHistoryBranchInfoRequest struct {
-		// The original branch token
-		BranchToken []byte
-		// The branch info to update with
-		BranchInfo *persistencespb.HistoryBranch
-	}
-
-	UpdateHistoryBranchInfoResponse struct {
-		// The newly updated branch token
-		BranchToken []byte
-	}
-
-	NewHistoryBranchRequest struct {
-		// The tree ID for the new branch token
-		TreeID string
-		// optional: can specify BranchID or allow random UUID to be generated
-		BranchID *string
-		// optional: can specify Ancestors to leave as empty
-		Ancestors []*persistencespb.HistoryBranchRange
-
-		// optional: supply optionally configured workflow settings as hints
-		RunTimeout        *time.Duration
-		ExecutionTimeout  *time.Duration
-		RetentionDuration *time.Duration
-	}
-
-	NewHistoryBranchResponse struct {
-		// The newly created branch token
-		BranchToken []byte
-	}
-
 	// ReadHistoryBranchRequest is used to read a history branch
 	ReadHistoryBranchRequest struct {
 		// The shard to get history branch data
@@ -1048,6 +1006,7 @@ type (
 	ExecutionManager interface {
 		Closeable
 		GetName() string
+		GetHistoryBranchUtil() HistoryBranchUtil
 
 		CreateWorkflowExecution(ctx context.Context, request *CreateWorkflowExecutionRequest) (*CreateWorkflowExecutionResponse, error)
 		UpdateWorkflowExecution(ctx context.Context, request *UpdateWorkflowExecutionRequest) (*UpdateWorkflowExecutionResponse, error)
@@ -1083,12 +1042,6 @@ type (
 		AppendHistoryNodes(ctx context.Context, request *AppendHistoryNodesRequest) (*AppendHistoryNodesResponse, error)
 		// AppendRawHistoryNodes add a node of raw histories to history node table
 		AppendRawHistoryNodes(ctx context.Context, request *AppendRawHistoryNodesRequest) (*AppendHistoryNodesResponse, error)
-		// ParseHistoryBranchInfo parses the history branch for branch information
-		ParseHistoryBranchInfo(ctx context.Context, request *ParseHistoryBranchInfoRequest) (*ParseHistoryBranchInfoResponse, error)
-		// UpdateHistoryBranchInfo updates the history branch with branch information
-		UpdateHistoryBranchInfo(ctx context.Context, request *UpdateHistoryBranchInfoRequest) (*UpdateHistoryBranchInfoResponse, error)
-		// NewHistoryBranch initializes a new history branch
-		NewHistoryBranch(ctx context.Context, request *NewHistoryBranchRequest) (*NewHistoryBranchResponse, error)
 		// ReadHistoryBranch returns history node data for a branch
 		ReadHistoryBranch(ctx context.Context, request *ReadHistoryBranchRequest) (*ReadHistoryBranchResponse, error)
 		// ReadHistoryBranchByBatch returns history node data for a branch ByBatch
@@ -1227,43 +1180,6 @@ func UnixMilliseconds(t time.Time) int64 {
 	return unixNano / int64(time.Millisecond)
 }
 
-func ParseHistoryBranchToken(branchToken []byte) (*persistencespb.HistoryBranch, error) {
-	// TODO: instead of always using the implementation from the serialization package, this should be injected
-	return serialization.HistoryBranchFromBlob(branchToken, enumspb.ENCODING_TYPE_PROTO3.String())
-}
-
-func UpdateHistoryBranchToken(branchToken []byte, branchInfo *persistencespb.HistoryBranch) ([]byte, error) {
-	bi, err := ParseHistoryBranchToken(branchToken)
-	if err != nil {
-		return nil, err
-	}
-	bi.TreeId = branchInfo.TreeId
-	bi.BranchId = branchInfo.BranchId
-	bi.Ancestors = branchInfo.Ancestors
-
-	// TODO: instead of always using the implementation from the serialization package, this should be injected
-	blob, err := serialization.HistoryBranchToBlob(bi)
-	if err != nil {
-		return nil, err
-	}
-	return blob.Data, nil
-}
-
-// NewHistoryBranchToken return a new branch token
-func NewHistoryBranchToken(treeID, branchID string, ancestors []*persistencespb.HistoryBranchRange) ([]byte, error) {
-	bi := &persistencespb.HistoryBranch{
-		TreeId:    treeID,
-		BranchId:  branchID,
-		Ancestors: ancestors,
-	}
-	datablob, err := serialization.HistoryBranchToBlob(bi)
-	if err != nil {
-		return nil, err
-	}
-	token := datablob.Data
-	return token, nil
-}
-
 // BuildHistoryGarbageCleanupInfo combine the workflow identity information into a string
 func BuildHistoryGarbageCleanupInfo(namespaceID, workflowID, runID string) string {
 	return fmt.Sprintf("%v:%v:%v", namespaceID, workflowID, runID)

common/persistence/execution_manager.go

@@ -79,6 +79,10 @@ func (m *executionManagerImpl) GetName() string {
 	return m.persistence.GetName()
 }
 
+func (m *executionManagerImpl) GetHistoryBranchUtil() HistoryBranchUtil {
+	return m.persistence.GetHistoryBranchUtil()
+}
+
 // The below three APIs are related to serialization/deserialization
 
 func (m *executionManagerImpl) CreateWorkflowExecution(