Перейти к основному содержимому

Distributed Locking — Архитектура и Практическое Руководство

Назначение​

Документ описывает, когда и как использовать distributed locking в Pin через библиотеку madelson/DistributedLock. Включает выбор backend (Redis vs PostgreSQL), правила использования, anti-patterns и примеры для типичных enterprise use cases.

Distributed Lock vs Оптимистичный Concurrency​

Выбор между двумя подходами зависит от природы конфликта:

КритерийDistributed LockOptimistic Concurrency (xmin / rowversion)
Частота конфликтовВысокая (много writers)Низкая (редкие коллизии)
ОперацияДорогая (нельзя повторить дёшево)Дешёвая (повторить легко)
DurationНесколько секунд (external call)Миллисекунды (DB write)
Примеры в PinБонус-активация, scheduled job, миграцияОбновление баланса, статус KYC
Penalty при конфликтеLock wait / fail fastRetry по exception

Правило: предпочитай оптимистичный concurrency. Distributed lock — только когда повтор операции недопустим или слишком дорог.

Пакет​

src/building-blocks/
  Talent.Common.Locking/

NuGet dependencies:

<PackageReference Include="DistributedLock.Redis"      Version="1.*" />
<PackageReference Include="DistributedLock.Postgres"   Version="1.*" />
<PackageReference Include="DistributedLock.Core"       Version="1.*" />

Провайдеры​

Pin поддерживает два backend для locks:

Redis Lock (рекомендуется для операционных locks)​

  • Быстрый, низкая latency (<1ms).
  • Подходит для: incentive claim, job deduplication, idempotency write lock.
  • Недостаток: при Redis failover lock может быть потерян (Redlock не используется — слишком сложный).

PostgreSQL Advisory Lock (рекомендуется для migration lock)​

  • Гарантированная durability (survives Redis restart).
  • Подходит для: database migrations, long-running reconciliation.
  • Медленнее Redis, но надёжнее при критичных операциях.

IPinLockProvider​

// Talent.Common.Locking/IPinLockProvider.cs

/// <summary>
/// Абстракция distributed lock. Скрывает backend (Redis или PostgreSQL).
/// Регистрируется как Singleton.
/// </summary>
public interface IPinLockProvider
{
    /// <summary>
    /// Попытка захватить lock. Если lock занят — ждём до <paramref name="timeout"/>.
    /// Возвращает null если lock не удалось захватить.
    /// </summary>
    Task<IDistributedLockHandle?> TryAcquire(
        string resource,
        TimeSpan timeout = default,
        CancellationToken ct = default);

    /// <summary>Захватить lock или выбросить <see cref="DistributedLockTimeoutException"/>.</summary>
    Task<IDistributedLockHandle> Acquire(
        string resource,
        TimeSpan timeout,
        CancellationToken ct = default);
}

public interface IDistributedLockHandle : IAsyncDisposable
{
    string Resource { get; }
    string OwnerId { get; }    // уникальный ID держателя (для диагностики)
    DateTimeOffset AcquiredAt { get; }
    DateTimeOffset ExpiresAt { get; }
}

Реализация через DistributedLock​

// Talent.Common.Locking/RedisLockProvider.cs

public sealed class RedisLockProvider : IPinLockProvider
{
    private readonly IConnectionMultiplexer _redis;
    private readonly ILogger<RedisLockProvider> _logger;
    private readonly LockOptions _options;

    public async Task<IDistributedLockHandle?> TryAcquire(
        string resource,
        TimeSpan timeout = default,
        CancellationToken ct = default)
    {
        var lockName = NormalizeResource(resource);
        var @lock    = new RedisDistributedLock(lockName, _redis.GetDatabase());

        var handle = await @lock.TryAcquireAsync(
            timeout:           timeout == default ? TimeSpan.Zero : timeout,
            cancellationToken: ct);

        if (handle is null)
        {
            _logger.LogWarning(
                "DistributedLock: failed to acquire {Resource} within {Timeout}ms",
                resource, timeout.TotalMilliseconds);
            return null;
        }

        _logger.LogDebug("DistributedLock: acquired {Resource}", resource);
        return new DistributedLockHandle(handle, resource, _logger);
    }

