diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index f9afe08498..b4dc336583 100644 --- a/CHANGELOG_PENDING.md +++ b/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 diff --git a/byoc/job_orchestrator.go b/byoc/job_orchestrator.go index 210474992e..54fb70eb3e 100644 --- a/byoc/job_orchestrator.go +++ b/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) } } diff --git a/byoc/stream_orchestrator.go b/byoc/stream_orchestrator.go index 359e269a61..62ff1b7c54 100644 --- a/byoc/stream_orchestrator.go +++ b/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) } diff --git a/byoc/trickle.go b/byoc/trickle.go index 8a4d98869d..72a5dc9fa5 100644 --- a/byoc/trickle.go +++ b/byoc/trickle.go @@ -404,7 +404,11 @@ 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 ctx.Err() != nil { + clog.V(common.DEBUG).Infof(ctx, "Process err=%v output: %s", err, output) + } else { + clog.Infof(ctx, "Process err=%v output: %s", err, output) + } select { case <-ctx.Done(): diff --git a/core/ai_orchestrator.go b/core/ai_orchestrator.go index 0155fe85da..6746073150 100644 --- a/core/ai_orchestrator.go +++ b/core/ai_orchestrator.go @@ -1139,51 +1139,52 @@ 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() + 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) { diff --git a/core/external_capabilities.go b/core/external_capabilities.go index 27ba79e45f..517fccbae0 100644 --- a/core/external_capabilities.go +++ b/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 diff --git a/core/external_capabilities_test.go b/core/external_capabilities_test.go index ed0ff1e469..79473f985b 100644 --- a/core/external_capabilities_test.go +++ b/core/external_capabilities_test.go @@ -220,6 +220,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,9 +304,18 @@ 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 diff --git a/core/orchestrator.go b/core/orchestrator.go index 0d4d8417a2..7f93dfaefc 100644 --- a/core/orchestrator.go +++ b/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) diff --git a/media/rtmp2segment.go b/media/rtmp2segment.go index 5655476b56..6130ab9195 100644 --- a/media/rtmp2segment.go +++ b/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 { diff --git a/server/ai_live_video.go b/server/ai_live_video.go index 5d38f44e17..8773fc5897 100644 --- a/server/ai_live_video.go +++ b/server/ai_live_video.go @@ -450,7 +450,11 @@ 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 ctx.Err() != nil { + clog.V(common.DEBUG).Infof(ctx, "Process err=%v output: %s", err, output) + } else { + clog.Infof(ctx, "Process err=%v output: %s", err, output) + } select { case <-ctx.Done(): diff --git a/trickle/trickle_server.go b/trickle/trickle_server.go index 23fcb95a7c..ce15086516 100644 --- a/trickle/trickle_server.go +++ b/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 } } diff --git a/trickle/trickle_subscriber.go b/trickle/trickle_subscriber.go index 420dc2a9ea..a1f4dcd6aa 100644 --- a/trickle/trickle_subscriber.go +++ b/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 } diff --git a/trickle/trickle_test.go b/trickle/trickle_test.go index d5bd2a08ae..8e621bfd1e 100644 --- a/trickle/trickle_test.go +++ b/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()