Распределённые локи

<?php

declare(strict_types=1);

namespace App\Lock;

/**
 * Redis-backed distributed lock.
 *
 * Gotchas:
 *  - Not safe in absence of fencing tokens if holder pauses > TTL.
 *  - Assumes a single Redis primary; failover can lose the lock.
 */
final class RedisLock
{
    public function __construct(
        private readonly \Redis $redis,
        private readonly int $defaultTtlMs = 30_000,
    ) {}

    /**
     * Try to acquire a lock. Returns lock handle or null.
     */
    public function acquire(string $resource, ?int $ttlMs = null): ?LockHandle
    {
        $ttl = $ttlMs ?? $this->defaultTtlMs;
        $token = bin2hex(random_bytes(16));
        $key = "lock:$resource";

        $ok = $this->redis->set($key, $token, ['NX', 'PX' => $ttl]);
        if (!$ok) {
            return null;
        }

        // Fencing token: monotonically increasing counter
        $fence = $this->redis->incr('lock:fence:' . $resource);

        return new LockHandle($this->redis, $key, $token, $fence);
    }

    /**
     * Block until lock is acquired or timeout elapsed.
     */
    public function acquireBlocking(string $resource, int $waitMs, ?int $ttlMs = null): ?LockHandle
    {
        $deadline = microtime(true) + $waitMs / 1000;
        $backoff = 10;
        do {
            $h = $this->acquire($resource, $ttlMs);
            if ($h !== null) {
                return $h;
            }
            usleep($backoff * 1000 + random_int(0, $backoff * 1000));
            $backoff = min($backoff * 2, 200);
        } while (microtime(true) < $deadline);

        return null;
    }
}

final class LockHandle
{
    public function __construct(
        private readonly \Redis $redis,
        public readonly string $key,
        public readonly string $token,
        public readonly int $fencingToken,
    ) {}

    /**
     * Atomic release: only delete if we still own the lock.
     */
    public function release(): bool
    {
        $script = <<<'LUA'
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            end
            return 0
        LUA;
        return (bool) $this->redis->eval($script, [$this->key, $this->token], 1);
    }

    /**
     * Extend TTL (keep lock alive for long-running operations).
     */
    public function extend(int $ttlMs): bool
    {
        $script = <<<'LUA'
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('PEXPIRE', KEYS[1], ARGV[2])
            end
            return 0
        LUA;
        return (bool) $this->redis->eval(
            $script,
            [$this->key, $this->token, (string) $ttlMs],
            1,
        );
    }
}

// Usage:
// $lock = new RedisLock($redis);
// $h = $lock->acquireBlocking('cron:nightly-report', waitMs: 5000, ttlMs: 60_000);
// if ($h === null) { return; }
// try {
//     runReport(fencingToken: $h->fencingToken);
// } finally {
//     $h->release();
// }

package lock

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"math/big"
	"time"

	"github.com/redis/go-redis/v9"
)

var releaseScript = redis.NewScript(`
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    end
    return 0
`)

var extendScript = redis.NewScript(`
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('PEXPIRE', KEYS[1], ARGV[2])
    end
    return 0
`)

type Lock struct {
	rdb *redis.Client
}

type Handle struct {
	rdb     *redis.Client
	key     string
	token   string
	Fencing int64
}

func New(rdb *redis.Client) *Lock { return &Lock{rdb: rdb} }

// Acquire tries to take the lock once; returns nil handle if already held.
func (l *Lock) Acquire(ctx context.Context, resource string, ttl time.Duration) (*Handle, error) {
	token, err := randToken()
	if err != nil {
		return nil, err
	}
	key := "lock:" + resource

	ok, err := l.rdb.SetNX(ctx, key, token, ttl).Result()
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, nil
	}

	fence, err := l.rdb.Incr(ctx, "lock:fence:"+resource).Result()
	if err != nil {
		// Best-effort release and report
		_, _ = releaseScript.Run(ctx, l.rdb, []string{key}, token).Result()
		return nil, err
	}

	return &Handle{rdb: l.rdb, key: key, token: token, Fencing: fence}, nil
}

// AcquireBlocking waits until acquired or timeout reached.
func (l *Lock) AcquireBlocking(ctx context.Context, resource string, ttl, wait time.Duration) (*Handle, error) {
	deadline := time.Now().Add(wait)
	backoff := 10 * time.Millisecond
	for {
		h, err := l.Acquire(ctx, resource, ttl)
		if err != nil {
			return nil, err
		}
		if h != nil {
			return h, nil
		}
		if time.Now().After(deadline) {
			return nil, errors.New("lock acquire timeout")
		}
		// Jittered backoff to avoid thundering herd
		jitter, _ := rand.Int(rand.Reader, big.NewInt(int64(backoff)))
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(backoff + time.Duration(jitter.Int64())):
		}
		backoff *= 2
		if backoff > 200*time.Millisecond {
			backoff = 200 * time.Millisecond
		}
	}
}

