rename working pool to avoid naming confusion

julienrbrt · Nov 26, 2025 · Nov 27, 2025 · Nov 28, 2025 · Nov 28, 2025 · Nov 28, 2025
commit aac3f47928eb98fc7d6f22acdded33b26b739df9
diff --git a/block/internal/syncing/async_da_retriever.go → ...ernal/syncing/da_retrieval_worker_pool.go b/block/internal/syncing/async_da_retriever.go → ...ernal/syncing/da_retrieval_worker_pool.go
@@ -8,8 +8,10 @@ import (
 	"github.com/rs/zerolog"
 )
 
-// AsyncDARetriever handles concurrent DA retrieval operations.
-type AsyncDARetriever struct {
+// DARetrievalWorkerPool handles concurrent on-demand DA retrieval operations
+// triggered by P2P hints. Unlike the background prefetcher (AsyncBlockRetriever),
+// this worker pool processes explicit retrieval requests for specific DA heights.
+type DARetrievalWorkerPool struct {
 	retriever DARetriever
 	resultCh  chan<- common.DAHeightEvent
 	workCh    chan uint64
@@ -21,44 +23,44 @@ type AsyncDARetriever struct {
 	cancel    context.CancelFunc
 }
 
-// NewAsyncDARetriever creates a new AsyncDARetriever.
-func NewAsyncDARetriever(
+// NewDARetrievalWorkerPool creates a new DARetrievalWorkerPool.
+func NewDARetrievalWorkerPool(
 	retriever DARetriever,
 	resultCh chan<- common.DAHeightEvent,
 	logger zerolog.Logger,
-) *AsyncDARetriever {
-	return &AsyncDARetriever{
+) *DARetrievalWorkerPool {
+	return &DARetrievalWorkerPool{
 		retriever: retriever,
 		resultCh:  resultCh,
 		workCh:    make(chan uint64, 100), // Buffer size 100
 		inFlight:  make(map[uint64]struct{}),
-		logger:    logger.With().Str("component", "async_da_retriever").Logger(),
+		logger:    logger.With().Str("component", "da_retrieval_worker_pool").Logger(),
 	}
 }
 
 // Start starts the worker pool.
-func (r *AsyncDARetriever) Start(ctx context.Context) {
+func (r *DARetrievalWorkerPool) Start(ctx context.Context) {
 	r.ctx, r.cancel = context.WithCancel(ctx)
 	// Start 5 workers
 	for i := 0; i < 5; i++ {
 		r.wg.Add(1)
 		go r.worker()
 	}
-	r.logger.Info().Msg("AsyncDARetriever started")
+	r.logger.Info().Msg("DARetrievalWorkerPool started")
 }
 
 // Stop stops the worker pool.
-func (r *AsyncDARetriever) Stop() {
+func (r *DARetrievalWorkerPool) Stop() {
 	if r.cancel != nil {
 		r.cancel()
 	}
 	r.wg.Wait()
-	r.logger.Info().Msg("AsyncDARetriever stopped")
+	r.logger.Info().Msg("DARetrievalWorkerPool stopped")
 }
 
 // RequestRetrieval requests a DA retrieval for the given height.
 // It is non-blocking and idempotent.
-func (r *AsyncDARetriever) RequestRetrieval(height uint64) {
+func (r *DARetrievalWorkerPool) RequestRetrieval(height uint64) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -75,7 +77,7 @@ func (r *AsyncDARetriever) RequestRetrieval(height uint64) {
 	}
 }
 
-func (r *AsyncDARetriever) worker() {
+func (r *DARetrievalWorkerPool) worker() {
 	defer r.wg.Done()
 
 	for {
@@ -88,7 +90,7 @@ func (r *AsyncDARetriever) worker() {
 	}
 }
 
-func (r *AsyncDARetriever) processRetrieval(height uint64) {
+func (r *DARetrievalWorkerPool) processRetrieval(height uint64) {
 	defer func() {
 		r.mu.Lock()
 		delete(r.inFlight, height)
@@ -97,7 +99,7 @@ func (r *AsyncDARetriever) processRetrieval(height uint64) {
 
 	events, err := r.retriever.RetrieveFromDA(r.ctx, height)
 	if err != nil {
-		r.logger.Debug().Err(err).Uint64("height", height).Msg("async DA retrieval failed")
+		r.logger.Debug().Err(err).Uint64("height", height).Msg("DA retrieval failed")
 		return
 	}
 

diff --git a/...ternal/syncing/async_da_retriever_test.go → .../syncing/da_retrieval_worker_pool_test.go b/...ternal/syncing/async_da_retriever_test.go → .../syncing/da_retrieval_worker_pool_test.go
@@ -11,23 +11,23 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-func TestAsyncDARetriever_RequestRetrieval(t *testing.T) {
+func TestDARetrievalWorkerPool_RequestRetrieval(t *testing.T) {
 	logger := zerolog.Nop()
 	mockRetriever := NewMockDARetriever(t)
 	resultCh := make(chan common.DAHeightEvent, 10)
 
-	asyncRetriever := NewAsyncDARetriever(mockRetriever, resultCh, logger)
+	workerPool := NewDARetrievalWorkerPool(mockRetriever, resultCh, logger)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	asyncRetriever.Start(ctx)
-	defer asyncRetriever.Stop()
+	workerPool.Start(ctx)
+	defer workerPool.Stop()
 
 	// 1. Test successful retrieval
 	height1 := uint64(100)
 	mockRetriever.EXPECT().RetrieveFromDA(mock.Anything, height1).Return([]common.DAHeightEvent{{DaHeight: height1}}, nil).Once()
 
-	asyncRetriever.RequestRetrieval(height1)
+	workerPool.RequestRetrieval(height1)
 
 	select {
 	case event := <-resultCh:
@@ -52,7 +52,7 @@ func TestAsyncDARetriever_RequestRetrieval(t *testing.T) {
 	}).Once() // Should be called only once despite multiple requests
 
 	// Send first request
-	asyncRetriever.RequestRetrieval(height2)
+	workerPool.RequestRetrieval(height2)
 
 	// Wait for the worker to pick it up
 	select {
@@ -62,8 +62,8 @@ func TestAsyncDARetriever_RequestRetrieval(t *testing.T) {
 	}
 
 	// Send duplicate requests while the first one is still in flight
-	asyncRetriever.RequestRetrieval(height2)
-	asyncRetriever.RequestRetrieval(height2)
+	workerPool.RequestRetrieval(height2)
+	workerPool.RequestRetrieval(height2)
 
 	// Unblock the worker
 	close(unblockCh)
@@ -84,17 +84,17 @@ func TestAsyncDARetriever_RequestRetrieval(t *testing.T) {
 	}
 }
 
-func TestAsyncDARetriever_WorkerPoolLimit(t *testing.T) {
+func TestDARetrievalWorkerPool_WorkerPoolLimit(t *testing.T) {
 	logger := zerolog.Nop()
 	mockRetriever := NewMockDARetriever(t)
 	resultCh := make(chan common.DAHeightEvent, 100)
 
-	asyncRetriever := NewAsyncDARetriever(mockRetriever, resultCh, logger)
+	workerPool := NewDARetrievalWorkerPool(mockRetriever, resultCh, logger)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	asyncRetriever.Start(ctx)
-	defer asyncRetriever.Stop()
+	workerPool.Start(ctx)
+	defer workerPool.Stop()
 
 	// We have 5 workers. We'll block them all.
 	unblockCh := make(chan struct{})
@@ -106,7 +106,7 @@ func TestAsyncDARetriever_WorkerPoolLimit(t *testing.T) {
 			<-unblockCh
 			return []common.DAHeightEvent{{DaHeight: h}}, nil
 		}).Once()
-		asyncRetriever.RequestRetrieval(h)
+		workerPool.RequestRetrieval(h)
 	}
 
 	// Give workers time to pick up tasks
@@ -120,7 +120,7 @@ func TestAsyncDARetriever_WorkerPoolLimit(t *testing.T) {
 		return []common.DAHeightEvent{{DaHeight: h}}, nil
 	}).Once()
 
-	asyncRetriever.RequestRetrieval(height6)
+	workerPool.RequestRetrieval(height6)
 
 	// Ensure 6th request is NOT processed yet
 	select {

diff --git a/block/internal/syncing/syncer.go b/block/internal/syncing/syncer.go
@@ -127,8 +127,8 @@ type Syncer struct {
 	// defaults to self, but can be wrapped with tracing.
 	blockSyncer BlockSyncer
 
-	// Async DA retriever
-	asyncDARetriever *AsyncDARetriever
+	// Worker pool for on-demand DA retrieval triggered by P2P hints
+	daRetrievalWorkerPool *DARetrievalWorkerPool
 }
 
 // pendingForcedInclusionTx represents a forced inclusion transaction that hasn't been included yet
@@ -216,8 +216,8 @@ func (s *Syncer) Start(ctx context.Context) error {
 
 	// Initialize handlers
 	s.daRetriever = NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
-	s.asyncDARetriever = NewAsyncDARetriever(s.daRetriever, s.heightInCh, s.logger)
-	s.asyncDARetriever.Start(s.ctx)
+	s.daRetrievalWorkerPool = NewDARetrievalWorkerPool(s.daRetriever, s.heightInCh, s.logger)
+	s.daRetrievalWorkerPool.Start(s.ctx)
 	if s.config.Instrumentation.IsTracingEnabled() {
 		s.daRetriever = WithTracingDARetriever(s.daRetriever)
 	}
@@ -634,7 +634,7 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 					Msg("P2P event with DA height hint, triggering targeted DA retrieval")
 
 				// Trigger targeted DA retrieval in background via worker pool
-				s.asyncDARetriever.RequestRetrieval(daHeightHint)
+				s.daRetrievalWorkerPool.RequestRetrieval(daHeightHint)
 			}
 		}
 	}

diff --git a/block/internal/syncing/syncer_test.go b/block/internal/syncing/syncer_test.go
@@ -741,14 +741,14 @@ func TestProcessHeightEvent_TriggersAsyncDARetrieval(t *testing.T) {
 	require.NoError(t, s.initializeState())
 	s.ctx = context.Background()
 
-	// Mock AsyncDARetriever
+	// Mock DARetrievalWorkerPool
 	mockRetriever := NewMockDARetriever(t)
-	asyncRetriever := NewAsyncDARetriever(mockRetriever, s.heightInCh, zerolog.Nop())
-	// We don't start the async retriever to avoid race conditions in test,
+	workerPool := NewDARetrievalWorkerPool(mockRetriever, s.heightInCh, zerolog.Nop())
+	// We don't start the worker pool to avoid race conditions in test,
 	// we just want to verify RequestRetrieval queues the request.
 	// However, RequestRetrieval writes to a channel, so we need a consumer or a buffered channel.
 	// The workCh is buffered (100), so we are good.
-	s.asyncDARetriever = asyncRetriever
+	s.daRetrievalWorkerPool = workerPool
 
 	// Create event with DA height hint
 	evt := common.DAHeightEvent{
@@ -778,9 +778,9 @@ func TestProcessHeightEvent_TriggersAsyncDARetrieval(t *testing.T) {
 
 	s.processHeightEvent(t.Context(), &evt)
 
-	// Verify that the request was queued in the async retriever
+	// Verify that the request was queued in the worker pool
 	select {
-	case h := <-asyncRetriever.workCh:
+	case h := <-workerPool.workCh:
 		assert.Equal(t, uint64(100), h)
 	default:
 		t.Fatal("expected DA retrieval request to be queued")