Multi-instance safety: Redis cache invalidation resilience & exactly-once scheduled tasks

Directory.Packages.props

@@ -65,6 +65,7 @@
     <PackageVersion Include="Verify.MSTest" Version="31.19.1" />
     <PackageVersion Include="Aspire.Hosting.AppHost" Version="13.4.3" />
     <PackageVersion Include="Aspire.Hosting.MongoDB" Version="13.4.3" />
+    <PackageVersion Include="Aspire.Hosting.Redis" Version="13.4.3" />
     <PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="10.7.0" />
     <PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="10.7.0" />
     <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.16.0" />

src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj

@@ -16,6 +16,7 @@
 	<ItemGroup>
 		<PackageReference Include="Aspire.Hosting.AppHost" />
 		<PackageReference Include="Aspire.Hosting.MongoDB" />
+		<PackageReference Include="Aspire.Hosting.Redis" />
 	</ItemGroup>
 	<ItemGroup>
 		<PackageReference Include="Aspire.Dashboard.Sdk.$(NETCoreSdkRuntimeIdentifier)" />

src/Aspire/Aspire.AppHost/Program.cs

@@ -4,6 +4,9 @@
 
 var mongo = builder.AddMongoDB("mongo").WithLifetime(ContainerLifetime.Persistent);
 var mongodb = mongo.AddDatabase("Mongodb");
-builder.ConfigureGrandWebProject(mongodb);
+
+var redis = builder.AddRedis("redis").WithLifetime(ContainerLifetime.Persistent);
+
+builder.ConfigureGrandWebProject(mongodb, redis);
 
 await builder.Build().RunAsync();

src/Aspire/Aspire.AppHost/ProjectConfiguration.cs

@@ -1,13 +1,34 @@
-namespace Aspire.AppHost;
+namespace Aspire.AppHost;
 
 public static class ProjectConfiguration
 {
-    public static void ConfigureGrandWebProject(this IDistributedApplicationBuilder builder, IResourceBuilder<MongoDBDatabaseResource> mongodb)
+    public static void ConfigureGrandWebProject(this IDistributedApplicationBuilder builder,
+        IResourceBuilder<MongoDBDatabaseResource> mongodb,
+        IResourceBuilder<RedisResource> redis)
+    {
+        //two instances of the same application (multi-pod simulation) sharing one database
+        //and synchronizing their memory cache through the Redis pub/sub message bus
+        builder.AddGrandWebInstance("grand-web", 80, mongodb, redis);
+        builder.AddGrandWebInstance("grand-web-2", 8080, mongodb, redis);
+    }
+
+    private static void AddGrandWebInstance(this IDistributedApplicationBuilder builder, string name, int port,
+        IResourceBuilder<MongoDBDatabaseResource> mongodb,
+        IResourceBuilder<RedisResource> redis)
     {
         builder
-            .AddProject<Projects.Grand_Web>("grand-web")
-            .WithHttpEndpoint(80, name: "front")
+            .AddProject<Projects.Grand_Web>(name)
+            .WithHttpEndpoint(port, name: "front")
             .WithReference(mongodb)
-            .WaitFor(mongodb);
+            .WaitFor(mongodb)
+            .WaitFor(redis)
+            //both instances share one content root - shadow copy would make them fight
+            //over the same Plugins/bin folder (files locked by the other instance)
+            .WithEnvironment("Extensions__PluginShadowCopy", "false")
+            .WithEnvironment("Redis__RedisPubSubEnabled", "true")
+            //show publish/receive debug logs of the cache message bus in the dashboard
+            .WithEnvironment("Logging__LogLevel__Grand.Infrastructure.Caching.Redis", "Debug")
+            .WithEnvironment("Redis__RedisPubSubChannel", "grandnode-cache")
+            .WithEnvironment("Redis__RedisPubSubConnectionString", redis.Resource.ConnectionStringExpression);
     }
-}
+}

src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs

@@ -36,6 +36,18 @@ public interface IScheduleTaskService
     /// <param name="task">Task</param>
     Task UpdateTask(ScheduleTask task);
 
+    /// <summary>
+    ///     Atomically claims a single run of the task, so that only one application instance
+    ///     executes it. The claim succeeds only if LastStartUtc still has the value the caller
+    ///     read (compare-and-set) - a concurrent claim by another instance makes it fail.
+    /// </summary>
+    /// <param name="taskId">Task identifier</param>
+    /// <param name="expectedLastStartUtc">LastStartUtc value read before the claim</param>
+    /// <param name="runStartUtc">New LastStartUtc value (start of the claimed run)</param>
+    /// <param name="instanceId">Unique identifier of the claiming application instance</param>
+    /// <returns>true if this instance won the claim and should execute the task</returns>
+    Task<bool> TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, string instanceId);
+
     /// <summary>
     ///     Delete the task
     /// </summary>

src/Core/Grand.Domain/Tasks/ScheduleTask.cs

@@ -10,6 +10,7 @@ public class ScheduleTask : BaseEntity
     public DateTime? LastSuccessUtc { get; set; }
     public string LeasedByMachineName { get; set; }
     public DateTime? LeasedUntilUtc { get; set; }
+    public string LeasedByInstance { get; set; }
     public int TimeInterval { get; set; }
     public string StoreId { get; set; }
 }

src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageBus.cs

@@ -1,66 +1,67 @@
-using Grand.Infrastructure.Caching.Message;
+using Grand.Infrastructure.Caching.Message;
 using Grand.Infrastructure.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using StackExchange.Redis;
-using System.Diagnostics;
 using System.Text.Json;
 
 namespace Grand.Infrastructure.Caching.Redis;
 
-public sealed class RedisMessageBus : IMessageBus
+public sealed class RedisMessageBus : IMessageBus, IDisposable
 {
     private static readonly string ClientId = Guid.NewGuid().ToString("N");
+
+    private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromSeconds(1);
+    private static readonly TimeSpan MaxRetryDelay = TimeSpan.FromSeconds(30);
+
+    private readonly IConnectionMultiplexer _connection;
     private readonly RedisConfig _redisConfig;
     private readonly IServiceProvider _serviceProvider;
     private readonly ISubscriber _subscriber;
+    private readonly ILogger<RedisMessageBus> _logger;
+    private readonly CancellationTokenSource _cts = new();
 
-    public RedisMessageBus(ISubscriber subscriber, IServiceProvider serviceProvider, RedisConfig redisConfig)
+    public RedisMessageBus(IConnectionMultiplexer connection, IServiceProvider serviceProvider,
+        RedisConfig redisConfig, ILogger<RedisMessageBus> logger)
     {
-        _subscriber = subscriber;
+        _connection = connection;
+        _subscriber = connection.GetSubscriber();
         _serviceProvider = serviceProvider;
         _redisConfig = redisConfig;
-        SubscribeAsync();
+        _logger = logger;
+
+        _connection.ConnectionFailed += OnConnectionFailed;
+        _connection.ConnectionRestored += OnConnectionRestored;
+
+        _ = Task.Run(() => SubscribeWithRetryAsync(_cts.Token));
     }
 
     public async Task PublishAsync<TMessage>(TMessage msg) where TMessage : IMessageEvent
     {
+        var message = JsonSerializer.Serialize(new MessageEventClient {
+            ClientId = ClientId,
+            Key = msg.Key,
+            MessageType = msg.MessageType
+        });
         try
         {
-            var client = new MessageEventClient {
-                ClientId = ClientId,
-                Key = msg.Key,
-                MessageType = msg.MessageType
-            };
-            var message = JsonSerializer.Serialize(client);
-            await _subscriber.PublishAsync(RedisChannel.Literal(_redisConfig.RedisPubSubChannel), message);
+            var receivers =
+                await _subscriber.PublishAsync(RedisChannel.Literal(_redisConfig.RedisPubSubChannel), message);
+            _logger.LogDebug(
+                "Published cache invalidation message (type: {MessageType}, key: {Key}), delivered to {Receivers} subscriber(s)",
+                (MessageEventType)msg.MessageType, msg.Key, receivers);
         }
         catch (Exception ex)
         {
-            Debug.WriteLine(ex.Message);
+            _logger.LogError(ex,
+                "Failed to publish cache invalidation message (type: {MessageType}, key: {Key}) - other instances may serve stale data until cache expiration",
+                (MessageEventType)msg.MessageType, msg.Key);
         }
     }
 
     public Task SubscribeAsync()
     {
-        _ = _subscriber.SubscribeAsync(RedisChannel.Literal(_redisConfig.RedisPubSubChannel), (_, redisValue) =>
-        {
-            try
-            {
-                MessageEventClient message = null;
-                if (!redisValue.IsNull)
-                {
-                    // Use the string overload explicitly to resolve ambiguity
-                    message = JsonSerializer.Deserialize<MessageEventClient>(redisValue.ToString());
-                }
-                if (message != null && message.ClientId != ClientId)
-                    OnSubscriptionChanged(message);
-            }
-            catch (Exception ex)
-            {
-                Debug.WriteLine(ex.Message);
-            }
-        });
-        return Task.CompletedTask;
+        return _subscriber.SubscribeAsync(RedisChannel.Literal(_redisConfig.RedisPubSubChannel), OnMessage);
     }
 
     public void OnSubscriptionChanged(IMessageEvent message)
@@ -80,4 +81,88 @@
                 break;
         }
     }
