replay blocks to sync app and rollup

Author: gupadhyaya

block/manager.go

@@ -157,7 +157,7 @@ func NewManager(
 	genesis *cmtypes.GenesisDoc,
 	store store.Store,
 	mempool mempool.Mempool,
-	proxyApp proxy.AppConnConsensus,
+	proxyApp proxy.AppConns,
 	dalc *da.DAClient,
 	eventBus *cmtypes.EventBus,
 	logger log.Logger,
@@ -217,7 +217,7 @@ func NewManager(
 	// allow buffer for the block header and protocol encoding
 	maxBlobSize -= blockProtocolOverhead
 
-	exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp, eventBus, maxBlobSize, logger, execMetrics, valSet.Hash())
+	exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp.Consensus(), eventBus, maxBlobSize, logger, execMetrics, valSet.Hash())
 	if s.LastBlockHeight+1 == uint64(genesis.InitialHeight) {
 		res, err := exec.InitChain(genesis)
 		if err != nil {
@@ -268,6 +268,14 @@ func NewManager(
 		metrics:       seqMetrics,
 		isProposer:    isProposer,
 	}
+
+	if agg.conf.Replay {
+		// Handshake to ensure that app and rollup are in sync
+		if err := agg.Handshake(context.Background(), proxyApp); err != nil {
+			return nil, err
+		}
+	}
+
 	return agg, nil
 }
 
@@ -1166,3 +1174,124 @@ func updateState(s *types.State, res *abci.ResponseInitChain) {
 	s.LastResultsHash = merkle.HashFromByteSlices(nil)
 
 }
+
+// Handshake performs the ABCI handshake with the application.
+func (m *Manager) Handshake(ctx context.Context, proxyApp proxy.AppConns) error {
+	// Handshake is done via ABCI Info on the query conn.
+	res, err := proxyApp.Query().Info(ctx, proxy.RequestInfo)
+	if err != nil {
+		return fmt.Errorf("error calling Info: %v", err)
+	}
+
+	blockHeight := res.LastBlockHeight
+	if blockHeight < 0 {
+		return fmt.Errorf("got a negative last block height (%d) from the app", blockHeight)
+	}
+	appHash := res.LastBlockAppHash
+
+	m.logger.Info("ABCI Handshake App Info",
+		"height", blockHeight,
+		"hash", fmt.Sprintf("%X", appHash),
+		"software-version", res.Version,
+		"protocol-version", res.AppVersion,
+	)
+
+	// Replay blocks up to the latest in the blockstore.
+	appHash, err = m.ReplayBlocks(ctx, appHash, blockHeight, proxyApp)
+	if err != nil {
+		return fmt.Errorf("error on replay: %v", err)
+	}
+
+	m.logger.Info("Completed ABCI Handshake - CometBFT and App are synced",
+		"appHeight", blockHeight, "appHash", fmt.Sprintf("%X", appHash))
+
+	// TODO: (on restart) replay mempool
+
+	return nil
+}
+
+// ReplayBlocks replays blocks from the last state to the app's last block height.
+func (m *Manager) ReplayBlocks(
+	ctx context.Context,
+	appHash []byte,
+	appBlockHeight int64,
+	proxyApp proxy.AppConns,
+) ([]byte, error) {
+	state := m.lastState
+	stateBlockHeight := m.lastState.LastBlockHeight
+	m.logger.Info(
+		"ABCI Replay Blocks",
+		"appHeight",
+		appBlockHeight,
+		"stateHeight",
+		stateBlockHeight)
+
+	if appBlockHeight < int64(stateBlockHeight) {
+		// the app is behind, so replay blocks
+		return m.replayBlocks(ctx, state, proxyApp, uint64(appBlockHeight), stateBlockHeight)
+	} else if appBlockHeight == int64(stateBlockHeight) {
+		// We're good!
+		assertAppHashEqualsOneFromState(appHash, state)
+		return appHash, nil
+	}
+
+	panic(fmt.Sprintf("uncovered case! app height higher than state height, possibly need app rollback; appHeight: %d, stateHeight: %d", appBlockHeight, stateBlockHeight))
+}
+
+func (m *Manager) replayBlocks(
+	ctx context.Context,
+	state types.State,
+	proxyApp proxy.AppConns,
+	appBlockHeight,
+	stateBlockHeight uint64,
+) ([]byte, error) {
+	var appHash []byte
+	finalBlock := stateBlockHeight
+	firstBlock := appBlockHeight + 1
+	if firstBlock == 1 {
+		firstBlock = state.InitialHeight
+	}
+	for i := firstBlock; i <= finalBlock; i++ {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+		}
+
+		m.logger.Info("Applying block", "height", i)
+		block, err := m.store.GetBlock(ctx, i)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get block data for height %d: %w", i, err)
+		}
+		appHash, err = m.executor.ExecCommitBlock(proxyApp.Consensus(), block, m.logger, state, m.store)
+		if err != nil {
+			return nil, err
+		}
+		// Extra check to ensure the app was not changed in a way it shouldn't have.
+		if len(appHash) > 0 {
+			assertAppHashEqualsOneFromBlock(appHash, block)
+		}
+	}
+
+	assertAppHashEqualsOneFromState(appHash, state)
+	return appHash, nil
+}
+
+func assertAppHashEqualsOneFromBlock(appHash []byte, block *types.Block) {
+	if !bytes.Equal(appHash, block.SignedHeader.AppHash) {
+		panic(fmt.Sprintf(`block.AppHash does not match AppHash after replay. Got %X, expected %X.
+Block: %v
+`,
+			appHash, block.SignedHeader.AppHash, block))
+	}
+}
+
+func assertAppHashEqualsOneFromState(appHash []byte, state types.State) {
+	if !bytes.Equal(appHash, state.AppHash) {
+		panic(fmt.Sprintf(`state.AppHash does not match AppHash after replay. Got
+%X, expected %X.
+State: %v
+Did you reset CometBFT without resetting your application's data?`,
+			appHash, state.AppHash, state))
+	}
+}

