From 51210cde45dbe4e51fe4a8108413a1281dd3a2d9 Mon Sep 17 00:00:00 2001 From: Mike Zupper Date: Sat, 31 Oct 2026 09:13:03 -0400 Subject: [PATCH 1/2] Observe payer funds once per signer request with counting RPC coverage Add a counting RPC fixture and paid-request baseline, separate the signer's funds observation from policy checks, and reuse that observation for the batch recheck. Squash Mike Zupper's three original commits. Source-Repository: https://github.com/BlueClawNetwork/livepeer-node Source-Pull-Request: https://github.com/BlueClawNetwork/livepeer-node/pull/1 Original-Commit: 285eda984d7cce4e3030ef0510d63ae142534dd7 Original-Commit: f3f1e4bb416a3116d8cfba48a7bc555b87d6af7a Original-Commit: b8cad3d341a030e873963f5b4b6f7dbc0f5becfe --- internal/test/chainrpc.go | 466 ++++++++++++++++++++++++++++++++++++++ signer/payment.go | 35 +-- signer/payment_test.go | 174 ++++++++++++++ signer/service_test.go | 43 +++- 4 files changed, 704 insertions(+), 14 deletions(-) create mode 100644 internal/test/chainrpc.go diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go new file mode 100644 index 0000000..19632b6 --- /dev/null +++ b/internal/test/chainrpc.go @@ -0,0 +1,466 @@ +package test + +import ( + "encoding/hex" + "encoding/json" + "fmt" + "maps" + "math/big" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" +) + +// BatchMode selects how ChainRPC answers a JSON-RPC batch, covering the ways +// providers and load balancers refuse them. +type BatchMode int + +const ( + BatchAccepted BatchMode = iota + BatchRefusedHTTP // HTTP 400 for any JSON array. + BatchRefusedObject // HTTP 200 with one error object instead of an array. + BatchRefusedEach // An array with -32601 for every element. + BatchTruncated // An array holding only the first response. +) + +// ChainSender is a TicketBroker sender. A nil Claimable entry uses Reserve. +type ChainSender struct { + Deposit, Reserve, WithdrawRound *big.Int + Claimable map[common.Address]*big.Int +} + +func (s ChainSender) clone() ChainSender { + s.Claimable = maps.Clone(s.Claimable) + return s +} + +type chainRounds struct { + Last int64 + Hashes map[int64]common.Hash +} + +// ChainState is the contract state of one block. State belongs to a contract +// address, so a retired contract keeps answering with what it last held. +type ChainState struct { + Number, L1Block int64 // A zero L1Block omits l1BlockNumber, as on Ethereum. + Contracts map[string]common.Address + rounds map[common.Address]chainRounds + active map[common.Address]map[common.Address]bool + senders map[common.Address]map[common.Address]ChainSender +} + +func (s *ChainState) clone() *ChainState { + next := &ChainState{Number: s.Number, L1Block: s.L1Block, Contracts: maps.Clone(s.Contracts), rounds: map[common.Address]chainRounds{}, active: map[common.Address]map[common.Address]bool{}, senders: map[common.Address]map[common.Address]ChainSender{}} + for address, rounds := range s.rounds { + next.rounds[address] = chainRounds{rounds.Last, maps.Clone(rounds.Hashes)} + } + for address, active := range s.active { + next.active[address] = maps.Clone(active) + } + for address, senders := range s.senders { + next.senders[address] = maps.Clone(senders) + } + return next +} + +// SetRound initializes a round on the current RoundsManager. +func (s *ChainState) SetRound(round int64, hash common.Hash) { + address := s.Contracts["RoundsManager"] + rounds := s.rounds[address] + rounds.Last, rounds.Hashes[round] = round, hash + s.rounds[address] = rounds +} + +// SetActive sets recipient status on the current BondingManager. +func (s *ChainState) SetActive(recipient common.Address, active bool) { + s.active[s.Contracts["BondingManager"]][recipient] = active +} + +// SetSender sets a payer's funds on the current TicketBroker. +func (s *ChainState) SetSender(payer common.Address, sender ChainSender) { + s.senders[s.Contracts["TicketBroker"]][payer] = sender.clone() +} + +// Move deploys a contract at a new address, starting from its current state. +// The old address keeps answering with the state it held. +func (s *ChainState) Move(contract string, address common.Address) { + previous := s.Contracts[contract] + switch contract { + case "RoundsManager": + s.rounds[address] = chainRounds{s.rounds[previous].Last, maps.Clone(s.rounds[previous].Hashes)} + case "BondingManager": + s.active[address] = maps.Clone(s.active[previous]) + case "TicketBroker": + s.senders[address] = maps.Clone(s.senders[previous]) + } + s.Contracts[contract] = address +} + +// ChainRPC is a counting Ethereum JSON-RPC HTTP server for the payment reads. +// Reads pinned to a block hash see that block's state and fail once the block +// is no longer canonical, so a test can observe mixed-block reads directly. +type ChainRPC struct { + URL string + Controller common.Address + mu sync.Mutex + blocks map[common.Hash]*ChainState + head common.Hash + orphaned map[common.Hash]bool + nextHash uint64 + latency time.Duration + down bool + batchMode BatchMode + failCall func(function string, input []byte) bool + trips int + batches int + calls map[string]int + unpinned int + batchSizes []int +} + +const chainRPCController = "0x00000000000000000000000000000000000c0001" + +var chainSelectors = func() map[string]string { + selectors := map[string]string{} + for name, signature := range map[string]string{ + "Controller.getContract": "getContract(bytes32)", + "RoundsManager.lastInitializedRound": "lastInitializedRound()", + "RoundsManager.blockHashForRound": "blockHashForRound(uint256)", + "BondingManager.isActiveTranscoder": "isActiveTranscoder(address)", + "TicketBroker.getSenderInfo": "getSenderInfo(address)", + "TicketBroker.claimableReserve": "claimableReserve(address,address)", + } { + selectors[hex.EncodeToString(crypto.Keccak256([]byte(signature))[:4])] = name + } + return selectors +}() + +// ChainRoundHash is the deterministic initial block hash of a round. +func ChainRoundHash(round int64) common.Hash { + return crypto.Keccak256Hash([]byte("round"), big.NewInt(round).Bytes()) +} + +// NewChainRPC starts a chain at round 5 with its contracts registered. +func NewChainRPC(t testing.TB) *ChainRPC { + t.Helper() + state := &ChainState{Number: 0x100000, L1Block: 1000, Contracts: map[string]common.Address{ + "RoundsManager": common.HexToAddress("0xa1"), "BondingManager": common.HexToAddress("0xa2"), "TicketBroker": common.HexToAddress("0xa3"), + }, rounds: map[common.Address]chainRounds{}, active: map[common.Address]map[common.Address]bool{}, senders: map[common.Address]map[common.Address]ChainSender{}} + state.rounds[state.Contracts["RoundsManager"]] = chainRounds{Hashes: map[int64]common.Hash{}} + state.active[state.Contracts["BondingManager"]] = map[common.Address]bool{} + state.senders[state.Contracts["TicketBroker"]] = map[common.Address]ChainSender{} + state.SetRound(5, ChainRoundHash(5)) + c := &ChainRPC{Controller: common.HexToAddress(chainRPCController), blocks: map[common.Hash]*ChainState{}, orphaned: map[common.Hash]bool{}, calls: map[string]int{}} + c.head = c.addBlock(state) + server := httptest.NewServer(http.HandlerFunc(c.serve)) + t.Cleanup(server.Close) + c.URL = server.URL + return c +} + +// addBlock stores a state under a new unique hash. Caller holds c.mu, or the +// server is not yet running. +func (c *ChainRPC) addBlock(state *ChainState) common.Hash { + c.nextHash++ + hash := crypto.Keccak256Hash([]byte("block"), new(big.Int).SetUint64(c.nextHash).Bytes()) + c.blocks[hash] = state + return hash +} + +// Mine appends a block whose state is the head's state after mutate. +func (c *ChainRPC) Mine(mutate func(*ChainState)) common.Hash { + c.mu.Lock() + defer c.mu.Unlock() + state := c.blocks[c.head].clone() + state.Number++ + if mutate != nil { + mutate(state) + } + c.head = c.addBlock(state) + return c.head +} + +// Reorg replaces the head with a sibling at the same height. The old head stops +// answering pinned reads, as requireCanonical demands. +func (c *ChainRPC) Reorg(mutate func(*ChainState)) common.Hash { + c.mu.Lock() + defer c.mu.Unlock() + state := c.blocks[c.head].clone() + if mutate != nil { + mutate(state) + } + c.orphaned[c.head] = true + c.head = c.addBlock(state) + return c.head +} + +// Rewind makes an earlier block the head without orphaning anything, as a load +// balancer does when a request lands on a lagging backend. +func (c *ChainRPC) Rewind(block common.Hash) { + c.mu.Lock() + defer c.mu.Unlock() + c.head = block +} + +// Head returns the current head hash. +func (c *ChainRPC) Head() common.Hash { + c.mu.Lock() + defer c.mu.Unlock() + return c.head +} + +// SetLatency delays every HTTP round trip. +func (c *ChainRPC) SetLatency(latency time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.latency = latency +} + +// SetDown makes every request fail with HTTP 503. +func (c *ChainRPC) SetDown(down bool) { + c.mu.Lock() + defer c.mu.Unlock() + c.down = down +} + +// SetBatchMode selects how batches are answered. +func (c *ChainRPC) SetBatchMode(mode BatchMode) { + c.mu.Lock() + defer c.mu.Unlock() + c.batchMode = mode +} + +// FailCalls makes matching eth_calls return an execution error. A nil filter +// clears it. +func (c *ChainRPC) FailCalls(filter func(function string, input []byte) bool) { + c.mu.Lock() + defer c.mu.Unlock() + c.failCall = filter +} + +// ChainRPCStats counts traffic since the last Reset. RoundTrips are HTTP +// requests, so a batch is one; Calls counts each JSON-RPC call inside them, by +// method or Contract.function. +type ChainRPCStats struct { + RoundTrips, Batches, Unpinned int + Calls map[string]int + BatchSizes []int +} + +// Stats returns a copy of the current counts. +func (c *ChainRPC) Stats() ChainRPCStats { + c.mu.Lock() + defer c.mu.Unlock() + return ChainRPCStats{RoundTrips: c.trips, Batches: c.batches, Unpinned: c.unpinned, Calls: maps.Clone(c.calls), BatchSizes: append([]int(nil), c.batchSizes...)} +} + +// Reset clears the counts. +func (c *ChainRPC) Reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.trips, c.batches, c.unpinned, c.calls, c.batchSizes = 0, 0, 0, map[string]int{}, nil +} + +type chainRPCRequest struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` +} + +func (c *ChainRPC) serve(w http.ResponseWriter, r *http.Request) { + var body json.RawMessage + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + c.mu.Lock() + latency, down, mode := c.latency, c.down, c.batchMode + c.trips++ + c.mu.Unlock() + time.Sleep(latency) + if down { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + if len(body) > 0 && body[0] == '[' { + var requests []chainRPCRequest + if err := json.Unmarshal(body, &requests); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + c.mu.Lock() + c.batches++ + c.batchSizes = append(c.batchSizes, len(requests)) + c.mu.Unlock() + switch mode { + case BatchRefusedHTTP: + http.Error(w, "batch requests are not supported", http.StatusBadRequest) + return + case BatchRefusedObject: + _ = json.NewEncoder(w).Encode(chainRPCError(nil, -32600, "batch requests are not supported")) + return + case BatchRefusedEach: + responses := make([]map[string]any, len(requests)) + for i, request := range requests { + responses[i] = chainRPCError(request.ID, -32601, "batch requests are not supported") + } + _ = json.NewEncoder(w).Encode(responses) + return + case BatchTruncated: + requests = requests[:1] + } + responses := make([]map[string]any, len(requests)) + for i, request := range requests { + responses[i] = c.answer(request) + } + _ = json.NewEncoder(w).Encode(responses) + return + } + var request chainRPCRequest + if err := json.Unmarshal(body, &request); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + _ = json.NewEncoder(w).Encode(c.answer(request)) +} + +func chainRPCError(id json.RawMessage, code int, message string) map[string]any { + return map[string]any{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": code, "message": message}} +} + +func (c *ChainRPC) answer(request chainRPCRequest) map[string]any { + c.mu.Lock() + defer c.mu.Unlock() + result := func(value any) map[string]any { + return map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": value} + } + switch request.Method { + case "eth_chainId": + c.calls[request.Method]++ + return result("0x1") + case "eth_getBalance": + c.calls[request.Method]++ + return result("0x1") + case "eth_getBlockByNumber": + c.calls[request.Method]++ + state := c.blocks[c.head] + header := map[string]string{"number": "0x" + strconv.FormatInt(state.Number, 16), "hash": c.head.Hex()} + if state.L1Block != 0 { + header["l1BlockNumber"] = "0x" + strconv.FormatInt(state.L1Block, 16) + } + return result(header) + case "eth_call": + return c.call(request) + } + c.calls[request.Method]++ + return chainRPCError(request.ID, -32601, "unexpected method "+request.Method) +} + +// call answers one eth_call. Caller holds c.mu. +func (c *ChainRPC) call(request chainRPCRequest) map[string]any { + var args struct { + To common.Address `json:"to"` + Input string `json:"input"` + } + var pin struct { + BlockHash common.Hash `json:"blockHash"` + RequireCanonical bool `json:"requireCanonical"` + } + input := []byte(nil) + if len(request.Params) > 0 && json.Unmarshal(request.Params[0], &args) == nil { + input = common.FromHex(args.Input) + } + function := "unknown" + if len(input) >= 4 { + if name, ok := chainSelectors[hex.EncodeToString(input[:4])]; ok { + function = name + } + } + c.calls[function]++ + block := c.head + if len(request.Params) > 1 && json.Unmarshal(request.Params[1], &pin) == nil && pin.BlockHash != (common.Hash{}) && pin.RequireCanonical { + block = pin.BlockHash + } else { + c.unpinned++ + } + state := c.blocks[block] + if state == nil { + return chainRPCError(request.ID, -32000, "header not found") + } + if c.orphaned[block] { + return chainRPCError(request.ID, -32000, "block hash is not currently canonical") + } + if c.failCall != nil && c.failCall(function, input) { + return chainRPCError(request.ID, 3, "execution reverted") + } + output := state.execute(function, args.To, input) + return map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": "0x" + output} +} + +func chainWord(value *big.Int) string { + if value == nil { + value = new(big.Int) + } + return fmt.Sprintf("%064x", value) +} + +func chainArgAddress(input []byte, index int) common.Address { + offset := 4 + 32*index + if len(input) < offset+32 { + return common.Address{} + } + return common.BytesToAddress(input[offset : offset+32]) +} + +// execute returns hex output, empty when the address holds no such contract. +func (s *ChainState) execute(function string, to common.Address, input []byte) string { + switch function { + case "Controller.getContract": + if len(input) < 36 { + return "" + } + for name, address := range s.Contracts { + if crypto.Keccak256Hash([]byte(name)) == common.BytesToHash(input[4:36]) { + return chainWord(new(big.Int).SetBytes(address.Bytes())) + } + } + return chainWord(nil) + case "RoundsManager.lastInitializedRound": + if rounds, ok := s.rounds[to]; ok { + return chainWord(big.NewInt(rounds.Last)) + } + case "RoundsManager.blockHashForRound": + if rounds, ok := s.rounds[to]; ok && len(input) >= 36 { + return strings.TrimPrefix(rounds.Hashes[new(big.Int).SetBytes(input[4:36]).Int64()].Hex(), "0x") + } + case "BondingManager.isActiveTranscoder": + if active, ok := s.active[to]; ok { + if active[chainArgAddress(input, 0)] { + return chainWord(big.NewInt(1)) + } + return chainWord(nil) + } + case "TicketBroker.getSenderInfo": + if senders, ok := s.senders[to]; ok { + sender := senders[chainArgAddress(input, 0)] + return chainWord(sender.Deposit) + chainWord(sender.WithdrawRound) + chainWord(sender.Reserve) + chainWord(nil) + } + case "TicketBroker.claimableReserve": + if senders, ok := s.senders[to]; ok { + sender := senders[chainArgAddress(input, 0)] + if claimable, ok := sender.Claimable[chainArgAddress(input, 1)]; ok { + return chainWord(claimable) + } + return chainWord(sender.Reserve) + } + } + return "" +} diff --git a/signer/payment.go b/signer/payment.go index f3c55b0..cf33d0a 100644 --- a/signer/payment.go +++ b/signer/payment.go @@ -100,7 +100,11 @@ func (s *Service) preparePayment(ctx context.Context, paymentType, manifest stri } // Precheck one ticket before generating the batch. // Observation/funding failures here are client errors. - if err := s.validateTicketParams(ctx, ¶ms, 1, "precheck"); err != nil { + funds, err := s.observePayer(ctx, ¶ms, "precheck") + if err == nil { + err = s.checkTicketParams(ctx, ¶ms, 1, funds, "precheck") + } + if err != nil { if errors.Is(err, pm.ErrRefreshRequired) { return draftPayment{}, err } @@ -117,9 +121,10 @@ func (s *Service) preparePayment(ctx context.Context, paymentType, manifest stri slog.With("component", "signer", "event", "payment", "operation", "payment").WarnContext(ctx, "signer ticket batch rejected", "outcome", "error", "error_kind", observability.ErrorKind(err)) return draftPayment{}, invalid("invalid ticket batch") } - // Recheck the full batch before authorization. A later observation/funding failure - // is an internal generation failure; invalid exposure remains a client error. - if err := s.validateTicketParams(ctx, ¶ms, count, "generation"); err != nil { + // Recheck the full batch before authorization, on the precheck's observation: + // only the ticket count differs. A funding failure here is an internal generation + // failure; invalid exposure remains a client error. + if err := s.checkTicketParams(ctx, ¶ms, count, funds, "generation"); err != nil { return draftPayment{}, err } batch, remaining, err := pm.DraftRemoteBatch(params, s.key.Address(), state.TicketNonce, fee, balance) @@ -178,30 +183,34 @@ func (p draftPayment) Sign(ctx context.Context, signer pm.TicketSigner, metrics return signedPayment{Payment: base64.StdEncoding.EncodeToString(wire.EncodePayment(message)), SegCreds: base64.StdEncoding.EncodeToString(wire.EncodeSegData(segment))}, nil } -// Adapted from go-livepeer/pm/sender.go's validateTicketParams, by Yondon Fu, -// Nico Vergauwen and Rafał Leszko. Supplied creation-round metadata is retained; -// refresh is governed by parameter expiry on the L1 clock, not the current round. -func (s *Service) validateTicketParams(ctx context.Context, ticketParams *pm.TicketParams, numTickets int, phase string) error { +func (s *Service) observePayer(ctx context.Context, ticketParams *pm.TicketParams, phase string) (pm.PayerFunds, error) { if ticketParams == nil { - return invalid("ticketParams is nil") + return pm.PayerFunds{}, invalid("ticketParams is nil") } if ticketParams.ExpirationBlock.Int64() == 0 { - return invalid("ticketParams expiration block is 0") + return pm.PayerFunds{}, invalid("ticketParams expiration block is 0") } if s.paymentChain == nil { slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain is not configured", "phase", phase) - return errors.Join(pm.ErrPayerUnavailable, errPaymentService) + return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } funds, err := s.paymentChain.PayerFunds(ctx, s.key.Address(), ticketParams.Recipient) if err != nil { slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation failed", "phase", phase, "outcome", "error", "error_kind", observability.ErrorKind(err)) - return errors.Join(pm.ErrPayerUnavailable, errPaymentService) + return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } if funds.Snapshot.Block == nil { slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation has no L1 block", "phase", phase) - return errors.Join(pm.ErrPayerUnavailable, errPaymentService) + return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } s.observeFunds(funds, false) + return funds, nil +} + +// Adapted from go-livepeer/pm/sender.go's validateTicketParams, by Yondon Fu, +// Nico Vergauwen and Rafał Leszko. Supplied creation-round metadata is retained; +// refresh is governed by parameter expiry on the L1 clock, not the current round. +func (s *Service) checkTicketParams(ctx context.Context, ticketParams *pm.TicketParams, numTickets int, funds pm.PayerFunds, phase string) error { latestL1Block := funds.Snapshot.Block currentBuffer := new(big.Int).Sub(ticketParams.ExpirationBlock, latestL1Block).Int64() if currentBuffer <= paramsExpiryBuffer { diff --git a/signer/payment_test.go b/signer/payment_test.go index 01db9f0..c2e7beb 100644 --- a/signer/payment_test.go +++ b/signer/payment_test.go @@ -1,13 +1,19 @@ package signer import ( + "bytes" "context" "encoding/base64" "encoding/json" "math/big" + "net/http" + "net/http/httptest" "os" "path/filepath" + "slices" + "strconv" "sync" + "sync/atomic" "testing" "time" @@ -158,3 +164,171 @@ func TestRecipientReservesWinningLiabilityAcrossConcurrentSessions(t *testing.T) require.NoError(t, err) require.Equal(t, 2, count, "rejected batches must roll back tickets and credit") } + +// paidPath is one paid request end to end against a counting JSON-RPC server: +// the orchestrator's challenge, the signer's payment and the orchestrator's receipt. +type paidPath struct { + rpc *test.ChainRPC + engine *orchestrator.PaymentEngine + service *Service + payer *eth.Key + recipient ethcommon.Address + direct eth.PaymentChain + next atomic.Int64 +} + +func newPaidPath(t testing.TB) *paidPath { + t.Helper() + payer, recipient := paidPathKey(t), ethcommon.HexToAddress("0x1234") + rpcServer := test.NewChainRPC(t) + rpcServer.Mine(func(state *test.ChainState) { + state.SetActive(recipient, true) + state.SetSender(payer.Address(), test.ChainSender{Deposit: big.NewInt(1e18), Reserve: big.NewInt(1e18), WithdrawRound: new(big.Int)}) + }) + client, err := eth.OpenRPC(rpcServer.URL) + require.NoError(t, err) + t.Cleanup(client.Close) + contracts, err := eth.NewContracts(client, rpcServer.Controller) + require.NoError(t, err) + store, err := orchestrator.OpenRedeemerDB(filepath.Join(t.TempDir(), "payments.sqlite")) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + direct := eth.PaymentChain{Contracts: contracts} + maximum := new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1)) + // One ticket covers the price and almost never wins, keeping winner storage + // out of the measurements of chain reads. + engine, err := orchestrator.NewPaymentEngine(store, pm.EthereumChain{Client: direct}, recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) + require.NoError(t, err) + service := newService(payer) + t.Cleanup(func() { _ = service.Close() }) + service.SetPaymentChain(pm.EthereumChain{Client: direct}) + return &paidPath{rpc: rpcServer, engine: engine, service: service, payer: payer, recipient: recipient, direct: direct} +} + +func paidPathKey(t testing.TB) *eth.Key { + t.Helper() + keyFile, passwordPath := test.WriteKeystore(t, nil) + key, err := eth.OpenKeystoreFile(keyFile, passwordPath) + require.NoError(t, err) + return key +} + +func (p *paidPath) manifest() string { + return "session-" + strconv.FormatInt(p.next.Add(1), 10) +} + +func (p *paidPath) challenge(t testing.TB, ctx context.Context, manifest string) wire.OrchestratorInfo { + t.Helper() + _, err := p.engine.MakeChallenge(ctx, "runner", manifest, p.payer.Address(), 9, "fixed", "https://orch.example") + require.NoError(t, err) + info, err := p.engine.ChallengeInfo(manifest) + require.NoError(t, err) + return info +} + +func (p *paidPath) sign(t testing.TB, info wire.OrchestratorInfo) (payment, segCreds string) { + t.Helper() + body, err := json.Marshal(map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) + require.NoError(t, err) + w := httptest.NewRecorder() + p.service.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/generate-live-payment", bytes.NewReader(body))) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + var signed paymentResponse + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &signed)) + return signed.Payment, signed.SegCreds +} + +func (p *paidPath) receive(ctx context.Context, manifest, payment, segCreds string) error { + _, _, err := p.engine.Receive(ctx, "runner", manifest, payment, segCreds) + return err +} + +// The expected log is the one derived from the code in the issue this test pins. +func TestPaidRequestRPCCalls(t *testing.T) { + path := newPaidPath(t) + manifest := path.manifest() + snapshot := map[string]int{"eth_getBlockByNumber": 1, "Controller.getContract": 1, "RoundsManager.lastInitializedRound": 1, "RoundsManager.blockHashForRound": 1} + isActive := map[string]int{"Controller.getContract": 1, "BondingManager.isActiveTranscoder": 1} + funds := map[string]int{"Controller.getContract": 1, "TicketBroker.getSenderInfo": 1, "TicketBroker.claimableReserve": 1} + sum := func(parts ...map[string]int) map[string]int { + total := map[string]int{} + for _, part := range parts { + for name, count := range part { + total[name] += count + } + } + return total + } + + path.rpc.Reset() + info := path.challenge(t, t.Context(), manifest) + stats := path.rpc.Stats() + require.Equal(t, 6, stats.RoundTrips, "challenge: snapshot (4) + IsActiveAt (2)") + require.Equal(t, sum(snapshot, isActive), stats.Calls) + require.Zero(t, stats.Unpinned) + + path.rpc.Reset() + payment, segCreds := path.sign(t, info) + stats = path.rpc.Stats() + require.Equal(t, 7, stats.RoundTrips, "signer: payer funds (7), reused by the batch recheck") + require.Equal(t, sum(snapshot, funds), stats.Calls) + require.Zero(t, stats.Unpinned) + + path.rpc.Reset() + require.NoError(t, path.receive(t.Context(), manifest, payment, segCreds)) + stats = path.rpc.Stats() + require.Equal(t, 9, stats.RoundTrips, "receive: payer funds (7) + IsActiveAt (2)") + require.Equal(t, sum(snapshot, funds, isActive), stats.Calls) + require.Zero(t, stats.Unpinned) +} + +// paidRequestLatency is a measurement rather than an assertion: RPC latency is +// injected, so the result shows how the round trips of a request add up. +const paidRequestLatency = 110 * time.Millisecond + +func (p *paidPath) request(t testing.TB, ctx context.Context) { + t.Helper() + manifest := p.manifest() + info := p.challenge(t, ctx, manifest) + payment, segCreds := p.sign(t, info) + require.NoError(t, p.receive(ctx, manifest, payment, segCreds)) +} + +func benchmarkPaidRequests(b *testing.B, path *paidPath, workers int) { + path.rpc.SetLatency(paidRequestLatency) + path.rpc.Reset() + var ( + next atomic.Int64 + mu sync.Mutex + latencies []time.Duration + running sync.WaitGroup + ) + b.ResetTimer() + for range workers { + running.Go(func() { + for next.Add(1) <= int64(b.N) { + start := time.Now() + path.request(b, context.Background()) + elapsed := time.Since(start) + mu.Lock() + latencies = append(latencies, elapsed) + mu.Unlock() + } + }) + } + running.Wait() + b.StopTimer() + slices.Sort(latencies) + b.ReportMetric(float64(path.rpc.Stats().RoundTrips)/float64(b.N), "rpc_trips/op") + b.ReportMetric(float64(latencies[len(latencies)/2].Milliseconds()), "p50_ms") + b.ReportMetric(float64(latencies[len(latencies)*99/100].Milliseconds()), "p99_ms") +} + +// Run with: go test ./signer -run '^$' -bench '^BenchmarkPaidRequest$' -benchtime 64x +func BenchmarkPaidRequest(b *testing.B) { + for _, workers := range []int{1, 32} { + b.Run("direct/workers="+strconv.Itoa(workers), func(b *testing.B) { + benchmarkPaidRequests(b, newPaidPath(b), workers) + }) + } +} diff --git a/signer/service_test.go b/signer/service_test.go index 10c4e57..bb1127b 100644 --- a/signer/service_test.go +++ b/signer/service_test.go @@ -426,7 +426,7 @@ func TestPaymentFailureStatuses(t *testing.T) { for _, phase := range []struct { name string failOn, status int - }{{"precheck", 1, 400}, {"generation", 2, 500}} { + }{{"precheck", 1, 400}} { for _, cause := range []string{"deposit", "reserve", "withdrawal", "chain observation"} { t.Run(phase.name+"/"+cause, func(t *testing.T) { logs.Reset() @@ -465,6 +465,30 @@ func TestPaymentFailureStatuses(t *testing.T) { }) } } + // A request observes funds once, so the batch recheck cannot see a new funding + // failure over HTTP. Its handling stays distinct: not a client error, so 500. + for _, cause := range []string{"deposit", "reserve", "withdrawal"} { + t.Run("generation/"+cause, func(t *testing.T) { + logs.Reset() + funds, err := (fundedPayer{}).PayerFunds(t.Context(), s.key.Address(), ethcommon.Address{}) + require.NoError(t, err) + switch cause { + case "deposit": + funds.Deposit = new(big.Int) + case "reserve": + funds.Reserve = new(big.Int) + case "withdrawal": + funds.WithdrawRound = new(big.Int).Add(funds.Snapshot.Round, big.NewInt(1)) + } + params := pm.TicketParams{ExpirationBlock: new(big.Int).SetBytes(original.TicketParams.ExpirationBlock)} + err = s.checkTicketParams(t.Context(), ¶ms, 2, funds, "generation") + require.ErrorIs(t, err, pm.ErrPayerUnavailable) + _, clientError := errors.AsType[paymentFailure](err) + require.False(t, clientError, "generate answers 500 for anything but a paymentFailure") + require.Equal(t, "rejected", paymentOutcome(err)) + require.Contains(t, logs.String(), `"phase":"generation"`) + }) + } s.SetPaymentChain(fundedPayer{}) t.Run("batch exposure", func(t *testing.T) { info := original @@ -517,6 +541,23 @@ func TestPaymentFailureStatuses(t *testing.T) { } } +func TestBatchRecheckReusesPrecheckObservation(t *testing.T) { + s, info := testService(t) + defer s.Close() + calls := 0 + s.SetPaymentChain(payerChainFunc(func(ctx context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { + calls++ + funds, err := (fundedPayer{}).PayerFunds(ctx, payer, recipient) + if calls > 1 { + funds.Deposit = new(big.Int) + } + return funds, err + })) + response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) + require.Equal(t, http.StatusOK, response.Code, response.Body.String()) + require.Equal(t, 1, calls) +} + func TestCancelledPaymentOutcome(t *testing.T) { s, info := testService(t) defer s.Close() From fccf209dc491b438fcc49868eeae8d3b3f05bc07 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Tue, 3 Nov 2026 00:08:33 -0800 Subject: [PATCH 2/2] Observe payer funds once per signer request Adapt the imported counting RPC fixture and payment tests to the Node integration. Retain one funds observation per request, consolidate failure coverage, and correct the resulting HTTP failure documentation. --- docs/http-api.md | 2 +- internal/test/chainrpc.go | 503 ++++++-------------------------------- signer/payment.go | 20 +- signer/payment_test.go | 193 +++------------ signer/service_test.go | 141 ++++++----- 5 files changed, 183 insertions(+), 676 deletions(-) diff --git a/docs/http-api.md b/docs/http-api.md index c0cdb61..4460238 100644 --- a/docs/http-api.md +++ b/docs/http-api.md @@ -227,7 +227,7 @@ example `"X-List: a,b",Authorization: Bearer credential`. | 480 | Refresh payment parameters or orchestrator authentication. | | 481 | The price exceeds a ceiling or the initial session price. | | 482 | No tickets are needed; skip this payment cycle. | -| 500 | Payment generation or signing failed, including payer funds/chain failures detected after precheck. | +| 500 | Payment generation or signing failed. | | 502 | Restore the authorization webhook or correct its response. | | 503 | Retry after payment dependencies, discovery sources, or accounting storage recover. | diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go index 19632b6..1b8cdd7 100644 --- a/internal/test/chainrpc.go +++ b/internal/test/chainrpc.go @@ -3,464 +3,105 @@ package test import ( "encoding/hex" "encoding/json" - "fmt" "maps" - "math/big" "net/http" "net/http/httptest" - "strconv" "strings" "sync" "testing" - "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/stretchr/testify/require" ) -// BatchMode selects how ChainRPC answers a JSON-RPC batch, covering the ways -// providers and load balancers refuse them. -type BatchMode int - -const ( - BatchAccepted BatchMode = iota - BatchRefusedHTTP // HTTP 400 for any JSON array. - BatchRefusedObject // HTTP 200 with one error object instead of an array. - BatchRefusedEach // An array with -32601 for every element. - BatchTruncated // An array holding only the first response. -) - -// ChainSender is a TicketBroker sender. A nil Claimable entry uses Reserve. -type ChainSender struct { - Deposit, Reserve, WithdrawRound *big.Int - Claimable map[common.Address]*big.Int -} - -func (s ChainSender) clone() ChainSender { - s.Claimable = maps.Clone(s.Claimable) - return s -} - -type chainRounds struct { - Last int64 - Hashes map[int64]common.Hash -} - -// ChainState is the contract state of one block. State belongs to a contract -// address, so a retired contract keeps answering with what it last held. -type ChainState struct { - Number, L1Block int64 // A zero L1Block omits l1BlockNumber, as on Ethereum. - Contracts map[string]common.Address - rounds map[common.Address]chainRounds - active map[common.Address]map[common.Address]bool - senders map[common.Address]map[common.Address]ChainSender -} - -func (s *ChainState) clone() *ChainState { - next := &ChainState{Number: s.Number, L1Block: s.L1Block, Contracts: maps.Clone(s.Contracts), rounds: map[common.Address]chainRounds{}, active: map[common.Address]map[common.Address]bool{}, senders: map[common.Address]map[common.Address]ChainSender{}} - for address, rounds := range s.rounds { - next.rounds[address] = chainRounds{rounds.Last, maps.Clone(rounds.Hashes)} - } - for address, active := range s.active { - next.active[address] = maps.Clone(active) - } - for address, senders := range s.senders { - next.senders[address] = maps.Clone(senders) - } - return next -} - -// SetRound initializes a round on the current RoundsManager. -func (s *ChainState) SetRound(round int64, hash common.Hash) { - address := s.Contracts["RoundsManager"] - rounds := s.rounds[address] - rounds.Last, rounds.Hashes[round] = round, hash - s.rounds[address] = rounds -} - -// SetActive sets recipient status on the current BondingManager. -func (s *ChainState) SetActive(recipient common.Address, active bool) { - s.active[s.Contracts["BondingManager"]][recipient] = active -} - -// SetSender sets a payer's funds on the current TicketBroker. -func (s *ChainState) SetSender(payer common.Address, sender ChainSender) { - s.senders[s.Contracts["TicketBroker"]][payer] = sender.clone() -} - -// Move deploys a contract at a new address, starting from its current state. -// The old address keeps answering with the state it held. -func (s *ChainState) Move(contract string, address common.Address) { - previous := s.Contracts[contract] - switch contract { - case "RoundsManager": - s.rounds[address] = chainRounds{s.rounds[previous].Last, maps.Clone(s.rounds[previous].Hashes)} - case "BondingManager": - s.active[address] = maps.Clone(s.active[previous]) - case "TicketBroker": - s.senders[address] = maps.Clone(s.senders[previous]) - } - s.Contracts[contract] = address -} - -// ChainRPC is a counting Ethereum JSON-RPC HTTP server for the payment reads. -// Reads pinned to a block hash see that block's state and fail once the block -// is no longer canonical, so a test can observe mixed-block reads directly. +// Adapted from BlueClawNetwork/livepeer-node internal/test/chainrpc.go, +// 285eda984d7cce4e3030ef0510d63ae142534dd7, by Mike Zupper (MIT; see LICENSE). +// ChainRPC serves the payment read methods at one canonical block. Other tests +// can extend this fixture without duplicating the payment RPC protocol. type ChainRPC struct { URL string Controller common.Address mu sync.Mutex - blocks map[common.Hash]*ChainState - head common.Hash - orphaned map[common.Hash]bool - nextHash uint64 - latency time.Duration - down bool - batchMode BatchMode - failCall func(function string, input []byte) bool - trips int - batches int - calls map[string]int - unpinned int - batchSizes []int + stats ChainRPCStats } -const chainRPCController = "0x00000000000000000000000000000000000c0001" - -var chainSelectors = func() map[string]string { - selectors := map[string]string{} - for name, signature := range map[string]string{ - "Controller.getContract": "getContract(bytes32)", - "RoundsManager.lastInitializedRound": "lastInitializedRound()", - "RoundsManager.blockHashForRound": "blockHashForRound(uint256)", - "BondingManager.isActiveTranscoder": "isActiveTranscoder(address)", - "TicketBroker.getSenderInfo": "getSenderInfo(address)", - "TicketBroker.claimableReserve": "claimableReserve(address,address)", - } { - selectors[hex.EncodeToString(crypto.Keccak256([]byte(signature))[:4])] = name - } - return selectors -}() - -// ChainRoundHash is the deterministic initial block hash of a round. -func ChainRoundHash(round int64) common.Hash { - return crypto.Keccak256Hash([]byte("round"), big.NewInt(round).Bytes()) +type ChainRPCStats struct { + RoundTrips int + Calls map[string]int } -// NewChainRPC starts a chain at round 5 with its contracts registered. func NewChainRPC(t testing.TB) *ChainRPC { t.Helper() - state := &ChainState{Number: 0x100000, L1Block: 1000, Contracts: map[string]common.Address{ - "RoundsManager": common.HexToAddress("0xa1"), "BondingManager": common.HexToAddress("0xa2"), "TicketBroker": common.HexToAddress("0xa3"), - }, rounds: map[common.Address]chainRounds{}, active: map[common.Address]map[common.Address]bool{}, senders: map[common.Address]map[common.Address]ChainSender{}} - state.rounds[state.Contracts["RoundsManager"]] = chainRounds{Hashes: map[int64]common.Hash{}} - state.active[state.Contracts["BondingManager"]] = map[common.Address]bool{} - state.senders[state.Contracts["TicketBroker"]] = map[common.Address]ChainSender{} - state.SetRound(5, ChainRoundHash(5)) - c := &ChainRPC{Controller: common.HexToAddress(chainRPCController), blocks: map[common.Hash]*ChainState{}, orphaned: map[common.Hash]bool{}, calls: map[string]int{}} - c.head = c.addBlock(state) - server := httptest.NewServer(http.HandlerFunc(c.serve)) - t.Cleanup(server.Close) - c.URL = server.URL - return c -} - -// addBlock stores a state under a new unique hash. Caller holds c.mu, or the -// server is not yet running. -func (c *ChainRPC) addBlock(state *ChainState) common.Hash { - c.nextHash++ - hash := crypto.Keccak256Hash([]byte("block"), new(big.Int).SetUint64(c.nextHash).Bytes()) - c.blocks[hash] = state - return hash -} - -// Mine appends a block whose state is the head's state after mutate. -func (c *ChainRPC) Mine(mutate func(*ChainState)) common.Hash { - c.mu.Lock() - defer c.mu.Unlock() - state := c.blocks[c.head].clone() - state.Number++ - if mutate != nil { - mutate(state) - } - c.head = c.addBlock(state) - return c.head -} - -// Reorg replaces the head with a sibling at the same height. The old head stops -// answering pinned reads, as requireCanonical demands. -func (c *ChainRPC) Reorg(mutate func(*ChainState)) common.Hash { - c.mu.Lock() - defer c.mu.Unlock() - state := c.blocks[c.head].clone() - if mutate != nil { - mutate(state) - } - c.orphaned[c.head] = true - c.head = c.addBlock(state) - return c.head -} - -// Rewind makes an earlier block the head without orphaning anything, as a load -// balancer does when a request lands on a lagging backend. -func (c *ChainRPC) Rewind(block common.Hash) { - c.mu.Lock() - defer c.mu.Unlock() - c.head = block -} - -// Head returns the current head hash. -func (c *ChainRPC) Head() common.Hash { - c.mu.Lock() - defer c.mu.Unlock() - return c.head -} - -// SetLatency delays every HTTP round trip. -func (c *ChainRPC) SetLatency(latency time.Duration) { - c.mu.Lock() - defer c.mu.Unlock() - c.latency = latency -} - -// SetDown makes every request fail with HTTP 503. -func (c *ChainRPC) SetDown(down bool) { - c.mu.Lock() - defer c.mu.Unlock() - c.down = down -} - -// SetBatchMode selects how batches are answered. -func (c *ChainRPC) SetBatchMode(mode BatchMode) { - c.mu.Lock() - defer c.mu.Unlock() - c.batchMode = mode -} - -// FailCalls makes matching eth_calls return an execution error. A nil filter -// clears it. -func (c *ChainRPC) FailCalls(filter func(function string, input []byte) bool) { - c.mu.Lock() - defer c.mu.Unlock() - c.failCall = filter -} - -// ChainRPCStats counts traffic since the last Reset. RoundTrips are HTTP -// requests, so a batch is one; Calls counts each JSON-RPC call inside them, by -// method or Contract.function. -type ChainRPCStats struct { - RoundTrips, Batches, Unpinned int - Calls map[string]int - BatchSizes []int -} - -// Stats returns a copy of the current counts. -func (c *ChainRPC) Stats() ChainRPCStats { - c.mu.Lock() - defer c.mu.Unlock() - return ChainRPCStats{RoundTrips: c.trips, Batches: c.batches, Unpinned: c.unpinned, Calls: maps.Clone(c.calls), BatchSizes: append([]int(nil), c.batchSizes...)} -} - -// Reset clears the counts. -func (c *ChainRPC) Reset() { - c.mu.Lock() - defer c.mu.Unlock() - c.trips, c.batches, c.unpinned, c.calls, c.batchSizes = 0, 0, 0, map[string]int{}, nil -} - -type chainRPCRequest struct { - ID json.RawMessage `json:"id"` - Method string `json:"method"` - Params []json.RawMessage `json:"params"` -} - -func (c *ChainRPC) serve(w http.ResponseWriter, r *http.Request) { - var body json.RawMessage - if err := json.NewDecoder(r.Body).Decode(&body); err != nil { - http.Error(w, "invalid JSON", http.StatusBadRequest) - return - } - c.mu.Lock() - latency, down, mode := c.latency, c.down, c.batchMode - c.trips++ - c.mu.Unlock() - time.Sleep(latency) - if down { - http.Error(w, "unavailable", http.StatusServiceUnavailable) - return + c := &ChainRPC{Controller: common.HexToAddress("0xc001"), stats: ChainRPCStats{Calls: map[string]int{}}} + blockHash := common.HexToHash("0xabcd").Hex() + word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value } + addresses := map[string]string{} + for name, address := range map[string]string{"RoundsManager": "a1", "BondingManager": "a2", "TicketBroker": "a3"} { + addresses[hex.EncodeToString(crypto.Keccak256([]byte(name)))] = word(address) + } + methods := map[string]string{} + for signature, name := range map[string]string{ + "getContract(bytes32)": "Controller.getContract", "lastInitializedRound()": "RoundsManager.lastInitializedRound", + "blockHashForRound(uint256)": "RoundsManager.blockHashForRound", "isActiveTranscoder(address)": "BondingManager.isActiveTranscoder", + "getSenderInfo(address)": "TicketBroker.getSenderInfo", "claimableReserve(address,address)": "TicketBroker.claimableReserve", + } { + methods[hex.EncodeToString(crypto.Keccak256([]byte(signature))[:4])] = name } - w.Header().Set("Content-Type", "application/json") - if len(body) > 0 && body[0] == '[' { - var requests []chainRPCRequest - if err := json.Unmarshal(body, &requests); err != nil { - http.Error(w, "invalid JSON", http.StatusBadRequest) - return + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request struct { + ID json.RawMessage + Method string + Params []json.RawMessage } - c.mu.Lock() - c.batches++ - c.batchSizes = append(c.batchSizes, len(requests)) - c.mu.Unlock() - switch mode { - case BatchRefusedHTTP: - http.Error(w, "batch requests are not supported", http.StatusBadRequest) - return - case BatchRefusedObject: - _ = json.NewEncoder(w).Encode(chainRPCError(nil, -32600, "batch requests are not supported")) - return - case BatchRefusedEach: - responses := make([]map[string]any, len(requests)) - for i, request := range requests { - responses[i] = chainRPCError(request.ID, -32601, "batch requests are not supported") + require.NoError(t, json.NewDecoder(r.Body).Decode(&request)) + name := request.Method + var result any + switch request.Method { + case "eth_getBlockByNumber": + result = map[string]string{"number": "0x100000", "l1BlockNumber": "0x32", "hash": blockHash} + case "eth_call": + require.Len(t, request.Params, 2) + 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) + var call struct{ Input string } + require.NoError(t, json.Unmarshal(request.Params[0], &call)) + require.GreaterOrEqual(t, len(call.Input), 10) + name = methods[call.Input[2:10]] + switch name { + case "Controller.getContract": + require.Contains(t, addresses, call.Input[10:]) + result = "0x" + addresses[call.Input[10:]] + case "RoundsManager.lastInitializedRound": + result = "0x" + word("5") + case "RoundsManager.blockHashForRound": + result = common.HexToHash("0x1234").Hex() + case "BondingManager.isActiveTranscoder": + result = "0x" + word("1") + case "TicketBroker.getSenderInfo": + result = "0x" + word("de0b6b3a7640000") + word("0") + word("de0b6b3a7640000") + word("0") + case "TicketBroker.claimableReserve": + result = "0x" + word("de0b6b3a7640000") + default: + t.Errorf("unexpected payment contract call") } - _ = json.NewEncoder(w).Encode(responses) - return - case BatchTruncated: - requests = requests[:1] - } - responses := make([]map[string]any, len(requests)) - for i, request := range requests { - responses[i] = c.answer(request) + default: + t.Errorf("unexpected payment RPC method %q", request.Method) } - _ = json.NewEncoder(w).Encode(responses) - return - } - var request chainRPCRequest - if err := json.Unmarshal(body, &request); err != nil { - http.Error(w, "invalid JSON", http.StatusBadRequest) - return - } - _ = json.NewEncoder(w).Encode(c.answer(request)) -} - -func chainRPCError(id json.RawMessage, code int, message string) map[string]any { - return map[string]any{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": code, "message": message}} + c.mu.Lock() + c.stats.RoundTrips++ + c.stats.Calls[name]++ + c.mu.Unlock() + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result})) + })) + t.Cleanup(server.Close) + c.URL = server.URL + return c } -func (c *ChainRPC) answer(request chainRPCRequest) map[string]any { +func (c *ChainRPC) Stats() ChainRPCStats { c.mu.Lock() defer c.mu.Unlock() - result := func(value any) map[string]any { - return map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": value} - } - switch request.Method { - case "eth_chainId": - c.calls[request.Method]++ - return result("0x1") - case "eth_getBalance": - c.calls[request.Method]++ - return result("0x1") - case "eth_getBlockByNumber": - c.calls[request.Method]++ - state := c.blocks[c.head] - header := map[string]string{"number": "0x" + strconv.FormatInt(state.Number, 16), "hash": c.head.Hex()} - if state.L1Block != 0 { - header["l1BlockNumber"] = "0x" + strconv.FormatInt(state.L1Block, 16) - } - return result(header) - case "eth_call": - return c.call(request) - } - c.calls[request.Method]++ - return chainRPCError(request.ID, -32601, "unexpected method "+request.Method) -} - -// call answers one eth_call. Caller holds c.mu. -func (c *ChainRPC) call(request chainRPCRequest) map[string]any { - var args struct { - To common.Address `json:"to"` - Input string `json:"input"` - } - var pin struct { - BlockHash common.Hash `json:"blockHash"` - RequireCanonical bool `json:"requireCanonical"` - } - input := []byte(nil) - if len(request.Params) > 0 && json.Unmarshal(request.Params[0], &args) == nil { - input = common.FromHex(args.Input) - } - function := "unknown" - if len(input) >= 4 { - if name, ok := chainSelectors[hex.EncodeToString(input[:4])]; ok { - function = name - } - } - c.calls[function]++ - block := c.head - if len(request.Params) > 1 && json.Unmarshal(request.Params[1], &pin) == nil && pin.BlockHash != (common.Hash{}) && pin.RequireCanonical { - block = pin.BlockHash - } else { - c.unpinned++ - } - state := c.blocks[block] - if state == nil { - return chainRPCError(request.ID, -32000, "header not found") - } - if c.orphaned[block] { - return chainRPCError(request.ID, -32000, "block hash is not currently canonical") - } - if c.failCall != nil && c.failCall(function, input) { - return chainRPCError(request.ID, 3, "execution reverted") - } - output := state.execute(function, args.To, input) - return map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": "0x" + output} -} - -func chainWord(value *big.Int) string { - if value == nil { - value = new(big.Int) - } - return fmt.Sprintf("%064x", value) -} - -func chainArgAddress(input []byte, index int) common.Address { - offset := 4 + 32*index - if len(input) < offset+32 { - return common.Address{} - } - return common.BytesToAddress(input[offset : offset+32]) -} - -// execute returns hex output, empty when the address holds no such contract. -func (s *ChainState) execute(function string, to common.Address, input []byte) string { - switch function { - case "Controller.getContract": - if len(input) < 36 { - return "" - } - for name, address := range s.Contracts { - if crypto.Keccak256Hash([]byte(name)) == common.BytesToHash(input[4:36]) { - return chainWord(new(big.Int).SetBytes(address.Bytes())) - } - } - return chainWord(nil) - case "RoundsManager.lastInitializedRound": - if rounds, ok := s.rounds[to]; ok { - return chainWord(big.NewInt(rounds.Last)) - } - case "RoundsManager.blockHashForRound": - if rounds, ok := s.rounds[to]; ok && len(input) >= 36 { - return strings.TrimPrefix(rounds.Hashes[new(big.Int).SetBytes(input[4:36]).Int64()].Hex(), "0x") - } - case "BondingManager.isActiveTranscoder": - if active, ok := s.active[to]; ok { - if active[chainArgAddress(input, 0)] { - return chainWord(big.NewInt(1)) - } - return chainWord(nil) - } - case "TicketBroker.getSenderInfo": - if senders, ok := s.senders[to]; ok { - sender := senders[chainArgAddress(input, 0)] - return chainWord(sender.Deposit) + chainWord(sender.WithdrawRound) + chainWord(sender.Reserve) + chainWord(nil) - } - case "TicketBroker.claimableReserve": - if senders, ok := s.senders[to]; ok { - sender := senders[chainArgAddress(input, 0)] - if claimable, ok := sender.Claimable[chainArgAddress(input, 1)]; ok { - return chainWord(claimable) - } - return chainWord(sender.Reserve) - } - } - return "" + return ChainRPCStats{RoundTrips: c.stats.RoundTrips, Calls: maps.Clone(c.stats.Calls)} } diff --git a/signer/payment.go b/signer/payment.go index cf33d0a..c547e1f 100644 --- a/signer/payment.go +++ b/signer/payment.go @@ -100,7 +100,7 @@ func (s *Service) preparePayment(ctx context.Context, paymentType, manifest stri } // Precheck one ticket before generating the batch. // Observation/funding failures here are client errors. - funds, err := s.observePayer(ctx, ¶ms, "precheck") + funds, err := s.observePayer(ctx, ¶ms) if err == nil { err = s.checkTicketParams(ctx, ¶ms, 1, funds, "precheck") } @@ -121,9 +121,7 @@ func (s *Service) preparePayment(ctx context.Context, paymentType, manifest stri slog.With("component", "signer", "event", "payment", "operation", "payment").WarnContext(ctx, "signer ticket batch rejected", "outcome", "error", "error_kind", observability.ErrorKind(err)) return draftPayment{}, invalid("invalid ticket batch") } - // Recheck the full batch before authorization, on the precheck's observation: - // only the ticket count differs. A funding failure here is an internal generation - // failure; invalid exposure remains a client error. + // Recheck full-batch exposure on the same observation before authorization. if err := s.checkTicketParams(ctx, ¶ms, count, funds, "generation"); err != nil { return draftPayment{}, err } @@ -183,7 +181,7 @@ func (p draftPayment) Sign(ctx context.Context, signer pm.TicketSigner, metrics return signedPayment{Payment: base64.StdEncoding.EncodeToString(wire.EncodePayment(message)), SegCreds: base64.StdEncoding.EncodeToString(wire.EncodeSegData(segment))}, nil } -func (s *Service) observePayer(ctx context.Context, ticketParams *pm.TicketParams, phase string) (pm.PayerFunds, error) { +func (s *Service) observePayer(ctx context.Context, ticketParams *pm.TicketParams) (pm.PayerFunds, error) { if ticketParams == nil { return pm.PayerFunds{}, invalid("ticketParams is nil") } @@ -191,25 +189,25 @@ func (s *Service) observePayer(ctx context.Context, ticketParams *pm.TicketParam return pm.PayerFunds{}, invalid("ticketParams expiration block is 0") } if s.paymentChain == nil { - slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain is not configured", "phase", phase) + slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain is not configured", "phase", "precheck") return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } funds, err := s.paymentChain.PayerFunds(ctx, s.key.Address(), ticketParams.Recipient) if err != nil { - slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation failed", "phase", phase, "outcome", "error", "error_kind", observability.ErrorKind(err)) + slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation failed", "phase", "precheck", "outcome", "error", "error_kind", observability.ErrorKind(err)) return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } if funds.Snapshot.Block == nil { - slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation has no L1 block", "phase", phase) + slog.With("component", "signer", "event", "payment", "operation", "payment").ErrorContext(ctx, "signer payer chain observation has no L1 block", "phase", "precheck") return pm.PayerFunds{}, errors.Join(pm.ErrPayerUnavailable, errPaymentService) } s.observeFunds(funds, false) return funds, nil } -// Adapted from go-livepeer/pm/sender.go's validateTicketParams, by Yondon Fu, -// Nico Vergauwen and Rafał Leszko. Supplied creation-round metadata is retained; -// refresh is governed by parameter expiry on the L1 clock, not the current round. +// Adapted from livepeer/go-livepeer pm/sender.go (63df25b4ce, 2cc84ee216, +// 4b6ede31f0), by Yondon Fu, Nico Vergauwen and Rafał Leszko (MIT; see LICENSE). +// Supplied creation-round metadata is retained; refresh uses the L1 clock. func (s *Service) checkTicketParams(ctx context.Context, ticketParams *pm.TicketParams, numTickets int, funds pm.PayerFunds, phase string) error { latestL1Block := funds.Snapshot.Block currentBuffer := new(big.Int).Sub(ticketParams.ExpirationBlock, latestL1Block).Int64() diff --git a/signer/payment_test.go b/signer/payment_test.go index c2e7beb..a55e73e 100644 --- a/signer/payment_test.go +++ b/signer/payment_test.go @@ -1,19 +1,13 @@ package signer import ( - "bytes" "context" "encoding/base64" "encoding/json" "math/big" - "net/http" - "net/http/httptest" "os" "path/filepath" - "slices" - "strconv" "sync" - "sync/atomic" "testing" "time" @@ -165,170 +159,45 @@ func TestRecipientReservesWinningLiabilityAcrossConcurrentSessions(t *testing.T) require.Equal(t, 2, count, "rejected batches must roll back tickets and credit") } -// paidPath is one paid request end to end against a counting JSON-RPC server: -// the orchestrator's challenge, the signer's payment and the orchestrator's receipt. -type paidPath struct { - rpc *test.ChainRPC - engine *orchestrator.PaymentEngine - service *Service - payer *eth.Key - recipient ethcommon.Address - direct eth.PaymentChain - next atomic.Int64 -} - -func newPaidPath(t testing.TB) *paidPath { - t.Helper() - payer, recipient := paidPathKey(t), ethcommon.HexToAddress("0x1234") - rpcServer := test.NewChainRPC(t) - rpcServer.Mine(func(state *test.ChainState) { - state.SetActive(recipient, true) - state.SetSender(payer.Address(), test.ChainSender{Deposit: big.NewInt(1e18), Reserve: big.NewInt(1e18), WithdrawRound: new(big.Int)}) - }) - client, err := eth.OpenRPC(rpcServer.URL) +// 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) - t.Cleanup(client.Close) - contracts, err := eth.NewContracts(client, rpcServer.Controller) + 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) store, err := orchestrator.OpenRedeemerDB(filepath.Join(t.TempDir(), "payments.sqlite")) require.NoError(t, err) - t.Cleanup(func() { _ = store.Close() }) - direct := eth.PaymentChain{Contracts: contracts} + defer store.Close() + recipient := ethcommon.HexToAddress("0x1234") maximum := new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1)) - // One ticket covers the price and almost never wins, keeping winner storage - // out of the measurements of chain reads. - engine, err := orchestrator.NewPaymentEngine(store, pm.EthereumChain{Client: direct}, recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) + engine, err := orchestrator.NewPaymentEngine(store, direct, recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) require.NoError(t, err) - service := newService(payer) - t.Cleanup(func() { _ = service.Close() }) - service.SetPaymentChain(pm.EthereumChain{Client: direct}) - return &paidPath{rpc: rpcServer, engine: engine, service: service, payer: payer, recipient: recipient, direct: direct} -} - -func paidPathKey(t testing.TB) *eth.Key { - t.Helper() - keyFile, passwordPath := test.WriteKeystore(t, nil) - key, err := eth.OpenKeystoreFile(keyFile, passwordPath) - require.NoError(t, err) - return key -} - -func (p *paidPath) manifest() string { - return "session-" + strconv.FormatInt(p.next.Add(1), 10) -} - -func (p *paidPath) challenge(t testing.TB, ctx context.Context, manifest string) wire.OrchestratorInfo { - t.Helper() - _, err := p.engine.MakeChallenge(ctx, "runner", manifest, p.payer.Address(), 9, "fixed", "https://orch.example") + _, err = engine.MakeChallenge(t.Context(), "runner", "manifest", s.key.Address(), 9, "fixed", "https://orch.example") require.NoError(t, err) - info, err := p.engine.ChallengeInfo(manifest) + info, err := engine.ChallengeInfo("manifest") require.NoError(t, err) - return info -} + require.Equal(t, 6, rpc.Stats().RoundTrips, "challenge: snapshot (4) + activity (2)") -func (p *paidPath) sign(t testing.TB, info wire.OrchestratorInfo) (payment, segCreds string) { - t.Helper() - body, err := json.Marshal(map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) - require.NoError(t, err) - w := httptest.NewRecorder() - p.service.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/generate-live-payment", bytes.NewReader(body))) - require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + response := postPayment(t, s, 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(w.Body.Bytes(), &signed)) - return signed.Payment, signed.SegCreds -} - -func (p *paidPath) receive(ctx context.Context, manifest, payment, segCreds string) error { - _, _, err := p.engine.Receive(ctx, "runner", manifest, payment, segCreds) - return err -} - -// The expected log is the one derived from the code in the issue this test pins. -func TestPaidRequestRPCCalls(t *testing.T) { - path := newPaidPath(t) - manifest := path.manifest() - snapshot := map[string]int{"eth_getBlockByNumber": 1, "Controller.getContract": 1, "RoundsManager.lastInitializedRound": 1, "RoundsManager.blockHashForRound": 1} - isActive := map[string]int{"Controller.getContract": 1, "BondingManager.isActiveTranscoder": 1} - funds := map[string]int{"Controller.getContract": 1, "TicketBroker.getSenderInfo": 1, "TicketBroker.claimableReserve": 1} - sum := func(parts ...map[string]int) map[string]int { - total := map[string]int{} - for _, part := range parts { - for name, count := range part { - total[name] += count - } - } - return total - } - - path.rpc.Reset() - info := path.challenge(t, t.Context(), manifest) - stats := path.rpc.Stats() - require.Equal(t, 6, stats.RoundTrips, "challenge: snapshot (4) + IsActiveAt (2)") - require.Equal(t, sum(snapshot, isActive), stats.Calls) - require.Zero(t, stats.Unpinned) - - path.rpc.Reset() - payment, segCreds := path.sign(t, info) - stats = path.rpc.Stats() - require.Equal(t, 7, stats.RoundTrips, "signer: payer funds (7), reused by the batch recheck") - require.Equal(t, sum(snapshot, funds), stats.Calls) - require.Zero(t, stats.Unpinned) - - path.rpc.Reset() - require.NoError(t, path.receive(t.Context(), manifest, payment, segCreds)) - stats = path.rpc.Stats() - require.Equal(t, 9, stats.RoundTrips, "receive: payer funds (7) + IsActiveAt (2)") - require.Equal(t, sum(snapshot, funds, isActive), stats.Calls) - require.Zero(t, stats.Unpinned) -} - -// paidRequestLatency is a measurement rather than an assertion: RPC latency is -// injected, so the result shows how the round trips of a request add up. -const paidRequestLatency = 110 * time.Millisecond - -func (p *paidPath) request(t testing.TB, ctx context.Context) { - t.Helper() - manifest := p.manifest() - info := p.challenge(t, ctx, manifest) - payment, segCreds := p.sign(t, info) - require.NoError(t, p.receive(ctx, manifest, payment, segCreds)) -} - -func benchmarkPaidRequests(b *testing.B, path *paidPath, workers int) { - path.rpc.SetLatency(paidRequestLatency) - path.rpc.Reset() - var ( - next atomic.Int64 - mu sync.Mutex - latencies []time.Duration - running sync.WaitGroup - ) - b.ResetTimer() - for range workers { - running.Go(func() { - for next.Add(1) <= int64(b.N) { - start := time.Now() - path.request(b, context.Background()) - elapsed := time.Since(start) - mu.Lock() - latencies = append(latencies, elapsed) - mu.Unlock() - } - }) - } - running.Wait() - b.StopTimer() - slices.Sort(latencies) - b.ReportMetric(float64(path.rpc.Stats().RoundTrips)/float64(b.N), "rpc_trips/op") - b.ReportMetric(float64(latencies[len(latencies)/2].Milliseconds()), "p50_ms") - b.ReportMetric(float64(latencies[len(latencies)*99/100].Milliseconds()), "p99_ms") -} - -// Run with: go test ./signer -run '^$' -bench '^BenchmarkPaidRequest$' -benchtime 64x -func BenchmarkPaidRequest(b *testing.B) { - for _, workers := range []int{1, 32} { - b.Run("direct/workers="+strconv.Itoa(workers), func(b *testing.B) { - benchmarkPaidRequests(b, newPaidPath(b), workers) - }) - } + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &signed)) + require.Equal(t, 13, rpc.Stats().RoundTrips, "signer adds one funds observation (7)") + _, _, 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") + require.Equal(t, map[string]int{ + "eth_getBlockByNumber": 3, "Controller.getContract": 7, + "RoundsManager.lastInitializedRound": 3, "RoundsManager.blockHashForRound": 3, + "BondingManager.isActiveTranscoder": 2, "TicketBroker.getSenderInfo": 2, "TicketBroker.claimableReserve": 2, + }, stats.Calls) } diff --git a/signer/service_test.go b/signer/service_test.go index bb1127b..ceef660 100644 --- a/signer/service_test.go +++ b/signer/service_test.go @@ -423,80 +423,67 @@ func TestPaymentFailureStatuses(t *testing.T) { s, original := testService(t) logs := captureSignerLogs(t) key := s.key - for _, phase := range []struct { - name string - failOn, status int - }{{"precheck", 1, 400}} { - for _, cause := range []string{"deposit", "reserve", "withdrawal", "chain observation"} { - t.Run(phase.name+"/"+cause, func(t *testing.T) { - logs.Reset() - calls := 0 - s.SetPaymentChain(payerChainFunc(func(ctx context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { - calls++ - funds, err := (fundedPayer{}).PayerFunds(ctx, payer, recipient) - if calls == phase.failOn { - switch cause { - case "deposit": - funds.Deposit = new(big.Int) - case "reserve": - funds.Reserve = new(big.Int) - case "withdrawal": - funds.WithdrawRound = new(big.Int).Add(funds.Snapshot.Round, big.NewInt(1)) - case "chain observation": - err = errors.New("RPC https://rpc.internal/?token=private-rpc-detail failed") - } - } - return funds, err - })) - outcome := "rejected" - if cause == "chain observation" { - outcome = "error" - } - before := test.MetricSum(t, s.metrics.Registry, "signer_payments_total", map[string]string{"outcome": outcome}) - response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(original), "type": "fixed"}) - requirePaymentFailure(t, response, phase.status) - require.Equal(t, before+1, test.MetricSum(t, s.metrics.Registry, "signer_payments_total", map[string]string{"outcome": outcome})) - require.NotContains(t, response.Body.String(), "rpc.internal") - require.NotContains(t, response.Body.String(), "private-rpc-detail") - require.Contains(t, logs.String(), `"phase":"`+phase.name+`"`) - if cause == "chain observation" { - require.NotContains(t, logs.String(), "private-rpc-detail") - } - }) - } - } - // A request observes funds once, so the batch recheck cannot see a new funding - // failure over HTTP. Its handling stays distinct: not a client error, so 500. - for _, cause := range []string{"deposit", "reserve", "withdrawal"} { - t.Run("generation/"+cause, func(t *testing.T) { + for _, cause := range []string{"deposit", "reserve", "withdrawal", "chain observation"} { + t.Run("precheck/"+cause, func(t *testing.T) { logs.Reset() - funds, err := (fundedPayer{}).PayerFunds(t.Context(), s.key.Address(), ethcommon.Address{}) - require.NoError(t, err) - switch cause { - case "deposit": - funds.Deposit = new(big.Int) - case "reserve": - funds.Reserve = new(big.Int) - case "withdrawal": - funds.WithdrawRound = new(big.Int).Add(funds.Snapshot.Round, big.NewInt(1)) + s.SetPaymentChain(payerChainFunc(func(ctx context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { + funds, err := (fundedPayer{}).PayerFunds(ctx, payer, recipient) + switch cause { + case "deposit": + funds.Deposit = new(big.Int) + case "reserve": + funds.Reserve = new(big.Int) + case "withdrawal": + funds.WithdrawRound = new(big.Int).Add(funds.Snapshot.Round, big.NewInt(1)) + case "chain observation": + err = errors.New("RPC https://rpc.internal/?token=private-rpc-detail failed") + } + return funds, err + })) + outcome := "rejected" + if cause == "chain observation" { + outcome = "error" + } + before := test.MetricSum(t, s.metrics.Registry, "signer_payments_total", map[string]string{"outcome": outcome}) + response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(original), "type": "fixed"}) + requirePaymentFailure(t, response, 400) + require.Equal(t, before+1, test.MetricSum(t, s.metrics.Registry, "signer_payments_total", map[string]string{"outcome": outcome})) + require.NotContains(t, response.Body.String(), "rpc.internal") + require.NotContains(t, response.Body.String(), "private-rpc-detail") + require.Contains(t, logs.String(), `"phase":"precheck"`) + if cause == "chain observation" { + require.NotContains(t, logs.String(), "private-rpc-detail") } - params := pm.TicketParams{ExpirationBlock: new(big.Int).SetBytes(original.TicketParams.ExpirationBlock)} - err = s.checkTicketParams(t.Context(), ¶ms, 2, funds, "generation") - require.ErrorIs(t, err, pm.ErrPayerUnavailable) - _, clientError := errors.AsType[paymentFailure](err) - require.False(t, clientError, "generate answers 500 for anything but a paymentFailure") - require.Equal(t, "rejected", paymentOutcome(err)) - require.Contains(t, logs.String(), `"phase":"generation"`) }) } s.SetPaymentChain(fundedPayer{}) t.Run("batch exposure", func(t *testing.T) { - info := original + s, info := testService(t) + defer s.Close() + logs.Reset() info.Price.PricePerUnit = 20 // One ticket passes precheck; the full batch exceeds the limit. s.payerPolicy.MaxBatchEV = big.NewRat(15, 1) + countingKey := &failingPaymentSigner{TicketSigner: s.key} + s.key = countingKey + calls := 0 + s.SetPaymentChain(payerChainFunc(func(ctx context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { + calls++ + return (fundedPayer{}).PayerFunds(ctx, payer, recipient) + })) + var authCalls atomic.Int32 + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + authCalls.Add(1) + _, _ = io.WriteString(w, `{"status":200}`) + })) + defer webhook.Close() + require.NoError(t, s.SetAuthWebhook(testURL(t, webhook.URL), nil)) + s.events = eventSinkFunc(func(context.Context, signingEvent) error { t.Error("rejected payment emitted an event"); return nil }) requirePaymentFailure(t, postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}), 400) + require.Contains(t, logs.String(), `"phase":"generation"`) + require.Equal(t, 1, calls) + require.Zero(t, countingKey.calls.Load()) + require.Zero(t, authCalls.Load()) }) - s.payerPolicy = pm.DefaultPayerPolicy() for _, tc := range []struct { name string log string @@ -544,18 +531,30 @@ func TestPaymentFailureStatuses(t *testing.T) { func TestBatchRecheckReusesPrecheckObservation(t *testing.T) { s, info := testService(t) defer s.Close() + info.Price.PricePerUnit = 20 // Both request types need multiple tickets. + funds, err := (fundedPayer{}).PayerFunds(t.Context(), s.key.Address(), ethcommon.Address{}) + require.NoError(t, err) + want, err := (fundedPayer{}).PayerFunds(t.Context(), s.key.Address(), ethcommon.Address{}) + require.NoError(t, err) calls := 0 - s.SetPaymentChain(payerChainFunc(func(ctx context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { + s.SetPaymentChain(payerChainFunc(func(_ context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { calls++ - funds, err := (fundedPayer{}).PayerFunds(ctx, payer, recipient) + require.Equal(t, s.key.Address(), payer) + require.Equal(t, ethcommon.BytesToAddress(info.TicketParams.Recipient), recipient) if calls > 1 { - funds.Deposit = new(big.Int) + return pm.PayerFunds{}, errors.New("unexpected second observation") } - return funds, err + return funds, nil })) - response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": "fixed"}) - require.Equal(t, http.StatusOK, response.Code, response.Body.String()) - require.Equal(t, 1, calls) + for _, kind := range []string{"fixed", "live"} { + t.Run(kind, func(t *testing.T) { + calls = 0 + response := postPayment(t, s, map[string]any{"orchestrator": wire.EncodeOrchestratorInfo(info), "type": kind}) + require.Equal(t, http.StatusOK, response.Code, response.Body.String()) + require.Equal(t, 1, calls) + require.Equal(t, want, funds, "validation must not mutate the observation") + }) + } } func TestCancelledPaymentOutcome(t *testing.T) {