diff --git a/doc/live-runner.md b/doc/live-runner.md index c4753c2649..627d9d07db 100644 --- a/doc/live-runner.md +++ b/doc/live-runner.md @@ -386,16 +386,13 @@ The proxy ID must be the final template path segment, so a template such as ### Payments On-chain runner sessions use Livepeer probabilistic micropayments. Reserving a -priced persistent runner without payment material returns `402 Payment -Required` with payment parameters. The client retries with -`Livepeer-Payment` and `Livepeer-Segment`, then periodically refreshes payment at +runner without payment material returns `402 Payment Required` with payment +parameters. The client retries the same request 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. 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. - Offchain runners do not issue payment challenges. For the underlying ticket protocol, see [Payments](payments.md). @@ -541,7 +538,7 @@ session. | `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`. 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` | +| `ANY /apps/{runner_id}/app/{app_path...}` | Client; no application-level authentication | Reserves a single-shot session, proxies one request, then releases it. | Upstream status, `400`, `404`, `503`, `502` | The `runner_id` path value may be a static label when `routing` is `label`. diff --git a/server/ai_http.go b/server/ai_http.go index 32398236c3..7008112719 100644 --- a/server/ai_http.go +++ b/server/ai_http.go @@ -245,9 +245,9 @@ func (h *lphttp) ReserveLiveRunnerSession(w http.ResponseWriter, r *http.Request return } } else { - var reservationOK bool - sessionID, appURL, reservationOK = h.reservePaidLiveRunnerSession(ctx, w, r, manager, runnerID, priceInfo) - if !reservationOK { + var reserved bool + sessionID, appURL, reserved = h.reservePaidLiveRunnerSession(ctx, w, r, manager, runnerID, priceInfo, nil) + if !reserved { return } } @@ -267,6 +267,7 @@ func (h *lphttp) reservePaidLiveRunnerSession( manager liveRunnerManager, runnerID string, priceInfo *runner.LiveRunnerPriceInfo, + cancelRequest func(), ) (string, string, bool) { // This helper owns all challenge and error responses for paid reservations. fixedPayment := strings.EqualFold(strings.TrimSpace(priceInfo.Unit), "fixed") @@ -349,6 +350,9 @@ func (h *lphttp) reservePaidLiveRunnerSession( if releaseErr := manager.ReleaseSession(runnerID, sessionID); releaseErr != nil { clog.Errorf(monitorCtx, "Error releasing live runner session after payment failure err=%v", releaseErr) } + if cancelRequest != nil { + cancelRequest() + } // Stop both the ticker loop below and the LivePaymentProcessor goroutine. cancel() } @@ -762,11 +766,30 @@ func (h *lphttp) ProxyLiveRunnerSingleShot(w http.ResponseWriter, r *http.Reques return } - sessionID, endpoint, err := manager.ReserveSession(runnerID) + priceInfo, err := manager.PaymentInfo(runnerID) if err != nil { respondWithLiveRunnerError(w, err) return } + var ( + sessionID string + endpoint string + ) + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + if priceInfo == nil { + sessionID, endpoint, err = manager.ReserveSession(runnerID) + if err != nil { + respondWithLiveRunnerError(w, err) + return + } + } else { + var reserved bool + sessionID, endpoint, reserved = h.reservePaidLiveRunnerSession(ctx, w, r, manager, runnerID, priceInfo, cancel) + if !reserved { + return + } + } defer func() { if err := manager.ReleaseSession(runnerID, sessionID); err != nil { slog.Error("error releasing single-shot session", "runner_id", runnerID, "session_id", sessionID, "err", err) @@ -778,7 +801,7 @@ func (h *lphttp) ProxyLiveRunnerSingleShot(w http.ResponseWriter, r *http.Reques respondWithLiveRunnerError(w, err) return } - h.proxyLiveRunner(w, r, runnerID, sessionID, sessionToken, endpoint, r.PathValue("app_path")) + h.proxyLiveRunner(w, r.Clone(ctx), runnerID, sessionID, sessionToken, endpoint, r.PathValue("app_path")) } func (h *lphttp) tryLiveRunnerProxy(w http.ResponseWriter, r *http.Request) bool { diff --git a/server/ai_http_test.go b/server/ai_http_test.go index dd5862aaa5..da28afe7fc 100644 --- a/server/ai_http_test.go +++ b/server/ai_http_test.go @@ -877,7 +877,7 @@ func TestReservePaidLiveRunnerSessionContextCancellationStopsBilling(t *testing. 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) + sessionID, _, ok := lp.reservePaidLiveRunnerSession(ctx, w, req, manager, "runner-1", priceInfo, nil) require.True(t, ok, w.Body.String()) balance := orch.Balance(orch.Address(), core.ManifestID(sessionID)) @@ -946,6 +946,41 @@ func TestLiveRunnerPaidSessionMonitorReleasesOnInsufficientBalance(t *testing.T) }) } +func TestLiveRunnerPaidSessionMonitorCancelsRequestOnInsufficientBalance(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) + defer manager.Stop() + + orch := lp.orchestrator.(*stubOrchestrator) + orch.balances = make(map[ethcommon.Address]map[core.ManifestID]*big.Rat) + orch.paymentCredit = big.NewRat(0, 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()) + defer cancel() + _, _, ok := lp.reservePaidLiveRunnerSession(ctx, w, req, manager, "runner-1", priceInfo, cancel) + require.True(t, ok, w.Body.String()) + + time.Sleep(time.Second) + synctest.Wait() + select { + case <-ctx.Done(): + default: + t.Fatal("expected payment failure to cancel request context") + } + }) +} + func TestLiveRunnerPaidSessionMonitorExitsAfterManualStop(t *testing.T) { synctest.Test(t, func(t *testing.T) { lp := newLiveRunnerHTTPOnchain(t)