config/config.go

@@ -36,6 +36,8 @@ const (
 	FlagLazyAggregator = "rollkit.lazy_aggregator"
 	// FlagMaxPendingBlocks is a flag to pause aggregator in case of large number of blocks pending DA submission
 	FlagMaxPendingBlocks = "rollkit.max_pending_blocks"
+	// FlagReplay is a flag for replaying blocks
+	FlagReplay = "rollkit.replay"
 )
 
 // NodeConfig stores Rollkit node configuration.
@@ -82,6 +84,8 @@ type BlockManagerConfig struct {
 	// LazyBlockTime defines how often new blocks are produced in lazy mode
 	// even if there are no transactions
 	LazyBlockTime time.Duration `mapstructure:"lazy_block_time"`
+	// Replay defines whether to replay blocks
+	Replay bool `mapstructure:"replay"`
 }
 
 // GetNodeConfig translates Tendermint's configuration into Rollkit configuration.
@@ -129,6 +133,7 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 	nc.TrustedHash = v.GetString(FlagTrustedHash)
 	nc.TrustedHash = v.GetString(FlagTrustedHash)
 	nc.MaxPendingBlocks = v.GetUint64(FlagMaxPendingBlocks)
+	nc.Replay = v.GetBool(FlagReplay)
 	return nil
 }
 
@@ -150,4 +155,5 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(FlagLight, def.Light, "run light client")
 	cmd.Flags().String(FlagTrustedHash, def.TrustedHash, "initial trusted hash to start the header exchange service")
 	cmd.Flags().Uint64(FlagMaxPendingBlocks, def.MaxPendingBlocks, "limit of blocks pending DA submission (0 for no limit)")
+	cmd.Flags().Bool(FlagReplay, def.Replay, "replay blocks")
 }

config/defaults.go

@@ -26,6 +26,7 @@ var DefaultNodeConfig = NodeConfig{
 		BlockTime:     1 * time.Second,
 		DABlockTime:   15 * time.Second,
 		LazyBlockTime: 60 * time.Second,
+		Replay:        false,
 	},
 	DAAddress:       "http://localhost:26658",
 	DAGasPrice:      -1,

