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

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Bug Fixes
 
+- Delay proposals after an incomplete block-sync handover with a bounded catch-up window, base the stall fallback on peers that can serve the needed block, and validate application responses before persisting synced blocks (#1416).
 - Bound stalled block-response cleanup while allowing actively progressing transfers to continue.
 - Authenticate state-sync validator thresholds against locally stored genesis parameters.
 - Reject partially populated validator public-key shares in light blocks.

internal/blocksync/applier.go

@@ -91,17 +91,26 @@ func (e *blockApplier) Apply(ctx context.Context, block *types.Block, commit *ty
 	}
 	verifyTime := time.Since(start)
 
+	// Validate the app response before persisting; save before FinalizeBlock so
+	// crash recovery never finds the block store behind the application.
+	start = time.Now()
+	uncommittedState, err := e.blockExec.ProcessProposal(ctx, block, commit.Round, e.state, true)
+	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 := e.observeSince("save", 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 := e.observeSince("exec", start)
+	execTime := processTime + time.Since(start)
+	e.metrics.ObserveBlockSyncStage("exec", execTime)
 
 	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

@@ -7,18 +7,65 @@ import (
 	"testing"
 	"time"
 
+	dbm "github.com/cometbft/cometbft-db"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	abciclient "github.com/dashpay/tenderdash/abci/client"
+	abci "github.com/dashpay/tenderdash/abci/types"
+	"github.com/dashpay/tenderdash/crypto"
 	"github.com/dashpay/tenderdash/internal/consensus"
+	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/store"
 	"github.com/dashpay/tenderdash/internal/test/factory"
 	"github.com/dashpay/tenderdash/internal/test/metricspy"
 	tmrequire "github.com/dashpay/tenderdash/internal/test/require"
+	"github.com/dashpay/tenderdash/libs/log"
 	"github.com/dashpay/tenderdash/types"
 )
 
+func TestBlockApplierChecksAppResponseBeforeSave(t *testing.T) {
+	ctx := context.Background()
+	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
+	appHash := make([]byte, crypto.DefaultAppHashSize)
+	appHash[0] = 1
+	app := &inconsistentProposalApp{appHash: appHash}
+	client := abciclient.NewLocalClient(log.NewNopLogger(), app)
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+	stateStore := sm.NewStore(dbm.NewMemDB())
+	require.NoError(t, stateStore.Save(initialState))
+	executor := sm.NewBlockExecutor(stateStore, client, nil, sm.EmptyEvidencePool{}, blockStore, nil)
+	applier := newBlockApplier(executor, blockStore, applierWithState(initialState))
+
+	require.Panics(t, func() { _ = applier.Apply(ctx, block, commit) })
+	require.Zero(t, blockStore.Height(), "an inconsistent app response must not advance the block store")
+	require.Equal(t, initialState.LastBlockHeight, applier.State().LastBlockHeight)
+	loaded, err := stateStore.Load()
+	require.NoError(t, err)
+	require.Equal(t, initialState.LastBlockHeight, loaded.LastBlockHeight)
+}
+
+type inconsistentProposalApp struct {
+	abci.BaseApplication
+	appHash []byte
+}
+
+func (app *inconsistentProposalApp) ProcessProposal(
+	_ context.Context, req *abci.RequestProcessProposal,
+) (*abci.ResponseProcessProposal, error) {
+	return &abci.ResponseProcessProposal{
+		Status:    abci.ResponseProcessProposal_ACCEPT,
+		AppHash:   app.appHash,
+		TxResults: factory.ExecTxResults(types.NewTxs(req.Txs)),
+	}, nil
+}
+
 func TestBlockApplierApply(t *testing.T) {
 	ctx := context.Background()
 	mockBlockExec := mocks.NewExecutor(t)
@@ -50,7 +97,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, true).
+					Once().
+					Return(sm.CurrentRoundState{}, nil)
+				mockBlockExec.
+					On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, blockH1ID, blockH1, commitH1).
 					Once().
 					Return(state, nil)
 			},
@@ -76,7 +127,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, true).
+					Once().
+					Return(sm.CurrentRoundState{}, nil)
+				mockBlockExec.
+					On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, blockH1ID, blockH1, commitH1).
 					Once().
 					Return(state, errors.New("eeeeeeeee"))
 			},
@@ -152,6 +207,78 @@ func TestApplyStatsSubMillisecondPreserved(t *testing.T) {
 	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, true).
+		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, true).
+		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)
+}
+
 // TestBlockApplierRecordsStageMetrics checks that a successful apply records
 // one sample per stage of the pipeline, that the verify stage is split into
 // the commit signature check and block validation, and that the idle time
