diff --git a/fnos/cmd/main b/fnos/cmd/main index 057b266..3ff417f 100755 --- a/fnos/cmd/main +++ b/fnos/cmd/main @@ -15,7 +15,10 @@ CA_BUNDLE="$RUNTIME_ROOT/etc/ssl/certs/ca-certificates.crt" RUN_ROOT="$DATA_ROOT/run" LOG_ROOT="$DATA_ROOT/log" APP_PID_FILE="$RUN_ROOT/liverecorder.pid" +LOG_MONITOR_PID_FILE="$RUN_ROOT/liverecorder-log-monitor.pid" APP_LOG="$LOG_ROOT/liverecorder.log" +APP_LOG_MAX_BYTES=52428800 +APP_LOG_RETAINED_FILES=3 ADMIN_PASSWORD_FILE="$DATA_ROOT/admin-password.seed" POSTGRES_ENROLLMENT_TOKEN_FILE="$DATA_ROOT/postgres-enrollment-token.seed" POSTGRES_CREDENTIALS_FILE="$DATA_ROOT/postgres-client.conf" @@ -36,6 +39,49 @@ log_message() { printf '%s - %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$1" >>"$APP_LOG" } +rotate_app_log() { + [ -f "$APP_LOG" ] || return 0 + size=$(wc -c <"$APP_LOG" 2>/dev/null | tr -d '[:space:]') + case "$size" in ''|*[!0-9]*) return 0 ;; esac + [ "$size" -ge "$APP_LOG_MAX_BYTES" ] || return 0 + + index=$APP_LOG_RETAINED_FILES + rm -f "$APP_LOG.$index" + while [ "$index" -gt 1 ]; do + previous=$((index - 1)) + if [ -f "$APP_LOG.$previous" ]; then + mv "$APP_LOG.$previous" "$APP_LOG.$index" + fi + index=$previous + done + + tail -c "$APP_LOG_MAX_BYTES" "$APP_LOG" >"$APP_LOG.1.tmp" 2>/dev/null || return 0 + mv "$APP_LOG.1.tmp" "$APP_LOG.1" + : >"$APP_LOG" +} + +start_log_monitor() { + rm -f "$LOG_MONITOR_PID_FILE" + ( + while app_pid >/dev/null 2>&1; do + rotate_app_log + sleep 60 + done + ) & + printf '%s\n' "$!" >"$LOG_MONITOR_PID_FILE" +} + +stop_log_monitor() { + if [ -f "$LOG_MONITOR_PID_FILE" ]; then + monitor_pid=$(sed -n '1p' "$LOG_MONITOR_PID_FILE" | tr -d '[:space:]') + case "$monitor_pid" in + ''|*[!0-9]*) ;; + *) kill "$monitor_pid" 2>/dev/null || true ;; + esac + fi + rm -f "$LOG_MONITOR_PID_FILE" +} + app_pid() { if [ -f "$APP_PID_FILE" ]; then pid=$(sed -n '1p' "$APP_PID_FILE" | tr -d '[:space:]') @@ -199,10 +245,13 @@ start_app() { fi rm -f "$APP_PID_FILE" + stop_log_monitor + rotate_app_log select_database_connection || return 1 launch_attempt=1 launch_app_process pid=$APP_PROCESS_PID + start_log_monitor attempt=0 while [ "$attempt" -lt 90 ]; do @@ -231,6 +280,7 @@ start_app() { } stop_app() { + stop_log_monitor if pid=$(app_pid); then kill "$pid" 2>/dev/null || true attempt=0 diff --git a/src/LiveRecorder.Infrastructure/Services/CompletionDispatchService.cs b/src/LiveRecorder.Infrastructure/Services/CompletionDispatchService.cs index f678013..054baf7 100644 --- a/src/LiveRecorder.Infrastructure/Services/CompletionDispatchService.cs +++ b/src/LiveRecorder.Infrastructure/Services/CompletionDispatchService.cs @@ -11,6 +11,9 @@ namespace LiveRecorder.Infrastructure.Services; public sealed class CompletionDispatchService { private static readonly TimeSpan RetryDelay = TimeSpan.FromMinutes(1); + private static readonly SemaphoreSlim[] SessionDispatchLocks = Enumerable.Range(0, 64) + .Select(static _ => new SemaphoreSlim(1, 1)) + .ToArray(); private readonly LiveRecorderDbContext _dbContext; private readonly IEventScriptService _eventScriptService; private readonly RecordUploadService _recordUploadService; @@ -70,8 +73,27 @@ public sealed class CompletionDispatchService return; } + var dispatchLock = GetSessionDispatchLock(task.RecordSessionId); + await dispatchLock.WaitAsync(cancellationToken); try { + await _dbContext.Entry(dispatch).ReloadAsync(cancellationToken); + if (dispatch.CompletedAt.HasValue) + { + return; + } + + await _dbContext.Entry(task).ReloadAsync(cancellationToken); + if (task.Result is not null) + { + await _dbContext.Entry(task.Result).ReloadAsync(cancellationToken); + } + + if (task.RecordSession is not null) + { + await _dbContext.Entry(task.RecordSession).ReloadAsync(cancellationToken); + } + var consolidation = await _shortFragmentConsolidationService.PrepareDispatchAsync(task.Id, cancellationToken); if (consolidation.SkipDispatch) { @@ -90,15 +112,17 @@ public sealed class CompletionDispatchService await _dbContext.Entry(task).ReloadAsync(cancellationToken); await _dbContext.Entry(task).Reference(item => item.Result).LoadAsync(cancellationToken); + var recordSession = task.RecordSession ?? throw new InvalidOperationException(); + var recordResult = task.Result ?? throw new InvalidOperationException(); if (!dispatch.ScriptDispatched) { var scriptResult = await _eventScriptService.RunSegmentCompletedAsync( task.LiveRoom, - task.RecordSession, + recordSession, task, - task.Result, - task.Result.FilePath, + recordResult, + recordResult.FilePath, task.EndedAt ?? DateTimeOffset.UtcNow, eventId: dispatch.Id, cancellationToken: cancellationToken); @@ -125,6 +149,16 @@ public sealed class CompletionDispatchService dispatch.ScheduleRetry(ex.Message, DateTimeOffset.UtcNow.Add(RetryDelay), DateTimeOffset.UtcNow); await _dbContext.SaveChangesAsync(cancellationToken); } + finally + { + dispatchLock.Release(); + } + } + + private static SemaphoreSlim GetSessionDispatchLock(Guid recordSessionId) + { + var hash = recordSessionId.GetHashCode() & int.MaxValue; + return SessionDispatchLocks[hash % SessionDispatchLocks.Length]; } } diff --git a/src/LiveRecorder.Infrastructure/Services/OpenListUploadQueueService.cs b/src/LiveRecorder.Infrastructure/Services/OpenListUploadQueueService.cs index e724371..009ec58 100644 --- a/src/LiveRecorder.Infrastructure/Services/OpenListUploadQueueService.cs +++ b/src/LiveRecorder.Infrastructure/Services/OpenListUploadQueueService.cs @@ -114,7 +114,6 @@ public sealed class OpenListUploadQueueService !item.IsHiddenArtifactSource && item.Result != null && item.Result.UploadStatus == RecordArtifactUploadStatus.NotUploaded && - item.UploadJob == null && item.UpdatedAt <= completedBefore && !string.IsNullOrWhiteSpace(item.Result.FilePath)) .OrderBy(static item => item.UpdatedAt) @@ -122,6 +121,28 @@ public sealed class OpenListUploadQueueService .Take(Math.Clamp(take, 1, MaxAutomaticRecoveryBatchSize)) .ToArrayAsync(cancellationToken); + var changedArtifactTaskIds = await _dbContext.RecordTasks + .AsNoTracking() + .Where(item => + (item.Status == RecordTaskStatus.Completed || item.Status == RecordTaskStatus.Stopped) && + !item.IsHiddenArtifactSource && + item.Result != null && + item.UploadJob != null && + item.Result.UploadStatus == RecordArtifactUploadStatus.Succeeded && + item.UploadJob.Status == RecordArtifactUploadStatus.Succeeded && + item.Result.FileSizeBytes != item.UploadJob.VideoSizeBytes && + !string.IsNullOrWhiteSpace(item.Result.FilePath)) + .OrderBy(static item => item.UpdatedAt) + .Select(static item => item.Id) + .Take(Math.Clamp(take, 1, MaxAutomaticRecoveryBatchSize)) + .ToArrayAsync(cancellationToken); + + taskIds = taskIds + .Concat(changedArtifactTaskIds) + .Distinct() + .Take(Math.Clamp(take, 1, MaxAutomaticRecoveryBatchSize)) + .ToArray(); + var recovered = 0; foreach (var taskId in taskIds) { diff --git a/src/LiveRecorder.Infrastructure/Services/ShortFragmentConsolidationService.cs b/src/LiveRecorder.Infrastructure/Services/ShortFragmentConsolidationService.cs index 1db7527..72a3d4f 100644 --- a/src/LiveRecorder.Infrastructure/Services/ShortFragmentConsolidationService.cs +++ b/src/LiveRecorder.Infrastructure/Services/ShortFragmentConsolidationService.cs @@ -51,6 +51,7 @@ public sealed class ShortFragmentConsolidationService var tasks = await _dbContext.RecordTasks .Include(item => item.Result) + .Include(item => item.UploadJob) .Where(item => item.RecordSessionId == task.RecordSessionId && !item.IsHiddenArtifactSource) .OrderBy(item => item.SegmentIndex) .ThenBy(item => item.CreatedAt) @@ -159,11 +160,22 @@ public sealed class ShortFragmentConsolidationService private static List ResolveMergeGroup(IReadOnlyList tasks, int index) { - var durations = tasks.Select(item => IsReadableTerminal(item) ? item.Result!.DurationSeconds : null).ToArray(); + var durations = tasks + .Select(item => IsReadableTerminal(item) && IsMergeEligible(item.Result!.UploadStatus, item.UploadJob is not null) + ? item.Result.DurationSeconds + : null) + .ToArray(); var (start, end) = ResolveMergeWindow(durations, index); - return tasks.Skip(start).Take(end - start + 1).Where(IsReadableTerminal).ToList(); + return tasks + .Skip(start) + .Take(end - start + 1) + .Where(item => IsReadableTerminal(item) && IsMergeEligible(item.Result!.UploadStatus, item.UploadJob is not null)) + .ToList(); } + internal static bool IsMergeEligible(RecordArtifactUploadStatus uploadStatus, bool hasUploadJob) => + uploadStatus == RecordArtifactUploadStatus.NotUploaded && !hasUploadJob; + internal static (int Start, int End) ResolveMergeWindow(IReadOnlyList durations, int index) { if (durations.Count == 0 || index < 0 || index >= durations.Count) return (-1, -1); diff --git a/tests/LiveRecorder.Tests/OpenListUploadTests.cs b/tests/LiveRecorder.Tests/OpenListUploadTests.cs index 3469913..1377a69 100644 --- a/tests/LiveRecorder.Tests/OpenListUploadTests.cs +++ b/tests/LiveRecorder.Tests/OpenListUploadTests.cs @@ -553,6 +553,40 @@ public sealed class OpenListUploadTests Assert.Equal(1, await fixture.Context.RecordUploadJobs.CountAsync()); } + [Fact] + public async Task AutomaticRecovery_RequeuesSucceededJobWhenMergedArtifactDrifts() + { + await using var fixture = await QueueFixture.CreateAsync(); + await fixture.Queue.EnqueueAsync(fixture.RecordTaskId); + var originalJob = await fixture.Context.RecordUploadJobs.AsNoTracking().SingleAsync(); + fixture.OpenList.Objects[originalJob.TargetVideoPath] = new OpenListObjectInfo( + "segment.mp4", + originalJob.VideoSizeBytes, + false, + new Dictionary + { + ["sha256"] = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes("video-content"))).ToLowerInvariant() + }); + Assert.True(await fixture.Queue.ProcessNextAsync()); + + var result = await fixture.Context.RecordResults.SingleAsync(); + var mergedPath = Path.Combine(Path.GetDirectoryName(result.FilePath)!, "segment-merged-recovery.mp4"); + await File.WriteAllBytesAsync(mergedPath, Encoding.UTF8.GetBytes("merged-video-content")); + result.Update(mergedPath, new FileInfo(mergedPath).Length, 61, null, 0, RecordTaskStatus.Completed, null); + await fixture.Context.SaveChangesAsync(); + + var recovered = await fixture.Queue.RecoverPendingAutomaticUploadsAsync(); + + Assert.Equal(1, recovered); + fixture.Context.ChangeTracker.Clear(); + var refreshedJob = await fixture.Context.RecordUploadJobs.SingleAsync(); + var refreshedResult = await fixture.Context.RecordResults.SingleAsync(); + Assert.Equal(RecordArtifactUploadStatus.Queued, refreshedJob.Status); + Assert.Equal(RecordArtifactUploadStatus.Queued, refreshedResult.UploadStatus); + Assert.EndsWith("segment-merged-recovery.mp4", refreshedJob.TargetVideoPath, StringComparison.Ordinal); + Assert.Equal(originalJob.TargetVideoPath, refreshedJob.PendingCleanupVideoPath); + } + [Fact] public async Task CompletionOutbox_IsCreatedWhenRecordResultWasWrittenOutsideEfTracking() { diff --git a/tests/LiveRecorder.Tests/RecordingContinuityTests.cs b/tests/LiveRecorder.Tests/RecordingContinuityTests.cs index c4fea0f..04fd48a 100644 --- a/tests/LiveRecorder.Tests/RecordingContinuityTests.cs +++ b/tests/LiveRecorder.Tests/RecordingContinuityTests.cs @@ -94,6 +94,18 @@ public sealed class RecordingContinuityTests Assert.Equal((0, 2), window); } + [Theory] + [InlineData(RecordArtifactUploadStatus.NotUploaded, false, true)] + [InlineData(RecordArtifactUploadStatus.NotUploaded, true, false)] + [InlineData(RecordArtifactUploadStatus.Queued, true, false)] + [InlineData(RecordArtifactUploadStatus.Uploading, true, false)] + [InlineData(RecordArtifactUploadStatus.Succeeded, true, false)] + public void ShortFragmentMerge_RejectsArtifactsThatEnteredUpload( + RecordArtifactUploadStatus uploadStatus, + bool hasUploadJob, + bool expected) => + Assert.Equal(expected, ShortFragmentConsolidationService.IsMergeEligible(uploadStatus, hasUploadJob)); + [Fact] public void DanmakuMerge_ShiftsRelativeOffset_ButPreservesAbsoluteTimestamp() {