fix: reorg detector and syncer issues on develop (#272)

* fix: downloader

* fix: driver handling closing download channel

* fix: e2e tests

* fix: increase log level

* fix: new logic

* fix: block finality config

* feat: add an init log to downloader

* fix: comments

* fix: lint

* fix: comments about not adding finalized blocks to reorg detector

* fix: UTs

* fix: comments

bridgesync/bridgesync.go

@@ -12,8 +12,8 @@ import (
 )
 
 const (
-	bridgeSyncL1       = "L1"
-	bridgeSyncL2       = "L2"
+	bridgeSyncL1       = "BridgeSyncL1"
+	bridgeSyncL2       = "BridgeSyncL2"
 	downloadBufferSize = 1000
 )
 
@@ -45,6 +45,7 @@ func NewL1(
 	maxRetryAttemptsAfterError int,
 	originNetwork uint32,
 	syncFullClaims bool,
+	finalizedBlockType etherman.BlockNumberFinality,
 ) (*BridgeSync, error) {
 	return newBridgeSync(
 		ctx,
@@ -61,6 +62,7 @@ func NewL1(
 		maxRetryAttemptsAfterError,
 		originNetwork,
 		syncFullClaims,
+		finalizedBlockType,
 	)
 }
 
@@ -79,6 +81,7 @@ func NewL2(
 	maxRetryAttemptsAfterError int,
 	originNetwork uint32,
 	syncFullClaims bool,
+	finalizedBlockType etherman.BlockNumberFinality,
 ) (*BridgeSync, error) {
 	return newBridgeSync(
 		ctx,
@@ -95,6 +98,7 @@ func NewL2(
 		maxRetryAttemptsAfterError,
 		originNetwork,
 		syncFullClaims,
+		finalizedBlockType,
 	)
 }
 
@@ -113,6 +117,7 @@ func newBridgeSync(
 	maxRetryAttemptsAfterError int,
 	originNetwork uint32,
 	syncFullClaims bool,
+	finalizedBlockType etherman.BlockNumberFinality,
 ) (*BridgeSync, error) {
 	logger := log.WithFields("bridge-syncer", layerID)
 	processor, err := newProcessor(dbPath, logger)
@@ -151,6 +156,7 @@ func newBridgeSync(
 		appender,
 		[]common.Address{bridge},
 		rh,
+		finalizedBlockType,
 	)
 	if err != nil {
 		return nil, err

bridgesync/bridgesync_test.go

@@ -56,6 +56,7 @@ func TestNewLx(t *testing.T) {
 		maxRetryAttemptsAfterError,
 		originNetwork,
 		false,
+		blockFinalityType,
 	)
 
 	assert.NoError(t, err)
@@ -77,6 +78,7 @@ func TestNewLx(t *testing.T) {
 		maxRetryAttemptsAfterError,
 		originNetwork,
 		false,
+		blockFinalityType,
 	)
 
 	assert.NoError(t, err)

bridgesync/e2e_test.go

@@ -88,6 +88,8 @@ func TestBridgeEventE2E(t *testing.T) {
 		}
 	}
 
+	helpers.CommitBlocks(t, setup.L1Environment.SimBackend, 11, blockTime)
+
 	// Wait for syncer to catch up
 	time.Sleep(time.Second * 2) // sleeping since the processor could be up to date, but have pending reorgs
 	lb, err := setup.L1Environment.SimBackend.Client().BlockNumber(ctx)

cmd/run.go

@@ -526,6 +526,7 @@ func runL1InfoTreeSyncerIfNeeded(
 		cfg.L1InfoTreeSync.RetryAfterErrorPeriod.Duration,
 		cfg.L1InfoTreeSync.MaxRetryAttemptsAfterError,
 		l1infotreesync.FlagNone,
+		etherman.FinalizedBlock,
 	)
 	if err != nil {
 		log.Fatal(err)
@@ -727,6 +728,7 @@ func runBridgeSyncL1IfNeeded(
 		cfg.MaxRetryAttemptsAfterError,
 		rollupID,
 		false,
+		etherman.FinalizedBlock,
 	)
 	if err != nil {
 		log.Fatalf("error creating bridgeSyncL1: %s", err)
@@ -762,6 +764,7 @@ func runBridgeSyncL2IfNeeded(
 		cfg.MaxRetryAttemptsAfterError,
 		rollupID,
 		true,
+		etherman.LatestBlock,
 	)
 	if err != nil {
 		log.Fatalf("error creating bridgeSyncL2: %s", err)

etherman/types.go

@@ -120,6 +120,10 @@ func (b *BlockNumberFinality) ToBlockNum() (*big.Int, error) {
 	}
 }
 
+func (b BlockNumberFinality) IsFinalized() bool {
+	return b == FinalizedBlock
+}
+
 type BlockNumber int64
 
 const (

l1infotreesync/e2e_test.go

@@ -70,8 +70,8 @@ func TestE2E(t *testing.T) {
 	rdm.On("AddBlockToTrack", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	client, auth, gerAddr, verifyAddr, gerSc, verifySC := newSimulatedClient(t)
-	syncer, err := l1infotreesync.New(ctx, dbPath, gerAddr, verifyAddr, 10, etherman.LatestBlock, rdm, client.Client(), time.Millisecond, 0, 100*time.Millisecond, 3,
-		l1infotreesync.FlagAllowWrongContractsAddrs)
+	syncer, err := l1infotreesync.New(ctx, dbPath, gerAddr, verifyAddr, 10, etherman.LatestBlock, rdm, client.Client(), time.Millisecond, 0, 100*time.Millisecond, 25,
+		l1infotreesync.FlagAllowWrongContractsAddrs, etherman.SafeBlock)
 	require.NoError(t, err)
 
 	go syncer.Start(ctx)
@@ -159,7 +159,7 @@ func TestWithReorgs(t *testing.T) {
 	require.NoError(t, rd.Start(ctx))
 
 	syncer, err := l1infotreesync.New(ctx, dbPathSyncer, gerAddr, verifyAddr, 10, etherman.LatestBlock, rd, client.Client(), time.Millisecond, 0, time.Second, 25,
-		l1infotreesync.FlagAllowWrongContractsAddrs)
+		l1infotreesync.FlagAllowWrongContractsAddrs, etherman.SafeBlock)
 	require.NoError(t, err)
 	go syncer.Start(ctx)
 
@@ -221,9 +221,6 @@ func TestWithReorgs(t *testing.T) {
 	// Block 4, 5, 6 after the fork
 	helpers.CommitBlocks(t, client, 3, time.Millisecond*500)
 
-	// Make sure syncer is up to date
-	waitForSyncerToCatchUp(ctx, t, syncer, client)
-
 	// Assert rollup exit root after the fork - should be zero since there are no events in the block after the fork
 	expectedRollupExitRoot, err = verifySC.GetRollupExitRoot(&bind.CallOpts{Pending: false})
 	require.NoError(t, err)
@@ -236,11 +233,12 @@ func TestWithReorgs(t *testing.T) {
 	require.NoError(t, err)
 	time.Sleep(time.Millisecond * 500)
 
+	helpers.CommitBlocks(t, client, 1, time.Millisecond*100)
+
 	// create some events and update the trees
 	updateL1InfoTreeAndRollupExitTree(2, 1)
 
-	// Block 4, 5, 6, 7 after the fork
-	helpers.CommitBlocks(t, client, 4, time.Millisecond*100)
+	helpers.CommitBlocks(t, client, 1, time.Millisecond*100)
 
 	// Make sure syncer is up to date
 	waitForSyncerToCatchUp(ctx, t, syncer, client)
@@ -274,7 +272,7 @@ func TestStressAndReorgs(t *testing.T) {
 	require.NoError(t, rd.Start(ctx))
 
 	syncer, err := l1infotreesync.New(ctx, dbPathSyncer, gerAddr, verifyAddr, 10, etherman.LatestBlock, rd, client.Client(), time.Millisecond, 0, time.Second, 100,
-		l1infotreesync.FlagAllowWrongContractsAddrs)
+		l1infotreesync.FlagAllowWrongContractsAddrs, etherman.SafeBlock)
 	require.NoError(t, err)
 	go syncer.Start(ctx)
 
@@ -305,7 +303,7 @@ func TestStressAndReorgs(t *testing.T) {
 		}
 	}
 
-	helpers.CommitBlocks(t, client, 1, time.Millisecond*10)
+	helpers.CommitBlocks(t, client, 11, time.Millisecond*10)
 
 	waitForSyncerToCatchUp(ctx, t, syncer, client)
 

l1infotreesync/l1infotreesync.go

@@ -47,6 +47,7 @@ func New(
 	retryAfterErrorPeriod time.Duration,
 	maxRetryAttemptsAfterError int,
 	flags CreationFlags,
+	finalizedBlockType etherman.BlockNumberFinality,
 ) (*L1InfoTreeSync, error) {
 	processor, err := newProcessor(dbPath)
 	if err != nil {
@@ -83,6 +84,7 @@ func New(
 		appender,
 		[]common.Address{globalExitRoot, rollupManager},
 		rh,
+		finalizedBlockType,
 	)
 	if err != nil {
 		return nil, err

l1infotreesync/processor.go

@@ -242,7 +242,6 @@ func (p *processor) Reorg(ctx context.Context, firstReorgedBlock uint64) error {
 	shouldRollback := true
 	defer func() {
 		if shouldRollback {
-			log.Debugf("rolling back reorg, first reorged block: %d", firstReorgedBlock)
 			if errRllbck := tx.Rollback(); errRllbck != nil {
 				log.Errorf("error while rolling back tx %v", errRllbck)
 			}
@@ -269,6 +268,9 @@ func (p *processor) Reorg(ctx context.Context, firstReorgedBlock uint64) error {
 	if err := tx.Commit(); err != nil {
 		return err
 	}
+
+	shouldRollback = false
+
 	sync.UnhaltIfAffectedRows(&p.halted, &p.haltedReason, &p.mu, rowsAffected)
 	return nil
 }