Muhannad/visibility errors - #28
Conversation
…entation and error queue support
📝 WalkthroughWalkthroughThe PR introduces error queue reading capabilities for RabbitMQ, extends bus configuration options significantly, refactors consumer statistics tracking with caching and timestamp awareness, and updates service registration to wire HTTP clients for the new readers. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant ConsumerReader
participant MemoryCache
participant ManagementAPI
Client->>ConsumerReader: GetConsumerCounts(consumer, message)
activate ConsumerReader
ConsumerReader->>MemoryCache: Check cache for queue
alt Cache Hit
MemoryCache-->>ConsumerReader: Cached queue data
else Cache Miss
ConsumerReader->>ManagementAPI: Fetch queue data
ManagementAPI-->>ConsumerReader: Queue info
ConsumerReader->>MemoryCache: Store with expiration
ConsumerReader->>ConsumerReader: Update lastUpdatedUtc
end
ConsumerReader-->>Client: Return consumer counts
deactivate ConsumerReader
Client->>ConsumerReader: GetLastUpdated()
ConsumerReader-->>Client: lastUpdatedUtc as Task<DateTime>
sequenceDiagram
participant Client
participant ErrorQueueReader
participant ConsumerDiscovery
participant ManagementAPI
Client->>ErrorQueueReader: Peek<TConsumer, TMessage>(ErrorQueueType, count)
activate ErrorQueueReader
ErrorQueueReader->>ConsumerDiscovery: Resolve consumer definition
activate ConsumerDiscovery
ConsumerDiscovery-->>ErrorQueueReader: Consumer metadata
deactivate ConsumerDiscovery
ErrorQueueReader->>ErrorQueueReader: Compute queue name (RetryQueueName or BadQueueName)
ErrorQueueReader->>ManagementAPI: Fetch messages from queue
activate ManagementAPI
ManagementAPI-->>ErrorQueueReader: Raw messages
deactivate ManagementAPI
ErrorQueueReader->>ErrorQueueReader: Map to ErrorMessage (extract headers, correlationId, exceptions)
ErrorQueueReader-->>Client: Return IEnumerable<ErrorMessage>
deactivate ErrorQueueReader
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
SW.Bus/ConsumerReader.cs (1)
228-231: Silent exception swallowing may hide issues.The empty
catchblock silently swallows all exceptions, including connection failures, authentication errors, and timeouts. Consider logging at debug/warning level to aid troubleshooting when queue lookups fail unexpectedly.🔧 Suggested improvement
+using Microsoft.Extensions.Logging; // In constructor, add ILogger<ConsumerReader> logger parameter catch { + // Optionally log the exception for debugging + // logger?.LogDebug(ex, "Failed to get queue info for {QueueName}", queueName); return null; }SW.Bus/IServiceCollectionExtensions.cs (2)
74-74: Conflicting service registrations forIConsumerReader.Line 74 registers
IConsumerReaderas a Singleton, but lines 89-95 useAddHttpClient<IConsumerReader, ConsumerReader>()which registers it as Transient by default. TheAddHttpClientregistration will override the Singleton, potentially causing unexpected behavior where each request gets a fresh instance (losing cached state likelastUpdatedUtc).Remove the Singleton registration on line 74 since
AddHttpClienthandles the registration, or explicitly specify the lifetime onAddHttpClient.🐛 Remove duplicate registration
- services.AddSingleton<IConsumerReader, ConsumerReader>(); services.AddSingleton<ConsumerDiscovery>(); services.AddMemoryCache();If Singleton behavior is needed, consider:
services.AddHttpClient<IConsumerReader, ConsumerReader>(...) .ConfigureHttpClient(...) .AddTypedClient<IConsumerReader>((client, sp) => sp.GetRequiredService<ConsumerReader>()); // Plus additional wiring for singleton lifetimeAlso applies to: 89-95
127-128: RabbitMQ connection and model are never disposed.The
connandmodelcreated here are assigned but never disposed, leading to resource leaks. These should be registered as singletons and disposed during application shutdown.🐛 Suggested fix - register as singletons
- var conn = factory.CreateConnection(); - var model = conn.CreateModel(); + services.AddSingleton<IConnection>(sp => + { + var rabbitUrl = sp.GetRequiredService<IConfiguration>().GetConnectionString("RabbitMQ"); + var busOpts = sp.GetRequiredService<BusOptions>(); + var factory = new ConnectionFactory + { + Uri = new Uri(rabbitUrl), + ClientProvidedName = $"{Assembly.GetCallingAssembly().GetName().Name} Publisher" + }; + return factory.CreateConnection(); + }); + + services.AddSingleton<IModel>(sp => sp.GetRequiredService<IConnection>().CreateModel()); - return services.AddScoped(serviceProvider => new BasicPublisher( - model, + return services.AddScoped(serviceProvider => new BasicPublisher( + serviceProvider.GetRequiredService<IModel>(), serviceProvider.GetRequiredService<BusOptions>(), serviceProvider.GetRequiredService<RequestContext>()))
🤖 Fix all issues with AI agents
In @SW.Bus/ErrorQueueReader.cs:
- Around line 69-70: The call to consumerDiscovery.Load(true) inside the peek
flow causes expensive reflection on every call; cache the Load() result (e.g.,
store the returned definitions collection in a private field or a MemoryCache
inside ErrorQueueReader) and use the cached definitions for subsequent lookups
of definition = definitions.FirstOrDefault(...) in the peek method; add a
clear/invalidate method or hook to refresh the cached value when consumers
register/unregister (or on a configurable TTL) so you only call
consumerDiscovery.Load(true) when the cache is empty or invalidated.
In @SW.Bus/SW.Bus.csproj:
- Around line 29-32: Update the RabbitMQ.Client PackageReference in
SW.Bus.csproj from Version="6.8.1" to Version="7.2.0" so the project targets the
native net8.0 feed; locate the <PackageReference Include="RabbitMQ.Client".../>
entry and change its Version attribute to "7.2.0", then run dotnet restore and
rebuild/run tests to ensure no API breaks (adjust any code using RabbitMQ.Client
APIs if compiler errors appear).
🧹 Nitpick comments (5)
SW.Bus.RabbitMqExtensions/IConsumerReader.cs (1)
40-43: Consider using a non-async property signature.The
Task<DateTime>return type suggests asynchronous computation, but the implementation inConsumerReader.cssimply returnsTask.FromResult(lastUpdatedUtc). A plainDateTime LastUpdated { get; }would be simpler and more accurately represent the synchronous nature of this property.♻️ Suggested simplification
- public Task<DateTime> LastUpdated { get; } + public DateTime LastUpdated { get; }SW.Bus.RabbitMqExtensions/IErrorQueueReader.cs (2)
43-46: Exception key matching may be too broad.The
StartsWith("exception")filter could match unintended header keys like"exceptionType","exceptionHandler", etc. If the expected pattern isexception0,exception1, etc., consider a more precise regex or suffix check.♻️ More precise matching
public IEnumerable<string?> ExceptionHistory => Headers - .Where(h => h.Key.StartsWith("exception", StringComparison.OrdinalIgnoreCase)) + .Where(h => h.Key.StartsWith("exception", StringComparison.OrdinalIgnoreCase) + && h.Key.Length > 9 + && char.IsDigit(h.Key[9])) .OrderBy(h => GetExceptionIndex(h.Key)) .Select(h => h.Value?.ToString());
54-55: Index parsing fallback may cause incorrect ordering.When
GetExceptionIndexfails to parse the numeric suffix, it returns0. This could cause non-exception keys that slip through the filter to appear first, or cause genuinely malformed keys to sort unexpectedly.♻️ Consider returning -1 for invalid keys
private static int GetExceptionIndex(string key) => - int.TryParse(key.Replace("exception", "", StringComparison.OrdinalIgnoreCase), out var index) ? index : 0; + int.TryParse(key.Replace("exception", "", StringComparison.OrdinalIgnoreCase), out var index) ? index : int.MaxValue;This would push malformed keys to the end of the sorted list rather than the beginning.
SW.Bus/ConsumerReader.cs (1)
31-31: Consider thread-safety forlastUpdatedUtc.The
lastUpdatedUtcfield can be updated concurrently from multiple cache-miss callbacks. While DateTime assignment is atomic on 64-bit systems, consider usingvolatileorInterlockedfor explicit thread-safety guarantees if precision matters.Also applies to: 61-61
SW.Bus/IServiceCollectionExtensions.cs (1)
28-32:BuildServiceProvider()during configuration is an anti-pattern.Calling
BuildServiceProvider()creates an orphaned service provider that won't be disposed properly and may cause issues with singleton instances. Consider using theIServiceProviderfromIHostApplicationBuilderor deferring configuration access.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
SW.Bus.RabbitMqExtensions/IConsumerReader.csSW.Bus.RabbitMqExtensions/IErrorQueueReader.csSW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csprojSW.Bus/BusOptions.csSW.Bus/ConsumerReader.csSW.Bus/ErrorQueueReader.csSW.Bus/IServiceCollectionExtensions.csSW.Bus/SW.Bus.csproj
🧰 Additional context used
🧬 Code graph analysis (4)
SW.Bus/ConsumerReader.cs (3)
SW.Bus/BusOptions.cs (2)
BusOptions(11-246)BusOptions(23-37)SW.Bus/ConsumerDiscovery.cs (1)
ConsumerDiscovery(12-118)SW.Bus/ConsumerDefinition.cs (3)
ConsumerDefinition(9-77)ConsumerDefinition(15-21)ConsumerDefinition(23-32)
SW.Bus/ErrorQueueReader.cs (2)
SW.Bus/ConsumerDiscovery.cs (1)
ConsumerDiscovery(12-118)SW.Bus.RabbitMqExtensions/IErrorQueueReader.cs (3)
Task(99-101)Task(120-121)Task(137-137)
SW.Bus/IServiceCollectionExtensions.cs (4)
SW.Bus/ConsumerReader.cs (2)
ConsumerReader(19-234)ConsumerReader(40-58)SW.Bus/BusOptions.cs (2)
BusOptions(11-246)BusOptions(23-37)SW.Bus/ErrorQueueReader.cs (2)
ErrorQueueReader(17-119)ErrorQueueReader(29-34)SW.Bus/MockPublisher.cs (2)
MockPublisher(7-33)MockPublisher(11-14)
SW.Bus.RabbitMqExtensions/IConsumerReader.cs (1)
SW.Bus/ConsumerReader.cs (7)
Task(64-76)Task(79-84)Task(87-101)Task(104-108)Task(117-153)Task(179-205)Task(214-233)
🔇 Additional comments (9)
SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj (1)
5-6: LGTM!Enabling XML documentation generation with suppression of warning 1591 is a reasonable approach for gradual documentation adoption.
SW.Bus.RabbitMqExtensions/IErrorQueueReader.cs (1)
77-138: LGTM!The
IErrorQueueReaderinterface is well-designed with comprehensive XML documentation. The peek methods provide good flexibility with typed and untyped access patterns.SW.Bus/BusOptions.cs (3)
88-94: Consider credential handling practices.
ManagementPasswordis stored as a plain string. While this is often unavoidable for runtime use, ensure this value is sourced from secure configuration (e.g., Azure Key Vault, environment variables) and not logged or serialized inadvertently.
39-64: LGTM!The
MonitoringCacheSecondsproperty with value clamping between 3-60 seconds is a good defensive pattern that prevents misconfiguration.
215-244: LGTM!The
AddQueueOptionmethod provides a clean API for per-queue configuration overrides with comprehensive documentation.SW.Bus/ConsumerReader.cs (1)
181-205: LGTM!Parallel fetching of main, retry, and bad queue statistics with
Task.WhenAllis efficient and well-structured.SW.Bus/ErrorQueueReader.cs (3)
17-34: LGTM!The constructor follows proper dependency injection patterns. The
HttpClientis injected and managed externally byIHttpClientFactory, so theManagementClientdoesn't need to handle disposal.
36-42: LGTM!Clean implementation that properly resolves the message type name and delegates to the internal method.
88-118: LGTM!Well-structured implementation with:
- Proper input validation including sensible upper bound on count
- Correct use of
AckRequeueTruefor non-destructive message inspection- Appropriate exception wrapping with helpful context for network-level failures
The decision to only catch
HttpRequestExceptionis reasonable, as other exceptions (e.g., auth failures, JSON parsing) will bubble up with their original context.
| var definitions = await consumerDiscovery.Load(true); | ||
| var definition = definitions.FirstOrDefault(d => d.ServiceType == consumerType && d.MessageTypeName == messageName); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Find ConsumerDiscovery.cs file
find . -name "ConsumerDiscovery.cs" -type fRepository: simplify9/SW-Bus
Length of output: 89
🏁 Script executed:
# Once we locate the file, examine the Load method implementation
git ls-files | grep -i discovererRepository: simplify9/SW-Bus
Length of output: 42
🏁 Script executed:
# Read ConsumerDiscovery.cs to examine the Load() method
cat ./SW.Bus/ConsumerDiscovery.csRepository: simplify9/SW-Bus
Length of output: 5134
Cache consumerDiscovery.Load() results to avoid repeated reflection-based discovery.
consumerDiscovery.Load(true) is called on every peek operation, but the Load() method performs expensive reflection-based discovery with no caching—creating service scopes, scanning implemented interfaces, and invoking methods via reflection each time. Consider caching the discovery results and invalidating only when necessary (e.g., on consumer registration changes).
🤖 Prompt for AI Agents
In @SW.Bus/ErrorQueueReader.cs around lines 69 - 70, The call to
consumerDiscovery.Load(true) inside the peek flow causes expensive reflection on
every call; cache the Load() result (e.g., store the returned definitions
collection in a private field or a MemoryCache inside ErrorQueueReader) and use
the cached definitions for subsequent lookups of definition =
definitions.FirstOrDefault(...) in the peek method; add a clear/invalidate
method or hook to refresh the cached value when consumers register/unregister
(or on a configurable TTL) so you only call consumerDiscovery.Load(true) when
the cache is empty or invalidated.
| <PackageReference Include="EasyNetQ.Management.Client" Version="3.0.1"/> | ||
| <PackageReference Include="RabbitMQ.Client" Version="6.8.1"/> | ||
| <PackageReference Include="Scrutor" Version="4.2.2"/> | ||
| <PackageReference Include="SimplyWorks.HttpExtensions" Version="5.0.0"/> |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
RabbitMQ.Client NuGet latest version 2025 2026
💡 Result:
The latest RabbitMQ.Client NuGet version as of January 8, 2026 is 7.2.0. [NuGet listing]. [1]
Sources:
- NuGet: RabbitMQ.Client 7.2.0. [1]
- RabbitMQ .NET client info (distribution via NuGet). [2]
References:
[1] NuGet Gallery — RabbitMQ.Client 7.2.0.
[2] RabbitMQ — .NET/C# client library (NuGet distribution).
🌐 Web query:
EasyNetQ.Management.Client NuGet latest version 2025 2026
💡 Result:
Latest EasyNetQ.Management.Client NuGet version: 3.0.1 — published October 17, 2025. [1][2]
Sources:
[1] NuGet profile for EasyNetQ (package list showing EasyNetQ.Management.Client 3.0.1, last updated 10/17/2025).
[2] NuGetMustHaves package page for EasyNetQ.Management.Client (version 3.0.1, last update 10/17/2025).
🌐 Web query:
RabbitMQ.Client 6.8.1 .NET 8 compatibility
💡 Result:
Short answer: Yes — RabbitMQ.Client 6.8.1 (targets .NET Standard 2.0) runs on .NET 8, but it does not include a native net8.0 target. For native net8.0 builds use the 7.x series (which includes a net8.0 target).
Sources:
- RabbitMQ.Client 6.8.1 NuGet (targets .NETStandard2.0 — compatible with .NET 8). [1]
- RabbitMQ .NET client release-series info (6.x is a .NET Standard release series; 7.x has newer framework targeting). [2]
- RabbitMQ.Client 7.2.0 NuGet (explicit net8.0 target in 7.x). [3]
If you want, I can list potential runtime caveats (dependency version mismatches) and steps to test/upgrade.
Upgrade RabbitMQ.Client to 7.2.0 for proper .NET 8 support.
RabbitMQ.Client 6.8.1 is outdated; version 7.2.0 is available. While 6.8.1 runs on .NET 8 via .NET Standard 2.0 compatibility, it lacks a native net8.0 target. Upgrade to 7.2.0 for explicit .NET 8 framework support. EasyNetQ.Management.Client 3.0.1 is already at the latest version.
🤖 Prompt for AI Agents
In @SW.Bus/SW.Bus.csproj around lines 29 - 32, Update the RabbitMQ.Client
PackageReference in SW.Bus.csproj from Version="6.8.1" to Version="7.2.0" so the
project targets the native net8.0 feed; locate the <PackageReference
Include="RabbitMQ.Client".../> entry and change its Version attribute to
"7.2.0", then run dotnet restore and rebuild/run tests to ensure no API breaks
(adjust any code using RabbitMQ.Client APIs if compiler errors appear).
2 participants
Summary by CodeRabbit
Release Notes
✏️ Tip: You can customize this high-level summary in your review settings.