[FAB-7290] Handle Nil pointer panic in blocks iterator

This CR add a simple nil check before calling close function
on blockfile stream instance. In addition, a couple of tests
are added.

Change-Id: I053b289b32c5dfeab5824eb866a954dc4aa5cfa9
Signed-off-by: manish <[email protected]>
Signed-off-by: Gari Singh <[email protected]>

Author: manish-sethi

common/ledger/blkstorage/fsblkstorage/blocks_itr.go

@@ -77,6 +77,7 @@ func (itr *blocksItr) Next() (ledger.QueryResult, error) {
 		return nil, nil
 	}
 	if itr.stream == nil {
+		logger.Debugf("Initializing block stream for iterator. itr.maxBlockNumAvailable=%d", itr.maxBlockNumAvailable)
 		if err := itr.initStream(); err != nil {
 			return nil, err
 		}
@@ -97,5 +98,7 @@ func (itr *blocksItr) Close() {
 	itr.mgr.cpInfoCond.L.Lock()
 	defer itr.mgr.cpInfoCond.L.Unlock()
 	itr.mgr.cpInfoCond.Broadcast()
-	itr.stream.close()
+	if itr.stream != nil {
+		itr.stream.close()
+	}
 }

common/ledger/blkstorage/fsblkstorage/blocks_itr_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package fsblkstorage
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -74,6 +75,68 @@ func TestBlockItrClose(t *testing.T) {
 	testutil.AssertNil(t, bh)
 }
 
+func TestBlockItrCloseWithoutRetrieve(t *testing.T) {
+	env := newTestEnv(t, NewConf(testPath(), 0))
+	defer env.Cleanup()
+	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
+	defer blkfileMgrWrapper.close()
+	blkfileMgr := blkfileMgrWrapper.blockfileMgr
+	blocks := testutil.ConstructTestBlocks(t, 5)
+	blkfileMgrWrapper.addBlocks(blocks)
+
+	itr, err := blkfileMgr.retrieveBlocks(2)
+	testutil.AssertNoError(t, err, "")
+	itr.Close()
+}
+
+func TestCloseMultipleItrsWaitForFutureBlock(t *testing.T) {
+	env := newTestEnv(t, NewConf(testPath(), 0))
+	defer env.Cleanup()
+	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
+	defer blkfileMgrWrapper.close()
+	blkfileMgr := blkfileMgrWrapper.blockfileMgr
+	blocks := testutil.ConstructTestBlocks(t, 10)
+	blkfileMgrWrapper.addBlocks(blocks[:5])
+
+	wg := &sync.WaitGroup{}
+	wg.Add(2)
+	itr1, err := blkfileMgr.retrieveBlocks(7)
+	testutil.AssertNoError(t, err, "")
+	// itr1 does not retrieve any block because it closes before new blocks are added
+	go iterateInBackground(t, itr1, 9, wg, []uint64{})
+
+	itr2, err := blkfileMgr.retrieveBlocks(8)
+	testutil.AssertNoError(t, err, "")
+	// itr2 retrieves two blocks 8 and 9. Because it started waiting for 8 and quits at 9
+	go iterateInBackground(t, itr2, 9, wg, []uint64{8, 9})
+
+	// sleep for the background iterators to get started
+	time.Sleep(2 * time.Second)
+	itr1.Close()
+	blkfileMgrWrapper.addBlocks(blocks[5:])
+	wg.Wait()
+}
+
+func iterateInBackground(t *testing.T, itr *blocksItr, quitAfterBlkNum uint64, wg *sync.WaitGroup, expectedBlockNums []uint64) {
+	defer wg.Done()
+	retrievedBlkNums := []uint64{}
+	defer func() { testutil.AssertEquals(t, retrievedBlkNums, expectedBlockNums) }()
+
+	for {
+		blk, err := itr.Next()
+		testutil.AssertNoError(t, err, "")
+		if blk == nil {
+			return
+		}
+		blkNum := blk.(*common.Block).Header.Number
+		retrievedBlkNums = append(retrievedBlkNums, blkNum)
+		t.Logf("blk.Num=%d", blk.(*common.Block).Header.Number)
+		if blkNum == quitAfterBlkNum {
+			return
+		}
+	}
+}
+
 func testIterateAndVerify(t *testing.T, itr *blocksItr, blocks []*common.Block, doneChan chan bool) {
 	blocksIterated := 0
 	for {