Add CCTP rebalancing to RFQ relayer

services/rfq/relayer/listener/export_test.go → ethergo/chain/listener/export_test.go

@@ -2,10 +2,10 @@ package listener
 
 import (
 	"context"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/synapsecns/sanguine/core/metrics"
 	"github.com/synapsecns/sanguine/ethergo/client"
-	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 )
 
@@ -21,18 +21,19 @@ func (c chainListener) GetMetadata(ctx context.Context) (startBlock, chainID uin
 }
 
 type TestChainListenerArgs struct {
-	Address  common.Address
-	Client   client.EVM
-	Contract *fastbridge.FastBridgeRef
-	Store    reldb.Service
-	Handler  metrics.Handler
+	Address      common.Address
+	InitialBlock uint64
+	Client       client.EVM
+	Store        reldb.Service
+	Handler      metrics.Handler
 }
 
 func NewTestChainListener(args TestChainListenerArgs) TestChainListener {
 	return &chainListener{
-		client:   args.Client,
-		contract: args.Contract,
-		store:    args.Store,
-		handler:  args.Handler,
+		client:       args.Client,
+		address:      args.Address,
+		initialBlock: args.InitialBlock,
+		store:        args.Store,
+		handler:      args.Handler,
 	}
 }

services/rfq/relayer/listener/listener.go → ethergo/chain/listener/listener.go

@@ -4,21 +4,19 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
+	"time"
+
 	"github.com/ethereum/go-ethereum"
-	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ipfs/go-log"
 	"github.com/jpillora/backoff"
 	"github.com/synapsecns/sanguine/core/metrics"
 	"github.com/synapsecns/sanguine/ethergo/client"
-	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
-	"golang.org/x/sync/errgroup"
-	"math/big"
-	"time"
 )
 
 // ContractListener listens for chain events and calls HandleLog.
@@ -37,11 +35,12 @@ type ContractListener interface {
 type HandleLog func(ctx context.Context, log types.Log) error
 
 type chainListener struct {
-	client   client.EVM
-	contract *fastbridge.FastBridgeRef
-	store    reldb.Service
-	handler  metrics.Handler
-	backoff  *backoff.Backoff
+	client       client.EVM
+	address      common.Address
+	initialBlock uint64
+	store        reldb.Service
+	handler      metrics.Handler
+	backoff      *backoff.Backoff
 	// IMPORTANT! These fields cannot be used until they has been set. They are NOT
 	// set in the constructor
 	startBlock, chainID, latestBlock uint64
@@ -52,18 +51,14 @@ type chainListener struct {
 var logger = log.Logger("chainlistener-logger")
 
 // NewChainListener creates a new chain listener.
-func NewChainListener(omnirpcClient client.EVM, store reldb.Service, address common.Address, handler metrics.Handler) (ContractListener, error) {
-	fastBridge, err := fastbridge.NewFastBridgeRef(address, omnirpcClient)
-	if err != nil {
-		return nil, fmt.Errorf("could not create fast bridge contract: %w", err)
-	}
-
+func NewChainListener(omnirpcClient client.EVM, store reldb.Service, address common.Address, initialBlock uint64, handler metrics.Handler) (ContractListener, error) {
 	return &chainListener{
-		handler:  handler,
-		store:    store,
-		client:   omnirpcClient,
-		contract: fastBridge,
-		backoff:  newBackoffConfig(),
+		handler:      handler,
+		address:      address,
+		initialBlock: initialBlock,
+		store:        store,
+		client:       omnirpcClient,
+		backoff:      newBackoffConfig(),
 	}, nil
 }
 
@@ -91,13 +86,12 @@ func (c *chainListener) Listen(ctx context.Context, handler HandleLog) (err erro
 			if err != nil {
 				logger.Warn(err)
 			}
-
 		}
 	}
 }
 
 func (c *chainListener) Address() common.Address {
-	return c.contract.Address()
+	return c.address
 }
 
 func (c *chainListener) LatestBlock() uint64 {
@@ -125,9 +119,8 @@ func (c *chainListener) doPoll(parentCtx context.Context, handler HandleLog) (er
 	}
 
 	// Check if latest block is the same as start block (for chains with slow block times)
-
 	if c.latestBlock == c.startBlock {
-		return
+		return nil
 	}
 
 	// Handle if the listener is more than one get logs range behind the head
@@ -164,7 +157,7 @@ func (c *chainListener) doPoll(parentCtx context.Context, handler HandleLog) (er
 }
 
 func (c chainListener) getMetadata(parentCtx context.Context) (startBlock, chainID uint64, err error) {
-	var deployBlock, lastIndexed uint64
+	var lastIndexed uint64
 	ctx, span := c.handler.Tracer().Start(parentCtx, "getMetadata")
 
 	defer func() {
@@ -173,48 +166,29 @@ func (c chainListener) getMetadata(parentCtx context.Context) (startBlock, chain
 
 	// TODO: consider some kind of backoff here in case rpcs are down at boot.
 	// this becomes more of an issue as we add more chains
-	g, ctx := errgroup.WithContext(ctx)
-	g.Go(func() error {
-		deployBlock, err := c.contract.DeployBlock(&bind.CallOpts{Context: ctx})
-		if err != nil {
-			return fmt.Errorf("could not get deploy block: %w", err)
-		}
-
-		startBlock = deployBlock.Uint64()
-		return nil
-	})
-
-	g.Go(func() error {
-		// TODO: one thing I've been going back and forth on is whether or not this method should be chain aware
-		// passing in the chain ID would allow us to pull everything directly from the config, but be less testable
-		// for now, this is probably the best solution for testability, but it's certainly a bit annoying we need to do
-		// an rpc call in order to get the chain id
-		//
-		rpcChainID, err := c.client.ChainID(ctx)
-		if err != nil {
-			return fmt.Errorf("could not get chain ID: %w", err)
-		}
-		chainID = rpcChainID.Uint64()
-
-		lastIndexed, err = c.store.LatestBlockForChain(ctx, chainID)
-		if errors.Is(err, reldb.ErrNoLatestBlockForChainID) {
-			// TODO: consider making this negative 1, requires type change
-			lastIndexed = 0
-			return nil
-		}
-		if err != nil {
-			return fmt.Errorf("could not get the latest block for chainID: %w", err)
-		}
-		return nil
-	})
-
-	err = g.Wait()
+	// TODO: one thing I've been going back and forth on is whether or not this method should be chain aware
+	// passing in the chain ID would allow us to pull everything directly from the config, but be less testable
+	// for now, this is probably the best solution for testability, but it's certainly a bit annoying we need to do
+	// an rpc call in order to get the chain id
+	//
+	rpcChainID, err := c.client.ChainID(ctx)
 	if err != nil {
-		return 0, 0, fmt.Errorf("could not get metadata: %w", err)
+		return 0, 0, fmt.Errorf("could not get chain ID: %w", err)
+	}
+	chainID = rpcChainID.Uint64()
+
+	lastIndexed, err = c.store.LatestBlockForChain(ctx, chainID)
+	if errors.Is(err, reldb.ErrNoLatestBlockForChainID) {
+		// TODO: consider making this negative 1, requires type change
+		lastIndexed = 0
+	} else if err != nil {
+		return 0, 0, fmt.Errorf("could not get the latest block for chainID: %w", err)
 	}
 
-	if lastIndexed > deployBlock {
+	if lastIndexed > c.startBlock {
 		startBlock = lastIndexed
+	} else {
+		startBlock = c.initialBlock
 	}
 
 	return startBlock, chainID, nil
@@ -233,6 +207,6 @@ func (c chainListener) buildFilterQuery(fromBlock, toBlock uint64) ethereum.Filt
 	return ethereum.FilterQuery{
 		FromBlock: new(big.Int).SetUint64(fromBlock),
 		ToBlock:   new(big.Int).SetUint64(toBlock),
-		Addresses: []common.Address{c.contract.Address()},
+		Addresses: []common.Address{c.address},
 	}
 }

services/rfq/relayer/listener/listener_test.go → ethergo/chain/listener/listener_test.go

@@ -7,11 +7,12 @@ import (
 	"time"
 
 	"github.com/brianvoe/gofakeit/v6"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/synapsecns/sanguine/ethergo/chain/listener"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/testcontracts/fastbridgemock"
-	"github.com/synapsecns/sanguine/services/rfq/relayer/listener"
 )
 
 func (l *ListenerTestSuite) TestListenForEvents() {
@@ -27,7 +28,7 @@ func (l *ListenerTestSuite) TestListenForEvents() {
 			testAddress := common.BigToAddress(big.NewInt(int64(i)))
 			auth := l.backend.GetTxContext(l.GetTestContext(), nil)
 
-			//nolint: typecheck
+			//nolint:typecheck
 			txID := [32]byte(crypto.Keccak256(testAddress.Bytes()))
 			bridgeRequestTX, err := handle.MockBridgeRequest(auth.TransactOpts, txID, testAddress, fastbridgemock.IFastBridgeBridgeParams{
 				DstChainId:   gofakeit.Uint32(),
@@ -72,7 +73,10 @@ func (l *ListenerTestSuite) TestListenForEvents() {
 
 	wg.Wait()
 
-	cl, err := listener.NewChainListener(l.backend, l.store, handle.Address(), l.metrics)
+	startBlock, err := handle.DeployBlock(&bind.CallOpts{Context: l.GetTestContext()})
+	l.NoError(err)
+
+	cl, err := listener.NewChainListener(l.backend, l.store, handle.Address(), uint64(startBlock.Int64()), l.metrics)
 	l.NoError(err)
 
 	eventCount := 0

services/rfq/relayer/listener/suite_test.go → ethergo/chain/listener/suite_test.go

@@ -1,22 +1,22 @@
 package listener_test
 
 import (
+	"math/big"
+	"testing"
+
 	"github.com/Flaque/filet"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/stretchr/testify/suite"
 	"github.com/synapsecns/sanguine/core/metrics"
 	"github.com/synapsecns/sanguine/core/testsuite"
 	"github.com/synapsecns/sanguine/ethergo/backends"
 	"github.com/synapsecns/sanguine/ethergo/backends/geth"
+	"github.com/synapsecns/sanguine/ethergo/chain/listener"
 	"github.com/synapsecns/sanguine/ethergo/contracts"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
-	"github.com/synapsecns/sanguine/services/rfq/relayer/listener"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb/sqlite"
 	"github.com/synapsecns/sanguine/services/rfq/testutil"
-	"math/big"
-	"testing"
 )
 
 const chainID = 10
@@ -55,31 +55,30 @@ func (l *ListenerTestSuite) SetupTest() {
 }
 
 func (l *ListenerTestSuite) TestGetMetadataNoStore() {
+	deployBlock, err := l.fastBridge.DeployBlock(&bind.CallOpts{Context: l.GetTestContext()})
+	l.NoError(err)
+
 	// nothing stored, should use start block
 	cl := listener.NewTestChainListener(listener.TestChainListenerArgs{
-		Address:  common.Address{},
-		Client:   l.backend,
-		Contract: l.fastBridge,
-		Store:    l.store,
-		Handler:  l.metrics,
+		Address:      l.fastBridge.Address(),
+		InitialBlock: deployBlock.Uint64(),
+		Client:       l.backend,
+		Store:        l.store,
+		Handler:      l.metrics,
 	})
 
 	startBlock, myChainID, err := cl.GetMetadata(l.GetTestContext())
 	l.NoError(err)
 	l.Equal(myChainID, uint64(chainID))
-
-	deployBlock, err := l.fastBridge.DeployBlock(&bind.CallOpts{Context: l.GetTestContext()})
-	l.NoError(err)
 	l.Equal(startBlock, deployBlock.Uint64())
 }
 
 func (l *ListenerTestSuite) TestStartBlock() {
 	cl := listener.NewTestChainListener(listener.TestChainListenerArgs{
-		Address:  common.Address{},
-		Client:   l.backend,
-		Contract: l.fastBridge,
-		Store:    l.store,
-		Handler:  l.metrics,
+		Address: l.fastBridge.Address(),
+		Client:  l.backend,
+		Store:   l.store,
+		Handler: l.metrics,
 	})
 
 	deployBlock, err := l.fastBridge.DeployBlock(&bind.CallOpts{Context: l.GetTestContext()})