Instrument payment phases and simplify shared chain paths

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
 

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 {

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)

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,16 +82,20 @@ 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
 }
 
 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,15 +169,13 @@ 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
 		}
 		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