align p2p handler with p2p broadcast

Author: julienrbrt

block/internal/syncing/p2p_handler.go

@@ -77,29 +77,23 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 			continue
 		}
 
-		// Get corresponding data
+		// Get corresponding data (empty data are still broadcasted by peers)
 		var data *types.Data
-		if bytes.Equal(header.DataHash, common.DataHashForEmptyTxs) {
-			// Create empty data for headers with empty data hash
-			data = h.createEmptyDataForHeader(ctx, header)
-		} else {
-			// Try to get data from data store with timeout
-			timeoutCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
-			retrievedData, err := h.dataStore.GetByHeight(timeoutCtx, height)
-			cancel()
-
-			if err != nil {
-				if errors.Is(err, context.DeadlineExceeded) {
-					h.logger.Debug().Uint64("height", height).Msg("timeout waiting for data from store, will retry later")
-					// Don't continue processing if data is not available
-					// Store event with header only for later processing
-					continue
-				}
-				h.logger.Debug().Uint64("height", height).Err(err).Msg("could not retrieve data for header from data store")
+		timeoutCtx, cancel = context.WithTimeout(ctx, 500*time.Millisecond)
+		retrievedData, err := h.dataStore.GetByHeight(timeoutCtx, height)
+		cancel()
+
+		if err != nil {
+			if errors.Is(err, context.DeadlineExceeded) {
+				h.logger.Debug().Uint64("height", height).Msg("timeout waiting for data from store, will retry later")
+				// Don't continue processing if data is not available
+				// Store event with header only for later processing
 				continue
 			}
-			data = retrievedData
+			h.logger.Debug().Uint64("height", height).Err(err).Msg("could not retrieve data for header from data store")
+			continue
 		}
+		data = retrievedData
 
 		// further header validation (signature) is done in validateBlock.
 		// we need to be sure that the previous block n-1 was executed before validating block n
@@ -201,34 +195,3 @@ func (h *P2PHandler) assertExpectedProposer(proposerAddr []byte) error {
 	}
 	return nil
 }
-
-// createEmptyDataForHeader creates empty data for headers with empty data hash
-func (h *P2PHandler) createEmptyDataForHeader(ctx context.Context, header *types.SignedHeader) *types.Data {
-	headerHeight := header.Height()
-	var lastDataHash types.Hash
-
-	if headerHeight > 1 {
-		// Try to get previous data hash with a short timeout, but don't fail if not available
-		timeoutCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
-		prevData, err := h.dataStore.GetByHeight(timeoutCtx, headerHeight-1)
-		cancel()
-
-		if err == nil && prevData != nil {
-			lastDataHash = prevData.Hash()
-		} else {
-			h.logger.Debug().Uint64("current_height", headerHeight).Uint64("previous_height", headerHeight-1).
-				Msg("previous block not available, using empty last data hash")
-		}
-	}
-
-	metadata := &types.Metadata{
-		ChainID:      header.ChainID(),
-		Height:       headerHeight,
-		Time:         header.BaseHeader.Time,
-		LastDataHash: lastDataHash,
-	}
-
-	return &types.Data{
-		Metadata: metadata,
-	}
-}

block/internal/syncing/p2p_handler_test.go

@@ -222,46 +222,6 @@ func TestP2PHandler_ProposerMismatch_Rejected(t *testing.T) {
 	require.Len(t, events, 0)
 }
 
-func TestP2PHandler_CreateEmptyDataForHeader_UsesPreviousDataHash(t *testing.T) {
-	p2pData := setupP2P(t)
-	ctx := context.Background()
-
-	// Prepare a header at height 10
-	signedHeader := p2pMakeSignedHeader(t, p2pData.Genesis.ChainID, 10, p2pData.ProposerAddr, p2pData.ProposerPub, p2pData.Signer)
-	signedHeader.DataHash = common.DataHashForEmptyTxs
-
-	// Mock previous data at height 9 so handler can propagate its hash
-	previousData := makeData(p2pData.Genesis.ChainID, 9, 1)
-	p2pData.DataStore.EXPECT().GetByHeight(mock.Anything, uint64(9)).Return(previousData, nil).Once()
-
-	emptyData := p2pData.Handler.createEmptyDataForHeader(ctx, signedHeader)
-	require.NotNil(t, emptyData, "handler should synthesize empty data when header declares empty data hash")
-	require.Equal(t, p2pData.Genesis.ChainID, emptyData.ChainID(), "synthesized data should carry header chain ID")
-	require.Equal(t, uint64(10), emptyData.Height(), "synthesized data should carry header height")
-	require.Equal(t, signedHeader.BaseHeader.Time, emptyData.Metadata.Time, "synthesized data should carry header time")
-	require.Equal(t, previousData.Hash(), emptyData.LastDataHash, "synthesized data should propagate previous data hash")
-}
-
-func TestP2PHandler_CreateEmptyDataForHeader_NoPreviousData(t *testing.T) {
-	p2pData := setupP2P(t)
-	ctx := context.Background()
-
-	// Prepare a header at height 2 (previous height exists but will return error)
-	signedHeader := p2pMakeSignedHeader(t, p2pData.Genesis.ChainID, 2, p2pData.ProposerAddr, p2pData.ProposerPub, p2pData.Signer)
-	signedHeader.DataHash = common.DataHashForEmptyTxs
-
-	// Mock previous data fetch failure
-	p2pData.DataStore.EXPECT().GetByHeight(mock.Anything, uint64(1)).Return(nil, errors.New("not available")).Once()
-
-	emptyData := p2pData.Handler.createEmptyDataForHeader(ctx, signedHeader)
-	require.NotNil(t, emptyData, "handler should synthesize empty data even when previous data is unavailable")
-	require.Equal(t, p2pData.Genesis.ChainID, emptyData.ChainID(), "synthesized data should carry header chain ID")
-	require.Equal(t, uint64(2), emptyData.Height(), "synthesized data should carry header height")
-	require.Equal(t, signedHeader.BaseHeader.Time, emptyData.Metadata.Time, "synthesized data should carry header time")
-	// When no previous data is available, LastDataHash should be zero value
-	require.Equal(t, (types.Hash)(nil), emptyData.LastDataHash, "last data hash should be empty when previous data is not available")
-}
-
 func TestP2PHandler_ProcessHeaderRange_MultipleHeightsHappyPath(t *testing.T) {
 	p2pData := setupP2P(t)
 	ctx := context.Background()

block/internal/syncing/syncer.go

@@ -273,7 +273,7 @@ func (s *Syncer) syncLoop() {
 		select {
 		case <-s.ctx.Done():
 			return
-		case <-time.After(min(100*time.Millisecond, s.config.Node.BlockTime.Duration)):
+		case <-time.After(min(25*time.Millisecond, s.config.Node.BlockTime.Duration)):
 		}
 	}
 }

block/internal/syncing/syncer_backoff_test.go

@@ -154,7 +154,7 @@ func TestSyncer_BackoffOnDAError(t *testing.T) {
 				assert.GreaterOrEqual(t, callCount, 2, "should continue without significant delay")
 				if len(callTimes) >= 2 {
 					timeBetweenCalls := callTimes[1].Sub(callTimes[0])
-					assert.Less(t, timeBetweenCalls, 100*time.Millisecond,
+					assert.Less(t, timeBetweenCalls, 120*time.Millisecond,
 						"should not have backoff delay for ErrBlobNotFound")
 				}
 			}