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)