Blockchain v2 processor (#4012)

* Add processor prototype

* Change processor API

    + expose a simple `handle` function which mutates internal state

* processor tests

* fix gofmt and ohter golangci issues

* scopelint var on range scope

* add check for short block received

* fix formatting

* small test reorg

* ignore unused for now

* ci fix changes

* go.mod revert

blockchain/v2/processor.go

@@ -0,0 +1,186 @@
+package v2
+
+import (
+	"fmt"
+
+	"github.com/tendermint/tendermint/p2p"
+	tdState "github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/types"
+)
+
+type peerError struct {
+	priorityHigh
+	peerID p2p.ID
+}
+
+type pcDuplicateBlock struct {
+	priorityNormal
+}
+
+type pcShortBlock struct {
+	priorityNormal
+}
+
+type bcBlockResponse struct {
+	priorityNormal
+	peerID p2p.ID
+	block  *types.Block
+	height int64
+}
+
+type pcBlockVerificationFailure struct {
+	priorityNormal
+	peerID p2p.ID
+	height int64
+}
+
+type pcBlockProcessed struct {
+	priorityNormal
+	height int64
+	peerID p2p.ID
+}
+
+type pcProcessBlock struct {
+	priorityNormal
+}
+
+type pcStop struct {
+	priorityNormal
+}
+
+type pcFinished struct {
+	priorityNormal
+	height       int64
+	blocksSynced int64
+}
+
+func (p pcFinished) Error() string {
+	return "finished"
+}
+
+type queueItem struct {
+	block  *types.Block
+	peerID p2p.ID
+}
+
+type blockQueue map[int64]queueItem
+
+type pcState struct {
+	height       int64      // height of the last synced block
+	queue        blockQueue // blocks waiting to be processed
+	chainID      string
+	blocksSynced int64
+	draining     bool
+	tdState      tdState.State
+	context      processorContext
+}
+
+func (state *pcState) String() string {
+	return fmt.Sprintf("height: %d queue length: %d draining: %v blocks synced: %d",
+		state.height, len(state.queue), state.draining, state.blocksSynced)
+}
+
+// newPcState returns a pcState initialized with the last verified block enqueued
+func newPcState(initHeight int64, tdState tdState.State, chainID string, context processorContext) *pcState {
+	return &pcState{
+		height:       initHeight,
+		queue:        blockQueue{},
+		chainID:      chainID,
+		draining:     false,
+		blocksSynced: 0,
+		context:      context,
+		tdState:      tdState,
+	}
+}
+
+// nextTwo returns the next two unverified blocks
+func (state *pcState) nextTwo() (queueItem, queueItem, error) {
+	if first, ok := state.queue[state.height+1]; ok {
+		if second, ok := state.queue[state.height+2]; ok {
+			return first, second, nil
+		}
+	}
+	return queueItem{}, queueItem{}, fmt.Errorf("not found")
+}
+
+// synced returns true when at most the last verified block remains in the queue
+func (state *pcState) synced() bool {
+	return len(state.queue) <= 1
+}
+
+func (state *pcState) advance() {
+	state.height++
+	delete(state.queue, state.height)
+	state.blocksSynced++
+}
+
+func (state *pcState) enqueue(peerID p2p.ID, block *types.Block, height int64) error {
+	if _, ok := state.queue[height]; ok {
+		return fmt.Errorf("duplicate queue item")
+	}
+	state.queue[height] = queueItem{block: block, peerID: peerID}
+	return nil
+}
+
+// purgePeer moves all unprocessed blocks from the queue
+func (state *pcState) purgePeer(peerID p2p.ID) {
+	// what if height is less than state.height?
+	for height, item := range state.queue {
+		if item.peerID == peerID {
+			delete(state.queue, height)
+		}
+	}
+}
+
+// handle processes FSM events
+func (state *pcState) handle(event Event) (Event, error) {
+	switch event := event.(type) {
+	case *bcBlockResponse:
+		if event.height <= state.height {
+			return pcShortBlock{}, nil
+		}
+		err := state.enqueue(event.peerID, event.block, event.height)
+		if err != nil {
+			return pcDuplicateBlock{}, nil
+		}
+
+	case pcProcessBlock:
+		firstItem, secondItem, err := state.nextTwo()
+		if err != nil {
+			if state.draining {
+				return noOp, pcFinished{height: state.height}
+			}
+			return noOp, nil
+		}
+		first, second := firstItem.block, secondItem.block
+
+		firstParts := first.MakePartSet(types.BlockPartSizeBytes)
+		firstPartsHeader := firstParts.Header()
+		firstID := types.BlockID{Hash: first.Hash(), PartsHeader: firstPartsHeader}
+
+		err = state.context.verifyCommit(state.chainID, firstID, first.Height, second.LastCommit)
+		if err != nil {
+			return pcBlockVerificationFailure{peerID: firstItem.peerID, height: first.Height}, nil
+		}
+
+		state.context.saveBlock(first, firstParts, second.LastCommit)
+
+		state.tdState, err = state.context.applyBlock(state.tdState, firstID, first)
+		if err != nil {
+			panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err))
+		}
+		state.advance()
+		return pcBlockProcessed{height: first.Height, peerID: firstItem.peerID}, nil
+
+	case *peerError:
+		state.purgePeer(event.peerID)
+
+	case pcStop:
+		if state.synced() {
+			return noOp, pcFinished{height: state.height, blocksSynced: state.blocksSynced}
+		}
+		state.draining = true
+	}
+
+	return noOp, nil
+}

