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..712c934fe4 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("expected static session price snapshot with price 10 and unit fixed, got: %+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..32398236c3 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,146 @@ 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) + } 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 +535,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))