[blockindex] introduce indexergroup

blockindex/indexer_group.go

@@ -0,0 +1,124 @@
+// Copyright (c) 2023 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package blockindex
+
+import (
+	"context"
+
+	"github.com/iotexproject/iotex-core/blockchain/block"
+	"github.com/iotexproject/iotex-core/blockchain/blockdao"
+)
+
+// IndexerGroup is a special index that includes multiple indexes,
+// which stay in sync when blocks are added.
+type IndexerGroup struct {
+	indexers       []blockdao.BlockIndexer
+	startHeights   []uint64
+	minStartHeight uint64
+}
+
+// NewIndexerGroup creates a new indexer group
+func NewIndexerGroup(indexers ...blockdao.BlockIndexer) *IndexerGroup {
+	return &IndexerGroup{indexers: indexers}
+}
+
+// Start starts the indexer group
+func (ig *IndexerGroup) Start(ctx context.Context) error {
+	for _, indexer := range ig.indexers {
+		if err := indexer.Start(ctx); err != nil {
+			return err
+		}
+	}
+	return ig.initStartHeight()
+}
+
+// Stop stops the indexer group
+func (ig *IndexerGroup) Stop(ctx context.Context) error {
+	for _, indexer := range ig.indexers {
+		if err := indexer.Stop(ctx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// PutBlock puts a block into the indexers in the group
+func (ig *IndexerGroup) PutBlock(ctx context.Context, blk *block.Block) error {
+	for i, indexer := range ig.indexers {
+		// check if the block is higher than the indexer's start height
+		if blk.Height() < ig.startHeights[i] {
+			continue
+		}
+		// check if the block is higher than the indexer's height
+		height, err := indexer.Height()
+		if err != nil {
+			return err
+		}
+		if blk.Height() <= height {
+			continue
+		}
+		// put block
+		if err := indexer.PutBlock(ctx, blk); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// DeleteTipBlock deletes the tip block from the indexers in the group
+func (ig *IndexerGroup) DeleteTipBlock(ctx context.Context, blk *block.Block) error {
+	for _, indexer := range ig.indexers {
+		if err := indexer.DeleteTipBlock(ctx, blk); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// StartHeight returns the minimum start height of the indexers in the group
+func (ig *IndexerGroup) StartHeight() uint64 {
+	return ig.minStartHeight
+}
+
+// Height returns the minimum height of the indexers in the group
+func (ig *IndexerGroup) Height() (uint64, error) {
+	var height uint64
+	for i, indexer := range ig.indexers {
+		h, err := indexer.Height()
+		if err != nil {
+			return 0, err
+		}
+		if i == 0 || h < height {
+			height = h
+		}
+	}
+	return height, nil
+}
+
+// initStartHeight initializes the start height of the indexers in the group
+// for every indexer, the start height is the maximum of tipheight+1 and startheight
+func (ig *IndexerGroup) initStartHeight() error {
+	ig.minStartHeight = 0
+	ig.startHeights = make([]uint64, len(ig.indexers))
+	for i, indexer := range ig.indexers {
+		tipHeight, err := indexer.Height()
+		if err != nil {
+			return err
+		}
+		indexStartHeight := tipHeight + 1
+		if indexerWithStart, ok := indexer.(blockdao.BlockIndexerWithStart); ok {
+			startHeight := indexerWithStart.StartHeight()
+			if startHeight > indexStartHeight {
+				indexStartHeight = startHeight
+			}
+		}
+		ig.startHeights[i] = indexStartHeight
+		if i == 0 || indexStartHeight < ig.minStartHeight {
+			ig.minStartHeight = indexStartHeight
+		}
+	}
+	return nil
+}

blockindex/indexer_group_test.go

@@ -0,0 +1,72 @@
+// Copyright (c) 2023 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package blockindex
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/iotexproject/iotex-core/blockchain/blockdao"
+	"github.com/iotexproject/iotex-core/test/mock/mock_blockdao"
+)
+
+func TestIndexerGroup_StartHeight(t *testing.T) {
+	require := require.New(t)
+	ctrl := gomock.NewController(t)
+
+	cases := []struct {
+		name     string
+		indexers [][2]uint64 // [startHeight, height]
+		expect   uint64
+	}{
+		{"no indexers", nil, 0},
+		{"one indexer without start height", [][2]uint64{{0, 100}}, 101},
+		{"one indexer with start height I", [][2]uint64{{100, 200}}, 201},
+		{"one indexer with start height II", [][2]uint64{{300, 200}}, 300},
+		{"two indexers with start height I", [][2]uint64{{100, 200}, {200, 300}}, 201},
+		{"two indexers with start height II", [][2]uint64{{100, 200}, {400, 300}}, 201},
+		{"two indexers with start height III", [][2]uint64{{100, 350}, {400, 300}}, 351},
+		{"two indexers one with start height I", [][2]uint64{{0, 1}, {150, 1}}, 2},
+		{"two indexers one with start height II", [][2]uint64{{0, 1}, {150, 200}}, 2},
+		{"two indexers one with start height III", [][2]uint64{{0, 200}, {250, 1}}, 201},
+		{"two indexers one with start height IV", [][2]uint64{{0, 200}, {150, 1}}, 150},
+		{"two indexers I", [][2]uint64{{0, 5}, {0, 1}}, 2},
+		{"two indexers II", [][2]uint64{{0, 5}, {0, 5}}, 6},
+		{"two indexers III", [][2]uint64{{0, 5}, {0, 6}}, 6},
+		{"multiple indexers I", [][2]uint64{{0, 5}, {0, 6}, {0, 7}}, 6},
+		{"multiple indexers II", [][2]uint64{{0, 5}, {10, 6}, {0, 7}}, 6},
+		{"multiple indexers III", [][2]uint64{{10, 5}, {0, 6}, {0, 7}}, 7},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			var indexers []blockdao.BlockIndexer
+			for _, indexer := range c.indexers {
+				if indexer[0] > 0 {
+					mockIndexerWithStart := mock_blockdao.NewMockBlockIndexerWithStart(ctrl)
+					mockIndexerWithStart.EXPECT().Start(gomock.Any()).Return(nil).Times(1)
+					mockIndexerWithStart.EXPECT().StartHeight().Return(indexer[0]).Times(1)
+					mockIndexerWithStart.EXPECT().Height().Return(indexer[1], nil).Times(1)
+					indexers = append(indexers, mockIndexerWithStart)
+				} else {
+					mockIndexer := mock_blockdao.NewMockBlockIndexer(ctrl)
+					mockIndexer.EXPECT().Start(gomock.Any()).Return(nil).Times(1)
+					mockIndexer.EXPECT().Height().Return(indexer[1], nil).Times(1)
+					indexers = append(indexers, mockIndexer)
+				}
+			}
+			ig := NewIndexerGroup(indexers...)
+			err := ig.Start(context.Background())
+			require.NoError(err)
+			height := ig.StartHeight()
+			require.Equal(c.expect, height)
+		})
+	}
+
+}

blockindex/sgd_indexer_test.go

@@ -9,14 +9,15 @@ import (
 
 	"github.com/iotexproject/go-pkgs/hash"
 	"github.com/iotexproject/iotex-address/address"
+	"github.com/stretchr/testify/require"
+
 	"github.com/iotexproject/iotex-core/action"
 	"github.com/iotexproject/iotex-core/blockchain/block"
 	"github.com/iotexproject/iotex-core/blockchain/genesis"
 	"github.com/iotexproject/iotex-core/db"
 	"github.com/iotexproject/iotex-core/state"
 	"github.com/iotexproject/iotex-core/test/identityset"
 	"github.com/iotexproject/iotex-core/testutil"
-	"github.com/stretchr/testify/require"
 )
 
 const (

chainservice/builder.go

@@ -257,19 +257,25 @@ func (builder *Builder) buildBlockDAO(forTest bool) error {
 	}
 
 	var indexers []blockdao.BlockIndexer
-	indexers = append(indexers, builder.cs.factory)
+	// indexers in synchronizedIndexers will be putblock one by one for every block
+	synchronizedIndexers := []blockdao.BlockIndexer{builder.cs.factory}
+	if builder.cs.contractStakingIndexer != nil {
+		synchronizedIndexers = append(synchronizedIndexers, builder.cs.contractStakingIndexer)
+	}
+	if builder.cs.sgdIndexer != nil {
+		synchronizedIndexers = append(synchronizedIndexers, builder.cs.sgdIndexer)
+	}
+	if len(synchronizedIndexers) > 1 {
+		indexers = append(indexers, blockindex.NewIndexerGroup(synchronizedIndexers...))
+	} else {
+		indexers = append(indexers, builder.cs.factory)
+	}
 	if !builder.cfg.Chain.EnableAsyncIndexWrite && builder.cs.indexer != nil {
 		indexers = append(indexers, builder.cs.indexer)
 	}
 	if builder.cs.bfIndexer != nil {
 		indexers = append(indexers, builder.cs.bfIndexer)
 	}
-	if builder.cs.sgdIndexer != nil {
-		indexers = append(indexers, builder.cs.sgdIndexer)
-	}
-	if builder.cs.contractStakingIndexer != nil {
-		indexers = append(indexers, builder.cs.contractStakingIndexer)
-	}
 	if forTest {
 		builder.cs.blockdao = blockdao.NewBlockDAOInMemForTest(indexers)
 	} else {

go.mod

@@ -63,6 +63,7 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.34.0
 	golang.org/x/exp v0.0.0-20230206171751-46f607a40771
 	golang.org/x/text v0.7.0
+	gonum.org/v1/gonum v0.6.0
 )
 
 require (

go.sum

@@ -1624,6 +1624,7 @@ golang.org/x/tools v0.2.0 h1:G6AHpWxTMGY1KyEYoAQ5WTtIekUUvDNjan3ugu60JvE=
 golang.org/x/xerrors v0.0.0-20190212162355-a5947ffaace3/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
 gonum.org/v1/gonum v0.0.0-20181121035319-3f7ecaa7e8ca/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
+gonum.org/v1/gonum v0.6.0 h1:DJy6UzXbahnGUf1ujUNkh/NEtK14qMo2nvlBPs4U5yw=
 gonum.org/v1/gonum v0.6.0/go.mod h1:9mxDZsDKxgMAuccQkewq682L+0eCu4dCN2yonUJTCLU=
 gonum.org/v1/netlib v0.0.0-20181029234149-ec6d1f5cefe6/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
 gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=