Add scanner invariants (uber#3257)

common/persistence/versionHistory.go

@@ -449,7 +449,7 @@ func (h *VersionHistories) GetVersionHistory(
 	branchIndex int,
 ) (*VersionHistory, error) {
 
-	if branchIndex < 0 || branchIndex > len(h.Histories) {
+	if branchIndex < 0 || branchIndex >= len(h.Histories) {
 		return nil, &shared.BadRequestError{Message: "invalid branch index."}
 	}
 

service/frontend/adminHandler_test.go

@@ -92,7 +92,7 @@ func (s *adminHandlerSuite) SetupTest() {
 		},
 	}
 	config := &Config{
-		EnableAdminProtection: dynamicconfig.GetBoolPropertyFn(false),
+		EnableAdminProtection:        dynamicconfig.GetBoolPropertyFn(false),
 		EnableCleanupReplicationTask: dynamicconfig.GetBoolPropertyFn(false),
 	}
 	s.handler = NewAdminHandler(s.mockResource, params, config)
@@ -553,8 +553,8 @@ func (s *adminHandlerSuite) Test_AddSearchAttribute_Permission() {
 	ctx := context.Background()
 	handler := s.handler
 	handler.config = &Config{
-		EnableAdminProtection: dynamicconfig.GetBoolPropertyFn(true),
-		AdminOperationToken:   dynamicconfig.GetStringPropertyFn(common.DefaultAdminOperationToken),
+		EnableAdminProtection:        dynamicconfig.GetBoolPropertyFn(true),
+		AdminOperationToken:          dynamicconfig.GetStringPropertyFn(common.DefaultAdminOperationToken),
 		EnableCleanupReplicationTask: dynamicconfig.GetBoolPropertyFn(false),
 	}
 

service/history/config/config.go

@@ -204,8 +204,8 @@ type Config struct {
 	ReplicationTaskProcessorCleanupInterval          dynamicconfig.DurationPropertyFnWithShardIDFilter
 	ReplicationTaskProcessorCleanupJitterCoefficient dynamicconfig.FloatPropertyFnWithShardIDFilter
 	// TODO: those two flags are for migration. Consider remove them after the migration complete
-	EnableRPCReplication   dynamicconfig.BoolPropertyFn
-	EnableKafkaReplication dynamicconfig.BoolPropertyFn
+	EnableRPCReplication         dynamicconfig.BoolPropertyFn
+	EnableKafkaReplication       dynamicconfig.BoolPropertyFn
 	EnableCleanupReplicationTask dynamicconfig.BoolPropertyFn
 
 	// The following are used by consistent query

service/worker/scanner/executions/common/blobstoreIterator.go

@@ -87,6 +87,9 @@ func getBlobstoreFetchPageFn(
 			if err := json.Unmarshal(p, &soe); err != nil {
 				return pagination.Page{}, err
 			}
+			if err := ValidateExecution(&soe.Execution); err != nil {
+				return pagination.Page{}, err
+			}
 			executions = append(executions, &soe.Execution)
 		}
 		var nextPageToken interface{} = index + 1

service/worker/scanner/executions/common/interfaces.go

@@ -49,7 +49,7 @@ type (
 	// It can also be used to fix the invariant for an execution.
 	Invariant interface {
 		Check(Execution, *InvariantResourceBag) CheckResult
-		Fix(Execution, *InvariantResourceBag) FixResult
+		Fix(Execution) FixResult
 		InvariantType() InvariantType
 	}
 

service/worker/scanner/executions/common/persistenceIterator.go

@@ -23,6 +23,7 @@
 package common
 
 import (
+	"github.com/uber/cadence/common/codec"
 	"github.com/uber/cadence/common/pagination"
 	"github.com/uber/cadence/common/persistence"
 )
@@ -40,15 +41,15 @@ func NewPersistenceIterator(
 	shardID int,
 ) ExecutionIterator {
 	return &persistenceIterator{
-		itr: pagination.NewIterator(nil, getPersistenceFetchPageFn(pr, pageSize, shardID)),
+		itr: pagination.NewIterator(nil, getPersistenceFetchPageFn(pr, codec.NewThriftRWEncoder(), pageSize, shardID)),
 	}
 }
 
 // Next returns the next execution
 func (i *persistenceIterator) Next() (*Execution, error) {
 	exec, err := i.itr.Next()
 	if exec != nil {
-		return exec.(*Execution), err
+		return exec.(*Execution), nil
 	}
 	return nil, err
 }
@@ -60,6 +61,7 @@ func (i *persistenceIterator) HasNext() bool {
 
 func getPersistenceFetchPageFn(
 	pr PersistenceRetryer,
+	encoder *codec.ThriftRWEncoder,
 	pageSize int,
 	shardID int,
 ) pagination.FetchFn {
@@ -76,22 +78,24 @@ func getPersistenceFetchPageFn(
 		}
 		executions := make([]pagination.Entity, len(resp.Executions), len(resp.Executions))
 		for i, e := range resp.Executions {
-			branchToken := e.ExecutionInfo.BranchToken
-			if e.VersionHistories != nil {
-				versionHistory, err := e.VersionHistories.GetCurrentVersionHistory()
-				if err != nil {
-					return pagination.Page{}, err
-				}
-				branchToken = versionHistory.GetBranchToken()
+			branchToken, treeID, branchID, err := GetBranchToken(e, encoder)
+			if err != nil {
+				return pagination.Page{}, err
 			}
-			executions[i] = &Execution{
+			exec := &Execution{
 				ShardID:     shardID,
 				DomainID:    e.ExecutionInfo.DomainID,
 				WorkflowID:  e.ExecutionInfo.WorkflowID,
 				RunID:       e.ExecutionInfo.RunID,
 				BranchToken: branchToken,
+				TreeID:      treeID,
+				BranchID:    branchID,
 				State:       e.ExecutionInfo.State,
 			}
+			if err := ValidateExecution(exec); err != nil {
+				return pagination.Page{}, err
+			}
+			executions[i] = exec
 		}
 		var nextToken interface{} = resp.PageToken
 		if len(resp.PageToken) == 0 {