imporve

l1infotreesync/e2e_test.go

@@ -263,7 +263,7 @@ func TestWithReorgs(t *testing.T) {
 
 func TestStressAndReorgs(t *testing.T) {
 	const (
-		totalIterations       = 3
+		totalIterations       = 300
 		blocksInIteration     = 140
 		reorgEveryXIterations = 70
 		reorgSizeInBlocks     = 2

l1infotreesync/l1infotreesync.go

@@ -106,6 +106,9 @@ func (s *L1InfoTreeSync) Start(ctx context.Context) {
 
 // GetL1InfoTreeMerkleProof creates a merkle proof for the L1 Info tree
 func (s *L1InfoTreeSync) GetL1InfoTreeMerkleProof(ctx context.Context, index uint32) (types.Proof, types.Root, error) {
+	if s.processor.halted {
+		return types.Proof{}, types.Root{}, sync.ErrInconsistentState
+	}
 	return s.processor.GetL1InfoTreeMerkleProof(ctx, index)
 }
 
@@ -115,6 +118,9 @@ func (s *L1InfoTreeSync) GetRollupExitTreeMerkleProof(
 	networkID uint32,
 	root common.Hash,
 ) (types.Proof, error) {
+	if s.processor.halted {
+		return types.Proof{}, sync.ErrInconsistentState
+	}
 	if networkID == 0 {
 		return tree.EmptyProof, nil
 	}
@@ -135,38 +141,59 @@ func translateError(err error) error {
 // - ErrBlockNotProcessed,
 // - ErrNotFound
 func (s *L1InfoTreeSync) GetLatestInfoUntilBlock(ctx context.Context, blockNum uint64) (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	leaf, err := s.processor.GetLatestInfoUntilBlock(ctx, blockNum)
 	return leaf, translateError(err)
 }
 
 // GetInfoByIndex returns the value of a leaf (not the hash) of the L1 info tree
 func (s *L1InfoTreeSync) GetInfoByIndex(ctx context.Context, index uint32) (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetInfoByIndex(ctx, index)
 }
 
 // GetL1InfoTreeRootByIndex returns the root of the L1 info tree at the moment the leaf with the given index was added
 func (s *L1InfoTreeSync) GetL1InfoTreeRootByIndex(ctx context.Context, index uint32) (types.Root, error) {
+	if s.processor.halted {
+		return types.Root{}, sync.ErrInconsistentState
+	}
 	return s.processor.l1InfoTree.GetRootByIndex(ctx, index)
 }
 
 // GetLastRollupExitRoot return the last rollup exit root processed
 func (s *L1InfoTreeSync) GetLastRollupExitRoot(ctx context.Context) (types.Root, error) {
+	if s.processor.halted {
+		return types.Root{}, sync.ErrInconsistentState
+	}
 	return s.processor.rollupExitTree.GetLastRoot(nil)
 }
 
 // GetLastL1InfoTreeRoot return the last root and index processed from the L1 Info tree
 func (s *L1InfoTreeSync) GetLastL1InfoTreeRoot(ctx context.Context) (types.Root, error) {
+	if s.processor.halted {
+		return types.Root{}, sync.ErrInconsistentState
+	}
 	return s.processor.l1InfoTree.GetLastRoot(nil)
 }
 
 // GetLastProcessedBlock return the last processed block
 func (s *L1InfoTreeSync) GetLastProcessedBlock(ctx context.Context) (uint64, error) {
+	if s.processor.halted {
+		return 0, sync.ErrInconsistentState
+	}
 	return s.processor.GetLastProcessedBlock(ctx)
 }
 
 func (s *L1InfoTreeSync) GetLocalExitRoot(
 	ctx context.Context, networkID uint32, rollupExitRoot common.Hash,
 ) (common.Hash, error) {
+	if s.processor.halted {
+		return common.Hash{}, sync.ErrInconsistentState
+	}
 	if networkID == 0 {
 		return common.Hash{}, errors.New("network 0 is not a rollup, and it's not part of the rollup exit tree")
 	}
@@ -175,45 +202,75 @@ func (s *L1InfoTreeSync) GetLocalExitRoot(
 }
 
 func (s *L1InfoTreeSync) GetLastVerifiedBatches(rollupID uint32) (*VerifyBatches, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetLastVerifiedBatches(rollupID)
 }
 
 func (s *L1InfoTreeSync) GetFirstVerifiedBatches(rollupID uint32) (*VerifyBatches, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetFirstVerifiedBatches(rollupID)
 }
 
 func (s *L1InfoTreeSync) GetFirstVerifiedBatchesAfterBlock(rollupID uint32, blockNum uint64) (*VerifyBatches, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetFirstVerifiedBatchesAfterBlock(rollupID, blockNum)
 }
 
 func (s *L1InfoTreeSync) GetFirstL1InfoWithRollupExitRoot(rollupExitRoot common.Hash) (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetFirstL1InfoWithRollupExitRoot(rollupExitRoot)
 }
 
 func (s *L1InfoTreeSync) GetLastInfo() (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetLastInfo()
 }
 
 func (s *L1InfoTreeSync) GetFirstInfo() (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetFirstInfo()
 }
 
 func (s *L1InfoTreeSync) GetFirstInfoAfterBlock(blockNum uint64) (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetFirstInfoAfterBlock(blockNum)
 }
 
 func (s *L1InfoTreeSync) GetInfoByGlobalExitRoot(ger common.Hash) (*L1InfoTreeLeaf, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetInfoByGlobalExitRoot(ger)
 }
 
 // GetL1InfoTreeMerkleProofFromIndexToRoot creates a merkle proof for the L1 Info tree
 func (s *L1InfoTreeSync) GetL1InfoTreeMerkleProofFromIndexToRoot(
 	ctx context.Context, index uint32, root common.Hash,
 ) (types.Proof, error) {
+	if s.processor.halted {
+		return types.Proof{}, sync.ErrInconsistentState
+	}
 	return s.processor.l1InfoTree.GetProof(ctx, index, root)
 }
 
 // GetInitL1InfoRootMap returns the initial L1 info root map, nil if no root map has been set
 func (s *L1InfoTreeSync) GetInitL1InfoRootMap(ctx context.Context) (*L1InfoTreeInitial, error) {
+	if s.processor.halted {
+		return nil, sync.ErrInconsistentState
+	}
 	return s.processor.GetInitL1InfoRootMap(nil)
 }

l1infotreesync/processor.go

@@ -279,7 +279,7 @@ func (p *processor) Reorg(ctx context.Context, firstReorgedBlock uint64) error {
 func (p *processor) ProcessBlock(ctx context.Context, block sync.Block) error {
 	if p.halted {
 		log.Errorf("processor is halted due to: %s", p.haltedReason)
-		return nil
+		return sync.ErrInconsistentState
 	}
 	tx, err := db.NewTx(ctx, p.db)
 	if err != nil {
@@ -352,7 +352,6 @@ func (p *processor) ProcessBlock(ctx context.Context, block sync.Block) error {
 			// failed due to a reorg. Hopefully, this is the case, eventually the reorg will get detected,
 			// and the syncer will get unhalted. Otherwise, this means that the syncer has an inconsistent state
 			// compared to the contracts, and this will need manual intervention.
-			// Do not return an error to give the driver the opportunity to detect a reorg
 			if root.Hash != event.UpdateL1InfoTreeV2.CurrentL1InfoRoot {
 				errStr := fmt.Sprintf(
 					"unexpected root when checking UpdateL1InfoTreeV2: %s vs %s. Happened on block %d",
@@ -361,7 +360,7 @@ func (p *processor) ProcessBlock(ctx context.Context, block sync.Block) error {
 				log.Error(errStr)
 				p.haltedReason = errStr
 				p.halted = true
-				return nil
+				return sync.ErrInconsistentState
 			}
 			if root.Index+1 != event.UpdateL1InfoTreeV2.LeafCount {
 				errStr := fmt.Sprintf(
@@ -371,7 +370,7 @@ func (p *processor) ProcessBlock(ctx context.Context, block sync.Block) error {
 				log.Error(errStr)
 				p.haltedReason = errStr
 				p.halted = true
-				return nil
+				return sync.ErrInconsistentState
 			}
 		}
 		if event.VerifyBatches != nil {

sync/driver.go

@@ -1,6 +1,11 @@
 package sync
 
-import "context"
+import (
+	"context"
+	"errors"
+)
+
+var ErrInconsistentState = errors.New("state is inconsistent, try again later once the state is fixed")
 
 type Block struct {
 	Num    uint64

sync/evmdriver.go

@@ -98,7 +98,7 @@ reset:
 			return
 		case b := <-downloadCh:
 			d.log.Debugf("handleNewBlock, blockNum: %d, blockHash: %s", b.Num, b.Hash)
-			d.handleNewBlock(ctx, b)
+			d.handleNewBlock(ctx, cancel, b)
 		case firstReorgedBlock := <-d.reorgSub.ReorgedBlock:
 			d.log.Debug("handleReorg from block: ", firstReorgedBlock)
 			d.handleReorg(ctx, cancel, firstReorgedBlock)
@@ -107,7 +107,7 @@ reset:
 	}
 }
 
-func (d *EVMDriver) handleNewBlock(ctx context.Context, b EVMBlock) {
+func (d *EVMDriver) handleNewBlock(ctx context.Context, cancel context.CancelFunc, b EVMBlock) {
 	attempts := 0
 	for {
 		err := d.reorgDetector.AddBlockToTrack(ctx, d.reorgDetectorID, b.Num, b.Hash)
@@ -127,6 +127,11 @@ func (d *EVMDriver) handleNewBlock(ctx context.Context, b EVMBlock) {
 		}
 		err := d.processor.ProcessBlock(ctx, blockToProcess)
 		if err != nil {
+			if err == ErrInconsistentState {
+				log.Warn("state got inconsistent after processing this block. Stopping downloader until there is a reorg")
+				cancel()
+				return
+			}
 			attempts++
 			d.log.Errorf("error processing events for block %d, err: ", b.Num, err)
 			d.rh.Handle("handleNewBlock", attempts)

sync/evmdriver_test.go

@@ -144,7 +144,7 @@ func TestHandleNewBlock(t *testing.T) {
 		Return(nil)
 	pm.On("ProcessBlock", ctx, Block{Num: b1.Num, Events: b1.Events}).
 		Return(nil)
-	driver.handleNewBlock(ctx, b1)
+	driver.handleNewBlock(ctx, nil, b1)
 
 	// reorg deteector fails once
 	b2 := EVMBlock{
@@ -161,7 +161,7 @@ func TestHandleNewBlock(t *testing.T) {
 		Return(nil).Once()
 	pm.On("ProcessBlock", ctx, Block{Num: b2.Num, Events: b2.Events}).
 		Return(nil)
-	driver.handleNewBlock(ctx, b2)
+	driver.handleNewBlock(ctx, nil, b2)
 
 	// processor fails once
 	b3 := EVMBlock{
@@ -177,7 +177,7 @@ func TestHandleNewBlock(t *testing.T) {
 		Return(errors.New("foo")).Once()
 	pm.On("ProcessBlock", ctx, Block{Num: b3.Num, Events: b3.Events}).
 		Return(nil).Once()
-	driver.handleNewBlock(ctx, b3)
+	driver.handleNewBlock(ctx, nil, b3)
 }
 
 func TestHandleReorg(t *testing.T) {