BYOC #3727

ad-astra-video · 2025-09-02T13:25:34Z

What does this pull request do? Explain your changes. (required)

Adds configurable streaming for BYOC entrypoint to go-livepeer. Uses trickle protocol to handle streaming for similar entrypoints and outputs from go-livepeer as live-video-to-video.

Streams can be any or a mix of the following:

video ingress via WHIP (with Gateway) or RTMP (with MediaMTX)
video egress via RTMP (with MediaMTX) or WHEP (with mediamtx)
SSE data output

Control and Events channels are created for every stream.

Streams are created with a POST request to /ai/stream/start that will start the stream and reserve the capacity with an Orchestrator that is providing the BYOC capability. If video ingress is enabled, the client should then start a stream with WHIP or RTMP to the provided ingress URLs provided in the response. URLs for egress video, data, updates (control) and events are also included in the response as well as the stream_id. The stream_id is an integral part of the URLs provided to interact with the stream and is combined with a provided stream name in the /ai/stream/start request.

Streams are stopped with a POST request to /ai/stream/stop. Orchestrators and Gateways track payment balance and the Gateway adjusts to the Orchestrators provided balance in new JobTokens provided at each payment interval every minute. Orchestrators will shutdown a stream when payment balance is zero.

Specific updates (required)

Add job_stream.go and job_stream_test.go
refactor job_rpc.go to reuse stream setup where made sense
updates go routines to ignore to enable tests to pass in common/testutil.go.

How did you test each of these updates (required)

Used byoc-stream to test end to end: https://github.com/ad-astra-video/livepeer-app-pipelines/tree/main/byoc-stream
Added tests to job_stream_test.go and some additional tests to job_rpc_test.go.

Does this pull request close any open issues?

Checklist:

Read the contribution guide
make runs successfully
All tests in ./test.sh pass
README and other documentation updated
Pending changelog updated

codecov · 2025-09-02T13:40:27Z

Codecov Report

❌ Patch coverage is 52.68631% with 819 lines in your changes missing coverage. Please review.
✅ Project coverage is 33.37411%. Comparing base (d095d96) to head (1c97122).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
server/job_stream.go	61.05169%	369 Missing and 68 partials ⚠️
server/job_rpc.go	52.55255%	128 Missing and 30 partials ⚠️
server/ai_live_video.go	31.25000%	82 Missing and 6 partials ⚠️
core/external_capabilities.go	3.48837%	83 Missing ⚠️
core/livepeernode.go	0.00000%	29 Missing ⚠️
common/testutil.go	0.00000%	13 Missing ⚠️
core/ai_orchestrator.go	0.00000%	4 Missing ⚠️
server/ai_mediaserver.go	69.23077%	3 Missing and 1 partial ⚠️
server/rpc.go	0.00000%	3 Missing ⚠️

Additional details and impacted files

@@                 Coverage Diff                 @@
##              master       #3727         +/-   ##
===================================================
+ Coverage   31.94301%   33.37411%   +1.43110%     
===================================================
  Files            158         159          +1     
  Lines          47519       49050       +1531     
===================================================
+ Hits           15179       16370       +1191     
- Misses         31437       31633        +196     
- Partials         903        1047        +144

Files with missing lines	Coverage Δ
server/ai_process.go	`1.65975% <ø> (ø)`
server/rpc.go	`68.66029% <0.00000%> (+0.46752%)`	⬆️
core/ai_orchestrator.go	`29.24820% <0.00000%> (-0.12098%)`	⬇️
server/ai_mediaserver.go	`9.63542% <69.23077%> (+2.79331%)`	⬆️
common/testutil.go	`14.08451% <0.00000%> (-2.58216%)`	⬇️
core/livepeernode.go	`51.90476% <0.00000%> (-8.31623%)`	⬇️
core/external_capabilities.go	`28.45528% <3.48837%> (-58.38683%)`	⬇️
server/ai_live_video.go	`31.61209% <31.25000%> (+31.61209%)`	⬆️
server/job_rpc.go	`43.79277% <52.55255%> (+13.42325%)`	⬆️
server/job_stream.go	`61.05169% <61.05169%> (ø)`

... and 4 files with indirect coverage changes

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 19ded51...1c97122. Read the comment docs.

Files with missing lines	Coverage Δ
server/ai_process.go	`1.65975% <ø> (ø)`
server/rpc.go	`68.66029% <0.00000%> (+0.46752%)`	⬆️
core/ai_orchestrator.go	`29.24820% <0.00000%> (-0.12098%)`	⬇️
server/ai_mediaserver.go	`9.63542% <69.23077%> (+2.79331%)`	⬆️
common/testutil.go	`14.08451% <0.00000%> (-2.58216%)`	⬇️
core/livepeernode.go	`51.90476% <0.00000%> (-8.31623%)`	⬇️
core/external_capabilities.go	`28.45528% <3.48837%> (-58.38683%)`	⬇️
server/ai_live_video.go	`31.61209% <31.25000%> (+31.61209%)`	⬆️
server/job_rpc.go	`43.79277% <52.55255%> (+13.42325%)`	⬆️
server/job_stream.go	`61.05169% <61.05169%> (ø)`

... and 4 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

core/external_capabilities.go

server/job_rpc.go

server/job_stream.go

core/external_capabilities.go

server/job_stream.go

pschroedl · 2025-09-09T18:24:18Z

server/job_stream.go

+		}
+		stopJob.sign() //no changes to make, sign job
+
+		token, err := sessionToToken(params.liveParams.sess)


lets check/log this err here

Still hoping we can check and log this err

Added an error check and log in d73fe37. Note that this function cannot return an error currently, should I just update to only return the JobToken. I think I added an error here thinking as I built it out something could cause an error possibly.

…able live payments

pschroedl · 2025-09-11T19:18:21Z

server/job_stream.go

+		}
+		stopJob.sign() //no changes to make, sign job
+
+		token, err := sessionToToken(params.liveParams.sess)


Still hoping we can check and log this err

pschroedl · 2025-09-11T19:38:54Z

server/job_stream.go

+
+			// fetch new JobToken with each payment
+			// update the session for the LivePipeline with new token
+			newToken, err := getToken(ctx, 3*time.Second, token.ServiceAddr, stream.Pipeline, jobSender.Addr, jobSender.Sig)


Can we make this 3 second param a constant, say TokenRefreshTimeout = 3 * time.second?

Also, it would be good to have a few retries here

Updated in bfb0135. Added 3 retries in 18db960 for the getToken request with a reasonable backoff with a check to stay within the timeout provided.

pschroedl · 2025-09-11T19:39:35Z

server/job_stream.go

