[admin] pause/unpause blockchain (#4656)

---------

Co-authored-by: CoderZhi <[email protected]>

Author: envestcc

blockchain/blockchain.go

@@ -57,6 +57,8 @@ var (
 	ErrInsufficientGas = errors.New("insufficient intrinsic gas value")
 	// ErrBalance indicates the error of balance
 	ErrBalance = errors.New("invalid balance")
+	// ErrPaused indicates the error of blockchain is paused
+	ErrPaused = errors.New("blockchain is paused")
 )
 
 func init() {
@@ -112,6 +114,8 @@ type (
 
 		// RemoveSubscriber make you listen to every single produced block
 		RemoveSubscriber(BlockCreationSubscriber) error
+		//  Pause pauses the blockchain
+		Pause(bool)
 	}
 
 	// BlockMinter is the block minter interface
@@ -133,7 +137,8 @@ type (
 		timerFactory   *prometheustimer.TimerFactory
 
 		// used by account-based model
-		bbf BlockMinter
+		bbf   BlockMinter
+		pause bool
 	}
 )
 
@@ -264,6 +269,12 @@ func (bc *blockchain) Stop(ctx context.Context) error {
 	return bc.lifecycle.OnStop(ctx)
 }
 
+func (bc *blockchain) Pause(pause bool) {
+	bc.mu.Lock()
+	defer bc.mu.Unlock()
+	bc.pause = pause
+}
+
 func (bc *blockchain) BlockHeaderByHeight(height uint64) (*block.Header, error) {
 	return bc.dao.HeaderByHeight(height)
 }
@@ -473,6 +484,9 @@ func (bc *blockchain) MintNewBlock(timestamp time.Time, opts ...MintOption) (*bl
 func (bc *blockchain) CommitBlock(blk *block.Block) error {
 	bc.mu.Lock()
 	defer bc.mu.Unlock()
+	if bc.pause {
+		return errors.Wrapf(ErrPaused, "blockchain is paused, cannot commit block %d, %x", blk.Height(), blk.HashBlock())
+	}
 	timer := bc.timerFactory.NewTimer("CommitBlock")
 	defer timer.End()
 	return bc.commitBlock(blk)

blocksync/blocksync.go

@@ -18,6 +18,7 @@ import (
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/iotexproject/iotex-core/v2/blockchain"
 	"github.com/iotexproject/iotex-core/v2/blockchain/block"
 	"github.com/iotexproject/iotex-core/v2/blockchain/blockdao"
 	"github.com/iotexproject/iotex-core/v2/pkg/fastrand"
@@ -169,6 +170,8 @@ func (bs *blockSyncer) commitBlocks(blks []*peerBlock) bool {
 			return true
 		case blockdao.ErrRemoteHeightTooLow:
 			log.L().Info("remote height too low", zap.Uint64("height", blk.block.Height()))
+		case blockchain.ErrPaused:
+			log.L().Info("blockchain is paused, skip committing block", zap.Uint64("height", blk.block.Height()))
 		default:
 			bs.blockP2pPeer(blk.pid)
 			log.L().Error("failed to commit block", zap.Error(err), zap.Uint64("height", blk.block.Height()), zap.String("peer", blk.pid))

chainservice/chainservice.go

@@ -83,6 +83,7 @@ type ChainService struct {
 	minter                   *factory.Minter
 
 	lastReceivedBlockHeight uint64
+	paused                  atomic.Bool
 }
 
 // Start starts the server
@@ -101,6 +102,9 @@ func (cs *ChainService) Start(ctx context.Context) error {
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
+				if cs.paused.Load() {
+					continue
+				}
 				currentHeight := cs.chain.TipHeight()
 				lrbh := atomic.LoadUint64(&cs.lastReceivedBlockHeight)
 				if currentHeight == lastHeight && lastReceivedBlockHeight != lrbh {
@@ -119,6 +123,10 @@ func (cs *ChainService) Stop(ctx context.Context) error {
 	return cs.lifecycle.OnStopSequentially(ctx)
 }
 
+func (cs *ChainService) Pause(pause bool) {
+	cs.paused.Store(pause)
+}
+
 // ReportFullness switch on or off block sync
 func (cs *ChainService) ReportFullness(_ context.Context, messageType iotexrpc.MessageType, fullness float32) {
 	_blockchainFullnessMtc.WithLabelValues(iotexrpc.MessageType_name[int32(messageType)]).Set(float64(fullness))

consensus/scheme/rolldpos/rolldposctx.go

@@ -593,6 +593,9 @@ func (ctx *rollDPoSCtx) Commit(msg interface{}) (bool, error) {
 	switch err := ctx.chain.CommitBlock(pendingBlock); errors.Cause(err) {
 	case blockchain.ErrInvalidTipHeight:
 		return true, nil
+	case blockchain.ErrPaused:
+		ctx.logger().Info("chain is paused, block will not be committed")
+		return false, nil
 	case nil:
 		break
 	default:

consensus/scheme/standalone.go

@@ -12,11 +12,12 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+
 	"github.com/iotexproject/iotex-core/v2/blockchain"
 	"github.com/iotexproject/iotex-core/v2/blockchain/block"
 	"github.com/iotexproject/iotex-core/v2/pkg/log"
 	"github.com/iotexproject/iotex-core/v2/pkg/routine"
-	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 )
 
 // Standalone is the consensus scheme that periodically create blocks
@@ -39,6 +40,10 @@ func (s *standaloneHandler) Run() {
 	}
 
 	if err := s.commitCb(blk); err != nil {
+		if errors.Is(err, blockchain.ErrPaused) {
+			log.L().Info("Consensus is paused, skip committing block", zap.Uint64("height", blk.Height()))
+			return
+		}
 		log.L().Error("Failed to commit.", zap.Error(err))
 		return
 	}

server/itx/pauseable.go

@@ -0,0 +1,36 @@
+package itx
+
+import (
+	"net/http"
+)
+
+type (
+	Pauseable interface {
+		Pause(bool)
+	}
+	PauseMgr struct {
+		pauses []Pauseable
+	}
+)
+
+func NewPauseMgr(pauses ...Pauseable) *PauseMgr {
+	return &PauseMgr{
+		pauses: pauses,
+	}
+}
+
+func (pm *PauseMgr) Pause(pause bool) {
+	for _, p := range pm.pauses {
+		p.Pause(pause)
+	}
+}
+
+func (pm *PauseMgr) HandlePause(w http.ResponseWriter, r *http.Request) {
+	pm.Pause(true)
+	w.WriteHeader(http.StatusOK)
+}
+
+func (pm *PauseMgr) HandleUnPause(w http.ResponseWriter, r *http.Request) {
+	pm.Pause(false)
+	w.WriteHeader(http.StatusOK)
+}

server/itx/server.go

@@ -43,6 +43,7 @@ type Server struct {
 	p2pAgent             p2p.Agent
 	dispatcher           dispatcher.Dispatcher
 	nodeStats            *nodestats.NodeStats
+	pauseMgr             *PauseMgr
 	initializedSubChains map[uint32]bool
 	mutex                sync.RWMutex
 	subModuleCancel      context.CancelFunc
@@ -120,6 +121,7 @@ func newServer(cfg config.Config, testing bool) (*Server, error) {
 		return nil, errors.Wrap(err, "fail to create chain service")
 	}
 	nodeStats := nodestats.NewNodeStats(rpcStats, cs.BlockSync(), p2pAgent)
+	pauseMgr := NewPauseMgr(cs.Blockchain(), cs)
 	apiServer, err := cs.NewAPIServer(cfg.API, cfg.Chain.EnableArchiveMode)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create api server")
@@ -141,6 +143,7 @@ func newServer(cfg config.Config, testing bool) (*Server, error) {
 		chainservices:        chains,
 		apiServers:           apiServers,
 		nodeStats:            nodeStats,
+		pauseMgr:             pauseMgr,
 		initializedSubChains: map[uint32]bool{},
 	}
 	// Setup sub-chain starter
@@ -302,6 +305,8 @@ func StartServer(ctx context.Context, svr *Server, probeSvr *probe.Server, cfg c
 		mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
 		mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
 		mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
+		mux.Handle("/pause", http.HandlerFunc(svr.pauseMgr.HandlePause))
+		mux.Handle("/unpause", http.HandlerFunc(svr.pauseMgr.HandleUnPause))
 
 		port := fmt.Sprintf(":%d", cfg.System.HTTPAdminPort)
 		adminserv = httputil.NewServer(port, mux)