1901 lines
76 KiB
C#
1901 lines
76 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Diagnostics;
|
|
using System.Text.RegularExpressions;
|
|
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.Abstractions.Storage;
|
|
using LiveRecorder.Application.Models.Media;
|
|
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 : IFfmpegService
|
|
{
|
|
private const string ArtifactRepairMarker = "[artifact-repair]";
|
|
private const string StalePendingStartupFailure =
|
|
"Recording startup was interrupted before stream initialization.";
|
|
private static readonly TimeSpan StalePendingThreshold = TimeSpan.FromMinutes(10);
|
|
private static readonly Regex SegmentOpeningRegex = new(
|
|
"""Opening '([^']+)' for writing""",
|
|
RegexOptions.Compiled | RegexOptions.CultureInvariant | RegexOptions.IgnoreCase);
|
|
|
|
private static readonly TimeSpan StartupFailureNotificationCooldown = TimeSpan.FromMinutes(30);
|
|
private static readonly TimeSpan StartupFailureMaxBackoff = TimeSpan.FromMinutes(15);
|
|
private static readonly TimeSpan StartupFailureBaseBackoff = TimeSpan.FromSeconds(30);
|
|
private static readonly TimeSpan RuntimeFailureMaxBackoff = TimeSpan.FromMinutes(5);
|
|
private static readonly TimeSpan RuntimeFailureBaseBackoff = TimeSpan.FromSeconds(15);
|
|
|
|
private readonly ConcurrentDictionary<Guid, SessionProcessRuntime> _processes = new();
|
|
private readonly ConcurrentDictionary<Guid, SessionTransitionRuntime> _sessionTransitions = new();
|
|
private readonly ConcurrentDictionary<Guid, PostProcessRuntimeEntry> _postProcessStates = new();
|
|
private readonly ConcurrentDictionary<Guid, Process> _postProcessProcesses = new();
|
|
private readonly ConcurrentDictionary<Guid, RoomStartupFailureState> _roomStartupFailureStates = new();
|
|
private readonly ConcurrentDictionary<Guid, RoomStartupFailureState> _roomRuntimeFailureStates = new();
|
|
private readonly SemaphoreSlim _recoveryEncoderProbeGate = new(1, 1);
|
|
private readonly object _transcodeConcurrencyLock = new();
|
|
private readonly SemaphoreSlim _orphanRecoveryGate = new(1, 1);
|
|
private readonly Queue<TaskCompletionSource<IDisposable>> _transcodeWaiters = new();
|
|
private readonly CancellationTokenSource _shutdownCts = new();
|
|
private int _activeTranscodeTasks;
|
|
private int _maxConcurrentTranscodeTasks = 1;
|
|
private bool _hasProbedRecoveryVideoEncoder;
|
|
private RecoveryVideoEncoderSelection _cachedRecoveryVideoEncoder = RecoveryVideoEncoderSelection.Software;
|
|
private readonly IServiceScopeFactory _serviceScopeFactory;
|
|
private readonly IStorageGuardService _storageGuardService;
|
|
private readonly ILiveRoomPollingSignal _liveRoomPollingSignal;
|
|
private readonly ILogger<FfmpegService> _logger;
|
|
|
|
public FfmpegService(
|
|
IServiceScopeFactory serviceScopeFactory,
|
|
IStorageGuardService storageGuardService,
|
|
ILiveRoomPollingSignal liveRoomPollingSignal,
|
|
ILogger<FfmpegService> logger)
|
|
{
|
|
_serviceScopeFactory = serviceScopeFactory;
|
|
_storageGuardService = storageGuardService;
|
|
_liveRoomPollingSignal = liveRoomPollingSignal;
|
|
_logger = logger;
|
|
}
|
|
|
|
public bool IsRunning(Guid recordSessionId) =>
|
|
IsSessionRuntimeActive(
|
|
_processes.ContainsKey(recordSessionId),
|
|
_sessionTransitions.ContainsKey(recordSessionId));
|
|
|
|
internal static bool IsSessionRuntimeActive(bool hasProcess, bool hasTransition) =>
|
|
hasProcess || hasTransition;
|
|
|
|
internal static bool ShouldReuseRecoveryTask(bool hasMedia, RecordTaskStatus status) =>
|
|
!hasMedia && status is RecordTaskStatus.Pending or RecordTaskStatus.Starting or RecordTaskStatus.Running or RecordTaskStatus.Stopping;
|
|
|
|
public IReadOnlyDictionary<Guid, RecordTaskRuntimeState> GetTaskRuntimeStates(IReadOnlyCollection<Guid> recordTaskIds)
|
|
{
|
|
if (recordTaskIds.Count == 0)
|
|
{
|
|
return new Dictionary<Guid, RecordTaskRuntimeState>();
|
|
}
|
|
|
|
var snapshot = new Dictionary<Guid, RecordTaskRuntimeState>();
|
|
foreach (var taskId in recordTaskIds)
|
|
{
|
|
if (_postProcessStates.TryGetValue(taskId, out var state))
|
|
{
|
|
snapshot[taskId] = state.State;
|
|
}
|
|
}
|
|
|
|
return snapshot;
|
|
}
|
|
|
|
public async Task<int> ResumeRecoveringSessionsAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
using var lookupScope = _serviceScopeFactory.CreateScope();
|
|
var lookupDb = lookupScope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var candidates = await lookupDb.RecordSessions
|
|
.AsNoTracking()
|
|
.Where(item => item.Status == RecordSessionStatus.Starting &&
|
|
item.ErrorMessage != null &&
|
|
item.ErrorMessage.StartsWith("[runtime-recovery]"))
|
|
.OrderBy(item => item.UpdatedAt)
|
|
.Select(item => item.Id)
|
|
.Take(20)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var resumed = 0;
|
|
foreach (var sessionId in candidates)
|
|
{
|
|
if (IsRunning(sessionId))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
var transition = new SessionTransitionRuntime(sessionId);
|
|
if (!_sessionTransitions.TryAdd(sessionId, transition))
|
|
{
|
|
transition.Dispose();
|
|
continue;
|
|
}
|
|
|
|
try
|
|
{
|
|
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 == sessionId, cancellationToken);
|
|
if (session?.LiveRoom is null || session.Status != RecordSessionStatus.Starting)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(session.OutputPathPattern))
|
|
{
|
|
_logger.LogWarning("Recovering session {RecordSessionId} has no output path pattern and cannot be resumed.", session.Id);
|
|
continue;
|
|
}
|
|
|
|
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
|
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
|
|
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId, cancellationToken);
|
|
if (!liveStatus.IsLive)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
var stream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality, cancellationToken);
|
|
var latest = session.RecordTasks
|
|
.OrderByDescending(item => item.SegmentIndex)
|
|
.ThenByDescending(item => item.CreatedAt)
|
|
.FirstOrDefault();
|
|
var hasMedia = latest?.Result is { DurationSeconds: > 0 } result && File.Exists(result.FilePath);
|
|
RecordTask task;
|
|
if (latest is not null && ShouldReuseRecoveryTask(hasMedia, latest.Status))
|
|
{
|
|
task = latest;
|
|
}
|
|
else
|
|
{
|
|
var nextIndex = Math.Max(1, (latest?.SegmentIndex ?? 0) + 1);
|
|
task = new RecordTask(session.LiveRoomId, session.Id, nextIndex, session.PreferredQuality, session.OutputFormat, DateTimeOffset.UtcNow);
|
|
var output = NormalizeAbsolutePath(ResolveSegmentOutputPath(session.OutputPathPattern, session.SaveMode, nextIndex));
|
|
task.MarkStarting(stream.SelectedUrl, output, DateTimeOffset.UtcNow);
|
|
await dbContext.RecordTasks.AddAsync(task, cancellationToken);
|
|
}
|
|
|
|
var settingsResolver = scope.ServiceProvider.GetRequiredService<LiveRoomRecordingSettingsResolver>();
|
|
var recordingSettings = await settingsResolver.ResolveAsync(session.LiveRoom, cancellationToken);
|
|
var taskOutputPath = string.IsNullOrWhiteSpace(task.OutputFilePath)
|
|
? NormalizeAbsolutePath(ResolveSegmentOutputPath(session.OutputPathPattern, session.SaveMode, task.SegmentIndex))
|
|
: task.OutputFilePath;
|
|
session.MarkStarting(stream.SelectedUrl, session.OutputPathPattern, DateTimeOffset.UtcNow);
|
|
session.ActivateSegment(task.SegmentIndex, DateTimeOffset.UtcNow);
|
|
task.MarkStarting(stream.SelectedUrl, taskOutputPath, DateTimeOffset.UtcNow);
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
|
|
await StartInternalAsync(
|
|
session,
|
|
task,
|
|
stream,
|
|
recordingSettings,
|
|
InitialRecoveryContext with { AttemptCount = 1, HasRetriedWithRefreshedStream = true },
|
|
cancellationToken);
|
|
session.MarkRunning(DateTimeOffset.UtcNow);
|
|
task.MarkRunning(DateTimeOffset.UtcNow);
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
resumed++;
|
|
}
|
|
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to resume persisted recorder recovery for session {RecordSessionId}", sessionId);
|
|
}
|
|
finally
|
|
{
|
|
_sessionTransitions.TryRemove(sessionId, out _);
|
|
transition.Completion.TrySetResult(true);
|
|
transition.Dispose();
|
|
}
|
|
}
|
|
|
|
return resumed;
|
|
}
|
|
|
|
public Task CompleteAsync(Guid recordSessionId, CancellationToken cancellationToken = default) =>
|
|
RequestStopAsync(recordSessionId, markAsCompletedOnExit: true, cancellationToken);
|
|
|
|
public Task StartAsync(
|
|
RecordSession recordSession,
|
|
RecordTask initialTask,
|
|
StreamUrlResult streamUrlResult,
|
|
RecordingExecutionSettings recordingSettings,
|
|
CancellationToken cancellationToken = default) =>
|
|
StartInternalAsync(
|
|
recordSession,
|
|
initialTask,
|
|
streamUrlResult,
|
|
recordingSettings,
|
|
InitialRecoveryContext,
|
|
cancellationToken);
|
|
|
|
private async Task StartInternalAsync(
|
|
RecordSession recordSession,
|
|
RecordTask initialTask,
|
|
StreamUrlResult streamUrlResult,
|
|
RecordingExecutionSettings recordingSettings,
|
|
FfmpegRecoveryContext recoveryContext,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(recordSession);
|
|
ArgumentNullException.ThrowIfNull(initialTask);
|
|
ArgumentNullException.ThrowIfNull(streamUrlResult);
|
|
ArgumentNullException.ThrowIfNull(recordingSettings);
|
|
|
|
if (string.IsNullOrWhiteSpace(streamUrlResult.SelectedUrl) || string.IsNullOrWhiteSpace(recordSession.OutputPathPattern))
|
|
{
|
|
throw new InvalidOperationException("Recording session is missing stream URL or output path.");
|
|
}
|
|
|
|
using var settingsScope = _serviceScopeFactory.CreateScope();
|
|
var settingsService = settingsScope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var settings = await settingsService.GetAsync(cancellationToken);
|
|
var requiresVideoEncoding = recoveryContext.InputOptionProfile == FfmpegInputOptionProfile.TimestampTranscode ||
|
|
recordingSettings.RecordingTemplate == RecordingTemplateType.BalancedMp4;
|
|
var recoveryVideoEncoder = requiresVideoEncoding
|
|
? await ResolveVideoEncoderAsync(settings.FfmpegPath, recoveryContext.ForceSoftwareEncoder, cancellationToken)
|
|
: RecoveryVideoEncoderSelection.Software;
|
|
|
|
var outputPathPattern = Path.IsPathRooted(recordSession.OutputPathPattern)
|
|
? recordSession.OutputPathPattern
|
|
: Path.GetFullPath(recordSession.OutputPathPattern, AppContext.BaseDirectory);
|
|
var recorderOutputPath = GetRecorderOutputPath(outputPathPattern, recordSession.OutputFormat, recordSession.SaveMode);
|
|
Directory.CreateDirectory(Path.GetDirectoryName(recorderOutputPath)!);
|
|
|
|
var process = new Process
|
|
{
|
|
StartInfo = new ProcessStartInfo
|
|
{
|
|
FileName = settings.FfmpegPath,
|
|
UseShellExecute = false,
|
|
RedirectStandardInput = true,
|
|
RedirectStandardOutput = true,
|
|
RedirectStandardError = true,
|
|
CreateNoWindow = true
|
|
},
|
|
EnableRaisingEvents = true
|
|
};
|
|
|
|
var runtime = new SessionProcessRuntime(
|
|
recordSession.Id,
|
|
recordSession.LiveRoomId,
|
|
streamUrlResult.SelectedUrl,
|
|
outputPathPattern,
|
|
recorderOutputPath,
|
|
recordSession.OutputFormat,
|
|
recordSession.SaveMode,
|
|
initialTask.Id,
|
|
Math.Max(1, initialTask.SegmentIndex),
|
|
ResolveRecorderSegmentOutputPath(outputPathPattern, recordSession.OutputFormat, recordSession.SaveMode, Math.Max(1, initialTask.SegmentIndex)),
|
|
streamUrlResult.SelectedQuality,
|
|
streamUrlResult.SelectedProtocol,
|
|
streamUrlResult.SelectedVideoCodec,
|
|
streamUrlResult.InputHeaders,
|
|
recordingSettings,
|
|
recoveryContext,
|
|
recoveryVideoEncoder);
|
|
|
|
foreach (var argument in BuildArgumentList(
|
|
streamUrlResult.SelectedUrl,
|
|
recorderOutputPath,
|
|
recordSession.OutputFormat,
|
|
recordSession.SaveMode,
|
|
Math.Max(1, initialTask.SegmentIndex),
|
|
recordingSettings.RecordingTemplate,
|
|
recordingSettings.EnableAutoReconnect,
|
|
recordingSettings.ReconnectDelayMaxSeconds,
|
|
recordingSettings.ReadWriteTimeoutMilliseconds,
|
|
recordingSettings.SegmentDurationMinutes,
|
|
streamUrlResult.InputHeaders,
|
|
streamUrlResult.SelectedProtocol,
|
|
streamUrlResult.SelectedVideoCodec,
|
|
recoveryContext.InputOptionProfile,
|
|
recoveryVideoEncoder,
|
|
recoveryContext.UseCurlFallback))
|
|
{
|
|
process.StartInfo.ArgumentList.Add(argument);
|
|
}
|
|
|
|
process.OutputDataReceived += (_, args) => _ = HandleProcessOutputAsync(runtime, args.Data, isError: false);
|
|
process.ErrorDataReceived += (_, args) => _ = HandleProcessOutputAsync(runtime, args.Data, isError: true);
|
|
process.Exited += (_, _) => _ = HandleProcessExitedAsync(runtime, process);
|
|
|
|
if (!process.Start())
|
|
{
|
|
throw new InvalidOperationException("ffmpeg failed to start.");
|
|
}
|
|
|
|
runtime.AttachProcess(process);
|
|
if (!_processes.TryAdd(recordSession.Id, runtime))
|
|
{
|
|
process.Kill(true);
|
|
process.Dispose();
|
|
throw new InvalidOperationException("A running ffmpeg process already exists for the recording session.");
|
|
}
|
|
|
|
// curl is a one-shot compatibility fallback for native FLV inputs whose response
|
|
// headers exceed FFmpeg's built-in HTTP header limit. Normal FLV and all HLS
|
|
// inputs stay on FFmpeg's native HTTP stack so native reconnect remains active.
|
|
if (ShouldUseCurlPipe(
|
|
streamUrlResult.SelectedUrl,
|
|
streamUrlResult.SelectedProtocol,
|
|
recoveryContext.UseCurlFallback))
|
|
{
|
|
var curlProcess = new Process
|
|
{
|
|
StartInfo = new ProcessStartInfo
|
|
{
|
|
FileName = "curl",
|
|
UseShellExecute = false,
|
|
RedirectStandardOutput = true,
|
|
RedirectStandardError = true,
|
|
CreateNoWindow = true
|
|
},
|
|
EnableRaisingEvents = true
|
|
};
|
|
|
|
foreach (var arg in BuildCurlArgumentList(streamUrlResult.SelectedUrl, streamUrlResult.InputHeaders))
|
|
{
|
|
curlProcess.StartInfo.ArgumentList.Add(arg);
|
|
}
|
|
|
|
curlProcess.Exited += (_, _) =>
|
|
{
|
|
_logger.LogInformation(
|
|
"curl fallback connection ended for session {RecordSessionId} with exit code {ExitCode}.",
|
|
recordSession.Id,
|
|
runtime.CurlExitCode?.ToString() ?? "unknown");
|
|
};
|
|
|
|
curlProcess.ErrorDataReceived += (_, args) =>
|
|
{
|
|
if (!string.IsNullOrWhiteSpace(args.Data))
|
|
{
|
|
runtime.RememberCurlErrorLine(args.Data);
|
|
_logger.LogDebug("curl[{SessionId}] {Line}", recordSession.Id, args.Data);
|
|
}
|
|
};
|
|
|
|
if (!curlProcess.Start())
|
|
{
|
|
process.Kill(true);
|
|
process.Dispose();
|
|
throw new InvalidOperationException("curl failed to start.");
|
|
}
|
|
|
|
curlProcess.BeginErrorReadLine();
|
|
runtime.AttachCurlProcess(curlProcess);
|
|
|
|
// Pipe curl stdout → FFmpeg stdin in the background.
|
|
// When curl exits or is killed, the pipe closes and FFmpeg sees EOF on stdin,
|
|
// which causes it to exit gracefully after finalizing the output.
|
|
_ = Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
await curlProcess.StandardOutput.BaseStream.CopyToAsync(
|
|
process.StandardInput.BaseStream);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogDebug(ex, "curl→ffmpeg pipe ended for session {RecordSessionId}", recordSession.Id);
|
|
}
|
|
finally
|
|
{
|
|
try { process.StandardInput.Close(); } catch { /* stdin may already be closed */ }
|
|
}
|
|
});
|
|
}
|
|
|
|
await PersistProcessBindingAsync(recordSession.Id, initialTask.Id, process.Id, cancellationToken);
|
|
await PersistStartupProfileAsync(runtime, cancellationToken);
|
|
await StartDanmakuAsync(runtime, initialTask, cancellationToken);
|
|
|
|
process.BeginOutputReadLine();
|
|
process.BeginErrorReadLine();
|
|
}
|
|
|
|
public Task StopAsync(Guid recordSessionId, CancellationToken cancellationToken = default) =>
|
|
RequestStopAsync(recordSessionId, markAsCompletedOnExit: false, cancellationToken);
|
|
|
|
public async Task<bool> StopAndWaitAsync(
|
|
Guid recordSessionId,
|
|
bool markAsCompletedOnExit,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
if (!_processes.TryGetValue(recordSessionId, out var runtime))
|
|
{
|
|
if (_sessionTransitions.TryGetValue(recordSessionId, out var transition))
|
|
{
|
|
transition.RequestStop(markAsCompletedOnExit);
|
|
return await WaitForTransitionAsync(transition, timeout, cancellationToken);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
await RequestStopAsync(recordSessionId, markAsCompletedOnExit, cancellationToken);
|
|
return await WaitForExitAsync(runtime, timeout, cancellationToken);
|
|
}
|
|
|
|
public async Task<bool> KillAndWaitAsync(
|
|
Guid recordSessionId,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
if (!_processes.TryGetValue(recordSessionId, out var runtime))
|
|
{
|
|
if (_sessionTransitions.TryGetValue(recordSessionId, out var transition))
|
|
{
|
|
transition.RequestStop(markAsCompletedOnExit: false);
|
|
return await WaitForTransitionAsync(transition, timeout, cancellationToken);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
var process = runtime.Process;
|
|
var curlProcess = runtime.CurlProcess;
|
|
if (curlProcess is not null)
|
|
{
|
|
try
|
|
{
|
|
if (!curlProcess.HasExited)
|
|
{
|
|
curlProcess.Kill(true);
|
|
}
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
}
|
|
}
|
|
|
|
if (process is not null)
|
|
{
|
|
try
|
|
{
|
|
if (!process.HasExited)
|
|
{
|
|
runtime.MarkForceKilled();
|
|
process.Kill(true);
|
|
}
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
}
|
|
}
|
|
|
|
return await WaitForExitAsync(runtime, timeout, cancellationToken);
|
|
}
|
|
|
|
public async Task<int> StopAllAndWaitAsync(
|
|
TimeSpan gracefulTimeout,
|
|
TimeSpan forceKillTimeout,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var runtimes = _processes.Values.ToArray();
|
|
var transitions = _sessionTransitions.Values.ToArray();
|
|
foreach (var runtime in runtimes)
|
|
{
|
|
// Mark the captured runtime before looking it up again. The process may exit
|
|
// between the snapshot and the stop signal, but its exit handler must still
|
|
// observe that this was an application shutdown rather than a stream failure.
|
|
runtime.MarkShutdownRequested();
|
|
}
|
|
|
|
await Task.WhenAll(runtimes.Select(runtime =>
|
|
RequestStopAsync(
|
|
runtime.RecordSessionId,
|
|
markAsCompletedOnExit: true,
|
|
cancellationToken,
|
|
shutdownRequested: true)));
|
|
|
|
foreach (var transition in transitions)
|
|
{
|
|
transition.RequestStop(markAsCompletedOnExit: true);
|
|
}
|
|
|
|
await WaitForRuntimeCompletionsAsync(runtimes, gracefulTimeout, cancellationToken);
|
|
await WaitForTransitionCompletionsAsync(transitions, gracefulTimeout, cancellationToken);
|
|
|
|
foreach (var runtime in runtimes.Where(static runtime => !runtime.ExitCompletion.Task.IsCompleted))
|
|
{
|
|
var process = runtime.Process;
|
|
try
|
|
{
|
|
if (process is not null && !process.HasExited)
|
|
{
|
|
runtime.MarkForceKilled();
|
|
process.Kill(true);
|
|
}
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
}
|
|
}
|
|
|
|
await WaitForRuntimeCompletionsAsync(runtimes, forceKillTimeout, CancellationToken.None);
|
|
|
|
_shutdownCts.Cancel();
|
|
foreach (var process in _postProcessProcesses.Values.ToArray())
|
|
{
|
|
try
|
|
{
|
|
if (!process.HasExited)
|
|
{
|
|
process.Kill(true);
|
|
}
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
}
|
|
}
|
|
|
|
await WaitForAllProcessesAsync(forceKillTimeout, CancellationToken.None);
|
|
return runtimes.Length + transitions.Length;
|
|
}
|
|
|
|
private static async Task WaitForTransitionCompletionsAsync(
|
|
IReadOnlyCollection<SessionTransitionRuntime> transitions,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (transitions.Count == 0 || transitions.All(static item => item.Completion.Task.IsCompleted)) return;
|
|
var completion = Task.WhenAll(transitions.Select(static item => item.Completion.Task));
|
|
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
var delay = Task.Delay(timeout, timeoutCts.Token);
|
|
if (await Task.WhenAny(completion, delay) == completion)
|
|
{
|
|
timeoutCts.Cancel();
|
|
await completion;
|
|
}
|
|
}
|
|
|
|
private static async Task WaitForRuntimeCompletionsAsync(
|
|
IReadOnlyCollection<SessionProcessRuntime> runtimes,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (runtimes.Count == 0 || runtimes.All(static runtime => runtime.ExitCompletion.Task.IsCompleted))
|
|
{
|
|
return;
|
|
}
|
|
|
|
var completionTask = Task.WhenAll(runtimes.Select(static runtime => runtime.ExitCompletion.Task));
|
|
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
var delayTask = Task.Delay(timeout, timeoutCts.Token);
|
|
if (await Task.WhenAny(completionTask, delayTask) == completionTask)
|
|
{
|
|
timeoutCts.Cancel();
|
|
await completionTask;
|
|
}
|
|
}
|
|
|
|
private async Task WaitForAllProcessesAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
|
{
|
|
var deadline = DateTimeOffset.UtcNow + timeout;
|
|
while ((!_processes.IsEmpty || !_sessionTransitions.IsEmpty || !_postProcessProcesses.IsEmpty || !_postProcessStates.IsEmpty) &&
|
|
DateTimeOffset.UtcNow < deadline)
|
|
{
|
|
await Task.Delay(TimeSpan.FromMilliseconds(200), cancellationToken);
|
|
}
|
|
}
|
|
|
|
public Task<bool> TryReconcileInactiveSessionAsync(Guid recordSessionId, CancellationToken cancellationToken = default) =>
|
|
ReconcileInactiveSessionAsync(recordSessionId, allowTerminalSession: false, cancellationToken);
|
|
|
|
public async Task<int> RecoverOrphanedTerminalSessionTasksAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
if (!await _orphanRecoveryGate.WaitAsync(0, cancellationToken))
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var candidateIds = await dbContext.RecordSessions
|
|
.AsNoTracking()
|
|
.Where(session =>
|
|
(session.Status == RecordSessionStatus.Completed ||
|
|
session.Status == RecordSessionStatus.Failed ||
|
|
session.Status == RecordSessionStatus.Stopped) &&
|
|
session.RecordTasks.Any(task =>
|
|
task.Status == RecordTaskStatus.Starting ||
|
|
task.Status == RecordTaskStatus.Running ||
|
|
task.Status == RecordTaskStatus.Stopping))
|
|
.OrderBy(session => session.UpdatedAt)
|
|
.Select(session => session.Id)
|
|
.Take(20)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var recovered = 0;
|
|
foreach (var candidateId in candidateIds)
|
|
{
|
|
if (await ReconcileInactiveSessionAsync(candidateId, allowTerminalSession: true, cancellationToken))
|
|
{
|
|
recovered++;
|
|
}
|
|
}
|
|
|
|
return recovered;
|
|
}
|
|
finally
|
|
{
|
|
_orphanRecoveryGate.Release();
|
|
}
|
|
}
|
|
|
|
public async Task<int> RecoverStalePendingSessionsAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
if (!await _orphanRecoveryGate.WaitAsync(0, cancellationToken))
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var cutoff = DateTimeOffset.UtcNow - StalePendingThreshold;
|
|
var candidateIds = await dbContext.RecordSessions
|
|
.AsNoTracking()
|
|
.Where(session =>
|
|
session.Status == RecordSessionStatus.Pending &&
|
|
session.UpdatedAt <= cutoff)
|
|
.OrderBy(session => session.UpdatedAt)
|
|
.Select(session => session.Id)
|
|
.Take(50)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var recovered = 0;
|
|
foreach (var candidateId in candidateIds)
|
|
{
|
|
if (IsRunning(candidateId))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
using var candidateScope = _serviceScopeFactory.CreateScope();
|
|
var candidateDbContext = candidateScope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var session = await candidateDbContext.RecordSessions
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == candidateId, cancellationToken);
|
|
if (session is null ||
|
|
session.Status != RecordSessionStatus.Pending ||
|
|
session.UpdatedAt > cutoff ||
|
|
IsRunning(candidateId))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (HasRecoverablePendingMedia(session))
|
|
{
|
|
if (await ReconcileInactiveSessionAsync(candidateId, allowTerminalSession: false, cancellationToken))
|
|
{
|
|
recovered++;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
var failedAt = DateTimeOffset.UtcNow;
|
|
foreach (var task in session.RecordTasks.Where(task => task.Status == RecordTaskStatus.Pending))
|
|
{
|
|
task.MarkFailed(StalePendingStartupFailure, failedAt);
|
|
}
|
|
|
|
session.MarkFailed(StalePendingStartupFailure, failedAt);
|
|
await candidateDbContext.SaveChangesAsync(cancellationToken);
|
|
|
|
var logService = candidateScope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"RecordSession",
|
|
"A stale pending recording session was marked failed.",
|
|
StalePendingStartupFailure,
|
|
session.LiveRoomId,
|
|
session.Id,
|
|
session.RecordTasks.OrderBy(task => task.SegmentIndex).FirstOrDefault()?.Id,
|
|
cancellationToken);
|
|
recovered++;
|
|
}
|
|
|
|
return recovered;
|
|
}
|
|
finally
|
|
{
|
|
_orphanRecoveryGate.Release();
|
|
}
|
|
}
|
|
|
|
private static bool HasRecoverablePendingMedia(RecordSession session)
|
|
{
|
|
if (DiscoverRecoverableSegments(session.OutputPathPattern, session.OutputFormat, session.SaveMode).Count > 0)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
foreach (var task in session.RecordTasks)
|
|
{
|
|
var candidates = new[]
|
|
{
|
|
task.Result?.FilePath,
|
|
task.OutputFilePath,
|
|
string.IsNullOrWhiteSpace(task.OutputFilePath)
|
|
? null
|
|
: GetRecorderOutputPath(task.OutputFilePath, session.OutputFormat, session.SaveMode)
|
|
};
|
|
|
|
if (candidates.Any(path =>
|
|
!string.IsNullOrWhiteSpace(path) &&
|
|
File.Exists(path) &&
|
|
new FileInfo(path).Length > 0))
|
|
{
|
|
return true;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
public async Task<bool> TryRecoverOrphanedTerminalTaskAsync(
|
|
Guid recordTaskId,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
await _orphanRecoveryGate.WaitAsync(cancellationToken);
|
|
try
|
|
{
|
|
using var lookupScope = _serviceScopeFactory.CreateScope();
|
|
var lookupDbContext = lookupScope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var sessionId = await lookupDbContext.RecordTasks
|
|
.AsNoTracking()
|
|
.Where(task =>
|
|
task.Id == recordTaskId &&
|
|
(task.Status == RecordTaskStatus.Starting ||
|
|
task.Status == RecordTaskStatus.Running ||
|
|
task.Status == RecordTaskStatus.Stopping) &&
|
|
task.RecordSession != null &&
|
|
(task.RecordSession.Status == RecordSessionStatus.Completed ||
|
|
task.RecordSession.Status == RecordSessionStatus.Failed ||
|
|
task.RecordSession.Status == RecordSessionStatus.Stopped))
|
|
.Select(task => (Guid?)task.RecordSessionId)
|
|
.FirstOrDefaultAsync(cancellationToken);
|
|
|
|
return sessionId.HasValue &&
|
|
await ReconcileInactiveSessionAsync(sessionId.Value, allowTerminalSession: true, cancellationToken);
|
|
}
|
|
finally
|
|
{
|
|
_orphanRecoveryGate.Release();
|
|
}
|
|
}
|
|
|
|
private async Task<bool> ReconcileInactiveSessionAsync(
|
|
Guid recordSessionId,
|
|
bool allowTerminalSession,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (IsRunning(recordSessionId))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
|
|
var recordSession = await dbContext.RecordSessions
|
|
.Include(item => item.LiveRoom)
|
|
.Include(item => item.RecordTasks)
|
|
.ThenInclude(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordSessionId, cancellationToken);
|
|
|
|
if (recordSession is null ||
|
|
(!IsActiveSessionStatus(recordSession.Status) &&
|
|
!(allowTerminalSession && IsTerminalSessionStatus(recordSession.Status))))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var hasActiveTasks = recordSession.RecordTasks.Any(task => IsActiveTaskStatus(task.Status));
|
|
if (allowTerminalSession && !hasActiveTasks)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var discoveredSegments = DiscoverRecoverableSegments(recordSession.OutputPathPattern, recordSession.OutputFormat, recordSession.SaveMode);
|
|
var existingSegmentIndexes = recordSession.RecordTasks
|
|
.Select(static task => task.SegmentIndex)
|
|
.ToHashSet();
|
|
var addedMissingTasks = false;
|
|
foreach (var segment in discoveredSegments.Where(segment => !existingSegmentIndexes.Contains(segment.SegmentIndex)))
|
|
{
|
|
var discoveredAt = File.GetLastWriteTimeUtc(segment.RecorderPath);
|
|
var createdAt = discoveredAt == DateTime.MinValue
|
|
? DateTimeOffset.UtcNow
|
|
: new DateTimeOffset(DateTime.SpecifyKind(discoveredAt, DateTimeKind.Utc));
|
|
var missingTask = new RecordTask(
|
|
recordSession.LiveRoomId,
|
|
recordSession.Id,
|
|
segment.SegmentIndex,
|
|
recordSession.PreferredQuality,
|
|
recordSession.OutputFormat,
|
|
createdAt);
|
|
var recoveryStartedAt = DateTimeOffset.UtcNow;
|
|
missingTask.MarkStarting(recordSession.StreamUrl ?? string.Empty, segment.OutputPath, recoveryStartedAt);
|
|
missingTask.MarkRunning(recoveryStartedAt);
|
|
await dbContext.RecordTasks.AddAsync(missingTask, cancellationToken);
|
|
recordSession.RecordTasks.Add(missingTask);
|
|
existingSegmentIndexes.Add(segment.SegmentIndex);
|
|
addedMissingTasks = true;
|
|
}
|
|
|
|
// RecordResult is upserted with raw SQL below, so newly discovered tasks must
|
|
// exist first to satisfy the RecordTaskId foreign key.
|
|
if (addedMissingTasks)
|
|
{
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
var settings = await settingsService.GetAsync(cancellationToken);
|
|
var endedAt = DateTimeOffset.UtcNow;
|
|
var tasks = recordSession.RecordTasks
|
|
.OrderBy(static item => item.SegmentIndex)
|
|
.ThenBy(static item => item.CreatedAt)
|
|
.ToList();
|
|
|
|
var anyUsableOutput = tasks.Any(static item => item.Status == RecordTaskStatus.Completed);
|
|
var hasBackgroundPostProcessing = false;
|
|
string? sessionFinalizationError = null;
|
|
var newlyCompletedTasks = new List<(RecordTask Task, string OutputPath)>();
|
|
|
|
foreach (var task in tasks)
|
|
{
|
|
if (IsTaskUnderPostProcessing(task.Id))
|
|
{
|
|
hasBackgroundPostProcessing = true;
|
|
continue;
|
|
}
|
|
|
|
if (!IsActiveTaskStatus(task.Status))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
var durationSeconds = task.StartedAt.HasValue
|
|
? (double?)Math.Max(0, (endedAt - task.StartedAt.Value).TotalSeconds)
|
|
: null;
|
|
var finalizationResult = await TryFinalizeTaskOutputAsync(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
recordSession,
|
|
task,
|
|
durationSeconds,
|
|
cancellationToken);
|
|
var effectiveOutputPath = finalizationResult.OutputPath;
|
|
var taskFinalizationError = finalizationResult.ErrorMessage;
|
|
var mediaValidation = await ValidateMediaArtifactAsync(
|
|
effectiveOutputPath,
|
|
recordSession.OutputFormat,
|
|
cancellationToken);
|
|
durationSeconds = mediaValidation.DurationSeconds;
|
|
if (!IsLowStoragePauseError(taskFinalizationError))
|
|
{
|
|
sessionFinalizationError ??= taskFinalizationError;
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(taskFinalizationError) && !mediaValidation.IsValid)
|
|
{
|
|
sessionFinalizationError ??= mediaValidation.ErrorMessage;
|
|
}
|
|
|
|
var fileSize = CalculateFileSize(effectiveOutputPath);
|
|
var danmakuPath = task.Result?.DanmakuFilePath ?? GuessDanmakuPath(task.OutputFilePath);
|
|
var danmakuCount = CountDanmakuMessages(danmakuPath);
|
|
|
|
if (IsLowStoragePauseError(taskFinalizationError))
|
|
{
|
|
task.MarkProcessing(taskFinalizationError, endedAt);
|
|
hasBackgroundPostProcessing = true;
|
|
}
|
|
else if (!string.IsNullOrWhiteSpace(taskFinalizationError))
|
|
{
|
|
task.MarkFailed(taskFinalizationError, endedAt, durationSeconds);
|
|
}
|
|
else if (mediaValidation.IsValid)
|
|
{
|
|
task.MarkCompleted(endedAt, durationSeconds);
|
|
anyUsableOutput = true;
|
|
newlyCompletedTasks.Add((task, effectiveOutputPath));
|
|
}
|
|
else
|
|
{
|
|
task.MarkFailed(mediaValidation.ErrorMessage!, endedAt, durationSeconds);
|
|
}
|
|
|
|
await UpsertRecordResultAsync(
|
|
task,
|
|
dbContext,
|
|
effectiveOutputPath,
|
|
fileSize,
|
|
durationSeconds,
|
|
danmakuPath,
|
|
danmakuCount,
|
|
endedAt,
|
|
mediaValidatedForDispatch: mediaValidation.IsValid,
|
|
cancellationToken: cancellationToken);
|
|
ClearPostProcessState(task.Id);
|
|
}
|
|
|
|
recordSession.SyncSegmentCount(tasks.Count == 0 ? 0 : tasks.Max(static task => task.SegmentIndex), endedAt);
|
|
if (!string.IsNullOrWhiteSpace(sessionFinalizationError))
|
|
{
|
|
recordSession.MarkFailed(sessionFinalizationError, endedAt);
|
|
}
|
|
else if (hasBackgroundPostProcessing)
|
|
{
|
|
recordSession.MarkStopped(
|
|
endedAt,
|
|
"Recording process was no longer running when the session was reconciled. Post-processing will continue in background.");
|
|
}
|
|
else if (anyUsableOutput)
|
|
{
|
|
recordSession.MarkCompleted(endedAt);
|
|
}
|
|
else
|
|
{
|
|
recordSession.MarkStopped(endedAt, "Recording process was no longer running when the session was reconciled.");
|
|
}
|
|
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
|
|
if (newlyCompletedTasks.Count > 0 && recordSession.LiveRoom is not null)
|
|
{
|
|
var completionDispatchService = scope.ServiceProvider.GetRequiredService<CompletionDispatchService>();
|
|
foreach (var completed in newlyCompletedTasks)
|
|
{
|
|
try
|
|
{
|
|
await completionDispatchService.TryDispatchTaskAsync(completed.Task.Id, cancellationToken);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(
|
|
ex,
|
|
"Recovered segment completion hooks failed for task {RecordTaskId}",
|
|
completed.Task.Id);
|
|
}
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
public async Task<bool> StartManualFinalizeTaskAsync(Guid recordTaskId, CancellationToken cancellationToken = default)
|
|
{
|
|
if (_postProcessStates.ContainsKey(recordTaskId))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var recordTask = await dbContext.RecordTasks
|
|
.Include(item => item.RecordSession)
|
|
.Include(item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
|
|
|
|
if (recordTask?.RecordSession is null)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (recordTask.OutputFormat != RecordOutputFormat.Mp4 ||
|
|
IsActiveTaskStatus(recordTask.Status) ||
|
|
IsRunning(recordTask.RecordSessionId))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(recordTask.OutputFilePath))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var recorderOutputPaths = ResolveManualFinalizeSourcePaths(recordTask, recordTask.RecordSession);
|
|
if (recorderOutputPaths.Count == 0)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
SetPostProcessState(
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id,
|
|
"Queued",
|
|
null,
|
|
$"Manual MP4 finalization queued for {Path.GetFileName(recordTask.OutputFilePath)}");
|
|
|
|
_ = Task.Run(
|
|
async () => await RunManualFinalizeTaskAsync(recordTask.Id),
|
|
CancellationToken.None);
|
|
|
|
return true;
|
|
}
|
|
|
|
public async Task<TranscodeMediaFileResultDto> StartManualFinalizeFileAsync(
|
|
string sourceFilePath,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(sourceFilePath))
|
|
{
|
|
return new TranscodeMediaFileResultDto
|
|
{
|
|
Success = false,
|
|
Message = "The selected source file is empty."
|
|
};
|
|
}
|
|
|
|
var absoluteSourcePath = NormalizeAbsolutePath(sourceFilePath);
|
|
if (!File.Exists(absoluteSourcePath))
|
|
{
|
|
return new TranscodeMediaFileResultDto
|
|
{
|
|
Success = false,
|
|
Message = "The selected .ts file does not exist.",
|
|
SourcePath = absoluteSourcePath
|
|
};
|
|
}
|
|
|
|
if (!absoluteSourcePath.EndsWith(".ts", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
return new TranscodeMediaFileResultDto
|
|
{
|
|
Success = false,
|
|
Message = "Only .ts files can be transcoded to MP4.",
|
|
SourcePath = absoluteSourcePath
|
|
};
|
|
}
|
|
|
|
var absoluteTargetPath = Path.ChangeExtension(absoluteSourcePath, ".mp4");
|
|
if (File.Exists(absoluteTargetPath))
|
|
{
|
|
return new TranscodeMediaFileResultDto
|
|
{
|
|
Success = false,
|
|
Message = "A target MP4 file already exists for the selected .ts file.",
|
|
SourcePath = absoluteSourcePath,
|
|
OutputPath = absoluteTargetPath
|
|
};
|
|
}
|
|
|
|
using var settingsScope = _serviceScopeFactory.CreateScope();
|
|
var settingsService = settingsScope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var settings = await settingsService.GetAsync(cancellationToken);
|
|
|
|
var syntheticSessionId = Guid.NewGuid();
|
|
var syntheticTaskId = Guid.NewGuid();
|
|
SetPostProcessState(
|
|
syntheticSessionId,
|
|
syntheticTaskId,
|
|
"Queued",
|
|
null,
|
|
$"Manual file transcode queued for {Path.GetFileName(absoluteSourcePath)}");
|
|
|
|
_ = Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
var result = await TryFinalizeMp4Async(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
syntheticSessionId,
|
|
syntheticTaskId,
|
|
[absoluteSourcePath],
|
|
absoluteTargetPath,
|
|
expectedDurationSeconds: null,
|
|
segmentsManifestPath: null,
|
|
CancellationToken.None);
|
|
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
|
|
if (string.IsNullOrWhiteSpace(result.ErrorMessage))
|
|
{
|
|
await logService.WriteAsync(
|
|
Domain.Enums.SystemLogLevel.Info,
|
|
"FFmpeg",
|
|
"Manual file transcode completed.",
|
|
$"source={absoluteSourcePath}; output={result.OutputPath}",
|
|
cancellationToken: CancellationToken.None);
|
|
}
|
|
else
|
|
{
|
|
await logService.WriteAsync(
|
|
Domain.Enums.SystemLogLevel.Warning,
|
|
"FFmpeg",
|
|
"Manual file transcode failed.",
|
|
$"source={absoluteSourcePath}; output={result.OutputPath}; error={result.ErrorMessage}",
|
|
cancellationToken: CancellationToken.None);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
Domain.Enums.SystemLogLevel.Error,
|
|
"FFmpeg",
|
|
"Manual file transcode crashed.",
|
|
$"source={absoluteSourcePath}; output={absoluteTargetPath}; error={ex}",
|
|
cancellationToken: CancellationToken.None);
|
|
}
|
|
finally
|
|
{
|
|
ClearPostProcessState(syntheticTaskId);
|
|
}
|
|
}, CancellationToken.None);
|
|
|
|
return new TranscodeMediaFileResultDto
|
|
{
|
|
Success = true,
|
|
Message = "Manual file transcode started. Refresh later to verify the output file.",
|
|
SourcePath = absoluteSourcePath,
|
|
OutputPath = absoluteTargetPath
|
|
};
|
|
}
|
|
|
|
public async Task<int> ResumePausedFinalizationsAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
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(cancellationToken);
|
|
|
|
var storageCheck = _storageGuardService.CheckCanStartOrResume(settings);
|
|
var candidates = await dbContext.RecordTasks
|
|
.Include(item => item.RecordSession)
|
|
.ThenInclude(item => item!.RecordTasks)
|
|
.Include(item => item.Result)
|
|
.Where(item => item.OutputFormat == RecordOutputFormat.Mp4 &&
|
|
(item.Status == RecordTaskStatus.Processing ||
|
|
item.Status == RecordTaskStatus.Completed ||
|
|
item.Status == RecordTaskStatus.Failed &&
|
|
item.ErrorMessage == FfprobeUnreadableArtifactError))
|
|
// Deployment-paused work must not be starved by a large backlog of
|
|
// older legacy failures that may no longer have local source files.
|
|
.OrderBy(static item => item.Status == RecordTaskStatus.Processing
|
|
? 0
|
|
: item.Status == RecordTaskStatus.Completed
|
|
? 1
|
|
: 2)
|
|
.ThenBy(static item => item.UpdatedAt)
|
|
.Take(100)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var queuedTaskIds = new List<Guid>();
|
|
var recoveredValidatedTaskIds = new List<Guid>();
|
|
var repairedInterruptedTasks = 0;
|
|
var repairedLegacyShutdownFailures = 0;
|
|
var now = DateTimeOffset.UtcNow;
|
|
foreach (var candidate in candidates)
|
|
{
|
|
if (candidate.Status == RecordTaskStatus.Processing)
|
|
{
|
|
queuedTaskIds.Add(candidate.Id);
|
|
continue;
|
|
}
|
|
|
|
if (candidate.Status == RecordTaskStatus.Failed)
|
|
{
|
|
var recoverySources = candidate.RecordSession is null
|
|
? Array.Empty<string>()
|
|
: ResolveManualFinalizeSourcePaths(candidate, candidate.RecordSession);
|
|
if (recoverySources.Count > 0)
|
|
{
|
|
candidate.MarkProcessing(
|
|
"A deployment-interrupted MP4 finalization was recovered and queued after restart.",
|
|
now);
|
|
candidate.RecordSession!.MarkStopped(
|
|
now,
|
|
"A deployment-interrupted MP4 finalization is continuing in the background.");
|
|
queuedTaskIds.Add(candidate.Id);
|
|
repairedLegacyShutdownFailures++;
|
|
continue;
|
|
}
|
|
|
|
var existingOutputPath = candidate.Result?.FilePath ?? candidate.OutputFilePath;
|
|
var mediaValidation = await ValidateMediaArtifactAsync(
|
|
existingOutputPath,
|
|
candidate.OutputFormat,
|
|
cancellationToken);
|
|
if (!mediaValidation.IsValid)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
candidate.MarkCompleted(now, mediaValidation.DurationSeconds);
|
|
if (candidate.RecordSession is not null &&
|
|
candidate.RecordSession.RecordTasks.All(static task =>
|
|
task.Status is RecordTaskStatus.Completed or RecordTaskStatus.Stopped))
|
|
{
|
|
candidate.RecordSession.MarkCompleted(now);
|
|
}
|
|
|
|
await UpsertRecordResultAsync(
|
|
candidate,
|
|
dbContext,
|
|
existingOutputPath,
|
|
CalculateFileSize(existingOutputPath),
|
|
mediaValidation.DurationSeconds,
|
|
candidate.Result?.DanmakuFilePath,
|
|
candidate.Result?.DanmakuMessageCount ?? 0,
|
|
now,
|
|
mediaValidatedForDispatch: true,
|
|
cancellationToken: cancellationToken);
|
|
recoveredValidatedTaskIds.Add(candidate.Id);
|
|
repairedLegacyShutdownFailures++;
|
|
continue;
|
|
}
|
|
|
|
if (!NeedsInterruptedMp4Finalization(candidate))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
candidate.MarkProcessing("MP4 finalization was interrupted before completion. Re-queued after restart.", now);
|
|
queuedTaskIds.Add(candidate.Id);
|
|
repairedInterruptedTasks++;
|
|
}
|
|
|
|
if (repairedInterruptedTasks > 0 || repairedLegacyShutdownFailures > 0)
|
|
{
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
if (recoveredValidatedTaskIds.Count > 0)
|
|
{
|
|
var completionDispatchService = scope.ServiceProvider.GetRequiredService<CompletionDispatchService>();
|
|
foreach (var taskId in recoveredValidatedTaskIds)
|
|
{
|
|
await completionDispatchService.TryDispatchTaskAsync(taskId, cancellationToken);
|
|
}
|
|
}
|
|
|
|
var started = 0;
|
|
foreach (var taskId in queuedTaskIds
|
|
.Where(taskId => !IsTaskUnderPostProcessing(taskId))
|
|
.Take(20))
|
|
{
|
|
if (await StartManualFinalizeTaskAsync(taskId, cancellationToken))
|
|
{
|
|
started++;
|
|
}
|
|
}
|
|
|
|
if (started > 0 || repairedLegacyShutdownFailures > 0)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"Storage",
|
|
"Resumed paused MP4 finalizations and repaired deployment-interrupted recording results.",
|
|
$"count={started}; repairedInterrupted={repairedInterruptedTasks}; repairedLegacyShutdown={repairedLegacyShutdownFailures}; {storageCheck.Message}",
|
|
cancellationToken: cancellationToken);
|
|
}
|
|
|
|
return started;
|
|
}
|
|
|
|
public async Task<bool> StartArtifactRepairAsync(
|
|
Guid recordTaskId,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
if (_postProcessStates.ContainsKey(recordTaskId))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var recordTask = await dbContext.RecordTasks
|
|
.Include(static item => item.RecordSession)
|
|
.Include(static item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
|
|
if (recordTask?.Result is null || recordTask.RecordSession is null ||
|
|
recordTask.Status is not (RecordTaskStatus.Failed or RecordTaskStatus.Processing) ||
|
|
recordTask.Status == RecordTaskStatus.Processing &&
|
|
recordTask.ErrorMessage?.StartsWith(ArtifactRepairMarker, StringComparison.Ordinal) != true)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var sourcePath = NormalizeAbsolutePath(recordTask.Result.FilePath);
|
|
if (!File.Exists(sourcePath) || new FileInfo(sourcePath).Length <= 0)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var settings = await settingsService.GetAsync(cancellationToken);
|
|
var storage = _storageGuardService.CheckCanStartOrResume(settings);
|
|
if (!storage.IsAvailable || storage.AvailableBytes < Math.Max(storage.RequiredBytes, new FileInfo(sourcePath).Length))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var targetPath = BuildRecoveredArtifactPath(sourcePath, recordTaskId);
|
|
var now = DateTimeOffset.UtcNow;
|
|
if (!_postProcessStates.TryAdd(
|
|
recordTask.Id,
|
|
new PostProcessRuntimeEntry(
|
|
recordTask.RecordSessionId,
|
|
new RecordTaskRuntimeState(
|
|
RecordTaskStatus.Processing,
|
|
"Queued",
|
|
0,
|
|
$"Waiting to repair {Path.GetFileName(sourcePath)}"))))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
try
|
|
{
|
|
recordTask.MarkProcessing(
|
|
$"{ArtifactRepairMarker} 正在生成非破坏性恢复文件：{Path.GetFileName(targetPath)}",
|
|
now);
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
catch
|
|
{
|
|
ClearPostProcessState(recordTask.Id);
|
|
throw;
|
|
}
|
|
|
|
_ = Task.Run(
|
|
async () => await RunArtifactRepairAsync(recordTaskId),
|
|
CancellationToken.None);
|
|
return true;
|
|
}
|
|
|
|
public async Task<int> ResumeArtifactRepairsAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var taskIds = await dbContext.RecordTasks
|
|
.AsNoTracking()
|
|
.Where(item => item.Status == RecordTaskStatus.Processing &&
|
|
item.ErrorMessage != null &&
|
|
item.ErrorMessage.StartsWith(ArtifactRepairMarker))
|
|
.OrderBy(static item => item.UpdatedAt)
|
|
.Select(static item => item.Id)
|
|
.Take(20)
|
|
.ToArrayAsync(cancellationToken);
|
|
|
|
var started = 0;
|
|
foreach (var taskId in taskIds)
|
|
{
|
|
if (await StartArtifactRepairAsync(taskId, cancellationToken))
|
|
{
|
|
started++;
|
|
}
|
|
}
|
|
|
|
return started;
|
|
}
|
|
|
|
private async Task RunArtifactRepairAsync(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(static item => item.RecordSession)
|
|
.ThenInclude(static item => item!.RecordTasks)
|
|
.Include(static item => item.Result)
|
|
.FirstOrDefaultAsync(item => item.Id == recordTaskId);
|
|
if (recordTask?.Result is null || recordTask.RecordSession is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var sourcePath = NormalizeAbsolutePath(recordTask.Result.FilePath);
|
|
var targetPath = BuildRecoveredArtifactPath(sourcePath, recordTaskId);
|
|
var originalError = recordTask.ErrorMessage;
|
|
var settings = await settingsService.GetAsync();
|
|
SetPostProcessState(
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id,
|
|
"Repairing",
|
|
null,
|
|
$"Repairing {Path.GetFileName(sourcePath)} without replacing the source");
|
|
|
|
var repair = await TryRepairArtifactFileAsync(
|
|
settings.FfmpegPath,
|
|
settings.MaxConcurrentFfmpegTranscodeTasks,
|
|
settings.Mp4FinalizeTimeoutMinutes,
|
|
recordTask.Id,
|
|
sourcePath,
|
|
targetPath,
|
|
_shutdownCts.Token);
|
|
if (_shutdownCts.IsCancellationRequested)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var now = DateTimeOffset.UtcNow;
|
|
var metadata = string.IsNullOrWhiteSpace(repair.ErrorMessage)
|
|
? await scope.ServiceProvider.GetRequiredService<IVideoMetadataService>()
|
|
.ExtractMetadataAsync(targetPath)
|
|
: null;
|
|
if (metadata?.DurationSeconds is > 0 && !string.IsNullOrWhiteSpace(metadata.VideoCodec))
|
|
{
|
|
recordTask.MarkCompleted(now, metadata.DurationSeconds);
|
|
recordTask.Result.Update(
|
|
targetPath,
|
|
new FileInfo(targetPath).Length,
|
|
metadata.DurationSeconds,
|
|
recordTask.Result.DanmakuFilePath,
|
|
recordTask.Result.DanmakuMessageCount,
|
|
RecordTaskStatus.Completed,
|
|
null);
|
|
recordTask.Result.ResetUploadForRecoveredArtifact();
|
|
if (recordTask.RecordSession.RecordTasks.All(static task =>
|
|
task.Status is RecordTaskStatus.Completed or RecordTaskStatus.Stopped))
|
|
{
|
|
recordTask.RecordSession.MarkCompleted(now);
|
|
}
|
|
|
|
await dbContext.SaveChangesAsync();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"Recovery",
|
|
"录制失败产物已修复，原文件已保留。",
|
|
$"source={sourcePath}; recovered={targetPath}; originalError={originalError}",
|
|
recordTask.LiveRoomId,
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id);
|
|
SetPostProcessState(recordTask.RecordSessionId, recordTask.Id, "Completed", 100, "Recovered file is ready for manual upload");
|
|
}
|
|
else
|
|
{
|
|
var error = repair.ErrorMessage ?? "恢复输出仍无法识别有效视频流。";
|
|
recordTask.MarkFailed($"录制产物修复失败：{error}", now, recordTask.DurationSeconds);
|
|
await dbContext.SaveChangesAsync();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Recovery",
|
|
"录制失败产物修复未成功，原文件保持不变。",
|
|
$"source={sourcePath}; error={error}",
|
|
recordTask.LiveRoomId,
|
|
recordTask.RecordSessionId,
|
|
recordTask.Id);
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (_shutdownCts.IsCancellationRequested)
|
|
{
|
|
// The persisted Processing marker is intentionally kept for restart recovery.
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Artifact repair failed for task {RecordTaskId}", recordTaskId);
|
|
}
|
|
finally
|
|
{
|
|
ClearPostProcessState(recordTaskId);
|
|
}
|
|
}
|
|
|
|
private static string BuildRecoveredArtifactPath(string sourcePath, Guid recordTaskId)
|
|
{
|
|
var directory = Path.GetDirectoryName(sourcePath) ?? AppContext.BaseDirectory;
|
|
var stem = Path.GetFileNameWithoutExtension(sourcePath);
|
|
return Path.Combine(directory, $"{stem}.{recordTaskId.ToString("N")[..8]}.recovered.mp4");
|
|
}
|
|
|
|
private static bool NeedsInterruptedMp4Finalization(RecordTask recordTask)
|
|
{
|
|
if (recordTask.Status != RecordTaskStatus.Completed ||
|
|
recordTask.RecordSession is null ||
|
|
recordTask.OutputFormat != RecordOutputFormat.Mp4 ||
|
|
recordTask.RecordSession.OutputFormat != RecordOutputFormat.Mp4 ||
|
|
string.IsNullOrWhiteSpace(recordTask.OutputFilePath))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
var finalOutputPath = NormalizeAbsolutePath(recordTask.OutputFilePath);
|
|
if (HasUsableOutput(finalOutputPath, CalculateFileSize(finalOutputPath)))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
return ResolveManualFinalizeSourcePaths(recordTask, recordTask.RecordSession).Count > 0;
|
|
}
|
|
|
|
private static string ResolveManualFinalizeSourcePath(RecordTask recordTask, RecordSession recordSession)
|
|
=> ResolveManualFinalizeSourcePaths(recordTask, recordSession).FirstOrDefault() ?? string.Empty;
|
|
|
|
private static IReadOnlyList<string> ResolveManualFinalizeSourcePaths(RecordTask recordTask, RecordSession recordSession)
|
|
{
|
|
if (!string.IsNullOrWhiteSpace(recordTask.OutputFilePath))
|
|
{
|
|
var manifestPaths = ReadRecorderSegmentsManifest(recordTask.OutputFilePath);
|
|
if (manifestPaths.Count > 0)
|
|
{
|
|
return manifestPaths;
|
|
}
|
|
}
|
|
|
|
var resultPath = recordTask.Result?.FilePath;
|
|
if (!string.IsNullOrWhiteSpace(resultPath) &&
|
|
resultPath.EndsWith(".ts", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
var normalizedResultPath = NormalizeAbsolutePath(resultPath);
|
|
if (File.Exists(normalizedResultPath))
|
|
{
|
|
return [normalizedResultPath];
|
|
}
|
|
}
|
|
|
|
var defaultPath = NormalizeAbsolutePath(
|
|
GetRecorderOutputPath(
|
|
recordTask.OutputFilePath ?? resultPath ?? string.Empty,
|
|
recordSession.OutputFormat,
|
|
recordSession.SaveMode));
|
|
return File.Exists(defaultPath) ? [defaultPath] : [];
|
|
}
|
|
|
|
private async Task PersistStartupProfileAsync(SessionProcessRuntime runtime, CancellationToken cancellationToken)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"FFmpeg",
|
|
$"ffmpeg input profile={runtime.InputOptionProfile}.",
|
|
detail: $"stream={runtime.SelectedProtocol}:{runtime.SelectedQuality}; inputTransport={(runtime.RecoveryContext.UseCurlFallback ? "curl-fallback" : "native-http")}; nativeReconnect={(runtime.RecordingSettings.EnableAutoReconnect && !runtime.RecoveryContext.UseCurlFallback ? "enabled" : "disabled")}; videoCodec={runtime.SelectedVideoCodec ?? "unknown"}; recoveryEncoder={runtime.RecoveryVideoEncoder.Kind}; recoveryDevice={runtime.RecoveryVideoEncoder.DevicePath ?? "default"}; attempt={runtime.RetryAttemptCount}; compatibilityRetry={runtime.HasRetriedWithCompatibilityProfile}; refreshRetry={runtime.HasRetriedWithRefreshedStream}; alternateProtocolRetry={runtime.HasRetriedWithAlternateProtocol}; curlFallbackTried={runtime.RecoveryContext.HasTriedCurlFallback}; softwareEncoderRetry={runtime.RecoveryContext.HasRetriedWithSoftwareEncoder}",
|
|
liveRoomId: runtime.LiveRoomId,
|
|
recordSessionId: runtime.RecordSessionId,
|
|
recordTaskId: runtime.CurrentTaskId,
|
|
cancellationToken: cancellationToken);
|
|
}
|
|
|
|
private static async Task<bool> WaitForExitAsync(
|
|
SessionProcessRuntime runtime,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
var delayTask = Task.Delay(timeout, timeoutCts.Token);
|
|
var completed = await Task.WhenAny(runtime.ExitCompletion.Task, delayTask);
|
|
if (completed == runtime.ExitCompletion.Task)
|
|
{
|
|
timeoutCts.Cancel();
|
|
await runtime.ExitCompletion.Task;
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
private static async Task<bool> WaitForTransitionAsync(
|
|
SessionTransitionRuntime transition,
|
|
TimeSpan timeout,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
var delayTask = Task.Delay(timeout, timeoutCts.Token);
|
|
var completed = await Task.WhenAny(transition.Completion.Task, delayTask);
|
|
if (completed != transition.Completion.Task)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
timeoutCts.Cancel();
|
|
await transition.Completion.Task;
|
|
return true;
|
|
}
|
|
|
|
private async Task RequestStopAsync(
|
|
Guid recordSessionId,
|
|
bool markAsCompletedOnExit,
|
|
CancellationToken cancellationToken,
|
|
bool shutdownRequested = false)
|
|
{
|
|
if (!_processes.TryGetValue(recordSessionId, out var runtime))
|
|
{
|
|
if (_sessionTransitions.TryGetValue(recordSessionId, out var transition))
|
|
{
|
|
transition.RequestStop(markAsCompletedOnExit || shutdownRequested);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
if (shutdownRequested)
|
|
{
|
|
runtime.MarkShutdownRequested();
|
|
}
|
|
else
|
|
{
|
|
runtime.MarkStopRequested(markAsCompletedOnExit);
|
|
}
|
|
var process = runtime.Process;
|
|
if (process is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
if (process.HasExited)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// When curl is piping the stream, kill curl first so the pipe closes
|
|
// and FFmpeg sees EOF on stdin, causing a graceful exit.
|
|
var curlProcess = runtime.CurlProcess;
|
|
if (curlProcess is not null && !curlProcess.HasExited)
|
|
{
|
|
try
|
|
{
|
|
curlProcess.Kill(true);
|
|
}
|
|
catch (Exception curlEx)
|
|
{
|
|
_logger.LogWarning(curlEx, "Kill curl process failed for session {RecordSessionId}", recordSessionId);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
await process.StandardInput.WriteLineAsync("q");
|
|
await process.StandardInput.FlushAsync();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Stop ffmpeg process failed for session {RecordSessionId}", recordSessionId);
|
|
}
|
|
}
|
|
|
|
private void SetPostProcessState(
|
|
Guid recordSessionId,
|
|
Guid recordTaskId,
|
|
string stage,
|
|
double? progressPercent,
|
|
string? detail = null)
|
|
{
|
|
var normalizedProgress = progressPercent.HasValue
|
|
? Math.Clamp(progressPercent.Value, 0d, 100d)
|
|
: (double?)null;
|
|
_postProcessStates[recordTaskId] = new PostProcessRuntimeEntry(
|
|
recordSessionId,
|
|
new RecordTaskRuntimeState(
|
|
RecordTaskStatus.Processing,
|
|
stage,
|
|
normalizedProgress,
|
|
detail));
|
|
}
|
|
|
|
private void ClearPostProcessState(Guid recordTaskId) =>
|
|
_postProcessStates.TryRemove(recordTaskId, out _);
|
|
|
|
private bool IsTaskUnderPostProcessing(Guid recordTaskId) =>
|
|
_postProcessStates.ContainsKey(recordTaskId);
|
|
|
|
private async Task<IDisposable> AcquireTranscodeSlotAsync(int maxConcurrentTasks, CancellationToken cancellationToken)
|
|
{
|
|
var normalizedMaxConcurrentTasks = Math.Clamp(maxConcurrentTasks, 1, 16);
|
|
TaskCompletionSource<IDisposable> waiter;
|
|
|
|
lock (_transcodeConcurrencyLock)
|
|
{
|
|
_maxConcurrentTranscodeTasks = normalizedMaxConcurrentTasks;
|
|
if (_transcodeWaiters.Count == 0 && _activeTranscodeTasks < _maxConcurrentTranscodeTasks)
|
|
{
|
|
_activeTranscodeTasks++;
|
|
return new TranscodeSlotLease(this);
|
|
}
|
|
|
|
waiter = new TaskCompletionSource<IDisposable>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
_transcodeWaiters.Enqueue(waiter);
|
|
}
|
|
|
|
using var cancellationRegistration = cancellationToken.Register(
|
|
static state => ((TaskCompletionSource<IDisposable>)state!).TrySetCanceled(),
|
|
waiter);
|
|
return await waiter.Task.ConfigureAwait(false);
|
|
}
|
|
|
|
private void ReleaseTranscodeSlot()
|
|
{
|
|
lock (_transcodeConcurrencyLock)
|
|
{
|
|
if (_activeTranscodeTasks > 0)
|
|
{
|
|
_activeTranscodeTasks--;
|
|
}
|
|
|
|
while (_activeTranscodeTasks < _maxConcurrentTranscodeTasks && _transcodeWaiters.Count > 0)
|
|
{
|
|
var waiter = _transcodeWaiters.Dequeue();
|
|
if (waiter.Task.IsCompleted)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
_activeTranscodeTasks++;
|
|
if (waiter.TrySetResult(new TranscodeSlotLease(this)))
|
|
{
|
|
return;
|
|
}
|
|
|
|
_activeTranscodeTasks--;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Records a startup failure for the given live room and returns the backoff delay
|
|
/// that should be applied before the next re-poll attempt. Uses exponential backoff
|
|
/// to prevent tight fail→retry→fail→re-poll loops from flooding notifications.
|
|
/// </summary>
|
|
private TimeSpan RecordStartupFailure(Guid liveRoomId)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
var state = _roomStartupFailureStates.AddOrUpdate(
|
|
liveRoomId,
|
|
_ => new RoomStartupFailureState { ConsecutiveFailures = 1, FirstFailureAt = now, LastFailureAt = now },
|
|
(_, existing) =>
|
|
{
|
|
existing.ConsecutiveFailures++;
|
|
existing.LastFailureAt = now;
|
|
return existing;
|
|
});
|
|
|
|
var backoffSeconds = StartupFailureBaseBackoff.TotalSeconds * Math.Pow(2, Math.Min(state.ConsecutiveFailures - 1, 5));
|
|
var backoff = TimeSpan.FromSeconds(Math.Min(backoffSeconds, StartupFailureMaxBackoff.TotalSeconds));
|
|
_logger.LogWarning(
|
|
"Startup failure backoff for room {LiveRoomId}: {ConsecutiveFailures} consecutive failures, next poll delayed by {BackoffSeconds:F0}s",
|
|
liveRoomId, state.ConsecutiveFailures, backoff.TotalSeconds);
|
|
return backoff;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Resets the startup failure backoff counter when a session successfully opens its first segment.
|
|
/// </summary>
|
|
private void ResetStartupFailureBackoff(Guid liveRoomId)
|
|
{
|
|
if (_roomStartupFailureStates.TryRemove(liveRoomId, out var state) && state.ConsecutiveFailures > 1)
|
|
{
|
|
_logger.LogInformation(
|
|
"Startup failure backoff reset for room {LiveRoomId} after {ConsecutiveFailures} failures",
|
|
liveRoomId, state.ConsecutiveFailures);
|
|
}
|
|
}
|
|
|
|
private TimeSpan RecordRuntimeFailure(Guid liveRoomId)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
var state = _roomRuntimeFailureStates.AddOrUpdate(
|
|
liveRoomId,
|
|
_ => new RoomStartupFailureState { ConsecutiveFailures = 1, FirstFailureAt = now, LastFailureAt = now },
|
|
(_, existing) =>
|
|
{
|
|
existing.ConsecutiveFailures++;
|
|
existing.LastFailureAt = now;
|
|
return existing;
|
|
});
|
|
|
|
var backoffSeconds = RuntimeFailureBaseBackoff.TotalSeconds *
|
|
Math.Pow(2, Math.Min(state.ConsecutiveFailures - 1, 5));
|
|
var backoff = TimeSpan.FromSeconds(Math.Min(backoffSeconds, RuntimeFailureMaxBackoff.TotalSeconds));
|
|
_logger.LogWarning(
|
|
"Short runtime failure backoff for room {LiveRoomId}: {ConsecutiveFailures} consecutive failures, next poll delayed by {BackoffSeconds:F0}s",
|
|
liveRoomId,
|
|
state.ConsecutiveFailures,
|
|
backoff.TotalSeconds);
|
|
return backoff;
|
|
}
|
|
|
|
private void ResetRuntimeFailureBackoff(Guid liveRoomId)
|
|
{
|
|
if (_roomRuntimeFailureStates.TryRemove(liveRoomId, out var state) && state.ConsecutiveFailures > 1)
|
|
{
|
|
_logger.LogInformation(
|
|
"Short runtime failure backoff reset for room {LiveRoomId} after {ConsecutiveFailures} failures",
|
|
liveRoomId,
|
|
state.ConsecutiveFailures);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns true if the failure notification for this room should be throttled
|
|
/// (i.e., at most one notification per <see cref="StartupFailureNotificationCooldown"/>).
|
|
/// </summary>
|
|
private bool ShouldThrottleStartupFailureNotification(Guid liveRoomId)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
var state = _roomStartupFailureStates.GetOrAdd(
|
|
liveRoomId,
|
|
_ => new RoomStartupFailureState { ConsecutiveFailures = 0, FirstFailureAt = now, LastFailureAt = now });
|
|
|
|
// Always allow the first failure notification through
|
|
if (state.LastNotificationAt is null)
|
|
{
|
|
state.LastNotificationAt = now;
|
|
return false;
|
|
}
|
|
|
|
if (now - state.LastNotificationAt.Value < StartupFailureNotificationCooldown)
|
|
{
|
|
_logger.LogWarning(
|
|
"Suppressed duplicate startup failure notification for room {LiveRoomId}. " +
|
|
"Last notification was at {LastNotificationAt} (cooldown={CooldownMinutes}min)",
|
|
liveRoomId, state.LastNotificationAt, StartupFailureNotificationCooldown.TotalMinutes);
|
|
return true;
|
|
}
|
|
|
|
state.LastNotificationAt = now;
|
|
return false;
|
|
}
|
|
|
|
private sealed class RoomStartupFailureState
|
|
{
|
|
public int ConsecutiveFailures;
|
|
public DateTimeOffset FirstFailureAt;
|
|
public DateTimeOffset LastFailureAt;
|
|
public DateTimeOffset? LastNotificationAt;
|
|
}
|
|
|
|
private sealed record PostProcessRuntimeEntry(Guid RecordSessionId, RecordTaskRuntimeState State);
|
|
|
|
private sealed class TranscodeSlotLease : IDisposable
|
|
{
|
|
private readonly FfmpegService _owner;
|
|
private int _disposed;
|
|
|
|
public TranscodeSlotLease(FfmpegService owner)
|
|
{
|
|
_owner = owner;
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (Interlocked.Exchange(ref _disposed, 1) == 0)
|
|
{
|
|
_owner.ReleaseTranscodeSlot();
|
|
}
|
|
}
|
|
}
|
|
}
|