23297885341

Committed 19 Mar 2026 01:44PM UTC coverage: 92.827%. First build

Build # 23297885341

Build Type

Pull #1656

github

Committed by

web-flow

Commit Message

Merge ebe61c872 into 1c4219cf8

Pull Request Pull Request #1656: Handle -READONLY as a redirection signal for Redis Cluster (AWS ElastiCache support)

Coverage Stats

1 of 8 new or added lines in 1 file covered. (12.5%)

8347 of 8992 relevant lines covered (92.83%)

114.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.59

/src/Connection/Cluster/RedisCluster.php

<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2026 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Connection\Cluster;

use ArrayIterator;
use Countable;
use IteratorAggregate;
use OutOfBoundsException;
use Predis\ClientException;
use Predis\Cluster\RedisStrategy as RedisClusterStrategy;
use Predis\Cluster\SlotMap;
use Predis\Cluster\StrategyInterface;
use Predis\Command\Command;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\Connection\AbstractAggregateConnection;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\ParametersInterface;
use Predis\Connection\RelayFactory;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ServerException;
use Predis\Retry\Retry;
use Predis\Retry\Strategy\ExponentialBackoff;
use Predis\TimeoutException;
use ReturnTypeWillChange;
use Throwable;
use Traversable;

/**
 * Abstraction for a Redis-backed cluster of nodes (Redis >= 3.0.0).
 *
 * This connection backend offers smart support for redis-cluster by handling
 * automatic slots map (re)generation upon -MOVED or -ASK responses returned by
 * Redis when redirecting a client to a different node.
 *
 * The cluster can be pre-initialized using only a subset of the actual nodes in
 * the cluster, Predis will do the rest by adjusting the slots map and creating
 * the missing underlying connection instances on the fly.
 *
 * It is possible to pre-associate connections to a slots range with the "slots"
 * parameter in the form "$first-$last". This can greatly reduce runtime node
 * guessing and redirections.
 *
 * It is also possible to ask for the full and updated slots map directly to one
 * of the nodes and optionally enable such a behaviour upon -MOVED redirections.
 * Asking for the cluster configuration to Redis is actually done by issuing a
 * CLUSTER SLOTS command to a random node in the pool.
 */
class RedisCluster extends AbstractAggregateConnection implements ClusterInterface, IteratorAggregate, Countable
{
    public $useClusterSlots = true;

    /**
     * @var NodeConnectionInterface[]
     */
    private $pool = [];
    private $slots = [];
    private $slotmap;
    private $strategy;
    private $connections;
    private $retryLimit = 5;
    private $retryInterval = 10;

    /**
     * @var int
     */
    private $readTimeout = 1000;

    /**
     * @var ParametersInterface
     */
    private $connectionParameters;

    /**
     * @param FactoryInterface       $connections Optional connection factory.
     * @param StrategyInterface|null $strategy    Optional cluster strategy.
     * @param int|null               $readTimeout Optional read timeout
     */
    public function __construct(
        FactoryInterface $connections,
        ParametersInterface $parameters,
        ?StrategyInterface $strategy = null,
        ?int $readTimeout = null
    ) {
        $this->connections = $connections;
        $this->connectionParameters = $parameters;
        $this->strategy = $strategy ?: new RedisClusterStrategy();
        $this->slotmap = new SlotMap();

        if (!is_null($readTimeout)) {
            $this->readTimeout = $readTimeout;
        }
    }

    /**
     * Sets the maximum number of retries for commands upon server failure.
     *
     * -1 = unlimited retry attempts
     *  0 = no retry attempts (fails immediately)
     *  n = fail only after n retry attempts
     *
     * @param int $retry Number of retry attempts.
     */
    public function setRetryLimit($retry)
    {
        $this->retryLimit = (int) $retry;
    }

    /**
     * Sets the initial retry interval (milliseconds).
     *
     * @param int $retryInterval Milliseconds between retries.
     */
    public function setRetryInterval($retryInterval)
    {
        $this->retryInterval = (int) $retryInterval;
    }

