refactor(store): replace go-header store by ev-node store

.mockery.yaml

@@ -44,23 +44,11 @@ packages:
           filename: batch.go
   github.com/celestiaorg/go-header:
     interfaces:
-      Exchange:
-        config:
-          dir: ./test/mocks
-          pkgname: mocks
-          filename: external/hexchange.go
       Store:
         config:
           dir: ./test/mocks
           pkgname: mocks
           filename: external/hstore.go
-  github.com/evstack/ev-node/pkg/sync:
-    interfaces:
-      P2PExchange:
-        config:
-          dir: ./test/mocks
-          pkgname: mocks
-          filename: external/p2pexchange.go
   github.com/evstack/ev-node/block/internal/syncing:
     interfaces:
       DARetriever:

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Improve cache handling when there is a significant backlog of pending headers and data. ([#3030](https://github.com/evstack/ev-node/pull/3030))
 - Decrease MaxBytesSize to `5MB` to increase compatibility with public nodes. ([#3030](https://github.com/evstack/ev-node/pull/3030))
+- Replace `go-header` store by `ev-node` store. This avoid duplication of all blocks in `go-header` and `ev-node` store. Thanks to the cached store from #3030, this should improve p2p performance as well.
 
 ## v1.0.0-rc.1
 

apps/evm/cmd/rollback.go

@@ -3,19 +3,16 @@ package cmd
 import (
 	"bytes"
 	"context"
-	"errors"
 	"fmt"
 	"os"
 
 	"github.com/ethereum/go-ethereum/common"
 	ds "github.com/ipfs/go-datastore"
 	"github.com/spf13/cobra"
 
-	goheaderstore "github.com/celestiaorg/go-header/store"
 	"github.com/evstack/ev-node/execution/evm"
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/store"
-	"github.com/evstack/ev-node/types"
 )
 
 // NewRollbackCmd creates a command to rollback ev-node state by one height.
@@ -64,6 +61,9 @@ func NewRollbackCmd() *cobra.Command {
 			}
 
 			// rollback ev-node main state
+			// Note: With the unified store approach, the ev-node store is the single source of truth.
+			// The store adapters (HeaderStoreAdapter/DataStoreAdapter) read from this store,
+			// so rolling back the ev-node store automatically affects P2P sync operations.
 			if err := evolveStore.Rollback(goCtx, height, !syncNode); err != nil {
 				return fmt.Errorf("failed to rollback ev-node state: %w", err)
 			}
@@ -79,50 +79,12 @@ func NewRollbackCmd() *cobra.Command {
 				cmd.Printf("Rolled back execution layer to height %d\n", height)
 			}
 
-			// rollback ev-node goheader state
-			headerStore, err := goheaderstore.NewStore[*types.SignedHeader](
-				evolveDB,
-				goheaderstore.WithStorePrefix("headerSync"),
-				goheaderstore.WithMetrics(),
-			)
-			if err != nil {
-				return err
-			}
-
-			dataStore, err := goheaderstore.NewStore[*types.Data](
-				evolveDB,
-				goheaderstore.WithStorePrefix("dataSync"),
-				goheaderstore.WithMetrics(),
-			)
-			if err != nil {
-				return err
-			}
-
-			if err := headerStore.Start(goCtx); err != nil {
-				return fmt.Errorf("failed to start header store: %w", err)
-			}
-			defer headerStore.Stop(goCtx)
-
-			if err := dataStore.Start(goCtx); err != nil {
-				return fmt.Errorf("failed to start data store: %w", err)
-			}
-			defer dataStore.Stop(goCtx)
-
-			var errs error
-			if err := headerStore.DeleteRange(goCtx, height+1, headerStore.Height()); err != nil {
-				errs = errors.Join(errs, fmt.Errorf("failed to rollback header sync service state: %w", err))
-			}
-
-			if err := dataStore.DeleteRange(goCtx, height+1, dataStore.Height()); err != nil {
-				errs = errors.Join(errs, fmt.Errorf("failed to rollback data sync service state: %w", err))
-			}
-
 			cmd.Printf("Rolled back ev-node state to height %d\n", height)
 			if syncNode {
 				fmt.Println("Restart the node with the `--evnode.clear_cache` flag")
 			}
 
-			return errs
+			return nil
 		},
 	}
 

apps/evm/go.mod

@@ -8,7 +8,6 @@ replace (
 )
 
 require (
-	github.com/celestiaorg/go-header v0.8.1
 	github.com/ethereum/go-ethereum v1.16.8
 	github.com/evstack/ev-node v1.0.0-rc.1
 	github.com/evstack/ev-node/core v1.0.0-rc.1
@@ -30,6 +29,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.20.0 // indirect
 	github.com/boltdb/bolt v1.3.1 // indirect
+	github.com/celestiaorg/go-header v0.8.1 // indirect
 	github.com/celestiaorg/go-libp2p-messenger v0.2.2 // indirect
 	github.com/celestiaorg/go-square/merkle v0.0.0-20240627094109-7d01436067a3 // indirect
 	github.com/celestiaorg/go-square/v3 v3.0.2 // indirect

apps/testapp/cmd/rollback.go

@@ -2,15 +2,12 @@ package cmd
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	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/evstack/ev-node/types"
 
-	goheaderstore "github.com/celestiaorg/go-header/store"
 	"github.com/spf13/cobra"
 )
 
@@ -65,59 +62,24 @@ func NewRollbackCmd() *cobra.Command {
 			}
 
 			// rollback ev-node main state
+			// Note: With the unified store approach, the ev-node store is the single source of truth.
+			// The store adapters (HeaderStoreAdapter/DataStoreAdapter) read from this store,
+			// so rolling back the ev-node store automatically affects P2P sync operations.
 			if err := evolveStore.Rollback(goCtx, height, !syncNode); err != nil {
 				return fmt.Errorf("failed to rollback ev-node state: %w", err)
 			}
 
-			// rollback ev-node goheader state
-			headerStore, err := goheaderstore.NewStore[*types.SignedHeader](
-				evolveDB,
-				goheaderstore.WithStorePrefix("headerSync"),
-				goheaderstore.WithMetrics(),
-			)
-			if err != nil {
-				return err
-			}
-
-			dataStore, err := goheaderstore.NewStore[*types.Data](
-				evolveDB,
-				goheaderstore.WithStorePrefix("dataSync"),
-				goheaderstore.WithMetrics(),
-			)
-			if err != nil {
-				return err
-			}
-
-			if err := headerStore.Start(goCtx); err != nil {
-				return err
-			}
-			defer headerStore.Stop(goCtx)
-
-			if err := dataStore.Start(goCtx); err != nil {
-				return err
-			}
-			defer dataStore.Stop(goCtx)
-
-			var errs error
-			if err := headerStore.DeleteRange(goCtx, height+1, headerStore.Height()); err != nil {
-				errs = errors.Join(errs, fmt.Errorf("failed to rollback header sync service state: %w", err))
-			}
-
-			if err := dataStore.DeleteRange(goCtx, height+1, dataStore.Height()); err != nil {
-				errs = errors.Join(errs, fmt.Errorf("failed to rollback data sync service state: %w", err))
-			}
-
 			// rollback execution store
 			if err := executor.Rollback(goCtx, height); err != nil {
-				errs = errors.Join(errs, fmt.Errorf("rollback failed: %w", err))
+				return fmt.Errorf("failed to rollback executor state: %w", err)
 			}
 
 			fmt.Printf("Rolled back ev-node state to height %d\n", height)
 			if syncNode {
 				fmt.Println("Restart the node with the `--evnode.clear_cache` flag")
 			}
 
-			return errs
+			return nil
 		},
 	}
 

