Skip to content

Content Service: fix MessageConsumer hanging on shutdown #710

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
YuriUfimtsev merged 1 commit into from
Nov 8, 2026
Merged

Content Service: fix MessageConsumer hanging on shutdown #710

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 47 additions & 36 deletions HwProj.ContentService/HwProj.ContentService.API/Services/MessageConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System.Collections.Concurrent;
using System.Threading.Channels;
using HwProj.ContentService.API.Models.Database;
using HwProj.ContentService.API.Models.DTO;
Expand All @@ -12,7 +13,7 @@ public class MessageConsumer : BackgroundService
{
private readonly ChannelReader<IProcessFileMessage> _channelReader;
private readonly ILogger<MessageConsumer> _logger;
private readonly HashSet<long> _filesInProcessing = new();
private readonly ConcurrentDictionary<Task, long> _backgroundTasks = new();

private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IMessageProducer _messageProducer;
Expand Down Expand Up @@ -66,21 +67,33 @@ private async Task ProcessMessage(IProcessFileMessage message, IFileRecordReposi
public override async Task StopAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("Приступаем к завершению работы обработчика сообщений MessageConsumer");

await base.StopAsync(cancellationToken);

using var scope = _serviceScopeFactory.CreateScope();
var fileRecordRepository = scope.ServiceProvider.GetRequiredService<IFileRecordRepository>();

// Обрабатываем оставшиеся сообщений в канале
// Обрабатываем оставшиеся сообщения в канале (при этом могут запуститься новые фоновые задачи)
while (_channelReader.TryRead(out var message))
{
await ProcessMessage(message, fileRecordRepository);
}

// Ждем завершения всех дополнительных задач
while (_filesInProcessing.Count > 0)
// Ждем завершения всех фоновых задач
try
{
await Task.WhenAll(_backgroundTasks.Keys).WaitAsync(cancellationToken);
}
catch (OperationCanceledException)
{
_logger.LogWarning(
"Истекло время ожидания остановки обработчика MessageConsumer. Не завершена обработка файлов: {fileIds}. " +
"Файлы будут обработаны повторно при следующем запуске сервиса",
string.Join(", ", _backgroundTasks.Values));
return;
}

// Ещё раз обрабатываем оставшиеся сообщения в канале, поскольку некоторые задачи содержат отправку сообщения
// Ещё раз обрабатываем оставшиеся сообщения в канале, поскольку фоновые задачи завершаются отправкой сообщения.
while (_channelReader.TryRead(out var message))
{
await ProcessMessage(message, fileRecordRepository);
Expand Down Expand Up @@ -126,11 +139,7 @@ await fileRecordRepository.UpdateAsync(fileRecordId,
ExternalKey: s3Key,
UploaderId: message.SenderId
);

// А также сигнализируем, что для fileRecordId есть задача, завершения которой нужно дождаться при остановке сервиса
_filesInProcessing.Add(fileRecordId);

UploadFileTask(uploadFileTaskDto);
RunInBackground(fileRecordId, () => UploadFileTask(uploadFileTaskDto));
}

private async Task HandleReUploadFileMessage(ReUploadFileMessage message, IFileRecordRepository fileRecordRepository)
Expand All @@ -144,11 +153,8 @@ private async Task HandleReUploadFileMessage(ReUploadFileMessage message, IFileR
return;
}

// Сигнализируем, что для fileRecord.Id есть задача, завершения которой нужно дождаться при остановке сервиса
_filesInProcessing.Add(fileRecord.Id);

// Не блокируя Consumer, начинаем взаимодействие с удаленным S3 по вопросу этого файла
Task.Run(async () =>
RunInBackground(fileRecord.Id, async () =>
{
// Если файл уже есть в s3, осталось просто обновить статус
var isInS3 = await _s3FilesService.CheckFileExistence(fileRecord.ExternalKey!);
Expand All @@ -164,10 +170,10 @@ private async Task HandleReUploadFileMessage(ReUploadFileMessage message, IFileR
return;
}

// Иначе формируем модель и отправляем стандартную задачу на загрузку
// Иначе формируем модель и выполняем стандартную задачу на загрузку
var uploadFileTaskDto = new UploadFileTaskDto(
FileRecordId: fileRecord.Id,
LocalFilePath: fileRecord.LocalPath,
LocalFilePath: fileRecord.LocalPath!,
ContentType: fileRecord.ContentType,
ExternalKey: fileRecord.ExternalKey!,
UploaderId: message.SenderId
Expand Down Expand Up @@ -232,10 +238,8 @@ await fileRecordRepository.UpdateAsync(message.FileId,
"Статус файла {fileId} успешно обновлён на {status} по запросу пользователя {senderId}",
message.FileId, fileRecord.Status.ToString(), message.SenderId);

// Сигнализируем, что для message.FileId есть задача, завершения которой нужно дождаться при остановке сервиса
_filesInProcessing.Add(message.FileId);

DeleteFileTask(message.FileId, fileRecord.ExternalKey!, message.SenderId);
// Не блокируя Consumer, отправляем задачу на удаление файла
RunInBackground(message.FileId, () => DeleteFileTask(message.FileId, fileRecord.ExternalKey!, message.SenderId));
}

private async Task HandleReDeleteFileMessage(ReDeleteFileMessage message, IFileRecordRepository fileRecordRepository)
Expand All @@ -248,12 +252,9 @@ private async Task HandleReDeleteFileMessage(ReDeleteFileMessage message, IFileR
message.SenderId, message.FileId);
return;
}

// Сигнализируем, что для fileRecord.Id есть задача, завершения которой нужно дождаться при остановке сервиса
_filesInProcessing.Add(fileRecord.Id);

// Не блокируя Consumer, начинаем взаимодействие с удаленным S3 по вопросу этого файла
Task.Run(async () =>
RunInBackground(fileRecord.Id, async () =>
{
// Если файла уже нет в s3, осталось просто отправить сообщение для удаления записи файла из БД
var isInS3 = await _s3FilesService.CheckFileExistence(fileRecord.ExternalKey!);
Expand All @@ -267,7 +268,7 @@ private async Task HandleReDeleteFileMessage(ReDeleteFileMessage message, IFileR
return;
}

// Иначе отправляем стандартную задачу на удаление
// Иначе выполняем стандартную задачу на удаление
await DeleteFileTask(fileRecord.Id, fileRecord.ExternalKey!, message.SenderId);
});
}
Expand All @@ -280,18 +281,37 @@ private async Task HandleFileDeletedMessage(FileDeletedMessage message, IFileRec
message.FileId, message.SenderId);
}

// Запускает задачу вне потока Consumer и запоминает её, чтобы дождаться её завершения при остановке сервиса.
// Задача удаляется из коллекции при любом исходе, в том числе при исключении
private void RunInBackground(long fileId, Func<Task> action)
{
var task = Task.Run(async () =>
{
try
{
await action();
}
catch (Exception e)
{
_logger.LogError(e, "Ошибка при фоновой обработке файла {fileId}", fileId);
}
});

_backgroundTasks.TryAdd(task, fileId);
task.ContinueWith(completedTask => _backgroundTasks.TryRemove(completedTask, out _), TaskScheduler.Default);
}

private async Task UploadFileTask(UploadFileTaskDto uploadFileTaskDto)
{
// Загружаем в S3
var fileReadStream = _localFilesService.GetFileStream(uploadFileTaskDto.LocalFilePath);
await using var fileReadStream = _localFilesService.GetFileStream(uploadFileTaskDto.LocalFilePath);
var uploadFileToS3Dto = new UploadFileToS3Dto(
FileStream: fileReadStream,
ContentType: uploadFileTaskDto.ContentType,
ExternalKey: uploadFileTaskDto.ExternalKey,
UploaderId: uploadFileTaskDto.UploaderId
);
var s3UploadingResult = await _s3FilesService.UploadFileAsync(uploadFileToS3Dto);
await fileReadStream.DisposeAsync();

if (!s3UploadingResult.Succeeded)
_logger.LogError("Не удалось загрузить файл {fileId} во внешнее хранилище. Ошибка: {error}",
Expand All @@ -305,9 +325,6 @@ private async Task UploadFileTask(UploadFileTaskDto uploadFileTaskDto)
SenderId: uploadFileToS3Dto.UploaderId
);
await _messageProducer.PushUpdateFileStatusMessage(updateStatusMessage);

// Задача для fileId завершена, новое сообщение отправлено в канал
_filesInProcessing.Remove(uploadFileTaskDto.FileRecordId);
}

private async Task DeleteFileTask(long fileId, string s3Key, string senderId)
Expand All @@ -317,9 +334,6 @@ private async Task DeleteFileTask(long fileId, string s3Key, string senderId)
{
var fileDeletedMessage = new FileDeletedMessage(fileId, senderId);
await _messageProducer.PushFileDeletedMessage(fileDeletedMessage);

// Задача для message.FileId завершена, новое сообщение отправлено в канал
_filesInProcessing.Remove(fileId);
return;
}

Expand All @@ -332,8 +346,5 @@ private async Task DeleteFileTask(long fileId, string s3Key, string senderId)
SenderId: senderId
);
await _messageProducer.PushUpdateFileStatusMessage(updateStatusMessage);

// Задача для message.FileId завершена, новое сообщение отправлено в канал
_filesInProcessing.Remove(fileId);
}
}
Loading