refactor(infra): S3 client singleton, ffmpeg pipe ordering, tidy streams/jobs

- Register IAmazonS3 as a singleton instead of scoped: the AWS SDK client is
  thread-safe and manages its own connection pool, so one instance per DI scope
  (per request / per job iteration) was wasteful and risked socket exhaustion
- AudioTranscoderService: start reading stdout/stderr before writing stdin to
  remove the latent pipe-buffer deadlock shape
- Dispose the resized-picture MemoryStream in ResizePictureActivity
- Background jobs use PeriodicTimer instead of while + Task.Delay

Verified: Worker boots against RabbitMQ + Postgres in Docker with no captive-
dependency error (dev scope validation) and both jobs start.

Build: 0 warnings, 0 errors.
This commit is contained in:
2026-07-05 15:07:49 +02:00
parent 93ee8be0cd
commit 6e4b8fad56
5 changed files with 20 additions and 13 deletions
@@ -19,7 +19,9 @@ namespace Musify.Infrastructure.Jobs
uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds, uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds,
uploadIntentConfiguration.TemporalRootPrefix); uploadIntentConfiguration.TemporalRootPrefix);
while (!stoppingToken.IsCancellationRequested) using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds));
do
{ {
try try
{ {
@@ -29,9 +31,8 @@ namespace Musify.Infrastructure.Jobs
{ {
logger.LogError(exception, "Error in TemporalUploadsCleanUpJob"); logger.LogError(exception, "Error in TemporalUploadsCleanUpJob");
} }
await Task.Delay(TimeSpan.FromSeconds(uploadIntentConfiguration.TemporalCleanUpJobIntervalSeconds), stoppingToken);
} }
while (await timer.WaitForNextTickAsync(stoppingToken));
} }
async Task RunAsync(CancellationToken cancellationToken) async Task RunAsync(CancellationToken cancellationToken)
@@ -18,7 +18,9 @@ namespace Musify.Infrastructure.Jobs
{ {
logger.LogInformation("UploadIntentExpirationJob started (interval: {Interval}s)", uploadIntentConfiguration.ExpirationJobIntervalSeconds); logger.LogInformation("UploadIntentExpirationJob started (interval: {Interval}s)", uploadIntentConfiguration.ExpirationJobIntervalSeconds);
while (!stoppingToken.IsCancellationRequested) using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.ExpirationJobIntervalSeconds));
do
{ {
try try
{ {
@@ -28,9 +30,8 @@ namespace Musify.Infrastructure.Jobs
{ {
logger.LogError(exception, "Error in UploadIntentExpirationJob"); logger.LogError(exception, "Error in UploadIntentExpirationJob");
} }
await Task.Delay(TimeSpan.FromSeconds(uploadIntentConfiguration.ExpirationJobIntervalSeconds), stoppingToken);
} }
while (await timer.WaitForNextTickAsync(stoppingToken));
} }
async Task RunAsync(CancellationToken cancellationToken) async Task RunAsync(CancellationToken cancellationToken)
@@ -30,7 +30,7 @@ namespace Musify.Infrastructure.MassTransit.Activities.Pictures
await using var fileStream = File.OpenRead(sourceFilePath); await using var fileStream = File.OpenRead(sourceFilePath);
var resizedPicture = await pictureHandler.ResizePictureAsWebpAsync( await using var resizedPicture = await pictureHandler.ResizePictureAsWebpAsync(
fileStream, fileStream,
executeContext.Arguments.Width, executeContext.Arguments.Width,
executeContext.Arguments.Height, executeContext.Arguments.Height,
@@ -103,13 +103,16 @@ namespace Musify.Infrastructure.Services
logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory); logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory);
var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
if (inputStream.CanSeek) if (inputStream.CanSeek)
inputStream.Position = 0; inputStream.Position = 0;
await inputStream.CopyToAsync(process.StandardInput.BaseStream, cancellationToken); await inputStream.CopyToAsync(process.StandardInput.BaseStream, cancellationToken);
process.StandardInput.Close(); process.StandardInput.Close();
return await WaitForProcessResultAsync(process, operationName, cancellationToken); return await WaitForProcessResultAsync(process, outputTask, errorTask, operationName, cancellationToken);
} }
async Task<FfmpegExecutionResult> ExecuteFfmpegAsync( async Task<FfmpegExecutionResult> ExecuteFfmpegAsync(
@@ -123,7 +126,10 @@ namespace Musify.Infrastructure.Services
logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory); logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory);
return await WaitForProcessResultAsync(process, operationName, cancellationToken); var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
return await WaitForProcessResultAsync(process, outputTask, errorTask, operationName, cancellationToken);
} }
Process BuildProcess(string arguments, string workingDirectory, bool redirectStandardInput) Process BuildProcess(string arguments, string workingDirectory, bool redirectStandardInput)
@@ -144,12 +150,11 @@ namespace Musify.Infrastructure.Services
async Task<FfmpegExecutionResult> WaitForProcessResultAsync( async Task<FfmpegExecutionResult> WaitForProcessResultAsync(
Process process, Process process,
Task<string> outputTask,
Task<string> errorTask,
string operationName, string operationName,
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
using var linkedCancellationTokens = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); using var linkedCancellationTokens = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
linkedCancellationTokens.CancelAfter(audioTranscoderConfiguration.TranscodingTimeout); linkedCancellationTokens.CancelAfter(audioTranscoderConfiguration.TranscodingTimeout);
@@ -21,7 +21,7 @@ namespace Musify.Infrastructure.Services
serviceDescriptors.AddSingleton(serviceProvider => serviceDescriptors.AddSingleton(serviceProvider =>
serviceProvider.GetRequiredService<IOptions<InfrastructureStorageConfiguration>>().Value); serviceProvider.GetRequiredService<IOptions<InfrastructureStorageConfiguration>>().Value);
serviceDescriptors.AddScoped<IAmazonS3>(serviceProvider => serviceDescriptors.AddSingleton<IAmazonS3>(serviceProvider =>
{ {
var configuration = serviceProvider.GetRequiredService<InfrastructureStorageConfiguration>(); var configuration = serviceProvider.GetRequiredService<InfrastructureStorageConfiguration>();
var s3Configuration = new AmazonS3Config var s3Configuration = new AmazonS3Config