Fix rollback bug

FAB-15963 #done

Change-Id: I3bb168e8e678530ee809acc4fbfee103f228526c
Signed-off-by: manish <manish.sethi@gmail.com>
Signed-off-by: senthil <cendhu@gmail.com>
(cherry picked from commit 61199ed)

Author: manish-sethi

common/ledger/blkstorage/fsblkstorage/rollback.go

@@ -145,6 +145,14 @@ func populateBlockInfoWithDuplicateTxids(blockInfo *serializedBlockInfo, placeme
 
 	for _, txOffset := range blockInfo.txOffsets {
 		blockLoc, err := indexStore.getBlockLocByTxID(txOffset.txID)
+		// There is a situations where the txid entries for a config transaction may not present in the index. This is caused
+		// by the fact that in the data produced by a release proior to 1.4.2, the txID for a config transaction is used as
+		// an empty string. However, in the data produced by release 1.4.2 (and up), the real txID is used by computing in the
+		// indexing code. So, a mismatch is possible where the generated txID is not present in the index
+		if err == blkstorage.ErrNotFoundInIndex {
+			logger.Warnf("TxID [%s] not found in index... skipping this TxID", txOffset.txID)
+			continue
+		}
 		if err != nil {
 			return err
 		}

common/ledger/blkstorage/fsblkstorage/rollback_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/hyperledger/fabric/common/ledger/blkstorage"
 	"github.com/hyperledger/fabric/common/ledger/testutil"
+	"github.com/hyperledger/fabric/common/ledger/util/leveldbhelper"
 	"github.com/hyperledger/fabric/common/metrics/disabled"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/peer"
@@ -300,6 +301,48 @@ func TestDuplicateTxIDDuringRollback(t *testing.T) {
 	blkfileMgrWrapper.testGetTransactionByTxID("tx0", blocks[2].Data.Data[0], nil)
 }
 
+func TestRollbackTxIDMissingFromIndex(t *testing.T) {
+	// Polpulate block store with four blocks
+	path := testPath()
+	blocks := testutil.ConstructTestBlocks(t, 4)
+	testutil.SetTxID(t, blocks[3], 0, "blk3_tx0")
+
+	env := newTestEnv(t, NewConf(path, 0))
+	defer env.Cleanup()
+	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
+
+	blkfileMgrWrapper.addBlocks(blocks)
+	expectedBlockchainInfo := &common.BlockchainInfo{
+		Height:            4,
+		CurrentBlockHash:  protoutil.BlockHeaderHash(blocks[3].Header),
+		PreviousBlockHash: protoutil.BlockHeaderHash(blocks[2].Header),
+	}
+	actualBlockchainInfo := blkfileMgrWrapper.blockfileMgr.getBlockchainInfo()
+	assert.Equal(t, expectedBlockchainInfo, actualBlockchainInfo)
+	blkfileMgrWrapper.testGetTransactionByTxID("blk3_tx0", blocks[3].Data.Data[0], nil)
+
+	// Delete index entries for a tx
+	testutilDeleteTxFromIndex(t, blkfileMgrWrapper.blockfileMgr.db, 3, 0, "blk3_tx0")
+	env.provider.Close()
+	blkfileMgrWrapper.close()
+
+	// Rollback to block 2
+	indexConfig := &blkstorage.IndexConfig{AttrsToIndex: attrsToIndex}
+	err := Rollback(path, "testLedger", 2, indexConfig)
+	assert.NoError(t, err)
+
+	// Reopen blockstorage and assert basic functionality
+	env = newTestEnv(t, NewConf(path, 0))
+	blkfileMgrWrapper = newTestBlockfileWrapper(env, "testLedger")
+	expectedBlockchainInfo = &common.BlockchainInfo{
+		Height:            3,
+		CurrentBlockHash:  protoutil.BlockHeaderHash(blocks[2].Header),
+		PreviousBlockHash: protoutil.BlockHeaderHash(blocks[1].Header),
+	}
+	actualBlockchainInfo = blkfileMgrWrapper.blockfileMgr.getBlockchainInfo()
+	assert.Equal(t, expectedBlockchainInfo, actualBlockchainInfo)
+}
+
 func assertBlockStoreRollback(t *testing.T, path, ledgerID string, maxFileSize int, blocks []*common.Block,
 	rollbackedToBlkNum uint64, lastFileSuffixNum int, indexConfig *blkstorage.IndexConfig) {
 
@@ -350,3 +393,27 @@ func assertBlockStoreRollback(t *testing.T, path, ledgerID string, maxFileSize i
 	env.provider.Close()
 	blkfileMgrWrapper.close()
 }
+
+func testutilDeleteTxFromIndex(t *testing.T, db *leveldbhelper.DBHandle, blkNum, txNum uint64, txID string) {
+	indexKeys := [][]byte{
+		constructTxIDKey(txID),
+		constructBlockNumTranNumKey(blkNum, txNum),
+		constructBlockTxIDKey(txID),
+		constructTxValidationCodeIDKey(txID),
+	}
+
+	batch := leveldbhelper.NewUpdateBatch()
+	for _, key := range indexKeys {
+		value, err := db.Get(key)
+		assert.NoError(t, err)
+		assert.NotNil(t, value)
+		batch.Delete(key)
+	}
+	db.WriteBatch(batch, true)
+
+	for _, key := range indexKeys {
+		value, err := db.Get(key)
+		assert.NoError(t, err)
+		assert.Nil(t, value)
+	}
+}