statesync: implement p2p state provider (#6807)

node/node.go

@@ -28,7 +28,6 @@ import (
 	"github.com/tendermint/tendermint/libs/service"
 	"github.com/tendermint/tendermint/libs/strings"
 	tmtime "github.com/tendermint/tendermint/libs/time"
-	"github.com/tendermint/tendermint/light"
 	"github.com/tendermint/tendermint/privval"
 	tmgrpc "github.com/tendermint/tendermint/privval/grpc"
 	"github.com/tendermint/tendermint/proxy"
@@ -328,6 +327,8 @@ func makeNode(config *cfg.Config,
 	}
 
 	stateSyncReactor = statesync.NewReactor(
+		genDoc.ChainID,
+		genDoc.InitialHeight,
 		*config.StateSync,
 		stateSyncReactorShim.Logger,
 		proxyApp.Snapshot(),
@@ -335,6 +336,7 @@ func makeNode(config *cfg.Config,
 		channels[statesync.SnapshotChannel],
 		channels[statesync.ChunkChannel],
 		channels[statesync.LightBlockChannel],
+		channels[statesync.ParamsChannel],
 		peerUpdates,
 		stateStore,
 		blockStore,
@@ -671,6 +673,8 @@ func (n *nodeImpl) OnStart() error {
 	}
 
 	// Run state sync
+	// TODO: We shouldn't run state sync if we already have state that has a
+	// LastBlockHeight that is not InitialHeight
 	if n.stateSync {
 		bcR, ok := n.bcReactor.(cs.BlockSyncReactor)
 		if !ok {
@@ -683,17 +687,52 @@ func (n *nodeImpl) OnStart() error {
 			return fmt.Errorf("unable to derive state: %w", err)
 		}
 
-		ssc := n.config.StateSync
-		sp, err := constructStateProvider(ssc, state, n.Logger.With("module", "light"))
-
-		if err != nil {
-			return fmt.Errorf("failed to set up light client state provider: %w", err)
+		// TODO: we may want to move these events within the respective
+		// reactors.
+		// At the beginning of the statesync start, we use the initialHeight as the event height
+		// because of the statesync doesn't have the concreate state height before fetched the snapshot.
+		d := types.EventDataStateSyncStatus{Complete: false, Height: state.InitialHeight}
+		if err := n.eventBus.PublishEventStateSyncStatus(d); err != nil {
+			n.eventBus.Logger.Error("failed to emit the statesync start event", "err", err)
 		}
 
-		if err := startStateSync(n.stateSyncReactor, bcR, n.consensusReactor, sp,
-			ssc, n.config.FastSyncMode, state.InitialHeight, n.eventBus); err != nil {
-			return fmt.Errorf("failed to start state sync: %w", err)
-		}
+		// FIXME: We shouldn't allow state sync to silently error out without
+		// bubbling up the error and gracefully shutting down the rest of the node
+		go func() {
+			n.Logger.Info("starting state sync")
+			state, err := n.stateSyncReactor.Sync(context.TODO())
+			if err != nil {
+				n.Logger.Error("state sync failed", "err", err)
+				return
+			}
+
+			n.consensusReactor.SetStateSyncingMetrics(0)
+
+			d := types.EventDataStateSyncStatus{Complete: true, Height: state.LastBlockHeight}
+			if err := n.eventBus.PublishEventStateSyncStatus(d); err != nil {
+				n.eventBus.Logger.Error("failed to emit the statesync start event", "err", err)
+			}
+
+			// TODO: Some form of orchestrator is needed here between the state
+			// advancing reactors to be able to control which one of the three
+			// is running
+			if n.config.FastSyncMode {
+				// FIXME Very ugly to have these metrics bleed through here.
+				n.consensusReactor.SetBlockSyncingMetrics(1)
+				if err := bcR.SwitchToBlockSync(state); err != nil {
+					n.Logger.Error("failed to switch to block sync", "err", err)
+					return
+				}
+
+				d := types.EventDataBlockSyncStatus{Complete: false, Height: state.LastBlockHeight}
+				if err := n.eventBus.PublishEventBlockSyncStatus(d); err != nil {
+					n.eventBus.Logger.Error("failed to emit the block sync starting event", "err", err)
+				}
+
+			} else {
+				n.consensusReactor.SwitchToConsensus(state, true)
+			}
+		}()
 	}
 
 	return nil
@@ -978,67 +1017,6 @@ func (n *nodeImpl) NodeInfo() types.NodeInfo {
 	return n.nodeInfo
 }
 
-// startStateSync starts an asynchronous state sync process, then switches to block sync mode.
-func startStateSync(
-	ssR statesync.SyncReactor,
-	bcR cs.BlockSyncReactor,
-	conR cs.ConsSyncReactor,
-	sp statesync.StateProvider,
-	config *cfg.StateSyncConfig,
-	blockSync bool,
-	stateInitHeight int64,
-	eb *types.EventBus,
-) error {
-	stateSyncLogger := eb.Logger.With("module", "statesync")
-
-	stateSyncLogger.Info("starting state sync...")
-
-	// at the beginning of the statesync start, we use the initialHeight as the event height
-	// because of the statesync doesn't have the concreate state height before fetched the snapshot.
-	d := types.EventDataStateSyncStatus{Complete: false, Height: stateInitHeight}
-	if err := eb.PublishEventStateSyncStatus(d); err != nil {
-		stateSyncLogger.Error("failed to emit the statesync start event", "err", err)
-	}
-
-	go func() {
-		state, err := ssR.Sync(context.TODO(), sp, config.DiscoveryTime)
-		if err != nil {
-			stateSyncLogger.Error("state sync failed", "err", err)
-			return
-		}
-
-		if err := ssR.Backfill(state); err != nil {
-			stateSyncLogger.Error("backfill failed; node has insufficient history to verify all evidence;"+
-				" proceeding optimistically...", "err", err)
-		}
-
-		conR.SetStateSyncingMetrics(0)
-
-		d := types.EventDataStateSyncStatus{Complete: true, Height: state.LastBlockHeight}
-		if err := eb.PublishEventStateSyncStatus(d); err != nil {
-			stateSyncLogger.Error("failed to emit the statesync start event", "err", err)
-		}
-
-		if blockSync {
-			// FIXME Very ugly to have these metrics bleed through here.
-			conR.SetBlockSyncingMetrics(1)
-			if err := bcR.SwitchToBlockSync(state); err != nil {
-				stateSyncLogger.Error("failed to switch to block sync", "err", err)
-				return
-			}
-
-			d := types.EventDataBlockSyncStatus{Complete: false, Height: state.LastBlockHeight}
-			if err := eb.PublishEventBlockSyncStatus(d); err != nil {
-				stateSyncLogger.Error("failed to emit the block sync starting event", "err", err)
-			}
-
-		} else {
-			conR.SwitchToConsensus(state, true)
-		}
-	}()
-	return nil
-}
-
 // genesisDocProvider returns a GenesisDoc.
 // It allows the GenesisDoc to be pulled from sources other than the
 // filesystem, for instance from a distributed key-value store cluster.
@@ -1221,24 +1199,3 @@ func getChannelsFromShim(reactorShim *p2p.ReactorShim) map[p2p.ChannelID]*p2p.Ch
 
 	return channels
 }
-
-func constructStateProvider(
-	ssc *cfg.StateSyncConfig,
-	state sm.State,
-	logger log.Logger,
-) (statesync.StateProvider, error) {
-	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
-	defer cancel()
-
-	to := light.TrustOptions{
-		Period: ssc.TrustPeriod,
-		Height: ssc.TrustHeight,
-		Hash:   ssc.TrustHashBytes(),
-	}
-
-	return statesync.NewLightClientStateProvider(
-		ctx,
-		state.ChainID, state.Version, state.InitialHeight,
-		ssc.RPCServers, to, logger,
-	)
-}

node/node_test.go

@@ -21,16 +21,12 @@ import (
 	"github.com/tendermint/tendermint/crypto"
 	"github.com/tendermint/tendermint/crypto/ed25519"
 	"github.com/tendermint/tendermint/crypto/tmhash"
-	consmocks "github.com/tendermint/tendermint/internal/consensus/mocks"
-	ssmocks "github.com/tendermint/tendermint/internal/statesync/mocks"
 
 	"github.com/tendermint/tendermint/internal/evidence"
 	"github.com/tendermint/tendermint/internal/mempool"
 	mempoolv0 "github.com/tendermint/tendermint/internal/mempool/v0"
-	statesync "github.com/tendermint/tendermint/internal/statesync"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
-	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	tmtime "github.com/tendermint/tendermint/libs/time"
 	"github.com/tendermint/tendermint/privval"
@@ -669,65 +665,3 @@ func loadStatefromGenesis(t *testing.T) sm.State {
 
 	return state
 }
-
-func TestNodeStartStateSync(t *testing.T) {
-	mockSSR := &statesync.MockSyncReactor{}
-	mockFSR := &consmocks.BlockSyncReactor{}
-	mockCSR := &consmocks.ConsSyncReactor{}
-	mockSP := &ssmocks.StateProvider{}
-	state := loadStatefromGenesis(t)
-	config := cfg.ResetTestRoot("load_state_from_genesis")
-
-	eventBus, err := createAndStartEventBus(log.TestingLogger())
-	defer func() {
-		err := eventBus.Stop()
-		require.NoError(t, err)
-	}()
-
-	require.NoError(t, err)
-	require.NotNil(t, eventBus)
-
-	sub, err := eventBus.Subscribe(context.Background(), "test-client", types.EventQueryStateSyncStatus, 10)
-	require.NoError(t, err)
-	require.NotNil(t, sub)
-
-	cfgSS := config.StateSync
-
-	mockSSR.On("Sync", context.TODO(), mockSP, cfgSS.DiscoveryTime).Return(state, nil).
-		On("Backfill", state).Return(nil)
-	mockCSR.On("SetStateSyncingMetrics", float64(0)).Return().
-		On("SwitchToConsensus", state, true).Return()
-
-	require.NoError(t,
-		startStateSync(mockSSR, mockFSR, mockCSR, mockSP, config.StateSync, false, state.InitialHeight, eventBus))
-
-	for cnt := 0; cnt < 2; {
-		select {
-		case <-time.After(3 * time.Second):
-			t.Errorf("StateSyncStatus timeout")
-		case msg := <-sub.Out():
-			if cnt == 0 {
-				ensureStateSyncStatus(t, msg, false, state.InitialHeight)
-				cnt++
-			} else {
-				// the state height = 0 because we are not actually update the state in this test
-				ensureStateSyncStatus(t, msg, true, 0)
-				cnt++
-			}
-		}
-	}
-
-	mockSSR.AssertNumberOfCalls(t, "Sync", 1)
-	mockSSR.AssertNumberOfCalls(t, "Backfill", 1)
-	mockCSR.AssertNumberOfCalls(t, "SetStateSyncingMetrics", 1)
-	mockCSR.AssertNumberOfCalls(t, "SwitchToConsensus", 1)
-}
-
-func ensureStateSyncStatus(t *testing.T, msg tmpubsub.Message, complete bool, height int64) {
-	t.Helper()
-	status, ok := msg.Data().(types.EventDataStateSyncStatus)
-
-	require.True(t, ok)
-	require.Equal(t, complete, status.Complete)
-	require.Equal(t, height, status.Height)
-}

node/setup.go

@@ -706,6 +706,7 @@ func makeNodeInfo(
 			byte(statesync.SnapshotChannel),
 			byte(statesync.ChunkChannel),
 			byte(statesync.LightBlockChannel),
+			byte(statesync.ParamsChannel),
 		},
 		Moniker: config.Moniker,
 		Other: types.NodeInfoOther{