feat: currentStreamBatchRaw sanity check (#131) (#132)

0xPolygon · Oct 17, 2024 · 08888a1 · 08888a1
1 parent 99dc357
commit 08888a1
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/aggregator/aggregator.go b/aggregator/aggregator.go
@@ -425,6 +425,12 @@ func (a *Aggregator) handleReceivedDataStream(
 
 			switch entry.Type {
 			case datastreamer.EntryType(datastream.EntryType_ENTRY_TYPE_BATCH_START):
+				// Check currentStreamBatchRaw is empty as sanity check
+				if len(a.currentStreamBatchRaw.Blocks) > 0 {
+					a.logger.Errorf("currentStreamBatchRaw should be empty, "+
+						"but it contains %v blocks", len(a.currentStreamBatchRaw.Blocks))
+					a.resetCurrentBatchData()
+				}
 				batch := &datastream.BatchStart{}
 				err := proto.Unmarshal(entry.Data, batch)
 				if err != nil {