Cache configured recipient activity in payment observations

eth/contracts_test.go

@@ -79,7 +79,7 @@ func TestRedemptionSubmission(t *testing.T) {
 	require.Equal(t, sentHash, hash)
 	// Redemption resolves the live broker independently of an uninitialized observer.
 	broker = common.HexToAddress("0x2001")
-	chain := PaymentChain{Contracts: contracts, Observer: NewPaymentObserver(PaymentChain{Contracts: contracts}, false)}
+	chain := PaymentChain{Contracts: contracts, Observer: NewPaymentObserver(PaymentChain{Contracts: contracts}, false, common.Address{})}
 	hash, err = chain.Redeem(t.Context(), key, big.NewInt(1), ticket)
 	require.NoError(t, err)
 	require.Equal(t, sentHash, hash)

eth/payment.go

@@ -30,25 +30,34 @@ func (c PaymentChain) IsActive(ctx context.Context, recipient ethcommon.Address)
 }
 
 func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) {
+	if err := ctx.Err(); err != nil {
+		return false, err
+	}
 	if err := snapshot.Validate(); err != nil {
 		return false, err
 	}
+	if c.Observer != nil || snapshot.observer != nil {
+		if snapshot.observer != c.Observer || c.Contracts != c.Observer.chain.Contracts || recipient == (ethcommon.Address{}) || recipient != c.Observer.recipient {
+			return false, ErrObservationUnavailable
+		}
+		return snapshot.recipientActive, snapshot.recipientErr
+	}
 	if snapshot.blockReference == nil {
 		return false, errors.New("canonical snapshot required")
 	}
 	address, err := c.Contracts.paymentAddress(ctx, snapshot, "bondingManager")
 	if err != nil {
 		return false, err
 	}
+	return c.activeAt(ctx, address, recipient, snapshot)
+}
+
+func (c PaymentChain) activeAt(ctx context.Context, address, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) {
 	bonding, err := contracts.NewBondingManagerCaller(address, c.Contracts.callerAt(snapshot.blockReference))
 	if err != nil {
 		return false, err
 	}
-	active, err := bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient)
-	if err == nil {
-		err = snapshot.Validate()
-	}
-	return active, err
+	return bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient)
 }
 
 // Receipt reports a submitted redemption without retrying or resubmitting it.
@@ -81,16 +90,18 @@ func (c PaymentChain) Receipt(ctx context.Context, hash ethcommon.Hash) (confirm
 }
 
 type ChainSnapshot struct {
-	Block          *big.Int // Protocol L1 block, including on Arbitrum.
-	Round          *big.Int // Last initialized round.
-	RoundHash      ethcommon.Hash
-	CurrentRound   *big.Int // Protocol round; may exceed the initialized Round.
-	blockReference map[string]any
-	addresses      paymentContracts
-	observer       *PaymentObserver
-	generation     uint64
-	observedAt     time.Time
-	blockTime      time.Time
+	Block           *big.Int // Protocol L1 block, including on Arbitrum.
+	Round           *big.Int // Last initialized round.
+	RoundHash       ethcommon.Hash
+	CurrentRound    *big.Int // Protocol round; may exceed the initialized Round.
+	blockReference  map[string]any
+	addresses       paymentContracts
+	observer        *PaymentObserver
+	generation      uint64
+	observedAt      time.Time
+	blockTime       time.Time
+	recipientActive bool
+	recipientErr    error
 }
 
 func (c PaymentChain) Snapshot(ctx context.Context) (ChainSnapshot, error) {

eth/payment_contracts_test.go

@@ -20,6 +20,7 @@ func TestObserverConfirmedPaymentContracts(t *testing.T) {
 	}{{"RoundsManager", 5}, {"TicketBroker", 5}, {"BondingManager", 5}, {"RoundsManager", 9}} {
 		t.Run(tc.contract+big.NewInt(int64(tc.round)).String(), func(t *testing.T) {
 			f, o, now := newTestObserver(t)
+			o.recipient = common.HexToAddress("0x3000")
 			require.NoError(t, o.refresh(t.Context()))
 			before, err := o.Snapshot(t.Context())
 			require.NoError(t, err)
@@ -68,6 +69,7 @@ func TestObserverConfirmedPaymentContracts(t *testing.T) {
 				}
 			}
 			require.Zero(t, f.Stats().Calls["Controller.getContract"], "warm payment reads retain confirmed addresses")
+			require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "replacement activity belongs to the supplied observation")
 		})
 	}
 }
