fix(consensus): don't propose at historical heights after a bad block-sync handover

internal/blocksync/applier.go

@@ -80,17 +80,35 @@ func (e *blockApplier) Apply(ctx context.Context, block *types.Block, commit *ty
 	}
 	verifyTime := time.Since(start)
 
+	// The two halves of sm.Executor.ApplyBlock are run separately so the block
+	// store is advanced between them. An application that refuses the block does
+	// so here, and refusing it must leave nothing behind: a block persisted for a
+	// height the application never processed is re-processed by the handshake on
+	// every later start, by the same application that already refused it
+	// (dashpay/tenderdash#1413). Saving it before the application commits is what
+	// keeps the store from ever falling behind the application, which the
+	// handshake rejects outright.
+	//
+	// verify is false because FinalizeBlock runs ValidateBlockWithRoundState with
+	// the same arguments; verifying here as well costs a second threshold
+	// signature verification of block.LastCommit per block.
+	start = time.Now()
+	uncommittedState, err := e.blockExec.ProcessProposal(ctx, block, commit.Round, e.state, false)
+	if err != nil {
+		panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", block.Height, block.Hash(), err))
+	}
+	processTime := time.Since(start)
+
 	start = time.Now()
 	e.store.SaveBlock(block, blockParts, commit)
 	saveTime := time.Since(start)
 
 	start = time.Now()
-	// TODO: Same thing for app - but we would need a way to get the hash without persisting the state.
-	e.state, err = e.blockExec.ApplyBlock(ctx, e.state, blockID, block, commit)
+	e.state, err = e.blockExec.FinalizeBlock(ctx, e.state, uncommittedState, blockID, block, commit)
 	if err != nil {
-		panic(fmt.Sprintf("failed to process committed block (%d:%X): %v", block.Height, block.Hash(), err))
+		panic(fmt.Sprintf("failed to finalize committed block (%d:%X): %v", block.Height, block.Hash(), err))
 	}
-	execTime := time.Since(start)
+	execTime := processTime + time.Since(start)
 
 	e.stats.add(partSetTime, verifyTime, saveTime, execTime)
 	// ByteSize is the size of the serialized block we just built, so the metric

internal/blocksync/applier_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	sm "github.com/dashpay/tenderdash/internal/state"
 	"github.com/dashpay/tenderdash/internal/state/mocks"
 	statefactory "github.com/dashpay/tenderdash/internal/state/test/factory"
 	"github.com/dashpay/tenderdash/internal/test/factory"
@@ -48,7 +49,11 @@ func TestBlockApplierApply(t *testing.T) {
 					Once().
 					Return(nil)
 				mockBlockExec.
-					On("ApplyBlock", mock.Anything, initialState, blockH1ID, blockH1, commitH1).
+					On("ProcessProposal", mock.Anything, blockH1, commitH1.Round, initialState, false).
+					Once().
+					Return(sm.CurrentRoundState{}, nil)
+				mockBlockExec.
+					On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, blockH1ID, blockH1, commitH1).
 					Once().
 					Return(state, nil)
 			},
@@ -74,7 +79,11 @@ func TestBlockApplierApply(t *testing.T) {
 					Once().
 					Return(nil)
 				mockBlockExec.
-					On("ApplyBlock", mock.Anything, initialState, blockH1ID, blockH1, commitH1).
+					On("ProcessProposal", mock.Anything, blockH1, commitH1.Round, initialState, false).
+					Once().
+					Return(sm.CurrentRoundState{}, nil)
+				mockBlockExec.
+					On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, blockH1ID, blockH1, commitH1).
 					Once().
 					Return(state, errors.New("eeeeeeeee"))
 			},
@@ -149,3 +158,75 @@ func TestApplyStatsSubMillisecondPreserved(t *testing.T) {
 	require.Equal(t, 300*time.Microsecond, timings.PartSet)
 	require.Zero(t, timings.PartSet.Milliseconds(), "the value this test exists to protect")
 }
