refactor: extend local-da to produce blocks

Author: julienrbrt

block/internal/syncing/syncer.go

@@ -749,7 +749,7 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 	}
 
 	// Verify forced inclusion transactions if configured.
-	// The checks is actually only performed on DA only enabled nodes.
+	// The check is actually only performed on DA only enabled nodes.
 	// P2P nodes aren't actually able to verify forced inclusions txs as DA inclusion happens later (so DA hints are not available) and DA hints cannot be trusted. This is a known limitation described in the ADR.
 	// In the future we should verify force inclusion txs completely asynchronously, while still waiting for block n-1 execution.
 	if event.Source == common.SourceDA {

block/public.go

@@ -63,9 +63,13 @@ func NewDAClient(
 	return base
 }
 
-// ErrForceInclusionNotConfigured is returned when force inclusion is not configured.
-// It is exported because sequencers needs to check for this error.
-var ErrForceInclusionNotConfigured = da.ErrForceInclusionNotConfigured
+// The following errors are exported because sequencers needs to check for this error.
+var (
+	// ErrForceInclusionNotConfigured is returned when force inclusion is not configured.
+	ErrForceInclusionNotConfigured = da.ErrForceInclusionNotConfigured
+	// ErrNoBatch is returned when no batch is available.
+	ErrNoBatch = common.ErrNoBatch
+)
 
 // ForcedInclusionEvent represents forced inclusion transactions retrieved from DA
 type ForcedInclusionEvent = da.ForcedInclusionEvent

pkg/sequencers/based/sequencer.go

@@ -253,7 +253,7 @@ func (s *BasedSequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64)
 	// Cache the transactions for this DA epoch
 	s.currentBatchTxs = validTxs
 
-	return forcedTxsEvent.Timestamp.UTC(), forcedTxsEvent.EndDaHeight, nil
+	return forcedTxsEvent.Timestamp, forcedTxsEvent.EndDaHeight, nil
 }
 
 // VerifyBatch verifies a batch of transactions

test/e2e/evm_force_inclusion_e2e_test.go