@@ -121,7 +123,7 @@ func TestObserverRegistryScopeAndFinalizedReads(t *testing.T) {
 	require.NoError(t, err)
 	f.Reset()
 	_, err = (PaymentChain{Contracts: other}).IsActiveAt(t.Context(), common.HexToAddress("0x3000"), s)
-	require.ErrorContains(t, err, "another contract client")
+	require.ErrorIs(t, err, ErrObservationUnavailable)
 	require.Zero(t, f.Stats().RoundTrips)
 	advanceObserver(f, *now)
 	f.Update(func(state *test.ChainState) {

eth/payment_observer.go

@@ -34,11 +34,12 @@ type PaymentObserver struct {
 	contact, progress time.Time
 	l1Progress        time.Time
 	l1HighWater       *big.Int
+	recipient         common.Address
 }
 
-func NewPaymentObserver(chain PaymentChain, requireL1 bool) *PaymentObserver {
+func NewPaymentObserver(chain PaymentChain, requireL1 bool, recipient common.Address) *PaymentObserver {
 	chain.Observer = nil
-	return &PaymentObserver{chain: chain, interval: 5 * time.Second, maxAge: 30 * time.Second, now: time.Now, requireL1: requireL1 || chain.Contracts.RPC.requireL1}
+	return &PaymentObserver{chain: chain, interval: 5 * time.Second, maxAge: 30 * time.Second, now: time.Now, requireL1: requireL1 || chain.Contracts.RPC.requireL1, recipient: recipient}
 }
 
 // Run has one service-owned caller. Every RPC inherits the service lifetime.
@@ -62,13 +63,14 @@ func (o *PaymentObserver) Run(ctx context.Context) {
 // A single provider cannot establish an absolute bound on protocol L1 lag.
 type ObserverStatus struct {
 	ObservedAt, ContactAt, HeadProgressAt, L1ProgressAt time.Time
-	Healthy                                             bool
+	Healthy, RecipientActive                            bool
 }
 
 func (o *PaymentObserver) Status() ObserverStatus {
 	o.mu.RLock()
 	defer o.mu.RUnlock()
-	return ObserverStatus{o.current.observedAt, o.contact, o.progress, o.l1Progress, o.validate(o.current) == nil}
+	healthy := o.validate(o.current) == nil
+	return ObserverStatus{o.current.observedAt, o.contact, o.progress, o.l1Progress, healthy, healthy && o.recipient != (common.Address{}) && o.current.recipientErr == nil && o.current.recipientActive}
 }
 
 func (o *PaymentObserver) Snapshot(ctx context.Context) (ChainSnapshot, error) {
@@ -180,6 +182,14 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	// Local admission follows go-livepeer/core/orchestrator.go isActive at
+	// 597f882c (Nico Vergauwen, 5bfee8f7; MIT, see LICENSE). Retain node's
+	// canonical contract predicate rather than duplicating round arithmetic.
+	// Per-entry failure isolation adapts Mike Zupper's PR #1 eth/chainview.go,
+	// 68c7834921c7218d16dbd883e2b7e87f6d7804d5 (MIT, see LICENSE).
+	if o.recipient != (common.Address{}) {
+		s.recipientActive, s.recipientErr = o.chain.activeAt(ctx, s.addresses.bonding, o.recipient, s)
+	}
 	if err := ctx.Err(); err != nil {
 		return err
 	}

eth/payment_observer_test.go

@@ -25,7 +25,7 @@ func newTestObserver(t *testing.T) (*test.ChainRPC, *PaymentObserver, *time.Time
 	t.Cleanup(rpc.Close)
 	c, err := NewContracts(rpc, f.Controller)
 	require.NoError(t, err)
-	o := NewPaymentObserver(PaymentChain{Contracts: c}, true)
+	o := NewPaymentObserver(PaymentChain{Contracts: c}, true, common.Address{})
 	now := time.Now().Truncate(time.Second)
 	o.now = func() time.Time { return now }
 	return f, o, &now
@@ -69,13 +69,13 @@ func TestObserverExplicitArbitrumChainIDRequiresL1(t *testing.T) {
 	f.Update(func(s *test.ChainState) { s.ChainID, s.OmitL1 = 42161, true })
 	rpc := original.chain.Contracts.RPC
 	require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(42161)))
-	o := NewPaymentObserver(original.chain, false) // No named network preset.
+	o := NewPaymentObserver(original.chain, false, common.Address{}) // No named network preset.
 	require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
 	// A non-Arbitrum custom chain continues to use its native block clock.
 	f.Update(func(s *test.ChainState) { s.ChainID = 1 })
 	rpc.requireL1 = false
 	require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(1)))
-	o = NewPaymentObserver(original.chain, false)
+	o = NewPaymentObserver(original.chain, false, common.Address{})
 	require.NoError(t, o.refresh(t.Context()))
 }
 
@@ -337,3 +337,74 @@ func TestObserverGapKeepsCanonicalStateAvailable(t *testing.T) {
 		}
 	}
 }
+
+// Round boundary/removal cases adapt Nico Vergauwen's go-livepeer
+// eth/watchers/orchestratorwatcher_test.go at 597f882cb9f4d42ad8430b0f592afd5929af4a6b.
+// Local/pinned failure isolation adapts Mike Zupper's PR #1 eth/chainview_test.go
+// at bbe126e5eddbf59f9cc9f94d3c37703215c26850. MIT notice in LICENSE.
+func TestObserverRecipientActivity(t *testing.T) {
+	f, o, now := newTestObserver(t)
+	o.recipient = common.HexToAddress("0x1234")
+	c := PaymentChain{Contracts: o.chain.Contracts, Observer: o}
+	f.Update(func(s *test.ChainState) { s.Recipient, s.Activation, s.Deactivation = o.recipient, 6, 8 })
+	_, err := c.IsActive(t.Context(), o.recipient)
+	require.ErrorIs(t, err, ErrObservationUnavailable)
+	for _, step := range []struct {
+		round  uint64
+		active bool
+	}{{5, false}, {6, true}, {7, true}, {8, false}} {
+		f.Update(func(s *test.ChainState) { s.CurrentRound = step.round })
+		require.NoError(t, o.refresh(t.Context()))
+		snapshot, err := c.Snapshot(t.Context())
+		require.NoError(t, err)
+		require.EqualValues(t, 5, snapshot.Round.Int64(), "activity uses current round even before initialization")
+		before := f.Stats()
+		active, err := c.IsActiveAt(t.Context(), o.recipient, snapshot)
+		require.NoError(t, err)
+		require.Equal(t, step.active, active)
+		_, err = c.IsActiveAt(t.Context(), common.HexToAddress("0x9999"), snapshot)
+		require.ErrorIs(t, err, ErrObservationUnavailable)
+		unowned := snapshot
+		unowned.observer = nil
+		_, err = c.IsActiveAt(t.Context(), o.recipient, unowned)
+		require.ErrorIs(t, err, ErrObservationUnavailable)
+		cancelled, cancel := context.WithCancel(t.Context())
+		cancel()
+		_, err = c.IsActiveAt(cancelled, o.recipient, snapshot)
+		require.ErrorIs(t, err, context.Canceled)
+		require.Equal(t, before, f.Stats(), "local checks cannot start activity or lookup RPCs")
+		advanceObserver(f, *now)
+		f.Update(func(s *test.ChainState) { s.Deactivation = 0 })
+		require.NoError(t, o.refresh(t.Context()))
+		active, err = c.IsActiveAt(t.Context(), o.recipient, snapshot)
+		require.NoError(t, err)
+		require.Equal(t, step.active, active, "a newer activity result cannot change a held funds observation")
+		advanceObserver(f, *now)
+		f.Update(func(s *test.ChainState) { s.Deactivation = 8 })
+	}
+
+	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)
+	require.NoError(t, err)
+	_, err = c.IsActive(t.Context(), o.recipient)
+	require.Error(t, err, "unknown must remain distinct from known inactive")
+	advanceObserver(f, *now)
+	f.Update(func(s *test.ChainState) { s.Fail, s.CurrentRound, s.Deactivation = "", 6, 8 })
+	require.NoError(t, o.refresh(t.Context()))
+	old, err := c.Snapshot(t.Context())
+	require.NoError(t, err)
+	f.Update(func(s *test.ChainState) {
+		s.Hash, s.Parent, s.Deactivation = common.HexToHash("0xdead"), common.Hash{}, 6
+	})
+	require.NoError(t, o.refresh(t.Context()))
+	_, err = c.IsActiveAt(t.Context(), o.recipient, old)
+	require.ErrorIs(t, err, ErrObservationUnavailable)
+	active, err := c.IsActive(t.Context(), o.recipient)
+	require.NoError(t, err)
+	require.False(t, active, "removed activation state is replaced from the canonical block")
+	*now = now.Add(30 * time.Second)
+	_, err = c.IsActive(t.Context(), o.recipient)
+	require.ErrorIs(t, err, ErrObservationUnavailable)
+}

