Выбор лидера

<?php

declare(strict_types=1);

/**
 * Simplified leader election using Redis (practical PHP approach).
 * Production systems use etcd/ZooKeeper, but Redis works for simpler cases.
 */
final class RedisLeaderElection
{
    private readonly string $lockKey;

    public function __construct(
        private readonly \Redis $redis,
        private readonly string $nodeId,
        private readonly string $electionName,
        private readonly int $ttlSeconds = 30,
        private readonly int $renewIntervalSec = 10,
    ) {
        $this->lockKey = "leader:{$this->electionName}";
    }

    /**
     * Try to become the leader.
     * Uses SET NX (set if not exists) + TTL for automatic expiration.
     */
    public function tryBecomeLeader(): bool
    {
        // SET key value NX EX ttl
        // NX = only set if key doesn't exist
        // EX = expire after ttl seconds
        $result = $this->redis->set(
            $this->lockKey,
            $this->nodeId,
            ['NX', 'EX' => $this->ttlSeconds],
        );

        return $result === true;
    }

    /**
     * Renew leadership (extend TTL).
     * Only succeeds if this node is still the leader.
     */
    public function renewLeadership(): bool
    {
        // Lua script for atomic check-and-extend
        $script = <<<'LUA'
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("EXPIRE", KEYS[1], ARGV[2])
            end
            return 0
        LUA;

        $result = $this->redis->eval(
            $script,
            [$this->lockKey, $this->nodeId, $this->ttlSeconds],
            1,
        );

        return $result === 1;
    }

    /**
     * Get current leader node ID.
     */
    public function getCurrentLeader(): ?string
    {
        $leader = $this->redis->get($this->lockKey);
        return $leader !== false ? $leader : null;
    }

    /**
     * Am I the current leader?
     */
    public function isLeader(): bool
    {
        return $this->getCurrentLeader() === $this->nodeId;
    }

    /**
     * Step down from leadership.
     * Only releases if this node is the leader (atomic).
     */
    public function stepDown(): bool
    {
        $script = <<<'LUA'
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            end
            return 0
        LUA;

        return $this->redis->eval($script, [$this->lockKey, $this->nodeId], 1) === 1;
    }

    /**
     * Run leader election loop.
     * Continuously tries to become leader and renews leadership.
     */
    public function run(callable $onBecomeLeader, callable $onLoseLeadership): void
    {
        $wasLeader = false;

        while (true) {
            if ($this->isLeader()) {
                if (!$wasLeader) {
                    $onBecomeLeader($this->nodeId);
                    $wasLeader = true;
                }
                $this->renewLeadership();
            } else {
                if ($wasLeader) {
                    $onLoseLeadership($this->nodeId);
                    $wasLeader = false;
                }
                $this->tryBecomeLeader();
            }

            sleep($this->renewIntervalSec);
        }
    }
}

// Usage: scheduled task coordinator
$election = new RedisLeaderElection(
    redis: new \Redis(),
    nodeId: gethostname() . ':' . getmypid(),
    electionName: 'cron-scheduler',
    ttlSeconds: 30,
    renewIntervalSec: 10,
);

$election->run(
    onBecomeLeader: function (string $nodeId) {
        echo "Node {$nodeId} is now the leader. Starting scheduler.\n";
    },
    onLoseLeadership: function (string $nodeId) {
        echo "Node {$nodeId} lost leadership. Stopping scheduler.\n";
    },
);

package election

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

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

// RedisLeaderElection implements leader election using Redis SET NX + TTL.
type RedisLeaderElection struct {
	rdb          *redis.Client
	nodeID       string
	lockKey      string
	ttl          time.Duration
	renewInterval time.Duration
}

func New(rdb *redis.Client, nodeID, electionName string, ttl, renewInterval time.Duration) *RedisLeaderElection {
	return &RedisLeaderElection{
		rdb:           rdb,
		nodeID:        nodeID,
		lockKey:       fmt.Sprintf("leader:%s", electionName),
		ttl:           ttl,
		renewInterval: renewInterval,
	}
}

// TryBecomeLeader attempts to acquire leadership using SET NX EX.
func (e *RedisLeaderElection) TryBecomeLeader(ctx context.Context) (bool, error) {
	ok, err := e.rdb.SetNX(ctx, e.lockKey, e.nodeID, e.ttl).Result()
	return ok, err
}