blockchain/v2/processor_context.go

@@ -0,0 +1,76 @@
+package v2
+
+import (
+	"fmt"
+
+	"github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/store"
+	"github.com/tendermint/tendermint/types"
+)
+
+type processorContext interface {
+	applyBlock(state state.State, blockID types.BlockID, block *types.Block) (state.State, error)
+	verifyCommit(chainID string, blockID types.BlockID, height int64, commit *types.Commit) error
+	saveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit)
+}
+
+// nolint:unused
+type pContext struct {
+	store    *store.BlockStore
+	executor *state.BlockExecutor
+	state    *state.State
+}
+
+// nolint:unused,deadcode
+func newProcessorContext(st *store.BlockStore, ex *state.BlockExecutor, s *state.State) *pContext {
+	return &pContext{
+		store:    st,
+		executor: ex,
+		state:    s,
+	}
+}
+
+func (pc *pContext) applyBlock(state state.State, blockID types.BlockID, block *types.Block) (state.State, error) {
+	return pc.executor.ApplyBlock(state, blockID, block)
+}
+
+func (pc *pContext) verifyCommit(chainID string, blockID types.BlockID, height int64, commit *types.Commit) error {
+	return pc.state.Validators.VerifyCommit(chainID, blockID, height, commit)
+}
+
+func (pc *pContext) saveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) {
+	pc.store.SaveBlock(block, blockParts, seenCommit)
+}
+
+type mockPContext struct {
+	applicationBL  []int64
+	verificationBL []int64
+}
+
+func newMockProcessorContext(verificationBlackList []int64, applicationBlackList []int64) *mockPContext {
+	return &mockPContext{
+		applicationBL:  applicationBlackList,
+		verificationBL: verificationBlackList,
+	}
+}
+
+func (mpc *mockPContext) applyBlock(state state.State, blockID types.BlockID, block *types.Block) (state.State, error) {
+	for _, h := range mpc.applicationBL {
+		if h == block.Height {
+			return state, fmt.Errorf("generic application error")
+		}
+	}
+	return state, nil
+}
+
+func (mpc *mockPContext) verifyCommit(chainID string, blockID types.BlockID, height int64, commit *types.Commit) error {
+	for _, h := range mpc.verificationBL {
+		if h == height {
+			return fmt.Errorf("generic verification error")
+		}
+	}
+	return nil
+}
+
+func (mpc *mockPContext) saveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) {
+}

blockchain/v2/processor_test.go