    private static string NormalizeResource(string resource)
        => $"Pin:lock:{resource.ToLowerInvariant().Replace('/', ':')}";
}

Соглашения Именования Lock Keys​

Pin:lock:incentive:claim:{userId}:{incentiveOfferId}
Pin:lock:job:scheduled:{jobName}
Pin:lock:migration:{serviceName}
Pin:lock:withdrawal:approve:{withdrawalId}
Pin:lock:payment:callback:{providerRef}
Pin:lock:user:session:{userId}
Pin:lock:reconciliation:{date}:{providerId}

Правила:

  • Все ключи начинаются с Pin:lock:.
  • Включают entity type и entity id — чтобы один игрок не блокировал другого.
  • Нижний регистр, : как разделитель.

Use Cases​

1. Бонус Claim Lock — один игрок одновременно один бонус​

// Pin.Incentive.Application/Commands/ClaimIncentive/ClaimIncentiveCommandHandler.cs

public async Task<Result<ClaimIncentiveResult>> Handle(ClaimIncentiveCommand cmd, CancellationToken ct)
{
    var lockKey = $"incentive:claim:{cmd.UserId.Value}:{cmd.IncentiveOfferId.Value}";

    // Пытаемся захватить lock на 10 секунд — не ждём, если занято
    await using var handle = await _lockProvider.TryAcquire(lockKey, timeout: TimeSpan.Zero, ct);
    if (handle is null)
        return Result.Fail(IncentiveErrors.ClaimAlreadyInProgress);

    // Безопасно: только один handler для данной пары (user, offer)
    var existing = await _userIncentiveRepo.Find(cmd.UserId, cmd.IncentiveOfferId, ct);
    if (existing is not null)
        return Result.Fail(IncentiveErrors.AlreadyClaimed);

    var incentive = await _incentiveOfferRepo.Get(cmd.IncentiveOfferId, ct);
    var result = incentive.ClaimFor(cmd.UserId, _clock.UtcNow);
    if (result.IsFailed) return result.ToResult<ClaimIncentiveResult>();

    await _userIncentiveRepo.Add(result.Value, ct);
    return Result.Ok(new ClaimIncentiveResult(result.Value.Id));
}
// Lock освобождается при выходе из using (IAsyncDisposable)

2. Scheduled Job Lock — один pod запускает job​

// Talent.Common.Infrastructure/Jobs/DistributedJobRunner.cs

public sealed class DistributedJobRunner
{
    private readonly IPinLockProvider _lockProvider;

    /// <summary>
    /// Запускает <paramref name="job"/> только если удалось захватить lock.
    /// Гарантирует, что при нескольких pod-ах job выполняется ровно один раз.
    /// </summary>
    public async Task RunExclusive(
        string jobName,
        Func<CancellationToken, Task> job,
        TimeSpan lockTimeout,
        CancellationToken ct = default)
    {
        var lockKey = $"job:scheduled:{jobName}";
        await using var handle = await _lockProvider.TryAcquire(lockKey, TimeSpan.Zero, ct);

        if (handle is null)
        {
            _logger.LogInformation("Job {JobName}: skipped, another instance holds lock", jobName);
            return;
        }

        _logger.LogInformation("Job {JobName}: starting", jobName);
        await job(ct);
        _logger.LogInformation("Job {JobName}: completed", jobName);
    }
}

// Пример использования в background worker:
public sealed class IncentiveExpiryJob : IJob
{
    private readonly DistributedJobRunner _runner;
    private readonly IIncentiveExpiryService _service;

    public async Task Execute(IJobExecutionContext context)
    {
        await _runner.RunExclusive(
            jobName:     "incentive-expiry-daily",
            job:         ct => _service.ExpireOverdueIncentivees(ct),
            lockTimeout: TimeSpan.FromMinutes(30),
            ct:          context.CancellationToken);
    }
}