// RenewLeadership extends TTL only if this node is still the leader (atomic Lua).
func (e *RedisLeaderElection) RenewLeadership(ctx context.Context) (bool, error) {
	script := redis.NewScript(`
		if redis.call("GET", KEYS[1]) == ARGV[1] then
			return redis.call("EXPIRE", KEYS[1], ARGV[2])
		end
		return 0
	`)
	result, err := script.Run(ctx, e.rdb, []string{e.lockKey}, e.nodeID, int(e.ttl.Seconds())).Int()
	return result == 1, err
}

// GetCurrentLeader returns the node ID of the current leader.
func (e *RedisLeaderElection) GetCurrentLeader(ctx context.Context) (string, error) {
	return e.rdb.Get(ctx, e.lockKey).Result()
}

// IsLeader checks if this node is the current leader.
func (e *RedisLeaderElection) IsLeader(ctx context.Context) bool {
	leader, err := e.GetCurrentLeader(ctx)
	return err == nil && leader == e.nodeID
}

// StepDown releases leadership only if this node holds it (atomic).
func (e *RedisLeaderElection) StepDown(ctx context.Context) (bool, error) {
	script := redis.NewScript(`
		if redis.call("GET", KEYS[1]) == ARGV[1] then
			return redis.call("DEL", KEYS[1])
		end
		return 0
	`)
	result, err := script.Run(ctx, e.rdb, []string{e.lockKey}, e.nodeID).Int()
	return result == 1, err
}

// Run starts the leader election loop with callbacks.
func (e *RedisLeaderElection) Run(ctx context.Context, onBecomeLeader, onLoseLeadership func(nodeID string)) {
	wasLeader := false
	ticker := time.NewTicker(e.renewInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			if wasLeader {
				e.StepDown(context.Background())
				onLoseLeadership(e.nodeID)
			}
			return
		case <-ticker.C:
			if e.IsLeader(ctx) {
				if !wasLeader {
					onBecomeLeader(e.nodeID)
					wasLeader = true
				}
				e.RenewLeadership(ctx)
			} else {
				if wasLeader {
					onLoseLeadership(e.nodeID)
					wasLeader = false
				}
				e.TryBecomeLeader(ctx)
			}
		}
	}
}

// Example usage:
// hostname, _ := os.Hostname()
// nodeID := fmt.Sprintf("%s:%d", hostname, os.Getpid())
// el := New(rdb, nodeID, "cron-scheduler", 30*time.Second, 10*time.Second)
// el.Run(ctx,
//     func(id string) { log.Printf("Node %s is leader", id) },
//     func(id string) { log.Printf("Node %s lost leadership", id) },
// )
var _ = os.Hostname // reference for import
var _ = log.Println

namespace Election;

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;

// RedisLeaderElection implements leader election using Redis SET NX + TTL.
public sealed class RedisLeaderElection(
    IDatabase redis,
    string nodeId,
    string electionName,
    TimeSpan ttl)
{
    private readonly string _lockKey = $"leader:{electionName}";

    public string NodeId => nodeId;

    // Try to acquire leadership: SET key value NX EX ttl.
    public Task<bool> TryBecomeLeaderAsync() =>
        redis.StringSetAsync(_lockKey, nodeId, ttl, When.NotExists);

    // Extend TTL only if this node is still the leader (atomic Lua).
    public async Task<bool> RenewLeadershipAsync()
    {
        const string script = """
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("EXPIRE", KEYS[1], ARGV[2])
            end
            return 0
            """;

        var result = await redis.ScriptEvaluateAsync(
            script,
            [_lockKey],
            [nodeId, (int)ttl.TotalSeconds]);

        return (int)result == 1;
    }

    // Get current leader node ID, or null if there is none.
    public async Task<string?> GetCurrentLeaderAsync() =>
        await redis.StringGetAsync(_lockKey);

    public async Task<bool> IsLeaderAsync() =>
        await GetCurrentLeaderAsync() == nodeId;

    // Release leadership only if this node holds it (atomic).
    public async Task<bool> StepDownAsync()
    {
        const string script = """
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            end
            return 0
            """;

        var result = await redis.ScriptEvaluateAsync(script, [_lockKey], [nodeId]);
        return (int)result == 1;
    }
}

