Extract standalone Scope runtime#3989
Conversation
github-actions
Bot
added
go
📝 WalkthroughWalkthroughAdds a serverless websocket worker for live video transformation, startup configuration and model validation, a ChangesServerless AI execution
Logging helper formatting
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant StartScope
participant Orchestrator
participant TrickleServer
participant ServerlessWorker
Client->>StartScope: POST /scope
StartScope->>Orchestrator: validate or process payment
StartScope->>TrickleServer: create control and events channels
StartScope->>ServerlessWorker: start LiveVideoToVideo
ServerlessWorker->>Orchestrator: perform websocket handshake
TrickleServer-->>StartScope: deliver event segments
StartScope-->>Client: return manifest and channel URLs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ai/worker/serverless_worker.go`:
- Around line 544-546: Update the websocket logging around the message-receipt
log and the handshake log to stop emitting complete user or protocol payloads.
Log only the message type, request ID, and payload size, explicitly redact
identifiers and URLs, and remove fields containing raw handshake parameters,
user IDs, or message contents.
- Around line 291-299: Update LiveVideoToVideo’s locked reservation block to
check whether f.inUse >= f.capacity before incrementing, and return the
caller-expected capacity error when full. Keep the trickleSrv nil validation and
ensure the capacity check and f.inUse increment occur atomically under f.mu.
- Around line 241-270: Update the wsURL override validation in the serverless
handshake flow and its downstream dial at the symbol handling params.ws_url:
parse the candidate and configured LIVE_AI_WS_PREFIX, compare scheme and exact
canonical host/port rather than using strings.HasPrefix, and enforce a path
boundary for the configured prefix. Preserve the original override URL casing
after validation, and add regression tests covering allowed-host.evil and
allowed-host@evil bypass attempts.
- Around line 487-507: The websocket reader goroutine’s direct send to
messageChan can block after the outer loop exits. Update the message delivery in
the goroutine around websocketConn.ReadMessage to use a cancellation-aware
handoff tied to the worker’s cleanup or done signal, while preserving
termination on read errors; alternatively buffer messageChan sufficiently to
prevent the blocked send.
- Around line 587-615: Update the handshake retry flow around performHandshake
to stop and join the existing pinger before retrying, ensuring the stop
operation waits for its goroutine to exit. Quiesce the ping loop before
performHandshake, then restart it afterward using startPinging, preserving the
existing missedPongs reset and error handling.
- Around line 336-357: Bound the readiness and handshake reads in the connection
setup flow by setting a read deadline before the first websocketConn.ReadMessage
and retaining it through performHandshake. After performHandshake succeeds,
clear the deadline so normal session reads are not constrained; preserve the
existing cleanup and error handling for failed reads.
In `@cmd/livepeer/starter/starter.go`:
- Around line 1393-1414: The AI worker startup flow must treat AIServerless as
an independent worker mode. Update the surrounding AIWorker activation and
startup logic so aiServerless enables the worker without requiring AIWorker, and
branch into the serverless path before local runner cleanup, GPU discovery,
model/image setup, or other local-worker initialization. Preserve the existing
NewServerlessWorker validation and construction while keeping NewWorker behavior
unchanged for non-serverless mode.
In `@server/ai_http.go`:
- Around line 191-194: Reject requests in the server/ai_http.go range 191-194
when segData.ManifestID differs from segData.AuthToken.SessionId, returning 403
before any processing or balance mutation; update server/ai_http_test.go range
241-270 to assert the legacy mismatch receives 403 Forbidden and verify neither
processing nor balance state changes.
- Around line 459-460: Make the live-video-to-video admission check atomic by
changing the capacity flow around CheckAICapacity and
ServerlessWorker.LiveVideoToVideo: reserve a worker slot while validating
capacity, or add a bounded acquire under the same synchronization used to
increment inUse, and release it on completion or failure. Ensure concurrent
Scope requests cannot all pass HasCapacity and exceed the configured limit.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3e9afe66-153b-4832-b796-7ed046a312a2
📒 Files selected for processing (9)
ai/worker/serverless_worker.goai/worker/serverless_worker_test.goclog/clog.gocmd/livepeer/starter/flags.gocmd/livepeer/starter/starter.gocmd/livepeer/starter/starter_test.goserver/ai_http.goserver/ai_http_test.goserver/rpc_test.go
| wsURL := f.wsURL | ||
| wsURLPrefix := strings.ToLower(strings.TrimSpace(os.Getenv("LIVE_AI_WS_PREFIX"))) | ||
| if req.Params != nil && wsURLPrefix != "" { | ||
| if wsURLRaw, ok := (*req.Params)["ws_url"]; ok { | ||
| wsURLOverride, ok := wsURLRaw.(string) | ||
| if !ok { | ||
| return nil, &ServerlessHandshakeError{ | ||
| StatusCode: http.StatusBadRequest, | ||
| Code: "INVALID_WS", | ||
| Message: "invalid params.ws_url: must be a string", | ||
| } | ||
| } | ||
| if wsURLOverride != "" { | ||
| wsURLOverride = strings.ToLower(wsURLOverride) | ||
| if !strings.HasPrefix(wsURLOverride, wsURLPrefix) { | ||
| return nil, &ServerlessHandshakeError{ | ||
| StatusCode: http.StatusBadRequest, | ||
| Code: "INVALID_WS", | ||
| Message: "unacceptable params.ws_url", | ||
| } | ||
| } | ||
| parsedWSURL, err := url.Parse(wsURLOverride) | ||
| if err != nil || parsedWSURL.Host == "" { | ||
| return nil, &ServerlessHandshakeError{ | ||
| StatusCode: http.StatusBadRequest, | ||
| Code: "INVALID_WS", | ||
| Message: "invalid params.ws_url", | ||
| } | ||
| } | ||
| wsURL = wsURLOverride |
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | ⚡ Quick win
Parse and compare the WebSocket authority; string prefixes expose FAL_API_KEY.
Line 255 accepts overrides such as ws://allowed-host.evil/... or ws://allowed-host@evil/.... Line 313 then dials the attacker-controlled host with the authorization header. Compare parsed scheme and exact canonical host/port, enforce any path boundary, and preserve the original URL’s case. Add both bypasses as regression tests.
Also applies to: 303-313
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 241 - 270, Update the wsURL
override validation in the serverless handshake flow and its downstream dial at
the symbol handling params.ws_url: parse the candidate and configured
LIVE_AI_WS_PREFIX, compare scheme and exact canonical host/port rather than
using strings.HasPrefix, and enforce a path boundary for the configured prefix.
Preserve the original override URL casing after validation, and add regression
tests covering allowed-host.evil and allowed-host@evil bypass attempts.
| f.mu.Lock() | ||
| trickleSrv := f.trickleSrv | ||
| if trickleSrv == nil { | ||
| f.mu.Unlock() | ||
| return nil, fmt.Errorf("serverless worker requires trickle server before live-video-to-video") | ||
| } | ||
| f.inUse++ | ||
| currentInUse := f.inUse | ||
| f.mu.Unlock() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Reserve capacity atomically inside LiveVideoToVideo.
CheckAICapacity and this unconditional increment are separate operations, so concurrent /scope requests can all pass the check and exceed configured capacity. Check f.inUse >= f.capacity under this same lock before incrementing, returning the capacity error contract expected by the caller.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 291 - 299, Update
LiveVideoToVideo’s locked reservation block to check whether f.inUse >=
f.capacity before incrementing, and return the caller-expected capacity error
when full. Keep the trickleSrv nil validation and ensure the capacity check and
f.inUse increment occur atomically under f.mu.
| // Wait for connection readiness before starting background processing or returning. | ||
| _, readyMsg, err := websocketConn.ReadMessage() | ||
| if err != nil { | ||
| _ = websocketConn.Close() | ||
| f.mu.Lock() | ||
| f.inUse-- | ||
| remainingInUse := f.inUse | ||
| f.mu.Unlock() | ||
| clog.Error(ctx, "Failed to read ready message", "error", err, "inUse", remainingInUse, "capacity", f.capacity) | ||
| return nil, fmt.Errorf("failed to read ready message: %w", err) | ||
| } | ||
|
|
||
| err = performHandshake( | ||
| ctx, | ||
| readyMsg, | ||
| func() ([]byte, error) { | ||
| _, msg, readErr := websocketConn.ReadMessage() | ||
| return msg, readErr | ||
| }, | ||
| websocketConn, | ||
| reqJSON, | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the relevant regions around the comment locations
sed -n '300,390p' ai/worker/serverless_worker.go
printf '\n==== later region ====\\n'
sed -n '800,850p' ai/worker/serverless_worker.go
printf '\n==== search for performHandshake and websocket deadlines ====\\n'
rg -n "performHandshake|SetReadDeadline|ReadDeadline|SetDeadline|deadline|timeout" ai/worker/serverless_worker.go ai/worker -g '!**/*_test.go'Repository: livepeer/go-livepeer
Length of output: 10849
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the handshake implementation and any surrounding timeout logic in the repo.
rg -n "func performHandshake|performHandshake\\(|SetReadDeadline\\(|ReadMessage\\(|session timeout|ReadTimeout|WriteTimeout|context.WithTimeout|deadline" .Repository: livepeer/go-livepeer
Length of output: 20402
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Narrow inspection for the handshake path and any timeout setup around it.
file=ai/worker/serverless_worker.go
printf '=== around first handshake ===\n'
sed -n '320,370p' "$file"
printf '\n=== around second handshake / later path ===\n'
sed -n '815,840p' "$file"
printf '\n=== direct timeout/deadline calls in this file ===\n'
rg -n "SetReadDeadline|SetDeadline|context.WithTimeout|WithDeadline|ReadMessage\\(" "$file"Repository: livepeer/go-livepeer
Length of output: 2938
Bound the handshake reads. The session timeout only starts after ready/started, so a peer that upgrades successfully but never sends either message can hold the request and capacity slot indefinitely. Set a read deadline before the first ReadMessage, keep it through performHandshake, and clear it once the handshake succeeds.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 336 - 357, Bound the readiness
and handshake reads in the connection setup flow by setting a read deadline
before the first websocketConn.ReadMessage and retaining it through
performHandshake. After performHandshake succeeds, clear the deadline so normal
session reads are not constrained; preserve the existing cleanup and error
handling for failed reads.
| // Create a channel to receive messages | ||
| messageChan := make(chan struct { | ||
| messageType int | ||
| message []byte | ||
| err error | ||
| }) | ||
|
|
||
| // Start goroutine to read websocket messages | ||
| go func() { | ||
| for { | ||
| messageType, message, err := websocketConn.ReadMessage() | ||
| messageChan <- struct { | ||
| messageType int | ||
| message []byte | ||
| err error | ||
| }{messageType, message, err} | ||
| if err != nil { | ||
| return | ||
| } | ||
| } | ||
| }() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- file outline ---\n'
ast-grep outline ai/worker/serverless_worker.go --view expanded || true
printf '\n--- relevant occurrences ---\n'
rg -n "messageChan|websocketConn|ReadMessage|session-done|done channel|cancel" ai/worker/serverless_worker.go || true
printf '\n--- nearby lines 430-560 ---\n'
sed -n '430,560p' ai/worker/serverless_worker.goRepository: livepeer/go-livepeer
Length of output: 249
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- relevant occurrences ---'
rg -n "messageChan|websocketConn|ReadMessage|session-done|done channel|cancel" ai/worker/serverless_worker.go || true
echo
echo '--- nearby lines 430-560 ---'
sed -n '430,560p' ai/worker/serverless_worker.goRepository: livepeer/go-livepeer
Length of output: 5236
Make reader delivery cancellation-aware. The goroutine can still block forever on messageChan <- ... after the outer loop returns (for timeout or any other exit), even though ReadMessage() was unblocked by closing the socket. Add a done/cancel path or buffer the handoff so cleanup can terminate the reader cleanly.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 487 - 507, The websocket reader
goroutine’s direct send to messageChan can block after the outer loop exits.
Update the message delivery in the goroutine around websocketConn.ReadMessage to
use a cancellation-aware handoff tied to the worker’s cleanup or done signal,
while preserving termination on read errors; alternatively buffer messageChan
sufficiently to prevent the blocked send.
| clog.Info(ctx, "Received message from websocket", | ||
| "type", msg.messageType, | ||
| "message", string(msg.message)) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not log complete user and protocol payloads.
The handshake request includes arbitrary parameters and daydream_user_id; WebSocket messages may contain further user data. Log only message type, request ID, and payload size, with explicit redaction for identifiers and URLs.
Also applies to: 822-822
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 544 - 546, Update the websocket
logging around the message-receipt log and the handshake log to stop emitting
complete user or protocol payloads. Log only the message type, request ID, and
payload size, explicitly redact identifiers and URLs, and remove fields
containing raw handshake parameters, user IDs, or message contents.
| // Perform the handshake | ||
| err := performHandshake( | ||
| ctx, | ||
| msg.message, | ||
| func() ([]byte, error) { | ||
| nextMsg := <-messageChan | ||
| if nextMsg.err != nil { | ||
| return nil, nextMsg.err | ||
| } | ||
| return nextMsg.message, nil | ||
| }, | ||
| websocketConn, | ||
| reqJSON, | ||
| ) | ||
| if err != nil { | ||
| clog.Error(ctx, "Handshake retry failed", "error", err) | ||
| return | ||
| } | ||
| if stopPings != nil { | ||
| stopPings() | ||
| } | ||
| stopPings = startPinging( | ||
| ctx, | ||
| pingInterval, | ||
| maxMissedPongs, | ||
| &missedPongs, | ||
| websocketConn, | ||
| ) | ||
| missedPongs.Store(0) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the file structure first.
ast-grep outline ai/worker/serverless_worker.go --view expanded > /tmp/serverless_outline.txt
sed -n '1,220p' /tmp/serverless_outline.txt
echo
echo '--- relevant source slices ---'
nl -ba ai/worker/serverless_worker.go | sed -n '520,700p'
echo '---'
nl -ba ai/worker/serverless_worker.go | sed -n '820,940p'
echo
echo '--- symbol search ---'
rg -n "startPinging|performHandshake|stopPings|missedPongs|pong|ready|started" ai/worker/serverless_worker.goRepository: livepeer/go-livepeer
Length of output: 4696
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
path = Path("ai/worker/serverless_worker.go")
lines = path.read_text().splitlines()
for start, end in [(520, 700), (790, 930)]:
print(f"\n--- {path} lines {start}-{end} ---")
for i in range(start, min(end, len(lines)) + 1):
print(f"{i:4d}: {lines[i-1]}")
PY
echo
echo '--- symbol search ---'
python3 - <<'PY'
from pathlib import Path
import re
text = Path("ai/worker/serverless_worker.go").read_text().splitlines()
for idx, line in enumerate(text, 1):
if re.search(r'performHandshake|startPinging|stopPings|missedPongs|pong|ready|started', line):
print(f"{idx:4d}: {line}")
PYRepository: livepeer/go-livepeer
Length of output: 14829
Stop and join the pinger before retrying the handshake.
The retry path can still read a pong between ready and started, and performHandshake treats that as an unexpected message. Quiesce the ping loop before re-handshaking, and make the stop path wait for the goroutine to exit before restarting it.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker.go` around lines 587 - 615, Update the handshake
retry flow around performHandshake to stop and join the existing pinger before
retrying, ensuring the stop operation waits for its goroutine to exit. Quiesce
the ping loop before performHandshake, then restart it afterward using
startPinging, preserving the existing missedPongs reset and error handling.
| if *cfg.AIServerless { | ||
| if len(aiModelConfigs) != 1 { | ||
| glog.Errorf("Serverless requires exactly one AI model config, got %d", len(aiModelConfigs)) | ||
| return | ||
| } | ||
| config := aiModelConfigs[0] | ||
| if config.Pipeline != "live-video-to-video" || config.ModelID != "scope" { | ||
| glog.Errorf("Serverless only supports live-video-to-video/scope, got %s/%s", config.Pipeline, config.ModelID) | ||
| return | ||
| } | ||
|
|
||
| n.AIWorker, err = worker.NewServerlessWorker(strings.TrimSpace(config.URL), config.Capacity) | ||
| if err != nil { | ||
| glog.Errorf("Error starting Serverless AI worker: %v", err) | ||
| return | ||
| } | ||
| } else { | ||
| n.AIWorker, err = worker.NewWorker(imageOverrides, *cfg.AIVerboseLogs, gpus, modelsDir, containerCreatorID) | ||
| if err != nil { | ||
| glog.Errorf("Error starting AI worker: %v", err) | ||
| return | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Treat serverless as a complete worker mode, not only an alternate constructor.
This branch is reached only under if *cfg.AIWorker, so -aiServerless alone never starts a worker. When both flags are supplied, startup still removes existing runner containers and performs local GPU/model/image setup before reaching this branch. Normalize serverless mode to activate AIWorker, then branch before all local-worker cleanup and initialization.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@cmd/livepeer/starter/starter.go` around lines 1393 - 1414, The AI worker
startup flow must treat AIServerless as an independent worker mode. Update the
surrounding AIWorker activation and startup logic so aiServerless enables the
worker without requiring AIWorker, and branch into the serverless path before
local runner cleanup, GPU discovery, model/image setup, or other local-worker
initialization. Preserve the existing NewServerlessWorker validation and
construction while keeping NewWorker behavior unchanged for non-serverless mode.
| if string(segData.ManifestID) != segData.AuthToken.SessionId { | ||
| clog.Info(ctx, "Legacy Scope payment manifest mismatch", "manifest_id", segData.ManifestID, "auth_session_id", segData.AuthToken.SessionId) | ||
| } | ||
| manifestID := string(segData.ManifestID) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Bind paid Scope retries to the auth token’s session. The production path accepts cross-session token reuse, and the new test codifies it.
server/ai_http.go#L191-L194: reject requests wheresegData.ManifestIDdiffers fromsegData.AuthToken.SessionId.server/ai_http_test.go#L241-L270: replace the successful legacy-mismatch case with a403 Forbiddenassertion and verify no processing or balance mutation occurs.
📍 Affects 2 files
server/ai_http.go#L191-L194(this comment)server/ai_http_test.go#L241-L270
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@server/ai_http.go` around lines 191 - 194, Reject requests in the
server/ai_http.go range 191-194 when segData.ManifestID differs from
segData.AuthToken.SessionId, returning 403 before any processing or balance
mutation; update server/ai_http_test.go range 241-270 to assert the legacy
mismatch receives 403 Forbidden and verify neither processing nor balance state
changes.
| // Check if there is capacity for the request | ||
| hasCapacity, _ := orch.CheckAICapacity(pipeline, modelID) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "CheckAICapacity|inUse|tryAcquire|LiveVideoToVideo|capacity" server . --glob '!**/vendor/**'Repository: livepeer/go-livepeer
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- server/ai_http.go matches ---\n'
rg -n "CheckAICapacity|LiveVideoToVideo|tryAcquire|inUse|capacity" server/ai_http.go server . --glob '!**/vendor/**' || true
printf '\n--- file list candidates ---\n'
fd -a ".*(worker|capacity|orch).*" server . --hidden --exclude vendor || trueRepository: livepeer/go-livepeer
Length of output: 249
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '430,540p' server/ai_http.goRepository: livepeer/go-livepeer
Length of output: 3926
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- server/ai_http.go:420-470 ---'
sed -n '420,470p' server/ai_http.go
echo
echo '--- core/ai_orchestrator.go:402-438 ---'
sed -n '402,438p' core/ai_orchestrator.go
echo
echo '--- ai/worker/serverless_worker.go:1-70 ---'
sed -n '1,70p' ai/worker/serverless_worker.go
echo
echo '--- ai/worker/serverless_worker.go:207-420 ---'
sed -n '207,420p' ai/worker/serverless_worker.go
echo
echo '--- ai/worker/serverless_worker.go:740-795 ---'
sed -n '740,795p' ai/worker/serverless_worker.goRepository: livepeer/go-livepeer
Length of output: 13175
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- core/ai_orchestrator.go:428-438 and 600-620 ---'
sed -n '428,438p' core/ai_orchestrator.go
printf '\n'
sed -n '600,620p' core/ai_orchestrator.go
printf '\n--- server/ai_http.go:788-946 ---\n'
sed -n '788,946p' server/ai_http.go
printf '\n--- search for CheckAICapacity callers in server/ai_http.go ---\n'
rg -n "CheckAICapacity\\(" server/ai_http.goRepository: livepeer/go-livepeer
Length of output: 8328
Make live-video-to-video admission atomic. The local AIWorker path only checks HasCapacity() here, while ServerlessWorker.LiveVideoToVideo increments inUse later under a separate lock. Concurrent Scope requests can all pass this gate and start over capacity; reserve capacity in CheckAICapacity() or add a bounded acquire/release in ServerlessWorker.LiveVideoToVideo.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@server/ai_http.go` around lines 459 - 460, Make the live-video-to-video
admission check atomic by changing the capacity flow around CheckAICapacity and
ServerlessWorker.LiveVideoToVideo: reserve a worker slot while validating
capacity, or add a bounded acquire under the same synchronization used to
increment inUse, and release it on completion or failure. Ensure concurrent
Scope requests cannot all pass HasCapacity and exceed the configured limit.
0ce3ea3 to
074e937
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@server/ai_http_test.go`:
- Around line 241-269: Update
TestScopePaidRetryAllowsSegmentManifestToDifferFromAuthToken and the
corresponding processScopePayment validation to reject requests when
segData.ManifestID differs from segData.AuthToken.SessionId. Change the test to
assert the request is rejected and verify no payment is credited under the
arbitrary manifest ID or auth-token session ID.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3cf81266-0968-436d-8944-d9e99839c7c8
📒 Files selected for processing (9)
ai/worker/serverless_worker.goai/worker/serverless_worker_test.goclog/clog.gocmd/livepeer/starter/flags.gocmd/livepeer/starter/starter.gocmd/livepeer/starter/starter_test.goserver/ai_http.goserver/ai_http_test.goserver/rpc_test.go
🚧 Files skipped from review as they are similar to previous changes (8)
- cmd/livepeer/starter/flags.go
- clog/clog.go
- cmd/livepeer/starter/starter.go
- server/rpc_test.go
- cmd/livepeer/starter/starter_test.go
- server/ai_http.go
- ai/worker/serverless_worker_test.go
- ai/worker/serverless_worker.go
| func TestScopePaidRetryAllowsSegmentManifestToDifferFromAuthToken(t *testing.T) { | ||
| lp := newScopeHTTP(t) | ||
| orch := lp.orchestrator.(*stubOrchestrator) | ||
| orch.balances = make(map[ethcommon.Address]map[core.ManifestID]*big.Rat) | ||
| orch.paymentCredit = big.NewRat(100, 1) | ||
| _, oInfo := requestScopePaymentChallenge(t, lp) | ||
| legacyManifestID := "different-manifest" | ||
| headers := scopePaymentHeaders(t, orch, oInfo.GetAuthToken(), legacyManifestID) | ||
|
|
||
| w := httptest.NewRecorder() | ||
| req := httptest.NewRequest(http.MethodPost, "/scope", strings.NewReader(`{"model_id":"scope"}`)) | ||
| setRequestHeaders(req, headers) | ||
| lp.ServeHTTP(w, req) | ||
|
|
||
| require.Equal(t, http.StatusOK, w.Code) | ||
| var resp worker.LiveVideoToVideoResponse | ||
| require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) | ||
| require.NotNil(t, resp.ManifestId) | ||
| require.Equal(t, legacyManifestID, *resp.ManifestId) | ||
| require.NotNil(t, resp.ControlUrl) | ||
| require.Contains(t, *resp.ControlUrl, legacyManifestID+"-control") | ||
| require.NotNil(t, resp.EventsUrl) | ||
| require.Contains(t, *resp.EventsUrl, legacyManifestID+"-events") | ||
| closeScopeEvents(t, lp, legacyManifestID) | ||
|
|
||
| balance := orch.Balance(orch.Address(), core.ManifestID(legacyManifestID)) | ||
| require.NotNil(t, balance) | ||
| require.Equal(t, "100", balance.FloatString(0)) | ||
| require.Nil(t, orch.Balance(orch.Address(), core.ManifestID(oInfo.GetAuthToken().GetSessionId()))) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Reject manifest IDs that differ from the payment challenge.
This test codifies an unsafe bypass: a payer with a valid challenge token can select an arbitrary manifest ID, which processScopePayment then uses for accounting, response identity, and trickle channel names. Require segData.ManifestID == segData.AuthToken.SessionId and test for rejection instead, preventing session namespace collisions.
Proposed test direction
-func TestScopePaidRetryAllowsSegmentManifestToDifferFromAuthToken(t *testing.T) {
+func TestScopePaidRetryRejectsSegmentManifestThatDiffersFromAuthToken(t *testing.T) {
...
- require.Equal(t, http.StatusOK, w.Code)
+ require.Equal(t, http.StatusForbidden, w.Code)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@server/ai_http_test.go` around lines 241 - 269, Update
TestScopePaidRetryAllowsSegmentManifestToDifferFromAuthToken and the
corresponding processScopePayment validation to reject requests when
segData.ManifestID differs from segData.AuthToken.SessionId. Change the test to
assert the request is rejected and verify no payment is credited under the
arbitrary manifest ID or auth-token session ID.
074e937 to
d36450e
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@trickle/trickle_server.go`:
- Around line 345-351: Update writeHookError so unexpected errors are logged
server-side and the HTTP response uses a fixed generic message instead of
err.Error(). Preserve the existing RequestError handling and status behavior,
while keeping unexpected failures mapped to http.StatusInternalServerError.
- Around line 332-338: Invoke sm.config.BeforeCreate in handlePost before the
getOrCreateStream call when the POST may autocreate a stream, applying the same
r and streamName arguments and writeHookError/early-return behavior used by the
existing hook block. Ensure autocreated streams cannot bypass the create-time
policy while preserving normal publishing flow.
In `@trickle/trickle_test.go`:
- Around line 575-617: Update the asynchronous reader and writer goroutines
around unblocked and writeDone to return typed result values that include any
io.ReadAll, write, or close errors instead of discarding them. Replace the
unbounded receives with select statements using a test timeout, and assert the
returned errors and data so reset or writer failures produce actionable test
failures without hanging.
- Around line 638-640: Update the test flow around sub.Read() to check err
before accessing resp or deferring resp.Body.Close(). Return or fail the test
with the read error when present, and only close the response body after
confirming resp is valid.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 15163ad0-ef4c-4892-a78a-648b6270c2ec
📒 Files selected for processing (15)
ai/worker/serverless_worker.goai/worker/serverless_worker_test.goclog/clog.gocmd/livepeer/starter/flags.gocmd/livepeer/starter/starter.gocmd/livepeer/starter/starter_test.goserver/ai_http.goserver/ai_http_test.goserver/rpc_test.gotrickle/README.mdtrickle/local_publisher.gotrickle/segment_buffer.gotrickle/segment_buffer_test.gotrickle/trickle_server.gotrickle/trickle_test.go
🚧 Files skipped from review as they are similar to previous changes (8)
- cmd/livepeer/starter/starter_test.go
- clog/clog.go
- cmd/livepeer/starter/starter.go
- server/rpc_test.go
- ai/worker/serverless_worker_test.go
- server/ai_http.go
- server/ai_http_test.go
- ai/worker/serverless_worker.go
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@trickle/trickle_server.go`:
- Around line 345-351: Update writeHookError so unexpected errors are logged
server-side and the HTTP response uses a fixed generic message instead of
err.Error(). Preserve the existing RequestError handling and status behavior,
while keeping unexpected failures mapped to http.StatusInternalServerError.
- Around line 332-338: Invoke sm.config.BeforeCreate in handlePost before the
getOrCreateStream call when the POST may autocreate a stream, applying the same
r and streamName arguments and writeHookError/early-return behavior used by the
existing hook block. Ensure autocreated streams cannot bypass the create-time
policy while preserving normal publishing flow.
In `@trickle/trickle_test.go`:
- Around line 575-617: Update the asynchronous reader and writer goroutines
around unblocked and writeDone to return typed result values that include any
io.ReadAll, write, or close errors instead of discarding them. Replace the
unbounded receives with select statements using a test timeout, and assert the
returned errors and data so reset or writer failures produce actionable test
failures without hanging.
- Around line 638-640: Update the test flow around sub.Read() to check err
before accessing resp or deferring resp.Body.Close(). Return or fail the test
with the read error when present, and only close the response body after
confirming resp is valid.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 15163ad0-ef4c-4892-a78a-648b6270c2ec
📒 Files selected for processing (15)
ai/worker/serverless_worker.goai/worker/serverless_worker_test.goclog/clog.gocmd/livepeer/starter/flags.gocmd/livepeer/starter/starter.gocmd/livepeer/starter/starter_test.goserver/ai_http.goserver/ai_http_test.goserver/rpc_test.gotrickle/README.mdtrickle/local_publisher.gotrickle/segment_buffer.gotrickle/segment_buffer_test.gotrickle/trickle_server.gotrickle/trickle_test.go
🚧 Files skipped from review as they are similar to previous changes (8)
- cmd/livepeer/starter/starter_test.go
- clog/clog.go
- cmd/livepeer/starter/starter.go
- server/rpc_test.go
- ai/worker/serverless_worker_test.go
- server/ai_http.go
- server/ai_http_test.go
- ai/worker/serverless_worker.go
🛑 Comments failed to post (4)
trickle/trickle_server.go (2)
332-338: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash ast-grep outline trickle/trickle_server.go --items all --match 'handlePost|BeforeCreate|getOrCreateStream' --view expanded rg -n -C8 'func \(sm \*Server\) handlePost|BeforeCreate|getOrCreateStream' trickle/trickle_server.goRepository: livepeer/go-livepeer
Length of output: 3035
🏁 Script executed:
#!/bin/bash sed -n '345,430p' trickle/trickle_server.go printf '\n---\n' rg -n -C4 'handleCreate|handlePost|Autocreate|BeforeCreate|PathValue\("idx"\)' trickle/trickle_server.goRepository: livepeer/go-livepeer
Length of output: 5784
Call
BeforeCreatefromhandlePost
POST /stream/{idx}can autocreate streams viagetOrCreateStream(..., false), but it skipssm.config.BeforeCreate. That lets publishers bypass create-time policy by creating streams directly through publish requests.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_server.go` around lines 332 - 338, Invoke sm.config.BeforeCreate in handlePost before the getOrCreateStream call when the POST may autocreate a stream, applying the same r and streamName arguments and writeHookError/early-return behavior used by the existing hook block. Ensure autocreated streams cannot bypass the create-time policy while preserving normal publishing flow.
345-351: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not expose generic hook errors in HTTP responses.
Line 351 returns
err.Error()for unexpected failures, potentially leaking database, payment, or infrastructure details. Log the error server-side and return a fixed message.Proposed fix
func writeHookError(w http.ResponseWriter, err error) { var requestErr *RequestError if errors.As(err, &requestErr) { http.Error(w, requestErr.Error(), requestErr.httpStatus()) return } - http.Error(w, err.Error(), http.StatusInternalServerError) + slog.Error("Trickle lifecycle hook failed", "err", err) + http.Error(w, "Internal server error", http.StatusInternalServerError) }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.func writeHookError(w http.ResponseWriter, err error) { var requestErr *RequestError if errors.As(err, &requestErr) { http.Error(w, requestErr.Error(), requestErr.httpStatus()) return } slog.Error("Trickle lifecycle hook failed", "err", err) http.Error(w, "Internal server error", http.StatusInternalServerError) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_server.go` around lines 345 - 351, Update writeHookError so unexpected errors are logged server-side and the HTTP response uses a fixed generic message instead of err.Error(). Preserve the existing RequestError handling and status behavior, while keeping unexpected failures mapped to http.StatusInternalServerError.trickle/trickle_test.go (2)
575-617: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Bound asynchronous waits and propagate goroutine errors.
<-unblockedand<-writeDonecan hang the suite indefinitely, while both goroutines discard operational errors. Return results through typed channels and useselectwith a timeout so reset failures produce actionable test failures.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_test.go` around lines 575 - 617, Update the asynchronous reader and writer goroutines around unblocked and writeDone to return typed result values that include any io.ReadAll, write, or close errors instead of discarding them. Replace the unbounded receives with select statements using a test timeout, and assert the returned errors and data so reset or writer failures produce actionable test failures without hanging.
638-640: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Check the read error before dereferencing
resp.If
sub.Read()fails, Line 639 can panic on a nil response and hide the actual failure.Proposed fix
resp, err := sub.Read() + require.NoError(err) defer resp.Body.Close()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.resp, err := sub.Read() require.NoError(err) defer resp.Body.Close()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_test.go` around lines 638 - 640, Update the test flow around sub.Read() to check err before accessing resp or deferring resp.Body.Close(). Return or fail the test with the read error when present, and only close the response body after confirming resp is valid.
d36450e to
48cb31c
Compare
48cb31c to
fed20c7
Compare
| return nil, fmt.Errorf("ServerlessWorker.ImageToText not implemented") | ||
| } | ||
|
|
||
| func (f *ServerlessWorker) TextToSpeech(ctx context.Context, req GenTextToSpeechJSONRequestBody) (*AudioResponse, error) { |
There was a problem hiding this comment.
@j0sh when we have everybody switched over maybe we can deprecate these batch pipelines.
There was a problem hiding this comment.
@j0sh The code appears well isolated from the other workloads. All tests pass, and Go-Livepeer builds successfully. I’m also able to send transcoding and live-video-to-video requests (i.e. via BOX) through it.
Overall, the code looks reasonable. I did not perform a full end-to-end test, so for a comprehensive E2E review, please ask someone on your team. Approved from my side.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ai/worker/serverless_worker_test.go`:
- Around line 368-413: The TestStartPingingRestartsWithFullIntervalGrace test
relies on wall-clock sleeps near timer boundaries and can flake under load.
Replace the time.Sleep-based assertions and polling with a controllable
clock/ticker or synctest, while preserving verification that no ping occurs
before each full interval and exactly the first ping occurs after the restarted
interval.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c88da3a1-7af7-409c-800f-0ef70cc4af9e
📒 Files selected for processing (8)
ai/worker/serverless_worker.goai/worker/serverless_worker_test.goclog/clog.gocmd/livepeer/starter/flags.gocmd/livepeer/starter/starter.goserver/ai_http.goserver/ai_http_test.goserver/rpc_test.go
🚧 Files skipped from review as they are similar to previous changes (7)
- cmd/livepeer/starter/flags.go
- cmd/livepeer/starter/starter.go
- server/rpc_test.go
- clog/clog.go
- server/ai_http_test.go
- ai/worker/serverless_worker.go
- server/ai_http.go
| func TestStartPingingRestartsWithFullIntervalGrace(t *testing.T) { | ||
| ws, _, cleanup := newTestLockedWebSocket(t) | ||
| defer cleanup() | ||
|
|
||
| const interval = 40 * time.Millisecond | ||
| var missedPongs atomic.Int32 | ||
|
|
||
| stopPings := startPinging( | ||
| context.Background(), | ||
| interval, | ||
| 100, // keep this test focused on timer behavior | ||
| &missedPongs, | ||
| ws, | ||
| ) | ||
|
|
||
| time.Sleep(interval - 10*time.Millisecond) | ||
| if got := missedPongs.Load(); got != 0 { | ||
| t.Fatalf("expected no ping before first interval, got=%d", got) | ||
| } | ||
|
|
||
| // Simulate restart: stop the existing pinger before its next tick. | ||
| stopPings() | ||
|
|
||
| // Simulate successful re-handshake: start a fresh pinger. | ||
| stopPings = startPinging( | ||
| context.Background(), | ||
| interval, | ||
| 100, | ||
| &missedPongs, | ||
| ws, | ||
| ) | ||
| defer stopPings() | ||
|
|
||
| time.Sleep(interval - 10*time.Millisecond) | ||
| if got := missedPongs.Load(); got != 0 { | ||
| t.Fatalf("expected full interval grace after restart, got=%d", got) | ||
| } | ||
|
|
||
| deadline := time.Now().Add(250 * time.Millisecond) | ||
| for time.Now().Before(deadline) { | ||
| if missedPongs.Load() == 1 { | ||
| return | ||
| } | ||
| time.Sleep(5 * time.Millisecond) | ||
| } | ||
| t.Fatalf("expected first ping after full interval, got=%d", missedPongs.Load()) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Make the pinger timing test deterministic.
Lines 383 and 401 assume a 30ms sleep returns before a 40ms timer. time.Sleep may overshoot under CI load, causing intermittent failures. Use a controllable clock/ticker or synctest instead of wall-clock boundary assertions.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ai/worker/serverless_worker_test.go` around lines 368 - 413, The
TestStartPingingRestartsWithFullIntervalGrace test relies on wall-clock sleeps
near timer boundaries and can flake under load. Replace the time.Sleep-based
assertions and polling with a controllable clock/ticker or synctest, while
preserving verification that no ping occurs before each full interval and
exactly the first ping occurs after the restarted interval.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3989 +/- ##
============================================
Coverage ? 33.71058%
============================================
Files ? 173
Lines ? 43120
Branches ? 0
============================================
Hits ? 14536
Misses ? 27473
Partials ? 1111
Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
2 participants
Summary
/scope, including request size limiting, public control/events trickle URLs, serverless handshake error mapping, and serverless worker trickle wiring.ai/worker.ServerlessWorkerwith websocket handshake/restart/ping/timeout/URL validation/capacity tests.-aiServerlessstarter support for exactly onelive-video-to-video/scopeAI model config.Excluded
/refresh-payment.livepricing changes.Tests
go test ./ai/worker -count=1go test ./clog -count=1git diff --checkThe requested target command is blocked locally by the existing ffmpeg install, not by this branch:
env GOCACHE=/private/tmp/go-livepeer-scope-gocache CGO_CFLAGS=-Wno-error=implicit-function-declaration go test ./server ./ai/worker ./cmd/livepeer/starter -count=1./ai/workerpasses in that command, but./serverand./cmd/livepeer/starterfail to link because/Users/josh/compiled/lib/libavfilter.ais missing_avfilter_compare_sign_bybuffand_avfilter_compare_sign_bypath. I reproduced the same link failure on pristinemasterin/private/tmp/go-livepeer-master-check.Extraction Check
ai/worker/serverless_worker.gomatchesja/live-runner.ja/live-runneris intentionally non-empty only where this branch excludes Live Runner APIs/discovery,/refresh-payment, remote signer live pricing, trickle edits, and other non-Scope branch changes.Summary by CodeRabbit
New Features
/scopeendpoint for live video processing, payment challenges, and segment-based billing.-aiServerlessoption to enable serverless AI processing.Bug Fixes
Tests