feat(signer): bill BYOC live payments from CapabilitiesPrices

server/remote_discovery.go

@@ -391,15 +391,8 @@ func capabilityPrice(info *common.OrchNetworkCapabilities, capability core.Capab
 	if info == nil {
 		return nil
 	}
-	// Check per-capability price if it exists
-	for _, capPrice := range info.CapabilitiesPrices {
-		if capPrice == nil || capPrice.PixelsPerUnit <= 0 || core.Capability(capPrice.Capability) != capability {
-			continue
-		}
-		price := new(big.Rat).SetFrac64(capPrice.PricePerUnit, capPrice.PixelsPerUnit)
-		if capPrice.Constraint == modelID {
-			return price
-		}
+	if capPrice := findCapPriceInfo(info.CapabilitiesPrices, capability, modelID, false); capPrice != nil {
+		return new(big.Rat).SetFrac64(capPrice.PricePerUnit, capPrice.PixelsPerUnit)
 	}
 	// Global fallback if no per-capability price is available.
 	if info.PriceInfo == nil || info.PriceInfo.PixelsPerUnit <= 0 {

server/remote_signer.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"math/big"
 	"net/http"
 	"net/url"
@@ -355,6 +356,38 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS
 	return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason)
 }
 
+// findCapPriceInfo returns the first CapabilitiesPrices entry matching
+// capability+modelID. requirePositiveRate skips non-positive PricePerUnit and
+// keeps scanning; otherwise a zero rate is allowed.
+func findCapPriceInfo(prices []*net.PriceInfo, capability core.Capability, modelID string, requirePositiveRate bool) *net.PriceInfo {
+	for _, p := range prices {
+		if p == nil || core.Capability(p.Capability) != capability || p.Constraint != modelID {
+			continue
+		}
+		if p.PixelsPerUnit <= 0 {
+			continue
+		}
+		if requirePositiveRate && p.PricePerUnit <= 0 {
+			continue
+		}
+		return &net.PriceInfo{PricePerUnit: p.PricePerUnit, PixelsPerUnit: p.PixelsPerUnit}
+	}
+	return nil
+}
+
+// resolveByocPrice looks up the BYOC model constraint from caps in
+// oInfo.CapabilitiesPrices. Returns nil when no usable price matches.
+func resolveByocPrice(caps *core.Capabilities, oInfo *net.OrchestratorInfo) *net.PriceInfo {
+	if caps == nil || oInfo == nil {
+		return nil
+	}
+	constraint := caps.ModelIDForCapability(core.Capability_BYOC)
+	if constraint == "" {
+		return nil
+	}
+	return findCapPriceInfo(oInfo.CapabilitiesPrices, core.Capability_BYOC, constraint, true)
+}
+
 // GenerateLivePayment handles remote generation of a payment for live streams.
 func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Request) {
 	requestID := string(core.RandomManifestID())
@@ -389,14 +422,35 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
-	priceInfo := oInfo.PriceInfo
-	if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 {
-		err := fmt.Errorf("missing or zero priceInfo")
+	if oInfo.TicketParams == nil {
+		err := fmt.Errorf("missing ticketParams in OrchestratorInfo")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
-	if oInfo.TicketParams == nil {
-		err := fmt.Errorf("missing ticketParams in OrchestratorInfo")
+
+	var reqCaps *core.Capabilities
+	if len(req.Capabilities) > 0 {
+		var caps net.Capabilities
+		if err := proto.Unmarshal(req.Capabilities, &caps); err != nil {
+			clog.Errorf(ctx, "Failed to unmarshal capabilities err=%q", err)
+			respondJsonError(ctx, w, err, http.StatusBadRequest)
+			return
+		}
+		reqCaps = core.CapabilitiesFromNetCapabilities(&caps)
+	}
+
+	// BYOC caps: use per-capability price (and bill compute-seconds below).
+	// Otherwise keep base PriceInfo / lv2v pixel pricing. Write back so state,
+	// ExpectedPrice, and validatePrice all see the same rate.
+	priceInfo := oInfo.PriceInfo
+	useByocPricing := false
+	if capPrice := resolveByocPrice(reqCaps, &oInfo); capPrice != nil {
+		priceInfo = capPrice
+		oInfo.PriceInfo = capPrice
+		useByocPricing = true
+	}
+	if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 {
+		err := fmt.Errorf("missing or zero priceInfo")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
@@ -454,16 +508,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 
 	streamParams := &core.StreamParameters{
 		// Embedded within genSegCreds, may be used by orch for payment accounting
-		ManifestID: core.ManifestID(manifestID),
-	}
-	if len(req.Capabilities) > 0 {
-		var caps net.Capabilities
-		if err := proto.Unmarshal(req.Capabilities, &caps); err != nil {
-			clog.Errorf(ctx, "Failed to unmarshal capabilities err=%q", err)
-			respondJsonError(ctx, w, err, http.StatusBadRequest)
-			return
-		}
-		streamParams.Capabilities = core.CapabilitiesFromNetCapabilities(&caps)
+		ManifestID:   core.ManifestID(manifestID),
+		Capabilities: reqCaps,
 	}
 
 	pmParams := pmTicketParams(oInfo.TicketParams)
@@ -528,15 +574,22 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		lastUpdate = now
 	}
 	billableSecs := now.Sub(lastUpdate).Seconds()
-	if req.Type == RemoteType_LiveVideoToVideo {
+	switch {
+	case useByocPricing:
+		// BYOC prices are per compute-second; bill seconds instead of lv2v pixels.
+		if billableSecs <= 0 {
+			billableSecs = (60 * time.Second).Seconds()
+		}
+		pixels = int64(math.Ceil(billableSecs))
+	case req.Type == RemoteType_LiveVideoToVideo:
 		info := defaultSegInfo
 		if billableSecs <= 0 {
 			// preload with 60 seconds of data for LV2V
 			billableSecs = (60 * time.Second).Seconds()
 		}
 		pixelsPerSec := float64(info.Height) * float64(info.Width) * float64(info.FPS)
 		pixels = int64(pixelsPerSec * billableSecs) // pixels to charge for
-	} else if req.Type != "" {
+	case req.Type != "":
 		err = errors.New("invalid job type")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return

server/remote_signer_test.go

@@ -1692,3 +1692,234 @@ func TestGetOrchInfoSig_SendsConfiguredHeaders(t *testing.T) {
 	require.Equal([]byte{0x12, 0x34}, []byte(resp.Address))
 	require.Equal([]byte{0xab, 0xcd}, []byte(resp.Signature))
 }
+
+func byocCaps(t *testing.T, modelID string) *core.Capabilities {
+	t.Helper()
+	if modelID == "" {
+		return nil
+	}
+	caps := core.NewCapabilities([]core.Capability{core.Capability_BYOC}, nil)
+	caps.SetPerCapabilityConstraints(core.PerCapabilityConstraints{
+		core.Capability_BYOC: &core.CapabilityConstraints{
+			Models: map[string]*core.ModelConstraint{
+				modelID: {},
+			},
+		},
+	})
+	return caps
+}
+
+func byocCapsBlob(t *testing.T, modelID string) []byte {
+	t.Helper()
+	caps := byocCaps(t, modelID)
+	if caps == nil {
+		return nil
+	}
+	blob, err := proto.Marshal(caps.ToNetCapabilities())
+	require.NoError(t, err)
+	return blob
+}
+
+func TestResolveByocPrice(t *testing.T) {
+	require := require.New(t)
+
+	byocCap := uint32(core.Capability_BYOC)
+	oInfo := &net.OrchestratorInfo{
+		PriceInfo: &net.PriceInfo{PricePerUnit: 100, PixelsPerUnit: 1},
+		CapabilitiesPrices: []*net.PriceInfo{
+			{Capability: byocCap, Constraint: "nano-banana", PricePerUnit: 10, PixelsPerUnit: 1},
+			{Capability: byocCap, Constraint: "recraft-v4", PricePerUnit: 20, PixelsPerUnit: 1},
+			{Capability: uint32(core.Capability_LiveVideoToVideo), Constraint: "nano-banana", PricePerUnit: 999, PixelsPerUnit: 1},
+			{Capability: byocCap, Constraint: "free-cap", PricePerUnit: 0, PixelsPerUnit: 1},
+			{Capability: byocCap, Constraint: "dup-cap", PricePerUnit: 0, PixelsPerUnit: 1},
+			{Capability: byocCap, Constraint: "dup-cap", PricePerUnit: 30, PixelsPerUnit: 1},
+		},
+	}
+
+	tests := []struct {
+		name       string
+		capability string
+		oInfo      *net.OrchestratorInfo
+		want       *net.PriceInfo
+	}{
+		{
+			name:       "resolves per capability",
+			capability: "recraft-v4",
+			oInfo:      oInfo,
+			want:       &net.PriceInfo{PricePerUnit: 20, PixelsPerUnit: 1},
+		},
+		{
+			name:       "resolves the other capability",
+			capability: "nano-banana",
+			oInfo:      oInfo,
+			want:       &net.PriceInfo{PricePerUnit: 10, PixelsPerUnit: 1},
+		},
+		{
+			name:       "unknown capability falls back (nil)",
+			capability: "does-not-exist",
+			oInfo:      oInfo,
+			want:       nil,
+		},
+		{
+			name:       "empty capability falls back (nil)",
+			capability: "",
+			oInfo:      oInfo,
+			want:       nil,
+		},
+		{
+			name:       "zero/invalid matched rate falls back (nil)",
+			capability: "free-cap",
+			oInfo:      oInfo,
+			want:       nil,
+		},
+		{
+			name:       "skips invalid duplicate, honors later valid entry",
+			capability: "dup-cap",
+			oInfo:      oInfo,
+			want:       &net.PriceInfo{PricePerUnit: 30, PixelsPerUnit: 1},
+		},
+		{
+			name:       "no capabilities prices falls back (nil)",
+			capability: "nano-banana",
+			oInfo:      &net.OrchestratorInfo{PriceInfo: &net.PriceInfo{PricePerUnit: 100, PixelsPerUnit: 1}},
+			want:       nil,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := resolveByocPrice(byocCaps(t, tc.capability), tc.oInfo)
+			if tc.want == nil {
+				require.Nil(got)
+				return
+			}
+			require.NotNil(got)
+			require.Equal(tc.want.PricePerUnit, got.PricePerUnit, "PricePerUnit")
+			require.Equal(tc.want.PixelsPerUnit, got.PixelsPerUnit, "PixelsPerUnit")
+		})
+	}
+}
+
+func TestGenerateLivePayment_ByocCapabilityPricing(t *testing.T) {
+	require := require.New(t)
+
+	ethClient := newTestEthClient(t)
+	node, _ := core.NewLivepeerNode(ethClient, "", nil)
+	node.Balances = core.NewAddressBalances(1 * time.Minute)
+
+	// Large EV keeps the lv2v fee under the 100-ticket cap.
+	ev := big.NewRat(10_000_000_000, 1)
+	var totalTickets uint32
+	sender := newMockSender(mockSenderConfig{
+		ev: ev,
+		createTicketBatchFn: func(args mock.Arguments, batch *pm.TicketBatch) {
+			size := args.Int(1)
+			*batch = *defaultTicketBatch()
+			var baseSig []byte
+			if len(batch.SenderParams) > 0 && batch.SenderParams[0] != nil {
+				baseSig = batch.SenderParams[0].Sig
+			}
+			batch.SenderParams = make([]*pm.TicketSenderParams, size)
+			for i := 0; i < size; i++ {
+				totalTickets++
+				batch.SenderParams[i] = &pm.TicketSenderParams{SenderNonce: totalTickets, Sig: baseSig}
+			}
+		},
+	})
+	node.Sender = sender
+	ls := &LivepeerServer{LivepeerNode: node}
+
+	autoPrice, err := core.NewAutoConvertedPrice("", big.NewRat(1_000, 1), nil)
+	require.NoError(err)
+	BroadcastCfg.SetMaxPrice(autoPrice)
+	defer BroadcastCfg.SetMaxPrice(nil)
+
+	// base >> 2x caps so leaving oInfo.PriceInfo at base would trip the doubling guard.
+	const basePPU, capNanoPPU, capRecraftPPU = 100, 10, 20
+	oInfo := &net.OrchestratorInfo{
+		Address:   ethClient.addr.Bytes(),
+		PriceInfo: &net.PriceInfo{PricePerUnit: basePPU, PixelsPerUnit: 1},
+		TicketParams: &net.TicketParams{
+			Recipient: pm.RandAddress().Bytes(),
+		},
+		AuthToken: stubAuthToken,
+		CapabilitiesPrices: []*net.PriceInfo{
+			{Capability: uint32(core.Capability_BYOC), Constraint: "nano-banana", PricePerUnit: capNanoPPU, PixelsPerUnit: 1},
+			{Capability: uint32(core.Capability_BYOC), Constraint: "recraft-v4", PricePerUnit: capRecraftPPU, PixelsPerUnit: 1},
+		},
+	}
+	orchBlob, err := proto.Marshal(oInfo)
+	require.NoError(err)
+
+	const preloadSecs = 60
+	lv2vPixels := int64(defaultSegInfo.Height) * int64(defaultSegInfo.Width) * int64(defaultSegInfo.FPS) * preloadSecs
+
+	doPayment := func(capability string) RemotePaymentState {
+		reqBody, err := json.Marshal(RemotePaymentRequest{
+			Orchestrator: orchBlob,
+			Type:         RemoteType_LiveVideoToVideo,
+			Capabilities: byocCapsBlob(t, capability),
+		})
+		require.NoError(err)
+		req := httptest.NewRequest(http.MethodPost, "/generate-live-payment", bytes.NewReader(reqBody))
+		rr := httptest.NewRecorder()
+		ls.GenerateLivePayment(rr, req)
+		require.Equal(http.StatusOK, rr.Code, "body=%s", rr.Body.String())
+		var resp RemotePaymentResponse
+		require.NoError(json.NewDecoder(rr.Body).Decode(&resp))
+		var state RemotePaymentState
+		require.NoError(json.Unmarshal(resp.State.State, &state))
+		return state
+	}
+
+	feeFromState := func(state RemotePaymentState) *big.Rat {
+		bal := new(big.Rat)
+		_, ok := bal.SetString(state.Balance)
+		require.True(ok, "parse balance %q", state.Balance)
+		price := big.NewRat(state.InitialPricePerUnit, state.InitialPixelsPerUnit)
+		var pixels int64
+		if state.InitialPricePerUnit == basePPU && state.InitialPixelsPerUnit == 1 {
+			pixels = lv2vPixels
+		} else {
+			pixels = preloadSecs
+		}
+		return new(big.Rat).Mul(price, big.NewRat(pixels, 1))
+	}
+
+	assertBalanceConsistent := func(state RemotePaymentState, fee *big.Rat) {
+		smc := fee
+		if ev.Cmp(smc) > 0 {
+			smc = ev
+		}
+		q := new(big.Rat).Quo(smc, ev)
+		nt := new(big.Int).Quo(q.Num(), q.Denom())
+		if new(big.Int).Rem(q.Num(), q.Denom()).Sign() != 0 {
+			nt.Add(nt, big.NewInt(1))
+		}
+		wantBal := new(big.Rat).Sub(new(big.Rat).Mul(new(big.Rat).SetInt(nt), ev), fee)
+		gotBal := new(big.Rat)
+		_, ok := gotBal.SetString(state.Balance)
+		require.True(ok)
+		require.Zero(gotBal.Cmp(wantBal), "balance got=%s want=%s fee=%s", gotBal.RatString(), wantBal.RatString(), fee.RatString())
+	}
+
+	nanoState := doPayment("nano-banana")
+	require.EqualValues(capNanoPPU, nanoState.InitialPricePerUnit, "must lock the resolved cap price")
+	require.EqualValues(1, nanoState.InitialPixelsPerUnit)
+	nanoFee := big.NewRat(capNanoPPU*preloadSecs, 1)
+	require.Zero(feeFromState(nanoState).Cmp(nanoFee), "nano fee")
+	assertBalanceConsistent(nanoState, nanoFee)
+
+	recraftState := doPayment("recraft-v4")
+	require.EqualValues(capRecraftPPU, recraftState.InitialPricePerUnit)
+	recraftFee := big.NewRat(capRecraftPPU*preloadSecs, 1)
+	require.Zero(feeFromState(recraftState).Cmp(recraftFee), "recraft fee")
+	assertBalanceConsistent(recraftState, recraftFee)
+	require.Zero(recraftFee.Cmp(new(big.Rat).Mul(nanoFee, big.NewRat(2, 1))), "recraft fee must be 2x nano fee")
+
+	unknownState := doPayment("totally-unknown-cap")
+	require.EqualValues(basePPU, unknownState.InitialPricePerUnit, "unknown cap must fall back to base")
+	fallbackFee := new(big.Rat).Mul(big.NewRat(basePPU, 1), big.NewRat(lv2vPixels, 1))
+	require.Zero(feeFromState(unknownState).Cmp(fallbackFee))
+	assertBalanceConsistent(unknownState, fallbackFee)
+}