Observe payer funds once per signer request

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. |
 

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)}
+}

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, &params, 1, "precheck"); err != nil {
+	funds, err := s.observePayer(ctx, &params)
+	if err == nil {
+		err = s.checkTicketParams(ctx, &params, 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, &params, count, "generation"); err != nil {
+	// Recheck full-batch exposure on the same observation before authorization.
+	if err := s.checkTicketParams(ctx, &params, 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 {

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)
+}

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()