refactor: dedupe inbox registration, drop dead options validation, and cover retention paths (#349)

Vulthil · web-flow · commit 8c72721d6ead · 2026-07-21T21:14:18.000+02:00
diff --git a/src/Vulthil.Messaging.Inbox.Cosmos/CosmosIdempotencyStore.cs b/src/Vulthil.Messaging.Inbox.Cosmos/CosmosIdempotencyStore.cs
@@ -34,10 +34,34 @@ public async Task<int> DeleteProcessedAsync(DateTimeOffset olderThanUtc, int bat
         }
 
         dbContext.InboxMessages.RemoveRange(markers);
-        await dbContext.SaveChangesAsync(cancellationToken);
+        await SaveRemovedMarkersAsync(cancellationToken);
         return markers.Count;
     }
 
+    /// <summary>
+    /// Saves the pending removals, treating a concurrency conflict as progress rather than a failure: it means a
+    /// concurrent sweeper already deleted the same marker, so the goal (the row being gone) is already achieved.
+    /// </summary>
+    private async Task SaveRemovedMarkersAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            await dbContext.SaveChangesAsync(cancellationToken);
+        }
+        catch (DbUpdateConcurrencyException exception)
+        {
+            DetachConflictingEntries(exception);
+        }
+    }
+
+    private static void DetachConflictingEntries(DbUpdateConcurrencyException exception)
+    {
+        foreach (var entry in exception.Entries)
+        {
+            entry.State = EntityState.Detached;
+        }
+    }
+
     public async Task<bool> ProcessAsync(string idempotencyKey, IMessageContext context, Func<CancellationToken, Task> process, CancellationToken cancellationToken)
     {
         ArgumentException.ThrowIfNullOrEmpty(idempotencyKey);
diff --git a/src/Vulthil.Messaging.Inbox.Cosmos/CosmosInboxExtensions.cs b/src/Vulthil.Messaging.Inbox.Cosmos/CosmosInboxExtensions.cs
@@ -30,16 +30,8 @@ public static IServiceCollection AddCosmosInbox<TContext>(
     {
         ArgumentNullException.ThrowIfNull(services);
 
-        services.TryAddSingleton(TimeProvider.System);
         services.TryAddScoped<IIdempotencyStore, CosmosIdempotencyStore<TContext>>();
-        services.AddInboxRetention(configure);
-
-        var options = new InboxOptions();
-        configure?.Invoke(options);
-        if (options.EnableMetrics)
-        {
-            services.AddOpenTelemetry().WithMetrics(metrics => metrics.AddVulthilInboxInstrumentation());
-        }
+        services.AddInboxCore(configure);
 
         return services;
     }
diff --git a/src/Vulthil.Messaging.Inbox.Cosmos/Vulthil.Messaging.Inbox.Cosmos.csproj b/src/Vulthil.Messaging.Inbox.Cosmos/Vulthil.Messaging.Inbox.Cosmos.csproj
@@ -12,7 +12,6 @@
 
 	<ItemGroup>
 	  <PackageReference Include="Microsoft.EntityFrameworkCore.Cosmos" />
-	  <PackageReference Include="OpenTelemetry.Extensions.Hosting" />
 	</ItemGroup>
 
 </Project>
diff --git a/src/Vulthil.Messaging.Inbox.Relational/RelationalInboxExtensions.cs b/src/Vulthil.Messaging.Inbox.Relational/RelationalInboxExtensions.cs
@@ -30,16 +30,8 @@ public static IServiceCollection AddRelationalInbox<TContext>(
     {
         ArgumentNullException.ThrowIfNull(services);
 
-        services.TryAddSingleton(TimeProvider.System);
         services.TryAddScoped<IIdempotencyStore, RelationalIdempotencyStore<TContext>>();
-        services.AddInboxRetention(configure);
-
-        var options = new InboxOptions();
-        configure?.Invoke(options);
-        if (options.EnableMetrics)
-        {
-            services.AddOpenTelemetry().WithMetrics(metrics => metrics.AddVulthilInboxInstrumentation());
-        }
+        services.AddInboxCore(configure);
 
         return services;
     }
diff --git a/src/Vulthil.Messaging.Inbox.Relational/Vulthil.Messaging.Inbox.Relational.csproj b/src/Vulthil.Messaging.Inbox.Relational/Vulthil.Messaging.Inbox.Relational.csproj
@@ -12,7 +12,6 @@
 
 	<ItemGroup>
 	  <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" />
-	  <PackageReference Include="OpenTelemetry.Extensions.Hosting" />
 	</ItemGroup>
 
 </Project>
diff --git a/src/Vulthil.Messaging.Inbox/IdempotentConsumeFilter.cs b/src/Vulthil.Messaging.Inbox/IdempotentConsumeFilter.cs
@@ -24,14 +24,15 @@ internal sealed class IdempotentConsumeFilter<TMessage>(
     ILogger<IdempotentConsumeFilter<TMessage>> logger) : IConsumeFilter<TMessage>
     where TMessage : notnull
 {
+    private static readonly string _messageType = typeof(TMessage).FullName ?? typeof(TMessage).Name;
+
     private readonly IIdempotencyStore _store = store;
     private readonly IInboxKeySelector<TMessage> _keySelector = keySelector;
     private readonly InboxOptions _options = options.Value;
     private readonly ILogger<IdempotentConsumeFilter<TMessage>> _logger = logger;
 
     public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegate<TMessage> next)
     {
-        var messageType = typeof(TMessage).FullName ?? typeof(TMessage).Name;
         var key = _keySelector.GetKey(context);
         key = string.IsNullOrEmpty(key) ? context.MessageId : key;
 
@@ -42,7 +43,7 @@ public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegat
                 throw new MissingIdempotencyKeyException(typeof(TMessage));
             }
 
-            InboxLog.MissingKeyAllowed(_logger, messageType);
+            InboxLog.MissingKeyAllowed(_logger, _messageType);
             InboxTelemetry.MissingKey.Add(1);
             await next(context);
             return;
@@ -56,7 +57,7 @@ public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegat
         }
         else
         {
-            InboxLog.DuplicateSkipped(_logger, messageType, key);
+            InboxLog.DuplicateSkipped(_logger, _messageType, key);
             InboxTelemetry.DuplicateSkipped.Add(1);
         }
     }
diff --git a/src/Vulthil.Messaging.Inbox/InboxCoreServiceCollectionExtensions.cs b/src/Vulthil.Messaging.Inbox/InboxCoreServiceCollectionExtensions.cs
@@ -0,0 +1,40 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+
+namespace Vulthil.Messaging.Inbox;
+
+/// <summary>
+/// Registration shared by every inbox store package (<c>AddRelationalInbox</c>, <c>AddCosmosInbox</c>). The store
+/// extensions differ only in which <see cref="IIdempotencyStore"/> implementation they register; the retention
+/// sweep, OpenTelemetry metrics, and <see cref="TimeProvider"/> registration are identical regardless of the store,
+/// so they live here once instead of being duplicated per store package. Internal, and visible to the store
+/// packages via <c>InternalsVisibleTo</c>.
+/// </summary>
+internal static class InboxCoreServiceCollectionExtensions
+{
+    /// <summary>
+    /// Registers everything an <c>AddXyzInbox</c> store extension needs beyond its own store type: the shared
+    /// <see cref="TimeProvider"/>, the retention sweep (gated by <see cref="InboxRetentionOptions.Enabled"/>), and
+    /// OpenTelemetry metrics (gated by <see cref="InboxOptions.EnableMetrics"/>). <paramref name="configure"/> is
+    /// invoked exactly once here to evaluate both gates from a single materialized <see cref="InboxOptions"/>
+    /// instance, plus once more when the options system resolves <see cref="Microsoft.Extensions.Options.IOptions{TOptions}"/>
+    /// for injected consumers.
+    /// </summary>
+    internal static IServiceCollection AddInboxCore(this IServiceCollection services, Action<InboxOptions>? configure)
+    {
+        ArgumentNullException.ThrowIfNull(services);
+
+        var options = new InboxOptions();
+        configure?.Invoke(options);
+
+        services.TryAddSingleton(TimeProvider.System);
+        services.RegisterInboxRetention(configure, options);
+
+        if (options.EnableMetrics)
+        {
+            services.AddOpenTelemetry().WithMetrics(metrics => metrics.AddVulthilInboxInstrumentation());
+        }
+
+        return services;
+    }
+}
diff --git a/src/Vulthil.Messaging.Inbox/InboxRetentionOptions.cs b/src/Vulthil.Messaging.Inbox/InboxRetentionOptions.cs
@@ -1,5 +1,3 @@
-using System.ComponentModel.DataAnnotations;
-
 namespace Vulthil.Messaging.Inbox;
 
 /// <summary>
@@ -29,6 +27,5 @@ public sealed class InboxRetentionOptions
     /// Gets or sets the maximum number of markers deleted per batch within a sweep; a sweep keeps deleting batches
     /// until fewer than this many markers remain. Default is 1000.
     /// </summary>
-    [Range(1, int.MaxValue)]
     public int BatchSize { get; set; } = 1000;
 }
