BYOC: fix orchestrator stream setup when fails

CHANGELOG_PENDING.md

@@ -19,6 +19,7 @@
 * [#3814](https://github.com/livepeer/go-livepeer/pull/3814) ai/worker: Add scope pipeline support to worker and build scripts (@victorges)
 * [#3823](https://github.com/livepeer/go-livepeer/pull/3823) ai/worker: Add sd15-v2v image support (@victorges)
 * [#3843](https://github.com/livepeer/go-livepeer/pull/3843) ai/worker: Add sdxl-v2v image support (@victorges)
+* [#3849](https://github.com/livepeer/go-livepeer/pull/3849) byoc: fix orchestrator stream setup when fails (@ad-astra-video)
 
 #### Transcoder
 

byoc/stream_orchestrator.go

@@ -70,6 +70,26 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 			eventsCh     *trickle.TrickleLocalPublisher
 			dataCh       *trickle.TrickleLocalPublisher
 		)
+		failedToStartStream := false
+
+		// reset trickle channels and release capacity on failure
+		defer func() {
+			if failedToStartStream {
+				bso.orch.FreeExternalCapabilityCapacity(orchJob.Req.Capability)
+				//close the trickle channels
+				if pubCh != nil {
+					pubCh.Close()
+				}
+				if subCh != nil {
+					subCh.Close()
+				}
+				if dataCh != nil {
+					dataCh.Close()
+				}
+				controlPubCh.Close()
+				eventsCh.Close()
+			}
+		}()
 
 		reqBodyForRunner := make(map[string]interface{})
 		reqBodyForRunner["gateway_request_id"] = mid
@@ -115,6 +135,7 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 		if err := json.Unmarshal(body, &bodyJSON); err != nil {
 			clog.Errorf(ctx, "Failed to parse body as JSON: %v", err)
 			http.Error(w, "Invalid JSON body", http.StatusBadRequest)
+			failedToStartStream = true
 			return
 		}
 		for key, value := range bodyJSON {
@@ -125,6 +146,7 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 		if err != nil {
 			clog.Errorf(ctx, "Failed to marshal request body err=%v", err)
 			http.Error(w, "Failed to marshal request body", http.StatusInternalServerError)
+			failedToStartStream = true
 			return
 		}
 
@@ -140,13 +162,15 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 		if err != nil {
 			clog.Errorf(ctx, "Error sending request to worker %v: %v", workerRoute, err)
 			respondWithError(w, "Error sending request to worker", http.StatusInternalServerError)
+			failedToStartStream = true
 			return
 		}
 
 		respBody, err := io.ReadAll(resp.Body)
 		if err != nil {
 			clog.Errorf(ctx, "Error reading response body: %v", err)
 			respondWithError(w, "Error reading response body", http.StatusInternalServerError)
+			failedToStartStream = true
 			return
 		}
 		defer resp.Body.Close()
@@ -160,6 +184,7 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 			//return error response from the worker
 			w.WriteHeader(resp.StatusCode)
 			w.Write(respBody)
+			failedToStartStream = true
 			return
 		}
 
@@ -173,6 +198,7 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler {
 		if err != nil {
 			clog.Errorf(ctx, "Error adding stream to external capabilities: %v", err)
 			respondWithError(w, "Error adding stream to external capabilities", http.StatusInternalServerError)
+			failedToStartStream = true
 			return
 		}