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. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||||
|
|||||||
+1
-1
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user