+
+// TestBlockApplierDoesNotSaveBlockRejectedByApp checks that a block the
+// application refuses never reaches the block store. Persisting it leaves the
+// store a height ahead of both the state and the app, and every later start has
+// to re-process that block through an application that already rejected it once
+// - the restart-proof failure of dashpay/tenderdash#1413.
+func TestBlockApplierDoesNotSaveBlockRejectedByApp(t *testing.T) {
+	ctx := context.Background()
+	mockBlockExec := mocks.NewExecutor(t)
+	// no SaveBlock expectation: the store must not be touched at all, and the
+	// mock fails the test on any call it was not told to expect
+	mockBlockStore := mocks.NewBlockStore(t)
+	valSet, privVals := factory.MockValidatorSet()
+	initialState := fakeInitialState(valSet)
+	state := initialState.Copy()
+	blocks := statefactory.MakeBlocks(ctx, t, 2, &state, privVals, 1)
+	block, commit := blocks[0], blocks[1].LastCommit
+
+	mockBlockExec.
+		On("ValidateBlock", mock.Anything, initialState, block).
+		Once().
+		Return(nil)
+	mockBlockExec.
+		On("ProcessProposal", mock.Anything, block, commit.Round, initialState, false).
+		Once().
+		Return(sm.CurrentRoundState{}, errors.New("app rejected the block"))
+
+	applier := newBlockApplier(mockBlockExec, mockBlockStore, applierWithState(initialState))
+	require.Panics(t, func() { _ = applier.Apply(ctx, block, commit) })
+}
+
+// TestBlockApplierSavesBlockBeforeFinalize pins the order the handshake relies
+// on: the block is in the store before the application commits it. The store may
+// be one height ahead of the state - replayer.go replays that last block on the
+// next start - but a store behind the app or the state is a case it rejects
+// outright, so a crash must never be able to leave one.
+func TestBlockApplierSavesBlockBeforeFinalize(t *testing.T) {
+	ctx := context.Background()
+	mockBlockExec := mocks.NewExecutor(t)
+	mockBlockStore := mocks.NewBlockStore(t)
+	valSet, privVals := factory.MockValidatorSet()
+	initialState := fakeInitialState(valSet)
+	state := initialState.Copy()
+	blocks := statefactory.MakeBlocks(ctx, t, 2, &state, privVals, 1)
+	block, commit := blocks[0], blocks[1].LastCommit
+	blockParts, err := block.MakePartSet(types.BlockPartSizeBytes)
+	require.NoError(t, err)
+
+	var calls []string
+	mockBlockExec.
+		On("ValidateBlock", mock.Anything, initialState, block).
+		Once().
+		Return(nil)
+	mockBlockExec.
+		On("ProcessProposal", mock.Anything, block, commit.Round, initialState, false).
+		Once().
+		Run(func(mock.Arguments) { calls = append(calls, "process") }).
+		Return(sm.CurrentRoundState{}, nil)
+	mockBlockStore.
+		On("SaveBlock", block, blockParts, commit).
+		Once().
+		Run(func(mock.Arguments) { calls = append(calls, "save") })
+	mockBlockExec.
+		On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, block.BlockID(blockParts), block, commit).
+		Once().
+		Run(func(mock.Arguments) { calls = append(calls, "finalize") }).
+		Return(state, nil)
+
+	applier := newBlockApplier(mockBlockExec, mockBlockStore, applierWithState(initialState))
+	require.NoError(t, applier.Apply(ctx, block, commit))
+	require.Equal(t, []string{"process", "save", "finalize"}, calls)
+}

internal/blocksync/reactor.go

