[FAB-2214] 1st block in chain is block 0, not 1

Made the following changes to fsblkstorage provider:

* Block numbers start at 0, not 1.

* Blockchain height should be last block number + 1.

* A block's Block.Header.Number should be equal to
  the Blockchain height when AddBlock() is called.

* Updated ledger tests to use random temporary dir.

* Update core endorser tests to generate valid block
  numbers.

Change-Id: I0208cd6a4b9da462c29d06126fa651f4c0dcd2fc
Signed-off-by: Luis Sanchez <sanchezl@us.ibm.com>

common/ledger/blkstorage/fsblkstorage/block_stream_test.go

@@ -30,7 +30,7 @@ func TestBlockfileStream(t *testing.T) {
 }
 
 func testBlockfileStream(t *testing.T, numBlocks int) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	ledgerid := "testledger"
 	w := newTestBlockfileWrapper(env, ledgerid)
@@ -71,7 +71,7 @@ func TestBlockFileStreamUnexpectedEOF(t *testing.T) {
 }
 
 func testBlockFileStreamUnexpectedEOF(t *testing.T, numBlocks int, partialBlockBytes []byte) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	w := newTestBlockfileWrapper(env, "testLedger")
 	blockfileMgr := w.blockfileMgr
@@ -100,7 +100,7 @@ func TestBlockStream(t *testing.T) {
 }
 
 func testBlockStream(t *testing.T, numFiles int) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	w := newTestBlockfileWrapper(env, "testLedger")
 	defer w.close()

common/ledger/blkstorage/fsblkstorage/blockfile_mgr.go