internal/test/chainrpc.go

@@ -36,6 +36,8 @@ type ChainState struct {
 	ChainID                    uint64
 	Hash, Parent, RoundHash    common.Hash
 	Round, CurrentRound        uint64
+	Recipient                  common.Address
+	Activation, Deactivation   uint64
 	OmitL1, OmitTimestamp      bool
 	Fail                       string
 	Before                     func(context.Context, string)
@@ -56,7 +58,7 @@ type ChainRPCStats struct {
 func NewChainRPC(t testing.TB) *ChainRPC {
 	t.Helper()
 	c := &ChainRPC{Controller: common.HexToAddress("0xc001"), stats: ChainRPCStats{Calls: map[string]int{}}}
-	c.state = ChainState{Number: 0x100000, L1Block: 50, Timestamp: uint64(time.Now().Unix()), Hash: common.HexToHash("0xabcd"), RoundHash: common.HexToHash("0x1234"), Round: 5, CurrentRound: 5}
+	c.state = ChainState{Number: 0x100000, L1Block: 50, Timestamp: uint64(time.Now().Unix()), Hash: common.HexToHash("0xabcd"), RoundHash: common.HexToHash("0x1234"), Round: 5, CurrentRound: 5, Deactivation: ^uint64(0)}
 	word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value }
 	contractNames := map[string]string{}
 	c.state.Contracts = map[string]common.Address{}
@@ -153,7 +155,13 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 			case "RoundsManager.blockHashForRound":
 				result = state.RoundHash.Hex()
 			case "BondingManager.isActiveTranscoder":
-				result = "0x" + word("1")
+				if state.Recipient != (common.Address{}) {
+					require.Equal(t, word(hex.EncodeToString(state.Recipient.Bytes())), call.Input[10:])
+				}
+				result = "0x" + word("0")
+				if state.Activation <= state.CurrentRound && state.CurrentRound < state.Deactivation {
+					result = "0x" + word("1")
+				}
 			case "TicketBroker.getSenderInfo":
 				result = "0x" + word("de0b6b3a7640000") + word("0") + word("de0b6b3a7640000") + word("0")
 			case "TicketBroker.claimableReserve":

orchestrator/app.go

@@ -284,7 +284,6 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 		}
 		paymentChain = eth.PaymentChain{Contracts: contracts}
 		observedChain = paymentChain
-		observedChain.Observer = eth.NewPaymentObserver(paymentChain, p.Network == nodeconfig.Mainnet)
 		if p.ETHUSDFeed != nil {
 			rate, until, err := contracts.WeiPerUSD(parent, *p.ETHUSDFeed, p.PriceMaxAge)
 			if err != nil {
@@ -298,6 +297,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 		if p.OrchestratorAddress != nil {
 			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())
 		if err != nil {
 			return fmt.Errorf("initialize payment engine (check --account, --ticket-face-value and --ticket-win-prob): %w", err)

orchestrator/app_test.go

@@ -101,6 +101,9 @@ func TestPaidHTTPStartup(t *testing.T) {
 					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)
 					}

orchestrator/metrics.go

@@ -131,10 +131,11 @@ type paidReadiness struct {
 func (p *paidReadiness) ready(registry *Registry) bool {
 	p.mu.Lock()
 	limit := max(time.Minute, 3*p.interval)
-	ok := p.chain && p.active && p.storage && !p.completed.IsZero() && time.Since(p.completed) <= limit && !p.workerCompleted.IsZero() && time.Since(p.workerCompleted) <= limit
+	chainReady := p.chain && p.active
 	if p.observer != nil {
-		ok = ok && p.observer.Status().Healthy
+		chainReady = p.observer.Status().RecipientActive
 	}
+	ok := chainReady && p.storage && !p.completed.IsZero() && time.Since(p.completed) <= limit && !p.workerCompleted.IsZero() && time.Since(p.workerCompleted) <= limit
 	p.mu.Unlock()
 	registry.mu.Lock()
 	fresh := registry.rateUntil.IsZero() || time.Now().Before(registry.rateUntil)

orchestrator/metrics_test.go

@@ -40,7 +40,7 @@ func TestPaidReadinessUsesCurrentPublication(t *testing.T) {
 	defer rpc.Close()
 	contracts, err := eth.NewContracts(rpc, f.Controller)
 	require.NoError(t, err)
-	o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true)
+	o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true, common.HexToAddress("0x1234"))
 	ctx, cancel := context.WithCancel(t.Context())
 	done := make(chan struct{})
 	go func() { o.Run(ctx); close(done) }()

orchestrator/payment_recipient_test.go

@@ -471,3 +471,65 @@ func TestLiveChargeObservesOnlyDebitAttempts(t *testing.T) {
 	require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "charge", "outcome": "success"}))
 	require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "charge", "outcome": "insufficient_balance"}))
 }
