[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 <[email protected]>

Author: Luis Sanchez

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,15 +159,15 @@ 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))
 		}
 		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
 }