Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion .github/copilot-instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
- When modifying a public member, make sure to update the XML Documentation and the corresponding docs in the docs folder and the README file if applicable.
- When modifying a public member, make sure to check the Public.API files for the affected assembly and update them if necessary.
- Do not ignore CS1591 warnings; analyze and add missing XML comments instead.
- Do not add comments inside methods; for complex logic, consider extracting it into a separate method with a descriptive name instead of adding comments.
- Comments inside method bodies must explain *why*, not *what*: a why-comment states a constraint, invariant, or rationale the code cannot express on its own (e.g. concurrency/ordering requirements, the reason for a workaround, a justification for otherwise-surprising deliberate behavior). Narration/what-comments — restating what the next line does, section-header comments, play-by-play commentary — are not allowed. For complex logic, prefer extracting it into a separate method with a descriptive name over adding a comment.

## Testing Guidelines
- Prefer using the Vulthil.xUnit testing framework for tests.
Expand Down
20 changes: 20 additions & 0 deletions src/Vulthil.Messaging/Queues/DeadLetterDefinition.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
namespace Vulthil.Messaging.Queues;

/// <summary>
/// Describes the dead letter queue and exchange configuration.
/// </summary>
public sealed record DeadLetterDefinition
{
/// <summary>
/// Gets or sets the dead letter queue name, or <see langword="null"/> to use the default.
/// </summary>
public string? QueueName { get; set; }
/// <summary>
/// Gets or sets the dead letter exchange name, or <see langword="null"/> to use the default.
/// </summary>
public string? ExchangeName { get; set; }
/// <summary>
/// Gets or sets a value indicating whether dead letter routing is enabled.
/// </summary>
public bool Enabled { get; set; }
}
261 changes: 0 additions & 261 deletions src/Vulthil.Messaging/Queues/IQueueConfigurator.cs
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
using System.Reflection;
using System.Security.Cryptography;
using Vulthil.Messaging.Abstractions.Consumers;

namespace Vulthil.Messaging.Queues;
Expand Down Expand Up @@ -79,263 +78,3 @@ public interface IQueueConfigurator
/// </summary>
IQueueConfigurator UseSingleActiveConsumer();
}

/// <summary>
/// Defines a retry policy including intervals, jitter, and exception filtering.
/// </summary>
public sealed record RetryPolicyDefinition
{
/// <summary>
/// Gets or sets the maximum number of retry attempts.
/// </summary>
public int MaxRetryCount { get; set; }
/// <summary>
/// Gets or sets the jitter factor (0.0–1.0) applied to retry intervals.
/// </summary>
public double JitterFactor { get; set; }
/// <summary>
/// Gets or sets a value indicating whether retries run in-memory — the consumer is re-invoked in-process
/// while the delivery is held — rather than via delayed re-delivery through the retry queue. In-memory
/// retries preserve message order (a later message cannot overtake the one being retried), so they are
/// used automatically for partitioned queues. Defaults to <see langword="false"/> (delayed re-delivery).
/// </summary>
public bool InMemory { get; set; }
/// <summary>
/// Gets the delay intervals between successive retry attempts.
/// </summary>
public ICollection<TimeSpan> Intervals { get; } = [];
/// <summary>
/// Gets the fully-qualified type names of exceptions that should be excluded from retry attempts.
/// Names outside the core library must be assembly-qualified to resolve; a name that cannot be
/// resolved is skipped (the RabbitMQ transport logs a startup warning for each such name).
/// </summary>
public ICollection<string> IgnoreExceptions { get; } = [];

private readonly object _ignoredTypesGate = new();
private readonly HashSet<Type> _ignoredTypes = [];
private HashSet<Type>? _resolvedIgnoredTypes;

internal void AddIgnoredType(Type type) => _ignoredTypes.Add(type);
/// <summary>
/// Gets the resolved CLR exception types that should be excluded from retry attempts. The set is built
/// once, on first access, from the fluently ignored types and the resolvable names in
/// <see cref="IgnoreExceptions"/>; the resolution is thread-safe and later mutations of either
/// collection do not change the result.
/// </summary>
public HashSet<Type> GetIgnoredExceptionTypes()
{
lock (_ignoredTypesGate)
{
return _resolvedIgnoredTypes ??= ResolveIgnoredExceptionTypes();
}
}

private HashSet<Type> ResolveIgnoredExceptionTypes()
{
var resolved = new HashSet<Type>(_ignoredTypes);
foreach (var name in IgnoreExceptions)
{
if (Type.GetType(name, false) is { } type)
{
resolved.Add(type);
}
}

return resolved;
}

/// <summary>
/// Calculates the delay for the specified retry attempt, applying jitter when configured.
/// Attempts beyond the last configured interval reuse that interval (capped back-off).
/// </summary>
/// <param name="attempt">The zero-based attempt index.</param>
/// <returns>The delay before the next retry.</returns>
public TimeSpan GetDelay(int attempt)
{
if (Intervals.Count == 0)
{
return TimeSpan.Zero;
}

var interval = GetIntervalForAttempt(attempt);
if (JitterFactor <= 0.0)
{
return interval;
}

var jitterMultiplier = RandomNumberGeneratorExtensions.GetDouble() * 2 * JitterFactor - JitterFactor;
var finalDelay = interval.TotalMilliseconds * (1 + jitterMultiplier);
return TimeSpan.FromMilliseconds(Math.Max(0, finalDelay));
}

private TimeSpan GetIntervalForAttempt(int attempt)
{
var index = Math.Min(attempt, Intervals.Count - 1);
return Intervals is IList<TimeSpan> list ? list[index] : Intervals.ElementAt(index);
}
}

