Fix deterministic check for searchable change version (#889)

This PR includes:
Fix non-deterministic error for searchable change version.
Fix first upsertSearchAttributes contains redundant attrs from following upsert calls.

Author: vancexu

internal/internal_event_handlers.go

@@ -317,7 +317,12 @@ func (wc *workflowEnvironmentImpl) updateWorkflowInfoWithSearchAttributes(attrib
 
 func mergeSearchAttributes(current, upsert *shared.SearchAttributes) *shared.SearchAttributes {
 	if current == nil || len(current.IndexedFields) == 0 {
-		return upsert
+		if upsert == nil || len(upsert.IndexedFields) == 0 {
+			return nil
+		}
+		current = &shared.SearchAttributes{
+			IndexedFields: make(map[string][]byte),
+		}
 	}
 
 	fields := current.IndexedFields
@@ -571,9 +576,6 @@ func validateVersion(changeID string, version, minSupported, maxSupported Versio
 func (wc *workflowEnvironmentImpl) GetVersion(changeID string, minSupported, maxSupported Version) Version {
 	if version, ok := wc.changeVersions[changeID]; ok {
 		validateVersion(changeID, version, minSupported, maxSupported)
-		if wc.isReplay {
-			wc.UpsertSearchAttributes(createSearchAttributesForChangeVersion(changeID, version, wc.changeVersions))
-		}
 		return version
 	}
 

internal/internal_event_handlers_test.go

@@ -194,13 +194,13 @@ func Test_MergeSearchAttributes(t *testing.T) {
 			name:     "currentIsNil",
 			current:  nil,
 			upsert:   &s.SearchAttributes{},
-			expected: &s.SearchAttributes{},
+			expected: nil,
 		},
 		{
 			name:     "currentIsEmpty",
 			current:  &s.SearchAttributes{IndexedFields: make(map[string][]byte)},
 			upsert:   &s.SearchAttributes{},
-			expected: &s.SearchAttributes{},
+			expected: nil,
 		},
 		{
 			name: "normalMerge",

internal/internal_task_handlers.go

@@ -1064,6 +1064,20 @@ func skipDeterministicCheckForEvent(e *s.HistoryEvent) bool {
 	return false
 }
 
+// special check for upsert change version event
+func skipDeterministicCheckForUpsertChangeVersion(events []*s.HistoryEvent, idx int) bool {
+	e := events[idx]
+	if e.GetEventType() == s.EventTypeMarkerRecorded &&
+		e.MarkerRecordedEventAttributes.GetMarkerName() == versionMarkerName &&
+		idx < len(events)-1 &&
+		events[idx+1].GetEventType() == s.EventTypeUpsertWorkflowSearchAttributes {
+		if _, ok := events[idx+1].UpsertWorkflowSearchAttributesEventAttributes.SearchAttributes.IndexedFields[CadenceChangeVersion]; ok {
+			return true
+		}
+	}
+	return false
+}
+
 func matchReplayWithHistory(replayDecisions []*s.Decision, historyEvents []*s.HistoryEvent) error {
 	di := 0
 	hi := 0
@@ -1074,6 +1088,10 @@ matchLoop:
 		var e *s.HistoryEvent
 		if hi < hSize {
 			e = historyEvents[hi]
+			if skipDeterministicCheckForUpsertChangeVersion(historyEvents, hi) {
+				hi += 2
+				continue matchLoop
+			}
 			if skipDeterministicCheckForEvent(e) {
 				hi++
 				continue matchLoop
@@ -1289,7 +1307,10 @@ func isDecisionMatchEvent(d *s.Decision, e *s.HistoryEvent, strictMode bool) boo
 		}
 		eventAttributes := e.UpsertWorkflowSearchAttributesEventAttributes
 		decisionAttributes := d.UpsertWorkflowSearchAttributesDecisionAttributes
-		return isSearchAttributesMatched(eventAttributes.SearchAttributes, decisionAttributes.SearchAttributes)
+		if strictMode && !isSearchAttributesMatched(eventAttributes.SearchAttributes, decisionAttributes.SearchAttributes) {
+			return false
+		}
+		return true
 	}
 
 	return false

internal/internal_task_handlers_test.go

@@ -1067,6 +1067,7 @@ func Test_NonDeterministicCheck(t *testing.T) {
 func Test_IsDecisionMatchEvent_UpsertWorkflowSearchAttributes(t *testing.T) {
 	diType := s.DecisionTypeUpsertWorkflowSearchAttributes
 	eType := s.EventTypeUpsertWorkflowSearchAttributes
+	strictMode := false
 
 	testCases := []struct {
 		name     string
@@ -1097,7 +1098,7 @@ func Test_IsDecisionMatchEvent_UpsertWorkflowSearchAttributes(t *testing.T) {
 				EventType: &eType,
 				UpsertWorkflowSearchAttributesEventAttributes: &s.UpsertWorkflowSearchAttributesEventAttributes{},
 			},
-			expected: false,
+			expected: true,
 		},
 		{
 			name: "attributes match",
@@ -1119,7 +1120,37 @@ func Test_IsDecisionMatchEvent_UpsertWorkflowSearchAttributes(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			require.Equal(t, testCase.expected, isDecisionMatchEvent(testCase.decision, testCase.event, false))
+			require.Equal(t, testCase.expected, isDecisionMatchEvent(testCase.decision, testCase.event, strictMode))
+		})
+	}
+
+	strictMode = true
+
+	testCases = []struct {
+		name     string
+		decision *s.Decision
+		event    *s.HistoryEvent
+		expected bool
+	}{
+		{
+			name: "attributes not match",
+			decision: &s.Decision{
+				DecisionType: &diType,
+				UpsertWorkflowSearchAttributesDecisionAttributes: &s.UpsertWorkflowSearchAttributesDecisionAttributes{
+					SearchAttributes: &s.SearchAttributes{},
+				},
+			},
+			event: &s.HistoryEvent{
+				EventType: &eType,
+				UpsertWorkflowSearchAttributesEventAttributes: &s.UpsertWorkflowSearchAttributesEventAttributes{},
+			},
+			expected: false,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			require.Equal(t, testCase.expected, isDecisionMatchEvent(testCase.decision, testCase.event, strictMode))
 		})
 	}
 }

internal/internal_worker.go

@@ -1036,6 +1036,7 @@ func (aw *aggregatedWorker) Start() error {
 var binaryChecksum string
 var binaryChecksumLock sync.Mutex
 
+// SetBinaryChecksum set binary checksum
 func SetBinaryChecksum(checksum string) {
 	binaryChecksumLock.Lock()
 	defer binaryChecksumLock.Unlock()