1512 lines
64 KiB
C#
1512 lines
64 KiB
C#
using System.Diagnostics;
|
|
using LiveRecorder.Application.Abstractions.Logging;
|
|
using LiveRecorder.Application.Abstractions.Notifications;
|
|
using LiveRecorder.Application.Abstractions.Platforms;
|
|
using LiveRecorder.Application.Abstractions.Recording;
|
|
using LiveRecorder.Application.Abstractions.Scripting;
|
|
using LiveRecorder.Application.Abstractions.Settings;
|
|
using LiveRecorder.Application.Services;
|
|
using LiveRecorder.Domain.Entities;
|
|
using LiveRecorder.Domain.Enums;
|
|
using LiveRecorder.Infrastructure.Persistence;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace LiveRecorder.Infrastructure.Services;
|
|
|
|
public sealed partial class FfmpegService
|
|
{
|
|
private static readonly TimeSpan RuntimeSourceFailureWindow = TimeSpan.FromSeconds(20);
|
|
private static readonly TimeSpan RuntimeOfflineVerificationStopTimeout = TimeSpan.FromSeconds(20);
|
|
private static readonly TimeSpan RuntimeOfflineVerificationKillTimeout = TimeSpan.FromSeconds(8);
|
|
|
|
private async Task HandleProcessOutputAsync(SessionProcessRuntime runtime, string? line, bool isError)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(line))
|
|
{
|
|
return;
|
|
}
|
|
|
|
_logger.LogDebug("ffmpeg[{SessionId}] {Line}", runtime.RecordSessionId, line);
|
|
if (TryParseSegmentOpenPath(line, out var openedPath))
|
|
{
|
|
await HandleSegmentOpenedAsync(runtime, openedPath);
|
|
return;
|
|
}
|
|
|
|
if (!runtime.HasOpenedFirstSegment && IsOptionCompatibilityFailureLine(line))
|
|
{
|
|
runtime.MarkStartupFailure(StartupFailureKind.InputOptionCompatibility, line);
|
|
}
|
|
else if (!runtime.HasOpenedFirstSegment && IsRetryableStartupFailureLine(line))
|
|
{
|
|
runtime.MarkStartupFailure(StartupFailureKind.StreamHandshake, line);
|
|
}
|
|
|
|
if (TryClassifyPersistedFfmpegLine(line, isError, out var level))
|
|
{
|
|
await PersistFfmpegLineAsync(runtime, line, level);
|
|
}
|
|
|
|
if (IsRuntimeSourceFailureLine(line) &&
|
|
runtime.RegisterRuntimeSourceFailure(DateTimeOffset.UtcNow, RuntimeSourceFailureWindow))
|
|
{
|
|
_ = ValidateRuntimeSourceFailureAsync(runtime, line);
|
|
}
|
|
}
|
|
|
|
private static bool TryClassifyPersistedFfmpegLine(string line, bool isError, out SystemLogLevel level)
|
|
{
|
|
if (IsRecoverableFfmpegWarningLine(line))
|
|
{
|
|
level = SystemLogLevel.Warning;
|
|
return true;
|
|
}
|
|
|
|
if (line.Contains("error", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("fail", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("timed out", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
level = isError ? SystemLogLevel.Error : SystemLogLevel.Warning;
|
|
return true;
|
|
}
|
|
|
|
level = default;
|
|
return false;
|
|
}
|
|
|
|
private static bool IsRecoverableFfmpegWarningLine(string line)
|
|
{
|
|
return line.Contains("Will reconnect at", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("keepalive request failed", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("retrying with new connection", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("IO error: Connection reset by peer", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Error in the push function", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Network is unreachable", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("HTTP error 404", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("404 Not Found", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to open segment", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to resolve hostname", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("error=Connection reset by peer", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("error=End of file", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Concatenated FLV detected", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("might fail to demux, decode and seek", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to parse temporal unit", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to read unit", StringComparison.OrdinalIgnoreCase);
|
|
}
|
|
|
|
private static bool IsRuntimeSourceFailureLine(string line)
|
|
{
|
|
if (line.Contains("Will reconnect at", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("retrying with new connection", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
return line.Contains("HTTP error 404", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("404 Not Found", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Input/output error", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to open segment", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Network is unreachable", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Failed to resolve hostname", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Connection to tcp://", StringComparison.OrdinalIgnoreCase) &&
|
|
line.Contains("failed", StringComparison.OrdinalIgnoreCase);
|
|
}
|
|
|
|
private async Task PersistFfmpegLineAsync(SessionProcessRuntime runtime, string line, SystemLogLevel level)
|
|
{
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
level,
|
|
"FFmpeg",
|
|
"ffmpeg reported a warning or error line.",
|
|
line,
|
|
runtime.LiveRoomId,
|
|
runtime.RecordSessionId,
|
|
runtime.CurrentTaskId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Persist ffmpeg output failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
}
|
|
|
|
private async Task ValidateRuntimeSourceFailureAsync(SessionProcessRuntime runtime, string failureLine)
|
|
{
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
|
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
|
|
var session = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
|
if (session?.LiveRoom is null || !IsActiveSessionStatus(session.Status) || !_processes.ContainsKey(session.Id))
|
|
{
|
|
return;
|
|
}
|
|
|
|
var observedAt = DateTimeOffset.UtcNow;
|
|
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
|
|
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId);
|
|
await liveRoomStatusService.ApplySnapshotAsync(session.LiveRoom, liveStatus, observedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
if (liveStatus.IsLive)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Repeated stream errors were detected, but the live room is still online.",
|
|
failureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
runtime.CurrentTaskId);
|
|
return;
|
|
}
|
|
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Repeated stream errors indicate the live room is offline. Completing the active recording session.",
|
|
failureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
runtime.CurrentTaskId);
|
|
|
|
var stopped = await StopAndWaitAsync(
|
|
session.Id,
|
|
markAsCompletedOnExit: true,
|
|
RuntimeOfflineVerificationStopTimeout,
|
|
CancellationToken.None);
|
|
if (!stopped)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"The recorder did not stop gracefully after offline verification. Force killing the ffmpeg process.",
|
|
failureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
runtime.CurrentTaskId);
|
|
|
|
var killed = await KillAndWaitAsync(
|
|
session.Id,
|
|
RuntimeOfflineVerificationKillTimeout,
|
|
CancellationToken.None);
|
|
if (!killed)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"The recorder did not exit before the forced-stop wait expired; the session will be finalized when the process exits.",
|
|
failureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
runtime.CurrentTaskId);
|
|
return;
|
|
}
|
|
}
|
|
|
|
await TryReconcileInactiveSessionAsync(session.Id, CancellationToken.None);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Runtime source failure verification failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
finally
|
|
{
|
|
runtime.CompleteRuntimeSourceFailureVerification();
|
|
}
|
|
}
|
|
|
|
private async Task HandleSegmentOpenedAsync(SessionProcessRuntime runtime, string openedPath)
|
|
{
|
|
await runtime.Gate.WaitAsync();
|
|
try
|
|
{
|
|
var normalizedOpenedPath = NormalizeAbsolutePath(openedPath);
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var session = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
|
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var now = DateTimeOffset.UtcNow;
|
|
var currentTask = session.RecordTasks.FirstOrDefault(item => item.Id == runtime.CurrentTaskId);
|
|
if (currentTask is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (runtime.SaveMode == RecordSaveMode.SingleFile)
|
|
{
|
|
runtime.HasOpenedFirstSegment = true;
|
|
currentTask.AttachProcess(runtime.ProcessId, now);
|
|
currentTask.MarkRunning(now);
|
|
session.AttachProcess(runtime.ProcessId, now);
|
|
session.MarkRunning(now);
|
|
session.ActivateSegment(1, now);
|
|
runtime.ResetCurrentRecorderSegmentPaths(normalizedOpenedPath);
|
|
if (!runtime.HasInitializedDanmaku)
|
|
{
|
|
await EnsureDanmakuSegmentAsync(runtime, currentTask, now);
|
|
runtime.HasInitializedDanmaku = true;
|
|
}
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
return;
|
|
}
|
|
|
|
var detectedSegmentIndex = ExtractSegmentIndex(openedPath);
|
|
var segmentIndex = !runtime.HasOpenedFirstSegment
|
|
? Math.Max(1, detectedSegmentIndex ?? runtime.CurrentSegmentIndex)
|
|
: Math.Max(
|
|
1,
|
|
detectedSegmentIndex ??
|
|
(string.Equals(
|
|
NormalizeAbsolutePath(openedPath),
|
|
NormalizeAbsolutePath(runtime.CurrentOutputFilePath),
|
|
StringComparison.OrdinalIgnoreCase)
|
|
? runtime.CurrentSegmentIndex
|
|
: runtime.CurrentSegmentIndex + 1));
|
|
if (!runtime.HasInitializedDanmaku)
|
|
{
|
|
await EnsureDanmakuSegmentAsync(runtime, currentTask, now);
|
|
runtime.HasInitializedDanmaku = true;
|
|
}
|
|
|
|
if (segmentIndex == runtime.CurrentSegmentIndex)
|
|
{
|
|
runtime.HasOpenedFirstSegment = true;
|
|
var finalOutputPath = NormalizeAbsolutePath(
|
|
currentTask.OutputFilePath ??
|
|
ResolveSegmentOutputPath(session.OutputPathPattern ?? runtime.OutputPathPattern, session.SaveMode, segmentIndex));
|
|
currentTask.MarkStarting(runtime.StreamUrl, finalOutputPath, currentTask.StartedAt ?? now);
|
|
currentTask.AttachProcess(runtime.ProcessId, now);
|
|
currentTask.MarkRunning(now);
|
|
session.AttachProcess(runtime.ProcessId, now);
|
|
session.MarkRunning(now);
|
|
session.ActivateSegment(segmentIndex, now);
|
|
runtime.CurrentOutputFilePath = normalizedOpenedPath;
|
|
runtime.TrackCurrentRecorderSegment(normalizedOpenedPath);
|
|
await dbContext.SaveChangesAsync();
|
|
return;
|
|
}
|
|
|
|
var previousTask = currentTask;
|
|
var previousTaskId = previousTask.Id;
|
|
var previousDurationSeconds = previousTask.StartedAt.HasValue
|
|
? Math.Max(0, (now - previousTask.StartedAt.Value).TotalSeconds)
|
|
: (double?)null;
|
|
var previousEffectiveOutputPath = previousTask.OutputFilePath ??
|
|
NormalizeAbsolutePath(ResolveSegmentOutputPath(session.OutputPathPattern ?? runtime.OutputPathPattern, session.SaveMode, previousTask.SegmentIndex));
|
|
|
|
if (ShouldTreatAsUnexpectedFragmentRollover(runtime, previousDurationSeconds))
|
|
{
|
|
runtime.CurrentOutputFilePath = normalizedOpenedPath;
|
|
runtime.TrackCurrentRecorderSegment(normalizedOpenedPath);
|
|
currentTask.AttachProcess(runtime.ProcessId, now);
|
|
currentTask.MarkRunning(now);
|
|
session.AttachProcess(runtime.ProcessId, now);
|
|
session.MarkRunning(now);
|
|
session.ActivateSegment(runtime.CurrentSegmentIndex, now);
|
|
await dbContext.SaveChangesAsync();
|
|
return;
|
|
}
|
|
|
|
var previousRecorderSegmentPaths = runtime.CaptureCurrentRecorderSegments();
|
|
|
|
var newTask = new RecordTask(
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
segmentIndex,
|
|
session.PreferredQuality,
|
|
session.OutputFormat,
|
|
now);
|
|
var newTaskOutputPath = NormalizeAbsolutePath(
|
|
ResolveSegmentOutputPath(session.OutputPathPattern ?? runtime.OutputPathPattern, session.SaveMode, segmentIndex));
|
|
newTask.MarkStarting(runtime.StreamUrl, newTaskOutputPath, now);
|
|
newTask.AttachProcess(runtime.ProcessId, now);
|
|
newTask.MarkRunning(now);
|
|
await dbContext.RecordTasks.AddAsync(newTask);
|
|
|
|
if (session.OutputFormat == RecordOutputFormat.Mp4)
|
|
{
|
|
previousTask.MarkProcessing("MP4 finalization queued.", now);
|
|
}
|
|
else
|
|
{
|
|
previousTask.MarkCompleted(now, previousDurationSeconds);
|
|
}
|
|
|
|
previousTask.DetachProcess(now);
|
|
|
|
session.AttachProcess(runtime.ProcessId, now);
|
|
session.MarkRunning(now);
|
|
session.ActivateSegment(segmentIndex, now);
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
runtime.CurrentTaskId = newTask.Id;
|
|
runtime.CurrentSegmentIndex = segmentIndex;
|
|
runtime.CurrentOutputFilePath = normalizedOpenedPath;
|
|
runtime.ResetCurrentRecorderSegmentPaths(normalizedOpenedPath);
|
|
runtime.HasOpenedFirstSegment = true;
|
|
|
|
if (runtime.DanmakuRecorder is not null)
|
|
{
|
|
await runtime.DanmakuRecorder.StartSegmentAsync(
|
|
newTask.Id,
|
|
segmentIndex,
|
|
NormalizeAbsolutePath(newTask.OutputFilePath ?? newTaskOutputPath),
|
|
now);
|
|
}
|
|
|
|
var previousDanmakuSummary = runtime.DanmakuRecorder?.TakeSummary(previousTaskId);
|
|
if (session.OutputFormat == RecordOutputFormat.Mp4)
|
|
{
|
|
SetPostProcessState(
|
|
session.Id,
|
|
previousTaskId,
|
|
"Queued",
|
|
0,
|
|
$"Waiting for {Path.GetFileName(GetRecorderOutputPath(previousEffectiveOutputPath, session.OutputFormat, session.SaveMode))} to finish writing");
|
|
_ = FinalizeCompletedSegmentAsync(
|
|
session.Id,
|
|
previousTaskId,
|
|
previousDurationSeconds,
|
|
now,
|
|
previousDanmakuSummary,
|
|
previousRecorderSegmentPaths);
|
|
}
|
|
else
|
|
{
|
|
await UpsertRecordResultAsync(
|
|
previousTask,
|
|
dbContext,
|
|
previousEffectiveOutputPath,
|
|
CalculateFileSize(previousEffectiveOutputPath),
|
|
previousTask.DurationSeconds,
|
|
previousDanmakuSummary?.FilePath,
|
|
previousDanmakuSummary?.MessageCount ?? 0,
|
|
now);
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
ClearPostProcessState(previousTask.Id);
|
|
var previousResult = await LoadRecordResultAsync(dbContext, previousTask.Id);
|
|
var eventScriptService = scope.ServiceProvider.GetRequiredService<IEventScriptService>();
|
|
await eventScriptService.RunSegmentCompletedAsync(
|
|
session.LiveRoom,
|
|
session,
|
|
previousTask,
|
|
previousResult,
|
|
previousEffectiveOutputPath,
|
|
now);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Handle segment open failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
finally
|
|
{
|
|
runtime.Gate.Release();
|
|
}
|
|
}
|
|
|
|
private async Task HandleProcessExitedAsync(SessionProcessRuntime runtime, Process process)
|
|
{
|
|
_processes.TryRemove(runtime.RecordSessionId, out _);
|
|
|
|
try
|
|
{
|
|
runtime.DanmakuCancellation.Cancel();
|
|
if (runtime.DanmakuPumpTask is not null)
|
|
{
|
|
try
|
|
{
|
|
await runtime.DanmakuPumpTask;
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Danmaku pump ended with error for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
}
|
|
|
|
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary = null;
|
|
if (runtime.DanmakuRecorder is not null)
|
|
{
|
|
activeDanmakuSummary = await runtime.DanmakuRecorder.CompleteActiveSegmentAsync();
|
|
await runtime.DanmakuRecorder.DisposeAsync();
|
|
}
|
|
|
|
if (runtime.DanmakuConnection is not null)
|
|
{
|
|
await runtime.DanmakuConnection.DisposeAsync();
|
|
}
|
|
|
|
if (await TryRecoverStartupFailureAsync(runtime))
|
|
{
|
|
return;
|
|
}
|
|
|
|
await FinalizeExitedSessionAsync(runtime, process, activeDanmakuSummary);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Handle ffmpeg exit failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
finally
|
|
{
|
|
runtime.ExitCompletion.TrySetResult(true);
|
|
runtime.Dispose();
|
|
process.Dispose();
|
|
}
|
|
}
|
|
|
|
private async Task<bool> TryRecoverStartupFailureAsync(SessionProcessRuntime runtime)
|
|
{
|
|
if (runtime.HasOpenedFirstSegment ||
|
|
runtime.StartupFailureKind == StartupFailureKind.None ||
|
|
string.IsNullOrWhiteSpace(runtime.LastStartupFailureLine))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
|
|
var session = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
|
|
|
if (session?.LiveRoom is null || !IsActiveSessionStatus(session.Status))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var currentTask = session.RecordTasks
|
|
.OrderBy(item => item.SegmentIndex)
|
|
.ThenBy(item => item.CreatedAt)
|
|
.FirstOrDefault(item => item.Id == runtime.CurrentTaskId);
|
|
|
|
if (currentTask is null || !IsActiveTaskStatus(currentTask.Status))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var observedAt = DateTimeOffset.UtcNow;
|
|
if (runtime.StartupFailureKind == StartupFailureKind.InputOptionCompatibility)
|
|
{
|
|
if (runtime.HasRetriedWithCompatibilityProfile)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Initial ffmpeg input profile was rejected. Retrying once with minimal compatibility options.",
|
|
runtime.LastStartupFailureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
currentTask.Id);
|
|
|
|
var retryStream = new StreamUrlResult(
|
|
runtime.SelectedQuality,
|
|
runtime.SelectedProtocol,
|
|
runtime.StreamUrl,
|
|
runtime.InputHeaders,
|
|
Array.Empty<StreamQualityOption>(),
|
|
runtime.SelectedVideoCodec);
|
|
|
|
session.MarkStarting(retryStream.SelectedUrl, session.OutputPathPattern ?? runtime.OutputPathPattern, observedAt);
|
|
session.ActivateSegment(Math.Max(1, runtime.CurrentSegmentIndex), observedAt);
|
|
currentTask.MarkStarting(retryStream.SelectedUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, observedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
await StartInternalAsync(
|
|
session,
|
|
currentTask,
|
|
retryStream,
|
|
runtime.RecordingSettings,
|
|
FfmpegInputOptionProfile.Minimal,
|
|
hasRetriedWithCompatibilityProfile: true,
|
|
hasRetriedWithRefreshedStream: runtime.HasRetriedWithRefreshedStream,
|
|
runtime.RetryAttemptCount + 1);
|
|
|
|
var restartedAt = DateTimeOffset.UtcNow;
|
|
session.MarkRunning(restartedAt);
|
|
currentTask.MarkRunning(restartedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"FFmpeg",
|
|
"ffmpeg startup retry succeeded with minimal compatibility options.",
|
|
liveRoomId: session.LiveRoomId,
|
|
recordSessionId: session.Id,
|
|
recordTaskId: currentTask.Id);
|
|
|
|
return true;
|
|
}
|
|
|
|
if (runtime.StartupFailureKind != StartupFailureKind.StreamHandshake || runtime.HasRetriedWithRefreshedStream)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
|
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
|
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
|
|
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId);
|
|
await liveRoomStatusService.ApplySnapshotAsync(session.LiveRoom, liveStatus, observedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
if (!liveStatus.IsLive)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Initial ffmpeg input handshake failed. Refreshing stream URL and retrying once.",
|
|
runtime.LastStartupFailureLine,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
currentTask.Id);
|
|
|
|
var refreshedStream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality);
|
|
session.MarkStarting(refreshedStream.SelectedUrl, session.OutputPathPattern ?? runtime.OutputPathPattern, observedAt);
|
|
session.ActivateSegment(Math.Max(1, runtime.CurrentSegmentIndex), observedAt);
|
|
currentTask.MarkStarting(refreshedStream.SelectedUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, observedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
await StartInternalAsync(
|
|
session,
|
|
currentTask,
|
|
refreshedStream,
|
|
runtime.RecordingSettings,
|
|
runtime.InputOptionProfile,
|
|
hasRetriedWithCompatibilityProfile: runtime.HasRetriedWithCompatibilityProfile,
|
|
hasRetriedWithRefreshedStream: true,
|
|
runtime.RetryAttemptCount + 1);
|
|
|
|
var refreshedRetryStartedAt = DateTimeOffset.UtcNow;
|
|
session.MarkRunning(refreshedRetryStartedAt);
|
|
currentTask.MarkRunning(refreshedRetryStartedAt);
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"FFmpeg",
|
|
"ffmpeg startup retry succeeded with a refreshed stream URL.",
|
|
liveRoomId: session.LiveRoomId,
|
|
recordSessionId: session.Id,
|
|
recordTaskId: currentTask.Id);
|
|
|
|
return true;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Recover ffmpeg startup failure failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private async Task FinalizeExitedSessionAsync(
|
|
SessionProcessRuntime runtime,
|
|
Process process,
|
|
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var emailNotificationService = scope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
|
var webhookNotificationService = scope.ServiceProvider.GetRequiredService<IWebhookNotificationService>();
|
|
var settings = await settingsService.GetAsync();
|
|
|
|
var session = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
|
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var currentTask = session.RecordTasks.FirstOrDefault(item => item.Id == runtime.CurrentTaskId)
|
|
?? session.RecordTasks.OrderByDescending(static item => item.SegmentIndex).FirstOrDefault();
|
|
if (currentTask is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var endedAt = DateTimeOffset.UtcNow;
|
|
string? finalizationError = null;
|
|
var effectiveOutputPath = currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath;
|
|
var durationSeconds = currentTask.StartedAt.HasValue
|
|
? (double?)Math.Max(0, (endedAt - currentTask.StartedAt.Value).TotalSeconds)
|
|
: null;
|
|
|
|
if (session.OutputFormat == RecordOutputFormat.Mp4)
|
|
{
|
|
var recorderOutputPath = NormalizeAbsolutePath(GetRecorderOutputPath(
|
|
currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath,
|
|
session.OutputFormat,
|
|
session.SaveMode));
|
|
|
|
if (runtime.StopRequested && runtime.WasForceKilled)
|
|
{
|
|
effectiveOutputPath = File.Exists(recorderOutputPath)
|
|
? recorderOutputPath
|
|
: effectiveOutputPath;
|
|
finalizationError = "The active segment did not close gracefully before the recorder process was terminated. The intermediate TS file was kept without MP4 finalization.";
|
|
}
|
|
else
|
|
{
|
|
var recorderSegmentPaths = runtime.CaptureCurrentRecorderSegments();
|
|
await WaitForFileToStabilizeAsync(recorderOutputPath, CancellationToken.None);
|
|
var finalizationResult = await TryFinalizeTaskOutputAsync(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
session,
|
|
currentTask,
|
|
durationSeconds,
|
|
recorderSegmentPaths);
|
|
effectiveOutputPath = finalizationResult.OutputPath;
|
|
finalizationError = finalizationResult.ErrorMessage;
|
|
}
|
|
}
|
|
|
|
var fileSize = CalculateFileSize(effectiveOutputPath);
|
|
var hasUsableOutput = HasUsableOutput(effectiveOutputPath, fileSize);
|
|
var toleratedNonZeroExit = false;
|
|
var danmakuPath = activeDanmakuSummary?.FilePath ?? currentTask.Result?.DanmakuFilePath ?? GuessDanmakuPath(effectiveOutputPath);
|
|
var danmakuMessageCount = activeDanmakuSummary?.MessageCount ?? CountDanmakuMessages(danmakuPath);
|
|
|
|
if (IsLowStoragePauseError(finalizationError))
|
|
{
|
|
currentTask.MarkProcessing(finalizationError, endedAt);
|
|
session.MarkStopped(endedAt, finalizationError);
|
|
}
|
|
else if (runtime.StopRequested)
|
|
{
|
|
currentTask.MarkStopped(endedAt, durationSeconds, finalizationError);
|
|
session.MarkStopped(endedAt, finalizationError);
|
|
}
|
|
else if (!string.IsNullOrWhiteSpace(finalizationError))
|
|
{
|
|
currentTask.MarkFailed(finalizationError, endedAt);
|
|
session.MarkFailed(finalizationError, endedAt);
|
|
}
|
|
else if ((process.ExitCode == 0 || runtime.CompletionRequested && hasUsableOutput) &&
|
|
(runtime.CompletionRequested || !runtime.StopRequested))
|
|
{
|
|
toleratedNonZeroExit = process.ExitCode != 0;
|
|
currentTask.MarkCompleted(endedAt, durationSeconds);
|
|
session.MarkCompleted(endedAt);
|
|
}
|
|
else
|
|
{
|
|
var errorMessage = $"ffmpeg exit code: {process.ExitCode}";
|
|
currentTask.MarkFailed(errorMessage, endedAt);
|
|
session.MarkFailed(errorMessage, endedAt);
|
|
}
|
|
|
|
currentTask.DetachProcess(endedAt);
|
|
session.SyncSegmentCount(session.RecordTasks.Count, endedAt);
|
|
await UpsertRecordResultAsync(currentTask, dbContext, effectiveOutputPath, fileSize, durationSeconds, danmakuPath, danmakuMessageCount, endedAt);
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
ClearPostProcessState(currentTask.Id);
|
|
|
|
if (currentTask.Status == RecordTaskStatus.Completed)
|
|
{
|
|
var recordResult = await LoadRecordResultAsync(dbContext, currentTask.Id);
|
|
var eventScriptService = scope.ServiceProvider.GetRequiredService<IEventScriptService>();
|
|
await eventScriptService.RunSegmentCompletedAsync(
|
|
session.LiveRoom,
|
|
session,
|
|
currentTask,
|
|
recordResult,
|
|
effectiveOutputPath,
|
|
endedAt);
|
|
}
|
|
|
|
await logService.WriteAsync(
|
|
session.Status == RecordSessionStatus.Completed ? SystemLogLevel.Info :
|
|
session.Status == RecordSessionStatus.Stopped ? SystemLogLevel.Warning :
|
|
SystemLogLevel.Error,
|
|
"FFmpeg",
|
|
$"Recording session exited with status={session.Status}.",
|
|
toleratedNonZeroExit
|
|
? $"exitCode={process.ExitCode}; output={effectiveOutputPath}; recorderOutput={runtime.RecorderOutputPath}; non-zero exit was tolerated because completion was requested and the output is usable"
|
|
: $"exitCode={process.ExitCode}; output={effectiveOutputPath}; recorderOutput={runtime.RecorderOutputPath}",
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
currentTask.Id);
|
|
|
|
if (session.Status == RecordSessionStatus.Failed && session.LiveRoom is not null)
|
|
{
|
|
await emailNotificationService.SendExceptionAsync(
|
|
"FFmpeg",
|
|
"Recording session exited abnormally.",
|
|
$"exitCode={process.ExitCode}; output={effectiveOutputPath}",
|
|
session.LiveRoom,
|
|
currentTask);
|
|
await webhookNotificationService.SendExceptionAsync(
|
|
"FFmpeg",
|
|
"Recording session exited abnormally.",
|
|
$"exitCode={process.ExitCode}; output={effectiveOutputPath}",
|
|
session.LiveRoom,
|
|
currentTask);
|
|
}
|
|
}
|
|
|
|
private async Task FinalizeCompletedSegmentAsync(
|
|
Guid recordSessionId,
|
|
Guid recordTaskId,
|
|
double? durationSeconds,
|
|
DateTimeOffset endedAt,
|
|
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? danmakuSummary,
|
|
IReadOnlyList<string>? recorderSegmentPaths)
|
|
{
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
var settings = await settingsService.GetAsync();
|
|
|
|
var session = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordSessionId);
|
|
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var recordTask = session.RecordTasks.FirstOrDefault(item => item.Id == recordTaskId);
|
|
if (recordTask is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var effectiveOutputPath = recordTask.OutputFilePath;
|
|
if (string.IsNullOrWhiteSpace(effectiveOutputPath))
|
|
{
|
|
if (string.IsNullOrWhiteSpace(session.OutputPathPattern))
|
|
{
|
|
return;
|
|
}
|
|
|
|
effectiveOutputPath = NormalizeAbsolutePath(ResolveSegmentOutputPath(
|
|
session.OutputPathPattern,
|
|
session.SaveMode,
|
|
recordTask.SegmentIndex));
|
|
}
|
|
|
|
await WaitForFileToStabilizeAsync(
|
|
GetRecorderOutputPath(effectiveOutputPath, session.OutputFormat, session.SaveMode),
|
|
CancellationToken.None);
|
|
|
|
var finalizationResult = await TryFinalizeTaskOutputAsync(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
session,
|
|
recordTask,
|
|
durationSeconds,
|
|
recorderSegmentPaths);
|
|
effectiveOutputPath = finalizationResult.OutputPath;
|
|
var finalizationError = finalizationResult.ErrorMessage;
|
|
var fileSize = CalculateFileSize(effectiveOutputPath);
|
|
var danmakuPath = danmakuSummary?.FilePath ?? recordTask.Result?.DanmakuFilePath ?? GuessDanmakuPath(recordTask.OutputFilePath);
|
|
var danmakuMessageCount = danmakuSummary?.MessageCount ?? recordTask.Result?.DanmakuMessageCount ?? CountDanmakuMessages(danmakuPath);
|
|
|
|
if (IsLowStoragePauseError(finalizationError))
|
|
{
|
|
recordTask.MarkProcessing(finalizationError, endedAt);
|
|
}
|
|
else if (string.IsNullOrWhiteSpace(finalizationError))
|
|
{
|
|
recordTask.MarkCompleted(endedAt, durationSeconds);
|
|
}
|
|
else
|
|
{
|
|
recordTask.MarkFailed(finalizationError, endedAt);
|
|
}
|
|
|
|
await UpsertRecordResultAsync(
|
|
recordTask,
|
|
dbContext,
|
|
effectiveOutputPath,
|
|
fileSize,
|
|
durationSeconds,
|
|
danmakuPath,
|
|
danmakuMessageCount,
|
|
endedAt);
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
if (recordTask.Status == RecordTaskStatus.Completed)
|
|
{
|
|
var recordResult = await LoadRecordResultAsync(dbContext, recordTask.Id);
|
|
var eventScriptService = scope.ServiceProvider.GetRequiredService<IEventScriptService>();
|
|
await eventScriptService.RunSegmentCompletedAsync(
|
|
session.LiveRoom,
|
|
session,
|
|
recordTask,
|
|
recordResult,
|
|
effectiveOutputPath,
|
|
endedAt);
|
|
}
|
|
|
|
if (!string.IsNullOrWhiteSpace(finalizationError))
|
|
{
|
|
await logService.WriteAsync(
|
|
IsLowStoragePauseError(finalizationError) ? SystemLogLevel.Warning : SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
IsLowStoragePauseError(finalizationError)
|
|
? "Segment MP4 finalization paused because storage is below threshold."
|
|
: "Segment MP4 finalization failed after rollover.",
|
|
finalizationError,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
recordTask.Id);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Finalize completed segment failed for session {RecordSessionId}, task {RecordTaskId}", recordSessionId, recordTaskId);
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
var session = await dbContext.RecordSessions
|
|
.AsNoTracking()
|
|
.FirstOrDefaultAsync(item => item.Id == recordSessionId);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Error,
|
|
"FFmpeg",
|
|
"Background segment finalization failed.",
|
|
ex.ToString(),
|
|
session?.LiveRoomId,
|
|
recordSessionId,
|
|
recordTaskId);
|
|
}
|
|
catch (Exception logEx)
|
|
{
|
|
_logger.LogWarning(logEx, "Persist segment finalization failure log failed for session {RecordSessionId}, task {RecordTaskId}", recordSessionId, recordTaskId);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
}
|
|
}
|
|
|
|
private async Task PersistProcessBindingAsync(Guid recordSessionId, Guid recordTaskId, int processId, CancellationToken cancellationToken)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var session = await dbContext.RecordSessions.FirstOrDefaultAsync(item => item.Id == recordSessionId, cancellationToken);
|
|
var task = await dbContext.RecordTasks.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
|
|
|
|
var now = DateTimeOffset.UtcNow;
|
|
session?.AttachProcess(processId, now);
|
|
task?.AttachProcess(processId, now);
|
|
if (session is not null || task is not null)
|
|
{
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
}
|
|
|
|
private async Task StartDanmakuAsync(SessionProcessRuntime runtime, RecordTask initialTask, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
LiveRoom? liveRoom = null;
|
|
var includeNonChatEvents = true;
|
|
var firstWrittenEventLogged = 0;
|
|
|
|
using (var readScope = _serviceScopeFactory.CreateScope())
|
|
{
|
|
var dbContext = readScope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var adapterFactory = readScope.ServiceProvider.GetRequiredService<ILiveDanmakuAdapterFactory>();
|
|
|
|
if (!runtime.RecordingSettings.EnableDanmakuRecording)
|
|
{
|
|
return;
|
|
}
|
|
|
|
liveRoom = await dbContext.LiveRooms
|
|
.AsNoTracking()
|
|
.FirstOrDefaultAsync(item => item.Id == runtime.LiveRoomId, cancellationToken);
|
|
if (liveRoom is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
includeNonChatEvents = runtime.RecordingSettings.DanmakuIncludeNonChatEvents;
|
|
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
|
|
runtime.DanmakuConnection = await adapter.ConnectAsync(
|
|
new DanmakuConnectionContext(
|
|
liveRoom.Id,
|
|
runtime.RecordSessionId,
|
|
liveRoom.Platform,
|
|
liveRoom.RoomId,
|
|
liveRoom.AnchorName,
|
|
liveRoom.Title,
|
|
liveRoom.SourceUrl,
|
|
runtime.RecordingSettings.DanmakuMinPollIntervalMilliseconds,
|
|
runtime.RecordingSettings.DanmakuRetryDelayMaxSeconds),
|
|
cancellationToken);
|
|
}
|
|
|
|
runtime.DanmakuRecorder = new SessionDanmakuXmlRecorder(
|
|
liveRoom.Platform,
|
|
liveRoom.Id,
|
|
runtime.RecordSessionId,
|
|
liveRoom.RoomId);
|
|
await runtime.DanmakuRecorder.StartSegmentAsync(
|
|
initialTask.Id,
|
|
initialTask.SegmentIndex,
|
|
NormalizeAbsolutePath(initialTask.OutputFilePath ?? runtime.CurrentOutputFilePath),
|
|
initialTask.StartedAt ?? DateTimeOffset.UtcNow);
|
|
runtime.HasInitializedDanmaku = true;
|
|
|
|
runtime.DanmakuPumpTask = Task.Run(
|
|
async () =>
|
|
{
|
|
try
|
|
{
|
|
await runtime.DanmakuConnection.StartAsync(
|
|
async danmakuEvent =>
|
|
{
|
|
if (runtime.DanmakuRecorder is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (!includeNonChatEvents &&
|
|
!string.Equals(danmakuEvent.Type, "chat", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return;
|
|
}
|
|
|
|
await runtime.DanmakuRecorder.AppendAsync(danmakuEvent);
|
|
if (Interlocked.Exchange(ref firstWrittenEventLogged, 1) == 0)
|
|
{
|
|
await WriteDanmakuSystemLogAsync(
|
|
SystemLogLevel.Info,
|
|
"First danmaku event was written to XML.",
|
|
$"type={danmakuEvent.Type}; user={TruncateDanmakuPreview(danmakuEvent.User, 80)}; content={TruncateDanmakuPreview(danmakuEvent.Content, 160)}",
|
|
runtime.LiveRoomId,
|
|
runtime.RecordSessionId,
|
|
initialTask.Id,
|
|
runtime.DanmakuCancellation.Token);
|
|
}
|
|
},
|
|
runtime.DanmakuCancellation.Token);
|
|
}
|
|
catch (OperationCanceledException) when (runtime.DanmakuCancellation.IsCancellationRequested)
|
|
{
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await WriteDanmakuSystemLogAsync(
|
|
SystemLogLevel.Warning,
|
|
"Danmaku pump stopped unexpectedly.",
|
|
ex.ToString(),
|
|
runtime.LiveRoomId,
|
|
runtime.RecordSessionId,
|
|
runtime.CurrentTaskId,
|
|
CancellationToken.None);
|
|
throw;
|
|
}
|
|
},
|
|
runtime.DanmakuCancellation.Token);
|
|
|
|
await WriteDanmakuSystemLogAsync(
|
|
SystemLogLevel.Info,
|
|
"Danmaku capture started for the recording session.",
|
|
null,
|
|
liveRoom.Id,
|
|
runtime.RecordSessionId,
|
|
initialTask.Id,
|
|
cancellationToken);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Start danmaku capture failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
|
|
try
|
|
{
|
|
await WriteDanmakuSystemLogAsync(
|
|
SystemLogLevel.Warning,
|
|
"Danmaku capture startup failed. Recording will continue without live comments until retry succeeds.",
|
|
ex.ToString(),
|
|
runtime.LiveRoomId,
|
|
runtime.RecordSessionId,
|
|
runtime.CurrentTaskId,
|
|
cancellationToken);
|
|
}
|
|
catch (Exception logEx)
|
|
{
|
|
_logger.LogWarning(logEx, "Persist danmaku startup failure log failed for session {RecordSessionId}", runtime.RecordSessionId);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task WriteDanmakuSystemLogAsync(
|
|
SystemLogLevel level,
|
|
string message,
|
|
string? detail,
|
|
Guid? liveRoomId,
|
|
Guid? recordSessionId,
|
|
Guid? recordTaskId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
level,
|
|
"Danmaku",
|
|
message,
|
|
detail,
|
|
liveRoomId,
|
|
recordSessionId,
|
|
recordTaskId,
|
|
cancellationToken);
|
|
}
|
|
|
|
private static string TruncateDanmakuPreview(string? value, int maxLength)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(value))
|
|
{
|
|
return string.Empty;
|
|
}
|
|
|
|
var normalized = value.Trim().Replace(Environment.NewLine, " ", StringComparison.Ordinal);
|
|
return normalized.Length <= maxLength ? normalized : normalized[..maxLength];
|
|
}
|
|
|
|
private static bool IsRetryableStartupFailureLine(string line) =>
|
|
line.Contains("Error reading HTTP response", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("unexpected EOF", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Connection reset", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Connection refused", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("I/O error", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Invalid data found when processing input", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Server returned 4", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Server returned 5", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("HTTP error 4", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("HTTP error 5", StringComparison.OrdinalIgnoreCase);
|
|
|
|
private static bool IsOptionCompatibilityFailureLine(string line) =>
|
|
line.Contains("Option not found", StringComparison.OrdinalIgnoreCase) ||
|
|
line.Contains("Unrecognized option", StringComparison.OrdinalIgnoreCase);
|
|
|
|
private static bool ShouldTreatAsUnexpectedFragmentRollover(
|
|
SessionProcessRuntime runtime,
|
|
double? previousDurationSeconds)
|
|
{
|
|
if (!runtime.HasOpenedFirstSegment || !previousDurationSeconds.HasValue)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var expectedSegmentDurationSeconds = Math.Max(60, runtime.RecordingSettings.SegmentDurationMinutes * 60);
|
|
var rolloverToleranceSeconds = Math.Min(30, Math.Max(5, expectedSegmentDurationSeconds * 0.1));
|
|
return previousDurationSeconds.Value + rolloverToleranceSeconds < expectedSegmentDurationSeconds;
|
|
}
|
|
|
|
private async Task EnsureDanmakuSegmentAsync(SessionProcessRuntime runtime, RecordTask recordTask, DateTimeOffset startedAt)
|
|
{
|
|
if (runtime.DanmakuRecorder is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await runtime.DanmakuRecorder.StartSegmentAsync(
|
|
recordTask.Id,
|
|
recordTask.SegmentIndex,
|
|
NormalizeAbsolutePath(recordTask.OutputFilePath ?? runtime.CurrentOutputFilePath),
|
|
startedAt);
|
|
}
|
|
|
|
private async Task RunManualFinalizeTaskAsync(Guid recordTaskId)
|
|
{
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
|
|
var recordTask = await dbContext.RecordTasks
|
|
.Include(item => item.RecordSession)
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordTaskId);
|
|
|
|
if (recordTask?.RecordSession is null)
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
return;
|
|
}
|
|
|
|
var session = recordTask.RecordSession;
|
|
var finalOutputPath = recordTask.OutputFilePath;
|
|
if (string.IsNullOrWhiteSpace(finalOutputPath))
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
return;
|
|
}
|
|
|
|
var recorderOutputPath = ResolveManualFinalizeSourcePath(recordTask, session);
|
|
if (!File.Exists(recorderOutputPath))
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Manual MP4 finalization was skipped because the intermediate file was missing.",
|
|
recorderOutputPath,
|
|
recordTask.LiveRoomId,
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id);
|
|
return;
|
|
}
|
|
|
|
await WaitForFileToStabilizeAsync(recorderOutputPath, CancellationToken.None);
|
|
|
|
var settings = await settingsService.GetAsync();
|
|
var endedAt = recordTask.EndedAt ?? DateTimeOffset.UtcNow;
|
|
var durationSeconds = recordTask.DurationSeconds ?? recordTask.Result?.DurationSeconds;
|
|
var finalizationResult = await TryFinalizeTaskOutputAsync(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
session,
|
|
recordTask,
|
|
durationSeconds,
|
|
[recorderOutputPath]);
|
|
|
|
var effectiveOutputPath = finalizationResult.OutputPath;
|
|
var finalizationError = finalizationResult.ErrorMessage;
|
|
var fileSize = CalculateFileSize(effectiveOutputPath);
|
|
var danmakuPath = recordTask.Result?.DanmakuFilePath ?? GuessDanmakuPath(recordTask.OutputFilePath);
|
|
var danmakuMessageCount = recordTask.Result?.DanmakuMessageCount ?? CountDanmakuMessages(danmakuPath);
|
|
|
|
if (IsLowStoragePauseError(finalizationError))
|
|
{
|
|
recordTask.MarkProcessing(finalizationError, endedAt);
|
|
}
|
|
else if (string.IsNullOrWhiteSpace(finalizationError))
|
|
{
|
|
if (recordTask.Status == RecordTaskStatus.Stopped)
|
|
{
|
|
recordTask.MarkStopped(endedAt, durationSeconds);
|
|
}
|
|
else
|
|
{
|
|
recordTask.MarkCompleted(endedAt, durationSeconds);
|
|
}
|
|
}
|
|
else if (recordTask.Status == RecordTaskStatus.Stopped)
|
|
{
|
|
recordTask.MarkStopped(endedAt, durationSeconds, finalizationError);
|
|
}
|
|
else
|
|
{
|
|
recordTask.MarkFailed(finalizationError, endedAt);
|
|
}
|
|
|
|
await UpsertRecordResultAsync(
|
|
recordTask,
|
|
dbContext,
|
|
effectiveOutputPath,
|
|
fileSize,
|
|
durationSeconds,
|
|
danmakuPath,
|
|
danmakuMessageCount,
|
|
endedAt);
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
|
|
if (recordTask.Status == RecordTaskStatus.Completed)
|
|
{
|
|
var recordResult = await LoadRecordResultAsync(dbContext, recordTask.Id);
|
|
var eventScriptService = scope.ServiceProvider.GetRequiredService<IEventScriptService>();
|
|
await eventScriptService.RunSegmentCompletedAsync(
|
|
recordTask.LiveRoom,
|
|
session,
|
|
recordTask,
|
|
recordResult,
|
|
effectiveOutputPath,
|
|
endedAt);
|
|
}
|
|
|
|
await logService.WriteAsync(
|
|
string.IsNullOrWhiteSpace(finalizationError) ? SystemLogLevel.Info : SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
string.IsNullOrWhiteSpace(finalizationError)
|
|
? "Manual MP4 finalization completed."
|
|
: "Manual MP4 finalization finished with warnings.",
|
|
string.IsNullOrWhiteSpace(finalizationError)
|
|
? effectiveOutputPath
|
|
: $"{effectiveOutputPath}; {finalizationError}",
|
|
recordTask.LiveRoomId,
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Manual MP4 finalization failed for task {RecordTaskId}", recordTaskId);
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Manual MP4 finalization failed to start or complete.",
|
|
ex.ToString(),
|
|
recordTaskId: recordTaskId);
|
|
}
|
|
catch (Exception logEx)
|
|
{
|
|
_logger.LogWarning(logEx, "Persist manual MP4 finalization failure log failed for task {RecordTaskId}", recordTaskId);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
}
|
|
}
|
|
|
|
private sealed class SessionProcessRuntime : IDisposable
|
|
{
|
|
public SessionProcessRuntime(
|
|
Guid recordSessionId,
|
|
Guid liveRoomId,
|
|
string streamUrl,
|
|
string outputPathPattern,
|
|
string recorderOutputPath,
|
|
RecordOutputFormat outputFormat,
|
|
RecordSaveMode saveMode,
|
|
Guid currentTaskId,
|
|
int currentSegmentIndex,
|
|
string currentOutputFilePath,
|
|
string selectedQuality,
|
|
string selectedProtocol,
|
|
string? selectedVideoCodec,
|
|
StreamInputHeaders? inputHeaders,
|
|
RecordingExecutionSettings recordingSettings,
|
|
FfmpegInputOptionProfile inputOptionProfile,
|
|
bool hasRetriedWithCompatibilityProfile,
|
|
bool hasRetriedWithRefreshedStream,
|
|
int retryAttemptCount)
|
|
{
|
|
RecordSessionId = recordSessionId;
|
|
LiveRoomId = liveRoomId;
|
|
StreamUrl = streamUrl;
|
|
OutputPathPattern = outputPathPattern;
|
|
RecorderOutputPath = recorderOutputPath;
|
|
OutputFormat = outputFormat;
|
|
SaveMode = saveMode;
|
|
CurrentTaskId = currentTaskId;
|
|
CurrentSegmentIndex = currentSegmentIndex;
|
|
CurrentOutputFilePath = currentOutputFilePath;
|
|
SelectedQuality = selectedQuality;
|
|
SelectedProtocol = selectedProtocol;
|
|
SelectedVideoCodec = selectedVideoCodec;
|
|
InputHeaders = inputHeaders;
|
|
RecordingSettings = recordingSettings;
|
|
InputOptionProfile = inputOptionProfile;
|
|
HasRetriedWithCompatibilityProfile = hasRetriedWithCompatibilityProfile;
|
|
HasRetriedWithRefreshedStream = hasRetriedWithRefreshedStream;
|
|
RetryAttemptCount = Math.Max(0, retryAttemptCount);
|
|
}
|
|
|
|
public Guid RecordSessionId { get; }
|
|
public Guid LiveRoomId { get; }
|
|
public string StreamUrl { get; }
|
|
public string OutputPathPattern { get; }
|
|
public string RecorderOutputPath { get; }
|
|
public RecordOutputFormat OutputFormat { get; }
|
|
public RecordSaveMode SaveMode { get; }
|
|
public Guid CurrentTaskId { get; set; }
|
|
public int CurrentSegmentIndex { get; set; }
|
|
public string CurrentOutputFilePath { get; set; }
|
|
public string SelectedQuality { get; }
|
|
public string SelectedProtocol { get; }
|
|
public string? SelectedVideoCodec { get; }
|
|
public StreamInputHeaders? InputHeaders { get; }
|
|
public RecordingExecutionSettings RecordingSettings { get; }
|
|
public FfmpegInputOptionProfile InputOptionProfile { get; }
|
|
public bool HasRetriedWithCompatibilityProfile { get; }
|
|
public bool HasRetriedWithRefreshedStream { get; }
|
|
public Process? Process { get; private set; }
|
|
public int ProcessId => Process?.Id ?? 0;
|
|
public bool CompletionRequested { get; private set; }
|
|
public bool StopRequested { get; private set; }
|
|
public bool WasForceKilled { get; private set; }
|
|
public bool HasInitializedDanmaku { get; set; }
|
|
public bool HasOpenedFirstSegment { get; set; }
|
|
public int RetryAttemptCount { get; }
|
|
public StartupFailureKind StartupFailureKind { get; private set; }
|
|
public string? LastStartupFailureLine { get; private set; }
|
|
public SemaphoreSlim Gate { get; } = new(1, 1);
|
|
public CancellationTokenSource DanmakuCancellation { get; } = new();
|
|
public TaskCompletionSource<bool> ExitCompletion { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
public SessionDanmakuXmlRecorder? DanmakuRecorder { get; set; }
|
|
public ILiveDanmakuConnection? DanmakuConnection { get; set; }
|
|
public Task? DanmakuPumpTask { get; set; }
|
|
private List<string> CurrentRecorderSegmentPaths { get; } = [];
|
|
private object RuntimeSourceFailureSync { get; } = new();
|
|
private DateTimeOffset RuntimeSourceFailureWindowStartedAt { get; set; }
|
|
private int RuntimeSourceFailureCount { get; set; }
|
|
private bool RuntimeSourceFailureVerificationInProgress { get; set; }
|
|
|
|
public void AttachProcess(Process process) => Process = process;
|
|
|
|
public void MarkStopRequested(bool markAsCompletedOnExit)
|
|
{
|
|
if (markAsCompletedOnExit)
|
|
{
|
|
CompletionRequested = true;
|
|
StopRequested = false;
|
|
}
|
|
else
|
|
{
|
|
CompletionRequested = false;
|
|
StopRequested = true;
|
|
}
|
|
}
|
|
|
|
public void MarkStartupFailure(StartupFailureKind kind, string line)
|
|
{
|
|
if (StartupFailureKind == StartupFailureKind.InputOptionCompatibility &&
|
|
kind != StartupFailureKind.InputOptionCompatibility)
|
|
{
|
|
return;
|
|
}
|
|
|
|
StartupFailureKind = kind;
|
|
LastStartupFailureLine = line;
|
|
}
|
|
|
|
public void MarkForceKilled() => WasForceKilled = true;
|
|
|
|
public void ResetCurrentRecorderSegmentPaths(string openedPath)
|
|
{
|
|
CurrentRecorderSegmentPaths.Clear();
|
|
CurrentRecorderSegmentPaths.Add(openedPath);
|
|
}
|
|
|
|
public void TrackCurrentRecorderSegment(string openedPath)
|
|
{
|
|
if (CurrentRecorderSegmentPaths.Count == 0 ||
|
|
!string.Equals(CurrentRecorderSegmentPaths[^1], openedPath, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
CurrentRecorderSegmentPaths.Add(openedPath);
|
|
}
|
|
}
|
|
|
|
public IReadOnlyList<string> CaptureCurrentRecorderSegments()
|
|
{
|
|
if (CurrentRecorderSegmentPaths.Count == 0)
|
|
{
|
|
return [CurrentOutputFilePath];
|
|
}
|
|
|
|
return CurrentRecorderSegmentPaths.ToArray();
|
|
}
|
|
|
|
public bool RegisterRuntimeSourceFailure(DateTimeOffset observedAt, TimeSpan window)
|
|
{
|
|
lock (RuntimeSourceFailureSync)
|
|
{
|
|
if (RuntimeSourceFailureVerificationInProgress)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (RuntimeSourceFailureWindowStartedAt == default ||
|
|
observedAt - RuntimeSourceFailureWindowStartedAt > window)
|
|
{
|
|
RuntimeSourceFailureWindowStartedAt = observedAt;
|
|
RuntimeSourceFailureCount = 0;
|
|
}
|
|
|
|
RuntimeSourceFailureCount++;
|
|
if (RuntimeSourceFailureCount < 3)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
RuntimeSourceFailureVerificationInProgress = true;
|
|
RuntimeSourceFailureWindowStartedAt = observedAt;
|
|
RuntimeSourceFailureCount = 0;
|
|
return true;
|
|
}
|
|
}
|
|
|
|
public void CompleteRuntimeSourceFailureVerification()
|
|
{
|
|
lock (RuntimeSourceFailureSync)
|
|
{
|
|
RuntimeSourceFailureVerificationInProgress = false;
|
|
RuntimeSourceFailureWindowStartedAt = DateTimeOffset.UtcNow;
|
|
RuntimeSourceFailureCount = 0;
|
|
}
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
DanmakuCancellation.Dispose();
|
|
Gate.Dispose();
|
|
}
|
|
}
|
|
}
|