internal static class RandomNumberGeneratorExtensions
{
/// <summary>
/// Generates a cryptographically random <see langword="double"/> in the range [0, 1). Retry jitter does
/// not need cryptographic strength, but the repository enforces CA5394 (no <see cref="Random"/>) as an
/// error, so the secure generator is used here as well.
/// </summary>
public static double GetDouble()
{
var bytes = RandomNumberGenerator.GetBytes(8);
var ul = BitConverter.ToUInt64(bytes, 0) / (1 << 11);
return ul / (double)(1UL << 53);
}
}

/// <summary>
/// Describes the dead letter queue and exchange configuration.
/// </summary>
public sealed record DeadLetterDefinition
{
/// <summary>
/// Gets or sets the dead letter queue name, or <see langword="null"/> to use the default.
/// </summary>
public string? QueueName { get; set; }
/// <summary>
/// Gets or sets the dead letter exchange name, or <see langword="null"/> to use the default.
/// </summary>
public string? ExchangeName { get; set; }
/// <summary>
/// Gets or sets a value indicating whether dead letter routing is enabled.
/// </summary>
public bool Enabled { get; set; }
}

/// <summary>
/// Fluent builder for constructing a <see cref="RetryPolicyDefinition"/>.
/// </summary>
public sealed class RetryPolicyConfigurator
{
private readonly HashSet<Type> _ignoredExceptions = [];
private List<TimeSpan> _intervals = [];
/// <summary>
/// Gets the maximum number of retry attempts configured via one of the interval methods.
/// </summary>
public int RetryLimit { get; private set; }
private double _jitterFactor;
private bool _inMemory;
/// <summary>
/// Gets the configured delay intervals between successive retry attempts.
/// </summary>
public IReadOnlyList<TimeSpan> Intervals => _intervals;
/// <summary>
/// Gets the exception types that should be excluded from retry attempts and immediately surfaced.
/// </summary>
public IReadOnlySet<Type> IgnoredExceptions => _ignoredExceptions;

/// <summary>
/// Configures immediate retries with zero delay.
/// </summary>
/// <param name="retryCount">The number of retries.</param>
public void Immediate(int retryCount)
{
RetryLimit = retryCount;
_intervals = [.. Enumerable.Repeat(TimeSpan.Zero, retryCount)];
}

/// <summary>
/// Configures explicit retry intervals.
/// </summary>
/// <param name="intervals">The delay between each retry attempt.</param>
public void SetIntervals(params TimeSpan[] intervals)
{
RetryLimit = intervals.Length;
_intervals = [.. intervals];
}
/// <summary>
/// Adds random jitter to retry intervals.
/// </summary>
/// <param name="factor">The jitter factor between 0.0 and 1.0.</param>
/// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="factor"/> is outside the valid range.</exception>
public void UseJitter(double factor)
{
if (factor is < 0 or > 1.0)
{
throw new ArgumentOutOfRangeException(nameof(factor), "Jitter must be between 0.0 and 1.0");
}

_jitterFactor = factor;
}
/// <summary>
/// Configures exponential back-off retry intervals.
/// </summary>
/// <param name="retryCount">The number of retries.</param>
/// <param name="initialInterval">The delay for the first retry.</param>
/// <param name="maxInterval">The maximum delay cap.</param>
/// <param name="multiplier">The multiplier applied to each successive interval. Default is 2.0.</param>
public void Exponential(
int retryCount,
TimeSpan initialInterval,
TimeSpan maxInterval,
double multiplier = 2.0)
{
RetryLimit = retryCount;
var intervals = new List<TimeSpan>();
var currentInterval = initialInterval;

for (int i = 0; i < retryCount; i++)
{
intervals.Add(currentInterval);

// Calculate next interval
double nextTicks = currentInterval.Ticks * multiplier;
currentInterval = TimeSpan.FromTicks((long)nextTicks);

// Cap it at the max interval
if (currentInterval > maxInterval)
{
currentInterval = maxInterval;
}
}

_intervals = [.. intervals];
}

/// <summary>
/// Adds an exception type to the ignore list; matching exceptions will not trigger retries.
/// </summary>
/// <typeparam name="TException">The exception type to ignore.</typeparam>
public void Ignore<TException>() where TException : Exception => _ignoredExceptions.Add(typeof(TException));

/// <summary>
/// Configures retries to run in-memory: the consumer is re-invoked in-process while the delivery is held,
/// instead of being re-delivered later through the retry queue. This preserves message order, so it is the
/// behavior used automatically for partitioned queues.
/// </summary>
public void InMemory() => _inMemory = true;

/// <summary>
/// Builds the <see cref="RetryPolicyDefinition"/> from the current configuration.
/// </summary>
/// <returns>The constructed retry policy definition.</returns>
public RetryPolicyDefinition Build()
{
var definition = new RetryPolicyDefinition
{
MaxRetryCount = RetryLimit,
JitterFactor = _jitterFactor,
InMemory = _inMemory
};

foreach (var interval in _intervals)
{
definition.Intervals.Add(interval);
}

foreach (var type in _ignoredExceptions)
{
definition.AddIgnoredType(type);
definition.IgnoreExceptions.Add(type.AssemblyQualifiedName!);
}

return definition;
}
}
12 changes: 4 additions & 8 deletions src/Vulthil.Messaging/Queues/QueueConfigurator.cs
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ public IQueueConfigurator AddRequestConsumer<TConsumer>(Action<IRequestConfigura
var reqType = new MessageType(args[0]);
var resType = args[1];

// Preservation of your original validation
if (!_messagingOptions.RegisterRequestType(reqType))
{
throw new InvalidOperationException($"Request '{reqType.Name}' is already handled elsewhere.");
Expand Down Expand Up @@ -168,10 +167,9 @@ public IQueueConfigurator UseSingleActiveConsumer()
/// </exception>
internal void Build()
{
// 1. Auto-subscribe any concrete TMessage from consumer registrations not yet subscribed.
// Routing keys are a Subscription-level concern, so auto-subscribed
// subscriptions get a null routing key (broker uses an empty pattern). For direct/topic exchanges that
// require a specific pattern, the caller must explicitly q.Subscribe<TConcrete>("pattern") first.
// Routing keys are a Subscription-level concern, so auto-subscribed subscriptions get a null routing
// key (broker uses an empty pattern). For direct/topic exchanges that require a specific pattern, the
// caller must explicitly q.Subscribe<TConcrete>("pattern") first.
var concreteConsumerMessageTypes = _queueDefinition.Registrations
.Select(r => r.MessageType)
.Where(m => m.Type is { IsAbstract: false, IsInterface: false });
Expand All @@ -181,7 +179,7 @@ internal void Build()
_queueDefinition.AddSubscription(new Subscription(messageType));
}

// 2. Request consumers cannot be polymorphic — the response type is fixed and can't be selected
// Request consumers cannot be polymorphic — the response type is fixed and can't be selected
// by the incoming concrete type.
foreach (var rpc in _queueDefinition.Registrations.OfType<RequestConsumerRegistration>())
{
Expand All @@ -194,7 +192,6 @@ internal void Build()
}
}

// 3. Every consumer must have at least one matching concrete subscription.
var orphanConsumer = _queueDefinition.Registrations.FirstOrDefault(
r => !_queueDefinition.Subscriptions.Any(s => r.MessageType.Type.IsAssignableFrom(s.MessageType.Type)));
if (orphanConsumer is not null)
Expand All @@ -205,7 +202,6 @@ internal void Build()
"Call q.Subscribe<TConcrete>() or q.SubscribeAll<TInterface>(assembly) for at least one implementer.");
}

// 4. Every subscription must have at least one matching consumer.
var orphanSubscription = _queueDefinition.Subscriptions.FirstOrDefault(
s => !_queueDefinition.Registrations.Any(r => r.MessageType.Type.IsAssignableFrom(s.MessageType.Type)));
if (orphanSubscription is not null)
Expand Down
18 changes: 18 additions & 0 deletions src/Vulthil.Messaging/Queues/RandomNumberGeneratorExtensions.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
using System.Security.Cryptography;

namespace Vulthil.Messaging.Queues;

internal static class RandomNumberGeneratorExtensions
{
/// <summary>
/// Generates a cryptographically random <see langword="double"/> in the range [0, 1). Retry jitter does
/// not need cryptographic strength, but the repository enforces CA5394 (no <see cref="Random"/>) as an
/// error, so the secure generator is used here as well.
/// </summary>
public static double GetDouble()
{
var bytes = RandomNumberGenerator.GetBytes(8);
var ul = BitConverter.ToUInt64(bytes, 0) / (1 << 11);
return ul / (double)(1UL << 53);
}
}
Loading