From 023d25e9eceffd63aff33e7072cc0ab9c38a5f99 Mon Sep 17 00:00:00 2001 From: seanhanca Date: Fri, 26 Jun 2026 20:21:15 -0700 Subject: [PATCH 1/3] fix(signer): meter real BYOC capability + model_id for usage events The remote signer hardcoded the create_signed_ticket `pipeline` to the lv2v constant whenever the gateway sent `type:"lv2v"` (which BYOC jobs always do for fee/pixel routing), and derived `model_id` only from the LiveVideoToVideo capability constraints. For BYOC capabilities (e.g. nano-banana) this emitted pipeline=live-video-to-video and an empty model_id, which the OpenMeter collector recorded as live-video-to-video / unknown. Add two additive, backward-compatible fields to RemotePaymentRequest: `capability` and `model_id`. When set by the gateway, they override the metered pipeline + model_id labels, decoupling usage attribution from `Type` (which still drives fee/pixel routing). When empty, behavior is byte-identical to before (lv2v constant + capabilities-derived model id; collector defaults empties to "unknown"). Label resolution is extracted into a pure resolveUsageLabels helper with a hermetic table test (TestResolveUsageLabels) that runs without the CGO ffmpeg toolchain. --- server/remote_signer.go | 47 +++++++++++++++++--- server/remote_signer_test.go | 86 ++++++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 6 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index 2c167150ce..c2678a98ca 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -247,6 +247,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 +369,32 @@ 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 +} + // GenerateLivePayment handles remote generation of a payment for live streams. func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Request) { requestID := string(core.RandomManifestID()) @@ -680,12 +720,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req if state.SequenceNumber == 0 { sessionStatus = "new" } - pipeline := "" - modelID := "" - if req.Type == RemoteType_LiveVideoToVideo { - pipeline = PipelineLiveVideoToVideo - modelID = streamParams.Capabilities.ModelIDForCapability(core.Capability_LiveVideoToVideo) - } + 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, diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 9e2d7a154a..504b736e80 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1692,3 +1692,89 @@ 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 +} + +// 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) { + require := require.New(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: 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: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + req := tc.req + pipeline, modelID := resolveUsageLabels(&req, tc.caps) + require.Equal(tc.wantPipeline, pipeline, "pipeline") + require.Equal(tc.wantModelID, modelID, "model_id") + }) + } +} From 597dbc621b922008bbbad414ee2dca9c838320fc Mon Sep 17 00:00:00 2001 From: qianghan Date: Mon, 29 Jun 2026 10:50:11 -0700 Subject: [PATCH 2/3] feat(signer): charge BYOC live payments at the per-capability price The remote signer's GenerateLivePayment charged every BYOC generation a flat, model-independent fee: it read only oInfo.PriceInfo (the base price) and synthesized lv2v pixels (1280*720*30 * billableSecs), so nano-banana, recraft-v4, ltx-*, etc. all cost the same regardless of the per-capability prices the orchestrator already advertises in oInfo.CapabilitiesPrices. Resolve the fee per capability instead, keyed on req.Capability (added for metering by the stacked #3966): scan oInfo.CapabilitiesPrices for the Capability_BYOC entry whose Constraint matches the capability and charge that USD->wei/sec rate over compute-seconds (pixels = ceil(billableSecs)). This aligns the gateway's paid amount with the orchestrator's existing per-capability per-second accounting (byoc.JobPriceInfo * seconds). 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 orch 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 and preventing a false "price more than doubled" rejection when base and cap diverge. Gated behind a new default-OFF flag (-byocPerCapPricing / LivepeerNode.ByocPerCapPricing). When OFF, or when req.Capability is empty, or when no usable cap price matches, behavior is byte-identical to the base-price path (zero regression). resolveByocPrice is a pure, hermetically testable function; ticket params (faceValue/winProb) are passed through unchanged so EV/ticket math stays valid. Tests: TestResolveByocPrice (resolution, fallback, non-BYOC/zero-rate ignored) and TestGenerateLivePayment_ByocPerCapPricing (flag-off zero-regression, per-cap seconds fee, 2:1 tariff ratio, unknown-cap fallback, doubling-guard not tripped), both CGO-free. Co-authored-by: Cursor --- cmd/livepeer/starter/flags.go | 1 + cmd/livepeer/starter/starter.go | 6 + core/livepeernode.go | 1 + server/remote_signer.go | 68 ++++++++- server/remote_signer_test.go | 244 ++++++++++++++++++++++++++++++++ 5 files changed, 319 insertions(+), 1 deletion(-) diff --git a/cmd/livepeer/starter/flags.go b/cmd/livepeer/starter/flags.go index 1c66006891..c29022eb71 100644 --- a/cmd/livepeer/starter/flags.go +++ b/cmd/livepeer/starter/flags.go @@ -148,6 +148,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig { cfg.RemoteSignerWebhookURL = fs.String("remoteSignerWebhookUrl", *cfg.RemoteSignerWebhookURL, "Authentication webhook URL called by remote signer during GenerateLivePayment") cfg.RemoteSignerWebhookHeaders = fs.String("remoteSignerWebhookHeaders", *cfg.RemoteSignerWebhookHeaders, "Map of headers to use for remote signer webhook requests. e.g. 'header:val,header2:val2'") 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 9dcce9469a..87ac1ed74a 100755 --- a/cmd/livepeer/starter/starter.go +++ b/cmd/livepeer/starter/starter.go @@ -179,6 +179,7 @@ type LivepeerConfig struct { RemoteSignerWebhookURL *string RemoteSignerWebhookHeaders *string RemoteDiscovery *bool + ByocPerCapPricing *bool AIRunnerImage *string AIRunnerImageOverrides *string AIVerboseLogs *bool @@ -324,6 +325,7 @@ func DefaultLivepeerConfig() LivepeerConfig { defaultRemoteSignerWebhookURL := "" defaultRemoteSignerWebhookHeaders := "" defaultRemoteDiscovery := false + defaultByocPerCapPricing := false // Gateway logs defaultKafkaBootstrapServers := "" @@ -455,6 +457,7 @@ func DefaultLivepeerConfig() LivepeerConfig { RemoteSignerWebhookURL: &defaultRemoteSignerWebhookURL, RemoteSignerWebhookHeaders: &defaultRemoteSignerWebhookHeaders, RemoteDiscovery: &defaultRemoteDiscovery, + ByocPerCapPricing: &defaultByocPerCapPricing, // Gateway logs KafkaBootstrapServers: &defaultKafkaBootstrapServers, @@ -1878,6 +1881,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 c2678a98ca..8956334d4a 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" @@ -395,6 +396,43 @@ func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pip 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: } (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()) @@ -429,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 + } + } if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 { err := fmt.Errorf("missing or zero priceInfo") respondJsonError(ctx, w, err, http.StatusBadRequest) @@ -568,7 +624,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 { + // 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 504b736e80..37e006200c 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1778,3 +1778,247 @@ func TestResolveUsageLabels(t *testing.T) { }) } } + +// 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}, + }, + } + + 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: "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) +} From 84c706aef5f56d0446a3c907d5b98e1d9dc9a91a Mon Sep 17 00:00:00 2001 From: seanhanca Date: Tue, 30 Jun 2026 16:00:18 -0700 Subject: [PATCH 3/3] fix(signer): harden BYOC per-cap pricing gating + duplicate price scan Addresses Copilot review feedback on PR #3967: - resolveByocPrice now skips a matched-but-invalid (non-positive rate) entry and continues scanning instead of returning nil, so a later valid duplicate entry for the same constraint (e.g. from a misconfiguration) is still honored. Added a regression test case. (Copilot, remote_signer.go:431) - BYOC per-capability pricing is now additionally gated on Type==RemoteType_LiveVideoToVideo. Enabling it changes the billing basis (overrides req.InPixels and bypasses lv2v pixel synthesis), so it must only apply to the lv2v job type that BYOC jobs always use, never to a non-lv2v request that happens to carry a capability. (Copilot, remote_signer.go:488) --- server/remote_signer.go | 25 +++++++++++++++++-------- server/remote_signer_test.go | 10 ++++++++++ 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index 8956334d4a..305d5f54e9 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -408,10 +408,13 @@ func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pip // 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. +// 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 @@ -423,10 +426,11 @@ func resolveByocPrice(req *RemotePaymentRequest, oInfo *net.OrchestratorInfo) *n 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. + // 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 { - return nil + continue } return &net.PriceInfo{PricePerUnit: p.PricePerUnit, PixelsPerUnit: p.PixelsPerUnit} } @@ -479,7 +483,12 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req // byte-identical to the base-price path. priceInfo := oInfo.PriceInfo useByocPricing := false - if ls.LivepeerNode.ByocPerCapPricing && req.Capability != "" { + // Additionally gate on Type=="lv2v": enabling BYOC pricing changes the + // billing basis (it overrides req.InPixels and bypasses lv2v pixel + // synthesis), so it must only apply to the live (lv2v) job type that BYOC + // jobs always use — never to a non-lv2v request that happens to carry a + // capability. + if ls.LivepeerNode.ByocPerCapPricing && req.Capability != "" && req.Type == RemoteType_LiveVideoToVideo { if capPrice := resolveByocPrice(&req, &oInfo); capPrice != nil { priceInfo = capPrice oInfo.PriceInfo = capPrice diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 37e006200c..cd0580a850 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1799,6 +1799,10 @@ func TestResolveByocPrice(t *testing.T) { {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}, }, } @@ -1838,6 +1842,12 @@ func TestResolveByocPrice(t *testing.T) { 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",