feat(relayer): only process profitable transactions (#408)

packages/relayer/.golangci.yml

@@ -28,7 +28,7 @@ linters:
 
 linters-settings:
   funlen:
-    lines: 117
+    lines: 123
     statements: 50
   gocognit:
     min-complexity: 37

packages/relayer/cli/cli.go

@@ -48,7 +48,13 @@ var (
 	defaultConfirmations       = 15
 )
 
-func Run(mode relayer.Mode, watchMode relayer.WatchMode, layer relayer.Layer, httpOnly relayer.HTTPOnly) {
+func Run(
+	mode relayer.Mode,
+	watchMode relayer.WatchMode,
+	layer relayer.Layer,
+	httpOnly relayer.HTTPOnly,
+	profitableOnly relayer.ProfitableOnly,
+) {
 	if err := loadAndValidateEnv(); err != nil {
 		log.Fatal(err)
 	}
@@ -95,7 +101,7 @@ func Run(mode relayer.Mode, watchMode relayer.WatchMode, layer relayer.Layer, ht
 	}()
 
 	if !httpOnly {
-		indexers, closeFunc, err := makeIndexers(layer, db)
+		indexers, closeFunc, err := makeIndexers(layer, db, profitableOnly)
 		if err != nil {
 			sqlDB.Close()
 			log.Fatal(err)
@@ -116,7 +122,11 @@ func Run(mode relayer.Mode, watchMode relayer.WatchMode, layer relayer.Layer, ht
 	<-forever
 }
 
-func makeIndexers(layer relayer.Layer, db relayer.DB) ([]*indexer.Service, func(), error) {
+func makeIndexers(
+	layer relayer.Layer,
+	db relayer.DB,
+	profitableOnly relayer.ProfitableOnly,
+) ([]*indexer.Service, func(), error) {
 	eventRepository, err := repo.NewEventRepository(db)
 	if err != nil {
 		return nil, nil, err
@@ -192,6 +202,7 @@ func makeIndexers(layer relayer.Layer, db relayer.DB) ([]*indexer.Service, func(
 			NumGoroutines:       numGoroutines,
 			SubscriptionBackoff: subscriptionBackoff,
 			Confirmations:       uint64(confirmations),
+			ProfitableOnly:      profitableOnly,
 		})
 		if err != nil {
 			log.Fatal(err)
@@ -218,6 +229,7 @@ func makeIndexers(layer relayer.Layer, db relayer.DB) ([]*indexer.Service, func(
 			NumGoroutines:       numGoroutines,
 			SubscriptionBackoff: subscriptionBackoff,
 			Confirmations:       uint64(confirmations),
+			ProfitableOnly:      profitableOnly,
 		})
 		if err != nil {
 			log.Fatal(err)

packages/relayer/cli/cli_test.go

@@ -203,7 +203,7 @@ func Test_makeIndexers(t *testing.T) {
 				defer reset()
 			}
 
-			indexers, cancel, err := makeIndexers(tt.layer, tt.dbFunc(t))
+			indexers, cancel, err := makeIndexers(tt.layer, tt.dbFunc(t), relayer.ProfitableOnly(true))
 			if cancel != nil {
 				defer cancel()
 			}

packages/relayer/cmd/main.go

@@ -36,6 +36,12 @@ func main() {
 	  false: run an http server and index blocks
 	`)
 
+	profitableOnlyPtr := flag.Bool("profitable-only", false, `only process profitable transactions. 
+	options:
+	  true:
+	  false:
+	`)
+
 	flag.Parse()
 
 	if !relayer.IsInSlice(relayer.Mode(*modePtr), relayer.Modes) {
@@ -51,5 +57,6 @@ func main() {
 		relayer.WatchMode(*watchModePtr),
 		relayer.Layer(*layersPtr),
 		relayer.HTTPOnly(*httpOnlyPtr),
+		relayer.ProfitableOnly(*profitableOnlyPtr),
 	)
 }

packages/relayer/errors.go

@@ -21,5 +21,10 @@ var (
 		"ERR_INVALID_CONFIRMATIONS",
 		"Confirmations amount is invalid, must be numerical and > 0",
 	)
-	ErrInvalidMode = errors.Validation.NewWithKeyAndDetail("ERR_INVALID_MODE", "Mode not supported")
+	ErrInvalidMode  = errors.Validation.NewWithKeyAndDetail("ERR_INVALID_MODE", "Mode not supported")
+	ErrUnprofitable = errors.Validation.NewWithKeyAndDetail("ERR_UNPROFITABLE", "Transaction is unprofitable to process")
+	ErrNotReceived  = errors.BadRequest.NewWithKeyAndDetail(
+		"ERR_NOT_RECEIVED",
+		"Message not received on destination chain",
+	)
 )

packages/relayer/event.go

@@ -51,5 +51,13 @@ type SaveEventOpts struct {
 type EventRepository interface {
 	Save(ctx context.Context, opts SaveEventOpts) (*Event, error)
 	UpdateStatus(ctx context.Context, id int, status EventStatus) error
-	FindAllByAddress(ctx context.Context, chainID *big.Int, address common.Address) ([]*Event, error)
+	FindAllByAddressAndChainID(
+		ctx context.Context,
+		chainID *big.Int,
+		address common.Address,
+	) ([]*Event, error)
+	FindAllByAddress(
+		ctx context.Context,
+		address common.Address,
+	) ([]*Event, error)
 }

packages/relayer/flags.go

@@ -27,3 +27,5 @@ var (
 )
 
 type HTTPOnly bool
+
+type ProfitableOnly bool

packages/relayer/http/get_events_by_address.go

@@ -1,25 +1,38 @@
 package http
 
 import (
-	"errors"
 	"html"
 	"math/big"
 	"net/http"
 
 	"github.com/cyberhorsey/webutils"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/labstack/echo/v4"
+	"github.com/taikoxyz/taiko-mono/packages/relayer"
 )
 
 func (srv *Server) GetEventsByAddress(c echo.Context) error {
 	chainID, ok := new(big.Int).SetString(c.QueryParam("chainID"), 10)
-	if !ok {
-		return webutils.LogAndRenderErrors(c, http.StatusUnprocessableEntity, errors.New("invalid chain id"))
-	}
 
 	address := html.EscapeString(c.QueryParam("address"))
 
-	events, err := srv.eventRepo.FindAllByAddress(c.Request().Context(), chainID, common.HexToAddress(address))
+	var events []*relayer.Event
+
+	var err error
+
+	if ok {
+		events, err = srv.eventRepo.FindAllByAddressAndChainID(
+			c.Request().Context(),
+			chainID,
+			common.HexToAddress(address),
+		)
+	} else {
+		events, err = srv.eventRepo.FindAllByAddress(
+			c.Request().Context(),
+			common.HexToAddress(address),
+		)
+	}
+
 	if err != nil {
 		return webutils.LogAndRenderErrors(c, http.StatusUnprocessableEntity, err)
 	}

packages/relayer/http/get_events_by_address_test.go

@@ -48,6 +48,14 @@ func Test_GetEventsByAddress(t *testing.T) {
 			[]string{`[{"id":780800018316137516,"name":"name",
 			"data":{"Owner":"0x0000000000000000000000000000000000000123"},"status":0,"chainID":167001}]`},
 		},
+		{
+			"successNoChainID",
+			"0x0000000000000000000000000000000000000123",
+			"",
+			http.StatusOK,
+			[]string{`[{"id":780800018316137516,"name":"name",
+			"data":{"Owner":"0x0000000000000000000000000000000000000123"},"status":0,"chainID":167001}]`},
+		},
 	}
 
 	for _, tt := range tests {

packages/relayer/indexer/filter_then_subscribe.go

@@ -42,11 +42,12 @@ func (svc *Service) FilterThenSubscribe(
 	}
 
 	if svc.processingBlockHeight == header.Number.Uint64() {
-		log.Info("caught up, subscribing to new incoming events")
+		log.Infof("chain ID %v caught up, subscribing to new incoming events", chainID.Uint64())
 		return svc.subscribe(ctx, chainID)
 	}
 
-	log.Infof("getting events between %v and %v in batches of %v",
+	log.Infof("chain ID %v getting events between %v and %v in batches of %v",
+		chainID.Uint64(),
 		svc.processingBlockHeight,
 		header.Number.Int64(),
 		svc.blockBatchSize,
@@ -111,7 +112,10 @@ func (svc *Service) FilterThenSubscribe(
 		}
 	}
 
-	log.Info("indexer fully caught up, checking latest block number to see if it's advanced")
+	log.Infof(
+		"chain id %v indexer fully caught up, checking latest block number to see if it's advanced",
+		chainID.Uint64(),
+	)
 
 	latestBlock, err := svc.ethClient.HeaderByNumber(ctx, nil)
 	if err != nil {

packages/relayer/indexer/service.go

@@ -65,6 +65,7 @@ type NewServiceOpts struct {
 	NumGoroutines       int
 	SubscriptionBackoff time.Duration
 	Confirmations       uint64
+	ProfitableOnly      relayer.ProfitableOnly
 }
 
 func NewService(opts NewServiceOpts) (*Service, error) {
@@ -153,6 +154,7 @@ func NewService(opts NewServiceOpts) (*Service, error) {
 		RelayerAddress:   relayerAddr,
 		Confirmations:    opts.Confirmations,
 		SrcETHClient:     opts.EthClient,
+		ProfitableOnly:   opts.ProfitableOnly,
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "message.NewProcessor")

packages/relayer/message/get_latest_nonce.go

@@ -0,0 +1,23 @@
+package message
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+)
+
+func (p *Processor) getLatestNonce(ctx context.Context, auth *bind.TransactOpts) error {
+	pendingNonce, err := p.destEthClient.PendingNonceAt(ctx, p.relayerAddr)
+	if err != nil {
+		return err
+	}
+
+	if pendingNonce > p.destNonce {
+		p.setLatestNonce(pendingNonce)
+	}
+
+	auth.Nonce = big.NewInt(int64(p.destNonce))
+
+	return nil
+}

packages/relayer/message/get_latest_nonce_test.go

@@ -0,0 +1,19 @@
+package message
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/stretchr/testify/assert"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/mock"
+)
+
+func Test_getLatestNonce(t *testing.T) {
+	p := newTestProcessor(true)
+
+	err := p.getLatestNonce(context.Background(), &bind.TransactOpts{})
+	assert.Nil(t, err)
+
+	assert.Equal(t, p.destNonce, mock.PendingNonce)
+}

packages/relayer/message/is_profitable.go

@@ -0,0 +1,41 @@
+package message
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/pkg/errors"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/contracts"
+)
+
+func (p *Processor) isProfitable(ctx context.Context, message contracts.IBridgeMessage, proof []byte) (bool, error) {
+	processingFee := message.ProcessingFee
+
+	if processingFee == nil || processingFee.Cmp(big.NewInt(0)) != 1 {
+		return false, nil
+	}
+
+	auth, err := bind.NewKeyedTransactorWithChainID(p.ecdsaKey, message.DestChainId)
+	if err != nil {
+		return false, errors.Wrap(err, "bind.NewKeyedTransactorWithChainID")
+	}
+
+	auth.NoSend = true
+
+	auth.Context = ctx
+
+	// estimate gas with auth.NoSend set to true
+	tx, err := p.destBridge.ProcessMessage(auth, message, proof)
+	if err != nil {
+		return false, errors.Wrap(err, "p.destBridge.ProcessMessage")
+	}
+
+	cost := tx.Cost()
+
+	if processingFee.Cmp(cost) != 1 {
+		return false, nil
+	}
+
+	return true, nil
+}

packages/relayer/message/is_profitable_test.go

@@ -0,0 +1,73 @@
+package message
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/contracts"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/mock"
+)
+
+func Test_isProfitable(t *testing.T) {
+	p := newTestProcessor(true)
+
+	tests := []struct {
+		name           string
+		message        contracts.IBridgeMessage
+		proof          []byte
+		wantProfitable bool
+		wantErr        error
+	}{
+		{
+			"zeroProcessingFee",
+			contracts.IBridgeMessage{
+				ProcessingFee: big.NewInt(0),
+			},
+			nil,
+			false,
+			nil,
+		},
+		{
+			"nilProcessingFee",
+			contracts.IBridgeMessage{},
+			nil,
+			false,
+			nil,
+		},
+		{
+			"lowProcessingFee",
+			contracts.IBridgeMessage{
+				ProcessingFee: new(big.Int).Sub(mock.ProcessMessageTx.Cost(), big.NewInt(1)),
+				DestChainId:   big.NewInt(167001),
+			},
+			nil,
+			false,
+			nil,
+		},
+		{
+			"profitableProcessingFee",
+			contracts.IBridgeMessage{
+				ProcessingFee: new(big.Int).Add(mock.ProcessMessageTx.Cost(), big.NewInt(1)),
+				DestChainId:   big.NewInt(167001),
+			},
+			nil,
+			true,
+			nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			profitable, err := p.isProfitable(
+				context.Background(),
+				tt.message,
+				tt.proof,
+			)
+
+			assert.Equal(t, tt.wantProfitable, profitable)
+			assert.Equal(t, tt.wantErr, err)
+		})
+	}
+}