refactor(block): move da in common, simplify start flow (#2650)

ref: #2644 (comment)

Author: julienrbrt

block/internal/syncing/da_handler.go block/internal/common/da_handler.goblock/internal/syncing/da_handler.go renamed to block/internal/common/da_handler.go

@@ -1,4 +1,4 @@
-package syncing
+package common
 
 import (
 	"bytes"
@@ -12,7 +12,6 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/evstack/ev-node/block/internal/cache"
-	"github.com/evstack/ev-node/block/internal/common"
 	coreda "github.com/evstack/ev-node/core/da"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
@@ -28,13 +27,21 @@ const (
 	maxSubmitAttempts = 30
 )
 
+// HeightEvent represents a block height event with header and data
+type HeightEvent struct {
+	Header                 *types.SignedHeader
+	Data                   *types.Data
+	DaHeight               uint64
+	HeaderDaIncludedHeight uint64
+}
+
 // DAHandler handles all DA layer operations for the syncer
 type DAHandler struct {
 	da      coreda.DA
 	cache   cache.Manager
 	config  config.Config
 	genesis genesis.Genesis
-	options common.BlockOptions
+	options BlockOptions
 	logger  zerolog.Logger
 }
 
@@ -44,7 +51,7 @@ func NewDAHandler(
 	cache cache.Manager,
 	config config.Config,
 	genesis genesis.Genesis,
-	options common.BlockOptions,
+	options BlockOptions,
 	logger zerolog.Logger,
 ) *DAHandler {
 	return &DAHandler{
@@ -294,7 +301,7 @@ func (h *DAHandler) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 
 // isEmptyDataExpected checks if empty data is expected for a header
 func (h *DAHandler) isEmptyDataExpected(header *types.SignedHeader) bool {
-	return len(header.DataHash) == 0 || !bytes.Equal(header.DataHash, common.DataHashForEmptyTxs)
+	return len(header.DataHash) == 0 || !bytes.Equal(header.DataHash, DataHashForEmptyTxs)
 }
 
 // createEmptyDataForHeader creates empty data for a header

…nternal/syncing/da_handler_logic_test.go …internal/common/da_handler_logic_test.goblock/internal/syncing/da_handler_logic_test.go renamed to block/internal/common/da_handler_logic_test.go block/internal/syncing/da_handler_logic_test.go renamed to block/internal/common/da_handler_logic_test.go

@@ -1,4 +1,4 @@
-package syncing
+package common
 
 import (
 	"context"
@@ -14,7 +14,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/evstack/ev-node/block/internal/cache"
-	"github.com/evstack/ev-node/block/internal/common"
 	coreda "github.com/evstack/ev-node/core/da"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
@@ -74,7 +73,7 @@ func TestDAHandler_SubmitHeadersAndData_MarksInclusionAndUpdatesLastSubmitted(t
 	// Dummy DA
 	dummyDA := coreda.NewDummyDA(10_000_000, 0, 0, 10*time.Millisecond)
 
-	handler := NewDAHandler(dummyDA, cm, cfg, gen, common.DefaultBlockOptions(), zerolog.Nop())
+	handler := NewDAHandler(dummyDA, cm, cfg, gen, DefaultBlockOptions(), zerolog.Nop())
 
 	// Submit headers and data
 	require.NoError(t, handler.SubmitHeaders(context.Background(), cm))

block/internal/executing/executor.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/evstack/ev-node/block/internal/cache"
 	"github.com/evstack/ev-node/block/internal/common"
+	coreda "github.com/evstack/ev-node/core/da"
 	coreexecutor "github.com/evstack/ev-node/core/execution"
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
 	"github.com/evstack/ev-node/pkg/config"
@@ -32,13 +33,20 @@ type broadcaster[T any] interface {
 	WriteToStoreAndBroadcast(ctx context.Context, payload T) error
 }
 
+// daHandler is the expected interface for the DA publishing
+type daHandler interface {
+	SubmitHeaders(ctx context.Context, cache cache.Manager) error
+	SubmitData(ctx context.Context, cache cache.Manager, signer signer.Signer, genesis genesis.Genesis) error
+}
+
 // Executor handles block production, transaction processing, and state management
 type Executor struct {
 	// Core components
 	store     store.Store
 	exec      coreexecutor.Executor
 	sequencer coresequencer.Sequencer
 	signer    signer.Signer
+	da        coreda.DA
 
 	// Shared components
 	cache   cache.Manager
@@ -63,6 +71,9 @@ type Executor struct {
 	// Reaper for transaction processing
 	reaper *Reaper
 
+	// DA handler for submissions
+	daHandler daHandler
+
 	// Logging
 	logger zerolog.Logger
 
@@ -72,12 +83,18 @@ type Executor struct {
 	wg     sync.WaitGroup
 }
 
-// NewExecutor creates a new block executor
+// NewExecutor creates a new block executor.
+// The executor is responsible for:
+// - Block production from sequencer batches
+// - State transitions and validation
+// - P2P broadcasting of produced blocks
+// - DA submission of headers and data
 func NewExecutor(
 	store store.Store,
 	exec coreexecutor.Executor,
 	sequencer coresequencer.Sequencer,
 	signer signer.Signer,
+	da coreda.DA,
 	cache cache.Manager,
 	metrics *common.Metrics,
 	config config.Config,
@@ -96,6 +113,7 @@ func NewExecutor(
 		exec:              exec,
 		sequencer:         sequencer,
 		signer:            signer,
+		da:                da,
 		cache:             cache,
 		metrics:           metrics,
 		config:            config,
@@ -125,6 +143,9 @@ func (e *Executor) Start(ctx context.Context) error {
 	}
 	e.reaper = NewReaper(e.ctx, e.exec, e.sequencer, e.genesis.ChainID, DefaultInterval, e.logger, reaperStore, e)
 
+	// Initialize DA handler
+	e.daHandler = common.NewDAHandler(e.da, e.cache, e.config, e.genesis, e.options, e.logger)
+
 	// Start execution loop
 	e.wg.Add(1)
 	go func() {
@@ -139,6 +160,13 @@ func (e *Executor) Start(ctx context.Context) error {
 		e.reaper.Start(e.ctx)
 	}()
 
+	// Start DA submission loop
+	e.wg.Add(1)
+	go func() {
+		defer e.wg.Done()
+		e.daSubmissionLoop()
+	}()
+
 	e.logger.Info().Msg("executor started")
 	return nil
 }
@@ -628,6 +656,36 @@ func (e *Executor) recordBlockMetrics(data *types.Data) {
 	e.metrics.CommittedHeight.Set(float64(data.Metadata.Height))
 }
 
+// daSubmissionLoop handles submission of headers and data to DA layer for aggregator nodes.
+func (e *Executor) daSubmissionLoop() {
+	e.logger.Info().Msg("starting DA submission loop")
+	defer e.logger.Info().Msg("DA submission loop stopped")
+
+	ticker := time.NewTicker(e.config.DA.BlockTime.Duration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-e.ctx.Done():
+			return
+		case <-ticker.C:
+			// Submit headers
+			if e.cache.NumPendingHeaders() != 0 {
+				if err := e.daHandler.SubmitHeaders(e.ctx, e.cache); err != nil {
+					e.logger.Error().Err(err).Msg("failed to submit headers")
+				}
+			}
+
+			// Submit data
+			if e.cache.NumPendingData() != 0 {
+				if err := e.daHandler.SubmitData(e.ctx, e.cache, e.signer, e.genesis); err != nil {
+					e.logger.Error().Err(err).Msg("failed to submit data")
+				}
+			}
+		}
+	}
+}
+
 // BatchData represents batch data from sequencer
 type BatchData struct {
 	*coresequencer.Batch

block/internal/executing/executor_logic_test.go

@@ -76,6 +76,7 @@ func TestProduceBlock_EmptyBatch_SetsEmptyDataHash(t *testing.T) {
 		mockExec,
 		mockSeq,
 		signerWrapper,
+		nil, // DA not needed for this test
 		cacheManager,
 		metrics,
 		cfg,
@@ -156,6 +157,7 @@ func TestPendingLimit_SkipsProduction(t *testing.T) {
 		mockExec,
 		mockSeq,
 		signerWrapper,
+		nil, // DA not needed for this test
 		cacheManager,
 		metrics,
 		cfg,

block/internal/executing/executor_test.go

@@ -43,7 +43,7 @@ func TestExecutor_BroadcasterIntegration(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create metrics
-	metrics := common.PrometheusMetrics("test_executor")
+	metrics := common.NopMetrics()
 
 	// Create genesis
 	gen := genesis.Genesis{
@@ -69,6 +69,7 @@ func TestExecutor_BroadcasterIntegration(t *testing.T) {
 		nil,        // nil executor (we're not testing execution)
 		nil,        // nil sequencer (we're not testing sequencing)
 		testSigner, // test signer (required for executor)
+		nil,        // nil DA (we're not testing DA operations)
 		cacheManager,
 		metrics,
 		config.DefaultConfig,
@@ -101,7 +102,7 @@ func TestExecutor_NilBroadcasters(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create metrics
-	metrics := common.PrometheusMetrics("test_executor_nil")
+	metrics := common.NopMetrics()
 
 	// Create genesis
 	gen := genesis.Genesis{
@@ -123,6 +124,7 @@ func TestExecutor_NilBroadcasters(t *testing.T) {
 		nil,        // nil executor
 		nil,        // nil sequencer
 		testSigner, // test signer (required for executor)
+		nil,        // nil DA
 		cacheManager,
 		metrics,
 		config.DefaultConfig,

block/internal/executing/reaper_test.go

@@ -37,6 +37,7 @@ func newTestExecutor(t *testing.T) *Executor {
 		nil, // core executor (unused)
 		nil, // sequencer (unused)
 		s,   // signer (required)
+		nil, // DA (unused)
 		nil, // cache (unused)
 		nil, // metrics (unused)
 		config.DefaultConfig,

block/internal/syncing/p2p_handler.go

@@ -9,11 +9,10 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/block/internal/cache"
+	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/signer"
 	"github.com/evstack/ev-node/types"
-
-	"github.com/evstack/ev-node/block/internal/common"
 )
 
 // P2PHandler handles all P2P operations for the syncer
@@ -49,12 +48,12 @@ func NewP2PHandler(
 }
 
 // ProcessHeaderRange processes headers from the header store within the given range
-func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHeight uint64) []HeightEvent {
+func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHeight uint64) []common.HeightEvent {
 	if startHeight > endHeight {
 		return nil
 	}
 
-	var events []HeightEvent
+	var events []common.HeightEvent
 
 	for height := startHeight; height <= endHeight; height++ {
 		select {
@@ -97,7 +96,7 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 		}
 
 		// Create height event
-		event := HeightEvent{
+		event := common.HeightEvent{
 			Header:   header,
 			Data:     data,
 			DaHeight: 0, // P2P events don't have DA height context
@@ -112,12 +111,12 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 }
 
 // ProcessDataRange processes data from the data store within the given range
-func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeight uint64) []HeightEvent {
+func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeight uint64) []common.HeightEvent {
 	if startHeight > endHeight {
 		return nil
 	}
 
-	var events []HeightEvent
+	var events []common.HeightEvent
 
 	for height := startHeight; height <= endHeight; height++ {
 		select {
@@ -152,7 +151,7 @@ func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeigh
 		}
 
 		// Create height event
-		event := HeightEvent{
+		event := common.HeightEvent{
 			Header:   header,
 			Data:     data,
 			DaHeight: 0, // P2P events don't have DA height context