From 0b137729bcb0f4507eea2607d81c488cbefdd078 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Tue, 3 Nov 2026 02:40:00 -0800 Subject: [PATCH] Instrument payment phases and simplify shared chain paths Instrument payment phases and authoritative RPC work. Consolidate shared chain paths and RPC fixtures, retain CI regressions, and remove the local payment benchmarks and historical comparison harness. --- docs/metrics.md | 11 +- eth/payment.go | 43 +++++-- eth/payment_contracts.go | 48 ------- eth/payment_contracts_test.go | 2 +- eth/payment_observer.go | 35 +++--- eth/payment_observer_test.go | 71 ++++------- eth/payment_test.go | 167 ++++++++----------------- eth/pricefeed_test.go | 86 ++++++------- eth/rpc.go | 78 ++++++++++-- eth/rpc_test.go | 10 +- eth/senderwatcher.go | 4 +- eth/senderwatcher_test.go | 120 ++++++++---------- eth/transactionmanager.go | 2 +- eth/transactionmanager_test.go | 9 +- internal/observability/metrics.go | 66 +++++++++- internal/observability/metrics_test.go | 106 ++++++++++++++++ internal/test/chainrpc.go | 70 ++++++----- orchestrator/app.go | 10 +- orchestrator/app_test.go | 69 ++++------ orchestrator/metrics.go | 6 + orchestrator/metrics_test.go | 41 ++++-- orchestrator/payment_engine.go | 3 + orchestrator/payment_recipient_test.go | 31 +---- pm/ethchain.go | 22 +--- signer/app.go | 7 +- signer/app_test.go | 97 +++++--------- signer/payment_test.go | 144 ++++++++++++--------- signer/service.go | 4 + signer/service_test.go | 28 +++++ 29 files changed, 739 insertions(+), 651 deletions(-) delete mode 100644 eth/payment_contracts.go diff --git a/docs/metrics.md b/docs/metrics.md index 1fedb5e..c17c002 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -82,19 +82,18 @@ gauges; timestamps use Unix seconds. Most event families appear on first use. | Inventory | `runner_registered`, `runner_ready`, `runner_capacity`, `runner_capacity_available`, `runner_sessions{state="pending\|active"}`, `runner_last_heartbeat_timestamp_seconds`, all with app identity. Unavailable runners expose zero available capacity; heartbeat timestamps apply to dynamic runners. | | Lifecycle | `runner_events_total{event}`, `runner_health_checks_total{outcome}`, `session_reservations_total{outcome}`, `session_ends_total{reason}`, `session_duration_seconds`, with app identity. Reservations include allocations pending payment; only activated sessions contribute lifetime observations. | | Streams | `trickle_channels{kind="control\|session"}`, `trickle_buffered_bytes` for retained published segment bytes; `stream_transfer_bytes_total{app,app_id,runner_id,mode,protocol,direction}`, `stream_errors_total{app,app_id,runner_id,mode,protocol,outcome}`. Trickle requests are transport activity, not extra app invocations. | -| Payments | `payment_operations_total{operation,outcome}` for challenge/receive/refresh/charge; `payment_tickets_total{kind="accepted\|winning"}`, `payment_credit_eth_total`, `payment_charge_eth_total`, `payment_billed_seconds_total`, `payment_fixed_charges_total`, with app identity. Credit is accepted expected ticket value; charges are successfully debited service usage. | -| Signer | `signer_payments_total{type,outcome}`, `signer_tickets_total{type}`, `signer_sign_operations_total{kind,outcome}`. Signing kinds are ticket, segment, state, and orchestrator_info. Partial signing before a failed request remains visible. Successful payment/ticket totals follow required outbox persistence, even if response delivery is interrupted. | +| Payments | `payment_operations_total{operation,outcome}` for challenge/receive/refresh/charge; `payment_operation_duration_seconds{operation}` for challenge/receive/refresh; `payment_tickets_total{kind="accepted\|winning"}`, `payment_credit_eth_total`, `payment_charge_eth_total`, `payment_billed_seconds_total`, `payment_fixed_charges_total`, with app identity. Credit is accepted expected ticket value; charges are successfully debited service usage. | +| Signer | `signer_payments_total{type,outcome}`, `signer_payment_duration_seconds{type}`, `signer_tickets_total{type}`, `signer_sign_operations_total{kind,outcome}`. Signing kinds are ticket, segment, state, and orchestrator_info. Partial signing before a failed request remains visible. Successful payment/ticket totals follow required outbox persistence, even if response delivery is interrupted. | | Authorization | `signer_auth_webhook_requests_total{status,outcome}`, `signer_auth_webhook_duration_seconds`, `signer_auth_cache_hits_total`. Status zero means no webhook response; outcomes distinguish success, denied, and error. | | Outbox | Existing `signer_kafka_pending_events`, `signer_kafka_pending_bytes`, `signer_kafka_oldest_event_age_seconds`, `signer_kafka_outbox_healthy`, `signer_kafka_publish_errors_total`, `signer_kafka_enqueue_errors_total`, and `signer_kafka_storage_errors_total`. New `signer_kafka_delivered_events_total` and `signer_kafka_last_delivery_timestamp_seconds` follow successful local acknowledgement; delivery remains at least once. | | Redemption queue | `redemption_queue_tickets{state}`, `redemption_oldest_ticket_timestamp_seconds{state}`: safe parameter-expiry `waiting`, clock-eligible `eligible`, `in_flight`, `awaiting_finality`, or `unknown` without a fresh clock observation. Each ticket counts once, regardless of replacements. | | Transactions | `redemption_unresolved_transactions{phase}`, `redemption_oldest_transaction_timestamp_seconds{phase}` include historical retries/replacements until the identity is confirmed, reverted, or superseded, or its ticket is settled (`confirmed`/`used`). Ticket expiry or invalidation alone does not remove an unresolved identity. `sending` appears as `broadcast`. | | Redemption activity | `redemption_transitions_total{event}`, `redemption_broadcasts_total{outcome}`, `redemption_errors_total{error_kind}`, `redemption_settlements_total`, `redemption_settled_eth_total`, `redemption_settlement_duration_seconds`, `redemption_last_settlement_timestamp_seconds`, `redemption_last_success_timestamp_seconds`, `redemption_worker_last_completed_timestamp_seconds`. Settlement counts once per ticket, including `used` → `confirmed` upgrades. | -| Chain/pricing | `rpc_requests_total{method,outcome}`, `rpc_request_duration_seconds{method}` include direct contract calls and JSON-RPC errors over HTTP 200. `chain_snapshot_block`, `chain_snapshot_round`, `chain_snapshot_last_success_timestamp_seconds`, `price_feed_updates_total{outcome}`, `price_feed_valid_until_timestamp_seconds`, `dependency_healthy{dependency}`, `signer_deposit_eth`, `signer_reserve_eth`, `redeemer_gas_balance_eth`. Block uses the protocol L1 clock; signer reserve is total reserve observed by readiness. | +| Chain/pricing | `rpc_requests_total{method,outcome}`, `rpc_request_duration_seconds{method}`, `rpc_contract_calls_total{contract,function}`, `rpc_work_total{work,kind}` include direct contract calls and JSON-RPC errors over HTTP 200. `chain_snapshot_block`, `chain_snapshot_round`, `chain_snapshot_last_success_timestamp_seconds`, `price_feed_updates_total{outcome}`, `price_feed_valid_until_timestamp_seconds`, `dependency_healthy{dependency}`, `signer_deposit_eth`, `signer_reserve_eth`, `redeemer_gas_balance_eth`. Block uses the protocol L1 clock; signer reserve is total reserve observed by readiness. | | Discovery | `discovery_refreshes_total{source_index,outcome}`, `discovery_refresh_duration_seconds{source_index}`, `discovery_source_state{source_index,state}`, `discovery_source_failures{source_index}`, `discovery_cached_runners{source_index}`, `discovery_last_success_timestamp_seconds{source_index}`. States distinguish healthy, retained, uninitialized, and evicted sources. | -Empty queues have zero counts/oldest timestamps. Fixed exchange rates have a -validity timestamp of zero. Chain/fund gauges retain the last observation; -check snapshot freshness and dependency health alongside them. +Empty queues and fixed-rate validity timestamps are zero. Check chain/fund freshness and health alongside retained values. +Measure end-to-end performance with [e2e-runner](https://github.com/livepeer/e2e-runner). ## Queries and starting alerts diff --git a/eth/payment.go b/eth/payment.go index b12fc8f..3e7bb6f 100644 --- a/eth/payment.go +++ b/eth/payment.go @@ -45,7 +45,7 @@ func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Addres if snapshot.blockReference == nil { return false, errors.New("canonical snapshot required") } - address, err := c.Contracts.paymentAddress(ctx, snapshot, "bondingManager") + address, err := c.Contracts.ResolveAt(ctx, snapshot.blockReference, "bondingManager") if err != nil { return false, err } @@ -89,6 +89,14 @@ func (c PaymentChain) FinalizedReceipt(ctx context.Context, hash ethcommon.Hash) return receipt, nil } +// Address retention adapts livepeer/go-livepeer eth/client.go setContracts at +// 597f882cb9f4d42ad8430b0f592afd5929af4a6b (Nico Vergauwen, 829cd9047c46aef03ff5bf07b3e34be810694b74; +// Rafał Leszko, 3209dd946bb0376308044055b6caf262fd25cf61). Confirmation before +// dependent reads adapts Mike Zupper's BlueClawNetwork/livepeer-node +// eth/chainview.go, 2d95d40b8aa4699c7ca40838b6a9c312704c93db. +// Copyright (c) 2017 Livepeer; MIT notice in LICENSE. +type paymentContracts struct{ rounds, broker, bonding ethcommon.Address } + type ChainSnapshot struct { Block *big.Int // Protocol L1 block, including on Arbitrum. Number *big.Int // Execution-chain height, distinct from the L1 clock. @@ -132,14 +140,19 @@ func (c PaymentChain) snapshotAt(ctx context.Context, header *miniHeader, curren // Bind every read to the canonical block, including Controller resolution. ref := map[string]any{"blockHash": header.Hash.Hex(), "requireCanonical": true} var addresses paymentContracts - var err error - if currentRound { // Observer candidates confirm the complete registry first. - addresses, err = c.Contracts.paymentContractsAt(ctx, ref) - } else { - addresses.rounds, err = c.Contracts.ResolveAt(ctx, ref, "roundsManager") - } - if err != nil { - return ChainSnapshot{}, err + registry := []struct { + name string + to *ethcommon.Address + }{{"roundsManager", &addresses.rounds}, {"ticketBroker", &addresses.broker}, {"bondingManager", &addresses.bonding}} + if !currentRound { + registry = registry[:1] + } + for _, entry := range registry { // Confirm the complete scope before dependent reads. + address, err := c.Contracts.ResolveAt(ctx, ref, entry.name) + if err != nil { + return ChainSnapshot{}, err + } + *entry.to = address } rounds, err := contracts.NewRoundsManagerCaller(addresses.rounds, c.Contracts.callerAt(ref)) if err != nil { @@ -184,7 +197,7 @@ type SenderInfo struct { Deposit, Reserve, WithdrawRound *big.Int } -func (c PaymentChain) SenderInfo(ctx context.Context, sender, recipient ethcommon.Address) (SenderInfo, error) { +func (c PaymentChain) PayerFunds(ctx context.Context, sender, recipient ethcommon.Address) (SenderInfo, error) { if c.Observer != nil { if c.Contracts != c.Observer.chain.Contracts { return SenderInfo{}, errors.New("payment observer belongs to another contract client") @@ -199,9 +212,13 @@ func (c PaymentChain) SenderInfo(ctx context.Context, sender, recipient ethcommo } func (c PaymentChain) senderInfoAt(ctx context.Context, sender, recipient ethcommon.Address, snapshot ChainSnapshot) (SenderInfo, error) { - address, err := c.Contracts.paymentAddress(ctx, snapshot, "ticketBroker") - if err != nil { - return SenderInfo{}, err + address := snapshot.addresses.broker + if snapshot.observer == nil { + var err error + address, err = c.Contracts.ResolveAt(ctx, snapshot.blockReference, "ticketBroker") + if err != nil { + return SenderInfo{}, err + } } broker, err := contracts.NewTicketBrokerCaller(address, c.Contracts.callerAt(snapshot.blockReference)) if err != nil { diff --git a/eth/payment_contracts.go b/eth/payment_contracts.go deleted file mode 100644 index 62ab8a4..0000000 --- a/eth/payment_contracts.go +++ /dev/null @@ -1,48 +0,0 @@ -package eth - -import ( - "context" - "errors" - - "github.com/ethereum/go-ethereum/common" -) - -// Address retention adapts livepeer/go-livepeer eth/client.go setContracts at -// 597f882cb9f4d42ad8430b0f592afd5929af4a6b (Nico Vergauwen, 829cd9047c46aef03ff5bf07b3e34be810694b74; -// Rafał Leszko, 3209dd946bb0376308044055b6caf262fd25cf61). Confirmation before -// dependent reads adapts Mike Zupper's BlueClawNetwork/livepeer-node -// eth/chainview.go, 2d95d40b8aa4699c7ca40838b6a9c312704c93db. -// Copyright (c) 2017 Livepeer; MIT notice in LICENSE. -type paymentContracts struct{ rounds, broker, bonding common.Address } - -func (c *Contracts) paymentContractsAt(ctx context.Context, ref any) (paymentContracts, error) { - var addresses paymentContracts - for _, entry := range []struct { - name string - to *common.Address - }{{"roundsManager", &addresses.rounds}, {"ticketBroker", &addresses.broker}, {"bondingManager", &addresses.bonding}} { - address, err := c.ResolveAt(ctx, ref, entry.name) - if err != nil { - return paymentContracts{}, err - } - *entry.to = address - } - return addresses, nil -} - -func (c *Contracts) paymentAddress(ctx context.Context, s ChainSnapshot, name string) (common.Address, error) { - if s.observer == nil { - return c.ResolveAt(ctx, s.blockReference, name) - } - if s.observer.chain.Contracts != c { - return common.Address{}, errors.New("payment snapshot belongs to another contract client") - } - switch name { - case "ticketBroker": - return s.addresses.broker, nil - case "bondingManager": - return s.addresses.bonding, nil - default: - return common.Address{}, errors.New("unsupported payment contract") - } -} diff --git a/eth/payment_contracts_test.go b/eth/payment_contracts_test.go index 607ecba..57c9704 100644 --- a/eth/payment_contracts_test.go +++ b/eth/payment_contracts_test.go @@ -57,7 +57,7 @@ func TestObserverConfirmedPaymentContracts(t *testing.T) { chain.Observer = o f.Reset() for range 3 { - funds, err := chain.SenderInfo(t.Context(), common.HexToAddress("0x2000"), common.HexToAddress("0x3000")) + funds, err := chain.PayerFunds(t.Context(), common.HexToAddress("0x2000"), common.HexToAddress("0x3000")) require.NoError(t, err) require.Equal(t, after, funds.Snapshot) active, err := chain.IsActiveAt(t.Context(), common.HexToAddress("0x3000"), funds.Snapshot) diff --git a/eth/payment_observer.go b/eth/payment_observer.go index 00036ba..807d78e 100644 --- a/eth/payment_observer.go +++ b/eth/payment_observer.go @@ -22,6 +22,7 @@ var ErrObservationUnavailable = errors.New("chain observation unavailable") // PaymentObserver publishes complete canonical observations. Gaps and reorgs // rebuild state at the new hash instead of replaying uncertain event history. type PaymentObserver struct { + OnRefresh func(time.Duration, error) // Install before Run; called serially. chain PaymentChain interval, maxAge time.Duration now func() time.Time @@ -30,7 +31,6 @@ type PaymentObserver struct { header *miniHeader generation uint64 requireL1 bool - invalid, stopped bool contact, progress time.Time l1Progress time.Time l1HighWater *big.Int @@ -48,23 +48,27 @@ func NewPaymentObserver(chain PaymentChain, requireL1 bool, recipient common.Add // Run has one service-owned caller. Every RPC inherits the service lifetime. func (o *PaymentObserver) Run(ctx context.Context) { + ctx = WithRPCWork(ctx, "background") o.mu.Lock() lifecycle, cancelReads := context.WithCancel(ctx) o.lifecycle = lifecycle o.mu.Unlock() defer func() { - cancelReads() o.mu.Lock() - o.stopped = true + cancelReads() o.mu.Unlock() o.payerReads.Wait() }() ticker := time.NewTicker(o.interval) defer ticker.Stop() for { + start := time.Now() read, cancel := context.WithTimeout(ctx, o.maxAge) - _ = o.refresh(read) + err := o.refresh(read) cancel() + if o.OnRefresh != nil { + o.OnRefresh(time.Since(start), err) + } select { case <-ctx.Done(): return @@ -78,6 +82,7 @@ func (o *PaymentObserver) Run(ctx context.Context) { type ObserverStatus struct { ObservedAt, ContactAt, HeadProgressAt, L1ProgressAt time.Time Healthy, RecipientActive bool + RecipientTracked, RecipientFailed bool PayerEntries, PayerReads, PayerFailures int } @@ -85,9 +90,12 @@ func (o *PaymentObserver) Status() ObserverStatus { o.mu.RLock() defer o.mu.RUnlock() healthy := o.validate(o.current) == nil - status := ObserverStatus{ObservedAt: o.current.observedAt, ContactAt: o.contact, HeadProgressAt: o.progress, L1ProgressAt: o.l1Progress, Healthy: healthy, RecipientActive: healthy && o.recipient != (common.Address{}) && o.current.recipientErr == nil && o.current.recipientActive, PayerEntries: len(o.payers), PayerReads: o.activeReads} + status := ObserverStatus{ + ObservedAt: o.current.observedAt, ContactAt: o.contact, HeadProgressAt: o.progress, L1ProgressAt: o.l1Progress, + RecipientActive: healthy && o.recipient != (common.Address{}) && o.current.recipientErr == nil && o.current.recipientActive, PayerEntries: len(o.payers), PayerReads: o.activeReads, Healthy: healthy, RecipientTracked: o.recipient != (common.Address{}), RecipientFailed: o.current.recipientErr != nil, + } for _, entry := range o.payers { - if !entry.valid && entry.failure != nil { + if !entry.valid && entry.failure != nil && entry.failedAt == o.current.BlockHash && entry.failedGeneration == o.current.generation { status.PayerFailures++ } } @@ -132,7 +140,7 @@ func (s ChainSnapshot) Validate() error { } func (o *PaymentObserver) validate(s ChainSnapshot) error { - if o.stopped || o.invalid || (o.lifecycle != nil && o.lifecycle.Err() != nil) || s.generation != o.generation || o.expired(s, o.l1Progress) { + if (o.lifecycle != nil && o.lifecycle.Err() != nil) || s.generation != o.generation || o.expired(s, o.l1Progress) { return ErrObservationUnavailable } return nil @@ -161,7 +169,6 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { if prior != nil { distance := new(big.Int).Sub((*big.Int)(head.Number), (*big.Int)(prior.Number)) if distance.Sign() < 0 { - o.invalid = true o.generation++ o.mu.Unlock() return ErrObservationUnavailable @@ -169,7 +176,6 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { gap = head.Hash != prior.Hash && distance.Cmp(big.NewInt(1)) > 0 continuous = distance.Cmp(big.NewInt(1)) == 0 && head.ParentHash == prior.Hash if head.Hash != prior.Hash && !gap && (distance.Sign() == 0 || head.ParentHash != prior.Hash) { - o.invalid = true o.generation++ } } @@ -187,7 +193,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { if prior != nil && head.Hash == prior.Hash { // A lagging backend may return N, N-1, then N again. Restore the // immutable observation without renewing any freshness deadline. - o.current.generation, o.invalid = o.generation, false + o.current.generation = o.generation err := o.validate(o.current) o.mu.Unlock() return err @@ -199,10 +205,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { anchor, err := o.chain.Contracts.RPC.header(ctx, hexutil.EncodeBig((*big.Int)(prior.Number))) continuous = err == nil && anchor.Hash == prior.Hash && anchor.Number != nil && (*big.Int)(anchor.Number).Cmp((*big.Int)(prior.Number)) == 0 if !continuous { - o.mu.Lock() - o.invalid = true - o.generation++ - o.mu.Unlock() + o.invalidate() } } s, err := o.chain.snapshotAt(ctx, head, true) @@ -243,7 +246,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { l1Progress, highWater = at, s.Block } s.observer, s.generation, s.observedAt, s.blockTime = o, o.generation, at, blockTime - if o.stopped || o.expired(s, l1Progress) { + if (o.lifecycle != nil && o.lifecycle.Err() != nil) || o.expired(s, l1Progress) { return ErrObservationUnavailable } for key, entry := range o.payers { @@ -252,7 +255,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { entry.valid, entry.failure = false, nil } } - o.current, o.header, o.invalid = s, head, false + o.current, o.header = s, head o.progress, o.l1Progress, o.l1HighWater = progress, l1Progress, highWater o.requireL1 = o.requireL1 || head.L1BlockNumber != nil return nil diff --git a/eth/payment_observer_test.go b/eth/payment_observer_test.go index 1b179d0..24d470f 100644 --- a/eth/payment_observer_test.go +++ b/eth/payment_observer_test.go @@ -3,7 +3,6 @@ package eth import ( "context" "math/big" - "sync" "testing" "time" @@ -130,8 +129,12 @@ func TestObserverInvalidHeaders(t *testing.T) { } func TestObserverRejectedHeadersInvalidateConflictingIdentity(t *testing.T) { - for _, metadata := range []string{"stale", "missing timestamp", "missing L1"} { - for _, mode := range []string{"replacement", "rollback", "parent", "gap", "same"} { + for metadata, change := range map[string]func(*test.ChainState, time.Time){ + "stale": func(s *test.ChainState, now time.Time) { s.Timestamp = uint64(now.Add(-31 * time.Second).Unix()) }, + "missing timestamp": func(s *test.ChainState, _ time.Time) { s.OmitTimestamp = true }, + "missing L1": func(s *test.ChainState, _ time.Time) { s.OmitL1 = true }, + } { + for mode, distance := range map[string]int64{"replacement": 0, "rollback": -1, "parent": 1, "gap": 10, "same": 0} { t.Run(metadata+"/"+mode, func(t *testing.T) { f, o, now := newTestObserver(t) require.NoError(t, o.refresh(t.Context())) @@ -141,22 +144,8 @@ func TestObserverRejectedHeadersInvalidateConflictingIdentity(t *testing.T) { if mode != "same" { s.Hash, s.Parent = common.HexToHash("0xdead"), common.HexToHash("0xbad") } - switch mode { - case "rollback": - s.Number-- - case "parent": - s.Number++ - case "gap": - s.Number += 10 - } - switch metadata { - case "stale": - s.Timestamp = uint64(now.Add(-31 * time.Second).Unix()) - case "missing timestamp": - s.OmitTimestamp = true - case "missing L1": - s.OmitL1 = true - } + s.Number = uint64(int64(s.Number) + distance) + change(s, *now) }) f.Reset() require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable) @@ -239,11 +228,10 @@ func TestObserverPinnedBackfillAndReorg(t *testing.T) { require.NoError(t, err) require.Equal(t, old.Round, s.Round, "failed candidates publish no partial round state") f.Update(func(s *test.ChainState) { s.Number += 10; s.Hash = common.HexToHash("0xbeef"); s.Fail = "" }) - var once sync.Once f.Update(func(s *test.ChainState) { s.Before = func(_ context.Context, name string) { if name == "RoundsManager.blockHashForRound" { - once.Do(func() { f.Update(func(s *test.ChainState) { s.Hash = common.HexToHash("0xbad") }) }) + f.Update(func(s *test.ChainState) { s.Hash = common.HexToHash("0xbad") }) } } }) @@ -293,10 +281,7 @@ func TestObserverGapKeepsCanonicalStateAvailable(t *testing.T) { for _, mode := range []string{"gap", "replacement", "parent"} { old, err := o.Snapshot(t.Context()) require.NoError(t, err) - entered, unblock := make(chan struct{}), make(chan struct{}) - notify := sync.OnceFunc(func() { close(entered) }) - release := sync.OnceFunc(func() { close(unblock) }) - t.Cleanup(release) + checked := false f.Update(func(s *test.ChainState) { if mode == "gap" { s.Number += 10 @@ -312,31 +297,23 @@ func TestObserverGapKeepsCanonicalStateAvailable(t *testing.T) { } s.Before = func(_ context.Context, name string) { if name == "RoundsManager.lastInitializedRound" { - notify() - <-unblock + checked = true + if mode != "gap" { + require.ErrorIs(t, old.Validate(), ErrObservationUnavailable) + require.False(t, o.Status().Healthy) + require.Equal(t, 1, f.Stats().Calls["eth_getBlockByNumber"], "proven reorgs must invalidate before any anchor probe") + } else { + require.NoError(t, old.Validate()) + require.True(t, o.Status().Healthy) + _, err := o.Snapshot(t.Context()) + require.NoError(t, err) + } } } }) f.Reset() - result := make(chan error, 1) - go func() { result <- o.refresh(t.Context()) }() - select { - case <-entered: - case <-time.After(time.Second): - t.Fatal("backfill did not start") - } - if mode != "gap" { - require.ErrorIs(t, old.Validate(), ErrObservationUnavailable) - require.False(t, o.Status().Healthy) - require.Equal(t, 1, f.Stats().Calls["eth_getBlockByNumber"], "proven reorgs must invalidate before any anchor probe") - } else { - require.NoError(t, old.Validate()) - require.True(t, o.Status().Healthy) - _, err = o.Snapshot(t.Context()) - require.NoError(t, err) - } - release() - require.NoError(t, <-result) + require.NoError(t, o.refresh(t.Context())) + require.True(t, checked, "availability must be checked before the dependent read completes") f.Update(func(s *test.ChainState) { s.Before = nil }) if mode == "replacement" { require.Equal(t, int64(4), o.current.Round.Int64(), "removed round state is backfilled") @@ -392,7 +369,7 @@ func TestObserverRecipientActivity(t *testing.T) { f.Update(func(s *test.ChainState) { s.Fail = "BondingManager.isActiveTranscoder" }) require.NoError(t, o.refresh(t.Context()), "a recipient failure must not reject unrelated chain state") require.True(t, o.Status().Healthy) - _, err = c.SenderInfo(t.Context(), common.HexToAddress("0x2000"), o.recipient) + _, err = c.PayerFunds(t.Context(), common.HexToAddress("0x2000"), o.recipient) require.NoError(t, err) _, err = c.IsActive(t.Context(), o.recipient) require.Error(t, err, "unknown must remain distinct from known inactive") diff --git a/eth/payment_test.go b/eth/payment_test.go index 8846872..14ce7c1 100644 --- a/eth/payment_test.go +++ b/eth/payment_test.go @@ -1,85 +1,34 @@ package eth import ( - "encoding/hex" + "context" "encoding/json" "math/big" - "net/http" - "net/http/httptest" "strings" - "sync/atomic" "testing" 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/livepeer/node/internal/test" "github.com/stretchr/testify/require" ) func TestCanonicalPaymentSnapshotAndCollateral(t *testing.T) { for _, l2 := range []bool{false, true} { t.Run(map[bool]string{false: "ethereum", true: "arbitrum"}[l2], func(t *testing.T) { - blockHash := ethcommon.HexToHash("0xabcd").Hex() sender, recipient := ethcommon.HexToAddress("0x2000"), ethcommon.HexToAddress("0x3000") - var reorg atomic.Bool - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - var request struct { - Method string - Params []json.RawMessage - } - require.NoError(t, json.NewDecoder(r.Body).Decode(&request)) - var result any - switch request.Method { - case "eth_getBlockByNumber": - header := map[string]string{"number": "0x100000", "hash": blockHash} - if l2 { - header["l1BlockNumber"] = "0x32" - } - result = header - case "eth_call": - var ref map[string]any - require.NoError(t, json.Unmarshal(request.Params[1], &ref)) - require.Equal(t, map[string]any{"blockHash": blockHash, "requireCanonical": true}, ref) - if reorg.Load() { - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32000}}) - return - } - var call map[string]string - require.NoError(t, json.Unmarshal(request.Params[0], &call)) - selector := func(sig string) bool { - return strings.HasPrefix(call["input"], "0x"+hex.EncodeToString(crypto.Keccak256([]byte(sig))[:4])) - } - switch { - case selector("getContract(bytes32)"): - result = "0x" + hex.EncodeToString(ethcommon.LeftPadBytes([]byte{1}, 32)) - case selector("lastInitializedRound()"): - result = "0x" + hex.EncodeToString(ethcommon.LeftPadBytes([]byte{5}, 32)) - case selector("blockHashForRound(uint256)"): - require.True(t, strings.HasSuffix(call["input"], strings.Repeat("0", 63)+"5")) - result = blockHash - case selector("isActiveTranscoder(address)"): - require.Equal(t, calldata("isActiveTranscoder(address)", addressWord(recipient)), call["input"]) - result = "0x" + word(1) - case selector("getSenderInfo(address)"): - require.Equal(t, calldata("getSenderInfo(address)", addressWord(sender)), call["input"]) - result = "0x" + word(10) + word(9) + word(50) + word(0) - case selector("claimableReserve(address,address)"): - require.Equal(t, calldata("claimableReserve(address,address)", addressWord(sender), addressWord(recipient)), call["input"]) - result = "0x" + word(7) - default: - t.Errorf("unexpected call %s", call["input"]) - } - default: - t.Errorf("unexpected RPC %s", request.Method) - } - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": result}) - })) - defer server.Close() - rpc, err := OpenRPC(server.URL) + fixture := test.NewChainRPC(t) + fixture.Update(func(s *test.ChainState) { + s.OmitL1 = !l2 + s.Payer, s.Recipient = sender, recipient + s.Deposit, s.Reserve, s.Claimable, s.WithdrawRound = 10, 50, 7, 9 + }) + blockHash := fixture.State().RoundHash.Hex() + rpc, err := OpenRPC(fixture.URL) require.NoError(t, err) defer rpc.Close() - contracts, err := OpenContracts(rpc, ethcommon.HexToAddress("0x1000").Hex()) + contracts, err := NewContracts(rpc, fixture.Controller) require.NoError(t, err) chain := PaymentChain{Contracts: contracts} snapshot, err := chain.Snapshot(t.Context()) @@ -94,7 +43,7 @@ func TestCanonicalPaymentSnapshotAndCollateral(t *testing.T) { active, err := chain.IsActive(t.Context(), recipient) require.NoError(t, err) require.True(t, active) - funds, err := chain.SenderInfo(t.Context(), sender, recipient) + funds, err := chain.PayerFunds(t.Context(), sender, recipient) require.NoError(t, err) require.Equal(t, snapshot.Block, funds.Snapshot.Block) require.Equal(t, snapshot.Round, funds.Snapshot.Round) @@ -103,10 +52,16 @@ func TestCanonicalPaymentSnapshotAndCollateral(t *testing.T) { require.Equal(t, big.NewInt(10), funds.Deposit) require.Equal(t, big.NewInt(9), funds.WithdrawRound) require.Equal(t, big.NewInt(7), funds.Reserve, "acceptance uses this recipient's claimable reserve") - readiness, err := chain.SenderInfo(t.Context(), sender, ethcommon.Address{}) + readiness, err := chain.PayerFunds(t.Context(), sender, ethcommon.Address{}) require.NoError(t, err) require.Equal(t, big.NewInt(50), readiness.Reserve, "readiness uses total remaining reserve") - reorg.Store(true) + fixture.Update(func(s *test.ChainState) { + s.Before = func(_ context.Context, name string) { + if name == "Controller.getContract" { + fixture.Update(func(s *test.ChainState) { s.Hash = ethcommon.HexToHash("0xdead") }) + } + } + }) _, err = chain.Snapshot(t.Context()) require.Error(t, err, "a reorg must not produce mixed observations") }) @@ -117,56 +72,44 @@ func TestRedemptionReceiptRequiresCanonicalFinality(t *testing.T) { for _, scenario := range []string{"pending", "unfinalized", "reorg", "confirmed", "reverted", "invalid-missing-status", "invalid-null-status", "invalid-status", "invalid-hash"} { t.Run(scenario, func(t *testing.T) { hash := ethcommon.HexToHash("0xabcd").Hex() - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - var request struct { - Method string - Params []json.RawMessage - } - require.NoError(t, json.NewDecoder(r.Body).Decode(&request)) - var result any - if request.Method == "eth_getTransactionReceipt" { - if scenario != "pending" { - status := uint64(1) - if scenario == "reverted" { - status = 0 - } - fields := map[string]any{"transactionHash": ethcommon.HexToHash("0x1234"), "status": hexutil.EncodeUint64(status), "blockNumber": "0x10", "blockHash": hash, - "cumulativeGasUsed": "0x0", "gasUsed": "0x0", "logsBloom": types.Bloom{}, "logs": []any{}} - switch scenario { - case "invalid-missing-status": - delete(fields, "status") - case "invalid-null-status": - fields["status"] = nil - case "invalid-status": - fields["status"] = "0x2" - case "invalid-hash": - fields["transactionHash"] = hash - } - result = fields + rpc := testRPC(t, func(method string, params []json.RawMessage) any { + if method == "eth_getTransactionReceipt" { + if scenario == "pending" { + return nil + } + status := uint64(1) + if scenario == "reverted" { + status = 0 } - } else { - var block string - require.NoError(t, json.Unmarshal(request.Params[0], &block)) - if block == "finalized" { - number := "0x10" - if scenario == "unfinalized" { - number = "0xf" - } - result = map[string]string{"number": number} - } else { - canonical := hash - if scenario == "reorg" { - canonical = ethcommon.HexToHash("0xffff").Hex() - } - result = map[string]string{"hash": canonical} + fields := map[string]any{"transactionHash": ethcommon.HexToHash("0x1234"), "status": hexutil.EncodeUint64(status), "blockNumber": "0x10", "blockHash": hash, + "cumulativeGasUsed": "0x0", "gasUsed": "0x0", "logsBloom": types.Bloom{}, "logs": []any{}} + switch scenario { + case "invalid-missing-status": + delete(fields, "status") + case "invalid-null-status": + fields["status"] = nil + case "invalid-status": + fields["status"] = "0x2" + case "invalid-hash": + fields["transactionHash"] = hash } + return fields } - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": result}) - })) - defer server.Close() - rpc, err := OpenRPC(server.URL) - require.NoError(t, err) - defer rpc.Close() + var block string + require.NoError(t, json.Unmarshal(params[0], &block)) + if block == "finalized" { + number := "0x10" + if scenario == "unfinalized" { + number = "0xf" + } + return map[string]string{"number": number} + } + canonical := hash + if scenario == "reorg" { + canonical = ethcommon.HexToHash("0xffff").Hex() + } + return map[string]string{"hash": canonical} + }) contracts := &Contracts{RPC: rpc} receipt, err := (PaymentChain{Contracts: contracts}).FinalizedReceipt(t.Context(), ethcommon.HexToHash("0x1234")) if strings.HasPrefix(scenario, "invalid") { diff --git a/eth/pricefeed_test.go b/eth/pricefeed_test.go index 977a8bc..8648593 100644 --- a/eth/pricefeed_test.go +++ b/eth/pricefeed_test.go @@ -4,8 +4,6 @@ import ( "encoding/hex" "encoding/json" "math/big" - "net/http" - "net/http/httptest" "strings" "testing" "time" @@ -20,57 +18,45 @@ func TestPriceFeedConversionAndRejection(t *testing.T) { t.Run(scenario, func(t *testing.T) { contract, err := chainlink.AggregatorV3InterfaceMetaData.GetAbi() require.NoError(t, err) - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - var req struct { - Method string - Params []json.RawMessage + rpc := testRPC(t, func(rpcMethod string, params []json.RawMessage) any { + if rpcMethod == "eth_getBlockByNumber" { + return map[string]string{"hash": ethcommon.HexToHash("0x1234").Hex()} } - require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) - var result any - if req.Method == "eth_getBlockByNumber" { - result = map[string]string{"hash": ethcommon.HexToHash("0x1234").Hex()} - } else { - var call map[string]string - require.NoError(t, json.Unmarshal(req.Params[0], &call)) - data, err := hex.DecodeString(strings.TrimPrefix(call["input"], "0x")) - require.NoError(t, err) - method, err := contract.MethodById(data[:4]) - require.NoError(t, err) - var args []any - switch method.Name { - case "description": - pair := "ETH / USD" - if scenario == "wrong-pair" { - pair = "BTC / USD" - } - args = []any{pair} - case "decimals": - args = []any{uint8(8)} - case "latestRoundData": - answer, updated, answered := big.NewInt(2000_00000000), time.Now().Add(-time.Minute).Unix(), big.NewInt(10) - switch scenario { - case "stale": - updated = time.Now().Add(-3 * time.Hour).Unix() - case "future": - updated = time.Now().Add(time.Hour).Unix() - case "negative": - answer.Neg(answer) - case "incomplete": - answered = big.NewInt(9) - } - args = []any{big.NewInt(10), answer, big.NewInt(updated), big.NewInt(updated), answered} + var call map[string]string + require.NoError(t, json.Unmarshal(params[0], &call)) + data, err := hex.DecodeString(strings.TrimPrefix(call["input"], "0x")) + require.NoError(t, err) + method, err := contract.MethodById(data[:4]) + require.NoError(t, err) + var args []any + switch method.Name { + case "description": + pair := "ETH / USD" + if scenario == "wrong-pair" { + pair = "BTC / USD" } - packed, err := method.Outputs.Pack(args...) - require.NoError(t, err) - result = "0x" + hex.EncodeToString(packed) + args = []any{pair} + case "decimals": + args = []any{uint8(8)} + case "latestRoundData": + answer, updated, answered := big.NewInt(2000_00000000), time.Now().Add(-time.Minute).Unix(), big.NewInt(10) + switch scenario { + case "stale": + updated = time.Now().Add(-3 * time.Hour).Unix() + case "future": + updated = time.Now().Add(time.Hour).Unix() + case "negative": + answer.Neg(answer) + case "incomplete": + answered = big.NewInt(9) + } + args = []any{big.NewInt(10), answer, big.NewInt(updated), big.NewInt(updated), answered} } - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": result}) - })) - defer server.Close() - rpc, err := OpenRPC(server.URL) - require.NoError(t, err) - contracts, err := OpenContracts(rpc, ethcommon.HexToAddress("0x1000").Hex()) - require.NoError(t, err) + packed, err := method.Outputs.Pack(args...) + require.NoError(t, err) + return "0x" + hex.EncodeToString(packed) + }) + contracts := &Contracts{RPC: rpc} rate, until, err := contracts.WeiPerUSD(t.Context(), ethcommon.HexToAddress("0x2000"), 2*time.Hour) if scenario != "valid" { require.Error(t, err) diff --git a/eth/rpc.go b/eth/rpc.go index 6c79499..9a9e5ac 100644 --- a/eth/rpc.go +++ b/eth/rpc.go @@ -11,6 +11,7 @@ import ( "net/http" "net/url" "strings" + "sync" "time" "github.com/ethereum/go-ethereum" @@ -28,13 +29,30 @@ const maxRPCResponse = 1 << 20 type RPC struct { ethereum *ethclient.Client http *http.Client - observer func(string, time.Duration, error) + observer func(RPCObservation) requireL1 bool // Set by the startup chain-ID check, before concurrent use. } +// RPCObservation contains fixed labels and redacted errors, never RPC payloads. +type RPCObservation struct { + Method, Work, Contract, Function string + Elapsed time.Duration + Err error +} + +type rpcWorkKey struct{} + +// WithRPCWork distinguishes request-induced work from scheduled chain work. +func WithRPCWork(ctx context.Context, work string) context.Context { + if work != "foreground" && work != "background" { + work = "other" + } + return context.WithValue(ctx, rpcWorkKey{}, work) +} + // SetObserver installs a hook before the client is used. Observations contain // allowlisted methods and redacted errors, never RPC payloads or URLs. -func (r *RPC) SetObserver(observer func(string, time.Duration, error)) { r.observer = observer } +func (r *RPC) SetObserver(observer func(RPCObservation)) { r.observer = observer } func OpenRPC(endpoint string) (*RPC, error) { u, err := destination.ParseURL(endpoint) @@ -82,16 +100,24 @@ func (t boundedRPCTransport) RoundTrip(request *http.Request) (response *http.Re observer := t.rpc.observer var observed error if observer != nil { - start, method := time.Now(), "unknown" + start := time.Now() + o := RPCObservation{Method: "unknown", Work: "other"} + if work, ok := request.Context().Value(rpcWorkKey{}).(string); ok { + o.Work = work + } if request.GetBody != nil { if body, err := request.GetBody(); err == nil { var envelope struct { - Method string `json:"method"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` } if json.NewDecoder(body).Decode(&envelope) == nil { switch envelope.Method { - case "eth_chainId", "eth_call", "eth_getCode", "eth_blockNumber", "eth_getBalance", "eth_getTransactionCount", "eth_getTransactionReceipt", "eth_getTransactionByHash", "eth_getBlockByNumber", "eth_estimateGas", "eth_sendRawTransaction", "eth_maxPriorityFeePerGas", "eth_gasPrice": - method = envelope.Method + case "eth_chainId", "eth_call", "eth_getCode", "eth_getLogs", "eth_blockNumber", "eth_getBalance", "eth_getTransactionCount", "eth_getTransactionReceipt", "eth_getTransactionByHash", "eth_getBlockByNumber", "eth_estimateGas", "eth_sendRawTransaction", "eth_maxPriorityFeePerGas", "eth_gasPrice": + o.Method = envelope.Method + } + if o.Method == "eth_call" { + o.Contract, o.Function = contractCallLabel(envelope.Params) } } _ = body.Close() @@ -101,7 +127,8 @@ func (t boundedRPCTransport) RoundTrip(request *http.Request) (response *http.Re if err != nil { observed = err } - observer(method, time.Since(start), safeRPCError(observed)) + o.Elapsed, o.Err = time.Since(start), safeRPCError(observed) + observer(o) }() } response, err = t.RoundTripper.RoundTrip(request) @@ -130,6 +157,43 @@ func (t boundedRPCTransport) RoundTrip(request *http.Request) (response *http.Re return response, nil } +// Adapted from BlueClawNetwork/livepeer-node eth/rpc.go, commit +// 2c2ce3cf8a51f5464ce4961bbb6c2e640f348ae0, by Mike Zupper (MIT; see LICENSE). +var contractCallLabels = sync.OnceValue(func() map[[4]byte][2]string { + labels := map[[4]byte][2]string{} + for binding, functions := range map[string][]string{ + "controller": {"getContract"}, "roundsManager": {"lastInitializedRound", "currentRound", "blockHashForRound"}, + "bondingManager": {"isActiveTranscoder"}, "ticketBroker": {"getSenderInfo", "claimableReserve", "usedTickets", "redeemWinningTicket"}, + "priceFeed": {"description", "decimals", "latestRoundData"}, + } { + parsed, err := bindingMetadata[binding].GetAbi() + if err != nil { + continue + } + for _, function := range functions { + if method, ok := parsed.Methods[function]; ok { + labels[[4]byte(method.ID)] = [2]string{binding, function} + } + } + } + return labels +}) + +func contractCallLabel(params []json.RawMessage) (string, string) { + var call struct{ Input, Data hexutil.Bytes } + if len(params) > 0 && json.Unmarshal(params[0], &call) == nil { + if len(call.Input) == 0 { + call.Input = call.Data + } + if len(call.Input) >= 4 { + if label, ok := contractCallLabels()[[4]byte(call.Input[:4])]; ok { + return label[0], label[1] + } + } + } + return "unknown", "unknown" +} + func (t boundedRPCTransport) CloseIdleConnections() { if closer, ok := t.RoundTripper.(interface{ CloseIdleConnections() }); ok { closer.CloseIdleConnections() diff --git a/eth/rpc_test.go b/eth/rpc_test.go index be58e16..b1b0722 100644 --- a/eth/rpc_test.go +++ b/eth/rpc_test.go @@ -48,12 +48,12 @@ func TestRPCTransportPolicy(t *testing.T) { require.NoError(t, err) defer rpc.Close() observed := 0 - rpc.SetObserver(func(method string, elapsed time.Duration, observedErr error) { + rpc.SetObserver(func(o RPCObservation) { observed++ - require.Equal(t, "eth_chainId", method) - require.GreaterOrEqual(t, elapsed, time.Duration(0)) - require.Error(t, observedErr) - require.NotContains(t, observedErr.Error(), "secret-") + require.Equal(t, "eth_chainId", o.Method) + require.GreaterOrEqual(t, o.Elapsed, time.Duration(0)) + require.Error(t, o.Err) + require.NotContains(t, o.Err.Error(), "secret-") }) _, err = rpc.ChainID(t.Context()) require.Equal(t, 1, observed) diff --git a/eth/senderwatcher.go b/eth/senderwatcher.go index bc329aa..0f8e7a8 100644 --- a/eth/senderwatcher.go +++ b/eth/senderwatcher.go @@ -99,7 +99,8 @@ func (o *PaymentObserver) senderInfo(ctx context.Context, payer, recipient commo o.mu.Unlock() return SenderInfo{}, ErrObservationUnavailable } - read, cancel := context.WithTimeout(o.lifecycle, 15*time.Second) + work, _ := ctx.Value(rpcWorkKey{}).(string) + read, cancel := context.WithTimeout(WithRPCWork(o.lifecycle, work), 15*time.Second) job = &payerRead{done: make(chan struct{}), cancel: cancel} entry.job = job o.activeReads++ @@ -164,7 +165,6 @@ func (o *PaymentObserver) readPayer(ctx context.Context, key payerKey, entry *pa func (o *PaymentObserver) invalidate() { o.mu.Lock() - o.invalid = true o.generation++ o.mu.Unlock() } diff --git a/eth/senderwatcher_test.go b/eth/senderwatcher_test.go index bfd7872..978bb2c 100644 --- a/eth/senderwatcher_test.go +++ b/eth/senderwatcher_test.go @@ -70,7 +70,7 @@ func TestPayerCacheWarmCopiesAndEntryFailure(t *testing.T) { info.Snapshot.Number.SetInt64(0) before := f.Stats() for range 20 { - info, err := c.SenderInfo(t.Context(), payer, recipient) + info, err := c.PayerFunds(t.Context(), payer, recipient) require.NoError(t, err) require.Positive(t, info.Deposit.Sign()) require.Positive(t, info.Snapshot.Number.Sign()) @@ -80,29 +80,36 @@ func TestPayerCacheWarmCopiesAndEntryFailure(t *testing.T) { } require.Equal(t, before, f.Stats()) foreign := PaymentChain{Contracts: &Contracts{}, Observer: o} - _, err := foreign.SenderInfo(t.Context(), payer, recipient) + _, err := foreign.PayerFunds(t.Context(), payer, recipient) require.ErrorContains(t, err, "another contract client") require.Equal(t, before, f.Stats()) - total, err := c.SenderInfo(t.Context(), payer, common.Address{}) + total, err := c.PayerFunds(t.Context(), payer, common.Address{}) require.NoError(t, err) require.Equal(t, int64(50), total.Reserve.Int64()) bad := common.HexToAddress("0xbad") f.Update(func(s *test.ChainState) { s.FailPayer = bad }) - _, err = c.SenderInfo(t.Context(), bad, recipient) + _, err = c.PayerFunds(t.Context(), bad, recipient) require.Error(t, err) + require.Equal(t, 1, o.Status().PayerFailures) before = f.Stats() - _, err = c.SenderInfo(t.Context(), bad, recipient) + _, err = c.PayerFunds(t.Context(), bad, recipient) require.Error(t, err) - _, err = c.SenderInfo(t.Context(), payer, recipient) + _, err = c.PayerFunds(t.Context(), payer, recipient) require.NoError(t, err) require.Equal(t, before, f.Stats(), "one failing entry cannot cause repeated or unrelated reads") *now = now.Add(o.maxAge) - _, err = c.SenderInfo(t.Context(), payer, recipient) + _, err = c.PayerFunds(t.Context(), payer, recipient) require.ErrorIs(t, err, ErrObservationUnavailable) + advanceObserver(f, *now) + require.NoError(t, o.refresh(t.Context())) + require.Zero(t, o.Status().PayerFailures, "a new head makes the old failure retryable") + _, err = c.PayerFunds(t.Context(), bad, recipient) + require.Error(t, err) + require.Equal(t, 1, o.Status().PayerFailures, "a failed retry belongs to the new head") f.Update(func(s *test.ChainState) { s.FailPayer = common.Address{} }) advanceObserver(f, *now) require.NoError(t, o.refresh(t.Context())) - _, err = c.SenderInfo(t.Context(), bad, recipient) + _, err = c.PayerFunds(t.Context(), bad, recipient) require.NoError(t, err) } @@ -291,6 +298,7 @@ func TestPayerColdReadRecoversAtSameHead(t *testing.T) { if timing == "rpc-after" { release() } + require.Zero(t, o.Status().PayerFailures, "a restored head must not count an old-generation failure") _, err := o.senderInfo(t.Context(), payer, common.Address{}) require.NoError(t, err, "a failure from an old epoch must not poison a restored-head payer") }) @@ -434,12 +442,21 @@ func TestPayerReorgDuringLogsRejectsCandidate(t *testing.T) { }) client := o.chain.Contracts.RPC.http next := client.Transport - client.Transport = reorgLogTransport{next, func() { - f.Update(func(s *test.ChainState) { - s.Hash, s.Deposit, s.Fail = common.HexToHash("0x201"), 0, "" - s.Logs = []types.Log{{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["Withdrawal"].ID, common.BytesToHash(payer.Bytes())}}} - }) - }} + client.Transport = payerHistoryRoute(func(r *http.Request) (*http.Response, error) { + request, err := readPayerHistoryCall(r) + if err != nil { + return nil, err + } + response, err := next.RoundTrip(r) + if request.Method == "eth_getLogs" { + // Hash filters may still return retained orphan logs after a reorg. + f.Update(func(s *test.ChainState) { + s.Hash, s.Deposit, s.Fail = common.HexToHash("0x201"), 0, "" + s.Logs = []types.Log{{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["Withdrawal"].ID, common.BytesToHash(payer.Bytes())}}} + }) + } + return response, err + }) require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable) require.Equal(t, old.BlockHash, o.current.BlockHash, "orphaned candidates publish no state") require.ErrorIs(t, old.Validate(), ErrObservationUnavailable) @@ -459,24 +476,6 @@ func TestPayerReorgDuringLogsRejectsCandidate(t *testing.T) { } } -type reorgLogTransport struct { - http.RoundTripper - reorg func() -} - -func (tr reorgLogTransport) RoundTrip(r *http.Request) (*http.Response, error) { - request, err := readPayerHistoryCall(r) - if err != nil { - return nil, err - } - response, err := tr.RoundTripper.RoundTrip(r) - if request.Method == "eth_getLogs" { - // Hash filters may still return retained orphan logs after a reorg. - tr.reorg() - } - return response, err -} - type payerHistoryRoute func(*http.Request) (*http.Response, error) func (fn payerHistoryRoute) RoundTrip(r *http.Request) (*http.Response, error) { return fn(r) } @@ -655,34 +654,20 @@ func TestPayerHistoryCostAndFreshness(t *testing.T) { } func TestPayerMalformedBloom(t *testing.T) { - for _, intermediate := range []bool{false, true} { - t.Run(fmt.Sprint(intermediate), func(t *testing.T) { - f, o, now := newTestObserver(t) - payer := common.HexToAddress("0x11") - held := warmPayer(t, o, payer, common.Address{}) - advanceObserver(f, *now) - f.Update(func(s *test.ChainState) { - payerHistoryBloom(s, "malformed") - s.Deposit = 0 - }) - if intermediate { - advanceObserver(f, *now) - f.Update(func(s *test.ChainState) { payerHistoryBloom(s, "empty") }) - require.NoError(t, o.refresh(t.Context())) - require.ErrorIs(t, held.Snapshot.Validate(), ErrObservationUnavailable) - funds, err := o.senderInfo(t.Context(), payer, common.Address{}) - require.NoError(t, err) - require.Zero(t, funds.Deposit.Sign(), "malformed history forces exact funds recovery") - } else { - require.Error(t, o.refresh(t.Context())) - require.Equal(t, held.Snapshot.BlockHash, o.current.BlockHash, "malformed candidate cannot publish") - } - }) - } + f, o, now := newTestObserver(t) + payer := common.HexToAddress("0x11") + held := warmPayer(t, o, payer, common.Address{}) + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { + payerHistoryBloom(s, "malformed") + s.Deposit = 0 + }) + require.Error(t, o.refresh(t.Context())) + require.Equal(t, held.Snapshot.BlockHash, o.current.BlockHash, "malformed candidate cannot publish") } func TestPayerIncompleteHistoryResets(t *testing.T) { - for _, mode := range []string{"parent-height", "parent-hash", "log-height", "log-hash"} { + for _, mode := range []string{"parent-height", "parent-hash", "parent-bloom", "log-height", "log-hash"} { t.Run(mode, func(t *testing.T) { f, o, now := newTestObserver(t) payer := common.HexToAddress("0x11") @@ -691,18 +676,23 @@ func TestPayerIncompleteHistoryResets(t *testing.T) { advanceObserver(f, *now) parent := f.State() advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { s.Deposit = 0 }) head := f.State() method, ref := "eth_getLogs", "" var result any - if mode == "parent-height" || mode == "parent-hash" { + if mode == "parent-height" || mode == "parent-hash" || mode == "parent-bloom" { method, ref = "eth_getBlockByNumber", hexutil.EncodeUint64(parent.Number) height, hash := parent.Number, parent.Hash - if mode == "parent-height" { + bloom := parent.LogsBloom + switch mode { + case "parent-height": height++ - } else { + case "parent-hash": hash = anchor.Hash + case "parent-bloom": + bloom = "0x01" } - result = map[string]any{"number": hexutil.EncodeUint64(height), "hash": hash, "parentHash": parent.Parent} + result = map[string]any{"number": hexutil.EncodeUint64(height), "hash": hash, "parentHash": parent.Parent, "logsBloom": bloom} } else { log := types.Log{Address: head.Contracts["TicketBroker"], BlockHash: head.Hash, BlockNumber: head.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["DepositFunded"].ID, common.HexToHash("0x12")}} if mode == "log-height" { @@ -712,8 +702,6 @@ func TestPayerIncompleteHistoryResets(t *testing.T) { } result = []types.Log{log} } - payload, err := json.Marshal(result) - require.NoError(t, err) client := o.chain.Contracts.RPC.http next := client.Transport client.Transport = payerHistoryRoute(func(r *http.Request) (*http.Response, error) { @@ -733,7 +721,7 @@ func TestPayerIncompleteHistoryResets(t *testing.T) { return next.RoundTrip(r) } } - data, err := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": json.RawMessage(payload)}) + data, err := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}) if err != nil { return nil, err } @@ -742,8 +730,10 @@ func TestPayerIncompleteHistoryResets(t *testing.T) { require.NoError(t, o.refresh(t.Context())) require.ErrorIs(t, held.Snapshot.Validate(), ErrObservationUnavailable) before := f.Stats().Calls["TicketBroker.getSenderInfo"] - _, err = o.senderInfo(t.Context(), payer, common.Address{}) + funds, err := o.senderInfo(t.Context(), payer, common.Address{}) require.NoError(t, err) + require.Zero(t, funds.Deposit.Sign(), "malformed history forces exact funds recovery") + require.Equal(t, head.Hash, funds.Snapshot.BlockHash) require.Equal(t, before+1, f.Stats().Calls["TicketBroker.getSenderInfo"], "incomplete proof must reset even an unaffected payer") }) } diff --git a/eth/transactionmanager.go b/eth/transactionmanager.go index d603db4..0efa449 100644 --- a/eth/transactionmanager.go +++ b/eth/transactionmanager.go @@ -87,7 +87,7 @@ type transactionReceipt struct { } func NewTransactionManager(eth transactionSenderReader, gpm feeCeiling, signer transactionSigner, txTimeout time.Duration, maxReplacements int) *TransactionManager { - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(WithRPCWork(context.Background(), "background")) return &TransactionManager{ ctx: ctx, cancel: cancel, cond: sync.NewCond(&sync.Mutex{}), diff --git a/eth/transactionmanager_test.go b/eth/transactionmanager_test.go index 19faff2..86deb89 100644 --- a/eth/transactionmanager_test.go +++ b/eth/transactionmanager_test.go @@ -29,6 +29,7 @@ type stubTransactionSenderReader struct { callsToTxByHash int lookupErrors []error minedAfter int + checkContext func(context.Context) } func (stm *stubTransactionSenderReader) SendTransaction(ctx context.Context, tx *types.Transaction) error { @@ -48,6 +49,9 @@ func (stm *stubTransactionSenderReader) TransactionByHash(ctx context.Context, t } func (stm *stubTransactionSenderReader) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) { + if stm.checkContext != nil { + stm.checkContext(ctx) + } if stm.receipt == nil { return nil, ethereum.NotFound } @@ -89,8 +93,9 @@ func TestTransactionManager_Wait(t *testing.T) { assert := assert.New(t) eth := &stubTransactionSenderReader{ - err: make(map[string]error), - pending: true, + err: make(map[string]error), + pending: true, + checkContext: func(ctx context.Context) { assert.Equal("background", ctx.Value(rpcWorkKey{})) }, } tm := NewTransactionManager(eth, nil, nil, 2*time.Second, 0) defer tm.Stop() diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 4752917..a7ab63f 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "io" + "log/slog" "math/big" "net" "net/http" @@ -22,6 +23,7 @@ import ( "github.com/felixge/httpsnoop" "github.com/livepeer/node/cmd/version" + "github.com/livepeer/node/eth" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/collectors" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -205,10 +207,64 @@ func ErrorKind(err error) string { return "error" } -func (m *Metrics) RPC(method string, elapsed time.Duration, err error) { - labels := Labels{"method": method} - m.Add("rpc_requests_total", 1, labels.With("outcome", ErrorKind(err))) - m.Observe("rpc_request_duration_seconds", elapsed.Seconds(), labels) +// Contract/refresh metrics adapt Mike Zupper's BlueClawNetwork/livepeer-node +// internal/observability/metrics.go at 2c2ce3cf8a51f5464ce4961bbb6c2e640f348ae0 (MIT; see LICENSE). +func (m *Metrics) RPC(o eth.RPCObservation) { + labels := Labels{"method": o.Method} + m.Add("rpc_requests_total", 1, labels.With("outcome", ErrorKind(o.Err))) + m.Observe("rpc_request_duration_seconds", o.Elapsed.Seconds(), labels) + m.Add("rpc_work_total", 1, Labels{"work": o.Work, "kind": "transport"}) + if o.Method == "eth_call" { + m.Add("rpc_contract_calls_total", 1, Labels{"contract": o.Contract, "function": o.Function}) + m.Add("rpc_work_total", 1, Labels{"work": o.Work, "kind": "contract_call"}) + } +} + +// PaymentObserver collects authoritative memory-only state. The callback only +// remembers the last logged health transition; it does not decide freshness. +func (m *Metrics) PaymentObserver(o *eth.PaymentObserver, logger *slog.Logger) { + last := "" + o.OnRefresh = func(elapsed time.Duration, err error) { + s := o.Status() + outcome := ErrorKind(err) + if err == nil && (s.RecipientFailed || s.PayerFailures > 0) { + outcome = "partial" + } + m.Add("chain_view_refreshes_total", 1, Labels{"outcome": outcome}) + m.Observe("chain_view_refresh_duration_seconds", elapsed.Seconds(), nil) + state := "unavailable" + if s.Healthy { + state = "healthy" + if err != nil || s.RecipientFailed || s.PayerFailures > 0 { + state = "degraded" + } + } + if state != last && logger != nil { + logger.Info("chain observation state changed", "event", "chain_view", "operation", "refresh", "outcome", state, "error_kind", outcome) + } + last = state + } + m.Snapshot(func(emit Emit) { + s := o.Status() + if !s.ObservedAt.IsZero() { + emit("chain_view_age_seconds", max(0, time.Since(s.ObservedAt).Seconds()), nil) + } + for name, at := range map[string]time.Time{ + "chain_view_last_contact_timestamp_seconds": s.ContactAt, + "chain_view_head_progress_timestamp_seconds": s.HeadProgressAt, + "chain_view_l1_progress_timestamp_seconds": s.L1ProgressAt, + } { + if !at.IsZero() { + emit(name, float64(at.Unix()), nil) + } + } + emit("chain_view_healthy", Bool(s.Healthy), nil) + emit("chain_view_tracked_entries", float64(s.PayerEntries), Labels{"kind": "payer"}) + emit("chain_view_failed_entries", float64(s.PayerFailures), Labels{"kind": "payer"}) + emit("chain_view_payer_reads_in_flight", float64(s.PayerReads), nil) + emit("chain_view_tracked_entries", Bool(s.RecipientTracked), Labels{"kind": "recipient"}) + emit("chain_view_failed_entries", Bool(s.RecipientFailed), Labels{"kind": "recipient"}) + }) } func (m *Metrics) Discovery(source int, elapsed time.Duration, err error) { @@ -390,7 +446,7 @@ func httpMethod(method string) string { // httpsnoop preserves the exact optional ResponseWriter interfaces. func (m *Metrics) HTTP(w http.ResponseWriter, r *http.Request, serve http.HandlerFunc) { s := &Request{metrics: m, Operation: "unknown", Protocol: "http", started: time.Now()} - r = r.WithContext(context.WithValue(r.Context(), requestKey{}, s)) + r = r.WithContext(eth.WithRPCWork(context.WithValue(r.Context(), requestKey{}, s), "foreground")) if r.Body != nil { r.Body = reader{r.Body, s} } diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go index 423949b..e51645d 100644 --- a/internal/observability/metrics_test.go +++ b/internal/observability/metrics_test.go @@ -1,19 +1,125 @@ package observability_test import ( + "bytes" "context" + "encoding/json" + "errors" "fmt" "io" + "log/slog" "net/http" "net/http/httptest" "strings" "testing" + "time" + "github.com/ethereum/go-ethereum/common" + "github.com/livepeer/node/eth" "github.com/livepeer/node/internal/observability" "github.com/livepeer/node/internal/test" "github.com/stretchr/testify/require" ) +func TestObserverMetricsAreLocalAndLogTransitions(t *testing.T) { + f := test.NewChainRPC(t) + rpc, err := eth.OpenRPC(f.URL) + require.NoError(t, err) + defer rpc.Close() + contracts, err := eth.NewContracts(rpc, f.Controller) + require.NoError(t, err) + o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true, common.HexToAddress("0x1234")) + m := observability.New("test") + rpc.SetObserver(m.RPC) + f.Update(func(s *test.ChainState) { s.Fail = "BondingManager.isActiveTranscoder" }) + var logs bytes.Buffer + m.PaymentObserver(o, slog.New(slog.NewTextHandler(&logs, nil))) + require.Empty(t, test.MetricSamples(t, m.Registry, "chain_view_age_seconds", nil)) + stopped := make(chan struct{}) + hook := o.OnRefresh + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { defer close(stopped); o.Run(ctx) }() + require.Eventually(t, func() bool { + return test.MetricSum(t, m.Registry, "chain_view_refresh_duration_seconds", nil) == 1 + }, 5*time.Second, time.Millisecond) + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "chain_view_healthy", nil)) + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "chain_view_refresh_duration_seconds", nil)) + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "chain_view_refreshes_total", map[string]string{"outcome": "partial"})) + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "chain_view_failed_entries", map[string]string{"kind": "recipient"})) + require.EqualValues(t, f.Stats().RoundTrips, test.MetricSum(t, m.Registry, "rpc_work_total", map[string]string{"work": "background", "kind": "transport"})) + f.Update(func(s *test.ChainState) { s.FailPayer = common.HexToAddress("0xbad") }) + _, err = (eth.PaymentChain{Contracts: contracts, Observer: o}).PayerFunds(eth.WithRPCWork(t.Context(), "foreground"), common.HexToAddress("0xbad"), common.Address{}) + require.Error(t, err) + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "chain_view_failed_entries", map[string]string{"kind": "payer"})) + before := f.Stats() + for range 3 { + require.NotEmpty(t, test.MetricSamples(t, m.Registry, "chain_view_age_seconds", nil)) + } + require.Equal(t, before, f.Stats(), "scraping must not perform chain reads") + cancel() + <-stopped + hook(time.Millisecond, errors.New("private-provider-detail")) + length := logs.Len() + hook(time.Millisecond, errors.New("private-provider-detail")) + require.Equal(t, length, logs.Len(), "unchanged failure must not log again") + require.NotContains(t, logs.String(), "private-") + require.EqualValues(t, 2, test.MetricSum(t, m.Registry, "chain_view_refreshes_total", map[string]string{"outcome": "error"})) +} + +func TestRPCAccountingPreservesAggregateLabels(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct{ ID json.RawMessage } + require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) + _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "error": map[string]any{"code": -32000, "message": "private-provider-text"}}) + })) + defer server.Close() + rpc, err := eth.OpenRPC(server.URL) + require.NoError(t, err) + defer rpc.Close() + m := observability.New("test") + rpc.SetObserver(m.RPC) + contracts, err := eth.NewContracts(rpc, common.Address{}) + require.NoError(t, err) + for _, tc := range []struct{ work, contract, function string }{ + {"foreground", "ticketBroker", "usedTickets"}, + {"background", "roundsManager", "currentRound"}, + {"unbounded-private-value", "roundsManager", "roundLength"}, + } { + ctx := eth.WithRPCWork(t.Context(), tc.work) + var args []any + if tc.function == "usedTickets" { + args = []any{common.Hash{}} + } + _, err := contracts.CallAt(ctx, "latest", tc.contract, common.Address{}, tc.function, args...) + require.Error(t, err) + } + require.EqualValues(t, 3, test.MetricSum(t, m.Registry, "rpc_requests_total", map[string]string{"method": "eth_call", "outcome": "error"})) + for _, work := range []string{"foreground", "background", "other"} { + for _, kind := range []string{"transport", "contract_call"} { + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "rpc_work_total", map[string]string{"work": work, "kind": kind})) + } + } + require.EqualValues(t, 1, test.MetricSum(t, m.Registry, "rpc_contract_calls_total", map[string]string{"contract": "unknown", "function": "unknown"})) + families, err := m.Registry.Gather() + require.NoError(t, err) + for _, family := range families { + for _, sample := range family.Metric { + var names []string + for _, label := range sample.Label { + names = append(names, label.GetName()) + } + switch family.GetName() { + case "livepeer_rpc_requests_total": + require.ElementsMatch(t, []string{"component", "method", "outcome"}, names) + case "livepeer_rpc_request_duration_seconds": + require.ElementsMatch(t, []string{"component", "method"}, names) + } + } + require.NotContains(t, family.String(), "private-") + } +} + // ResponseRecorder commits informational responses; a real server permits a // later final response. Preserve that behavior in this controlled writer. type provisionalRecorder struct{ *httptest.ResponseRecorder } diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go index d7abc3b..fe7cfbf 100644 --- a/internal/test/chainrpc.go +++ b/internal/test/chainrpc.go @@ -15,7 +15,6 @@ import ( "time" "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/stretchr/testify/require" @@ -39,7 +38,7 @@ type ChainState struct { ChainID uint64 Hash, Parent, RoundHash common.Hash Round, CurrentRound uint64 - Recipient common.Address + Payer, Recipient common.Address Activation, Deactivation uint64 OmitL1, OmitTimestamp bool Fail string @@ -83,6 +82,8 @@ func NewChainRPC(t testing.TB) *ChainRPC { "blockHashForRound(uint256)": "RoundsManager.blockHashForRound", "isActiveTranscoder(address)": "BondingManager.isActiveTranscoder", "getSenderInfo(address)": "TicketBroker.getSenderInfo", "claimableReserve(address,address)": "TicketBroker.claimableReserve", "usedTickets(bytes32)": "TicketBroker.usedTickets", + "description()": "PriceFeed.description", + "redeemWinningTicket((address,address,uint256,uint256,uint256,bytes32,bytes),bytes,uint256)": "TicketBroker.redeemWinningTicket", } { methods[hex.EncodeToString(crypto.Keccak256([]byte(signature))[:4])] = name } @@ -116,7 +117,7 @@ func NewChainRPC(t testing.TB) *ChainRPC { fail := func() { _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": request.ID, "error": map[string]any{"code": -32000}}) } - if state.Fail == name { + if state.Fail == name || state.Fail == request.Method { fail() return } @@ -124,40 +125,30 @@ func NewChainRPC(t testing.TB) *ChainRPC { switch request.Method { case "eth_getLogs": var query struct { - BlockHash common.Hash - FromBlock, ToBlock string - Address []common.Address + BlockHash common.Hash + Address []common.Address } require.NoError(t, json.Unmarshal(request.Params[0], &query)) - from, to := uint64(0), state.Number - if query.BlockHash == (common.Hash{}) { - var err error - from, err = hexutil.DecodeUint64(query.FromBlock) - require.NoError(t, err) - to, err = hexutil.DecodeUint64(query.ToBlock) - require.NoError(t, err) - } else if _, ok := c.stateAt(query.BlockHash.Hex()); !ok { + logs := []types.Log{} + c.mu.Lock() + frame, ok := c.history[query.BlockHash] + if !ok { + c.mu.Unlock() fail() return } - logs := []types.Log{} - c.mu.Lock() c.stats.LogAddresses = query.Address - for _, frame := range c.history { - for _, log := range frame.Logs { - matches := query.BlockHash == log.BlockHash - if query.BlockHash == (common.Hash{}) { - matches = log.BlockNumber >= from && log.BlockNumber <= to - } - if matches && slices.Contains(query.Address, log.Address) { - logs = append(logs, log) - } + for _, log := range frame.Logs { + if log.BlockHash == query.BlockHash && slices.Contains(query.Address, log.Address) { + logs = append(logs, log) } } c.mu.Unlock() result = logs case "eth_chainId": result = quantity(state.ChainID) + case "eth_getBalance": + result = "0x1" case "eth_getBlockByNumber": var ref string require.NoError(t, json.Unmarshal(request.Params[0], &ref)) @@ -180,14 +171,18 @@ func NewChainRPC(t testing.TB) *ChainRPC { } result = header case "eth_call": - var ref map[string]any - require.NoError(t, json.Unmarshal(request.Params[1], &ref)) - require.Equal(t, true, ref["requireCanonical"]) - var ok bool - state, ok = c.stateAt(ref["blockHash"].(string)) - if !ok { - fail() - return + directRead := (name == "Controller.getContract" || name == "TicketBroker.usedTickets") && string(request.Params[1]) == `"latest"` + simulation := name == "TicketBroker.redeemWinningTicket" && string(request.Params[1]) == `"pending"` + if !directRead && !simulation { + var ref map[string]any + require.NoError(t, json.Unmarshal(request.Params[1], &ref), "%s block reference %s", name, request.Params[1]) + require.Equal(t, true, ref["requireCanonical"]) + var ok bool + state, ok = c.stateAt(ref["blockHash"].(string)) + if !ok { + fail() + return + } } switch name { case "Controller.getContract": @@ -199,6 +194,7 @@ func NewChainRPC(t testing.TB) *ChainRPC { case "RoundsManager.currentRound": result = "0x" + word(quantity(state.CurrentRound)[2:]) case "RoundsManager.blockHashForRound": + require.Equal(t, word(quantity(state.Round)[2:]), call.Input[10:]) result = state.RoundHash.Hex() case "BondingManager.isActiveTranscoder": if state.Recipient != (common.Address{}) { @@ -209,15 +205,23 @@ func NewChainRPC(t testing.TB) *ChainRPC { result = "0x" + word("1") } case "TicketBroker.getSenderInfo": + if state.Payer != (common.Address{}) { + require.Equal(t, word(hex.EncodeToString(state.Payer.Bytes())), call.Input[10:]) + } if state.FailPayer != (common.Address{}) && common.HexToAddress(call.Input[10:74]) == state.FailPayer { fail() return } result = "0x" + word(quantity(state.Deposit)[2:]) + word(quantity(state.WithdrawRound)[2:]) + word(quantity(state.Reserve)[2:]) + word("0") case "TicketBroker.claimableReserve": + if state.Payer != (common.Address{}) { + require.Equal(t, word(hex.EncodeToString(state.Payer.Bytes()))+word(hex.EncodeToString(state.Recipient.Bytes())), call.Input[10:]) + } result = "0x" + word(quantity(state.Claimable)[2:]) case "TicketBroker.usedTickets": result = "0x" + word("0") + case "TicketBroker.redeemWinningTicket": + result = "0x" default: t.Errorf("unexpected payment contract call") } diff --git a/orchestrator/app.go b/orchestrator/app.go index e15b2d5..23a6f50 100644 --- a/orchestrator/app.go +++ b/orchestrator/app.go @@ -26,7 +26,6 @@ import ( "github.com/livepeer/node/internal/observability" "github.com/livepeer/node/nodeconfig" "github.com/livepeer/node/nodeconfig/migrations" - "github.com/livepeer/node/pm" "github.com/spf13/cobra" ) @@ -298,7 +297,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { recipient = *p.OrchestratorAddress } observedChain.Observer = eth.NewPaymentObserver(paymentChain, p.Network == nodeconfig.Mainnet, recipient) - engine, err = NewPaymentEngine(redeemerDB, pm.EthereumChain{Client: observedChain}, recipient, p.TicketFaceValue.ToBig(), p.TicketWinProb.ToBig()) + engine, err = NewPaymentEngine(redeemerDB, observedChain, recipient, p.TicketFaceValue.ToBig(), p.TicketWinProb.ToBig()) if err != nil { return fmt.Errorf("initialize payment engine (check --account, --ticket-face-value and --ticket-win-prob): %w", err) } @@ -311,6 +310,9 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { } app := NewServer(registry, runnerPolicy, proxyPolicy, logger) logger = app.logger + if engine != nil { + registry.metrics.PaymentObserver(observedChain.Observer, logger) + } defer app.Close() app.SetPayment(engine) readiness := &paidReadiness{interval: p.HeartbeatInterval} @@ -348,7 +350,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { ctx, cancel := signal.NotifyContext(parent, syscall.SIGINT, syscall.SIGTERM) defer cancel() if engine != nil { - if err := StartRedemptionRecovery(ctx, redeemerDB, paymentChain, redeemerKey, RedemptionOptions{Timeout: p.RedeemerTxTimeout, MaxReplacements: p.RedeemerMaxTxReplacements}); err != nil { + if err := StartRedemptionRecovery(eth.WithRPCWork(ctx, "background"), redeemerDB, paymentChain, redeemerKey, RedemptionOptions{Timeout: p.RedeemerTxTimeout, MaxReplacements: p.RedeemerMaxTxReplacements}); err != nil { return err } } @@ -370,7 +372,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { case <-ctx.Done(): return case <-ticker.C: - tickCtx, done := context.WithTimeout(ctx, timeout) + tickCtx, done := context.WithTimeout(eth.WithRPCWork(ctx, "background"), timeout) work(tickCtx) done() } diff --git a/orchestrator/app_test.go b/orchestrator/app_test.go index dda87bf..be07d1d 100644 --- a/orchestrator/app_test.go +++ b/orchestrator/app_test.go @@ -56,6 +56,7 @@ func freeTCPPort(t *testing.T) int { func TestPaidHTTPStartup(t *testing.T) { for _, separate := range []bool{false, true} { t.Run(fmt.Sprintf("separate-wallet=%t", separate), func(t *testing.T) { + client := &http.Client{Transport: &http.Transport{DisableKeepAlives: true}} mainPort, metricsPort := freeTCPPort(t), freeTCPPort(t) address := fmt.Sprintf("http://127.0.0.1:%d", mainPort) p := paymentKeystoreParams(t, "http://127.0.0.1:1") @@ -71,50 +72,23 @@ func TestPaidHTTPStartup(t *testing.T) { var requests atomic.Int32 dependencies := make(chan struct{}) releaseDependencies := sync.OnceFunc(func() { close(dependencies) }) - rpc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - var req struct { - Method string - ID json.RawMessage - Params []json.RawMessage - } - require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) - if req.Method != "eth_chainId" { - select { - case <-dependencies: - case <-r.Context().Done(): - return + rpc := test.NewChainRPC(t) + rpc.Update(func(s *test.ChainState) { + s.ChainID, s.Round, s.CurrentRound = 1, 1, 1 + s.Recipient, s.Fail = recipient, "eth_estimateGas" + s.Before = func(ctx context.Context, method string) { + if method != "eth_chainId" { + select { + case <-dependencies: + case <-ctx.Done(): + return + } } + requests.Add(1) } - requests.Add(1) - var result any = "0x1" - switch req.Method { - case "eth_chainId", "eth_getBalance", "eth_gasPrice", "eth_maxPriorityFeePerGas": - case "eth_getTransactionCount": - result = "0x0" - case "eth_estimateGas": - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "error": map[string]any{"code": -32000, "message": "ticket estimation unavailable"}}) - return - case "eth_getBlockByNumber": - result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0xabcd").Hex(), "timestamp": fmt.Sprintf("0x%x", time.Now().Unix())} - case "eth_call": - // Reuse the 32-byte encoding of 1 for calls returning an address, round, hash, or bool. - result = fmt.Sprintf("0x%064x", 1) - var call struct{ Input string } - require.NoError(t, json.Unmarshal(req.Params[0], &call)) - if strings.HasPrefix(call.Input, fmt.Sprintf("0x%x", crypto.Keccak256([]byte("isActiveTranscoder(address)"))[:4])) { - require.True(t, strings.HasSuffix(call.Input, fmt.Sprintf("%064x", recipient.Bytes())), "observe the configured ticket recipient, including a separate wallet") - } - if strings.HasPrefix(call.Input, fmt.Sprintf("0x%x", crypto.Keccak256([]byte("usedTickets(bytes32)"))[:4])) { - result = fmt.Sprintf("0x%064x", 0) - } - default: - t.Errorf("unexpected RPC %s", req.Method) - } - require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "result": result})) - })) - defer rpc.Close() + }) defer releaseDependencies() - p.RPCURL = mustURL(t, rpc.URL) + p.RPCURL, p.Controller = mustURL(t, rpc.URL), &rpc.Controller require.NoError(t, p.Listen.UnmarshalText(fmt.Appendf(nil, "unsafe@0.0.0.0:%d", mainPort))) require.NoError(t, p.MetricsListen.UnmarshalText(fmt.Appendf(nil, "unsafe@0.0.0.0:%d", metricsPort))) metricsURL := fmt.Sprintf("http://127.0.0.1:%d", metricsPort) @@ -134,7 +108,7 @@ func TestPaidHTTPStartup(t *testing.T) { } }) require.Eventually(t, func() bool { - response, err := http.Get(address + "/external/discovery") + response, err := client.Get(address + "/external/discovery") if err != nil { return false } @@ -143,7 +117,7 @@ func TestPaidHTTPStartup(t *testing.T) { }, 5*time.Second, 20*time.Millisecond) require.EqualValues(t, 1, requests.Load()) require.FileExists(t, p.RedeemerDB) - response, err := http.Get(metricsURL + "/readyz") + response, err := client.Get(metricsURL + "/readyz") require.NoError(t, err) response.Body.Close() require.Equal(t, http.StatusServiceUnavailable, response.StatusCode) @@ -156,7 +130,7 @@ func TestPaidHTTPStartup(t *testing.T) { require.NoError(t, store.StoreWinningTicket(&pm.SignedTicket{Ticket: &pm.Ticket{PayerAddress: recipient, Recipient: recipient, FaceValue: big.NewInt(1000), WinProb: big.NewInt(100), TicketNonce: 1, RecipientRandHash: crypto.Keccak256Hash([]byte("failed ticket")), CreationRound: 1, ParamsExpirationBlock: big.NewInt(0)}, Sig: make([]byte, 65), RecipientRand: big.NewInt(1)})) } releaseDependencies() - response, err = http.Get(address + "/external/discovery") + response, err = client.Get(address + "/external/discovery") require.NoError(t, err) defer response.Body.Close() var entries []discoveryEntry @@ -168,7 +142,7 @@ func TestPaidHTTPStartup(t *testing.T) { require.NoError(t, err) request.Header.Set("Livepeer-Payer-Address", payer.Address().Hex()) require.Eventually(t, func() bool { - response, err = http.DefaultClient.Do(request) + response, err = client.Do(request) if err != nil { return false } @@ -189,19 +163,20 @@ func TestPaidHTTPStartup(t *testing.T) { require.Equal(t, recipient.Bytes(), info.Address) require.Equal(t, recipient.Bytes(), info.TicketParams.Recipient) require.Eventually(t, func() bool { - ready, err := http.Get(metricsURL + "/readyz") + ready, err := client.Get(metricsURL + "/readyz") if err != nil { return false } defer ready.Body.Close() return ready.StatusCode == http.StatusOK }, 7*time.Second, 20*time.Millisecond, "paid readiness must become healthy after the first dependency and worker observations") - response, err = http.Get(metricsURL + "/metrics") + response, err = client.Get(metricsURL + "/metrics") require.NoError(t, err) metrics, err := io.ReadAll(response.Body) response.Body.Close() require.NoError(t, err) require.Equal(t, separate, strings.Contains(string(metrics), "livepeer_redemption_errors_total{"), "readiness must include completion of the first worker pass, including ticket failure") + require.Equal(t, separate, rpc.Stats().Calls["eth_estimateGas"] > 0, "the queued ticket reaches the injected estimation failure") }) } diff --git a/orchestrator/metrics.go b/orchestrator/metrics.go index 77a4b6e..ace96ac 100644 --- a/orchestrator/metrics.go +++ b/orchestrator/metrics.go @@ -258,3 +258,9 @@ func (e *PaymentEngine) recordPayment(operation string, err error, labels observ e.store.logger.Info("payment observed", "event", "payment", "operation", operation, "outcome", outcome, "app", labels["app"], "runner_id", labels["runner_id"], "app_id", labels["app_id"]) } } + +// Adapted from Mike Zupper's BlueClawNetwork/livepeer-node orchestrator/metrics.go, +// commit 2c2ce3cf8a51f5464ce4961bbb6c2e640f348ae0 (MIT; see LICENSE). +func (e *PaymentEngine) observeDuration(operation string, start time.Time, labels observability.Labels) { + e.metrics.Observe("payment_operation_duration_seconds", time.Since(start).Seconds(), labels.With("operation", operation)) +} diff --git a/orchestrator/metrics_test.go b/orchestrator/metrics_test.go index 3f6f9c7..9c0bdbd 100644 --- a/orchestrator/metrics_test.go +++ b/orchestrator/metrics_test.go @@ -33,20 +33,38 @@ type snapshotReader struct { func (c snapshotReader) Snapshot(ctx context.Context) (pm.ChainSnapshot, error) { return c.read(ctx) } -func TestPaidReadinessUsesCurrentPublication(t *testing.T) { +func newObservedPaymentChain(t *testing.T, recipient common.Address) (*test.ChainRPC, eth.PaymentChain, func()) { + t.Helper() f := test.NewChainRPC(t) + f.Update(func(s *test.ChainState) { s.Recipient = recipient }) rpc, err := eth.OpenRPC(f.URL) require.NoError(t, err) - defer rpc.Close() + t.Cleanup(rpc.Close) contracts, err := eth.NewContracts(rpc, f.Controller) require.NoError(t, err) - o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true, common.HexToAddress("0x1234")) + chain := eth.PaymentChain{Contracts: contracts} + chain.Observer = eth.NewPaymentObserver(chain, true, recipient) ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) - go func() { o.Run(ctx); close(done) }() - defer func() { cancel(); <-done }() - var snapshot eth.ChainSnapshot - require.Eventually(t, func() bool { snapshot, err = o.Snapshot(ctx); return err == nil }, time.Second, time.Millisecond) + go func() { chain.Observer.Run(ctx); close(done) }() + stop := func() { + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Error("payment observer did not stop") + } + } + t.Cleanup(stop) + require.Eventually(t, func() bool { _, err := chain.Snapshot(t.Context()); return err == nil }, time.Second, time.Millisecond) + return f, chain, stop +} + +func TestPaidReadinessUsesCurrentPublication(t *testing.T) { + f, chain, stop := newObservedPaymentChain(t, common.HexToAddress("0x1234")) + o := chain.Observer + snapshot, err := chain.Snapshot(t.Context()) + require.NoError(t, err) p := &paidReadiness{chain: true, active: true, storage: true, completed: time.Now(), workerCompleted: time.Now(), observer: o, snapshot: snapshot} registry := NewRegistry("", "http://localhost", time.Second, time.Minute) f.Update(func(s *test.ChainState) { @@ -57,7 +75,7 @@ func TestPaidReadinessUsesCurrentPublication(t *testing.T) { require.Eventually(t, func() bool { return o.Status().ObservedAt.After(snapshot.ObservedAt()) }, 7*time.Second, 10*time.Millisecond) require.ErrorIs(t, snapshot.Validate(), eth.ErrObservationUnavailable) require.True(t, p.ready(registry), "an older heartbeat must not conceal a healthy completed backfill") - _, store, ticket, chain, _ := newRecoveryTest(t, 15) + _, store, ticket, direct, _ := newRecoveryTest(t, 15) store.readiness = p stats, err := store.redemptionStats(t.Context()) require.NoError(t, err) @@ -83,15 +101,14 @@ func TestPaidReadinessUsesCurrentPublication(t *testing.T) { e.mu.Unlock() t.Fatal("pruner did not capture snapshot") } - cancel() - <-done + stop() e.mu.Unlock() require.ErrorIs(t, <-pruned, eth.ErrObservationUnavailable) require.Equal(t, snapshot.Block, e.lastSeenBlock) require.Len(t, e.ticketNonces, 1) require.Equal(t, 1, e.nonceCount) - chain.Observer = o - p.check(t.Context(), store, chain, &PaymentEngine{recipient: ticket.Recipient}, observability.New("orchestrator"), slog.New(slog.NewTextHandler(io.Discard, nil))) + direct.Observer = o + p.check(t.Context(), store, direct, &PaymentEngine{recipient: ticket.Recipient}, observability.New("orchestrator"), slog.New(slog.NewTextHandler(io.Discard, nil))) require.True(t, p.storage, "chain unavailability must not mark healthy storage unavailable") require.False(t, p.ready(registry)) } diff --git a/orchestrator/payment_engine.go b/orchestrator/payment_engine.go index 538f8b6..f9ccce2 100644 --- a/orchestrator/payment_engine.go +++ b/orchestrator/payment_engine.go @@ -89,6 +89,7 @@ func (e *PaymentEngine) MakeChallenge(ctx context.Context, appID, manifest strin func (e *PaymentEngine) makeAppChallenge(ctx context.Context, appID, manifest string, payer ethcommon.Address, price int64, unit, service string, registration *appLifetime) (challenge Challenge, result error) { labels := e.paymentLabels(appID, manifest) defer func() { e.recordPayment("challenge", result, labels) }() + defer e.observeDuration("challenge", time.Now(), labels) return e.makeChallenge(ctx, appID, manifest, payer, price, unit, service, registration) } @@ -167,6 +168,7 @@ func (e *PaymentEngine) ChallengeForManifest(appID, manifest, service string) (C func (e *PaymentEngine) RefreshChallenge(ctx context.Context, manifest string, payer ethcommon.Address, service string) (challenge Challenge, result error) { labels := e.paymentLabels("", manifest) defer func() { e.recordPayment("refresh", result, labels) }() + defer e.observeDuration("refresh", time.Now(), labels) e.mu.Lock() current := e.sessions[manifest] if current == nil { @@ -223,6 +225,7 @@ func decodeHeaders(paymentHeader, segmentHeader string) (wire.Payment, wire.SegD func (e *PaymentEngine) Receive(ctx context.Context, appID, manifest, paymentHeader, segmentHeader string) (payerResult ethcommon.Address, credit *big.Rat, result error) { labels := e.paymentLabels(appID, manifest) defer func() { e.recordPayment("receive", result, labels) }() + defer e.observeDuration("receive", time.Now(), labels) payment, segment, err := decodeHeaders(paymentHeader, segmentHeader) if err != nil { return ethcommon.Address{}, nil, err diff --git a/orchestrator/payment_recipient_test.go b/orchestrator/payment_recipient_test.go index 79541e7..0f13154 100644 --- a/orchestrator/payment_recipient_test.go +++ b/orchestrator/payment_recipient_test.go @@ -252,6 +252,8 @@ func TestRecipientBatchReplayAndExpiry(t *testing.T) { require.NoError(t, err) require.Equal(t, challenges, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "challenge"})) require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "refresh", "outcome": "success"})) + require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operation_duration_seconds", map[string]string{"operation": "refresh"})) + require.Equal(t, challenges, test.MetricSum(t, e.metrics.Registry, "payment_operation_duration_seconds", map[string]string{"operation": "challenge"})) // A newer valid auth token cannot reset nonce tracking for old PM params. info.Auth = e.authToken("session", time.Now().Add(2*time.Hour).Unix()) require.ErrorIs(t, receiveRecipientPayment(t, e, key, info, 1), ErrInvalidPayment) @@ -444,9 +446,11 @@ func TestPaidFailureOutcomes(t *testing.T) { request = request.WithContext(ctx) } response := httptest.NewRecorder() + beforeDuration := test.MetricSum(t, e.metrics.Registry, "payment_operation_duration_seconds", map[string]string{"operation": operation}) server.ServeHTTP(response, request) require.Equal(t, status, response.Code, response.Body.String()) require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": operation, "outcome": outcome})) + require.Equal(t, beforeDuration+1, test.MetricSum(t, e.metrics.Registry, "payment_operation_duration_seconds", map[string]string{"operation": operation})) require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "app_calls_total", map[string]string{"outcome": appOutcome})) require.Zero(t, test.MetricSum(t, e.metrics.Registry, "upstream_requests_total", nil)) }) @@ -477,31 +481,8 @@ func TestLiveChargeObservesOnlyDebitAttempts(t *testing.T) { // Copyright (c) 2017 Livepeer; MIT notice in LICENSE. func TestPaymentsAndReadinessUseObservedRecipient(t *testing.T) { e, _, key, _ := newRecipientTestEngine(t) - f := test.NewChainRPC(t) - f.Update(func(s *test.ChainState) { s.Recipient = e.recipient }) - rpc, err := eth.OpenRPC(f.URL) - require.NoError(t, err) - t.Cleanup(rpc.Close) - contracts, err := eth.NewContracts(rpc, f.Controller) - require.NoError(t, err) - chain := eth.PaymentChain{Contracts: contracts} - chain.Observer = eth.NewPaymentObserver(chain, true, e.recipient) - e.chain = pm.EthereumChain{Client: chain} - ctx, cancel := context.WithCancel(t.Context()) - done := make(chan struct{}) - go func() { chain.Observer.Run(ctx); close(done) }() - t.Cleanup(func() { - cancel() - select { - case <-done: - case <-time.After(time.Second): - t.Error("recipient observation did not stop") - } - }) - require.Eventually(t, func() bool { - _, err := chain.Snapshot(t.Context()) - return err == nil - }, time.Second, time.Millisecond) + f, chain, _ := newObservedPaymentChain(t, e.recipient) + e.chain = chain registry := NewRegistry("", "https://orch.example", time.Second, time.Minute) ready := &paidReadiness{storage: true, completed: time.Now(), workerCompleted: time.Now(), observer: chain.Observer} f.Reset() diff --git a/pm/ethchain.go b/pm/ethchain.go index 86eb4d2..2624351 100644 --- a/pm/ethchain.go +++ b/pm/ethchain.go @@ -23,24 +23,4 @@ type PaymentChain interface { type PayerFunds = eth.SenderInfo -// EthereumChain adapts retained Ethereum contract reads to the payment -// engine's interface. Contract ABI and RPC logic remain in eth. -type EthereumChain struct{ Client eth.PaymentChain } - -func (c EthereumChain) Snapshot(ctx context.Context) (ChainSnapshot, error) { - state, err := c.Client.Snapshot(ctx) - if err != nil { - return ChainSnapshot{}, err - } - return state, nil -} - -func (c EthereumChain) PayerFunds(ctx context.Context, payer, recipient ethcommon.Address) (PayerFunds, error) { - return c.Client.SenderInfo(ctx, payer, recipient) -} - -func (c EthereumChain) IsActiveAt(ctx context.Context, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) { - return c.Client.IsActiveAt(ctx, recipient, snapshot) -} - -var _ PaymentChain = EthereumChain{} +var _ PaymentChain = eth.PaymentChain{} diff --git a/signer/app.go b/signer/app.go index 5a61bce..9212d66 100644 --- a/signer/app.go +++ b/signer/app.go @@ -187,8 +187,9 @@ func serve(parent context.Context, p Params) error { } chain := eth.PaymentChain{Contracts: contracts} observer := eth.NewPaymentObserver(chain, p.Network == nodeconfig.Mainnet, ethcommon.Address{}) + service.metrics.PaymentObserver(observer, slog.With("component", "signer")) chain.Observer = observer - service.SetPaymentChain(pm.EthereumChain{Client: chain}) + service.SetPaymentChain(chain) policy, _ := newPricePolicy(p.MaxHourlyPrice, p.MaxFixedPrice) policy.metrics = service.metrics if p.WeiPerUSD == nil { @@ -240,7 +241,7 @@ func serve(parent context.Context, p Params) error { _, _ = io.WriteString(w, "ok\n") }) metricsMux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) { - ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(eth.WithRPCWork(r.Context(), "background"), 5*time.Second) defer cancel() funds, err := service.paymentChain.PayerFunds(ctx, key.Address(), ethcommon.Address{}) if err == nil { @@ -269,7 +270,7 @@ func serve(parent context.Context, p Params) error { workers.Go(func() { service.RunDiscovery(ctx) }) if p.WeiPerUSD == nil { workers.Go(func() { - refreshPriceFeed(ctx, time.Hour, service.pricePolicy, func(readCtx context.Context) (*big.Rat, time.Time, error) { + refreshPriceFeed(eth.WithRPCWork(ctx, "background"), time.Hour, service.pricePolicy, func(readCtx context.Context) (*big.Rat, time.Time, error) { return contracts.WeiPerUSD(readCtx, p.ETHUSDFeed, p.ETHUSDMaxAge) }) }) diff --git a/signer/app_test.go b/signer/app_test.go index 4d9a97f..e8ec4c5 100644 --- a/signer/app_test.go +++ b/signer/app_test.go @@ -2,7 +2,6 @@ package signer import ( "context" - "encoding/hex" "encoding/json" "fmt" "io" @@ -14,13 +13,11 @@ import ( "net/url" "os" "path/filepath" - "strings" "sync/atomic" "testing" "time" ethcommon "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/crypto" "github.com/j0sh/boa/pkg/boa" "github.com/livepeer/node/internal/test" "github.com/spf13/cobra" @@ -284,70 +281,32 @@ func TestSignerReadiness(t *testing.T) { } key, keyFile, passwordPath := testSignerKey(t) p.KeystoreFile, p.KeystorePasswordFile = keyFile, passwordPath - var failure, collateralReads atomic.Int32 - failure.Store(tc.failure) - webhookCalled := make(chan struct{}, 1) - rpc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/list" { - webhookCalled <- struct{}{} - _, _ = io.WriteString(w, "[]") - return - } - var req struct { - Method string - Params []json.RawMessage + rpc := test.NewChainRPC(t) + rpc.Update(func(s *test.ChainState) { + s.ChainID, s.Payer = 1, key.Address() + s.Deposit, s.Reserve = 1000, 1000 + switch tc.failure { + case 1: + s.Fail = "eth_call" + case 2: + s.Deposit = 0 + case 3: + s.Reserve = 0 + case 4: + s.WithdrawRound = 6 } - require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) - var result any - switch req.Method { - case "eth_chainId": - result = "0x1" - case "eth_getBlockByNumber": - if failure.Load() != 1 { - result = map[string]string{"number": "0x32", "hash": ethcommon.HexToHash("0x1234").Hex(), "timestamp": fmt.Sprintf("0x%x", time.Now().Unix())} - } - case "eth_call": - if failure.Load() == 1 { - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32000}}) - return - } - var call struct { - Data string `json:"input"` - } - require.NoError(t, json.Unmarshal(req.Params[0], &call)) - selector := func(sig string) bool { - return strings.HasPrefix(call.Data, "0x"+hex.EncodeToString(crypto.Keccak256([]byte(sig))[:4])) - } - switch { - case selector("getContract(bytes32)"): - result = fmt.Sprintf("0x%064x", 1) - case selector("lastInitializedRound()"), selector("currentRound()"): - result = fmt.Sprintf("0x%064x", 5) - case selector("blockHashForRound(uint256)"): - result = ethcommon.HexToHash("0x1234").Hex() - case selector("getSenderInfo(address)"): - collateralReads.Add(1) - deposit, reserve, withdraw := 1000, 1000, 0 - switch failure.Load() { - case 2: - deposit = 0 - case 3: - reserve = 0 - case 4: - withdraw = 6 - } - result = fmt.Sprintf("0x%064x%064x%064x%064x", deposit, withdraw, reserve, 0) - case selector("description()"): - result = "0x" // The oracle is unavailable during startup. - default: - t.Errorf("unexpected RPC call: %s", call.Data) - } + if !fixedRate { + s.Fail = "PriceFeed.description" } - _ = json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": result}) + }) + webhookCalled := make(chan struct{}, 1) + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + webhookCalled <- struct{}{} + _, _ = io.WriteString(w, "[]") })) - defer rpc.Close() - p.RPCURL = testURL(t, rpc.URL) - p.OrchestratorWebhook = testURL(t, rpc.URL+"/list") + defer webhook.Close() + p.RPCURL, p.Controller = testURL(t, rpc.URL), rpc.Controller + p.OrchestratorWebhook = testURL(t, webhook.URL) p.OrchestratorWebhookRefreshInterval = time.Hour p.DiscoveryGrants = []string{"127.0.0.1"} free := func() boa.HTTPBind { @@ -381,7 +340,7 @@ func TestSignerReadiness(t *testing.T) { t.Error("signer did not stop") } }) - client := &http.Client{Timeout: time.Second} + client := &http.Client{Timeout: time.Second, Transport: &http.Transport{DisableKeepAlives: true}} select { case <-webhookCalled: case <-time.After(3 * time.Second): @@ -396,15 +355,17 @@ func TestSignerReadiness(t *testing.T) { return response.StatusCode } if !fixedRate || tc.failure != 0 { - require.Eventually(t, func() bool { return get("/readyz") == 503 && (tc.failure == 1 || collateralReads.Load() > 0) }, 3*time.Second, 10*time.Millisecond) + require.Eventually(t, func() bool { + return get("/readyz") == 503 && (tc.failure == 1 || rpc.Stats().Calls["TicketBroker.getSenderInfo"] > 0) + }, 3*time.Second, 10*time.Millisecond) require.Equal(t, 200, get("/healthz")) return } require.Eventually(t, func() bool { return get("/readyz") == 200 }, 3*time.Second, 10*time.Millisecond) - failure.Store(1) + rpc.Update(func(s *test.ChainState) { s.Fail = "eth_call" }) require.Equal(t, 200, get("/readyz"), "warm readiness retains a fresh observation during an RPC outage") require.Equal(t, 200, get("/healthz")) - failure.Store(0) + rpc.Update(func(s *test.ChainState) { s.Fail = "" }) _, err := outbox.enqueue(t.Context(), testEvent()) require.NoError(t, err) require.Equal(t, 503, get("/readyz"), "a full outbox must make the signer unready") diff --git a/signer/payment_test.go b/signer/payment_test.go index 96818bb..93705b2 100644 --- a/signer/payment_test.go +++ b/signer/payment_test.go @@ -14,6 +14,7 @@ import ( ethcommon "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/livepeer/node/eth" + "github.com/livepeer/node/internal/observability" "github.com/livepeer/node/internal/test" "github.com/livepeer/node/orchestrator" "github.com/livepeer/node/pm" @@ -161,37 +162,79 @@ func TestRecipientReservesWinningLiabilityAcrossConcurrentSessions(t *testing.T) // Adapted from BlueClawNetwork/livepeer-node signer/payment_test.go, // 285eda984d7cce4e3030ef0510d63ae142534dd7, by Mike Zupper (MIT; see LICENSE). -func TestPaidRequestDirectRPCBudget(t *testing.T) { - s, _ := testService(t) - defer s.Close() - rpc := test.NewChainRPC(t) - client, err := eth.OpenRPC(rpc.URL) - require.NoError(t, err) - defer client.Close() - contracts, err := eth.NewContracts(client, rpc.Controller) - require.NoError(t, err) - // Explicitly use the direct adapter: cached configurations can do less work. - direct := pm.EthereumChain{Client: eth.PaymentChain{Contracts: contracts}} - s.SetPaymentChain(direct) +type paidPath struct { + service *Service + engine *orchestrator.PaymentEngine + rpc *test.ChainRPC + recipient ethcommon.Address + metrics *observability.Metrics + stop func() +} + +func newPaidPath(t *testing.T, observed bool) *paidPath { + t.Helper() + key, _, _ := testSignerKey(t) + p := &paidPath{service: newService(key), rpc: test.NewChainRPC(t), recipient: ethcommon.HexToAddress("0x1234"), metrics: observability.New("orchestrator")} + t.Cleanup(func() { _ = p.service.Close() }) + ctx, cancel := context.WithCancel(t.Context()) + var observers sync.WaitGroup + p.stop = func() { cancel(); observers.Wait() } + t.Cleanup(p.stop) + newChain := func(metrics *observability.Metrics, recipient ethcommon.Address) eth.PaymentChain { + client, err := eth.OpenRPC(p.rpc.URL) + require.NoError(t, err) + t.Cleanup(func() { p.stop(); client.Close() }) + client.SetObserver(metrics.RPC) + contracts, err := eth.NewContracts(client, p.rpc.Controller) + require.NoError(t, err) + chain := eth.PaymentChain{Contracts: contracts} + if observed { + o := eth.NewPaymentObserver(chain, true, recipient) + chain.Observer = o + metrics.PaymentObserver(o, nil) + observers.Go(func() { o.Run(ctx) }) + require.Eventually(t, func() bool { return o.Status().Healthy }, 5*time.Second, time.Millisecond) + } + return chain + } + direct := newChain(p.metrics, p.recipient) + p.service.SetPaymentChain(newChain(p.service.metrics, ethcommon.Address{})) store, err := orchestrator.OpenRedeemerDB(filepath.Join(t.TempDir(), "payments.sqlite")) require.NoError(t, err) - defer store.Close() - recipient := ethcommon.HexToAddress("0x1234") + t.Cleanup(func() { _ = store.Close() }) maximum := new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1)) - engine, err := orchestrator.NewPaymentEngine(store, direct, recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) + p.engine, err = orchestrator.NewPaymentEngine(store, direct, p.recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) require.NoError(t, err) - _, err = engine.MakeChallenge(t.Context(), "runner", "manifest", s.key.Address(), 9, "fixed", "https://orch.example") + return p +} + +func (p *paidPath) challenge(t *testing.T, ctx context.Context, manifest string) wire.OrchestratorInfo { + t.Helper() + _, err := p.engine.MakeChallenge(ctx, "runner", manifest, p.service.key.Address(), 9, "fixed", "https://orch.example") require.NoError(t, err) - info, err := engine.ChallengeInfo("manifest") + info, err := p.engine.ChallengeInfo(manifest) require.NoError(t, err) - require.Equal(t, 6, rpc.Stats().RoundTrips, "challenge: snapshot (4) + activity (2)") + return info +} - response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) +func (p *paidPath) sign(t *testing.T, info wire.OrchestratorInfo) paymentResponse { + t.Helper() + response := postPayment(t, p.service, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) require.Equal(t, 200, response.Code, response.Body.String()) var signed paymentResponse require.NoError(t, json.Unmarshal(response.Body.Bytes(), &signed)) + return signed +} + +func TestPaidRequestDirectRPCBudget(t *testing.T) { + p := newPaidPath(t, false) + s, engine, rpc, recipient := p.service, p.engine, p.rpc, p.recipient + info := p.challenge(t, t.Context(), "manifest") + require.Equal(t, 6, rpc.Stats().RoundTrips, "challenge: snapshot (4) + activity (2)") + + signed := p.sign(t, info) require.Equal(t, 13, rpc.Stats().RoundTrips, "signer adds one funds observation (7)") - _, _, err = engine.Receive(t.Context(), "runner", "manifest", signed.Payment, signed.SegCreds) + _, _, err := engine.Receive(t.Context(), "runner", "manifest", signed.Payment, signed.SegCreds) require.NoError(t, err) stats := rpc.Stats() require.Equal(t, 22, stats.RoundTrips, "receipt adds funds (7) + activity (2); baseline was 29") @@ -203,11 +246,8 @@ func TestPaidRequestDirectRPCBudget(t *testing.T) { // A signer's retained observation cannot prove that the recipient is behind // expiry. The recipient must still enforce its own clock at the boundary. - _, err = engine.MakeChallenge(t.Context(), "runner", "divergent", s.key.Address(), 9, "fixed", "https://orch.example") - require.NoError(t, err) - info, err = engine.ChallengeInfo("divergent") - require.NoError(t, err) - older, err := direct.PayerFunds(t.Context(), s.key.Address(), recipient) + info = p.challenge(t, t.Context(), "divergent") + older, err := s.paymentChain.PayerFunds(t.Context(), s.key.Address(), recipient) require.NoError(t, err) s.SetPaymentChain(payerChainFunc(func(context.Context, ethcommon.Address, ethcommon.Address) (pm.PayerFunds, error) { return older, nil })) rpc.Update(func(state *test.ChainState) { @@ -215,41 +255,29 @@ func TestPaidRequestDirectRPCBudget(t *testing.T) { state.Number++ state.Hash = ethcommon.HexToHash("0xeeee") }) - response = postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) - require.Equal(t, 200, response.Code, response.Body.String()) - require.NoError(t, json.Unmarshal(response.Body.Bytes(), &signed)) + signed = p.sign(t, info) _, _, err = engine.Receive(t.Context(), "runner", "divergent", signed.Payment, signed.SegCreds) require.ErrorIs(t, err, orchestrator.ErrInvalidPayment) +} - const manifest = "invalidated-signing" - _, err = engine.MakeChallenge(t.Context(), "runner", manifest, s.key.Address(), 9, "fixed", "https://orch.example") - require.NoError(t, err) - info, err = engine.ChallengeInfo(manifest) +func TestPaidRequestCachedRPCBudget(t *testing.T) { + p := newPaidPath(t, true) + before := p.rpc.Stats().RoundTrips + p.request(t, t.Context(), "cold") + require.Equal(t, 4, p.rpc.Stats().RoundTrips-before, "one cold pair read in each process") + require.EqualValues(t, 4, test.MetricSum(t, p.metrics.Registry, "rpc_work_total", map[string]string{"work": "foreground", "kind": "transport"})+test.MetricSum(t, p.service.metrics.Registry, "rpc_work_total", map[string]string{"work": "foreground", "kind": "transport"})) + before = p.rpc.Stats().RoundTrips + p.request(t, t.Context(), "warm") + require.Equal(t, before, p.rpc.Stats().RoundTrips, "new manifests for a known payer read memory") +} + +// Adapted from Mike Zupper's BlueClawNetwork/livepeer-node signer/payment_test.go, +// a4fe74293a04937c5c5514b1a3706718a502eba0 (MIT; see LICENSE). +func (p *paidPath) request(t *testing.T, ctx context.Context, manifest string) { + t.Helper() + ctx = eth.WithRPCWork(ctx, "foreground") + info := p.challenge(t, ctx, manifest) + signed := p.sign(t, info) + _, _, err := p.engine.Receive(ctx, "runner", manifest, signed.Payment, signed.SegCreds) require.NoError(t, err) - for _, during := range []bool{false, true} { - observer := eth.NewPaymentObserver(direct.Client, false, ethcommon.Address{}) - ctx, cancel := context.WithCancel(t.Context()) - done := make(chan struct{}) - go func() { observer.Run(ctx); close(done) }() - stop := func() { cancel(); <-done } - defer stop() - require.Eventually(t, func() bool { _, err := observer.Snapshot(ctx); return err == nil }, time.Second, time.Millisecond) - observed := direct - observed.Client.Observer = observer - s.SetPaymentChain(observed) - draft, err := s.preparePayment(t.Context(), "fixed", manifest, info, paymentState{}, -1, time.Now()) - require.NoError(t, err) - key := &failingPaymentSigner{TicketSigner: s.key} - if during { - key.before = stop - } else { - stop() - } - payment, err := draft.Sign(t.Context(), key, s.metrics) - require.ErrorIs(t, err, eth.ErrObservationUnavailable) - require.Empty(t, payment) - if !during { - require.Zero(t, key.calls.Load(), "invalidated observations must not produce signatures") - } - } } diff --git a/signer/service.go b/signer/service.go index 043acb8..49bf650 100644 --- a/signer/service.go +++ b/signer/service.go @@ -345,6 +345,7 @@ func checkMaxPrice(req paymentRequest, price wire.PriceInfo) error { func (s *Service) generate(w http.ResponseWriter, r *http.Request) { paymentType, outcome := "unknown", "rejected" + start := time.Now() defer func() { if outcome != "success" && errors.Is(r.Context().Err(), context.Canceled) { outcome = "client_cancelled" @@ -352,6 +353,9 @@ func (s *Service) generate(w http.ResponseWriter, r *http.Request) { outcome = "error" } s.metrics.Add("signer_payments_total", 1, observability.Labels{"type": paymentType, "outcome": outcome}) + // Adapted from Mike Zupper's BlueClawNetwork/livepeer-node signer/service.go, + // commit 2c2ce3cf8a51f5464ce4961bbb6c2e640f348ae0 (MIT; see LICENSE). + s.metrics.Observe("signer_payment_duration_seconds", time.Since(start).Seconds(), observability.Labels{"type": paymentType}) }() // MaxBytesReader needs net/http's original writer to close oversized requests. body, err := io.ReadAll(http.MaxBytesReader(httpsnoop.Unwrap(w), r.Body, 1<<20)) diff --git a/signer/service_test.go b/signer/service_test.go index 6cb07ee..c9dd446 100644 --- a/signer/service_test.go +++ b/signer/service_test.go @@ -379,7 +379,9 @@ func TestPaymentParamsValidation(t *testing.T) { t.Run(tc.name, func(t *testing.T) { info := original tc.change(&info) + before := test.MetricSum(t, s.metrics.Registry, "signer_payment_duration_seconds", nil) w := postPayment(t, s, map[string]any{"orchestrator": base64.StdEncoding.EncodeToString(wire.EncodeOrchestratorInfo(info)), "type": "fixed"}) + require.Equal(t, before+1, test.MetricSum(t, s.metrics.Registry, "signer_payment_duration_seconds", nil)) if tc.status == 200 { require.Equal(t, 200, w.Code, w.Body.String()) var response paymentResponse @@ -424,6 +426,32 @@ func (s *failingPaymentSigner) SignMessage(message []byte) ([]byte, error) { return s.TicketSigner.SignMessage(message) } +func TestPaidRequestInvalidatedSigning(t *testing.T) { + for _, during := range []bool{false, true} { + t.Run(fmt.Sprint(during), func(t *testing.T) { + p := newPaidPath(t, true) + s := p.service + info := p.challenge(t, t.Context(), "invalidated-signing") + draft, err := s.preparePayment(t.Context(), "fixed", "invalidated-signing", info, paymentState{}, -1, time.Now()) + require.NoError(t, err) + key := &failingPaymentSigner{TicketSigner: s.key} + if during { + key.before = p.stop + } else { + p.stop() + } + payment, err := draft.Sign(t.Context(), key, s.metrics) + require.ErrorIs(t, err, eth.ErrObservationUnavailable) + require.Empty(t, payment) + if during { + require.Positive(t, key.calls.Load(), "invalidate inside signing") + } else { + require.Zero(t, key.calls.Load(), "invalidated observations must not produce signatures") + } + }) + } +} + func TestPaymentFailureStatuses(t *testing.T) { s, original := testService(t) logs := captureSignerLogs(t)