feat: optionally skip processProposal while blocksyncing (#2658)

first part of #2657

adds a config and flag to the blocksync config that enables a user to
select if they call ProcessProposal while blocksyncing or not. This can
significantly increase syncing rate, especially with large blocks. When
state syncing, users also suffer from a minimal security decrease.

when verifying is disabled, we are only comparing the partsetheader, not
the data root.

Author: evan-forbes

blocksync/reactor.go

@@ -42,6 +42,16 @@ type consensusReactor interface {
 	SwitchToConsensus(state sm.State, skipWAL bool)
 }
 
+// ReactorOption defines a function argument for Reactor.
+type ReactorOption func(*Reactor)
+
+// ReactorVerifyData sets the verifyData field of the reactor.
+func ReactorVerifyData(verifyData bool) ReactorOption {
+	return func(bcR *Reactor) {
+		bcR.verifyData = verifyData
+	}
+}
+
 type peerError struct {
 	err    error
 	peerID p2p.ID
@@ -63,6 +73,7 @@ type Reactor struct {
 	pool          *BlockPool
 	traceClient   trace.Tracer
 	blockSync     bool
+	verifyData    bool
 	localAddr     crypto.Address
 	poolRoutineWg sync.WaitGroup
 
@@ -76,14 +87,14 @@ type Reactor struct {
 
 // NewReactor returns new reactor instance.
 func NewReactor(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore,
-	blockSync bool, metrics *Metrics, offlineStateSyncHeight int64,
+	blockSync bool, metrics *Metrics, offlineStateSyncHeight int64, options ...ReactorOption,
 ) *Reactor {
-	return NewReactorWithAddr(state, blockExec, store, blockSync, nil, metrics, offlineStateSyncHeight, trace.NoOpTracer())
+	return NewReactorWithAddr(state, blockExec, store, blockSync, nil, metrics, offlineStateSyncHeight, trace.NoOpTracer(), options...)
 }
 
 // Function added to keep existing API.
 func NewReactorWithAddr(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore,
-	blockSync bool, localAddr crypto.Address, metrics *Metrics, offlineStateSyncHeight int64, traceClient trace.Tracer,
+	blockSync bool, localAddr crypto.Address, metrics *Metrics, offlineStateSyncHeight int64, traceClient trace.Tracer, options ...ReactorOption,
 ) *Reactor {
 
 	storeHeight := store.Height()
@@ -120,13 +131,19 @@ func NewReactorWithAddr(state sm.State, blockExec *sm.BlockExecutor, store *stor
 		store:        store,
 		pool:         pool,
 		blockSync:    blockSync,
+		verifyData:   true,
 		localAddr:    localAddr,
 		requestsCh:   requestsCh,
 		errorsCh:     errorsCh,
 		metrics:      metrics,
 		traceClient:  traceClient,
 	}
 	bcR.BaseReactor = *p2p.NewBaseReactor("BlockSync", bcR, p2p.WithIncomingQueueSize(ReactorIncomingMessageQueueSize))
+
+	for _, option := range options {
+		option(bcR)
+	}
+
 	return bcR
 }
 
@@ -514,7 +531,7 @@ FOR_LOOP:
 				err = bcR.blockExec.ValidateBlock(state, first)
 			}
 
-			if err == nil {
+			if err == nil && bcR.verifyData {
 				var stateMachineValid bool
 				// Block sync doesn't check that the `Data` in a block is valid.
 				// Since celestia-core can't determine if the `Data` in a block

blocksync/reactor_test.go

@@ -67,6 +67,18 @@ func newReactor(
 	privVals []types.PrivValidator,
 	maxBlockHeight int64,
 	incorrectData ...int64,
+) ReactorPair {
+	return newReactorWithConfig(t, logger, genDoc, privVals, maxBlockHeight, true, incorrectData...)
+}
+
+func newReactorWithConfig(
+	t *testing.T,
+	logger log.Logger,
+	genDoc *types.GenesisDoc,
+	privVals []types.PrivValidator,
+	maxBlockHeight int64,
+	verifyData bool,
+	incorrectData ...int64,
 ) ReactorPair {
 	if len(privVals) != 1 {
 		panic("only support one validator")
@@ -177,12 +189,39 @@ func newReactor(
 		}
 	}
 
-	bcReactor := NewByzantineReactor(incorrectBlock, NewReactor(state.Copy(), blockExec, blockStore, blockSync, NopMetrics(), 0))
+	bcReactor := NewByzantineReactor(incorrectBlock, NewReactor(state.Copy(), blockExec, blockStore, blockSync, NopMetrics(), 0, ReactorVerifyData(verifyData)))
 	bcReactor.SetLogger(logger.With("module", "blocksync"))
 
 	return ReactorPair{bcReactor, proxyApp}
 }
 
+func TestVerifyData(t *testing.T) {
+	config = test.ResetTestRoot("blocksync_verify_data_test")
+	defer os.RemoveAll(config.RootDir)
+	genDoc, privVals := randGenesisDoc(1, false, 30)
+
+	maxBlockHeight := int64(10)
+
+	// Case 1: verifyData = true. Normal behavior.
+	// We cannot easily mock ProcessProposal failure here because we are using a real app (NewBaseApplication) via proxy.
+	// But we can observe that it works.
+	// To test failure, we would need to mock the app or the proxy connection.
+	// The current newReactor setup uses a real BaseApplication.
+	// So, passing verifyData=false should also work for normal blocks.
+	// To really test this, we might need a way to make ProcessProposal fail.
+
+	// Let's verify that we can construct reactors with both settings.
+	rp1 := newReactorWithConfig(t, log.TestingLogger(), genDoc, privVals, maxBlockHeight, true)
+	rp2 := newReactorWithConfig(t, log.TestingLogger(), genDoc, privVals, maxBlockHeight, false)
+
+	require.NotNil(t, rp1.reactor)
+	require.NotNil(t, rp2.reactor)
+
+	// We can check the internal field
+	assert.True(t, rp1.reactor.verifyData)
+	assert.False(t, rp2.reactor.verifyData)
+}
+
 func TestNoBlockResponse(t *testing.T) {
 	config = test.ResetTestRoot("blocksync_reactor_test")
 	defer os.RemoveAll(config.RootDir)

config/config.go

@@ -1035,13 +1035,15 @@ func (cfg *StateSyncConfig) ValidateBasic() error {
 
 // BlockSyncConfig (formerly known as FastSync) defines the configuration for the CometBFT block sync service
 type BlockSyncConfig struct {
-	Version string `mapstructure:"version"`
+	Version    string `mapstructure:"version"`
+	VerifyData bool   `mapstructure:"verify_data"`
 }
 
 // DefaultBlockSyncConfig returns a default configuration for the block sync service
 func DefaultBlockSyncConfig() *BlockSyncConfig {
 	return &BlockSyncConfig{
-		Version: "v0",
+		Version:    "v0",
+		VerifyData: true,
 	}
 }
 

config/toml.go

@@ -499,6 +499,11 @@ chunk_fetchers = "{{ .StateSync.ChunkFetchers }}"
 #   1) "v0" - the default block sync implementation
 version = "{{ .BlockSync.Version }}"
 
+# If true, the node will verify the application data in the block via ProcessProposal
+# during block sync. If false, this verification is skipped.
+# Default: true
+verify_data = {{ .BlockSync.VerifyData }}
+
 #######################################################
 ###         Consensus Configuration Options         ###
 #######################################################

node/setup.go

@@ -352,7 +352,7 @@ func createBlocksyncReactor(config *cfg.Config,
 ) (bcReactor p2p.Reactor, err error) {
 	switch config.BlockSync.Version {
 	case "v0":
-		bcReactor = blocksync.NewReactorWithAddr(state.Copy(), blockExec, blockStore, blockSync, localAddr, metrics, offlineStateSyncHeight, tracer)
+		bcReactor = blocksync.NewReactorWithAddr(state.Copy(), blockExec, blockStore, blockSync, localAddr, metrics, offlineStateSyncHeight, tracer, blocksync.ReactorVerifyData(config.BlockSync.VerifyData))
 	case "v1", "v2":
 		return nil, fmt.Errorf("block sync version %s has been deprecated. Please use v0", config.BlockSync.Version)
 	default: