Backports

Author: alpe

.mockery.yaml

@@ -30,6 +30,11 @@ packages:
           dir: ./test/mocks
           pkgname: mocks
           filename: store.go
+      Batch:
+        config:
+          dir: ./test/mocks
+          pkgname: mocks
+          filename: batch.go
   github.com/celestiaorg/go-header:
     interfaces:
       Store:

block/internal/cache/pending_data.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 
 	"github.com/rs/zerolog"
 
@@ -28,8 +29,17 @@ type PendingData struct {
 	base *pendingBase[*types.Data]
 }
 
+var errInFlightData = errors.New("inflight data")
+
 func fetchData(ctx context.Context, store store.Store, height uint64) (*types.Data, error) {
 	_, data, err := store.GetBlockData(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	// in the executor, WIP data is temporary stored. skip them until the process is completed
+	if data.Height() == 0 {
+		return nil, errInFlightData
+	}
 	return data, err
 }
 

block/internal/cache/pending_headers.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 
 	"github.com/rs/zerolog"
 
@@ -25,8 +26,17 @@ type PendingHeaders struct {
 	base *pendingBase[*types.SignedHeader]
 }
 
+var errInFlightHeader = errors.New("inflight header")
+
 func fetchSignedHeader(ctx context.Context, store storepkg.Store, height uint64) (*types.SignedHeader, error) {
 	header, err := store.GetHeader(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	// in the executor, WIP headers are temporary stored. skip them until the process is completed
+	if header.Height() == 0 {
+		return nil, errInFlightHeader
+	}
 	return header, err
 }
 

block/internal/executing/executor.go

@@ -227,6 +227,9 @@ func (e *Executor) initializeState() error {
 	if err := batch.SetHeight(state.LastBlockHeight); err != nil {
 		return fmt.Errorf("failed to set store height: %w", err)
 	}
+	if err := batch.UpdateState(state); err != nil {
+		return fmt.Errorf("failed to update state: %w", err)
+	}
 	if err := batch.Commit(); err != nil {
 		return fmt.Errorf("failed to commit batch: %w", err)
 	}
@@ -236,7 +239,8 @@ func (e *Executor) initializeState() error {
 
 	// Sync execution layer with store on startup
 	execReplayer := common.NewReplayer(e.store, e.exec, e.genesis, e.logger)
-	if err := execReplayer.SyncToHeight(e.ctx, state.LastBlockHeight); err != nil {
+	syncTargetHeight := state.LastBlockHeight
+	if err := execReplayer.SyncToHeight(e.ctx, syncTargetHeight); err != nil {
 		e.sendCriticalError(fmt.Errorf("failed to sync execution layer: %w", err))
 		return fmt.Errorf("failed to sync execution layer: %w", err)
 	}
@@ -281,7 +285,7 @@ func (e *Executor) executionLoop() {
 	}
 	txsAvailable := false
 
-	for {
+	for e.ctx.Err() == nil {
 		select {
 		case <-e.ctx.Done():
 			return
@@ -316,6 +320,10 @@ func (e *Executor) executionLoop() {
 
 // ProduceBlock creates, validates, and stores a new block.
 func (e *Executor) ProduceBlock(ctx context.Context) error {
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
+
 	start := time.Now()
 	defer func() {
 		if e.metrics.OperationDuration["block_production"] != nil {
@@ -582,7 +590,7 @@ func (e *Executor) CreateBlock(ctx context.Context, height uint64, batchData *Ba
 	}
 
 	for i, tx := range batchData.Transactions {
-		data.Txs[i] = types.Tx(tx)
+		data.Txs[i] = tx
 	}
 
 	// Set data hash

block/internal/submitting/submitter.go

@@ -108,6 +108,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()
@@ -130,7 +131,18 @@ func (s *Submitter) Stop() error {
 	if s.cancel != nil {
 		s.cancel()
 	}
-	s.wg.Wait()
+	// Wait for goroutines to finish with a timeout to prevent hanging
+	done := make(chan struct{})
+	go func() {
+		s.wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		// All goroutines finished cleanly
+	case <-time.After(5 * time.Second):
+		s.logger.Warn().Msg("submitter shutdown timed out waiting for goroutines, proceeding anyway")
+	}
 	s.logger.Info().Msg("submitter stopped")
 	return nil
 }
@@ -153,7 +165,9 @@ func (s *Submitter) daSubmissionLoop() {
 				s.logger.Debug().Time("t", time.Now()).Uint64("headers", headersNb).Msg("Submitting headers")
 				if s.headerSubmissionMtx.TryLock() {
 					s.logger.Debug().Time("t", time.Now()).Uint64("headers", headersNb).Msg("Header submission in progress")
+					s.wg.Add(1)
 					go func() {
+						defer s.wg.Done()
 						defer func() {
 							s.logger.Debug().Time("t", time.Now()).Uint64("headers", headersNb).Msg("Header submission completed")
 							s.headerSubmissionMtx.Unlock()
@@ -177,7 +191,9 @@ func (s *Submitter) daSubmissionLoop() {
 				s.logger.Debug().Time("t", time.Now()).Uint64("data", dataNb).Msg("Submitting data")
 				if s.dataSubmissionMtx.TryLock() {
 					s.logger.Debug().Time("t", time.Now()).Uint64("data", dataNb).Msg("Data submission in progress")
+					s.wg.Add(1)
 					go func() {
+						defer s.wg.Done()
 						defer func() {
 							s.logger.Debug().Time("t", time.Now()).Uint64("data", dataNb).Msg("Data submission completed")
 							s.dataSubmissionMtx.Unlock()

block/internal/syncing/assert.go

@@ -0,0 +1,43 @@
+package syncing
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/types"
+)
+
+func assertExpectedProposer(genesis genesis.Genesis, proposerAddr []byte) error {
+	if string(proposerAddr) != string(genesis.ProposerAddress) {
+		return fmt.Errorf("unexpected proposer: got %x, expected %x",
+			proposerAddr, genesis.ProposerAddress)
+	}
+	return nil
+}
+
+func assertValidSignedData(signedData *types.SignedData, genesis genesis.Genesis) error {
+	if signedData == nil || signedData.Txs == nil {
+		return errors.New("empty signed data")
+	}
+
+	if err := assertExpectedProposer(genesis, signedData.Signer.Address); err != nil {
+		return err
+	}
+
+	dataBytes, err := signedData.Data.MarshalBinary()
+	if err != nil {
+		return fmt.Errorf("failed to get signed data payload: %w", err)
+	}
+
+	valid, err := signedData.Signer.PubKey.Verify(dataBytes, signedData.Signature)
+	if err != nil {
+		return fmt.Errorf("failed to verify signature: %w", err)
+	}
+
+	if !valid {
+		return fmt.Errorf("invalid signature")
+	}
+
+	return nil
+}

block/internal/syncing/da_retriever.go

@@ -346,38 +346,12 @@ func (r *daRetriever) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 
 // assertExpectedProposer validates the proposer address
 func (r *daRetriever) assertExpectedProposer(proposerAddr []byte) error {
-	if string(proposerAddr) != string(r.genesis.ProposerAddress) {
-		return fmt.Errorf("unexpected proposer: got %x, expected %x",
-			proposerAddr, r.genesis.ProposerAddress)
-	}
-	return nil
+	return assertExpectedProposer(r.genesis, proposerAddr)
 }
 
 // assertValidSignedData validates signed data using the configured signature provider
 func (r *daRetriever) assertValidSignedData(signedData *types.SignedData) error {
-	if signedData == nil || signedData.Txs == nil {
-		return errors.New("empty signed data")
-	}
-
-	if err := r.assertExpectedProposer(signedData.Signer.Address); err != nil {
-		return err
-	}
-
-	dataBytes, err := signedData.Data.MarshalBinary()
-	if err != nil {
-		return fmt.Errorf("failed to get signed data payload: %w", err)
-	}
-
-	valid, err := signedData.Signer.PubKey.Verify(dataBytes, signedData.Signature)
-	if err != nil {
-		return fmt.Errorf("failed to verify signature: %w", err)
-	}
-
-	if !valid {
-		return fmt.Errorf("invalid signature")
-	}
-
-	return nil
+	return assertValidSignedData(signedData, r.genesis)
 }
 
 // isEmptyDataExpected checks if empty data is expected for a header