22723904332

Committed 05 Mar 2026 03:01PM UTC coverage: 92.905% (+0.01%) from 92.892%

Build # 22723904332

Build Type

push

github

Committed by

web-flow

Commit Message

Fix sentinel discovery methods not catching StreamInitException (#1650)

* Fix sentinel discovery methods not catching StreamInitException

The getMaster(), getSlaves(), and updateSentinels() methods only catch
ConnectionException when connecting to Sentinel nodes. StreamInitException
(thrown by stream_socket_client() failures) extends PredisException directly,
not ConnectionException, so it propagates uncaught — preventing fallback to
the next Sentinel node.

This causes complete application failure when any single Sentinel is
unreachable, even if other Sentinels are healthy.

Add StreamInitException to the catch clause using a union type in all three
methods. This is more precise than catching PredisException, which would
also swallow ServerException (e.g. "ERR No such master with that name")
and mask configuration errors.

See also #1577 which applied a similar fix to retryCommandOnFailure().

* Fix coding standards and add changelog entry

- Remove spaces around | in union catch (php-cs-fixer)
- Add CHANGELOG.md entry for #1650

* Codestlye fixes

---------

Co-authored-by: Wolfgang Kerschbaumer <wolfgang.kerschbaumer@bergfex.at>
Co-authored-by: vladvildanov <vladyslav.vildanov@redis.com>

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

20 existing lines in 1 file now uncovered.

8328 of 8964 relevant lines covered (92.9%)

113.97 hits per line

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

91.76

/src/Connection/Replication/SentinelReplication.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\Replication;

use InvalidArgumentException;
use Predis\Command\Command;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\CommunicationException;
use Predis\Connection\AbstractAggregateConnection;
use Predis\Connection\ConnectionException;
use Predis\Connection\FactoryInterface as ConnectionFactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\Parameters;
use Predis\Connection\ParametersInterface;
use Predis\Connection\RelayFactory;
use Predis\Connection\Resource\Exception\StreamInitException;
use Predis\Replication\ReplicationStrategy;
use Predis\Replication\RoleException;
use Predis\Response\Error;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ServerException;
use Predis\Retry\Retry;
use Predis\Retry\Strategy\ExponentialBackoff;
use Throwable;

/**
 * @author Daniele Alessandri <suppakilla@gmail.com>
 * @author Ville Mattila <ville@eventio.fi>
 */
class SentinelReplication extends AbstractAggregateConnection implements ReplicationInterface
{
    /**
     * @var NodeConnectionInterface
     */
    protected $master;

    /**
     * @var NodeConnectionInterface[]
     */
    protected $slaves = [];

    /**
     * @var NodeConnectionInterface[]
     */
    protected $pool = [];

    /**
     * @var NodeConnectionInterface
     */
    protected $current;

    /**
     * @var string
     */
    protected $service;

    /**
     * @var ConnectionFactoryInterface
     */
    protected $connectionFactory;

    /**
     * @var ReplicationStrategy
     */
    protected $strategy;

    /**
     * Sentinel connection parameters.
     *
     * Can contain:
     * - String URIs (e.g., "tcp://127.0.0.1:26379")
     * - Arrays of connection parameters (e.g., ['host' => '127.0.0.1', 'port' => 26379])
     * - ParametersInterface objects
     * - NodeConnectionInterface objects
     *
     * @var array<string|array|ParametersInterface|NodeConnectionInterface>
     */
    protected $sentinels = [];

    /**
     * @var int
     */
    protected $sentinelIndex = 0;

    /**
     * @var NodeConnectionInterface
     */
    protected $sentinelConnection;

    /**
     * @var float
     */
    protected $sentinelTimeout = 0.100;

    /**
     * Max number of automatic retries of commands upon server failure.
     *
     * -1 = unlimited retry attempts
     *  0 = no retry attempts (fails immediately)
     *  n = fail only after n retry attempts
     *
     * @var int
     */
    protected $retryLimit = 20;

    /**
     * Time to wait in milliseconds before fetching a new configuration from one
     * of the sentinel servers.
     *
     * @var int
     */
    protected $retryWait = 1000;

    /**
     * Flag for automatic fetching of available sentinels.
     *
     * @var bool
     */
    protected $updateSentinels = false;

    /**
     * @param string                     $service           Name of the service for autodiscovery.
     * @param array                      $sentinels         Sentinel servers connection parameters.
     * @param ConnectionFactoryInterface $connectionFactory Connection factory instance.
     * @param ReplicationStrategy|null   $strategy          Replication strategy instance.
     */
    public function __construct(
        $service,
        array $sentinels,
        ConnectionFactoryInterface $connectionFactory,
        ?ReplicationStrategy $strategy = null
    ) {
        $this->sentinels = $sentinels;
        $this->service = $service;
        $this->connectionFactory = $connectionFactory;
        $this->strategy = $strategy ?: new ReplicationStrategy();
    }

    /**
     * Sets a default timeout for connections to sentinels.
     *
     * When "timeout" is present in the connection parameters of sentinels, its
     * value overrides the default sentinel timeout.
     *
     * @param float $timeout Timeout value.
     */
    public function setSentinelTimeout($timeout)
    {
        $this->sentinelTimeout = (float) $timeout;
    }

    /**
     * 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 time to wait (in milliseconds) before fetching a new configuration
     * from one of the sentinels.
     *
     * @param float $milliseconds Time to wait before the next attempt.
     */
    public function setRetryWait($milliseconds)
    {
        $this->retryWait = (float) $milliseconds;
    }

    /**
     * Set automatic fetching of available sentinels.
     *
     * @param bool $update Enable or disable automatic updates.
     */
    public function setUpdateSentinels($update)
    {
        $this->updateSentinels = (bool) $update;
    }

    /**
     * Resets the current connection.
     */
    protected function reset()
    {
        $this->current = null;
    }

    /**
     * Wipes the current list of master and slaves nodes.
     */
    protected function wipeServerList()
    {
        $this->reset();

        $this->master = null;
        $this->slaves = [];
        $this->pool = [];
    }

    /**
     * {@inheritdoc}
     */
    public function add(NodeConnectionInterface $connection)
    {
        $parameters = $connection->getParameters();
        $role = $parameters->role;

        if ('master' === $role) {
            $this->master = $connection;
        } elseif ('sentinel' === $role) {
            $this->sentinels[] = $connection;

            // sentinels are not considered part of the pool.
            return;
        } else {
            // everything else is considered a slave.
            $this->slaves[] = $connection;
        }

        $this->pool[(string) $connection] = $connection;

        $this->reset();
    }

    /**
     * {@inheritdoc}
     */
    public function remove(NodeConnectionInterface $connection)
    {
        if ($connection === $this->master) {
            $this->master = null;
        } elseif (false !== $id = array_search($connection, $this->slaves, true)) {
            unset($this->slaves[$id]);
        } elseif (false !== $id = array_search($connection, $this->sentinels, true)) {
            unset($this->sentinels[$id]);

            return true;
        } else {
            return false;
        }

        unset($this->pool[(string) $connection]);

        $this->reset();

        return true;
    }

    /**
     * Creates a new connection to a sentinel server.
     *
     * @return NodeConnectionInterface
     */
    protected function createSentinelConnection($parameters)
    {
        if ($parameters instanceof NodeConnectionInterface) {
            return $parameters;
        }

        if (is_string($parameters)) {
            $parameters = Parameters::parse($parameters);
        }

        if (is_array($parameters)) {
            // NOTE: sentinels do not accept SELECT command so we must
            // explicitly set it to NULL to avoid problems when using default
            // parameters set via client options.
            $parameters['database'] = null;

            // don't leak password from between configurations
            // https://github.com/predis/predis/pull/807/#discussion_r985764770
            if (!isset($parameters['password'])) {
                $parameters['password'] = null;
            }

            if (!isset($parameters['timeout'])) {
                $parameters['timeout'] = $this->sentinelTimeout;
            }
        }

        return $this->connectionFactory->create($parameters);
    }

    /**
     * Returns the current sentinel connection.
     *
     * If there is no active sentinel connection, a new connection is created.
     *
     * @return NodeConnectionInterface
     */
    public function getSentinelConnection()
    {
        if (!$this->sentinelConnection) {
            if ($this->sentinelIndex >= count($this->sentinels)) {
                $this->sentinelIndex = 0;
                throw new \Predis\ClientException('No sentinel server available for autodiscovery.');
            }

            $sentinel = $this->sentinels[$this->sentinelIndex];
            ++$this->sentinelIndex;
            $this->sentinelConnection = $this->createSentinelConnection($sentinel);
        }

        return $this->sentinelConnection;
    }

    /**
     * Fetches an updated list of sentinels from a sentinel.
     */
    public function updateSentinels()
    {
        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $payload = $sentinel->executeCommand(
                    RawCommand::create('SENTINEL', 'sentinels', $this->service)
                );

                $this->sentinels = [];
                $this->sentinelIndex = 0;
                // NOTE: sentinel server does not return itself, so we add it back.
                $this->sentinels[] = $sentinel->getParameters()->toArray();

                foreach ($payload as $sentinel) {
                    $this->sentinels[] = [
                        'host' => $sentinel[3],
                        'port' => $sentinel[5],
                        'role' => 'sentinel',
                    ];
                }
            } catch (ConnectionException|StreamInitException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }
    }

    /**
     * Fetches the details for the master and slave servers from a sentinel.
     */
    public function querySentinel()
    {
        $this->wipeServerList();

        $this->updateSentinels();
        $this->getMaster();
        $this->getSlaves();
    }

    /**
     * Handles error responses returned by redis-sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param ErrorResponseInterface  $error    Error response.
     */
    private function handleSentinelErrorResponse(NodeConnectionInterface $sentinel, ErrorResponseInterface $error)
    {
        if ($error->getErrorType() === 'IDONTKNOW') {
            throw new ConnectionException($sentinel, $error->getMessage());
        }
        throw new ServerException($error->getMessage());
    }

    /**
     * Fetches the details for the master server from a sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param string                  $service  Name of the service.
     *
     * @return array
     */
    protected function querySentinelForMaster(NodeConnectionInterface $sentinel, $service)
    {
        $payload = $sentinel->executeCommand(
            RawCommand::create('SENTINEL', 'get-master-addr-by-name', $service)
        );

        if ($payload === null) {
            throw new ServerException('ERR No such master with that name');
        }

        if ($payload instanceof ErrorResponseInterface) {
            $this->handleSentinelErrorResponse($sentinel, $payload);
        }

        return [
            'host' => $payload[0],
            'port' => $payload[1],
            'role' => 'master',
        ];
    }

    /**
     * Fetches the details for the slave servers from a sentinel.
     *
     * @param NodeConnectionInterface $sentinel Connection to a sentinel server.
     * @param string                  $service  Name of the service.
     *
     * @return array
     */
    protected function querySentinelForSlaves(NodeConnectionInterface $sentinel, $service)
    {
        $slaves = [];

        $payload = $sentinel->executeCommand(
            RawCommand::create('SENTINEL', 'slaves', $service)
        );

        if ($payload instanceof ErrorResponseInterface) {
            $this->handleSentinelErrorResponse($sentinel, $payload);
        }

        foreach ($payload as $slave) {
            $flags = explode(',', $slave[9]);

            if (array_intersect($flags, ['s_down', 'o_down', 'disconnected'])) {
                continue;
            }

            // ensure `master-link-status` is ok
            if (isset($slave[31]) && $slave[31] === 'err') {
                continue;
            }

            $slaves[] = [
                'host' => $slave[3],
                'port' => $slave[5],
                'role' => 'slave',
            ];
        }

        return $slaves;
    }

    /**
     * {@inheritdoc}
     */
    public function getCurrent()
    {
        return $this->current;
    }

    /**
     * {@inheritdoc}
     */
    public function getMaster()
    {
        if ($this->master) {
            return $this->master;
        }

        if ($this->updateSentinels) {
            $this->updateSentinels();
        }

        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $masterParameters = $this->querySentinelForMaster($sentinel, $this->service);
                $masterConnection = $this->connectionFactory->create($masterParameters);

                $this->add($masterConnection);
            } catch (ConnectionException|StreamInitException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }

        return $masterConnection;
    }

    /**
     * {@inheritdoc}
     */
    public function getSlaves()
    {
        if ($this->slaves) {
            return array_values($this->slaves);
        }

        if ($this->updateSentinels) {
            $this->updateSentinels();
        }

        SENTINEL_QUERY: {
            $sentinel = $this->getSentinelConnection();

            try {
                $slavesParameters = $this->querySentinelForSlaves($sentinel, $this->service);

                foreach ($slavesParameters as $slaveParameters) {
                    $this->add($this->connectionFactory->create($slaveParameters));
                }
            } catch (ConnectionException|StreamInitException $exception) {
                $this->sentinelConnection = null;

                goto SENTINEL_QUERY;
            }
        }

        return array_values($this->slaves);
    }

    /**
     * Returns a random slave.
     *
     * @return NodeConnectionInterface|null
     */
    protected function pickSlave()
    {
        $slaves = $this->getSlaves();

        return $slaves
            ? $slaves[rand(1, count($slaves)) - 1]
            : null;
    }

    /**
     * Returns the connection instance in charge for the given command.
     *
     * @param CommandInterface $command Command instance.
     *
     * @return NodeConnectionInterface
     */
    private function getConnectionInternal(CommandInterface $command)
    {
        if (!$this->current) {
            if ($this->strategy->isReadOperation($command) && $slave = $this->pickSlave()) {
                $this->current = $slave;
            } else {
                $this->current = $this->getMaster();
            }

            return $this->current;
        }

        if ($this->current === $this->master) {
            return $this->current;
        }

        if (!$this->strategy->isReadOperation($command)) {
            $this->current = $this->getMaster();
        }

        return $this->current;
    }

    /**
     * Asserts that the specified connection matches an expected role.
     *
     * @param NodeConnectionInterface $connection Connection to a redis server.
     * @param string                  $role       Expected role of the server ("master", "slave" or "sentinel").
     *
     * @throws RoleException|ConnectionException
     */
    protected function assertConnectionRole(NodeConnectionInterface $connection, $role)
    {
        $role = strtolower($role);
        $retry = $connection->getParameters()->retry;
        $actualRole = $retry->callWithRetry(function () use ($connection) {
            return $connection->executeCommand(RawCommand::create('ROLE'));
        });

        if ($actualRole instanceof Error) {
            throw new ConnectionException($connection, $actualRole->getMessage());
        }

        if ($role !== $actualRole[0]) {
            throw new RoleException($connection, "Expected $role but got $actualRole[0] [$connection]");
        }
    }

    /**
     * {@inheritdoc}
     */
    public function getConnectionByCommand(CommandInterface $command)
    {
        $connection = $this->getConnectionInternal($command);

        if (!$connection->isConnected()) {
            // When we do not have any available slave in the pool we can expect
            // read-only operations to hit the master server.
            $expectedRole = $this->strategy->isReadOperation($command) && $this->slaves ? 'slave' : 'master';
            $this->assertConnectionRole($connection, $expectedRole);
        }

        return $connection;
    }

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

    /**
     * Returns a connection by its role.
     *
     * @param string $role Connection role (`master`, `slave` or `sentinel`)
     *
     * @return NodeConnectionInterface|null
     */
    public function getConnectionByRole($role)
    {
        if ($role === 'master') {
            return $this->getMaster();
        } elseif ($role === 'slave') {
            return $this->pickSlave();
        } elseif ($role === 'sentinel') {
            return $this->getSentinelConnection();
        }

        return null;
    }

    /**
     * Switches the internal connection in use by the backend.
     *
     * Sentinel connections are not considered as part of the pool, meaning that
     * trying to switch to a sentinel will throw an exception.
     *
     * @param NodeConnectionInterface $connection Connection instance in the pool.
     */
    public function switchTo(NodeConnectionInterface $connection)
    {
        if ($connection && $connection === $this->current) {
            return;
        }

        if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) {
            throw new InvalidArgumentException('Invalid connection or connection not found.');
        }

        $connection->connect();

        if ($this->current) {
            $this->current->disconnect();
        }

        $this->current = $connection;
    }

    /**
     * {@inheritdoc}
     */
    public function switchToMaster()
    {
        $connection = $this->getConnectionByRole('master');
        $this->switchTo($connection);
    }

    /**
     * {@inheritdoc}
     */
    public function switchToSlave()
    {
        $connection = $this->getConnectionByRole('slave');
        $this->switchTo($connection);
    }

    /**
     * {@inheritdoc}
     */
    public function isConnected()
    {
        return $this->current ? $this->current->isConnected() : false;
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        if (!$this->current) {
            if (!$this->current = $this->pickSlave()) {
                $this->current = $this->getMaster();
            }
        }

        $this->current->connect();
    }

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

    /**
     * Retries the execution of a command upon server failure after asking a new
     * configuration to one of the sentinels.
     *
     * @param CommandInterface $command Command instance.
     * @param string           $method  Actual method.
     *
     * @return mixed
     */
    private function retryCommandOnFailure(CommandInterface $command, $method)
    {
        $parameters = $this->getParameters();

        if ($parameters->isDisabledRetry() || $this->connectionFactory instanceof RelayFactory) {
            // Override default parameters, for backward-compatibility
            // with current behaviour
            $retry = new Retry(
                new ExponentialBackoff($this->retryWait * 1000, -1),
                $this->retryLimit
            );
        } else {
            $retry = $parameters->retry;
        }
        $retry->updateCatchableExceptions([Throwable::class]);

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

            if ($response instanceof Error && $response->getErrorType() === 'LOADING') {
                throw new ConnectionException($this->current, $response->getMessage());
            }

            return $response;
        };

        $failCallback = function (Throwable $exception) {
            $this->wipeServerList();

            if ($exception instanceof CommunicationException) {
                $exception->getConnection()->disconnect();
            }
        };

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

    /**
     * {@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)
    {
        return $this->retryCommandOnFailure($command, __FUNCTION__);
    }

    /**
     * Returns the underlying replication strategy.
     *
     * @return ReplicationStrategy
     */
    public function getReplicationStrategy()
    {
        return $this->strategy;
    }

    /**
     * {@inheritdoc}
     */
    public function __sleep()
    {
        return [
            'master', 'slaves', 'pool', 'service', 'sentinels', 'connectionFactory', 'strategy',
        ];
    }

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

        if (!empty($this->slaves)) {
            return $this->slaves[0]->getParameters();
        }

        if (!empty($this->sentinels)) {
            $sentinel = $this->sentinels[0];

            // Handle string URIs (e.g., "tcp://127.0.0.1:26379")
            if (is_string($sentinel)) {
                return new Parameters(Parameters::parse($sentinel));
            }

            // After querySentinels(), sentinels array contains plain arrays instead of connection objects
            if (is_array($sentinel)) {
                return new Parameters($sentinel);
            }

            if ($sentinel instanceof ParametersInterface) {
                return $sentinel;
            }

            return $sentinel->getParameters();
        }

        return null;
    }
}

predis / predis / 22723904332

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