feat(signer): bill BYOC live payments from CapabilitiesPrices

cmd/livepeer/starter/flags.go

@@ -149,6 +149,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	cfg.RemoteSignerWebhookHeaders = fs.String("remoteSignerWebhookHeaders", *cfg.RemoteSignerWebhookHeaders, "Map of headers to use for remote signer webhook requests. e.g. 'header:val,header2:val2'")
 	cfg.RemoteSignerAllowNoAuth = fs.Bool("remoteSignerAllowNoAuth", *cfg.RemoteSignerAllowNoAuth, "Allow an unauthenticated remote signer on a public -httpAddr (no webhook). UNSAFE: signs payments from this node's deposit for any reachable caller; restrict access externally (proxy/private network).")
 	cfg.RemoteDiscovery = fs.Bool("remoteDiscovery", *cfg.RemoteDiscovery, "Enable orchestrator discovery on remote signers")
+	cfg.ByocPerCapPricing = fs.Bool("byocPerCapPricing", *cfg.ByocPerCapPricing, "Remote signer: charge BYOC live payments from CapabilitiesPrices instead of the base price (default OFF)")
 
 	// Gateway metrics
 	cfg.KafkaBootstrapServers = fs.String("kafkaBootstrapServers", *cfg.KafkaBootstrapServers, "URL of Kafka Bootstrap Servers")

cmd/livepeer/starter/starter.go

@@ -180,6 +180,7 @@ type LivepeerConfig struct {
 	RemoteSignerWebhookHeaders *string
 	RemoteSignerAllowNoAuth    *bool
 	RemoteDiscovery            *bool
+	ByocPerCapPricing          *bool
 	AIRunnerImage              *string
 	AIRunnerImageOverrides     *string
 	AIVerboseLogs              *bool
@@ -326,6 +327,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	defaultRemoteSignerWebhookHeaders := ""
 	defaultRemoteSignerAllowNoAuth := false
 	defaultRemoteDiscovery := false
+	defaultByocPerCapPricing := false
 
 	// Gateway logs
 	defaultKafkaBootstrapServers := ""
@@ -458,6 +460,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		RemoteSignerWebhookHeaders: &defaultRemoteSignerWebhookHeaders,
 		RemoteSignerAllowNoAuth:    &defaultRemoteSignerAllowNoAuth,
 		RemoteDiscovery:            &defaultRemoteDiscovery,
+		ByocPerCapPricing:          &defaultByocPerCapPricing,
 
 		// Gateway logs
 		KafkaBootstrapServers: &defaultKafkaBootstrapServers,
@@ -1881,6 +1884,9 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	if cfg.RemoteDiscovery != nil {
 		n.RemoteDiscovery = *cfg.RemoteDiscovery
 	}
+	if cfg.ByocPerCapPricing != nil {
+		n.ByocPerCapPricing = *cfg.ByocPerCapPricing
+	}
 	if cfg.LiveAIHeartbeatHeaders != nil {
 		n.LiveAIHeartbeatHeaders = parseHeaderMap(*cfg.LiveAIHeartbeatHeaders)
 	}

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              // remote signer: use CapabilitiesPrices for BYOC (default OFF)
 
 	// Thread safety for config fields
 	mu                  sync.RWMutex

server/remote_discovery.go

@@ -391,15 +391,8 @@ func capabilityPrice(info *common.OrchNetworkCapabilities, capability core.Capab
 	if info == nil {
 		return nil
 	}
-	// Check per-capability price if it exists
-	for _, capPrice := range info.CapabilitiesPrices {
-		if capPrice == nil || capPrice.PixelsPerUnit <= 0 || core.Capability(capPrice.Capability) != capability {
-			continue
-		}
-		price := new(big.Rat).SetFrac64(capPrice.PricePerUnit, capPrice.PixelsPerUnit)
-		if capPrice.Constraint == modelID {
-			return price
-		}
+	if capPrice := findCapPriceInfo(info.CapabilitiesPrices, capability, modelID, false); capPrice != nil {
+		return new(big.Rat).SetFrac64(capPrice.PricePerUnit, capPrice.PixelsPerUnit)
 	}
 	// Global fallback if no per-capability price is available.
 	if info.PriceInfo == nil || info.PriceInfo.PixelsPerUnit <= 0 {

server/remote_signer.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"math/big"
 	"net/http"
 	"net/url"
@@ -355,6 +356,38 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS
 	return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason)
 }
 
+// findCapPriceInfo returns the first CapabilitiesPrices entry matching
+// capability+modelID. requirePositiveRate skips non-positive PricePerUnit and
+// keeps scanning; otherwise a zero rate is allowed.
+func findCapPriceInfo(prices []*net.PriceInfo, capability core.Capability, modelID string, requirePositiveRate bool) *net.PriceInfo {
+	for _, p := range prices {
+		if p == nil || core.Capability(p.Capability) != capability || p.Constraint != modelID {
+			continue
+		}
+		if p.PixelsPerUnit <= 0 {
+			continue
+		}
+		if requirePositiveRate && p.PricePerUnit <= 0 {
+			continue
+		}
+		return &net.PriceInfo{PricePerUnit: p.PricePerUnit, PixelsPerUnit: p.PixelsPerUnit}
+	}
+	return nil
+}
+
+// resolveByocPrice looks up the BYOC model constraint from caps in
+// oInfo.CapabilitiesPrices. Returns nil when no usable price matches.
+func resolveByocPrice(caps *core.Capabilities, oInfo *net.OrchestratorInfo) *net.PriceInfo {
+	if caps == nil || oInfo == nil {
+		return nil
+	}
+	constraint := caps.ModelIDForCapability(core.Capability_BYOC)
+	if constraint == "" {
+		return nil
+	}
+	return findCapPriceInfo(oInfo.CapabilitiesPrices, core.Capability_BYOC, constraint, true)
+}
+
 // GenerateLivePayment handles remote generation of a payment for live streams.
 func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Request) {
 	requestID := string(core.RandomManifestID())
@@ -389,14 +422,37 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
-	priceInfo := oInfo.PriceInfo
-	if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 {
-		err := fmt.Errorf("missing or zero priceInfo")
+	if oInfo.TicketParams == nil {
+		err := fmt.Errorf("missing ticketParams in OrchestratorInfo")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
-	if oInfo.TicketParams == nil {
-		err := fmt.Errorf("missing ticketParams in OrchestratorInfo")
+
+	var reqCaps *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
+		}
+		reqCaps = core.CapabilitiesFromNetCapabilities(&caps)
+	}
+
+	// BYOC caps: use per-capability price (and bill compute-seconds below).
+	// Otherwise keep base PriceInfo / lv2v pixel pricing. Write back so state,
+	// ExpectedPrice, and validatePrice all see the same rate.
+	priceInfo := oInfo.PriceInfo
+	useByocPricing := false
+	if ls.LivepeerNode.ByocPerCapPricing {
+		if capPrice := resolveByocPrice(reqCaps, &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)
 		return
 	}
@@ -454,16 +510,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 
 	streamParams := &core.StreamParameters{
 		// 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)
+		ManifestID:   core.ManifestID(manifestID),
+		Capabilities: reqCaps,
 	}
 
 	pmParams := pmTicketParams(oInfo.TicketParams)
@@ -528,7 +576,13 @@ 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 prices are per compute-second; bill seconds instead of lv2v pixels.
+		if billableSecs <= 0 {
+			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