From 10ba11a69f85f7a818c3b4042816f96d4050c0d5 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Thu, 23 Jul 2026 21:18:45 -0700 Subject: [PATCH 1/2] runner: Add fixed pricing Add `fixed` pricing for live runners: a single payment charged when a request or session is created, rather than payment metered over time or processed pixels. Support fixed prices in dynamic and static registrations, discovery, remote signers, and orchestrator accounting, documentation and tests. --- ai/runner/live_runner.go | 64 ++++++++--- ai/runner/live_runner_test.go | 9 ++ doc/live-runner.md | 16 +-- doc/remote-signer.md | 2 +- server/ai_http.go | 198 +++++++++++++++++++++------------- server/ai_http_test.go | 126 ++++++++++++++++++++-- server/remote_discovery.go | 2 +- server/remote_signer.go | 7 +- server/remote_signer_test.go | 98 +++++++++++++++++ 9 files changed, 417 insertions(+), 105 deletions(-) diff --git a/ai/runner/live_runner.go b/ai/runner/live_runner.go index 77949eb41d..d9fd186fb5 100644 --- a/ai/runner/live_runner.go +++ b/ai/runner/live_runner.go @@ -115,8 +115,9 @@ func normalizeLiveRunnerPriceInfo(priceInfo LiveRunnerPriceInfo) (LiveRunnerPric switch unit { case "hour": case "720p": + case "fixed": default: - return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.unit must be hour or 720p") + return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.unit must be hour, 720p, or fixed") } priceInfo.Currency = currency priceInfo.Unit = unit @@ -243,6 +244,7 @@ type liveRunner struct { type liveRunnerSession struct { // Protected by the parent liveRunner.mu. createdAt time.Time + priceInfo LiveRunnerPriceInfo channels map[string]*liveRunnerTrickleChannel proxies map[string]proxyTarget } @@ -957,9 +959,6 @@ func (r *LiveRunnerRegistry) ReserveSession(runnerID string, optSessionID ...str if !isReadyStatus(runner.Status) { return "", "", &RunnerError{StatusCode: http.StatusNotFound, Message: "runner not found"} } - if len(runner.sessions) >= runner.Capacity { - return "", "", &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: "no capacity available for runner"} - } id := "" if len(optSessionID) > 0 { // use supplied id if given - usually ticket param's auth token id / manifest id @@ -967,9 +966,14 @@ func (r *LiveRunnerRegistry) ReserveSession(runnerID string, optSessionID ...str } if id != "" { if _, exists := runner.sessions[id]; exists { + // Consider another status code if session enumeration becomes a concern. return "", "", &RunnerError{StatusCode: http.StatusConflict, Message: "session already exists"} } - } else { + } + if len(runner.sessions) >= runner.Capacity { + return "", "", &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: "no capacity available for runner"} + } + if id == "" { id = "session_" + randomStr(liveRunnerIDRandomBytes) for { if _, exists := runner.sessions[id]; !exists { @@ -980,6 +984,7 @@ func (r *LiveRunnerRegistry) ReserveSession(runnerID string, optSessionID ...str } session := &liveRunnerSession{ createdAt: time.Now(), + priceInfo: runner.PriceInfo, channels: make(map[string]*liveRunnerTrickleChannel), proxies: make(map[string]proxyTarget), } @@ -1015,14 +1020,27 @@ func (r *LiveRunnerRegistry) PaymentInfo(runnerID string) (*LiveRunnerPriceInfo, } priceInfo, err := runner.convertPrice() if err != nil { - return nil, err + return nil, &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: fmt.Sprintf("live runner price unavailable: %v", err)} } if _, err := priceInfo.priceRat(); err != nil { - return nil, nil + return nil, &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: fmt.Sprintf("live runner price unavailable: %v", err)} } return &priceInfo, nil } +func (r *LiveRunnerRegistry) SessionPriceInfo(runnerID, sessionID string) (LiveRunnerPriceInfo, error) { + runner, unlock, err := r.lockLiveRunner(runnerID) + if err != nil { + return LiveRunnerPriceInfo{}, err + } + defer unlock() + session, err := runner.liveRunnerSessionLocked(sessionID) + if err != nil { + return LiveRunnerPriceInfo{}, err + } + return session.priceInfo, nil +} + func (r *LiveRunnerRegistry) ReleaseSession(runnerID, sessionID string) error { runner, unlock, err := r.lockLiveRunner(runnerID) if err != nil { @@ -1753,15 +1771,19 @@ func newConverterForRunner(priceInfo LiveRunnerPriceInfo) (*core.AutoConvertedPr return nil, err } - usdPerHour, err := priceInfo.priceRat() + usdPrice, err := priceInfo.priceRat() if err != nil { return nil, err } - denom := int64(3600) - if priceInfo.Unit == "720p" { - denom *= 1280 * 720 * 30 // 720p @ 30fps + switch priceInfo.Unit { + case "hour": + usdPrice = new(big.Rat).Quo(usdPrice, new(big.Rat).SetInt64(3600)) + case "720p": + denom := int64(3600 * 1280 * 720 * 30) // 1h of 720p @ 30fps + usdPrice = new(big.Rat).Quo(usdPrice, new(big.Rat).SetInt64(denom)) + case "fixed": + // Fixed prices are already denominated per request. } - usdPrice := new(big.Rat).Quo(usdPerHour, new(big.Rat).SetInt64(denom)) return core.NewAutoConvertedPrice("USD", usdPrice, nil) } @@ -1773,17 +1795,27 @@ func (runner *liveRunner) convertPrice() (LiveRunnerPriceInfo, error) { if err != nil { return LiveRunnerPriceInfo{}, err } - convertedUnit := "seconds" - if runner.PriceInfo.Unit == "720p" { - convertedUnit = "720p-pixel-seconds" + if price.Sign() <= 0 { + return LiveRunnerPriceInfo{}, fmt.Errorf("converted live runner price must be at least one wei") } return LiveRunnerPriceInfo{ Price: json.Number(fmt.Sprintf("%d", price.Num().Int64())), Currency: "wei", - Unit: convertedUnit, + Unit: convertedLiveRunnerPriceUnit(runner.PriceInfo.Unit), }, nil } +func convertedLiveRunnerPriceUnit(unit string) string { + switch strings.ToLower(strings.TrimSpace(unit)) { + case "720p": + return "720p-pixel-seconds" + case "fixed": + return "fixed" + default: + return "seconds" + } +} + func isReadyStatus(status string) bool { return status == "" || status == "ready" } diff --git a/ai/runner/live_runner_test.go b/ai/runner/live_runner_test.go index bf27697a5b..699d9bafdd 100644 --- a/ai/runner/live_runner_test.go +++ b/ai/runner/live_runner_test.go @@ -1118,6 +1118,7 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t App: "live-video-to-video/scope", Capacity: 2, HealthURL: healthSrv.URL, + PriceInfo: LiveRunnerPriceInfo{Price: json.Number("10"), Unit: "fixed"}, } resp1, err := registry.RegisterStaticRunners(StaticLiveRunnerConfig{Runners: []StaticLiveRunnerConfigEntry{entry}}) if err != nil { @@ -1130,6 +1131,7 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t } entry.RunnerURL = "https://runner-updated.example.com" + entry.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("20"), Unit: "hour"} resp2, err := registry.RegisterStaticRunners(StaticLiveRunnerConfig{Runners: []StaticLiveRunnerConfigEntry{entry}}) if err != nil { t.Fatal(err) @@ -1140,6 +1142,13 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t if _, err := registry.RunnerEndpointForSession(runnerID, sessionID); err != nil { t.Fatalf("expected active session to survive healthy static upsert: %v", err) } + sessionPriceInfo, err := registry.SessionPriceInfo(runnerID, sessionID) + if err != nil { + t.Fatal(err) + } + if sessionPriceInfo.Price.String() != "10" || sessionPriceInfo.Unit != "fixed" { + t.Fatalf("unexpected static session price snapshot: %+v", sessionPriceInfo) + } } func TestLiveRunnerRegistry_RegisterStaticRunnersConcurrentUpserts(t *testing.T) { diff --git a/doc/live-runner.md b/doc/live-runner.md index 8e908cf6c1..c4753c2649 100644 --- a/doc/live-runner.md +++ b/doc/live-runner.md @@ -194,10 +194,12 @@ On an on-chain network, every runner must provide a positive `price_info.price`. The registration price is denominated in USD: - `currency` defaults to `usd`; no other registration currency is accepted. -- `unit` defaults to `hour` and accepts `hour` or `720p`. +- `unit` defaults to `hour` and accepts `hour`, `720p`, or `fixed`. - `hour` is converted to a wei-per-second discovery/payment price. - `720p` is converted using 720p at 30 fps and is advertised in `720p-pixel-seconds`. +- `fixed` is converted directly from USD to wei without a time or pixel + divisor and is advertised as `fixed`. The conversion follows the node's USD/ETH price feed. On an offchain network, runner prices are not advertised and no payment challenge is required. @@ -293,9 +295,7 @@ UTF-8 and at most 1,024 UTF-8-encoded bytes. An empty value clears previously registered metadata and is omitted from discovery. On-chain discovery includes the converted wei price. Offchain runner discovery -omits it. If the orchestrator also has a serverless AI worker, the response may -contain synthetic live-video-to-video runner records alongside registered live -runners. +omits it. #### Remote signer discovery @@ -391,7 +391,7 @@ Required` with payment parameters. The client retries with `Livepeer-Payment` and `Livepeer-Segment`, then periodically refreshes payment at `POST /apps/{runner_id}/session/{session_id}/payment`. The orchestrator accounts usage on its configured payment interval and releases the session if payment -fails. +fails. If the payment unit is `fixed` then payment is only processed once. The current single-shot proxy path does not perform a payment challenge or accounting. Use persistent mode when on-chain payment enforcement is required. @@ -420,7 +420,7 @@ are currently ignored. Labels in one submitted batch must be unique. | `mode` | string | Default: `persistent` | `persistent`, `single-shot`, or the normalized alias `single_shot`. | | `capacity` | integer | Default: `1` when zero | Maximum concurrent sessions. Negative values are invalid. | | `gpu` | object or integer | Optional | Object fields are `id`, `name`, and `vram_mb`. An integer is treated as a local device index; a negative index records only that numeric ID and skips hardware lookup. | -| `price_info` | object | Required onchain; ignored for offchain payment/discovery | `price` is a positive decimal. `currency` defaults to and must equal `usd`. `unit` defaults to `hour` and accepts `hour` or `720p`. | +| `price_info` | object | Required onchain; ignored for offchain payment/discovery | `price` is a positive decimal. `currency` defaults to and must equal `usd`. `unit` defaults to `hour` and accepts `hour`, `720p`, or `fixed`. | ### CLI flags @@ -537,9 +537,9 @@ session. | Method and path | Caller / authentication | Description | Principal responses | | --- | --- | --- | --- | -| `POST /apps/{runner_id}/session` | Client; no auth offchain. Onchain uses `Livepeer-Payer-Address` for the initial challenge and `Livepeer-Payment` plus `Livepeer-Segment` to reserve. | Reserves a persistent session. Returns `session_id`, `app_url`, and `control_url`; `proxy: true` registrations receive a random proxy-template `app_url`. A priced runner first returns a payment challenge. | `200`, `402`, `404`, `503` | +| `POST /apps/{runner_id}/session` | Client; no auth offchain. Onchain uses `Livepeer-Payer-Address` for the initial challenge and `Livepeer-Payment` plus `Livepeer-Segment` to reserve. | Reserves a persistent session. Returns `session_id`, `app_url`, and `control_url`; `proxy: true` registrations receive a random proxy-template `app_url`. A priced runner first returns a payment challenge. | `200`, `400`, `402`, `404`, `409`, `503` | | `POST /apps/{runner_id}/session/{session_id}/stop` | Client; session is identified by the URL | Releases a persistent session, its channels, and proxies. | `204`, `404` | -| `POST /apps/{runner_id}/session/{session_id}/payment` | Paying client; `Livepeer-Payment` and `Livepeer-Segment` | Adds payment for an active session. The payment manifest must match `session_id`. | `200`, `400`, `403`, `404` | +| `POST /apps/{runner_id}/session/{session_id}/payment` | Paying client; `Livepeer-Payment` and `Livepeer-Segment` | Adds payment for an active session. The payment manifest must match `session_id`. Fixed-payment sessions do not need to call this. | `200`, `400`, `403`, `404`, `409` | | `ANY /apps/{runner_id}/session/{session_id}/app/{app_path...}` | Client; access is by the reserved public URL | Proxies any HTTP method, SSE response, or WebSocket upgrade to a persistent runner. | Upstream status, `404`, `502` | | `ANY /apps/{runner_id}/app/{app_path...}` | Client; no application-level authentication | Reserves a single-shot session, proxies one request, then releases it. The current path does not enforce runner payment. | Upstream status, `400`, `404`, `503`, `502` | diff --git a/doc/remote-signer.md b/doc/remote-signer.md index d8a2709bb0..10f8e3e430 100644 --- a/doc/remote-signer.md +++ b/doc/remote-signer.md @@ -125,7 +125,7 @@ Remote discovery filters data before exposing it: configured capability/model maximum from `-maxPricePerCapability`, falling back to the global maximum when applicable; - runner records require a URL, a non-empty `app`, and a positive wei price; -- runner price units must be `seconds` or `720p-pixel-seconds`; and +- runner price units must be `seconds`, `720p-pixel-seconds`, or `fixed`; and - the global `-maxPricePerUnit`, when set, also limits runner prices. Each valid runner's `app` is added to the address's capability list and can be diff --git a/server/ai_http.go b/server/ai_http.go index 67d9a5560d..6998292b40 100644 --- a/server/ai_http.go +++ b/server/ai_http.go @@ -116,6 +116,7 @@ type liveRunnerManager interface { PaymentInfo(runnerID string) (*runner.LiveRunnerPriceInfo, error) ReserveSession(runnerID string, sessionID ...string) (string, string, error) ReleaseSession(runnerID, sessionID string) error + SessionPriceInfo(runnerID, sessionID string) (runner.LiveRunnerPriceInfo, error) RunnerMode(runnerID string) (string, error) RunnerEndpointForSession(runnerID, sessionID string) (string, error) SessionTokenForSession(runnerID, sessionID string) (string, error) @@ -218,6 +219,7 @@ func (h *lphttp) ReserveLiveRunnerSession(w http.ResponseWriter, r *http.Request return } runnerID := r.PathValue("runner_id") + ctx := clog.AddVal(context.Background(), "runner_id", runnerID) mode, err := manager.RunnerMode(runnerID) if err != nil { respondWithLiveRunnerError(w, err) @@ -232,100 +234,147 @@ func (h *lphttp) ReserveLiveRunnerSession(w http.ResponseWriter, r *http.Request respondWithLiveRunnerError(w, err) return } - paymentRequired := priceInfo != nil - var newPaymentProcessor func(context.Context, time.Duration, func(int64) error) *LivePaymentProcessor - if paymentRequired { - newPaymentProcessor, err = preparePaymentProcessor(priceInfo.Unit) - if err != nil { - respondWithError(w, err.Error(), http.StatusInternalServerError) - return - } - } - if paymentRequired && r.Header.Get(paymentHeader) == "" && r.Header.Get(segmentHeader) == "" { - h.runnerChallenge(w, r, priceInfo) - return - } var ( - payment lpnet.Payment - segData *core.SegTranscodingMetadata - ctx = r.Context() sessionID string appURL string ) - if paymentRequired { - var err error - payment, segData, ctx, err = h.processPaymentAndSegmentHeaders(w, r) + if priceInfo == nil { + sessionID, appURL, err = manager.ReserveSession(runnerID) if err != nil { + respondWithLiveRunnerError(w, err) return } - if string(segData.ManifestID) != segData.AuthToken.SessionId { - respondWithError(w, "mismatched manifest and auth token", http.StatusForbidden) + ctx = clog.AddVal(ctx, "session_id", sessionID) + } else { + var reservationOK bool + sessionID, appURL, reservationOK = h.reservePaidLiveRunnerSession(ctx, w, r, manager, runnerID, priceInfo) + if !reservationOK { return } - // for easier correlation across orch, gw, signer - sessionID = string(segData.ManifestID) } - sessionID, appURL, err = manager.ReserveSession(runnerID, sessionID) + controlURL := h.orchestrator.ServiceURI().JoinPath("apps", runnerID, "session", sessionID).String() + data, err := json.Marshal(liveRunnerSessionResponse{SessionID: sessionID, AppURL: appURL, ControlURL: controlURL}) if err != nil { - respondWithLiveRunnerError(w, err) + respondWithError(w, err.Error(), http.StatusInternalServerError) return } - ctx = clog.AddVal(ctx, "runner_id", runnerID) + respondJsonOk(w, data) +} + +func (h *lphttp) reservePaidLiveRunnerSession( + ctx context.Context, + w http.ResponseWriter, + r *http.Request, + manager liveRunnerManager, + runnerID string, + priceInfo *runner.LiveRunnerPriceInfo, +) (string, string, bool) { + // This helper owns all challenge and error responses for paid reservations. + fixedPayment := strings.EqualFold(strings.TrimSpace(priceInfo.Unit), "fixed") + var newPaymentProcessor func(context.Context, time.Duration, func(int64) error) *LivePaymentProcessor + + if r.Header.Get(paymentHeader) == "" && r.Header.Get(segmentHeader) == "" { + h.runnerChallenge(w, r, priceInfo) + return "", "", false + } + payment, segData, _, err := h.processPaymentAndSegmentHeaders(w, r) + if err != nil { + return "", "", false + } + if string(segData.ManifestID) != segData.AuthToken.SessionId { + respondWithError(w, "mismatched manifest and auth token", http.StatusForbidden) + return "", "", false + } + + if fixedPayment { + expectedPrice, priceErr := priceInfo.Price.Int64() + if priceErr != nil || payment.GetExpectedPrice() == nil || payment.GetExpectedPrice().GetPricePerUnit() != expectedPrice || payment.GetExpectedPrice().GetPixelsPerUnit() != 1 { + respondWithError(w, "payment price does not match live runner price", http.StatusBadRequest) + return "", "", false + } + } else { + newPaymentProcessor, err = preparePaymentProcessor(priceInfo.Unit) + if err != nil { + respondWithError(w, err.Error(), http.StatusInternalServerError) + return "", "", false + } + + } + + // For easier correlation across orchestrator, gateway, and signer. + sessionID := string(segData.ManifestID) + sessionID, appURL, err := manager.ReserveSession(runnerID, sessionID) + if err != nil { + respondWithLiveRunnerError(w, err) + return "", "", false + } ctx = clog.AddVal(ctx, "session_id", sessionID) - if paymentRequired { - if err := h.orchestrator.ProcessPayment(ctx, payment, segData.ManifestID); err != nil { + + if err := h.orchestrator.ProcessPayment(ctx, payment, segData.ManifestID); err != nil { + if releaseErr := manager.ReleaseSession(runnerID, sessionID); releaseErr != nil { + clog.Errorf(ctx, "Error releasing live runner session after payment failure err=%v", releaseErr) + } + respondWithError(w, err.Error(), http.StatusBadRequest) + return "", "", false + } + + paymentReceiver := livePaymentReceiver{orchestrator: h.orchestrator} + if fixedPayment { + err := paymentReceiver.AccountPayment(ctx, &SegmentInfoReceiver{ + sender: getPaymentSender(payment), + units: 1, + priceInfo: payment.GetExpectedPrice(), + sessionID: string(segData.ManifestID), + }) + if err != nil { if releaseErr := manager.ReleaseSession(runnerID, sessionID); releaseErr != nil { - clog.Errorf(ctx, "Error releasing live runner session after payment failure err=%v", releaseErr) + clog.Errorf(ctx, "Error releasing fixed-price live runner session after accounting failure err=%v", releaseErr) } respondWithError(w, err.Error(), http.StatusBadRequest) - return + return "", "", false } - monitorCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) - paymentReceiver := livePaymentReceiver{orchestrator: h.orchestrator} - accountPaymentFunc := func(units int64) error { - err := paymentReceiver.AccountPayment(monitorCtx, &SegmentInfoReceiver{ - sender: getPaymentSender(payment), - units: units, - priceInfo: payment.GetExpectedPrice(), - sessionID: string(segData.ManifestID), - }) - if err != nil { - clog.Errorf(monitorCtx, "Error accounting live runner payment, releasing session err=%v", err) - if releaseErr := manager.ReleaseSession(runnerID, sessionID); releaseErr != nil { - clog.Errorf(monitorCtx, "Error releasing live runner session after payment failure err=%v", releaseErr) - } - // Stop both the ticker loop below and the LivePaymentProcessor goroutine. - cancel() + return sessionID, appURL, true + } + + // time based payments + monitorCtx, cancel := context.WithCancel(ctx) + accountPaymentFunc := func(units int64) error { + err := paymentReceiver.AccountPayment(monitorCtx, &SegmentInfoReceiver{ + sender: getPaymentSender(payment), + units: units, + priceInfo: payment.GetExpectedPrice(), + sessionID: string(segData.ManifestID), + }) + if err != nil { + clog.Errorf(monitorCtx, "Error accounting live runner payment, releasing session err=%v", err) + if releaseErr := manager.ReleaseSession(runnerID, sessionID); releaseErr != nil { + clog.Errorf(monitorCtx, "Error releasing live runner session after payment failure err=%v", releaseErr) } - return err + // Stop both the ticker loop below and the LivePaymentProcessor goroutine. + cancel() } - paymentProcessor := newPaymentProcessor(monitorCtx, h.node.LivePaymentInterval, accountPaymentFunc) - go func() { - ticker := time.NewTicker(h.node.LivePaymentInterval) - defer ticker.Stop() - defer cancel() - for { - select { - case <-ticker.C: - // Stop monitoring once the live runner session has been released - // by an explicit stop, runner cleanup, expiry, or payment failure. - if _, err := manager.RunnerEndpointForSession(runnerID, sessionID); err != nil { - return - } - paymentProcessor.process(monitorCtx) - case <-monitorCtx.Done(): + return err + } + paymentProcessor := newPaymentProcessor(monitorCtx, h.node.LivePaymentInterval, accountPaymentFunc) + go func() { + ticker := time.NewTicker(h.node.LivePaymentInterval) + defer ticker.Stop() + defer cancel() + for { + select { + case <-ticker.C: + // Stop monitoring once the live runner session has been released + // by an explicit stop, runner cleanup, expiry, or payment failure. + if _, err := manager.RunnerEndpointForSession(runnerID, sessionID); err != nil { return } + paymentProcessor.process(monitorCtx) + case <-monitorCtx.Done(): + return } - }() - } - controlURL := h.orchestrator.ServiceURI().JoinPath("apps", runnerID, "session", sessionID).String() - data, err := json.Marshal(liveRunnerSessionResponse{SessionID: sessionID, AppURL: appURL, ControlURL: controlURL}) - if err != nil { - respondWithError(w, err.Error(), http.StatusInternalServerError) - return - } - respondJsonOk(w, data) + } + }() + return sessionID, appURL, true } func preparePaymentProcessor(unit string) (func(context.Context, time.Duration, func(int64) error) *LivePaymentProcessor, error) { @@ -487,10 +536,15 @@ func (h *lphttp) PaymentForLiveRunnerSession(w http.ResponseWriter, r *http.Requ runnerID := r.PathValue("runner_id") sessionID := r.PathValue("session_id") - if _, err := manager.RunnerEndpointForSession(runnerID, sessionID); err != nil { + priceInfo, err := manager.SessionPriceInfo(runnerID, sessionID) + if err != nil { respondWithLiveRunnerError(w, err) return } + if strings.EqualFold(strings.TrimSpace(priceInfo.Unit), "fixed") { + respondWithError(w, "fixed-price live runner sessions do not accept follow-up payments", http.StatusConflict) + return + } payment, segData, ctx, err := h.processPaymentAndSegmentHeaders(w, r) if err != nil { diff --git a/server/ai_http_test.go b/server/ai_http_test.go index 039ec09ccb..dd5862aaa5 100644 --- a/server/ai_http_test.go +++ b/server/ai_http_test.go @@ -673,6 +673,73 @@ func TestLiveRunnerReserveSessionOnchainReturnsPaymentChallenge(t *testing.T) { require.Nil(t, oInfo.GetCapabilities()) } +func TestLiveRunnerFixedPriceSessionAccountsOnce(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lp := newLiveRunnerHTTPOnchain(t) + lp.node.LivePaymentInterval = time.Second + registerLiveRunnerForSession(t, lp, &liveRunnerRegistrationOptions{ + PriceInfo: &runner.LiveRunnerPriceInfo{Price: json.Number("0.001"), Unit: "fixed"}, + }) + + orch := lp.orchestrator.(*stubOrchestrator) + orch.balances = make(map[ethcommon.Address]map[core.ManifestID]*big.Rat) + + challenge, oInfo := requestLiveRunnerPaymentChallenge(t, lp, "runner-1") + orch.paymentCredit = big.NewRat(oInfo.GetPriceInfo().GetPricePerUnit(), 1) + headers := liveRunnerReservationPaymentHeadersWithPrice(t, orch, oInfo.GetAuthToken(), challenge.ManifestID, oInfo.GetPriceInfo()) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/apps/runner-1/session", nil) + setRequestHeaders(req, headers) + lp.ServeHTTP(w, req) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + + balance := orch.Balance(orch.Address(), core.ManifestID(challenge.ManifestID)) + require.NotNil(t, balance) + require.Zero(t, balance.Sign()) + + time.Sleep(time.Second) + synctest.Wait() + + manager := lp.node.LiveRunnerManager.(*runner.LiveRunnerRegistry) + _, err := manager.RunnerEndpointForSession("runner-1", challenge.ManifestID) + require.NoError(t, err) + balance = orch.Balance(orch.Address(), core.ManifestID(challenge.ManifestID)) + require.NotNil(t, balance) + require.Zero(t, balance.Sign()) + + w = httptest.NewRecorder() + req = httptest.NewRequest(http.MethodPost, "/apps/runner-1/session/"+challenge.ManifestID+"/payment", nil) + setRequestHeaders(req, headers) + lp.ServeHTTP(w, req) + require.Equal(t, http.StatusConflict, w.Code) + }) +} + +func TestLiveRunnerFixedPriceSessionRejectsMismatchedPriceBeforeReservation(t *testing.T) { + lp := newLiveRunnerHTTPOnchain(t) + registerLiveRunnerForSession(t, lp, &liveRunnerRegistrationOptions{ + PriceInfo: &runner.LiveRunnerPriceInfo{Price: json.Number("0.001"), Unit: "fixed"}, + }) + + challenge, oInfo := requestLiveRunnerPaymentChallenge(t, lp, "runner-1") + orch := lp.orchestrator.(*stubOrchestrator) + mismatchedPrice := proto.Clone(oInfo.GetPriceInfo()).(*lpnet.PriceInfo) + mismatchedPrice.PricePerUnit-- + headers := liveRunnerReservationPaymentHeadersWithPrice(t, orch, oInfo.GetAuthToken(), challenge.ManifestID, mismatchedPrice) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/apps/runner-1/session", nil) + setRequestHeaders(req, headers) + lp.ServeHTTP(w, req) + + require.Equal(t, http.StatusBadRequest, w.Code) + require.Contains(t, w.Body.String(), "payment price does not match live runner price") + manager := lp.node.LiveRunnerManager.(*runner.LiveRunnerRegistry) + _, err := manager.RunnerEndpointForSession("runner-1", challenge.ManifestID) + require.Error(t, err) +} + func TestLiveRunnerSessionPaymentAcceptsPayment(t *testing.T) { oldStorage := drivers.NodeStorage drivers.NodeStorage = drivers.NewMemoryDriver(nil) @@ -790,6 +857,47 @@ func TestLiveRunnerPaidSessionMonitorDebitsBalance(t *testing.T) { }) } +func TestReservePaidLiveRunnerSessionContextCancellationStopsBilling(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + lp := newLiveRunnerHTTPOnchain(t) + lp.node.LivePaymentInterval = time.Second + registerLiveRunnerForSession(t, lp, nil) + manager := lp.node.LiveRunnerManager.(*runner.LiveRunnerRegistry) + + orch := lp.orchestrator.(*stubOrchestrator) + orch.balances = make(map[ethcommon.Address]map[core.ManifestID]*big.Rat) + orch.paymentCredit = big.NewRat(3, 1) + + challenge, oInfo := requestLiveRunnerPaymentChallenge(t, lp, "runner-1") + headers := liveRunnerReservationPaymentHeadersWithPrice(t, orch, oInfo.GetAuthToken(), challenge.ManifestID, oInfo.GetPriceInfo()) + priceInfo, err := manager.PaymentInfo("runner-1") + require.NoError(t, err) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/apps/runner-1/session", nil) + setRequestHeaders(req, headers) + ctx, cancel := context.WithCancel(context.Background()) + sessionID, _, ok := lp.reservePaidLiveRunnerSession(ctx, w, req, manager, "runner-1", priceInfo) + require.True(t, ok, w.Body.String()) + + balance := orch.Balance(orch.Address(), core.ManifestID(sessionID)) + require.NotNil(t, balance) + require.Equal(t, "3", balance.FloatString(0)) + + // An active monitor would debit one unit per second. Canceling its parent + // context must stop billing without releasing the reserved session. + cancel() + time.Sleep(2 * time.Second) + synctest.Wait() + + _, err = manager.RunnerEndpointForSession("runner-1", sessionID) + require.NoError(t, err) + balance = orch.Balance(orch.Address(), core.ManifestID(sessionID)) + require.NotNil(t, balance) + require.Equal(t, "3", balance.FloatString(0)) + }) +} + func TestPreparePaymentProcessor(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -1992,9 +2100,10 @@ type liveRunnerRegistrationOptions struct { Capacity int Mode string Proxy bool + PriceInfo *runner.LiveRunnerPriceInfo } -func registerLiveRunnerForSession(t *testing.T, lp *lphttp, opts *liveRunnerRegistrationOptions) { +func registerLiveRunnerForSession(t *testing.T, lp *lphttp, opts *liveRunnerRegistrationOptions) *runner.LiveRunnerHeartbeatResponse { t.Helper() if lp == nil { require.FailNow(t, "live runner lphttp is required") @@ -2026,9 +2135,16 @@ func registerLiveRunnerForSession(t *testing.T, lp *lphttp, opts *liveRunnerRegi if opts.Mode != "" && opts.Mode != runner.LiveRunnerModePersistent && opts.Mode != runner.LiveRunnerModeSingleShot { require.FailNow(t, "invalid live runner mode", "mode=%q", opts.Mode) } + priceInfo := runner.LiveRunnerPriceInfo{ + Price: json.Number("0.001990656"), + Unit: "720p", + } + if opts.PriceInfo != nil { + priceInfo = *opts.PriceInfo + } manager, ok := lp.liveRunnerManager() require.True(t, ok) - _, err := manager.Heartbeat(runner.LiveRunnerHeartbeatRequest{ + resp, err := manager.Heartbeat(runner.LiveRunnerHeartbeatRequest{ RunnerID: runnerID, Proxy: opts.Proxy, RunnerURL: runnerURL, @@ -2036,12 +2152,10 @@ func registerLiveRunnerForSession(t *testing.T, lp *lphttp, opts *liveRunnerRegi Mode: opts.Mode, App: "live-video-to-video/scope", Capacity: capacity, - PriceInfo: runner.LiveRunnerPriceInfo{ - Price: json.Number("0.001990656"), - Unit: "720p", - }, + PriceInfo: priceInfo, }, lp.orchestrator.RegistrationSecret()) require.NoError(t, err) + return resp } func reserveLiveRunnerSession(t *testing.T, lp *lphttp, runnerID string) string { diff --git a/server/remote_discovery.go b/server/remote_discovery.go index 3968de7e09..499b5f9f7a 100644 --- a/server/remote_discovery.go +++ b/server/remote_discovery.go @@ -322,7 +322,7 @@ func validateRunnerPrice(priceInfo *runner.LiveRunnerPriceInfo) (*big.Rat, error return nil, errors.New("unsupported currency") } unit := strings.ToLower(strings.TrimSpace(priceInfo.Unit)) - if unit != "seconds" && unit != "720p-pixel-seconds" { + if unit != "seconds" && unit != "720p-pixel-seconds" && unit != "fixed" { return nil, errors.New("unsupported unit") } return price, nil diff --git a/server/remote_signer.go b/server/remote_signer.go index 245bf5001c..3949d503fa 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_Live = "live" const RemoteType_LiveVideoToVideo = "lv2v" +const RemoteType_Fixed = "fixed" const PipelineLiveVideoToVideo = "live-video-to-video" const remoteSignerAuthIDHeader = "Signer-Auth-Id" @@ -179,7 +180,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: `live`, `lv2v`. Optional. + // Job type to automatically calculate payments. Valid values: `live`, `lv2v`, `fixed`. Optional. Type string `json:"type"` // Capabilities to include in the ticket. Optional; may be set for the lv2v job type. @@ -487,6 +488,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req billableSecs = (10 * time.Second).Seconds() } billableUnits = int64(math.Ceil(billableSecs)) // seconds to charge for + } else if req.Type == RemoteType_Fixed { + billableUnits = 1 } else if req.Type != "" { err = errors.New("invalid job type") respondJsonError(ctx, w, err, http.StatusBadRequest) @@ -636,6 +639,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req pipeline = PipelineLiveVideoToVideo } else if req.Type == RemoteType_Live { pipeline = RemoteType_Live + } else if req.Type == RemoteType_Fixed { + pipeline = RemoteType_Fixed } // NB: This could could drop events if tha Kafka queue is full! monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{ diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index cc6e976637..e03989df42 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -276,6 +276,20 @@ func TestGenerateLivePayment_RequestValidationErrors(t *testing.T) { wantStatus: HTTPStatusPriceExceeded, wantMsg: "orchestrator price", }, + { + name: "fixed orchestrator price uses existing max price check", + req: func() RemotePaymentRequest { + oInfo := proto.Clone(baseOrchInfo).(*net.OrchestratorInfo) + oInfo.PriceInfo = &net.PriceInfo{PricePerUnit: 1000, PixelsPerUnit: 1} + return RemotePaymentRequest{ + Orchestrator: makeOrchBlob(oInfo), + ManifestID: "fixed-manifest", + Type: RemoteType_Fixed, + } + }(), + wantStatus: HTTPStatusPriceExceeded, + wantMsg: "orchestrator price", + }, { name: "orchestrator price exceeds per-capability max price", req: func() RemotePaymentRequest { @@ -787,6 +801,80 @@ func TestGenerateLivePayment_LiveBillsElapsedSeconds(t *testing.T) { }) } +func TestGenerateLivePayment_FixedBillsOneUnitAndAllowsState(t *testing.T) { + require := require.New(t) + + ethClient := newTestEthClient(t) + node, _ := core.NewLivepeerNode(ethClient, "", nil) + node.Balances = core.NewAddressBalances(time.Minute) + defer node.Balances.StopCleanup() + var ticketNonce uint32 + node.Sender = newMockSender(mockSenderConfig{ + ev: big.NewRat(3, 1), + createTicketBatchFn: func(args mock.Arguments, batch *pm.TicketBatch) { + require.Equal(1, args.Int(1)) + *batch = *defaultTicketBatch() + ticketNonce++ + batch.SenderParams = []*pm.TicketSenderParams{{ + SenderNonce: ticketNonce, + Sig: pm.RandBytes(42), + }} + }, + }) + ls := &LivepeerServer{LivepeerNode: node} + + oInfo := &net.OrchestratorInfo{ + Address: ethClient.addr.Bytes(), + PriceInfo: &net.PriceInfo{PricePerUnit: 3, PixelsPerUnit: 1}, + TicketParams: &net.TicketParams{ + Recipient: pm.RandAddress().Bytes(), + }, + AuthToken: stubAuthToken, + } + orchBlob, err := proto.Marshal(oInfo) + require.NoError(err) + + doPayment := func(manifestID string, state RemotePaymentStateSig) (RemotePaymentResponse, net.Payment) { + body, err := json.Marshal(RemotePaymentRequest{ + Orchestrator: orchBlob, + ManifestID: manifestID, + InPixels: 1_000_000, + Type: RemoteType_Fixed, + State: state, + }) + require.NoError(err) + rr := httptest.NewRecorder() + ls.GenerateLivePayment(rr, httptest.NewRequest(http.MethodPost, "/generate-live-payment", bytes.NewReader(body))) + require.Equal(http.StatusOK, rr.Code, rr.Body.String()) + + var resp RemotePaymentResponse + require.NoError(json.NewDecoder(rr.Body).Decode(&resp)) + paymentBytes, err := base64.StdEncoding.DecodeString(resp.Payment) + require.NoError(err) + var payment net.Payment + require.NoError(proto.Unmarshal(paymentBytes, &payment)) + return resp, payment + } + + // Like other payment types, an initial request may omit the manifest ID. + first, firstPayment := doPayment("", RemotePaymentStateSig{}) + require.Len(firstPayment.TicketSenderParams, 1) + var firstState RemotePaymentState + require.NoError(json.Unmarshal(first.State.State, &firstState)) + require.Equal(RemoteType_Fixed, firstState.Type) + require.EqualValues(0, firstState.SequenceNumber) + require.Equal("0", firstState.Balance) + + second, secondPayment := doPayment("fixed-manifest", first.State) + require.Len(secondPayment.TicketSenderParams, 1) + require.NotEqual(firstPayment.TicketSenderParams[0].SenderNonce, secondPayment.TicketSenderParams[0].SenderNonce) + var secondState RemotePaymentState + require.NoError(json.Unmarshal(second.State.State, &secondState)) + require.Equal(RemoteType_Fixed, secondState.Type) + require.EqualValues(1, secondState.SequenceNumber) + require.Equal("0", secondState.Balance) +} + func TestGenerateLivePayment_WebhookCallback(t *testing.T) { require := require.New(t) @@ -1604,6 +1692,16 @@ func discoveryRaw(t *testing.T, data string) json.RawMessage { return json.RawMessage(data) } +func TestValidateRunnerPriceAcceptsFixedUnit(t *testing.T) { + price, err := validateRunnerPrice(&runner.LiveRunnerPriceInfo{ + Price: json.Number("7"), + Currency: "wei", + Unit: "fixed", + }) + require.NoError(t, err) + require.Zero(t, price.Cmp(big.NewRat(7, 1))) +} + func discoveryRunnerApps(t *testing.T, resp discoveryResponse) []string { t.Helper() apps := make([]string, 0, len(resp.Runners)) From 922932a1990f506662ad1cbc73b8e138b18ff3c7 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 24 Jul 2026 13:56:05 -0700 Subject: [PATCH 2/2] PR feedback --- ai/runner/live_runner_test.go | 2 +- server/ai_http.go | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/ai/runner/live_runner_test.go b/ai/runner/live_runner_test.go index 699d9bafdd..712c934fe4 100644 --- a/ai/runner/live_runner_test.go +++ b/ai/runner/live_runner_test.go @@ -1147,7 +1147,7 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t t.Fatal(err) } if sessionPriceInfo.Price.String() != "10" || sessionPriceInfo.Unit != "fixed" { - t.Fatalf("unexpected static session price snapshot: %+v", sessionPriceInfo) + t.Fatalf("expected static session price snapshot with price 10 and unit fixed, got: %+v", sessionPriceInfo) } } diff --git a/server/ai_http.go b/server/ai_http.go index 6998292b40..32398236c3 100644 --- a/server/ai_http.go +++ b/server/ai_http.go @@ -244,7 +244,6 @@ func (h *lphttp) ReserveLiveRunnerSession(w http.ResponseWriter, r *http.Request respondWithLiveRunnerError(w, err) return } - ctx = clog.AddVal(ctx, "session_id", sessionID) } else { var reservationOK bool sessionID, appURL, reservationOK = h.reservePaidLiveRunnerSession(ctx, w, r, manager, runnerID, priceInfo)