close data segment writer after right and put segment writer back to 5 segment history

Author: ad-astra-video

server/ai_live_video.go

@@ -864,6 +864,7 @@ func startDataSubscribe(ctx context.Context, url *url.URL, params aiRequestParam
 			scanner := bufio.NewScanner(segment.Body)
 			for scanner.Scan() {
 				writer, err := dataWriter.Next()
+				clog.V(8).Infof(ctx, "data subscribe writing seq=%d", seq)
 				if err != nil {
 					if err != io.EOF {
 						stopProcessing(ctx, params, fmt.Errorf("data subscribe could not get next: %w", err))
@@ -876,6 +877,8 @@ func startDataSubscribe(ctx context.Context, url *url.URL, params aiRequestParam
 				}
 				readBytes += n
 				readMessages += 1
+
+				writer.Close()
 			}
 			if err := scanner.Err(); err != nil {
 				clog.InfofErr(ctx, "data subscribe error reading seq=%d", seq, err)

server/job_stream.go

@@ -598,7 +598,7 @@ func (ls *LivepeerServer) setupStream(ctx context.Context, r *http.Request, job
 		return nil, http.StatusBadRequest, errors.New("invalid job parameters")
 	}
 	if jobParams.EnableDataOutput {
-		params.liveParams.dataWriter = media.NewSegmentWriter(1)
+		params.liveParams.dataWriter = media.NewSegmentWriter(5)
 	}
 
 	//check if stream exists