Node: Update Celo watcher (#4210)

* Node: Update Celo watcher

* Code review rework

.github/workflows/build.yml

@@ -332,7 +332,7 @@ jobs:
           go-version: "1.23.3"
       # The go-ethereum and celo-blockchain packages both implement secp256k1 using the exact same header, but that causes duplicate symbols.
       - name: Run golang tests
-        run: cd node && go test -v -timeout 5m -race -ldflags '-extldflags "-Wl,--allow-multiple-definition" ' ./...
+        run: cd node && go test -v -timeout 5m -race ./...
 
   # Run Rust lints and tests
   rust-lint-and-tests:

Makefile

@@ -38,7 +38,6 @@ node: $(BIN)/guardiand
 .PHONY: $(BIN)/guardiand
 $(BIN)/guardiand: CGO_ENABLED=1
 $(BIN)/guardiand: dirs generate
-	@# The go-ethereum and celo-blockchain packages both implement secp256k1 using the exact same header, but that causes duplicate symbols.
-	cd node && go build -ldflags "-X github.com/certusone/wormhole/node/pkg/version.version=${VERSION} -extldflags -Wl,--allow-multiple-definition" \
+	cd node && go build -ldflags "-X github.com/certusone/wormhole/node/pkg/version.version=${VERSION}" \
 	  -mod=readonly -o ../$(BIN)/guardiand \
 	  github.com/certusone/wormhole/node

node/Dockerfile

@@ -19,7 +19,7 @@ ARG GO_BUILD_ARGS=-race
 RUN --mount=type=cache,target=/root/.cache --mount=type=cache,target=/go \
   export CGO_ENABLED=1 && \
   cd node && \
-  go build ${GO_BUILD_ARGS} -gcflags="all=-N -l" --ldflags '-extldflags "-Wl,--allow-multiple-definition" -X "github.com/certusone/wormhole/node/cmd/guardiand.Build=dev"' -mod=readonly -o /guardiand github.com/certusone/wormhole/node && \
+  go build ${GO_BUILD_ARGS} -gcflags="all=-N -l" --ldflags '-X "github.com/certusone/wormhole/node/cmd/guardiand.Build=dev"' -mod=readonly -o /guardiand github.com/certusone/wormhole/node && \
   go get github.com/CosmWasm/wasmvm@v1.1.1 && \
   cp /go/pkg/mod/github.com/!cosm!wasm/wasmvm@v1.1.1/internal/api/libwasmvm.*.so /usr/lib/
 

node/Makefile

@@ -1,13 +1,3 @@
-.PHONY test
+.PHONY: test
 test: 
-# Use this command on amd64 systems
-	go test -v -ldflags '-extldflags "-Wl,--allow-multiple-definition" ' ./...
-
-.PHONY test-arm64
-test-arm64:
-# Use this command on arm64, otherwise you will encounter linker errors.
-# It's not perfect: it will fail due to 'undefined symbols' errors
-# for packges using cgo. Still, it will get you farther than running
-# the default command.
-# To test a single package, use these -ldflags with e.g. ./pkg/governor
-	go test -ldflags '-extldflags "-Wl,-ld_classic " ' ./...
+	go test -v ./...

node/hack/parse_eth_tx/parse_eth_tx.go

@@ -1,5 +1,5 @@
 // To compile:
-//   go build --ldflags '-extldflags "-Wl,--allow-multiple-definition"' -o parse_eth_tx
+//   go build -o parse_eth_tx
 // Usage:
 //   ./parse_eth_tx -chainID=14 -ethRPC=wss://alfajores-forno.celo-testnet.org/ws -contractAddr=0x88505117CA88e7dd2eC6EA1E13f0948db2D50D56 -tx=0x20a1e7e491dd82b6b33db0820e88a96b58bac28d65770ea73af80e457745aab1
 
@@ -39,16 +39,9 @@ func main() {
 
 	var ethIntf connectors.Connector
 	var err error
-	if chainID == vaa.ChainIDCelo {
-		ethIntf, err = connectors.NewCeloConnector(ctx, "", *flagEthRPC, contractAddr, zap.L())
-		if err != nil {
-			log.Fatalf("dialing eth client failed: %v", err)
-		}
-	} else {
-		ethIntf, err = connectors.NewEthereumBaseConnector(ctx, "", *flagEthRPC, contractAddr, zap.L())
-		if err != nil {
-			log.Fatalf("dialing eth client failed: %v", err)
-		}
+	ethIntf, err = connectors.NewEthereumBaseConnector(ctx, "", *flagEthRPC, contractAddr, zap.L())
+	if err != nil {
+		log.Fatalf("dialing eth client failed: %v", err)
 	}
 
 	transactionHash := ethCommon.HexToHash(*flagTx)

node/hack/query/ccqlistener/Dockerfile

@@ -18,7 +18,7 @@ ARG GO_BUILD_ARGS=-race
 RUN --mount=type=cache,target=/root/.cache --mount=type=cache,target=/go \
   export CGO_ENABLED=1 && \
   cd node/hack/query/ccqlistener && \
-  go build ${GO_BUILD_ARGS} -gcflags="all=-N -l" --ldflags '-extldflags "-Wl,--allow-multiple-definition" -X "github.com/certusone/wormhole/node/cmd/guardiand.Build=dev"' -mod=readonly -o /ccqlistener ccqlistener.go && \
+  go build ${GO_BUILD_ARGS} -gcflags="all=-N -l" --ldflags '-X "github.com/certusone/wormhole/node/cmd/guardiand.Build=dev"' -mod=readonly -o /ccqlistener ccqlistener.go && \
   go get github.com/CosmWasm/wasmvm@v1.1.1 && \
   cp /go/pkg/mod/github.com/!cosm!wasm/wasmvm@v1.1.1/internal/api/libwasmvm.x86_64.so /usr/lib/
 

node/pkg/node/node_test.go

@@ -1146,7 +1146,7 @@ func BenchmarkCrypto(b *testing.B) {
 
 // How to run:
 //
-//	go test -v -ldflags '-extldflags "-Wl,--allow-multiple-definition" ' -bench ^BenchmarkConsensus -benchtime=1x -count 1 -run ^$ > bench.log; tail bench.log
+//	go test -v -bench ^BenchmarkConsensus -benchtime=1x -count 1 -run ^$ > bench.log; tail bench.log
 func BenchmarkConsensus(b *testing.B) {
 	require.Equal(b, b.N, 1)
 	//CONSOLE_LOG_LEVEL = zap.DebugLevel

node/pkg/watchers/evm/config.go

@@ -53,9 +53,7 @@ func (wc *WatcherConfig) Create(
 		setWriteC = setC
 	}
 
-	var devMode bool = (env == common.UnsafeDevNet)
-
-	watcher := NewEthWatcher(wc.Rpc, eth_common.HexToAddress(wc.Contract), string(wc.NetworkID), wc.ChainID, msgC, setWriteC, obsvReqC, queryReqC, queryResponseC, devMode, wc.CcqBackfillCache)
+	watcher := NewEthWatcher(wc.Rpc, eth_common.HexToAddress(wc.Contract), string(wc.NetworkID), wc.ChainID, msgC, setWriteC, obsvReqC, queryReqC, queryResponseC, env, wc.CcqBackfillCache)
 	watcher.SetL1Finalizer(wc.l1Finalizer)
 	return watcher, watcher.Run, nil
 }

node/pkg/watchers/evm/connectors/celo.go

@@ -1,291 +0,0 @@
-package connectors
-
-import (
-	"context"
-	"time"
-
-	celoAbi "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/celoabi"
-	ethAbi "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/ethabi"
-
-	celoBind "github.com/celo-org/celo-blockchain/accounts/abi/bind"
-	celoCommon "github.com/celo-org/celo-blockchain/common"
-	celoTypes "github.com/celo-org/celo-blockchain/core/types"
-	celoClient "github.com/celo-org/celo-blockchain/ethclient"
-	celoRpc "github.com/celo-org/celo-blockchain/rpc"
-
-	ethereum "github.com/ethereum/go-ethereum"
-	ethCommon "github.com/ethereum/go-ethereum/common"
-	ethTypes "github.com/ethereum/go-ethereum/core/types"
-	ethClient "github.com/ethereum/go-ethereum/ethclient"
-	ethEvent "github.com/ethereum/go-ethereum/event"
-	ethRpc "github.com/ethereum/go-ethereum/rpc"
-
-	"github.com/certusone/wormhole/node/pkg/common"
-	"go.uber.org/zap"
-)
-
-// CeloConnector implements EVM network query capabilities for the Celo network. It's almost identical to
-// EthereumConnector except it's using the Celo fork and provides shims between their respective types.
-type CeloConnector struct {
-	networkName string
-	address     ethCommon.Address
-	logger      *zap.Logger
-	client      *celoClient.Client
-	rawClient   *celoRpc.Client
-	filterer    *celoAbi.AbiFilterer
-	caller      *celoAbi.AbiCaller
-}
-
-func NewCeloConnector(ctx context.Context, networkName, rawUrl string, address ethCommon.Address, logger *zap.Logger) (*CeloConnector, error) {
-	rawClient, err := celoRpc.DialContext(ctx, rawUrl)
-	if err != nil {
-		return nil, err
-	}
-	client := celoClient.NewClient(rawClient)
-
-	filterer, err := celoAbi.NewAbiFilterer(celoCommon.BytesToAddress(address.Bytes()), client)
-	if err != nil {
-		panic(err)
-	}
-	caller, err := celoAbi.NewAbiCaller(celoCommon.BytesToAddress(address.Bytes()), client)
-	if err != nil {
-		panic(err)
-	}
-
-	return &CeloConnector{
-		networkName: networkName,
-		address:     address,
-		logger:      logger,
-		client:      client,
-		rawClient:   rawClient,
-		filterer:    filterer,
-		caller:      caller,
-	}, nil
-}
-
-func (c *CeloConnector) NetworkName() string {
-	return c.networkName
-}
-
-func (c *CeloConnector) ContractAddress() ethCommon.Address {
-	return c.address
-}
-
-func (c *CeloConnector) GetCurrentGuardianSetIndex(ctx context.Context) (uint32, error) {
-	opts := &celoBind.CallOpts{Context: ctx}
-	return c.caller.GetCurrentGuardianSetIndex(opts)
-}
-
-func (c *CeloConnector) GetGuardianSet(ctx context.Context, index uint32) (ethAbi.StructsGuardianSet, error) {
-	opts := &celoBind.CallOpts{Context: ctx}
-	celoGs, err := c.caller.GetGuardianSet(opts, index)
-	if err != nil {
-		return ethAbi.StructsGuardianSet{}, err
-	}
-
-	ethKeys := make([]ethCommon.Address, len(celoGs.Keys))
-	for n, k := range celoGs.Keys {
-		ethKeys[n] = ethCommon.BytesToAddress(k.Bytes())
-	}
-
-	return ethAbi.StructsGuardianSet{
-		Keys:           ethKeys,
-		ExpirationTime: celoGs.ExpirationTime,
-	}, err
-}
-
-func (c *CeloConnector) WatchLogMessagePublished(ctx context.Context, errC chan error, sink chan<- *ethAbi.AbiLogMessagePublished) (ethEvent.Subscription, error) {
-	timeout, cancel := context.WithTimeout(ctx, 15*time.Second)
-	defer cancel()
-	messageC := make(chan *celoAbi.AbiLogMessagePublished, 2)
-	messageSub, err := c.filterer.WatchLogMessagePublished(&celoBind.WatchOpts{Context: timeout}, messageC, nil)
-	if err != nil {
-		return messageSub, err
-	}
-
-	// The purpose of this is to map events from the Celo log message channel to the Eth log message channel.
-	common.RunWithScissors(ctx, errC, "celo_connector_watch_log", func(ctx context.Context) error {
-		for {
-			select {
-			// This will return when the subscription is unsubscribed as the error channel gets closed
-			case <-messageSub.Err():
-				return nil
-			case celoEvent := <-messageC:
-				sink <- convertCeloEventToEth(celoEvent)
-			}
-		}
-	})
-
-	return messageSub, err
-}
-
-func (c *CeloConnector) TransactionReceipt(ctx context.Context, txHash ethCommon.Hash) (*ethTypes.Receipt, error) {
-	celoReceipt, err := c.client.TransactionReceipt(ctx, celoCommon.BytesToHash(txHash.Bytes()))
-	if err != nil {
-		return nil, err
-	}
-
-	return convertCeloReceiptToEth(celoReceipt), err
-}
-
-func (c *CeloConnector) TimeOfBlockByHash(ctx context.Context, hash ethCommon.Hash) (uint64, error) {
-	block, err := c.client.HeaderByHash(ctx, celoCommon.BytesToHash(hash.Bytes()))
-	if err != nil {
-		return 0, err
-	}
-
-	return block.Time, err
-}
-
-func (c *CeloConnector) ParseLogMessagePublished(ethLog ethTypes.Log) (*ethAbi.AbiLogMessagePublished, error) {
-	celoEvent, err := c.filterer.ParseLogMessagePublished(*convertCeloLogFromEth(&ethLog))
-	if err != nil {
-		return nil, err
-	}
-
-	return convertCeloEventToEth(celoEvent), err
-}
-
-func (c *CeloConnector) SubscribeForBlocks(ctx context.Context, errC chan error, sink chan<- *NewBlock) (ethereum.Subscription, error) {
-	headSink := make(chan *celoTypes.Header, 2)
-	headerSubscription, err := c.client.SubscribeNewHead(ctx, headSink)
-	if err != nil {
-		return headerSubscription, err
-	}
-
-	// The purpose of this is to map events from the Celo event channel to the new block event channel.
-	common.RunWithScissors(ctx, errC, "celo_connector_subscribe_for_block", func(ctx context.Context) error {
-		for {
-			select {
-			case <-ctx.Done():
-				return nil
-			case ev := <-headSink:
-				if ev == nil {
-					c.logger.Error("new header event is nil")
-					continue
-				}
-				if ev.Number == nil {
-					c.logger.Error("new header block number is nil")
-					continue
-				}
-				hash := ethCommon.BytesToHash(ev.Hash().Bytes())
-				sink <- &NewBlock{
-					Number:   ev.Number,
-					Hash:     hash,
-					Time:     ev.Time,
-					Finality: Finalized,
-				}
-				sink <- &NewBlock{
-					Number:   ev.Number,
-					Hash:     hash,
-					Time:     ev.Time,
-					Finality: Safe,
-				}
-				sink <- &NewBlock{
-					Number:   ev.Number,
-					Hash:     hash,
-					Time:     ev.Time,
-					Finality: Latest,
-				}
-			}
-		}
-	})
-
-	return headerSubscription, err
-}
-
-func (c *CeloConnector) RawCallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
-	return c.rawClient.CallContext(ctx, result, method, args...)
-}
-
-func (c *CeloConnector) RawBatchCallContext(ctx context.Context, b []ethRpc.BatchElem) error {
-	celoB := make([]celoRpc.BatchElem, len(b))
-	for i, v := range b {
-		celoB[i] = celoRpc.BatchElem{
-			Method: v.Method,
-			Args:   v.Args,
-			Result: v.Result,
-			Error:  v.Error,
-		}
-	}
-	return c.rawClient.BatchCallContext(ctx, celoB)
-}
-
-func (c *CeloConnector) Client() *ethClient.Client {
-	panic("unimplemented")
-}
-
-func (c *CeloConnector) SubscribeNewHead(ctx context.Context, ch chan<- *ethTypes.Header) (ethereum.Subscription, error) {
-	panic("unimplemented")
-}
-
-func convertCeloEventToEth(ev *celoAbi.AbiLogMessagePublished) *ethAbi.AbiLogMessagePublished {
-	return &ethAbi.AbiLogMessagePublished{
-		Sender:           ethCommon.BytesToAddress(ev.Sender.Bytes()),
-		Sequence:         ev.Sequence,
-		Nonce:            ev.Nonce,
-		Payload:          ev.Payload,
-		ConsistencyLevel: ev.ConsistencyLevel,
-		Raw:              *convertCeloLogToEth(&ev.Raw),
-	}
-}
-
-func convertCeloLogToEth(l *celoTypes.Log) *ethTypes.Log {
-	topics := make([]ethCommon.Hash, len(l.Topics))
-	for n, t := range l.Topics {
-		topics[n] = ethCommon.BytesToHash(t.Bytes())
-	}
-
-	return &ethTypes.Log{
-		Address:     ethCommon.BytesToAddress(l.Address.Bytes()),
-		Topics:      topics,
-		Data:        l.Data,
-		BlockNumber: l.BlockNumber,
-		TxHash:      ethCommon.BytesToHash(l.TxHash.Bytes()),
-		TxIndex:     l.TxIndex,
-		BlockHash:   ethCommon.BytesToHash(l.BlockHash.Bytes()),
-		Index:       l.Index,
-		Removed:     l.Removed,
-	}
-}
-
-func convertCeloReceiptToEth(celoReceipt *celoTypes.Receipt) *ethTypes.Receipt {
-	ethLogs := make([]*ethTypes.Log, len(celoReceipt.Logs))
-	for n, l := range celoReceipt.Logs {
-		ethLogs[n] = convertCeloLogToEth(l)
-	}
-
-	return &ethTypes.Receipt{
-		Type:              celoReceipt.Type,
-		PostState:         celoReceipt.PostState,
-		Status:            celoReceipt.Status,
-		CumulativeGasUsed: celoReceipt.CumulativeGasUsed,
-		Bloom:             ethTypes.BytesToBloom(celoReceipt.Bloom.Bytes()),
-		Logs:              ethLogs,
-		TxHash:            ethCommon.BytesToHash(celoReceipt.TxHash.Bytes()),
-		ContractAddress:   ethCommon.BytesToAddress(celoReceipt.ContractAddress.Bytes()),
-		GasUsed:           celoReceipt.GasUsed,
-		BlockHash:         ethCommon.BytesToHash(celoReceipt.BlockHash.Bytes()),
-		BlockNumber:       celoReceipt.BlockNumber,
-		TransactionIndex:  celoReceipt.TransactionIndex,
-	}
-}
-
-func convertCeloLogFromEth(l *ethTypes.Log) *celoTypes.Log {
-	topics := make([]celoCommon.Hash, len(l.Topics))
-	for n, t := range l.Topics {
-		topics[n] = celoCommon.BytesToHash(t.Bytes())
-	}
-
-	return &celoTypes.Log{
-		Address:     celoCommon.BytesToAddress(l.Address.Bytes()),
-		Topics:      topics,
-		Data:        l.Data,
-		BlockNumber: l.BlockNumber,
-		TxHash:      celoCommon.BytesToHash(l.TxHash.Bytes()),
-		TxIndex:     l.TxIndex,
-		BlockHash:   celoCommon.BytesToHash(l.BlockHash.Bytes()),
-		Index:       l.Index,
-		Removed:     l.Removed,
-	}
-}

node/pkg/watchers/evm/watcher.go

@@ -120,8 +120,8 @@ type (
 		currentGuardianSet *uint32
 
 		// Interface to the chain specific ethereum library.
-		ethConn       connectors.Connector
-		unsafeDevMode bool
+		ethConn connectors.Connector
+		env     common.Environment
 
 		latestBlockNumber          uint64
 		latestSafeBlockNumber      uint64
@@ -163,7 +163,7 @@ func NewEthWatcher(
 	obsvReqC <-chan *gossipv1.ObservationRequest,
 	queryReqC <-chan *query.PerChainQueryInternal,
 	queryResponseC chan<- *query.PerChainQueryResponseInternal,
-	unsafeDevMode bool,
+	env common.Environment,
 	ccqBackfillCache bool,
 ) *Watcher {
 	return &Watcher{
@@ -178,7 +178,7 @@ func NewEthWatcher(
 		queryReqC:          queryReqC,
 		queryResponseC:     queryResponseC,
 		pending:            map[pendingKey]*pendingMessage{},
-		unsafeDevMode:      unsafeDevMode,
+		env:                env,
 		ccqConfig:          query.GetPerChainConfig(chainID),
 		ccqMaxBlockNumber:  big.NewInt(0).SetUint64(math.MaxUint64),
 		ccqBackfillCache:   ccqBackfillCache,
@@ -197,7 +197,7 @@ func (w *Watcher) Run(parentCtx context.Context) error {
 		zap.String("contract", w.contract.String()),
 		zap.String("networkName", w.networkName),
 		zap.String("chainID", w.chainID.String()),
-		zap.Bool("unsafeDevMode", w.unsafeDevMode),
+		zap.String("env", string(w.env)),
 	)
 
 	// later on we will spawn multiple go-routines through `RunWithScissors`, i.e. catching panics.
@@ -232,15 +232,6 @@ func (w *Watcher) Run(parentCtx context.Context) error {
 			return fmt.Errorf("dialing eth client failed: %w", err)
 		}
 		w.ethConn = connectors.NewBatchPollConnector(ctx, logger, baseConnector, safePollingSupported, 1000*time.Millisecond)
-	} else if w.chainID == vaa.ChainIDCelo {
-		// When we are running in mainnet or testnet, we need to use the Celo ethereum library rather than go-ethereum.
-		// However, in devnet, we currently run the standard ETH node for Celo, so we need to use the standard go-ethereum.
-		w.ethConn, err = connectors.NewCeloConnector(timeout, w.networkName, w.url, w.contract, logger)
-		if err != nil {
-			ethConnectionErrors.WithLabelValues(w.networkName, "dial_error").Inc()
-			p2p.DefaultRegistry.AddErrorCount(w.chainID, 1)
-			return fmt.Errorf("dialing eth client failed: %w", err)
-		}
 	} else {
 		// Everything else is instant finality.
 		logger.Info("assuming instant finality")
@@ -259,7 +250,7 @@ func (w *Watcher) Run(parentCtx context.Context) error {
 	}
 
 	if w.ccqConfig.TimestampCacheSupported {
-		w.ccqTimestampCache = NewBlocksByTimestamp(BTS_MAX_BLOCKS, w.unsafeDevMode)
+		w.ccqTimestampCache = NewBlocksByTimestamp(BTS_MAX_BLOCKS, (w.env == common.UnsafeDevNet))
 	}
 
 	errC := make(chan error)
@@ -706,7 +697,7 @@ func (w *Watcher) getFinality(ctx context.Context) (bool, bool, error) {
 	safe := false
 
 	// Tilt supports polling for both finalized and safe.
-	if w.unsafeDevMode {
+	if w.env == common.UnsafeDevNet {
 		finalized = true
 		safe = true
 
@@ -737,9 +728,11 @@ func (w *Watcher) getFinality(ctx context.Context) (bool, bool, error) {
 		finalized = true
 		safe = true
 
-		// The following chains have their own specialized finalizers.
 	} else if w.chainID == vaa.ChainIDCelo {
-		return false, false, nil
+		// TODO: Celo testnet now supports finalized and safe. As of January 2025, mainnet doesn't yet support safe. Once Celo mainnet cuts over, Celo can
+		// be added to the list above. That change won't be super urgent since we'll just continue to publish safe as finalized, which is not a huge deal.
+		finalized = true
+		safe = w.env != common.MainNet
 
 		// Polygon now supports polling for finalized but not safe.
 		// https://forum.polygon.technology/t/optimizing-decentralized-apps-ux-with-milestones-a-significantly-accelerated-finality-solution/13154