+ add logs to batches

synchronizer/l1_rollup_info_consumer.go

@@ -44,6 +44,7 @@ type l1RollupInfoConsumer struct {
 	ctx                   context.Context
 	statistics            l1RollupInfoConsumerStatistics
 	lastEthBlockSynced    *state.Block
+	highestBlockProcessed uint64
 }
 
 func newL1RollupInfoConsumer(cfg configConsumer,
@@ -62,22 +63,37 @@ func newL1RollupInfoConsumer(cfg configConsumer,
 			startTime: time.Now(),
 			cfg:       cfg,
 		},
+		highestBlockProcessed: invalidBlockNumber,
 	}
 }
 
 func (l *l1RollupInfoConsumer) Start(ctx context.Context, lastEthBlockSynced *state.Block) error {
 	l.ctx = ctx
 	l.lastEthBlockSynced = lastEthBlockSynced
+	if l.highestBlockProcessed == invalidBlockNumber && lastEthBlockSynced != nil {
+		log.Infof("consumer: Starting consumer. setting HighestBlockProcessed: %d (lastEthBlockSynced)", lastEthBlockSynced.BlockNumber)
+		l.highestBlockProcessed = lastEthBlockSynced.BlockNumber
+	}
+	log.Infof("consumer: Starting consumer. HighestBlockProcessed: %d", l.highestBlockProcessed)
 	l.statistics.onStart()
 	err := l.step()
 	for ; err == nil; err = l.step() {
 	}
 	if err != errConsumerStopped && err != errConsumerStoppedBecauseIsSynchronized {
 		return err
 	}
-	// The errConsumerStopped is not an error, so we return nil meaning that the process finished in a normal way
+	// The errConsumerStopped||errConsumerStoppedBecauseIsSynchronized are not an error, so we return nil meaning that the process finished in a normal way
 	return nil
 }
