Fix processing pipeline: report album failures, fault handling on create slips, config key names, audio duration

This commit is contained in:
2026-09-24 05:02:04 +02:00
parent c8e9c7790e
commit ab388e6ad6
15 changed files with 119 additions and 32 deletions
@@ -45,7 +45,7 @@ public class RequestAlbumPictureUploadCommandHandler(
var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}"; var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}";
var tempKey = albumConfiguration.Routes.BuildTempPicturePath( var tempKey = albumConfiguration.Routes.BuildTempPicturePath(
uploadIntentConfiguration.TemporalRootPrefix, request.UserId, objectName); uploadIntentConfiguration.TempRootPrefix, request.UserId, objectName);
try try
{ {
@@ -31,11 +31,11 @@ public class UploadIntentConfiguration
public int ExpiredIntentsRetentionDays { get; set; } = 7; public int ExpiredIntentsRetentionDays { get; set; } = 7;
[Required, MinLength(1)] [Required, MinLength(1)]
public string TemporalRootPrefix { get; set; } = "temporal"; public string TempRootPrefix { get; set; } = "temp";
[Range(1, 30)] [Range(1, 30)]
public int TemporalUploadsRetentionDays { get; set; } = 3; public int TempUploadsRetentionDays { get; set; } = 3;
[Range(60, 86400)] [Range(60, 86400)]
public int TemporalCleanUpJobIntervalSeconds { get; set; } = 3600; public int TempCleanupJobIntervalSeconds { get; set; } = 3600;
} }
@@ -45,7 +45,7 @@ public class RequestPlayListPictureUploadCommandHandler(
var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}"; var objectName = $"{Guid.NewGuid()}.{request.FileType.TrimStart('.').ToLowerInvariant()}";
var tempKey = playListConfiguration.Routes.BuildTempPicturePath( var tempKey = playListConfiguration.Routes.BuildTempPicturePath(
uploadIntentConfiguration.TemporalRootPrefix, request.UserId, objectName); uploadIntentConfiguration.TempRootPrefix, request.UserId, objectName);
try try
{ {
@@ -51,9 +51,9 @@ public class RequestTrackUploadUrlsCommandHandler(
var audioObjectName = $"{Guid.NewGuid()}.{request.AudioFileType.TrimStart('.').ToLowerInvariant()}"; var audioObjectName = $"{Guid.NewGuid()}.{request.AudioFileType.TrimStart('.').ToLowerInvariant()}";
var tempPictureKey = trackConfiguration.Routes.BuildTempPicturePath( var tempPictureKey = trackConfiguration.Routes.BuildTempPicturePath(
uploadIntentConfiguration.TemporalRootPrefix, request.UserId, pictureObjectName); uploadIntentConfiguration.TempRootPrefix, request.UserId, pictureObjectName);
var tempAudioKey = trackConfiguration.Routes.BuildTempAudioPath( var tempAudioKey = trackConfiguration.Routes.BuildTempAudioPath(
uploadIntentConfiguration.TemporalRootPrefix, request.UserId, audioObjectName); uploadIntentConfiguration.TempRootPrefix, request.UserId, audioObjectName);
try try
{ {
@@ -16,10 +16,10 @@ public class TemporalUploadsCleanUpJob(
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
logger.LogInformation("TemporalUploadsCleanUpJob started (interval: {Interval}s, prefix: {Prefix})", logger.LogInformation("TemporalUploadsCleanUpJob started (interval: {Interval}s, prefix: {Prefix})",
uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds, uploadIntentConfiguration.TempCleanupJobIntervalSeconds,
uploadIntentConfiguration.TemporalRootPrefix); uploadIntentConfiguration.TempRootPrefix);
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds)); using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TempCleanupJobIntervalSeconds));
do do
{ {
@@ -40,13 +40,13 @@ public class TemporalUploadsCleanUpJob(
await using var scope = scopeFactory.CreateAsyncScope(); await using var scope = scopeFactory.CreateAsyncScope();
var storageService = scope.ServiceProvider.GetRequiredService<IStorageService>(); var storageService = scope.ServiceProvider.GetRequiredService<IStorageService>();
var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TemporalUploadsRetentionDays); var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TempUploadsRetentionDays);
var deleted = 0; var deleted = 0;
var errors = 0; var errors = 0;
await foreach (var (key, lastModified) in storageService.ListObjectsAsync( await foreach (var (key, lastModified) in storageService.ListObjectsAsync(
storageConfiguration.Bucket, storageConfiguration.Bucket,
uploadIntentConfiguration.TemporalRootPrefix, uploadIntentConfiguration.TempRootPrefix,
cancellationToken)) cancellationToken))
{ {
if (lastModified >= cutOff) if (lastModified >= cutOff)
@@ -55,7 +55,7 @@ internal class TranscodeAudioActivity(
var log = new TranscodeAudioLog(workingDirectory); var log = new TranscodeAudioLog(workingDirectory);
return executeContext.CompletedWithVariables(log, new Dictionary<string, object> return executeContext.CompletedWithVariables(log, new Dictionary<string, object>
{ {
[RoutingSlipVariableNames.Audio.DurationSeconds] = (int)Math.Round(transcodeResult.Duration.TotalSeconds) [RoutingSlipVariableNames.Audio.DurationSeconds] = transcodeResult.Duration.TotalSeconds
}); });
} }
catch (Exception exception) catch (Exception exception)
@@ -38,11 +38,7 @@ internal class UpdateTrackAudioActivity(
if (string.IsNullOrWhiteSpace(audioFolderKey)) if (string.IsNullOrWhiteSpace(audioFolderKey))
{ {
logger.LogWarning("Audio folder key is empty for track {TrackId}; skipping update", track.Id); throw new InvalidOperationException($"Audio folder key is empty for track {track.Id}");
await publishEndpoint.Publish(
new TrackAudioProcessed(executeContext.Arguments.TrackId),
executeContext.CancellationToken);
return executeContext.Completed();
} }
try try
@@ -32,6 +32,28 @@ public class ProcessingSlipFaultConsumer : IConsumer<RoutingSlipFaulted>
case RoutingSlipVariableNames.ProcessKinds.PlayListPicture: case RoutingSlipVariableNames.ProcessKinds.PlayListPicture:
await context.Publish(new PlayListPictureProcessingFailed(subjectId)); await context.Publish(new PlayListPictureProcessingFailed(subjectId));
break; break;
case RoutingSlipVariableNames.ProcessKinds.AlbumPicture:
await context.Publish(new AlbumPictureProcessingFailed(subjectId));
break;
case RoutingSlipVariableNames.ProcessKinds.TrackCreation:
await context.Publish(new TrackProcessingFailed(
subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket),
Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey),
Variable(variables, RoutingSlipVariableNames.Workflow.AudioKey)));
break;
case RoutingSlipVariableNames.ProcessKinds.AlbumCreation:
await context.Publish(new AlbumProcessingFailed(
subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket),
Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey)));
break;
case RoutingSlipVariableNames.ProcessKinds.PlayListCreation:
await context.Publish(new PlayListProcessingFailed(
subjectId, Variable(variables, RoutingSlipVariableNames.Workflow.Bucket),
Variable(variables, RoutingSlipVariableNames.Workflow.PictureKey)));
break;
} }
} }
private static string? Variable(IDictionary<string, object> variables, string name) =>
variables.TryGetValue(name, out var value) ? value?.ToString() : null;
} }
@@ -22,7 +22,8 @@ public class AlbumPictureSourceRoutingSlipBuilder
message.Small, message.Small,
message.Medium, message.Medium,
message.Large, message.Large,
correlationId); correlationId,
RoutingSlipVariableNames.ProcessKinds.AlbumCreation);
public RoutingSlipBuilder Build(UpdateAlbumPictureSourceEvent message, Guid? correlationId) public RoutingSlipBuilder Build(UpdateAlbumPictureSourceEvent message, Guid? correlationId)
=> BuildSlip( => BuildSlip(
@@ -34,7 +35,8 @@ public class AlbumPictureSourceRoutingSlipBuilder
message.Small, message.Small,
message.Medium, message.Medium,
message.Large, message.Large,
correlationId); correlationId,
RoutingSlipVariableNames.ProcessKinds.AlbumPicture);
private static RoutingSlipBuilder BuildSlip( private static RoutingSlipBuilder BuildSlip(
Guid albumId, Guid albumId,
@@ -45,7 +47,8 @@ public class AlbumPictureSourceRoutingSlipBuilder
ImageSize small, ImageSize small,
ImageSize medium, ImageSize medium,
ImageSize large, ImageSize large,
Guid? correlationId) Guid? correlationId,
string processKind)
{ {
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty); routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
@@ -79,6 +82,14 @@ public class AlbumPictureSourceRoutingSlipBuilder
medium, medium,
large)); large));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, albumId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder; return routingSlipBuilder;
} }
} }
@@ -57,6 +57,15 @@ public class CreateTrackRoutingSlipBuilder
message.Medium, message.Medium,
message.Large)); message.Large));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackCreation);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, message.Bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, message.PictureDestinationKey);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.AudioKey, message.AudioDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder; return routingSlipBuilder;
} }
} }
@@ -22,7 +22,8 @@ public class PlayListPictureSourceRoutingSlipBuilder
message.Small, message.Small,
message.Medium, message.Medium,
message.Large, message.Large,
correlationId); correlationId,
RoutingSlipVariableNames.ProcessKinds.PlayListCreation);
public RoutingSlipBuilder Build(UpdatePlayListPictureSourceEvent message, Guid? correlationId) public RoutingSlipBuilder Build(UpdatePlayListPictureSourceEvent message, Guid? correlationId)
=> BuildSlip( => BuildSlip(
@@ -34,7 +35,8 @@ public class PlayListPictureSourceRoutingSlipBuilder
message.Small, message.Small,
message.Medium, message.Medium,
message.Large, message.Large,
correlationId); correlationId,
RoutingSlipVariableNames.ProcessKinds.PlayListPicture);
private static RoutingSlipBuilder BuildSlip( private static RoutingSlipBuilder BuildSlip(
Guid playListId, Guid playListId,
@@ -45,7 +47,8 @@ public class PlayListPictureSourceRoutingSlipBuilder
ImageSize small, ImageSize small,
ImageSize medium, ImageSize medium,
ImageSize large, ImageSize large,
Guid? correlationId) Guid? correlationId,
string processKind)
{ {
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty); routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
@@ -79,6 +82,14 @@ public class PlayListPictureSourceRoutingSlipBuilder
medium, medium,
large)); large));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, playListId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder; return routingSlipBuilder;
} }
} }
@@ -8,6 +8,9 @@ internal static class RoutingSlipVariableNames
internal const string CorrelationId = "Workflow.CorrelationId"; internal const string CorrelationId = "Workflow.CorrelationId";
internal const string SubjectId = "Workflow.SubjectId"; internal const string SubjectId = "Workflow.SubjectId";
internal const string ProcessKind = "Workflow.ProcessKind"; internal const string ProcessKind = "Workflow.ProcessKind";
internal const string Bucket = "Workflow.Bucket";
internal const string PictureKey = "Workflow.PictureKey";
internal const string AudioKey = "Workflow.AudioKey";
} }
internal static class ProcessKinds internal static class ProcessKinds
@@ -16,6 +19,9 @@ internal static class RoutingSlipVariableNames
internal const string TrackAudio = "TrackAudio"; internal const string TrackAudio = "TrackAudio";
internal const string PlayListPicture = "PlayListPicture"; internal const string PlayListPicture = "PlayListPicture";
internal const string AlbumPicture = "AlbumPicture"; internal const string AlbumPicture = "AlbumPicture";
internal const string TrackCreation = "TrackCreation";
internal const string AlbumCreation = "AlbumCreation";
internal const string PlayListCreation = "PlayListCreation";
} }
internal static class Audio internal static class Audio
@@ -17,8 +17,16 @@ public sealed class AlbumProcessingStateMachine : MassTransitStateMachine<AlbumP
InstanceState(state => state.CurrentState); InstanceState(state => state.CurrentState);
Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.AlbumId)); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.AlbumId));
Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.AlbumId)); Event(() => PictureProcessed, config =>
Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.AlbumId)); {
config.CorrelateById(context => context.Message.AlbumId);
config.OnMissingInstance(instance => instance.Discard());
});
Event(() => PictureFailed, config =>
{
config.CorrelateById(context => context.Message.AlbumId);
config.OnMissingInstance(instance => instance.Discard());
});
Initially( Initially(
When(ProcessingStarted) When(ProcessingStarted)
@@ -17,8 +17,16 @@ public sealed class PlayListProcessingStateMachine : MassTransitStateMachine<Pla
InstanceState(state => state.CurrentState); InstanceState(state => state.CurrentState);
Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.PlayListId)); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.PlayListId));
Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.PlayListId)); Event(() => PictureProcessed, config =>
Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.PlayListId)); {
config.CorrelateById(context => context.Message.PlayListId);
config.OnMissingInstance(instance => instance.Discard());
});
Event(() => PictureFailed, config =>
{
config.CorrelateById(context => context.Message.PlayListId);
config.OnMissingInstance(instance => instance.Discard());
});
Initially( Initially(
When(ProcessingStarted) When(ProcessingStarted)
@@ -19,10 +19,26 @@ public sealed class TrackProcessingStateMachine : MassTransitStateMachine<TrackP
InstanceState(state => state.CurrentState); InstanceState(state => state.CurrentState);
Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.TrackId)); Event(() => ProcessingStarted, config => config.CorrelateById(context => context.Message.TrackId));
Event(() => PictureProcessed, config => config.CorrelateById(context => context.Message.TrackId)); Event(() => PictureProcessed, config =>
Event(() => AudioProcessed, config => config.CorrelateById(context => context.Message.TrackId)); {
Event(() => PictureFailed, config => config.CorrelateById(context => context.Message.TrackId)); config.CorrelateById(context => context.Message.TrackId);
Event(() => AudioFailed, config => config.CorrelateById(context => context.Message.TrackId)); config.OnMissingInstance(instance => instance.Discard());
});
Event(() => AudioProcessed, config =>
{
config.CorrelateById(context => context.Message.TrackId);
config.OnMissingInstance(instance => instance.Discard());
});
Event(() => PictureFailed, config =>
{
config.CorrelateById(context => context.Message.TrackId);
config.OnMissingInstance(instance => instance.Discard());
});
Event(() => AudioFailed, config =>
{
config.CorrelateById(context => context.Message.TrackId);
config.OnMissingInstance(instance => instance.Discard());
});
Initially( Initially(
When(ProcessingStarted) When(ProcessingStarted)