func (h *Handle) Release(ctx context.Context) error {
	_, err := releaseScript.Run(ctx, h.rdb, []string{h.key}, h.token).Result()
	return err
}

func (h *Handle) Extend(ctx context.Context, ttl time.Duration) error {
	_, err := extendScript.Run(ctx, h.rdb, []string{h.key}, h.token, ttl.Milliseconds()).Result()
	return err
}

func randToken() (string, error) {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		return "", err
	}
	return hex.EncodeToString(b), nil
}

using System.Security.Cryptography;
using StackExchange.Redis;

namespace App.Lock;

/// Redis-backed distributed lock.
///
/// Gotchas:
///  - Not safe without fencing tokens if the holder pauses longer than the TTL.
///  - Assumes a single Redis primary; failover can lose the lock.
public sealed class RedisLock(IDatabase redis, int defaultTtlMs = 30_000)
{
    /// Tries to acquire the lock once. Returns a handle or null.
    public async Task<LockHandle?> AcquireAsync(string resource, int? ttlMs = null)
    {
        var ttl = TimeSpan.FromMilliseconds(ttlMs ?? defaultTtlMs);
        var token = Convert.ToHexString(RandomNumberGenerator.GetBytes(16));
        var key = (RedisKey)$"lock:{resource}";

        if (!await redis.StringSetAsync(key, token, ttl, When.NotExists))
        {
            return null;
        }

        // Fencing token: monotonically increasing counter
        var fence = await redis.StringIncrementAsync($"lock:fence:{resource}");

        return new LockHandle(redis, key, token, fence);
    }

    /// Blocks until the lock is acquired or the wait budget elapses.
    public async Task<LockHandle?> AcquireBlockingAsync(
        string resource, int waitMs, int? ttlMs = null, CancellationToken ct = default)
    {
        var deadline = DateTimeOffset.UtcNow.AddMilliseconds(waitMs);
        var backoffMs = 10;

        do
        {
            var handle = await AcquireAsync(resource, ttlMs);
            if (handle is not null)
            {
                return handle;
            }

            // Jittered backoff to avoid a thundering herd
            await Task.Delay(backoffMs + RandomNumberGenerator.GetInt32(backoffMs), ct);
            backoffMs = Math.Min(backoffMs * 2, 200);
        }
        while (DateTimeOffset.UtcNow < deadline);

        return null;
    }
}

public sealed class LockHandle(IDatabase redis, RedisKey key, string token, long fencingToken)
    : IAsyncDisposable
{
    private const string ReleaseScript = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        end
        return 0
        """;

    private const string ExtendScript = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('PEXPIRE', KEYS[1], ARGV[2])
        end
        return 0
        """;

    public long FencingToken { get; } = fencingToken;

    /// Atomic release: only delete if we still own the lock.
    public async Task<bool> ReleaseAsync()
    {
        var result = await redis.ScriptEvaluateAsync(
            ReleaseScript, [key], [token]);
        return (long)result == 1;
    }

    /// Extend the TTL (keeps the lock alive for long-running operations).
    public async Task<bool> ExtendAsync(int ttlMs)
    {
        var result = await redis.ScriptEvaluateAsync(
            ExtendScript, [key], [token, ttlMs]);
        return (long)result == 1;
    }

    public async ValueTask DisposeAsync() => await ReleaseAsync();
}

// Usage:
// var locker = new RedisLock(redis);
// await using var handle = await locker.AcquireBlockingAsync(
//     "cron:nightly-report", waitMs: 5000, ttlMs: 60_000);
// if (handle is null) { return; }
// await RunReportAsync(handle.FencingToken);

import asyncio
import secrets
from types import TracebackType

import redis.asyncio as redis

RELEASE_SCRIPT = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    end
    return 0
"""

EXTEND_SCRIPT = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('PEXPIRE', KEYS[1], ARGV[2])
    end
    return 0
"""

