From 15e4480ab0fafdfef020a9aadaef143a883b945d Mon Sep 17 00:00:00 2001 From: lbtsm Date: Tue, 18 Aug 2026 17:14:20 +0800 Subject: [PATCH 1/3] security: read TRON API key from config instead of the source tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The TronGrid API key was hardcoded in Connect, so it shipped in every binary and could not be rotated without a release, and every environment shared one credential. Take it from the chain's `apiKey` opt instead. It stays optional: a self hosted FullNode gRPC endpoint needs no key, so an empty value only warns rather than failing startup. Deployments that talk to a hosted gateway must now set `apiKey` in config — and the leaked key needs rotating regardless. Co-Authored-By: Claude Opus 5 (1M context) --- chains/tron/chain.go | 2 +- chains/tron/config.go | 12 +++++++++--- chains/tron/conn.go | 17 ++++++++++++++--- config.json.example | 15 +++++++++++++++ internal/chain/config.go | 1 + 5 files changed, 40 insertions(+), 7 deletions(-) diff --git a/chains/tron/chain.go b/chains/tron/chain.go index f255e0ea..149fcd0c 100644 --- a/chains/tron/chain.go +++ b/chains/tron/chain.go @@ -48,7 +48,7 @@ func (c *Chain) createChain(chainCfg *core.ChainConfig, logger log15.Logger, sys return nil, err } - conn := NewConnection(config.Endpoint, logger) + conn := NewConnection(config.Endpoint, config.APIKey, logger) err = conn.Connect() if err != nil { return nil, err diff --git a/chains/tron/config.go b/chains/tron/config.go index f23cd151..3192ebb1 100644 --- a/chains/tron/config.go +++ b/chains/tron/config.go @@ -13,9 +13,12 @@ import ( type Config struct { chain.Config LightNode, RentNode, FeeKey, FeeType, EnergySupply string - EthFrom common.Address - McsContract []string - Rent bool + // APIKey authenticates against hosted TRON gateways such as TronGrid. It is + // deployment specific and must come from config, never from the source tree. + APIKey string + EthFrom common.Address + McsContract []string + Rent bool } func parseCfg(chainCfg *core.ChainConfig) (*Config, error) { @@ -38,6 +41,9 @@ func parseCfg(chainCfg *core.ChainConfig) (*Config, error) { } } + if ele, ok := chainCfg.Opts[chain.ApiKey]; ok && ele != "" { + ret.APIKey = ele + } if ele, ok := chainCfg.Opts[chain.RentNode]; ok && ele != "" { ret.RentNode = ele } diff --git a/chains/tron/conn.go b/chains/tron/conn.go index b6ec6ab7..0dadd345 100644 --- a/chains/tron/conn.go +++ b/chains/tron/conn.go @@ -13,19 +13,23 @@ import ( ethcommon "github.com/ethereum/go-ethereum/common" "github.com/lbtsm/gotron-sdk/pkg/client" "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/pkg/errors" ) type Connection struct { - endpoint string + endpoint, apiKey string cli *client.GrpcClient log log15.Logger stop chan int reqTime, cacheBlockNumber int64 } -func NewConnection(endpoint string, log log15.Logger) *Connection { +// NewConnection builds a TRON gRPC connection. apiKey may be empty: a self hosted +// FullNode needs none, a hosted gateway such as TronGrid does. +func NewConnection(endpoint, apiKey string, log log15.Logger) *Connection { return &Connection{ endpoint: endpoint, + apiKey: apiKey, log: log, stop: make(chan int), } @@ -34,7 +38,14 @@ func NewConnection(endpoint string, log log15.Logger) *Connection { // Connect starts the ethereum WS connection func (c *Connection) Connect() error { c.cli = client.NewGrpcClient(c.endpoint) - c.cli.SetAPIKey("ebd4248d-cd13-4f41-9622-acdf84bca55f") + if c.apiKey != "" { + if err := c.cli.SetAPIKey(c.apiKey); err != nil { + return errors.Wrap(err, "set tron api key failed") + } + } else if c.log != nil { + c.log.Warn("Tron connection has no apiKey configured, hosted gateways will reject it", + "endpoint", c.endpoint) + } err := c.cli.Start(grpc.WithInsecure()) if err != nil { return err diff --git a/config.json.example b/config.json.example index 5cf9c0f6..ee7f08d6 100644 --- a/config.json.example +++ b/config.json.example @@ -97,6 +97,21 @@ "syncToMap": "true", "eth2Url": "http://18.138.248.113:9596" } + }, + { + "name": "tron", + "type": "tron", + "id": "728126428", + "endpoint": "grpc.trongrid.io:50051", + "from": "TXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX", + "opts": { + "tronmcs": "0x0000000000000000000000000000000000000000", + "lightnode": "TXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX", + "http": "true", + "eth2Url": "http://18.138.248.113:8545", + "apiKey": "your-trongrid-api-key", + "syncToMap": "true" + } } ] } \ No newline at end of file diff --git a/internal/chain/config.go b/internal/chain/config.go index d4b53d2f..b1f7a657 100644 --- a/internal/chain/config.go +++ b/internal/chain/config.go @@ -50,6 +50,7 @@ var ( Rent = "rent" Addr = "addr" Validate = "validate" + ApiKey = "apiKey" ) // Config encapsulates all necessary parameters in ethereum compatible forms From ada302f32b3b50536ad21fcf24f400b78fcb9110 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Tue, 18 Aug 2026 17:14:47 +0800 Subject: [PATCH 2/3] refactor: own cross-chain runtime state in an injectable Registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MAP client, per-chain contract callers, height providers and the light manager lived in package level vars in internal/mapprotocol, written by five chain constructors and by the expose proof API. Gin serves each proof request on its own goroutine, and chains.Proffer hands every request the same prototype, so those plain maps were written concurrently — a concurrent map write away from taking the process down — and tests could only work by mutating and restoring globals. Introduce mapprotocol.Registry: one owner for that state, every accessor guarded, snapshots handed to callers that iterate. Missing entries now report ErrNotRegistered instead of panicking (Map2OtherHeight was indexed and called in one expression) or silently returning a nil height (the Get2MapHeight/GetNodeTypeByManager function vars defaulted to (nil, nil) until startup swapped them out). The four Init* calls that installed those closures collapse into one LightManager object built from a client and an address, which a small ContractCaller interface makes unit testable. Also fold the five near-identical Proffer.Connect bodies into chain.ProofConnector. Each wrapped its registration in a sync.OnceFunc allocated inside the call, so the Once was fresh every time and deduped nothing: every proof request re-registered shared state and leaked a new connection. The connector caches per chain and endpoint, registers once, and stops swallowing ABI parse errors. Registry is still reached through Default() at the call sites, so the process wide instance is unchanged; what changed is that it is now a value that can be passed in, which is what the next step needs. Co-Authored-By: Claude Opus 5 (1M context) --- chains/bsc/chain.go | 37 +--- chains/eth2/chain.go | 47 ++-- chains/eth2/maintainer.go | 4 +- chains/ethereum/chain.go | 63 +++--- chains/matic/chain.go | 38 +--- chains/sol/chain.go | 2 +- chains/tron/chain.go | 36 +--- cmd/compass/main.go | 16 +- internal/chain/block_2map.go | 2 +- internal/chain/block_map2other.go | 2 +- internal/chain/chain.go | 13 +- internal/chain/common.go | 6 +- internal/chain/maintainer.go | 4 +- internal/chain/mcs.go | 4 +- internal/chain/messenger.go | 2 +- internal/chain/oracle.go | 6 +- internal/chain/proof_connector.go | 116 ++++++++++ internal/chain/proof_connector_test.go | 154 +++++++++++++ internal/mapo/callcontract.go | 2 +- internal/mapprotocol/light_manager.go | 135 ++++++++++++ internal/mapprotocol/mapprotocol.go | 133 +----------- internal/mapprotocol/params.go | 5 - internal/mapprotocol/registry.go | 286 +++++++++++++++++++++++++ internal/mapprotocol/registry_test.go | 169 +++++++++++++++ 24 files changed, 958 insertions(+), 324 deletions(-) create mode 100644 internal/chain/proof_connector.go create mode 100644 internal/chain/proof_connector_test.go create mode 100644 internal/mapprotocol/light_manager.go create mode 100644 internal/mapprotocol/registry.go create mode 100644 internal/mapprotocol/registry_test.go diff --git a/chains/bsc/chain.go b/chains/bsc/chain.go index 5ed4af0b..9e985a95 100644 --- a/chains/bsc/chain.go +++ b/chains/bsc/chain.go @@ -4,13 +4,9 @@ import ( "fmt" "math/big" "strconv" - "sync" - "github.com/ethereum/go-ethereum/common" "github.com/mapprotocol/compass/internal/constant" "github.com/mapprotocol/compass/internal/mapprotocol" - "github.com/mapprotocol/compass/pkg/abi" - "github.com/mapprotocol/compass/pkg/contract" "github.com/mapprotocol/compass/pkg/msg" "github.com/pkg/errors" @@ -26,10 +22,13 @@ import ( ) type Chain struct { + proofConn *chain.ProofConnector } func New() *Chain { - return &Chain{} + return &Chain{ + proofConn: chain.NewProofConnector(mapprotocol.Default(), chain.DefaultProofDialer, nil), + } } func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, role mapprotocol.Role) (core.Chain, error) { @@ -47,7 +46,7 @@ func (c *Chain) syncHeaderToMap(m *chain.Maintainer, latestBlock *big.Int) error return nil } // synced height check - syncedHeight, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(m.Cfg.Id) if err != nil { m.Log.Error("Get current synced Height failed", "err", err) return err @@ -112,29 +111,7 @@ func (c *Chain) assembleProof(m *chain.Messenger, log *types.Log, proofType int6 } func (c *Chain) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { - conn := connection.NewConnection(endpoint, true, nil, nil, big.NewInt(chain.DefaultGasLimit), - big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) - err := conn.Connect() - if err != nil { - return nil, err - } - - fn := sync.OnceFunc(func() { - idInt, _ := strconv.ParseUint(id, 10, 64) - oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) - call := contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi) - mapprotocol.ContractMapping[msg.ChainId(idInt)] = call - - oAbi, _ := abi.New(mapprotocol.SignerJson) - oracleCall := contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, oAbi) - mapprotocol.SingMapping[msg.ChainId(idInt)] = oracleCall - - fn := mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), common.HexToAddress(lightNode), conn.Client()) - mapprotocol.Map2OtherHeight[msg.ChainId(idInt)] = fn - }) - fn() - - return conn.Client(), nil + return c.proofConn.Connect(id, endpoint, mcs, lightNode, oracleNode) } func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, proofType int64, selfId, @@ -187,7 +164,7 @@ func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, } func (c *Chain) Maintainer(client *ethclient.Client, selfId, toChainId uint64, srcEndpoint string) ([]byte, error) { - syncedHeight, err := mapprotocol.Get2MapHeight(msg.ChainId(selfId)) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(msg.ChainId(selfId)) if err != nil { return nil, errors.Wrap(err, "unable to get synced height") } diff --git a/chains/eth2/chain.go b/chains/eth2/chain.go index 096d0af2..46f05327 100644 --- a/chains/eth2/chain.go +++ b/chains/eth2/chain.go @@ -3,8 +3,6 @@ package eth2 import ( "fmt" "math/big" - "strconv" - "sync" "github.com/ChainSafe/log15" "github.com/ethereum/go-ethereum/common" @@ -13,7 +11,6 @@ import ( "github.com/mapprotocol/compass/connections/eth2" "github.com/mapprotocol/compass/core" "github.com/mapprotocol/compass/internal/chain" - "github.com/mapprotocol/compass/internal/constant" ieth "github.com/mapprotocol/compass/internal/eth2" "github.com/mapprotocol/compass/internal/mapprotocol" "github.com/mapprotocol/compass/internal/tx" @@ -33,10 +30,17 @@ type Chain struct { writer *chain.Writer // The writer of the chain listen core.Listener // The listener of this chain stop chan<- int + + proofConn *chain.ProofConnector } func New() *Chain { - return &Chain{} + return &Chain{ + proofConn: chain.NewProofConnector(mapprotocol.Default(), func(endpoint string) core.Connection { + return eth2.NewConnection(endpoint, "", true, nil, nil, big.NewInt(chain.DefaultGasLimit), + big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) + }, nil), + } } func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, role mapprotocol.Role) (core.Chain, error) { @@ -71,6 +75,7 @@ func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan // simplified a little bit var listen core.Listener + registry := mapprotocol.Default() cs := chain.NewCommonSync(conn, cfg, logger, stop, sysErr, bs, chain.OptOfOracleHandler(chain.DefaultOracleHandler)) cs.RegisterState(cfg.Name, string(role)) @@ -82,22 +87,22 @@ func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan return nil, errors.Wrap(err, "eth2 get init headerHeight failed") } logger.Info("map2eth2 Current situation", "height", height, "lightNode", cfg.LightNode) - mapprotocol.SyncOtherMap[cfg.Id] = height - mapprotocol.Map2OtherHeight[cfg.Id] = fn + registry.SetMap2OtherInitHeight(cfg.Id, height) + registry.SetMap2OtherHeight(cfg.Id, fn) listen = NewMaintainer(cs, conn.Eth2Client()) case mapprotocol.RoleOfMessenger: oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) call := contract.New(conn, cfg.McsContract, oracleAbi) - mapprotocol.ContractMapping[cfg.Id] = call + registry.SetMosCall(cfg.Id, call) listen = NewMessenger(cs) case mapprotocol.RoleOfOracle: oAbi, _ := abi.New(mapprotocol.SignerJson) oracleCall := contract.New(conn, []common.Address{cfg.OracleNode}, oAbi) - mapprotocol.SingMapping[cfg.Id] = oracleCall + registry.SetSignerCall(cfg.Id, oracleCall) otherAbi, _ := abi.New(mapprotocol.OtherAbi) call := contract.New(conn, []common.Address{cfg.LightNode}, otherAbi) - mapprotocol.LightNodeMapping[cfg.Id] = call + registry.SetLightNodeCall(cfg.Id, call) listen = chain.NewOracle(cs) } wri := chain.NewWriter(conn, cfg, logger, stop, sysErr) @@ -153,29 +158,7 @@ func (c *Chain) Conn() core.Connection { } func (c *Chain) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { - conn := eth2.NewConnection(endpoint, "", true, nil, nil, big.NewInt(chain.DefaultGasLimit), - big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) - err := conn.Connect() - if err != nil { - return nil, err - } - - fn := sync.OnceFunc(func() { - idInt, _ := strconv.ParseUint(id, 10, 64) - oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) - call := contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi) - mapprotocol.ContractMapping[msg.ChainId(idInt)] = call - - oAbi, _ := abi.New(mapprotocol.SignerJson) - oracleCall := contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, oAbi) - mapprotocol.SingMapping[msg.ChainId(idInt)] = oracleCall - - fn := mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), common.HexToAddress(lightNode), conn.Client()) - mapprotocol.Map2OtherHeight[msg.ChainId(idInt)] = fn - }) - fn() - - return conn.Client(), nil + return c.proofConn.Connect(id, endpoint, mcs, lightNode, oracleNode) } func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, proofType int64, selfId, diff --git a/chains/eth2/maintainer.go b/chains/eth2/maintainer.go index 9c9c5cce..28227591 100644 --- a/chains/eth2/maintainer.go +++ b/chains/eth2/maintainer.go @@ -81,7 +81,7 @@ func (m *Maintainer) sync() error { continue } - startNumber, endNumber, err := mapprotocol.GetEth22MapNumber(m.Cfg.Id) + startNumber, endNumber, err := mapprotocol.Default().Eth2MapNumber(m.Cfg.Id) if err != nil { m.Log.Error("Get startNumber failed", "err", err) time.Sleep(constant.BlockRetryInterval) @@ -153,7 +153,7 @@ func (m *Maintainer) sync() error { } func (m *Maintainer) updateSyncHeight() error { - syncedHeight, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(m.Cfg.Id) if err != nil { m.Log.Error("Get synced Height failed", "err", err) return err diff --git a/chains/ethereum/chain.go b/chains/ethereum/chain.go index 689edfca..93f17687 100644 --- a/chains/ethereum/chain.go +++ b/chains/ethereum/chain.go @@ -7,7 +7,6 @@ import ( "math/big" "strconv" "strings" - "sync" "github.com/ChainSafe/log15" "github.com/ethereum/go-ethereum/common" @@ -20,17 +19,32 @@ import ( "github.com/mapprotocol/compass/internal/mapo" "github.com/mapprotocol/compass/internal/mapprotocol" "github.com/mapprotocol/compass/internal/tx" - "github.com/mapprotocol/compass/pkg/abi" - "github.com/mapprotocol/compass/pkg/contract" "github.com/mapprotocol/compass/pkg/ethclient" "github.com/mapprotocol/compass/pkg/msg" ) type Chain struct { + proofConn *chain.ProofConnector } func New() *Chain { - return &Chain{} + c := &Chain{} + c.proofConn = chain.NewProofConnector(mapprotocol.Default(), chain.DefaultProofDialer, c.registerHeight) + return c +} + +// registerHeight publishes the height providers a proof request needs. The MAP +// chain is the one that answers "how far have you followed chain X", every other +// chain answers "how far have I followed MAP". +func (c *Chain) registerHeight(registry *mapprotocol.Registry, id msg.ChainId, conn core.Connection, lightNode string) error { + if uint64(id) == constant.MapChainId { + registry.SetMapConn(conn.Client()) + registry.SetLightManager(mapprotocol.NewLightManager(conn.Client(), common.HexToAddress(lightNode))) + return nil + } + registry.SetMap2OtherHeight(id, mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), + common.HexToAddress(lightNode), conn.Client())) + return nil } func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, @@ -38,7 +52,7 @@ func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan opts := make([]chain.SyncOpt, 0) opts = append(opts, chain.OptOfInitHeight(mapprotocol.HeaderOneCount)) - if strconv.FormatUint(uint64(chainCfg.Id), 10) == mapprotocol.MapId { + if strconv.FormatUint(uint64(chainCfg.Id), 10) == mapprotocol.Default().MapID() { opts = append(opts, chain.OptOfSync2Map(c.mapToOther)) opts = append(opts, chain.OptOfInitHeight(mapprotocol.EpochOfMap)) } else { @@ -77,14 +91,14 @@ func (c *Chain) mapToOther(m *chain.Maintainer, latestBlock *big.Int) error { msgpayload := []interface{}{input} waitCount := len(m.Cfg.SyncChainIDList) for _, cid := range m.Cfg.SyncChainIDList { - if v, ok := mapprotocol.SyncOtherMap[cid]; ok && latestBlock.Cmp(v) <= 0 { + if v, ok := mapprotocol.Default().Map2OtherInitHeight(cid); ok && latestBlock.Cmp(v) <= 0 { waitCount-- m.Log.Info("map to other current less than synchronized headerHeight", "toChainId", cid, "synced height", v, "current height", latestBlock) continue } // Query the latest height for comparison - if fn, ok := mapprotocol.Map2OtherHeight[cid]; ok { + if fn, ok := mapprotocol.Default().Map2OtherHeight(cid); ok { height, err := fn() if err != nil { return fmt.Errorf("get headerHeight failed, cid(%d),err is %v", cid, err) @@ -96,7 +110,7 @@ func (c *Chain) mapToOther(m *chain.Maintainer, latestBlock *big.Int) error { continue } } - if name, ok := mapprotocol.OnlineChaId[cid]; ok && strings.ToLower(name) == "near" { + if name, ok := mapprotocol.Default().ChainName(cid); ok && strings.ToLower(name) == "near" { param := map[string]interface{}{ "header": mapprotocol.ConvertNearNeedHeader(header), "agg_pk": map[string]interface{}{ @@ -127,7 +141,7 @@ func (c *Chain) mapToOther(m *chain.Maintainer, latestBlock *big.Int) error { } func (c *Chain) headerToMap(m *chain.Maintainer, latestBlock *big.Int) error { - syncedHeight, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(m.Cfg.Id) if err != nil { m.Log.Error("Get synced Height failed", "err", err) return err @@ -219,34 +233,7 @@ func (c *Chain) rlpEthereumHeaders(source, destination msg.ChainId, headers []ty } func (c *Chain) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { - conn := connection.NewConnection(endpoint, true, nil, nil, big.NewInt(chain.DefaultGasLimit), - big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) - err := conn.Connect() - if err != nil { - return nil, err - } - - fn := sync.OnceFunc(func() { - idInt, _ := strconv.ParseUint(id, 10, 64) - oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) - call := contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi) - mapprotocol.ContractMapping[msg.ChainId(idInt)] = call - - oAbi, _ := abi.New(mapprotocol.SignerJson) - oracleCall := contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, oAbi) - mapprotocol.SingMapping[msg.ChainId(idInt)] = oracleCall - - if idInt == constant.MapChainId { - mapprotocol.GlobalMapConn = conn.Client() - mapprotocol.InitOtherChain2MapHeight(common.HexToAddress(lightNode)) - } else { - fn := mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), common.HexToAddress(lightNode), conn.Client()) - mapprotocol.Map2OtherHeight[msg.ChainId(idInt)] = fn - } - }) - fn() - - return conn.Client(), nil + return c.proofConn.Connect(id, endpoint, mcs, lightNode, oracleNode) } func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, proofType int64, selfId, @@ -296,7 +283,7 @@ func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, func (c *Chain) Maintainer(client *ethclient.Client, selfId, toChainId uint64, srcEndpoint string) ([]byte, error) { ret := make([]byte, 0) if selfId == constant.MapChainId { - syncedHeight, err := mapprotocol.Map2OtherHeight[msg.ChainId(toChainId)]() + syncedHeight, err := mapprotocol.Default().Map2OtherSyncedHeight(msg.ChainId(toChainId)) if err != nil { return nil, err } diff --git a/chains/matic/chain.go b/chains/matic/chain.go index 6474488c..6847397d 100644 --- a/chains/matic/chain.go +++ b/chains/matic/chain.go @@ -5,31 +5,29 @@ import ( "fmt" "math/big" "strconv" - "sync" "github.com/ChainSafe/log15" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" connection "github.com/mapprotocol/compass/connections/ethereum" "github.com/mapprotocol/compass/core" "github.com/mapprotocol/compass/internal/chain" - "github.com/mapprotocol/compass/internal/constant" "github.com/mapprotocol/compass/internal/mapprotocol" "github.com/mapprotocol/compass/internal/matic" "github.com/mapprotocol/compass/internal/proof" "github.com/mapprotocol/compass/internal/tx" - "github.com/mapprotocol/compass/pkg/abi" - "github.com/mapprotocol/compass/pkg/contract" "github.com/mapprotocol/compass/pkg/ethclient" "github.com/mapprotocol/compass/pkg/msg" "github.com/pkg/errors" ) type Chain struct { + proofConn *chain.ProofConnector } func New() *Chain { - return &Chain{} + return &Chain{ + proofConn: chain.NewProofConnector(mapprotocol.Default(), chain.DefaultProofDialer, nil), + } } func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, role mapprotocol.Role) (core.Chain, error) { @@ -46,7 +44,7 @@ func (c *Chain) syncHeaderToMap(m *chain.Maintainer, latestBlock *big.Int) error if remainder.Cmp(mapprotocol.Big0) != 0 { return nil } - syncedHeight, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(m.Cfg.Id) if err != nil { m.Log.Error("Get current synced Height failed", "err", err) return err @@ -114,29 +112,7 @@ func (c *Chain) assembleProof(m *chain.Messenger, log *types.Log, proofType int6 } func (c *Chain) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { - conn := connection.NewConnection(endpoint, true, nil, nil, big.NewInt(chain.DefaultGasLimit), - big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) - err := conn.Connect() - if err != nil { - return nil, err - } - - fn := sync.OnceFunc(func() { - idInt, _ := strconv.ParseUint(id, 10, 64) - oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) - call := contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi) - mapprotocol.ContractMapping[msg.ChainId(idInt)] = call - - oAbi, _ := abi.New(mapprotocol.SignerJson) - oracleCall := contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, oAbi) - mapprotocol.SingMapping[msg.ChainId(idInt)] = oracleCall - - fn := mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), common.HexToAddress(lightNode), conn.Client()) - mapprotocol.Map2OtherHeight[msg.ChainId(idInt)] = fn - }) - fn() - - return conn.Client(), nil + return c.proofConn.Connect(id, endpoint, mcs, lightNode, oracleNode) } func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, proofType int64, selfId, @@ -191,7 +167,7 @@ func (c *Chain) Proof(client *ethclient.Client, log *types.Log, endpoint string, } func (c *Chain) Maintainer(client *ethclient.Client, selfId, toChainId uint64, srcEndpoint string) ([]byte, error) { - syncedHeight, err := mapprotocol.Get2MapHeight(msg.ChainId(selfId)) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(msg.ChainId(selfId)) if err != nil { return nil, errors.Wrap(err, "unable to get synced height") } diff --git a/chains/sol/chain.go b/chains/sol/chain.go index fba6d6ad..762e36c0 100644 --- a/chains/sol/chain.go +++ b/chains/sol/chain.go @@ -73,7 +73,7 @@ func createChain(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- case mapprotocol.RoleOfOracle: listen = newSync(cs, oracleHandler, conn, config, conn.cli) } - mapprotocol.MosMapping[config.Id] = config.McsContract[0] + mapprotocol.Default().SetMosAddress(config.Id, config.McsContract[0]) return &Chain{ conn: conn, diff --git a/chains/tron/chain.go b/chains/tron/chain.go index 149fcd0c..69bbebd7 100644 --- a/chains/tron/chain.go +++ b/chains/tron/chain.go @@ -4,7 +4,6 @@ import ( "fmt" "math/big" "strconv" - gosync "sync" "github.com/mapprotocol/compass/internal/mapprotocol" "github.com/mapprotocol/compass/pkg/keystore" @@ -17,7 +16,6 @@ import ( connection "github.com/mapprotocol/compass/connections/ethereum" "github.com/mapprotocol/compass/core" "github.com/mapprotocol/compass/internal/chain" - "github.com/mapprotocol/compass/internal/constant" "github.com/mapprotocol/compass/internal/proof" "github.com/mapprotocol/compass/internal/tx" "github.com/mapprotocol/compass/pkg/abi" @@ -32,10 +30,14 @@ type Chain struct { writer *Writer stop chan<- int listen core.Listener + + proofConn *chain.ProofConnector } func New() *Chain { - return &Chain{} + return &Chain{ + proofConn: chain.NewProofConnector(mapprotocol.Default(), chain.DefaultProofDialer, nil), + } } func (c *Chain) New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, role mapprotocol.Role) (core.Chain, error) { @@ -91,11 +93,11 @@ func (c *Chain) createChain(chainCfg *core.ChainConfig, logger log15.Logger, sys case mapprotocol.RoleOfOracle: oAbi, _ := abi.New(mapprotocol.SignerJson) oracleCall := contract.New(ethConn, []common.Address{config.OracleNode}, oAbi) - mapprotocol.SingMapping[config.Id] = oracleCall + mapprotocol.Default().SetSignerCall(config.Id, oracleCall) otherAbi, _ := abi.New(mapprotocol.OtherAbi) call := contract.New(conn, []common.Address{common.HexToAddress(config.LightNode)}, otherAbi) - mapprotocol.LightNodeMapping[config.Id] = call + mapprotocol.Default().SetLightNodeCall(config.Id, call) handler := oracle if config.Filter { handler = filterOracle @@ -149,29 +151,7 @@ func (c *Chain) Conn() core.Connection { } func (c *Chain) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { - conn := connection.NewConnection(endpoint, true, nil, nil, big.NewInt(chain.DefaultGasLimit), - big.NewInt(chain.DefaultGasPrice), chain.DefaultGasMultiplier) - err := conn.Connect() - if err != nil { - return nil, err - } - - fn := gosync.OnceFunc(func() { - idInt, _ := strconv.ParseUint(id, 10, 64) - oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) - call := contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi) - mapprotocol.ContractMapping[msg.ChainId(idInt)] = call - - oAbi, _ := abi.New(mapprotocol.SignerJson) - oracleCall := contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, oAbi) - mapprotocol.SingMapping[msg.ChainId(idInt)] = oracleCall - - fn := mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), common.HexToAddress(lightNode), conn.Client()) - mapprotocol.Map2OtherHeight[msg.ChainId(idInt)] = fn - }) - fn() - - return conn.Client(), nil + return c.proofConn.Connect(id, endpoint, mcs, lightNode, oracleNode) } func (c *Chain) Proof(client *ethclient.Client, l *types.Log, endpoint string, proofType int64, selfId, diff --git a/cmd/compass/main.go b/cmd/compass/main.go index ac4ce9c4..ff0cd435 100644 --- a/cmd/compass/main.go +++ b/cmd/compass/main.go @@ -209,6 +209,9 @@ func run(ctx *cli.Context, role mapprotocol.Role) error { allChains = append(allChains, cfg.MapChain) allChains = append(allChains, cfg.Chains...) + // registry owns the cross-chain runtime state the chains below register into. + registry := mapprotocol.Default() + for idx, ele := range allChains { ks := ele.KeystorePath if ks == "" { @@ -218,7 +221,7 @@ func run(ctx *cli.Context, role mapprotocol.Role) error { if err != nil { return err } - mapprotocol.MapId = cfg.MapChain.Id + registry.SetMapID(cfg.MapChain.Id) ele.Opts[config.MapChainID] = cfg.MapChain.Id chainConfig := &core.ChainConfig{ Name: ele.Name, @@ -259,20 +262,19 @@ func run(ctx *cli.Context, role mapprotocol.Role) error { } if idx == 0 { - mapprotocol.GlobalMapConn = newChain.(*chain2.Chain).EthClient() + mapClient := newChain.(*chain2.Chain).EthClient() + registry.SetMapConn(mapClient) validateAbi, err := abi.New(mapprotocol.ValidateJson) if err != nil { return err } contract.InitDefaultValidator(contract2.New(newChain.(*chain2.Chain).Conn(), []common.Address{common.HexToAddress(chainConfig.Opts[chain2.Validate])}, validateAbi)) - mapprotocol.Init2GetEth22MapNumber(common.HexToAddress(chainConfig.Opts[chain2.LightNode])) - mapprotocol.InitOtherChain2MapHeight(common.HexToAddress(chainConfig.Opts[chain2.LightNode])) - mapprotocol.InitLightManager(common.HexToAddress(chainConfig.Opts[chain2.LightNode])) - mapprotocol.LightManagerNodeType(common.HexToAddress(chainConfig.Opts[chain2.LightNode])) + registry.SetLightManager(mapprotocol.NewLightManager(mapClient, + common.HexToAddress(chainConfig.Opts[chain2.LightNode]))) } - mapprotocol.OnlineChaId[chainConfig.Id] = chainConfig.Name + registry.SetChainName(chainConfig.Id, chainConfig.Name) c.AddChain(newChain) } c.Start() diff --git a/internal/chain/block_2map.go b/internal/chain/block_2map.go index 10b152e8..069265a7 100644 --- a/internal/chain/block_2map.go +++ b/internal/chain/block_2map.go @@ -45,7 +45,7 @@ func (w *Writer) execToMapMsg(m msg.Message) bool { errorCount++ if errorCount >= 10 { util.Alarm(context.Background(), fmt.Sprintf("%s2map updateHeader failed, err is %s", - mapprotocol.OnlineChaId[m.Source], err.Error())) + mapprotocol.Default().ChainNameOf(m.Source), err.Error())) errorCount = 0 } continue diff --git a/internal/chain/block_map2other.go b/internal/chain/block_map2other.go index 77ea4753..c5d58d00 100644 --- a/internal/chain/block_map2other.go +++ b/internal/chain/block_map2other.go @@ -59,7 +59,7 @@ func (w *Writer) execMap2OtherMsg(m msg.Message) bool { needNonce = w.needNonce(err) errorCount++ if errorCount >= 10 { - util.Alarm(context.Background(), fmt.Sprintf("map2%s updateHeader failed, err is %s", mapprotocol.OnlineChaId[m.Destination], err.Error())) + util.Alarm(context.Background(), fmt.Sprintf("map2%s updateHeader failed, err is %s", mapprotocol.Default().ChainNameOf(m.Destination), err.Error())) errorCount = 0 } time.Sleep(constant.TxRetryInterval) diff --git a/internal/chain/chain.go b/internal/chain/chain.go index 52fc54a4..882ddc89 100644 --- a/internal/chain/chain.go +++ b/internal/chain/chain.go @@ -53,6 +53,7 @@ func New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, r } var listen core.Listener + registry := mapprotocol.Default() cs := NewCommonSync(conn, cfg, logger, stop, sysErr, bs, opts...) cs.RegisterState(cfg.Name, string(role)) switch role { @@ -64,26 +65,26 @@ func New(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, r return nil, errors.Wrap(err, "Map2Other get init headerHeight failed") } logger.Info("Map2other Current situation", "id", cfg.Id, "height", height, "lightNode", cfg.LightNode) - mapprotocol.SyncOtherMap[cfg.Id] = height - mapprotocol.Map2OtherHeight[cfg.Id] = fn + registry.SetMap2OtherInitHeight(cfg.Id, height) + registry.SetMap2OtherHeight(cfg.Id, fn) } listen = NewMaintainer(cs) case mapprotocol.RoleOfMessenger: oracleAbi, _ := abi.New(mapprotocol.OracleAbiJson) call := contract.New(conn, cfg.McsContract, oracleAbi) - mapprotocol.ContractMapping[cfg.Id] = call + registry.SetMosCall(cfg.Id, call) listen = NewMessenger(cs) case mapprotocol.RoleOfOracle: otherAbi, _ := abi.New(mapprotocol.OtherAbi) call := contract.New(conn, []common.Address{cfg.LightNode}, otherAbi) - mapprotocol.LightNodeMapping[cfg.Id] = call + registry.SetLightNodeCall(cfg.Id, call) listen = NewOracle(cs) } oAbi, _ := abi.New(mapprotocol.SignerJson) oracleCall := contract.New(conn, []common.Address{cfg.OracleNode}, oAbi) - mapprotocol.SingMapping[cfg.Id] = oracleCall + registry.SetSignerCall(cfg.Id, oracleCall) wri := NewWriter(conn, cfg, logger, stop, sysErr) - mapprotocol.MosMapping[cfg.Id] = cfg.McsContract[0].String() + registry.SetMosAddress(cfg.Id, cfg.McsContract[0].String()) return &Chain{ cfg: chainCfg, diff --git a/internal/chain/common.go b/internal/chain/common.go index 02b677be..2067b5b7 100644 --- a/internal/chain/common.go +++ b/internal/chain/common.go @@ -26,7 +26,7 @@ type OrderStatusResp struct { } func OrderStatus(idx int, selfChainId, toChainID uint64, blockNumber *big.Int, orderId []byte) (*OrderStatusResp, error) { - call, ok := mapprotocol.ContractMapping[msg.ChainId(toChainID)] + call, ok := mapprotocol.Default().MosCall(msg.ChainId(toChainID)) if !ok { return nil, ContractNotExist } @@ -65,7 +65,7 @@ type MulSignInfoResp struct { } func MulSignInfo(idx int, toChainID uint64) (*MulSignInfoResp, error) { - call, ok := mapprotocol.SingMapping[msg.ChainId(toChainID)] + call, ok := mapprotocol.Default().SignerCall(msg.ChainId(toChainID)) if !ok { return nil, ContractNotExist } @@ -85,7 +85,7 @@ type ProposalInfoResp struct { } func ProposalInfo(idx int, selfChainId, toChainID uint64, blockNumber *big.Int, receipt common.Hash, version [32]byte) (*ProposalInfoResp, error) { - call, ok := mapprotocol.SingMapping[msg.ChainId(toChainID)] + call, ok := mapprotocol.Default().SignerCall(msg.ChainId(toChainID)) if !ok { return nil, ContractNotExist } diff --git a/internal/chain/maintainer.go b/internal/chain/maintainer.go index 1934c252..857cd2ba 100644 --- a/internal/chain/maintainer.go +++ b/internal/chain/maintainer.go @@ -47,7 +47,7 @@ func (m *Maintainer) sync() error { m.Log.Info("Polling Blocks...", "block", currentBlock) if m.Cfg.SyncToMap { - syncedHeight, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + syncedHeight, err := mapprotocol.Default().Chain2MapHeight(m.Cfg.Id) if err != nil { m.Log.Error("Get synced Height failed", "err", err) return err @@ -63,7 +63,7 @@ func (m *Maintainer) sync() error { } } else if m.Cfg.Id == m.Cfg.MapChainID { minHeight := big.NewInt(0) - for cId, height := range mapprotocol.SyncOtherMap { + for cId, height := range mapprotocol.Default().Map2OtherInitHeights() { if minHeight.Uint64() == 0 || minHeight.Cmp(height) == 1 { m.Log.Info("map to other chain find min sync height ", "chainId", cId, "syncedHeight", minHeight, "currentHeight", height) diff --git a/internal/chain/mcs.go b/internal/chain/mcs.go index ff033a0d..09a9b083 100644 --- a/internal/chain/mcs.go +++ b/internal/chain/mcs.go @@ -243,8 +243,8 @@ func (w *Writer) proposal(m msg.Message) bool { } func (w *Writer) mosAlarm(m msg.Message, tx interface{}, err error) { - util.Alarm(context.Background(), fmt.Sprintf("mos %s2%s failed, srcHash=%v err is %s", mapprotocol.OnlineChaId[m.Source], - mapprotocol.OnlineChaId[m.Destination], tx, err.Error())) + util.Alarm(context.Background(), fmt.Sprintf("mos %s2%s failed, srcHash=%v err is %s", mapprotocol.Default().ChainNameOf(m.Source), + mapprotocol.Default().ChainNameOf(m.Destination), tx, err.Error())) } func (w *Writer) call(toAddress *common.Address, input []byte, useAbi abi.ABI, method string) error { diff --git a/internal/chain/messenger.go b/internal/chain/messenger.go index bd99b3eb..312642e3 100644 --- a/internal/chain/messenger.go +++ b/internal/chain/messenger.go @@ -175,7 +175,7 @@ func log2Msg(m *Messenger, log *types.Log, idx int) (int, error) { if m.Cfg.Id == m.Cfg.MapChainID { toChainID = big.NewInt(0).SetBytes(log.Topics[2].Bytes()[8:16]).Uint64() } - chainName, ok := mapprotocol.OnlineChaId[msg.ChainId(toChainID)] + chainName, ok := mapprotocol.Default().ChainName(msg.ChainId(toChainID)) if !ok { m.Log.Info("Map Found a log that is not the current task ", "blockNumber", log.BlockNumber, "toChainID", toChainID) return 0, nil diff --git a/internal/chain/oracle.go b/internal/chain/oracle.go index 60411d20..c8317db2 100644 --- a/internal/chain/oracle.go +++ b/internal/chain/oracle.go @@ -187,7 +187,7 @@ func log2Oracle(m *Oracle, logs []types.Log, blockNumber *big.Int, filterId int6 toChainID := uint64(m.Cfg.MapChainID) if m.Cfg.Id == m.Cfg.MapChainID { toChainID = getToChainId(log.Topics) - if _, ok := mapprotocol.OnlineChaId[msg.ChainId(toChainID)]; !ok { + if _, ok := mapprotocol.Default().ChainName(msg.ChainId(toChainID)); !ok { m.Log.Info("Map Oracle Found a log that is not the current task", "blockNumber", log.BlockNumber, "toChainID", toChainID) continue } @@ -200,7 +200,7 @@ func log2Oracle(m *Oracle, logs []types.Log, blockNumber *big.Int, filterId int6 return errors.Wrap(err, "Get2OtherNodeType failed") } } else { - nodeType, err = mapprotocol.GetNodeTypeByManager(mapprotocol.MethodOfNodeType, big.NewInt(int64(m.Cfg.Id))) + nodeType, err = mapprotocol.Default().LightNodeType(mapprotocol.MethodOfNodeType, big.NewInt(int64(m.Cfg.Id))) if err != nil { return errors.Wrap(err, "GetOther2MapNodeType failed") } @@ -356,7 +356,7 @@ func GetMap2OtherNodeType(idx int, toChainID uint64) (*big.Int, error) { } - call, ok := mapprotocol.LightNodeMapping[msg.ChainId(toChainID)] + call, ok := mapprotocol.Default().LightNodeCall(msg.ChainId(toChainID)) if !ok { return nil, ContractNotExist } diff --git a/internal/chain/proof_connector.go b/internal/chain/proof_connector.go new file mode 100644 index 00000000..c7de382b --- /dev/null +++ b/internal/chain/proof_connector.go @@ -0,0 +1,116 @@ +package chain + +import ( + "math/big" + "strconv" + "sync" + + "github.com/ethereum/go-ethereum/common" + connection "github.com/mapprotocol/compass/connections/ethereum" + "github.com/mapprotocol/compass/core" + "github.com/mapprotocol/compass/internal/constant" + "github.com/mapprotocol/compass/internal/mapprotocol" + "github.com/mapprotocol/compass/pkg/abi" + "github.com/mapprotocol/compass/pkg/contract" + "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/mapprotocol/compass/pkg/msg" + "github.com/pkg/errors" +) + +// Dialer opens a read-only connection to a chain endpoint. Each chain package +// supplies its own, which is the only part of proof connection setup that differs +// between them. +type Dialer func(endpoint string) core.Connection + +// HeightRegistrar records how the MAP chain and a peer chain observe each other's +// height. It runs once per chain, right after the connection is established. +type HeightRegistrar func(registry *mapprotocol.Registry, id msg.ChainId, conn core.Connection, lightNode string) error + +// DefaultProofDialer dials an EVM endpoint the way the proof API needs it: +// over HTTP, without a keypair, with the package gas defaults. +func DefaultProofDialer(endpoint string) core.Connection { + return connection.NewConnection(endpoint, true, nil, nil, big.NewInt(DefaultGasLimit), + big.NewInt(DefaultGasPrice), DefaultGasMultiplier) +} + +// ProofConnector serves the Proffer.Connect half of the expose proof API: it opens +// the connection a proof request needs and registers that chain's contract callers. +// +// It replaces five near-identical Connect bodies, each of which wrapped its +// registration in a sync.OnceFunc that was allocated inside the call and therefore +// never actually deduplicated anything. Every proof request re-ran the +// registration (writing shared state from an HTTP handler goroutine) and leaked a +// fresh connection. Here the connection is cached per chain and endpoint, so +// registration happens once and later requests reuse the client. +type ProofConnector struct { + mu sync.Mutex + conns map[string]core.Connection + + registry *mapprotocol.Registry + dial Dialer + register HeightRegistrar +} + +// NewProofConnector returns a connector that dials with the given dialer and +// publishes what it opens into registry. When register is nil the connector +// installs the standard "ask the peer's light node how far it followed MAP" +// provider. +func NewProofConnector(registry *mapprotocol.Registry, dial Dialer, register HeightRegistrar) *ProofConnector { + return &ProofConnector{ + conns: make(map[string]core.Connection), + registry: registry, + dial: dial, + register: register, + } +} + +// Connect returns a client for the chain, opening and registering it on first use. +func (p *ProofConnector) Connect(id, endpoint, mcs, lightNode, oracleNode string) (*ethclient.Client, error) { + chainID, err := strconv.ParseUint(id, 10, 64) + if err != nil { + return nil, errors.Wrapf(err, "proof connect: invalid chain id %q", id) + } + + key := id + "|" + endpoint + p.mu.Lock() + defer p.mu.Unlock() + if conn, ok := p.conns[key]; ok { + return conn.Client(), nil + } + + conn := p.dial(endpoint) + if err := conn.Connect(); err != nil { + return nil, err + } + + if err := p.registerLocked(msg.ChainId(chainID), conn, mcs, lightNode, oracleNode); err != nil { + conn.Close() + return nil, err + } + + p.conns[key] = conn + return conn.Client(), nil +} + +// registerLocked publishes the chain's contract callers and height provider. +func (p *ProofConnector) registerLocked(id msg.ChainId, conn core.Connection, mcs, lightNode, oracleNode string) error { + oracleAbi, err := abi.New(mapprotocol.OracleAbiJson) + if err != nil { + return errors.Wrap(err, "proof connect: parse mos abi") + } + signerAbi, err := abi.New(mapprotocol.SignerJson) + if err != nil { + return errors.Wrap(err, "proof connect: parse signer abi") + } + + registry := p.registry + registry.SetMosCall(id, contract.New(conn, []common.Address{common.HexToAddress(mcs)}, oracleAbi)) + registry.SetSignerCall(id, contract.New(conn, []common.Address{common.HexToAddress(oracleNode)}, signerAbi)) + + if p.register != nil { + return p.register(registry, id, conn, lightNode) + } + registry.SetMap2OtherHeight(id, mapprotocol.Map2EthHeight(constant.ZeroAddress.Hex(), + common.HexToAddress(lightNode), conn.Client())) + return nil +} diff --git a/internal/chain/proof_connector_test.go b/internal/chain/proof_connector_test.go new file mode 100644 index 00000000..7e4f236c --- /dev/null +++ b/internal/chain/proof_connector_test.go @@ -0,0 +1,154 @@ +package chain + +import ( + "math/big" + "sync" + "sync/atomic" + "testing" + + "github.com/ethereum/go-ethereum/accounts/abi/bind" + ethkeystore "github.com/ethereum/go-ethereum/accounts/keystore" + "github.com/ethereum/go-ethereum/common" + "github.com/mapprotocol/compass/core" + "github.com/mapprotocol/compass/internal/mapprotocol" + "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/mapprotocol/compass/pkg/msg" + "github.com/pkg/errors" +) + +// fakeConn is a core.Connection that records whether it was opened and closed. Only +// Connect/Client/Close matter to the proof connector. +type fakeConn struct { + connected atomic.Int32 + closed atomic.Int32 +} + +func (f *fakeConn) Connect() error { f.connected.Add(1); return nil } +func (f *fakeConn) Keypair() *ethkeystore.Key { return nil } +func (f *fakeConn) Opts() *bind.TransactOpts { return nil } +func (f *fakeConn) CallOpts() *bind.CallOpts { return nil } +func (f *fakeConn) LockAndUpdateOpts(bool) error { return nil } +func (f *fakeConn) UnlockOpts() {} +func (f *fakeConn) Client() *ethclient.Client { return nil } +func (f *fakeConn) EnsureHasBytecode(common.Address) error { return nil } +func (f *fakeConn) LatestBlock() (*big.Int, error) { return big.NewInt(0), nil } +func (f *fakeConn) WaitForBlock(*big.Int, *big.Int) error { return nil } +func (f *fakeConn) Close() { f.closed.Add(1) } + +func TestProofConnectorReusesConnection(t *testing.T) { + var dialed atomic.Int32 + conn := &fakeConn{} + registry := mapprotocol.NewRegistry() + + pc := NewProofConnector(registry, func(string) core.Connection { + dialed.Add(1) + return conn + }, nil) + + for i := 0; i < 3; i++ { + if _, err := pc.Connect("97", "http://node", "0x1", "0x2", "0x3"); err != nil { + t.Fatalf("Connect #%d: %v", i, err) + } + } + + // Before this change every proof request dialed again and re-ran registration. + if got := dialed.Load(); got != 1 { + t.Fatalf("dialed %d times, want 1", got) + } + if got := conn.connected.Load(); got != 1 { + t.Fatalf("Connect called %d times on the connection, want 1", got) + } + if _, ok := registry.MosCall(97); !ok { + t.Fatal("mos caller was not registered") + } + if _, ok := registry.SignerCall(97); !ok { + t.Fatal("signer caller was not registered") + } + if _, ok := registry.Map2OtherHeight(97); !ok { + t.Fatal("map2other height provider was not registered") + } +} + +func TestProofConnectorSeparatesEndpoints(t *testing.T) { + var dialed atomic.Int32 + registry := mapprotocol.NewRegistry() + pc := NewProofConnector(registry, func(string) core.Connection { + dialed.Add(1) + return &fakeConn{} + }, nil) + + if _, err := pc.Connect("97", "http://a", "0x1", "0x2", "0x3"); err != nil { + t.Fatalf("Connect a: %v", err) + } + if _, err := pc.Connect("97", "http://b", "0x1", "0x2", "0x3"); err != nil { + t.Fatalf("Connect b: %v", err) + } + if got := dialed.Load(); got != 2 { + t.Fatalf("dialed %d times, want 2: a changed endpoint is a different connection", got) + } +} + +func TestProofConnectorRejectsBadChainID(t *testing.T) { + var dialed atomic.Int32 + pc := NewProofConnector(mapprotocol.NewRegistry(), func(string) core.Connection { + dialed.Add(1) + return &fakeConn{} + }, nil) + + if _, err := pc.Connect("not-a-number", "http://node", "0x1", "0x2", "0x3"); err == nil { + t.Fatal("Connect with a non-numeric chain id returned no error") + } + if got := dialed.Load(); got != 0 { + t.Fatalf("dialed %d times for an invalid chain id, want 0", got) + } +} + +func TestProofConnectorClosesConnectionWhenRegistrationFails(t *testing.T) { + conn := &fakeConn{} + boom := errors.New("boom") + pc := NewProofConnector(mapprotocol.NewRegistry(), func(string) core.Connection { return conn }, + func(*mapprotocol.Registry, msg.ChainId, core.Connection, string) error { return boom }) + + _, err := pc.Connect("97", "http://node", "0x1", "0x2", "0x3") + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } + if got := conn.closed.Load(); got != 1 { + t.Fatalf("connection closed %d times after a failed registration, want 1", got) + } + + // A failed attempt must not be cached, so the next request retries. + if _, err := pc.Connect("97", "http://node", "0x1", "0x2", "0x3"); !errors.Is(err, boom) { + t.Fatalf("second attempt err = %v, want boom", err) + } + if got := conn.connected.Load(); got != 2 { + t.Fatalf("Connect called %d times, want 2: the failed attempt must not be cached", got) + } +} + +// TestProofConnectorConcurrentConnect mirrors what the expose proof API does: many +// HTTP handler goroutines asking for the same chain at once. Run with -race. +func TestProofConnectorConcurrentConnect(t *testing.T) { + var dialed atomic.Int32 + conn := &fakeConn{} + pc := NewProofConnector(mapprotocol.NewRegistry(), func(string) core.Connection { + dialed.Add(1) + return conn + }, nil) + + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := pc.Connect("97", "http://node", "0x1", "0x2", "0x3"); err != nil { + t.Errorf("Connect: %v", err) + } + }() + } + wg.Wait() + + if got := dialed.Load(); got != 1 { + t.Fatalf("dialed %d times under concurrency, want 1", got) + } +} diff --git a/internal/mapo/callcontract.go b/internal/mapo/callcontract.go index 2a86dd19..2e014d20 100644 --- a/internal/mapo/callcontract.go +++ b/internal/mapo/callcontract.go @@ -164,7 +164,7 @@ func AssembleMapProof(cli *ethclient.Client, log *types.Log, receipts []*types.R ek := util.Key2Hex(key, len(prf)) var payloads []byte - name, _ := mapprotocol.OnlineChaId[msg.ChainId(uToChainID)] + name := mapprotocol.Default().ChainNameOf(msg.ChainId(uToChainID)) switch name { default: istanbulExtra := mapprotocol.ConvertIstanbulExtra(ist) diff --git a/internal/mapprotocol/light_manager.go b/internal/mapprotocol/light_manager.go new file mode 100644 index 00000000..831e5b24 --- /dev/null +++ b/internal/mapprotocol/light_manager.go @@ -0,0 +1,135 @@ +// Copyright 2021 Compass Systems +// SPDX-License-Identifier: LGPL-3.0-only + +package mapprotocol + +import ( + "context" + "math/big" + + goeth "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/mapprotocol/compass/internal/constant" + "github.com/mapprotocol/compass/pkg/msg" + "github.com/pkg/errors" +) + +// ContractCaller is the only thing the light manager needs from a chain client. +// Keeping it this small is what lets the reads below be unit tested without an RPC +// endpoint. +type ContractCaller interface { + CallContract(ctx context.Context, call goeth.CallMsg, blockNumber *big.Int) ([]byte, error) +} + +// LightManager reads the MAP light-client manager contract. +// +// It replaces four package level function variables that were each installed by an +// Init* call at startup: the connection and the contract address are now fields of +// one object instead of values captured in closures that other goroutines could +// swap out mid-flight. +type LightManager struct { + caller ContractCaller + addr common.Address +} + +// NewLightManager binds a light manager reader to a client and contract address. +func NewLightManager(caller ContractCaller, addr common.Address) *LightManager { + return &LightManager{caller: caller, addr: addr} +} + +// Data performs a raw read and returns the manager's bytes output. +func (l *LightManager) Data(method string, params ...interface{}) ([]byte, error) { + output, err := l.call(method, params...) + if err != nil { + return nil, err + } + + outputs := LightManger.Methods[method].Outputs + unpack, err := outputs.Unpack(output) + if err != nil { + return nil, err + } + ret := make([]byte, 0) + if err = outputs.Copy(&ret, unpack); err != nil { + return nil, err + } + return ret, nil +} + +// NodeType returns the light node type the manager holds for a chain. +func (l *LightManager) NodeType(method string, params ...interface{}) (*big.Int, error) { + output, err := l.call(method, params...) + if err != nil { + return nil, err + } + + outputs := LightManger.Methods[method].Outputs + unpack, err := outputs.Unpack(output) + if err != nil { + return nil, err + } + ret := new(big.Int) + if err = outputs.Copy(&ret, unpack); err != nil { + return nil, err + } + return ret, nil +} + +// HeaderHeight reports how far the MAP chain has followed the given chain. +func (l *LightManager) HeaderHeight(chainID msg.ChainId) (*big.Int, error) { + output, err := l.call(MethodOfHeaderHeight, big.NewInt(int64(chainID))) + if err != nil { + return nil, errors.Wrap(err, "get other2map headerHeight by lightManager failed") + } + return UnpackHeaderHeightOutput(output) +} + +// Eth2MapNumber reports the block range the MAP chain expects next from an eth2 +// style chain. +func (l *LightManager) Eth2MapNumber(chainID msg.ChainId) (*big.Int, *big.Int, error) { + output, err := l.call(MethodClientState, big.NewInt(int64(chainID))) + if err != nil { + return nil, nil, err + } + + outputs := LightManger.Methods[MethodClientState].Outputs + unpack, err := outputs.Unpack(output) + if err != nil { + return nil, nil, err + } + + back := make([]byte, 0) + if err = outputs.Copy(&back, unpack); err != nil { + return nil, nil, err + } + + ret := struct { + StartNumber *big.Int + EndNumber *big.Int + }{} + analysis, err := Eth2.Methods[MethodClientStateAnalysis].Outputs.Unpack(back) + if err != nil { + return nil, nil, errors.Wrap(err, "analysis") + } + if err = Eth2.Methods[MethodClientStateAnalysis].Outputs.Copy(&ret, analysis); err != nil { + return nil, nil, errors.Wrap(err, "analysis copy") + } + + return ret.StartNumber, ret.EndNumber, nil +} + +// call packs the arguments for a light manager method and performs the read. +func (l *LightManager) call(method string, params ...interface{}) ([]byte, error) { + if l.caller == nil { + return nil, errors.New("map light manager has no client") + } + input, err := PackInput(LightManger, method, params...) + if err != nil { + return nil, errors.Wrapf(err, "pack lightManager %s input failed", method) + } + return l.caller.CallContract( + context.Background(), + goeth.CallMsg{From: constant.ZeroAddress, To: &l.addr, Data: input}, + nil, + ) +} diff --git a/internal/mapprotocol/mapprotocol.go b/internal/mapprotocol/mapprotocol.go index ee9e7042..1b9cb7e0 100644 --- a/internal/mapprotocol/mapprotocol.go +++ b/internal/mapprotocol/mapprotocol.go @@ -8,8 +8,6 @@ import ( "encoding/json" "fmt" "github.com/mapprotocol/compass/internal/constant" - "github.com/mapprotocol/compass/pkg/contract" - "github.com/mapprotocol/compass/pkg/msg" nearclient "github.com/mapprotocol/near-api-go/pkg/client" "math/big" @@ -26,104 +24,6 @@ import ( type GetHeight func() (*big.Int, error) type GetVerifyRange func() (*big.Int, *big.Int, error) -var ( - MapId string - GlobalMapConn *ethclient.Client - SyncOtherMap = make(map[msg.ChainId]*big.Int) // map to other chain init height - Map2OtherHeight = make(map[msg.ChainId]GetHeight) // get map to other height function collect - ContractMapping = make(map[msg.ChainId]*contract.Call) - LightNodeMapping = make(map[msg.ChainId]*contract.Call) - SingMapping = make(map[msg.ChainId]*contract.Call) - MosMapping = make(map[msg.ChainId]string) - Get2MapHeight = func(chainId msg.ChainId) (*big.Int, error) { return nil, nil } // get other chain to map height - GetEth22MapNumber = func(chainId msg.ChainId) (*big.Int, *big.Int, error) { return nil, nil, nil } // can reform, return data is []byte - GetDataByManager = func(string, ...interface{}) ([]byte, error) { return nil, nil } - GetNodeTypeByManager = func(string, ...interface{}) (*big.Int, error) { return nil, nil } -) - -func InitLightManager(lightNode common.Address) { - GetDataByManager = func(method string, params ...interface{}) ([]byte, error) { - input, err := PackInput(LightManger, method, params...) - if err != nil { - return nil, errors.Wrap(err, "get other2map packInput failed") - } - output, err := GlobalMapConn.CallContract( - context.Background(), - goeth.CallMsg{From: constant.ZeroAddress, To: &lightNode, Data: input}, - nil, - ) - if err != nil { - return nil, err - } - outputs := LightManger.Methods[method].Outputs - unpack, err := outputs.Unpack(output) - if err != nil { - return nil, err - } - ret := make([]byte, 0) - if err = outputs.Copy(&ret, unpack); err != nil { - return nil, err - } - - return ret, nil - } -} - -func Init2GetEth22MapNumber(lightNode common.Address) { - GetEth22MapNumber = func(chainId msg.ChainId) (*big.Int, *big.Int, error) { - input, err := PackInput(LightManger, MethodClientState, big.NewInt(int64(chainId))) - if err != nil { - return nil, nil, errors.Wrap(err, "get eth22map packInput failed") - } - - output, err := GlobalMapConn.CallContract(context.Background(), - goeth.CallMsg{From: constant.ZeroAddress, To: &lightNode, Data: input}, nil) - if err != nil { - return nil, nil, err - } - - outputs := LightManger.Methods[MethodClientState].Outputs - unpack, err := outputs.Unpack(output) - if err != nil { - return nil, nil, err - } - - back := make([]byte, 0) - if err = outputs.Copy(&back, unpack); err != nil { - return nil, nil, err - } - - ret := struct { - StartNumber *big.Int - EndNumber *big.Int - }{} - analysis, err := Eth2.Methods[MethodClientStateAnalysis].Outputs.Unpack(back) - if err != nil { - return nil, nil, errors.Wrap(err, "analysis") - } - if err = Eth2.Methods[MethodClientStateAnalysis].Outputs.Copy(&ret, analysis); err != nil { - return nil, nil, errors.Wrap(err, "analysis copy") - } - - return ret.StartNumber, ret.EndNumber, nil - } -} - -func InitOtherChain2MapHeight(lightManager common.Address) { - Get2MapHeight = func(chainId msg.ChainId) (*big.Int, error) { - input, err := PackInput(LightManger, MethodOfHeaderHeight, big.NewInt(int64(chainId))) - if err != nil { - return nil, errors.Wrap(err, "get other2map by manager packInput failed") - } - - height, err := HeaderHeight(lightManager, input) - if err != nil { - return nil, errors.Wrap(err, "get other2map headerHeight by lightManager failed") - } - return height, nil - } -} - func Map2EthHeight(fromUser string, lightNode common.Address, client *ethclient.Client) GetHeight { return func() (*big.Int, error) { from := common.HexToAddress(fromUser) @@ -192,8 +92,9 @@ func UnpackHeaderHeightOutput(output []byte) (*big.Int, error) { return height, nil } -func HeaderHeight(to common.Address, input []byte) (*big.Int, error) { - output, err := GlobalMapConn.CallContract(context.Background(), goeth.CallMsg{From: constant.ZeroAddress, To: &to, Data: input}, nil) +// HeaderHeight reads a headerHeight style getter on the MAP chain. +func HeaderHeight(caller ContractCaller, to common.Address, input []byte) (*big.Int, error) { + output, err := caller.CallContract(context.Background(), goeth.CallMsg{From: constant.ZeroAddress, To: &to, Data: input}, nil) if err != nil { return nil, err } @@ -203,31 +104,3 @@ func HeaderHeight(to common.Address, input []byte) (*big.Int, error) { } return height, nil } - -func LightManagerNodeType(lightNode common.Address) { - GetNodeTypeByManager = func(method string, params ...interface{}) (*big.Int, error) { - input, err := PackInput(LightManger, method, params...) - if err != nil { - return nil, errors.Wrap(err, "get other2map packInput failed") - } - output, err := GlobalMapConn.CallContract( - context.Background(), - goeth.CallMsg{From: constant.ZeroAddress, To: &lightNode, Data: input}, - nil, - ) - if err != nil { - return nil, err - } - outputs := LightManger.Methods[method].Outputs - unpack, err := outputs.Unpack(output) - if err != nil { - return nil, err - } - ret := new(big.Int) - if err = outputs.Copy(&ret, unpack); err != nil { - return nil, err - } - - return ret, nil - } -} diff --git a/internal/mapprotocol/params.go b/internal/mapprotocol/params.go index 07a148bf..eb57c815 100644 --- a/internal/mapprotocol/params.go +++ b/internal/mapprotocol/params.go @@ -1,7 +1,6 @@ package mapprotocol import ( - "github.com/mapprotocol/compass/pkg/msg" "math/big" "strings" @@ -96,10 +95,6 @@ var ( RoleOfOracle Role = "oracle" ) -var ( - OnlineChaId = map[msg.ChainId]string{} -) - var ( ConfirmsOfMatic = big.NewInt(10) HeaderLengthOfEth2 = 20 diff --git a/internal/mapprotocol/registry.go b/internal/mapprotocol/registry.go new file mode 100644 index 00000000..0604d22f --- /dev/null +++ b/internal/mapprotocol/registry.go @@ -0,0 +1,286 @@ +// Copyright 2021 Compass Systems +// SPDX-License-Identifier: LGPL-3.0-only + +package mapprotocol + +import ( + "math/big" + "sync" + + "github.com/mapprotocol/compass/pkg/contract" + "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/mapprotocol/compass/pkg/msg" + "github.com/pkg/errors" +) + +// ErrNotRegistered is returned when a chain was never registered for the role the +// caller needs. It replaces the old "missing map key" behaviour, which either +// panicked or silently produced a nil height. +var ErrNotRegistered = errors.New("chain is not registered") + +// Registry owns the cross-chain runtime state that a relay process discovers while +// it assembles chains: the MAP connection, the per-chain contract callers, the +// height providers and the light-client manager. +// +// It replaces the package level maps this state used to live in. Every accessor +// takes the lock, because the state is written during chain construction and by +// the expose proof API (one goroutine per HTTP request) while the +// maintainer/messenger/oracle loops read it. +type Registry struct { + mu sync.RWMutex + mapID string + mapConn *ethclient.Client + chainNames map[msg.ChainId]string + map2OtherInit map[msg.ChainId]*big.Int + map2OtherHeight map[msg.ChainId]GetHeight + mosCall map[msg.ChainId]*contract.Call + lightNodeCall map[msg.ChainId]*contract.Call + signerCall map[msg.ChainId]*contract.Call + mosAddress map[msg.ChainId]string + lightManager *LightManager +} + +// NewRegistry returns an empty registry. Tests build their own instead of reaching +// for Default, which is what keeps them isolated from each other. +func NewRegistry() *Registry { + return &Registry{ + chainNames: make(map[msg.ChainId]string), + map2OtherInit: make(map[msg.ChainId]*big.Int), + map2OtherHeight: make(map[msg.ChainId]GetHeight), + mosCall: make(map[msg.ChainId]*contract.Call), + lightNodeCall: make(map[msg.ChainId]*contract.Call), + signerCall: make(map[msg.ChainId]*contract.Call), + mosAddress: make(map[msg.ChainId]string), + } +} + +// defaultRegistry is the process wide registry the composition root fills in. +// It stays a singleton for now so this change keeps the current call graph; the +// point of routing everything through a *Registry is that the next step can pass +// one in explicitly. +var defaultRegistry = NewRegistry() + +// Default returns the process wide registry. +func Default() *Registry { + return defaultRegistry +} + +// SetMapID records the configured MAP relay chain id. +func (r *Registry) SetMapID(id string) { + r.mu.Lock() + defer r.mu.Unlock() + r.mapID = id +} + +// MapID returns the configured MAP relay chain id. +func (r *Registry) MapID() string { + r.mu.RLock() + defer r.mu.RUnlock() + return r.mapID +} + +// SetMapConn records the client used for every MAP chain contract read. +func (r *Registry) SetMapConn(client *ethclient.Client) { + r.mu.Lock() + defer r.mu.Unlock() + r.mapConn = client +} + +// MapConn returns the MAP chain client, or nil when no MAP chain was built yet. +func (r *Registry) MapConn() *ethclient.Client { + r.mu.RLock() + defer r.mu.RUnlock() + return r.mapConn +} + +// SetChainName records the human readable name of an online chain. +func (r *Registry) SetChainName(id msg.ChainId, name string) { + r.mu.Lock() + defer r.mu.Unlock() + r.chainNames[id] = name +} + +// ChainName returns the name of an online chain. +func (r *Registry) ChainName(id msg.ChainId) (string, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + name, ok := r.chainNames[id] + return name, ok +} + +// ChainNameOf returns the name of an online chain, or an empty string when the +// chain is unknown. Alarm and log messages use it, where a missing name is not an +// error worth handling. +func (r *Registry) ChainNameOf(id msg.ChainId) string { + name, _ := r.ChainName(id) + return name +} + +// SetMosCall registers the MOS contract caller of a chain (messenger role). +func (r *Registry) SetMosCall(id msg.ChainId, call *contract.Call) { + r.mu.Lock() + defer r.mu.Unlock() + r.mosCall[id] = call +} + +// MosCall returns the MOS contract caller of a chain. +func (r *Registry) MosCall(id msg.ChainId) (*contract.Call, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + call, ok := r.mosCall[id] + return call, ok +} + +// SetLightNodeCall registers the light node contract caller of a chain (oracle role). +func (r *Registry) SetLightNodeCall(id msg.ChainId, call *contract.Call) { + r.mu.Lock() + defer r.mu.Unlock() + r.lightNodeCall[id] = call +} + +// LightNodeCall returns the light node contract caller of a chain. +func (r *Registry) LightNodeCall(id msg.ChainId) (*contract.Call, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + call, ok := r.lightNodeCall[id] + return call, ok +} + +// SetSignerCall registers the oracle signer contract caller of a chain. +func (r *Registry) SetSignerCall(id msg.ChainId, call *contract.Call) { + r.mu.Lock() + defer r.mu.Unlock() + r.signerCall[id] = call +} + +// SignerCall returns the oracle signer contract caller of a chain. +func (r *Registry) SignerCall(id msg.ChainId) (*contract.Call, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + call, ok := r.signerCall[id] + return call, ok +} + +// SetMosAddress records the MOS contract address of a chain. +func (r *Registry) SetMosAddress(id msg.ChainId, addr string) { + r.mu.Lock() + defer r.mu.Unlock() + r.mosAddress[id] = addr +} + +// MosAddress returns the MOS contract address of a chain. +func (r *Registry) MosAddress(id msg.ChainId) (string, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + addr, ok := r.mosAddress[id] + return addr, ok +} + +// SetMap2OtherInitHeight records the height a chain had synced from MAP when the +// maintainer for that chain started. +func (r *Registry) SetMap2OtherInitHeight(id msg.ChainId, height *big.Int) { + r.mu.Lock() + defer r.mu.Unlock() + r.map2OtherInit[id] = height +} + +// Map2OtherInitHeights returns a snapshot of the recorded start heights. Callers +// iterate the copy, so a chain registering itself concurrently cannot corrupt it. +func (r *Registry) Map2OtherInitHeights() map[msg.ChainId]*big.Int { + r.mu.RLock() + defer r.mu.RUnlock() + out := make(map[msg.ChainId]*big.Int, len(r.map2OtherInit)) + for id, height := range r.map2OtherInit { + out[id] = height + } + return out +} + +// Map2OtherInitHeight returns the recorded start height of a chain. +func (r *Registry) Map2OtherInitHeight(id msg.ChainId) (*big.Int, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + height, ok := r.map2OtherInit[id] + return height, ok +} + +// SetMap2OtherHeight registers the provider that reports how far a chain's light +// node has followed the MAP chain. +func (r *Registry) SetMap2OtherHeight(id msg.ChainId, fn GetHeight) { + r.mu.Lock() + defer r.mu.Unlock() + r.map2OtherHeight[id] = fn +} + +// Map2OtherHeight returns the height provider registered for a chain. +func (r *Registry) Map2OtherHeight(id msg.ChainId) (GetHeight, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + fn, ok := r.map2OtherHeight[id] + return fn, ok +} + +// Map2OtherSyncedHeight asks a chain's light node how far it followed MAP. It +// reports ErrNotRegistered instead of panicking when the chain was never +// registered, which is what indexing the old map did. +func (r *Registry) Map2OtherSyncedHeight(id msg.ChainId) (*big.Int, error) { + fn, ok := r.Map2OtherHeight(id) + if !ok { + return nil, errors.Wrapf(ErrNotRegistered, "map2other height provider for chain %d", id) + } + return fn() +} + +// SetLightManager records the MAP light-client manager reader. +func (r *Registry) SetLightManager(lm *LightManager) { + r.mu.Lock() + defer r.mu.Unlock() + r.lightManager = lm +} + +// LightManager returns the MAP light-client manager reader. +func (r *Registry) LightManager() (*LightManager, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if r.lightManager == nil { + return nil, errors.New("map light manager is not initialized") + } + return r.lightManager, nil +} + +// Chain2MapHeight reports how far the MAP chain has followed the given chain. +func (r *Registry) Chain2MapHeight(id msg.ChainId) (*big.Int, error) { + lm, err := r.LightManager() + if err != nil { + return nil, err + } + return lm.HeaderHeight(id) +} + +// Eth2MapNumber reports the block range the MAP chain expects next from an eth2 +// style chain. +func (r *Registry) Eth2MapNumber(id msg.ChainId) (*big.Int, *big.Int, error) { + lm, err := r.LightManager() + if err != nil { + return nil, nil, err + } + return lm.Eth2MapNumber(id) +} + +// LightNodeType reports the light node type the manager holds for a chain. +func (r *Registry) LightNodeType(method string, params ...interface{}) (*big.Int, error) { + lm, err := r.LightManager() + if err != nil { + return nil, err + } + return lm.NodeType(method, params...) +} + +// LightManagerData performs a raw read against the light-client manager. +func (r *Registry) LightManagerData(method string, params ...interface{}) ([]byte, error) { + lm, err := r.LightManager() + if err != nil { + return nil, err + } + return lm.Data(method, params...) +} diff --git a/internal/mapprotocol/registry_test.go b/internal/mapprotocol/registry_test.go new file mode 100644 index 00000000..df699b0b --- /dev/null +++ b/internal/mapprotocol/registry_test.go @@ -0,0 +1,169 @@ +package mapprotocol + +import ( + "context" + "errors" + "math/big" + "sync" + "testing" + + goeth "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/mapprotocol/compass/pkg/msg" +) + +// fakeCaller answers contract reads from a canned output, so the light manager can +// be exercised without an RPC endpoint. +type fakeCaller struct { + output []byte + err error + + mu sync.Mutex + calls int + to common.Address +} + +func (f *fakeCaller) CallContract(_ context.Context, call goeth.CallMsg, _ *big.Int) ([]byte, error) { + f.mu.Lock() + f.calls++ + if call.To != nil { + f.to = *call.To + } + f.mu.Unlock() + return f.output, f.err +} + +func TestRegistryRoundTrip(t *testing.T) { + r := NewRegistry() + + r.SetMapID("212") + if got := r.MapID(); got != "212" { + t.Fatalf("MapID = %q, want 212", got) + } + + r.SetChainName(97, "bsc") + if name, ok := r.ChainName(97); !ok || name != "bsc" { + t.Fatalf("ChainName(97) = %q, %v; want bsc, true", name, ok) + } + if got := r.ChainNameOf(999); got != "" { + t.Fatalf("ChainNameOf(unknown) = %q, want empty", got) + } + + r.SetMosAddress(97, "0xabc") + if addr, ok := r.MosAddress(97); !ok || addr != "0xabc" { + t.Fatalf("MosAddress(97) = %q, %v; want 0xabc, true", addr, ok) + } + + if _, ok := r.MosCall(97); ok { + t.Fatal("MosCall(97) reported a caller that was never registered") + } +} + +func TestRegistryMap2OtherSyncedHeight(t *testing.T) { + r := NewRegistry() + + // The old code indexed a map and called the result, so an unregistered chain + // panicked. It must be an error instead. + if _, err := r.Map2OtherSyncedHeight(97); !errors.Is(err, ErrNotRegistered) { + t.Fatalf("Map2OtherSyncedHeight on unregistered chain err = %v, want ErrNotRegistered", err) + } + + r.SetMap2OtherHeight(97, func() (*big.Int, error) { return big.NewInt(1234), nil }) + height, err := r.Map2OtherSyncedHeight(97) + if err != nil { + t.Fatalf("Map2OtherSyncedHeight: %v", err) + } + if height.Int64() != 1234 { + t.Fatalf("height = %d, want 1234", height.Int64()) + } +} + +func TestRegistryMap2OtherInitHeightsIsSnapshot(t *testing.T) { + r := NewRegistry() + r.SetMap2OtherInitHeight(97, big.NewInt(10)) + + snapshot := r.Map2OtherInitHeights() + r.SetMap2OtherInitHeight(1, big.NewInt(20)) + + if len(snapshot) != 1 { + t.Fatalf("snapshot len = %d, want 1: callers must iterate a copy", len(snapshot)) + } +} + +func TestRegistryWithoutLightManager(t *testing.T) { + r := NewRegistry() + + // Before this change these reads went through function variables that returned + // (nil, nil) until startup installed the real ones, so the caller dereferenced + // a nil height. + if _, err := r.Chain2MapHeight(97); err == nil { + t.Fatal("Chain2MapHeight without a light manager returned no error") + } + if _, err := r.LightNodeType(MethodOfNodeType, big.NewInt(97)); err == nil { + t.Fatal("LightNodeType without a light manager returned no error") + } + if _, _, err := r.Eth2MapNumber(97); err == nil { + t.Fatal("Eth2MapNumber without a light manager returned no error") + } +} + +func TestRegistryChain2MapHeightUsesLightManager(t *testing.T) { + packed, err := Height.Methods[MethodOfHeaderHeight].Outputs.Pack(big.NewInt(4242)) + if err != nil { + t.Fatalf("pack headerHeight output: %v", err) + } + + addr := common.HexToAddress("0x000068656164657273746F726541646472657373") + caller := &fakeCaller{output: packed} + + r := NewRegistry() + r.SetLightManager(NewLightManager(caller, addr)) + + height, err := r.Chain2MapHeight(97) + if err != nil { + t.Fatalf("Chain2MapHeight: %v", err) + } + if height.Int64() != 4242 { + t.Fatalf("height = %d, want 4242", height.Int64()) + } + if caller.to != addr { + t.Fatalf("called %s, want the light manager address %s", caller.to, addr) + } +} + +// TestRegistryConcurrentAccess is the reason this type exists: the expose proof API +// registers chains from one goroutine per HTTP request while the sync loops read +// the same state. Run with -race. +func TestRegistryConcurrentAccess(t *testing.T) { + r := NewRegistry() + const workers = 16 + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + id := msg.ChainId(i) + wg.Add(1) + go func() { + defer wg.Done() + for n := 0; n < 100; n++ { + r.SetMap2OtherInitHeight(id, big.NewInt(int64(n))) + r.SetMap2OtherHeight(id, func() (*big.Int, error) { return big.NewInt(int64(n)), nil }) + r.SetChainName(id, "chain") + r.SetMosAddress(id, "0xabc") + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for n := 0; n < 100; n++ { + r.Map2OtherInitHeights() + _, _ = r.Map2OtherInitHeight(id) + _, _ = r.Map2OtherHeight(id) + _ = r.ChainNameOf(id) + _, _ = r.MosCall(id) + _, _ = r.SignerCall(id) + _, _ = r.LightNodeCall(id) + } + }() + } + wg.Wait() +} From 3e3967affed237f617274a56912e979802484e18 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Sat, 22 Aug 2026 10:32:25 +0800 Subject: [PATCH 3/3] feat --- internal/mapo/callcontract.go | 221 +---------------------- internal/mapo/op_receipt_builder.go | 131 ++++++++++++++ internal/mapo/op_receipt_builder_test.go | 118 ++++++++++++ internal/mapo/proof_assembler.go | 218 ++++++++++++++++++++++ internal/mapo/proof_assembler_test.go | 76 ++++++++ internal/tx/eth_client.go | 119 +----------- internal/tx/eth_client_test.go | 45 +++++ internal/tx/receipt_fetcher.go | 143 +++++++++++++++ internal/tx/receipt_fetcher_test.go | 152 ++++++++++++++++ pkg/ethclient/ethclient.go | 17 +- pkg/ethclient/ethclient_test.go | 68 +++++++ tests/update_near_header_map_test.go | 21 --- 12 files changed, 979 insertions(+), 350 deletions(-) create mode 100644 internal/mapo/op_receipt_builder.go create mode 100644 internal/mapo/op_receipt_builder_test.go create mode 100644 internal/mapo/proof_assembler.go create mode 100644 internal/mapo/proof_assembler_test.go create mode 100644 internal/tx/eth_client_test.go create mode 100644 internal/tx/receipt_fetcher.go create mode 100644 internal/tx/receipt_fetcher_test.go create mode 100644 pkg/ethclient/ethclient_test.go delete mode 100644 tests/update_near_header_map_test.go diff --git a/internal/mapo/callcontract.go b/internal/mapo/callcontract.go index 2e014d20..167d70a2 100644 --- a/internal/mapo/callcontract.go +++ b/internal/mapo/callcontract.go @@ -4,224 +4,23 @@ package mapo import ( - "context" - "fmt" - "math/big" - "strings" - "sync" - "time" - - "github.com/mapprotocol/compass/internal/mapprotocol" - "github.com/mapprotocol/compass/pkg/msg" - - "github.com/ethereum/go-ethereum/ethdb/memorydb" - "github.com/mapprotocol/compass/internal/arb" - "github.com/mapprotocol/compass/internal/constant" - "github.com/mapprotocol/compass/internal/op" - "github.com/mapprotocol/compass/pkg/util" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/rlp" - "github.com/ethereum/go-ethereum/trie" maptypes "github.com/mapprotocol/atlas/core/types" - "github.com/mapprotocol/compass/internal/proof" + "github.com/mapprotocol/compass/internal/mapprotocol" "github.com/mapprotocol/compass/pkg/ethclient" - "github.com/pkg/errors" + "github.com/mapprotocol/compass/pkg/msg" ) func AssembleEthProof(conn *ethclient.Client, log *types.Log, receipts []*types.Receipt, header *maptypes.Header, - method string, fId msg.ChainId, proofType int64, sign [][]byte) ([]byte, error) { - var ( - pack []byte - err error - orderId = log.Topics[1] - ) - idx := 0 - for i, ele := range receipts[log.TxIndex].Logs { - if ele.Index != log.Index { - continue - } - idx = i - } - receipt, err := mapprotocol.GetTxReceipt(receipts[log.TxIndex]) - if err != nil { - return nil, err - } - prf, receiptHash, err := ethProof(conn, fId, log.TxIndex, receipts) - if err != nil { - return nil, err - } - - var key []byte - key = rlp.AppendUint64(key[:0], uint64(log.TxIndex)) - - switch proofType { - case constant.ProofTypeOfOrigin: - case constant.ProofTypeOfZk: - case constant.ProofTypeOfOracle: - pack, err = proof.Oracle(log.BlockNumber, receipt, key, prf, fId, method, idx, - mapprotocol.ProofAbi, orderId, false) - case constant.ProofTypeOfNewOracle: - if receiptHash.Hex() != header.ReceiptHash.Hex() { - fmt.Println("Evm generate receiptHash ", receiptHash, "oracle", header.ReceiptHash.Hex(), " not same") - return nil, errors.New("receiptHash not same") - } - fallthrough - case constant.ProofTypeOfLogOracle: - pack, err = proof.SignOracle(&maptypes.Header{ - ReceiptHash: receiptHash, - Number: big.NewInt(int64(log.BlockNumber)), - }, receipt, key, prf, fId, idx, method, sign, log, proofType) - } - if err != nil { - return nil, err - } - - return pack, nil -} - -func ethProof(conn *ethclient.Client, fId msg.ChainId, txIdx uint, receipts []*types.Receipt) ([][]byte, common.Hash, error) { - var dls proof.DerivableList - switch fId { - case constant.ArbChainId, constant.Robinhood: - pr := arb.Receipts{} - for _, r := range receipts { - pr = append(pr, &arb.Receipt{Receipt: r}) - } - dls = pr - case constant.BaseChainId, constant.OptimismChainId: - pr := op.Receipts{} - results := make([]*op.Receipt, len(receipts)) - var wg sync.WaitGroup - for i, r := range receipts { - wg.Add(1) - go func(i int, r *types.Receipt) { - defer wg.Done() - tmp, err := conn.OpReceipt(context.Background(), r.TxHash) - if err != nil { - return - } - if tmp == nil { - return - } - vptr := uint64(0) - nptr := uint64(0) - if tmp.DepositReceiptVersion != "" { - version, _ := big.NewInt(0).SetString(strings.TrimPrefix(tmp.DepositReceiptVersion, "0x"), 16) - vptr = version.Uint64() - } - if tmp.DepositNonce != "" { - nonce, _ := big.NewInt(0).SetString(strings.TrimPrefix(tmp.DepositNonce, "0x"), 16) - nptr = nonce.Uint64() - } - results[i] = &op.Receipt{Receipt: r, DepositReceiptVersion: &vptr, DepositNonce: &nptr} - }(i, r) - if i%30 == 0 { - time.Sleep(100 * time.Millisecond) - } - } - wg.Wait() - for _, result := range results { - if result == nil { - continue - } - pr = append(pr, result) - } - dls = pr - default: - dls = types.Receipts(receipts) - } - ret, err := proof.Get(dls, txIdx) - if err != nil { - return nil, constant.ZeroAddress.Hash(), err - } - - tr, _ := trie.New(common.Hash{}, trie.NewDatabase(memorydb.New())) - tr = proof.DeriveTire(dls, tr) - - return ret, tr.Hash(), nil + method string, fromChainID msg.ChainId, proofType int64, sign [][]byte) ([]byte, error) { + return NewProofAssembler(conn).AssembleEth(log, receipts, header, method, fromChainID, proofType, sign) } -func AssembleMapProof(cli *ethclient.Client, log *types.Log, receipts []*types.Receipt, - header *maptypes.Header, fId msg.ChainId, method, zkUrl string, proofType int64, sign [][]byte) (uint64, []byte, error) { - uToChainID := big.NewInt(0).SetBytes(log.Topics[2].Bytes()[8:16]).Uint64() - txIndex := log.TxIndex - orderId := log.Topics[1] - aggPK, ist, _, err := mapprotocol.GetAggPK(cli, new(big.Int).Sub(header.Number, big.NewInt(1)), header.Extra) - if err != nil { - return 0, nil, err - } - - receipt, err := mapprotocol.GetTxReceipt(receipts[txIndex]) - prf, err := proof.Get(types.Receipts(receipts), txIndex) - if err != nil { - return 0, nil, err - } - - var key []byte - key = rlp.AppendUint64(key[:0], uint64(txIndex)) - ek := util.Key2Hex(key, len(prf)) - - var payloads []byte - name := mapprotocol.Default().ChainNameOf(msg.ChainId(uToChainID)) - switch name { - default: - istanbulExtra := mapprotocol.ConvertIstanbulExtra(ist) - nr := mapprotocol.MapTxReceipt{ - PostStateOrStatus: receipt.PostStateOrStatus, - CumulativeGasUsed: receipt.CumulativeGasUsed, - Bloom: receipt.Bloom, - Logs: receipt.Logs, - } - - nrRlp, err := rlp.EncodeToBytes(nr) - if err != nil { - return 0, nil, err - } - rp := mapprotocol.NewMapReceiptProof{ - Header: mapprotocol.ConvertHeader(header), - AggPk: aggPK, - KeyIndex: ek, - Proof: prf, - Ist: *istanbulExtra, - TxReceiptRlp: mapprotocol.TxReceiptRlp{ - ReceiptType: receipt.ReceiptType, - ReceiptRlp: nrRlp, - }, - } - - idx := 0 - for i, ele := range receipts[txIndex].Logs { - if ele.Index != log.Index { - continue - } - idx = i - } - - switch proofType { - case constant.ProofTypeOfZk: - zkProof, err := mapprotocol.GetZkProof(zkUrl, fId, header.Number.Uint64()) - if err != nil { - return 0, nil, errors.Wrap(err, "GetZkProof failed") - } - payloads, err = proof.Pack(fId, method, mapprotocol.Mcs, rp, zkProof) - case constant.ProofTypeOfOracle: - payloads, err = proof.Oracle(header.Number.Uint64(), receipt, key, prf, fId, - method, idx, mapprotocol.ProofAbi, orderId, false) - case constant.ProofTypeOfNewOracle: - fallthrough - case constant.ProofTypeOfLogOracle: - payloads, err = proof.SignOracle(header, receipt, key, prf, fId, idx, method, sign, log, proofType) - default: - payloads, err = proof.V3Pack(fId, method, mapprotocol.Map2Other, idx, orderId, true, rp) - } - if err != nil { - return 0, nil, err - } - - } - return uToChainID, payloads, nil +func AssembleMapProof(conn *ethclient.Client, log *types.Log, receipts []*types.Receipt, + header *maptypes.Header, fromChainID msg.ChainId, method, zkURL string, + proofType int64, sign [][]byte) (uint64, []byte, error) { + return NewProofAssembler(conn).AssembleMap(log, receipts, header, fromChainID, method, zkURL, proofType, sign) } func Key2Hex(str []byte, proofLength int) []byte { @@ -255,8 +54,8 @@ func ConvertNearReceipt(h *mapprotocol.TxReceipt) *TxReceipt { logs := make([]TxLog, 0, len(h.Logs)) for _, log := range h.Logs { topics := make([]string, 0, len(log.Topics)) - for _, t := range log.Topics { - topics = append(topics, "0x"+common.Bytes2Hex(t)) + for _, topic := range log.Topics { + topics = append(topics, "0x"+common.Bytes2Hex(topic)) } logs = append(logs, TxLog{ Address: log.Addr, diff --git a/internal/mapo/op_receipt_builder.go b/internal/mapo/op_receipt_builder.go new file mode 100644 index 00000000..55fac76e --- /dev/null +++ b/internal/mapo/op_receipt_builder.go @@ -0,0 +1,131 @@ +package mapo + +import ( + "context" + "fmt" + "math/big" + "strings" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/mapprotocol/compass/internal/op" + "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/pkg/errors" +) + +const maxOPReceiptWorkers = 30 + +type OPReceiptReader interface { + OpReceipt(context.Context, common.Hash) (*ethclient.OpReceipt, error) +} + +type OPReceiptBuilder struct { + reader OPReceiptReader + workerLimit int +} + +func NewOPReceiptBuilder(reader OPReceiptReader) *OPReceiptBuilder { + return &OPReceiptBuilder{ + reader: reader, + workerLimit: maxOPReceiptWorkers, + } +} + +func (b *OPReceiptBuilder) Build(ctx context.Context, receipts []*types.Receipt) (op.Receipts, error) { + if len(receipts) == 0 { + return op.Receipts{}, nil + } + type job struct { + idx int + receipt *types.Receipt + } + type result struct { + idx int + receipt *op.Receipt + err error + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + jobs := make(chan job) + results := make(chan result) + workerCount := min(b.workerLimit, len(receipts)) + var workers sync.WaitGroup + workers.Add(workerCount) + for range workerCount { + go func() { + defer workers.Done() + for current := range jobs { + receipt, err := b.buildOne(ctx, current.receipt) + select { + case results <- result{idx: current.idx, receipt: receipt, err: err}: + case <-ctx.Done(): + return + } + } + }() + } + go func() { + defer close(jobs) + for idx, receipt := range receipts { + select { + case jobs <- job{idx: idx, receipt: receipt}: + case <-ctx.Done(): + return + } + } + }() + go func() { + workers.Wait() + close(results) + }() + + ret := make(op.Receipts, len(receipts)) + for result := range results { + if result.err != nil { + return nil, fmt.Errorf("build OP receipt %s at index %d: %w", + receipts[result.idx].TxHash, result.idx, result.err) + } + ret[result.idx] = result.receipt + } + if err := ctx.Err(); err != nil { + return nil, err + } + return ret, nil +} + +func (b *OPReceiptBuilder) buildOne(ctx context.Context, receipt *types.Receipt) (*op.Receipt, error) { + raw, err := b.reader.OpReceipt(ctx, receipt.TxHash) + if err != nil { + return nil, err + } + if raw == nil { + return nil, errors.New("empty OP receipt") + } + version, err := parseOptionalHexUint64("depositReceiptVersion", raw.DepositReceiptVersion) + if err != nil { + return nil, err + } + nonce, err := parseOptionalHexUint64("depositNonce", raw.DepositNonce) + if err != nil { + return nil, err + } + return &op.Receipt{ + Receipt: receipt, + DepositReceiptVersion: version, + DepositNonce: nonce, + }, nil +} + +func parseOptionalHexUint64(field, value string) (*uint64, error) { + if value == "" { + return nil, nil + } + parsed, ok := new(big.Int).SetString(strings.TrimPrefix(value, "0x"), 16) + if !ok || !parsed.IsUint64() { + return nil, fmt.Errorf("invalid %s %q", field, value) + } + ret := parsed.Uint64() + return &ret, nil +} diff --git a/internal/mapo/op_receipt_builder_test.go b/internal/mapo/op_receipt_builder_test.go new file mode 100644 index 00000000..6f866c3d --- /dev/null +++ b/internal/mapo/op_receipt_builder_test.go @@ -0,0 +1,118 @@ +package mapo + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/mapprotocol/compass/pkg/ethclient" +) + +type opReceiptReaderFunc func(context.Context, common.Hash) (*ethclient.OpReceipt, error) + +func (f opReceiptReaderFunc) OpReceipt(ctx context.Context, hash common.Hash) (*ethclient.OpReceipt, error) { + return f(ctx, hash) +} + +func TestOPReceiptBuilderObjectAPI(t *testing.T) { + receipts := []*types.Receipt{{TxHash: common.HexToHash("0x1")}} + reader := opReceiptReaderFunc(func(context.Context, common.Hash) (*ethclient.OpReceipt, error) { + return ðclient.OpReceipt{DepositNonce: "0x2"}, nil + }) + builder := NewOPReceiptBuilder(reader) + + got, err := builder.Build(context.Background(), receipts) + if err != nil { + t.Fatalf("OPReceiptBuilder.Build returned error: %v", err) + } + if len(got) != 1 || got[0].DepositNonce == nil || *got[0].DepositNonce != 2 { + t.Fatalf("OPReceiptBuilder.Build returned %+v, want nonce 2", got) + } +} + +func TestOPReceiptBuilderReturnsFetchError(t *testing.T) { + wantErr := errors.New("rpc failed") + receipts := []*types.Receipt{ + {TxHash: common.HexToHash("0x1")}, + {TxHash: common.HexToHash("0x2")}, + } + reader := opReceiptReaderFunc(func(_ context.Context, hash common.Hash) (*ethclient.OpReceipt, error) { + if hash == receipts[1].TxHash { + return nil, wantErr + } + return ðclient.OpReceipt{}, nil + }) + + _, err := NewOPReceiptBuilder(reader).Build(context.Background(), receipts) + if !errors.Is(err, wantErr) { + t.Fatalf("OPReceiptBuilder.Build error = %v, want wrapped %v", err, wantErr) + } +} + +func TestOPReceiptBuilderParsesOptionalFieldsAndPreservesOrder(t *testing.T) { + receipts := []*types.Receipt{ + {TxHash: common.HexToHash("0x1")}, + {TxHash: common.HexToHash("0x2")}, + } + reader := opReceiptReaderFunc(func(_ context.Context, hash common.Hash) (*ethclient.OpReceipt, error) { + if hash == receipts[0].TxHash { + return ðclient.OpReceipt{ + DepositNonce: "0x2", + DepositReceiptVersion: "0x1", + }, nil + } + return ðclient.OpReceipt{}, nil + }) + + got, err := NewOPReceiptBuilder(reader).Build(context.Background(), receipts) + if err != nil { + t.Fatalf("OPReceiptBuilder.Build returned error: %v", err) + } + if len(got) != len(receipts) || got[0].Receipt != receipts[0] || got[1].Receipt != receipts[1] { + t.Fatalf("OPReceiptBuilder.Build did not preserve receipt order: %+v", got) + } + if got[0].DepositNonce == nil || *got[0].DepositNonce != 2 { + t.Fatalf("deposit nonce = %v, want 2", got[0].DepositNonce) + } + if got[0].DepositReceiptVersion == nil || *got[0].DepositReceiptVersion != 1 { + t.Fatalf("deposit receipt version = %v, want 1", got[0].DepositReceiptVersion) + } + if got[1].DepositNonce != nil || got[1].DepositReceiptVersion != nil { + t.Fatalf("missing optional fields should remain nil: %+v", got[1]) + } +} + +func TestOPReceiptBuilderFetchesConcurrently(t *testing.T) { + receipts := []*types.Receipt{ + {TxHash: common.HexToHash("0x1")}, + {TxHash: common.HexToHash("0x2")}, + } + started := make(chan struct{}, len(receipts)) + release := make(chan struct{}) + reader := opReceiptReaderFunc(func(context.Context, common.Hash) (*ethclient.OpReceipt, error) { + started <- struct{}{} + <-release + return ðclient.OpReceipt{}, nil + }) + + done := make(chan error, 1) + go func() { + _, err := NewOPReceiptBuilder(reader).Build(context.Background(), receipts) + done <- err + }() + <-started + select { + case <-started: + close(release) + case <-time.After(25 * time.Millisecond): + close(release) + <-done + t.Fatal("OP receipt fetches ran serially") + } + if err := <-done; err != nil { + t.Fatalf("OPReceiptBuilder.Build returned error: %v", err) + } +} diff --git a/internal/mapo/proof_assembler.go b/internal/mapo/proof_assembler.go new file mode 100644 index 00000000..2a1855d5 --- /dev/null +++ b/internal/mapo/proof_assembler.go @@ -0,0 +1,218 @@ +package mapo + +import ( + "context" + "fmt" + "math/big" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb/memorydb" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" + maptypes "github.com/mapprotocol/atlas/core/types" + "github.com/mapprotocol/compass/internal/arb" + "github.com/mapprotocol/compass/internal/constant" + "github.com/mapprotocol/compass/internal/mapprotocol" + "github.com/mapprotocol/compass/internal/proof" + "github.com/mapprotocol/compass/pkg/ethclient" + "github.com/mapprotocol/compass/pkg/msg" + "github.com/mapprotocol/compass/pkg/util" + "github.com/pkg/errors" +) + +const defaultOPReceiptTimeout = 30 * time.Second + +type ProofAssembler struct { + client *ethclient.Client + opReceiptBuilder *OPReceiptBuilder + opReceiptTimeout time.Duration +} + +func NewProofAssembler(client *ethclient.Client) *ProofAssembler { + return &ProofAssembler{ + client: client, + opReceiptBuilder: NewOPReceiptBuilder(client), + opReceiptTimeout: defaultOPReceiptTimeout, + } +} + +func (a *ProofAssembler) AssembleEth(log *types.Log, receipts []*types.Receipt, header *maptypes.Header, + method string, fromChainID msg.ChainId, proofType int64, sign [][]byte) ([]byte, error) { + if log == nil { + return nil, errors.New("nil log") + } + if len(log.Topics) < 2 { + return nil, fmt.Errorf("log has %d topics, want at least 2", len(log.Topics)) + } + sourceReceipt, logIndex, err := a.receiptAndLogIndex(receipts, log) + if err != nil { + return nil, err + } + + receipt, err := mapprotocol.GetTxReceipt(sourceReceipt) + if err != nil { + return nil, err + } + proofNodes, receiptHash, err := a.ethProof(fromChainID, log.TxIndex, receipts) + if err != nil { + return nil, err + } + + var key []byte + key = rlp.AppendUint64(key[:0], uint64(log.TxIndex)) + orderID := log.Topics[1] + var pack []byte + switch proofType { + case constant.ProofTypeOfOrigin: + case constant.ProofTypeOfZk: + case constant.ProofTypeOfOracle: + pack, err = proof.Oracle(log.BlockNumber, receipt, key, proofNodes, fromChainID, method, logIndex, + mapprotocol.ProofAbi, orderID, false) + case constant.ProofTypeOfNewOracle: + if receiptHash.Hex() != header.ReceiptHash.Hex() { + return nil, fmt.Errorf("receipt hash %s does not match header receipt hash %s", receiptHash, header.ReceiptHash) + } + fallthrough + case constant.ProofTypeOfLogOracle: + pack, err = proof.SignOracle(&maptypes.Header{ + ReceiptHash: receiptHash, + Number: big.NewInt(int64(log.BlockNumber)), + }, receipt, key, proofNodes, fromChainID, logIndex, method, sign, log, proofType) + } + if err != nil { + return nil, err + } + return pack, nil +} + +func (a *ProofAssembler) AssembleMap(log *types.Log, receipts []*types.Receipt, header *maptypes.Header, + fromChainID msg.ChainId, method, zkURL string, proofType int64, sign [][]byte) (uint64, []byte, error) { + if log == nil { + return 0, nil, errors.New("nil log") + } + if len(log.Topics) < 3 { + return 0, nil, fmt.Errorf("log has %d topics, want at least 3", len(log.Topics)) + } + sourceReceipt, logIndex, err := a.receiptAndLogIndex(receipts, log) + if err != nil { + return 0, nil, err + } + + toChainID := big.NewInt(0).SetBytes(log.Topics[2].Bytes()[8:16]).Uint64() + orderID := log.Topics[1] + aggPK, ist, _, err := mapprotocol.GetAggPK(a.client, new(big.Int).Sub(header.Number, big.NewInt(1)), header.Extra) + if err != nil { + return 0, nil, err + } + receipt, err := mapprotocol.GetTxReceipt(sourceReceipt) + if err != nil { + return 0, nil, err + } + proofNodes, err := proof.Get(types.Receipts(receipts), log.TxIndex) + if err != nil { + return 0, nil, err + } + + var key []byte + key = rlp.AppendUint64(key[:0], uint64(log.TxIndex)) + keyIndex := util.Key2Hex(key, len(proofNodes)) + istanbulExtra := mapprotocol.ConvertIstanbulExtra(ist) + mapReceipt := mapprotocol.MapTxReceipt{ + PostStateOrStatus: receipt.PostStateOrStatus, + CumulativeGasUsed: receipt.CumulativeGasUsed, + Bloom: receipt.Bloom, + Logs: receipt.Logs, + } + receiptRLP, err := rlp.EncodeToBytes(mapReceipt) + if err != nil { + return 0, nil, err + } + receiptProof := mapprotocol.NewMapReceiptProof{ + Header: mapprotocol.ConvertHeader(header), + AggPk: aggPK, + KeyIndex: keyIndex, + Proof: proofNodes, + Ist: *istanbulExtra, + TxReceiptRlp: mapprotocol.TxReceiptRlp{ + ReceiptType: receipt.ReceiptType, + ReceiptRlp: receiptRLP, + }, + } + + var payload []byte + switch proofType { + case constant.ProofTypeOfZk: + zkProof, err := mapprotocol.GetZkProof(zkURL, fromChainID, header.Number.Uint64()) + if err != nil { + return 0, nil, errors.Wrap(err, "GetZkProof failed") + } + payload, err = proof.Pack(fromChainID, method, mapprotocol.Mcs, receiptProof, zkProof) + case constant.ProofTypeOfOracle: + payload, err = proof.Oracle(header.Number.Uint64(), receipt, key, proofNodes, fromChainID, + method, logIndex, mapprotocol.ProofAbi, orderID, false) + case constant.ProofTypeOfNewOracle: + fallthrough + case constant.ProofTypeOfLogOracle: + payload, err = proof.SignOracle(header, receipt, key, proofNodes, fromChainID, logIndex, method, sign, log, proofType) + default: + payload, err = proof.V3Pack(fromChainID, method, mapprotocol.Map2Other, logIndex, orderID, true, receiptProof) + } + if err != nil { + return 0, nil, err + } + return toChainID, payload, nil +} + +func (a *ProofAssembler) receiptAndLogIndex(receipts []*types.Receipt, target *types.Log) (*types.Receipt, int, error) { + if target.TxIndex >= uint(len(receipts)) { + return nil, 0, fmt.Errorf("receipt index %d out of range for %d receipts", target.TxIndex, len(receipts)) + } + receipt := receipts[target.TxIndex] + if receipt == nil { + return nil, 0, errors.New("nil receipt") + } + for idx, entry := range receipt.Logs { + if entry != nil && entry.Index == target.Index { + return receipt, idx, nil + } + } + return nil, 0, fmt.Errorf("log index %d not found in receipt %s", target.Index, target.TxHash) +} + +func (a *ProofAssembler) ethProof(fromChainID msg.ChainId, txIndex uint, + receipts []*types.Receipt) ([][]byte, common.Hash, error) { + ctx, cancel := context.WithTimeout(context.Background(), a.opReceiptTimeout) + defer cancel() + return a.buildEthProof(ctx, fromChainID, txIndex, receipts) +} + +func (a *ProofAssembler) buildEthProof(ctx context.Context, fromChainID msg.ChainId, txIndex uint, + receipts []*types.Receipt) ([][]byte, common.Hash, error) { + var derivable proof.DerivableList + switch fromChainID { + case constant.ArbChainId, constant.Robinhood: + arbReceipts := arb.Receipts{} + for _, receipt := range receipts { + arbReceipts = append(arbReceipts, &arb.Receipt{Receipt: receipt}) + } + derivable = arbReceipts + case constant.BaseChainId, constant.OptimismChainId: + opReceipts, err := a.opReceiptBuilder.Build(ctx, receipts) + if err != nil { + return nil, constant.ZeroAddress.Hash(), err + } + derivable = opReceipts + default: + derivable = types.Receipts(receipts) + } + + proofNodes, err := proof.Get(derivable, txIndex) + if err != nil { + return nil, constant.ZeroAddress.Hash(), err + } + receiptTrie, _ := trie.New(common.Hash{}, trie.NewDatabase(memorydb.New())) + receiptTrie = proof.DeriveTire(derivable, receiptTrie) + return proofNodes, receiptTrie.Hash(), nil +} diff --git a/internal/mapo/proof_assembler_test.go b/internal/mapo/proof_assembler_test.go new file mode 100644 index 00000000..250b9cb6 --- /dev/null +++ b/internal/mapo/proof_assembler_test.go @@ -0,0 +1,76 @@ +package mapo + +import ( + "context" + "errors" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/mapprotocol/compass/internal/constant" + "github.com/mapprotocol/compass/pkg/ethclient" +) + +func TestProofAssemblerReturnsOPReceiptFetchError(t *testing.T) { + wantErr := errors.New("rpc failed") + receipts := []*types.Receipt{{TxHash: common.HexToHash("0x1")}} + reader := opReceiptReaderFunc(func(context.Context, common.Hash) (*ethclient.OpReceipt, error) { + return nil, wantErr + }) + assembler := &ProofAssembler{opReceiptBuilder: NewOPReceiptBuilder(reader)} + + _, _, err := assembler.buildEthProof(context.Background(), constant.BaseChainId, 0, receipts) + if !errors.Is(err, wantErr) { + t.Fatalf("ProofAssembler.buildEthProof error = %v, want wrapped %v", err, wantErr) + } +} + +func TestProofAssemblerReturnsErrorWhenLogIsMissing(t *testing.T) { + receipt := &types.Receipt{Logs: []*types.Log{{Index: 1}}} + target := &types.Log{Index: 2, TxHash: common.HexToHash("0x2")} + + _, _, err := (&ProofAssembler{}).receiptAndLogIndex([]*types.Receipt{receipt}, target) + if err == nil { + t.Fatal("ProofAssembler.receiptAndLogIndex returned nil error for a missing log") + } +} + +func TestProofAssemblerReturnsPositionWithinReceipt(t *testing.T) { + receipt := &types.Receipt{Logs: []*types.Log{ + {Index: 3}, + {Index: 7}, + }} + target := &types.Log{Index: 7} + + gotReceipt, gotIndex, err := (&ProofAssembler{}).receiptAndLogIndex([]*types.Receipt{receipt}, target) + if err != nil { + t.Fatalf("ProofAssembler.receiptAndLogIndex returned error: %v", err) + } + if gotReceipt != receipt || gotIndex != 1 { + t.Fatalf("receiptAndLogIndex = (%p, %d), want (%p, 1)", gotReceipt, gotIndex, receipt) + } +} + +func TestProofAssemblerObjectAPIRejectsMissingLog(t *testing.T) { + assembler := NewProofAssembler(nil) + log := &types.Log{ + Index: 2, + Topics: []common.Hash{{}, common.HexToHash("0x1")}, + } + receipts := []*types.Receipt{{Logs: []*types.Log{{Index: 1}}}} + + _, err := assembler.AssembleEth(log, receipts, nil, "", 0, 0, nil) + if err == nil { + t.Fatal("ProofAssembler.AssembleEth returned nil error for a missing log") + } +} + +func TestProofAssemblerObjectAPIRejectsInvalidMapLog(t *testing.T) { + assembler := NewProofAssembler(nil) + log := &types.Log{Topics: []common.Hash{{}, common.HexToHash("0x1")}} + + _, _, err := assembler.AssembleMap(log, nil, nil, 0, "", "", 0, nil) + if err == nil { + t.Fatal("ProofAssembler.AssembleMap returned nil error for an invalid log") + } +} diff --git a/internal/tx/eth_client.go b/internal/tx/eth_client.go index 8a8a37b3..6b45d328 100644 --- a/internal/tx/eth_client.go +++ b/internal/tx/eth_client.go @@ -2,9 +2,7 @@ package tx import ( "context" - "errors" "math/big" - "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" @@ -25,122 +23,9 @@ func GetTxsHashByBlockNumber(conn *ethclient.Client, number *big.Int) ([]common. } func GetReceiptsByTxsHash(conn *ethclient.Client, txsHash []common.Hash) ([]*types.Receipt, error) { - type ele struct { - r *types.Receipt - idx int - } - var ( - count = len(txsHash) - errReceive = make(chan error) - receive = make(chan *ele, len(txsHash)) - rs = make([]*types.Receipt, len(txsHash)) - ) - go func() { - for idx, h := range txsHash { - tmpIdx := idx - tmpHash := h - go func(i int, tx common.Hash) { - for { - r, err := conn.TransactionReceipt(context.Background(), tx) - if err != nil { - if err.Error() == "not found" { - time.Sleep(time.Millisecond * 100) - continue - } - errReceive <- err - return - } - receive <- &ele{ - r: r, - idx: i, - } - break - } - }(tmpIdx, tmpHash) - - if idx%30 == 0 { - time.Sleep(time.Millisecond * 500) - } - } - }() - - for { - select { - case v, ok := <-receive: - if !ok { - return nil, errors.New("receive chan is closed") - } - if v != nil { - rs[v.idx] = v.r - } - count-- - if count == 0 { - return rs, nil - } - case err := <-errReceive: - return nil, err - } - } + return NewReceiptFetcher(conn, RetryNotFound).Fetch(context.Background(), txsHash) } func GetMaticReceiptsByTxsHash(conn *ethclient.Client, txsHash []common.Hash) ([]*types.Receipt, error) { - type ele struct { - r *types.Receipt - idx int - } - var ( - count = len(txsHash) - errReceive = make(chan error) - receive = make(chan *ele, len(txsHash)) - rs = make([]*types.Receipt, len(txsHash)) - ) - go func() { - for idx, h := range txsHash { - tmpIdx := idx - tmpHash := h - go func(i int, tx common.Hash) { - for { - r, err := conn.TransactionReceipt(context.Background(), tx) - if err != nil { - if err.Error() == "not found" { - receive <- &ele{ - r: nil, - idx: i, - } - break - } - errReceive <- err - return - } - receive <- &ele{ - r: r, - idx: i, - } - break - } - }(tmpIdx, tmpHash) - - if idx%30 == 0 { - time.Sleep(time.Millisecond * 500) - } - } - }() - - for { - select { - case v, ok := <-receive: - if !ok { - return nil, errors.New("receive chan is closed") - } - if v != nil { - rs[v.idx] = v.r - } - count-- - if count == 0 { - return rs, nil - } - case err := <-errReceive: - return nil, err - } - } + return NewReceiptFetcher(conn, PreserveNotFound).Fetch(context.Background(), txsHash) } diff --git a/internal/tx/eth_client_test.go b/internal/tx/eth_client_test.go new file mode 100644 index 00000000..f0d68e67 --- /dev/null +++ b/internal/tx/eth_client_test.go @@ -0,0 +1,45 @@ +package tx + +import ( + "errors" + "testing" + "time" + + "github.com/ethereum/go-ethereum/core/types" +) + +func TestPublicReceiptFetchersReturnForEmptyInput(t *testing.T) { + tests := []struct { + name string + fetch func() ([]*types.Receipt, error) + }{ + {name: "standard", fetch: func() ([]*types.Receipt, error) { + return GetReceiptsByTxsHash(nil, nil) + }}, + {name: "matic", fetch: func() ([]*types.Receipt, error) { + return GetMaticReceiptsByTxsHash(nil, nil) + }}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + done := make(chan error, 1) + go func() { + receipts, err := tt.fetch() + if err == nil && len(receipts) != 0 { + err = errors.New("empty input returned non-empty receipts") + } + done <- err + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("empty input returned error: %v", err) + } + case <-time.After(25 * time.Millisecond): + t.Fatal("empty input did not return promptly") + } + }) + } +} diff --git a/internal/tx/receipt_fetcher.go b/internal/tx/receipt_fetcher.go new file mode 100644 index 00000000..edce7abc --- /dev/null +++ b/internal/tx/receipt_fetcher.go @@ -0,0 +1,143 @@ +package tx + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" +) + +type ReceiptReader interface { + TransactionReceipt(context.Context, common.Hash) (*types.Receipt, error) +} + +type MissingReceiptPolicy uint8 + +const ( + RetryNotFound MissingReceiptPolicy = iota + PreserveNotFound + + maxReceiptWorkers = 30 + receiptRetryInterval = 100 * time.Millisecond + receiptFetchTimeout = 30 * time.Second +) + +type ReceiptFetcher struct { + reader ReceiptReader + policy MissingReceiptPolicy + workerLimit int + timeout time.Duration + retryInterval time.Duration +} + +func NewReceiptFetcher(reader ReceiptReader, policy MissingReceiptPolicy) *ReceiptFetcher { + return &ReceiptFetcher{ + reader: reader, + policy: policy, + workerLimit: maxReceiptWorkers, + timeout: receiptFetchTimeout, + retryInterval: receiptRetryInterval, + } +} + +func (f *ReceiptFetcher) Fetch(ctx context.Context, hashes []common.Hash) ([]*types.Receipt, error) { + if len(hashes) == 0 { + return []*types.Receipt{}, nil + } + ctx, cancel := context.WithTimeout(ctx, f.timeout) + defer cancel() + + type job struct { + idx int + hash common.Hash + } + type result struct { + idx int + receipt *types.Receipt + err error + } + + ctx, stopWorkers := context.WithCancel(ctx) + defer stopWorkers() + jobs := make(chan job) + results := make(chan result) + workerCount := min(f.workerLimit, len(hashes)) + var workers sync.WaitGroup + workers.Add(workerCount) + for range workerCount { + go func() { + defer workers.Done() + for current := range jobs { + receipt, err := f.fetchOne(ctx, current.hash) + select { + case results <- result{idx: current.idx, receipt: receipt, err: err}: + case <-ctx.Done(): + return + } + } + }() + } + + go func() { + defer close(jobs) + for idx, hash := range hashes { + select { + case jobs <- job{idx: idx, hash: hash}: + case <-ctx.Done(): + return + } + } + }() + go func() { + workers.Wait() + close(results) + }() + + receipts := make([]*types.Receipt, len(hashes)) + for result := range results { + if result.err != nil { + return nil, fmt.Errorf("fetch receipt %s at index %d: %w", hashes[result.idx], result.idx, result.err) + } + receipts[result.idx] = result.receipt + } + if err := ctx.Err(); err != nil { + return nil, err + } + return receipts, nil +} + +func (f *ReceiptFetcher) fetchOne(ctx context.Context, hash common.Hash) (*types.Receipt, error) { + for { + receipt, err := f.reader.TransactionReceipt(ctx, hash) + if err == nil { + if receipt == nil && f.policy == RetryNotFound { + return nil, errors.New("empty receipt") + } + return receipt, nil + } + if !errors.Is(err, ethereum.NotFound) { + return nil, err + } + if f.policy == PreserveNotFound { + return nil, nil + } + + timer := time.NewTimer(f.retryInterval) + select { + case <-timer.C: + case <-ctx.Done(): + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return nil, ctx.Err() + } + } +} diff --git a/internal/tx/receipt_fetcher_test.go b/internal/tx/receipt_fetcher_test.go new file mode 100644 index 00000000..ad7ea3be --- /dev/null +++ b/internal/tx/receipt_fetcher_test.go @@ -0,0 +1,152 @@ +package tx + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" +) + +type receiptReaderFunc func(context.Context, common.Hash) (*types.Receipt, error) + +func (f receiptReaderFunc) TransactionReceipt(ctx context.Context, hash common.Hash) (*types.Receipt, error) { + return f(ctx, hash) +} + +func TestReceiptFetcherObjectAPI(t *testing.T) { + hashes := []common.Hash{common.HexToHash("0x1"), common.HexToHash("0x2")} + reader := receiptReaderFunc(func(_ context.Context, hash common.Hash) (*types.Receipt, error) { + return &types.Receipt{TxHash: hash}, nil + }) + + receipts, err := NewReceiptFetcher(reader, RetryNotFound).Fetch(context.Background(), hashes) + if err != nil { + t.Fatalf("ReceiptFetcher.Fetch returned error: %v", err) + } + for idx, receipt := range receipts { + if receipt == nil || receipt.TxHash != hashes[idx] { + t.Fatalf("receipt %d = %+v, want hash %s", idx, receipt, hashes[idx]) + } + } +} + +func TestReceiptFetcherEmpty(t *testing.T) { + called := false + reader := receiptReaderFunc(func(context.Context, common.Hash) (*types.Receipt, error) { + called = true + return nil, nil + }) + + receipts, err := NewReceiptFetcher(reader, RetryNotFound).Fetch(context.Background(), nil) + if err != nil { + t.Fatalf("ReceiptFetcher.Fetch returned error: %v", err) + } + if len(receipts) != 0 { + t.Fatalf("ReceiptFetcher.Fetch returned %d receipts, want 0", len(receipts)) + } + if called { + t.Fatal("ReceiptFetcher.Fetch called reader for empty input") + } +} + +func TestReceiptFetcherPreservesOrder(t *testing.T) { + hashes := []common.Hash{ + common.HexToHash("0x1"), + common.HexToHash("0x2"), + common.HexToHash("0x3"), + } + reader := receiptReaderFunc(func(_ context.Context, hash common.Hash) (*types.Receipt, error) { + return &types.Receipt{TxHash: hash}, nil + }) + + receipts, err := NewReceiptFetcher(reader, PreserveNotFound).Fetch(context.Background(), hashes) + if err != nil { + t.Fatalf("ReceiptFetcher.Fetch returned error: %v", err) + } + if len(receipts) != len(hashes) { + t.Fatalf("ReceiptFetcher.Fetch returned %d receipts, want %d", len(receipts), len(hashes)) + } + for idx, receipt := range receipts { + if receipt == nil || receipt.TxHash != hashes[idx] { + t.Fatalf("receipt %d = %+v, want hash %s", idx, receipt, hashes[idx]) + } + } +} + +func TestReceiptFetcherRetriesNotFound(t *testing.T) { + hash := common.HexToHash("0x1") + calls := 0 + reader := receiptReaderFunc(func(_ context.Context, got common.Hash) (*types.Receipt, error) { + calls++ + if calls == 1 { + return nil, ethereum.NotFound + } + return &types.Receipt{TxHash: got}, nil + }) + + receipts, err := NewReceiptFetcher(reader, RetryNotFound).Fetch(context.Background(), []common.Hash{hash}) + if err != nil { + t.Fatalf("ReceiptFetcher.Fetch returned error: %v", err) + } + if calls != 2 { + t.Fatalf("reader called %d times, want 2", calls) + } + if len(receipts) != 1 || receipts[0] == nil || receipts[0].TxHash != hash { + t.Fatalf("ReceiptFetcher.Fetch returned %+v, want receipt %s", receipts, hash) + } +} + +func TestReceiptFetcherPreservesMaticNotFoundPosition(t *testing.T) { + hashes := []common.Hash{common.HexToHash("0x1"), common.HexToHash("0x2")} + reader := receiptReaderFunc(func(_ context.Context, hash common.Hash) (*types.Receipt, error) { + if hash == hashes[0] { + return nil, ethereum.NotFound + } + return &types.Receipt{TxHash: hash}, nil + }) + + receipts, err := NewReceiptFetcher(reader, PreserveNotFound).Fetch(context.Background(), hashes) + if err != nil { + t.Fatalf("ReceiptFetcher.Fetch returned error: %v", err) + } + if len(receipts) != len(hashes) || receipts[0] != nil { + t.Fatalf("Matic missing receipt position was not preserved: %+v", receipts) + } + if receipts[1] == nil || receipts[1].TxHash != hashes[1] { + t.Fatalf("receipt 1 = %+v, want hash %s", receipts[1], hashes[1]) + } +} + +func TestReceiptFetcherStopsRetryWhenContextCanceled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + called := make(chan struct{}, 1) + reader := receiptReaderFunc(func(ctx context.Context, _ common.Hash) (*types.Receipt, error) { + called <- struct{}{} + if err := ctx.Err(); err != nil { + return nil, err + } + return nil, ethereum.NotFound + }) + + done := make(chan error, 1) + go func() { + _, err := NewReceiptFetcher(reader, RetryNotFound).Fetch(ctx, []common.Hash{{1}}) + done <- err + }() + <-called + cancel() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("ReceiptFetcher.Fetch error = %v, want context.Canceled", err) + } + case <-time.After(25 * time.Millisecond): + err := <-done + t.Fatalf("ReceiptFetcher.Fetch did not stop promptly after cancellation; eventual error: %v", err) + } +} diff --git a/pkg/ethclient/ethclient.go b/pkg/ethclient/ethclient.go index 14014f58..59235ea1 100644 --- a/pkg/ethclient/ethclient.go +++ b/pkg/ethclient/ethclient.go @@ -622,7 +622,12 @@ type OpReceipt struct { func (ec *Client) OpReceipt(ctx context.Context, txHash common.Hash) (*OpReceipt, error) { s := fmt.Sprintf("{\"jsonrpc\": \"2.0\",\"method\": \"eth_getTransactionReceipt\",\"params\": [\"%s\"],\"id\": 1}", txHash.Hex()) body := strings.NewReader(s) - resp, err := ec.cli.Post(ec.url, "application/json", body) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, ec.url, body) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := ec.cli.Do(req) if err != nil { return nil, err } @@ -632,6 +637,16 @@ func (ec *Client) OpReceipt(ctx context.Context, txHash common.Hash) (*OpReceipt if err := json.NewDecoder(resp.Body).Decode(&respmsg); err != nil { return nil, err } + if len(respmsg.Error) > 0 && string(respmsg.Error) != "null" { + var rpcErr struct { + Code int `json:"code"` + Message string `json:"message"` + } + if err := json.Unmarshal(respmsg.Error, &rpcErr); err != nil { + return nil, fmt.Errorf("eth_getTransactionReceipt RPC error: %s", respmsg.Error) + } + return nil, fmt.Errorf("eth_getTransactionReceipt RPC error %d: %s", rpcErr.Code, rpcErr.Message) + } data := make([]byte, 0, len(respmsg.Result)) for _, res := range respmsg.Result { diff --git a/pkg/ethclient/ethclient_test.go b/pkg/ethclient/ethclient_test.go new file mode 100644 index 00000000..f5ebc4a3 --- /dev/null +++ b/pkg/ethclient/ethclient_test.go @@ -0,0 +1,68 @@ +package ethclient + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" +) + +func TestOpReceiptHonorsContext(t *testing.T) { + requestStarted := make(chan struct{}) + releaseRequest := make(chan struct{}) + var releaseOnce sync.Once + release := func() { + releaseOnce.Do(func() { close(releaseRequest) }) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(requestStarted) + select { + case <-r.Context().Done(): + return + case <-releaseRequest: + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{}}`)) + } + })) + defer server.Close() + defer release() + + client := NewClient(nil, server.URL, server.Client()) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := client.OpReceipt(ctx, common.Hash{}) + done <- err + }() + + <-requestStarted + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("OpReceipt error = %v, want context.Canceled", err) + } + case <-time.After(25 * time.Millisecond): + release() + <-done + t.Fatal("OpReceipt did not stop after context cancellation") + } +} + +func TestOpReceiptReturnsJSONRPCError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"receipt unavailable"}}`)) + })) + defer server.Close() + + client := NewClient(nil, server.URL, server.Client()) + _, err := client.OpReceipt(context.Background(), common.Hash{}) + if err == nil || !strings.Contains(err.Error(), "receipt unavailable") { + t.Fatalf("OpReceipt error = %v, want JSON-RPC error message", err) + } +} diff --git a/tests/update_near_header_map_test.go b/tests/update_near_header_map_test.go deleted file mode 100644 index 3b97eede..00000000 --- a/tests/update_near_header_map_test.go +++ /dev/null @@ -1,21 +0,0 @@ -package tests - -import ( - "github.com/lbtsm/gotron-sdk/pkg/client" - "github.com/stretchr/testify/require" - "google.golang.org/grpc" - "testing" -) - -func Test_grpc(t *testing.T) { - conn := client.NewGrpcClient("grpc.trongrid.io:50051") - err := conn.Start(grpc.WithInsecure()) - require.Nil(t, err) - - // `[{"bytes32":"1eee75d90926c3470877c4da9c21d52c6e762225fec1386f7a526bf3f1ce440e"}]` - tx, err := conn.TriggerConstantContract("TNoZuAuL83PSh8TG4W92AvLqkA4E2dKSNm", - "TYMpgB8Q9vSoGtkyE3hXsvUrpte3KCDGj6", - "orderList(bytes32)", `[{"bytes32":"[30 238 117 217 9 38 195 71 8 119 196 218 156 33 213 44 110 118 34 37 254 193 56 111 122 82 107 243 241 206 68 14]"}]`) - t.Log("err", err) - t.Log("tx", tx.ConstantResult[0]) -}