diff --git a/CHANGELOG.md b/CHANGELOG.md index 47c9a56e5..f2d419082 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/blocksync/continuous.go b/blocksync/continuous.go new file mode 100644 index 000000000..236662234 --- /dev/null +++ b/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 } +} diff --git a/blocksync/metrics.gen.go b/blocksync/metrics.gen.go index ae1b48cb3..8d02b45e1 100644 --- a/blocksync/metrics.gen.go +++ b/blocksync/metrics.gen.go @@ -44,6 +44,18 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics { Name: "latest_block_height", Help: "The height of the latest block.", }, labels).With(labelsAndValues...), + MaxPeerHeight: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: MetricsSubsystem, + Name: "max_peer_height", + Help: "The highest block height reported by connected peers.", + }, labels).With(labelsAndValues...), + ContinuousMode: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: MetricsSubsystem, + Name: "continuous_mode", + Help: "Whether continuous block sync mode is configured.", + }, labels).With(labelsAndValues...), AlreadyIncludedBlocks: prometheus.NewCounterFrom(stdprometheus.CounterOpts{ Namespace: namespace, Subsystem: MetricsSubsystem, @@ -74,6 +86,8 @@ func NopMetrics() *Metrics { TotalTxs: discard.NewGauge(), BlockSizeBytes: discard.NewGauge(), LatestBlockHeight: discard.NewGauge(), + MaxPeerHeight: discard.NewGauge(), + ContinuousMode: discard.NewGauge(), AlreadyIncludedBlocks: discard.NewCounter(), IngestedBlocks: discard.NewCounter(), IngestedBlockDuration: discard.NewHistogram(), diff --git a/blocksync/metrics.go b/blocksync/metrics.go index f39852378..d13e6e620 100644 --- a/blocksync/metrics.go +++ b/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"` diff --git a/blocksync/reactor.go b/blocksync/reactor.go index 04cdf93ed..1ce150745 100644 --- a/blocksync/reactor.go +++ b/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 } diff --git a/blocksync/reactor_test.go b/blocksync/reactor_test.go index a6af9bdf7..824608acf 100644 --- a/blocksync/reactor_test.go +++ b/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)) +} diff --git a/config/config.go b/config/config.go index 72521096a..d5f2974fe 100644 --- a/config/config.go +++ b/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 } //----------------------------------------------------------------------------- diff --git a/config/config_test.go b/config/config_test.go index a30958c6b..0ba55447c 100644 --- a/config/config_test.go +++ b/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") +} diff --git a/config/toml.go b/config/toml.go index ee4732deb..a2fad4de4 100644 --- a/config/toml.go +++ b/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 ### ####################################################### diff --git a/config/toml_test.go b/config/toml_test.go index e94cf7ac4..c79171cb8 100644 --- a/config/toml_test.go +++ b/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) diff --git a/mempool/reactor.go b/mempool/reactor.go index c5951dc49..fb5c75251 100644 --- a/mempool/reactor.go +++ b/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 diff --git a/mempool/reactor_lifecycle_test.go b/mempool/reactor_lifecycle_test.go new file mode 100644 index 000000000..27e7dae31 --- /dev/null +++ b/mempool/reactor_lifecycle_test.go @@ -0,0 +1,143 @@ +package mempool + +import ( + "testing" + "time" + + "github.com/go-kit/kit/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cometbft/cometbft/abci/example/kvstore" + cfg "github.com/cometbft/cometbft/config" + "github.com/cometbft/cometbft/p2p" + "github.com/cometbft/cometbft/p2p/mock" + "github.com/cometbft/cometbft/proxy" + "github.com/cometbft/cometbft/types" +) + +type outboundConnectionsObserver struct { + metrics.Gauge + changes chan float64 +} + +func (g outboundConnectionsObserver) Add(delta float64) { + g.Gauge.Add(delta) + g.changes <- delta +} + +func TestReactorDisconnectWhileSyncing(t *testing.T) { + config := cfg.TestConfig() + mp, cleanup := newMempoolWithApp(proxy.NewLocalClientCreator(kvstore.NewInMemoryApplication())) + t.Cleanup(cleanup) + changes := make(chan float64, 2) + mp.metrics.ActiveOutboundConnections = outboundConnectionsObserver{ + Gauge: mp.metrics.ActiveOutboundConnections, + changes: changes, + } + waitForChange := func(want float64) { + t.Helper() + select { + case delta := <-changes: + require.Equal(t, want, delta) + case <-time.After(time.Second): + t.Fatal("broadcast routine did not update the active connection count") + } + } + reactor := NewReactor(config.Mempool, mp, true) + sw := p2p.NewSwitch(config.P2P, nil) + sw.AddReactor("MEMPOOL", reactor) + require.NoError(t, reactor.Start()) + t.Cleanup(func() { assert.NoError(t, reactor.Stop()) }) + + // Keep sync pending across reconnects, as in continuous block sync. + for i := 0; i < 20; i++ { + peer := mock.NewPeer(nil) + peer.Set(types.PeerStateKey, peerState{1}) + reactor.InitPeer(peer) + reactor.AddPeer(peer) + waitForChange(1) + require.NoError(t, peer.Stop()) + reactor.RemovePeer(peer, nil) + waitForChange(-1) + require.True(t, reactor.WaitSync()) + } +} + +func TestBroadcastTxRoutineStopsWhileWaiting(t *testing.T) { + for _, waitSync := range []bool{true, false} { + stage := "peer_state" + if waitSync { + stage = "sync" + } + for _, stopPeer := range []bool{true, false} { + stop := "reactor" + if stopPeer { + stop = "peer" + } + t.Run(stage+"/"+stop, func(t *testing.T) { + reactor := NewReactor(cfg.TestConfig().Mempool, nil, waitSync) + require.NoError(t, reactor.Start()) + t.Cleanup(func() { + if reactor.IsRunning() { + assert.NoError(t, reactor.Stop()) + } + }) + peer := mock.NewPeer(nil) + if waitSync { + peer.Set(types.PeerStateKey, peerState{1}) + } + t.Cleanup(func() { + if peer.IsRunning() { + assert.NoError(t, peer.Stop()) + } + }) + done := make(chan struct{}) + go func() { + reactor.broadcastTxRoutine(peer) + close(done) + }() + require.Never(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, 50*time.Millisecond, time.Millisecond) + if stopPeer { + require.NoError(t, peer.Stop()) + } else { + require.NoError(t, reactor.Stop()) + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("broadcast routine did not exit") + } + }) + } + } +} + +func TestReactorBroadcastAfterSync(t *testing.T) { + reactors, switches := makeAndConnectReactorsWithWaitSync(cfg.TestConfig(), 2, true) + t.Cleanup(func() { + for _, sw := range switches { + assert.NoError(t, sw.Stop()) + } + }) + for _, reactor := range reactors { + for _, peer := range reactor.Switch.Peers().Copy() { + peer.Set(types.PeerStateKey, peerState{1}) + } + } + txs := addRandomTxs(t, reactors[0].mempool, 3, UnknownPeerID) + require.Never(t, func() bool { return reactors[1].mempool.Size() != 0 }, + 50*time.Millisecond, time.Millisecond) + for _, reactor := range reactors { + reactor.EnableInOutTxs() + reactor.EnableInOutTxs() + } + waitForTxsOnReactors(t, txs, reactors) +} diff --git a/mempool/reactor_test.go b/mempool/reactor_test.go index 5d64cf8c0..5bc8d1392 100644 --- a/mempool/reactor_test.go +++ b/mempool/reactor_test.go @@ -244,6 +244,10 @@ func mempoolLogger() log.Logger { // connect N mempool reactors through N switches func makeAndConnectReactors(config *cfg.Config, n int) ([]*Reactor, []*p2p.Switch) { + return makeAndConnectReactorsWithWaitSync(config, n, false) +} + +func makeAndConnectReactorsWithWaitSync(config *cfg.Config, n int, waitSync bool) ([]*Reactor, []*p2p.Switch) { reactors := make([]*Reactor, n) logger := mempoolLogger() for i := 0; i < n; i++ { @@ -252,7 +256,7 @@ func makeAndConnectReactors(config *cfg.Config, n int) ([]*Reactor, []*p2p.Switc mempool, cleanup := newMempoolWithApp(cc) defer cleanup() - reactors[i] = NewReactor(config.Mempool, mempool, false) // so we dont start the consensus states + reactors[i] = NewReactor(config.Mempool, mempool, waitSync) reactors[i].SetLogger(logger.With("validator", i)) } diff --git a/node/setup.go b/node/setup.go index a6d7f5a8b..a2f4453dc 100644 --- a/node/setup.go +++ b/node/setup.go @@ -316,6 +316,7 @@ func createBlocksyncReactor( localAddr, offlineStateSyncHeight, metrics, + blocksync.ReactorContinuousMode(config.BlockSync.Mode == cfg.BlockSyncModeContinuous), ) bcReactor.SetLogger(logger.With("module", "blocksync"))