fix(identity): Lock when handling user sessions

aspnet-core/modules/identity/LINGYUN.Abp.Identity.Domain/LINGYUN/Abp/Identity/AbpIdentityDomainModule.cs

@@ -5,6 +5,7 @@
 using Volo.Abp;
 using Volo.Abp.AutoMapper;
 using Volo.Abp.BackgroundWorkers;
+using Volo.Abp.DistributedLocking;
 using Volo.Abp.Domain.Entities.Events.Distributed;
 using Volo.Abp.Identity;
 using Volo.Abp.Modularity;
@@ -14,6 +15,7 @@ namespace LINGYUN.Abp.Identity;
 
 [DependsOn(
     typeof(AbpIdentityDomainSharedModule),
+    typeof(AbpDistributedLockingAbstractionsModule),
     typeof(Volo.Abp.Identity.AbpIdentityDomainModule))]
 public class AbpIdentityDomainModule : AbpModule
 {

aspnet-core/modules/identity/LINGYUN.Abp.Identity.Domain/LINGYUN/Abp/Identity/Session/IdentitySessionCacheItemSynchronizer.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Threading.Tasks;
 using Volo.Abp.DependencyInjection;
+using Volo.Abp.DistributedLocking;
 using Volo.Abp.Domain.Entities.Events;
 using Volo.Abp.Domain.Entities.Events.Distributed;
 using Volo.Abp.EventBus;
@@ -22,15 +23,18 @@ public class IdentitySessionCacheItemSynchronizer :
 {
     public ILogger<IdentitySessionCacheItemSynchronizer> Logger { protected get; set; }
     protected ISettingProvider SettingProvider { get; }
+    protected IAbpDistributedLock DistributedLock { get; }
     protected IIdentitySessionCache IdentitySessionCache { get; }
     protected IIdentitySessionStore IdentitySessionStore { get; }
 
     public IdentitySessionCacheItemSynchronizer(
         ISettingProvider settingProvider,
+        IAbpDistributedLock distributedLock,
         IIdentitySessionCache identitySessionCache, 
         IIdentitySessionStore identitySessionStore)
     {
         SettingProvider = settingProvider;
+        DistributedLock = distributedLock;
         IdentitySessionCache = identitySessionCache;
         IdentitySessionStore = identitySessionStore;
 
@@ -45,34 +49,71 @@ public async virtual Task HandleEventAsync(EntityDeletedEto<IdentitySessionEto>
     [UnitOfWork]
     public async virtual Task HandleEventAsync(EntityCreatedEto<IdentitySessionEto> eventData)
     {
-        await RefreshSessionCache(eventData.Entity);
-        await CheckConcurrentLoginStrategy(eventData.Entity);
+        var lockKey = $"{nameof(IdentitySessionCacheItemSynchronizer)}_{nameof(EntityCreatedEto<IdentitySessionEto>)}";
+        await using (var handle = await DistributedLock.TryAcquireAsync(lockKey))
+        {
+            Logger.LogInformation($"Lock is acquired for {lockKey}");
+
+            if (handle == null)
+            {
+                Logger.LogInformation($"Handle is null because of the locking for : {lockKey}");
+                return;
+            }
+
+            await RefreshSessionCache(eventData.Entity);
+            await CheckConcurrentLoginStrategy(eventData.Entity);
+        }
     }
 
+    [UnitOfWork]
     public async virtual Task HandleEventAsync(IdentitySessionChangeAccessedEvent eventData)
     {
-        var idetitySession = await IdentitySessionStore.FindAsync(eventData.SessionId);
-        if (idetitySession != null)
+        var lockKey = $"{nameof(IdentitySessionCacheItemSynchronizer)}_{nameof(IdentitySessionChangeAccessedEvent)}";
+        await using (var handle = await DistributedLock.TryAcquireAsync(lockKey))
         {
-            if (!eventData.IpAddresses.IsNullOrWhiteSpace())
+            Logger.LogInformation($"Lock is acquired for {lockKey}");
+
+            if (handle == null)
             {
-                idetitySession.SetIpAddresses(eventData.IpAddresses.Split(","));
+                Logger.LogInformation($"Handle is null because of the locking for : {lockKey}");
+                return;
             }
-            idetitySession.UpdateLastAccessedTime(eventData.LastAccessed);
 
-            await IdentitySessionStore.UpdateAsync(idetitySession);
-        }
-        else
-        {
-            // 数据库中不存在会话, 清理缓存, 后续请求会话失效
-            await IdentitySessionCache.RemoveAsync(eventData.SessionId);
+            var idetitySession = await IdentitySessionStore.FindAsync(eventData.SessionId);
+            if (idetitySession != null)
+            {
+                if (!eventData.IpAddresses.IsNullOrWhiteSpace())
+                {
+                    idetitySession.SetIpAddresses(eventData.IpAddresses.Split(","));
+                }
+                idetitySession.UpdateLastAccessedTime(eventData.LastAccessed);
+
+                await IdentitySessionStore.UpdateAsync(idetitySession);
+            }
+            else
+            {
+                // 数据库中不存在会话, 清理缓存, 后续请求会话失效
+                await IdentitySessionCache.RemoveAsync(eventData.SessionId);
+            }
         }
     }
 
     public async virtual Task HandleEventAsync(EntityDeletedEventData<IdentityUser> eventData)
     {
-        // 用户被删除, 移除所有会话
-        await IdentitySessionStore.RevokeAllAsync(eventData.Entity.Id);
+        var lockKey = $"{nameof(IdentitySessionCacheItemSynchronizer)}_{nameof(EntityDeletedEventData<IdentityUser>)}";
+        await using (var handle = await DistributedLock.TryAcquireAsync(lockKey))
+        {
+            Logger.LogInformation($"Lock is acquired for {lockKey}");
+
+            if (handle == null)
+            {
+                Logger.LogInformation($"Handle is null because of the locking for : {lockKey}");
+                return;
+            }
+
+            // 用户被删除, 移除所有会话
+            await IdentitySessionStore.RevokeAllAsync(eventData.Entity.Id);
+        }
     }
 
     protected async virtual Task RefreshSessionCache(IdentitySessionEto session)

aspnet-core/modules/identity/LINGYUN.Abp.Identity.Domain/LINGYUN/Abp/Identity/Session/IdentitySessionCleanupBackgroundWorker.cs

@@ -1,20 +1,25 @@
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using System.Threading.Tasks;
 using Volo.Abp.BackgroundWorkers;
+using Volo.Abp.DistributedLocking;
 using Volo.Abp.Threading;
 
 namespace LINGYUN.Abp.Identity.Session;
 
 public class IdentitySessionCleanupBackgroundWorker : AsyncPeriodicBackgroundWorkerBase
 {
+    protected IAbpDistributedLock DistributedLock { get; }
     protected IdentitySessionCleanupOptions Options { get; }
     public IdentitySessionCleanupBackgroundWorker(
         AbpAsyncTimer timer,
         IServiceScopeFactory serviceScopeFactory,
-        IOptions<IdentitySessionCleanupOptions> options)
+        IOptions<IdentitySessionCleanupOptions> options,
+        IAbpDistributedLock distributedLock)
         : base(timer, serviceScopeFactory)
     {
+        DistributedLock = distributedLock;
         Options = options.Value;
         timer.Period = Options.CleanupPeriod;
     }
@@ -26,9 +31,22 @@ protected async override Task DoWorkAsync(PeriodicBackgroundWorkerContext worker
             return;
         }
 
-        await workerContext
-             .ServiceProvider
-             .GetRequiredService<IdentitySessionCleanupService>()
-             .CleanAsync();
+        await using (var handle = await DistributedLock.TryAcquireAsync(nameof(IdentitySessionCleanupBackgroundWorker)))
+        {
+            Logger.LogInformation($"Lock is acquired for {nameof(IdentitySessionCleanupBackgroundWorker)}");
+
+            if (handle != null)
+            {
+                await workerContext
+                    .ServiceProvider
+                    .GetRequiredService<IdentitySessionCleanupService>()
+                    .CleanAsync();
+
+                Logger.LogInformation($"Lock is released for {nameof(IdentitySessionCleanupBackgroundWorker)}");
+                return;
+            }
+
+            Logger.LogInformation($"Handle is null because of the locking for : {nameof(IdentitySessionCleanupBackgroundWorker)}");
+        }
     }
 }