Merge branch 'master' into CDNC-8019

common/persistence/nosql/nosql_history_store_test.go

@@ -24,11 +24,13 @@ package nosql
 
 import (
 	ctx "context"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/log"
@@ -171,3 +173,155 @@ func TestAppendHistoryNodes_NewBranch(t *testing.T) {
 
 	assert.NoError(t, err)
 }
+
+const (
+	testMinNodeID         = 111
+	testMaxNodeID         = 222
+	testRequestLastNodeID = 333
+	testLastTransactionID = 444
+
+	// These needs to be greater than testRequestLastNodeID
+	testRowNodeID1 = int64(334)
+	testRowNodeID2 = int64(335)
+
+	// These needs to be greater than testLastTransactionID
+	testRowTxnID1 = int64(445)
+	testRowTxnID2 = int64(446)
+)
+
+func validInternalReadHistoryBranchRequest() *persistence.InternalReadHistoryBranchRequest {
+	return &persistence.InternalReadHistoryBranchRequest{
+		TreeID:            "TestTreeID",
+		BranchID:          "TestBranchID",
+		MinNodeID:         testMinNodeID,
+		MaxNodeID:         testMaxNodeID,
+		PageSize:          0,
+		NextPageToken:     nil,
+		LastNodeID:        testRequestLastNodeID,
+		LastTransactionID: testLastTransactionID,
+		ShardID:           testShardID,
+	}
+}
+
+func expectedHistoryNodeFilter() *nosqlplugin.HistoryNodeFilter {
+	return &nosqlplugin.HistoryNodeFilter{
+		ShardID:       testShardID,
+		TreeID:        "TestTreeID",
+		BranchID:      "TestBranchID",
+		MinNodeID:     testMinNodeID,
+		MaxNodeID:     testMaxNodeID,
+		NextPageToken: nil,
+		PageSize:      0,
+	}
+}
+
+func validHistoryNodeRows() []*nosqlplugin.HistoryNodeRow {
+	return []*nosqlplugin.HistoryNodeRow{
+		{
+			TreeID:       "TestTreeID",
+			BranchID:     "TestBranchID",
+			NodeID:       testRowNodeID1,
+			TxnID:        common.Ptr(testRowTxnID1),
+			Data:         []byte("TestEvents"),
+			DataEncoding: string(common.EncodingTypeThriftRW),
+			ShardID:      testShardID,
+		},
+		{
+			TreeID:       "TestTreeID",
+			BranchID:     "TestBranchID",
+			NodeID:       testRowNodeID2,
+			TxnID:        common.Ptr(testRowTxnID2),
+			Data:         []byte("TestEvents2"),
+			DataEncoding: string(common.EncodingTypeThriftRW),
+			ShardID:      testShardID,
+		},
+	}
+}
+
+func TestReadHistoryBranch(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := validInternalReadHistoryBranchRequest()
+	rows := validHistoryNodeRows()
+	// Append a rowID with a lower transaction ID to test that it is discarded
+	badRow := *rows[0]
+	badRow.TxnID = common.Ptr[int64](testLastTransactionID - 1)
+	rows = append(rows, &badRow)
+
+	// Expect to read the history branch
+	dbMock.EXPECT().SelectFromHistoryNode(gomock.Any(), expectedHistoryNodeFilter()).
+		Return(rows, nil, nil).Times(1)
+
+	resp, err := store.ReadHistoryBranch(ctx.Background(), request)
+	require.NoError(t, err)
+
+	// Asset that we got the history for all the nodes
+	assert.Equal(t, 2, len(resp.History))
+	assert.Equal(t, rows[0].Data, resp.History[0].Data)
+	assert.Equal(t, rows[1].Data, resp.History[1].Data)
+	assert.Equal(t, common.EncodingTypeThriftRW, resp.History[0].Encoding)
+	assert.Equal(t, common.EncodingTypeThriftRW, resp.History[1].Encoding)
+
+	assert.Nil(t, resp.NextPageToken)
+
+	// Assert that these ids corresponds to the last node and transaction id
+	assert.Equal(t, testRowNodeID2, resp.LastNodeID)
+	assert.Equal(t, testRowTxnID2, resp.LastTransactionID)
+}
+
+func TestReadHistoryBranch_ErrorIfSelectFromHistoryNodeErrors(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := validInternalReadHistoryBranchRequest()
+
+	testError := fmt.Errorf("test error")
+
+	dbMock.EXPECT().SelectFromHistoryNode(gomock.Any(), expectedHistoryNodeFilter()).
+		Return(nil, nil, testError).Times(1)
+	dbMock.EXPECT().IsNotFoundError(testError).Return(true).Times(1)
+
+	_, err := store.ReadHistoryBranch(ctx.Background(), request)
+
+	var notExistsErr *types.EntityNotExistsError
+	assert.ErrorAs(t, err, &notExistsErr)
+	assert.ErrorContains(t, err, "SelectFromHistoryNode")
+	assert.ErrorContains(t, err, "test error")
+}
+
+func TestReadHistoryBranch_ErrorIfDecreasingNodeID(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := validInternalReadHistoryBranchRequest()
+	rows := validHistoryNodeRows()
+	// Set the first row to have a node id that is less than the last node id in the request
+	rows[0].NodeID = 1
+
+	// Expect to read the history branch
+	dbMock.EXPECT().SelectFromHistoryNode(gomock.Any(), expectedHistoryNodeFilter()).
+		Return(rows, nil, nil).Times(1)
+
+	_, err := store.ReadHistoryBranch(ctx.Background(), request)
+
+	var dataError *types.InternalDataInconsistencyError
+	assert.ErrorAs(t, err, &dataError)
+	assert.ErrorContains(t, err, "corrupted data, nodeID cannot decrease")
+}
+
+func TestReadHistoryBranch_ErrorIfSameNodeID(t *testing.T) {
+	store, dbMock := setUpMocks(t)
+
+	request := validInternalReadHistoryBranchRequest()
+	rows := validHistoryNodeRows()
+	// Set the second row to have the same node id as the first row
+	rows[1].NodeID = rows[0].NodeID
+
+	// Expect to read the history branch
+	dbMock.EXPECT().SelectFromHistoryNode(gomock.Any(), expectedHistoryNodeFilter()).
+		Return(rows, nil, nil).Times(1)
+
+	_, err := store.ReadHistoryBranch(ctx.Background(), request)
+
+	var dataError *types.InternalDataInconsistencyError
+	assert.ErrorAs(t, err, &dataError)
+	assert.ErrorContains(t, err, "corrupted data, same nodeID must have smaller txnID")
+}