feat(signer, orch): BYOC per-capability pricing and payment alignment

core/orch_test.go

@@ -1752,6 +1752,79 @@ func TestBYOCExternalCapsSenderPricing(t *testing.T) {
 	assert.Equal(t, int64(10), getBYOCPrice(addr3), "falls back to default")
 }
 
+func TestPriceInfoForCaps_BYOCUsesJobPrice(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	n, _ := NewLivepeerNode(nil, "", nil)
+	n.AutoAdjustPrice = false
+	n.SetBasePrice("default", NewFixedPrice(big.NewRat(1, 1)))
+	n.Recipient = new(pm.MockRecipient)
+	orch := NewOrchestrator(n, nil)
+
+	sender := ethcommon.HexToAddress("0x1000000000000000000000000000000000000000")
+	n.ExternalCapabilities.Capabilities["flux-schnell"] = &ExternalCapability{Name: "flux-schnell"}
+	n.SetPriceForExternalCapability("default", "flux-schnell", big.NewRat(42, 1))
+	n.SetPriceForExternalCapability(sender.Hex(), "flux-schnell", big.NewRat(99, 1))
+	// Built-in cap price must not be used for BYOC.
+	n.SetBasePriceForCap("default", Capability_BYOC, "flux-schnell", NewFixedPrice(big.NewRat(7, 1)))
+
+	byocCaps := NewCapabilities([]Capability{Capability_BYOC}, nil)
+	byocCaps.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_BYOC: &CapabilityConstraints{
+			Models: map[string]*ModelConstraint{
+				"flux-schnell": {Warm: true, Capacity: 1},
+			},
+		},
+	})
+	netCaps := byocCaps.ToNetCapabilities()
+
+	price, err := orch.PriceInfoForCaps(sender, "", netCaps)
+	require.Nil(err)
+	require.NotNil(price)
+	assert.Equal(int64(99), price.PricePerUnit)
+	assert.Equal(int64(1), price.PixelsPerUnit)
+
+	other := ethcommon.HexToAddress("0x2000000000000000000000000000000000000000")
+	price, err = orch.PriceInfoForCaps(other, "", netCaps)
+	require.Nil(err)
+	require.NotNil(price)
+	assert.Equal(int64(42), price.PricePerUnit, "falls back to default job price")
+
+	// Session-pinned price wins over job price.
+	n.Balances = NewAddressBalances(time.Minute)
+	n.Balances.Credit(sender, ManifestID("sess-1"), big.NewRat(0, 1))
+	n.Balances.SetFixedPrice(sender, ManifestID("sess-1"), big.NewRat(5, 2))
+	price, err = orch.PriceInfoForCaps(sender, ManifestID("sess-1"), netCaps)
+	require.Nil(err)
+	require.NotNil(price)
+	assert.Equal(int64(5), price.PricePerUnit)
+	assert.Equal(int64(2), price.PixelsPerUnit)
+
+	// TicketParams must be minted from the same PriceInfoForCaps rate (not base/cap).
+	recipient := n.Recipient.(*pm.MockRecipient)
+	jobPrice := big.NewRat(99, 1)
+	recipient.On("TicketParams", sender, mock.MatchedBy(func(p *big.Rat) bool {
+		return p != nil && p.Cmp(jobPrice) == 0
+	})).Return(&pm.TicketParams{
+		Recipient:         sender,
+		FaceValue:         big.NewInt(100),
+		WinProb:           big.NewInt(100),
+		RecipientRandHash: pm.RandHash(),
+		Seed:              big.NewInt(1),
+		ExpirationBlock:   big.NewInt(100),
+		PricePerPixel:     jobPrice,
+		ExpirationParams:  &pm.TicketExpirationParams{},
+	}, nil).Once()
+
+	price, err = orch.PriceInfoForCaps(sender, "", netCaps)
+	require.Nil(err)
+	params, err := orch.TicketParams(sender, price)
+	require.Nil(err)
+	require.NotNil(params)
+	recipient.AssertExpectations(t)
+}
+
 func TestBYOCExternalCapsPriceEdgeCases(t *testing.T) {
 	addr := "0x1000000000000000000000000000000000000000"
 

core/orchestrator.go

@@ -368,11 +368,37 @@ func (orch *orchestrator) PriceInfoForCaps(sender ethcommon.Address, manifestID
 		return nil, nil
 	}
 
+	if fixedPrice := orch.sessionFixedPrice(sender, manifestID); fixedPrice != nil {
+		return priceInfoFromRat(fixedPrice)
+	}
+
+	// BYOC jobs are priced via GetPriceForJob / JobPriceInfo, not GetBasePriceForCap.
+	// When GetOrchestrator is called with BYOC caps, TicketParams must be minted at
+	// that same rate so Payment.ExpectedPrice matches recipientRandHash.
+	if caps != nil {
+		coreCaps := CapabilitiesFromNetCapabilities(caps)
+		if modelID := coreCaps.ModelIDForCapability(Capability_BYOC); modelID != "" {
+			return orch.JobPriceInfo(sender, modelID)
+		}
+	}
+
 	price, err := orch.priceInfo(sender, manifestID, caps)
 	if err != nil {
 		return nil, err
 	}
 
+	return priceInfoFromRat(price)
+}
+
+func (orch *orchestrator) sessionFixedPrice(sender ethcommon.Address, manifestID ManifestID) *big.Rat {
+	if manifestID == "" || orch.node.Balances == nil {
+		return nil
+	}
+
+	return orch.node.Balances.FixedPrice(sender, manifestID)
+}
+
+func priceInfoFromRat(price *big.Rat) (*net.PriceInfo, error) {
 	if !price.Num().IsInt64() || !price.Denom().IsInt64() {
 		fixedPrice, err := common.PriceToInt64(price)
 		if err != nil {
@@ -390,11 +416,8 @@ func (orch *orchestrator) PriceInfoForCaps(sender ethcommon.Address, manifestID
 // priceInfo returns price per pixel as a fixed point number wrapped in a big.Rat
 func (orch *orchestrator) priceInfo(sender ethcommon.Address, manifestID ManifestID, caps *net.Capabilities) (*big.Rat, error) {
 	// If there is already a fixed price for the given session, use this price
-	if manifestID != "" {
-		fixedPrice := orch.node.Balances.FixedPrice(sender, manifestID)
-		if fixedPrice != nil {
-			return fixedPrice, nil
-		}
+	if fixedPrice := orch.sessionFixedPrice(sender, manifestID); fixedPrice != nil {
+		return fixedPrice, nil
 	}
 
 	transcodePrice := orch.node.GetBasePrice(sender.String())

pm/recipient_test.go

@@ -182,6 +182,41 @@ func TestReceiveTicket_InvalidSignature(t *testing.T) {
 	assert.False(ok)
 }
 
+// TestBYOCAlignedPrice_RecipientRandAccepted is the payment-alignment probe:
+// TicketParams minted at the orch job price accept matching ExpectedPrice, and
+// reject a divergent CapabilitiesPrices rate (the recipientRandHash failure mode).
+func TestBYOCAlignedPrice_RecipientRandAccepted(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	sender, b, _, gm, sm, tm, cfg, sig := newRecipientFixtureOrFatal(t)
+	sv := &stubSigVerifier{}
+	sv.SetVerifyResult(true)
+	v := NewValidator(sv, tm)
+	secret := [32]byte{3}
+	r := NewRecipientWithSecret(RandAddress(), b, v, gm, sm, tm, secret, cfg)
+
+	jobPrice := big.NewRat(99, 1)     // GetPriceForJob / PriceInfo
+	divergentCapPrice := big.NewRat(7, 1) // CapabilitiesPrices / base-cap price
+
+	params, err := r.TicketParams(sender, jobPrice)
+	require.NoError(err)
+	require.Zero(params.PricePerPixel.Cmp(jobPrice))
+
+	ticket := newTicket(sender, params, 0)
+	_, _, err = r.ReceiveTicket(ticket, sig, params.Seed)
+	require.NoError(err, "payment at orch job price must validate recipientRand")
+
+	// Signer/gateway substituting CapabilitiesPrices into ExpectedPrice.
+	badTicket := newTicket(sender, params, 1)
+	badTicket.PricePerPixel = divergentCapPrice
+	_, _, err = r.ReceiveTicket(badTicket, sig, params.Seed)
+	require.Error(err)
+	assert.Equal(errInvalidTicketRecipientRand.Error(), err.Error())
+	_, fatal := err.(*FatalReceiveErr)
+	assert.True(fatal)
+}
+
 func TestReceiveTicket_InvalidSender(t *testing.T) {
 	assert := assert.New(t)
 	sender, b, v, gm, sm, tm, cfg, sig := newRecipientFixtureOrFatal(t)

server/remote_discovery.go

@@ -392,15 +392,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

@@ -18,7 +18,6 @@ import (
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/proto"
 	"github.com/livepeer/go-livepeer/ai/runner"
-	"github.com/livepeer/go-livepeer/byoc"
 	"github.com/livepeer/go-livepeer/clog"
 	"github.com/livepeer/go-livepeer/common"
 	"github.com/livepeer/go-livepeer/core"
@@ -74,7 +73,9 @@ func (ls *LivepeerServer) SignOrchestratorInfo(w http.ResponseWriter, r *http.Re
 	_ = json.NewEncoder(w).Encode(results)
 }
 
-// SignBYOCJobRequest signs a BYOC job using the V1 binary format (FlattenBYOCJob).
+// SignBYOCJobRequest signs a BYOC job the same way gatewayJob.sign() and
+// network orchestrator verifyJobCreds do today: eth-sign over
+// request+parameters. (V1 FlattenBYOCJob is not yet deployed on network orchs.)
 type SignBYOCJobRequestInput struct {
 	ID             string `json:"id"`
 	Capability     string `json:"capability"`
@@ -113,15 +114,7 @@ func (ls *LivepeerServer) SignBYOCJobRequest(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	sigPayload := byoc.FlattenBYOCJob(&byoc.BYOCJobSigningInput{
-		ID:             req.ID,
-		Capability:     req.Capability,
-		Request:        req.Request,
-		Parameters:     req.Parameters,
-		TimeoutSeconds: req.TimeoutSeconds,
-	})
-
-	sig, err := gw.Sign(sigPayload)
+	sig, err := gw.Sign([]byte(req.Request + req.Parameters))
 	if err != nil {
 		clog.Errorf(ctx, "Failed to sign BYOC job request err=%q", err)
 		respondJsonError(ctx, w, err, http.StatusInternalServerError)
@@ -239,13 +232,14 @@ type RemotePaymentRequest struct {
 	// Set if an ID is needed to tie into orch accounting for a session. Optional
 	ManifestID string
 
-	// Number of pixels to generate a ticket for. Required if `type` is not set.
+	// Number of billable units to generate tickets for (e.g. compute-seconds
+	// for BYOC, or pixels for other callers). Required if `type` is not set.
 	InPixels int64 `json:"inPixels"`
 
 	// Job type to automatically calculate payments. Valid values: `lv2v`. Optional.
 	Type string `json:"type"`
 
-	// Capabilities to include in the ticket. Optional; may be set for the lv2v job type.
+	// Capabilities to include in segment credentials / max-price policy. Optional.
 	Capabilities []byte `json:"capabilities"`
 }
 
@@ -355,6 +349,25 @@ 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
+}
+
 // 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 +402,29 @@ 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)
+	}
+
+	// ExpectedPrice must match the rate baked into TicketParams.recipientRandHash.
+	// Never overwrite PriceInfo from CapabilitiesPrices — the orch is the sole
+	// rate source (via GetOrchestrator / JobPriceInfo).
+	priceInfo := oInfo.PriceInfo
+	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 +482,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,21 +548,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 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
 	}
 	if pixels <= 0 {
-		err = errors.New("missing pixels or job type")
+		err = errors.New("missing billable units or job type")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}