feat(payment): add per-user in-memory lock for gateway operations
Build and Deploy to Kubernetes / build-and-deploy (push) Failing after 9m58s
Build and Deploy to Kubernetes / build-and-deploy (push) Failing after 9m58s
Introduce IUserPaymentLock to serialize payment initiate and verify flows per user, preventing concurrent duplicate gateway requests across services. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,143 @@
|
||||
using System.Collections.Concurrent;
|
||||
using CMSMicroservice.Application.Common.Exceptions;
|
||||
using CMSMicroservice.Application.Common.Interfaces;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace CMSMicroservice.Infrastructure.Services;
|
||||
|
||||
/// <summary>
|
||||
/// In-process keyed semaphore lock for payment operations.
|
||||
/// Safe for multi-threaded gRPC/MediatR handlers within a single CMS instance.
|
||||
/// </summary>
|
||||
public sealed class UserPaymentLockService : IUserPaymentLock, IDisposable
|
||||
{
|
||||
private static readonly TimeSpan VerifyWaitTimeout = TimeSpan.FromSeconds(30);
|
||||
private static readonly TimeSpan StaleEntryAge = TimeSpan.FromMinutes(30);
|
||||
private static readonly TimeSpan CleanupInterval = TimeSpan.FromMinutes(5);
|
||||
|
||||
private readonly ConcurrentDictionary<string, LockEntry> _entries = new();
|
||||
private readonly ILogger<UserPaymentLockService> _logger;
|
||||
private readonly Timer _cleanupTimer;
|
||||
private int _disposed;
|
||||
|
||||
public UserPaymentLockService(ILogger<UserPaymentLockService> logger)
|
||||
{
|
||||
_logger = logger;
|
||||
_cleanupTimer = new Timer(_ => CleanupStaleEntries(), null, CleanupInterval, CleanupInterval);
|
||||
}
|
||||
|
||||
public Task<T> ExecuteAsync<T>(
|
||||
string scope,
|
||||
PaymentLockStrategy strategy,
|
||||
Func<CancellationToken, Task<T>> action,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(scope);
|
||||
ArgumentNullException.ThrowIfNull(action);
|
||||
|
||||
return ExecuteCoreAsync(scope, strategy, action, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task ExecuteAsync(
|
||||
string scope,
|
||||
PaymentLockStrategy strategy,
|
||||
Func<CancellationToken, Task> action,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(scope);
|
||||
ArgumentNullException.ThrowIfNull(action);
|
||||
|
||||
await ExecuteCoreAsync(
|
||||
scope,
|
||||
strategy,
|
||||
async ct =>
|
||||
{
|
||||
await action(ct);
|
||||
return true;
|
||||
},
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<T> ExecuteCoreAsync<T>(
|
||||
string scope,
|
||||
PaymentLockStrategy strategy,
|
||||
Func<CancellationToken, Task<T>> action,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var entry = _entries.GetOrAdd(scope, static _ => new LockEntry());
|
||||
entry.Touch();
|
||||
|
||||
var waitTimeout = strategy == PaymentLockStrategy.FailFast
|
||||
? TimeSpan.Zero
|
||||
: VerifyWaitTimeout;
|
||||
|
||||
Interlocked.Increment(ref entry.WaiterCount);
|
||||
|
||||
var acquired = false;
|
||||
try
|
||||
{
|
||||
acquired = await entry.Semaphore.WaitAsync(waitTimeout, cancellationToken);
|
||||
if (!acquired)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Payment lock busy. Scope={Scope}, Strategy={Strategy}",
|
||||
scope,
|
||||
strategy);
|
||||
|
||||
throw new PaymentInProgressException(
|
||||
strategy == PaymentLockStrategy.FailFast
|
||||
? "درخواست پرداخت دیگری در حال پردازش است. لطفاً صبر کنید."
|
||||
: "تأیید پرداخت در حال انجام است. لطفاً چند لحظه صبر کنید.");
|
||||
}
|
||||
|
||||
entry.Touch();
|
||||
return await action(cancellationToken);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (acquired)
|
||||
entry.Semaphore.Release();
|
||||
|
||||
if (Interlocked.Decrement(ref entry.WaiterCount) == 0 && entry.CanRemove)
|
||||
_entries.TryRemove(scope, out _);
|
||||
}
|
||||
}
|
||||
|
||||
private void CleanupStaleEntries()
|
||||
{
|
||||
if (Interlocked.CompareExchange(ref _disposed, 0, 0) == 1)
|
||||
return;
|
||||
|
||||
var cutoff = DateTime.UtcNow - StaleEntryAge;
|
||||
foreach (var (scope, entry) in _entries)
|
||||
{
|
||||
if (entry.LastUsedUtc < cutoff && entry.CanRemove)
|
||||
_entries.TryRemove(scope, out _);
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Interlocked.Exchange(ref _disposed, 1) == 1)
|
||||
return;
|
||||
|
||||
_cleanupTimer.Dispose();
|
||||
|
||||
foreach (var entry in _entries.Values)
|
||||
entry.Semaphore.Dispose();
|
||||
|
||||
_entries.Clear();
|
||||
}
|
||||
|
||||
private sealed class LockEntry
|
||||
{
|
||||
public SemaphoreSlim Semaphore { get; } = new(1, 1);
|
||||
public int WaiterCount;
|
||||
public DateTime LastUsedUtc { get; private set; } = DateTime.UtcNow;
|
||||
|
||||
public void Touch() => LastUsedUtc = DateTime.UtcNow;
|
||||
|
||||
public bool CanRemove =>
|
||||
Volatile.Read(ref WaiterCount) == 0 && Semaphore.CurrentCount == 1;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user