more clean-ups

Author: julienrbrt

block/internal/syncing/syncer.go

@@ -52,8 +52,8 @@ type Syncer struct {
 	// State management
 	lastState *atomic.Pointer[types.State]
 
-	// DA state
-	daHeight *atomic.Uint64
+	// DA retriever height
+	daRetrieverHeight *atomic.Uint64
 
 	// P2P stores
 	headerStore common.Broadcaster[*types.SignedHeader]
@@ -95,21 +95,21 @@ func NewSyncer(
 	errorCh chan<- error,
 ) *Syncer {
 	return &Syncer{
-		store:       store,
-		exec:        exec,
-		da:          da,
-		cache:       cache,
-		metrics:     metrics,
-		config:      config,
-		genesis:     genesis,
-		options:     options,
-		headerStore: headerStore,
-		dataStore:   dataStore,
-		lastState:   &atomic.Pointer[types.State]{},
-		daHeight:    &atomic.Uint64{},
-		heightInCh:  make(chan common.DAHeightEvent, 1_000),
-		errorCh:     errorCh,
-		logger:      logger.With().Str("component", "syncer").Logger(),
+		store:             store,
+		exec:              exec,
+		da:                da,
+		cache:             cache,
+		metrics:           metrics,
+		config:            config,
+		genesis:           genesis,
+		options:           options,
+		headerStore:       headerStore,
+		dataStore:         dataStore,
+		lastState:         &atomic.Pointer[types.State]{},
+		daRetrieverHeight: &atomic.Uint64{},
+		heightInCh:        make(chan common.DAHeightEvent, 1_000),
+		errorCh:           errorCh,
+		logger:            logger.With().Str("component", "syncer").Logger(),
 	}
 }
 
@@ -175,16 +175,6 @@ func (s *Syncer) SetLastState(state types.State) {
 	s.lastState.Store(&state)
 }
 
-// GetDAHeight returns the current DA height
-func (s *Syncer) GetDAHeight() uint64 {
-	return s.daHeight.Load()
-}
-
-// SetDAHeight updates the DA height
-func (s *Syncer) SetDAHeight(height uint64) {
-	s.daHeight.Store(height)
-}
-
 // initializeState loads the current sync state
 func (s *Syncer) initializeState() error {
 	// Load state from store
@@ -218,11 +208,11 @@ func (s *Syncer) initializeState() error {
 
 	// Set DA height to the maximum of the genesis start height, the state's DA height, and the cached DA height.
 	// This ensures we resume from the highest known DA height, even if the cache is cleared on restart. If the DA height is too high because of a user error, reset it with --evnode.clear_cache. The DA height will be back to the last highest known executed DA height for a height.
-	s.SetDAHeight(max(s.genesis.DAStartHeight, s.cache.DaHeight(), state.DAHeight))
+	s.daRetrieverHeight.Store(max(s.genesis.DAStartHeight, s.cache.DaHeight(), state.DAHeight))
 
 	s.logger.Info().
 		Uint64("height", state.LastBlockHeight).
-		Uint64("da_height", s.GetDAHeight()).
+		Uint64("da_height", s.daRetrieverHeight.Load()).
 		Str("chain_id", state.ChainID).
 		Msg("initialized syncer state")
 
@@ -296,13 +286,13 @@ func (s *Syncer) fetchDAUntilCaughtUp() error {
 		default:
 		}
 
-		daHeight := max(s.GetDAHeight(), s.cache.DaHeight())
+		daHeight := max(s.daRetrieverHeight.Load(), s.cache.DaHeight())
 
 		events, err := s.daRetriever.RetrieveFromDA(s.ctx, daHeight)
 		if err != nil {
 			switch {
 			case errors.Is(err, coreda.ErrBlobNotFound):
-				s.SetDAHeight(daHeight + 1)
+				s.daRetrieverHeight.Store(daHeight + 1)
 				continue // Fetch next height immediately
 			case errors.Is(err, coreda.ErrHeightFromFuture):
 				s.logger.Debug().Err(err).Uint64("da_height", daHeight).Msg("DA is ahead of local target; backing off future height requests")
@@ -327,8 +317,8 @@ func (s *Syncer) fetchDAUntilCaughtUp() error {
 			}
 		}
 
-		// increment DA height on successful retrieval
-		s.SetDAHeight(daHeight + 1)
+		// increment DA retrieval height on successful retrieval
+		s.daRetrieverHeight.Store(daHeight + 1)
 	}
 }
 
@@ -534,13 +524,14 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 		return err
 	}
 
-	// Apply block
 	newState, err := s.applyBlock(header.Header, data, currentState)
 	if err != nil {
 		return fmt.Errorf("failed to apply block: %w", err)
 	}
 
 	// Update DA height if needed
+	// This height is only updated when an height is process from DA as P2P
+	// events do not contain DA height information
 	if event.DaHeight > newState.DAHeight {
 		newState.DAHeight = event.DaHeight
 	}
@@ -666,15 +657,6 @@ func (s *Syncer) sendCriticalError(err error) {
 	}
 }
 
