diff --git a/eth/contracts_test.go b/eth/contracts_test.go index 300f977..21acec7 100644 --- a/eth/contracts_test.go +++ b/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) } diff --git a/eth/payment.go b/eth/payment.go index 8fb2f07..8d55615 100644 --- a/eth/payment.go +++ b/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 } diff --git a/eth/payment_contracts.go b/eth/payment_contracts.go new file mode 100644 index 0000000..62ab8a4 --- /dev/null +++ b/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") + } +} diff --git a/eth/payment_contracts_test.go b/eth/payment_contracts_test.go new file mode 100644 index 0000000..218afca --- /dev/null +++ b/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"]) +} diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go index 1738f27..91471ee 100644 --- a/internal/test/chainrpc.go +++ b/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 {