feat: improve recording automation and task workflows
This commit is contained in:
@@ -1,12 +1,16 @@
|
||||
using System.Collections.Concurrent;
|
||||
using LiveRecorder.Application.Abstractions.Logging;
|
||||
using LiveRecorder.Application.Abstractions.Notifications;
|
||||
using LiveRecorder.Application.Abstractions.Platforms;
|
||||
using LiveRecorder.Application.Abstractions.Recording;
|
||||
using LiveRecorder.Application.Abstractions.Settings;
|
||||
using LiveRecorder.Application.Abstractions.Storage;
|
||||
using LiveRecorder.Application.Models.Settings;
|
||||
using LiveRecorder.Application.Models.RecordTasks;
|
||||
using LiveRecorder.Application.Services;
|
||||
using LiveRecorder.Domain.Enums;
|
||||
using LiveRecorder.Infrastructure.Persistence;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
@@ -16,8 +20,13 @@ namespace LiveRecorder.Infrastructure.Services;
|
||||
|
||||
public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
{
|
||||
private static readonly TimeSpan OfflineGracefulStopTimeout = TimeSpan.FromSeconds(20);
|
||||
private static readonly TimeSpan OfflineForcedStopTimeout = TimeSpan.FromSeconds(8);
|
||||
private static readonly TimeSpan ExceptionEmailCooldown = TimeSpan.FromHours(6);
|
||||
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
private readonly ILogger<LiveRoomPollingBackgroundService> _logger;
|
||||
private readonly ConcurrentDictionary<string, DateTimeOffset> _exceptionEmailSentAt = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
public LiveRoomPollingBackgroundService(
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
@@ -38,9 +47,15 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
||||
var settings = await settingsService.GetAsync(stoppingToken);
|
||||
var emailNotificationService = scope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
||||
|
||||
delaySeconds = settings.PollingIntervalSeconds;
|
||||
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
|
||||
var storageGuardService = scope.ServiceProvider.GetRequiredService<IStorageGuardService>();
|
||||
if (storageGuardService.CheckCanStartOrResume(settings).HasEnoughSpace)
|
||||
{
|
||||
await ffmpegService.ResumePausedFinalizationsAsync(stoppingToken);
|
||||
}
|
||||
|
||||
if (!settings.EnableBackgroundPolling)
|
||||
{
|
||||
await DelayAsync(delaySeconds, stoppingToken);
|
||||
@@ -48,104 +63,23 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
}
|
||||
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
||||
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
|
||||
var recordService = scope.ServiceProvider.GetRequiredService<RecordService>();
|
||||
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
|
||||
var liveRooms = (await dbContext.LiveRooms.ToListAsync(stoppingToken))
|
||||
.Where(static item => item.IsEnabled)
|
||||
var liveRoomIds = (await dbContext.LiveRooms
|
||||
.AsNoTracking()
|
||||
.Where(static item => item.IsEnabled)
|
||||
.Select(static item => new { item.Id, item.UpdatedAt })
|
||||
.ToListAsync(stoppingToken))
|
||||
.OrderBy(static item => item.UpdatedAt)
|
||||
.Select(static item => item.Id)
|
||||
.ToList();
|
||||
|
||||
foreach (var liveRoom in liveRooms)
|
||||
foreach (var liveRoomId in liveRoomIds)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
|
||||
var liveStatus = await adapter.GetLiveStatusAsync(liveRoom.RoomId, stoppingToken);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
await liveRoomStatusService.ApplySnapshotAsync(liveRoom, liveStatus, now, stoppingToken);
|
||||
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
|
||||
if (!liveStatus.IsLive)
|
||||
{
|
||||
await CompleteActiveSessionsForOfflineRoomAsync(
|
||||
dbContext,
|
||||
ffmpegService,
|
||||
logService,
|
||||
liveRoom.Id,
|
||||
stoppingToken);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!settings.AutoStartRecordingOnLive)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var hasRunningSession = await dbContext.RecordSessions.AnyAsync(
|
||||
item => item.LiveRoomId == liveRoom.Id &&
|
||||
(item.Status == RecordSessionStatus.Starting ||
|
||||
item.Status == RecordSessionStatus.Running ||
|
||||
item.Status == RecordSessionStatus.Stopping),
|
||||
stoppingToken);
|
||||
|
||||
if (hasRunningSession)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Auto-start recording for live room {RoomId}", liveRoom.RoomId);
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"Scheduler",
|
||||
"Live detected by background poller. Auto-starting recording task.",
|
||||
liveRoomId: liveRoom.Id,
|
||||
cancellationToken: stoppingToken);
|
||||
|
||||
await recordService.StartAsync(
|
||||
new StartRecordTaskRequest
|
||||
{
|
||||
LiveRoomId = liveRoom.Id,
|
||||
PreferredQuality = settings.DefaultQuality,
|
||||
OutputFormat = settings.DefaultOutputFormat
|
||||
},
|
||||
stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Polling room {RoomId} failed", liveRoom.RoomId);
|
||||
|
||||
var isTransient = IsTransientPollingException(ex, stoppingToken);
|
||||
await logService.WriteAsync(
|
||||
isTransient ? SystemLogLevel.Warning : SystemLogLevel.Error,
|
||||
"Scheduler",
|
||||
isTransient
|
||||
? "Transient background polling failure. The room will be retried on the next cycle."
|
||||
: "Background polling failed for a live room.",
|
||||
ex.ToString(),
|
||||
liveRoomId: liveRoom.Id,
|
||||
cancellationToken: stoppingToken);
|
||||
|
||||
if (!isTransient)
|
||||
{
|
||||
await emailNotificationService.SendExceptionAsync(
|
||||
"Scheduler",
|
||||
"Background polling failed for a live room.",
|
||||
ex.ToString(),
|
||||
liveRoom,
|
||||
cancellationToken: stoppingToken);
|
||||
}
|
||||
}
|
||||
await PollLiveRoomAsync(liveRoomId, settings, stoppingToken);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
@@ -160,11 +94,14 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
{
|
||||
using var notificationScope = _serviceScopeFactory.CreateScope();
|
||||
var emailNotificationService = notificationScope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
||||
await emailNotificationService.SendExceptionAsync(
|
||||
"Scheduler",
|
||||
"Background live room polling failed.",
|
||||
ex.ToString(),
|
||||
cancellationToken: stoppingToken);
|
||||
if (ShouldSendExceptionEmail(BuildExceptionEmailKey("background-loop", ex)))
|
||||
{
|
||||
await emailNotificationService.SendExceptionAsync(
|
||||
"Scheduler",
|
||||
"Background live room polling failed.",
|
||||
ex.ToString(),
|
||||
cancellationToken: stoppingToken);
|
||||
}
|
||||
}
|
||||
catch (Exception notificationEx)
|
||||
{
|
||||
@@ -179,6 +116,128 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
private static Task DelayAsync(int delaySeconds, CancellationToken cancellationToken) =>
|
||||
Task.Delay(TimeSpan.FromSeconds(Math.Clamp(delaySeconds, 10, 3600)), cancellationToken);
|
||||
|
||||
private async Task PollLiveRoomAsync(Guid liveRoomId, SystemSettingsDto settings, CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
||||
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
|
||||
var recordService = scope.ServiceProvider.GetRequiredService<RecordService>();
|
||||
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
var emailNotificationService = scope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
||||
var storageGuardService = scope.ServiceProvider.GetRequiredService<IStorageGuardService>();
|
||||
|
||||
var liveRoom = await dbContext.LiveRooms.FirstOrDefaultAsync(item => item.Id == liveRoomId, cancellationToken);
|
||||
if (liveRoom is null || !liveRoom.IsEnabled)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
|
||||
var liveStatus = await adapter.GetLiveStatusAsync(liveRoom.RoomId, cancellationToken);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
await liveRoomStatusService.ApplySnapshotAsync(liveRoom, liveStatus, now, cancellationToken);
|
||||
await SaveChangesWithRetryAsync(dbContext, cancellationToken);
|
||||
|
||||
if (!liveStatus.IsLive)
|
||||
{
|
||||
await CompleteActiveSessionsForOfflineRoomAsync(
|
||||
dbContext,
|
||||
ffmpegService,
|
||||
logService,
|
||||
liveRoom.Id,
|
||||
cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var pauseCheck = storageGuardService.CheckShouldPause(settings);
|
||||
if (!pauseCheck.HasEnoughSpace)
|
||||
{
|
||||
await PauseActiveSessionsForLowStorageAsync(
|
||||
dbContext,
|
||||
ffmpegService,
|
||||
logService,
|
||||
liveRoom.Id,
|
||||
pauseCheck.Message,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
if (!settings.AutoStartRecordingOnLive)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var startCheck = storageGuardService.CheckCanStartOrResume(settings);
|
||||
if (!startCheck.HasEnoughSpace)
|
||||
{
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"Storage",
|
||||
"Auto-start recording skipped because storage is below resume threshold.",
|
||||
startCheck.Message,
|
||||
liveRoomId: liveRoom.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var hasRunningSession = await dbContext.RecordSessions.AnyAsync(
|
||||
item => item.LiveRoomId == liveRoom.Id &&
|
||||
(item.Status == RecordSessionStatus.Starting ||
|
||||
item.Status == RecordSessionStatus.Running ||
|
||||
item.Status == RecordSessionStatus.Stopping),
|
||||
cancellationToken);
|
||||
|
||||
if (hasRunningSession)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Auto-start recording for live room {RoomId}", liveRoom.RoomId);
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"Scheduler",
|
||||
"Live detected by background poller. Auto-starting recording task.",
|
||||
liveRoomId: liveRoom.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await recordService.StartAsync(
|
||||
new StartRecordTaskRequest
|
||||
{
|
||||
LiveRoomId = liveRoom.Id
|
||||
},
|
||||
cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Polling room {RoomId} failed", liveRoom.RoomId);
|
||||
|
||||
var isTransient = IsTransientPollingException(ex, cancellationToken);
|
||||
await logService.WriteAsync(
|
||||
isTransient ? SystemLogLevel.Warning : SystemLogLevel.Error,
|
||||
"Scheduler",
|
||||
isTransient
|
||||
? "Transient background polling failure. The room will be retried on the next cycle."
|
||||
: "Background polling failed for a live room.",
|
||||
ex.ToString(),
|
||||
liveRoomId: liveRoom.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
if (!isTransient && ShouldSendExceptionEmail(BuildExceptionEmailKey("live-room-poll", ex, liveRoom.Id)))
|
||||
{
|
||||
await emailNotificationService.SendExceptionAsync(
|
||||
"Scheduler",
|
||||
"Background polling failed for a live room.",
|
||||
ex.ToString(),
|
||||
liveRoom,
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task CompleteActiveSessionsForOfflineRoomAsync(
|
||||
LiveRecorderDbContext dbContext,
|
||||
IFfmpegService ffmpegService,
|
||||
@@ -201,7 +260,40 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
{
|
||||
if (ffmpegService.IsRunning(activeSession.Id))
|
||||
{
|
||||
await ffmpegService.CompleteAsync(activeSession.Id, cancellationToken);
|
||||
var stopped = await ffmpegService.StopAndWaitAsync(
|
||||
activeSession.Id,
|
||||
markAsCompletedOnExit: true,
|
||||
OfflineGracefulStopTimeout,
|
||||
cancellationToken);
|
||||
|
||||
if (!stopped)
|
||||
{
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"Scheduler",
|
||||
"Live room is offline, but the recorder did not stop gracefully in time. Force killing the ffmpeg process.",
|
||||
liveRoomId: liveRoomId,
|
||||
recordSessionId: activeSession.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
var killed = await ffmpegService.KillAndWaitAsync(
|
||||
activeSession.Id,
|
||||
OfflineForcedStopTimeout,
|
||||
cancellationToken);
|
||||
if (!killed)
|
||||
{
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Error,
|
||||
"Scheduler",
|
||||
"Live room is offline, but the active recording session is still shutting down.",
|
||||
liveRoomId: liveRoomId,
|
||||
recordSessionId: activeSession.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
await ffmpegService.TryReconcileInactiveSessionAsync(activeSession.Id, cancellationToken);
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"Scheduler",
|
||||
@@ -225,6 +317,57 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task PauseActiveSessionsForLowStorageAsync(
|
||||
LiveRecorderDbContext dbContext,
|
||||
IFfmpegService ffmpegService,
|
||||
ISystemLogService logService,
|
||||
Guid liveRoomId,
|
||||
string detail,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var activeSessions = await dbContext.RecordSessions
|
||||
.Where(item => item.LiveRoomId == liveRoomId &&
|
||||
(item.Status == RecordSessionStatus.Starting ||
|
||||
item.Status == RecordSessionStatus.Running ||
|
||||
item.Status == RecordSessionStatus.Stopping))
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
foreach (var activeSession in activeSessions.OrderBy(static item => item.CreatedAt))
|
||||
{
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"Storage",
|
||||
"Storage is below threshold. Pausing active recording session.",
|
||||
detail,
|
||||
liveRoomId: liveRoomId,
|
||||
recordSessionId: activeSession.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
if (ffmpegService.IsRunning(activeSession.Id))
|
||||
{
|
||||
var stopped = await ffmpegService.StopAndWaitAsync(
|
||||
activeSession.Id,
|
||||
markAsCompletedOnExit: false,
|
||||
OfflineGracefulStopTimeout,
|
||||
cancellationToken);
|
||||
if (!stopped)
|
||||
{
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"Storage",
|
||||
"Recorder did not stop gracefully after low storage pause. Force killing the ffmpeg process.",
|
||||
detail,
|
||||
liveRoomId,
|
||||
activeSession.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
await ffmpegService.KillAndWaitAsync(activeSession.Id, OfflineForcedStopTimeout, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
await ffmpegService.TryReconcileInactiveSessionAsync(activeSession.Id, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsTransientPollingException(Exception exception, CancellationToken cancellationToken)
|
||||
{
|
||||
if (exception is OperationCanceledException && cancellationToken.IsCancellationRequested)
|
||||
@@ -237,7 +380,99 @@ public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
||||
return true;
|
||||
}
|
||||
|
||||
if (exception is DbUpdateException dbUpdateException &&
|
||||
IsSqliteLockException(dbUpdateException))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (exception is SqliteException sqliteException &&
|
||||
IsSqliteLockException(sqliteException))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return exception.InnerException is not null &&
|
||||
IsTransientPollingException(exception.InnerException, cancellationToken);
|
||||
}
|
||||
|
||||
private bool ShouldSendExceptionEmail(string key)
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
while (true)
|
||||
{
|
||||
if (_exceptionEmailSentAt.TryGetValue(key, out var lastSentAt))
|
||||
{
|
||||
if (now - lastSentAt < ExceptionEmailCooldown)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Suppressed repeated scheduler exception email. Key={Key}; CooldownMinutes={CooldownMinutes}",
|
||||
key,
|
||||
ExceptionEmailCooldown.TotalMinutes);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (_exceptionEmailSentAt.TryUpdate(key, now, lastSentAt))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
if (_exceptionEmailSentAt.TryAdd(key, now))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static string BuildExceptionEmailKey(string scope, Exception exception, Guid? liveRoomId = null)
|
||||
{
|
||||
if (IsSqliteStorageFullException(exception))
|
||||
{
|
||||
return $"{scope}:sqlite-storage-full";
|
||||
}
|
||||
|
||||
var root = exception.GetBaseException();
|
||||
var message = root.Message.Length > 160 ? root.Message[..160] : root.Message;
|
||||
return $"{scope}:{liveRoomId?.ToString() ?? "global"}:{root.GetType().FullName}:{message}";
|
||||
}
|
||||
|
||||
private static bool IsSqliteStorageFullException(Exception exception)
|
||||
{
|
||||
if (exception is SqliteException sqliteException &&
|
||||
(sqliteException.SqliteErrorCode == 13 ||
|
||||
sqliteException.Message.Contains("database or disk is full", StringComparison.OrdinalIgnoreCase)))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return exception.InnerException is not null && IsSqliteStorageFullException(exception.InnerException);
|
||||
}
|
||||
|
||||
private static async Task SaveChangesWithRetryAsync(LiveRecorderDbContext dbContext, CancellationToken cancellationToken)
|
||||
{
|
||||
for (var attempt = 1; attempt <= 5; attempt++)
|
||||
{
|
||||
try
|
||||
{
|
||||
await dbContext.SaveChangesAsync(cancellationToken);
|
||||
return;
|
||||
}
|
||||
catch (DbUpdateException ex) when (attempt < 5 && IsSqliteLockException(ex))
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(300 * Math.Pow(2, attempt - 1)), cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsSqliteLockException(DbUpdateException exception) =>
|
||||
exception.InnerException is SqliteException sqliteException &&
|
||||
IsSqliteLockException(sqliteException);
|
||||
|
||||
private static bool IsSqliteLockException(SqliteException exception) =>
|
||||
exception.SqliteErrorCode is 5 or 6 ||
|
||||
exception.Message.Contains("database is locked", StringComparison.OrdinalIgnoreCase) ||
|
||||
exception.Message.Contains("database table is locked", StringComparison.OrdinalIgnoreCase);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user