From 0cff1bd08718ee1a6676667d028fac95aeb89913 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Wed, 15 Jul 2026 01:19:07 -0700 Subject: [PATCH 1/2] Stop filtering remote discovery runners by capability --- server/remote_discovery.go | 45 ++++-------------- server/remote_signer_test.go | 88 +++++++++++++++++++++++------------- 2 files changed, 65 insertions(+), 68 deletions(-) diff --git a/server/remote_discovery.go b/server/remote_discovery.go index d2c2b2f626..546d73f74b 100644 --- a/server/remote_discovery.go +++ b/server/remote_discovery.go @@ -131,8 +131,6 @@ func (p *remoteDiscoveryPool) refresh() { // A node is one discovery `address` value. A single orchestrator may expose // multiple nodes/addresses through its /discovery response. validNodes := make(map[string]bool) - rejectedCapabilityNodes := make(map[string]bool) - validApps := make(map[string]map[string]bool) // Create or reuse the accumulator for a node. The same node address can be // seen from normal polling and from multiple /discovery responses, so merge // everything for that address into one final /discover-orchestrators entry. @@ -164,10 +162,8 @@ func (p *remoteDiscoveryPool) refresh() { // Treat root service URIs with and without a trailing slash as the same address. key := strings.TrimRight(orchURL.String(), "/") eligibleCaps := make(map[string]bool) - advertisedCaps := 0 // Capabilities in discovery are price-eligible capabilities only. forEachRemoteDiscoveryCapability(orchCaps, func(key string, capability core.Capability, modelID string) { - advertisedCaps++ price := capabilityPrice(orchCaps, capability, modelID) if price == nil { return @@ -180,9 +176,6 @@ func (p *remoteDiscoveryPool) refresh() { }) // Drop orchestrators if it doesn't have any eligible capabilities if len(eligibleCaps) == 0 { - if advertisedCaps > 0 { - rejectedCapabilityNodes[key] = true - } continue } validNodes[key] = true @@ -191,7 +184,6 @@ func (p *remoteDiscoveryPool) refresh() { entry.orch.Capabilities = append(entry.orch.Capabilities, cap) } sort.Strings(entry.orch.Capabilities) - validApps[key] = eligibleCaps } for _, orchCaps := range networkCaps { @@ -205,12 +197,7 @@ func (p *remoteDiscoveryPool) refresh() { // Treat root service URIs with and without a trailing slash as the same address. key := strings.TrimRight(discovery.Address, "/") if validNodes[key] { - mergeDiscoveryRunners(merged[key], discovery, validApps[key]) - continue - } - // If normal polling saw this address but price/capability filtering rejected it, - // do not let runner discovery publish it anyway. - if rejectedCapabilityNodes[key] { + mergeDiscoveryRunners(merged[key], discovery) continue } orchURL, err := url.ParseRequestURI(discovery.Address) @@ -218,7 +205,7 @@ func (p *remoteDiscoveryPool) refresh() { continue } entry := ensureMerged(discovery.Address, orchURL) - mergeDiscoveryRunners(entry, discovery, nil) + mergeDiscoveryRunners(entry, discovery) } } @@ -254,14 +241,14 @@ func cloneRemoteDiscoveryOrchestrators(cached []remoteDiscoveryOrchestrator) []r return cloned } -func mergeDiscoveryRunners(entry *remoteDiscoveryMergeEntry, discovery remoteDiscoveryEntry, eligibleCaps map[string]bool) { +func mergeDiscoveryRunners(entry *remoteDiscoveryMergeEntry, discovery remoteDiscoveryEntry) { // For each address, merge runners by URL. First wins; matching duplicates are // ignored and conflicting duplicates are logged before keeping the first. for _, r := range discovery.Runners { if r.URL == "" { continue } - if !entry.remoteDiscoveryRunnerEligible(r, eligibleCaps) { + if !entry.remoteDiscoveryRunnerEligible(r) { continue } if !slices.Contains(entry.orch.Capabilities, r.App) { @@ -299,8 +286,8 @@ func remoteDiscoveryEntries(raw json.RawMessage) []remoteDiscoveryEntry { return entries } -func (entry *remoteDiscoveryMergeEntry) remoteDiscoveryRunnerEligible(runner runner.LiveRunnerDiscoveryRunner, eligibleCaps map[string]bool) bool { - if eligibleCaps != nil && !eligibleCaps[runner.App] { +func (entry *remoteDiscoveryMergeEntry) remoteDiscoveryRunnerEligible(runner runner.LiveRunnerDiscoveryRunner) bool { + if strings.TrimSpace(runner.App) == "" { return false } if runner.PriceInfo == nil { @@ -321,17 +308,13 @@ func (entry *remoteDiscoveryMergeEntry) remoteDiscoveryRunnerEligible(runner run return false } - capability, modelID, ok := capabilityModelFromApp(runner.App) - if !ok { - return false - } - maxPrice := capabilityMaxPrice(capability, modelID) + maxPrice := BroadcastCfg.MaxPrice() if maxPrice == nil { return true } price := new(big.Rat).SetFrac64(runner.PriceInfo.PricePerUnit, runner.PriceInfo.PixelsPerUnit) if price.Cmp(maxPrice) > 0 { - clog.Warningf(context.Background(), "Rejecting remote discovery runner above max price orch=%s runner=%s app=%s price_per_unit=%d pixels_per_unit=%d unit=%s price=%s max_price=%s", + clog.Warningf(context.Background(), "Rejecting remote discovery runner above global max price orch=%s runner=%s app=%s price_per_unit=%d pixels_per_unit=%d unit=%s price=%s max_price=%s", entry.orch.URL.String(), runner.URL, runner.App, @@ -346,18 +329,6 @@ func (entry *remoteDiscoveryMergeEntry) remoteDiscoveryRunnerEligible(runner run return true } -func capabilityModelFromApp(app string) (core.Capability, string, bool) { - pipeline, modelID, ok := strings.Cut(app, "/") - if !ok || pipeline == "" || modelID == "" { - return core.Capability_Unused, "", false - } - capability, err := core.PipelineToCapability(pipeline) - if err != nil { - return core.Capability_Unused, "", false - } - return capability, modelID, true -} - func forEachRemoteDiscoveryCapability(info *common.OrchNetworkCapabilities, f func(key string, capability core.Capability, modelID string)) { if info == nil || info.Capabilities == nil || info.Capabilities.Constraints == nil { return diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index cb793b86b5..f1031a9171 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -1218,22 +1218,26 @@ func TestRemoteSigner_Discovery(t *testing.T) { body := rr.Body.Bytes() var resp []discoveryResponse require.NoError(json.Unmarshal(body, &resp)) - require.Len(resp, 3) + require.Len(resp, 4) orch1Resp := discoveryResponseByAddress(t, resp, "https://orch1.example.com:8935") require.Equal(float32(common.Score_Trusted), orch1Resp.Score) - require.Equal([]string{"live-video-to-video/model-a"}, orch1Resp.Capabilities) - require.Equal([]string{"live-video-to-video/model-a"}, discoveryRunnerApps(t, orch1Resp)) + require.Equal([]string{"live-video-to-video/model-a", "text-to-image/model-b"}, orch1Resp.Capabilities) + require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "text-to-image/model-b"}, discoveryRunnerApps(t, orch1Resp)) requireNotContainsDiscoveryField(t, body, "unknown_field") discoveredResp := discoveryResponseByAddress(t, resp, "https://discovered.example.com:8935") require.Equal(float32(common.Score_Trusted), discoveredResp.Score) require.Equal([]string{"live-video-to-video/model-a"}, discoveredResp.Capabilities) - require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveredResp)) + require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveredResp)) require.Equal(1, discoveredResp.Runners[1].Capacity) require.Equal(0, discoveredResp.Runners[1].CapacityUsed) orch2Resp := discoveryResponseByAddress(t, resp, "https://orch2.example.com:8935") require.Equal(float32(common.Score_Trusted), orch2Resp.Score) - require.Equal([]string{"text-to-image/model-b"}, orch2Resp.Capabilities) - require.Equal([]string{"text-to-image/model-b"}, discoveryRunnerApps(t, orch2Resp)) + require.Equal([]string{"live-video-to-video/model-a", "text-to-image/model-b"}, orch2Resp.Capabilities) + require.Equal([]string{"text-to-image/model-b", "live-video-to-video/model-a"}, discoveryRunnerApps(t, orch2Resp)) + orch3Resp := discoveryResponseByAddress(t, resp, "https://orch3.example.com:8935") + require.Equal(float32(common.Score_Trusted), orch3Resp.Score) + require.Equal([]string{"live-video-to-video/model-a"}, orch3Resp.Capabilities) + require.Equal([]string{"live-video-to-video/model-a"}, discoveryRunnerApps(t, orch3Resp)) capsReq := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=live-video-to-video/model-a", nil) capsRR := httptest.NewRecorder() @@ -1241,9 +1245,11 @@ func TestRemoteSigner_Discovery(t *testing.T) { require.Equal(http.StatusOK, capsRR.Code) var capsResp []discoveryResponse require.NoError(json.NewDecoder(capsRR.Body).Decode(&capsResp)) - require.Len(capsResp, 2) - require.Equal([]string{"live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://orch1.example.com:8935"))) - require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://discovered.example.com:8935"))) + require.Len(capsResp, 4) + require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "text-to-image/model-b"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://orch1.example.com:8935"))) + require.Equal([]string{"live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a", "live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://discovered.example.com:8935"))) + require.Equal([]string{"text-to-image/model-b", "live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://orch2.example.com:8935"))) + require.Equal([]string{"live-video-to-video/model-a"}, discoveryRunnerApps(t, discoveryResponseByAddress(t, capsResp, "https://orch3.example.com:8935"))) repeatedReq := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=live-video-to-video/model-a&caps=text-to-image/model-b", nil) repeatedRR := httptest.NewRecorder() @@ -1251,10 +1257,11 @@ func TestRemoteSigner_Discovery(t *testing.T) { require.Equal(http.StatusOK, repeatedRR.Code) var repeatedResp []discoveryResponse require.NoError(json.NewDecoder(repeatedRR.Body).Decode(&repeatedResp)) - require.Len(repeatedResp, 3) + require.Len(repeatedResp, 4) discoveryResponseByAddress(t, repeatedResp, "https://orch1.example.com:8935") discoveryResponseByAddress(t, repeatedResp, "https://discovered.example.com:8935") discoveryResponseByAddress(t, repeatedResp, "https://orch2.example.com:8935") + discoveryResponseByAddress(t, repeatedResp, "https://orch3.example.com:8935") // If refresh only receives invalid or ineligible network capability entries, // cache should remain empty and discovery should return service unavailable. @@ -1375,7 +1382,7 @@ func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t Discovery: discoveryRaw(t, `[{ "address": "https://runner-derived.example.com:8935", "runners": [ - {"url":"https://runner-derived.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} + {"url":"https://runner-derived.example.com:8935/apps/runner/session","app":"custom-runner-v1","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} ] }]`), }, @@ -1387,7 +1394,7 @@ func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t } ls := &LivepeerServer{} - req := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=live-video-to-video/scope", nil) + req := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=custom-runner-v1", nil) rr := httptest.NewRecorder() ls.GetOrchestrators(rdp, rr, req) @@ -1396,11 +1403,11 @@ func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t require.NoError(json.NewDecoder(rr.Body).Decode(&resp)) require.Len(resp, 1) require.Equal("https://runner-derived.example.com:8935", resp[0].Address) - require.Equal([]string{"live-video-to-video/scope"}, resp[0].Capabilities) - require.Equal([]string{"live-video-to-video/scope"}, discoveryRunnerApps(t, resp[0])) + require.Equal([]string{"custom-runner-v1"}, resp[0].Capabilities) + require.Equal([]string{"custom-runner-v1"}, discoveryRunnerApps(t, resp[0])) } -func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsEligibleRunnersWhenRPCCapabilitiesMissing(t *testing.T) { +func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsValidRunnersWhenRPCCapabilitiesMissing(t *testing.T) { require := require.New(t) capability := core.Capability_LiveVideoToVideo @@ -1419,6 +1426,7 @@ func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsEligibleRunnersWhenRPCCapabi {"url":"https://mixed-runner-derived.example.com:8935/usd","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"USD"}}, {"url":"https://mixed-runner-derived.example.com:8935/missing-price","app":"live-video-to-video/scope"}, {"url":"https://mixed-runner-derived.example.com:8935/invalid-pixels","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":0,"unit":"WEI"}}, + {"url":"https://mixed-runner-derived.example.com:8935/empty-app","app":" ","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}}, {"url":"https://mixed-runner-derived.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} ] }]`), @@ -1441,9 +1449,10 @@ func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsEligibleRunnersWhenRPCCapabi require.Len(resp, 1) require.Equal("https://mixed-runner-derived.example.com:8935", resp[0].Address) require.Equal([]string{"live-video-to-video/scope"}, resp[0].Capabilities) - require.Len(resp[0].Runners, 1) - require.Equal("https://mixed-runner-derived.example.com:8935/apps/runner/session", resp[0].Runners[0].URL) - require.Equal([]string{"live-video-to-video/scope"}, discoveryRunnerApps(t, resp[0])) + require.Len(resp[0].Runners, 2) + require.Equal("https://mixed-runner-derived.example.com:8935/too-expensive", resp[0].Runners[0].URL) + require.Equal("https://mixed-runner-derived.example.com:8935/apps/runner/session", resp[0].Runners[1].URL) + require.Equal([]string{"live-video-to-video/scope", "live-video-to-video/scope"}, discoveryRunnerApps(t, resp[0])) } func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { @@ -1451,7 +1460,11 @@ func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { capability := core.Capability_LiveVideoToVideo modelID := "scope" - BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(200, 995328000000))) + BroadcastCfg.SetMaxPrice(core.NewFixedPrice(big.NewRat(200, 995328000000))) + defer BroadcastCfg.SetMaxPrice(nil) + // Capability/model max prices do not apply to runner discovery. This would + // reject price-ok under the old runner capability lookup path. + BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(1, 995328000000))) defer BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, nil) node := &core.LivepeerNode{} @@ -1462,10 +1475,12 @@ func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { "address": "https://filtered.example.com:8935", "runners": [ {"url":"https://filtered.example.com:8935/too-expensive","app":"live-video-to-video/scope","price_info":{"price_per_unit":201,"pixels_per_unit":995328000000,"unit":"WEI"}}, + {"url":"https://filtered.example.com:8935/price-ok","app":"live-video-to-video/scope","price_info":{"price_per_unit":199,"pixels_per_unit":995328000000,"unit":"WEI"}}, {"url":"https://filtered.example.com:8935/usd","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"USD"}}, {"url":"https://filtered.example.com:8935/missing-price","app":"live-video-to-video/scope"}, {"url":"https://filtered.example.com:8935/non-positive-price","app":"live-video-to-video/scope","price_info":{"price_per_unit":0,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://filtered.example.com:8935/non-positive-pixels","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":0,"unit":"WEI"}} + {"url":"https://filtered.example.com:8935/non-positive-pixels","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":0,"unit":"WEI"}}, + {"url":"https://filtered.example.com:8935/empty-app","app":"","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} ] }]`), }, @@ -1481,8 +1496,13 @@ func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { rr := httptest.NewRecorder() ls.GetOrchestrators(rdp, rr, req) - require.Equal(http.StatusServiceUnavailable, rr.Code) + require.Equal(http.StatusOK, rr.Code) require.Equal("application/json", rr.Header().Get("Content-Type")) + var resp []discoveryResponse + require.NoError(json.NewDecoder(rr.Body).Decode(&resp)) + require.Len(resp, 1) + require.Len(resp[0].Runners, 1) + require.Equal("https://filtered.example.com:8935/price-ok", resp[0].Runners[0].URL) } func discoveryRaw(t *testing.T, data string) json.RawMessage { @@ -1544,7 +1564,7 @@ func TestRemoteDiscoveryRunnerDuplicateComparison(t *testing.T) { require.False(t, sameRemoteDiscoveryRunner(base, capacityChanged)) } -func TestRemoteSigner_Discovery_RequiresAdvertisedPriceWithoutMaxPrice(t *testing.T) { +func TestRemoteSigner_Discovery_RunnersDoNotRequireTopLevelAdvertisedPrice(t *testing.T) { require := require.New(t) capability := core.Capability_LiveVideoToVideo @@ -1611,10 +1631,13 @@ func TestRemoteSigner_Discovery_RequiresAdvertisedPriceWithoutMaxPrice(t *testin require.Equal(http.StatusOK, rr.Code) var resp []discoveryResponse require.NoError(json.NewDecoder(rr.Body).Decode(&resp)) - require.Len(resp, 1) - require.Equal("https://priced.example.com:8935", resp[0].Address) - require.Equal([]string{"live-video-to-video/model-priced"}, resp[0].Capabilities) - require.Equal([]string{"live-video-to-video/model-priced"}, discoveryRunnerApps(t, resp[0])) + require.Len(resp, 2) + pricedResp := discoveryResponseByAddress(t, resp, "https://priced.example.com:8935") + require.Equal([]string{"live-video-to-video/model-priced", "live-video-to-video/model-unpriced"}, pricedResp.Capabilities) + require.Equal([]string{"live-video-to-video/model-priced", "live-video-to-video/model-unpriced"}, discoveryRunnerApps(t, pricedResp)) + unpricedNodeResp := discoveryResponseByAddress(t, resp, "https://unpriced.example.com:8935") + require.Equal([]string{"live-video-to-video/model-priced", "live-video-to-video/model-unpriced"}, unpricedNodeResp.Capabilities) + require.Equal([]string{"live-video-to-video/model-priced", "live-video-to-video/model-unpriced"}, discoveryRunnerApps(t, unpricedNodeResp)) capsReq := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=live-video-to-video/model-priced", nil) capsRR := httptest.NewRecorder() @@ -1622,16 +1645,19 @@ func TestRemoteSigner_Discovery_RequiresAdvertisedPriceWithoutMaxPrice(t *testin require.Equal(http.StatusOK, capsRR.Code) var capsResp []discoveryResponse require.NoError(json.NewDecoder(capsRR.Body).Decode(&capsResp)) - require.Len(capsResp, 1) - require.Equal("https://priced.example.com:8935", capsResp[0].Address) + require.Len(capsResp, 2) + discoveryResponseByAddress(t, capsResp, "https://priced.example.com:8935") + discoveryResponseByAddress(t, capsResp, "https://unpriced.example.com:8935") unpricedReq := httptest.NewRequest(http.MethodGet, "/discover-orchestrators?caps=live-video-to-video/model-unpriced", nil) unpricedRR := httptest.NewRecorder() ls.GetOrchestrators(rdp, unpricedRR, unpricedReq) require.Equal(http.StatusOK, unpricedRR.Code) - var unpricedResp []discoveryResponse - require.NoError(json.NewDecoder(unpricedRR.Body).Decode(&unpricedResp)) - require.Empty(unpricedResp) + var unpricedCapsResp []discoveryResponse + require.NoError(json.NewDecoder(unpricedRR.Body).Decode(&unpricedCapsResp)) + require.Len(unpricedCapsResp, 2) + discoveryResponseByAddress(t, unpricedCapsResp, "https://priced.example.com:8935") + discoveryResponseByAddress(t, unpricedCapsResp, "https://unpriced.example.com:8935") } func TestRemoteSigner_Discovery_RefreshesAfterInterval(t *testing.T) { From 004450d24c821a527d570e486010a7eb5e89b124 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Wed, 15 Jul 2026 12:39:57 -0700 Subject: [PATCH 2/2] Add 'live' pricing type --- ai/runner/live_runner.go | 119 ++++++++++++-------- ai/runner/live_runner_test.go | 182 +++++++++++++++++++++++++------ server/ai_http.go | 14 ++- server/ai_http_test.go | 109 +++++++++++++------ server/cliserver_test.go | 6 +- server/remote_discovery.go | 81 +++++++------- server/remote_signer.go | 33 ++++-- server/remote_signer_test.go | 198 +++++++++++++++++++++++++--------- 8 files changed, 532 insertions(+), 210 deletions(-) diff --git a/ai/runner/live_runner.go b/ai/runner/live_runner.go index b08d37ec37..b51f7f3d7b 100644 --- a/ai/runner/live_runner.go +++ b/ai/runner/live_runner.go @@ -79,9 +79,47 @@ type LiveRunnerGPU struct { } type LiveRunnerPriceInfo struct { - PricePerUnit int64 `json:"price_per_unit"` - PixelsPerUnit int64 `json:"pixels_per_unit"` - Unit string `json:"unit,omitempty"` + Price json.Number `json:"price"` + Currency string `json:"currency,omitempty"` + Unit string `json:"unit,omitempty"` +} + +func (p LiveRunnerPriceInfo) priceRat() (*big.Rat, error) { + price := strings.TrimSpace(p.Price.String()) + if price == "" { + return nil, fmt.Errorf("price_info.price is required") + } + rat, ok := new(big.Rat).SetString(price) + if !ok || rat.Sign() <= 0 { + return nil, fmt.Errorf("price_info.price must be a positive decimal") + } + return rat, nil +} + +func normalizeLiveRunnerPriceInfo(priceInfo LiveRunnerPriceInfo) (LiveRunnerPriceInfo, error) { + if _, err := priceInfo.priceRat(); err != nil { + return LiveRunnerPriceInfo{}, err + } + currency := strings.ToLower(strings.TrimSpace(priceInfo.Currency)) + if currency == "" { + currency = "usd" + } + if currency != "usd" { + return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.currency must be usd") + } + unit := strings.ToLower(strings.TrimSpace(priceInfo.Unit)) + if unit == "" { + unit = "hour" + } + switch unit { + case "hour": + case "720p": + default: + return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.unit must be hour or 720p") + } + priceInfo.Currency = currency + priceInfo.Unit = unit + return priceInfo, nil } type LiveRunnerHeartbeatRequest struct { @@ -831,14 +869,11 @@ func (r *LiveRunnerRegistry) normalizeHeartbeat(runnerID string, req LiveRunnerH return req, nil } - if req.PriceInfo.PixelsPerUnit <= 0 || req.PriceInfo.PricePerUnit <= 0 { - return LiveRunnerHeartbeatRequest{}, fmt.Errorf("runner must include positive price_info") - } - unit := strings.ToUpper(strings.TrimSpace(req.PriceInfo.Unit)) - if unit != "USD" && unit != "WEI" { - return LiveRunnerHeartbeatRequest{}, fmt.Errorf("price_info.unit must be USD or WEI") + priceInfo, err := normalizeLiveRunnerPriceInfo(req.PriceInfo) + if err != nil { + return LiveRunnerHeartbeatRequest{}, err } - req.PriceInfo.Unit = unit + req.PriceInfo = priceInfo return req, nil } @@ -923,15 +958,11 @@ func (r *LiveRunnerRegistry) PaymentInfo(runnerID string) (*LiveRunnerPriceInfo, if runner.offchain { return nil, nil } - priceInfo := runner.PriceInfo - if runner.converter != nil { - converted, err := convertedPriceInfo(runner.converter) - if err != nil { - return nil, err - } - priceInfo = converted + priceInfo, err := runner.convertPrice() + if err != nil { + return nil, err } - if priceInfo == (LiveRunnerPriceInfo{}) || priceInfo.PricePerUnit <= 0 || priceInfo.PixelsPerUnit <= 0 { + if _, err := priceInfo.priceRat(); err != nil { return nil, nil } return &priceInfo, nil @@ -1577,14 +1608,9 @@ func validSecret(got, want string) bool { func (runner *liveRunner) discoveryRunner() LiveRunnerDiscoveryRunner { // discoveryRunner requires runner.mu. - priceInfo := runner.PriceInfo - if runner.converter != nil { - converted, err := convertedPriceInfo(runner.converter) - if err != nil { - slog.Error("error reading converted live runner price", "app", runner.App, "endpoint", runner.RunnerURL, "err", err) - } else { - priceInfo = converted - } + priceInfo, err := runner.convertPrice() + if err != nil { + slog.Error("error reading converted live runner price", "app", runner.App, "endpoint", runner.RunnerURL, "err", err) } var discoveryPriceInfo *LiveRunnerPriceInfo if !runner.offchain && priceInfo != (LiveRunnerPriceInfo{}) { @@ -1623,7 +1649,7 @@ func (runner *liveRunner) updatePriceConverterLocked() { runner.converter = converter if priceChanged { slog.Debug("live runner price set", "runner_id", runner.RunnerID, "app", runner.App, - "price_per_unit", runner.PriceInfo.PricePerUnit, "pixels_per_unit", runner.PriceInfo.PixelsPerUnit, "price_unit", runner.PriceInfo.Unit) + "price", runner.PriceInfo.Price, "currency", runner.PriceInfo.Currency, "unit", runner.PriceInfo.Unit) } } @@ -1637,32 +1663,39 @@ func (runner *liveRunner) stopPriceConverterLocked() { } func newConverterForRunner(priceInfo LiveRunnerPriceInfo) (*core.AutoConvertedPrice, error) { - if strings.ToUpper(priceInfo.Unit) != "USD" { - return nil, nil + priceInfo, err := normalizeLiveRunnerPriceInfo(priceInfo) + if err != nil { + return nil, err } - usdPerPixel := usdPerPixelFromUSDPerHour(priceInfo) - return core.NewAutoConvertedPrice("USD", usdPerPixel, nil) -} - -func usdPerPixelFromUSDPerHour(priceInfo LiveRunnerPriceInfo) *big.Rat { - pixelsPerHour := 1280 * 720 * 30 * 3600 // 720p @ 30fps - usdPerHour := new(big.Rat).SetFrac64(priceInfo.PricePerUnit, priceInfo.PixelsPerUnit) - return new(big.Rat).Quo(usdPerHour, new(big.Rat).SetInt64(int64(pixelsPerHour))) + usdPerHour, err := priceInfo.priceRat() + if err != nil { + return nil, err + } + denom := int64(3600) + if priceInfo.Unit == "720p" { + denom *= 1280 * 720 * 30 // 720p @ 30fps + } + usdPrice := new(big.Rat).Quo(usdPerHour, new(big.Rat).SetInt64(denom)) + return core.NewAutoConvertedPrice("USD", usdPrice, nil) } -func convertedPriceInfo(converter *core.AutoConvertedPrice) (LiveRunnerPriceInfo, error) { - if converter == nil { +func (runner *liveRunner) convertPrice() (LiveRunnerPriceInfo, error) { + if runner.converter == nil { return LiveRunnerPriceInfo{}, nil } - price, err := common.PriceToInt64(converter.Value()) + price, err := common.PriceToInt64(runner.converter.Value()) if err != nil { return LiveRunnerPriceInfo{}, err } + convertedUnit := "seconds" + if runner.PriceInfo.Unit == "720p" { + convertedUnit = "720p-pixel-seconds" + } return LiveRunnerPriceInfo{ - PricePerUnit: price.Num().Int64(), - PixelsPerUnit: price.Denom().Int64(), - Unit: "WEI", + Price: json.Number(fmt.Sprintf("%d", price.Num().Int64())), + Currency: "wei", + Unit: convertedUnit, }, nil } diff --git a/ai/runner/live_runner_test.go b/ai/runner/live_runner_test.go index 3730c55ded..36fab44c80 100644 --- a/ai/runner/live_runner_test.go +++ b/ai/runner/live_runner_test.go @@ -583,18 +583,82 @@ func TestLiveRunnerRegistry_OnchainRequiresStaticRunnerPriceInfo(t *testing.T) { } } -func TestLiveRunnerRegistry_OnchainAcceptsWEIPrice(t *testing.T) { +func TestLiveRunnerRegistry_OnchainDefaultsPriceCurrencyAndUnit(t *testing.T) { registry := newOnchainLiveRunnerTestRegistry() - req := liveRunnerTestHeartbeat("runner-wei-price") - req.PriceInfo = LiveRunnerPriceInfo{PricePerUnit: 10, PixelsPerUnit: 2, Unit: "wei"} + req := liveRunnerTestHeartbeat("runner-usd-price") + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10")} liveRunnerTestRegister(t, registry, req) - runners := registry.Runners() - if len(runners) != 1 { - t.Fatalf("expected one runner, got %d", len(runners)) + runner, unlock, err := registry.lockLiveRunner("runner-usd-price") + if err != nil { + t.Fatal(err) } - if runners[0].PriceInfo == nil || *runners[0].PriceInfo != (LiveRunnerPriceInfo{PricePerUnit: 10, PixelsPerUnit: 2, Unit: "WEI"}) { - t.Fatalf("unexpected runner price info: %+v", runners[0].PriceInfo) + defer unlock() + if runner.PriceInfo != (LiveRunnerPriceInfo{Price: json.Number("10"), Currency: "usd", Unit: "hour"}) { + t.Fatalf("unexpected runner price info: %+v", runner.PriceInfo) + } +} + +func TestLiveRunnerRegistry_OnchainNormalizes720pPriceUnit(t *testing.T) { + registry := newOnchainLiveRunnerTestRegistry() + req := liveRunnerTestHeartbeat("runner-720p-price") + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10"), Unit: "720p"} + + liveRunnerTestRegister(t, registry, req) + runner, unlock, err := registry.lockLiveRunner("runner-720p-price") + if err != nil { + t.Fatal(err) + } + defer unlock() + if runner.PriceInfo != (LiveRunnerPriceInfo{Price: json.Number("10"), Currency: "usd", Unit: "720p"}) { + t.Fatalf("unexpected runner price info: %+v", runner.PriceInfo) + } +} + +func TestLiveRunnerRegistry_OnchainRejectsInvalidPriceInfo(t *testing.T) { + tests := []struct { + name string + priceInfo LiveRunnerPriceInfo + wantErr string + }{ + { + name: "zero price", + priceInfo: LiveRunnerPriceInfo{Price: json.Number("0")}, + wantErr: "price_info.price", + }, + { + name: "negative price", + priceInfo: LiveRunnerPriceInfo{Price: json.Number("-1")}, + wantErr: "price_info.price", + }, + { + name: "unsupported currency", + priceInfo: LiveRunnerPriceInfo{Price: json.Number("10"), Currency: "eur"}, + wantErr: "price_info.currency", + }, + { + name: "unsupported unit", + priceInfo: LiveRunnerPriceInfo{Price: json.Number("10"), Unit: "minute"}, + wantErr: "price_info.unit", + }, + { + name: "invalid decimal", + priceInfo: LiveRunnerPriceInfo{Price: json.Number("wat")}, + wantErr: "price_info.price", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + registry := newOnchainLiveRunnerTestRegistry() + req := liveRunnerTestHeartbeat("runner-invalid-price") + req.PriceInfo = tt.priceInfo + + _, err := registry.Heartbeat(req, liveRunnerTestBootstrapSecret) + if !isRunnerErrorStatus(err, http.StatusBadRequest) || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("expected bad request containing %q, got %v", tt.wantErr, err) + } + }) } } @@ -605,7 +669,7 @@ func TestLiveRunnerRegistry_OffchainIgnoresUSDPriceWatcher(t *testing.T) { registry := newLiveRunnerTestRegistry() req := liveRunnerTestHeartbeat("runner-offchain-usd") - req.PriceInfo = LiveRunnerPriceInfo{PricePerUnit: 10, PixelsPerUnit: 1, Unit: "USD"} + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10")} liveRunnerTestRegister(t, registry, req) runners := registry.Runners() @@ -1090,26 +1154,84 @@ func TestLiveRunnerRegistry_ConvertsUSDToWEI(t *testing.T) { registry := newOnchainLiveRunnerTestRegistry() req := liveRunnerTestHeartbeat("runner-1") - req.PriceInfo = LiveRunnerPriceInfo{PricePerUnit: 10, PixelsPerUnit: 1, Unit: "USD"} + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10")} liveRunnerTestRegister(t, registry, req) - runners := registry.Runners() - if len(runners) != 1 { - t.Fatalf("expected one runner, got %d", len(runners)) + paymentInfo, err := registry.PaymentInfo("runner-1") + if err != nil { + t.Fatal(err) } - if runners[0].PriceInfo == nil { - t.Fatal("expected runner price info") + if paymentInfo == nil { + t.Fatal("expected payment price info") } - got := big.NewRat(runners[0].PriceInfo.PricePerUnit, runners[0].PriceInfo.PixelsPerUnit) - expected, err := common.PriceToInt64(big.NewRat(5_000_000_000_000_000, 1280*720*30*3600)) + expected, err := common.PriceToInt64(big.NewRat(5_000_000_000_000_000, 3600)) if err != nil { t.Fatal(err) } - if got.Cmp(expected) != 0 { - t.Fatalf("unexpected converted price: got=%s want=%s", got, expected) + if got := paymentInfo.Price.String(); got != expected.Num().String() { + t.Fatalf("unexpected converted price: got=%s want=%s", got, expected.Num().String()) } - if runners[0].PriceInfo.Unit != "WEI" { - t.Fatalf("unexpected converted unit: %s", runners[0].PriceInfo.Unit) + if paymentInfo.Currency != "wei" || paymentInfo.Unit != "seconds" { + t.Fatalf("unexpected converted price info: %+v", paymentInfo) + } +} + +func TestLiveRunnerRegistry_Converts720pUnitAsPerPixelPrice(t *testing.T) { + prevWatcher := core.PriceFeedWatcher + core.PriceFeedWatcher = stubPriceFeedWatcher{price: eth.PriceData{Price: big.NewRat(2000, 1)}} + defer func() { core.PriceFeedWatcher = prevWatcher }() + + registry := newOnchainLiveRunnerTestRegistry() + req := liveRunnerTestHeartbeat("runner-720p") + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10"), Unit: "720p"} + liveRunnerTestRegister(t, registry, req) + + paymentInfo, err := registry.PaymentInfo("runner-720p") + if err != nil { + t.Fatal(err) + } + if paymentInfo == nil { + t.Fatal("expected payment price info") + } + expected, err := common.PriceToInt64(big.NewRat(5_000_000_000_000_000, 3600*1280*720*30)) + if err != nil { + t.Fatal(err) + } + if got := paymentInfo.Price.String(); got != expected.Num().String() { + t.Fatalf("unexpected converted price: got=%s want=%s", got, expected.Num().String()) + } + if paymentInfo.Currency != "wei" || paymentInfo.Unit != "720p-pixel-seconds" { + t.Fatalf("unexpected converted 720p price info: %+v", paymentInfo) + } +} + +func TestLiveRunnerRegistry_ScopeRunnerUsesLivePriceConversion(t *testing.T) { + prevWatcher := core.PriceFeedWatcher + core.PriceFeedWatcher = stubPriceFeedWatcher{price: eth.PriceData{Price: big.NewRat(2000, 1)}} + defer func() { core.PriceFeedWatcher = prevWatcher }() + + registry := newOnchainLiveRunnerTestRegistry() + req := liveRunnerTestHeartbeat("scope-runner") + req.App = "live-video-to-video/scope" + req.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10")} + liveRunnerTestRegister(t, registry, req) + + paymentInfo, err := registry.PaymentInfo("scope-runner") + if err != nil { + t.Fatal(err) + } + if paymentInfo == nil { + t.Fatal("expected payment price info") + } + expected, err := common.PriceToInt64(big.NewRat(5_000_000_000_000_000, 3600)) + if err != nil { + t.Fatal(err) + } + if got := paymentInfo.Price.String(); got != expected.Num().String() { + t.Fatalf("unexpected converted scope live runner price: got=%s want=%s", got, expected.Num().String()) + } + if paymentInfo.Currency != "wei" || paymentInfo.Unit != "seconds" { + t.Fatalf("unexpected converted scope live runner price info: %+v", paymentInfo) } } @@ -1117,24 +1239,22 @@ func TestLiveRunnerRegistry_SharedURLAppKeepsPerRunnerPrices(t *testing.T) { registry := newOnchainLiveRunnerTestRegistry() req1 := liveRunnerTestHeartbeat("runner-1") - req1.PriceInfo = LiveRunnerPriceInfo{PricePerUnit: 10, PixelsPerUnit: 1, Unit: "WEI"} + req1.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("10")} req2 := liveRunnerTestHeartbeat("runner-2") - req2.PriceInfo = LiveRunnerPriceInfo{PricePerUnit: 20, PixelsPerUnit: 1, Unit: "WEI"} + req2.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("20")} resp1 := liveRunnerTestRegister(t, registry, req1) liveRunnerTestRegister(t, registry, req2) if err := registry.Unregister("runner-1", resp1.HeartbeatSecret); err != nil { t.Fatal(err) } - runners := registry.Runners() - if len(runners) != 1 { - t.Fatalf("expected one remaining runner, got %d", len(runners)) - } - if runners[0].PriceInfo == nil { - t.Fatal("expected runner price info") + runner, unlock, err := registry.lockLiveRunner("runner-2") + if err != nil { + t.Fatal(err) } - if got := runners[0].PriceInfo.PricePerUnit; got != 20 { - t.Fatalf("expected remaining runner price to stay isolated, got %d", got) + defer unlock() + if got := runner.PriceInfo.Price.String(); got != "20" { + t.Fatalf("expected remaining runner price to stay isolated, got %s", got) } } diff --git a/server/ai_http.go b/server/ai_http.go index a23ddf2ea5..e606c27e31 100644 --- a/server/ai_http.go +++ b/server/ai_http.go @@ -411,9 +411,13 @@ func (h *lphttp) runnerOrchInfo(sender ethcommon.Address, priceInfo *runner.Live if priceInfo == nil { return nil, fmt.Errorf("missing live runner price info") } + pricePerUnit, err := priceInfo.Price.Int64() + if err != nil || pricePerUnit <= 0 { + return nil, fmt.Errorf("invalid live runner price info") + } netPriceInfo := &lpnet.PriceInfo{ - PricePerUnit: priceInfo.PricePerUnit, - PixelsPerUnit: priceInfo.PixelsPerUnit, + PricePerUnit: pricePerUnit, + PixelsPerUnit: 1, } params, err := h.orchestrator.TicketParams(sender, netPriceInfo) if err != nil { @@ -810,9 +814,9 @@ func (h *lphttp) DiscoverLiveRunners(w http.ResponseWriter, r *http.Request) { glog.Errorf("error converting discovery price for capability %v modelID=%v err=%v", core.CapabilityNameLookup[capability], modelID, err) } else { priceInfo = &runner.LiveRunnerPriceInfo{ - PricePerUnit: priceInt64.Num().Int64(), - PixelsPerUnit: priceInt64.Denom().Int64(), - Unit: "WEI", + Price: json.Number(strconv.FormatInt(priceInt64.Num().Int64(), 10)), + Currency: "wei", + Unit: "720p-pixel-seconds", } } } diff --git a/server/ai_http_test.go b/server/ai_http_test.go index 6de93305fc..d5741371c6 100644 --- a/server/ai_http_test.go +++ b/server/ai_http_test.go @@ -193,9 +193,7 @@ func TestLiveRunnerDiscoveryEndpoint(t *testing.T) { App: "live-video-to-video/scope", Capacity: 2, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "USD", + Price: json.Number("10"), }, }, lp.orchestrator.RegistrationSecret()) require.NoError(t, err) @@ -300,9 +298,7 @@ func TestLiveRunnerDiscoveryOnchainIncludesPriceInfo(t *testing.T) { App: "live-video-to-video/scope", Capacity: 2, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "USD", + Price: json.Number("10"), }, }, lp.orchestrator.RegistrationSecret()) require.NoError(t, err) @@ -321,10 +317,51 @@ func TestLiveRunnerDiscoveryOnchainIncludesPriceInfo(t *testing.T) { require.Equal(t, 0, resp[0].Runners[0].CapacityUsed) require.Equal(t, 2, resp[0].Runners[0].CapacityAvailable) require.NotNil(t, resp[0].Runners[0].PriceInfo) - require.NotZero(t, resp[0].Runners[0].PriceInfo.PricePerUnit) + require.Equal(t, "1388888888888", resp[0].Runners[0].PriceInfo.Price.String()) + require.Equal(t, "wei", resp[0].Runners[0].PriceInfo.Currency) + require.Equal(t, "seconds", resp[0].Runners[0].PriceInfo.Unit) require.Contains(t, w.Body.String(), "price_info") } +func TestLiveRunnerDiscoveryOnchainConverts720pPriceToPixel(t *testing.T) { + lp := newLiveRunnerHTTP(t, true) + manager := runner.NewLiveRunnerRegistry(runner.LiveRunnerRegistryConfig{Host: testLiveRunnerHost{orchestrator: lp.orchestrator}, Onchain: true}) + t.Cleanup(manager.Stop) + lp.node.LiveRunnerManager = manager + trickleBaseURL := lp.orchestrator.ServiceURI().JoinPath(TrickleHTTPPath).String() + manager.SetTrickleServer(lp.trickleSrv, trickleBaseURL, trickleBaseURL) + prevWatcher := core.PriceFeedWatcher + core.PriceFeedWatcher = stubPriceFeedWatcher{price: eth.PriceData{Price: big.NewRat(2000, 1)}} + defer func() { core.PriceFeedWatcher = prevWatcher }() + _, err := manager.Heartbeat(runner.LiveRunnerHeartbeatRequest{ + RunnerID: t.Name(), + RunnerURL: "https://runner.example.com", + Version: "1.2.3", + Status: "ready", + App: "live-video-to-video/scope", + Capacity: 2, + PriceInfo: runner.LiveRunnerPriceInfo{ + Price: json.Number("0.001990656"), + Unit: "720p", + }, + }, lp.orchestrator.RegistrationSecret()) + require.NoError(t, err) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/discovery", nil) + lp.ServeHTTP(w, req) + + require.Equal(t, http.StatusOK, w.Code) + var resp []liveRunnerDiscoveryEntry + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.Len(t, resp, 1) + require.Len(t, resp[0].Runners, 1) + require.NotNil(t, resp[0].Runners[0].PriceInfo) + require.Equal(t, "10", resp[0].Runners[0].PriceInfo.Price.String()) + require.Equal(t, "wei", resp[0].Runners[0].PriceInfo.Currency) + require.Equal(t, "720p-pixel-seconds", resp[0].Runners[0].PriceInfo.Unit) +} + func TestLiveRunnerDiscoveryServerlessWorker(t *testing.T) { lp := newServerlessLiveRunnerHTTP(t, false, 3) lp.node.SetBasePriceForCap("default", core.Capability_LiveVideoToVideo, "scope", core.NewFixedPrice(big.NewRat(7, 1))) @@ -350,8 +387,9 @@ func TestLiveRunnerDiscoveryServerlessWorker(t *testing.T) { require.Equal(t, 0, discoveryRunner.CapacityUsed) require.Equal(t, 3, discoveryRunner.CapacityAvailable) require.NotNil(t, discoveryRunner.PriceInfo) - require.Equal(t, int64(7), discoveryRunner.PriceInfo.PricePerUnit) - require.Equal(t, int64(1), discoveryRunner.PriceInfo.PixelsPerUnit) + require.Equal(t, "7", discoveryRunner.PriceInfo.Price.String()) + require.Equal(t, "wei", discoveryRunner.PriceInfo.Currency) + require.Equal(t, "720p-pixel-seconds", discoveryRunner.PriceInfo.Unit) } func TestLiveRunnerDiscoveryReturnsHeartbeatAndServerlessRunners(t *testing.T) { @@ -370,9 +408,7 @@ func TestLiveRunnerDiscoveryReturnsHeartbeatAndServerlessRunners(t *testing.T) { App: "live-video-to-video/scope", Capacity: 1, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "USD", + Price: json.Number("10"), }, }, lp.orchestrator.RegistrationSecret()) require.NoError(t, err) @@ -405,9 +441,7 @@ func TestLiveRunnerHeartbeat(t *testing.T) { App: "live-video-to-video/scope", Capacity: 3, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "USD", + Price: json.Number("10"), }, SessionIDs: []string{"runner-reported-session"}, }) @@ -494,9 +528,7 @@ func TestLiveRunnerHeartbeatUsesLiveRunnerAddr(t *testing.T) { App: "live-video-to-video/scope", Capacity: 1, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "WEI", + Price: json.Number("10"), }, }) require.NoError(t, err) @@ -512,26 +544,22 @@ func TestLiveRunnerHeartbeatUsesLiveRunnerAddr(t *testing.T) { require.Equal(t, "http://go-livepeer:8935", resp.Orchestrator) } -func TestLiveRunnerHeartbeatRejectsMissingPriceInfoUnit(t *testing.T) { - lp := newLiveRunnerHTTP(t, true) - manager := runner.NewLiveRunnerRegistry(runner.LiveRunnerRegistryConfig{Host: testLiveRunnerHost{orchestrator: lp.orchestrator}, Onchain: true}) - t.Cleanup(manager.Stop) - lp.node.LiveRunnerManager = manager +func TestLiveRunnerHeartbeatDefaultsMissingPriceInfoCurrencyAndUnit(t *testing.T) { + lp := newLiveRunnerHTTPOnchain(t) body := []byte(fmt.Sprintf(`{ "runner_id":%q, "runner_url":"https://runner.example.com", - "price_info":{"price_per_unit":1,"pixels_per_unit":1}, + "price_info":{"price":1}, "app":"live-video-to-video/scope", - "capacity":1, - "pricing":{"usd_per_hour":10} + "capacity":1 }`, t.Name())) w := httptest.NewRecorder() req := httptest.NewRequest(http.MethodPost, "/runners/heartbeat", bytes.NewReader(body)) + req.Header.Set("Authorization", lp.orchestrator.RegistrationSecret()) lp.ServeHTTP(w, req) - require.Equal(t, http.StatusBadRequest, w.Code) - require.Contains(t, w.Body.String(), "price_info.unit") + require.Equal(t, http.StatusOK, w.Code) } func TestLiveRunnerEndpointsUnsupportedWithoutManager(t *testing.T) { @@ -604,6 +632,22 @@ func TestLiveRunnerReserveSessionOnchainReturnsPaymentChallenge(t *testing.T) { require.Nil(t, oInfo.GetCapabilities()) } +func TestRunnerOrchInfoUsesWeiSecondsPrice(t *testing.T) { + lp := newLiveRunnerHTTPOnchain(t) + orch := lp.orchestrator.(*stubOrchestrator) + + oInfo, err := lp.runnerOrchInfo(orch.Address(), &runner.LiveRunnerPriceInfo{ + Price: json.Number("10"), + Currency: "wei", + Unit: "seconds", + }) + + require.NoError(t, err) + require.NotNil(t, oInfo.GetPriceInfo()) + require.Equal(t, int64(10), oInfo.GetPriceInfo().GetPricePerUnit()) + require.Equal(t, int64(1), oInfo.GetPriceInfo().GetPixelsPerUnit()) +} + func TestLiveRunnerReserveSessionOnchainAcceptsPaidReservation(t *testing.T) { lp := newLiveRunnerHTTPOnchain(t) registerLiveRunnerForSession(t, lp, nil) @@ -1707,6 +1751,10 @@ func newLiveRunnerHTTP(t *testing.T, withManager bool) *lphttp { func newLiveRunnerHTTPOnchain(t *testing.T) *lphttp { t.Helper() + prevWatcher := core.PriceFeedWatcher + core.PriceFeedWatcher = stubPriceFeedWatcher{price: eth.PriceData{Price: big.NewRat(2000, 1)}} + t.Cleanup(func() { core.PriceFeedWatcher = prevWatcher }) + node, err := core.NewLivepeerNode(nil, t.TempDir(), nil) require.NoError(t, err) node.LivePaymentInterval = 5 * time.Second @@ -1930,9 +1978,8 @@ func registerLiveRunnerForSession(t *testing.T, lp *lphttp, opts *liveRunnerRegi App: "live-video-to-video/scope", Capacity: capacity, PriceInfo: runner.LiveRunnerPriceInfo{ - PricePerUnit: 10, - PixelsPerUnit: 1, - Unit: "WEI", + Price: json.Number("0.001990656"), + Unit: "720p", }, }, lp.orchestrator.RegistrationSecret()) require.NoError(t, err) diff --git a/server/cliserver_test.go b/server/cliserver_test.go index 57df172cd0..98c9fc18da 100644 --- a/server/cliserver_test.go +++ b/server/cliserver_test.go @@ -230,7 +230,7 @@ func TestRegisterLiveRunnersCLIEndpoint(t *testing.T) { srv := httptest.NewServer(s.cliWebServerHandlers("addr")) defer srv.Close() - body := fmt.Sprintf(`{"runners":[{"label":"runner-a","runner_url":"https://runner.example.com","app":"live-video-to-video/scope","capacity":1,"price_info":{"price_per_unit":10,"pixels_per_unit":1,"unit":"WEI"},"health_url":%q}]}`, healthSrv.URL) + body := fmt.Sprintf(`{"runners":[{"label":"runner-a","runner_url":"https://runner.example.com","app":"live-video-to-video/scope","capacity":1,"price_info":{"price":10},"health_url":%q}]}`, healthSrv.URL) res, err := http.Post(srv.URL+"/registerLiveRunners", "application/json", strings.NewReader(body)) require.NoError(t, err) defer res.Body.Close() @@ -255,7 +255,7 @@ func TestRegisterLiveRunnersCLIEndpointInvalidBatchDoesNotMutate(t *testing.T) { orch := newStubOrchestrator() manager := runner.NewLiveRunnerRegistry(runner.LiveRunnerRegistryConfig{Host: orch}) t.Cleanup(manager.Stop) - initialConfig := fmt.Sprintf(`{"runners":[{"label":"runner-a","runner_url":"https://runner.example.com","app":"live-video-to-video/scope","price_info":{"price_per_unit":10,"pixels_per_unit":1,"unit":"WEI"},"health_url":%q}]}`, healthSrv.URL) + initialConfig := fmt.Sprintf(`{"runners":[{"label":"runner-a","runner_url":"https://runner.example.com","app":"live-video-to-video/scope","price_info":{"price":10},"health_url":%q}]}`, healthSrv.URL) _, err = manager.RegisterStaticRunnersJSON([]byte(initialConfig)) require.NoError(t, err) n.LiveRunnerManager = manager @@ -263,7 +263,7 @@ func TestRegisterLiveRunnersCLIEndpointInvalidBatchDoesNotMutate(t *testing.T) { srv := httptest.NewServer(s.cliWebServerHandlers("addr")) defer srv.Close() - body := fmt.Sprintf(`{"runners":[{"label":"runner-b","runner_url":"https://runner-two.example.com","app":"live-video-to-video/scope","price_info":{"price_per_unit":10,"pixels_per_unit":1,"unit":"WEI"},"health_url":%q},{"runner_url":"https://runner-three.example.com","app":"live-video-to-video/scope","price_info":{"price_per_unit":10,"pixels_per_unit":1,"unit":"WEI"},"health_url":%q}]}`, healthSrv.URL, healthSrv.URL) + body := fmt.Sprintf(`{"runners":[{"label":"runner-b","runner_url":"https://runner-two.example.com","app":"live-video-to-video/scope","price_info":{"price":10},"health_url":%q},{"runner_url":"https://runner-three.example.com","app":"live-video-to-video/scope","price_info":{"price":10},"health_url":%q}]}`, healthSrv.URL, healthSrv.URL) res, err := http.Post(srv.URL+"/registerLiveRunners", "application/json", strings.NewReader(body)) require.NoError(t, err) defer res.Body.Close() diff --git a/server/remote_discovery.go b/server/remote_discovery.go index 546d73f74b..5787fdf638 100644 --- a/server/remote_discovery.go +++ b/server/remote_discovery.go @@ -3,6 +3,7 @@ package server import ( "context" "encoding/json" + "errors" "math/big" "net/url" "reflect" @@ -245,10 +246,12 @@ func mergeDiscoveryRunners(entry *remoteDiscoveryMergeEntry, discovery remoteDis // For each address, merge runners by URL. First wins; matching duplicates are // ignored and conflicting duplicates are logged before keeping the first. for _, r := range discovery.Runners { - if r.URL == "" { - continue - } - if !entry.remoteDiscoveryRunnerEligible(r) { + if err := entry.validateRunner(r); err != nil { + orchURL := "" + if entry != nil && entry.orch.URL != nil { + orchURL = entry.orch.URL.String() + } + clog.Info(context.Background(), "Filtered remote discovery runner", "orch", orchURL, "runner", r.URL, "app", r.App, "reason", err.Error()) continue } if !slices.Contains(entry.orch.Capabilities, r.App) { @@ -286,47 +289,51 @@ func remoteDiscoveryEntries(raw json.RawMessage) []remoteDiscoveryEntry { return entries } -func (entry *remoteDiscoveryMergeEntry) remoteDiscoveryRunnerEligible(runner runner.LiveRunnerDiscoveryRunner) bool { +func (entry *remoteDiscoveryMergeEntry) validateRunner(runner runner.LiveRunnerDiscoveryRunner) error { + if runner.URL == "" { + return errors.New("missing URL") + } if strings.TrimSpace(runner.App) == "" { - return false + return errors.New("missing app") + } + price, err := validateRunnerPrice(runner.PriceInfo) + if err != nil { + return err + } + maxPrice := BroadcastCfg.MaxPrice() + if maxPrice != nil { + if price.Cmp(maxPrice) > 0 { + return errors.New("above global max price") + } } - if runner.PriceInfo == nil { - return false + return nil +} + +func validateRunnerPrice(priceInfo *runner.LiveRunnerPriceInfo) (*big.Rat, error) { + price, ok := runnerPrice(priceInfo) + if !ok { + return nil, errors.New("invalid price") } - if runner.PriceInfo.PricePerUnit <= 0 || runner.PriceInfo.PixelsPerUnit <= 0 { - return false + currency := strings.ToLower(strings.TrimSpace(priceInfo.Currency)) + if currency != "wei" { + return nil, errors.New("unsupported currency") } - if !strings.EqualFold(strings.TrimSpace(runner.PriceInfo.Unit), "WEI") { - clog.Warningf(context.Background(), "Rejecting remote discovery runner with unsupported price unit orch=%s runner=%s app=%s price_per_unit=%d pixels_per_unit=%d unit=%s", - entry.orch.URL.String(), - runner.URL, - runner.App, - runner.PriceInfo.PricePerUnit, - runner.PriceInfo.PixelsPerUnit, - runner.PriceInfo.Unit, - ) - return false + unit := strings.ToLower(strings.TrimSpace(priceInfo.Unit)) + if unit != "seconds" && unit != "720p-pixel-seconds" { + return nil, errors.New("unsupported unit") } + return price, nil +} - maxPrice := BroadcastCfg.MaxPrice() - if maxPrice == nil { - return true +func runnerPrice(priceInfo *runner.LiveRunnerPriceInfo) (*big.Rat, bool) { + if priceInfo == nil { + return nil, false } - price := new(big.Rat).SetFrac64(runner.PriceInfo.PricePerUnit, runner.PriceInfo.PixelsPerUnit) - if price.Cmp(maxPrice) > 0 { - clog.Warningf(context.Background(), "Rejecting remote discovery runner above global max price orch=%s runner=%s app=%s price_per_unit=%d pixels_per_unit=%d unit=%s price=%s max_price=%s", - entry.orch.URL.String(), - runner.URL, - runner.App, - runner.PriceInfo.PricePerUnit, - runner.PriceInfo.PixelsPerUnit, - runner.PriceInfo.Unit, - price.RatString(), - maxPrice.RatString(), - ) - return false + price, ok := new(big.Rat).SetString(strings.TrimSpace(priceInfo.Price.String())) + if !ok || price.Sign() <= 0 { + return nil, false } - return true + return price, true } func forEachRemoteDiscoveryCapability(info *common.OrchNetworkCapabilities, f func(key string, capability core.Capability, modelID string)) { diff --git a/server/remote_signer.go b/server/remote_signer.go index fd926cf574..df8ccd1142 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "math" "math/big" "net/http" "net/url" @@ -32,6 +33,7 @@ const HTTPStatusRefreshSession = 480 const HTTPStatusPriceExceeded = 481 const HTTPStatusNoTickets = 482 const RefreshSessionOrchestratorURLHeader = "Livepeer-Orchestrator-URL" +const RemoteType_Live = "live" const RemoteType_LiveVideoToVideo = "lv2v" const PipelineLiveVideoToVideo = "live-video-to-video" const remoteSignerAuthIDHeader = "Signer-Auth-Id" @@ -218,6 +220,7 @@ type RemotePaymentState struct { Balance string InitialPricePerUnit int64 InitialPixelsPerUnit int64 + Type string SequenceNumber uint64 AuthID string } @@ -242,7 +245,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: `live`, `lv2v`. Optional. Type string `json:"type"` // Capabilities to include in the ticket. Optional; may be set for the lv2v job type. @@ -426,6 +429,12 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req respondJsonError(ctx, w, err, http.StatusBadRequest) return } + if state.Type != "" && state.Type != req.Type { + err := fmt.Errorf("job type mismatch") + respondJsonError(ctx, w, err, http.StatusBadRequest) + return + } + state.Type = req.Type state.SequenceNumber++ } else { state = &RemotePaymentState{ @@ -433,6 +442,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req OrchestratorAddress: orchAddr, InitialPricePerUnit: priceInfo.PricePerUnit, InitialPixelsPerUnit: priceInfo.PixelsPerUnit, + Type: req.Type, } } @@ -521,7 +531,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req return } - pixels := req.InPixels + pixels := int64(0) + billableUnits := int64(req.InPixels) now := time.Now() lastUpdate := state.LastUpdate if lastUpdate.IsZero() { @@ -536,13 +547,19 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req } pixelsPerSec := float64(info.Height) * float64(info.Width) * float64(info.FPS) pixels = int64(pixelsPerSec * billableSecs) // pixels to charge for + billableUnits = pixels + } else if req.Type == RemoteType_Live { + if billableSecs <= 0 { + billableSecs = (10 * time.Second).Seconds() + } + billableUnits = int64(math.Ceil(billableSecs)) // seconds to charge for } else if req.Type != "" { err = errors.New("invalid job type") respondJsonError(ctx, w, err, http.StatusBadRequest) return } - if pixels <= 0 { - err = errors.New("missing pixels or job type") + if billableUnits <= 0 { + err = errors.New("missing billable unit or job type") respondJsonError(ctx, w, err, http.StatusBadRequest) return } @@ -558,7 +575,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req } // Compute required fee using initial price - fee := calculateFee(pixels, initialPrice) + fee := calculateFee(billableUnits, initialPrice) // Create balance update balUpdate, err := newBalanceUpdate(sess, fee) @@ -627,7 +644,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req return } state.Balance = newBal.RatString() - state.LastUpdate = time.Now() + state.LastUpdate = now state.PMSessionID = sess.PMSessionID state.SenderNonce, err = sender.Nonce(sess.PMSessionID) if err != nil { @@ -683,6 +700,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req pipeline := "" if req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo + } else if req.Type == RemoteType_Live { + pipeline = RemoteType_Live } // NB: This could could drop events if tha Kafka queue is full! monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{ @@ -702,7 +721,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req "pixels": pixels, "session_balance": newBal.FloatString(0), "computed_fee": fee.FloatString(0), - "cost_per_pixel": orchPrice.FloatString(10), + "cost": orchPrice.FloatString(10), "sequence_number": state.SequenceNumber, "num_tickets": balUpdate.NumTickets, "auth_id": state.AuthID, diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index f1031a9171..cf0238d637 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -238,7 +238,7 @@ func TestGenerateLivePayment_RequestValidationErrors(t *testing.T) { return r }(), wantStatus: http.StatusBadRequest, - wantMsg: "missing pixels or job type", + wantMsg: "missing billable unit or job type", }, { name: "num tickets exceeds limit", @@ -375,6 +375,7 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) { stateBytes []byte stateSig []byte orchInfo *net.OrchestratorInfo + reqType string omitManifestID bool wantStatus int wantMsg string @@ -408,6 +409,23 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) { wantStatus: http.StatusBadRequest, wantMsg: "invalid state", }, + { + name: "job type mismatch", + stateBytes: func() []byte { + state, err := json.Marshal(RemotePaymentState{ + StateID: "state", + OrchestratorAddress: ethcommon.BytesToAddress(orchInfo.Address), + InitialPricePerUnit: 1, + InitialPixelsPerUnit: 1, + Type: RemoteType_LiveVideoToVideo, + }) + require.NoError(err) + return state + }(), + reqType: RemoteType_Live, + wantStatus: http.StatusBadRequest, + wantMsg: "job type mismatch", + }, { name: "orchestrator address mismatch", stateBytes: func() []byte { @@ -479,6 +497,7 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) { Orchestrator: orchBlob, ManifestID: manifestID, InPixels: 1, + Type: tt.reqType, State: RemotePaymentStateSig{State: tt.stateBytes, Sig: stateSig}, }) require.NoError(err) @@ -681,6 +700,92 @@ func TestGenerateLivePayment_LV2V_Succeeds(t *testing.T) { } +func TestGenerateLivePayment_LiveBillsElapsedSeconds(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + require := require.New(t) + + ethClient := newTestEthClient(t) + node, _ := core.NewLivepeerNode(ethClient, "", nil) + node.Balances = core.NewAddressBalances(1 * time.Minute) + defer node.Balances.StopCleanup() + node.Sender = newMockSender(mockSenderConfig{ev: big.NewRat(35, 1)}) + 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(reqPayload RemotePaymentRequest) (RemotePaymentResponse, net.Payment) { + reqBody, err := json.Marshal(reqPayload) + 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, rr.Body.String()) + + var resp RemotePaymentResponse + require.NoError(json.NewDecoder(rr.Body).Decode(&resp)) + require.NotEmpty(resp.Payment) + + paymentBytes, err := base64.StdEncoding.DecodeString(resp.Payment) + require.NoError(err) + var payment net.Payment + require.NoError(proto.Unmarshal(paymentBytes, &payment)) + + return resp, payment + } + + parseBalance := func(stateBytes []byte) (RemotePaymentState, *big.Rat) { + var state RemotePaymentState + require.NoError(json.Unmarshal(stateBytes, &state)) + bal := new(big.Rat) + _, ok := bal.SetString(state.Balance) + require.True(ok, "failed to parse balance: %q", state.Balance) + return state, bal + } + + firstUpdate := time.Now() + resp, payment := doPayment(RemotePaymentRequest{ + Orchestrator: orchBlob, + Type: RemoteType_Live, + }) + require.Len(payment.TicketSenderParams, 1) + + state, bal := parseBalance(resp.State.State) + require.Equal(RemoteType_Live, state.Type) + require.EqualValues(0, state.SequenceNumber) + require.Equal(firstUpdate, state.LastUpdate) + // First live payment bills the 10-second minimum: 10s * 3 wei/s = 30 wei. + require.Zero(bal.Cmp(big.NewRat(5, 1)), "unexpected initial balance: %s", bal.RatString()) + + time.Sleep(12 * time.Second) + secondUpdate := time.Now() + resp2, payment2 := doPayment(RemotePaymentRequest{ + Orchestrator: orchBlob, + ManifestID: "live-manifest", + Type: RemoteType_Live, + State: resp.State, + }) + require.Len(payment2.TicketSenderParams, 1) + + state2, bal2 := parseBalance(resp2.State.State) + require.Equal(RemoteType_Live, state2.Type) + require.EqualValues(1, state2.SequenceNumber) + require.Equal(secondUpdate, state2.LastUpdate) + // Follow-up live payment bills elapsed seconds: 12s * 3 wei/s = 36 wei. + require.Zero(bal2.Cmp(big.NewRat(4, 1)), "unexpected follow-up balance: %s", bal2.RatString()) + }) +} + func TestGenerateLivePayment_WebhookCallback(t *testing.T) { require := require.New(t) @@ -1107,29 +1212,29 @@ func TestRemoteSigner_Discovery(t *testing.T) { "capabilities": ["evil/app"], "unknown_field": "must-not-leak", "runners": [ - {"url":"https://orch1.example.com:8935/scope-ok","app":"live-video-to-video/model-a","price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}}, + {"url":"https://orch1.example.com:8935/scope-ok","app":"live-video-to-video/model-a","price_info":{"price":80,"currency":"wei","unit":"seconds"}}, {"url":"https://orch1.example.com:8935/scope-missing-price","app":"live-video-to-video/model-a"}, - {"url":"https://orch1.example.com:8935/scope-too-expensive","app":"live-video-to-video/model-a","price_info":{"price_per_unit":120,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://orch1.example.com:8935/scope-usd","app":"live-video-to-video/model-a","price_info":{"price_per_unit":1,"pixels_per_unit":1,"unit":"USD"}}, - {"url":"https://orch1.example.com:8935/scope-invalid-price","app":"live-video-to-video/model-a","price_info":{"price_per_unit":1,"pixels_per_unit":0,"unit":"WEI"}}, - {"url":"https://orch1.example.com:8935/scope-missing-unit","app":"live-video-to-video/model-a","price_info":{"price_per_unit":1,"pixels_per_unit":1}}, - {"url":"https://orch1.example.com:8935/txt","app":"text-to-image/model-b","price_info":{"price_per_unit":1,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://orch1.example.com:8935/scope-too-expensive","app":"live-video-to-video/model-a","price_info":{"price":120,"currency":"wei","unit":"seconds"}}, + {"url":"https://orch1.example.com:8935/scope-unsupported-unit","app":"live-video-to-video/model-a","price_info":{"price":1,"currency":"usd","unit":"seconds"}}, + {"url":"https://orch1.example.com:8935/scope-invalid-price","app":"live-video-to-video/model-a","price_info":{"price":0,"currency":"wei","unit":"seconds"}}, + {"url":"https://orch1.example.com:8935/scope-invalid-pixels","app":"live-video-to-video/model-a","price_info":{"price":0,"currency":"wei","unit":"seconds"}}, + {"url":"https://orch1.example.com:8935/txt","app":"text-to-image/model-b","price_info":{"price":1,"currency":"wei","unit":"seconds"}} ] }, { "address": "https://discovered.example.com:8935", "runners": [ - {"url":"https://discovered.example.com:8935/scope","app":"live-video-to-video/model-a","capacity":3,"price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":1,"price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}}, + {"url":"https://discovered.example.com:8935/scope","app":"live-video-to-video/model-a","capacity":3,"price_info":{"price":80,"currency":"wei","unit":"seconds"}}, + {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":1,"price_info":{"price":80,"currency":"wei","unit":"seconds"}}, {"url":"https://discovered.example.com:8935/missing-price","app":"live-video-to-video/model-a"}, - {"url":"https://discovered.example.com:8935/too-expensive","app":"live-video-to-video/model-a","price_info":{"price_per_unit":120,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://discovered.example.com:8935/too-expensive","app":"live-video-to-video/model-a","price_info":{"price":120,"currency":"wei","unit":"seconds"}} ] }, { "address": "https://discovered.example.com:8935/", "runners": [ - {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":1,"capacity_used":1,"capacity_available":0,"price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":2,"price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":1,"capacity_used":1,"capacity_available":0,"price_info":{"price":80,"currency":"wei","unit":"seconds"}}, + {"url":"https://discovered.example.com:8935/dupe","app":"live-video-to-video/model-a","capacity":2,"price_info":{"price":80,"currency":"wei","unit":"seconds"}} ] } ]`), @@ -1159,14 +1264,14 @@ func TestRemoteSigner_Discovery(t *testing.T) { { "address": "https://orch2.example.com:8935", "runners": [ - {"url":"https://orch2.example.com:8935/txt","app":"text-to-image/model-b","price_info":{"price_per_unit":999,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://orch2.example.com:8935/scope","app":"live-video-to-video/model-a","price_info":{"price_per_unit":1,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://orch2.example.com:8935/txt","app":"text-to-image/model-b","price_info":{"price":999,"currency":"wei","unit":"seconds"}}, + {"url":"https://orch2.example.com:8935/scope","app":"live-video-to-video/model-a","price_info":{"price":1,"currency":"wei","unit":"seconds"}} ] }, { "address": "https://discovered.example.com:8935", "runners": [ - {"url":"https://discovered.example.com:8935/from-orch2","app":"live-video-to-video/model-a","price_info":{"price_per_unit":80,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://discovered.example.com:8935/from-orch2","app":"live-video-to-video/model-a","price_info":{"price":80,"currency":"wei","unit":"seconds"}} ] } ]`), @@ -1188,7 +1293,7 @@ func TestRemoteSigner_Discovery(t *testing.T) { Discovery: discoveryRaw(t, `[{ "address": "https://orch3.example.com:8935", "runners": [ - {"url":"https://orch3.example.com:8935/scope","app":"live-video-to-video/model-a","price_info":{"price_per_unit":1,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://orch3.example.com:8935/scope","app":"live-video-to-video/model-a","price_info":{"price":1,"currency":"wei","unit":"seconds"}} ] }]`), }, @@ -1321,11 +1426,6 @@ func TestRemoteSigner_Discovery(t *testing.T) { func TestRemoteSigner_Discovery_EmptyCacheRetriesBeforeInterval(t *testing.T) { require := require.New(t) - capability := core.Capability_LiveVideoToVideo - modelID := "scope" - BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(200, 995328000000))) - defer BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, nil) - // Node starts with no network capabilities (pool poll has not populated the cache yet). node := &core.LivepeerNode{} @@ -1349,7 +1449,7 @@ func TestRemoteSigner_Discovery_EmptyCacheRetriesBeforeInterval(t *testing.T) { Discovery: discoveryRaw(t, `[{ "address": "https://late.example.com:8935", "runners": [ - {"url":"https://late.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} + {"url":"https://late.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price":5,"currency":"wei","unit":"seconds"}} ] }]`), }, @@ -1370,11 +1470,6 @@ func TestRemoteSigner_Discovery_EmptyCacheRetriesBeforeInterval(t *testing.T) { func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t *testing.T) { require := require.New(t) - capability := core.Capability_LiveVideoToVideo - modelID := "scope" - BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(200, 995328000000))) - defer BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, nil) - node := &core.LivepeerNode{} require.NoError(node.UpdateNetworkCapabilities([]*common.OrchNetworkCapabilities{ { @@ -1382,7 +1477,7 @@ func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t Discovery: discoveryRaw(t, `[{ "address": "https://runner-derived.example.com:8935", "runners": [ - {"url":"https://runner-derived.example.com:8935/apps/runner/session","app":"custom-runner-v1","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} + {"url":"https://runner-derived.example.com:8935/apps/runner/session","app":"custom-runner-v1","capacity":1,"price_info":{"price":5,"currency":"wei","unit":"seconds"}} ] }]`), }, @@ -1410,11 +1505,6 @@ func TestRemoteSigner_Discovery_UsesRunnerDiscoveryWhenRPCCapabilitiesMissing(t func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsValidRunnersWhenRPCCapabilitiesMissing(t *testing.T) { require := require.New(t) - capability := core.Capability_LiveVideoToVideo - modelID := "scope" - BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(200, 995328000000))) - defer BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, nil) - node := &core.LivepeerNode{} require.NoError(node.UpdateNetworkCapabilities([]*common.OrchNetworkCapabilities{ { @@ -1422,12 +1512,12 @@ func TestRemoteSigner_Discovery_RunnerDiscoveryKeepsValidRunnersWhenRPCCapabilit Discovery: discoveryRaw(t, `[{ "address": "https://mixed-runner-derived.example.com:8935", "runners": [ - {"url":"https://mixed-runner-derived.example.com:8935/too-expensive","app":"live-video-to-video/scope","price_info":{"price_per_unit":201,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://mixed-runner-derived.example.com:8935/usd","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"USD"}}, + {"url":"https://mixed-runner-derived.example.com:8935/too-expensive","app":"live-video-to-video/scope","price_info":{"price":201,"currency":"wei","unit":"seconds"}}, + {"url":"https://mixed-runner-derived.example.com:8935/unsupported-unit","app":"live-video-to-video/scope","price_info":{"price":5,"currency":"usd","unit":"seconds"}}, {"url":"https://mixed-runner-derived.example.com:8935/missing-price","app":"live-video-to-video/scope"}, - {"url":"https://mixed-runner-derived.example.com:8935/invalid-pixels","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":0,"unit":"WEI"}}, - {"url":"https://mixed-runner-derived.example.com:8935/empty-app","app":" ","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://mixed-runner-derived.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} + {"url":"https://mixed-runner-derived.example.com:8935/invalid-pixels","app":"live-video-to-video/scope","price_info":{"price":0,"currency":"wei","unit":"seconds"}}, + {"url":"https://mixed-runner-derived.example.com:8935/empty-app","app":" ","price_info":{"price":5,"currency":"wei","unit":"seconds"}}, + {"url":"https://mixed-runner-derived.example.com:8935/apps/runner/session","app":"live-video-to-video/scope","capacity":1,"price_info":{"price":5,"currency":"wei","unit":"seconds"}} ] }]`), }, @@ -1460,11 +1550,11 @@ func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { capability := core.Capability_LiveVideoToVideo modelID := "scope" - BroadcastCfg.SetMaxPrice(core.NewFixedPrice(big.NewRat(200, 995328000000))) + BroadcastCfg.SetMaxPrice(core.NewFixedPrice(big.NewRat(200, 1))) defer BroadcastCfg.SetMaxPrice(nil) // Capability/model max prices do not apply to runner discovery. This would // reject price-ok under the old runner capability lookup path. - BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(1, 995328000000))) + BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, core.NewFixedPrice(big.NewRat(1, 1))) defer BroadcastCfg.SetCapabilityMaxPrice(capability, modelID, nil) node := &core.LivepeerNode{} @@ -1474,13 +1564,15 @@ func TestRemoteSigner_Discovery_FiltersRunnerDiscoveryPricing(t *testing.T) { Discovery: discoveryRaw(t, `[{ "address": "https://filtered.example.com:8935", "runners": [ - {"url":"https://filtered.example.com:8935/too-expensive","app":"live-video-to-video/scope","price_info":{"price_per_unit":201,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://filtered.example.com:8935/price-ok","app":"live-video-to-video/scope","price_info":{"price_per_unit":199,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://filtered.example.com:8935/usd","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"USD"}}, + {"url":"https://filtered.example.com:8935/too-expensive","app":"live-video-to-video/scope","price_info":{"price":201,"currency":"wei","unit":"720p-pixel-seconds"}}, + {"url":"https://filtered.example.com:8935/price-ok","app":"live-video-to-video/scope","price_info":{"price":199,"currency":"wei","unit":"720p-pixel-seconds"}}, + {"url":"https://filtered.example.com:8935/unsupported-unit","app":"live-video-to-video/scope","price_info":{"price":5,"currency":"usd","unit":"720p-pixel-seconds"}}, {"url":"https://filtered.example.com:8935/missing-price","app":"live-video-to-video/scope"}, - {"url":"https://filtered.example.com:8935/non-positive-price","app":"live-video-to-video/scope","price_info":{"price_per_unit":0,"pixels_per_unit":995328000000,"unit":"WEI"}}, - {"url":"https://filtered.example.com:8935/non-positive-pixels","app":"live-video-to-video/scope","price_info":{"price_per_unit":5,"pixels_per_unit":0,"unit":"WEI"}}, - {"url":"https://filtered.example.com:8935/empty-app","app":"","price_info":{"price_per_unit":5,"pixels_per_unit":995328000000,"unit":"WEI"}} + {"url":"https://filtered.example.com:8935/missing-currency","app":"live-video-to-video/scope","price_info":{"price":5,"unit":"720p-pixel-seconds"}}, + {"url":"https://filtered.example.com:8935/missing-unit","app":"live-video-to-video/scope","price_info":{"price":5,"currency":"wei"}}, + {"url":"https://filtered.example.com:8935/non-positive-price","app":"live-video-to-video/scope","price_info":{"price":0,"currency":"wei","unit":"720p-pixel-seconds"}}, + {"url":"https://filtered.example.com:8935/non-positive-pixels","app":"live-video-to-video/scope","price_info":{"price":0,"currency":"wei","unit":"720p-pixel-seconds"}}, + {"url":"https://filtered.example.com:8935/empty-app","app":"","price_info":{"price":5,"currency":"wei","unit":"720p-pixel-seconds"}} ] }]`), }, @@ -1548,9 +1640,9 @@ func TestRemoteDiscoveryRunnerDuplicateComparison(t *testing.T) { CapacityUsed: 1, CapacityAvailable: 1, PriceInfo: &runner.LiveRunnerPriceInfo{ - PricePerUnit: 7, - PixelsPerUnit: 1, - Unit: "WEI", + Price: json.Number("7"), + Currency: "wei", + Unit: "seconds", }, } @@ -1600,8 +1692,8 @@ func TestRemoteSigner_Discovery_RunnersDoNotRequireTopLevelAdvertisedPrice(t *te Discovery: discoveryRaw(t, `[{ "address": "https://priced.example.com:8935", "runners": [ - {"url":"https://priced.example.com:8935/priced","app":"live-video-to-video/model-priced","price_info":{"price_per_unit":7,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://priced.example.com:8935/unpriced","app":"live-video-to-video/model-unpriced","price_info":{"price_per_unit":7,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://priced.example.com:8935/priced","app":"live-video-to-video/model-priced","price_info":{"price":7,"currency":"wei","unit":"seconds"}}, + {"url":"https://priced.example.com:8935/unpriced","app":"live-video-to-video/model-unpriced","price_info":{"price":7,"currency":"wei","unit":"seconds"}} ] }]`), }, @@ -1611,8 +1703,8 @@ func TestRemoteSigner_Discovery_RunnersDoNotRequireTopLevelAdvertisedPrice(t *te Discovery: discoveryRaw(t, `[{ "address": "https://unpriced.example.com:8935", "runners": [ - {"url":"https://unpriced.example.com:8935/priced","app":"live-video-to-video/model-priced","price_info":{"price_per_unit":7,"pixels_per_unit":1,"unit":"WEI"}}, - {"url":"https://unpriced.example.com:8935/unpriced","app":"live-video-to-video/model-unpriced","price_info":{"price_per_unit":7,"pixels_per_unit":1,"unit":"WEI"}} + {"url":"https://unpriced.example.com:8935/priced","app":"live-video-to-video/model-priced","price_info":{"price":7,"currency":"wei","unit":"seconds"}}, + {"url":"https://unpriced.example.com:8935/unpriced","app":"live-video-to-video/model-unpriced","price_info":{"price":7,"currency":"wei","unit":"seconds"}} ] }]`), },