+		clog.Infof(ctx, "Starting stream processing")
+		//refresh the token if not first Orch to confirm capacity and new ticket params
+		if firstProcessed {
+			newToken, err := getToken(ctx, 3*time.Second, orch.ServiceAddr, gatewayJob.Job.Req.Capability, gatewayJob.Job.Req.Sender, gatewayJob.Job.Req.Sig)


Same comment RE: constant for timeout

Updated in bfb0135

pschroedl · 2025-09-11T19:45:28Z

server/job_stream.go

+
+	// Setup request body to be able to preserve for retries
+	// Read the entire body first with 10MB limit
+	bodyBytes, err := io.ReadAll(io.LimitReader(r.Body, 10<<20))


Since io.ReadALL truncates to 10MB here, maybe we want to log that?

Suggested change

bodyBytes, err := io.ReadAll(io.LimitReader(r.Body, 10<<20))

if len(bodyBytes) > 10<<20 {

fmt.Errorf("request body too large, maximum size is %d bytes", 10<<20)

}

Updated how the r.Body is limited to 10MB so that can get a specific error in 67cb9cd. Looked like from docs that limit reader would just stop reading and provide no error.

pschroedl · 2025-09-11T19:47:52Z

server/job_rpc.go

+	clog.V(common.DEBUG).Infof(ctx, "job price=%v units=%v", jobPrice.PricePerUnit, jobPrice.PixelsPerUnit)
+
+	//no payment included, confirm if balance remains
+	jobPriceRat := big.NewRat(jobPrice.PricePerUnit, jobPrice.PixelsPerUnit)


I think we should extract this payment logic here into another method(s), a bit hard to reason about this with so all the nested conditions

Plus, Adding test(s) around that new method would be fantastic!

pschroedl · 2025-09-11T19:56:23Z

server/job_stream.go

+
+				jobPriceRat := big.NewRat(orchJob.JobPrice.PricePerUnit, orchJob.JobPrice.PixelsPerUnit)
+				if jobPriceRat.Cmp(big.NewRat(0, 1)) > 0 {
+					h.orchestrator.DebitFees(orchJob.Sender, core.ManifestID(orchJob.Req.Capability), orchJob.JobPrice, int64(pmtCheckDur.Seconds()))


This concerns me a bit as it seems like a couple things could happen here since these two operations ( debit and check ) are not one atomic operation
Between the debit and balance check, other goroutines can create race conditions by:

Crediting the account (incoming payments)

Debit the account (other streams)

Change the balance state(maybe somehow?)

I think we can pull these into another method debit and check and using a mutex lock to prevent this

pschroedl · 2025-09-11T19:58:06Z

server/job_stream.go

+	stream, err := h.node.ExternalCapabilities.AddStream(orchJob.Req.ID, orchJob.Req.Capability, reqBodyBytes)
+	if err != nil {
+		clog.Errorf(ctx, "Error adding stream to external capabilities: %v", err)
+		respondWithError(w, "Error adding stream to external capabilities", http.StatusInternalServerError)


should we be calling RemoveStream here in case of error?

pschroedl · 2025-09-11T20:02:26Z

core/external_capabilities.go

+		return false
+	}
+
+	if sd.controlChannel == nil {


I think we should/need to acquire the mutex sd.sdm.lock() here before reading or there is a possiblility of reading nil, old, or partial data

Fixed in d73fe37

pschroedl · 2025-09-11T20:17:52Z

core/external_capabilities.go

+		<-ctx.Done()
+
+		//orchestrator channels shutdown
+		if stream.pubChannel != nil {


Lets add some error handling here just in case we call close() twice ( even though adding locks to the other channel operations should prevent this )

added error if stmt to catch error and log it in d73fe37

pschroedl · 2025-09-11T21:12:28Z

server/job_stream.go

+			delete(ls.LivepeerNode.LivePipelines, streamId)
+			return
+		case <-pmtTicker.C:
+			if !params.inputStreamExists() {


nit: suggest breaking this case out into a new method

pschroedl · 2025-09-11T21:21:05Z

core/ai_orchestrator.go

+	//ensure price numerator and denominator can be int64
+	jobPrice, err = common.PriceToInt64(jobPrice)
+	if err != nil {
+		return nil, err


return nil, fmt.Errorf("invalid job price: %w", err)

Updated in d73fe37

* Add SDXL and SDXL FaceID streamdiffusion images Co-authored-by: victorgelias <[email protected]> * Refactor: Clarify docker image descriptions Co-authored-by: victorgelias <[email protected]> --------- Co-authored-by: Cursor Agent <[email protected]>

…gateway->sendPaymentForStream

…5 segment history

ad-astra-video requested a review from pschroedl September 2, 2025 13:25

github-actions bot added go Pull requests that update Go code AI Issues and PR related to the AI-video branch. labels Sep 2, 2025

pschroedl reviewed Sep 3, 2025

View reviewed changes

core/external_capabilities.go Outdated Show resolved Hide resolved

server/job_rpc.go Show resolved Hide resolved

server/job_stream.go Outdated Show resolved Hide resolved

core/external_capabilities.go Outdated Show resolved Hide resolved

pschroedl reviewed Sep 3, 2025

View reviewed changes

server/job_stream.go Outdated Show resolved Hide resolved

ad-astra-video force-pushed the av/add-byoc-streaming branch from fdc7fc4 to 34e53b9 Compare September 5, 2025 17:50

pschroedl reviewed Sep 9, 2025

View reviewed changes

ad-astra-video force-pushed the av/add-byoc-streaming branch from e862a20 to 33fda51 Compare September 9, 2025 20:30

ad-astra-video and others added 21 commits September 9, 2025 15:39

add data channel

a139b2f

update to new trickle subscriber api

b2507bf

update ai-runner bindings

8b7c61b

remove Orch data url short circuit

00fd90c

move datastore items to separte file similar to status

3a6d24a

update to remove datastore and use ringbuffer reader

a1323ee

ai/live: Read JSONL from a time delimited data channel.

642814d

copy dataWriter in newParams

5a07065

update data channel mimetype

35ff4ac

update code gen for data_url added to LiveVideoToVideoResponse

aa2f0b8

update to use live video response DataUrl field

3e166c5

update to make data channel optional

699f778

add byoc streaming

bc624ee

refactor to move job setup to separate function

9ac928d

fix

ba8525c

refactor to reuse job request send to orch

33c05ae

fix

e52582c

capabilities: add live-ai external capabilities to Capabilities to en…

8a7ec3f

…able live payments

capabilities: fix

fa105de

updates to add streaming to byoc

5e8676b

continue build out of gateway and orchestrator buildout

b78405d

ad-astra-video added 8 commits September 9, 2025 15:42

fix rtmp streaming and storing orch url

9aacc0d

add error logging to retrying stream

12f8895

gateway job_stream refactor

8246825

remove job_trickle.go

358cea2

slim down StreamInfo in external capabilities used by Orchestrator

ec00c5e

get new token for stop request

85e5c25

some cleanup

9182ef8

fix merge update

55666ef

ad-astra-video force-pushed the av/add-byoc-streaming branch from 769b8e6 to 55666ef Compare September 9, 2025 20:47

pschroedl marked this pull request as ready for review September 11, 2025 18:48

pschroedl reviewed Sep 11, 2025

View reviewed changes

victorges and others added 9 commits September 19, 2025 18:01

error handling, logging, lock stream info when checking active

f79d2fe

create get new token timeout

30af123

add retries to getToken

d648a59

add logging end response for too large body

da54cbf

move payment processing to separate functions: orch->processPayment, …

dcc98b1

…gateway->sendPaymentForStream

fix error check, reduce segment writer history, fix stop stream

5881ea9

close data segment writer after right and put segment writer back to …

6fb5308

…5 segment history

tests tests tests

1ca67c4

ad-astra-video force-pushed the av/add-byoc-streaming branch from ddf2612 to 1ca67c4 Compare September 19, 2025 23:03

ad-astra-video and others added 5 commits September 19, 2025 18:07

Merge branch 'master' into av/add-byoc-streaming

1c97122

make stream payment processing more lenient

a6c8d88

add startTime to liveParams for First Segment Delay calcs

b1771b7

Merge branch 'master' into av/add-byoc-streaming

3d01cc3

fix spacing

9c94845

eliteprox mentioned this pull request Oct 3, 2025

server(api): change default server path livepeer/pytrickle#47

Draft

-	bodyBytes, err := io.ReadAll(io.LimitReader(r.Body, 10<<20))
+if len(bodyBytes) > 10<<20 {
+	fmt.Errorf("request body too large, maximum size is %d bytes", 10<<20)
+}

BYOC #3727

Are you sure you want to change the base?

BYOC #3727

Conversation

ad-astra-video commented Sep 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented Sep 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ad-astra-video Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ad-astra-video Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ad-astra-video Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pschroedl Sep 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pschroedl Sep 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ad-astra-video Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ad-astra-video commented Sep 2, 2025 •

edited

Loading

codecov bot commented Sep 2, 2025 •

edited

Loading

ad-astra-video Sep 12, 2025 •

edited

Loading

ad-astra-video Sep 12, 2025 •

edited

Loading

ad-astra-video Sep 12, 2025 •

edited

Loading

pschroedl Sep 11, 2025 •

edited

Loading

pschroedl Sep 11, 2025 •

edited

Loading

ad-astra-video Sep 12, 2025 •

edited

Loading