+
+// Warm admission and refusal cases adapt Mike Zupper's PR #1
+// orchestrator/chainview_test.go at bbe126e5eddbf59f9cc9f94d3c37703215c26850.
+// 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)
+	registry := NewRegistry("", "https://orch.example", time.Second, time.Minute)
+	ready := &paidReadiness{storage: true, completed: time.Now(), workerCompleted: time.Now(), observer: chain.Observer}
+	f.Reset()
+	info := issueRecipientParams(t, e, key)
+	require.NoError(t, receiveRecipientPayment(t, e, key, info, 1))
+	require.True(t, ready.ready(registry), "current local activity overrides an older unavailable dependency check")
+	require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "warm challenge, receipt, and readiness activity must be local")
+	before, err := e.Balance("session")
+	require.NoError(t, err)
+	beforeNonces := e.nonceCount
+	ready.chain, ready.active = true, true // Simulate the preceding periodic check.
+	f.Update(func(s *test.ChainState) {
+		s.Parent, s.Hash = s.Hash, ethcommon.HexToHash("0xfeed")
+		s.Number++
+		s.L1Block++
+		s.Timestamp = uint64(time.Now().Unix())
+		s.Fail = "BondingManager.isActiveTranscoder"
+	})
+	require.Eventually(t, func() bool {
+		snapshot, err := chain.Snapshot(t.Context())
+		return err == nil && snapshot.Block.Int64() == 51
+	}, 7*time.Second, 10*time.Millisecond)
+	f.Reset()
+	require.False(t, ready.ready(registry), "readiness must refuse between periodic checks")
+	require.Error(t, receiveRecipientPayment(t, e, key, info, 2))
+	after, err := e.Balance("session")
+	require.NoError(t, err)
+	require.Equal(t, before.RatString(), after.RatString())
+	require.Equal(t, beforeNonces, e.nonceCount, "unavailable activity must not consume replay guards")
+	require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "unknown state must refuse without a request-triggered retry")
+}