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 f255e0ea..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) { @@ -48,7 +50,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 @@ -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/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/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/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/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/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 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..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.OnlineChaId[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/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() +} 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]) -}