-}
+
+    public void Dispose()
+    {
+        _connection.ConnectionFailed -= OnConnectionFailed;
+        _connection.ConnectionRestored -= OnConnectionRestored;
+        _cts.Cancel();
+        _cts.Dispose();
+    }
+
+    private async Task SubscribeWithRetryAsync(CancellationToken cancellationToken)
+    {
+        var delay = InitialRetryDelay;
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                await SubscribeAsync();
+                _logger.LogInformation("Subscribed to Redis pub/sub channel {Channel}",
+                    _redisConfig.RedisPubSubChannel);
+                return;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex,
+                    "Failed to subscribe to Redis pub/sub channel {Channel}, retrying in {Delay}s",
+                    _redisConfig.RedisPubSubChannel, delay.TotalSeconds);
+                try
+                {
+                    await Task.Delay(delay, cancellationToken);
+                }
+                catch (OperationCanceledException)
+                {
+                    return;
+                }
+                delay = delay * 2 > MaxRetryDelay ? MaxRetryDelay : delay * 2;
+            }
+        }
+    }
+
+    private void OnMessage(RedisChannel channel, RedisValue redisValue)
+    {
+        try
+        {
+            if (redisValue.IsNull) return;
+            var message = JsonSerializer.Deserialize<MessageEventClient>(redisValue.ToString());
+            if (message != null && message.ClientId != ClientId)
+            {
+                _logger.LogDebug(
+                    "Received cache invalidation message (type: {MessageType}, key: {Key}) from client {ClientId}",
+                    (MessageEventType)message.MessageType, message.Key, message.ClientId);
+                OnSubscriptionChanged(message);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Failed to process cache invalidation message from Redis");
+        }
+    }
+
+    private void OnConnectionFailed(object sender, ConnectionFailedEventArgs e)
+    {
+        _logger.LogWarning(e.Exception,
+            "Redis connection failed ({ConnectionType}, {FailureType}) - cache invalidation messages may be lost while disconnected",
+            e.ConnectionType, e.FailureType);
+    }
+
+    private void OnConnectionRestored(object sender, ConnectionFailedEventArgs e)
+    {
+        //invalidation messages published while this instance was disconnected are lost
+        //(Redis pub/sub has no replay), so the local cache can no longer be trusted
+        if (e.ConnectionType != ConnectionType.Subscription) return;
+        _logger.LogWarning(
+            "Redis subscription connection restored - clearing local cache to drop entries that may have missed invalidation");
+        try
+        {
+            using var scope = _serviceProvider.CreateScope();
+            var cache = scope.ServiceProvider.GetRequiredService<ICacheBase>();
+            _ = cache.Clear(false);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Failed to clear local cache after Redis connection was restored");
+        }
+    }
+}

src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageCacheManager.cs

@@ -1,4 +1,4 @@
-using Grand.Infrastructure.Caching.Message;
+using Grand.Infrastructure.Caching.Message;
 using Grand.Infrastructure.Configuration;
 using MediatR;
 using Microsoft.Extensions.Caching.Memory;