// The election loop runs as a hosted service -- the .NET equivalent of a
// long-lived background goroutine, with graceful shutdown via the token.
public sealed class LeaderElectionService(
    RedisLeaderElection election,
    ILogger<LeaderElectionService> logger,
    TimeSpan renewInterval) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var wasLeader = false;
        using var timer = new PeriodicTimer(renewInterval);

        try
        {
            while (await timer.WaitForNextTickAsync(stoppingToken))
            {
                if (await election.IsLeaderAsync())
                {
                    if (!wasLeader)
                    {
                        logger.LogInformation("Node {NodeId} is now the leader", election.NodeId);
                        wasLeader = true;
                    }

                    await election.RenewLeadershipAsync();
                }
                else
                {
                    if (wasLeader)
                    {
                        logger.LogInformation("Node {NodeId} lost leadership", election.NodeId);
                        wasLeader = false;
                    }

                    await election.TryBecomeLeaderAsync();
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Expected on shutdown -- fall through to release the lock
        }

        if (wasLeader)
        {
            await election.StepDownAsync();
            logger.LogInformation("Node {NodeId} stepped down on shutdown", election.NodeId);
        }
    }
}

import asyncio
import contextlib
import logging
import os
import socket
from collections.abc import Callable

import redis.asyncio as redis

logger = logging.getLogger(__name__)

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

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

class RedisLeaderElection:
    """Leader election using Redis SET NX + TTL."""

    def __init__(
        self,
        client: redis.Redis,
        node_id: str,
        election_name: str,
        ttl_seconds: int = 30,
        renew_interval_sec: int = 10,
    ) -> None:
        self._redis = client
        self._node_id = node_id
        self._lock_key = f"leader:{election_name}"
        self._ttl_seconds = ttl_seconds
        self._renew_interval_sec = renew_interval_sec
        self._renew = client.register_script(RENEW_SCRIPT)
        self._release = client.register_script(RELEASE_SCRIPT)

    async def try_become_leader(self) -> bool:
        """Set the lock only if absent (nx), expiring after ttl (ex)."""
        return bool(
            await self._redis.set(
                self._lock_key, self._node_id, nx=True, ex=self._ttl_seconds
            )
        )

    async def renew_leadership(self) -> bool:
        """Extend TTL only if this node is still the leader (atomic)."""
        result = await self._renew(
            keys=[self._lock_key], args=[self._node_id, self._ttl_seconds]
        )
        return result == 1

    async def get_current_leader(self) -> str | None:
        value = await self._redis.get(self._lock_key)
        return value.decode() if value is not None else None

    async def is_leader(self) -> bool:
        return await self.get_current_leader() == self._node_id

    async def step_down(self) -> bool:
        """Release the lock only if this node holds it (atomic)."""
        result = await self._release(keys=[self._lock_key], args=[self._node_id])
        return result == 1

    async def run(
        self,
        on_become_leader: Callable[[str], None],
        on_lose_leadership: Callable[[str], None],
    ) -> None:
        """Run the election loop until cancelled, stepping down on exit."""
        was_leader = False
        try:
            while True:
                if await self.is_leader():
                    if not was_leader:
                        on_become_leader(self._node_id)
                        was_leader = True
                    await self.renew_leadership()
                else:
                    if was_leader:
                        on_lose_leadership(self._node_id)
                        was_leader = False
                    await self.try_become_leader()

                await asyncio.sleep(self._renew_interval_sec)
        except asyncio.CancelledError:
            # Cancellation is the normal shutdown path -- release the lock first
            if was_leader:
                with contextlib.suppress(redis.RedisError):
                    await self.step_down()
                on_lose_leadership(self._node_id)
            raise

# Usage: scheduled task coordinator
async def main() -> None:
    election = RedisLeaderElection(
        client=redis.Redis(),
        node_id=f"{socket.gethostname()}:{os.getpid()}",
        election_name="cron-scheduler",
        ttl_seconds=30,
        renew_interval_sec=10,
    )

    await election.run(
        on_become_leader=lambda node_id: logger.info(
            "Node %s is now the leader. Starting scheduler.", node_id
        ),
        on_lose_leadership=lambda node_id: logger.info(
            "Node %s lost leadership. Stopping scheduler.", node_id
        ),
    )