Retain confirmed payment contract bindings

eth/contracts_test.go

@@ -77,4 +77,10 @@ func TestRedemptionSubmission(t *testing.T) {
 	require.NoError(t, err)
 	require.NotEqual(t, common.Hash{}, hash)
 	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)}
+	hash, err = chain.Redeem(t.Context(), key, big.NewInt(1), ticket)
+	require.NoError(t, err)
+	require.Equal(t, sentHash, hash)
 }

eth/payment.go

@@ -36,7 +36,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.ResolveAt(ctx, snapshot.blockReference, "bondingManager")
+	address, err := c.Contracts.paymentAddress(ctx, snapshot, "bondingManager")
 	if err != nil {
 		return false, err
 	}
@@ -86,6 +86,7 @@ type ChainSnapshot struct {
 	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
@@ -117,11 +118,17 @@ func (c PaymentChain) snapshotAt(ctx context.Context, header *miniHeader, curren
 	block := new(big.Int).Set((*big.Int)(clock))
 	// Bind every read to the canonical block, including Controller resolution.
 	ref := map[string]any{"blockHash": header.Hash.Hex(), "requireCanonical": true}
-	address, err := c.Contracts.ResolveAt(ctx, ref, "roundsManager")
+	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
 	}
-	rounds, err := contracts.NewRoundsManagerCaller(address, c.Contracts.callerAt(ref))
+	rounds, err := contracts.NewRoundsManagerCaller(addresses.rounds, c.Contracts.callerAt(ref))
 	if err != nil {
 		return ChainSnapshot{}, err
 	}
@@ -140,7 +147,7 @@ func (c PaymentChain) snapshotAt(ctx context.Context, header *miniHeader, curren
 	if hash == ([32]byte{}) {
 		return ChainSnapshot{}, errors.New("invalid round block hash")
 	}
-	snapshot := ChainSnapshot{Block: block, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref}
+	snapshot := ChainSnapshot{Block: block, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref, addresses: addresses}
 	if currentRound {
 		snapshot.CurrentRound, err = rounds.CurrentRound(opts)
 		if err != nil {
@@ -169,7 +176,7 @@ func (c PaymentChain) SenderInfo(ctx context.Context, sender, recipient ethcommo
 	if err != nil {
 		return SenderInfo{}, err
 	}
-	address, err := c.Contracts.ResolveAt(ctx, snapshot.blockReference, "ticketBroker")
+	address, err := c.Contracts.paymentAddress(ctx, snapshot, "ticketBroker")
 	if err != nil {
 		return SenderInfo{}, err
 	}

eth/payment_contracts.go

@@ -0,0 +1,48 @@
+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")
+	}
+}

eth/payment_contracts_test.go

@@ -0,0 +1,137 @@
+package eth
+
+import (
+	"context"
+	"math/big"
+	"sync/atomic"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/livepeer/node/internal/test"
+	"github.com/stretchr/testify/require"
+)
+
+// Registry replacement cases adapt Mike Zupper's BlueClawNetwork/livepeer-node
+// eth/chainview_test.go, 7e3a51e7076aef558cffcbadd01149b727618005 (MIT; see LICENSE).
+func TestObserverConfirmedPaymentContracts(t *testing.T) {
+	for _, tc := range []struct {
+		contract string
+		round    uint64
+	}{{"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)
+			require.NoError(t, o.refresh(t.Context()))
+			before, err := o.Snapshot(t.Context())
+			require.NoError(t, err)
+			old := f.State().Contracts[tc.contract]
+			moved := common.HexToAddress("0xb1")
+			advanceObserver(f, *now)
+			f.Update(func(s *test.ChainState) {
+				s.Contracts[tc.contract] = moved
+				s.Round, s.CurrentRound = tc.round, tc.round+1
+				s.Results = map[test.ChainCall]string{}
+				switch tc.contract {
+				case "RoundsManager":
+					s.RoundHash = common.HexToHash("0x9876")
+					s.Results[test.ChainCall{To: old, Function: "RoundsManager.lastInitializedRound"}] = "0x" + word(5)
+					s.Results[test.ChainCall{To: old, Function: "RoundsManager.currentRound"}] = "0x" + word(5)
+					s.Results[test.ChainCall{To: old, Function: "RoundsManager.blockHashForRound"}] = before.RoundHash.Hex()
+				case "TicketBroker":
+					s.Results[test.ChainCall{To: moved, Function: "TicketBroker.getSenderInfo"}] = "0x" + word(7) + word(0) + word(8) + word(0)
+					s.Results[test.ChainCall{To: moved, Function: "TicketBroker.claimableReserve"}] = "0x" + word(6)
+				case "BondingManager":
+					s.Results[test.ChainCall{To: moved, Function: "BondingManager.isActiveTranscoder"}] = "0x" + word(0)
+				}
+			})
+			f.Reset()
+			require.NoError(t, o.refresh(t.Context()))
+			require.Equal(t, 3, f.Stats().Calls["Controller.getContract"])
+			after, err := o.Snapshot(t.Context())
+			require.NoError(t, err)
+			require.NotEqual(t, before.addresses, after.addresses)
+			require.EqualValues(t, tc.round, after.Round.Uint64())
+			require.EqualValues(t, tc.round+1, after.CurrentRound.Uint64(), "protocol round can advance before initialization")
+			require.Equal(t, f.State().RoundHash, after.RoundHash)
+			chain := o.chain
+			chain.Observer = o
+			f.Reset()
+			for range 3 {
+				funds, err := chain.SenderInfo(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)
+				require.NoError(t, err)
+				require.Equal(t, tc.contract != "BondingManager", active)
+				if tc.contract == "TicketBroker" {
+					require.Equal(t, big.NewInt(7), funds.Deposit)
+					require.Equal(t, big.NewInt(6), funds.Reserve)
+				}
+			}
+			require.Zero(t, f.Stats().Calls["Controller.getContract"], "warm payment reads retain confirmed addresses")
+		})
+	}
+}
+
+func TestObserverRejectsIncompleteRegistry(t *testing.T) {
+	for _, failure := range []string{"RPC failure", "zero address", "malformed address", "reorg"} {
+		t.Run(failure, func(t *testing.T) {
+			f, o, now := newTestObserver(t)
+			require.NoError(t, o.refresh(t.Context()))
+			before, err := o.Snapshot(t.Context())
+			require.NoError(t, err)
+			advanceObserver(f, *now)
+			var calls atomic.Int32
+			f.Update(func(s *test.ChainState) {
+				switch failure {
+				case "RPC failure":
+					s.Fail = "Controller.getContract"
+				case "zero address":
+					s.Contracts["TicketBroker"] = common.Address{}
+				case "malformed address":
+					s.Results = map[test.ChainCall]string{{To: f.Controller, Function: "Controller.getContract"}: "0x01"}
+				case "reorg":
+					s.Before = func(_ context.Context, name string) {
+						if name == "Controller.getContract" && calls.Add(1) == 3 {
+							f.Update(func(s *test.ChainState) { s.Hash = common.HexToHash("0xdead") })
+						}
+					}
+				}
+			})
+			f.Reset()
+			require.Error(t, o.refresh(t.Context()))
+			after, err := o.Snapshot(t.Context())
+			require.NoError(t, err)
+			require.Equal(t, before, after, "a failed registry candidate cannot publish any state or renew age")
+			require.Zero(t, f.Stats().Calls["RoundsManager.lastInitializedRound"], "confirm the entire registry before dependent reads")
+			f.Update(func(s *test.ChainState) {
+				s.Fail, s.Results, s.Before = "", nil, nil
+				s.Contracts["TicketBroker"] = common.HexToAddress("0xa3")
+			})
+			require.NoError(t, o.refresh(t.Context()), "a later complete candidate recovers")
+		})
+	}
+}
+
+func TestObserverRegistryScopeAndFinalizedReads(t *testing.T) {
+	f, o, now := newTestObserver(t)
+	require.NoError(t, o.refresh(t.Context()))
+	s, err := o.Snapshot(t.Context())
+	require.NoError(t, err)
+	other, err := NewContracts(o.chain.Contracts.RPC, common.HexToAddress("0xc002"))
+	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.Zero(t, f.Stats().RoundTrips)
+	advanceObserver(f, *now)
+	f.Update(func(state *test.ChainState) {
+		state.Contracts["TicketBroker"] = common.HexToAddress("0xb3")
+		state.Results = map[test.ChainCall]string{{To: s.addresses.broker, Function: "TicketBroker.usedTickets"}: "0x" + word(1)}
+	})
+	chain := o.chain
+	chain.Observer = o // Its retained registry is deliberately still the old one.
+	used, err := chain.IsUsedTicket(t.Context(), common.HexToHash("0x9999"), "finalized")
+	require.NoError(t, err)
+	require.False(t, used, "finalized settlement resolves its own canonical registry")
+	require.Equal(t, 1, f.Stats().Calls["Controller.getContract"])
+}

internal/test/chainrpc.go

@@ -39,6 +39,13 @@ type ChainState struct {
 	OmitL1, OmitTimestamp      bool
 	Fail                       string
 	Before                     func(context.Context, string)
+	Contracts                  map[string]common.Address
+	Results                    map[ChainCall]string
+}
+
+type ChainCall struct {
+	To       common.Address
+	Function string
 }
 
 type ChainRPCStats struct {
@@ -50,18 +57,21 @@ 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.history = map[common.Hash]ChainState{c.state.Hash: c.state}
 	word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value }
-	addresses := map[string]string{}
+	contractNames := map[string]string{}
+	c.state.Contracts = map[string]common.Address{}
 	for name, address := range map[string]string{"RoundsManager": "a1", "BondingManager": "a2", "TicketBroker": "a3"} {
-		addresses[hex.EncodeToString(crypto.Keccak256([]byte(name)))] = word(address)
+		contractNames[hex.EncodeToString(crypto.Keccak256([]byte(name)))] = name
+		c.state.Contracts[name] = common.HexToAddress(address)
 	}
+	c.history = map[common.Hash]ChainState{c.state.Hash: c.state}
 	methods := map[string]string{}
 	for signature, name := range map[string]string{
 		"getContract(bytes32)": "Controller.getContract", "lastInitializedRound()": "RoundsManager.lastInitializedRound",
 		"currentRound()":             "RoundsManager.currentRound",
 		"blockHashForRound(uint256)": "RoundsManager.blockHashForRound", "isActiveTranscoder(address)": "BondingManager.isActiveTranscoder",
 		"getSenderInfo(address)": "TicketBroker.getSenderInfo", "claimableReserve(address,address)": "TicketBroker.claimableReserve",
+		"usedTickets(bytes32)": "TicketBroker.usedTickets",
 	} {
 		methods[hex.EncodeToString(crypto.Keccak256([]byte(signature))[:4])] = name
 	}
@@ -73,7 +83,10 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 		}
 		require.NoError(t, json.NewDecoder(r.Body).Decode(&request))
 		name := request.Method
-		var call struct{ Input string }
+		var call struct {
+			Input string
+			To    common.Address
+		}
 		if name == "eth_call" {
 			require.Len(t, request.Params, 2)
 			require.NoError(t, json.Unmarshal(request.Params[0], &call))
@@ -103,7 +116,7 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 		case "eth_getBlockByNumber":
 			var ref string
 			require.NoError(t, json.Unmarshal(request.Params[0], &ref))
-			if ref != "latest" {
+			if ref != "latest" && ref != "finalized" {
 				var ok bool
 				state, ok = c.stateAt(ref)
 				if !ok {
@@ -130,8 +143,9 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 			}
 			switch name {
 			case "Controller.getContract":
-				require.Contains(t, addresses, call.Input[10:])
-				result = "0x" + addresses[call.Input[10:]]
+				require.Equal(t, c.Controller, call.To)
+				require.Contains(t, contractNames, call.Input[10:])
+				result = "0x" + word(state.Contracts[contractNames[call.Input[10:]]].Hex()[2:])
 			case "RoundsManager.lastInitializedRound":
 				result = "0x" + word(quantity(state.Round)[2:])
 			case "RoundsManager.currentRound":
@@ -144,9 +158,14 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 				result = "0x" + word("de0b6b3a7640000") + word("0") + word("de0b6b3a7640000") + word("0")
 			case "TicketBroker.claimableReserve":
 				result = "0x" + word("de0b6b3a7640000")
+			case "TicketBroker.usedTickets":
+				result = "0x" + word("0")
 			default:
 				t.Errorf("unexpected payment contract call")
 			}
+			if override, ok := state.Results[ChainCall{call.To, name}]; ok {
+				result = override
+			}
 		default:
 			t.Errorf("unexpected payment RPC method %q", request.Method)
 		}
@@ -160,12 +179,16 @@ func NewChainRPC(t testing.TB) *ChainRPC {
 func (c *ChainRPC) State() ChainState {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	return c.state
+	s := c.state
+	s.Contracts, s.Results = maps.Clone(s.Contracts), maps.Clone(s.Results)
+	return s
 }
 
 func (c *ChainRPC) Update(update func(*ChainState)) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	c.state.Contracts = maps.Clone(c.state.Contracts)
+	c.state.Results = maps.Clone(c.state.Results)
 	update(&c.state)
 	for h, state := range c.history {
 		if state.Number >= c.state.Number {