fix errors

Author: tac0turtle

block/internal/syncing/p2p_handler.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"sync"
+	"sync/atomic"
 
 	goheader "github.com/celestiaorg/go-header"
 	"github.com/rs/zerolog"
@@ -28,8 +28,7 @@ type P2PHandler struct {
 	genesis     genesis.Genesis
 	logger      zerolog.Logger
 
-	mu              sync.Mutex
-	processedHeight uint64 // highest block fully applied by the syncer
+	processedHeight atomic.Uint64
 }
 
 // NewP2PHandler creates a new P2P handler.
@@ -51,22 +50,22 @@ func NewP2PHandler(
 
 // SetProcessedHeight updates the highest processed block height.
 func (h *P2PHandler) SetProcessedHeight(height uint64) {
-	h.mu.Lock()
-	if height > h.processedHeight {
-		h.processedHeight = height
+	for {
+		current := h.processedHeight.Load()
+		if height <= current {
+			return
+		}
+		if h.processedHeight.CompareAndSwap(current, height) {
+			return
+		}
 	}
-	h.mu.Unlock()
 }
 
 // ProcessHeight retrieves and validates both header and data for the given height from P2P stores.
 // It blocks until both are available, validates consistency (proposer address and data hash match),
 // then emits the event to heightInCh or stores it as pending. Updates processedHeight on success.
 func (h *P2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInCh chan<- common.DAHeightEvent) error {
-	h.mu.Lock()
-	shouldProcess := height > h.processedHeight
-	h.mu.Unlock()
-
-	if !shouldProcess {
+	if height <= h.processedHeight.Load() {
 		return nil
 	}
 
@@ -110,11 +109,7 @@ func (h *P2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInC
 		h.cache.SetPendingEvent(event.Header.Height(), &event)
 	}
 
-	h.mu.Lock()
-	if height > h.processedHeight {
-		h.processedHeight = height
-	}
-	h.mu.Unlock()
+	h.SetProcessedHeight(height)
 
 	h.logger.Debug().Uint64("height", height).Msg("processed event from P2P")
 	return nil

block/internal/syncing/syncer.go

@@ -76,8 +76,7 @@ type Syncer struct {
 	wg     sync.WaitGroup
 
 	// P2P wait coordination
-	p2pWaitMu    sync.Mutex
-	p2pWaitState p2pWaitState
+	p2pWaitState atomic.Value // stores p2pWaitState
 }
 
 // NewSyncer creates a new block syncer
@@ -257,10 +256,6 @@ func (s *Syncer) startSyncWorkers() {
 	go s.p2pWorkerLoop()
 }
 
-const (
-	futureHeightBackoff = 6 * time.Second // current celestia block time
-)
-
 func (s *Syncer) daWorkerLoop() {
 	defer s.wg.Done()
 
@@ -277,7 +272,7 @@ func (s *Syncer) daWorkerLoop() {
 		var backoff time.Duration
 		if err == nil {
 			// No error, means we are caught up.
-			backoff = futureHeightBackoff
+			backoff = s.config.DA.BlockTime.Duration
 		} else {
 			// Error, back off for a shorter duration.
 			backoff = s.config.DA.BlockTime.Duration
@@ -772,20 +767,21 @@ type p2pWaitState struct {
 }
 
 func (s *Syncer) setP2PWaitState(height uint64, cancel context.CancelFunc) {
-	s.p2pWaitMu.Lock()
-	defer s.p2pWaitMu.Unlock()
-	s.p2pWaitState = p2pWaitState{
-		height: height,
-		cancel: cancel,
-	}
+	s.p2pWaitState.Store(p2pWaitState{height: height, cancel: cancel})
 }
 
 func (s *Syncer) cancelP2PWait(height uint64) {
-	s.p2pWaitMu.Lock()
-	defer s.p2pWaitMu.Unlock()
+	val := s.p2pWaitState.Load()
+	if val == nil {
+		return
+	}
+	state, ok := val.(p2pWaitState)
+	if !ok || state.cancel == nil {
+		return
+	}
 
-	if s.p2pWaitState.cancel != nil && (height == 0 || s.p2pWaitState.height <= height) {
-		s.p2pWaitState.cancel()
-		s.p2pWaitState = p2pWaitState{}
+	if height == 0 || state.height <= height {
+		s.p2pWaitState.Store(p2pWaitState{})
+		state.cancel()
 	}
 }