nanxun/live_recorder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recover merged uploads and rotate logs

Browse Source
This commit is contained in:
nanxun
2026-08-14 13:09:49 +08:00
parent 7a33b2c6e3
commit 51e4c41ee2
6 changed files with 169 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fnos/cmd/main
+50
View File
@@ -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
src/LiveRecorder.Infrastructure/Services/CompletionDispatchService.cs
+37 -3
View File
@@ -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];
}
}
src/LiveRecorder.Infrastructure/Services/OpenListUploadQueueService.cs
+22 -1
View File
@@ -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)
{
src/LiveRecorder.Infrastructure/Services/ShortFragmentConsolidationService.cs
+14 -2
View File
@@ -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<RecordTask> ResolveMergeGroup(IReadOnlyList<RecordTask> 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<double?> durations, int index)
{
if (durations.Count == 0 || index < 0 || index >= durations.Count) return (-1, -1);
tests/LiveRecorder.Tests/OpenListUploadTests.cs
+34
View File
@@ -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<string, string>
{
["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()
{
tests/LiveRecorder.Tests/RecordingContinuityTests.cs
+12
View File
@@ -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()
{