apps/testapp/go.mod

@@ -3,7 +3,6 @@ module github.com/evstack/ev-node/apps/testapp
 go 1.25.0
 
 require (
-	github.com/celestiaorg/go-header v0.8.1
 	github.com/evstack/ev-node v1.0.0-rc.1
 	github.com/evstack/ev-node/core v1.0.0-rc.1
 	github.com/ipfs/go-datastore v0.9.0
@@ -19,6 +18,7 @@ require (
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/boltdb/bolt v1.3.1 // indirect
+	github.com/celestiaorg/go-header v0.8.1 // indirect
 	github.com/celestiaorg/go-libp2p-messenger v0.2.2 // indirect
 	github.com/celestiaorg/go-square/merkle v0.0.0-20240627094109-7d01436067a3 // indirect
 	github.com/celestiaorg/go-square/v3 v3.0.2 // indirect

node/failover.go

@@ -46,7 +46,6 @@ func newSyncMode(
 	da block.DAClient,
 	logger zerolog.Logger,
 	rktStore store.Store,
-	mainKV ds.Batching,
 	blockMetrics *block.Metrics,
 	nodeOpts NodeOptions,
 	raftNode *raft.Node,
@@ -66,7 +65,7 @@ func newSyncMode(
 			raftNode,
 		)
 	}
-	return setupFailoverState(nodeConfig, nodeKey, rootDB, daStore, genesis, logger, mainKV, rktStore, blockComponentsFn, raftNode)
+	return setupFailoverState(nodeConfig, nodeKey, rootDB, daStore, genesis, logger, rktStore, blockComponentsFn, raftNode)
 }
 
 func newAggregatorMode(
@@ -81,7 +80,6 @@ func newAggregatorMode(
 	da block.DAClient,
 	logger zerolog.Logger,
 	rktStore store.Store,
-	mainKV ds.Batching,
 	blockMetrics *block.Metrics,
 	nodeOpts NodeOptions,
 	raftNode *raft.Node,
@@ -105,7 +103,7 @@ func newAggregatorMode(
 		)
 	}
 
-	return setupFailoverState(nodeConfig, nodeKey, rootDB, daStore, genesis, logger, mainKV, rktStore, blockComponentsFn, raftNode)
+	return setupFailoverState(nodeConfig, nodeKey, rootDB, daStore, genesis, logger, rktStore, blockComponentsFn, raftNode)
 }
 
 func setupFailoverState(
@@ -115,7 +113,6 @@ func setupFailoverState(
 	daStore store.Store,
 	genesis genesispkg.Genesis,
 	logger zerolog.Logger,
-	mainKV ds.Batching,
 	rktStore store.Store,
 	buildComponentsFn func(headerSyncService *evsync.HeaderSyncService, dataSyncService *evsync.DataSyncService) (*block.Components, error),
 	raftNode *raft.Node,
@@ -125,12 +122,12 @@ func setupFailoverState(
 		return nil, err
 	}
 
-	headerSyncService, err := evsync.NewHeaderSyncService(mainKV, daStore, nodeConfig, genesis, p2pClient, logger.With().Str("component", "HeaderSyncService").Logger())
+	headerSyncService, err := evsync.NewHeaderSyncService(daStore, nodeConfig, genesis, p2pClient, logger.With().Str("component", "HeaderSyncService").Logger())
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
 	}
 
-	dataSyncService, err := evsync.NewDataSyncService(mainKV, daStore, nodeConfig, genesis, p2pClient, logger.With().Str("component", "DataSyncService").Logger())
+	dataSyncService, err := evsync.NewDataSyncService(daStore, nodeConfig, genesis, p2pClient, logger.With().Str("component", "DataSyncService").Logger())
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing DataSyncService: %w", err)
 	}

node/full.go

@@ -107,12 +107,12 @@ func newFullNode(
 		logger.Info().Msg("Starting aggregator-MODE")
 		nodeConfig.Node.Aggregator = true
 		nodeConfig.P2P.Peers = "" // peers are not supported in aggregator mode
-		return newAggregatorMode(nodeConfig, nodeKey, signer, genesis, database, evstore, exec, sequencer, daClient, logger, evstore, mainKV, blockMetrics, nodeOpts, raftNode)
+		return newAggregatorMode(nodeConfig, nodeKey, signer, genesis, database, evstore, exec, sequencer, daClient, logger, evstore, blockMetrics, nodeOpts, raftNode)
 	}
 	followerFactory := func() (raftpkg.Runnable, error) {
 		logger.Info().Msg("Starting sync-MODE")
 		nodeConfig.Node.Aggregator = false
-		return newSyncMode(nodeConfig, nodeKey, genesis, database, evstore, exec, daClient, logger, evstore, mainKV, blockMetrics, nodeOpts, raftNode)
+		return newSyncMode(nodeConfig, nodeKey, genesis, database, evstore, exec, daClient, logger, evstore, blockMetrics, nodeOpts, raftNode)
 	}
 
 	// Initialize raft node if enabled (for both aggregator and sync nodes)

node/light.go

@@ -57,7 +57,7 @@ func newLightNode(
 		return nil, fmt.Errorf("failed to create cached store: %w", err)
 	}
 
-	headerSyncService, err := sync.NewHeaderSyncService(database, cachedStore, conf, genesis, p2pClient, componentLogger)
+	headerSyncService, err := sync.NewHeaderSyncService(cachedStore, conf, genesis, p2pClient, componentLogger)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
 	}

pkg/store/README.md

@@ -176,3 +176,7 @@ batch.Delete(ctx, key3)
 // Commit all operations atomically
 err = batch.Commit(ctx)
 ```
+
+## Store Adapters for P2P Integration
+
+The store package provides adapter implementations that wrap the ev-node store to satisfy the `header.Store[H]` interface from the `go-header` library. This enables the ev-node store to be used directly by go-header's P2P infrastructure, eliminating data duplication.