chore: use interfaces

Author: julienrbrt

block/internal/da/async_epoch_fetcher.go

@@ -16,9 +16,16 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
-// AsyncEpochFetcher handles background prefetching of DA epoch data
+// AsyncEpochFetcher provides background prefetching of DA epoch data
+type AsyncEpochFetcher interface {
+	Start()
+	Stop()
+	GetCachedEpoch(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error)
+}
+
+// asyncEpochFetcher handles background prefetching of DA epoch data
 // to speed up processing at epoch boundaries.
-type AsyncEpochFetcher struct {
+type asyncEpochFetcher struct {
 	client        Client
 	logger        zerolog.Logger
 	daEpochSize   uint64
@@ -35,7 +42,6 @@ type AsyncEpochFetcher struct {
 
 	// Current DA height tracking
 	currentDAHeight uint64
-	heightMu        sync.RWMutex
 
 	// Prefetch window - how many epochs ahead to prefetch
 	prefetchWindow uint64
@@ -51,7 +57,7 @@ func NewAsyncEpochFetcher(
 	daStartHeight, daEpochSize uint64,
 	prefetchWindow uint64,
 	pollInterval time.Duration,
-) *AsyncEpochFetcher {
+) AsyncEpochFetcher {
 	if prefetchWindow == 0 {
 		prefetchWindow = 1 // Default: prefetch next epoch
 	}
@@ -61,7 +67,7 @@ func NewAsyncEpochFetcher(
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	return &AsyncEpochFetcher{
+	return &asyncEpochFetcher{
 		client:          client,
 		logger:          logger.With().Str("component", "async_epoch_fetcher").Logger(),
 		daStartHeight:   daStartHeight,
@@ -76,7 +82,7 @@ func NewAsyncEpochFetcher(
 }
 
 // Start begins the background prefetching process.
-func (f *AsyncEpochFetcher) Start() {
+func (f *asyncEpochFetcher) Start() {
 	f.wg.Add(1)
 	go f.backgroundFetchLoop()
 	f.logger.Info().
@@ -88,35 +94,16 @@ func (f *AsyncEpochFetcher) Start() {
 }
 
 // Stop gracefully stops the background prefetching process.
-func (f *AsyncEpochFetcher) Stop() {
+func (f *asyncEpochFetcher) Stop() {
 	f.logger.Info().Msg("stopping async epoch fetcher")
 	f.cancel()
 	f.wg.Wait()
 	f.logger.Info().Msg("async epoch fetcher stopped")
 }
 
-// SetDAHeight updates the current DA height being processed.
-// This is called by sequencers to inform the fetcher of progress.
-func (f *AsyncEpochFetcher) SetDAHeight(height uint64) {
-	f.heightMu.Lock()
-	defer f.heightMu.Unlock()
-
-	if height > f.currentDAHeight {
-		f.currentDAHeight = height
-		f.logger.Debug().Uint64("da_height", height).Msg("updated current DA height")
-	}
-}
-
-// GetDAHeight returns the current DA height.
-func (f *AsyncEpochFetcher) GetDAHeight() uint64 {
-	f.heightMu.RLock()
-	defer f.heightMu.RUnlock()
-	return f.currentDAHeight
-}
-
 // GetCachedEpoch retrieves a cached epoch from memory.
 // Returns nil if the epoch is not cached.
-func (f *AsyncEpochFetcher) GetCachedEpoch(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error) {
+func (f *asyncEpochFetcher) GetCachedEpoch(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error) {
 	if !f.client.HasForcedInclusionNamespace() {
 		return nil, ErrForceInclusionNotConfigured
 	}
@@ -167,7 +154,7 @@ func (f *AsyncEpochFetcher) GetCachedEpoch(ctx context.Context, daHeight uint64)
 }
 
 // backgroundFetchLoop runs in the background and prefetches epochs ahead of time.
-func (f *AsyncEpochFetcher) backgroundFetchLoop() {
+func (f *asyncEpochFetcher) backgroundFetchLoop() {
 	defer f.wg.Done()
 
 	ticker := time.NewTicker(f.pollInterval)
@@ -184,15 +171,13 @@ func (f *AsyncEpochFetcher) backgroundFetchLoop() {
 }
 
 // prefetchEpochs prefetches epochs within the prefetch window.
-func (f *AsyncEpochFetcher) prefetchEpochs() {
+func (f *asyncEpochFetcher) prefetchEpochs() {
 	if !f.client.HasForcedInclusionNamespace() {
 		return
 	}
 
-	currentHeight := f.GetDAHeight()
-
 	// Calculate the current epoch and epochs to prefetch
-	_, currentEpochEnd, _ := types.CalculateEpochBoundaries(currentHeight, f.daStartHeight, f.daEpochSize)
+	_, currentEpochEnd, _ := types.CalculateEpochBoundaries(f.currentDAHeight, f.daStartHeight, f.daEpochSize)
 
 	// Prefetch upcoming epochs
 	for i := uint64(0); i < f.prefetchWindow; i++ {
@@ -218,7 +203,7 @@ func (f *AsyncEpochFetcher) prefetchEpochs() {
 }
 
 // fetchAndCacheEpoch fetches an epoch and stores it in the cache.
-func (f *AsyncEpochFetcher) fetchAndCacheEpoch(epochEnd uint64) {
+func (f *asyncEpochFetcher) fetchAndCacheEpoch(epochEnd uint64) {
 	epochStart := epochEnd - (f.daEpochSize - 1)
 	if epochStart < f.daStartHeight {
 		epochStart = f.daStartHeight
@@ -313,7 +298,7 @@ func (f *AsyncEpochFetcher) fetchAndCacheEpoch(epochEnd uint64) {
 }
 
 // processForcedInclusionBlobs processes blobs from a single DA height for forced inclusion.
-func (f *AsyncEpochFetcher) processForcedInclusionBlobs(
+func (f *asyncEpochFetcher) processForcedInclusionBlobs(
 	event *ForcedInclusionEvent,
 	result datypes.ResultRetrieve,
 	height uint64,
@@ -347,7 +332,7 @@ func (f *AsyncEpochFetcher) processForcedInclusionBlobs(
 }
 
 // cleanupOldEpochs removes epochs older than the current epoch from cache.
-func (f *AsyncEpochFetcher) cleanupOldEpochs(currentEpochEnd uint64) {
+func (f *asyncEpochFetcher) cleanupOldEpochs(currentEpochEnd uint64) {
 	// Remove epochs older than current - 1
 	// Keep current and previous in case of reorgs or restarts
 	cleanupThreshold := currentEpochEnd - f.daEpochSize

block/internal/da/async_epoch_fetcher_test.go

@@ -15,34 +15,6 @@ import (
 	mocks "github.com/evstack/ev-node/test/mocks"
 )
 
-func TestAsyncEpochFetcher_Creation(t *testing.T) {
-	client := &mocks.MockClient{}
-	logger := zerolog.Nop()
-
-	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 2, 100*time.Millisecond)
-	assert.NotNil(t, fetcher)
-	assert.Equal(t, uint64(100), fetcher.daStartHeight)
-	assert.Equal(t, uint64(10), fetcher.daEpochSize)
-	assert.Equal(t, uint64(2), fetcher.prefetchWindow)
-	assert.Equal(t, 100*time.Millisecond, fetcher.pollInterval)
-}
-
-func TestAsyncEpochFetcher_SetAndGetDAHeight(t *testing.T) {
-	client := &mocks.MockClient{}
-	logger := zerolog.Nop()
-
-	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 1, time.Second)
-
-	assert.Equal(t, uint64(100), fetcher.GetDAHeight())
-
-	fetcher.SetDAHeight(150)
-	assert.Equal(t, uint64(150), fetcher.GetDAHeight())
-
-	// Should not decrease
-	fetcher.SetDAHeight(120)
-	assert.Equal(t, uint64(150), fetcher.GetDAHeight())
-}
-
 func TestAsyncEpochFetcher_GetCachedEpoch_NotAtEpochEnd(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
@@ -122,16 +94,27 @@ func TestAsyncEpochFetcher_FetchAndCache(t *testing.T) {
 	}
 
 	logger := zerolog.Nop()
-	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 1, time.Second)
-
-	// Manually trigger fetch
-	fetcher.fetchAndCacheEpoch(109)
+	// Use a short poll interval for faster test execution
+	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 1, 50*time.Millisecond)
+	fetcher.Start()
+	defer fetcher.Stop()
 
-	// Now try to get from cache
+	// Wait for the background fetch to complete by polling the cache
 	ctx := context.Background()
-	event, err := fetcher.GetCachedEpoch(ctx, 109)
-	require.NoError(t, err)
-	assert.NotNil(t, event)
+	var event *ForcedInclusionEvent
+	var err error
+
+	// Poll for up to 2 seconds for the epoch to be cached
+	for i := 0; i < 40; i++ {
+		event, err = fetcher.GetCachedEpoch(ctx, 109)
+		require.NoError(t, err)
+		if event != nil {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	require.NotNil(t, event, "epoch should be cached after background fetch")
 	assert.Equal(t, 3, len(event.Txs))
 	assert.Equal(t, testBlobs[0], event.Txs[0])
 	assert.Equal(t, uint64(100), event.StartDaHeight)
@@ -176,7 +159,6 @@ func TestAsyncEpochFetcher_BackgroundPrefetch(t *testing.T) {
 	logger := zerolog.Nop()
 	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 1, 50*time.Millisecond)
 
-	fetcher.SetDAHeight(100)
 	fetcher.Start()
 	defer fetcher.Stop()
 
@@ -250,9 +232,8 @@ func TestAsyncEpochFetcher_HeightFromFuture(t *testing.T) {
 
 	logger := zerolog.Nop()
 	fetcher := NewAsyncEpochFetcher(client, logger, 100, 10, 1, time.Second)
-
-	// This should not panic and should handle gracefully
-	fetcher.fetchAndCacheEpoch(109)
+	fetcher.Start()
+	defer fetcher.Stop()
 
 	// Cache should be empty
 	ctx := context.Background()

block/internal/da/forced_inclusion_retriever.go

@@ -21,7 +21,7 @@ type ForcedInclusionRetriever struct {
 	logger        zerolog.Logger
 	daEpochSize   uint64
 	daStartHeight uint64
-	asyncFetcher  *AsyncEpochFetcher // Required for async prefetching
+	asyncFetcher  AsyncEpochFetcher // Required for async prefetching
 }
 
 // ForcedInclusionEvent contains forced inclusion transactions retrieved from DA.
@@ -33,12 +33,11 @@ type ForcedInclusionEvent struct {
 }
 
 // NewForcedInclusionRetriever creates a new forced inclusion retriever.
-// The asyncFetcher parameter is required for background prefetching of DA epoch data.
 func NewForcedInclusionRetriever(
 	client Client,
 	logger zerolog.Logger,
 	daStartHeight, daEpochSize uint64,
-	asyncFetcher *AsyncEpochFetcher,
+	asyncFetcher AsyncEpochFetcher,
 ) *ForcedInclusionRetriever {
 	return &ForcedInclusionRetriever{
 		client:        client,
@@ -51,7 +50,6 @@ func NewForcedInclusionRetriever(
 
 // RetrieveForcedIncludedTxs retrieves forced inclusion transactions at the given DA height.
 // It respects epoch boundaries and only fetches at epoch start.
-// If an async fetcher is configured, it will try to use cached data first for better performance.
 func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error) {
 	// when daStartHeight is not set or no namespace is configured, we retrieve nothing.
 	if !r.client.HasForcedInclusionNamespace() {

block/internal/da/forced_inclusion_retriever_test.go

@@ -14,8 +14,8 @@ import (
 	"github.com/evstack/ev-node/test/mocks"
 )
 
-// createTestAsyncFetcher creates a minimal async fetcher for tests (without starting it)
-func createTestAsyncFetcher(client Client, gen genesis.Genesis) *AsyncEpochFetcher {
+// createTestAsyncFetcher creates a minimal async fetcher for tests
+func createTestAsyncFetcher(client Client, gen genesis.Genesis) AsyncEpochFetcher {
 	return NewAsyncEpochFetcher(
 		client,
 		zerolog.Nop(),

block/internal/syncing/syncer.go

@@ -194,8 +194,8 @@ func (s *Syncer) Start(ctx context.Context) error {
 		s.logger,
 		s.genesis.DAStartHeight,
 		s.genesis.DAEpochForcedInclusion,
-		1,             // prefetch 1 epoch ahead
-		2*time.Second, // check every 2 seconds
+		1, // prefetch 1 epoch ahead
+		s.config.DA.BlockTime.Duration,
 	)
 	asyncFetcher.Start()
 

block/internal/syncing/syncer_forced_inclusion_test.go

@@ -24,7 +24,7 @@ import (
 )
 
 // createTestAsyncFetcher creates a minimal async fetcher for tests (without starting it)
-func createTestAsyncFetcherForSyncer(client da.Client, gen genesis.Genesis) *da.AsyncEpochFetcher {
+func createTestAsyncFetcherForSyncer(client da.Client, gen genesis.Genesis) da.AsyncEpochFetcher {
 	return da.NewAsyncEpochFetcher(
 		client,
 		zerolog.Nop(),

block/public.go

@@ -67,7 +67,11 @@ var ErrForceInclusionNotConfigured = da.ErrForceInclusionNotConfigured
 type ForcedInclusionEvent = da.ForcedInclusionEvent
 
 // AsyncEpochFetcher provides background prefetching of DA epoch data
-type AsyncEpochFetcher = da.AsyncEpochFetcher
+type AsyncEpochFetcher interface {
+	Start()
+	Stop()
+	GetCachedEpoch(ctx context.Context, daHeight uint64) (*da.ForcedInclusionEvent, error)
+}
 
 // ForcedInclusionRetriever defines the interface for retrieving forced inclusion transactions from DA
 type ForcedInclusionRetriever interface {
@@ -88,7 +92,7 @@ func NewAsyncEpochFetcher(
 	daStartHeight, daEpochSize uint64,
 	prefetchWindow uint64,
 	pollInterval time.Duration,
-) *AsyncEpochFetcher {
+) AsyncEpochFetcher {
 	return da.NewAsyncEpochFetcher(client, logger, daStartHeight, daEpochSize, prefetchWindow, pollInterval)
 }
 
@@ -98,7 +102,7 @@ func NewForcedInclusionRetriever(
 	client DAClient,
 	logger zerolog.Logger,
 	daStartHeight, daEpochSize uint64,
-	asyncFetcher *AsyncEpochFetcher,
+	asyncFetcher AsyncEpochFetcher,
 ) ForcedInclusionRetriever {
 	return da.NewForcedInclusionRetriever(client, logger, daStartHeight, daEpochSize, asyncFetcher)
 }

pkg/sequencers/based/sequencer.go

@@ -25,7 +25,7 @@ var _ coresequencer.Sequencer = (*BasedSequencer)(nil)
 type BasedSequencer struct {
 	logger zerolog.Logger
 
-	asyncFetcher    *block.AsyncEpochFetcher
+	asyncFetcher    block.AsyncEpochFetcher
 	fiRetriever     block.ForcedInclusionRetriever
 	daHeight        atomic.Uint64
 	checkpointStore *seqcommon.CheckpointStore

pkg/sequencers/single/sequencer.go

@@ -40,7 +40,7 @@ type Sequencer struct {
 	queue     *BatchQueue // single queue for immediate availability
 
 	// Forced inclusion support
-	asyncFetcher    *block.AsyncEpochFetcher
+	asyncFetcher    block.AsyncEpochFetcher
 	fiRetriever     block.ForcedInclusionRetriever
 	daHeight        atomic.Uint64
 	daStartHeight   atomic.Uint64