Set proper max reset points

service/history/execution/mutable_state_builder.go

@@ -1982,11 +1982,8 @@ func (e *mutableStateBuilder) addBinaryCheckSumIfNotExists(
 		}
 	}
 
-	if len(currResetPoints) == maxResetPoints {
-		// If exceeding the max limit, do rotation by taking the oldest one out.
-		currResetPoints = currResetPoints[1:]
-		recentBinaryChecksums = recentBinaryChecksums[1:]
-	}
+	recentBinaryChecksums, currResetPoints = trimBinaryChecksums(recentBinaryChecksums, currResetPoints, maxResetPoints)
+
 	// Adding current version of the binary checksum.
 	recentBinaryChecksums = append(recentBinaryChecksums, binChecksum)
 

service/history/execution/mutable_state_decision_task_manager.go

@@ -26,7 +26,6 @@ package execution
 
 import (
 	"fmt"
-	"math"
 	"time"
 
 	"github.com/uber/cadence/common"
@@ -254,7 +253,12 @@ func (m *mutableStateDecisionTaskManagerImpl) ReplicateDecisionTaskCompletedEven
 	event *types.HistoryEvent,
 ) error {
 	m.beforeAddDecisionTaskCompletedEvent()
-	return m.afterAddDecisionTaskCompletedEvent(event, math.MaxInt32)
+	maxResetPoints := common.DefaultHistoryMaxAutoResetPoints // use default when it is not set in the config
+	if m.msb.GetDomainEntry() != nil && m.msb.GetDomainEntry().GetInfo() != nil && m.msb.config != nil {
+		domainName := m.msb.GetDomainEntry().GetInfo().Name
+		maxResetPoints = m.msb.config.MaxAutoResetPoints(domainName)
+	}
+	return m.afterAddDecisionTaskCompletedEvent(event, maxResetPoints)
 }
 
 func (m *mutableStateDecisionTaskManagerImpl) ReplicateDecisionTaskFailedEvent() error {

service/history/execution/mutable_state_decision_task_manager_test.go

@@ -0,0 +1,75 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package execution
+
+import (
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/types"
+	"github.com/uber/cadence/service/history/config"
+	"github.com/uber/cadence/service/history/constants"
+	"github.com/uber/cadence/service/history/shard"
+)
+
+func TestReplicateDecisionTaskCompletedEvent(t *testing.T) {
+	mockShard := shard.NewTestContext(
+		t,
+		gomock.NewController(t),
+		&persistence.ShardInfo{
+			ShardID:          0,
+			RangeID:          1,
+			TransferAckLevel: 0,
+		},
+		config.NewForTest(),
+	)
+	mockShard.GetConfig().MutableStateChecksumGenProbability = func(domain string) int { return 100 }
+	mockShard.GetConfig().MutableStateChecksumVerifyProbability = func(domain string) int { return 100 }
+	logger := mockShard.GetLogger()
+	mockShard.Resource.DomainCache.EXPECT().GetDomainID(constants.TestDomainName).Return(constants.TestDomainID, nil).AnyTimes()
+
+	m := &mutableStateDecisionTaskManagerImpl{
+		msb: newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry),
+	}
+	eventType := types.EventTypeActivityTaskCompleted
+	e := &types.HistoryEvent{
+		ID:        1,
+		EventType: &eventType,
+	}
+	err := m.ReplicateDecisionTaskCompletedEvent(e)
+	assert.NoError(t, err)
+
+	// test when domainEntry is missed
+	m.msb.domainEntry = nil
+	err = m.ReplicateDecisionTaskCompletedEvent(e)
+	assert.NoError(t, err)
+
+	// test when config is nil
+	m.msb = newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry)
+	m.msb.config = nil
+	err = m.ReplicateDecisionTaskCompletedEvent(e)
+	assert.NoError(t, err)
+}

service/history/execution/mutable_state_util.go

@@ -552,3 +552,16 @@ func GetChildExecutionDomainEntry(
 
 	return parentDomainEntry, nil
 }
+
+func trimBinaryChecksums(recentBinaryChecksums []string, currResetPoints []*types.ResetPointInfo, maxResetPoints int) ([]string, []*types.ResetPointInfo) {
+	numResetPoints := len(currResetPoints)
+	if numResetPoints >= maxResetPoints {
+		// If exceeding the max limit, do rotation by taking the oldest ones out.
+		// startIndex plus one here because it needs to make space for the new binary checksum for the current run
+		startIndex := numResetPoints - maxResetPoints + 1
+		currResetPoints = currResetPoints[startIndex:]
+		recentBinaryChecksums = recentBinaryChecksums[startIndex:]
+	}
+
+	return recentBinaryChecksums, currResetPoints
+}

service/history/execution/mutable_state_util_test.go

@@ -148,3 +148,76 @@ func TestFindAutoResetPoint(t *testing.T) {
 	})
 	assert.Equal(t, pt, pt5)
 }
+
+func TestTrimBinaryChecksums(t *testing.T) {
+	testCases := []struct {
+		name           string
+		maxResetPoints int
+		expected       []string
+	}{
+		{
+			name:           "not reach limit",
+			maxResetPoints: 6,
+			expected:       []string{"checksum1", "checksum2", "checksum3", "checksum4", "checksum5"},
+		},
+		{
+			name:           "reach at limit",
+			maxResetPoints: 5,
+			expected:       []string{"checksum2", "checksum3", "checksum4", "checksum5"},
+		},
+		{
+			name:           "exceeds limit",
+			maxResetPoints: 2,
+			expected:       []string{"checksum5"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			currResetPoints := []*types.ResetPointInfo{
+				{
+					BinaryChecksum: "checksum1",
+					RunID:          "run1",
+				},
+				{
+					BinaryChecksum: "checksum2",
+					RunID:          "run2",
+				},
+				{
+					BinaryChecksum: "checksum3",
+					RunID:          "run3",
+				},
+				{
+					BinaryChecksum: "checksum4",
+					RunID:          "run4",
+				},
+				{
+					BinaryChecksum: "checksum5",
+					RunID:          "run5",
+				},
+			}
+			var recentBinaryChecksums []string
+			for _, rp := range currResetPoints {
+				recentBinaryChecksums = append(recentBinaryChecksums, rp.GetBinaryChecksum())
+			}
+			recentBinaryChecksums, currResetPoints = trimBinaryChecksums(recentBinaryChecksums, currResetPoints, tc.maxResetPoints)
+			assert.Equal(t, tc.expected, recentBinaryChecksums)
+			assert.Equal(t, len(tc.expected), len(currResetPoints))
+		})
+	}
+
+	// test empty case
+	currResetPoints := make([]*types.ResetPointInfo, 0, 1)
+	var recentBinaryChecksums []string
+	for _, rp := range currResetPoints {
+		recentBinaryChecksums = append(recentBinaryChecksums, rp.GetBinaryChecksum())
+	}
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("The function panicked: %v", r)
+		}
+	}()
+	trimedBinaryChecksums, trimedResetPoints := trimBinaryChecksums(recentBinaryChecksums, currResetPoints, 2)
+	assert.Equal(t, recentBinaryChecksums, trimedBinaryChecksums)
+	assert.Equal(t, currResetPoints, trimedResetPoints)
+}