    /**
     * Returns the retry interval (milliseconds).
     *
     * @return int Milliseconds between retries.
     */
    public function getRetryInterval()
    {
        return (int) $this->retryInterval;
    }

    /**
     * {@inheritdoc}
     */
    public function isConnected()
    {
        foreach ($this->pool as $connection) {
            if ($connection->isConnected()) {
                return true;
            }
        }

        return false;
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        foreach ($this->pool as $connection) {
            $connection->connect();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        foreach ($this->pool as $connection) {
            $connection->disconnect();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function add(NodeConnectionInterface $connection)
    {
        $this->pool[(string) $connection] = $connection;
        $this->slotmap->reset();
    }

    /**
     * {@inheritdoc}
     */
    public function remove(NodeConnectionInterface $connection)
    {
        if (false !== $id = array_search($connection, $this->pool, true)) {
            $this->slotmap->reset();
            $this->slots = array_diff($this->slots, [$connection]);
            unset($this->pool[$id]);

            return true;
        }

        return false;
    }

    /**
     * Removes a connection instance by using its identifier.
     *
     * @param string $connectionID Connection identifier.
     *
     * @return bool True if the connection was in the pool.
     */
    public function removeById($connectionID)
    {
        if (isset($this->pool[$connectionID])) {
            $this->slotmap->reset();
            $this->slots = array_diff($this->slots, [$connectionID]);
            unset($this->pool[$connectionID]);

            return true;
        }

        return false;
    }

    /**
     * Generates the current slots map by guessing the cluster configuration out
     * of the connection parameters of the connections in the pool.
     *
     * Generation is based on the same algorithm used by Redis to generate the
     * cluster, so it is most effective when all of the connections supplied on
     * initialization have the "slots" parameter properly set accordingly to the
     * current cluster configuration.
     */
    public function buildSlotMap()
    {
        $this->slotmap->reset();

        foreach ($this->pool as $connectionID => $connection) {
            $parameters = $connection->getParameters();

            if (!isset($parameters->slots)) {
                continue;
            }

            foreach (explode(',', $parameters->slots) as $slotRange) {
                $slots = explode('-', $slotRange, 2);

                if (!isset($slots[1])) {
                    $slots[1] = $slots[0];
                }

                $this->slotmap->setSlots($slots[0], $slots[1], $connectionID);
            }
        }
    }

    /**
     * Queries the specified node of the cluster to fetch the updated slots map.
     *
     * When the connection fails, this method tries to execute the same command
     * on a different connection picked at random from the pool of known nodes,
     * up until the retry limit is reached.
     *
     * @param NodeConnectionInterface $connection Connection to a node of the cluster.
     *
     * @return mixed
     */
    private function queryClusterNodeForSlotMap(NodeConnectionInterface $connection)
    {
        // Backward-compatible hardcoded retry
        $retry = new Retry(
            new ExponentialBackoff($this->retryInterval * 1000, -1),
            $this->retryLimit,
            [ConnectionException::class]
        );

        $command = RawCommand::create('CLUSTER', 'SLOTS');

        $doCallback = static function () use (&$connection, $command) {
            return $connection->executeCommand($command);
        };

        $failCallback = function (ConnectionException $exception) use (&$connection) {
            $connection = $exception->getConnection();
            $connection->disconnect();

            $this->remove($connection);

            if (!$connection = $this->getRandomConnection()) {
                throw new ClientException('No connections left in the pool for `CLUSTER SLOTS`');
            }
        };

        return $retry->callWithRetry($doCallback, $failCallback);
    }

    /**
     * Generates an updated slots map fetching the cluster configuration using
     * the CLUSTER SLOTS command against the specified node or a random one from
     * the pool.
     *
     * @param NodeConnectionInterface|null $connection Optional connection instance.
     */
    public function askSlotMap(?NodeConnectionInterface $connection = null)
    {
        if (!$connection && !$connection = $this->getRandomConnection()) {
            return;
        }

        $this->slotmap->reset();

        $response = $this->queryClusterNodeForSlotMap($connection);

        foreach ($response as $slots) {
            // We only support master servers for now, so we ignore subsequent
            // elements in the $slots array identifying slaves.
            [$start, $end, $master] = $slots;

            if ($master[0] === '') {
                $this->slotmap->setSlots($start, $end, (string) $connection);
            } else {
                $this->slotmap->setSlots($start, $end, "{$master[0]}:{$master[1]}");
            }
        }
    }

    /**
     * Guesses the correct node associated to a given slot using a precalculated
     * slots map, falling back to the same logic used by Redis to initialize a
     * cluster (best-effort).
     *
     * @param int $slot Slot index.
     *
     * @return string Connection ID.
     */
    protected function guessNode($slot)
    {
        if (!$this->pool) {
            throw new ClientException('No connections available in the pool');
        }

        if ($this->slotmap->isEmpty()) {
            $this->buildSlotMap();
        }

        if ($node = $this->slotmap[$slot]) {
            return $node;
        }

        $count = count($this->pool);
        $index = min((int) ($slot / (int) (16384 / $count)), $count - 1);
        $nodes = array_keys($this->pool);

        return $nodes[$index];
    }

    /**
     * Creates a new connection instance from the given connection ID.
     *
     * @param string $connectionID Identifier for the connection.
     *
     * @return NodeConnectionInterface
     */
    protected function createConnection($connectionID)
    {
        $separator = strrpos($connectionID, ':');

        return $this->connections->create([
            'host' => substr($connectionID, 0, $separator),
            'port' => substr($connectionID, $separator + 1),
        ]);
    }

    /**
     * {@inheritdoc}
     */
    public function getConnectionByCommand(CommandInterface $command)
    {
        $slot = $this->strategy->getSlot($command);

        if (!isset($slot)) {
            throw new NotSupportedException(
                "Cannot use '{$command->getId()}' with redis-cluster."
            );
        }

        if (isset($this->slots[$slot])) {
            return $this->slots[$slot];
        }

        return $this->getConnectionBySlot($slot);
    }

    /**
     * Returns the connection currently associated to a given slot.
     *
     * @param int $slot Slot index.
     *
     * @return NodeConnectionInterface
     * @throws OutOfBoundsException
     */
    public function getConnectionBySlot($slot)
    {
        if (!SlotMap::isValid($slot)) {
            throw new OutOfBoundsException("Invalid slot [$slot].");
        }

        if (isset($this->slots[$slot])) {
            return $this->slots[$slot];
        }

        $connectionID = $this->guessNode($slot);

        if (!$connection = $this->getConnectionById($connectionID)) {
            $connection = $this->createConnection($connectionID);
            $this->pool[$connectionID] = $connection;
        }

        return $this->slots[$slot] = $connection;
    }

    /**
     * {@inheritdoc}
     */
    public function getConnectionById($connectionID)
    {
        return $this->pool[$connectionID] ?? null;
    }

    /**
     * Returns a random connection from the pool.
     *
     * @return NodeConnectionInterface|null
     */
    protected function getRandomConnection()
    {
        if (!$this->pool) {
            return null;
        }

        return $this->pool[array_rand($this->pool)];
    }

    /**
     * Permanently associates the connection instance to a new slot.
     * The connection is added to the connections pool if not yet included.
     *
     * @param NodeConnectionInterface $connection Connection instance.
     * @param int                     $slot       Target slot index.
     */
    protected function move(NodeConnectionInterface $connection, $slot)
    {
        $this->pool[(string) $connection] = $connection;
        $this->slots[(int) $slot] = $connection;
        $this->slotmap[(int) $slot] = $connection;
    }

    /**
     * Handles -ERR responses returned by Redis.
     *
     * @param CommandInterface       $command Command that generated the -ERR response.
     * @param ErrorResponseInterface $error   Redis error response object.
     *
     * @return mixed
     */
    protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $error)
    {
        $details = explode(' ', $error->getMessage(), 2);

        switch ($details[0]) {
            case 'READONLY':
                return $this->onReadOnlyResponse($command);

            case 'MOVED':
                return $this->onMovedResponse($command, $details[1]);

            case 'ASK':
                return $this->onAskResponse($command, $details[1]);

            default:
                return $error;
        }
    }

    /**
     * Handles -READONLY responses by disconnecting the current node's connection
     * and refreshing the slots map (when cluster slots are enabled), then
     * re-executing the command so it is routed to the updated primary node.
     *
     * This is a workaround for AWS ElastiCache Redis OSS, which may return
     * -READONLY errors during failover events. Standard Redis clusters do not
     * exhibit this behavior.
     *
     * @param CommandInterface $command Command that generated the -READONLY response.
     *
     * @return mixed
     */
    protected function onReadOnlyResponse(CommandInterface $command)
    {
        if ($this->useClusterSlots) {
            $connection = $this->getConnectionByCommand($command);
            $connection->disconnect();
            $this->askSlotMap();
        }

        return $this->executeCommand($command);
    }

    /**
     * Handles -MOVED responses by executing again the command against the node
     * indicated by the Redis response.
     *
     * @param CommandInterface $command Command that generated the -MOVED response.
     * @param string           $details Parameters of the -MOVED response.
     *
     * @return mixed
     */
    protected function onMovedResponse(CommandInterface $command, $details)
    {
        [$slot, $connectionID] = explode(' ', $details, 2);

        // Handle connection ID in the form of "IP:port (details about exception)"
        // by trimming everything after first space (including the space)
        $startPositionOfExtraDetails = strpos($connectionID, ' ');

        if ($startPositionOfExtraDetails !== false) {
            $connectionID = substr($connectionID, 0, $startPositionOfExtraDetails);
        }

        if (!$connection = $this->getConnectionById($connectionID)) {
            $connection = $this->createConnection($connectionID);
        }

        if ($this->useClusterSlots) {
            $this->askSlotMap($connection);
        }

        $this->move($connection, $slot);

        return $this->executeCommand($command);
    }

    /**
     * Handles -ASK responses by executing again the command against the node
     * indicated by the Redis response.
     *
     * @param CommandInterface $command Command that generated the -ASK response.
     * @param string           $details Parameters of the -ASK response.
     *
     * @return mixed
     */
    protected function onAskResponse(CommandInterface $command, $details)
    {
        [$slot, $connectionID] = explode(' ', $details, 2);

        if (!$connection = $this->getConnectionById($connectionID)) {
            $connection = $this->createConnection($connectionID);
        }

        $connection->executeCommand(RawCommand::create('ASKING'));

        return $connection->executeCommand($command);
    }

    /**
     * Ensures that a command is executed one more time on connection failure.
     *
     * The connection to the node that generated the error is evicted from the
     * pool before trying to fetch an updated slots map from another node. If
     * the new slots map points to an unreachable server the client gives up and
     * throws the exception as the nodes participating in the cluster may still
     * have to agree that something changed in the configuration of the cluster.
     *
     * @param CommandInterface $command Command instance.
     * @param string           $method  Actual method.
     *
     * @return mixed
     * @throws Throwable
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
        if ($this->connectionParameters->isDisabledRetry() || $this->connections instanceof RelayFactory) {
            // Override default parameters, for backward-compatibility
            // with current behaviour
            $retry = new Retry(
                new ExponentialBackoff($this->retryInterval * 1000, -1),
                $this->retryLimit
            );
        } else {
            $retry = $this->connectionParameters->retry;
        }
        $retry->updateCatchableExceptions([ServerException::class]);

        $doCallback = function () use ($command, $method) {
            $response = $this->getConnectionByCommand($command)->$method($command);

            if ($response instanceof ErrorResponse) {
                $message = $response->getMessage();

                if (strpos($message, 'CLUSTERDOWN') !== false) {
                    throw new ServerException($message);
                }
            }

            return $response;
        };

        return $retry->callWithRetry(
            $doCallback,
            function (Throwable $e) {
                $this->onFailCallback($e);
            }
        );
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function readResponse(CommandInterface $command)
    {
        return $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function executeCommand(CommandInterface $command)
    {
        $response = $this->retryCommandOnFailure($command, __FUNCTION__);

        if ($response instanceof ErrorResponseInterface) {
            return $this->onErrorResponse($command, $response);
        }

        return $response;
    }

    /**
     * {@inheritdoc}
     */
    public function executeCommandOnEachNode(CommandInterface $command): array
    {
        $responses = [];

        foreach ($this->pool as $connection) {
            $responses[] = $connection->executeCommand($command);
        }

        return $responses;
    }

    /**
     * @return int
     */
    #[ReturnTypeWillChange]
    public function count()
    {
        return count($this->pool);
    }

    /**
     * @return Traversable<string, NodeConnectionInterface>
     */
    #[ReturnTypeWillChange]
    public function getIterator()
    {
        if ($this->slotmap->isEmpty()) {
            $this->useClusterSlots ? $this->askSlotMap() : $this->buildSlotMap();
        }

        $connections = [];

        foreach ($this->slotmap->getNodes() as $node) {
            if (!$connection = $this->getConnectionById($node)) {
                $this->add($connection = $this->createConnection($node));
            }

            $connections[] = $connection;
        }

        return new ArrayIterator($connections);
    }

    /**
     * Returns the underlying slot map.
     *
     * @return SlotMap
     */
    public function getSlotMap()
    {
        return $this->slotmap;
    }

    /**
     * {@inheritDoc}
     */
    public function getClusterStrategy(): StrategyInterface
    {
        return $this->strategy;
    }

    /**
     * Returns the underlying connection factory used to create new connection
     * instances to Redis nodes indicated by redis-cluster.
     *
     * @return FactoryInterface
     */
    public function getConnectionFactory()
    {
        return $this->connections;
    }

    /**
     * Enables automatic fetching of the current slots map from one of the nodes
     * using the CLUSTER SLOTS command. This option is enabled by default as
     * asking the current slots map to Redis upon -MOVED responses may reduce
     * overhead by eliminating the trial-and-error nature of the node guessing
     * procedure, mostly when targeting many keys that would end up in a lot of
     * redirections.
     *
     * The slots map can still be manually fetched using the askSlotMap()
     * method whether or not this option is enabled.
     *
     * @param bool $value Enable or disable the use of CLUSTER SLOTS.
     */
    public function useClusterSlots($value)
    {
        $this->useClusterSlots = (bool) $value;
    }

    /**
     * {@inheritdoc}
     */
    public function getParameters(): ?ParametersInterface
    {
        return $this->connectionParameters;
    }

    /**
     * Loop over connections until there's data to read.
     *
     * @return mixed
     */
    public function read()
    {
        while (true) {
            foreach ($this->pool as $connection) {
                if ($connection->hasDataToRead()) {
                    return $connection->read();
                }
            }

            usleep($this->readTimeout);
        }
    }

    /**
     * Handle exceptions.
     *
     * @param  Throwable $exception
     * @return void
     */
    private function onFailCallback(Throwable $exception)
    {
        if ($exception instanceof ConnectionException) {
            $connection = $exception->getConnection();

            if ($connection) {
                $connection->disconnect();
                $this->remove($connection);
            }

            if ($this->useClusterSlots) {
                $this->askSlotMap();
            }
        }

        if ($exception instanceof TimeoutException) {
            $connection = $exception->getConnection();

            if ($connection) {
                $connection->disconnect();
            }
        }
    }
}

predis / predis / 23297885341

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous