Merge branch 'main' into julien/sync-p2p

Author: julienrbrt

block/internal/syncing/da_retriever.go

@@ -189,8 +189,8 @@ func (r *DARetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 
 		// Handle empty data case
 		if data == nil {
-			if r.isEmptyDataExpected(header) {
-				data = r.createEmptyDataForHeader(ctx, header)
+			if isEmptyDataExpected(header) {
+				data = createEmptyDataForHeader(ctx, header)
 				delete(r.pendingHeaders, height)
 			} else {
 				// keep header in pending headers until data lands
@@ -289,24 +289,6 @@ func (r *DARetriever) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 	return &signedData.Data
 }
 
-// isEmptyDataExpected checks if empty data is expected for a header
-func (r *DARetriever) isEmptyDataExpected(header *types.SignedHeader) bool {
-	return len(header.DataHash) == 0 || bytes.Equal(header.DataHash, common.DataHashForEmptyTxs)
-}
-
-// createEmptyDataForHeader creates empty data for a header
-func (r *DARetriever) createEmptyDataForHeader(ctx context.Context, header *types.SignedHeader) *types.Data {
-	return &types.Data{
-		Txs: make(types.Txs, 0),
-		Metadata: &types.Metadata{
-			ChainID:      header.ChainID(),
-			Height:       header.Height(),
-			Time:         header.BaseHeader.Time,
-			LastDataHash: nil, // LastDataHash must be filled in the syncer, as it is not available here, block n-1 has not been processed yet.
-		},
-	}
-}
-
 // assertExpectedProposer validates the proposer address
 func (r *DARetriever) assertExpectedProposer(proposerAddr []byte) error {
 	if string(proposerAddr) != string(r.genesis.ProposerAddress) {
@@ -342,3 +324,21 @@ func (r *DARetriever) assertValidSignedData(signedData *types.SignedData) error
 
 	return nil
 }
+
+// isEmptyDataExpected checks if empty data is expected for a header
+func isEmptyDataExpected(header *types.SignedHeader) bool {
+	return len(header.DataHash) == 0 || bytes.Equal(header.DataHash, common.DataHashForEmptyTxs)
+}
+
+// createEmptyDataForHeader creates empty data for a header
+func createEmptyDataForHeader(ctx context.Context, header *types.SignedHeader) *types.Data {
+	return &types.Data{
+		Txs: make(types.Txs, 0),
+		Metadata: &types.Metadata{
+			ChainID:      header.ChainID(),
+			Height:       header.Height(),
+			Time:         header.BaseHeader.Time,
+			LastDataHash: nil, // LastDataHash must be filled in the syncer, as it is not available here, block n-1 has not been processed yet.
+		},
+	}
+}

block/internal/syncing/da_retriever_test.go

@@ -238,16 +238,6 @@ func TestDARetriever_TryDecodeHeaderAndData_Basic(t *testing.T) {
 	assert.Nil(t, r.tryDecodeData([]byte("junk"), 1))
 }
 
-func TestDARetriever_isEmptyDataExpected(t *testing.T) {
-	r := &DARetriever{}
-	h := &types.SignedHeader{}
-	// when DataHash is nil/empty -> expected empty
-	assert.True(t, r.isEmptyDataExpected(h))
-	// when equals to predefined emptyTxs hash -> expected empty
-	h.DataHash = common.DataHashForEmptyTxs
-	assert.True(t, r.isEmptyDataExpected(h))
-}
-
 func TestDARetriever_tryDecodeData_InvalidSignatureOrProposer(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)
@@ -422,3 +412,12 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	require.Len(t, r.pendingHeaders, 0, "all headers should be processed")
 	require.Len(t, r.pendingData, 0, "all data should be processed")
 }
+
+func Test_isEmptyDataExpected(t *testing.T) {
+	h := &types.SignedHeader{}
+	// when DataHash is nil/empty -> expected empty
+	assert.True(t, isEmptyDataExpected(h))
+	// when equals to predefined emptyTxs hash -> expected empty
+	h.DataHash = common.DataHashForEmptyTxs
+	assert.True(t, isEmptyDataExpected(h))
+}

block/internal/syncing/syncer.go

@@ -241,33 +241,15 @@ func (s *Syncer) syncLoop() {
 	nextDARequestAt := &time.Time{}
 
 	for {
-		wg := sync.WaitGroup{}
-
 		select {
 		case <-s.ctx.Done():
 			return
 		default:
 		}
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			s.processPendingEvents()
-		}()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			s.tryFetchFromP2P()
-		}()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			s.tryFetchFromDA(nextDARequestAt)
-		}()
-
-		// wait for pending events processing, p2p and da fetching
-		wg.Wait()
+
+		s.processPendingEvents()
+		s.tryFetchFromP2P()
+		s.tryFetchFromDA(nextDARequestAt)
 
 		// Prevent busy-waiting when no events are processed
 		select {
@@ -310,6 +292,7 @@ func (s *Syncer) tryFetchFromDA(nextDARequestAt *time.Time) {
 		*nextDARequestAt = now.Add(backoffDelay)
 
 		s.logger.Error().Err(err).Dur("delay", backoffDelay).Uint64("da_height", daHeight).Msg("failed to retrieve from DA; backing off DA requests")
+
 		return
 	}
 

test/e2e/evm_full_node_e2e_test.go

@@ -426,7 +426,7 @@ func TestEvmSequencerWithFullNodeE2E(t *testing.T) {
 	t.Logf("Full node block height before DA inclusion wait: %d", fnBlockHeightBeforeWait)
 
 	// Wait a few seconds to allow DA inclusion to process
-	waitTime := 5 * time.Second
+	waitTime := 4 * time.Second
 	t.Logf("Waiting %v for DA inclusion to process...", waitTime)
 	time.Sleep(waitTime)