Revert "BYOC: add streaming" (#3804)

This commit reverts #3804 because it was merged with a failing test and without review from members of the other core teams. The plan is to resubmit it for proper review and cross-team discussion.

Author: rickstaa

CHANGELOG_PENDING.md

@@ -23,6 +23,5 @@
 * [#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,11 +82,6 @@ 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",
@@ -98,12 +93,6 @@ 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
@@ -115,6 +104,7 @@ 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,10 +66,9 @@ func (b *Balance) Balance() *big.Rat {
 
 // AddressBalances holds credit balances for ETH addresses
 type AddressBalances struct {
-	balances     map[ethcommon.Address]*Balances
-	mtx          sync.Mutex
-	sharedBalMtx sync.Mutex
-	ttl          time.Duration
+	balances map[ethcommon.Address]*Balances
+	mtx      sync.Mutex
+	ttl      time.Duration
 }
 
 // NewAddressBalances creates a new AddressBalances instance
@@ -100,47 +99,6 @@ 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,97 +265,3 @@ 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,12 +1200,6 @@ 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(),