Reduplicative ations with same hash shows in action list 1386 (#1390)

Author: coderbradlee

blockchain/indexbuilder.go

@@ -44,10 +44,16 @@ type IndexBuilder struct {
 	cancelChan   chan interface{}
 	timerFactory *prometheustimer.TimerFactory
 	dao          *blockDAO
+	reindex      bool
+}
+
+type actionDelta struct {
+	senderDelta    map[hash.Hash160]uint64
+	recipientDelta map[hash.Hash160]uint64
 }
 
 // NewIndexBuilder instantiates an index builder
-func NewIndexBuilder(chain Blockchain) (*IndexBuilder, error) {
+func NewIndexBuilder(chain Blockchain, reindex bool) (*IndexBuilder, error) {
 	bc, ok := chain.(*blockchain)
 	if !ok {
 		log.S().Panic("unexpected blockchain implementation")
@@ -67,6 +73,7 @@ func NewIndexBuilder(chain Blockchain) (*IndexBuilder, error) {
 		cancelChan:   make(chan interface{}),
 		timerFactory: timerFactory,
 		dao:          bc.dao,
+		reindex:      reindex,
 	}, nil
 }
 
@@ -147,14 +154,16 @@ func (ib *IndexBuilder) getStartHeightAndIndex() (startHeight, startIndex uint64
 	}
 	return
 }
-func (ib *IndexBuilder) commitBatchAndClear(tipIndex, tipHeight uint64, batch db.KVStoreBatch) error {
+func (ib *IndexBuilder) commitBatchAndClear(tipIndex, tipHeight uint64, batch db.KVStoreBatch, actDelta *actionDelta) error {
 	tipIndexBytes := byteutil.Uint64ToBytes(tipIndex)
 	batch.Put(blockActionBlockMappingNS, indexActionsTipIndexKey, tipIndexBytes, "failed to put tip index of actions")
 	tipHeightBytes := byteutil.Uint64ToBytes(tipHeight)
 	batch.Put(blockActionBlockMappingNS, indexActionsTipHeightKey, tipHeightBytes, "failed to put tip height")
 	if err := ib.store.Commit(batch); err != nil {
 		return err
 	}
+	actDelta.senderDelta = make(map[hash.Hash160]uint64)
+	actDelta.recipientDelta = make(map[hash.Hash160]uint64)
 	return nil
 }
 func (ib *IndexBuilder) initAndLoadActions() error {
@@ -166,12 +175,19 @@ func (ib *IndexBuilder) initAndLoadActions() error {
 	if err != nil {
 		return err
 	}
-	startHeight, startIndex, err := ib.getStartHeightAndIndex()
-	if err != nil {
-		return err
+	startHeight, startIndex := uint64(1), uint64(0)
+	if !ib.reindex {
+		startHeight, startIndex, err = ib.getStartHeightAndIndex()
+		if err != nil {
+			return err
+		}
 	}
 	zap.L().Info("Loading actions", zap.Uint64("startHeight", startHeight), zap.Uint64("startIndex", startIndex))
 	batch := db.NewBatch()
+	actDelta := &actionDelta{
+		senderDelta:    make(map[hash.Hash160]uint64),
+		recipientDelta: make(map[hash.Hash160]uint64),
+	}
 	i := startHeight
 	for ; i <= tipHeight; i++ {
 		hash, err := ib.dao.getBlockHash(i)
@@ -185,7 +201,7 @@ func (ib *IndexBuilder) initAndLoadActions() error {
 		blk := &block.Block{
 			Body: *body,
 		}
-		err = indexBlockHash(startIndex, hash, ib.store, blk, batch)
+		err = indexBlockHash(startIndex, hash, ib.store, blk, batch, actDelta)
 		if err != nil {
 			return err
 		}
@@ -198,7 +214,7 @@ func (ib *IndexBuilder) initAndLoadActions() error {
 		startIndex += uint64(len(blk.Actions))
 		// commit once every 10000 heights
 		if i%10000 == 0 || i == tipHeight {
-			if err := ib.commitBatchAndClear(startIndex, i, batch); err != nil {
+			if err := ib.commitBatchAndClear(startIndex, i, batch, actDelta); err != nil {
 				return err
 			}
 		}
@@ -226,7 +242,7 @@ func getNextHeight(store db.KVStore) (uint64, error) {
 	return nextHeight, nil
 }
 func indexBlock(store db.KVStore, blk *block.Block, batch db.KVStoreBatch) error {
-	hash := blk.HashBlock()
+	hashBlock := blk.HashBlock()
 	// get index that already builded
 	startIndex, err := getNextIndex(store)
 	if err != nil {
@@ -237,7 +253,11 @@ func indexBlock(store db.KVStore, blk *block.Block, batch db.KVStoreBatch) error
 	if err != nil {
 		return err
 	}
-	if err = indexBlockHash(startIndex, hash, store, blk, batch); err != nil {
+	actDelta := &actionDelta{
+		senderDelta:    make(map[hash.Hash160]uint64),
+		recipientDelta: make(map[hash.Hash160]uint64),
+	}
+	if err = indexBlockHash(startIndex, hashBlock, store, blk, batch, actDelta); err != nil {
 		return err
 	}
 	tipIndexBytes := byteutil.Uint64ToBytes(startIndex + uint64(len(blk.Actions)))
@@ -246,21 +266,20 @@ func indexBlock(store db.KVStore, blk *block.Block, batch db.KVStoreBatch) error
 	batch.Put(blockActionBlockMappingNS, indexActionsTipHeightKey, tipHeightBytes, "failed to put tip height")
 	return nil
 }
-func indexBlockHash(startActionsNum uint64, blkHash hash.Hash256, store db.KVStore, blk *block.Block, batch db.KVStoreBatch) error {
+func indexBlockHash(startActionsNum uint64, blkHash hash.Hash256, store db.KVStore, blk *block.Block, batch db.KVStoreBatch, actDelta *actionDelta) error {
 	for i, elp := range blk.Actions {
 		actHash := elp.Hash()
 		batch.Put(blockActionBlockMappingNS, actHash[hashOffset:], blkHash[:], "failed to put action hash %x", actHash)
 		indexActionsBytes := byteutil.Uint64ToBytes(startActionsNum + uint64(i))
 		batch.Put(blockActionBlockMappingNS, indexActionsBytes, actHash[:], "failed to put index of actions %x", actHash)
 	}
 
-	return putActions(store, blk, batch)
+	return putActions(store, blk, batch, actDelta)
 }
 
-func putActions(store db.KVStore, blk *block.Block, batch db.KVStoreBatch) error {
-	senderDelta := make(map[hash.Hash160]uint64)
-	recipientDelta := make(map[hash.Hash160]uint64)
-
+func putActions(store db.KVStore, blk *block.Block, batch db.KVStoreBatch, actDelta *actionDelta) error {
+	senderDelta := actDelta.senderDelta
+	recipientDelta := actDelta.recipientDelta
 	for _, selp := range blk.Actions {
 		actHash := selp.Hash()
 		callerAddrBytes := hash.BytesToHash160(selp.SrcPubkey().Hash())

chainservice/chainservice.go

@@ -140,7 +140,7 @@ func New(
 
 	var indexBuilder *blockchain.IndexBuilder
 	if _, ok := cfg.Plugins[config.GatewayPlugin]; ok && cfg.Chain.EnableAsyncIndexWrite {
-		if indexBuilder, err = blockchain.NewIndexBuilder(chain); err != nil {
+		if indexBuilder, err = blockchain.NewIndexBuilder(chain, cfg.Reindex); err != nil {
 			return nil, errors.Wrap(err, "failed to create index builder")
 		}
 		if err := chain.AddSubscriber(indexBuilder); err != nil {

config/config.go

@@ -35,6 +35,7 @@ func init() {
 	flag.StringVar(&_secretPath, "secret-path", "", "Secret path")
 	flag.StringVar(&_subChainPath, "sub-config-path", "", "Sub chain Config path")
 	flag.Var(&_plugins, "plugin", "Plugin of the node")
+	flag.BoolVar(&_reindex, "reindex", false, "reindex actions")
 }
 
 var (
@@ -44,6 +45,7 @@ var (
 	_secretPath   string
 	_subChainPath string
 	_plugins      strs
+	_reindex      bool
 )
 
 const (
@@ -166,6 +168,7 @@ var (
 			},
 		},
 		Genesis: genesis.Default,
+		Reindex: false,
 	}
 
 	// ErrInvalidCfg indicates the invalid config value
@@ -351,6 +354,7 @@ type (
 		Log        log.GlobalConfig            `yaml:"log"`
 		SubLogs    map[string]log.GlobalConfig `yaml:"subLogs"`
 		Genesis    genesis.Genesis             `yaml:"genesis"`
+		Reindex    bool                        `yaml:"reindex"`
 	}
 
 	// Validate is the interface of validating the config
@@ -379,7 +383,7 @@ func New(validates ...Validate) (Config, error) {
 	if err := yaml.Get(uconfig.Root).Populate(&cfg); err != nil {
 		return Config{}, errors.Wrap(err, "failed to unmarshal YAML config to struct")
 	}
-
+	cfg.Reindex = _reindex
 	// set network master key to private key
 	if cfg.Network.MasterKey == "" {
 		cfg.Network.MasterKey = cfg.Chain.ProducerPrivKey