-// sendNonBlockingSignal sends a signal without blocking
-func (s *Syncer) sendNonBlockingSignal(ch chan struct{}, name string) {
-	select {
-	case ch <- struct{}{}:
-	default:
-		s.logger.Debug().Str("channel", name).Msg("channel full, signal dropped")
-	}
-}
-
 // processPendingEvents fetches and processes pending events from cache
 // optimistically fetches the next events from cache until no matching heights are found
 func (s *Syncer) processPendingEvents() {

block/internal/syncing/syncer_benchmark_test.go

@@ -58,7 +58,7 @@ func BenchmarkSyncerIO(b *testing.B) {
 				}
 				require.Len(b, fixt.s.heightInCh, 0)
 
-				assert.Equal(b, spec.heights+daHeightOffset, fixt.s.daHeight)
+				assert.Equal(b, spec.heights+daHeightOffset, fixt.s.daRetrieverHeight)
 				gotStoreHeight, err := fixt.s.store.Height(b.Context())
 				require.NoError(b, err)
 				assert.Equal(b, spec.heights, gotStoreHeight)

block/internal/syncing/syncer_test.go

@@ -274,23 +274,6 @@ func TestSequentialBlockSync(t *testing.T) {
 	requireEmptyChan(t, errChan)
 }
 
-func TestSyncer_sendNonBlockingSignal(t *testing.T) {
-	s := &Syncer{logger: zerolog.Nop()}
-	ch := make(chan struct{}, 1)
-	ch <- struct{}{}
-	done := make(chan struct{})
-	go func() {
-		s.sendNonBlockingSignal(ch, "test")
-		close(done)
-	}()
-	select {
-	case <-done:
-		// ok
-	case <-time.After(200 * time.Millisecond):
-		t.Fatal("sendNonBlockingSignal blocked unexpectedly")
-	}
-}
-
 func TestSyncer_processPendingEvents(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)
@@ -432,7 +415,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 	requireEmptyChan(t, errorCh)
 
 	t.Log("sync workers on instance1 completed")
-	require.Equal(t, myFutureDAHeight, syncerInst1.GetDAHeight())
+	require.Equal(t, myFutureDAHeight, syncerInst1.daRetrieverHeight.Load())
 
 	// wait for all events consumed
 	require.NoError(t, cacheMgr.SaveToDisk())
@@ -482,7 +465,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 		Run(func(arg mock.Arguments) {
 			cancel()
 			// retrieve last one again
-			assert.Equal(t, syncerInst2.GetDAHeight(), arg.Get(1).(uint64))
+			assert.Equal(t, syncerInst2.daRetrieverHeight.Load(), arg.Get(1).(uint64))
 		}).
 		Return(nil, nil)
 
@@ -623,14 +606,14 @@ func TestSyncer_InitializeState_CallsReplayer(t *testing.T) {
 
 	// Create syncer with minimal dependencies
 	syncer := &Syncer{
-		store:     mockStore,
-		exec:      mockExec,
-		genesis:   gen,
-		lastState: &atomic.Pointer[types.State]{},
-		daHeight:  &atomic.Uint64{},
-		logger:    zerolog.Nop(),
-		ctx:       context.Background(),
-		cache:     cm,
+		store:             mockStore,
+		exec:              mockExec,
+		genesis:           gen,
+		lastState:         &atomic.Pointer[types.State]{},
+		daRetrieverHeight: &atomic.Uint64{},
+		logger:            zerolog.Nop(),
+		ctx:               context.Background(),
+		cache:             cm,
 	}
 
 	// Initialize state - this should call Replayer