3. Payment Callback Lock — idempotency при повторных callbacks​

// При получении callback от провайдера — lock на providerReference
var lockKey = $"payment:callback:{providerReference}";
await using var handle = await _lockProvider.TryAcquire(
    lockKey, timeout: TimeSpan.FromSeconds(5), ct);

if (handle is null)
    return Result.Fail(PaymentErrors.CallbackProcessingInProgress);

// Проверяем inbox — был ли уже обработан
var existing = await _inboxRepo.Find(providerReference, ct);
if (existing is not null)
    return Result.Ok(/* idempotent replay */);

// Обрабатываем callback...

4. Migration Lock — PostgreSQL advisory lock​

// Pin.Kernel.Migrator/DistributedMigrationLock.cs

public sealed class DistributedMigrationLock
{
    private readonly NpgsqlConnection _connection;
    private readonly string _serviceName;

    public async Task<IAsyncDisposable> Acquire(CancellationToken ct)
    {
        await _connection.OpenAsync(ct);

        // PostgreSQL advisory lock — уникален для сервиса по hash имени
        var lockId = Math.Abs(_serviceName.GetHashCode());
        var @lock  = new PostgresDistributedLock(
            new PostgresAdvisoryLockKey(lockId, isGlobal: true),
            _connection);

        var handle = await @lock.AcquireAsync(
            timeout:           TimeSpan.FromMinutes(5),
            cancellationToken: ct);

        return handle;
    }
}

Правила​

  • Lock key всегда включает entity id — не блокируй весь тип, только конкретный объект.
  • Timeout: операционные locks — до 10 секунд; job locks — минуты; migration lock — 5 минут.
  • Всегда используй await using (IAsyncDisposable) — lock освобождается даже при исключении.
  • Никогда не держи lock дольше времени HTTP request (>30s — anti-pattern).
  • При TryAcquire возвращает null → return Result.Fail с typed error, не retry.
  • Lock не заменяет database transaction — используй оба: lock предотвращает race, transaction обеспечивает atomicity.
  • Не используй distributed lock для оптимистичного concurrency (xmin/rowversion достаточно).
  • Логируй lock acquire/release с ключом и duration на уровне Debug.
  • В тестах используй InProcessDistributedLock из DistributedLock.Core (без Redis/PostgreSQL).

Anti-Patterns​

// ПЛОХО: lock на тип без id — блокирует всех игроков
var lockKey = "incentive:claim"; // ← слишком широкий

// ПЛОХО: lock без timeout — может висеть бесконечно
await _lockProvider.Acquire("some:resource", TimeSpan.Zero);

// ПЛОХО: lock снаружи транзакции
await using var handle = await _lockProvider.Acquire(key, ...);
// ... много бизнес-логики без транзакции ...
await _repo.Update(entity); // нет транзакции!

// ХОРОШО: lock + transaction
await using var handle = await _lockProvider.Acquire(key, TimeSpan.FromSeconds(5), ct);
await using var tx = await _transactionBoundary.Begin(ct);
// ...
await tx.Commit(ct);

Регистрация​

// AddPinLocking.cs — в Talent.Common.Locking

public static IPinInfrastructureBuilder AddPinLocking(
    this IPinInfrastructureBuilder builder,
    LockingBackend backend = LockingBackend.Redis)
{
    switch (backend)
    {
        case LockingBackend.Redis:
            builder.Services.AddSingleton<IPinLockProvider, RedisLockProvider>();
            break;
        case LockingBackend.Postgres:
            builder.Services.AddSingleton<IPinLockProvider, PostgresLockProvider>();
            break;
    }
    builder.Services.AddSingleton<DistributedJobRunner>();
    return builder;
}

public enum LockingBackend { Redis, Postgres }

Ссылки​

Обратные ссылки​

Автоссылки: где используется этот документ.