fix: wait for finalization (#455)

bridge/standard/pkg/node/node.go

@@ -6,8 +6,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"math/big"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -20,6 +20,7 @@ import (
 	"github.com/primev/mev-commit/bridge/standard/pkg/store"
 	l1gateway "github.com/primev/mev-commit/contracts-abi/clients/L1Gateway"
 	settlementgateway "github.com/primev/mev-commit/contracts-abi/clients/SettlementGateway"
+	"github.com/primev/mev-commit/x/contracts/ethwrapper"
 	"github.com/primev/mev-commit/x/contracts/events"
 	"github.com/primev/mev-commit/x/contracts/events/publisher"
 	"github.com/primev/mev-commit/x/contracts/transactor"
@@ -243,9 +244,29 @@ func (n *Node) createGatewayContract(
 		return fmt.Errorf("failed to parse contract ABI: %w", err)
 	}
 
+	opts := []ethwrapper.EthClientOptions{
+		ethwrapper.EthClientWithMaxRetries(3),
+	}
+
+	if component == "l1" {
+		opts = append(
+			opts,
+			ethwrapper.EthClientWithBlockNumOverride(finalizedBlockNumGetter),
+		)
+	}
+
+	wrappedClient, err := ethwrapper.NewClient(
+		logger.With("component", fmt.Sprintf("%s/ethwrapper", component)),
+		[]string{rpcURL},
+		opts...,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create wrapped client: %w", err)
+	}
+
 	monitor := txmonitor.New(
 		signer.GetAddress(),
-		client,
+		wrappedClient,
 		txmonitor.NewEVMHelperWithLogger(
 			client,
 			logger.With("component", fmt.Sprintf("%s/evmhelper", component)),
@@ -292,51 +313,23 @@ func (n *Node) createGatewayContract(
 	)
 	n.metrics.MustRegister(evtMgr.Metrics()...)
 
-	parsedURL, err := url.Parse(rpcURL)
-	if err != nil {
-		return fmt.Errorf("failed to parse URL: %w", err)
-	}
-
-	switch parsedURL.Scheme {
-	case "ws", "wss":
-		p := publisher.NewWSPublisher(
-			st,
-			logger,
-			client,
-			evtMgr,
-		)
-		n.startables = append(
-			n.startables,
-			StartableObjWithDesc{
-				Startable: StartableFunc(
-					func(ctx context.Context) <-chan struct{} {
-						return p.Start(ctx, contractAddr)
-					},
-				),
-				Desc: fmt.Sprintf("%s/publisher", component),
-			},
-		)
-	case "http", "https":
-		p := publisher.NewHTTPPublisher(
-			st,
-			logger,
-			client,
-			evtMgr,
-		)
-		n.startables = append(
-			n.startables,
-			StartableObjWithDesc{
-				Startable: StartableFunc(
-					func(ctx context.Context) <-chan struct{} {
-						return p.Start(ctx, contractAddr)
-					},
-				),
-				Desc: fmt.Sprintf("%s/publisher", component),
-			},
-		)
-	default:
-		return fmt.Errorf("unsupported scheme: %s", parsedURL.Scheme)
-	}
+	p := publisher.NewHTTPPublisher(
+		st,
+		logger,
+		wrappedClient,
+		evtMgr,
+	)
+	n.startables = append(
+		n.startables,
+		StartableObjWithDesc{
+			Startable: StartableFunc(
+				func(ctx context.Context) <-chan struct{} {
+					return p.Start(ctx, contractAddr)
+				},
+			),
+			Desc: fmt.Sprintf("%s/publisher", component),
+		},
+	)
 
 	switch component {
 	case "l1":
@@ -399,3 +392,29 @@ func setupMetricsNamespace(namespace string) {
 	events.Namespace = namespace
 	transactor.Namespace = namespace
 }
+
+// BlockResponse is used to get the block number from the eth_getBlockByNumber RPC response.
+type BlockResponse struct {
+	Number string `json:"number"`
+}
+
+// FinalizedBlockNumGetter gets the finalized block number from the Ethereum client.
+func finalizedBlockNumGetter(ctx context.Context, cli ethwrapper.EthClient) (uint64, error) {
+	client, ok := cli.(*ethclient.Client)
+	if !ok {
+		return cli.BlockNumber(ctx)
+	}
+	var blockResponse BlockResponse
+
+	// Call the eth_getBlockByNumber RPC with "finalized"
+	err := client.Client().CallContext(ctx, &blockResponse, "eth_getBlockByNumber", "finalized", false)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get the finalized block: %w", err)
+	}
+
+	// Convert the hex block number to a big.Int
+	blockNumber := new(big.Int)
+	blockNumber.SetString(blockResponse.Number[2:], 16) // Strip "0x" and convert from hex
+
+	return blockNumber.Uint64(), nil
+}

oracle/pkg/node/node.go

@@ -25,6 +25,7 @@ import (
 	"github.com/primev/mev-commit/oracle/pkg/l1Listener"
 	"github.com/primev/mev-commit/oracle/pkg/store"
 	"github.com/primev/mev-commit/oracle/pkg/updater"
+	"github.com/primev/mev-commit/x/contracts/ethwrapper"
 	"github.com/primev/mev-commit/x/contracts/events"
 	"github.com/primev/mev-commit/x/contracts/events/publisher"
 	"github.com/primev/mev-commit/x/contracts/transactor"
@@ -197,19 +198,19 @@ func NewNode(opts *Options) (*Node, error) {
 		TransactOpts: *tOpts,
 	}
 
-	l1ClientOpts := []l1ClientOptions{
-		l1ClientWithMaxRetries(30),
+	l1ClientOpts := []ethwrapper.EthClientOptions{
+		ethwrapper.EthClientWithMaxRetries(30),
 	}
 	if opts.LaggerdMode > 0 {
 		l1ClientOpts = append(
 			l1ClientOpts,
-			l1ClientWithBlockNumberDrift(opts.LaggerdMode),
+			ethwrapper.EthClientWithBlockNumberDrift(opts.LaggerdMode),
 		)
 	}
 	if len(opts.OverrideWinners) > 0 {
 		l1ClientOpts = append(
 			l1ClientOpts,
-			l1ClientWithWinnersOverride(opts.OverrideWinners),
+			ethwrapper.EthClientWithWinnersOverride(opts.OverrideWinners),
 		)
 		for _, winner := range opts.OverrideWinners {
 			nd.logger.Info("setting builder mapping", "builderName", winner, "builderAddress", winner)
@@ -228,7 +229,7 @@ func NewNode(opts *Options) (*Node, error) {
 			}
 		}
 	}
-	l1Client, err := newL1Client(
+	l1Client, err := ethwrapper.NewClient(
 		nd.logger,
 		opts.L1RPCUrls,
 		l1ClientOpts...,
@@ -249,13 +250,20 @@ func NewNode(opts *Options) (*Node, error) {
 	l1LisClosed := l1Lis.Start(ctx)
 	healthChecker.Register(health.CloseChannelHealthCheck("l1_listener", l1LisClosed))
 
+	rawClient := l1Client.RawClient()
+	if rawClient == nil {
+		nd.logger.Error("failed to get raw client")
+		cancel()
+		return nil, errors.New("failed to get ethclient")
+	}
+
 	updtr, err := updater.NewUpdater(
 		nd.logger.With("component", "updater"),
 		l1Client,
 		st,
 		evtMgr,
 		oracleTransactorSession,
-		txmonitor.NewEVMHelperWithLogger(l1Client.clients[0].cli.(*ethclient.Client), nd.logger, contracts),
+		txmonitor.NewEVMHelperWithLogger(rawClient, nd.logger, contracts),
 	)
 	if err != nil {
 		nd.logger.Error("failed to instantiate updater", "error", err)