+
+func (l *l1RollupInfoConsumer) Reset(startingBlockNumber uint64) {
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+	l.highestBlockProcessed = startingBlockNumber
+	l.lastEthBlockSynced = nil
+	l.statistics.onReset()
+}
+
 func (l *l1RollupInfoConsumer) step() error {
 	l.statistics.onStartStep()
 	var err error
@@ -125,10 +141,12 @@ func (l *l1RollupInfoConsumer) processIncommingRollupInfoData(rollupInfo rollupI
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
 	var err error
-	if (l.lastEthBlockSynced != nil) && (l.lastEthBlockSynced.BlockNumber+1 != rollupInfo.blockRange.fromBlock) {
-		log.Infof("consumer: received a rollupInfo with a wrong block range.  Ignoring it. Last block synced: %d. RollupInfo block range: %s",
-			l.lastEthBlockSynced.BlockNumber, rollupInfo.blockRange.String())
+	if (l.highestBlockProcessed != invalidBlockNumber) && (l.highestBlockProcessed+1 != rollupInfo.blockRange.fromBlock) {
+		log.Warnf("consumer: received a rollupInfo with a wrong block range.  Ignoring it. Highest block synced: %d. RollupInfo block range: %s",
+			l.highestBlockProcessed, rollupInfo.blockRange.String())
+		return nil
 	}
+	l.highestBlockProcessed = rollupInfo.blockRange.toBlock
 	// Uncommented that line to produce a infinite loop of errors, and resets! (just for develop)
 	//return errors.New("forcing an continuous error!")
 	statisticsMsg := l.statistics.onStartProcessIncommingRollupInfoData(rollupInfo)
@@ -137,7 +155,7 @@ func (l *l1RollupInfoConsumer) processIncommingRollupInfoData(rollupInfo rollupI
 	l.lastEthBlockSynced, err = l.processUnsafe(rollupInfo)
 	l.statistics.onFinishProcessIncommingRollupInfoData(rollupInfo, time.Since(timeProcessingStart), err)
 	if err != nil {
-		log.Info("consumer: error processing rollupInfo. Error: ", err)
+		log.Infof("consumer: error processing rollupInfo %s. Error: %s", rollupInfo.blockRange.String(), err.Error())
 		return err
 	}
 	l.statistics.numProcessedBlocks += uint64(len(rollupInfo.blocks))
@@ -168,31 +186,31 @@ func (l *l1RollupInfoConsumer) processUnsafe(rollupInfo rollupInfoByBlockRangeRe
 	blocks := rollupInfo.blocks
 	order := rollupInfo.order
 	var lastEthBlockSynced *state.Block
-	err := l.synchronizer.processBlockRange(blocks, order)
-	if err != nil {
-		log.Info("consumer: Error processing block range: ", rollupInfo.blockRange, " err:", err)
-		return nil, err
-	}
-	if len(blocks) > 0 {
-		tmpStateBlock := convertEthmanBlockToStateBlock(&blocks[len(blocks)-1])
-		lastEthBlockSynced = &tmpStateBlock
-		logBlocks(blocks)
-	}
+
 	if len(blocks) == 0 {
 		lb := rollupInfo.lastBlockOfRange
 		if lb == nil {
 			log.Warn("consumer: Error processing block range: ", rollupInfo.blockRange, " err: need the last block of range and got a nil")
 			return nil, errMissingLastBlock
 		}
 		b := convertL1BlockToEthBlock(lb)
-		err = l.synchronizer.processBlockRange([]etherman.Block{b}, order)
+		err := l.synchronizer.processBlockRange([]etherman.Block{b}, order)
 		if err != nil {
 			log.Error("consumer: Error processing last block of range: ", rollupInfo.blockRange, " err:", err)
 			return nil, err
 		}
 		block := convertL1BlockToStateBlock(lb)
 		lastEthBlockSynced = &block
 		log.Debug("consumer: Storing empty block. BlockNumber: ", b.BlockNumber, ". BlockHash: ", b.BlockHash)
+	} else {
+		tmpStateBlock := convertEthmanBlockToStateBlock(&blocks[len(blocks)-1])
+		lastEthBlockSynced = &tmpStateBlock
+		logBlocks(blocks)
+		err := l.synchronizer.processBlockRange(blocks, order)
+		if err != nil {
+			log.Info("consumer: Error processing block range: ", rollupInfo.blockRange, " err:", err)
+			return nil, err
+		}
 	}
 	return lastEthBlockSynced, nil
 }

synchronizer/l1_rollup_info_consumer_statistics.go

@@ -9,35 +9,54 @@ import (
 )
 
 type l1RollupInfoConsumerStatistics struct {
-	numProcessedRollupInfo         uint64
-	numProcessedBlocks             uint64
-	startTime                      time.Time
-	timePreviousProcessingDuration time.Duration
-	startStepTime                  time.Time
-	cfg                            configConsumer
+	numProcessedRollupInfo             uint64
+	numProcessedRollupInfoForCheckTime uint64
+	numProcessedBlocks                 uint64
+	startTime                          time.Time
+	timePreviousProcessingDuration     time.Duration
+	startStepTime                      time.Time
+	cfg                                configConsumer
 }
 
 func (l *l1RollupInfoConsumerStatistics) onStart() {
 	l.startTime = time.Now()
 	l.startStepTime = time.Time{}
+	l.numProcessedRollupInfoForCheckTime = 0
 }
 
 func (l *l1RollupInfoConsumerStatistics) onStartStep() {
 	l.startStepTime = time.Now()
 }
 
+func (l *l1RollupInfoConsumerStatistics) onReset() {
+	l.numProcessedRollupInfoForCheckTime = 0
+	l.startStepTime = time.Time{}
+}
+
 func (l *l1RollupInfoConsumerStatistics) onStartProcessIncommingRollupInfoData(rollupInfo rollupInfoByBlockRangeResult) string {
 	now := time.Now()
 	// Time have have been blocked in the select statement
 	waitingTimeForData := now.Sub(l.startStepTime)
 	blocksPerSecond := float64(l.numProcessedBlocks) / time.Since(l.startTime).Seconds()
-	if l.numProcessedRollupInfo > uint64(l.cfg.numIterationsBeforeStartCheckingTimeWaitingForNewRollupInfoData) && waitingTimeForData > l.cfg.acceptableTimeWaitingForNewRollupInfoData {
+	generatedWarning := false
+	if l.numProcessedRollupInfoForCheckTime > uint64(l.cfg.numIterationsBeforeStartCheckingTimeWaitingForNewRollupInfoData) && waitingTimeForData > l.cfg.acceptableTimeWaitingForNewRollupInfoData {
 		msg := fmt.Sprintf("wasted waiting for new rollupInfo from L1: %s last_process: %s new range: %s block_per_second: %f",
 			waitingTimeForData, l.timePreviousProcessingDuration, rollupInfo.blockRange.String(), blocksPerSecond)
 		log.Warnf("consumer:: Too much wasted time (waiting to receive a new data):%s", msg)
+		generatedWarning = true
 	}
 	l.numProcessedRollupInfo++
-	msg := fmt.Sprintf("wasted_time_waiting_for_data [%s] last_process_time [%s] block_per_second [%f]", waitingTimeForData.Round(time.Second).String(), l.timePreviousProcessingDuration, blocksPerSecond)
+	l.numProcessedRollupInfoForCheckTime++
+	msg := fmt.Sprintf("wasted_time_waiting_for_data [%s] last_process_time [%s] block_per_second [%f]",
+		waitingTimeForData.Round(time.Second).String(),
+		l.timePreviousProcessingDuration,
+		blocksPerSecond)
+	if waitingTimeForData > l.cfg.acceptableTimeWaitingForNewRollupInfoData {
+		msg = msg + " WASTED_TIME_EXCEED "
+	}
+	if generatedWarning {
+		msg = msg + " WARNING_WASTED_TIME "
+	}
 	return msg
 }
 

synchronizer/l1_rollup_info_consumer_statistics_test.go

@@ -0,0 +1,117 @@
+package synchronizer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/0xPolygonHermez/zkevm-node/etherman"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestL1RollupInfoConsumerStatistics(t *testing.T) {
+	cfg := configConsumer{
+		numIterationsBeforeStartCheckingTimeWaitingForNewRollupInfoData: 10,
+		acceptableTimeWaitingForNewRollupInfoData:                       5 * time.Second,
+	}
+	stats := l1RollupInfoConsumerStatistics{
+		cfg: cfg,
+	}
+
+	stats.onStart()
+	stats.onStartStep()
+
+	// Test onFinishProcessIncommingRollupInfoData
+	rollupInfo := rollupInfoByBlockRangeResult{
+		blockRange: blockRange{
+			fromBlock: 1,
+			toBlock:   10,
+		},
+		blocks: []etherman.Block{},
+	}
+	executionTime := 2 * time.Second
+	stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	stats.onFinishProcessIncommingRollupInfoData(rollupInfo, executionTime, error(nil))
+	assert.Equal(t, stats.timePreviousProcessingDuration, executionTime)
+	assert.Equal(t, stats.numProcessedRollupInfo, uint64(1))
+	assert.Equal(t, stats.numProcessedBlocks, uint64(len(rollupInfo.blocks)))
+
+	stats.onStart()
+	stats.onStartStep()
+
+	msg := stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	assert.Contains(t, msg, "wasted_time_waiting_for_data")
+	assert.Contains(t, msg, "last_process_time")
+	assert.Contains(t, msg, "block_per_second")
+	assert.NotContains(t, msg, "WASTED_TIME_EXCEED")
+	assert.NotContains(t, msg, "WARNING_WASTED_TIME")
+}
+
+func TestL1RollupInfoConsumerStatisticsWithExceedTimeButNoWarningGenerated(t *testing.T) {
+	cfg := configConsumer{
+		numIterationsBeforeStartCheckingTimeWaitingForNewRollupInfoData: 10,
+		acceptableTimeWaitingForNewRollupInfoData:                       0 * time.Second,
+	}
+	stats := l1RollupInfoConsumerStatistics{
+		cfg: cfg,
+	}
+
+	stats.onStart()
+	stats.onStartStep()
+
+	// Test onFinishProcessIncommingRollupInfoData
+	rollupInfo := rollupInfoByBlockRangeResult{
+		blockRange: blockRange{
+			fromBlock: 1,
+			toBlock:   10,
+		},
+		blocks: []etherman.Block{},
+	}
+	executionTime := 2 * time.Second
+	err := error(nil)
+	stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	stats.onFinishProcessIncommingRollupInfoData(rollupInfo, executionTime, err)
+
+	stats.onStartStep()
+	msg := stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	assert.Contains(t, msg, "wasted_time_waiting_for_data")
+	assert.Contains(t, msg, "last_process_time")
+	assert.Contains(t, msg, "block_per_second")
+	assert.Contains(t, msg, "WASTED_TIME_EXCEED")
+	assert.NotContains(t, msg, "WARNING_WASTED_TIME")
+}
+
+func TestL1RollupInfoConsumerStatisticsWithExceedTimeButAndWarningGenerated(t *testing.T) {
+	cfg := configConsumer{
+		numIterationsBeforeStartCheckingTimeWaitingForNewRollupInfoData: 1,
+		acceptableTimeWaitingForNewRollupInfoData:                       0 * time.Second,
+	}
+	stats := l1RollupInfoConsumerStatistics{
+		cfg: cfg,
+	}
+
+	stats.onStart()
+	stats.onStartStep()
+
+	// Test onFinishProcessIncommingRollupInfoData
+	rollupInfo := rollupInfoByBlockRangeResult{
+		blockRange: blockRange{
+			fromBlock: 1,
+			toBlock:   10,
+		},
+		blocks: []etherman.Block{},
+	}
+	executionTime := 2 * time.Second
+	err := error(nil)
+	stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	stats.onFinishProcessIncommingRollupInfoData(rollupInfo, executionTime, err)
+	stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	stats.onFinishProcessIncommingRollupInfoData(rollupInfo, executionTime, err)
+
+	stats.onStartStep()
+	msg := stats.onStartProcessIncommingRollupInfoData(rollupInfo)
+	assert.Contains(t, msg, "wasted_time_waiting_for_data")
+	assert.Contains(t, msg, "last_process_time")
+	assert.Contains(t, msg, "block_per_second")
+	assert.Contains(t, msg, "WASTED_TIME_EXCEED")
+	assert.Contains(t, msg, "WARNING_WASTED_TIME")
+}