From 0c75f6969807f1d043dc6c5c8edfd62e52ce05ba Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Thu, 11 Jun 2026 10:10:47 -0400 Subject: [PATCH 01/14] feat(ai_mediaserver): Make RTMP output optional in AI media server (#3946) - Introduced a new function `LiveAIDisableRTMPOutput` to determine if RTMP egress should be skipped based on the environment variable `LIVE_AI_DISABLE_RTMP_OUTPUT`. - Updated the RTMP output collection logic in `setupStream` and `StartLiveVideo` methods to conditionally append outputs based on the new function. - Improved logging to indicate whether RTMP output is disabled, providing better visibility into the streaming configuration. These changes optimize the handling of RTMP outputs, particularly for WHEP scenarios that utilize an in-memory ring buffer instead. --- byoc/stream_gateway.go | 6 +++--- media/whip_server.go | 10 ++++++++++ server/ai_mediaserver.go | 30 +++++++++++++++++------------- 3 files changed, 30 insertions(+), 16 deletions(-) diff --git a/byoc/stream_gateway.go b/byoc/stream_gateway.go index 6aba915c15..cdac9cd99e 100644 --- a/byoc/stream_gateway.go +++ b/byoc/stream_gateway.go @@ -567,9 +567,9 @@ func (bsg *BYOCGatewayServer) setupStream(ctx context.Context, r *http.Request, ctx = clog.AddVal(ctx, "stream_id", streamID) clog.Infof(ctx, "Received live video AI request") - // collect all RTMP outputs + // collect all RTMP outputs (optional; WHEP uses the ring buffer instead) var rtmpOutputs []string - if job.Job.Params.EnableVideoEgress { + if job.Job.Params.EnableVideoEgress && !media.LiveAIDisableRTMPOutput() { if outputURL != "" { rtmpOutputs = append(rtmpOutputs, outputURL) } @@ -578,7 +578,7 @@ func (bsg *BYOCGatewayServer) setupStream(ctx context.Context, r *http.Request, } } - clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs) + clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs, "disabled", media.LiveAIDisableRTMPOutput()) // Clear any previous gateway status bsg.statusStore.Clear(streamID) diff --git a/media/whip_server.go b/media/whip_server.go index 8c4eecb372..bf608587f5 100644 --- a/media/whip_server.go +++ b/media/whip_server.go @@ -576,6 +576,16 @@ func disableTSCorrection() bool { return v } +// LiveAIDisableRTMPOutput skips ffmpeg RTMP egress. WHEP reads processed video from +// the in-memory ring buffer and does not require RTMP outputs. +func LiveAIDisableRTMPOutput() bool { + v, err := strconv.ParseBool(os.Getenv("LIVE_AI_DISABLE_RTMP_OUTPUT")) + if err != nil { + return false + } + return v +} + func getUDPListenerAddr(addrStr string) (*net.UDPAddr, error) { if addrStr == "" { // Default to all interfaces, port 7290 diff --git a/server/ai_mediaserver.go b/server/ai_mediaserver.go index 36173e7afc..8b80e4e8e7 100644 --- a/server/ai_mediaserver.go +++ b/server/ai_mediaserver.go @@ -567,15 +567,17 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler { ctx = clog.AddVal(ctx, "stream_id", streamID) clog.Infof(ctx, "Received live video AI request for %s. pipelineParams=%v", streamName, pipelineParams) - // collect all RTMP outputs + // collect all RTMP outputs (optional; WHEP uses the ring buffer instead) var rtmpOutputs []string - if outputURL != "" { - rtmpOutputs = append(rtmpOutputs, outputURL) - } - if mediaMTXOutputURL != "" { - rtmpOutputs = append(rtmpOutputs, mediaMTXOutputURL, mediaMTXOutputAlias) + if !media.LiveAIDisableRTMPOutput() { + if outputURL != "" { + rtmpOutputs = append(rtmpOutputs, outputURL) + } + if mediaMTXOutputURL != "" { + rtmpOutputs = append(rtmpOutputs, mediaMTXOutputURL, mediaMTXOutputAlias) + } } - clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs) + clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs, "disabled", media.LiveAIDisableRTMPOutput()) // channel that blocks until after orch selection is complete // avoids a race condition with closing the control channel @@ -1107,13 +1109,15 @@ func (ls *LivepeerServer) CreateWhip(server *media.WHIPServer) http.Handler { }) }() - if outputURL != "" { - rtmpOutputs = append(rtmpOutputs, outputURL) - } - if mediamtxOutputURL != "" { - rtmpOutputs = append(rtmpOutputs, mediamtxOutputURL, mediaMTXOutputAlias) + if !media.LiveAIDisableRTMPOutput() { + if outputURL != "" { + rtmpOutputs = append(rtmpOutputs, outputURL) + } + if mediamtxOutputURL != "" { + rtmpOutputs = append(rtmpOutputs, mediamtxOutputURL, mediaMTXOutputAlias) + } } - clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs) + clog.Info(ctx, "RTMP outputs", "destinations", rtmpOutputs, "disabled", media.LiveAIDisableRTMPOutput()) params := aiRequestParams{ node: ls.LivepeerNode, From cb51ef8db82a59406ec5306ed3856fba3622ba28 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Mon, 22 Jun 2026 20:22:01 +0100 Subject: [PATCH 02/14] docs(changelog): Document ticket params zero-expiration fix (#3956) Record the security fix from a6c4b1e in CHANGELOG_PENDING.md. The merge commit ("Merge commit from fork") did not document the user-facing change: validateTicketParams returned early on ExpirationBlock == 0, skipping the economic caps; it now rejects such tickets. --- CHANGELOG_PENDING.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index 3e3ac497c9..c96ba10adc 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -18,4 +18,8 @@ #### General +#### Broadcaster + +* [a6c4b1e](https://github.com/livepeer/go-livepeer/commit/a6c4b1ef70d8f4d3da0e7d8164ac8d1faf80ad0e) pm: Reject ticket params with a zero expiration block so they are subjected to the economic caps (@rickstaa) + #### CLI From 8b9a45083d46e159da685b22a02f917e5e391e48 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Tue, 23 Jun 2026 08:50:25 +0100 Subject: [PATCH 03/14] release: v0.8.11 (#3957) Bump VERSION to 0.8.11 and document the PRs merged since v0.8.10 in the changelog. --- CHANGELOG.md | 26 ++++++++++++++++++++++++++ VERSION | 2 +- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9fc519e26..e07e3b7209 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +## v0.8.11 + +### Features ⚒ + +#### General + +* [#3889](https://github.com/livepeer/go-livepeer/pull/3889) Remove x11 from MacOS builds (@j0sh) +* [#3925](https://github.com/livepeer/go-livepeer/pull/3925) trickle: Add a //next endpoint (@rickstaa) +* [#3882](https://github.com/livepeer/go-livepeer/pull/3882) box: add webcam input and fix bash conditionals (@rickstaa) + +#### AI – LV2V + +* [#3946](https://github.com/livepeer/go-livepeer/pull/3946) ai/live: Make RTMP output optional in AI media server (@eliteprox) +* [#3897](https://github.com/livepeer/go-livepeer/pull/3897) signer: Add auth webhook callback (@j0sh) +* [#3877](https://github.com/livepeer/go-livepeer/pull/3877) ai/live: Send Kafka event for tracking payments (@j0sh) + +#### AI – BYOC + +* [#3878](https://github.com/livepeer/go-livepeer/pull/3878) byoc: add auth token to byoc worker registration (@ad-astra-video) + +### Bug Fixes 🐞 + +#### Broadcaster + +* [a6c4b1e](https://github.com/livepeer/go-livepeer/commit/a6c4b1ef70d8f4d3da0e7d8164ac8d1faf80ad0e) pm: Reject ticket params with a zero expiration block so they are subjected to the economic caps (@rickstaa) + ## v0.8.10 #### General diff --git a/VERSION b/VERSION index e6663d4cee..83ce05d72f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.8.10 \ No newline at end of file +0.8.11 From 2be78dbca1ca3b856826b73f4892bfd4ef75c8c9 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Sat, 13 Jun 2026 02:18:42 -0400 Subject: [PATCH 04/14] feat(capabilities): Add ModelIDForCapability function and corresponding tests - Implemented the ModelIDForCapability method in the Capabilities struct to return the constrained model ID for a given capability, ensuring deterministic results when multiple models are present. - Added unit tests for ModelIDForCapability to cover various scenarios, including nil receivers, no constraints, single constrained models, and handling of ambiguous model cases. - Updated the remote signer to include the model ID in the event payload for LiveVideoToVideo capabilities. --- core/capabilities.go | 25 ++++++++++++++++++ core/capabilities_test.go | 54 +++++++++++++++++++++++++++++++++++++++ server/remote_signer.go | 3 +++ 3 files changed, 82 insertions(+) diff --git a/core/capabilities.go b/core/capabilities.go index e1ecd62576..6bef2c5480 100644 --- a/core/capabilities.go +++ b/core/capabilities.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "sync" "github.com/Masterminds/semver/v3" @@ -791,6 +792,30 @@ func (bcast *Capabilities) PerCapability() PerCapabilityConstraints { return nil } +// ModelIDForCapability returns the constrained model ID for the given capability. +// When multiple models are constrained it returns the lexicographically first one +// so the result is deterministic. Returns "" when no model is constrained. +func (bcast *Capabilities) ModelIDForCapability(cap Capability) string { + if bcast == nil { + return "" + } + capConstraints, ok := bcast.constraints.perCapability[cap] + if !ok || capConstraints == nil { + return "" + } + modelIDs := make([]string, 0, len(capConstraints.Models)) + for modelID := range capConstraints.Models { + if modelID != "" { + modelIDs = append(modelIDs, modelID) + } + } + if len(modelIDs) == 0 { + return "" + } + sort.Strings(modelIDs) + return modelIDs[0] +} + func (bcast *Capabilities) SetMinVersionConstraint(minVersionConstraint string) { if bcast != nil { bcast.constraints.minVersion = minVersionConstraint diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 59f45b0582..5eba55217c 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -751,6 +751,60 @@ func TestMinRunnerVersion(t *testing.T) { assert.Equal("", c.MinRunnerVersionConstraint(Capability_LiveVideoToVideo, "other")) } +func TestModelIDForCapability(t *testing.T) { + assert := assert.New(t) + + // nil receiver is safe + var nilCaps *Capabilities + assert.Equal("", nilCaps.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // no constraints for the capability + c := NewCapabilities([]Capability{Capability_LiveVideoToVideo}, nil) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // single constrained model + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{"streamdiffusion": &ModelConstraint{Warm: true}}, + }, + }) + assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // different capability has no constraint + assert.Equal("", c.ModelIDForCapability(Capability_TextToImage)) + + // empty model IDs are ignored + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{"": &ModelConstraint{}}, + }, + }) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // one real model plus empty key resolves to the real model + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "": &ModelConstraint{}, + "streamdiffusion": &ModelConstraint{Warm: true}, + }, + }, + }) + assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // multiple models are ambiguous, so no model is returned + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "comfyui": &ModelConstraint{}, + "streamdiffusion": &ModelConstraint{}, + "noop": &ModelConstraint{}, + }, + }, + }) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) +} + func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) { // the capability should be added by AddCapacity for modelID, modelConstraint := range constraint.Models { diff --git a/server/remote_signer.go b/server/remote_signer.go index 537f30978b..4ba9a61612 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -663,14 +663,17 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req sessionStatus = "new" } pipeline := "" + modelID := "" if req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo + modelID = streamParams.Capabilities.ModelIDForCapability(core.Capability_LiveVideoToVideo) } // NB: This could could drop events if tha Kafka queue is full! monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{ "session_id": state.StateID, "session_status": sessionStatus, "pipeline": pipeline, + "model_id": modelID, "request_id": requestID, "orch_address": orchAddr.Hex(), "orch_url": oInfo.Transcoder, From 6bad18350f9171353156e050a15bebe8d7ed00d2 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Thu, 25 Jun 2026 17:16:09 -0400 Subject: [PATCH 05/14] fix(capabilities): Update ModelIDForCapability test for lexicographical resolution - Modified the test for ModelIDForCapability to reflect that multiple models resolve to the lexicographically first ID instead of returning an empty string. - Adjusted the expected output in the test case to return "comfyui" for the Capability_LiveVideoToVideo capability. --- core/capabilities_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 5eba55217c..7f602dd98b 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -792,7 +792,7 @@ func TestModelIDForCapability(t *testing.T) { }) assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) - // multiple models are ambiguous, so no model is returned + // multiple models resolve to the lexicographically first ID c.SetPerCapabilityConstraints(PerCapabilityConstraints{ Capability_LiveVideoToVideo: &CapabilityConstraints{ Models: ModelConstraints{ @@ -802,7 +802,7 @@ func TestModelIDForCapability(t *testing.T) { }, }, }) - assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + assert.Equal("comfyui", c.ModelIDForCapability(Capability_LiveVideoToVideo)) } func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) { From 7b220bbfceb4d8aee758e6833cdc403e3115b9d0 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Thu, 25 Jun 2026 17:23:36 -0400 Subject: [PATCH 06/14] ci: trigger docker build for PR #3947 From ddb4866453a799c3ebbf0eabd3a1943602a56ba6 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 26 Jun 2026 13:30:44 -0400 Subject: [PATCH 07/14] Enhance remote signer payment handling with auth ID support - Updated the payment processing logic to include an optional `AuthID` in the `RemotePaymentState` and `authResponse` structures. - Modified the `doPaymentWithStateAndAuthID` function to accept and set the `AuthID` from the request header or callback response. - Added tests to verify the correct handling of `AuthID` in various scenarios, including fallback behavior and error handling when the `AuthID` changes unexpectedly. - Ensured that the `AuthID` is included in the response payload for better tracking and debugging. --- server/remote_signer.go | 18 ++++++ server/remote_signer_test.go | 114 ++++++++++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 2 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index 4ba9a61612..7779651555 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -34,6 +34,7 @@ const HTTPStatusNoTickets = 482 const RefreshSessionOrchestratorURLHeader = "Livepeer-Orchestrator-URL" const RemoteType_LiveVideoToVideo = "lv2v" const PipelineLiveVideoToVideo = "live-video-to-video" +const remoteSignerAuthIDHeader = "Signer-Auth-Id" // SignOrchestratorInfo handles signing GetOrchestratorInfo requests for multiple orchestrators func (ls *LivepeerServer) SignOrchestratorInfo(w http.ResponseWriter, r *http.Request) { @@ -218,6 +219,7 @@ type RemotePaymentState struct { InitialPricePerUnit int64 InitialPixelsPerUnit int64 SequenceNumber uint64 + AuthID string } type RemotePaymentStateSig struct { @@ -267,6 +269,8 @@ type authResponse struct { // Unix timestamp (seconds) until which auth is considered valid. // Allows for skipping webhook callbacks until this time is exceeded. Expiry int64 `json:"expiry,omitempty"` + // Optional opaque identifier for metering attribution. + AuthID string `json:"auth_id,omitempty"` } // Signs the serialized state with the remote signer's Ethereum key. @@ -639,6 +643,19 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req if callbackResp != nil { state.AuthExpiry = callbackResp.Expiry } + authID := r.Header.Get(remoteSignerAuthIDHeader) + if callbackResp != nil && callbackResp.AuthID != "" { + authID = callbackResp.AuthID + } + if authID != "" && state.AuthID != authID { + if state.AuthID != "" { + clog.Warningf(ctx, "Remote signer auth ID changed oldAuthID=%s newAuthID=%s", state.AuthID, authID) + respondJsonError(ctx, w, errors.New("remote signer auth ID changed"), http.StatusInternalServerError) + return + } + state.AuthID = authID + } + ctx = clog.AddVal(ctx, "auth_id", state.AuthID) // Encode and sign updated state stateBytes, err := json.Marshal(state) @@ -690,6 +707,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req "cost_per_pixel": orchPrice.FloatString(10), "sequence_number": state.SequenceNumber, "num_tickets": balUpdate.NumTickets, + "auth_id": state.AuthID, }) } diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 3be1e5e837..9e2d7a154a 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -701,7 +701,7 @@ func TestGenerateLivePayment_WebhookCallback(t *testing.T) { orchBlob, err := proto.Marshal(oInfo) require.NoError(err) - doPaymentWithState := func(requestHeader string, state RemotePaymentStateSig) *httptest.ResponseRecorder { + doPaymentWithStateAndAuthID := func(requestHeader string, authID string, state RemotePaymentStateSig) *httptest.ResponseRecorder { reqBody, err := json.Marshal(RemotePaymentRequest{ Orchestrator: orchBlob, ManifestID: "manifest", @@ -712,10 +712,19 @@ func TestGenerateLivePayment_WebhookCallback(t *testing.T) { req := httptest.NewRequest(http.MethodPost, "/generate-live-payment", bytes.NewReader(reqBody)) req.Header.Set("X-Request-ID", requestHeader) + if authID != "" { + req.Header.Set(remoteSignerAuthIDHeader, authID) + } rr := httptest.NewRecorder() ls.GenerateLivePayment(rr, req) return rr } + doPaymentWithState := func(requestHeader string, state RemotePaymentStateSig) *httptest.ResponseRecorder { + return doPaymentWithStateAndAuthID(requestHeader, "", state) + } + doPaymentWithAuthID := func(requestHeader string, authID string) *httptest.ResponseRecorder { + return doPaymentWithStateAndAuthID(requestHeader, authID, RemotePaymentStateSig{}) + } doPayment := func(requestHeader string) *httptest.ResponseRecorder { return doPaymentWithState(requestHeader, RemotePaymentStateSig{}) } @@ -800,13 +809,67 @@ func TestGenerateLivePayment_WebhookCallback(t *testing.T) { require.Equal(expiry, state.AuthExpiry) }) + t.Run("callback auth id sets state", func(t *testing.T) { + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":200,"auth_id":"webhook-auth-id"}`)) + })) + defer webhook.Close() + + webhookURL, err := url.Parse(webhook.URL) + require.NoError(err) + ls.LivepeerNode.RemoteSignerWebhookURL = webhookURL + ls.LivepeerNode.RemoteSignerWebhookHeaders = nil + + rr := doPayment("req-auth-id") + require.Equal(http.StatusOK, rr.Code) + state := parseResponseState(rr) + require.Equal("webhook-auth-id", state.AuthID) + }) + + t.Run("request auth id header is fallback when callback omits auth id", func(t *testing.T) { + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":200}`)) + })) + defer webhook.Close() + + webhookURL, err := url.Parse(webhook.URL) + require.NoError(err) + ls.LivepeerNode.RemoteSignerWebhookURL = webhookURL + ls.LivepeerNode.RemoteSignerWebhookHeaders = nil + + rr := doPaymentWithAuthID("req-auth-id-fallback", "header-auth-id") + require.Equal(http.StatusOK, rr.Code) + state := parseResponseState(rr) + require.Equal("header-auth-id", state.AuthID) + }) + + t.Run("callback auth id wins over request header", func(t *testing.T) { + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":200,"auth_id":"webhook-auth-id"}`)) + })) + defer webhook.Close() + + webhookURL, err := url.Parse(webhook.URL) + require.NoError(err) + ls.LivepeerNode.RemoteSignerWebhookURL = webhookURL + ls.LivepeerNode.RemoteSignerWebhookHeaders = nil + + rr := doPaymentWithAuthID("req-auth-id-priority", "header-auth-id") + require.Equal(http.StatusOK, rr.Code) + state := parseResponseState(rr) + require.Equal("webhook-auth-id", state.AuthID) + }) + t.Run("sequential requests skip callback while auth expiry still valid", func(t *testing.T) { callbackCalls := 0 expiry := time.Now().Add(5 * time.Minute).Unix() webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { callbackCalls++ w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte(fmt.Sprintf(`{"status":200,"expiry":%d}`, expiry))) + _, _ = w.Write([]byte(fmt.Sprintf(`{"status":200,"expiry":%d,"auth_id":"cached-auth-id"}`, expiry))) })) defer webhook.Close() @@ -820,8 +883,24 @@ func TestGenerateLivePayment_WebhookCallback(t *testing.T) { var firstResp RemotePaymentResponse require.NoError(json.NewDecoder(first.Body).Decode(&firstResp)) + var firstState RemotePaymentState + require.NoError(json.Unmarshal(firstResp.State.State, &firstState)) + require.Equal("cached-auth-id", firstState.AuthID) + second := doPaymentWithState("req-skip-second", firstResp.State) require.Equal(http.StatusOK, second.Code) + var secondResp RemotePaymentResponse + require.NoError(json.NewDecoder(second.Body).Decode(&secondResp)) + var secondState RemotePaymentState + require.NoError(json.Unmarshal(secondResp.State.State, &secondState)) + require.Equal("cached-auth-id", secondState.AuthID) + require.Equal(1, callbackCalls) + + third := doPaymentWithStateAndAuthID("req-skip-third", "new-header-auth-id", secondResp.State) + require.Equal(http.StatusInternalServerError, third.Code) + var apiErr apiErrorResponse + require.NoError(json.NewDecoder(third.Body).Decode(&apiErr)) + require.Equal("Internal Server Error", apiErr.Error.Message) require.Equal(1, callbackCalls) }) @@ -850,6 +929,37 @@ func TestGenerateLivePayment_WebhookCallback(t *testing.T) { require.Equal(2, callbackCalls) }) + t.Run("callback auth id change returns 500", func(t *testing.T) { + webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":200,"auth_id":"updated-auth-id"}`)) + })) + defer webhook.Close() + + webhookURL, err := url.Parse(webhook.URL) + require.NoError(err) + ls.LivepeerNode.RemoteSignerWebhookURL = webhookURL + ls.LivepeerNode.RemoteSignerWebhookHeaders = nil + + initialState := RemotePaymentState{ + StateID: string(core.RandomManifestID()), + OrchestratorAddress: ethcommon.BytesToAddress(oInfo.Address), + InitialPricePerUnit: oInfo.PriceInfo.PricePerUnit, + InitialPixelsPerUnit: oInfo.PriceInfo.PixelsPerUnit, + AuthID: "cached-auth-id", + } + stateBytes, err := json.Marshal(initialState) + require.NoError(err) + stateSig, err := signState(ls, stateBytes) + require.NoError(err) + + rr := doPaymentWithState("req-auth-id-replaced", RemotePaymentStateSig{State: stateBytes, Sig: stateSig}) + require.Equal(http.StatusInternalServerError, rr.Code) + var apiErr apiErrorResponse + require.NoError(json.NewDecoder(rr.Body).Decode(&apiErr)) + require.Equal("Internal Server Error", apiErr.Error.Message) + }) + t.Run("callback 200 missing status returns 500", func(t *testing.T) { webhook := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) From 82844a920040965059dd51085acc3bcf7d64dd03 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 26 Jun 2026 22:15:20 -0400 Subject: [PATCH 08/14] ci(docker): rebuild PR images without shared registry cache Shared livepeerci/build:cache caused PR Docker tags (pr-3947, feat-add-model-id-signer-kafka) to point at an older livepeer binary while metadata listed the PR head SHA. Skip registry cache on pull_request builds and pass GIT_REVISION to bust the compile layer. --- .github/workflows/docker.yaml | 7 +++++-- docker/Dockerfile | 3 +++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 5ecfc8379c..56813d0d07 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -82,6 +82,7 @@ jobs: with: build-args: | BUILD_TAGS=${{ steps.build-tag.outputs.build-tags }} + GIT_REVISION=${{ github.event.pull_request.head.sha || github.sha }} context: . provenance: mode=max sbom: true @@ -92,8 +93,10 @@ jobs: tags: ${{ steps.meta.outputs.tags }} annotations: ${{ steps.meta.outputs.annotations }} labels: ${{ steps.meta.outputs.labels }} - cache-from: type=registry,ref=livepeerci/build:cache - cache-to: type=registry,ref=livepeerci/build:cache,mode=max + # PR builds must not reuse the shared registry cache — it can serve a + # stale livepeer binary while metadata tags the PR head SHA (see #3947). + cache-from: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache' || '' }} + cache-to: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache,mode=max' || '' }} builder: name: go-livepeer builder docker image generation diff --git a/docker/Dockerfile b/docker/Dockerfile index 6ed4a49705..00f799c065 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -60,6 +60,9 @@ COPY ./install_ffmpeg.sh ./install_ffmpeg.sh ARG BUILD_TAGS ENV BUILD_TAGS=${BUILD_TAGS} +ARG GIT_REVISION=unknown +RUN echo "Building livepeer at ${GIT_REVISION}" + COPY go.mod go.sum ./ RUN go mod download From 023d25e9eceffd63aff33e7072cc0ab9c38a5f99 Mon Sep 17 00:00:00 2001 From: seanhanca Date: Fri, 26 Jun 2026 20:21:15 -0700 Subject: [PATCH 09/14] 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 10/14] 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 11/14] 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", From cae4e7316c19a91db1dae361056b654f10bd6710 Mon Sep 17 00:00:00 2001 From: seanhanca Date: Tue, 30 Jun 2026 15:56:41 -0700 Subject: [PATCH 12/14] 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 305d5f54e9..c78afcc0f3 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -370,28 +370,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 cd0580a850..f6e627194f 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") From 410aeebc47521daecd2d40bd1bd46807ed942e0d Mon Sep 17 00:00:00 2001 From: seanhanca <103605970+seanhanca@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:13:06 -0700 Subject: [PATCH 13/14] fix(byoc): restore type:byoc billing and V1 job-creds verify (#3980) Align staging orch with DMZ FlattenBYOCJob signing and accept explicit type:byoc on /generate-live-payment alongside the existing lv2v path. Co-authored-by: Cursor --- byoc/job_orchestrator.go | 28 +++++++++------ server/remote_signer.go | 69 ++++++++++++++++++++++++++---------- server/remote_signer_test.go | 51 ++++++++++++++++++++++++++ 3 files changed, 120 insertions(+), 28 deletions(-) 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/server/remote_signer.go b/server/remote_signer.go index 32f93d525e..c6ea711f71 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -34,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" @@ -243,7 +244,7 @@ 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. @@ -407,7 +408,15 @@ func sanitizeUsageLabel(s string) string { func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pipeline, modelID string) { if req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo - modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo) + 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 = m + modelID = m + } } if c := sanitizeUsageLabel(req.Capability); c != "" { pipeline = c @@ -418,6 +427,23 @@ func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pip 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 @@ -505,13 +531,26 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req // byte-identical to the base-price path. priceInfo := oInfo.PriceInfo useByocPricing := false - // 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 { + 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 @@ -583,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) @@ -655,7 +688,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req lastUpdate = now } billableSecs := now.Sub(lastUpdate).Seconds() - if useByocPricing { + 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 diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index f6e627194f..b333d552f3 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1708,6 +1708,25 @@ func lv2vCapsWithModel(t *testing.T, modelID string) *core.Capabilities { 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); @@ -1738,6 +1757,15 @@ func TestResolveUsageLabels(t *testing.T) { wantPipeline: PipelineLiveVideoToVideo, wantModelID: "streamdiffusion", }, + { + name: "byoc type: labels from capabilities protobuf", + req: RemotePaymentRequest{ + Type: RemoteType_BYOC, + }, + caps: byocCapsWithModel(t, "flux-schnell"), + wantPipeline: "flux-schnell", + wantModelID: "flux-schnell", + }, { name: "byoc: real capability + model override lv2v defaults", req: RemotePaymentRequest{ @@ -2053,4 +2081,27 @@ func TestGenerateLivePayment_ByocPerCapPricing(t *testing.T) { 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") } From 34038ffa7a46e8592a4fa6ff9c587a590feec272 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 10 Jul 2026 22:24:06 -0400 Subject: [PATCH 14/14] fix(remote_signer): update pipeline resolution for BYOC capability Changed the pipeline assignment in the remote signer to use the new capability-to-pipeline conversion function, ensuring that the correct pipeline is returned for BYOC capabilities. Updated the corresponding test to reflect this change in expected output. --- server/remote_signer.go | 2 +- server/remote_signer_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index c6ea711f71..8713116bce 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -414,7 +414,7 @@ func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pip } if req.Type == RemoteType_BYOC && caps != nil { if m := caps.ModelIDForCapability(core.Capability_BYOC); m != "" { - pipeline = m + pipeline = core.CapabilityToPipeline(core.Capability_BYOC) modelID = m } } diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index b333d552f3..99a0b65a33 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1763,7 +1763,7 @@ func TestResolveUsageLabels(t *testing.T) { Type: RemoteType_BYOC, }, caps: byocCapsWithModel(t, "flux-schnell"), - wantPipeline: "flux-schnell", + wantPipeline: "byoc", wantModelID: "flux-schnell", }, {