tendermintp2p,raftp2p: implement block rollback

- wasmx-as-contracts commit f9cac107f76ed85ca085e18d7328d65ce825999d

Author: loredanacirstea

tests/testdata/tinygo/wasmx-blocks/lib/calldata.go

@@ -7,6 +7,7 @@ type CallData struct {
 	SetConsensusParams          *CallDataSetConsensusParams          `json:"setConsensusParams,omitempty"`
 	SetIndexedTransactionByHash *CallDataSetIndexedTransactionByHash `json:"setIndexedTransactionByHash,omitempty"`
 	BootstrapAfterStateSync     *CallDataBootstrap                   `json:"bootstrapAfterStateSync,omitempty"`
+	Rollback                    *CalldataRollback                    `json:"rollback,omitempty"`
 
 	GetIndexedData              *CallDataGetIndexedData              `json:"getIndexedData,omitempty"`
 	GetLastBlockIndex           *CallDataGetLastBlockIndex           `json:"getLastBlockIndex,omitempty"`
@@ -84,3 +85,10 @@ type CallDataBootstrap struct {
 	LastHeightChanged int64  `json:"last_height_changed"`
 	Params            []byte `json:"params"`
 }
+
+type CalldataRollback struct {
+	Height        int64          `json:"height"`
+	Hash          []byte         `json:"hash"`
+	TxHashes      [][]byte       `json:"txhashes"`
+	IndexedTopics []IndexedTopic `json:"indexed_topics"`
+}

tests/testdata/tinygo/wasmx-consensus-utils/lib/utils.go

@@ -201,7 +201,7 @@ type IndexedTopic struct {
 	Values []string `json:"values"`
 }
 
-func ExtractIndexedTopics(finalizeResp consensus.ResponseFinalizeBlock, txhashes [][]byte) []IndexedTopic {
+func ExtractIndexedTopics(txResults []consensus.ExecTxResult, txhashes [][]byte) []IndexedTopic {
 	topicMap := map[string][]string{}
 	push := func(topic string, txhash []byte) {
 		if len(topic) > MaxKeyLength {
@@ -211,8 +211,8 @@ func ExtractIndexedTopics(finalizeResp consensus.ResponseFinalizeBlock, txhashes
 		arr = append(arr, base64.StdEncoding.EncodeToString(txhash))
 		topicMap[topic] = arr
 	}
-	for i := range finalizeResp.TxResults {
-		res := finalizeResp.TxResults[i]
+	for i := range txResults {
+		res := txResults[i]
 		for _, ev := range res.Events {
 			for _, attr := range ev.Attributes {
 				if attr.Index {

tests/testdata/tinygo/wasmx-raft-lib/lib/action_utils.go

@@ -424,8 +424,8 @@ func updateConsensusParams(height int64, updates *typestnd.ConsensusParams) erro
 }
 
 // Expose select consensus-utils helpers
-func extractIndexedTopics(resp typestnd.ResponseFinalizeBlock, txhashes [][]byte) []blocks.IndexedTopic {
-	topics := consutils.ExtractIndexedTopics(resp, txhashes)
+func extractIndexedTopics(txResults []typestnd.ExecTxResult, txhashes [][]byte) []blocks.IndexedTopic {
+	topics := consutils.ExtractIndexedTopics(txResults, txhashes)
 	out := make([]blocks.IndexedTopic, len(topics))
 	for i := range topics {
 		out[i] = blocks.IndexedTopic{Topic: topics[i].Topic, Values: topics[i].Values}
@@ -1116,3 +1116,30 @@ func WeAreNotAloneInternal(nodes []p2p.NodeInfo, state CurrentState) bool {
 	}
 	return state.WeAreNotAlone
 }
+
+func RollbackBlockData(height int64, hash []byte, txhashes [][]byte, indexedTopics []blocks.IndexedTopic) error {
+	calldata := blocks.CalldataRollback{
+		Height:        height,
+		Hash:          hash,
+		TxHashes:      txhashes,
+		IndexedTopics: indexedTopics,
+	}
+
+	payload := map[string]any{
+		"rollback": calldata,
+	}
+	calldatastr, err := json.Marshal(payload)
+	if err != nil {
+		return fmt.Errorf("failed to marshal rollback data: %s", err.Error())
+	}
+
+	// Call storage
+	resp, err := callStorage(string(calldatastr), false)
+	if err != nil {
+		return fmt.Errorf("could not rollback finalized block: %s: %s", resp.Data, err.Error())
+	}
+	if resp.Success > 0 {
+		return fmt.Errorf("could not rollback finalized block: %s", resp.Data)
+	}
+	return nil
+}

tests/testdata/tinygo/wasmx-raft-lib/lib/actions.go

@@ -1855,7 +1855,7 @@ func StartBlockFinalizationInternal(entryobj *LogEntryAggregate, retry bool) (bo
 	if err != nil {
 		return false, err
 	}
-	indexedTopics := extractIndexedTopics(*finalizeResp, txHashBytes)
+	indexedTopics := extractIndexedTopics(finalizeResp.TxResults, txHashBytes)
 	if err := setFinalizedBlock(string(blockData), base64.StdEncoding.EncodeToString(processReq.Hash), txHashBytes, indexedTopics); err != nil {
 		return false, err
 	}
@@ -2256,3 +2256,130 @@ func VerifyCommitLight(params []fsm.ActionParam, event fsm.EventObject) error {
 	wasmx.SetFinishData(respbz)
 	return nil
 }
+
+// Leader usually commits one block in advance
+// so if this node is a follower, it should only remove the block in-waiting to be committed
+// TODO fixme; now we need to go back 2 blocks with the Leader & Followers, otherwise
+// the AppHash has a mismatch
+func Rollback(params []fsm.ActionParam, event fsm.EventObject) error {
+	// Extract commit parameter
+	height := int64(0)
+	for _, p := range params {
+		if p.Key == "height" {
+			height = parseI64(p.Value, height)
+			break
+		}
+	}
+	if height == 0 {
+		for _, p := range event.Params {
+			if p.Key == "data" {
+				height = parseI64(p.Value, height)
+				break
+			}
+		}
+	}
+	lastCommited, err := GetLastBlockIndex()
+	if err != nil {
+		return err
+	}
+	if height == 0 {
+		height = lastCommited
+	}
+
+	LoggerInfo("rolling back block", []string{"height", fmt.Sprintf("%d", height), "lastCommited", fmt.Sprintf("%d", lastCommited)})
+
+	if height > lastCommited+1 {
+		return nil
+	}
+	if height < lastCommited {
+		Revert(fmt.Sprintf(`must roll back last block first: %d`, height))
+	}
+
+	newHeight := height - 1
+
+	if height == lastCommited {
+		LoggerInfo("rolling back block", []string{"height", fmt.Sprint(height)})
+
+		data, err := getFinalBlock(height)
+		if err != nil {
+			return err
+		}
+		if data == "" {
+			return nil
+		}
+
+		var blockData blocks.BlockEntry
+		if err := json.Unmarshal([]byte(data), &blockData); err != nil {
+			return fmt.Errorf("failed to parse BlockEntry: %v", err)
+		}
+
+		var processReqWithMeta typestnd.RequestProcessProposalWithMetaInfo
+		if err := json.Unmarshal(blockData.Data, &processReqWithMeta); err != nil {
+			return fmt.Errorf("failed to unmarshal RequestProcessProposalWithMetaInfo: %v", err)
+		}
+		processReq := processReqWithMeta.Request
+
+		var finalizeResp typestnd.ResponseFinalizeBlock
+		if err := json.Unmarshal(blockData.Result, &finalizeResp); err != nil {
+			return fmt.Errorf("failed to unmarshal ResponseFinalizeBlock: %v", err)
+		}
+
+		var blockCommit typestnd.BlockCommit
+		if err := json.Unmarshal(blockData.LastCommit, &blockCommit); err != nil {
+			return fmt.Errorf("failed to unmarshal BlockCommit: %v", err)
+		}
+
+		var header typestnd.Header
+		if err := json.Unmarshal(blockData.Header, &header); err != nil {
+			return fmt.Errorf("failed to unmarshal Header: %v", err)
+		}
+
+		hash, err := consutils.GetHeaderHash(header)
+		if err != nil {
+			return err
+		}
+
+		// Rollback block from storage
+		var txhashes [][]byte
+		for _, tx := range processReq.Txs {
+			txhash := wasmx.Sha256(tx)
+			txhashes = append(txhashes, txhash)
+		}
+		indexedTopics := extractIndexedTopics(finalizeResp.TxResults, txhashes)
+
+		RollbackBlockData(height, hash, txhashes, indexedTopics)
+		LoggerInfo("rolled back block data", []string{"height", fmt.Sprint(height)})
+
+		// Update consensus state
+		lastCommitHash, _ := hex.DecodeString(strings.ToLower(string(header.LastCommitHash)))
+		lastResultsHash, _ := hex.DecodeString(strings.ToLower(string(header.LastResultsHash)))
+
+		state, err := GetCurrentState()
+		if err != nil {
+			return err
+		}
+		state.NextHeight = height
+		state.NextHash = []byte{}
+		state.AppHash = finalizeResp.AppHash
+		state.LastBlockID = GetBlockID(hash)
+		state.LastCommitHash = lastCommitHash
+		state.LastResultsHash = lastResultsHash
+		state.LastTime = processReq.Time
+		state.ValidValue = 0
+		state.ValidRound = 0
+		state.LockedValue = 0
+		state.LockedRound = 0
+		// state.LastBlockSignatures = blockCommit.Signatures
+
+		SetCurrentState(state)
+		LoggerInfo("rolledback consensus state", []string{"new_height", fmt.Sprint(newHeight), "app_hash", hex.EncodeToString(state.AppHash)})
+	}
+
+	// Reset indexes
+	RemoveLogEntry(height)
+	SetLastLogIndex(newHeight)
+	LoggerInfo("rolled back consensus data", []string{"new_height", fmt.Sprint(newHeight)})
+
+	LoggerInfo("rollback consensus completed", []string{"new_height", fmt.Sprint(newHeight)})
+	return nil
+}

tests/testdata/tinygo/wasmx-raftp2p-lib/cmd/main.go

@@ -313,6 +313,11 @@ func main() {
 			lib.Revert("VerifyCommitLight failed: " + err.Error())
 			return
 		}
+	case "rollback":
+		if err := raft.Rollback(calld.Params, calld.Event); err != nil {
+			lib.Revert("Rollback failed: " + err.Error())
+			return
+		}
 	default:
 		wasmx.Revert(append([]byte("invalid function call data: "), databz...))
 		return

wasmx/server/rollback.go

@@ -1,21 +1,26 @@
 package server
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/spf13/cobra"
 
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	sdkserver "github.com/cosmos/cosmos-sdk/server"
 	"github.com/cosmos/cosmos-sdk/server/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	mapp "github.com/loredanacirstea/wasmx/app"
 	"github.com/loredanacirstea/wasmx/multichain"
+	networktypes "github.com/loredanacirstea/wasmx/x/network/types"
+	wasmxtypes "github.com/loredanacirstea/wasmx/x/wasmx/types"
 )
 
-// NewRollbackCmd creates a command to rollback CometBFT and multistore state by one height.
+// NewRollbackCmd creates a command to rollback block, and multistore state by one height.
 func NewRollbackCmd(appCreator types.AppCreator, defaultNodeHome string) *cobra.Command {
-	var removeBlock bool
+	// var removeBlock bool
+	var blockHeightPtr *int64
 
 	cmd := &cobra.Command{
 		Use:   "rollback",
@@ -37,31 +42,55 @@ application.
 				return err
 			}
 			app := appCreator(ctx.Logger, db, nil, ctx.Viper)
-			// TODO maybe rollback blocks too?
-			// // rollback CometBFT state
-			// height, hash, err := cmtcmd.RollbackState(ctx.Config, removeBlock)
-			// if err != nil {
-			// 	return fmt.Errorf("failed to rollback CometBFT state: %w", err)
-			// }
-
 			baseapp := app.(*mapp.App)
-			height := baseapp.LastBlockHeight() - 1
-			fmt.Printf("rolling back to version: %d \n", height)
-
-			// rollback the multistore
-			if err := app.CommitMultiStore().RollbackToVersion(height); err != nil {
-				return fmt.Errorf("failed to rollback to version: %w", err)
+			height := baseapp.LastBlockHeight()
+			newheight := height - 1
+			blockHeight := int64(0)
+			if blockHeightPtr != nil {
+				blockHeight = *blockHeightPtr
+			}
+			if blockHeight == int64(0) {
+				blockHeight = height
+			}
+			if blockHeight < height {
+				return fmt.Errorf("can only roll back last block: %d", height)
+			}
+			if blockHeight == height {
+				fmt.Printf("rolling back to version: %d \n", newheight)
+				// rollback the multistore
+				if err := app.CommitMultiStore().RollbackToVersion(newheight); err != nil {
+					return fmt.Errorf("failed to rollback to version: %w", err)
+				}
 			}
 
 			height = baseapp.LastBlockHeight()
 			hash := baseapp.LastCommitID().Hash
 			fmt.Printf("Rolled back state to height %d and hash %X", height, hash)
-			return nil
+
+			_, goctx := getCtx(ctx, true)
+			cb := func(goctx context.Context) (any, error) {
+				ctx := sdk.UnwrapSDKContext(goctx)
+				msg := []byte(fmt.Sprintf(`{"execute":{"action": {"type": "rollback", "params": [{"key":"height","value":"%d"}],"event":null}}}`, blockHeight))
+				execmsg := &networktypes.MsgExecuteContract{
+					Sender:   wasmxtypes.ROLE_CONSENSUS,
+					Contract: wasmxtypes.ROLE_CONSENSUS,
+					Msg:      msg,
+				}
+				res, err := baseapp.NetworkKeeper.ExecuteContractInternal(ctx, execmsg)
+				if err != nil {
+					return nil, err
+				}
+				return res, nil
+			}
+
+			_, err = baseapp.GetActionExecutor().Execute(goctx, height, sdk.ExecModeFinalize, cb)
+			return err
 		},
 	}
 
 	cmd.Flags().String(flags.FlagHome, defaultNodeHome, "The application home directory")
-	cmd.Flags().BoolVar(&removeBlock, "hard", false, "remove last block as well as state")
+	// cmd.Flags().BoolVar(&removeBlock, "hard", false, "remove last block as well as state")
+	blockHeightPtr = cmd.Flags().Int64("height", 0, "block height to remove, optional")
 	cmd.Flags().String(flags.FlagChainID, "testnet", "Specify Chain ID for sending Tx")
 	cmd.Flags().String(multichain.FlagRegistryChainId, "", "multichain registry chain id")
 	return cmd