diff --git a/go.mod b/go.mod index 4351d9b..f77acd4 100644 --- a/go.mod +++ b/go.mod @@ -102,6 +102,7 @@ require ( github.com/koron/go-ssdp v0.0.5 // indirect github.com/kr/pretty v0.3.1 // indirect github.com/kr/text v0.2.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/libp2p/go-buffer-pool v0.1.0 // indirect github.com/libp2p/go-cidranger v1.1.0 // indirect github.com/libp2p/go-flow-metrics v0.2.0 // indirect diff --git a/internal/router/router.go b/internal/router/router.go index e7fd6d9..6a7b466 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -13,6 +13,7 @@ import ( "github.com/shutter-network/shutter-api/docs" "github.com/shutter-network/shutter-api/internal/middleware" "github.com/shutter-network/shutter-api/internal/service" + "github.com/shutter-network/shutter-api/internal/usecase" swaggerFiles "github.com/swaggo/files" ginSwagger "github.com/swaggo/gin-swagger" ) @@ -22,6 +23,7 @@ func NewRouter( db *pgxpool.Pool, contract *common.Contract, ethClient *ethclient.Client, + txManager usecase.TxManagerInterface, config *common.Config, ) *gin.Engine { router := gin.New() @@ -30,7 +32,7 @@ func NewRouter( router.Use(cors.Default()) router.Use(middleware.ErrorHandler()) - cryptoService := service.NewCryptoService(db, contract, ethClient, config) + cryptoService := service.NewCryptoService(db, contract, ethClient, txManager, config) docs.SwaggerInfo.BasePath = "/api" api := router.Group("/api") { diff --git a/internal/service/crypto.go b/internal/service/crypto.go index ed529f4..f36cb58 100644 --- a/internal/service/crypto.go +++ b/internal/service/crypto.go @@ -32,10 +32,11 @@ func NewCryptoService( db *pgxpool.Pool, contract *common.Contract, ethClient *ethclient.Client, + txManager usecase.TxManagerInterface, config *common.Config, ) *CryptoService { return &CryptoService{ - CryptoUsecase: usecase.NewCryptoUsecase(db, contract.ShutterRegistryContract, contract.ShutterEventRegistryContract, contract.KeyperSetManagerContract, contract.KeyBroadcastContract, ethClient, config), + CryptoUsecase: usecase.NewCryptoUsecase(db, contract.ShutterRegistryContract, contract.ShutterEventRegistryContract, contract.KeyperSetManagerContract, contract.KeyBroadcastContract, ethClient, txManager, config), } } @@ -233,6 +234,7 @@ func (svc *CryptoService) GetDataForEncryptionEvent(ctx *gin.Context) { // @Failure 400 {object} error.Http "Invalid Register identity request." // @Failure 429 {object} error.Http "Too many requests. Rate limited." // @Failure 500 {object} error.Http "Internal server error." +// @Failure 503 {object} error.Http "Too many registrations in flight. Retry." // @Security BearerAuth // @Router /time/register_identity [post] func (svc *CryptoService) RegisterIdentity(ctx *gin.Context) { @@ -405,6 +407,7 @@ func CompileEventTriggerDefinition(ctx *gin.Context) { // @Failure 429 {object} error.Http "Too many requests. Rate limited." // @Failure 500 {object} error.Http "Internal server error." // @Failure 501 {object} error.Http "Event API is disabled on this deployment." +// @Failure 503 {object} error.Http "Too many registrations in flight. Retry." // @Security BearerAuth // @Router /event/register_identity [post] func (svc *CryptoService) RegisterEventIdentity(ctx *gin.Context) { diff --git a/internal/txmgr/manager.go b/internal/txmgr/manager.go new file mode 100644 index 0000000..04d0b87 --- /dev/null +++ b/internal/txmgr/manager.go @@ -0,0 +1,726 @@ +// Package txmgr serializes transaction submission for a single signer account. +// +// Every registration is signed by the same key, so all of them compete for one +// nonce sequence. Left to the generated bindings, a nil TransactOpts.Nonce makes +// each call resolve its own nonce, so two concurrent requests sign two +// transactions with the same one and only one of them can be mined. +// +// The Manager closes that window structurally rather than with locks: Send only +// queues a request, and a single goroutine owns nonce assignment, submission and +// everything in flight. That goroutine also polls for receipts, so this package +// holds no mutexes. +// +// A request is watched until it is mined, and if something else mines its nonce +// first it is carried over to a free one rather than failed. A transaction that +// goes RebroadcastAfter without being mined is sent again, priced to outbid the +// version it replaces, which doubles as gas bumping. Recomputing from the node's +// suggestion alone would not do: geth demands a bump on the tip as well as the +// fee cap, and the tip oracle does not move when the base fee does, so a +// transaction stranded below a risen base fee could never be replaced. +// MaxFeeCapPercent bounds how far that escalation goes. +// +// # Known limits +// +// Pending state is in memory and lost on restart. An account with no gas money +// is watched indefinitely and wedges registration, which shows up in +// pending_transactions and signer_balance_ether and clears once it is refilled. +// A transaction sent from this key by anything other than this process will +// collide with ours. Carrying a request over abandons a transaction that cannot +// be proven dead, so if the old one is still included the request lands twice: +// the time registry reverts the second with AlreadyRegistered, the event +// registry has no such error and registers the identity twice. +package txmgr + +import ( + "context" + "crypto/ecdsa" + "errors" + "fmt" + "math/big" + "slices" + "strings" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi/bind" + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/rs/zerolog/log" + "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service" + "github.com/shutter-network/shutter-api/metrics" +) + +const ( + // basefeeWiggleMultiplier mirrors the constant of the same name in + // accounts/abi/bind/v2/base.go, so that owning gas pricing here changes who + // calculates it without changing what gets paid. + basefeeWiggleMultiplier = 2 + + // priceBumpPercent is what geth's pool demands of a replacement, from + // core/txpool/legacypool.DefaultConfig.PriceBump. Both the tip and the fee + // cap must clear it, and both must be strictly greater than the incumbent's + // (core/txpool/legacypool/list.go, list.Add). + priceBumpPercent = 110 +) + +// Client is the subset of ethclient.Client the Manager needs. +type Client interface { + NonceAt(ctx context.Context, account ecommon.Address, blockNumber *big.Int) (uint64, error) + TransactionReceipt(ctx context.Context, txHash ecommon.Hash) (*types.Receipt, error) + SuggestGasTipCap(ctx context.Context) (*big.Int, error) + HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) +} + +// SubmitFunc signs and sends one contract call, which is what a generated +// binding method does. The Manager supplies opts with From, Signer, Nonce and +// both fee caps already set, so the closure only binds the call's own arguments +// and stays independent of any particular contract. +// +// It is invoked again for every resubmission, so it must be safe to call +// repeatedly. Note that it sends a transaction; it is not an eth_call. +type SubmitFunc func(opts *bind.TransactOpts) (*types.Transaction, error) + +// The Manager's own reasons for ending a request. Any other error on an Err +// event comes from the node or from reading the chain. +// +// They are distinguished because they mean different things to a caller deciding +// what to tell its own client: the first two are proof that nothing was signed, +// while the third is not. +var ( + // ErrQueueFull means the worker is not keeping up and the request was turned + // away without being signed. + ErrQueueFull = errors.New("transaction queue is full") + // ErrShutdown means the Manager stopped while the request was still queued. + // Nothing was signed, so nothing can land. + ErrShutdown = errors.New("transaction manager stopped before submitting") + // ErrAbandoned means the Manager stopped while watching a transaction it had + // already sent. That transaction is in the pool and may still be mined, so + // the request cannot be reported as having failed, only as unknown. + ErrAbandoned = errors.New("transaction abandoned while awaiting a receipt") +) + +// Event is one thing that happened to a send request. Exactly one field is set. +// +// A caller that only needs something to report can take the first event and walk +// away. A caller that needs the outcome reads to the end, since Receipt and Err +// are terminal: one of them is always the last event, and the channel is closed +// after it. +type Event struct { + // Tx is the transaction being watched on the request's behalf from now on. It + // arrives once for the first submission and again for every resubmission that + // replaces it, so the newest is the one to poll for and the earlier ones are + // history rather than mistakes: a replacement should evict its predecessor, + // but propagation is not atomic and any version may still be the one mined. + Tx *types.Transaction + // Receipt is the receipt of whichever version was mined. Whether the call + // itself succeeded is in Receipt.Status: that is the contract's verdict rather + // than the Manager's, and a caller holding the receipt can read it. TxHash + // names the version that won, which need not be the last one announced. + Receipt *types.Receipt + // Err is the reason the Manager stopped working on the request. Every error + // but ErrAbandoned means no transaction of ours can be mined; that one means + // the Manager stopped looking rather than that anything failed. + Err error +} + +// eventBufferSize is how many events a request may hold for a caller that has +// stopped reading. Generous because a request produces one event per +// resubmission and those arrive for as long as it goes unmined, and cheap +// because the buffer only ever holds what a caller has not taken yet. +const eventBufferSize = 64 + +// Config tunes the Manager. +type Config struct { + // PollInterval is how often pending transactions are checked for a receipt. + PollInterval time.Duration + // RebroadcastAfter is how long a transaction may go without a submission + // attempt before it is sent again. It is an age per transaction, not a + // period: a freshly submitted transaction is never resubmitted just because + // an older one was due, which would replace a healthy transaction with a + // more expensive one for no reason. + RebroadcastAfter time.Duration + // QueueSize bounds how many requests may wait for submission. Reaching it + // means the worker is not keeping up, which is worth reporting rather than + // absorbing. + QueueSize int + // MaxFeeCapPercent bounds what a resubmission may pay, as a percentage of + // the fee cap the node currently suggests. 200 means never more than twice + // the going rate. Capping the fee cap is enough to bound the spend, because + // it also bounds the tip actually paid. + // + // A resubmission that cannot outbid the incumbent within it is skipped, and + // the transaction waits for fees to fall. Without a ceiling the escalation + // below would compound without limit. + MaxFeeCapPercent uint64 +} + +// DefaultConfig polls once per block at Gnosis' five second block time. +func DefaultConfig() Config { + return Config{ + PollInterval: 5 * time.Second, + RebroadcastAfter: time.Minute, + QueueSize: 1024, + MaxFeeCapPercent: 200, + } +} + +// sendRequest represents a caller's request to send a transaction. It stores a +// channel to report what happens to it. +type sendRequest struct { + submit SubmitFunc + events chan Event +} + +// publish announces an intermediate event, and drops it if the caller has left +// the buffer full. Dropping rather than blocking because this runs on the worker, +// and one caller that stopped reading must not be able to stall nonce assignment +// for every other request. +// +// The last slot is reserved so that finish always has room, which is what makes +// the terminal event the one thing a caller cannot miss. +func (r sendRequest) publish(ev Event) { + if len(r.events) >= cap(r.events)-1 { + log.Warn().Int("buffer", cap(r.events)). + Msg("dropping transaction event, caller is not reading its channel") + return + } + r.events <- ev +} + +// finish announces the terminal event and closes the channel. The send cannot +// block: publish never occupies the last slot, and this is the last event, so +// nothing follows it into the buffer. +func (r sendRequest) finish(ev Event) { + r.events <- ev + close(r.events) +} + +// pendingTransaction is a submitted request awaiting the chain's verdict. It +// outlives any single transaction, because a resubmission replaces the +// transaction while the request stays the same. +type pendingTransaction struct { + sendRequest + // txs is every version submitted for this request, oldest first. A + // replacement should evict its predecessor, but propagation is not atomic + // across nodes, so an earlier version can still be the one that gets mined. + // All of them are watched, otherwise that would look like a stranger taking + // our nonce and the request would be carried over and land twice. + txs []*types.Transaction + // lastAttempt is when the transaction was last sent, successfully or not. + // Failed attempts count, otherwise a rejected resubmission would be retried + // on every poll instead of once per RebroadcastAfter. + lastAttempt time.Time +} + +// tx is the version submitted most recently, and the one whose nonce counts. +func (p *pendingTransaction) tx() *types.Transaction { + return p.txs[len(p.txs)-1] +} + +func (p *pendingTransaction) nonce() uint64 { + return p.tx().Nonce() +} + +// Manager assigns nonces for one account and watches what it submits. +type Manager struct { + client Client + from ecommon.Address + signer bind.SignerFn + cfg Config + requests chan sendRequest + + // Owned exclusively by the goroutine Start launches. Not guarded, because + // nothing else touches it. + // + // A queue in nonce order rather than a map, because only the first can be + // mined next and that is the only entry ever acted on. It stays ordered by + // construction: new sends take the next nonce, and a carried-over request + // takes one above the last. + pending []*pendingTransaction +} + +// NewManager builds a Manager for the given signing key. The chain ID is read +// once here rather than per request. +func NewManager(client Client, signingKey *ecdsa.PrivateKey, chainID *big.Int, cfg Config) (*Manager, error) { + signer, err := bind.NewKeyedTransactorWithChainID(signingKey, chainID) + if err != nil { + return nil, err + } + defaults := DefaultConfig() + if cfg.PollInterval <= 0 { + cfg.PollInterval = defaults.PollInterval + } + if cfg.RebroadcastAfter <= 0 { + cfg.RebroadcastAfter = defaults.RebroadcastAfter + } + if cfg.QueueSize <= 0 { + cfg.QueueSize = defaults.QueueSize + } + if cfg.MaxFeeCapPercent <= 100 { + cfg.MaxFeeCapPercent = defaults.MaxFeeCapPercent + } + return &Manager{ + client: client, + from: signer.From, + signer: signer.Signer, + cfg: cfg, + requests: make(chan sendRequest, cfg.QueueSize), + }, nil +} + +// From is the address every transaction is sent from. Identities are derived +// from it, so callers need it even when they are not sending. +func (m *Manager) From() ecommon.Address { + return m.from +} + +// Send queues a call and returns immediately, without making any RPC calls. +// Nothing has been sent yet when it returns, so neither a transaction hash nor a +// submission error exists until the first Event arrives. +// +// The returned channel reports every transaction sent for the request and ends +// with either a receipt or an error, whatever happens, so a caller waiting for +// the outcome is never left waiting forever. It is buffered, so a caller that +// takes what it needs and stops reading costs nothing; events beyond the buffer +// are dropped rather than delaying the worker, except the terminal one, which is +// always kept. +func (m *Manager) Send(submit SubmitFunc) <-chan Event { + req := sendRequest{submit: submit, events: make(chan Event, eventBufferSize)} + + select { + case m.requests <- req: + default: + m.reject(req, fmt.Errorf("%w (size %d)", ErrQueueFull, m.cfg.QueueSize)) + } + return req.events +} + +// Start runs the Manager until ctx is cancelled. It satisfies the +// rolling-shutter service interface. +// +// Submission and polling share one goroutine deliberately: that is what makes +// nonce assignment safe without locking, and a submission delaying a receipt +// check by a few hundred milliseconds costs nothing, since polling is only +// observation. +func (m *Manager) Start(ctx context.Context, runner service.Runner) error { + runner.Go(func() error { + ticker := time.NewTicker(m.cfg.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + m.shutdown() + return ctx.Err() + case req := <-m.requests: + m.submit(ctx, req) + case <-ticker.C: + m.poll(ctx) + } + } + }) + return nil +} + +// submit assigns a nonce, sends one queued request and starts watching it. +func (m *Manager) submit(ctx context.Context, req sendRequest) { + nonce, err := m.assignNonce(ctx) + if err != nil { + m.reject(req, fmt.Errorf("failed to determine the next nonce: %w", err)) + return + } + + tip, feeCap, err := m.gasPrices(ctx) + if err != nil { + m.reject(req, fmt.Errorf("failed to calculate gas prices: %w", err)) + return + } + + tx, err := req.submit(m.transactOpts(ctx, nonce, tip, feeCap)) + if err != nil { + // Nothing is tracked, so the nonce this request was given goes to the + // next one instead of leaving a hole that later transactions would queue + // behind forever. Safe without any interlock because this goroutine is + // the only one that assigns nonces. + m.reject(req, fmt.Errorf("failed to submit transaction: %w", err)) + return + } + + m.pending = append(m.pending, &pendingTransaction{ + sendRequest: req, + txs: []*types.Transaction{tx}, + lastAttempt: time.Now(), + }) + req.publish(Event{Tx: tx}) + + metrics.PendingTransactions.Set(float64(len(m.pending))) + log.Info().Str("tx_hash", tx.Hash().Hex()).Uint64("nonce", nonce). + Msg("transaction submitted") +} + +// assignNonce returns the nonce for the next submission: one above the last +// queued transaction, or the account's mined nonce when nothing is in flight. +// +// NonceAt rather than PendingNonceAt on purpose. After a restart our own +// transactions may still sit in the pool with nothing left to resubmit them, and +// deferring to them would queue everything new behind a blocker that no longer +// has anyone escalating it. Taking the mined frontier adopts that nonce and +// drives it forward instead. +func (m *Manager) assignNonce(ctx context.Context) (uint64, error) { + if len(m.pending) > 0 { + return m.lastPendingTransaction().nonce() + 1, nil + } + + nonce, err := m.client.NonceAt(ctx, m.from, nil) + if err != nil { + metrics.FailedRPCCalls.Inc() + return 0, err + } + return nonce, nil +} + +// gasPrices reproduces what the bindings compute when the fee fields are left +// nil: the suggested tip, and a cap covering a doubling of the base fee. +func (m *Manager) gasPrices(ctx context.Context) (tip, feeCap *big.Int, err error) { + tip, err = m.client.SuggestGasTipCap(ctx) + if err != nil { + metrics.FailedRPCCalls.Inc() + return nil, nil, err + } + head, err := m.client.HeaderByNumber(ctx, nil) + if err != nil { + metrics.FailedRPCCalls.Inc() + return nil, nil, err + } + if head.BaseFee == nil { + return nil, nil, errors.New("chain is not EIP-1559 ready: header has no base fee") + } + feeCap = new(big.Int).Add( + tip, + new(big.Int).Mul(head.BaseFee, big.NewInt(basefeeWiggleMultiplier)), + ) + return tip, feeCap, nil +} + +// replacementPrices raises the suggested prices until the pool will accept the +// new transaction in place of the incumbent, and reports whether that stays +// within the ceiling. +// +// Recomputing from the node's suggestion is not enough on its own. The pool +// demands a bump on the tip as well as the fee cap, and SuggestGasTipCap is a +// tip oracle that does not move when the base fee does. So on a chain with +// steady tips every resubmission would be rejected as underpriced, and a +// transaction left below a risen base fee could never be replaced, wedging the +// whole queue behind it. +func replacementPrices(suggestedTip, suggestedFeeCap *big.Int, old *types.Transaction, maxFeeCapPercent uint64) (tip, feeCap *big.Int, ok bool) { + tip = maxInt(suggestedTip, outbid(old.GasTipCap())) + feeCap = maxInt(suggestedFeeCap, outbid(old.GasFeeCap())) + + ceiling := percentOf(suggestedFeeCap, maxFeeCapPercent) + if feeCap.Cmp(ceiling) > 0 { + return nil, nil, false + } + return tip, feeCap, true +} + +// outbid is the least a field may be to displace an incumbent holding old: the +// percentage threshold, and at least one wei more, since the threshold rounds +// down and the pool also requires a strict increase. +func outbid(old *big.Int) *big.Int { + return maxInt(percentOf(old, priceBumpPercent), new(big.Int).Add(old, big.NewInt(1))) +} + +func percentOf(value *big.Int, percent uint64) *big.Int { + scaled := new(big.Int).Mul(value, new(big.Int).SetUint64(percent)) + return scaled.Div(scaled, big.NewInt(100)) +} + +func maxInt(a, b *big.Int) *big.Int { + if a.Cmp(b) >= 0 { + return new(big.Int).Set(a) + } + return new(big.Int).Set(b) +} + +// transactOpts constructs opts with ctx, from address, signer, nonce, tip, +// and fee cap set. Gas limit is not set as the manager cannot compute it. +func (m *Manager) transactOpts(ctx context.Context, nonce uint64, tip, feeCap *big.Int) *bind.TransactOpts { + return &bind.TransactOpts{ + From: m.from, + Signer: m.signer, + Nonce: new(big.Int).SetUint64(nonce), + GasTipCap: tip, + GasFeeCap: feeCap, + Context: ctx, + } +} + +// poll advances every pending transaction once: those the chain has decided are +// resolved, and the first in the queue is sent again if stale. +func (m *Manager) poll(ctx context.Context) { + m.checkReceipts(ctx) + m.resubmitFirstPending(ctx) +} + +func (m *Manager) firstPendingTransaction() *pendingTransaction { + return m.pending[0] +} + +func (m *Manager) lastPendingTransaction() *pendingTransaction { + return m.pending[len(m.pending)-1] +} + +// remove drops a transaction from the queue, keeping the rest in nonce order. +func (m *Manager) remove(p *pendingTransaction) { + for i, candidate := range m.pending { + if candidate == p { + m.pending = append(m.pending[:i], m.pending[i+1:]...) + return + } + } +} + +// checkReceipts resolves every pending transaction the chain has decided on. +func (m *Manager) checkReceipts(ctx context.Context) { + for _, p := range slices.Clone(m.pending) { + for _, tx := range p.txs { + receipt, err := m.client.TransactionReceipt(ctx, tx.Hash()) + if err == nil { + m.resolve(p, Event{Receipt: receipt}) + break + } + + // A missing receipt is the normal case for a transaction that has + // not been mined yet. Any other error is a problem with the node, + // worth reporting but not evidence about the transaction. + if !errors.Is(err, ethereum.NotFound) { + metrics.FailedRPCCalls.Inc() + log.Err(err).Str("tx_hash", tx.Hash().Hex()). + Msg("failed to query transaction receipt") + } + } + } +} + +// resubmitHead sends the lowest-nonce pending transaction again, if it has gone +// RebroadcastAfter without a submission attempt. +// +// Only that one, and only one per poll. Transactions are mined in nonce order, +// so the lowest is the only one the chain can accept next, and repricing the +// others cannot make them move until it does. Once it is mined or carried over +// the next takes its place and gets its turn, so nothing is starved. Prices are +// therefore always read immediately before the submission that uses them. +func (m *Manager) resubmitFirstPending(ctx context.Context) { + if len(m.pending) == 0 { + return + } + first := m.firstPendingTransaction() + if time.Since(first.lastAttempt) < m.cfg.RebroadcastAfter { + return + } + + nonce := first.nonce() + if accountNonce, err := m.client.NonceAt(ctx, m.from, nil); err != nil { + // Without the mined nonce there is no telling whether this transaction + // still holds a live one, so resubmit it as it is and look again next poll. + metrics.FailedRPCCalls.Inc() + log.Err(err).Msg("failed to read the account nonce") + } else if nonce < accountNonce { + // Something mined this nonce, and a whole RebroadcastAfter of receipt + // checks has not turned up a receipt of ours, so it was not us. Carry + // the request over to a free nonce. + // + // The transaction abandoned here is all but certainly dead, having lost + // its nonce to a mined one, but that cannot be proven: if it is somehow + // still included the request lands twice. On the time registry the + // second one reverts with AlreadyRegistered; on the event registry both + // succeed and the identity is registered twice. + nonce = m.nextFreeNonce(accountNonce) + } + + suggestedTip, suggestedFeeCap, err := m.gasPrices(ctx) + if err != nil { + log.Err(err).Msg("failed to calculate gas prices, skipping resubmission") + return + } + + // Only a resubmission at the same nonce has an incumbent to outbid. A + // carried-over one takes a free nonce, so the suggestion stands. + tip, feeCap := suggestedTip, suggestedFeeCap + if nonce == first.nonce() { + var withinCeiling bool + tip, feeCap, withinCeiling = replacementPrices( + suggestedTip, suggestedFeeCap, first.tx(), m.cfg.MaxFeeCapPercent) + if !withinCeiling { + // Outbidding would cost more than the ceiling allows, so wait for + // fees to fall. lastAttempt still moves, otherwise this would be + // recomputed on every poll rather than once per interval. + first.lastAttempt = time.Now() + log.Warn().Str("tx_hash", first.tx().Hash().Hex()).Uint64("nonce", nonce). + Uint64("max_fee_cap_percent", m.cfg.MaxFeeCapPercent). + Msg("cannot outbid own pending transaction within the fee ceiling") + return + } + } + + first.lastAttempt = time.Now() + tx, err := first.submit(m.transactOpts(ctx, nonce, tip, feeCap)) + switch { + case err == nil: + m.replaceTx(first, tx) + case isBenignResubmitError(err): + // Nothing to do, and nothing to record: the next poll reads the account + // nonce and the receipt again, which is all these outcomes would have + // told us. + log.Debug().Err(err).Str("tx_hash", first.tx().Hash().Hex()).Uint64("nonce", nonce). + Msg("resubmission was a no-op") + default: + metrics.FailedRPCCalls.Inc() + log.Err(err).Str("tx_hash", first.tx().Hash().Hex()).Uint64("nonce", nonce). + Msg("failed to resubmit transaction") + } +} + +// nextFreeNonce returns a nonce that no pending transaction occupies and that the +// chain has not already mined, so a carried-over request lands above the queue +// rather than colliding with a transaction of ours that is still live. +func (m *Manager) nextFreeNonce(accountNonce uint64) uint64 { + if afterTail := m.lastPendingTransaction().nonce() + 1; afterTail > accountNonce { + return afterTail + } + return accountNonce +} + +// replaceTx records the transaction a successful resubmission produced, rekeying +// it if the request was carried over to a new nonce. +func (m *Manager) replaceTx(p *pendingTransaction, tx *types.Transaction) { + old := p.tx() + if tx.Hash() == old.Hash() { + return + } + p.txs = append(p.txs, tx) + p.publish(Event{Tx: tx}) + + event := log.Info().Str("old_tx_hash", old.Hash().Hex()).Str("tx_hash", tx.Hash().Hex()) + if tx.Nonce() == old.Nonce() { + event.Uint64("nonce", tx.Nonce()).Msg("resubmission replaced transaction at a higher price") + return + } + + // The new nonce is above every other in the queue, so the request belongs at + // the end. Moving it keeps the queue ordered without sorting, and means a + // carried-over request can never displace one that is still live. + m.remove(p) + m.pending = append(m.pending, p) + + event.Uint64("old_nonce", old.Nonce()).Uint64("nonce", tx.Nonce()). + Msg("carried request over to a new nonce after losing the old one") +} + +// resolve reports the chain's verdict and stops watching. +func (m *Manager) resolve(p *pendingTransaction, ev Event) { + m.remove(p) + + label := statusLabel(ev) + metrics.PendingTransactions.Set(float64(len(m.pending))) + metrics.TransactionsResolved.WithLabelValues(label).Inc() + + // The mined version need not be the last one announced, so the receipt names + // the transaction this is about whenever there is one. + hash := p.tx().Hash() + if ev.Receipt != nil { + hash = ev.Receipt.TxHash + } + + event := log.Info() + if label != metrics.TxStatusConfirmed { + event = log.Error() + } + event.Str("tx_hash", hash.Hex()). + Uint64("nonce", p.nonce()). + Str("status", label). + Msg("transaction resolved") + + p.finish(ev) +} + +// statusLabel is the metric label for a terminal event. It is finer grained than +// the events themselves on purpose: a revert is the contract's verdict rather +// than the Manager's, so callers read it off the receipt, but a revert rate is +// still worth alerting on. +func statusLabel(ev Event) string { + switch { + case ev.Receipt == nil: + if errors.Is(ev.Err, ErrAbandoned) { + return metrics.TxStatusAbandoned + } + return metrics.TxStatusRejected + case ev.Receipt.Status == types.ReceiptStatusFailed: + return metrics.TxStatusReverted + default: + return metrics.TxStatusConfirmed + } +} + +// reject ends a request that never made it onto the chain. It was never tracked, +// so there is nothing to stop watching. The counterpart to resolve: between them +// every request ends in exactly one place, counted and logged once. +func (m *Manager) reject(req sendRequest, err error) { + ev := Event{Err: err} + metrics.TransactionsResolved.WithLabelValues(statusLabel(ev)).Inc() + log.Err(err).Msg("request rejected before submission") + req.finish(ev) +} + +// shutdown ends everything outstanding, so the guarantee of a terminal event +// followed by a close holds even for a Manager that is going away and no caller +// waits on a channel that will never fire. +func (m *Manager) shutdown() { + for _, p := range slices.Clone(m.pending) { + m.resolve(p, Event{Err: ErrAbandoned}) + } + for { + select { + case req := <-m.requests: + m.reject(req, ErrShutdown) + default: + return + } + } +} + +// isBenignResubmitError reports whether a failed resubmission needs no action. +// +// "already known" is routine: the pool has the transaction and the resubmission +// changed nothing about it. "replacement transaction underpriced" means the +// outbid was not enough, which should be rare now that resubmissions price +// against the incumbent, but can still happen if the node saw a higher version +// than the one we hold. +// +// "nonce too low" is different: the nonce is read immediately before +// submitting, so it only appears when the nonce is mined in the moment between +// the two calls. It is ignored rather than expected. Reporting it as a failure +// would be wrong, since nothing failed on our side and the next poll reads the +// nonce again and carries the request over, and the only cost of the race is +// that the carry-over waits one more RebroadcastAfter. +// +// These are geth's error strings, from core/txpool.ErrAlreadyKnown, +// core/txpool.ErrReplaceUnderpriced and core.ErrNonceTooLow. They arrive over +// RPC as plain text rather than typed errors, so they cannot be compared with +// errors.Is, and a non-geth node may word them differently. +func isBenignResubmitError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + for _, expected := range []string{ + "already known", + "replacement transaction underpriced", + "nonce too low", + } { + if strings.Contains(msg, expected) { + return true + } + } + return false +} diff --git a/internal/txmgr/manager_test.go b/internal/txmgr/manager_test.go new file mode 100644 index 0000000..33e237b --- /dev/null +++ b/internal/txmgr/manager_test.go @@ -0,0 +1,980 @@ +package txmgr + +import ( + "context" + "errors" + "math/big" + "os" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi/bind" + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/rs/zerolog" + "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service" + "github.com/shutter-network/shutter-api/metrics" + "github.com/stretchr/testify/require" +) + +// TestMain silences the manager's logging, which is per transaction per poll and +// would otherwise bury the test output. +func TestMain(m *testing.M) { + zerolog.SetGlobalLevel(zerolog.Disabled) + os.Exit(m.Run()) +} + +// Fixed so failures reproduce. Any valid secp256k1 scalar will do. +const testSigningKey = "b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291" + +var ( + testChainID = big.NewInt(100) // Gnosis + testTo = ecommon.HexToAddress("0x1111111111111111111111111111111111111111") + testTip = big.NewInt(1_000_000_000) + testBaseFee = big.NewInt(7_000_000_000) +) + +// fakeClient stands in for ethclient.Client. A fake is used rather than a +// simulated chain because the interesting cases here are node behaviours we +// cannot easily provoke for real: a nonce stolen by someone else, a receipt that +// has not propagated yet, a pool that forgot a transaction. +type fakeClient struct { + mu sync.Mutex + + accountNonce uint64 + nonceAtCalls int + nonceAtErr error + + receipts map[ecommon.Hash]*types.Receipt + receiptErr error + + tip *big.Int + baseFee *big.Int +} + +func newFakeClient() *fakeClient { + return &fakeClient{ + receipts: make(map[ecommon.Hash]*types.Receipt), + tip: new(big.Int).Set(testTip), + baseFee: new(big.Int).Set(testBaseFee), + } +} + +func (c *fakeClient) NonceAt(_ context.Context, _ ecommon.Address, _ *big.Int) (uint64, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.nonceAtCalls++ + if c.nonceAtErr != nil { + return 0, c.nonceAtErr + } + return c.accountNonce, nil +} + +func (c *fakeClient) TransactionReceipt(_ context.Context, hash ecommon.Hash) (*types.Receipt, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.receiptErr != nil { + return nil, c.receiptErr + } + if receipt, ok := c.receipts[hash]; ok { + return receipt, nil + } + return nil, ethereum.NotFound +} + +func (c *fakeClient) SuggestGasTipCap(_ context.Context) (*big.Int, error) { + c.mu.Lock() + defer c.mu.Unlock() + return new(big.Int).Set(c.tip), nil +} + +func (c *fakeClient) HeaderByNumber(_ context.Context, _ *big.Int) (*types.Header, error) { + c.mu.Lock() + defer c.mu.Unlock() + return &types.Header{BaseFee: new(big.Int).Set(c.baseFee)}, nil +} + +func (c *fakeClient) setReceipt(hash ecommon.Hash, status uint64) { + c.mu.Lock() + defer c.mu.Unlock() + c.receipts[hash] = &types.Receipt{Status: status, BlockNumber: big.NewInt(42), TxHash: hash} +} + +func (c *fakeClient) setBaseFee(wei int64) { + c.mu.Lock() + defer c.mu.Unlock() + c.baseFee = big.NewInt(wei) +} + +func (c *fakeClient) setReceiptErr(err error) { + c.mu.Lock() + defer c.mu.Unlock() + c.receiptErr = err +} + +func (c *fakeClient) setAccountNonce(nonce uint64) { + c.mu.Lock() + defer c.mu.Unlock() + c.accountNonce = nonce +} + +func (c *fakeClient) chainReads() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.nonceAtCalls +} + +// submitter is a SubmitFunc that records the opts it was handed and can be told +// to fail. It stands in for a generated contract binding. +type submitter struct { + mu sync.Mutex + seen []*bind.TransactOpts + built []*types.Transaction + err error +} + +func (s *submitter) submit(opts *bind.TransactOpts) (*types.Transaction, error) { + s.mu.Lock() + defer s.mu.Unlock() + + s.seen = append(s.seen, opts) + if s.err != nil { + return nil, s.err + } + + tx := types.NewTx(&types.DynamicFeeTx{ + ChainID: testChainID, + Nonce: opts.Nonce.Uint64(), + GasTipCap: opts.GasTipCap, + GasFeeCap: opts.GasFeeCap, + Gas: 21000, + To: &testTo, + }) + s.built = append(s.built, tx) + return tx, nil +} + +func (s *submitter) fail(err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.err = err +} + +// nonces is every nonce submission was attempted with, in order, so a test can +// tell which transactions a resubmission round touched. +func (s *submitter) nonces() []uint64 { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]uint64, 0, len(s.seen)) + for _, opts := range s.seen { + out = append(out, opts.Nonce.Uint64()) + } + return out +} + +func (s *submitter) attempts() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.seen) +} + +// lastTx is how a test running against Start learns a hash, since the pending +// queue belongs to the worker goroutine and must not be read from outside it. +func (s *submitter) lastTx() *types.Transaction { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.built) == 0 { + return nil + } + return s.built[len(s.built)-1] +} + +func newTestManager(t *testing.T, client Client) *Manager { + t.Helper() + key, err := crypto.HexToECDSA(testSigningKey) + require.NoError(t, err) + m, err := NewManager(client, key, testChainID, DefaultConfig()) + require.NoError(t, err) + return m +} + +// drain submits every queued request. Production code does this in the Start +// loop; tests call it directly so they can drive the Manager one step at a time, +// on the same goroutine that owns its state. +func drain(m *Manager) { + for { + select { + case req := <-m.requests: + m.submit(context.Background(), req) + default: + return + } + } +} + +// send queues a call and submits it, which is what Start would have done. +func send(m *Manager, submit SubmitFunc) <-chan Event { + events := m.Send(submit) + drain(m) + return events +} + +// lastEvent reads a request's events to the end and returns the last one, also +// checking the contract every request is meant to honour: the last event is +// terminal, and the channel is closed after it. +func lastEvent(t *testing.T, events <-chan Event) Event { + t.Helper() + var last Event + for { + select { + case ev, open := <-events: + if !open { + require.True(t, last.Receipt != nil || last.Err != nil, + "the last event before the channel closed must be a receipt or an error") + return last + } + last = ev + case <-time.After(2 * time.Second): + t.Fatal("the request never ended") + } + } +} + +// requireWatched requires that a request has not ended, consuming the +// transactions announced so far. Every event before the last is a transaction, so +// anything else here means the request resolved when it should not have. +func requireWatched(t *testing.T, events <-chan Event) { + t.Helper() + for { + select { + case ev, open := <-events: + require.True(t, open, "the request ended when it should still be watched") + require.Nil(t, ev.Receipt, "the request was resolved when it should still be watched") + require.NoError(t, ev.Err, "the request was failed when it should still be watched") + require.NotNil(t, ev.Tx, "an event set no field") + default: + return + } + } +} + +// announcedTxs is every transaction a request has announced so far. +func announcedTxs(t *testing.T, events <-chan Event) []*types.Transaction { + t.Helper() + var out []*types.Transaction + for { + select { + case ev, open := <-events: + require.True(t, open, "the request ended") + require.NotNil(t, ev.Tx, "expected a transaction event") + out = append(out, ev.Tx) + default: + return out + } + } +} + +// makeStale backdates a pending transaction so the next poll resubmits it. +func makeStale(t *testing.T, m *Manager, nonce uint64) { + t.Helper() + for _, p := range m.pending { + if p.nonce() == nonce { + p.lastAttempt = time.Now().Add(-2 * m.cfg.RebroadcastAfter) + return + } + } + t.Fatalf("no pending transaction with nonce %d", nonce) +} + +// pendingWithNonce returns the queued transaction holding a given nonce. +func pendingWithNonce(t *testing.T, m *Manager, nonce uint64) *pendingTransaction { + t.Helper() + for _, p := range m.pending { + if p.nonce() == nonce { + return p + } + } + t.Fatalf("no pending transaction with nonce %d", nonce) + return nil +} + +// pendingNonces returns the nonces currently in flight, in order. +func pendingNonces(m *Manager) []uint64 { + out := make([]uint64, 0, len(m.pending)) + for _, p := range m.pending { + out = append(out, p.nonce()) + } + return out +} + +// onlyPending returns the single pending transaction. +func onlyPending(t *testing.T, m *Manager) *pendingTransaction { + t.Helper() + require.Len(t, m.pending, 1) + return m.pending[0] +} + +// TestConcurrentSendsGetDistinctSequentialNonces is the regression test for the +// bug this package exists to fix. Against the old code, which left +// TransactOpts.Nonce nil and let every call resolve its own, this fails with the +// same nonce issued repeatedly. +func TestConcurrentSendsGetDistinctSequentialNonces(t *testing.T) { + const senders = 50 + + client := newFakeClient() + client.accountNonce = 7 + m := newTestManager(t, client) + s := &submitter{} + + var wg sync.WaitGroup + for i := 0; i < senders; i++ { + wg.Add(1) + go func() { + defer wg.Done() + m.Send(s.submit) + }() + } + wg.Wait() + drain(m) + + seen := make(map[uint64]bool) + for _, nonce := range s.nonces() { + require.False(t, seen[nonce], "nonce %d issued twice", nonce) + seen[nonce] = true + } + require.Len(t, seen, senders) + for nonce := uint64(7); nonce < 7+senders; nonce++ { + require.True(t, seen[nonce], "nonce %d missing from the sequence", nonce) + } + + // Only the first submission finds nothing in flight, so the chain is read + // once no matter how many registrations arrive together. + require.Equal(t, 1, client.chainReads()) +} + +func TestSendDoesNotTouchTheNetwork(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := m.Send(s.submit) + + require.Zero(t, client.chainReads(), "Send must not make RPC calls") + require.Zero(t, s.attempts(), "Send must not build a transaction") + require.Empty(t, events, "Send must not report anything before the worker runs") + + drain(m) + require.Equal(t, 1, s.attempts()) + require.Equal(t, s.lastTx().Hash(), (<-events).Tx.Hash(), + "submitting must announce the transaction") +} + +func TestChainIsReadOnlyWhenNothingIsInFlight(t *testing.T) { + client := newFakeClient() + client.accountNonce = 3 + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + first := onlyPending(t, m) + require.Equal(t, uint64(3), first.nonce()) + require.Equal(t, 1, client.chainReads()) + + send(m, s.submit) + require.Equal(t, 1, client.chainReads(), "chain read again while work was in flight") + + // Empty the set, and the next submission goes back to the chain. + client.setReceipt(first.tx().Hash(), types.ReceiptStatusSuccessful) + m.poll(context.Background()) + require.NotNil(t, lastEvent(t, events).Receipt) + + m.shutdown() + send(m, s.submit) + require.Equal(t, 2, client.chainReads()) +} + +func TestFailedSubmissionReusesTheNonce(t *testing.T) { + client := newFakeClient() + client.accountNonce = 11 + m := newTestManager(t, client) + s := &submitter{} + + submitErr := errors.New("connection refused") + s.fail(submitErr) + events := send(m, s.submit) + require.ErrorIs(t, lastEvent(t, events).Err, submitErr, + "the caller should be told why submission failed") + require.Empty(t, m.pending, "a failed submission must not be watched") + + s.fail(nil) + send(m, s.submit) + require.Equal(t, uint64(11), onlyPending(t, m).nonce(), + "a failed submission must not consume its nonce") +} + +func TestRequestIsRejectedWhenChainNonceUnavailable(t *testing.T) { + client := newFakeClient() + client.nonceAtErr = errors.New("rpc down") + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + + require.ErrorIs(t, lastEvent(t, events).Err, client.nonceAtErr) + require.Zero(t, s.attempts(), "no transaction should be built without a nonce") +} + +func TestFullQueueRejectsImmediately(t *testing.T) { + client := newFakeClient() + key, err := crypto.HexToECDSA(testSigningKey) + require.NoError(t, err) + m, err := NewManager(client, key, testChainID, Config{QueueSize: 1}) + require.NoError(t, err) + s := &submitter{} + + require.Empty(t, m.Send(s.submit), "the first request fits in the queue") + + // Nothing has drained the queue, so this one has nowhere to go. + events := m.Send(s.submit) + ev := lastEvent(t, events) + require.ErrorIs(t, ev.Err, ErrQueueFull) + require.Nil(t, ev.Tx, "a request that was never sent has no transaction") +} + +func TestGasPricesFollowTheBindingFormula(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + + s.mu.Lock() + opts := s.seen[0] + s.mu.Unlock() + + require.Equal(t, testTip, opts.GasTipCap) + // tip + 2*basefee, matching basefeeWiggleMultiplier in the bindings. + want := new(big.Int).Add(testTip, new(big.Int).Mul(testBaseFee, big.NewInt(2))) + require.Equal(t, want, opts.GasFeeCap) + require.Zero(t, opts.GasLimit, "the gas limit is the bindings' business to estimate, not ours") + require.Equal(t, m.From(), opts.From) + require.NotNil(t, opts.Signer) +} + +// A revert is the contract's verdict, not the Manager's, so both outcomes +// resolve as confirmed and the caller reads Receipt.Status to tell them apart. +func TestPollResolvesAnyMinedTransactionAsConfirmed(t *testing.T) { + for _, tc := range []struct { + name string + status uint64 + }{ + {"mined successfully", types.ReceiptStatusSuccessful}, + {"mined but reverted", types.ReceiptStatusFailed}, + } { + t.Run(tc.name, func(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + p := onlyPending(t, m) + + m.poll(context.Background()) + requireWatched(t, events) + + client.setReceipt(p.tx().Hash(), tc.status) + m.poll(context.Background()) + + ev := lastEvent(t, events) + require.NotNil(t, ev.Receipt) + require.Equal(t, tc.status, ev.Receipt.Status) + require.NoError(t, ev.Err, "a mined transaction is not an error, whatever it did") + }) + } +} + +// Something else mining our nonce must retry the request at a free one rather +// than fail it. +func TestStaleTransactionWhoseNonceWasMinedIsCarriedOver(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + require.Equal(t, uint64(0), onlyPending(t, m).nonce()) + + // Something else mined nonce 0, so ours never can be. + client.setAccountNonce(1) + makeStale(t, m, 0) + m.poll(context.Background()) + + requireWatched(t, events) // the request must be retried, not failed + require.Equal(t, []uint64{0, 1}, s.nonces(), "the request should move to the free nonce") + require.Equal(t, []uint64{1}, pendingNonces(m)) + + // It is the same request, so the original caller still gets the answer. + client.setReceipt(onlyPending(t, m).tx().Hash(), types.ReceiptStatusSuccessful) + m.poll(context.Background()) + require.NotNil(t, lastEvent(t, events).Receipt) +} + +// Staleness is the grace period: a transaction is only carried over once a whole +// RebroadcastAfter of receipt checks has come up empty. A fresh transaction whose +// nonce merely looks mined must be left alone, since the most likely explanation +// is that it was mined by us and the receipt has not propagated. +func TestFreshTransactionWhoseNonceWasMinedIsLeftAlone(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + client.setAccountNonce(1) + + m.poll(context.Background()) + + require.Equal(t, []uint64{0}, s.nonces(), "a fresh transaction must not be carried over") + require.Equal(t, []uint64{0}, pendingNonces(m)) +} + +// A rejected resubmission still counts as an attempt. Otherwise the transaction +// stays stale and is retried on every poll, turning a one minute interval into a +// five second one. +func TestRejectedResubmissionStillResetsTheClock(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + s.fail(errors.New("already known")) + makeStale(t, m, onlyPending(t, m).nonce()) + + m.poll(context.Background()) + require.Equal(t, 2, s.attempts(), "the stale transaction should be resubmitted once") + + m.poll(context.Background()) + require.Equal(t, 2, s.attempts(), "a failed attempt must still count as an attempt") +} + +// A replacement is a new transaction for the same request, so it must not become +// a second queue entry or a second caller. +func TestAReplacementStaysTheSameRequest(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + p := onlyPending(t, m) + original := p.tx() + + makeStale(t, m, p.nonce()) + m.poll(context.Background()) + + require.Len(t, m.pending, 1, "the replacement must not become a second entry") + require.Equal(t, original.Nonce(), p.nonce(), "a replacement keeps the nonce") + require.NotEqual(t, original.Hash(), p.tx().Hash(), "the tracked transaction should have changed") + + // The caller's channel survives the swap and reports the winning version. + client.setReceipt(p.tx().Hash(), types.ReceiptStatusSuccessful) + m.poll(context.Background()) + require.Equal(t, p.tx().Hash(), lastEvent(t, events).Receipt.TxHash) +} + +// A caller that wants to know what to watch has to be told every time that +// changes, otherwise it is left polling a hash the Manager has already given up +// on. +func TestEveryVersionSentIsAnnounced(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + for round := 0; round < 2; round++ { + makeStale(t, m, 0) + m.poll(context.Background()) + } + require.Equal(t, 3, s.attempts(), "expected an original and two replacements") + + announced := announcedTxs(t, events) + require.Len(t, announced, 3, "every version sent must be announced") + + s.mu.Lock() + built := s.built + s.mu.Unlock() + for i, tx := range announced { + require.Equal(t, built[i].Hash(), tx.Hash(), "version %d announced out of order", i) + } + + // The last event is still the outcome, and it names the version that won. + client.setReceipt(announced[2].Hash(), types.ReceiptStatusSuccessful) + m.poll(context.Background()) + require.Equal(t, announced[2].Hash(), lastEvent(t, events).Receipt.TxHash) +} + +// The buffer is finite and the worker must never wait on a caller, so an +// abandoned channel loses intermediate events. The outcome is not intermediate, +// and losing it would leave a caller that comes back later unable to tell what +// happened, so a slot is reserved for it. +func TestAFullChannelStillDeliversTheOutcome(t *testing.T) { + req := sendRequest{events: make(chan Event, eventBufferSize)} + + for i := 0; i < eventBufferSize*2; i++ { + req.publish(Event{Tx: types.NewTx(&types.DynamicFeeTx{Nonce: uint64(i)})}) + } + req.finish(Event{Err: ErrAbandoned}) + + require.Len(t, req.events, eventBufferSize, "the reserved slot should have been used by finish") + + var last Event + for ev := range req.events { + last = ev + } + require.ErrorIs(t, last.Err, ErrAbandoned, "the outcome must survive a full buffer") +} + +// Transactions are mined in nonce order, so the lowest is the only one the chain +// can accept next. Repricing the ones behind it cannot make them move and would +// pay more gas for nothing. +func TestOnlyTheLowestNoncePendingTransactionIsResubmitted(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + send(m, s.submit) + send(m, s.submit) + for nonce := uint64(0); nonce < 3; nonce++ { + makeStale(t, m, nonce) + } + + m.poll(context.Background()) + require.Equal(t, []uint64{0, 1, 2, 0}, s.nonces()) + + // Its turn is over until it goes stale again, and the ones behind it still + // wait on it, so a second poll changes nothing. + m.poll(context.Background()) + require.Equal(t, []uint64{0, 1, 2, 0}, s.nonces()) +} + +// Once the head is mined the next takes its place, so nothing behind it is +// starved by only ever resubmitting the head. +func TestTheNextTransactionTakesItsTurnOnceTheHeadIsMined(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + send(m, s.submit) + makeStale(t, m, 0) + makeStale(t, m, 1) + + m.poll(context.Background()) + require.Equal(t, []uint64{0, 1, 0}, s.nonces(), "only the head is resubmitted") + + client.setReceipt(pendingWithNonce(t, m, 0).tx().Hash(), types.ReceiptStatusSuccessful) + client.setAccountNonce(1) + m.poll(context.Background()) + + require.Equal(t, []uint64{0, 1, 0, 1}, s.nonces(), "nonce 1 is the head now and was stale") + require.Equal(t, []uint64{1}, pendingNonces(m)) +} + +func TestAFreshHeadIsNotResubmitted(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + send(m, s.submit) + // Only the one behind the head is stale, and it cannot move until the head + // does, so nothing should be sent. + makeStale(t, m, 1) + + m.poll(context.Background()) + require.Equal(t, []uint64{0, 1}, s.nonces()) +} + +// The pool rejects a resubmission on every calm-chain poll, so those rejections +// must not read as RPC failures. Otherwise the failure metric climbs steadily +// while nothing is wrong, and a real problem is invisible in the noise. A nonce +// mined between reading it and submitting is rarer, but equally not our failure. +func TestBenignResubmitOutcomesAreNotCountedAsFailures(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + + for _, expected := range []string{ + "already known", + "replacement transaction underpriced", + "nonce too low", + } { + s.fail(errors.New(expected)) + makeStale(t, m, 0) + + before := testutil.ToFloat64(metrics.FailedRPCCalls) + m.poll(context.Background()) + + require.Equal(t, before, testutil.ToFloat64(metrics.FailedRPCCalls), + "%q must not count as an RPC failure", expected) + requireWatched(t, events) + require.Len(t, m.pending, 1, "%q must leave the request being watched", expected) + } + + // Anything else does. + s.fail(errors.New("intrinsic gas too low")) + makeStale(t, m, 0) + + before := testutil.ToFloat64(metrics.FailedRPCCalls) + m.poll(context.Background()) + require.Equal(t, before+1, testutil.ToFloat64(metrics.FailedRPCCalls)) +} + +// geth rejects a replacement unless BOTH the tip and the fee cap are strictly +// greater than the incumbent's and clear 110% (list.Add). SuggestGasTipCap is a +// tip oracle that does not move with the base fee, so pricing a resubmission +// from the suggestion alone would leave the tip flat and every replacement would +// be rejected, stranding a transaction under a risen base fee forever. +func TestResubmissionOutbidsTheIncumbentEvenWhenSuggestionsAreFlat(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + original := onlyPending(t, m).tx() + + // Suggestions unchanged, exactly as a calm-tip chain behaves. + makeStale(t, m, 0) + m.poll(context.Background()) + + bumped := onlyPending(t, m).tx() + require.Positive(t, bumped.GasTipCap().Cmp(original.GasTipCap()), "tip must rise") + require.Positive(t, bumped.GasFeeCap().Cmp(original.GasFeeCap()), "fee cap must rise") + requireClearsGethThreshold(t, original, bumped) +} + +// requireClearsGethThreshold applies geth's own replacement rule, so the test +// fails if a bump would be rejected by the pool. +func requireClearsGethThreshold(t *testing.T, old, replacement *types.Transaction) { + t.Helper() + + require.Positive(t, replacement.GasFeeCap().Cmp(old.GasFeeCap()), "fee cap must be strictly greater") + require.Positive(t, replacement.GasTipCap().Cmp(old.GasTipCap()), "tip must be strictly greater") + + hundred := big.NewInt(100) + bump := big.NewInt(priceBumpPercent) + feeCapThreshold := new(big.Int).Div(new(big.Int).Mul(bump, old.GasFeeCap()), hundred) + tipThreshold := new(big.Int).Div(new(big.Int).Mul(bump, old.GasTipCap()), hundred) + + require.GreaterOrEqual(t, replacement.GasFeeCap().Cmp(feeCapThreshold), 0, "fee cap below the 110% threshold") + require.GreaterOrEqual(t, replacement.GasTipCap().Cmp(tipThreshold), 0, "tip below the 110% threshold") +} + +// Escalation compounds, so without a ceiling a transaction that never mines +// would bid the account dry. +func TestResubmissionStopsAtTheFeeCeiling(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + send(m, s.submit) + + // Offer it the chance to escalate far more often than the ceiling allows at + // 110% a round. It has to stop bidding of its own accord. + stopped := false + for round := 0; round < 50 && !stopped; round++ { + before := s.attempts() + makeStale(t, m, 0) + m.poll(context.Background()) + stopped = s.attempts() == before + } + require.True(t, stopped, "escalation never stopped, the ceiling is not enforced") + + suggestedFeeCap := new(big.Int).Add(testTip, new(big.Int).Mul(testBaseFee, big.NewInt(2))) + ceiling := new(big.Int).Div( + new(big.Int).Mul(suggestedFeeCap, new(big.Int).SetUint64(m.cfg.MaxFeeCapPercent)), + big.NewInt(100), + ) + paid := onlyPending(t, m).tx().GasFeeCap() + require.LessOrEqual(t, paid.Cmp(ceiling), 0, "paid %s, ceiling %s", paid, ceiling) + + // Still watched, so it resolves if fees fall and it is finally mined. + require.Len(t, m.pending, 1) +} + +// A replacement should evict its predecessor, but propagation is not atomic. If +// an earlier version is the one mined, that must read as our transaction +// succeeding, not as a stranger taking the nonce. +func TestAnEarlierVersionBeingMinedResolvesTheRequest(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + original := onlyPending(t, m).tx() + + makeStale(t, m, 0) + m.poll(context.Background()) + bumped := onlyPending(t, m).tx() + require.NotEqual(t, original.Hash(), bumped.Hash(), "the resubmission should have replaced it") + + // The version we stopped tracking is the one that lands. + client.setReceipt(original.Hash(), types.ReceiptStatusSuccessful) + client.setAccountNonce(1) + m.poll(context.Background()) + + // Checked before reading the channel so that failing to notice the earlier + // version reports itself here, rather than blocking on an outcome that a + // manager watching only the newest version would never deliver. + require.Empty(t, m.pending, "a request mined as an earlier version must be resolved, not still watched") + + require.Equal(t, original.Hash(), lastEvent(t, events).Receipt.TxHash, + "the mined version should be reported") +} + +// A request that loses its nonce moves above the queue rather than colliding +// with a live transaction of ours. +func TestCarriedOverRequestTakesANonceAboveTheOthers(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + first := send(m, s.submit) // nonce 0 + send(m, s.submit) // nonce 1 + send(m, s.submit) // nonce 2 + + // Something else mined nonce 0, leaving 1 and 2 healthy. + client.setAccountNonce(1) + for nonce := uint64(0); nonce < 3; nonce++ { + makeStale(t, m, nonce) + } + + m.poll(context.Background()) + + require.Equal(t, []uint64{1, 2, 3}, pendingNonces(m)) + + // Nonce 3 is the carried request, so its original caller gets the answer. + client.setReceipt(pendingWithNonce(t, m, 3).tx().Hash(), types.ReceiptStatusSuccessful) + m.poll(context.Background()) + require.NotNil(t, lastEvent(t, first).Receipt) +} + +func TestPollKeepsWatchingWhenReceiptLookupFails(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + events := send(m, s.submit) + + // A node error is not evidence about the transaction, so nothing resolves. + client.setReceiptErr(errors.New("rpc timeout")) + m.poll(context.Background()) + + requireWatched(t, events) + require.Len(t, m.pending, 1) +} + +// The two outstanding cases mean different things to a caller: a queued request +// was never signed and cannot land, while a submitted one is in the pool and +// still might. +func TestShutdownResolvesEverythingOutstanding(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + submitted := send(m, s.submit) + // Queued but never submitted, so shutdown has to resolve it too. + queued := m.Send(s.submit) + + m.shutdown() + + require.ErrorIs(t, lastEvent(t, submitted).Err, ErrAbandoned) + require.ErrorIs(t, lastEvent(t, queued).Err, ErrShutdown) + require.Empty(t, m.pending) +} + +func TestIgnoredEventChannelDoesNotBlockTheWorker(t *testing.T) { + client := newFakeClient() + m := newTestManager(t, client) + s := &submitter{} + + // Deliberately drop the channel: most callers have nothing to do on + // confirmation and the worker must not care. + _ = send(m, s.submit) + client.setReceipt(onlyPending(t, m).tx().Hash(), types.ReceiptStatusSuccessful) + + done := make(chan struct{}) + go func() { + m.poll(context.Background()) + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("poll blocked on events nobody is reading") + } + require.Empty(t, m.pending) +} + +// The tests above drive submit and poll directly. These two run the real Start +// loop, the only thing covering the select over requests, ticks and +// cancellation. +func TestStartDrivesARequestToConfirmation(t *testing.T) { + client := newFakeClient() + client.accountNonce = 5 + key, err := crypto.HexToECDSA(testSigningKey) + require.NoError(t, err) + m, err := NewManager(client, key, testChainID, Config{PollInterval: 5 * time.Millisecond}) + require.NoError(t, err) + s := &submitter{} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + _, deferFn := service.RunBackground(ctx, m) + defer deferFn() + + events := m.Send(s.submit) + + require.Eventually(t, func() bool { return s.lastTx() != nil }, + 2*time.Second, 5*time.Millisecond, "the worker never submitted the request") + tx := s.lastTx() + require.Equal(t, uint64(5), tx.Nonce()) + client.setReceipt(tx.Hash(), types.ReceiptStatusSuccessful) + + require.Equal(t, tx.Hash(), lastEvent(t, events).Receipt.TxHash) +} + +func TestStartResolvesOutstandingWorkOnCancellation(t *testing.T) { + client := newFakeClient() + key, err := crypto.HexToECDSA(testSigningKey) + require.NoError(t, err) + m, err := NewManager(client, key, testChainID, Config{PollInterval: time.Hour}) + require.NoError(t, err) + s := &submitter{} + + ctx, cancel := context.WithCancel(context.Background()) + _, deferFn := service.RunBackground(ctx, m) + defer deferFn() + + events := m.Send(s.submit) + require.Eventually(t, func() bool { return s.lastTx() != nil }, + 2*time.Second, 5*time.Millisecond) + + cancel() + + // lastEvent fails the test rather than hanging if shutdown leaves a caller + // waiting on a channel that never fires. + require.ErrorIs(t, lastEvent(t, events).Err, ErrAbandoned) +} + +func TestFromIsDerivedFromTheSigningKey(t *testing.T) { + key, err := crypto.HexToECDSA(testSigningKey) + require.NoError(t, err) + m := newTestManager(t, newFakeClient()) + require.Equal(t, crypto.PubkeyToAddress(key.PublicKey), m.From()) +} diff --git a/internal/usecase/crypto.go b/internal/usecase/crypto.go index 215697f..c258ce8 100644 --- a/internal/usecase/crypto.go +++ b/internal/usecase/crypto.go @@ -5,7 +5,6 @@ import ( "encoding/hex" "fmt" "io" - "math/big" "net/http" "net/url" "strings" @@ -14,11 +13,9 @@ import ( cryptorand "crypto/rand" "github.com/ethereum/go-ethereum/accounts/abi/bind" - ecommon "github.com/ethereum/go-ethereum/common" ethCommon "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/crypto" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/pkg/errors" @@ -57,8 +54,6 @@ type KeyBroadcastInterface interface { type EthClientInterface interface { BlockNumber(ctx context.Context) (uint64, error) - ChainID(ctx context.Context) (*big.Int, error) - TransactionReceipt(ctx context.Context, txHash ecommon.Hash) (*types.Receipt, error) } type GetDecryptionKeyResponse struct { @@ -91,6 +86,7 @@ type CryptoUsecase struct { keyperSetManagerContract KeyperSetManagerInterface keyBroadcastContract KeyBroadcastInterface ethClient EthClientInterface + txManager TxManagerInterface config *common.Config } @@ -101,6 +97,7 @@ func NewCryptoUsecase( keyperSetManagerContract KeyperSetManagerInterface, keyBroadcastContract KeyBroadcastInterface, ethClient EthClientInterface, + txManager TxManagerInterface, config *common.Config, ) *CryptoUsecase { return &CryptoUsecase{ @@ -111,38 +108,11 @@ func NewCryptoUsecase( keyperSetManagerContract: keyperSetManagerContract, keyBroadcastContract: keyBroadcastContract, ethClient: ethClient, + txManager: txManager, config: config, } } -// getSigner returns the signer for the API signer address. -func (uc *CryptoUsecase) getSigner(ctx context.Context) (*bind.TransactOpts, *httpError.Http) { - chainID, err := uc.ethClient.ChainID(ctx) - if err != nil { - log.Err(err).Msg("err encountered while querying chain id") - metrics.TotalFailedRPCCalls.Inc() - err := httpError.NewHttpError( - "error encountered while querying chain id", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - - newSigner, err := bind.NewKeyedTransactorWithChainID(uc.config.SigningKey, chainID) - if err != nil { - log.Err(err).Msg("err encountered while creating signer") - err := httpError.NewHttpError( - "error encountered while creating signer", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - - return newSigner, nil -} - func (uc *CryptoUsecase) GetDecryptionKey(ctx context.Context, identity string) (*GetDecryptionKeyResponse, *httpError.Http) { identityBytes, err := hex.DecodeString(strings.TrimPrefix(string(identity), "0x")) if err != nil { @@ -168,7 +138,7 @@ func (uc *CryptoUsecase) GetDecryptionKey(ctx context.Context, identity string) registrationData, err := uc.shutterRegistryContract.Registrations(nil, [32]byte(identityBytes)) if err != nil { log.Err(err).Msg("err encountered while querying contract") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error while querying for identity from the contract", "", @@ -301,7 +271,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin blockNumber, err := uc.ethClient.BlockNumber(ctx) if err != nil { log.Err(err).Msg("err encountered while querying for recent block") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for recent block", "", @@ -313,7 +283,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin eon, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber) if err != nil { log.Err(err).Msg("err encountered while querying keyper set index") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for keyper set index", "", @@ -325,7 +295,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin eonKeyBytes, err := uc.keyBroadcastContract.GetEonKey(nil, eon) if err != nil { log.Err(err).Msg("err encountered while querying for eon key") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for eon key", "", @@ -381,12 +351,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin // GetDataForEncryptionEvent is the event-based variant which uses the API signer address to compute the identity. func (uc *CryptoUsecase) GetDataForEncryptionEvent(ctx context.Context, identityPrefixStringified string, triggerDefinitionHex string) (*GetDataForEncryptionResponse, *httpError.Http) { - newSigner, httpErr := uc.getSigner(ctx) - if httpErr != nil { - return nil, httpErr - } - - return uc.GetDataForEncryption(ctx, newSigner.From.Hex(), identityPrefixStringified, triggerDefinitionHex) + return uc.GetDataForEncryption(ctx, uc.txManager.From().Hex(), identityPrefixStringified, triggerDefinitionHex) } func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimestamp uint64, identityPrefixStringified string) (*RegisterIdentityResponse, *httpError.Http) { @@ -442,7 +407,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta blockNumber, err := uc.ethClient.BlockNumber(ctx) if err != nil { log.Err(err).Msg("err encountered while querying for recent block") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for recent block", "", @@ -454,7 +419,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta eon, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber) if err != nil { log.Err(err).Msg("err encountered while querying keyper set index") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for keyper set index", "", @@ -466,7 +431,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta eonKeyBytes, err := uc.keyBroadcastContract.GetEonKey(nil, eon) if err != nil { log.Err(err).Msg("err encountered while querying for eon key") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for eon key", "", @@ -486,17 +451,12 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta return nil, &err } - newSigner, httpErr := uc.getSigner(ctx) - if httpErr != nil { - return nil, httpErr - } - - identity := common.ComputeIdentity(identityPrefix[:], newSigner.From) + identity := common.ComputeIdentity(identityPrefix[:], uc.txManager.From()) registrationData, err := uc.shutterRegistryContract.Registrations(nil, [32]byte(identity)) if err != nil { log.Err(err).Msg("err encountered while querying contract") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error while querying for registrations from the contract", "", @@ -515,35 +475,27 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta return nil, &err } - publicAddress := crypto.PubkeyToAddress(*uc.config.PublicKey) + events := uc.txManager.Send(func(opts *bind.TransactOpts) (*types.Transaction, error) { + return uc.shutterRegistryContract.Register(opts, eon, identityPrefix, decryptionTimestamp) + }) - opts := bind.TransactOpts{ - From: publicAddress, - Signer: newSigner.Signer, + txHash, httpErr := awaitSubmission(ctx, events) + if httpErr != nil { + return nil, httpErr } - tx, err := uc.shutterRegistryContract.Register(&opts, eon, identityPrefix, decryptionTimestamp) - if err != nil { - log.Err(err).Msg("failed to send transaction") - metrics.TotalFailedRPCCalls.Inc() - err := httpError.NewHttpError( - "failed to register identity", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - // not launching a routine to monitor the transaction - // we return the transaction hash in response to allow - // users the ability to monitor it themselves + // The rest of the events go unread on purpose. The response is committed to a + // hash, so a later version of the transaction cannot change it, and the + // transaction manager logs how the registration ends either way. Clients are + // expected to follow the hash themselves. - metrics.TotalSuccessfulIdentityRegistration.Inc() + metrics.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ Eon: eon, Identity: common.PrefixWith0x(hex.EncodeToString(identity)), IdentityPrefix: common.PrefixWith0x(hex.EncodeToString(identityPrefix[:])), EonKey: common.PrefixWith0x(hex.EncodeToString(eonKeyBytes)), - TxHash: tx.Hash().Hex(), + TxHash: txHash.Hex(), }, nil } diff --git a/internal/usecase/eventtrigger.go b/internal/usecase/eventtrigger.go index 4d29e7f..572d621 100644 --- a/internal/usecase/eventtrigger.go +++ b/internal/usecase/eventtrigger.go @@ -9,12 +9,12 @@ import ( "net/http" "slices" "strings" - "time" "github.com/defiweb/go-sigparser" "github.com/ethereum/go-ethereum/accounts/abi/bind" ecommon "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/jackc/pgx/v5" "github.com/rs/zerolog/log" @@ -23,6 +23,7 @@ import ( "github.com/shutter-network/shutter-api/internal/data" httpError "github.com/shutter-network/shutter-api/internal/error" sherror "github.com/shutter-network/shutter-api/internal/error" + "github.com/shutter-network/shutter-api/internal/txmgr" "github.com/shutter-network/shutter-api/metrics" "github.com/shutter-network/shutter/shlib/shcrypto" ) @@ -288,7 +289,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger blockNumber, err := uc.ethClient.BlockNumber(ctx) if err != nil { log.Err(err).Msg("err encountered while querying for recent block") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for recent block", "", @@ -300,7 +301,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger eon, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber) if err != nil { log.Err(err).Msg("err encountered while querying keyper set index") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for keyper set index", "", @@ -312,7 +313,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger eonKeyBytes, err := uc.keyBroadcastContract.GetEonKey(nil, eon) if err != nil { log.Err(err).Msg("err encountered while querying for eon key") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for eon key", "", @@ -332,18 +333,6 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger return nil, &err } - chainId, err := uc.ethClient.ChainID(ctx) - if err != nil { - log.Err(err).Msg("err encountered while quering chain id") - metrics.TotalFailedRPCCalls.Inc() - err := httpError.NewHttpError( - "error encountered while querying chain id", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - eventTriggerDefinition, err := hexutil.Decode(eventTriggerDefinitionHex) if err != nil { err := httpError.NewHttpError( @@ -365,18 +354,8 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger return nil, &err } - newSigner, err := bind.NewKeyedTransactorWithChainID(uc.config.SigningKey, chainId) - if err != nil { - log.Err(err).Msg("err encountered while creating signer") - err := httpError.NewHttpError( - "error encountered while registering identity", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - - identity := common.ComputeEventIdentity(identityPrefix[:], newSigner.From, eventTriggerDefinition) + sender := uc.txManager.From() + identity := common.ComputeEventIdentity(identityPrefix[:], sender, eventTriggerDefinition) _, err = uc.dbQuery.GetEventIdentityRegistration(ctx, data.GetEventIdentityRegistrationParams{ Eon: int64(eon), @@ -401,83 +380,121 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger return nil, &err } - publicAddress := crypto.PubkeyToAddress(*uc.config.PublicKey) - - opts := bind.TransactOpts{ - From: publicAddress, - Signer: newSigner.Signer, - } - - tx, err := uc.shutterEventRegistryContract.Register(&opts, eon, identityPrefix, eventTriggerDefinition, ttl) - if err != nil { - log.Err(err).Msg("failed to send transaction") - metrics.TotalFailedRPCCalls.Inc() - err := httpError.NewHttpError( - "failed to register identity", - "", - http.StatusInternalServerError, - ) - return nil, &err - } - // not launching a routine to monitor the transaction - // we return the transaction hash in response to allow - // users the ability to monitor it themselves - - // Store the registration in database - txHashBytes := tx.Hash().Bytes() - err = uc.dbQuery.InsertEventIdentityRegistration(ctx, data.InsertEventIdentityRegistrationParams{ + events := uc.txManager.Send(func(opts *bind.TransactOpts) (*types.Transaction, error) { + return uc.shutterEventRegistryContract.Register(opts, eon, identityPrefix, eventTriggerDefinition, ttl) + }) + registration := data.InsertEventIdentityRegistrationParams{ Eon: int64(eon), Identity: identity, IdentityPrefix: identityPrefix[:], - Sender: newSigner.From.Hex(), + Sender: sender.Hex(), EventTriggerDefinition: eventTriggerDefinition, - TxHash: txHashBytes, - }) + } + + txHash, httpErr := awaitSubmission(ctx, events) + if httpErr != nil { + // Giving up waiting does not stop the transaction, so an identity may yet + // be registered on chain that the API has no record of. Hand the request + // over instead of dropping it, and let the row be written when the + // transaction turns up. + go uc.recordEventRegistration(events, registration, ttl, false) + return nil, httpErr + } + + registration.TxHash = txHash.Bytes() + err = uc.dbQuery.InsertEventIdentityRegistration(ctx, registration) if err != nil { + // The transaction is already on its way, so the registration happens + // whether or not this worked. Answering with an error would be a lie; + // recordEventRegistration gets another attempt at the row instead. log.Err(err).Msg("err encountered while storing event identity registration") - // Note: Transaction already sent, so we log the error but don't fail the request - // The registration is on-chain even if DB insert fails } - go uc.updateEventIdentityExpirationBlockNumber(tx.Hash(), eon, identity, ttl) + go uc.recordEventRegistration(events, registration, ttl, err == nil) - metrics.TotalSuccessfulIdentityRegistration.Inc() + metrics.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ Eon: eon, Identity: common.PrefixWith0x(hex.EncodeToString(identity)), IdentityPrefix: common.PrefixWith0x(hex.EncodeToString(identityPrefix[:])), EonKey: common.PrefixWith0x(hex.EncodeToString(eonKeyBytes)), - TxHash: tx.Hash().Hex(), + TxHash: txHash.Hex(), }, nil } -func (uc *CryptoUsecase) updateEventIdentityExpirationBlockNumber(txHash ecommon.Hash, eon uint64, identity []byte, ttl uint64) { +// recordEventRegistration follows a registration to its conclusion and keeps the +// database in step with it. It runs in a goroutine of its own because it outlives +// the request by design: the response is committed as soon as there is a hash, +// while the transaction is only mined blocks later. +// +// inserted says whether the row already exists. A request that gave up waiting, or +// whose own insert failed, leaves it to this, so that an identity registered on +// chain is not one the API has no record of. +func (uc *CryptoUsecase) recordEventRegistration( + events <-chan txmgr.Event, + registration data.InsertEventIdentityRegistrationParams, + ttl uint64, + inserted bool, +) { + // Deliberately not the request's context, which is cancelled once the response + // is written. Reading stops on its own when the manager ends the request. ctx := context.Background() - ticker := time.NewTicker(1 * time.Second) - defer ticker.Stop() - - for { - receipt, err := uc.ethClient.TransactionReceipt(ctx, txHash) - if err == nil { - if receipt.Status == 0 { - log.Error().Str("tx_hash", txHash.Hex()).Msg("event identity registration transaction failed") - return - } - - expirationBlockNumber := receipt.BlockNumber.Uint64() + ttl - err = uc.dbQuery.UpdateEventIdentityRegistrationExpirationBlockNumber(ctx, data.UpdateEventIdentityRegistrationExpirationBlockNumberParams{ - ExpirationBlockNumber: int64(expirationBlockNumber), - Eon: int64(eon), - Identity: identity, - }) - if err != nil { - log.Err(err).Str("tx_hash", txHash.Hex()).Msg("failed to update expiration block number") + for ev := range events { + switch { + case ev.Tx != nil: + registration.TxHash = ev.Tx.Hash().Bytes() + if !inserted { + inserted = uc.insertEventRegistration(ctx, registration) + } + case ev.Receipt != nil: + registration.TxHash = ev.Receipt.TxHash.Bytes() + if !inserted { + uc.insertEventRegistration(ctx, registration) } - return + uc.recordExpirationBlockNumber(ctx, registration, ttl, ev.Receipt) + case ev.Err != nil: + log.Err(ev.Err). + Str("identity", hex.EncodeToString(registration.Identity)). + Msg("event identity registration was not mined") } + } +} - <-ticker.C +// insertEventRegistration writes the registration row and reports whether that +// worked. A failure is logged rather than retried, because the next event is +// another chance at it. +func (uc *CryptoUsecase) insertEventRegistration(ctx context.Context, registration data.InsertEventIdentityRegistrationParams) bool { + if err := uc.dbQuery.InsertEventIdentityRegistration(ctx, registration); err != nil { + log.Err(err).Str("tx_hash", hex.EncodeToString(registration.TxHash)). + Msg("err encountered while storing event identity registration") + return false + } + return true +} + +// recordExpirationBlockNumber fills in the block the registration expires at, +// which cannot be known before the transaction is mined. +func (uc *CryptoUsecase) recordExpirationBlockNumber( + ctx context.Context, + registration data.InsertEventIdentityRegistrationParams, + ttl uint64, + receipt *types.Receipt, +) { + if receipt.Status == types.ReceiptStatusFailed { + log.Error().Str("tx_hash", receipt.TxHash.Hex()). + Msg("event identity registration transaction reverted") + return + } + + err := uc.dbQuery.UpdateEventIdentityRegistrationExpirationBlockNumber(ctx, data.UpdateEventIdentityRegistrationExpirationBlockNumberParams{ + ExpirationBlockNumber: int64(receipt.BlockNumber.Uint64() + ttl), + Eon: registration.Eon, + Identity: registration.Identity, + }) + if err != nil { + log.Err(err).Str("tx_hash", receipt.TxHash.Hex()). + Msg("failed to update expiration block number") } } @@ -503,8 +520,7 @@ func (uc *CryptoUsecase) GetEventTriggerExpirationBlock(ctx context.Context, eon return nil, &err } - address := crypto.PubkeyToAddress(uc.config.SigningKey.PublicKey) - sender := address.Hex() + sender := uc.txManager.From().Hex() expirationBlockNumber, err := uc.dbQuery.GetEventTriggerExpirationBlockNumber(ctx, data.GetEventTriggerExpirationBlockNumberParams{ Eon: int64(eon), @@ -560,7 +576,7 @@ func (uc *CryptoUsecase) GetEventDecryptionKey(ctx context.Context, identity str blockNumber, err := uc.ethClient.BlockNumber(ctx) if err != nil { log.Err(err).Msg("err encountered while querying for recent block") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying for recent block", "", @@ -572,7 +588,7 @@ func (uc *CryptoUsecase) GetEventDecryptionKey(ctx context.Context, identity str eonUint, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber) if err != nil { log.Err(err).Msg("err encountered while querying current eon") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying current eon", "", diff --git a/internal/usecase/submit.go b/internal/usecase/submit.go new file mode 100644 index 0000000..4e6a8fd --- /dev/null +++ b/internal/usecase/submit.go @@ -0,0 +1,96 @@ +package usecase + +import ( + "context" + "errors" + "net/http" + "time" + + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/rs/zerolog/log" + httpError "github.com/shutter-network/shutter-api/internal/error" + "github.com/shutter-network/shutter-api/internal/txmgr" + "github.com/shutter-network/shutter-api/metrics" +) + +// TxManagerInterface is the part of txmgr.Manager the registration endpoints use. +type TxManagerInterface interface { + // From is the address registrations are sent from, and the one identities are + // derived from. + From() ecommon.Address + // Send queues a call and returns without sending it. Nothing has reached the + // node when it returns, so the transaction hash arrives on the channel rather + // than from the call. + Send(submit txmgr.SubmitFunc) <-chan txmgr.Event +} + +// submitTimeout bounds how long a registration waits for its transaction to reach +// the node. It has to cover assigning a nonce, reading gas prices and +// broadcasting, so a handful of RPC round trips, plus the wait behind any +// submissions already queued in front of it, since one goroutine performs them in +// sequence. It does not cover waiting for a block, because the response carries a +// transaction hash rather than a receipt. +// +// Generous rather than tight, because giving up does not cancel anything. The +// transaction is still sent, so a client told its registration failed may find it +// on chain regardless, and waiting a little longer costs far less than that does. +const submitTimeout = 5 * time.Second + +// awaitSubmission waits for the transaction a registration produced and returns +// its hash, or the error to answer the request with. +// +// The reason never reaches the client, only the log: it describes our node and our +// queue, which is nothing a client can act on. A full queue is the exception, +// because "overloaded, come back later" is something it can. +// +// Later events are left on the channel for whoever wants them, including the one +// that says how the transaction ended. +func awaitSubmission(ctx context.Context, events <-chan txmgr.Event) (ecommon.Hash, *httpError.Http) { + ctx, cancel := context.WithTimeout(ctx, submitTimeout) + defer cancel() + + select { + case ev, open := <-events: + switch { + case !open: + // The manager reports an outcome before closing, always, so this is a + // bug in it rather than a state to handle. + log.Error().Msg("transaction manager closed a request without reporting anything") + case ev.Tx != nil: + return ev.Tx.Hash(), nil + case ev.Receipt != nil: + // Already mined, which is unlikely this early but not a problem: the + // receipt names the transaction just as well as the transaction does. + return ev.Receipt.TxHash, nil + case errors.Is(ev.Err, txmgr.ErrQueueFull): + log.Err(ev.Err).Msg("rejecting registration, too many are already in flight") + err := httpError.NewHttpError( + "too many registrations in flight, please retry", + "", + http.StatusServiceUnavailable, + ) + return ecommon.Hash{}, &err + default: + log.Err(ev.Err).Msg("failed to submit registration") + } + case <-ctx.Done(): + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + // The request is still queued and will be submitted, so the + // registration may land after this answer says it did not. Counted + // because that is a discrepancy worth reconciling rather than a + // failure worth ignoring. + metrics.SubmissionTimeouts.Inc() + log.Error().Dur("timeout", submitTimeout). + Msg("gave up waiting for a registration to be submitted, it may still be sent") + } else { + log.Debug().Msg("caller went away before the registration was submitted") + } + } + + err := httpError.NewHttpError( + "failed to register identity", + "", + http.StatusInternalServerError, + ) + return ecommon.Hash{}, &err +} diff --git a/main.go b/main.go index ed850af..01b3cef 100644 --- a/main.go +++ b/main.go @@ -9,6 +9,7 @@ import ( shutterAPICommon "github.com/shutter-network/shutter-api/common" "github.com/shutter-network/shutter-api/common/database" "github.com/shutter-network/shutter-api/internal/router" + "github.com/shutter-network/shutter-api/internal/txmgr" "github.com/shutter-network/shutter-api/metrics" "github.com/ethereum/go-ethereum/common" @@ -194,13 +195,29 @@ func main() { if config.DisableEventAPI { log.Info().Msg("Event API disabled: SHUTTER_EVENT_REGISTRY_CONTRACT_ADDRESS not configured") } - app := router.NewRouter(ctx, db, contract, client, config) + // One manager for the whole process: it owns the nonce sequence of the signing + // key, which only works if every registration goes through the same one. + chainID, err := client.ChainID(ctx) + if err != nil { + log.Err(err).Msg("failed to query chain id") + return + } + txManager, err := txmgr.NewManager(client, signingKey, chainID, txmgr.DefaultConfig()) + if err != nil { + log.Err(err).Msg("failed to instantiate transaction manager") + return + } + + app := router.NewRouter(ctx, db, contract, client, txManager, config) watcher := watcher.NewWatcher(config, db) - group, deferFn := service.RunBackground(ctx, watcher) + group, deferFn := service.RunBackground(ctx, watcher, txManager) defer deferFn() if metricsConfig.Enabled { - group, deferFn := service.RunBackground(ctx, metricsServer) + signerAddress := crypto.PubkeyToAddress(*config.PublicKey) + balancePoller := metrics.NewBalancePoller(client, signerAddress) + + group, deferFn := service.RunBackground(ctx, metricsServer, balancePoller) defer deferFn() go func() { if err := group.Wait(); err != nil { diff --git a/metrics/balance.go b/metrics/balance.go new file mode 100644 index 0000000..cf79c74 --- /dev/null +++ b/metrics/balance.go @@ -0,0 +1,100 @@ +package metrics + +import ( + "context" + "math/big" + "time" + + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/prometheus/client_golang/prometheus" + "github.com/rs/zerolog/log" + "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service" +) + +const ( + balancePollInterval = 60 * time.Second + balancePollTimeout = 10 * time.Second +) + +// weiPerEther is the divisor turning a wei balance into ether. +var weiPerEther = new(big.Float).SetFloat64(1e18) + +var SignerBalanceEther = newSignerBalanceGauge() + +// A GaugeVec with no labels, so the series stays absent until a balance has +// been read. A plain Gauge would be registered holding 0, and a restart while +// the RPC endpoint is down would publish 0 ether and fire the low-balance +// alert. The exposed series is the same either way. +func newSignerBalanceGauge() *prometheus.GaugeVec { + return prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "shutter_api", + Name: "signer_balance_ether", + Help: "Balance of the signer account in ether.", + }, + []string{}, + ) +} + +func initBalanceMetrics() { + prometheus.MustRegister(SignerBalanceEther) +} + +// BalanceReader reads an account balance from the chain. It is the single +// method of ethclient.Client that the poller needs. +type BalanceReader interface { + BalanceAt(ctx context.Context, account ecommon.Address, blockNumber *big.Int) (*big.Int, error) +} + +// BalancePoller publishes the signer's balance as a gauge. The signer pays gas +// for every identity registration, so an empty account takes registration down; +// without this the service spends from an account it cannot observe. +type BalancePoller struct { + client BalanceReader + address ecommon.Address +} + +func NewBalancePoller(client BalanceReader, address ecommon.Address) *BalancePoller { + return &BalancePoller{client: client, address: address} +} + +func (p *BalancePoller) Start(ctx context.Context, runner service.Runner) error { + runner.Go(func() error { + ticker := time.NewTicker(balancePollInterval) + defer ticker.Stop() + + // Publish once up front so the series exists from the first scrape + // rather than only after a full interval. + p.poll(ctx) + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + p.poll(ctx) + } + } + }) + return nil +} + +// poll reads the balance and updates the gauge. A failed read leaves the gauge +// at its previous value: publishing a zero would look like an empty account and +// fire the very alert this metric exists to raise. Errors are never returned, +// because the poller shares an error group with the API and a transient RPC +// failure must not shut the service down. +func (p *BalancePoller) poll(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, balancePollTimeout) + defer cancel() + + wei, err := p.client.BalanceAt(ctx, p.address, nil) + if err != nil { + log.Err(err).Str("address", p.address.Hex()).Msg("failed to query signer balance") + FailedRPCCalls.Inc() + return + } + + ether, _ := new(big.Float).Quo(new(big.Float).SetInt(wei), weiPerEther).Float64() + SignerBalanceEther.WithLabelValues().Set(ether) +} diff --git a/metrics/balance_test.go b/metrics/balance_test.go new file mode 100644 index 0000000..601a4f4 --- /dev/null +++ b/metrics/balance_test.go @@ -0,0 +1,83 @@ +package metrics + +import ( + "context" + "errors" + "math/big" + "testing" + "time" + + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// mockBalanceReader is local to this test: the poller needs a single method, +// while tests/mock.MockEthClient serves the usecase's wider interface. +type mockBalanceReader struct { + mock.Mock +} + +func (m *mockBalanceReader) BalanceAt(ctx context.Context, account ecommon.Address, blockNumber *big.Int) (*big.Int, error) { + args := m.Called(ctx, account, blockNumber) + balance, _ := args.Get(0).(*big.Int) + return balance, args.Error(1) +} + +// Deliberately not a real signer address. +var testSignerAddress = ecommon.HexToAddress("0x1111111111111111111111111111111111111111") + +// A registered-but-never-read balance must expose nothing at all. Publishing a +// zero would be read as an empty signer account. +func TestBalanceIsAbsentBeforeTheFirstRead(t *testing.T) { + require.Zero(t, testutil.CollectAndCount(newSignerBalanceGauge())) +} + +func TestPollPublishesBalanceInEther(t *testing.T) { + // 1.5 ether, to catch a big.Int quotient truncating the fraction away. + wei := new(big.Int).Add( + new(big.Int).SetUint64(1e18), + new(big.Int).SetUint64(5e17), + ) + + client := &mockBalanceReader{} + client.On("BalanceAt", mock.Anything, testSignerAddress, (*big.Int)(nil)).Return(wei, nil) + + NewBalancePoller(client, testSignerAddress).poll(context.Background()) + + require.InDelta(t, 1.5, testutil.ToFloat64(SignerBalanceEther.WithLabelValues()), 1e-9) + client.AssertExpectations(t) +} + +func TestPollLeavesGaugeUntouchedOnError(t *testing.T) { + SignerBalanceEther.WithLabelValues().Set(2) + failuresBefore := testutil.ToFloat64(FailedRPCCalls) + + client := &mockBalanceReader{} + client.On("BalanceAt", mock.Anything, testSignerAddress, (*big.Int)(nil)). + Return(nil, errors.New("rpc unavailable")) + + NewBalancePoller(client, testSignerAddress).poll(context.Background()) + + // A zero here would read as an empty account and fire a false alert. + require.InDelta(t, 2.0, testutil.ToFloat64(SignerBalanceEther.WithLabelValues()), 1e-9) + require.Equal(t, failuresBefore+1, testutil.ToFloat64(FailedRPCCalls)) +} + +// The read must carry its own deadline, so a hung RPC cannot stall the loop +// and stop every later poll. +func TestPollBoundsTheReadWithATimeout(t *testing.T) { + var got context.Context + + client := &mockBalanceReader{} + client.On("BalanceAt", mock.Anything, testSignerAddress, (*big.Int)(nil)). + Run(func(args mock.Arguments) { got = args.Get(0).(context.Context) }). + Return(big.NewInt(0), nil) + + NewBalancePoller(client, testSignerAddress).poll(context.Background()) + + deadline, ok := got.Deadline() + require.True(t, ok, "read was made without a deadline") + require.LessOrEqual(t, time.Until(deadline), balancePollTimeout) +} diff --git a/metrics/metrics.go b/metrics/metrics.go index b1cb12c..d03a8ce 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -2,32 +2,34 @@ package metrics import "github.com/prometheus/client_golang/prometheus" -var TotalSuccessfulIdentityRegistration = prometheus.NewGauge( - prometheus.GaugeOpts{ +var SuccessfulIdentityRegistrations = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", - Name: "total_successful_identities_registration", - Help: "counter of successful identity registration", + Name: "successful_identity_registrations_total", + Help: "Count of successful identity registrations.", }, ) -var TotalDecryptionKeysReceived = prometheus.NewGauge( - prometheus.GaugeOpts{ +var DecryptionKeysReceived = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", - Name: "total_decryption_keys_received", - Help: "counter of total dec keys received", + Name: "decryption_keys_received_total", + Help: "Count of decryption keys received from the keypers.", }, ) -var TotalFailedRPCCalls = prometheus.NewGauge( - prometheus.GaugeOpts{ +var FailedRPCCalls = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", - Name: "total_failed_rpc_calls", - Help: "Counter of failed rpc calls", + Name: "failed_rpc_calls_total", + Help: "Count of failed RPC calls.", }, ) func InitMetrics() { - prometheus.MustRegister(TotalSuccessfulIdentityRegistration) - prometheus.MustRegister(TotalDecryptionKeysReceived) - prometheus.MustRegister(TotalFailedRPCCalls) + prometheus.MustRegister(SuccessfulIdentityRegistrations) + prometheus.MustRegister(DecryptionKeysReceived) + prometheus.MustRegister(FailedRPCCalls) + initBalanceMetrics() + initTransactionMetrics() } diff --git a/metrics/transactions.go b/metrics/transactions.go new file mode 100644 index 0000000..7d6ec31 --- /dev/null +++ b/metrics/transactions.go @@ -0,0 +1,76 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +// Terminal states a send request can reach, used as the status label on +// TransactionsResolved. Exported so the transaction manager and the tests agree +// on the spelling. +// +// These are finer grained than what the transaction manager reports to its +// callers, which is a receipt or an error. A revert is the contract's verdict +// rather than the manager's, so callers read it off the receipt, but a revert +// rate is still worth alerting on. Rejected and abandoned are one error to a +// caller that only wants to know it will not be mined, and two series here. +const ( + TxStatusConfirmed = "confirmed" + TxStatusReverted = "reverted" + TxStatusAbandoned = "abandoned" + TxStatusRejected = "rejected" +) + +// txStatuses is every value TransactionsResolved is ever incremented with. +var txStatuses = []string{ + TxStatusConfirmed, + TxStatusReverted, + TxStatusAbandoned, + TxStatusRejected, +} + +// TransactionsResolved counts send requests by the state they finished in. +// SuccessfulIdentityRegistrations counts submissions, which says nothing about +// whether a transaction was mined; this is the counter that does. +var TransactionsResolved = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "shutter_api", + Name: "transactions_resolved_total", + Help: "Count of send requests by terminal state.", + }, + []string{"status"}, +) + +// SubmissionTimeouts counts requests answered with an error while their +// transaction was still on its way to the node. Giving up waiting does not stop +// the transaction, so each one of these is a client told its registration failed +// that may have landed on chain anyway, and a candidate for reconciliation. +var SubmissionTimeouts = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: "shutter_api", + Name: "submission_timeouts_total", + Help: "Count of requests that gave up waiting for their transaction to be submitted.", + }, +) + +// PendingTransactions is the number of submitted transactions still waiting for +// a receipt. The transaction manager never gives up on one, so a value that +// climbs and does not fall means registrations are being accepted but not +// mined, and every later registration is queued behind them. +var PendingTransactions = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: "shutter_api", + Name: "pending_transactions", + Help: "Number of submitted transactions still awaiting a receipt.", + }, +) + +// initTransactionMetrics registers the transaction metrics and creates every +// status series up front, so a rate() over a state that has not happened yet +// reads as zero instead of returning no data. +func initTransactionMetrics() { + prometheus.MustRegister(TransactionsResolved) + prometheus.MustRegister(SubmissionTimeouts) + prometheus.MustRegister(PendingTransactions) + + for _, status := range txStatuses { + TransactionsResolved.WithLabelValues(status) + } +} diff --git a/tests/init_test.go b/tests/init_test.go index 8a3bfe8..e81f1eb 100644 --- a/tests/init_test.go +++ b/tests/init_test.go @@ -27,6 +27,7 @@ type TestShutterService struct { keyperSetManagerContract *mock.MockKeyperSetManager keyBroadcastContract *mock.MockKeyBroadcast ethClient *mock.MockEthClient + txManager *mock.MockTxManager } func TestShutterServiceSuite(t *testing.T) { @@ -59,7 +60,8 @@ func (s *TestShutterService) SetupSuite() { s.keyBroadcastContract = new(mock.MockKeyBroadcast) s.keyperSetManagerContract = new(mock.MockKeyperSetManager) s.ethClient = new(mock.MockEthClient) - s.cryptoUsecase = usecase.NewCryptoUsecase(s.testDB.DbInstance, s.shutterRegistryContract, s.shutterEventRegistryContract, s.keyperSetManagerContract, s.keyBroadcastContract, s.ethClient, s.config) + s.txManager = mock.NewMockTxManager(crypto.PubkeyToAddress(*publicKey)) + s.cryptoUsecase = usecase.NewCryptoUsecase(s.testDB.DbInstance, s.shutterRegistryContract, s.shutterEventRegistryContract, s.keyperSetManagerContract, s.keyBroadcastContract, s.ethClient, s.txManager, s.config) } func (s *TestShutterService) BeforeTest(suiteName, testName string) { @@ -68,6 +70,7 @@ func (s *TestShutterService) BeforeTest(suiteName, testName string) { s.keyBroadcastContract.ExpectedCalls = nil s.keyperSetManagerContract.ExpectedCalls = nil s.ethClient.ExpectedCalls = nil + s.txManager.Reset() } func generateRandomETHAccount() (*ecdsa.PrivateKey, *ecdsa.PublicKey, string, error) { diff --git a/tests/integration/init_test.go b/tests/integration/init_test.go index e61affe..444e738 100644 --- a/tests/integration/init_test.go +++ b/tests/integration/init_test.go @@ -23,6 +23,7 @@ import ( "github.com/shutter-network/shutter-api/common/database" "github.com/shutter-network/shutter-api/internal/data" "github.com/shutter-network/shutter-api/internal/router" + "github.com/shutter-network/shutter-api/internal/txmgr" "github.com/shutter-network/shutter-api/watcher" "github.com/stretchr/testify/suite" ) @@ -105,12 +106,21 @@ func (s *TestShutterService) SetupSuite() { s.Require().NoError(database.RunMigrations(ctx, dbURL, migrationsPath)) watcher := watcher.NewWatcher(s.config, s.db) - group, deferFn := service.RunBackground(ctx, watcher) + + // The real manager against the real chain, as production runs it: the + // registration endpoints have no transaction to report until its worker + // submits one. + chainID, err := s.ethClient.ChainID(ctx) + s.Require().NoError(err) + txManager, err := txmgr.NewManager(s.ethClient, signingKey, chainID, txmgr.DefaultConfig()) + s.Require().NoError(err) + + group, deferFn := service.RunBackground(ctx, watcher, txManager) defer deferFn() go func() { s.Require().NoError(group.Wait()) }() - s.router = router.NewRouter(ctx, s.db, s.contract, s.ethClient, s.config) + s.router = router.NewRouter(ctx, s.db, s.contract, s.ethClient, txManager, s.config) s.testServer = httptest.NewServer(s.router) } diff --git a/tests/mock/tx_manager.go b/tests/mock/tx_manager.go new file mode 100644 index 0000000..ea7a3e5 --- /dev/null +++ b/tests/mock/tx_manager.go @@ -0,0 +1,67 @@ +package mock + +import ( + "github.com/ethereum/go-ethereum/accounts/abi/bind" + ecommon "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/shutter-network/shutter-api/internal/txmgr" +) + +// MockTxManager stands in for txmgr.Manager. A fake rather than a testify mock, +// because what a test needs from it is the timing rather than the return value: +// Send has to invoke the submit closure and announce what comes back, which is +// what the real manager's worker goroutine does. +type MockTxManager struct { + from ecommon.Address + // Err is returned instead of submitting, standing in for a manager that could + // not get the request onto the chain. + Err error + // Events is every channel Send handed out, so a test can drive a request past + // submission by publishing a receipt on it. + Events []chan txmgr.Event +} + +func NewMockTxManager(from ecommon.Address) *MockTxManager { + return &MockTxManager{from: from} +} + +func (m *MockTxManager) From() ecommon.Address { + return m.from +} + +// Send submits immediately and announces the result, so a caller waiting for a +// hash finds one already there. The channel is left open, as the real one is +// until the transaction resolves. +func (m *MockTxManager) Send(submit txmgr.SubmitFunc) <-chan txmgr.Event { + events := make(chan txmgr.Event, 8) + m.Events = append(m.Events, events) + + if m.Err != nil { + events <- txmgr.Event{Err: m.Err} + close(events) + return events + } + + tx, err := submit(&bind.TransactOpts{From: m.from, Nonce: nil}) + if err != nil { + events <- txmgr.Event{Err: err} + close(events) + return events + } + events <- txmgr.Event{Tx: tx} + return events +} + +// Resolve ends the most recent request with a receipt, as the manager does once +// the transaction is mined. +func (m *MockTxManager) Resolve(receipt *types.Receipt) { + events := m.Events[len(m.Events)-1] + events <- txmgr.Event{Receipt: receipt} + close(events) +} + +// Reset drops the requests remembered from earlier tests. +func (m *MockTxManager) Reset() { + m.Events = nil + m.Err = nil +} diff --git a/tests/register_event_identity_test.go b/tests/register_event_identity_test.go index b111254..89e4a4d 100644 --- a/tests/register_event_identity_test.go +++ b/tests/register_event_identity_test.go @@ -8,10 +8,12 @@ import ( "time" "github.com/ethereum/go-ethereum/accounts/abi/bind" + ethCommon "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/core/types" "github.com/shutter-network/shutter-api/common" "github.com/shutter-network/shutter-api/internal/data" + "github.com/shutter-network/shutter-api/internal/usecase" "github.com/stretchr/testify/mock" ) @@ -26,7 +28,7 @@ func (s *TestShutterService) TestRegisterEventIdentity() { eon := rand.Uint64() // Hardcoded valid event trigger definition - eventTriggerDefinitionHex := "0x01f86694953a0425accee2e05f22e78999c595ed2ee7183cf84fe480e205a0ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3efe401e205a0000000000000000000000000812a6755975485c6e340f97de6790b34a94d1430c404c20402" + eventTriggerDefinitionHex := s.validEventTriggerDefinition() eventTriggerDefinitionBytes, err := hexutil.Decode(eventTriggerDefinitionHex) s.Require().NoError(err) @@ -54,33 +56,28 @@ func (s *TestShutterService) TestRegisterEventIdentity() { Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - s.shutterEventRegistryContract. On("Register", mock.Anything, eon, [32]byte(identityPrefix), eventTriggerDefinitionBytes, ttl). Return(randomTx, nil). Once() - // Mock transaction receipt - use a different block number for the receipt - txBlockNumber := blockNumber + 5 // Transaction mined in a later block + // The transaction is mined in a later block, which is where the expiration is + // measured from. + txBlockNumber := blockNumber + 5 receipt := &types.Receipt{ Status: types.ReceiptStatusSuccessful, BlockNumber: big.NewInt(int64(txBlockNumber)), + TxHash: randomTx.Hash(), } - - s.ethClient. - On("TransactionReceipt", mock.Anything, randomTx.Hash()). - Return(receipt, nil). - Once() - expectedExpirationBlockNumber := int64(txBlockNumber + ttl) response, err := s.cryptoUsecase.RegisterEventIdentity(ctx, eventTriggerDefinitionHex, identityPrefixStringified, ttl) s.Require().Nil(err) + // The expiration block is only knowable once the transaction is mined, so the + // manager reports it after the response has already gone out. + s.txManager.Resolve(receipt) + s.Require().Equal(response.Eon, eon) s.Require().Equal(common.PrefixWith0x(hex.EncodeToString(identity)), response.Identity) s.Require().Equal(common.PrefixWith0x(hex.EncodeToString(identityPrefix)), response.IdentityPrefix) @@ -112,7 +109,7 @@ func (s *TestShutterService) TestRegisterEventIdentity() { func (s *TestShutterService) TestRegisterEventIdentity_InvalidIdentityPrefix() { ctx := context.Background() ttl := uint64(100) - eventTriggerDefinitionHex := "0x01f86694953a0425accee2e05f22e78999c595ed2ee7183cf84fe480e205a0ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3efe401e205a0000000000000000000000000812a6755975485c6e340f97de6790b34a94d1430c404c20402" + eventTriggerDefinitionHex := s.validEventTriggerDefinition() // Test with invalid identity prefix length invalidIdentityPrefix := "0x1234" // Too short @@ -162,11 +159,6 @@ func (s *TestShutterService) TestRegisterEventIdentity_InvalidEventTriggerDefini Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - _, httpErr := s.cryptoUsecase.RegisterEventIdentity(ctx, invalidEventTriggerDefinitionHex, identityPrefixStringified, ttl) s.Require().NotNil(httpErr) @@ -211,11 +203,6 @@ func (s *TestShutterService) TestRegisterEventIdentity_TriggerDefinitionWithout0 Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - _, httpErr := s.cryptoUsecase.RegisterEventIdentity(ctx, eventTriggerDefinitionHexWithoutPrefix, identityPrefixStringified, ttl) s.Require().NotNil(httpErr) @@ -261,11 +248,6 @@ func (s *TestShutterService) TestRegisterEventIdentity_ZeroBytesEventTriggerDefi Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - _, httpErr := s.cryptoUsecase.RegisterEventIdentity(ctx, zeroEventTriggerDefinitionHex, identityPrefixStringified, ttl) s.Require().NotNil(httpErr) @@ -282,7 +264,7 @@ func (s *TestShutterService) TestRegisterEventIdentity_EmptyIdentityPrefix() { eon := rand.Uint64() // Hardcoded valid event trigger definition - eventTriggerDefinitionHex := "0x01f86694953a0425accee2e05f22e78999c595ed2ee7183cf84fe480e205a0ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3efe401e205a0000000000000000000000000812a6755975485c6e340f97de6790b34a94d1430c404c20402" + eventTriggerDefinitionHex := s.validEventTriggerDefinition() eventTriggerDefinitionBytes, err := hexutil.Decode(eventTriggerDefinitionHex) s.Require().NoError(err) @@ -316,11 +298,6 @@ func (s *TestShutterService) TestRegisterEventIdentity_EmptyIdentityPrefix() { Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - // Mock will be called with the generated identity prefix (we can't predict it, so use mock.MatchedBy) s.shutterEventRegistryContract. On("Register", mock.Anything, eon, mock.MatchedBy(func(prefix [32]byte) bool { @@ -363,7 +340,7 @@ func (s *TestShutterService) TestRegisterEventIdentity_AlreadyRegistered() { eon := rand.Uint64() // Hardcoded valid event trigger definition - eventTriggerDefinitionHex := "0x01f86694953a0425accee2e05f22e78999c595ed2ee7183cf84fe480e205a0ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3efe401e205a0000000000000000000000000812a6755975485c6e340f97de6790b34a94d1430c404c20402" + eventTriggerDefinitionHex := s.validEventTriggerDefinition() eventTriggerDefinitionBytes, err := hexutil.Decode(eventTriggerDefinitionHex) s.Require().NoError(err) @@ -392,11 +369,6 @@ func (s *TestShutterService) TestRegisterEventIdentity_AlreadyRegistered() { Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - s.shutterEventRegistryContract. On("Register", mock.Anything, eon, [32]byte(identityPrefix), eventTriggerDefinitionBytes, ttl). Return(randomTx, nil). @@ -445,14 +417,24 @@ func (s *TestShutterService) TestRegisterEventIdentity_AlreadyRegistered() { Return(eonPublicKey.Marshal(), nil). Once() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Once() - // Second registration should fail with "event identity already registered" _, httpErr := s.cryptoUsecase.RegisterEventIdentity(ctx, eventTriggerDefinitionHex, identityPrefixStringified, ttl) s.Require().NotNil(httpErr) s.Require().Equal("event identity already registered", httpErr.Description) s.Require().Equal(400, httpErr.StatusCode) } + +// validEventTriggerDefinition compiles a definition instead of hardcoding one, so +// the fixture cannot fall behind the encoding version the keypers expect. +func (s *TestShutterService) validEventTriggerDefinition() string { + resp, errs := usecase.CompileEventTriggerDefinitionInternal(usecase.EventTriggerDefinitionRequest{ + ContractAddress: ethCommon.HexToAddress("0x953A0425ACCee2E05f22E78999c595eD2eE7183c"), + EventSignature: "event Transfer(address indexed from, address indexed to, uint256 amount)", + Arguments: []usecase.EventArgument{ + {Name: "from", Operator: "eq", Bytes: "0x812a6755975485C6E340F97dE6790B34a94D1430"}, + {Name: "amount", Operator: "gte", Number: "2"}, + }, + }) + s.Require().Empty(errs) + return resp.EventTriggerDefinition +} diff --git a/tests/register_identity_test.go b/tests/register_identity_test.go index b7a2654..127c1d8 100644 --- a/tests/register_identity_test.go +++ b/tests/register_identity_test.go @@ -48,11 +48,6 @@ func (s *TestShutterService) TestRegisterIdentity() { Return(eonPublicKey.Marshal(), nil). Twice() - s.ethClient. - On("ChainID", ctx). - Return(big.NewInt(GnosisMainnetChainID), nil). - Twice() - s.shutterRegistryContract. On("Registrations", mock.AnythingOfType("*bind.CallOpts"), [32]byte(identity)). Return(struct { diff --git a/watcher/watcher.go b/watcher/watcher.go index d5603dc..a282fd9 100644 --- a/watcher/watcher.go +++ b/watcher/watcher.go @@ -46,7 +46,7 @@ func (w *Watcher) Start(ctx context.Context, runner service.Runner) error { }); err != nil { log.Err(err).Msg("failed to insert decryption key") } - metrics.TotalDecryptionKeysReceived.Inc() + metrics.DecryptionKeysReceived.Inc() } } }