fix: preserve DA data until block acceptance

Author: randygrok

block/internal/syncing/da_retriever.go

@@ -28,6 +28,10 @@ type DARetriever interface {
 	ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent
 }
 
+type pendingDataCleaner interface {
+	removePendingData(height uint64)
+}
+
 // daRetriever handles DA retrieval operations for syncing
 type daRetriever struct {
 	client  da.Client
@@ -213,7 +217,6 @@ func (r *daRetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 			}
 		} else {
 			delete(r.pendingHeaders, height)
-			delete(r.pendingData, height)
 		}
 
 		// Create height event
@@ -245,6 +248,13 @@ func (r *daRetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 	return events
 }
 
+func (r *daRetriever) removePendingData(height uint64) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	delete(r.pendingData, height)
+}
+
 // tryDecodeHeader attempts to decode a blob as a header
 func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedHeader {
 	header := new(types.SignedHeader)

block/internal/syncing/da_retriever_test.go

@@ -307,9 +307,35 @@ func TestDARetriever_ProcessBlobs_CrossDAHeightMatching(t *testing.T) {
 	assert.Equal(t, uint64(5), event.Data.Height())
 	assert.Equal(t, uint64(102), event.DaHeight, "DaHeight should be the height where data was processed")
 
-	// Verify pending maps are cleared
+	// Verify the header is consumed, while data remains available until the
+	// candidate block is accepted by the syncer.
 	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.Contains(t, r.pendingData, uint64(5), "data should remain pending until accepted")
+
+	r.removePendingData(5)
+	require.NotContains(t, r.pendingData, uint64(5), "accepted data should be removed from pending")
+}
+
+func TestDARetriever_ProcessBlobs_KeepsDataForLaterHeaderAfterCandidateEvent(t *testing.T) {
+	expectedAddr, expectedPub, expectedSigner := buildSyncTestSigner(t)
+	wrongAddr, wrongPub, wrongSigner := buildSyncTestSigner(t)
+	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: expectedAddr}
+
+	r := newTestDARetriever(t, nil, config.DefaultConfig(), gen)
+
+	dataBin, data := makeSignedDataBytes(t, gen.ChainID, 5, expectedAddr, expectedPub, expectedSigner, 2)
+	wrongHeaderBin, wrongHeader := makeSignedHeaderBytes(t, gen.ChainID, 5, wrongAddr, wrongPub, wrongSigner, nil, &data.Data, nil)
+	correctHeaderBin, correctHeader := makeSignedHeaderBytes(t, gen.ChainID, 5, expectedAddr, expectedPub, expectedSigner, nil, &data.Data, nil)
+
+	events := r.processBlobs(context.Background(), [][]byte{wrongHeaderBin, dataBin}, 100)
+	require.Len(t, events, 1)
+	require.Equal(t, wrongHeader.Hash().String(), events[0].Header.Hash().String())
+	require.Contains(t, r.pendingData, uint64(5), "data should stay available until the candidate block is accepted")
+
+	events = r.processBlobs(context.Background(), [][]byte{correctHeaderBin}, 101)
+	require.Len(t, events, 1)
+	require.Equal(t, correctHeader.Hash().String(), events[0].Header.Hash().String())
+	require.Equal(t, data.Data.DACommitment().String(), events[0].Data.DACommitment().String())
 }
 
 func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testing.T) {
@@ -355,6 +381,8 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	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)
+	r.removePendingData(3)
+	r.removePendingData(5)
 
 	// Verify header 4 is still pending (no matching data yet)
 	require.Contains(t, r.pendingHeaders, uint64(4), "header 4 should still be pending")
@@ -369,6 +397,7 @@ 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)
+	r.removePendingData(4)
 
 	// Verify all pending maps are now clear
 	require.NotContains(t, r.pendingHeaders, uint64(4), "header 4 should be removed from pending")

block/internal/syncing/da_retriever_tracing.go

@@ -59,3 +59,9 @@ func (t *tracedDARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64)
 func (t *tracedDARetriever) ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
 	return t.inner.ProcessBlobs(ctx, blobs, daHeight)
 }
+
+func (t *tracedDARetriever) removePendingData(height uint64) {
+	if cleaner, ok := t.inner.(pendingDataCleaner); ok {
+		cleaner.removePendingData(height)
+	}
+}

block/internal/syncing/syncer.go

@@ -827,6 +827,12 @@ func (s *Syncer) trySyncNextBlockWithState(ctx context.Context, event *common.DA
 		s.p2pHandler.SetProcessedHeight(newState.LastBlockHeight)
 	}
 
+	if event.Source == common.SourceDA {
+		if cleaner, ok := s.daRetriever.(pendingDataCleaner); ok {
+			cleaner.removePendingData(nextHeight)
+		}
+	}
+
 	return nil
 }
 

block/internal/syncing/syncer_test.go

@@ -425,15 +425,18 @@ func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 	lastState := s.getLastState()
 	data := makeData(gen.ChainID, 1, 0)
 	_, hdr := makeSignedHeaderBytes(t, gen.ChainID, 1, addr, pub, signer, lastState.AppHash, data, nil)
+	daRetriever := &daRetriever{pendingData: map[uint64]*types.Data{1: data}}
+	s.daRetriever = daRetriever
 
 	// Expect ExecuteTxs call for height 1
 	mockExec.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(1), mock.Anything, lastState.AppHash).
 		Return([]byte("app1"), nil).Once()
 
-	evt := common.DAHeightEvent{Header: hdr, Data: data, DaHeight: 1}
+	evt := common.DAHeightEvent{Header: hdr, Data: data, Source: common.SourceDA, DaHeight: 1}
 	s.processHeightEvent(t.Context(), &evt)
 
 	requireEmptyChan(t, errChan)
+	require.NotContains(t, daRetriever.pendingData, uint64(1), "accepted DA data should be removed from the retriever pending data")
 	h, err := st.Height(t.Context())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(1), h)