Address PR comments, and some cleanup

statediff/service/service.go

@@ -53,12 +53,12 @@ func (sds *StateDiffService) Loop(chainEventCh chan core.ChainEvent) {
 	chainEventSub := sds.BlockChain.SubscribeChainEvent(chainEventCh)
 	defer chainEventSub.Unsubscribe()
 
-	blocksCh := make(chan *types.Block)
+	blocksCh := make(chan *types.Block, 10)
 	errCh := chainEventSub.Err()
 	quitCh := make(chan struct{})
 
 	go func() {
-	HandleLoop:
+	HandleChainEventChLoop:
 		for {
 			select {
 			//Notify chain event channel of events
@@ -67,14 +67,15 @@ func (sds *StateDiffService) Loop(chainEventCh chan core.ChainEvent) {
 				blocksCh <- chainEvent.Block
 			//if node stopped
 			case err := <-errCh:
-				log.Debug("Error from chain event subscription, breaking loop.", "error", err)
-				break HandleLoop
+				log.Warn("Error from chain event subscription, breaking loop.", "error", err)
+				break HandleChainEventChLoop
 			}
 		}
 		close(quitCh)
 	}()
 
 	//loop through chain events until no more
+HandleBlockChLoop:
 	for {
 		select {
 		case block := <-blocksCh:
@@ -82,10 +83,10 @@ func (sds *StateDiffService) Loop(chainEventCh chan core.ChainEvent) {
 			parentHash := currentBlock.ParentHash()
 			parentBlock := sds.BlockChain.GetBlockByHash(parentHash)
 			if parentBlock == nil {
-				log.Warn("Parent block is nil, skipping this block",
+				log.Error("Parent block is nil, skipping this block",
 					"parent block hash", parentHash.String(),
-						"current block number", currentBlock.Number())
-				break
+					"current block number", currentBlock.Number())
+				break HandleBlockChLoop
 			}
 
 			stateDiffLocation, err := sds.Extractor.ExtractStateDiff(*parentBlock, *currentBlock)
@@ -95,6 +96,7 @@ func (sds *StateDiffService) Loop(chainEventCh chan core.ChainEvent) {
 				log.Info("Statediff extracted", "block number", currentBlock.Number(), "location", stateDiffLocation)
 			}
 		case <-quitCh:
+			log.Debug("Quitting the statediff block channel")
 			return
 		}
 	}

statediff/service/service_test.go

@@ -14,11 +14,12 @@ import (
 )
 
 func TestServiceLoop(t *testing.T) {
-	testServiceLoop(t)
+	testErrorInChainEventLoop(t)
+	testErrorInBlockLoop(t)
 }
 
 var (
-	eventsChannel = make(chan core.ChainEvent)
+	eventsChannel = make(chan core.ChainEvent, 1)
 
 	parentHeader1 = types.Header{Number: big.NewInt(rand.Int63())}
 	parentHeader2 = types.Header{Number: big.NewInt(rand.Int63())}
@@ -31,18 +32,20 @@ var (
 
 	header1 = types.Header{ParentHash: parentHash1}
 	header2 = types.Header{ParentHash: parentHash2}
+	header3 = types.Header{ParentHash: common.HexToHash("parent hash")}
 
 	block1 = types.NewBlock(&header1, nil, nil, nil)
 	block2 = types.NewBlock(&header2, nil, nil, nil)
+	block3 = types.NewBlock(&header3, nil, nil, nil)
 
 	event1 = core.ChainEvent{Block: block1}
 	event2 = core.ChainEvent{Block: block2}
+	event3 = core.ChainEvent{Block: block3}
 )
 
-func testServiceLoop(t *testing.T) {
-
+func testErrorInChainEventLoop(t *testing.T) {
+	//the first chain event causes and error (in blockchain mock)
 	extractor := mocks.Extractor{}
-	//close(eventsChannel)
 
 	blockChain := mocks.BlockChain{}
 	service := s.StateDiffService{
@@ -51,8 +54,8 @@ func testServiceLoop(t *testing.T) {
 		BlockChain: &blockChain,
 	}
 
-	blockChain.SetParentBlockToReturn([]*types.Block{parentBlock1, parentBlock2})
-	blockChain.SetChainEvents([]core.ChainEvent{event1, event2})
+	blockChain.SetParentBlocksToReturn([]*types.Block{parentBlock1, parentBlock2})
+	blockChain.SetChainEvents([]core.ChainEvent{event1, event2, event3})
 	service.Loop(eventsChannel)
 
 	//parent and current blocks are passed to the extractor
@@ -74,3 +77,31 @@ func testServiceLoop(t *testing.T) {
 		t.Logf("Actual does not equal expected.\nactual:%+v\nexpected: %+v", blockChain.ParentHashesLookedUp, expectedHashes)
 	}
 }
+
+func testErrorInBlockLoop(t *testing.T) {
+	//second block's parent block can't be found
+	extractor := mocks.Extractor{}
+
+	blockChain := mocks.BlockChain{}
+	service := s.StateDiffService{
+		Builder:    nil,
+		Extractor:  &extractor,
+		BlockChain: &blockChain,
+	}
+
+	blockChain.SetParentBlocksToReturn([]*types.Block{parentBlock1, nil})
+	blockChain.SetChainEvents([]core.ChainEvent{event1, event2})
+	service.Loop(eventsChannel)
+
+	//only the first current block (and it's parent) are passed to the extractor
+	expectedCurrentBlocks := []types.Block{*block1}
+	if !reflect.DeepEqual(extractor.CurrentBlocks, expectedCurrentBlocks) {
+		t.Error("Test failure:", t.Name())
+		t.Logf("Actual does not equal expected.\nactual:%+v\nexpected: %+v", extractor.CurrentBlocks, expectedCurrentBlocks)
+	}
+	expectedParentBlocks := []types.Block{*parentBlock1}
+	if !reflect.DeepEqual(extractor.ParentBlocks, expectedParentBlocks) {
+		t.Error("Test failure:", t.Name())
+		t.Logf("Actual does not equal expected.\nactual:%+v\nexpected: %+v", extractor.CurrentBlocks, expectedParentBlocks)
+	}
+}

statediff/testhelpers/mocks/blockchain.go

@@ -2,6 +2,7 @@ package mocks
 
 import (
 	"errors"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -15,20 +16,20 @@ type BlockChain struct {
 	ChainEvents          []core.ChainEvent
 }
 
-func (mc *BlockChain) SetParentBlockToReturn(blocks []*types.Block) {
+func (mc *BlockChain) SetParentBlocksToReturn(blocks []*types.Block) {
 	mc.parentBlocksToReturn = blocks
 }
 
 func (mc *BlockChain) GetBlockByHash(hash common.Hash) *types.Block {
 	mc.ParentHashesLookedUp = append(mc.ParentHashesLookedUp, hash)
 
-	var parentBlock types.Block
+	var parentBlock *types.Block
 	if len(mc.parentBlocksToReturn) > 0 {
-		parentBlock = *mc.parentBlocksToReturn[mc.callCount]
+		parentBlock = mc.parentBlocksToReturn[mc.callCount]
 	}
 
 	mc.callCount++
-	return &parentBlock
+	return parentBlock
 }
 
 func (bc *BlockChain) SetChainEvents(chainEvents []core.ChainEvent) {