@@ -32,18 +32,29 @@ const (
 	// consider block sync stalled after this duration of inactivity
 	syncTimeout = 60 * time.Second
 
-	// hand over to consensus after this duration of inactivity even when peers
+	// hand over to consensus after this duration of inactivity, even when peers
 	// still report higher blocks, so that a wedged synchronizer cannot keep the
-	// node out of consensus forever
+	// node out of consensus forever - but only within maxCatchupGap of the tip
 	maxSyncStall = 10 * time.Minute
+
+	// maxCatchupGap is how far behind the highest height any peer claims the node
+	// may be and still hand over to consensus on the stall backstop above.
+	//
+	// What is left after the handover is covered by consensus catch-up, which
+	// moves about one block per gossip cycle: it closes a gap of a few blocks in
+	// seconds and a gap of thousands never. Beyond this, block sync retrying an
+	// unproductive peer set is the only route to the tip that exists, and giving
+	// up on it puts a validator at heights the network committed long ago
+	// (dashpay/tenderdash#1413).
+	maxCatchupGap int64 = 10
 )
 
 type ReactorOption func(*Reactor)
 
 type consensusReactor interface {
 	// For when we switch from block sync reactor to the consensus
 	// machine.
-	SwitchToConsensus(ctx context.Context, state sm.State, skipWAL bool)
+	SwitchToConsensus(ctx context.Context, state sm.State, skipWAL bool, behind bool)
 }
 
 // Reactor handles long-term catchup syncing.
@@ -292,12 +303,18 @@ func (r *Reactor) requestRoutine(ctx context.Context, p2pClient *client.Client)
 // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
 func (r *Reactor) poolRoutine(ctx context.Context, stateSynced bool) {
 	caughtUp := r.synchronizer.WaitForSync(ctx)
+	state := r.executor.State()
+	// Read before the synchronizer stops, and only as evidence that the node is
+	// behind: a peer claiming a height above ours is the one thing that says so.
+	// Where no peer claims one - a solo validator, a network with no peers at all
+	// - the node is not held back at all.
+	behind := !caughtUp && r.synchronizer.MaxPeerHeight() > state.LastBlockHeight
 	r.synchronizer.Stop()
 	r.blockSyncFlag.Store(false)
 	if r.consReactor != nil {
 		// caughtUp is what WaitForSync actually decided on, rather than a second
 		// IsCaughtUp call that races the synchronizer we just stopped
-		r.consReactor.SwitchToConsensus(ctx, r.executor.State(), caughtUp || stateSynced)
+		r.consReactor.SwitchToConsensus(ctx, state, caughtUp || stateSynced, behind)
 	}
 }
 

internal/blocksync/synchronizer.go

