feat(blocksync): preserve continuous mode for Cronos v1.8

CHANGELOG.md

@@ -6,6 +6,8 @@
 
 ### BUG FIXES
 
+- `[mempool]` stop transaction broadcast routines when peers disconnect during sync
+  or when peers or the reactor stop while waiting for peer state.
 - `[blocksync]` tolerate late BlockResponse from honest peers after switching to consensus
   ([\#5959](https://github.com/cometbft/cometbft/pull/5959))
 - `[consensus]` correct loop condition in checkDoubleSigningRisk for height=1
@@ -44,6 +46,9 @@
 
 ### FEATURES
 
+- `[blocksync]` Preserve opt-in continuous sync for non-validator Cronos v1.8 nodes;
+  reject simultaneous adaptive sync and retain peer-height metrics.
+
 - `[config]` Add EventBusBufferCapacity setting.
   ([\#5849](https://github.com/cometbft/cometbft/pull/5849))
 

blocksync/continuous.go

@@ -0,0 +1,9 @@
+package blocksync
+
+// ReactorOption configures a block sync reactor.
+type ReactorOption func(*Reactor)
+
+// ReactorContinuousMode keeps non-validator nodes in block sync after catching up.
+func ReactorContinuousMode(enabled bool) ReactorOption {
+	return func(r *Reactor) { r.continuous = enabled }
+}

blocksync/metrics.go

@@ -25,6 +25,10 @@ type Metrics struct {
 	BlockSizeBytes metrics.Gauge
 	// The height of the latest block.
 	LatestBlockHeight metrics.Gauge
+	// The highest block height reported by connected peers.
+	MaxPeerHeight metrics.Gauge
+	// Whether continuous block sync mode is configured.
+	ContinuousMode metrics.Gauge
 
 	// AlreadyIncludedBlocks blocks that were already included in the chain
 	AlreadyIncludedBlocks metrics.Counter `metrics_name:"already_included_blocks"`

blocksync/reactor.go

@@ -74,6 +74,7 @@ type Reactor struct {
 
 	blockExec     *sm.BlockExecutor
 	store         sm.BlockStore
+	continuous    bool
 	pool          *BlockPool
 	localAddr     crypto.Address
 	poolRoutineWg sync.WaitGroup
@@ -100,6 +101,7 @@ func NewReactor(
 	localAddr crypto.Address,
 	offlineStateSyncHeight int64,
 	metrics *Metrics,
+	options ...ReactorOption,
 ) *Reactor {
 	storeHeight := store.Height()
 	if storeHeight == 0 {
@@ -156,6 +158,12 @@ func NewReactor(
 		intervalStatusUpdate:      intervalStatusUpdate,
 	}
 
+	for _, option := range options {
+		option(r)
+	}
+	if r.continuous {
+		r.intervalStatusUpdate = time.Second
+	}
 	r.BaseReactor = *p2p.NewBaseReactor("Blocksync", r)
 
 	return r
@@ -169,6 +177,15 @@ func (r *Reactor) SetLogger(l log.Logger) {
 
 // OnStart implements service.Service.
 func (r *Reactor) OnStart() error {
+	if r.continuous && r.localNodeHasVotingPower(r.initialState) {
+		return errors.New("continuous block sync mode cannot be used by a validator")
+	}
+	if r.continuous && r.adaptiveSyncEnabled {
+		return errors.New("continuous block sync mode cannot be combined with adaptive sync")
+	}
+	if r.continuous {
+		r.metrics.ContinuousMode.Set(1)
+	}
 	// noop
 	if !r.enabled.Load() {
 		return nil
@@ -220,6 +237,9 @@ func (r *Reactor) runPool(stateSynced bool) error {
 
 // Enable is called by the state sync reactor when switching to block sync.
 func (r *Reactor) Enable(state sm.State) error {
+	if r.continuous && r.localNodeHasVotingPower(state) {
+		return errors.New("continuous block sync mode cannot be used by a validator")
+	}
 	if !r.enabled.CompareAndSwap(false, true) {
 		return ErrAlreadyEnabled
 	}
@@ -277,6 +297,7 @@ func (r *Reactor) AddPeer(peer p2p.Peer) {
 // RemovePeer implements Reactor by removing peer from the pool.
 func (r *Reactor) RemovePeer(peer p2p.Peer, _ any) {
 	r.pool.RemovePeer(peer.ID())
+	r.metrics.MaxPeerHeight.Set(float64(r.pool.MaxPeerHeight()))
 }
 
 // respondToPeer loads a block and sends it to the requesting peer,
@@ -444,6 +465,7 @@ func (r *Reactor) Receive(e p2p.Envelope) {
 	case *bcproto.StatusResponse:
 		// Got a peer status. Unverified.
 		r.pool.SetPeerRange(e.Src.ID(), msg.Base, msg.Height)
+		r.metrics.MaxPeerHeight.Set(float64(r.pool.MaxPeerHeight()))
 	case *bcproto.NoBlockResponse:
 		r.Logger.Debug("Peer does not have requested block", "peer", e.Src, "height", msg.Height)
 		r.pool.RedoRequestFrom(msg.Height, e.Src.ID())
@@ -461,6 +483,21 @@ func (r *Reactor) localNodeBlocksTheChain(state sm.State) bool {
 	return val.VotingPower >= total/3
 }
 
+func (r *Reactor) localNodeHasVotingPower(state sm.State) bool {
+	if len(r.localAddr) == 0 {
+		return false
+	}
+	_, val := state.Validators.GetByAddress(r.localAddr)
+	return val != nil && val.VotingPower > 0
+}
+
+func (r *Reactor) shouldSwitchToConsensus(state sm.State, caughtUp bool) bool {
+	if r.localNodeBlocksTheChain(state) {
+		return true
+	}
+	return caughtUp && (!r.continuous || r.localNodeHasVotingPower(state))
+}
+
 // Handle messages from the poolReactor telling the reactor what to do.
 // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
 func (r *Reactor) poolRoutine(stateSynced bool) {
@@ -546,8 +583,9 @@ FOR_LOOP:
 				continue FOR_LOOP
 			}
 
-			// keep syncing
-			if !r.pool.IsCaughtUp() && !r.localNodeBlocksTheChain(state) {
+			// Keep syncing in continuous mode, including after catching up.
+			r.metrics.MaxPeerHeight.Set(float64(r.pool.MaxPeerHeight()))
+			if !r.shouldSwitchToConsensus(state, r.pool.IsCaughtUp()) {
 				continue FOR_LOOP
 			}
 

blocksync/reactor_test.go

@@ -67,6 +67,15 @@ type ReactorPair struct {
 	app     proxy.AppConns
 }
 
+type advertisedHeightBlockStore struct {
+	sm.BlockStore
+	height atomic.Int64
+}
+
+func (s *advertisedHeightBlockStore) Height() int64 {
+	return s.height.Load()
+}
+
 type reactorOpts struct {
 	corruptedBlock          int64
 	allAbsentExtCommitBlock int64
@@ -1006,3 +1015,71 @@ func TestPeerNotDisconnectedOnLateBlockResponseAfterConsensusSwitch(t *testing.T
 	}, 200*time.Millisecond, 5*time.Millisecond,
 		"serving peer was incorrectly disconnected by a late in-flight BlockResponse")
 }
+
+func TestContinuousBlockSyncRemainsActiveAndResumes(t *testing.T) {
+	config = test.ResetTestRoot("blocksync_reactor_continuous_test")
+	defer os.RemoveAll(config.RootDir)
+	genDoc, privVals := genesisDocWithValsPowers([]int64{30})
+
+	const initialAdvertisedHeight = int64(20)
+	const finalAdvertisedHeight = int64(40)
+	logger := log.NewNopLogger()
+
+	reactorPairs := []ReactorPair{
+		newReactor(t, logger, genDoc, privVals, finalAdvertisedHeight),
+		newReactor(t, logger, genDoc, privVals, 0),
+	}
+	peerStore := &advertisedHeightBlockStore{BlockStore: reactorPairs[0].reactor.store}
+	peerStore.height.Store(initialAdvertisedHeight)
+	reactorPairs[0].reactor.store = peerStore
+	reactorPairs[1].reactor.continuous = true
+	reactorPairs[1].reactor.intervalSwitchToConsensus = 50 * time.Millisecond
+
+	p2p.MakeConnectedSwitches(config.P2P, 2, func(i int, s *p2p.Switch) *p2p.Switch {
+		s.AddReactor("BLOCKSYNC", reactorPairs[i].reactor)
+		return s
+	}, p2p.Connect2Switches)
+
+	defer func() {
+		for _, r := range reactorPairs {
+			require.NoError(t, r.reactor.Stop())
+			require.NoError(t, r.app.Stop())
+		}
+	}()
+
+	require.Eventually(t, func() bool {
+		return reactorPairs[1].reactor.store.Height() >= initialAdvertisedHeight-1
+	}, 30*time.Second, 20*time.Millisecond)
+	require.Eventually(t, reactorPairs[1].reactor.pool.IsCaughtUp, 5*time.Second, 20*time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
+	require.True(t, reactorPairs[1].reactor.pool.IsRunning())
+
+	peerStore.height.Store(finalAdvertisedHeight)
+
+	require.Eventually(t, func() bool {
+		return reactorPairs[1].reactor.store.Height() >= finalAdvertisedHeight-1
+	}, 30*time.Second, 20*time.Millisecond)
+	require.True(t, reactorPairs[1].reactor.pool.IsRunning())
+}
+
+func TestContinuousBlockSyncRejectsValidator(t *testing.T) {
+	config = test.ResetTestRoot("blocksync_reactor_continuous_validator_test")
+	defer os.RemoveAll(config.RootDir)
+	genDoc, privVals := genesisDocWithValsPowers([]int64{30})
+
+	reactorPair := newReactor(t, log.TestingLogger(), genDoc, privVals, 0)
+	defer func() {
+		require.NoError(t, reactorPair.app.Stop())
+	}()
+
+	pubKey, err := privVals[0].GetPubKey()
+	require.NoError(t, err)
+	reactorPair.reactor.localAddr = pubKey.Address()
+	reactorPair.reactor.continuous = true
+
+	require.ErrorContains(t, reactorPair.reactor.OnStart(), "cannot be used by a validator")
+	require.True(t, reactorPair.reactor.shouldSwitchToConsensus(reactorPair.reactor.initialState, true))
+
+	reactorPair.reactor.localAddr = nil
+	require.False(t, reactorPair.reactor.shouldSwitchToConsensus(reactorPair.reactor.initialState, true))
+}

config/config.go

@@ -1207,8 +1207,14 @@ func (cfg *StateSyncConfig) ValidateBasic() error {
 //-----------------------------------------------------------------------------
 // BlockSyncConfig
 
+const (
+	BlockSyncModeAuto       = "auto"
+	BlockSyncModeContinuous = "continuous"
+)
+
 // BlockSyncConfig defines the configuration for the CometBFT block sync service
 type BlockSyncConfig struct {
+	Mode         string `mapstructure:"mode"`
 	Version      string `mapstructure:"version"`
 	AdaptiveSync bool   `mapstructure:"adaptive_sync"`
 }
@@ -1218,6 +1224,7 @@ func DefaultBlockSyncConfig() *BlockSyncConfig {
 	return &BlockSyncConfig{
 		Version:      "v0",
 		AdaptiveSync: false,
+		Mode:         BlockSyncModeAuto,
 	}
 }
 
@@ -1230,12 +1237,21 @@ func TestBlockSyncConfig() *BlockSyncConfig {
 func (cfg *BlockSyncConfig) ValidateBasic() error {
 	switch cfg.Version {
 	case v0:
-		return nil
 	case v1, v2:
 		return ErrDeprecatedBlocksyncVersion{Version: cfg.Version, Allowed: []string{v0}}
 	default:
 		return ErrUnknownBlocksyncVersion{cfg.Version}
 	}
+	switch cfg.Mode {
+	case "", BlockSyncModeAuto:
+	case BlockSyncModeContinuous:
+		if cfg.AdaptiveSync {
+			return errors.New("continuous block sync mode cannot be combined with adaptive sync")
+		}
+	default:
+		return fmt.Errorf("unknown blocksync mode %s", cfg.Mode)
+	}
+	return nil
 }
 
 //-----------------------------------------------------------------------------

config/config_test.go

@@ -321,6 +321,14 @@ func TestStateSyncConfigValidateBasic(t *testing.T) {
 func TestBlockSyncConfigValidateBasic(t *testing.T) {
 	cfg := config.TestBlockSyncConfig()
 	assert.NoError(t, cfg.ValidateBasic())
+	assert.Equal(t, config.BlockSyncModeAuto, cfg.Mode)
+
+	cfg.Mode = config.BlockSyncModeContinuous
+	assert.NoError(t, cfg.ValidateBasic())
+
+	cfg.Mode = "invalid"
+	assert.Error(t, cfg.ValidateBasic())
+	cfg.Mode = config.BlockSyncModeAuto
 
 	// tamper with version
 	cfg.Version = "v1"
@@ -383,3 +391,10 @@ func TestInstrumentationConfigValidateBasic(t *testing.T) {
 	cfg.MaxOpenConnections = -1
 	assert.Error(t, cfg.ValidateBasic())
 }
+
+func TestContinuousModeRejectsAdaptiveSync(t *testing.T) {
+	cfg := config.DefaultBlockSyncConfig()
+	cfg.Mode = config.BlockSyncModeContinuous
+	cfg.AdaptiveSync = true
+	require.ErrorContains(t, cfg.ValidateBasic(), "cannot be combined with adaptive sync")
+}

config/toml.go

@@ -571,6 +571,9 @@ version = "{{ .BlockSync.Version }}"
 # Run both BLOCKSYNC and CONSENSUS for improved liveness, connectivity, and performance.
 adaptive_sync = {{ .BlockSync.AdaptiveSync }}
 
+# "auto" switches to consensus after catching up; "continuous" is for non-validator nodes.
+mode = "{{ .BlockSync.Mode }}"
+
 #######################################################
 ###         Consensus Configuration Options         ###
 #######################################################

config/toml_test.go

@@ -76,6 +76,7 @@ func assertValidConfig(t *testing.T, configFile string) {
 		"propose",
 		"max",
 		"genesis",
+		"mode = \"auto\"",
 	}
 	for _, e := range elems {
 		assert.Contains(t, configFile, e)

mempool/reactor.go

@@ -230,6 +230,8 @@ func (memR *Reactor) broadcastTxRoutine(peer p2p.Peer) {
 		select {
 		case <-memR.waitSyncCh:
 			// EnableInOutTxs() has set WaitSync() to false.
+		case <-peer.Quit():
+			return
 		case <-memR.Quit():
 			return
 		}
@@ -242,6 +244,9 @@ func (memR *Reactor) broadcastTxRoutine(peer p2p.Peer) {
 	// reactor. We should wait a few milliseconds and retry. We assume the pointer to the state is
 	// set once and never unset.
 	for {
+		if !memR.IsRunning() || !peer.IsRunning() {
+			return
+		}
 		if ps, ok := peer.Get(types.PeerStateKey).(PeerState); ok {
 			peerState = ps
 			break