@@ -117,7 +117,7 @@ func newBlockfileMgr(id string, conf *Conf, indexConfig *blkstorage.IndexConfig,
 		panic(fmt.Sprintf("Could not get block file info for current block file from db: %s", err))
 	}
 	if cpInfo == nil { //if no cpInfo stored in db initiate to zero
-		cpInfo = &checkpointInfo{latestFileChunkSuffixNum: 0, latestFileChunksize: 0}
+		cpInfo = &checkpointInfo{0, 0, true, 0}
 		err = mgr.saveCurrentInfo(cpInfo, true)
 		if err != nil {
 			panic(fmt.Sprintf("Could not save next block file info to db: %s", err))
@@ -159,7 +159,7 @@ func newBlockfileMgr(id string, conf *Conf, indexConfig *blkstorage.IndexConfig,
 		PreviousBlockHash: nil}
 
 	//If start up is a restart of an existing storage, update BlockchainInfo for external API's
-	if cpInfo.lastBlockNumber > 0 {
+	if !cpInfo.isChainEmpty {
 		lastBlockHeader, err := mgr.retrieveBlockHeaderByNumber(cpInfo.lastBlockNumber)
 		if err != nil {
 			panic(fmt.Sprintf("Could not retrieve header of the last block form file: %s", err))
@@ -167,7 +167,7 @@ func newBlockfileMgr(id string, conf *Conf, indexConfig *blkstorage.IndexConfig,
 		lastBlockHash := lastBlockHeader.Hash()
 		previousBlockHash := lastBlockHeader.PreviousHash
 		bcInfo = &common.BlockchainInfo{
-			Height:            cpInfo.lastBlockNumber,
+			Height:            cpInfo.lastBlockNumber + 1,
 			CurrentBlockHash:  lastBlockHash,
 			PreviousBlockHash: previousBlockHash}
 	}
@@ -202,9 +202,17 @@ func syncCPInfoFromFS(rootDir string, cpInfo *checkpointInfo) {
 	if err != nil {
 		panic(fmt.Sprintf("Could not open current file for detecting last block in the file: %s", err))
 	}
-	//Updates the checkpoint info for the actual last block number stored and it's end location
-	cpInfo.lastBlockNumber += uint64(numBlocks)
 	cpInfo.latestFileChunksize = int(endOffsetLastBlock)
+	if numBlocks == 0 {
+		return
+	}
+	//Updates the checkpoint info for the actual last block number stored and it's end location
+	if cpInfo.isChainEmpty {
+		cpInfo.lastBlockNumber = uint64(numBlocks - 1)
+	} else {
+		cpInfo.lastBlockNumber += uint64(numBlocks)
+	}
+	cpInfo.isChainEmpty = false
 	logger.Debugf("Checkpoint after updates by scanning the last file segment:%s", cpInfo)
 }
 
@@ -242,6 +250,9 @@ func (mgr *blockfileMgr) moveToNextFile() {
 }
 
 func (mgr *blockfileMgr) addBlock(block *common.Block) error {
+	if block.Header.Number != mgr.getBlockchainInfo().Height {
+		return fmt.Errorf("Block number should have been %d but was %d", mgr.getBlockchainInfo().Height, block.Header.Number)
+	}
 	blockBytes, info, err := serializeBlock(block)
 	if err != nil {
 		return fmt.Errorf("Error while serializing block: %s", err)
@@ -282,6 +293,7 @@ func (mgr *blockfileMgr) addBlock(block *common.Block) error {
 	newCPInfo := &checkpointInfo{
 		latestFileChunkSuffixNum: currentCPInfo.latestFileChunkSuffixNum,
 		latestFileChunksize:      currentCPInfo.latestFileChunksize + totalBytesToAppend,
+		isChainEmpty:             false,
 		lastBlockNumber:          block.Header.Number}
 	//save the checkpoint information in the database
 	if err = mgr.saveCurrentInfo(newCPInfo, false); err != nil {
@@ -312,20 +324,24 @@ func (mgr *blockfileMgr) addBlock(block *common.Block) error {
 
 func (mgr *blockfileMgr) syncIndex() error {
 	var lastBlockIndexed uint64
+	var indexEmpty bool
 	var err error
 	//from the database, get the last block that was indexed
 	if lastBlockIndexed, err = mgr.index.getLastBlockIndexed(); err != nil {
-		return err
+		if err != errIndexEmpty {
+			return err
+		}
+		indexEmpty = true
 	}
-	//initialize index to file number:zero, offset:zero and block:1
+	//initialize index to file number:zero, offset:zero and blockNum:0
 	startFileNum := 0
 	startOffset := 0
-	blockNum := uint64(1)
+	blockNum := uint64(0)
 	skipFirstBlock := false
 	//get the last file that blocks were added to using the checkpoint info
 	endFileNum := mgr.cpInfo.latestFileChunkSuffixNum
 	//if the index stored in the db has value, update the index information with those values
-	if lastBlockIndexed != 0 {
+	if !indexEmpty {
 		var flp *fileLocPointer
 		if flp, err = mgr.index.getBlockLocByBlockNum(lastBlockIndexed); err != nil {
 			return err
@@ -428,7 +444,7 @@ func (mgr *blockfileMgr) retrieveBlockByNumber(blockNum uint64) (*common.Block,
 
 	// interpret math.MaxUint64 as a request for last block
 	if blockNum == math.MaxUint64 {
-		blockNum = mgr.getBlockchainInfo().Height
+		blockNum = mgr.getBlockchainInfo().Height - 1
 	}
 
 	loc, err := mgr.index.getBlockLocByBlockNum(blockNum)
@@ -597,6 +613,7 @@ func scanForLastCompleteBlock(rootDir string, fileNum int, startingOffset int64)
 type checkpointInfo struct {
 	latestFileChunkSuffixNum int
 	latestFileChunksize      int
+	isChainEmpty             bool
 	lastBlockNumber          uint64
 }
 
@@ -612,12 +629,20 @@ func (i *checkpointInfo) marshal() ([]byte, error) {
 	if err = buffer.EncodeVarint(i.lastBlockNumber); err != nil {
 		return nil, err
 	}
+	var chainEmptyMarker uint64
+	if i.isChainEmpty {
+		chainEmptyMarker = 1
+	}
+	if err = buffer.EncodeVarint(chainEmptyMarker); err != nil {
+		return nil, err
+	}
 	return buffer.Bytes(), nil
 }
 
 func (i *checkpointInfo) unmarshal(b []byte) error {
 	buffer := proto.NewBuffer(b)
 	var val uint64
+	var chainEmptyMarker uint64
 	var err error
 
 	if val, err = buffer.DecodeVarint(); err != nil {
@@ -634,11 +659,14 @@ func (i *checkpointInfo) unmarshal(b []byte) error {
 		return err
 	}
 	i.lastBlockNumber = val
-
+	if chainEmptyMarker, err = buffer.DecodeVarint(); err != nil {
+		return err
+	}
+	i.isChainEmpty = chainEmptyMarker == 1
 	return nil
 }
 
 func (i *checkpointInfo) String() string {
-	return fmt.Sprintf("latestFileChunkSuffixNum=[%d], latestFileChunksize=[%d], lastBlockNumber=[%d]",
-		i.latestFileChunkSuffixNum, i.latestFileChunksize, i.lastBlockNumber)
+	return fmt.Sprintf("latestFileChunkSuffixNum=[%d], latestFileChunksize=[%d], isChainEmpty=[%t], lastBlockNumber=[%d]",
+		i.latestFileChunkSuffixNum, i.latestFileChunksize, i.isChainEmpty, i.lastBlockNumber)
 }

common/ledger/blkstorage/fsblkstorage/blockfile_mgr_test.go

@@ -28,14 +28,14 @@ import (
 )
 
 func TestBlockfileMgrBlockReadWrite(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
 	blocks := testutil.ConstructTestBlocks(t, 10)
 	blkfileMgrWrapper.addBlocks(blocks)
 	blkfileMgrWrapper.testGetBlockByHash(blocks)
-	blkfileMgrWrapper.testGetBlockByNumber(blocks, 1)
+	blkfileMgrWrapper.testGetBlockByNumber(blocks, 0)
 }
 
 func TestBlockfileMgrCrashDuringWriting(t *testing.T) {
@@ -47,7 +47,7 @@ func TestBlockfileMgrCrashDuringWriting(t *testing.T) {
 
 func testBlockfileMgrCrashDuringWriting(t *testing.T, numBlocksBeforeCheckpoint int,
 	numBlocksAfterCheckpoint int, numLastBlockBytes int, numPartialBytesToWrite int) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	ledgerid := "testLedger"
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
@@ -58,6 +58,7 @@ func testBlockfileMgrCrashDuringWriting(t *testing.T, numBlocksBeforeCheckpoint
 	cpInfo1 := &checkpointInfo{
 		currentCPInfo.latestFileChunkSuffixNum,
 		currentCPInfo.latestFileChunksize,
+		currentCPInfo.isChainEmpty,
 		currentCPInfo.lastBlockNumber}
 
 	blocksAfterCP := bg.NextTestBlocks(numBlocksAfterCheckpoint)
@@ -88,17 +89,17 @@ func testBlockfileMgrCrashDuringWriting(t *testing.T, numBlocksBeforeCheckpoint
 	allBlocks = append(allBlocks, blocksBeforeCP...)
 	allBlocks = append(allBlocks, blocksAfterCP...)
 	allBlocks = append(allBlocks, blocksAfterRestart...)
-	testBlockfileMgrBlockIterator(t, blkfileMgrWrapper.blockfileMgr, 1, len(allBlocks), allBlocks)
+	testBlockfileMgrBlockIterator(t, blkfileMgrWrapper.blockfileMgr, 0, len(allBlocks)-1, allBlocks)
 }
 
 func TestBlockfileMgrBlockIterator(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
 	blocks := testutil.ConstructTestBlocks(t, 10)
 	blkfileMgrWrapper.addBlocks(blocks)
-	testBlockfileMgrBlockIterator(t, blkfileMgrWrapper.blockfileMgr, 1, 8, blocks[0:8])
+	testBlockfileMgrBlockIterator(t, blkfileMgrWrapper.blockfileMgr, 0, 7, blocks[0:8])
 }
 
 func testBlockfileMgrBlockIterator(t *testing.T, blockfileMgr *blockfileMgr,
@@ -120,7 +121,7 @@ func testBlockfileMgrBlockIterator(t *testing.T, blockfileMgr *blockfileMgr,
 }
 
 func TestBlockfileMgrBlockchainInfo(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
@@ -135,7 +136,7 @@ func TestBlockfileMgrBlockchainInfo(t *testing.T) {
 }
 
 func TestBlockfileMgrGetTxById(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
@@ -156,7 +157,7 @@ func TestBlockfileMgrGetTxById(t *testing.T) {
 }
 
 func TestBlockfileMgrGetTxByBlockNumTranNum(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
@@ -165,7 +166,7 @@ func TestBlockfileMgrGetTxByBlockNumTranNum(t *testing.T) {
 	for blockIndex, blk := range blocks {
 		for tranIndex, txEnvelopeBytes := range blk.Data.Data {
 			// blockNum starts with 1, tranNum starts with 1
-			txEnvelopeFromFileMgr, err := blkfileMgrWrapper.blockfileMgr.retrieveTransactionByBlockNumTranNum(uint64(blockIndex+1), uint64(tranIndex+1))
+			txEnvelopeFromFileMgr, err := blkfileMgrWrapper.blockfileMgr.retrieveTransactionByBlockNumTranNum(uint64(blockIndex), uint64(tranIndex+1))
 			testutil.AssertNoError(t, err, "Error while retrieving tx from blkfileMgr")
 			txEnvelope, err := putil.GetEnvelopeFromBlock(txEnvelopeBytes)
 			testutil.AssertNoError(t, err, "Error while unmarshalling tx")
@@ -175,24 +176,27 @@ func TestBlockfileMgrGetTxByBlockNumTranNum(t *testing.T) {
 }
 
 func TestBlockfileMgrRestart(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	ledgerid := "testLedger"
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
 	blocks := testutil.ConstructTestBlocks(t, 10)
 	blkfileMgrWrapper.addBlocks(blocks)
+	expectedHeight := uint64(10)
+	testutil.AssertEquals(t, blkfileMgrWrapper.blockfileMgr.getBlockchainInfo().Height, expectedHeight)
 	blkfileMgrWrapper.close()
 
 	blkfileMgrWrapper = newTestBlockfileWrapper(env, ledgerid)
 	defer blkfileMgrWrapper.close()
-	testutil.AssertEquals(t, int(blkfileMgrWrapper.blockfileMgr.cpInfo.lastBlockNumber), 10)
+	testutil.AssertEquals(t, int(blkfileMgrWrapper.blockfileMgr.cpInfo.lastBlockNumber), 9)
 	blkfileMgrWrapper.testGetBlockByHash(blocks)
+	testutil.AssertEquals(t, blkfileMgrWrapper.blockfileMgr.getBlockchainInfo().Height, expectedHeight)
 }
 
 func TestBlockfileMgrFileRolling(t *testing.T) {
-	blocks := testutil.ConstructTestBlocks(t, 100)
+	blocks := testutil.ConstructTestBlocks(t, 200)
 	size := 0
-	for _, block := range blocks {
+	for _, block := range blocks[:100] {
 		by, _, err := serializeBlock(block)
 		testutil.AssertNoError(t, err, "Error while serializing block")
 		blockBytesSize := len(by)
@@ -201,31 +205,31 @@ func TestBlockfileMgrFileRolling(t *testing.T) {
 	}
 
 	maxFileSie := int(0.75 * float64(size))
-	env := newTestEnv(t, NewConf(testPath, maxFileSie))
+	env := newTestEnv(t, NewConf(testPath(), maxFileSie))
 	defer env.Cleanup()
 	ledgerid := "testLedger"
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
-	blkfileMgrWrapper.addBlocks(blocks)
+	blkfileMgrWrapper.addBlocks(blocks[:100])
 	testutil.AssertEquals(t, blkfileMgrWrapper.blockfileMgr.cpInfo.latestFileChunkSuffixNum, 1)
-	blkfileMgrWrapper.testGetBlockByHash(blocks)
+	blkfileMgrWrapper.testGetBlockByHash(blocks[:100])
 	blkfileMgrWrapper.close()
 
 	blkfileMgrWrapper = newTestBlockfileWrapper(env, ledgerid)
 	defer blkfileMgrWrapper.close()
-	blkfileMgrWrapper.addBlocks(blocks)
+	blkfileMgrWrapper.addBlocks(blocks[100:])
 	testutil.AssertEquals(t, blkfileMgrWrapper.blockfileMgr.cpInfo.latestFileChunkSuffixNum, 2)
-	blkfileMgrWrapper.testGetBlockByHash(blocks)
+	blkfileMgrWrapper.testGetBlockByHash(blocks[100:])
 }
 
 func TestBlockfileMgrGetBlockByTxID(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
 	defer blkfileMgrWrapper.close()
 	blocks := testutil.ConstructTestBlocks(t, 10)
 	blkfileMgrWrapper.addBlocks(blocks)
 	for _, blk := range blocks {
-		for j, _ := range blk.Data.Data {
+		for j := range blk.Data.Data {
 			// blockNum starts with 1
 			txID, err := extractTxID(blk.Data.Data[j])
 			testutil.AssertNoError(t, err, "")

common/ledger/blkstorage/fsblkstorage/blockfile_scan_test.go

@@ -27,7 +27,7 @@ import (
 )
 
 func TestBlockFileScanSmallTxOnly(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	ledgerid := "testLedger"
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
@@ -50,7 +50,7 @@ func TestBlockFileScanSmallTxOnly(t *testing.T) {
 }
 
 func TestBlockFileScanSmallTxLastTxIncomplete(t *testing.T) {
-	env := newTestEnv(t, NewConf(testPath, 0))
+	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()
 	ledgerid := "testLedger"
 	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)

common/ledger/blkstorage/fsblkstorage/blockindex.go

@@ -18,6 +18,7 @@ package fsblkstorage
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 
 	"github.com/golang/protobuf/proto"
@@ -36,6 +37,7 @@ const (
 )
 
 var indexCheckpointKey = []byte(indexCheckpointKeyStr)
+var errIndexEmpty = errors.New("NoBlockIndexed")
 
 type index interface {
 	getLastBlockIndexed() (uint64, error)
@@ -75,6 +77,9 @@ func (index *blockIndex) getLastBlockIndexed() (uint64, error) {
 	if blockNumBytes, err = index.db.Get(indexCheckpointKey); err != nil {
 		return 0, nil
 	}
+	if blockNumBytes == nil {
+		return 0, errIndexEmpty
+	}
 	return decodeBlockNum(blockNumBytes), nil
 }
 

common/ledger/blkstorage/fsblkstorage/blockindex_test.go

@@ -59,53 +59,56 @@ func TestBlockIndexSync(t *testing.T) {
 }
 
 func testBlockIndexSync(t *testing.T, numBlocks int, numBlocksToIndex int, syncByRestart bool) {
-	env := newTestEnv(t, NewConf(testPath, 0))
-	defer env.Cleanup()
-	ledgerid := "testledger"
-	blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
-	defer blkfileMgrWrapper.close()
-	blkfileMgr := blkfileMgrWrapper.blockfileMgr
-	origIndex := blkfileMgr.index
-	// construct blocks for testing
-	blocks := testutil.ConstructTestBlocks(t, numBlocks)
-	// add a few blocks
-	blkfileMgrWrapper.addBlocks(blocks[:numBlocksToIndex])
-
-	// Plug-in a noop index and add remaining blocks
-	blkfileMgr.index = &noopIndex{}
-	blkfileMgrWrapper.addBlocks(blocks[numBlocksToIndex:])
-
-	// Plug-in back the original index
-	blkfileMgr.index = origIndex
-	// The first set of blocks should be present in the orginal index
-	for i := 1; i <= numBlocksToIndex; i++ {
-		block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
-		testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
-		testutil.AssertEquals(t, block, blocks[i-1])
-	}
-
-	// The last set of blocks should not be present in the original index
-	for i := numBlocksToIndex + 1; i <= numBlocks; i++ {
-		_, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
-		testutil.AssertSame(t, err, blkstorage.ErrNotFoundInIndex)
-	}
-
-	// perform index sync
-	if syncByRestart {
-		blkfileMgrWrapper.close()
-		blkfileMgrWrapper = newTestBlockfileWrapper(env, ledgerid)
+	testName := fmt.Sprintf("%v/%v/%v", numBlocks, numBlocksToIndex, syncByRestart)
+	t.Run(testName, func(t *testing.T) {
+		env := newTestEnv(t, NewConf(testPath(), 0))
+		defer env.Cleanup()
+		ledgerid := "testledger"
+		blkfileMgrWrapper := newTestBlockfileWrapper(env, ledgerid)
 		defer blkfileMgrWrapper.close()
-		blkfileMgr = blkfileMgrWrapper.blockfileMgr
-	} else {
-		blkfileMgr.syncIndex()
-	}
-
-	// Now, last set of blocks should also be present in original index
-	for i := numBlocksToIndex + 1; i <= numBlocks; i++ {
-		block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
-		testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
-		testutil.AssertEquals(t, block, blocks[i-1])
-	}
+		blkfileMgr := blkfileMgrWrapper.blockfileMgr
+		origIndex := blkfileMgr.index
+		// construct blocks for testing
+		blocks := testutil.ConstructTestBlocks(t, numBlocks)
+		// add a few blocks
+		blkfileMgrWrapper.addBlocks(blocks[:numBlocksToIndex])
+
+		// Plug-in a noop index and add remaining blocks
+		blkfileMgr.index = &noopIndex{}
+		blkfileMgrWrapper.addBlocks(blocks[numBlocksToIndex:])
+
+		// Plug-in back the original index
+		blkfileMgr.index = origIndex
+		// The first set of blocks should be present in the orginal index
+		for i := 0; i < numBlocksToIndex; i++ {
+			block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
+			testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
+			testutil.AssertEquals(t, block, blocks[i])
+		}
+
+		// The last set of blocks should not be present in the original index
+		for i := numBlocksToIndex + 1; i <= numBlocks; i++ {
+			_, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
+			testutil.AssertSame(t, err, blkstorage.ErrNotFoundInIndex)
+		}
+
+		// perform index sync
+		if syncByRestart {
+			blkfileMgrWrapper.close()
+			blkfileMgrWrapper = newTestBlockfileWrapper(env, ledgerid)
+			defer blkfileMgrWrapper.close()
+			blkfileMgr = blkfileMgrWrapper.blockfileMgr
+		} else {
+			blkfileMgr.syncIndex()
+		}
+
+		// Now, last set of blocks should also be present in original index
+		for i := numBlocksToIndex; i < numBlocks; i++ {
+			block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
+			testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
+			testutil.AssertEquals(t, block, blocks[i])
+		}
+	})
 }
 
 func TestBlockIndexSelectiveIndexing(t *testing.T) {
@@ -119,69 +122,75 @@ func TestBlockIndexSelectiveIndexing(t *testing.T) {
 }
 
 func testBlockIndexSelectiveIndexing(t *testing.T, indexItems []blkstorage.IndexableAttr) {
-	env := newTestEnvSelectiveIndexing(t, NewConf(testPath, 0), indexItems)
-	defer env.Cleanup()
-	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testledger")
-	defer blkfileMgrWrapper.close()
-
-	blocks := testutil.ConstructTestBlocks(t, 3)
-	// add test blocks
-	blkfileMgrWrapper.addBlocks(blocks)
-	blockfileMgr := blkfileMgrWrapper.blockfileMgr
-
-	// if index has been configured for an indexItem then the item should be indexed else not
-	// test 'retrieveBlockByHash'
-	block, err := blockfileMgr.retrieveBlockByHash(blocks[0].Header.Hash())
-	if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockHash) {
-		testutil.AssertNoError(t, err, "Error while retrieving block by hash")
-		testutil.AssertEquals(t, block, blocks[0])
-	} else {
-		testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
-	}
-
-	// test 'retrieveBlockByNumber'
-	block, err = blockfileMgr.retrieveBlockByNumber(1)
-	if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockNum) {
-		testutil.AssertNoError(t, err, "Error while retrieving block by number")
-		testutil.AssertEquals(t, block, blocks[0])
-	} else {
-		testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
+	var testName string
+	for _, s := range indexItems {
+		testName = testName + string(s)
 	}
+	t.Run(testName, func(t *testing.T) {
+		env := newTestEnvSelectiveIndexing(t, NewConf(testPath(), 0), indexItems)
+		defer env.Cleanup()
+		blkfileMgrWrapper := newTestBlockfileWrapper(env, "testledger")
+		defer blkfileMgrWrapper.close()
 
-	// test 'retrieveTransactionByID'
-	txid, err := extractTxID(blocks[0].Data.Data[0])
-	testutil.AssertNoError(t, err, "")
-	txEnvelope, err := blockfileMgr.retrieveTransactionByID(txid)
-	if testutil.Contains(indexItems, blkstorage.IndexableAttrTxID) {
-		testutil.AssertNoError(t, err, "Error while retrieving tx by id")
-		txEnvelopeBytes := blocks[0].Data.Data[0]
-		txEnvelopeOrig, err := putil.GetEnvelopeFromBlock(txEnvelopeBytes)
+		blocks := testutil.ConstructTestBlocks(t, 3)
+		// add test blocks
+		blkfileMgrWrapper.addBlocks(blocks)
+		blockfileMgr := blkfileMgrWrapper.blockfileMgr
+
+		// if index has been configured for an indexItem then the item should be indexed else not
+		// test 'retrieveBlockByHash'
+		block, err := blockfileMgr.retrieveBlockByHash(blocks[0].Header.Hash())
+		if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockHash) {
+			testutil.AssertNoError(t, err, "Error while retrieving block by hash")
+			testutil.AssertEquals(t, block, blocks[0])
+		} else {
+			testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
+		}
+
+		// test 'retrieveBlockByNumber'
+		block, err = blockfileMgr.retrieveBlockByNumber(0)
+		if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockNum) {
+			testutil.AssertNoError(t, err, "Error while retrieving block by number")
+			testutil.AssertEquals(t, block, blocks[0])
+		} else {
+			testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
+		}
+
+		// test 'retrieveTransactionByID'
+		txid, err := extractTxID(blocks[0].Data.Data[0])
 		testutil.AssertNoError(t, err, "")
-		testutil.AssertEquals(t, txEnvelope, txEnvelopeOrig)
-	} else {
-		testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
-	}
-
-	//test 'retrieveTrasnactionsByBlockNumTranNum
-	txEnvelope2, err := blockfileMgr.retrieveTransactionByBlockNumTranNum(1, 1)
-	if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockNumTranNum) {
-		testutil.AssertNoError(t, err, "Error while retrieving tx by blockNum and tranNum")
-		txEnvelopeBytes2 := blocks[0].Data.Data[0]
-		txEnvelopeOrig2, err2 := putil.GetEnvelopeFromBlock(txEnvelopeBytes2)
-		testutil.AssertNoError(t, err2, "")
-		testutil.AssertEquals(t, txEnvelope2, txEnvelopeOrig2)
-	} else {
-		testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
-	}
-
-	// test 'retrieveBlockByTxID'
-	txid, err = extractTxID(blocks[0].Data.Data[0])
-	testutil.AssertNoError(t, err, "")
-	block, err = blockfileMgr.retrieveBlockByTxID(txid)
-	if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockTxID) {
-		testutil.AssertNoError(t, err, "Error while retrieving block by txID")