error handling, logging, lock stream info when checking active

Author: ad-astra-video

core/ai_orchestrator.go

@@ -1203,7 +1203,7 @@ func (orch *orchestrator) JobPriceInfo(sender ethcommon.Address, jobCapability s
 	//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)
 	}
 
 	return &net.PriceInfo{

core/external_capabilities.go

@@ -64,6 +64,8 @@ type StreamInfo struct {
 }
 
 func (sd *StreamInfo) IsActive() bool {
+	sd.sdm.Lock()
+	defer sd.sdm.Unlock()
 	if sd.StreamCtx.Err() != nil {
 		return false
 	}
@@ -128,19 +130,29 @@ func (extCaps *ExternalCapabilities) AddStream(streamID string, capability strin
 
 		//orchestrator channels shutdown
 		if stream.pubChannel != nil {
-			stream.pubChannel.Close()
+			if err := stream.pubChannel.Close(); err != nil {
+				glog.Errorf("error closing pubChannel for stream=%s: %v", streamID, err)
+			}
 		}
 		if stream.subChannel != nil {
-			stream.subChannel.Close()
+			if err := stream.subChannel.Close(); err != nil {
+				glog.Errorf("error closing subChannel for stream=%s: %v", streamID, err)
+			}
 		}
 		if stream.controlChannel != nil {
-			stream.controlChannel.Close()
+			if err := stream.controlChannel.Close(); err != nil {
+				glog.Errorf("error closing controlChannel for stream=%s: %v", streamID, err)
+			}
 		}
 		if stream.eventsChannel != nil {
-			stream.eventsChannel.Close()
+			if err := stream.eventsChannel.Close(); err != nil {
+				glog.Errorf("error closing eventsChannel for stream=%s: %v", streamID, err)
+			}
 		}
 		if stream.dataChannel != nil {
-			stream.dataChannel.Close()
+			if err := stream.dataChannel.Close(); err != nil {
+				glog.Errorf("error closing dataChannel for stream=%s: %v", streamID, err)
+			}
 		}
 		return
 	}()

server/job_stream.go

@@ -100,6 +100,11 @@ func (ls *LivepeerServer) StopStream() http.Handler {
 		stopJob.sign() //no changes to make, sign job
 
 		token, err := sessionToToken(params.liveParams.sess)
+		if err != nil {
+			clog.Errorf(ctx, "Error converting session to token: %s", err)
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
 		newToken, err := getToken(ctx, 3*time.Second, token.ServiceAddr, stopJob.Job.Req.Capability, stopJob.Job.Req.Sender, stopJob.Job.Req.Sig)
 		if err != nil {
 			clog.Errorf(ctx, "Error converting session to token: %s", err)