feat(signer): charge BYOC live payments at the per-capability price #3967
+465
−7
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,6 +7,7 @@ import ( | |
| "errors" | ||
| "fmt" | ||
| "io" | ||
| "math" | ||
| "math/big" | ||
| "net/http" | ||
| "net/url" | ||
|
|
@@ -247,6 +248,20 @@ type RemotePaymentRequest struct { | |
|
|
||
| // Capabilities to include in the ticket. Optional; may be set for the lv2v job type. | ||
| Capabilities []byte `json:"capabilities"` | ||
|
|
||
| // Capability is the real BYOC capability name (e.g. "nano-banana"). When | ||
| // set, it overrides the metered `pipeline` label for the emitted | ||
| // create_signed_ticket usage event, decoupling usage attribution from | ||
| // `Type` (which stays "lv2v" for fee/pixel routing). Optional; empty | ||
| // preserves the previous behavior (pipeline falls back to the lv2v | ||
| // constant when Type=="lv2v"). | ||
| Capability string `json:"capability,omitempty"` | ||
|
|
||
| // ModelID is the specific provider model from the request | ||
| // payload/parameters. When set, it overrides the metered `model_id` | ||
| // label. Optional; empty preserves the previous behavior (the | ||
| // capabilities-derived model id, which defaults to "unknown" downstream). | ||
| ModelID string `json:"model_id,omitempty"` | ||
| } | ||
|
|
||
| // Returned by the remote signer and includes a new payment plus updated state. | ||
|
|
@@ -355,6 +370,69 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS | |
| return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason) | ||
| } | ||
|
|
||
| // resolveUsageLabels derives the metering `pipeline` and `model_id` labels for | ||
| // the emitted create_signed_ticket usage event. | ||
| // | ||
| // The real BYOC capability/model supplied by the gateway (req.Capability / | ||
| // req.ModelID) take precedence so BYOC jobs are attributed to their true | ||
| // pipeline + model. When those additive fields are empty, the labels fall back | ||
| // to the previous behavior: for lv2v the pipeline is the live-video-to-video | ||
| // constant and the model id is derived from the request capabilities; for any | ||
| // other request both remain empty (the collector then defaults them to | ||
| // "unknown"). This makes non-BYOC (lv2v) callers byte-identical to before and | ||
| // keeps usage attribution decoupled from `Type`, which still drives fee/pixel | ||
| // routing. | ||
| func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pipeline, modelID string) { | ||
| if req.Type == RemoteType_LiveVideoToVideo { | ||
| pipeline = PipelineLiveVideoToVideo | ||
| modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo) | ||
| } | ||
| if req.Capability != "" { | ||
| pipeline = req.Capability | ||
| } | ||
| if req.ModelID != "" { | ||
| modelID = req.ModelID | ||
| } | ||
| return pipeline, modelID | ||
| } | ||
|
|
||
| // resolveByocPrice resolves the per-capability BYOC price advertised in the | ||
| // orchestrator's OrchestratorInfo.CapabilitiesPrices, keyed on the request | ||
| // capability name (req.Capability). The orchestrator appends external BYOC | ||
| // capability prices to CapabilitiesPrices as | ||
| // {Capability: Capability_BYOC, Constraint: <capability name>} (see | ||
| // core/orchestrator.go GetCapabilitiesPrices), so a match is a BYOC entry whose | ||
| // Constraint equals req.Capability. | ||
| // | ||
| // Granularity is per-capability only: req.ModelID is intentionally NOT used for | ||
| // price selection (model_id still flows through for metering attribution). | ||
| // | ||
| // Returns nil when there is no usable per-capability price (no capability set, | ||
| // no matching entry, or a non-positive rate). Callers fall back to the base | ||
| // oInfo.PriceInfo in that case, preserving today's behavior for unconfigured or | ||
| // misconfigured capabilities. This is a pure function (no flag, no IO) so it is | ||
| // hermetically testable without the CGO/ffmpeg toolchain. | ||
| func resolveByocPrice(req *RemotePaymentRequest, oInfo *net.OrchestratorInfo) *net.PriceInfo { | ||
| if req == nil || oInfo == nil || req.Capability == "" { | ||
| return nil | ||
| } | ||
| for _, p := range oInfo.CapabilitiesPrices { | ||
| if p == nil { | ||
| continue | ||
| } | ||
| if p.Capability != uint32(core.Capability_BYOC) || p.Constraint != req.Capability { | ||
| continue | ||
| } | ||
| // Only honor a valid positive rate; otherwise fall back to base so a | ||
| // zero/invalid advertised cap price never zeros out or breaks the fee. | ||
| if p.PricePerUnit <= 0 || p.PixelsPerUnit <= 0 { | ||
| return nil | ||
| } | ||
| 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,7 +467,25 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req | |
| respondJsonError(ctx, w, err, http.StatusBadRequest) | ||
| return | ||
| } | ||
| // BYOC per-capability pricing (flag-gated, default OFF). When enabled and the | ||
| // gateway supplied a real capability, resolve the per-capability price from | ||
| // the orchestrator's advertised CapabilitiesPrices and use it instead of the | ||
| // flat base price. The resolved price is written back to oInfo.PriceInfo so it | ||
| // is the single source for: state init, initialPrice, the max-price ceiling, | ||
| // the payment's ExpectedPrice (which the orchestrator uses to set its fixed | ||
| // per-session price), and the price-doubling guard in validatePrice (which | ||
| // reads sess.OrchestratorInfo.PriceInfo) — keeping all of them cap-vs-cap | ||
| // consistent. When the flag is OFF or no usable cap price matches, behavior is | ||
| // byte-identical to the base-price path. | ||
| priceInfo := oInfo.PriceInfo | ||
| useByocPricing := false | ||
| if ls.LivepeerNode.ByocPerCapPricing && req.Capability != "" { | ||
| if capPrice := resolveByocPrice(&req, &oInfo); capPrice != nil { | ||
| priceInfo = capPrice | ||
| oInfo.PriceInfo = capPrice | ||
| useByocPricing = true | ||
| } | ||
| } | ||
|
Comment on lines
484
to
+497
Author
There was a problem hiding this comment. Addressed the first part in 84c706a: BYOC pricing is now additionally gated on On persisting the pricing mode in signed state: I left that out intentionally for now. The price is re-derived per request from |
||
| if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 { | ||
| err := fmt.Errorf("missing or zero priceInfo") | ||
| respondJsonError(ctx, w, err, http.StatusBadRequest) | ||
|
|
@@ -528,7 +624,17 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req | |
| lastUpdate = now | ||
| } | ||
| billableSecs := now.Sub(lastUpdate).Seconds() | ||
| if useByocPricing { | ||
| // BYOC per-capability tariff is denominated per compute-second (wei/sec), | ||
| // so charge on seconds directly rather than synthesizing lv2v pixels. With | ||
| // pixels = ceil(billableSecs) and the resolved per-cap price kept as its | ||
| // reduced wei/sec rational, calculateFee yields capPriceWeiPerSec * seconds. | ||
| if billableSecs <= 0 { | ||
| // preload with 60 seconds, mirroring the lv2v first-call behavior | ||
| billableSecs = (60 * time.Second).Seconds() | ||
| } | ||
| pixels = int64(math.Ceil(billableSecs)) | ||
| } else if req.Type == RemoteType_LiveVideoToVideo { | ||
| info := defaultSegInfo | ||
| if billableSecs <= 0 { | ||
| // preload with 60 seconds of data for LV2V | ||
|
|
@@ -680,12 +786,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req | |
| if state.SequenceNumber == 0 { | ||
| sessionStatus = "new" | ||
| } | ||
| pipeline, modelID := resolveUsageLabels(&req, streamParams.Capabilities) | ||
| // NB: This could could drop events if tha Kafka queue is full! | ||
| monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{ | ||
| "session_id": state.StateID, | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.