node/full.go

@@ -264,7 +264,7 @@ func initBlockSyncService(ctx context.Context, mainKV ds.TxnDatastore, nodeConfi
 }
 
 func initBlockManager(signingKey crypto.PrivKey, nodeConfig config.NodeConfig, genesis *cmtypes.GenesisDoc, store store.Store, mempool mempool.Mempool, proxyApp proxy.AppConns, dalc *da.DAClient, eventBus *cmtypes.EventBus, logger log.Logger, blockSyncService *block.BlockSyncService, seqMetrics *block.Metrics, execMetrics *state.Metrics) (*block.Manager, error) {
-	blockManager, err := block.NewManager(signingKey, nodeConfig.BlockManagerConfig, genesis, store, mempool, proxyApp.Consensus(), dalc, eventBus, logger.With("module", "BlockManager"), blockSyncService.BlockStore(), seqMetrics, execMetrics)
+	blockManager, err := block.NewManager(signingKey, nodeConfig.BlockManagerConfig, genesis, store, mempool, proxyApp, dalc, eventBus, logger.With("module", "BlockManager"), blockSyncService.BlockStore(), seqMetrics, execMetrics)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing BlockManager: %w", err)
 	}

state/executor.go

@@ -14,6 +14,7 @@ import (
 	cmtypes "github.com/cometbft/cometbft/types"
 
 	"github.com/rollkit/rollkit/mempool"
+	"github.com/rollkit/rollkit/store"
 	"github.com/rollkit/rollkit/third_party/log"
 	"github.com/rollkit/rollkit/types"
 	abciconv "github.com/rollkit/rollkit/types/abci"
@@ -505,3 +506,60 @@ func fromRollkitTxs(rollkitTxs types.Txs) cmtypes.Txs {
 	}
 	return txs
 }
+
+// ----------------------------------------------------------------------------------------------------
+// Execute block without state. TODO: eliminate
+
+// ExecCommitBlock executes and commits a block on the proxyApp without validating or mutating the state.
+// It returns the application root hash (result of abci.Commit).
+func (e *BlockExecutor) ExecCommitBlock(
+	appConnConsensus proxy.AppConnConsensus,
+	block *types.Block,
+	logger log.Logger,
+	state types.State,
+	store store.Store,
+) ([]byte, error) {
+	abciHeader, err := abciconv.ToABCIHeaderPB(&block.SignedHeader.Header)
+	if err != nil {
+		return nil, err
+	}
+	abciHeader.ChainID = e.chainID
+	abciBlock, err := abciconv.ToABCIBlock(block)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := e.proxyApp.FinalizeBlock(context.TODO(), &abci.RequestFinalizeBlock{
+		Hash:               block.Hash(),
+		NextValidatorsHash: e.valsetHash,
+		ProposerAddress:    abciHeader.ProposerAddress,
+		Height:             abciHeader.Height,
+		Time:               abciHeader.Time,
+		DecidedLastCommit: abci.CommitInfo{
+			Round: 0,
+			Votes: nil,
+		},
+		Misbehavior: abciBlock.Evidence.Evidence.ToABCI(),
+		Txs:         abciBlock.Txs.ToSliceOfBytes(),
+	})
+	if err != nil {
+		logger.Error("Error in proxyAppConn.FinalizeBlock", "err", err)
+		return nil, err
+	}
+
+	// Assert that the application correctly returned tx results for each of the transactions provided in the block
+	if len(block.Data.Txs) != len(resp.TxResults) {
+		return nil, fmt.Errorf("expected tx results length to match size of transactions in block. Expected %d, got %d", len(block.Data.Txs), len(resp.TxResults))
+	}
+
+	logger.Info("Executed block", "height", block.Height, "app_hash", fmt.Sprintf("%X", resp.AppHash))
+
+	// Commit block
+	_, err = e.proxyApp.Commit(context.TODO())
+	if err != nil {
+		logger.Error("Client error during proxyAppConn.Commit", "err", err)
+		return nil, err
+	}
+
+	// ResponseCommit has no error or log
+	return resp.AppHash, nil
+}