From 42e83452f363d6bedda5f6993069ee85f0a2c813 Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Wed, 12 Aug 2026 19:04:25 +0200 Subject: [PATCH 1/6] metrics: use counters instead of gauges for monotonic totals TotalSuccessfulIdentityRegistration, TotalDecryptionKeysReceived and TotalFailedRPCCalls are only ever incremented, so Counter is the correct type. Series names are unchanged, so existing queries keep working, and rate() over them is now legitimate rather than an accident. --- metrics/metrics.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/metrics/metrics.go b/metrics/metrics.go index b1cb12c..5266130 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -2,24 +2,24 @@ package metrics import "github.com/prometheus/client_golang/prometheus" -var TotalSuccessfulIdentityRegistration = prometheus.NewGauge( - prometheus.GaugeOpts{ +var TotalSuccessfulIdentityRegistration = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", Name: "total_successful_identities_registration", Help: "counter of successful identity registration", }, ) -var TotalDecryptionKeysReceived = prometheus.NewGauge( - prometheus.GaugeOpts{ +var TotalDecryptionKeysReceived = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", Name: "total_decryption_keys_received", Help: "counter of total dec keys received", }, ) -var TotalFailedRPCCalls = prometheus.NewGauge( - prometheus.GaugeOpts{ +var TotalFailedRPCCalls = prometheus.NewCounter( + prometheus.CounterOpts{ Namespace: "shutter_api", Name: "total_failed_rpc_calls", Help: "Counter of failed rpc calls", From adc5105130b3606c2dcc4700878a792d2ef58414 Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Fri, 14 Aug 2026 21:31:39 +0200 Subject: [PATCH 2/6] metrics: rename counters to the _total suffix convention Prometheus counters carry _total as a suffix, not a prefix, and the plural belongs on the thing being counted: shutter_api_total_successful_identities_registration -> shutter_api_successful_identity_registrations_total shutter_api_total_decryption_keys_received -> shutter_api_decryption_keys_received_total shutter_api_total_failed_rpc_calls -> shutter_api_failed_rpc_calls_total Go identifiers drop their now-redundant Total prefix to match. Existing series keep their history under the old names but stop being written to, so any dashboard or alert rule querying them needs updating. --- internal/usecase/crypto.go | 22 +++++++++++----------- internal/usecase/eventtrigger.go | 16 ++++++++-------- metrics/metrics.go | 24 ++++++++++++------------ watcher/watcher.go | 2 +- 4 files changed, 32 insertions(+), 32 deletions(-) diff --git a/internal/usecase/crypto.go b/internal/usecase/crypto.go index 215697f..883aa64 100644 --- a/internal/usecase/crypto.go +++ b/internal/usecase/crypto.go @@ -120,7 +120,7 @@ func (uc *CryptoUsecase) getSigner(ctx context.Context) (*bind.TransactOpts, *ht chainID, err := uc.ethClient.ChainID(ctx) if err != nil { log.Err(err).Msg("err encountered while querying chain id") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying chain id", "", @@ -168,7 +168,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 +301,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 +313,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 +325,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", "", @@ -442,7 +442,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 +454,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 +466,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", "", @@ -496,7 +496,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta 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", "", @@ -525,7 +525,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta tx, err := uc.shutterRegistryContract.Register(&opts, eon, identityPrefix, decryptionTimestamp) if err != nil { log.Err(err).Msg("failed to send transaction") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "failed to register identity", "", @@ -537,7 +537,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta // we return the transaction hash in response to allow // users the ability to monitor it themselves - metrics.TotalSuccessfulIdentityRegistration.Inc() + metrics.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ Eon: eon, Identity: common.PrefixWith0x(hex.EncodeToString(identity)), diff --git a/internal/usecase/eventtrigger.go b/internal/usecase/eventtrigger.go index 4d29e7f..f3a08e5 100644 --- a/internal/usecase/eventtrigger.go +++ b/internal/usecase/eventtrigger.go @@ -288,7 +288,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 +300,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 +312,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", "", @@ -335,7 +335,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger chainId, err := uc.ethClient.ChainID(ctx) if err != nil { log.Err(err).Msg("err encountered while quering chain id") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "error encountered while querying chain id", "", @@ -411,7 +411,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger tx, err := uc.shutterEventRegistryContract.Register(&opts, eon, identityPrefix, eventTriggerDefinition, ttl) if err != nil { log.Err(err).Msg("failed to send transaction") - metrics.TotalFailedRPCCalls.Inc() + metrics.FailedRPCCalls.Inc() err := httpError.NewHttpError( "failed to register identity", "", @@ -441,7 +441,7 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger go uc.updateEventIdentityExpirationBlockNumber(tx.Hash(), eon, identity, ttl) - metrics.TotalSuccessfulIdentityRegistration.Inc() + metrics.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ Eon: eon, Identity: common.PrefixWith0x(hex.EncodeToString(identity)), @@ -560,7 +560,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 +572,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/metrics/metrics.go b/metrics/metrics.go index 5266130..5d4488c 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -2,32 +2,32 @@ package metrics import "github.com/prometheus/client_golang/prometheus" -var TotalSuccessfulIdentityRegistration = prometheus.NewCounter( +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.NewCounter( +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.NewCounter( +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) } 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() } } } From c9096d5f3a01448788d8c07478e7ead6d7d651e8 Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Sun, 16 Aug 2026 16:55:01 +0200 Subject: [PATCH 3/6] metrics: add a gauge for the signer's balance The signer pays gas for every identity registration, but the service never queried its balance, so an account draining to empty was only visible once registrations started failing. At that point it surfaced as failed_rpc_calls_total from the transaction send sites, indistinguishable from an RPC outage. shutter_api_signer_balance_ether is published by a new BalancePoller running as a service alongside the metrics server, so it is gated on METRICS_ENABLED. It reads once at startup and then every 60s, each read bounded by a 10s timeout so a hung endpoint cannot stall the loop. A failed read logs, increments failed_rpc_calls_total and leaves the gauge alone; it never returns an error, because a transient RPC failure must not bring the service down through the error group. The metric is a GaugeVec with no labels rather than a plain Gauge. A plain Gauge is registered holding 0, so a restart while the RPC endpoint was down would publish 0 ether and fire the low-balance alert this metric exists to raise. With no value set the series is simply absent, and the exposed series is otherwise identical. Balance is reported in ether rather than wei so that alert thresholds are readable; the conversion goes through big.Float, as an integer quotient would truncate everything below 1 ether to zero. Known gap: a reading that stops being refreshed keeps its last value indefinitely, so a dead poller looks healthy. Detection relies on failed_rpc_calls_total, which now also moves for background polls and can therefore rise with no traffic. Co-Authored-By: Claude --- go.mod | 1 + main.go | 5 +- metrics/balance.go | 100 ++++++++++++++++++++++++++++++++++++++++ metrics/balance_test.go | 83 +++++++++++++++++++++++++++++++++ metrics/metrics.go | 1 + 5 files changed, 189 insertions(+), 1 deletion(-) create mode 100644 metrics/balance.go create mode 100644 metrics/balance_test.go 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/main.go b/main.go index ed850af..0d26815 100644 --- a/main.go +++ b/main.go @@ -200,7 +200,10 @@ func main() { 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 5d4488c..1a1f09b 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -30,4 +30,5 @@ func InitMetrics() { prometheus.MustRegister(SuccessfulIdentityRegistrations) prometheus.MustRegister(DecryptionKeysReceived) prometheus.MustRegister(FailedRPCCalls) + initBalanceMetrics() } From 8c6faef5e38d382bbe8c9fea5b2ff9bc38d4f549 Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Mon, 24 Aug 2026 11:36:14 +0200 Subject: [PATCH 4/6] metrics: add metrics for the transaction lifecycle successful_identity_registrations_total counts submissions, which says nothing about whether a transaction was mined. These three cover what happens after the response goes out. transactions_resolved_total is labelled by terminal state, and is finer grained than what the transaction manager reports to its callers. 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 nothing will be mined, and two series here. Every status series is created at registration so that rate() over a state that has not happened yet reads as zero instead of returning no data. submission_timeouts_total 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 is a client told its registration failed that may have landed on chain anyway, and a candidate for reconciliation rather than a plain failure. pending_transactions is a gauge because it is not monotonic. A value that climbs and does not fall means registrations are being accepted but not mined, and that every later registration is queued behind them. Co-Authored-By: Claude Opus 5 (1M context) --- metrics/metrics.go | 1 + metrics/transactions.go | 76 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) create mode 100644 metrics/transactions.go diff --git a/metrics/metrics.go b/metrics/metrics.go index 1a1f09b..d03a8ce 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -31,4 +31,5 @@ func InitMetrics() { 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) + } +} From f33e3b0e628ae9395315981cd01a0071f9dd1a1b Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Mon, 24 Aug 2026 11:36:29 +0200 Subject: [PATCH 5/6] txmgr: serialize transaction submission for the signing key 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. Under any real load the API was losing registrations to that race. The Manager closes the 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 the package holds no mutexes. Polling is only observation, so a submission delaying a receipt check by a few hundred milliseconds costs nothing next to what sharing the nonce sequence would. 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 and would wedge the whole queue behind it. MaxFeeCapPercent bounds how far that escalation goes. Only the lowest-nonce transaction is resubmitted, and only one per poll. Transactions are mined in nonce order, so repricing the others cannot make them move until it does. Known limits: pending state is in memory and lost on restart, so transactions still in the pool are adopted by nonce rather than resubmitted. An account with no gas money is watched indefinitely and wedges registration, visible 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. Co-Authored-By: Claude Opus 5 (1M context) --- internal/txmgr/manager.go | 726 ++++++++++++++++++++++++ internal/txmgr/manager_test.go | 980 +++++++++++++++++++++++++++++++++ 2 files changed, 1706 insertions(+) create mode 100644 internal/txmgr/manager.go create mode 100644 internal/txmgr/manager_test.go 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()) +} From e15d8635c1df3f130c83dcd6e4c72e94f1f10476 Mon Sep 17 00:00:00 2001 From: Jannik Luhn Date: Mon, 24 Aug 2026 11:36:47 +0200 Subject: [PATCH 6/6] usecase: submit registrations through the transaction manager Both registration endpoints signed and sent their own transaction with a nil nonce, so concurrent registrations raced for the same one. They now hand a closure to a single process-wide manager, which owns the nonce sequence of the signing key. One manager for the whole process is the point: it only works if every registration goes through the same one. Submission is no longer synchronous, so the endpoints wait for the first event on the request's channel to get a hash to answer with. awaitSubmission bounds that wait at five seconds, which has to cover assigning a nonce, reading gas prices and broadcasting, plus the wait behind any submissions already queued in front of it. It does not cover waiting for a block, since the response carries a hash rather than a receipt. The bound is 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, which costs far more than waiting a little longer. A full queue is the one failure reported to the client as such, with 503, because "overloaded, come back later" is the only one a client can act on. The event endpoint's receipt poller becomes recordEventRegistration, following the same channel instead of polling TransactionReceipt on a one-second ticker. It also now covers the case where the request gave up waiting or its insert failed, so an identity registered on chain is no longer one the API has no record of. The signer address comes from the manager rather than being derived from the config key at each call site, which also removes ChainID from the per-request path: it is read once at startup instead. ChainID and TransactionReceipt leave EthClientInterface, since nothing in the usecase layer reads the chain directly any more. Co-Authored-By: Claude Opus 5 (1M context) --- internal/router/router.go | 4 +- internal/service/crypto.go | 5 +- internal/usecase/crypto.go | 80 +++--------- internal/usecase/eventtrigger.go | 178 ++++++++++++++------------ internal/usecase/submit.go | 96 ++++++++++++++ main.go | 18 ++- tests/init_test.go | 5 +- tests/integration/init_test.go | 14 +- tests/mock/tx_manager.go | 67 ++++++++++ tests/register_event_identity_test.go | 76 +++++------ tests/register_identity_test.go | 5 - 11 files changed, 344 insertions(+), 204 deletions(-) create mode 100644 internal/usecase/submit.go create mode 100644 tests/mock/tx_manager.go 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/usecase/crypto.go b/internal/usecase/crypto.go index 883aa64..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.FailedRPCCalls.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 { @@ -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) { @@ -486,12 +451,7 @@ 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 { @@ -515,27 +475,19 @@ 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.FailedRPCCalls.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.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ @@ -543,7 +495,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta 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 f3a08e5..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" ) @@ -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.FailedRPCCalls.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,45 +380,37 @@ 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.FailedRPCCalls.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.SuccessfulIdentityRegistrations.Inc() return &RegisterIdentityResponse{ @@ -447,37 +418,83 @@ func (uc *CryptoUsecase) RegisterEventIdentity(ctx context.Context, eventTrigger 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) } - return + case ev.Receipt != nil: + registration.TxHash = ev.Receipt.TxHash.Bytes() + if !inserted { + uc.insertEventRegistration(ctx, registration) + } + 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), 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 0d26815..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,9 +195,22 @@ 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 { 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 {