Quiet expected teardown logs and lock capability reads

CHANGELOG_PENDING.md

@@ -18,6 +18,9 @@
 
 #### General
 
+- \#3922 Demote expected cancellation logs during stream teardown (@vavo)
+- \#3885 Protect external capability map reads with the capability lock (@vavo)
+
 #### Broadcaster
 
 #### CLI

byoc/job_orchestrator.go

@@ -274,9 +274,12 @@ func (bso *BYOCOrchestratorServer) processJob(ctx context.Context, w http.Respon
 	req.Header.Add("Content-Type", r.Header.Get("Content-Type"))
 
 	// Add Authorization header if auth token is set for this capability
-	if extCap, ok := bso.node.ExternalCapabilities.Capabilities[orchJob.Req.Capability]; ok {
-		if extCap.AuthToken != "" {
-			req.Header.Add("Authorization", "Bearer "+extCap.AuthToken)
+	if extCap, ok := bso.node.ExternalCapabilities.GetCapability(orchJob.Req.Capability); ok {
+		extCap.Mu.RLock()
+		authToken := extCap.AuthToken
+		extCap.Mu.RUnlock()
+		if authToken != "" {
+			req.Header.Add("Authorization", "Bearer "+authToken)
 		}
 	}
 

byoc/stream_orchestrator.go

@@ -236,7 +236,7 @@ func (bso *BYOCOrchestratorServer) monitorOrchStream(job *orchJob) {
 			return
 		case <-pmtTicker.C:
 			// Check payment status
-			extCap, ok := bso.node.ExternalCapabilities.Capabilities[capability]
+			extCap, ok := bso.node.ExternalCapabilities.GetCapability(capability)
 			if !ok {
 				clog.Errorf(ctx, "Capability not found for payment monitoring, exiting monitoring capability=%s", capability)
 				return
@@ -414,7 +414,9 @@ func (bso *BYOCOrchestratorServer) createWorkerReq(ctx context.Context, workerRo
 	}
 
 	// Add Authorization header if auth token is set for this capability
-	if extCap, ok := bso.node.ExternalCapabilities.Capabilities[capability]; ok {
+	if extCap, ok := bso.node.ExternalCapabilities.GetCapability(capability); ok {
+		extCap.Mu.RLock()
+		defer extCap.Mu.RUnlock()
 		if extCap.AuthToken != "" {
 			req.Header.Add("Authorization", "Bearer "+extCap.AuthToken)
 		}

byoc/trickle.go

@@ -404,7 +404,13 @@ func (bsg *BYOCGatewayServer) ffmpegOutput(ctx context.Context, outputUrl string
 		cmd.WaitDelay = 5 * time.Second
 		cmd.Stdin = outWriter.MakeReader() // start at leading edge of output for each retry
 		output, err := cmd.CombinedOutput()
-		clog.Infof(ctx, "Process err=%v output: %s", err, output)
+		if err != nil && ctx.Err() != nil {
+			clog.V(common.DEBUG).Infof(ctx, "Process err=%v output: %s", err, output)
+		} else if err != nil {
+			clog.Errorf(ctx, "Process err=%v output: %s", err, output)
+		} else {
+			clog.Infof(ctx, "Process err=%v output: %s", err, output)
+		}
 
 		select {
 		case <-ctx.Done():

core/ai_orchestrator.go

@@ -1139,51 +1139,55 @@ func (orch *orchestrator) RemoveExternalCapability(extCapability string) error {
 }
 
 func (orch *orchestrator) GetUrlForCapability(extCapability string) string {
-	for _, capability := range orch.node.ExternalCapabilities.Capabilities {
-		if capability.Name == extCapability {
-			return capability.Url
-		}
+	capability, ok := orch.node.ExternalCapabilities.GetCapability(extCapability)
+	if !ok {
+		return ""
 	}
 
-	return ""
+	capability.Mu.RLock()
+	defer capability.Mu.RUnlock()
+	return capability.Url
 }
 
 func (orch *orchestrator) CheckExternalCapabilityCapacity(extCapability string) int64 {
-	if cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]; !ok {
+	cap, ok := orch.node.ExternalCapabilities.GetCapability(extCapability)
+	if !ok {
 		return 0
-	} else {
-		if cap.Load < cap.Capacity {
-			return int64(cap.Capacity - cap.Load)
-		} else {
-			return 0
-		}
 	}
+
+	cap.Mu.RLock()
+	defer cap.Mu.RUnlock()
+	if cap.Load < cap.Capacity {
+		return int64(cap.Capacity - cap.Load)
+	}
+	return 0
 }
 
 func (orch *orchestrator) ReserveExternalCapabilityCapacity(extCapability string) error {
-	cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]
-	if ok {
-		cap.Mu.Lock()
-		defer cap.Mu.Unlock()
-
-		cap.Load++
-		return nil
-	} else {
+	cap, ok := orch.node.ExternalCapabilities.GetCapability(extCapability)
+	if !ok {
 		return errors.New("external capability not found")
 	}
+
+	cap.Mu.Lock()
+	defer cap.Mu.Unlock()
+	if cap.Load >= cap.Capacity {
+		return errors.New("external capability capacity exhausted")
+	}
+	cap.Load++
+	return nil
 }
 
 func (orch *orchestrator) FreeExternalCapabilityCapacity(extCapability string) error {
-	cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]
-	if ok {
-		cap.Mu.Lock()
-		defer cap.Mu.Unlock()
-
-		cap.Load--
-		return nil
-	} else {
+	cap, ok := orch.node.ExternalCapabilities.GetCapability(extCapability)
+	if !ok {
 		return errors.New("external capability not found")
 	}
+
+	cap.Mu.Lock()
+	defer cap.Mu.Unlock()
+	cap.Load--
+	return nil
 }
 
 func (orch *orchestrator) JobPriceInfo(sender ethcommon.Address, jobCapability string) (*net.PriceInfo, error) {

core/external_capabilities.go

@@ -103,7 +103,7 @@ func (sd *StreamInfo) cleanup() {
 }
 
 type ExternalCapabilities struct {
-	capm         sync.Mutex
+	capm         sync.RWMutex
 	Capabilities map[string]*ExternalCapability
 	Streams      map[string]*StreamInfo
 }
@@ -189,6 +189,25 @@ func (extCaps *ExternalCapabilities) StreamExists(streamID string) bool {
 	return ok
 }
 
+func (extCaps *ExternalCapabilities) GetCapability(name string) (*ExternalCapability, bool) {
+	extCaps.capm.RLock()
+	defer extCaps.capm.RUnlock()
+
+	capability, ok := extCaps.Capabilities[name]
+	return capability, ok
+}
+
+func (extCaps *ExternalCapabilities) GetCapabilityNames() []string {
+	extCaps.capm.RLock()
+	defer extCaps.capm.RUnlock()
+
+	names := make([]string, 0, len(extCaps.Capabilities))
+	for name := range extCaps.Capabilities {
+		names = append(names, name)
+	}
+	return names
+}
+
 func (extCaps *ExternalCapabilities) RemoveCapability(extCap string) {
 	extCaps.capm.Lock()
 	defer extCaps.capm.Unlock()
@@ -220,10 +239,12 @@ func (extCaps *ExternalCapabilities) RegisterCapability(extCapability string) (*
 		panic(fmt.Errorf("error converting price: %v", err))
 	}
 	if cap, ok := extCaps.Capabilities[extCap.Name]; ok {
+		cap.Mu.Lock()
 		cap.Url = extCap.Url
 		cap.Capacity = extCap.Capacity
 		cap.price = extCap.price
 		cap.AuthToken = extCap.AuthToken
+		cap.Mu.Unlock()
 	}
 
 	extCaps.Capabilities[extCap.Name] = &extCap

core/external_capabilities_test.go

@@ -3,6 +3,7 @@ package core
 import (
 	"encoding/json"
 	"math/big"
+	"sync"
 	"testing"
 
 	"github.com/livepeer/go-livepeer/eth"
@@ -220,6 +221,17 @@ func TestExternalCapability_GetPrice(t *testing.T) {
 	})
 }
 
+func TestExternalCapabilities_GetCapability(t *testing.T) {
+	extCaps := NewExternalCapabilities()
+	capability := &ExternalCapability{Name: "test-cap"}
+	extCaps.Capabilities[capability.Name] = capability
+
+	got, ok := extCaps.GetCapability(capability.Name)
+	assert.True(t, ok)
+	assert.Same(t, capability, got)
+	assert.Equal(t, []string{capability.Name}, extCaps.GetCapabilityNames())
+}
+
 func TestExternalCapabilities_MarshalJSON(t *testing.T) {
 	extCaps := NewExternalCapabilities()
 
@@ -293,11 +305,57 @@ func TestExternalCapabilities_Concurrency(t *testing.T) {
 			done <- true
 		}()
 
+		go func() {
+			for i := 0; i < 100; i++ {
+				extCaps.GetCapability("concurrent-test-" + string(rune('A'+i%26)))
+				extCaps.GetCapabilityNames()
+			}
+			done <- true
+		}()
+
 		// Wait for both goroutines to finish
 		<-done
 		<-done
+		<-done
 
 		// No assertions needed - if there are no race conditions during build with -race flag,
 		// then the test passes
 	})
 }
+
+func TestOrchestrator_ReserveExternalCapabilityCapacityIsAtomic(t *testing.T) {
+	extCaps := NewExternalCapabilities()
+	extCap := &ExternalCapability{Name: "test-cap", Capacity: 1}
+	extCaps.Capabilities[extCap.Name] = extCap
+	orch := &orchestrator{node: &LivepeerNode{ExternalCapabilities: extCaps}}
+
+	const requests = 32
+	start := make(chan struct{})
+	results := make(chan error, requests)
+	var wg sync.WaitGroup
+	for i := 0; i < requests; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			results <- orch.ReserveExternalCapabilityCapacity(extCap.Name)
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(results)
+
+	reserved := 0
+	for err := range results {
+		if err == nil {
+			reserved++
+		}
+	}
+
+	assert.Equal(t, 1, reserved)
+	extCap.Mu.RLock()
+	load := extCap.Load
+	extCap.Mu.RUnlock()
+	assert.Equal(t, 1, load)
+	assert.Zero(t, orch.CheckExternalCapabilityCapacity(extCap.Name))
+}

core/orchestrator.go

@@ -318,7 +318,7 @@ func (orch *orchestrator) GetCapabilitiesPrices(sender ethcommon.Address) ([]*ne
 	// The registered capability name is set as the Constraint, making BYOC
 	// pricing seamless alongside built-in capabilities like LiveVideoToVideo.
 	if orch.node != nil && orch.node.ExternalCapabilities != nil {
-		for name := range orch.node.ExternalCapabilities.Capabilities {
+		for _, name := range orch.node.ExternalCapabilities.GetCapabilityNames() {
 			price := orch.node.GetPriceForJob(ethAddr, name)
 			if price == nil {
 				price = orch.node.GetPriceForJob("default", name)

media/rtmp2segment.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/cenkalti/backoff"
 	"github.com/livepeer/go-livepeer/clog"
+	"github.com/livepeer/go-livepeer/common"
 	"golang.org/x/sys/unix"
 )
 
@@ -69,7 +70,11 @@ func (ms *MediaSegmenter) RunSegmentation(ctx context.Context, in string, segmen
 			return nil
 		}, backoff.WithMaxRetries(newExponentialBackOff(), 3))
 		if err != nil {
-			clog.Errorf(ctx, "Stopping segmentation in=%s err=%s", in, err)
+			if ctx.Err() != nil {
+				clog.V(common.DEBUG).Infof(ctx, "Stopping segmentation in=%s err=%s", in, err)
+			} else {
+				clog.Errorf(ctx, "Stopping segmentation in=%s err=%s", in, err)
+			}
 			break
 		}
 		if retryCount > 0 {
@@ -90,7 +95,11 @@ func (ms *MediaSegmenter) RunSegmentation(ctx context.Context, in string, segmen
 		cmd.WaitDelay = 5 * time.Second
 		output, err := cmd.CombinedOutput()
 		if err != nil {
-			clog.Errorf(ctx, "Error receiving RTMP: %v ffmpeg output: %s", err, output)
+			if ctx.Err() != nil {
+				clog.V(common.DEBUG).Infof(ctx, "Error receiving RTMP: %v ffmpeg output: %s", err, output)
+			} else {
+				clog.Errorf(ctx, "Error receiving RTMP: %v ffmpeg output: %s", err, output)
+			}
 			break
 		}
 		clog.Infof(ctx, "Segmentation stopped, will retry. retryCount=%d ffmpeg output: %s", retryCount, output)

server/ai_live_video.go

@@ -450,7 +450,13 @@ func ffmpegOutput(ctx context.Context, outputUrl string, outWriter *media.RingBu
 		cmd.WaitDelay = 5 * time.Second
 		cmd.Stdin = outWriter.MakeReader() // start at leading edge of output for each retry
 		output, err := cmd.CombinedOutput()
-		clog.Infof(ctx, "Process err=%v output: %s", err, output)
+		if err != nil && ctx.Err() != nil {
+			clog.V(common.DEBUG).Infof(ctx, "Process err=%v output: %s", err, output)
+		} else if err != nil {
+			clog.Errorf(ctx, "Process err=%v output: %s", err, output)
+		} else {
+			clog.Infof(ctx, "Process err=%v output: %s", err, output)
+		}
 
 		select {
 		case <-ctx.Done():

trickle/trickle_server.go

@@ -671,7 +671,11 @@ func (s *Stream) handleGet(w http.ResponseWriter, r *http.Request, idx int) {
 
 	if n, err := sendData(); err != nil {
 		// Handle write error or client disconnect
-		slog.Error("Error sending data to client", "stream", s.name, "idx", segment.idx, "sentBytes", n, "err", err)
+		if r.Context().Err() != nil {
+			slog.Debug("Error sending data to client", "stream", s.name, "idx", segment.idx, "sentBytes", n, "err", err)
+		} else {
+			slog.Error("Error sending data to client", "stream", s.name, "idx", segment.idx, "sentBytes", n, "err", err)
+		}
 		return
 	}
 }

trickle/trickle_subscriber.go

@@ -274,7 +274,11 @@ func (c *TrickleSubscriber) Read() (*http.Response, error) {
 		defer c.mu.Unlock()
 		nextConn, err := c.preconnect()
 		if err != nil {
-			slog.Error("failed to preconnect next segment", "url", c.url, "idx", c.idx, "err", err)
+			if errors.Is(err, context.Canceled) || c.baseCtx.Err() != nil {
+				slog.Debug("failed to preconnect next segment", "url", c.url, "idx", c.idx, "err", err)
+			} else {
+				slog.Error("failed to preconnect next segment", "url", c.url, "idx", c.idx, "err", err)
+			}
 			c.preconnectErrorCount++
 			return
 		}

trickle/trickle_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"sync"
@@ -15,6 +16,30 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestTrickleServer_CanceledClientDisconnectIsDebug(t *testing.T) {
+	var logs bytes.Buffer
+	previousLogger := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
+	defer slog.SetDefault(previousLogger)
+
+	stream := &Stream{
+		segments:  make([]*Segment, maxSegmentsPerStream),
+		name:      "test",
+		mimeType:  "video/MP2T",
+		nextWrite: 1,
+	}
+	stream.segments[0] = newSegment(0)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	req := httptest.NewRequest(http.MethodGet, "/test/0", nil).WithContext(ctx)
+	stream.handleGet(httptest.NewRecorder(), req, 0)
+
+	output := logs.String()
+	require.Contains(t, output, "level=DEBUG")
+	require.NotContains(t, output, "level=ERROR")
+}
+
 func TestTrickle_Close(t *testing.T) {
 	require := require.New(t)
 	mux := http.NewServeMux()