@@ -0,0 +1,336 @@
+package v2
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tendermint/tendermint/p2p"
+	tdState "github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/types"
+)
+
+// pcBlock is a test helper structure with simple types. Its purpose is to help with test readability.
+type pcBlock struct {
+	pid    string
+	height int64
+}
+
+// params is a test structure used to create processor state.
+type params struct {
+	height       int64
+	items        []pcBlock
+	blocksSynced int64
+	verBL        []int64
+	appBL        []int64
+	draining     bool
+}
+
+// makePcBlock makes an empty block.
+func makePcBlock(height int64) *types.Block {
+	return &types.Block{Header: types.Header{Height: height}}
+}
+
+// makeState takes test parameters and creates a specific processor state.
+func makeState(p *params) *pcState {
+	var (
+		tdState = tdState.State{}
+		context = newMockProcessorContext(p.verBL, p.appBL)
+	)
+	state := newPcState(p.height, tdState, "test", context)
+
+	for _, item := range p.items {
+		_ = state.enqueue(p2p.ID(item.pid), makePcBlock(item.height), item.height)
+	}
+
+	state.blocksSynced = p.blocksSynced
+	state.draining = p.draining
+	return state
+}
+
+func mBlockResponse(peerID p2p.ID, height int64) *bcBlockResponse {
+	return &bcBlockResponse{
+		peerID: peerID,
+		block:  makePcBlock(height),
+		height: height,
+	}
+}
+
+type pcFsmMakeStateValues struct {
+	currentState  *params
+	event         Event
+	wantState     *params
+	wantNextEvent Event
+	wantErr       error
+	wantPanic     bool
+}
+
+type testFields struct {
+	name  string
+	steps []pcFsmMakeStateValues
+}
+
+func executeProcessorTests(t *testing.T, tests []testFields) {
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			var state *pcState
+			for _, step := range tt.steps {
+				defer func() {
+					r := recover()
+					if (r != nil) != step.wantPanic {
+						t.Errorf("recover = %v, wantPanic = %v", r, step.wantPanic)
+					}
+				}()
+
+				// First step must always initialise the currentState as state.
+				if step.currentState != nil {
+					state = makeState(step.currentState)
+				}
+				if state == nil {
+					panic("Bad (initial?) step")
+				}
+
+				nextEvent, err := state.handle(step.event)
+				t.Log(state)
+				assert.Equal(t, step.wantErr, err)
+				assert.Equal(t, makeState(step.wantState), state)
+				assert.Equal(t, step.wantNextEvent, nextEvent)
+				// Next step may use the wantedState as their currentState.
+				state = makeState(step.wantState)
+			}
+		})
+	}
+}
+
+func TestPcBlockResponse(t *testing.T) {
+	tests := []testFields{
+
+		{
+			name: "add one block",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{}, event: mBlockResponse("P1", 1),
+					wantState: &params{items: []pcBlock{{"P1", 1}}}, wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "add two blocks",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{}, event: mBlockResponse("P1", 3),
+					wantState: &params{items: []pcBlock{{"P1", 3}}}, wantNextEvent: noOp,
+				},
+				{ // use previous wantState as currentState,
+					event:     mBlockResponse("P1", 4),
+					wantState: &params{items: []pcBlock{{"P1", 3}, {"P1", 4}}}, wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "add duplicate block from same peer",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{}, event: mBlockResponse("P1", 3),
+					wantState: &params{items: []pcBlock{{"P1", 3}}}, wantNextEvent: noOp,
+				},
+				{ // use previous wantState as currentState,
+					event:     mBlockResponse("P1", 3),
+					wantState: &params{items: []pcBlock{{"P1", 3}}}, wantNextEvent: pcDuplicateBlock{},
+				},
+			},
+		},
+		{
+			name: "add duplicate block from different peer",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{}, event: mBlockResponse("P1", 3),
+					wantState: &params{items: []pcBlock{{"P1", 3}}}, wantNextEvent: noOp,
+				},
+				{ // use previous wantState as currentState,
+					event:     mBlockResponse("P2", 3),
+					wantState: &params{items: []pcBlock{{"P1", 3}}}, wantNextEvent: pcDuplicateBlock{},
+				},
+			},
+		},
+		{
+			name: "attempt to add block with height equal to state.height",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 2, items: []pcBlock{{"P1", 3}}}, event: mBlockResponse("P1", 2),
+					wantState: &params{height: 2, items: []pcBlock{{"P1", 3}}}, wantNextEvent: pcShortBlock{},
+				},
+			},
+		},
+		{
+			name: "attempt to add block with height smaller than state.height",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 2, items: []pcBlock{{"P1", 3}}}, event: mBlockResponse("P1", 1),
+					wantState: &params{height: 2, items: []pcBlock{{"P1", 3}}}, wantNextEvent: pcShortBlock{},
+				},
+			},
+		},
+	}
+
+	executeProcessorTests(t, tests)
+}
+
+func TestPcProcessBlockSuccess(t *testing.T) {
+	tests := []testFields{
+		{
+			name: "noop - no blocks over current height",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{}, event: pcProcessBlock{},
+					wantState: &params{}, wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "noop - high new blocks",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 5, items: []pcBlock{{"P1", 30}, {"P2", 31}}}, event: pcProcessBlock{},
+					wantState: &params{height: 5, items: []pcBlock{{"P1", 30}, {"P2", 31}}}, wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "blocks H+1 and H+2 present",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}}, event: pcProcessBlock{},
+					wantState:     &params{height: 1, items: []pcBlock{{"P2", 2}}, blocksSynced: 1},
+					wantNextEvent: pcBlockProcessed{height: 1, peerID: "P1"},
+				},
+			},
+		},
+		{
+			name: "blocks H+1 and H+2 present after draining",
+			steps: []pcFsmMakeStateValues{
+				{ // some contiguous blocks - on stop check draining is set
+					currentState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}, {"P1", 4}}}, event: pcStop{},
+					wantState:     &params{items: []pcBlock{{"P1", 1}, {"P2", 2}, {"P1", 4}}, draining: true},
+					wantNextEvent: noOp,
+				},
+				{
+					event:         pcProcessBlock{},
+					wantState:     &params{height: 1, items: []pcBlock{{"P2", 2}, {"P1", 4}}, blocksSynced: 1, draining: true},
+					wantNextEvent: pcBlockProcessed{height: 1, peerID: "P1"},
+				},
+				{ // finish when H+1 or/and H+2 are missing
+					event:         pcProcessBlock{},
+					wantState:     &params{height: 1, items: []pcBlock{{"P2", 2}, {"P1", 4}}, blocksSynced: 1, draining: true},
+					wantNextEvent: noOp,
+					wantErr:       pcFinished{height: 1},
+				},
+			},
+		},
+	}
+
+	executeProcessorTests(t, tests)
+}
+
+func TestPcProcessBlockFailures(t *testing.T) {
+	tests := []testFields{
+		{
+			name: "blocks H+1 and H+2 present - H+1 verification fails ",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}, verBL: []int64{1}}, event: pcProcessBlock{},
+					wantState:     &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}, verBL: []int64{1}},
+					wantNextEvent: pcBlockVerificationFailure{peerID: "P1", height: 1},
+				},
+			},
+		},
+		{
+			name: "blocks H+1 and H+2 present - H+1 applyBlock fails ",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}, appBL: []int64{1}}, event: pcProcessBlock{},
+					wantState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}, appBL: []int64{1}}, wantPanic: true,
+				},
+			},
+		},
+	}
+
+	executeProcessorTests(t, tests)
+}
+
+func TestPcPeerError(t *testing.T) {
+	tests := []testFields{
+		{
+			name: "peer not present",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}}, event: &peerError{peerID: "P3"},
+					wantState:     &params{items: []pcBlock{{"P1", 1}, {"P2", 2}}},
+					wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "some blocks are from errored peer",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 100}, {"P1", 99}, {"P2", 101}}}, event: &peerError{peerID: "P1"},
+					wantState:     &params{items: []pcBlock{{"P2", 101}}},
+					wantNextEvent: noOp,
+				},
+			},
+		},
+		{
+			name: "all blocks are from errored peer",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{items: []pcBlock{{"P1", 100}, {"P1", 99}}}, event: &peerError{peerID: "P1"},
+					wantState:     &params{},
+					wantNextEvent: noOp,
+				},
+			},
+		},
+	}
+
+	executeProcessorTests(t, tests)
+}
+
+func TestStop(t *testing.T) {
+	tests := []testFields{
+		{
+			name: "no blocks",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 100, items: []pcBlock{}, blocksSynced: 100}, event: pcStop{},
+					wantState:     &params{height: 100, items: []pcBlock{}, blocksSynced: 100},
+					wantNextEvent: noOp,
+					wantErr:       pcFinished{height: 100, blocksSynced: 100},
+				},
+			},
+		},
+		{
+			name: "maxHeight+1 block present",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 100, items: []pcBlock{{"P1", 101}}, blocksSynced: 100}, event: pcStop{},
+					wantState:     &params{height: 100, items: []pcBlock{{"P1", 101}}, blocksSynced: 100},
+					wantNextEvent: noOp,
+					wantErr:       pcFinished{height: 100, blocksSynced: 100},
+				},
+			},
+		},
+		{
+			name: "more blocks present",
+			steps: []pcFsmMakeStateValues{
+				{
+					currentState: &params{height: 100, items: []pcBlock{{"P1", 101}, {"P1", 102}}, blocksSynced: 100}, event: pcStop{},
+					wantState:     &params{height: 100, items: []pcBlock{{"P1", 101}, {"P1", 102}}, blocksSynced: 100, draining: true},
+					wantNextEvent: noOp,
+					wantErr:       nil,
+				},
+			},
+		},
+	}
+
+	executeProcessorTests(t, tests)
+}

blockchain/v2/reactor.go

@@ -69,6 +69,7 @@ func (r *Reactor) Start() {
 	}()
 }
 
+// XXX: How to make this deterministic?
 // XXX: Would it be possible here to provide some kind of type safety for the types
 // of events that each routine can produce and consume?
 func (r *Reactor) demux() {

blockchain/v2/schedule.go

@@ -164,6 +164,7 @@ func (sc *schedule) removePeer(peerID p2p.ID) error {
 	return nil
 }
 
+// TODO - keep track of highest height
 func (sc *schedule) setPeerHeight(peerID p2p.ID, height int64) error {
 	peer, ok := sc.peers[peerID]
 	if !ok {