@@ -7,13 +7,11 @@ namespace Grand.Infrastructure.Caching.Redis;
 
 public class RedisMessageCacheManager : MemoryCacheBase, ICacheBase
 {
-    private readonly IMemoryCache _cache;
     private readonly IMessageBus _messageBus;
 
     public RedisMessageCacheManager(IMemoryCache cache, IMediator mediator, IMessageBus messageBus, CacheConfig config)
         : base(cache, mediator, config)
     {
-        _cache = cache;
         _messageBus = messageBus;
     }
 
@@ -22,31 +20,27 @@ public RedisMessageCacheManager(IMemoryCache cache, IMediator mediator, IMessage
     /// </summary>
     /// <param name="key">Key of cached item</param>
     /// <param name="publisher">Publisher</param>
-    public override Task RemoveAsync(string key, bool publisher = true)
+    public override async Task RemoveAsync(string key, bool publisher = true)
     {
-        _cache.Remove(key);
+        await base.RemoveAsync(key, false);
 
         if (publisher)
-            _messageBus.PublishAsync(new MessageEvent { Key = key, MessageType = (int)MessageEventType.RemoveKey });
-
-        return Task.CompletedTask;
+            await _messageBus.PublishAsync(new MessageEvent
+                { Key = key, MessageType = (int)MessageEventType.RemoveKey });
     }
 
     /// <summary>
     ///     Removes items by key prefix
     /// </summary>
     /// <param name="prefix">String prefix</param>
     /// <param name="publisher">publisher</param>
-    public override Task RemoveByPrefix(string prefix, bool publisher = true)
+    public override async Task RemoveByPrefix(string prefix, bool publisher = true)
     {
-        var entriesToRemove = CacheEntries.Where(x => x.Key.StartsWith(prefix, StringComparison.OrdinalIgnoreCase));
-        foreach (var cacheEntries in entriesToRemove) _cache.Remove(cacheEntries.Key);
+        await base.RemoveByPrefix(prefix, false);
 
         if (publisher)
-            _messageBus.PublishAsync(new MessageEvent
+            await _messageBus.PublishAsync(new MessageEvent
                 { Key = prefix, MessageType = (int)MessageEventType.RemoveByPrefix });
-
-        return Task.CompletedTask;
     }
 
     /// <summary>
@@ -56,8 +50,9 @@ public override Task RemoveByPrefix(string prefix, bool publisher = true)
     public override async Task Clear(bool publisher = true)
     {
         await base.Clear(publisher);
-        if (publisher)
-            await _messageBus.PublishAsync(new MessageEvent { Key = "", MessageType = (int)MessageEventType.ClearCache });
 
+        if (publisher)
+            await _messageBus.PublishAsync(new MessageEvent
+                { Key = "", MessageType = (int)MessageEventType.ClearCache });
     }
-}
+}

src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs

@@ -78,6 +78,29 @@ public virtual async Task UpdateTask(ScheduleTask task)
         await _taskRepository.UpdateAsync(task);
     }
 
+    /// <summary>
+    ///     Atomically claims a single run of the task (compare-and-set on LastStartUtc)
+    /// </summary>
+    public virtual async Task<bool> TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc,
+        DateTime runStartUtc, string instanceId)
+    {
+        //truncate to milliseconds so the value survives the database round-trip unchanged
+        runStartUtc = new DateTime(runStartUtc.Ticks - runStartUtc.Ticks % TimeSpan.TicksPerMillisecond,
+            DateTimeKind.Utc);
+
+        //conditional update - matches only when no other instance has claimed the run
+        //in the meantime; on MongoDB a single update is atomic, so exactly one instance wins
+        await _taskRepository.UpdateOneAsync(
+            x => x.Id == taskId && x.LastStartUtc == expectedLastStartUtc,
+            UpdateBuilder<ScheduleTask>.Create()
+                .Set(x => x.LastStartUtc, runStartUtc)
+                .Set(x => x.LeasedByInstance, instanceId));
+
+        //UpdateOneAsync does not report whether the filter matched - read back the winner
+        var task = await GetTaskById(taskId);
+        return task != null && task.LeasedByInstance == instanceId && task.LastStartUtc == runStartUtc;
+    }
+
     /// <summary>
     ///     Delete the task
     /// </summary>