remove worker pool

Author: julienrbrt

block/internal/syncing/da_retrieval_worker_pool.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
+	"sync"
 
 	"github.com/rs/zerolog"
 	"google.golang.org/protobuf/proto"
@@ -20,7 +22,13 @@ import (
 
 // DARetriever defines the interface for retrieving events from the DA layer
 type DARetriever interface {
+	// RetrieveFromDA retrieves blocks from the specified DA height and returns height events
 	RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
+	// QueuePriorityHeight queues a DA height for priority retrieval (from P2P hints).
+	// These heights take precedence over sequential fetching.
+	QueuePriorityHeight(daHeight uint64)
+	// PopPriorityHeight returns the next priority height to fetch, or 0 if none.
+	PopPriorityHeight() uint64
 }
 
 // daRetriever handles DA retrieval operations for syncing
@@ -38,6 +46,12 @@ type daRetriever struct {
 	// strictMode indicates if the node has seen a valid DAHeaderEnvelope
 	// and should now reject all legacy/unsigned headers.
 	strictMode bool
+
+	// priorityMu protects priorityHeights from concurrent access
+	priorityMu sync.Mutex
+	// priorityHeights holds DA heights from P2P hints that should be fetched
+	// before continuing sequential retrieval. Sorted in ascending order.
+	priorityHeights []uint64
 }
 
 // NewDARetriever creates a new DA retriever
@@ -48,14 +62,46 @@ func NewDARetriever(
 	logger zerolog.Logger,
 ) *daRetriever {
 	return &daRetriever{
-		client:         client,
-		cache:          cache,
-		genesis:        genesis,
-		logger:         logger.With().Str("component", "da_retriever").Logger(),
-		pendingHeaders: make(map[uint64]*types.SignedHeader),
-		pendingData:    make(map[uint64]*types.Data),
-		strictMode:     false,
+		client:          client,
+		cache:           cache,
+		genesis:         genesis,
+		logger:          logger.With().Str("component", "da_retriever").Logger(),
+		pendingHeaders:  make(map[uint64]*types.SignedHeader),
+		pendingData:     make(map[uint64]*types.Data),
+		strictMode:      false,
+		priorityHeights: make([]uint64, 0),
+	}
+}
+
+// QueuePriorityHeight queues a DA height for priority retrieval.
+// Heights from P2P hints take precedence over sequential fetching.
+func (r *daRetriever) QueuePriorityHeight(daHeight uint64) {
+	r.priorityMu.Lock()
+	defer r.priorityMu.Unlock()
+
+	// Skip if already queued
+	if slices.Contains(r.priorityHeights, daHeight) {
+		return
+	}
+
+	r.priorityHeights = append(r.priorityHeights, daHeight)
+	// Keep sorted in ascending order so we process lower heights first
+	slices.Sort(r.priorityHeights)
+}
+
+// PopPriorityHeight returns the next priority height to fetch, or 0 if none.
+func (r *daRetriever) PopPriorityHeight() uint64 {
+	r.priorityMu.Lock()
+	defer r.priorityMu.Unlock()
+
+	if len(r.priorityHeights) == 0 {
+		return 0
 	}
+
+	height := r.priorityHeights[0]
+	r.priorityHeights = r.priorityHeights[1:]
+
+	return height
 }
 
 // RetrieveFromDA retrieves blocks from the specified DA height and returns height events