@@ -409,8 +409,9 @@ func (s *Synchronizer) IsCaughtUp() bool {
 // while it is still thousands of blocks behind stays behind. As long as some
 // peer holds the block we are waiting for there is something to retry, so keep
 // going and say so loudly. Give up on the stall only once no peer can serve that
-// block, or once the stall has outlasted maxSyncStall, so that a wedged
-// synchronizer can still hand over rather than blocking forever.
+// block, or once the stall has outlasted maxSyncStall within maxCatchupGap of
+// the tip, so that a wedged synchronizer can still hand over rather than
+// blocking forever without handing over a node consensus cannot catch up.
 func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 	ticker := s.clock.NewTicker(switchToConsensusIntervalSeconds * time.Second)
 	defer ticker.Stop()
@@ -422,11 +423,8 @@ func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 			if s.IsCaughtUp() {
 				return true
 			}
-			height, stalledFor, servable := s.stallSnapshot()
-			// read separately because nothing is decided on it: it only gives the
-			// log lines below the number an operator compares against
-			maxPeerHeight := s.MaxPeerHeight()
-			switch stallVerdictFor(servable, stalledFor) {
+			height, stalledFor, servable, maxPeerHeight := s.stallSnapshot()
+			switch stallVerdictFor(servable, maxPeerHeight-height, stalledFor) {
 			case stopNothingToFetch:
 				if maxPeerHeight > height {
 					// Peers claim to be ahead yet none of them holds the block we
@@ -455,7 +453,7 @@ func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 					"height", height,
 					"max_peer_height", maxPeerHeight,
 					"stalled_for", stalledFor,
-					"giving_up_in", maxSyncStall-stalledFor,
+					"behind_by", maxPeerHeight-height,
 				)
 				continue
 			}
@@ -471,15 +469,16 @@ func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 
 // stallSnapshot reads everything the stall verdict is formed from as one
 // observation: the height block sync is waiting for, how long it has been
-// waiting for it, and whether any peer can serve it.
+// waiting for it, whether any peer can serve it, and the highest height any peer
+// claims.
 //
 // Blocks are applied in order, so the current height is the only one that can
-// move us forward, and whether a peer can serve that one is what decides
-// whether waiting is worth anything. The highest height anyone claims decides
-// nothing: a peer whose blocks start above us has nothing we can use however
-// high it claims to be.
+// move us forward, and whether a peer can serve that one is what decides whether
+// waiting is worth anything. The highest height anyone claims cannot decide that
+// - a peer whose blocks start above us has nothing we can use however high it
+// claims to be - it only says how far there is left to go.
 //
-// The three are read under one lock because advance() stamps the height and the
+// They are read under one lock because advance() stamps the height and the
 // advance time together under that same lock. A block applied concurrently
 // therefore lands either wholly inside the snapshot or wholly outside it, and
 // the verdict can never pair a height with a staleness or a servability
@@ -488,10 +487,10 @@ func (s *Synchronizer) WaitForSync(ctx context.Context) (caughtUp bool) {
 // fetch while the height we had by then moved on to is served. Ending block
 // sync is a one-way door, so a stop assembled from two inconsistent readings
 // leaves the node in consensus catch-up it cannot leave.
-func (s *Synchronizer) stallSnapshot() (height int64, stalledFor time.Duration, servable bool) {
+func (s *Synchronizer) stallSnapshot() (height int64, stalledFor time.Duration, servable bool, maxPeerHeight int64) {
 	s.mtx.RLock()
 	defer s.mtx.RUnlock()
-	return s.height, s.clock.Since(s.lastAdvance), s.peerStore.HasPeerForHeight(s.height)
+	return s.height, s.clock.Since(s.lastAdvance), s.peerStore.HasPeerForHeight(s.height), s.peerStore.MaxHeight()
 }
 
 // stallVerdict says what a lack of progress in block sync means.
@@ -507,26 +506,29 @@ const (
 )
 
 // stallVerdictFor decides what to do when block sync has made no progress for
-// stalledFor, given whether any peer holds the block we are waiting for.
+// stalledFor, given whether any peer holds the block we are waiting for and how
+// many blocks behind the highest height any peer claims we are.
 //
 // Waiting on a block someone has is a reason to keep retrying, not to stop:
 // handing over to consensus is effectively irreversible, so stopping while
 // behind leaves the node grinding through consensus catch-up instead. Only a
-// stall on a block nobody has, or one long enough to look like a wedge, ends
-// block sync.
+// stall on a block nobody has, or one long enough to look like a wedge with the
+// tip within consensus catch-up's reach, ends block sync.
 //
 // servable is judged from what peers advertise about themselves, so it cannot
 // distinguish a peer that has the block from one that says it does and never
 // answers. maxSyncStall stays as the wall-clock backstop for that case, and for
-// a synchronizer wedged on our own side, where peers are willing and able and
-// no block arrives anyway.
-func stallVerdictFor(servable bool, stalledFor time.Duration) stallVerdict {
+// a synchronizer wedged on our own side, where peers are willing and able and no
+// block arrives anyway - but only within maxCatchupGap of the tip. Further back
+// the backstop buys nothing and costs what dashpay/tenderdash#1413 describes, so
+// block sync keeps retrying and says so every interval.
+func stallVerdictFor(servable bool, behindBy int64, stalledFor time.Duration) stallVerdict {
 	switch {
 	case stalledFor <= syncTimeout:
 		return keepSyncing
 	case !servable:
 		return stopNothingToFetch
-	case stalledFor > maxSyncStall:
+	case stalledFor > maxSyncStall && behindBy <= maxCatchupGap:
 		return stopStalledTooLong
 	default:
 		return keepSyncing