fix: blockhash comparison in stage batches processor

zk/datastream/client/stream_client.go

@@ -101,7 +101,6 @@ func (c *StreamClient) GetL2BlockByNumber(blockNum uint64) (*types.FullL2Block,
 	if _, err := c.EnsureConnected(); err != nil {
 		return nil, -1, err
 	}
-	defer c.Stop()
 
 	var (
 		l2Block   *types.FullL2Block
@@ -160,7 +159,6 @@ func (c *StreamClient) GetLatestL2Block() (l2Block *types.FullL2Block, err error
 	if _, err := c.EnsureConnected(); err != nil {
 		return nil, err
 	}
-	defer c.Stop()
 
 	h, err := c.GetHeader()
 	if err != nil {

zk/stages/stage_batches.go

@@ -219,7 +219,12 @@ func SpawnStageBatches(
 		return rollback(logPrefix, eriDb, hermezDb, dsQueryClient, unwindBlock, tx, u)
 	}
 
-	batchProcessor, err := NewBatchesProcessor(ctx, logPrefix, tx, hermezDb, eriDb, cfg.zkCfg.SyncLimit, cfg.zkCfg.DebugLimit, cfg.zkCfg.DebugStepAfter, cfg.zkCfg.DebugStep, stageProgressBlockNo, stageProgressBatchNo, dsQueryClient, progressChan, cfg.chainConfig, cfg.miningConfig, unwindFn)
+	lastProcessedBlockHash, err := eriDb.ReadCanonicalHash(stageProgressBlockNo)
+	if err != nil {
+		return fmt.Errorf("failed to read canonical hash for block %d: %w", stageProgressBlockNo, err)
+	}
+
+	batchProcessor, err := NewBatchesProcessor(ctx, logPrefix, tx, hermezDb, eriDb, cfg.zkCfg.SyncLimit, cfg.zkCfg.DebugLimit, cfg.zkCfg.DebugStepAfter, cfg.zkCfg.DebugStep, stageProgressBlockNo, stageProgressBatchNo, lastProcessedBlockHash, dsQueryClient, progressChan, cfg.chainConfig, cfg.miningConfig, unwindFn)
 	if err != nil {
 		return err
 	}

zk/stages/stage_batches_datastream.go

@@ -74,7 +74,6 @@ func (r *DatastreamClientRunner) StopRead() {
 	for r.dsClient.GetStreamingAtomic().Load() {
 		time.Sleep(10 * time.Microsecond)
 	}
-	r.dsClient.Stop()
 }
 
 func (r *DatastreamClientRunner) RestartReadFromBlock(fromBlock uint64) error {

zk/stages/stage_batches_processor.go

@@ -96,6 +96,7 @@ func NewBatchesProcessor(
 	hermezDb ProcessorHermezDb,
 	eriDb ProcessorErigonDb,
 	syncBlockLimit, debugBlockLimit, debugStepAfter, debugStep, stageProgressBlockNo, stageProgressBatchNo uint64,
+	lastProcessedBlockHash common.Hash,
 	dsQueryClient DsQueryClient,
 	progressChan chan uint64,
 	chainConfig *chain.Config,
@@ -128,7 +129,7 @@ func NewBatchesProcessor(
 		highestVerifiedBatch: highestVerifiedBatch,
 		dsQueryClient:        dsQueryClient,
 		progressChan:         progressChan,
-		lastBlockHash:        emptyHash,
+		lastBlockHash:        lastProcessedBlockHash,
 		lastBlockRoot:        emptyHash,
 		lastForkId:           lastForkId,
 		unwindFn:             unwindFn,
@@ -270,28 +271,12 @@ func (p *BatchesProcessor) processFullBlock(blockEntry *types.FullL2Block) (endL
 		}
 	}
 
-	dsParentBlockHash := p.lastBlockHash
-	dsBlockNumber := p.lastBlockHeight
-	if dsParentBlockHash == emptyHash {
-		parentBlockDS, _, err := p.dsQueryClient.GetL2BlockByNumber(blockEntry.L2BlockNumber - 1)
-		if err != nil {
-			return false, err
-		}
-
-		if parentBlockDS != nil {
-			dsParentBlockHash = parentBlockDS.L2Blockhash
-			if parentBlockDS.L2BlockNumber > 0 {
-				dsBlockNumber = parentBlockDS.L2BlockNumber
-			}
-		}
-	}
-
-	if blockEntry.L2BlockNumber > 1 && dbParentBlockHash != dsParentBlockHash {
+	if p.lastBlockHeight > 0 && dbParentBlockHash != p.lastBlockHash {
 		// unwind/rollback blocks until the latest common ancestor block
 		log.Warn(fmt.Sprintf("[%s] Parent block hashes mismatch on block %d. Triggering unwind...", p.logPrefix, blockEntry.L2BlockNumber),
 			"db parent block hash", dbParentBlockHash,
-			"ds parent block number", dsBlockNumber,
-			"ds parent block hash", dsParentBlockHash,
+			"ds parent block number", p.lastBlockHeight,
+			"ds parent block hash", p.lastBlockHash,
 			"ds parent block number", blockEntry.L2BlockNumber-1,
 		)
 		//parent blockhash is wrong, so unwind to it, then restat stream from it to get the correct one