Merge pull request #3727 from ad-astra-video/av/add-byoc-streaming

BYOC: add streaming

Author: pschroedl

CHANGELOG_PENDING.md

@@ -23,5 +23,6 @@
 * [#3777](https://github.com/livepeer/go-livepeer/pull/3777) docker: Forcefully SIGKILL runners after timeout (@pwilczynskiclearcode)
 * [#3779](https://github.com/livepeer/go-livepeer/pull/3779) worker: Fix orphaned containers on node shutdown (@victorges)
 * [#3781](https://github.com/livepeer/go-livepeer/pull/3781) worker/docker: Destroy containers from watch routines (@victorges)
+* [#3727](https://github.com/livepeer/go-livepeer/pull/3727) BYOC: add streaming for BYOC pipelines using trickle (@ad-astra-video)
 
 #### CLI

common/testutil.go

@@ -82,6 +82,11 @@ func (s *StubServerStream) Send(n *net.NotifySegment) error {
 func IgnoreRoutines() []goleak.Option {
 	// goleak works by making list of all running goroutines and reporting error if it finds any
 	// this list tells goleak to ignore these goroutines - we're not interested in these particular goroutines
+	// following added for job_stream_tests, believe related to open connections on trickle server that are cleaned up periodically
+	//   net/http.(*persistConn).mapRoundTripError
+	//   net/http.(*persistConn).readLoop
+	//   net/http.(*persistConn).writeLoop
+	//   io.(*pipe).read
 	funcs2ignore := []string{"github.com/golang/glog.(*loggingT).flushDaemon", "go.opencensus.io/stats/view.(*worker).start",
 		"github.com/rjeczalik/notify.(*recursiveTree).dispatch", "github.com/rjeczalik/notify._Cfunc_CFRunLoopRun", "github.com/ethereum/go-ethereum/metrics.(*meterArbiter).tick",
 		"github.com/ethereum/go-ethereum/consensus/ethash.(*Ethash).remote", "github.com/ethereum/go-ethereum/core.(*txSenderCacher).cache",
@@ -93,6 +98,12 @@ func IgnoreRoutines() []goleak.Option {
 		"github.com/livepeer/go-livepeer/core.(*Balances).StartCleanup",
 		"internal/synctest.Run",
 		"testing/synctest.testingSynctestTest",
+		"github.com/livepeer/go-livepeer/server.startTrickleSubscribe.func2",
+		"net/http.(*persistConn).mapRoundTripError",
+		"net/http.(*persistConn).readLoop",
+		"net/http.(*persistConn).writeLoop",
+		"io.(*pipe).read",
+		"github.com/livepeer/go-livepeer/media.gatherIncomingTracks",
 	}
 	ignoreAnywhereFuncs := []string{
 		// glog’s file flusher often has syscall/os.* on top
@@ -104,7 +115,6 @@ func IgnoreRoutines() []goleak.Option {
 		res = append(res, goleak.IgnoreTopFunction(f))
 	}
 	for _, f := range ignoreAnywhereFuncs {
-		// ignore if these function signatures appear anywhere in the call stack
 		res = append(res, goleak.IgnoreAnyFunction(f))
 	}
 	return res

core/accounting.go

@@ -66,9 +66,10 @@ func (b *Balance) Balance() *big.Rat {
 
 // AddressBalances holds credit balances for ETH addresses
 type AddressBalances struct {
-	balances map[ethcommon.Address]*Balances
-	mtx      sync.Mutex
-	ttl      time.Duration
+	balances     map[ethcommon.Address]*Balances
+	mtx          sync.Mutex
+	sharedBalMtx sync.Mutex
+	ttl          time.Duration
 }
 
 // NewAddressBalances creates a new AddressBalances instance
@@ -99,6 +100,47 @@ func (a *AddressBalances) Balance(addr ethcommon.Address, id ManifestID) *big.Ra
 	return a.balancesForAddr(addr).Balance(id)
 }
 
+// compares expected balance with current balance and updates accordingly with the expected balance being the target
+// returns the difference and if minimum balance was covered
+// also returns if balance was reset to zero because expected was zero
+func (a *AddressBalances) CompareAndUpdateBalance(addr ethcommon.Address, id ManifestID, expected *big.Rat, minimumBal *big.Rat) (*big.Rat, *big.Rat, bool, bool) {
+	a.sharedBalMtx.Lock()
+	defer a.sharedBalMtx.Unlock()
+	current := a.balancesForAddr(addr).Balance(id)
+	if current == nil {
+		//create a balance of 1 to start tracking
+		a.Debit(addr, id, big.NewRat(0, 1))
+		current = a.balancesForAddr(addr).Balance(id)
+	}
+	if expected == nil {
+		expected = big.NewRat(0, 1)
+	}
+	diff := new(big.Rat).Sub(expected, current)
+
+	if diff.Sign() > 0 {
+		a.Credit(addr, id, diff)
+	} else {
+		a.Debit(addr, id, new(big.Rat).Abs(diff))
+	}
+
+	var resetToZero bool
+	if expected.Sign() == 0 {
+		a.Debit(addr, id, current)
+
+		resetToZero = true
+	}
+
+	//get updated balance after changes
+	current = a.balancesForAddr(addr).Balance(id)
+
+	var minimumBalCovered bool
+	if current.Cmp(minimumBal) >= 0 {
+		minimumBalCovered = true
+	}
+
+	return current, diff, minimumBalCovered, resetToZero
+}
+
 // StopCleanup stops the cleanup loop for all balances
 func (a *AddressBalances) StopCleanup() {
 	a.mtx.Lock()

core/accounting_test.go

@@ -265,3 +265,97 @@ func TestBalancesCleanup(t *testing.T) {
 	// Now balance for mid1 should be cleaned as well
 	assert.Nil(b.Balance(mid1))
 }
+
+func TestAddressBalances_CompareAndUpdateBalance(t *testing.T) {
+	addr := ethcommon.BytesToAddress([]byte("foo"))
+	mid := ManifestID("some manifestID")
+	balances := NewAddressBalances(1 * time.Minute)
+	defer balances.StopCleanup()
+
+	assert := assert.New(t)
+
+	// Test 1: Balance doesn't exist - should initialize to 1 and then update to expected
+	expected := big.NewRat(10, 1)
+	minimumBal := big.NewRat(5, 1)
+	current, diff, minimumBalCovered, resetToZero := balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to expected value")
+	assert.Zero(big.NewRat(10, 1).Cmp(diff), "Diff should be expected - initial (10 - 1)")
+	assert.True(minimumBalCovered, "Minimum balance should be covered when going from 1 to 10")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 2: Expected > Current (Credit scenario)
+	expected = big.NewRat(20, 1)
+	minimumBal = big.NewRat(15, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to expected value")
+	assert.Zero(big.NewRat(10, 1).Cmp(diff), "Diff should be 20 - 10 = 10")
+	assert.True(minimumBalCovered, "Minimum balance should be covered when crossing threshold")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 3: Expected < Current (Debit scenario)
+	expected = big.NewRat(5, 1)
+	minimumBal = big.NewRat(3, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to expected value")
+	assert.Zero(big.NewRat(-15, 1).Cmp(diff), "Diff should be 5 - 20 = -15")
+	assert.True(minimumBalCovered, "Minimum balance should still be covered")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 4: Expected == Current (No change)
+	expected = big.NewRat(5, 1)
+	minimumBal = big.NewRat(3, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should remain the same")
+	assert.Zero(big.NewRat(0, 1).Cmp(diff), "Diff should be 0")
+	assert.True(minimumBalCovered, "Minimum balance should still be covered")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 5: Reset to zero (current > 0, expected = 0)
+	balances.Credit(addr, mid, big.NewRat(5, 1)) // Set current to 10
+	expected = big.NewRat(0, 1)
+	minimumBal = big.NewRat(3, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be reset to zero")
+	assert.Zero(big.NewRat(-10, 1).Cmp(diff), "Diff should be 0 - 10 = -10")
+	assert.False(minimumBalCovered, "Minimum balance should not be covered when resetting to zero")
+	assert.True(resetToZero, "Should be marked as reset to zero")
+
+	// Test 6: Minimum balance covered threshold - just below to just above
+	expected = big.NewRat(2, 1)
+	minimumBal = big.NewRat(5, 1)
+	balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal) // Set to 2
+
+	expected = big.NewRat(5, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to 5")
+	assert.Zero(big.NewRat(3, 1).Cmp(diff), "Diff should be 5 - 2 = 3")
+	assert.True(minimumBalCovered, "Minimum balance should be covered when crossing from below to at threshold")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 7: Minimum balance not covered - already above threshold
+	expected = big.NewRat(10, 1)
+	minimumBal = big.NewRat(5, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to 10")
+	assert.Zero(big.NewRat(5, 1).Cmp(diff), "Diff should be 10 - 5 = 5")
+	assert.True(minimumBalCovered, "Minimum balance should still be covered")
+	assert.False(resetToZero, "Should not be reset to zero")
+
+	// Test 8: Negative balance handling
+	balances.Debit(addr, mid, big.NewRat(20, 1)) // Force negative: 10 - 20 = -10
+	expected = big.NewRat(5, 1)
+	minimumBal = big.NewRat(3, 1)
+	current, diff, minimumBalCovered, resetToZero = balances.CompareAndUpdateBalance(addr, mid, expected, minimumBal)
+
+	assert.Zero(expected.Cmp(current), "Balance should be updated to expected value")
+	assert.Zero(big.NewRat(15, 1).Cmp(diff), "Diff should be 5 - (-10) = 15")
+	assert.True(minimumBalCovered, "Minimum balance should be covered when going from negative to positive above minimum")
+	assert.False(resetToZero, "Should not be reset to zero")
+}

core/ai_orchestrator.go

@@ -1163,8 +1163,8 @@ func (orch *orchestrator) CheckExternalCapabilityCapacity(extCapability string)
 func (orch *orchestrator) ReserveExternalCapabilityCapacity(extCapability string) error {
 	cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]
 	if ok {
-		cap.mu.Lock()
-		defer cap.mu.Unlock()
+		cap.Mu.Lock()
+		defer cap.Mu.Unlock()
 
 		cap.Load++
 		return nil
@@ -1176,8 +1176,8 @@ func (orch *orchestrator) ReserveExternalCapabilityCapacity(extCapability string
 func (orch *orchestrator) FreeExternalCapabilityCapacity(extCapability string) error {
 	cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]
 	if ok {
-		cap.mu.Lock()
-		defer cap.mu.Unlock()
+		cap.Mu.Lock()
+		defer cap.Mu.Unlock()
 
 		cap.Load--
 		return nil
@@ -1200,6 +1200,12 @@ func (orch *orchestrator) JobPriceInfo(sender ethcommon.Address, jobCapability s
 		return nil, err
 	}
 
+	//ensure price numerator and denominator can be int64
+	jobPrice, err = common.PriceToInt64(jobPrice)
+	if err != nil {
+		return nil, fmt.Errorf("invalid job price: %w", err)
+	}
+
 	return &net.PriceInfo{
 		PricePerUnit:  jobPrice.Num().Int64(),
 		PixelsPerUnit: jobPrice.Denom().Int64(),