feedback part 1.

Author: julienrbrt

block/manager.go

@@ -48,7 +48,7 @@ const (
 	namespaceMigrationKey = "namespace_migration_completed"
 
 	// Applies to the headerInCh and dataInCh, 20000 is a large enough number for headers per DA block.
-	eventInChLength = 20000
+	eventInChLength = 10000
 )
 
 var (
@@ -609,13 +609,13 @@ func (m *Manager) retrieveBatch(ctx context.Context) (*BatchData, error) {
 	return nil, ErrNoBatch
 }
 
-// isUsingExpectedSingleSequencer checks if the header is using the expected single sequencer.
-func (m *Manager) isUsingExpectedSingleSequencer(proposerAddress []byte) (bool, error) {
+// assertUsingExpectedSingleSequencer checks if the header is using the expected single sequencer.
+func (m *Manager) assertUsingExpectedSingleSequencer(proposerAddress []byte) error {
 	if !bytes.Equal(m.genesis.ProposerAddress, proposerAddress) {
-		return false, fmt.Errorf("proposer address is not the same as the genesis proposer address %x != %x", proposerAddress, m.genesis.ProposerAddress)
+		return fmt.Errorf("proposer address is not the same as the genesis proposer address %x != %x", proposerAddress, m.genesis.ProposerAddress)
 	}
 
-	return true, nil
+	return nil
 }
 
 // publishBlockInternal is the internal implementation for publishing a block.
@@ -870,7 +870,7 @@ func (m *Manager) execCreateBlock(_ context.Context, height uint64, lastSignatur
 		return nil, nil, fmt.Errorf("failed to get proposer address: %w", err)
 	}
 
-	if _, err := m.isUsingExpectedSingleSequencer(address); err != nil {
+	if err := m.assertUsingExpectedSingleSequencer(address); err != nil {
 		return nil, nil, err
 	}
 
@@ -1108,7 +1108,7 @@ func (m *Manager) isValidSignedData(signedData *types.SignedData) bool {
 		return false
 	}
 
-	if ok, _ := m.isUsingExpectedSingleSequencer(signedData.Signer.Address); !ok {
+	if err := m.assertUsingExpectedSingleSequencer(signedData.Signer.Address); err != nil {
 		return false
 	}
 

block/manager_test.go

@@ -713,12 +713,12 @@ func TestUtilityFunctions(t *testing.T) {
 		require.ErrorContains(err, "signer is nil; cannot sign header")
 	})
 
-	t.Run("IsUsingExpectedSingleSequencer", func(t *testing.T) {
+	t.Run("AssertUsingExpectedSingleSequencer", func(t *testing.T) {
 		require := require.New(t)
 		m, _ := getManager(t, mocks.NewMockDA(t), -1, -1)
 
 		// Create genesis data for the test
-		genesisData, privKey, _ := types.GetGenesisWithPrivkey("TestIsUsingExpectedSingleSequencer")
+		genesisData, privKey, _ := types.GetGenesisWithPrivkey("TestAssertUsingExpectedSingleSequencer")
 		m.genesis = genesisData
 
 		// Create a signer
@@ -751,15 +751,13 @@ func TestUtilityFunctions(t *testing.T) {
 		header.Signature = signature
 
 		// Should return true for valid header with correct proposer
-		ok, err := m.isUsingExpectedSingleSequencer(header.ProposerAddress)
+		err = m.assertUsingExpectedSingleSequencer(header.ProposerAddress)
 		require.NoError(err)
-		require.True(ok)
 
 		// Should return false for header with wrong proposer address
 		header.ProposerAddress = []byte("wrong-proposer")
-		ok, err = m.isUsingExpectedSingleSequencer(header.ProposerAddress)
+		err = m.assertUsingExpectedSingleSequencer(header.ProposerAddress)
 		require.Error(err)
-		require.False(ok)
 	})
 }
 

block/retriever_da.go

@@ -24,8 +24,8 @@ const (
 // daRetriever encapsulates DA retrieval with pending events management
 type daRetriever struct {
 	manager       *Manager
+	mutex         sync.RWMutex // mutex for pendingEvents
 	pendingEvents map[uint64][]pendingDAEvent
-	mutex         sync.RWMutex
 }
 
 // pendingDAEvent represents a DA event waiting for processing
@@ -200,7 +200,7 @@ func (m *Manager) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedHeade
 	}
 
 	// early validation to reject junk headers
-	if ok, err := m.isUsingExpectedSingleSequencer(header.ProposerAddress); !ok {
+	if err := m.assertUsingExpectedSingleSequencer(header.ProposerAddress); err != nil {
 		m.logger.Debug().
 			Uint64("headerHeight", header.Height()).
 			Str("headerHash", header.Hash().String()).
@@ -253,9 +253,8 @@ func (m *Manager) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 	return &signedData.Data
 }
 
-// isAtHeight checks if a height is available without blocking
-// waitForHeightNonBlocking checks if a height is available without blocking
-func (m *Manager) waitForHeightNonBlocking(ctx context.Context, height uint64) error {
+// isAtHeight checks if a height is available.
+func (m *Manager) isAtHeight(ctx context.Context, height uint64) error {
 	currentHeight, err := m.GetStoreHeight(ctx)
 	if err != nil {
 		return err
@@ -282,8 +281,8 @@ func (dr *daRetriever) sendHeightEventIfValid(ctx context.Context, header *types
 		return
 	}
 
-	// Check if we can validate this height immediately (non-blocking check)
-	if err := dr.manager.waitForHeightNonBlocking(ctx, header.Height()-1); err != nil {
+	// Check if we can validate this height immediately
+	if err := dr.manager.isAtHeight(ctx, header.Height()-1); err != nil {
 		// Queue this event for later processing when the prerequisite height is available
 		dr.queuePendingEvent(header, data, daHeight)
 		return
@@ -316,7 +315,7 @@ func (dr *daRetriever) queuePendingEvent(header *types.SignedHeader, data *types
 func (dr *daRetriever) processEvent(ctx context.Context, header *types.SignedHeader, data *types.Data, daHeight uint64) {
 	headerHash := header.Hash().String()
 
-	// Validate header with its data - this requires previous height to be stored
+	// Validate header with its data - some execution environment may require previous height to be stored
 	if err := header.ValidateBasicWithData(data); err != nil {
 		dr.manager.logger.Debug().Uint64("height", header.Height()).Err(err).Msg("header validation with data failed")
 		return
@@ -365,7 +364,7 @@ func (dr *daRetriever) processPendingEvents(ctx context.Context) {
 	defer dr.mutex.Unlock()
 
 	for height, events := range dr.pendingEvents {
-		if height-1 <= currentHeight {
+		if height <= currentHeight+1 {
 			for _, event := range events {
 				dr.manager.logger.Debug().
 					Uint64("height", height).

block/retriever_store.go

@@ -72,7 +72,7 @@ func (m *Manager) processHeaderStoreRange(ctx context.Context, startHeight, endH
 		}
 
 		// early validation to reject junk headers
-		if ok, _ := m.isUsingExpectedSingleSequencer(header.ProposerAddress); !ok {
+		if err := m.assertUsingExpectedSingleSequencer(header.ProposerAddress); err != nil {
 			continue
 		}
 
@@ -127,7 +127,7 @@ func (m *Manager) processDataStoreRange(ctx context.Context, startHeight, endHei
 		}
 
 		// early validation to reject junk headers
-		if ok, _ := m.isUsingExpectedSingleSequencer(header.ProposerAddress); !ok {
+		if err := m.assertUsingExpectedSingleSequencer(header.ProposerAddress); err != nil {
 			continue
 		}