From 8ddd08ea34f6c14870f64826db2dd582a7172d7d Mon Sep 17 00:00:00 2001 From: qianghan Date: Fri, 10 Jul 2026 16:41:47 -0700 Subject: [PATCH] fix(byoc): restore type:byoc billing and V1 job-creds verify 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 c78afcc0f3..9eb3565afe 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") }