refactor(syncing)!: simplify cache and syncing logic (#2679)

<!--
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

- Fix setting header da inclusion at the correct place (after header
validation)
- Simplify cache by deleting SetHeader and SetData, as we sync
sequentially, so we can always take the next event
- other code cleanup

<!-- 
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>
-->

---------

Co-authored-by: Marko <marko@baricevic.me>

Author: julienrbrt

block/internal/cache/manager.go

@@ -38,16 +38,12 @@ func registerGobTypes() {
 // Manager provides centralized cache management for both executing and syncing components
 type Manager interface {
 	// Header operations
-	GetHeader(height uint64) *types.SignedHeader
-	SetHeader(height uint64, header *types.SignedHeader)
 	IsHeaderSeen(hash string) bool
 	SetHeaderSeen(hash string)
 	GetHeaderDAIncluded(hash string) (uint64, bool)
 	SetHeaderDAIncluded(hash string, daHeight uint64)
 
 	// Data operations
-	GetData(height uint64) *types.Data
-	SetData(height uint64, data *types.Data)
 	IsDataSeen(hash string) bool
 	SetDataSeen(hash string)
 	GetDataDAIncluded(hash string) (uint64, bool)
@@ -58,18 +54,15 @@ type Manager interface {
 	GetPendingData(ctx context.Context) ([]*types.SignedData, error)
 	SetLastSubmittedHeaderHeight(ctx context.Context, height uint64)
 	SetLastSubmittedDataHeight(ctx context.Context, height uint64)
-
 	NumPendingHeaders() uint64
 	NumPendingData() uint64
 
-	// Pending events for DA coordination
+	// Pending events syncing coordination
 	SetPendingEvent(height uint64, event *common.DAHeightEvent)
 	GetPendingEvents() map[uint64]*common.DAHeightEvent
 	DeletePendingEvent(height uint64)
 
 	// Cleanup operations
-	ClearProcessedHeader(height uint64)
-	ClearProcessedData(height uint64)
 	SaveToDisk() error
 	LoadFromDisk() error
 }
@@ -122,14 +115,6 @@ func NewManager(cfg config.Config, store store.Store, logger zerolog.Logger) (Ma
 }
 
 // Header operations
-func (m *implementation) GetHeader(height uint64) *types.SignedHeader {
-	return m.headerCache.getItem(height)
-}
-
-func (m *implementation) SetHeader(height uint64, header *types.SignedHeader) {
-	m.headerCache.setItem(height, header)
-}
-
 func (m *implementation) IsHeaderSeen(hash string) bool {
 	return m.headerCache.isSeen(hash)
 }
@@ -147,14 +132,6 @@ func (m *implementation) SetHeaderDAIncluded(hash string, daHeight uint64) {
 }
 
 // Data operations
-func (m *implementation) GetData(height uint64) *types.Data {
-	return m.dataCache.getItem(height)
-}
-
-func (m *implementation) SetData(height uint64, data *types.Data) {
-	m.dataCache.setItem(height, data)
-}
-
 func (m *implementation) IsDataSeen(hash string) bool {
 	return m.dataCache.isSeen(hash)
 }
@@ -236,15 +213,6 @@ func (m *implementation) DeletePendingEvent(height uint64) {
 	m.pendingEventsCache.deleteItem(height)
 }
 
-// Cleanup operations
-func (m *implementation) ClearProcessedHeader(height uint64) {
-	m.headerCache.deleteItem(height)
-}
-
-func (m *implementation) ClearProcessedData(height uint64) {
-	m.dataCache.deleteItem(height)
-}
-
 func (m *implementation) SaveToDisk() error {
 	cfgDir := filepath.Join(m.config.RootDir, "data")
 

block/internal/cache/manager_test.go

@@ -38,16 +38,6 @@ func TestManager_HeaderDataOperations(t *testing.T) {
 	m, err := NewManager(cfg, st, zerolog.Nop())
 	require.NoError(t, err)
 
-	// construct simple header and data
-	header := &types.SignedHeader{Header: types.Header{BaseHeader: types.BaseHeader{ChainID: "test", Height: 1}}}
-	data := &types.Data{Metadata: &types.Metadata{ChainID: "test", Height: 1}}
-
-	// set and get by height
-	m.SetHeader(1, header)
-	m.SetData(1, data)
-	assert.Equal(t, header, m.GetHeader(1))
-	assert.Equal(t, data, m.GetData(1))
-
 	// seen & DA included flags
 	m.SetHeaderSeen("h1")
 	m.SetDataSeen("d1")
@@ -60,12 +50,6 @@ func TestManager_HeaderDataOperations(t *testing.T) {
 	assert.True(t, ok)
 	_, ok = m.GetDataDAIncluded("d1")
 	assert.True(t, ok)
-
-	// cleanup of processed
-	m.ClearProcessedHeader(1)
-	m.ClearProcessedData(1)
-	assert.Nil(t, m.GetHeader(1))
-	assert.Nil(t, m.GetData(1))
 }
 
 func TestManager_PendingEventsCRUD(t *testing.T) {
@@ -114,8 +98,6 @@ func TestManager_SaveAndLoadFromDisk(t *testing.T) {
 	// populate caches
 	hdr := &types.SignedHeader{Header: types.Header{BaseHeader: types.BaseHeader{ChainID: "c", Height: 2}}}
 	dat := &types.Data{Metadata: &types.Metadata{ChainID: "c", Height: 2}}
-	m1.SetHeader(2, hdr)
-	m1.SetData(2, dat)
 	m1.SetHeaderSeen("H2")
 	m1.SetDataSeen("D2")
 	m1.SetHeaderDAIncluded("H2", 100)
@@ -131,8 +113,6 @@ func TestManager_SaveAndLoadFromDisk(t *testing.T) {
 	require.NoError(t, err)
 
 	// check loaded items
-	assert.NotNil(t, m2.GetHeader(2))
-	assert.NotNil(t, m2.GetData(2))
 	assert.True(t, m2.IsHeaderSeen("H2"))
 	assert.True(t, m2.IsDataSeen("D2"))
 	_, ok := m2.GetHeaderDAIncluded("H2")

block/internal/executing/executor.go

@@ -353,6 +353,8 @@ func (e *Executor) produceBlock() error {
 		return fmt.Errorf("failed to apply block: %w", err)
 	}
 
+	// signing the header is done after applying the block
+	// as for signing, the state of the block may be required by the signature payload provider.
 	signature, err := e.signHeader(header.Header)
 	if err != nil {
 		return fmt.Errorf("failed to sign header: %w", err)

block/internal/syncing/da_retriever.go

@@ -76,7 +76,6 @@ func (r *DARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]co
 			r.logger.Debug().Int("blobs", len(blobsResp.Data)).Uint64("da_height", daHeight).Msg("retrieved blob data")
 			events := r.processBlobs(ctx, blobsResp.Data, daHeight)
 			return events, nil
-
 		}
 
 		if strings.Contains(fetchErr.Error(), coreda.ErrHeightFromFuture.Error()) {
@@ -247,15 +246,9 @@ func (r *DARetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 		return nil
 	}
 
-	// Mark as DA included
-	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("header marked as DA included")
+	// note, we cannot mark the header as DA included
+	// we haven't done any signature verification check here
+	// signature verification happens with data.
 
 	return header
 }
@@ -293,7 +286,6 @@ func (r *DARetriever) tryDecodeData(bz []byte, daHeight uint64) *types.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)
 }
 

block/internal/syncing/da_retriever_test.go

@@ -118,7 +118,7 @@ func TestDARetriever_ProcessBlobs_HeaderAndData_Success(t *testing.T) {
 	_, lastState := types.State{}, types.State{}
 	_ = lastState // placeholder to keep parity with helper pattern
 
-	hdrBin, hdr := makeSignedHeaderBytes(t, gen.ChainID, 2, addr, pub, signer, nil)
+	hdrBin, _ := makeSignedHeaderBytes(t, gen.ChainID, 2, addr, pub, signer, nil)
 	dataBin, _ := makeSignedDataBytes(t, gen.ChainID, 2, addr, pub, signer, 2)
 
 	events := r.processBlobs(context.Background(), [][]byte{hdrBin, dataBin}, 77)
@@ -127,11 +127,6 @@ func TestDARetriever_ProcessBlobs_HeaderAndData_Success(t *testing.T) {
 	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)
-
-	// DA included markers should be set by tryDecodeHeader/tryDecodeData
-	if hHeight, ok := cm.GetHeaderDAIncluded(hdr.Hash().String()); assert.True(t, ok) {
-		assert.Equal(t, uint64(77), hHeight)
-	}
 }
 
 func TestDARetriever_ProcessBlobs_HeaderOnly_EmptyDataExpected(t *testing.T) {

block/internal/syncing/p2p_handler.go

@@ -60,8 +60,8 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 			continue
 		}
 
-		// Validate header
-		if err := h.validateHeader(header); err != nil {
+		// basic header validation
+		if err := h.assertExpectedProposer(header.ProposerAddress); err != nil {
 			h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P")
 			continue
 		}
@@ -81,17 +81,15 @@ func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHei
 			data = retrievedData
 		}
 
-		// Validate header with data
-		if err := header.ValidateBasicWithData(data); err != nil {
-			h.logger.Debug().Uint64("height", height).Err(err).Msg("header validation with data failed")
-			continue
-		}
+		// 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
 
 		// Create height event
 		event := common.DAHeightEvent{
-			Header:   header,
-			Data:     data,
-			DaHeight: 0, // P2P events don't have DA height context
+			Header:                 header,
+			Data:                   data,
+			DaHeight:               0, // P2P events don't have DA height context
+			HeaderDaIncludedHeight: 0, // P2P events don't have DA height context
 		}
 
 		events = append(events, event)
@@ -130,23 +128,21 @@ func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeigh
 			continue
 		}
 
-		// Validate header
-		if err := h.validateHeader(header); err != nil {
+		// basic header validation
+		if err := h.assertExpectedProposer(header.ProposerAddress); err != nil {
 			h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P")
 			continue
 		}
 
-		// Validate header with data
-		if err := header.ValidateBasicWithData(data); err != nil {
-			h.logger.Debug().Uint64("height", height).Err(err).Msg("header validation with data failed")
-			continue
-		}
+		// 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
 
 		// Create height event
 		event := common.DAHeightEvent{
-			Header:   header,
-			Data:     data,
-			DaHeight: 0, // P2P events don't have DA height context
+			Header:                 header,
+			Data:                   data,
+			DaHeight:               0, // P2P events don't have DA height context
+			HeaderDaIncludedHeight: 0, // P2P events don't have DA height context
 		}
 
 		events = append(events, event)
@@ -157,16 +153,6 @@ func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeigh
 	return events
 }
 
-// validateHeader performs basic validation on a header from P2P
-func (h *P2PHandler) validateHeader(header *types.SignedHeader) error {
-	// Check proposer address
-	if err := h.assertExpectedProposer(header.ProposerAddress); err != nil {
-		return fmt.Errorf("unexpected proposer: %w", err)
-	}
-
-	return nil
-}
-
 // assertExpectedProposer validates the proposer address
 func (h *P2PHandler) assertExpectedProposer(proposerAddr []byte) error {
 	if !bytes.Equal(h.genesis.ProposerAddress, proposerAddr) {

block/internal/syncing/p2p_handler_test.go

@@ -272,20 +272,3 @@ func TestP2PHandler_ProcessDataRange_HeaderValidateHeaderFails(t *testing.T) {
 	events := p2pData.Handler.ProcessDataRange(ctx, 3, 3)
 	require.Len(t, events, 0, "validateHeader failure should drop event")
 }
-
-func TestP2PHandler_ProcessDataRange_ValidateBasicWithDataFails(t *testing.T) {
-	p2pData := setupP2P(t)
-	ctx := context.Background()
-
-	// Data exists at height 4
-	blockData := p2pMakeData(t, p2pData.Genesis.ChainID, 4, 1)
-	p2pData.DataStore.EXPECT().GetByHeight(ctx, uint64(4)).Return(blockData, nil).Once()
-
-	// Header proposer matches genesis, but signature is empty -> ValidateBasicWithData should fail
-	header := p2pMakeSignedHeader(t, p2pData.Genesis.ChainID, 4, p2pData.ProposerAddr, p2pData.ProposerPub, p2pData.Signer)
-	header.Signature = nil // force signature validation failure
-	p2pData.HeaderStore.EXPECT().GetByHeight(ctx, uint64(4)).Return(header, nil).Once()
-
-	events := p2pData.Handler.ProcessDataRange(ctx, 4, 4)
-	require.Len(t, events, 0, "ValidateBasicWithData failure should drop event")
-}