pass E2E

bridgesync/bridgesync.go

@@ -37,6 +37,7 @@ func NewL1(
 	rd sync.ReorgDetector,
 	ethClient EthClienter,
 	initialBlock uint64,
+	waitForNewBlocksPeriod time.Duration,
 ) (*BridgeSync, error) {
 	return new(
 		ctx,
@@ -49,6 +50,7 @@ func NewL1(
 		initialBlock,
 		dbPrefixL1,
 		reorgDetectorIDL1,
+		waitForNewBlocksPeriod,
 	)
 }
 
@@ -62,6 +64,7 @@ func NewL2(
 	rd sync.ReorgDetector,
 	ethClient EthClienter,
 	initialBlock uint64,
+	waitForNewBlocksPeriod time.Duration,
 ) (*BridgeSync, error) {
 	return new(
 		ctx,
@@ -74,6 +77,7 @@ func NewL2(
 		initialBlock,
 		dbPrefixL1,
 		reorgDetectorIDL1,
+		waitForNewBlocksPeriod,
 	)
 }
 
@@ -87,6 +91,7 @@ func new(
 	ethClient EthClienter,
 	initialBlock uint64,
 	dbPrefix, reorgDetectorID string,
+	waitForNewBlocksPeriod time.Duration,
 ) (*BridgeSync, error) {
 	processor, err := newProcessor(ctx, dbPath, dbPrefix)
 	if err != nil {

bridgesync/downloader.go

@@ -3,7 +3,6 @@ package bridgesync
 import (
 	"fmt"
 	"math/big"
-	"time"
 
 	"github.com/0xPolygon/cdk-contracts-tooling/contracts/etrog/polygonzkevmbridge"
 	"github.com/0xPolygon/cdk-contracts-tooling/contracts/etrog/polygonzkevmbridgev2"
@@ -15,10 +14,6 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 )
 
-const (
-	waitForNewBlocksPeriod = time.Millisecond * 100
-)
-
 var (
 	bridgeEventSignature        = crypto.Keccak256Hash([]byte("BridgeEvent(uint8,uint32,address,uint32,address,uint256,bytes,uint32)"))
 	claimEventSignature         = crypto.Keccak256Hash([]byte("ClaimEvent(uint256,uint32,address,address,uint256)"))

bridgesync/e2e_test.go

@@ -54,7 +54,7 @@ func TestBridgeEventE2E(t *testing.T) {
 	rd, err := reorgdetector.New(ctx, client.Client(), dbPathReorg)
 	go rd.Start(ctx)
 
-	syncer, err := bridgesync.NewL1(ctx, dbPathSyncer, bridgeAddr, 10, etherman.LatestBlock, rd, client.Client(), 0)
+	syncer, err := bridgesync.NewL1(ctx, dbPathSyncer, bridgeAddr, 10, etherman.LatestBlock, rd, client.Client(), 0, time.Millisecond*10)
 	require.NoError(t, err)
 	go syncer.Start(ctx)
 

bridgesync/processor.go

@@ -296,3 +296,21 @@ func (p *processor) updateLastProcessedBlock(tx kv.RwTx, blockNum uint64) error
 	blockNumBytes := dbCommon.Uint64ToBytes(blockNum)
 	return tx.Put(p.lastBlockTable, lastBlokcKey, blockNumBytes)
 }
+
+func GenerateGlobalIndex(mainnetFlag bool, rollupIndex uint, localExitRootIndex uint32) *big.Int {
+	var (
+		globalIndexBytes []byte
+		buf              [4]byte
+	)
+	if mainnetFlag {
+		globalIndexBytes = append(globalIndexBytes, big.NewInt(1).Bytes()...)
+		ri := big.NewInt(0).FillBytes(buf[:])
+		globalIndexBytes = append(globalIndexBytes, ri...)
+	} else {
+		ri := big.NewInt(0).SetUint64(uint64(rollupIndex)).FillBytes(buf[:])
+		globalIndexBytes = append(globalIndexBytes, ri...)
+	}
+	leri := big.NewInt(0).SetUint64(uint64(localExitRootIndex)).FillBytes(buf[:])
+	globalIndexBytes = append(globalIndexBytes, leri...)
+	return big.NewInt(0).SetBytes(globalIndexBytes)
+}

claimsponsor/claimsponsor.go

@@ -60,17 +60,25 @@ func (c *Claim) Key() []byte {
 type ClaimSender interface {
 	checkClaim(ctx context.Context, claim *Claim) error
 	sendClaim(ctx context.Context, claim *Claim) (string, error)
-	claimStatus(id string) (ClaimStatus, error)
+	claimStatus(ctx context.Context, id string) (ClaimStatus, error)
 }
 
 type ClaimSponsor struct {
 	db                    kv.RwDB
 	sender                ClaimSender
 	rh                    *sync.RetryHandler
 	waitTxToBeMinedPeriod time.Duration
+	waitOnEmptyQueue      time.Duration
 }
 
-func newClaimSponsor(dbPath string) (*ClaimSponsor, error) {
+func newClaimSponsor(
+	dbPath string,
+	sender ClaimSender,
+	retryAfterErrorPeriod time.Duration,
+	maxRetryAttemptsAfterError int,
+	waitTxToBeMinedPeriod time.Duration,
+	waitOnEmptyQueue time.Duration,
+) (*ClaimSponsor, error) {
 	tableCfgFunc := func(defaultBuckets kv.TableCfg) kv.TableCfg {
 		cfg := kv.TableCfg{
 			claimTable: {},
@@ -85,8 +93,16 @@ func newClaimSponsor(dbPath string) (*ClaimSponsor, error) {
 	if err != nil {
 		return nil, err
 	}
+	rh := &sync.RetryHandler{
+		MaxRetryAttemptsAfterError: maxRetryAttemptsAfterError,
+		RetryAfterErrorPeriod:      retryAfterErrorPeriod,
+	}
 	return &ClaimSponsor{
-		db: db,
+		db:                    db,
+		sender:                sender,
+		rh:                    rh,
+		waitTxToBeMinedPeriod: waitTxToBeMinedPeriod,
+		waitOnEmptyQueue:      waitOnEmptyQueue,
 	}, nil
 }
 
@@ -98,7 +114,7 @@ func (c *ClaimSponsor) Start(ctx context.Context) {
 	for {
 		if err != nil {
 			attempts++
-			c.rh.Handle("claimsponsor start", attempts)
+			c.rh.Handle("claimsponsor main loop", attempts)
 		}
 		tx, err2 := c.db.BeginRw(ctx)
 		if err2 != nil {
@@ -110,6 +126,12 @@ func (c *ClaimSponsor) Start(ctx context.Context) {
 		if err2 != nil {
 			err = err2
 			tx.Rollback()
+			if err == ErrNotFound {
+				log.Debugf("queue is empty")
+				err = nil
+				time.Sleep(c.waitOnEmptyQueue)
+				continue
+			}
 			log.Errorf("error calling getFirstQueueIndex: %v", err)
 			continue
 		}
@@ -182,6 +204,7 @@ func (c *ClaimSponsor) Start(ctx context.Context) {
 		}
 
 		attempts = 0
+		log.Error("wtf: ", err)
 	}
 }
 
@@ -192,7 +215,7 @@ func (c *ClaimSponsor) waitTxToBeSuccessOrFail(ctx context.Context, txID string)
 		case <-ctx.Done():
 			return "", errors.New("context cancelled")
 		case <-t.C:
-			status, err := c.sender.claimStatus(txID)
+			status, err := c.sender.claimStatus(ctx, txID)
 			if err != nil {
 				return "", err
 			}
@@ -216,8 +239,10 @@ func (c *ClaimSponsor) AddClaimToQueue(ctx context.Context, claim *Claim) error
 	_, err = getClaim(tx, claim.GlobalIndex)
 	if err != ErrNotFound {
 		if err != nil {
+			tx.Rollback()
 			return err
 		} else {
+			tx.Rollback()
 			return errors.New("claim already added")
 		}
 	}
@@ -228,12 +253,16 @@ func (c *ClaimSponsor) AddClaimToQueue(ctx context.Context, claim *Claim) error
 		return err
 	}
 
+	var queuePosition uint64
 	lastQueuePosition, _, err := getLastQueueIndex(tx)
-	if err != nil {
+	if err == ErrNotFound {
+		queuePosition = 0
+	} else if err != nil {
 		tx.Rollback()
 		return err
+	} else {
+		queuePosition = lastQueuePosition + 1
 	}
-	queuePosition := lastQueuePosition + 1
 	err = tx.Put(queueTable, dbCommon.Uint64ToBytes(queuePosition), claim.Key())
 	if err != nil {
 		tx.Rollback()

claimsponsor/e2e_test.go

@@ -2,6 +2,7 @@ package claimsponsor_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/big"
 	"testing"
@@ -21,23 +22,31 @@ func TestE2EL1toEVML2(t *testing.T) {
 	ctx := context.Background()
 	env := helpers.SetupAggoracleWithEVMChain(t)
 	dbPathBridgeSyncL1 := t.TempDir()
-	bridgeSyncL1, err := bridgesync.NewL1(ctx, dbPathBridgeSyncL1, env.BridgeL1Addr, 10, etherman.LatestBlock, env.ReorgDetector, env.L1Client.Client(), 0)
+	bridgeSyncL1, err := bridgesync.NewL1(ctx, dbPathBridgeSyncL1, env.BridgeL1Addr, 10, etherman.LatestBlock, env.ReorgDetector, env.L1Client.Client(), 0, time.Millisecond*10)
 	require.NoError(t, err)
 	go bridgeSyncL1.Start(ctx)
 
 	// start claim sponsor
 	dbPathClaimSponsor := t.TempDir()
-	ethTxMock := helpers.NewEthTxManagerMock(t)
-	claimer, err := claimsponsor.NewEVMClaimSponsor(dbPathClaimSponsor, env.L2Client.Client(), env.BridgeL2Addr, env.AuthL2.From, 100_000, 0, ethTxMock)
+	claimer, err := claimsponsor.NewEVMClaimSponsor(
+		dbPathClaimSponsor,
+		env.L2Client.Client(),
+		env.BridgeL2Addr,
+		env.AuthL2.From,
+		200_000,
+		0,
+		env.EthTxManMockL2,
+		0, 0, time.Millisecond*10, time.Millisecond*10,
+	)
 	require.NoError(t, err)
 	go claimer.Start(ctx)
 
 	// test
 	for i := 0; i < 10; i++ {
 		// Send bridges to L2, wait for GER to be injected on L2
 		amount := big.NewInt(int64(i) + 1)
+		env.AuthL1.Value = amount
 		_, err := env.BridgeL1Contract.BridgeAsset(env.AuthL1, env.NetworkIDL2, env.AuthL2.From, amount, common.Address{}, true, nil)
-		require.NoError(t, err)
 		env.L1Client.Commit()
 		time.Sleep(time.Millisecond * 50)
 		expectedGER, err := env.GERL1Contract.GetLastGlobalExitRoot(&bind.CallOpts{Pending: false})
@@ -54,11 +63,12 @@ func TestE2EL1toEVML2(t *testing.T) {
 		rollupProof, err := env.L1InfoTreeSync.GetRollupExitTreeMerkleProof(ctx, 0, common.Hash{})
 
 		// Request to sponsor claim
-		claimer.AddClaimToQueue(ctx, &claimsponsor.Claim{
+		globalIndex := bridgesync.GenerateGlobalIndex(true, 0, uint32(i))
+		err = claimer.AddClaimToQueue(ctx, &claimsponsor.Claim{
 			LeafType:            0,
 			ProofLocalExitRoot:  localProof,
 			ProofRollupExitRoot: rollupProof,
-			GlobalIndex:         nil, // TODO
+			GlobalIndex:         globalIndex,
 			MainnetExitRoot:     info.MainnetExitRoot,
 			RollupExitRoot:      info.RollupExitRoot,
 			OriginNetwork:       0,
@@ -68,8 +78,22 @@ func TestE2EL1toEVML2(t *testing.T) {
 			Amount:              amount,
 			Metadata:            nil,
 		})
+		require.NoError(t, err)
 
-		// TODO: Wait until success
+		// Wait until success
+		succeed := false
+		for i := 0; i < 10; i++ {
+			claim, err := claimer.GetClaim(ctx, globalIndex)
+			require.NoError(t, err)
+			if claim.Status == claimsponsor.FailedClaimStatus {
+				require.NoError(t, errors.New("claim failed"))
+			} else if claim.Status == claimsponsor.SuccessClaimStatus {
+				succeed = true
+				break
+			}
+			time.Sleep(50 * time.Millisecond)
+		}
+		require.True(t, succeed)
 
 		// Check on contract that is claimed
 		isClaimed, err := env.BridgeL2Contract.IsClaimed(&bind.CallOpts{Pending: false}, uint32(i), 0)

claimsponsor/evmclaimsponsor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"time"
 
 	"github.com/0xPolygon/cdk-contracts-tooling/contracts/etrog/polygonzkevmbridgev2"
 	"github.com/0xPolygonHermez/zkevm-ethtx-manager/ethtxmanager"
@@ -53,6 +54,10 @@ func NewEVMClaimSponsor(
 	sender common.Address,
 	maxGas, gasOffset uint64,
 	ethTxManager EthTxManager,
+	retryAfterErrorPeriod time.Duration,
+	maxRetryAttemptsAfterError int,
+	waitTxToBeMinedPeriod time.Duration,
+	waitOnEmptyQueue time.Duration,
 ) (*EVMClaimSponsor, error) {
 	contract, err := polygonzkevmbridgev2.NewPolygonzkevmbridgev2(bridge, l2Client)
 	if err != nil {
@@ -62,12 +67,7 @@ func NewEVMClaimSponsor(
 	if err != nil {
 		return nil, err
 	}
-	sponsor, err := newClaimSponsor(dbPath)
-	if err != nil {
-		return nil, err
-	}
-	return &EVMClaimSponsor{
-		ClaimSponsor:   sponsor,
+	evmSponsor := &EVMClaimSponsor{
 		l2Client:       l2Client,
 		bridgeABI:      abi,
 		bridgeAddr:     bridge,
@@ -76,7 +76,20 @@ func NewEVMClaimSponsor(
 		gasOffest:      gasOffset,
 		maxGas:         maxGas,
 		ethTxManager:   ethTxManager,
-	}, nil
+	}
+	baseSponsor, err := newClaimSponsor(
+		dbPath,
+		evmSponsor,
+		retryAfterErrorPeriod,
+		maxRetryAttemptsAfterError,
+		waitTxToBeMinedPeriod,
+		waitOnEmptyQueue,
+	)
+	if err != nil {
+		return nil, err
+	}
+	evmSponsor.ClaimSponsor = baseSponsor
+	return evmSponsor, nil
 }
 
 func (c *EVMClaimSponsor) checkClaim(ctx context.Context, claim *Claim) error {
@@ -98,7 +111,7 @@ func (c *EVMClaimSponsor) checkClaim(ctx context.Context, claim *Claim) error {
 	return nil
 }
 
-func (c *EVMClaimSponsor) sendTx(ctx context.Context, claim *Claim) (string, error) {
+func (c *EVMClaimSponsor) sendClaim(ctx context.Context, claim *Claim) (string, error) {
 	data, err := c.buildClaimTxData(claim)
 	if err != nil {
 		return "", err

l1bridge2infoindexsync/e2e_test.go

@@ -135,7 +135,7 @@ func TestE2E(t *testing.T) {
 	rd, err := reorgdetector.New(ctx, client.Client(), dbPathReorg)
 	go rd.Start(ctx)
 
-	bridgeSync, err := bridgesync.NewL1(ctx, dbPathBridgeSync, bridgeAddr, 10, etherman.LatestBlock, rd, client.Client(), 0)
+	bridgeSync, err := bridgesync.NewL1(ctx, dbPathBridgeSync, bridgeAddr, 10, etherman.LatestBlock, rd, client.Client(), 0, time.Millisecond*10)
 	require.NoError(t, err)
 	go bridgeSync.Start(ctx)