[DO NOT MERGRE] Re-use p2p-client

apps/evm/go.mod

@@ -2,10 +2,10 @@ module github.com/evstack/ev-node/apps/evm
 
 go 1.25.6
 
-//replace (
-//	github.com/evstack/ev-node => ../../
-//	github.com/evstack/ev-node/execution/evm => ../../execution/evm
-//)
+replace (
+	github.com/evstack/ev-node => ../../
+	github.com/evstack/ev-node/execution/evm => ../../execution/evm
+)
 
 require (
 	github.com/ethereum/go-ethereum v1.16.8

apps/evm/go.sum

@@ -411,12 +411,8 @@ github.com/ethereum/go-ethereum v1.16.8 h1:LLLfkZWijhR5m6yrAXbdlTeXoqontH+Ga2f9i
 github.com/ethereum/go-ethereum v1.16.8/go.mod h1:Fs6QebQbavneQTYcA39PEKv2+zIjX7rPUZ14DER46wk=
 github.com/ethereum/go-verkle v0.2.2 h1:I2W0WjnrFUIzzVPwm8ykY+7pL2d4VhlsePn4j7cnFk8=
 github.com/ethereum/go-verkle v0.2.2/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk=
-github.com/evstack/ev-node v1.0.0-rc.4 h1:Ju7pSETFdadBZxmAj0//4z7hHkXbSRDy9iTzhF60Dew=
-github.com/evstack/ev-node v1.0.0-rc.4/go.mod h1:xGCH5NCdGiYk6v3GVPm4NhzAtcKQgnaVnORg8b4tbOk=
 github.com/evstack/ev-node/core v1.0.0-rc.1 h1:Dic2PMUMAYUl5JW6DkDj6HXDEWYzorVJQuuUJOV0FjE=
 github.com/evstack/ev-node/core v1.0.0-rc.1/go.mod h1:n2w/LhYQTPsi48m6lMj16YiIqsaQw6gxwjyJvR+B3sY=
-github.com/evstack/ev-node/execution/evm v1.0.0-rc.3 h1:3o8H1TNywnst56lo2RlS2SXulDfp9yZJtkYYh7ZJrdM=
-github.com/evstack/ev-node/execution/evm v1.0.0-rc.3/go.mod h1:VUEEklKoclg45GL7dzLoDwu3UQ4ptT3rF8bw5zUmnRk=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw=

apps/testapp/go.mod

@@ -2,7 +2,7 @@ module github.com/evstack/ev-node/apps/testapp
 
 go 1.25.6
 
-//replace github.com/evstack/ev-node => ../../
+replace github.com/evstack/ev-node => ../../
 
 require (
 	github.com/evstack/ev-node v1.0.0-rc.4

apps/testapp/go.sum

@@ -367,8 +367,6 @@ github.com/envoyproxy/protoc-gen-validate v0.10.0/go.mod h1:DRjgyB0I43LtJapqN6Ni
 github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
 github.com/envoyproxy/protoc-gen-validate v1.0.1/go.mod h1:0vj8bNkYbSTNS2PIyH87KZaeN4x9zpL9Qt8fQC7d+vs=
 github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
-github.com/evstack/ev-node v1.0.0-rc.4 h1:Ju7pSETFdadBZxmAj0//4z7hHkXbSRDy9iTzhF60Dew=
-github.com/evstack/ev-node v1.0.0-rc.4/go.mod h1:xGCH5NCdGiYk6v3GVPm4NhzAtcKQgnaVnORg8b4tbOk=
 github.com/evstack/ev-node/core v1.0.0-rc.1 h1:Dic2PMUMAYUl5JW6DkDj6HXDEWYzorVJQuuUJOV0FjE=
 github.com/evstack/ev-node/core v1.0.0-rc.1/go.mod h1:n2w/LhYQTPsi48m6lMj16YiIqsaQw6gxwjyJvR+B3sY=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=

block/components_test.go

@@ -175,7 +175,7 @@ func TestNewAggregatorComponents_Creation(t *testing.T) {
 		zerolog.Nop(),
 		NopMetrics(),
 		DefaultBlockOptions(),
-		nil,
+		nil, // raftNode
 	)
 
 	require.NoError(t, err)
@@ -258,7 +258,7 @@ func TestExecutor_RealExecutionClientFailure_StopsNode(t *testing.T) {
 			zerolog.Nop(),
 			NopMetrics(),
 			DefaultBlockOptions(),
-			nil,
+			nil, // raftNode
 		)
 		require.NoError(t, err)
 

block/internal/executing/executor.go

@@ -429,12 +429,12 @@ func (e *Executor) ProduceBlock(ctx context.Context) error {
 
 	// Check if there's an already stored block at the newHeight
 	// If there is use that instead of creating a new block
-	pendingHeader, pendingData, err := e.store.GetBlockData(ctx, newHeight)
-	if err == nil {
+	pendingHeader, pendingData, err := e.getPendingBlock(ctx)
+	if err == nil && pendingHeader != nil && pendingHeader.Height() == newHeight {
 		e.logger.Info().Uint64("height", newHeight).Msg("using pending block")
 		header = pendingHeader
 		data = pendingData
-	} else if !errors.Is(err, datastore.ErrNotFound) {
+	} else if err != nil && !errors.Is(err, datastore.ErrNotFound) {
 		return fmt.Errorf("failed to get block data: %w", err)
 	} else {
 		// get batch from sequencer
@@ -452,18 +452,9 @@ func (e *Executor) ProduceBlock(ctx context.Context) error {
 		if err != nil {
 			return fmt.Errorf("failed to create block: %w", err)
 		}
-
-		// saved early for crash recovery, will be overwritten later with the final signature
-		batch, err := e.store.NewBatch(ctx)
-		if err != nil {
-			return fmt.Errorf("failed to create batch for early save: %w", err)
-		}
-		if err = batch.SaveBlockData(header, data, &types.Signature{}); err != nil {
+		if err := e.savePendingBlock(ctx, header, data); err != nil {
 			return fmt.Errorf("failed to save block data: %w", err)
 		}
-		if err = batch.Commit(); err != nil {
-			return fmt.Errorf("failed to commit early save batch: %w", err)
-		}
 	}
 
 	if e.raftNode != nil && !e.raftNode.HasQuorum() {
@@ -535,6 +526,10 @@ func (e *Executor) ProduceBlock(ctx context.Context) error {
 		}
 		e.logger.Debug().Uint64("height", newHeight).Msg("proposed block to raft")
 	}
+	if err := e.deletePendingBlock(e.ctx, batch); err != nil {
+		e.logger.Warn().Err(err).Uint64("height", newHeight).Msg("failed to delete pending block metadata")
+	}
+
 	if err := batch.Commit(); err != nil {
 		return fmt.Errorf("failed to commit batch: %w", err)
 	}

block/internal/executing/executor_restart_test.go

@@ -79,7 +79,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, exec1.initializeState())
 
 	// Set up context for first executor
-	exec1.ctx, exec1.cancel = context.WithCancel(context.Background())
+	exec1.ctx, exec1.cancel = context.WithCancel(t.Context())
 
 	// First executor produces a block normally
 	mockSeq1.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
@@ -101,12 +101,12 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify first block was produced
-	h1, err := memStore.Height(context.Background())
+	h1, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(1), h1)
 
 	// Store the produced block data for later verification
-	originalHeader, originalData, err := memStore.GetBlockData(context.Background(), 1)
+	originalHeader, originalData, err := memStore.GetBlockData(t.Context(), 1)
 	require.NoError(t, err)
 	assert.Equal(t, 2, len(originalData.Txs), "first block should have 2 transactions")
 
@@ -158,11 +158,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	pendingHeader.DataHash = pendingData.DACommitment()
 
 	// Save pending block data (this is what would happen during a crash)
-	batch, err := memStore.NewBatch(context.Background())
-	require.NoError(t, err)
-	err = batch.SaveBlockData(pendingHeader, pendingData, &types.Signature{})
-	require.NoError(t, err)
-	err = batch.Commit()
+	err = exec1.savePendingBlock(t.Context(), pendingHeader, pendingData)
 	require.NoError(t, err)
 
 	// Stop first executor (simulating crash/restart)
@@ -199,7 +195,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, exec2.initializeState())
 
 	// Set up context for second executor
-	exec2.ctx, exec2.cancel = context.WithCancel(context.Background())
+	exec2.ctx, exec2.cancel = context.WithCancel(t.Context())
 	defer exec2.cancel()
 
 	// Verify that the state is at height 1 (pending block at height 2 wasn't committed)
@@ -221,12 +217,12 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify height advanced to 2
-	h2, err := memStore.Height(context.Background())
+	h2, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(2), h2, "height should advance to 2 using pending block")
 
 	// Verify the block at height 2 matches the pending block data
-	finalHeader, finalData, err := memStore.GetBlockData(context.Background(), 2)
+	finalHeader, finalData, err := memStore.GetBlockData(t.Context(), 2)
 	require.NoError(t, err)
 	assert.Equal(t, 3, len(finalData.Txs), "should use pending block with 3 transactions")
 	assert.Equal(t, []byte("pending_tx1"), []byte(finalData.Txs[0]))
@@ -388,7 +384,7 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify normal operation
-	h, err := memStore.Height(context.Background())
+	h, err := memStore.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(numBlocks+1), h)
 

block/internal/executing/pending.go

@@ -0,0 +1,89 @@
+package executing
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/evstack/ev-node/pkg/store"
+	"github.com/evstack/ev-node/types"
+	ds "github.com/ipfs/go-datastore"
+)
+
+const (
+	headerKey = "pending_header"
+	dataKey   = "pending_data"
+)
+
+// getPendingBlock retrieves the pending block from metadata if it exists
+func (e *Executor) getPendingBlock(ctx context.Context) (*types.SignedHeader, *types.Data, error) {
+	headerBytes, err := e.store.GetMetadata(ctx, headerKey)
+	if err != nil {
+		if errors.Is(err, ds.ErrNotFound) {
+			return nil, nil, nil
+		}
+		return nil, nil, err
+	}
+
+	dataBytes, err := e.store.GetMetadata(ctx, dataKey)
+	if err != nil {
+		if errors.Is(err, ds.ErrNotFound) {
+			return nil, nil, nil
+		}
+		return nil, nil, err
+	}
+
+	header := new(types.SignedHeader)
+	if err := header.UnmarshalBinary(headerBytes); err != nil {
+		return nil, nil, fmt.Errorf("unmarshal pending header: %w", err)
+	}
+
+	data := new(types.Data)
+	if err := data.UnmarshalBinary(dataBytes); err != nil {
+		return nil, nil, fmt.Errorf("unmarshal pending data: %w", err)
+	}
+	return header, data, nil
+}
+
+// savePendingBlock saves a block to metadata as pending
+func (e *Executor) savePendingBlock(ctx context.Context, header *types.SignedHeader, data *types.Data) error {
+	headerBytes, err := header.MarshalBinary()
+	if err != nil {
+		return fmt.Errorf("marshal header: %w", err)
+	}
+
+	dataBytes, err := data.MarshalBinary()
+	if err != nil {
+		return fmt.Errorf("marshal data: %w", err)
+	}
+
+	batch, err := e.store.NewBatch(ctx)
+	if err != nil {
+		return fmt.Errorf("create batch for early save: %w", err)
+	}
+
+	if err := batch.Put(ds.NewKey(store.GetMetaKey(headerKey)), headerBytes); err != nil {
+		return fmt.Errorf("save pending header: %w", err)
+	}
+
+	if err := batch.Put(ds.NewKey(store.GetMetaKey(dataKey)), dataBytes); err != nil {
+		return fmt.Errorf("save pending data: %w", err)
+	}
+
+	if err := batch.Commit(); err != nil {
+		return fmt.Errorf("commit pending block: %w", err)
+	}
+	return nil
+}
+
+// deletePendingBlock removes pending block metadata
+func (e *Executor) deletePendingBlock(ctx context.Context, batch store.Batch) error {
+	if err := batch.Delete(ds.NewKey(store.GetMetaKey(headerKey))); err != nil {
+		return fmt.Errorf("delete pending header: %w", err)
+	}
+
+	if err := batch.Delete(ds.NewKey(store.GetMetaKey(dataKey))); err != nil {
+		return fmt.Errorf("delete pending data: %w", err)
+	}
+	return nil
+}