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 new file mode 100644 index 0000000..1b8cdd7 --- /dev/null +++ b/internal/test/chainrpc.go @@ -0,0 +1,107 @@ +package test + +import ( + "encoding/hex" + "encoding/json" + "maps" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/stretchr/testify/require" +) + +// 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 + stats ChainRPCStats +} + +type ChainRPCStats struct { + RoundTrips int + Calls map[string]int +} + +func NewChainRPC(t testing.TB) *ChainRPC { + t.Helper() + 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 + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request struct { + ID json.RawMessage + Method string + Params []json.RawMessage + } + 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") + } + default: + t.Errorf("unexpected payment RPC method %q", request.Method) + } + 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) Stats() ChainRPCStats { + c.mu.Lock() + defer c.mu.Unlock() + return ChainRPCStats{RoundTrips: c.stats.RoundTrips, Calls: maps.Clone(c.stats.Calls)} +} diff --git a/signer/payment.go b/signer/payment.go index f3c55b0..c547e1f 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) + 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,8 @@ 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 full-batch exposure on the same observation before authorization. + 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 +181,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) (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) + 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)) - return errors.Join(pm.ErrPayerUnavailable, errPaymentService) + 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) - return errors.Join(pm.ErrPayerUnavailable, errPaymentService) + 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 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() if currentBuffer <= paramsExpiryBuffer { diff --git a/signer/payment_test.go b/signer/payment_test.go index 01db9f0..a55e73e 100644 --- a/signer/payment_test.go +++ b/signer/payment_test.go @@ -158,3 +158,46 @@ func TestRecipientReservesWinningLiabilityAcrossConcurrentSessions(t *testing.T) require.NoError(t, err) require.Equal(t, 2, count, "rejected batches must roll back tickets and credit") } + +// 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) + 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) + defer store.Close() + recipient := ethcommon.HexToAddress("0x1234") + maximum := new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1)) + engine, err := orchestrator.NewPaymentEngine(store, direct, recipient, big.NewInt(1e12), new(big.Int).Quo(maximum, big.NewInt(1e5))) + require.NoError(t, err) + _, err = engine.MakeChallenge(t.Context(), "runner", "manifest", s.key.Address(), 9, "fixed", "https://orch.example") + require.NoError(t, err) + info, err := engine.ChallengeInfo("manifest") + require.NoError(t, err) + require.Equal(t, 6, rpc.Stats().RoundTrips, "challenge: snapshot (4) + activity (2)") + + 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(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 10c4e57..ceef660 100644 --- a/signer/service_test.go +++ b/signer/service_test.go @@ -423,56 +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}, {"generation", 2, 500}} { - 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") + for _, cause := range []string{"deposit", "reserve", "withdrawal", "chain observation"} { + t.Run("precheck/"+cause, func(t *testing.T) { + logs.Reset() + 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") + } + }) } 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 @@ -517,6 +528,35 @@ 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(_ context.Context, payer, recipient ethcommon.Address) (pm.PayerFunds, error) { + calls++ + require.Equal(t, s.key.Address(), payer) + require.Equal(t, ethcommon.BytesToAddress(info.TicketParams.Recipient), recipient) + if calls > 1 { + return pm.PayerFunds{}, errors.New("unexpected second observation") + } + return funds, nil + })) + 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) { s, info := testService(t) defer s.Close()