feat(syncer): fetch from p2p and da (#2712)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

#2712 (comment)

Always fetch both p2p and DA in order to mark height fetched by DA da
included.

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->

Author: julienrbrt

block/internal/cache/generic_cache.go

@@ -104,6 +104,11 @@ func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64) {
 	c.daIncluded.Store(hash, daHeight)
 }
 
+// removeDAIncluded removes the DA-included status of the hash
+func (c *Cache[T]) removeDAIncluded(hash string) {
+	c.daIncluded.Delete(hash)
+}
+
 const (
 	itemsByHeightFilename = "items_by_height.gob"
 	hashesFilename        = "hashes.gob"

block/internal/cache/manager.go

@@ -43,6 +43,7 @@ type Manager interface {
 	SetHeaderSeen(hash string)
 	GetHeaderDAIncluded(hash string) (uint64, bool)
 	SetHeaderDAIncluded(hash string, daHeight uint64)
+	RemoveHeaderDAIncluded(hash string)
 
 	// Data operations
 	IsDataSeen(hash string) bool
@@ -141,6 +142,10 @@ func (m *implementation) SetHeaderDAIncluded(hash string, daHeight uint64) {
 	m.headerCache.setDAIncluded(hash, daHeight)
 }
 
+func (m *implementation) RemoveHeaderDAIncluded(hash string) {
+	m.headerCache.removeDAIncluded(hash)
+}
+
 // Data operations
 func (m *implementation) IsDataSeen(hash string) bool {
 	return m.dataCache.isSeen(hash)

block/internal/common/types.go

@@ -8,6 +8,4 @@ type DAHeightEvent struct {
 	Data   *types.Data
 	// DaHeight corresponds to the highest DA included height between the Header and Data.
 	DaHeight uint64
-	// HeaderDaIncludedHeight corresponds to the DA height at which the Header was included.
-	HeaderDaIncludedHeight uint64
 }

block/internal/syncing/da_retriever.go

@@ -35,9 +35,8 @@ type DARetriever struct {
 
 	// transient cache, only full event need to be passed to the syncer
 	// on restart, will be refetch as da height is updated by syncer
-	pendingHeaders  map[uint64]*types.SignedHeader
-	pendingData     map[uint64]*types.Data
-	headerDAHeights map[uint64]uint64
+	pendingHeaders map[uint64]*types.SignedHeader
+	pendingData    map[uint64]*types.Data
 }
 
 // NewDARetriever creates a new DA retriever
@@ -59,7 +58,6 @@ func NewDARetriever(
 		namespaceDataBz: coreda.NamespaceFromString(config.DA.GetDataNamespace()).Bytes(),
 		pendingHeaders:  make(map[uint64]*types.SignedHeader),
 		pendingData:     make(map[uint64]*types.Data),
-		headerDAHeights: make(map[uint64]uint64), // Track DA height for each header
 	}
 }
 
@@ -166,7 +164,6 @@ func (r *DARetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 
 		if header := r.tryDecodeHeader(bz, daHeight); header != nil {
 			r.pendingHeaders[header.Height()] = header
-			r.headerDAHeights[header.Height()] = daHeight
 			continue
 		}
 
@@ -180,14 +177,12 @@ func (r *DARetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 	// Match headers with data and create events
 	for height, header := range r.pendingHeaders {
 		data := r.pendingData[height]
-		includedHeight := r.headerDAHeights[height]
 
 		// Handle empty data case
 		if data == nil {
 			if r.isEmptyDataExpected(header) {
 				data = r.createEmptyDataForHeader(ctx, header)
 				delete(r.pendingHeaders, height)
-				delete(r.headerDAHeights, height)
 			} else {
 				// keep header in pending headers until data lands
 				r.logger.Debug().Uint64("height", height).Msg("header found but no matching data")
@@ -196,15 +191,13 @@ func (r *DARetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 		} else {
 			delete(r.pendingHeaders, height)
 			delete(r.pendingData, height)
-			delete(r.headerDAHeights, height)
 		}
 
 		// Create height event
 		event := common.DAHeightEvent{
-			Header:                 header,
-			Data:                   data,
-			DaHeight:               daHeight,
-			HeaderDaIncludedHeight: includedHeight,
+			Header:   header,
+			Data:     data,
+			DaHeight: daHeight,
 		}
 
 		events = append(events, event)
@@ -240,9 +233,17 @@ func (r *DARetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 		return nil
 	}
 
-	// note, we cannot mark the header as DA included
-	// we haven't done any signature verification check here
-	// signature verification happens with data.
+	// Optimistically mark as DA included
+	// This has to be done for all fetched DA headers prior to validation because P2P does not confirm
+	// da inclusion. This is not an issue, as an invalid header will be rejected. There cannot be hash collisions.
+	headerHash := header.Hash().String()
+	r.cache.SetHeaderDAIncluded(headerHash, daHeight)
+
+	r.logger.Info().
+		Str("header_hash", headerHash).
+		Uint64("da_height", daHeight).
+		Uint64("height", header.Height()).
+		Msg("optimistically marked header as DA included")
 
 	return header
 }

block/internal/syncing/da_retriever_test.go

@@ -196,7 +196,14 @@ func TestDARetriever_ProcessBlobs_HeaderAndData_Success(t *testing.T) {
 	assert.Equal(t, uint64(2), events[0].Header.Height())
 	assert.Equal(t, uint64(2), events[0].Data.Height())
 	assert.Equal(t, uint64(77), events[0].DaHeight)
-	assert.Equal(t, uint64(77), events[0].HeaderDaIncludedHeight)
+
+	hHeight, ok := r.cache.GetHeaderDAIncluded(events[0].Header.Hash().String())
+	assert.True(t, ok)
+	assert.Equal(t, uint64(77), hHeight)
+
+	dHeight, ok := r.cache.GetDataDAIncluded(events[0].Data.DACommitment().String())
+	assert.True(t, ok)
+	assert.Equal(t, uint64(77), dHeight)
 }
 
 func TestDARetriever_ProcessBlobs_HeaderOnly_EmptyDataExpected(t *testing.T) {
@@ -217,6 +224,14 @@ func TestDARetriever_ProcessBlobs_HeaderOnly_EmptyDataExpected(t *testing.T) {
 	assert.Equal(t, uint64(3), events[0].Header.Height())
 	assert.NotNil(t, events[0].Data)
 	assert.Equal(t, uint64(88), events[0].DaHeight)
+
+	hHeight, ok := r.cache.GetHeaderDAIncluded(events[0].Header.Hash().String())
+	assert.True(t, ok)
+	assert.Equal(t, uint64(88), hHeight)
+
+	// empty data is not marked as data included (the submitter components does handle the empty data case)
+	_, ok = r.cache.GetDataDAIncluded(events[0].Data.DACommitment().String())
+	assert.False(t, ok)
 }
 
 func TestDARetriever_TryDecodeHeaderAndData_Basic(t *testing.T) {
@@ -346,8 +361,6 @@ func TestDARetriever_ProcessBlobs_CrossDAHeightMatching(t *testing.T) {
 
 	// Verify header is stored in pending headers
 	require.Contains(t, r.pendingHeaders, uint64(5), "header should be stored as pending")
-	require.Contains(t, r.headerDAHeights, uint64(5), "header DA height should be tracked")
-	assert.Equal(t, uint64(100), r.headerDAHeights[5])
 
 	// Process data from DA height 102
 	events2 := r.processBlobs(context.Background(), [][]byte{dataBin}, 102)
@@ -357,12 +370,10 @@ func TestDARetriever_ProcessBlobs_CrossDAHeightMatching(t *testing.T) {
 	assert.Equal(t, uint64(5), event.Header.Height())
 	assert.Equal(t, uint64(5), event.Data.Height())
 	assert.Equal(t, uint64(102), event.DaHeight, "DaHeight should be the height where data was processed")
-	assert.Equal(t, uint64(100), event.HeaderDaIncludedHeight, "HeaderDaIncludedHeight should be where header was included")
 
 	// Verify pending maps are cleared
 	require.NotContains(t, r.pendingHeaders, uint64(5), "header should be removed from pending")
 	require.NotContains(t, r.pendingData, uint64(5), "data should be removed from pending")
-	require.NotContains(t, r.headerDAHeights, uint64(5), "header DA height should be removed")
 }
 
 func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testing.T) {
@@ -393,9 +404,6 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	require.Contains(t, r.pendingHeaders, uint64(3), "header 3 should be pending")
 	require.Contains(t, r.pendingHeaders, uint64(4), "header 4 should be pending")
 	require.Contains(t, r.pendingHeaders, uint64(5), "header 5 should be pending")
-	assert.Equal(t, uint64(200), r.headerDAHeights[3])
-	assert.Equal(t, uint64(200), r.headerDAHeights[4])
-	assert.Equal(t, uint64(200), r.headerDAHeights[5])
 
 	// Process some data from DA height 203 - should create partial events
 	events2 := r.processBlobs(context.Background(), [][]byte{data3Bin, data5Bin}, 203)
@@ -410,13 +418,11 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	assert.Equal(t, uint64(3), events2[0].Header.Height())
 	assert.Equal(t, uint64(3), events2[0].Data.Height())
 	assert.Equal(t, uint64(203), events2[0].DaHeight)
-	assert.Equal(t, uint64(200), events2[0].HeaderDaIncludedHeight)
 
 	// Verify event for height 5
 	assert.Equal(t, uint64(5), events2[1].Header.Height())
 	assert.Equal(t, uint64(5), events2[1].Data.Height())
 	assert.Equal(t, uint64(203), events2[1].DaHeight)
-	assert.Equal(t, uint64(200), events2[1].HeaderDaIncludedHeight)
 
 	// Verify header 4 is still pending (no matching data yet)
 	require.Contains(t, r.pendingHeaders, uint64(4), "header 4 should still be pending")
@@ -431,11 +437,9 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	assert.Equal(t, uint64(4), events3[0].Header.Height())
 	assert.Equal(t, uint64(4), events3[0].Data.Height())
 	assert.Equal(t, uint64(205), events3[0].DaHeight)
-	assert.Equal(t, uint64(200), events3[0].HeaderDaIncludedHeight)
 
 	// Verify all pending maps are now clear
 	require.NotContains(t, r.pendingHeaders, uint64(4), "header 4 should be removed from pending")
 	require.Len(t, r.pendingHeaders, 0, "all headers should be processed")
 	require.Len(t, r.pendingData, 0, "all data should be processed")
-	require.Len(t, r.headerDAHeights, 0, "all header DA heights should be cleared")
 }

block/internal/syncing/p2p_handler.go

@@ -86,10 +86,9 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 
 		// Create height event
 		event := common.DAHeightEvent{
-			Header:                 header,
-			Data:                   data,
-			DaHeight:               0, // P2P events don't have DA height context
-			HeaderDaIncludedHeight: 0, // P2P events don't have DA height context
+			Header:   header,
+			Data:     data,
+			DaHeight: 0, // P2P events don't have DA height context
 		}
 
 		events = append(events, event)
@@ -139,10 +138,9 @@ func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeigh
 
 		// Create height event
 		event := common.DAHeightEvent{
-			Header:                 header,
-			Data:                   data,
-			DaHeight:               0, // P2P events don't have DA height context
-			HeaderDaIncludedHeight: 0, // P2P events don't have DA height context
+			Header:   header,
+			Data:     data,
+			DaHeight: 0, // P2P events don't have DA height context
 		}
 
 		events = append(events, event)