Clear previous gateway status on new requests (#3421)

Clear any previous gateway status and allow the stream status API to return the gateway status if no stream status from the runner has been received. This is to allow the front end to properly detect things like capacity errors

Author: mjh1

server/ai_live_video.go

@@ -550,7 +550,10 @@ func (s *SlowOrchChecker) GetCount() int {
 
 func LiveErrorEventSender(ctx context.Context, streamID string, event map[string]string) func(err error) {
 	return func(err error) {
-		GatewayStatus.Store(streamID, map[string]interface{}{"last_error": err.Error()})
+		GatewayStatus.Store(streamID, map[string]interface{}{
+			"last_error":      err.Error(),
+			"last_error_time": time.Now().UnixMilli(),
+		})
 
 		ev := maps.Clone(event)
 		ev["capability"] = clog.GetVal(ctx, "capability")

server/ai_mediaserver.go

@@ -508,6 +508,9 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 		ctx = clog.AddVal(ctx, "stream_id", streamID)
 		clog.Infof(ctx, "Received live video AI request for %s. pipelineParams=%v", streamName, pipelineParams)
 
+		// Clear any previous gateway status
+		GatewayStatus.Clear(streamID)
+
 		monitor.SendQueueEventAsync("stream_trace", map[string]interface{}{
 			"type":        "gateway_receive_stream_request",
 			"timestamp":   streamRequestTime,
@@ -663,15 +666,16 @@ func (ls *LivepeerServer) GetLiveVideoToVideoStatus() http.Handler {
 
 		// Get status for specific stream
 		status, exists := StreamStatusStore.Get(streamId)
-		if !exists {
+		gatewayStatus, gatewayExists := GatewayStatus.Get(streamId)
+		if !exists && !gatewayExists {
 			http.Error(w, "Stream not found", http.StatusNotFound)
 			return
 		}
-		gatewayStatus, exists := GatewayStatus.Get(streamId)
-		if exists {
-			for k, v := range gatewayStatus {
-				status["gateway_"+k] = v
+		if gatewayExists {
+			if status == nil {
+				status = make(map[string]any)
 			}
+			status["gateway_status"] = gatewayStatus
 		}
 
 		w.Header().Set("Content-Type", "application/json")