Convert ReplicationState/ExecutionState POGO to Proto (#622)

temporalio · Jul 28, 2020 · 17e3b67 · 17e3b67
1 parent 463fc8a
commit 17e3b67
Show file tree

Hide file tree

Showing 29 changed files with 1,599 additions and 856 deletions.
diff --git a/api/persistenceblobs/v1/message.pb.go b/api/persistenceblobs/v1/message.pb.go
diff --git a/common/persistence/cassandra/cassandraPersistence.go b/common/persistence/cassandra/cassandraPersistence.go
@@ -2806,7 +2806,7 @@ func (d *cassandraPersistence) RangeDeleteReplicationTaskFromDLQ(
 	return nil
 }
 
-func workflowExecutionFromRow(result map[string]interface{}) (*p.InternalWorkflowExecutionInfo, *p.ReplicationState, error) {
+func workflowExecutionFromRow(result map[string]interface{}) (*p.InternalWorkflowExecutionInfo, *persistenceblobs.ReplicationState, error) {
 	eiBytes, ok := result["execution"].([]byte)
 	if !ok {
 		return nil, nil, newPersistedTypeMismatchError("execution", "", eiBytes, result)
@@ -2834,7 +2834,7 @@ func workflowExecutionFromRow(result map[string]interface{}) (*p.InternalWorkflo
 
 	info := p.ProtoWorkflowExecutionToPartialInternalExecution(protoInfo, protoState, nextEventID)
 
-	var state *p.ReplicationState
+	var state *persistenceblobs.ReplicationState
 	if protoInfo.ReplicationData != nil {
 		protoReplVersions, err := ProtoReplicationVersionsFromResultMap(result)
 		if err != nil {

diff --git a/common/persistence/cassandra/cassandraPersistenceUtil.go b/common/persistence/cassandra/cassandraPersistenceUtil.go
@@ -425,7 +425,7 @@ func createExecution(
 	batch *gocql.Batch,
 	shardID int,
 	executionInfo *p.InternalWorkflowExecutionInfo,
-	replicationState *p.ReplicationState,
+	replicationState *persistenceblobs.ReplicationState,
 	versionHistories *serialization.DataBlob,
 	checksum checksum.Checksum,
 	cqlNowTimestampMillis int64,
@@ -506,7 +506,7 @@ func createExecution(
 			checksumDatablob.Data,
 			checksumDatablob.Encoding.String())
 	} else if replicationState != nil {
-		replicationVersions, err := serialization.ReplicationVersionsToBlob(replicationState.GenerateVersionProto())
+		replicationVersions, err := serialization.ReplicationVersionsToBlob(p.GenerateVersionProto(replicationState))
 		if err != nil {
 			return err
 		}
@@ -537,7 +537,7 @@ func updateExecution(
 	batch *gocql.Batch,
 	shardID int,
 	executionInfo *p.InternalWorkflowExecutionInfo,
-	replicationState *p.ReplicationState,
+	replicationState *persistenceblobs.ReplicationState,
 	versionHistories *serialization.DataBlob,
 	cqlNowTimestampMillis int64,
 	condition int64,
@@ -620,7 +620,7 @@ func updateExecution(
 			rowTypeExecutionTaskID,
 			condition)
 	} else if replicationState != nil {
-		replicationVersions, err := serialization.ReplicationVersionsToBlob(replicationState.GenerateVersionProto())
+		replicationVersions, err := serialization.ReplicationVersionsToBlob(p.GenerateVersionProto(replicationState))
 		if err != nil {
 			return err
 		}
@@ -1536,12 +1536,12 @@ func updateBufferedEvents(
 	}
 }
 
-func ReplicationStateFromProtos(wei *persistenceblobs.WorkflowExecutionInfo, rv *persistenceblobs.ReplicationVersions) *p.ReplicationState {
+func ReplicationStateFromProtos(wei *persistenceblobs.WorkflowExecutionInfo, rv *persistenceblobs.ReplicationVersions) *persistenceblobs.ReplicationState {
 	if rv == nil && wei.ReplicationData == nil {
 		return nil
 	}
 
-	info := &p.ReplicationState{}
+	info := &persistenceblobs.ReplicationState{}
 	info.CurrentVersion = wei.CurrentVersion
 
 	if rv != nil {
@@ -1551,7 +1551,7 @@ func ReplicationStateFromProtos(wei *persistenceblobs.WorkflowExecutionInfo, rv
 
 	if wei.ReplicationData != nil {
 		info.LastReplicationInfo = wei.ReplicationData.LastReplicationInfo
-		info.LastWriteEventID = wei.ReplicationData.LastWriteEventId
+		info.LastWriteEventId = wei.ReplicationData.LastWriteEventId
 	}
 
 	if info.LastReplicationInfo == nil {

diff --git a/common/persistence/dataInterfaces.go b/common/persistence/dataInterfaces.go
@@ -257,21 +257,6 @@ type (
 		CronSchedule string
 	}
 
-	// ExecutionStats is the statistics about workflow execution
-	ExecutionStats struct {
-		HistorySize int64
-	}
-
-	// ReplicationState represents mutable state information for global namespaces.
-	// This information is used by replication protocol when applying events from remote clusters
-	ReplicationState struct {
-		CurrentVersion      int64
-		StartVersion        int64
-		LastWriteVersion    int64
-		LastWriteEventID    int64
-		LastReplicationInfo map[string]*replicationspb.ReplicationInfo
-	}
-
 	// ReplicationTaskInfoWrapper describes a replication task.
 	ReplicationTaskInfoWrapper struct {
 		*persistenceblobs.ReplicationTaskInfo
@@ -490,8 +475,8 @@ type (
 		SignalInfos         map[int64]*persistenceblobs.SignalInfo
 		SignalRequestedIDs  map[string]struct{}
 		ExecutionInfo       *WorkflowExecutionInfo
-		ExecutionStats      *ExecutionStats
-		ReplicationState    *ReplicationState
+		ExecutionStats      *persistenceblobs.ExecutionStats
+		ReplicationState    *persistenceblobs.ReplicationState
 		BufferedEvents      []*historypb.HistoryEvent
 		VersionHistories    *VersionHistories
 		Checksum            checksum.Checksum
@@ -658,8 +643,8 @@ type (
 	// WorkflowMutation is used as generic workflow execution state mutation
 	WorkflowMutation struct {
 		ExecutionInfo    *WorkflowExecutionInfo
-		ExecutionStats   *ExecutionStats
-		ReplicationState *ReplicationState
+		ExecutionStats   *persistenceblobs.ExecutionStats
+		ReplicationState *persistenceblobs.ReplicationState
 		VersionHistories *VersionHistories
 
 		UpsertActivityInfos       []*persistenceblobs.ActivityInfo
@@ -688,8 +673,8 @@ type (
 	// WorkflowSnapshot is used as generic workflow execution state snapshot
 	WorkflowSnapshot struct {
 		ExecutionInfo    *WorkflowExecutionInfo
-		ExecutionStats   *ExecutionStats
-		ReplicationState *ReplicationState
+		ExecutionStats   *persistenceblobs.ExecutionStats
+		ReplicationState *persistenceblobs.ReplicationState
 		VersionHistories *VersionHistories
 
 		ActivityInfos       []*persistenceblobs.ActivityInfo
@@ -2168,7 +2153,7 @@ func NewGetReplicationTasksFromDLQRequest(
 	}
 }
 
-func (r *ReplicationState) GenerateVersionProto() *persistenceblobs.ReplicationVersions {
+func GenerateVersionProto(r *persistenceblobs.ReplicationState) *persistenceblobs.ReplicationVersions {
 	return &persistenceblobs.ReplicationVersions{
 		StartVersion:     &types.Int64Value{Value: r.StartVersion},
 		LastWriteVersion: &types.Int64Value{Value: r.LastWriteVersion},

diff --git a/common/persistence/executionStore.go b/common/persistence/executionStore.go
@@ -28,6 +28,7 @@ import (
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
 
+	"go.temporal.io/server/api/persistenceblobs/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/persistence/serialization"
@@ -109,7 +110,7 @@ func (m *executionManagerImpl) GetWorkflowExecution(
 
 func (m *executionManagerImpl) DeserializeExecutionInfo(
 	info *InternalWorkflowExecutionInfo,
-) (*WorkflowExecutionInfo, *ExecutionStats, error) {
+) (*WorkflowExecutionInfo, *persistenceblobs.ExecutionStats, error) {
 
 	completionEvent, err := m.serializer.DeserializeEvent(info.CompletionEvent)
 	if err != nil {
@@ -177,7 +178,7 @@ func (m *executionManagerImpl) DeserializeExecutionInfo(
 		SearchAttributes:                       info.SearchAttributes,
 		Memo:                                   info.Memo,
 	}
-	newStats := &ExecutionStats{
+	newStats := &persistenceblobs.ExecutionStats{
 		HistorySize: info.HistorySize,
 	}
 	return newInfo, newStats, nil
@@ -230,7 +231,7 @@ func (m *executionManagerImpl) UpdateWorkflowExecution(
 
 func (m *executionManagerImpl) SerializeExecutionInfo(
 	info *WorkflowExecutionInfo,
-	stats *ExecutionStats,
+	stats *persistenceblobs.ExecutionStats,
 	encoding common.EncodingType,
 ) (*InternalWorkflowExecutionInfo, error) {
 
@@ -691,7 +692,7 @@ func (m *executionManagerImpl) Close() {
 
 func getStartVersion(
 	versionHistories *VersionHistories,
-	replicationState *ReplicationState,
+	replicationState *persistenceblobs.ReplicationState,
 ) (int64, error) {
 
 	if replicationState == nil && versionHistories == nil {
@@ -715,7 +716,7 @@ func getStartVersion(
 
 func getLastWriteVersion(
 	versionHistories *VersionHistories,
-	replicationState *ReplicationState,
+	replicationState *persistenceblobs.ReplicationState,
 ) (int64, error) {
 
 	if replicationState == nil && versionHistories == nil {