feat: improve recording recovery and upload workflow

This commit is contained in:
2026-08-10 11:14:34 +08:00
parent e0d3969e46
commit 19d358b312
58 changed files with 5979 additions and 586 deletions
@@ -120,12 +120,14 @@ public interface IRecordResultRepository
Task<List<(RecordResult Result, RecordTask Task)>> ListUploadStatusAsync(
RecordArtifactUploadStatus? uploadStatusFilter,
string? query,
int skip,
int take,
CancellationToken cancellationToken = default);
Task<int> CountUploadStatusAsync(
RecordArtifactUploadStatus? uploadStatusFilter,
string? query,
CancellationToken cancellationToken = default);
Task AddAsync(RecordResult recordResult, CancellationToken cancellationToken = default);
@@ -47,6 +47,12 @@ public interface IFfmpegService
Task<int> ResumePausedFinalizationsAsync(CancellationToken cancellationToken = default);
Task<bool> StartArtifactRepairAsync(Guid recordTaskId, CancellationToken cancellationToken = default);
Task<int> ResumeArtifactRepairsAsync(CancellationToken cancellationToken = default);
Task<int> ResumeRecoveringSessionsAsync(CancellationToken cancellationToken = default);
bool IsRunning(Guid recordSessionId);
IReadOnlyDictionary<Guid, RecordTaskRuntimeState> GetTaskRuntimeStates(IReadOnlyCollection<Guid> recordTaskIds);
@@ -12,6 +12,8 @@ public sealed class RecordSessionDto
public required string LiveRoomTitle { get; init; }
public string? AnchorName { get; init; }
public required LivePlatformType Platform { get; init; }
public required string RoomId { get; init; }
@@ -32,6 +34,10 @@ public sealed class RecordSessionDto
public string? ErrorMessage { get; init; }
public bool IsRecovering { get; init; }
public string? RecoveryReason { get; init; }
public DateTimeOffset CreatedAt { get; init; }
public DateTimeOffset? StartedAt { get; init; }
@@ -85,6 +85,10 @@ public sealed class RecordTaskDto
public double? DurationSeconds { get; init; }
public bool IsHiddenArtifactSource { get; init; }
public Guid? MergedIntoRecordTaskId { get; init; }
public RecordArtifactUploadStatus? UploadStatus { get; init; }
public string? PostProcessStage { get; init; }
@@ -171,6 +175,13 @@ public sealed class RecordArtifactUploadBatchResultDto
public required IReadOnlyList<RecordArtifactUploadItemResultDto> Items { get; init; }
}
public sealed class RetryRecordArtifactUploadsRequest
{
public RecordArtifactUploadStatus? UploadStatus { get; set; }
public string? Query { get; set; }
}
public sealed class ManualSegmentCompletedTriggerResultDto
{
public Guid RecordTaskId { get; init; }
@@ -250,4 +261,21 @@ public sealed class UploadTaskListResponse
public int UploadingCount { get; init; }
public int WaitingRetryCount { get; init; }
public int MatchingRetryableCount { get; init; }
public required UploadQueueHealthDto QueueHealth { get; init; }
}
public sealed class UploadQueueHealthDto
{
public required string State { get; init; }
public bool IsPaused { get; init; }
public string? Reason { get; init; }
public DateTimeOffset? RetryAt { get; init; }
public DateTimeOffset? LastErrorAt { get; init; }
}
@@ -9,6 +9,21 @@ public sealed class RecoveryOverviewDto
public required IReadOnlyList<RecoverableLiveRoomDto> LiveRooms { get; init; }
public required IReadOnlyList<RecoverableFinalizationDto> Finalizations { get; init; }
public required IReadOnlyList<MergedArtifactRecordDto> MergedArtifacts { get; init; }
}
public sealed class MergedArtifactRecordDto
{
public Guid SourceRecordTaskId { get; init; }
public Guid RecordSessionId { get; init; }
public Guid? MergedIntoRecordTaskId { get; init; }
public string? SourceVideoPath { get; init; }
public string? MergedVideoPath { get; init; }
public string? RecoveryDirectory { get; init; }
public string? ManifestPath { get; init; }
public double? SourceDurationSeconds { get; init; }
public DateTimeOffset CreatedAt { get; init; }
}
public sealed class StorageGuardStatusDto
@@ -104,3 +119,58 @@ public sealed class RecoveryActionResultDto
public required IReadOnlyList<string> Messages { get; init; }
}
public sealed class RecordingFailureListResponse
{
public required IReadOnlyList<RecordingFailureItemDto> Items { get; init; }
public int TotalCount { get; init; }
}
public sealed class RecordingFailureItemDto
{
public Guid RecordTaskId { get; init; }
public Guid RecordSessionId { get; init; }
public Guid LiveRoomId { get; init; }
public required string LiveRoomTitle { get; init; }
public required string RoomId { get; init; }
public required string PlatformName { get; init; }
public int SegmentIndex { get; init; }
public required string FailureKind { get; init; }
public required string FailureLabel { get; init; }
public required string RecommendedAction { get; init; }
public string? ErrorMessage { get; init; }
public string? FilePath { get; init; }
public long? FileSizeBytes { get; init; }
public double? DurationSeconds { get; init; }
public bool FileExists { get; init; }
public bool CanAccept { get; init; }
public bool CanRepair { get; init; }
public bool CanRetryRoom { get; init; }
public bool IsRepairing { get; init; }
public DateTimeOffset CreatedAt { get; init; }
}
public sealed class AcceptRecordingArtifactRequest
{
public bool ConfirmShortArtifact { get; set; }
}
@@ -27,6 +27,8 @@ internal static class RecordModelMapper
StartedAt = recordTask.StartedAt,
EndedAt = recordTask.EndedAt,
DurationSeconds = recordTask.DurationSeconds,
IsHiddenArtifactSource = recordTask.IsHiddenArtifactSource,
MergedIntoRecordTaskId = recordTask.MergedIntoRecordTaskId,
UploadStatus = recordTask.Result?.UploadStatus,
PostProcessStage = runtimeState?.Stage,
PostProcessProgressPercent = runtimeState?.ProgressPercent,
@@ -58,6 +60,7 @@ internal static class RecordModelMapper
IReadOnlyDictionary<Guid, RecordTaskRuntimeState>? runtimeStates = null)
{
var orderedTasks = recordSession.RecordTasks
.Where(static item => !item.IsHiddenArtifactSource)
.OrderBy(static item => item.SegmentIndex)
.ThenBy(static item => item.CreatedAt)
.ToList();
@@ -73,6 +76,7 @@ internal static class RecordModelMapper
Id = recordSession.Id,
LiveRoomId = recordSession.LiveRoomId,
LiveRoomTitle = recordSession.LiveRoom?.Title ?? recordSession.LiveRoom?.AnchorName ?? recordSession.LiveRoom?.RoomId ?? "Unknown Room",
AnchorName = recordSession.LiveRoom?.AnchorName,
Platform = recordSession.LiveRoom?.Platform ?? LivePlatformType.Unknown,
RoomId = recordSession.LiveRoom?.RoomId ?? string.Empty,
Status = recordSession.Status,
@@ -83,6 +87,12 @@ internal static class RecordModelMapper
SegmentCount = Math.Max(recordSession.SegmentCount, orderedTasks.Count),
RecorderProcessId = recordSession.RecorderProcessId,
ErrorMessage = recordSession.ErrorMessage,
IsRecovering = recordSession.Status == RecordSessionStatus.Starting &&
recordSession.ErrorMessage?.StartsWith("[runtime-recovery]", StringComparison.Ordinal) == true,
RecoveryReason = recordSession.Status == RecordSessionStatus.Starting &&
recordSession.ErrorMessage?.StartsWith("[runtime-recovery]", StringComparison.Ordinal) == true
? recordSession.ErrorMessage["[runtime-recovery]".Length..].TrimStart(' ', ':', ';')
: null,
CreatedAt = recordSession.CreatedAt,
StartedAt = recordSession.StartedAt,
EndedAt = recordSession.EndedAt,
@@ -126,6 +136,8 @@ internal static class RecordModelMapper
StartedAt = recordTask.StartedAt,
EndedAt = recordTask.EndedAt,
DurationSeconds = recordTask.DurationSeconds,
IsHiddenArtifactSource = recordTask.IsHiddenArtifactSource,
MergedIntoRecordTaskId = recordTask.MergedIntoRecordTaskId,
UploadStatus = recordTask.Result?.UploadStatus,
PostProcessStage = runtimeState?.Stage,
PostProcessProgressPercent = runtimeState?.ProgressPercent,
@@ -49,6 +49,16 @@ public sealed class RecordCompletionDispatch
UpdatedAt = updatedAt;
}
public void MarkSkipped(string reason, DateTimeOffset updatedAt)
{
ScriptDispatched = true;
UploadDispatched = true;
LastError = string.IsNullOrWhiteSpace(reason) ? null : reason.Trim();
NextAttemptAt = null;
CompletedAt = updatedAt;
UpdatedAt = updatedAt;
}
private void CompleteIfReady(DateTimeOffset updatedAt)
{
if (!ScriptDispatched || !UploadDispatched)
@@ -149,6 +149,17 @@ public class RecordResult
DeletedLocalFilesAfterUpload = false;
}
public void ResetUploadForRecoveredArtifact()
{
UploadStatus = RecordArtifactUploadStatus.NotUploaded;
LastUploadProvider = null;
RemoteVideoPath = null;
RemoteDanmakuPath = null;
LastUploadedAt = null;
UploadErrorMessage = null;
DeletedLocalFilesAfterUpload = false;
}
private static string? NormalizeNullable(string? value) =>
string.IsNullOrWhiteSpace(value) ? null : value.Trim();
}
@@ -81,6 +81,14 @@ public class RecordSession
UpdatedAt = updatedAt;
}
public void MarkRecovering(string reason, DateTimeOffset updatedAt)
{
Status = RecordSessionStatus.Starting;
RecorderProcessId = null;
ErrorMessage = string.IsNullOrWhiteSpace(reason) ? "[runtime-recovery]" : reason.Trim();
UpdatedAt = updatedAt;
}
public void MarkStopping(DateTimeOffset updatedAt)
{
Status = RecordSessionStatus.Stopping;
@@ -67,6 +67,10 @@ public class RecordTask
public RecordUploadJob? UploadJob { get; private set; }
public bool IsHiddenArtifactSource { get; private set; }
public Guid? MergedIntoRecordTaskId { get; private set; }
public void AssignToSession(Guid recordSessionId, int segmentIndex, DateTimeOffset updatedAt)
{
RecordSessionId = recordSessionId;
@@ -153,4 +157,25 @@ public class RecordTask
RecorderProcessId = null;
UpdatedAt = endedAt;
}
public void MarkMergedSource(Guid mergedIntoRecordTaskId, DateTimeOffset updatedAt)
{
if (mergedIntoRecordTaskId == Guid.Empty || mergedIntoRecordTaskId == Id)
{
throw new ArgumentException("Merged target must be another recording task.", nameof(mergedIntoRecordTaskId));
}
IsHiddenArtifactSource = true;
MergedIntoRecordTaskId = mergedIntoRecordTaskId;
RecorderProcessId = null;
UpdatedAt = updatedAt;
}
public void MarkRecoveryOnly(DateTimeOffset updatedAt)
{
IsHiddenArtifactSource = true;
MergedIntoRecordTaskId = null;
RecorderProcessId = null;
UpdatedAt = updatedAt;
}
}
@@ -136,6 +136,34 @@ public sealed class RecordUploadJob
UpdatedAt = updatedAt;
}
public void RequestImmediateRetry(DateTimeOffset requestedAt)
{
if (Status is not (RecordArtifactUploadStatus.Failed or RecordArtifactUploadStatus.WaitingRetry))
{
throw new InvalidOperationException("Only failed or waiting upload jobs can be retried immediately.");
}
Status = RecordArtifactUploadStatus.Queued;
AttemptCount = 0;
NextAttemptAt = null;
ErrorMessage = null;
CompletedAt = null;
RequestedAt = requestedAt;
UpdatedAt = requestedAt;
}
public void SuspendForProviderFailure(string? errorMessage, DateTimeOffset updatedAt, bool rollbackAttempt)
{
Status = RecordArtifactUploadStatus.WaitingRetry;
if (rollbackAttempt)
{
AttemptCount = Math.Max(0, AttemptCount - 1);
}
NextAttemptAt = null;
ErrorMessage = NormalizeNullable(errorMessage);
UpdatedAt = updatedAt;
}
public void BeginAttempt(DateTimeOffset updatedAt)
{
MarkProcessing(updatedAt);
@@ -75,6 +75,8 @@ public sealed class LiveRecorderDbContext : DbContext, IUnitOfWork
builder.Property(static x => x.StreamUrl).HasMaxLength(2048);
builder.Property(static x => x.OutputFilePath).HasMaxLength(2048);
builder.Property(static x => x.ErrorMessage).HasMaxLength(2048);
builder.Property(static x => x.IsHiddenArtifactSource).HasDefaultValue(false);
builder.HasIndex(static x => x.MergedIntoRecordTaskId);
builder.HasOne(static x => x.LiveRoom)
.WithMany(static x => x.RecordTasks)
.HasForeignKey(static x => x.LiveRoomId)
@@ -228,6 +230,7 @@ public sealed class LiveRecorderDbContext : DbContext, IUnitOfWork
.Where(entry => entry.State is EntityState.Added or EntityState.Modified)
.Select(static entry => entry.Entity)
.Where(static task => task.Status is Domain.Enums.RecordTaskStatus.Completed or Domain.Enums.RecordTaskStatus.Stopped)
.Where(static task => !task.IsHiddenArtifactSource)
.ToArray();
foreach (var task in terminalTasks)
{
@@ -239,7 +242,7 @@ public sealed class LiveRecorderDbContext : DbContext, IUnitOfWork
.FirstOrDefaultAsync(item => item.RecordTaskId == task.Id, cancellationToken);
var path = result?.FilePath;
if (string.IsNullOrWhiteSpace(path) ||
result?.DurationSeconds is null or < 5 ||
result?.DurationSeconds is null or <= 0 ||
(task.OutputFormat == Domain.Enums.RecordOutputFormat.Mp4 && !path.EndsWith(".mp4", StringComparison.OrdinalIgnoreCase)))
{
continue;
@@ -0,0 +1,40 @@
using LiveRecorder.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace LiveRecorder.Infrastructure.Persistence.Migrations;
[DbContext(typeof(LiveRecorderDbContext))]
[Migration("20260805143000_AddMergedArtifactTracking")]
public sealed class AddMergedArtifactTracking : Migration
{
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.AddColumn<bool>(
name: "IsHiddenArtifactSource",
table: "RecordTasks",
type: "boolean",
nullable: false,
defaultValue: false);
migrationBuilder.AddColumn<Guid>(
name: "MergedIntoRecordTaskId",
table: "RecordTasks",
type: "uuid",
nullable: true);
migrationBuilder.CreateIndex(
name: "IX_RecordTasks_MergedIntoRecordTaskId",
table: "RecordTasks",
column: "MergedIntoRecordTaskId");
}
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropIndex(name: "IX_RecordTasks_MergedIntoRecordTaskId", table: "RecordTasks");
migrationBuilder.DropColumn(name: "IsHiddenArtifactSource", table: "RecordTasks");
migrationBuilder.DropColumn(name: "MergedIntoRecordTaskId", table: "RecordTasks");
}
}
@@ -192,6 +192,11 @@ namespace LiveRecorder.Infrastructure.Persistence.Migrations
.HasMaxLength(2048)
.HasColumnType("character varying(2048)");
b.Property<bool>("IsHiddenArtifactSource")
.ValueGeneratedOnAdd()
.HasColumnType("boolean")
.HasDefaultValue(false);
b.Property<string>("LastAutoStartDecisionSummary")
.HasMaxLength(256)
.HasColumnType("character varying(256)");
@@ -411,6 +416,9 @@ namespace LiveRecorder.Infrastructure.Persistence.Migrations
b.Property<Guid>("LiveRoomId")
.HasColumnType("uuid");
b.Property<Guid?>("MergedIntoRecordTaskId")
.HasColumnType("uuid");
b.Property<int>("OutputFormat")
.HasColumnType("integer");
@@ -514,6 +522,8 @@ namespace LiveRecorder.Infrastructure.Persistence.Migrations
b.HasIndex("LiveRoomId");
b.HasIndex("MergedIntoRecordTaskId");
b.HasIndex("RecordSessionId", "SegmentIndex")
.IsUnique();
@@ -103,6 +103,8 @@ public sealed class RecordTaskRepository : IRecordTaskRepository
.Include(item => item.UploadJob)
.AsNoTracking();
query = query.Where(static item => !item.IsHiddenArtifactSource);
if (liveRoomId.HasValue)
{
query = query.Where(item => item.LiveRoomId == liveRoomId.Value);
@@ -143,11 +145,11 @@ public sealed class RecordTaskRepository : IRecordTaskRepository
public Task<double> SumDurationSecondsAsync(DateTimeOffset startedFrom, DateTimeOffset startedTo, CancellationToken cancellationToken = default) =>
_dbContext.RecordTasks
.Include(item => item.RecordSession)
.Where(item => item.RecordSession != null && item.RecordSession.StartedAt >= startedFrom && item.RecordSession.StartedAt <= startedTo)
.Where(item => !item.IsHiddenArtifactSource && item.RecordSession != null && item.RecordSession.StartedAt >= startedFrom && item.RecordSession.StartedAt <= startedTo)
.SumAsync(item => item.DurationSeconds ?? 0, cancellationToken);
public Task<int> CountByStatusAsync(RecordTaskStatus status, CancellationToken cancellationToken = default) =>
_dbContext.RecordTasks.CountAsync(item => item.Status == status, cancellationToken);
_dbContext.RecordTasks.CountAsync(item => !item.IsHiddenArtifactSource && item.Status == status, cancellationToken);
public Task AddAsync(RecordTask recordTask, CancellationToken cancellationToken = default) =>
_dbContext.RecordTasks.AddAsync(recordTask, cancellationToken).AsTask();
@@ -350,10 +352,11 @@ public sealed class RecordSessionRepository : IRecordSessionRepository
cancellationToken);
var totalTaskCount = await sessions
.SelectMany(static item => item.RecordTasks)
.Where(static item => !item.IsHiddenArtifactSource)
.CountAsync(cancellationToken);
var totalDanmakuCount = await sessions
.SelectMany(static item => item.RecordTasks)
.Where(static item => item.Result != null)
.Where(static item => !item.IsHiddenArtifactSource && item.Result != null)
.SumAsync(static item => (int?)item.Result!.DanmakuMessageCount, cancellationToken) ?? 0;
return new RecordSessionOverviewTotals(
@@ -530,6 +533,7 @@ public sealed class RecordResultRepository : IRecordResultRepository
public async Task<List<(RecordResult Result, RecordTask Task)>> ListUploadStatusAsync(
RecordArtifactUploadStatus? uploadStatusFilter,
string? searchQuery,
int skip,
int take,
CancellationToken cancellationToken = default)
@@ -546,6 +550,8 @@ public sealed class RecordResultRepository : IRecordResultRepository
query = query.Where(item => item.UploadStatus == uploadStatusFilter.Value);
}
query = ApplyUploadSearch(query, searchQuery);
var results = await query
.OrderByDescending(item => item.CreatedAt)
.Skip(skip)
@@ -560,6 +566,7 @@ public sealed class RecordResultRepository : IRecordResultRepository
public Task<int> CountUploadStatusAsync(
RecordArtifactUploadStatus? uploadStatusFilter,
string? searchQuery,
CancellationToken cancellationToken = default)
{
var query = _dbContext.RecordResults.AsQueryable();
@@ -569,9 +576,33 @@ public sealed class RecordResultRepository : IRecordResultRepository
query = query.Where(item => item.UploadStatus == uploadStatusFilter.Value);
}
query = ApplyUploadSearch(query, searchQuery);
return query.CountAsync(cancellationToken);
}
private static IQueryable<RecordResult> ApplyUploadSearch(
IQueryable<RecordResult> query,
string? searchQuery)
{
var normalized = searchQuery?.Trim().ToLowerInvariant();
if (string.IsNullOrWhiteSpace(normalized))
{
return query;
}
var parsedTaskId = Guid.TryParse(normalized, out var taskId) ? taskId : (Guid?)null;
return query.Where(item =>
parsedTaskId.HasValue && item.RecordTaskId == parsedTaskId.Value ||
item.FilePath.ToLower().Contains(normalized) ||
item.RemoteVideoPath != null && item.RemoteVideoPath.ToLower().Contains(normalized) ||
item.UploadErrorMessage != null && item.UploadErrorMessage.ToLower().Contains(normalized) ||
item.RecordTask != null && item.RecordTask.LiveRoom != null &&
((item.RecordTask.LiveRoom.Title != null && item.RecordTask.LiveRoom.Title.ToLower().Contains(normalized)) ||
item.RecordTask.LiveRoom.AnchorName != null && item.RecordTask.LiveRoom.AnchorName.ToLower().Contains(normalized) ||
item.RecordTask.LiveRoom.RoomId.ToLower().Contains(normalized)));
}
public Task AddAsync(RecordResult recordResult, CancellationToken cancellationToken = default) =>
_dbContext.RecordResults.AddAsync(recordResult, cancellationToken).AsTask();
@@ -14,15 +14,18 @@ public sealed class CompletionDispatchService
private readonly LiveRecorderDbContext _dbContext;
private readonly IEventScriptService _eventScriptService;
private readonly RecordUploadService _recordUploadService;
private readonly ShortFragmentConsolidationService _shortFragmentConsolidationService;
public CompletionDispatchService(
LiveRecorderDbContext dbContext,
IEventScriptService eventScriptService,
RecordUploadService recordUploadService)
RecordUploadService recordUploadService,
ShortFragmentConsolidationService shortFragmentConsolidationService)
{
_dbContext = dbContext;
_eventScriptService = eventScriptService;
_recordUploadService = recordUploadService;
_shortFragmentConsolidationService = shortFragmentConsolidationService;
}
public async Task<bool> TryDispatchNextAsync(CancellationToken cancellationToken = default)
@@ -69,6 +72,25 @@ public sealed class CompletionDispatchService
try
{
var consolidation = await _shortFragmentConsolidationService.PrepareDispatchAsync(task.Id, cancellationToken);
if (consolidation.SkipDispatch)
{
dispatch.MarkSkipped(consolidation.Reason ?? "该分片无需独立分发。", DateTimeOffset.UtcNow);
await _dbContext.SaveChangesAsync(cancellationToken);
return;
}
if (!consolidation.CanDispatch)
{
var retryAt = DateTimeOffset.UtcNow.Add(consolidation.RetryDelay <= TimeSpan.Zero ? TimeSpan.FromSeconds(2) : consolidation.RetryDelay);
dispatch.ScheduleRetry(consolidation.Reason ?? "等待短分片判定。", retryAt, DateTimeOffset.UtcNow);
await _dbContext.SaveChangesAsync(cancellationToken);
return;
}
await _dbContext.Entry(task).ReloadAsync(cancellationToken);
await _dbContext.Entry(task).Reference(item => item.Result).LoadAsync(cancellationToken);
if (!dispatch.ScriptDispatched)
{
var scriptResult = await _eventScriptService.RunSegmentCompletedAsync(
@@ -19,6 +19,7 @@ namespace LiveRecorder.Infrastructure.Services;
public sealed partial class FfmpegService
{
private const int MaxInSessionRetryAttempts = 3;
private const string RuntimeRecoveryMarker = "[runtime-recovery]";
private const string ShortUnexpectedExitArtifactError =
"Unexpected recorder exit produced only a short fragment. The file was kept locally and excluded from automatic upload.";
internal const string FfprobeUnreadableArtifactError = "ffprobe could not read the recorded media file.";
@@ -29,6 +30,15 @@ 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 static readonly TimeSpan OfflineConfirmationDelay = TimeSpan.FromSeconds(10);
private static readonly TimeSpan[] RuntimeRecoveryBackoff =
[
TimeSpan.FromSeconds(5),
TimeSpan.FromSeconds(15),
TimeSpan.FromSeconds(30),
TimeSpan.FromSeconds(60),
TimeSpan.FromSeconds(120)
];
private async Task HandleProcessOutputAsync(SessionProcessRuntime runtime, string? line, bool isError)
{
@@ -221,6 +231,9 @@ public sealed partial class FfmpegService
internal static bool IsMeaningfulUnexpectedExitArtifact(double? durationSeconds, long? fileSizeBytes) =>
durationSeconds >= MinimumUnexpectedExitArtifactDuration.TotalSeconds;
internal static bool CanResetInSessionRetryBudget(DateTimeOffset? processStartedAt, DateTimeOffset observedAt) =>
processStartedAt.HasValue && observedAt - processStartedAt.Value >= StableRuntimeResetThreshold;
private static bool IsRuntimeSourceFailureLine(string line)
{
if (line.Contains("Will reconnect at", StringComparison.OrdinalIgnoreCase) ||
@@ -567,6 +580,19 @@ public sealed partial class FfmpegService
private async Task HandleProcessExitedAsync(SessionProcessRuntime runtime, Process process)
{
var transition = new SessionTransitionRuntime(runtime.RecordSessionId);
if (!_sessionTransitions.TryAdd(runtime.RecordSessionId, transition))
{
transition.Dispose();
if (_sessionTransitions.TryGetValue(runtime.RecordSessionId, out var activeTransition))
{
await activeTransition.Completion.Task;
}
await HandleProcessExitedAsync(runtime, process);
return;
}
_processes.TryRemove(runtime.RecordSessionId, out _);
try
@@ -604,7 +630,16 @@ public sealed partial class FfmpegService
return;
}
if (await TryRecoverUnexpectedExitAsync(runtime, activeDanmakuSummary))
if (!runtime.HasOpenedFirstSegment &&
!runtime.StopRequested &&
!runtime.CompletionRequested &&
!runtime.ShutdownRequested &&
await TryRecoverStartupUntilAvailableAsync(runtime, transition))
{
return;
}
if (await TryRecoverUnexpectedExitAsync(runtime, activeDanmakuSummary, transition))
{
return;
}
@@ -617,12 +652,110 @@ public sealed partial class FfmpegService
}
finally
{
_sessionTransitions.TryRemove(runtime.RecordSessionId, out _);
transition.Completion.TrySetResult(true);
transition.Dispose();
runtime.ExitCompletion.TrySetResult(true);
runtime.Dispose();
process.Dispose();
}
}
private async Task<bool> TryRecoverStartupUntilAvailableAsync(
SessionProcessRuntime runtime,
SessionTransitionRuntime transition)
{
var attempt = transition.NextAttempt(Math.Max(1, runtime.RetryAttemptCount + 1));
while (!transition.Token.IsCancellationRequested)
{
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 == runtime.RecordSessionId, transition.Token);
if (session?.LiveRoom is null || !IsActiveSessionStatus(session.Status))
{
return false;
}
var currentTask = session.RecordTasks.FirstOrDefault(item => item.Id == runtime.CurrentTaskId)
?? session.RecordTasks.OrderByDescending(item => item.SegmentIndex).FirstOrDefault();
if (currentTask is null)
{
return false;
}
var delay = GetRuntimeRecoveryDelay(attempt);
session.MarkRecovering(
$"{RuntimeRecoveryMarker} attempt={attempt}; startup did not open a media segment",
DateTimeOffset.UtcNow);
currentTask.MarkStarting(runtime.StreamUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, DateTimeOffset.UtcNow);
await dbContext.SaveChangesAsync(transition.Token);
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
await logService.WriteAsync(
SystemLogLevel.Warning,
"FFmpeg",
$"Recorder startup is recovering in the same session (attempt {attempt}).",
$"delaySeconds={delay.TotalSeconds:0}; ffmpegExit={runtime.Process?.ExitCode}; curlExit={runtime.CurlExitCode?.ToString() ?? "unknown"}; output={runtime.GetRecentOutputSummary()}; curl={runtime.GetRecentCurlErrorSummary()}",
session.LiveRoomId,
session.Id,
currentTask.Id,
transition.Token);
await Task.Delay(delay, transition.Token);
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId, transition.Token);
if (!liveStatus.IsLive)
{
await Task.Delay(OfflineConfirmationDelay, transition.Token);
liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId, transition.Token);
}
if (!liveStatus.IsLive)
{
runtime.MarkStopRequested(markAsCompletedOnExit: true);
return false;
}
var stream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality, transition.Token);
var selected = SelectRetryStreamForCurrentSession(stream, runtime);
var context = runtime.RecoveryContext with
{
AttemptCount = attempt,
HasRetriedWithRefreshedStream = true
};
session.MarkStarting(selected.SelectedUrl, session.OutputPathPattern ?? runtime.OutputPathPattern, DateTimeOffset.UtcNow);
currentTask.MarkStarting(selected.SelectedUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, DateTimeOffset.UtcNow);
await dbContext.SaveChangesAsync(transition.Token);
await StartInternalAsync(session, currentTask, selected, runtime.RecordingSettings, context, transition.Token);
var restartedAt = DateTimeOffset.UtcNow;
session.MarkRunning(restartedAt);
currentTask.MarkRunning(restartedAt);
await dbContext.SaveChangesAsync(transition.Token);
return true;
}
catch (OperationCanceledException) when (transition.Token.IsCancellationRequested)
{
runtime.MarkStopRequested(transition.MarkAsCompletedOnExit);
return false;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Recorder startup recovery attempt {Attempt} failed for session {RecordSessionId}", attempt, runtime.RecordSessionId);
attempt = transition.NextAttempt(attempt + 1);
}
}
return false;
}
private async Task<bool> TryRecoverStartupFailureAsync(SessionProcessRuntime runtime)
{
if (runtime.ShutdownRequested ||
@@ -885,14 +1018,14 @@ public sealed partial class FfmpegService
private async Task<bool> TryRecoverUnexpectedExitAsync(
SessionProcessRuntime runtime,
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary)
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary,
SessionTransitionRuntime transition)
{
if (!runtime.HasOpenedFirstSegment ||
runtime.StopRequested ||
runtime.CompletionRequested ||
runtime.ShutdownRequested ||
runtime.SaveMode != RecordSaveMode.Segmented ||
runtime.RetryAttemptCount >= MaxInSessionRetryAttempts)
runtime.SaveMode != RecordSaveMode.Segmented)
{
return false;
}
@@ -929,15 +1062,20 @@ public sealed partial class FfmpegService
var observedAt = DateTimeOffset.UtcNow;
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId);
if (!liveStatus.IsLive)
{
await Task.Delay(OfflineConfirmationDelay, transition.Token);
liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId, transition.Token);
}
await liveRoomStatusService.ApplySnapshotAsync(session.LiveRoom, liveStatus, observedAt);
await dbContext.SaveChangesAsync();
if (!liveStatus.IsLive)
{
runtime.MarkStopRequested(markAsCompletedOnExit: true);
return false;
}
var retryAttempt = runtime.RetryAttemptCount + 1;
var refreshedStream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality);
var retryStream = SelectRetryStreamForCurrentSession(refreshedStream, runtime);
var nextSegmentIndex = Math.Max(currentTask.SegmentIndex + 1, session.ActiveSegmentIndex + 1);
@@ -957,11 +1095,17 @@ public sealed partial class FfmpegService
await dbContext.SaveChangesAsync();
var retryInputOptionProfile = ResolveRetryInputOptionProfile(runtime);
var stableBeforeExit = runtime.ProcessStartedAt.HasValue &&
observedAt - runtime.ProcessStartedAt.Value >= StableRuntimeResetThreshold;
var recoveryBase = stableBeforeExit
? InitialRecoveryContext
: runtime.RecoveryContext;
var recoveryContext = AdvanceRecoveryContext(
runtime.RecoveryContext,
recoveryBase,
retryInputOptionProfile,
runtime.SelectedProtocol,
retryStream.SelectedProtocol);
var retryAttempt = recoveryContext.AttemptCount;
try
{
await StartInternalAsync(
@@ -1063,7 +1207,7 @@ public sealed partial class FfmpegService
await logService.WriteAsync(
SystemLogLevel.Warning,
"FFmpeg",
$"ffmpeg exited unexpectedly. Retrying within the current session ({retryAttempt}/{MaxInSessionRetryAttempts}).",
$"ffmpeg exited unexpectedly. Retrying within the current session (attempt {retryAttempt}).",
$"transition={runtime.InputOptionProfile}/{runtime.SelectedProtocol} -> {retryInputOptionProfile}/{retryStream.SelectedProtocol}; output={runtime.GetRecentOutputSummary()}",
session.LiveRoomId,
session.Id,
@@ -1074,10 +1218,82 @@ public sealed partial class FfmpegService
catch (Exception ex)
{
_logger.LogWarning(ex, "In-session ffmpeg retry failed for session {RecordSessionId}", runtime.RecordSessionId);
return false;
return await WaitAndRetryRecoveringSessionAsync(runtime, activeDanmakuSummary, transition, ex);
}
}
private async Task<bool> WaitAndRetryRecoveringSessionAsync(
SessionProcessRuntime runtime,
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary,
SessionTransitionRuntime transition,
Exception failure)
{
var attempt = transition.NextAttempt(Math.Max(1, runtime.RetryAttemptCount + 1));
while (!transition.Token.IsCancellationRequested && !runtime.ShutdownRequested)
{
try
{
using (var scope = _serviceScopeFactory.CreateScope())
{
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
var session = await dbContext.RecordSessions.FirstOrDefaultAsync(
item => item.Id == runtime.RecordSessionId,
CancellationToken.None);
if (session is null || !IsActiveSessionStatus(session.Status))
{
return false;
}
var reason = $"{RuntimeRecoveryMarker} attempt={attempt}; {failure.GetBaseException().Message}";
session.MarkRecovering(reason, DateTimeOffset.UtcNow);
await dbContext.SaveChangesAsync(CancellationToken.None);
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
await logService.WriteAsync(
SystemLogLevel.Warning,
"FFmpeg",
$"Recorder recovery is waiting before attempt {attempt}.",
$"delaySeconds={GetRuntimeRecoveryDelay(attempt).TotalSeconds:0}; failure={failure}",
runtime.LiveRoomId,
runtime.RecordSessionId,
runtime.CurrentTaskId,
CancellationToken.None);
}
await Task.Delay(GetRuntimeRecoveryDelay(attempt), transition.Token);
if (transition.MarkAsCompletedOnExit)
{
runtime.MarkStopRequested(markAsCompletedOnExit: true);
return false;
}
if (transition.StopRequested)
{
runtime.MarkStopRequested(markAsCompletedOnExit: false);
return false;
}
// Re-enter the normal recovery path with a fresh platform status and stream URL.
return await TryRecoverUnexpectedExitAsync(runtime, activeDanmakuSummary, transition);
}
catch (OperationCanceledException) when (transition.Token.IsCancellationRequested)
{
runtime.MarkStopRequested(transition.MarkAsCompletedOnExit);
return false;
}
catch (Exception ex)
{
failure = ex;
attempt = transition.NextAttempt(attempt + 1);
}
}
return false;
}
internal static TimeSpan GetRuntimeRecoveryDelay(int attempt) =>
RuntimeRecoveryBackoff[Math.Clamp(attempt - 1, 0, RuntimeRecoveryBackoff.Length - 1)];
private async Task<bool> TryRetryWithAlternateProtocolAsync(
SessionProcessRuntime runtime,
RecordSession session,
@@ -1543,7 +1759,9 @@ public sealed partial class FfmpegService
$"exitCode={exitCode}",
$"output={effectiveOutputPath}",
$"recorderOutput={runtime.RecorderOutputPath}",
$"shutdownRequested={runtime.ShutdownRequested}"
$"shutdownRequested={runtime.ShutdownRequested}",
$"lifetimeSeconds={(runtime.ProcessStartedAt.HasValue ? Math.Max(0, (DateTimeOffset.UtcNow - runtime.ProcessStartedAt.Value).TotalSeconds).ToString("F1", CultureInfo.InvariantCulture) : "unknown")}",
$"curlExitCode={runtime.CurlExitCode?.ToString(CultureInfo.InvariantCulture) ?? "unknown"}"
};
if (toleratedNonZeroExit)
@@ -1567,6 +1785,12 @@ public sealed partial class FfmpegService
parts.Add($"recentOutput={recentOutput}");
}
var recentCurlError = runtime.GetRecentCurlErrorSummary();
if (!string.IsNullOrWhiteSpace(recentCurlError))
{
parts.Add($"curlStderr={recentCurlError}");
}
return string.Join("; ", parts);
}
@@ -2280,6 +2504,7 @@ public sealed partial class FfmpegService
private object RuntimeSourceFailureSync { get; } = new();
private object RecentOutputSync { get; } = new();
private Queue<string> RecentOutputLines { get; } = new();
private Queue<string> RecentCurlErrorLines { get; } = new();
private DateTimeOffset RuntimeSourceFailureWindowStartedAt { get; set; }
private int RuntimeSourceFailureCount { get; set; }
private bool RuntimeSourceFailureVerificationInProgress { get; set; }
@@ -2290,6 +2515,14 @@ public sealed partial class FfmpegService
ProcessStartedAt = DateTimeOffset.UtcNow;
}
public void AttachCurlProcess(Process curlProcess) => CurlProcess = curlProcess;
public int? CurlExitCode
{
get
{
try { return CurlProcess?.HasExited == true ? CurlProcess.ExitCode : null; }
catch { return null; }
}
}
public void MarkTimestampDiscontinuityFailure() => HasTimestampDiscontinuityFailure = true;
public void MarkTimestampMuxerFailure() => HasTimestampMuxerFailure = true;
public void MarkHlsOverlongHeadersFailure() => HasHlsOverlongHeadersFailure = true;
@@ -2353,6 +2586,28 @@ public sealed partial class FfmpegService
}
}
public void RememberCurlErrorLine(string line)
{
lock (RecentOutputSync)
{
RecentCurlErrorLines.Enqueue(line.Trim());
while (RecentCurlErrorLines.Count > 8)
{
RecentCurlErrorLines.Dequeue();
}
}
}
public string? GetRecentCurlErrorSummary()
{
lock (RecentOutputSync)
{
return RecentCurlErrorLines.Count == 0
? null
: string.Join(" | ", RecentCurlErrorLines);
}
}
public void ResetCurrentRecorderSegmentPaths(string openedPath)
{
CurrentRecorderSegmentPaths.Clear();
@@ -2428,6 +2683,41 @@ public sealed partial class FfmpegService
Gate.Dispose();
}
}
private sealed class SessionTransitionRuntime : IDisposable
{
private readonly CancellationTokenSource _cancellation = new();
public SessionTransitionRuntime(Guid recordSessionId) => RecordSessionId = recordSessionId;
public Guid RecordSessionId { get; }
public bool StopRequested { get; private set; }
public bool MarkAsCompletedOnExit { get; private set; }
public CancellationToken Token => _cancellation.Token;
public TaskCompletionSource<bool> Completion { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
private int _attemptCount;
public int NextAttempt(int minimum)
{
var next = Interlocked.Increment(ref _attemptCount);
if (next >= minimum)
{
return next;
}
Interlocked.Exchange(ref _attemptCount, minimum);
return minimum;
}
public void RequestStop(bool markAsCompletedOnExit)
{
StopRequested = true;
MarkAsCompletedOnExit = markAsCompletedOnExit;
_cancellation.Cancel();
}
public void Dispose() => _cancellation.Dispose();
}
}
internal enum ExitedRecordingDisposition
@@ -678,6 +678,119 @@ public sealed partial class FfmpegService
return arguments;
}
private async Task<(string OutputPath, string? ErrorMessage)> TryRepairArtifactFileAsync(
string ffmpegPath,
int maxConcurrentTranscodeTasks,
int timeoutMinutes,
Guid recordTaskId,
string sourcePath,
string targetPath,
CancellationToken cancellationToken)
{
if (File.Exists(targetPath))
{
var existing = await ValidateMediaArtifactAsync(targetPath, RecordOutputFormat.Mp4, cancellationToken);
if (existing.IsValid)
{
return (targetPath, null);
}
File.Delete(targetPath);
}
var temporaryPath = $"{targetPath}.repairing";
string? lastError = null;
foreach (var strategy in new[] { Mp4FinalizeStrategy.StreamCopy, Mp4FinalizeStrategy.RepairTranscode })
{
if (File.Exists(temporaryPath))
{
File.Delete(temporaryPath);
}
using var transcodeSlot = await AcquireTranscodeSlotAsync(maxConcurrentTranscodeTasks, cancellationToken);
using var process = new Process
{
StartInfo = new ProcessStartInfo
{
FileName = ffmpegPath,
UseShellExecute = false,
RedirectStandardError = true,
RedirectStandardOutput = true,
CreateNoWindow = true
}
};
var processStarted = false;
foreach (var argument in BuildMp4FinalizeArgumentList(sourcePath, temporaryPath, strategy))
{
process.StartInfo.ArgumentList.Add(argument);
}
try
{
process.Start();
processStarted = true;
_postProcessProcesses[recordTaskId] = process;
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
timeout.CancelAfter(TimeSpan.FromMinutes(Math.Clamp(timeoutMinutes, 1, 1440)));
var stdout = process.StandardOutput.ReadToEndAsync(timeout.Token);
var stderr = process.StandardError.ReadToEndAsync(timeout.Token);
await process.WaitForExitAsync(timeout.Token);
await stdout;
lastError = await stderr;
if (process.ExitCode == 0 && File.Exists(temporaryPath))
{
var validation = await ValidateMediaArtifactAsync(
temporaryPath,
RecordOutputFormat.Mp4,
cancellationToken);
if (validation.IsValid)
{
File.Move(temporaryPath, targetPath, overwrite: true);
return (targetPath, null);
}
lastError = validation.ErrorMessage;
}
}
catch (OperationCanceledException)
{
if (cancellationToken.IsCancellationRequested)
{
throw;
}
lastError = $"FFmpeg 修复超过 {Math.Clamp(timeoutMinutes, 1, 1440)} 分钟限制。";
}
finally
{
_postProcessProcesses.TryRemove(recordTaskId, out _);
if (processStarted && !process.HasExited)
{
try
{
process.Kill(true);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to stop artifact repair process for task {RecordTaskId}", recordTaskId);
}
}
if (File.Exists(temporaryPath))
{
File.Delete(temporaryPath);
}
}
}
return (
targetPath,
string.IsNullOrWhiteSpace(lastError)
? "FFmpeg 无法从该文件生成可读取的视频。"
: lastError.Trim());
}
internal static bool IsRepairableMp4FinalizeError(string errorDetail)
{
if (IsLowStoragePauseError(errorDetail))
@@ -1067,13 +1180,18 @@ public sealed partial class FfmpegService
return MediaArtifactValidation.Invalid("The recorded media file does not contain a video stream.", metadata.DurationSeconds);
}
if (!metadata.DurationSeconds.HasValue ||
metadata.DurationSeconds.Value < MinimumUnexpectedExitArtifactDuration.TotalSeconds)
if (!IsReadableMediaDuration(metadata.DurationSeconds))
{
return MediaArtifactValidation.Invalid(ShortUnexpectedExitArtifactError, metadata.DurationSeconds);
return MediaArtifactValidation.Invalid("The recorded media file has no readable duration.", metadata.DurationSeconds);
}
return new MediaArtifactValidation(true, metadata.DurationSeconds, null);
return new MediaArtifactValidation(
true,
metadata.DurationSeconds,
IsStandaloneMediaDuration(metadata.DurationSeconds),
!IsStandaloneMediaDuration(metadata.DurationSeconds)
? ShortUnexpectedExitArtifactError
: null);
}
private static async Task UpsertRecordResultAsync(
@@ -1458,6 +1576,10 @@ public sealed partial class FfmpegService
private static bool IsTerminalSessionStatus(RecordSessionStatus status) =>
status is RecordSessionStatus.Completed or RecordSessionStatus.Failed or RecordSessionStatus.Stopped;
internal static bool IsReadableMediaDuration(double? durationSeconds) => durationSeconds is > 0;
internal static bool IsStandaloneMediaDuration(double? durationSeconds) => durationSeconds is >= 5;
private static bool TryParseFfmpegProgressSeconds(string line, out double seconds)
{
if (line.StartsWith("out_time=", StringComparison.OrdinalIgnoreCase))
@@ -1495,10 +1617,14 @@ public sealed partial class FfmpegService
TimestampTranscode = 3
}
private sealed record MediaArtifactValidation(bool IsValid, double? DurationSeconds, string? ErrorMessage)
private sealed record MediaArtifactValidation(
bool IsValid,
double? DurationSeconds,
bool IsStandaloneEligible,
string? ErrorMessage)
{
public static MediaArtifactValidation Invalid(string errorMessage, double? durationSeconds = null) =>
new(false, durationSeconds, errorMessage);
new(false, durationSeconds, false, errorMessage);
}
private enum StartupFailureKind
@@ -21,6 +21,7 @@ namespace LiveRecorder.Infrastructure.Services;
public sealed partial class FfmpegService : IFfmpegService
{
private const string ArtifactRepairMarker = "[artifact-repair]";
private static readonly Regex SegmentOpeningRegex = new(
"""Opening '([^']+)' for writing""",
RegexOptions.Compiled | RegexOptions.CultureInvariant | RegexOptions.IgnoreCase);
@@ -32,6 +33,7 @@ public sealed partial class FfmpegService : IFfmpegService
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();
@@ -62,7 +64,16 @@ public sealed partial class FfmpegService : IFfmpegService
_logger = logger;
}
public bool IsRunning(Guid recordSessionId) => _processes.ContainsKey(recordSessionId);
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)
{
@@ -83,6 +94,120 @@ public sealed partial class FfmpegService : IFfmpegService
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);
@@ -229,6 +354,7 @@ public sealed partial class FfmpegService : IFfmpegService
{
if (!string.IsNullOrWhiteSpace(args.Data))
{
runtime.RememberCurlErrorLine(args.Data);
_logger.LogDebug("curl[{SessionId}] {Line}", recordSession.Id, args.Data);
}
};
@@ -283,6 +409,12 @@ public sealed partial class FfmpegService : IFfmpegService
{
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;
}
@@ -297,6 +429,12 @@ public sealed partial class FfmpegService : IFfmpegService
{
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;
}
@@ -340,6 +478,7 @@ public sealed partial class FfmpegService : IFfmpegService
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
@@ -355,7 +494,13 @@ public sealed partial class FfmpegService : IFfmpegService
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))
{
@@ -391,7 +536,23 @@ public sealed partial class FfmpegService : IFfmpegService
}
await WaitForAllProcessesAsync(forceKillTimeout, CancellationToken.None);
return runtimes.Length;
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(
@@ -417,7 +578,7 @@ public sealed partial class FfmpegService : IFfmpegService
private async Task WaitForAllProcessesAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
var deadline = DateTimeOffset.UtcNow + timeout;
while ((!_processes.IsEmpty || !_postProcessProcesses.IsEmpty || !_postProcessStates.IsEmpty) &&
while ((!_processes.IsEmpty || !_sessionTransitions.IsEmpty || !_postProcessProcesses.IsEmpty || !_postProcessStates.IsEmpty) &&
DateTimeOffset.UtcNow < deadline)
{
await Task.Delay(TimeSpan.FromMilliseconds(200), cancellationToken);
@@ -1020,6 +1181,215 @@ public sealed partial class FfmpegService : IFfmpegService
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 ||
@@ -1081,7 +1451,7 @@ public sealed partial class FfmpegService : IFfmpegService
SystemLogLevel.Info,
"FFmpeg",
$"ffmpeg input profile={runtime.InputOptionProfile}.",
detail: $"stream={runtime.SelectedProtocol}:{runtime.SelectedQuality}; videoCodec={runtime.SelectedVideoCodec ?? "unknown"}; recoveryEncoder={runtime.RecoveryVideoEncoder.Kind}; recoveryDevice={runtime.RecoveryVideoEncoder.DevicePath ?? "default"}; attempt={runtime.RetryAttemptCount}/{MaxInSessionRetryAttempts}; compatibilityRetry={runtime.HasRetriedWithCompatibilityProfile}; refreshRetry={runtime.HasRetriedWithRefreshedStream}; alternateProtocolRetry={runtime.HasRetriedWithAlternateProtocol}; softwareEncoderRetry={runtime.RecoveryContext.HasRetriedWithSoftwareEncoder}",
detail: $"stream={runtime.SelectedProtocol}:{runtime.SelectedQuality}; 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}; softwareEncoderRetry={runtime.RecoveryContext.HasRetriedWithSoftwareEncoder}",
liveRoomId: runtime.LiveRoomId,
recordSessionId: runtime.RecordSessionId,
recordTaskId: runtime.CurrentTaskId,
@@ -1106,6 +1476,24 @@ public sealed partial class FfmpegService : IFfmpegService
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,
@@ -1114,6 +1502,11 @@ public sealed partial class FfmpegService : IFfmpegService
{
if (!_processes.TryGetValue(recordSessionId, out var runtime))
{
if (_sessionTransitions.TryGetValue(recordSessionId, out var transition))
{
transition.RequestStop(markAsCompletedOnExit || shutdownRequested);
}
return;
}
@@ -29,6 +29,7 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR
private static readonly TimeSpan PollDispatchSpacing = TimeSpan.FromMilliseconds(400);
private static readonly TimeSpan MinimumIdleDelay = TimeSpan.FromSeconds(2);
private static readonly TimeSpan PerRoomPollingTimeout = TimeSpan.FromMinutes(2);
private static readonly TimeSpan OfflineConfirmationDelay = TimeSpan.FromSeconds(10);
private const int MaxConcurrentLiveRoomPolls = 2;
private readonly IServiceScopeFactory _serviceScopeFactory;
@@ -74,6 +75,7 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
var settings = await settingsService.GetAsync(stoppingToken);
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
await ffmpegService.ResumeRecoveringSessionsAsync(stoppingToken);
var recoveredOrphanedSessions = await ffmpegService.RecoverOrphanedTerminalSessionTasksAsync(stoppingToken);
if (recoveredOrphanedSessions > 0)
{
@@ -88,6 +90,9 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR
// ResumePausedFinalizationsAsync internally no-ops only when space is truly
// insufficient to remux, and it is retried every poll cycle.
await ffmpegService.ResumePausedFinalizationsAsync(stoppingToken);
await ffmpegService.ResumeArtifactRepairsAsync(stoppingToken);
var shortFragmentRecovery = scope.ServiceProvider.GetRequiredService<ShortFragmentConsolidationService>();
await shortFragmentRecovery.ResumeInterruptedMergesAsync(stoppingToken);
if (!settings.EnableBackgroundPolling)
{
@@ -237,6 +242,8 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR
private static int GetEffectivePollingIntervalSeconds(int globalIntervalSeconds, int? overrideIntervalSeconds) =>
Math.Clamp(overrideIntervalSeconds ?? globalIntervalSeconds, 10, 3600);
internal static bool RequiresOfflineConfirmation(bool isLive) => !isLive;
private async Task PollLiveRoomsAsync(
IReadOnlyList<PollCandidate> liveRooms,
SystemSettingsDto settings,
@@ -359,6 +366,20 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR
{
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
var liveStatus = await adapter.GetLiveStatusAsync(liveRoom.RoomId, cancellationToken);
if (RequiresOfflineConfirmation(liveStatus.IsLive))
{
await Task.Delay(OfflineConfirmationDelay, cancellationToken);
var confirmation = await adapter.GetLiveStatusAsync(liveRoom.RoomId, cancellationToken);
if (confirmation.IsLive)
{
_logger.LogInformation(
"Ignored a transient offline sample for live room {RoomId}; the confirmation check is live.",
liveRoom.RoomId);
}
liveStatus = confirmation;
}
var now = DateTimeOffset.UtcNow;
await liveRoomStatusService.ApplySnapshotAsync(liveRoom, liveStatus, now, cancellationToken: cancellationToken);
@@ -10,6 +10,11 @@ namespace LiveRecorder.Infrastructure.Services;
public interface IOpenListClient
{
Task EnsureAuthenticatedAsync(
OpenListConnectionRequest connection,
bool forceRefresh = false,
CancellationToken cancellationToken = default);
Task<OpenListConnectionTestDto> TestConnectionAsync(
OpenListConnectionRequest connection,
CancellationToken cancellationToken = default);
@@ -72,16 +77,73 @@ public sealed record OpenListTaskInfo(
string Status,
string? Error);
public sealed class OpenListApiException : InvalidOperationException
{
public OpenListApiException(
string operation,
int apiCode,
HttpStatusCode httpStatusCode,
string apiMessage,
TimeSpan? retryAfter = null)
: base($"{operation} failed with code {apiCode}: {apiMessage}")
{
Operation = operation;
ApiCode = apiCode;
HttpStatusCode = httpStatusCode;
ApiMessage = apiMessage;
RetryAfter = retryAfter;
}
public string Operation { get; }
public int ApiCode { get; }
public HttpStatusCode HttpStatusCode { get; }
public string ApiMessage { get; }
public TimeSpan? RetryAfter { get; }
public bool IsRateLimited =>
HttpStatusCode == HttpStatusCode.TooManyRequests ||
ApiCode == 429 ||
ApiMessage.Contains("too many", StringComparison.OrdinalIgnoreCase) ||
ApiMessage.Contains("尝试过多", StringComparison.OrdinalIgnoreCase) ||
ApiMessage.Contains("请求过于频繁", StringComparison.OrdinalIgnoreCase);
public bool IsAuthenticationFailure =>
HttpStatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden ||
ApiCode is 401 or 403 ||
Operation.Contains("login", StringComparison.OrdinalIgnoreCase) &&
(
ApiMessage.Contains("password", StringComparison.OrdinalIgnoreCase) ||
ApiMessage.Contains("credential", StringComparison.OrdinalIgnoreCase) ||
ApiMessage.Contains("用户名", StringComparison.OrdinalIgnoreCase) ||
ApiMessage.Contains("密码", StringComparison.OrdinalIgnoreCase));
}
public sealed class OpenListClient : IOpenListClient
{
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
private readonly IHttpClientFactory _httpClientFactory;
private readonly OpenListUploadHealthState _healthState;
private readonly ConcurrentDictionary<string, TokenCacheEntry> _tokens = new(StringComparer.Ordinal);
private readonly SemaphoreSlim _loginGate = new(1, 1);
public OpenListClient(IHttpClientFactory httpClientFactory)
public OpenListClient(
IHttpClientFactory httpClientFactory,
OpenListUploadHealthState? healthState = null)
{
_httpClientFactory = httpClientFactory;
_healthState = healthState ?? new OpenListUploadHealthState();
}
public async Task EnsureAuthenticatedAsync(
OpenListConnectionRequest connection,
bool forceRefresh = false,
CancellationToken cancellationToken = default)
{
_ = await GetTokenAsync(connection, forceRefresh, cancellationToken);
}
public async Task<OpenListConnectionTestDto> TestConnectionAsync(
@@ -533,6 +595,7 @@ public sealed class OpenListClient : IOpenListClient
Func<HttpRequestMessage> requestFactory,
CancellationToken cancellationToken)
{
ApiEnvelope? lastUnauthorized = null;
for (var attempt = 0; attempt < 2; attempt++)
{
var forceRefresh = attempt > 0;
@@ -549,10 +612,24 @@ public sealed class OpenListClient : IOpenListClient
return envelope;
}
lastUnauthorized = envelope;
InvalidateToken(connection);
}
throw new InvalidOperationException("OpenList 登录状态无效,请检查账号或密码。");
var failure = lastUnauthorized ?? new ApiEnvelope(
401,
"OpenList 登录状态无效,请检查账号或密码。",
null,
HttpStatusCode.Unauthorized,
null);
var exception = new OpenListApiException(
"OpenList authenticated request",
failure.Code,
failure.HttpStatusCode,
failure.Message,
failure.RetryAfter);
_healthState.MarkProviderFailure(exception, DateTimeOffset.UtcNow);
throw exception;
}
private async Task<string> GetTokenAsync(
@@ -592,6 +669,7 @@ public sealed class OpenListClient : IOpenListClient
var token = tokenElement.GetString()!;
_tokens[cacheKey] = new TokenCacheEntry(token, DateTimeOffset.UtcNow.AddMinutes(20));
_healthState.MarkHealthy();
return token;
}
finally
@@ -617,10 +695,11 @@ public sealed class OpenListClient : IOpenListClient
private static async Task<ApiEnvelope> ReadEnvelopeAsync(HttpResponseMessage response, CancellationToken cancellationToken)
{
var retryAfter = response.Headers.RetryAfter?.Delta;
var body = await response.Content.ReadAsStringAsync(cancellationToken);
if (string.IsNullOrWhiteSpace(body))
{
return new ApiEnvelope((int)response.StatusCode, response.ReasonPhrase ?? "Empty response", null);
return new ApiEnvelope((int)response.StatusCode, response.ReasonPhrase ?? "Empty response", null, response.StatusCode, retryAfter);
}
try
@@ -636,7 +715,16 @@ public sealed class OpenListClient : IOpenListClient
JsonElement? data = root.TryGetProperty("data", out var dataElement)
? dataElement.Clone()
: null;
return new ApiEnvelope(code, message, data);
return new ApiEnvelope(code, message, data, response.StatusCode, retryAfter);
}
catch (JsonException) when (!response.IsSuccessStatusCode)
{
return new ApiEnvelope(
(int)response.StatusCode,
body,
null,
response.StatusCode,
retryAfter);
}
catch (JsonException ex)
{
@@ -644,12 +732,25 @@ public sealed class OpenListClient : IOpenListClient
}
}
private static void EnsureSuccess(ApiEnvelope envelope, string operation)
private void EnsureSuccess(ApiEnvelope envelope, string operation)
{
if (envelope.Code != 200)
if (envelope.Code != 200 || (int)envelope.HttpStatusCode >= 400)
{
throw new InvalidOperationException($"{operation} failed with code {envelope.Code}: {envelope.Message}");
var exception = new OpenListApiException(
operation,
envelope.Code,
envelope.HttpStatusCode,
envelope.Message,
envelope.RetryAfter);
if (exception.IsRateLimited || exception.IsAuthenticationFailure)
{
_healthState.MarkProviderFailure(exception, DateTimeOffset.UtcNow);
}
throw exception;
}
_healthState.MarkHealthy();
}
private static bool ContainsAny(string? value, params string[] candidates) =>
@@ -681,5 +782,10 @@ public sealed class OpenListClient : IOpenListClient
private sealed record TokenCacheEntry(string Token, DateTimeOffset ExpiresAt);
private sealed record ApiEnvelope(int Code, string Message, JsonElement? Data);
private sealed record ApiEnvelope(
int Code,
string Message,
JsonElement? Data,
HttpStatusCode HttpStatusCode,
TimeSpan? RetryAfter);
}
@@ -0,0 +1,126 @@
namespace LiveRecorder.Infrastructure.Services;
public enum OpenListQueueHealthStatus
{
Healthy,
RateLimited,
AuthenticationBlocked,
Disabled
}
public sealed record OpenListQueueHealthSnapshot(
OpenListQueueHealthStatus Status,
string? Reason,
DateTimeOffset? RetryAt,
DateTimeOffset? LastErrorAt)
{
public bool IsPaused => Status != OpenListQueueHealthStatus.Healthy;
}
/// <summary>
/// Process-wide OpenList availability guard. Upload jobs are persisted in the database,
/// while provider-wide authentication/rate-limit failures are deliberately kept out of
/// individual retry budgets.
/// </summary>
public sealed class OpenListUploadHealthState
{
private static readonly TimeSpan DefaultRateLimitDelay = TimeSpan.FromMinutes(15);
private static readonly TimeSpan MaximumRateLimitDelay = TimeSpan.FromHours(1);
private readonly object _gate = new();
private OpenListQueueHealthStatus _status = OpenListQueueHealthStatus.Healthy;
private string? _reason;
private DateTimeOffset? _retryAt;
private DateTimeOffset? _lastErrorAt;
private int _consecutiveRateLimits;
public OpenListQueueHealthSnapshot GetSnapshot()
{
lock (_gate)
{
return new OpenListQueueHealthSnapshot(_status, _reason, _retryAt, _lastErrorAt);
}
}
public bool CanProcess(DateTimeOffset now)
{
lock (_gate)
{
return _status switch
{
OpenListQueueHealthStatus.Healthy => true,
OpenListQueueHealthStatus.RateLimited => _retryAt.HasValue && _retryAt <= now,
_ => false
};
}
}
public void MarkHealthy()
{
lock (_gate)
{
_status = OpenListQueueHealthStatus.Healthy;
_reason = null;
_retryAt = null;
_lastErrorAt = null;
_consecutiveRateLimits = 0;
}
}
public void MarkEnabled()
{
lock (_gate)
{
if (_status == OpenListQueueHealthStatus.Disabled)
{
_status = OpenListQueueHealthStatus.Healthy;
_reason = null;
_retryAt = null;
}
}
}
public void MarkDisabled(string reason)
{
lock (_gate)
{
_status = OpenListQueueHealthStatus.Disabled;
_reason = reason;
_retryAt = null;
}
}
public void MarkProviderFailure(OpenListApiException exception, DateTimeOffset now)
{
lock (_gate)
{
_lastErrorAt = now;
_reason = exception.Message;
if (exception.IsRateLimited)
{
_consecutiveRateLimits++;
var multiplier = Math.Pow(2, Math.Clamp(_consecutiveRateLimits - 1, 0, 2));
var calculated = TimeSpan.FromTicks((long)(DefaultRateLimitDelay.Ticks * multiplier));
var delay = exception.RetryAfter.GetValueOrDefault(calculated);
if (delay <= TimeSpan.Zero)
{
delay = DefaultRateLimitDelay;
}
if (delay > MaximumRateLimitDelay)
{
delay = MaximumRateLimitDelay;
}
_status = OpenListQueueHealthStatus.RateLimited;
_retryAt = now.Add(delay);
return;
}
if (exception.IsAuthenticationFailure)
{
_status = OpenListQueueHealthStatus.AuthenticationBlocked;
_retryAt = null;
}
}
}
}
@@ -34,6 +34,7 @@ public sealed class OpenListUploadQueueService
private readonly LiveRecorderDbContext _dbContext;
private readonly ISystemSettingsService _settingsService;
private readonly IOpenListClient _openListClient;
private readonly OpenListUploadHealthState _healthState;
private readonly ISystemLogService _systemLogService;
private readonly IVideoMetadataService _videoMetadataService;
@@ -41,16 +42,43 @@ public sealed class OpenListUploadQueueService
LiveRecorderDbContext dbContext,
ISystemSettingsService settingsService,
IOpenListClient openListClient,
OpenListUploadHealthState healthState,
ISystemLogService systemLogService,
IVideoMetadataService videoMetadataService)
{
_dbContext = dbContext;
_settingsService = settingsService;
_openListClient = openListClient;
_healthState = healthState;
_systemLogService = systemLogService;
_videoMetadataService = videoMetadataService;
}
public OpenListQueueHealthSnapshot GetHealthSnapshot() => _healthState.GetSnapshot();
public async Task<OpenListQueueHealthSnapshot> ResumeProviderAsync(
CancellationToken cancellationToken = default)
{
var settings = await _settingsService.GetAsync(cancellationToken);
if (!settings.EnableFileUpload || settings.UploadTarget != UploadTargetType.OpenList)
{
_healthState.MarkDisabled("OpenList 上传未启用。");
return _healthState.GetSnapshot();
}
ValidateSettings(settings.OpenListUpload);
_healthState.MarkEnabled();
var connection = new OpenListConnectionRequest
{
BaseUrl = settings.OpenListUpload.BaseUrl,
Username = settings.OpenListUpload.Username,
Password = settings.OpenListUpload.Password
};
await _openListClient.EnsureAuthenticatedAsync(connection, forceRefresh: true, cancellationToken);
_healthState.MarkHealthy();
return _healthState.GetSnapshot();
}
public async Task<RecordArtifactUploadItemResultDto?> TryEnqueueAutomaticAsync(
Guid recordTaskId,
CancellationToken cancellationToken = default)
@@ -83,6 +111,7 @@ public sealed class OpenListUploadQueueService
.AsNoTracking()
.Where(item =>
(item.Status == RecordTaskStatus.Completed || item.Status == RecordTaskStatus.Stopped) &&
!item.IsHiddenArtifactSource &&
item.Result != null &&
item.Result.UploadStatus == RecordArtifactUploadStatus.NotUploaded &&
item.UploadJob == null &&
@@ -148,7 +177,7 @@ public sealed class OpenListUploadQueueService
{
var taskIds = await _dbContext.RecordTasks
.AsNoTracking()
.Where(item => item.RecordSessionId == recordSessionId)
.Where(item => item.RecordSessionId == recordSessionId && !item.IsHiddenArtifactSource)
.OrderBy(static item => item.SegmentIndex)
.ThenBy(static item => item.CreatedAt)
.Select(static item => item.Id)
@@ -180,9 +209,113 @@ public sealed class OpenListUploadQueueService
};
}
public async Task<RecordArtifactUploadItemResultDto> RetryAsync(
Guid recordTaskId,
CancellationToken cancellationToken = default)
{
var settings = await _settingsService.GetAsync(cancellationToken);
if (!settings.EnableFileUpload || settings.UploadTarget != UploadTargetType.OpenList)
{
return Failure(recordTaskId, "OpenList 上传未启用。", "openlist");
}
var recordTask = await _dbContext.RecordTasks
.Include(static item => item.Result)
.Include(static item => item.UploadJob)
.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
if (recordTask?.Result is null || recordTask.UploadJob is null || recordTask.IsHiddenArtifactSource)
{
return Failure(recordTaskId, "该任务没有可重试的上传作业。", "openlist");
}
if (recordTask.UploadJob.Status == RecordArtifactUploadStatus.Failed)
{
return await EnqueueInternalAsync(recordTaskId, settings, cancellationToken);
}
if (recordTask.UploadJob.Status != RecordArtifactUploadStatus.WaitingRetry)
{
return Failure(recordTaskId, "只有上传失败或等待重试的任务可以立即重试。", "openlist");
}
var now = DateTimeOffset.UtcNow;
recordTask.UploadJob.RequestImmediateRetry(now);
recordTask.Result.MarkUploadQueued("openlist", now);
await _dbContext.SaveChangesAsync(cancellationToken);
return QueuedResult(recordTaskId, recordTask.Result, recordTask.UploadJob, "已重置重试次数并立即加入队列。");
}
public async Task<RecordArtifactUploadBatchResultDto> RetryMatchingAsync(
RecordArtifactUploadStatus? uploadStatus,
string? query,
CancellationToken cancellationToken = default)
{
if (uploadStatus.HasValue && uploadStatus is not (RecordArtifactUploadStatus.Failed or RecordArtifactUploadStatus.WaitingRetry))
{
return new RecordArtifactUploadBatchResultDto
{
RequestedCount = 0,
SuccessCount = 0,
FailedCount = 1,
Items = [Failure(Guid.Empty, "批量重试仅支持上传失败和等待重试状态。", "openlist")]
};
}
var statuses = uploadStatus.HasValue
? new[] { uploadStatus.Value }
: new[] { RecordArtifactUploadStatus.Failed, RecordArtifactUploadStatus.WaitingRetry };
var normalizedQuery = query?.Trim().ToLowerInvariant();
var taskQuery = _dbContext.RecordResults
.AsNoTracking()
.Where(item => statuses.Contains(item.UploadStatus) && item.RecordTask != null);
if (!string.IsNullOrWhiteSpace(normalizedQuery))
{
var parsedTaskId = Guid.TryParse(normalizedQuery, out var taskId) ? taskId : (Guid?)null;
taskQuery = taskQuery.Where(item =>
parsedTaskId.HasValue && item.RecordTaskId == parsedTaskId.Value ||
item.FilePath.ToLower().Contains(normalizedQuery) ||
item.RemoteVideoPath != null && item.RemoteVideoPath.ToLower().Contains(normalizedQuery) ||
item.UploadErrorMessage != null && item.UploadErrorMessage.ToLower().Contains(normalizedQuery) ||
item.RecordTask!.LiveRoom != null &&
((item.RecordTask.LiveRoom.Title != null && item.RecordTask.LiveRoom.Title.ToLower().Contains(normalizedQuery)) ||
item.RecordTask.LiveRoom.RoomId.ToLower().Contains(normalizedQuery)));
}
var taskIds = await taskQuery
.OrderBy(static item => item.CreatedAt)
.Select(static item => item.RecordTaskId)
.ToArrayAsync(cancellationToken);
var items = new List<RecordArtifactUploadItemResultDto>(taskIds.Length);
foreach (var taskId in taskIds)
{
items.Add(await RetryAsync(taskId, cancellationToken));
}
return new RecordArtifactUploadBatchResultDto
{
RequestedCount = taskIds.Length,
SuccessCount = items.Count(static item => item.Success),
FailedCount = items.Count(static item => !item.Success),
Items = items
};
}
public async Task<bool> ProcessNextAsync(CancellationToken cancellationToken = default)
{
var now = DateTimeOffset.UtcNow;
var settings = await _settingsService.GetAsync(cancellationToken);
if (!settings.EnableFileUpload || settings.UploadTarget != UploadTargetType.OpenList)
{
_healthState.MarkDisabled("OpenList 上传已在设置中关闭。");
return false;
}
_healthState.MarkEnabled();
if (!_healthState.CanProcess(now))
{
return false;
}
var jobs = _dbContext.RecordUploadJobs
.Include(static item => item.RecordTask)
.ThenInclude(static item => item!.Result)
@@ -215,6 +348,38 @@ public sealed class OpenListUploadQueueService
}
var result = job.RecordTask.Result;
var connection = new OpenListConnectionRequest
{
BaseUrl = job.ProviderEndpoint,
Username = settings.OpenListUpload.Username,
Password = settings.OpenListUpload.Password
};
try
{
await _openListClient.EnsureAuthenticatedAsync(connection, cancellationToken: cancellationToken);
}
catch (OpenListApiException ex) when (ex.IsRateLimited || ex.IsAuthenticationFailure)
{
await _systemLogService.WriteAsync(
SystemLogLevel.Warning,
"Upload",
"OpenList 上传队列已暂停,任务重试次数未消耗。",
ex.Message,
cancellationToken: cancellationToken);
return true;
}
catch (Exception ex)
{
if (job.Status != RecordArtifactUploadStatus.Uploading)
{
job.BeginAttempt(now);
result.MarkUploadStarted("openlist", now);
}
await ScheduleRetryOrFailAsync(job, result, ex.Message, clearExternalTask: false, cancellationToken);
return true;
}
var validationError = await GetUploadValidationErrorAsync(
job.RecordTask,
NormalizeAbsolutePath(result.FilePath),
@@ -225,7 +390,8 @@ public sealed class OpenListUploadQueueService
return true;
}
if (job.Status != RecordArtifactUploadStatus.Uploading)
var attemptStartedThisRun = job.Status != RecordArtifactUploadStatus.Uploading;
if (attemptStartedThisRun)
{
job.BeginAttempt(now);
result.MarkUploadStarted("openlist", now);
@@ -234,13 +400,6 @@ public sealed class OpenListUploadQueueService
try
{
var settings = await _settingsService.GetAsync(cancellationToken);
var connection = new OpenListConnectionRequest
{
BaseUrl = job.ProviderEndpoint,
Username = settings.OpenListUpload.Username,
Password = settings.OpenListUpload.Password
};
await ProcessJobStepAsync(job, result, connection, cancellationToken);
}
catch (OpenListUploadConflictException ex)
@@ -251,6 +410,19 @@ public sealed class OpenListUploadQueueService
{
throw;
}
catch (OpenListApiException ex) when (ex.IsRateLimited || ex.IsAuthenticationFailure)
{
var pausedAt = DateTimeOffset.UtcNow;
job.SuspendForProviderFailure(ex.Message, pausedAt, attemptStartedThisRun);
result.MarkUploadWaitingRetry("openlist", ex.Message, pausedAt);
await _dbContext.SaveChangesAsync(cancellationToken);
await _systemLogService.WriteAsync(
SystemLogLevel.Warning,
"Upload",
"OpenList 上传队列已暂停,当前任务重试次数已回退。",
ex.Message,
cancellationToken: cancellationToken);
}
catch (Exception ex)
{
await ScheduleRetryOrFailAsync(job, result, ex.Message, clearExternalTask: false, cancellationToken);
@@ -274,6 +446,11 @@ public sealed class OpenListUploadQueueService
return Failure(recordTaskId, "录制结果尚未生成,不能上传。", "openlist");
}
if (recordTask.IsHiddenArtifactSource)
{
return Failure(recordTaskId, "该源分片已被短片合并流程收纳,不能单独上传。", "openlist");
}
var result = recordTask.Result;
var localVideoPath = NormalizeAbsolutePath(result.FilePath);
if (string.IsNullOrWhiteSpace(localVideoPath) || !File.Exists(localVideoPath))
@@ -63,6 +63,38 @@ public sealed class RecordUploadService
return await UploadTaskInternalAsync(recordTaskId, settings, automatic: false, cancellationToken);
}
public async Task<RecordArtifactUploadItemResultDto> RetryTaskAsync(
Guid recordTaskId,
CancellationToken cancellationToken = default)
{
var settings = await _systemSettingsService.GetAsync(cancellationToken);
if (settings.UploadTarget != UploadTargetType.OpenList)
{
return CreateFailureResult(recordTaskId, "当前上传目标不支持队列重试。");
}
return await _openListUploadQueue.RetryAsync(recordTaskId, cancellationToken);
}
public async Task<RecordArtifactUploadBatchResultDto> RetryMatchingAsync(
RetryRecordArtifactUploadsRequest request,
CancellationToken cancellationToken = default)
{
var settings = await _systemSettingsService.GetAsync(cancellationToken);
if (settings.UploadTarget != UploadTargetType.OpenList)
{
return new RecordArtifactUploadBatchResultDto
{
RequestedCount = 0,
SuccessCount = 0,
FailedCount = 1,
Items = [CreateFailureResult(Guid.Empty, "当前上传目标不支持队列重试。")]
};
}
return await _openListUploadQueue.RetryMatchingAsync(request.UploadStatus, request.Query, cancellationToken);
}
public async Task<RecordArtifactUploadBatchResultDto> UploadSessionAsync(
Guid recordSessionId,
CancellationToken cancellationToken = default)
@@ -98,6 +130,7 @@ public sealed class RecordUploadService
}
var taskIds = session.RecordTasks
.Where(static item => !item.IsHiddenArtifactSource)
.OrderBy(static item => item.SegmentIndex)
.ThenBy(static item => item.CreatedAt)
.Select(static item => item.Id)
@@ -135,6 +168,11 @@ public sealed class RecordUploadService
return CreateFailureResult(recordTaskId, "Recording result is not ready for upload.");
}
if (recordTask.IsHiddenArtifactSource)
{
return CreateFailureResult(recordTaskId, "该分片已并入相邻录像或转存恢复目录,不能单独上传。");
}
if (!settings.EnableFileUpload || settings.UploadTarget == UploadTargetType.None)
{
return CreateFailureResult(recordTaskId, "File upload is disabled or no upload target is configured.");
@@ -1,4 +1,5 @@
using LiveRecorder.Application.Abstractions.Recording;
using LiveRecorder.Application.Abstractions.Logging;
using LiveRecorder.Application.Abstractions.Settings;
using LiveRecorder.Application.Abstractions.Storage;
using LiveRecorder.Application.Common;
@@ -19,19 +20,25 @@ public sealed class RecoveryService
private readonly IStorageGuardService _storageGuardService;
private readonly IFfmpegService _ffmpegService;
private readonly RecordService _recordService;
private readonly IVideoMetadataService _videoMetadataService;
private readonly ISystemLogService _systemLogService;
public RecoveryService(
LiveRecorderDbContext dbContext,
ISystemSettingsService systemSettingsService,
IStorageGuardService storageGuardService,
IFfmpegService ffmpegService,
RecordService recordService)
RecordService recordService,
IVideoMetadataService videoMetadataService,
ISystemLogService systemLogService)
{
_dbContext = dbContext;
_systemSettingsService = systemSettingsService;
_storageGuardService = storageGuardService;
_ffmpegService = ffmpegService;
_recordService = recordService;
_videoMetadataService = videoMetadataService;
_systemLogService = systemLogService;
}
public async Task<RecoveryOverviewDto> GetOverviewAsync(CancellationToken cancellationToken = default)
@@ -40,6 +47,7 @@ public sealed class RecoveryService
var storage = _storageGuardService.CheckCanStartOrResume(settings);
var liveRooms = await ListRecoverableLiveRoomsAsync(cancellationToken);
var finalizations = await ListRecoverableFinalizationsAsync(cancellationToken);
var mergedArtifacts = await ListMergedArtifactsAsync(cancellationToken);
return new RecoveryOverviewDto
{
@@ -61,10 +69,69 @@ public sealed class RecoveryService
RedThresholdPercent = storage.RedThresholdPercent
},
LiveRooms = liveRooms,
Finalizations = finalizations
Finalizations = finalizations,
MergedArtifacts = mergedArtifacts
};
}
private async Task<IReadOnlyList<MergedArtifactRecordDto>> ListMergedArtifactsAsync(CancellationToken cancellationToken)
{
var sources = await _dbContext.RecordTasks
.AsNoTracking()
.Include(item => item.Result)
.Where(item => item.IsHiddenArtifactSource)
.OrderByDescending(item => item.UpdatedAt)
.Take(100)
.ToListAsync(cancellationToken);
var targetIds = sources.Where(item => item.MergedIntoRecordTaskId.HasValue)
.Select(item => item.MergedIntoRecordTaskId!.Value)
.Distinct()
.ToArray();
var targets = await _dbContext.RecordTasks
.AsNoTracking()
.Include(item => item.Result)
.Where(item => targetIds.Contains(item.Id))
.ToDictionaryAsync(item => item.Id, cancellationToken);
return sources.Select(source =>
{
var sourcePath = source.Result?.FilePath ?? source.OutputFilePath;
var parent = string.IsNullOrWhiteSpace(sourcePath) ? null : Path.GetDirectoryName(Path.GetFullPath(sourcePath));
var recoveryBase = parent is null ? null : Path.Combine(parent, ".liverecorder-recovery", source.RecordSessionId.ToString("N"));
var manifest = FindRecoveryManifest(recoveryBase, source.Id);
return new MergedArtifactRecordDto
{
SourceRecordTaskId = source.Id,
RecordSessionId = source.RecordSessionId,
MergedIntoRecordTaskId = source.MergedIntoRecordTaskId,
SourceVideoPath = sourcePath,
MergedVideoPath = source.MergedIntoRecordTaskId.HasValue && targets.TryGetValue(source.MergedIntoRecordTaskId.Value, out var target)
? target.Result?.FilePath
: null,
RecoveryDirectory = manifest is null ? recoveryBase : Path.GetDirectoryName(manifest),
ManifestPath = manifest,
SourceDurationSeconds = source.Result?.DurationSeconds ?? source.DurationSeconds,
CreatedAt = source.UpdatedAt
};
}).ToList();
}
private static string? FindRecoveryManifest(string? recoveryBase, Guid recordTaskId)
{
if (string.IsNullOrWhiteSpace(recoveryBase) || !Directory.Exists(recoveryBase)) return null;
try
{
foreach (var path in Directory.EnumerateFiles(recoveryBase, "manifest.json", SearchOption.AllDirectories).Take(100))
{
if (File.ReadAllText(path).Contains(recordTaskId.ToString(), StringComparison.OrdinalIgnoreCase)) return path;
}
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
}
return null;
}
public async Task<RecoveryActionResultDto> RetryLiveRoomAsync(Guid liveRoomId, CancellationToken cancellationToken = default)
{
var room = await _dbContext.LiveRooms.FirstOrDefaultAsync(item => item.Id == liveRoomId, cancellationToken);
@@ -225,6 +292,211 @@ public sealed class RecoveryService
};
}
public async Task<RecordingFailureListResponse> ListRecordingFailuresAsync(
string? failureKind,
int skip,
int take,
CancellationToken cancellationToken = default)
{
var tasks = await _dbContext.RecordTasks
.AsNoTracking()
.Include(static item => item.LiveRoom)
.Include(static item => item.Result)
.Where(item => item.Status == RecordTaskStatus.Failed ||
item.Status == RecordTaskStatus.Processing &&
item.ErrorMessage != null &&
item.ErrorMessage.StartsWith("[artifact-repair]"))
.OrderByDescending(static item => item.UpdatedAt)
.ToListAsync(cancellationToken);
var items = tasks.Select(MapRecordingFailure).AsEnumerable();
if (!string.IsNullOrWhiteSpace(failureKind))
{
items = items.Where(item => item.FailureKind.Equals(failureKind.Trim(), StringComparison.OrdinalIgnoreCase));
}
var materialized = items.ToList();
return new RecordingFailureListResponse
{
TotalCount = materialized.Count,
Items = materialized
.Skip(Math.Max(0, skip))
.Take(Math.Clamp(take, 1, 100))
.ToList()
};
}
public async Task<RecoveryActionResultDto> AcceptRecordingArtifactAsync(
Guid recordTaskId,
bool confirmShortArtifact,
CancellationToken cancellationToken = default)
{
var task = await _dbContext.RecordTasks
.Include(static item => item.RecordSession)
.ThenInclude(static item => item!.RecordTasks)
.Include(static item => item.Result)
.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
if (task?.Result is null || task.Status != RecordTaskStatus.Failed)
{
return FailureResult("该任务不是可认领的录制失败产物。");
}
var path = NormalizeAbsolutePath(task.Result.FilePath);
if (!File.Exists(path) || new FileInfo(path).Length <= 0)
{
return FailureResult("本地媒体文件不存在或为空,无法认领。");
}
var metadata = await _videoMetadataService.ExtractMetadataAsync(path, cancellationToken);
if (metadata is null || string.IsNullOrWhiteSpace(metadata.VideoCodec) || metadata.DurationSeconds is not > 0)
{
return FailureResult("媒体仍无法读取,请先使用非破坏修复。");
}
if (metadata.DurationSeconds < 5 && !confirmShortArtifact)
{
return FailureResult($"媒体只有 {metadata.DurationSeconds:0.###} 秒,需要确认短分片后才能认领。");
}
var originalError = task.ErrorMessage;
var now = DateTimeOffset.UtcNow;
task.MarkCompleted(task.EndedAt ?? now, metadata.DurationSeconds);
task.Result.Update(
path,
new FileInfo(path).Length,
metadata.DurationSeconds,
task.Result.DanmakuFilePath,
task.Result.DanmakuMessageCount,
RecordTaskStatus.Completed,
null);
task.Result.ResetUploadForRecoveredArtifact();
if (task.RecordSession is not null && task.RecordSession.RecordTasks.All(static item =>
item.Status is RecordTaskStatus.Completed or RecordTaskStatus.Stopped))
{
task.RecordSession.MarkCompleted(now);
}
await _dbContext.SaveChangesAsync(cancellationToken);
await _systemLogService.WriteAsync(
SystemLogLevel.Info,
"Recovery",
"录制失败产物已由管理员确认有效,等待手动上传。",
$"path={path}; duration={metadata.DurationSeconds:0.###}; originalError={originalError}",
task.LiveRoomId,
task.RecordSessionId,
task.Id,
cancellationToken);
return new RecoveryActionResultDto
{
RequestedCount = 1,
SuccessCount = 1,
FailedCount = 0,
Messages = ["分片已认领为有效产物,可前往上传任务页面手动上传。"]
};
}
public async Task<RecoveryActionResultDto> RepairRecordingArtifactAsync(
Guid recordTaskId,
CancellationToken cancellationToken = default)
{
var started = await _ffmpegService.StartArtifactRepairAsync(recordTaskId, cancellationToken);
return new RecoveryActionResultDto
{
RequestedCount = 1,
SuccessCount = started ? 1 : 0,
FailedCount = started ? 0 : 1,
Messages =
[
started
? "非破坏修复已排队;原文件不会被覆盖或删除。"
: "无法启动修复:任务可能正在处理、文件缺失或可用空间不足。"
]
};
}
private static RecordingFailureItemDto MapRecordingFailure(RecordTask task)
{
var result = task.Result;
var path = result?.FilePath;
var normalizedPath = string.IsNullOrWhiteSpace(path) ? null : NormalizeAbsolutePath(path);
var fileExists = normalizedPath is not null && File.Exists(normalizedPath) && new FileInfo(normalizedPath).Length > 0;
var classification = ClassifyRecordingFailure(task, fileExists);
var isRepairing = task.Status == RecordTaskStatus.Processing &&
task.ErrorMessage?.StartsWith("[artifact-repair]", StringComparison.Ordinal) == true;
return new RecordingFailureItemDto
{
RecordTaskId = task.Id,
RecordSessionId = task.RecordSessionId,
LiveRoomId = task.LiveRoomId,
LiveRoomTitle = task.LiveRoom?.Title ?? task.LiveRoom?.AnchorName ?? task.LiveRoom?.RoomId ?? "未知直播间",
RoomId = task.LiveRoom?.RoomId ?? "-",
PlatformName = task.LiveRoom?.Platform.ToString() ?? "Unknown",
SegmentIndex = task.SegmentIndex,
FailureKind = classification.Kind,
FailureLabel = classification.Label,
RecommendedAction = classification.Action,
ErrorMessage = task.ErrorMessage ?? result?.ErrorMessage,
FilePath = normalizedPath,
FileSizeBytes = result?.FileSizeBytes,
DurationSeconds = result?.DurationSeconds ?? task.DurationSeconds,
FileExists = fileExists,
CanAccept = !isRepairing && fileExists && classification.Kind is "ReadableFragment" or "TooShort" or "Unknown",
CanRepair = !isRepairing && fileExists && classification.Kind is "UnreadableMedia" or "FinalizationFailed" or "Unknown",
CanRetryRoom = !isRepairing && task.LiveRoom?.AvailabilityStatus == LiveRoomAvailabilityStatus.Live,
IsRepairing = isRepairing,
CreatedAt = task.CreatedAt
};
}
private static (string Kind, string Label, string Action) ClassifyRecordingFailure(RecordTask task, bool fileExists)
{
var error = task.ErrorMessage ?? task.Result?.ErrorMessage ?? string.Empty;
if (task.Status == RecordTaskStatus.Processing && error.StartsWith("[artifact-repair]", StringComparison.Ordinal))
{
return ("Repairing", "正在修复", "等待修复完成,应用重启后会自动续排。");
}
if (!fileExists)
{
return ("MissingMedia", "文件缺失", "历史媒体文件不存在,无法恢复;若直播仍在线可重新开录。");
}
if (error.Contains("offline", StringComparison.OrdinalIgnoreCase) || error.Contains("已离线", StringComparison.OrdinalIgnoreCase))
{
return ("SourceOffline", "直播已离线", "历史时段无法补录;直播重新在线后可重新开录。");
}
if (error.Contains("播放地址", StringComparison.OrdinalIgnoreCase) || error.Contains("stream url", StringComparison.OrdinalIgnoreCase))
{
return ("StreamUrlUnavailable", "未取得播放地址", "等待下一次巡检刷新播放地址,在线时可手动重试开录。");
}
if (error.Contains("short fragment", StringComparison.OrdinalIgnoreCase) ||
task.Result?.DurationSeconds is > 0 and < 5)
{
return ("TooShort", "短分片", "确认内容有价值后可人工认领;不足 5 秒需要二次确认。");
}
if (error.Contains("ffprobe", StringComparison.OrdinalIgnoreCase) ||
error.Contains("does not contain a video", StringComparison.OrdinalIgnoreCase))
{
return ("UnreadableMedia", "媒体不可读", "尝试生成新的恢复文件;原文件保持不变。");
}
if (error.Contains("finaliz", StringComparison.OrdinalIgnoreCase) ||
error.Contains("mux", StringComparison.OrdinalIgnoreCase))
{
return ("FinalizationFailed", "封装或收尾失败", "先重新封装,失败后再转码修复。");
}
if (task.Result?.DurationSeconds is > 0)
{
return ("ReadableFragment", "异常退出分片", "重新校验媒体;确认有效后转为待上传。");
}
return ("Unknown", "待检测", "可先尝试校验认领;无法读取时再执行非破坏修复。");
}
private async Task<IReadOnlyList<RecoverableLiveRoomDto>> ListRecoverableLiveRoomsAsync(CancellationToken cancellationToken)
{
var activeLiveRoomIds = await _dbContext.RecordSessions
@@ -0,0 +1,422 @@
using System.Diagnostics;
using System.Globalization;
using System.Text.Json;
using System.Xml.Linq;
using LiveRecorder.Application.Abstractions.Recording;
using LiveRecorder.Application.Abstractions.Settings;
using LiveRecorder.Domain.Entities;
using LiveRecorder.Domain.Enums;
using LiveRecorder.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
namespace LiveRecorder.Infrastructure.Services;
public sealed class ShortFragmentConsolidationService
{
internal const double MinimumStandaloneDurationSeconds = 5;
private static readonly TimeSpan ProcessTimeout = TimeSpan.FromMinutes(20);
private readonly LiveRecorderDbContext _dbContext;
private readonly ISystemSettingsService _settingsService;
private readonly IVideoMetadataService _metadataService;
private readonly ILogger<ShortFragmentConsolidationService> _logger;
public ShortFragmentConsolidationService(
LiveRecorderDbContext dbContext,
ISystemSettingsService settingsService,
IVideoMetadataService metadataService,
ILogger<ShortFragmentConsolidationService> logger)
{
_dbContext = dbContext;
_settingsService = settingsService;
_metadataService = metadataService;
_logger = logger;
}
public async Task<FragmentDispatchDecision> PrepareDispatchAsync(Guid recordTaskId, CancellationToken cancellationToken)
{
var task = await _dbContext.RecordTasks
.Include(item => item.RecordSession)
.Include(item => item.Result)
.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
if (task?.RecordSession is null || task.Result is null)
{
return FragmentDispatchDecision.Wait("录制结果仍在写入。", TimeSpan.FromSeconds(2));
}
if (task.IsHiddenArtifactSource)
{
return FragmentDispatchDecision.Skip("该分片已并入相邻分片。", task.MergedIntoRecordTaskId);
}
var tasks = await _dbContext.RecordTasks
.Include(item => item.Result)
.Where(item => item.RecordSessionId == task.RecordSessionId && !item.IsHiddenArtifactSource)
.OrderBy(item => item.SegmentIndex)
.ThenBy(item => item.CreatedAt)
.ToListAsync(cancellationToken);
var index = tasks.FindIndex(item => item.Id == recordTaskId);
if (index < 0)
{
return FragmentDispatchDecision.Skip("该分片已被恢复流程收纳。", task.MergedIntoRecordTaskId);
}
var next = index + 1 < tasks.Count ? tasks[index + 1] : null;
if (next is not null && IsActive(next.Status))
{
var runtime = next.StartedAt.HasValue ? DateTimeOffset.UtcNow - next.StartedAt.Value : TimeSpan.Zero;
if (runtime < TimeSpan.FromSeconds(MinimumStandaloneDurationSeconds) || IsShort(task))
{
return FragmentDispatchDecision.Wait("等待下一分片完成短片判定。", TimeSpan.FromSeconds(2));
}
}
else if (next is null && IsActive(task.RecordSession.Status))
{
return FragmentDispatchDecision.Wait("等待下一分片,避免短片被提前上传。", TimeSpan.FromSeconds(2));
}
var group = ResolveMergeGroup(tasks, index);
if (group.Count == 1 && !IsShort(task))
{
return FragmentDispatchDecision.Ready;
}
var totalDuration = group.Sum(item => item.Result?.DurationSeconds ?? 0);
var target = group.FirstOrDefault(item => !IsShort(item));
if (target is null)
{
if (IsActive(task.RecordSession.Status))
{
return FragmentDispatchDecision.Wait("短分片正在等待相邻有效分片。", TimeSpan.FromSeconds(2));
}
if (totalDuration < MinimumStandaloneDurationSeconds)
{
await MoveRecoveryOnlyAsync(group, cancellationToken);
return FragmentDispatchDecision.Skip("本次会话仅产生不足 5 秒的可读媒体,已转存到恢复目录。", null);
}
target = group[^1];
}
var merged = await MergeAsync(task.RecordSession, target, group, cancellationToken);
return merged
? target.Id == task.Id
? FragmentDispatchDecision.Ready
: FragmentDispatchDecision.Skip("该分片已并入相邻分片。", target.Id)
: FragmentDispatchDecision.Wait("短分片合并失败,原文件已保留,将自动重试。", TimeSpan.FromMinutes(1));
}
public async Task<int> ResumeInterruptedMergesAsync(CancellationToken cancellationToken)
{
var sources = await _dbContext.RecordTasks
.AsNoTracking()
.Include(item => item.Result)
.Where(item => item.IsHiddenArtifactSource && item.Result != null)
.OrderByDescending(item => item.UpdatedAt)
.Take(100)
.ToListAsync(cancellationToken);
var manifests = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
foreach (var source in sources)
{
var parent = Path.GetDirectoryName(Path.GetFullPath(source.Result!.FilePath));
var recoveryBase = parent is null ? null : Path.Combine(parent, ".liverecorder-recovery", source.RecordSessionId.ToString("N"));
if (recoveryBase is null || !Directory.Exists(recoveryBase)) continue;
foreach (var path in Directory.EnumerateFiles(recoveryBase, "manifest.json", SearchOption.AllDirectories).Take(100)) manifests.Add(path);
}
var resumed = 0;
foreach (var path in manifests)
{
try
{
var manifest = JsonSerializer.Deserialize<MergeRecoveryManifest>(await File.ReadAllTextAsync(path, cancellationToken));
if (manifest is null || manifest.Phase == "completed" || string.IsNullOrWhiteSpace(manifest.MergedPath)) continue;
var targetPath = await _dbContext.RecordTasks
.AsNoTracking()
.Where(item => item.Id == manifest.TargetRecordTaskId && item.Result != null)
.Select(item => item.Result!.FilePath)
.FirstOrDefaultAsync(cancellationToken);
if (!File.Exists(manifest.MergedPath) || !string.Equals(Path.GetFullPath(targetPath ?? string.Empty), Path.GetFullPath(manifest.MergedPath), StringComparison.OrdinalIgnoreCase)) continue;
foreach (var source in manifest.Sources)
{
CopyToRecovery(source.VideoPath, manifest.RecoveryDirectory);
CopyToRecovery(source.DanmakuPath, manifest.RecoveryDirectory);
DeleteOriginal(source.VideoPath, manifest.MergedPath);
DeleteOriginal(source.DanmakuPath, Path.ChangeExtension(manifest.MergedPath, ".xml"));
}
await WriteManifestAsync(manifest.RecoveryDirectory, manifest with { Phase = "completed" }, cancellationToken);
resumed++;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or JsonException)
{
_logger.LogWarning(ex, "Unable to resume short-fragment recovery manifest {ManifestPath}", path);
}
}
return resumed;
}
private static List<RecordTask> ResolveMergeGroup(IReadOnlyList<RecordTask> tasks, int index)
{
var durations = tasks.Select(item => IsReadableTerminal(item) ? item.Result!.DurationSeconds : null).ToArray();
var (start, end) = ResolveMergeWindow(durations, index);
return tasks.Skip(start).Take(end - start + 1).Where(IsReadableTerminal).ToList();
}
internal static (int Start, int End) ResolveMergeWindow(IReadOnlyList<double?> durations, int index)
{
if (durations.Count == 0 || index < 0 || index >= durations.Count) return (-1, -1);
var start = index;
var end = index;
while (start > 0 && durations[start - 1] is > 0 and < MinimumStandaloneDurationSeconds) start--;
while (end + 1 < durations.Count && durations[end + 1] is > 0 and < MinimumStandaloneDurationSeconds) end++;
if (durations[index] is > 0 and < MinimumStandaloneDurationSeconds)
{
if (start > 0 && durations[start - 1] is >= MinimumStandaloneDurationSeconds) start--;
else if (end + 1 < durations.Count && durations[end + 1] is >= MinimumStandaloneDurationSeconds) end++;
}
return (start, end);
}
private async Task<bool> MergeAsync(
RecordSession session,
RecordTask target,
IReadOnlyList<RecordTask> group,
CancellationToken cancellationToken)
{
var sources = group.Select(item => item.Result!.FilePath).Where(File.Exists).ToArray();
if (sources.Length != group.Count || sources.Length < 2) return false;
var mergeId = $"{DateTimeOffset.UtcNow:yyyyMMddHHmmss}-{Guid.NewGuid():N}";
var directory = Path.GetDirectoryName(Path.GetFullPath(target.Result!.FilePath))!;
var extension = Path.GetExtension(target.Result.FilePath);
var mergedPath = Path.Combine(directory, $"{Path.GetFileNameWithoutExtension(target.Result.FilePath)}-merged-{mergeId[..14]}{extension}");
var stagingPath = mergedPath + ".partial.mp4";
var recoveryDirectory = Path.Combine(directory, ".liverecorder-recovery", session.Id.ToString("N"), mergeId);
Directory.CreateDirectory(recoveryDirectory);
var manifest = new MergeRecoveryManifest(
mergeId,
session.Id,
target.Id,
"staging",
mergedPath,
recoveryDirectory,
group.Select(item => new MergeRecoverySource(item.Id, item.Result!.FilePath, item.Result.DanmakuFilePath, item.Result.DurationSeconds ?? 0)).ToArray(),
DateTimeOffset.UtcNow);
await WriteManifestAsync(recoveryDirectory, manifest, cancellationToken);
var settings = await _settingsService.GetAsync(cancellationToken);
var concatFile = Path.Combine(recoveryDirectory, "concat.txt");
await File.WriteAllLinesAsync(concatFile, sources.Select(path => $"file '{EscapeConcatPath(Path.GetFullPath(path))}'"), cancellationToken);
var copied = await RunFfmpegAsync(settings.FfmpegPath,
["-hide_banner", "-loglevel", "warning", "-f", "concat", "-safe", "0", "-i", concatFile, "-map", "0", "-c", "copy", "-movflags", "+faststart", "-y", stagingPath],
cancellationToken);
if (!copied)
{
copied = await RunFfmpegAsync(settings.FfmpegPath,
["-hide_banner", "-loglevel", "warning", "-f", "concat", "-safe", "0", "-i", concatFile, "-map", "0:v:0", "-map", "0:a?", "-c:v", "libx264", "-preset", "fast", "-crf", "20", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", "-y", stagingPath],
cancellationToken);
}
var metadata = copied ? await _metadataService.ExtractMetadataAsync(stagingPath, cancellationToken) : null;
if (metadata?.DurationSeconds is null or <= 0 || string.IsNullOrWhiteSpace(metadata.VideoCodec))
{
return false;
}
File.Move(stagingPath, mergedPath, overwrite: false);
var mergedDanmakuPath = await MergeDanmakuAsync(group, target, mergedPath, cancellationToken);
foreach (var source in group)
{
CopyToRecovery(source.Result!.FilePath, recoveryDirectory);
CopyToRecovery(source.Result.DanmakuFilePath, recoveryDirectory);
}
var now = DateTimeOffset.UtcNow;
target.Result.Update(
mergedPath,
new FileInfo(mergedPath).Length,
metadata.DurationSeconds,
mergedDanmakuPath,
group.Sum(item => item.Result!.DanmakuMessageCount),
RecordTaskStatus.Completed,
null);
target.MarkCompleted(now, metadata.DurationSeconds);
foreach (var source in group.Where(item => item.Id != target.Id))
{
source.MarkMergedSource(target.Id, now);
var dispatch = await _dbContext.RecordCompletionDispatches.FirstOrDefaultAsync(item => item.RecordTaskId == source.Id, cancellationToken);
dispatch?.MarkSkipped($"Merged into {target.Id}", now);
}
if (!await _dbContext.RecordCompletionDispatches.AnyAsync(item => item.RecordTaskId == target.Id, cancellationToken))
{
await _dbContext.RecordCompletionDispatches.AddAsync(new RecordCompletionDispatch(target.Id, now), cancellationToken);
}
await _dbContext.SaveChangesAsync(cancellationToken);
foreach (var source in manifest.Sources)
{
DeleteOriginal(source.VideoPath, mergedPath);
DeleteOriginal(source.DanmakuPath, mergedDanmakuPath);
}
await WriteManifestAsync(recoveryDirectory, manifest with { Phase = "completed" }, cancellationToken);
return true;
}
private async Task MoveRecoveryOnlyAsync(IReadOnlyList<RecordTask> group, CancellationToken cancellationToken)
{
if (group.Count == 0) return;
var sessionId = group[0].RecordSessionId;
var mergeId = $"recovery-only-{DateTimeOffset.UtcNow:yyyyMMddHHmmss}-{Guid.NewGuid():N}";
var directory = Path.GetDirectoryName(Path.GetFullPath(group[0].Result!.FilePath))!;
var recoveryDirectory = Path.Combine(directory, ".liverecorder-recovery", sessionId.ToString("N"), mergeId);
Directory.CreateDirectory(recoveryDirectory);
var manifest = new MergeRecoveryManifest(
mergeId,
sessionId,
group[0].Id,
"staging",
string.Empty,
recoveryDirectory,
group.Select(item => new MergeRecoverySource(item.Id, item.Result!.FilePath, item.Result.DanmakuFilePath, item.Result.DurationSeconds ?? 0)).ToArray(),
DateTimeOffset.UtcNow);
await WriteManifestAsync(recoveryDirectory, manifest, cancellationToken);
foreach (var item in group)
{
CopyToRecovery(item.Result!.FilePath, recoveryDirectory);
CopyToRecovery(item.Result.DanmakuFilePath, recoveryDirectory);
item.MarkRecoveryOnly(DateTimeOffset.UtcNow);
var dispatch = await _dbContext.RecordCompletionDispatches.FirstOrDefaultAsync(d => d.RecordTaskId == item.Id, cancellationToken);
dispatch?.MarkSkipped("Recovery-only short fragment", DateTimeOffset.UtcNow);
}
await _dbContext.SaveChangesAsync(cancellationToken);
foreach (var item in group)
{
DeleteOriginal(item.Result!.FilePath, null);
DeleteOriginal(item.Result.DanmakuFilePath, null);
}
await WriteManifestAsync(recoveryDirectory, manifest with { Phase = "completed" }, cancellationToken);
}
private static async Task<string?> MergeDanmakuAsync(
IReadOnlyList<RecordTask> group,
RecordTask target,
string mergedVideoPath,
CancellationToken cancellationToken)
{
var available = group.Where(item => !string.IsNullOrWhiteSpace(item.Result?.DanmakuFilePath) && File.Exists(item.Result.DanmakuFilePath)).ToList();
if (available.Count == 0) return null;
var root = new XElement("i",
new XAttribute("recordSessionId", target.RecordSessionId),
new XAttribute("recordTaskId", target.Id),
new XAttribute("merged", true));
var cumulative = 0d;
foreach (var item in group)
{
if (!string.IsNullOrWhiteSpace(item.Result?.DanmakuFilePath) && File.Exists(item.Result.DanmakuFilePath))
{
var source = await XDocument.LoadAsync(File.OpenRead(item.Result.DanmakuFilePath), LoadOptions.None, cancellationToken);
foreach (var element in source.Root?.Elements() ?? [])
{
root.Add(AdjustDanmakuElementOffsets(element, cumulative));
}
}
cumulative += item.Result?.DurationSeconds ?? 0;
}
var output = Path.ChangeExtension(mergedVideoPath, ".xml");
await using var stream = File.Create(output);
await new XDocument(new XDeclaration("1.0", "UTF-8", null), root).SaveAsync(stream, SaveOptions.None, cancellationToken);
return output;
}
internal static XElement AdjustDanmakuElementOffsets(XElement element, double cumulativeSeconds)
{
var clone = new XElement(element);
if (clone.Name.LocalName == "d" && clone.Attribute("p") is { } p)
{
var parts = p.Value.Split(',');
if (parts.Length > 0 && double.TryParse(parts[0], NumberStyles.Float, CultureInfo.InvariantCulture, out var offset))
{
parts[0] = (offset + cumulativeSeconds).ToString("F1", CultureInfo.InvariantCulture);
p.Value = string.Join(',', parts);
}
}
else if (clone.Attribute("offset") is { } offsetAttribute &&
double.TryParse(offsetAttribute.Value, NumberStyles.Float, CultureInfo.InvariantCulture, out var offset))
{
offsetAttribute.Value = (offset + cumulativeSeconds).ToString("F1", CultureInfo.InvariantCulture);
}
return clone;
}
private async Task<bool> RunFfmpegAsync(string path, IReadOnlyList<string> arguments, CancellationToken cancellationToken)
{
using var process = new Process { StartInfo = new ProcessStartInfo(path) { UseShellExecute = false, RedirectStandardError = true, RedirectStandardOutput = true, CreateNoWindow = true } };
foreach (var argument in arguments) process.StartInfo.ArgumentList.Add(argument);
try
{
process.Start();
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
timeout.CancelAfter(ProcessTimeout);
var stderr = process.StandardError.ReadToEndAsync(timeout.Token);
var stdout = process.StandardOutput.ReadToEndAsync(timeout.Token);
await process.WaitForExitAsync(timeout.Token);
await Task.WhenAll(stderr, stdout);
if (process.ExitCode == 0) return true;
_logger.LogWarning("Short fragment merge ffmpeg failed: {Error}", await stderr);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Short fragment merge process failed");
try { if (!process.HasExited) process.Kill(true); } catch { }
}
return false;
}
private static bool IsShort(RecordTask task) =>
IsReadableTerminal(task) && task.Result!.DurationSeconds!.Value < MinimumStandaloneDurationSeconds;
private static bool IsReadableTerminal(RecordTask task) =>
task.Status is RecordTaskStatus.Completed or RecordTaskStatus.Stopped &&
task.Result?.DurationSeconds is > 0 && File.Exists(task.Result.FilePath);
private static bool IsActive(RecordTaskStatus status) => status is RecordTaskStatus.Pending or RecordTaskStatus.Starting or RecordTaskStatus.Running or RecordTaskStatus.Stopping or RecordTaskStatus.Processing;
private static bool IsActive(RecordSessionStatus status) => status is RecordSessionStatus.Pending or RecordSessionStatus.Starting or RecordSessionStatus.Running or RecordSessionStatus.Stopping;
private static string EscapeConcatPath(string path) => path.Replace("'", "'\\''", StringComparison.Ordinal);
private static void CopyToRecovery(string? source, string directory)
{
if (string.IsNullOrWhiteSpace(source) || !File.Exists(source)) return;
var destination = Path.Combine(directory, Path.GetFileName(source));
if (!File.Exists(destination)) File.Copy(source, destination);
}
private static void DeleteOriginal(string? source, string? preserved)
{
if (string.IsNullOrWhiteSpace(source) || string.Equals(Path.GetFullPath(source), preserved is null ? null : Path.GetFullPath(preserved), StringComparison.OrdinalIgnoreCase)) return;
if (File.Exists(source)) File.Delete(source);
}
private static async Task WriteManifestAsync(string directory, MergeRecoveryManifest manifest, CancellationToken cancellationToken)
{
var path = Path.Combine(directory, "manifest.json");
var temp = path + ".tmp";
await File.WriteAllTextAsync(temp, JsonSerializer.Serialize(manifest, new JsonSerializerOptions { WriteIndented = true }), cancellationToken);
File.Move(temp, path, true);
}
}
public sealed record FragmentDispatchDecision(bool CanDispatch, bool SkipDispatch, string? Reason, TimeSpan RetryDelay, Guid? MergedIntoRecordTaskId)
{
public static readonly FragmentDispatchDecision Ready = new(true, false, null, TimeSpan.Zero, null);
public static FragmentDispatchDecision Wait(string reason, TimeSpan delay) => new(false, false, reason, delay, null);
public static FragmentDispatchDecision Skip(string reason, Guid? target) => new(false, true, reason, TimeSpan.Zero, target);
}
internal sealed record MergeRecoveryManifest(string MergeId, Guid RecordSessionId, Guid TargetRecordTaskId, string Phase, string MergedPath, string RecoveryDirectory, IReadOnlyList<MergeRecoverySource> Sources, DateTimeOffset CreatedAt);
internal sealed record MergeRecoverySource(Guid RecordTaskId, string VideoPath, string? DanmakuPath, double DurationSeconds);
@@ -19,6 +19,7 @@ public sealed class RecordTasksController : ControllerBase
private readonly IRecordMediaService _recordMediaService;
private readonly IDanmakuService _danmakuService;
private readonly LinkGenerator _linkGenerator;
private readonly OpenListUploadQueueService _openListUploadQueue;
public RecordTasksController(
RecordService recordService,
@@ -26,7 +27,8 @@ public sealed class RecordTasksController : ControllerBase
IRecordResultRepository recordResultRepository,
IRecordMediaService recordMediaService,
IDanmakuService danmakuService,
LinkGenerator linkGenerator)
LinkGenerator linkGenerator,
OpenListUploadQueueService openListUploadQueue)
{
_recordService = recordService;
_recordUploadService = recordUploadService;
@@ -34,6 +36,7 @@ public sealed class RecordTasksController : ControllerBase
_recordMediaService = recordMediaService;
_danmakuService = danmakuService;
_linkGenerator = linkGenerator;
_openListUploadQueue = openListUploadQueue;
}
[HttpGet]
@@ -43,6 +46,7 @@ public sealed class RecordTasksController : ControllerBase
[HttpGet("upload-status")]
public async Task<ActionResult<UploadTaskListResponse>> ListUploadStatus(
[FromQuery] int? uploadStatus,
[FromQuery] string? query,
[FromQuery] int skip = 0,
[FromQuery] int take = 50,
CancellationToken cancellationToken = default)
@@ -51,15 +55,27 @@ public sealed class RecordTasksController : ControllerBase
? (RecordArtifactUploadStatus)uploadStatus.Value
: (RecordArtifactUploadStatus?)null;
var items = await _recordResultRepository.ListUploadStatusAsync(filter, skip, take, cancellationToken);
var totalCount = await _recordResultRepository.CountUploadStatusAsync(filter, cancellationToken);
var items = await _recordResultRepository.ListUploadStatusAsync(filter, query, skip, take, cancellationToken);
var totalCount = await _recordResultRepository.CountUploadStatusAsync(filter, query, cancellationToken);
var notUploadedCount = await _recordResultRepository.CountPendingUploadAsync(cancellationToken);
var failedArtifactCount = await _recordResultRepository.CountFailedArtifactAsync(cancellationToken);
var succeededCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Succeeded, cancellationToken);
var failedCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Failed, cancellationToken);
var queuedCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Queued, cancellationToken);
var uploadingCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Uploading, cancellationToken);
var waitingRetryCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.WaitingRetry, cancellationToken);
var succeededCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Succeeded, null, cancellationToken);
var failedCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Failed, null, cancellationToken);
var queuedCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Queued, null, cancellationToken);
var uploadingCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Uploading, null, cancellationToken);
var waitingRetryCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.WaitingRetry, null, cancellationToken);
var hasQuery = !string.IsNullOrWhiteSpace(query);
var matchingFailedCount = filter is null or RecordArtifactUploadStatus.Failed
? hasQuery
? await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Failed, query, cancellationToken)
: failedCount
: 0;
var matchingWaitingRetryCount = filter is null or RecordArtifactUploadStatus.WaitingRetry
? hasQuery
? await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.WaitingRetry, query, cancellationToken)
: waitingRetryCount
: 0;
var health = _openListUploadQueue.GetHealthSnapshot();
return Ok(new UploadTaskListResponse
{
@@ -71,7 +87,16 @@ public sealed class RecordTasksController : ControllerBase
FailedCount = failedCount,
QueuedCount = queuedCount,
UploadingCount = uploadingCount,
WaitingRetryCount = waitingRetryCount
WaitingRetryCount = waitingRetryCount,
MatchingRetryableCount = matchingFailedCount + matchingWaitingRetryCount,
QueueHealth = new UploadQueueHealthDto
{
State = health.Status.ToString(),
IsPaused = health.IsPaused,
Reason = health.Reason,
RetryAt = health.RetryAt,
LastErrorAt = health.LastErrorAt
}
});
}
@@ -180,6 +205,22 @@ public sealed class RecordTasksController : ControllerBase
: Ok(result);
}
[HttpPost("{id:guid}/upload/retry")]
public async Task<ActionResult<RecordArtifactUploadItemResultDto>> RetryUpload(Guid id, CancellationToken cancellationToken)
{
var result = await _recordUploadService.RetryTaskAsync(id, cancellationToken);
return result.Success ? Accepted(result) : BadRequest(result);
}
[HttpPost("upload/retry")]
public async Task<ActionResult<RecordArtifactUploadBatchResultDto>> RetryUploads(
[FromBody] RetryRecordArtifactUploadsRequest request,
CancellationToken cancellationToken)
{
var result = await _recordUploadService.RetryMatchingAsync(request, cancellationToken);
return result.FailedCount > 0 && result.SuccessCount == 0 ? BadRequest(result) : Accepted(result);
}
[HttpGet("{id:guid}/danmaku")]
public async Task<ActionResult<DanmakuResponseDto>> GetDanmaku(Guid id, CancellationToken cancellationToken)
{
@@ -34,4 +34,25 @@ public sealed class RecoveryController : ControllerBase
[HttpPost("finalizations/resume-all")]
public async Task<ActionResult<RecoveryActionResultDto>> ResumeAllFinalizations(CancellationToken cancellationToken) =>
Ok(await _recoveryService.ResumeAllFinalizationsAsync(cancellationToken));
[HttpGet("recording-failures")]
public async Task<ActionResult<RecordingFailureListResponse>> ListRecordingFailures(
[FromQuery] string? failureKind,
[FromQuery] int skip = 0,
[FromQuery] int take = 20,
CancellationToken cancellationToken = default) =>
Ok(await _recoveryService.ListRecordingFailuresAsync(failureKind, skip, take, cancellationToken));
[HttpPost("recording-failures/{taskId:guid}/accept")]
public async Task<ActionResult<RecoveryActionResultDto>> AcceptRecordingFailure(
Guid taskId,
[FromBody] AcceptRecordingArtifactRequest request,
CancellationToken cancellationToken) =>
Ok(await _recoveryService.AcceptRecordingArtifactAsync(taskId, request.ConfirmShortArtifact, cancellationToken));
[HttpPost("recording-failures/{taskId:guid}/repair")]
public async Task<ActionResult<RecoveryActionResultDto>> RepairRecordingFailure(
Guid taskId,
CancellationToken cancellationToken) =>
Ok(await _recoveryService.RepairRecordingArtifactAsync(taskId, cancellationToken));
}
@@ -0,0 +1,44 @@
using LiveRecorder.Application.Models.RecordTasks;
using LiveRecorder.Infrastructure.Services;
using Microsoft.AspNetCore.Mvc;
namespace LiveRecorder.WebApi.Controllers;
[ApiController]
[Route("api/upload-queue")]
public sealed class UploadQueueController : ControllerBase
{
private readonly OpenListUploadQueueService _queue;
public UploadQueueController(OpenListUploadQueueService queue)
{
_queue = queue;
}
[HttpPost("openlist/resume")]
public async Task<ActionResult<UploadQueueHealthDto>> ResumeOpenList(CancellationToken cancellationToken)
{
try
{
var health = await _queue.ResumeProviderAsync(cancellationToken);
return Ok(Map(health));
}
catch (Exception ex)
{
return BadRequest(new
{
message = ex.Message,
queueHealth = Map(_queue.GetHealthSnapshot())
});
}
}
private static UploadQueueHealthDto Map(OpenListQueueHealthSnapshot health) => new()
{
State = health.Status.ToString(),
IsPaused = health.IsPaused,
Reason = health.Reason,
RetryAt = health.RetryAt,
LastErrorAt = health.LastErrorAt
};
}
+2
View File
@@ -211,7 +211,9 @@ builder.Services.AddScoped<PlatformHttpClientFactory>();
builder.Services.AddScoped<PlatformHttpRequestService>();
builder.Services.AddScoped<RecordUploadService>();
builder.Services.AddScoped<CompletionDispatchService>();
builder.Services.AddScoped<ShortFragmentConsolidationService>();
builder.Services.AddScoped<OpenListUploadQueueService>();
builder.Services.AddSingleton<OpenListUploadHealthState>();
builder.Services.AddSingleton<IOpenListClient, OpenListClient>();
builder.Services.AddScoped<IDanmakuService, DanmakuService>();
builder.Services.AddScoped<IVideoMetadataService, FfmpegVideoMetadataService>();