diff --git a/frontend/src/components/ui/StorageCapacity.vue b/frontend/src/components/ui/StorageCapacity.vue index 9dab83f..b2305e1 100644 --- a/frontend/src/components/ui/StorageCapacity.vue +++ b/frontend/src/components/ui/StorageCapacity.vue @@ -5,6 +5,7 @@ interface StorageCapacityStatus { isEnabled: boolean; isAvailable: boolean; checkedPath: string; + volumeRoot: string; totalBytes: number; usedBytes: number; availableBytes: number; @@ -83,6 +84,9 @@ function formatBytes(bytes: number) { {{ statusLabel }}

{{ status.checkedPath || "未配置输出路径" }}

+

+ 检测卷：{{ status.volumeRoot || "--" }} +

已使用
{{ status.isAvailable ? formatBytes(status.usedBytes) : "--" }}
@@ -103,6 +107,7 @@ function formatBytes(bytes: number) { .storage-capacity__summary { display: flex; align-items: center; justify-content: space-between; gap: 12px; } .storage-capacity__summary > strong { min-width: 0; color: var(--text-primary); font-size: 13px; font-variant-numeric: tabular-nums; } .storage-capacity__path { margin: 7px 0 0; overflow: hidden; color: var(--text-secondary); font-family: var(--font-mono); font-size: 10px; text-overflow: ellipsis; white-space: nowrap; } +.storage-capacity__volume { margin: 3px 0 0; overflow: hidden; color: var(--text-muted); font-family: var(--font-mono); font-size: 9px; text-overflow: ellipsis; white-space: nowrap; } .storage-capacity__metrics { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: 8px; margin: 13px 0 0; } .storage-capacity__metrics > div { min-width: 0; padding: 9px; border-radius: var(--radius-sm); background: var(--surface-muted); } .storage-capacity__metrics dt { color: var(--text-muted); font-size: 9px; } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index bac1006..b273865 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -703,6 +703,7 @@ export interface StorageGuardStatus { isAvailable: boolean; hasEnoughSpace: boolean; checkedPath: string; + volumeRoot: string; totalBytes: number; usedBytes: number; availableBytes: number; @@ -1011,6 +1012,7 @@ export interface DashboardStorageStatus { hasEnoughSpace: boolean; message: string; checkedPath: string; + volumeRoot: string; totalBytes: number; usedBytes: number; availableBytes: number; diff --git a/src/LiveRecorder.Application/Abstractions/Persistence/PersistenceContracts.cs b/src/LiveRecorder.Application/Abstractions/Persistence/PersistenceContracts.cs index d0d83d8..2180a9b 100644 --- a/src/LiveRecorder.Application/Abstractions/Persistence/PersistenceContracts.cs +++ b/src/LiveRecorder.Application/Abstractions/Persistence/PersistenceContracts.cs @@ -116,6 +116,8 @@ public interface IRecordResultRepository Task SumPendingUploadBytesAsync(CancellationToken cancellationToken = default); + Task GetPendingUploadMetricsAsync(CancellationToken cancellationToken = default); + Task CountFailedArtifactAsync(CancellationToken cancellationToken = default); Task> ListUploadStatusAsync( @@ -202,6 +204,8 @@ public interface IUserSessionRepository void Update(UserSession session); } +public sealed record PendingUploadMetrics(int Count, long TotalBytes); + public interface IUnitOfWork { Task SaveChangesAsync(CancellationToken cancellationToken = default); diff --git a/src/LiveRecorder.Application/Abstractions/Recording/IFfmpegService.cs b/src/LiveRecorder.Application/Abstractions/Recording/IFfmpegService.cs index 46becc6..ab18e05 100644 --- a/src/LiveRecorder.Application/Abstractions/Recording/IFfmpegService.cs +++ b/src/LiveRecorder.Application/Abstractions/Recording/IFfmpegService.cs @@ -37,6 +37,8 @@ public interface IFfmpegService Task RecoverOrphanedTerminalSessionTasksAsync(CancellationToken cancellationToken = default); + Task RecoverStalePendingSessionsAsync(CancellationToken cancellationToken = default); + Task TryRecoverOrphanedTerminalTaskAsync(Guid recordTaskId, CancellationToken cancellationToken = default); Task StartManualFinalizeTaskAsync(Guid recordTaskId, CancellationToken cancellationToken = default); diff --git a/src/LiveRecorder.Application/Abstractions/Storage/IStorageGuardService.cs b/src/LiveRecorder.Application/Abstractions/Storage/IStorageGuardService.cs index e3db747..da09273 100644 --- a/src/LiveRecorder.Application/Abstractions/Storage/IStorageGuardService.cs +++ b/src/LiveRecorder.Application/Abstractions/Storage/IStorageGuardService.cs @@ -23,6 +23,16 @@ public interface IStorageGuardService StorageGuardResult CheckCanFinalize(SystemSettingsDto settings, long estimatedTemporaryBytes); } +public interface IStorageCapacityProvider +{ + StorageVolumeCapacity GetCapacity(string outputPath); +} + +public sealed record StorageVolumeCapacity( + string VolumeRoot, + long TotalBytes, + long AvailableBytes); + public sealed record StorageGuardResult( bool IsEnabled, bool HasEnoughSpace, @@ -33,6 +43,8 @@ public sealed record StorageGuardResult( { public bool IsAvailable { get; init; } + public string VolumeRoot { get; init; } = string.Empty; + public long TotalBytes { get; init; } public long UsedBytes { get; init; } diff --git a/src/LiveRecorder.Application/Models/Recovery/RecoveryModels.cs b/src/LiveRecorder.Application/Models/Recovery/RecoveryModels.cs index fbcac4b..567e2bd 100644 --- a/src/LiveRecorder.Application/Models/Recovery/RecoveryModels.cs +++ b/src/LiveRecorder.Application/Models/Recovery/RecoveryModels.cs @@ -36,6 +36,8 @@ public sealed class StorageGuardStatusDto public required string CheckedPath { get; init; } + public string VolumeRoot { get; init; } = string.Empty; + public long AvailableBytes { get; init; } public long TotalBytes { get; init; } diff --git a/src/LiveRecorder.Application/Models/Reports/DashboardModels.cs b/src/LiveRecorder.Application/Models/Reports/DashboardModels.cs index 8522db3..2fddd67 100644 --- a/src/LiveRecorder.Application/Models/Reports/DashboardModels.cs +++ b/src/LiveRecorder.Application/Models/Reports/DashboardModels.cs @@ -101,6 +101,7 @@ public sealed class StorageStatusDto public bool HasEnoughSpace { get; init; } public string Message { get; init; } = string.Empty; public string CheckedPath { get; init; } = string.Empty; + public string VolumeRoot { get; init; } = string.Empty; public long TotalBytes { get; init; } public long UsedBytes { get; init; } public long AvailableBytes { get; init; } diff --git a/src/LiveRecorder.Application/Services/DashboardService.cs b/src/LiveRecorder.Application/Services/DashboardService.cs index 7d4269d..1d8bf21 100644 --- a/src/LiveRecorder.Application/Services/DashboardService.cs +++ b/src/LiveRecorder.Application/Services/DashboardService.cs @@ -65,8 +65,7 @@ public sealed class DashboardService var settings = await _systemSettingsService.GetAsync(cancellationToken); var storageCheck = _storageGuardService.CheckCanStartOrResume(settings); var pendingTranscodeCount = await _recordTaskRepository.CountByStatusAsync(RecordTaskStatus.Processing, cancellationToken); - var pendingUploadCount = await _recordResultRepository.CountPendingUploadAsync(cancellationToken); - var queuedDataBytes = await _recordResultRepository.SumPendingUploadBytesAsync(cancellationToken); + var pendingUploadMetrics = await _recordResultRepository.GetPendingUploadMetricsAsync(cancellationToken); var operationsMetrics = await _operationsMetricsRepository.GetAsync(cancellationToken); return new DashboardDto @@ -88,6 +87,7 @@ public sealed class DashboardService HasEnoughSpace = storageCheck.HasEnoughSpace, Message = storageCheck.Message ?? string.Empty, CheckedPath = storageCheck.CheckedPath, + VolumeRoot = storageCheck.VolumeRoot, TotalBytes = storageCheck.TotalBytes, UsedBytes = storageCheck.UsedBytes, AvailableBytes = storageCheck.AvailableBytes, @@ -99,8 +99,8 @@ public sealed class DashboardService RedThresholdPercent = storageCheck.RedThresholdPercent }, PendingTranscodeCount = pendingTranscodeCount, - PendingUploadCount = pendingUploadCount, - QueuedDataBytes = queuedDataBytes, + PendingUploadCount = pendingUploadMetrics.Count, + QueuedDataBytes = pendingUploadMetrics.TotalBytes, OldestTranscodeUpdatedAt = operationsMetrics.OldestTranscodeUpdatedAt, OldestUploadProgressAt = operationsMetrics.OldestUploadProgressAt, StalledUploadCount = operationsMetrics.StalledUploadCount, diff --git a/src/LiveRecorder.Application/Services/RecordService.cs b/src/LiveRecorder.Application/Services/RecordService.cs index 26d00fa..640f87b 100644 --- a/src/LiveRecorder.Application/Services/RecordService.cs +++ b/src/LiveRecorder.Application/Services/RecordService.cs @@ -352,8 +352,9 @@ public sealed class RecordService } catch (Exception ex) { - initialTask.MarkFailed(ex.Message, DateTimeOffset.UtcNow); - recordSession.MarkFailed(ex.Message, DateTimeOffset.UtcNow); + var failedAt = DateTimeOffset.UtcNow; + initialTask.MarkFailed(ex.Message, failedAt); + recordSession.MarkFailed(ex.Message, failedAt); if (trackAutoStartDecision) { ApplyAutoStartDecision( @@ -363,7 +364,8 @@ public sealed class RecordService ex.Message); } - await _unitOfWork.SaveChangesAsync(cancellationToken); + using var terminalStateCts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await _unitOfWork.SaveChangesAsync(terminalStateCts.Token); await _systemLogService.WriteAsync( SystemLogLevel.Error, @@ -373,7 +375,12 @@ public sealed class RecordService liveRoom.Id, recordSession.Id, initialTask.Id, - cancellationToken); + terminalStateCts.Token); + + if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested) + { + throw; + } await _emailNotificationService.SendExceptionAsync( "RecordSession", diff --git a/src/LiveRecorder.Infrastructure/Persistence/Repositories/Repositories.cs b/src/LiveRecorder.Infrastructure/Persistence/Repositories/Repositories.cs index a5a4b86..a6a11b5 100644 --- a/src/LiveRecorder.Infrastructure/Persistence/Repositories/Repositories.cs +++ b/src/LiveRecorder.Infrastructure/Persistence/Repositories/Repositories.cs @@ -452,9 +452,9 @@ public sealed class RecordSessionRepository : IRecordSessionRepository public Task CountActiveAsync(CancellationToken cancellationToken = default) => _dbContext.RecordSessions.CountAsync(item => - item.Status == RecordSessionStatus.Pending || item.Status == RecordSessionStatus.Starting || - item.Status == RecordSessionStatus.Running, cancellationToken); + item.Status == RecordSessionStatus.Running || + item.Status == RecordSessionStatus.Stopping, cancellationToken); public async Task> ListRecentAsync(int take, CancellationToken cancellationToken = default) => await _dbContext.RecordSessions @@ -508,21 +508,49 @@ public sealed class RecordResultRepository : IRecordResultRepository return (result?.TotalBytes ?? 0L, result?.TotalDanmaku ?? 0); } - public Task CountPendingUploadAsync(CancellationToken cancellationToken = default) => - _dbContext.RecordResults.CountAsync( - item => item.UploadStatus == RecordArtifactUploadStatus.NotUploaded && - item.RecordTask != null && - (item.RecordTask.Status == RecordTaskStatus.Completed || - item.RecordTask.Status == RecordTaskStatus.Stopped), - cancellationToken); + public async Task CountPendingUploadAsync(CancellationToken cancellationToken = default) => + (await GetPendingUploadMetricsAsync(cancellationToken)).Count; - public Task SumPendingUploadBytesAsync(CancellationToken cancellationToken = default) => - _dbContext.RecordResults + public async Task SumPendingUploadBytesAsync(CancellationToken cancellationToken = default) => + (await GetPendingUploadMetricsAsync(cancellationToken)).TotalBytes; + + public async Task GetPendingUploadMetricsAsync(CancellationToken cancellationToken = default) + { + var candidatePaths = await _dbContext.RecordResults + .AsNoTracking() .Where(item => item.UploadStatus == RecordArtifactUploadStatus.NotUploaded && item.RecordTask != null && + !item.RecordTask.IsHiddenArtifactSource && + item.RecordTask.MergedIntoRecordTaskId == null && (item.RecordTask.Status == RecordTaskStatus.Completed || - item.RecordTask.Status == RecordTaskStatus.Stopped)) - .SumAsync(item => item.FileSizeBytes ?? 0L, cancellationToken); + item.RecordTask.Status == RecordTaskStatus.Stopped) && + item.FilePath != null && + item.FilePath != string.Empty) + .Select(item => item.FilePath) + .ToListAsync(cancellationToken); + + var count = 0; + long totalBytes = 0; + foreach (var path in candidatePaths) + { + try + { + var file = new FileInfo(path); + if (!file.Exists || file.Length <= 0) + { + continue; + } + + count++; + totalBytes += file.Length; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or ArgumentException) + { + } + } + + return new PendingUploadMetrics(count, totalBytes); + } public Task CountFailedArtifactAsync(CancellationToken cancellationToken = default) => _dbContext.RecordResults.CountAsync( diff --git a/src/LiveRecorder.Infrastructure/Services/DriveInfoStorageCapacityProvider.cs b/src/LiveRecorder.Infrastructure/Services/DriveInfoStorageCapacityProvider.cs new file mode 100644 index 0000000..abe159a --- /dev/null +++ b/src/LiveRecorder.Infrastructure/Services/DriveInfoStorageCapacityProvider.cs @@ -0,0 +1,89 @@ +using LiveRecorder.Application.Abstractions.Storage; + +namespace LiveRecorder.Infrastructure.Services; + +public sealed class DriveInfoStorageCapacityProvider : IStorageCapacityProvider +{ + public StorageVolumeCapacity GetCapacity(string outputPath) + { + var volumes = new List(); + foreach (var drive in DriveInfo.GetDrives()) + { + try + { + if (drive.IsReady) + { + volumes.Add(new StorageVolumeCapacity( + drive.RootDirectory.FullName, + drive.TotalSize, + drive.AvailableFreeSpace)); + } + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + } + } + + var selected = SelectBestVolume(outputPath, volumes); + if (selected is not null) + { + return selected; + } + + var fallbackRoot = Path.GetPathRoot(outputPath); + if (string.IsNullOrWhiteSpace(fallbackRoot)) + { + throw new InvalidOperationException($"No mounted volume contains output path '{outputPath}'."); + } + + var fallback = new DriveInfo(fallbackRoot); + if (!fallback.IsReady) + { + throw new IOException($"Storage volume '{fallbackRoot}' is not ready."); + } + + return new StorageVolumeCapacity( + fallback.RootDirectory.FullName, + fallback.TotalSize, + fallback.AvailableFreeSpace); + } + + internal static StorageVolumeCapacity? SelectBestVolume( + string outputPath, + IReadOnlyCollection volumes) + { + var normalizedPath = NormalizeForMountComparison(outputPath); + return volumes + .Select(volume => new + { + Volume = volume, + NormalizedRoot = NormalizeForMountComparison(volume.VolumeRoot) + }) + .Where(item => IsPathWithinMount(normalizedPath, item.NormalizedRoot)) + .OrderByDescending(item => item.NormalizedRoot.Length) + .Select(item => item.Volume) + .FirstOrDefault(); + } + + private static bool IsPathWithinMount(string path, string mountRoot) + { + if (mountRoot == "/") + { + return path.StartsWith("/", StringComparison.Ordinal); + } + + return string.Equals(path, mountRoot, StringComparison.OrdinalIgnoreCase) || + path.StartsWith(mountRoot + "/", StringComparison.OrdinalIgnoreCase); + } + + private static string NormalizeForMountComparison(string path) + { + var normalized = path.Trim().Replace((char)92, '/'); + if (normalized.Length > 1) + { + normalized = normalized.TrimEnd('/'); + } + + return normalized; + } +} diff --git a/src/LiveRecorder.Infrastructure/Services/FfmpegService.Recovery.cs b/src/LiveRecorder.Infrastructure/Services/FfmpegService.Recovery.cs index bc6e3a7..803da18 100644 --- a/src/LiveRecorder.Infrastructure/Services/FfmpegService.Recovery.cs +++ b/src/LiveRecorder.Infrastructure/Services/FfmpegService.Recovery.cs @@ -49,6 +49,21 @@ public sealed partial class FfmpegService !hasTriedCurlFallback && string.Equals(selectedProtocol, "flv", StringComparison.OrdinalIgnoreCase); + internal static FfmpegRecoveryContext PrepareRefreshedStreamRecoveryContext( + FfmpegRecoveryContext current, + string selectedProtocol, + bool hasOverlongHeadersFailure) + { + var useCurlFallback = hasOverlongHeadersFailure && + string.Equals(selectedProtocol, "flv", StringComparison.OrdinalIgnoreCase); + + return current with + { + UseCurlFallback = useCurlFallback, + HasTriedCurlFallback = current.HasTriedCurlFallback || useCurlFallback + }; + } + internal static bool ShouldApplyRuntimeFailureBackoff( RecordSessionStatus status, TimeSpan processRuntime) => diff --git a/src/LiveRecorder.Infrastructure/Services/FfmpegService.Runtime.cs b/src/LiveRecorder.Infrastructure/Services/FfmpegService.Runtime.cs index 63667d4..fa63486 100644 --- a/src/LiveRecorder.Infrastructure/Services/FfmpegService.Runtime.cs +++ b/src/LiveRecorder.Infrastructure/Services/FfmpegService.Runtime.cs @@ -37,7 +37,10 @@ public sealed partial class FfmpegService TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(60), - TimeSpan.FromSeconds(120) + TimeSpan.FromSeconds(120), + TimeSpan.FromSeconds(300), + TimeSpan.FromSeconds(600), + TimeSpan.FromSeconds(900) ]; private async Task HandleProcessOutputAsync(SessionProcessRuntime runtime, string? line, bool isError) @@ -85,7 +88,10 @@ public sealed partial class FfmpegService runtime.MarkStartupFailure(StartupFailureKind.StreamHandshake, line); } - if (TryClassifyPersistedFfmpegLine(line, isError, out var level)) + if (ShouldPersistFfmpegLine( + runtime.HasOpenedFirstSegment, + runtime.StartupFailureKind == StartupFailureKind.StreamHandshake) && + TryClassifyPersistedFfmpegLine(line, isError, out var level)) { await PersistFfmpegLineAsync(runtime, line, level); } @@ -728,11 +734,13 @@ public sealed partial class FfmpegService var stream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality, transition.Token); var selected = SelectRetryStreamForCurrentSession(stream, runtime); - var context = runtime.RecoveryContext with + var context = PrepareRefreshedStreamRecoveryContext(runtime.RecoveryContext with { AttemptCount = attempt, HasRetriedWithRefreshedStream = true - }; + }, + selected.SelectedProtocol, + runtime.HasHlsOverlongHeadersFailure || runtime.RecoveryContext.HasTriedCurlFallback); 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); @@ -1036,12 +1044,15 @@ public sealed partial class FfmpegService currentTask, streamForRetry, runtime.RecordingSettings, - AdvanceRecoveryContext( - runtime.RecoveryContext, - runtime.InputOptionProfile, - runtime.SelectedProtocol, + PrepareRefreshedStreamRecoveryContext( + AdvanceRecoveryContext( + runtime.RecoveryContext, + runtime.InputOptionProfile, + runtime.SelectedProtocol, + streamForRetry.SelectedProtocol, + refreshedStream: true), streamForRetry.SelectedProtocol, - refreshedStream: true)); + runtime.HasHlsOverlongHeadersFailure || runtime.RecoveryContext.HasTriedCurlFallback)); var refreshedRetryStartedAt = DateTimeOffset.UtcNow; session.MarkRunning(refreshedRetryStartedAt); @@ -1362,6 +1373,11 @@ public sealed partial class FfmpegService internal static TimeSpan GetRuntimeRecoveryDelay(int attempt) => RuntimeRecoveryBackoff[Math.Clamp(attempt - 1, 0, RuntimeRecoveryBackoff.Length - 1)]; + internal static bool ShouldPersistFfmpegLine( + bool hasOpenedFirstSegment, + bool isStreamHandshakeFailure) => + hasOpenedFirstSegment || !isStreamHandshakeFailure; + private async Task TryRetryWithAlternateProtocolAsync( SessionProcessRuntime runtime, RecordSession session, diff --git a/src/LiveRecorder.Infrastructure/Services/FfmpegService.cs b/src/LiveRecorder.Infrastructure/Services/FfmpegService.cs index f8208f1..c1729fb 100644 --- a/src/LiveRecorder.Infrastructure/Services/FfmpegService.cs +++ b/src/LiveRecorder.Infrastructure/Services/FfmpegService.cs @@ -22,6 +22,9 @@ namespace LiveRecorder.Infrastructure.Services; public sealed partial class FfmpegService : IFfmpegService { private const string ArtifactRepairMarker = "[artifact-repair]"; + private const string StalePendingStartupFailure = + "Recording startup was interrupted before stream initialization."; + private static readonly TimeSpan StalePendingThreshold = TimeSpan.FromMinutes(10); private static readonly Regex SegmentOpeningRegex = new( """Opening '([^']+)' for writing""", RegexOptions.Compiled | RegexOptions.CultureInvariant | RegexOptions.IgnoreCase); @@ -644,6 +647,120 @@ public sealed partial class FfmpegService : IFfmpegService } } + public async Task RecoverStalePendingSessionsAsync(CancellationToken cancellationToken = default) + { + if (!await _orphanRecoveryGate.WaitAsync(0, cancellationToken)) + { + return 0; + } + + try + { + using var scope = _serviceScopeFactory.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + var cutoff = DateTimeOffset.UtcNow - StalePendingThreshold; + var candidateIds = await dbContext.RecordSessions + .AsNoTracking() + .Where(session => + session.Status == RecordSessionStatus.Pending && + session.UpdatedAt <= cutoff) + .OrderBy(session => session.UpdatedAt) + .Select(session => session.Id) + .Take(50) + .ToListAsync(cancellationToken); + + var recovered = 0; + foreach (var candidateId in candidateIds) + { + if (IsRunning(candidateId)) + { + continue; + } + + using var candidateScope = _serviceScopeFactory.CreateScope(); + var candidateDbContext = candidateScope.ServiceProvider.GetRequiredService(); + var session = await candidateDbContext.RecordSessions + .Include(item => item.RecordTasks) + .ThenInclude(item => item.Result) + .FirstOrDefaultAsync(item => item.Id == candidateId, cancellationToken); + if (session is null || + session.Status != RecordSessionStatus.Pending || + session.UpdatedAt > cutoff || + IsRunning(candidateId)) + { + continue; + } + + if (HasRecoverablePendingMedia(session)) + { + if (await ReconcileInactiveSessionAsync(candidateId, allowTerminalSession: false, cancellationToken)) + { + recovered++; + } + + continue; + } + + var failedAt = DateTimeOffset.UtcNow; + foreach (var task in session.RecordTasks.Where(task => task.Status == RecordTaskStatus.Pending)) + { + task.MarkFailed(StalePendingStartupFailure, failedAt); + } + + session.MarkFailed(StalePendingStartupFailure, failedAt); + await candidateDbContext.SaveChangesAsync(cancellationToken); + + var logService = candidateScope.ServiceProvider.GetRequiredService(); + await logService.WriteAsync( + SystemLogLevel.Warning, + "RecordSession", + "A stale pending recording session was marked failed.", + StalePendingStartupFailure, + session.LiveRoomId, + session.Id, + session.RecordTasks.OrderBy(task => task.SegmentIndex).FirstOrDefault()?.Id, + cancellationToken); + recovered++; + } + + return recovered; + } + finally + { + _orphanRecoveryGate.Release(); + } + } + + private static bool HasRecoverablePendingMedia(RecordSession session) + { + if (DiscoverRecoverableSegments(session.OutputPathPattern, session.OutputFormat, session.SaveMode).Count > 0) + { + return true; + } + + foreach (var task in session.RecordTasks) + { + var candidates = new[] + { + task.Result?.FilePath, + task.OutputFilePath, + string.IsNullOrWhiteSpace(task.OutputFilePath) + ? null + : GetRecorderOutputPath(task.OutputFilePath, session.OutputFormat, session.SaveMode) + }; + + if (candidates.Any(path => + !string.IsNullOrWhiteSpace(path) && + File.Exists(path) && + new FileInfo(path).Length > 0)) + { + return true; + } + } + + return false; + } + public async Task TryRecoverOrphanedTerminalTaskAsync( Guid recordTaskId, CancellationToken cancellationToken = default) diff --git a/src/LiveRecorder.Infrastructure/Services/LiveRoomPollingBackgroundService.cs b/src/LiveRecorder.Infrastructure/Services/LiveRoomPollingBackgroundService.cs index 9ed2794..3a7a8b2 100644 --- a/src/LiveRecorder.Infrastructure/Services/LiveRoomPollingBackgroundService.cs +++ b/src/LiveRecorder.Infrastructure/Services/LiveRoomPollingBackgroundService.cs @@ -76,6 +76,13 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService, ILiveR var settings = await settingsService.GetAsync(stoppingToken); var ffmpegService = scope.ServiceProvider.GetRequiredService(); await ffmpegService.ResumeRecoveringSessionsAsync(stoppingToken); + var recoveredPendingSessions = await ffmpegService.RecoverStalePendingSessionsAsync(stoppingToken); + if (recoveredPendingSessions > 0) + { + _logger.LogWarning( + "Recovered {SessionCount} stale pending recording sessions.", + recoveredPendingSessions); + } var recoveredOrphanedSessions = await ffmpegService.RecoverOrphanedTerminalSessionTasksAsync(stoppingToken); if (recoveredOrphanedSessions > 0) { diff --git a/src/LiveRecorder.Infrastructure/Services/RecoveryService.cs b/src/LiveRecorder.Infrastructure/Services/RecoveryService.cs index a30d745..5bf4052 100644 --- a/src/LiveRecorder.Infrastructure/Services/RecoveryService.cs +++ b/src/LiveRecorder.Infrastructure/Services/RecoveryService.cs @@ -57,6 +57,7 @@ public sealed class RecoveryService IsAvailable = storage.IsAvailable, HasEnoughSpace = storage.HasEnoughSpace, CheckedPath = storage.CheckedPath, + VolumeRoot = storage.VolumeRoot, TotalBytes = storage.TotalBytes, UsedBytes = storage.UsedBytes, AvailableBytes = storage.AvailableBytes, diff --git a/src/LiveRecorder.Infrastructure/Services/StorageGuardService.cs b/src/LiveRecorder.Infrastructure/Services/StorageGuardService.cs index 1e92d00..fc154f1 100644 --- a/src/LiveRecorder.Infrastructure/Services/StorageGuardService.cs +++ b/src/LiveRecorder.Infrastructure/Services/StorageGuardService.cs @@ -8,10 +8,14 @@ public sealed class StorageGuardService : IStorageGuardService { private const long Megabyte = 1024L * 1024L; private readonly ILogger _logger; + private readonly IStorageCapacityProvider _capacityProvider; - public StorageGuardService(ILogger logger) + public StorageGuardService( + ILogger logger, + IStorageCapacityProvider capacityProvider) { _logger = logger; + _capacityProvider = capacityProvider; } public StorageGuardResult CheckCanStartOrResume(SystemSettingsDto settings, long additionalRequiredBytes = 0) => @@ -44,9 +48,9 @@ public sealed class StorageGuardService : IStorageGuardService try { checkedPath = ResolveOutputRoot(settings.OutputRoot ?? string.Empty); - var drive = new DriveInfo(Path.GetPathRoot(checkedPath) ?? checkedPath); - var availableBytes = drive.AvailableFreeSpace; - var totalBytes = drive.TotalSize; + var capacity = _capacityProvider.GetCapacity(checkedPath); + var availableBytes = capacity.AvailableBytes; + var totalBytes = capacity.TotalBytes; var usedBytes = Math.Max(0, totalBytes - availableBytes); var usagePercent = totalBytes > 0 ? (double)usedBytes / totalBytes * 100.0 : 0; var freePercent = totalBytes > 0 ? (double)availableBytes / totalBytes * 100.0 : 0; @@ -81,6 +85,7 @@ public sealed class StorageGuardService : IStorageGuardService return new StorageGuardResult(isEnabled, hasEnoughSpace, checkedPath, availableBytes, requiredBytes, message) { IsAvailable = totalBytes > 0, + VolumeRoot = capacity.VolumeRoot, Tier = tier, TotalBytes = totalBytes, UsedBytes = usedBytes, @@ -108,9 +113,10 @@ public sealed class StorageGuardService : IStorageGuardService private static string ResolveOutputRoot(string outputRoot) { var root = string.IsNullOrWhiteSpace(outputRoot) ? "records" : outputRoot.Trim(); - return Path.IsPathRooted(root) + var resolved = Path.IsPathRooted(root) ? root : Path.GetFullPath(root, AppContext.BaseDirectory); + return Path.GetFullPath(resolved); } private static string FormatBytes(long bytes) diff --git a/src/LiveRecorder.WebApi/Controllers/RecordTasksController.cs b/src/LiveRecorder.WebApi/Controllers/RecordTasksController.cs index 2cdce5e..338bfee 100644 --- a/src/LiveRecorder.WebApi/Controllers/RecordTasksController.cs +++ b/src/LiveRecorder.WebApi/Controllers/RecordTasksController.cs @@ -57,7 +57,8 @@ public sealed class RecordTasksController : ControllerBase 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 pendingUploadMetrics = await _recordResultRepository.GetPendingUploadMetricsAsync(cancellationToken); + var notUploadedCount = pendingUploadMetrics.Count; var failedArtifactCount = await _recordResultRepository.CountFailedArtifactAsync(cancellationToken); var succeededCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Succeeded, null, cancellationToken); var failedCount = await _recordResultRepository.CountUploadStatusAsync(RecordArtifactUploadStatus.Failed, null, cancellationToken); diff --git a/src/LiveRecorder.WebApi/Program.cs b/src/LiveRecorder.WebApi/Program.cs index 0836354..585caad 100644 --- a/src/LiveRecorder.WebApi/Program.cs +++ b/src/LiveRecorder.WebApi/Program.cs @@ -243,6 +243,7 @@ builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddScoped(); builder.Services.AddSingleton(); diff --git a/tests/LiveRecorder.Tests/FfmpegFailureClassificationTests.cs b/tests/LiveRecorder.Tests/FfmpegFailureClassificationTests.cs index 404930d..9edca11 100644 --- a/tests/LiveRecorder.Tests/FfmpegFailureClassificationTests.cs +++ b/tests/LiveRecorder.Tests/FfmpegFailureClassificationTests.cs @@ -272,7 +272,7 @@ public sealed class FfmpegFailureClassificationTests } [Fact] - public void RecoveryContext_LeavesCurlAfterExitWithoutResettingOneShotBudget() + public void RefreshedFlvRecovery_ReEnablesCurlAfterOverlongHeaders() { var curlContext = FfmpegService.InitialRecoveryContext with { @@ -280,18 +280,65 @@ public sealed class FfmpegFailureClassificationTests HasTriedCurlFallback = true }; - var nextContext = FfmpegService.AdvanceRecoveryContext( - curlContext, - curlContext.InputOptionProfile, + var nextContext = FfmpegService.PrepareRefreshedStreamRecoveryContext( + FfmpegService.AdvanceRecoveryContext( + curlContext, + curlContext.InputOptionProfile, + "flv", + "flv", + refreshedStream: true), "flv", - "flv", - refreshedStream: true); + hasOverlongHeadersFailure: true); - Assert.False(nextContext.UseCurlFallback); + Assert.True(nextContext.UseCurlFallback); Assert.True(nextContext.HasTriedCurlFallback); Assert.True(nextContext.HasRetriedWithRefreshedStream); Assert.False(FfmpegService.ShouldFallbackToCurl("flv", true, nextContext.HasTriedCurlFallback)); } + + [Fact] + public void RefreshedNonFlvRecovery_DisablesCurl() + { + var nextContext = FfmpegService.PrepareRefreshedStreamRecoveryContext( + FfmpegService.InitialRecoveryContext with + { + UseCurlFallback = true, + HasTriedCurlFallback = true + }, + "hls", + hasOverlongHeadersFailure: true); + + Assert.False(nextContext.UseCurlFallback); + Assert.True(nextContext.HasTriedCurlFallback); + } + + [Fact] + public void RefreshedFlvRecovery_PreservesCurlRequirementAcrossLaterProcesses() + { + var priorContext = FfmpegService.InitialRecoveryContext with + { + UseCurlFallback = false, + HasTriedCurlFallback = true + }; + + var nextContext = FfmpegService.PrepareRefreshedStreamRecoveryContext( + priorContext, + "flv", + hasOverlongHeadersFailure: priorContext.HasTriedCurlFallback); + + Assert.True(nextContext.UseCurlFallback); + Assert.True(nextContext.HasTriedCurlFallback); + } + + [Theory] + [InlineData(false, true, false)] + [InlineData(false, false, true)] + [InlineData(true, true, true)] + public void FfmpegLinePersistence_SuppressesRecoverableStartupHandshakeNoise( + bool hasOpenedFirstSegment, + bool isStreamHandshakeFailure, + bool expected) => + Assert.Equal(expected, FfmpegService.ShouldPersistFfmpegLine(hasOpenedFirstSegment, isStreamHandshakeFailure)); [Fact] public void RecoveryContext_PreservesBudgetAcrossHlsFallbackAndReachesTimestampTranscode() { diff --git a/tests/LiveRecorder.Tests/OpenListUploadTests.cs b/tests/LiveRecorder.Tests/OpenListUploadTests.cs index 1377a69..81161a4 100644 --- a/tests/LiveRecorder.Tests/OpenListUploadTests.cs +++ b/tests/LiveRecorder.Tests/OpenListUploadTests.cs @@ -615,6 +615,62 @@ public sealed class OpenListUploadTests Assert.Equal(1, await repository.CountFailedArtifactAsync()); } + [Fact] + public async Task PendingUploadMetrics_IncludeValidVisibleArtifactWithActualFileSize() + { + await using var fixture = await QueueFixture.CreateAsync(); + var result = await fixture.Context.RecordResults.AsNoTracking().SingleAsync(); + var repository = new RecordResultRepository(fixture.Context); + + var metrics = await repository.GetPendingUploadMetricsAsync(); + + Assert.Equal(1, metrics.Count); + Assert.Equal(new FileInfo(result.FilePath).Length, metrics.TotalBytes); + } + + [Fact] + public async Task PendingUploadMetrics_ExcludeHiddenMergedSource() + { + await using var fixture = await QueueFixture.CreateAsync(); + var task = await fixture.Context.RecordTasks.SingleAsync(); + task.MarkMergedSource(Guid.NewGuid(), DateTimeOffset.UtcNow); + await fixture.Context.SaveChangesAsync(); + var repository = new RecordResultRepository(fixture.Context); + + var metrics = await repository.GetPendingUploadMetricsAsync(); + + Assert.Equal(0, metrics.Count); + Assert.Equal(0, metrics.TotalBytes); + } + + [Fact] + public async Task PendingUploadMetrics_ExcludeMissingLocalFile() + { + await using var fixture = await QueueFixture.CreateAsync(); + var result = await fixture.Context.RecordResults.AsNoTracking().SingleAsync(); + File.Delete(result.FilePath); + var repository = new RecordResultRepository(fixture.Context); + + var metrics = await repository.GetPendingUploadMetricsAsync(); + + Assert.Equal(0, metrics.Count); + Assert.Equal(0, metrics.TotalBytes); + } + + [Fact] + public async Task PendingUploadMetrics_ExcludeEmptyLocalFile() + { + await using var fixture = await QueueFixture.CreateAsync(); + var result = await fixture.Context.RecordResults.AsNoTracking().SingleAsync(); + File.WriteAllBytes(result.FilePath, []); + var repository = new RecordResultRepository(fixture.Context); + + var metrics = await repository.GetPendingUploadMetricsAsync(); + + Assert.Equal(0, metrics.Count); + Assert.Equal(0, metrics.TotalBytes); + } + private static OpenListClient CreateClient(HttpMessageHandler handler) => new(new StubHttpClientFactory(handler)); diff --git a/tests/LiveRecorder.Tests/RecordingContinuityTests.cs b/tests/LiveRecorder.Tests/RecordingContinuityTests.cs index 04fd48a..51a7110 100644 --- a/tests/LiveRecorder.Tests/RecordingContinuityTests.cs +++ b/tests/LiveRecorder.Tests/RecordingContinuityTests.cs @@ -20,8 +20,11 @@ public sealed class RecordingContinuityTests [InlineData(3, 30)] [InlineData(4, 60)] [InlineData(5, 120)] - [InlineData(50, 120)] - public void RuntimeRecoveryBackoff_IsBoundedAtTwoMinutes(int attempt, int seconds) => + [InlineData(6, 300)] + [InlineData(7, 600)] + [InlineData(8, 900)] + [InlineData(50, 900)] + public void RuntimeRecoveryBackoff_IsBoundedAtFifteenMinutes(int attempt, int seconds) => Assert.Equal(TimeSpan.FromSeconds(seconds), FfmpegService.GetRuntimeRecoveryDelay(attempt)); [Theory] diff --git a/tests/LiveRecorder.Tests/StalePendingRecoveryTests.cs b/tests/LiveRecorder.Tests/StalePendingRecoveryTests.cs new file mode 100644 index 0000000..00c7827 --- /dev/null +++ b/tests/LiveRecorder.Tests/StalePendingRecoveryTests.cs @@ -0,0 +1,120 @@ +using LiveRecorder.Application.Abstractions.Logging; +using LiveRecorder.Application.Models.Logs; +using LiveRecorder.Domain.Entities; +using LiveRecorder.Domain.Enums; +using LiveRecorder.Infrastructure.Persistence; +using LiveRecorder.Infrastructure.Services; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Storage; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; + +namespace LiveRecorder.Tests; + +public sealed class StalePendingRecoveryTests +{ + [Fact] + public async Task StalePendingSessionWithoutMedia_IsFailedAndRetainedForAudit() + { + var services = new ServiceCollection(); + var databaseRoot = new InMemoryDatabaseRoot(); + var databaseName = $"stale-pending-{Guid.NewGuid():N}"; + services.AddDbContext(options => + options.UseInMemoryDatabase(databaseName, databaseRoot)); + var logService = new CapturingSystemLogService(); + services.AddSingleton(logService); + await using var provider = services.BuildServiceProvider(); + + var createdAt = DateTimeOffset.UtcNow.AddMinutes(-11); + Guid sessionId; + Guid taskId; + await using (var scope = provider.CreateAsyncScope()) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + var liveRoom = new LiveRoom( + LivePlatformType.Douyin, + "https://live.example/room", + "room-1", + "https://live.example/room", + createdAt); + var session = new RecordSession( + liveRoom.Id, + "origin", + RecordOutputFormat.Mp4, + RecordSaveMode.Segmented, + createdAt); + var task = new RecordTask( + liveRoom.Id, + session.Id, + 1, + "origin", + RecordOutputFormat.Mp4, + createdAt); + sessionId = session.Id; + taskId = task.Id; + dbContext.AddRange(liveRoom, session, task); + await dbContext.SaveChangesAsync(); + } + + var service = new FfmpegService( + provider.GetRequiredService(), + null!, + null!, + NullLogger.Instance); + + await using (var preconditionScope = provider.CreateAsyncScope()) + { + var preconditionContext = preconditionScope.ServiceProvider.GetRequiredService(); + var persisted = await preconditionContext.RecordSessions.SingleAsync(); + Assert.Equal(RecordSessionStatus.Pending, persisted.Status); + Assert.True( + persisted.UpdatedAt <= DateTimeOffset.UtcNow.AddMinutes(-10), + $"updatedAt={persisted.UpdatedAt:O}"); + } + + var recovered = await service.RecoverStalePendingSessionsAsync(); + + Assert.Equal(1, recovered); + await using var verificationScope = provider.CreateAsyncScope(); + var verificationContext = verificationScope.ServiceProvider.GetRequiredService(); + var persistedSession = await verificationContext.RecordSessions.SingleAsync(item => item.Id == sessionId); + var persistedTask = await verificationContext.RecordTasks + .Include(item => item.Result) + .SingleAsync(item => item.Id == taskId); + Assert.Equal(RecordSessionStatus.Failed, persistedSession.Status); + Assert.Equal(RecordTaskStatus.Failed, persistedTask.Status); + Assert.Contains("interrupted before stream initialization", persistedSession.ErrorMessage); + Assert.Null(persistedTask.Result); + Assert.Single(logService.Entries); + Assert.Equal(SystemLogLevel.Warning, logService.Entries[0].Level); + } + + private sealed class CapturingSystemLogService : ISystemLogService + { + public List<(SystemLogLevel Level, string Message)> Entries { get; } = []; + + public Task WriteAsync( + SystemLogLevel level, + string category, + string message, + string? detail = null, + Guid? liveRoomId = null, + Guid? recordSessionId = null, + Guid? recordTaskId = null, + CancellationToken cancellationToken = default) + { + Entries.Add((level, message)); + return Task.CompletedTask; + } + + public Task> ListAsync( + Guid? liveRoomId = null, + Guid? recordSessionId = null, + Guid? recordTaskId = null, + SystemLogLevel? level = null, + string? content = null, + int take = 200, + CancellationToken cancellationToken = default) => + Task.FromResult>([]); + } +} diff --git a/tests/LiveRecorder.Tests/StorageGuardServiceTests.cs b/tests/LiveRecorder.Tests/StorageGuardServiceTests.cs index 16aa6ea..8c9b4f7 100644 --- a/tests/LiveRecorder.Tests/StorageGuardServiceTests.cs +++ b/tests/LiveRecorder.Tests/StorageGuardServiceTests.cs @@ -7,7 +7,26 @@ namespace LiveRecorder.Tests; public sealed class StorageGuardServiceTests { - private readonly StorageGuardService _service = new(NullLogger.Instance); + private readonly StorageGuardService _service = new( + NullLogger.Instance, + new DriveInfoStorageCapacityProvider()); + + [Fact] + public void CapacityProvider_SelectsLongestContainingMountRoot() + { + var selected = DriveInfoStorageCapacityProvider.SelectBestVolume( + "/vol00/live-recorder/Record", + [ + new StorageVolumeCapacity("/", 100, 10), + new StorageVolumeCapacity("/vol00", 200, 20), + new StorageVolumeCapacity("/vol00/live-recorder", 300, 30), + new StorageVolumeCapacity("/vol0", 400, 40) + ]); + + Assert.NotNull(selected); + Assert.Equal("/vol00/live-recorder", selected.VolumeRoot); + Assert.Equal(300, selected.TotalBytes); + } [Fact] public void DisabledProtection_StillReportsActualCapacityWithoutBlockingRecording()