@@ -170,7 +297,9 @@ func TestBlockApplierRecordsStageMetrics(t *testing.T) {
 
 	mockBlockStore.On("SaveBlock", blockH1, mock.Anything, commitH1).Twice()
 	mockBlockExec.On("ValidateBlock", mock.Anything, mock.Anything, blockH1).Twice().Return(nil)
-	mockBlockExec.On("ApplyBlock", mock.Anything, mock.Anything, mock.Anything, blockH1, commitH1).
+	mockBlockExec.On("ProcessProposal", mock.Anything, blockH1, commitH1.Round, initialState, true).
+		Twice().Return(sm.CurrentRoundState{}, nil)
+	mockBlockExec.On("FinalizeBlock", mock.Anything, initialState, sm.CurrentRoundState{}, mock.Anything, blockH1, commitH1).
 		Twice().Return(initialState, nil)
 
 	hist := metricspy.NewHistogram("stage")

internal/blocksync/handover_test.go

@@ -0,0 +1,101 @@
+package blocksync
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+
+	sm "github.com/dashpay/tenderdash/internal/state"
+	"github.com/dashpay/tenderdash/libs/log"
+	"github.com/dashpay/tenderdash/libs/service"
+)
+
+func TestWaitForSyncRetainsObservedTarget(t *testing.T) {
+	clock := clockwork.NewFakeClock()
+	observed := make(chan struct{}, 1)
+	synchronizer := NewSynchronizer(100, nil, nil, WithClock(clock),
+		WithLogger(&handoverObservationLogger{Logger: log.NewNopLogger(), observed: observed}))
+	synchronizer.lastAdvance = clock.Now()
+	synchronizer.AddPeer(newPeerData("peer", 1, 101))
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	result := make(chan int64, 1)
+	go func() {
+		_, target := synchronizer.WaitForSync(ctx)
+		result <- target
+	}()
+	require.NoError(t, clock.BlockUntilContext(ctx, 1))
+	clock.Advance(switchToConsensusIntervalSeconds * time.Second)
+	select {
+	case <-observed:
+	case <-ctx.Done():
+		t.Fatal("synchronizer did not observe its peer")
+	}
+	synchronizer.RemovePeer("peer")
+	clock.Advance(syncTimeout + time.Second)
+	select {
+	case target := <-result:
+		require.Equal(t, int64(101), target, "disconnecting a peer must not erase its observed handover target")
+	case <-ctx.Done():
+		t.Fatal("synchronizer did not finish")
+	}
+}
+
+type handoverObservationLogger struct {
+	log.Logger
+	observed chan<- struct{}
+}
+
+func (l *handoverObservationLogger) Info(string, ...interface{}) {
+	select {
+	case l.observed <- struct{}{}:
+	default:
+	}
+}
+
+func TestHandoverUsesFinalAppliedState(t *testing.T) {
+	for _, applyDuringStop := range []bool{false, true} {
+		t.Run(map[bool]string{false: "one block remains", true: "last block applied during stop"}[applyDuringStop], func(t *testing.T) {
+			applier := newBlockApplier(nil, nil, applierWithState(sm.State{LastBlockHeight: 99}))
+			synchronizer := NewSynchronizer(100, nil, applier)
+			synchronizer.AddPeer(newPeerData("peer", 1, 100))
+			stop := &handoverStopHook{fn: func() {
+				if applyDuringStop {
+					applier.UpdateState(sm.State{LastBlockHeight: 100})
+				}
+			}}
+			synchronizer.BaseService = *service.NewBaseService(log.NewNopLogger(), "handover", stop)
+			serviceCtx, cancelService := context.WithCancel(context.Background())
+			defer cancelService()
+			require.NoError(t, synchronizer.Start(serviceCtx))
+			capture := &handoverCapture{}
+			reactor := &Reactor{executor: applier, synchronizer: synchronizer, consReactor: capture, blockSyncFlag: new(atomic.Bool)}
+			ctx, cancel := context.WithCancel(context.Background())
+			cancel()
+			reactor.poolRoutine(ctx, false)
+			require.True(t, capture.called)
+			require.True(t, capture.skipWAL)
+			require.Equal(t, applier.State().LastBlockHeight, capture.state.LastBlockHeight)
+			require.Equal(t, int64(100), capture.targetHeight)
+		})
+	}
+}
+
+type handoverCapture struct {
+	state           sm.State
+	called, skipWAL bool
+	targetHeight    int64
+}
+
+func (c *handoverCapture) SwitchToConsensus(_ context.Context, state sm.State, skipWAL bool, targetHeight int64) {
+	c.called, c.state, c.skipWAL, c.targetHeight = true, state, skipWAL, targetHeight
+}
+
+type handoverStopHook struct{ fn func() }
+
+func (*handoverStopHook) OnStart(context.Context) error { return nil }
+func (h *handoverStopHook) OnStop()                     { h.fn() }

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, targetHeight int64)
 }
 
 // Reactor handles long-term catchup syncing.
@@ -308,13 +319,15 @@ 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)
+	caughtUp, targetHeight := r.synchronizer.WaitForSync(ctx)
 	r.synchronizer.Stop()
+	// Wait for an application holding the applier lock and use its final state.
+	state := r.executor.State()
 	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, targetHeight)
 	}
 }