diff --git a/src/Vulthil.Messaging.Inbox/InboxRetentionServiceCollectionExtensions.cs b/src/Vulthil.Messaging.Inbox/InboxRetentionServiceCollectionExtensions.cs
@@ -1,4 +1,5 @@
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 
 namespace Vulthil.Messaging.Inbox;
 
@@ -14,9 +15,31 @@ public static class InboxRetentionServiceCollectionExtensions
     /// do).
     /// </summary>
     /// <param name="services">The service collection.</param>
-    /// <param name="configure">An optional action to configure <see cref="InboxOptions"/>.</param>
+    /// <param name="configure">
+    /// An optional action to configure <see cref="InboxOptions"/>. Invoked once eagerly (to evaluate
+    /// <see cref="InboxRetentionOptions.Enabled"/>) and once more when the options system materializes
+    /// <see cref="Microsoft.Extensions.Options.IOptions{TOptions}"/> for injected consumers.
+    /// </param>
     /// <returns>The same service collection, for chaining.</returns>
-    public static IServiceCollection AddInboxRetention(this IServiceCollection services, Action<InboxOptions>? configure)
+    public static IServiceCollection AddInboxRetention(this IServiceCollection services, Action<InboxOptions>? configure = null)
+    {
+        ArgumentNullException.ThrowIfNull(services);
+
+        services.TryAddSingleton(TimeProvider.System);
+
+        var options = new InboxOptions();
+        configure?.Invoke(options);
+
+        return services.RegisterInboxRetention(configure, options);
+    }
+
+    /// <summary>
+    /// Registers the options validation and the conditional hosted-service registration for the inbox retention
+    /// sweep, using an already-materialized <paramref name="options"/> instance so a shared caller (e.g.
+    /// <c>AddInboxCore</c>) does not need to invoke <paramref name="configure"/> a second time just to re-evaluate
+    /// the <see cref="InboxRetentionOptions.Enabled"/> gate.
+    /// </summary>
+    internal static IServiceCollection RegisterInboxRetention(this IServiceCollection services, Action<InboxOptions>? configure, InboxOptions options)
     {
         services.AddOptions<InboxOptions>()
             .Configure(configure ?? (static _ => { }))
@@ -26,11 +49,8 @@ public static IServiceCollection AddInboxRetention(this IServiceCollection servi
                     && o.Retention.SweepInterval > TimeSpan.Zero
                     && o.Retention.BatchSize >= 1,
                 "Inbox retention requires RetentionPeriod and SweepInterval greater than zero and BatchSize of at least 1 when enabled.")
-            .ValidateDataAnnotations()
             .ValidateOnStart();
 
-        var options = new InboxOptions();
-        configure?.Invoke(options);
         if (options.Retention.Enabled)
         {
             services.AddHostedService<InboxRetentionBackgroundService>();
diff --git a/src/Vulthil.Messaging.Inbox/PublicAPI.Shipped.txt b/src/Vulthil.Messaging.Inbox/PublicAPI.Shipped.txt
@@ -5,7 +5,7 @@ Vulthil.Messaging.Inbox.InboxTelemetry
 Vulthil.Messaging.Inbox.MeterProviderBuilderExtensions
 static readonly Vulthil.Messaging.Inbox.InboxTelemetry.MeterName -> string!
 static Vulthil.Messaging.Inbox.MeterProviderBuilderExtensions.AddVulthilInboxInstrumentation(this OpenTelemetry.Metrics.MeterProviderBuilder! builder) -> OpenTelemetry.Metrics.MeterProviderBuilder!
-static Vulthil.Messaging.Inbox.InboxRetentionServiceCollectionExtensions.AddInboxRetention(this Microsoft.Extensions.DependencyInjection.IServiceCollection! services, System.Action<Vulthil.Messaging.Inbox.InboxOptions!>? configure) -> Microsoft.Extensions.DependencyInjection.IServiceCollection!
+static Vulthil.Messaging.Inbox.InboxRetentionServiceCollectionExtensions.AddInboxRetention(this Microsoft.Extensions.DependencyInjection.IServiceCollection! services, System.Action<Vulthil.Messaging.Inbox.InboxOptions!>? configure = null) -> Microsoft.Extensions.DependencyInjection.IServiceCollection!
 Vulthil.Messaging.Inbox.IIdempotencyStore
 Vulthil.Messaging.Inbox.IIdempotencyStore.ProcessAsync(string! idempotencyKey, Vulthil.Messaging.Abstractions.Consumers.IMessageContext! context, System.Func<System.Threading.CancellationToken, System.Threading.Tasks.Task!>! process, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task<bool>!
 Vulthil.Messaging.Inbox.IInboxKeySelector<TMessage>
diff --git a/src/Vulthil.Messaging.Inbox/Vulthil.Messaging.Inbox.csproj b/src/Vulthil.Messaging.Inbox/Vulthil.Messaging.Inbox.csproj
@@ -7,12 +7,17 @@
 
 	<ItemGroup>
 		<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
-		<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" />
 		<PackageReference Include="OpenTelemetry.Api" />
+		<PackageReference Include="OpenTelemetry.Extensions.Hosting" />
 	</ItemGroup>
 
 	<ItemGroup>
 	  <ProjectReference Include="..\Vulthil.Messaging\Vulthil.Messaging.csproj" />
 	</ItemGroup>
-	
+
+	<ItemGroup>
+		<InternalsVisibleTo Include="Vulthil.Messaging.Inbox.Relational" />
+		<InternalsVisibleTo Include="Vulthil.Messaging.Inbox.Cosmos" />
+	</ItemGroup>
+
 </Project>
diff --git a/src/Vulthil.SharedKernel.Outbox/OutboxEngineServiceCollectionExtensions.cs b/src/Vulthil.SharedKernel.Outbox/OutboxEngineServiceCollectionExtensions.cs
@@ -62,7 +62,21 @@ public static IServiceCollection AddOutboxEngine(
             .Validate(
                 o => o.MaxDelaySeconds >= o.OutboxProcessingDelaySeconds,
                 "MaxDelaySeconds must be greater than or equal to OutboxProcessingDelaySeconds.")
-            .ValidateDataAnnotations()
+            .Validate(
+                o => o.OutboxProcessingDelaySeconds is >= 1 and <= 100,
+                "OutboxProcessingDelaySeconds must be between 1 and 100.")
+            .Validate(
+                o => o.MaxDelaySeconds is >= 1 and <= 300,
+                "MaxDelaySeconds must be between 1 and 300.")
+            .Validate(
+                o => o.BatchSize >= 1,
+                "BatchSize must be at least 1.")
+            .Validate(
+                o => o.MaxRetries >= 1,
+                "MaxRetries must be at least 1.")
+            .Validate(
+                o => o.MaxDegreeOfParallelism is >= 1 and <= 100,
+                "MaxDegreeOfParallelism must be between 1 and 100.")
             .ValidateOnStart();
 
         var options = new OutboxProcessingOptions();
diff --git a/src/Vulthil.SharedKernel.Outbox/OutboxProcessingOptions.cs b/src/Vulthil.SharedKernel.Outbox/OutboxProcessingOptions.cs
@@ -1,5 +1,3 @@
-using System.ComponentModel.DataAnnotations;
-
 namespace Vulthil.SharedKernel.Outbox;
 
 /// <summary>
@@ -10,23 +8,19 @@ public sealed class OutboxProcessingOptions
     /// <summary>
     /// Gets the base polling delay in seconds between outbox processing cycles. Default is 2.
     /// </summary>
-    [Range(1, 100)]
     public int OutboxProcessingDelaySeconds { get; set; } = 2;
     /// <summary>
     /// Gets the maximum back-off delay in seconds when no messages are found. Default is 60.
     /// </summary>
-    [Range(1, 300)]
     public int MaxDelaySeconds { get; set; } = 60;
     /// <summary>
     /// Gets the maximum number of outbox messages to process per cycle. Default is 10.
     /// </summary>
-    [Range(1, int.MaxValue)]
     public int BatchSize { get; set; } = 10;
     /// <summary>
     /// Gets the maximum number of delivery attempts before a message is dead-lettered — its <c>FailedOnUtc</c>
     /// is set and it is no longer relayed. Default is 3.
     /// </summary>
-    [Range(1, int.MaxValue)]
     public int MaxRetries { get; set; } = 3;
     /// <summary>
     /// Gets a value indicating whether messages within a batch should be published in parallel.
@@ -38,7 +32,6 @@ public sealed class OutboxProcessingOptions
     /// Gets the maximum number of messages dispatched concurrently when <see cref="EnableParallelPublishing"/> is
     /// enabled. Default is 4. Ignored when parallel publishing is disabled.
     /// </summary>
-    [Range(1, 100)]
     public int MaxDegreeOfParallelism { get; set; } = 4;
 
     /// <summary>
diff --git a/src/Vulthil.SharedKernel.Outbox/OutboxRetentionOptions.cs b/src/Vulthil.SharedKernel.Outbox/OutboxRetentionOptions.cs
@@ -1,5 +1,3 @@
-using System.ComponentModel.DataAnnotations;
-
 namespace Vulthil.SharedKernel.Outbox;
 
 /// <summary>
@@ -28,6 +26,5 @@ public sealed class OutboxRetentionOptions
     /// Gets or sets the maximum number of rows deleted per batch within a sweep; a sweep keeps deleting batches until
     /// fewer than this many rows remain. Default is 1000.
     /// </summary>
-    [Range(1, int.MaxValue)]
     public int BatchSize { get; set; } = 1000;
 }
diff --git a/src/Vulthil.SharedKernel.Outbox/Vulthil.SharedKernel.Outbox.csproj b/src/Vulthil.SharedKernel.Outbox/Vulthil.SharedKernel.Outbox.csproj
@@ -8,7 +8,6 @@
 	<ItemGroup>
 		<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
 		<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" />
-		<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" />
 		<PackageReference Include="OpenTelemetry.Api" />
 		<PackageReference Include="OpenTelemetry.Extensions.Hosting" />
 	</ItemGroup>
diff --git a/tests/Vulthil.Messaging.Inbox.Relational.Tests/RelationalInboxExtensionsTests.cs b/tests/Vulthil.Messaging.Inbox.Relational.Tests/RelationalInboxExtensionsTests.cs
diff --git a/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionBackgroundServiceTests.cs b/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionBackgroundServiceTests.cs
diff --git a/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionServiceCollectionExtensionsTests.cs b/tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionServiceCollectionExtensionsTests.cs
diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs

Original file line number	Diff line number	Diff line change
`@@ -24,14 +24,15 @@ internal sealed class IdempotentConsumeFilter<TMessage>(`
`24`	`24`	`ILogger<IdempotentConsumeFilter<TMessage>> logger) : IConsumeFilter<TMessage>`
`25`	`25`	`where TMessage : notnull`
`26`	`26`	`{`
	`27`	`+ private static readonly string _messageType = typeof(TMessage).FullName ?? typeof(TMessage).Name;`
	`28`	`+`
`27`	`29`	`private readonly IIdempotencyStore _store = store;`
`28`	`30`	`private readonly IInboxKeySelector<TMessage> _keySelector = keySelector;`
`29`	`31`	`private readonly InboxOptions _options = options.Value;`
`30`	`32`	`private readonly ILogger<IdempotentConsumeFilter<TMessage>> _logger = logger;`
`31`	`33`
`32`	`34`	`public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegate<TMessage> next)`
`33`	`35`	`{`
`34`		`- var messageType = typeof(TMessage).FullName ?? typeof(TMessage).Name;`
`35`	`36`	`var key = _keySelector.GetKey(context);`
`36`	`37`	`key = string.IsNullOrEmpty(key) ? context.MessageId : key;`
`37`	`38`
`@@ -42,7 +43,7 @@ public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegat`
`42`	`43`	`throw new MissingIdempotencyKeyException(typeof(TMessage));`
`43`	`44`	`}`
`44`	`45`
`45`		`- InboxLog.MissingKeyAllowed(_logger, messageType);`
	`46`	`+ InboxLog.MissingKeyAllowed(_logger, _messageType);`
`46`	`47`	`InboxTelemetry.MissingKey.Add(1);`
`47`	`48`	`await next(context);`
`48`	`49`	`return;`
`@@ -56,7 +57,7 @@ public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegat`
`56`	`57`	`}`
`57`	`58`	`else`
`58`	`59`	`{`
`59`		`- InboxLog.DuplicateSkipped(_logger, messageType, key);`
	`60`	`+ InboxLog.DuplicateSkipped(_logger, _messageType, key);`
`60`	`61`	`InboxTelemetry.DuplicateSkipped.Add(1);`
`61`	`62`	`}`
`62`	`63`	`}`