Implementation

 Please enter the commit message for your changes. Lines starting

Author: alpe

.mockery.yaml

@@ -65,8 +65,3 @@ packages:
           dir: ./block/internal/common
           pkgname: common
           filename: broadcaster_mock.go
-      p2pHandler:
-        config:
-          dir: ./block/internal/syncing
-          pkgname: syncing
-          filename: syncer_mock.go

apps/grpc/single/cmd/run.go

@@ -14,7 +14,6 @@ import (
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/config"
 	rollgenesis "github.com/evstack/ev-node/pkg/genesis"
-	"github.com/evstack/ev-node/pkg/p2p"
 	"github.com/evstack/ev-node/pkg/p2p/key"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/sequencers/single"
@@ -100,14 +99,8 @@ The execution client must implement the Evolve execution gRPC interface.`,
 			return err
 		}
 
-		// Create P2P client
-		p2pClient, err := p2p.NewClient(nodeConfig.P2P, nodeKey.PrivKey, datastore, genesis.ChainID, logger, nil)
-		if err != nil {
-			return err
-		}
-
 		// Start the node
-		return rollcmd.StartNode(logger, cmd, executor, sequencer, &daJrpc.DA, p2pClient, datastore, nodeConfig, genesis, node.NodeOptions{})
+		return rollcmd.StartNode(logger, cmd, executor, sequencer, &daJrpc.DA, nodeKey, datastore, nodeConfig, genesis, node.NodeOptions{})
 	},
 }
 

block/components.go

@@ -119,6 +119,11 @@ func (bc *Components) Stop() error {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop submitter: %w", err))
 		}
 	}
+	if bc.Cache != nil {
+		if err := bc.Cache.SaveToDisk(); err != nil {
+			errs = errors.Join(errs, fmt.Errorf("failed to save caches: %w", err))
+		}
+	}
 
 	return errs
 }
@@ -139,7 +144,6 @@ func NewSyncComponents(
 	blockOpts BlockOptions,
 	raftNode common.RaftNode,
 ) (*Components, error) {
-	logger.Info().Msg("Starting in sync-mode")
 	cacheManager, err := cache.NewManager(config, store, logger)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create cache manager: %w", err)
@@ -205,7 +209,6 @@ func NewAggregatorComponents(
 	blockOpts BlockOptions,
 	raftNode common.RaftNode,
 ) (*Components, error) {
-	logger.Info().Msg("Starting in aggregator-mode")
 	cacheManager, err := cache.NewManager(config, store, logger)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create cache manager: %w", err)

block/internal/common/broadcaster_mock.go

@@ -7,8 +7,9 @@ import (
 	"github.com/celestiaorg/go-header"
 )
 
-// broadcaster interface for P2P broadcasting
+// Broadcaster interface for P2P broadcasting
 type Broadcaster[H header.Header[H]] interface {
 	WriteToStoreAndBroadcast(ctx context.Context, payload H, opts ...pubsub.PubOpt) error
 	Store() header.Store[H]
+	Height() uint64
 }

block/internal/common/expected_interfaces.go

@@ -9,14 +9,9 @@ import (
 // RaftNode interface for raft consensus integration
 type RaftNode interface {
 	IsLeader() bool
-	NodeID() string
 	GetState() *raft.RaftBlockState
 
 	Broadcast(ctx context.Context, state *raft.RaftBlockState) error
 
 	SetApplyCallback(ch chan<- raft.RaftApplyMsg)
-	Shutdown() error
-
-	AddPeer(nodeID, addr string) error
-	RemovePeer(nodeID string) error
 }

block/internal/common/raft.go

@@ -10,6 +10,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/evstack/ev-node/pkg/raft"
 	"github.com/ipfs/go-datastore"
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
@@ -20,7 +21,6 @@ import (
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
-	"github.com/evstack/ev-node/pkg/raft"
 	"github.com/evstack/ev-node/pkg/signer"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
@@ -100,6 +100,9 @@ func NewExecutor(
 	if !bytes.Equal(addr, genesis.ProposerAddress) {
 		return nil, common.ErrNotProposer
 	}
+	if raftNode != nil && reflect.ValueOf(raftNode).IsNil() {
+		raftNode = nil
+	}
 
 	return &Executor{
 		store:             store,
@@ -210,6 +213,12 @@ func (e *Executor) initializeState() error {
 		}
 	}
 
+	if e.raftNode != nil {
+		// ensure node is fully synced before producing any blocks
+		if raftState := e.raftNode.GetState(); raftState != nil && raftState.Height != state.LastBlockHeight {
+			return fmt.Errorf("invalid state: node is not synced with the chain: raft %d != %d state", raftState.Height, state.LastBlockHeight)
+		}
+	}
 	e.setLastState(state)
 
 	// Initialize store height using batch for atomicity
@@ -318,7 +327,7 @@ func (e *Executor) produceBlock() error {
 	}()
 
 	// Check raft leadership if raft is enabled
-	if !reflect.ValueOf(e.raftNode).IsNil() && !e.raftNode.IsLeader() {
+	if e.raftNode != nil && !e.raftNode.IsLeader() {
 		return errors.New("not raft leader")
 	}
 
@@ -422,16 +431,8 @@ func (e *Executor) produceBlock() error {
 		return fmt.Errorf("failed to update state: %w", err)
 	}
 
-	if err := batch.Commit(); err != nil {
-		return fmt.Errorf("failed to commit batch: %w", err)
-	}
-
-	// Update in-memory state after successful commit
-	e.setLastState(newState)
-	e.metrics.Height.Set(float64(newState.LastBlockHeight))
-
-	// Propose block to raft before p2p broadcast if raft is enabled
-	if !reflect.ValueOf(e.raftNode).IsNil() {
+	// Propose block to raft to share state in the cluster
+	if e.raftNode != nil {
 		headerBytes, err := header.MarshalBinary()
 		if err != nil {
 			return fmt.Errorf("failed to marshal header: %w", err)
@@ -452,8 +453,19 @@ func (e *Executor) produceBlock() error {
 			return fmt.Errorf("failed to propose block to raft: %w", err)
 		}
 		e.logger.Debug().Uint64("height", newHeight).Msg("proposed block to raft")
+
+	}
+	if err := batch.Commit(); err != nil {
+		return fmt.Errorf("failed to commit batch: %w", err)
+	}
+	if err := e.store.Sync(context.Background()); err != nil {
+		return fmt.Errorf("failed to sync store: %w", err)
 	}
 
+	// Update in-memory state after successful commit
+	e.setLastState(newState)
+	e.metrics.Height.Set(float64(newState.LastBlockHeight))
+
 	// broadcast header and data to P2P network
 	g, ctx := errgroup.WithContext(e.ctx)
 	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) })
@@ -694,6 +706,15 @@ func (e *Executor) recordBlockMetrics(data *types.Data) {
 	e.metrics.CommittedHeight.Set(float64(data.Metadata.Height))
 }
 
+// IsSynced checks if the last block height in the stored state matches the expected height and returns true if they are equal.
+func (e *Executor) IsSynced(expHeight uint64) bool {
+	state, err := e.store.GetState(e.ctx)
+	if err != nil {
+		return false
+	}
+	return state.LastBlockHeight == expHeight
+}
+
 // BatchData represents batch data from sequencer
 type BatchData struct {
 	*coresequencer.Batch

block/internal/executing/executor.go

@@ -2,6 +2,7 @@ package executing
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -256,6 +257,8 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	cfg.Node.BlockTime = config.DurationWrapper{Duration: 10 * time.Millisecond}
 	cfg.Node.MaxPendingHeadersAndData = 1000
 
+	const numBlocks = 5
+
 	gen := genesis.Genesis{
 		ChainID:         "test-chain",
 		InitialHeight:   1,
@@ -294,24 +297,29 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 		Return(initStateRoot, uint64(1024), nil).Once()
 	require.NoError(t, exec1.initializeState())
 
-	exec1.ctx, exec1.cancel = context.WithCancel(context.Background())
+	exec1.ctx, exec1.cancel = context.WithCancel(t.Context())
 
-	// Produce first block
+	// Produce n blocks
 	mockSeq1.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
 		RunAndReturn(func(ctx context.Context, req coreseq.GetNextBatchRequest) (*coreseq.GetNextBatchResponse, error) {
 			return &coreseq.GetNextBatchResponse{
 				Batch: &coreseq.Batch{
-					Transactions: [][]byte{[]byte("tx1")},
+					Transactions: [][]byte{[]byte("any_tx")},
 				},
 				Timestamp: time.Now(),
 			}, nil
-		}).Once()
+		}).Times(numBlocks)
 
-	mockExec1.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(1), mock.AnythingOfType("time.Time"), initStateRoot).
-		Return([]byte("new_root_1"), uint64(1024), nil).Once()
+	lastStateRoot := initStateRoot
+	for i := range numBlocks {
+		newStateRoot := []byte(fmt.Sprintf("new_root_%d", i+1))
+		mockExec1.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, gen.InitialHeight+uint64(i), mock.AnythingOfType("time.Time"), lastStateRoot).
+			Return(newStateRoot, uint64(1024), nil).Once()
+		lastStateRoot = newStateRoot
 
-	err = exec1.produceBlock()
-	require.NoError(t, err)
+		require.NoError(t, exec1.produceBlock())
+	}
+	require.Equal(t, uint64(numBlocks), exec1.GetLastState().LastBlockHeight)
 
 	// Stop first executor
 	exec1.cancel()
@@ -343,12 +351,12 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	require.NoError(t, exec2.initializeState())
-	exec2.ctx, exec2.cancel = context.WithCancel(context.Background())
+	exec2.ctx, exec2.cancel = context.WithCancel(t.Context())
 	defer exec2.cancel()
 
 	// Verify state loaded correctly
 	state := exec2.getLastState()
-	assert.Equal(t, uint64(1), state.LastBlockHeight)
+	require.Equal(t, uint64(numBlocks), state.LastBlockHeight)
 
 	// Now produce next block - should go through normal sequencer flow since no pending block
 	mockSeq2.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
@@ -361,16 +369,16 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 			}, nil
 		}).Once()
 
-	mockExec2.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(2), mock.AnythingOfType("time.Time"), []byte("new_root_1")).
-		Return([]byte("new_root_2"), uint64(1024), nil).Once()
+	mockExec2.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(numBlocks+1), mock.AnythingOfType("time.Time"), lastStateRoot).
+		Return([]byte("new_root_after_restart"), uint64(1024), nil).Once()
 
 	err = exec2.produceBlock()
 	require.NoError(t, err)
 
 	// Verify normal operation
 	h, err := memStore.Height(context.Background())
 	require.NoError(t, err)
-	assert.Equal(t, uint64(2), h)
+	assert.Equal(t, uint64(numBlocks+1), h)
 
 	// Verify sequencer was called (normal flow)
 	mockSeq2.AssertCalled(t, "GetNextBatch", mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest"))

block/internal/executing/executor_restart_test.go

@@ -104,6 +104,7 @@ func (s *Submitter) Start(ctx context.Context) error {
 
 	// Start DA submission loop if signer is available (aggregator nodes only)
 	if s.signer != nil {
+		s.logger.Info().Msg("starting DA submission loop")
 		s.wg.Add(1)
 		go func() {
 			defer s.wg.Done()

block/internal/submitting/submitter.go

@@ -23,7 +23,6 @@ func (e eventProcessorFn) handle(ctx context.Context, event common.DAHeightEvent
 }
 
 type raftRetriever struct {
-	// Raft consensus
 	raftNode       common.RaftNode
 	wg             sync.WaitGroup
 	logger         zerolog.Logger