From dd54ad4bb6f9db1309f5f13924209d55b61dbf7d Mon Sep 17 00:00:00 2001 From: corver Date: Mon, 25 Nov 2024 12:18:38 +0200 Subject: [PATCH] chore(solver/app): start event streamers --- e2e/app/definition.go | 2 - e2e/app/setup.go | 8 +- lib/ethclient/ethbackend/backends.go | 34 +++++++ solver/app/app.go | 138 +++++++++++++++++---------- solver/app/bindings.go | 1 - solver/app/cursor.go | 1 - solver/app/definition.go | 12 +-- solver/app/deps.go | 1 - solver/app/helpers.go | 33 +++++++ solver/app/targets.go | 1 - 10 files changed, 160 insertions(+), 71 deletions(-) create mode 100644 solver/app/helpers.go diff --git a/e2e/app/definition.go b/e2e/app/definition.go index 1dfbaa6f8..7556b52de 100644 --- a/e2e/app/definition.go +++ b/e2e/app/definition.go @@ -188,8 +188,6 @@ func newBackends(ctx context.Context, cfg DefinitionConfig, testnet types.Testne return ethbackend.Backends{}, errors.Wrap(err, "new fireblocks") } - // TODO(corver): Fireblocks keys need to be funded on private/internal chains we deploy. - return ethbackend.NewFireBackends(ctx, testnet, fireCl) } diff --git a/e2e/app/setup.go b/e2e/app/setup.go index 13595d3d0..cd55f0f35 100644 --- a/e2e/app/setup.go +++ b/e2e/app/setup.go @@ -498,7 +498,7 @@ func writeRelayerConfig(ctx context.Context, def Definition, logCfg log.Config) return nil } -func writeSolverConfig(_ context.Context, def Definition, logCfg log.Config) error { +func writeSolverConfig(ctx context.Context, def Definition, logCfg log.Config) error { confRoot := filepath.Join(def.Testnet.Dir, "solver") const ( @@ -516,10 +516,10 @@ func writeSolverConfig(_ context.Context, def Definition, logCfg log.Config) err endpoints = ExternalEndpoints(def) } - // TODO(corver): save proper private key - privKey, err := ethcrypto.GenerateKey() + // Save private key + privKey, err := eoa.PrivateKey(ctx, def.Testnet.Network, eoa.RoleSolver) if err != nil { - return errors.Wrap(err, "generate private key") + return errors.Wrap(err, "get relayer key") } if err := ethcrypto.SaveECDSA(filepath.Join(confRoot, privKeyFile), privKey); err != nil { return errors.Wrap(err, "write private key") diff --git a/lib/ethclient/ethbackend/backends.go b/lib/ethclient/ethbackend/backends.go index 77db10553..8ad6b4c00 100644 --- a/lib/ethclient/ethbackend/backends.go +++ b/lib/ethclient/ethbackend/backends.go @@ -11,6 +11,7 @@ import ( "github.com/omni-network/omni/lib/fireblocks" "github.com/omni-network/omni/lib/netconf" "github.com/omni-network/omni/lib/txmgr" + "github.com/omni-network/omni/lib/xchain" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" @@ -82,6 +83,30 @@ func NewFireBackends(ctx context.Context, testnet types.Testnet, fireCl firebloc }, nil } +func BackendsFromNetwork(network netconf.Network, endpoints xchain.RPCEndpoints, privKeys ...*ecdsa.PrivateKey) (Backends, error) { + inner := make(map[uint64]*Backend) + for _, chain := range network.EVMChains() { + endpoint, err := endpoints.ByNameOrID(chain.Name, chain.ID) + if err != nil { + return Backends{}, err + } + + ethCl, err := ethclient.Dial(chain.Name, endpoint) + if err != nil { + return Backends{}, errors.Wrap(err, "dial") + } + + inner[chain.ID], err = NewBackend(chain.Name, chain.ID, chain.BlockPeriod, ethCl, privKeys...) + if err != nil { + return Backends{}, errors.Wrap(err, "new backend") + } + } + + return Backends{ + backends: inner, + }, nil +} + // NewBackends returns a multi-backends backed by in-memory keys that supports configured all chains. func NewBackends(ctx context.Context, testnet types.Testnet, deployKeyFile string) (Backends, error) { var err error @@ -164,6 +189,15 @@ func (b Backends) All() map[uint64]*Backend { return b.backends } +func (b Backends) Clients() map[uint64]ethclient.Client { + clients := make(map[uint64]ethclient.Client) + for chainID, backend := range b.backends { + clients[chainID] = backend.Client + } + + return clients +} + func (b Backends) Backend(sourceChainID uint64) (*Backend, error) { backend, ok := b.backends[sourceChainID] if !ok { diff --git a/solver/app/app.go b/solver/app/app.go index f7272addc..334d938b0 100644 --- a/solver/app/app.go +++ b/solver/app/app.go @@ -1,4 +1,3 @@ -//nolint:unused // It will be used in PRs. package app import ( @@ -10,6 +9,7 @@ import ( "github.com/omni-network/omni/contracts/bindings" "github.com/omni-network/omni/halo/genutil/evm/predeploys" "github.com/omni-network/omni/lib/buildinfo" + "github.com/omni-network/omni/lib/contracts" "github.com/omni-network/omni/lib/errors" "github.com/omni-network/omni/lib/ethclient" "github.com/omni-network/omni/lib/ethclient/ethbackend" @@ -19,7 +19,9 @@ import ( "github.com/omni-network/omni/lib/xchain" xprovider "github.com/omni-network/omni/lib/xchain/provider" + "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" + ethcrypto "github.com/ethereum/go-ethereum/crypto" "github.com/prometheus/client_golang/prometheus/promhttp" ) @@ -50,17 +52,37 @@ func Run(ctx context.Context, cfg Config) error { return err } - ethClients, err := initializeEthClients(network.EVMChains(), cfg.RPCEndpoints) + if cfg.PrivateKey == "" { + return errors.New("private key not set") + } + privKey, err := ethcrypto.LoadECDSA(cfg.PrivateKey) + if err != nil { + return errors.Wrap(err, "load private key") + } + solverAddr := ethcrypto.PubkeyToAddress(privKey.PublicKey) + log.Debug(ctx, "Using solver address", "address", solverAddr.Hex()) + + backends, err := ethbackend.BackendsFromNetwork(network, cfg.RPCEndpoints, privKey) if err != nil { return err } - _, err = newSolverDB(cfg.DBDir) + xprov := xprovider.New(network, backends.Clients(), nil) + + db, err := newSolverDB(cfg.DBDir) if err != nil { return err } - _ = xprovider.New(network, ethClients, nil) + cursors, err := newCursors(db) + if err != nil { + return errors.Wrap(err, "create cursor store") + } + + err = startEventStreams(ctx, network, xprov, backends, solverAddr, cursors) + if err != nil { + return errors.Wrap(err, "start event streams") + } select { case <-ctx.Done(): @@ -119,68 +141,82 @@ func makePortalRegistry(network netconf.ID, endpoints xchain.RPCEndpoints) (*bin return resp, nil } -// initializeEthClients initializes the RPC clients for the given chains. -func initializeEthClients(chains []netconf.Chain, endpoints xchain.RPCEndpoints) (map[uint64]ethclient.Client, error) { - rpcClientPerChain := make(map[uint64]ethclient.Client) - for _, chain := range chains { - rpc, err := endpoints.ByNameOrID(chain.Name, chain.ID) - if err != nil { - return nil, err - } - c, err := ethclient.Dial(chain.Name, rpc) - if err != nil { - return nil, errors.Wrap(err, "dial rpc", "chain_name", chain.Name, "chain_id", chain.ID, "rpc_url", rpc) - } - rpcClientPerChain[chain.ID] = c - } - - return rpcClientPerChain, nil -} - +// startEventStreams starts the event streams for the solver. +// TODO(corver): Make this robust against chains not be available on startup. func startEventStreams( ctx context.Context, network netconf.Network, xprov xchain.Provider, backends ethbackend.Backends, - def Definition, solverAddr common.Address, cursors *cursors, ) error { + addrs, err := contracts.GetAddresses(ctx, network.ID) + if err != nil { + return errors.Wrap(err, "get contract addresses") + } + + inboxChains, err := detectContractChains(ctx, network, backends, addrs.SolveInbox) + if err != nil { + return errors.Wrap(err, "detect inbox chains") + } + inboxContracts := make(map[uint64]*bindings.SolveInbox) - outboxContracts := make(map[uint64]*bindings.SolveOutbox) - for _, chain := range network.EVMChains() { - // Maybe init cursor store with deploy heights - chainVer := chainVerFromID(chain.ID) + for _, chain := range inboxChains { + name := network.ChainName(chain) + chainVer := chainVerFromID(chain) + log.Debug(ctx, "Using inbox contract", "chain", name, "address", addrs.SolveInbox.Hex()) + + backend, err := backends.Backend(chain) + if err != nil { + return err + } + + inbox, err := bindings.NewSolveInbox(addrs.SolveInbox, backend) + if err != nil { + return errors.Wrap(err, "create inbox contract", "chain", name) + } + inboxContracts[chain] = inbox + + // Check if cursor store should be initialized with deploy height if _, ok, err := cursors.Get(ctx, chainVer); err != nil { - return errors.Wrap(err, "get cursor", "chain_id", chain.ID) - } else if !ok { - height, ok := def.InboxDeployHeights[chain.ID] - if !ok { - return errors.New("missing inbox deploy height", "chain_id", chain.ID) - } - err := cursors.Set(ctx, chainVer, height) - if err != nil { - return err - } + return errors.Wrap(err, "get cursor", "chain", name) + } else if ok { // Cursor already set, skip + continue } - backend, err := backends.Backend(chain.ID) + height, err := inbox.DeployedAt(&bind.CallOpts{Context: ctx}) if err != nil { + return errors.New("get inbox deploy height", "chain", name) + } + + log.Info(ctx, "Initializing inbox cursor", "chain", name, "deployed_at", height) + + if err := cursors.Set(ctx, chainVer, height.Uint64()); err != nil { return err } + } - inbox, err := bindings.NewSolveInbox(def.InboxAddress, backend) + outboxChains, err := detectContractChains(ctx, network, backends, addrs.SolveOutbox) + if err != nil { + return errors.Wrap(err, "detect outbox chains") + } + + outboxContracts := make(map[uint64]*bindings.SolveOutbox) + for _, chain := range outboxChains { + name := network.ChainName(chain) + log.Debug(ctx, "Using outbox contract", "chain", name, "address", addrs.SolveInbox.Hex()) + + backend, err := backends.Backend(chain) if err != nil { - return errors.Wrap(err, "create inbox contract", "chain_id", chain.ID) + return err } - outbox, err := bindings.NewSolveOutbox(def.OutboxAddress, backend) + outbox, err := bindings.NewSolveOutbox(addrs.SolveOutbox, backend) if err != nil { - return errors.Wrap(err, "create outbox contract", "chain_id", chain.ID) + return errors.Wrap(err, "create outbox contract", "chain", name) } - - inboxContracts[chain.ID] = inbox - outboxContracts[chain.ID] = outbox + outboxContracts[chain] = outbox } cursorSetter := func(ctx context.Context, chainID uint64, height uint64) error { @@ -190,7 +226,7 @@ func startEventStreams( deps := procDeps{ ParseID: newIDParser(inboxContracts), GetRequest: newRequestGetter(inboxContracts), - ShouldReject: newRequestValidator(def), + ShouldReject: newRequestValidator(), Accept: newAcceptor(network.ID, inboxContracts, backends, solverAddr), Reject: newRejector(inboxContracts, backends, solverAddr), Fulfill: newFulfiller(outboxContracts, backends, solverAddr), @@ -198,20 +234,22 @@ func startEventStreams( SetCursor: cursorSetter, } - for _, chain := range network.EVMChains() { - go streamEventsForever(ctx, chain.ID, xprov, deps, def, cursors) + for _, chain := range inboxChains { + log.Info(ctx, "Starting inbox event stream", "chain", network.ChainName(chain)) + go streamEventsForever(ctx, chain, xprov, deps, cursors, addrs.SolveInbox) } return nil } +// streamEventsForever streams events from the inbox contract on the given chain. func streamEventsForever( ctx context.Context, chainID uint64, xprov xchain.Provider, deps procDeps, - def Definition, cursors *cursors, + inboxAddr common.Address, ) { backoff := expbackoff.New(ctx, expbackoff.WithPeriodicConfig(time.Second*5)) for { @@ -227,7 +265,7 @@ func streamEventsForever( ChainID: chainID, Height: from, ConfLevel: confLevel, - FilterAddress: def.InboxAddress, + FilterAddress: inboxAddr, FilterTopics: allEventTopics, } err = xprov.StreamEventLogs(ctx, req, newEventProcessor(deps, chainID)) diff --git a/solver/app/bindings.go b/solver/app/bindings.go index 9fe1a21d6..0a30ed5dc 100644 --- a/solver/app/bindings.go +++ b/solver/app/bindings.go @@ -83,7 +83,6 @@ var ( return resp }() - //nolint:unused // False positive allEventTopics = func() []common.Hash { resp := make([]common.Hash, 0, len(allEvents)) for _, e := range allEvents { diff --git a/solver/app/cursor.go b/solver/app/cursor.go index fdcaaab13..e79c89ec6 100644 --- a/solver/app/cursor.go +++ b/solver/app/cursor.go @@ -1,4 +1,3 @@ -//nolint:unused // It will be used package app import ( diff --git a/solver/app/definition.go b/solver/app/definition.go index 5ee885813..555b33295 100644 --- a/solver/app/definition.go +++ b/solver/app/definition.go @@ -5,19 +5,9 @@ import ( "github.com/omni-network/omni/contracts/bindings" "github.com/omni-network/omni/lib/errors" - - "github.com/ethereum/go-ethereum/common" ) -// Definition defines the solver rules and addresses and config per network. -type Definition struct { - InboxAddress common.Address - OutboxAddress common.Address - InboxDeployHeights map[uint64]uint64 -} - -//nolint:unused // False positive. -func newRequestValidator(_ Definition) func(ctx context.Context, chainID uint64, req bindings.SolveRequest) (uint8, bool, error) { +func newRequestValidator() func(ctx context.Context, chainID uint64, req bindings.SolveRequest) (uint8, bool, error) { return func(context.Context, uint64, bindings.SolveRequest) (uint8, bool, error) { return 0, false, errors.New("not implemented") } diff --git a/solver/app/deps.go b/solver/app/deps.go index f01b41480..e2ab0aaf6 100644 --- a/solver/app/deps.go +++ b/solver/app/deps.go @@ -1,4 +1,3 @@ -//nolint:unused // Some functions are unused but are kept for future use package app import ( diff --git a/solver/app/helpers.go b/solver/app/helpers.go new file mode 100644 index 000000000..a80be16dc --- /dev/null +++ b/solver/app/helpers.go @@ -0,0 +1,33 @@ +package app + +import ( + "context" + + "github.com/omni-network/omni/lib/errors" + "github.com/omni-network/omni/lib/ethclient/ethbackend" + "github.com/omni-network/omni/lib/netconf" + + "github.com/ethereum/go-ethereum/common" +) + +// detectContractChains returns the chains on which the contract is deployed at the provided address. +func detectContractChains(ctx context.Context, network netconf.Network, backends ethbackend.Backends, address common.Address) ([]uint64, error) { + var resp []uint64 + for _, chain := range network.EVMChains() { + backend, err := backends.Backend(chain.ID) + if err != nil { + return nil, err + } + + code, err := backend.CodeAt(ctx, address, nil) + if err != nil { + return nil, errors.Wrap(err, "get code", "chain", chain.Name) + } else if len(code) == 0 { + continue + } + + resp = append(resp, chain.ID) + } + + return resp, nil +} diff --git a/solver/app/targets.go b/solver/app/targets.go index 892358589..8b3f6e804 100644 --- a/solver/app/targets.go +++ b/solver/app/targets.go @@ -1,4 +1,3 @@ -//nolint:unused // Some functions are unused but are kept for future use package app import (