@@ -808,13 +808,7 @@ func TestEvmSequencerCatchUpBasedSequencerE2E(t *testing.T) {
 
 	// Hack for local-da for advancing past multiple epochs
 	t.Log("Advancing DA past multiple epochs...")
-	for i := 0; i < 5; i++ {
-		dummyData := [][]byte{[]byte(fmt.Sprintf("dummy-%d", i))}
-		result := daClient.Submit(ctx, dummyData, -1, daClient.GetDataNamespace(), nil)
-		require.Equal(t, da.StatusSuccess, result.Code, "Failed to submit dummy data %d to DA: %s", i+1, result.Message)
-		time.Sleep(100 * time.Millisecond)
-	}
-	t.Log("DA advanced past multiple epochs")
+	time.Sleep(6 * time.Second)
 
 	// ===== PHASE 6: Verify Based Sequencer Includes Forced Txs =====
 	t.Log("Phase 6: Verify Based Sequencer Includes Forced Txs")
@@ -1058,13 +1052,7 @@ func TestEvmBasedSequencerBaselineE2E(t *testing.T) {
 	// Advance DA past epoch boundary by submitting dummy data
 	// With epoch=2, we need at least 2 DA blocks per epoch
 	t.Log("Advancing DA past epoch boundary...")
-	for i := 0; i < 5; i++ {
-		dummyData := [][]byte{[]byte(fmt.Sprintf("dummy-%d", i))}
-		result := daClient.Submit(ctx, dummyData, -1, daClient.GetDataNamespace(), nil)
-		require.Equal(t, da.StatusSuccess, result.Code, "Failed to submit dummy data %d to DA: %s", i+1, result.Message)
-		time.Sleep(100 * time.Millisecond)
-	}
-	t.Log("DA advanced past epoch boundary")
+	time.Sleep(4 * time.Second)
 
 	// ===== VERIFY BASED SEQUENCER INCLUDES FORCED TXS =====
 	t.Log("Waiting for based sequencer to include forced inclusion txs")

test/e2e/evm_test_common.go

@@ -554,7 +554,7 @@ func setupCommonEVMEnv(t testing.TB, sut *SystemUnderTest, client tastoratypes.T
 	if evmSingleBinaryPath != "evm" {
 		localDABinary = filepath.Join(filepath.Dir(evmSingleBinaryPath), "local-da")
 	}
-	sut.ExecCmd(localDABinary, "-port", dynEndpoints.DAPort)
+	sut.ExecCmd(localDABinary, "-port", dynEndpoints.DAPort, "-block-time", "1s")
 	t.Logf("Started local DA on port %s", dynEndpoints.DAPort)
 
 	require.NotNil(t, client, "docker client is required")

tools/local-da/local.go

@@ -19,8 +19,13 @@ import (
 	datypes "github.com/evstack/ev-node/pkg/da/types"
 )
 
-// DefaultMaxBlobSize is the default max blob size
-const DefaultMaxBlobSize uint64 = 7 * 1024 * 1024 // 7MB
+const (
+	// DefaultMaxBlobSize is the default max blob size
+	DefaultMaxBlobSize uint64 = 7 * 1024 * 1024 // 7MB
+
+	// DefaultBlockTime is the default time between empty blocks
+	DefaultBlockTime = 1 * time.Second
+)
 
 // LocalDA is a simple implementation of in-memory DA. Not production ready! Intended only for testing!
 //
@@ -35,6 +40,8 @@ type LocalDA struct {
 	height      uint64
 	privKey     ed25519.PrivateKey
 	pubKey      ed25519.PublicKey
+	blockTime   time.Duration
+	lastTime    time.Time // tracks last timestamp to ensure monotonicity
 
 	logger zerolog.Logger
 }
@@ -51,6 +58,8 @@ func NewLocalDA(logger zerolog.Logger, opts ...func(*LocalDA) *LocalDA) *LocalDA
 		timestamps:  make(map[uint64]time.Time),
 		blobData:    make(map[uint64][]*blobrpc.Blob),
 		maxBlobSize: DefaultMaxBlobSize,
+		blockTime:   DefaultBlockTime,
+		lastTime:    time.Now(),
 		logger:      logger,
 	}
 	for _, f := range opts {
@@ -194,7 +203,7 @@ func (d *LocalDA) SubmitWithOptions(ctx context.Context, blobs []datypes.Blob, g
 	defer d.mu.Unlock()
 	ids := make([]datypes.ID, len(blobs))
 	d.height += 1
-	d.timestamps[d.height] = time.Now()
+	d.timestamps[d.height] = d.monotonicTime()
 	for i, blob := range blobs {
 		ids[i] = append(d.nextID(), d.getHash(blob)...)
 
@@ -224,7 +233,7 @@ func (d *LocalDA) Submit(ctx context.Context, blobs []datypes.Blob, gasPrice flo
 	defer d.mu.Unlock()
 	ids := make([]datypes.ID, len(blobs))
 	d.height += 1
-	d.timestamps[d.height] = time.Now()
+	d.timestamps[d.height] = d.monotonicTime()
 	for i, blob := range blobs {
 		ids[i] = append(d.nextID(), d.getHash(blob)...)
 
@@ -274,10 +283,57 @@ func (d *LocalDA) getProof(id, blob []byte) []byte {
 	return sign
 }
 
+// monotonicTime returns a timestamp that is guaranteed to be after the last recorded timestamp.
+func (d *LocalDA) monotonicTime() time.Time {
+	now := time.Now()
+	if now.After(d.lastTime) {
+		d.lastTime = now
+		return now
+	}
+	d.lastTime = d.lastTime.Add(1)
+	return d.lastTime
+}
+
 // WithMaxBlobSize returns a function that sets the max blob size of LocalDA
 func WithMaxBlobSize(maxBlobSize uint64) func(*LocalDA) *LocalDA {
 	return func(da *LocalDA) *LocalDA {
 		da.maxBlobSize = maxBlobSize
 		return da
 	}
 }
+
+// WithBlockTime returns a function that sets the block time for empty block production
+func WithBlockTime(blockTime time.Duration) func(*LocalDA) *LocalDA {
+	return func(da *LocalDA) *LocalDA {
+		da.blockTime = blockTime
+		return da
+	}
+}
+
+// Start begins producing empty blocks at the configured block time interval.
+func (d *LocalDA) Start(ctx context.Context) {
+	go func() {
+		ticker := time.NewTicker(d.blockTime)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				d.logger.Info().Msg("LocalDA: stopping empty block production")
+				return
+			case <-ticker.C:
+				d.produceEmptyBlock()
+			}
+		}
+	}()
+	d.logger.Info().Dur("blockTime", d.blockTime).Msg("LocalDA: started empty block production")
+}
+
+// produceEmptyBlock creates a new empty block at the next height.
+func (d *LocalDA) produceEmptyBlock() {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.height++
+	d.timestamps[d.height] = d.monotonicTime()
+	d.logger.Debug().Uint64("height", d.height).Msg("produced empty block")
+}

tools/local-da/main.go

@@ -23,11 +23,13 @@ func main() {
 		port        string
 		listenAll   bool
 		maxBlobSize uint64
+		blockTime   time.Duration
 	)
 	flag.StringVar(&port, "port", defaultPort, "listening port")
 	flag.StringVar(&host, "host", defaultHost, "listening address")
 	flag.BoolVar(&listenAll, "listen-all", false, "listen on all network interfaces (0.0.0.0) instead of just localhost")
 	flag.Uint64Var(&maxBlobSize, "max-blob-size", DefaultMaxBlobSize, "maximum blob size in bytes")
+	flag.DurationVar(&blockTime, "block-time", DefaultBlockTime, "time between empty blocks (e.g., 1s, 500ms)")
 	flag.Parse()
 
 	if listenAll {
@@ -43,25 +45,33 @@ func main() {
 	if maxBlobSize != DefaultMaxBlobSize {
 		opts = append(opts, WithMaxBlobSize(maxBlobSize))
 	}
+	if blockTime != DefaultBlockTime {
+		opts = append(opts, WithBlockTime(blockTime))
+	}
 	da := NewLocalDA(logger, opts...)
 
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	da.Start(ctx)
+
 	addr := fmt.Sprintf("%s:%s", host, port)
 	srv, err := startBlobServer(logger, addr, da)
 	if err != nil {
 		logger.Error().Err(err).Msg("error while creating blob RPC server")
 		os.Exit(1)
 	}
 
-	logger.Info().Str("host", host).Str("port", port).Uint64("maxBlobSize", maxBlobSize).Msg("Listening on")
+	logger.Info().Str("host", host).Str("port", port).Uint64("maxBlobSize", maxBlobSize).Dur("blockTime", blockTime).Msg("Listening on")
 
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt, syscall.SIGINT)
 	<-interrupt
 	fmt.Println("\nCtrl+C pressed. Exiting...")
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	if err := srv.Shutdown(ctx); err != nil {
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer shutdownCancel()
+	if err := srv.Shutdown(shutdownCtx); err != nil {
 		logger.Error().Err(err).Msg("error shutting down server")
 	}
 	os.Exit(0)