Add Service Bus distributed cache expiration

Author: pwelter34

.github/workflows/dotnet.yml

@@ -46,7 +46,22 @@ jobs:
         run: dotnet build --no-restore --configuration Release
 
       - name: Run Test
-        run: dotnet test --no-build --configuration Release
+        run: dotnet test --no-build --configuration Release --treenode-filter "/*/*/*/*[Category!=LocalOnly]" -- --coverage
+
+      - name: Collect Coverage
+        if: success()
+        id: coverage
+        run: |
+          FILES=$(find ${{github.workspace}} -name '*.coverage' -print | tr '\n' ' ')
+          echo "files=$FILES" >> $GITHUB_OUTPUT
+
+      - name: Report Coverage
+        if: success() && steps.coverage.outputs.files != ''
+        uses: coverallsapp/github-action@v2
+        continue-on-error: true
+        with:
+          files: ${{ steps.coverage.outputs.files }}
+          format: cobertura
 
       - name: Create Packages
         if: success() && github.event_name != 'pull_request'

docs/guide/behaviors/caching.md

@@ -122,6 +122,87 @@ public override string? GetCacheTag() => $"Products Category:{CategoryId}";
 public string? GetCacheTag() => $"Products Category:{CategoryId}";
 ```
 
+### Multiple Cache Tags
+
+`ICacheExpire` exposes `GetCacheTags()` for commands that need to expire more than one tag for a single change. The default implementation returns the single value from `GetCacheTag()`, so existing implementations continue to work without modification. Override `GetCacheTags()` to invalidate multiple tags at once:
+
+```csharp
+public record MoveProductCommand : PrincipalCommandBase<ProductReadModel>, ICacheExpire
+{
+    public MoveProductCommand(ClaimsPrincipal principal, int fromCategoryId, int toCategoryId)
+        : base(principal)
+    {
+        FromCategoryId = fromCategoryId;
+        ToCategoryId = toCategoryId;
+    }
+
+    public int FromCategoryId { get; }
+
+    public int ToCategoryId { get; }
+
+    public string? GetCacheKey() => null;
+
+    public string? GetCacheTag() => $"Products Category:{FromCategoryId}";
+
+    // expire both the source and destination category tags
+    public override IEnumerable<string> GetCacheTags() =>
+    [
+        $"Products Category:{FromCategoryId}",
+        $"Products Category:{ToCategoryId}",
+    ];
+}
+```
+
+### Distributed Cache Expiration over Azure Service Bus
+
+When the same application is hosted across multiple processes (for example, a load-balanced web farm or several independent services), each process keeps its own local `HybridCache` tier. The `Arbiter.Messaging.ServiceBus` package provides an opt-in feature that publishes cache expiration messages to a Service Bus topic so every subscribed process expires the matching entries.
+
+The feature is built on top of the local `HybridCacheExpireBehavior`:
+
+- The publisher behavior (`ServiceBusCacheExpireBehavior`) expires the local cache first, then publishes a lightweight `CacheExpireMessage` (cache key, tags, and a per-process `SourceId`) to the configured topic.
+- The subscriber processor (`CacheExpireProcessor`) receives messages from a topic subscription and expires the matching key and tags from its own local cache. Messages originating from the same process are skipped using the `SourceId`, since that process already expired its cache locally.
+
+The topic and subscription must be declared in the `AddServiceBus` configuration so the initializer provisions them and registers a sender for the topic:
+
+```csharp
+services.AddServiceBus(
+    serviceName: "Default",
+    nameOrConnectionString: "ServiceBus",
+    configureBus: bus => bus.AddTopic("cache-expire", "app-instance"));
+```
+
+Three extension methods opt in to the feature:
+
+```csharp
+// Publisher only: expire local cache and publish expiration messages
+services.AddServiceBusCacheExpirePublisher("cache-expire");
+
+// Subscriber only: receive expiration messages and expire local cache
+services.AddServiceBusCacheExpireSubscriber(
+    serviceName: "Default",
+    topicName: "cache-expire",
+    subscriptionName: "app-instance");
+
+// Both publisher and subscriber
+services.AddServiceBusCacheExpire(
+    serviceName: "Default",
+    topicName: "cache-expire",
+    subscriptionName: "app-instance");
+```
+
+Each process should use a unique topic subscription so it receives its own copy of every expiration message. The per-process `SourceId` defaults to a unique value, but can be set explicitly through the optional `configureOptions` delegate:
+
+```csharp
+services.AddServiceBusCacheExpire(
+    serviceName: "Default",
+    topicName: "cache-expire",
+    subscriptionName: "app-instance",
+    configureOptions: options => options.SourceId = Environment.MachineName);
+```
+
+> [!NOTE]
+> Distributed expiration complements, and does not replace, the shared distributed cache tier (such as Redis). It is most useful for expiring the fast local L1 cache in each process when data changes elsewhere.
+
 ## Configuration Options
 
 ### Hybrid Cache Configuration

src/Arbiter.CommandQuery/Behaviors/HybridCacheExpireBehavior.cs

@@ -14,8 +14,6 @@ namespace Arbiter.CommandQuery.Behaviors;
 public class HybridCacheExpireBehavior<TRequest, TResponse> : PipelineBehaviorBase<TRequest, TResponse>
     where TRequest : class, IRequest<TResponse>, ICacheExpire
 {
-    private readonly HybridCache _hybridCache;
-
     /// <summary>
     /// Initializes a new instance of the <see cref="HybridCacheExpireBehavior{TRequest, TResponse}"/> class.
     /// </summary>
@@ -29,9 +27,14 @@ public HybridCacheExpireBehavior(
     {
         ArgumentNullException.ThrowIfNull(hybridCache);
 
-        _hybridCache = hybridCache;
+        HybridCache = hybridCache;
     }
 
+    /// <summary>
+    /// Gets the <see cref="HybridCache"/> used to expire cache entries.
+    /// </summary>
+    protected HybridCache HybridCache { get; }
+
     /// <inheritdoc />
     protected override async ValueTask<TResponse?> Process(
         TRequest request,
@@ -44,17 +47,31 @@ public HybridCacheExpireBehavior(
         var response = await next(cancellationToken).ConfigureAwait(false);
 
         // expire cache
-        if (request is not ICacheExpire cacheRequest)
-            return response;
+        if (request is ICacheExpire cacheRequest)
+            await ExpireCache(cacheRequest, cancellationToken).ConfigureAwait(false);
+
+        return response;
+    }
+
+    /// <summary>
+    /// Removes the cache key and cache tags of the specified <paramref name="cacheRequest"/> from the local <see cref="HybridCache"/>.
+    /// </summary>
+    /// <param name="cacheRequest">The request describing the cache key and tags to expire.</param>
+    /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
+    /// <returns>A <see cref="ValueTask"/> that represents the asynchronous expire operation.</returns>
+    /// <exception cref="ArgumentNullException">When <paramref name="cacheRequest"/> is null</exception>
+    protected async ValueTask ExpireCache(ICacheExpire cacheRequest, CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(cacheRequest);
 
         var cacheKey = cacheRequest.GetCacheKey();
         if (!string.IsNullOrEmpty(cacheKey))
-            await _hybridCache.RemoveAsync(cacheKey, cancellationToken).ConfigureAwait(false);
-
-        var cacheTag = cacheRequest.GetCacheTag();
-        if (!string.IsNullOrEmpty(cacheTag))
-            await _hybridCache.RemoveByTagAsync(cacheTag, cancellationToken).ConfigureAwait(false);
+            await HybridCache.RemoveAsync(cacheKey, cancellationToken).ConfigureAwait(false);
 
-        return response;
+        foreach (var cacheTag in cacheRequest.GetCacheTags())
+        {
+            if (!string.IsNullOrEmpty(cacheTag))
+                await HybridCache.RemoveByTagAsync(cacheTag, cancellationToken).ConfigureAwait(false);
+        }
     }
 }

src/Arbiter.CommandQuery/Definitions/ICacheExpire.cs

@@ -16,4 +16,22 @@ public interface ICacheExpire
     /// </summary>
     /// <returns>The cache tag for this entity</returns>
     string? GetCacheTag();
+
+    /// <summary>
+    /// Gets the cache tags for this entity.
+    /// </summary>
+    /// <returns>
+    /// The cache tags for this entity. The default implementation returns the single tag from
+    /// <see cref="GetCacheTag"/> as a one-item sequence, or an empty sequence when no tag is available.
+    /// </returns>
+    /// <remarks>
+    /// Override this member to expire multiple cache tags for a single change. The default implementation
+    /// preserves backward compatibility with <see cref="GetCacheTag"/>, so existing implementers do not
+    /// need to change.
+    /// </remarks>
+    IEnumerable<string> GetCacheTags()
+    {
+        var tag = GetCacheTag();
+        return string.IsNullOrEmpty(tag) ? [] : [tag];
+    }
 }

src/Arbiter.Messaging.ServiceBus/Cache/CacheExpireMessage.cs

@@ -0,0 +1,35 @@
+using System.Text.Json.Serialization;
+
+namespace Arbiter.Messaging.ServiceBus.Cache;
+
+/// <summary>
+/// A message describing a cache entry to expire across applications subscribed to a Service Bus topic.
+/// </summary>
+/// <remarks>
+/// The message is published when a command expires the local cache and is received by other applications so they can
+/// remove the matching key and tags from their own <see cref="Microsoft.Extensions.Caching.Hybrid.HybridCache"/>.
+/// </remarks>
+public sealed record CacheExpireMessage
+{
+    /// <summary>
+    /// Gets the cache key to remove, or <see langword="null"/> when no specific key should be removed.
+    /// </summary>
+    [JsonPropertyName("key")]
+    public string? Key { get; init; }
+
+    /// <summary>
+    /// Gets the cache tags to remove. Empty when no tags should be removed.
+    /// </summary>
+    [JsonPropertyName("tags")]
+    public IReadOnlyList<string> Tags { get; init; } = [];
+
+    /// <summary>
+    /// Gets the identifier of the application instance that published this message.
+    /// </summary>
+    /// <remarks>
+    /// Used by receivers to skip expiration for messages they originally published, since the publishing
+    /// application has already expired its own local cache.
+    /// </remarks>
+    [JsonPropertyName("sourceId")]
+    public string? SourceId { get; init; }
+}

src/Arbiter.Messaging.ServiceBus/Cache/CacheExpireOptions.cs

@@ -0,0 +1,21 @@
+namespace Arbiter.Messaging.ServiceBus.Cache;
+
+/// <summary>
+/// Options that control distributed cache expiration over Azure Service Bus.
+/// </summary>
+public sealed class CacheExpireOptions
+{
+    /// <summary>
+    /// Gets or sets the name of the Service Bus topic used to publish and receive cache expiration messages.
+    /// </summary>
+    public required string TopicName { get; set; }
+
+    /// <summary>
+    /// Gets or sets the identifier of this application instance.
+    /// </summary>
+    /// <remarks>
+    /// Stamped on outgoing messages and compared on incoming messages so an application does not redundantly
+    /// expire cache entries it already expired locally. Defaults to a unique value per process.
+    /// </remarks>
+    public string SourceId { get; set; } = Guid.NewGuid().ToString("N");
+}

src/Arbiter.Messaging.ServiceBus/Cache/CacheExpireProcessor.cs

@@ -0,0 +1,105 @@
+using System.Text.Json;
+
+using Azure.Messaging.ServiceBus;
+
+using Microsoft.Extensions.Caching.Hybrid;
+using Microsoft.Extensions.Logging;
+
+namespace Arbiter.Messaging.ServiceBus.Cache;
+
+/// <summary>
+/// Processes <see cref="CacheExpireMessage"/> messages from a Service Bus topic subscription and expires the
+/// matching entries from the local <see cref="HybridCache"/>.
+/// </summary>
+public sealed partial class CacheExpireProcessor : ServiceBusProcessorBase
+{
+    private readonly HybridCache _hybridCache;
+    private readonly CacheExpireOptions _options;
+    private readonly ILogger<CacheExpireProcessor> _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CacheExpireProcessor"/> class.
+    /// </summary>
+    /// <param name="processor">The Service Bus processor used to receive messages.</param>
+    /// <param name="logger">The logger used to write processing messages.</param>
+    /// <param name="hybridCache">The hybrid cache to expire.</param>
+    /// <param name="options">The cache expiration options.</param>
+    /// <exception cref="ArgumentNullException">When <paramref name="hybridCache"/> or <paramref name="options"/> is null.</exception>
+    public CacheExpireProcessor(
+        ServiceBusProcessor processor,
+        ILogger<CacheExpireProcessor> logger,
+        HybridCache hybridCache,
+        CacheExpireOptions options)
+        : base(processor, logger)
+    {
+        ArgumentNullException.ThrowIfNull(hybridCache);
+        ArgumentNullException.ThrowIfNull(options);
+
+        _hybridCache = hybridCache;
+        _options = options;
+        _logger = logger;
+    }
+
+    /// <inheritdoc />
+    /// <remarks>
+    /// Message settlement is handled automatically by the processor (<see cref="ServiceBusProcessorOptions.AutoCompleteMessages" />);
+    /// returning successfully completes the message, while throwing abandons it for redelivery. A message with an
+    /// unparsable body is non-retryable and is dead-lettered explicitly rather than retried until the delivery count is exhausted.
+    /// </remarks>
+    protected override async Task ProcessMessageAsync(ProcessMessageEventArgs args)
+    {
+        ArgumentNullException.ThrowIfNull(args);
+
+        CacheExpireMessage? message;
+
+        try
+        {
+            message = args.ReadFromJson<CacheExpireMessage>();
+        }
+        catch (JsonException ex)
+        {
+            // non-retryable: a malformed body never succeeds, so dead-letter immediately instead of
+            // abandoning until MaxDeliveryCount is reached
+            LogDeadLetteringMessage(_logger, ex, args.Message.MessageId);
+
+            await args.DeadLetterMessageAsync(
+                args.Message,
+                deadLetterReason: "DeserializationFailed",
+                deadLetterErrorDescription: ex.Message,
+                args.CancellationToken).ConfigureAwait(false);
+
+            return;
+        }
+
+        if (message is null)
+            return;
+
+        // skip messages this application published; it already expired its own local cache
+        if (!string.IsNullOrEmpty(message.SourceId)
+            && string.Equals(message.SourceId, _options.SourceId, StringComparison.Ordinal))
+        {
+            LogSkippingOwnMessage(_logger, message.SourceId);
+            return;
+        }
+
+        if (!string.IsNullOrEmpty(message.Key))
+            await _hybridCache.RemoveAsync(message.Key, args.CancellationToken).ConfigureAwait(false);
+
+        foreach (var tag in message.Tags)
+        {
+            if (!string.IsNullOrEmpty(tag))
+                await _hybridCache.RemoveByTagAsync(tag, args.CancellationToken).ConfigureAwait(false);
+        }
+
+        LogExpiredCache(_logger, message.Key, message.Tags.Count);
+    }
+
+    [LoggerMessage(Level = LogLevel.Debug, Message = "Skipping cache expiration message from own source '{SourceId}'")]
+    private static partial void LogSkippingOwnMessage(ILogger logger, string sourceId);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "Dead-lettering cache expiration message '{MessageId}' with unparsable body")]
+    private static partial void LogDeadLetteringMessage(ILogger logger, Exception exception, string messageId);
+
+    [LoggerMessage(Level = LogLevel.Debug, Message = "Expired local cache for key '{Key}' and {TagCount} tag(s)")]
+    private static partial void LogExpiredCache(ILogger logger, string? key, int tagCount);
+}