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 692f79783ad3a825025feeb297b74cba36da50bc Mon Sep 17 00:00:00 2001 From: seanhanca Date: Fri, 26 Jun 2026 21:06:37 -0700 Subject: [PATCH 2/3] ci: re-trigger build (transient zlib.net/gpg infra flake) From f7d2f83d611da37f51a7b5ffbaa3a06866b62bbe Mon Sep 17 00:00:00 2001 From: seanhanca Date: Tue, 30 Jun 2026 15:56:41 -0700 Subject: [PATCH 3/3] fix(signer): sanitize gateway usage labels + correct docstring/test Addresses Copilot review feedback on PR #3966: - resolveUsageLabels now sanitizes the gateway-supplied capability/model_id override labels (TrimSpace + cap at 128 runes via sanitizeUsageLabel) before they are emitted to the create_signed_ticket usage event, bounding Kafka payload size and downstream label cardinality. req.Capability/req.ModelID come from the request body, so an unbounded value could otherwise inflate cardinality (pipeline was previously a small fixed set). - Corrected the resolveUsageLabels docstring: the capability/model_id overrides apply regardless of Type, not only for lv2v; clarified the empty-field fallback wording. - TestResolveUsageLabels now constructs a fresh require.New(t) inside each subtest so failures are attributed to the correct subtest, and adds cases for whitespace-only (ignored) and oversized (length-capped) override labels. --- server/remote_signer.go | 46 ++++++++++++++++++++++++++---------- server/remote_signer_test.go | 26 ++++++++++++++++++-- 2 files changed, 58 insertions(+), 14 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index c2678a98ca..3a317ff56e 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -369,28 +369,50 @@ 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 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. +// 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 modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo) } - if req.Capability != "" { - pipeline = req.Capability + if c := sanitizeUsageLabel(req.Capability); c != "" { + pipeline = c } - if req.ModelID != "" { - modelID = req.ModelID + if m := sanitizeUsageLabel(req.ModelID); m != "" { + modelID = m } return pipeline, modelID } diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 504b736e80..4cebec7d80 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" @@ -1716,8 +1717,6 @@ func lv2vCapsWithModel(t *testing.T, modelID string) *core.Capabilities { // 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 @@ -1767,10 +1766,33 @@ func TestResolveUsageLabels(t *testing.T) { 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")