diff --git a/server/remote_discovery.go b/server/remote_discovery.go index bd08efd665..dbc4589e8f 100644 --- a/server/remote_discovery.go +++ b/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 { diff --git a/server/remote_signer.go b/server/remote_signer.go index f34cdead52..e32214964b 100644 --- a/server/remote_signer.go +++ b/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,7 +574,14 @@ 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 @@ -536,7 +589,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req } 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 diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 9e2d7a154a..46cf6d7857 100644 --- a/server/remote_signer_test.go +++ b/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) +}