diff --git a/Directory.Packages.props b/Directory.Packages.props index 8b8e6d5e6..a9f87a784 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -65,6 +65,7 @@ + diff --git a/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj b/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj index 550a7d6b8..aef766843 100644 --- a/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj +++ b/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj @@ -16,6 +16,7 @@ + diff --git a/src/Aspire/Aspire.AppHost/Program.cs b/src/Aspire/Aspire.AppHost/Program.cs index a575b90d6..efb79fa91 100644 --- a/src/Aspire/Aspire.AppHost/Program.cs +++ b/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(); diff --git a/src/Aspire/Aspire.AppHost/ProjectConfiguration.cs b/src/Aspire/Aspire.AppHost/ProjectConfiguration.cs index 5f8053133..336c30f5c 100644 --- a/src/Aspire/Aspire.AppHost/ProjectConfiguration.cs +++ b/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 mongodb) + public static void ConfigureGrandWebProject(this IDistributedApplicationBuilder builder, + IResourceBuilder mongodb, + IResourceBuilder 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 mongodb, + IResourceBuilder redis) { builder - .AddProject("grand-web") - .WithHttpEndpoint(80, name: "front") + .AddProject(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); } -} \ No newline at end of file +} diff --git a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs index 89f7d02d9..8c78591b9 100644 --- a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs +++ b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs @@ -36,6 +36,18 @@ public interface IScheduleTaskService /// Task Task UpdateTask(ScheduleTask task); + /// + /// 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. + /// + /// Task identifier + /// LastStartUtc value read before the claim + /// New LastStartUtc value (start of the claimed run) + /// Unique identifier of the claiming application instance + /// true if this instance won the claim and should execute the task + Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, string instanceId); + /// /// Delete the task /// diff --git a/src/Core/Grand.Domain/Tasks/ScheduleTask.cs b/src/Core/Grand.Domain/Tasks/ScheduleTask.cs index d19979140..6d773f6f6 100644 --- a/src/Core/Grand.Domain/Tasks/ScheduleTask.cs +++ b/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; } } \ No newline at end of file diff --git a/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageBus.cs b/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageBus.cs index 7117afb2c..17bf89f5c 100644 --- a/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageBus.cs +++ b/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 _logger; + private readonly CancellationTokenSource _cts = new(); - public RedisMessageBus(ISubscriber subscriber, IServiceProvider serviceProvider, RedisConfig redisConfig) + public RedisMessageBus(IConnectionMultiplexer connection, IServiceProvider serviceProvider, + RedisConfig redisConfig, ILogger 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 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(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 @@ public void OnSubscriptionChanged(IMessageEvent message) break; } } -} \ No newline at end of file + + 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(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(); + _ = cache.Clear(false); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to clear local cache after Redis connection was restored"); + } + } +} diff --git a/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageCacheManager.cs b/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageCacheManager.cs index 6fc492267..954162361 100644 --- a/src/Core/Grand.Infrastructure/Caching/Redis/RedisMessageCacheManager.cs +++ b/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,14 +20,13 @@ public RedisMessageCacheManager(IMemoryCache cache, IMediator mediator, IMessage /// /// Key of cached item /// Publisher - 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 }); } /// @@ -37,16 +34,13 @@ public override Task RemoveAsync(string key, bool publisher = true) /// /// String prefix /// publisher - 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; } /// @@ -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 }); } -} \ No newline at end of file +} diff --git a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs index c17036276..bc4670bb4 100644 --- a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs +++ b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs @@ -78,6 +78,29 @@ public virtual async Task UpdateTask(ScheduleTask task) await _taskRepository.UpdateAsync(task); } + /// + /// Atomically claims a single run of the task (compare-and-set on LastStartUtc) + /// + public virtual async Task 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.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; + } + /// /// Delete the task /// diff --git a/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageBusTests.cs b/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageBusTests.cs new file mode 100644 index 000000000..a10674633 --- /dev/null +++ b/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageBusTests.cs @@ -0,0 +1,156 @@ +using Grand.Infrastructure.Caching; +using Grand.Infrastructure.Caching.Message; +using Grand.Infrastructure.Caching.Redis; +using Grand.Infrastructure.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Moq; +using StackExchange.Redis; +using System.Net; +using System.Text.Json; + +namespace Grand.Infrastructure.Tests.Caching.Redis; + +[TestClass] +public class RedisMessageBusTests +{ + private Mock _connectionMock; + private Mock _subscriberMock; + private Mock _cacheMock; + private RedisConfig _config; + private RedisMessageBus _bus; + + [TestInitialize] + public void Init() + { + _subscriberMock = new Mock(); + _subscriberMock.Setup(s => s.SubscribeAsync(It.IsAny(), + It.IsAny>(), It.IsAny())) + .Returns(Task.CompletedTask); + + _connectionMock = new Mock(); + _connectionMock.Setup(c => c.GetSubscriber(It.IsAny())).Returns(_subscriberMock.Object); + + _cacheMock = new Mock(); + _cacheMock.Setup(c => c.RemoveAsync(It.IsAny(), It.IsAny())).Returns(Task.CompletedTask); + _cacheMock.Setup(c => c.RemoveByPrefix(It.IsAny(), It.IsAny())).Returns(Task.CompletedTask); + _cacheMock.Setup(c => c.Clear(It.IsAny())).Returns(Task.CompletedTask); + + var services = new ServiceCollection(); + services.AddSingleton(_cacheMock.Object); + var serviceProvider = services.BuildServiceProvider(); + + _config = new RedisConfig { RedisPubSubChannel = "test-channel" }; + _bus = new RedisMessageBus(_connectionMock.Object, serviceProvider, _config, + new Mock>().Object); + } + + [TestCleanup] + public void Cleanup() + { + _bus.Dispose(); + } + + [TestMethod] + public async Task Constructor_SubscribesToConfiguredChannel() + { + //subscription starts on a background task - wait for it + for (var i = 0; i < 100; i++) + { + try + { + _subscriberMock.Verify(s => s.SubscribeAsync( + It.Is(c => c == RedisChannel.Literal("test-channel")), + It.IsAny>(), It.IsAny()), Times.AtLeastOnce); + return; + } + catch (MockException) + { + await Task.Delay(20); + } + } + + Assert.Fail("SubscribeAsync was not called on the configured channel"); + } + + [TestMethod] + public async Task PublishAsync_SendsSerializedMessageToConfiguredChannel() + { + RedisValue publishedValue = default; + _subscriberMock.Setup(s => + s.PublishAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((_, value, _) => publishedValue = value) + .ReturnsAsync(1); + + await _bus.PublishAsync(new MessageEvent { Key = "key", MessageType = (int)MessageEventType.RemoveKey }); + + _subscriberMock.Verify(s => s.PublishAsync( + It.Is(c => c == RedisChannel.Literal("test-channel")), + It.IsAny(), It.IsAny()), Times.Once); + + var message = JsonSerializer.Deserialize(publishedValue.ToString()); + Assert.IsNotNull(message); + Assert.AreEqual("key", message.Key); + Assert.AreEqual((int)MessageEventType.RemoveKey, message.MessageType); + Assert.IsFalse(string.IsNullOrEmpty(message.ClientId)); + } + + [TestMethod] + public async Task PublishAsync_SubscriberThrows_DoesNotPropagateException() + { + _subscriberMock.Setup(s => + s.PublishAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ThrowsAsync(new RedisConnectionException(ConnectionFailureType.SocketFailure, "connection lost")); + + await _bus.PublishAsync(new MessageEvent { Key = "key", MessageType = (int)MessageEventType.RemoveKey }); + } + + [TestMethod] + public void OnSubscriptionChanged_RemoveKey_RemovesFromLocalCacheWithoutRepublishing() + { + _bus.OnSubscriptionChanged(new MessageEventClient + { ClientId = "other", Key = "key", MessageType = (int)MessageEventType.RemoveKey }); + + _cacheMock.Verify(c => c.RemoveAsync("key", false), Times.Once); + } + + [TestMethod] + public void OnSubscriptionChanged_RemoveByPrefix_RemovesFromLocalCacheWithoutRepublishing() + { + _bus.OnSubscriptionChanged(new MessageEventClient + { ClientId = "other", Key = "prefix", MessageType = (int)MessageEventType.RemoveByPrefix }); + + _cacheMock.Verify(c => c.RemoveByPrefix("prefix", false), Times.Once); + } + + [TestMethod] + public void OnSubscriptionChanged_ClearCache_ClearsLocalCacheWithoutRepublishing() + { + _bus.OnSubscriptionChanged(new MessageEventClient + { ClientId = "other", Key = "", MessageType = (int)MessageEventType.ClearCache }); + + _cacheMock.Verify(c => c.Clear(false), Times.Once); + } + + [TestMethod] + public void ConnectionRestored_SubscriptionConnection_ClearsLocalCache() + { + //messages published while disconnected are lost - the local cache must be dropped + _connectionMock.Raise(c => c.ConnectionRestored += null, + new ConnectionFailedEventArgs(_connectionMock.Object, new DnsEndPoint("localhost", 6379), + ConnectionType.Subscription, ConnectionFailureType.SocketClosed, null, "test")); + + _cacheMock.Verify(c => c.Clear(false), Times.Once); + } + + [TestMethod] + public void ConnectionRestored_InteractiveConnection_DoesNotClearLocalCache() + { + _connectionMock.Raise(c => c.ConnectionRestored += null, + new ConnectionFailedEventArgs(_connectionMock.Object, new DnsEndPoint("localhost", 6379), + ConnectionType.Interactive, ConnectionFailureType.SocketClosed, null, "test")); + + _cacheMock.Verify(c => c.Clear(It.IsAny()), Times.Never); + } +} diff --git a/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageCacheManagerTests.cs b/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageCacheManagerTests.cs new file mode 100644 index 000000000..cc2c7b064 --- /dev/null +++ b/src/Tests/Grand.Infrastructure.Tests/Caching/Redis/RedisMessageCacheManagerTests.cs @@ -0,0 +1,102 @@ +using Grand.Infrastructure.Caching.Message; +using Grand.Infrastructure.Caching.Redis; +using Grand.Infrastructure.Configuration; +using MediatR; +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Moq; + +namespace Grand.Infrastructure.Tests.Caching.Redis; + +[TestClass] +public class RedisMessageCacheManagerTests +{ + private Mock _messageBusMock; + private IMemoryCache _memoryCache; + private RedisMessageCacheManager _service; + + [TestInitialize] + public void Init() + { + var services = new ServiceCollection(); + services.AddMemoryCache(); + _memoryCache = services.BuildServiceProvider().GetService(); + _messageBusMock = new Mock(); + _service = new RedisMessageCacheManager(_memoryCache, new Mock().Object, + _messageBusMock.Object, new CacheConfig { DefaultCacheTimeMinutes = 1 }); + } + + [TestMethod] + public async Task RemoveAsync_RemovesKeyAndPublishesMessage() + { + _service.Get("key", () => "value"); + + await _service.RemoveAsync("key"); + + Assert.IsFalse(_memoryCache.TryGetValue("key", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.Is(e => + e.Key == "key" && e.MessageType == (int)MessageEventType.RemoveKey)), Times.Once); + } + + [TestMethod] + public async Task RemoveAsync_PublisherDisabled_DoesNotPublish() + { + _service.Get("key", () => "value"); + + await _service.RemoveAsync("key", false); + + Assert.IsFalse(_memoryCache.TryGetValue("key", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.IsAny()), Times.Never); + } + + [TestMethod] + public async Task RemoveByPrefix_RemovesMatchingKeysAndPublishesMessage() + { + _service.Get("prefix-1", () => "value"); + _service.Get("prefix-2", () => "value"); + _service.Get("other", () => "value"); + + await _service.RemoveByPrefix("prefix"); + + Assert.IsFalse(_memoryCache.TryGetValue("prefix-1", out _)); + Assert.IsFalse(_memoryCache.TryGetValue("prefix-2", out _)); + Assert.IsTrue(_memoryCache.TryGetValue("other", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.Is(e => + e.Key == "prefix" && e.MessageType == (int)MessageEventType.RemoveByPrefix)), Times.Once); + } + + [TestMethod] + public async Task RemoveByPrefix_PublisherDisabled_DoesNotPublish() + { + _service.Get("prefix-1", () => "value"); + + await _service.RemoveByPrefix("prefix", false); + + Assert.IsFalse(_memoryCache.TryGetValue("prefix-1", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.IsAny()), Times.Never); + } + + [TestMethod] + public async Task Clear_PublishesClearCacheMessage() + { + _service.Get("key", () => "value"); + + await _service.Clear(); + + Assert.IsFalse(_memoryCache.TryGetValue("key", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.Is(e => + e.MessageType == (int)MessageEventType.ClearCache)), Times.Once); + } + + [TestMethod] + public async Task Clear_PublisherDisabled_DoesNotPublish() + { + _service.Get("key", () => "value"); + + await _service.Clear(false); + + Assert.IsFalse(_memoryCache.TryGetValue("key", out _)); + _messageBusMock.Verify(m => m.PublishAsync(It.IsAny()), Times.Never); + } +} diff --git a/src/Tests/Grand.Modules.Tests/Grand.Modules.Tests.csproj b/src/Tests/Grand.Modules.Tests/Grand.Modules.Tests.csproj index aa8e24bc4..dc6e1d9a2 100644 --- a/src/Tests/Grand.Modules.Tests/Grand.Modules.Tests.csproj +++ b/src/Tests/Grand.Modules.Tests/Grand.Modules.Tests.csproj @@ -21,6 +21,7 @@ + diff --git a/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs new file mode 100644 index 000000000..7663112f3 --- /dev/null +++ b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs @@ -0,0 +1,83 @@ +using Grand.Data.Tests.MongoDb; +using Grand.Domain.Tasks; +using Grand.Module.ScheduledTasks.BackgroundServices; +using Microsoft.VisualStudio.TestTools.UnitTesting; + +namespace Grand.Modules.Tests.Services.BackgroundService; + +[TestClass] +public class ScheduleTaskServiceTests +{ + private MongoDBRepositoryTest _repository; + private ScheduleTaskService _service; + + [TestInitialize] + public void Init() + { + _repository = new MongoDBRepositoryTest(); + _service = new ScheduleTaskService(_repository); + } + + private async Task InsertTask(DateTime? lastStartUtc = null) + { + var task = new ScheduleTask { + ScheduleTaskName = "TestTask", + Enabled = true, + TimeInterval = 1, + LastStartUtc = lastStartUtc + }; + await _service.InsertTask(task); + //read back so LastStartUtc has the value as stored in the database (ms precision) + return await _service.GetTaskById(task.Id); + } + + [TestMethod] + public async Task TryClaimTaskRun_FirstClaim_ReturnsTrueAndSetsLease() + { + var task = await InsertTask(); + var runStartUtc = DateTime.UtcNow; + + var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, runStartUtc, "instance-a"); + + Assert.IsTrue(claimed); + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-a", updated.LeasedByInstance); + Assert.IsNotNull(updated.LastStartUtc); + } + + [TestMethod] + public async Task TryClaimTaskRun_StaleExpectedValue_ReturnsFalseAndKeepsWinner() + { + var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10)); + + //instance A claims the run + var claimedByA = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-a"); + //instance B still holds the old LastStartUtc value - its claim must fail + var claimedByB = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-b"); + + Assert.IsTrue(claimedByA); + Assert.IsFalse(claimedByB); + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-a", updated.LeasedByInstance); + } + + [TestMethod] + public async Task TryClaimTaskRun_ConcurrentClaims_ExactlyOneWins() + { + var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10)); + + var claims = await Task.WhenAll( + _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-1"), + _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-2")); + + Assert.AreEqual(1, claims.Count(x => x)); + } + + [TestMethod] + public async Task TryClaimTaskRun_NonExistingTask_ReturnsFalse() + { + var claimed = await _service.TryClaimTaskRun("000000000000000000000000", null, DateTime.UtcNow, "instance-a"); + + Assert.IsFalse(claimed); + } +} diff --git a/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs new file mode 100644 index 000000000..eba0394a8 --- /dev/null +++ b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs @@ -0,0 +1,145 @@ +using Grand.Business.Core.Interfaces.System.ScheduleTasks; +using Grand.Domain.Tasks; +using Grand.Infrastructure; +using Grand.Web.Common.Infrastructure; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Moq; + +namespace Grand.Web.Common.Tests.Infrastructure; + +[TestClass] +public class BackgroundServiceTaskTests +{ + private const string TaskName = "TestTask"; + + private Mock _scheduleTaskServiceMock; + private Mock _scheduleTaskMock; + private ScheduleTask _task; + private IServiceProvider _serviceProvider; + + [TestInitialize] + public void Init() + { + _task = new ScheduleTask { + Id = "1", + ScheduleTaskName = TaskName, + Enabled = true, + TimeInterval = 60 + }; + + _scheduleTaskServiceMock = new Mock(); + _scheduleTaskServiceMock.Setup(s => s.GetTaskByName(TaskName)).ReturnsAsync(_task); + _scheduleTaskServiceMock.Setup(s => s.UpdateTask(It.IsAny())) + .Returns(Task.CompletedTask); + + _scheduleTaskMock = new Mock(); + _scheduleTaskMock.Setup(t => t.Execute()).Returns(Task.CompletedTask); + + var storeContextSetter = new Mock(); + storeContextSetter.Setup(s => s.InitializeStoreContext(It.IsAny())) + .ReturnsAsync(Mock.Of()); + var workContextSetter = new Mock(); + workContextSetter.Setup(s => s.InitializeWorkContext(It.IsAny())) + .ReturnsAsync(Mock.Of()); + + var services = new ServiceCollection(); + services.AddLogging(); + services.AddSingleton(_scheduleTaskServiceMock.Object); + services.AddKeyedSingleton(TaskName, _scheduleTaskMock.Object); + services.AddSingleton(Mock.Of()); + services.AddSingleton(storeContextSetter.Object); + services.AddSingleton(workContextSetter.Object); + _serviceProvider = services.BuildServiceProvider(); + } + + [TestMethod] + public async Task ExecuteAsync_ClaimWon_ExecutesTaskAndPersistsResult() + { + _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.UpdateTask))); + cts.Cancel(); + + _scheduleTaskMock.Verify(t => t.Execute(), Times.Once); + _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.Is(x => + x.LastSuccessUtc != null && x.LastStartUtc != null)), Times.Once); + } + + [TestMethod] + public async Task ExecuteAsync_ClaimLost_DoesNotExecuteAndDoesNotPersist() + { + _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(false); + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.TryClaimTaskRun))); + //give the loop a moment to (incorrectly) execute the task if the guard is broken + await Task.Delay(200); + cts.Cancel(); + + _scheduleTaskMock.Verify(t => t.Execute(), Times.Never); + //losing instance must not overwrite the winner's data with a stale entity + _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.IsAny()), Times.Never); + } + + [TestMethod] + public async Task ExecuteAsync_TaskNotDueYet_DoesNotClaimNorExecute() + { + _task.LastStartUtc = DateTime.UtcNow; + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.GetTaskByName))); + await Task.Delay(200); + cts.Cancel(); + + _scheduleTaskServiceMock.Verify(s => s.TryClaimTaskRun(It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny()), Times.Never); + _scheduleTaskMock.Verify(t => t.Execute(), Times.Never); + _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.IsAny()), Times.Never); + } + + [TestMethod] + public async Task ExecuteAsync_TaskLeasedByOtherMachine_DoesNotExecute() + { + _task.LeasedByMachineName = "other-machine"; + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.GetTaskByName))); + await Task.Delay(200); + cts.Cancel(); + + _scheduleTaskMock.Verify(t => t.Execute(), Times.Never); + } + + private static async Task WaitFor(Func condition) + { + for (var i = 0; i < 300; i++) + { + if (condition()) return; + await Task.Delay(10); + } + + Assert.Fail("Condition was not met within the timeout"); + } +} diff --git a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs index dcc0e885e..9ff473bf8 100644 --- a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs +++ b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs @@ -9,6 +9,10 @@ namespace Grand.Web.Common.Infrastructure; public class BackgroundServiceTask : BackgroundService { + //unique per process - pod/machine names are not stable across scaling, so a random + //id generated at startup is all that is needed to tell instances apart + private static readonly string InstanceId = Guid.NewGuid().ToString("N"); + private readonly IServiceProvider _serviceProvider; private readonly string Name; @@ -40,6 +44,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) machineName == task.LeasedByMachineName)) { + var updateTask = false; var scheduleTask = serviceProvider.GetRequiredKeyedService(task.ScheduleTaskName); if (scheduleTask != null) { @@ -65,32 +70,56 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) if (runTask) { - task.LastStartUtc = DateTime.UtcNow; - try - { - logger.LogInformation($"Task {Name} execute"); - await scheduleTask.Execute(); - task.LastSuccessUtc = DateTime.UtcNow; - task.LastNonSuccessEndUtc = null; + //claim this run atomically - when several instances race, + //only one wins and executes the task (no duplicated e-mails etc.) + var runStartUtc = DateTime.UtcNow; + var claimed = await scheduleTaskService.TryClaimTaskRun(task.Id, task.LastStartUtc, + runStartUtc, InstanceId); + if (claimed) + { + updateTask = true; + task.LastStartUtc = runStartUtc; + task.LeasedByInstance = InstanceId; + try + { + logger.LogInformation($"Task {Name} execute"); + await scheduleTask.Execute(); + task.LastSuccessUtc = DateTime.UtcNow; + task.LastNonSuccessEndUtc = null; + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + //application shutdown - do not classify as a task failure + throw; + } + catch (Exception exc) + { + task.LastNonSuccessEndUtc = DateTime.UtcNow; + task.Enabled = !task.StopOnError; + logger.LogError(exc, + "Error while running the \'{TaskScheduleTaskName}\' schedule task", + task.ScheduleTaskName); + } } - catch (Exception exc) + else { - task.LastNonSuccessEndUtc = DateTime.UtcNow; - task.Enabled = !task.StopOnError; - logger.LogError(exc, - "Error while running the \'{TaskScheduleTaskName}\' schedule task", - task.ScheduleTaskName); + //another instance executes this run - check again on the next interval + logger.LogDebug("Task {TaskName} claimed by another instance, skipping", Name); } } } else { + updateTask = true; task.Enabled = !task.StopOnError; task.LastNonSuccessEndUtc = DateTime.UtcNow; logger.LogError("Type {TaskName} is not registered", Name); } - await scheduleTaskService.UpdateTask(task); + //persist only when this instance actually ran the task - an unconditional + //write would overwrite the claim/results of the winning instance with stale data + if (updateTask) + await scheduleTaskService.UpdateTask(task); await Task.Delay(TimeSpan.FromMinutes(timeInterval), stoppingToken); } else diff --git a/src/Web/Grand.Web.Common/Startup/StartupApplication.cs b/src/Web/Grand.Web.Common/Startup/StartupApplication.cs index 943280079..c4bcf966b 100644 --- a/src/Web/Grand.Web.Common/Startup/StartupApplication.cs +++ b/src/Web/Grand.Web.Common/Startup/StartupApplication.cs @@ -60,16 +60,20 @@ private static void RegisterCache(IServiceCollection serviceCollection, IConfigu var config = new RedisConfig(); configuration.GetSection("Redis").Bind(config); - serviceCollection.AddSingleton(); - if (config.RedisPubSubEnabled) { - var redis = ConnectionMultiplexer.Connect(config.RedisPubSubConnectionString); - serviceCollection.AddSingleton(_ => redis.GetSubscriber()); + //AbortOnConnectFail=false lets the multiplexer keep retrying in the background + //instead of crashing the instance when Redis is temporarily unavailable at startup + var options = ConfigurationOptions.Parse(config.RedisPubSubConnectionString); + options.AbortOnConnectFail = false; + var redis = ConnectionMultiplexer.Connect(options); + serviceCollection.AddSingleton(redis); serviceCollection.AddSingleton(); serviceCollection.AddSingleton(); return; } + + serviceCollection.AddSingleton(); } private static void RegisterContextService(IServiceCollection serviceCollection)