Skip to content

Fix Redis BRPOP timeout crashing background services #267

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
ModerRAS merged 1 commit into from
Apr 17, 2026
Merged

Fix Redis BRPOP timeout crashing background services #267

Show file tree
Hide file tree
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
17 changes: 15 additions & 2 deletions TelegramSearchBot/Service/AI/LLM/AgentRegistryService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,8 +184,21 @@ public async Task RunMaintenanceOnceAsync(CancellationToken cancellationToken =

protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
while (!stoppingToken.IsCancellationRequested) {
await RunMaintenanceOnceAsync(stoppingToken);
await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
try {
await RunMaintenanceOnceAsync(stoppingToken);
} catch (OperationCanceledException) {
break;
} catch (RedisException ex) {
_logger.LogWarning(ex, "Redis error in AgentRegistryService maintenance, retrying in 5 s");
} catch (Exception ex) {
_logger.LogError(ex, "Unexpected error in AgentRegistryService maintenance");
}

try {
await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
Comment on lines +192 to +198

@coderabbitai coderabbitai Bot Apr 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Log message understates the actual retry delay.

The warning hard-codes "retrying in 5 s", but the subsequent delay is Math.Max(5, Env.AgentHeartbeatIntervalSeconds). If AgentHeartbeatIntervalSeconds is configured above 5, the log will misreport the retry interval and make field diagnostics harder.

💡 Proposed tweak 
-                } catch (RedisException ex) {
-                    _logger.LogWarning(ex, "Redis error in AgentRegistryService maintenance, retrying in 5 s");
+                } catch (RedisException ex) {
+                    var retrySeconds = Math.Max(5, Env.AgentHeartbeatIntervalSeconds);
+                    _logger.LogWarning(ex, "Redis error in AgentRegistryService maintenance, retrying in {RetrySeconds} s", retrySeconds);
                 } catch (Exception ex) {
                     _logger.LogError(ex, "Unexpected error in AgentRegistryService maintenance");
                 }
 
                 try {
-                    await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
+                    await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
_logger.LogWarning(ex, "Redis error in AgentRegistryService maintenance, retrying in 5 s");
} catch (Exception ex) {
_logger.LogError(ex, "Unexpected error in AgentRegistryService maintenance");
}
try {
await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
} catch (RedisException ex) {
var retrySeconds = Math.Max(5, Env.AgentHeartbeatIntervalSeconds);
_logger.LogWarning(ex, "Redis error in AgentRegistryService maintenance, retrying in {RetrySeconds} s", retrySeconds);
} catch (Exception ex) {
_logger.LogError(ex, "Unexpected error in AgentRegistryService maintenance");
}
try {
await Task.Delay(TimeSpan.FromSeconds(Math.Max(5, Env.AgentHeartbeatIntervalSeconds)), stoppingToken);
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@TelegramSearchBot/Service/AI/LLM/AgentRegistryService.cs` around lines 192 -
198, The warning message in AgentRegistryService currently hardcodes "retrying
in 5 s" while the actual delay uses Math.Max(5,
Env.AgentHeartbeatIntervalSeconds); update the maintenance loop to compute the
delay (e.g., var retryDelay = TimeSpan.FromSeconds(Math.Max(5,
Env.AgentHeartbeatIntervalSeconds))) and use that variable in both the await
Task.Delay(...) and in the _logger.LogWarning call so the log shows the real
retryDelay value (reference: AgentRegistryService,
Env.AgentHeartbeatIntervalSeconds, _logger.LogWarning).

} catch (OperationCanceledException) {
break;
}
}
}

Expand Down
92 changes: 50 additions & 42 deletions TelegramSearchBot/Service/AI/LLM/TelegramTaskConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,57 +30,65 @@ public TelegramTaskConsumer(

protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
while (!stoppingToken.IsCancellationRequested) {
var result = await _redis.GetDatabase().ExecuteAsync("BRPOP", LlmAgentRedisKeys.TelegramTaskQueue, 5);
if (result.IsNull) {
continue;
}
try {
// Use a 2-second block time so BRPOP returns well within SE.Redis's 5 s async timeout.
var result = await _redis.GetDatabase().ExecuteAsync("BRPOP", LlmAgentRedisKeys.TelegramTaskQueue, 2);
if (result.IsNull) {
continue;
}

var parts = (RedisResult[])result!;
if (parts.Length != 2) {
continue;
}
var parts = (RedisResult[])result!;
if (parts.Length != 2) {
continue;
}

var payload = parts[1].ToString();
if (string.IsNullOrWhiteSpace(payload)) {
continue;
}
var payload = parts[1].ToString();
if (string.IsNullOrWhiteSpace(payload)) {
continue;
}

var task = JsonConvert.DeserializeObject<TelegramAgentToolTask>(payload);
if (task == null) {
continue;
}
var task = JsonConvert.DeserializeObject<TelegramAgentToolTask>(payload);
if (task == null) {
continue;
}

var response = new TelegramAgentToolResult {
RequestId = task.RequestId,
Success = false
};
var response = new TelegramAgentToolResult {
RequestId = task.RequestId,
Success = false
};

try {
if (!task.ToolName.Equals("send_message", StringComparison.OrdinalIgnoreCase)) {
throw new InvalidOperationException($"Unsupported telegram tool: {task.ToolName}");
}
try {
if (!task.ToolName.Equals("send_message", StringComparison.OrdinalIgnoreCase)) {
throw new InvalidOperationException($"Unsupported telegram tool: {task.ToolName}");
}

if (!task.Arguments.TryGetValue("text", out var text) || string.IsNullOrWhiteSpace(text)) {
throw new InvalidOperationException("send_message 缺少 text 参数。");
}
if (!task.Arguments.TryGetValue("text", out var text) || string.IsNullOrWhiteSpace(text)) {
throw new InvalidOperationException("send_message 缺少 text 参数。");
}

var chatId = task.Arguments.TryGetValue("chatId", out var chatIdString) && long.TryParse(chatIdString, out var parsedChatId)
? parsedChatId
: task.ChatId;
var chatId = task.Arguments.TryGetValue("chatId", out var chatIdString) && long.TryParse(chatIdString, out var parsedChatId)
? parsedChatId
: task.ChatId;

var sent = await _sendMessage.AddTaskWithResult(() => _botClient.SendMessage(chatId, text, cancellationToken: stoppingToken), chatId);
response.Success = true;
response.TelegramMessageId = sent.MessageId;
response.Result = sent.MessageId.ToString();
} catch (Exception ex) {
_logger.LogError(ex, "Failed to execute telegram task {RequestId}", task.RequestId);
response.ErrorMessage = ex.Message;
}
var sent = await _sendMessage.AddTaskWithResult(() => _botClient.SendMessage(chatId, text, cancellationToken: stoppingToken), chatId);
response.Success = true;
response.TelegramMessageId = sent.MessageId;
response.Result = sent.MessageId.ToString();
} catch (Exception ex) when (ex is not OperationCanceledException) {
_logger.LogError(ex, "Failed to execute telegram task {RequestId}", task.RequestId);
response.ErrorMessage = ex.Message;
}

await _redis.GetDatabase().StringSetAsync(
LlmAgentRedisKeys.TelegramResult(task.RequestId),
JsonConvert.SerializeObject(response),
TimeSpan.FromMinutes(5));
await _redis.GetDatabase().StringSetAsync(
LlmAgentRedisKeys.TelegramResult(task.RequestId),
JsonConvert.SerializeObject(response),
TimeSpan.FromMinutes(5));
} catch (OperationCanceledException) {
break;
} catch (RedisException ex) {
_logger.LogWarning(ex, "Redis error in TelegramTaskConsumer, retrying in 1 s");
await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken);
}
}
}
}
Expand Down
Loading