diff --git a/byoc/job_orchestrator.go b/byoc/job_orchestrator.go index 210474992e..5a7b28812d 100644 --- a/byoc/job_orchestrator.go +++ b/byoc/job_orchestrator.go @@ -566,18 +566,26 @@ func (bso *BYOCOrchestratorServer) verifyJobCreds(ctx context.Context, jobCreds return nil, errSegSig } - if !bso.orch.VerifySig(ethcommon.HexToAddress(jobData.Sender), jobData.Request+jobData.Parameters, sigByte) { - clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender) - return nil, errSegSig - } - - if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil { - return nil, errZeroCapacity + sender := ethcommon.HexToAddress(jobData.Sender) + + // Verify V1 structured binary format (matches DMZ /sign-byoc-job signing). + v1Payload := FlattenBYOCJob(&BYOCJobSigningInput{ + ID: jobData.ID, + Capability: jobData.Capability, + Request: jobData.Request, + Parameters: jobData.Parameters, + TimeoutSeconds: jobData.Timeout, + }) + if bso.orch.VerifySig(sender, string(v1Payload), sigByte) { + if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil { + return nil, errZeroCapacity + } + jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability) + return jobData, nil } - jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability) - - return jobData, nil + clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender) + return nil, errSegSig } func (bso *BYOCOrchestratorServer) verifyTokenCreds(ctx context.Context, tokenCreds string) (*JobSender, error) { diff --git a/cmd/livepeer/starter/flags.go b/cmd/livepeer/starter/flags.go index af2d7eafd7..1fff92adfa 100644 --- a/cmd/livepeer/starter/flags.go +++ b/cmd/livepeer/starter/flags.go @@ -149,6 +149,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig { cfg.RemoteSignerWebhookHeaders = fs.String("remoteSignerWebhookHeaders", *cfg.RemoteSignerWebhookHeaders, "Map of headers to use for remote signer webhook requests. e.g. 'header:val,header2:val2'") cfg.RemoteSignerAllowNoAuth = fs.Bool("remoteSignerAllowNoAuth", *cfg.RemoteSignerAllowNoAuth, "Allow an unauthenticated remote signer on a public -httpAddr (no webhook). UNSAFE: signs payments from this node's deposit for any reachable caller; restrict access externally (proxy/private network).") cfg.RemoteDiscovery = fs.Bool("remoteDiscovery", *cfg.RemoteDiscovery, "Enable orchestrator discovery on remote signers") + cfg.ByocPerCapPricing = fs.Bool("byocPerCapPricing", *cfg.ByocPerCapPricing, "Remote signer: resolve BYOC live-payment fee from the orchestrator's per-capability CapabilitiesPrices (keyed on the request capability) instead of the flat base price. Default OFF (byte-identical to base-price behavior).") // Gateway metrics cfg.KafkaBootstrapServers = fs.String("kafkaBootstrapServers", *cfg.KafkaBootstrapServers, "URL of Kafka Bootstrap Servers") diff --git a/cmd/livepeer/starter/starter.go b/cmd/livepeer/starter/starter.go index 6c7741e7c3..2a00ad7d35 100755 --- a/cmd/livepeer/starter/starter.go +++ b/cmd/livepeer/starter/starter.go @@ -180,6 +180,7 @@ type LivepeerConfig struct { RemoteSignerWebhookHeaders *string RemoteSignerAllowNoAuth *bool RemoteDiscovery *bool + ByocPerCapPricing *bool AIRunnerImage *string AIRunnerImageOverrides *string AIVerboseLogs *bool @@ -326,6 +327,7 @@ func DefaultLivepeerConfig() LivepeerConfig { defaultRemoteSignerWebhookHeaders := "" defaultRemoteSignerAllowNoAuth := false defaultRemoteDiscovery := false + defaultByocPerCapPricing := false // Gateway logs defaultKafkaBootstrapServers := "" @@ -458,6 +460,7 @@ func DefaultLivepeerConfig() LivepeerConfig { RemoteSignerWebhookHeaders: &defaultRemoteSignerWebhookHeaders, RemoteSignerAllowNoAuth: &defaultRemoteSignerAllowNoAuth, RemoteDiscovery: &defaultRemoteDiscovery, + ByocPerCapPricing: &defaultByocPerCapPricing, // Gateway logs KafkaBootstrapServers: &defaultKafkaBootstrapServers, @@ -1881,6 +1884,9 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) { if cfg.RemoteDiscovery != nil { n.RemoteDiscovery = *cfg.RemoteDiscovery } + if cfg.ByocPerCapPricing != nil { + n.ByocPerCapPricing = *cfg.ByocPerCapPricing + } if cfg.LiveAIHeartbeatHeaders != nil { n.LiveAIHeartbeatHeaders = parseHeaderMap(*cfg.LiveAIHeartbeatHeaders) } diff --git a/core/livepeernode.go b/core/livepeernode.go index 31cc4d108a..0ace42dd55 100644 --- a/core/livepeernode.go +++ b/core/livepeernode.go @@ -158,6 +158,7 @@ type LivepeerNode struct { RemoteEthAddr ethcommon.Address // eth address of the remote signer InfoSig []byte // sig over eth address for the OrchestratorInfo request RemoteDiscovery bool // expose remote discovery endpoint when enabled + ByocPerCapPricing bool // resolve BYOC fee from per-capability CapabilitiesPrices instead of base price (remote signer; default OFF) // Thread safety for config fields mu sync.RWMutex diff --git a/server/remote_signer.go b/server/remote_signer.go index f34cdead52..8713116bce 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" @@ -33,6 +34,7 @@ const HTTPStatusPriceExceeded = 481 const HTTPStatusNoTickets = 482 const RefreshSessionOrchestratorURLHeader = "Livepeer-Orchestrator-URL" const RemoteType_LiveVideoToVideo = "lv2v" +const RemoteType_BYOC = "byoc" const PipelineLiveVideoToVideo = "live-video-to-video" const remoteSignerAuthIDHeader = "Signer-Auth-Id" @@ -242,11 +244,25 @@ type RemotePaymentRequest struct { // Number of pixels to generate a ticket for. Required if `type` is not set. InPixels int64 `json:"inPixels"` - // Job type to automatically calculate payments. Valid values: `lv2v`. Optional. + // Job type to automatically calculate payments. Valid values: `lv2v`, `byoc`. Optional. Type string `json:"type"` // 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 +371,120 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason) } +// maxUsageLabelLen bounds the length of gateway-supplied usage labels +// (pipeline/model_id) before they are emitted to the metering pipeline. +// req.Capability / req.ModelID come straight from the request body, so a buggy +// or malicious caller could otherwise send very long / high-cardinality values +// that inflate Kafka payload size and downstream label cardinality (the +// pipeline label was previously a small fixed set). 128 runes is generous for +// real capability/model identifiers. +const maxUsageLabelLen = 128 + +// sanitizeUsageLabel trims surrounding whitespace and caps the length (in +// runes, to never emit invalid UTF-8) of a gateway-supplied metering label. +func sanitizeUsageLabel(s string) string { + s = strings.TrimSpace(s) + if r := []rune(s); len(r) > maxUsageLabelLen { + return string(r[:maxUsageLabelLen]) + } + return s +} + +// 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 and override the labels whenever they are set, +// regardless of `Type` — this decouples usage attribution from `Type`, which +// still drives fee/pixel routing. The gateway-supplied values are sanitized +// (trimmed + length-capped) to bound payload size and label cardinality. +// +// 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 the +// untouched label(s) remain empty (the collector then defaults them to +// "unknown"). This makes non-BYOC (lv2v) callers with no override +// byte-identical to before. +func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pipeline, modelID string) { + if req.Type == RemoteType_LiveVideoToVideo { + pipeline = PipelineLiveVideoToVideo + if caps != nil { + modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo) + } + } + if req.Type == RemoteType_BYOC && caps != nil { + if m := caps.ModelIDForCapability(core.Capability_BYOC); m != "" { + pipeline = core.CapabilityToPipeline(core.Capability_BYOC) + modelID = m + } + } + if c := sanitizeUsageLabel(req.Capability); c != "" { + pipeline = c + } + if m := sanitizeUsageLabel(req.ModelID); m != "" { + modelID = m + } + return pipeline, modelID +} + +// byocCapabilityName returns the BYOC capability constraint used for per-cap +// pricing lookup. The gateway may supply it as req.Capability (lv2v path) or +// encode it in the BYOC capabilities protobuf (type:"byoc" path). +func byocCapabilityName(req *RemotePaymentRequest, caps *core.Capabilities) string { + if c := sanitizeUsageLabel(req.Capability); c != "" { + return c + } + if caps != nil { + return caps.ModelIDForCapability(core.Capability_BYOC) + } + return "" +} + +func isByocBillingType(t string) bool { + return t == RemoteType_BYOC || t == RemoteType_LiveVideoToVideo +} + +// 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: } (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, +// or no matching entry with a positive rate). A matching entry with a +// non-positive rate is skipped rather than aborting the scan, so a later valid +// duplicate entry for the same constraint (e.g. due to misconfiguration) is +// still honored. Callers fall back to the base oInfo.PriceInfo when nil is +// returned, 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. Skip invalid/zero entries and keep + // scanning so a later valid duplicate for the same constraint isn't + // shadowed; if none is found we fall back to base (never zeroing the fee). + if p.PricePerUnit <= 0 || p.PixelsPerUnit <= 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,7 +519,43 @@ 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 + var parsedCaps *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 + } + parsedCaps = core.CapabilitiesFromNetCapabilities(&caps) + } + capName := byocCapabilityName(&req, parsedCaps) + // BYOC per-cap pricing changes the billing basis (compute-seconds instead + // of lv2v pixels). Apply for lv2v or explicit byoc job types when a + // capability constraint is present. + if ls.LivepeerNode.ByocPerCapPricing && capName != "" && isByocBillingType(req.Type) { + priceReq := req + if priceReq.Capability == "" { + priceReq.Capability = capName + } + if capPrice := resolveByocPrice(&priceReq, &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) @@ -456,14 +622,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req // 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) + if parsedCaps != nil { + streamParams.Capabilities = parsedCaps } pmParams := pmTicketParams(oInfo.TicketParams) @@ -528,7 +688,17 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req lastUpdate = now } billableSecs := now.Sub(lastUpdate).Seconds() - if req.Type == RemoteType_LiveVideoToVideo { + if useByocPricing || req.Type == RemoteType_BYOC { + // 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 diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 9e2d7a154a..99a0b65a33 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -11,6 +11,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "strings" "testing" "testing/synctest" "time" @@ -1692,3 +1693,415 @@ func TestGetOrchInfoSig_SendsConfiguredHeaders(t *testing.T) { require.Equal([]byte{0x12, 0x34}, []byte(resp.Address)) require.Equal([]byte{0xab, 0xcd}, []byte(resp.Signature)) } + +// lv2vCapsWithModel builds a core.Capabilities advertising a single warm model +// for the live-video-to-video capability (used to exercise the existing +// capabilities-derived model_id fallback). +func lv2vCapsWithModel(t *testing.T, modelID string) *core.Capabilities { + t.Helper() + c := core.NewCapabilities([]core.Capability{core.Capability_LiveVideoToVideo}, nil) + c.SetPerCapabilityConstraints(core.PerCapabilityConstraints{ + core.Capability_LiveVideoToVideo: &core.CapabilityConstraints{ + Models: core.ModelConstraints{modelID: &core.ModelConstraint{Warm: true}}, + }, + }) + return c +} + +func byocCapsWithModel(t *testing.T, modelID string) *core.Capabilities { + t.Helper() + c := core.NewCapabilities([]core.Capability{core.Capability_BYOC}, nil) + c.SetPerCapabilityConstraints(core.PerCapabilityConstraints{ + core.Capability_BYOC: &core.CapabilityConstraints{ + Models: core.ModelConstraints{modelID: &core.ModelConstraint{Warm: true}}, + }, + }) + return c +} + +func byocCapsProtoBytes(t *testing.T, modelID string) []byte { + t.Helper() + caps := byocCapsWithModel(t, modelID) + b, err := proto.Marshal(caps.ToNetCapabilities()) + require.NoError(t, err) + return b +} + +// TestResolveUsageLabels asserts the additive usage-attribution contract: +// - when the gateway supplies the real BYOC capability/model, the metered +// pipeline + model_id reflect them (the root-cause fix); +// - when those fields are empty, behavior is unchanged (lv2v constant + +// capabilities-derived model id, or empty → "unknown" downstream). +// +// This is hermetic: it exercises the pure label-resolution helper without the +// CGO ffmpeg toolchain, the remote-signer HTTP handler, or Kafka. +func TestResolveUsageLabels(t *testing.T) { + tests := []struct { + name string + req RemotePaymentRequest + caps *core.Capabilities + wantPipeline string + wantModelID string + }{ + { + name: "lv2v unchanged: no new fields, no caps", + req: RemotePaymentRequest{Type: RemoteType_LiveVideoToVideo}, + caps: nil, + wantPipeline: PipelineLiveVideoToVideo, + wantModelID: "", + }, + { + name: "lv2v unchanged: model derived from capabilities", + req: RemotePaymentRequest{Type: RemoteType_LiveVideoToVideo}, + caps: lv2vCapsWithModel(t, "streamdiffusion"), + wantPipeline: PipelineLiveVideoToVideo, + wantModelID: "streamdiffusion", + }, + { + name: "byoc type: labels from capabilities protobuf", + req: RemotePaymentRequest{ + Type: RemoteType_BYOC, + }, + caps: byocCapsWithModel(t, "flux-schnell"), + wantPipeline: "byoc", + wantModelID: "flux-schnell", + }, + { + name: "byoc: real capability + model override lv2v defaults", + req: RemotePaymentRequest{ + Type: RemoteType_LiveVideoToVideo, + Capability: "nano-banana", + ModelID: "google/nano-banana", + }, + caps: lv2vCapsWithModel(t, "streamdiffusion"), + wantPipeline: "nano-banana", + wantModelID: "google/nano-banana", + }, + { + name: "byoc: capability set, empty model falls back to caps-derived", + req: RemotePaymentRequest{ + Type: RemoteType_LiveVideoToVideo, + Capability: "nano-banana", + }, + caps: lv2vCapsWithModel(t, "streamdiffusion"), + wantPipeline: "nano-banana", + wantModelID: "streamdiffusion", + }, + { + name: "non-lv2v with no fields: both empty (collector defaults to unknown)", + req: RemotePaymentRequest{}, + caps: nil, + wantPipeline: "", + wantModelID: "", + }, + { + name: "byoc: whitespace-only override is ignored, lv2v defaults kept", + req: RemotePaymentRequest{ + Type: RemoteType_LiveVideoToVideo, + Capability: " ", + ModelID: "\t\n", + }, + caps: lv2vCapsWithModel(t, "streamdiffusion"), + wantPipeline: PipelineLiveVideoToVideo, + wantModelID: "streamdiffusion", + }, + { + name: "byoc: oversized override labels are length-capped", + req: RemotePaymentRequest{ + Type: RemoteType_LiveVideoToVideo, + Capability: strings.Repeat("c", maxUsageLabelLen+50), + ModelID: strings.Repeat("m", maxUsageLabelLen+50), + }, + caps: nil, + wantPipeline: strings.Repeat("c", maxUsageLabelLen), + wantModelID: strings.Repeat("m", maxUsageLabelLen), + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + require := require.New(t) + req := tc.req + pipeline, modelID := resolveUsageLabels(&req, tc.caps) + require.Equal(tc.wantPipeline, pipeline, "pipeline") + require.Equal(tc.wantModelID, modelID, "model_id") + }) + } +} + +// TestResolveByocPrice covers the pure per-capability BYOC price resolver: +// it matches a Capability_BYOC entry whose Constraint equals req.Capability, +// ignores model_id and non-BYOC entries, and returns nil (→ caller falls back +// to base) for no-match / empty-capability / non-positive rate. +// +// Hermetic: exercises the pure resolver without the CGO ffmpeg toolchain, the +// remote-signer HTTP handler, or Kafka. +func TestResolveByocPrice(t *testing.T) { + require := require.New(t) + + byocCap := uint32(core.Capability_BYOC) + oInfo := &net.OrchestratorInfo{ + PriceInfo: &net.PriceInfo{PricePerUnit: 100, PixelsPerUnit: 1}, // base (unused by resolver) + CapabilitiesPrices: []*net.PriceInfo{ + {Capability: byocCap, Constraint: "nano-banana", PricePerUnit: 10, PixelsPerUnit: 1}, + {Capability: byocCap, Constraint: "recraft-v4", PricePerUnit: 20, PixelsPerUnit: 1}, + // decoy: a non-BYOC capability sharing the constraint name must NOT match + {Capability: uint32(core.Capability_LiveVideoToVideo), Constraint: "nano-banana", PricePerUnit: 999, PixelsPerUnit: 1}, + // decoy: zero/invalid BYOC rate must be treated as no-match (fallback) + {Capability: byocCap, Constraint: "free-cap", PricePerUnit: 0, PixelsPerUnit: 1}, + // misconfig: an early invalid duplicate must NOT shadow a later valid + // entry for the same constraint (scan continues past invalid rates). + {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) { + req := &RemotePaymentRequest{Capability: tc.capability} + got := resolveByocPrice(req, 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") + }) + } +} + +// TestGenerateLivePayment_ByocPerCapPricing drives the full GenerateLivePayment +// handler (no CGO needed) and proves, via the resulting signed balance/state: +// - flag OFF: a matching cap is ignored; the fee is the byte-identical lv2v +// synthetic-pixel base fee (zero-regression); +// - flag ON: the fee is the resolved per-capability rate * compute-seconds; +// - flag ON: two caps in a 2:1 tariff produce a 2:1 fee ratio; +// - flag ON, unknown cap: falls back to the base lv2v fee; +// - flag ON with base >> 2x cap returns 200 (no false "price more than +// doubled"), proving the oInfo.PriceInfo = capPrice correction. +func TestGenerateLivePayment_ByocPerCapPricing(t *testing.T) { + require := require.New(t) + + ethClient := newTestEthClient(t) + node, _ := core.NewLivepeerNode(ethClient, "", nil) + node.Balances = core.NewAddressBalances(1 * time.Minute) + + // Large per-ticket EV keeps the lv2v fee within the handler's 100-ticket cap + // (lv2v fee = base * 1280*720*30*60 is large), while the small per-second + // BYOC fees resolve to a single ticket. + 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} + + // Generous global max price so neither base nor cap prices trip the ceiling. + autoPrice, err := core.NewAutoConvertedPrice("", big.NewRat(1_000, 1), nil) + require.NoError(err) + BroadcastCfg.SetMaxPrice(autoPrice) + defer BroadcastCfg.SetMaxPrice(nil) + + // base is intentionally >> 2x the cap rates so a regression that left + // oInfo.PriceInfo at base (while locking the cap price into state) would trip + // the >2x doubling guard. With the correction it must NOT trip. + 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) + + // Fresh-session preload bills 60 seconds (lv2v) of data. + 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, + Capability: 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 + } + + // fee = numTickets*ev - balance for a fresh session; recover it from the + // signed state's balance and the locked initial price (also returned). + feeFromState := func(state RemotePaymentState) *big.Rat { + bal := new(big.Rat) + _, ok := bal.SetString(state.Balance) + require.True(ok, "parse balance %q", state.Balance) + // numTickets*ev = balance + fee, and numTickets*ev is the smallest + // multiple of ev that is >= max(fee, ev); recover fee = k*ev - balance + // where k = round((balance)/ev upward to the enclosing ticket). Simpler: + // derive fee from initialPrice * pixels directly using the locked state. + price := big.NewRat(state.InitialPricePerUnit, state.InitialPixelsPerUnit) + var pixels int64 + if state.InitialPricePerUnit == basePPU && state.InitialPixelsPerUnit == 1 { + pixels = lv2vPixels // base path: lv2v synthetic pixels + } else { + pixels = preloadSecs // byoc path: compute-seconds + } + return new(big.Rat).Mul(price, big.NewRat(pixels, 1)) + } + + assertBalanceConsistent := func(state RemotePaymentState, fee *big.Rat) { + // Independent check: balance must equal numTickets*ev - fee with + // numTickets = ceil(max(fee,ev)/ev), validating the fee end-to-end. + 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()) + } + + // --- flag OFF: cap present but ignored → base lv2v synthetic-pixel fee --- + ls.LivepeerNode.ByocPerCapPricing = false + offState := doPayment("nano-banana") + require.EqualValues(basePPU, offState.InitialPricePerUnit, "flag OFF must lock the base price") + require.EqualValues(1, offState.InitialPixelsPerUnit) + offFee := new(big.Rat).Mul(big.NewRat(basePPU, 1), big.NewRat(lv2vPixels, 1)) + require.Zero(feeFromState(offState).Cmp(offFee)) + assertBalanceConsistent(offState, offFee) + + // --- flag ON --- + ls.LivepeerNode.ByocPerCapPricing = true + + // nano-banana: fee = capNanoPPU * 60 (seconds basis); base>>2x cap must NOT + // trip the doubling guard (proves oInfo.PriceInfo = capPrice). + nanoState := doPayment("nano-banana") + require.EqualValues(capNanoPPU, nanoState.InitialPricePerUnit, "flag ON 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) + + // recraft-v4 priced 2x nano → fee ratio is exactly 2:1. + 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") + + // unknown cap with flag ON → fall back to base lv2v fee (zero-regression). + unknownState := doPayment("totally-unknown-cap") + require.EqualValues(basePPU, unknownState.InitialPricePerUnit, "unknown cap must fall back to base") + require.Zero(feeFromState(unknownState).Cmp(offFee)) + assertBalanceConsistent(unknownState, offFee) + + // type:"byoc" + capabilities protobuf (no capability string) must bill like BYOC. + doPaymentByocType := func(modelID string) RemotePaymentState { + reqBody, err := json.Marshal(RemotePaymentRequest{ + Orchestrator: orchBlob, + Type: RemoteType_BYOC, + Capabilities: byocCapsProtoBytes(t, modelID), + }) + 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 + } + + byocTypeState := doPaymentByocType("nano-banana") + require.EqualValues(capNanoPPU, byocTypeState.InitialPricePerUnit, "type byoc must resolve per-cap price from capabilities proto") + require.Zero(feeFromState(byocTypeState).Cmp(nanoFee), "type byoc fee") +}