class LockHandle:
    """Owns one acquired lock. Release is atomic via a Lua CAS."""

    def __init__(
        self, client: redis.Redis, key: str, token: str, fencing_token: int
    ) -> None:
        self._redis = client
        self._key = key
        self._token = token
        self.fencing_token = fencing_token

    async def release(self) -> bool:
        """Only delete if we still own the lock."""
        result = await self._redis.eval(RELEASE_SCRIPT, 1, self._key, self._token)
        return bool(result)

    async def extend(self, ttl_ms: int) -> bool:
        """Extend TTL to keep the lock alive for long-running operations."""
        result = await self._redis.eval(
            EXTEND_SCRIPT, 1, self._key, self._token, ttl_ms
        )
        return bool(result)

    async def __aenter__(self) -> "LockHandle":
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self.release()

class RedisLock:
    """Redis-backed distributed lock.

    Gotchas:
      - Not safe without fencing tokens if the holder pauses longer than the TTL.
      - Assumes a single Redis primary; failover can lose the lock.
    """

    def __init__(self, client: redis.Redis, default_ttl_ms: int = 30_000) -> None:
        self._redis = client
        self._default_ttl_ms = default_ttl_ms

    async def acquire(self, resource: str, ttl_ms: int | None = None) -> LockHandle | None:
        """Try to acquire once; return a handle or None."""
        ttl = ttl_ms or self._default_ttl_ms
        token = secrets.token_hex(16)
        key = f"lock:{resource}"

        if not await self._redis.set(key, token, nx=True, px=ttl):
            return None

        # Fencing token: monotonically increasing counter
        fence = await self._redis.incr(f"lock:fence:{resource}")
        return LockHandle(self._redis, key, token, fence)

    async def acquire_blocking(
        self, resource: str, wait_ms: int, ttl_ms: int | None = None
    ) -> LockHandle | None:
        """Block until acquired or the wait budget elapses."""
        loop = asyncio.get_running_loop()
        deadline = loop.time() + wait_ms / 1000
        backoff = 0.01

        while True:
            handle = await self.acquire(resource, ttl_ms)
            if handle is not None:
                return handle
            if loop.time() >= deadline:
                return None

            # Jittered backoff to avoid a thundering herd
            await asyncio.sleep(backoff + secrets.randbelow(1000) / 1000 * backoff)
            backoff = min(backoff * 2, 0.2)

# Usage:
# locker = RedisLock(client)
# handle = await locker.acquire_blocking("cron:nightly-report", wait_ms=5000, ttl_ms=60_000)
# if handle is None:
#     return
# async with handle:
#     await run_report(fencing_token=handle.fencing_token)

<?php

declare(strict_types=1);

final class NightlyReportCommand
{
    public function __construct(
        private readonly RedisLock $lock,
        private readonly ReportService $report,
        private readonly LoggerInterface $log,
    ) {}

    public function run(): int
    {
        $h = $this->lock->acquire('cron:nightly-report', ttlMs: 60 * 60 * 1000);
        if ($h === null) {
            $this->log->info('Another worker is running this job; exiting');
            return 0;
        }

        try {
            // Long-running job: extend lock periodically via a heartbeat thread/goroutine
            $this->report->generate(fence: $h->fencingToken);
        } finally {
            $h->release();
        }
        return 0;
    }
}

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace App.Cron;

/// Runs on every replica; only the lock holder does the work.
public sealed class NightlyReportJob(
    RedisLock lockService,
    IReportService report,
    ILogger<NightlyReportJob> log) : BackgroundService
{
    private static readonly TimeSpan LockTtl = TimeSpan.FromHours(1);

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        var handle = await lockService.AcquireAsync(
            "cron:nightly-report", (int)LockTtl.TotalMilliseconds);

        if (handle is null)
        {
            log.LogInformation("Another worker is running this job; exiting");
            return;
        }

        try
        {
            // Long-running job: extend the lock periodically via a heartbeat task
            await report.GenerateAsync(handle.FencingToken, ct);
        }
        finally
        {
            await handle.ReleaseAsync();
        }
    }
}

import logging

log = logging.getLogger(__name__)

LOCK_TTL_MS = 60 * 60 * 1000

class NightlyReportCommand:
    """Runs on every replica; only the lock holder does the work."""

    def __init__(self, lock: RedisLock, report: ReportService) -> None:
        self._lock = lock
        self._report = report

    async def run(self) -> int:
        handle = await self._lock.acquire("cron:nightly-report", ttl_ms=LOCK_TTL_MS)
        if handle is None:
            log.info("Another worker is running this job; exiting")
            return 0

        try:
            # Long-running job: extend the lock periodically via a heartbeat task
            await self._report.generate(fence=handle.fencing_token)
        finally:
            await handle.release()

        return 0