fix(syncing): clean up already processed DA data (#3413)

fix: clean up processed DA data

Author: jgimeno

block/internal/syncing/syncer.go

@@ -589,6 +589,7 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 
 	// Skip if already processed
 	if height <= currentHeight || s.cache.IsHeaderSeen(headerHash) {
+		s.removePendingDAData(event)
 		s.logger.Debug().
 			Uint64("height", height).
 			Str("source", string(event.Source)).
@@ -844,15 +845,21 @@ 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)
-		}
-	}
+	s.removePendingDAData(event)
 
 	return nil
 }
 
+func (s *Syncer) removePendingDAData(event *common.DAHeightEvent) {
+	if event.Source != common.SourceDA {
+		return
+	}
+
+	if cleaner, ok := s.daRetriever.(pendingDataCleaner); ok {
+		cleaner.removePendingData(event.Header.Height())
+	}
+}
+
 // ApplyBlock applies a block to get the new state
 func (s *Syncer) ApplyBlock(ctx context.Context, header types.Header, data *types.Data, currentState types.State) (types.State, error) {
 	// Prepare transactions

block/internal/syncing/syncer_test.go

@@ -622,6 +622,51 @@ func TestProcessHeightEvent_SyncsAndUpdatesState(t *testing.T) {
 	assert.Equal(t, uint64(1), st1.LastBlockHeight)
 }
 
+func TestProcessHeightEvent_AlreadyProcessedDADataDoesNotAccumulate(t *testing.T) {
+	const alreadyProcessedBlocks = 8
+
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+
+	batch, err := st.NewBatch(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, batch.SetHeight(alreadyProcessedBlocks))
+	require.NoError(t, batch.Commit())
+
+	addr, pub, signer := buildSyncTestSigner(t)
+	gen := genesis.Genesis{
+		ChainID:         "tchain",
+		InitialHeight:   1,
+		StartTime:       time.Now().Add(-time.Second),
+		ProposerAddress: addr,
+	}
+	daRetriever := newTestDARetriever(t, nil, config.DefaultConfig(), gen)
+	s := &Syncer{
+		store:       st,
+		daRetriever: daRetriever,
+		logger:      zerolog.Nop(),
+	}
+
+	futureHeight := uint64(alreadyProcessedBlocks + 1)
+	futureDataBin, _ := makeSignedDataBytes(t, gen.ChainID, futureHeight, addr, pub, signer, 8)
+	require.Empty(t, daRetriever.processBlobs(t.Context(), [][]byte{futureDataBin}, futureHeight))
+	require.Contains(t, daRetriever.pendingData, futureHeight)
+
+	for height := uint64(1); height <= alreadyProcessedBlocks; height++ {
+		dataBin, data := makeSignedDataBytes(t, gen.ChainID, height, addr, pub, signer, 8)
+		headerBin, _ := makeSignedHeaderBytes(t, gen.ChainID, height, addr, pub, signer, nil, &data.Data, nil)
+
+		events := daRetriever.processBlobs(t.Context(), [][]byte{headerBin, dataBin}, height)
+		require.Len(t, events, 1)
+
+		s.processHeightEvent(t.Context(), &events[0])
+		require.NotContains(t, daRetriever.pendingData, height, "processed DA data must be removed immediately")
+	}
+
+	require.Len(t, daRetriever.pendingData, 1)
+	require.Contains(t, daRetriever.pendingData, futureHeight, "DA data above the current height must remain pending")
+}
+
 func TestProcessHeightEvent_UnexpectedProposerFromDAIsNotCriticalStateError(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)