feat: add block rollback support

.gitignore

@@ -16,7 +16,7 @@ coverage.out
 execution/evm/jwttoken
 target
 /.claude/settings.local.json
-
+apps/testapp/testapp
 
 docs/.vitepress/dist
 node_modules

apps/testapp/cmd/rollback.go

@@ -0,0 +1,65 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+
+	kvexecutor "github.com/evstack/ev-node/apps/testapp/kv"
+	rollcmd "github.com/evstack/ev-node/pkg/cmd"
+	"github.com/evstack/ev-node/pkg/store"
+	"github.com/spf13/cobra"
+)
+
+var RollbackCmd = &cobra.Command{
+	Use: "rollback <height>",
+	Short: "Rollback the testapp node",
+	Args: cobra.RangeArgs(0, 1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+ nodeConfig, err := rollcmd.ParseConfig(cmd)
+ if err != nil {
+ return err
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ datastore, err := store.NewDefaultKVStore(nodeConfig.RootDir, nodeConfig.DBPath, "testapp")
+ if err != nil {
+ return err
+ }
+ storeWrapper := store.New(datastore)
+
+ executor, err := kvexecutor.NewKVExecutor(nodeConfig.RootDir, nodeConfig.DBPath)
+ if err != nil {
+ return err
+ }
+
+ cmd.Println("Starting rollback operation")
+ currentHeight, err := storeWrapper.Height(ctx)
+ if err != nil {
+ return fmt.Errorf("failed to get current height: %w", err)
+ }
+
+ var targetHeight uint64 = currentHeight - 1
+ if len(args) > 0 {
+ targetHeight, err = strconv.ParseUint(args[0], 10, 64)
+ if err != nil {
+ return fmt.Errorf("failed to parse target height: %w", err)
+ }
+ }
+
+ // rollback ev-node store
+ if err := storeWrapper.Rollback(ctx, targetHeight); err != nil {
+ return fmt.Errorf("rollback failed: %w", err)
+ }
+
+ // rollback execution store
+ if err := executor.Rollback(ctx, targetHeight); err != nil {
+ return fmt.Errorf("rollback failed: %w", err)
+ }
+
+ cmd.Println("Rollback completed successfully")
+ return nil
+	},
+}

apps/testapp/cmd/root.go

@@ -9,9 +9,8 @@ import (
 const (
 	// AppName is the name of the application, the name of the command, and the name of the home directory.
 	AppName = "testapp"
-)
 
-const (
+	// flagKVEndpoint is the flag for the KV endpoint
 	flagKVEndpoint = "kv-endpoint"
 )
 

apps/testapp/kv/http_server_test.go

@@ -277,3 +277,40 @@ func TestHTTPServerContextCancellation(t *testing.T) {
  t.Fatal("Expected connection error after shutdown, but got none")
 	}
 }
+
+func TestHTTPIntegration_GetKVWithMultipleHeights(t *testing.T) {
+	exec, err := NewKVExecutor(t.TempDir(), "testdb")
+	if err != nil {
+ t.Fatalf("Failed to create KVExecutor: %v", err)
+	}
+	ctx := context.Background()
+
+	// Execute transactions at different heights for the same key
+	txsHeight1 := [][]byte{[]byte("testkey=original_value")}
+	_, _, err = exec.ExecuteTxs(ctx, txsHeight1, 1, time.Now(), []byte(""))
+	if err != nil {
+ t.Fatalf("ExecuteTxs failed for height 1: %v", err)
+	}
+
+	txsHeight2 := [][]byte{[]byte("testkey=updated_value")}
+	_, _, err = exec.ExecuteTxs(ctx, txsHeight2, 2, time.Now(), []byte(""))
+	if err != nil {
+ t.Fatalf("ExecuteTxs failed for height 2: %v", err)
+	}
+
+	server := NewHTTPServer(exec, ":0")
+
+	// Test GET request - should return the latest value
+	req := httptest.NewRequest(http.MethodGet, "/kv?key=testkey", nil)
+	rr := httptest.NewRecorder()
+
+	server.handleKV(rr, req)
+
+	if rr.Code != http.StatusOK {
+ t.Errorf("expected status 200, got %d", rr.Code)
+	}
+
+	if rr.Body.String() != "updated_value" {
+ t.Errorf("expected body 'updated_value', got %q", rr.Body.String())
+	}
+}

apps/testapp/kv/kvexecutor.go

@@ -16,6 +16,7 @@ import (
 var (
 	genesisInitializedKey = ds.NewKey("/genesis/initialized")
 	genesisStateRootKey = ds.NewKey("/genesis/stateroot")
+	heightKeyPrefix = ds.NewKey("/height")
 	finalizedHeightKey = ds.NewKey("/finalizedHeight")
 	// Define a buffer size for the transaction channel
 	txChannelBufferSize = 10000
@@ -49,18 +50,56 @@ func NewKVExecutor(rootdir, dbpath string) (*KVExecutor, error) {
 }
 
 // GetStoreValue is a helper for the HTTP interface to retrieve the value for a key from the database.
+// It searches across all block heights to find the latest value for the given key.
 func (k *KVExecutor) GetStoreValue(ctx context.Context, key string) (string, bool) {
-	dsKey := ds.NewKey(key)
-	valueBytes, err := k.db.Get(ctx, dsKey)
-	if errors.Is(err, ds.ErrNotFound) {
+	// Query all keys to find height-prefixed versions of this key
+	q := query.Query{}
+	results, err := k.db.Query(ctx, q)
+	if err != nil {
+ fmt.Printf("Error querying DB for key '%s': %v\n", key, err)
  return "", false
 	}
-	if err != nil {
- // Log the error or handle it appropriately
- fmt.Printf("Error getting value from DB: %v\n", err)
+	defer results.Close()
+
+	heightPrefix := heightKeyPrefix.String()
+	var latestValue string
+	var latestHeight uint64
+	found := false
+
+	for result := range results.Next() {
+ if result.Error != nil {
+ fmt.Printf("Error iterating query results for key '%s': %v\n", key, result.Error)
+ return "", false
+ }
+
+ resultKey := result.Key
+ // Check if this is a height-prefixed key that matches our target key
+ if strings.HasPrefix(resultKey, heightPrefix+"/") {
+ // Extract height and actual key: /height/{height}/{actual_key}
+ parts := strings.Split(strings.TrimPrefix(resultKey, heightPrefix+"/"), "/")
+ if len(parts) >= 2 {
+ var keyHeight uint64
+ if _, err := fmt.Sscanf(parts[0], "%d", &keyHeight); err == nil {
+ // Reconstruct the actual key by joining all parts after the height
+ actualKey := strings.Join(parts[1:], "/")
+ if actualKey == key {
+ // This key matches - check if it's the latest height
+ if !found || keyHeight > latestHeight {
+ latestHeight = keyHeight
+ latestValue = string(result.Value)
+ found = true
+ }
+ }
+ }
+ }
+ }
+	}
+
+	if !found {
  return "", false
 	}
-	return string(valueBytes), true
+
+	return latestValue, true
 }
 
 // computeStateRoot computes a deterministic state root by querying all keys, sorting them,
@@ -206,11 +245,14 @@ func (k *KVExecutor) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight u
  if key == "" {
  return nil, 0, errors.New("empty key in transaction")
  }
- dsKey := ds.NewKey(key)
+
+ dsKey := getTxKey(blockHeight, key)
+
  // Prevent writing reserved keys via transactions
  if reservedKeys[dsKey] {
  return nil, 0, fmt.Errorf("transaction attempts to modify reserved key: %s", key)
  }
+
  err = batch.Put(ctx, dsKey, []byte(value))
  if err != nil {
  // This error is unlikely for Put unless the context is cancelled.
@@ -263,3 +305,96 @@ func (k *KVExecutor) InjectTx(tx []byte) {
  // Consider adding metrics here
 	}
 }
+
+// Rollback reverts the state to the previous block height.
+func (k *KVExecutor) Rollback(ctx context.Context, height uint64) error {
+	select {
+	case <-ctx.Done():
+ return ctx.Err()
+	default:
+	}
+
+	// Validate height constraints
+	if height == 0 {
+ return fmt.Errorf("cannot rollback to height 0: invalid height")
+	}
+
+	// Create a batch for atomic rollback operation
+	batch, err := k.db.Batch(ctx)
+	if err != nil {
+ return fmt.Errorf("failed to create batch for rollback: %w", err)
+	}
+
+	// Query all keys to find those with height > target height
+	q := query.Query{}
+	results, err := k.db.Query(ctx, q)
+	if err != nil {
+ return fmt.Errorf("failed to query keys for rollback: %w", err)
+	}
+	defer results.Close()
+
+	keysToDelete := make([]ds.Key, 0)
+	heightPrefix := heightKeyPrefix.String()
+
+	for result := range results.Next() {
+ if result.Error != nil {
+ return fmt.Errorf("error iterating query results during rollback: %w", result.Error)
+ }
+
+ key := result.Key
+ // Check if this is a height-prefixed key
+ if strings.HasPrefix(key, heightPrefix+"/") {
+ // Extract height from key: /height/{height}/{actual_key} (see getTxKey)
+ parts := strings.Split(strings.TrimPrefix(key, heightPrefix+"/"), "/")
+ if len(parts) > 0 {
+ var keyHeight uint64
+ if _, err := fmt.Sscanf(parts[0], "%d", &keyHeight); err == nil {
+ // If this key's height is greater than target, mark for deletion
+ if keyHeight > height {
+ keysToDelete = append(keysToDelete, ds.NewKey(key))
+ }
+ }
+ }
+ }
+	}
+
+	// Delete all keys with height > target height
+	for _, key := range keysToDelete {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ default:
+ }
+
+ err = batch.Delete(ctx, key)
+ if err != nil {
+ return fmt.Errorf("failed to stage delete operation for key '%s' during rollback: %w", key.String(), err)
+ }
+	}
+
+	// Update finalized height if necessary - it should not exceed rollback height
+	finalizedHeightKey := ds.NewKey("/finalizedHeight")
+	if finalizedHeightBytes, err := k.db.Get(ctx, finalizedHeightKey); err == nil {
+ var finalizedHeight uint64
+ if _, err := fmt.Sscanf(string(finalizedHeightBytes), "%d", &finalizedHeight); err == nil {
+ if finalizedHeight > height {
+ err = batch.Put(ctx, finalizedHeightKey, fmt.Appendf([]byte{}, "%d", height))
+ if err != nil {
+ return fmt.Errorf("failed to update finalized height during rollback: %w", err)
+ }
+ }
+ }
+	}
+
+	// Commit the batch atomically
+	err = batch.Commit(ctx)
+	if err != nil {
+ return fmt.Errorf("failed to commit rollback batch: %w", err)
+	}
+
+	return nil
+}
+
+func getTxKey(height uint64, txKey string) ds.Key {
+	return heightKeyPrefix.Child(ds.NewKey(fmt.Sprintf("%d/%s", height, txKey)))
+}

apps/testapp/main.go

@@ -20,6 +20,7 @@ func main() {
  rollcmd.NetInfoCmd,
  rollcmd.StoreUnsafeCleanCmd,
  rollcmd.KeysCmd(),
+ cmds.RollbackCmd,
  initCmd,
 	)
 

block/publish_block_p2p_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
@@ -28,6 +29,7 @@ import (
 	"github.com/evstack/ev-node/pkg/signer/noop"
 	"github.com/evstack/ev-node/pkg/store"
 	evSync "github.com/evstack/ev-node/pkg/sync"
+	"github.com/evstack/ev-node/test/mocks"
 	"github.com/evstack/ev-node/types"
 )
 
@@ -200,13 +202,18 @@ func setupBlockManager(t *testing.T, ctx context.Context, workDir string, mainKV
 	require.NoError(t, err)
 	require.NoError(t, dataSyncService.Start(ctx))
 
+	mockExecutor := mocks.NewMockExecutor(t)
+	mockExecutor.On("InitChain", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(bytesN(32), uint64(10_000), nil).Maybe()
+	mockExecutor.On("ExecuteTxs", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(bytesN(32), uint64(10_000), nil).Maybe()
+	mockExecutor.On("SetFinal", mock.Anything, mock.Anything).Return(nil).Maybe()
+
 	result, err := NewManager(
  ctx,
  signer,
  nodeConfig,
  genesisDoc,
  store.New(mainKV),
-&mockExecutor{},
+ mockExecutor,
  coresequencer.NewDummySequencer(),
  nil,
  blockManagerLogger,
@@ -221,24 +228,6 @@ func setupBlockManager(t *testing.T, ctx context.Context, workDir string, mainKV
 	return result, headerSyncService, dataSyncService
 }
 
-type mockExecutor struct{}
-
-func (m mockExecutor) InitChain(ctx context.Context, genesisTime time.Time, initialHeight uint64, chainID string) (stateRoot []byte, maxBytes uint64, err error) {
-	return bytesN(32), 10_000, nil
-}
-
-func (m mockExecutor) GetTxs(ctx context.Context) ([][]byte, error) {
-	panic("implement me")
-}
-
-func (m mockExecutor) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight uint64, timestamp time.Time, prevStateRoot []byte) (updatedStateRoot []byte, maxBytes uint64, err error) {
-	return bytesN(32), 10_000, nil
-}
-
-func (m mockExecutor) SetFinal(ctx context.Context, blockHeight uint64) error {
-	return nil
-}
-
 var rnd = rand.New(rand.NewSource(1)) //nolint:gosec // test code only
 
 func bytesN(n int) []byte {

pkg/rpc/server/server.go

@@ -276,8 +276,6 @@ func NewServiceHandler(store store.Store, peerManager p2p.P2PRPC, logger zerolog
 
 	mux := http.NewServeMux()
 
-	fmt.Println("Registering gRPC reflection service...")
-
 	compress1KB := connect.WithCompressMinBytes(1024)
 	reflector := grpcreflect.NewStaticReflector(
  rpc.StoreServiceName,